laskoviymishka commented on code in PR #1906:
URL: https://github.com/apache/iceberg-go/pull/1906#discussion_r3874737643
##########
table/dv/deletion_vector.go:
##########
@@ -227,23 +229,78 @@ func ReadDVs(fs iceio.IO, dvFiles []iceberg.DataFile)
([]*RoaringPositionBitmap,
defer f.Close()
blobsByOffset := indexBlobMetadataByOffset(reader.Blobs())
- bitmaps := make([]*RoaringPositionBitmap, len(dvFiles))
+ type dvBlobRead struct {
+ index int
+ blob puffin.BlobMetadata
+ offset int64
+ manifestCardinality int64
+ }
+
+ reads := make([]dvBlobRead, 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)
+
+ manifestCardinality, err := validateDVBlobMetadata(blob,
dvFile, offset, manifestReferencedDataFile)
if err != nil {
return nil, err
}
+
+ reads[i] = dvBlobRead{
+ index: i,
+ blob: blob,
+ offset: offset,
+ manifestCardinality: manifestCardinality,
+ }
+ }
+
+ slices.SortFunc(reads, func(a, b dvBlobRead) int {
+ return cmp.Compare(a.offset, b.offset)
+ })
+
+ bitmaps := make([]*RoaringPositionBitmap, len(dvFiles))
+ for start := 0; start < len(reads); {
+ end := start + 1
+ rangeStart := reads[start].offset
+ rangeEnd := rangeStart + reads[start].blob.Length
+
+ for end < len(reads) {
+ next := reads[end]
+ nextEnd := next.offset + next.blob.Length
+ if next.offset > rangeEnd || nextEnd-rangeStart >
maxCoalescedDVRangeSize {
Review Comment:
This breaks coalescing on any gap between blobs. A single byte of alignment
padding between two DV blobs and we fall back to separate reads, silently, and
the puffin spec doesn't guarantee gap-free packing, so I'm a little wary of the
win quietly evaporating on files we didn't write.
Would it be worth tolerating a small gap (read across it and discard the
slack), or at least a `slog.Debug` when a gap blocks a merge? Totally fine to
punt if contiguous-only is the intended scope, I'd just want it to be a
deliberate call. wdyt?
##########
table/dv/deletion_vector.go:
##########
@@ -227,23 +229,78 @@ func ReadDVs(fs iceio.IO, dvFiles []iceberg.DataFile)
([]*RoaringPositionBitmap,
defer f.Close()
blobsByOffset := indexBlobMetadataByOffset(reader.Blobs())
- bitmaps := make([]*RoaringPositionBitmap, len(dvFiles))
+ type dvBlobRead struct {
+ index int
+ blob puffin.BlobMetadata
+ offset int64
+ manifestCardinality int64
+ }
+
+ reads := make([]dvBlobRead, 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)
+
+ manifestCardinality, err := validateDVBlobMetadata(blob,
dvFile, offset, manifestReferencedDataFile)
if err != nil {
return nil, err
}
+
+ reads[i] = dvBlobRead{
+ index: i,
+ blob: blob,
+ offset: offset,
+ manifestCardinality: manifestCardinality,
+ }
+ }
+
+ slices.SortFunc(reads, func(a, b dvBlobRead) int {
+ return cmp.Compare(a.offset, b.offset)
+ })
+
+ bitmaps := make([]*RoaringPositionBitmap, len(dvFiles))
+ for start := 0; start < len(reads); {
+ end := start + 1
+ rangeStart := reads[start].offset
+ rangeEnd := rangeStart + reads[start].blob.Length
+
+ for end < len(reads) {
+ next := reads[end]
+ nextEnd := next.offset + next.blob.Length
+ if next.offset > rangeEnd || nextEnd-rangeStart >
maxCoalescedDVRangeSize {
+ break
+ }
+
+ rangeEnd = max(rangeEnd, nextEnd)
+ end++
+ }
+
+ rangeData := make([]byte, rangeEnd-rangeStart)
+ if _, err := reader.ReadAt(rangeData, rangeStart); err != nil {
+ return nil, fmt.Errorf("read DV blob range at offset
%d: %w", rangeStart, err)
+ }
+
+ for _, read := range reads[start:end] {
+ blobStart := read.offset - rangeStart
+ blobEnd := blobStart + read.blob.Length
+ bitmaps[read.index], err =
DeserializeDV(rangeData[blobStart:blobEnd], read.manifestCardinality)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ start = end
}
return bitmaps, nil
}
+const maxCoalescedDVRangeSize int64 = 8 << 20
Review Comment:
I'd move this up into the const block with `DVMagicNumber` and friends.
Right now it's declared after its only caller, so you read the whole coalescing
loop before finding out what the cap actually is.
While we're here, the cap only gates whether a second blob gets merged into
the window. A lone blob still allocates its full length, bounded only by
`validateDVFile`'s 256 MiB check, so the description reading like 8 MiB bounds
peak allocation isn't quite right. A one-line comment saying it only bounds
multi-blob extension would save the next reader the same double-take.
##########
table/dv/deletion_vector.go:
##########
@@ -227,23 +229,78 @@ func ReadDVs(fs iceio.IO, dvFiles []iceberg.DataFile)
([]*RoaringPositionBitmap,
defer f.Close()
blobsByOffset := indexBlobMetadataByOffset(reader.Blobs())
- bitmaps := make([]*RoaringPositionBitmap, len(dvFiles))
+ type dvBlobRead struct {
+ index int
+ blob puffin.BlobMetadata
+ offset int64
+ manifestCardinality int64
+ }
+
+ reads := make([]dvBlobRead, 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)
+
+ manifestCardinality, err := validateDVBlobMetadata(blob,
dvFile, offset, manifestReferencedDataFile)
if err != nil {
return nil, err
}
+
+ reads[i] = dvBlobRead{
+ index: i,
+ blob: blob,
+ offset: offset,
+ manifestCardinality: manifestCardinality,
+ }
+ }
+
+ slices.SortFunc(reads, func(a, b dvBlobRead) int {
+ return cmp.Compare(a.offset, b.offset)
+ })
+
+ bitmaps := make([]*RoaringPositionBitmap, len(dvFiles))
+ for start := 0; start < len(reads); {
+ end := start + 1
+ rangeStart := reads[start].offset
+ rangeEnd := rangeStart + reads[start].blob.Length
+
+ for end < len(reads) {
+ next := reads[end]
+ nextEnd := next.offset + next.blob.Length
+ if next.offset > rangeEnd || nextEnd-rangeStart >
maxCoalescedDVRangeSize {
+ break
+ }
+
+ rangeEnd = max(rangeEnd, nextEnd)
+ end++
+ }
+
+ rangeData := make([]byte, rangeEnd-rangeStart)
+ if _, err := reader.ReadAt(rangeData, rangeStart); err != nil {
+ return nil, fmt.Errorf("read DV blob range at offset
%d: %w", rangeStart, err)
+ }
+
+ for _, read := range reads[start:end] {
+ blobStart := read.offset - rangeStart
+ blobEnd := blobStart + read.blob.Length
+ bitmaps[read.index], err =
DeserializeDV(rangeData[blobStart:blobEnd], read.manifestCardinality)
Review Comment:
Two things here. `err` is the function-scoped variable from `openDVReader`
up top, so a future `:=` added in between would silently rebind it out from
under this assignment. And the bare `return nil, err` drops the file/offset
context that the `ReadAt` error just above bothers to attach, so one corrupt
blob in a batch read surfaces opaquely.
A local binding cleans up both:
```go
bitmap, err := DeserializeDV(rangeData[blobStart:blobEnd],
read.manifestCardinality)
if err != nil {
return nil, fmt.Errorf("deserialize DV blob at offset %d: %w",
read.offset, err)
}
bitmaps[read.index] = bitmap
```
##########
table/dv/deletion_vector_bench_test.go:
##########
@@ -0,0 +1,111 @@
+// 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 dv
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/puffin"
+)
+
+func BenchmarkReadDVs(b *testing.B) {
+ for _, numDVs := range []int{2, 16, 64} {
+ b.Run(fmt.Sprintf("dvs=%d", numDVs), func(b *testing.B) {
+ files := benchmarkDVFiles(b, numDVs)
+ fs := &countingReadIO{base: iceio.LocalFS{}}
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ fs.reads = 0
+ bitmaps, err := ReadDVs(fs, files)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if len(bitmaps) != numDVs {
+ b.Fatalf("got %d bitmaps, want %d",
len(bitmaps), numDVs)
+ }
+ }
+ b.ReportMetric(float64(fs.reads), "range-reads/op")
+ b.StopTimer()
+ })
+ }
+}
+
+func benchmarkDVFiles(b *testing.B, numDVs int) []iceberg.DataFile {
+ b.Helper()
+
+ path := filepath.Join(b.TempDir(), "deletion-vectors.puffin")
+ f, err := os.Create(path)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ writer, err := puffin.NewWriter(f)
+ if err != nil {
+ _ = f.Close()
+ b.Fatal(err)
+ }
+
+ files := make([]iceberg.DataFile, numDVs)
+ for i := range numDVs {
+ bitmap := NewRoaringPositionBitmap()
+ bitmap.Set(uint64(i))
+ data, err := SerializeDV(bitmap)
+ if err != nil {
+ _ = f.Close()
+ b.Fatal(err)
+ }
+
+ referencedDataFile := fmt.Sprintf("data-%03d.parquet", i)
+ meta, err := writer.AddBlob(puffin.BlobMetadataInput{
+ Type: puffin.BlobTypeDeletionVector,
+ SnapshotID: -1,
+ SequenceNumber: -1,
+ Fields: []int32{},
Review Comment:
Inert today, but every other fixture in this package uses `Fields:
[]int32{2147483546}` and this one passes an empty slice. I'd match the
convention (or pull it into a shared constant) so the bench fixture doesn't
look like it's exercising something different.
##########
table/dv/deletion_vector_bench_test.go:
##########
@@ -0,0 +1,111 @@
+// 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 dv
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/puffin"
+)
+
+func BenchmarkReadDVs(b *testing.B) {
+ for _, numDVs := range []int{2, 16, 64} {
+ b.Run(fmt.Sprintf("dvs=%d", numDVs), func(b *testing.B) {
+ files := benchmarkDVFiles(b, numDVs)
+ fs := &countingReadIO{base: iceio.LocalFS{}}
+
+ b.ReportAllocs()
+ b.ResetTimer()
Review Comment:
`b.Loop()` resets the timer itself, so this `ResetTimer()` and the
`StopTimer()` after the loop are both no-ops. I'd drop them. Setup is already
outside the loop, and leaving them in tends to get copied into `b.N`-style
benches where the placement does matter.
##########
table/dv/deletion_vector_test.go:
##########
@@ -531,6 +560,53 @@ func TestReadDVs(t *testing.T) {
})
}
+func TestReadDVsCoalescesAdjacentBlobReads(t *testing.T) {
Review Comment:
This is the only test that positively asserts coalescing, and it only covers
the contiguous happy path, so neither break in the loop gets exercised. I'd add
two cases: two adjacent blobs whose combined size crosses
`maxCoalescedDVRangeSize` (expect them read separately), and two blobs with a
gap between them (expect separate reads). Without those, a `>` to `>=` slip in
either condition, or a change to the cap, lands green.
##########
table/dv/deletion_vector_test.go:
##########
@@ -531,6 +560,53 @@ func TestReadDVs(t *testing.T) {
})
}
+func TestReadDVsCoalescesAdjacentBlobReads(t *testing.T) {
+ dir := t.TempDir()
+ first := NewRoaringPositionBitmap()
+ first.Set(1)
+ firstData, err := SerializeDV(first)
+ require.NoError(t, err)
+
+ second := NewRoaringPositionBitmap()
+ second.Set(2)
+ secondData, err := SerializeDV(second)
+ require.NoError(t, err)
+
+ path, metas := writePuffinWithDVBlobs(t, dir,
+ testPuffinBlobInput{
+ data: firstData,
+ props: map[string]string{
+ dvReferencedDataFileProperty:
"data-001.parquet",
+ dvCardinalityProperty: "1",
+ },
+ },
+ testPuffinBlobInput{
+ data: secondData,
+ props: map[string]string{
+ dvReferencedDataFileProperty:
"data-002.parquet",
+ dvCardinalityProperty: "1",
+ },
+ },
+ )
+
+ firstOffset, firstSize := metas[0].Offset, metas[0].Length
+ secondOffset, secondSize := metas[1].Offset, metas[1].Length
+ files := []iceberg.DataFile{
+ newDVTestFile(path, 1, &secondOffset, &secondSize),
+ newDVTestFile(path, 1, &firstOffset, &firstSize),
+ }
+ files[0].(*mockDVFile).referencedDataFile = strPtr("data-002.parquet")
+ files[1].(*mockDVFile).referencedDataFile = strPtr("data-001.parquet")
+
+ fs := &countingReadIO{base: iceio.LocalFS{}}
+ bitmaps, err := ReadDVs(fs, files)
+ require.NoError(t, err)
+ require.Len(t, bitmaps, 2)
+ assert.True(t, bitmaps[0].Contains(2))
+ assert.True(t, bitmaps[1].Contains(1))
+ assert.Equal(t, 3, fs.reads)
Review Comment:
The 3 here isn't only the coalesced data read, it folds in
`puffin.NewReader`'s own header and footer reads, so this assertion is really
pinned to puffin's internal read strategy. If puffin ever folds its header into
the footer read the count drops and this fails even though coalescing is fine;
conversely coalescing could regress from one blob read to two while puffin
happens to save an init read, and this still passes.
I'd assert the property we actually care about rather than the absolute
total: something like `fs.reads < len(files) + puffinInitReads`, or wrap the
`*puffin.Reader` in `countingReadFile` so it only counts reads on the
coalescing path. Either way a short comment breaking down where the number
comes from would help, since `3` reads as magic right now.
--
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]
