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 89a9104ec perf(table): filter position-delete reads by file path
(#1937)
89a9104ec is described below
commit 89a9104ec3639d38ab07e4461721cb5270e95571
Author: Minh Vu <[email protected]>
AuthorDate: Mon Sep 28 13:00:13 2026 +0200
perf(table): filter position-delete reads by file path (#1937)
perf(table): filter lazy position-delete reads by file path
---
table/arrow_scanner.go | 231 ++++++-
table/arrow_scanner_lazy_delete_regression_test.go | 18 +-
table/arrow_scanner_nested_delete_test.go | 71 +++
table/arrow_scanner_posdelete_bench_test.go | 111 +++-
table/arrow_scanner_posdelete_regression_test.go | 684 ++++++++++++++++++++-
table/arrow_scanner_test.go | 11 +-
6 files changed, 1076 insertions(+), 50 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 7b5469902..4d29a69e4 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -57,15 +57,16 @@ const (
var PositionalDeleteArrowSchema, _ =
SchemaToArrowSchema(iceberg.PositionalDeleteSchema, nil, true, false)
type (
- positionDeletes = []*arrow.Chunked
- perFilePosDeletes = map[string]positionDeletes
+ positionDeletes = []*arrow.Chunked
+ perFilePosDeletes = map[string]positionDeletes
+ perDeleteFileTargets = map[string]map[string]struct{}
)
// releasePerFilePosDeletes releases every Arrow chunk in a positional-delete
// map. Required on every error return between readAllDeleteFiles and the
// iterator returned by createIterator — Arrow allocations are not freed by
// GC, so dropping the map on the floor leaks the chunks. Safe to call on a
-// nil map; the nil-chunk guard is defensive — readDeletes never inserts a
+// nil map; the nil-chunk guard is defensive — position-delete readers never
insert a
// nil *arrow.Chunked, but the guard keeps callers safe if that invariant
// ever changes.
func releasePerFilePosDeletes(deletesPerFile perFilePosDeletes) {
@@ -78,25 +79,51 @@ func releasePerFilePosDeletes(deletesPerFile
perFilePosDeletes) {
}
}
-// readAllDeleteFiles reads every referenced positional-delete file up front.
It
-// remains the path used by Transaction.makePositionDeleteRecordsForFilter and
by
-// the eager-vs-lazy benchmark; arrowScan.GetRecords uses
lazyPositionDeleteLoader.
-func readAllDeleteFiles(ctx context.Context, fs iceio.IO, tasks
[]FileScanTask, concurrency int) (perFilePosDeletes, error) {
- deletesPerFile := make(perFilePosDeletes)
+// collectPositionDeleteFilesAndTargets indexes each positional-delete file and
+// the data-file paths of tasks that reference it. A nil target set preserves
+// the whole-file fallback when any referencing task lacks a usable data path.
+func collectPositionDeleteFilesAndTargets(tasks []FileScanTask)
(map[string]iceberg.DataFile, perDeleteFileTargets) {
uniqueDeletes := make(map[string]iceberg.DataFile)
+ targetsByDelete := make(perDeleteFileTargets)
- for _, t := range tasks {
- for _, d := range t.DeleteFiles {
- if d.ContentType() != iceberg.EntryContentPosDeletes {
+ for _, task := range tasks {
+ for _, deleteFile := range task.DeleteFiles {
+ if deleteFile.ContentType() !=
iceberg.EntryContentPosDeletes {
+ continue
+ }
+
+ deletePath := deleteFile.FilePath()
+ if _, ok := uniqueDeletes[deletePath]; !ok {
+ uniqueDeletes[deletePath] = deleteFile
+ }
+
+ targets, ok := targetsByDelete[deletePath]
+ if !ok {
+ targets = make(map[string]struct{})
+ targetsByDelete[deletePath] = targets
+ }
+ if targets == nil {
continue
}
+ if task.File == nil || task.File.FilePath() == "" {
+ targetsByDelete[deletePath] = nil
- if _, ok := uniqueDeletes[d.FilePath()]; !ok {
- uniqueDeletes[d.FilePath()] = d
+ continue
}
+ targets[task.File.FilePath()] = struct{}{}
}
}
+ return uniqueDeletes, targetsByDelete
+}
+
+// readAllDeleteFiles reads every referenced positional-delete file up front.
It
+// remains the path used by Transaction.makePositionDeleteRecordsForFilter and
by
+// the eager-vs-lazy benchmark; arrowScan.GetRecords uses
lazyPositionDeleteLoader.
+func readAllDeleteFiles(ctx context.Context, fs iceio.IO, tasks
[]FileScanTask, concurrency int) (perFilePosDeletes, error) {
+ deletesPerFile := make(perFilePosDeletes)
+ uniqueDeletes, targetsByDelete :=
collectPositionDeleteFilesAndTargets(tasks)
+
if len(uniqueDeletes) == 0 {
return deletesPerFile, nil
}
@@ -112,7 +139,7 @@ func readAllDeleteFiles(ctx context.Context, fs iceio.IO,
tasks []FileScanTask,
defer close(perFileChan)
for _, v := range uniqueDeletes {
g.Go(func() error {
- deletes, err := readDeletes(gctx, fs, v)
+ deletes, err := readDeletesForPaths(gctx, fs,
v, targetsByDelete[v.FilePath()])
if err != nil {
return fmt.Errorf("read position
deletes from %s: %w", v.FilePath(), err)
}
@@ -167,6 +194,7 @@ var errPositionDeleteLoaderReleased = errors.New("position
delete loader already
type lazyPositionDeleteFile struct {
dataFile iceberg.DataFile
+ targets map[string]struct{}
once sync.Once
deletes map[string]*arrow.Chunked
@@ -174,21 +202,16 @@ type lazyPositionDeleteFile struct {
}
func newLazyPositionDeleteLoader(fs iceio.IO, tasks []FileScanTask)
*lazyPositionDeleteLoader {
+ uniqueDeletes, targetsByDelete :=
collectPositionDeleteFilesAndTargets(tasks)
loader := &lazyPositionDeleteLoader{
fs: fs,
- files: make(map[string]*lazyPositionDeleteFile),
+ files: make(map[string]*lazyPositionDeleteFile,
len(uniqueDeletes)),
}
- for _, task := range tasks {
- for _, deleteFile := range task.DeleteFiles {
- if deleteFile.ContentType() !=
iceberg.EntryContentPosDeletes {
- continue
- }
-
- path := deleteFile.FilePath()
- if _, ok := loader.files[path]; !ok {
- loader.files[path] =
&lazyPositionDeleteFile{dataFile: deleteFile}
- }
+ for path, deleteFile := range uniqueDeletes {
+ loader.files[path] = &lazyPositionDeleteFile{
+ dataFile: deleteFile,
+ targets: targetsByDelete[path],
}
}
@@ -234,9 +257,9 @@ func (l *lazyPositionDeleteLoader) load(ctx
context.Context, task FileScanTask)
}
cached.once.Do(func() {
- cached.deletes, cached.err = readDeletes(ctx, l.fs,
cached.dataFile)
+ cached.deletes, cached.err = readDeletesForPaths(ctx,
l.fs, cached.dataFile, cached.targets)
if cached.err != nil {
- // readDeletes currently returns nil on errors.
Release defensively
+ // readDeletesForPaths currently returns nil on
errors. Release defensively
// in case a future reader returns partial
Arrow ownership.
releasePosDeletes(cached.deletes)
cached.deletes = nil
@@ -505,12 +528,14 @@ func (c *posDeleteCursor) next() (int64, bool) {
type posDeleteAccumulator struct {
mem memory.Allocator
+ targets map[string]struct{}
builders map[string]*array.Int64Builder
}
-func newPosDeleteAccumulator(ctx context.Context) *posDeleteAccumulator {
+func newPosDeleteAccumulator(ctx context.Context, targets map[string]struct{})
*posDeleteAccumulator {
return &posDeleteAccumulator{
mem: compute.GetAllocator(ctx),
+ targets: targets,
builders: make(map[string]*array.Int64Builder),
}
}
@@ -611,6 +636,12 @@ func (a *posDeleteAccumulator) appendFilePathChunk(ctx
context.Context, filePath
}
path := paths.Value(i)
+ if len(a.targets) > 0 {
+ if _, ok := a.targets[path]; !ok {
+ continue
+ }
+ }
+
builder, ok := a.builders[path]
if !ok {
path = strings.Clone(path)
@@ -649,6 +680,7 @@ func (a *posDeleteAccumulator) appendChunked(ctx
context.Context, filePathCol, p
return ctx.Err()
}
+// appendRecord requires columns projected in file_path, pos order.
func (a *posDeleteAccumulator) appendRecord(ctx context.Context, record
arrow.RecordBatch) error {
if record.NumCols() != 2 {
return fmt.Errorf("%w: projected position delete record has %d
columns, expected 2",
@@ -661,8 +693,15 @@ func (a *posDeleteAccumulator) appendRecord(ctx
context.Context, record arrow.Re
posCol.DataType(), posCol.NullN()); err != nil {
return err
}
+ if err := validatePosDeleteColumnLengths(filePathCol.Len(),
posCol.Len()); err != nil {
+ return err
+ }
- posArr := posCol.(*array.Int64)
+ posArr, ok := posCol.(*array.Int64)
+ if !ok {
+ return fmt.Errorf("%w: unsupported pos record array type %T in
position delete file",
+ iceberg.ErrInvalidSchema, posCol)
+ }
posCursor := posDeleteCursor{chunks: []*array.Int64{posArr}}
if err := a.appendFilePathChunk(ctx, filePathCol, &posCursor); err !=
nil {
return err
@@ -672,7 +711,7 @@ func (a *posDeleteAccumulator) appendRecord(ctx
context.Context, record arrow.Re
}
func groupPosDeletesByFilePath(ctx context.Context, filePathCol, posCol
*arrow.Chunked) (results map[string]*arrow.Chunked, err error) {
- acc := newPosDeleteAccumulator(ctx)
+ acc := newPosDeleteAccumulator(ctx, nil)
defer func() {
if err != nil {
acc.release()
@@ -746,7 +785,13 @@ func releasePosDeletes(deletes map[string]*arrow.Chunked) {
}
}
-func readDeletes(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile)
(_ map[string]*arrow.Chunked, err error) {
+func readDeletes(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile)
(map[string]*arrow.Chunked, error) {
+ return readDeletesForPaths(ctx, fs, dataFile, nil)
+}
+
+func readDeletesForPaths(ctx context.Context, fs iceio.IO, dataFile
iceberg.DataFile,
+ targets map[string]struct{},
+) (_ map[string]*arrow.Chunked, err error) {
src, err := tblutils.GetFile(ctx, fs, dataFile, true)
if err != nil {
return nil, err
@@ -768,7 +813,12 @@ func readDeletes(ctx context.Context, fs iceio.IO,
dataFile iceberg.DataFile) (_
return nil, err
}
- records, err := rdr.GetRecords(ctx, columns, nil)
+ tester, err := newPositionDeleteRowGroupTester(schema, targets)
+ if err != nil {
+ return nil, err
+ }
+
+ records, err := rdr.GetRecords(ctx, columns, tester)
if err != nil {
return nil, err
}
@@ -776,7 +826,7 @@ func readDeletes(ctx context.Context, fs iceio.IO, dataFile
iceberg.DataFile) (_
// string values, so independent dictionaries across batches are safe.
defer records.Release()
- acc := newPosDeleteAccumulator(ctx)
+ acc := newPosDeleteAccumulator(ctx, targets)
defer func() {
// Returning an error assigns the named return value before
deferred
// functions run, which releases builders on every error path.
@@ -800,6 +850,121 @@ func readDeletes(ctx context.Context, fs iceio.IO,
dataFile iceberg.DataFile) (_
return acc.finish(), nil
}
+func newPositionDeleteRowGroupTester(schema *arrow.Schema, targets
map[string]struct{}) (*tblutils.ParquetRowGroupTester, error) {
+ if len(targets) == 0 || len(targets) > inPredicateLimit {
+ return nil, nil
+ }
+ pruningEnabled, err := positionDeletePruningEnabled(schema)
+ if err != nil {
+ return nil, err
+ }
+ if !pruningEnabled {
+ return nil, nil
+ }
+
+ paths := make([]string, 0, len(targets))
+ for path := range targets {
+ paths = append(paths, path)
+ }
+
+ var filter iceberg.BooleanExpression
+ if len(paths) == 1 {
+ // A single target is the common case. EqualTo avoids building
the
+ // set literal used by IsIn and gives the stats/bloom planners
the
+ // simpler predicate directly.
+ filter = iceberg.EqualTo(iceberg.Reference("file_path"),
paths[0])
+ } else {
+ slices.Sort(paths)
+ filter = iceberg.IsIn(iceberg.Reference("file_path"), paths...)
+ }
+ filter, err = iceberg.BindExpr(iceberg.PositionalDeleteSchema, filter,
true)
+ if err != nil {
+ return nil, err
+ }
+
+ statsFn, err :=
newParquetRowGroupStatsEvaluator(iceberg.PositionalDeleteSchema, filter, false)
+ if err != nil {
+ return nil, err
+ }
+ bloomPreds, err := newBloomFilterPredicates(filter)
+ if err != nil {
+ return nil, err
+ }
+
+ return &tblutils.ParquetRowGroupTester{
+ StatsFn: statsFn,
+ BloomPreds: bloomPreds,
+ }, nil
+}
+
+func positionDeletePruningEnabled(schema *arrow.Schema) (bool, error) {
+ // Row-group stats and Bloom predicates are keyed by Parquet physical
field
+ // IDs, while projection resolves these columns by their spec-defined
names.
+ // pqarrow carries the Parquet IDs into Arrow metadata, so only enable
+ // pushdown when those two views agree for the reserved delete columns.
+ physicalIDs := make(map[int]int)
+ var collectIDs func([]arrow.Field)
+ collectIDs = func(fields []arrow.Field) {
+ for _, field := range fields {
+ if id := getFieldID(field); id != nil {
+ physicalIDs[*id]++
+ }
+ if nested, ok := field.Type.(arrow.NestedType); ok {
+ collectIDs(nested.Fields())
+ }
+ }
+ }
+ collectIDs(schema.Fields())
+ if len(physicalIDs) == 0 {
+ // External position-delete files are allowed to omit Iceberg
field IDs.
+ // The name-based projection and row-level target filter remain
safe, but
+ // stats and Bloom pruning cannot be trusted without the IDs.
+ return false, nil
+ }
+
+ deleteFields := iceberg.PositionalDeleteSchema.Fields()
+ for _, field := range deleteFields {
+ if physicalIDs[field.ID] > 1 {
+ return false, fmt.Errorf("%w: position delete field ID
%d is not unique",
+ iceberg.ErrInvalidSchema, field.ID)
+ }
+ }
+
+ pruningEnabled := true
+ for _, want := range deleteFields {
+ indices := schema.FieldIndices(want.Name)
+ if len(indices) != 1 {
+ return false, fmt.Errorf("%w: position delete file must
contain exactly one %q column, found %d",
+ iceberg.ErrInvalidSchema, want.Name,
len(indices))
+ }
+
+ fieldID := getFieldID(schema.Field(indices[0]))
+ if fieldID == nil {
+ if physicalIDs[want.ID] != 0 {
+ return false, fmt.Errorf("%w: position delete
field ID %d is assigned to another column instead of %q",
+ iceberg.ErrInvalidSchema, want.ID,
want.Name)
+ }
+ // Missing IDs only disable pruning. Keep checking the
remaining
+ // columns so this fallback cannot hide an invalid ID
mapping.
+ pruningEnabled = false
+
+ continue
+ }
+ if *fieldID != want.ID {
+ if physicalIDs[want.ID] != 0 {
+ return false, fmt.Errorf("%w: position delete
column %q has field ID %d, want %d; field ID %d is assigned to another column",
+ iceberg.ErrInvalidSchema, want.Name,
*fieldID, want.ID, want.ID)
+ }
+ // A non-canonical ID is safe for name-based reading,
but not for
+ // stats or Bloom pruning. Keep the pre-pruning read
compatible with
+ // external writers that renumber fields.
+ pruningEnabled = false
+ }
+ }
+
+ return pruningEnabled, nil
+}
+
func positionDeleteProjectionIndices(schema *arrow.Schema, reader
tblutils.FileReader) ([]int, error) {
filePathIndex, posIndex, err := positionDeleteColumnIndices(schema)
if err != nil {
diff --git a/table/arrow_scanner_lazy_delete_regression_test.go
b/table/arrow_scanner_lazy_delete_regression_test.go
index ba0dc0bf8..a1ba540e3 100644
--- a/table/arrow_scanner_lazy_delete_regression_test.go
+++ b/table/arrow_scanner_lazy_delete_regression_test.go
@@ -114,17 +114,19 @@ func
TestLazyPositionDeleteLoaderDefersReadsAndSharesResults(t *testing.T) {
ctx := compute.WithAllocator(t.Context(), mem)
const (
- deletePath = "mem://bucket/deletes/shared.parquet"
- dataPathA = "mem://bucket/data/a.parquet"
- dataPathB = "mem://bucket/data/b.parquet"
+ deletePath = "mem://bucket/deletes/shared.parquet"
+ dataPathA = "mem://bucket/data/a.parquet"
+ dataPathB = "mem://bucket/data/b.parquet"
+ unrelatedPath = "mem://bucket/data/unrelated.parquet"
)
memFS := &countingOpenMemFS{MemFS: iceio.NewMemFS()}
writePosDeleteParquetToMemFS(t, memFS.MemFS, deletePath, `[
{"file_path": "`+dataPathA+`", "pos": 1},
- {"file_path": "`+dataPathB+`", "pos": 3}
+ {"file_path": "`+dataPathB+`", "pos": 3},
+ {"file_path": "`+unrelatedPath+`", "pos": 5}
]`)
- deleteFile := newPosDeleteFile(t, deletePath, 2, 128)
+ deleteFile := newPosDeleteFile(t, deletePath, 3, 128)
tasks := []FileScanTask{
{
File: newLazyDataFile(t, dataPathA),
@@ -145,6 +147,12 @@ func
TestLazyPositionDeleteLoaderDefersReadsAndSharesResults(t *testing.T) {
assert.Equal(t, []int64{1}, int64Values(gotA[0]))
assert.Equal(t, int64(1), memFS.opens.Load())
+ cached := loader.files[deletePath]
+ require.NotNil(t, cached)
+ assert.Equal(t, map[string]struct{}{dataPathA: {}, dataPathB: {}},
cached.targets)
+ assert.NotContains(t, cached.deletes, unrelatedPath,
+ "lazy delete reads must filter paths that are not referenced by
scan tasks")
+
gotB, err := loader.load(ctx, tasks[1])
require.NoError(t, err)
require.Len(t, gotB, 1)
diff --git a/table/arrow_scanner_nested_delete_test.go
b/table/arrow_scanner_nested_delete_test.go
new file mode 100644
index 000000000..e976c6b97
--- /dev/null
+++ b/table/arrow_scanner_nested_delete_test.go
@@ -0,0 +1,71 @@
+// 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/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/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestReadDeletesForPathsRejectsNestedBloomFieldIDCollision(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ fields := PositionalDeleteArrowSchema.Fields()
+ fields = append(fields, arrow.Field{Name: "row", Type:
arrow.ListOfField(arrow.Field{
+ Name: "element", Type: arrow.BinaryTypes.String,
+ Metadata: fields[0].Metadata,
+ })})
+ schema := arrow.NewSchema(fields, nil)
+ record := mustLoadRecordBatchFromJSON(schema,
+ `[{"file_path":"data.parquet","pos":1,"row":["unrelated"]}]`)
+ defer record.Release()
+ tbl := array.NewTableFromRecords(schema, []arrow.RecordBatch{record})
+ defer tbl.Release()
+
+ fs := iceio.NewMemFS()
+ const deletePath = "mem://bucket/deletes/nested-bloom.parquet"
+ writer, err := fs.Create(deletePath)
+ require.NoError(t, err)
+ require.NoError(t, pqarrow.WriteTable(tbl, writer, 1,
+ parquet.NewWriterProperties(parquet.WithStats(true),
parquet.WithBloomFilterEnabled(true)),
+ pqarrow.DefaultWriterProps()))
+ require.NoError(t, writer.Close())
+
+ file := newPosDeleteFile(t, deletePath, 1, 128)
+ allDeletes, err := readDeletesForPaths(ctx, fs, file, nil)
+ require.NoError(t, err)
+ defer releasePosDeletes(allDeletes)
+ assert.Equal(t, []int64{1}, int64Values(allDeletes["data.parquet"]))
+
+ filtered, err := readDeletesForPaths(ctx, fs, file,
map[string]struct{}{"data.parquet": {}})
+ defer releasePosDeletes(filtered)
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ assert.Nil(t, filtered)
+}
diff --git a/table/arrow_scanner_posdelete_bench_test.go
b/table/arrow_scanner_posdelete_bench_test.go
index 91dfcaece..6f2073d1a 100644
--- a/table/arrow_scanner_posdelete_bench_test.go
+++ b/table/arrow_scanner_posdelete_bench_test.go
@@ -35,6 +35,111 @@ import (
tblutils "github.com/apache/iceberg-go/table/internal"
)
+func BenchmarkReadDeletesWithFilePathFilter(b *testing.B) {
+ const (
+ numPaths = 1_000
+ rowsPerPath = 1_024
+ )
+
+ memFS, deleteFile, pathNames := benchmarkPositionDeleteFileWithPaths(b,
numPaths, rowsPerPath)
+ ctx := tblutils.WithTableProperties(context.Background(),
iceberg.Properties{
+ ParquetBatchSizeKey: "65536",
+ })
+
+ b.Run("all paths", func(b *testing.B) {
+ benchmarkReadDeletesWithFilePathFilter(b, ctx, memFS,
deleteFile, nil)
+ })
+ for _, targetCount := range []int{1, 10, 100, 200, 201, 1_000} {
+ b.Run(fmt.Sprintf("targets=%d", targetCount), func(b
*testing.B) {
+ targets := make(map[string]struct{}, targetCount)
+ for _, path := range pathNames[:targetCount] {
+ targets[path] = struct{}{}
+ }
+
+ benchmarkReadDeletesWithFilePathFilter(b, ctx, memFS,
deleteFile, targets)
+ })
+ }
+}
+
+func benchmarkReadDeletesWithFilePathFilter(
+ b *testing.B,
+ ctx context.Context,
+ fs iceio.IO,
+ dataFile iceberg.DataFile,
+ targets map[string]struct{},
+) {
+ b.Helper()
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ for b.Loop() {
+ deletes, err := readDeletesForPaths(ctx, fs, dataFile, targets)
+ if err != nil {
+ b.Fatal(err)
+ }
+ releasePosDeletes(deletes)
+ }
+}
+
+func benchmarkPositionDeleteFileWithPaths(b *testing.B, numPaths, rowsPerPath
int) (*iceio.MemFS, iceberg.DataFile, []string) {
+ b.Helper()
+
+ mem := memory.DefaultAllocator
+ pathNames := make([]string, numPaths)
+ for i := range numPaths {
+ pathNames[i] =
fmt.Sprintf("mem://bucket/data/data-%04d.parquet", i)
+ }
+
+ pathBuilder := array.NewStringBuilder(mem)
+ posBuilder := array.NewInt64Builder(mem)
+ for pathIdx, path := range pathNames {
+ for pos := range rowsPerPath {
+ pathBuilder.Append(path)
+ posBuilder.Append(int64(pathIdx*rowsPerPath + pos))
+ }
+ }
+ paths := pathBuilder.NewStringArray()
+ pathBuilder.Release()
+ positions := posBuilder.NewInt64Array()
+ posBuilder.Release()
+ record := array.NewRecordBatch(PositionalDeleteArrowSchema,
+ []arrow.Array{paths, positions}, int64(numPaths*rowsPerPath))
+ paths.Release()
+ positions.Release()
+ defer record.Release()
+ tbl := array.NewTableFromRecords(PositionalDeleteArrowSchema,
[]arrow.RecordBatch{record})
+ defer tbl.Release()
+
+ deletePath := "mem://bucket/deletes/file-path-filter-benchmark.parquet"
+ memFS := iceio.NewMemFS()
+ fw, err := memFS.Create(deletePath)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if err := pqarrow.WriteTable(tbl, fw, int64(rowsPerPath),
+ parquet.NewWriterProperties(
+ parquet.WithStats(true),
+ parquet.WithMaxRowGroupLength(int64(rowsPerPath)),
+ ),
+ pqarrow.DefaultWriterProps()); err != nil {
+ _ = fw.Close()
+ b.Fatal(err)
+ }
+ if err := fw.Close(); err != nil {
+ b.Fatal(err)
+ }
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ deletePath, iceberg.ParquetFile, nil, nil, nil,
+ int64(numPaths*rowsPerPath), 128)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ return memFS, builder.Build(), pathNames
+}
+
func BenchmarkReadDeletesProjected(b *testing.B) {
for _, numRows := range []int{100_000, 1_000_000} {
b.Run(fmt.Sprintf("rows=%d", numRows), func(b *testing.B) {
@@ -47,7 +152,7 @@ func BenchmarkReadDeletesProjected(b *testing.B) {
benchmarkReadDeletes(b, ctx, memFS, dataFile,
readDeletesBefore)
})
b.Run("after", func(b *testing.B) {
- benchmarkReadDeletes(b, ctx, memFS, dataFile,
readDeletes)
+ benchmarkReadDeletes(b, ctx, memFS, dataFile,
readDeletesAfter)
})
})
}
@@ -73,6 +178,10 @@ func benchmarkReadDeletes(
}
}
+func readDeletesAfter(ctx context.Context, fs iceio.IO, dataFile
iceberg.DataFile) (map[string]*arrow.Chunked, error) {
+ return readDeletesForPaths(ctx, fs, dataFile, nil)
+}
+
func benchmarkPositionDeleteFile(b *testing.B, numRows int) (*iceio.MemFS,
iceberg.DataFile) {
b.Helper()
diff --git a/table/arrow_scanner_posdelete_regression_test.go
b/table/arrow_scanner_posdelete_regression_test.go
index ad26269d3..46a2cb5a0 100644
--- a/table/arrow_scanner_posdelete_regression_test.go
+++ b/table/arrow_scanner_posdelete_regression_test.go
@@ -18,7 +18,9 @@
package table
import (
+ "bytes"
"context"
+ "strconv"
"testing"
"github.com/apache/arrow-go/v18/arrow"
@@ -26,6 +28,8 @@ import (
"github.com/apache/arrow-go/v18/arrow/compute"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/file"
+ "github.com/apache/arrow-go/v18/parquet/metadata"
"github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
@@ -91,12 +95,379 @@ func TestReadDeletesRejectsMissingFilePath(t *testing.T) {
pqarrow.DefaultWriterProps()))
require.NoError(t, fw.Close())
- deletes, err := readDeletes(ctx, memFS, newPosDeleteFile(t, deletePath,
1, 128))
+ deletes, err := readDeletesForPaths(ctx, memFS, newPosDeleteFile(t,
deletePath, 1, 128), nil)
require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
assert.Nil(t, deletes)
assert.Contains(t, err.Error(), `exactly one "file_path" column, found
0`)
}
+func TestReadDeletesForPathsFiltersUnneededRows(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ deletePath := "mem://bucket/deletes/filtered.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ otherPath := "mem://bucket/data/unneeded.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFS(t, memFS, deletePath, `[
+ {"file_path": "`+dataPath+`", "pos": 10},
+ {"file_path": "`+otherPath+`", "pos": 20},
+ {"file_path": "`+dataPath+`", "pos": 30}
+ ]`)
+
+ deletes, err := readDeletesForPaths(ctx, memFS, newPosDeleteFile(t,
deletePath, 3, 128), map[string]struct{}{dataPath: {}})
+ require.NoError(t, err)
+ defer releasePosDeletes(deletes)
+
+ assert.Equal(t, []int64{10, 30}, int64Values(deletes[dataPath]))
+ assert.NotContains(t, deletes, otherPath)
+}
+
+func TestReadDeletesForPathsTreatsEmptyTargetsAsUnfiltered(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ deletePath := "mem://bucket/deletes/empty-targets.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ otherPath := "mem://bucket/data/other.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFS(t, memFS, deletePath, `[
+ {"file_path": "`+dataPath+`", "pos": 10},
+ {"file_path": "`+otherPath+`", "pos": 20}
+ ]`)
+
+ for _, targets := range []map[string]struct{}{nil, {}} {
+ deletes, err := readDeletesForPaths(ctx, memFS,
+ newPosDeleteFile(t, deletePath, 2, 128), targets)
+ require.NoError(t, err)
+ assert.Equal(t, []int64{10}, int64Values(deletes[dataPath]))
+ assert.Equal(t, []int64{20}, int64Values(deletes[otherPath]))
+ releasePosDeletes(deletes)
+ }
+}
+
+func TestReadDeletesForPathsMatchesFilePathsExactly(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ deletePath := "mem://bucket/deletes/exact-target.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFS(t, memFS, deletePath, `[
+ {"file_path": "`+dataPath+`", "pos": 10}
+ ]`)
+
+ deletes, err := readDeletesForPaths(ctx, memFS,
+ newPosDeleteFile(t, deletePath, 1, 128),
+ map[string]struct{}{dataPath + "/": {}})
+ require.NoError(t, err)
+ defer releasePosDeletes(deletes)
+ assert.Empty(t, deletes)
+}
+
+func TestReadDeletesForPathsHandlesDictionaryFilePath(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := internal.WithTableProperties(
+ compute.WithAllocator(t.Context(), mem),
+ iceberg.Properties{internal.ParquetBatchSizeKey: "2"},
+ )
+ defer mem.AssertSize(t, 0)
+
+ deletePath := "mem://bucket/deletes/dictionary-filtered.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ otherPath := "mem://bucket/data/unneeded.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFSWithSchema(t, memFS, deletePath,
PositionalDeleteArrowSchema, `[
+ {"file_path": "`+dataPath+`", "pos": 10},
+ {"file_path": "`+otherPath+`", "pos": 20},
+ {"file_path": "`+dataPath+`", "pos": 30},
+ {"file_path": "`+dataPath+`", "pos": 40}
+ ]`)
+
+ deletes, err := readDeletesForPaths(ctx, memFS, newPosDeleteFile(t,
deletePath, 4, 128), map[string]struct{}{dataPath: {}})
+ require.NoError(t, err)
+ defer releasePosDeletes(deletes)
+
+ assert.Equal(t, []int64{10, 30, 40}, int64Values(deletes[dataPath]))
+ assert.NotContains(t, deletes, otherPath)
+}
+
+func TestReadDeletesForPathsHandlesReversedPhysicalSchema(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ deleteSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "pos", Type: arrow.PrimitiveTypes.Int64, Nullable:
false},
+ {Name: "file_path", Type: arrow.BinaryTypes.String, Nullable:
false},
+ }, nil)
+ deletePath := "mem://bucket/deletes/reversed-filtered.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ otherPath := "mem://bucket/data/unneeded.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFSWithSchema(t, memFS, deletePath,
deleteSchema, `[
+ {"pos": 10, "file_path": "`+dataPath+`"},
+ {"pos": 20, "file_path": "`+otherPath+`"},
+ {"pos": 30, "file_path": "`+dataPath+`"}
+ ]`)
+
+ deletes, err := readDeletesForPaths(ctx, memFS, newPosDeleteFile(t,
deletePath, 3, 128), map[string]struct{}{dataPath: {}})
+ require.NoError(t, err)
+ defer releasePosDeletes(deletes)
+
+ assert.Equal(t, []int64{10, 30}, int64Values(deletes[dataPath]))
+ assert.NotContains(t, deletes, otherPath)
+}
+
+func TestReadAllDeleteFilesUsesTaskDataFilePaths(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ deletePath := "mem://bucket/deletes/task-target.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ otherPath := "mem://bucket/data/unneeded.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFS(t, memFS, deletePath, `[
+ {"file_path": "`+dataPath+`", "pos": 10},
+ {"file_path": "`+otherPath+`", "pos": 20}
+ ]`)
+
+ dataBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+ dataPath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+
+ deletes, err := readAllDeleteFiles(ctx, memFS, []FileScanTask{{
+ File: dataBuilder.Build(),
+ DeleteFiles: []iceberg.DataFile{
+ newPosDeleteFile(t, deletePath, 2, 128),
+ },
+ }}, 1)
+ require.NoError(t, err)
+ defer releasePerFilePosDeletes(deletes)
+
+ assert.Equal(t, []int64{10}, int64Values(deletes[dataPath][0]))
+ assert.NotContains(t, deletes, otherPath)
+}
+
+func TestReadAllDeleteFilesFallsBackForMixedTasks(t *testing.T) {
+ for _, missingFile := range []struct {
+ name string
+ file iceberg.DataFile
+ }{
+ {name: "nil file"},
+ {name: "empty path", file: &mockDataFile{}},
+ } {
+ for _, missingFirst := range []bool{false, true} {
+ t.Run(missingFile.name+"/missing
first="+strconv.FormatBool(missingFirst), func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ deletePath :=
"mem://bucket/deletes/mixed-tasks.parquet"
+ dataPath := "mem://bucket/data/data-A.parquet"
+ otherPath := "mem://bucket/data/data-B.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFS(t, memFS,
deletePath, `[
+ {"file_path": "`+dataPath+`", "pos":
10},
+ {"file_path": "`+otherPath+`", "pos":
20}
+ ]`)
+ deleteFile := newPosDeleteFile(t, deletePath,
2, 128)
+ tasks := []FileScanTask{
+ {File: &mockDataFile{path: dataPath},
DeleteFiles: []iceberg.DataFile{deleteFile}},
+ {File: missingFile.file, DeleteFiles:
[]iceberg.DataFile{deleteFile}},
+ }
+ if missingFirst {
+ tasks[0], tasks[1] = tasks[1], tasks[0]
+ }
+
+ deletes, err := readAllDeleteFiles(ctx, memFS,
tasks, 2)
+ require.NoError(t, err)
+ defer releasePerFilePosDeletes(deletes)
+ require.Len(t, deletes, 2)
+ require.Len(t, deletes[dataPath], 1)
+ require.Len(t, deletes[otherPath], 1)
+ assert.Equal(t, []int64{10},
int64Values(deletes[dataPath][0]))
+ assert.Equal(t, []int64{20},
int64Values(deletes[otherPath][0]))
+ })
+ }
+ }
+}
+
+func TestReadDeletesForPathsAtPredicateLimit(t *testing.T) {
+ for _, targetCount := range []int{inPredicateLimit - 1,
inPredicateLimit, inPredicateLimit + 1} {
+ t.Run(strconv.Itoa(targetCount), func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ targets := make(map[string]struct{}, targetCount)
+ for i := range targetCount {
+
targets["mem://bucket/data/target-"+strconv.Itoa(i)+".parquet"] = struct{}{}
+ }
+ tester, err :=
newPositionDeleteRowGroupTester(PositionalDeleteArrowSchema, targets)
+ require.NoError(t, err)
+ if targetCount > inPredicateLimit {
+ assert.Nil(t, tester)
+ } else {
+ require.NotNil(t, tester)
+ }
+
+ deletePath :=
"mem://bucket/deletes/predicate-limit.parquet"
+ dataPath := "mem://bucket/data/target-0.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFS(t, memFS, deletePath, `[
+ {"file_path": "`+dataPath+`", "pos": 10},
+ {"file_path":
"mem://bucket/data/unneeded.parquet", "pos": 20},
+ {"file_path": "`+dataPath+`", "pos": 30}
+ ]`)
+ deletes, err := readDeletesForPaths(ctx, memFS,
newPosDeleteFile(t, deletePath, 3, 128), targets)
+ require.NoError(t, err)
+ defer releasePosDeletes(deletes)
+ require.Len(t, deletes, 1)
+ require.Contains(t, deletes, dataPath)
+ assert.Equal(t, []int64{10, 30},
int64Values(deletes[dataPath]))
+ })
+ }
+}
+
+func TestReadDeletesForPathsFallsBackForPartialFieldIDs(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ missing []int
+ }{
+ {name: "file_path ID absent", missing: []int{0}},
+ {name: "pos ID absent", missing: []int{1}},
+ {name: "IDs only on row", missing: []int{0, 1}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ fields := PositionalDeleteArrowSchema.Fields()
+ for _, index := range tc.missing {
+ fields[index].Metadata = arrow.Metadata{}
+ }
+ fields = append(fields, arrow.Field{
+ Name: "row", Type: arrow.StructOf(arrow.Field{
+ Name: "id", Type:
arrow.PrimitiveTypes.Int64,
+ Metadata:
arrow.MetadataFrom(map[string]string{ArrowParquetFieldIDKey: "101"}),
+ }),
+ Metadata:
arrow.MetadataFrom(map[string]string{ArrowParquetFieldIDKey: "100"}),
+ })
+ schema := arrow.NewSchema(fields, nil)
+ deletePath := "mem://bucket/deletes/partial-ids.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ targets := map[string]struct{}{dataPath: {}}
+ tester, err := newPositionDeleteRowGroupTester(schema,
targets)
+ require.NoError(t, err)
+ assert.Nil(t, tester)
+
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFSWithSchema(t, memFS,
deletePath, schema, `[
+ {"file_path": "`+dataPath+`", "pos": 10, "row":
{"id": 1}},
+ {"file_path":
"mem://bucket/data/unneeded.parquet", "pos": 20, "row": {"id": 2}},
+ {"file_path": "`+dataPath+`", "pos": 30, "row":
{"id": 3}}
+ ]`, parquet.WithBloomFilterEnabled(true))
+ deletes, err := readDeletesForPaths(ctx, memFS,
newPosDeleteFile(t, deletePath, 3, 128), targets)
+ require.NoError(t, err)
+ defer releasePosDeletes(deletes)
+ require.Len(t, deletes, 1)
+ require.Contains(t, deletes, dataPath)
+ assert.Equal(t, []int64{10, 30},
int64Values(deletes[dataPath]))
+ })
+ }
+}
+
+func TestReadDeletesForPathsFallsBackForNoncanonicalFieldIDs(t *testing.T) {
+ filePathField, _ :=
iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ posField, _ := iceberg.PositionalDeleteSchema.FindFieldByName("pos")
+ for _, tc := range []struct {
+ name string
+ filePathID int
+ posID int
+ }{
+ {name: "both IDs renumbered", filePathID: 1, posID: 2},
+ {name: "file_path ID renumbered", filePathID: 1, posID:
posField.ID},
+ {name: "pos ID renumbered", filePathID: filePathField.ID,
posID: 2},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ deletePath :=
"mem://bucket/deletes/noncanonical-ids.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ otherPath := "mem://bucket/data/unneeded.parquet"
+ targets := map[string]struct{}{dataPath: {}}
+ schema :=
positionDeleteSchemaWithFieldIDs(tc.filePathID, tc.posID)
+ tester, err := newPositionDeleteRowGroupTester(schema,
targets)
+ require.NoError(t, err)
+ assert.Nil(t, tester)
+
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFSWithSchema(t, memFS,
deletePath, schema, `[
+ {"file_path": "`+dataPath+`", "pos": 10},
+ {"file_path": "`+otherPath+`", "pos": 20},
+ {"file_path": "`+dataPath+`", "pos": 30}
+ ]`)
+ deletes, err := readDeletesForPaths(ctx, memFS,
+ newPosDeleteFile(t, deletePath, 3, 128),
targets)
+ require.NoError(t, err)
+ defer releasePosDeletes(deletes)
+ assert.Equal(t, []int64{10, 30},
int64Values(deletes[dataPath]))
+ assert.NotContains(t, deletes, otherPath)
+ })
+ }
+}
+
+func TestPositionDeleteRowGroupTesterUsesFilePathStats(t *testing.T) {
+ dataPath := "mem://bucket/data/needed.parquet"
+ otherPath := "mem://bucket/data/unneeded.parquet"
+ rec := mustLoadRecordBatchFromJSON(PositionalDeleteArrowSchema, `[
+ {"file_path": "`+otherPath+`", "pos": 10},
+ {"file_path": "`+otherPath+`", "pos": 20},
+ {"file_path": "`+dataPath+`", "pos": 30},
+ {"file_path": "`+dataPath+`", "pos": 40}
+ ]`)
+ defer rec.Release()
+
+ var buf bytes.Buffer
+ writer, err := pqarrow.NewFileWriter(
+ PositionalDeleteArrowSchema, &buf,
+ parquet.NewWriterProperties(
+ parquet.WithStats(true),
+ parquet.WithMaxRowGroupLength(2),
+ ),
+ pqarrow.DefaultWriterProps(),
+ )
+ require.NoError(t, err)
+ require.NoError(t, writer.Write(rec))
+ require.NoError(t, writer.Close())
+
+ reader, err := file.NewParquetReader(bytes.NewReader(buf.Bytes()))
+ require.NoError(t, err)
+ defer reader.Close()
+ assert.Equal(t, 2, reader.NumRowGroups())
+
+ tester, err :=
newPositionDeleteRowGroupTester(PositionalDeleteArrowSchema,
map[string]struct{}{dataPath: {}})
+ require.NoError(t, err)
+ require.NotNil(t, tester)
+
+ use, err := tester.StatsFn(reader.MetaData().RowGroup(0), []int{0, 1})
+ require.NoError(t, err)
+ assert.False(t, use)
+
+ use, err = tester.StatsFn(reader.MetaData().RowGroup(1), []int{0, 1})
+ require.NoError(t, err)
+ assert.True(t, use)
+}
+
func TestReadDeletesProjectsColumnsAndAccumulatesBatches(t *testing.T) {
mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
ctx := compute.WithAllocator(t.Context(), mem)
@@ -150,7 +521,7 @@ func TestReadDeletesProjectsColumnsAndAccumulatesBatches(t
*testing.T) {
projected.Release()
assert.Equal(t, 3, batchCount)
- deletes, err := readDeletes(ctx, memFS, dataFile)
+ deletes, err := readDeletesForPaths(ctx, memFS, dataFile, nil)
require.NoError(t, err)
defer releasePosDeletes(deletes)
@@ -198,7 +569,7 @@ func TestReadDeletesHandlesDictionaryEncodedFilePath(t
*testing.T) {
require.NoError(t, records.Err())
assert.Greater(t, dictionaryBatches, 0)
- deletes, err := readDeletes(ctx, memFS, dataFile)
+ deletes, err := readDeletesForPaths(ctx, memFS, dataFile, nil)
require.NoError(t, err)
defer releasePosDeletes(deletes)
@@ -225,7 +596,7 @@ func TestReadDeletesHandlesReversedPhysicalSchema(t
*testing.T) {
{"pos": 30, "file_path": "`+dataPath+`"}
]`)
- deletes, err := readDeletes(ctx, memFS, newPosDeleteFile(t, deletePath,
3, 128))
+ deletes, err := readDeletesForPaths(ctx, memFS, newPosDeleteFile(t,
deletePath, 3, 128), nil)
require.NoError(t, err)
defer releasePosDeletes(deletes)
@@ -234,7 +605,7 @@ func TestReadDeletesHandlesReversedPhysicalSchema(t
*testing.T) {
}
func TestPosDeleteAccumulatorFinishAfterReleasePanics(t *testing.T) {
- acc := newPosDeleteAccumulator(t.Context())
+ acc := newPosDeleteAccumulator(t.Context(), nil)
acc.release()
assert.PanicsWithValue(t, "position delete accumulator is already
finished or released", func() {
@@ -242,6 +613,248 @@ func TestPosDeleteAccumulatorFinishAfterReleasePanics(t
*testing.T) {
})
}
+func TestPositionDeleteRowGroupTesterUsesFilePathBloomFilters(t *testing.T) {
+ dataPath := "mem://bucket/data/needed.parquet"
+ otherPath := "mem://bucket/data/unneeded.parquet"
+ deletePath := "mem://bucket/deletes/bloom-filtered.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFSWithSchema(t, memFS, deletePath,
PositionalDeleteArrowSchema, `[
+ {"file_path": "`+otherPath+`", "pos": 10},
+ {"file_path": "`+otherPath+`", "pos": 20},
+ {"file_path": "`+dataPath+`", "pos": 30},
+ {"file_path": "`+dataPath+`", "pos": 40}
+ ]`, parquet.WithStats(false),
parquet.WithBloomFilterEnabledFor("file_path", true),
+ parquet.WithMaxRowGroupLength(2))
+ deleteFile := newPosDeleteFile(t, deletePath, 4, 128)
+
+ for _, useBloom := range []bool{false, true} {
+ t.Run("bloom="+strconv.FormatBool(useBloom), func(t *testing.T)
{
+ ctx := t.Context()
+ source, err := internal.GetFile(ctx, memFS, deleteFile,
true)
+ require.NoError(t, err)
+ reader, err := source.GetReader(ctx)
+ require.NoError(t, err)
+ defer reader.Close()
+ fileMetadata :=
reader.Metadata().(*metadata.FileMetaData)
+ require.Len(t, fileMetadata.RowGroups, 2)
+ schema, err := reader.Schema()
+ require.NoError(t, err)
+ tester, err := newPositionDeleteRowGroupTester(schema,
map[string]struct{}{dataPath: {}})
+ require.NoError(t, err)
+ require.NotNil(t, tester)
+ require.NotEmpty(t, tester.BloomPreds)
+ if !useBloom {
+ tester.BloomPreds = nil
+ }
+ for i := range fileMetadata.RowGroups {
+ use, err :=
tester.StatsFn(fileMetadata.RowGroup(i), []int{0, 1})
+ require.NoError(t, err)
+ require.True(t, use, "statistics must not prune
either row group")
+ }
+
+ var survivors []internal.RowGroupSpan
+ tester.Survivors = &survivors
+ records, err := reader.GetRecords(ctx, []int{0, 1},
tester)
+ require.NoError(t, err)
+ defer records.Release()
+ var positions []int64
+ for records.Next() {
+ positions = append(positions,
records.RecordBatch().Column(1).(*array.Int64).Int64Values()...)
+ }
+ require.NoError(t, records.Err())
+ if useBloom {
+ assert.Equal(t, []int64{30, 40}, positions)
+ assert.Equal(t,
[]internal.RowGroupSpan{{FirstRowPos: 2, NumRows: 2}}, survivors)
+ deletes, err := readDeletesForPaths(ctx, memFS,
deleteFile, map[string]struct{}{dataPath: {}})
+ require.NoError(t, err)
+ defer releasePosDeletes(deletes)
+ require.Len(t, deletes, 1)
+ require.Contains(t, deletes, dataPath)
+ assert.Equal(t, []int64{30, 40},
int64Values(deletes[dataPath]))
+ } else {
+ assert.Equal(t, []int64{10, 20, 30, 40},
positions)
+ assert.Nil(t, survivors)
+ }
+ })
+ }
+}
+
+func TestPositionDeleteRowGroupTesterValidatesPartialFieldIDs(t *testing.T) {
+ for _, missingIndex := range []int{0, 1} {
+ for _, tc := range []struct {
+ name string
+ fields func([]arrow.Field) []arrow.Field
+ wantErr bool
+ }{
+ {
+ name: "reserved ID on another column",
+ wantErr: true,
+ fields: func(fields []arrow.Field)
[]arrow.Field {
+ return append(fields, arrow.Field{
+ Name: "row", Type:
arrow.BinaryTypes.String,
+ Metadata:
PositionalDeleteArrowSchema.Field(missingIndex).Metadata,
+ })
+ },
+ },
+ {
+ name: "duplicate reserved ID",
+ wantErr: true,
+ fields: func(fields []arrow.Field)
[]arrow.Field {
+ return append(fields, arrow.Field{
+ Name: "row", Type:
arrow.BinaryTypes.String,
+ Metadata:
fields[1-missingIndex].Metadata,
+ })
+ },
+ },
+ {
+ name: "noncanonical ID on other delete column",
+ fields: func(fields []arrow.Field)
[]arrow.Field {
+ fields[1-missingIndex].Metadata =
arrow.MetadataFrom(map[string]string{ArrowParquetFieldIDKey: "100"})
+
+ return fields
+ },
+ },
+ } {
+
t.Run("missing="+strconv.Itoa(missingIndex)+"/"+tc.name, func(t *testing.T) {
+ fields := PositionalDeleteArrowSchema.Fields()
+ fields[missingIndex].Metadata = arrow.Metadata{}
+ tester, err := newPositionDeleteRowGroupTester(
+ arrow.NewSchema(tc.fields(fields),
nil), map[string]struct{}{"data.parquet": {}})
+ if !tc.wantErr {
+ require.NoError(t, err)
+ assert.Nil(t, tester)
+
+ return
+ }
+
+ require.ErrorIs(t, err,
iceberg.ErrInvalidSchema)
+ assert.Nil(t, tester)
+ })
+ }
+ }
+}
+
+func TestPositionDeleteRowGroupTesterValidatesPhysicalFieldIDs(t *testing.T) {
+ t.Parallel()
+
+ filePathField, _ :=
iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ posField, _ := iceberg.PositionalDeleteSchema.FindFieldByName("pos")
+ dataPath := "mem://bucket/data/needed.parquet"
+
+ tests := []struct {
+ name string
+ schema *arrow.Schema
+ wantTester bool
+ wantErr string
+ }{
+ {
+ name: "canonical IDs",
+ schema:
positionDeleteSchemaWithFieldIDs(filePathField.ID, posField.ID),
+ wantTester: true,
+ },
+ {
+ name: "IDs absent",
+ schema: positionDeleteSchemaWithoutFieldIDs(),
+ },
+ {
+ name: "noncanonical IDs",
+ schema: positionDeleteSchemaWithFieldIDs(1, 2),
+ },
+ {
+ name: "swapped IDs",
+ schema: positionDeleteSchemaWithFieldIDs(posField.ID,
filePathField.ID),
+ wantErr: `position delete column "file_path" has field
ID`,
+ },
+ {
+ name: "duplicate IDs",
+ schema:
positionDeleteSchemaWithFieldIDs(filePathField.ID, filePathField.ID),
+ wantErr: "is not unique",
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tester, err :=
newPositionDeleteRowGroupTester(tt.schema, map[string]struct{}{dataPath: {}})
+ if tt.wantErr != "" {
+ require.ErrorIs(t, err,
iceberg.ErrInvalidSchema)
+ require.ErrorContains(t, err, tt.wantErr)
+
+ return
+ }
+
+ require.NoError(t, err)
+ if tt.wantTester {
+ require.NotNil(t, tester)
+ assert.NotEmpty(t, tester.BloomPreds)
+ } else {
+ assert.Nil(t, tester)
+ }
+ })
+ }
+}
+
+func TestReadDeletesForPathsRejectsSwappedPhysicalFieldIDs(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ filePathField, _ :=
iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ posField, _ := iceberg.PositionalDeleteSchema.FindFieldByName("pos")
+ deletePath := "mem://bucket/deletes/swapped-ids.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFSWithSchema(t, memFS, deletePath,
+ positionDeleteSchemaWithFieldIDs(posField.ID, filePathField.ID),
+ `[{"file_path": "`+dataPath+`", "pos": 0}]`)
+
+ deletes, err := readDeletesForPaths(ctx, memFS, newPosDeleteFile(t,
deletePath, 1, 128),
+ map[string]struct{}{dataPath: {}})
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ assert.Nil(t, deletes)
+}
+
+func TestReadDeletesForPathsRejectsDuplicatePhysicalFieldIDs(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ defer mem.AssertSize(t, 0)
+
+ filePathField, _ :=
iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ deletePath := "mem://bucket/deletes/duplicate-ids.parquet"
+ dataPath := "mem://bucket/data/needed.parquet"
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFSWithSchema(t, memFS, deletePath,
+ positionDeleteSchemaWithFieldIDs(filePathField.ID,
filePathField.ID),
+ `[{"file_path": "`+dataPath+`", "pos": 0}]`)
+
+ deletes, err := readDeletesForPaths(ctx, memFS, newPosDeleteFile(t,
deletePath, 1, 128),
+ map[string]struct{}{dataPath: {}})
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ assert.Nil(t, deletes)
+}
+
+func positionDeleteSchemaWithFieldIDs(filePathID, posID int) *arrow.Schema {
+ return arrow.NewSchema([]arrow.Field{
+ {
+ Name: "file_path",
+ Type: arrow.BinaryTypes.String,
+ Nullable: false,
+ Metadata:
arrow.MetadataFrom(map[string]string{ArrowParquetFieldIDKey:
strconv.Itoa(filePathID)}),
+ },
+ {
+ Name: "pos",
+ Type: arrow.PrimitiveTypes.Int64,
+ Nullable: false,
+ Metadata:
arrow.MetadataFrom(map[string]string{ArrowParquetFieldIDKey:
strconv.Itoa(posID)}),
+ },
+ }, nil)
+}
+
+func positionDeleteSchemaWithoutFieldIDs() *arrow.Schema {
+ return arrow.NewSchema([]arrow.Field{
+ {Name: "file_path", Type: arrow.BinaryTypes.String, Nullable:
false},
+ {Name: "pos", Type: arrow.PrimitiveTypes.Int64, Nullable:
false},
+ }, nil)
+}
+
func TestGroupPosDeletesByFilePathSupportsStringLayouts(t *testing.T) {
for _, tc := range []struct {
name string
@@ -454,6 +1067,37 @@ func
TestGroupPosDeletesByFilePathRejectsMismatchedLengths(t *testing.T) {
assert.Contains(t, err.Error(), "file_path and pos columns have
different lengths: 2 and 1")
}
+func TestPosDeleteAccumulatorAppendRecordRejectsMismatchedLengths(t
*testing.T) {
+ for _, tc := range []struct {
+ name string
+ filePaths []string
+ positions []int64
+ }{
+ {name: "extra positions", filePaths:
[]string{"file-a.parquet"}, positions: []int64{1, 2}},
+ {name: "extra file paths", filePaths:
[]string{"file-a.parquet", "file-b.parquet"}, positions: []int64{1}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+
+ filePathArr := stringArray(mem, tc.filePaths...)
+ defer filePathArr.Release()
+ posArr := int64Array(mem, tc.positions...)
+ defer posArr.Release()
+ record :=
array.NewRecordBatch(PositionalDeleteArrowSchema,
+ []arrow.Array{filePathArr, posArr},
int64(min(len(tc.filePaths), len(tc.positions))))
+ defer record.Release()
+
+ acc := newPosDeleteAccumulator(ctx, nil)
+ defer acc.release()
+ err := acc.appendRecord(ctx, record)
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ assert.Contains(t, err.Error(), "file_path and pos
columns have different lengths")
+ })
+ }
+}
+
func TestGroupPosDeletesByFilePathRejectsNegativePositions(t *testing.T) {
mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
defer mem.AssertSize(t, 0)
@@ -835,7 +1479,7 @@ func TestReadDeletesRejectsNullPos(t *testing.T) {
pqarrow.DefaultWriterProps()))
require.NoError(t, fw.Close())
- deletes, err := readDeletes(ctx, memFS, newPosDeleteFile(t, deletePath,
1, 128))
+ deletes, err := readDeletesForPaths(ctx, memFS, newPosDeleteFile(t,
deletePath, 1, 128), nil)
require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
assert.Nil(t, deletes)
assert.Contains(t, err.Error(), "null pos in position delete file")
@@ -1091,10 +1735,36 @@ func
TestReadDeletesProjectsLeafColumnsAroundNestedRow(t *testing.T) {
{"file_path": "other.parquet", "pos": 2, "row": {"id": 20,
"name": "b"}},
{"file_path": "`+dataPath+`", "pos": 3, "row": {"id": 30,
"name": "c"}}
]`)
- deletes, err := readDeletes(ctx, memFS,
newPosDeleteFile(t, deletePath, 3, 128))
+ deletes, err := readDeletesForPaths(ctx, memFS,
newPosDeleteFile(t, deletePath, 3, 128), map[string]struct{}{dataPath: {}})
require.NoError(t, err)
defer releasePosDeletes(deletes)
assert.Equal(t, []int64{1, 3},
int64Values(deletes[dataPath]))
})
}
}
+
+func TestPositionDeleteRowGroupTesterRejectsNestedDuplicateFieldIDs(t
*testing.T) {
+ duplicate := arrow.Field{
+ Name: "value", Type: arrow.BinaryTypes.String,
+ Metadata: PositionalDeleteArrowSchema.Field(0).Metadata,
+ }
+ for _, tc := range []struct {
+ name string
+ typ arrow.DataType
+ }{
+ {"struct", arrow.StructOf(duplicate)},
+ {"list", arrow.ListOfField(duplicate)},
+ {"large list", arrow.LargeListOfField(duplicate)},
+ {"map key", arrow.MapOfFields(duplicate, arrow.Field{Name:
"value", Type: arrow.BinaryTypes.String})},
+ {"map value", arrow.MapOfFields(arrow.Field{Name: "key", Type:
arrow.BinaryTypes.String}, duplicate)},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ fields := PositionalDeleteArrowSchema.Fields()
+ fields = append(fields, arrow.Field{Name: "row", Type:
tc.typ})
+ tester, err := newPositionDeleteRowGroupTester(
+ arrow.NewSchema(fields, nil),
map[string]struct{}{"data.parquet": {}})
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ require.Nil(t, tester)
+ })
+ }
+}
diff --git a/table/arrow_scanner_test.go b/table/arrow_scanner_test.go
index e85de0855..c35e41d35 100644
--- a/table/arrow_scanner_test.go
+++ b/table/arrow_scanner_test.go
@@ -305,12 +305,11 @@ func mustLoadRecordBatchFromJSON(schema *arrow.Schema,
content string) arrow.Rec
func writePosDeleteParquetToMemFS(t *testing.T, memFS *iceio.MemFS, path,
content string) {
t.Helper()
-
writePosDeleteParquetToMemFSWithSchema(t, memFS, path,
PositionalDeleteArrowSchema, content)
}
func writePosDeleteParquetToMemFSWithSchema(t *testing.T, memFS *iceio.MemFS,
path string,
- schema *arrow.Schema, content string,
+ schema *arrow.Schema, content string, props ...parquet.WriterProperty,
) {
t.Helper()
@@ -323,8 +322,12 @@ func writePosDeleteParquetToMemFSWithSchema(t *testing.T,
memFS *iceio.MemFS, pa
fw, err := memFS.Create(path)
require.NoError(t, err)
+ props = append([]parquet.WriterProperty{
+ parquet.WithDictionaryDefault(true),
+ parquet.WithStats(true),
+ }, props...)
require.NoError(t, pqarrow.WriteTable(tbl, fw, rec.NumRows(),
-
parquet.NewWriterProperties(parquet.WithDictionaryDefault(true),
parquet.WithStats(true)),
+ parquet.NewWriterProperties(props...),
pqarrow.DefaultWriterProps()))
require.NoError(t, fw.Close())
}
@@ -431,7 +434,7 @@ func TestReleasePerFilePosDeletes(t *testing.T) {
t.Run("nil chunk in slice does not panic", func(t *testing.T) {
// Defensive: production code paths never insert a nil
*arrow.Chunked
// into the map. This subtest pins the guard so a future caller
(the
- // in-flight delete-file merger, or a refactor of readDeletes)
+ // in-flight delete-file merger, or a refactor of
readDeletesForPaths)
// can't silently NPE the cleanup path.
mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
defer mem.AssertSize(t, 0)
