This is an automated email from the ASF dual-hosted git repository.
luwei16 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 54300f922fd [fix](compaction) Keep cumulative inputs mergeable after
score trimming (#65470)
54300f922fd is described below
commit 54300f922fddbeac6e679a83d7829444d4a72f77
Author: Jamie <[email protected]>
AuthorDate: Mon Aug 3 11:57:42 2026 +0800
[fix](compaction) Keep cumulative inputs mergeable after score trimming
(#65470)
### What problem does this PR solve?
Issue Number: None
Related PR: #59268
Problem Summary:
When the cumulative compaction max score is reduced, back trimming can
leave a single non-overlapping rowset. It cannot be compacted alone, so
compaction repeatedly returns `CUMULATIVE_NO_SUITABLE_VERSION` without
advancing the cumulative point.
This change remembers the last trimmed rowset in both cloud and local
size-based policies. If trimming strands a single non-overlapping
rowset, it restores the direct successor and allows one max-score
overshoot so the input remains mergeable. Existing overlapping-singleton
and cloud empty-rowset behavior is unchanged.
### Release note
Prevent max-score trimming from leaving an unmergeable cumulative
compaction input.
---
.../cloud/cloud_cumulative_compaction_policy.cpp | 15 +++-
.../compaction/cumulative_compaction_policy.cpp | 15 +++-
.../cloud_cumulative_compaction_policy_test.cpp | 62 ++++++++++++++++
.../cumulative_compaction_policy_test.cpp | 84 ++++++++++++++++++++++
4 files changed, 170 insertions(+), 6 deletions(-)
diff --git a/be/src/cloud/cloud_cumulative_compaction_policy.cpp
b/be/src/cloud/cloud_cumulative_compaction_policy.cpp
index ca422be9b24..465f2c2c31e 100644
--- a/be/src/cloud/cloud_cumulative_compaction_policy.cpp
+++ b/be/src/cloud/cloud_cumulative_compaction_policy.cpp
@@ -122,6 +122,7 @@ int64_t
CloudSizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
*compaction_score = 0;
int64_t total_size = 0;
bool skip_trim = false; // Skip trim for Empty Rowset Compaction
+ RowsetSharedPtr last_popped;
// DEFER: trim input_rowsets from back if score > max_compaction_score
// This ensures we don't return more rowsets than allowed by
max_compaction_score,
@@ -134,11 +135,19 @@ int64_t
CloudSizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
// Keep at least 1 rowset to avoid removing the only rowset
(consistent with fallback branch)
while (input_rowsets->size() > 1 &&
*compaction_score > static_cast<size_t>(max_compaction_score)) {
- auto& last_rowset = input_rowsets->back();
- *compaction_score -=
last_rowset->rowset_meta()->get_compaction_score();
- total_size -= last_rowset->rowset_meta()->total_disk_size();
+ last_popped = std::move(input_rowsets->back());
+ *compaction_score -=
last_popped->rowset_meta()->get_compaction_score();
+ total_size -= last_popped->rowset_meta()->total_disk_size();
input_rowsets->pop_back();
}
+ // A single non-overlapping rowset cannot be compacted by itself.
Restore the direct
+ // successor and accept a one-off max-score overshoot to keep the
input mergeable.
+ if (input_rowsets->size() == 1 && last_popped != nullptr &&
+ !input_rowsets->front()->rowset_meta()->is_segments_overlapping())
{
+ *compaction_score +=
last_popped->rowset_meta()->get_compaction_score();
+ total_size += last_popped->rowset_meta()->total_disk_size();
+ input_rowsets->push_back(std::move(last_popped));
+ }
});
for (auto& rowset : candidate_rowsets) {
diff --git a/be/src/storage/compaction/cumulative_compaction_policy.cpp
b/be/src/storage/compaction/cumulative_compaction_policy.cpp
index a120d941e6f..7ab7a8e34ef 100644
--- a/be/src/storage/compaction/cumulative_compaction_policy.cpp
+++ b/be/src/storage/compaction/cumulative_compaction_policy.cpp
@@ -269,6 +269,7 @@ int SizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
int transient_size = 0;
*compaction_score = 0;
int64_t total_size = 0;
+ RowsetSharedPtr last_popped;
// DEFER: trim input_rowsets from back if score > max_compaction_score
// This ensures we don't return more rowsets than allowed by
max_compaction_score,
@@ -278,11 +279,19 @@ int
SizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
// Keep at least 1 rowset to avoid removing the only rowset
(consistent with fallback branch)
while (input_rowsets->size() > 1 &&
*compaction_score > static_cast<size_t>(max_compaction_score)) {
- auto& last_rowset = input_rowsets->back();
- *compaction_score -=
last_rowset->rowset_meta()->get_compaction_score();
- total_size -= last_rowset->rowset_meta()->total_disk_size();
+ last_popped = std::move(input_rowsets->back());
+ *compaction_score -=
last_popped->rowset_meta()->get_compaction_score();
+ total_size -= last_popped->rowset_meta()->total_disk_size();
input_rowsets->pop_back();
}
+ // A single non-overlapping rowset cannot be compacted by itself.
Restore the direct
+ // successor and accept a one-off max-score overshoot to keep the
input mergeable.
+ if (input_rowsets->size() == 1 && last_popped != nullptr &&
+ !input_rowsets->front()->rowset_meta()->is_segments_overlapping())
{
+ *compaction_score +=
last_popped->rowset_meta()->get_compaction_score();
+ total_size += last_popped->rowset_meta()->total_disk_size();
+ input_rowsets->push_back(std::move(last_popped));
+ }
});
for (auto& rowset : candidate_rowsets) {
diff --git a/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp
b/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp
index 08799c900d2..2a20b5b3cba 100644
--- a/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp
+++ b/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp
@@ -145,6 +145,22 @@ static int64_t total_disk_size(const
std::vector<RowsetSharedPtr>& rowsets) {
return total_size;
}
+static std::vector<RowsetSharedPtr> create_max_score_trim_candidates(bool
include_stranded_head) {
+ std::vector<RowsetSharedPtr> candidate_rowsets;
+ if (include_stranded_head) {
+ candidate_rowsets.push_back(create_rowset(Version(13, 56), 0, false,
0));
+ }
+ candidate_rowsets.push_back(create_rowset(Version(57, 57), 192, true, 256
* 1024 * 1024));
+ candidate_rowsets.push_back(create_rowset(Version(58, 58), 0, false, 0));
+ candidate_rowsets.push_back(create_rowset(Version(59, 59), 0, false, 0));
+ candidate_rowsets.push_back(create_rowset(Version(60, 60), 150, true, 256
* 1024 * 1024));
+ for (int64_t version = 61; version <= 67; ++version) {
+ candidate_rowsets.push_back(
+ create_rowset(Version(version, version), 1, false, 1024 *
1024));
+ }
+ return candidate_rowsets;
+}
+
TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy, new_cumulative_point) {
std::vector<RowsetMetaSharedPtr> rs_metas;
init_rs_meta_small_base(&rs_metas);
@@ -267,6 +283,52 @@ TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy,
EXPECT_EQ(0, compaction_score);
}
+TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy,
+ pick_input_rowsets_restores_successor_for_non_overlapping_singleton) {
+ CloudTablet tablet(_engine, _tablet_meta);
+ tablet._base_size = 1024L * 1024 * 1024;
+ tablet._tablet_meta->_enable_unique_key_merge_on_write = true;
+ auto candidate_rowsets = create_max_score_trim_candidates(true);
+ ASSERT_EQ(12, candidate_rowsets.size());
+
+ std::vector<RowsetSharedPtr> input_rowsets;
+ Version last_delete_version {-1, -1};
+ size_t compaction_score = 0;
+ CloudSizeBasedCumulativeCompactionPolicy policy;
+ policy.pick_input_rowsets(&tablet, candidate_rowsets, 100, 5,
&input_rowsets,
+ &last_delete_version, &compaction_score, true);
+
+ ASSERT_EQ(2, input_rowsets.size());
+ EXPECT_EQ(Version(13, 56), input_rowsets[0]->version());
+ EXPECT_EQ(Version(57, 57), input_rowsets[1]->version());
+ EXPECT_EQ(193, compaction_score);
+
+ auto output_rowset = create_rowset(Version(13, 57), 1, false, 256 * 1024 *
1024);
+ ASSERT_NE(nullptr, output_rowset);
+ EXPECT_EQ(58, policy.new_cumulative_point(&tablet, output_rowset,
last_delete_version, 13));
+}
+
+TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy,
+ pick_input_rowsets_keeps_single_overlapping_rowset_after_trim) {
+ CloudTablet tablet(_engine, _tablet_meta);
+ tablet._base_size = 1024L * 1024 * 1024;
+ auto candidate_rowsets = create_max_score_trim_candidates(false);
+ ASSERT_EQ(11, candidate_rowsets.size());
+
+ std::vector<RowsetSharedPtr> input_rowsets;
+ Version last_delete_version {-1, -1};
+ size_t compaction_score = 0;
+ CloudSizeBasedCumulativeCompactionPolicy policy;
+ policy.pick_input_rowsets(&tablet, candidate_rowsets, 100, 5,
&input_rowsets,
+ &last_delete_version, &compaction_score, true);
+
+ // The tail was trimmed, but the overlapping head remains mergeable by
itself.
+ ASSERT_EQ(1, input_rowsets.size());
+ EXPECT_GT(candidate_rowsets.size(), input_rowsets.size());
+ EXPECT_EQ(Version(57, 57), input_rowsets.front()->version());
+ EXPECT_EQ(192, compaction_score);
+}
+
// Test case: Empty rowset compaction with skip_trim
TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy,
pick_input_rowsets_empty_rowset_compaction) {
// Save original config values
diff --git a/be/test/storage/compaction/cumulative_compaction_policy_test.cpp
b/be/test/storage/compaction/cumulative_compaction_policy_test.cpp
index 58b18128b86..ce722df19b6 100644
--- a/be/test/storage/compaction/cumulative_compaction_policy_test.cpp
+++ b/be/test/storage/compaction/cumulative_compaction_policy_test.cpp
@@ -329,6 +329,32 @@ public:
rs_metas->push_back(ptr5);
}
+ std::vector<RowsetMetaSharedPtr> create_rs_meta_max_score_trim(bool
include_stranded_head) {
+ std::vector<RowsetMetaSharedPtr> rs_metas;
+ auto add_rowset = [&](int64_t start_version, int64_t end_version, int
num_segments,
+ bool overlapping, int64_t total_disk_size) {
+ RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
+ init_rs_meta(rowset_meta, start_version, end_version);
+ rowset_meta->set_num_segments(num_segments);
+ rowset_meta->set_segments_overlap(overlapping ? OVERLAPPING :
NONOVERLAPPING);
+ rowset_meta->set_total_disk_size(total_disk_size);
+ rs_metas.push_back(rowset_meta);
+ };
+
+ add_rowset(0, include_stranded_head ? 12 : 56, 1, false, 1024L * 1024
* 1024);
+ if (include_stranded_head) {
+ add_rowset(13, 56, 0, false, 0);
+ }
+ add_rowset(57, 57, 192, true, 256L * 1024 * 1024);
+ add_rowset(58, 58, 0, false, 0);
+ add_rowset(59, 59, 0, false, 0);
+ add_rowset(60, 60, 150, true, 256L * 1024 * 1024);
+ for (int64_t version = 61; version <= 67; ++version) {
+ add_rowset(version, version, 1, false, 1024L * 1024);
+ }
+ return rs_metas;
+ }
+
protected:
std::string _json_rowset_meta;
TabletMetaSharedPtr _tablet_meta;
@@ -1301,6 +1327,64 @@ TEST_F(TestSizeBasedCumulativeCompactionPolicy,
EXPECT_EQ(0, compaction_score);
}
+TEST_F(TestSizeBasedCumulativeCompactionPolicy,
+ pick_input_rowsets_restores_successor_for_non_overlapping_singleton) {
+ auto rs_metas = create_rs_meta_max_score_trim(true);
+ for (auto& rowset : rs_metas) {
+ static_cast<void>(_tablet_meta->add_rs_meta(rowset));
+ }
+
+ TabletSharedPtr tablet(
+ new Tablet(_engine, _tablet_meta, nullptr,
CUMULATIVE_SIZE_BASED_POLICY));
+ static_cast<void>(tablet->init());
+ tablet->calculate_cumulative_point();
+ EXPECT_EQ(13, tablet->cumulative_layer_point());
+ EXPECT_EQ(64L * 1024 * 1024, tablet->cumulative_promotion_size());
+
+ auto candidate_rowsets =
tablet->pick_candidate_rowsets_to_cumulative_compaction();
+ ASSERT_EQ(12, candidate_rowsets.size());
+ std::vector<RowsetSharedPtr> input_rowsets;
+ Version last_delete_version {-1, -1};
+ size_t compaction_score = 0;
+ tablet->_cumulative_compaction_policy->pick_input_rowsets(
+ tablet.get(), candidate_rowsets, 100, 5, &input_rowsets,
&last_delete_version,
+ &compaction_score, config::enable_delete_when_cumu_compaction);
+
+ ASSERT_EQ(2, input_rowsets.size());
+ EXPECT_EQ(Version(13, 56), input_rowsets[0]->version());
+ EXPECT_EQ(Version(57, 57), input_rowsets[1]->version());
+ EXPECT_EQ(193, compaction_score);
+}
+
+TEST_F(TestSizeBasedCumulativeCompactionPolicy,
+ pick_input_rowsets_keeps_single_overlapping_rowset_after_trim) {
+ auto rs_metas = create_rs_meta_max_score_trim(false);
+ for (auto& rowset : rs_metas) {
+ static_cast<void>(_tablet_meta->add_rs_meta(rowset));
+ }
+
+ TabletSharedPtr tablet(
+ new Tablet(_engine, _tablet_meta, nullptr,
CUMULATIVE_SIZE_BASED_POLICY));
+ static_cast<void>(tablet->init());
+ tablet->calculate_cumulative_point();
+ EXPECT_EQ(57, tablet->cumulative_layer_point());
+
+ auto candidate_rowsets =
tablet->pick_candidate_rowsets_to_cumulative_compaction();
+ ASSERT_EQ(11, candidate_rowsets.size());
+ std::vector<RowsetSharedPtr> input_rowsets;
+ Version last_delete_version {-1, -1};
+ size_t compaction_score = 0;
+ tablet->_cumulative_compaction_policy->pick_input_rowsets(
+ tablet.get(), candidate_rowsets, 100, 5, &input_rowsets,
&last_delete_version,
+ &compaction_score, config::enable_delete_when_cumu_compaction);
+
+ // The tail was trimmed, but the overlapping head remains mergeable by
itself.
+ ASSERT_EQ(1, input_rowsets.size());
+ EXPECT_GT(candidate_rowsets.size(), input_rowsets.size());
+ EXPECT_EQ(Version(57, 57), input_rowsets.front()->version());
+ EXPECT_EQ(192, compaction_score);
+}
+
// Test case: Trim with varying scores (high score rowsets at tail)
TEST_F(TestSizeBasedCumulativeCompactionPolicy,
pick_input_rowsets_trim_high_score_tail) {
std::vector<RowsetMetaSharedPtr> rs_metas;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
