You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "fallintoplace (via GitHub)" <gi...@apache.org> on 2026/10/07 11:06:02 UTC
[PR] perf(table): reuse bitmap indexes for positional deletes [iceberg-go]
fallintoplace opened a new pull request, #2146:
URL: https://github.com/apache/iceberg-go/pull/2146
**What**
- Reuse bitmap position-delete indexes across scan splits.
**Why**
- Split tasks for one data file currently rebuild the same position hash set. Large delete sets repeat this work for every split.
**Implementation**
- Keep lazy delete-file reads. Cache per-target Roaring bitmaps.
- Union applicable delete files once per data path and delete-file set.
- Share immutable indexes across split tasks. Keep row-position tracking, `Take` filtering and DV precedence.
- Keep the existing transaction delete path.
**Benchmark**
Apple M1 Pro, darwin/arm64, Go 1.26.3. Median of 3 runs, 300ms each, `-cpu=1`.
100,000 consecutive positions in one in-memory Parquet delete file. Includes delete decoding and index construction for all splits; excludes data-file scanning and row filtering.
| Splits | Hash sets ms/op | Shared bitmap ms/op | B/op before → after | Allocs/op before → after |
| --- | ---: | ---: | ---: | ---: |
| 1 | 8.73 | 7.91 | 15,137,237 → 12,886,688 | 877 → 684 |
| 4 | 14.16 | 7.87 | 21,965,778 → 12,790,773 | 1,653 → 707 |
| 16 | 37.00 | 7.97 | 50,494,199 → 12,679,164 | 4,755 → 802 |
At 16 splits: **4.6x faster setup**, **75% fewer allocated bytes**.
```sh
go test ./table -run '^$' -bench '^BenchmarkPositionDeleteBitmapReuseAcrossSplits$' -benchmem -benchtime=300ms -count=3 -cpu=1
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] perf(table): reuse bitmap indexes for positional deletes [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka merged PR #2146:
URL: https://github.com/apache/iceberg-go/pull/2146
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] perf(table): reuse bitmap indexes for positional deletes [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] perf(table): reuse bitmap indexes for positional deletes [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org