This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new ff7df2e56 perf(table): reuse bitmap indexes for positional deletes
(#2146)
ff7df2e56 is described below
commit ff7df2e56ed3391e810281b772bf27ea74d03256
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 12 21:23:20 2026 +0200
perf(table): reuse bitmap indexes for positional deletes (#2146)
---
table/arrow_scanner.go | 384 ++++++++++++++++----
table/arrow_scanner_lazy_delete_bench_test.go | 220 +++++++++++-
table/arrow_scanner_lazy_delete_regression_test.go | 397 +++++++++++++++++++--
table/arrow_scanner_posdelete_bitmap_test.go | 282 +++++++++++++++
table/dv_scanner_end_to_end_test.go | 32 ++
5 files changed, 1217 insertions(+), 98 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index a482a2964..935a949eb 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -172,19 +172,19 @@ func readAllDeleteFiles(ctx context.Context, fs iceio.IO,
tasks []FileScanTask,
// lazyPositionDeleteLoader indexes positional-delete metadata for a scan, but
// waits to open each delete file until a worker reaches a task that references
-// it. A delete file can apply to more than one data file, so the cache keeps
-// the complete grouped result for the delete file rather than caching only one
-// task's positions.
+// it. Delete-file reads are cached by delete path, while the merged position
+// bitmap is cached by target data file plus its delete-file set. Split tasks
+// therefore share one immutable bitmap instead of rebuilding a hash set per
+// task. Unions of multiple non-empty inputs are retained alongside the
+// per-file bitmaps until release.
//
-// The grouped Arrow chunks are owned by the loader until release. The iterator
-// calls release after all workers have stopped, which keeps shared chunks
alive
-// while multiple tasks use them and also covers early iterator termination.
-// The loader has the lifetime of exactly one scan. Each file's first load also
-// locks in its result, including context errors, for every caller; a loader
-// must not be reused for a retry with a different context.
+// The loader has the lifetime of exactly one scan. Each delete file and merged
+// bitmap locks in its first result, including context errors, for every
caller;
+// a loader must not be reused for a retry with a different context.
type lazyPositionDeleteLoader struct {
- fs iceio.IO
- files map[string]*lazyPositionDeleteFile
+ fs iceio.IO
+ files map[string]*lazyPositionDeleteFile
+ indexes []positionDeleteTaskIndex
releaseOnce sync.Once
released atomic.Bool
@@ -197,15 +197,73 @@ type lazyPositionDeleteFile struct {
targets map[string]struct{}
once sync.Once
- deletes map[string]*arrow.Chunked
+ bitmaps map[string]*dv.RoaringPositionBitmap
err error
}
+type lazyPositionDeleteIndex struct {
+ targetPath string
+ deletePaths []string
+
+ once sync.Once
+ bitmap *dv.RoaringPositionBitmap
+ err error
+}
+
+type positionDeleteTaskIndex struct {
+ targetPath string
+ index *lazyPositionDeleteIndex
+}
+
+func positionDeleteIndexKey(task FileScanTask) (key, targetPath string,
deletePaths []string) {
+ if task.File == nil {
+ return "", "", nil
+ }
+ targetPath = task.File.FilePath()
+ if targetPath == "" {
+ return "", "", nil
+ }
+ // An applicable DV contains all earlier position deletes for this file.
+ // Skip superseded files before loading or validating their contents.
+ if len(task.DeletionVectorFiles) > 0 {
+ return "", targetPath, nil
+ }
+
+ for _, deleteFile := range task.DeleteFiles {
+ if deleteFile.ContentType() == iceberg.EntryContentPosDeletes {
+ deletePaths = append(deletePaths, deleteFile.FilePath())
+ }
+ }
+ if len(deletePaths) == 0 {
+ return "", targetPath, nil
+ }
+ slices.Sort(deletePaths)
+ deletePaths = slices.Compact(deletePaths)
+
+ var builder strings.Builder
+ writePart := func(part string) {
+ builder.WriteString(strconv.Itoa(len(part)))
+ builder.WriteByte(':')
+ builder.WriteString(part)
+ }
+ writePart(targetPath)
+ for _, path := range deletePaths {
+ builder.WriteByte('|')
+ writePart(path)
+ }
+
+ return builder.String(), targetPath, deletePaths
+}
+
func newLazyPositionDeleteLoader(fs iceio.IO, tasks []FileScanTask)
*lazyPositionDeleteLoader {
uniqueDeletes, targetsByDelete :=
collectPositionDeleteFilesAndTargets(tasks)
+ if len(uniqueDeletes) == 0 {
+ return nil
+ }
loader := &lazyPositionDeleteLoader{
- fs: fs,
- files: make(map[string]*lazyPositionDeleteFile,
len(uniqueDeletes)),
+ fs: fs,
+ files: make(map[string]*lazyPositionDeleteFile,
len(uniqueDeletes)),
+ indexes: make([]positionDeleteTaskIndex, len(tasks)),
}
for path, deleteFile := range uniqueDeletes {
@@ -214,69 +272,124 @@ 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)
+ loader.indexes[i].targetPath = targetPath
+ if key == "" {
+ continue
+ }
+ index := indexesByKey[key]
+ if index == nil {
+ index = &lazyPositionDeleteIndex{
+ targetPath: targetPath,
+ deletePaths: deletePaths,
+ }
+ indexesByKey[key] = index
+ }
+ loader.indexes[i].index = 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. The target path
rejects
+// ordinals paired with another data file after reordering or filtering tasks,
+// including ordinals that originally had no position deletes.
+//
+// The returned bitmap is shared and read-only. Callers must not mutate it with
+// Set, Or, or RunLengthEncode; other split tasks may be reading the same
bitmap.
+func (l *lazyPositionDeleteLoader) load(ctx context.Context, taskIndex int,
targetPath string) (*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
+ task := l.indexes[taskIndex]
+ if task.targetPath != targetPath {
+ return nil, fmt.Errorf("%w: position delete task index %d
targets %q, got %q",
+ iceberg.ErrInvalidArgument, taskIndex, task.targetPath,
targetPath)
+ }
+ index := task.index
+ 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
}
func (l *lazyPositionDeleteLoader) release() {
@@ -287,8 +400,12 @@ func (l *lazyPositionDeleteLoader) release() {
l.releaseOnce.Do(func() {
l.released.Store(true)
for _, cached := range l.files {
- releasePosDeletes(cached.deletes)
- cached.deletes = nil
+ cached.bitmaps = nil
+ }
+ for _, task := range l.indexes {
+ if task.index != nil {
+ task.index.bitmap = nil
+ }
}
})
}
@@ -727,6 +844,8 @@ func groupPosDeletesByFilePath(ctx context.Context,
filePathCol, posCol *arrow.C
const maxPositionalDeletePreallocation = 64 * 1024
+// collectPosDeletePositions serves producePosDeletesFromTask when a
+// transaction creates new delete records. Scan tasks use cached bitmaps
instead.
func collectPosDeletePositions(positionalDeletes positionDeletes) (set[int64],
error) {
totalPositions := 0
for _, chunk := range positionalDeletes {
@@ -785,6 +904,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})
+ if err != nil {
+ return nil, err
+ }
+ bitmaps[path] = bitmap
+ }
+
+ return bitmaps, nil
+}
+
func readDeletes(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile)
(map[string]*arrow.Chunked, error) {
return readDeletesForPaths(ctx, fs, dataFile, nil)
}
@@ -1063,6 +1240,77 @@ 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)
+ if i > 0 {
+ // This row is deleted. Resize avoids
rounding the maximum
+ // survivor count up to a power of two
as Reserve would.
+ bldr.Resize(int(nrows - 1))
+ }
+ for j := range i {
+ bldr.UnsafeAppend(j)
+ }
+ }
+
+ continue
+ }
+
+ if bldr != nil {
+ // A leading run of deletes needs no values buffer
until a row
+ // survives. Size it once for the remaining possible
survivors.
+ if bldr.Cap() == 0 {
+ bldr.Resize(int(nrows - i))
+ }
+ bldr.UnsafeAppend(i)
+ }
+ }
+
+ if bldr == nil {
+ return nil
+ }
+ defer bldr.Release()
+
+ return bldr.NewArray()
+}
+
+func processPositionalDeleteBitmap(
+ ctx context.Context,
+ deletes *dv.RoaringPositionBitmap,
+ cursor *rowPositionCursor,
+) recProcessFn {
+ mem := compute.GetAllocator(ctx)
+
+ return func(r arrow.RecordBatch) (arrow.RecordBatch, error) {
+ defer r.Release()
+
+ indices := combinePositionalDeleteBitmap(mem, deletes, cursor,
r.NumRows())
+ if indices == nil {
+ r.Retain()
+
+ return r, nil
+ }
+ defer indices.Release()
+
+ out, err := compute.Take(ctx, *compute.DefaultTakeOptions(),
+ compute.NewDatumWithoutOwning(r),
compute.NewDatumWithoutOwning(indices))
+ if err != nil {
+ return nil, err
+ }
+
+ return out.(*compute.RecordDatum).Value, nil
+ }
+}
+
func processPositionalDeletes(ctx context.Context, deletes set[int64], cursor
*rowPositionCursor) recProcessFn {
mem := compute.GetAllocator(ctx)
@@ -2154,7 +2402,7 @@ func pruningFilterHasMissingInitialDefault(
return false, nil
}
-func (as *arrowScan) recordsFromTask(ctx context.Context, task
tblutils.Enumerated[FileScanTask], sink *recordSink, positionalDeletes
positionDeletes, dvBitmap *dv.RoaringPositionBitmap, eqDeleteSets
[]*equalityDeleteSet, invariants *arrowScanInvariants) (err error) {
+func (as *arrowScan) recordsFromTask(ctx context.Context, task
tblutils.Enumerated[FileScanTask], sink *recordSink, positionalDeleteBitmap
*dv.RoaringPositionBitmap, dvBitmap *dv.RoaringPositionBitmap, eqDeleteSets
[]*equalityDeleteSet, invariants *arrowScanInvariants) (err error) {
defer func() {
if err != nil {
sink.fail(task, err)
@@ -2215,7 +2463,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
var posSource *rowPositionSource
if synthesizeRowID || applyPosDeletes || hasDV {
posSource = &rowPositionSource{}
@@ -2239,12 +2487,8 @@ func (as *arrowScan) recordsFromTask(ctx
context.Context, task tblutils.Enumerat
}
if applyPosDeletes {
- deletes, err := collectPosDeletePositions(positionalDeletes)
- if err != nil {
- return err
- }
-
- pipeline = append(pipeline, processPositionalDeletes(ctx,
deletes, posSource.cursor()))
+ pipeline = append(pipeline,
+ processPositionalDeleteBitmap(ctx,
positionalDeleteBitmap, posSource.cursor()))
}
if hasDV {
@@ -2526,10 +2770,10 @@ func (as *arrowScan)
recordBatchesFromTasksAndDeletes(ctx context.Context, tasks
}
filePath :=
task.Value.File.FilePath()
- var positionalDeletes
positionDeletes
+ var positionalDeleteBitmap
*dv.RoaringPositionBitmap
if positionDeleteLoader != nil {
var err error
- positionalDeletes, err
= positionDeleteLoader.load(scanCtx, task.Value)
+ positionalDeleteBitmap,
err = positionDeleteLoader.load(scanCtx, task.Index, filePath)
if err != nil {
sink.fail(task,
err)
cancel(err)
@@ -2554,7 +2798,7 @@ func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx
context.Context, tasks
}
if err :=
as.recordsFromTask(scanCtx, task, sink,
- positionalDeletes,
+ positionalDeleteBitmap,
dvBitmap,
eqDeleteSets,
invariants); err != nil
{
diff --git a/table/arrow_scanner_lazy_delete_bench_test.go
b/table/arrow_scanner_lazy_delete_bench_test.go
index b25243a87..0ad127398 100644
--- a/table/arrow_scanner_lazy_delete_bench_test.go
+++ b/table/arrow_scanner_lazy_delete_bench_test.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table/dv"
)
const (
@@ -146,14 +147,227 @@ func BenchmarkLazyPositionDeleteLoading(b *testing.B) {
b.ResetTimer()
for b.Loop() {
loader := newLazyPositionDeleteLoader(fixture.fs,
fixture.tasks)
- deletes, err := loader.load(b.Context(),
fixture.tasks[0])
+ bitmap, err := loader.load(b.Context(), 0,
fixture.tasks[0].File.FilePath())
if err != nil {
b.Fatal(err)
}
- if len(deletes) != 1 {
- b.Fatalf("expected one delete chunk for the
first task, got %d", len(deletes))
+ if bitmap == nil || bitmap.Cardinality() != 1 {
+ b.Fatalf("expected one indexed delete for the
first task")
}
loader.release()
}
})
}
+
+var positionDeleteSplitReuseBenchmarkSink int64
+
+func BenchmarkPositionDeleteBitmapNoDeletes(b *testing.B) {
+ for _, taskCount := range []int{1, 10_000} {
+ b.Run(fmt.Sprintf("tasks=%d", taskCount), func(b *testing.B) {
+ tasks := make([]FileScanTask, taskCount)
+ dataFile := &mockDataFile{path:
"mem://benchmark/data/no-deletes.parquet"}
+ for i := range tasks {
+ tasks[i].File = dataFile
+ }
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ loader := newLazyPositionDeleteLoader(nil,
tasks)
+ if loader != nil {
+ b.Fatal("expected no position delete
loader")
+ }
+ loader.release()
+ }
+ })
+ }
+}
+
+// BenchmarkPositionDeleteBitmapUnion isolates merging already decoded delete
+// files. Separate high-bit buckets make repeated copies of the growing union
+// visible as the number of applicable delete files increases.
+func BenchmarkPositionDeleteBitmapUnion(b *testing.B) {
+ const (
+ dataPath = "mem://benchmark/data/union.parquet"
+ positionsPerFile = 16_384
+ )
+
+ for _, fileCount := range []int{1, 4, 16} {
+ b.Run(fmt.Sprintf("files=%d", fileCount), func(b *testing.B) {
+ loader := &lazyPositionDeleteLoader{
+ files: make(map[string]*lazyPositionDeleteFile,
fileCount),
+ }
+ paths := make([]string, fileCount)
+ for i := range paths {
+ paths[i] = fmt.Sprintf("delete-%d.parquet", i)
+ bitmap := dv.NewRoaringPositionBitmap()
+ for position := range positionsPerFile {
+ bitmap.Set(uint64(i)<<32 |
uint64(position*37))
+ }
+ cached := &lazyPositionDeleteFile{
+ bitmaps:
map[string]*dv.RoaringPositionBitmap{dataPath: bitmap},
+ }
+ cached.once.Do(func() {})
+ loader.files[paths[i]] = cached
+ }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ bitmap, err := loader.buildIndex(b.Context(),
dataPath, paths)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if bitmap.Cardinality() !=
int64(fileCount*positionsPerFile) {
+ b.Fatal("merged bitmap has unexpected
cardinality")
+ }
+ positionDeleteSplitReuseBenchmarkSink =
bitmap.Cardinality()
+ }
+ })
+ }
+}
+
+// BenchmarkPositionDeleteBitmapCachedLoad measures the per-split lookup after
+// the delete files and merged bitmap are already cached. Each operation visits
+// 16 splits of the same data file without any Parquet decoding or row
filtering.
+func BenchmarkPositionDeleteBitmapCachedLoad(b *testing.B) {
+ const (
+ dataPath =
"mem://benchmark/data/cached-position-delete-index.parquet"
+ splits = 16
+ )
+
+ for _, fileCount := range []int{1, 4, 16} {
+ b.Run(fmt.Sprintf("files=%d", fileCount), func(b *testing.B) {
+ deleteFiles := make([]iceberg.DataFile, fileCount)
+ for i := range deleteFiles {
+ deleteFiles[i] = &mockDataFile{
+ path:
fmt.Sprintf("mem://benchmark/deletes/cached-index-%02d.parquet", i),
+ contentType:
iceberg.EntryContentPosDeletes,
+ }
+ }
+ tasks := make([]FileScanTask, splits)
+ for i := range tasks {
+ tasks[i] = FileScanTask{
+ File: &mockDataFile{path: dataPath},
DeleteFiles: deleteFiles,
+ Start: int64(i * 128), Length: 128,
+ }
+ }
+ loader := newLazyPositionDeleteLoader(nil, tasks)
+ defer loader.release()
+ for i, deleteFile := range deleteFiles {
+ cached := loader.files[deleteFile.FilePath()]
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.Set(uint64(i))
+ cached.bitmaps =
map[string]*dv.RoaringPositionBitmap{dataPath: bitmap}
+ cached.once.Do(func() {})
+ }
+ if _, err := loader.load(b.Context(), 0, dataPath); err
!= nil {
+ b.Fatal(err)
+ }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ for taskIndex := range tasks {
+ bitmap, err := loader.load(b.Context(),
taskIndex, dataPath)
+ if err != nil {
+ b.Fatal(err)
+ }
+ positionDeleteSplitReuseBenchmarkSink =
bitmap.Cardinality()
+ }
+ }
+ })
+ }
+}
+
+func BenchmarkPositionDeleteBitmapReuseAcrossSplits(b *testing.B) {
+ const deleteRows = 100_000
+
+ deletePath := "mem://benchmark/deletes/split-reuse.parquet"
+ dataPath := "mem://benchmark/data/split-reuse.parquet"
+ fs := iceio.NewMemFS()
+
+ var content strings.Builder
+ content.Grow(deleteRows * 72)
+ content.WriteByte('[')
+ for pos := range deleteRows {
+ if pos > 0 {
+ content.WriteByte(',')
+ }
+ fmt.Fprintf(&content, `{"file_path":"%s","pos":%d}`, dataPath,
pos)
+ }
+ content.WriteByte(']')
+ benchmarkWritePosDeleteParquet(b, fs, deletePath, content.String())
+
+ deleteBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ deletePath, iceberg.ParquetFile, nil, nil, nil, deleteRows, 128)
+ if err != nil {
+ b.Fatal(err)
+ }
+ deleteFile := deleteBuilder.Build()
+
+ dataBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+ dataPath, iceberg.ParquetFile, nil, nil, nil, deleteRows*2,
16*128)
+ if err != nil {
+ b.Fatal(err)
+ }
+ dataFile := dataBuilder.Build()
+ targets := map[string]struct{}{dataPath: {}}
+
+ for _, splitCount := range []int{1, 4, 16} {
+ b.Run(fmt.Sprintf("splits=%d", splitCount), func(b *testing.B) {
+ tasks := make([]FileScanTask, splitCount)
+ for i := range tasks {
+ tasks[i] = FileScanTask{
+ File: dataFile,
+ DeleteFiles:
[]iceberg.DataFile{deleteFile},
+ Start: int64(i * 128),
+ Length: 128,
+ }
+ }
+
+ b.Run("task_local_hash_sets", func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ deletes, err :=
readDeletesForPaths(b.Context(), fs, deleteFile, targets)
+ if err != nil {
+ b.Fatal(err)
+ }
+ positions :=
positionDeletes{deletes[dataPath]}
+ for range splitCount {
+ set, err :=
collectPosDeletePositions(positions)
+ if err != nil {
+
releasePosDeletes(deletes)
+ b.Fatal(err)
+ }
+
positionDeleteSplitReuseBenchmarkSink = int64(len(set))
+ }
+ releasePosDeletes(deletes)
+ }
+ })
+
+ b.Run("shared_roaring_bitmap", func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ loader :=
newLazyPositionDeleteLoader(fs, tasks)
+ for taskIndex := range tasks {
+ bitmap, err :=
loader.load(b.Context(), taskIndex, dataPath)
+ if err != nil {
+ loader.release()
+ b.Fatal(err)
+ }
+ if bitmap == nil {
+ loader.release()
+ b.Fatal("expected
position delete bitmap")
+ }
+
positionDeleteSplitReuseBenchmarkSink = bitmap.Cardinality()
+ }
+ loader.release()
+ }
+ })
+ })
+ }
+}
diff --git a/table/arrow_scanner_lazy_delete_regression_test.go
b/table/arrow_scanner_lazy_delete_regression_test.go
index a1ba540e3..87ff58848 100644
--- a/table/arrow_scanner_lazy_delete_regression_test.go
+++ b/table/arrow_scanner_lazy_delete_regression_test.go
@@ -35,6 +35,7 @@ import (
"github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table/dv"
"github.com/apache/iceberg-go/table/internal"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -141,27 +142,358 @@ func
TestLazyPositionDeleteLoaderDefersReadsAndSharesResults(t *testing.T) {
assert.Zero(t, memFS.opens.Load(), "constructing the scan loader must
not open delete files")
- gotA, err := loader.load(ctx, tasks[0])
+ gotA, err := loader.load(ctx, 0, dataPathA)
require.NoError(t, err)
- require.Len(t, gotA, 1, "duplicate delete references must be read once
per task")
- assert.Equal(t, []int64{1}, int64Values(gotA[0]))
+ require.NotNil(t, gotA)
+ assert.Equal(t, int64(1), gotA.Cardinality(),
+ "duplicate delete references must be indexed once per task")
+ assert.True(t, gotA.Contains(1))
assert.Equal(t, int64(1), memFS.opens.Load())
cached := loader.files[deletePath]
require.NotNil(t, cached)
assert.Equal(t, map[string]struct{}{dataPathA: {}, dataPathB: {}},
cached.targets)
- assert.NotContains(t, cached.deletes, unrelatedPath,
+ assert.NotContains(t, cached.bitmaps, unrelatedPath,
"lazy delete reads must filter paths that are not referenced by
scan tasks")
- gotB, err := loader.load(ctx, tasks[1])
+ gotB, err := loader.load(ctx, 1, dataPathB)
require.NoError(t, err)
- require.Len(t, gotB, 1)
- assert.Equal(t, []int64{3}, int64Values(gotB[0]))
+ require.NotNil(t, gotB)
+ assert.Equal(t, int64(1), gotB.Cardinality())
+ assert.True(t, gotB.Contains(3))
assert.Equal(t, int64(1), memFS.opens.Load(), "shared delete files must
use one read")
loader.release()
}
+func TestLazyPositionDeleteLoaderSkipsPlansWithoutPositionDeletes(t
*testing.T) {
+ for _, tc := range []struct {
+ name string
+ equalityDeleteFiles []iceberg.DataFile
+ }{
+ {name: "no deletes"},
+ {name: "equality deletes only", equalityDeleteFiles:
[]iceberg.DataFile{&mockDataFile{
+ path: "mem://bucket/deletes/equality.parquet",
contentType: iceberg.EntryContentEqDeletes,
+ }}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ tasks := make([]FileScanTask, 10_000)
+ dataFile := newLazyDataFile(t,
"mem://bucket/data/a.parquet")
+ for i := range tasks {
+ tasks[i] = FileScanTask{File: dataFile,
EqualityDeleteFiles: tc.equalityDeleteFiles}
+ }
+ loader := newLazyPositionDeleteLoader(nil, tasks)
+ require.Nil(t, loader, "a scan without position deletes
must not allocate a task-sized index")
+ loader.release()
+ })
+ }
+}
+
+func TestLazyPositionDeleteLoaderKeepsOriginalTaskOrdinals(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ const (
+ dataPathA = "mem://bucket/data/a.parquet"
+ dataPathB = "mem://bucket/data/b.parquet"
+ deletePath = "mem://bucket/deletes/task-ordinals.parquet"
+ )
+ fs := &countingOpenMemFS{MemFS: iceio.NewMemFS()}
+ writePosDeleteParquetToMemFS(t, fs.MemFS, deletePath, `[
+ {"file_path":"`+dataPathA+`","pos":1},
+ {"file_path":"`+dataPathB+`","pos":2}
+ ]`)
+ deleteFile := newPosDeleteFile(t, deletePath, 2, 128)
+ tasks := []FileScanTask{
+ {File: newLazyDataFile(t, dataPathA)},
+ {File: newLazyDataFile(t, dataPathB), DeleteFiles:
[]iceberg.DataFile{deleteFile}},
+ {File: newLazyDataFile(t, dataPathA), DeleteFiles:
[]iceberg.DataFile{deleteFile}},
+ {File: newLazyDataFile(t, dataPathB), DeleteFiles:
[]iceberg.DataFile{deleteFile}},
+ }
+ loader := newLazyPositionDeleteLoader(fs, tasks)
+ defer loader.release()
+ for _, taskIndex := range []int{3, 0, 2, 1} {
+ bitmap, err := loader.load(ctx, taskIndex,
tasks[taskIndex].File.FilePath())
+ require.NoError(t, err)
+ if taskIndex == 0 {
+ assert.Nil(t, bitmap, "tasks without deletes must keep
their place in the index slice")
+
+ continue
+ }
+ require.NotNil(t, bitmap)
+ assert.Equal(t, int64(1), bitmap.Cardinality())
+ if taskIndex == 2 {
+ assert.True(t, bitmap.Contains(1))
+ } else {
+ assert.True(t, bitmap.Contains(2))
+ }
+ }
+ assert.Same(t, loader.indexes[1].index, loader.indexes[3].index, "split
tasks must share the same cached index")
+ assert.Equal(t, int64(1), fs.opens.Load())
+ for _, taskIndex := range []int{-1, len(tasks)} {
+ bitmap, err := loader.load(ctx, taskIndex, dataPathA)
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ assert.Nil(t, bitmap)
+ }
+ for _, taskIndex := range []int{0, 2} {
+ bitmap, err := loader.load(ctx, taskIndex, dataPathB)
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument,
+ "a reordered task must fail even when its original
ordinal had no deletes")
+ assert.Nil(t, bitmap)
+ assert.ErrorContains(t, err, dataPathA)
+ assert.ErrorContains(t, err, dataPathB)
+ }
+}
+
+func TestLazyPositionDeleteLoaderSkipsDVSupersededFiles(t *testing.T) {
+ const (
+ dataPath = "mem://bucket/data/dv.parquet"
+ deletePath = "mem://bucket/deletes/superseded.parquet"
+ )
+ fs := &countingOpenMemFS{MemFS: iceio.NewMemFS()}
+ writePosDeleteParquetToMemFS(t, fs.MemFS, deletePath,
+ `[{"file_path":"`+dataPath+`","pos":-1}]`)
+ task := FileScanTask{
+ File: newLazyDataFile(t, dataPath),
+ DeleteFiles: []iceberg.DataFile{newPosDeleteFile(t,
deletePath, 1, 128)},
+ DeletionVectorFiles: []iceberg.DataFile{&mockDataFile{path:
"mem://bucket/deletes/dv.puffin"}},
+ }
+ loader := newLazyPositionDeleteLoader(fs, []FileScanTask{task})
+ defer loader.release()
+ bitmap, err := loader.load(t.Context(), 0, dataPath)
+ require.NoError(t, err)
+ assert.Nil(t, bitmap)
+ assert.Zero(t, fs.opens.Load(), "superseded position deletes must not
be opened or validated")
+
+ // Prove that the fixture really fails validation without DV precedence.
+ task.DeletionVectorFiles = nil
+ activeLoader := newLazyPositionDeleteLoader(fs, []FileScanTask{task})
+ defer activeLoader.release()
+ _, err = activeLoader.load(t.Context(), 0, dataPath)
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ assert.Equal(t, int64(1), fs.opens.Load())
+}
+
+func TestLazyPositionDeleteLoaderConcurrentReadsWhileBuildingUnions(t
*testing.T) {
+ const (
+ dataPath = "mem://bucket/data/shared-bitmap.parquet"
+ positions = 8_192
+ readers = 4
+ )
+ deleteFiles := []iceberg.DataFile{
+ newPosDeleteFile(t, "mem://bucket/deletes/first.parquet",
positions, 128),
+ newPosDeleteFile(t, "mem://bucket/deletes/second.parquet",
positions, 128),
+ }
+ tasks := []FileScanTask{
+ {File: newLazyDataFile(t, dataPath), DeleteFiles:
deleteFiles[:1]},
+ {File: newLazyDataFile(t, dataPath), DeleteFiles:
deleteFiles[1:]},
+ {File: newLazyDataFile(t, dataPath), DeleteFiles: deleteFiles},
+ }
+ loader := newLazyPositionDeleteLoader(nil, tasks)
+ defer loader.release()
+ paths := []string{deleteFiles[0].FilePath(), deleteFiles[1].FilePath()}
+ for i, path := range paths {
+ bitmap := dv.NewRoaringPositionBitmap()
+ for position := range positions {
+ bitmap.Set(uint64(position*2 + i))
+ }
+ cached := loader.files[path]
+ cached.bitmaps = map[string]*dv.RoaringPositionBitmap{dataPath:
bitmap}
+ cached.once.Do(func() {})
+ }
+ first, err := loader.load(t.Context(), 0, dataPath)
+ require.NoError(t, err)
+ second, err := loader.load(t.Context(), 1, dataPath)
+ require.NoError(t, err)
+ require.Same(t, loader.files[paths[0]].bitmaps[dataPath], first)
+ require.Same(t, loader.files[paths[1]].bitmaps[dataPath], second)
+
+ start := make(chan struct{})
+ var wg sync.WaitGroup
+ wg.Add(readers + 1)
+ for range readers {
+ go func() {
+ defer wg.Done()
+ <-start
+ for i := range positions * 16 {
+ position := uint64(i % positions * 2)
+ if !first.Contains(position) ||
first.Contains(position+1) ||
+ second.Contains(position) ||
!second.Contains(position+1) {
+ t.Error("building a union mutated a
shared input bitmap")
+
+ return
+ }
+ }
+ }()
+ }
+ go func() {
+ defer wg.Done()
+ <-start
+ for range 32 {
+ merged, err := loader.buildIndex(t.Context(), dataPath,
paths)
+ if !assert.NoError(t, err) {
+ return
+ }
+ assert.Equal(t, int64(positions*2),
merged.Cardinality())
+ }
+ }()
+ close(start)
+ wg.Wait()
+ assert.Equal(t, int64(positions), first.Cardinality())
+ assert.Equal(t, int64(positions), second.Cardinality())
+}
+
+func TestLazyPositionDeleteLoaderReusesMergedBitmapAcrossSplitTasks(t
*testing.T) {
+ mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+
+ const (
+ deletePathA = "mem://bucket/deletes/split-a.parquet"
+ deletePathB = "mem://bucket/deletes/split-b.parquet"
+ deletePathC = "mem://bucket/deletes/split-c.parquet"
+ dataPath = "mem://bucket/data/split.parquet"
+ )
+ fs := &countingOpenMemFS{MemFS: iceio.NewMemFS()}
+ writePosDeleteParquetToMemFS(t, fs.MemFS, deletePathA,
+ `[{"file_path":"`+dataPath+`","pos":1}]`)
+ writePosDeleteParquetToMemFS(t, fs.MemFS, deletePathB,
+ `[{"file_path":"`+dataPath+`","pos":3}]`)
+ writePosDeleteParquetToMemFS(t, fs.MemFS, deletePathC,
+ `[{"file_path":"`+dataPath+`","pos":5}]`)
+
+ deleteA := newPosDeleteFile(t, deletePathA, 1, 128)
+ deleteB := newPosDeleteFile(t, deletePathB, 1, 128)
+ deleteC := newPosDeleteFile(t, deletePathC, 1, 128)
+ dataFile := newLazyDataFile(t, dataPath)
+ tasks := []FileScanTask{
+ {File: dataFile, DeleteFiles: []iceberg.DataFile{deleteA,
deleteB, deleteC}, Start: 0, Length: 64},
+ {File: dataFile, DeleteFiles: []iceberg.DataFile{deleteC,
deleteB, deleteA}, Start: 64, Length: 64},
+ {File: dataFile, DeleteFiles: []iceberg.DataFile{deleteA}},
+ {File: dataFile, DeleteFiles: []iceberg.DataFile{deleteB,
deleteC}},
+ }
+ loader := newLazyPositionDeleteLoader(fs, tasks)
+
+ first, err := loader.load(ctx, 0, dataPath)
+ require.NoError(t, err)
+ require.NotNil(t, first)
+ second, err := loader.load(ctx, 1, dataPath)
+ require.NoError(t, err)
+ require.NotNil(t, second)
+
+ assert.Same(t, first, second, "split tasks must reuse the merged
position index")
+ assert.Equal(t, int64(3), first.Cardinality())
+ assert.True(t, first.Contains(1))
+ assert.True(t, first.Contains(3))
+ assert.True(t, first.Contains(5))
+
+ // Merging three or more files must not change cached source bitmaps
used
+ // by another task with a smaller set of applicable delete files.
+ single, err := loader.load(ctx, 2, dataPath)
+ require.NoError(t, err)
+ require.NotNil(t, single)
+ assert.Equal(t, int64(1), single.Cardinality())
+ assert.True(t, single.Contains(1))
+ subset, err := loader.load(ctx, 3, dataPath)
+ require.NoError(t, err)
+ require.NotNil(t, subset)
+ assert.Equal(t, int64(2), subset.Cardinality())
+ assert.True(t, subset.Contains(3))
+ assert.True(t, subset.Contains(5))
+ assert.Equal(t, int64(3), fs.opens.Load(), "each delete file must be
read once")
+
+ loader.release()
+}
+
+func TestLazyPositionDeleteLoaderUnionSkipsEmptyTargetsWithoutMutatingInputs(t
*testing.T) {
+ for _, emptyIndex := range []int{0, 1} {
+ t.Run(fmt.Sprintf("empty file %d", emptyIndex), func(t
*testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.NewGoAllocator())
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ const dataPath = "mem://bucket/data/union.parquet"
+ fs := &countingOpenMemFS{MemFS: iceio.NewMemFS()}
+ deleteFiles := make([]iceberg.DataFile, 3)
+ var want []uint64
+ for i := range deleteFiles {
+ path := dataPath
+ if i == emptyIndex {
+ path =
"mem://bucket/data/unrelated.parquet"
+ } else {
+ want = append(want, uint64(i))
+ }
+ deletePath :=
fmt.Sprintf("mem://bucket/deletes/union-%d.parquet", i)
+ writePosDeleteParquetToMemFS(t, fs.MemFS,
deletePath,
+
fmt.Sprintf(`[{"file_path":%q,"pos":%d}]`, path, i))
+ deleteFiles[i] = newPosDeleteFile(t,
deletePath, 1, 128)
+ }
+ firstPresent := int(want[0])
+ tasks := []FileScanTask{
+ {File: newLazyDataFile(t, dataPath),
DeleteFiles: deleteFiles},
+ {File: newLazyDataFile(t, dataPath),
DeleteFiles: []iceberg.DataFile{deleteFiles[firstPresent]}},
+ }
+ loader := newLazyPositionDeleteLoader(fs, tasks)
+ defer loader.release()
+
+ single, err := loader.load(ctx, 1, dataPath)
+ require.NoError(t, err)
+ merged, err := loader.load(ctx, 0, dataPath)
+ require.NoError(t, err)
+ require.NotNil(t, merged)
+ assert.Equal(t, int64(len(want)), merged.Cardinality())
+ for _, position := range want {
+ assert.True(t, merged.Contains(position))
+ }
+ cached :=
loader.files[deleteFiles[firstPresent].FilePath()].bitmaps[dataPath]
+ assert.Same(t, single, cached)
+ assert.Equal(t, int64(1), cached.Cardinality(), "the
first non-empty input must stay immutable")
+ assert.True(t, cached.Contains(uint64(firstPresent)))
+ assert.Equal(t, int64(3), fs.opens.Load())
+ })
+ }
+}
+
+func TestLazyPositionDeleteLoaderPreservesOtherTargetValidation(t *testing.T) {
+ for _, invalidPos := range []string{"-1", "null"} {
+ t.Run(invalidPos, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.NewGoAllocator())
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ const (
+ dataPathA = "mem://bucket/data/a.parquet"
+ dataPathB = "mem://bucket/data/b.parquet"
+ deletePath =
"mem://bucket/deletes/invalid-other-target.parquet"
+ )
+ fs := &countingOpenMemFS{MemFS: iceio.NewMemFS()}
+ fields := PositionalDeleteArrowSchema.Fields()
+ fields[1].Nullable = true
+ writePosDeleteParquetToMemFSWithSchema(t, fs.MemFS,
deletePath, arrow.NewSchema(fields, nil), `[
+ {"file_path":"`+dataPathA+`","pos":1},
+
{"file_path":"`+dataPathB+`","pos":`+invalidPos+`}
+ ]`)
+ deleteFile := newPosDeleteFile(t, deletePath, 2, 128)
+ tasks := []FileScanTask{{
+ File: newLazyDataFile(t, dataPathA),
DeleteFiles: []iceberg.DataFile{deleteFile},
+ }}
+
+ // The existing Arrow reader validates every decoded
row before target
+ // filtering. Bitmap conversion must preserve that
error behavior.
+ deletes, err := readDeletesForPaths(ctx, fs,
deleteFile, map[string]struct{}{dataPathA: {}})
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ assert.Nil(t, deletes)
+
+ loader := newLazyPositionDeleteLoader(fs, tasks)
+ defer loader.release()
+ bitmap, err := loader.load(ctx, 0, dataPathA)
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ assert.Nil(t, bitmap)
+ _, cachedErr := loader.load(ctx, 0, dataPathA)
+ assert.ErrorIs(t, cachedErr, err)
+ assert.Equal(t, int64(2), fs.opens.Load(), "one read
per loader, including the legacy reader")
+ })
+ }
+}
+
func TestLazyPositionDeleteLoaderSingleflightsConcurrentLoads(t *testing.T) {
mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
defer mem.AssertSize(t, 0)
@@ -182,14 +514,14 @@ func
TestLazyPositionDeleteLoaderSingleflightsConcurrentLoads(t *testing.T) {
loader := newLazyPositionDeleteLoader(fs, []FileScanTask{task})
const callers = 8
- results := make(chan positionDeletes, callers)
+ results := make(chan *dv.RoaringPositionBitmap, callers)
errs := make(chan error, callers)
var wg sync.WaitGroup
wg.Add(callers)
for range callers {
go func() {
defer wg.Done()
- deletes, err := loader.load(ctx, task)
+ deletes, err := loader.load(ctx, 0, dataPath)
results <- deletes
errs <- err
}()
@@ -201,9 +533,16 @@ func
TestLazyPositionDeleteLoaderSingleflightsConcurrentLoads(t *testing.T) {
for err := range errs {
require.NoError(t, err)
}
- for deletes := range results {
- require.Len(t, deletes, 1)
- assert.Equal(t, []int64{7}, int64Values(deletes[0]))
+ var first *dv.RoaringPositionBitmap
+ for bitmap := range results {
+ require.NotNil(t, bitmap)
+ assert.Equal(t, int64(1), bitmap.Cardinality())
+ assert.True(t, bitmap.Contains(7))
+ if first == nil {
+ first = bitmap
+ } else {
+ assert.Same(t, first, bitmap, "concurrent loads must
share one position index")
+ }
}
assert.Equal(t, int64(1), fs.opens.Load(), "concurrent users must share
one delete-file read")
@@ -219,13 +558,13 @@ func TestLazyPositionDeleteLoaderCachesErrors(t
*testing.T) {
}
loader := newLazyPositionDeleteLoader(fs, []FileScanTask{task})
- first, err := loader.load(context.Background(), task)
+ first, err := loader.load(context.Background(), 0, task.File.FilePath())
require.Error(t, err)
assert.Nil(t, first)
assert.ErrorContains(t, err, "read position deletes from
"+deleteFile.FilePath())
assert.Equal(t, int64(1), fs.opens.Load())
- second, secondErr := loader.load(context.Background(), task)
+ second, secondErr := loader.load(context.Background(), 0,
task.File.FilePath())
require.Error(t, secondErr)
assert.Nil(t, second)
assert.ErrorIs(t, secondErr, err)
@@ -244,7 +583,7 @@ func
TestLazyPositionDeleteLoaderWrapsErrorsWithoutFilePath(t *testing.T) {
}
loader := newLazyPositionDeleteLoader(fs, []FileScanTask{task})
- _, err := loader.load(context.Background(), task)
+ _, err := loader.load(context.Background(), 0, task.File.FilePath())
require.Error(t, err)
assert.ErrorContains(t, err, "read position deletes from "+deletePath)
loader.release()
@@ -266,9 +605,9 @@ func TestLazyPositionDeleteLoaderCachesCancellation(t
*testing.T) {
}
loader := newLazyPositionDeleteLoader(fs, []FileScanTask{task})
- _, firstErr := loader.load(ctx, task)
+ _, firstErr := loader.load(ctx, 0, task.File.FilePath())
require.ErrorIs(t, firstErr, context.Canceled)
- _, secondErr := loader.load(context.Background(), task)
+ _, secondErr := loader.load(context.Background(), 0,
task.File.FilePath())
require.ErrorIs(t, secondErr, context.Canceled)
assert.Equal(t, int64(1), fs.opens.Load(), "cancellation must not cause
a second read")
@@ -448,17 +787,22 @@ func
TestArrowScanRowLimitStopsBeforeLazyPositionDeleteError(t *testing.T) {
"a row limit may finish before a later task's positional-delete
error")
}
-func TestLazyPositionDeleteLoaderReleasesChunksWhenIteratorStops(t *testing.T)
{
+func TestLazyPositionDeleteLoaderClearsBitmapsWhenIteratorStops(t *testing.T) {
mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
defer mem.AssertSize(t, 0)
- loader := &lazyPositionDeleteLoader{files:
map[string]*lazyPositionDeleteFile{
- "mem://bucket/deletes/one.parquet": {
- deletes: map[string]*arrow.Chunked{
- "mem://bucket/data/a.parquet":
chunkedPosDelete(t, mem, []int64{1}),
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.Set(1)
+ loader := &lazyPositionDeleteLoader{
+ files: map[string]*lazyPositionDeleteFile{
+ "mem://bucket/deletes/one.parquet": {
+ bitmaps: map[string]*dv.RoaringPositionBitmap{
+ "mem://bucket/data/a.parquet": bitmap,
+ },
},
},
- }}
+ indexes: []positionDeleteTaskIndex{{index:
&lazyPositionDeleteIndex{bitmap: bitmap}}},
+ }
batch := checkedInt64RecordBatch(mem, 1)
records := make(chan enumeratedRecord, 1)
@@ -482,7 +826,10 @@ func
TestLazyPositionDeleteLoaderReleasesChunksWhenIteratorStops(t *testing.T) {
}
for _, cached := range loader.files {
- assert.Nil(t, cached.deletes, "release must clear cached delete
chunks")
+ assert.Nil(t, cached.bitmaps, "release must clear cached delete
bitmaps")
+ }
+ for _, index := range loader.indexes {
+ assert.Nil(t, index.index.bitmap, "release must clear merged
position indexes")
}
}
@@ -490,7 +837,7 @@ func TestLazyPositionDeleteLoaderRejectsLoadAfterRelease(t
*testing.T) {
loader := &lazyPositionDeleteLoader{}
loader.release()
- _, err := loader.load(context.Background(), FileScanTask{})
+ _, err := loader.load(context.Background(), 0, "")
require.ErrorIs(t, err, errPositionDeleteLoaderReleased)
}
diff --git a/table/arrow_scanner_posdelete_bitmap_test.go
b/table/arrow_scanner_posdelete_bitmap_test.go
new file mode 100644
index 000000000..0db2b23f2
--- /dev/null
+++ b/table/arrow_scanner_posdelete_bitmap_test.go
@@ -0,0 +1,282 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "math"
+ "slices"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/compute"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table/dv"
+ "github.com/apache/iceberg-go/table/internal"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestCombinePositionalDeleteBitmap(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ deletes []uint64
+ spans []internal.RowGroupSpan
+ batches []int64
+ want [][]int64
+ wantNext int64
+ }{
+ {name: "empty batch", batches: []int64{0}, want:
[][]int64{nil}},
+ {name: "no deletes", batches: []int64{4}, want: [][]int64{nil},
wantNext: 4},
+ {name: "deletes outside batch", deletes: []uint64{10}, batches:
[]int64{4}, want: [][]int64{nil}, wantNext: 4},
+ {name: "first row", deletes: []uint64{0}, batches: []int64{4},
want: [][]int64{{1, 2, 3}}, wantNext: 4},
+ {name: "backfill", deletes: []uint64{2}, batches: []int64{4},
want: [][]int64{{0, 1, 3}}, wantNext: 4},
+ {name: "last row", deletes: []uint64{3}, batches: []int64{4},
want: [][]int64{{0, 1, 2}}, wantNext: 4},
+ {name: "all rows", deletes: []uint64{0, 1, 2}, batches:
[]int64{3}, want: [][]int64{{}}, wantNext: 3},
+ {name: "across batches", deletes: []uint64{3}, batches:
[]int64{3, 2}, want: [][]int64{nil, {1}}, wantNext: 5},
+ {
+ name: "pruned row groups", deletes: []uint64{4},
+ spans: []internal.RowGroupSpan{{FirstRowPos: 0,
NumRows: 2}, {FirstRowPos: 4, NumRows: 2}},
+ batches: []int64{2, 2}, want: [][]int64{nil, {1}},
wantNext: 6,
+ },
+ {
+ name: "64-bit positions", deletes: []uint64{1 << 32},
+ spans: []internal.RowGroupSpan{{FirstRowPos: 1<<32 -
1, NumRows: 3}},
+ batches: []int64{3}, want: [][]int64{{0, 2}}, wantNext:
1<<32 + 2,
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ bitmap := dv.NewRoaringPositionBitmap()
+ for _, position := range tc.deletes {
+ bitmap.Set(position)
+ }
+ cursor := (&rowPositionSource{spans: tc.spans}).cursor()
+ for i, nrows := range tc.batches {
+ indices := combinePositionalDeleteBitmap(mem,
bitmap, cursor, nrows)
+ if tc.want[i] == nil {
+ require.Nil(t, indices, "a clean batch
must use the passthrough path")
+
+ continue
+ }
+ require.NotNil(t, indices, "an empty array must
still remove an entirely deleted batch")
+ if len(tc.want[i]) == 0 {
+ assert.Zero(t, indices.Len())
+ } else {
+ assert.Equal(t, tc.want[i],
indices.(*array.Int64).Int64Values())
+ }
+ indices.Release()
+ }
+ assert.Equal(t, tc.wantNext, cursor.next(), "every
input row must advance the shared position source")
+ })
+ }
+}
+
+func TestProcessPositionalDeleteBitmapAcrossBatches(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.Set(3)
+ process := processPositionalDeleteBitmap(ctx, bitmap,
(&rowPositionSource{}).cursor())
+
+ for i, values := range [][]int64{{10, 11, 12}, {13, 14}} {
+ batch := checkedInt64RecordBatch(mem, values...)
+ out, err := process(batch)
+ require.NoError(t, err)
+ if i == 0 {
+ assert.Same(t, batch, out, "the clean first batch must
retain its input")
+ assert.Equal(t, values,
out.Column(0).(*array.Int64).Int64Values())
+ } else {
+ assert.Equal(t, []int64{14},
out.Column(0).(*array.Int64).Int64Values())
+ }
+ out.Release()
+ }
+}
+
+func TestCombinePositionalDeleteBitmapLargeBatches(t *testing.T) {
+ const nrows = 1_024
+ for _, tc := range []struct {
+ name string
+ start, end uint64
+ }{
+ {name: "first row", start: 0, end: 1},
+ {name: "leading run", start: 0, end: 512},
+ {name: "middle row", start: 512, end: 513},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.SetRange(tc.start, tc.end)
+ var want []int64
+ for i := range uint64(nrows) {
+ if i < tc.start || i >= tc.end {
+ want = append(want, int64(i))
+ }
+ }
+
+ indices := combinePositionalDeleteBitmap(mem, bitmap,
(&rowPositionSource{}).cursor(), nrows)
+ require.NotNil(t, indices)
+ defer indices.Release()
+ assert.Equal(t, want,
indices.(*array.Int64).Int64Values())
+ })
+ }
+}
+
+type positionDeleteAllocationTracker struct {
+ memory.Allocator
+ largest int
+}
+
+func (a *positionDeleteAllocationTracker) Allocate(size int) []byte {
+ a.largest = max(a.largest, size)
+
+ return a.Allocator.Allocate(size)
+}
+
+func (a *positionDeleteAllocationTracker) Reallocate(size int, b []byte)
[]byte {
+ a.largest = max(a.largest, size)
+
+ return a.Allocator.Reallocate(size, b)
+}
+
+func TestCombinePositionalDeleteBitmapAllocationBounds(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ nrows int64
+ start, end uint64
+ }{
+ {name: "clean", nrows: 1_024},
+ {name: "first row", nrows: 1_024, start: 0, end: 1},
+ {name: "middle row", nrows: 1_024, start: 512, end: 513},
+ {name: "first row non-power-of-two batch", nrows: 1_025, start:
0, end: 1},
+ {name: "middle row non-power-of-two batch", nrows: 1_025,
start: 512, end: 513},
+ {name: "leading run", nrows: 1_024, start: 0, end: 512},
+ {name: "all rows", nrows: 1_024, start: 0, end: 1_024},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ checked :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer checked.AssertSize(t, 0)
+ mem := &positionDeleteAllocationTracker{Allocator:
checked}
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.SetRange(tc.start, tc.end)
+ indices := combinePositionalDeleteBitmap(mem, bitmap,
(&rowPositionSource{}).cursor(), tc.nrows)
+ if tc.start == tc.end {
+ require.Nil(t, indices)
+ assert.Zero(t, mem.largest, "a clean batch must
not allocate Arrow buffers")
+
+ return
+ }
+ require.NotNil(t, indices)
+ defer indices.Release()
+ survivors := int(tc.nrows) - int(tc.end-tc.start)
+ assert.Equal(t, survivors, indices.Len())
+ if survivors == 0 {
+ assert.Zero(t, mem.largest, "a fully deleted
batch must not allocate Arrow buffers")
+ } else {
+ // Arrow aligns buffers to 64 bytes. Check the
largest request,
+ // before NewArray can shrink an oversized
values buffer.
+ assert.LessOrEqual(t, mem.largest,
arrow.Int64Traits.BytesRequired(survivors)+63)
+ }
+ })
+ }
+}
+
+func TestCollectPosDeleteBitmapRejectsInvalidPositions(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ column func(memory.Allocator) arrow.Array
+ want string
+ }{
+ {name: "nil chunk", want: "nil pos column chunk"},
+ {name: "wrong type", column: func(mem memory.Allocator)
arrow.Array {
+ return int32Array(mem, 1)
+ }, want: "unsupported pos column type"},
+ {name: "negative", column: func(mem memory.Allocator)
arrow.Array {
+ return int64Array(mem, -1)
+ }, want: "negative pos -1"},
+ {name: "null", column: func(mem memory.Allocator) arrow.Array {
+ builder := array.NewInt64Builder(mem)
+ defer builder.Release()
+ builder.AppendNull()
+
+ return builder.NewArray()
+ }, want: "null pos in position delete file"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ var chunk *arrow.Chunked
+ if tc.column != nil {
+ column := tc.column(mem)
+ defer column.Release()
+ chunk = arrow.NewChunked(column.DataType(),
[]arrow.Array{column})
+ defer chunk.Release()
+ }
+
+ bitmap, err :=
collectPosDeleteBitmap(positionDeletes{chunk})
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ assert.ErrorContains(t, err, tc.want)
+ assert.Nil(t, bitmap)
+ })
+ }
+}
+
+func TestCollectPosDeleteBitmapDeduplicates64BitPositions(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ first := int64Array(mem, 0, 2, 2, 1<<32)
+ defer first.Release()
+ second := int64Array(mem, 1<<32, math.MaxInt64)
+ defer second.Release()
+ column := arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{first, second})
+ defer column.Release()
+
+ bitmap, err := collectPosDeleteBitmap(positionDeletes{column, column})
+ require.NoError(t, err)
+ assert.Equal(t, []uint64{0, 2, 1 << 32, math.MaxInt64},
slices.Collect(bitmap.Positions()))
+}
+
+func BenchmarkCombinePositionalDeleteBitmap(b *testing.B) {
+ const nrows = 65_536
+ for _, name := range []string{"clean", "first", "middle", "all"} {
+ b.Run(name, func(b *testing.B) {
+ bitmap := dv.NewRoaringPositionBitmap()
+ switch name {
+ case "first":
+ bitmap.Set(0)
+ case "middle":
+ bitmap.Set(nrows / 2)
+ case "all":
+ bitmap.SetRange(0, nrows)
+ }
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ indices :=
combinePositionalDeleteBitmap(memory.DefaultAllocator, bitmap,
(&rowPositionSource{}).cursor(), nrows)
+ if indices != nil {
+ positionDeleteSplitReuseBenchmarkSink =
int64(indices.Len())
+ indices.Release()
+ }
+ }
+ })
+ }
+}
diff --git a/table/dv_scanner_end_to_end_test.go
b/table/dv_scanner_end_to_end_test.go
index 1506ca980..807d01bc7 100644
--- a/table/dv_scanner_end_to_end_test.go
+++ b/table/dv_scanner_end_to_end_test.go
@@ -19,6 +19,7 @@ package table
import (
"context"
+ "fmt"
"path/filepath"
"slices"
"strconv"
@@ -216,6 +217,37 @@ func TestDVScanEndToEnd(t *testing.T) {
"first and last positions must be removable")
})
+ for _, position := range []string{"0", "-1", "null"} {
+ t.Run("DV supersedes position delete "+position, func(t
*testing.T) {
+ fs, tmp, tbl := newDVScanTestEnv(t)
+ dataPath := filepath.Join(tmp,
"data-with-both-deletes.parquet")
+ df := writeIntParquetWithFieldID(t, fs, dataPath, 0, 5)
+ deletePath := filepath.Join(tmp,
"position-deletes.parquet")
+ fields := PositionalDeleteArrowSchema.Fields()
+ fields[1].Nullable = true
+ deleteSchema := arrow.NewSchema(fields, nil)
+ record := mustLoadRecordBatchFromJSON(deleteSchema,
+ fmt.Sprintf(`[{"file_path":%q,"pos":%s}]`,
dataPath, position))
+ defer record.Release()
+ deleteTable := array.NewTableFromRecords(deleteSchema,
[]arrow.RecordBatch{record})
+ defer deleteTable.Release()
+ deleteFile, err := fs.Create(deletePath)
+ require.NoError(t, err)
+ defer deleteFile.Close()
+ require.NoError(t, pqarrow.WriteTable(deleteTable,
deleteFile, 1, nil, pqarrow.DefaultWriterProps()))
+
+ // A DV takes precedence for hand-built tasks too.
Superseded
+ // position files must not affect rows or fail
validation.
+ puffinPath, offset, length, card :=
writeDVPuffinFixture(t, []uint64{3}, dataPath)
+ got := collectDVScanRows(t, tbl.Scan(), []FileScanTask{{
+ File: df,
+ DeleteFiles:
[]iceberg.DataFile{newPosDeleteFile(t, deletePath, 1, 128)},
+ DeletionVectorFiles:
[]iceberg.DataFile{newDVMockDataFile(puffinPath, dataPath, offset, length,
card)},
+ }})
+ assert.Equal(t, []int64{0, 1, 2, 4}, got)
+ })
+ }
+
t.Run("DV deletes all rows", func(t *testing.T) {
fs, tmp, tbl := newDVScanTestEnv(t)
