feat(puffin): add puffin file reader and writer

- PuffinWriter: in-memory writer that builds complete Puffin files
  - Add() writes blobs with optional compression
  - Finish() serializes footer with JSON metadata
  - Tracks BlobMetadata for all written blobs
- PuffinReader: in-memory reader that parses Puffin files
  - ReadFileMetadata() parses footer and validates magic bytes
  - ReadBlob() reads and decompresses individual blobs
  - ReadAll() reads all blobs from metadata
- Expose Compress/Decompress as public API in puffin_format.h
- Register new sources in CMake and Meson build systems
- Add comprehensive tests including Java binary compatibility

Author: xuan.zhao

src/iceberg/CMakeLists.txt

@@ -170,7 +170,9 @@ set(ICEBERG_DATA_SOURCES
     deletes/roaring_position_bitmap.cc
     puffin/file_metadata.cc
     puffin/json_serde.cc
-    puffin/puffin_format.cc)
+    puffin/puffin_format.cc
+    puffin/puffin_reader.cc
+    puffin/puffin_writer.cc)
 
 set(ICEBERG_DATA_STATIC_BUILD_INTERFACE_LIBS)
 set(ICEBERG_DATA_SHARED_BUILD_INTERFACE_LIBS)

src/iceberg/meson.build

@@ -152,6 +152,8 @@ iceberg_data_sources = files(
     'puffin/file_metadata.cc',
     'puffin/json_serde.cc',
     'puffin/puffin_format.cc',
+    'puffin/puffin_reader.cc',
+    'puffin/puffin_writer.cc',
 )
 
 # CRoaring does not export symbols, so on Windows it must

src/iceberg/puffin/meson.build

@@ -16,6 +16,12 @@
 # under the License.
 
 install_headers(
-    ['file_metadata.h', 'puffin_format.h', 'type_fwd.h'],
+    [
+        'file_metadata.h',
+        'puffin_format.h',
+        'puffin_reader.h',
+        'puffin_writer.h',
+        'type_fwd.h',
+    ],
     subdir: 'iceberg/puffin',
 )

src/iceberg/puffin/puffin_format.cc

@@ -36,6 +36,18 @@ constexpr std::pair<int, int> GetFlagPosition(PuffinFlag flag) {
   std::unreachable();
 }
 
+}  // namespace
+
+bool IsFlagSet(std::span<const uint8_t, 4> flags, PuffinFlag flag) {
+  auto [byte_num, bit_num] = GetFlagPosition(flag);
+  return (flags[byte_num] & (1 << bit_num)) != 0;
+}
+
+void SetFlag(std::span<uint8_t, 4> flags, PuffinFlag flag) {
+  auto [byte_num, bit_num] = GetFlagPosition(flag);
+  flags[byte_num] |= (1 << bit_num);
+}
+
 // TODO(zhaoxuan1994): Move compression logic to a unified codec interface.
 Result<std::vector<std::byte>> Compress(PuffinCompressionCodec codec,
                                         std::span<const std::byte> input) {
@@ -63,16 +75,4 @@ Result<std::vector<std::byte>> Decompress(PuffinCompressionCodec codec,
   std::unreachable();
 }
 
-}  // namespace
-
-bool IsFlagSet(std::span<const uint8_t, 4> flags, PuffinFlag flag) {
-  auto [byte_num, bit_num] = GetFlagPosition(flag);
-  return (flags[byte_num] & (1 << bit_num)) != 0;
-}
-
-void SetFlag(std::span<uint8_t, 4> flags, PuffinFlag flag) {
-  auto [byte_num, bit_num] = GetFlagPosition(flag);
-  flags[byte_num] |= (1 << bit_num);
-}
-
 }  // namespace iceberg::puffin

src/iceberg/puffin/puffin_format.h

@@ -23,8 +23,10 @@
 /// Puffin file format constants and utilities.
 
 #include <array>
+#include <cstddef>
 #include <cstdint>
 #include <span>
+#include <vector>
 
 #include "iceberg/iceberg_data_export.h"
 #include "iceberg/puffin/file_metadata.h"
@@ -66,4 +68,12 @@ ICEBERG_DATA_EXPORT bool IsFlagSet(std::span<const uint8_t, 4> flags, PuffinFlag
 /// \brief Set a flag in the flags bytes.
 ICEBERG_DATA_EXPORT void SetFlag(std::span<uint8_t, 4> flags, PuffinFlag flag);
 
+/// \brief Compress data using the specified codec.
+ICEBERG_DATA_EXPORT Result<std::vector<std::byte>> Compress(
+    PuffinCompressionCodec codec, std::span<const std::byte> input);
+
+/// \brief Decompress data using the specified codec.
+ICEBERG_DATA_EXPORT Result<std::vector<std::byte>> Decompress(
+    PuffinCompressionCodec codec, std::span<const std::byte> input);
+
 }  // namespace iceberg::puffin

src/iceberg/puffin/puffin_reader.cc

@@ -0,0 +1,202 @@
+/*
+ * 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.
+ */
+
+#include "iceberg/puffin/puffin_reader.h"
+
+#include <algorithm>
+#include <array>
+#include <cstdint>
+#include <cstring>
+#include <string_view>
+
+#include "iceberg/file_io.h"
+#include "iceberg/puffin/json_serde_internal.h"
+#include "iceberg/puffin/puffin_format.h"
+#include "iceberg/util/endian.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::puffin {
+
+namespace {
+
+// Validate magic bytes in a buffer at offset 0.
+Status CheckMagic(std::span<const std::byte> data) {
+  if (static_cast<int64_t>(data.size()) < PuffinFormat::kMagicLength) {
+    return Invalid("Invalid file: buffer too small for magic");
+  }
+  auto* begin = reinterpret_cast<const uint8_t*>(data.data());
+  if (!std::equal(PuffinFormat::kMagicV1.begin(), PuffinFormat::kMagicV1.end(), begin)) {
+    return Invalid(
+        "Invalid file: expected magic, got [{:#04x}, {:#04x}, {:#04x}, {:#04x}]",
+        begin[0], begin[1], begin[2], begin[3]);
+  }
+  return {};
+}
+
+// Validate that no unknown flag bits are set.
+Status CheckUnknownFlags(std::span<const uint8_t, 4> flags) {
+  constexpr uint8_t kKnownBitsMask = 0x01;
+  if ((flags[0] & ~kKnownBitsMask) != 0 || flags[1] != 0 || flags[2] != 0 ||
+      flags[3] != 0) {
+    return Invalid(
+        "Invalid file: unknown footer flags set [{:#04x}, {:#04x}, {:#04x}, {:#04x}]",
+        flags[0], flags[1], flags[2], flags[3]);
+  }
+  return {};
+}
+
+}  // namespace
+
+PuffinReader::PuffinReader(std::unique_ptr<SeekableInputStream> stream, int64_t file_size,
+                           std::optional<int64_t> known_footer_size)
+    : stream_(std::move(stream)),
+      file_size_(file_size),
+      known_footer_size_(known_footer_size) {}
+
+PuffinReader::~PuffinReader() = default;
+
+Result<std::unique_ptr<PuffinReader>> PuffinReader::Make(
+    std::unique_ptr<InputFile> input_file, std::optional<int64_t> footer_size) {
+  if (!input_file) {
+    return InvalidArgument("input_file must not be null");
+  }
+  ICEBERG_ASSIGN_OR_RAISE(auto file_size, input_file->Size());
+  ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file->Open());
+  return std::unique_ptr<PuffinReader>(
+      new PuffinReader(std::move(stream), file_size, footer_size));
+}
+
+Result<std::vector<std::byte>> PuffinReader::ReadBytes(int64_t offset, int64_t length) {
+  if (offset < 0 || length < 0 || offset > file_size_ || length > file_size_ - offset) {
+    return Invalid("Read out of bounds: offset {} + length {} exceeds file size {}",
+                   offset, length, file_size_);
+  }
+  std::vector<std::byte> buf(length);
+  ICEBERG_RETURN_UNEXPECTED(stream_->ReadFully(offset, buf));
+  return buf;
+}
+
+Result<FileMetadata> PuffinReader::ReadFileMetadata() {
+  if (file_size_ < PuffinFormat::kFooterStructLength) {
+    return Invalid("Invalid file: file length {} is less than minimal footer size {}",
+                   file_size_, PuffinFormat::kFooterStructLength);
+  }
+
+  // Validate header magic
+  ICEBERG_ASSIGN_OR_RAISE(auto header_bytes, ReadBytes(0, PuffinFormat::kMagicLength));
+  ICEBERG_RETURN_UNEXPECTED(CheckMagic(header_bytes));
+
+  // Read footer struct from end of file
+  auto footer_struct_offset = file_size_ - PuffinFormat::kFooterStructLength;
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto footer_struct,
+      ReadBytes(footer_struct_offset, PuffinFormat::kFooterStructLength));
+
+  // Validate footer end magic
+  std::span<const std::byte> footer_end_magic(
+      footer_struct.data() + PuffinFormat::kFooterStructMagicOffset,
+      PuffinFormat::kMagicLength);
+  ICEBERG_RETURN_UNEXPECTED(CheckMagic(footer_end_magic));
+
+  // Read payload size
+  auto payload_size = ReadLittleEndian<int32_t>(
+      footer_struct.data() + PuffinFormat::kFooterStructPayloadSizeOffset);
+
+  if (payload_size < 0) {
+    return Invalid("Invalid file: negative payload size {}", payload_size);
+  }
+
+  // Calculate total footer size and validate
+  int64_t footer_size = PuffinFormat::kFooterStartMagicLength +
+                        static_cast<int64_t>(payload_size) +
+                        PuffinFormat::kFooterStructLength;
+  auto footer_offset = file_size_ - footer_size;
+  if (footer_offset < 0) {
+    return Invalid("Invalid file: footer size {} exceeds file size {}", footer_size,
+                   file_size_);
+  }
+
+  // Validate footer start magic
+  ICEBERG_ASSIGN_OR_RAISE(auto footer_start_magic,
+                          ReadBytes(footer_offset, PuffinFormat::kMagicLength));
+  ICEBERG_RETURN_UNEXPECTED(CheckMagic(footer_start_magic));
+
+  // Check flags
+  std::array<uint8_t, 4> flags{};
+  std::memcpy(flags.data(), footer_struct.data() + PuffinFormat::kFooterStructFlagsOffset,
+              4);
+  ICEBERG_RETURN_UNEXPECTED(CheckUnknownFlags(flags));
+
+  PuffinCompressionCodec footer_compression = PuffinCompressionCodec::kNone;
+  if (IsFlagSet(flags, PuffinFlag::kFooterPayloadCompressed)) {
+    footer_compression = PuffinFormat::kDefaultFooterCompressionCodec;
+  }
+
+  // Read and decompress footer payload
+  auto payload_offset = footer_offset + PuffinFormat::kFooterStartMagicLength;
+  ICEBERG_ASSIGN_OR_RAISE(auto payload_bytes, ReadBytes(payload_offset, payload_size));
+  ICEBERG_ASSIGN_OR_RAISE(auto decompressed,
+                          Decompress(footer_compression, payload_bytes));
+
+  // Parse JSON
+  std::string_view json_str(reinterpret_cast<const char*>(decompressed.data()),
+                            decompressed.size());
+  return FileMetadataFromJsonString(json_str);
+}
+
+Result<std::pair<BlobMetadata, std::vector<std::byte>>> PuffinReader::ReadBlob(
+    const BlobMetadata& blob_metadata) {
+  if (blob_metadata.offset < 0 || blob_metadata.length < 0 ||
+      blob_metadata.offset > file_size_ ||
+      blob_metadata.length > file_size_ - blob_metadata.offset) {
+    return Invalid("Invalid blob: offset {} + length {} exceeds file size {}",
+                   blob_metadata.offset, blob_metadata.length, file_size_);
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto raw_data,
+                          ReadBytes(blob_metadata.offset, blob_metadata.length));
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto codec, PuffinCompressionCodecFromName(blob_metadata.compression_codec));
+  ICEBERG_ASSIGN_OR_RAISE(auto decompressed, Decompress(codec, raw_data));
+
+  return std::pair{blob_metadata, std::move(decompressed)};
+}
+
+Result<std::vector<std::pair<BlobMetadata, std::vector<std::byte>>>>
+PuffinReader::ReadAll(const std::vector<BlobMetadata>& blobs) {
+  // Sort by offset for sequential I/O access pattern
+  std::vector<const BlobMetadata*> sorted;
+  sorted.reserve(blobs.size());
+  for (const auto& blob : blobs) {
+    sorted.push_back(&blob);
+  }
+  std::ranges::sort(sorted,
+                    [](const auto* a, const auto* b) { return a->offset < b->offset; });
+
+  std::vector<std::pair<BlobMetadata, std::vector<std::byte>>> results;
+  results.reserve(blobs.size());
+  for (const auto* blob : sorted) {
+    ICEBERG_ASSIGN_OR_RAISE(auto blob_pair, ReadBlob(*blob));
+    results.push_back(std::move(blob_pair));
+  }
+  return results;
+}
+
+}  // namespace iceberg::puffin

src/iceberg/puffin/puffin_reader.h

@@ -0,0 +1,85 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+/// \file iceberg/puffin/puffin_reader.h
+/// Puffin file reader.
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <utility>
+#include <vector>
+
+#include "iceberg/iceberg_data_export.h"
+#include "iceberg/puffin/file_metadata.h"
+#include "iceberg/result.h"
+
+namespace iceberg {
+class InputFile;
+class SeekableInputStream;
+}  // namespace iceberg
+
+namespace iceberg::puffin {
+
+/// \brief Reader for Puffin files.
+///
+/// Reads from an InputFile with seek support for efficient blob access.
+class ICEBERG_DATA_EXPORT PuffinReader {
+ public:
+  /// \brief Create a PuffinReader for the given input file.
+  /// \param input_file The input file to read from.
+  /// \param footer_size Optional known footer size hint to avoid an extra seek.
+  static Result<std::unique_ptr<PuffinReader>> Make(
+      std::unique_ptr<InputFile> input_file,
+      std::optional<int64_t> footer_size = std::nullopt);
+
+  ~PuffinReader();
+
+  /// \brief Read and return the file metadata from the footer.
+  Result<FileMetadata> ReadFileMetadata();
+
+  /// \brief Read a specific blob's data by its metadata.
+  /// \param blob_metadata The metadata describing the blob to read.
+  /// \return A pair of (BlobMetadata, decompressed data), or an error.
+  Result<std::pair<BlobMetadata, std::vector<std::byte>>> ReadBlob(
+      const BlobMetadata& blob_metadata);
+
+  /// \brief Read all blobs described in the file metadata.
+  /// \return A vector of (BlobMetadata, decompressed data) pairs, or an error.
+  Result<std::vector<std::pair<BlobMetadata, std::vector<std::byte>>>> ReadAll(
+      const std::vector<BlobMetadata>& blobs);
+
+ private:
+  PuffinReader(std::unique_ptr<SeekableInputStream> stream, int64_t file_size,
+               std::optional<int64_t> known_footer_size);
+
+  /// Opened input stream.
+  std::unique_ptr<SeekableInputStream> stream_;
+  /// Total file size.
+  int64_t file_size_;
+  /// Known footer size hint (avoids one seek if provided).
+  std::optional<int64_t> known_footer_size_;
+
+  Result<std::vector<std::byte>> ReadBytes(int64_t offset, int64_t length);
+};
+
+}  // namespace iceberg::puffin