diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go index cda8ad341..f28b3fe56 100644 --- a/table/arrow_scanner.go +++ b/table/arrow_scanner.go @@ -54,15 +54,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 (e.g. when readAllDeletionVectors lands and starts merging // into the same map). @@ -79,6 +80,7 @@ func releasePerFilePosDeletes(deletesPerFile perFilePosDeletes) { func readAllDeleteFiles(ctx context.Context, fs iceio.IO, tasks []FileScanTask, concurrency int) (perFilePosDeletes, error) { deletesPerFile := make(perFilePosDeletes) uniqueDeletes := make(map[string]iceberg.DataFile) + targetsByDelete := make(perDeleteFileTargets) for _, t := range tasks { for _, d := range t.DeleteFiles { @@ -86,9 +88,27 @@ func readAllDeleteFiles(ctx context.Context, fs iceio.IO, tasks []FileScanTask, continue } - if _, ok := uniqueDeletes[d.FilePath()]; !ok { - uniqueDeletes[d.FilePath()] = d + deletePath := d.FilePath() + if _, ok := uniqueDeletes[deletePath]; !ok { + uniqueDeletes[deletePath] = d } + + targets, ok := targetsByDelete[deletePath] + if !ok { + targets = make(map[string]struct{}) + targetsByDelete[deletePath] = targets + } + // A nil target set means that at least one task did not carry a + // usable data-file path. Keep the old whole-file read in that case. + if targets == nil { + continue + } + if t.File == nil || t.File.FilePath() == "" { + targetsByDelete[deletePath] = nil + + continue + } + targets[t.File.FilePath()] = struct{}{} } } @@ -107,7 +127,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 err } @@ -366,12 +386,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), } } @@ -472,6 +494,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) @@ -510,6 +538,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", @@ -522,8 +551,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 @@ -533,7 +569,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() @@ -607,7 +643,9 @@ 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 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 @@ -629,7 +667,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 } @@ -637,7 +680,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. @@ -661,6 +704,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_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 5dbe358ea..249504a77 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()) } @@ -430,7 +433,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 readAllDeletionVectors merger, or a refactor of readDeletes) + // in-flight readAllDeletionVectors merger, or a refactor of readDeletesForPaths) // can't silently NPE the cleanup path. mem := memory.NewCheckedAllocator(memory.NewGoAllocator()) defer mem.AssertSize(t, 0)