imay commented on a change in pull request #1576: Change cumulative compaction
for decoupling storage from compution
URL: https://github.com/apache/incubator-doris/pull/1576#discussion_r312031427
##########
File path: be/src/olap/base_compaction.cpp
##########
@@ -16,511 +16,151 @@
// under the License.
#include "olap/base_compaction.h"
-
-#include <algorithm>
-#include <list>
-#include <map>
-#include <string>
-#include <vector>
-
-#include "olap/delete_handler.h"
-#include "olap/merger.h"
-#include "olap/rowset/column_data.h"
-#include "olap/storage_engine.h"
-#include "olap/tablet_meta.h"
-#include "olap/rowset/segment_group.h"
-#include "olap/tablet.h"
-#include "olap/utils.h"
#include "util/doris_metrics.h"
-using std::list;
-using std::map;
-using std::string;
-using std::vector;
-
namespace doris {
-OLAPStatus BaseCompaction::init(TabletSharedPtr tablet, bool
is_manual_trigger) {
- // 表在首次查询或PUSH等操作时，会被加载到内存
- // 如果表没有被加载，表明该表上目前没有任何操作，所以不进行BE操作
- if (!tablet->init_succeeded()) {
- return OLAP_ERR_INPUT_PARAMETER_ERROR;
- }
-
- LOG(INFO) << "init base compaction handler. [tablet=" <<
tablet->full_name() << "]";
-
- _tablet = tablet;
-
- // 1. 尝试取得base compaction的锁
- if (!_try_base_compaction_lock()) {
- LOG(WARNING) << "another base compaction is running. tablet=" <<
tablet->full_name();
- return OLAP_ERR_BE_TRY_BE_LOCK_ERROR;
- }
-
- // 2. 检查是否满足base compaction触发策略
- VLOG(3) << "check whether satisfy base compaction policy.";
- bool is_policy_satisfied = false;
- vector<Version> candidate_versions;
- is_policy_satisfied = _check_whether_satisfy_policy(is_manual_trigger,
&candidate_versions);
-
- // 2.1 如果不满足触发策略，则直接释放base compaction锁, 返回错误码
- if (!is_policy_satisfied) {
- _release_base_compaction_lock();
- return OLAP_ERR_BE_NO_SUITABLE_VERSION;
- }
-
- // 2.2 如果满足触发策略，触发base compaction
- // 不释放base compaction锁, 在run()完成之后再释放
- if (!_validate_need_merged_versions(candidate_versions)) {
- LOG(FATAL) << "error! invalid need merged versions";
- _release_base_compaction_lock();
- return OLAP_ERR_BE_INVALID_NEED_MERGED_VERSIONS;
- }
-
- _need_merged_versions = candidate_versions;
-
- return OLAP_SUCCESS;
-}
-
-OLAPStatus BaseCompaction::run() {
- LOG(INFO) << "start base compaction. tablet=" << _tablet->full_name()
- << ", old_base_version=" << _old_base_version.second
- << ", new_base_version=" << _new_base_version.second;
-
- OLAPStatus res = OLAP_SUCCESS;
- OlapStopWatch stage_watch;
-
- // 1. 计算新base的version hash
- VersionHash new_base_version_hash;
- res = _tablet->compute_all_versions_hash(_need_merged_versions,
&new_base_version_hash);
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to calculate new base version hash. tablet=" <<
_tablet->full_name()
- << ", new_base_version=" << _new_base_version.second;
- _garbage_collection();
- return res;
- }
-
- VLOG(10) << "new_base_version_hash:" << new_base_version_hash;
-
- // 2. 获取生成新base需要的data sources
- vector<RowsetSharedPtr> rowsets;
- res = _tablet->capture_consistent_rowsets(_need_merged_versions, &rowsets);
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to acquire need data sources. tablet=" <<
_tablet->full_name()
- << ", version=" << _new_base_version.second;
- _garbage_collection();
- return res;
- }
-
- {
-
DorisMetrics::base_compaction_deltas_total.increment(_need_merged_versions.size());
- int64_t merge_bytes = 0;
- for (auto& rowset : rowsets) {
- merge_bytes += rowset->data_disk_size();
- }
- DorisMetrics::base_compaction_bytes_total.increment(merge_bytes);
- }
+BaseCompaction::BaseCompaction(TabletSharedPtr tablet)
+ : Compaction(tablet) { }
- // 3. 执行base compaction
- // 执行过程可能会持续比较长时间
- stage_watch.reset();
- RowsetId rowset_id = 0;
- RETURN_NOT_OK(_tablet->next_rowset_id(&rowset_id));
- RowsetWriterContext context;
- context.rowset_id = rowset_id;
- context.tablet_uid = _tablet->tablet_uid();
- context.tablet_id = _tablet->tablet_id();
- context.partition_id = _tablet->partition_id();
- context.tablet_schema_hash = _tablet->schema_hash();
- context.rowset_type = ALPHA_ROWSET;
- context.rowset_path_prefix = _tablet->tablet_path();
- context.tablet_schema = &(_tablet->tablet_schema());
- context.rowset_state = VISIBLE;
- context.data_dir = _tablet->data_dir();
- context.version = _new_base_version;
- context.version_hash = new_base_version_hash;
+BaseCompaction::~BaseCompaction() { }
- _rs_writer.reset(new (std::nothrow)AlphaRowsetWriter());
- if (_rs_writer == nullptr) {
- LOG(WARNING) << "fail to new rowset.";
- _garbage_collection();
- return OLAP_ERR_MALLOC_ERROR;
- }
- RETURN_NOT_OK(_rs_writer->init(context));
- res = _do_base_compaction(new_base_version_hash, rowsets);
- _tablet->data_dir()->remove_pending_ids(ROWSET_ID_PREFIX +
std::to_string(_rs_writer->rowset_id()));
- // 释放不再使用的ColumnData对象
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to do base version. tablet=" <<
_tablet->full_name()
- << ", version=" << _new_base_version.second;
- _garbage_collection();
- return res;
+OLAPStatus BaseCompaction::compact() {
+ if (!_tablet->init_succeeded()) {
+ return OLAP_ERR_INPUT_PARAMETER_ERROR;
}
- // validate that delete action is right
- // if error happened, sleep 1 hour. Report a fatal log every 1 minute
- if (_validate_delete_file_action() != OLAP_SUCCESS) {
- LOG(WARNING) << "failed to do base compaction. delete action has
error.";
- _garbage_collection();
- return OLAP_ERR_BE_ERROR_DELETE_ACTION;
+ MutexLock lock(_tablet->get_base_lock(), TRY_LOCK);
+ if (!lock.own_lock()) {
+ LOG(WARNING) << "another base compaction is running. tablet=" <<
_tablet->full_name();
+ return OLAP_ERR_BE_TRY_BE_LOCK_ERROR;
}
- // 4. make new versions visable.
- // If success, remove files belong to old versions;
- // If fail, gc files belong to new versions.
- vector<RowsetSharedPtr> unused_rowsets;
- vector<Version> unused_versions;
- _get_unused_versions(&unused_versions);
- res = _tablet->capture_consistent_rowsets(unused_versions,
&unused_rowsets);
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to capture consistent rowsets. tablet=" <<
_tablet->full_name()
- << ", version=" << _new_base_version.second;
- _garbage_collection();
- return res;
- }
+ // 1. pick rowsets to compact
+ RETURN_NOT_OK(pick_rowsets_to_compact());
- res = _update_header(unused_rowsets);
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to update header. tablet=" <<
_tablet->full_name()
- << ", version=" << _new_base_version.first << "-" <<
_new_base_version.second;
- _garbage_collection();
- return res;
- }
- _delete_old_files(&unused_rowsets);
+ // 2. do base compaction, merge rowsets
+ RETURN_NOT_OK(do_compaction());
- _release_base_compaction_lock();
+ // 3. set state to success
+ _state = CompactionState::SUCCESS;
- LOG(INFO) << "succeed to do base compaction. tablet=" <<
_tablet->full_name()
- << ", base_version=" << _new_base_version.first << "-" <<
_new_base_version.second
- << ". elapsed time of doing base compaction"
- << ", time=" << stage_watch.get_elapse_second() << "s";
+ // 4. garbage collect input rowsets after base compaction
+ RETURN_NOT_OK(gc_unused_rowsets());
return OLAP_SUCCESS;
}
-static bool version_comparator(const Version& lhs, const Version& rhs) {
- return lhs.second < rhs.second;
-}
-
-bool BaseCompaction::_check_whether_satisfy_policy(bool is_manual_trigger,
- vector<Version>*
candidate_versions) {
- ReadLock rdlock(_tablet->get_header_lock_ptr());
- int64_t cumulative_layer_point = _tablet->cumulative_layer_point();
- if (cumulative_layer_point == -1) {
- LOG(FATAL) << "tablet has an unreasonable cumulative layer point.
[tablet='" << _tablet->full_name()
- << "' cumulative_layer_point=" << cumulative_layer_point <<
"]";
- return false;
- }
-
- // 为了后面计算方便，我们在这里先将cumulative_layer_point减1
- --cumulative_layer_point;
-
- vector<Version> path_versions;
- if (OLAP_SUCCESS != _tablet->capture_consistent_versions(Version(0,
cumulative_layer_point), &path_versions)) {
- LOG(WARNING) << "fail to select shortest version path. start=0, end="
<< cumulative_layer_point;
- return false;
- }
-
- //
base_compaction_layer_point应该为cumulative_layer_point之前，倒数第2个cumulative文件的end
version
- int64_t base_creation_time = 0;
- size_t base_size = 0;
- int32_t base_compaction_layer_point = -1;
- for (unsigned int index = 0; index < path_versions.size(); ++index) {
- Version temp = path_versions[index];
- // base文件
- if (temp.first == 0) {
- _old_base_version = temp;
- base_size = _tablet->get_rowset_size_by_version(temp);
- base_creation_time =
_tablet->get_rowset_by_version(temp)->creation_time();
- continue;
- }
-
- if (temp.second == cumulative_layer_point) {
- base_compaction_layer_point = temp.first - 1;
- _latest_cumulative = temp;
- _new_base_version = Version(0, base_compaction_layer_point);
- }
- }
-
- // 只有1个base文件和1个delta文件
- if (base_compaction_layer_point == -1) {
- VLOG(3) << "can't do base compaction: no cumulative files."
- << "tablet=" << _tablet->full_name()
- << ", base_version=0-" << _old_base_version.second
- << ", cumulative_layer_point=" << cumulative_layer_point + 1;
- return false;
- }
-
- // 只有1个cumulative文件
- if (base_compaction_layer_point == _old_base_version.second) {
- VLOG(3) << "can't do base compaction: only one cumulative file."
- << "tablet=" << _tablet->full_name()
- << ", base_version=0-" << _old_base_version.second
- << ", cumulative_layer_point=" << cumulative_layer_point + 1;
- return false;
- }
-
- // 使用最短路径算法，选择可合并的cumulative版本
- if (OLAP_SUCCESS !=
_tablet->capture_consistent_versions(_new_base_version, candidate_versions)) {
- LOG(WARNING) << "fail to select shortest version path."
- << "start=" << _new_base_version.first
- << ", end=" << _new_base_version.second;
- return false;
+OLAPStatus BaseCompaction::pick_rowsets_to_compact() {
+ _input_rowsets.clear();
+ _tablet->pick_candicate_rowsets_to_base_compaction(&_input_rowsets);
+ if (_input_rowsets.size() <= 1) {
+ LOG(INFO) << "There is no enough rowsets to cumulative compaction."
+ << ", the size of rowsets to compact=" <<
_input_rowsets.size()
+ << ", cumulative_point=" <<
_tablet->cumulative_layer_point();
+ return OLAP_ERR_CUMULATIVE_NO_SUITABLE_VERSIONS;
}
- std::sort(candidate_versions->begin(), candidate_versions->end(),
version_comparator);
+ std::sort(_input_rowsets.begin(), _input_rowsets.end(),
Rowset::comparator);
+ RETURN_NOT_OK(check_version_continuity(_input_rowsets));
- // 如果是手动执行START_BASE_COMPACTION命令，则不检查base compaction policy,
- // 也不考虑删除版本过期问题, 只要有可以合并的cumulative，就执行base compaction
- if (is_manual_trigger) {
- VLOG(3) << "manual trigger base compaction. tablet=" <<
_tablet->full_name();
- return true;
- }
-
- // 统计可合并cumulative版本文件的总大小
- size_t cumulative_total_size = 0;
- for (vector<Version>::const_iterator version_iter =
candidate_versions->begin();
- version_iter != candidate_versions->end(); ++version_iter) {
- Version temp = *version_iter;
- // 跳过base文件
- if (temp.first == 0) {
- continue;
- }
- // cumulative文件
- cumulative_total_size += _tablet->get_rowset_size_by_version(temp);
- }
-
- // 检查是否满足base compaction的触发条件
- // 满足以下条件时触发base compaction: 触发条件1 || 触发条件2 || 触发条件3
- // 触发条件1：cumulative文件个数超过一个阈值
- const uint32_t base_compaction_num_cumulative_deltas
- = config::base_compaction_num_cumulative_deltas;
- // candidate_versions中包含base文件，所以这里减1
- if (candidate_versions->size() - 1 >=
base_compaction_num_cumulative_deltas) {
+ // 1. cumulative rowset must reach base_compaction_num_cumulative_deltas
threshold
+ if (_input_rowsets.size() > config::base_compaction_num_cumulative_deltas)
{
LOG(INFO) << "satisfy the base compaction policy. tablet="<<
_tablet->full_name()
- << ", num_cumulative_deltas=" << candidate_versions->size() - 1
- << ", base_compaction_num_cumulative_deltas=" <<
base_compaction_num_cumulative_deltas;
- return true;
+ << ", num_cumulative_rowsets=" << _input_rowsets.size() - 1
+ << ", base_compaction_num_cumulative_rowsets=" <<
config::base_compaction_num_cumulative_deltas;
+ return OLAP_SUCCESS;
+ }
+
+ // 2. the ratio between base rowset and all input cumulative rowsets
reachs the threshold
+ int64_t base_size = 0;
+ int64_t cumulative_total_size = 0;
+ for (auto& rowset : _input_rowsets) {
+ if (rowset->start_version() != 0) {
+ cumulative_total_size += rowset->data_disk_size();
+ } else {
+ base_size = rowset->data_disk_size();
+ }
}
- // 触发条件2：所有cumulative文件的大小超过base文件大小的某一比例
- const double base_cumulative_delta_ratio =
config::base_cumulative_delta_ratio;
+ double base_cumulative_delta_ratio = config::base_cumulative_delta_ratio;
double cumulative_base_ratio = static_cast<double>(cumulative_total_size)
/ base_size;
+
if (cumulative_base_ratio > base_cumulative_delta_ratio) {
LOG(INFO) << "satisfy the base compaction policy. tablet=" <<
_tablet->full_name()
- << ", cumualtive_total_size=" << cumulative_total_size
- << ", base_size=" << base_size
- << ", cumulative_base_ratio=" << cumulative_base_ratio
- << ", policy_ratio=" << base_cumulative_delta_ratio;
- return true;
- }
-
- // 触发条件3：距离上一次进行base compaction已经超过设定的间隔时间
- const uint32_t interval_since_last_operation =
config::base_compaction_interval_seconds_since_last_operation;
- int64_t interval_since_last_be = time(NULL) - base_creation_time;
- if (interval_since_last_be > interval_since_last_operation) {
+ << ", cumualtive_total_size=" << cumulative_total_size
+ << ", base_size=" << base_size
+ << ", cumulative_base_ratio=" << cumulative_base_ratio
+ << ", policy_ratio=" << base_cumulative_delta_ratio;
+ return OLAP_SUCCESS;
+ }
+
+ // 3. the interval since last base compaction reachs 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_since_last_base_compaction = time(NULL) -
base_creation_time;
+ if (interval_since_last_base_compaction > interval_threshold) {
LOG(INFO) << "satisfy the base compaction policy. tablet=" <<
_tablet->full_name()
- << ", interval_since_last_be=" << interval_since_last_be
- << ", policy_interval=" << interval_since_last_operation;
- return true;
+ << ", interval_since_last_base_compaction=" <<
interval_since_last_base_compaction
+ << ", interval_threshold=" << interval_threshold;
+ return OLAP_SUCCESS;
}
- VLOG(3) << "don't satisfy the base compaction policy. tablet=" <<
_tablet->full_name()
- << ", cumulative_files_number=" << candidate_versions->size() - 1
- << ", cumulative_base_ratio=" << cumulative_base_ratio
- << ", interval_since_last_be=" << interval_since_last_be;
-
- return false;
+ LOG(INFO) << "don't satisfy the base compaction policy. tablet=" <<
_tablet->full_name()
+ << ", num_cumulative_rowsets=" << _input_rowsets.size() - 1
+ << ", cumulative_base_ratio=" << cumulative_base_ratio
+ << ", interval_since_last_base_compaction=" <<
interval_since_last_base_compaction;
+ return OLAP_ERR_BE_NO_SUITABLE_VERSION;
}
-OLAPStatus BaseCompaction::_do_base_compaction(VersionHash
new_base_version_hash,
- const vector<RowsetSharedPtr>&
rowsets) {
- OlapStopWatch watch;
- vector<RowsetReaderSharedPtr> rs_readers;
- for (auto& rowset : rowsets) {
- RowsetReaderSharedPtr rs_reader(rowset->create_reader());
- if (rs_reader == nullptr) {
- LOG(WARNING) << "rowset create reader failed. rowset:" <<
rowset->rowset_id();
- return OLAP_ERR_ROWSET_CREATE_READER;
- }
- rs_readers.push_back(rs_reader);
+OLAPStatus BaseCompaction::do_compaction() {
Review comment:
I think it's better to encapsulate all the following logic into base class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
