laskoviymishka commented on code in PR #2149: URL: https://github.com/apache/iceberg-go/pull/2149#discussion_r4234106982
########## table/scan_task_finalization.go: ########## @@ -0,0 +1,199 @@ +// 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" + "sync" + + "github.com/apache/iceberg-go" +) + +// Keep enough work in each finalization worker to amortize goroutine and +// result-merging overhead on small scans. +const scanTaskFinalizationMinTasksPerWorker = 64 + +type scanTaskSplitReplacement struct { + index int + tasks []FileScanTask +} + +type scanTaskFinalizationResult struct { + metrics scanMetricsAccumulator + replacements []scanTaskSplitReplacement + err error +} + +// finalizePlannedTasks applies task residuals, records result metrics, and +// expands Parquet byte-range splits. Workers own disjoint task ranges, so base +// tasks can be updated in place. Only tasks that actually split need replacement +// slices; unsplit scans return plannedTasks directly without another O(files) +// copy. +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), max(1, len(plannedTasks)/scanTaskFinalizationMinTasksPerWorker)) Review Comment: `scan.concurrency` is the manifest-IO fan-out knob and can easily be 32 or more, but this work is CPU-bound, so with no `GOMAXPROCS` bound we can spawn far more CPU-bound goroutines than cores. The benchmark already shows the cost: `identity_residual` is 11.2ms at 4 workers and 14.2ms at 8, so 8 is slower. I'd clamp the worker count to `runtime.GOMAXPROCS(0)`, and given that regression raise the per-worker floor well above 64. ########## table/scan_task_finalization.go: ########## @@ -0,0 +1,199 @@ +// 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" + "sync" + + "github.com/apache/iceberg-go" +) + +// Keep enough work in each finalization worker to amortize goroutine and +// result-merging overhead on small scans. +const scanTaskFinalizationMinTasksPerWorker = 64 + +type scanTaskSplitReplacement struct { + index int + tasks []FileScanTask +} + +type scanTaskFinalizationResult struct { + metrics scanMetricsAccumulator + replacements []scanTaskSplitReplacement + err error +} + +// finalizePlannedTasks applies task residuals, records result metrics, and +// expands Parquet byte-range splits. Workers own disjoint task ranges, so base +// tasks can be updated in place. Only tasks that actually split need replacement Review Comment: The doc says tasks "can be updated in place" but not that `finalizePlannedTasks` consumes the slice. It writes `Residual` into the caller's array, including on the error path where it leaves the slice half-updated, and the unsplit path returns that same backing array. That's safe today because `planFilesLocal` owns the slice, but this reads like a general helper on `Scan`, and the incremental and changelog planners use similar shapes. I'd spell the ownership out: callers hand off `plannedTasks` and must not reuse it after the call, and the returned slice may alias it. ########## table/scan_task_finalization.go: ########## @@ -0,0 +1,199 @@ +// 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" + "sync" + + "github.com/apache/iceberg-go" +) + +// Keep enough work in each finalization worker to amortize goroutine and +// result-merging overhead on small scans. +const scanTaskFinalizationMinTasksPerWorker = 64 + +type scanTaskSplitReplacement struct { + index int + tasks []FileScanTask +} + +type scanTaskFinalizationResult struct { + metrics scanMetricsAccumulator + replacements []scanTaskSplitReplacement + err error +} + +// finalizePlannedTasks applies task residuals, records result metrics, and +// expands Parquet byte-range splits. Workers own disjoint task ranges, so base +// tasks can be updated in place. Only tasks that actually split need replacement +// slices; unsplit scans return plannedTasks directly without another O(files) +// copy. +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), max(1, len(plannedTasks)/scanTaskFinalizationMinTasksPerWorker)) + results := make([]scanTaskFinalizationResult, workerCount) + chunkSize := (len(plannedTasks) + workerCount - 1) / workerCount + + finalizeRange := func(worker, start, end int) { + results[worker] = scan.finalizeTaskRange( + plannedTasks, start, end, residualEvaluators, splitTargetSize) + } + + if workerCount == 1 { + finalizeRange(0, 0, len(plannedTasks)) + } else { + var wg sync.WaitGroup + for worker := range workerCount { + start := worker * chunkSize + end := min(start+chunkSize, len(plannedTasks)) + + wg.Add(1) + go func(worker, start, end int) { + defer wg.Done() + finalizeRange(worker, start, end) + }(worker, start, end) + } + wg.Wait() + } + + // Each worker stops at its first error, and results follow contiguous task + // ranges. The first error in result order is therefore also the first error + // in task order, regardless of which worker finished first. + for i := range results { + if results[i].err != nil { + return nil, results[i].err + } + } + + totalTasks := len(plannedTasks) + for i := range results { + mergeTaskFinalizationMetrics(acc, &results[i].metrics) + for _, replacement := range results[i].replacements { + totalTasks += len(replacement.tasks) - 1 + } + } + if totalTasks == len(plannedTasks) { + return plannedTasks, nil + } + + finalized := make([]FileScanTask, 0, totalTasks) Review Comment: On the split path we hold `plannedTasks`, every per-split replacement slice, and `finalized` live at the same time, then copy every task again into `finalized`, so peak footprint is roughly double on exactly the large-table, many-splits case. That's the case `planFilesLocal` just worked to shrink by releasing the delete indexes first, and the split benchmark barely scales (22ms to 17ms across 1 to 8 workers). I'd have each worker emit its range already in final order, unsplit tasks and splits inline, and concatenate the per-worker slices once, so there's a single copy and no separate replacement bookkeeping. ########## table/scan_task_finalization_test.go: ########## @@ -0,0 +1,317 @@ +// 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" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFinalizePlannedTasksParallelMatchesSerial(t *testing.T) { Review Comment: Two gaps here, both on the most intricate code in the PR. Every file splits into four (offsets `[8,25,50,75]`, target 25), so `plannedTasks[next:replacement.index]` is always empty and the gap-stitching in reassembly never runs; deleting that copy line would still pass this test. And "matches serial" compares `concurrency: 8` against `concurrency: 1` of the same function, so a regression that dropped a gap or a metric would pass both sides. I'd build an independent serial reference straight from `splitParquetScanTask` and assert against that, with a mix of split and unsplit tasks (splits at index 0, the last index, adjacent, and straddling a chunk boundary). Varying the task count across the 64-per-worker threshold (0, 1, 63, 128, 200) while here would cover the chunk arithmetic too. ########## table/scan_task_finalization.go: ########## @@ -0,0 +1,199 @@ +// 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" + "sync" + + "github.com/apache/iceberg-go" +) + +// Keep enough work in each finalization worker to amortize goroutine and +// result-merging overhead on small scans. +const scanTaskFinalizationMinTasksPerWorker = 64 + +type scanTaskSplitReplacement struct { + index int + tasks []FileScanTask +} + +type scanTaskFinalizationResult struct { + metrics scanMetricsAccumulator + replacements []scanTaskSplitReplacement + err error +} + +// finalizePlannedTasks applies task residuals, records result metrics, and +// expands Parquet byte-range splits. Workers own disjoint task ranges, so base +// tasks can be updated in place. Only tasks that actually split need replacement +// slices; unsplit scans return plannedTasks directly without another O(files) +// copy. +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), max(1, len(plannedTasks)/scanTaskFinalizationMinTasksPerWorker)) + results := make([]scanTaskFinalizationResult, workerCount) + chunkSize := (len(plannedTasks) + workerCount - 1) / workerCount + + finalizeRange := func(worker, start, end int) { + results[worker] = scan.finalizeTaskRange( + plannedTasks, start, end, residualEvaluators, splitTargetSize) + } + + if workerCount == 1 { + finalizeRange(0, 0, len(plannedTasks)) + } else { + var wg sync.WaitGroup + for worker := range workerCount { + start := worker * chunkSize + end := min(start+chunkSize, len(plannedTasks)) + + wg.Add(1) + go func(worker, start, end int) { Review Comment: A panic in `residual` or `splitParquetScanTask` now unwinds this anonymous goroutine with no recover, so it takes down the process. Before, it unwound on the caller's goroutine, where a `recover` around `PlanFiles` could catch it. It's not a regression in kind, since the manifest `errgroup` has the same gap, but it does widen the exposure to callers that recover. I'd add a `defer` in the worker that converts a panic into `result.err` so it flows through the same first-error path. ########## table/scan_task_finalization_test.go: ########## @@ -0,0 +1,317 @@ +// 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" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFinalizePlannedTasksParallelMatchesSerial(t *testing.T) { + 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, 256) + 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} Review Comment: These cases populate `DeleteFiles` and `EqualityDeleteFiles` but never a deletion vector, so the `dvs` count and the `content_size_in_bytes` size path through the per-worker merge go untested; a regression that dropped either from `mergeTaskFinalizationMetrics` would still pass. I'd add a case with a Puffin DV delete file and assert `dvs`, `totalDeleteFileSize`, and `resultDeleteFiles`. ########## table/scan_task_finalization_test.go: ########## @@ -0,0 +1,317 @@ +// 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" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFinalizePlannedTasksParallelMatchesSerial(t *testing.T) { + 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, 256) + 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} + } + } + + serialScan := &Scan{ + metadata: metadata, rowFilter: iceberg.AlwaysTrue{}, caseSensitive: true, concurrency: 1, + } + parallelScan := &Scan{ + metadata: metadata, rowFilter: iceberg.AlwaysTrue{}, caseSensitive: true, concurrency: 8, + } + serialInput := append([]FileScanTask(nil), tasks...) + parallelInput := append([]FileScanTask(nil), tasks...) + var serialMetrics, parallelMetrics scanMetricsAccumulator + + serial, err := serialScan.finalizePlannedTasks(serialInput, schema, &serialMetrics) + require.NoError(t, err) + parallel, err := parallelScan.finalizePlannedTasks(parallelInput, 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(86), parallelMetrics.positionalDeleteFiles) + assert.Equal(t, int64(52), parallelMetrics.equalityDeleteFiles) + assert.Equal(t, int64(138), parallelMetrics.resultDeleteFiles) + assert.Equal(t, int64(13_800), parallelMetrics.totalDeleteFileSize) +} + +func TestFinalizePlannedTasksParallelResidualsMatchSerial(t *testing.T) { + 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, 256) + 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)) + serialScan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive: true, concurrency: 1} + parallelScan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive: true, concurrency: 8} + serialInput := append([]FileScanTask(nil), tasks...) + parallelInput := append([]FileScanTask(nil), tasks...) + var serialMetrics, parallelMetrics scanMetricsAccumulator + + serial, err := serialScan.finalizePlannedTasks(serialInput, schema, &serialMetrics) + require.NoError(t, err) + parallel, err := parallelScan.finalizePlannedTasks(parallelInput, 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, 256) + for i := range tasks { + var partitionValue any = int32(7) + if i == 1 || i == 128 { + // Put failures in separate worker ranges. Both must retain the + // first file's diagnostic even if a later range finishes first. + 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} { + scan := &Scan{ + metadata: metadata, concurrency: concurrency, caseSensitive: true, + rowFilter: iceberg.EqualTo(iceberg.Reference("id"), int32(7)), + } + result, err := scan.finalizePlannedTasks(tasks, schema, &scanMetricsAccumulator{}) + require.ErrorContains(t, err, "evaluate partition residual for mem://table/data/file-1.parquet") + assert.Nil(t, result) + } +} + +func TestFinalizePlannedTasksUnsplitReusesInput(t *testing.T) { + 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 := []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, 1) + assert.Same(t, &tasks[0], &got[0]) +} + +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 + +func BenchmarkFinalizePlannedTasks(b *testing.B) { + schema := simpleSchema() + spec := partitionedSpec() + + for _, tc := range []struct { + name string + rowFilter iceberg.BooleanExpression + offsets []int64 + targetSize string + fileCount int + }{ + { + 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, + }, + } { + 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) + } + tasks := make([]FileScanTask, tc.fileCount) + for i := range tasks { + file := finalizationTestDataFile( + b, + spec, + fmt.Sprintf("mem://benchmark/data/file-%d.parquet", i), + map[int]any{1000: int32(7)}, + 2048, + tc.offsets, + ) + tasks[i] = FileScanTask{File: file, Start: 0, Length: file.FileSizeBytes()} + } + + 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(tc.fileCount), "files/op") + b.ResetTimer() Review Comment: `b.Loop()` resets the timer on its first call, so this `b.ResetTimer()` is a no-op and can go. ########## table/scan_task_finalization_test.go: ########## @@ -0,0 +1,317 @@ +// 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" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFinalizePlannedTasksParallelMatchesSerial(t *testing.T) { + 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, 256) + 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} + } + } + + serialScan := &Scan{ + metadata: metadata, rowFilter: iceberg.AlwaysTrue{}, caseSensitive: true, concurrency: 1, + } + parallelScan := &Scan{ + metadata: metadata, rowFilter: iceberg.AlwaysTrue{}, caseSensitive: true, concurrency: 8, + } + serialInput := append([]FileScanTask(nil), tasks...) + parallelInput := append([]FileScanTask(nil), tasks...) + var serialMetrics, parallelMetrics scanMetricsAccumulator + + serial, err := serialScan.finalizePlannedTasks(serialInput, schema, &serialMetrics) + require.NoError(t, err) + parallel, err := parallelScan.finalizePlannedTasks(parallelInput, 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(86), parallelMetrics.positionalDeleteFiles) + assert.Equal(t, int64(52), parallelMetrics.equalityDeleteFiles) + assert.Equal(t, int64(138), parallelMetrics.resultDeleteFiles) + assert.Equal(t, int64(13_800), parallelMetrics.totalDeleteFileSize) +} + +func TestFinalizePlannedTasksParallelResidualsMatchSerial(t *testing.T) { + 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, 256) + 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)) + serialScan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive: true, concurrency: 1} + parallelScan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive: true, concurrency: 8} + serialInput := append([]FileScanTask(nil), tasks...) + parallelInput := append([]FileScanTask(nil), tasks...) + var serialMetrics, parallelMetrics scanMetricsAccumulator + + serial, err := serialScan.finalizePlannedTasks(serialInput, schema, &serialMetrics) + require.NoError(t, err) + parallel, err := parallelScan.finalizePlannedTasks(parallelInput, 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, 256) + for i := range tasks { + var partitionValue any = int32(7) + if i == 1 || i == 128 { + // Put failures in separate worker ranges. Both must retain the + // first file's diagnostic even if a later range finishes first. + 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} { + scan := &Scan{ + metadata: metadata, concurrency: concurrency, caseSensitive: true, + rowFilter: iceberg.EqualTo(iceberg.Reference("id"), int32(7)), + } + result, err := scan.finalizePlannedTasks(tasks, schema, &scanMetricsAccumulator{}) + require.ErrorContains(t, err, "evaluate partition residual for mem://table/data/file-1.parquet") + assert.Nil(t, result) + } +} + +func TestFinalizePlannedTasksUnsplitReusesInput(t *testing.T) { + 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 := []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, 1) + assert.Same(t, &tasks[0], &got[0]) Review Comment: This pins "same backing array" as a contract on a single-task input, which only ever hits the `workerCount==1` path, and it actively blocks the memory-optimal reassembly above: a change that stopped aliasing the input would read as a regression here. If the aliasing is a real guarantee I'd document it on the function and test it at a multi-worker size; otherwise I'd drop the pointer-identity check and assert that `Residual` was written instead. ########## table/scan_task_finalization_test.go: ########## @@ -0,0 +1,317 @@ +// 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" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFinalizePlannedTasksParallelMatchesSerial(t *testing.T) { + 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, 256) + 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} + } + } + + serialScan := &Scan{ + metadata: metadata, rowFilter: iceberg.AlwaysTrue{}, caseSensitive: true, concurrency: 1, + } + parallelScan := &Scan{ + metadata: metadata, rowFilter: iceberg.AlwaysTrue{}, caseSensitive: true, concurrency: 8, + } + serialInput := append([]FileScanTask(nil), tasks...) + parallelInput := append([]FileScanTask(nil), tasks...) + var serialMetrics, parallelMetrics scanMetricsAccumulator + + serial, err := serialScan.finalizePlannedTasks(serialInput, schema, &serialMetrics) + require.NoError(t, err) + parallel, err := parallelScan.finalizePlannedTasks(parallelInput, 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(86), parallelMetrics.positionalDeleteFiles) + assert.Equal(t, int64(52), parallelMetrics.equalityDeleteFiles) + assert.Equal(t, int64(138), parallelMetrics.resultDeleteFiles) + assert.Equal(t, int64(13_800), parallelMetrics.totalDeleteFileSize) +} + +func TestFinalizePlannedTasksParallelResidualsMatchSerial(t *testing.T) { + 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, 256) + 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)) + serialScan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive: true, concurrency: 1} + parallelScan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive: true, concurrency: 8} + serialInput := append([]FileScanTask(nil), tasks...) + parallelInput := append([]FileScanTask(nil), tasks...) + var serialMetrics, parallelMetrics scanMetricsAccumulator + + serial, err := serialScan.finalizePlannedTasks(serialInput, schema, &serialMetrics) + require.NoError(t, err) + parallel, err := parallelScan.finalizePlannedTasks(parallelInput, 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, 256) + for i := range tasks { + var partitionValue any = int32(7) + if i == 1 || i == 128 { + // Put failures in separate worker ranges. Both must retain the + // first file's diagnostic even if a later range finishes first. + 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} { Review Comment: The ordering guarantee this names, "even if a later range finishes first", isn't actually forced: worker 0 fails at index 1 and so almost always finishes first, so an implementation that returned the last error would still pass most runs. I'd move the first failure late in range 0 and early in range 1 so a wrong selection fails deterministically. Separately, both concurrency values run against the same `tasks` slice that the function mutates, and `err` here shadows the outer one, so I'd clone per iteration and wrap each concurrency in its own `t.Run`. ########## table/scan_task_finalization_test.go: ########## @@ -0,0 +1,317 @@ +// 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" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFinalizePlannedTasksParallelMatchesSerial(t *testing.T) { + 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, 256) + 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} + } + } + + serialScan := &Scan{ + metadata: metadata, rowFilter: iceberg.AlwaysTrue{}, caseSensitive: true, concurrency: 1, + } + parallelScan := &Scan{ + metadata: metadata, rowFilter: iceberg.AlwaysTrue{}, caseSensitive: true, concurrency: 8, + } + serialInput := append([]FileScanTask(nil), tasks...) + parallelInput := append([]FileScanTask(nil), tasks...) + var serialMetrics, parallelMetrics scanMetricsAccumulator + + serial, err := serialScan.finalizePlannedTasks(serialInput, schema, &serialMetrics) + require.NoError(t, err) + parallel, err := parallelScan.finalizePlannedTasks(parallelInput, 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(86), parallelMetrics.positionalDeleteFiles) + assert.Equal(t, int64(52), parallelMetrics.equalityDeleteFiles) + assert.Equal(t, int64(138), parallelMetrics.resultDeleteFiles) + assert.Equal(t, int64(13_800), parallelMetrics.totalDeleteFileSize) +} + +func TestFinalizePlannedTasksParallelResidualsMatchSerial(t *testing.T) { + 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, 256) + 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)) + serialScan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive: true, concurrency: 1} + parallelScan := &Scan{metadata: metadata, rowFilter: filter, caseSensitive: true, concurrency: 8} + serialInput := append([]FileScanTask(nil), tasks...) + parallelInput := append([]FileScanTask(nil), tasks...) + var serialMetrics, parallelMetrics scanMetricsAccumulator + + serial, err := serialScan.finalizePlannedTasks(serialInput, schema, &serialMetrics) + require.NoError(t, err) + parallel, err := parallelScan.finalizePlannedTasks(parallelInput, 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, 256) + for i := range tasks { + var partitionValue any = int32(7) + if i == 1 || i == 128 { + // Put failures in separate worker ranges. Both must retain the + // first file's diagnostic even if a later range finishes first. + 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} { + scan := &Scan{ + metadata: metadata, concurrency: concurrency, caseSensitive: true, + rowFilter: iceberg.EqualTo(iceberg.Reference("id"), int32(7)), + } + result, err := scan.finalizePlannedTasks(tasks, schema, &scanMetricsAccumulator{}) + require.ErrorContains(t, err, "evaluate partition residual for mem://table/data/file-1.parquet") + assert.Nil(t, result) + } +} + +func TestFinalizePlannedTasksUnsplitReusesInput(t *testing.T) { + 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 := []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, 1) + assert.Same(t, &tasks[0], &got[0]) +} + +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 + +func BenchmarkFinalizePlannedTasks(b *testing.B) { Review Comment: This benchmarks the new function against itself at different worker counts, so it can't show whether we're actually faster than the serial loop we removed, and `concurrency=1` here isn't that loop (it adds the cache indirection and the two-phase merge). To justify this much concurrency machinery I'd want a before/after against the old path, ideally an end-to-end `PlanFiles` benchmark, since local planning is dominated by manifest IO and a few ms on 100K files sits in the noise. -- 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]
