Skip to content

Commit 8487f49

Browse files
lxy-9602lxy-9602
committed
add ForceUpLevel0Compaction
1 parent e5331f4 commit 8487f49

6 files changed

Lines changed: 261 additions & 0 deletions

File tree

‎src/paimon/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -528,6 +528,7 @@ if(PAIMON_BUILD_TESTS)
528528
core/mergetree/compact/off_peak_hours_test.cpp
529529
core/mergetree/compact/early_full_compaction_test.cpp
530530
core/mergetree/compact/universal_compaction_test.cpp
531+
core/mergetree/compact/force_up_level0_compaction_test.cpp
531532
core/mergetree/compact/compact_strategy_test.cpp
532533
core/mergetree/drop_delete_reader_test.cpp
533534
core/mergetree/merge_tree_writer_test.cpp

‎src/paimon/core/mergetree/compact/force_up_level0_compaction.h‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#pragma once
18+
19+
#include <atomic>
20+
21+
#include "paimon/core/mergetree/compact/compact_strategy.h"
22+
#include "paimon/core/mergetree/compact/universal_compaction.h"
23+
24+
namespace paimon {
25+
/// A `CompactStrategy` to force compacting level 0 files.
26+
class ForceUpLevel0Compaction : public CompactStrategy {
27+
public:
28+
ForceUpLevel0Compaction(const std::shared_ptr<UniversalCompaction>& universal,
29+
const std::optional<int32_t>& max_compact_interval)
30+
: universal_(universal), max_compact_interval_(max_compact_interval) {
31+
assert(universal_);
32+
if (max_compact_interval_) {
33+
compact_trigger_count_ = std::make_unique<std::atomic<int32_t>>(0);
34+
}
35+
}
36+
37+
std::optional<int32_t> MaxCompactInterval() const {
38+
return max_compact_interval_;
39+
}
40+
41+
Result<std::optional<CompactUnit>> Pick(int32_t num_levels,
42+
const std::vector<LevelSortedRun>& runs) override {
43+
PAIMON_ASSIGN_OR_RAISE(std::optional<CompactUnit> unit, universal_->Pick(num_levels, runs));
44+
if (unit) {
45+
return unit;
46+
}
47+
if (!max_compact_interval_ || !compact_trigger_count_) {
48+
return universal_->ForcePickL0(num_levels, runs);
49+
}
50+
51+
compact_trigger_count_->fetch_add(1);
52+
// We must copy max_compact_interval because compare_exchange_strong(T& expected, T desired)
53+
// modifies 'expected' to the current actual value of the atomic if the comparison fails.
54+
int32_t expected_compact_interval = max_compact_interval_.value();
55+
if (compact_trigger_count_->compare_exchange_strong(expected_compact_interval, 0)) {
56+
// Universal compaction due to max lookup compaction interval
57+
return universal_->ForcePickL0(num_levels, runs);
58+
} else {
59+
// Skip universal compaction due to lookup compaction trigger count is less than the max
60+
// interval
61+
return std::optional<CompactUnit>();
62+
}
63+
}
64+
65+
private:
66+
std::shared_ptr<UniversalCompaction> universal_;
67+
std::optional<int32_t> max_compact_interval_;
68+
std::unique_ptr<std::atomic<int32_t>> compact_trigger_count_;
69+
};
70+
} // namespace paimon

‎src/paimon/core/mergetree/compact/force_up_level0_compaction_test.cpp‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/core/mergetree/compact/force_up_level0_compaction.h"
18+
19+
#include "paimon/core/mergetree/compact/universal_compaction.h"
20+
#include "paimon/testing/utils/testharness.h"
21+
22+
namespace paimon::test {
23+
class ForceUpLevel0CompactionTest : public testing::Test {
24+
public:
25+
LevelSortedRun CreateLevelSortedRun(int32_t level, int64_t size) const {
26+
auto file_meta = std::make_shared<DataFileMeta>(
27+
"fake.data", /*file_size=*/size, /*row_count=*/1,
28+
/*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/
29+
BinaryRow::EmptyRow(),
30+
/*key_stats=*/
31+
SimpleStats::EmptyStats(),
32+
/*value_stats=*/
33+
SimpleStats::EmptyStats(),
34+
/*min_sequence_number=*/0, /*max_sequence_number=*/6, /*schema_id=*/0,
35+
/*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
36+
/*creation_time=*/Timestamp(0ll, 0),
37+
/*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),
38+
/*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
39+
/*first_row_id=*/std::nullopt,
40+
/*write_cols=*/std::nullopt);
41+
return {level, SortedRun::FromSingle(file_meta)};
42+
}
43+
44+
std::vector<LevelSortedRun> CreateRunsWithLevelAndSize(
45+
const std::vector<int32_t>& levels, const std::vector<int64_t>& sizes) const {
46+
EXPECT_EQ(levels.size(), sizes.size());
47+
std::vector<LevelSortedRun> runs;
48+
for (size_t i = 0; i < levels.size(); i++) {
49+
runs.push_back(CreateLevelSortedRun(levels[i], sizes[i]));
50+
}
51+
return runs;
52+
}
53+
};
54+
55+
TEST_F(ForceUpLevel0CompactionTest, TestForceCompaction0) {
56+
auto universal =
57+
std::make_shared<UniversalCompaction>(/*max_size_amp=*/200, /*size_ratio=*/1,
58+
/*num_run_compaction_trigger=*/5, nullptr, nullptr);
59+
ForceUpLevel0Compaction compaction(universal, /*max_compact_interval=*/std::nullopt);
60+
61+
ASSERT_OK_AND_ASSIGN(
62+
auto unit, compaction.Pick(/*num_levels=*/3, CreateRunsWithLevelAndSize({0, 0}, {1, 1})));
63+
ASSERT_TRUE(unit);
64+
ASSERT_EQ(unit.value().output_level, 2);
65+
66+
ASSERT_OK_AND_ASSIGN(
67+
unit, compaction.Pick(/*num_levels=*/3, CreateRunsWithLevelAndSize({0, 1}, {1, 10})));
68+
ASSERT_TRUE(unit);
69+
ASSERT_EQ(unit.value().output_level, 2);
70+
71+
ASSERT_OK_AND_ASSIGN(
72+
unit, compaction.Pick(/*num_levels=*/3, CreateRunsWithLevelAndSize({0, 0, 2}, {1, 5, 10})));
73+
ASSERT_TRUE(unit);
74+
ASSERT_EQ(unit.value().output_level, 1);
75+
76+
ASSERT_OK_AND_ASSIGN(unit,
77+
compaction.Pick(/*num_levels=*/3, CreateRunsWithLevelAndSize({2}, {10})));
78+
ASSERT_FALSE(unit);
79+
80+
ASSERT_OK_AND_ASSIGN(unit,
81+
compaction.Pick(/*num_levels=*/3,
82+
CreateRunsWithLevelAndSize({0, 0, 0, 0}, {1, 5, 10, 20})));
83+
ASSERT_TRUE(unit);
84+
ASSERT_EQ(unit.value().output_level, 2);
85+
}
86+
} // namespace paimon::test

‎src/paimon/core/mergetree/compact/universal_compaction.cpp‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,21 @@ Result<std::optional<CompactUnit>> UniversalCompaction::Pick(
6262
return std::optional<CompactUnit>();
6363
}
6464

65+
Result<std::optional<CompactUnit>> UniversalCompaction::ForcePickL0(
66+
int32_t num_levels, const std::vector<LevelSortedRun>& runs) {
67+
// collect all level 0 files
68+
int32_t candidate_count = 0;
69+
for (; static_cast<size_t>(candidate_count) < runs.size(); ++candidate_count) {
70+
if (runs[candidate_count].level > 0) {
71+
break;
72+
}
73+
}
74+
if (candidate_count == 0) {
75+
return std::optional<CompactUnit>();
76+
}
77+
return PickForSizeRatio(num_levels - 1, runs, candidate_count, /*force_pick=*/true);
78+
}
79+
6580
std::optional<CompactUnit> UniversalCompaction::PickForSizeAmp(
6681
int32_t max_level, const std::vector<LevelSortedRun>& runs) {
6782
if (runs.size() < static_cast<size_t>(num_run_compaction_trigger_)) {

‎src/paimon/core/mergetree/compact/universal_compaction.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,9 @@ class UniversalCompaction : public CompactStrategy {
3535
Result<std::optional<CompactUnit>> Pick(int32_t num_levels,
3636
const std::vector<LevelSortedRun>& runs) override;
3737

38+
Result<std::optional<CompactUnit>> ForcePickL0(int32_t num_levels,
39+
const std::vector<LevelSortedRun>& runs);
40+
3841
private:
3942
std::optional<CompactUnit> PickForSizeAmp(int32_t max_level,
4043
const std::vector<LevelSortedRun>& runs);

‎src/paimon/core/mergetree/compact/universal_compaction_test.cpp‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
#include "paimon/core/mergetree/compact/universal_compaction.h"
1818

19+
#include "paimon/core/mergetree/compact/force_up_level0_compaction.h"
1920
#include "paimon/testing/utils/testharness.h"
2021

2122
namespace paimon::test {
@@ -358,4 +359,89 @@ TEST_F(UniversalCompactionTest, TestSizeRatioThreshold) {
358359
ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()), std::vector<int64_t>({8, 9, 10}));
359360
}
360361
}
362+
363+
TEST_F(UniversalCompactionTest, TestLookup) {
364+
auto universal =
365+
std::make_shared<UniversalCompaction>(/*max_size_amp=*/25, /*size_ratio=*/1,
366+
/*num_run_compaction_trigger=*/3, nullptr, nullptr);
367+
ForceUpLevel0Compaction compaction(universal, /*max_compact_interval=*/std::nullopt);
368+
369+
// level 0 to max level
370+
ASSERT_OK_AND_ASSIGN(auto unit,
371+
compaction.Pick(/*num_levels=*/3, CreateRunsWithSize({1, 2, 2, 2})));
372+
ASSERT_TRUE(unit);
373+
ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()), std::vector<int64_t>({1, 2, 2, 2}));
374+
ASSERT_EQ(unit.value().output_level, 2);
375+
376+
// level 0 force pick
377+
ASSERT_OK_AND_ASSIGN(
378+
unit, compaction.Pick(/*num_levels=*/3, CreateRunsWithLevelAndSize(/*levels=*/{0, 1, 2},
379+
/*sizes=*/{1, 2, 2})));
380+
ASSERT_TRUE(unit);
381+
ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()), std::vector<int64_t>({1, 2, 2}));
382+
ASSERT_EQ(unit.value().output_level, 2);
383+
384+
// level 0 to empty level
385+
ASSERT_OK_AND_ASSIGN(
386+
unit, compaction.Pick(/*num_levels=*/3,
387+
CreateRunsWithLevelAndSize(/*levels=*/{0, 2}, /*sizes=*/{1, 2})));
388+
ASSERT_TRUE(unit);
389+
ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()), std::vector<int64_t>({1}));
390+
ASSERT_EQ(unit.value().output_level, 1);
391+
}
392+
393+
TEST_F(UniversalCompactionTest, TestForcePickL0) {
394+
int32_t max_compact_interval = 5;
395+
auto universal =
396+
std::make_shared<UniversalCompaction>(/*max_size_amp=*/25, /*size_ratio=*/1,
397+
/*num_run_compaction_trigger=*/5, nullptr, nullptr);
398+
ForceUpLevel0Compaction compaction(universal, max_compact_interval);
399+
400+
// level 0 to max level
401+
auto level0_to_max = CreateRunsWithSize({1, 2, 2, 2});
402+
std::optional<CompactUnit> unit;
403+
for (int32_t i = 1; i <= max_compact_interval; i++) {
404+
// level 0 to max level triggered
405+
ASSERT_OK_AND_ASSIGN(unit, compaction.Pick(/*num_levels=*/3, level0_to_max));
406+
if (i == max_compact_interval) {
407+
ASSERT_TRUE(unit);
408+
ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()),
409+
std::vector<int64_t>({1, 2, 2, 2}));
410+
ASSERT_EQ(unit.value().output_level, 2);
411+
} else {
412+
// compact skipped
413+
ASSERT_FALSE(unit);
414+
}
415+
}
416+
417+
// level 0 force pick
418+
auto level0_force_pick = CreateRunsWithLevelAndSize(/*levels=*/{0, 1, 2}, /*sizes=*/{2, 2, 2});
419+
for (int32_t i = 1; i <= max_compact_interval; i++) {
420+
ASSERT_OK_AND_ASSIGN(unit, compaction.Pick(/*num_levels=*/3, level0_force_pick));
421+
if (i == max_compact_interval) {
422+
// level 0 force pick triggered
423+
ASSERT_TRUE(unit);
424+
ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()), std::vector<int64_t>({2, 2, 2}));
425+
ASSERT_EQ(unit.value().output_level, 2);
426+
} else {
427+
// compact skipped
428+
ASSERT_FALSE(unit);
429+
}
430+
}
431+
432+
// level 0 to empty level
433+
auto level0_to_empty = CreateRunsWithLevelAndSize(/*levels=*/{0, 2}, /*sizes=*/{1, 2});
434+
for (int32_t i = 1; i <= max_compact_interval; i++) {
435+
ASSERT_OK_AND_ASSIGN(unit, compaction.Pick(/*num_levels=*/3, level0_to_empty));
436+
if (i == max_compact_interval) {
437+
// level 0 force pick triggered
438+
ASSERT_TRUE(unit);
439+
ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()), std::vector<int64_t>({1}));
440+
ASSERT_EQ(unit.value().output_level, 1);
441+
} else {
442+
// compact skipped
443+
ASSERT_FALSE(unit);
444+
}
445+
}
446+
}
361447
} // namespace paimon::test

0 commit comments

Comments
 (0)