Re-enable the rowserrcheck linter

Seems to work fine on Go 1.23.4 ¯\_(ツ)_/¯

Signed-off-by: Mihai Todor <todormihai@gmail.com>

.golangci.yml

@@ -89,6 +89,6 @@ linters:
     - tenv
     - predeclared
     - mirror
-    # - rowserrcheck
+    - rowserrcheck
     - bodyclose
     - nolintlint

internal/impl/mysql/input_mysql_stream.go

@@ -313,11 +313,12 @@ func (i *mysqlStreamInput) readSnapshot(ctx context.Context, snapshot *Snapshot)
 			return err
 		}
 		i.logger.Tracef("primary keys for table %s: %v", table, tablePks)
-		var numRowsProcessed int
 		lastSeenPksValues := map[string]any{}
 		for _, pk := range tablePks {
 			lastSeenPksValues[pk] = nil
 		}
+
+		var numRowsProcessed int
 		for {
 			var batchRows *sql.Rows
 			if numRowsProcessed == 0 {
@@ -326,21 +327,19 @@ func (i *mysqlStreamInput) readSnapshot(ctx context.Context, snapshot *Snapshot)
 				batchRows, err = snapshot.querySnapshotTable(ctx, table, tablePks, &lastSeenPksValues, i.fieldSnapshotMaxBatchSize)
 			}
 			if err != nil {
-				return err
+				return fmt.Errorf("failed to execute snapshot table query: %s", err)
 			}
 
 			types, err := batchRows.ColumnTypes()
 			if err != nil {
-				_ = batchRows.Close()
-				return err
+				return fmt.Errorf("failed to fetch column types: %s", err)
 			}
 
 			values, mappers := prepSnapshotScannerAndMappers(types)
 
 			columns, err := batchRows.Columns()
 			if err != nil {
-				_ = batchRows.Close()
-				return err
+				return fmt.Errorf("failed to fetch columns: %s", err)
 			}
 
 			var batchRowsCount int
@@ -349,7 +348,6 @@ func (i *mysqlStreamInput) readSnapshot(ctx context.Context, snapshot *Snapshot)
 				batchRowsCount++
 
 				if err := batchRows.Scan(values...); err != nil {
-					_ = batchRows.Close()
 					return err
 				}
 
@@ -376,6 +374,11 @@ func (i *mysqlStreamInput) readSnapshot(ctx context.Context, snapshot *Snapshot)
 					return ctx.Err()
 				}
 			}
+
+			if err := batchRows.Err(); err != nil {
+				return fmt.Errorf("failed to iterate snapshot table: %s", err)
+			}
+
 			// TODO(cdc): Save checkpoint
 			if batchRowsCount < i.fieldSnapshotMaxBatchSize {
 				break

internal/impl/mysql/snapshot.go

@@ -144,6 +144,10 @@ ORDER BY ORDINAL_POSITION;
 		pks = append(pks, pk)
 	}
 
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("failed to iterate table: %s", err)
+	}
+
 	if len(pks) == 0 {
 		return nil, fmt.Errorf("unable to find primary key for table %s - does the table exist and does it have a primary key set?", table)
 	}

internal/impl/timeplus/driver/driver.go

@@ -56,6 +56,10 @@ func (d *driver) Run(sql string) error {
 		return err
 	}
 
+	if err := rows.Err(); err != nil {
+		return err
+	}
+
 	columnTypes, err := rows.ColumnTypes()
 	if err != nil {
 		return err