This is an automated email from the ASF dual-hosted git repository.
dataroaring 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 3dde97bff1 (compaction) opt compaction task producer and quick
compaction (#13495) (#14535)
3dde97bff1 is described below
commit 3dde97bff1bb19fd293e2df60d3df443e0a1a776
Author: yixiutt <[email protected]>
AuthorDate: Fri Dec 2 10:07:44 2022 +0800
(compaction) opt compaction task producer and quick compaction (#13495)
(#14535)
1.remove quick_compaction's rowset pick policy, call cu compaction when
trigger
quick compaction
2. skip tablet's compaction task when compaction score is too small
Co-authored-by: yixiutt <[email protected]>
---
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 | 5 --
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 | 29 ++++---
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, 203 insertions(+), 432 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 3f5ca98021..aa07c6834f 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -726,29 +726,25 @@ void
TaskWorkerPool::_publish_version_worker_thread_callback() {
.error(status);
finish_task_request.__set_error_tablet_ids(error_tablet_ids);
} else {
- 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];
+ 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;
}
+ } 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 d2687b83a2..9faf33ef00 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -260,53 +260,35 @@ CONF_Int32(vertical_compaction_max_row_source_memory_mb,
"200");
// In ordered data compaction, min segment size for input rowset
CONF_mInt32(ordered_data_compaction_min_segment_size, "10485760");
-// check the configuration of auto compaction in seconds when auto compaction
disabled
-CONF_mInt32(check_auto_compaction_interval_seconds, "5");
+// 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");
-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");
-
-// dup key not compaction big files
-CONF_Bool(enable_dup_key_base_compaction_skip_big_file, "true");
+CONF_mInt64(base_compaction_min_rowset_num, "5");
+CONF_mDouble(base_compaction_min_data_ratio, "0.3");
CONF_mInt64(base_compaction_dup_key_max_file_size_mbytes, "1024");
-// In size_based policy, output rowset of cumulative compaction total disk
size exceed this config size,
+// 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(cumulative_size_based_promotion_size_mbytes, "1024");
+CONF_mInt64(compaction_promotion_size_mbytes, "1024");
-// In size_based policy, output rowset of cumulative compaction total disk
size exceed this config ratio of
+// 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(cumulative_size_based_promotion_ratio, "0.05");
+CONF_mDouble(compaction_promotion_ratio, "0.05");
-// In size_based policy, the smallest size of rowset promotion. When the
rowset is less than this config, this
+// 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(cumulative_size_based_promotion_min_size_mbytes, "64");
+CONF_mInt64(compaction_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(cumulative_size_based_compaction_lower_size_mbytes, "64");
+CONF_mInt64(compaction_min_size_mbytes, "64");
// cumulative compaction policy: min and max delta file's number
-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");
+CONF_mInt64(cumulative_compaction_min_deltas, "5");
+CONF_mInt64(cumulative_compaction_max_deltas, "1000");
// This config can be set to limit thread number in segcompaction thread pool.
CONF_mInt32(seg_compaction_max_threads, "10");
@@ -315,7 +297,7 @@ CONF_mInt32(seg_compaction_max_threads, "10");
CONF_mInt64(total_permits_for_compaction_score, "10000");
// sleep interval in ms after generated compaction tasks
-CONF_mInt32(generate_compaction_tasks_min_interval_ms, "10");
+CONF_mInt32(generate_compaction_tasks_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.
@@ -329,23 +311,17 @@ 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");
@@ -807,14 +783,7 @@ CONF_mInt32(orc_natural_read_size_mb, "8");
// if it is lower than a specific threshold, the predicate will be disabled.
CONF_mInt32(bloom_filter_predicate_check_row_num, "204800");
-//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");
+CONF_Bool(enable_decimalv3, "false");
// cooldown task configs
CONF_Int32(cooldown_thread_num, "5");
diff --git a/be/src/olap/base_compaction.cpp b/be/src/olap/base_compaction.cpp
index 0e1d737b15..ba71d2fd5d 100644
--- a/be/src/olap/base_compaction.cpp
+++ b/be/src/olap/base_compaction.cpp
@@ -87,10 +87,9 @@ Status BaseCompaction::execute_compact_impl() {
}
void BaseCompaction::_filter_input_rowset() {
- // if enable dup key skip big file and no delete predicate
+ // if dup_key and no delete predicate
// we skip big files too save resources
- if (!config::enable_dup_key_base_compaction_skip_big_file ||
- _tablet->keys_type() != KeysType::DUP_KEYS ||
_tablet->delete_predicates().size() != 0) {
+ if (_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;
@@ -144,11 +143,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_num_cumulative_deltas)
{
+ if (_input_rowsets.size() > config::base_compaction_min_rowset_num) {
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_num_cumulative_deltas;
+ << config::base_compaction_min_rowset_num;
return Status::OK();
}
@@ -160,7 +159,7 @@ Status BaseCompaction::pick_rowsets_to_compact() {
cumulative_total_size += (*it)->data_disk_size();
}
- double base_cumulative_delta_ratio = config::base_cumulative_delta_ratio;
+ double min_data_ratio = config::base_compaction_min_data_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
@@ -168,18 +167,18 @@ Status BaseCompaction::pick_rowsets_to_compact() {
}
double cumulative_base_ratio = static_cast<double>(cumulative_total_size)
/ base_size;
- if (cumulative_base_ratio > base_cumulative_delta_ratio) {
+ if (cumulative_base_ratio > min_data_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_ratio=" << base_cumulative_delta_ratio;
+ << ", policy_min_data_ratio=" << min_data_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 =
config::base_compaction_interval_seconds_since_last_operation;
+ int64_t interval_threshold = 86400;
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 65797a8914..364fb976ed 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -57,57 +57,6 @@ 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");
uint32_t checksum_before;
@@ -391,6 +340,7 @@ 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()
<< " is_vertical=" << vertical_compaction << ". tablet=" <<
_tablet->full_name()
<< ", output_version=" << _output_version
@@ -399,8 +349,7 @@ 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 == nullptr ? "quick" : cumu_policy->name())
+ << "s. cumulative_compaction_policy=" << 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 bf11629fac..08cdcd34a4 100644
--- a/be/src/olap/compaction.h
+++ b/be/src/olap/compaction.h
@@ -48,7 +48,6 @@ 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 4736454c09..b2fca016c7 100644
--- a/be/src/olap/cumulative_compaction.cpp
+++ b/be/src/olap/cumulative_compaction.cpp
@@ -117,10 +117,9 @@ 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::max_cumulative_compaction_num_singleton_deltas,
- config::min_cumulative_compaction_num_singleton_deltas,
&_input_rowsets,
- &_last_delete_version, &compaction_score);
+ _tablet.get(), candidate_rowsets,
config::cumulative_compaction_max_deltas,
+ config::cumulative_compaction_min_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):
@@ -143,8 +142,7 @@ 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 =
-
config::base_compaction_interval_seconds_since_last_operation * 1000;
+ int64_t interval_threshold = 86400 * 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 599fad92be..35d65be797 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 size_based_promotion_size, double size_based_promotion_ratio,
- int64_t size_based_promotion_min_size, int64_t
size_based_compaction_lower_bound_size)
+ int64_t promotion_size, double promotion_ratio, int64_t
promotion_min_size,
+ int64_t compaction_min_size)
: CumulativeCompactionPolicy(),
- _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;
+ _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;
while (i_size >= 1024) {
_levels.push_back(i_size);
@@ -122,20 +122,19 @@ 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 * _size_based_promotion_ratio;
+ *promotion_size = base_size * _promotion_ratio;
- // 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;
+ // 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;
}
- _refresh_tablet_size_based_promotion_size(*promotion_size);
+ _refresh_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::_refresh_tablet_promotion_size(int64_t
promotion_size) {
+ _tablet_promotion_size = promotion_size;
}
void SizeBasedCumulativeCompactionPolicy::update_cumulative_point(
@@ -152,7 +151,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_size_based_promotion_size) {
+ if (total_size >= _tablet_promotion_size) {
tablet->set_cumulative_layer_point(output_rowset->end_version() +
1);
}
}
@@ -240,7 +239,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_size_based_promotion_size;
+ size_t promotion_size = _tablet_promotion_size;
int transient_size = 0;
*compaction_score = 0;
int64_t total_size = 0;
@@ -319,12 +318,10 @@ 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 < _size_based_compaction_lower_bound_size &&
- *compaction_score < min_compaction_score) {
+ if (total_size < _compaction_min_size && *compaction_score <
min_compaction_score) {
input_rowsets->clear();
*compaction_score = 0;
- } else if (total_size >= _size_based_compaction_lower_bound_size &&
- input_rowsets->size() == 1) {
+ } else if (total_size >= _compaction_min_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 16e9b3b19b..15454c1892 100644
--- a/be/src/olap/cumulative_compaction_policy.h
+++ b/be/src/olap/cumulative_compaction_policy.h
@@ -117,13 +117,10 @@ public:
/// it needs tablet pointer to access tablet method.
/// param tablet, the shared pointer of tablet
SizeBasedCumulativeCompactionPolicy(
- 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);
+ 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);
/// Destructor function of SizeBasedCumulativeCompactionPolicy.
~SizeBasedCumulativeCompactionPolicy() {}
@@ -166,24 +163,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 cumulative_size_based_promotion_min_size_mbytes
- /// and cumulative_size_based_promotion_size_mbytes
+ /// between compaction_promotion_min_size_mbytes
+ /// and compaction_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_size_based_promotion_size(int64_t promotion_size);
+ void _refresh_tablet_promotion_size(int64_t promotion_size);
private:
/// cumulative compaction promotion size, unit is byte.
- int64_t _size_based_promotion_size;
+ int64_t _promotion_size;
/// cumulative compaction promotion ratio of base rowset total disk size.
- double _size_based_promotion_ratio;
+ double _promotion_ratio;
/// cumulative compaction promotion min size, unit is byte.
- int64_t _size_based_promotion_min_size;
+ int64_t _promotion_min_size;
/// lower bound size to do compaction compaction.
- int64_t _size_based_compaction_lower_bound_size;
+ int64_t _compaction_min_size;
/// record tablet promotion size, it is updated each time when calculate
cumulative_compaction_score
- int64_t _tablet_size_based_promotion_size;
+ int64_t _tablet_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 d330e430fe..d75e0b63d4 100644
--- a/be/src/olap/delta_writer.cpp
+++ b/be/src/olap/delta_writer.cpp
@@ -111,15 +111,17 @@ Status DeltaWriter::init() {
}
// check tablet version number
- if (_tablet->version_count() > config::max_tablet_version_num) {
- //trigger quick compaction
- if (config::enable_quick_compaction) {
- StorageEngine::instance()->submit_quick_compaction_task(_tablet);
+ 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);
}
- 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 46bb15c96d..9217a88316 100644
--- a/be/src/olap/merger.cpp
+++ b/be/src/olap/merger.cpp
@@ -83,10 +83,6 @@ 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 6e39b0f093..a266d643b3 100644
--- a/be/src/olap/olap_server.cpp
+++ b/be/src/olap/olap_server.cpp
@@ -81,10 +81,6 @@ 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);
if (config::enable_segcompaction && config::enable_storage_vectorization) {
ThreadPoolBuilder("SegCompactionTaskThreadPool")
.set_min_threads(config::seg_compaction_max_threads)
@@ -251,22 +247,19 @@ void StorageEngine::_disk_stat_monitor_thread_callback() {
}
void StorageEngine::check_cumulative_compaction_config() {
- 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;
+ 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;
// 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_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";
+ 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";
}
}
@@ -417,7 +410,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_min_interval_ms;
+ int64_t interval = config::generate_compaction_tasks_interval_ms;
do {
if (!config::disable_auto_compaction) {
_adjust_compaction_thread_num();
@@ -477,9 +470,9 @@ void StorageEngine::_compaction_tasks_producer_callback() {
<< tablet->tablet_id() << ", err: " <<
st.get_error_msg();
}
}
- interval = config::generate_compaction_tasks_min_interval_ms;
+ interval = config::generate_compaction_tasks_interval_ms;
} else {
- interval = config::check_auto_compaction_interval_seconds * 1000;
+ interval = 5000; // 5s to check disable_auto_compaction
}
} while
(!_stop_background_threads_latch.wait_for(std::chrono::milliseconds(interval)));
}
@@ -541,18 +534,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 &&
-
!tablet->tablet_meta()->tablet_schema()->disable_auto_compaction()) {
- if (need_pick_tablet) {
- tablets_compaction.emplace_back(tablet);
+ 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.";
}
- 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.";
}
}
}
@@ -671,29 +664,10 @@ 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();
-}
-
Status StorageEngine::_handle_seg_compaction(BetaRowsetWriter* writer,
SegCompactionCandidatesSharedPtr
segments) {
writer->compact_segments(segments);
diff --git a/be/src/olap/storage_engine.cpp b/be/src/olap/storage_engine.cpp
index e08b3d562e..05a6b24a1e 100644
--- a/be/src/olap/storage_engine.cpp
+++ b/be/src/olap/storage_engine.cpp
@@ -135,14 +135,9 @@ StorageEngine::~StorageEngine() {
_cumu_compaction_thread_pool->shutdown();
}
- if (_quick_compaction_thread_pool) {
- _quick_compaction_thread_pool->shutdown();
- }
-
if (_seg_compaction_thread_pool) {
_seg_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 6de80ead25..3e9c5d5857 100644
--- a/be/src/olap/storage_engine.h
+++ b/be/src/olap/storage_engine.h
@@ -186,7 +186,6 @@ public:
void check_cumulative_compaction_config();
Status submit_compaction_task(TabletSharedPtr tablet, CompactionType
compaction_type);
- Status submit_quick_compaction_task(TabletSharedPtr tablet);
Status submit_seg_compaction_task(BetaRowsetWriter* writer,
SegCompactionCandidatesSharedPtr
segments);
@@ -267,8 +266,6 @@ 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();
@@ -366,7 +363,6 @@ 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;
std::unique_ptr<ThreadPool> _seg_compaction_thread_pool;
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
index 1c8e1a0905..d741203f1c 100644
--- a/be/src/olap/tablet.cpp
+++ b/be/src/olap/tablet.cpp
@@ -92,8 +92,6 @@ 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) {
@@ -939,56 +937,6 @@ 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);
@@ -1482,18 +1430,6 @@ 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;
@@ -2238,4 +2174,31 @@ 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 4a66e8824c..0990de435b 100644
--- a/be/src/olap/tablet.h
+++ b/be/src/olap/tablet.h
@@ -80,8 +80,6 @@ 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);
@@ -210,10 +208,6 @@ 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;
@@ -263,8 +257,6 @@ 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);
@@ -365,6 +357,11 @@ 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;
@@ -446,7 +443,6 @@ 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;
@@ -455,14 +451,6 @@ 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
@@ -477,6 +465,12 @@ 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 c877a8d0ec..ad16257544 100644
--- a/be/src/olap/tablet_manager.cpp
+++ b/be/src/olap/tablet_manager.cpp
@@ -641,14 +641,16 @@ TabletSharedPtr
TabletManager::find_best_tablet_to_compaction(
int64_t now_ms = UnixMillis();
const string& compaction_type_str =
compaction_type == CompactionType::BASE_COMPACTION ? "base" :
"cumulative";
- double highest_score = 0.0;
+ uint32_t highest_score = 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;
}
@@ -662,7 +664,11 @@ 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 <=
config::min_compaction_failure_interval_sec * 1000) {
+ if (now_ms - last_failure_ms <= 5000) {
+ VLOG_DEBUG << "Too often to check compaction, skip it. "
+ << "compaction_type=" << compaction_type_str
+ << ", last_failure_time_ms=" << last_failure_ms
+ << ", tablet_id=" << tablet_ptr->tablet_id();
continue;
}
@@ -684,19 +690,13 @@ TabletSharedPtr
TabletManager::find_best_tablet_to_compaction(
uint32_t current_compaction_score =
tablet_ptr->calc_compaction_score(
compaction_type, cumulative_compaction_policy);
-
- double scan_frequency = 0.0;
- if (config::compaction_tablet_scan_frequency_factor != 0) {
- scan_frequency = tablet_ptr->calculate_scan_frequency();
+ 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 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;
+ if (current_compaction_score > highest_score) {
+ highest_score = current_compaction_score;
compaction_score = current_compaction_score;
- tablet_scan_frequency = scan_frequency;
best_tablet = tablet_ptr;
}
}
@@ -707,7 +707,6 @@ 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 e0425c36b8..3f46c508d1 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::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;
+ 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;
_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_size_based_promotion_size);
+ EXPECT_EQ(1073741824, policy->_tablet_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_size_based_promotion_size);
+ EXPECT_EQ(67108864, policy->_tablet_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 6e5eda0d1a..83596c9c4d 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_min_interval_ms`
+### `generate_compaction_tasks_interval_ms`
Default: 10 (ms)
@@ -117,13 +117,7 @@ Default: true
Whether to enable vectorized compaction
-### `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`
+### `base_compaction_min_rowset_num`
Default: 5
@@ -160,7 +154,7 @@ Default: 5（MB）
Maximum disk write speed per second of BaseCompaction task
-### `base_cumulative_delta_ratio`
+### `base_compaction_min_data_ratio`
Default: 0.3 （30%）
@@ -224,12 +218,6 @@ 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
@@ -351,34 +339,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.
-### `cumulative_size_based_promotion_size_mbytes`
+### `compaction_promotion_size_mbytes`
* Type: int64
-* 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.
+* 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.
* 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.
-### `cumulative_size_based_promotion_ratio`
+### `compaction_promotion_ratio`
* Type: double
-* 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.
+* 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.
* Default value: 0.05
Generally, it is recommended that the configuration should not be higher than
0.1 and lower than 0.02.
-### `cumulative_size_based_promotion_min_size_mbytes`
+### `compaction_promotion_min_size_mbytes`
* Type: int64
-* 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.
+* 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.
* 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.
-### `cumulative_size_based_compaction_lower_size_mbytes`
+### `compaction_min_size_mbytes`
* Type: int64
-* 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.
+* 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.
* Default value: 64
Generally, the configuration is within 128m. Over configuration will cause
more cumulative compaction write amplification.
@@ -752,13 +740,13 @@ Default: 3
The maximum number of consumers in a data consumer group, used for routine load
-### `min_cumulative_compaction_num_singleton_deltas`
+### `cumulative_compaction_min_deltas`
Default: 5
Cumulative compaction strategy: the minimum number of incremental files
-### `max_cumulative_compaction_num_singleton_deltas`
+### `cumulative_compaction_max_deltas`
Default: 1000
@@ -880,13 +868,6 @@ 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
@@ -1085,13 +1066,6 @@ 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 40277d6a8e..ef51e44e9d 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_min_interval_ms`
+### `generate_compaction_tasks_interval_ms`
默认值：10 （ms）
@@ -113,13 +113,7 @@ BE 的配置项有两种方式进行配置：
是否开启向量化compaction
-### `base_compaction_interval_seconds_since_last_operation`
-
-默认值：86400
-
-BaseCompaction触发条件之一：上一次BaseCompaction距今的间隔
-
-### `base_compaction_num_cumulative_deltas`
+### `base_compaction_min_rowset_num`
默认值：5
@@ -131,7 +125,7 @@ BaseCompaction触发条件之一：Cumulative文件数目要达到的限制，
BaseCompaction任务每秒写磁盘最大速度
-### `base_cumulative_delta_ratio`
+### `base_compaction_min_data_ratio`
默认值：0.3 （30%）
@@ -217,12 +211,6 @@ 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
@@ -347,34 +335,34 @@ BaseCompaction触发条件之一：Singleton文件大小限制，100MB
如果设置为true，`cumulative_compaction_trace_threshold` 和
`base_compaction_trace_threshold` 将不起作用。并且trace日志将关闭。
-### `cumulative_size_based_promotion_size_mbytes`
+### `compaction_promotion_size_mbytes`
* 类型：int64
-* 描述：在size_based策略下，cumulative compaction的输出rowset总磁盘大小超过了此配置大小，该rowset将用于base
compaction。单位是m字节。
+* 描述：cumulative compaction的输出rowset总磁盘大小超过了此配置大小，该rowset将用于base
compaction。单位是m字节。
* 默认值：1024
一般情况下，配置在2G以内，为了防止cumulative compaction时间过长，导致版本积压。
-### `cumulative_size_based_promotion_ratio`
+### `compaction_promotion_ratio`
* 类型：double
-* 描述：在size_based策略下，cumulative
compaction的输出rowset总磁盘大小超过base版本rowset的配置比例时，该rowset将用于base compaction。
+* 描述：cumulative compaction的输出rowset总磁盘大小超过base版本rowset的配置比例时，该rowset将用于base
compaction。
* 默认值：0.05
一般情况下，建议配置不要高于0.1，低于0.02。
-### `cumulative_size_based_promotion_min_size_mbytes`
+### `compaction_promotion_min_size_mbytes`
* 类型：int64
-* 描述：在size_based策略下，cumulative compaction的输出rowset总磁盘大小低于此配置大小，该rowset将不进行base
compaction，仍然处于cumulative compaction流程中。单位是m字节。
+* 描述：Cumulative compaction的输出rowset总磁盘大小低于此配置大小，该rowset将不进行base
compaction，仍然处于cumulative compaction流程中。单位是m字节。
* 默认值：64
一般情况下，配置在512m以内，配置过大会导致base版本早期的大小过小，一直不进行base compaction。
-### `cumulative_size_based_compaction_lower_size_mbytes`
+### `compaction_min_size_mbytes`
* 类型：int64
-* 描述：在size_based策略下，cumulative
compaction进行合并时，选出的要进行合并的rowset的总磁盘大小大于此配置时，才按级别策略划分合并。小于这个配置时，直接执行合并。单位是m字节。
+* 描述：cumulative
compaction进行合并时，选出的要进行合并的rowset的总磁盘大小大于此配置时，才按级别策略划分合并。小于这个配置时，直接执行合并。单位是m字节。
* 默认值：64
一般情况下，配置在128m以内，配置过大会导致cumulative compaction写放大较多。
@@ -753,13 +741,13 @@ soft limit是指站单节点导入内存上限的比例。例如所有导入任
一个数据消费者组中的最大消费者数量，用于routine load
-### `min_cumulative_compaction_num_singleton_deltas`
+### `cumulative_compaction_min_deltas`
默认值：5
cumulative compaction策略：最小增量文件的数量
-### `max_cumulative_compaction_num_singleton_deltas`
+### `cumulative_compaction_max_deltas`
默认值：1000
@@ -881,13 +869,6 @@ txn 管理器中每个 txn_partition_map 的最大 txns 数，这是一种自我
最小读取缓冲区大小（以字节为单位）
-### `min_compaction_failure_interval_sec`
-
-* 类型：int32
-* 描述：在 cumulative compaction 过程中，当选中的 tablet
没能成功的进行版本合并，则会等待一段时间后才会再次有可能被选中。等待的这段时间就是这个配置的值。
-* 默认值：5
-* 单位：秒
-
### `min_compaction_threads`
* 类型：int32
@@ -1086,13 +1067,6 @@ 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: [email protected]
For additional commands, e-mail: [email protected]
