Align deletion vector writer with Java behavior

- Reuse `WriteResult` for delete writers and report referenced/rewritten delete files.
- Add shared DV read helper and refactor DV loading/writing paths.
- Preserve source delete files in `PositionDeleteIndex` and tighten DV serialization.
- Generate Iceberg-CPP `created-by` metadata consistently in CMake and Meson.
- Fold DV writer tests into existing test targets.

Author: wgtmac

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

@@ -197,6 +197,7 @@ set(ICEBERG_DATA_SOURCES
     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

@@ -58,7 +58,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 +98,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 +127,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

@@ -31,6 +31,7 @@
 
 #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"
@@ -88,17 +89,31 @@ 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)) {}
 
 DeleteLoader::~DeleteLoader() = default;
 
-Status DeleteLoader::LoadPositionDelete(const std::shared_ptr<DataFile>& file,
-                                        PositionDeleteIndex& index,
+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.
-  ICEBERG_ASSIGN_OR_RAISE(auto reader, OpenDeleteFile(*file, PosDeleteSchema(), io_));
+  // 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());
   internal::ArrowSchemaGuard schema_guard(&arrow_schema);
@@ -116,19 +131,15 @@ Status DeleteLoader::LoadPositionDelete(const std::shared_ptr<DataFile>& file,
   // `ForEachPositionDelete`. Trusts the hint -- spec-compliant writers
   // only set it when all rows share one data file.
   const bool use_referenced_data_file_fast_path =
-      file->referenced_data_file.has_value() &&
-      file->referenced_data_file.value() == data_file_path;
+      file.referenced_data_file.has_value() &&
+      file.referenced_data_file.value() == data_file_path;
 
   // Filter-path staging buffer; reused across batches via `clear()`.
   std::vector<int64_t> positions;
   // Scratch buffer for `ForEachPositionDelete`'s bulk dispatch path;
   // reused across batches and across both routing branches.
   std::vector<uint32_t> bulk_scratch;
 
-  // Whether any position for the target data file came from this file, so the
-  // source delete file is recorded once.
-  bool contributed = false;
-
   while (true) {
     ICEBERG_ASSIGN_OR_RAISE(auto batch_opt, reader->Next());
     if (!batch_opt.has_value()) break;
@@ -160,9 +171,6 @@ Status DeleteLoader::LoadPositionDelete(const std::shared_ptr<DataFile>& file,
     const int64_t* pos_data = Int64ValuesBuffer(pos_view);
 
     if (use_referenced_data_file_fast_path) {
-      // Reaching here means length > 0 (empty batches are skipped above), so
-      // this file contributes positions for the target data file.
-      contributed = true;
       ForEachPositionDelete(std::span<const int64_t>(pos_data, length), index,
                             bulk_scratch);
       continue;
@@ -177,67 +185,10 @@ Status DeleteLoader::LoadPositionDelete(const std::shared_ptr<DataFile>& file,
         positions.push_back(pos_data[i]);
       }
     }
-    if (!positions.empty()) {
-      contributed = true;
-    }
     ForEachPositionDelete(positions, index, bulk_scratch);
   }
 
-  ICEBERG_RETURN_UNEXPECTED(reader->Close());
-
-  // Record the source delete file so callers (e.g. DeletionVectorWriter) can
-  // report file-scoped position deletes as rewritten.
-  if (contributed) {
-    index.AddDeleteFile(file);
-  }
-  return {};
-}
-
-Status DeleteLoader::LoadDV(const std::shared_ptr<DataFile>& file,
-                            PositionDeleteIndex& index,
-                            std::string_view data_file_path) const {
-  // A deletion vector must reference exactly one data file; without it the
-  // caller cannot know which data file the positions apply to.
-  ICEBERG_PRECHECK(file->referenced_data_file.has_value(),
-                   "Deletion vector requires referenced_data_file: {}", file->file_path);
-
-  // The DV must reference the data file being read. A mismatch means the caller
-  // grouped delete files incorrectly; failing here avoids silently applying the
-  // wrong DV or hiding bad metadata behind an empty index.
-  ICEBERG_PRECHECK(file->referenced_data_file.value() == data_file_path,
-                   "Deletion vector references {}, not the requested data file {}",
-                   file->referenced_data_file.value(), data_file_path);
-
-  // For deletion vectors, content_offset and content_size_in_bytes point directly
-  // at the DV blob bytes within the Puffin file and are required by the spec.
-  ICEBERG_PRECHECK(
-      file->content_offset.has_value() && file->content_size_in_bytes.has_value(),
-      "Deletion vector requires content_offset and content_size_in_bytes: {}",
-      file->file_path);
-
-  const int64_t offset = file->content_offset.value();
-  const int64_t length = file->content_size_in_bytes.value();
-  ICEBERG_PRECHECK(offset >= 0 && length >= 0,
-                   "Invalid deletion vector offset/length: offset={}, length={}", offset,
-                   length);
-  ICEBERG_PRECHECK(length <= std::numeric_limits<int32_t>::max(),
-                   "Cannot read deletion vector larger than 2GB: {}", length);
-
-  ICEBERG_ASSIGN_OR_RAISE(auto input_file, io_->NewInputFile(file->file_path));
-  ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file->Open());
-
-  std::vector<std::byte> bytes(static_cast<size_t>(length));
-  ICEBERG_RETURN_UNEXPECTED(stream->ReadFully(offset, bytes));
-  ICEBERG_RETURN_UNEXPECTED(stream->Close());
-
-  std::span<const uint8_t> blob(reinterpret_cast<const uint8_t*>(bytes.data()),
-                                bytes.size());
-  // Deserialize validates the blob length and cardinality against `file` and
-  // retains it as the source delete file.
-  ICEBERG_ASSIGN_OR_RAISE(auto dv, PositionDeleteIndex::Deserialize(blob, file));
-
-  index.Merge(dv);
-  return {};
+  return reader->Close();
 }
 
 Result<PositionDeleteIndex> DeleteLoader::LoadPositionDeletes(
@@ -247,19 +198,14 @@ Result<PositionDeleteIndex> DeleteLoader::LoadPositionDeletes(
     ICEBERG_PRECHECK(file != nullptr, "Delete file must not be null");
   }
 
-  PositionDeleteIndex index;
-
-  // A single deletion vector replaces all other deletes for a data file, so it
-  // is read through the dedicated DV path and validated against the requested
-  // data file.
   if (ContentFileUtil::ContainsSingleDV(delete_files)) {
-    ICEBERG_RETURN_UNEXPECTED(LoadDV(delete_files.front(), index, data_file_path));
-    return index;
+    const auto& dv = delete_files.front();
+    ICEBERG_RETURN_UNEXPECTED(ValidateDV(*dv, data_file_path));
+    return DVUtil::ReadDV(dv, io_);
   }
 
-  // Otherwise all entries must be position delete files. A DV must not be mixed
-  // with position deletes: once a DV applies, readers ignore matching position
-  // deletes, so a mixed list means the caller grouped delete files incorrectly.
+  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: {}",
@@ -268,14 +214,18 @@ Result<PositionDeleteIndex> DeleteLoader::LoadPositionDeletes(
                      "Expected position delete file but got content type {}",
                      ToString(file->content));
 
-    // A file-scoped position delete for another data file has nothing for the
-    // target path; skip it as an optimization.
     if (file->referenced_data_file.has_value() &&
         file->referenced_data_file.value() != data_file_path) {
       continue;
     }
 
-    ICEBERG_RETURN_UNEXPECTED(LoadPositionDelete(file, index, data_file_path));
+    position_delete_files.push_back(file);
+  }
+
+  PositionDeleteIndex index(position_delete_files);
+
+  for (const auto& file : position_delete_files) {
+    ICEBERG_RETURN_UNEXPECTED(LoadPositionDelete(*file, index, data_file_path));
   }
 
   return index;

src/iceberg/data/delete_loader.h

@@ -67,14 +67,9 @@ class ICEBERG_DATA_EXPORT DeleteLoader {
 
  private:
   /// \brief Load a single position delete file into the index.
-  Status LoadPositionDelete(const std::shared_ptr<DataFile>& file,
-                            PositionDeleteIndex& index,
+  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 std::shared_ptr<DataFile>& file, PositionDeleteIndex& index,
-                std::string_view data_file_path) const;
-
   std::shared_ptr<FileIO> io_;
 };