zeroshade commented on code in PR #1938:
URL: https://github.com/apache/iceberg-go/pull/1938#discussion_r3883475169
##########
table/arrow_scanner.go:
##########
@@ -1769,59 +1890,81 @@ func createIterator(ctx context.Context, numWorkers
uint, records <-chan enumera
}
}
-func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context,
tasks []FileScanTask, deletesPerFile perFilePosDeletes, dvBitmaps
perFileDVBitmaps, eqDeleteSets map[int][]*equalityDeleteSet, invariants
*arrowScanInvariants) iter.Seq2[arrow.RecordBatch, error] {
- extSet := substrait.NewExtensionSet()
-
- ctx, cancel := context.WithCancelCause(exprs.WithExtensionIDSet(ctx,
extSet))
- taskChan := make(chan tblutils.Enumerated[FileScanTask], len(tasks))
-
- // numWorkers := 1
- numWorkers := min(as.concurrency, len(tasks))
- records := make(chan enumeratedRecord, numWorkers)
+func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context,
tasks []FileScanTask, positionDeleteLoader *lazyPositionDeleteLoader, dvBitmaps
perFileDVBitmaps, eqDeleteSets map[int][]*equalityDeleteSet, invariants
*arrowScanInvariants) iter.Seq2[arrow.RecordBatch, error] {
+ return func(yield func(arrow.RecordBatch, error) bool) {
+ extSet := substrait.NewExtensionSet()
+ scanCtx, cancel :=
context.WithCancelCause(exprs.WithExtensionIDSet(ctx, extSet))
+ numWorkers := min(as.concurrency, len(tasks))
+ taskChan := make(chan tblutils.Enumerated[FileScanTask],
len(tasks))
+ records := make(chan enumeratedRecord, numWorkers)
+
+ var wg sync.WaitGroup
+ wg.Add(numWorkers)
+ for range numWorkers {
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-scanCtx.Done():
+ return
+ case task, ok := <-taskChan:
+ if !ok {
+ return
+ }
+ if scanCtx.Err() != nil {
+ return
+ }
+
+ filePath :=
task.Value.File.FilePath()
+ var positionalDeletes
positionDeletes
+ if positionDeleteLoader != nil {
+ var err error
+ positionalDeletes, err
= positionDeleteLoader.load(scanCtx, task.Value)
+ if err != nil {
+ records <-
enumeratedRecord{Task: task, Err: err}
+ cancel(err)
+
+ return
+ }
+ }
+
+ if err :=
as.recordsFromTask(scanCtx, task, records,
+ positionalDeletes,
+ dvBitmaps[filePath],
+
eqDeleteSets[task.Index],
+ invariants); err != nil
{
+ cancel(err)
+
+ return
+ }
+ }
+ }
+ }()
+ }
- var wg sync.WaitGroup
- wg.Add(numWorkers)
- for range numWorkers {
go func() {
- defer wg.Done()
- for {
+ for i, t := range tasks {
select {
- case <-ctx.Done():
+ case <-scanCtx.Done():
Review Comment:
**[P1] Cancellation can deadlock iteration indefinitely.** This return
bypasses `close(taskChan)`, `wg.Wait()`, and `close(records)`.
`createIteratorWithCleanup` then cancels and drains the sequenced channel, but
`MakeSequencedChan` cannot close because its `records` source remains open. I
reproduced this with one valid task and an already-canceled context: iteration
failed to return within 200 ms on the first attempt. Please make producer
teardown unconditional (for example, defer closing `taskChan`, waiting for
workers, and closing `records`) and add pre-canceled and early-termination
regression tests.
##########
table/arrow_scanner.go:
##########
@@ -1769,59 +1890,81 @@ func createIterator(ctx context.Context, numWorkers
uint, records <-chan enumera
}
}
-func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context,
tasks []FileScanTask, deletesPerFile perFilePosDeletes, dvBitmaps
perFileDVBitmaps, eqDeleteSets map[int][]*equalityDeleteSet, invariants
*arrowScanInvariants) iter.Seq2[arrow.RecordBatch, error] {
- extSet := substrait.NewExtensionSet()
-
- ctx, cancel := context.WithCancelCause(exprs.WithExtensionIDSet(ctx,
extSet))
- taskChan := make(chan tblutils.Enumerated[FileScanTask], len(tasks))
-
- // numWorkers := 1
- numWorkers := min(as.concurrency, len(tasks))
- records := make(chan enumeratedRecord, numWorkers)
+func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context,
tasks []FileScanTask, positionDeleteLoader *lazyPositionDeleteLoader, dvBitmaps
perFileDVBitmaps, eqDeleteSets map[int][]*equalityDeleteSet, invariants
*arrowScanInvariants) iter.Seq2[arrow.RecordBatch, error] {
+ return func(yield func(arrow.RecordBatch, error) bool) {
+ extSet := substrait.NewExtensionSet()
+ scanCtx, cancel :=
context.WithCancelCause(exprs.WithExtensionIDSet(ctx, extSet))
+ numWorkers := min(as.concurrency, len(tasks))
+ taskChan := make(chan tblutils.Enumerated[FileScanTask],
len(tasks))
+ records := make(chan enumeratedRecord, numWorkers)
+
+ var wg sync.WaitGroup
+ wg.Add(numWorkers)
+ for range numWorkers {
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-scanCtx.Done():
+ return
+ case task, ok := <-taskChan:
+ if !ok {
+ return
+ }
+ if scanCtx.Err() != nil {
+ return
+ }
+
+ filePath :=
task.Value.File.FilePath()
+ var positionalDeletes
positionDeletes
+ if positionDeleteLoader != nil {
+ var err error
+ positionalDeletes, err
= positionDeleteLoader.load(scanCtx, task.Value)
+ if err != nil {
+ records <-
enumeratedRecord{Task: task, Err: err}
Review Comment:
**[P2] Deferred positional-delete errors can leak out-of-order Arrow
batches.** A later task can emit a batch while an earlier task is loading its
delete file. If the earlier load then fails here, the sequencer emits the error
but abandons batches still held in its priority queue when `records` closes;
iterator cleanup never sees those records to release them. A checked-allocator
probe with task 1 queued ahead of task 0’s error reproduced a 128-byte Arrow
leak. Please give the sequencer an error/close discard path that releases
queued record batches, with a deterministic two-worker regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
