This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 366dc2041 perf(puffin): sort blob indexes when reading all blobs
(#2143)
366dc2041 is described below
commit 366dc20410649bad7d24aa12bfcd684937064ad6
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 12 20:24:44 2026 +0200
perf(puffin): sort blob indexes when reading all blobs (#2143)
---
puffin/puffin_reader.go | 30 +++++-----
puffin/puffin_reader_bench_test.go | 112 +++++++++++++++++++++++++++++++++++++
puffin/puffin_test.go | 31 ++++++++++
3 files changed, 156 insertions(+), 17 deletions(-)
diff --git a/puffin/puffin_reader.go b/puffin/puffin_reader.go
index c7b3b58ce..186876105 100644
--- a/puffin/puffin_reader.go
+++ b/puffin/puffin_reader.go
@@ -20,6 +20,7 @@ package puffin
import (
"bufio"
"bytes"
+ "cmp"
"encoding/binary"
"encoding/json"
"errors"
@@ -28,7 +29,6 @@ import (
"maps"
"math"
"slices"
- "sort"
"github.com/pierrec/lz4/v4"
)
@@ -620,29 +620,25 @@ func (r *Reader) ReadAllBlobs() ([]*BlobData, error) {
return nil, nil
}
- // Create index mapping to preserve original order
- type indexedBlob struct {
- index int
- meta BlobMetadata
+ // Keep only indexes in the temporary read-order slice.
+ // Results retain defensive metadata copies in footer order.
+ order := make([]int, len(footer.Blobs))
+ for i := range order {
+ order[i] = i
}
- indexed := make([]indexedBlob, len(footer.Blobs))
- for i, meta := range footer.Blobs {
- indexed[i] = indexedBlob{index: i, meta: meta}
- }
-
- // Sort by offset for sequential I/O
- sort.Slice(indexed, func(i, j int) bool {
- return indexed[i].meta.Offset < indexed[j].meta.Offset
+ slices.SortFunc(order, func(a, b int) int {
+ return cmp.Compare(footer.Blobs[a].Offset,
footer.Blobs[b].Offset)
})
// Read blobs in offset order, store in original order
results := make([]*BlobData, len(footer.Blobs))
- for _, ib := range indexed {
- data, err := r.readBlobData(ib.meta)
+ for _, index := range order {
+ meta := footer.Blobs[index]
+ data, err := r.readBlobData(meta)
if err != nil {
- return nil, fmt.Errorf("puffin: read blob %d: %w",
ib.index, err)
+ return nil, fmt.Errorf("puffin: read blob %d: %w",
index, err)
}
- results[ib.index] = &BlobData{Metadata:
cloneBlobMetadata(ib.meta), Data: data}
+ results[index] = &BlobData{Metadata: cloneBlobMetadata(meta),
Data: data}
}
return results, nil
diff --git a/puffin/puffin_reader_bench_test.go
b/puffin/puffin_reader_bench_test.go
new file mode 100644
index 000000000..3513a5bf6
--- /dev/null
+++ b/puffin/puffin_reader_bench_test.go
@@ -0,0 +1,112 @@
+// 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 puffin_test
+
+import (
+ "bytes"
+ "encoding/binary"
+ "encoding/json"
+ "fmt"
+ "math/rand/v2"
+ "testing"
+
+ "github.com/apache/iceberg-go/puffin"
+)
+
+func BenchmarkReadAllBlobs(b *testing.B) {
+ for _, blobCount := range []int{16, 256, 1024} {
+ b.Run(fmt.Sprintf("blobs=%d", blobCount), func(b *testing.B) {
+ var buf bytes.Buffer
+ writer, err := puffin.NewWriter(&buf)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ payload := make([]byte, 64)
+ for i := range blobCount {
+ binary.LittleEndian.PutUint64(payload,
uint64(i))
+ _, err :=
writer.AddBlob(puffin.BlobMetadataInput{
+ Type:
puffin.BlobTypeDataSketchesTheta,
+ SnapshotID: int64(i),
+ Fields: []int32{1},
+ Properties: map[string]string{"ndv":
"1"},
+ }, payload)
+ if err != nil {
+ b.Fatal(err)
+ }
+ }
+ dataEnd := buf.Len()
+ if err := writer.Finish(); err != nil {
+ b.Fatal(err)
+ }
+
+ reader, err :=
puffin.NewReader(bytes.NewReader(buf.Bytes()))
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ // Keep payloads in physical order and shuffle only the
footer metadata.
+ // This exercises the sort beyond its already-sorted
fast path.
+ footer := puffin.Footer{Blobs: reader.Blobs(),
Properties: reader.Properties()}
+ rand.New(rand.NewPCG(0, 0)).Shuffle(len(footer.Blobs),
func(i, j int) {
+ footer.Blobs[i], footer.Blobs[j] =
footer.Blobs[j], footer.Blobs[i]
+ })
+ footerPayload, err := json.Marshal(footer)
+ if err != nil {
+ b.Fatal(err)
+ }
+ shuffled := append(bytes.Clone(buf.Bytes()[:dataEnd]),
fileWithFooterPayload(footerPayload)[puffin.MagicSize:]...)
+
+ for _, test := range []struct {
+ name string
+ data []byte
+ }{
+ {name: "sorted", data: buf.Bytes()},
+ {name: "shuffled", data: shuffled},
+ } {
+ b.Run(test.name, func(b *testing.B) {
+ reader, err :=
puffin.NewReader(bytes.NewReader(test.data))
+ if err != nil {
+ b.Fatal(err)
+ }
+ metadata := reader.Blobs()
+ var blobs []*puffin.BlobData
+
+ b.ReportAllocs()
+ for b.Loop() {
+ blobs, err =
reader.ReadAllBlobs()
+ if err != nil {
+ b.Fatal(err)
+ }
+ }
+
+ if len(blobs) != blobCount {
+ b.Fatalf("got %d blobs, want
%d", len(blobs), blobCount)
+ }
+ for i, blob := range blobs {
+ if len(blob.Data) !=
len(payload) ||
+
binary.LittleEndian.Uint64(blob.Data) != uint64(metadata[i].SnapshotID) ||
+
blob.Metadata.SnapshotID != metadata[i].SnapshotID {
+ b.Fatalf("blob %d does
not match its footer entry", i)
+ }
+ }
+ })
+ }
+ })
+ }
+}
diff --git a/puffin/puffin_test.go b/puffin/puffin_test.go
index 907c8212d..b2c70bbd4 100644
--- a/puffin/puffin_test.go
+++ b/puffin/puffin_test.go
@@ -20,6 +20,7 @@ package puffin_test
import (
"bytes"
"encoding/binary"
+ "encoding/json"
"errors"
"math"
"os"
@@ -968,6 +969,36 @@ func TestReaderBlobAccess(t *testing.T) {
})
}
+func TestReadAllBlobsPreservesFooterOrder(t *testing.T) {
+ // Blob metadata may be listed in a different order from the payloads.
+ footer := puffin.Footer{Blobs: []puffin.BlobMetadata{
+ {Type: "third", Fields: []int32{3}, Offset: 15, Length: 5},
+ {Type: "first", Fields: []int32{1}, Offset: 4, Length: 5},
+ {Type: "second", Fields: []int32{2}, Offset: 9, Length: 6},
+ }}
+ payload, err := json.Marshal(footer)
+ require.NoError(t, err)
+ data := append([]byte("PFA1firstsecondthird"),
fileWithFooterPayload(payload)[puffin.MagicSize:]...)
+ input := &countingReaderAtSeeker{Reader: bytes.NewReader(data)}
+ r, err := puffin.NewReader(input)
+ require.NoError(t, err)
+ input.reads = nil
+
+ blobs, err := r.ReadAllBlobs()
+ require.NoError(t, err)
+ require.Len(t, blobs, len(footer.Blobs))
+ for i, expected := range []string{"third", "first", "second"} {
+ assert.Equal(t, expected, string(blobs[i].Data))
+ assert.Equal(t, footer.Blobs[i], blobs[i].Metadata)
+ }
+ assert.Equal(t, []readCall{
+ {offset: 4, length: 5},
+ {offset: 9, length: 6},
+ {offset: 15, length: 5},
+ }, input.reads)
+ assert.Equal(t, footer.Blobs, r.Blobs())
+}
+
// TestReadBlobByMetadataValidation verifies validation of blob metadata.
// Prevents reading garbage data when metadata is corrupted or crafted
maliciously.
func TestReadBlobByMetadataValidation(t *testing.T) {
