refactor: rename LeveledMerger to SpillFileMerger and improve code clarity

- Rename class LeveledMerger → SpillFileMerger, files leveled_merger → spill_file_merger
- Rename Compact/Compaction methods to Merge (RunMergeIfNeeded, RunFinalMergeIfNeeded, etc.)
- Move kMinFanIn validation from CoreOptions to ExternalSortBuffer::Create
- Improve variable names (l→level_idx, a/b→lhs/rhs, n→files_to_merge, f→file)
- Convert 6 spill tests from TEST_P to TEST_F (no parameterization needed)
- Fix 4 misused TEST_P in btree_global_index and file_system tests
- Make test assertions exact instead of range-based

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: zjw1111

src/paimon/CMakeLists.txt

@@ -253,7 +253,7 @@ set(PAIMON_CORE_SRCS
     core/mergetree/merge_tree_writer.cpp
     core/mergetree/in_memory_sort_buffer.cpp
     core/mergetree/external_sort_buffer.cpp
-    core/mergetree/leveled_merger.cpp
+    core/mergetree/spill_file_merger.cpp
     core/mergetree/write_buffer.cpp
     core/mergetree/levels.cpp
     core/mergetree/lookup_file.cpp
@@ -649,7 +649,7 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/merge_tree_writer_test.cpp
                     core/mergetree/write_buffer_test.cpp
                     core/mergetree/sort_buffer_test.cpp
-                    core/mergetree/leveled_merger_test.cpp
+                    core/mergetree/spill_file_merger_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/fs/file_system_test.cpp

@@ -173,7 +173,7 @@ class FileSystemTest : public ::testing::Test, public ::testing::WithParamInterf
     std::string test_root_;
 };
 
-TEST_P(FileSystemTest, TestNoneFileSystemFactory) {
+TEST(FileSystemStaticTest, TestNoneFileSystemFactory) {
     std::map<std::string, std::string> fs_options;
     Result<std::unique_ptr<FileSystem>> fs =
         FileSystemFactory::Get("not exist", "/tmp", fs_options);
@@ -516,7 +516,7 @@ TEST_P(FileSystemTest, TestReadAndWriteAndAtomicStoreFile) {
     ASSERT_EQ(read_content, new_content);
 }
 
-TEST_P(FileSystemTest, TestIsObjectStore) {
+TEST(FileSystemStaticTest, TestIsObjectStore) {
     ASSERT_OK_AND_ASSIGN(bool is_object_store, FileSystem::IsObjectStore("file:///tmp/test.txt"));
     ASSERT_FALSE(is_object_store);
     ASSERT_OK_AND_ASSIGN(is_object_store, FileSystem::IsObjectStore("/tmp/test.txt"));

src/paimon/common/global_index/btree/btree_global_index_integration_test.cpp

@@ -1491,7 +1491,7 @@ TEST_P(BTreeGlobalIndexIntegrationTest, WriteAndReadAllNull) {
     }
 }
 
-TEST_P(BTreeGlobalIndexIntegrationTest, WriteAndReadLargeDataWithSmallBlocks) {
+TEST_F(BTreeGlobalIndexIntegrationTest, WriteAndReadLargeDataWithSmallBlocks) {
     // Use very small block size and cache size to force multiple block evictions
     auto file_writer = std::make_shared<FakeGlobalIndexFileWriter>(fs_, base_path_);
     auto field = arrow::field("int_field", arrow::int32());
@@ -1710,7 +1710,7 @@ TEST_P(BTreeGlobalIndexIntegrationTest, CreateReaderWithMultiFieldSchema) {
                         "supposed to have single field");
 }
 
-TEST_P(BTreeGlobalIndexIntegrationTest, CreateWriterWithMissingField) {
+TEST_F(BTreeGlobalIndexIntegrationTest, CreateWriterWithMissingField) {
     auto file_writer = std::make_shared<FakeGlobalIndexFileWriter>(fs_, base_path_);
     auto type = arrow::struct_({arrow::field("existing_field", arrow::int32())});
     auto struct_type = std::dynamic_pointer_cast<arrow::StructType>(type);

src/paimon/core/core_options.cpp

@@ -498,11 +498,6 @@ struct CoreOptions::Impl {
         // Parse local-sort.max-num-file-handles - spill file handle cap for local merge
         PAIMON_RETURN_NOT_OK(parser.Parse(Options::LOCAL_SORT_MAX_NUM_FILE_HANDLES,
                                           &local_sort_max_num_file_handles));
-        if (local_sort_max_num_file_handles < kLocalSortFileHandlesMinimalLimit) {
-            return Status::Invalid(fmt::format(
-                "invalid '{}': {}, must be at least {}", Options::LOCAL_SORT_MAX_NUM_FILE_HANDLES,
-                local_sort_max_num_file_handles, kLocalSortFileHandlesMinimalLimit));
-        }
         // Parse spill-compression - compression codec for spill files
         PAIMON_RETURN_NOT_OK(
             parser.Parse(Options::SPILL_COMPRESSION, &spill_compress_options.compress));

src/paimon/core/core_options.h

@@ -181,8 +181,6 @@ class PAIMON_EXPORT CoreOptions {
 
     const std::map<std::string, std::string>& ToMap() const;
 
-    static constexpr int32_t kLocalSortFileHandlesMinimalLimit = 2;
-
  private:
     std::optional<std::string> GetDataFileExternalPaths() const;
     std::optional<std::string> GetGlobalIndexExternalPath() const;

src/paimon/core/mergetree/external_sort_buffer.cpp

@@ -48,6 +48,11 @@ Result<std::unique_ptr<ExternalSortBuffer>> ExternalSortBuffer::Create(
     const std::shared_ptr<FieldsComparator>& user_defined_seq_comparator,
     const CoreOptions& options, const std::shared_ptr<IOManager>& io_manager,
     const std::shared_ptr<MemoryPool>& pool) {
+    if (options.GetLocalSortMaxNumFileHandles() < kSpillMinFanIn) {
+        return Status::Invalid(fmt::format(
+            "invalid '{}': {}, must be at least {}", Options::LOCAL_SORT_MAX_NUM_FILE_HANDLES,
+            options.GetLocalSortMaxNumFileHandles(), kSpillMinFanIn));
+    }
     arrow::FieldVector key_fields;
     key_fields.reserve(trimmed_primary_keys.size());
     for (const auto& primary_key : trimmed_primary_keys) {
@@ -84,7 +89,7 @@ ExternalSortBuffer::ExternalSortBuffer(
       max_fan_in_(options.GetLocalSortMaxNumFileHandles()),
       spill_channel_manager_(
           std::make_shared<SpillChannelManager>(options_.GetFileSystem(), max_fan_in_)),
-      leveled_merger_(std::make_unique<LeveledMerger>(max_fan_in_)),
+      spill_merger_(std::make_unique<SpillFileMerger>(max_fan_in_)),
       spill_channel_enumerator_(spill_channel_enumerator),
       actual_max_fan_in_(max_fan_in_),
       spill_batch_size_(options_.GetWriteBatchSize()) {}
@@ -102,7 +107,7 @@ void ExternalSortBuffer::DoClear() {
 
     spill_channel_manager_->Reset();
     total_spill_disk_bytes_ = 0;
-    leveled_merger_->Clear();
+    spill_merger_->Clear();
 }
 
 void ExternalSortBuffer::Clear() {
@@ -113,7 +118,7 @@ uint64_t ExternalSortBuffer::GetMemorySize() const {
     return in_memory_buffer_->GetMemorySize();
 }
 
-void ExternalSortBuffer::EstimateSpillParameters() {
+void ExternalSortBuffer::UpdateSpillParameters() {
     int64_t estimated_row_size = in_memory_buffer_->GetEstimateMemoryUseForEachRow();
     if (estimated_row_size <= 0) {
         return;
@@ -128,22 +133,21 @@ void ExternalSortBuffer::EstimateSpillParameters() {
     spill_batch_size_ = std::clamp(spill_batch_size_, min_batch_size, max_batch_size);
 
     actual_max_fan_in_ = merge_budget / (spill_batch_size_ * estimated_row_size);
-    actual_max_fan_in_ =
-        std::clamp(actual_max_fan_in_, CoreOptions::kLocalSortFileHandlesMinimalLimit, max_fan_in_);
+    actual_max_fan_in_ = std::clamp(actual_max_fan_in_, kSpillMinFanIn, max_fan_in_);
 
     // Re-derive spill_batch_size_ from the clamped actual_max_fan_in_ to stay within merge_budget.
     spill_batch_size_ = merge_budget / (actual_max_fan_in_ * estimated_row_size);
     spill_batch_size_ = std::clamp(spill_batch_size_, 1, max_batch_size);
 
-    leveled_merger_->SetMaxFanIn(actual_max_fan_in_);
+    spill_merger_->SetMaxFanIn(actual_max_fan_in_);
 }
 
 Result<bool> ExternalSortBuffer::FlushMemory() {
     if (!in_memory_buffer_->HasData()) {
         return true;
     }
 
-    EstimateSpillParameters();
+    UpdateSpillParameters();
     PAIMON_ASSIGN_OR_RAISE(std::vector<std::unique_ptr<KeyValueRecordReader>> memory_buffer_readers,
                            in_memory_buffer_->CreateReaders());
     PAIMON_RETURN_NOT_OK(SpillMemoryBuffer(std::move(memory_buffer_readers)));
@@ -168,9 +172,9 @@ Result<std::vector<std::unique_ptr<KeyValueRecordReader>>> ExternalSortBuffer::C
 
     int32_t max_spill_files = actual_max_fan_in_ - 1;
     PAIMON_RETURN_NOT_OK(
-        leveled_merger_->RunFinalCleanupIfNeeded(max_spill_files, CreateMergeFn()));
+        spill_merger_->RunFinalMergeIfNeeded(max_spill_files, CreateSpillFileMergeFn()));
     PAIMON_ASSIGN_OR_RAISE(std::vector<std::unique_ptr<KeyValueRecordReader>> readers,
-                           CreateSpillReaders(leveled_merger_->GetAllFiles()));
+                           CreateSpillReaders(spill_merger_->GetAllFiles()));
     readers.insert(readers.end(), std::make_move_iterator(memory_readers.begin()),
                    std::make_move_iterator(memory_readers.end()));
     return readers;
@@ -242,11 +246,11 @@ Status ExternalSortBuffer::SpillMemoryBuffer(
     PAIMON_ASSIGN_OR_RAISE(FileChannelInfo file_info,
                            SpillToDisk(std::move(readers), spill_batch_size_));
     total_spill_disk_bytes_ += file_info.file_size;
-    leveled_merger_->AddFile(file_info);
-    return leveled_merger_->RunCompactionIfNeeded(CreateMergeFn());
+    spill_merger_->AddFile(file_info);
+    return spill_merger_->RunMergeIfNeeded(CreateSpillFileMergeFn());
 }
 
-LeveledMerger::MergeFn ExternalSortBuffer::CreateMergeFn() {
+SpillFileMerger::MergeFn ExternalSortBuffer::CreateSpillFileMergeFn() {
     return [this](const std::vector<FileChannelInfo>& files) -> Result<FileChannelInfo> {
         return MergeAndReplaceFiles(files);
     };

src/paimon/core/mergetree/external_sort_buffer.h

@@ -25,8 +25,8 @@
 #include "paimon/core/core_options.h"
 #include "paimon/core/disk/file_io_channel.h"
 #include "paimon/core/mergetree/in_memory_sort_buffer.h"
-#include "paimon/core/mergetree/leveled_merger.h"
 #include "paimon/core/mergetree/sort_buffer.h"
+#include "paimon/core/mergetree/spill_file_merger.h"
 #include "paimon/record_batch.h"
 #include "paimon/result.h"
 #include "paimon/status.h"
@@ -64,16 +64,17 @@ class ExternalSortBuffer : public SortBuffer {
     bool HasData() const override;
 
  private:
+    static constexpr int32_t kSpillMinFanIn = 2;
     static constexpr int32_t kSpillMinBatchSize = 256;
 
     void DoClear();
-    void EstimateSpillParameters();
+    void UpdateSpillParameters();
     bool HasSpilledData() const;
     Result<std::vector<std::unique_ptr<KeyValueRecordReader>>> CreateSpillReaders(
         const std::vector<FileChannelInfo>& files) const;
     Result<FileChannelInfo> SpillToDisk(
         std::vector<std::unique_ptr<KeyValueRecordReader>>&& readers, int32_t write_batch_size);
-    LeveledMerger::MergeFn CreateMergeFn();
+    SpillFileMerger::MergeFn CreateSpillFileMergeFn();
     Result<FileChannelInfo> MergeAndReplaceFiles(const std::vector<FileChannelInfo>& files);
     Status SpillMemoryBuffer(std::vector<std::unique_ptr<KeyValueRecordReader>>&& readers);
 
@@ -98,7 +99,7 @@ class ExternalSortBuffer : public SortBuffer {
     const int32_t max_fan_in_;
     const std::shared_ptr<SpillChannelManager> spill_channel_manager_;
 
-    std::unique_ptr<LeveledMerger> leveled_merger_;
+    std::unique_ptr<SpillFileMerger> spill_merger_;
     std::shared_ptr<FileIOChannel::Enumerator> spill_channel_enumerator_;
     int64_t total_spill_disk_bytes_ = 0;
     int32_t actual_max_fan_in_;

src/paimon/core/mergetree/merge_tree_writer_test.cpp

@@ -1076,7 +1076,7 @@ TEST_P(MergeTreeWriterTest, TestCloseSkipsDeleteForUpgradedFilesInCompactAfter)
         << "Intermediate file should be deleted because it's not in compact_before_";
 }
 
-TEST_P(MergeTreeWriterTest, TestSpillWithSameKeyDeduplicate) {
+TEST_F(MergeTreeWriterTest, TestSpillWithSameKeyDeduplicate) {
     ASSERT_OK_AND_ASSIGN(CoreOptions options,
                          CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
                                                {Options::WRITE_BUFFER_SIZE, "1"},
@@ -1111,9 +1111,8 @@ TEST_P(MergeTreeWriterTest, TestSpillWithSameKeyDeduplicate) {
 
     WriteBatch(batch1, /*row_kinds=*/{}, merge_writer.get());
     WriteBatch(batch2, /*row_kinds=*/{}, merge_writer.get());
-    // Leveled compaction may merge 2 files into 1 when actual_max_fan_in_ is low.
-    ASSERT_GE(2u, TestHelper::CountChannelFiles(file_system_, dir->Str() + "/tmp"));
-    ASSERT_GE(TestHelper::CountChannelFiles(file_system_, dir->Str() + "/tmp"), 1u);
+    // actual_max_fan_in_=2 with 1-byte budget, so 2 spill files compact to 1.
+    ASSERT_EQ(1u, TestHelper::CountChannelFiles(file_system_, dir->Str() + "/tmp"));
 
     std::shared_ptr<arrow::Array> batch3 =
         arrow::ipc::internal::json::ArrayFromJSON(value_type_, R"([
@@ -1142,7 +1141,7 @@ TEST_P(MergeTreeWriterTest, TestSpillWithSameKeyDeduplicate) {
     CheckFileContent(expected_data_file_path, expected_array);
 }
 
-TEST_P(MergeTreeWriterTest, TestIntermediateMergeSpillFileBound) {
+TEST_F(MergeTreeWriterTest, TestIntermediateMergeSpillFileBound) {
     ASSERT_OK_AND_ASSIGN(CoreOptions options,
                          CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
                                                {Options::WRITE_BUFFER_SIZE, "1"},
@@ -1186,9 +1185,8 @@ TEST_P(MergeTreeWriterTest, TestIntermediateMergeSpillFileBound) {
     ASSERT_EQ(1u, TestHelper::CountChannelFiles(file_system_, dir->Str() + "/tmp"));
 
     WriteBatch(batch3, /*row_kinds=*/{}, merge_writer.get());
-    // Leveled compaction keeps files across levels, so file count may be > 1.
-    ASSERT_LE(TestHelper::CountChannelFiles(file_system_, dir->Str() + "/tmp"), 2u);
-    ASSERT_GE(TestHelper::CountChannelFiles(file_system_, dir->Str() + "/tmp"), 1u);
+    // After 3rd spill: 1 file at level 0 (new) + 1 file at level 1 (from prior compact) = 2.
+    ASSERT_EQ(2u, TestHelper::CountChannelFiles(file_system_, dir->Str() + "/tmp"));
 
     ASSERT_OK_AND_ASSIGN(CommitIncrement commit_increment,
                          merge_writer->PrepareCommit(/*wait_compaction=*/false));
@@ -1207,7 +1205,7 @@ TEST_P(MergeTreeWriterTest, TestIntermediateMergeSpillFileBound) {
     CheckFileContent(expected_data_file_path, expected_array);
 }
 
-TEST_P(MergeTreeWriterTest, TestDiskQuotaExhaustedFallsBackToFlushWriteBuffer) {
+TEST_F(MergeTreeWriterTest, TestDiskQuotaExhaustedFallsBackToFlushWriteBuffer) {
     ASSERT_OK_AND_ASSIGN(CoreOptions options,
                          CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
                                                {Options::WRITE_BUFFER_SIZE, "1"},
@@ -1281,7 +1279,7 @@ TEST_P(MergeTreeWriterTest, TestDiskQuotaExhaustedFallsBackToFlushWriteBuffer) {
     }
 }
 
-TEST_P(MergeTreeWriterTest, TestFlushMemoryQuotaExhaustedFallsBackToFlushWriteBuffer) {
+TEST_F(MergeTreeWriterTest, TestFlushMemoryQuotaExhaustedFallsBackToFlushWriteBuffer) {
     // WRITE_BUFFER_SIZE is large enough so WriteBatch does NOT auto-spill.
     // SPILL_MAX_DISK_SIZE is tiny so the first FlushMemory() exhausts the quota,
     // triggering the fallback path: FlushMemory() -> quota exhausted -> FlushWriteBuffer.
@@ -1328,7 +1326,7 @@ TEST_P(MergeTreeWriterTest, TestFlushMemoryQuotaExhaustedFallsBackToFlushWriteBu
     ASSERT_OK(merge_writer->Close());
 }
 
-TEST_P(MergeTreeWriterTest, TestCloseDeletesSpillTempFiles) {
+TEST_F(MergeTreeWriterTest, TestCloseDeletesSpillTempFiles) {
     ASSERT_OK_AND_ASSIGN(CoreOptions options,
                          CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
                                                {Options::WRITE_BUFFER_SIZE, "1"},
@@ -1360,7 +1358,7 @@ TEST_P(MergeTreeWriterTest, TestCloseDeletesSpillTempFiles) {
     ASSERT_EQ(0u, TestHelper::CountChannelFiles(file_system_, dir->Str() + "/tmp"));
 }
 
-TEST_P(MergeTreeWriterTest, TestMultiplePrepareCommitWithSpill) {
+TEST_F(MergeTreeWriterTest, TestMultiplePrepareCommitWithSpill) {
     ASSERT_OK_AND_ASSIGN(
         CoreOptions options,
         CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}, {Options::WRITE_ONLY, "true"}}));