feat(puffin): support deletion-vector-v1 blob read/write (#777)

feat(puffin): support deletion-vector-v1 blob read/write

Add end-to-end support for the `deletion-vector-v1` Puffin blob type on
top of the existing PuffinReader/PuffinWriter and RoaringPositionBitmap.

Encoding (puffin/deletion_vector.{h,cc}):
- SerializeDeletionVectorBlob / DeserializeDeletionVectorBlob implement
the DV blob framing: 4-byte big-endian length, 0xD1D33964 magic, the
portable Roaring vector, and a trailing big-endian CRC-32. Reads validate
the magic and checksum.
- MakeDeletionVectorBlob builds a spec-compliant Blob: fields set to the
  row-position metadata column id, snapshot-id/sequence-number = -1, no
compression, and the required referenced-data-file/cardinality properties.

Writing (data/deletion_vector_writer.{h,cc}):
- DeletionVectorWriter accumulates deleted positions per data file,
run-length
encodes each bitmap before serializing, writes one blob per data file,
and produces a DataFile per blob carrying content_offset /
content_size_in_bytes,  referenced_data_file and record_count for manifest
registration.

Reading (data/delete_loader.cc):
- Implement DeleteLoader::LoadDV: read the blob bytes referenced by
content_offset/content_size_in_bytes, validate referenced_data_file, the
2GB limit and cardinality == record_count, then apply positions to the
index. This wires DV deletes into DeleteFilter::ComputeAliveRows.

Puffin writer convenience:
- PuffinWriter::Make now fills a default `created-by` property when the
caller does not provide one, matching the Java writer.

Behavior matches the Java implementation (BaseDVFileWriter /
BaseDeleteLoader /
BitmapPositionDeleteIndex): magic, byte order, CRC coverage, blob
fields, RLE and the read-side validations are all aligned.

Tests:
- DV blob framing round-trip and error cases (bad magic, corrupted CRC,
  truncated blob, size mismatch).
- DeletionVectorWriter -> DeleteLoader write/read round trip and guard
cases.
- delete_loader: load DV, referenced-file filtering, cardinality
mismatch,
  mixed DV + position-delete loading.
- delete_filter: end-to-end ComputeAliveRows filtering with a real
Puffin DV  file over Arrow FileIO.
- PuffinWriter created-by default and caller-precedence.

Not included (deletion-vector upper layer, follow-up): merging with
previously written DVs (compaction), the bulk delete(PositionDeleteIndex)
API, and per-data-file partition/spec in a single DV file.

---------

Co-authored-by: xuan.zhao <xuan.zhao@clickzetta.com>
Co-authored-by: Gang Wu <ustcwg@gmail.com>

Author: 3 people

.clang-tidy

@@ -43,6 +43,6 @@ CheckOptions:
   - key:    modernize-use-scoped-lock.WarnOnSingleLocks
     value:  'false'
   - key:    misc-include-cleaner.IgnoreHeaders
-    value:  'arrow/.*;avro/.*;aws/.*;cpr/.*;gmock/.*;gtest/.*;nanoarrow/.*;nlohmann/.*;parquet/.*;roaring/.*;spdlog/.*;sqlpp23/.*;sqlite3\.h;thrift/.*;utf8proc\.h;zlib\.h'
+    value:  'arrow/.*;avro/.*;aws/.*;cpr/.*;gmock/.*;gtest/.*;nanoarrow/.*;nlohmann/.*;parquet/.*;roaring/.*;spdlog/.*;sqlpp23/.*;thrift/.*'
 
 HeaderFilterRegex: 'src/iceberg|example'

CMakeLists.txt

@@ -28,10 +28,9 @@ project(Iceberg
         DESCRIPTION "Iceberg C++ Project"
         LANGUAGES CXX)
 
-set(ICEBERG_VERSION_SUFFIX "-SNAPSHOT")
-set(ICEBERG_VERSION_STRING "${PROJECT_VERSION}${ICEBERG_VERSION_SUFFIX}")
-configure_file("${CMAKE_CURRENT_SOURCE_DIR}/src/iceberg/version.h.in"
-               "${CMAKE_BINARY_DIR}/src/iceberg/version.h")
+include(IcebergBuildUtils)
+iceberg_configure_version_header("${CMAKE_CURRENT_SOURCE_DIR}/src/iceberg/version.h.in"
+                                 "${CMAKE_BINARY_DIR}/src/iceberg/version.h")
 
 set(CMAKE_CXX_STANDARD 23)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
@@ -79,7 +78,6 @@ if(ICEBERG_BUILD_REST_INTEGRATION_TESTS AND WIN32)
 endif()
 
 include(CMakeParseArguments)
-include(IcebergBuildUtils)
 include(IcebergSanitizer)
 include(IcebergSccache)
 include(IcebergThirdpartyToolchain)

cmake_modules/IcebergBuildUtils.cmake

@@ -20,6 +20,28 @@
 
 include(CMakePackageConfigHelpers)
 
+function(iceberg_configure_version_header INPUT_FILE OUTPUT_FILE)
+  set(ICEBERG_VERSION_SUFFIX "-SNAPSHOT")
+  set(ICEBERG_VERSION_STRING "${PROJECT_VERSION}${ICEBERG_VERSION_SUFFIX}")
+  set(ICEBERG_GIT_COMMIT_ID "unknown")
+
+  find_package(Git QUIET)
+  if(GIT_FOUND)
+    execute_process(COMMAND "${GIT_EXECUTABLE}" rev-parse HEAD
+                    WORKING_DIRECTORY "${PROJECT_SOURCE_DIR}"
+                    OUTPUT_VARIABLE ICEBERG_GIT_COMMIT_ID
+                    RESULT_VARIABLE ICEBERG_GIT_RESULT
+                    OUTPUT_STRIP_TRAILING_WHITESPACE ERROR_QUIET)
+    if(NOT ICEBERG_GIT_RESULT EQUAL 0 OR NOT ICEBERG_GIT_COMMIT_ID)
+      set(ICEBERG_GIT_COMMIT_ID "unknown")
+    endif()
+  endif()
+
+  set(ICEBERG_FULL_VERSION_STRING
+      "Apache Iceberg-CPP ${ICEBERG_VERSION_STRING} (commit ${ICEBERG_GIT_COMMIT_ID})")
+  configure_file("${INPUT_FILE}" "${OUTPUT_FILE}" @ONLY)
+endfunction()
+
 function(iceberg_install_cmake_package PACKAGE_NAME EXPORT_NAME)
   set(CONFIG_CMAKE "${PACKAGE_NAME}-config.cmake")
   set(BUILT_CONFIG_CMAKE "${CMAKE_CURRENT_BINARY_DIR}/${CONFIG_CMAKE}")

src/iceberg/CMakeLists.txt

@@ -196,6 +196,8 @@ set(ICEBERG_DATA_SOURCES
     data/data_writer.cc
     data/delete_filter.cc
     data/delete_loader.cc
+    data/deletion_vector_writer.cc
+    data/dv_util.cc
     data/equality_delete_writer.cc
     data/file_scan_task_reader.cc
     data/position_delete_writer.cc

src/iceberg/data/data_writer.cc

@@ -20,7 +20,9 @@
 #include "iceberg/data/data_writer.h"
 
 #include <map>
+#include <utility>
 
+#include "iceberg/data/writer.h"  // IWYU pragma: keep
 #include "iceberg/file_writer.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/partition_spec.h"
@@ -58,7 +60,7 @@ class DataWriter::Impl {
     return {};
   }
 
-  Result<FileWriter::WriteResult> Metadata() {
+  Result<WriteResult> Metadata() {
     ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer");
 
     ICEBERG_ASSIGN_OR_RAISE(auto metrics, writer_->metrics());
@@ -98,7 +100,7 @@ class DataWriter::Impl {
             options_.spec ? std::make_optional(options_.spec->spec_id()) : std::nullopt,
     });
 
-    FileWriter::WriteResult result;
+    WriteResult result;
     result.data_files.push_back(std::move(data_file));
     return result;
   }
@@ -127,6 +129,6 @@ Result<int64_t> DataWriter::Length() const { return impl_->Length(); }
 
 Status DataWriter::Close() { return impl_->Close(); }
 
-Result<FileWriter::WriteResult> DataWriter::Metadata() { return impl_->Metadata(); }
+Result<WriteResult> DataWriter::Metadata() { return impl_->Metadata(); }
 
 }  // namespace iceberg

src/iceberg/data/delete_loader.cc

@@ -19,23 +19,29 @@
 
 #include "iceberg/data/delete_loader.h"
 
+#include <cstdint>
 #include <cstring>
+#include <limits>
 #include <span>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include <nanoarrow/nanoarrow.h>
 
 #include "iceberg/arrow/nanoarrow_status_internal.h"
 #include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/data/dv_util_internal.h"
 #include "iceberg/deletes/position_delete_index.h"
 #include "iceberg/deletes/position_delete_range_consumer.h"
+#include "iceberg/file_io.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/metadata_columns.h"
 #include "iceberg/result.h"
 #include "iceberg/row/arrow_array_wrapper.h"
 #include "iceberg/schema.h"
+#include "iceberg/util/content_file_util.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/struct_like_set.h"
 
@@ -83,6 +89,21 @@ bool StringEquals(const ArrowArrayView* view, int64_t row_idx, std::string_view
   return sv.data != nullptr && std::memcmp(sv.data, target.data(), target.size()) == 0;
 }
 
+Status ValidateDV(const DataFile& dv, std::string_view data_file_path) {
+  ICEBERG_PRECHECK(dv.content_offset.has_value(), "Invalid DV, offset cannot be null: {}",
+                   ContentFileUtil::DVDesc(dv));
+  ICEBERG_PRECHECK(dv.content_size_in_bytes.has_value(), "Invalid DV, length is null: {}",
+                   ContentFileUtil::DVDesc(dv));
+  ICEBERG_PRECHECK(
+      dv.content_size_in_bytes.value() <= std::numeric_limits<int32_t>::max(),
+      "Can't read DV larger than 2GB: {}", dv.content_size_in_bytes.value());
+  ICEBERG_PRECHECK(dv.referenced_data_file.has_value() &&
+                       dv.referenced_data_file.value() == data_file_path,
+                   "DV is expected to reference {}, not {}", data_file_path,
+                   dv.referenced_data_file.value_or(""));
+  return {};
+}
+
 }  // namespace
 
 DeleteLoader::DeleteLoader(std::shared_ptr<FileIO> io) : io_(std::move(io)) {}
@@ -91,7 +112,7 @@ DeleteLoader::~DeleteLoader() = default;
 
 Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteIndex& index,
                                         std::string_view data_file_path) const {
-  // TODO(gangwu): push down path filter to open the file.
+  // TODO(gangwu): Add cache hooks, worker pool, and filter pushdown.
   ICEBERG_ASSIGN_OR_RAISE(auto reader, OpenDeleteFile(file, PosDeleteSchema(), io_));
 
   ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, reader->Schema());
@@ -170,30 +191,40 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde
   return reader->Close();
 }
 
-Status DeleteLoader::LoadDV(const DataFile& file, PositionDeleteIndex& index) const {
-  return NotSupported("Loading deletion vectors is not yet supported");
-}
-
 Result<PositionDeleteIndex> DeleteLoader::LoadPositionDeletes(
     std::span<const std::shared_ptr<DataFile>> delete_files,
     std::string_view data_file_path) const {
-  PositionDeleteIndex index;
+  for (const auto& file : delete_files) {
+    ICEBERG_PRECHECK(file != nullptr, "Delete file must not be null");
+  }
+
+  if (ContentFileUtil::ContainsSingleDV(delete_files)) {
+    const auto& dv = delete_files.front();
+    ICEBERG_RETURN_UNEXPECTED(ValidateDV(*dv, data_file_path));
+    return DVUtil::ReadDV(dv, io_);
+  }
+
+  std::vector<std::shared_ptr<DataFile>> position_delete_files;
 
   for (const auto& file : delete_files) {
+    ICEBERG_PRECHECK(!file->IsDeletionVector(),
+                     "Deletion vector cannot be mixed with position delete files: {}",
+                     file->file_path);
+    ICEBERG_PRECHECK(file->content == DataFile::Content::kPositionDeletes,
+                     "Expected position delete file but got content type {}",
+                     ToString(file->content));
+
     if (file->referenced_data_file.has_value() &&
         file->referenced_data_file.value() != data_file_path) {
       continue;
     }
 
-    if (file->IsDeletionVector()) {
-      ICEBERG_RETURN_UNEXPECTED(LoadDV(*file, index));
-      continue;
-    }
+    position_delete_files.push_back(file);
+  }
 
-    ICEBERG_PRECHECK(file->content == DataFile::Content::kPositionDeletes,
-                     "Expected position delete file but got content type {}",
-                     ToString(file->content));
+  PositionDeleteIndex index(position_delete_files);
 
+  for (const auto& file : position_delete_files) {
     ICEBERG_RETURN_UNEXPECTED(LoadPositionDelete(*file, index, data_file_path));
   }
 

src/iceberg/data/delete_loader.h

@@ -70,9 +70,6 @@ class ICEBERG_DATA_EXPORT DeleteLoader {
   Status LoadPositionDelete(const DataFile& file, PositionDeleteIndex& index,
                             std::string_view data_file_path) const;
 
-  /// \brief Load a single deletion vector file into the index.
-  Status LoadDV(const DataFile& file, PositionDeleteIndex& index) const;
-
   std::shared_ptr<FileIO> io_;
 };