feat: add spill-to-disk write buffer path

Author: zjw1111

.gitignore

@@ -16,6 +16,7 @@
 build
 build-release
 build-debug
+build-test
 output
 
 # IDE settings

include/paimon/defs.h

@@ -178,6 +178,23 @@ struct PAIMON_EXPORT Options {
     /// on-disk file. The default value is 256 mb
     static const char WRITE_BUFFER_SIZE[];
 
+    /// "write-buffer-spillable" - Whether the write buffer can be spillable. Default value is true.
+    static const char WRITE_BUFFER_SPILLABLE[];
+
+    /// "write-buffer-spill.max-disk-size" - The max disk to use for write buffer spill. This only
+    /// work when the write buffer spill is enabled. Default value is unlimited.
+    static const char WRITE_BUFFER_SPILL_MAX_DISK_SIZE[];
+
+    /// "local-sort.max-num-file-handles" - The maximal fan-in for external merge sort. It limits
+    /// the number of file handles. If it is too small, may cause intermediate merging. But if it is
+    /// too large, it will cause too many files opened at the same time, consume memory and lead to
+    /// random reading. Default value is 128.
+    static const char LOCAL_SORT_MAX_NUM_FILE_HANDLES[];
+
+    /// "spill-compression" - Compression for spill.
+    /// Default value is zstd.
+    static const char SPILL_COMPRESSION[];
+
     /// "snapshot.num-retained.min" - The minimum number of completed snapshots to retain. Should be
     /// greater than or equal to 1. Default value is 10
     static const char SNAPSHOT_NUM_RETAINED_MIN[];

src/paimon/CMakeLists.txt

@@ -213,6 +213,7 @@ set(PAIMON_CORE_SRCS
     core/io/key_value_data_file_record_reader.cpp
     core/io/key_value_data_file_writer.cpp
     core/io/key_value_in_memory_record_reader.cpp
+    core/io/merged_key_value_record_reader.cpp
     core/io/key_value_meta_projection_consumer.cpp
     core/io/key_value_projection_consumer.cpp
     core/io/key_value_projection_reader.cpp
@@ -241,12 +242,16 @@ set(PAIMON_CORE_SRCS
     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/raw_sort_merge_reader_with_min_heap.cpp
     core/mergetree/compact/sort_merge_reader_with_loser_tree.cpp
     core/mergetree/compact/sort_merge_reader_with_min_heap.cpp
     core/mergetree/compact/lookup_merge_tree_compact_rewriter.cpp
     core/mergetree/compact/changelog_merge_tree_rewriter.cpp
     core/mergetree/merge_tree_writer.cpp
+    core/mergetree/binary_in_memory_sort_buffer.cpp
+    core/mergetree/binary_external_sort_buffer.cpp
     core/mergetree/write_buffer.cpp
+    core/mergetree/writer_memory_manager.cpp
     core/mergetree/levels.cpp
     core/mergetree/lookup_file.cpp
     core/mergetree/lookup_levels.cpp
@@ -564,6 +569,7 @@ if(PAIMON_BUILD_TESTS)
                     core/io/key_value_data_file_record_reader_test.cpp
                     core/io/key_value_projection_reader_test.cpp
                     core/io/key_value_in_memory_record_reader_test.cpp
+                    core/io/merged_key_value_record_reader_test.cpp
                     core/io/complete_row_tracking_fields_reader_test.cpp
                     core/io/data_file_meta_test.cpp
                     core/io/file_index_evaluator_test.cpp
@@ -624,6 +630,7 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/sorted_run_test.cpp
                     core/mergetree/spill_channel_manager_test.cpp
                     core/mergetree/spill_reader_writer_test.cpp
+                    core/mergetree/writer_memory_manager_test.cpp
                     core/migrate/file_meta_utils_test.cpp
                     core/operation/metrics/compaction_metrics_test.cpp
                     core/operation/data_evolution_file_store_scan_test.cpp

src/paimon/common/defs.cpp

@@ -50,6 +50,10 @@ const char Options::SCAN_MODE[] = "scan.mode";
 const char Options::READ_BATCH_SIZE[] = "read.batch-size";
 const char Options::WRITE_BATCH_SIZE[] = "write.batch-size";
 const char Options::WRITE_BUFFER_SIZE[] = "write-buffer-size";
+const char Options::WRITE_BUFFER_SPILLABLE[] = "write-buffer-spillable";
+const char Options::WRITE_BUFFER_SPILL_MAX_DISK_SIZE[] = "write-buffer-spill.max-disk-size";
+const char Options::LOCAL_SORT_MAX_NUM_FILE_HANDLES[] = "local-sort.max-num-file-handles";
+const char Options::SPILL_COMPRESSION[] = "spill-compression";
 const char Options::SNAPSHOT_NUM_RETAINED_MIN[] = "snapshot.num-retained.min";
 const char Options::SNAPSHOT_NUM_RETAINED_MAX[] = "snapshot.num-retained.max";
 const char Options::SNAPSHOT_TIME_RETAINED[] = "snapshot.time-retained";

src/paimon/common/utils/string_utils_test.cpp

@@ -326,6 +326,12 @@ TEST_F(StringUtilsTest, TestSplit) {
             StringUtils::Split("key1=value1//key3=value3", std::string("/"), std::string("="));
         ASSERT_EQ(expect, result);
     }
+    {
+        std::vector<std::vector<std::string>> expect = {};
+        std::vector<std::vector<std::string>> result =
+            StringUtils::Split("", std::string("/"), std::string("="));
+        ASSERT_EQ(expect, result);
+    }
 }
 
 TEST_F(StringUtilsTest, TestStringToValueSimple) {