source-*: check rows.Err() after all rows.Next() loops

source-oracle-batch/discovery.go

@@ -239,6 +239,9 @@ func discoverTables(ctx context.Context, db *sql.DB, discoverSchemas []string) (
 			Name:  tableName,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered tables: %w", err)
+	}
 	return tables, nil
 }
 
@@ -337,6 +340,9 @@ func discoverColumns(ctx context.Context, db *sql.DB, discoverSchemas []string)
 		}
 		columns = append(columns, column)
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered columns: %w", err)
+	}
 	return columns, nil
 }
 

source-oracle/backfill.go

@@ -297,6 +297,9 @@ func (db *oracleDatabase) fetchBackfillRowIDRange(ctx context.Context, schemaNam
 
 		db.backfillRowIDRanges[streamID] = append(db.backfillRowIDRanges[streamID], rowidStart)
 	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("iterating rowid ranges: %w", err)
+	}
 
 	var row = db.conn.QueryRowContext(ctx, fmt.Sprintf(`SELECT MAX(ROWID) FROM "%s"."%s"`, schemaName, tableName))
 	if err != nil {
@@ -516,6 +519,10 @@ func (db *oracleDatabase) explainQuery(ctx context.Context, streamID sqlcapture.
 
 		outputLines = append(outputLines, outputLine)
 	}
+	if err := rows.Err(); err != nil {
+		logrus.WithFields(logrus.Fields{"id": streamID, "err": err}).Error("error iterating explain output")
+		return
+	}
 
 	logrus.WithFields(logrus.Fields{
 		"streamID": streamID,

source-oracle/prerequisites.go

@@ -145,6 +145,9 @@ func (db *oracleDatabase) prerequisiteTableAndColumnNameLengths(ctx context.Cont
 				}).Warn("column names longer than 30 characters are not supported for capturing, column will be ignored.")
 			}
 		}
+		if err := rows.Err(); err != nil {
+			return fmt.Errorf("iterating columns for table %q.%q: %w", owner, tableName, err)
+		}
 	}
 
 	return nil

source-postgres-batch/discovery.go

@@ -264,6 +264,9 @@ func discoverTables(ctx context.Context, db *sql.DB, discoverSchemas []string) (
 			Type:   tableType,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered tables: %w", err)
+	}
 	return tables, nil
 }
 
@@ -419,6 +422,9 @@ func discoverColumns(ctx context.Context, db *sql.DB, discoverSchemas []string)
 			DataType:   &dataType,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered columns: %w", err)
+	}
 	return columns, nil
 }
 
@@ -482,6 +488,9 @@ func discoverPrimaryKeys(ctx context.Context, db *sql.DB, discoverSchemas []stri
 			return nil, fmt.Errorf("primary key column %q (of table %q) appears out of order", columnName, tableID)
 		}
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered primary keys: %w", err)
+	}
 
 	var keys []*discoveredPrimaryKey
 	for _, key := range keysByTable {

source-postgres/database.go

@@ -59,6 +59,9 @@ func listPublishedTables(ctx context.Context, conn *pgxpool.Pool, publicationNam
 		}
 		publicationStatus[sqlcapture.JoinStreamID(schema, table)] = true
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error listing tables in publication %q: %w", publicationName, err)
+	}
 	return publicationStatus, nil
 }
 

source-postgres/discovery.go

@@ -798,6 +798,9 @@ func getGeneratedColumns(ctx context.Context, conn *pgxpool.Pool) (map[sqlcaptur
 		var streamID = sqlcapture.JoinStreamID(tableSchema, tableName)
 		generatedColumns[streamID] = append(generatedColumns[streamID], columnName)
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating generated columns: %w", err)
+	}
 	return generatedColumns, nil
 }
 

source-redshift-batch/discovery.go

@@ -304,6 +304,9 @@ func discoverTables(ctx context.Context, db *sql.DB, discoverSchemas []string) (
 			Type:   tableType,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered tables: %w", err)
+	}
 	return tables, nil
 }
 
@@ -464,6 +467,9 @@ func discoverColumns(ctx context.Context, db *sql.DB, discoverSchemas []string)
 			DataType:   &dataType,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered columns: %w", err)
+	}
 	return columns, nil
 }
 
@@ -523,6 +529,9 @@ func discoverPrimaryKeys(ctx context.Context, db *sql.DB, discoverSchemas []stri
 			Index:  ordinalPosition,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered primary keys: %w", err)
+	}
 	return keys, nil
 }
 

source-sqlserver-batch/discovery.go

@@ -253,6 +253,9 @@ func discoverTables(ctx context.Context, db *sql.DB, discoverSchemas []string) (
 			Type:   tableType,
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered tables: %w", err)
+	}
 	return tables, nil
 }
 
@@ -411,6 +414,9 @@ func discoverColumns(ctx context.Context, db *sql.DB, discoverSchemas []string)
 		}
 		columns = append(columns, column)
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered columns: %w", err)
+	}
 	return columns, nil
 }
 
@@ -478,6 +484,9 @@ func discoverPrimaryKeys(ctx context.Context, db *sql.DB, discoverSchemas []stri
 			return nil, fmt.Errorf("primary key column %q (of table %q) appears out of order", columnName, tableID)
 		}
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered primary keys: %w", err)
+	}
 
 	var keys []*discoveredPrimaryKey
 	for _, key := range keysByTable {

source-sqlserver-ct/diagnostics.go

@@ -42,6 +42,9 @@ func (db *sqlserverDatabase) ReplicationDiagnostics(ctx context.Context) error {
 			}
 			log.WithFields(logFields).Info("got diagnostic row")
 		}
+		if err := rows.Err(); err != nil {
+			logEntry.WithField("err", err).Error("error iterating diagnostic rows")
+		}
 		if numResults == 0 {
 			logEntry.Info("no results")
 		}

source-sqlserver-ct/discovery.go

@@ -173,6 +173,9 @@ func getTables(ctx context.Context, conn *sql.DB, opts discoveryOptions) ([]*sql
 			ExtraDetails: &sqlserverTableDiscoveryDetails{},
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered tables: %w", err)
+	}
 	return tables, nil
 }
 
@@ -246,6 +249,9 @@ func getColumns(ctx context.Context, conn *sql.DB, opts discoveryOptions) ([]sql
 		}
 		columns = append(columns, ci)
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered columns: %w", err)
+	}
 	return columns, nil
 }
 
@@ -305,6 +311,9 @@ func getPrimaryKeys(ctx context.Context, conn *sql.DB, opts discoveryOptions) (m
 			return nil, fmt.Errorf("primary key column %q (of table %q) appears out of order", columnName, streamID)
 		}
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered primary keys: %w", err)
+	}
 	return keys, nil
 }
 
@@ -342,6 +351,9 @@ func getComputedColumns(ctx context.Context, conn *sql.DB, opts discoveryOptions
 		var streamID = sqlcapture.JoinStreamID(tableSchema, tableName)
 		computedColumns[streamID] = append(computedColumns[streamID], columnName)
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered computed columns: %w", err)
+	}
 	return computedColumns, nil
 }
 

source-sqlserver/discovery.go

@@ -268,6 +268,9 @@ func getTables(ctx context.Context, conn *sql.DB, opts discoveryOptions) ([]*sql
 			ExtraDetails: &sqlserverTableDiscoveryDetails{},
 		})
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered tables: %w", err)
+	}
 	return tables, nil
 }
 
@@ -347,6 +350,9 @@ func getColumns(ctx context.Context, conn *sql.DB, opts discoveryOptions) ([]sql
 		}
 		columns = append(columns, ci)
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered columns: %w", err)
+	}
 	return columns, nil
 }
 
@@ -420,6 +426,9 @@ func getPrimaryKeys(ctx context.Context, conn *sql.DB, opts discoveryOptions) (m
 			return nil, fmt.Errorf("primary key column %q (of table %q) appears out of order", columnName, streamID)
 		}
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered primary keys: %w", err)
+	}
 	return keys, nil
 }
 
@@ -472,7 +481,10 @@ func getSecondaryIndexes(ctx context.Context, conn *sql.DB, opts discoveryOption
 			return nil, fmt.Errorf("internal error: secondary index key ordering failure: index %q on stream %q: column %q appears out of order", indexName, streamID, columnName)
 		}
 	}
-	return streamIndexColumns, err
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered secondary indexes: %w", err)
+	}
+	return streamIndexColumns, nil
 }
 
 func queryDiscoverComputedColumns(opts discoveryOptions) string {
@@ -509,6 +521,9 @@ func getComputedColumns(ctx context.Context, conn *sql.DB, opts discoveryOptions
 		var streamID = sqlcapture.JoinStreamID(tableSchema, tableName)
 		computedColumns[streamID] = append(computedColumns[streamID], columnName)
 	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating discovered computed columns: %w", err)
+	}
 	return computedColumns, nil
 }