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

Author: xuan.zhao

src/iceberg/CMakeLists.txt

@@ -188,13 +188,15 @@ 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
     data/writer.cc
     deletes/position_delete_index.cc
     deletes/position_delete_range_consumer.cc
     deletes/roaring_position_bitmap.cc
+    puffin/deletion_vector.cc
     puffin/file_metadata.cc
     puffin/json_serde.cc
     puffin/puffin_format.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,9 +33,12 @@
 #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/deletes/roaring_position_bitmap.h"
+#include "iceberg/file_io.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/metadata_columns.h"
+#include "iceberg/puffin/deletion_vector.h"
 #include "iceberg/result.h"
 #include "iceberg/row/arrow_array_wrapper.h"
 #include "iceberg/schema.h"
@@ -171,7 +177,44 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde
 }
 
 Status DeleteLoader::LoadDV(const DataFile& file, PositionDeleteIndex& index) const {
-  return NotSupported("Loading deletion vectors is not yet supported");
+  // 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());
+  ICEBERG_ASSIGN_OR_RAISE(auto bitmap, puffin::DeserializeDeletionVectorBlob(blob));
+
+  // The bitmap cardinality must match the record count recorded in metadata.
+  ICEBERG_PRECHECK(std::cmp_equal(bitmap.Cardinality(), file.record_count),
+                   "Deletion vector cardinality {} does not match record count {}: {}",
+                   bitmap.Cardinality(), file.record_count, file.file_path);
+
+  bitmap.ForEach([&index](int64_t pos) { index.Delete(pos); });
+  return {};
 }
 
 Result<PositionDeleteIndex> DeleteLoader::LoadPositionDeletes(

src/iceberg/data/deletion_vector_writer.cc

@@ -0,0 +1,157 @@
+/*
+ * 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/data/deletion_vector_writer.h"
+
+#include <map>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "iceberg/deletes/roaring_position_bitmap.h"
+#include "iceberg/file_io.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/puffin/deletion_vector.h"
+#include "iceberg/puffin/puffin_writer.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+class DeletionVectorWriter::Impl {
+ public:
+  explicit Impl(DeletionVectorWriterOptions options) : options_(std::move(options)) {}
+
+  Status Delete(std::string_view referenced_data_file, int64_t pos) {
+    ICEBERG_CHECK(!closed_, "Cannot delete after the writer is closed");
+    ICEBERG_PRECHECK(!referenced_data_file.empty(),
+                     "Deletion vector requires a non-empty referenced data file");
+    ICEBERG_PRECHECK(pos >= 0 && pos <= RoaringPositionBitmap::kMaxPosition,
+                     "Invalid deletion vector position: {}", pos);
+    bitmaps_[std::string(referenced_data_file)].Add(pos);
+    return {};
+  }
+
+  Status Close() {
+    if (closed_) {
+      return {};
+    }
+
+    // No deletes: skip creating an orphan Puffin file that no metadata would
+    // reference, matching the Java DV writer.
+    if (bitmaps_.empty()) {
+      closed_ = true;
+      return {};
+    }
+
+    ICEBERG_ASSIGN_OR_RAISE(auto output_file, options_.io->NewOutputFile(options_.path));
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto writer,
+        puffin::PuffinWriter::Make(std::move(output_file), options_.properties));
+
+    // One blob per referenced data file, in deterministic (sorted) order.
+    struct Entry {
+      std::string referenced_data_file;
+      int64_t offset;
+      int64_t length;
+      int64_t cardinality;
+    };
+    std::vector<Entry> entries;
+    entries.reserve(bitmaps_.size());
+    for (auto& [referenced_data_file, bitmap] : bitmaps_) {
+      // Run-length encode before serializing for space efficiency, matching the
+      // Java DV writer.
+      bitmap.Optimize();
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto blob, puffin::MakeDeletionVectorBlob(bitmap, referenced_data_file));
+      ICEBERG_ASSIGN_OR_RAISE(auto blob_metadata, writer->Write(blob));
+      entries.push_back(Entry{
+          .referenced_data_file = referenced_data_file,
+          .offset = blob_metadata.offset,
+          .length = blob_metadata.length,
+          .cardinality = static_cast<int64_t>(bitmap.Cardinality()),
+      });
+    }
+
+    ICEBERG_RETURN_UNEXPECTED(writer->Finish());
+    ICEBERG_ASSIGN_OR_RAISE(file_size_, writer->FileSize());
+
+    const auto spec_id =
+        options_.spec ? std::make_optional(options_.spec->spec_id()) : std::nullopt;
+
+    for (auto& entry : entries) {
+      data_files_.push_back(std::make_shared<DataFile>(DataFile{
+          .content = DataFile::Content::kPositionDeletes,
+          .file_path = options_.path,
+          .file_format = FileFormatType::kPuffin,
+          .partition = options_.partition,
+          .record_count = entry.cardinality,
+          .file_size_in_bytes = file_size_,
+          .referenced_data_file = std::move(entry.referenced_data_file),
+          .content_offset = entry.offset,
+          .content_size_in_bytes = entry.length,
+          .partition_spec_id = spec_id,
+      }));
+    }
+
+    closed_ = true;
+    return {};
+  }
+
+  Result<FileWriter::WriteResult> Metadata() {
+    ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer");
+    FileWriter::WriteResult result;
+    result.data_files = data_files_;
+    return result;
+  }
+
+ private:
+  DeletionVectorWriterOptions options_;
+  // Ordered by referenced data file path for deterministic blob layout.
+  std::map<std::string, RoaringPositionBitmap> bitmaps_;
+  std::vector<std::shared_ptr<DataFile>> data_files_;
+  int64_t file_size_ = -1;
+  bool closed_ = false;
+};
+
+DeletionVectorWriter::DeletionVectorWriter(std::unique_ptr<Impl> impl)
+    : impl_(std::move(impl)) {}
+
+DeletionVectorWriter::~DeletionVectorWriter() = default;
+
+Result<std::unique_ptr<DeletionVectorWriter>> DeletionVectorWriter::Make(
+    DeletionVectorWriterOptions options) {
+  ICEBERG_PRECHECK(options.io != nullptr, "DeletionVectorWriter requires a FileIO");
+  ICEBERG_PRECHECK(!options.path.empty(), "DeletionVectorWriter requires a path");
+  return std::unique_ptr<DeletionVectorWriter>(
+      new DeletionVectorWriter(std::make_unique<Impl>(std::move(options))));
+}
+
+Status DeletionVectorWriter::Delete(std::string_view referenced_data_file, int64_t pos) {
+  return impl_->Delete(referenced_data_file, pos);
+}
+
+Status DeletionVectorWriter::Close() { return impl_->Close(); }
+
+Result<FileWriter::WriteResult> DeletionVectorWriter::Metadata() {
+  return impl_->Metadata();
+}
+
+}  // namespace iceberg

src/iceberg/data/deletion_vector_writer.h

@@ -0,0 +1,92 @@
+/*
+ * 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/data/deletion_vector_writer.h
+/// Writer that emits deletion vectors as `deletion-vector-v1` blobs in a Puffin file.
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+
+#include "iceberg/data/writer.h"
+#include "iceberg/iceberg_data_export.h"
+#include "iceberg/result.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Options for creating a DeletionVectorWriter.
+struct ICEBERG_DATA_EXPORT DeletionVectorWriterOptions {
+  /// Output Puffin file location.
+  std::string path;
+  /// FileIO used to create the Puffin file.
+  std::shared_ptr<FileIO> io;
+  /// Partition spec the referenced data files belong to (optional).
+  std::shared_ptr<PartitionSpec> spec;
+  /// Partition the referenced data files belong to.
+  PartitionValues partition;
+  /// File-level Puffin properties (e.g. "created-by").
+  std::unordered_map<std::string, std::string> properties;
+};
+
+/// \brief Writes one or more deletion vectors into a single Puffin file.
+///
+/// Each referenced data file gets its own `deletion-vector-v1` blob. After
+/// Close(), Metadata() returns one DataFile per blob, each carrying the
+/// content_offset/content_size_in_bytes and referenced_data_file required to
+/// register the deletion vector in a manifest.
+///
+/// \note All referenced data files are assumed to belong to the single
+/// partition supplied in the options.
+class ICEBERG_DATA_EXPORT DeletionVectorWriter {
+ public:
+  ~DeletionVectorWriter();
+
+  DeletionVectorWriter(const DeletionVectorWriter&) = delete;
+  DeletionVectorWriter& operator=(const DeletionVectorWriter&) = delete;
+
+  /// \brief Create a new DeletionVectorWriter.
+  static Result<std::unique_ptr<DeletionVectorWriter>> Make(
+      DeletionVectorWriterOptions options);
+
+  /// \brief Mark a row position as deleted for the given data file.
+  ///
+  /// Positions are accumulated per data file and serialized on Close().
+  Status Delete(std::string_view referenced_data_file, int64_t pos);
+
+  /// \brief Write all accumulated deletion vectors to the Puffin file and close it.
+  Status Close();
+
+  /// \brief Metadata for the DataFiles produced (one per referenced data file).
+  /// \note Must be called after Close().
+  Result<FileWriter::WriteResult> Metadata();
+
+ private:
+  class Impl;
+  std::unique_ptr<Impl> impl_;
+
+  explicit DeletionVectorWriter(std::unique_ptr<Impl> impl);
+};
+
+}  // namespace iceberg

src/iceberg/data/meson.build

@@ -20,6 +20,7 @@ install_headers(
         'data_writer.h',
         'delete_filter.h',
         'delete_loader.h',
+        'deletion_vector_writer.h',
         'equality_delete_writer.h',
         'file_scan_task_reader.h',
         'position_delete_writer.h',

src/iceberg/meson.build

@@ -170,13 +170,15 @@ iceberg_data_sources = files(
     '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',
     'data/writer.cc',
     'deletes/position_delete_index.cc',
     'deletes/position_delete_range_consumer.cc',
     'deletes/roaring_position_bitmap.cc',
+    'puffin/deletion_vector.cc',
     'puffin/file_metadata.cc',
     'puffin/json_serde.cc',
     'puffin/puffin_format.cc',