zeroshade commented on code in PR #2401: URL: https://github.com/apache/iceberg-go/pull/2401#discussion_r4470412518
########## puffin/blob_writer_compression.go: ########## @@ -0,0 +1,113 @@ +// 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" + "encoding/binary" + "fmt" + + "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) { Review Comment: Until #2400 lands, `Reader.ReadBlob` rejects every blob this produces. Please sequence after #2400 and have the tests read back through `ReadBlob` instead of decoding frames directly. ########## puffin/blob_writer_compression.go: ########## @@ -0,0 +1,113 @@ +// 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" + "encoding/binary" + "fmt" + + "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 "lz4": + // 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 "zstd": + options := []zstd.EOption{zstd.WithEncoderConcurrency(1), zstd.WithZeroFrames(true)} + // Non-single-segment frames omit content size for inputs below 256 bytes. + if len(data) < 256 { + options = append(options, zstd.WithSingleSegment(true)) + } + encoder, err := zstd.NewWriter(nil, options...) + if err != nil { + return nil, fmt.Errorf("puffin: configure zstd compression: %w", err) + } + result := encoder.EncodeAll(data, nil) + if err := encoder.Close(); err != nil { + return nil, fmt.Errorf("puffin: finalize zstd blob: %w", err) + } + + return result, 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. A conventional lz4.Writer cannot emit this header using SizeOption(0). +func emptyLZ4Frame() []byte { + frame := make([]byte, 4+2+8+1+4) + binary.LittleEndian.PutUint32(frame[:4], lz4FrameMagic) + frame[4] = lz4FrameVersionOne | 0x20 | lz4FrameContentSizeFlag + frame[5] = 0x70 // 4 MiB maximum block size + frame[14] = emptyLZ4HeaderChecksum(frame[4:14]) + + return frame +} + +// XXH32 with seed zero for a 10-byte LZ4 frame descriptor. +func emptyLZ4HeaderChecksum(descriptor []byte) byte { Review Comment: Since the empty frame's header is fixed, its checksum byte could be a precomputed constant, verified by the existing `lz4.ValidFrameHeader` test, instead of a hand-written XXH32. ########## puffin/blob_writer_compression.go: ########## @@ -0,0 +1,113 @@ +// 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" + "encoding/binary" + "fmt" + + "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 "lz4": + // 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 "zstd": + options := []zstd.EOption{zstd.WithEncoderConcurrency(1), zstd.WithZeroFrames(true)} + // Non-single-segment frames omit content size for inputs below 256 bytes. + if len(data) < 256 { + options = append(options, zstd.WithSingleSegment(true)) + } + encoder, err := zstd.NewWriter(nil, options...) Review Comment: `zstd.NewWriter` allocates sizeable encoder state on every call; `EncodeAll` is safe to call concurrently, so two package-level encoders (with and without single-segment) would avoid this per-blob cost. ########## 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 + Type BlobType + // CompressionCodec is "", "lz4", or "zstd". An empty value writes raw bytes. + // Deletion vectors must not be compressed per the Puffin specification. + CompressionCodec string Review Comment: Inserting this second breaks struct literals written without field names; adding it as the last field avoids that. Exported codec constants (shared with #2400's reader) would also be easier to use than a bare string. -- 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]
