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

Author: xuan.zhao

src/iceberg/CMakeLists.txt

@@ -188,6 +188,7 @@ set(ICEBERG_DATA_SOURCES
     data/data_writer.cc
     data/delete_filter.cc
     data/delete_loader.cc
+    data/deletion_vector_writer.cc
     data/equality_delete_writer.cc
     data/file_scan_task_reader.cc
     data/position_delete_writer.cc

src/iceberg/data/delete_loader.cc

@@ -19,9 +19,12 @@
 
 #include "iceberg/data/delete_loader.h"
 
+#include <cstdint>
 #include <cstring>
+#include <limits>
 #include <span>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include <nanoarrow/nanoarrow.h>
@@ -30,6 +33,7 @@
 #include "iceberg/arrow_c_data_guard_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"
@@ -89,10 +93,11 @@ DeleteLoader::DeleteLoader(std::shared_ptr<FileIO> io) : io_(std::move(io)) {}
 
 DeleteLoader::~DeleteLoader() = default;
 
-Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteIndex& index,
+Status DeleteLoader::LoadPositionDelete(const std::shared_ptr<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_));
+  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);
@@ -110,15 +115,19 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde
   // `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;
@@ -150,6 +159,9 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde
     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;
@@ -164,14 +176,59 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde
         positions.push_back(pos_data[i]);
       }
     }
+    if (!positions.empty()) {
+      contributed = true;
+    }
     ForEachPositionDelete(positions, index, bulk_scratch);
   }
 
-  return reader->Close();
+  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 DataFile& file, PositionDeleteIndex& index) const {
-  return NotSupported("Loading deletion vectors is not yet supported");
+Status DeleteLoader::LoadDV(const std::shared_ptr<DataFile>& file,
+                            PositionDeleteIndex& index) 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);
+
+  // 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 {};
 }
 
 Result<PositionDeleteIndex> DeleteLoader::LoadPositionDeletes(
@@ -180,21 +237,23 @@ Result<PositionDeleteIndex> DeleteLoader::LoadPositionDeletes(
   PositionDeleteIndex index;
 
   for (const auto& file : delete_files) {
+    ICEBERG_PRECHECK(file != nullptr, "Delete file must not be null");
+
     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));
+      ICEBERG_RETURN_UNEXPECTED(LoadDV(file, index));
       continue;
     }
 
     ICEBERG_PRECHECK(file->content == DataFile::Content::kPositionDeletes,
                      "Expected position delete file but got content type {}",
                      ToString(file->content));
 
-    ICEBERG_RETURN_UNEXPECTED(LoadPositionDelete(*file, index, data_file_path));
+    ICEBERG_RETURN_UNEXPECTED(LoadPositionDelete(file, index, data_file_path));
   }
 
   return index;

src/iceberg/data/delete_loader.h

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