feat(compaction): support compaction for key table in framework (alibaba#195)

Author: lucasfang

include/paimon/defs.h

@@ -203,6 +203,43 @@ struct PAIMON_EXPORT Options {
     /// "commit.max-retries" - Maximum number of retries when commit failed. Default value is 10.
     static const char COMMIT_MAX_RETRIES[];
 
+    /// "compaction.max-size-amplification-percent" - The size amplification is defined as the
+    /// amount (in percentage) of additional storage needed to store a single byte of data in the
+    /// merge tree for changelog mode table. Default value is 200.
+    static const char COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT[];
+
+    /// "compaction.size-ratio" - Percentage flexibility while comparing sorted run size for
+    /// changelog mode table. If the candidate sorted run(s) size is 1% smaller than the next
+    /// sorted run's size, then include next sorted run into this candidate set. Default value is 1.
+    static const char COMPACTION_SIZE_RATIO[];
+
+    /// "num-sorted-run.compaction-trigger" - The sorted run number to trigger compaction. Includes
+    /// level0 files (one file one sorted run) and high-level runs (one level one sorted run).
+    /// Default value is 5.
+    static const char NUM_SORTED_RUNS_COMPACTION_TRIGGER[];
+
+    /// "num-sorted-run.stop-trigger" - The number of sorted runs that trigger the stopping of
+    /// writes, the default value is 'num-sorted-run.compaction-trigger' + 3.
+    static const char NUM_SORTED_RUNS_STOP_TRIGGER[];
+
+    /// "num-levels" - Total level number, for example, there are 3 levels, including 0,1,2 levels.
+    /// No default value.
+    static const char NUM_LEVELS[];
+
+    /// "lookup-compact" - Lookup compact mode used for lookup compaction. Default value is
+    /// LookupCompactMode::RADICAL.
+    static const char LOOKUP_COMPACT[];
+
+    /// "compaction.force-up-level-0" - If set to true, compaction strategy will always include all
+    /// level 0 files in candidates. Default value is false.
+    static const char COMPACTION_FORCE_UP_LEVEL_0[];
+
+    /// "lookup-compact.max-interval" - The max interval for a gentle mode lookup compaction to be
+    /// triggered. For every interval, a forced lookup compaction will be performed to flush L0
+    /// files to higher level. This option is only valid when lookup-compact mode is gentle. No
+    /// default value.
+    static const char LOOKUP_COMPACT_MAX_INTERVAL[];
+
     /// "sequence.field" - The field that generates the sequence number for primary key table, the
     /// sequence number determines which data is the most recent. Value use "," as delimiter.
     static const char SEQUENCE_FIELD[];
@@ -304,7 +341,6 @@ struct PAIMON_EXPORT Options {
     static const char SCAN_TAG_NAME[];
     /// "write-only" - If set to "true", compactions and snapshot expiration will be skipped. This
     /// option is used along with dedicated compact jobs. Default value is "false".
-    /// @note: This option will be ignore until compaction is supported.
     static const char WRITE_ONLY[];
     /// "compaction.min.file-num" - For file set [f_0,...,f_N], the minimum file number to trigger a
     /// compaction for append-only table. Default value is 5.

include/paimon/write_context.h

@@ -29,6 +29,7 @@
 
 namespace paimon {
 class Executor;
+class IOManager;
 class MemoryPool;
 
 /// `WriteContext` is some configuration for write operations.
@@ -44,6 +45,7 @@ class PAIMON_EXPORT WriteContext {
                  const std::vector<std::string>& write_schema,
                  const std::shared_ptr<MemoryPool>& memory_pool,
                  const std::shared_ptr<Executor>& executor,
+                 const std::shared_ptr<IOManager>& io_manager,
                  const std::shared_ptr<FileSystem>& specific_file_system,
                  const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
                  const std::map<std::string, std::string>& options);
@@ -98,6 +100,10 @@ class PAIMON_EXPORT WriteContext {
         return executor_;
     }
 
+    std::shared_ptr<IOManager> GetIOManager() const {
+        return io_manager_;
+    }
+
     std::shared_ptr<FileSystem> GetSpecificFileSystem() const {
         return specific_file_system_;
     }
@@ -113,6 +119,7 @@ class PAIMON_EXPORT WriteContext {
     std::vector<std::string> write_schema_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
+    std::shared_ptr<IOManager> io_manager_;
     std::shared_ptr<FileSystem> specific_file_system_;
     std::map<std::string, std::string> fs_scheme_to_identifier_map_;
     std::map<std::string, std::string> options_;
@@ -163,6 +170,11 @@ class PAIMON_EXPORT WriteContextBuilder {
     /// @return Reference to this builder for method chaining.
     WriteContextBuilder& WithExecutor(const std::shared_ptr<Executor>& executor);
 
+    /// Set custom IO manager for lookup and external disk spill operations.
+    /// @param io_manager The IO manager to use.
+    /// @return Reference to this builder for method chaining.
+    WriteContextBuilder& WithIOManager(const std::shared_ptr<IOManager>& io_manager);
+
     /// For postpone bucket mode in pk table, `WithWriteId()` supposed to be used.
     ///
     /// Each worker must have its own unique `write_id` within a task, which is used as the prefix

src/paimon/CMakeLists.txt

@@ -229,7 +229,9 @@ set(PAIMON_CORE_SRCS
     core/mergetree/compact/aggregate/field_sum_agg.cpp
     core/mergetree/compact/interval_partition.cpp
     core/mergetree/compact/loser_tree.cpp
+    core/mergetree/compact/merge_tree_compact_manager.cpp
     core/mergetree/compact/merge_tree_compact_rewriter.cpp
+    core/mergetree/compact/merge_tree_compact_task.cpp
     core/mergetree/compact/partial_update_merge_function.cpp
     core/mergetree/compact/sort_merge_reader_with_loser_tree.cpp
     core/mergetree/compact/sort_merge_reader_with_min_heap.cpp
@@ -510,6 +512,7 @@ if(PAIMON_BUILD_TESTS)
                     core/catalog/identifier_test.cpp
                     core/compact/compact_deletion_file_test.cpp
                     core/core_options_test.cpp
+                    core/options/lookup_strategy_test.cpp
                     core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
                     core/deletionvectors/bitmap_deletion_vector_test.cpp
                     core/deletionvectors/bucketed_dv_maintainer_test.cpp
@@ -577,6 +580,7 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/compact/universal_compaction_test.cpp
                     core/mergetree/compact/force_up_level0_compaction_test.cpp
                     core/mergetree/compact/compact_strategy_test.cpp
+                    core/mergetree/compact/merge_tree_compact_manager_test.cpp
                     core/mergetree/compact/merge_tree_compact_rewriter_test.cpp
                     core/mergetree/lookup/persist_processor_test.cpp
                     core/mergetree/drop_delete_reader_test.cpp

src/paimon/common/defs.cpp

@@ -102,4 +102,14 @@ const char Options::LOOKUP_CACHE_SPILL_COMPRESSION[] = "lookup.cache-spill-compr
 const char Options::SPILL_COMPRESSION_ZSTD_LEVEL[] = "spill-compression.zstd-level";
 const char Options::CACHE_PAGE_SIZE[] = "cache-page-size";
 const char Options::FILE_FORMAT_PER_LEVEL[] = "file.format.per.level";
+const char Options::COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT[] =
+    "compaction.max-size-amplification-percent";
+const char Options::COMPACTION_SIZE_RATIO[] = "compaction.size-ratio";
+const char Options::NUM_SORTED_RUNS_COMPACTION_TRIGGER[] = "num-sorted-run.compaction-trigger";
+const char Options::NUM_SORTED_RUNS_STOP_TRIGGER[] = "num-sorted-run.stop-trigger";
+const char Options::NUM_LEVELS[] = "num-levels";
+const char Options::COMPACTION_FORCE_UP_LEVEL_0[] = "compaction.force-up-level-0";
+const char Options::LOOKUP_COMPACT[] = "lookup-compact";
+const char Options::LOOKUP_COMPACT_MAX_INTERVAL[] = "lookup-compact.max-interval";
+
 }  // namespace paimon

src/paimon/core/append/append_only_writer.cpp

@@ -140,7 +140,7 @@ Status AppendOnlyWriter::Flush(bool wait_for_latest_compaction, bool forced_full
     }
     // add new generated files
     for (const auto& flushed_file : flushed_files) {
-        compact_manager_->AddNewFile(flushed_file);
+        PAIMON_RETURN_NOT_OK(compact_manager_->AddNewFile(flushed_file));
     }
     PAIMON_RETURN_NOT_OK(TrySyncLatestCompaction(wait_for_latest_compaction));
     PAIMON_RETURN_NOT_OK(compact_manager_->TriggerCompaction(forced_full_compaction));

src/paimon/core/append/bucketed_append_compact_manager.cpp

@@ -16,6 +16,8 @@
 
 #include "paimon/core/append/bucketed_append_compact_manager.h"
 
+#include <cassert>
+
 #include "paimon/common/executor/future.h"
 
 namespace paimon {
@@ -26,7 +28,7 @@ BucketedAppendCompactManager::BucketedAppendCompactManager(
     const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer, int32_t min_file_num,
     int64_t target_file_size, int64_t compaction_file_size, bool force_rewrite_all_files,
     CompactRewriter rewriter, const std::shared_ptr<CompactionMetrics::Reporter>& reporter,
-    const std::shared_ptr<std::atomic_bool>& cancel_flag)
+    const std::shared_ptr<CancellationController>& cancellation_controller)
     : executor_(executor),
       dv_maintainer_(dv_maintainer),
       min_file_num_(min_file_num),
@@ -39,15 +41,16 @@ BucketedAppendCompactManager::BucketedAppendCompactManager(
           [](const std::shared_ptr<DataFileMeta>& lhs, const std::shared_ptr<DataFileMeta>& rhs) {
               return lhs->min_sequence_number > rhs->min_sequence_number;
           }),
-      cancel_flag_(cancel_flag ? cancel_flag : std::make_shared<std::atomic_bool>(false)),
+      cancellation_controller_(cancellation_controller),
       logger_(Logger::GetLogger("BucketedAppendCompactManager")) {
+    assert(cancellation_controller_ != nullptr);
     for (const auto& file : restored) {
         to_compact_.push(file);
     }
 }
 
 void BucketedAppendCompactManager::CancelCompaction() {
-    cancel_flag_->store(true, std::memory_order_relaxed);
+    cancellation_controller_->Cancel();
     CompactFutureManager::CancelCompaction();
 }
 
@@ -78,7 +81,7 @@ Status BucketedAppendCompactManager::TriggerFullCompaction() {
         compacting.push_back(to_compact_.top());
         to_compact_.pop();
     }
-    cancel_flag_->store(false, std::memory_order_relaxed);
+    cancellation_controller_->Reset();
     auto compact_task = std::make_shared<FullCompactTask>(reporter_, dv_maintainer_, compacting,
                                                           compaction_file_size_,
                                                           force_rewrite_all_files_, rewriter_);
@@ -95,7 +98,7 @@ void BucketedAppendCompactManager::TriggerCompactionWithBestEffort() {
     }
     std::optional<std::vector<std::shared_ptr<DataFileMeta>>> picked = PickCompactBefore();
     if (picked) {
-        cancel_flag_->store(false, std::memory_order_relaxed);
+        cancellation_controller_->Reset();
         compacting_ = picked.value();
         auto compact_task = std::make_shared<AutoCompactTask>(reporter_, dv_maintainer_,
                                                               compacting_.value(), rewriter_);

src/paimon/core/append/bucketed_append_compact_manager.h

@@ -16,7 +16,6 @@
 
 #pragma once
 
-#include <atomic>
 #include <deque>
 #include <functional>
 #include <memory>
@@ -25,6 +24,7 @@
 #include <vector>
 
 #include "paimon/common/executor/future.h"
+#include "paimon/core/compact/cancellation_controller.h"
 #include "paimon/core/compact/compact_deletion_file.h"
 #include "paimon/core/compact/compact_future_manager.h"
 #include "paimon/core/compact/compact_task.h"
@@ -67,14 +67,13 @@ class BucketedAppendCompactManager : public CompactFutureManager {
         };
     }
 
-    BucketedAppendCompactManager(const std::shared_ptr<Executor>& executor,
-                                 const std::vector<std::shared_ptr<DataFileMeta>>& restored,
-                                 const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer,
-                                 int32_t min_file_num, int64_t target_file_size,
-                                 int64_t compaction_file_size, bool force_rewrite_all_files,
-                                 CompactRewriter rewriter,
-                                 const std::shared_ptr<CompactionMetrics::Reporter>& reporter,
-                                 const std::shared_ptr<std::atomic_bool>& cancel_flag);
+    BucketedAppendCompactManager(
+        const std::shared_ptr<Executor>& executor,
+        const std::vector<std::shared_ptr<DataFileMeta>>& restored,
+        const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer, int32_t min_file_num,
+        int64_t target_file_size, int64_t compaction_file_size, bool force_rewrite_all_files,
+        CompactRewriter rewriter, const std::shared_ptr<CompactionMetrics::Reporter>& reporter,
+        const std::shared_ptr<CancellationController>& cancellation_controller);
     ~BucketedAppendCompactManager() override = default;
 
     void CancelCompaction() override;
@@ -88,8 +87,9 @@ class BucketedAppendCompactManager : public CompactFutureManager {
         return false;
     }
 
-    void AddNewFile(const std::shared_ptr<DataFileMeta>& file) override {
+    Status AddNewFile(const std::shared_ptr<DataFileMeta>& file) override {
         to_compact_.push(file);
+        return Status::OK();
     }
 
     std::vector<std::shared_ptr<DataFileMeta>> AllFiles() const override;
@@ -199,7 +199,7 @@ class BucketedAppendCompactManager : public CompactFutureManager {
     std::shared_ptr<CompactionMetrics::Reporter> reporter_;
     std::optional<std::vector<std::shared_ptr<DataFileMeta>>> compacting_;
     DataFileMetaPriorityQueue to_compact_;
-    std::shared_ptr<std::atomic_bool> cancel_flag_;
+    std::shared_ptr<CancellationController> cancellation_controller_;
     std::unique_ptr<Logger> logger_;
 };
 

src/paimon/core/append/bucketed_append_compact_manager_test.cpp

@@ -16,7 +16,6 @@
 
 #include "paimon/core/append/bucketed_append_compact_manager.h"
 
-#include <atomic>
 #include <chrono>
 #include <future>
 #include <optional>
@@ -76,7 +75,7 @@ class BucketedAppendCompactManagerTest : public testing::Test {
             executor_, to_compact_before_pick,
             /*dv_maintainer=*/nullptr, min_file_num, target_file_size, threshold,
             /*force_rewrite_all_files=*/false, /*rewriter=*/nullptr, /*reporter=*/nullptr,
-            /*cancel_flag=*/std::make_shared<std::atomic_bool>(false));
+            /*cancellation_controller=*/std::make_shared<CancellationController>());
         auto actual = manager.PickCompactBefore();
         if (expected_present) {
             ASSERT_TRUE(actual.has_value());
@@ -267,14 +266,14 @@ TEST_F(BucketedAppendCompactManagerTest, TestPick) {
 }
 
 TEST_F(BucketedAppendCompactManagerTest, TestCancelCompactionPropagatesToRewriteLoop) {
-    auto cancel_flag = std::make_shared<std::atomic_bool>(false);
+    auto cancellation_controller = std::make_shared<CancellationController>();
     auto exit_signal = std::make_shared<std::promise<void>>();
     auto exit_future = exit_signal->get_future();
 
-    auto rewriter = [cancel_flag,
+    auto rewriter = [cancellation_controller,
                      exit_signal](const std::vector<std::shared_ptr<DataFileMeta>>& to_compact)
         -> Result<std::vector<std::shared_ptr<DataFileMeta>>> {
-        while (!cancel_flag->load(std::memory_order_relaxed)) {
+        while (!cancellation_controller->IsCancelled()) {
             std::this_thread::sleep_for(std::chrono::milliseconds(1));
         }
         exit_signal->set_value();
@@ -288,7 +287,7 @@ TEST_F(BucketedAppendCompactManagerTest, TestCancelCompactionPropagatesToRewrite
         /*target_file_size=*/1024,
         /*compaction_file_size=*/700,
         /*force_rewrite_all_files=*/false, rewriter,
-        /*reporter=*/nullptr, cancel_flag);
+        /*reporter=*/nullptr, cancellation_controller);
 
     ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
     manager.CancelCompaction();
@@ -297,7 +296,8 @@ TEST_F(BucketedAppendCompactManagerTest, TestCancelCompactionPropagatesToRewrite
 }
 
 TEST_F(BucketedAppendCompactManagerTest, TestTriggerCompactionResetsCancelFlag) {
-    auto cancel_flag = std::make_shared<std::atomic_bool>(true);
+    auto cancellation_controller = std::make_shared<CancellationController>();
+    cancellation_controller->Cancel();
     auto rewriter = [](const std::vector<std::shared_ptr<DataFileMeta>>& to_compact)
         -> Result<std::vector<std::shared_ptr<DataFileMeta>>> { return to_compact; };
 
@@ -308,10 +308,10 @@ TEST_F(BucketedAppendCompactManagerTest, TestTriggerCompactionResetsCancelFlag)
         /*target_file_size=*/1024,
         /*compaction_file_size=*/700,
         /*force_rewrite_all_files=*/false, rewriter,
-        /*reporter=*/nullptr, cancel_flag);
+        /*reporter=*/nullptr, cancellation_controller);
 
     ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
-    EXPECT_FALSE(cancel_flag->load(std::memory_order_relaxed));
+    EXPECT_FALSE(cancellation_controller->IsCancelled());
 }
 
 }  // namespace paimon::test

src/paimon/core/compact/cancellation_controller.h

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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
+
+#include <atomic>
+
+namespace paimon {
+
+class CancellationController {
+ public:
+    void Cancel() {
+        cancelled_.store(true, std::memory_order_relaxed);
+    }
+
+    void Reset() {
+        cancelled_.store(false, std::memory_order_relaxed);
+    }
+
+    bool IsCancelled() const {
+        return cancelled_.load(std::memory_order_relaxed);
+    }
+
+ private:
+    std::atomic_bool cancelled_{false};
+};
+
+}  // namespace paimon