@@ -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
 
@@ -0,0 +1,279 @@
+/*
+ * 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 <cmath>
+#include <compare>
+#include <limits>
+#include <map>
+#include <string>
+#include <string_view>
+#include <tuple>
+#include <utility>
+
+#include "iceberg/expression/literal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/table_scan.h"
+#include "iceberg/util/content_file_util.h"
+#include "iceberg/util/data_file_set.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+struct CanonicalPartitionKey {
+  int32_t spec_id;
+  std::string values;
+
+  auto operator<=>(const CanonicalPartitionKey&) const = default;
+};
+
+struct Candidate {
+  CompactionFile file;
+  bool has_delete_pressure;
+};
+
+struct PartitionCandidates {
+  int32_t spec_id;
+  PartitionValues partition;
+  std::vector<Candidate> files;
+};
+
+Status ValidateConfig(const CompactionPlannerConfig& config) {
+  ICEBERG_PRECHECK(config.target_file_size_bytes > 0,
+                   "target_file_size_bytes must be greater than zero");
+  ICEBERG_PRECHECK(std::isfinite(config.min_file_size_ratio) &&
+                       config.min_file_size_ratio >= 0 && config.min_file_size_ratio <= 1,
+                   "min_file_size_ratio must be finite and in [0, 1]");
+  ICEBERG_PRECHECK(
+      std::isfinite(config.max_file_size_ratio) && config.max_file_size_ratio >= 1,
+      "max_file_size_ratio must be finite and at least 1");
+  ICEBERG_PRECHECK(config.min_input_files > 0,
+                   "min_input_files must be greater than zero");
+  ICEBERG_PRECHECK(config.delete_file_threshold >= 0,
+                   "delete_file_threshold must not be negative");
+  ICEBERG_PRECHECK(std::isfinite(config.delete_ratio_threshold) &&
+                       config.delete_ratio_threshold >= 0 &&
+                       config.delete_ratio_threshold <= 1,
+                   "delete_ratio_threshold must be finite and in [0, 1]");
+  return {};
+}
+
+Result<int64_t> CheckedAddNonNegative(int64_t lhs, int64_t rhs,
+                                      std::string_view description) {
+  ICEBERG_PRECHECK(lhs >= 0 && rhs >= 0, "{} must not be negative", description);
+  ICEBERG_PRECHECK(lhs <= std::numeric_limits<int64_t>::max() - rhs, "{} overflow",
+                   description);
+  return lhs + rhs;
+}
+
+void AppendFramed(std::string& output, std::string_view value) {
+  output.append(std::to_string(value.size()));
+  output.push_back(':');
+  output.append(value);
+}
+
+Result<CanonicalPartitionKey> MakePartitionKey(const DataFile& data_file) {
+  ICEBERG_PRECHECK(data_file.partition_spec_id.has_value(),
+                   "Data file '{}' is missing partition_spec_id", data_file.file_path);
+
+  std::string encoded;
+  encoded.append(std::to_string(data_file.partition.num_fields()));
+  encoded.push_back(':');
+  for (const auto& literal : data_file.partition.values()) {
+    AppendFramed(encoded, literal.type()->ToString());
+    if (literal.IsNull()) {
+      encoded.push_back('N');
+      continue;
+    }
+
+    encoded.push_back('V');
+    ICEBERG_ASSIGN_OR_RAISE(auto bytes, literal.Serialize());
+    std::string_view serialized;
+    if (!bytes.empty()) {
+      serialized =
+          std::string_view(reinterpret_cast<const char*>(bytes.data()), bytes.size());
+    }
+    AppendFramed(encoded, serialized);
+  }
+  return CanonicalPartitionKey{.spec_id = *data_file.partition_spec_id,
+                               .values = std::move(encoded)};
+}
+
+Result<CompactionFile> BuildCompactionFile(
+    const std::shared_ptr<FileScanTask>& scan_task) {
+  CompactionFile result{.scan_task = scan_task};
+  const auto& data_file = scan_task->data_file();
+  ICEBERG_PRECHECK(data_file->record_count >= 0,
+                   "Data file '{}' has negative record count", data_file->file_path);
+  ICEBERG_PRECHECK(data_file->file_size_in_bytes >= 0,
+                   "Data file '{}' has negative file size", data_file->file_path);
+
+  DeleteFileSet applicable_deletes;
+  for (const auto& delete_file : scan_task->delete_files()) {
+    ICEBERG_PRECHECK(delete_file != nullptr, "Data file '{}' has a null delete file",
+                     data_file->file_path);
+    if (delete_file->content != DataFile::Content::kPositionDeletes) {
+      continue;
+    }
+
+    ICEBERG_ASSIGN_OR_RAISE(auto referenced_file,
+                            ContentFileUtil::ReferencedDataFile(*delete_file));
+    if (!referenced_file.has_value() || *referenced_file != data_file->file_path) {
+      continue;
+    }
+
+    if (!applicable_deletes.insert(delete_file).second) {
+      continue;
+    }
+
+    ICEBERG_PRECHECK(delete_file->record_count >= 0,
+                     "Delete file '{}' has negative record count",
+                     delete_file->file_path);
+    if (delete_file->IsDeletionVector()) {
+      ICEBERG_PRECHECK(
+          delete_file->record_count <= data_file->record_count,
+          "Deletion vector '{}' cardinality {} exceeds data file '{}' record count {}",
+          delete_file->file_path, delete_file->record_count, data_file->file_path,
+          data_file->record_count);
+    }
+
+    ICEBERG_ASSIGN_OR_RAISE(result.file_scoped_delete_count,
+                            CheckedAddNonNegative(result.file_scoped_delete_count, 1,
+                                                  "File-scoped delete-file count"));
+    ICEBERG_ASSIGN_OR_RAISE(result.file_scoped_delete_record_count,
+                            CheckedAddNonNegative(result.file_scoped_delete_record_count,
+                                                  delete_file->record_count,
+                                                  "File-scoped delete-record count"));
+  }
+
+  result.file_scoped_delete_record_count =
+      std::min(result.file_scoped_delete_record_count, data_file->record_count);
+  return result;
+}
+
+bool CandidateLess(const Candidate& lhs, const Candidate& rhs) {
+  const auto& lhs_data = *lhs.file.scan_task->data_file();
+  const auto& rhs_data = *rhs.file.scan_task->data_file();
+  return std::tie(lhs_data.file_path, lhs_data.file_size_in_bytes, lhs_data.record_count,
+                  lhs.file.file_scoped_delete_count,
+                  lhs.file.file_scoped_delete_record_count) <
+         std::tie(rhs_data.file_path, rhs_data.file_size_in_bytes, rhs_data.record_count,
+                  rhs.file.file_scoped_delete_count,
+                  rhs.file.file_scoped_delete_record_count);
+}
+
+Result<void> AddToGroup(CompactionGroup& group, Candidate candidate,
+                        bool& has_delete_pressure) {
+  const auto& data_file = *candidate.file.scan_task->data_file();
+  ICEBERG_ASSIGN_OR_RAISE(
+      group.data_file_size_bytes,
+      CheckedAddNonNegative(group.data_file_size_bytes, data_file.file_size_in_bytes,
+                            "Compaction group data-file size"));
+  ICEBERG_ASSIGN_OR_RAISE(
+      group.file_scoped_delete_record_count,
+      CheckedAddNonNegative(group.file_scoped_delete_record_count,
+                            candidate.file.file_scoped_delete_record_count,
+                            "Compaction group delete-record count"));
+  has_delete_pressure |= candidate.has_delete_pressure;
+  group.files.push_back(std::move(candidate.file));
+  return {};
+}
+
+void FinishGroup(CompactionGroup& group, bool has_delete_pressure, size_t min_input_files,
+                 CompactionPlan& plan) {
+  if (!group.files.empty() &&
+      (has_delete_pressure || group.files.size() >= min_input_files)) {
+    plan.groups.push_back(std::move(group));
+  }
+}
+
+}  // namespace
+
+Result<CompactionPlan> CompactionPlanner::Plan(
+    int64_t source_snapshot_id, std::span<const std::shared_ptr<FileScanTask>> scan_tasks,
+    const CompactionPlannerConfig& config) {
+  ICEBERG_RETURN_UNEXPECTED(ValidateConfig(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::map<CanonicalPartitionKey, PartitionCandidates> partitions;
+
+  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 partition_key, MakePartitionKey(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 =
+        config.delete_file_threshold > 0 &&
+        file.file_scoped_delete_count >= config.delete_file_threshold;
+    const bool has_high_delete_ratio =
+        config.delete_ratio_threshold > 0 && data_file.record_count > 0 &&
+        static_cast<double>(file.file_scoped_delete_record_count) /
+                data_file.record_count >=
+            config.delete_ratio_threshold;
+    const bool has_delete_pressure = has_many_delete_files || has_high_delete_ratio;
+    const bool should_compact =
+        is_oversized ? has_delete_pressure : is_small || has_delete_pressure;
+    if (!should_compact) {
+      continue;
+    }
+
+    auto partition =
+        partitions
+            .try_emplace(std::move(partition_key),
+                         PartitionCandidates{.spec_id = *data_file.partition_spec_id,
+                                             .partition = data_file.partition})
+            .first;
+    partition->second.files.push_back(
+        Candidate{.file = std::move(file), .has_delete_pressure = has_delete_pressure});
+  }
+
+  CompactionPlan plan{.source_snapshot_id = source_snapshot_id};
+  for (auto& [_, partition] : partitions) {
+    std::ranges::sort(partition.files, CandidateLess);
+
+    CompactionGroup group{.partition_spec_id = partition.spec_id,
+                          .partition = partition.partition};
+    bool has_delete_pressure = false;
+    for (auto& candidate : partition.files) {
+      const int64_t file_size = candidate.file.scan_task->data_file()->file_size_in_bytes;
+      if (!group.files.empty() &&
+          (group.data_file_size_bytes > config.target_file_size_bytes ||
+           file_size > config.target_file_size_bytes - group.data_file_size_bytes)) {
+        FinishGroup(group, has_delete_pressure, config.min_input_files, plan);
+        group = CompactionGroup{.partition_spec_id = partition.spec_id,
+                                .partition = partition.partition};
+        has_delete_pressure = false;
+      }
+      ICEBERG_RETURN_UNEXPECTED(
+          AddToGroup(group, std::move(candidate), has_delete_pressure));
+    }
+    FinishGroup(group, has_delete_pressure, config.min_input_files, plan);
+  }
+  return plan;
+}
+
+}  // namespace iceberg