You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/21 19:33:35 UTC
(iceberg-go) branch main updated: perf(table): group deletion vector reads by Puffin file (#1764)
This is an automated email from the ASF dual-hosted git repository.
zeroshade 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 b4a3e9d1e perf(table): group deletion vector reads by Puffin file (#1764)
b4a3e9d1e is described below
commit b4a3e9d1eec16e88baddc61f008ed4960e5cb6f5
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Fri Aug 21 21:33:29 2026 +0200
perf(table): group deletion vector reads by Puffin file (#1764)
---
table/arrow_scanner.go | 45 ++++++++++--
table/dv/deletion_vector.go | 149 ++++++++++++++++++++++++++++++++-------
table/dv/deletion_vector_test.go | 65 +++++++++++++++++
table/dv_scanner_read_test.go | 93 ++++++++++++++++++++++++
4 files changed, 321 insertions(+), 31 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index f25eeb778..33b81995c 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -191,10 +191,39 @@ func readAllDeletionVectors(ctx context.Context, fs iceio.IO, tasks []FileScanTa
if len(uniqueDVs) == 0 {
return out, nil
}
+ if len(uniqueDVs) == 1 {
+ for ref, dvFile := range uniqueDVs {
+ bitmap, err := dv.ReadDV(fs, dvFile)
+ if err != nil {
+ return nil, fmt.Errorf("read deletion vector %s: %w", dvFile.FilePath(), err)
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ out[ref] = bitmap
+ }
+
+ 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)
+ }
type dvResult struct {
- referencedDataFile string
- bitmap *dv.RoaringPositionBitmap
+ referencedDataFiles []string
+ bitmaps []*dv.RoaringPositionBitmap
}
g, gctx := errgroup.WithContext(ctx)
@@ -207,14 +236,14 @@ func readAllDeletionVectors(ctx context.Context, fs iceio.IO, tasks []FileScanTa
// panic on send to a closed channel. The error is collected via
// the outer g.Wait() below, not stored on a shared variable.
defer close(resultsChan)
- for ref, dvFile := range uniqueDVs {
+ for puffinPath, group := range groups {
g.Go(func() error {
- bitmap, err := dv.ReadDV(fs, dvFile)
+ bitmaps, err := dv.ReadDVs(fs, group.files)
if err != nil {
- return fmt.Errorf("read deletion vector %s: %w", dvFile.FilePath(), err)
+ return fmt.Errorf("read deletion vectors from %s: %w", puffinPath, err)
}
select {
- case resultsChan <- dvResult{referencedDataFile: ref, bitmap: bitmap}:
+ case resultsChan <- dvResult{referencedDataFiles: group.referencedDataFiles, bitmaps: bitmaps}:
return nil
case <-gctx.Done():
return gctx.Err()
@@ -225,7 +254,9 @@ func readAllDeletionVectors(ctx context.Context, fs iceio.IO, tasks []FileScanTa
}()
for r := range resultsChan {
- out[r.referencedDataFile] = r.bitmap
+ for i, ref := range r.referencedDataFiles {
+ out[ref] = r.bitmaps[i]
+ }
}
if err := g.Wait(); err != nil {
diff --git a/table/dv/deletion_vector.go b/table/dv/deletion_vector.go
index 06991be98..94ed4c291 100644
--- a/table/dv/deletion_vector.go
+++ b/table/dv/deletion_vector.go
@@ -173,39 +173,115 @@ func SerializeDV(bitmap *RoaringPositionBitmap) ([]byte, error) {
// than rejected — the Go writer always emits the property, but third-party
// writers may not, and the per-byte CRC check in DeserializeDV still applies.
func ReadDV(fs iceio.IO, dvFile iceberg.DataFile) (*RoaringPositionBitmap, error) {
+ if err := validateDVFile(dvFile); err != nil {
+ return nil, err
+ }
+
+ reader, f, err := openDVReader(fs, dvFile.FilePath())
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+
+ _, _, manifestReferencedDataFile, contentOffset, contentSize := iceberginternal.BorrowedDataFilePointers(dvFile)
+ offset, size := *contentOffset, *contentSize
+ blob, err := findBlobMetadataByRange(reader.Blobs(), offset, size)
+ if err != nil {
+ return nil, fmt.Errorf("%w: DV file %s: %w", ErrInvalidDeletionVector, dvFile.FilePath(), err)
+ }
+
+ return readDV(reader, blob, dvFile, offset, manifestReferencedDataFile)
+}
+
+// ReadDVs reads multiple deletion vectors stored in the same Puffin file.
+// The returned bitmaps have the same order as dvFiles. The Puffin file is
+// opened and its footer is decoded once for the complete batch.
+func ReadDVs(fs iceio.IO, dvFiles []iceberg.DataFile) ([]*RoaringPositionBitmap, error) {
+ if len(dvFiles) == 0 {
+ return nil, nil
+ }
+ if len(dvFiles) == 1 {
+ bitmap, err := ReadDV(fs, dvFiles[0])
+ if err != nil {
+ return nil, err
+ }
+
+ return []*RoaringPositionBitmap{bitmap}, nil
+ }
+
+ filePath := dvFiles[0].FilePath()
+ for i, dvFile := range dvFiles {
+ if dvFile.FilePath() != filePath {
+ return nil, fmt.Errorf("%w: deletion vector at index %d uses Puffin file %q, expected %q",
+ iceberg.ErrInvalidArgument, i, dvFile.FilePath(), filePath)
+ }
+ if err := validateDVFile(dvFile); err != nil {
+ return nil, err
+ }
+ }
+
+ reader, f, err := openDVReader(fs, filePath)
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+
+ blobsByOffset := indexBlobMetadataByOffset(reader.Blobs())
+ bitmaps := make([]*RoaringPositionBitmap, len(dvFiles))
+ for i, dvFile := range dvFiles {
+ _, _, manifestReferencedDataFile, contentOffset, contentSize := iceberginternal.BorrowedDataFilePointers(dvFile)
+ offset, size := *contentOffset, *contentSize
+ blob, err := findIndexedBlobMetadataByRange(blobsByOffset, offset, size)
+ if err != nil {
+ return nil, fmt.Errorf("%w: DV file %s: %w", ErrInvalidDeletionVector, dvFile.FilePath(), err)
+ }
+ bitmaps[i], err = readDV(reader, blob, dvFile, offset, manifestReferencedDataFile)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ return bitmaps, nil
+}
+
+func validateDVFile(dvFile iceberg.DataFile) error {
if dvFile.FileFormat() != iceberg.PuffinFile {
- return nil, fmt.Errorf("expected PUFFIN format for deletion vector, got %s", dvFile.FileFormat())
+ return fmt.Errorf("expected PUFFIN format for deletion vector, got %s", dvFile.FileFormat())
}
_, _, manifestReferencedDataFile, contentOffset, contentSize := iceberginternal.BorrowedDataFilePointers(dvFile)
if contentOffset == nil || contentSize == nil {
- return nil, fmt.Errorf("DV file %s missing ContentOffset/ContentSizeInBytes", dvFile.FilePath())
+ return fmt.Errorf("DV file %s missing ContentOffset/ContentSizeInBytes", dvFile.FilePath())
}
if manifestReferencedDataFile == nil || *manifestReferencedDataFile == "" {
- return nil, fmt.Errorf("%w: DV file %s missing or empty %s property", ErrInvalidDeletionVector, dvFile.FilePath(), dvReferencedDataFileProperty)
+ return fmt.Errorf("%w: DV file %s missing or empty %s property", ErrInvalidDeletionVector, dvFile.FilePath(), dvReferencedDataFileProperty)
}
size := *contentSize
if size < 0 || size > int64(puffin.DefaultMaxBlobSize) {
- return nil, fmt.Errorf("DV blob size %d out of valid range [0, %d]", size, puffin.DefaultMaxBlobSize)
+ return fmt.Errorf("DV blob size %d out of valid range [0, %d]", size, puffin.DefaultMaxBlobSize)
}
- f, err := fs.Open(dvFile.FilePath())
+ return nil
+}
+
+func openDVReader(fs iceio.IO, filePath string) (*puffin.Reader, iceio.File, error) {
+ f, err := fs.Open(filePath)
if err != nil {
- return nil, fmt.Errorf("open DV file %s: %w", dvFile.FilePath(), err)
+ return nil, nil, fmt.Errorf("open DV file %s: %w", filePath, err)
}
- defer f.Close()
reader, err := puffin.NewReader(f)
if err != nil {
- return nil, fmt.Errorf("create puffin reader for %s: %w", dvFile.FilePath(), err)
- }
+ _ = f.Close()
- offset := *contentOffset
- blob, err := findBlobMetadataByRange(reader.Blobs(), offset, size)
- if err != nil {
- return nil, fmt.Errorf("%w: DV file %s: %w", ErrInvalidDeletionVector, dvFile.FilePath(), err)
+ return nil, nil, fmt.Errorf("create puffin reader for %s: %w", filePath, err)
}
+
+ return reader, f, nil
+}
+
+func readDV(reader *puffin.Reader, blob puffin.BlobMetadata, dvFile iceberg.DataFile, offset int64, manifestReferencedDataFile *string) (*RoaringPositionBitmap, error) {
if blob.Type != puffin.BlobTypeDeletionVector {
return nil, fmt.Errorf("%w: DV file %s: blob at offset %d has type %q, expected %q", ErrInvalidDeletionVector,
dvFile.FilePath(), offset, blob.Type, puffin.BlobTypeDeletionVector)
@@ -255,7 +331,7 @@ func ReadDV(fs iceio.IO, dvFile iceberg.DataFile) (*RoaringPositionBitmap, error
"dv_file", dvFile.FilePath(), "offset", offset)
}
- blobData := make([]byte, size)
+ blobData := make([]byte, blob.Length)
if _, err := reader.ReadAt(blobData, offset); err != nil {
return nil, fmt.Errorf("read DV blob at offset %d: %w", offset, err)
}
@@ -266,28 +342,53 @@ func ReadDV(fs iceio.IO, dvFile iceberg.DataFile) (*RoaringPositionBitmap, error
return DeserializeDV(blobData, manifestCardinality)
}
+func indexBlobMetadataByOffset(blobs []puffin.BlobMetadata) map[int64]puffin.BlobMetadata {
+ indexed := make(map[int64]puffin.BlobMetadata, len(blobs))
+ for _, blob := range blobs {
+ if _, exists := indexed[blob.Offset]; !exists {
+ indexed[blob.Offset] = blob
+ }
+ }
+
+ return indexed
+}
+
// findBlobMetadataByRange returns the footer entry identified by the
// manifest's content offset and size. On error it returns a zero-value
// puffin.BlobMetadata; callers must check the error before reading the value.
func findBlobMetadataByRange(blobs []puffin.BlobMetadata, offset, size int64) (puffin.BlobMetadata, error) {
- for _, b := range blobs {
- if b.Offset != offset {
+ for _, blob := range blobs {
+ if blob.Offset != offset {
continue
}
- if b.Length != size {
- // Same starting offset, different length: the manifest entry
- // disagrees with the puffin footer on how big this blob is.
- // Surface that distinct condition rather than rolling it into
- // "no blob at offset" — different writer bug, different fix.
- return puffin.BlobMetadata{}, fmt.Errorf("blob at offset %d has length %d, manifest says %d", offset, b.Length, size)
- }
- return b, nil
+ return validateBlobMetadataSize(blob, offset, size)
}
return puffin.BlobMetadata{}, fmt.Errorf("no blob in puffin footer at offset %d, size %d", offset, size)
}
+func findIndexedBlobMetadataByRange(blobsByOffset map[int64]puffin.BlobMetadata, offset, size int64) (puffin.BlobMetadata, error) {
+ blob, ok := blobsByOffset[offset]
+ if !ok {
+ return puffin.BlobMetadata{}, fmt.Errorf("no blob in puffin footer at offset %d, size %d", offset, size)
+ }
+
+ return validateBlobMetadataSize(blob, offset, size)
+}
+
+func validateBlobMetadataSize(blob puffin.BlobMetadata, offset, size int64) (puffin.BlobMetadata, error) {
+ if blob.Length != size {
+ // Same starting offset, different length: the manifest entry
+ // disagrees with the puffin footer on how big this blob is.
+ // Surface that distinct condition rather than rolling it into
+ // "no blob at offset" — different writer bug, different fix.
+ return puffin.BlobMetadata{}, fmt.Errorf("blob at offset %d has length %d, manifest says %d", offset, blob.Length, size)
+ }
+
+ return blob, nil
+}
+
// blobCardinality returns the cardinality declared by a Puffin blob. The bool
// indicates whether the property was present.
func blobCardinality(blob puffin.BlobMetadata) (int64, bool, error) {
diff --git a/table/dv/deletion_vector_test.go b/table/dv/deletion_vector_test.go
index 181c3d8b7..d384ed666 100644
--- a/table/dv/deletion_vector_test.go
+++ b/table/dv/deletion_vector_test.go
@@ -22,6 +22,7 @@ import (
"encoding/binary"
"encoding/json"
"errors"
+ "fmt"
"hash/crc32"
"os"
"path/filepath"
@@ -466,6 +467,70 @@ func TestReadDV(t *testing.T) {
assert.False(t, bm.Contains(2))
}
+func TestReadDVs(t *testing.T) {
+ dvBlobBytes := readDVTestData(t, "small-alternating-values-position-index.bin")
+ dir := t.TempDir()
+ path, metas := writePuffinWithDVBlobs(t, dir,
+ testPuffinBlobInput{
+ data: dvBlobBytes,
+ props: map[string]string{
+ dvReferencedDataFileProperty: "s3://bucket/data/data-001.parquet",
+ dvCardinalityProperty: "5",
+ },
+ },
+ testPuffinBlobInput{
+ data: dvBlobBytes,
+ props: map[string]string{
+ dvReferencedDataFileProperty: "s3://bucket/data/data-002.parquet",
+ dvCardinalityProperty: "5",
+ },
+ },
+ )
+
+ files := make([]iceberg.DataFile, len(metas))
+ for i, meta := range metas {
+ offset, size := meta.Offset, meta.Length
+ file := newDVTestFile(path, 5, &offset, &size)
+ file.referencedDataFile = strPtr(fmt.Sprintf("s3://bucket/data/data-%03d.parquet", i+1))
+ files[i] = file
+ }
+
+ bitmaps, err := ReadDVs(iceio.LocalFS{}, files)
+ require.NoError(t, err)
+ require.Len(t, bitmaps, 2)
+ for _, bitmap := range bitmaps {
+ assert.Equal(t, int64(5), bitmap.Cardinality())
+ assert.True(t, bitmap.Contains(1))
+ assert.True(t, bitmap.Contains(9))
+ }
+
+ t.Run("empty batch", func(t *testing.T) {
+ bitmaps, err := ReadDVs(iceio.LocalFS{}, nil)
+ require.NoError(t, err)
+ assert.Nil(t, bitmaps)
+ })
+
+ t.Run("different Puffin files", func(t *testing.T) {
+ first := files[0].(*mockDVFile)
+ second := *files[1].(*mockDVFile)
+ second.path = filepath.Join(t.TempDir(), "other.puffin")
+
+ _, err := ReadDVs(iceio.LocalFS{}, []iceberg.DataFile{first, &second})
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ assert.ErrorContains(t, err, "uses Puffin file")
+ })
+
+ t.Run("invalid blob identity", func(t *testing.T) {
+ first := files[0].(*mockDVFile)
+ second := *files[1].(*mockDVFile)
+ second.referencedDataFile = strPtr("s3://bucket/data/wrong.parquet")
+
+ _, err := ReadDVs(iceio.LocalFS{}, []iceberg.DataFile{first, &second})
+ require.ErrorIs(t, err, ErrInvalidDeletionVector)
+ assert.ErrorContains(t, err, "manifest referenced_data_file")
+ })
+}
+
// Why: ReadDV should reject callers that pass the wrong file type before doing any I/O.
// Condition: DataFile reports Parquet format instead of Puffin.
// Assertion: returns an error containing "expected PUFFIN format".
diff --git a/table/dv_scanner_read_test.go b/table/dv_scanner_read_test.go
index 2be2fff07..3696363d3 100644
--- a/table/dv_scanner_read_test.go
+++ b/table/dv_scanner_read_test.go
@@ -23,6 +23,7 @@ import (
"os"
"path/filepath"
"strconv"
+ "sync/atomic"
"testing"
"github.com/apache/arrow-go/v18/arrow"
@@ -101,6 +102,42 @@ func newDVMockDataFile(puffinPath, referencedDataFile string, offset, contentSiz
}
}
+func writeSharedDVPuffinFixture(t testing.TB, count int) []iceberg.DataFile {
+ t.Helper()
+
+ location := filepath.Join(t.TempDir(), "shared-dv.puffin")
+ w := dv.NewDVWriter(iceio.LocalFS{}, func(specID int32) *iceberg.PartitionSpec {
+ if specID == 0 {
+ return iceberg.UnpartitionedSpec
+ }
+
+ return nil
+ })
+ for i := range count {
+ path := "file:///table/data/data-" + strconv.Itoa(i) + ".parquet"
+ require.NoError(t, w.Add(path, []int64{int64(i), int64(i + count)}, 0, nil))
+ }
+
+ files, err := w.Flush(context.Background(), location)
+ require.NoError(t, err)
+
+ return files
+}
+
+type countingDVOpenIO struct {
+ iceio.LocalFS
+ opens atomic.Int64
+}
+
+func (c *countingDVOpenIO) Open(name string) (iceio.File, error) {
+ f, err := c.LocalFS.Open(name)
+ if err == nil {
+ c.opens.Add(1)
+ }
+
+ return f, err
+}
+
// Compile-time assertion of readAllDeletionVectors' signature — pins the
// `perFileDVBitmaps` return type so GetRecords' downstream wiring through
// recordBatchesFromTasksAndDeletes can't silently shift to a different shape
@@ -168,6 +205,40 @@ func TestReadAllDeletionVectors(t *testing.T) {
assert.Equal(t, int64(2), got[dataFilePath].Cardinality())
})
+ t.Run("opens a shared Puffin file once", func(t *testing.T) {
+ const count = 100
+ files := writeSharedDVPuffinFixture(t, count)
+ fs := &countingDVOpenIO{}
+
+ got, err := readAllDeletionVectors(ctx, fs,
+ []FileScanTask{{DeletionVectorFiles: files}}, 8)
+ require.NoError(t, err)
+ require.Len(t, got, count)
+ assert.Equal(t, int64(1), fs.opens.Load())
+ for i := range count {
+ path := "file:///table/data/data-" + strconv.Itoa(i) + ".parquet"
+ bitmap := got[path]
+ require.NotNil(t, bitmap)
+ assert.True(t, bitmap.Contains(uint64(i)))
+ assert.True(t, bitmap.Contains(uint64(i+count)))
+ }
+ })
+
+ t.Run("canceled singleton read returns the context error", func(t *testing.T) {
+ const dataFilePath = "file:///table/data/data-canceled.parquet"
+ puffinPath, offset, length, card := writeDVPuffinFixture(t, []uint64{1}, dataFilePath)
+ canceledCtx, cancel := context.WithCancel(ctx)
+ cancel()
+
+ got, err := readAllDeletionVectors(canceledCtx, fs, []FileScanTask{{
+ DeletionVectorFiles: []iceberg.DataFile{
+ newDVMockDataFile(puffinPath, dataFilePath, offset, length, card),
+ },
+ }}, 1)
+ require.ErrorIs(t, err, context.Canceled)
+ assert.Nil(t, got)
+ })
+
t.Run("no tasks -> empty map, no error", func(t *testing.T) {
got, err := readAllDeletionVectors(ctx, fs, nil, 1)
require.NoError(t, err)
@@ -278,6 +349,28 @@ func TestReadAllDeletionVectors(t *testing.T) {
})
}
+func BenchmarkReadAllDeletionVectorsSharedPuffin(b *testing.B) {
+ for _, count := range []int{1, 10, 100} {
+ b.Run(strconv.Itoa(count), func(b *testing.B) {
+ files := writeSharedDVPuffinFixture(b, count)
+ tasks := []FileScanTask{{DeletionVectorFiles: files}}
+ fs := iceio.LocalFS{}
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ bitmaps, err := readAllDeletionVectors(context.Background(), fs, tasks, 8)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if len(bitmaps) != count {
+ b.Fatalf("got %d bitmaps, expected %d", len(bitmaps), count)
+ }
+ }
+ })
+ }
+}
+
// TestFilterByDeletionVector pins the per-batch pipeline step that applies
// a RoaringPositionBitmap to a record batch via Boolean keep-mask +
// compute.Filter — the load-bearing optimization over materializing the