[fix](cloud) Stabilize grouped single-rowset compaction

Issue Number: None

Related PR: #65907

Problem Summary: Single-rowset grouped compaction produces rowsets whose segments are non-overlapping only within recorded groups. Follow-up compaction, vertical reading, row ID conversion, cloud snapshot conversion, and schema change did not consistently preserve or interpret that layout. Repeated compaction also selected input by logical group count while planning ranges by physical segment count, so it could rewrite a rowset without reducing groups. Preserve and validate grouped metadata across cloud paths, read ordered segments lazily within each group, plan follow-up merges from logical group boundaries, use logical merge-way counts, and skip configurations that cannot make progress. Run the regression case non-concurrently only in cloud mode on the existing cloud cluster, temporarily disable SQL cache and mutable BE settings with guaranteed restoration, and use the configured MetaService endpoint. Reuse ReadSchema in the repeated-compaction unit test so it uses the current storage reader API.

None

- Test: No tests executed (targeted BE UT build was previously blocked during CMake configuration because OpenMP_C was unavailable; later runs were skipped per request); git diff --check passed
- Behavior changed: Yes (grouped rowsets are preserved and read correctly, repeated compaction reduces logical groups, and the cloud regression case is skipped outside cloud mode)
- Does this need documentation: No

Author: mymeiyi

be/src/cloud/cloud_cumulative_compaction.cpp

@@ -17,6 +17,8 @@
 
 #include "cloud/cloud_cumulative_compaction.h"
 
+#include <fmt/format.h>
+#include <fmt/ranges.h>
 #include <gen_cpp/cloud.pb.h>
 
 #include <random>
@@ -47,9 +49,13 @@ 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;
+    const auto& rowset_meta = rowset->rowset_meta();
+    const int64_t overlap_unit_count =
+            rowset_meta->segments_overlap() == NONOVERLAPPING_WITHIN_GROUP
+                    ? static_cast<int64_t>(rowset_meta->segment_group_sizes().size())
+                    : rowset->num_segments();
+    return !rowset_meta->has_delete_predicate() && rowset_meta->is_segments_overlapping() &&
+           overlap_unit_count >= config::cloud_single_rowset_compaction_min_segments;
 }
 
 bool should_use_single_rowset_grouped_compaction(const std::vector<RowsetSharedPtr>& input_rowsets,
@@ -61,6 +67,51 @@ bool should_use_single_rowset_grouped_compaction(const std::vector<RowsetSharedP
            is_single_rowset_compaction_candidate(input_rowsets.front());
 }
 
+std::vector<SegmentGroupMergeRange> build_segment_group_merge_ranges(const RowsetMeta& rowset_meta,
+                                                                     int64_t segment_group_size) {
+    DORIS_CHECK_GT(segment_group_size, 1);
+    DORIS_CHECK_GT(rowset_meta.num_segments(), 0);
+
+    std::vector<SegmentGroupMergeRange> ranges;
+    if (rowset_meta.segments_overlap() == NONOVERLAPPING_WITHIN_GROUP) {
+        const auto& input_segment_group_sizes = rowset_meta.segment_group_sizes();
+        const int64_t input_group_count = cast_set<int64_t>(input_segment_group_sizes.size());
+        DORIS_CHECK_GT(input_group_count, 0);
+        ranges.reserve(cast_set<size_t>((input_group_count + segment_group_size - 1) /
+                                        segment_group_size));
+
+        int64_t segment_end = 0;
+        for (int64_t group_start = 0; group_start < input_group_count;
+             group_start += segment_group_size) {
+            const int64_t group_end = std::min(group_start + segment_group_size, input_group_count);
+            const int64_t segment_start = segment_end;
+            for (int64_t group_index = group_start; group_index < group_end; ++group_index) {
+                const int32_t input_group_size =
+                        input_segment_group_sizes.Get(cast_set<int>(group_index));
+                DORIS_CHECK_GT(input_group_size, 0);
+                segment_end += input_group_size;
+            }
+
+            ranges.push_back({.segment_start = segment_start,
+                              .segment_end = segment_end,
+                              .merge_way_num = group_end - group_start});
+        }
+        DORIS_CHECK_EQ(segment_end, rowset_meta.num_segments());
+    } else {
+        ranges.reserve(cast_set<size_t>((rowset_meta.num_segments() + segment_group_size - 1) /
+                                        segment_group_size));
+        for (int64_t segment_start = 0; segment_start < rowset_meta.num_segments();
+             segment_start += segment_group_size) {
+            const int64_t segment_end =
+                    std::min(segment_start + segment_group_size, rowset_meta.num_segments());
+            ranges.push_back({.segment_start = segment_start,
+                              .segment_end = segment_end,
+                              .merge_way_num = segment_end - segment_start});
+        }
+    }
+    return ranges;
+}
+
 } // namespace cloud
 
 bvar::Adder<uint64_t> cumu_output_size("cumu_compaction", "output_size");
@@ -273,6 +324,18 @@ Status CloudCumulativeCompaction::execute_compact() {
     return st;
 }
 
+bool CloudCumulativeCompaction::should_calculate_new_cumulative_point(
+        int64_t input_cumulative_point) const {
+    if (!_single_rowset_compaction_segment_group_size.has_value()) {
+        return true;
+    }
+
+    DORIS_CHECK_EQ(_input_rowsets.size(), 1);
+    DORIS_CHECK(_output_rowset != nullptr);
+    return _input_rowsets.front()->start_version() == input_cumulative_point &&
+           _output_rowset->rowset_meta()->segments_overlap() == NONOVERLAPPING;
+}
+
 Status CloudCumulativeCompaction::modify_rowsets() {
     // calculate new cumulative point
     int64_t input_cumulative_point;
@@ -290,7 +353,7 @@ Status CloudCumulativeCompaction::modify_rowsets() {
     }
     auto compaction_policy = cloud_tablet()->tablet_meta()->compaction_policy();
     int64_t new_cumulative_point = input_cumulative_point;
-    if (!_single_rowset_compaction_segment_group_size.has_value()) {
+    if (should_calculate_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.
@@ -672,14 +735,15 @@ 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) {
+        const int64_t segment_group_size =
+                config::cloud_single_rowset_compaction_segment_group_size;
+        if (config::enable_cloud_single_rowset_compaction && segment_group_size > 1) {
             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;
+                    _single_rowset_compaction_segment_group_size = segment_group_size;
                     return Status::OK();
                 }
             }
@@ -754,11 +818,7 @@ Status CloudCumulativeCompaction::prepare_merge_input_rowsets(MergeInputRowsetsR
     }
 
     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);
-    }
+    DORIS_CHECK_GT(segment_group_size, 1);
     result->is_segment_grouped = true;
     result->segment_group_size = segment_group_size;
     return Status::OK();
@@ -773,12 +833,10 @@ Status CloudCumulativeCompaction::do_merge_input_rowsets(
 
     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 auto segment_ranges = cloud::build_segment_group_merge_ranges(
+            *input_rowset->rowset_meta(), segment_group_size);
+    for (size_t range_index = 0; range_index < segment_ranges.size(); ++range_index) {
+        const auto& range = segment_ranges[range_index];
         const int32_t output_segment_start = _output_rs_writer->get_allocated_segment_id();
 
         RowsetReaderSharedPtr rs_reader;
@@ -788,10 +846,10 @@ Status CloudCumulativeCompaction::do_merge_input_rowsets(
 
         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}));
+        RETURN_IF_ERROR(execute_merge(group_readers, range.merge_way_num, &group_stats,
+                                      std::make_pair(range.segment_start, range.segment_end),
+                                      {.total_ranges = cast_set<int64_t>(segment_ranges.size()),
+                                       .range_index = cast_set<int64_t>(range_index)}));
 
         _stats.output_rows += group_stats.output_rows;
         _stats.merged_rows += group_stats.merged_rows;
@@ -805,7 +863,7 @@ Status CloudCumulativeCompaction::do_merge_input_rowsets(
         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;
+            result->output_segment_group_sizes.push_back(output_group_size);
         }
     }
     return Status::OK();
@@ -816,8 +874,9 @@ void CloudCumulativeCompaction::update_output_rowset_after_build(
     if (!result.is_segment_grouped) {
         return;
     }
-    if (result.output_segment_group_count > 1) {
-        _output_rowset->rowset_meta()->set_segments_overlap(OVERLAPPING);
+    if (result.output_segment_group_sizes.size() > 1) {
+        _output_rowset->rowset_meta()->set_segments_overlap(NONOVERLAPPING_WITHIN_GROUP);
+        _output_rowset->rowset_meta()->set_segment_group_sizes(result.output_segment_group_sizes);
     }
 
     const auto& input_rowset = _input_rowsets.front();
@@ -827,7 +886,9 @@ void CloudCumulativeCompaction::update_output_rowset_after_build(
             .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);
+            .tag("output_groups", result.output_segment_group_sizes.size())
+            .tag("output_segment_group_sizes",
+                 fmt::format("[{}]", fmt::join(result.output_segment_group_sizes, ", ")));
 }
 
 void CloudCumulativeCompaction::update_cumulative_point(int64_t input_cumulative_point,

be/src/cloud/cloud_cumulative_compaction.h

@@ -31,14 +31,25 @@
 
 namespace doris {
 
+class RowsetMeta;
+
 namespace cloud {
 
+struct SegmentGroupMergeRange {
+    int64_t segment_start;
+    int64_t segment_end;
+    int64_t merge_way_num;
+};
+
 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);
 
+std::vector<SegmentGroupMergeRange> build_segment_group_merge_ranges(const RowsetMeta& rowset_meta,
+                                                                     int64_t segment_group_size);
+
 } // namespace cloud
 
 class CloudCumulativeCompaction : public CloudCompactionMixin {
@@ -73,6 +84,8 @@ class CloudCumulativeCompaction : public CloudCompactionMixin {
 
     void update_output_rowset_after_build(const MergeInputRowsetsResult& result) override;
 
+    bool should_calculate_new_cumulative_point(int64_t input_cumulative_point) const;
+
     std::string_view compaction_name() const override { return "CloudCumulativeCompaction"; }
 
 protected:

be/src/cloud/cloud_schema_change_job.cpp

@@ -389,7 +389,13 @@ Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParam
         context.txn_expiration = _expiration;
         context.version = rs_reader->version();
         context.rowset_state = VISIBLE;
-        context.segments_overlap = rs_reader->rowset()->rowset_meta()->segments_overlap();
+        const auto input_segments_overlap = rs_reader->rowset()->rowset_meta()->segments_overlap();
+        // Cloud schema change rewrites remote rowsets, so the input group layout is no longer
+        // applicable. Fall back to the conservative overlap state without assuming that the
+        // rewritten segments are globally ordered.
+        context.segments_overlap = input_segments_overlap == NONOVERLAPPING_WITHIN_GROUP
+                                           ? OVERLAPPING
+                                           : input_segments_overlap;
         context.tablet_schema = _new_tablet->tablet_schema();
         context.newest_write_timestamp = rs_reader->newest_write_timestamp();
         context.storage_resource = _cloud_storage_engine.get_storage_resource(sc_params.vault_id);

be/src/cloud/cloud_snapshot_mgr.cpp

@@ -272,6 +272,8 @@ Status CloudSnapshotMgr::_create_rowset_meta(
     new_rowset_meta_pb->set_num_segments(source_meta_pb.num_segments());
     new_rowset_meta_pb->mutable_segment_ids()->CopyFrom(source_meta_pb.segment_ids());
     new_rowset_meta_pb->set_rowset_state(source_meta_pb.rowset_state());
+    new_rowset_meta_pb->mutable_segment_group_sizes()->CopyFrom(
+            source_meta_pb.segment_group_sizes());
 
     new_rowset_meta_pb->clear_segments_key_bounds();
     for (const auto& key_bound : source_meta_pb.segments_key_bounds()) {

be/src/cloud/pb_convert.cpp

@@ -80,6 +80,7 @@ void doris_rowset_meta_to_cloud(RowsetMetaCloudPB* out, const RowsetMetaPB& in)
     }
     out->set_txn_expiration(in.txn_expiration());
     out->set_segments_overlap_pb(in.segments_overlap_pb());
+    out->mutable_segment_group_sizes()->CopyFrom(in.segment_group_sizes());
     if (in.has_segments_key_bounds_truncated()) {
         out->set_segments_key_bounds_truncated(in.segments_key_bounds_truncated());
     }
@@ -179,6 +180,7 @@ void doris_rowset_meta_to_cloud(RowsetMetaCloudPB* out, RowsetMetaPB&& in) {
     }
     out->set_txn_expiration(in.txn_expiration());
     out->set_segments_overlap_pb(in.segments_overlap_pb());
+    out->mutable_segment_group_sizes()->Swap(in.mutable_segment_group_sizes());
     if (in.has_segments_key_bounds_truncated()) {
         out->set_segments_key_bounds_truncated(in.segments_key_bounds_truncated());
     }
@@ -290,6 +292,7 @@ void cloud_rowset_meta_to_doris(RowsetMetaPB* out, const RowsetMetaCloudPB& in)
     }
     out->set_txn_expiration(in.txn_expiration());
     out->set_segments_overlap_pb(in.segments_overlap_pb());
+    out->mutable_segment_group_sizes()->CopyFrom(in.segment_group_sizes());
     if (in.has_segments_key_bounds_truncated()) {
         out->set_segments_key_bounds_truncated(in.segments_key_bounds_truncated());
     }
@@ -389,6 +392,7 @@ void cloud_rowset_meta_to_doris(RowsetMetaPB* out, RowsetMetaCloudPB&& in) {
     }
     out->set_txn_expiration(in.txn_expiration());
     out->set_segments_overlap_pb(in.segments_overlap_pb());
+    out->mutable_segment_group_sizes()->Swap(in.mutable_segment_group_sizes());
     if (in.has_segments_key_bounds_truncated()) {
         out->set_segments_key_bounds_truncated(in.segments_key_bounds_truncated());
     }

be/src/storage/compaction/compaction.h

@@ -106,7 +106,7 @@ class Compaction {
     struct MergeInputRowsetsResult {
         bool is_segment_grouped = false;
         int64_t segment_group_size = 0;
-        int64_t output_segment_group_count = 0;
+        std::vector<int32_t> output_segment_group_sizes;
     };
 
     Status merge_input_rowsets();