This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 e1a198b6d65 [fix](compaction) Keep latest 10 versions unmerged during
schema change (#66337)
e1a198b6d65 is described below
commit e1a198b6d657885c5f77e9efa99e31fc02f2a37b
Author: Jamie <[email protected]>
AuthorDate: Thu Aug 6 16:59:32 2026 +0800
[fix](compaction) Keep latest 10 versions unmerged during schema change
(#66337)
Related PR: #63112
Problem Summary: This PR takes over #63112 and adds explicit local and
Cloud unit-test coverage.
While a schema change target tablet is `TABLET_NOTREADY`, cumulative
compaction should merge only older rowsets and leave the newest 10
versions unmerged. The filter used the inverse comparison, skipping
older rowsets and selecting the newest versions. A compaction output
could then cross the base tablet's maximum version and prevent
incremental schema-change conversion with `VERSION_ALREADY_MERGED`.
This change reverses the comparison in both local and Cloud size-based
cumulative compaction policies. The new tests verify that versions 2
through 10 are selected and versions 11 through 20 remain unmerged.
### Release note
Fix schema changes that could fail when cumulative compaction merged the
latest versions on the new tablet.
---
.../cloud/cloud_cumulative_compaction_policy.cpp | 2 +-
.../compaction/cumulative_compaction_policy.cpp | 2 +-
.../cloud_cumulative_compaction_policy_test.cpp | 34 ++++++++++++++++
.../cumulative_compaction_policy_test.cpp | 47 ++++++++++++++++++++++
4 files changed, 83 insertions(+), 2 deletions(-)
diff --git a/be/src/cloud/cloud_cumulative_compaction_policy.cpp
b/be/src/cloud/cloud_cumulative_compaction_policy.cpp
index 465f2c2c31e..890527b7e17 100644
--- a/be/src/cloud/cloud_cumulative_compaction_policy.cpp
+++ b/be/src/cloud/cloud_cumulative_compaction_policy.cpp
@@ -169,7 +169,7 @@ int64_t
CloudSizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
if (tablet->tablet_state() == TABLET_NOTREADY) {
// If tablet under alter, keep latest 10 version so that base
tablet max version
// not merged in new tablet, and then we can copy data from base
tablet
- if (rowset->version().second < max_version - 10) {
+ if (rowset->version().second > max_version - 10) {
continue;
}
}
diff --git a/be/src/storage/compaction/cumulative_compaction_policy.cpp
b/be/src/storage/compaction/cumulative_compaction_policy.cpp
index 7ab7a8e34ef..503fca42066 100644
--- a/be/src/storage/compaction/cumulative_compaction_policy.cpp
+++ b/be/src/storage/compaction/cumulative_compaction_policy.cpp
@@ -313,7 +313,7 @@ int SizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
if (tablet->tablet_state() == TABLET_NOTREADY) {
// If tablet under alter, keep latest 10 version so that base
tablet max version
// not merged in new tablet, and then we can copy data from base
tablet
- if (rowset->version().second < max_version - 10) {
+ if (rowset->version().second > max_version - 10) {
continue;
}
}
diff --git a/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp
b/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp
index 2a20b5b3cba..743391b3af7 100644
--- a/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp
+++ b/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp
@@ -178,6 +178,40 @@ TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy,
new_cumulative_point) {
EXPECT_EQ(policy.new_cumulative_point(&_tablet, output_rowset, version,
2), 6);
}
+TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy,
+ pick_input_rowsets_notready_keeps_latest_versions) {
+ auto base_rowset = create_rowset(Version(0, 1), 1, false, kGiB);
+ ASSERT_NE(nullptr, base_rowset);
+ ASSERT_TRUE(_tablet_meta->add_rs_meta(base_rowset->rowset_meta()).ok());
+
+ std::vector<RowsetSharedPtr> candidate_rowsets;
+ for (int64_t version = 2; version <= 20; ++version) {
+ auto rowset = create_rowset(Version(version, version), 1, true, kMiB);
+ ASSERT_NE(nullptr, rowset);
+ ASSERT_TRUE(_tablet_meta->add_rs_meta(rowset->rowset_meta()).ok());
+ candidate_rowsets.push_back(rowset);
+ }
+
+ CloudTablet tablet(_engine, _tablet_meta);
+ ASSERT_TRUE(tablet.set_tablet_state(TABLET_NOTREADY).ok());
+ tablet._base_size = kGiB;
+
+ std::vector<RowsetSharedPtr> input_rowsets;
+ Version last_delete_version {-1, -1};
+ size_t compaction_score = 0;
+
+ CloudSizeBasedCumulativeCompactionPolicy policy;
+ auto picked_size = policy.pick_input_rowsets(&tablet, candidate_rowsets,
100, 5, &input_rowsets,
+ &last_delete_version,
&compaction_score, true);
+
+ EXPECT_EQ(9, picked_size);
+ ASSERT_EQ(9, input_rowsets.size());
+ EXPECT_EQ(9, compaction_score);
+ EXPECT_EQ(2, input_rowsets.front()->start_version());
+ EXPECT_EQ(10, input_rowsets.back()->end_version());
+ EXPECT_EQ(Version(-1, -1), last_delete_version);
+}
+
TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy,
pick_input_rowsets_large_head_not_repeated_when_output_below_promotion)
{
CloudTablet _tablet(_engine, _tablet_meta);
diff --git a/be/test/storage/compaction/cumulative_compaction_policy_test.cpp
b/be/test/storage/compaction/cumulative_compaction_policy_test.cpp
index ce722df19b6..ad1c536451d 100644
--- a/be/test/storage/compaction/cumulative_compaction_policy_test.cpp
+++ b/be/test/storage/compaction/cumulative_compaction_policy_test.cpp
@@ -506,6 +506,53 @@ TEST_F(TestSizeBasedCumulativeCompactionPolicy,
pick_input_rowsets_normal) {
EXPECT_EQ(-1, last_delete_version.second);
}
+TEST_F(TestSizeBasedCumulativeCompactionPolicy,
pick_input_rowsets_notready_keeps_latest_versions) {
+ std::vector<RowsetMetaSharedPtr> rs_metas;
+
+ RowsetMetaSharedPtr base_rowset(new RowsetMeta());
+ init_rs_meta(base_rowset, 0, 1);
+ base_rowset->set_total_disk_size(kGiB);
+ base_rowset->set_segments_overlap(NONOVERLAPPING);
+ rs_metas.push_back(base_rowset);
+
+ for (int64_t version = 2; version <= 20; ++version) {
+ RowsetMetaSharedPtr rowset(new RowsetMeta());
+ init_rs_meta(rowset, version, version);
+ rowset->set_total_disk_size(kMiB);
+ rowset->set_num_segments(1);
+ rowset->set_segments_overlap(OVERLAPPING);
+ rs_metas.push_back(rowset);
+ }
+
+ for (const auto& rowset : rs_metas) {
+ ASSERT_TRUE(_tablet_meta->add_rs_meta(rowset).ok());
+ }
+
+ TabletSharedPtr tablet(
+ new Tablet(_engine, _tablet_meta, nullptr,
CUMULATIVE_SIZE_BASED_POLICY));
+ ASSERT_TRUE(tablet->init().ok());
+ ASSERT_TRUE(tablet->set_tablet_state(TABLET_NOTREADY).ok());
+ tablet->calculate_cumulative_point();
+
+ EXPECT_EQ(2, tablet->cumulative_layer_point());
+ auto candidate_rowsets =
tablet->pick_candidate_rowsets_to_cumulative_compaction();
+ ASSERT_EQ(19, candidate_rowsets.size());
+
+ std::vector<RowsetSharedPtr> input_rowsets;
+ Version last_delete_version {-1, -1};
+ size_t compaction_score = 0;
+ auto picked_size =
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);
+
+ EXPECT_EQ(9, picked_size);
+ ASSERT_EQ(9, input_rowsets.size());
+ EXPECT_EQ(9, compaction_score);
+ EXPECT_EQ(2, input_rowsets.front()->start_version());
+ EXPECT_EQ(10, input_rowsets.back()->end_version());
+ EXPECT_EQ(Version(-1, -1), last_delete_version);
+}
+
TEST_F(TestSizeBasedCumulativeCompactionPolicy, pick_input_rowsets_big_base) {
std::vector<RowsetMetaSharedPtr> rs_metas;
init_rs_meta_big_base(&rs_metas);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
