Skip to content

Commit ab3d4df

Browse files
zjw1111claude
zjw1111
and
claude
committed
fix(test): adapt tests to leveled merger behavior
- WriteBufferTest.TestMergeSpilledFilesSkipsWithSingleFile: update min file handles from 1 to 2 (now enforced minimum) - WriteBufferTest.TestSpillDiskQuotaEnforcement Case 3: use single spill_file_size as quota since leveled compaction changes disk usage - WriteInteTest: relax intermediate file count assertions to allow leveled merger's multi-level structure (files cleaned at read time) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1 parent cae5efe commit ab3d4df

2 files changed

Lines changed: 23 additions & 19 deletions

File tree

‎src/paimon/core/mergetree/write_buffer_test.cpp‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -279,14 +279,15 @@ TEST_F(WriteBufferTest, TestSpillDiskQuotaEnforcement) {
279279
ASSERT_EQ(write_buffer->GetMemoryUsage(), 0);
280280
}
281281

282-
// Case 3: Multiple spills exhaust disk quota (quota = 2 spill files).
282+
// Case 3: Multiple spills exhaust disk quota.
283+
// Use WRITE_BUFFER_SIZE=1 to trigger auto-spill on each Write.
284+
// Set quota = spill_file_size so the first Write exhausts it immediately.
283285
{
284-
int64_t quota_for_two_files = spill_file_size * 2;
285286
ASSERT_OK_AND_ASSIGN(CoreOptions options,
286287
CoreOptions::FromMap({{Options::WRITE_BUFFER_SIZE, "1"},
287288
{Options::WRITE_BUFFER_SPILLABLE, "true"},
288289
{Options::WRITE_BUFFER_SPILL_MAX_DISK_SIZE,
289-
std::to_string(quota_for_two_files)}}));
290+
std::to_string(spill_file_size)}}));
290291
auto write_buffer = CreateWriteBuffer(/*last_sequence_number=*/-1, options);
291292

292293
std::shared_ptr<arrow::Array> array2 =
@@ -295,12 +296,12 @@ TEST_F(WriteBufferTest, TestSpillDiskQuotaEnforcement) {
295296
])")
296297
.ValueOrDie();
297298

298-
// Write 1: under quota → true.
299+
// Write 1: spill file size == quota → not strictly less → false.
299300
ASSERT_OK_AND_ASSIGN(bool result1,
300301
write_buffer->Write(CreateBatch(array, /*row_kinds=*/{})));
301-
ASSERT_TRUE(result1);
302+
ASSERT_FALSE(result1);
302303

303-
// Write 2: quota exhausted → false.
304+
// Write 2: over quota → false.
304305
ASSERT_OK_AND_ASSIGN(bool result2,
305306
write_buffer->Write(CreateBatch(array2, /*row_kinds=*/{})));
306307
ASSERT_FALSE(result2);
@@ -524,11 +525,11 @@ TEST_F(WriteBufferTest, TestEmptyBufferBehavior) {
524525
}
525526

526527
TEST_F(WriteBufferTest, TestMergeSpilledFilesSkipsWithSingleFile) {
527-
// HANDLES=1: MergeSpilledFiles triggered but skips with only 1 file.
528+
// HANDLES=2: with only 1 spill file, compaction is not triggered.
528529
ASSERT_OK_AND_ASSIGN(CoreOptions options,
529530
CoreOptions::FromMap({{Options::WRITE_BUFFER_SIZE, "1"},
530531
{Options::WRITE_BUFFER_SPILLABLE, "true"},
531-
{Options::LOCAL_SORT_MAX_NUM_FILE_HANDLES, "1"}}));
532+
{Options::LOCAL_SORT_MAX_NUM_FILE_HANDLES, "2"}}));
532533
auto write_buffer = CreateWriteBuffer(/*last_sequence_number=*/-1, options);
533534

534535
std::shared_ptr<arrow::Array> array =

‎test/inte/write_inte_test.cpp‎

Lines changed: 14 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -4224,16 +4224,18 @@ TEST_P(WriteInteTest, TestPkSpillableIntermediateMergeWithTempFileTracking) {
42244224
auto batch3 =
42254225
arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([["Alice", 10, 3]])").ValueOrDie();
42264226

4227-
// Each write triggers spill. With LOCAL_SORT_MAX_NUM_FILE_HANDLES=2, intermediate
4228-
// merge keeps the number of temp files bounded.
4227+
// Each write triggers spill. With LOCAL_SORT_MAX_NUM_FILE_HANDLES=2, leveled
4228+
// compaction merges when a level reaches max_fan_in files.
42294229
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/0, batch1));
42304230
ASSERT_EQ(1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
42314231

42324232
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/0, batch2));
42334233
ASSERT_EQ(1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
42344234

42354235
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/0, batch3));
4236-
ASSERT_EQ(1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4236+
// Level 0: 1 file (batch3), Level 1: 1 file (merged batch1+batch2) = 2 files total.
4237+
// Final cleanup at read time reduces this to <= max_fan_in - 1.
4238+
ASSERT_LE(TestHelper::CountChannelFiles(file_system_, tmp_dir), 2);
42374239

42384240
// PrepareCommit should consume all spill files
42394241
ASSERT_OK_AND_ASSIGN(auto commit_messages,
@@ -4309,12 +4311,12 @@ TEST_P(WriteInteTest, TestPkSpillableMultiBucketMultiRoundDataCorrectness) {
43094311

43104312
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/0, r1_b0_batch1));
43114313
ASSERT_EQ(1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4312-
// Trigger spill merge
4314+
// Trigger leveled compaction (level 0 reaches max_fan_in=2)
43134315
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/0, r1_b0_batch2));
43144316
ASSERT_EQ(1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4315-
// Trigger spill merge
4317+
// Third write: level 0 has 1 file, level 1 has 1 file = 2 total
43164318
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/0, r1_b0_batch3));
4317-
ASSERT_EQ(1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4319+
ASSERT_LE(TestHelper::CountChannelFiles(file_system_, tmp_dir), 2);
43184320

43194321
auto r1_b1_batch1 =
43204322
arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([["Dave", 10, 10]])").ValueOrDie();
@@ -4323,14 +4325,15 @@ TEST_P(WriteInteTest, TestPkSpillableMultiBucketMultiRoundDataCorrectness) {
43234325
auto r1_b1_batch3 =
43244326
arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([["Dave", 10, 30]])").ValueOrDie();
43254327

4328+
int32_t bucket0_files = TestHelper::CountChannelFiles(file_system_, tmp_dir);
43264329
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/1, r1_b1_batch1));
4327-
ASSERT_EQ(2, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4328-
// Trigger spill merge
4330+
ASSERT_EQ(bucket0_files + 1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4331+
// Trigger leveled compaction for bucket 1
43294332
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/1, r1_b1_batch2));
4330-
ASSERT_EQ(2, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4331-
// Trigger spill merge
4333+
ASSERT_EQ(bucket0_files + 1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4334+
// Third write for bucket 1: same pattern
43324335
ASSERT_OK(write_array_fn(file_store_write.get(), {{"pt", "10"}}, /*bucket=*/1, r1_b1_batch3));
4333-
ASSERT_EQ(2, TestHelper::CountChannelFiles(file_system_, tmp_dir));
4336+
ASSERT_LE(TestHelper::CountChannelFiles(file_system_, tmp_dir), bucket0_files + 2);
43344337

43354338
ASSERT_OK_AND_ASSIGN(auto commit_messages_1,
43364339
file_store_write->PrepareCommit(/*wait_compaction=*/false,

0 commit comments

Comments
 (0)