Function scanPgStatActivityRow

tests/integration/pg_stat_activity_test.go:15–67

scanPgStatActivityRow scans a full pg_stat_activity row into a map. All 21 columns are always returned regardless of the SELECT column list, so we must scan all of them.

(rows *sql.Rows)

// All 21 columns are always returned regardless of the SELECT column list,
// so we must scan all of them.
func scanPgStatActivityRow(rows *sql.Rows) (map[string]interface{}, error) {
	var (
		datid, pid, usesysid, clientPort, workerID             int
		datname, usename, appName, clientAddr, state, query    string
		backendType                                            string
		backendStart                                           time.Time
		xactStart, queryStart, stateChange                     sql.NullTime
		waitEventType, waitEvent                               sql.NullString
		backendXid, backendXmin, leaderPid                     sql.NullInt32
		queryProgress                                          float64
		rowsProcessed, totalRowsToProcess                      int64
	)

	err := rows.Scan(
		&datid, &datname, &pid, &usesysid, &usename,
		&appName, &clientAddr, &clientPort,
		&backendStart, &xactStart, &queryStart, &stateChange,
		&waitEventType, &waitEvent, &state,
		&backendXid, &backendXmin, &query,
		&backendType, &leaderPid, &workerID,
		&queryProgress, &rowsProcessed, &totalRowsToProcess,
	)
	if err != nil {
		return nil, err
	}

	return map[string]interface{}{
		"datid":                  datid,
		"datname":                datname,
		"pid":                    pid,
		"usesysid":               usesysid,
		"usename":                usename,
		"application_name":       appName,
		"client_addr":            clientAddr,
		"client_port":            clientPort,
		"backend_start":          backendStart,
		"xact_start":             xactStart,
		"query_start":            queryStart,
		"state_change":           stateChange,
		"wait_event_type":        waitEventType,
		"wait_event":             waitEvent,
		"state":                  state,
		"backend_xid":            backendXid,
		"backend_xmin":           backendXmin,
		"query":                  query,
		"backend_type":           backendType,
		"leader_pid":             leaderPid,
		"worker_id":              workerID,
		"query_progress":         queryProgress,
		"rows_processed":         rowsProcessed,
		"total_rows_to_process":  totalRowsToProcess,
	}, nil
}

// queryPgStatActivity returns all rows from pg_stat_activity as maps.
func queryPgStatActivity(db *sql.DB) ([]map[string]interface{}, error) {