feat(mergetree): support spill-to-disk write buffer and writer memory manager

Introduce ExternalSortBuffer that wraps an InMemorySortBuffer and spills
sorted runs to disk via IOManager channels when the in-memory budget is
reached. Spilled runs are merged back through the existing sort-merge
reader path on flush.

Also add WriterMemoryManager to coordinate write-buffer memory across
multiple BatchWriters by picking the largest writer to flush when the
global budget is exceeded.

New CoreOptions:
  - write-buffer-spillable
  - write-buffer-spill.max-disk-size
  - local-sort.max-num-file-handles
  - spill-compression
  - spill-compression.zstd-level (moved next to other spill options)

Tests:
  - sort_buffer_test: external sort buffer spill & merge behaviour
  - writer_memory_manager_test: memory accounting and shrink-to-limit
  - extend sort_merge_reader_test / core_options_test / data_generator_test

Generated-by: Aone Copilot (claude-4.7-opus)

Author: zjw1111

include/paimon/defs.h

@@ -178,6 +178,27 @@ 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[];
+
+    /// "spill-compression.zstd-level" - Default spill compression zstd level. For higher
+    /// compression rates, it can be configured to 9, but the read and write speed will
+    /// significantly decrease. Default value is 1.
+    static const char SPILL_COMPRESSION_ZSTD_LEVEL[];
+
     /// "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[];
@@ -405,10 +426,6 @@ struct PAIMON_EXPORT Options {
     /// lz4 are supported. Default value is zstd.
     /// Noted that java paimon also supports lzo which paimon-cpp does not support for now.
     static const char LOOKUP_CACHE_SPILL_COMPRESSION[];
-    /// "spill-compression.zstd-level" - Default spill compression zstd level. For higher
-    /// compression rates, it can be configured to 9, but the read and write speed will
-    /// significantly decrease. Default value is 1.
-    static const char SPILL_COMPRESSION_ZSTD_LEVEL[];
     /// "cache-page-size" - Memory page size for caching. Default value is 64 kb.
     static const char CACHE_PAGE_SIZE[];
     /// "file.format.per.level" - Define different file format for different level, you can add the

src/paimon/CMakeLists.txt

@@ -251,7 +251,9 @@ set(PAIMON_CORE_SRCS
     core/mergetree/compact/changelog_merge_tree_rewriter.cpp
     core/mergetree/merge_tree_writer.cpp
     core/mergetree/in_memory_sort_buffer.cpp
+    core/mergetree/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
@@ -633,9 +635,11 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/drop_delete_reader_test.cpp
                     core/mergetree/merge_tree_writer_test.cpp
                     core/mergetree/write_buffer_test.cpp
+                    core/mergetree/sort_buffer_test.cpp
                     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,11 @@ 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::SPILL_COMPRESSION_ZSTD_LEVEL[] = "spill-compression.zstd-level";
 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";
@@ -104,7 +109,6 @@ const char Options::LOOKUP_CACHE_BLOOM_FILTER_FPP[] = "lookup.cache.bloom.filter
 const char Options::LOOKUP_REMOTE_FILE_ENABLED[] = "lookup.remote-file.enabled";
 const char Options::LOOKUP_REMOTE_LEVEL_THRESHOLD[] = "lookup.remote-file.level-threshold";
 const char Options::LOOKUP_CACHE_SPILL_COMPRESSION[] = "lookup.cache-spill-compression";
-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::FILE_COMPRESSION_PER_LEVEL[] = "file.compression.per.level";

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) {