feat: implement merge multiple DVs

Author: HuaHuaY

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