Skip to content

perf(table): filter position-delete reads by file path #1937

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fallintoplace wants to merge 7 commits into
base: main
Choose a base branch
from
Open

perf(table): filter position-delete reads by file path #1937

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
819aaca
perf(table): filter position-delete reads by file path
fallintoplace Aug 28, 2026
b0cc2d1
perf(table): specialize single-target delete pruning
fallintoplace Aug 28, 2026
7b31629
fix(table): guard position delete pruning field IDs
fallintoplace Aug 28, 2026
c90c455
fix(table): resolve position delete columns and nested IDs
fallintoplace Aug 30, 2026
27f5b81
fix(table): fall back for missing position delete field IDs
fallintoplace Aug 30, 2026
89aeb79
fix(table): make position delete pruning fallback conservative
fallintoplace Sep 2, 2026
ab57a44
fix(table): validate streamed position delete batches
fallintoplace Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
182 changes: 170 additions & 12 deletions table/arrow_scanner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand All @@ -79,16 +80,35 @@ 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 {
if d.ContentType() != iceberg.EntryContentPosDeletes {
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

@laskoviymishka laskoviymishka Aug 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The nil sentinel here is the whole-file fallback for the mixed-task case, and it's a different code path from the filtered read, but I don't see a test that exercises it end to end. TestReadAllDeleteFilesUsesTaskDataFilePaths only covers a single task with a valid path.

The construction I'd want: two tasks sharing this delete file, one with File.FilePath() set to data-A.parquet and one with File == nil, then assert readAllDeleteFiles returns rows for both data-A and data-B, i.e. the whole file, not just data-A. That also locks in the ordering here, since the targets == nil check has to stay ahead of the t.File == nil check and nothing tests that today. A refactor that reorders them, or swaps nil for an empty map, would silently drop the other task's deletes with everything still green.


continue
}
targets[t.File.FilePath()] = struct{}{}
}

@zeroshade zeroshade Sep 4, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

minor — Filtering also drops deletes from a delete file not assigned to the task — a semantic change, not just perf

The target set is built only from tasks that reference a given delete file, so rows in that file addressing a data file whose task did not list it are now discarded. Previously the whole file was read and those rows landed in deletesPerFile, where the other task's lookup would apply them. I believe the new behaviour is the spec-correct one (delete-to-data assignment is the planner's job, and applying an unassigned delete is over-deletion), and I confirmed no under-deletion is possible because targets always contains t.File.FilePath() for every task that references the file. Still, the PR is framed as pure perf with an explicit 'keeps the existing result shape' claim, and this changes observable output. Worth a sentence in the PR body and a regression test pinning the intended semantics.

}

Expand All @@ -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
}
Expand Down Expand Up @@ -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),
}
}
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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",
Expand All @@ -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
Expand All @@ -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()
Expand Down Expand Up @@ -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
Expand All @@ -629,15 +667,20 @@ 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
}
// Do not unify dictionaries here: appendFilePathChunk decodes each batch to
// 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.
Expand All @@ -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 {

@laskoviymishka laskoviymishka Aug 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

len(targets) > inPredicateLimit disables the tester, but there's no test at that boundary. I'd add one with inPredicateLimit+1 targets against a delete file that has both target and non-target rows, asserting the non-target rows are still filtered out. That confirms the tester goes nil above the cap and the row-level filter still carries correctness on its own.

return nil, nil

@zeroshade zeroshade Sep 4, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

minor — Field-ID validation is skipped by the early return, so the same corrupt file errors or reads depending on query shape

newPositionDeleteRowGroupTester returns (nil, nil) before calling positionDeletePruningEnabled whenever len(targets)==0 or len(targets)>inPredicateLimit. A delete file with swapped or duplicated reserved field IDs therefore fails with ErrInvalidSchema when a query touches 1..200 data files, but reads successfully when it touches >200, or when any task lacks a usable path and the nil whole-file fallback kicks in. Neither path is unsafe (validation only gates pushdown, and the unvalidated paths do no pruning), so this is a consistency/support concern rather than a correctness one — but a scan that fails only for some query shapes is hard to diagnose. Consider validating unconditionally, or documenting that validation is deliberately scoped to the pushdown path.

}
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)

@laskoviymishka laskoviymishka Aug 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

BloomPreds is asserted non-empty in the field-ID test, but I didn't find a test that writes a delete file with a bloom filter on file_path and confirms a row group actually gets pruned through this path. Worth adding one if it isn't already covered by the data-scan bloom tests, fine to skip if it is.

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 {

@laskoviymishka laskoviymishka Aug 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The guard you added is the right call for the genuinely corrupt cases, but I think this nil branch is a touch too strict.

A delete file from a mixed-version writer that stamps a field ID on some other column (say a v3 row field) but not on file_path/pos lands right here: len(physicalIDs) isn't 0, so the all-absent fallback above doesn't fire, and then fieldID == nil aborts the whole read. Before this PR that file read fine, just without pruning, and Java or PyIceberg would still read it. I'd rather degrade than fail the scan:

if fieldID == nil {
    // IDs present on other columns but not on file_path/pos:
    // fall back to name-based reading with no pruning.
    return false, nil
}

The murkier one is the *fieldID != want.id branch just below. A custom-but-valid writer that maps file_path to its own ID is indistinguishable from a genuinely swapped file at this check, and right now both fail the read even though the first is safe to read by name. I don't think you need to solve that here, but it's worth deciding whether non-canonical positive IDs should also degrade rather than error. wdyt?

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 {
Expand Down
71 changes: 71 additions & 0 deletions table/arrow_scanner_nested_delete_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
}
Loading
Loading