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 94501ec07 perf(table): append remote Parquet splits without per-file
buffers (#2253)
94501ec07 is described below
commit 94501ec074d50fbccc070b1b7d447008fe280fea
Author: Minh Vu <[email protected]>
AuthorDate: Wed Oct 21 22:51:45 2026 +0200
perf(table): append remote Parquet splits without per-file buffers (#2253)
* perf(table): append remote parquet splits directly
Reuse the final scan task buffer across large Parquet files instead of
allocating a split slice for each input file. Preserve split boundaries,
pass-through tasks, and caller-owned task slices.
Add multi-file regression coverage and batch planning benchmarks.
Signed-off-by: Minh Vu <[email protected]>
* style(table): format remote split batch benchmark
Signed-off-by: Minh Vu <[email protected]>
* style(table): satisfy nlreturn for remote Parquet splits
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/scan_splits.go | 109 +++++++++++++++++++++++++---------------
table/scan_splits_bench_test.go | 63 +++++++++++++++++++++++
table/scan_splits_test.go | 50 ++++++++++++++++++
3 files changed, 181 insertions(+), 41 deletions(-)
diff --git a/table/scan_splits.go b/table/scan_splits.go
index 83de6abfe..0fe38a563 100644
--- a/table/scan_splits.go
+++ b/table/scan_splits.go
@@ -22,54 +22,48 @@ import (
iceinternal "github.com/apache/iceberg-go/internal"
)
-// splitParquetScanTask returns coalesced byte-range tasks when a complete,
-// large-file task can be split safely. The boolean is false when the original
-// task should be used unchanged. Split offsets are absolute byte positions in
-// the data file and each safe range ends at the next supplied offset, or at
the
-// end of the file. The first range starts at byte zero so sparse split-offset
-// lists cannot leave leading row groups unassigned.
-//
-// Tasks that are already partial, small, non-Parquet, or missing usable split
-// offsets stay unchanged. Keeping the original task in those cases preserves
-// the existing behavior for remote tasks and older files.
-func splitParquetScanTask(task FileScanTask, targetSize int64)
([]FileScanTask, bool) {
+// splittableParquetOffsets validates the source task before any output is
+// appended. In particular, malformed split offsets must not leave a partial
+// result behind when a different task in the batch has already been split.
+func splittableParquetOffsets(task FileScanTask, targetSize int64) ([]int64,
int64, bool) {
file := task.File
if file == nil || file.FileFormat() != iceberg.ParquetFile ||
targetSize <= 0 {
- return nil, false
+ return nil, 0, false
}
fileSize := file.FileSizeBytes()
if fileSize <= targetSize || task.Start != 0 || task.Length != fileSize
{
- return nil, false
+ return nil, 0, false
}
offsets := iceinternal.BorrowedDataFileSplitOffsets(file)
if len(offsets) < 2 {
- return nil, false
+ return nil, 0, false
}
-
for i, offset := range offsets {
if offset < 0 || offset >= fileSize || (i > 0 && offset <=
offsets[i-1]) {
- return nil, false
+ return nil, 0, false
}
}
- // Each pair of adjacent output tasks covers more than targetSize bytes.
- // Bound the initial capacity to avoid reserving one task per row group
when
- // most row groups will be coalesced. Compare before multiplying to
avoid
- // overflow for very large files.
- capacity := len(offsets)
+ return offsets, fileSize, true
+}
+
+// parquetSplitCapacity bounds the reserved output size for coalesced groups.
+// Compare before multiplying to avoid overflow for very large files.
+func parquetSplitCapacity(offsetCount int, fileSize, targetSize int64) int {
+ capacity := offsetCount
if groups := fileSize / targetSize; groups < int64(capacity/2) {
capacity = int(2*groups + 1)
}
- result := make([]FileScanTask, 0, capacity)
- appendTask := func(start, end int64) {
- split := task
- split.Start = start
- split.Length = end - start
- result = append(result, split)
- }
+ return capacity
+}
+
+// appendParquetTaskSplits appends the ranges directly to the destination.
+// The offsets have already been checked as strictly increasing, non-negative,
+// and below the file size, so every emitted range has positive length.
+func appendParquetTaskSplits(result []FileScanTask, task FileScanTask, offsets
[]int64, fileSize, targetSize int64) []FileScanTask {
var taskStart, taskEnd int64
for i := range offsets {
chunkStart := int64(0)
@@ -81,36 +75,69 @@ func splitParquetScanTask(task FileScanTask, targetSize
int64) ([]FileScanTask,
if i+1 < len(offsets) {
chunkEnd = offsets[i+1]
}
- if chunkEnd <= chunkStart {
- return nil, false
- }
if taskEnd > taskStart && chunkEnd-taskStart > targetSize {
- appendTask(taskStart, taskEnd)
+ split := task
+ split.Start = taskStart
+ split.Length = taskEnd - taskStart
+ result = append(result, split)
taskStart = chunkStart
}
taskEnd = chunkEnd
}
- if taskEnd <= taskStart {
+
+ split := task
+ split.Start = taskStart
+ split.Length = taskEnd - taskStart
+
+ return append(result, split)
+}
+
+// splitParquetScanTask returns coalesced byte-range tasks when a complete,
+// large-file task can be split safely. The boolean is false when the original
+// task should be used unchanged. Split offsets are absolute byte positions in
+// the data file and each safe range ends at the next supplied offset, or at
the
+// end of the file. The first range starts at byte zero so sparse split-offset
+// lists cannot leave leading row groups unassigned.
+//
+// Tasks that are already partial, small, non-Parquet, or missing usable split
+// offsets stay unchanged. Keeping the original task in those cases preserves
+// the existing behavior for remote tasks and older files.
+func splitParquetScanTask(task FileScanTask, targetSize int64)
([]FileScanTask, bool) {
+ offsets, fileSize, ok := splittableParquetOffsets(task, targetSize)
+ if !ok {
return nil, false
}
- appendTask(taskStart, taskEnd)
- return result, true
+ result := make([]FileScanTask, 0, parquetSplitCapacity(len(offsets),
fileSize, targetSize))
+
+ return appendParquetTaskSplits(result, task, offsets, fileSize,
targetSize), true
}
// splitRemoteScanTasks allocates a new task slice only if a file needs
splitting.
+// All split ranges are written directly into that slice instead of allocating
+// and copying one temporary slice for each large Parquet file.
func splitRemoteScanTasks(tasks []FileScanTask, targetSize int64)
[]FileScanTask {
var result []FileScanTask
for i, task := range tasks {
- if splits, ok := splitParquetScanTask(task, targetSize); ok {
- if result == nil {
- result = append(make([]FileScanTask, 0,
len(tasks)+len(splits)-1), tasks[:i]...)
+ offsets, fileSize, ok := splittableParquetOffsets(task,
targetSize)
+ if !ok {
+ if result != nil {
+ result = append(result, task)
+ }
+
+ continue
+ }
+
+ if result == nil {
+ capacity := len(tasks)
+ extra := parquetSplitCapacity(len(offsets), fileSize,
targetSize) - 1
+ if extra <= int(^uint(0)>>1)-capacity {
+ capacity += extra
}
- result = append(result, splits...)
- } else if result != nil {
- result = append(result, task)
+ result = append(make([]FileScanTask, 0, capacity),
tasks[:i]...)
}
+ result = appendParquetTaskSplits(result, task, offsets,
fileSize, targetSize)
}
if result == nil {
return tasks
diff --git a/table/scan_splits_bench_test.go b/table/scan_splits_bench_test.go
index a60c889a9..f708e197a 100644
--- a/table/scan_splits_bench_test.go
+++ b/table/scan_splits_bench_test.go
@@ -116,3 +116,66 @@ func BenchmarkSplitRemoteScanTasksSplitOffsets(b
*testing.B) {
// Embedding only the public DataFile interface intentionally hides
// DataFileSplitOffsetsRef, forcing the defensive-copy fallback for comparison.
type splitOffsetsPublicBenchmarkFile struct{ iceberg.DataFile }
+
+// BenchmarkSplitRemoteScanTasksBatch measures the split planner with many
+// Parquet files and shared row-group layouts. Input files are built before
+// timing, and the benchmark reports the complete output task slice.
+func BenchmarkSplitRemoteScanTasksBatch(b *testing.B) {
+ const rowGroupBytes int64 = 4096
+
+ for _, tc := range []struct {
+ files, groups, splittablePercent int
+ }{
+ {1024, 4, 100},
+ {1024, 32, 100},
+ {1024, 128, 100},
+ {8192, 4, 100},
+ {8192, 32, 100},
+ {8192, 32, 50},
+ {8192, 32, 0},
+ } {
+ 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
+ targetSize := int64(8192)
+ if tc.groups > 4 {
+ targetSize = 16384
+ }
+
+ tasks := make([]FileScanTask, tc.files)
+ for i := range tasks {
+ builder, err :=
iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentData,
+
fmt.Sprintf("mem://split-batch/data-%d.parquet", i),
+ iceberg.ParquetFile, nil, nil,
nil, 1, fileSize,
+ )
+ if err != nil {
+ b.Fatal(err)
+ }
+ task := FileScanTask{File:
builder.SplitOffsets(offsets).Build(), Length: fileSize}
+ if i%100 >= tc.splittablePercent {
+ task.Start = 1
+ task.Length--
+ }
+ tasks[i] = task
+ }
+
+ probe := splitRemoteScanTasks(tasks, targetSize)
+ if len(probe) < len(tasks) {
+ b.Fatalf("split planner dropped tasks:
got %d, want at least %d", len(probe), len(tasks))
+ }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ splitRemoteScanTasksBenchmarkSink =
splitRemoteScanTasks(tasks, targetSize)
+ }
+ b.ReportMetric(float64(tc.files), "files/op")
+ })
+ }
+}
diff --git a/table/scan_splits_test.go b/table/scan_splits_test.go
index b7d9937a9..f81c35322 100644
--- a/table/scan_splits_test.go
+++ b/table/scan_splits_test.go
@@ -525,3 +525,53 @@ func TestPrepareToReadClosesReaderOnSchemaError(t
*testing.T) {
})
}
}
+
+// TestSplitRemoteScanTasksManyFilesAndInvalidOffsets keeps unsafe tasks intact
+// even after a previous task has produced multiple split ranges.
+func TestSplitRemoteScanTasksManyFilesAndInvalidOffsets(t *testing.T) {
+ valid := splitTestDataFile(t, iceberg.ParquetFile, 100, []int64{8, 40,
70})
+ invalid := splitTestDataFile(t, iceberg.ParquetFile, 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...)
+
+ got := splitRemoteScanTasks(tasks, 50)
+
+ require.Len(t, got, 9)
+ assert.Equal(t, before, tasks, "planning must not modify caller-owned
tasks")
+ assert.Equal(t, before[0], got[0])
+ assert.Equal(t, before[2], got[4])
+ assert.Equal(t, before[3], got[5])
+
+ for _, first := range []int{1, 6} {
+ for splitIndex, wantStart := range []int64{0, 40, 70} {
+ split := got[first+splitIndex]
+ assert.Equal(t, wantStart, split.Start)
+ assert.Equal(t, []int64{40, 30, 30}[splitIndex],
split.Length)
+ assert.Equal(t, valid, split.File)
+ }
+ }
+ for _, split := range got[1:4] {
+ assert.Equal(t, &firstRowID, split.FirstRowID)
+ }
+ for _, split := range got[6:9] {
+ assert.Equal(t, iceberg.AlwaysTrue{}, split.Residual)
+ }
+}
+
+func TestSplitRemoteScanTasksAllInvalidOffsetsSharesInput(t *testing.T) {
+ invalid := splitTestDataFile(t, iceberg.ParquetFile, 100, []int64{8,
40, 40})
+ tasks := []FileScanTask{
+ {File: invalid, Length: 100},
+ {File: invalid, Length: 100},
+ }
+ got := splitRemoteScanTasks(tasks, 50)
+ assert.Equal(t, tasks, got)
+ assert.Same(t, &tasks[0], &got[0], "unsplit batches must keep the input
slice")
+}
