You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/10/21 20:51:52 UTC
(iceberg-go) branch main updated: perf(table): append remote Parquet splits without per-file buffers (#2253)
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 <vu...@gmail.com>
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 <vu...@gmail.com>
* style(table): format remote split batch benchmark
Signed-off-by: Minh Vu <vu...@gmail.com>
* style(table): satisfy nlreturn for remote Parquet splits
---------
Signed-off-by: Minh Vu <vu...@gmail.com>
---
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")
+}