feat: add deletion vector compaction planner

Add a metadata-only planner that selects rewrite candidates by file size and
file-scoped position-delete pressure, then groups candidates by partition.
Expose focused configuration and result models for later rewrite execution.

Author: wirybeaver

src/iceberg/CMakeLists.txt

@@ -26,6 +26,7 @@ set(ICEBERG_SOURCES
     catalog/memory/in_memory_catalog.cc
     catalog/session_catalog.cc
     catalog/session_context.cc
+    compaction_planner.cc
     delete_file_index.cc
     deletes/dv_util.cc
     deletes/dv_writer.cc

src/iceberg/compaction_planner.cc

@@ -0,0 +1,125 @@
+/*
+ * 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/compaction_planner.h"
+
+#include <algorithm>
+#include <utility>
+
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/table_scan.h"
+#include "iceberg/util/content_file_util.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+struct PendingGroup {
+  CompactionGroup group;
+  bool rewrite_independently = false;
+};
+
+Result<CompactionFile> BuildCompactionFile(
+    const std::shared_ptr<FileScanTask>& scan_task) {
+  CompactionFile result{.scan_task = scan_task};
+  const auto& data_file = scan_task->data_file();
+
+  for (const auto& delete_file : scan_task->delete_files()) {
+    if (delete_file->content != DataFile::Content::kPositionDeletes) {
+      continue;
+    }
+
+    ICEBERG_ASSIGN_OR_RAISE(auto referenced_file,
+                            ContentFileUtil::ReferencedDataFile(*delete_file));
+    if (referenced_file != data_file->file_path) {
+      continue;
+    }
+
+    ++result.file_scoped_delete_count;
+    result.file_scoped_delete_record_count += delete_file->record_count;
+  }
+
+  return result;
+}
+
+PendingGroup& FindOrAddGroup(std::vector<PendingGroup>& groups,
+                             const DataFile& data_file) {
+  auto group = std::ranges::find_if(groups, [&](const PendingGroup& candidate) {
+    return candidate.group.partition_spec_id == data_file.partition_spec_id &&
+           candidate.group.partition == data_file.partition;
+  });
+  if (group != groups.end()) {
+    return *group;
+  }
+
+  groups.push_back(PendingGroup{.group = CompactionGroup{
+                                    .partition_spec_id = data_file.partition_spec_id,
+                                    .partition = data_file.partition,
+                                }});
+  return groups.back();
+}
+
+}  // namespace
+
+Result<CompactionPlan> CompactionPlanner::Plan(
+    std::span<const std::shared_ptr<FileScanTask>> scan_tasks,
+    const CompactionPlannerConfig& config) {
+  const long double min_file_size =
+      config.target_file_size_bytes * config.min_file_size_ratio;
+  const long double max_file_size =
+      config.target_file_size_bytes * config.max_file_size_ratio;
+  std::vector<PendingGroup> pending_groups;
+
+  for (const auto& scan_task : scan_tasks) {
+    ICEBERG_PRECHECK(scan_task != nullptr, "File scan task must not be null");
+    const auto& data_file = *scan_task->data_file();
+    ICEBERG_ASSIGN_OR_RAISE(auto file, BuildCompactionFile(scan_task));
+
+    const bool is_small = data_file.file_size_in_bytes < min_file_size;
+    const bool is_oversized = data_file.file_size_in_bytes > max_file_size;
+    const bool has_many_delete_files =
+        file.file_scoped_delete_count >= config.delete_file_threshold;
+    const bool has_high_delete_ratio =
+        data_file.record_count > 0 &&
+        static_cast<double>(file.file_scoped_delete_record_count) /
+                data_file.record_count >=
+            config.delete_ratio_threshold;
+    if (!is_small && !is_oversized && !has_many_delete_files && !has_high_delete_ratio) {
+      continue;
+    }
+
+    auto& group = FindOrAddGroup(pending_groups, data_file);
+    group.rewrite_independently |=
+        is_oversized || has_many_delete_files || has_high_delete_ratio;
+    group.group.data_file_size_bytes += data_file.file_size_in_bytes;
+    group.group.file_scoped_delete_record_count += file.file_scoped_delete_record_count;
+    group.group.files.push_back(std::move(file));
+  }
+
+  CompactionPlan plan;
+  for (auto& pending : pending_groups) {
+    if (pending.rewrite_independently ||
+        pending.group.files.size() >= config.min_input_files) {
+      plan.groups.push_back(std::move(pending.group));
+    }
+  }
+  return plan;
+}
+
+}  // namespace iceberg

src/iceberg/compaction_planner.h

@@ -0,0 +1,78 @@
+/*
+ * 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/compaction_planner.h
+/// Plan data-file compaction without executing rewrites.
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <span>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Thresholds used to select data files for compaction.
+struct ICEBERG_EXPORT CompactionPlannerConfig {
+  int64_t target_file_size_bytes = int64_t{512} * 1024 * 1024;
+  double min_file_size_ratio = 0.75;
+  double max_file_size_ratio = 1.8;
+  size_t min_input_files = 5;
+  size_t delete_file_threshold = 2;
+  double delete_ratio_threshold = 0.3;
+};
+
+/// \brief A selected data file and its file-scoped delete pressure.
+struct ICEBERG_EXPORT CompactionFile {
+  std::shared_ptr<FileScanTask> scan_task;
+  size_t file_scoped_delete_count = 0;
+  int64_t file_scoped_delete_record_count = 0;
+};
+
+/// \brief Selected files from one partition that can be rewritten together.
+struct ICEBERG_EXPORT CompactionGroup {
+  std::optional<int32_t> partition_spec_id;
+  PartitionValues partition;
+  std::vector<CompactionFile> files;
+  int64_t data_file_size_bytes = 0;
+  int64_t file_scoped_delete_record_count = 0;
+};
+
+/// \brief Result of compaction planning.
+struct ICEBERG_EXPORT CompactionPlan {
+  std::vector<CompactionGroup> groups;
+};
+
+/// \brief Select and group scan tasks for data-file compaction.
+class ICEBERG_EXPORT CompactionPlanner {
+ public:
+  static Result<CompactionPlan> Plan(
+      std::span<const std::shared_ptr<FileScanTask>> scan_tasks,
+      const CompactionPlannerConfig& config = {});
+};
+
+}  // namespace iceberg

src/iceberg/meson.build

@@ -79,6 +79,7 @@ iceberg_sources = files(
     'catalog/memory/in_memory_catalog.cc',
     'catalog/session_catalog.cc',
     'catalog/session_context.cc',
+    'compaction_planner.cc',
     'delete_file_index.cc',
     'deletes/dv_util.cc',
     'deletes/dv_writer.cc',
@@ -319,6 +320,7 @@ install_headers(
     [
         'arrow_c_data.h',
         'catalog.h',
+        'compaction_planner.h',
         'constants.h',
         'delete_file_index.h',
         'exception.h',

src/iceberg/test/CMakeLists.txt

@@ -163,6 +163,8 @@ add_iceberg_test(puffin_test
                  puffin_json_test.cc
                  puffin_reader_writer_test.cc)
 
+add_iceberg_test(compaction_planner_test SOURCES compaction_planner_test.cc)
+
 if(ICEBERG_BUILD_BUNDLE)
   add_iceberg_test(avro_test
                    USE_BUNDLE