Take segments in singleton rowset into consideration upon cumulative compaction

be/src/olap/cumulative_compaction.cpp

@@ -78,29 +78,37 @@ OLAPStatus CumulativeCompaction::pick_rowsets_to_compact() {
     RETURN_NOT_OK(check_version_continuity(candidate_rowsets));
 
     std::vector<RowsetSharedPtr> transient_rowsets;
+    size_t num_overlapping_segments = 0;
     for (size_t i = 0; i < candidate_rowsets.size() - 1; ++i) {
         // VersionHash will calculated from chosen rowsets.
         // If ultimate singleton rowset is chosen, VersionHash
         // will be different from the value recorded in FE.
         // So the ultimate singleton rowset is revserved.
         RowsetSharedPtr rowset = candidate_rowsets[i];
         if (_tablet->version_for_delete_predicate(rowset->version())) {
-            if (transient_rowsets.size() > config::min_cumulative_compaction_num_singleton_deltas) {
+            if (num_overlapping_segments >= config::min_cumulative_compaction_num_singleton_deltas) {
                 _input_rowsets = transient_rowsets;
                 break;
             }
             transient_rowsets.clear();
+            num_overlapping_segments = 0;
             continue;
         }
 
-        if (transient_rowsets.size() >= config::max_cumulative_compaction_num_singleton_deltas) {
+        if (num_overlapping_segments >= config::max_cumulative_compaction_num_singleton_deltas) {
             // the threshold of files to compacted one time
             break;
         }
+
+        if (rowset->start_version() == rowset->end_version()) {
+            num_overlapping_segments += rowset->num_segments();
+        } else {
+            num_overlapping_segments++;
+        }
         transient_rowsets.push_back(rowset); 
     }
 
-    if (transient_rowsets.size() > config::min_cumulative_compaction_num_singleton_deltas) {
+    if (num_overlapping_segments >= config::min_cumulative_compaction_num_singleton_deltas) {
         _input_rowsets = transient_rowsets;
     }
 
@@ -113,7 +121,7 @@ OLAPStatus CumulativeCompaction::pick_rowsets_to_compact() {
 
     if (_input_rowsets.size() <= 1) {
         LOG(WARNING) << "There is no enough rowsets to cumulative compaction."
-                     << ", the size of rowsets to compact=" << candidate_rowsets.size()
+                     << " The size of rowsets to compact=" << candidate_rowsets.size()
                      << ", cumulative_point=" << _tablet->cumulative_layer_point();
         return OLAP_ERR_CUMULATIVE_NO_SUITABLE_VERSIONS;
     }

be/src/olap/tablet.cpp

@@ -609,6 +609,9 @@ const uint32_t Tablet::calc_cumulative_compaction_score() const {
         if (rs_meta->start_version() >= point) {
             score++;
         }
+        if (rs_meta->start_version() == rs_meta->end_version()) {
+            score += rs_meta->num_segments();
+        }
         if (rs_meta->start_version() == 0) {
             base_rowset_exist = true;
         }