This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 2b23a7a15 perf(table): bound scan task channel capacity (#2198)
2b23a7a15 is described below
commit 2b23a7a151b7860329346ae750637f49082af920
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 19 18:58:33 2026 +0200
perf(table): bound scan task channel capacity (#2198)
* perf(table): bound scan task channel to worker count
* test(table): cover bounded scan task dispatch and cancellation
* style(table): gofmt scan task channel test
---
table/arrow_scanner.go | 8 +++++++-
table/arrow_scanner_reorder_test.go | 23 +++++++++++++++++++++++
2 files changed, 30 insertions(+), 1 deletion(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 935a949eb..98556f2f4 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -2740,12 +2740,18 @@ func createIteratorWithCleanup(ctx context.Context,
numWorkers uint, records <-c
}
}
+// newScanTaskChannel buffers enough work to keep scan workers busy without
+// copying the entire task list into the channel for large scans.
+func newScanTaskChannel(taskCount, numWorkers int) chan
tblutils.Enumerated[FileScanTask] {
+ return make(chan tblutils.Enumerated[FileScanTask], min(taskCount,
2*numWorkers))
+}
+
func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context,
tasks []FileScanTask, positionDeleteLoader *lazyPositionDeleteLoader, dvLoader
*lazyDeletionVectorLoader, equalityDeleteLoader *lazyEqualityDeleteLoader,
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))
+ taskChan := newScanTaskChannel(len(tasks), numWorkers)
records := make(chan enumeratedRecord, numWorkers)
var wg sync.WaitGroup
diff --git a/table/arrow_scanner_reorder_test.go
b/table/arrow_scanner_reorder_test.go
index a754c759c..1cd5c90b9 100644
--- a/table/arrow_scanner_reorder_test.go
+++ b/table/arrow_scanner_reorder_test.go
@@ -218,6 +218,26 @@ func writeReorderDataFile(tb testing.TB, arrowSchema
*arrow.Schema, path string,
return bldr.Build()
}
+func TestScanTaskChannelCapacity(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ taskCount int
+ numWorkers int
+ wantCapacity int
+ }{
+ {name: "empty", taskCount: 0, numWorkers: 0, wantCapacity: 0},
+ {name: "single task", taskCount: 1, numWorkers: 1,
wantCapacity: 1},
+ {name: "fewer tasks than capacity", taskCount: 3, numWorkers:
3, wantCapacity: 3},
+ {name: "many tasks", taskCount: 100_000, numWorkers: 4,
wantCapacity: 8},
+ {name: "single worker", taskCount: 100_000, numWorkers: 1,
wantCapacity: 2},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ queue := newScanTaskChannel(tc.taskCount, tc.numWorkers)
+ require.Equal(t, tc.wantCapacity, cap(queue))
+ })
+ }
+}
+
func TestArrowScanReorderHeapBoundedWhileHeadTaskLags(t *testing.T) {
const (
files = 32
@@ -280,6 +300,9 @@ func
TestArrowScanReorderCancelWhileWorkersWaitForTaskCredits(t *testing.T) {
)
synctest.Test(t, func(t *testing.T) {
+ // With the first task gated, these tasks also fill and block
the producer.
+ require.Greater(t, files,
numWorkers+cap(newScanTaskChannel(files, numWorkers)))
+
dir := filepath.ToSlash(t.TempDir())
gateIO := &reorderGateIO{}
tbl, tasks := newReorderScanFixture(t, dir, files, rowsPerFile,
gateIO)
