laskoviymishka commented on code in PR #1937:
URL: https://github.com/apache/iceberg-go/pull/1937#discussion_r3960850313
##########
table/arrow_scanner.go:
##########
@@ -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) {
Review Comment:
Could we add a short doc comment here on what the nil return means? A nil
tester is "no row-group pruning" (0 or >200 targets, or IDs we can't trust),
not "skip filtering": the row-level filter in the accumulator is still doing
the actual path selection. Easy to misread the nil as bailing out of filtering
on a later read. Non-blocking.
##########
table/arrow_scanner.go:
##########
@@ -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 {
Review Comment:
I'd consider scoping this loop's `pruningEnabled = false` to the `file_path`
column. The predicate we build only touches `file_path`, so a non-canonical
`pos` ID disables row-group pruning even though it can't affect the file_path
stats/bloom check at all.
An external writer that stamps its own ID on `pos` but leaves `file_path`
canonical loses pruning here for no real reason. The first loop already guards
duplicate canonical IDs across both columns, so narrowing this one to
`want.Name == "file_path"` shouldn't weaken any of the safety we care about.
Real Iceberg files always carry canonical IDs so this never fires in
practice, purely a follow-up robustness thing, not something to hold the merge
for. wdyt?
##########
table/arrow_scanner_posdelete_regression_test.go:
##########
@@ -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, {}} {
Review Comment:
I'd wrap these two in `t.Run`: nil and empty-map are semantically distinct
cases here, and a bare-loop failure won't tell us which one broke. Matches the
subtest style the rest of the file already uses. Non-blocking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
