laskoviymishka commented on code in PR #1937:
URL: https://github.com/apache/iceberg-go/pull/1937#discussion_r3890696161
##########
table/arrow_scanner.go:
##########
@@ -76,16 +77,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
Review Comment:
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.
##########
table/arrow_scanner.go:
##########
@@ -361,106 +381,199 @@ func (c *posDeleteCursor) next() (int64, bool) {
return pos, true
}
-func groupPosDeletesByFilePath(ctx context.Context, filePathCol, posCol
*arrow.Chunked) (results map[string]*arrow.Chunked, err error) {
- if err := ctx.Err(); err != nil {
- return nil, err
+type posDeleteAccumulator struct {
+ mem memory.Allocator
+ targets map[string]struct{}
+ builders map[string]*array.Int64Builder
+}
+
+func newPosDeleteAccumulator(ctx context.Context, targets map[string]struct{})
*posDeleteAccumulator {
+ return &posDeleteAccumulator{
+ mem: compute.GetAllocator(ctx),
+ targets: targets,
+ builders: make(map[string]*array.Int64Builder),
}
- if filePathCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null file_path in position delete
file", iceberg.ErrInvalidSchema)
+}
+
+func (a *posDeleteAccumulator) release() {
+ for _, builder := range a.builders {
+ builder.Release()
}
- if filePathValueType(filePathCol.DataType()).ID() == arrow.STRING_VIEW {
- return nil, fmt.Errorf("%w: unsupported file_path column type
%s in position delete file",
- iceberg.ErrInvalidSchema, filePathCol.DataType())
+ a.builders = nil
+}
+
+func (a *posDeleteAccumulator) finish() map[string]*arrow.Chunked {
+ if a.builders == nil {
+ panic("position delete accumulator is already finished or
released")
}
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+
+ results := make(map[string]*arrow.Chunked, len(a.builders))
+ for path, builder := range a.builders {
+ positions := builder.NewInt64Array()
+ builder.Release()
+
+ results[path] = arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{positions})
+ positions.Release()
}
- if posCol.DataType().ID() != arrow.INT64 {
- return nil, fmt.Errorf("%w: unsupported pos column type %s in
position delete file",
- iceberg.ErrInvalidSchema, posCol.DataType())
+ a.builders = nil
+
+ return results
+}
+
+func validatePosDeleteColumns(filePathType arrow.DataType, filePathNulls int,
+ posType arrow.DataType, posNulls int,
+) error {
+ if filePathNulls > 0 {
+ return fmt.Errorf("%w: null file_path in position delete file",
iceberg.ErrInvalidSchema)
}
- if filePathCol.Len() != posCol.Len() {
- return nil, fmt.Errorf("%w: file_path and pos columns have
different lengths: %d and %d",
- iceberg.ErrInvalidSchema, filePathCol.Len(),
posCol.Len())
+ if filePathValueType(filePathType).ID() == arrow.STRING_VIEW {
+ return fmt.Errorf("%w: unsupported file_path column type %s in
position delete file",
+ iceberg.ErrInvalidSchema, filePathType)
+ }
+ if posNulls > 0 {
+ return fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ }
+ if posType.ID() != arrow.INT64 {
+ return fmt.Errorf("%w: unsupported pos column type %s in
position delete file",
+ iceberg.ErrInvalidSchema, posType)
}
- mem := compute.GetAllocator(ctx)
- posCursor, err := newPosDeleteCursor(posCol)
+ return nil
+}
+
+func validatePosDeleteColumnLengths(filePathLen, posLen int) error {
+ if filePathLen != posLen {
+ return fmt.Errorf("%w: file_path and pos columns have different
lengths: %d and %d",
+ iceberg.ErrInvalidSchema, filePathLen, posLen)
+ }
+
+ return nil
+}
+
+func (a *posDeleteAccumulator) appendFilePathChunk(ctx context.Context,
filePathChunk arrow.Array,
+ posCursor *posDeleteCursor,
+) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+
+ paths, err := filePathValues(filePathChunk)
if err != nil {
- return nil, err
+ return err
}
- builders := make(map[string]*array.Int64Builder)
- defer func() {
- if err != nil {
- for _, builder := range builders {
- builder.Release()
+ var dictionary arrow.Array
+ var indices *array.Dictionary
+ if dict, ok := filePathChunk.(*array.Dictionary); ok &&
dict.Dictionary().NullN() > 0 {
+ dictionary = dict.Dictionary()
+ indices = dict
+ }
+
+ for i := range filePathChunk.Len() {
+ if i&(positionalDeleteCancellationCheckInterval-1) == 0 {
+ if err := ctx.Err(); err != nil {
+ return err
}
}
- }()
- for _, filePathChunk := range filePathCol.Chunks() {
- if err := ctx.Err(); err != nil {
- return nil, err
+ pos, ok := posCursor.next()
+ if !ok {
+ return fmt.Errorf("%w: position delete columns ended
before file_path column",
+ iceberg.ErrInvalidSchema)
+ }
+ if pos < 0 {
+ return fmt.Errorf("%w: negative pos %d in position
delete file",
+ iceberg.ErrInvalidSchema, pos)
+ }
+ if dictionary != nil &&
dictionary.IsNull(indices.GetValueIndex(i)) {
+ return fmt.Errorf("%w: null file_path dictionary value
in position delete file",
+ iceberg.ErrInvalidSchema)
}
- paths, pathErr := filePathValues(filePathChunk)
- if pathErr != nil {
- return nil, pathErr
+ path := paths.Value(i)
+ if a.targets != nil {
+ if _, ok := a.targets[path]; !ok {
+ continue
+ }
}
- var dictionary arrow.Array
- var indices *array.Dictionary
- if dict, ok := filePathChunk.(*array.Dictionary); ok &&
dict.Dictionary().NullN() > 0 {
- dictionary = dict.Dictionary()
- indices = dict
+ builder, ok := a.builders[path]
+ if !ok {
+ path = strings.Clone(path)
+ builder = array.NewInt64Builder(a.mem)
+ a.builders[path] = builder
}
+ builder.Append(pos)
+ }
- for i := range filePathChunk.Len() {
- if i&(positionalDeleteCancellationCheckInterval-1) == 0
{
- if err := ctx.Err(); err != nil {
- return nil, err
- }
- }
+ return nil
+}
- pos, ok := posCursor.next()
- if !ok {
- return nil, fmt.Errorf("%w: position delete
columns ended before file_path column",
- iceberg.ErrInvalidSchema)
- }
- if pos < 0 {
- return nil, fmt.Errorf("%w: negative pos %d in
position delete file",
- iceberg.ErrInvalidSchema, pos)
- }
- if dictionary != nil &&
dictionary.IsNull(indices.GetValueIndex(i)) {
- return nil, fmt.Errorf("%w: null file_path
dictionary value in position delete file",
- iceberg.ErrInvalidSchema)
- }
+func (a *posDeleteAccumulator) appendChunked(ctx context.Context, filePathCol,
posCol *arrow.Chunked) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if err := validatePosDeleteColumns(filePathCol.DataType(),
filePathCol.NullN(),
+ posCol.DataType(), posCol.NullN()); err != nil {
+ return err
+ }
+ if err := validatePosDeleteColumnLengths(filePathCol.Len(),
posCol.Len()); err != nil {
+ return err
+ }
- path := paths.Value(i)
- builder, ok := builders[path]
- if !ok {
- path = strings.Clone(path)
- builder = array.NewInt64Builder(mem)
- builders[path] = builder
- }
- builder.Append(pos)
+ posCursor, err := newPosDeleteCursor(posCol)
+ if err != nil {
+ return err
+ }
+
+ for _, filePathChunk := range filePathCol.Chunks() {
+ if err := a.appendFilePathChunk(ctx, filePathChunk,
&posCursor); err != nil {
+ return err
}
}
- if err := ctx.Err(); err != nil {
- return nil, err
+
+ return ctx.Err()
+}
+
+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",
+ iceberg.ErrInvalidSchema, record.NumCols())
}
- results = make(map[string]*arrow.Chunked, len(builders))
- for path, builder := range builders {
- positions := builder.NewInt64Array()
- builder.Release()
+ filePathCol := record.Column(0)
Review Comment:
`appendRecord` hard-codes `Column(0)` as `file_path` and `Column(1)` as
`pos`. It's correct today because the only caller passes `[]int{filePathIndex,
posIndex}` to `GetRecords` and pqarrow preserves that order, but the contract
is invisible from in here. A one-line precondition comment, or looking the
columns up by name, would keep a future caller passing a different order from
silently mis-attributing paths and positions.
##########
table/arrow_scanner.go:
##########
@@ -517,29 +636,139 @@ func readDeletes(ctx context.Context, fs iceio.IO,
dataFile iceberg.DataFile) (_
}
defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
+ schema, err := rdr.Schema()
+ if err != nil {
+ return nil, err
+ }
+
+ filePathIndex, posIndex, err := positionDeleteColumnIndices(schema)
if err != nil {
return nil, err
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ tester, err := newPositionDeleteRowGroupTester(schema, targets)
if err != nil {
return nil, err
}
- defer tbl.Release()
- filePathIndex, posIndex, err :=
positionDeleteColumnIndices(tbl.Schema())
+ records, err := rdr.GetRecords(ctx, []int{filePathIndex, posIndex},
tester)
if err != nil {
return nil, err
}
- filePathCol := tbl.Column(filePathIndex).Data()
- posCol := tbl.Column(posIndex).Data()
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ defer records.Release()
+
+ acc := newPosDeleteAccumulator(ctx, targets)
+ defer func() {
+ if err != nil {
+ acc.release()
+ }
+ }()
+
+ for records.Next() {
+ if err := acc.appendRecord(ctx, records.RecordBatch()); err !=
nil {
+ return nil, err
+ }
+ }
+ if err := records.Err(); err != nil {
+ return nil, err
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ return acc.finish(), nil
+}
+
+func newPositionDeleteRowGroupTester(schema *arrow.Schema, targets
map[string]struct{}) (*tblutils.ParquetRowGroupTester, error) {
+ if len(targets) == 0 || len(targets) > inPredicateLimit {
Review Comment:
`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.
##########
table/arrow_scanner.go:
##########
@@ -517,29 +636,139 @@ func readDeletes(ctx context.Context, fs iceio.IO,
dataFile iceberg.DataFile) (_
}
defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
+ schema, err := rdr.Schema()
+ if err != nil {
+ return nil, err
+ }
+
+ filePathIndex, posIndex, err := positionDeleteColumnIndices(schema)
if err != nil {
return nil, err
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ tester, err := newPositionDeleteRowGroupTester(schema, targets)
if err != nil {
return nil, err
}
- defer tbl.Release()
- filePathIndex, posIndex, err :=
positionDeleteColumnIndices(tbl.Schema())
+ records, err := rdr.GetRecords(ctx, []int{filePathIndex, posIndex},
tester)
if err != nil {
return nil, err
}
- filePathCol := tbl.Column(filePathIndex).Data()
- posCol := tbl.Column(posIndex).Data()
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ defer records.Release()
+
+ acc := newPosDeleteAccumulator(ctx, targets)
+ defer func() {
+ if err != nil {
+ acc.release()
+ }
+ }()
+
+ for records.Next() {
+ if err := acc.appendRecord(ctx, records.RecordBatch()); err !=
nil {
+ return nil, err
+ }
+ }
+ if err := records.Err(); err != nil {
+ return nil, err
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ 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)
Review Comment:
`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.
##########
table/arrow_scanner.go:
##########
@@ -517,29 +636,139 @@ func readDeletes(ctx context.Context, fs iceio.IO,
dataFile iceberg.DataFile) (_
}
defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
+ schema, err := rdr.Schema()
+ if err != nil {
+ return nil, err
+ }
+
+ filePathIndex, posIndex, err := positionDeleteColumnIndices(schema)
if err != nil {
return nil, err
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ tester, err := newPositionDeleteRowGroupTester(schema, targets)
if err != nil {
return nil, err
}
- defer tbl.Release()
- filePathIndex, posIndex, err :=
positionDeleteColumnIndices(tbl.Schema())
+ records, err := rdr.GetRecords(ctx, []int{filePathIndex, posIndex},
tester)
if err != nil {
return nil, err
}
- filePathCol := tbl.Column(filePathIndex).Data()
- posCol := tbl.Column(posIndex).Data()
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ defer records.Release()
+
+ acc := newPosDeleteAccumulator(ctx, targets)
+ defer func() {
+ if err != nil {
+ acc.release()
+ }
+ }()
+
+ for records.Next() {
+ if err := acc.appendRecord(ctx, records.RecordBatch()); err !=
nil {
+ return nil, err
+ }
+ }
+ if err := records.Err(); err != nil {
+ return nil, err
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ 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) {
+ physicalIDs := indexArrowFieldsByMetadata(schema)
+ 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
+ }
+
+ filePathField, _ :=
iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
Review Comment:
Both `FindFieldByName` errors get dropped here. Stable today, but if either
name changes under a refactor the zero-value `NestedField` (ID 0) propagates
silently into the ID checks below. I'd assert on the error, or pull these from
named field-ID constants, so it fails loudly instead.
##########
table/arrow_scanner.go:
##########
@@ -517,29 +636,139 @@ func readDeletes(ctx context.Context, fs iceio.IO,
dataFile iceberg.DataFile) (_
}
defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
+ schema, err := rdr.Schema()
+ if err != nil {
+ return nil, err
+ }
+
+ filePathIndex, posIndex, err := positionDeleteColumnIndices(schema)
if err != nil {
return nil, err
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ tester, err := newPositionDeleteRowGroupTester(schema, targets)
if err != nil {
return nil, err
}
- defer tbl.Release()
- filePathIndex, posIndex, err :=
positionDeleteColumnIndices(tbl.Schema())
+ records, err := rdr.GetRecords(ctx, []int{filePathIndex, posIndex},
tester)
if err != nil {
return nil, err
}
- filePathCol := tbl.Column(filePathIndex).Data()
- posCol := tbl.Column(posIndex).Data()
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ defer records.Release()
+
+ acc := newPosDeleteAccumulator(ctx, targets)
+ defer func() {
+ if err != nil {
+ acc.release()
+ }
+ }()
+
+ for records.Next() {
+ if err := acc.appendRecord(ctx, records.RecordBatch()); err !=
nil {
+ return nil, err
+ }
+ }
+ if err := records.Err(); err != nil {
+ return nil, err
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ 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) {
+ physicalIDs := indexArrowFieldsByMetadata(schema)
+ 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
+ }
+
+ filePathField, _ :=
iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ posField, _ := iceberg.PositionalDeleteSchema.FindFieldByName("pos")
+ for _, field := range []iceberg.NestedField{filePathField, posField} {
+ if len(physicalIDs[field.ID]) > 1 {
+ return false, fmt.Errorf("%w: position delete field ID
%d is not unique",
+ iceberg.ErrInvalidSchema, field.ID)
+ }
+ }
+
+ for _, want := range []struct {
+ name string
+ id int
+ }{
+ {name: filePathField.Name, id: filePathField.ID},
+ {name: posField.Name, id: posField.ID},
+ } {
+ 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 {
Review Comment:
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:
```go
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?
##########
table/arrow_scanner_posdelete_bench_test.go:
##########
@@ -27,8 +27,120 @@ import (
"github.com/apache/arrow-go/v18/arrow/compute"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/arrow/scalar"
+ "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"
+ tblutils "github.com/apache/iceberg-go/table/internal"
)
+func BenchmarkReadDeletesWithFilePathFilter(b *testing.B) {
+ const (
+ numPaths = 1_000
+ rowsPerPath = 1_024
+ )
+
+ memFS, deleteFile, pathNames := benchmarkPositionDeleteFile(b,
numPaths, rowsPerPath)
+ ctx := tblutils.WithTableProperties(context.Background(),
iceberg.Properties{
+ ParquetBatchSizeKey: "65536",
+ })
+
+ for _, targetCount := range []int{1, 10, 100, 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{}{}
+ }
+
+ b.Run("all paths", func(b *testing.B) {
Review Comment:
The "all paths" sub-benchmark passes nil and is identical across every
`targetCount` iteration, so it re-measures the same baseline four times.
Hoisting it out of the `targetCount` loop runs it once and de-clutters the
output.
##########
table/arrow_scanner_test.go:
##########
@@ -273,19 +273,24 @@ func mustLoadRecordBatchFromJSON(schema *arrow.Schema,
content string) arrow.Rec
}
func writePosDeleteParquetToMemFS(t *testing.T, memFS *iceio.MemFS, path,
content string) {
+ writePosDeleteParquetToMemFSWithSchema(t, memFS, path,
PositionalDeleteArrowSchema, content)
Review Comment:
The extracted `WithSchema` helper keeps its `t.Helper()`, but this wrapper
dropped its own, so a failure inside the helper now points at this line instead
of the calling test. Adding `t.Helper()` as the first line of the wrapper puts
the attribution back.
--
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]
