This is an automated email from the ASF dual-hosted git repository.
Yukang-Lian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ee0aa56f7e1 [improvement](compaction) split time series compaction
score metric (#63900)
ee0aa56f7e1 is described below
commit ee0aa56f7e1561c92a8532e60a7fde374dd224ec
Author: Jamie <[email protected]>
AuthorDate: Tue Aug 11 15:51:38 2026 +0800
[improvement](compaction) split time series compaction score metric (#63900)
### What problem does this PR solve?
Before this change, BE exposes only the aggregate cumulative compaction
score:
```bash
curl -s http://<be_host>:<webserver_port>/metrics \
| grep 'doris_be_tablet_cumulative_max_compaction_score'
```
The aggregate value cannot show which cumulative compaction policy
contributes the maximum score.
After this change, observe the aggregate and per-policy scores with:
```bash
curl -s http://<be_host>:<webserver_port>/metrics \
| grep -E
'doris_be_tablet_(cumulative|size_based|time_series)_max_compaction_score'
```
| Metric | Meaning |
| --- | --- |
| `doris_be_tablet_cumulative_max_compaction_score` | Maximum score
across all cumulative compaction policies; semantics unchanged |
| `doris_be_tablet_size_based_max_compaction_score` | Maximum score from
the size-based policy |
| `doris_be_tablet_time_series_max_compaction_score` | Maximum score
from the time-series policy |
New behavior:
- Local and Cloud BEs publish the two per-policy metrics.
- The existing aggregate metric and BE report value `max(base,
cumulative)` remain unchanged.
- Compaction score calculation, tablet ranking, and tablet selection are
unchanged.
- A complete score check can reset a missing policy score to zero; an
incomplete local scan caused by capacity-limited disks does not lower
the cumulative score metrics.
### Release note
Add separate size-based and time-series cumulative compaction score
metrics while preserving the existing aggregate metric.
---
be/src/cloud/cloud_storage_engine.cpp | 36 +++-
be/src/cloud/cloud_storage_engine.h | 5 +
be/src/cloud/cloud_tablet_mgr.cpp | 21 +-
be/src/cloud/cloud_tablet_mgr.h | 4 +-
be/src/common/metrics/doris_metrics.cpp | 4 +
be/src/common/metrics/doris_metrics.h | 6 +-
be/src/storage/data_dir.h | 7 +
be/src/storage/olap_common.h | 7 +
be/src/storage/olap_server.cpp | 84 +++++---
be/src/storage/storage_engine.h | 20 +-
be/src/storage/tablet/tablet_manager.cpp | 41 +++-
be/src/storage/tablet/tablet_manager.h | 3 +-
be/test/cloud/cloud_compaction_test.cpp | 172 +++++++++++++++-
be/test/storage/tablet/tablet_mgr_test.cpp | 316 ++++++++++++++++++++++++++++-
be/test/util/doris_metrics_test.cpp | 12 ++
15 files changed, 663 insertions(+), 75 deletions(-)
diff --git a/be/src/cloud/cloud_storage_engine.cpp
b/be/src/cloud/cloud_storage_engine.cpp
index 19514d647d3..a8cebc7449d 100644
--- a/be/src/cloud/cloud_storage_engine.cpp
+++ b/be/src/cloud/cloud_storage_engine.cpp
@@ -761,9 +761,13 @@ bool CloudStorageEngine::register_index_change_compaction(
std::vector<CloudTabletSPtr>
CloudStorageEngine::_generate_cloud_compaction_tasks(
CompactionType compaction_type, bool check_score) {
+ DCHECK(compaction_type == CompactionType::BASE_COMPACTION ||
+ compaction_type == CompactionType::CUMULATIVE_COMPACTION ||
+ compaction_type == CompactionType::CUMU_BINLOG_COMPACTION);
std::vector<std::shared_ptr<CloudTablet>> tablets_compaction;
- int64_t max_compaction_score = 0;
+ CompactionScoreStats score_stats;
+ bool got_score_stats = false;
std::unordered_set<int64_t> tablet_preparing_cumu_compaction;
std::unordered_map<int64_t,
std::vector<std::shared_ptr<CloudCumulativeCompaction>>>
submitted_cumu_compactions;
@@ -860,25 +864,37 @@ std::vector<CloudTabletSPtr>
CloudStorageEngine::_generate_cloud_compaction_task
do {
std::vector<CloudTabletSPtr> tablets;
auto st = tablet_mgr().get_topn_tablets_to_compact(n, compaction_type,
filter_out, &tablets,
-
&max_compaction_score);
+ &score_stats);
if (!st.ok()) {
LOG(WARNING) << "failed to get tablets to compact, err=" << st;
break;
}
+ got_score_stats = true;
if (!need_pick_tablet) break;
tablets_compaction = std::move(tablets);
} while (false);
- if (max_compaction_score > 0) {
- if (compaction_type == CompactionType::BASE_COMPACTION) {
+ if (got_score_stats && score_stats.scanned) {
+ if (compaction_type == CompactionType::BASE_COMPACTION &&
score_stats.max_score > 0) {
DorisMetrics::instance()->tablet_base_max_compaction_score->set_value(
- max_compaction_score);
- } else if (compaction_type == CompactionType::CUMU_BINLOG_COMPACTION) {
+ score_stats.max_score);
+ } else if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
+ if (check_score || score_stats.max_score > 0) {
+
DorisMetrics::instance()->tablet_cumulative_max_compaction_score->set_value(
+ score_stats.max_score);
+ }
+ if (check_score || score_stats.size_based_max_score > 0) {
+
DorisMetrics::instance()->tablet_size_based_max_compaction_score->set_value(
+ score_stats.size_based_max_score);
+ }
+ if (check_score || score_stats.time_series_max_score > 0) {
+
DorisMetrics::instance()->tablet_time_series_max_compaction_score->set_value(
+ score_stats.time_series_max_score);
+ }
+ } else if (compaction_type == CompactionType::CUMU_BINLOG_COMPACTION &&
+ score_stats.max_score > 0) {
DorisMetrics::instance()->tablet_binlog_max_compaction_score->set_value(
- max_compaction_score);
- } else {
-
DorisMetrics::instance()->tablet_cumulative_max_compaction_score->set_value(
- max_compaction_score);
+ score_stats.max_score);
}
}
diff --git a/be/src/cloud/cloud_storage_engine.h
b/be/src/cloud/cloud_storage_engine.h
index b3f4de73137..330850b5ed8 100644
--- a/be/src/cloud/cloud_storage_engine.h
+++ b/be/src/cloud/cloud_storage_engine.h
@@ -199,6 +199,11 @@ public:
void set_cloud_warm_up_manager(std::unique_ptr<CloudWarmUpManager>
manager);
void init_calc_delete_bitmap_executor_for_UT();
+
+ std::vector<CloudTabletSPtr> generate_cloud_compaction_tasks_for_test(
+ CompactionType compaction_type, bool check_score) {
+ return _generate_cloud_compaction_tasks(compaction_type, check_score);
+ }
#endif
private:
diff --git a/be/src/cloud/cloud_tablet_mgr.cpp
b/be/src/cloud/cloud_tablet_mgr.cpp
index b46ebdc0429..3affe632996 100644
--- a/be/src/cloud/cloud_tablet_mgr.cpp
+++ b/be/src/cloud/cloud_tablet_mgr.cpp
@@ -29,6 +29,7 @@
#include "common/status.h"
#include "cpp/sync_point.h"
#include "runtime/memory/cache_policy.h"
+#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "util/debug_points.h"
#include "util/lru_cache.h"
#include "util/stack_util.h"
@@ -432,11 +433,12 @@ void CloudTabletMgr::sync_tablets(const CountDownLatch&
stop_latch) {
Status CloudTabletMgr::get_topn_tablets_to_compact(
int n, CompactionType compaction_type, const
std::function<bool(CloudTablet*)>& filter_out,
- std::vector<std::shared_ptr<CloudTablet>>* tablets, int64_t*
max_score) {
+ std::vector<std::shared_ptr<CloudTablet>>* tablets,
CompactionScoreStats* score_stats) {
DCHECK(compaction_type == CompactionType::BASE_COMPACTION ||
compaction_type == CompactionType::CUMULATIVE_COMPACTION ||
compaction_type == CompactionType::CUMU_BINLOG_COMPACTION);
- *max_score = 0;
+ *score_stats = {};
+ score_stats->scanned = true;
int64_t max_score_tablet_id = 0;
// clang-format off
auto score = [compaction_type](CloudTablet* t) {
@@ -503,9 +505,18 @@ Status CloudTabletMgr::get_topn_tablets_to_compact(
int64_t s = score(t.get());
if (s <= 0) { continue; }
- if (s > *max_score) {
+ if (s > score_stats->max_score) {
max_score_tablet_id = t->tablet_id();
- *max_score = s;
+ score_stats->max_score = s;
+ }
+ if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
+ int64_t* policy_max_score =
+ t->tablet_meta()->compaction_policy() ==
CUMULATIVE_TIME_SERIES_POLICY
+ ? &score_stats->time_series_max_score
+ : &score_stats->size_based_max_score;
+ if (s > *policy_max_score) {
+ *policy_max_score = s;
+ }
}
if (filter_out(t.get())) { ++num_filtered; continue; }
@@ -520,7 +531,7 @@ Status CloudTabletMgr::get_topn_tablets_to_compact(
LOG_EVERY_N(INFO, 1000) << "get_topn_compaction_score, n=" << n << "
type=" << compaction_type
<< " num_tablets=" << weak_tablets.size() << " num_skipped=" <<
num_skipped
<< " num_disabled=" << num_disabled << " num_filtered=" <<
num_filtered
- << " max_score=" << *max_score << " max_score_tablet=" <<
max_score_tablet_id
+ << " max_score=" << score_stats->max_score << "
max_score_tablet=" << max_score_tablet_id
<< " tablets=[" << [&buf] { std::stringstream ss; for (auto& i
: buf) ss << i.first->tablet_id() << ":" << i.second << ","; return ss.str();
}() << "]"
;
// clang-format on
diff --git a/be/src/cloud/cloud_tablet_mgr.h b/be/src/cloud/cloud_tablet_mgr.h
index 9894d97552b..c44f4f36f8c 100644
--- a/be/src/cloud/cloud_tablet_mgr.h
+++ b/be/src/cloud/cloud_tablet_mgr.h
@@ -79,13 +79,13 @@ public:
* @param filter_out a filter takes a tablet and return bool to check
* whether skipping the tablet, true for skip
* @param tablets output param
- * @param max_score output param, max score of existed tablets
+ * @param score_stats output param, max scores of existed tablets
* @return status of this call
*/
Status get_topn_tablets_to_compact(int n, CompactionType compaction_type,
const
std::function<bool(CloudTablet*)>& filter_out,
std::vector<std::shared_ptr<CloudTablet>>* tablets,
- int64_t* max_score);
+ CompactionScoreStats* score_stats);
/**
* Gets tablets info and total tablet num that are reported
diff --git a/be/src/common/metrics/doris_metrics.cpp
b/be/src/common/metrics/doris_metrics.cpp
index 36bebcddbb3..3799a95da05 100644
--- a/be/src/common/metrics/doris_metrics.cpp
+++ b/be/src/common/metrics/doris_metrics.cpp
@@ -195,6 +195,8 @@
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(process_fd_num_limit_soft, MetricUnit::NOUNIT
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(process_fd_num_limit_hard,
MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(tablet_cumulative_max_compaction_score,
MetricUnit::NOUNIT);
+DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(tablet_size_based_max_compaction_score,
MetricUnit::NOUNIT);
+DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(tablet_time_series_max_compaction_score,
MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(tablet_base_max_compaction_score,
MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(tablet_binlog_max_compaction_score,
MetricUnit::NOUNIT);
@@ -392,6 +394,8 @@ DorisMetrics::DorisMetrics() :
_metric_registry(_s_registry_name) {
INT_GAUGE_METRIC_REGISTER(_server_metric_entity,
process_fd_num_limit_hard);
INT_GAUGE_METRIC_REGISTER(_server_metric_entity,
tablet_cumulative_max_compaction_score);
+ INT_GAUGE_METRIC_REGISTER(_server_metric_entity,
tablet_size_based_max_compaction_score);
+ INT_GAUGE_METRIC_REGISTER(_server_metric_entity,
tablet_time_series_max_compaction_score);
INT_GAUGE_METRIC_REGISTER(_server_metric_entity,
tablet_base_max_compaction_score);
INT_GAUGE_METRIC_REGISTER(_server_metric_entity,
tablet_binlog_max_compaction_score);
diff --git a/be/src/common/metrics/doris_metrics.h
b/be/src/common/metrics/doris_metrics.h
index aadda83fa86..2973435305a 100644
--- a/be/src/common/metrics/doris_metrics.h
+++ b/be/src/common/metrics/doris_metrics.h
@@ -165,9 +165,11 @@ public:
IntGauge* process_fd_num_limit_hard = nullptr;
// the max compaction score of all tablets.
- // Record base and cumulative scores separately, because
- // we need to get the larger of the two.
+ // Keep the cumulative score as the aggregate for compatibility, and record
+ // size-based and time-series cumulative scores separately.
IntGauge* tablet_cumulative_max_compaction_score = nullptr;
+ IntGauge* tablet_size_based_max_compaction_score = nullptr;
+ IntGauge* tablet_time_series_max_compaction_score = nullptr;
IntGauge* tablet_base_max_compaction_score = nullptr;
IntGauge* tablet_binlog_max_compaction_score = nullptr;
diff --git a/be/src/storage/data_dir.h b/be/src/storage/data_dir.h
index 4598f3d8771..f5a2c57f33f 100644
--- a/be/src/storage/data_dir.h
+++ b/be/src/storage/data_dir.h
@@ -141,6 +141,13 @@ public:
(double)_disk_capacity_bytes;
}
+#ifdef BE_TEST
+ void set_capacity_for_test(size_t disk_capacity_bytes, size_t
available_bytes) {
+ _disk_capacity_bytes = disk_capacity_bytes;
+ _available_bytes = available_bytes;
+ }
+#endif
+
// Move tablet to trash.
Status move_to_trash(const std::string& tablet_path);
diff --git a/be/src/storage/olap_common.h b/be/src/storage/olap_common.h
index 12ca1d2e6a4..51f23f93cd1 100644
--- a/be/src/storage/olap_common.h
+++ b/be/src/storage/olap_common.h
@@ -64,6 +64,13 @@ enum CompactionType {
CUMU_BINLOG_COMPACTION = 4
};
+struct CompactionScoreStats {
+ int64_t max_score = 0;
+ int64_t size_based_max_score = 0;
+ int64_t time_series_max_score = 0;
+ bool scanned = false;
+};
+
enum DataDirType {
SPILL_DISK_DIR,
OLAP_DATA_DIR,
diff --git a/be/src/storage/olap_server.cpp b/be/src/storage/olap_server.cpp
index 07143c265ff..07f179f6348 100644
--- a/be/src/storage/olap_server.cpp
+++ b/be/src/storage/olap_server.cpp
@@ -150,11 +150,12 @@ bool
CompactionSubmitRegistry::has_compaction_task(DataDir* dir, CompactionType
std::vector<TabletCompactionContext>
CompactionSubmitRegistry::pick_topn_tablets_for_compaction(
TabletManager* tablet_mgr, DataDir* data_dir, CompactionType
compaction_type,
- const CumuCompactionPolicyTable& cumu_compaction_policies, uint32_t*
disk_max_score) {
+ const CumuCompactionPolicyTable& cumu_compaction_policies,
+ CompactionScoreStats* disk_score_stats) {
// non-lock, used in snapshot
return tablet_mgr->find_best_tablets_to_compaction(compaction_type,
data_dir,
_get_tablet_set(data_dir, compaction_type),
- disk_max_score,
cumu_compaction_policies);
+ disk_score_stats,
cumu_compaction_policies);
}
bool CompactionSubmitRegistry::insert(TabletSharedPtr tablet, CompactionType
compaction_type) {
@@ -845,7 +846,7 @@ bool need_generate_compaction_tasks(int task_cnt_per_disk,
int thread_per_disk,
return true;
}
-int get_concurrent_per_disk(int max_score, int thread_per_disk) {
+int get_concurrent_per_disk(int64_t max_score, int thread_per_disk) {
if (!config::enable_compaction_priority_scheduling) {
return thread_per_disk;
}
@@ -887,12 +888,16 @@ bool has_free_compaction_slot(CompactionSubmitRegistry*
registry, DataDir* dir,
std::vector<TabletCompactionContext> StorageEngine::_generate_compaction_tasks(
CompactionType compaction_type, std::vector<DataDir*>& data_dirs, bool
check_score) {
+ DCHECK(compaction_type == CompactionType::BASE_COMPACTION ||
+ compaction_type == CompactionType::CUMULATIVE_COMPACTION ||
+ compaction_type == CompactionType::CUMU_BINLOG_COMPACTION);
TEST_SYNC_POINT_RETURN_WITH_VALUE("olap_server::_generate_compaction_tasks.return_empty",
std::vector<TabletCompactionContext> {});
_update_cumulative_compaction_policy();
auto cumulative_compaction_policies =
_snapshot_cumulative_compaction_policy();
std::vector<TabletCompactionContext> tablet_compaction_contexts;
- uint32_t max_compaction_score = 0;
+ CompactionScoreStats max_score_stats;
+ bool skipped_capacity_limited_dir = false;
std::random_device rd;
std::mt19937 g(rd());
@@ -916,38 +921,61 @@ std::vector<TabletCompactionContext>
StorageEngine::_generate_compaction_tasks(
// Even if need_pick_tablet is false, we still need to call
find_best_tablet_to_compaction(),
// So that we can update the max_compaction_score metric.
- if (!data_dir->reach_capacity_limit(0)) {
- uint32_t disk_max_score = 0;
- auto tablet_contexts =
compaction_registry_snapshot.pick_topn_tablets_for_compaction(
- _tablet_manager.get(), data_dir, compaction_type,
- cumulative_compaction_policies, &disk_max_score);
- int concurrent_num = get_concurrent_per_disk(
- disk_max_score, disk_compaction_slot_num(*data_dir,
compaction_type));
- need_pick_tablet = need_generate_compaction_tasks(
- executing_task_num, concurrent_num, compaction_type,
- !compaction_registry_snapshot.has_compaction_task(
- data_dir, CompactionType::CUMULATIVE_COMPACTION));
- for (const auto& context : tablet_contexts) {
- if (context.tablet != nullptr) {
- if (need_pick_tablet) {
- tablet_compaction_contexts.emplace_back(context);
- }
- max_compaction_score = std::max(max_compaction_score,
disk_max_score);
+ if (data_dir->reach_capacity_limit(0)) {
+ skipped_capacity_limited_dir = true;
+ continue;
+ }
+
+ CompactionScoreStats disk_score_stats;
+ auto tablet_contexts =
compaction_registry_snapshot.pick_topn_tablets_for_compaction(
+ _tablet_manager.get(), data_dir, compaction_type,
cumulative_compaction_policies,
+ &disk_score_stats);
+ max_score_stats.scanned = max_score_stats.scanned ||
disk_score_stats.scanned;
+ max_score_stats.max_score = std::max(max_score_stats.max_score,
disk_score_stats.max_score);
+ max_score_stats.size_based_max_score =
std::max(max_score_stats.size_based_max_score,
+
disk_score_stats.size_based_max_score);
+ max_score_stats.time_series_max_score =
std::max(max_score_stats.time_series_max_score,
+
disk_score_stats.time_series_max_score);
+ int concurrent_num = get_concurrent_per_disk(
+ disk_score_stats.max_score,
disk_compaction_slot_num(*data_dir, compaction_type));
+ need_pick_tablet = need_generate_compaction_tasks(
+ executing_task_num, concurrent_num, compaction_type,
+ !compaction_registry_snapshot.has_compaction_task(
+ data_dir, CompactionType::CUMULATIVE_COMPACTION));
+ for (const auto& context : tablet_contexts) {
+ if (context.tablet != nullptr) {
+ if (need_pick_tablet) {
+ tablet_compaction_contexts.emplace_back(context);
}
}
}
}
- if (max_compaction_score > 0) {
- if (compaction_type == CompactionType::BASE_COMPACTION) {
+ if (max_score_stats.scanned) {
+ if (compaction_type == CompactionType::BASE_COMPACTION &&
max_score_stats.max_score > 0) {
DorisMetrics::instance()->tablet_base_max_compaction_score->set_value(
- max_compaction_score);
+ max_score_stats.max_score);
} else if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
-
DorisMetrics::instance()->tablet_cumulative_max_compaction_score->set_value(
- max_compaction_score);
- } else if (compaction_type == CompactionType::CUMU_BINLOG_COMPACTION) {
+ auto update_policy_score = [skipped_capacity_limited_dir,
check_score](IntGauge* metric,
+
int64_t score) {
+ if (skipped_capacity_limited_dir) {
+ if (score > metric->value()) {
+ metric->set_value(score);
+ }
+ } else if (check_score || score > 0) {
+ metric->set_value(score);
+ }
+ };
+
update_policy_score(DorisMetrics::instance()->tablet_cumulative_max_compaction_score,
+ max_score_stats.max_score);
+
update_policy_score(DorisMetrics::instance()->tablet_size_based_max_compaction_score,
+ max_score_stats.size_based_max_score);
+
update_policy_score(DorisMetrics::instance()->tablet_time_series_max_compaction_score,
+ max_score_stats.time_series_max_score);
+ } else if (compaction_type == CompactionType::CUMU_BINLOG_COMPACTION &&
+ max_score_stats.max_score > 0) {
DorisMetrics::instance()->tablet_binlog_max_compaction_score->set_value(
- max_compaction_score);
+ max_score_stats.max_score);
}
}
return tablet_compaction_contexts;
diff --git a/be/src/storage/storage_engine.h b/be/src/storage/storage_engine.h
index 5530b09dae2..c503fc6cb31 100644
--- a/be/src/storage/storage_engine.h
+++ b/be/src/storage/storage_engine.h
@@ -233,7 +233,8 @@ public:
std::vector<TabletCompactionContext> pick_topn_tablets_for_compaction(
TabletManager* tablet_mgr, DataDir* data_dir, CompactionType
compaction_type,
- const CumuCompactionPolicyTable& cumu_compaction_policies,
uint32_t* disk_max_score);
+ const CumuCompactionPolicyTable& cumu_compaction_policies,
+ CompactionScoreStats* disk_score_stats);
private:
TabletSet& _get_tablet_set(DataDir* dir, CompactionType compaction_type);
@@ -378,6 +379,23 @@ public:
int64_t get_compaction_num_per_round() const { return
_compaction_num_per_round; }
+#ifdef BE_TEST
+ std::vector<TabletSharedPtr> generate_compaction_tasks_for_test(
+ CompactionType compaction_type, std::vector<DataDir*>& data_dirs,
bool check_score) {
+ auto tablet_contexts = _generate_compaction_tasks(compaction_type,
data_dirs, check_score);
+ std::vector<TabletSharedPtr> tablets;
+ tablets.reserve(tablet_contexts.size());
+ for (auto& context : tablet_contexts) {
+ tablets.emplace_back(std::move(context.tablet));
+ }
+ return tablets;
+ }
+
+ CompactionSubmitRegistry& compaction_submit_registry_for_test() {
+ return _compaction_submit_registry;
+ }
+#endif
+
private:
// Instance should be inited from `static open()`
// MUST NOT be called in other circumstances.
diff --git a/be/src/storage/tablet/tablet_manager.cpp
b/be/src/storage/tablet/tablet_manager.cpp
index cea5d672daa..dea24b37de6 100644
--- a/be/src/storage/tablet/tablet_manager.cpp
+++ b/be/src/storage/tablet/tablet_manager.cpp
@@ -30,6 +30,7 @@
#include <algorithm>
#include <list>
#include <mutex>
+#include <optional>
#include <ostream>
#include <string_view>
@@ -43,6 +44,7 @@
#include "io/fs/local_file_system.h"
#include "runtime/exec_env.h"
#include "service/backend_options.h"
+#include "storage/compaction/cumulative_compaction_policy.h"
#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "storage/data_dir.h"
#include "storage/olap_common.h"
@@ -730,9 +732,13 @@ struct TabletScore {
std::vector<TabletCompactionContext>
TabletManager::find_best_tablets_to_compaction(
CompactionType compaction_type, DataDir* data_dir,
- const std::unordered_set<TabletSharedPtr>&
tablet_submitted_compaction, uint32_t* score,
+ const std::unordered_set<TabletSharedPtr>& tablet_submitted_compaction,
+ CompactionScoreStats* score_stats,
const std::unordered_map<std::string_view,
std::shared_ptr<CumulativeCompactionPolicy>>&
all_cumulative_compaction_policies) {
+ DCHECK(score_stats != nullptr);
+ *score_stats = {};
+ score_stats->scanned = true;
int64_t now_ms = UnixMillis();
const string& compaction_type_str = compaction_type ==
CompactionType::BASE_COMPACTION ? "base"
: compaction_type ==
CompactionType::CUMU_BINLOG_COMPACTION
@@ -793,8 +799,9 @@ std::vector<TabletCompactionContext>
TabletManager::find_best_tablets_to_compact
return;
}
}
- auto cumulative_compaction_policy =
all_cumulative_compaction_policies.at(
- tablet_ptr->tablet_meta()->compaction_policy());
+ const auto& compaction_policy =
tablet_ptr->tablet_meta()->compaction_policy();
+ auto cumulative_compaction_policy =
+ all_cumulative_compaction_policies.at(compaction_policy);
uint32_t current_compaction_score =
tablet_ptr->calc_compaction_score(compaction_type);
if (current_compaction_score < 5) {
tablet_ptr->set_skip_compaction(true, compaction_type,
UnixSeconds());
@@ -804,6 +811,24 @@ std::vector<TabletCompactionContext>
TabletManager::find_best_tablets_to_compact
return;
}
+ std::optional<bool> suitable_for_compaction;
+ auto is_suitable = [&]() {
+ if (!suitable_for_compaction.has_value()) {
+ suitable_for_compaction = tablet_ptr->suitable_for_compaction(
+ compaction_type, cumulative_compaction_policy);
+ }
+ return suitable_for_compaction.value();
+ };
+
+ if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
+ int64_t* policy_max_score = compaction_policy ==
CUMULATIVE_TIME_SERIES_POLICY
+ ?
&score_stats->time_series_max_score
+ :
&score_stats->size_based_max_score;
+ if (current_compaction_score > *policy_max_score && is_suitable())
{
+ *policy_max_score = current_compaction_score;
+ }
+ }
+
if (compaction_num_per_round > 1) {
TabletScore ts;
ts.score = current_compaction_score;
@@ -811,9 +836,7 @@ std::vector<TabletCompactionContext>
TabletManager::find_best_tablets_to_compact
if ((top_tablets.size() >= compaction_num_per_round &&
current_compaction_score > top_tablets.top().score) ||
top_tablets.size() < compaction_num_per_round) {
- bool ret = tablet_ptr->suitable_for_compaction(compaction_type,
-
cumulative_compaction_policy);
- if (ret) {
+ if (is_suitable()) {
top_tablets.push(ts);
if (top_tablets.size() > compaction_num_per_round) {
top_tablets.pop();
@@ -823,9 +846,7 @@ std::vector<TabletCompactionContext>
TabletManager::find_best_tablets_to_compact
}
} else {
if (current_compaction_score > highest_score) {
- bool ret = tablet_ptr->suitable_for_compaction(compaction_type,
-
cumulative_compaction_policy);
- if (ret) {
+ if (is_suitable()) {
highest_score = current_compaction_score;
best_tablet_context = {.tablet_ptr = tablet_ptr,
.score = current_compaction_score};
@@ -855,7 +876,7 @@ std::vector<TabletCompactionContext>
TabletManager::find_best_tablets_to_compact
picked_tablet_contexts.emplace_back(TabletCompactionContext {.tablet =
it->tablet_ptr});
}
- *score = highest_score;
+ score_stats->max_score = highest_score;
return picked_tablet_contexts;
}
diff --git a/be/src/storage/tablet/tablet_manager.h
b/be/src/storage/tablet/tablet_manager.h
index b507e79ada7..d1c6097cb42 100644
--- a/be/src/storage/tablet/tablet_manager.h
+++ b/be/src/storage/tablet/tablet_manager.h
@@ -82,7 +82,8 @@ public:
// single compaction tasks for the tablet.
std::vector<TabletCompactionContext> find_best_tablets_to_compaction(
CompactionType compaction_type, DataDir* data_dir,
- const std::unordered_set<TabletSharedPtr>&
tablet_submitted_compaction, uint32_t* score,
+ const std::unordered_set<TabletSharedPtr>&
tablet_submitted_compaction,
+ CompactionScoreStats* score_stats,
const std::unordered_map<std::string_view,
std::shared_ptr<CumulativeCompactionPolicy>>&
all_cumulative_compaction_policies);
diff --git a/be/test/cloud/cloud_compaction_test.cpp
b/be/test/cloud/cloud_compaction_test.cpp
index 0c21a845f59..78440a82de3 100644
--- a/be/test/cloud/cloud_compaction_test.cpp
+++ b/be/test/cloud/cloud_compaction_test.cpp
@@ -22,6 +22,7 @@
#include <gtest/gtest.h>
#include <memory>
+#include <string_view>
#include "cloud/cloud_base_compaction.h"
#include "cloud/cloud_cluster_info.h"
@@ -29,8 +30,10 @@
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
#include "cloud/config.h"
+#include "common/metrics/doris_metrics.h"
#include "io/fs/s3_file_system.h"
#include "json2pb/json_to_pb.h"
+#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "storage/olap_common.h"
#include "storage/rowset/rowset_factory.h"
#include "storage/rowset/rowset_meta.h"
@@ -145,13 +148,16 @@ TEST_F(CloudCompactionTest,
failure_base_compaction_tablet_sleep_test) {
tablet1->set_last_base_compaction_failure_time(0);
tablet1->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
tablet1->_approximate_num_rowsets = 10;
+ tablet1->_approximate_cumu_num_rowsets = 0;
mgr.put_tablet_for_UT(tablet1);
- int64_t max_score;
+ CompactionScoreStats score_stats;
std::vector<std::shared_ptr<CloudTablet>> tablets {};
Status st = mgr.get_topn_tablets_to_compact(1,
CompactionType::BASE_COMPACTION, filter_out,
- &tablets, &max_score);
+ &tablets, &score_stats);
ASSERT_EQ(st, Status::OK());
+ ASSERT_TRUE(score_stats.scanned);
+ ASSERT_EQ(score_stats.max_score, 10);
ASSERT_EQ(tablets.size(), 1);
tablet1->set_last_base_compaction_failure_time(
@@ -159,8 +165,10 @@ TEST_F(CloudCompactionTest,
failure_base_compaction_tablet_sleep_test) {
std::chrono::system_clock::now().time_since_epoch())
.count());
st = mgr.get_topn_tablets_to_compact(1, CompactionType::BASE_COMPACTION,
filter_out, &tablets,
- &max_score);
+ &score_stats);
ASSERT_EQ(st, Status::OK());
+ ASSERT_TRUE(score_stats.scanned);
+ ASSERT_EQ(score_stats.max_score, 10);
ASSERT_EQ(tablets.size(), 0);
}
@@ -186,11 +194,15 @@ TEST_F(CloudCompactionTest,
failure_cumu_compaction_tablet_sleep_test) {
tablet1->_approximate_cumu_num_deltas = 10;
mgr.put_tablet_for_UT(tablet1);
- int64_t max_score;
+ CompactionScoreStats score_stats;
std::vector<std::shared_ptr<CloudTablet>> tablets {};
Status st = mgr.get_topn_tablets_to_compact(1,
CompactionType::CUMULATIVE_COMPACTION,
- filter_out, &tablets,
&max_score);
+ filter_out, &tablets,
&score_stats);
ASSERT_EQ(st, Status::OK());
+ ASSERT_TRUE(score_stats.scanned);
+ ASSERT_EQ(score_stats.max_score, 10);
+ ASSERT_EQ(score_stats.size_based_max_score, 10);
+ ASSERT_EQ(score_stats.time_series_max_score, 0);
ASSERT_EQ(tablets.size(), 1);
tablet1->set_last_cumu_compaction_failure_time(
@@ -198,8 +210,10 @@ TEST_F(CloudCompactionTest,
failure_cumu_compaction_tablet_sleep_test) {
std::chrono::system_clock::now().time_since_epoch())
.count());
st = mgr.get_topn_tablets_to_compact(1, CompactionType::BASE_COMPACTION,
filter_out, &tablets,
- &max_score);
+ &score_stats);
ASSERT_EQ(st, Status::OK());
+ ASSERT_TRUE(score_stats.scanned);
+ ASSERT_EQ(score_stats.max_score, 0);
ASSERT_EQ(tablets.size(), 0);
}
@@ -223,15 +237,155 @@ TEST_F(CloudCompactionTest,
binlog_compaction_max_score_ignores_normal_tablets)
binlog_tablet->_approximate_cumu_num_deltas = 7;
mgr.put_tablet_for_UT(binlog_tablet);
- int64_t max_score = 0;
+ CompactionScoreStats score_stats;
std::vector<std::shared_ptr<CloudTablet>> tablets {};
Status st = mgr.get_topn_tablets_to_compact(1,
CompactionType::CUMU_BINLOG_COMPACTION,
- filter_out, &tablets,
&max_score);
+ filter_out, &tablets,
&score_stats);
ASSERT_EQ(st, Status::OK());
+ ASSERT_TRUE(score_stats.scanned);
ASSERT_EQ(tablets.size(), 1);
EXPECT_EQ(tablets.front()->tablet_id(), binlog_tablet->tablet_id());
- EXPECT_EQ(max_score, 7);
+ EXPECT_EQ(score_stats.max_score, 7);
+ EXPECT_EQ(score_stats.size_based_max_score, 0);
+ EXPECT_EQ(score_stats.time_series_max_score, 0);
+}
+
+TEST_F(CloudCompactionTest, split_cumu_compaction_score_stats_before_filter) {
+ CloudTabletMgr mgr(_engine);
+
+ auto create_tablet = [this, &mgr](int64_t tablet_id, std::string_view
compaction_policy,
+ int64_t score) {
+ TabletMetaSharedPtr tablet_meta(new TabletMeta(*_tablet_meta));
+ tablet_meta->_tablet_id = tablet_id;
+ tablet_meta->set_compaction_policy(std::string(compaction_policy));
+ auto tablet = std::make_shared<CloudTablet>(_engine, tablet_meta);
+
tablet->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
+ tablet->_approximate_cumu_num_deltas = score;
+ mgr.put_tablet_for_UT(tablet);
+ return tablet;
+ };
+
+ create_tablet(10000, CUMULATIVE_SIZE_BASED_POLICY, 7);
+ create_tablet(10001, CUMULATIVE_TIME_SERIES_POLICY, 13);
+
+ auto filter_time_series = [](CloudTablet* t) { return t->tablet_id() ==
10001; };
+ CompactionScoreStats score_stats;
+ std::vector<std::shared_ptr<CloudTablet>> tablets;
+ Status st = mgr.get_topn_tablets_to_compact(1,
CompactionType::CUMULATIVE_COMPACTION,
+ filter_time_series, &tablets,
&score_stats);
+ ASSERT_EQ(st, Status::OK());
+ ASSERT_TRUE(score_stats.scanned);
+ ASSERT_EQ(score_stats.max_score, 13);
+ ASSERT_EQ(score_stats.size_based_max_score, 7);
+ ASSERT_EQ(score_stats.time_series_max_score, 13);
+ ASSERT_EQ(tablets.size(), 1);
+ ASSERT_EQ(tablets[0]->tablet_id(), 10000);
+}
+
+TEST_F(CloudCompactionTest,
generate_cloud_compaction_tasks_updates_policy_metrics) {
+ CloudTabletMgr& mgr = _engine.tablet_mgr();
+ TabletMetaSharedPtr tablet_meta(new TabletMeta(*_tablet_meta));
+ tablet_meta->_tablet_id = 11000;
+
tablet_meta->set_compaction_policy(std::string(CUMULATIVE_SIZE_BASED_POLICY));
+ auto tablet = std::make_shared<CloudTablet>(_engine, tablet_meta);
+ tablet->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
+ tablet->_approximate_cumu_num_deltas = 7;
+ mgr.put_tablet_for_UT(tablet);
+
+ auto* metrics = DorisMetrics::instance();
+ metrics->tablet_cumulative_max_compaction_score->set_value(101);
+ metrics->tablet_size_based_max_compaction_score->set_value(102);
+ metrics->tablet_time_series_max_compaction_score->set_value(200);
+
+ auto tablets = _engine.generate_cloud_compaction_tasks_for_test(
+ CompactionType::CUMULATIVE_COMPACTION, false);
+ ASSERT_EQ(tablets.size(), 1);
+ ASSERT_EQ(tablets[0]->tablet_id(), 11000);
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 7);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 7);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 200);
+
+ tablets = _engine.generate_cloud_compaction_tasks_for_test(
+ CompactionType::CUMULATIVE_COMPACTION, true);
+ ASSERT_EQ(tablets.size(), 1);
+ ASSERT_EQ(tablets[0]->tablet_id(), 11000);
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 7);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 7);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 0);
+
+ TabletMetaSharedPtr time_series_meta(new TabletMeta(*_tablet_meta));
+ time_series_meta->_tablet_id = 11001;
+
time_series_meta->set_compaction_policy(std::string(CUMULATIVE_TIME_SERIES_POLICY));
+ auto time_series = std::make_shared<CloudTablet>(_engine,
time_series_meta);
+
time_series->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
+ time_series->_approximate_cumu_num_deltas = 13;
+ mgr.put_tablet_for_UT(time_series);
+
+ tablets = _engine.generate_cloud_compaction_tasks_for_test(
+ CompactionType::CUMULATIVE_COMPACTION, true);
+ ASSERT_FALSE(tablets.empty());
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 13);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 7);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 13);
+}
+
+TEST_F(CloudCompactionTest,
generate_cloud_binlog_compaction_tasks_updates_only_binlog_metric) {
+ CloudTabletMgr& mgr = _engine.tablet_mgr();
+
+ auto normal_meta = std::make_shared<TabletMeta>(*_tablet_meta);
+ normal_meta->_tablet_id = 11002;
+ normal_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_DATA);
+ auto normal_tablet = std::make_shared<CloudTablet>(_engine, normal_meta);
+
normal_tablet->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
+ normal_tablet->_approximate_cumu_num_deltas = 10;
+ mgr.put_tablet_for_UT(normal_tablet);
+
+ auto binlog_meta = std::make_shared<TabletMeta>(*_tablet_meta);
+ binlog_meta->_tablet_id = 11003;
+ binlog_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_ROW_BINLOG);
+ auto binlog_tablet = std::make_shared<CloudTablet>(_engine, binlog_meta);
+
binlog_tablet->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
+ binlog_tablet->_approximate_cumu_num_deltas = 7;
+ mgr.put_tablet_for_UT(binlog_tablet);
+
+ auto* metrics = DorisMetrics::instance();
+ metrics->tablet_cumulative_max_compaction_score->set_value(101);
+ metrics->tablet_size_based_max_compaction_score->set_value(102);
+ metrics->tablet_time_series_max_compaction_score->set_value(103);
+ metrics->tablet_binlog_max_compaction_score->set_value(104);
+
+ auto tablets = _engine.generate_cloud_compaction_tasks_for_test(
+ CompactionType::CUMU_BINLOG_COMPACTION, true);
+
+ ASSERT_EQ(tablets.size(), 1);
+ ASSERT_EQ(tablets[0]->tablet_id(), binlog_tablet->tablet_id());
+ ASSERT_EQ(metrics->tablet_binlog_max_compaction_score->value(), 7);
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 101);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 102);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 103);
+}
+
+TEST_F(CloudCompactionTest,
generate_cloud_compaction_tasks_clears_metrics_without_tablets) {
+ auto* metrics = DorisMetrics::instance();
+ metrics->tablet_cumulative_max_compaction_score->set_value(101);
+ metrics->tablet_size_based_max_compaction_score->set_value(102);
+ metrics->tablet_time_series_max_compaction_score->set_value(200);
+
+ auto tablets = _engine.generate_cloud_compaction_tasks_for_test(
+ CompactionType::CUMULATIVE_COMPACTION, false);
+ ASSERT_TRUE(tablets.empty());
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 101);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 102);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 200);
+
+ tablets = _engine.generate_cloud_compaction_tasks_for_test(
+ CompactionType::CUMULATIVE_COMPACTION, true);
+
+ ASSERT_TRUE(tablets.empty());
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 0);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 0);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 0);
}
static RowsetSharedPtr create_rowset(Version version, int num_segments, bool
overlapping,
diff --git a/be/test/storage/tablet/tablet_mgr_test.cpp
b/be/test/storage/tablet/tablet_mgr_test.cpp
index b3e15265fc9..d5e3dd1784b 100644
--- a/be/test/storage/tablet/tablet_mgr_test.cpp
+++ b/be/test/storage/tablet/tablet_mgr_test.cpp
@@ -26,9 +26,11 @@
#include <algorithm>
#include <memory>
#include <string>
+#include <string_view>
#include <vector>
#include "common/config.h"
+#include "common/metrics/doris_metrics.h"
#include "common/status.h"
#include "gtest/gtest_pred_impl.h"
#include "io/fs/local_file_system.h"
@@ -47,6 +49,8 @@
#include "storage/tablet/tablet_manager.h"
#include "storage/tablet/tablet_meta.h"
#include "storage/tablet/tablet_meta_manager.h"
+#include "util/debug_points.h"
+#include "util/defer_op.h"
#include "util/uid_util.h"
using ::testing::_;
@@ -88,6 +92,90 @@ public:
_tablet_mgr = nullptr;
config::compaction_num_per_round = 1;
}
+
+ TabletSharedPtr create_compaction_tablet(
+ int64_t tablet_id, int rowset_size,
+ std::string_view compaction_policy = CUMULATIVE_SIZE_BASED_POLICY,
+ DataDir* data_dir = nullptr) {
+ data_dir = data_dir == nullptr ? _data_dir : data_dir;
+ std::vector<TColumn> cols;
+ TColumn col1;
+ col1.column_type.type = TPrimitiveType::SMALLINT;
+ col1.__set_column_name("col1");
+ col1.__set_is_key(true);
+ cols.push_back(col1);
+
+ TColumn col2;
+ col2.column_type.type = TPrimitiveType::INT;
+ col2.__set_column_name(SEQUENCE_COL);
+ col2.__set_is_key(false);
+ col2.__set_aggregation_type(TAggregationType::REPLACE);
+ cols.push_back(col2);
+
+ TColumn col3;
+ col3.column_type.type = TPrimitiveType::INT;
+ col3.__set_column_name("v1");
+ col3.__set_is_key(false);
+ col3.__set_aggregation_type(TAggregationType::REPLACE);
+ cols.push_back(col3);
+
+ RuntimeProfile profile("CreateTablet");
+ TTabletSchema tablet_schema;
+ tablet_schema.__set_short_key_column_count(1);
+ tablet_schema.__set_schema_hash(3333);
+ tablet_schema.__set_keys_type(TKeysType::UNIQUE_KEYS);
+ tablet_schema.__set_storage_type(TStorageType::COLUMN);
+ tablet_schema.__set_columns(cols);
+ tablet_schema.__set_sequence_col_idx(1);
+ TCreateTabletReq create_tablet_req;
+ create_tablet_req.__set_tablet_schema(tablet_schema);
+ create_tablet_req.__set_tablet_id(tablet_id);
+ create_tablet_req.__set_version(1);
+ create_tablet_req.__set_replica_id(tablet_id * 10);
+
create_tablet_req.__set_compaction_policy(std::string(compaction_policy));
+ if (compaction_policy == CUMULATIVE_TIME_SERIES_POLICY) {
+
create_tablet_req.__set_time_series_compaction_file_count_threshold(1);
+ }
+ std::vector<DataDir*> data_dirs;
+ data_dirs.push_back(data_dir);
+ Status create_st = _tablet_mgr->create_tablet(create_tablet_req,
data_dirs, &profile);
+ if (!create_st.ok()) {
+ ADD_FAILURE() << create_st;
+ return nullptr;
+ }
+
+ TabletSharedPtr tablet = _tablet_mgr->get_tablet(tablet_id);
+ if (tablet == nullptr) {
+ ADD_FAILURE() << "failed to get tablet " << tablet_id;
+ return nullptr;
+ }
+
+ auto create_rowset = [=, this](int64_t start, int64_t end) {
+ auto rowset_meta = std::make_shared<RowsetMeta>();
+ Version version(start, end);
+ rowset_meta->set_version(version);
+ rowset_meta->set_tablet_id(tablet->tablet_id());
+ rowset_meta->set_tablet_uid(tablet->tablet_uid());
+ rowset_meta->set_rowset_id(k_engine->next_rowset_id());
+ return std::make_shared<BetaRowset>(tablet->tablet_schema(),
std::move(rowset_meta),
+ tablet->tablet_path());
+ };
+ auto st = tablet->init();
+ if (!st.ok()) {
+ ADD_FAILURE() << st;
+ return nullptr;
+ }
+ for (int i = 2; i <= rowset_size; ++i) {
+ auto rs = create_rowset(i, i);
+ st = tablet->add_inc_rowset(rs);
+ if (!st.ok()) {
+ ADD_FAILURE() << st;
+ return nullptr;
+ }
+ }
+ return tablet;
+ }
+
StorageEngine* k_engine;
private:
@@ -345,7 +433,8 @@ TEST_F(TabletMgrTest, GetRowsetId) {
}
TEST_F(TabletMgrTest, FindTabletWithCompact) {
- auto create_tablet = [this](int64_t tablet_id, int rowset_size) {
+ auto create_tablet = [this](int64_t tablet_id, int rowset_size,
+ std::string_view compaction_policy =
CUMULATIVE_SIZE_BASED_POLICY) {
std::vector<TColumn> cols;
TColumn col1;
col1.column_type.type = TPrimitiveType::SMALLINT;
@@ -380,6 +469,10 @@ TEST_F(TabletMgrTest, FindTabletWithCompact) {
create_tablet_req.__set_tablet_id(tablet_id);
create_tablet_req.__set_version(1);
create_tablet_req.__set_replica_id(tablet_id * 10);
+
create_tablet_req.__set_compaction_policy(std::string(compaction_policy));
+ if (compaction_policy == CUMULATIVE_TIME_SERIES_POLICY) {
+
create_tablet_req.__set_time_series_compaction_file_count_threshold(1);
+ }
std::vector<DataDir*> data_dirs;
data_dirs.push_back(_data_dir);
Status create_st = _tablet_mgr->create_tablet(create_tablet_req,
data_dirs, &profile);
@@ -433,20 +526,68 @@ TEST_F(TabletMgrTest, FindTabletWithCompact) {
cumulative_compaction_policies[CUMULATIVE_TIME_SERIES_POLICY] =
CumulativeCompactionPolicyFactory::create_cumulative_compaction_policy(
CUMULATIVE_TIME_SERIES_POLICY);
- uint32_t score = 0;
+ CompactionScoreStats score_stats;
auto compact_tablets = _tablet_mgr->find_best_tablets_to_compaction(
- CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set, &score,
+ CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score_stats,
cumulative_compaction_policies);
ASSERT_EQ(compact_tablets.size(), 1);
ASSERT_EQ(compact_tablets[0].tablet->tablet_id(), 10);
- ASSERT_EQ(score, 14);
+ ASSERT_TRUE(score_stats.scanned);
+ ASSERT_EQ(score_stats.max_score, 14);
+ ASSERT_EQ(score_stats.size_based_max_score, 14);
+ ASSERT_EQ(score_stats.time_series_max_score, 0);
+
+ // create 10 more tablets with higher compaction scores
+ for (int64_t id = 11; id <= 20; ++id) {
+ create_tablet(id, rowset_size++);
+ }
+
+ compact_tablets = _tablet_mgr->find_best_tablets_to_compaction(
+ CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score_stats,
+ cumulative_compaction_policies);
+ ASSERT_EQ(compact_tablets.size(), 1);
+ ASSERT_EQ(compact_tablets[0].tablet->tablet_id(), 20);
+ ASSERT_EQ(score_stats.max_score, 24);
+ ASSERT_EQ(score_stats.size_based_max_score, 24);
+ ASSERT_EQ(score_stats.time_series_max_score, 0);
+
+ create_tablet(21, rowset_size++);
+
+ compact_tablets = _tablet_mgr->find_best_tablets_to_compaction(
+ CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score_stats,
+ cumulative_compaction_policies);
+ ASSERT_EQ(compact_tablets.size(), 1);
+ ASSERT_EQ(compact_tablets[0].tablet->tablet_id(), 21);
+ ASSERT_EQ(score_stats.max_score, 25);
+ ASSERT_EQ(score_stats.size_based_max_score, 25);
+ ASSERT_EQ(score_stats.time_series_max_score, 0);
// drop all tablets
- for (int64_t id = 1; id <= 10; ++id) {
+ for (int64_t id = 1; id <= 21; ++id) {
Status drop_st = _tablet_mgr->drop_tablet(id, id * 10, false);
ASSERT_TRUE(drop_st.ok()) << drop_st;
}
+ {
+ create_tablet(40001, 8, CUMULATIVE_SIZE_BASED_POLICY);
+ create_tablet(40002, 12, CUMULATIVE_TIME_SERIES_POLICY);
+
+ compact_tablets = _tablet_mgr->find_best_tablets_to_compaction(
+ CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score_stats,
+ cumulative_compaction_policies);
+ ASSERT_TRUE(score_stats.scanned);
+ ASSERT_EQ(score_stats.max_score, 12);
+ ASSERT_EQ(score_stats.size_based_max_score, 8);
+ ASSERT_EQ(score_stats.time_series_max_score, 12);
+ ASSERT_EQ(compact_tablets.size(), 1);
+ ASSERT_EQ(compact_tablets[0].tablet->tablet_id(), 40002);
+
+ Status drop_st = _tablet_mgr->drop_tablet(40001, 400010, false);
+ ASSERT_TRUE(drop_st.ok()) << drop_st;
+ drop_st = _tablet_mgr->drop_tablet(40002, 400020, false);
+ ASSERT_TRUE(drop_st.ok()) << drop_st;
+ }
+
{
k_engine->_compaction_num_per_round = 10;
for (int64_t i = 1; i <= 100; ++i) {
@@ -454,7 +595,7 @@ TEST_F(TabletMgrTest, FindTabletWithCompact) {
}
compact_tablets = _tablet_mgr->find_best_tablets_to_compaction(
- CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score,
+ CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score_stats,
cumulative_compaction_policies);
ASSERT_EQ(compact_tablets.size(), 10);
int index = 0;
@@ -479,7 +620,7 @@ TEST_F(TabletMgrTest, FindTabletWithCompact) {
}
compact_tablets = _tablet_mgr->find_best_tablets_to_compaction(
- CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score,
+ CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score_stats,
cumulative_compaction_policies);
ASSERT_EQ(compact_tablets.size(), 5);
for (int i = 0; i < 5; ++i) {
@@ -501,6 +642,167 @@ TEST_F(TabletMgrTest, FindTabletWithCompact) {
ASSERT_TRUE(trash_st.ok()) << trash_st;
}
+TEST_F(TabletMgrTest, FindBestTabletsIgnoresUnsuitablePolicyScore) {
+ auto tablet = create_compaction_tablet(50001, 12,
CUMULATIVE_TIME_SERIES_POLICY);
+ ASSERT_TRUE(tablet != nullptr);
+
ASSERT_GT(tablet->calc_compaction_score(CompactionType::CUMULATIVE_COMPACTION),
5);
+
+ bool old_enable_debug_points = config::enable_debug_points;
+ config::enable_debug_points = true;
+ Defer restore_debug_points([&] { config::enable_debug_points =
old_enable_debug_points; });
+
DebugPoints::instance()->add("Tablet._calc_cumulative_compaction_score.return");
+ Defer clear_debug_point([] { DebugPoints::instance()->clear(); });
+
+ std::unordered_set<TabletSharedPtr> cumu_set;
+ std::unordered_map<std::string_view,
std::shared_ptr<CumulativeCompactionPolicy>>
+ cumulative_compaction_policies;
+ cumulative_compaction_policies[CUMULATIVE_SIZE_BASED_POLICY] =
+
CumulativeCompactionPolicyFactory::create_cumulative_compaction_policy(
+ CUMULATIVE_SIZE_BASED_POLICY);
+ cumulative_compaction_policies[CUMULATIVE_TIME_SERIES_POLICY] =
+
CumulativeCompactionPolicyFactory::create_cumulative_compaction_policy(
+ CUMULATIVE_TIME_SERIES_POLICY);
+
+ CompactionScoreStats score_stats;
+ auto compact_tablets = _tablet_mgr->find_best_tablets_to_compaction(
+ CompactionType::CUMULATIVE_COMPACTION, _data_dir, cumu_set,
&score_stats,
+ cumulative_compaction_policies);
+ ASSERT_TRUE(score_stats.scanned);
+ ASSERT_EQ(score_stats.max_score, 0);
+ ASSERT_EQ(score_stats.size_based_max_score, 0);
+ ASSERT_EQ(score_stats.time_series_max_score, 0);
+ ASSERT_TRUE(compact_tablets.empty());
+}
+
+TEST_F(TabletMgrTest, GenerateCompactionTasksClearsMissingPolicyScoreOnCheck) {
+ auto tablet = create_compaction_tablet(51001, 8,
CUMULATIVE_SIZE_BASED_POLICY);
+ ASSERT_TRUE(tablet != nullptr);
+ auto* metrics = DorisMetrics::instance();
+ metrics->tablet_cumulative_max_compaction_score->set_value(101);
+ metrics->tablet_size_based_max_compaction_score->set_value(102);
+ metrics->tablet_time_series_max_compaction_score->set_value(200);
+
+ std::vector<DataDir*> data_dirs {_data_dir};
+ auto tasks =
k_engine->generate_compaction_tasks_for_test(CompactionType::CUMULATIVE_COMPACTION,
+ data_dirs, true);
+
+ ASSERT_EQ(tasks.size(), 1);
+ ASSERT_EQ(tasks[0]->tablet_id(), 51001);
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 8);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 8);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 0);
+}
+
+TEST_F(TabletMgrTest,
GenerateCompactionTasksKeepsMissingPolicyScoreWithoutCheck) {
+ auto tablet = create_compaction_tablet(52001, 8,
CUMULATIVE_SIZE_BASED_POLICY);
+ ASSERT_TRUE(tablet != nullptr);
+ auto* metrics = DorisMetrics::instance();
+ metrics->tablet_cumulative_max_compaction_score->set_value(101);
+ metrics->tablet_size_based_max_compaction_score->set_value(102);
+ metrics->tablet_time_series_max_compaction_score->set_value(200);
+
+ std::vector<DataDir*> data_dirs {_data_dir};
+ auto tasks =
k_engine->generate_compaction_tasks_for_test(CompactionType::CUMULATIVE_COMPACTION,
+ data_dirs,
false);
+
+ ASSERT_EQ(tasks.size(), 1);
+ ASSERT_EQ(tasks[0]->tablet_id(), 52001);
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 8);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 8);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 200);
+}
+
+TEST_F(TabletMgrTest,
GenerateCompactionTasksDoesNotUpdateMetricWhenNoDirScanned) {
+ auto* metrics = DorisMetrics::instance();
+ metrics->tablet_cumulative_max_compaction_score->set_value(101);
+ metrics->tablet_size_based_max_compaction_score->set_value(102);
+ metrics->tablet_time_series_max_compaction_score->set_value(200);
+
+ std::vector<DataDir*> data_dirs;
+ auto tasks =
k_engine->generate_compaction_tasks_for_test(CompactionType::CUMULATIVE_COMPACTION,
+ data_dirs, true);
+
+ ASSERT_TRUE(tasks.empty());
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 101);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 102);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 200);
+}
+
+TEST_F(TabletMgrTest, GenerateCompactionTasksAggregatesScoreWhenNoSlot) {
+ auto dummy = create_compaction_tablet(53000, 5,
CUMULATIVE_SIZE_BASED_POLICY);
+ auto size_based = create_compaction_tablet(53001, 8,
CUMULATIVE_SIZE_BASED_POLICY);
+ auto time_series = create_compaction_tablet(53002, 12,
CUMULATIVE_TIME_SERIES_POLICY);
+ ASSERT_TRUE(dummy != nullptr);
+ ASSERT_TRUE(size_based != nullptr);
+ ASSERT_TRUE(time_series != nullptr);
+
+ std::vector<DataDir*> data_dirs {_data_dir};
+ auto& registry = k_engine->compaction_submit_registry_for_test();
+ registry.reset(data_dirs);
+ Defer reset_registry([&] { registry.reset(data_dirs); });
+ dummy->compaction_stage = CompactionStage::EXECUTING;
+ ASSERT_FALSE(registry.insert(dummy,
CompactionType::CUMULATIVE_COMPACTION));
+
+ int32_t old_compaction_task_num_per_disk =
config::compaction_task_num_per_disk;
+ config::compaction_task_num_per_disk = 1;
+ Defer restore_config(
+ [&] { config::compaction_task_num_per_disk =
old_compaction_task_num_per_disk; });
+ bool old_enable_compaction_priority_scheduling =
config::enable_compaction_priority_scheduling;
+ config::enable_compaction_priority_scheduling = false;
+ Defer restore_priority_scheduling([&] {
+ config::enable_compaction_priority_scheduling =
old_enable_compaction_priority_scheduling;
+ });
+
+ auto* metrics = DorisMetrics::instance();
+ metrics->tablet_cumulative_max_compaction_score->set_value(0);
+ metrics->tablet_size_based_max_compaction_score->set_value(0);
+ metrics->tablet_time_series_max_compaction_score->set_value(0);
+
+ auto tasks =
k_engine->generate_compaction_tasks_for_test(CompactionType::CUMULATIVE_COMPACTION,
+ data_dirs, true);
+
+ ASSERT_TRUE(tasks.empty());
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 12);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 8);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 12);
+}
+
+TEST_F(TabletMgrTest,
GenerateCompactionTasksDoesNotLowerPolicyScoreWhenDirFull) {
+ std::string full_dir_path =
"./be/test/storage/test_data/converter_test_data/tmp_full";
+
ASSERT_TRUE(io::global_local_filesystem()->delete_directory(full_dir_path).ok());
+
ASSERT_TRUE(io::global_local_filesystem()->create_directory(full_dir_path).ok());
+ ASSERT_TRUE(io::global_local_filesystem()->create_directory(full_dir_path
+ "/meta").ok());
+ Defer cleanup_full_dir([&] {
+
static_cast<void>(io::global_local_filesystem()->delete_directory(full_dir_path));
+ });
+
+ auto full_data_dir = std::make_unique<DataDir>(*k_engine, full_dir_path,
1000000000);
+ ASSERT_TRUE(full_data_dir->init().ok());
+ auto full_time_series =
+ create_compaction_tablet(54001, 12, CUMULATIVE_TIME_SERIES_POLICY,
full_data_dir.get());
+ auto size_based = create_compaction_tablet(54002, 8,
CUMULATIVE_SIZE_BASED_POLICY);
+ ASSERT_TRUE(full_time_series != nullptr);
+ ASSERT_TRUE(size_based != nullptr);
+ Defer drop_full_tablet(
+ [&] { static_cast<void>(_tablet_mgr->drop_tablet(54001, 540010,
false)); });
+ full_data_dir->set_capacity_for_test(100, 0);
+
+ auto* metrics = DorisMetrics::instance();
+ metrics->tablet_cumulative_max_compaction_score->set_value(200);
+ metrics->tablet_size_based_max_compaction_score->set_value(0);
+ metrics->tablet_time_series_max_compaction_score->set_value(200);
+
+ std::vector<DataDir*> data_dirs {_data_dir, full_data_dir.get()};
+ auto tasks =
k_engine->generate_compaction_tasks_for_test(CompactionType::CUMULATIVE_COMPACTION,
+ data_dirs, true);
+
+ ASSERT_EQ(tasks.size(), 1);
+ ASSERT_EQ(tasks[0]->tablet_id(), 54002);
+ ASSERT_EQ(metrics->tablet_cumulative_max_compaction_score->value(), 200);
+ ASSERT_EQ(metrics->tablet_size_based_max_compaction_score->value(), 8);
+ ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 200);
+}
+
TEST_F(TabletMgrTest, LoadTabletFromMeta) {
TTabletId tablet_id = 111;
TSchemaHash schema_hash = 3333;
diff --git a/be/test/util/doris_metrics_test.cpp
b/be/test/util/doris_metrics_test.cpp
index dfac2d557fe..a3269c9b04f 100644
--- a/be/test/util/doris_metrics_test.cpp
+++ b/be/test/util/doris_metrics_test.cpp
@@ -178,6 +178,18 @@ TEST_F(DorisMetricsTest, Normal) {
EXPECT_TRUE(metric != nullptr);
EXPECT_STREQ("31", metric->to_string().c_str());
}
+ {
+
DorisMetrics::instance()->tablet_size_based_max_compaction_score->set_value(41);
+ auto metric =
server_entity->get_metric("tablet_size_based_max_compaction_score");
+ EXPECT_TRUE(metric != nullptr);
+ EXPECT_STREQ("41", metric->to_string().c_str());
+ }
+ {
+
DorisMetrics::instance()->tablet_time_series_max_compaction_score->set_value(42);
+ auto metric =
server_entity->get_metric("tablet_time_series_max_compaction_score");
+ EXPECT_TRUE(metric != nullptr);
+ EXPECT_STREQ("42", metric->to_string().c_str());
+ }
{
DorisMetrics::instance()->base_compaction_bytes_total->increment(32);
auto metric =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
