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 f49e311e2 perf(table): append local Parquet splits without per-file
buffers (#2269)
f49e311e2 is described below
commit f49e311e21a5fdb0069477f1bc7fb1e5e3a203f2
Author: Minh Vu <[email protected]>
AuthorDate: Fri Oct 23 18:28:51 2026 +0200
perf(table): append local Parquet splits without per-file buffers (#2269)
* perf(table): append local Parquet splits without per-file buffers
Signed-off-by: Minh Vu <[email protected]>
* test(table): cover mixed local scan splits
Signed-off-by: Minh Vu <[email protected]>
* style(table): gofmt split benchmark constants
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/scan_task_finalization.go | 11 +--
table/scan_task_finalization_split_bench_test.go | 93 ++++++++++++++++++++++++
table/scan_task_finalization_split_test.go | 65 +++++++++++++++++
3 files changed, 164 insertions(+), 5 deletions(-)
diff --git a/table/scan_task_finalization.go b/table/scan_task_finalization.go
index 1e51b55f6..723a0d584 100644
--- a/table/scan_task_finalization.go
+++ b/table/scan_task_finalization.go
@@ -184,15 +184,16 @@ func (scan *Scan) finalizeTaskRange(
result.metrics.addResultDeleteMetrics(*task)
result.metrics.totalFileSize += task.File.FileSizeBytes()
- if splitTasks, split := splitParquetScanTask(*task,
splitTargetSize); split {
+ if offsets, fileSize, split := splittableParquetOffsets(*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)))
+ // A mixed range keeps the unsplit prefix.
Reserve space once
+ // without allocating a temporary slice for
each split file.
+ result.tasks = make([]FileScanTask, 0,
max((end-start)*2,
+
index-start+parquetSplitCapacity(len(offsets), fileSize, splitTargetSize)))
result.tasks = append(result.tasks,
plannedTasks[start:index]...)
}
- result.tasks = append(result.tasks, splitTasks...)
+ result.tasks = appendParquetTaskSplits(result.tasks,
*task, offsets, fileSize, splitTargetSize)
} else if result.expanded {
result.tasks = append(result.tasks, *task)
}
diff --git a/table/scan_task_finalization_split_bench_test.go
b/table/scan_task_finalization_split_bench_test.go
new file mode 100644
index 000000000..07f2a2c5f
--- /dev/null
+++ b/table/scan_task_finalization_split_bench_test.go
@@ -0,0 +1,93 @@
+// 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"
+)
+
+var finalizeTaskRangeSplitBenchmarkSink []FileScanTask
+
+// BenchmarkFinalizeTaskRangeParquetSplits isolates local scan task
finalization.
+// Files and split offsets are built before timing. No residual evaluation or
+// manifest I/O is included, so the result measures split/append overhead.
+func BenchmarkFinalizeTaskRangeParquetSplits(b *testing.B) {
+ const (
+ rowGroupBytes int64 = 4096
+ targetSize int64 = 16384
+ )
+
+ for _, tc := range []struct {
+ files, groups, splittablePercent int
+ }{
+ {8192, 32, 0},
+ {8192, 32, 50},
+ {8192, 32, 100},
+ {2048, 128, 100},
+ } {
+ b.Run(fmt.Sprintf("files=%d/groups=%d/splittable=%d",
+ tc.files, tc.groups, tc.splittablePercent), func(b
*testing.B) {
+ offsets := make([]int64, tc.groups)
+ for i := range offsets {
+ offsets[i] = int64(i) * rowGroupBytes
+ }
+ fileSize := int64(tc.groups) * rowGroupBytes
+ tasks := make([]FileScanTask, tc.files)
+ for i := range tasks {
+ file := finalizationTestDataFile(
+ b, *iceberg.UnpartitionedSpec,
+
fmt.Sprintf("mem://finalization-splits/data-%d.parquet", i),
+ nil, fileSize, offsets,
+ )
+ task := FileScanTask{File: file, Length:
fileSize}
+ if i%100 >= tc.splittablePercent {
+ task.Start = 1
+ task.Length--
+ }
+ tasks[i] = task
+ }
+
+ expectedTasks := 0
+ for _, task := range tasks {
+ if splits, ok := splitParquetScanTask(task,
targetSize); ok {
+ expectedTasks += len(splits)
+ } else {
+ expectedTasks++
+ }
+ }
+
+ scan := &Scan{}
+ b.ReportAllocs()
+ b.ReportMetric(float64(tc.files), "files/op")
+ b.ResetTimer()
+ for b.Loop() {
+ result := scan.finalizeTaskRange(tasks, 0,
len(tasks), nil, targetSize)
+ if result.err != nil {
+ b.Fatal(result.err)
+ }
+ if len(result.tasks) != expectedTasks {
+ b.Fatalf("got %d tasks, want %d",
len(result.tasks), expectedTasks)
+ }
+ finalizeTaskRangeSplitBenchmarkSink =
result.tasks
+ }
+ })
+ }
+}
diff --git a/table/scan_task_finalization_split_test.go
b/table/scan_task_finalization_split_test.go
new file mode 100644
index 000000000..953977d4b
--- /dev/null
+++ b/table/scan_task_finalization_split_test.go
@@ -0,0 +1,65 @@
+// 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 (
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestFinalizeTaskRangeMixedParquetSplitOffsets(t *testing.T) {
+ spec := *iceberg.UnpartitionedSpec
+ valid := finalizationTestDataFile(t, spec, "mem://valid.parquet", nil,
100, []int64{8, 40, 70})
+ invalid := finalizationTestDataFile(t, spec, "mem://invalid.parquet",
nil, 100, []int64{8, 40, 40})
+ firstRowID := int64(23)
+
+ tasks := []FileScanTask{
+ {File: invalid, Length: 100},
+ {File: valid, Length: 100, FirstRowID: &firstRowID},
+ {File: valid, Start: 40, Length: 60},
+ {File: invalid, Length: 100},
+ {File: valid, Length: 100, Residual: iceberg.AlwaysTrue{}},
+ }
+ before := append([]FileScanTask(nil), tasks...)
+ result := (&Scan{}).finalizeTaskRange(tasks, 0, len(tasks), nil, 50)
+
+ require.NoError(t, result.err)
+ assert.True(t, result.expanded)
+ require.Len(t, result.tasks, 9)
+ assert.Equal(t, before, tasks, "splitting must not mutate the source
tasks")
+ assert.Equal(t, before[0], result.tasks[0])
+ assert.Equal(t, before[2], result.tasks[4])
+ assert.Equal(t, before[3], result.tasks[5])
+ for _, first := range []int{1, 6} {
+ for splitIndex, start := range []int64{0, 40, 70} {
+ split := result.tasks[first+splitIndex]
+ assert.Equal(t, start, split.Start)
+ assert.Equal(t, []int64{40, 30, 30}[splitIndex],
split.Length)
+ assert.Equal(t, valid, split.File)
+ }
+ }
+ for _, task := range result.tasks[1:4] {
+ assert.Equal(t, &firstRowID, task.FirstRowID)
+ }
+ for _, task := range result.tasks[6:9] {
+ assert.Equal(t, iceberg.AlwaysTrue{}, task.Residual)
+ }
+}
