feat(spill): introduce external sort buffer and writer memory manager (#257)

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
+    /// works 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[];
@@ -417,10 +438,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

@@ -234,6 +234,7 @@ set(PAIMON_CORE_SRCS
     core/manifest/manifest_list.cpp
     core/manifest/partition_entry.cpp
     core/manifest/index_manifest_file_handler.cpp
+    core/memory/writer_memory_manager.cpp
     core/mergetree/compact/universal_compaction.cpp
     core/mergetree/compact/early_full_compaction.cpp
     core/mergetree/compact/aggregate/aggregate_merge_function.cpp
@@ -251,6 +252,7 @@ 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/levels.cpp
     core/mergetree/lookup_file.cpp
@@ -594,6 +596,7 @@ if(PAIMON_BUILD_TESTS)
                     core/manifest/file_entry_test.cpp
                     core/manifest/index_manifest_entry_serializer_test.cpp
                     core/manifest/index_manifest_file_handler_test.cpp
+                    core/memory/writer_memory_manager_test.cpp
                     core/mergetree/levels_test.cpp
                     core/mergetree/lookup_file_test.cpp
                     core/mergetree/lookup_levels_test.cpp
@@ -633,6 +636,7 @@ 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

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";
@@ -106,7 +111,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

@@ -327,6 +327,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) {