You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2022/10/31 10:39:27 UTC
[doris] 01/01: Revert "[enhancement](compaction) opt compaction task producer and quick compaction (#13495)"
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch revert-13495-compaction_v8
in repository https://gitbox.apache.org/repos/asf/doris.git
commit c1b63d4ea7843b7884320cf40cd29010fbb7dcaa
Author: Mingyu Chen <mo...@gmail.com>
AuthorDate: Mon Oct 31 18:39:19 2022 +0800
Revert "[enhancement](compaction) opt compaction task producer and quick compaction (#13495)"
This reverts commit 4f2ea0776ca3fe5315ab5ef7e00eefabfb5771a0.
---
be/src/agent/task_worker_pool.cpp | 36 +++++----
be/src/common/config.h | 69 +++++++++++-----
be/src/olap/base_compaction.cpp | 17 ++--
be/src/olap/compaction.cpp | 55 ++++++++++++-
be/src/olap/compaction.h | 1 +
be/src/olap/cumulative_compaction.cpp | 10 ++-
be/src/olap/cumulative_compaction_policy.cpp | 47 +++++------
be/src/olap/cumulative_compaction_policy.h | 27 ++++---
be/src/olap/delta_writer.cpp | 18 ++---
be/src/olap/merger.cpp | 4 +
be/src/olap/olap_server.cpp | 78 ++++++++++++-------
be/src/olap/storage_engine.cpp | 4 +
be/src/olap/storage_engine.h | 4 +
be/src/olap/tablet.cpp | 91 +++++++++++++++-------
be/src/olap/tablet.h | 28 ++++---
be/src/olap/tablet_manager.cpp | 25 +++---
be/test/olap/cumulative_compaction_policy_test.cpp | 12 +--
docs/en/docs/admin-manual/config/be-config.md | 52 +++++++++----
docs/zh-CN/docs/admin-manual/config/be-config.md | 52 +++++++++----
19 files changed, 432 insertions(+), 198 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp b/be/src/agent/task_worker_pool.cpp
index 20f8414841..5e482ef90f 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -731,25 +731,29 @@ void TaskWorkerPool::_publish_version_worker_thread_callback() {
.error(status);
finish_task_request.__set_error_tablet_ids(error_tablet_ids);
} else {
- for (int i = 0; i < succ_tablet_ids.size(); i++) {
- TabletSharedPtr tablet =
- StorageEngine::instance()->tablet_manager()->get_tablet(succ_tablet_ids[i]);
- if (tablet != nullptr) {
- tablet->publised_count++;
- if (tablet->publised_count % 10 == 0) {
- StorageEngine::instance()->submit_compaction_task(
- tablet, CompactionType::CUMULATIVE_COMPACTION);
- LOG(INFO) << "trigger compaction succ, tabletid:" << succ_tablet_ids[i]
- << ", publised:" << tablet->publised_count;
+ if (config::enable_quick_compaction && config::quick_compaction_batch_size > 0) {
+ for (int i = 0; i < succ_tablet_ids.size(); i++) {
+ TabletSharedPtr tablet =
+ StorageEngine::instance()->tablet_manager()->get_tablet(
+ succ_tablet_ids[i]);
+ if (tablet != nullptr) {
+ tablet->publised_count++;
+ if (tablet->publised_count % config::quick_compaction_batch_size == 0) {
+ StorageEngine::instance()->submit_quick_compaction_task(tablet);
+ LOG(INFO) << "trigger quick compaction succ, tabletid:"
+ << succ_tablet_ids[i]
+ << ", publised:" << tablet->publised_count;
+ }
+ } else {
+ LOG(WARNING) << "trigger quick compaction failed, tabletid:"
+ << succ_tablet_ids[i];
}
- } else {
- LOG(WARNING) << "trigger compaction failed, tabletid:" << succ_tablet_ids[i];
}
+ LOG_INFO("successfully publish version")
+ .tag("signature", agent_task_req.signature)
+ .tag("transaction_id", publish_version_req.transaction_id)
+ .tag("tablets_num", succ_tablet_ids.size());
}
- LOG_INFO("successfully publish version")
- .tag("signature", agent_task_req.signature)
- .tag("transaction_id", publish_version_req.transaction_id)
- .tag("tablets_num", succ_tablet_ids.size());
}
status.to_thrift(&finish_task_request.task_status);
diff --git a/be/src/common/config.h b/be/src/common/config.h
index 23634a3ee1..8a690c2c1d 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -253,41 +253,59 @@ CONF_Bool(enable_vectorized_compaction, "true");
// whether enable vectorized schema change/material-view/rollup task.
CONF_Bool(enable_vectorized_alter_table, "true");
-// This config can be set to limit thread number in compaction thread pool.
-CONF_mInt32(max_base_compaction_threads, "4");
-CONF_mInt32(max_cumu_compaction_threads, "10");
+// check the configuration of auto compaction in seconds when auto compaction disabled
+CONF_mInt32(check_auto_compaction_interval_seconds, "5");
+CONF_mInt64(base_compaction_num_cumulative_deltas, "5");
+CONF_mDouble(base_cumulative_delta_ratio, "0.3");
+CONF_mInt64(base_compaction_interval_seconds_since_last_operation, "86400");
+CONF_mInt32(base_compaction_write_mbytes_per_sec, "5");
CONF_Bool(enable_base_compaction_idle_sched, "true");
-CONF_mInt64(base_compaction_min_rowset_num, "5");
-CONF_mDouble(base_compaction_min_data_ratio, "0.3");
+
+// dup key not compaction big files
+CONF_Bool(enable_dup_key_base_compaction_skip_big_file, "true");
CONF_mInt64(base_compaction_dup_key_max_file_size_mbytes, "1024");
-// output rowset of cumulative compaction total disk size exceed this config size,
+// In size_based policy, output rowset of cumulative compaction total disk size exceed this config size,
// this rowset will be given to base compaction, unit is m byte.
-CONF_mInt64(compaction_promotion_size_mbytes, "1024");
+CONF_mInt64(cumulative_size_based_promotion_size_mbytes, "1024");
-// output rowset of cumulative compaction total disk size exceed this config ratio of
+// In size_based policy, output rowset of cumulative compaction total disk size exceed this config ratio of
// base rowset's total disk size, this rowset will be given to base compaction. The value must be between
// 0 and 1.
-CONF_mDouble(compaction_promotion_ratio, "0.05");
+CONF_mDouble(cumulative_size_based_promotion_ratio, "0.05");
-// the smallest size of rowset promotion. When the rowset is less than this config, this
+// In size_based policy, the smallest size of rowset promotion. When the rowset is less than this config, this
// rowset will be not given to base compaction. The unit is m byte.
-CONF_mInt64(compaction_promotion_min_size_mbytes, "64");
+CONF_mInt64(cumulative_size_based_promotion_min_size_mbytes, "64");
// The lower bound size to do cumulative compaction. When total disk size of candidate rowsets is less than
// this size, size_based policy may not do to cumulative compaction. The unit is m byte.
-CONF_mInt64(compaction_min_size_mbytes, "64");
+CONF_mInt64(cumulative_size_based_compaction_lower_size_mbytes, "64");
// cumulative compaction policy: min and max delta file's number
-CONF_mInt64(cumulative_compaction_min_deltas, "5");
-CONF_mInt64(cumulative_compaction_max_deltas, "1000");
+CONF_mInt64(min_cumulative_compaction_num_singleton_deltas, "5");
+CONF_mInt64(max_cumulative_compaction_num_singleton_deltas, "1000");
+
+// if compaction of a tablet failed, this tablet should not be chosen to
+// compaction until this interval passes.
+CONF_mInt64(min_compaction_failure_interval_sec, "5"); // 5 seconds
+
+// This config can be set to limit thread number in compaction thread pool.
+CONF_mInt32(max_base_compaction_threads, "4");
+CONF_mInt32(max_cumu_compaction_threads, "10");
+
+// This config can be set to limit thread number in smallcompaction thread pool.
+CONF_mInt32(quick_compaction_max_threads, "10");
+
+// Thread count to do tablet meta checkpoint, -1 means use the data directories count.
+CONF_Int32(max_meta_checkpoint_threads, "-1");
// The upper limit of "permits" held by all compaction tasks. This config can be set to limit memory consumption for compaction.
CONF_mInt64(total_permits_for_compaction_score, "10000");
// sleep interval in ms after generated compaction tasks
-CONF_mInt32(generate_compaction_tasks_interval_ms, "10");
+CONF_mInt32(generate_compaction_tasks_min_interval_ms, "10");
// Compaction task number per disk.
// Must be greater than 2, because Base compaction and Cumulative compaction have at least one thread each.
@@ -301,17 +319,23 @@ CONF_Validator(compaction_task_num_per_fast_disk,
// How many rounds of cumulative compaction for each round of base compaction when compaction tasks generation.
CONF_mInt32(cumulative_compaction_rounds_for_each_base_compaction_round, "9");
+// Merge log will be printed for each "row_step_for_compaction_merge_log" rows merged during compaction
+CONF_mInt64(row_step_for_compaction_merge_log, "0");
+
// Threshold to logging compaction trace, in seconds.
CONF_mInt32(base_compaction_trace_threshold, "60");
CONF_mInt32(cumulative_compaction_trace_threshold, "10");
CONF_mBool(disable_compaction_trace_log, "true");
-// Thread count to do tablet meta checkpoint, -1 means use the data directories count.
-CONF_Int32(max_meta_checkpoint_threads, "-1");
-
// Threshold to logging agent task trace, in seconds.
CONF_mInt32(agent_task_trace_threshold_sec, "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");
+
// This config can be set to limit thread number in tablet migration thread pool.
CONF_Int32(min_tablet_migration_threads, "1");
CONF_Int32(max_tablet_migration_threads, "1");
@@ -783,6 +807,15 @@ CONF_mInt32(bloom_filter_predicate_check_row_num, "204800");
CONF_Bool(enable_decimalv3, "false");
+//whether turn on quick compaction feature
+CONF_Bool(enable_quick_compaction, "false");
+// For continuous versions that rows less than quick_compaction_max_rows will trigger compaction quickly
+CONF_Int32(quick_compaction_max_rows, "1000");
+// min compaction versions
+CONF_Int32(quick_compaction_batch_size, "10");
+// do compaction min rowsets
+CONF_Int32(quick_compaction_min_rowsets, "10");
+
// cooldown task configs
CONF_Int32(cooldown_thread_num, "5");
CONF_mInt64(generate_cooldown_task_interval_sec, "20");
diff --git a/be/src/olap/base_compaction.cpp b/be/src/olap/base_compaction.cpp
index f1e46014ef..63da436d23 100644
--- a/be/src/olap/base_compaction.cpp
+++ b/be/src/olap/base_compaction.cpp
@@ -87,9 +87,10 @@ Status BaseCompaction::execute_compact_impl() {
}
void BaseCompaction::_filter_input_rowset() {
- // if dup_key and no delete predicate
+ // if enable dup key skip big file and no delete predicate
// we skip big files too save resources
- if (_tablet->keys_type() != KeysType::DUP_KEYS || _tablet->delete_predicates().size() != 0) {
+ if (!config::enable_dup_key_base_compaction_skip_big_file ||
+ _tablet->keys_type() != KeysType::DUP_KEYS || _tablet->delete_predicates().size() != 0) {
return;
}
int64_t max_size = config::base_compaction_dup_key_max_file_size_mbytes * 1024 * 1024;
@@ -143,11 +144,11 @@ Status BaseCompaction::pick_rowsets_to_compact() {
}
// 1. cumulative rowset must reach base_compaction_num_cumulative_deltas threshold
- if (_input_rowsets.size() > config::base_compaction_min_rowset_num) {
+ if (_input_rowsets.size() > config::base_compaction_num_cumulative_deltas) {
VLOG_NOTICE << "satisfy the base compaction policy. tablet=" << _tablet->full_name()
<< ", num_cumulative_rowsets=" << _input_rowsets.size() - 1
<< ", base_compaction_num_cumulative_rowsets="
- << config::base_compaction_min_rowset_num;
+ << config::base_compaction_num_cumulative_deltas;
return Status::OK();
}
@@ -159,7 +160,7 @@ Status BaseCompaction::pick_rowsets_to_compact() {
cumulative_total_size += (*it)->data_disk_size();
}
- double min_data_ratio = config::base_compaction_min_data_ratio;
+ double base_cumulative_delta_ratio = config::base_cumulative_delta_ratio;
if (base_size == 0) {
// base_size == 0 means this may be a base version [0-1], which has no data.
// set to 1 to void divide by zero
@@ -167,18 +168,18 @@ Status BaseCompaction::pick_rowsets_to_compact() {
}
double cumulative_base_ratio = static_cast<double>(cumulative_total_size) / base_size;
- if (cumulative_base_ratio > min_data_ratio) {
+ if (cumulative_base_ratio > base_cumulative_delta_ratio) {
VLOG_NOTICE << "satisfy the base compaction policy. tablet=" << _tablet->full_name()
<< ", cumulative_total_size=" << cumulative_total_size
<< ", base_size=" << base_size
<< ", cumulative_base_ratio=" << cumulative_base_ratio
- << ", policy_min_data_ratio=" << min_data_ratio;
+ << ", policy_ratio=" << base_cumulative_delta_ratio;
return Status::OK();
}
// 3. the interval since last base compaction reaches the threshold
int64_t base_creation_time = _input_rowsets[0]->creation_time();
- int64_t interval_threshold = 86400;
+ int64_t interval_threshold = config::base_compaction_interval_seconds_since_last_operation;
int64_t interval_since_last_base_compaction = time(nullptr) - base_creation_time;
if (interval_since_last_base_compaction > interval_threshold) {
VLOG_NOTICE << "satisfy the base compaction policy. tablet=" << _tablet->full_name()
diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
index 7c99e05a95..41b2ebd902 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -59,6 +59,57 @@ Status Compaction::execute_compact() {
return st;
}
+Status Compaction::quick_rowsets_compact() {
+ std::unique_lock<std::mutex> lock(_tablet->get_cumulative_compaction_lock(), std::try_to_lock);
+ if (!lock.owns_lock()) {
+ LOG(WARNING) << "The tablet is under cumulative compaction. tablet="
+ << _tablet->full_name();
+ return Status::OLAPInternalError(OLAP_ERR_CE_TRY_CE_LOCK_ERROR);
+ }
+
+ // Clone task may happen after compaction task is submitted to thread pool, and rowsets picked
+ // for compaction may change. In this case, current compaction task should not be executed.
+ if (_tablet->get_clone_occurred()) {
+ _tablet->set_clone_occurred(false);
+ return Status::OLAPInternalError(OLAP_ERR_CUMULATIVE_CLONE_OCCURRED);
+ }
+
+ _input_rowsets.clear();
+ int version_count = _tablet->version_count();
+ MonotonicStopWatch watch;
+ watch.start();
+ int64_t permits = 0;
+ _tablet->pick_quick_compaction_rowsets(&_input_rowsets, &permits);
+ std::vector<Version> missedVersions;
+ find_longest_consecutive_version(&_input_rowsets, &missedVersions);
+ if (missedVersions.size() != 0) {
+ LOG(WARNING) << "quick_rowsets_compaction, find missed version"
+ << ",input_size:" << _input_rowsets.size();
+ }
+ int nums = _input_rowsets.size();
+ if (_input_rowsets.size() >= config::quick_compaction_min_rowsets) {
+ Status st = check_version_continuity(_input_rowsets);
+ if (!st.ok()) {
+ LOG(WARNING) << "quick_rowsets_compaction failed, cause version not continuous";
+ return st;
+ }
+ st = do_compaction(permits);
+ if (!st.ok()) {
+ gc_output_rowset();
+ LOG(WARNING) << "quick_rowsets_compaction failed";
+ } else {
+ LOG(INFO) << "quick_compaction succ"
+ << ", before_versions:" << version_count
+ << ", after_versions:" << _tablet->version_count()
+ << ", cost:" << (watch.elapsed_time() / 1000 / 1000) << "ms"
+ << ", merged: " << nums << ", batch:" << config::quick_compaction_batch_size
+ << ", segments:" << permits << ", tabletid:" << _tablet->tablet_id();
+ _tablet->set_last_quick_compaction_success_time(UnixMillis());
+ }
+ }
+ return Status::OK();
+}
+
Status Compaction::do_compaction(int64_t permits) {
TRACE("start to do compaction");
_tablet->data_dir()->disks_compaction_score_increment(permits);
@@ -172,7 +223,6 @@ Status Compaction::do_compaction_impl(int64_t permits) {
}
auto cumu_policy = _tablet->cumulative_compaction_policy();
- DCHECK(cumu_policy);
LOG(INFO) << "succeed to do " << merge_type << compaction_name()
<< ". tablet=" << _tablet->full_name() << ", output_version=" << _output_version
<< ", current_max_version=" << current_max_version
@@ -180,7 +230,8 @@ Status Compaction::do_compaction_impl(int64_t permits) {
<< ", input_row_num=" << _input_row_num
<< ", output_row_num=" << _output_rowset->num_rows()
<< ". elapsed time=" << watch.get_elapse_second()
- << "s. cumulative_compaction_policy=" << cumu_policy->name()
+ << "s. cumulative_compaction_policy="
+ << (cumu_policy == nullptr ? "quick" : cumu_policy->name())
<< ", compact_row_per_second=" << int(_input_row_num / watch.get_elapse_second());
return Status::OK();
diff --git a/be/src/olap/compaction.h b/be/src/olap/compaction.h
index 42b1e60e5b..26d985f9c0 100644
--- a/be/src/olap/compaction.h
+++ b/be/src/olap/compaction.h
@@ -48,6 +48,7 @@ public:
// This is only for http CompactionAction
Status compact();
+ Status quick_rowsets_compact();
virtual Status prepare_compact() = 0;
Status execute_compact();
diff --git a/be/src/olap/cumulative_compaction.cpp b/be/src/olap/cumulative_compaction.cpp
index baf1382989..4461a240b5 100644
--- a/be/src/olap/cumulative_compaction.cpp
+++ b/be/src/olap/cumulative_compaction.cpp
@@ -117,9 +117,10 @@ Status CumulativeCompaction::pick_rowsets_to_compact() {
size_t compaction_score = 0;
int transient_size = _tablet->cumulative_compaction_policy()->pick_input_rowsets(
- _tablet.get(), candidate_rowsets, config::cumulative_compaction_max_deltas,
- config::cumulative_compaction_min_deltas, &_input_rowsets, &_last_delete_version,
- &compaction_score);
+ _tablet.get(), candidate_rowsets,
+ config::max_cumulative_compaction_num_singleton_deltas,
+ config::min_cumulative_compaction_num_singleton_deltas, &_input_rowsets,
+ &_last_delete_version, &compaction_score);
// Cumulative compaction will process with at least 1 rowset.
// So when there is no rowset being chosen, we should return Status::OLAPInternalError(OLAP_ERR_CUMULATIVE_NO_SUITABLE_VERSION):
@@ -142,7 +143,8 @@ Status CumulativeCompaction::pick_rowsets_to_compact() {
int64_t last_cumu = _tablet->last_cumu_compaction_success_time();
int64_t last_base = _tablet->last_base_compaction_success_time();
if (last_cumu != 0 || last_base != 0) {
- int64_t interval_threshold = 86400 * 1000;
+ int64_t interval_threshold =
+ config::base_compaction_interval_seconds_since_last_operation * 1000;
int64_t cumu_interval = now - last_cumu;
int64_t base_interval = now - last_base;
if (cumu_interval > interval_threshold && base_interval > interval_threshold) {
diff --git a/be/src/olap/cumulative_compaction_policy.cpp b/be/src/olap/cumulative_compaction_policy.cpp
index 35d65be797..599fad92be 100644
--- a/be/src/olap/cumulative_compaction_policy.cpp
+++ b/be/src/olap/cumulative_compaction_policy.cpp
@@ -25,16 +25,16 @@
namespace doris {
SizeBasedCumulativeCompactionPolicy::SizeBasedCumulativeCompactionPolicy(
- int64_t promotion_size, double promotion_ratio, int64_t promotion_min_size,
- int64_t compaction_min_size)
+ int64_t size_based_promotion_size, double size_based_promotion_ratio,
+ int64_t size_based_promotion_min_size, int64_t size_based_compaction_lower_bound_size)
: CumulativeCompactionPolicy(),
- _promotion_size(promotion_size),
- _promotion_ratio(promotion_ratio),
- _promotion_min_size(promotion_min_size),
- _compaction_min_size(compaction_min_size) {
- // init _levels by divide 2 between promotion_size and 1K
- // cu compaction handle file size less then promotion_size
- int64_t i_size = promotion_size / 2;
+ _size_based_promotion_size(size_based_promotion_size),
+ _size_based_promotion_ratio(size_based_promotion_ratio),
+ _size_based_promotion_min_size(size_based_promotion_min_size),
+ _size_based_compaction_lower_bound_size(size_based_compaction_lower_bound_size) {
+ // init _levels by divide 2 between size_based_compaction_lower_bound_size and 1K
+ // cu compaction handle file size less then size_based_compaction_lower_bound_size
+ int64_t i_size = size_based_promotion_size / 2;
while (i_size >= 1024) {
_levels.push_back(i_size);
@@ -122,19 +122,20 @@ void SizeBasedCumulativeCompactionPolicy::calculate_cumulative_point(
void SizeBasedCumulativeCompactionPolicy::_calc_promotion_size(RowsetMetaSharedPtr base_rowset_meta,
int64_t* promotion_size) {
int64_t base_size = base_rowset_meta->total_disk_size();
- *promotion_size = base_size * _promotion_ratio;
+ *promotion_size = base_size * _size_based_promotion_ratio;
- // promotion_size is between _promotion_size and _promotion_min_size
- if (*promotion_size >= _promotion_size) {
- *promotion_size = _promotion_size;
- } else if (*promotion_size <= _promotion_min_size) {
- *promotion_size = _promotion_min_size;
+ // promotion_size is between _size_based_promotion_size and _size_based_promotion_min_size
+ if (*promotion_size >= _size_based_promotion_size) {
+ *promotion_size = _size_based_promotion_size;
+ } else if (*promotion_size <= _size_based_promotion_min_size) {
+ *promotion_size = _size_based_promotion_min_size;
}
- _refresh_tablet_promotion_size(*promotion_size);
+ _refresh_tablet_size_based_promotion_size(*promotion_size);
}
-void SizeBasedCumulativeCompactionPolicy::_refresh_tablet_promotion_size(int64_t promotion_size) {
- _tablet_promotion_size = promotion_size;
+void SizeBasedCumulativeCompactionPolicy::_refresh_tablet_size_based_promotion_size(
+ int64_t promotion_size) {
+ _tablet_size_based_promotion_size = promotion_size;
}
void SizeBasedCumulativeCompactionPolicy::update_cumulative_point(
@@ -151,7 +152,7 @@ void SizeBasedCumulativeCompactionPolicy::update_cumulative_point(
// if rowsets have no delete version, check output_rowset total disk size
// satisfies promotion size.
size_t total_size = output_rowset->rowset_meta()->total_disk_size();
- if (total_size >= _tablet_promotion_size) {
+ if (total_size >= _tablet_size_based_promotion_size) {
tablet->set_cumulative_layer_point(output_rowset->end_version() + 1);
}
}
@@ -239,7 +240,7 @@ int SizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
const int64_t max_compaction_score, const int64_t min_compaction_score,
std::vector<RowsetSharedPtr>* input_rowsets, Version* last_delete_version,
size_t* compaction_score) {
- size_t promotion_size = _tablet_promotion_size;
+ size_t promotion_size = _tablet_size_based_promotion_size;
int transient_size = 0;
*compaction_score = 0;
int64_t total_size = 0;
@@ -318,10 +319,12 @@ int SizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
// if we have a sufficient number of segments, we should process the compaction.
// otherwise, we check number of segments and total_size whether can do compaction.
- if (total_size < _compaction_min_size && *compaction_score < min_compaction_score) {
+ if (total_size < _size_based_compaction_lower_bound_size &&
+ *compaction_score < min_compaction_score) {
input_rowsets->clear();
*compaction_score = 0;
- } else if (total_size >= _compaction_min_size && input_rowsets->size() == 1) {
+ } else if (total_size >= _size_based_compaction_lower_bound_size &&
+ input_rowsets->size() == 1) {
auto rs_meta = input_rowsets->front()->rowset_meta();
// if there is only one rowset and not overlapping,
// we do not need to do compaction
diff --git a/be/src/olap/cumulative_compaction_policy.h b/be/src/olap/cumulative_compaction_policy.h
index e2ab9f926e..4abdd3497a 100644
--- a/be/src/olap/cumulative_compaction_policy.h
+++ b/be/src/olap/cumulative_compaction_policy.h
@@ -117,10 +117,13 @@ public:
/// it needs tablet pointer to access tablet method.
/// param tablet, the shared pointer of tablet
SizeBasedCumulativeCompactionPolicy(
- int64_t promotion_size = config::compaction_promotion_size_mbytes * 1024 * 1024,
- double promotion_ratio = config::compaction_promotion_ratio,
- int64_t promotion_min_size = config::compaction_promotion_min_size_mbytes * 1024 * 1024,
- int64_t compaction_min_size = config::compaction_min_size_mbytes * 1024 * 1024);
+ int64_t size_based_promotion_size =
+ config::cumulative_size_based_promotion_size_mbytes * 1024 * 1024,
+ double size_based_promotion_ratio = config::cumulative_size_based_promotion_ratio,
+ int64_t size_based_promotion_min_size =
+ config::cumulative_size_based_promotion_min_size_mbytes * 1024 * 1024,
+ int64_t size_based_compaction_lower_bound_size =
+ config::cumulative_size_based_compaction_lower_size_mbytes * 1024 * 1024);
/// Destructor function of SizeBasedCumulativeCompactionPolicy.
~SizeBasedCumulativeCompactionPolicy() {}
@@ -163,24 +166,24 @@ private:
void _calc_promotion_size(RowsetMetaSharedPtr base_rowset_meta, int64_t* promotion_size);
/// calculate the disk size belong to which level, the level is divide by power of 2
- /// between compaction_promotion_min_size_mbytes
- /// and compaction_promotion_size_mbytes
+ /// between cumulative_size_based_promotion_min_size_mbytes
+ /// and cumulative_size_based_promotion_size_mbytes
int _level_size(const int64_t size);
/// when policy calculate cumulative_compaction_score, update promotion size at the same time
- void _refresh_tablet_promotion_size(int64_t promotion_size);
+ void _refresh_tablet_size_based_promotion_size(int64_t promotion_size);
private:
/// cumulative compaction promotion size, unit is byte.
- int64_t _promotion_size;
+ int64_t _size_based_promotion_size;
/// cumulative compaction promotion ratio of base rowset total disk size.
- double _promotion_ratio;
+ double _size_based_promotion_ratio;
/// cumulative compaction promotion min size, unit is byte.
- int64_t _promotion_min_size;
+ int64_t _size_based_promotion_min_size;
/// lower bound size to do compaction compaction.
- int64_t _compaction_min_size;
+ int64_t _size_based_compaction_lower_bound_size;
/// record tablet promotion size, it is updated each time when calculate cumulative_compaction_score
- int64_t _tablet_promotion_size;
+ int64_t _tablet_size_based_promotion_size;
/// levels division of disk size, same level rowsets can do compaction
std::vector<int64_t> _levels;
};
diff --git a/be/src/olap/delta_writer.cpp b/be/src/olap/delta_writer.cpp
index 8bd817a591..c8a36fbec7 100644
--- a/be/src/olap/delta_writer.cpp
+++ b/be/src/olap/delta_writer.cpp
@@ -112,17 +112,15 @@ Status DeltaWriter::init() {
_mem_tracker = std::make_shared<MemTrackerLimiter>(
-1, fmt::format("DeltaWriter:tabletId={}", _tablet->tablet_id()), _parent_tracker);
// check tablet version number
- if (_tablet->version_count() > config::max_tablet_version_num - 100) {
- //trigger compaction
- StorageEngine::instance()->submit_compaction_task(_tablet,
- CompactionType::CUMULATIVE_COMPACTION);
- if (_tablet->version_count() > config::max_tablet_version_num) {
- LOG(WARNING) << "failed to init delta writer. version count: "
- << _tablet->version_count()
- << ", exceed limit: " << config::max_tablet_version_num
- << ". tablet: " << _tablet->full_name();
- return Status::OLAPInternalError(OLAP_ERR_TOO_MANY_VERSION);
+ if (_tablet->version_count() > config::max_tablet_version_num) {
+ //trigger quick compaction
+ if (config::enable_quick_compaction) {
+ StorageEngine::instance()->submit_quick_compaction_task(_tablet);
}
+ LOG(WARNING) << "failed to init delta writer. version count: " << _tablet->version_count()
+ << ", exceed limit: " << config::max_tablet_version_num
+ << ". tablet: " << _tablet->full_name();
+ return Status::OLAPInternalError(OLAP_ERR_TOO_MANY_VERSION);
}
{
diff --git a/be/src/olap/merger.cpp b/be/src/olap/merger.cpp
index 94d72ccc9a..ace3d6b39a 100644
--- a/be/src/olap/merger.cpp
+++ b/be/src/olap/merger.cpp
@@ -81,6 +81,10 @@ Status Merger::merge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
dst_rowset_writer->add_row(row_cursor),
"failed to write row when merging rowsets of tablet " + tablet->full_name());
output_rows++;
+ LOG_IF(INFO, config::row_step_for_compaction_merge_log != 0 &&
+ output_rows % config::row_step_for_compaction_merge_log == 0)
+ << "Merge rowsets stay alive. "
+ << "tablet=" << tablet->full_name() << ", merged rows=" << output_rows;
// the memory allocate by mem pool has been copied,
// so we should release memory immediately
mem_pool->clear();
diff --git a/be/src/olap/olap_server.cpp b/be/src/olap/olap_server.cpp
index 5a49c9c99c..845031c229 100644
--- a/be/src/olap/olap_server.cpp
+++ b/be/src/olap/olap_server.cpp
@@ -79,6 +79,10 @@ Status StorageEngine::start_bg_threads() {
.set_min_threads(config::max_cumu_compaction_threads)
.set_max_threads(config::max_cumu_compaction_threads)
.build(&_cumu_compaction_thread_pool);
+ ThreadPoolBuilder("SmallCompactionTaskThreadPool")
+ .set_min_threads(config::quick_compaction_max_threads)
+ .set_max_threads(config::quick_compaction_max_threads)
+ .build(&_quick_compaction_thread_pool);
// compaction tasks producer thread
RETURN_IF_ERROR(Thread::create(
@@ -239,19 +243,22 @@ void StorageEngine::_disk_stat_monitor_thread_callback() {
}
void StorageEngine::check_cumulative_compaction_config() {
- int64_t promotion_size = config::compaction_promotion_size_mbytes;
- int64_t promotion_min_size = config::compaction_promotion_min_size_mbytes;
- int64_t compaction_min_size = config::compaction_min_size_mbytes;
+ int64_t size_based_promotion_size = config::cumulative_size_based_promotion_size_mbytes;
+ int64_t size_based_promotion_min_size = config::cumulative_size_based_promotion_min_size_mbytes;
+ int64_t size_based_compaction_lower_bound_size =
+ config::cumulative_size_based_compaction_lower_size_mbytes;
// check size_based_promotion_size must be greater than size_based_promotion_min_size and 2 * size_based_compaction_lower_bound_size
- int64_t should_min_promotion_size = std::max(promotion_min_size, 2 * compaction_min_size);
-
- if (promotion_size < should_min_promotion_size) {
- promotion_size = should_min_promotion_size;
- LOG(WARNING) << "the config promotion_size is adjusted to "
- "promotion_min_size or 2 * "
- "compaction_min_size "
- << should_min_promotion_size << ", because size_based_promotion_size is small";
+ int64_t should_min_size_based_promotion_size =
+ std::max(size_based_promotion_min_size, 2 * size_based_compaction_lower_bound_size);
+
+ if (size_based_promotion_size < should_min_size_based_promotion_size) {
+ size_based_promotion_size = should_min_size_based_promotion_size;
+ LOG(WARNING) << "the config size_based_promotion_size is adjusted to "
+ "size_based_promotion_min_size or 2 * "
+ "size_based_compaction_lower_bound_size "
+ << should_min_size_based_promotion_size
+ << ", because size_based_promotion_size is small";
}
}
@@ -402,7 +409,7 @@ void StorageEngine::_compaction_tasks_producer_callback() {
int64_t last_base_score_update_time = 0;
static const int64_t check_score_interval_ms = 5000; // 5 secs
- int64_t interval = config::generate_compaction_tasks_interval_ms;
+ int64_t interval = config::generate_compaction_tasks_min_interval_ms;
do {
if (!config::disable_auto_compaction) {
_adjust_compaction_thread_num();
@@ -462,9 +469,9 @@ void StorageEngine::_compaction_tasks_producer_callback() {
<< tablet->tablet_id() << ", err: " << st.get_error_msg();
}
}
- interval = config::generate_compaction_tasks_interval_ms;
+ interval = config::generate_compaction_tasks_min_interval_ms;
} else {
- interval = 5000; // 5s to check disable_auto_compaction
+ interval = config::check_auto_compaction_interval_seconds * 1000;
}
} while (!_stop_background_threads_latch.wait_for(std::chrono::milliseconds(interval)));
}
@@ -526,18 +533,18 @@ std::vector<TabletSharedPtr> StorageEngine::_generate_compaction_tasks(
? copied_cumu_map[data_dir]
: copied_base_map[data_dir],
&disk_max_score, _cumulative_compaction_policy);
- if (tablet != nullptr) {
- if (!tablet->tablet_meta()->tablet_schema()->disable_auto_compaction()) {
- if (need_pick_tablet) {
- tablets_compaction.emplace_back(tablet);
- }
- max_compaction_score = std::max(max_compaction_score, disk_max_score);
- } else {
- LOG_EVERY_N(INFO, 500)
- << "Tablet " << tablet->full_name()
- << " will be ignored by automatic compaction tasks since it's "
- << "set to disabled automatic compaction.";
+ if (tablet != nullptr &&
+ !tablet->tablet_meta()->tablet_schema()->disable_auto_compaction()) {
+ if (need_pick_tablet) {
+ tablets_compaction.emplace_back(tablet);
}
+ max_compaction_score = std::max(max_compaction_score, disk_max_score);
+ } else if (tablet != nullptr &&
+ tablet->tablet_meta()->tablet_schema()->disable_auto_compaction()) {
+ LOG_EVERY_N(INFO, 500)
+ << "Tablet " << tablet->full_name()
+ << " will be ignored by automatic compaction tasks since it's "
+ << "set to disabled automatic compaction.";
}
}
}
@@ -656,10 +663,29 @@ Status StorageEngine::submit_compaction_task(TabletSharedPtr tablet,
if (tablet->get_cumulative_compaction_policy() == nullptr) {
tablet->set_cumulative_compaction_policy(_cumulative_compaction_policy);
}
- tablet->set_skip_compaction(false);
return _submit_compaction_task(tablet, compaction_type);
}
+Status StorageEngine::_handle_quick_compaction(TabletSharedPtr tablet) {
+ CumulativeCompaction compact(tablet);
+ compact.quick_rowsets_compact();
+ _pop_tablet_from_submitted_compaction(tablet, CompactionType::CUMULATIVE_COMPACTION);
+ return Status::OK();
+}
+
+Status StorageEngine::submit_quick_compaction_task(TabletSharedPtr tablet) {
+ bool already_exist =
+ _push_tablet_into_submitted_compaction(tablet, CompactionType::CUMULATIVE_COMPACTION);
+ if (already_exist) {
+ return Status::AlreadyExist(
+ "compaction task has already been submitted, tablet_id={}, compaction_type={}.",
+ tablet->tablet_id(), CompactionType::CUMULATIVE_COMPACTION);
+ }
+ _quick_compaction_thread_pool->submit_func(
+ std::bind<void>(&StorageEngine::_handle_quick_compaction, this, tablet));
+ return Status::OK();
+}
+
void StorageEngine::_cooldown_tasks_producer_callback() {
int64_t interval = config::generate_cooldown_task_interval_sec;
do {
diff --git a/be/src/olap/storage_engine.cpp b/be/src/olap/storage_engine.cpp
index 4f50dd7cba..183e4fa5ba 100644
--- a/be/src/olap/storage_engine.cpp
+++ b/be/src/olap/storage_engine.cpp
@@ -153,6 +153,10 @@ StorageEngine::~StorageEngine() {
_cumu_compaction_thread_pool->shutdown();
}
+ if (_quick_compaction_thread_pool) {
+ _quick_compaction_thread_pool->shutdown();
+ }
+
if (_tablet_meta_checkpoint_thread_pool) {
_tablet_meta_checkpoint_thread_pool->shutdown();
}
diff --git a/be/src/olap/storage_engine.h b/be/src/olap/storage_engine.h
index d61625ac78..7d0dad4f6e 100644
--- a/be/src/olap/storage_engine.h
+++ b/be/src/olap/storage_engine.h
@@ -190,6 +190,7 @@ public:
void check_cumulative_compaction_config();
Status submit_compaction_task(TabletSharedPtr tablet, CompactionType compaction_type);
+ Status submit_quick_compaction_task(TabletSharedPtr tablet);
std::unique_ptr<ThreadPool>& tablet_publish_txn_thread_pool() {
return _tablet_publish_txn_thread_pool;
@@ -268,6 +269,8 @@ private:
Status _submit_compaction_task(TabletSharedPtr tablet, CompactionType compaction_type);
+ Status _handle_quick_compaction(TabletSharedPtr);
+
void _adjust_compaction_thread_num();
void _cooldown_tasks_producer_callback();
@@ -371,6 +374,7 @@ private:
HeartbeatFlags* _heartbeat_flags;
+ std::unique_ptr<ThreadPool> _quick_compaction_thread_pool;
std::unique_ptr<ThreadPool> _base_compaction_thread_pool;
std::unique_ptr<ThreadPool> _cumu_compaction_thread_pool;
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
index bd89c9428f..58f6f226d8 100644
--- a/be/src/olap/tablet.cpp
+++ b/be/src/olap/tablet.cpp
@@ -92,6 +92,8 @@ Tablet::Tablet(TabletMetaSharedPtr tablet_meta, DataDir* data_dir,
_newly_created_rowset_num(0),
_last_checkpoint_time(0),
_cumulative_compaction_type(cumulative_compaction_type),
+ _last_record_scan_count(0),
+ _last_record_scan_count_timestamp(time(nullptr)),
_is_clone_occurred(false),
_last_missed_version(-1),
_last_missed_time_s(0) {
@@ -932,6 +934,56 @@ void Tablet::calculate_cumulative_point() {
set_cumulative_layer_point(ret_cumulative_point);
}
+//find rowsets that rows less then "config::quick_compaction_max_rows"
+Status Tablet::pick_quick_compaction_rowsets(std::vector<RowsetSharedPtr>* input_rowsets,
+ int64_t* permits) {
+ int max_rows = config::quick_compaction_max_rows;
+ if (!config::enable_quick_compaction || max_rows <= 0) {
+ return Status::OK();
+ }
+ if (!init_succeeded()) {
+ return Status::OLAPInternalError(OLAP_ERR_CUMULATIVE_INVALID_PARAMETERS);
+ }
+ int max_series_num = 1000;
+
+ std::vector<std::vector<RowsetSharedPtr>> quick_compaction_rowsets(max_series_num);
+ int idx = 0;
+ std::shared_lock rdlock(_meta_lock);
+ std::vector<RowsetSharedPtr> sortedRowset;
+ for (auto& rs : _rs_version_map) {
+ sortedRowset.push_back(rs.second);
+ }
+ std::sort(sortedRowset.begin(), sortedRowset.end(), Rowset::comparator);
+ if (tablet_state() == TABLET_RUNNING) {
+ for (int i = 0; i < sortedRowset.size(); i++) {
+ bool is_delete = version_for_delete_predicate(sortedRowset[i]->version());
+ if (!is_delete && sortedRowset[i]->start_version() > 0 &&
+ sortedRowset[i]->start_version() > cumulative_layer_point()) {
+ if (sortedRowset[i]->num_rows() < max_rows) {
+ quick_compaction_rowsets[idx].push_back(sortedRowset[i]);
+ } else {
+ idx++;
+ if (idx > max_series_num) {
+ break;
+ }
+ }
+ }
+ }
+ if (quick_compaction_rowsets.size() == 0) return Status::OK();
+ std::vector<RowsetSharedPtr> result = quick_compaction_rowsets[0];
+ for (int i = 0; i < quick_compaction_rowsets.size(); i++) {
+ if (quick_compaction_rowsets[i].size() > result.size()) {
+ result = quick_compaction_rowsets[i];
+ }
+ }
+ for (int i = 0; i < result.size(); i++) {
+ *permits += result[i]->num_segments();
+ input_rowsets->push_back(result[i]);
+ }
+ }
+ return Status::OK();
+}
+
Status Tablet::split_range(const OlapTuple& start_key_strings, const OlapTuple& end_key_strings,
uint64_t request_block_row_count, std::vector<OlapTuple>* ranges) {
DCHECK(ranges != nullptr);
@@ -1425,6 +1477,18 @@ 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;
+}
+
Status Tablet::prepare_compaction_and_calculate_permits(CompactionType compaction_type,
TabletSharedPtr tablet, int64_t* permits) {
std::vector<RowsetSharedPtr> compaction_rowsets;
@@ -2153,31 +2217,4 @@ bool Tablet::check_all_rowset_segment() {
return true;
}
-void Tablet::set_skip_compaction(bool skip, CompactionType compaction_type, int64_t start) {
- if (!skip) {
- _skip_cumu_compaction = false;
- _skip_base_compaction = false;
- return;
- }
- if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
- _skip_cumu_compaction = true;
- _skip_cumu_compaction_ts = start;
- } else {
- DCHECK(compaction_type == CompactionType::BASE_COMPACTION);
- _skip_base_compaction = true;
- _skip_base_compaction_ts = start;
- }
-}
-
-bool Tablet::should_skip_compaction(CompactionType compaction_type, int64_t now) {
- if (compaction_type == CompactionType::CUMULATIVE_COMPACTION && _skip_cumu_compaction &&
- now < _skip_cumu_compaction_ts + 120) {
- return true;
- } else if (compaction_type == CompactionType::BASE_COMPACTION && _skip_base_compaction &&
- now < _skip_base_compaction_ts + 120) {
- return true;
- }
- return false;
-}
-
} // namespace doris
diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
index 353b5ad190..318fdc8748 100644
--- a/be/src/olap/tablet.h
+++ b/be/src/olap/tablet.h
@@ -80,6 +80,8 @@ public:
// Used in clone task, to update local meta when finishing a clone job
Status revise_tablet_meta(const std::vector<RowsetMetaSharedPtr>& rowsets_to_clone,
const std::vector<Version>& versions_to_delete);
+ Status pick_quick_compaction_rowsets(std::vector<RowsetSharedPtr>* input_rowsets,
+ int64_t* permits);
const int64_t cumulative_layer_point() const;
void set_cumulative_layer_point(int64_t new_point);
@@ -208,6 +210,10 @@ public:
_last_cumu_compaction_success_millis = millis;
}
+ void set_last_quick_compaction_success_time(int64_t millis) {
+ _last_quick_compaction_success_time_millis = millis;
+ }
+
int64_t last_base_compaction_success_time() { return _last_base_compaction_success_millis; }
void set_last_base_compaction_success_time(int64_t millis) {
_last_base_compaction_success_millis = millis;
@@ -257,6 +263,8 @@ public:
// return a json string to show the compaction status of this tablet
void get_compaction_status(std::string* json_result);
+ double calculate_scan_frequency();
+
Status prepare_compaction_and_calculate_permits(CompactionType compaction_type,
TabletSharedPtr tablet, int64_t* permits);
void execute_compaction(CompactionType compaction_type);
@@ -350,11 +358,6 @@ public:
void update_max_version_schema(const TabletSchemaSPtr& tablet_schema);
- void set_skip_compaction(bool skip,
- CompactionType compaction_type = CompactionType::CUMULATIVE_COMPACTION,
- int64_t start = -1);
- bool should_skip_compaction(CompactionType compaction_type, int64_t now);
-
private:
Status _init_once_action();
void _print_missed_versions(const std::vector<Version>& missed_versions) const;
@@ -436,6 +439,7 @@ private:
std::atomic<int64_t> _last_cumu_compaction_success_millis;
// timestamp of last base compaction success
std::atomic<int64_t> _last_base_compaction_success_millis;
+ std::atomic<int64_t> _last_quick_compaction_success_time_millis;
std::atomic<int64_t> _cumulative_point;
std::atomic<int32_t> _newly_created_rowset_num;
std::atomic<int64_t> _last_checkpoint_time;
@@ -444,6 +448,14 @@ private:
std::shared_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;
+
std::shared_ptr<CumulativeCompaction> _cumulative_compaction;
std::shared_ptr<BaseCompaction> _base_compaction;
// whether clone task occurred during the tablet is in thread pool queue to wait for compaction
@@ -458,12 +470,6 @@ private:
// Max schema_version schema from Rowset or FE
TabletSchemaSPtr _max_version_schema;
- bool _skip_cumu_compaction = false;
- int64_t _skip_cumu_compaction_ts;
-
- bool _skip_base_compaction = false;
- int64_t _skip_base_compaction_ts;
-
DISALLOW_COPY_AND_ASSIGN(Tablet);
public:
diff --git a/be/src/olap/tablet_manager.cpp b/be/src/olap/tablet_manager.cpp
index a1e368311e..7cc81b782e 100644
--- a/be/src/olap/tablet_manager.cpp
+++ b/be/src/olap/tablet_manager.cpp
@@ -641,16 +641,14 @@ TabletSharedPtr TabletManager::find_best_tablet_to_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 (const auto& tablets_shard : _tablets_shards) {
std::shared_lock rdlock(tablets_shard.lock);
for (const auto& tablet_map : tablets_shard.tablet_map) {
const TabletSharedPtr& tablet_ptr = tablet_map.second;
- if (tablet_ptr->should_skip_compaction(compaction_type, UnixSeconds())) {
- continue;
- }
if (!tablet_ptr->can_do_compaction(data_dir->path_hash(), compaction_type)) {
continue;
}
@@ -664,7 +662,7 @@ TabletSharedPtr TabletManager::find_best_tablet_to_compaction(
if (compaction_type == CompactionType::BASE_COMPACTION) {
last_failure_ms = tablet_ptr->last_base_compaction_failure_time();
}
- if (now_ms - last_failure_ms <= 5000) {
+ if (now_ms - last_failure_ms <= config::min_compaction_failure_interval_sec * 1000) {
VLOG_DEBUG << "Too often to check compaction, skip it. "
<< "compaction_type=" << compaction_type_str
<< ", last_failure_time_ms=" << last_failure_ms
@@ -690,13 +688,19 @@ TabletSharedPtr TabletManager::find_best_tablet_to_compaction(
uint32_t current_compaction_score = tablet_ptr->calc_compaction_score(
compaction_type, cumulative_compaction_policy);
- if (current_compaction_score < 5) {
- LOG(INFO) << "tablet set skip compaction, tablet_id: " << tablet_ptr->tablet_id();
- tablet_ptr->set_skip_compaction(true, compaction_type, UnixSeconds());
+
+ double scan_frequency = 0.0;
+ if (config::compaction_tablet_scan_frequency_factor != 0) {
+ scan_frequency = tablet_ptr->calculate_scan_frequency();
}
- if (current_compaction_score > highest_score) {
- highest_score = current_compaction_score;
+
+ 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;
}
}
@@ -707,6 +711,7 @@ 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;
*score = compaction_score;
}
diff --git a/be/test/olap/cumulative_compaction_policy_test.cpp b/be/test/olap/cumulative_compaction_policy_test.cpp
index 3f46c508d1..e0425c36b8 100644
--- a/be/test/olap/cumulative_compaction_policy_test.cpp
+++ b/be/test/olap/cumulative_compaction_policy_test.cpp
@@ -31,10 +31,10 @@ class TestSizeBasedCumulativeCompactionPolicy : public testing::Test {
public:
TestSizeBasedCumulativeCompactionPolicy() {}
void SetUp() {
- config::compaction_promotion_size_mbytes = 1024;
- config::compaction_promotion_ratio = 0.05;
- config::compaction_promotion_min_size_mbytes = 64;
- config::compaction_min_size_mbytes = 64;
+ config::cumulative_size_based_promotion_size_mbytes = 1024;
+ config::cumulative_size_based_promotion_ratio = 0.05;
+ config::cumulative_size_based_promotion_min_size_mbytes = 64;
+ config::cumulative_size_based_compaction_lower_size_mbytes = 64;
_tablet_meta = static_cast<TabletMetaSharedPtr>(new TabletMeta(
1, 2, 15673, 15674, 4, 5, TTabletSchema(), 6, {{7, 8}}, UniqueId(9, 10),
@@ -653,7 +653,7 @@ TEST_F(TestSizeBasedCumulativeCompactionPolicy, _calc_promotion_size_big) {
dynamic_cast<SizeBasedCumulativeCompactionPolicy*>(
_tablet->_cumulative_compaction_policy.get());
- EXPECT_EQ(1073741824, policy->_tablet_promotion_size);
+ EXPECT_EQ(1073741824, policy->_tablet_size_based_promotion_size);
}
TEST_F(TestSizeBasedCumulativeCompactionPolicy, _calc_promotion_size_small) {
@@ -671,7 +671,7 @@ TEST_F(TestSizeBasedCumulativeCompactionPolicy, _calc_promotion_size_small) {
SizeBasedCumulativeCompactionPolicy* policy =
dynamic_cast<SizeBasedCumulativeCompactionPolicy*>(
_tablet->_cumulative_compaction_policy.get());
- EXPECT_EQ(67108864, policy->_tablet_promotion_size);
+ EXPECT_EQ(67108864, policy->_tablet_size_based_promotion_size);
}
TEST_F(TestSizeBasedCumulativeCompactionPolicy, _level_size) {
diff --git a/docs/en/docs/admin-manual/config/be-config.md b/docs/en/docs/admin-manual/config/be-config.md
index 1e25375fcf..db36e24b8a 100644
--- a/docs/en/docs/admin-manual/config/be-config.md
+++ b/docs/en/docs/admin-manual/config/be-config.md
@@ -105,7 +105,7 @@ Default: 3
The number of threads making schema changes
-### `generate_compaction_tasks_interval_ms`
+### `generate_compaction_tasks_min_interval_ms`
Default: 10 (ms)
@@ -117,7 +117,13 @@ Default: true
Whether to enable vectorized compaction
-### `base_compaction_min_rowset_num`
+### `base_compaction_interval_seconds_since_last_operation`
+
+Default: 86400
+
+One of the triggering conditions of BaseCompaction: the interval since the last BaseCompaction
+
+### `base_compaction_num_cumulative_deltas`
Default: 5
@@ -154,7 +160,7 @@ Default: 5（MB）
Maximum disk write speed per second of BaseCompaction task
-### `base_compaction_min_data_ratio`
+### `base_cumulative_delta_ratio`
Default: 0.3 （30%）
@@ -218,6 +224,12 @@ Clean up pages that may be saved by the buffer pool
The maximum amount of memory available in the BE buffer pool. The buffer pool is a new memory management structure of BE, which manages the memory by the buffer page and enables spill data to disk. The memory for all concurrent queries will be allocated from the buffer pool. The current buffer pool only works on **AggregationNode** and **ExchangeNode**.
+### `check_auto_compaction_interval_seconds`
+
+* Type: int32
+* Description: Check the configuration of auto compaction in seconds when auto compaction disabled.
+* Default value: 5
+
### `check_consistency_worker_count`
Default: 1
@@ -339,34 +351,34 @@ Similar to `base_compaction_trace_threshold`.
If set to true, the `cumulative_compaction_trace_threshold` and `base_compaction_trace_threshold` won't work and log is disabled.
-### `compaction_promotion_size_mbytes`
+### `cumulative_size_based_promotion_size_mbytes`
* Type: int64
-* Description: The total disk size of the output rowset of cumulative compaction exceeds this configuration size, and the rowset will be used for base compaction. The unit is m bytes.
+* Description: Under the size_based policy, the total disk size of the output rowset of cumulative compaction exceeds this configuration size, and the rowset will be used for base compaction. The unit is m bytes.
* Default value: 1024
In general, if the configuration is less than 2G, in order to prevent the cumulative compression time from being too long, resulting in the version backlog.
-### `compaction_promotion_ratio`
+### `cumulative_size_based_promotion_ratio`
* Type: double
-* Description: When the total disk size of the cumulative compaction output rowset exceeds the configuration ratio of the base version rowset, the rowset will be used for base compaction.
+* Description: Under the size_based policy, when the total disk size of the cumulative compaction output rowset exceeds the configuration ratio of the base version rowset, the rowset will be used for base compaction.
* Default value: 0.05
Generally, it is recommended that the configuration should not be higher than 0.1 and lower than 0.02.
-### `compaction_promotion_min_size_mbytes`
+### `cumulative_size_based_promotion_min_size_mbytes`
* Type: int64
-* Description: If the total disk size of the output rowset of the cumulative compaction is lower than this configuration size, the rowset will not undergo base compaction and is still in the cumulative compaction process. The unit is m bytes.
+* Description: Under the size_based strategy, if the total disk size of the output rowset of the cumulative compaction is lower than this configuration size, the rowset will not undergo base compaction and is still in the cumulative compaction process. The unit is m bytes.
* Default value: 64
Generally, the configuration is within 512m. If the configuration is too large, the size of the early base version is too small, and base compaction has not been performed.
-### `compaction_min_size_mbytes`
+### `cumulative_size_based_compaction_lower_size_mbytes`
* Type: int64
-* Description: When the cumulative compaction is merged, the selected rowsets to be merged have a larger disk size than this configuration, then they are divided and merged according to the level policy. When it is smaller than this configuration, merge directly. The unit is m bytes.
+* Description: Under the size_based strategy, when the cumulative compaction is merged, the selected rowsets to be merged have a larger disk size than this configuration, then they are divided and merged according to the level policy. When it is smaller than this configuration, merge directly. The unit is m bytes.
* Default value: 64
Generally, the configuration is within 128m. Over configuration will cause more cumulative compaction write amplification.
@@ -740,13 +752,13 @@ Default: 3
The maximum number of consumers in a data consumer group, used for routine load
-### `cumulative_compaction_min_deltas`
+### `min_cumulative_compaction_num_singleton_deltas`
Default: 5
Cumulative compaction strategy: the minimum number of incremental files
-### `cumulative_compaction_max_deltas`
+### `max_cumulative_compaction_num_singleton_deltas`
Default: 1000
@@ -856,6 +868,13 @@ Default: 1024
Minimum read buffer size (in bytes)
+### `min_compaction_failure_interval_sec`
+
+* Type: int32
+* Description: During the cumulative compaction process, when the selected tablet fails to be merged successfully, it will wait for a period of time before it may be selected again. The waiting period is the value of this configuration.
+* Default value: 600
+* Unit: seconds
+
### `min_compaction_threads`
* Type: int32
@@ -1054,6 +1073,13 @@ Default: true
Check row nums for BE/CE and schema change. true is open, false is closed
+### `row_step_for_compaction_merge_log`
+
+* Type: int64
+* Description: Merge log will be printed for each "row_step_for_compaction_merge_log" rows merged during compaction. If the value is set to 0, merge log will not be printed.
+* Default value: 0
+* Dynamically modify: true
+
### `scan_context_gc_interval_min`
Default: 5
diff --git a/docs/zh-CN/docs/admin-manual/config/be-config.md b/docs/zh-CN/docs/admin-manual/config/be-config.md
index 64f1e18031..b6816b4c0e 100644
--- a/docs/zh-CN/docs/admin-manual/config/be-config.md
+++ b/docs/zh-CN/docs/admin-manual/config/be-config.md
@@ -101,7 +101,7 @@ BE 的配置项有两种方式进行配置：
进行schema change的线程数
-### `generate_compaction_tasks_interval_ms`
+### `generate_compaction_tasks_min_interval_ms`
默认值：10 （ms）
@@ -113,7 +113,13 @@ BE 的配置项有两种方式进行配置：
是否开启向量化compaction
-### `base_compaction_min_rowset_num`
+### `base_compaction_interval_seconds_since_last_operation`
+
+默认值：86400
+
+BaseCompaction触发条件之一：上一次BaseCompaction距今的间隔
+
+### `base_compaction_num_cumulative_deltas`
默认值：5
@@ -125,7 +131,7 @@ BaseCompaction触发条件之一：Cumulative文件数目要达到的限制，
BaseCompaction任务每秒写磁盘最大速度
-### `base_compaction_min_data_ratio`
+### `base_cumulative_delta_ratio`
默认值：0.3 （30%）
@@ -211,6 +217,12 @@ Metrics: {"filtered_rows":0,"input_row_num":3346807,"input_rowsets_count":42,"in
BE缓存池最大的内存可用量，buffer pool是BE新的内存管理结构，通过buffer page来进行内存管理，并能够实现数据的落盘。并发的所有查询的内存申请都会通过buffer pool来申请。当前buffer pool仅作用在**AggregationNode**与**ExchangeNode**。
+### `check_auto_compaction_interval_seconds`
+
+* 类型：int32
+* 描述：当自动执行compaction的功能关闭时，检查自动compaction开关是否被开启的时间间隔。
+* 默认值：5
+
### `check_consistency_worker_count`
默认值：1
@@ -335,34 +347,34 @@ BaseCompaction触发条件之一：Singleton文件大小限制，100MB
如果设置为true，`cumulative_compaction_trace_threshold` 和 `base_compaction_trace_threshold` 将不起作用。并且trace日志将关闭。
-### `compaction_promotion_size_mbytes`
+### `cumulative_size_based_promotion_size_mbytes`
* 类型：int64
-* 描述：cumulative compaction的输出rowset总磁盘大小超过了此配置大小，该rowset将用于base compaction。单位是m字节。
+* 描述：在size_based策略下，cumulative compaction的输出rowset总磁盘大小超过了此配置大小，该rowset将用于base compaction。单位是m字节。
* 默认值：1024
一般情况下，配置在2G以内，为了防止cumulative compaction时间过长，导致版本积压。
-### `compaction_promotion_ratio`
+### `cumulative_size_based_promotion_ratio`
* 类型：double
-* 描述：cumulative compaction的输出rowset总磁盘大小超过base版本rowset的配置比例时，该rowset将用于base compaction。
+* 描述：在size_based策略下，cumulative compaction的输出rowset总磁盘大小超过base版本rowset的配置比例时，该rowset将用于base compaction。
* 默认值：0.05
一般情况下，建议配置不要高于0.1，低于0.02。
-### `compaction_promotion_min_size_mbytes`
+### `cumulative_size_based_promotion_min_size_mbytes`
* 类型：int64
-* 描述：Cumulative compaction的输出rowset总磁盘大小低于此配置大小，该rowset将不进行base compaction，仍然处于cumulative compaction流程中。单位是m字节。
+* 描述：在size_based策略下，cumulative compaction的输出rowset总磁盘大小低于此配置大小，该rowset将不进行base compaction，仍然处于cumulative compaction流程中。单位是m字节。
* 默认值：64
一般情况下，配置在512m以内，配置过大会导致base版本早期的大小过小，一直不进行base compaction。
-### `compaction_min_size_mbytes`
+### `cumulative_size_based_compaction_lower_size_mbytes`
* 类型：int64
-* 描述：cumulative compaction进行合并时，选出的要进行合并的rowset的总磁盘大小大于此配置时，才按级别策略划分合并。小于这个配置时，直接执行合并。单位是m字节。
+* 描述：在size_based策略下，cumulative compaction进行合并时，选出的要进行合并的rowset的总磁盘大小大于此配置时，才按级别策略划分合并。小于这个配置时，直接执行合并。单位是m字节。
* 默认值：64
一般情况下，配置在128m以内，配置过大会导致cumulative compaction写放大较多。
@@ -741,13 +753,13 @@ soft limit是指站单节点导入内存上限的比例。例如所有导入任
一个数据消费者组中的最大消费者数量，用于routine load
-### `cumulative_compaction_min_deltas`
+### `min_cumulative_compaction_num_singleton_deltas`
默认值：5
cumulative compaction策略：最小增量文件的数量
-### `cumulative_compaction_max_deltas`
+### `max_cumulative_compaction_num_singleton_deltas`
默认值：1000
@@ -857,6 +869,13 @@ txn 管理器中每个 txn_partition_map 的最大 txns 数，这是一种自我
最小读取缓冲区大小（以字节为单位）
+### `min_compaction_failure_interval_sec`
+
+* 类型：int32
+* 描述：在 cumulative compaction 过程中，当选中的 tablet 没能成功的进行版本合并，则会等待一段时间后才会再次有可能被选中。等待的这段时间就是这个配置的值。
+* 默认值：5
+* 单位：秒
+
### `min_compaction_threads`
* 类型：int32
@@ -1055,6 +1074,13 @@ routine load任务的线程池大小。 这应该大于 FE 配置 'max_concurren
检查 BE/CE 和schema更改的行号。 true 是打开的，false 是关闭的。
+### `row_step_for_compaction_merge_log`
+
+* 类型：int64
+* 描述：Compaction执行过程中，每次合并row_step_for_compaction_merge_log行数据会打印一条LOG。如果该参数被设置为0，表示merge过程中不需要打印LOG。
+* 默认值： 0
+* 可动态修改：是
+
### `scan_context_gc_interval_min`
默认值：5
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org