kylesoda/go-migrate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: enhance error handling in ProcessPartition by tracking last row values

Browse Source
This commit is contained in:
Kylesoda
2026-05-11 10:02:58 -05:00
parent 604702ef43
commit 6f2e3e28f1
1 changed files with 10 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
internal/app/etl/extractors/process.go
View File

@@ -76,6 +76,7 @@ func (ex *GenericExtractor) ProcessPartition(
batchRows := make([]models.UnknownRowValues, 0, batchSize) batchRows := make([]models.UnknownRowValues, 0, batchSize)
var rowsRead int64 = 0 var rowsRead int64 = 0
var lastRow models.UnknownRowValues
for rows.Next() { for rows.Next() {
rowValues := make([]any, len(columns)) rowValues := make([]any, len(columns))
@@ -98,6 +99,7 @@ func (ex *GenericExtractor) ProcessPartition(
return rowsRead, errorFromLastPartitionRow(lastRow, indexPrimaryKey, partition, err) return rowsRead, errorFromLastPartitionRow(lastRow, indexPrimaryKey, partition, err)
} }
rowsRead++ rowsRead++
lastRow = rowValues
batchRows = append(batchRows, rowValues) batchRows = append(batchRows, rowValues)
if len(batchRows) >= batchSize { if len(batchRows) >= batchSize {
@@ -114,5 +116,12 @@ func (ex *GenericExtractor) ProcessPartition(
return rowsRead, err return rowsRead, err
} }
return rowsRead, rows.Err() if err := rows.Err(); err != nil {
if lastRow != nil {
return rowsRead, errorFromLastPartitionRow(lastRow, indexPrimaryKey, partition, err)
}
return rowsRead, err
}
return rowsRead, nil
} }