refactor: optimize row handling in mssql extractor and transformer
This commit is contained in:
@@ -68,7 +68,7 @@ func Consume(
|
|||||||
wgActivePartitions.Done()
|
wgActivePartitions.Done()
|
||||||
|
|
||||||
if rowsReadResult > 0 {
|
if rowsReadResult > 0 {
|
||||||
atomic.AddInt64(rowsRead, int64(rowsReadResult))
|
atomic.AddInt64(rowsRead, rowsReadResult)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -86,8 +86,6 @@ func Consume(
|
|||||||
case chErrorsOut <- custom_errors.JobError{ShouldCancelJob: false, Msg: err.Error(), Prev: err}:
|
case chErrorsOut <- custom_errors.JobError{ShouldCancelJob: false, Msg: err.Error(), Prev: err}:
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
continue
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -89,14 +89,9 @@ func (mssqlEx *MssqlExtractor) Exec(
|
|||||||
batchRows := make([]models.UnknownRowValues, 0, batchSize)
|
batchRows := make([]models.UnknownRowValues, 0, batchSize)
|
||||||
var rowsRead int64 = 0
|
var rowsRead int64 = 0
|
||||||
|
|
||||||
rowValues := make([]any, len(columns))
|
|
||||||
scanArgs := make([]any, len(columns))
|
|
||||||
for i := range rowValues {
|
|
||||||
scanArgs[i] = &rowValues[i]
|
|
||||||
}
|
|
||||||
|
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
if err := rows.Scan(scanArgs...); err != nil {
|
values, err := rows.Values()
|
||||||
|
if err != nil {
|
||||||
if len(batchRows) == 0 {
|
if len(batchRows) == 0 {
|
||||||
return rowsRead, err
|
return rowsRead, err
|
||||||
}
|
}
|
||||||
@@ -110,7 +105,7 @@ func (mssqlEx *MssqlExtractor) Exec(
|
|||||||
}
|
}
|
||||||
rowsRead++
|
rowsRead++
|
||||||
|
|
||||||
batchRows = append(batchRows, rowValues)
|
batchRows = append(batchRows, values)
|
||||||
if len(batchRows) >= batchSize {
|
if len(batchRows) >= batchSize {
|
||||||
if err := flush(ctx, &partition, batchSize, batchRows, chBatchesOut); err != nil {
|
if err := flush(ctx, &partition, batchSize, batchRows, chBatchesOut); err != nil {
|
||||||
return rowsRead, err
|
return rowsRead, err
|
||||||
|
|||||||
@@ -74,6 +74,10 @@ func (mssqlTr *MssqlTransformer) ProcessBatch(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if rowValues == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
for _, task := range transformationPlan {
|
for _, task := range transformationPlan {
|
||||||
val := rowValues[task.Index]
|
val := rowValues[task.Index]
|
||||||
if val == nil {
|
if val == nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user