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 a19082e2d feat(puffin): write compressed blobs (#2401)
a19082e2d is described below
commit a19082e2d50d24d27710968d759338d931ba8442
Author: Minh Vu <[email protected]>
AuthorDate: Wed Nov 4 19:05:04 2026 +0100
feat(puffin): write compressed blobs (#2401)
* feat(puffin): read compressed blobs
Signed-off-by: Minh Vu <[email protected]>
* fix(puffin): reject trailing zstd frames
Signed-off-by: Minh Vu <[email protected]>
* feat(puffin): write compressed blobs
Signed-off-by: Minh Vu <[email protected]>
* fix(puffin): validate metadata first and verify frame sizes
* fix(puffin): reject compressed deletion vectors on read
* test(puffin): reject compressed DV footer and external metadata
* test(puffin): verify offsets of mixed-codec blobs
* fix(puffin): allow common zstd windows for small blobs
Signed-off-by: Minh Vu <[email protected]>
* fix(puffin): defer DV compression rejection to blob reads
* test(puffin): preserve DV reader error contract
* style(puffin): use complete ASF license header
* style(puffin): use complete ASF license header
* docs(puffin): clarify per-blob compression input
* test(puffin): cover LZ4 frame checksum and raw blocks
* fix(puffin): reject explicit empty compression codec
* refactor(puffin): share LZ4 and Zstd codec constants
* fix(puffin): handle unused encoder cleanup result
---------
Signed-off-by: Minh Vu <[email protected]>
---
puffin/blob_writer_compression.go | 110 +++++++++++++++
puffin/blob_writer_compression_test.go | 249 +++++++++++++++++++++++++++++++++
puffin/puffin_writer.go | 25 +++-
3 files changed, 381 insertions(+), 3 deletions(-)
diff --git a/puffin/blob_writer_compression.go
b/puffin/blob_writer_compression.go
new file mode 100644
index 000000000..4203f64a6
--- /dev/null
+++ b/puffin/blob_writer_compression.go
@@ -0,0 +1,110 @@
+// 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
+
+import (
+ "bytes"
+ "fmt"
+ "sync"
+
+ "github.com/klauspost/compress/zstd"
+ "github.com/pierrec/lz4/v4"
+)
+
+// compressBlob returns one complete frame, including the uncompressed content
+// size required by the Puffin format. The caller must not compress DVs.
+func compressBlob(codec string, data []byte) ([]byte, error) {
+ switch codec {
+ case "":
+ return data, nil
+ case CompressionCodecLZ4:
+ // The lz4 library intentionally omits the content-size field
when
+ // SizeOption(0) is used. Write the empty frame explicitly
instead.
+ if len(data) == 0 {
+ return emptyLZ4Frame(), nil
+ }
+ var buf bytes.Buffer
+ writer := lz4.NewWriter(&buf)
+ if err := writer.Apply(lz4.SizeOption(uint64(len(data))),
lz4.ConcurrencyOption(1)); err != nil {
+ return nil, fmt.Errorf("puffin: configure lz4
compression: %w", err)
+ }
+ if _, err := writer.Write(data); err != nil {
+ return nil, fmt.Errorf("puffin: compress lz4 blob: %w",
err)
+ }
+ if err := writer.Close(); err != nil {
+ return nil, fmt.Errorf("puffin: finalize lz4 blob: %w",
err)
+ }
+
+ return buf.Bytes(), nil
+ case CompressionCodecZstd:
+ // Short Zstd frames need the single-segment flag to include a
content size.
+ encoder, err := sharedZstdBlobEncoder(len(data) < 256)
+ if err != nil {
+ return nil, err
+ }
+
+ return encoder.EncodeAll(data, nil), nil
+ default:
+ return nil, fmt.Errorf("puffin: unsupported blob compression
codec %q", codec)
+ }
+}
+
+// emptyLZ4Frame is an independent-block frame with an explicit zero content
+// size. LZ4 SizeOption(0) omits that field. The header checksum (0x12) is
+// verified by lz4.ValidFrameHeader in the writer tests.
+func emptyLZ4Frame() []byte {
+ return []byte{
+ 0x04, 0x22, 0x4d, 0x18, // frame magic
+ 0x68, 0x70, // frame flags and 4 MiB block size
+ 0, 0, 0, 0, 0, 0, 0, 0, // content size: zero
+ 0x12, // frame descriptor checksum
+ 0, 0, 0, 0, // end mark
+ }
+}
+
+var (
+ zstdBlobEncoderOnce sync.Once
+ zstdBlobEncoder *zstd.Encoder
+ zstdBlobSingleSegmentEncoder *zstd.Encoder
+ zstdBlobEncoderErr error
+)
+
+// sharedZstdBlobEncoder lazily creates two reusable encoders. EncodeAll can be
+// called concurrently and does not require resetting or closing the encoder.
+func sharedZstdBlobEncoder(singleSegment bool) (*zstd.Encoder, error) {
+ zstdBlobEncoderOnce.Do(func() {
+ options := []zstd.EOption{zstd.WithEncoderConcurrency(1),
zstd.WithZeroFrames(true)}
+ zstdBlobEncoder, zstdBlobEncoderErr = zstd.NewWriter(nil,
options...)
+ if zstdBlobEncoderErr != nil {
+ return
+ }
+ zstdBlobSingleSegmentEncoder, zstdBlobEncoderErr =
zstd.NewWriter(
+ nil, append(options, zstd.WithSingleSegment(true))...,
+ )
+ if zstdBlobEncoderErr != nil {
+ _ = zstdBlobEncoder.Close()
+ }
+ })
+ if zstdBlobEncoderErr != nil {
+ return nil, fmt.Errorf("puffin: configure zstd compression:
%w", zstdBlobEncoderErr)
+ }
+ if singleSegment {
+ return zstdBlobSingleSegmentEncoder, nil
+ }
+
+ return zstdBlobEncoder, nil
+}
diff --git a/puffin/blob_writer_compression_test.go
b/puffin/blob_writer_compression_test.go
new file mode 100644
index 000000000..02c49c216
--- /dev/null
+++ b/puffin/blob_writer_compression_test.go
@@ -0,0 +1,249 @@
+// 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"
+ "strconv"
+ "testing"
+
+ "github.com/apache/iceberg-go/puffin"
+ "github.com/klauspost/compress/zstd"
+ "github.com/pierrec/lz4/v4"
+ "github.com/stretchr/testify/require"
+)
+
+func TestWriteCompressedBlobs(t *testing.T) {
+ for _, codec := range []string{"", puffin.CompressionCodecLZ4,
puffin.CompressionCodecZstd} {
+ for _, input := range [][]byte{nil, []byte("hello, Iceberg!"),
bytes.Repeat([]byte("Iceberg"), 2000)} {
+ t.Run(codec+"/"+strconv.Itoa(len(input)), func(t
*testing.T) {
+ t.Parallel()
+ var file bytes.Buffer
+ writer, err := puffin.NewWriter(&file)
+ require.NoError(t, err)
+ inputMeta := puffin.BlobMetadataInput{
+ Type:
puffin.BlobTypeDataSketchesTheta,
+ Fields: []int32{12},
+ SnapshotID: 5,
+ SequenceNumber: 3,
+ CompressionCodec: codec,
+ }
+ meta, err := writer.AddBlob(inputMeta, input)
+ require.NoError(t, err)
+ require.NoError(t, writer.Finish())
+ require.Equal(t, int64(puffin.MagicSize),
meta.Offset)
+ if codec != "" {
+ require.Greater(t, meta.Length,
int64(0), "even an empty compressed frame has bytes")
+ }
+ if codec == "" {
+ require.Nil(t, meta.CompressionCodec)
+ } else {
+ require.Equal(t, codec,
*meta.CompressionCodec)
+ }
+ reader, err :=
puffin.NewReader(bytes.NewReader(file.Bytes()))
+ require.NoError(t, err)
+ blobs := reader.Blobs()
+ require.Len(t, blobs, 1)
+ require.Equal(t, meta, blobs[0])
+ encoded := file.Bytes()[meta.Offset :
meta.Offset+meta.Length]
+ switch codec {
+ case "":
+ require.Equal(t, string(input),
string(encoded))
+ case puffin.CompressionCodecLZ4:
+ // Puffin requires the LZ4 content-size
flag, including for empty blobs.
+ require.GreaterOrEqual(t, len(encoded),
19)
+ require.Equal(t, uint32(0x184D2204),
binary.LittleEndian.Uint32(encoded[:4]))
+ require.NotZero(t, encoded[4]&0x08)
+ require.Equal(t, uint64(len(input)),
binary.LittleEndian.Uint64(encoded[6:14]))
+ validHeader, headerErr :=
lz4.ValidFrameHeader(encoded[:15])
+ require.NoError(t, headerErr)
+ require.True(t, validHeader, "LZ4 frame
header checksum must be valid")
+ case puffin.CompressionCodecZstd:
+ var header zstd.Header
+ require.NoError(t,
header.Decode(encoded))
+ require.True(t, header.HasFCS)
+ require.Equal(t, uint64(len(input)),
header.FrameContentSize)
+ }
+
+ blob, err := reader.ReadBlob(0)
+ require.NoError(t, err)
+ require.Equal(t, string(input),
string(blob.Data))
+ require.Equal(t, meta, blob.Metadata)
+ byMetadata, err :=
reader.ReadBlobByMetadata(meta)
+ require.NoError(t, err)
+ require.Equal(t, string(input),
string(byMetadata))
+ all, err := reader.ReadAllBlobs()
+ require.NoError(t, err)
+ require.Len(t, all, 1)
+ require.Equal(t, string(input),
string(all[0].Data))
+ })
+ }
+ }
+}
+
+func TestZstdBlobFrameContentSizeBoundaries(t *testing.T) {
+ for _, size := range []int{0, 1, 15, 255, 256, 65535, 65536} {
+ t.Run(strconv.Itoa(size), func(t *testing.T) {
+ var output bytes.Buffer
+ writer, err := puffin.NewWriter(&output)
+ require.NoError(t, err)
+ meta, err := writer.AddBlob(puffin.BlobMetadataInput{
+ Type: puffin.BlobTypeDataSketchesTheta, Fields:
[]int32{1}, CompressionCodec: puffin.CompressionCodecZstd,
+ }, bytes.Repeat([]byte{42}, size))
+ require.NoError(t, err)
+ var header zstd.Header
+ require.NoError(t,
header.Decode(output.Bytes()[meta.Offset:meta.Offset+meta.Length]))
+ require.True(t, header.HasFCS)
+ require.Equal(t, uint64(size), header.FrameContentSize)
+ require.NoError(t, writer.Finish())
+ reader, err :=
puffin.NewReader(bytes.NewReader(output.Bytes()))
+ require.NoError(t, err)
+ blob, err := reader.ReadBlob(0)
+ require.NoError(t, err)
+ require.Equal(t, string(bytes.Repeat([]byte{42},
size)), string(blob.Data))
+ })
+ }
+}
+
+func TestCompressedWriterRejectsInvalidCodecWithoutWriting(t *testing.T) {
+ var buf bytes.Buffer
+ w, err := puffin.NewWriter(&buf)
+ require.NoError(t, err)
+ n := buf.Len()
+ _, err = w.AddBlob(puffin.BlobMetadataInput{
+ Type: puffin.BlobTypeDataSketchesTheta,
+ Fields: []int32{3},
+ CompressionCodec: "gzip",
+ }, []byte("payload"))
+ require.ErrorContains(t, err, "unsupported blob compression codec")
+ require.Equal(t, n, buf.Len())
+ _, err = w.AddBlob(puffin.BlobMetadataInput{
+ Type: puffin.BlobTypeDeletionVector,
+ Fields: []int32{},
+ SnapshotID: -1,
+ SequenceNumber: -1,
+ CompressionCodec: puffin.CompressionCodecLZ4,
+ }, []byte("payload"))
+ require.ErrorContains(t, err, "must not be compressed")
+ require.Equal(t, n, buf.Len())
+ _, err = w.AddBlob(puffin.BlobMetadataInput{Type:
puffin.BlobTypeDataSketchesTheta, Fields: []int32{}}, []byte("ok"))
+ require.NoError(t, err)
+ require.NoError(t, w.Finish())
+}
+
+func TestLZ4BlobSpanningMultipleBlocks(t *testing.T) {
+ data := bytes.Repeat([]byte("abcd1234"), (4<<20)/8+1)
+ var output bytes.Buffer
+ writer, err := puffin.NewWriter(&output)
+ require.NoError(t, err)
+ meta, err := writer.AddBlob(puffin.BlobMetadataInput{
+ Type: puffin.BlobTypeDataSketchesTheta, Fields: []int32{1},
CompressionCodec: puffin.CompressionCodecLZ4,
+ }, data)
+ require.NoError(t, err)
+ require.NoError(t, writer.Finish())
+ encoded := output.Bytes()[meta.Offset : meta.Offset+meta.Length]
+ require.NotZero(t, encoded[4]&0x08)
+ require.Equal(t, uint64(len(data)),
binary.LittleEndian.Uint64(encoded[6:14]))
+ reader, err := puffin.NewReader(bytes.NewReader(output.Bytes()))
+ require.NoError(t, err)
+ blob, err := reader.ReadBlob(0)
+ require.NoError(t, err)
+ require.Equal(t, data, blob.Data)
+}
+
+// Incompressible input exercises LZ4's uncompressed-block fallback, rather
+// than only the compressed blocks covered by the other writer tests.
+func TestLZ4BlobWithUncompressedBlock(t *testing.T) {
+ data := make([]byte, 64<<10)
+ state := uint32(0x12345678)
+ for i := range data {
+ state ^= state << 13
+ state ^= state >> 17
+ state ^= state << 5
+ data[i] = byte(state)
+ }
+
+ var file bytes.Buffer
+ writer, err := puffin.NewWriter(&file)
+ require.NoError(t, err)
+ meta, err := writer.AddBlob(puffin.BlobMetadataInput{
+ Type: puffin.BlobTypeDataSketchesTheta, Fields: []int32{1},
CompressionCodec: puffin.CompressionCodecLZ4,
+ }, data)
+ require.NoError(t, err)
+ require.NoError(t, writer.Finish())
+
+ frame := file.Bytes()[meta.Offset : meta.Offset+meta.Length]
+ require.GreaterOrEqual(t, len(frame), 19)
+ require.NotZero(t, binary.LittleEndian.Uint32(frame[15:19])&(1<<31),
+ "LZ4 should store incompressible blocks uncompressed")
+ reader, err := puffin.NewReader(bytes.NewReader(file.Bytes()))
+ require.NoError(t, err)
+ blob, err := reader.ReadBlob(0)
+ require.NoError(t, err)
+ require.Equal(t, data, blob.Data)
+}
+
+func TestMixedBlobCompressionUsesStoredOffsets(t *testing.T) {
+ var file bytes.Buffer
+ writer, err := puffin.NewWriter(&file)
+ require.NoError(t, err)
+
+ blobs := []struct {
+ codec string
+ data []byte
+ }{
+ {codec: "", data: []byte("raw payload")},
+ {codec: puffin.CompressionCodecLZ4, data:
bytes.Repeat([]byte("lz4 payload"), 512)},
+ {codec: puffin.CompressionCodecZstd, data:
bytes.Repeat([]byte("zstd payload"), 512)},
+ }
+ previousEnd := int64(puffin.MagicSize)
+ for _, blob := range blobs {
+ meta, err := writer.AddBlob(puffin.BlobMetadataInput{
+ Type: puffin.BlobTypeDataSketchesTheta,
+ Fields: []int32{1},
+ CompressionCodec: blob.codec,
+ }, blob.data)
+ require.NoError(t, err)
+ require.Equal(t, previousEnd, meta.Offset)
+ previousEnd = meta.Offset + meta.Length
+ }
+ require.NoError(t, writer.Finish())
+
+ reader, err := puffin.NewReader(bytes.NewReader(file.Bytes()))
+ require.NoError(t, err)
+ metadata := reader.Blobs()
+ require.Len(t, metadata, len(blobs))
+ for i, blob := range blobs {
+ meta := metadata[i]
+ if blob.codec == "" {
+ require.Nil(t, meta.CompressionCodec)
+ } else {
+ require.Equal(t, blob.codec, *meta.CompressionCodec)
+ }
+ got, err := reader.ReadBlob(i)
+ require.NoError(t, err)
+ require.Equal(t, blob.data, got.Data)
+ require.Equal(t, meta, got.Metadata)
+ byMetadata, err := reader.ReadBlobByMetadata(meta)
+ require.NoError(t, err)
+ require.Equal(t, blob.data, byMetadata)
+ }
+ all, err := reader.ReadAllBlobs()
+ require.NoError(t, err)
+ require.Len(t, all, len(blobs))
+}
diff --git a/puffin/puffin_writer.go b/puffin/puffin_writer.go
index 113206c11..4da2849da 100644
--- a/puffin/puffin_writer.go
+++ b/puffin/puffin_writer.go
@@ -61,13 +61,17 @@ type Writer struct {
}
// BlobMetadataInput contains fields the caller provides when adding a blob.
-// Offset, Length, and CompressionCodec are set by the writer.
+// Offset and Length are computed by the writer. CompressionCodec selects the
+// blob's compression and is omitted from footer metadata when empty.
type BlobMetadataInput struct {
Type BlobType
SnapshotID int64
SequenceNumber int64
Fields []int32
Properties map[string]string
+ // CompressionCodec selects CompressionCodecLZ4 or CompressionCodecZstd.
+ // An empty value writes raw bytes. Deletion vectors must not be
compressed.
+ CompressionCodec string
}
// NewWriter creates a new Writer and writes the file header magic.
@@ -155,6 +159,10 @@ func (w *Writer) AddBlob(input BlobMetadataInput, data
[]byte) (BlobMetadata, er
fields := slices.Clone(input.Fields)
properties := maps.Clone(input.Properties)
+ if input.Type == BlobTypeDeletionVector && input.CompressionCodec != ""
{
+ return BlobMetadata{}, errors.New("puffin: deletion-vector-v1
must not be compressed")
+ }
+
// Deletion vectors have special requirements per spec. Format matches
// the reader-side validateBlobs check so both sides surface the same
// "<type> requires X to be -1, got N" phrasing for log grep.
@@ -190,17 +198,28 @@ func (w *Writer) AddBlob(input BlobMetadataInput, data
[]byte) (BlobMetadata, er
}
}
+ // Validate metadata before compressing potentially large blobs. Encode
before
+ // writing any bytes so compression failures cannot leave partial
output.
+ encoded, err := compressBlob(input.CompressionCodec, data)
+ if err != nil {
+ return BlobMetadata{}, err
+ }
+
meta := BlobMetadata{
Type: input.Type,
SnapshotID: input.SnapshotID,
SequenceNumber: input.SequenceNumber,
Fields: fields,
Offset: w.offset,
- Length: int64(len(data)),
+ Length: int64(len(encoded)),
Properties: properties,
}
+ if input.CompressionCodec != "" {
+ codec := input.CompressionCodec
+ meta.CompressionCodec = &codec
+ }
- if err := w.write(data); err != nil {
+ if err := w.write(encoded); err != nil {
return BlobMetadata{}, fmt.Errorf("puffin: write blob: %w", err)
}
