feat(puffin): add format constants, utilities, and JSON serde (#603)

Author: zhaoxuan1994

src/iceberg/CMakeLists.txt

@@ -67,6 +67,8 @@ set(ICEBERG_SOURCES
     partition_spec.cc
     partition_summary.cc
     puffin/file_metadata.cc
+    puffin/puffin_format.cc
+    puffin/json_serde.cc
     row/arrow_array_wrapper.cc
     row/manifest_wrapper.cc
     row/partition_values.cc

src/iceberg/deletes/roaring_position_bitmap.cc

@@ -49,28 +49,6 @@ int64_t ToPosition(int32_t key, uint32_t pos32) {
   return (int64_t{key} << 32) | int64_t{pos32};
 }
 
-void WriteLE64(char* buf, int64_t value) {
-  auto le = ToLittleEndian(static_cast<uint64_t>(value));
-  std::memcpy(buf, &le, sizeof(le));
-}
-
-void WriteLE32(char* buf, int32_t value) {
-  auto le = ToLittleEndian(static_cast<uint32_t>(value));
-  std::memcpy(buf, &le, sizeof(le));
-}
-
-int64_t ReadLE64(const char* buf) {
-  uint64_t v;
-  std::memcpy(&v, buf, sizeof(v));
-  return static_cast<int64_t>(FromLittleEndian(v));
-}
-
-int32_t ReadLE32(const char* buf) {
-  uint32_t v;
-  std::memcpy(&v, buf, sizeof(v));
-  return static_cast<int32_t>(FromLittleEndian(v));
-}
-
 Status ValidatePosition(int64_t pos) {
   if (pos < 0 || pos > RoaringPositionBitmap::kMaxPosition) {
     return InvalidArgument("Bitmap supports positions that are >= 0 and <= {}: {}",
@@ -205,12 +183,12 @@ Result<std::string> RoaringPositionBitmap::Serialize() const {
   char* buf = result.data();
 
   // Write bitmap count (array length including empties)
-  WriteLE64(buf, static_cast<int64_t>(impl_->bitmaps.size()));
+  WriteLittleEndian(static_cast<int64_t>(impl_->bitmaps.size()), buf);
   buf += kBitmapCountSizeBytes;
 
   // Write each bitmap with its key
   for (int32_t key = 0; std::cmp_less(key, impl_->bitmaps.size()); ++key) {
-    WriteLE32(buf, key);
+    WriteLittleEndian(key, buf);
     buf += kBitmapKeySizeBytes;
     size_t written = impl_->bitmaps[key].write(buf, /*portable=*/true);
     buf += written;
@@ -226,7 +204,7 @@ Result<RoaringPositionBitmap> RoaringPositionBitmap::Deserialize(std::string_vie
   ICEBERG_PRECHECK(remaining >= kBitmapCountSizeBytes,
                    "Buffer too small for bitmap count: {} bytes", remaining);
 
-  int64_t bitmap_count = ReadLE64(buf);
+  auto bitmap_count = ReadLittleEndian<int64_t>(buf);
   buf += kBitmapCountSizeBytes;
   remaining -= kBitmapCountSizeBytes;
 
@@ -242,7 +220,7 @@ Result<RoaringPositionBitmap> RoaringPositionBitmap::Deserialize(std::string_vie
     ICEBERG_PRECHECK(remaining >= kBitmapKeySizeBytes,
                      "Buffer too small for bitmap key: {} bytes", remaining);
 
-    int32_t key = ReadLE32(buf);
+    auto key = ReadLittleEndian<int32_t>(buf);
     buf += kBitmapKeySizeBytes;
     remaining -= kBitmapKeySizeBytes;
 

src/iceberg/meson.build

@@ -89,6 +89,8 @@ iceberg_sources = files(
     'partition_spec.cc',
     'partition_summary.cc',
     'puffin/file_metadata.cc',
+    'puffin/json_serde.cc',
+    'puffin/puffin_format.cc',
     'row/arrow_array_wrapper.cc',
     'row/manifest_wrapper.cc',
     'row/partition_values.cc',

src/iceberg/puffin/json_serde.cc

@@ -0,0 +1,127 @@
+/*
+ * 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 <nlohmann/json.hpp>
+
+#include "iceberg/puffin/file_metadata.h"
+#include "iceberg/puffin/json_serde_internal.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::puffin {
+
+namespace {
+constexpr std::string_view kBlobs = "blobs";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kType = "type";
+constexpr std::string_view kFields = "fields";
+constexpr std::string_view kSnapshotId = "snapshot-id";
+constexpr std::string_view kSequenceNumber = "sequence-number";
+constexpr std::string_view kOffset = "offset";
+constexpr std::string_view kLength = "length";
+constexpr std::string_view kCompressionCodec = "compression-codec";
+}  // namespace
+
+nlohmann::json ToJson(const BlobMetadata& blob_metadata) {
+  nlohmann::json json;
+  json[kType] = blob_metadata.type;
+  json[kFields] = blob_metadata.input_fields;
+  json[kSnapshotId] = blob_metadata.snapshot_id;
+  json[kSequenceNumber] = blob_metadata.sequence_number;
+  json[kOffset] = blob_metadata.offset;
+  json[kLength] = blob_metadata.length;
+
+  SetOptionalStringField(json, kCompressionCodec, blob_metadata.compression_codec);
+  SetContainerField(json, kProperties, blob_metadata.properties);
+
+  return json;
+}
+
+Result<BlobMetadata> BlobMetadataFromJson(const nlohmann::json& json) {
+  BlobMetadata blob_metadata;
+
+  ICEBERG_ASSIGN_OR_RAISE(blob_metadata.type, GetJsonValue<std::string>(json, kType));
+  ICEBERG_ASSIGN_OR_RAISE(blob_metadata.input_fields,
+                          GetJsonValue<std::vector<int32_t>>(json, kFields));
+  ICEBERG_ASSIGN_OR_RAISE(blob_metadata.snapshot_id,
+                          GetJsonValue<int64_t>(json, kSnapshotId));
+  ICEBERG_ASSIGN_OR_RAISE(blob_metadata.sequence_number,
+                          GetJsonValue<int64_t>(json, kSequenceNumber));
+  ICEBERG_ASSIGN_OR_RAISE(blob_metadata.offset, GetJsonValue<int64_t>(json, kOffset));
+  ICEBERG_ASSIGN_OR_RAISE(blob_metadata.length, GetJsonValue<int64_t>(json, kLength));
+  ICEBERG_ASSIGN_OR_RAISE(blob_metadata.compression_codec,
+                          GetJsonValueOrDefault<std::string>(json, kCompressionCodec));
+  ICEBERG_ASSIGN_OR_RAISE(blob_metadata.properties,
+                          FromJsonMap<std::string>(json, kProperties));
+
+  return blob_metadata;
+}
+
+nlohmann::json ToJson(const FileMetadata& file_metadata) {
+  nlohmann::json json;
+
+  nlohmann::json blobs_json = nlohmann::json::array();
+  for (const auto& blob : file_metadata.blobs) {
+    blobs_json.push_back(ToJson(blob));
+  }
+  json[kBlobs] = std::move(blobs_json);
+
+  SetContainerField(json, kProperties, file_metadata.properties);
+
+  return json;
+}
+
+Result<FileMetadata> FileMetadataFromJson(const nlohmann::json& json) {
+  FileMetadata file_metadata;
+
+  ICEBERG_ASSIGN_OR_RAISE(auto blobs_json, GetJsonValue<nlohmann::json>(json, kBlobs));
+  if (!blobs_json.is_array()) {
+    return JsonParseError("Cannot parse blobs from non-array: {}",
+                          SafeDumpJson(blobs_json));
+  }
+
+  for (const auto& blob_json : blobs_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto blob, BlobMetadataFromJson(blob_json));
+    file_metadata.blobs.push_back(std::move(blob));
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(file_metadata.properties,
+                          FromJsonMap<std::string>(json, kProperties));
+
+  return file_metadata;
+}
+
+std::string ToJsonString(const FileMetadata& file_metadata, bool pretty) {
+  auto json = ToJson(file_metadata);
+  return pretty ? json.dump(2) : json.dump();
+}
+
+Result<FileMetadata> FileMetadataFromJsonString(std::string_view json_string) {
+  if (json_string.empty()) {
+    return JsonParseError("Cannot parse empty JSON string");
+  }
+  try {
+    auto json = nlohmann::json::parse(json_string);
+    return FileMetadataFromJson(json);
+  } catch (const nlohmann::json::parse_error& e) {
+    return JsonParseError("Failed to parse JSON: {}", e.what());
+  }
+}
+
+}  // namespace iceberg::puffin

src/iceberg/puffin/json_serde_internal.h

@@ -0,0 +1,56 @@
+/*
+ * 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/json_serde_internal.h
+/// JSON serialization/deserialization for Puffin file metadata.
+
+#include <string>
+#include <string_view>
+
+#include <nlohmann/json_fwd.hpp>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/puffin/type_fwd.h"
+#include "iceberg/result.h"
+
+namespace iceberg::puffin {
+
+/// \brief Serialize a BlobMetadata to JSON.
+ICEBERG_EXPORT nlohmann::json ToJson(const BlobMetadata& blob_metadata);
+
+/// \brief Deserialize a BlobMetadata from JSON.
+ICEBERG_EXPORT Result<BlobMetadata> BlobMetadataFromJson(const nlohmann::json& json);
+
+/// \brief Serialize a FileMetadata to JSON.
+ICEBERG_EXPORT nlohmann::json ToJson(const FileMetadata& file_metadata);
+
+/// \brief Deserialize a FileMetadata from JSON.
+ICEBERG_EXPORT Result<FileMetadata> FileMetadataFromJson(const nlohmann::json& json);
+
+/// \brief Serialize a FileMetadata to a JSON string.
+ICEBERG_EXPORT std::string ToJsonString(const FileMetadata& file_metadata,
+                                        bool pretty = false);
+
+/// \brief Deserialize a FileMetadata from a JSON string.
+ICEBERG_EXPORT Result<FileMetadata> FileMetadataFromJsonString(
+    std::string_view json_string);
+
+}  // namespace iceberg::puffin

src/iceberg/puffin/meson.build

@@ -15,4 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-install_headers(['file_metadata.h'], subdir: 'iceberg/puffin')
+install_headers(
+    ['file_metadata.h', 'puffin_format.h', 'type_fwd.h'],
+    subdir: 'iceberg/puffin',
+)

src/iceberg/puffin/puffin_format.cc

@@ -0,0 +1,78 @@
+/*
+ * 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_format.h"
+
+#include <cstddef>
+#include <utility>
+#include <vector>
+
+namespace iceberg::puffin {
+
+namespace {
+
+// Returns (byte_index, bit_index) for a given flag within the 4-byte flags field.
+constexpr std::pair<int, int> GetFlagPosition(PuffinFlag flag) {
+  switch (flag) {
+    case PuffinFlag::kFooterPayloadCompressed:
+      return {0, 0};
+  }
+  std::unreachable();
+}
+
+// TODO(zhaoxuan1994): Move compression logic to a unified codec interface.
+Result<std::vector<std::byte>> Compress(PuffinCompressionCodec codec,
+                                        std::span<const std::byte> input) {
+  switch (codec) {
+    case PuffinCompressionCodec::kNone:
+      return std::vector<std::byte>(input.begin(), input.end());
+    case PuffinCompressionCodec::kLz4:
+      return NotSupported("LZ4 compression is not yet supported");
+    case PuffinCompressionCodec::kZstd:
+      return NotSupported("Zstd compression is not yet supported");
+  }
+  std::unreachable();
+}
+
+Result<std::vector<std::byte>> Decompress(PuffinCompressionCodec codec,
+                                          std::span<const std::byte> input) {
+  switch (codec) {
+    case PuffinCompressionCodec::kNone:
+      return std::vector<std::byte>(input.begin(), input.end());
+    case PuffinCompressionCodec::kLz4:
+      return NotSupported("LZ4 decompression is not yet supported");
+    case PuffinCompressionCodec::kZstd:
+      return NotSupported("Zstd decompression is not yet supported");
+  }
+  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