This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 a2d559acf perf(table): parallelize scan task finalization (#2149)
a2d559acf is described below
commit a2d559acfb2215e1826b9f1401d8c5dfc54f9fb2
Author: Minh Vu <[email protected]>
AuthorDate: Wed Oct 21 22:51:14 2026 +0200
perf(table): parallelize scan task finalization (#2149)
---
table/scan_metrics.go | 9 +-
table/scan_task_finalization.go | 214 +++++++
.../scan_task_finalization_planfiles_bench_test.go | 196 +++++++
table/scan_task_finalization_test.go | 645 +++++++++++++++++++++
table/scanner.go | 54 +-
5 files changed, 1060 insertions(+), 58 deletions(-)
diff --git a/table/scan_metrics.go b/table/scan_metrics.go
index 94022576f..a541d548d 100644
--- a/table/scan_metrics.go
+++ b/table/scan_metrics.go
@@ -29,11 +29,10 @@ import (
"github.com/apache/iceberg-go/metrics"
)
-// scanMetricsAccumulator gathers scan-planning counts. Every field is written
-// from a single goroutine — the manifest counts in
-// fetchPartitionSpecFilteredManifestsWithSchema and the result/delete counts
in
-// planFilesLocal after its concurrent manifest/task barrier — so plain
-// integers are race-free; no field is touched from the concurrent workers.
+// scanMetricsAccumulator gathers scan-planning counts. The shared accumulator
+// is written from a single goroutine. Concurrent task-finalization workers use
+// independent local accumulators and merge them after their barrier, so these
+// fields remain plain integers without atomics.
//
// The scanned/skipped manifest counts measure partition-spec pruning (the
// filter applied while listing manifests). The per-entry skipped-data-files /
diff --git a/table/scan_task_finalization.go b/table/scan_task_finalization.go
new file mode 100644
index 000000000..1e51b55f6
--- /dev/null
+++ b/table/scan_task_finalization.go
@@ -0,0 +1,214 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "fmt"
+ "runtime"
+ "sync"
+
+ "github.com/apache/iceberg-go"
+)
+
+// Conservative worker threshold heuristic; the actual crossover depends on the
+// fraction of residual and split work. BenchmarkFinalizePlannedTasks sweeps
it.
+const scanTaskFinalizationMinTasksPerWorker = 256
+
+type scanTaskFinalizationResult struct {
+ metrics scanMetricsAccumulator
+ tasks []FileScanTask
+ expanded bool
+ err error
+}
+
+// finalizePlannedTasks applies residuals, records result metrics, and expands
+// Parquet byte-range splits. The caller transfers ownership of plannedTasks:
+// workers mutate residuals in that slice, even when an error is returned.
+// Callers must not reuse it after this call. If no task splits, the returned
+// slice aliases the original backing array.
+func (scan *Scan) finalizePlannedTasks(
+ plannedTasks []FileScanTask,
+ schema *iceberg.Schema,
+ acc *scanMetricsAccumulator,
+) ([]FileScanTask, error) {
+ var boundRowFilter iceberg.BooleanExpression
+ if scan.rowFilter != nil &&
!scan.rowFilter.Equals(iceberg.AlwaysTrue{}) {
+ var err error
+ boundRowFilter, err = iceberg.BindExpr(schema, scan.rowFilter,
scan.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ var residualEvaluators *keyDefaultMapErr[int,
*partitionResidualEvaluator]
+ if boundRowFilter != nil {
+ residualEvaluators = newKeyDefaultMapWrapErr(func(specID int)
(*partitionResidualEvaluator, error) {
+ return newPartitionResidualEvaluator(
+ schema,
scan.metadata.PartitionSpecByID(specID), boundRowFilter, scan.caseSensitive)
+ })
+ }
+
+ splitTargetSize := scan.metadata.Properties().GetInt64(
+ ReadSplitTargetSizeKey, ReadSplitTargetSizeDefault)
+ acc.resultDataFiles = int64(len(plannedTasks))
+ if len(plannedTasks) == 0 {
+ return []FileScanTask{}, nil
+ }
+
+ workerCount := min(max(scan.concurrency, 1), runtime.GOMAXPROCS(0),
+ max(1, len(plannedTasks)/scanTaskFinalizationMinTasksPerWorker))
+ results := make([]scanTaskFinalizationResult, workerCount)
+
+ if workerCount == 1 {
+ // Preserve the serial caller's ability to recover a panic and
inspect
+ // its stack. Only spawned goroutines need panic-to-error
conversion.
+ results[0] = scan.finalizeTaskRange(
+ plannedTasks, 0, len(plannedTasks), residualEvaluators,
splitTargetSize)
+ } else {
+ var wg sync.WaitGroup
+ chunkSize, remainder := len(plannedTasks)/workerCount,
len(plannedTasks)%workerCount
+ start := 0
+ for worker := range workerCount {
+ // Distribute the remainder instead of rounding every
range up:
+ // even large worker counts must stay within the input
slice.
+ end := start + chunkSize
+ if worker < remainder {
+ end++
+ }
+
+ wg.Add(1)
+ go func(worker, start, end int) {
+ defer wg.Done()
+ defer func() {
+ if recovered := recover(); recovered !=
nil {
+ results[worker].err =
fmt.Errorf("finalize scan tasks [%d:%d]: panic: %v",
+ start, end, recovered)
+ }
+ }()
+ results[worker] = scan.finalizeTaskRange(
+ plannedTasks, start, end,
residualEvaluators, splitTargetSize)
+ }(worker, start, end)
+ start = end
+ }
+ wg.Wait()
+ }
+
+ // Ranges follow task order, regardless of the order workers completed.
+ for i := range results {
+ if results[i].err != nil {
+ return nil, results[i].err
+ }
+ }
+
+ totalTasks := 0
+ anySplit := false
+ for i := range results {
+ mergeTaskFinalizationMetrics(acc, &results[i].metrics)
+ anySplit = anySplit || results[i].expanded
+ totalTasks += len(results[i].tasks)
+ }
+ if !anySplit {
+ return plannedTasks, nil
+ }
+ if workerCount == 1 {
+ return results[0].tasks, nil
+ }
+
+ // Workers that split files emit their ranges in task order. Unsplit
+ // workers reuse their input ranges without temporary replacement lists.
+ // Concatenate once; do not hold an allocation per individual split.
+ finalized := make([]FileScanTask, 0, totalTasks)
+ for i := range results {
+ finalized = append(finalized, results[i].tasks...)
+ }
+
+ return finalized, nil
+}
+
+func (scan *Scan) finalizeTaskRange(
+ plannedTasks []FileScanTask,
+ start, end int,
+ residualEvaluators *keyDefaultMapErr[int, *partitionResidualEvaluator],
+ splitTargetSize int64,
+) scanTaskFinalizationResult {
+ var result scanTaskFinalizationResult
+ var residualEvaluator *partitionResidualEvaluator
+ var cachedSpecID int
+ var hasCachedSpec bool
+ for index := start; index < end; index++ {
+ task := &plannedTasks[index]
+ if residualEvaluators != nil {
+ specID := int(task.File.SpecID())
+ var err error
+ if !hasCachedSpec || specID != cachedSpecID {
+ residualEvaluator, err =
residualEvaluators.Get(specID)
+ if err != nil {
+ result.err = fmt.Errorf(
+ "build partition residual
evaluator for spec %d: %w", specID, err)
+
+ return result
+ }
+ // Manifest-order runs normally share one spec.
Evaluators are
+ // immutable; a worker caches the last lookup
for a file run.
+ cachedSpecID, hasCachedSpec = specID, true
+ }
+ if residualEvaluator != nil {
+ var simplified bool
+ task.Residual, simplified, err =
residualEvaluator.residual(dataFilePartition(task.File))
+ if err != nil {
+ result.err = fmt.Errorf(
+ "evaluate partition residual
for %s: %w", task.File.FilePath(), err)
+
+ return result
+ }
+ if !simplified {
+ task.Residual = nil
+ }
+ }
+ }
+
+ result.metrics.addResultDeleteMetrics(*task)
+ result.metrics.totalFileSize += task.File.FileSizeBytes()
+ if splitTasks, split := splitParquetScanTask(*task,
splitTargetSize); split {
+ if !result.expanded {
+ result.expanded = true
+ // A mixed range keeps the unsplit prefix;
reserve extra space
+ // once to avoid repeated append growth when
more files split.
+ result.tasks = make([]FileScanTask, 0,
max((end-start)*2, index-start+len(splitTasks)))
+ result.tasks = append(result.tasks,
plannedTasks[start:index]...)
+ }
+ result.tasks = append(result.tasks, splitTasks...)
+ } else if result.expanded {
+ result.tasks = append(result.tasks, *task)
+ }
+ }
+ if !result.expanded {
+ result.tasks = plannedTasks[start:end]
+ }
+
+ return result
+}
+
+func mergeTaskFinalizationMetrics(acc, local *scanMetricsAccumulator) {
+ acc.totalFileSize += local.totalFileSize
+ acc.totalDeleteFileSize += local.totalDeleteFileSize
+ acc.positionalDeleteFiles += local.positionalDeleteFiles
+ acc.equalityDeleteFiles += local.equalityDeleteFiles
+ acc.dvs += local.dvs
+ acc.resultDeleteFiles = acc.positionalDeleteFiles +
acc.equalityDeleteFiles + acc.dvs
+}
diff --git a/table/scan_task_finalization_planfiles_bench_test.go
b/table/scan_task_finalization_planfiles_bench_test.go
new file mode 100644
index 000000000..c75178317
--- /dev/null
+++ b/table/scan_task_finalization_planfiles_bench_test.go
@@ -0,0 +1,196 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
+
+const (
+ planFilesFinalizationBenchmarkManifestCount = 8
+ planFilesFinalizationBenchmarkFilesPerManifest = 2_500
+)
+
+var planFilesFinalizationBenchmarkSink int
+
+// BenchmarkPlanFilesTaskFinalization exercises local planning over multiple
+// partition specs and partition values, with an optional mixture of split and
+// unsplit Parquet tasks. It uses in-memory FileIO and no delete manifests,
+// so this is not a full measure of remote IO or delete-index build cost.
+// Copy this fixture unchanged to the PR base worktree for before/after runs.
+func BenchmarkPlanFilesTaskFinalization(b *testing.B) {
+ const fileCount = planFilesFinalizationBenchmarkManifestCount *
planFilesFinalizationBenchmarkFilesPerManifest
+
+ // Four manifests are unpartitioned. Among the four identity-partitioned
+ // manifests, one in four files has id=7. In the split case one in four
+ // files expands into four tasks.
+ for _, tc := range []struct {
+ name string
+ rowFilter iceberg.BooleanExpression
+ splitOffsets bool
+ expectedTasks int
+ }{
+ {name: "plain", rowFilter: iceberg.AlwaysTrue{}, expectedTasks:
fileCount},
+ {name: "identity_residual", rowFilter:
iceberg.EqualTo(iceberg.Reference("id"), int32(7)), expectedTasks: fileCount *
5 / 8},
+ {name: "split_offsets", rowFilter: iceberg.AlwaysTrue{},
splitOffsets: true, expectedTasks: fileCount * 7 / 4},
+ } {
+ b.Run(tc.name, func(b *testing.B) {
+ tbl := newPlanFilesFinalizationBenchmarkTable(b,
tc.splitOffsets)
+ scan := tbl.Scan(WithMaxConcurrency(8),
WithRowFilter(tc.rowFilter))
+ b.ReportAllocs()
+ b.ReportMetric(float64(fileCount), "files/op")
+ for b.Loop() {
+ tasks, err :=
scan.PlanFiles(context.Background())
+ if err != nil {
+ b.Fatal(err)
+ }
+ if len(tasks) != tc.expectedTasks {
+ b.Fatalf("PlanFiles returned %d tasks
for %s, want %d", len(tasks), tc.name, tc.expectedTasks)
+ }
+ b.ReportMetric(float64(len(tasks)), "tasks/op")
+ planFilesFinalizationBenchmarkSink = len(tasks)
+ }
+ })
+ }
+}
+
+func newPlanFilesFinalizationBenchmarkTable(b testing.TB, splitOffsets bool)
*Table {
+ b.Helper()
+
+ const (
+ tableLocation = "mem://plan-files-finalization-benchmark"
+ sequenceNum = int64(1)
+ )
+ snapshotID := int64(1)
+
+ fs := newTrackingIO()
+ schema := simpleSchema()
+ spec := partitionedSpec()
+ properties := iceberg.Properties{}
+ if splitOffsets {
+ properties[ReadSplitTargetSizeKey] = "512"
+ }
+ meta, err := NewMetadata(schema, &spec, UnsortedSortOrder,
tableLocation, properties)
+ if err != nil {
+ b.Fatal(err)
+ }
+ builder, err := MetadataBuilderFromBase(meta, "")
+ if err != nil {
+ b.Fatal(err)
+ }
+ // Alternate the identity partition and unpartitioned specs to exercise
+ // residual evaluator lookup and cross-spec manifest planning.
+ unpartitioned := iceberg.NewPartitionSpecID(1)
+ requireNoErrorBenchmark(b, builder.AddPartitionSpec(&unpartitioned,
false))
+
+ manifests := make([]iceberg.ManifestFile,
planFilesFinalizationBenchmarkManifestCount)
+ for manifestIndex := range manifests {
+ entries := make([]iceberg.ManifestEntry,
planFilesFinalizationBenchmarkFilesPerManifest)
+ manifestSpec := spec
+ partition := map[int]any(nil)
+ if manifestIndex%2 == 1 {
+ manifestSpec = unpartitioned
+ }
+ for entryIndex := range entries {
+ dataPath := fmt.Sprintf("%s/data-%d-%d.parquet",
tableLocation, manifestIndex, entryIndex)
+ if manifestIndex%2 == 0 {
+ partition = map[int]any{1000: int32(6 +
entryIndex%4)}
+ }
+ fileBuilder, err := iceberg.NewDataFileBuilder(
+ manifestSpec,
+ iceberg.EntryContentData,
+ dataPath,
+ iceberg.ParquetFile,
+ partition,
+ nil,
+ nil,
+ 1,
+ 2048,
+ )
+ if err != nil {
+ b.Fatal(err)
+ }
+ if splitOffsets && entryIndex%4 == 0 {
+ fileBuilder.SplitOffsets([]int64{0, 512, 1024,
1536})
+ }
+ entries[entryIndex] = iceberg.NewManifestEntryBuilder(
+ iceberg.EntryStatusADDED,
+ &snapshotID,
+ fileBuilder.Build(),
+ ).SequenceNum(sequenceNum).Build()
+ }
+
+ manifestPath := fmt.Sprintf("%s/metadata/manifest-%d.avro",
tableLocation, manifestIndex)
+ var manifestBuffer bytes.Buffer
+ _, err := iceberg.WriteManifest(manifestPath, &manifestBuffer,
2, manifestSpec, schema, snapshotID, entries)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if err := fs.WriteFile(manifestPath, manifestBuffer.Bytes());
err != nil {
+ b.Fatal(err)
+ }
+ manifests[manifestIndex] = iceberg.NewManifestFile(
+ 2, manifestPath, int64(manifestBuffer.Len()),
int32(manifestSpec.ID()), snapshotID,
+ ).SequenceNum(sequenceNum, sequenceNum).
+
AddedFiles(int32(planFilesFinalizationBenchmarkFilesPerManifest)).
+
AddedRows(int64(planFilesFinalizationBenchmarkFilesPerManifest)).
+ Build()
+ }
+
+ manifestListPath := tableLocation + "/metadata/snap-1.avro"
+ var manifestListBuffer bytes.Buffer
+ requireNoErrorBenchmark(b, iceberg.WriteManifestList(
+ 2, &manifestListBuffer, snapshotID, nil, new(sequenceNum), 0,
manifests,
+ ))
+ requireNoErrorBenchmark(b, fs.WriteFile(manifestListPath,
manifestListBuffer.Bytes()))
+
+ schemaID := meta.CurrentSchema().ID
+ requireNoErrorBenchmark(b, builder.AddSnapshot(&Snapshot{
+ SnapshotID: snapshotID,
+ SequenceNumber: sequenceNum,
+ TimestampMs: meta.LastUpdatedMillis() + 1,
+ ManifestList: manifestListPath,
+ Summary: &Summary{Operation: OpAppend},
+ SchemaID: &schemaID,
+ }))
+ requireNoErrorBenchmark(b, builder.SetSnapshotRef(MainBranch,
snapshotID, BranchRef))
+ metadata, err := builder.Build()
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ return New(
+ Identifier{"db", "plan-files-finalization-benchmark"},
+ metadata,
+ tableLocation+"/metadata/metadata.json",
+ testFSF(fs),
+ nil,
+ )
+}
+
+func requireNoErrorBenchmark(b testing.TB, err error) {
+ b.Helper()
+ if err != nil {
+ b.Fatal(err)
+ }
+}
diff --git a/table/scan_task_finalization_test.go
b/table/scan_task_finalization_test.go
new file mode 100644
index 000000000..c7c264616
--- /dev/null
+++ b/table/scan_task_finalization_test.go
@@ -0,0 +1,645 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "fmt"
+ "runtime"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func setFinalizationTestProcs(t *testing.T, count int) {
+ t.Helper()
+ previous := runtime.GOMAXPROCS(count)
+ t.Cleanup(func() { runtime.GOMAXPROCS(previous) })
+}
+
+func TestFinalizePlannedTasksParallelMatchesSerial(t *testing.T) {
+ // Exercise the concurrent worker path even on single-core CI runners.
+ setFinalizationTestProcs(t, 4)
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64,
Required: true,
+ })
+ metadata, err := NewMetadata(
+ schema,
+ iceberg.UnpartitionedSpec,
+ UnsortedSortOrder,
+ "mem://table",
+ iceberg.Properties{ReadSplitTargetSizeKey: "25"},
+ )
+ require.NoError(t, err)
+
+ tasks := make([]FileScanTask, 1024)
+ for i := range tasks {
+ file := finalizationTestDataFile(
+ t, *iceberg.UnpartitionedSpec,
+ fmt.Sprintf("mem://table/data/file-%d.parquet", i), nil,
+ 100, []int64{8, 25, 50, 75})
+ tasks[i] = FileScanTask{File: file, Start: 0, Length:
file.FileSizeBytes()}
+ if i%3 == 0 {
+ tasks[i].DeleteFiles = []iceberg.DataFile{file}
+ }
+ if i%5 == 0 {
+ tasks[i].EqualityDeleteFiles = []iceberg.DataFile{file}
+ }
+ }
+
+ scan := &Scan{
+ metadata: metadata, rowFilter: iceberg.AlwaysTrue{},
caseSensitive: true, concurrency: 8,
+ }
+ serial, serialMetrics, err := serialFinalizationReference(
+ scan, append([]FileScanTask(nil), tasks...), schema)
+ require.NoError(t, err)
+ var parallelMetrics scanMetricsAccumulator
+ parallel, err := scan.finalizePlannedTasks(
+ append([]FileScanTask(nil), tasks...), schema, ¶llelMetrics)
+ require.NoError(t, err)
+
+ assert.Equal(t, serial, parallel)
+ assert.Equal(t, serialMetrics, parallelMetrics)
+ require.Len(t, parallel, len(tasks)*4)
+ for i := range tasks {
+ for split, start := range []int64{0, 25, 50, 75} {
+ task := parallel[i*4+split]
+ assert.Equal(t,
fmt.Sprintf("mem://table/data/file-%d.parquet", i), task.File.FilePath())
+ assert.Equal(t, start, task.Start)
+ assert.Equal(t, int64(25), task.Length)
+ }
+ }
+ assert.Equal(t, int64(len(tasks)), parallelMetrics.resultDataFiles)
+ assert.Equal(t, int64(len(tasks)*100), parallelMetrics.totalFileSize)
+ assert.Equal(t, int64(342), parallelMetrics.positionalDeleteFiles)
+ assert.Equal(t, int64(205), parallelMetrics.equalityDeleteFiles)
+ assert.Equal(t, int64(547), parallelMetrics.resultDeleteFiles)
+ assert.Equal(t, int64(54_700), parallelMetrics.totalDeleteFileSize)
+}
+
+func TestFinalizePlannedTasksParallelResidualsMatchSerial(t *testing.T) {
+ // Exercise the concurrent worker path even on single-core CI runners.
+ setFinalizationTestProcs(t, 4)
+ schema := simpleSchema()
+ spec := partitionedSpec()
+ unpartitioned := iceberg.NewPartitionSpecID(1)
+ metadata, err := NewMetadata(
+ schema,
+ &spec,
+ UnsortedSortOrder,
+ "mem://table",
+ nil,
+ )
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(metadata, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.AddPartitionSpec(&unpartitioned, false))
+ metadata, err = builder.Build()
+ require.NoError(t, err)
+
+ tasks := make([]FileScanTask, 1024)
+ for i := range tasks {
+ partitionValue := int32(7)
+ if i%2 != 0 {
+ partitionValue = 8
+ }
+ fileSpec := spec
+ if i%4 < 2 {
+ fileSpec = unpartitioned
+ }
+ file := finalizationTestDataFile(
+ t, fileSpec,
+ fmt.Sprintf("mem://table/data/file-%d.parquet", i),
+ map[int]any{1000: partitionValue}, 100, nil)
+ tasks[i] = FileScanTask{File: file, Start: 0, Length:
file.FileSizeBytes()}
+ }
+
+ filter := iceberg.EqualTo(iceberg.Reference("id"), int32(7))
+ scan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive:
true, concurrency: 8}
+ serial, serialMetrics, err := serialFinalizationReference(
+ scan, append([]FileScanTask(nil), tasks...), schema)
+ require.NoError(t, err)
+ var parallelMetrics scanMetricsAccumulator
+ parallel, err := scan.finalizePlannedTasks(
+ append([]FileScanTask(nil), tasks...), schema, ¶llelMetrics)
+ require.NoError(t, err)
+
+ assert.Equal(t, serial, parallel)
+ assert.Equal(t, serialMetrics, parallelMetrics)
+ for i, task := range parallel {
+ if i%4 < 2 {
+ assert.Nil(t, task.Residual, "task %d has no partition
projection", i)
+ } else if i%2 == 0 {
+ _, ok := task.Residual.(iceberg.AlwaysTrue)
+ assert.True(t, ok, "task %d should simplify to always
true", i)
+ } else {
+ _, ok := task.Residual.(iceberg.AlwaysFalse)
+ assert.True(t, ok, "task %d should simplify to always
false", i)
+ }
+ }
+}
+
+func TestFinalizePlannedTasksReturnsFirstErrorInTaskOrder(t *testing.T) {
+ schema := simpleSchema()
+ spec := partitionedSpec()
+ metadata, err := NewMetadata(schema, &spec, UnsortedSortOrder,
"mem://table", nil)
+ require.NoError(t, err)
+
+ tasks := make([]FileScanTask, 1024)
+ for i := range tasks {
+ var partitionValue any = int32(7)
+ if i == 255 || i == 256 {
+ // The earlier failing task is at the end of worker
range 0;
+ // worker 1 can fail first at the start of its own
range.
+ partitionValue = "invalid integer partition"
+ }
+ file := finalizationTestDataFile(t, spec,
+ fmt.Sprintf("mem://table/data/file-%d.parquet", i),
+ map[int]any{1000: partitionValue}, 100, nil)
+ tasks[i] = FileScanTask{File: file, Length:
file.FileSizeBytes()}
+ }
+
+ for _, concurrency := range []int{1, 8} {
+ t.Run(fmt.Sprintf("concurrency=%d", concurrency), func(t
*testing.T) {
+ scan := &Scan{
+ metadata: metadata, concurrency: concurrency,
caseSensitive: true,
+ rowFilter:
iceberg.EqualTo(iceberg.Reference("id"), int32(7)),
+ }
+ input := append([]FileScanTask(nil), tasks...)
+ if concurrency > 1 {
+ // With four CPU workers and 1024 tasks, worker
0 owns
+ // [0:256] and worker 1 owns [256:512].
+ previous := runtime.GOMAXPROCS(4)
+ defer runtime.GOMAXPROCS(previous)
+
+ laterErrorReached := make(chan struct{})
+ input[255].File = finalizationOrderFile{
+ DataFile: input[255].File, wait:
laterErrorReached,
+ }
+ input[256].File = finalizationOrderFile{
+ DataFile: input[256].File, signal:
laterErrorReached, signalOnce: new(sync.Once),
+ }
+ }
+ result, planErr := scan.finalizePlannedTasks(input,
schema, &scanMetricsAccumulator{})
+ require.ErrorContains(t, planErr, "evaluate partition
residual for mem://table/data/file-255.parquet")
+ assert.Nil(t, result)
+ })
+ }
+}
+
+// Controls which worker reaches its partition error first, without relying on
+// scheduler timing. The wrapper intentionally does not forward the built-in
+// borrowed-partition interface, so residual evaluation calls Partition.
+type finalizationOrderFile struct {
+ iceberg.DataFile
+ wait <-chan struct{}
+ signal chan struct{}
+ signalOnce *sync.Once
+}
+
+func (f finalizationOrderFile) Partition() map[int]any {
+ if f.signal != nil {
+ f.signalOnce.Do(func() { close(f.signal) })
+ }
+ if f.wait != nil {
+ select {
+ case <-f.wait:
+ case <-time.After(5 * time.Second):
+ panic("timed out waiting for later worker to reach its
error")
+ }
+ }
+
+ return f.DataFile.Partition()
+}
+
+func TestFinalizePlannedTasksUnsplitReusesInput(t *testing.T) {
+ // Exercise the concurrent worker path even on single-core CI runners.
+ setFinalizationTestProcs(t, 4)
+ schema := simpleSchema()
+ metadata, err := NewMetadata(
+ schema,
+ iceberg.UnpartitionedSpec,
+ UnsortedSortOrder,
+ "mem://table",
+ nil,
+ )
+ require.NoError(t, err)
+ file := finalizationTestDataFile(
+ t, *iceberg.UnpartitionedSpec, "mem://table/data/file.parquet",
nil, 100, nil)
+ tasks := make([]FileScanTask, 512)
+ for i := range tasks {
+ tasks[i] = FileScanTask{File: file, Start: 0, Length:
file.FileSizeBytes()}
+ }
+ scan := &Scan{metadata: metadata, rowFilter: iceberg.AlwaysTrue{},
caseSensitive: true, concurrency: 8}
+
+ got, err := scan.finalizePlannedTasks(tasks, schema,
&scanMetricsAccumulator{})
+ require.NoError(t, err)
+ require.Len(t, got, len(tasks))
+ assert.Same(t, &tasks[0], &got[0])
+ assert.Same(t, &tasks[len(tasks)-1], &got[len(got)-1])
+ assert.Nil(t, got[0].Residual)
+}
+
+func TestFinalizePlannedTasksManyWorkersCoverEveryTask(t *testing.T) {
+ // A rounded-up chunk size can put the last worker past the input when
+ // there are more workers than tasks per worker and a short remainder.
+ const workers = 300
+ previous := runtime.GOMAXPROCS(workers)
+ defer runtime.GOMAXPROCS(previous)
+
+ schema := simpleSchema()
+ metadata, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder,
+ "mem://table", iceberg.Properties{ReadSplitTargetSizeKey: "25"})
+ require.NoError(t, err)
+ file := finalizationTestDataFile(t, *iceberg.UnpartitionedSpec,
+ "mem://table/data/unsplit.parquet", nil, 100, nil)
+ splitFile := finalizationTestDataFile(t, *iceberg.UnpartitionedSpec,
+ "mem://table/data/split.parquet", nil, 100, []int64{8, 25, 50,
75})
+ lastFile := finalizationTestDataFile(t, *iceberg.UnpartitionedSpec,
+ "mem://table/data/last.parquet", nil, 100, nil)
+ tasks := make([]FileScanTask,
workers*scanTaskFinalizationMinTasksPerWorker+1)
+ firstRowIDs := make([]int64, len(tasks))
+ for i := range tasks {
+ firstRowIDs[i] = int64(i * 100)
+ tasks[i] = FileScanTask{File: file, Length: 100, FirstRowID:
&firstRowIDs[i]}
+ }
+ tasks[0].File = splitFile
+ tasks[len(tasks)-1].File = lastFile
+ scan := &Scan{metadata: metadata, rowFilter: iceberg.AlwaysTrue{},
concurrency: workers}
+
+ var acc scanMetricsAccumulator
+ got, err := scan.finalizePlannedTasks(tasks, schema, &acc)
+ require.NoError(t, err)
+ require.Len(t, got, len(tasks)+3)
+ for i := range 4 {
+ assert.Equal(t, splitFile.FilePath(), got[i].File.FilePath())
+ assert.Equal(t, int64(i*25), got[i].Start)
+ assert.Equal(t, int64(25), got[i].Length)
+ assert.Same(t, tasks[0].FirstRowID, got[i].FirstRowID)
+ }
+ for i, task := range got[4 : len(got)-1] {
+ if task.File != file || task.Start != 0 || task.Length != 100
|| task.FirstRowID != tasks[i+1].FirstRowID {
+ t.Fatalf("unsplit task %d changed or moved: %+v", i+1,
task)
+ }
+ }
+ assert.Equal(t, lastFile.FilePath(), got[len(got)-1].File.FilePath())
+ assert.Same(t, tasks[len(tasks)-1].FirstRowID,
got[len(got)-1].FirstRowID)
+ assert.Equal(t, int64(len(tasks)), acc.resultDataFiles)
+ assert.Equal(t, int64(len(tasks)*100), acc.totalFileSize)
+}
+
+// serialFinalizationReference reproduces the pre-parallelization loop. It is
+// deliberately independent of finalizeTaskRange and the worker merge logic.
+func serialFinalizationReference(
+ scan *Scan,
+ tasks []FileScanTask,
+ schema *iceberg.Schema,
+) ([]FileScanTask, scanMetricsAccumulator, error) {
+ var acc scanMetricsAccumulator
+ var filter iceberg.BooleanExpression
+ if scan.rowFilter != nil &&
!scan.rowFilter.Equals(iceberg.AlwaysTrue{}) {
+ var err error
+ filter, err = iceberg.BindExpr(schema, scan.rowFilter,
scan.caseSensitive)
+ if err != nil {
+ return nil, acc, err
+ }
+ }
+ evaluators := make(map[int]*partitionResidualEvaluator)
+ acc.resultDataFiles = int64(len(tasks))
+ result := make([]FileScanTask, 0, len(tasks))
+ target := scan.metadata.Properties().GetInt64(ReadSplitTargetSizeKey,
ReadSplitTargetSizeDefault)
+ for _, task := range tasks {
+ if filter != nil {
+ specID := int(task.File.SpecID())
+ evaluator, ok := evaluators[specID]
+ if !ok {
+ var err error
+ evaluator, err = newPartitionResidualEvaluator(
+ schema,
scan.metadata.PartitionSpecByID(specID), filter, scan.caseSensitive)
+ if err != nil {
+ return nil, acc, fmt.Errorf("build
partition residual evaluator for spec %d: %w", specID, err)
+ }
+ evaluators[specID] = evaluator
+ }
+ if evaluator != nil {
+ var err error
+ var simplified bool
+ task.Residual, simplified, err =
evaluator.residual(dataFilePartition(task.File))
+ if err != nil {
+ return nil, acc, fmt.Errorf(
+ "evaluate partition residual
for %s: %w", task.File.FilePath(), err)
+ }
+ if !simplified {
+ task.Residual = nil
+ }
+ }
+ }
+ acc.addResultDeleteMetrics(task)
+ acc.totalFileSize += task.File.FileSizeBytes()
+ if splits, ok := splitParquetScanTask(task, target); ok {
+ result = append(result, splits...)
+ } else {
+ result = append(result, task)
+ }
+ }
+
+ return result, acc, nil
+}
+
+func TestFinalizePlannedTasksMixedSplitsAgainstIndependentReference(t
*testing.T) {
+ // Exercise the concurrent worker path even on single-core CI runners.
+ setFinalizationTestProcs(t, 4)
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64,
Required: true,
+ })
+ metadata, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder,
+ "mem://table", iceberg.Properties{ReadSplitTargetSizeKey: "25"})
+ require.NoError(t, err)
+
+ dvBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ "mem://table/deletes.dv.puffin", iceberg.PuffinFile, nil, nil,
nil, 1, 2048)
+ require.NoError(t, err)
+ dv := dvBuilder.ContentSizeInBytes(17).ContentOffset(100).Build()
+
+ for _, count := range []int{0, 1, 255, 256, 257, 512, 1024} {
+ t.Run(fmt.Sprintf("files=%d", count), func(t *testing.T) {
+ tasks := make([]FileScanTask, count)
+ splits := 0
+ for i := range tasks {
+ offsets := []int64(nil)
+ if i%7 < 2 || i == count-1 || i == 255 || i ==
256 {
+ offsets = []int64{8, 25, 50, 75}
+ splits++
+ }
+ file := finalizationTestDataFile(t,
*iceberg.UnpartitionedSpec,
+
fmt.Sprintf("mem://table/data/file-%d.parquet", i), nil, 100, offsets)
+ tasks[i] = FileScanTask{File: file, Length: 100}
+ if i%9 == 0 {
+ tasks[i].DeleteFiles =
[]iceberg.DataFile{file}
+ }
+ if i%11 == 0 {
+ tasks[i].EqualityDeleteFiles =
[]iceberg.DataFile{file}
+ }
+ if i%13 == 0 {
+ tasks[i].DeletionVectorFiles =
[]iceberg.DataFile{dv}
+ }
+ }
+ scan := &Scan{
+ metadata: metadata, rowFilter:
iceberg.AlwaysTrue{},
+ caseSensitive: true, concurrency: 8,
+ }
+ expected, expectedMetrics, err :=
serialFinalizationReference(
+ scan, append([]FileScanTask(nil), tasks...),
schema)
+ require.NoError(t, err)
+ var actualMetrics scanMetricsAccumulator
+ actual, err := scan.finalizePlannedTasks(
+ append([]FileScanTask(nil), tasks...), schema,
&actualMetrics)
+ require.NoError(t, err)
+ assert.Equal(t, expected, actual)
+ assert.Equal(t, expectedMetrics, actualMetrics)
+ assert.Len(t, actual, count+3*splits)
+ assert.Equal(t, int64((count+12)/13), actualMetrics.dvs)
+ assert.Equal(t,
int64((count+8)/9*100+(count+10)/11*100+(count+12)/13*17),
+ actualMetrics.totalDeleteFileSize)
+ })
+ }
+}
+
+type panicSizeFinalizationFile struct {
+ iceberg.DataFile
+}
+
+func (panicSizeFinalizationFile) FileSizeBytes() int64 {
+ panic("injected file size panic")
+}
+
+func TestFinalizePlannedTasksWorkerPanicReturnsError(t *testing.T) {
+ // Exercise the concurrent worker path even on single-core CI runners.
+ setFinalizationTestProcs(t, 4)
+ schema := simpleSchema()
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, "mem://table", nil)
+ require.NoError(t, err)
+ file := finalizationTestDataFile(t, *iceberg.UnpartitionedSpec,
+ "mem://table/data/file.parquet", nil, 100, nil)
+ tasks := make([]FileScanTask, 512)
+ for i := range tasks {
+ tasks[i] = FileScanTask{File: file, Length: 100}
+ }
+ tasks[256].File = panicSizeFinalizationFile{DataFile: file}
+
+ scan := &Scan{metadata: meta, rowFilter: iceberg.AlwaysTrue{},
caseSensitive: true, concurrency: 8}
+ actual, err := scan.finalizePlannedTasks(tasks, schema,
&scanMetricsAccumulator{})
+ require.ErrorContains(t, err, "injected file size panic")
+ require.Nil(t, actual)
+}
+
+// A one-worker scan keeps the original caller-goroutine panic behavior.
+func TestFinalizePlannedTasksSingleWorkerPanicPropagates(t *testing.T) {
+ schema := simpleSchema()
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, "mem://table", nil)
+ require.NoError(t, err)
+ file := finalizationTestDataFile(t, *iceberg.UnpartitionedSpec,
+ "mem://table/data/file.parquet", nil, 100, nil)
+ tasks := []FileScanTask{{File: panicSizeFinalizationFile{DataFile:
file}, Length: 100}}
+ scan := &Scan{metadata: meta, rowFilter: iceberg.AlwaysTrue{},
caseSensitive: true, concurrency: 1}
+ assert.PanicsWithValue(t, "injected file size panic", func() {
+ _, _ = scan.finalizePlannedTasks(tasks, schema,
&scanMetricsAccumulator{})
+ })
+}
+
+func finalizationTestDataFile(
+ t testing.TB,
+ spec iceberg.PartitionSpec,
+ path string,
+ partition map[int]any,
+ size int64,
+ offsets []int64,
+) iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ spec,
+ iceberg.EntryContentData,
+ path,
+ iceberg.ParquetFile,
+ partition,
+ nil,
+ nil,
+ 100,
+ size,
+ )
+ require.NoError(t, err)
+ if offsets != nil {
+ builder.SplitOffsets(offsets)
+ }
+
+ return builder.Build()
+}
+
+var scanTaskFinalizationBenchmarkSink int
+
+// Compare the independent old serial loop with the new finalization after
+// identical per-iteration input clones. Smaller sizes exercise the
conservative
+// worker threshold; the large sizes measure throughput once workers amortize.
+// Mixed work covers variable partitions, partial splitting, positional
deletes,
+// and deletion vectors, without depending on manifest IO.
+func BenchmarkFinalizePlannedTasks(b *testing.B) {
+ schema := simpleSchema()
+ spec := partitionedSpec()
+
+ for _, tc := range []struct {
+ name string
+ rowFilter iceberg.BooleanExpression
+ offsets []int64
+ targetSize string
+ fileCount int
+ mixedWork bool
+ withDeletes bool
+ }{
+ {
+ name: "plain",
+ rowFilter: iceberg.AlwaysTrue{},
+ fileCount: 100_000,
+ },
+ {
+ name: "identity_residual",
+ rowFilter: iceberg.EqualTo(iceberg.Reference("id"),
int32(7)),
+ fileCount: 100_000,
+ },
+ {
+ name: "split_offsets",
+ rowFilter: iceberg.AlwaysTrue{},
+ offsets: benchmarkFinalizationOffsets(32, 32),
+ targetSize: "128",
+ fileCount: 20_000,
+ },
+ {
+ name: "mixed_split_deletes",
+ rowFilter: iceberg.EqualTo(iceberg.Reference("id"),
int32(7)),
+ offsets: benchmarkFinalizationOffsets(32, 32),
+ targetSize: "128",
+ fileCount: 20_000,
+ mixedWork: true,
+ withDeletes: true,
+ },
+ } {
+ b.Run(tc.name, func(b *testing.B) {
+ properties := iceberg.Properties{}
+ if tc.targetSize != "" {
+ properties[ReadSplitTargetSizeKey] =
tc.targetSize
+ }
+ metadata, err := NewMetadata(schema, &spec,
UnsortedSortOrder, "mem://benchmark", properties)
+ if err != nil {
+ b.Fatal(err)
+ }
+ var posDelete, dvDelete iceberg.DataFile
+ if tc.withDeletes {
+ posBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes,
"mem://benchmark/deletes/pos.parquet",
+ iceberg.ParquetFile, map[int]any{1000:
int32(7)}, nil, nil, 1, 128)
+ if err != nil {
+ b.Fatal(err)
+ }
+ posDelete = posBuilder.Build()
+ dvBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes,
"mem://benchmark/deletes/dv.puffin",
+ iceberg.PuffinFile, map[int]any{1000:
int32(7)}, nil, nil, 1, 256)
+ if err != nil {
+ b.Fatal(err)
+ }
+ dvDelete =
dvBuilder.ContentOffset(64).ContentSizeInBytes(17).Build()
+ }
+ tasks := make([]FileScanTask, tc.fileCount)
+ for i := range tasks {
+ offsets := tc.offsets
+ if tc.mixedWork && i%3 != 0 {
+ offsets = nil
+ }
+ file := finalizationTestDataFile(
+ b,
+ spec,
+
fmt.Sprintf("mem://benchmark/data/file-%d.parquet", i),
+ map[int]any{1000: int32(6 + i%4)},
+ 2048,
+ offsets,
+ )
+ tasks[i] = FileScanTask{File: file, Start: 0,
Length: file.FileSizeBytes()}
+ if tc.withDeletes && i%10 == 0 {
+ tasks[i].DeleteFiles =
[]iceberg.DataFile{posDelete}
+ }
+ if tc.withDeletes && i%13 == 0 {
+ tasks[i].DeletionVectorFiles =
[]iceberg.DataFile{dvDelete}
+ }
+ }
+
+ for _, fileCount := range []int{512, 1_024, 2_048,
4_096, tc.fileCount} {
+ b.Run(fmt.Sprintf("files=%d", fileCount),
func(b *testing.B) {
+ benchTasks := tasks[:fileCount]
+ b.Run("old_serial_loop", func(b
*testing.B) {
+ scan := &Scan{
+ metadata: metadata,
rowFilter: tc.rowFilter, caseSensitive: true, concurrency: 1,
+ }
+ b.ReportAllocs()
+
b.ReportMetric(float64(fileCount), "files/op")
+ for b.Loop() {
+ input :=
append([]FileScanTask(nil), benchTasks...)
+ result, _, err :=
serialFinalizationReference(scan, input, schema)
+ if err != nil {
+ b.Fatal(err)
+ }
+
scanTaskFinalizationBenchmarkSink += len(result)
+ }
+ })
+
+ for _, concurrency := range []int{1, 2,
4, 8} {
+
b.Run(fmt.Sprintf("concurrency=%d", concurrency), func(b *testing.B) {
+ scan := &Scan{
+ metadata:
metadata, rowFilter: tc.rowFilter,
+ caseSensitive:
true, concurrency: concurrency,
+ }
+ b.ReportAllocs()
+
b.ReportMetric(float64(fileCount), "files/op")
+ for b.Loop() {
+ input :=
append([]FileScanTask(nil), benchTasks...)
+ var acc
scanMetricsAccumulator
+ finalized, err
:= scan.finalizePlannedTasks(input, schema, &acc)
+ if err != nil {
+
b.Fatal(err)
+ }
+
scanTaskFinalizationBenchmarkSink += len(finalized)
+ }
+ })
+ }
+ })
+ }
+ })
+ }
+}
+
+func benchmarkFinalizationOffsets(count int, spacing int64) []int64 {
+ offsets := make([]int64, count)
+ for i := range offsets {
+ offsets[i] = int64(i+1) * spacing
+ }
+
+ return offsets
+}
diff --git a/table/scanner.go b/table/scanner.go
index b3c98e041..2bef72fed 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -1796,59 +1796,7 @@ func (scan *Scan) planFilesLocal(
eqDeleteIndex = nil
}
- var boundRowFilter iceberg.BooleanExpression
- if scan.rowFilter != nil &&
!scan.rowFilter.Equals(iceberg.AlwaysTrue{}) {
- boundRowFilter, err = iceberg.BindExpr(schema, scan.rowFilter,
scan.caseSensitive)
- if err != nil {
- return nil, err
- }
- }
- var residualEvaluators map[int]*partitionResidualEvaluator
- if boundRowFilter != nil {
- residualEvaluators = make(map[int]*partitionResidualEvaluator)
- }
-
- // Apply residuals, metrics, and parquet range splitting in manifest
order.
- // These operations stay outside the concurrent manifest workers so the
- // residual evaluator cache and scan metrics remain race-free.
- results = make([]FileScanTask, 0, len(plannedTasks))
- splitTargetSize := scan.metadata.Properties().GetInt64(
- ReadSplitTargetSizeKey, ReadSplitTargetSizeDefault)
- acc.resultDataFiles = int64(len(plannedTasks))
- for _, task := range plannedTasks {
- if boundRowFilter != nil {
- specID := int(task.File.SpecID())
- residualEvaluator, found := residualEvaluators[specID]
- if !found {
- residualEvaluator, err =
newPartitionResidualEvaluator(
- schema,
scan.metadata.PartitionSpecByID(specID), boundRowFilter, scan.caseSensitive)
- if err != nil {
- return nil, fmt.Errorf("build partition
residual evaluator for spec %d: %w", specID, err)
- }
- residualEvaluators[specID] = residualEvaluator
- }
- if residualEvaluator != nil {
- var simplified bool
- task.Residual, simplified, err =
residualEvaluator.residual(dataFilePartition(task.File))
- if err != nil {
- return nil, fmt.Errorf("evaluate
partition residual for %s: %w", task.File.FilePath(), err)
- }
- if !simplified {
- task.Residual = nil
- }
- }
- }
-
- acc.addResultDeleteMetrics(task)
- acc.totalFileSize += task.File.FileSizeBytes()
- if splitTasks, split := splitParquetScanTask(task,
splitTargetSize); split {
- results = append(results, splitTasks...)
- } else {
- results = append(results, task)
- }
- }
-
- return results, nil
+ return scan.finalizePlannedTasks(plannedTasks, schema, acc)
}
func dataFilesWithoutColumnStats(
