fix(compaction): make sure that only one task is running at a time, refactor compaction manager creation and add test (alibaba#201)

Author: lucasfang

include/paimon/defs.h

@@ -226,6 +226,10 @@ struct PAIMON_EXPORT Options {
     /// No default value.
     static const char NUM_LEVELS[];
 
+    /// "lookup-wait" - When need to lookup, commit will wait for compaction by lookup. Default
+    /// value is "true".
+    static const char LOOKUP_WAIT[];
+
     /// "lookup-compact" - Lookup compact mode used for lookup compaction. Default value is
     /// LookupCompactMode::RADICAL.
     static const char LOOKUP_COMPACT[];

src/paimon/CMakeLists.txt

@@ -230,6 +230,7 @@ set(PAIMON_CORE_SRCS
     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_manager_factory.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
@@ -513,6 +514,7 @@ if(PAIMON_BUILD_TESTS)
                     core/catalog/catalog_test.cpp
                     core/catalog/identifier_test.cpp
                     core/compact/compact_deletion_file_test.cpp
+                    core/compact/noop_compact_manager_test.cpp
                     core/core_options_test.cpp
                     core/options/lookup_strategy_test.cpp
                     core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
@@ -522,6 +524,7 @@ if(PAIMON_BUILD_TESTS)
                     core/deletionvectors/deletion_vector_test.cpp
                     core/deletionvectors/deletion_vector_index_file_writer_test.cpp
                     core/deletionvectors/deletion_vectors_index_file_test.cpp
+                    core/disk/io_manager_test.cpp
                     core/index/index_in_data_file_dir_path_factory_test.cpp
                     core/index/deletion_vector_meta_test.cpp
                     core/index/index_file_meta_serializer_test.cpp
@@ -582,7 +585,9 @@ 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/file_rewrite_compact_task_test.cpp
                     core/mergetree/compact/merge_tree_compact_manager_test.cpp
+                    core/mergetree/compact/merge_tree_compact_manager_factory_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

@@ -109,6 +109,7 @@ const char Options::NUM_SORTED_RUNS_COMPACTION_TRIGGER[] = "num-sorted-run.compa
 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_WAIT[] = "lookup-wait";
 const char Options::LOOKUP_COMPACT[] = "lookup-compact";
 const char Options::LOOKUP_COMPACT_MAX_INTERVAL[] = "lookup-compact.max-interval";
 

src/paimon/core/append/append_only_writer.cpp

@@ -248,8 +248,9 @@ Status AppendOnlyWriter::Sync() {
 }
 
 Status AppendOnlyWriter::Close() {
-    // cancel compaction so that it does not block job cancelling
-    compact_manager_->CancelCompaction();
+    // Request cancellation and wait for running compaction to exit.
+    // This avoids reusing cancellation state while an old task is still running.
+    compact_manager_->CancelAndWaitCompaction();
     PAIMON_RETURN_NOT_OK(Sync());
 
     PAIMON_RETURN_NOT_OK(compact_manager_->Close());