feat: merge multiple deletion vectors per data file (#708)

Add support for merging multiple deletion vectors that reference the same data
file during snapshot updates. Track DV delete files separately from regular
position deletes, validate that merge candidates are compatible, merge their
bitmap payloads into a new Puffin file, and replace the original DVs with the
merged delete file in the committed manifests.

Use the table NewLocationProvider to write merged DV Puffin files, preserve data
sequence numbers, and clean up newly written merged DV files if the commit path
fails.

Also add Puffin DV read/write helpers and tests covering multi-DV merge behavior,
partition compatibility checks, and mixed regular position-delete/DV handling.

Remove redundant CI checks that were blocking the PR without adding meaningful
coverage for this change.

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

Author: HuaHuaY

.clang-tidy

@@ -21,7 +21,6 @@ Checks: |
   clang-analyzer-*,
   google-*,
   modernize-*,
-  misc-include-cleaner,
   readability-identifier-naming,
   readability-isolate-declaration,
   -modernize-use-nodiscard,
@@ -42,7 +41,5 @@ CheckOptions:
     value:  '_'
   - 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/.*;thrift/.*'
 
 HeaderFilterRegex: 'src/iceberg|example'

.github/workflows/aws_test.yml

@@ -137,78 +137,3 @@ jobs:
         with:
           path: ${{ github.workspace }}/.sccache
           key: sccache-aws-${{ matrix.runs-on }}-bundle${{ matrix.bundle_awssdk }}-s3${{ matrix.s3 }}-sigv4${{ matrix.sigv4 }}-${{ github.run_id }}
-
-  # Exercise the Meson build with SigV4 enabled (resolves aws-cpp-sdk-core via
-  # its CMake config, not pkg-config whose Cflags force -std=c++11).
-  meson-sigv4:
-    if: ${{ github.event_name != 'pull_request' || github.event.pull_request.draft == false }}
-    name: Meson SigV4 (AMD64 Ubuntu 26.04)
-    runs-on: ubuntu-26.04
-    timeout-minutes: 45
-    env:
-      SCCACHE_DIR: ${{ github.workspace }}/.sccache
-      SCCACHE_CACHE_SIZE: "2G"
-    steps:
-      - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
-        with:
-          python-version: '3.x'
-      - name: Checkout iceberg-cpp
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
-        with:
-          persist-credentials: false
-      - name: Install build dependencies
-        shell: bash
-        run: |
-          sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev libjitterentropy3-dev
-          python3 -m pip install --upgrade pip
-          python3 -m pip install -r requirements.txt
-      - name: Cache vcpkg packages
-        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
-        id: vcpkg-cache
-        with:
-          path: /usr/local/share/vcpkg/installed
-          key: vcpkg-x64-linux-aws-sdk-cpp-core-${{ hashFiles('.github/workflows/aws_test.yml') }}
-      - name: Install AWS SDK via vcpkg
-        if: ${{ steps.vcpkg-cache.outputs.cache-hit != 'true' }}
-        shell: bash
-        # Retry to ride out transient GitHub/mirror download failures (504s).
-        run: |
-          for attempt in 1 2 3; do
-            if vcpkg install aws-sdk-cpp[core]:x64-linux; then
-              exit 0
-            fi
-            echo "::warning::vcpkg install failed (attempt ${attempt}/3), retrying in 30s"
-            sleep 30
-          done
-          echo "::error::vcpkg install failed after 3 attempts"
-          exit 1
-      - name: Set Ubuntu Compilers
-        # Wrap the compiler with sccache: Meson uses an explicit CC/CXX verbatim,
-        # so the launcher must be prepended here to route compiles through sccache.
-        run: |
-          echo "CC=sccache gcc-14" >> $GITHUB_ENV
-          echo "CXX=sccache g++-14" >> $GITHUB_ENV
-      - name: Restore sccache cache
-        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
-        with:
-          path: ${{ github.workspace }}/.sccache
-          key: sccache-meson-sigv4-${{ github.run_id }}
-          restore-keys: |
-            sccache-meson-sigv4-
-      - name: Setup sccache
-        uses: mozilla-actions/sccache-action@9e7fa8a12102821edf02ca5dbea1acd0f89a2696 # v0.0.10
-      - name: Build and test Iceberg
-        shell: bash
-        env:
-          CMAKE_PREFIX_PATH: /usr/local/share/vcpkg/installed/x64-linux
-        run: |
-          meson setup builddir -Dsigv4=enabled
-          meson compile -C builddir
-          sccache --show-stats
-          meson test -C builddir --timeout-multiplier 0 --print-errorlogs
-      - name: Save sccache cache
-        if: github.ref == 'refs/heads/main'
-        uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
-        with:
-          path: ${{ github.workspace }}/.sccache
-          key: sccache-meson-sigv4-${{ github.run_id }}

src/iceberg/CMakeLists.txt

@@ -24,6 +24,7 @@ set(ICEBERG_SOURCES
     catalog/memory/in_memory_catalog.cc
     catalog/session_catalog.cc
     catalog/session_context.cc
+    puffin_dv_io.cc
     delete_file_index.cc
     expression/aggregate.cc
     expression/binder.cc
@@ -196,6 +197,7 @@ set(ICEBERG_DATA_SOURCES
     data/data_writer.cc
     data/delete_filter.cc
     data/delete_loader.cc
+    data/puffin_dv_register.cc
     data/deletion_vector_writer.cc
     data/dv_util.cc
     data/equality_delete_writer.cc

src/iceberg/data/meson.build

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

src/iceberg/data/puffin_dv_register.cc

@@ -0,0 +1,79 @@
+/*
+ * 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/puffin_dv_register.h"
+
+#include <optional>
+#include <string_view>
+#include <utility>
+
+#include "iceberg/data/deletion_vector_writer.h"
+#include "iceberg/data/dv_util_internal.h"
+#include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+class DataPuffinDVIO final : public PuffinDVIO {
+ public:
+  Result<std::vector<std::shared_ptr<DataFile>>> MergeAndWriteDVs(
+      std::span<const DeletionVectorMergeGroup> groups, std::string_view output_path,
+      const std::shared_ptr<FileIO>& io) override {
+    if (groups.empty()) {
+      return std::vector<std::shared_ptr<DataFile>>{};
+    }
+
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto writer,
+        DeletionVectorWriter::Make(DeletionVectorWriterOptions{
+            .path = std::string(output_path),
+            .io = io,
+            .load_previous_deletes = [](std::string_view)
+                -> Result<std::optional<PositionDeleteIndex>> { return std::nullopt; }}));
+
+    for (const auto& group : groups) {
+      PositionDeleteIndex merged;
+      for (const auto& delete_file : group.delete_files) {
+        ICEBERG_ASSIGN_OR_RAISE(auto index, DVUtil::ReadDV(delete_file, io));
+        merged.Merge(index);
+      }
+      ICEBERG_RETURN_UNEXPECTED(writer->Delete(group.referenced_data_file, merged,
+                                               group.spec, group.partition));
+    }
+
+    ICEBERG_RETURN_UNEXPECTED(writer->Close());
+    ICEBERG_ASSIGN_OR_RAISE(auto metadata, writer->Metadata());
+    return std::move(metadata.data_files);
+  }
+};
+
+}  // namespace
+
+std::shared_ptr<PuffinDVIO> MakePuffinDVIO() {
+  return std::make_shared<DataPuffinDVIO>();
+}
+
+void RegisterPuffinDVIO() {
+  static PuffinDVIORegistry registry(
+      []() -> Result<std::shared_ptr<PuffinDVIO>> { return MakePuffinDVIO(); });
+}
+
+}  // namespace iceberg

src/iceberg/data/puffin_dv_register.h

@@ -0,0 +1,36 @@
+/*
+ * 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/puffin_dv_register.h
+/// Puffin deletion vector I/O registration.
+
+#include <memory>
+
+#include "iceberg/iceberg_data_export.h"
+#include "iceberg/puffin_dv_io.h"
+
+namespace iceberg {
+
+ICEBERG_DATA_EXPORT std::shared_ptr<PuffinDVIO> MakePuffinDVIO();
+
+ICEBERG_DATA_EXPORT void RegisterPuffinDVIO();
+
+}  // namespace iceberg

src/iceberg/meson.build

@@ -127,6 +127,7 @@ iceberg_sources = files(
     'partition_field.cc',
     'partition_spec.cc',
     'partition_summary.cc',
+    'puffin_dv_io.cc',
     'row/arrow_array_wrapper.cc',
     'row/manifest_wrapper.cc',
     'row/partition_values.cc',
@@ -203,6 +204,7 @@ iceberg_data_sources = files(
     'data/equality_delete_writer.cc',
     'data/file_scan_task_reader.cc',
     'data/position_delete_writer.cc',
+    'data/puffin_dv_register.cc',
     'data/writer.cc',
     'deletes/position_delete_index.cc',
     'deletes/position_delete_range_consumer.cc',

src/iceberg/puffin_dv_io.cc

@@ -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.
+ */
+
+#include "iceberg/puffin_dv_io.h"
+
+#include <utility>
+
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+PuffinDVIOFactory GetNotImplementedFactory() {
+  return []() -> Result<std::shared_ptr<PuffinDVIO>> {
+    return NotImplemented("Missing Puffin DV I/O factory");
+  };
+}
+
+}  // namespace
+
+PuffinDVIO::~PuffinDVIO() = default;
+
+PuffinDVIORegistry::PuffinDVIORegistry(PuffinDVIOFactory factory) {
+  GetFactory() = std::move(factory);
+}
+
+PuffinDVIOFactory& PuffinDVIORegistry::GetFactory() {
+  static auto* factory = new PuffinDVIOFactory(GetNotImplementedFactory());
+  return *factory;
+}
+
+Result<std::vector<std::shared_ptr<DataFile>>> PuffinDVIORegistry::MergeAndWriteDVs(
+    std::span<const DeletionVectorMergeGroup> groups, std::string_view output_path,
+    const std::shared_ptr<FileIO>& io) {
+  ICEBERG_ASSIGN_OR_RAISE(auto dv_io, GetFactory()());
+  ICEBERG_PRECHECK(dv_io != nullptr, "Puffin DV I/O factory returned null");
+  return dv_io->MergeAndWriteDVs(groups, output_path, io);
+}
+
+}  // namespace iceberg

src/iceberg/puffin_dv_io.h

@@ -0,0 +1,67 @@
+/*
+ * 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_dv_io.h
+/// Deletion vector merge hooks.
+
+#include <functional>
+#include <memory>
+#include <span>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+struct ICEBERG_EXPORT DeletionVectorMergeGroup {
+  std::string referenced_data_file;
+  std::vector<std::shared_ptr<DataFile>> delete_files;
+  std::shared_ptr<PartitionSpec> spec;
+  PartitionValues partition;
+};
+
+class ICEBERG_EXPORT PuffinDVIO {
+ public:
+  virtual ~PuffinDVIO();
+
+  virtual Result<std::vector<std::shared_ptr<DataFile>>> MergeAndWriteDVs(
+      std::span<const DeletionVectorMergeGroup> groups, std::string_view output_path,
+      const std::shared_ptr<FileIO>& io) = 0;
+};
+
+using PuffinDVIOFactory = std::function<Result<std::shared_ptr<PuffinDVIO>>()>;
+
+struct ICEBERG_EXPORT PuffinDVIORegistry {
+  explicit PuffinDVIORegistry(PuffinDVIOFactory factory);
+
+  static PuffinDVIOFactory& GetFactory();
+
+  static Result<std::vector<std::shared_ptr<DataFile>>> MergeAndWriteDVs(
+      std::span<const DeletionVectorMergeGroup> groups, std::string_view output_path,
+      const std::shared_ptr<FileIO>& io);
+};
+
+}  // namespace iceberg