Skip to content

Return connection errors from GetSelectedColumns and other pgx queries missing them #4657

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ilidemi merged 2 commits into from
Aug 5, 2026
Merged

Return connection errors from GetSelectedColumns and other pgx queries missing them #4657

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f42767c
Return connection errors from GetSelectedColumns
ilidemi Aug 4, 2026
269f6cc
CollectRows in places that don't handle errors
ilidemi Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 16 additions & 12 deletions flow/cmd/mirror_status.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -331,9 +331,8 @@ func (h *FlowRequestHandler) InitialLoadSummary(

defer rows.Close()

cloneStatuses := []*protos.CloneTableSummary{}
for rows.Next() {
if err := rows.Scan(
cloneStatuses, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (*protos.CloneTableSummary, error) {
if err := row.Scan(
&flowName,
&destinationTable,
&sourceTable,
Expand All @@ -345,7 +344,7 @@ func (h *FlowRequestHandler) InitialLoadSummary(
&numRowsSynced,
&avgTimePerPartitionMs,
); err != nil {
return nil, NewInternalApiError(fmt.Errorf("unable to scan initial load partition - %s: %w", parentMirrorName, err))
return nil, err
}

res := &protos.CloneTableSummary{
Expand Down Expand Up @@ -392,7 +391,10 @@ func (h *FlowRequestHandler) InitialLoadSummary(
res.AvgTimePerPartitionMs = int64(avgTimePerPartitionMs.Float64)
}

cloneStatuses = append(cloneStatuses, res)
return res, nil
})
if err != nil {
return nil, NewInternalApiError(fmt.Errorf("unable to scan initial load partition - %s: %w", parentMirrorName, err))
}
return &protos.InitialLoadSummaryResponse{
TableSummaries: cloneStatuses,
Expand Down Expand Up @@ -430,18 +432,15 @@ func (h *FlowRequestHandler) getPartitionStatuses(

defer rows.Close()

res := []*protos.PartitionStatus{}
var partitionId pgtype.Text
var startTime pgtype.Timestamp
var endTime pgtype.Timestamp
var numRowsInPartition pgtype.Int8
var numRowsSynced pgtype.Int8

for rows.Next() {
if err := rows.Scan(&partitionId, &startTime, &endTime, &numRowsInPartition, &numRowsSynced); err != nil {
slog.ErrorContext(ctx, "unable to scan qrep partition",
slog.String("flow", flowJobName), slog.Any("error", err))
return nil, fmt.Errorf("unable to scan qrep partition - %s: %w", flowJobName, err)
res, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (*protos.PartitionStatus, error) {
if err := row.Scan(&partitionId, &startTime, &endTime, &numRowsInPartition, &numRowsSynced); err != nil {
return nil, err
}

partitionStatus := &protos.PartitionStatus{}
Expand All @@ -462,7 +461,12 @@ func (h *FlowRequestHandler) getPartitionStatuses(
partitionStatus.RowsSynced = numRowsSynced.Int64
}

res = append(res, partitionStatus)
return partitionStatus, nil
})
if err != nil {
slog.ErrorContext(ctx, "unable to scan qrep partition",
slog.String("flow", flowJobName), slog.Any("error", err))
return nil, fmt.Errorf("unable to scan qrep partition - %s: %w", flowJobName, err)
}

return res, nil
Expand Down
12 changes: 5 additions & 7 deletions flow/connectors/postgres/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,8 +404,7 @@ func getSlotInfo(
return nil, fmt.Errorf("failed to read information for slots: %w", err)
}
defer rows.Close()
var slotInfoRows []*protos.SlotInfo
for rows.Next() {
return pgx.CollectRows(rows, func(row pgx.CollectableRow) (*protos.SlotInfo, error) {
var slotName pgtype.Text
var redoLSN pgtype.Text
var restartLSN pgtype.Text
Expand All @@ -427,7 +426,7 @@ func getSlotInfo(
var spillCount *int64
var spillBytes *int64

err := rows.Scan(
err := row.Scan(
&slotName,
&redoLSN,
&restartLSN,
Expand All @@ -453,7 +452,7 @@ func getSlotInfo(
return nil, err
}

slotInfoRows = append(slotInfoRows, &protos.SlotInfo{
return &protos.SlotInfo{
SlotName: slotName.String,
RedoLSN: redoLSN.String,
RestartLSN: restartLSN.String,
Expand All @@ -474,9 +473,8 @@ func getSlotInfo(
SpillTxns: spillTxns,
SpillCount: spillCount,
SpillBytes: spillBytes,
})
}
return slotInfoRows, nil
}, nil
})
}

// GetSlotInfo gets the information about the replication slot size and LSNs.
Expand Down
10 changes: 3 additions & 7 deletions flow/connectors/postgres/postgres_source.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -433,13 +433,9 @@ func (c *PostgresConnector) GetSelectedColumns(
return nil, fmt.Errorf("error getting selected columns for table %s: %w", sourceTable, err)
}

columns := make([]string, 0)
for rows.Next() {
var columnName string
if err := rows.Scan(&columnName); err != nil {
return nil, fmt.Errorf("error scanning column while getting selected columns: %w", err)
}
columns = append(columns, columnName)
columns, err := pgx.CollectRows(rows, pgx.RowTo[string])
if err != nil {
return nil, fmt.Errorf("error scanning columns while getting selected columns for table %s: %w", sourceTable, err)
}

return columns, nil
Expand Down
Loading