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 a879ce375 perf(table): avoid cloning parquet split offsets during
scans (#2118)
a879ce375 is described below
commit a879ce3755119aa64233a5fb1816c12045cbd78c
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 20 22:44:02 2026 +0200
perf(table): avoid cloning parquet split offsets during scans (#2118)
---
data_file_refs.go | 11 ++++
data_file_refs_test.go | 67 ++++++++++++++++++++++
internal/data_file_ref.go | 29 ++++++++++
internal/data_file_ref_test.go | 107 ++++++++++++++++++++++++++++++++++
table/arrow_scanner.go | 2 +-
table/arrow_scanner_bench_test.go | 65 ++++++++++++++++++---
table/data_file_stats_ref_test.go | 20 +++++++
table/internal/parquet_files.go | 8 ++-
table/scan_splits.go | 7 ++-
table/scan_splits_bench_test.go | 118 ++++++++++++++++++++++++++++++++++++++
10 files changed, 420 insertions(+), 14 deletions(-)
diff --git a/data_file_refs.go b/data_file_refs.go
index 723f400fa..96ec27f40 100644
--- a/data_file_refs.go
+++ b/data_file_refs.go
@@ -44,6 +44,17 @@ func (d *dataFile) DataFileStatsRef(_ internal.DataFileRef) (
return d.valCntMap, d.nullCntMap, d.nanCntMap, d.lowerBoundMap,
d.upperBoundMap
}
+// DataFileSplitOffsetsRef returns the data file's split offsets without
+// initializing unrelated column-statistics maps. The returned slice aliases
+// immutable DataFile state and must be treated as read-only.
+func (d *dataFile) DataFileSplitOffsetsRef(_ internal.DataFileRef) []int64 {
+ if d.Splits == nil {
+ return nil
+ }
+
+ return *d.Splits
+}
+
func (d *dataFile) DataFileCollectionsRef(_ internal.DataFileRef) (
map[int]int64, []byte, []int64, []int,
) {
diff --git a/data_file_refs_test.go b/data_file_refs_test.go
new file mode 100644
index 000000000..d2499a1e4
--- /dev/null
+++ b/data_file_refs_test.go
@@ -0,0 +1,67 @@
+// 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 iceberg
+
+import (
+ "testing"
+
+ "github.com/apache/iceberg-go/internal"
+ "github.com/stretchr/testify/require"
+)
+
+func TestBorrowedSplitOffsetsDoNotMaterializeColumnStats(t *testing.T) {
+ builder, err := NewDataFileBuilder(
+ *UnpartitionedSpec,
+ EntryContentData,
+ "file.parquet",
+ ParquetFile,
+ nil,
+ nil,
+ nil,
+ 1,
+ 128,
+ )
+ require.NoError(t, err)
+
+ file := builder.
+ ColumnSizes(map[int]int64{1: 8}).
+ ValueCounts(map[int]int64{1: 1}).
+ NullValueCounts(map[int]int64{1: 0}).
+ NaNValueCounts(map[int]int64{1: 0}).
+ LowerBoundValues(map[int][]byte{1: {1}}).
+ UpperBoundValues(map[int][]byte{1: {2}}).
+ SplitOffsets([]int64{0, 64}).
+ Build().(*dataFile)
+
+ require.Nil(t, file.colSizeMap)
+ require.Nil(t, file.valCntMap)
+ require.Nil(t, file.nullCntMap)
+ require.Nil(t, file.nanCntMap)
+ require.Nil(t, file.lowerBoundMap)
+ require.Nil(t, file.upperBoundMap)
+
+ offsets := internal.BorrowedDataFileSplitOffsets(file)
+ require.Equal(t, []int64{0, 64}, offsets)
+
+ require.Nil(t, file.colSizeMap)
+ require.Nil(t, file.valCntMap)
+ require.Nil(t, file.nullCntMap)
+ require.Nil(t, file.nanCntMap)
+ require.Nil(t, file.lowerBoundMap)
+ require.Nil(t, file.upperBoundMap)
+}
diff --git a/internal/data_file_ref.go b/internal/data_file_ref.go
index bca42702a..61a418fad 100644
--- a/internal/data_file_ref.go
+++ b/internal/data_file_ref.go
@@ -17,6 +17,8 @@
package internal
+import "slices"
+
// DataFileRef authorizes zero-copy access to immutable DataFile state from
// trusted packages within this module. Go's internal-package rule prevents
// external callers from constructing this token.
@@ -80,6 +82,33 @@ type DataFileCollectionsRef interface {
)
}
+// DataFileSplitOffsets is the public getter needed by the internal
split-offset
+// helper. It intentionally excludes unrelated collection and statistics
getters.
+type DataFileSplitOffsets interface {
+ SplitOffsets() []int64
+}
+
+// DataFileSplitOffsetsRef exposes only split offsets so callers that do not
+// need column statistics do not trigger their lazy materialization.
+type DataFileSplitOffsetsRef interface {
+ DataFileSplitOffsetsRef(DataFileRef) []int64
+}
+
+// BorrowedDataFileSplitOffsets returns split offsets without copying for the
+// built-in data file and falls back to the public getter for external
+// implementations. The returned slice is read-only and must not be mutated or
+// retained beyond the current planning or read operation.
+func BorrowedDataFileSplitOffsets(file DataFileSplitOffsets) []int64 {
+ var splitOffsets []int64
+ if ref, ok := file.(DataFileSplitOffsetsRef); ok {
+ splitOffsets = ref.DataFileSplitOffsetsRef(DataFileRef{})
+ } else {
+ splitOffsets = file.SplitOffsets()
+ }
+
+ return slices.Clip(splitOffsets)
+}
+
func BorrowedDataFileCollections(file DataFileCollections) (
columnSizes map[int]int64,
keyMetadata []byte,
diff --git a/internal/data_file_ref_test.go b/internal/data_file_ref_test.go
new file mode 100644
index 000000000..c174a7227
--- /dev/null
+++ b/internal/data_file_ref_test.go
@@ -0,0 +1,107 @@
+// 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 internal
+
+import (
+ "slices"
+ "testing"
+)
+
+type splitOffsetsTestFile struct {
+ offsets []int64
+ splitOffsetsCalls int
+}
+
+func (f *splitOffsetsTestFile) SplitOffsets() []int64 {
+ f.splitOffsetsCalls++
+
+ return f.offsets
+}
+
+type splitOffsetsReferenceTestFile struct {
+ splitOffsetsTestFile
+ borrowedOffsets []int64
+ refCalls int
+}
+
+func (f *splitOffsetsReferenceTestFile) DataFileSplitOffsetsRef(DataFileRef)
[]int64 {
+ f.refCalls++
+
+ return f.borrowedOffsets
+}
+
+func splitOffsetsTestCases() []struct {
+ name string
+ offsets []int64
+} {
+ populated := make([]int64, 3, 8)
+ copy(populated, []int64{12, 40, 96})
+
+ return []struct {
+ name string
+ offsets []int64
+ }{
+ {name: "nil"},
+ {name: "empty", offsets: []int64{}},
+ {name: "populated with spare capacity", offsets: populated},
+ }
+}
+
+func TestBorrowedDataFileSplitOffsetsUsesPublicFallbackOnly(t *testing.T) {
+ for _, tt := range splitOffsetsTestCases() {
+ t.Run(tt.name, func(t *testing.T) {
+ file := &splitOffsetsTestFile{offsets: tt.offsets}
+
+ got := BorrowedDataFileSplitOffsets(file)
+ if !slices.Equal(got, tt.offsets) || (got == nil) !=
(tt.offsets == nil) {
+ t.Fatalf("BorrowedDataFileSplitOffsets() = %v,
want %v", got, tt.offsets)
+ }
+ if cap(got) != len(got) {
+ t.Fatalf("BorrowedDataFileSplitOffsets()
capacity = %d, want %d", cap(got), len(got))
+ }
+ if file.splitOffsetsCalls != 1 {
+ t.Fatalf("SplitOffsets() called %d times, want
1", file.splitOffsetsCalls)
+ }
+ })
+ }
+}
+
+func TestBorrowedDataFileSplitOffsetsUsesBorrowedReference(t *testing.T) {
+ for _, tt := range splitOffsetsTestCases() {
+ t.Run(tt.name, func(t *testing.T) {
+ file := &splitOffsetsReferenceTestFile{borrowedOffsets:
tt.offsets}
+
+ got := BorrowedDataFileSplitOffsets(file)
+ if !slices.Equal(got, tt.offsets) || (got == nil) !=
(tt.offsets == nil) {
+ t.Fatalf("BorrowedDataFileSplitOffsets() = %v,
want %v", got, tt.offsets)
+ }
+ if len(got) > 0 && &got[0] != &tt.offsets[0] {
+ t.Fatal("BorrowedDataFileSplitOffsets() copied
the borrowed offsets")
+ }
+ if cap(got) != len(got) {
+ t.Fatalf("BorrowedDataFileSplitOffsets()
capacity = %d, want %d", cap(got), len(got))
+ }
+ if file.refCalls != 1 {
+ t.Fatalf("DataFileSplitOffsetsRef() called %d
times, want 1", file.refCalls)
+ }
+ if file.splitOffsetsCalls != 0 {
+ t.Fatal("public getter was called for borrowed
offsets")
+ }
+ })
+ }
+}
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 92364b795..cfa5cf4ba 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -2518,7 +2518,7 @@ func (as *arrowScan) processRecordsWithPlans(
(task.Value.Length != 0 && task.Value.Length !=
task.Value.File.FileSizeBytes()) {
tester.RangeSet = true
tester.Start, tester.Length =
adjustParquetTaskRange(task.Value, rdr.SourceFileSize())
- tester.PlanningSplitOffsets =
task.Value.File.SplitOffsets()
+ tester.PlanningSplitOffsets =
iceinternal.BorrowedDataFileSplitOffsets(task.Value.File)
}
if posSource != nil {
tester.Survivors = &posSource.spans
diff --git a/table/arrow_scanner_bench_test.go
b/table/arrow_scanner_bench_test.go
index b5c5dcc20..e5aec94bc 100644
--- a/table/arrow_scanner_bench_test.go
+++ b/table/arrow_scanner_bench_test.go
@@ -197,12 +197,13 @@ func BenchmarkArrowScanManyFilesAndBatches(b *testing.B) {
}
func benchmarkSplitParquetScan(b *testing.B) (Metadata, iceio.IO,
[]FileScanTask, []FileScanTask, int64) {
- b.Helper()
+ return benchmarkSplitParquetScanWithRowGroups(b, 8, 32768)
+}
- const (
- rowGroups = 8
- rowsPerGroup = 32768
- )
+func benchmarkSplitParquetScanWithRowGroups(
+ b *testing.B, rowGroups, rowsPerGroup int,
+) (Metadata, iceio.IO, []FileScanTask, []FileScanTask, int64) {
+ b.Helper()
rowCount := int64(rowGroups * rowsPerGroup)
schema := iceberg.NewSchema(0,
iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
@@ -233,9 +234,9 @@ func benchmarkSplitParquetScan(b *testing.B) (Metadata,
iceio.IO, []FileScanTask
var buf bytes.Buffer
writerProps := parquet.NewWriterProperties(
parquet.WithStats(true),
- parquet.WithMaxRowGroupLength(rowsPerGroup),
+ parquet.WithMaxRowGroupLength(int64(rowsPerGroup)),
)
- if err := pqarrow.WriteTable(table, &buf, rowsPerGroup, writerProps,
pqarrow.DefaultWriterProps()); err != nil {
+ if err := pqarrow.WriteTable(table, &buf, int64(rowsPerGroup),
writerProps, pqarrow.DefaultWriterProps()); err != nil {
b.Fatal(err)
}
@@ -276,8 +277,15 @@ func benchmarkSplitParquetScan(b *testing.B) (Metadata,
iceio.IO, []FileScanTask
dataFile := dataFileBuilder.SplitOffsets(offsets).Build()
fullTask := FileScanTask{File: dataFile, Start: 0, Length:
int64(buf.Len())}
splitTasks, split := splitParquetScanTask(fullTask, 1)
+ if rowGroups == 1 && !split {
+ // A one-row-group file has fewer than two split offsets and
cannot be split.
+ // Make the task partial so the benchmark still exercises range
planning.
+ partialTask := fullTask
+ partialTask.Length--
+ splitTasks, split = []FileScanTask{partialTask}, true
+ }
if !split || len(splitTasks) != rowGroups {
- b.Fatalf("expected one split task per row group, got %d",
len(splitTasks))
+ b.Fatalf("expected %d benchmark range tasks, got %d",
rowGroups, len(splitTasks))
}
metadata, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder,
@@ -334,6 +342,47 @@ func BenchmarkArrowScanLargeParquetFileSplitTasks(b
*testing.B) {
}
}
+func BenchmarkArrowScanPlanningSplitOffsets(b *testing.B) {
+ for _, rowGroups := range []int{1, 32, 128, 512} {
+ b.Run(fmt.Sprintf("offsets=%d", rowGroups), func(b *testing.B) {
+ metadata, fs, _, tasks, wantRows :=
benchmarkSplitParquetScanWithRowGroups(
+ b, rowGroups, 512/rowGroups)
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ scan := &arrowScan{
+ fs: fs,
+ metadata: metadata,
+ scanSchema:
metadata.CurrentSchema(),
+ projectedSchema:
metadata.CurrentSchema(),
+ boundRowFilter: iceberg.AlwaysTrue{},
+ rowLimit: -1,
+ concurrency: 8,
+ }
+ _, records, err := scan.GetRecords(b.Context(),
tasks)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ var rows int64
+ for record, err := range records {
+ if err != nil {
+ b.Fatal(err)
+ }
+ rows += record.NumRows()
+ record.Release()
+ }
+ if rows != wantRows {
+ b.Fatalf("unexpected row count: got %d,
want %d", rows, wantRows)
+ }
+ }
+ b.ReportMetric(float64(len(tasks)), "tasks/op")
+ b.ReportMetric(float64(rowGroups), "split_offsets/task")
+ })
+ }
+}
+
func benchmarkComplexBoundFilter(b *testing.B, schema *iceberg.Schema)
iceberg.BooleanExpression {
b.Helper()
diff --git a/table/data_file_stats_ref_test.go
b/table/data_file_stats_ref_test.go
index e5e1dee42..937e85491 100644
--- a/table/data_file_stats_ref_test.go
+++ b/table/data_file_stats_ref_test.go
@@ -183,6 +183,26 @@ func TestDataFileCollectionsUsesBorrowedView(t *testing.T)
{
assert.Equal(t, map[int]int64{1: 10}, measuredColumnSizes)
}
+func TestBorrowedDataFileSplitOffsetsUsesBuiltInBackingSlice(t *testing.T) {
+ file := testDataFileWithStats(t)
+ require.Implements(t, (*internal.DataFileSplitOffsetsRef)(nil), file)
+
+ first := internal.BorrowedDataFileSplitOffsets(file)
+ second := internal.BorrowedDataFileSplitOffsets(file)
+ public := file.SplitOffsets()
+ require.NotEmpty(t, first)
+ require.Equal(t, first, second)
+ require.Equal(t, first, public)
+ require.Equal(t, len(first), cap(first))
+
+ if &first[0] != &second[0] {
+ t.Fatal("borrowed split offsets do not share the built-in data
file backing slice")
+ }
+ if &first[0] == &public[0] {
+ t.Fatal("public SplitOffsets unexpectedly exposed the built-in
backing slice")
+ }
+}
+
func TestDataFileEqualityFieldIDsClonesBorrowedView(t *testing.T) {
file := testDataFileWithStats(t)
require.Implements(t, (*internal.DataFileCollectionsRef)(nil), file)
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index 9805878dd..e023739ce 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -1994,9 +1994,11 @@ type ParquetRowGroupTester struct {
// Parquet row-group offset falls in [Start, Start+Length).
Start, Length int64
// PlanningSplitOffsets contains the split offsets used when the scan
task
- // was planned. When it contains one valid offset per row group, range
- // selection uses these offsets instead of re-deriving them from the
opened
- // Parquet footer. Sparse split-offset lists fall back to the footer
offsets.
+ // was planned. It may be a borrowed view of DataFile state and must be
+ // treated as read-only for the lifetime of this tester. When it
contains one
+ // valid offset per row group, range selection uses these offsets
instead of
+ // re-deriving them from the opened Parquet footer. Sparse split-offset
lists
+ // fall back to the footer offsets.
PlanningSplitOffsets []int64
// Survivors, if non-nil, is reset and then filled with one span per
row group
// that survives pruning, in file order, with positions relative to the
full
diff --git a/table/scan_splits.go b/table/scan_splits.go
index a07037e03..83de6abfe 100644
--- a/table/scan_splits.go
+++ b/table/scan_splits.go
@@ -17,7 +17,10 @@
package table
-import "github.com/apache/iceberg-go"
+import (
+ "github.com/apache/iceberg-go"
+ 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
@@ -40,7 +43,7 @@ func splitParquetScanTask(task FileScanTask, targetSize
int64) ([]FileScanTask,
return nil, false
}
- offsets := file.SplitOffsets()
+ offsets := iceinternal.BorrowedDataFileSplitOffsets(file)
if len(offsets) < 2 {
return nil, false
}
diff --git a/table/scan_splits_bench_test.go b/table/scan_splits_bench_test.go
new file mode 100644
index 000000000..a60c889a9
--- /dev/null
+++ b/table/scan_splits_bench_test.go
@@ -0,0 +1,118 @@
+// 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 splitRemoteScanTasksBenchmarkSink []FileScanTask
+
+func BenchmarkSplitRemoteScanTasksSplitOffsets(b *testing.B) {
+ const filesPerOp = 1024
+
+ for _, offsetCount := range []int{1, 32, 128, 512} {
+ for _, publicGetter := range []bool{false, true} {
+ b.Run(fmt.Sprintf("offsets=%d/public=%t", offsetCount,
publicGetter), func(b *testing.B) {
+ fileSize := int64(offsetCount+1) * 64
+ offsets := make([]int64, offsetCount)
+ for i := range offsets {
+ offsets[i] = int64(i+1) * 64
+ }
+
+ makeFile := func(i int) iceberg.DataFile {
+ dataFileBuilder, err :=
iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentData,
+
fmt.Sprintf("mem://benchmark/split-offsets-%d.parquet", i),
+ iceberg.ParquetFile,
+ nil,
+ nil,
+ nil,
+ 1,
+ fileSize,
+ )
+ if err != nil {
+ b.Fatal(err)
+ }
+ file := dataFileBuilder.
+ ColumnSizes(map[int]int64{1:
fileSize}).
+ ValueCounts(map[int]int64{1:
1}).
+
NullValueCounts(map[int]int64{1: 0}).
+ NaNValueCounts(map[int]int64{1:
0}).
+
LowerBoundValues(map[int][]byte{1: {1}}).
+
UpperBoundValues(map[int][]byte{1: {2}}).
+ SplitOffsets(offsets).
+ Build()
+ if publicGetter {
+ file =
splitOffsetsPublicBenchmarkFile{DataFile: file}
+ }
+
+ return file
+ }
+ makeTasks := func() []FileScanTask {
+ tasks := make([]FileScanTask,
filesPerOp)
+ for i := range tasks {
+ tasks[i] = FileScanTask{File:
makeFile(i), Start: 0, Length: fileSize}
+ }
+
+ return tasks
+ }
+
+ probe := makeTasks()
+ wantTasksPerFile := 1
+ if offsetCount > 1 {
+ splits, ok :=
splitParquetScanTask(probe[0], fileSize/2)
+ if !ok {
+ b.Fatal("expected split offsets
to produce scan tasks")
+ }
+ wantTasksPerFile = len(splits)
+ }
+ wantTasks := filesPerOp * wantTasksPerFile
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ // Rebuild distinct files outside the
timer so each timed call measures
+ // first access to per-file metadata,
matching scan planning.
+ b.StopTimer()
+ tasks := makeTasks()
+ b.StartTimer()
+
+ result := splitRemoteScanTasks(tasks,
fileSize/2)
+
+ b.StopTimer()
+ if len(result) != wantTasks {
+
b.Fatalf("splitRemoteScanTasks() returned %d tasks, want %d", len(result),
wantTasks)
+ }
+ splitRemoteScanTasksBenchmarkSink =
result
+ b.StartTimer()
+ }
+ b.ReportMetric(filesPerOp, "files/op")
+ b.ReportMetric(float64(offsetCount),
"split_offsets/file")
+ })
+ }
+ }
+}
+
+// Embedding only the public DataFile interface intentionally hides
+// DataFileSplitOffsetsRef, forcing the defensive-copy fallback for comparison.
+type splitOffsetsPublicBenchmarkFile struct{ iceberg.DataFile }
