[feature](cloud) Support single rowset grouped compaction

Issue Number: None

Related PR: None

Problem Summary: Cloud cumulative compaction normally selects multiple rowsets, so a singleton rowset containing many overlapping segments can remain uncompacted. Add configurable selection and grouped merge support for eligible single rowsets, merging bounded segment ranges through the common compaction output lifecycle and preserving overlap semantics across non-empty output groups. Restrict the grouped path to size-based cumulative policy after normal policy filtering, exclude cluster-key tablets, retain the input cumulative point for this grouped path, and keep row ID conversion, vertical writer segment metadata, inverted index metadata, and vertical progress reporting consistent across all segment groups. Snapshot the grouped-mode decision and segment group size when input rowsets are selected so queued tasks preserve their selected execution semantics if dynamic settings change before merging begins. Add focused BE unit coverage and a cloud regression case for grouped compaction behavior.

Support configurable cloud cumulative compaction of a single overlapping rowset in bounded segment groups. Queued tasks preserve their selected grouped execution mode when dynamic settings change. Cluster-key tablets continue to use the existing compaction path.

- Test: Unit Test / Regression test
    - Added focused BE unit coverage and a cloud regression case.
    - Targeted CloudCompactionTest attempted with run-be-ut.sh; blocked during CMake configuration because OpenMP_C is unavailable.
- Behavior changed: Yes. Eligible cloud single rowsets with overlapping segments can be compacted in bounded groups while retaining their previous cumulative point and task-local grouped settings.
- Does this need documentation: No

Author: mymeiyi

be/src/cloud/cloud_cumulative_compaction.cpp

@@ -26,19 +26,43 @@
 #include "cloud/config.h"
 #include "common/config.h"
 #include "common/logging.h"
+#include "common/metrics/doris_metrics.h"
 #include "common/status.h"
 #include "cpp/sync_point.h"
 #include "service/backend_options.h"
 #include "storage/compaction/compaction.h"
 #include "storage/compaction/cumulative_compaction_policy.h"
 #include "storage/compaction/cumulative_compaction_time_series_policy.h"
+#include "storage/merger.h"
+#include "storage/rowset/rowset_reader.h"
+#include "storage/rowset/rowset_writer.h"
+#include "storage/tablet/tablet_schema.h"
 #include "util/debug_points.h"
 #include "util/trace.h"
 #include "util/uuid_generator.h"
 
 namespace doris {
 using namespace ErrorCode;
 
+namespace cloud {
+
+bool is_single_rowset_compaction_candidate(const RowsetSharedPtr& rowset) {
+    return !rowset->rowset_meta()->has_delete_predicate() &&
+           rowset->rowset_meta()->is_segments_overlapping() &&
+           rowset->num_segments() >= config::cloud_single_rowset_compaction_min_segments;
+}
+
+bool should_use_single_rowset_grouped_compaction(const std::vector<RowsetSharedPtr>& input_rowsets,
+                                                 const TabletSchema& tablet_schema,
+                                                 std::string_view compaction_policy) {
+    return compaction_policy == CUMULATIVE_SIZE_BASED_POLICY &&
+           tablet_schema.num_key_columns() > 0 && tablet_schema.cluster_key_uids().empty() &&
+           config::enable_cloud_single_rowset_compaction && input_rowsets.size() == 1 &&
+           is_single_rowset_compaction_candidate(input_rowsets.front());
+}
+
+} // namespace cloud
+
 bvar::Adder<uint64_t> cumu_output_size("cumu_compaction", "output_size");
 bvar::LatencyRecorder g_cu_compaction_hold_delete_bitmap_lock_time_ms(
         "cu_compaction_hold_delete_bitmap_lock_time_ms");
@@ -266,17 +290,19 @@ Status CloudCumulativeCompaction::modify_rowsets() {
     }
     auto compaction_policy = cloud_tablet()->tablet_meta()->compaction_policy();
     int64_t new_cumulative_point = input_cumulative_point;
-    if (!_enable_parallel_cumu_compaction && input_tablet_state == TABLET_NOTREADY &&
-        _output_rowset->start_version() > input_cumulative_point) {
-        // Historical rowsets are absent from a schema-change target until conversion finishes.
-        DORIS_CHECK_LE(input_cumulative_point, input_alter_version);
-        DORIS_CHECK_GT(_output_rowset->start_version(), input_alter_version);
-    } else if (!_enable_parallel_cumu_compaction ||
-               _output_rowset->start_version() == input_cumulative_point) {
-        new_cumulative_point =
-                _engine.cumu_compaction_policy(compaction_policy)
-                        ->new_cumulative_point(cloud_tablet(), _output_rowset, _last_delete_version,
-                                               input_cumulative_point);
+    if (!_single_rowset_compaction_segment_group_size.has_value()) {
+        if (!_enable_parallel_cumu_compaction && input_tablet_state == TABLET_NOTREADY &&
+            _output_rowset->start_version() > input_cumulative_point) {
+            // Historical rowsets are absent from a schema-change target until conversion finishes.
+            DORIS_CHECK_LE(input_cumulative_point, input_alter_version);
+            DORIS_CHECK_GT(_output_rowset->start_version(), input_alter_version);
+        } else if (!_enable_parallel_cumu_compaction ||
+                   _output_rowset->start_version() == input_cumulative_point) {
+            new_cumulative_point =
+                    _engine.cumu_compaction_policy(compaction_policy)
+                            ->new_cumulative_point(cloud_tablet(), _output_rowset,
+                                                   _last_delete_version, input_cumulative_point);
+        }
     }
     // commit compaction job
     cloud::TabletJobInfoPB job;
@@ -607,6 +633,7 @@ Status CloudCumulativeCompaction::advance_cumulative_point_before_pick(
 
 Status CloudCumulativeCompaction::pick_rowsets_to_compact() {
     _input_rowsets.clear();
+    _single_rowset_compaction_segment_group_size.reset();
 
     int64_t min_conflict_version = _min_conflict_version;
     int64_t max_conflict_version = _max_conflict_version;
@@ -645,6 +672,19 @@ Status CloudCumulativeCompaction::pick_rowsets_to_compact() {
                                      config::cumulative_compaction_min_deltas, &_input_rowsets,
                                      &_last_delete_version, &compaction_score);
 
+        if (config::enable_cloud_single_rowset_compaction) {
+            for (const auto& rowset : _input_rowsets) {
+                if (cloud::should_use_single_rowset_grouped_compaction(
+                            {rowset}, *cloud_tablet()->tablet_schema(), compaction_policy)) {
+                    auto grouped_input_rowset = rowset;
+                    _input_rowsets = {std::move(grouped_input_rowset)};
+                    _single_rowset_compaction_segment_group_size =
+                            config::cloud_single_rowset_compaction_segment_group_size;
+                    return Status::OK();
+                }
+            }
+        }
+
         if (_input_rowsets.empty()) {
             return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
                     "no suitable versions: input rowsets empty");
@@ -708,6 +748,88 @@ Status CloudCumulativeCompaction::pick_rowsets_to_compact() {
     return Status::OK();
 }
 
+Status CloudCumulativeCompaction::prepare_merge_input_rowsets(MergeInputRowsetsResult* result) {
+    if (!_single_rowset_compaction_segment_group_size.has_value()) {
+        return Status::OK();
+    }
+
+    const int64_t segment_group_size = *_single_rowset_compaction_segment_group_size;
+    if (segment_group_size <= 0) {
+        return Status::InvalidArgument(
+                "cloud_single_rowset_compaction_segment_group_size must be positive, value={}",
+                segment_group_size);
+    }
+    result->is_segment_grouped = true;
+    result->segment_group_size = segment_group_size;
+    return Status::OK();
+}
+
+Status CloudCumulativeCompaction::do_merge_input_rowsets(
+        const std::vector<RowsetReaderSharedPtr>& input_rs_readers,
+        MergeInputRowsetsResult* result) {
+    if (!result->is_segment_grouped) {
+        return Compaction::do_merge_input_rowsets(input_rs_readers, result);
+    }
+
+    const int64_t segment_group_size = result->segment_group_size;
+    const auto& input_rowset = _input_rowsets.front();
+    const int64_t segment_group_count =
+            (input_rowset->num_segments() + segment_group_size - 1) / segment_group_size;
+    for (int64_t segment_start = 0; segment_start < input_rowset->num_segments();
+         segment_start += segment_group_size) {
+        const int64_t segment_end =
+                std::min(segment_start + segment_group_size, input_rowset->num_segments());
+        const int32_t output_segment_start = _output_rs_writer->get_allocated_segment_id();
+
+        RowsetReaderSharedPtr rs_reader;
+        RETURN_IF_ERROR(input_rowset->create_reader(&rs_reader));
+        std::vector<RowsetReaderSharedPtr> group_readers;
+        group_readers.push_back(std::move(rs_reader));
+
+        Merger::Statistics group_stats;
+        group_stats.rowid_conversion = _stats.rowid_conversion;
+        RETURN_IF_ERROR(execute_merge(group_readers, segment_end - segment_start, &group_stats,
+                                      std::make_pair(segment_start, segment_end),
+                                      {.total_ranges = segment_group_count,
+                                       .range_index = segment_start / segment_group_size}));
+
+        _stats.output_rows += group_stats.output_rows;
+        _stats.merged_rows += group_stats.merged_rows;
+        _stats.filtered_rows += group_stats.filtered_rows;
+        _stats.bytes_read_from_local += group_stats.bytes_read_from_local;
+        _stats.bytes_read_from_remote += group_stats.bytes_read_from_remote;
+        _stats.cached_bytes_total += group_stats.cached_bytes_total;
+        _stats.cloud_local_read_time += group_stats.cloud_local_read_time;
+        _stats.cloud_remote_read_time += group_stats.cloud_remote_read_time;
+
+        const int32_t output_segment_end = _output_rs_writer->get_allocated_segment_id();
+        const int32_t output_group_size = output_segment_end - output_segment_start;
+        if (output_group_size > 0) {
+            ++result->output_segment_group_count;
+        }
+    }
+    return Status::OK();
+}
+
+void CloudCumulativeCompaction::update_output_rowset_after_build(
+        const MergeInputRowsetsResult& result) {
+    if (!result.is_segment_grouped) {
+        return;
+    }
+    if (result.output_segment_group_count > 1) {
+        _output_rowset->rowset_meta()->set_segments_overlap(OVERLAPPING);
+    }
+
+    const auto& input_rowset = _input_rowsets.front();
+    LOG_INFO("finish single rowset grouped compaction, tablet_id={}, version=[{}-{}]",
+             _tablet->tablet_id(), input_rowset->start_version(), input_rowset->end_version())
+            .tag("job_id", _uuid)
+            .tag("input_segments", input_rowset->num_segments())
+            .tag("segment_group_size", result.segment_group_size)
+            .tag("output_segments", _output_rowset->num_segments())
+            .tag("output_groups", result.output_segment_group_count);
+}
+
 void CloudCumulativeCompaction::update_cumulative_point(int64_t input_cumulative_point,
                                                         int64_t output_cumulative_point) {
     DORIS_CHECK_LT(input_cumulative_point, output_cumulative_point);

be/src/cloud/cloud_cumulative_compaction.h

@@ -20,14 +20,27 @@
 #include <limits>
 #include <memory>
 #include <optional>
+#include <string_view>
+#include <vector>
 
 #include "cloud/cloud_storage_engine.h"
 #include "cloud/cloud_tablet.h"
 #include "storage/compaction/compaction.h"
 #include "storage/compaction_task_tracker.h"
+#include "storage/tablet/tablet_fwd.h"
 
 namespace doris {
 
+namespace cloud {
+
+bool is_single_rowset_compaction_candidate(const RowsetSharedPtr& rowset);
+
+bool should_use_single_rowset_grouped_compaction(const std::vector<RowsetSharedPtr>& input_rowsets,
+                                                 const TabletSchema& tablet_schema,
+                                                 std::string_view compaction_policy);
+
+} // namespace cloud
+
 class CloudCumulativeCompaction : public CloudCompactionMixin {
 public:
     CloudCumulativeCompaction(CloudStorageEngine& engine, CloudTabletSPtr tablet);
@@ -53,6 +66,13 @@ class CloudCumulativeCompaction : public CloudCompactionMixin {
 
     Status pick_rowsets_to_compact();
 
+    Status prepare_merge_input_rowsets(MergeInputRowsetsResult* result) override;
+
+    Status do_merge_input_rowsets(const std::vector<RowsetReaderSharedPtr>& input_rs_readers,
+                                  MergeInputRowsetsResult* result) override;
+
+    void update_output_rowset_after_build(const MergeInputRowsetsResult& result) override;
+
     std::string_view compaction_name() const override { return "CloudCumulativeCompaction"; }
 
 protected:
@@ -75,6 +95,7 @@ class CloudCumulativeCompaction : public CloudCompactionMixin {
     int64_t _cumulative_compaction_cnt = 0;
     int64_t _picked_cumulative_point = 0;
     Version _last_delete_version {-1, -1};
+    std::optional<int64_t> _single_rowset_compaction_segment_group_size;
 };
 
 } // namespace doris

be/src/cloud/config.cpp

@@ -55,6 +55,9 @@ DEFINE_mInt32(max_base_compaction_task_num_per_disk, "2");
 DEFINE_mBool(prioritize_query_perf_in_compaction, "false");
 DEFINE_mInt32(compaction_max_rowset_count, "10000");
 DEFINE_mInt64(compaction_txn_max_size_bytes, "7340032"); // 7MB
+DEFINE_mBool(enable_cloud_single_rowset_compaction, "false");
+DEFINE_mInt32(cloud_single_rowset_compaction_min_segments, "512");
+DEFINE_mInt32(cloud_single_rowset_compaction_segment_group_size, "64");
 
 DEFINE_mInt32(refresh_s3_info_interval_s, "60");
 DEFINE_mInt32(vacuum_stale_rowsets_interval_s, "300");

be/src/cloud/config.h

@@ -96,6 +96,9 @@ DECLARE_mInt32(max_base_compaction_task_num_per_disk);
 DECLARE_mBool(prioritize_query_perf_in_compaction);
 DECLARE_mInt32(compaction_max_rowset_count);
 DECLARE_mInt64(compaction_txn_max_size_bytes);
+DECLARE_mBool(enable_cloud_single_rowset_compaction);
+DECLARE_mInt32(cloud_single_rowset_compaction_min_segments);
+DECLARE_mInt32(cloud_single_rowset_compaction_segment_group_size);
 
 // CloudStorageEngine config
 DECLARE_mInt32(refresh_s3_info_interval_s);

be/src/storage/compaction/compaction.cpp

@@ -282,6 +282,9 @@ int64_t Compaction::merge_way_num() {
 }
 
 Status Compaction::merge_input_rowsets() {
+    MergeInputRowsetsResult result;
+    RETURN_IF_ERROR(prepare_merge_input_rowsets(&result));
+
     std::vector<RowsetReaderSharedPtr> input_rs_readers;
     input_rs_readers.reserve(_input_rowsets.size());
     for (auto& rowset : _input_rowsets) {
@@ -307,38 +310,10 @@ Status Compaction::merge_input_rowsets() {
         _stats.rowid_conversion = _rowid_conversion.get();
     }
 
-    int64_t way_num = merge_way_num();
-
-    Status res;
     {
         SCOPED_TIMER(_merge_rowsets_latency_timer);
         // 1. Merge segment files and write bkd inverted index
-        // TODO implement vertical compaction for seq map
-        if (_is_vertical && !_tablet->tablet_schema()->has_seq_map()) {
-            if (!_tablet->tablet_schema()->cluster_key_uids().empty()) {
-                RETURN_IF_ERROR(update_delete_bitmap());
-            }
-            auto progress_cb = [compaction_id = this->_compaction_id](int64_t total,
-                                                                      int64_t completed) {
-                CompactionTaskTracker::instance()->update_progress(compaction_id, total, completed);
-            };
-            res = Merger::vertical_merge_rowsets(_tablet, compaction_type(), *_cur_tablet_schema,
-                                                 input_rs_readers, _output_rs_writer.get(),
-                                                 cast_set<uint32_t>(get_avg_segment_rows()),
-                                                 way_num, &_stats, progress_cb);
-        } else {
-            if (!_tablet->tablet_schema()->cluster_key_uids().empty()) {
-                return Status::InternalError(
-                        "mow table with cluster keys does not support non vertical compaction");
-            }
-            res = Merger::vmerge_rowsets(_tablet, compaction_type(), *_cur_tablet_schema,
-                                         input_rs_readers, _output_rs_writer.get(), &_stats);
-        }
-
-        _tablet->last_compaction_status = res;
-        if (!res.ok()) {
-            return res;
-        }
+        RETURN_IF_ERROR(do_merge_input_rowsets(input_rs_readers, &result));
         // 2. Merge the remaining inverted index files of the string type
         RETURN_IF_ERROR(do_inverted_index_compaction());
     }
@@ -364,6 +339,7 @@ Status Compaction::merge_input_rowsets() {
 
     //RETURN_IF_ERROR(_engine.meta_mgr().commit_rowset(*_output_rowset->rowset_meta().get()));
     set_delete_predicate_for_output_rowset();
+    update_output_rowset_after_build(result);
 
     _local_read_bytes_total = _stats.bytes_read_from_local;
     _remote_read_bytes_total = _stats.bytes_read_from_remote;
@@ -380,6 +356,46 @@ Status Compaction::merge_input_rowsets() {
     return check_correctness();
 }
 
+Status Compaction::do_merge_input_rowsets(
+        const std::vector<RowsetReaderSharedPtr>& input_rs_readers,
+        MergeInputRowsetsResult* /*result*/) {
+    return execute_merge(input_rs_readers, merge_way_num(), &_stats);
+}
+
+Status Compaction::execute_merge(const std::vector<RowsetReaderSharedPtr>& input_rs_readers,
+                                 int64_t merge_way_num, Merger::Statistics* stats,
+                                 std::optional<std::pair<int64_t, int64_t>> segment_range,
+                                 VerticalMergeProgressContext progress) {
+    Status status;
+    // TODO implement vertical compaction for seq map
+    if (_is_vertical && !_tablet->tablet_schema()->has_seq_map()) {
+        if (!_tablet->tablet_schema()->cluster_key_uids().empty() && !segment_range.has_value()) {
+            RETURN_IF_ERROR(update_delete_bitmap());
+        }
+        auto progress_cb = [compaction_id = this->_compaction_id, progress](int64_t total,
+                                                                            int64_t completed) {
+            CompactionTaskTracker::instance()->update_progress(
+                    compaction_id, total * progress.total_ranges,
+                    total * progress.range_index + completed);
+        };
+        status = Merger::vertical_merge_rowsets(_tablet, compaction_type(), *_cur_tablet_schema,
+                                                input_rs_readers, _output_rs_writer.get(),
+                                                cast_set<uint32_t>(get_avg_segment_rows()),
+                                                merge_way_num, stats, progress_cb, segment_range);
+    } else {
+        if (!_tablet->tablet_schema()->cluster_key_uids().empty()) {
+            return Status::InternalError(
+                    "mow table with cluster keys does not support non vertical compaction");
+        }
+        status = Merger::vmerge_rowsets(_tablet, compaction_type(), *_cur_tablet_schema,
+                                        input_rs_readers, _output_rs_writer.get(), stats,
+                                        segment_range);
+    }
+
+    _tablet->last_compaction_status = status;
+    return status;
+}
+
 void Compaction::set_delete_predicate_for_output_rowset() {
     // Now we support delete in cumu compaction, to make all data in rowsets whose version
     // is below output_version to be delete in the future base compaction, we should carry