laskoviymishka commented on code in PR #1968:
URL: https://github.com/apache/iceberg-go/pull/1968#discussion_r3903429250
##########
table/arrow_scanner.go:
##########
@@ -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 {
Review Comment:
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?
##########
table/lazy_deletion_vector_test.go:
##########
@@ -0,0 +1,226 @@
+// 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 (
+ "errors"
+ "path/filepath"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table/dv"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestLazyDeletionVectorLoaderLoadsSharedPuffinGroupsOnDemand(t *testing.T)
{
+ files := writeSharedDVPuffinFixture(t, 2)
+ refs := make([]string, len(files))
+ for i, file := range files {
+ require.NotNil(t, file.ReferencedDataFile())
+ refs[i] = *file.ReferencedDataFile()
+ }
+ tasks := []FileScanTask{
+ {DeletionVectorFiles: []iceberg.DataFile{files[0]}},
+ {DeletionVectorFiles: []iceberg.DataFile{files[1]}},
+ }
+ fs := &countingDVOpenIO{}
+
+ loader, err := newLazyDeletionVectorLoader(fs, tasks)
+ require.NoError(t, err)
+ require.Len(t, loader.groups, 1)
+ assert.Zero(t, fs.opens.Load(), "indexing must not open a Puffin file")
+
+ bitmap, err := loader.load(t.Context(), refs[0])
+ require.NoError(t, err)
+ require.NotNil(t, bitmap)
+ assert.True(t, bitmap.Contains(0))
+ assert.True(t, bitmap.Contains(2))
+ assert.Equal(t, int64(1), fs.opens.Load())
+
+ bitmap, err = loader.load(t.Context(), refs[1])
+ require.NoError(t, err)
+ require.NotNil(t, bitmap)
+ assert.True(t, bitmap.Contains(1))
+ assert.True(t, bitmap.Contains(3))
+ assert.Equal(t, int64(1), fs.opens.Load(), "a shared Puffin group must
be loaded once")
+}
+
+func TestLazyDeletionVectorLoaderRejectsEmptyReferencedDataFile(t *testing.T) {
+ dvFile := newDVMockDataFile("empty-ref.puffin", "", 0, 1, 1)
+
+ loader, err := newLazyDeletionVectorLoader(&countingDVOpenIO{},
[]FileScanTask{{
+ DeletionVectorFiles: []iceberg.DataFile{dvFile},
+ }})
+
+ assert.Nil(t, loader)
+ require.ErrorIs(t, err, dv.ErrInvalidDeletionVector)
+ assert.ErrorContains(t, err, "missing or empty referenced_data_file")
+}
+
+func TestLazyDeletionVectorLoaderSingleflightsConcurrentGroupLoads(t
*testing.T) {
+ files := writeSharedDVPuffinFixture(t, 2)
+ refs := make([]string, len(files))
+ for i, file := range files {
+ require.NotNil(t, file.ReferencedDataFile())
+ refs[i] = *file.ReferencedDataFile()
+ }
+ tasks := []FileScanTask{{DeletionVectorFiles: files}}
+ fs := &countingDVOpenIO{}
+ loader, err := newLazyDeletionVectorLoader(fs, tasks)
+ require.NoError(t, err)
+
+ const callers = 32
+ bitmaps := make([]*dv.RoaringPositionBitmap, callers)
+ errs := make([]error, callers)
+ var wg sync.WaitGroup
+ wg.Add(callers)
+ for i := range callers {
+ go func(i int) {
+ defer wg.Done()
+ bitmaps[i], errs[i] = loader.load(t.Context(),
refs[i%len(refs)])
+ }(i)
+ }
+ wg.Wait()
+
+ assert.Equal(t, int64(1), fs.opens.Load(), "concurrent callers must
share one group read")
+ for i := range callers {
+ require.NoError(t, errs[i])
+ require.NotNil(t, bitmaps[i])
+ }
+}
+
+type failingLazyDeletionVectorIO struct {
+ opens atomic.Int64
+ err error
+}
+
+func (f *failingLazyDeletionVectorIO) Open(string) (iceio.File, error) {
+ f.opens.Add(1)
+
+ return nil, f.err
+}
+
+func (f *failingLazyDeletionVectorIO) Remove(string) error { return nil }
+
+func TestLazyDeletionVectorLoaderCachesGroupErrors(t *testing.T) {
Review Comment:
Nice to see the new
`TestLazyDeletionVectorLoaderRejectsEmptyReferencedDataFile` covering the
empty-ref path. Once the `ctx.Err()`-inside-`once.Do` issue is fixed, I'd add a
sibling test right around here that pins it: load a group through an
already-cancelled context (or cancel right after `ReadDVs` returns), then load
the same group again with a fresh context and assert the second call returns
the real bitmaps rather than a cached cancellation. The singleflight and
error-caching tests are good, but neither exercises the succeed-then-poisoned
path, so that bug would regress silently.
##########
table/arrow_scanner.go:
##########
@@ -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)
Review Comment:
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?
##########
table/arrow_scanner.go:
##########
@@ -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 {
+ group.err = err
+
+ return
+ }
+ if len(bitmaps) != len(group.referencedDataFiles) {
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
