[fix](compaction) Keep cumulative compaction input mergeable when max-score trim strands a single non-overlapping rowset

be/src/cloud/cloud_cumulative_compaction_policy.cpp

@@ -121,6 +121,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,
@@ -133,11 +134,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) {

be/src/cloud/cloud_storage_engine.cpp

@@ -93,6 +93,19 @@ int get_base_thread_num() {
     return std::min(std::max(int(num_cores * config::base_compaction_thread_num_factor), 1), 10);
 }
 
+static void log_compaction_submit_failure(int64_t tablet_id, const Status& status) {
+    const bool debug_log_enabled = VLOG_DEBUG_IS_ON;
+    if (status.is<ErrorCode::CUMULATIVE_NO_SUITABLE_VERSION>() && !debug_log_enabled) {
+        LOG_EVERY_N(INFO, 100) << "failed to submit cumulative compaction task for tablet: "
+                               << tablet_id << ", err: " << status;
+        return;
+    }
+    if (!status.is<ErrorCode::BE_NO_SUITABLE_VERSION>() || debug_log_enabled) {
+        LOG(WARNING) << "failed to submit compaction task for tablet: " << tablet_id
+                     << ", err: " << status;
+    }
+}
+
 CloudStorageEngine::CloudStorageEngine(const EngineOptions& options)
         : BaseStorageEngine(Type::CLOUD, options.backend_uid),
           _meta_mgr(std::make_unique<cloud::CloudMetaMgr>()),
@@ -595,13 +608,10 @@ void CloudStorageEngine::_compaction_tasks_producer_callback() {
             /// thus cannot be collected by the garbage collector. (TabletManager::start_trash_sweep)
             for (const auto& tablet : tablets_compaction) {
                 Status status = submit_compaction_task(tablet, compaction_type);
-                if (status.ok()) continue;
-                if ((!status.is<ErrorCode::BE_NO_SUITABLE_VERSION>() &&
-                     !status.is<ErrorCode::CUMULATIVE_NO_SUITABLE_VERSION>()) ||
-                    VLOG_DEBUG_IS_ON) {
-                    LOG(WARNING) << "failed to submit compaction task for tablet: "
-                                 << tablet->tablet_id() << ", err: " << status;
+                if (status.ok()) {
+                    continue;
                 }
+                log_compaction_submit_failure(tablet->tablet_id(), status);
             }
             interval = config::generate_compaction_tasks_interval_ms;
         } else {

be/src/storage/compaction/cumulative_compaction_policy.cpp

@@ -268,6 +268,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,
@@ -277,11 +278,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) {

be/test/cloud/cloud_cumulative_compaction_policy_test.cpp

@@ -134,6 +134,22 @@ static RowsetSharedPtr create_rowset(Version version, int num_segments, bool ove
     return rowset;
 }
 
+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);
@@ -151,6 +167,52 @@ TEST_F(TestCloudSizeBasedCumulativeCompactionPolicy, new_cumulative_point) {
     EXPECT_EQ(policy.new_cumulative_point(&_tablet, output_rowset, version, 2), 6);
 }
 
+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

be/test/storage/compaction/cumulative_compaction_policy_test.cpp

@@ -325,6 +325,32 @@ class TestSizeBasedCumulativeCompactionPolicy : public testing::Test {
         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;
@@ -1085,6 +1111,64 @@ TEST_F(TestSizeBasedCumulativeCompactionPolicy, pick_input_rowsets_trim_after_pr
     EXPECT_EQ(100, input_rowsets.size());
 }
 
+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;