[Optimize] Take 'tablet scan frequency' into consideration when selecting a tablet for compaction

be/src/common/config.h

@@ -329,6 +329,13 @@ namespace config {
     CONF_mInt32(base_compaction_trace_threshold, "10");
     CONF_mInt32(cumulative_compaction_trace_threshold, "2");
 
+    // time interval to record tablet scan count in second for the purpose of calculating tablet scan frequency
+    CONF_mInt64(tablet_scan_frequency_time_node_interval_second, "300");
+    // coefficient for tablet scan frequency and compaction score when finding a tablet for compaction
+    CONF_mInt32(compaction_tablet_scan_frequency_factor, "0");
+    CONF_mInt32(compaction_tablet_compaction_score_factor, "1");
+
+
     // Port to start debug webserver on
     CONF_Int32(webserver_port, "8040");
     // Number of webserver workers

be/src/exec/olap_scanner.cpp

@@ -497,6 +497,7 @@ void OlapScanner::update_counter() {
 
     _tablet->query_scan_bytes->increment(_compressed_bytes_read);
     _tablet->query_scan_rows->increment(_raw_rows_read);
+    _tablet->query_scan_count->increment(1);
 
     _has_update_counter = true;
 }

be/src/olap/base_tablet.cpp

@@ -26,6 +26,7 @@ namespace doris {
 
 extern MetricPrototype METRIC_query_scan_bytes;
 extern MetricPrototype METRIC_query_scan_rows;
+extern MetricPrototype METRIC_query_scan_count;
 
 BaseTablet::BaseTablet(TabletMetaSharedPtr tablet_meta, DataDir* data_dir)
         : _state(tablet_meta->tablet_state()),
@@ -45,6 +46,7 @@ BaseTablet::BaseTablet(TabletMetaSharedPtr tablet_meta, DataDir* data_dir)
         MetricEntityType::kTablet);
     INT_COUNTER_METRIC_REGISTER(_metric_entity, query_scan_bytes);
     INT_COUNTER_METRIC_REGISTER(_metric_entity, query_scan_rows);
+    INT_COUNTER_METRIC_REGISTER(_metric_entity, query_scan_count);
 }
 
 BaseTablet::~BaseTablet() {

be/src/olap/base_tablet.h

@@ -82,6 +82,7 @@ class BaseTablet : public std::enable_shared_from_this<BaseTablet> {
 public:
     IntCounter* query_scan_bytes;
     IntCounter* query_scan_rows;
+    IntCounter* query_scan_count;
 
 private:
     DISALLOW_COPY_AND_ASSIGN(BaseTablet);

be/src/olap/tablet.cpp

@@ -66,7 +66,9 @@ Tablet::Tablet(TabletMetaSharedPtr tablet_meta, DataDir* data_dir,
         _last_cumu_compaction_success_millis(0),
         _last_base_compaction_success_millis(0),
         _cumulative_point(K_INVALID_CUMULATIVE_POINT),
-        _cumulative_compaction_type(cumulative_compaction_type) {
+        _cumulative_compaction_type(cumulative_compaction_type),
+        _last_record_scan_count(0),
+        _last_record_scan_count_timestamp(time(nullptr)) {
     // construct _timestamped_versioned_tracker from rs and stale rs meta
     _timestamped_version_tracker.construct_versioned_tracker(_tablet_meta->all_rs_metas(),
                                                              _tablet_meta->all_stale_rs_metas());
@@ -1311,4 +1313,16 @@ void Tablet::generate_tablet_meta_copy_unlocked(TabletMetaSharedPtr new_tablet_m
     new_tablet_meta->init_from_pb(tablet_meta_pb);
 }
 
+double Tablet::calculate_scan_frequency() {
+    time_t now = time(nullptr);
+    int64_t current_count = query_scan_count->value();
+    double interval = difftime(now, _last_record_scan_count_timestamp);
+    double scan_frequency = (current_count - _last_record_scan_count) * 60 / interval;
+    if (interval >= config::tablet_scan_frequency_time_node_interval_second) {
+        _last_record_scan_count = current_count;
+        _last_record_scan_count_timestamp = now;
+    }
+    return scan_frequency;
+}
+
 }  // namespace doris

be/src/olap/tablet.h

@@ -232,6 +232,8 @@ class Tablet : public BaseTablet {
     // return a json string to show the compaction status of this tablet
     void get_compaction_status(std::string* json_result);
 
+    double calculate_scan_frequency();
+
 private:
     OLAPStatus _init_once_action();
     void _print_missed_versions(const std::vector<Version>& missed_versions) const;
@@ -302,6 +304,15 @@ class Tablet : public BaseTablet {
     // cumulative compaction policy
     std::unique_ptr<CumulativeCompactionPolicy> _cumulative_compaction_policy;
     std::string _cumulative_compaction_type;
+
+    // the value of metric 'query_scan_count' and timestamp will be recorded when every time
+    // 'config::tablet_scan_frequency_time_node_interval_second' passed to calculate tablet
+    // scan frequency.
+    // the value of metric 'query_scan_count' for the last record.
+    int64_t _last_record_scan_count;
+    // the timestamp of the last record.
+    time_t _last_record_scan_count_timestamp;
+
     DISALLOW_COPY_AND_ASSIGN(Tablet);
 
 public:

be/src/olap/tablet_manager.cpp

@@ -684,7 +684,9 @@ TabletSharedPtr TabletManager::find_best_tablet_to_compaction(
         vector<TTabletId> &tablet_submitted_compaction) {
     int64_t now_ms = UnixMillis();
     const string& compaction_type_str = compaction_type == CompactionType::BASE_COMPACTION ? "base" : "cumulative";
-    uint32_t highest_score = 0;
+    double highest_score = 0.0;
+    uint32_t compaction_score = 0;
+    double tablet_scan_frequency = 0.0;
     TabletSharedPtr best_tablet;
     for (int32 i = 0; i < _tablet_map_lock_shard_size; i++) {
         ReadLock tablet_map_rdlock(&_tablet_map_lock_array[i]);
@@ -745,9 +747,18 @@ TabletSharedPtr TabletManager::find_best_tablet_to_compaction(
                     }
                 }
 
-                uint32_t table_score = tablet_ptr->calc_compaction_score(compaction_type);
-                if (table_score > highest_score) {
-                    highest_score = table_score;
+                uint32_t current_compaction_score = tablet_ptr->calc_compaction_score(compaction_type);
+
+                double scan_frequency = 0.0;
+                if (config::compaction_tablet_scan_frequency_factor != 0) {
+                    scan_frequency = tablet_ptr->calculate_scan_frequency();
+                }
+
+                double tablet_score = config::compaction_tablet_scan_frequency_factor * scan_frequency + config::compaction_tablet_compaction_score_factor * current_compaction_score;
+                if (tablet_score > highest_score) {
+                    highest_score = tablet_score;
+                    compaction_score = current_compaction_score;
+                    tablet_scan_frequency = scan_frequency;
                     best_tablet = tablet_ptr;
                 }
             }
@@ -759,13 +770,15 @@ TabletSharedPtr TabletManager::find_best_tablet_to_compaction(
                   << "compaction_type=" << compaction_type_str
                   << ", tablet_id=" << best_tablet->tablet_id()
                   << ", path=" << data_dir->path()
+                  << ", compaction_score=" << compaction_score
+                  << ", tablet_scan_frequency=" << tablet_scan_frequency
                   << ", highest_score=" << highest_score;
         // TODO(lingbin): Remove 'max' from metric name, it would be misunderstood as the
         // biggest in history(like peak), but it is really just the value at current moment.
         if (compaction_type == CompactionType::BASE_COMPACTION) {
-            DorisMetrics::instance()->tablet_base_max_compaction_score->set_value(highest_score);
+            DorisMetrics::instance()->tablet_base_max_compaction_score->set_value(compaction_score);
         } else {
-            DorisMetrics::instance()->tablet_cumulative_max_compaction_score->set_value(highest_score);
+            DorisMetrics::instance()->tablet_cumulative_max_compaction_score->set_value(compaction_score);
         }
     }
     return best_tablet;

be/src/util/doris_metrics.cpp

@@ -32,6 +32,7 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(http_requests_total, MetricUnit::REQUESTS);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(http_request_send_bytes, MetricUnit::BYTES);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_bytes, MetricUnit::BYTES);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_rows, MetricUnit::ROWS);
+DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_count, MetricUnit::NOUNIT);
 DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(push_requests_success_total, MetricUnit::REQUESTS, "", push_requests_total, Labels({{"status", "SUCCESS"}}));
 DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(push_requests_fail_total, MetricUnit::REQUESTS, "", push_requests_total, Labels({{"status", "FAIL"}}));
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(push_request_duration_us, MetricUnit::MICROSECONDS);

docs/en/administrator-guide/config/be_config.md

@@ -186,6 +186,23 @@ User can set this configuration to a larger value to get better QPS performance.
 
 ### `column_dictionary_key_size_threshold`
 
+### `compaction_tablet_compaction_score_factor`
+
+* Type: int32
+* Description: Coefficient for compaction score when calculating tablet score to find a tablet for compaction.
+* Default value: 1
+
+### `compaction_tablet_scan_frequency_factor`
+
+* Type: int32
+* Description: Coefficient for tablet scan frequency when calculating tablet score to find a tablet for compaction.
+* Default value: 0
+
+Tablet scan frequency can be taken into consideration when selecting an tablet for compaction and preferentially do compaction for those tablets which are scanned frequently during a latest period of time at the present.
+Tablet score can be calculated like this:
+
+tablet_score = compaction_tablet_scan_frequency_factor * tablet_scan_frequency + compaction_tablet_scan_frequency_factor * compaction_score
+
 ### `compress_rowbatches`
 
 ### `create_tablet_worker_count`
@@ -612,6 +629,12 @@ Some data formats, such as JSON, cannot be split. Doris must read all the data i
 
 ### `tablet_meta_checkpoint_min_new_rowsets_num`
 
+### `tablet_scan_frequency_time_node_interval_second`
+
+* Type: int64
+* Description: Time interval to record the metric 'query_scan_count' and timestamp in second for the purpose of  calculating tablet scan frequency during a latest period of time at the present.
+* Default: 300
+
 ### `tablet_stat_cache_update_interval_second`
 
 ### `tablet_rowset_stale_sweep_time_sec`

docs/zh-CN/administrator-guide/config/be_config.md

@@ -180,6 +180,24 @@ Metrics: {"filtered_rows":0,"input_row_num":3346807,"input_rowsets_count":42,"in
 
 ### `column_dictionary_key_size_threshold`
 
+### `compaction_tablet_compaction_score_factor`
+
+* 类型：int32
+* 描述：选择tablet进行compaction时，计算 tablet score 的公式中 compaction score的权重。
+* 默认值：1
+
+### `compaction_tablet_scan_frequency_factor`
+
+* 类型：int32
+* 描述：选择tablet进行compaction时，计算 tablet score 的公式中 tablet scan frequency 的权重。
+* 默认值：0
+
+选择一个tablet执行compaction任务时，可以将tablet的scan频率作为一个选择依据，对当前最近一段时间频繁scan的tablet优先执行compaction。
+tablet score可以通过以下公式计算：
+
+tablet_score = compaction_tablet_scan_frequency_factor * tablet_scan_frequency + compaction_tablet_scan_frequency_factor * compaction_score
+
+
 ### `compress_rowbatches`
 
 ### `create_tablet_worker_count`
@@ -607,6 +625,12 @@ Stream Load 一般适用于导入几个GB以内的数据，不适合导入过大
 
 ### `tablet_meta_checkpoint_min_new_rowsets_num`
 
+### `tablet_scan_frequency_time_node_interval_second`
+
+* 类型：int64
+* 描述：用来表示记录 metric 'query_scan_count' 的时间间隔。为了计算当前一段时间的tablet的scan频率，需要每隔一段时间记录一次 metric 'query_scan_count'。
+* 默认值：300
+
 ### `tablet_stat_cache_update_interval_second`
 
 ### `tablet_rowset_stale_sweep_time_sec`