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 0490c3f1a perf(table): cache dictionary paths for position deletes
(#2231)
0490c3f1a is described below
commit 0490c3f1aca665130413b311e98a6283ced81761
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 20 20:21:57 2026 +0200
perf(table): cache dictionary paths for position deletes (#2231)
* perf(table): cache dictionary file paths during position delete grouping
* perf(table): keep uncached delete path lean
* style(table): fix lint findings
* fix(table): reject null dictionary file paths in position deletes
* test(table): cover null dictionary indices in position deletes
* test(table): expect column validation for null dictionary indices
---
table/arrow_scanner.go | 108 +++++++++++++++-
table/arrow_scanner_posdelete_cache_test.go | 186 ++++++++++++++++++++++++++++
2 files changed, 291 insertions(+), 3 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 3a75df002..f9ca7d6b2 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -712,6 +712,10 @@ func validatePosDeleteColumnLengths(filePathLen, posLen
int) error {
return nil
}
+// A position delete file usually references few data files. Do not reserve
+// memory proportional to an unusually large Arrow dictionary.
+const maxPositionDeleteDictionaryCacheEntries = 4096
+
func (a *posDeleteAccumulator) appendFilePathChunk(ctx context.Context,
filePathChunk arrow.Array,
posCursor *posDeleteCursor,
) error {
@@ -726,12 +730,30 @@ func (a *posDeleteAccumulator) appendFilePathChunk(ctx
context.Context, filePath
var dictionary arrow.Array
var indices *array.Dictionary
- if dict, ok := filePathChunk.(*array.Dictionary); ok &&
dict.Dictionary().NullN() > 0 {
- dictionary = dict.Dictionary()
+ if dict, ok := filePathChunk.(*array.Dictionary); ok {
indices = dict
+ if dict.Dictionary().NullN() > 0 {
+ dictionary = dict.Dictionary()
+ }
+ if size := dict.Dictionary().Len(); size > 0 && size <=
maxPositionDeleteDictionaryCacheEntries {
+ return a.appendFilePathChunkWithDictionaryCache(
+ ctx, paths, dictionary, indices, dict,
posCursor)
+ }
}
- for i := range filePathChunk.Len() {
+ return a.appendFilePathChunkWithoutCache(
+ ctx, filePathChunk.Len(), paths, dictionary, indices, posCursor)
+}
+
+func (a *posDeleteAccumulator) appendFilePathChunkWithoutCache(
+ ctx context.Context,
+ length int,
+ paths arrow.TypedArray[string],
+ dictionary arrow.Array,
+ indices *array.Dictionary,
+ posCursor *posDeleteCursor,
+) error {
+ for i := range length {
if i&(positionalDeleteCancellationCheckInterval-1) == 0 {
if err := ctx.Err(); err != nil {
return err
@@ -747,6 +769,10 @@ func (a *posDeleteAccumulator) appendFilePathChunk(ctx
context.Context, filePath
return fmt.Errorf("%w: negative pos %d in position
delete file",
iceberg.ErrInvalidSchema, pos)
}
+ if indices != nil && indices.IsNull(i) {
+ return fmt.Errorf("%w: null file_path dictionary index
in position delete file",
+ iceberg.ErrInvalidSchema)
+ }
if dictionary != nil &&
dictionary.IsNull(indices.GetValueIndex(i)) {
return fmt.Errorf("%w: null file_path dictionary value
in position delete file",
iceberg.ErrInvalidSchema)
@@ -771,6 +797,82 @@ func (a *posDeleteAccumulator) appendFilePathChunk(ctx
context.Context, filePath
return nil
}
+func (a *posDeleteAccumulator) appendFilePathChunkWithDictionaryCache(
+ ctx context.Context,
+ paths arrow.TypedArray[string],
+ dictionary arrow.Array,
+ indices *array.Dictionary,
+ cachedIndices *array.Dictionary,
+ posCursor *posDeleteCursor,
+) error {
+ cachedBuilders := make([]*array.Int64Builder,
cachedIndices.Dictionary().Len())
+ cachedSeen := make([]bool, len(cachedBuilders))
+
+ for i := range cachedIndices.Len() {
+ if i&(positionalDeleteCancellationCheckInterval-1) == 0 {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ }
+
+ pos, ok := posCursor.next()
+ if !ok {
+ return fmt.Errorf("%w: position delete columns ended
before file_path column",
+ iceberg.ErrInvalidSchema)
+ }
+ if pos < 0 {
+ return fmt.Errorf("%w: negative pos %d in position
delete file",
+ iceberg.ErrInvalidSchema, pos)
+ }
+ if indices != nil && indices.IsNull(i) {
+ return fmt.Errorf("%w: null file_path dictionary index
in position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+ if dictionary != nil &&
dictionary.IsNull(indices.GetValueIndex(i)) {
+ return fmt.Errorf("%w: null file_path dictionary value
in position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+
+ // Resolve each dictionary index once per chunk. Several
indexes may
+ // refer to the same file path, so retain the existing
string-keyed
+ // builders as the source of truth.
+ cacheIndex := cachedIndices.GetValueIndex(i)
+ if cacheIndex >= 0 && cacheIndex < len(cachedSeen) &&
cachedSeen[cacheIndex] {
+ if builder := cachedBuilders[cacheIndex]; builder !=
nil {
+ builder.Append(pos)
+ }
+
+ continue
+ }
+
+ path := paths.Value(i)
+ if len(a.targets) > 0 {
+ if _, ok := a.targets[path]; !ok {
+ // A nil builder caches a rejected target as
well.
+ if cacheIndex >= 0 && cacheIndex <
len(cachedSeen) {
+ cachedSeen[cacheIndex] = true
+ }
+
+ continue
+ }
+ }
+
+ builder, ok := a.builders[path]
+ if !ok {
+ path = strings.Clone(path)
+ builder = array.NewInt64Builder(a.mem)
+ a.builders[path] = builder
+ }
+ if cacheIndex >= 0 && cacheIndex < len(cachedSeen) {
+ cachedSeen[cacheIndex] = true
+ cachedBuilders[cacheIndex] = builder
+ }
+ builder.Append(pos)
+ }
+
+ return nil
+}
+
func (a *posDeleteAccumulator) appendChunked(ctx context.Context, filePathCol,
posCol *arrow.Chunked) error {
if err := ctx.Err(); err != nil {
return err
diff --git a/table/arrow_scanner_posdelete_cache_test.go
b/table/arrow_scanner_posdelete_cache_test.go
new file mode 100644
index 000000000..ec9e287bc
--- /dev/null
+++ b/table/arrow_scanner_posdelete_cache_test.go
@@ -0,0 +1,186 @@
+// 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/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/compute"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/stretchr/testify/require"
+)
+
+func TestPosDeleteDictionaryPathCache(t *testing.T) {
+ for _, dictionarySize := range []int{3,
maxPositionDeleteDictionaryCacheEntries + 1} {
+ t.Run(fmt.Sprintf("dictionary-entries=%d", dictionarySize),
func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+
+ pathNames := make([]string, dictionarySize)
+ for i := range pathNames {
+ pathNames[i] = fmt.Sprintf("other-%d.parquet",
i)
+ }
+ pathNames[0] = "file-a.parquet"
+ pathNames[1] = "file-b.parquet"
+ pathNames[2] = "file-a.parquet" // Different indexes
with the same value.
+
+ dictionary := stringArray(mem, pathNames...)
+ defer dictionary.Release()
+ indexes := int32Array(mem, 0, 0, 1, 2, 1, 2, 0)
+ defer indexes.Release()
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int32,
+ ValueType: arrow.BinaryTypes.String,
+ }
+ pathArray := array.NewDictionaryArray(dictType,
indexes, dictionary)
+ defer pathArray.Release()
+ pathCol := arrow.NewChunked(dictType,
[]arrow.Array{pathArray})
+ defer pathCol.Release()
+
+ posArray := int64Array(mem, 11, 12, 13, 14, 15, 16, 17)
+ defer posArray.Release()
+ posCol := arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{posArray})
+ defer posCol.Release()
+
+ acc := newPosDeleteAccumulator(ctx,
map[string]struct{}{"file-a.parquet": {}})
+ defer acc.release()
+ require.NoError(t, acc.appendChunked(ctx, pathCol,
posCol))
+ grouped := acc.finish()
+ defer releasePosDeletes(grouped)
+ require.Len(t, grouped, 1)
+ require.Equal(t, []int64{11, 12, 14, 16, 17},
int64Values(grouped["file-a.parquet"]))
+ })
+ }
+}
+
+func TestPosDeleteDictionaryPathCacheValidatesEveryPosition(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+
+ dictionary := stringArray(mem, "file-a.parquet")
+ defer dictionary.Release()
+ indexes := int32Array(mem, 0, 0)
+ defer indexes.Release()
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int32,
+ ValueType: arrow.BinaryTypes.String,
+ }
+ pathArray := array.NewDictionaryArray(dictType, indexes, dictionary)
+ defer pathArray.Release()
+ pathCol := arrow.NewChunked(dictType, []arrow.Array{pathArray})
+ defer pathCol.Release()
+
+ posArray := int64Array(mem, 1, -2)
+ defer posArray.Release()
+ posCol := arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{posArray})
+ defer posCol.Release()
+
+ acc := newPosDeleteAccumulator(ctx, nil)
+ defer acc.release()
+ err := acc.appendChunked(ctx, pathCol, posCol)
+ require.ErrorContains(t, err, "negative pos -2")
+}
+
+func TestPosDeleteDictionaryPathCacheRejectsNullIndices(t *testing.T) {
+ for _, dictionarySize := range []int{1,
maxPositionDeleteDictionaryCacheEntries + 1} {
+ t.Run(fmt.Sprintf("dictionary-entries=%d", dictionarySize),
func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+
+ paths := make([]string, dictionarySize)
+ for i := range paths {
+ paths[i] = fmt.Sprintf("file-%d.parquet", i)
+ }
+ dictionary := stringArray(mem, paths...)
+ defer dictionary.Release()
+
+ indexBuilder := array.NewInt32Builder(mem)
+ indexBuilder.Append(0)
+ indexBuilder.AppendNull()
+ indexes := indexBuilder.NewArray()
+ indexBuilder.Release()
+ defer indexes.Release()
+
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int32,
+ ValueType: arrow.BinaryTypes.String,
+ }
+ pathArray := array.NewDictionaryArray(dictType,
indexes, dictionary)
+ defer pathArray.Release()
+ pathCol := arrow.NewChunked(dictType,
[]arrow.Array{pathArray})
+ defer pathCol.Release()
+
+ posArray := int64Array(mem, 1, 2)
+ defer posArray.Release()
+ posCol := arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{posArray})
+ defer posCol.Release()
+
+ acc := newPosDeleteAccumulator(ctx, nil)
+ defer acc.release()
+ err := acc.appendChunked(ctx, pathCol, posCol)
+ // Column-level validation rejects null dictionary
indices before either cache path runs.
+ require.ErrorContains(t, err, "null file_path in
position delete file")
+ })
+ }
+}
+
+func BenchmarkGroupPosDeletesByDictionaryIndex(b *testing.B) {
+ const numRows = 262_144
+ for _, numPaths := range []int{1, 16, 512,
maxPositionDeleteDictionaryCacheEntries + 1} {
+ b.Run(fmt.Sprintf("paths=%d", numPaths), func(b *testing.B) {
+ mem := memory.DefaultAllocator
+ pathNames := make([]string, numPaths)
+ for i := range pathNames {
+ pathNames[i] =
fmt.Sprintf("s3://bucket/table/data/%05d.parquet", i)
+ }
+ dictionary := stringArray(mem, pathNames...)
+ defer dictionary.Release()
+
+ indexes := make([]int32, numRows)
+ positions := make([]int64, numRows)
+ for i := range indexes {
+ indexes[i] = int32(i % numPaths)
+ positions[i] = int64(i)
+ }
+ indexArray := int32Array(mem, indexes...)
+ defer indexArray.Release()
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int32,
+ ValueType: arrow.BinaryTypes.String,
+ }
+ pathArray := array.NewDictionaryArray(dictType,
indexArray, dictionary)
+ defer pathArray.Release()
+ pathCol := arrow.NewChunked(dictType,
[]arrow.Array{pathArray})
+ defer pathCol.Release()
+
+ posArray := int64Array(mem, positions...)
+ defer posArray.Release()
+ posCol := arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{posArray})
+ defer posCol.Release()
+
+ ctx := compute.WithAllocator(b.Context(), mem)
+ benchmarkGroupPosDeletesByFilePath(b, ctx, pathCol,
posCol, groupPosDeletesByFilePath)
+ })
+ }
+}
