diff --git a/be/src/cloud/cloud_cumulative_compaction_policy.cpp b/be/src/cloud/cloud_cumulative_compaction_policy.cpp index 834b9b682ea27d..a78f0cd0a510c8 100644 --- a/be/src/cloud/cloud_cumulative_compaction_policy.cpp +++ b/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(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/cloud/cloud_storage_engine.cpp b/be/src/cloud/cloud_storage_engine.cpp index 803a738d6dafa0..76b9c9d87855e0 100644 --- a/be/src/cloud/cloud_storage_engine.cpp +++ b/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() && !debug_log_enabled) { + LOG_EVERY_N(INFO, 100) << "failed to submit cumulative compaction task for tablet: " + << tablet_id << ", err: " << status; + return; + } + if (!status.is() || 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()), @@ -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() && - !status.is()) || - 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 { diff --git a/be/src/storage/compaction/cumulative_compaction_policy.cpp b/be/src/storage/compaction/cumulative_compaction_policy.cpp index 4ab5970959096a..fc6d97ac447848 100644 --- a/be/src/storage/compaction/cumulative_compaction_policy.cpp +++ b/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(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 29358eaaee0901..a5b9d3d0a6e904 100644 --- a/be/test/cloud/cloud_cumulative_compaction_policy_test.cpp +++ b/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 create_max_score_trim_candidates(bool include_stranded_head) { + std::vector 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 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 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 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 b0573ead66b984..cf09f0045c14b3 100644 --- a/be/test/storage/compaction/cumulative_compaction_policy_test.cpp +++ b/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 create_rs_meta_max_score_trim(bool include_stranded_head) { + std::vector 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(_tablet_meta->add_rs_meta(rowset)); + } + + TabletSharedPtr tablet( + new Tablet(_engine, _tablet_meta, nullptr, CUMULATIVE_SIZE_BASED_POLICY)); + static_cast(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 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(_tablet_meta->add_rs_meta(rowset)); + } + + TabletSharedPtr tablet( + new Tablet(_engine, _tablet_meta, nullptr, CUMULATIVE_SIZE_BASED_POLICY)); + static_cast(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 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 rs_metas;