laskoviymishka commented on code in PR #2146:
URL: https://github.com/apache/iceberg-go/pull/2146#discussion_r4232288046
##########
table/arrow_scanner.go:
##########
@@ -214,69 +258,110 @@ func newLazyPositionDeleteLoader(fs iceio.IO, tasks
[]FileScanTask) *lazyPositio
targets: targetsByDelete[path],
}
}
+ for _, task := range tasks {
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
+ if key == "" {
+ continue
+ }
+ if _, ok := loader.indexes[key]; ok {
+ continue
+ }
+ loader.indexes[key] = &lazyPositionDeleteIndex{
+ targetPath: targetPath,
+ deletePaths: deletePaths,
+ }
+ }
return loader
}
-func (l *lazyPositionDeleteLoader) load(ctx context.Context, task
FileScanTask) (positionDeletes, error) {
- if l.released.Load() {
- return nil, errPositionDeleteLoaderReleased
- }
-
- if len(task.DeleteFiles) == 0 {
+func (l *lazyPositionDeleteLoader) loadDeleteFile(
+ ctx context.Context, path string,
+) (*lazyPositionDeleteFile, error) {
+ cached, ok := l.files[path]
+ if !ok {
return nil, nil
}
- targetPath := task.File.FilePath()
- deletes := make(positionDeletes, 0, len(task.DeleteFiles))
- // Most scan tasks carry one positional delete file. Avoid allocating a
- // deduplication map unless there can actually be duplicate entries.
- var seen map[string]struct{}
- if len(task.DeleteFiles) > 1 {
- seen = make(map[string]struct{}, len(task.DeleteFiles))
+ cached.once.Do(func() {
+ cached.bitmaps, cached.err = readPositionDeleteBitmapsForPaths(
+ ctx, l.fs, cached.dataFile, cached.targets)
+ if cached.err != nil {
+ cached.bitmaps = nil
+ cached.err = fmt.Errorf("read position deletes from %s:
%w",
+ cached.dataFile.FilePath(), cached.err)
+ }
+ })
+ if cached.err != nil {
+ return nil, cached.err
}
- for _, deleteFile := range task.DeleteFiles {
- if deleteFile.ContentType() != iceberg.EntryContentPosDeletes {
+
+ return cached, nil
+}
+
+func (l *lazyPositionDeleteLoader) buildIndex(
+ ctx context.Context, targetPath string, deletePaths []string,
+) (*dv.RoaringPositionBitmap, error) {
+ var bitmap *dv.RoaringPositionBitmap
+ owned := false
+ for _, path := range deletePaths {
+ cached, err := l.loadDeleteFile(ctx, path)
+ if err != nil {
+ return nil, err
+ }
+ if cached == nil {
continue
}
- path := deleteFile.FilePath()
- if seen != nil {
- if _, ok := seen[path]; ok {
- continue
- }
- seen[path] = struct{}{}
+ deleteBitmap := cached.bitmaps[targetPath]
+ if deleteBitmap == nil || deleteBitmap.IsEmpty() {
+ continue
}
+ if bitmap == nil {
+ bitmap = deleteBitmap
- cached, ok := l.files[path]
- if !ok {
- // The loader is normally built from the same task
slice supplied to
- // this method. Keep this guard so a malformed caller
cannot panic a
- // scan if it changes a task after loader construction.
continue
}
- cached.once.Do(func() {
- cached.deletes, cached.err = readDeletesForPaths(ctx,
l.fs, cached.dataFile, cached.targets)
- if cached.err != nil {
- // readDeletesForPaths currently returns nil on
errors. Release defensively
- // in case a future reader returns partial
Arrow ownership.
- releasePosDeletes(cached.deletes)
- cached.deletes = nil
- cached.err = fmt.Errorf("read position deletes
from %s: %w",
- cached.dataFile.FilePath(), cached.err)
- }
- })
- if cached.err != nil {
- return nil, cached.err
+ // Detach from the first cached bitmap once, then accumulate
into the
+ // private union. Cached inputs stay immutable for other split
tasks.
+ if !owned {
+ merged := dv.NewRoaringPositionBitmap()
+ merged.Or(bitmap)
+ bitmap = merged
+ owned = true
}
+ bitmap.Or(deleteBitmap)
+ }
- if chunk := cached.deletes[targetPath]; chunk != nil {
- deletes = append(deletes, chunk)
- }
+ return bitmap, nil
+}
+
+func (l *lazyPositionDeleteLoader) load(ctx context.Context, task
FileScanTask) (*dv.RoaringPositionBitmap, error) {
+ if l.released.Load() {
+ return nil, errPositionDeleteLoaderReleased
}
- return deletes, nil
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
Review Comment:
We compute `positionDeleteIndexKey(task)` once per task at construction and
then again here on every `load`, so a split-heavy scan pays the sort + Compact
+ `strings.Builder` cost twice per task, on the exact hot path this PR exists
to cut. I'd cache the key on the task (keyed by `*FileScanTask`, or by task
ordinal) so `load` just looks it up.
If caching the key is awkward, at least cheapen the rebuild: `Grow` the
builder up front, `strconv.AppendInt` into a stack buffer instead of `Itoa`,
and skip the sort/Compact when `len(deletePaths) == 1`, which is the common
single-delete-file case.
##########
table/arrow_scanner.go:
##########
@@ -785,6 +872,64 @@ func releasePosDeletes(deletes map[string]*arrow.Chunked) {
}
}
+func collectPosDeleteBitmap(positionalDeletes positionDeletes)
(*dv.RoaringPositionBitmap, error) {
+ bitmap := dv.NewRoaringPositionBitmap()
+ for _, chunk := range positionalDeletes {
+ if chunk == nil {
+ return nil, fmt.Errorf("%w: nil pos column chunk in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+ if chunk.DataType().ID() != arrow.INT64 {
+ return nil, fmt.Errorf("%w: unsupported pos column type
%s in position delete file",
+ iceberg.ErrInvalidSchema, chunk.DataType())
+ }
+ for _, arr := range chunk.Chunks() {
+ posArr, ok := arr.(*array.Int64)
+ if !ok {
+ return nil, fmt.Errorf("%w: unsupported pos
chunk array type %T in position delete file",
+ iceberg.ErrInvalidSchema, arr)
+ }
+ if posArr.NullN() > 0 {
+ return nil, fmt.Errorf("%w: null pos in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+ for _, position := range posArr.Int64Values() {
+ if position < 0 {
+ return nil, fmt.Errorf("%w: negative
pos %d in position delete file",
+ iceberg.ErrInvalidSchema,
position)
+ }
+ bitmap.Set(uint64(position))
+ }
+ }
+ }
+
+ return bitmap, nil
+}
+
+func readPositionDeleteBitmapsForPaths(
+ ctx context.Context,
+ fs iceio.IO,
+ dataFile iceberg.DataFile,
+ targets map[string]struct{},
+) (map[string]*dv.RoaringPositionBitmap, error) {
+ deletes, err := readDeletesForPaths(ctx, fs, dataFile, targets)
+ if err != nil {
+ return nil, err
+ }
+ defer releasePosDeletes(deletes)
+
+ bitmaps := make(map[string]*dv.RoaringPositionBitmap, len(deletes))
+ for path, positions := range deletes {
+ bitmap, err :=
collectPosDeleteBitmap(positionDeletes{positions})
Review Comment:
This validates every target in the delete file eagerly on first load and
caches the result for all callers, which is a behavior change: a malformed row
that only concerns `b.parquet` now fails a scan of `a.parquet` that merely
shares the delete file, and a pos-delete superseded by a DV gets validated even
though it's never applied. The old path only validated the target the current
task actually consumed.
It's probably mostly theoretical given the planner drops superseded deletes,
but it's a real strictness change with no coverage. I'd add a test with a bad
row for target B and a clean target A and assert scanning A alone still fails,
or decide that's wrong and validate per target-build instead.
##########
table/arrow_scanner.go:
##########
@@ -1058,6 +1203,68 @@ func combinePositionalDeletes(mem memory.Allocator,
deletes set[int64], cursor *
type recProcessFn func(arrow.RecordBatch) (arrow.RecordBatch, error)
+func combinePositionalDeleteBitmap(
Review Comment:
This is the function we swap onto the hot path and it has no direct unit
test, including the `Reserve`/backfill invariant above. I'd cover it directly:
a batch with no deletes returns nil, a delete at row 0, all rows deleted, and
the cursor offset carrying across two batches.
##########
table/arrow_scanner.go:
##########
@@ -1058,6 +1203,68 @@ func combinePositionalDeletes(mem memory.Allocator,
deletes set[int64], cursor *
type recProcessFn func(arrow.RecordBatch) (arrow.RecordBatch, error)
+func combinePositionalDeleteBitmap(
+ mem memory.Allocator,
+ deletes *dv.RoaringPositionBitmap,
+ cursor *rowPositionCursor,
+ nrows int64,
+) arrow.Array {
+ var bldr *array.Int64Builder
+
+ for i := range nrows {
+ if deletes.Contains(uint64(cursor.next())) {
+ if bldr == nil {
+ bldr = array.NewInt64Builder(mem)
+ bldr.Reserve(int(i))
Review Comment:
`Reserve(int(i))` only reserves up to the first deleted row's index, but we
can append up to `nrows` values after it, so a delete at row 0 reserves nothing
and every later append can regrow. I'd reserve the real upper bound:
```suggestion
bldr.Reserve(int(nrows))
```
##########
table/arrow_scanner.go:
##########
@@ -2188,7 +2395,7 @@ func (as *arrowScan) recordsFromTask(ctx context.Context,
task tblutils.Enumerat
// the planner (scanner.go) and Java's DeleteFileIndex, and keeps at
most one
// row-dropping position step in the pipeline so cursors never run over
a
// sequence an earlier step already shortened.
- applyPosDeletes := len(positionalDeletes) > 0 && !hasDV
+ applyPosDeletes := positionalDeleteBitmap != nil &&
!positionalDeleteBitmap.IsEmpty() && !hasDV
Review Comment:
This `!hasDV` gate is where DV precedence over position deletes lives, and
nothing tests it directly. I'd add a task that carries both a DV and a
position-delete bitmap and assert the position step is skipped, so a future
change to this condition can't silently resurrect superseded deletes.
##########
table/arrow_scanner.go:
##########
@@ -214,69 +258,110 @@ func newLazyPositionDeleteLoader(fs iceio.IO, tasks
[]FileScanTask) *lazyPositio
targets: targetsByDelete[path],
}
}
+ for _, task := range tasks {
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
+ if key == "" {
+ continue
+ }
+ if _, ok := loader.indexes[key]; ok {
+ continue
+ }
+ loader.indexes[key] = &lazyPositionDeleteIndex{
+ targetPath: targetPath,
+ deletePaths: deletePaths,
+ }
+ }
return loader
}
-func (l *lazyPositionDeleteLoader) load(ctx context.Context, task
FileScanTask) (positionDeletes, error) {
- if l.released.Load() {
- return nil, errPositionDeleteLoaderReleased
- }
-
- if len(task.DeleteFiles) == 0 {
+func (l *lazyPositionDeleteLoader) loadDeleteFile(
+ ctx context.Context, path string,
+) (*lazyPositionDeleteFile, error) {
+ cached, ok := l.files[path]
+ if !ok {
return nil, nil
}
- targetPath := task.File.FilePath()
- deletes := make(positionDeletes, 0, len(task.DeleteFiles))
- // Most scan tasks carry one positional delete file. Avoid allocating a
- // deduplication map unless there can actually be duplicate entries.
- var seen map[string]struct{}
- if len(task.DeleteFiles) > 1 {
- seen = make(map[string]struct{}, len(task.DeleteFiles))
+ cached.once.Do(func() {
+ cached.bitmaps, cached.err = readPositionDeleteBitmapsForPaths(
+ ctx, l.fs, cached.dataFile, cached.targets)
+ if cached.err != nil {
+ cached.bitmaps = nil
+ cached.err = fmt.Errorf("read position deletes from %s:
%w",
+ cached.dataFile.FilePath(), cached.err)
+ }
+ })
+ if cached.err != nil {
+ return nil, cached.err
}
- for _, deleteFile := range task.DeleteFiles {
- if deleteFile.ContentType() != iceberg.EntryContentPosDeletes {
+
+ return cached, nil
+}
+
+func (l *lazyPositionDeleteLoader) buildIndex(
+ ctx context.Context, targetPath string, deletePaths []string,
+) (*dv.RoaringPositionBitmap, error) {
+ var bitmap *dv.RoaringPositionBitmap
+ owned := false
+ for _, path := range deletePaths {
+ cached, err := l.loadDeleteFile(ctx, path)
+ if err != nil {
+ return nil, err
+ }
+ if cached == nil {
continue
}
- path := deleteFile.FilePath()
- if seen != nil {
- if _, ok := seen[path]; ok {
- continue
- }
- seen[path] = struct{}{}
+ deleteBitmap := cached.bitmaps[targetPath]
+ if deleteBitmap == nil || deleteBitmap.IsEmpty() {
+ continue
}
+ if bitmap == nil {
+ bitmap = deleteBitmap
- cached, ok := l.files[path]
- if !ok {
- // The loader is normally built from the same task
slice supplied to
- // this method. Keep this guard so a malformed caller
cannot panic a
- // scan if it changes a task after loader construction.
continue
}
- cached.once.Do(func() {
- cached.deletes, cached.err = readDeletesForPaths(ctx,
l.fs, cached.dataFile, cached.targets)
- if cached.err != nil {
- // readDeletesForPaths currently returns nil on
errors. Release defensively
- // in case a future reader returns partial
Arrow ownership.
- releasePosDeletes(cached.deletes)
- cached.deletes = nil
- cached.err = fmt.Errorf("read position deletes
from %s: %w",
- cached.dataFile.FilePath(), cached.err)
- }
- })
- if cached.err != nil {
- return nil, cached.err
+ // Detach from the first cached bitmap once, then accumulate
into the
+ // private union. Cached inputs stay immutable for other split
tasks.
+ if !owned {
Review Comment:
The ownership logic here (alias the first non-empty bitmap, clone on the
second merge) is the subtle part, and the regression test only builds the union
in one order. I'd add a case where the first delete file is empty for the
target but later ones aren't, which exercises the empty-skip before the `owned`
flip, then assert a sibling task's cached `files[B].bitmaps[data]` is untouched
after the merge. That's the exact shape a later "OR into the first bitmap"
refactor would silently break.
##########
table/arrow_scanner.go:
##########
@@ -785,6 +872,64 @@ func releasePosDeletes(deletes map[string]*arrow.Chunked) {
}
}
+func collectPosDeleteBitmap(positionalDeletes positionDeletes)
(*dv.RoaringPositionBitmap, error) {
Review Comment:
With this path in, `collectPosDeletePositions` / `combinePositionalDeletes`
/ `processPositionalDeletes` are reachable only from tests and benchmarks now,
so we have two parallel implementations in the production file and the live
path's error cases (null / negative / wrong-type pos) are the ones without
coverage. I'd migrate those error-case tests onto `collectPosDeleteBitmap` and
drop the dead helpers, rather than keep tests exercising code the scan no
longer calls.
##########
table/arrow_scanner.go:
##########
@@ -214,69 +258,110 @@ func newLazyPositionDeleteLoader(fs iceio.IO, tasks
[]FileScanTask) *lazyPositio
targets: targetsByDelete[path],
}
}
+ for _, task := range tasks {
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
+ if key == "" {
+ continue
+ }
+ if _, ok := loader.indexes[key]; ok {
+ continue
+ }
+ loader.indexes[key] = &lazyPositionDeleteIndex{
+ targetPath: targetPath,
+ deletePaths: deletePaths,
+ }
+ }
return loader
}
-func (l *lazyPositionDeleteLoader) load(ctx context.Context, task
FileScanTask) (positionDeletes, error) {
- if l.released.Load() {
- return nil, errPositionDeleteLoaderReleased
- }
-
- if len(task.DeleteFiles) == 0 {
+func (l *lazyPositionDeleteLoader) loadDeleteFile(
+ ctx context.Context, path string,
+) (*lazyPositionDeleteFile, error) {
+ cached, ok := l.files[path]
+ if !ok {
return nil, nil
}
- targetPath := task.File.FilePath()
- deletes := make(positionDeletes, 0, len(task.DeleteFiles))
- // Most scan tasks carry one positional delete file. Avoid allocating a
- // deduplication map unless there can actually be duplicate entries.
- var seen map[string]struct{}
- if len(task.DeleteFiles) > 1 {
- seen = make(map[string]struct{}, len(task.DeleteFiles))
+ cached.once.Do(func() {
+ cached.bitmaps, cached.err = readPositionDeleteBitmapsForPaths(
+ ctx, l.fs, cached.dataFile, cached.targets)
+ if cached.err != nil {
+ cached.bitmaps = nil
+ cached.err = fmt.Errorf("read position deletes from %s:
%w",
+ cached.dataFile.FilePath(), cached.err)
+ }
+ })
+ if cached.err != nil {
+ return nil, cached.err
}
- for _, deleteFile := range task.DeleteFiles {
- if deleteFile.ContentType() != iceberg.EntryContentPosDeletes {
+
+ return cached, nil
+}
+
+func (l *lazyPositionDeleteLoader) buildIndex(
+ ctx context.Context, targetPath string, deletePaths []string,
+) (*dv.RoaringPositionBitmap, error) {
+ var bitmap *dv.RoaringPositionBitmap
+ owned := false
+ for _, path := range deletePaths {
+ cached, err := l.loadDeleteFile(ctx, path)
+ if err != nil {
+ return nil, err
+ }
+ if cached == nil {
continue
}
- path := deleteFile.FilePath()
- if seen != nil {
- if _, ok := seen[path]; ok {
- continue
- }
- seen[path] = struct{}{}
+ deleteBitmap := cached.bitmaps[targetPath]
+ if deleteBitmap == nil || deleteBitmap.IsEmpty() {
+ continue
}
+ if bitmap == nil {
+ bitmap = deleteBitmap
- cached, ok := l.files[path]
- if !ok {
- // The loader is normally built from the same task
slice supplied to
- // this method. Keep this guard so a malformed caller
cannot panic a
- // scan if it changes a task after loader construction.
continue
}
- cached.once.Do(func() {
- cached.deletes, cached.err = readDeletesForPaths(ctx,
l.fs, cached.dataFile, cached.targets)
- if cached.err != nil {
- // readDeletesForPaths currently returns nil on
errors. Release defensively
- // in case a future reader returns partial
Arrow ownership.
- releasePosDeletes(cached.deletes)
- cached.deletes = nil
- cached.err = fmt.Errorf("read position deletes
from %s: %w",
- cached.dataFile.FilePath(), cached.err)
- }
- })
- if cached.err != nil {
- return nil, cached.err
+ // Detach from the first cached bitmap once, then accumulate
into the
+ // private union. Cached inputs stay immutable for other split
tasks.
+ if !owned {
+ merged := dv.NewRoaringPositionBitmap()
+ merged.Or(bitmap)
+ bitmap = merged
+ owned = true
}
+ bitmap.Or(deleteBitmap)
+ }
- if chunk := cached.deletes[targetPath]; chunk != nil {
- deletes = append(deletes, chunk)
- }
+ return bitmap, nil
+}
+
+func (l *lazyPositionDeleteLoader) load(ctx context.Context, task
FileScanTask) (*dv.RoaringPositionBitmap, error) {
+ if l.released.Load() {
+ return nil, errPositionDeleteLoaderReleased
}
- return deletes, nil
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
+ if key == "" {
+ return nil, nil
+ }
+
+ index := l.indexes[key]
+ if index == nil {
+ // The loader is normally built from the same task slice
supplied here.
+ // Keep mutated or hand-built tasks safe without mutating the
shared cache.
+ return l.buildIndex(ctx, targetPath, deletePaths)
Review Comment:
Small thing: this fallback builds the union without the per-index `once`, so
concurrent callers for an unknown key each build their own merged bitmap.
Delete-file reads stay single-flighted through `loadDeleteFile`, so it's not a
correctness issue. Worth a one-line comment that the branch is intentionally
un-cached for hand-built tasks, so nobody "fixes" it into the shared cache
later.
##########
table/arrow_scanner.go:
##########
@@ -214,69 +258,110 @@ func newLazyPositionDeleteLoader(fs iceio.IO, tasks
[]FileScanTask) *lazyPositio
targets: targetsByDelete[path],
}
}
+ for _, task := range tasks {
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
+ if key == "" {
+ continue
+ }
+ if _, ok := loader.indexes[key]; ok {
+ continue
+ }
+ loader.indexes[key] = &lazyPositionDeleteIndex{
+ targetPath: targetPath,
+ deletePaths: deletePaths,
+ }
+ }
return loader
}
-func (l *lazyPositionDeleteLoader) load(ctx context.Context, task
FileScanTask) (positionDeletes, error) {
- if l.released.Load() {
- return nil, errPositionDeleteLoaderReleased
- }
-
- if len(task.DeleteFiles) == 0 {
+func (l *lazyPositionDeleteLoader) loadDeleteFile(
+ ctx context.Context, path string,
+) (*lazyPositionDeleteFile, error) {
+ cached, ok := l.files[path]
+ if !ok {
return nil, nil
}
- targetPath := task.File.FilePath()
- deletes := make(positionDeletes, 0, len(task.DeleteFiles))
- // Most scan tasks carry one positional delete file. Avoid allocating a
- // deduplication map unless there can actually be duplicate entries.
- var seen map[string]struct{}
- if len(task.DeleteFiles) > 1 {
- seen = make(map[string]struct{}, len(task.DeleteFiles))
+ cached.once.Do(func() {
+ cached.bitmaps, cached.err = readPositionDeleteBitmapsForPaths(
+ ctx, l.fs, cached.dataFile, cached.targets)
+ if cached.err != nil {
+ cached.bitmaps = nil
+ cached.err = fmt.Errorf("read position deletes from %s:
%w",
+ cached.dataFile.FilePath(), cached.err)
+ }
+ })
+ if cached.err != nil {
+ return nil, cached.err
}
- for _, deleteFile := range task.DeleteFiles {
- if deleteFile.ContentType() != iceberg.EntryContentPosDeletes {
+
+ return cached, nil
+}
+
+func (l *lazyPositionDeleteLoader) buildIndex(
Review Comment:
Not a code change, but worth a sentence somewhere here: for 2+ applicable
delete files we hold every per-file bitmap in `files[*].bitmaps` plus a merged
copy per distinct (data file, delete set) until `release()`, so peak is roughly
O(targets x files). Roaring is usually far smaller than the old hash set so we
likely net ahead, but the trade-off is worth writing down.
--
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]
