zeroshade commented on code in PR #1968:
URL: https://github.com/apache/iceberg-go/pull/1968#discussion_r3935766101
##########
table/scanner.go:
##########
@@ -1872,6 +1872,10 @@ type FileScanTask struct {
// If an error is encountered, during the planning and setup then this will
return the
// error directly. If the error occurs while iterating the records, it will be
returned
// by the iterator.
+// Deletion-vector references are validated during setup. Deletion-vector
Puffin files
+// are loaded lazily, so a Puffin read error is returned by the iterator when
a task
Review Comment:
**minor** — 'validated during setup' overstates what is checked eagerly
collectUniqueDeletionVectors validates ref presence/non-emptiness,
content_offset/content_size presence, and duplicate-DV conflicts. It does NOT
check file_format or the content_size range — those live in dv.validateDVFile,
which now only runs inside ReadDVs at demand time. So a structurally invalid DV
entry (e.g. file_format=PARQUET) passes setup and, if a filter or row limit
prunes away every task referencing it, is never validated at all, where main
failed GetRecords unconditionally. Suggest softening to something like
'Deletion-vector manifest references are validated during setup; format and
blob-level validation happens with the Puffin read.' Same text at :1894 on
ReadTasks.
##########
table/lazy_deletion_vector_test.go:
##########
@@ -0,0 +1,294 @@
+// 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 (
+ "context"
+ "errors"
+ "os"
+ "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.byDataFile, 2)
+ 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 TestReadTasksRejectsEmptyReferencedDataFile(t *testing.T) {
+ fs := iceio.LocalFS{}
+ tmp := t.TempDir()
+ tbl := buildDVScanTestTable(t, fs, tmp)
+ dataPath := filepath.Join(tmp, "data.parquet")
+ dataFile := writeIntParquetWithFieldID(t, fs, dataPath, 0, 1)
+ dvFile := newDVMockDataFile("empty-ref.puffin", "", 0, 1, 1)
+
+ _, _, err := tbl.Scan().ReadTasks(t.Context(), []FileScanTask{
+ {File: dataFile, DeletionVectorFiles:
[]iceberg.DataFile{dvFile}},
+ })
+ require.ErrorIs(t, err, dv.ErrInvalidDeletionVector)
+}
+
+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) {
+ const dataFilePath = "file:///table/data/missing.parquet"
+ offset, size := int64(0), int64(1)
+ dvFile := newDVMockDataFile("missing.puffin", dataFilePath, offset,
size, 1)
+ fs := &failingLazyDeletionVectorIO{err: errors.New("boom")}
+ loader, err := newLazyDeletionVectorLoader(fs,
+ []FileScanTask{{DeletionVectorFiles:
[]iceberg.DataFile{dvFile}}})
+ require.NoError(t, err)
+
+ const callers = 8
+ errs := make([]error, callers)
+ var wg sync.WaitGroup
+ wg.Add(callers)
+ for i := range callers {
+ go func(i int) {
+ defer wg.Done()
+ _, errs[i] = loader.load(t.Context(), dataFilePath)
+ }(i)
+ }
+ wg.Wait()
+
+ assert.Equal(t, int64(1), fs.opens.Load())
+ for _, loadErr := range errs {
+ require.Error(t, loadErr)
+ assert.ErrorContains(t, loadErr, "read deletion vectors from
missing.puffin")
+ assert.ErrorContains(t, loadErr, "boom")
+ }
+}
+
+type cancelOnOpenIO struct {
+ iceio.LocalFS
+ cancel context.CancelFunc
+ opens atomic.Int64
+}
+
+func (f *cancelOnOpenIO) Open(name string) (iceio.File, error) {
+ file, err := f.LocalFS.Open(name)
+ if err == nil {
+ f.opens.Add(1)
+ f.cancel()
+ }
+
+ return file, err
+}
+
+func TestLazyDeletionVectorLoaderDoesNotCacheCallerCancellation(t *testing.T) {
+ files := writeSharedDVPuffinFixture(t, 1)
+ dataFilePath := *files[0].ReferencedDataFile()
+ ctx, cancel := context.WithCancel(t.Context())
+ fs := &cancelOnOpenIO{cancel: cancel}
+ loader, err := newLazyDeletionVectorLoader(fs, []FileScanTask{{
+ DeletionVectorFiles: files,
+ }})
+ require.NoError(t, err)
+
+ bitmap, err := loader.load(ctx, dataFilePath)
+ assert.ErrorIs(t, err, context.Canceled)
+ assert.Nil(t, bitmap)
+
+ bitmap, err = loader.load(context.Background(), dataFilePath)
+ require.NoError(t, err)
+ require.NotNil(t, bitmap)
+ assert.True(t, bitmap.Contains(0))
+ assert.Equal(t, int64(1), fs.opens.Load())
+}
+
+func TestLazyDeletionVectorLoaderSurfacesPuffinReadErrors(t *testing.T) {
+ const dataFilePath = "file:///table/data/truncated.parquet"
+ puffinPath, offset, length, card := writeDVPuffinFixture(t,
[]uint64{1}, dataFilePath)
+ require.NoError(t, os.Truncate(puffinPath, 1))
+ dvFile := newDVMockDataFile(puffinPath, dataFilePath, offset, length,
card)
+ loader, err := newLazyDeletionVectorLoader(iceio.LocalFS{},
[]FileScanTask{{
+ DeletionVectorFiles: []iceberg.DataFile{dvFile},
+ }})
+ require.NoError(t, err)
+
+ _, err = loader.load(t.Context(), dataFilePath)
+ require.Error(t, err)
+ assert.ErrorContains(t, err, "read deletion vectors from")
+}
+
+type countingPuffinOpenIO struct {
+ base iceio.IO
+ opens atomic.Int64
+}
+
+func (f *countingPuffinOpenIO) Open(name string) (iceio.File, error) {
+ file, err := f.base.Open(name)
+ if err == nil && strings.HasSuffix(name, ".puffin") {
+ f.opens.Add(1)
+ }
+
+ return file, err
+}
+
+func (f *countingPuffinOpenIO) Remove(name string) error { return
f.base.Remove(name) }
+
+func TestReadTasksDoesNotLoadDeletionVectorsBeforeIteration(t *testing.T) {
+ baseFS := iceio.LocalFS{}
Review Comment:
**minor** — No test abandons a partially-consumed iterator
TestReadTasksDoesNotLoadDeletionVectorsBeforeIteration covers the
never-iterated case (opens==0) and the fully-drained case (opens==1), but
nothing breaks out of an iterator that has already forced a Puffin load. That
is the shape where a worker can be parked inside once.Do while the consumer
walks away, so it is worth pinning. A Close-counting iceio.IO asserting
opens==closes after an early break would cover it in ~20 lines.
##########
table/lazy_deletion_vector_bench_test.go:
##########
@@ -0,0 +1,157 @@
+// 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 (
+ "context"
+ "fmt"
+ "path/filepath"
+ "sync/atomic"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table/dv"
+ "golang.org/x/sync/errgroup"
+)
+
+const (
+ lazyDVBenchmarkGroupCount = 100
+ lazyDVBenchmarkDVsPerGroup = 10
+ lazyDVBenchmarkTotalDVs = lazyDVBenchmarkGroupCount *
lazyDVBenchmarkDVsPerGroup
+)
+
+type lazyDVBenchmarkFixture struct {
+ tasks []FileScanTask
+ groupStarts []string
+}
+
+type countingLazyDVOpenIO struct {
+ iceio.LocalFS
+ opens atomic.Int64
+}
+
+func (f *countingLazyDVOpenIO) Open(name string) (iceio.File, error) {
+ file, err := f.LocalFS.Open(name)
+ if err == nil {
+ f.opens.Add(1)
+ }
+
+ return file, err
+}
+
+func newLazyDVBenchmarkFixture(b *testing.B) lazyDVBenchmarkFixture {
+ b.Helper()
+
+ baseFS := iceio.LocalFS{}
+ root := b.TempDir()
+ fixture := lazyDVBenchmarkFixture{
+ tasks: make([]FileScanTask, 0, lazyDVBenchmarkTotalDVs),
+ groupStarts: make([]string, 0, lazyDVBenchmarkGroupCount),
+ }
+
+ for groupIndex := range lazyDVBenchmarkGroupCount {
+ path := filepath.Join(root, fmt.Sprintf("dv-%03d.puffin",
groupIndex))
+ writer := dv.NewDVWriter(baseFS, func(int32)
*iceberg.PartitionSpec {
+ return iceberg.UnpartitionedSpec
+ })
+
+ for offset := range lazyDVBenchmarkDVsPerGroup {
+ dataIndex := groupIndex*lazyDVBenchmarkDVsPerGroup +
offset
+ ref :=
fmt.Sprintf("file:///benchmark/data/data-%04d.parquet", dataIndex)
+ if offset == 0 {
+ fixture.groupStarts =
append(fixture.groupStarts, ref)
+ }
+ if err := writer.Add(ref, []int64{int64(offset)}, 0,
nil); err != nil {
+ b.Fatal(err)
+ }
+ }
+
+ files, err := writer.Flush(context.Background(), path)
+ if err != nil {
+ b.Fatal(err)
+ }
+ for _, file := range files {
+ fixture.tasks = append(fixture.tasks, FileScanTask{
+ DeletionVectorFiles: []iceberg.DataFile{file},
+ })
+ }
+ }
+
+ return fixture
+}
+
+func BenchmarkLazyDeletionVectorLoading(b *testing.B) {
+ fixture := newLazyDVBenchmarkFixture(b)
Review Comment:
**nit** — Description's benchmark table cites an 'eager all groups' row that
is no longer reproducible
BenchmarkLazyDeletionVectorLoading only defines lazy_unread /
lazy_first_group / lazy_ten_groups / lazy_full_scan. The 'eager all groups'
baseline came from BenchmarkReadAllDeletionVectorsSharedPuffin, deleted in
17f771a. Worth a note in the PR body that the baseline row was measured against
the pre-PR tree, so a reviewer does not go looking for the benchmark.
##########
table/arrow_scanner_test.go:
##########
@@ -430,7 +430,7 @@ func TestReleasePerFilePosDeletes(t *testing.T) {
t.Run("nil chunk in slice does not panic", func(t *testing.T) {
// Defensive: production code paths never insert a nil
*arrow.Chunked
// into the map. This subtest pins the guard so a future caller
(the
- // in-flight readAllDeletionVectors merger, or a refactor of
readDeletes)
+ // in-flight delete-file merger, or a refactor of readDeletes)
// can't silently NPE the cleanup path.
Review Comment:
**nit** — Comment points at a plan that no longer exists
The comment was rewritten from 'the in-flight readAllDeletionVectors merger'
to 'the in-flight delete-file merger' when the eager loader was deleted. There
is no in-flight merger any more; the sentence now names nothing. Suggest just
describing the guard ('a future caller, or a refactor of readDeletes').
--
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]
