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_r311401908
##########
File path: be/src/olap/base_compaction.cpp
##########
@@ -16,511 +16,145 @@
// 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;
- }
+BaseCompaction::BaseCompaction(TabletSharedPtr tablet)
+ : Compaction(tablet) { }
- _need_merged_versions = candidate_versions;
+BaseCompaction::~BaseCompaction() { }
- 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);
- }
-
- // 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;
-
- _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;
+OLAPStatus BaseCompaction::pick_rowsets_to_compact() {
+ _input_rowsets.clear();
+ _tablet->pick_candicate_rowsets_to_base_compaction(&_input_rowsets);
+ if (_input_rowsets.size() == 0 || _input_rowsets.size() == 1) {
Review comment:
```suggestion
if (_input_rowsets.size() <= 1) {
```
----------------------------------------------------------------
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]
