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 24cd5be0f feat(puffin): read compressed blobs (#2400)
24cd5be0f is described below
commit 24cd5be0f33cdb0defd2c1a5cd77c0449f9cc9ec
Author: Minh Vu <[email protected]>
AuthorDate: Wed Nov 4 19:01:58 2026 +0100
feat(puffin): read compressed blobs (#2400)
* 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]>
* fix(puffin): reject compressed deletion vectors on read
* test(puffin): reject compressed DV footer and external metadata
* 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
* fix(puffin): reject explicit empty compression codec
* refactor(puffin): share LZ4 and Zstd codec constants
---------
Signed-off-by: Minh Vu <[email protected]>
---
puffin/blob_compression.go | 167 +++++++++++++++++++++++
puffin/blob_compression_test.go | 290 ++++++++++++++++++++++++++++++++++++++++
puffin/puffin_reader.go | 22 ++-
puffin/puffin_test.go | 25 ++--
4 files changed, 492 insertions(+), 12 deletions(-)
diff --git a/puffin/blob_compression.go b/puffin/blob_compression.go
new file mode 100644
index 000000000..0bc1b9328
--- /dev/null
+++ b/puffin/blob_compression.go
@@ -0,0 +1,167 @@
+// 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"
+ "errors"
+ "fmt"
+ "io"
+ "math"
+
+ "github.com/klauspost/compress/zstd"
+ "github.com/pierrec/lz4/v4"
+)
+
+// CompressionCodecLZ4 is the Puffin LZ4 frame codec.
+const CompressionCodecLZ4 = "lz4"
+
+// CompressionCodecZstd is the Puffin Zstandard frame codec.
+const CompressionCodecZstd = "zstd"
+
+// readCompressedBlob decompresses one Puffin blob frame. Puffin requires
+// compressed blobs to include their uncompressed content size in the frame.
+// Bound both the advertised size and the actual output to maxBlobSize.
+func readCompressedBlob(data []byte, codec string, maxBlobSize int64) ([]byte,
error) {
+ if maxBlobSize <= 0 {
+ return nil, fmt.Errorf("puffin: invalid maximum blob size %d",
maxBlobSize)
+ }
+
+ var (
+ decoded io.Reader
+ expectedSize uint64
+ closeDecoder func()
+ )
+
+ switch codec {
+ case CompressionCodecLZ4:
+ input := bytes.NewReader(data)
+ header, err := readLZ4FrameHeader(input)
+ if err != nil {
+ return nil, fmt.Errorf("puffin: invalid LZ4 blob frame:
%w", err)
+ }
+ if !header.hasContentSize {
+ return nil, errors.New("puffin: LZ4 blob frame is
missing content size")
+ }
+ expectedSize = header.contentSize
+ if expectedSize > uint64(maxBlobSize) {
+ return nil, fmt.Errorf("puffin: decompressed blob
exceeds maximum size %d", maxBlobSize)
+ }
+ if err := validateLZ4FrameEnvelope(
+ input, int64(len(data)), header.flags,
header.headerSize, header.blockMaxSize,
+ ); err != nil {
+ return nil, fmt.Errorf("puffin: invalid LZ4 blob frame:
%w", err)
+ }
+ decoded = lz4.NewReader(bytes.NewReader(data))
+
+ case CompressionCodecZstd:
+ var header zstd.Header
+ if err := header.Decode(data); err != nil {
+ return nil, fmt.Errorf("puffin: invalid zstd blob
frame: %w", err)
+ }
+ if header.Skippable || !header.HasFCS {
+ return nil, errors.New("puffin: zstd blob frame is
missing content size")
+ }
+ expectedSize = header.FrameContentSize
+ if expectedSize > uint64(maxBlobSize) {
+ return nil, fmt.Errorf("puffin: decompressed blob
exceeds maximum size %d", maxBlobSize)
+ }
+ if err := validateZstdFrameEnvelope(data, header); err != nil {
+ return nil, fmt.Errorf("puffin: invalid zstd blob
frame: %w", err)
+ }
+ // The decoder may need a window larger than a tiny caller-set
blob
+ // limit, but its output is still strictly bounded below. Allow
at least
+ // an 8 MiB window for small blobs written with standard
encoders.
+ decoder, err := zstd.NewReader(
+ bytes.NewReader(data),
+ zstd.WithDecoderConcurrency(1),
+ zstd.WithDecoderMaxMemory(max(uint64(maxBlobSize),
uint64(8<<20))),
+ )
+ if err != nil {
+ return nil, fmt.Errorf("puffin: create zstd blob
decoder: %w", err)
+ }
+ decoded = decoder
+ closeDecoder = decoder.Close
+
+ default:
+ return nil, fmt.Errorf("puffin: unsupported blob compression
codec %q", codec)
+ }
+ if closeDecoder != nil {
+ defer closeDecoder()
+ }
+
+ // Reading one byte past the limit catches corrupt content-size headers
and
+ // decompression bombs without allocating unbounded output buffers.
+ limit := maxBlobSize
+ if limit < math.MaxInt64 {
+ limit++
+ }
+ output, err := io.ReadAll(io.LimitReader(decoded, limit))
+ if err != nil {
+ return nil, fmt.Errorf("puffin: decompress %s blob: %w", codec,
err)
+ }
+ if int64(len(output)) > maxBlobSize {
+ return nil, fmt.Errorf("puffin: decompressed blob exceeds
maximum size %d", maxBlobSize)
+ }
+ if uint64(len(output)) != expectedSize {
+ return nil, fmt.Errorf("puffin: decompressed %s blob size %d
does not match frame content size %d", codec, len(output), expectedSize)
+ }
+
+ return output, nil
+}
+
+// validateZstdFrameEnvelope ensures the blob contains exactly one frame.
+// The decoder accepts concatenated or skippable frames, which Puffin does not.
+func validateZstdFrameEnvelope(data []byte, header zstd.Header) error {
+ offset := header.HeaderSize
+ for {
+ if len(data)-offset < 3 {
+ return errors.New("truncated zstd block header")
+ }
+ block := uint32(data[offset]) | uint32(data[offset+1])<<8 |
uint32(data[offset+2])<<16
+ offset += 3
+ last := block&1 != 0
+ kind := (block >> 1) & 3
+ blockSize := int(block >> 3)
+ switch kind {
+ case 0, 2:
+ case 1:
+ blockSize = 1
+ default:
+ return errors.New("reserved zstd block type")
+ }
+ if blockSize > len(data)-offset {
+ return errors.New("truncated zstd block")
+ }
+ offset += blockSize
+ if last {
+ break
+ }
+ }
+ if header.HasCheckSum {
+ if len(data)-offset < 4 {
+ return errors.New("truncated zstd content checksum")
+ }
+ offset += 4
+ }
+ if offset != len(data) {
+ return errors.New("zstd frame has trailing data")
+ }
+
+ return nil
+}
diff --git a/puffin/blob_compression_test.go b/puffin/blob_compression_test.go
new file mode 100644
index 000000000..0fc1d7620
--- /dev/null
+++ b/puffin/blob_compression_test.go
@@ -0,0 +1,290 @@
+// 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"
+ "testing"
+
+ "github.com/apache/iceberg-go/puffin"
+ "github.com/klauspost/compress/zstd"
+ "github.com/pierrec/lz4/v4"
+ "github.com/stretchr/testify/require"
+)
+
+func compressedBlobTestFile(t *testing.T, codec string, payload []byte) []byte
{
+ t.Helper()
+
+ return compressedBlobTestFileWithMetadata(t, payload,
puffin.BlobMetadata{
+ Type: puffin.BlobTypeDataSketchesTheta,
+ Fields: []int32{1},
+ SnapshotID: 42,
+ SequenceNumber: 7,
+ Offset: puffin.MagicSize,
+ Length: int64(len(payload)),
+ CompressionCodec: &codec,
+ })
+}
+
+func compressedBlobTestFileWithMetadata(t *testing.T, payload []byte, metadata
puffin.BlobMetadata) []byte {
+ t.Helper()
+ footer, err := json.Marshal(puffin.Footer{Blobs:
[]puffin.BlobMetadata{metadata}})
+ require.NoError(t, err)
+
+ var data bytes.Buffer
+ data.WriteString("PFA1")
+ data.Write(payload)
+ data.WriteString("PFA1")
+ data.Write(footer)
+ var trailer [12]byte
+ binary.LittleEndian.PutUint32(trailer[:4], uint32(len(footer)))
+ copy(trailer[8:], "PFA1")
+ data.Write(trailer[:])
+
+ return data.Bytes()
+}
+
+func compressedBlobTestPayload(t *testing.T, codec string, data []byte) []byte
{
+ t.Helper()
+
+ var output bytes.Buffer
+ switch codec {
+ case puffin.CompressionCodecLZ4:
+ writer := lz4.NewWriter(&output)
+ require.NoError(t,
writer.Apply(lz4.SizeOption(uint64(len(data)))))
+ _, err := writer.Write(data)
+ require.NoError(t, err)
+ require.NoError(t, writer.Close())
+
+ case puffin.CompressionCodecZstd:
+ writer, err := zstd.NewWriter(nil)
+ require.NoError(t, err)
+ defer writer.Close()
+
+ return writer.EncodeAll(data, nil)
+ }
+
+ return output.Bytes()
+}
+
+func TestReadCompressedPuffinBlobs(t *testing.T) {
+ raw := bytes.Repeat([]byte("iceberg blob content "), 256)
+ for _, codec := range []string{puffin.CompressionCodecLZ4,
puffin.CompressionCodecZstd} {
+ t.Run(codec, func(t *testing.T) {
+ stored := compressedBlobTestPayload(t, codec, raw)
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t, codec, stored)))
+ require.NoError(t, err)
+
+ blob, err := r.ReadBlob(0)
+ require.NoError(t, err)
+ require.Equal(t, raw, blob.Data)
+ require.Equal(t, int64(len(stored)),
blob.Metadata.Length)
+ require.Equal(t, codec, *blob.Metadata.CompressionCodec)
+
+ byMetadata, err := r.ReadBlobByMetadata(r.Blobs()[0])
+ require.NoError(t, err)
+ require.Equal(t, raw, byMetadata)
+
+ all, err := r.ReadAllBlobs()
+ require.NoError(t, err)
+ require.Len(t, all, 1)
+ require.Equal(t, raw, all[0].Data)
+ })
+ }
+}
+
+func TestReadCompressedPuffinBlobHonorsUncompressedLimit(t *testing.T) {
+ raw := bytes.Repeat([]byte("a"), 4096)
+ for _, codec := range []string{puffin.CompressionCodecLZ4,
puffin.CompressionCodecZstd} {
+ t.Run(codec, func(t *testing.T) {
+ stored := compressedBlobTestPayload(t, codec, raw)
+ require.Less(t, len(stored), 128)
+ r, err := puffin.NewReader(
+ bytes.NewReader(compressedBlobTestFile(t,
codec, stored)),
+ puffin.WithMaxBlobSize(128),
+ )
+ require.NoError(t, err)
+
+ _, err = r.ReadBlob(0)
+ require.ErrorContains(t, err, "decompressed blob
exceeds maximum size")
+ })
+ }
+}
+
+func TestReadCompressedPuffinBlobRejectsInvalidFrames(t *testing.T) {
+ raw := []byte("Puffin blob frame")
+ t.Run("unknown codec", func(t *testing.T) {
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t, "gzip", raw)))
+ require.NoError(t, err)
+ _, err = r.ReadBlob(0)
+ require.ErrorContains(t, err, "not supported")
+ })
+
+ t.Run("explicit empty codec", func(t *testing.T) {
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t, "", raw)))
+ require.NoError(t, err)
+
+ _, err = r.ReadBlob(0)
+ require.ErrorContains(t, err, `compression "" not supported`)
+ _, err = r.ReadBlobByMetadata(r.Blobs()[0])
+ require.ErrorContains(t, err, `compression "" not supported`)
+ _, err = r.ReadAllBlobs()
+ require.ErrorContains(t, err, `compression "" not supported`)
+ })
+
+ t.Run("lz4 missing content size", func(t *testing.T) {
+ var output bytes.Buffer
+ writer := lz4.NewWriter(&output)
+ _, err := writer.Write(raw)
+ require.NoError(t, err)
+ require.NoError(t, writer.Close())
+
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t,
puffin.CompressionCodecLZ4, output.Bytes())))
+ require.NoError(t, err)
+ _, err = r.ReadBlob(0)
+ require.ErrorContains(t, err, "missing content size")
+ })
+
+ t.Run("lz4 trailing frame", func(t *testing.T) {
+ frame := compressedBlobTestPayload(t,
puffin.CompressionCodecLZ4, raw)
+ frame = append(frame, 0xff)
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t,
puffin.CompressionCodecLZ4, frame)))
+ require.NoError(t, err)
+ _, err = r.ReadBlob(0)
+ require.ErrorContains(t, err, "trailing data")
+ })
+
+ t.Run("zstd missing content size", func(t *testing.T) {
+ var output bytes.Buffer
+ writer, err := zstd.NewWriter(&output)
+ require.NoError(t, err)
+ _, err = writer.Write(raw)
+ require.NoError(t, err)
+ require.NoError(t, writer.Close())
+
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t,
puffin.CompressionCodecZstd, output.Bytes())))
+ require.NoError(t, err)
+ _, err = r.ReadBlob(0)
+ require.ErrorContains(t, err, "missing content size")
+ })
+}
+
+func TestReadCompressedPuffinBlobRejectsZstdTrailingFrames(t *testing.T) {
+ data := bytes.Repeat([]byte("single frame"), 256)
+ frame := compressedBlobTestPayload(t, puffin.CompressionCodecZstd, data)
+ for _, tc := range []struct {
+ name string
+ trailing []byte
+ }{
+ {"empty frame", []byte{0x28, 0xb5, 0x2f, 0xfd, 0x20, 0x00,
0x01, 0x00, 0x00}},
+ {"skippable frame", []byte{0x50, 0x2a, 0x4d, 0x18, 0, 0, 0, 0}},
+ {"nonempty frame", compressedBlobTestPayload(t,
puffin.CompressionCodecZstd, []byte("other"))},
+ {"trailing byte", []byte{0}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ payload := append(bytes.Clone(frame), tc.trailing...)
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t,
puffin.CompressionCodecZstd, payload)))
+ require.NoError(t, err)
+ _, err = r.ReadBlob(0)
+ require.ErrorContains(t, err, "trailing data")
+ })
+ }
+}
+
+func TestReadCompressedPuffinBlobRejectsTruncatedZstd(t *testing.T) {
+ frame := compressedBlobTestPayload(t, puffin.CompressionCodecZstd,
bytes.Repeat([]byte("hello"), 100))
+ for n := 1; n < len(frame); n++ {
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t,
puffin.CompressionCodecZstd, frame[:n])))
+ require.NoError(t, err)
+ _, err = r.ReadBlob(0)
+ require.Error(t, err, "truncated at %d of %d", n, len(frame))
+ }
+}
+
+func TestReadCompressedPuffinBlobZstdMultipleBlocksAndEmpty(t *testing.T) {
+ large := make([]byte, 300000)
+ for i := range large {
+ large[i] = byte((i * 31) ^ (i >> 7))
+ }
+ for _, tc := range []struct {
+ name string
+ frame, want []byte
+ }{
+ {"multiple blocks", compressedBlobTestPayload(t,
puffin.CompressionCodecZstd, large), large},
+ {"empty", []byte{0x28, 0xb5, 0x2f, 0xfd, 0x20, 0x00, 0x01,
0x00, 0x00}, []byte{}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ r, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t,
puffin.CompressionCodecZstd, tc.frame)))
+ require.NoError(t, err)
+ blob, err := r.ReadBlob(0)
+ require.NoError(t, err)
+ require.Equal(t, tc.want, blob.Data)
+ })
+ }
+}
+
+func TestReadRejectsCompressedDeletionVectors(t *testing.T) {
+ for _, codec := range []string{puffin.CompressionCodecLZ4,
puffin.CompressionCodecZstd} {
+ t.Run(codec, func(t *testing.T) {
+ stored := compressedBlobTestPayload(t, codec,
[]byte("invalid compressed deletion vector"))
+ meta := puffin.BlobMetadata{
+ Type: puffin.BlobTypeDeletionVector,
+ Fields: []int32{},
+ SnapshotID: -1,
+ SequenceNumber: -1,
+ Offset: puffin.MagicSize,
+ Length: int64(len(stored)),
+ CompressionCodec: &codec,
+ }
+ reader, err :=
puffin.NewReader(bytes.NewReader(compressedBlobTestFileWithMetadata(t, stored,
meta)))
+ require.NoError(t, err)
+ _, err = reader.ReadBlob(0)
+ require.ErrorContains(t, err, "deletion-vector-v1 must
not be compressed")
+
+ reader, err =
puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t, codec, stored)))
+ require.NoError(t, err)
+ _, err = reader.ReadBlobByMetadata(meta)
+ require.ErrorContains(t, err, "deletion-vector-v1 must
not be compressed")
+ })
+ }
+}
+
+func TestReadSmallZstdBlobWithLargeWindow(t *testing.T) {
+ raw := bytes.Repeat([]byte("abc"), 100)
+ encoder, err := zstd.NewWriter(nil, zstd.WithWindowSize(8<<20),
zstd.WithSingleSegment(false))
+ require.NoError(t, err)
+ frame := encoder.EncodeAll(raw, nil)
+ encoder.Close()
+ var initial zstd.Header
+ require.NoError(t, initial.Decode(frame))
+ require.False(t, initial.SingleSegment)
+ // Set a valid 8 MiB window descriptor while retaining the original
blocks.
+ frame[5] = 0x68
+ var header zstd.Header
+ require.NoError(t, header.Decode(frame))
+ require.True(t, header.HasFCS)
+ require.Equal(t, uint64(8<<20), header.WindowSize)
+ require.Equal(t, uint64(len(raw)), header.FrameContentSize)
+ r, err := puffin.NewReader(bytes.NewReader(compressedBlobTestFile(t,
puffin.CompressionCodecZstd, frame)), puffin.WithMaxBlobSize(1024))
+ require.NoError(t, err)
+ blob, err := r.ReadBlob(0)
+ require.NoError(t, err)
+ require.Equal(t, raw, blob.Data)
+}
diff --git a/puffin/puffin_reader.go b/puffin/puffin_reader.go
index 0a5b0987f..e7db8731a 100644
--- a/puffin/puffin_reader.go
+++ b/puffin/puffin_reader.go
@@ -77,7 +77,8 @@ type ReaderOption func(*Reader)
// WithMaxBlobSize sets the maximum blob size allowed when reading.
// This prevents OOM attacks from malicious files with huge blob lengths.
-// Default is DefaultMaxBlobSize (256 MB).
+// Default is DefaultMaxBlobSize (256 MB). Zstd frames also have a window
+// limit equal to the larger of this size and 8 MiB.
func WithMaxBlobSize(size int64) ReaderOption {
return func(r *Reader) {
r.maxBlobSize = size
@@ -365,10 +366,18 @@ func (r *Reader) readBlobData(meta BlobMetadata) ([]byte,
error) {
if meta.Type == "" {
return nil, errors.New("puffin: cannot read blob: type is
required")
}
+ if meta.Type == BlobTypeDeletionVector && meta.CompressionCodec != nil {
+ return nil, errors.New("puffin: deletion-vector-v1 must not be
compressed")
+ }
- // Check for compressed blob (unsupported)
- if meta.CompressionCodec != nil && *meta.CompressionCodec != "" {
- return nil, fmt.Errorf("puffin: cannot read blob: compression
%q not supported", *meta.CompressionCodec)
+ // Reject unknown codecs before issuing range reads. An omitted codec
+ // means uncompressed; an explicitly empty codec is invalid.
+ codec := ""
+ if meta.CompressionCodec != nil {
+ codec = *meta.CompressionCodec
+ if codec != CompressionCodecLZ4 && codec !=
CompressionCodecZstd {
+ return nil, fmt.Errorf("puffin: cannot read blob:
compression %q not supported", codec)
+ }
}
// Validate offset/length
@@ -381,6 +390,9 @@ func (r *Reader) readBlobData(meta BlobMetadata) ([]byte,
error) {
if _, err := r.r.ReadAt(data, meta.Offset); err != nil {
return nil, fmt.Errorf("puffin: read blob data: %w", err)
}
+ if codec != "" {
+ return readCompressedBlob(data, codec, r.maxBlobSize)
+ }
return data, nil
}
@@ -502,6 +514,8 @@ func (r *Reader) validateBlobs(blobs []BlobMetadata,
footerStart int64) error {
// "<type> requires X to be -1, got N" so both sides share a
// grep-able canonical message.
if blob.Type == BlobTypeDeletionVector {
+ // Leave compression validation to ReadBlob and the
deletion-vector
+ // reader, which reports ErrInvalidDeletionVector to
its callers.
if blob.SnapshotID != -1 {
return fmt.Errorf("puffin: blob %d: %s requires
snapshot-id to be -1, got %d",
i, BlobTypeDeletionVector,
blob.SnapshotID)
diff --git a/puffin/puffin_test.go b/puffin/puffin_test.go
index aad597296..510f586d9 100644
--- a/puffin/puffin_test.go
+++ b/puffin/puffin_test.go
@@ -1247,20 +1247,29 @@ func TestReadJavaUncompressedMetricData(t *testing.T) {
assert.Len(t, blob1.Data, 83)
}
-// TestReadJavaCompressedMetricData verifies our reader correctly rejects
-// the zstd-compressed fixture since compression is not yet supported.
+// TestReadJavaCompressedMetricData verifies that Java-produced zstd blobs
+// decode to the same bytes as the canonical uncompressed Puffin fixture.
func TestReadJavaCompressedMetricData(t *testing.T) {
- data := readFixture(t, "sample-metric-data-compressed-zstd.bin")
- r, err := puffin.NewReader(bytes.NewReader(data))
+ compressed := readFixture(t, "sample-metric-data-compressed-zstd.bin")
+ r, err := puffin.NewReader(bytes.NewReader(compressed))
+ require.NoError(t, err)
+
+ plain := readFixture(t, "sample-metric-data-uncompressed.bin")
+ uncompressed, err := puffin.NewReader(bytes.NewReader(plain))
require.NoError(t, err)
- // Metadata should be readable (compression is per-blob, not footer)
blobs := r.Blobs()
require.Len(t, blobs, 2)
+ for i, meta := range blobs {
+ require.NotNil(t, meta.CompressionCodec)
+ assert.Equal(t, "zstd", *meta.CompressionCodec)
- // But reading blob data should fail due to compression
- _, err = r.ReadBlob(0)
- assert.ErrorContains(t, err, "compression")
+ got, err := r.ReadBlob(i)
+ require.NoError(t, err)
+ want, err := uncompressed.ReadBlob(i)
+ require.NoError(t, err)
+ assert.Equal(t, want.Data, got.Data)
+ }
}
// TestWriterBitIdenticalWithJava verifies that our writer produces
