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 db7645084 perf(table): reuse position delete indexes across adjacent
splits (#2255)
db7645084 is described below
commit db764508477e380e2a95d6aeef2ea4355a5abc07
Author: Minh Vu <[email protected]>
AuthorDate: Wed Oct 21 22:54:11 2026 +0200
perf(table): reuse position delete indexes across adjacent splits (#2255)
* perf(table): reuse positional-delete indexes across adjacent splits
Signed-off-by: Minh Vu <[email protected]>
* test(table): cover split index reuse and changing delete sets
Signed-off-by: Minh Vu <[email protected]>
* bench(table): compare positional-delete index construction A/B
Signed-off-by: Minh Vu <[email protected]>
* style(table): gofmt positional delete index tests
Signed-off-by: Minh Vu <[email protected]>
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/arrow_scanner.go | 39 ++++++
...row_scanner_posdelete_index_reuse_bench_test.go | 135 +++++++++++++++++++++
table/arrow_scanner_posdelete_index_reuse_test.go | 97 +++++++++++++++
3 files changed, 271 insertions(+)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index cfa5cf4ba..8c884041b 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -257,6 +257,25 @@ func positionDeleteIndexKey(task FileScanTask) (key,
targetPath string, deletePa
return builder.String(), targetPath, deletePaths
}
+// sameOrderedPositionDeleteFiles compares only the references used to build a
+// positional-delete index. Other delete types do not affect the index key.
+func sameOrderedPositionDeleteFiles(left, right []iceberg.DataFile) bool {
+ if len(left) != len(right) {
+ return false
+ }
+
+ for i, file := range left {
+ leftPosDelete := file.ContentType() ==
iceberg.EntryContentPosDeletes
+ rightPosDelete := right[i].ContentType() ==
iceberg.EntryContentPosDeletes
+ if leftPosDelete != rightPosDelete ||
+ (leftPosDelete && file.FilePath() !=
right[i].FilePath()) {
+ return false
+ }
+ }
+
+ return true
+}
+
func newLazyPositionDeleteLoader(fs iceio.IO, tasks []FileScanTask)
*lazyPositionDeleteLoader {
uniqueDeletes, targetsByDelete :=
collectPositionDeleteFilesAndTargets(tasks)
if len(uniqueDeletes) == 0 {
@@ -275,10 +294,29 @@ func newLazyPositionDeleteLoader(fs iceio.IO, tasks
[]FileScanTask) *lazyPositio
}
}
indexesByKey := make(map[string]*lazyPositionDeleteIndex)
+ var previousIndex *lazyPositionDeleteIndex
+ var previousTargetPath string
for i, task := range tasks {
+ // Consecutive splits of a data file usually share its delete
set.
+ // Avoid rebuilding and sorting the same long path key for each
split.
+ if i > 0 && previousIndex != nil && task.File != nil &&
+ len(task.DeletionVectorFiles) == 0 &&
+ task.File.FilePath() == previousTargetPath &&
+ sameOrderedPositionDeleteFiles(task.DeleteFiles,
tasks[i-1].DeleteFiles) {
+ loader.indexes[i] = positionDeleteTaskIndex{
+ targetPath: previousTargetPath,
+ index: previousIndex,
+ }
+
+ continue
+ }
+
key, targetPath, deletePaths := positionDeleteIndexKey(task)
loader.indexes[i].targetPath = targetPath
+ previousTargetPath = targetPath
if key == "" {
+ previousIndex = nil
+
continue
}
index := indexesByKey[key]
@@ -290,6 +328,7 @@ func newLazyPositionDeleteLoader(fs iceio.IO, tasks
[]FileScanTask) *lazyPositio
indexesByKey[key] = index
}
loader.indexes[i].index = index
+ previousIndex = index
}
return loader
diff --git a/table/arrow_scanner_posdelete_index_reuse_bench_test.go
b/table/arrow_scanner_posdelete_index_reuse_bench_test.go
new file mode 100644
index 000000000..ebc9fb9fa
--- /dev/null
+++ b/table/arrow_scanner_posdelete_index_reuse_bench_test.go
@@ -0,0 +1,135 @@
+// 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 benchmarkPositionDeleteIndexBuildSink *lazyPositionDeleteLoader
+
+// benchmarkPositionDeleteIndexBuildBefore keeps the previous key-generation
+// loop so the same benchmark invocation can measure before and after.
+func benchmarkPositionDeleteIndexBuildBefore(tasks []FileScanTask)
*lazyPositionDeleteLoader {
+ uniqueDeletes, targetsByDelete :=
collectPositionDeleteFilesAndTargets(tasks)
+ if len(uniqueDeletes) == 0 {
+ return nil
+ }
+ loader := &lazyPositionDeleteLoader{
+ files: make(map[string]*lazyPositionDeleteFile,
len(uniqueDeletes)),
+ indexes: make([]positionDeleteTaskIndex, len(tasks)),
+ }
+ for path, deleteFile := range uniqueDeletes {
+ loader.files[path] = &lazyPositionDeleteFile{
+ dataFile: deleteFile,
+ targets: targetsByDelete[path],
+ }
+ }
+ indexesByKey := make(map[string]*lazyPositionDeleteIndex)
+ for i, task := range tasks {
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
+ loader.indexes[i].targetPath = targetPath
+ if key == "" {
+ continue
+ }
+ index := indexesByKey[key]
+ if index == nil {
+ index = &lazyPositionDeleteIndex{
+ targetPath: targetPath,
+ deletePaths: deletePaths,
+ }
+ indexesByKey[key] = index
+ }
+ loader.indexes[i].index = index
+ }
+
+ return loader
+}
+
+func benchmarkPositionDeleteIndexTasks(deleteCount int, interleaved bool)
[]FileScanTask {
+ const fileCount, splitsPerFile = 1000, 10
+ dataFiles := make([]iceberg.DataFile, fileCount)
+ deleteFiles := make([][]iceberg.DataFile, fileCount)
+ for i := range dataFiles {
+ dataFiles[i] = &mockDataFile{
+ path:
fmt.Sprintf("s3://bucket/catalog/namespace/table/data/%016d-%032d.parquet", i,
i),
+ }
+ deleteFiles[i] = make([]iceberg.DataFile, deleteCount)
+ for j := range deleteFiles[i] {
+ deleteFiles[i][j] = &mockDataFile{
+ path:
fmt.Sprintf("s3://bucket/catalog/namespace/table/deletes/%016d-%012d-%032d.parquet",
i, j, j),
+ contentType: iceberg.EntryContentPosDeletes,
+ }
+ }
+ }
+ tasks := make([]FileScanTask, fileCount*splitsPerFile)
+ for i := range tasks {
+ fileIndex := i / splitsPerFile
+ if interleaved {
+ fileIndex = i % fileCount
+ }
+ tasks[i] = FileScanTask{
+ File: dataFiles[fileIndex],
+ DeleteFiles: deleteFiles[fileIndex],
+ Start: int64(i * 128),
+ Length: 128,
+ }
+ }
+
+ return tasks
+}
+
+// BenchmarkPositionDeleteIndexConstruction measures building the complete
+// scan-scoped index for 10,000 split tasks and 1,000 data files. The
interleaved
+// control prevents a best-case-only result. It excludes Parquet IO.
+func BenchmarkPositionDeleteIndexConstruction(b *testing.B) {
+ for _, deleteCount := range []int{1, 4, 16} {
+ for _, layout := range []struct {
+ name string
+ interleaved bool
+ }{
+ {name: "contiguous"},
+ {name: "interleaved", interleaved: true},
+ } {
+ tasks := benchmarkPositionDeleteIndexTasks(deleteCount,
layout.interleaved)
+ b.Run(fmt.Sprintf("%s/deletes=%d", layout.name,
deleteCount), func(b *testing.B) {
+ for _, implementation := range []struct {
+ name string
+ build func() *lazyPositionDeleteLoader
+ }{
+ {name: "before", build: func()
*lazyPositionDeleteLoader {
+ return
benchmarkPositionDeleteIndexBuildBefore(tasks)
+ }},
+ {name: "after", build: func()
*lazyPositionDeleteLoader {
+ return
newLazyPositionDeleteLoader(nil, tasks)
+ }},
+ } {
+ b.Run(implementation.name, func(b
*testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+
benchmarkPositionDeleteIndexBuildSink = implementation.build()
+ }
+ })
+ }
+ })
+ }
+ }
+}
diff --git a/table/arrow_scanner_posdelete_index_reuse_test.go
b/table/arrow_scanner_posdelete_index_reuse_test.go
new file mode 100644
index 000000000..5c4889ecd
--- /dev/null
+++ b/table/arrow_scanner_posdelete_index_reuse_test.go
@@ -0,0 +1,97 @@
+// 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/require"
+)
+
+func TestPositionDeleteIndexReuseAcrossSplits(t *testing.T) {
+ data := &mockDataFile{path: "s3://bucket/data/a.parquet"}
+ otherData := &mockDataFile{path: "s3://bucket/data/b.parquet"}
+ deleteA := &mockDataFile{path: "s3://bucket/delete/a.parquet",
contentType: iceberg.EntryContentPosDeletes}
+ deleteB := &mockDataFile{path: "s3://bucket/delete/b.parquet",
contentType: iceberg.EntryContentPosDeletes}
+ deleteC := &mockDataFile{path: "s3://bucket/delete/c.parquet",
contentType: iceberg.EntryContentPosDeletes}
+ otherDelete := &mockDataFile{path:
"s3://bucket/delete/not-pos.parquet", contentType: iceberg.EntryContentData}
+
+ tasks := []FileScanTask{
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteA, deleteB}},
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteA,
deleteB}}, // Adjacent identical set.
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteB,
deleteA}}, // Reordered set.
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteA,
deleteC}}, // Changed set.
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteA, deleteB,
deleteB}}, // Duplicate path.
+ {File: otherData, DeleteFiles: []iceberg.DataFile{deleteA,
deleteB}}, // Different target.
+ {
+ File: data, DeleteFiles: []iceberg.DataFile{deleteA,
deleteB},
+ DeletionVectorFiles: []iceberg.DataFile{deleteC}, //
Superseding DV.
+ },
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteA,
deleteB}}, // After DV.
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteA,
otherDelete}}, // Non-positional delete.
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteA}},
+ {File: data, DeleteFiles: []iceberg.DataFile{deleteA, deleteB}},
+ {DeleteFiles: []iceberg.DataFile{deleteA}}, // Missing data
file path.
+ {File: data}, // No delete files.
+ }
+
+ loader := newLazyPositionDeleteLoader(nil, tasks)
+ require.NotNil(t, loader)
+ defer loader.release()
+
+ for i, task := range tasks {
+ wantPath := ""
+ if task.File != nil {
+ wantPath = task.File.FilePath()
+ }
+ require.Equal(t, wantPath, loader.indexes[i].targetPath, "task
%d", i)
+ }
+
+ indexes := loader.indexes
+ require.NotNil(t, indexes[0].index)
+ for _, i := range []int{1, 2, 4, 7, 10} {
+ require.Same(t, indexes[0].index, indexes[i].index, "task %d
should share the canonical index", i)
+ }
+ require.Equal(t, []string{deleteA.FilePath(), deleteB.FilePath()},
indexes[0].index.deletePaths)
+ require.NotSame(t, indexes[0].index, indexes[3].index) // Changed
delete set.
+ require.NotSame(t, indexes[0].index, indexes[5].index) // Changed
target path.
+ require.Same(t, indexes[8].index, indexes[9].index) //
Non-positional entries are ignored.
+ require.NotSame(t, indexes[0].index, indexes[9].index)
+ for _, i := range []int{6, 11, 12} {
+ require.Nil(t, indexes[i].index, "task %d must not have an
index", i)
+ }
+}
+
+func TestSameOrderedPositionDeleteFiles(t *testing.T) {
+ a := &mockDataFile{path: "delete-a.parquet", contentType:
iceberg.EntryContentPosDeletes}
+ b := &mockDataFile{path: "delete-b.parquet", contentType:
iceberg.EntryContentPosDeletes}
+ nonPosA := &mockDataFile{path: "other-a.parquet", contentType:
iceberg.EntryContentData}
+ nonPosB := &mockDataFile{path: "other-b.parquet", contentType:
iceberg.EntryContentData}
+
+ require.True(t, sameOrderedPositionDeleteFiles(
+ []iceberg.DataFile{a, nonPosA}, []iceberg.DataFile{a, nonPosB}))
+ require.False(t, sameOrderedPositionDeleteFiles(
+ []iceberg.DataFile{a, b}, []iceberg.DataFile{b, a}))
+ require.False(t, sameOrderedPositionDeleteFiles(
+ []iceberg.DataFile{a}, []iceberg.DataFile{b}))
+ require.False(t, sameOrderedPositionDeleteFiles(
+ []iceberg.DataFile{a}, []iceberg.DataFile{nonPosA}))
+ require.False(t, sameOrderedPositionDeleteFiles(
+ []iceberg.DataFile{a}, nil))
+}
