Skip to content

perf(table): load deletion-vector Puffin groups lazily #1968

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fallintoplace wants to merge 2 commits into
base: main
Choose a base branch
from
+554 −42
Open

perf(table): load deletion-vector Puffin groups lazily #1968

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
63cb840
perf(table): load deletion-vector Puffin groups lazily
fallintoplace Aug 29, 2026
9b03e3e
fix(table): reject empty deletion vector references
fallintoplace Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
196 changes: 154 additions & 42 deletions table/arrow_scanner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,31 +144,109 @@ func readAllDeleteFiles(ctx context.Context, fs iceio.IO, tasks []FileScanTask,
// directly, instead of materializing positions into a set[int64] + Take.
type perFileDVBitmaps = map[string]*dv.RoaringPositionBitmap

// readAllDeletionVectors reads every deletion-vector puffin blob referenced
// by the input tasks and returns a perFileDVBitmaps map keyed by the
// referenced data-file path.
//
// Dedup is by referenced-data-file path, not by puffin file path: a single
// puffin file can carry multiple DV blobs (one per data file). Keying by the
// puffin path would silently drop all but the first blob. This matches Java's
// DeleteFileIndex.findDV, which keys by data-file path. As a side-effect we
// can detect spec violations: two distinct DV blobs targeting the same data
// file is rejected (mirrors Java's "Can't index multiple DVs for %s"
// ValidationException — over-deletion risk if silently unioned).
//
// Validation happens up front, before any goroutines are launched, so the
// goroutine fan-out has no early-exit path. (An early return after g.Go
// dispatches but before g.Wait would close resultsChan while in-flight
// workers were still sending, panicking with "send on closed channel".)
func readAllDeletionVectors(ctx context.Context, fs iceio.IO, tasks []FileScanTask, concurrency int) (perFileDVBitmaps, error) {
out := make(perFileDVBitmaps)
// lazyDeletionVectorLoader indexes deletion-vector metadata for a scan, but
// waits to read a Puffin file until a task for one of its data files is
// processed. Each Puffin group is loaded once and all of its bitmaps are kept
// in the scan-scoped cache so tasks sharing a file do not repeat the read.
type lazyDeletionVectorLoader struct {
fs iceio.IO

groups map[string]*lazyDeletionVectorGroup
byDataFile map[string]*lazyDeletionVectorGroup
}

type lazyDeletionVectorGroup struct {
puffinPath string
referencedDataFiles []string
files []iceberg.DataFile

once sync.Once
bitmaps perFileDVBitmaps
err error
}

func newLazyDeletionVectorLoader(fs iceio.IO, tasks []FileScanTask) (*lazyDeletionVectorLoader, error) {
uniqueDVs, err := collectUniqueDeletionVectors(tasks)
if err != nil {
return nil, err
}

groups := groupDeletionVectors(uniqueDVs)
loader := &lazyDeletionVectorLoader{
fs: fs,
groups: groups,
byDataFile: make(map[string]*lazyDeletionVectorGroup, len(uniqueDVs)),
}
for _, group := range groups {
for _, ref := range group.referencedDataFiles {
loader.byDataFile[ref] = group
}
}

return loader, nil
}

func (l *lazyDeletionVectorLoader) load(ctx context.Context, dataFilePath string) (*dv.RoaringPositionBitmap, error) {
if l == nil {
return nil, nil
}

group := l.byDataFile[dataFilePath]
if group == nil {
return nil, nil
}
if err := ctx.Err(); err != nil {
return nil, err
}

group.once.Do(func() {
bitmaps, err := dv.ReadDVs(l.fs, group.files)
if err != nil {
group.err = fmt.Errorf("read deletion vectors from %s: %w", group.puffinPath, err)

return
}
if err := ctx.Err(); err != nil {

@laskoviymishka laskoviymishka Sep 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This one survived the rebase, and I'd still pull it out of the once.Do body. It can turn a successful read into a permanent failure for the whole group.

Once ReadDVs has returned cleanly the bitmaps are correct and safe to cache. But the goroutine that wins the Once runs on scanCtx, which any other worker can cancel. So if worker C hits an I/O error and calls cancel(err) just after worker A finished ReadDVs, A sees the cancelled ctx here, writes group.err = context.Canceled, and drops the bitmaps it just read. Worker B, waiting on the same Once, then unblocks and gets a cancellation that originated in a completely unrelated file.

It also bites on plain double-iteration: range the returned iter.Seq2, break early (which fires cancel(nil)), then range it again. once is already spent, so every group touched in the first pass returns the stale cancellation even though the fresh pass has a live context.

The post-Do block just below already does if err := ctx.Err(); err != nil { return nil, err } against the current caller's ctx, which is the right place to surface cancellation. I'd delete this in-Once check entirely and let that handle it. wdyt?

group.err = err

return
}
if len(bitmaps) != len(group.referencedDataFiles) {

@laskoviymishka laskoviymishka Sep 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Small one: ReadDVs either returns len(dvFiles) bitmaps or an error, never a short slice on success, so this branch can't fire, and if it somehow did, the index wiring just below would panic out-of-bounds first anyway. I'd either drop it or leave a //nolint note that it's a belt-and-suspenders assertion so it doesn't read as load-bearing.

group.err = fmt.Errorf("read deletion vectors from %s: got %d bitmaps, expected %d",
group.puffinPath, len(bitmaps), len(group.referencedDataFiles))

return
}

group.bitmaps = make(perFileDVBitmaps, len(bitmaps))
for i, ref := range group.referencedDataFiles {
group.bitmaps[ref] = bitmaps[i]
}
})

if err := ctx.Err(); err != nil {
return nil, err
}
if group.err != nil {
return nil, group.err
}

return group.bitmaps[dataFilePath], nil
}

func collectUniqueDeletionVectors(tasks []FileScanTask) (map[string]iceberg.DataFile, error) {
uniqueDVs := make(map[string]iceberg.DataFile)

for _, t := range tasks {
for _, d := range t.DeletionVectorFiles {
_, _, ref, contentOffset, contentSize := iceinternal.BorrowedDataFilePointers(d)
if ref == nil {
return nil, fmt.Errorf("deletion vector %s missing referenced_data_file", d.FilePath())
return nil, fmt.Errorf("%w: deletion vector %s missing referenced_data_file",
dv.ErrInvalidDeletionVector, d.FilePath())
}
if *ref == "" {
return nil, fmt.Errorf("%w: deletion vector %s missing or empty referenced_data_file",
dv.ErrInvalidDeletionVector, d.FilePath())
}
if contentOffset == nil || contentSize == nil {
// Spec §Manifest Files: content_offset and content_size_in_
Expand All @@ -191,6 +269,48 @@ func readAllDeletionVectors(ctx context.Context, fs iceio.IO, tasks []FileScanTa
}
}

return uniqueDVs, nil
}

func groupDeletionVectors(uniqueDVs map[string]iceberg.DataFile) map[string]*lazyDeletionVectorGroup {
groups := make(map[string]*lazyDeletionVectorGroup)
for ref, dvFile := range uniqueDVs {
group := groups[dvFile.FilePath()]
if group == nil {
group = &lazyDeletionVectorGroup{puffinPath: dvFile.FilePath()}
groups[dvFile.FilePath()] = group
}

group.referencedDataFiles = append(group.referencedDataFiles, ref)
group.files = append(group.files, dvFile)
}

return groups
}

// readAllDeletionVectors reads every deletion-vector puffin blob referenced
// by the input tasks and returns a perFileDVBitmaps map keyed by the
// referenced data-file path.
//
// Dedup is by referenced-data-file path, not by puffin file path: a single
// puffin file can carry multiple DV blobs (one per data file). Keying by the
// puffin path would silently drop all but the first blob. This matches Java's
// DeleteFileIndex.findDV, which keys by data-file path. As a side-effect we
// can detect spec violations: two distinct DV blobs targeting the same data
// file is rejected (mirrors Java's "Can't index multiple DVs for %s"
// ValidationException — over-deletion risk if silently unioned).
//
// Validation happens up front, before any goroutines are launched, so the
// goroutine fan-out has no early-exit path. (An early return after g.Go
// dispatches but before g.Wait would close resultsChan while in-flight
// workers were still sending, panicking with "send on closed channel".)
func readAllDeletionVectors(ctx context.Context, fs iceio.IO, tasks []FileScanTask, concurrency int) (perFileDVBitmaps, error) {
out := make(perFileDVBitmaps)
uniqueDVs, err := collectUniqueDeletionVectors(tasks)
if err != nil {
return nil, err
}

if len(uniqueDVs) == 0 {
return out, nil
}
Expand All @@ -209,20 +329,7 @@ func readAllDeletionVectors(ctx context.Context, fs iceio.IO, tasks []FileScanTa
return out, nil
}

type dvGroup struct {
referencedDataFiles []string
files []iceberg.DataFile
}
groups := make(map[string]*dvGroup)
for ref, dvFile := range uniqueDVs {
group := groups[dvFile.FilePath()]
if group == nil {
group = &dvGroup{}
groups[dvFile.FilePath()] = group
}
group.referencedDataFiles = append(group.referencedDataFiles, ref)
group.files = append(group.files, dvFile)
}
groups := groupDeletionVectors(uniqueDVs)

type dvResult struct {
referencedDataFiles []string
Expand Down Expand Up @@ -2057,7 +2164,7 @@ func createIterator(ctx context.Context, numWorkers uint, records <-chan enumera
}
}

func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context, tasks []FileScanTask, deletesPerFile perFilePosDeletes, dvBitmaps perFileDVBitmaps, equalityDeleteLoader *lazyEqualityDeleteLoader, invariants *arrowScanInvariants) iter.Seq2[arrow.RecordBatch, error] {
func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context, tasks []FileScanTask, deletesPerFile perFilePosDeletes, dvLoader *lazyDeletionVectorLoader, equalityDeleteLoader *lazyEqualityDeleteLoader, invariants *arrowScanInvariants) iter.Seq2[arrow.RecordBatch, error] {
return func(yield func(arrow.RecordBatch, error) bool) {
extSet := substrait.NewExtensionSet()

Expand Down Expand Up @@ -2086,6 +2193,13 @@ func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context, tasks
}

filePath := task.Value.File.FilePath()
dvBitmap, err := dvLoader.load(scanCtx, filePath)
if err != nil {
records <- enumeratedRecord{Task: task, Err: err}
cancel(err)

return
}
eqDeleteSets, err := equalityDeleteLoader.load(scanCtx, task.Value)
if err != nil {
records <- enumeratedRecord{Task: task, Err: err}
Expand All @@ -2096,7 +2210,7 @@ func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context, tasks

if err := as.recordsFromTask(scanCtx, task, records,
deletesPerFile[filePath],
dvBitmaps[filePath],
dvBitmap,
eqDeleteSets,
invariants); err != nil {
cancel(err)
Expand Down Expand Up @@ -2177,12 +2291,10 @@ func (as *arrowScan) GetRecords(ctx context.Context, tasks []FileScanTask) (*arr
return nil, nil, err
}

// DV bitmaps stay in their native form rather than being materialized
// into int64 positions and merged with the Parquet pos-delete map.
// filterByDeletionVector applies the bitmap to each batch via a Boolean
// keep-mask + compute.Filter — O(1) Contains lookups, vectorized Filter,
// no intermediate position set.
dvBitmaps, err := readAllDeletionVectors(ctx, as.fs, tasks, as.concurrency)
// Index DV ownership up front, but defer Puffin reads until a task using a
// referenced data file enters the iterator. The loader keeps each shared
// Puffin group cached after its first read.
dvLoader, err := newLazyDeletionVectorLoader(as.fs, tasks)

@laskoviymishka laskoviymishka Sep 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new commit lands the eager-structural-validation half of what I was after here: newLazyDeletionVectorLoader runs collectUniqueDeletionVectors synchronously, so missing/empty-ref and duplicate-DV violations still fail GetRecords up front (and now with a nice ErrInvalidDeletionVector sentinel). That's the right split.

What's left is the Puffin I/O errors, which used to surface from GetRecords and now only surface on first iteration. Two consequences worth a line of docs: a caller using err from GetRecords as a gate (if err != nil { return }; use(iter)) silently stops seeing DV read errors, and if it never iterates the error is dropped entirely; and if a filter or row-limit prunes away every task referencing a corrupt DV, the read error never surfaces at all, which diverges from Java, where DeleteFileIndex validates every DV before emitting rows. I don't think lazy-by-default is wrong, but I'd document the changed timing in the GetRecords godoc so callers know the DV-read error now rides on the iterator. wdyt?

if err != nil {
releasePerFilePosDeletes(deletesPerFile)

Expand All @@ -2199,5 +2311,5 @@ func (as *arrowScan) GetRecords(ctx context.Context, tasks []FileScanTask) (*arr
equalityDeleteLoader.addFieldIDs(invariants.projectedIDs)

return resultSchema, as.recordBatchesFromTasksAndDeletes(ctx, tasks,
deletesPerFile, dvBitmaps, equalityDeleteLoader, invariants), nil
deletesPerFile, dvLoader, equalityDeleteLoader, invariants), nil
}
Loading
Loading