laskoviymishka commented on code in PR #2146:
URL: https://github.com/apache/iceberg-go/pull/2146#discussion_r4242741613
##########
table/arrow_scanner.go:
##########
@@ -2188,7 +2415,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 is the DV-superseded half of the validation thread from last round,
still open. By the time we drop the bitmap here, the loader has already read,
converted and validated the superseded delete file, so a malformed row in a
file a DV supersedes still fails the scan, and we pay the I/O for deletes we
throw away.
Now that the loader owns the key, `positionDeleteIndexKey` can return an
empty key when `len(task.DeletionVectorFiles) > 0` and skip the whole read. I'd
either do that, or add a line saying the eager validation is intentional, and
pin the DV-plus-malformed-superseded case with a test either way. That's the
decision-plus-test I was after in round 1.
##########
table/arrow_scanner.go:
##########
@@ -214,69 +262,113 @@ func newLazyPositionDeleteLoader(fs iceio.IO, tasks
[]FileScanTask) *lazyPositio
targets: targetsByDelete[path],
}
}
+ indexesByKey := make(map[string]*lazyPositionDeleteIndex)
+ for i, task := range tasks {
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
+ if key == "" {
+ continue
+ }
+ index := indexesByKey[key]
+ if index == nil {
+ index = &lazyPositionDeleteIndex{
+ targetPath: targetPath,
+ deletePaths: deletePaths,
+ }
+ indexesByKey[key] = index
+ }
+ loader.indexes[i] = index
+ }
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
+}
+
+// load uses the ordinal in the task slice passed to
newLazyPositionDeleteLoader.
+// Workers preserve that ordinal even when they finish out of order, so this
+// lookup does not rebuild or hash a key for every split.
+func (l *lazyPositionDeleteLoader) load(ctx context.Context, taskIndex int)
(*dv.RoaringPositionBitmap, error) {
Review Comment:
The ordinal is now the only thing tying a task to its index, and nothing
checks it against the task's own file. Today `recordBatchesFromTasksAndDeletes`
passes the same slice it built the loader from, so it holds, but a future
refactor that filters, sorts or limit-trims tasks between construction and the
scan loop would silently hand back another data file's bitmap: wrong rows
dropped, no error. The old `load(task)` was self-validating because it read
`task.File.FilePath()` itself.
I'd pass the target path into `load` and compare it to `index.targetPath`
(one string compare), returning `ErrInvalidArgument` on mismatch, with a test
for the mismatch. Cheap insurance against a silent-wrong-results seam.
##########
table/arrow_scanner.go:
##########
@@ -785,6 +883,64 @@ func releasePosDeletes(deletes map[string]*arrow.Chunked) {
}
}
+func collectPosDeleteBitmap(positionalDeletes positionDeletes)
(*dv.RoaringPositionBitmap, error) {
Review Comment:
The nil / type / null / negative checks here are a near-verbatim copy of
`collectPosDeletePositions`, so the two validators will drift on the next
change. Not blocking, but I'd either factor out one validator both consume, or
move `producePosDeletesFromTask` onto the bitmap and drop the set path, which
was the round-1 cleanup. The doc note pointing at `producePosDeletesFromTask`
helps either way.
##########
table/arrow_scanner.go:
##########
@@ -214,69 +262,113 @@ func newLazyPositionDeleteLoader(fs iceio.IO, tasks
[]FileScanTask) *lazyPositio
targets: targetsByDelete[path],
}
}
+ indexesByKey := make(map[string]*lazyPositionDeleteIndex)
+ for i, task := range tasks {
+ key, targetPath, deletePaths := positionDeleteIndexKey(task)
+ if key == "" {
+ continue
+ }
+ index := indexesByKey[key]
+ if index == nil {
+ index = &lazyPositionDeleteIndex{
+ targetPath: targetPath,
+ deletePaths: deletePaths,
+ }
+ indexesByKey[key] = index
+ }
+ loader.indexes[i] = index
+ }
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
+}
+
+// load uses the ordinal in the task slice passed to
newLazyPositionDeleteLoader.
+// Workers preserve that ordinal even when they finish out of order, so this
+// lookup does not rebuild or hash a key for every split.
+func (l *lazyPositionDeleteLoader) load(ctx context.Context, taskIndex int)
(*dv.RoaringPositionBitmap, error) {
+ if l.released.Load() {
+ return nil, errPositionDeleteLoaderReleased
+ }
+ if taskIndex < 0 || taskIndex >= len(l.indexes) {
+ return nil, fmt.Errorf("%w: position delete task index %d out
of range [0, %d)",
+ iceberg.ErrInvalidArgument, taskIndex, len(l.indexes))
}
- return deletes, nil
+ index := l.indexes[taskIndex]
+ if index == nil {
+ return nil, nil
+ }
+
+ index.once.Do(func() {
+ index.bitmap, index.err = l.buildIndex(ctx, index.targetPath,
index.deletePaths)
+ })
+ if index.err != nil {
+ return nil, index.err
+ }
+
+ return index.bitmap, nil
Review Comment:
With 0 or 1 non-empty inputs this returns the cached per-file bitmap
directly, so several split tasks share one `*RoaringPositionBitmap`, and it's
mutable and not safe for concurrent writes. Nothing enforces read-only use:
`recordsFromTask` takes it right next to `dvBitmap`, so a later edit that runs
`Or` / `Set` / `RunLengthEncode` on it would corrupt other splits, and neither
the compiler nor `-race` would catch it until a test happened to hit the window.
The copy-on-first-merge logic itself is correct, this is just about the
contract. I'd put a doc comment on `load` stating the returned bitmap is shared
and read-only, and ideally a `-race` test where several workers read `Contains`
while another builds a union from the same cached inputs.
--
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]
