You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by da...@apache.org on 2024/02/07 13:04:59 UTC
(doris) branch master updated: [feature](merge-cloud) Refactor Compaction (#30839)
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 43fb17e0ae2 [feature](merge-cloud) Refactor Compaction (#30839)
43fb17e0ae2 is described below
commit 43fb17e0ae2e02ce74b6202eefc150b1a50ef0a7
Author: plat1ko <pl...@gmail.com>
AuthorDate: Wed Feb 7 21:04:52 2024 +0800
[feature](merge-cloud) Refactor Compaction (#30839)
---
be/src/common/status.h | 2 +-
be/src/http/action/compaction_action.cpp | 21 +-
be/src/olap/base_compaction.cpp | 31 +-
be/src/olap/base_compaction.h | 20 +-
be/src/olap/base_tablet.cpp | 4 +-
be/src/olap/cold_data_compaction.cpp | 53 +-
be/src/olap/cold_data_compaction.h | 17 +-
be/src/olap/compaction.cpp | 988 +++++++++------------
be/src/olap/compaction.h | 122 +--
be/src/olap/cumulative_compaction.cpp | 89 +-
be/src/olap/cumulative_compaction.h | 21 +-
be/src/olap/full_compaction.cpp | 60 +-
be/src/olap/full_compaction.h | 23 +-
be/src/olap/merger.cpp | 100 ++-
be/src/olap/merger.h | 28 +-
be/src/olap/olap_server.cpp | 30 +-
be/src/olap/rowset/rowset.cpp | 16 +
be/src/olap/rowset/rowset.h | 3 +
be/src/olap/rowset/segcompaction.cpp | 8 +-
be/src/olap/schema_change.cpp | 2 +-
be/src/olap/single_replica_compaction.cpp | 53 +-
be/src/olap/single_replica_compaction.h | 20 +-
be/src/olap/snapshot_manager.cpp | 17 -
be/src/olap/tablet.cpp | 20 +-
be/src/olap/tablet.h | 11 +-
be/src/olap/txn_manager.cpp | 4 +-
be/src/olap/txn_manager.h | 2 +-
be/src/pipeline/task_scheduler.cpp | 3 +-
be/src/runtime/fragment_mgr.cpp | 2 +-
be/src/vec/core/sort_cursor.h | 2 +-
.../compaction_delete_bitmap_calculator_test.cpp | 2 +-
be/test/olap/cumulative_compaction_policy_test.cpp | 36 -
...mulative_compaction_time_series_policy_test.cpp | 35 -
be/test/olap/delta_writer_test.cpp | 1 -
.../olap/engine_storage_migration_task_test.cpp | 2 -
be/test/olap/memtable_memory_limiter_test.cpp | 1 -
be/test/olap/ordered_data_compaction_test.cpp | 9 +-
be/test/olap/rowid_conversion_test.cpp | 4 +-
be/test/runtime/load_stream_test.cpp | 2 -
be/test/vec/olap/vertical_compaction_test.cpp | 20 +-
40 files changed, 852 insertions(+), 1032 deletions(-)
diff --git a/be/src/common/status.h b/be/src/common/status.h
index e5dcfe4fe81..b2a526c9bc8 100644
--- a/be/src/common/status.h
+++ b/be/src/common/status.h
@@ -495,7 +495,7 @@ public:
friend std::ostream& operator<<(std::ostream& ostr, const Status& status);
- std::string msg() const { return _err_msg ? _err_msg->_msg : ""; }
+ std::string_view msg() const { return _err_msg ? _err_msg->_msg : std::string_view(""); }
private:
int _code;
diff --git a/be/src/http/action/compaction_action.cpp b/be/src/http/action/compaction_action.cpp
index 8e8113c66cc..468c935b1ef 100644
--- a/be/src/http/action/compaction_action.cpp
+++ b/be/src/http/action/compaction_action.cpp
@@ -49,6 +49,8 @@
namespace doris {
using namespace ErrorCode;
+namespace {}
+
const static std::string HEADER_JSON = "application/json";
CompactionAction::CompactionAction(CompactionActionType ctype, ExecEnv* exec_env,
@@ -155,8 +157,7 @@ Status CompactionAction::_handle_run_compaction(HttpRequest* req, std::string* j
}
LOG(INFO) << "Manual compaction task is successfully triggered";
*json_result =
- "{\"status\": \"Success\", \"msg\": \"compaction task is successfully triggered. Table "
- "id: " +
+ R"({"status": "Success", "msg": "compaction task is successfully triggered. Table id: )" +
std::to_string(table_id) + ". Tablet id: " + std::to_string(tablet_id) + "\"}";
return Status::OK();
}
@@ -238,17 +239,21 @@ Status CompactionAction::_execute_compaction_callback(TabletSharedPtr tablet,
tablet->set_cumulative_compaction_policy(cumulative_compaction_policy);
}
Status res = Status::OK();
+ auto do_compact = [](Compaction& compaction) {
+ RETURN_IF_ERROR(compaction.prepare_compact());
+ return compaction.execute_compact();
+ };
if (compaction_type == PARAM_COMPACTION_BASE) {
- BaseCompaction base_compaction(tablet);
- res = base_compaction.compact();
+ BaseCompaction base_compaction(_engine, tablet);
+ res = do_compact(base_compaction);
if (!res) {
if (!res.is<BE_NO_SUITABLE_VERSION>()) {
DorisMetrics::instance()->base_compaction_request_failed->increment(1);
}
}
} else if (compaction_type == PARAM_COMPACTION_CUMULATIVE) {
- CumulativeCompaction cumulative_compaction(tablet);
- res = cumulative_compaction.compact();
+ CumulativeCompaction cumulative_compaction(_engine, tablet);
+ res = do_compact(cumulative_compaction);
if (!res) {
if (res.is<CUMULATIVE_NO_SUITABLE_VERSION>()) {
// Ignore this error code.
@@ -261,8 +266,8 @@ Status CompactionAction::_execute_compaction_callback(TabletSharedPtr tablet,
}
}
} else if (compaction_type == PARAM_COMPACTION_FULL) {
- FullCompaction full_compaction(tablet);
- res = full_compaction.compact();
+ FullCompaction full_compaction(_engine, tablet);
+ res = do_compact(full_compaction);
if (!res) {
if (res.is<FULL_NO_SUITABLE_VERSION>()) {
// Ignore this error code.
diff --git a/be/src/olap/base_compaction.cpp b/be/src/olap/base_compaction.cpp
index 474909cbf45..436180c78ca 100644
--- a/be/src/olap/base_compaction.cpp
+++ b/be/src/olap/base_compaction.cpp
@@ -18,8 +18,6 @@
#include "olap/base_compaction.h"
#include <gen_cpp/olap_file.pb.h>
-#include <stdint.h>
-#include <time.h>
#include <memory>
#include <mutex>
@@ -27,8 +25,10 @@
#include "common/config.h"
#include "common/logging.h"
+#include "olap/compaction.h"
#include "olap/olap_define.h"
#include "olap/rowset/rowset_meta.h"
+#include "olap/tablet.h"
#include "runtime/thread_context.h"
#include "util/doris_metrics.h"
#include "util/thread.h"
@@ -37,17 +37,18 @@
namespace doris {
using namespace ErrorCode;
-BaseCompaction::BaseCompaction(const TabletSharedPtr& tablet)
- : Compaction(tablet, "BaseCompaction:" + std::to_string(tablet->tablet_id())) {}
+BaseCompaction::BaseCompaction(StorageEngine& engine, const TabletSharedPtr& tablet)
+ : CompactionMixin(engine, tablet, "BaseCompaction:" + std::to_string(tablet->tablet_id())) {
+}
BaseCompaction::~BaseCompaction() = default;
Status BaseCompaction::prepare_compact() {
- if (!_tablet->init_succeeded()) {
+ if (!tablet()->init_succeeded()) {
return Status::Error<INVALID_ARGUMENT, false>("_tablet init failed");
}
- std::unique_lock<std::mutex> lock(_tablet->get_base_compaction_lock(), std::try_to_lock);
+ std::unique_lock<std::mutex> lock(tablet()->get_base_compaction_lock(), std::try_to_lock);
if (!lock.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>(
"another base compaction is running. tablet={}", _tablet->tablet_id());
@@ -60,13 +61,13 @@ Status BaseCompaction::prepare_compact() {
return Status::OK();
}
-Status BaseCompaction::execute_compact_impl() {
+Status BaseCompaction::execute_compact() {
#ifndef __APPLE__
if (config::enable_base_compaction_idle_sched) {
Thread::set_idle_sched();
}
#endif
- std::unique_lock<std::mutex> lock(_tablet->get_base_compaction_lock(), std::try_to_lock);
+ std::unique_lock<std::mutex> lock(tablet()->get_base_compaction_lock(), std::try_to_lock);
if (!lock.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>(
"another base compaction is running. tablet={}", _tablet->tablet_id());
@@ -74,14 +75,10 @@ Status BaseCompaction::execute_compact_impl() {
SCOPED_ATTACH_TASK(_mem_tracker);
- // 2. do base compaction, merge rowsets
- int64_t permits = get_compaction_permits();
- RETURN_IF_ERROR(do_compaction(permits));
-
- // 3. set state to success
- _state = CompactionState::SUCCESS;
+ RETURN_IF_ERROR(CompactionMixin::execute_compact());
+ DCHECK_EQ(_state, CompactionState::SUCCESS);
- // 4. add metric to base compaction
+ tablet()->set_last_base_compaction_success_time(UnixMillis());
DorisMetrics::instance()->base_compaction_deltas_total->increment(_input_rowsets.size());
DorisMetrics::instance()->base_compaction_bytes_total->increment(_input_rowsets_size);
@@ -112,7 +109,7 @@ void BaseCompaction::_filter_input_rowset() {
}
Status BaseCompaction::pick_rowsets_to_compact() {
- _input_rowsets = _tablet->pick_candidate_rowsets_to_base_compaction();
+ _input_rowsets = tablet()->pick_candidate_rowsets_to_base_compaction();
RETURN_IF_ERROR(check_version_continuity(_input_rowsets));
_filter_input_rowset();
if (_input_rowsets.size() <= 1) {
@@ -133,7 +130,7 @@ Status BaseCompaction::pick_rowsets_to_compact() {
// rowset's start version is bigger than 2, we'll always remain the delete pred information inside
// the output rowset so the rowsets whose version is less than _input_rowsets.front()->start_version() > 0
// would apply the delete pred in the end.
- if (!allow_delete_in_cumu_compaction() && _input_rowsets.front()->start_version() > 0) {
+ if (!_allow_delete_in_cumu_compaction && _input_rowsets.front()->start_version() > 0) {
bool has_delete_predicate = false;
for (const auto& rs : _input_rowsets) {
if (rs->rowset_meta()->has_delete_predicate()) {
diff --git a/be/src/olap/base_compaction.h b/be/src/olap/base_compaction.h
index e86cb3330a5..91e17e8ea5a 100644
--- a/be/src/olap/base_compaction.h
+++ b/be/src/olap/base_compaction.h
@@ -17,16 +17,12 @@
#pragma once
-#include <butil/macros.h>
-
#include <string>
#include <vector>
#include "common/status.h"
#include "io/io_common.h"
#include "olap/compaction.h"
-#include "olap/rowset/rowset.h"
-#include "olap/tablet.h"
namespace doris {
@@ -35,26 +31,24 @@ namespace doris {
// 1. its policy to pick rowsets
// 2. do compaction to produce new rowset.
-class BaseCompaction : public Compaction {
+class BaseCompaction final : public CompactionMixin {
public:
- BaseCompaction(const TabletSharedPtr& tablet);
+ BaseCompaction(StorageEngine& engine, const TabletSharedPtr& tablet);
~BaseCompaction() override;
Status prepare_compact() override;
- Status execute_compact_impl() override;
-protected:
- Status pick_rowsets_to_compact() override;
- std::string compaction_name() const override { return "base compaction"; }
+ Status execute_compact() override;
+
+private:
+ Status pick_rowsets_to_compact();
+ std::string_view compaction_name() const override { return "base compaction"; }
ReaderType compaction_type() const override { return ReaderType::READER_BASE_COMPACTION; }
-private:
// filter input rowset in some case:
// 1. dup key without delete predicate
void _filter_input_rowset();
-
- DISALLOW_COPY_AND_ASSIGN(BaseCompaction);
};
} // namespace doris
diff --git a/be/src/olap/base_tablet.cpp b/be/src/olap/base_tablet.cpp
index 83c6bb36fbf..c0904bdeadc 100644
--- a/be/src/olap/base_tablet.cpp
+++ b/be/src/olap/base_tablet.cpp
@@ -321,7 +321,7 @@ Versions BaseTablet::get_missed_versions(int64_t spec_version) const {
existing_versions.emplace_back(rs->version());
}
}
- return calc_missed_versions(spec_version, existing_versions);
+ return calc_missed_versions(spec_version, std::move(existing_versions));
}
Versions BaseTablet::get_missed_versions_unlocked(int64_t spec_version) const {
@@ -331,7 +331,7 @@ Versions BaseTablet::get_missed_versions_unlocked(int64_t spec_version) const {
for (const auto& rs : _tablet_meta->all_rs_metas()) {
existing_versions.emplace_back(rs->version());
}
- return calc_missed_versions(spec_version, existing_versions);
+ return calc_missed_versions(spec_version, std::move(existing_versions));
}
Versions BaseTablet::calc_missed_versions(int64_t spec_version, Versions existing_versions) {
diff --git a/be/src/olap/cold_data_compaction.cpp b/be/src/olap/cold_data_compaction.cpp
index b4bd95ea80f..bc06b00addf 100644
--- a/be/src/olap/cold_data_compaction.cpp
+++ b/be/src/olap/cold_data_compaction.cpp
@@ -31,6 +31,9 @@
#include "olap/compaction.h"
#include "olap/olap_common.h"
#include "olap/rowset/rowset.h"
+#include "olap/rowset/rowset_writer_context.h"
+#include "olap/storage_policy.h"
+#include "olap/tablet.h"
#include "olap/tablet_meta.h"
#include "runtime/thread_context.h"
#include "util/thread.h"
@@ -40,37 +43,57 @@
namespace doris {
using namespace ErrorCode;
-ColdDataCompaction::ColdDataCompaction(const TabletSharedPtr& tablet)
- : Compaction(tablet, "ColdDataCompaction:" + std::to_string(tablet->tablet_id())) {}
+ColdDataCompaction::ColdDataCompaction(StorageEngine& engine, const TabletSharedPtr& tablet)
+ : CompactionMixin(engine, tablet,
+ "ColdDataCompaction:" + std::to_string(tablet->tablet_id())) {}
ColdDataCompaction::~ColdDataCompaction() = default;
Status ColdDataCompaction::prepare_compact() {
- if (UNLIKELY(!_tablet->init_succeeded())) {
+ if (UNLIKELY(!tablet()->init_succeeded())) {
return Status::Error<INVALID_ARGUMENT>("_tablet init failed");
}
return pick_rowsets_to_compact();
}
-Status ColdDataCompaction::execute_compact_impl() {
+Status ColdDataCompaction::execute_compact() {
#ifndef __APPLE__
if (config::enable_base_compaction_idle_sched) {
Thread::set_idle_sched();
}
#endif
SCOPED_ATTACH_TASK(_mem_tracker);
- int64_t permits = get_compaction_permits();
- std::shared_lock cooldown_conf_rlock(_tablet->get_cooldown_conf_lock());
- if (_tablet->cooldown_conf_unlocked().cooldown_replica_id != _tablet->replica_id()) {
+ std::shared_lock cooldown_conf_rlock(tablet()->get_cooldown_conf_lock());
+ if (tablet()->cooldown_conf_unlocked().cooldown_replica_id != tablet()->replica_id()) {
return Status::Aborted<false>("this replica is not cooldown replica");
}
- RETURN_IF_ERROR(do_compaction(permits));
- _state = CompactionState::SUCCESS;
+ RETURN_IF_ERROR(CompactionMixin::execute_compact());
+ DCHECK_EQ(_state, CompactionState::SUCCESS);
return Status::OK();
}
+Status ColdDataCompaction::construct_output_rowset_writer(RowsetWriterContext& ctx) {
+ // write output rowset to storage policy resource
+ auto storage_policy = get_storage_policy(tablet()->storage_policy_id());
+ if (storage_policy == nullptr) {
+ return Status::InternalError("could not find storage_policy, storage_policy_id={}",
+ tablet()->storage_policy_id());
+ }
+
+ auto resource = get_storage_resource(storage_policy->resource_id);
+ if (resource.fs == nullptr) {
+ return Status::InternalError("could not find resource, resouce_id={}",
+ storage_policy->resource_id);
+ }
+
+ DCHECK(atol(resource.fs->id().c_str()) == storage_policy->resource_id);
+ DCHECK(resource.fs->type() != io::FileSystemType::LOCAL);
+ ctx.fs = std::move(resource.fs);
+ return CompactionMixin::construct_output_rowset_writer(ctx);
+}
+
Status ColdDataCompaction::pick_rowsets_to_compact() {
- _tablet->traverse_rowsets([this](const auto& rs) {
+ tablet()->traverse_rowsets([this](const auto& rs) {
if (!rs->is_local()) {
_input_rowsets.push_back(rs);
}
@@ -79,23 +102,23 @@ Status ColdDataCompaction::pick_rowsets_to_compact() {
return check_version_continuity(_input_rowsets);
}
-Status ColdDataCompaction::modify_rowsets(const Merger::Statistics* stats) {
+Status ColdDataCompaction::modify_rowsets() {
UniqueId cooldown_meta_id = UniqueId::gen_uid();
{
std::lock_guard wlock(_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
// Merged cooldowned rowsets MUST NOT be managed by version graph, they will be reclaimed by `remove_unused_remote_files`.
- _tablet->delete_rowsets(_input_rowsets, false);
- _tablet->add_rowsets({_output_rowset});
+ tablet()->delete_rowsets(_input_rowsets, false);
+ tablet()->add_rowsets({_output_rowset});
// TODO(plat1ko): process primary key
_tablet->tablet_meta()->set_cooldown_meta_id(cooldown_meta_id);
}
{
std::shared_lock rlock(_tablet->get_header_lock());
- _tablet->save_meta();
+ tablet()->save_meta();
}
// write remote tablet meta
- Tablet::async_write_cooldown_meta(_tablet);
+ Tablet::async_write_cooldown_meta(std::static_pointer_cast<Tablet>(_tablet));
return Status::OK();
}
diff --git a/be/src/olap/cold_data_compaction.h b/be/src/olap/cold_data_compaction.h
index 0b2eddf234c..33647ea8c63 100644
--- a/be/src/olap/cold_data_compaction.h
+++ b/be/src/olap/cold_data_compaction.h
@@ -22,25 +22,26 @@
#include "common/status.h"
#include "io/io_common.h"
#include "olap/compaction.h"
-#include "olap/merger.h"
-#include "olap/tablet.h"
namespace doris {
-class ColdDataCompaction final : public Compaction {
+class ColdDataCompaction final : public CompactionMixin {
public:
- ColdDataCompaction(const TabletSharedPtr& tablet);
+ ColdDataCompaction(StorageEngine& engine, const TabletSharedPtr& tablet);
~ColdDataCompaction() override;
Status prepare_compact() override;
- Status execute_compact_impl() override;
+ Status execute_compact() override;
private:
- std::string compaction_name() const override { return "cold data compaction"; }
+ std::string_view compaction_name() const override { return "cold data compaction"; }
ReaderType compaction_type() const override { return ReaderType::READER_COLD_DATA_COMPACTION; }
- Status pick_rowsets_to_compact() override;
- Status modify_rowsets(const Merger::Statistics* stats = nullptr) override;
+ Status construct_output_rowset_writer(RowsetWriterContext& ctx) override;
+
+ Status pick_rowsets_to_compact();
+
+ Status modify_rowsets() override;
};
} // namespace doris
diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
index c2eba35d4c1..2290858be71 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -43,6 +43,7 @@
#include "olap/olap_define.h"
#include "olap/rowset/beta_rowset.h"
#include "olap/rowset/rowset.h"
+#include "olap/rowset/rowset_fwd.h"
#include "olap/rowset/rowset_meta.h"
#include "olap/rowset/rowset_writer.h"
#include "olap/rowset/rowset_writer_context.h"
@@ -65,19 +66,52 @@ using std::vector;
namespace doris {
using namespace ErrorCode;
-Compaction::Compaction(const TabletSharedPtr& tablet, const std::string& label)
- : _tablet(tablet),
- _input_rowsets_size(0),
- _input_row_num(0),
- _input_num_segments(0),
- _input_index_size(0),
- _state(CompactionState::INITED),
+namespace {
+
+bool is_rowset_tidy(std::string& pre_max_key, const RowsetSharedPtr& rhs) {
+ size_t min_tidy_size = config::ordered_data_compaction_min_segment_size;
+ if (rhs->num_segments() == 0) {
+ return true;
+ }
+ if (rhs->is_segments_overlapping()) {
+ return false;
+ }
+ // check segment size
+ auto* beta_rowset = reinterpret_cast<BetaRowset*>(rhs.get());
+ std::vector<size_t> segments_size;
+ static_cast<void>(beta_rowset->get_segments_size(&segments_size));
+ for (auto segment_size : segments_size) {
+ // is segment is too small, need to do compaction
+ if (segment_size < min_tidy_size) {
+ return false;
+ }
+ }
+ std::string min_key;
+ auto ret = rhs->min_key(&min_key);
+ if (!ret) {
+ return false;
+ }
+ if (min_key < pre_max_key) {
+ return false;
+ }
+ CHECK(rhs->max_key(&pre_max_key));
+
+ return true;
+}
+
+} // namespace
+
+Compaction::Compaction(BaseTabletSPtr tablet, const std::string& label)
+ : _mem_tracker(
+ std::make_shared<MemTrackerLimiter>(MemTrackerLimiter::Type::COMPACTION, label)),
+ _tablet(std::move(tablet)),
+ _is_vertical(config::enable_vertical_compaction),
_allow_delete_in_cumu_compaction(config::enable_delete_when_cumu_compaction) {
- _mem_tracker = std::make_shared<MemTrackerLimiter>(MemTrackerLimiter::Type::COMPACTION, label);
+ ;
init_profile(label);
}
-Compaction::~Compaction() {}
+Compaction::~Compaction() = default;
void Compaction::init_profile(const std::string& label) {
_profile = std::make_unique<RuntimeProfile>(label);
@@ -96,59 +130,80 @@ void Compaction::init_profile(const std::string& label) {
_merge_rowsets_latency_timer = ADD_TIMER(_profile, "merge_rowsets_latency");
}
-Status Compaction::compact() {
- RETURN_IF_ERROR(prepare_compact());
- RETURN_IF_ERROR(execute_compact());
- return Status::OK();
-}
-
-Status Compaction::execute_compact() {
- Status st = execute_compact_impl();
- if (!st.ok()) {
- gc_output_rowset();
+Status Compaction::merge_input_rowsets() {
+ std::vector<RowsetReaderSharedPtr> input_rs_readers;
+ input_rs_readers.reserve(_input_rowsets.size());
+ for (auto& rowset : _input_rowsets) {
+ RowsetReaderSharedPtr rs_reader;
+ RETURN_IF_ERROR(rowset->create_reader(&rs_reader));
+ input_rs_readers.push_back(std::move(rs_reader));
}
- return st;
-}
-Status Compaction::do_compaction(int64_t permits) {
- uint32_t checksum_before;
- uint32_t checksum_after;
- if (config::enable_compaction_checksum) {
- EngineChecksumTask checksum_task(ExecEnv::GetInstance()->storage_engine().to_local(),
- _tablet->tablet_id(), _tablet->schema_hash(),
- _input_rowsets.back()->end_version(), &checksum_before);
- RETURN_IF_ERROR(checksum_task.execute());
- }
+ RowsetWriterContext ctx;
+ RETURN_IF_ERROR(construct_output_rowset_writer(ctx));
- _tablet->data_dir()->disks_compaction_score_increment(permits);
- _tablet->data_dir()->disks_compaction_num_increment(1);
- Status st = do_compaction_impl(permits);
- _tablet->data_dir()->disks_compaction_score_increment(-permits);
- _tablet->data_dir()->disks_compaction_num_increment(-1);
+ // write merged rows to output rowset
+ // The test results show that merger is low-memory-footprint, there is no need to tracker its mem pool
+ // if ctx.skip_inverted_index.size() > 0, it means we need to do inverted index compaction.
+ // the row ID conversion matrix needs to be used for inverted index compaction.
+ if (!ctx.skip_inverted_index.empty() || (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
+ _tablet->enable_unique_key_merge_on_write())) {
+ _stats.rowid_conversion = &_rowid_conversion;
+ }
- if (config::enable_compaction_checksum) {
- EngineChecksumTask checksum_task(ExecEnv::GetInstance()->storage_engine().to_local(),
- _tablet->tablet_id(), _tablet->schema_hash(),
- _input_rowsets.back()->end_version(), &checksum_after);
- RETURN_IF_ERROR(checksum_task.execute());
- if (checksum_before != checksum_after) {
- LOG(WARNING) << "Compaction tablet=" << _tablet->tablet_id()
- << " checksum not consistent"
- << ", before=" << checksum_before << ", checksum_after=" << checksum_after;
+ Status res;
+ {
+ SCOPED_TIMER(_merge_rowsets_latency_timer);
+ if (_is_vertical) {
+ res = Merger::vertical_merge_rowsets(_tablet, compaction_type(), *_cur_tablet_schema,
+ input_rs_readers, _output_rs_writer.get(),
+ get_avg_segment_rows(), &_stats);
+ } else {
+ res = Merger::vmerge_rowsets(_tablet, compaction_type(), *_cur_tablet_schema,
+ input_rs_readers, _output_rs_writer.get(), &_stats);
}
}
- if (st.ok()) {
- _load_segment_to_cache();
+
+ if (!res.ok()) {
+ LOG(WARNING) << "fail to do " << compaction_name() << ". res=" << res
+ << ", tablet=" << _tablet->tablet_id()
+ << ", output_version=" << _output_version;
+ return res;
}
- return st;
-}
-bool Compaction::should_vertical_compaction() {
- // some conditions that not use vertical compaction
- if (!config::enable_vertical_compaction) {
- return false;
+ COUNTER_UPDATE(_merged_rows_counter, _stats.merged_rows);
+ COUNTER_UPDATE(_filtered_rows_counter, _stats.filtered_rows);
+
+ RETURN_NOT_OK_STATUS_WITH_WARN(_output_rs_writer->build(_output_rowset),
+ fmt::format("rowset writer build failed. output_version: {}",
+ _output_version.to_string()));
+
+ // Now we support delete in cumu compaction, to make all data in rowsets whose version
+ // is below output_version to be delete in the future base compaction, we should carry
+ // all delete predicate in the output rowset.
+ // Output start version > 2 means we must set the delete predicate in the output rowset
+ if (_allow_delete_in_cumu_compaction && _output_rowset->version().first > 2) {
+ DeletePredicatePB delete_predicate;
+ std::accumulate(_input_rowsets.begin(), _input_rowsets.end(), &delete_predicate,
+ [](DeletePredicatePB* delete_predicate, const RowsetSharedPtr& rs) {
+ if (rs->rowset_meta()->has_delete_predicate()) {
+ delete_predicate->MergeFrom(rs->rowset_meta()->delete_predicate());
+ }
+ return delete_predicate;
+ });
+ // now version in delete_predicate is deprecated
+ if (!delete_predicate.in_predicates().empty() ||
+ !delete_predicate.sub_predicates_v2().empty() ||
+ !delete_predicate.sub_predicates().empty()) {
+ _output_rowset->rowset_meta()->set_delete_predicate(std::move(delete_predicate));
+ }
}
- return true;
+
+ COUNTER_UPDATE(_output_rowset_data_size_counter, _output_rowset->data_disk_size());
+ COUNTER_UPDATE(_output_row_num_counter, _output_rowset->num_rows());
+ COUNTER_UPDATE(_output_segments_num_counter, _output_rowset->num_segments());
+
+ return check_correctness();
}
int64_t Compaction::get_avg_segment_rows() {
@@ -166,38 +221,27 @@ int64_t Compaction::get_avg_segment_rows() {
(_input_rowsets_size / (_input_row_num + 1) + 1);
}
-bool Compaction::is_rowset_tidy(std::string& pre_max_key, const RowsetSharedPtr& rhs) {
- size_t min_tidy_size = config::ordered_data_compaction_min_segment_size;
- if (rhs->num_segments() == 0) {
- return true;
- }
- if (rhs->is_segments_overlapping()) {
- return false;
- }
- // check segment size
- auto beta_rowset = reinterpret_cast<BetaRowset*>(rhs.get());
- std::vector<size_t> segments_size;
- static_cast<void>(beta_rowset->get_segments_size(&segments_size));
- for (auto segment_size : segments_size) {
- // is segment is too small, need to do compaction
- if (segment_size < min_tidy_size) {
- return false;
+CompactionMixin::CompactionMixin(StorageEngine& engine, TabletSharedPtr tablet,
+ const std::string& label)
+ : Compaction(tablet, label), _engine(engine) {}
+
+CompactionMixin::~CompactionMixin() {
+ if (_state != CompactionState::SUCCESS && _output_rowset != nullptr) {
+ if (!_output_rowset->is_local()) {
+ tablet()->record_unused_remote_rowset(_output_rowset->rowset_id(),
+ _output_rowset->rowset_meta()->resource_id(),
+ _output_rowset->num_segments());
+ return;
}
+ _engine.add_unused_rowset(_output_rowset);
}
- std::string min_key;
- auto ret = rhs->min_key(&min_key);
- if (!ret) {
- return false;
- }
- if (min_key < pre_max_key) {
- return false;
- }
- CHECK(rhs->max_key(&pre_max_key));
+}
- return true;
+Tablet* CompactionMixin::tablet() {
+ return static_cast<Tablet*>(_tablet.get());
}
-Status Compaction::do_compact_ordered_rowsets() {
+Status CompactionMixin::do_compact_ordered_rowsets() {
build_basic_info();
RowsetWriterContext ctx;
RETURN_IF_ERROR(construct_output_rowset_writer(ctx));
@@ -232,7 +276,7 @@ Status Compaction::do_compact_ordered_rowsets() {
return Status::OK();
}
-void Compaction::build_basic_info() {
+void CompactionMixin::build_basic_info() {
for (auto& rowset : _input_rowsets) {
_input_rowsets_size += rowset->data_disk_size();
_input_index_size += rowset->index_disk_size();
@@ -254,7 +298,7 @@ void Compaction::build_basic_info() {
_cur_tablet_schema = _tablet->tablet_schema_with_merged_max_schema_version(rowset_metas);
}
-bool Compaction::handle_ordered_data_compaction() {
+bool CompactionMixin::handle_ordered_data_compaction() {
if (!config::enable_ordered_data_compaction) {
return false;
}
@@ -293,37 +337,62 @@ bool Compaction::handle_ordered_data_compaction() {
// most rowset of current compaction is nonoverlapping
// just handle nonoverlappint rowsets
auto st = do_compact_ordered_rowsets();
+ return st.ok();
+}
+
+Status CompactionMixin::execute_compact() {
+ uint32_t checksum_before;
+ uint32_t checksum_after;
+ bool enable_compaction_checksum = config::enable_compaction_checksum;
+ if (enable_compaction_checksum) {
+ EngineChecksumTask checksum_task(_engine, _tablet->tablet_id(), _tablet->schema_hash(),
+ _input_rowsets.back()->end_version(), &checksum_before);
+ RETURN_IF_ERROR(checksum_task.execute());
+ }
+
+ auto* data_dir = tablet()->data_dir();
+ int64_t permits = get_compaction_permits();
+ data_dir->disks_compaction_score_increment(permits);
+ data_dir->disks_compaction_num_increment(1);
+
+ Status st = execute_compact_impl(permits);
+
+ data_dir->disks_compaction_score_increment(-permits);
+ data_dir->disks_compaction_num_increment(-1);
+
if (!st.ok()) {
- return false;
+ return st;
}
- return true;
+
+ if (enable_compaction_checksum) {
+ EngineChecksumTask checksum_task(_engine, _tablet->tablet_id(), _tablet->schema_hash(),
+ _input_rowsets.back()->end_version(), &checksum_after);
+ RETURN_IF_ERROR(checksum_task.execute());
+ if (checksum_before != checksum_after) {
+ return Status::InternalError(
+ "compaction tablet checksum not consistent, before={}, after={}, tablet_id={}",
+ checksum_before, checksum_after, _tablet->tablet_id());
+ }
+ }
+
+ _load_segment_to_cache();
+ return Status::OK();
}
-Status Compaction::do_compaction_impl(int64_t permits) {
+Status CompactionMixin::execute_compact_impl(int64_t permits) {
OlapStopWatch watch;
if (handle_ordered_data_compaction()) {
RETURN_IF_ERROR(modify_rowsets());
-
- int64_t now = UnixMillis();
- if (compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION) {
- _tablet->set_last_cumu_compaction_success_time(now);
- } else if (compaction_type() == ReaderType::READER_BASE_COMPACTION) {
- _tablet->set_last_base_compaction_success_time(now);
- } else if (compaction_type() == ReaderType::READER_FULL_COMPACTION) {
- _tablet->set_last_full_compaction_success_time(now);
- }
- auto cumu_policy = _tablet->cumulative_compaction_policy();
LOG(INFO) << "succeed to do ordered data " << compaction_name()
<< ". tablet=" << _tablet->tablet_id() << ", output_version=" << _output_version
- << ", disk=" << _tablet->data_dir()->path()
+ << ", disk=" << tablet()->data_dir()->path()
<< ", segments=" << _input_num_segments << ", input_row_num=" << _input_row_num
<< ", output_row_num=" << _output_rowset->num_rows()
<< ", input_rowset_size=" << _input_rowsets_size
<< ", output_rowset_size=" << _output_rowset->data_disk_size()
- << ". elapsed time=" << watch.get_elapse_second()
- << "s. cumulative_compaction_policy="
- << (cumu_policy == nullptr ? "quick" : cumu_policy->name());
+ << ". elapsed time=" << watch.get_elapse_second() << "s.";
+ _state = CompactionState::SUCCESS;
return Status::OK();
}
build_basic_info();
@@ -332,412 +401,276 @@ Status Compaction::do_compaction_impl(int64_t permits) {
LOG(INFO) << "start " << compaction_name() << ". tablet=" << _tablet->tablet_id()
<< ", output_version=" << _output_version << ", permits: " << permits;
- bool vertical_compaction = should_vertical_compaction();
- RowsetWriterContext ctx;
- RETURN_IF_ERROR(construct_input_rowset_readers());
- RETURN_IF_ERROR(construct_output_rowset_writer(ctx, vertical_compaction));
- // 2. write merged rows to output rowset
- // The test results show that merger is low-memory-footprint, there is no need to tracker its mem pool
- Merger::Statistics stats;
- // if ctx.skip_inverted_index.size() > 0, it means we need to do inverted index compaction.
- // the row ID conversion matrix needs to be used for inverted index compaction.
- if (ctx.skip_inverted_index.size() > 0 || (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
- _tablet->enable_unique_key_merge_on_write())) {
- stats.rowid_conversion = &_rowid_conversion;
- }
+ RETURN_IF_ERROR(merge_input_rowsets());
- Status res;
- {
- SCOPED_TIMER(_merge_rowsets_latency_timer);
- if (vertical_compaction) {
- res = Merger::vertical_merge_rowsets(_tablet, compaction_type(), _cur_tablet_schema,
- _input_rs_readers, _output_rs_writer.get(),
- get_avg_segment_rows(), &stats);
- } else {
- res = Merger::vmerge_rowsets(_tablet, compaction_type(), _cur_tablet_schema,
- _input_rs_readers, _output_rs_writer.get(), &stats);
- }
- }
+ RETURN_IF_ERROR(do_inverted_index_compaction());
- if (!res.ok()) {
- LOG(WARNING) << "fail to do " << compaction_name() << ". res=" << res
- << ", tablet=" << _tablet->tablet_id()
- << ", output_version=" << _output_version;
- return res;
+ RETURN_IF_ERROR(modify_rowsets());
+
+ auto* cumu_policy = tablet()->cumulative_compaction_policy();
+ DCHECK(cumu_policy);
+ LOG(INFO) << "succeed to do " << compaction_name() << " is_vertical=" << _is_vertical
+ << ". tablet=" << _tablet->tablet_id() << ", output_version=" << _output_version
+ << ", current_max_version=" << tablet()->max_version().second
+ << ", disk=" << tablet()->data_dir()->path() << ", segments=" << _input_num_segments
+ << ", input_rowset_size=" << _input_rowsets_size
+ << ", output_rowset_size=" << _output_rowset->data_disk_size()
+ << ", input_row_num=" << _input_row_num
+ << ", output_row_num=" << _output_rowset->num_rows()
+ << ", filtered_row_num=" << _stats.filtered_rows
+ << ", merged_row_num=" << _stats.merged_rows
+ << ". elapsed time=" << watch.get_elapse_second()
+ << "s. cumulative_compaction_policy=" << cumu_policy->name()
+ << ", compact_row_per_second=" << int(_input_row_num / watch.get_elapse_second());
+
+ _state = CompactionState::SUCCESS;
+
+ return Status::OK();
+}
+
+Status CompactionMixin::do_inverted_index_compaction() {
+ const auto& ctx = _output_rs_writer->context();
+ if (!config::inverted_index_compaction_enable || _input_row_num <= 0 ||
+ !_stats.rowid_conversion || ctx.skip_inverted_index.empty()) {
+ return Status::OK();
}
- COUNTER_UPDATE(_merged_rows_counter, stats.merged_rows);
- COUNTER_UPDATE(_filtered_rows_counter, stats.filtered_rows);
- RETURN_NOT_OK_STATUS_WITH_WARN(_output_rs_writer->build(_output_rowset),
- fmt::format("rowset writer build failed. output_version: {}",
- _output_version.to_string()));
- // Now we support delete in cumu compaction, to make all data in rowsets whose version
- // is below output_version to be delete in the future base compaction, we should carry
- // all delete predicate in the output rowset.
- // Output start version > 2 means we must set the delete predicate in the output rowset
- if (allow_delete_in_cumu_compaction() && _output_rowset->version().first > 2) {
- DeletePredicatePB delete_predicate;
- std::accumulate(
- _input_rs_readers.begin(), _input_rs_readers.end(), &delete_predicate,
- [](DeletePredicatePB* delete_predicate, const RowsetReaderSharedPtr& reader) {
- if (reader->rowset()->rowset_meta()->has_delete_predicate()) {
- delete_predicate->MergeFrom(
- reader->rowset()->rowset_meta()->delete_predicate());
- }
- return delete_predicate;
- });
- // now version in delete_predicate is deprecated
- if (!delete_predicate.in_predicates().empty() ||
- !delete_predicate.sub_predicates_v2().empty() ||
- !delete_predicate.sub_predicates().empty()) {
- _output_rowset->rowset_meta()->set_delete_predicate(std::move(delete_predicate));
+ OlapStopWatch inverted_watch;
+
+ Version version = tablet()->max_version();
+ DeleteBitmap output_rowset_delete_bitmap(_tablet->tablet_id());
+ std::set<RowLocation> missed_rows;
+ std::map<RowsetSharedPtr, std::list<std::pair<RowLocation, RowLocation>>> location_map;
+ // Convert the delete bitmap of the input rowsets to output rowset.
+ tablet()->calc_compaction_output_rowset_delete_bitmap(
+ _input_rowsets, _rowid_conversion, 0, version.second + 1, &missed_rows, &location_map,
+ _tablet->tablet_meta()->delete_bitmap(), &output_rowset_delete_bitmap);
+
+ if (!_allow_delete_in_cumu_compaction) {
+ if (compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION &&
+ _stats.merged_rows != missed_rows.size()) {
+ std::string err_msg = fmt::format(
+ "cumulative compaction: the merged rows({}) is not equal to missed "
+ "rows({}) in rowid conversion, tablet_id: {}, table_id:{}",
+ _stats.merged_rows, missed_rows.size(), _tablet->tablet_id(),
+ _tablet->table_id());
+ DCHECK(false) << err_msg;
+ LOG(WARNING) << err_msg;
}
}
- COUNTER_UPDATE(_output_rowset_data_size_counter, _output_rowset->data_disk_size());
- COUNTER_UPDATE(_output_row_num_counter, _output_rowset->num_rows());
- COUNTER_UPDATE(_output_segments_num_counter, _output_rowset->num_segments());
+ RETURN_IF_ERROR(tablet()->check_rowid_conversion(_output_rowset, location_map));
- // 3. check correctness
- RETURN_IF_ERROR(check_correctness(stats));
+ // translation vec
+ // <<dest_idx_num, dest_docId>>
+ // the first level vector: index indicates src segment.
+ // the second level vector: index indicates row id of source segment,
+ // value indicates row id of destination segment.
+ // <UINT32_MAX, UINT32_MAX> indicates current row not exist.
+ const auto& trans_vec = _stats.rowid_conversion->get_rowid_conversion_map();
- if (_input_row_num > 0 && stats.rowid_conversion && config::inverted_index_compaction_enable &&
- !ctx.skip_inverted_index.empty()) {
- OlapStopWatch inverted_watch;
+ // source rowset,segment -> index_id
+ const auto& src_seg_to_id_map = _stats.rowid_conversion->get_src_segment_to_id_map();
- // check rowid_conversion correctness
- Version version = _tablet->max_version();
- DeleteBitmap output_rowset_delete_bitmap(_tablet->tablet_id());
- std::set<RowLocation> missed_rows;
- std::map<RowsetSharedPtr, std::list<std::pair<RowLocation, RowLocation>>> location_map;
- // Convert the delete bitmap of the input rowsets to output rowset.
- std::size_t missed_rows_size = 0;
- _tablet->calc_compaction_output_rowset_delete_bitmap(
- _input_rowsets, _rowid_conversion, 0, version.second + 1, &missed_rows,
- &location_map, _tablet->tablet_meta()->delete_bitmap(),
- &output_rowset_delete_bitmap);
- if (!allow_delete_in_cumu_compaction()) {
- missed_rows_size = missed_rows.size();
- if (compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION &&
- stats.merged_rows != missed_rows_size) {
- std::string err_msg = fmt::format(
- "cumulative compaction: the merged rows({}) is not equal to missed "
- "rows({}) in rowid conversion, tablet_id: {}, table_id:{}",
- stats.merged_rows, missed_rows_size, _tablet->tablet_id(),
- _tablet->table_id());
- DCHECK(false) << err_msg;
- LOG(WARNING) << err_msg;
- }
- }
+ // dest rowset id
+ RowsetId dest_rowset_id = _stats.rowid_conversion->get_dst_rowset_id();
+ // dest segment id -> num rows
+ std::vector<uint32_t> dest_segment_num_rows;
+ RETURN_IF_ERROR(_output_rs_writer->get_segment_num_rows(&dest_segment_num_rows));
- RETURN_IF_ERROR(_tablet->check_rowid_conversion(_output_rowset, location_map));
-
- // translation vec
- // <<dest_idx_num, dest_docId>>
- // the first level vector: index indicates src segment.
- // the second level vector: index indicates row id of source segment,
- // value indicates row id of destination segment.
- // <UINT32_MAX, UINT32_MAX> indicates current row not exist.
- std::vector<std::vector<std::pair<uint32_t, uint32_t>>> trans_vec =
- stats.rowid_conversion->get_rowid_conversion_map();
-
- // source rowset,segment -> index_id
- std::map<std::pair<RowsetId, uint32_t>, uint32_t> src_seg_to_id_map =
- stats.rowid_conversion->get_src_segment_to_id_map();
- // dest rowset id
- RowsetId dest_rowset_id = stats.rowid_conversion->get_dst_rowset_id();
- // dest segment id -> num rows
- std::vector<uint32_t> dest_segment_num_rows;
- RETURN_IF_ERROR(_output_rs_writer->get_segment_num_rows(&dest_segment_num_rows));
-
- auto src_segment_num = src_seg_to_id_map.size();
- auto dest_segment_num = dest_segment_num_rows.size();
-
- if (dest_segment_num > 0) {
- // src index files
- // format: rowsetId_segmentId
- std::vector<std::string> src_index_files(src_segment_num);
- for (const auto& m : src_seg_to_id_map) {
- std::pair<RowsetId, uint32_t> p = m.first;
- src_index_files[m.second] = p.first.to_string() + "_" + std::to_string(p.second);
- }
+ auto src_segment_num = src_seg_to_id_map.size();
+ auto dest_segment_num = dest_segment_num_rows.size();
- // dest index files
- // format: rowsetId_segmentId
- std::vector<std::string> dest_index_files(dest_segment_num);
- for (int i = 0; i < dest_segment_num; ++i) {
- auto prefix = dest_rowset_id.to_string() + "_" + std::to_string(i);
- dest_index_files[i] = prefix;
- }
+ if (dest_segment_num <= 0) {
+ LOG(INFO) << "skip doing index compaction due to no output segments"
+ << ". tablet=" << _tablet->tablet_id() << ", input row number=" << _input_row_num
+ << ", output row number=" << _output_rowset->num_rows()
+ << ". elapsed time=" << inverted_watch.get_elapse_second() << "s.";
+ return Status::OK();
+ }
- // create index_writer to compaction indexes
- const auto& fs = _output_rowset->rowset_meta()->fs();
- const auto& tablet_path = _tablet->tablet_path();
-
- // we choose the first destination segment name as the temporary index writer path
- // Used to distinguish between different index compaction
- auto index_writer_path = tablet_path + "/" + dest_index_files[0];
- LOG(INFO) << "start index compaction"
- << ". tablet=" << _tablet->tablet_id()
- << ", source index size=" << src_segment_num
- << ", destination index size=" << dest_segment_num << ".";
- Status status = Status::OK();
- std::for_each(
- ctx.skip_inverted_index.cbegin(), ctx.skip_inverted_index.cend(),
- [&src_segment_num, &dest_segment_num, &index_writer_path, &src_index_files,
- &dest_index_files, &fs, &tablet_path, &trans_vec, &dest_segment_num_rows,
- &status, this](int32_t column_uniq_id) {
- auto index_id = _cur_tablet_schema->get_inverted_index(column_uniq_id, "")
- ->index_id();
- try {
- auto st = compact_column(index_id, src_segment_num, dest_segment_num,
- src_index_files, dest_index_files, fs,
- index_writer_path, tablet_path, trans_vec,
- dest_segment_num_rows);
- if (!st.ok()) {
- LOG(WARNING) << "failed to do index compaction"
- << ". tablet=" << _tablet->tablet_id()
- << ". column uniq id=" << column_uniq_id
- << ". index_id=" << index_id;
- for (auto& rowset : _input_rowsets) {
- rowset->set_skip_index_compaction(column_uniq_id);
- LOG(INFO) << "mark skipping inverted index compaction next time"
- << ". tablet=" << _tablet->tablet_id()
- << ", rowset=" << rowset->rowset_id()
- << ", column uniq id=" << column_uniq_id
- << ", index_id=" << index_id;
- }
- status = Status::Error<ErrorCode::INVERTED_INDEX_COMPACTION_ERROR>(
- st.msg());
- }
- } catch (CLuceneError& e) {
- LOG(WARNING) << "failed to do index compaction"
- << ". tablet=" << _tablet->tablet_id()
- << ", column uniq id=" << column_uniq_id
- << ", index_id=" << index_id;
- for (auto& rowset : _input_rowsets) {
- rowset->set_skip_index_compaction(column_uniq_id);
- LOG(INFO) << "mark skipping inverted index compaction next time"
- << ". tablet=" << _tablet->tablet_id()
- << ", rowset=" << rowset->rowset_id()
- << ", column uniq id=" << column_uniq_id
- << ", index_id=" << index_id;
- }
- status = Status::Error<ErrorCode::INVERTED_INDEX_COMPACTION_ERROR>(
- e.what());
- }
- });
-
- // check index compaction status. If status is not ok, we should return error and end this compaction round.
- if (!status.ok()) {
- return status;
- }
+ // src index files
+ // format: rowsetId_segmentId
+ std::vector<std::string> src_index_files(src_segment_num);
+ for (const auto& m : src_seg_to_id_map) {
+ std::pair<RowsetId, uint32_t> p = m.first;
+ src_index_files[m.second] = p.first.to_string() + "_" + std::to_string(p.second);
+ }
- LOG(INFO) << "succeed to do index compaction"
- << ". tablet=" << _tablet->tablet_id()
- << ", input row number=" << _input_row_num
- << ", output row number=" << _output_rowset->num_rows()
- << ", input_rowset_size=" << _input_rowsets_size
- << ", output_rowset_size=" << _output_rowset->data_disk_size()
- << ". elapsed time=" << inverted_watch.get_elapse_second() << "s.";
- } else {
- LOG(INFO) << "skip doing index compaction due to no output segments"
- << ". tablet=" << _tablet->tablet_id()
- << ", input row number=" << _input_row_num
- << ", output row number=" << _output_rowset->num_rows()
- << ". elapsed time=" << inverted_watch.get_elapse_second() << "s.";
- }
+ // dest index files
+ // format: rowsetId_segmentId
+ std::vector<std::string> dest_index_files(dest_segment_num);
+ for (int i = 0; i < dest_segment_num; ++i) {
+ auto prefix = dest_rowset_id.to_string() + "_" + std::to_string(i);
+ dest_index_files[i] = prefix;
}
- // 4. modify rowsets in memory
- RETURN_IF_ERROR(modify_rowsets(&stats));
+ // create index_writer to compaction indexes
+ const auto& fs = _output_rowset->rowset_meta()->fs();
+ const auto& tablet_path = _tablet->tablet_path();
- // 5. update last success compaction time
- int64_t now = UnixMillis();
- // TODO(yingchun): do the judge in Tablet class
- if (compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION) {
- _tablet->set_last_cumu_compaction_success_time(now);
- } else if (compaction_type() == ReaderType::READER_BASE_COMPACTION) {
- _tablet->set_last_base_compaction_success_time(now);
- } else if (compaction_type() == ReaderType::READER_FULL_COMPACTION) {
- _tablet->set_last_full_compaction_success_time(now);
- }
+ // we choose the first destination segment name as the temporary index writer path
+ // Used to distinguish between different index compaction
+ auto index_writer_path = tablet_path + "/" + dest_index_files[0];
+ LOG(INFO) << "start index compaction"
+ << ". tablet=" << _tablet->tablet_id() << ", source index size=" << src_segment_num
+ << ", destination index size=" << dest_segment_num << ".";
- int64_t current_max_version = -1;
- {
- std::shared_lock rdlock(_tablet->get_header_lock());
- current_max_version = -1;
- if (RowsetSharedPtr max_rowset = _tablet->get_rowset_with_max_version()) {
- current_max_version = max_rowset->end_version();
+ auto error_handler = [this](int64_t index_id, int64_t column_uniq_id) {
+ LOG(WARNING) << "failed to do index compaction"
+ << ". tablet=" << _tablet->tablet_id() << ". column uniq id=" << column_uniq_id
+ << ". index_id=" << index_id;
+ for (auto& rowset : _input_rowsets) {
+ rowset->set_skip_index_compaction(column_uniq_id);
+ LOG(INFO) << "mark skipping inverted index compaction next time"
+ << ". tablet=" << _tablet->tablet_id() << ", rowset=" << rowset->rowset_id()
+ << ", column uniq id=" << column_uniq_id << ", index_id=" << index_id;
+ }
+ };
+
+ for (auto&& column_uniq_id : ctx.skip_inverted_index) {
+ auto index_id = _cur_tablet_schema->get_inverted_index(column_uniq_id, "")->index_id();
+ try {
+ auto st = compact_column(index_id, src_segment_num, dest_segment_num, src_index_files,
+ dest_index_files, fs, index_writer_path, tablet_path,
+ trans_vec, dest_segment_num_rows);
+ if (!st.ok()) {
+ error_handler(index_id, column_uniq_id);
+ return Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(st.msg());
+ }
+ } catch (CLuceneError& e) {
+ error_handler(index_id, column_uniq_id);
+ return Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(e.what());
}
}
- auto* cumu_policy = _tablet->cumulative_compaction_policy();
- DCHECK(cumu_policy);
- LOG(INFO) << "succeed to do " << compaction_name() << " is_vertical=" << vertical_compaction
- << ". tablet=" << _tablet->tablet_id() << ", output_version=" << _output_version
- << ", current_max_version=" << current_max_version
- << ", disk=" << _tablet->data_dir()->path() << ", segments=" << _input_num_segments
+ LOG(INFO) << "succeed to do index compaction"
+ << ". tablet=" << _tablet->tablet_id() << ", input row number=" << _input_row_num
+ << ", output row number=" << _output_rowset->num_rows()
<< ", input_rowset_size=" << _input_rowsets_size
<< ", output_rowset_size=" << _output_rowset->data_disk_size()
- << ", input_row_num=" << _input_row_num
- << ", output_row_num=" << _output_rowset->num_rows()
- << ", filtered_row_num=" << stats.filtered_rows
- << ", merged_row_num=" << stats.merged_rows
- << ". elapsed time=" << watch.get_elapse_second()
- << "s. cumulative_compaction_policy=" << cumu_policy->name()
- << ", compact_row_per_second=" << int(_input_row_num / watch.get_elapse_second());
+ << ". elapsed time=" << inverted_watch.get_elapse_second() << "s.";
return Status::OK();
}
-Status Compaction::construct_output_rowset_writer(RowsetWriterContext& ctx, bool is_vertical) {
- ctx.version = _output_version;
- ctx.rowset_state = VISIBLE;
- ctx.segments_overlap = NONOVERLAPPING;
- ctx.tablet_schema = _cur_tablet_schema;
- ctx.newest_write_timestamp = _newest_write_timestamp;
- ctx.write_type = DataWriteType::TYPE_COMPACTION;
- if (config::inverted_index_compaction_enable &&
- ((_tablet->keys_type() == KeysType::UNIQUE_KEYS ||
- _tablet->keys_type() == KeysType::DUP_KEYS))) {
- for (const auto& index : _cur_tablet_schema->indexes()) {
- if (index.index_type() == IndexType::INVERTED) {
- auto col_unique_id = index.col_unique_ids()[0];
- //NOTE: here src_rs may be in building index progress, so it would not contain inverted index info.
- bool all_have_inverted_index = std::all_of(
- _input_rowsets.begin(), _input_rowsets.end(), [&](const auto& src_rs) {
- BetaRowsetSharedPtr rowset =
- std::static_pointer_cast<BetaRowset>(src_rs);
- if (rowset == nullptr) {
- LOG(WARNING) << "tablet[" << _tablet->tablet_id()
- << "] rowset is null, will skip index compaction";
- return false;
- }
- if (rowset->is_skip_index_compaction(col_unique_id)) {
- LOG(WARNING)
- << "tablet[" << _tablet->tablet_id() << "] rowset["
- << rowset->rowset_id() << "] column_unique_id["
- << col_unique_id
- << "] skip inverted index compaction due to last failure";
- return false;
- }
- auto fs = rowset->rowset_meta()->fs();
-
- const auto* index_meta =
- rowset->tablet_schema()->get_inverted_index(col_unique_id, "");
- if (index_meta == nullptr) {
- LOG(WARNING) << "tablet[" << _tablet->tablet_id()
- << "] column_unique_id[" << col_unique_id
- << "] index meta is null, will skip index compaction";
- return false;
- }
- for (auto i = 0; i < rowset->num_segments(); i++) {
- auto segment_file = rowset->segment_file_path(i);
- std::string inverted_index_src_file_path =
- InvertedIndexDescriptor::get_index_file_name(
- segment_file, index_meta->index_id(),
- index_meta->get_index_suffix());
- bool exists = false;
- if (!fs->exists(inverted_index_src_file_path, &exists).ok()) {
- LOG(ERROR)
- << inverted_index_src_file_path << " fs->exists error";
- return false;
- }
- if (!exists) {
- LOG(WARNING) << "tablet[" << _tablet->tablet_id()
- << "] column_unique_id[" << col_unique_id << "],"
- << inverted_index_src_file_path
- << " is not exists, will skip index compaction";
- return false;
- }
-
- // check idx file size
- int64_t file_size = 0;
- if (fs->file_size(inverted_index_src_file_path, &file_size) !=
- Status::OK()) {
- LOG(ERROR) << inverted_index_src_file_path
- << " fs->file_size error";
- return false;
- }
- if (file_size == 0) {
- LOG(WARNING) << "tablet[" << _tablet->tablet_id()
- << "] column_unique_id[" << col_unique_id << "],"
- << inverted_index_src_file_path
- << " is empty file, will skip index compaction";
- return false;
- }
-
- // check index meta
- std::filesystem::path p(inverted_index_src_file_path);
- std::string dir_str = p.parent_path().string();
- std::string file_str = p.filename().string();
- lucene::store::Directory* dir =
- DorisCompoundDirectoryFactory::getDirectory(
- fs, dir_str.c_str());
- DorisCompoundReader reader(dir, file_str.c_str());
- std::vector<std::string> files;
- reader.list(&files);
- reader.close();
-
- // why is 3?
- // bkd index will write at least 3 files
- if (files.size() < 3) {
- LOG(WARNING) << "tablet[" << _tablet->tablet_id()
- << "] column_unique_id[" << col_unique_id << "],"
- << inverted_index_src_file_path
- << " is corrupted, will skip index compaction";
- return false;
- }
- }
- return true;
- });
- if (all_have_inverted_index &&
- field_is_slice_type(_cur_tablet_schema->column_by_uid(col_unique_id).type())) {
- ctx.skip_inverted_index.insert(col_unique_id);
+void CompactionMixin::construct_skip_inverted_index(RowsetWriterContext& ctx) {
+ for (const auto& index : _cur_tablet_schema->indexes()) {
+ if (index.index_type() != IndexType::INVERTED) {
+ continue;
+ }
+
+ auto col_unique_id = index.col_unique_ids()[0];
+ auto has_inverted_index = [&](const RowsetSharedPtr& src_rs) {
+ auto rowset = static_cast<BetaRowset*>(src_rs.get());
+ if (rowset->is_skip_index_compaction(col_unique_id)) {
+ LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] rowset["
+ << rowset->rowset_id() << "] column_unique_id[" << col_unique_id
+ << "] skip inverted index compaction due to last failure";
+ return false;
+ }
+
+ auto& fs = rowset->rowset_meta()->fs();
+
+ const auto* index_meta = rowset->tablet_schema()->get_inverted_index(col_unique_id, "");
+ if (index_meta == nullptr) {
+ LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] column_unique_id["
+ << col_unique_id << "] index meta is null, will skip index compaction";
+ return false;
+ }
+
+ for (auto i = 0; i < rowset->num_segments(); i++) {
+ auto segment_file = rowset->segment_file_path(i);
+ std::string inverted_index_src_file_path =
+ InvertedIndexDescriptor::get_index_file_name(
+ segment_file, index_meta->index_id(),
+ index_meta->get_index_suffix());
+ bool exists = false;
+ if (!fs->exists(inverted_index_src_file_path, &exists).ok()) {
+ LOG(ERROR) << inverted_index_src_file_path << " fs->exists error";
+ return false;
+ }
+
+ if (!exists) {
+ LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] column_unique_id["
+ << col_unique_id << "]," << inverted_index_src_file_path
+ << " is not exists, will skip index compaction";
+ return false;
+ }
+
+ // check idx file size
+ int64_t file_size = 0;
+ if (fs->file_size(inverted_index_src_file_path, &file_size) != Status::OK()) {
+ LOG(ERROR) << inverted_index_src_file_path << " fs->file_size error";
+ return false;
+ }
+
+ if (file_size == 0) {
+ LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] column_unique_id["
+ << col_unique_id << "]," << inverted_index_src_file_path
+ << " is empty file, will skip index compaction";
+ return false;
+ }
+
+ // check index meta
+ std::filesystem::path p(inverted_index_src_file_path);
+ std::string dir_str = p.parent_path().string();
+ std::string file_str = p.filename().string();
+ lucene::store::Directory* dir =
+ DorisCompoundDirectoryFactory::getDirectory(fs, dir_str.c_str());
+ DorisCompoundReader reader(dir, file_str.c_str());
+ std::vector<std::string> files;
+ reader.list(&files);
+ reader.close();
+
+ // why is 3?
+ // bkd index will write at least 3 files
+ if (files.size() < 3) {
+ LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] column_unique_id["
+ << col_unique_id << "]," << inverted_index_src_file_path
+ << " is corrupted, will skip index compaction";
+ return false;
}
}
+ return true;
+ };
+
+ bool all_have_inverted_index = std::all_of(_input_rowsets.begin(), _input_rowsets.end(),
+ std::move(has_inverted_index));
+
+ if (all_have_inverted_index &&
+ field_is_slice_type(_cur_tablet_schema->column_by_uid(col_unique_id).type())) {
+ ctx.skip_inverted_index.insert(col_unique_id);
}
}
- if (compaction_type() == ReaderType::READER_COLD_DATA_COMPACTION) {
- // write output rowset to storage policy resource
- auto storage_policy = get_storage_policy(_tablet->storage_policy_id());
- if (storage_policy == nullptr) {
- return Status::InternalError("could not find storage_policy, storage_policy_id={}",
- _tablet->storage_policy_id());
- }
- auto resource = get_storage_resource(storage_policy->resource_id);
- if (resource.fs == nullptr) {
- return Status::InternalError("could not find resource, resouce_id={}",
- storage_policy->resource_id);
- }
- DCHECK(atol(resource.fs->id().c_str()) == storage_policy->resource_id);
- DCHECK(resource.fs->type() != io::FileSystemType::LOCAL);
- ctx.fs = std::move(resource.fs);
- }
- _output_rs_writer = DORIS_TRY(_tablet->create_rowset_writer(ctx, is_vertical));
- _pending_rs_guard = ExecEnv::GetInstance()->storage_engine().to_local().add_pending_rowset(ctx);
- return Status::OK();
}
-Status Compaction::construct_input_rowset_readers() {
- for (auto& rowset : _input_rowsets) {
- RowsetReaderSharedPtr rs_reader;
- RETURN_IF_ERROR(rowset->create_reader(&rs_reader));
- _input_rs_readers.push_back(std::move(rs_reader));
+Status CompactionMixin::construct_output_rowset_writer(RowsetWriterContext& ctx) {
+ if (config::inverted_index_compaction_enable &&
+ ((_tablet->keys_type() == KeysType::UNIQUE_KEYS ||
+ _tablet->keys_type() == KeysType::DUP_KEYS))) {
+ construct_skip_inverted_index(ctx);
}
+ ctx.version = _output_version;
+ ctx.rowset_state = VISIBLE;
+ ctx.segments_overlap = NONOVERLAPPING;
+ ctx.tablet_schema = _cur_tablet_schema;
+ ctx.newest_write_timestamp = _newest_write_timestamp;
+ ctx.write_type = DataWriteType::TYPE_COMPACTION;
+ _output_rs_writer = DORIS_TRY(_tablet->create_rowset_writer(ctx, _is_vertical));
+ _pending_rs_guard = _engine.add_pending_rowset(ctx);
return Status::OK();
}
-Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
+Status CompactionMixin::modify_rowsets() {
std::vector<RowsetSharedPtr> output_rowsets;
output_rowsets.push_back(_output_rowset);
if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
_tablet->enable_unique_key_merge_on_write() &&
_tablet->tablet_schema()->cluster_key_idxes().empty()) {
- Version version = _tablet->max_version();
+ Version version = tablet()->max_version();
DeleteBitmap output_rowset_delete_bitmap(_tablet->tablet_id());
std::set<RowLocation> missed_rows;
std::map<RowsetSharedPtr, std::list<std::pair<RowLocation, RowLocation>>> location_map;
@@ -747,18 +680,18 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
// of incremental data later.
// TODO(LiaoXin): check if there are duplicate keys
std::size_t missed_rows_size = 0;
- _tablet->calc_compaction_output_rowset_delete_bitmap(
+ tablet()->calc_compaction_output_rowset_delete_bitmap(
_input_rowsets, _rowid_conversion, 0, version.second + 1, &missed_rows,
&location_map, _tablet->tablet_meta()->delete_bitmap(),
&output_rowset_delete_bitmap);
- if (!allow_delete_in_cumu_compaction()) {
+ if (!_allow_delete_in_cumu_compaction) {
missed_rows_size = missed_rows.size();
- if (compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION && stats != nullptr &&
- stats->merged_rows != missed_rows_size) {
+ if (compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION &&
+ _stats.merged_rows != missed_rows_size) {
std::string err_msg = fmt::format(
"cumulative compaction: the merged rows({}) is not equal to missed "
"rows({}) in rowid conversion, tablet_id: {}, table_id:{}",
- stats->merged_rows, missed_rows_size, _tablet->tablet_id(),
+ _stats.merged_rows, missed_rows_size, _tablet->tablet_id(),
_tablet->table_id());
DCHECK(false) << err_msg;
LOG(WARNING) << err_msg;
@@ -766,12 +699,12 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
}
if (config::enable_rowid_conversion_correctness_check) {
- RETURN_IF_ERROR(_tablet->check_rowid_conversion(_output_rowset, location_map));
+ RETURN_IF_ERROR(tablet()->check_rowid_conversion(_output_rowset, location_map));
}
location_map.clear();
{
- std::lock_guard<std::mutex> wrlock_(_tablet->get_rowset_update_lock());
+ std::lock_guard<std::mutex> wrlock_(tablet()->get_rowset_update_lock());
std::lock_guard<std::shared_mutex> wrlock(_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
@@ -780,12 +713,8 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
// All rowsets which need to recalculate have been published so we don't need to acquire lock.
// Step1: collect this tablet's all committed rowsets' delete bitmaps
CommitTabletTxnInfoVec commit_tablet_txn_info_vec {};
- ExecEnv::GetInstance()
- ->storage_engine()
- .to_local()
- .txn_manager()
- ->get_all_commit_tablet_txn_info_by_tablet(_tablet,
- &commit_tablet_txn_info_vec);
+ _engine.txn_manager()->get_all_commit_tablet_txn_info_by_tablet(
+ *tablet(), &commit_tablet_txn_info_vec);
// Step2: calculate all rowsets' delete bitmaps which are published during compaction.
for (auto& it : commit_tablet_txn_info_vec) {
@@ -798,7 +727,7 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
continue;
}
DeleteBitmap txn_output_delete_bitmap(_tablet->tablet_id());
- _tablet->calc_compaction_output_rowset_delete_bitmap(
+ tablet()->calc_compaction_output_rowset_delete_bitmap(
_input_rowsets, _rowid_conversion, 0, UINT64_MAX, &missed_rows,
&location_map, *it.delete_bitmap.get(), &txn_output_delete_bitmap);
if (config::enable_merge_on_write_correctness_check) {
@@ -810,24 +739,20 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
it.delete_bitmap->merge(txn_output_delete_bitmap);
// Step3: write back updated delete bitmap and tablet info.
it.rowset_ids.insert(_output_rowset->rowset_id());
- ExecEnv::GetInstance()
- ->storage_engine()
- .to_local()
- .txn_manager()
- ->set_txn_related_delete_bitmap(it.partition_id, it.transaction_id,
- _tablet->tablet_id(), _tablet->tablet_uid(),
- true, it.delete_bitmap, it.rowset_ids,
- it.partial_update_info);
+ _engine.txn_manager()->set_txn_related_delete_bitmap(
+ it.partition_id, it.transaction_id, _tablet->tablet_id(),
+ tablet()->tablet_uid(), true, it.delete_bitmap, it.rowset_ids,
+ it.partial_update_info);
}
// Convert the delete bitmap of the input rowsets to output rowset for
// incremental data.
- _tablet->calc_compaction_output_rowset_delete_bitmap(
+ tablet()->calc_compaction_output_rowset_delete_bitmap(
_input_rowsets, _rowid_conversion, version.second, UINT64_MAX, &missed_rows,
&location_map, _tablet->tablet_meta()->delete_bitmap(),
&output_rowset_delete_bitmap);
- if (!allow_delete_in_cumu_compaction() &&
+ if (!_allow_delete_in_cumu_compaction &&
compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION) {
DCHECK_EQ(missed_rows.size(), missed_rows_size);
if (missed_rows.size() != missed_rows_size) {
@@ -837,34 +762,34 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
}
if (config::enable_rowid_conversion_correctness_check) {
- RETURN_IF_ERROR(_tablet->check_rowid_conversion(_output_rowset, location_map));
+ RETURN_IF_ERROR(tablet()->check_rowid_conversion(_output_rowset, location_map));
}
- _tablet->merge_delete_bitmap(output_rowset_delete_bitmap);
- RETURN_IF_ERROR(_tablet->modify_rowsets(output_rowsets, _input_rowsets, true));
+ tablet()->merge_delete_bitmap(output_rowset_delete_bitmap);
+ RETURN_IF_ERROR(tablet()->modify_rowsets(output_rowsets, _input_rowsets, true));
}
} else {
std::lock_guard<std::shared_mutex> wrlock(_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
- RETURN_IF_ERROR(_tablet->modify_rowsets(output_rowsets, _input_rowsets, true));
+ RETURN_IF_ERROR(tablet()->modify_rowsets(output_rowsets, _input_rowsets, true));
}
if (config::tablet_rowset_stale_sweep_by_size &&
_tablet->tablet_meta()->all_stale_rs_metas().size() >=
config::tablet_rowset_stale_sweep_threshold_size) {
- _tablet->delete_expired_stale_rowset();
+ tablet()->delete_expired_stale_rowset();
}
int64_t cur_max_version = 0;
{
std::shared_lock rlock(_tablet->get_header_lock());
cur_max_version = _tablet->max_version_unlocked();
- _tablet->save_meta();
+ tablet()->save_meta();
}
if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
_tablet->enable_unique_key_merge_on_write()) {
auto st = TabletMetaManager::remove_old_version_delete_bitmap(
- _tablet->data_dir(), _tablet->tablet_id(), cur_max_version);
+ tablet()->data_dir(), _tablet->tablet_id(), cur_max_version);
if (!st.ok()) {
LOG(WARNING) << "failed to remove old version delete bitmap, st: " << st;
}
@@ -872,7 +797,7 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
return Status::OK();
}
-bool Compaction::_check_if_includes_input_rowsets(
+bool CompactionMixin::_check_if_includes_input_rowsets(
const RowsetIdUnorderedSet& commit_rowset_ids_set) const {
std::vector<RowsetId> commit_rowset_ids {};
commit_rowset_ids.insert(commit_rowset_ids.end(), commit_rowset_ids_set.begin(),
@@ -887,79 +812,21 @@ bool Compaction::_check_if_includes_input_rowsets(
input_rowset_ids.begin(), input_rowset_ids.end());
}
-void Compaction::gc_output_rowset() {
- if (_state != CompactionState::SUCCESS && _output_rowset != nullptr) {
- if (!_output_rowset->is_local()) {
- _tablet->record_unused_remote_rowset(_output_rowset->rowset_id(),
- _output_rowset->rowset_meta()->resource_id(),
- _output_rowset->num_segments());
- return;
- }
- ExecEnv::GetInstance()->storage_engine().to_local().add_unused_rowset(_output_rowset);
- }
-}
-
-// Find the longest consecutive version path in "rowset", from beginning.
-// Two versions before and after the missing version will be saved in missing_version,
-// if missing_version is not null.
-Status Compaction::find_longest_consecutive_version(std::vector<RowsetSharedPtr>* rowsets,
- std::vector<Version>* missing_version) {
- if (rowsets->empty()) {
- return Status::OK();
- }
- RowsetSharedPtr prev_rowset = rowsets->front();
- size_t i = 1;
- for (; i < rowsets->size(); ++i) {
- RowsetSharedPtr rowset = (*rowsets)[i];
- if (rowset->start_version() != prev_rowset->end_version() + 1) {
- if (missing_version != nullptr) {
- missing_version->push_back(prev_rowset->version());
- missing_version->push_back(rowset->version());
- }
- break;
- }
- prev_rowset = rowset;
- }
-
- rowsets->resize(i);
- return Status::OK();
-}
-
-Status Compaction::check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets) {
- if (rowsets.empty()) {
- return Status::OK();
- }
- RowsetSharedPtr prev_rowset = rowsets.front();
- for (size_t i = 1; i < rowsets.size(); ++i) {
- RowsetSharedPtr rowset = rowsets[i];
- if (rowset->start_version() != prev_rowset->end_version() + 1) {
- return Status::Error<CUMULATIVE_MISS_VERSION>(
- "There are missed versions among rowsets. prev_rowset version={}-{}, rowset "
- "version={}-{}",
- prev_rowset->start_version(), prev_rowset->end_version(),
- rowset->start_version(), rowset->end_version());
- }
- prev_rowset = rowset;
- }
-
- return Status::OK();
-}
-
-Status Compaction::check_correctness(const Merger::Statistics& stats) {
+Status Compaction::check_correctness() {
// 1. check row number
- if (_input_row_num != _output_rowset->num_rows() + stats.merged_rows + stats.filtered_rows) {
+ if (_input_row_num != _output_rowset->num_rows() + _stats.merged_rows + _stats.filtered_rows) {
return Status::Error<CHECK_LINES_ERROR>(
"row_num does not match between cumulative input and output! tablet={}, "
"input_row_num={}, merged_row_num={}, filtered_row_num={}, output_row_num={}",
- _tablet->tablet_id(), _input_row_num, stats.merged_rows, stats.filtered_rows,
+ _tablet->tablet_id(), _input_row_num, _stats.merged_rows, _stats.filtered_rows,
_output_rowset->num_rows());
}
return Status::OK();
}
-int64_t Compaction::get_compaction_permits() {
+int64_t CompactionMixin::get_compaction_permits() {
int64_t permits = 0;
- for (auto rowset : _input_rowsets) {
+ for (auto&& rowset : _input_rowsets) {
permits += rowset->rowset_meta()->get_compaction_score();
}
return permits;
@@ -977,13 +844,4 @@ void Compaction::_load_segment_to_cache() {
}
}
-#ifdef BE_TEST
-void Compaction::set_input_rowset(const std::vector<RowsetSharedPtr>& rowsets) {
- _input_rowsets = rowsets;
-}
-
-RowsetSharedPtr Compaction::output_rowset() {
- return _output_rowset;
-}
-#endif
} // namespace doris
diff --git a/be/src/olap/compaction.h b/be/src/olap/compaction.h
index a8e3b1a5c28..6998856ee94 100644
--- a/be/src/olap/compaction.h
+++ b/be/src/olap/compaction.h
@@ -17,9 +17,7 @@
#pragma once
-#include <butil/macros.h>
-#include <stdint.h>
-
+#include <cstdint>
#include <memory>
#include <string>
#include <vector>
@@ -30,15 +28,16 @@
#include "olap/olap_common.h"
#include "olap/rowid_conversion.h"
#include "olap/rowset/pending_rowset_helper.h"
-#include "olap/rowset/rowset.h"
-#include "olap/rowset/rowset_reader.h"
-#include "olap/tablet.h"
-#include "olap/tablet_schema.h"
+#include "olap/rowset/rowset_fwd.h"
+#include "olap/tablet_fwd.h"
+#include "util/runtime_profile.h"
namespace doris {
class MemTrackerLimiter;
class RowsetWriter;
+struct RowsetWriterContext;
+class StorageEngine;
// This class is a base class for compaction.
// The entrance of this class is compact()
@@ -49,90 +48,62 @@ class RowsetWriter;
// 4. gc output rowset if failed
class Compaction {
public:
- Compaction(const TabletSharedPtr& tablet, const std::string& label);
+ Compaction(BaseTabletSPtr tablet, const std::string& label);
virtual ~Compaction();
- // This is only for http CompactionAction
- Status compact();
-
+ // Pick input rowsets satisfied to compact
virtual Status prepare_compact() = 0;
- Status execute_compact();
- virtual Status execute_compact_impl() = 0;
- const std::vector<RowsetSharedPtr>& input_rowsets() { return _input_rowsets; }
-#ifdef BE_TEST
- void set_input_rowset(const std::vector<RowsetSharedPtr>& rowsets);
- RowsetSharedPtr output_rowset();
-#endif
+ // Merge input rowsets to output rowset and modify tablet meta
+ virtual Status execute_compact() = 0;
RuntimeProfile* runtime_profile() const { return _profile.get(); }
virtual ReaderType compaction_type() const = 0;
- virtual std::string compaction_name() const = 0;
+ virtual std::string_view compaction_name() const = 0;
protected:
- virtual Status pick_rowsets_to_compact() = 0;
-
- Status do_compaction(int64_t permits);
- Status do_compaction_impl(int64_t permits);
+ Status merge_input_rowsets();
- virtual Status modify_rowsets(const Merger::Statistics* stats = nullptr);
- void gc_output_rowset();
+ virtual Status construct_output_rowset_writer(RowsetWriterContext& ctx) = 0;
- Status construct_output_rowset_writer(RowsetWriterContext& ctx, bool is_vertical = false);
- Status construct_input_rowset_readers();
-
- Status check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets);
- Status check_correctness(const Merger::Statistics& stats);
- Status find_longest_consecutive_version(std::vector<RowsetSharedPtr>* rowsets,
- std::vector<Version>* missing_version);
- int64_t get_compaction_permits();
+ Status check_correctness();
- bool should_vertical_compaction();
int64_t get_avg_segment_rows();
- bool handle_ordered_data_compaction();
- Status do_compact_ordered_rowsets();
- bool is_rowset_tidy(std::string& pre_max_key, const RowsetSharedPtr& rhs);
- void build_basic_info();
-
void init_profile(const std::string& label);
- [[nodiscard]] bool allow_delete_in_cumu_compaction() const {
- return _allow_delete_in_cumu_compaction;
- }
-private:
- bool _check_if_includes_input_rowsets(const RowsetIdUnorderedSet& commit_rowset_ids_set) const;
void _load_segment_to_cache();
-protected:
// the root tracker for this compaction
std::shared_ptr<MemTrackerLimiter> _mem_tracker;
- TabletSharedPtr _tablet;
+ BaseTabletSPtr _tablet;
std::vector<RowsetSharedPtr> _input_rowsets;
- std::vector<RowsetReaderSharedPtr> _input_rs_readers;
- int64_t _input_rowsets_size;
- int64_t _input_row_num;
- int64_t _input_num_segments;
- int64_t _input_index_size;
+ int64_t _input_rowsets_size {0};
+ int64_t _input_row_num {0};
+ int64_t _input_num_segments {0};
+ int64_t _input_index_size {0};
+
+ Merger::Statistics _stats;
RowsetSharedPtr _output_rowset;
- PendingRowsetGuard _pending_rs_guard;
std::unique_ptr<RowsetWriter> _output_rs_writer;
- enum CompactionState { INITED = 0, SUCCESS = 1 };
- CompactionState _state;
+ enum CompactionState : uint8_t { INITED = 0, SUCCESS = 1 };
+ CompactionState _state {CompactionState::INITED};
+
+ bool _is_vertical;
+ bool _allow_delete_in_cumu_compaction;
Version _output_version;
- int64_t _newest_write_timestamp;
+ int64_t _newest_write_timestamp {-1};
RowIdConversion _rowid_conversion;
TabletSchemaSPtr _cur_tablet_schema;
std::unique_ptr<RuntimeProfile> _profile;
- bool _allow_delete_in_cumu_compaction = false;
RuntimeProfile::Counter* _input_rowsets_data_size_counter = nullptr;
RuntimeProfile::Counter* _input_rowsets_counter = nullptr;
@@ -144,8 +115,45 @@ protected:
RuntimeProfile::Counter* _output_row_num_counter = nullptr;
RuntimeProfile::Counter* _output_segments_num_counter = nullptr;
RuntimeProfile::Counter* _merge_rowsets_latency_timer = nullptr;
+};
+
+// `StorageEngine` mixin for `Compaction`
+class CompactionMixin : public Compaction {
+public:
+ CompactionMixin(StorageEngine& engine, TabletSharedPtr tablet, const std::string& label);
+
+ ~CompactionMixin() override;
+
+ Status execute_compact() override;
- DISALLOW_COPY_AND_ASSIGN(Compaction);
+ int64_t get_compaction_permits();
+
+protected:
+ // Convert `_tablet` from `BaseTablet` to `Tablet`
+ Tablet* tablet();
+
+ Status construct_output_rowset_writer(RowsetWriterContext& ctx) override;
+
+ virtual Status modify_rowsets();
+
+ StorageEngine& _engine;
+
+private:
+ Status execute_compact_impl(int64_t permits);
+
+ void build_basic_info();
+
+ void construct_skip_inverted_index(RowsetWriterContext& ctx);
+
+ bool handle_ordered_data_compaction();
+
+ Status do_compact_ordered_rowsets();
+
+ Status do_inverted_index_compaction();
+
+ bool _check_if_includes_input_rowsets(const RowsetIdUnorderedSet& commit_rowset_ids_set) const;
+
+ PendingRowsetGuard _pending_rs_guard;
};
} // namespace doris
diff --git a/be/src/olap/cumulative_compaction.cpp b/be/src/olap/cumulative_compaction.cpp
index 1f54c1f3285..9de5837b935 100644
--- a/be/src/olap/cumulative_compaction.cpp
+++ b/be/src/olap/cumulative_compaction.cpp
@@ -17,9 +17,6 @@
#include "olap/cumulative_compaction.h"
-#include <stddef.h>
-#include <stdint.h>
-
#include <memory>
#include <mutex>
#include <ostream>
@@ -29,6 +26,7 @@
#include "olap/cumulative_compaction_policy.h"
#include "olap/olap_define.h"
#include "olap/rowset/rowset_meta.h"
+#include "olap/tablet.h"
#include "runtime/thread_context.h"
#include "util/doris_metrics.h"
#include "util/time.h"
@@ -37,36 +35,62 @@
namespace doris {
using namespace ErrorCode;
-CumulativeCompaction::CumulativeCompaction(const TabletSharedPtr& tablet)
- : Compaction(tablet, "CumulativeCompaction:" + std::to_string(tablet->tablet_id())) {}
+namespace {
+
+void find_longest_consecutive_version(std::vector<RowsetSharedPtr>* rowsets,
+ std::vector<Version>* missing_version) {
+ if (rowsets->empty()) {
+ return;
+ }
+
+ RowsetSharedPtr prev_rowset = rowsets->front();
+ size_t i = 1;
+ for (; i < rowsets->size(); ++i) {
+ RowsetSharedPtr rowset = (*rowsets)[i];
+ if (rowset->start_version() != prev_rowset->end_version() + 1) {
+ if (missing_version != nullptr) {
+ missing_version->push_back(prev_rowset->version());
+ missing_version->push_back(rowset->version());
+ }
+ break;
+ }
+ prev_rowset = rowset;
+ }
+
+ rowsets->resize(i);
+}
+
+} // namespace
+
+CumulativeCompaction::CumulativeCompaction(StorageEngine& engine, const TabletSharedPtr& tablet)
+ : CompactionMixin(engine, tablet,
+ "CumulativeCompaction:" + std::to_string(tablet->tablet_id())) {}
CumulativeCompaction::~CumulativeCompaction() = default;
Status CumulativeCompaction::prepare_compact() {
- if (!_tablet->init_succeeded()) {
+ if (!tablet()->init_succeeded()) {
return Status::Error<CUMULATIVE_INVALID_PARAMETERS, false>("_tablet init failed");
}
- std::unique_lock<std::mutex> lock(_tablet->get_cumulative_compaction_lock(), std::try_to_lock);
+ std::unique_lock<std::mutex> lock(tablet()->get_cumulative_compaction_lock(), std::try_to_lock);
if (!lock.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>(
"The tablet is under cumulative compaction. tablet={}", _tablet->tablet_id());
}
- // 1. calculate cumulative point
- _tablet->calculate_cumulative_point();
+ tablet()->calculate_cumulative_point();
VLOG_CRITICAL << "after calculate, current cumulative point is "
- << _tablet->cumulative_layer_point() << ", tablet=" << _tablet->tablet_id();
+ << tablet()->cumulative_layer_point() << ", tablet=" << _tablet->tablet_id();
- // 2. pick rowsets to compact
RETURN_IF_ERROR(pick_rowsets_to_compact());
COUNTER_UPDATE(_input_rowsets_counter, _input_rowsets.size());
return Status::OK();
}
-Status CumulativeCompaction::execute_compact_impl() {
- std::unique_lock<std::mutex> lock(_tablet->get_cumulative_compaction_lock(), std::try_to_lock);
+Status CumulativeCompaction::execute_compact() {
+ std::unique_lock<std::mutex> lock(tablet()->get_cumulative_compaction_lock(), std::try_to_lock);
if (!lock.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>(
"The tablet is under cumulative compaction. tablet={}", _tablet->tablet_id());
@@ -74,20 +98,15 @@ Status CumulativeCompaction::execute_compact_impl() {
SCOPED_ATTACH_TASK(_mem_tracker);
- // 3. do cumulative compaction, merge rowsets
- int64_t permits = get_compaction_permits();
- RETURN_IF_ERROR(do_compaction(permits));
-
- // 4. set state to success
- _state = CompactionState::SUCCESS;
+ RETURN_IF_ERROR(CompactionMixin::execute_compact());
+ DCHECK_EQ(_state, CompactionState::SUCCESS);
- // 5. set cumulative point
- _tablet->cumulative_compaction_policy()->update_cumulative_point(
- _tablet.get(), _input_rowsets, _output_rowset, _last_delete_version);
+ tablet()->cumulative_compaction_policy()->update_cumulative_point(
+ tablet(), _input_rowsets, _output_rowset, _last_delete_version);
VLOG_CRITICAL << "after cumulative compaction, current cumulative point is "
- << _tablet->cumulative_layer_point() << ", tablet=" << _tablet->tablet_id();
+ << tablet()->cumulative_layer_point() << ", tablet=" << _tablet->tablet_id();
- // 6. add metric to cumulative compaction
+ tablet()->set_last_cumu_compaction_success_time(UnixMillis());
DorisMetrics::instance()->cumulative_compaction_deltas_total->increment(_input_rowsets.size());
DorisMetrics::instance()->cumulative_compaction_bytes_total->increment(_input_rowsets_size);
@@ -95,7 +114,7 @@ Status CumulativeCompaction::execute_compact_impl() {
}
Status CumulativeCompaction::pick_rowsets_to_compact() {
- auto candidate_rowsets = _tablet->pick_candidate_rowsets_to_cumulative_compaction();
+ auto candidate_rowsets = tablet()->pick_candidate_rowsets_to_cumulative_compaction();
if (candidate_rowsets.empty()) {
return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>("candidate_rowsets is empty");
}
@@ -103,7 +122,7 @@ Status CumulativeCompaction::pick_rowsets_to_compact() {
// candidate_rowsets may not be continuous
// So we need to choose the longest continuous path from it.
std::vector<Version> missing_versions;
- RETURN_IF_ERROR(find_longest_consecutive_version(&candidate_rowsets, &missing_versions));
+ find_longest_consecutive_version(&candidate_rowsets, &missing_versions);
if (!missing_versions.empty()) {
DCHECK(missing_versions.size() == 2);
LOG(WARNING) << "There are missed versions among rowsets. "
@@ -113,10 +132,10 @@ Status CumulativeCompaction::pick_rowsets_to_compact() {
}
size_t compaction_score = 0;
- _tablet->cumulative_compaction_policy()->pick_input_rowsets(
- _tablet.get(), candidate_rowsets, config::cumulative_compaction_max_deltas,
+ tablet()->cumulative_compaction_policy()->pick_input_rowsets(
+ tablet(), candidate_rowsets, config::cumulative_compaction_max_deltas,
config::cumulative_compaction_min_deltas, &_input_rowsets, &_last_delete_version,
- &compaction_score, allow_delete_in_cumu_compaction());
+ &compaction_score, _allow_delete_in_cumu_compaction);
// Cumulative compaction will process with at least 1 rowset.
// So when there is no rowset being chosen, we should return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>():
@@ -125,7 +144,7 @@ Status CumulativeCompaction::pick_rowsets_to_compact() {
// we meet a delete version, should increase the cumulative point to let base compaction handle the delete version.
// plus 1 to skip the delete version.
// NOTICE: after that, the cumulative point may be larger than max version of this tablet, but it doesn't matter.
- _tablet->set_cumulative_layer_point(_last_delete_version.first + 1);
+ tablet()->set_cumulative_layer_point(_last_delete_version.first + 1);
return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
"_last_delete_version.first not equal to -1");
}
@@ -137,8 +156,8 @@ Status CumulativeCompaction::pick_rowsets_to_compact() {
// check both last success time of base and cumulative compaction
int64_t now = UnixMillis();
- int64_t last_cumu = _tablet->last_cumu_compaction_success_time();
- int64_t last_base = _tablet->last_base_compaction_success_time();
+ 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::pick_rowset_to_compact_interval_sec * 1000;
int64_t cumu_interval = now - last_cumu;
@@ -154,16 +173,16 @@ Status CumulativeCompaction::pick_rowsets_to_compact() {
}
// all candidate rowsets are non-overlapping, increase the cumulative point
- _tablet->set_cumulative_layer_point(candidate_rowsets.back()->start_version() + 1);
+ tablet()->set_cumulative_layer_point(candidate_rowsets.back()->start_version() + 1);
}
} else {
// init the compaction success time for first time
if (last_cumu == 0) {
- _tablet->set_last_cumu_compaction_success_time(now);
+ tablet()->set_last_cumu_compaction_success_time(now);
}
if (last_base == 0) {
- _tablet->set_last_base_compaction_success_time(now);
+ tablet()->set_last_base_compaction_success_time(now);
}
}
diff --git a/be/src/olap/cumulative_compaction.h b/be/src/olap/cumulative_compaction.h
index 7ea7fb383f1..14527bf2fab 100644
--- a/be/src/olap/cumulative_compaction.h
+++ b/be/src/olap/cumulative_compaction.h
@@ -17,8 +17,6 @@
#pragma once
-#include <butil/macros.h>
-
#include <string>
#include <vector>
@@ -26,30 +24,27 @@
#include "io/io_common.h"
#include "olap/compaction.h"
#include "olap/olap_common.h"
-#include "olap/rowset/rowset.h"
-#include "olap/tablet.h"
namespace doris {
-class CumulativeCompaction : public Compaction {
+class CumulativeCompaction final : public CompactionMixin {
public:
- CumulativeCompaction(const TabletSharedPtr& tablet);
+ CumulativeCompaction(StorageEngine& engine, const TabletSharedPtr& tablet);
+
~CumulativeCompaction() override;
Status prepare_compact() override;
- Status execute_compact_impl() override;
-protected:
- Status pick_rowsets_to_compact() override;
+ Status execute_compact() override;
- std::string compaction_name() const override { return "cumulative compaction"; }
+private:
+ std::string_view compaction_name() const override { return "cumulative compaction"; }
ReaderType compaction_type() const override { return ReaderType::READER_CUMULATIVE_COMPACTION; }
-private:
- Version _last_delete_version {-1, -1};
+ Status pick_rowsets_to_compact();
- DISALLOW_COPY_AND_ASSIGN(CumulativeCompaction);
+ Version _last_delete_version {-1, -1};
};
} // namespace doris
diff --git a/be/src/olap/full_compaction.cpp b/be/src/olap/full_compaction.cpp
index f98ac2a24e7..b247d6c0ea1 100644
--- a/be/src/olap/full_compaction.cpp
+++ b/be/src/olap/full_compaction.cpp
@@ -27,6 +27,7 @@
#include "common/config.h"
#include "common/status.h"
+#include "olap/compaction.h"
#include "olap/cumulative_compaction_policy.h"
#include "olap/olap_common.h"
#include "olap/olap_define.h"
@@ -41,18 +42,19 @@
namespace doris {
using namespace ErrorCode;
-FullCompaction::FullCompaction(const TabletSharedPtr& tablet)
- : Compaction(tablet, "FullCompaction:" + std::to_string(tablet->tablet_id())) {}
+FullCompaction::FullCompaction(StorageEngine& engine, const TabletSharedPtr& tablet)
+ : CompactionMixin(engine, tablet, "FullCompaction:" + std::to_string(tablet->tablet_id())) {
+}
-FullCompaction::~FullCompaction() {}
+FullCompaction::~FullCompaction() = default;
Status FullCompaction::prepare_compact() {
- if (!_tablet->init_succeeded()) {
+ if (!tablet()->init_succeeded()) {
return Status::Error<INVALID_ARGUMENT, false>("Full compaction init failed");
}
- std::unique_lock base_lock(_tablet->get_base_compaction_lock());
- std::unique_lock cumu_lock(_tablet->get_cumulative_compaction_lock());
+ std::unique_lock base_lock(tablet()->get_base_compaction_lock());
+ std::unique_lock cumu_lock(tablet()->get_cumulative_compaction_lock());
// 1. pick rowsets to compact
RETURN_IF_ERROR(pick_rowsets_to_compact());
@@ -60,31 +62,27 @@ Status FullCompaction::prepare_compact() {
return Status::OK();
}
-Status FullCompaction::execute_compact_impl() {
- std::unique_lock base_lock(_tablet->get_base_compaction_lock());
- std::unique_lock cumu_lock(_tablet->get_cumulative_compaction_lock());
+Status FullCompaction::execute_compact() {
+ std::unique_lock base_lock(tablet()->get_base_compaction_lock());
+ std::unique_lock cumu_lock(tablet()->get_cumulative_compaction_lock());
SCOPED_ATTACH_TASK(_mem_tracker);
- // 2. do full compaction, merge rowsets
- int64_t permits = get_compaction_permits();
- RETURN_IF_ERROR(do_compaction(permits));
-
- // 3. set state to success
- _state = CompactionState::SUCCESS;
+ RETURN_IF_ERROR(CompactionMixin::execute_compact());
- // 4. set cumulative point
Version last_version = _input_rowsets.back()->version();
- _tablet->cumulative_compaction_policy()->update_cumulative_point(_tablet.get(), _input_rowsets,
- _output_rowset, last_version);
+ tablet()->cumulative_compaction_policy()->update_cumulative_point(tablet(), _input_rowsets,
+ _output_rowset, last_version);
VLOG_CRITICAL << "after cumulative compaction, current cumulative point is "
- << _tablet->cumulative_layer_point() << ", tablet=" << _tablet->tablet_id();
+ << tablet()->cumulative_layer_point() << ", tablet=" << _tablet->tablet_id();
+
+ tablet()->set_last_full_compaction_success_time(UnixMillis());
return Status::OK();
}
Status FullCompaction::pick_rowsets_to_compact() {
- _input_rowsets = _tablet->pick_candidate_rowsets_to_full_compaction();
+ _input_rowsets = tablet()->pick_candidate_rowsets_to_full_compaction();
RETURN_IF_ERROR(check_version_continuity(_input_rowsets));
RETURN_IF_ERROR(_check_all_version(_input_rowsets));
if (_input_rowsets.size() <= 1) {
@@ -100,7 +98,7 @@ Status FullCompaction::pick_rowsets_to_compact() {
return Status::OK();
}
-Status FullCompaction::modify_rowsets(const Merger::Statistics* stats) {
+Status FullCompaction::modify_rowsets() {
if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
_tablet->enable_unique_key_merge_on_write()) {
RETURN_IF_ERROR(
@@ -108,10 +106,10 @@ Status FullCompaction::modify_rowsets(const Merger::Statistics* stats) {
}
std::vector<RowsetSharedPtr> output_rowsets(1, _output_rowset);
{
- std::lock_guard<std::mutex> rowset_update_wlock(_tablet->get_rowset_update_lock());
+ std::lock_guard<std::mutex> rowset_update_wlock(tablet()->get_rowset_update_lock());
std::lock_guard<std::shared_mutex> meta_wlock(_tablet->get_header_lock());
- RETURN_IF_ERROR(_tablet->modify_rowsets(output_rowsets, _input_rowsets, true));
- _tablet->save_meta();
+ RETURN_IF_ERROR(tablet()->modify_rowsets(output_rowsets, _input_rowsets, true));
+ tablet()->save_meta();
}
return Status::OK();
}
@@ -123,15 +121,15 @@ Status FullCompaction::_check_all_version(const std::vector<RowsetSharedPtr>& ro
}
const RowsetSharedPtr& last_rowset = rowsets.back();
const RowsetSharedPtr& first_rowset = rowsets.front();
- if (last_rowset->version() != _tablet->max_version() || first_rowset->version().first != 0) {
+ auto max_version = tablet()->max_version();
+ if (last_rowset->version() != max_version || first_rowset->version().first != 0) {
return Status::Error<FULL_MISS_VERSION, false>(
"Full compaction rowsets' versions not equal to all exist rowsets' versions. "
"full compaction rowsets max version={}-{}"
", current rowsets max version={}-{}"
"full compaction rowsets min version={}-{}, current rowsets min version=0-1",
- last_rowset->start_version(), last_rowset->end_version(),
- _tablet->max_version().first, _tablet->max_version().second,
- first_rowset->start_version(), first_rowset->end_version());
+ last_rowset->start_version(), last_rowset->end_version(), max_version.first,
+ max_version.second, first_rowset->start_version(), first_rowset->end_version());
}
return Status::OK();
}
@@ -147,7 +145,7 @@ Status FullCompaction::_full_compaction_update_delete_bitmap(const RowsetSharedP
return Status::OK();
}
- int64_t max_version = _tablet->max_version().second;
+ int64_t max_version = tablet()->max_version().second;
DCHECK(max_version >= rowset->version().second);
if (max_version > rowset->version().second) {
RETURN_IF_ERROR(_tablet->capture_consistent_rowsets_unlocked(
@@ -160,10 +158,10 @@ Status FullCompaction::_full_compaction_update_delete_bitmap(const RowsetSharedP
_full_compaction_calc_delete_bitmap(it, rowset, cur_version, rowset_writer));
}
- std::lock_guard rowset_update_lock(_tablet->get_rowset_update_lock());
+ std::lock_guard rowset_update_lock(tablet()->get_rowset_update_lock());
std::lock_guard header_lock(_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
- for (const auto& it : _tablet->rowset_map()) {
+ for (const auto& it : tablet()->rowset_map()) {
const int64_t& cur_version = it.first.first;
const RowsetSharedPtr& published_rowset = it.second;
if (cur_version > max_version) {
diff --git a/be/src/olap/full_compaction.h b/be/src/olap/full_compaction.h
index 631d901e846..cb11ac70aa3 100644
--- a/be/src/olap/full_compaction.h
+++ b/be/src/olap/full_compaction.h
@@ -25,27 +25,28 @@
#include "common/status.h"
#include "io/io_common.h"
#include "olap/compaction.h"
-#include "olap/rowset/rowset.h"
-#include "olap/tablet.h"
namespace doris {
-class FullCompaction : public Compaction {
+class FullCompaction final : public CompactionMixin {
public:
- FullCompaction(const TabletSharedPtr& tablet);
+ FullCompaction(StorageEngine& engine, const TabletSharedPtr& tablet);
+
~FullCompaction() override;
Status prepare_compact() override;
- Status execute_compact_impl() override;
- Status modify_rowsets(const Merger::Statistics* stats = nullptr) override;
-protected:
- Status pick_rowsets_to_compact() override;
- std::string compaction_name() const override { return "full compaction"; }
+ Status execute_compact() override;
+
+private:
+ Status pick_rowsets_to_compact();
+
+ Status modify_rowsets() override;
+
+ std::string_view compaction_name() const override { return "full compaction"; }
ReaderType compaction_type() const override { return ReaderType::READER_FULL_COMPACTION; }
-private:
Status _check_all_version(const std::vector<RowsetSharedPtr>& rowsets);
Status _full_compaction_update_delete_bitmap(const RowsetSharedPtr& rowset,
RowsetWriter* rowset_writer);
@@ -53,8 +54,6 @@ private:
const RowsetSharedPtr& rowset,
const int64_t& cur_version,
RowsetWriter* rowset_writer);
-
- DISALLOW_COPY_AND_ASSIGN(FullCompaction);
};
} // namespace doris
diff --git a/be/src/olap/merger.cpp b/be/src/olap/merger.cpp
index e2449cd8e57..33dd99ca7b6 100644
--- a/be/src/olap/merger.cpp
+++ b/be/src/olap/merger.cpp
@@ -51,9 +51,33 @@
#include "vec/olap/vertical_merge_iterator.h"
namespace doris {
+namespace {
+
+// for mow with cluster key table, the key group also contains cluster key columns.
+// the `key_group_cluster_key_idxes` marks the positions of cluster key columns in key group.
+void _generate_key_group_cluster_key_idxes(const TabletSchema& tablet_schema,
+ std::vector<std::vector<uint32_t>>& column_groups,
+ std::vector<uint32_t>& key_group_cluster_key_idxes) {
+ if (column_groups.empty() || tablet_schema.cluster_key_idxes().empty()) {
+ return;
+ }
+
+ auto& key_column_group = column_groups[0];
+ for (const auto& index_in_tablet_schema : tablet_schema.cluster_key_idxes()) {
+ for (auto j = 0; j < key_column_group.size(); ++j) {
+ auto cid = key_column_group[j];
+ if (cid == index_in_tablet_schema) {
+ key_group_cluster_key_idxes.emplace_back(j);
+ break;
+ }
+ }
+ }
+}
-Status Merger::vmerge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
- TabletSchemaSPtr cur_tablet_schema,
+} // namespace
+
+Status Merger::vmerge_rowsets(BaseTabletSPtr tablet, ReaderType reader_type,
+ const TabletSchema& cur_tablet_schema,
const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,
RowsetWriter* dst_rowset_writer, Statistics* stats_output) {
vectorized::BlockReader reader;
@@ -64,7 +88,7 @@ Status Merger::vmerge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
TabletReader::ReadSource read_source;
read_source.rs_splits.reserve(src_rowset_readers.size());
for (const RowsetReaderSharedPtr& rs_reader : src_rowset_readers) {
- read_source.rs_splits.emplace_back(RowSetSplits(rs_reader));
+ read_source.rs_splits.emplace_back(rs_reader);
}
read_source.fill_delete_predicates();
reader_params.set_read_source(std::move(read_source));
@@ -72,7 +96,7 @@ Status Merger::vmerge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
reader_params.version = dst_rowset_writer->version();
TabletSchemaSPtr merge_tablet_schema = std::make_shared<TabletSchema>();
- merge_tablet_schema->copy_from(*cur_tablet_schema);
+ merge_tablet_schema->copy_from(cur_tablet_schema);
// Merge the columns in delete predicate that not in latest schema in to current tablet schema
for (auto& del_pred_rs : reader_params.delete_predicates) {
@@ -87,7 +111,7 @@ Status Merger::vmerge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
reader_params.record_rowids = true;
}
- reader_params.return_columns.resize(cur_tablet_schema->num_columns());
+ reader_params.return_columns.resize(cur_tablet_schema.num_columns());
std::iota(reader_params.return_columns.begin(), reader_params.return_columns.end(), 0);
reader_params.origin_return_columns = &reader_params.return_columns;
RETURN_IF_ERROR(reader.init(reader_params));
@@ -103,7 +127,7 @@ Status Merger::vmerge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
}
}
- vectorized::Block block = cur_tablet_schema->create_block(reader_params.return_columns);
+ vectorized::Block block = cur_tablet_schema.create_block(reader_params.return_columns);
size_t output_rows = 0;
bool eof = false;
while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {
@@ -145,10 +169,10 @@ Status Merger::vmerge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
// split columns into several groups, make sure all keys in one group
// unique_key should consider sequence&delete column
-void Merger::vertical_split_columns(TabletSchemaSPtr tablet_schema,
+void Merger::vertical_split_columns(const TabletSchema& tablet_schema,
std::vector<std::vector<uint32_t>>* column_groups) {
- uint32_t num_key_cols = tablet_schema->num_key_columns();
- uint32_t total_cols = tablet_schema->num_columns();
+ uint32_t num_key_cols = tablet_schema.num_key_columns();
+ uint32_t total_cols = tablet_schema.num_columns();
std::vector<uint32_t> key_columns;
for (auto i = 0; i < num_key_cols; ++i) {
key_columns.emplace_back(i);
@@ -158,17 +182,17 @@ void Merger::vertical_split_columns(TabletSchemaSPtr tablet_schema,
int32_t delete_sign_idx = -1;
// in key column compaction, seq_col real index is _num_key_columns
// and delete_sign column is _block->columns() - 1
- if (tablet_schema->keys_type() == KeysType::UNIQUE_KEYS) {
- if (tablet_schema->has_sequence_col()) {
- sequence_col_idx = tablet_schema->sequence_col_idx();
+ if (tablet_schema.keys_type() == KeysType::UNIQUE_KEYS) {
+ if (tablet_schema.has_sequence_col()) {
+ sequence_col_idx = tablet_schema.sequence_col_idx();
key_columns.emplace_back(sequence_col_idx);
}
- delete_sign_idx = tablet_schema->field_index(DELETE_SIGN);
+ delete_sign_idx = tablet_schema.field_index(DELETE_SIGN);
if (delete_sign_idx != -1) {
key_columns.emplace_back(delete_sign_idx);
}
- if (!tablet_schema->cluster_key_idxes().empty()) {
- for (const auto& cid : tablet_schema->cluster_key_idxes()) {
+ if (!tablet_schema.cluster_key_idxes().empty()) {
+ for (const auto& cid : tablet_schema.cluster_key_idxes()) {
if (cid >= num_key_cols) {
key_columns.emplace_back(cid);
}
@@ -181,7 +205,7 @@ void Merger::vertical_split_columns(TabletSchemaSPtr tablet_schema,
if (!key_columns.empty()) {
column_groups->emplace_back(std::move(key_columns));
}
- auto&& cluster_key_idxes = tablet_schema->cluster_key_idxes();
+ auto&& cluster_key_idxes = tablet_schema.cluster_key_idxes();
for (uint32_t i = num_key_cols; i < total_cols; ++i) {
if (i == sequence_col_idx || i == delete_sign_idx ||
cluster_key_idxes.end() !=
@@ -196,8 +220,9 @@ void Merger::vertical_split_columns(TabletSchemaSPtr tablet_schema,
}
Status Merger::vertical_compact_one_group(
- TabletSharedPtr tablet, ReaderType reader_type, TabletSchemaSPtr tablet_schema, bool is_key,
- const std::vector<uint32_t>& column_group, vectorized::RowSourcesBuffer* row_source_buf,
+ BaseTabletSPtr tablet, ReaderType reader_type, const TabletSchema& tablet_schema,
+ bool is_key, const std::vector<uint32_t>& column_group,
+ vectorized::RowSourcesBuffer* row_source_buf,
const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,
RowsetWriter* dst_rowset_writer, int64_t max_rows_per_segment, Statistics* stats_output,
std::vector<uint32_t> key_group_cluster_key_idxes) {
@@ -213,7 +238,7 @@ Status Merger::vertical_compact_one_group(
TabletReader::ReadSource read_source;
read_source.rs_splits.reserve(src_rowset_readers.size());
for (const RowsetReaderSharedPtr& rs_reader : src_rowset_readers) {
- read_source.rs_splits.emplace_back(RowSetSplits(rs_reader));
+ read_source.rs_splits.emplace_back(rs_reader);
}
read_source.fill_delete_predicates();
reader_params.set_read_source(std::move(read_source));
@@ -221,7 +246,7 @@ Status Merger::vertical_compact_one_group(
reader_params.version = dst_rowset_writer->version();
TabletSchemaSPtr merge_tablet_schema = std::make_shared<TabletSchema>();
- merge_tablet_schema->copy_from(*tablet_schema);
+ merge_tablet_schema->copy_from(tablet_schema);
for (auto& del_pred_rs : reader_params.delete_predicates) {
merge_tablet_schema->merge_dropped_columns(*del_pred_rs->tablet_schema());
@@ -251,7 +276,7 @@ Status Merger::vertical_compact_one_group(
}
}
- vectorized::Block block = tablet_schema->create_block(reader_params.return_columns);
+ vectorized::Block block = tablet_schema.create_block(reader_params.return_columns);
size_t output_rows = 0;
bool eof = false;
while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {
@@ -289,8 +314,8 @@ Status Merger::vertical_compact_one_group(
}
// for segcompaction
-Status Merger::vertical_compact_one_group(TabletSharedPtr tablet, ReaderType reader_type,
- TabletSchemaSPtr tablet_schema, bool is_key,
+Status Merger::vertical_compact_one_group(int64_t tablet_id, ReaderType reader_type,
+ const TabletSchema& tablet_schema, bool is_key,
const std::vector<uint32_t>& column_group,
vectorized::RowSourcesBuffer* row_source_buf,
vectorized::VerticalBlockReader& src_block_reader,
@@ -300,27 +325,27 @@ Status Merger::vertical_compact_one_group(TabletSharedPtr tablet, ReaderType rea
// build tablet reader
VLOG_NOTICE << "vertical compact one group, max_rows_per_segment=" << max_rows_per_segment;
// TODO: record_rowids
- vectorized::Block block = tablet_schema->create_block(column_group);
+ vectorized::Block block = tablet_schema.create_block(column_group);
size_t output_rows = 0;
bool eof = false;
while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {
// Read one block from block reader
RETURN_NOT_OK_STATUS_WITH_WARN(src_block_reader.next_block_with_aggregation(&block, &eof),
"failed to read next block when merging rowsets of tablet " +
- std::to_string(tablet->tablet_id()));
+ std::to_string(tablet_id));
if (!block.rows()) {
break;
}
RETURN_NOT_OK_STATUS_WITH_WARN(dst_segment_writer.append_block(&block, 0, block.rows()),
"failed to write block when merging rowsets of tablet " +
- std::to_string(tablet->tablet_id()));
+ std::to_string(tablet_id));
output_rows += block.rows();
block.clear_column_data();
}
if (ExecEnv::GetInstance()->storage_engine().stopped()) {
return Status::Error<INTERNAL_ERROR>("tablet {} failed to do compaction, engine stopped",
- tablet->tablet_id());
+ tablet_id);
}
if (is_key && stats_output != nullptr) {
@@ -349,8 +374,8 @@ Status Merger::vertical_compact_one_group(TabletSharedPtr tablet, ReaderType rea
// 2. compact groups one by one, generate a row_source_buf when compact key group
// and use this row_source_buf to compact value column groups
// 3. build output rowset
-Status Merger::vertical_merge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
- TabletSchemaSPtr tablet_schema,
+Status Merger::vertical_merge_rowsets(BaseTabletSPtr tablet, ReaderType reader_type,
+ const TabletSchema& tablet_schema,
const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,
RowsetWriter* dst_rowset_writer, int64_t max_rows_per_segment,
Statistics* stats_output) {
@@ -385,21 +410,4 @@ Status Merger::vertical_merge_rowsets(TabletSharedPtr tablet, ReaderType reader_
return Status::OK();
}
-void Merger::_generate_key_group_cluster_key_idxes(
- TabletSchemaSPtr tablet_schema, std::vector<std::vector<uint32_t>>& column_groups,
- std::vector<uint32_t>& key_group_cluster_key_idxes) {
- if (column_groups.size() > 0 && !tablet_schema->cluster_key_idxes().empty()) {
- auto& key_column_group = column_groups[0];
- for (const auto& index_in_tablet_schema : tablet_schema->cluster_key_idxes()) {
- for (auto j = 0; j < key_column_group.size(); ++j) {
- auto cid = key_column_group[j];
- if (cid == index_in_tablet_schema) {
- key_group_cluster_key_idxes.emplace_back(j);
- break;
- }
- }
- }
- }
-}
-
} // namespace doris
diff --git a/be/src/olap/merger.h b/be/src/olap/merger.h
index ab948f55ed9..5749f518136 100644
--- a/be/src/olap/merger.h
+++ b/be/src/olap/merger.h
@@ -17,15 +17,12 @@
#pragma once
-#include <stdint.h>
-
#include <vector>
#include "common/status.h"
#include "io/io_common.h"
-#include "olap/rowset/rowset_reader.h"
-#include "olap/tablet.h"
-#include "olap/tablet_schema.h"
+#include "olap/rowset/rowset_fwd.h"
+#include "olap/tablet_fwd.h"
namespace doris {
class KeyBoundsPB;
@@ -55,22 +52,21 @@ public:
// return OK and set statistics into `*stats_output`.
// return others on error
- static Status vmerge_rowsets(TabletSharedPtr tablet, ReaderType reader_type,
- TabletSchemaSPtr cur_tablet_schema,
+ static Status vmerge_rowsets(BaseTabletSPtr tablet, ReaderType reader_type,
+ const TabletSchema& cur_tablet_schema,
const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,
RowsetWriter* dst_rowset_writer, Statistics* stats_output);
static Status vertical_merge_rowsets(
- TabletSharedPtr tablet, ReaderType reader_type, TabletSchemaSPtr tablet_schema,
+ BaseTabletSPtr tablet, ReaderType reader_type, const TabletSchema& tablet_schema,
const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,
RowsetWriter* dst_rowset_writer, int64_t max_rows_per_segment,
Statistics* stats_output);
-public:
// for vertical compaction
- static void vertical_split_columns(TabletSchemaSPtr tablet_schema,
+ static void vertical_split_columns(const TabletSchema& tablet_schema,
std::vector<std::vector<uint32_t>>* column_groups);
static Status vertical_compact_one_group(
- TabletSharedPtr tablet, ReaderType reader_type, TabletSchemaSPtr tablet_schema,
+ BaseTabletSPtr tablet, ReaderType reader_type, const TabletSchema& tablet_schema,
bool is_key, const std::vector<uint32_t>& column_group,
vectorized::RowSourcesBuffer* row_source_buf,
const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,
@@ -78,20 +74,14 @@ public:
std::vector<uint32_t> key_group_cluster_key_idxes);
// for segcompaction
- static Status vertical_compact_one_group(TabletSharedPtr tablet, ReaderType reader_type,
- TabletSchemaSPtr tablet_schema, bool is_key,
+ static Status vertical_compact_one_group(int64_t tablet_id, ReaderType reader_type,
+ const TabletSchema& tablet_schema, bool is_key,
const std::vector<uint32_t>& column_group,
vectorized::RowSourcesBuffer* row_source_buf,
vectorized::VerticalBlockReader& src_block_reader,
segment_v2::SegmentWriter& dst_segment_writer,
int64_t max_rows_per_segment, Statistics* stats_output,
uint64_t* index_size, KeyBoundsPB& key_bounds);
-
- // for mow with cluster key table, the key group also contains cluster key columns.
- // the `key_group_cluster_key_idxes` marks the positions of cluster key columns in key group.
- static void _generate_key_group_cluster_key_idxes(
- TabletSchemaSPtr tablet_schema, std::vector<std::vector<uint32_t>>& column_groups,
- std::vector<uint32_t>& key_group_cluster_key_idxes);
};
} // namespace doris
diff --git a/be/src/olap/olap_server.cpp b/be/src/olap/olap_server.cpp
index 00645b47339..97f0aab96b6 100644
--- a/be/src/olap/olap_server.cpp
+++ b/be/src/olap/olap_server.cpp
@@ -742,7 +742,7 @@ Status StorageEngine::_submit_single_replica_compaction_task(TabletSharedPtr tab
"compaction task has already been submitted, tablet_id={}", tablet->tablet_id());
}
- auto compaction = std::make_shared<SingleReplicaCompaction>(tablet, compaction_type);
+ auto compaction = std::make_shared<SingleReplicaCompaction>(*this, tablet, compaction_type);
auto st = compaction->prepare_compact();
auto clean_single_replica_compaction = [tablet, this]() {
@@ -947,7 +947,7 @@ Status StorageEngine::_submit_compaction_task(TabletSharedPtr tablet,
"compaction task has already been submitted, tablet_id={}, compaction_type={}.",
tablet->tablet_id(), compaction_type);
}
- std::shared_ptr<Compaction> compaction;
+ std::shared_ptr<CompactionMixin> compaction;
int64_t permits = 0;
Status st = Tablet::prepare_compaction_and_calculate_permits(compaction_type, tablet,
compaction, permits);
@@ -1361,27 +1361,37 @@ void StorageEngine::_cold_data_compaction_producer_callback() {
for (auto& [tablet, score] : tablet_to_compact) {
LOG(INFO) << "submit cold data compaction. tablet_id=" << tablet->tablet_id()
<< " score=" << score;
- static_cast<void>(
- _cold_data_compaction_thread_pool->submit_func([&, t = std::move(tablet)]() {
- auto compaction = std::make_shared<ColdDataCompaction>(t);
+ static_cast<void>(_cold_data_compaction_thread_pool->submit_func(
+ [&, t = std::move(tablet), this]() {
+ auto compaction = std::make_shared<ColdDataCompaction>(*this, t);
{
std::lock_guard lock(tablet_submitted_mtx);
tablet_submitted.insert(t->tablet_id());
}
+ Defer defer {[&] {
+ std::lock_guard lock(tablet_submitted_mtx);
+ tablet_submitted.erase(t->tablet_id());
+ }};
std::unique_lock cold_compaction_lock(t->get_cold_compaction_lock(),
std::try_to_lock);
if (!cold_compaction_lock.owns_lock()) {
LOG(WARNING) << "try cold_compaction_lock failed, tablet_id="
<< t->tablet_id();
+ return;
}
- auto st = compaction->compact();
- {
- std::lock_guard lock(tablet_submitted_mtx);
- tablet_submitted.erase(t->tablet_id());
+
+ auto st = compaction->prepare_compact();
+ if (!st.ok()) {
+ LOG(WARNING) << "failed to prepare cold data compaction. tablet_id="
+ << t->tablet_id() << " err=" << st;
+ return;
}
+
+ st = compaction->execute_compact();
if (!st.ok()) {
- LOG(WARNING) << "failed to do cold data compaction. tablet_id="
+ LOG(WARNING) << "failed to execute cold data compaction. tablet_id="
<< t->tablet_id() << " err=" << st;
+ return;
}
}));
}
diff --git a/be/src/olap/rowset/rowset.cpp b/be/src/olap/rowset/rowset.cpp
index d09fce730e5..bc946b3e06e 100644
--- a/be/src/olap/rowset/rowset.cpp
+++ b/be/src/olap/rowset/rowset.cpp
@@ -101,4 +101,20 @@ std::string Rowset::get_rowset_info_str() {
rowset_id().to_string(), disk_size);
}
+Status check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets) {
+ if (rowsets.size() < 2) {
+ return Status::OK();
+ }
+ auto prev = rowsets.begin();
+ for (auto it = rowsets.begin() + 1; it != rowsets.end(); ++it) {
+ if ((*prev)->end_version() + 1 != (*it)->start_version()) {
+ return Status::InternalError("versions are not continuity: prev={} cur={}",
+ (*prev)->version().to_string(),
+ (*it)->version().to_string());
+ }
+ prev = it;
+ }
+ return Status::OK();
+}
+
} // namespace doris
diff --git a/be/src/olap/rowset/rowset.h b/be/src/olap/rowset/rowset.h
index 4e0b781f4a2..d9da9907405 100644
--- a/be/src/olap/rowset/rowset.h
+++ b/be/src/olap/rowset/rowset.h
@@ -342,4 +342,7 @@ protected:
std::set<int32_t> skip_index_compaction;
};
+// `rs_metas` MUST already be sorted by `RowsetMeta::comparator`
+Status check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets);
+
} // namespace doris
diff --git a/be/src/olap/rowset/segcompaction.cpp b/be/src/olap/rowset/segcompaction.cpp
index e252870ae76..81a73b99ebb 100644
--- a/be/src/olap/rowset/segcompaction.cpp
+++ b/be/src/olap/rowset/segcompaction.cpp
@@ -220,7 +220,7 @@ Status SegcompactionWorker::_do_compact_segments(SegCompactionCandidatesSharedPt
auto tablet = std::static_pointer_cast<Tablet>(ctx.tablet);
std::vector<std::vector<uint32_t>> column_groups;
- Merger::vertical_split_columns(ctx.tablet_schema, &column_groups);
+ Merger::vertical_split_columns(*ctx.tablet_schema, &column_groups);
vectorized::RowSourcesBuffer row_sources_buf(tablet->tablet_id(), tablet->tablet_path(),
ReaderType::READER_SEGMENT_COMPACTION);
@@ -247,9 +247,9 @@ Status SegcompactionWorker::_do_compact_segments(SegCompactionCandidatesSharedPt
Merger::Statistics merger_stats;
RETURN_IF_ERROR(Merger::vertical_compact_one_group(
- tablet, ReaderType::READER_SEGMENT_COMPACTION, ctx.tablet_schema, is_key,
- column_ids, &row_sources_buf, *reader, *writer, INT_MAX, &merger_stats, &index_size,
- key_bounds));
+ tablet->tablet_id(), ReaderType::READER_SEGMENT_COMPACTION, *ctx.tablet_schema,
+ is_key, column_ids, &row_sources_buf, *reader, *writer, INT_MAX, &merger_stats,
+ &index_size, key_bounds));
total_index_size += index_size;
if (is_key) {
RETURN_IF_ERROR(row_sources_buf.flush());
diff --git a/be/src/olap/schema_change.cpp b/be/src/olap/schema_change.cpp
index 6e432b24096..d3005c7cb3a 100644
--- a/be/src/olap/schema_change.cpp
+++ b/be/src/olap/schema_change.cpp
@@ -646,7 +646,7 @@ Status VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
Merger::Statistics stats;
RETURN_IF_ERROR(Merger::vmerge_rowsets(new_tablet, ReaderType::READER_ALTER_TABLE,
- new_tablet_schema, rs_readers, rowset_writer, &stats));
+ *new_tablet_schema, rs_readers, rowset_writer, &stats));
_add_merged_rows(stats.merged_rows);
_add_filtered_rows(stats.filtered_rows);
return Status::OK();
diff --git a/be/src/olap/single_replica_compaction.cpp b/be/src/olap/single_replica_compaction.cpp
index 5d22ae5590b..fba76765b0a 100644
--- a/be/src/olap/single_replica_compaction.cpp
+++ b/be/src/olap/single_replica_compaction.cpp
@@ -44,26 +44,28 @@
namespace doris {
using namespace ErrorCode;
-SingleReplicaCompaction::SingleReplicaCompaction(const TabletSharedPtr& tablet,
- const CompactionType& compaction_type)
- : Compaction(tablet, "SingleReplicaCompaction:" + std::to_string(tablet->tablet_id())),
+SingleReplicaCompaction::SingleReplicaCompaction(StorageEngine& engine,
+ const TabletSharedPtr& tablet,
+ CompactionType compaction_type)
+ : CompactionMixin(engine, tablet,
+ "SingleReplicaCompaction:" + std::to_string(tablet->tablet_id())),
_compaction_type(compaction_type) {}
SingleReplicaCompaction::~SingleReplicaCompaction() = default;
Status SingleReplicaCompaction::prepare_compact() {
VLOG_CRITICAL << _tablet->tablet_id() << " prepare single replcia compaction and pick rowsets!";
- if (!_tablet->init_succeeded()) {
+ if (!tablet()->init_succeeded()) {
return Status::Error<CUMULATIVE_INVALID_PARAMETERS, false>("_tablet init failed");
}
- std::unique_lock<std::mutex> lock_cumu(_tablet->get_cumulative_compaction_lock(),
+ std::unique_lock<std::mutex> lock_cumu(tablet()->get_cumulative_compaction_lock(),
std::try_to_lock);
if (!lock_cumu.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>(
"The tablet is under cumulative compaction. tablet={}", _tablet->tablet_id());
}
- std::unique_lock<std::mutex> lock_base(_tablet->get_base_compaction_lock(), std::try_to_lock);
+ std::unique_lock<std::mutex> lock_base(tablet()->get_base_compaction_lock(), std::try_to_lock);
if (!lock_base.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>(
"another base compaction is running. tablet={}", _tablet->tablet_id());
@@ -79,7 +81,7 @@ Status SingleReplicaCompaction::prepare_compact() {
}
Status SingleReplicaCompaction::pick_rowsets_to_compact() {
- auto candidate_rowsets = _tablet->pick_candidate_rowsets_to_single_replica_compaction();
+ auto candidate_rowsets = tablet()->pick_candidate_rowsets_to_single_replica_compaction();
if (candidate_rowsets.empty()) {
return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>("candidate_rowsets is empty");
}
@@ -91,15 +93,15 @@ Status SingleReplicaCompaction::pick_rowsets_to_compact() {
return Status::OK();
}
-Status SingleReplicaCompaction::execute_compact_impl() {
- std::unique_lock<std::mutex> lock_cumu(_tablet->get_cumulative_compaction_lock(),
+Status SingleReplicaCompaction::execute_compact() {
+ std::unique_lock<std::mutex> lock_cumu(tablet()->get_cumulative_compaction_lock(),
std::try_to_lock);
if (!lock_cumu.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>(
"The tablet is under cumulative compaction. tablet={}", _tablet->tablet_id());
}
- std::unique_lock<std::mutex> lock_base(_tablet->get_base_compaction_lock(), std::try_to_lock);
+ std::unique_lock<std::mutex> lock_base(tablet()->get_base_compaction_lock(), std::try_to_lock);
if (!lock_base.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>(
"another base compaction is running. tablet={}", _tablet->tablet_id());
@@ -117,9 +119,9 @@ Status SingleReplicaCompaction::execute_compact_impl() {
}
Status SingleReplicaCompaction::_do_single_replica_compaction() {
- _tablet->data_dir()->disks_compaction_num_increment(1);
+ tablet()->data_dir()->disks_compaction_num_increment(1);
Status st = _do_single_replica_compaction_impl();
- _tablet->data_dir()->disks_compaction_num_increment(-1);
+ tablet()->data_dir()->disks_compaction_num_increment(-1);
return st;
}
@@ -128,8 +130,7 @@ Status SingleReplicaCompaction::_do_single_replica_compaction_impl() {
TReplicaInfo addr;
std::string token;
// 1. get peer replica info
- if (!ExecEnv::GetInstance()->storage_engine().to_local().get_peer_replica_info(
- _tablet->tablet_id(), &addr, &token)) {
+ if (!_engine.get_peer_replica_info(_tablet->tablet_id(), &addr, &token)) {
LOG(WARNING) << _tablet->tablet_id() << " tablet don't have peer replica";
return Status::Aborted("tablet don't have peer replica");
}
@@ -152,11 +153,11 @@ Status SingleReplicaCompaction::_do_single_replica_compaction_impl() {
// 6. update last success compaction time
if (compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION) {
- _tablet->set_last_cumu_compaction_success_time(UnixMillis());
+ tablet()->set_last_cumu_compaction_success_time(UnixMillis());
} else if (compaction_type() == ReaderType::READER_BASE_COMPACTION) {
- _tablet->set_last_base_compaction_success_time(UnixMillis());
+ tablet()->set_last_base_compaction_success_time(UnixMillis());
} else if (compaction_type() == ReaderType::READER_FULL_COMPACTION) {
- _tablet->set_last_full_compaction_success_time(UnixMillis());
+ tablet()->set_last_full_compaction_success_time(UnixMillis());
}
int64_t current_max_version = -1;
@@ -208,8 +209,7 @@ Status SingleReplicaCompaction::_get_rowset_verisons_from_peer(
return Status::Cancelled("no peer version");
}
for (int i = 0; i < response.versions_size(); ++i) {
- (*peer_versions)
- .emplace_back(Version(response.versions(i).first(), response.versions(i).second()));
+ (*peer_versions).emplace_back(response.versions(i).first(), response.versions(i).second());
}
return Status::OK();
}
@@ -217,7 +217,7 @@ Status SingleReplicaCompaction::_get_rowset_verisons_from_peer(
bool SingleReplicaCompaction::_find_rowset_to_fetch(const std::vector<Version>& peer_versions,
Version* proper_version) {
// get local versions
- std::vector<Version> local_versions = _tablet->get_all_versions();
+ std::vector<Version> local_versions = tablet()->get_all_versions();
std::sort(local_versions.begin(), local_versions.end(),
[](const Version& left, const Version& right) { return left.first < right.first; });
for (const auto& v : local_versions) {
@@ -293,7 +293,7 @@ Status SingleReplicaCompaction::_fetch_rowset(const TReplicaInfo& addr, const st
const Version& rowset_version) {
LOG(INFO) << "begin to fetch compaction result, tablet_id=" << _tablet->tablet_id()
<< ", addr=" << addr.host << ", version=" << rowset_version;
- std::shared_lock migration_rlock(_tablet->get_migration_lock(), std::try_to_lock);
+ std::shared_lock migration_rlock(tablet()->get_migration_lock(), std::try_to_lock);
if (!migration_rlock.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED, false>("got migration_rlock failed. tablet={}",
_tablet->tablet_id());
@@ -327,11 +327,10 @@ Status SingleReplicaCompaction::_fetch_rowset(const TReplicaInfo& addr, const st
<< _tablet->tablet_id() << "/" << _tablet->schema_hash() << "/";
remote_url_prefix = ss.str();
}
- RETURN_IF_ERROR(_download_files(_tablet->data_dir(), remote_url_prefix, local_path));
- _pending_rs_guards = DORIS_TRY(
- ExecEnv::GetInstance()->storage_engine().to_local().snapshot_mgr()->convert_rowset_ids(
- local_path, _tablet->tablet_id(), _tablet->replica_id(),
- _tablet->partition_id(), _tablet->schema_hash()));
+ RETURN_IF_ERROR(_download_files(tablet()->data_dir(), remote_url_prefix, local_path));
+ _pending_rs_guards = DORIS_TRY(_engine.snapshot_mgr()->convert_rowset_ids(
+ local_path, _tablet->tablet_id(), tablet()->replica_id(), _tablet->partition_id(),
+ _tablet->schema_hash()));
// 4: finish_clone: create output_rowset and link file
return _finish_clone(local_data_path, rowset_version);
}
@@ -520,7 +519,7 @@ Status SingleReplicaCompaction::_finish_clone(const string& clone_dir,
LOG(WARNING) << "version not found in cloned tablet meta when do single compaction";
return Status::InternalError("version not found in cloned tablet meta");
}
- res = _tablet->create_rowset(output_rs_meta, &_output_rowset);
+ res = tablet()->create_rowset(output_rs_meta, &_output_rowset);
if (!res.ok()) {
LOG(WARNING) << "fail to init rowset. version=" << output_version;
return res;
diff --git a/be/src/olap/single_replica_compaction.h b/be/src/olap/single_replica_compaction.h
index fd35da12645..068f177a8d1 100644
--- a/be/src/olap/single_replica_compaction.h
+++ b/be/src/olap/single_replica_compaction.h
@@ -22,23 +22,24 @@
#include "common/status.h"
#include "olap/compaction.h"
#include "olap/rowset/pending_rowset_helper.h"
-#include "olap/rowset/rowset.h"
-#include "olap/tablet.h"
namespace doris {
+class DataDir;
+
// SingleReplicaCompaction is used to fetch peer replica compaction result.
-class SingleReplicaCompaction : public Compaction {
+class SingleReplicaCompaction final : public CompactionMixin {
public:
- SingleReplicaCompaction(const TabletSharedPtr& tablet, const CompactionType& compaction_type);
+ SingleReplicaCompaction(StorageEngine& engine, const TabletSharedPtr& tablet,
+ CompactionType compaction_type);
~SingleReplicaCompaction() override;
Status prepare_compact() override;
- Status execute_compact_impl() override;
+ Status execute_compact() override;
protected:
- Status pick_rowsets_to_compact() override;
- std::string compaction_name() const override { return "single replica compaction"; }
+ Status pick_rowsets_to_compact();
+ std::string_view compaction_name() const override { return "single replica compaction"; }
ReaderType compaction_type() const override {
return (_compaction_type == CompactionType::CUMULATIVE_COMPACTION)
? ReaderType::READER_CUMULATIVE_COMPACTION
@@ -59,13 +60,10 @@ private:
Status _download_files(DataDir* data_dir, const std::string& remote_url_prefix,
const std::string& local_path);
Status _release_snapshot(const std::string& ip, int port, const std::string& snapshot_path);
- Status _finish_clone(const string& clone_dir, const Version& version);
+ Status _finish_clone(const std::string& clone_dir, const Version& version);
std::string _mask_token(const std::string& str);
CompactionType _compaction_type;
- DISALLOW_COPY_AND_ASSIGN(SingleReplicaCompaction);
-
-private:
std::vector<PendingRowsetGuard> _pending_rs_guards;
};
diff --git a/be/src/olap/snapshot_manager.cpp b/be/src/olap/snapshot_manager.cpp
index f8235a83af5..c5bf5a785db 100644
--- a/be/src/olap/snapshot_manager.cpp
+++ b/be/src/olap/snapshot_manager.cpp
@@ -337,23 +337,6 @@ Status SnapshotManager::_link_index_and_data_files(
return res;
}
-// `rs_metas` MUST already be sorted by `RowsetMeta::comparator`
-Status check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets) {
- if (rowsets.size() < 2) {
- return Status::OK();
- }
- auto prev = rowsets.begin();
- for (auto it = rowsets.begin() + 1; it != rowsets.end(); ++it) {
- if ((*prev)->end_version() + 1 != (*it)->start_version()) {
- return Status::InternalError("versions are not continuity: prev={} cur={}",
- (*prev)->version().to_string(),
- (*it)->version().to_string());
- }
- prev = it;
- }
- return Status::OK();
-}
-
Status SnapshotManager::_create_snapshot_files(const TabletSharedPtr& ref_tablet,
const TSnapshotRequest& request,
string* snapshot_path,
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
index 9730c149f74..71e0886ad1e 100644
--- a/be/src/olap/tablet.cpp
+++ b/be/src/olap/tablet.cpp
@@ -1589,15 +1589,14 @@ void Tablet::build_tablet_report_info(TTabletInfo* tablet_info,
}
}
-Status Tablet::prepare_compaction_and_calculate_permits(CompactionType compaction_type,
- const TabletSharedPtr& tablet,
- std::shared_ptr<Compaction>& compaction,
- int64_t& permits) {
+Status Tablet::prepare_compaction_and_calculate_permits(
+ CompactionType compaction_type, const TabletSharedPtr& tablet,
+ std::shared_ptr<CompactionMixin>& compaction, int64_t& permits) {
if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
MonotonicStopWatch watch;
watch.start();
- compaction = std::make_shared<CumulativeCompaction>(tablet);
+ compaction = std::make_shared<CumulativeCompaction>(tablet->_engine, tablet);
DorisMetrics::instance()->cumulative_compaction_request_total->increment(1);
Status res = compaction->prepare_compact();
if (!config::disable_compaction_trace_log &&
@@ -1625,7 +1624,7 @@ Status Tablet::prepare_compaction_and_calculate_permits(CompactionType compactio
MonotonicStopWatch watch;
watch.start();
- compaction = std::make_shared<BaseCompaction>(tablet);
+ compaction = std::make_shared<BaseCompaction>(tablet->_engine, tablet);
DorisMetrics::instance()->base_compaction_request_total->increment(1);
Status res = compaction->prepare_compact();
if (!config::disable_compaction_trace_log &&
@@ -1653,7 +1652,7 @@ Status Tablet::prepare_compaction_and_calculate_permits(CompactionType compactio
} else {
DCHECK_EQ(compaction_type, CompactionType::FULL_COMPACTION);
- compaction = std::make_shared<FullCompaction>(tablet);
+ compaction = std::make_shared<FullCompaction>(tablet->_engine, tablet);
Status res = compaction->prepare_compact();
if (!res.ok()) {
tablet->set_last_full_compaction_failure_time(UnixMillis());
@@ -1668,10 +1667,7 @@ Status Tablet::prepare_compaction_and_calculate_permits(CompactionType compactio
}
}
- permits = 0;
- for (auto&& rowset : compaction->input_rowsets()) {
- permits += rowset->rowset_meta()->get_compaction_score();
- }
+ permits = compaction->get_compaction_permits();
return Status::OK();
}
@@ -1700,7 +1696,7 @@ std::vector<Version> Tablet::get_all_versions() {
return local_versions;
}
-void Tablet::execute_compaction(Compaction& compaction) {
+void Tablet::execute_compaction(CompactionMixin& compaction) {
signal::tablet_id = tablet_id();
MonotonicStopWatch watch;
diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
index 7b993d1af32..551b2422381 100644
--- a/be/src/olap/tablet.h
+++ b/be/src/olap/tablet.h
@@ -59,7 +59,7 @@ namespace doris {
class Tablet;
class CumulativeCompactionPolicy;
-class Compaction;
+class CompactionMixin;
class SingleReplicaCompaction;
class RowsetWriter;
struct RowsetWriterContext;
@@ -275,12 +275,11 @@ public:
// return a json string to show the compaction status of this tablet
void get_compaction_status(std::string* json_result);
- static Status prepare_compaction_and_calculate_permits(CompactionType compaction_type,
- const TabletSharedPtr& tablet,
- std::shared_ptr<Compaction>& compaction,
- int64_t& permits);
+ static Status prepare_compaction_and_calculate_permits(
+ CompactionType compaction_type, const TabletSharedPtr& tablet,
+ std::shared_ptr<CompactionMixin>& compaction, int64_t& permits);
- void execute_compaction(Compaction& compaction);
+ void execute_compaction(CompactionMixin& compaction);
void execute_single_replica_compaction(SingleReplicaCompaction& compaction);
void set_cumulative_compaction_policy(
diff --git a/be/src/olap/txn_manager.cpp b/be/src/olap/txn_manager.cpp
index e3d22b6bc05..739e85dd0fd 100644
--- a/be/src/olap/txn_manager.cpp
+++ b/be/src/olap/txn_manager.cpp
@@ -730,11 +730,11 @@ void TxnManager::get_all_related_tablets(std::set<TabletInfo>* tablet_infos) {
}
void TxnManager::get_all_commit_tablet_txn_info_by_tablet(
- const TabletSharedPtr& tablet, CommitTabletTxnInfoVec* commit_tablet_txn_info_vec) {
+ const Tablet& tablet, CommitTabletTxnInfoVec* commit_tablet_txn_info_vec) {
for (int32_t i = 0; i < _txn_map_shard_size; i++) {
std::shared_lock txn_rdlock(_txn_map_locks[i]);
for (const auto& [txn_key, load_info_map] : _txn_tablet_maps[i]) {
- auto tablet_load_it = load_info_map.find(tablet->get_tablet_info());
+ auto tablet_load_it = load_info_map.find(tablet.get_tablet_info());
if (tablet_load_it != load_info_map.end()) {
const auto& [_, load_info] = *tablet_load_it;
const auto& rowset = load_info->rowset;
diff --git a/be/src/olap/txn_manager.h b/be/src/olap/txn_manager.h
index e33958a66aa..67a62d557aa 100644
--- a/be/src/olap/txn_manager.h
+++ b/be/src/olap/txn_manager.h
@@ -208,7 +208,7 @@ public:
const RowsetIdUnorderedSet& rowset_ids,
std::shared_ptr<PartialUpdateInfo> partial_update_info);
void get_all_commit_tablet_txn_info_by_tablet(
- const TabletSharedPtr& tablet, CommitTabletTxnInfoVec* commit_tablet_txn_info_vec);
+ const Tablet& tablet, CommitTabletTxnInfoVec* commit_tablet_txn_info_vec);
int64_t get_txn_by_tablet_version(int64_t tablet_id, int64_t version);
void update_tablet_version_txn(int64_t tablet_id, int64_t version, int64_t txn_id);
diff --git a/be/src/pipeline/task_scheduler.cpp b/be/src/pipeline/task_scheduler.cpp
index 77cdddfb4a7..fdab28a5f40 100644
--- a/be/src/pipeline/task_scheduler.cpp
+++ b/be/src/pipeline/task_scheduler.cpp
@@ -307,7 +307,8 @@ void TaskScheduler::_do_work(size_t index) {
// LOG(WARNING)<< "task:\n"<<task->debug_string();
// exec failed，cancel all fragment instance
- fragment_ctx->cancel(PPlanFragmentCancelReason::INTERNAL_ERROR, status.msg());
+ fragment_ctx->cancel(PPlanFragmentCancelReason::INTERNAL_ERROR,
+ std::string(status.msg()));
_close_task(task, PipelineTaskState::CANCELED, status);
continue;
}
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index 3a9a7c2a88d..9857233f689 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -422,7 +422,7 @@ void FragmentMgr::coordinator_callback(const ReportStatusRequest& req) {
print_id(req.fragment_instance_id), rpc_status.to_string());
// we need to cancel the execution of this fragment
static_cast<void>(req.update_fn(rpc_status));
- req.cancel_fn(PPlanFragmentCancelReason::INTERNAL_ERROR, rpc_status.msg());
+ req.cancel_fn(PPlanFragmentCancelReason::INTERNAL_ERROR, std::string(rpc_status.msg()));
}
}
diff --git a/be/src/vec/core/sort_cursor.h b/be/src/vec/core/sort_cursor.h
index 9e40f40cf18..a7be72eb9ab 100644
--- a/be/src/vec/core/sort_cursor.h
+++ b/be/src/vec/core/sort_cursor.h
@@ -243,7 +243,7 @@ struct BlockSupplierSortCursorImpl : public MergeSortCursorImpl {
MergeSortCursorImpl::reset(_block);
return status.ok();
} else if (!status.ok()) {
- throw std::runtime_error(status.msg());
+ throw std::runtime_error(std::string(status.msg()));
}
return false;
}
diff --git a/be/test/olap/compaction_delete_bitmap_calculator_test.cpp b/be/test/olap/compaction_delete_bitmap_calculator_test.cpp
index f1b5865c817..f20609a6d03 100644
--- a/be/test/olap/compaction_delete_bitmap_calculator_test.cpp
+++ b/be/test/olap/compaction_delete_bitmap_calculator_test.cpp
@@ -273,7 +273,7 @@ TEST_F(CompactionDeleteBitmapCalculatorTest, test) {
CommitTabletTxnInfoVec commit_tablet_txn_info_vec {};
engine_ref->txn_manager()->get_all_commit_tablet_txn_info_by_tablet(
- _tablet, &commit_tablet_txn_info_vec);
+ *_tablet, &commit_tablet_txn_info_vec);
EXPECT_EQ(commit_tablet_txn_info_vec.size(), 1);
}
} // namespace doris
diff --git a/be/test/olap/cumulative_compaction_policy_test.cpp b/be/test/olap/cumulative_compaction_policy_test.cpp
index 08d479727e1..e4775031c28 100644
--- a/be/test/olap/cumulative_compaction_policy_test.cpp
+++ b/be/test/olap/cumulative_compaction_policy_test.cpp
@@ -738,42 +738,6 @@ TEST_F(TestSizeBasedCumulativeCompactionPolicy, _level_size) {
EXPECT_EQ(1 << 19, policy->_level_size((1 << 20) - 100));
}
-TEST_F(TestSizeBasedCumulativeCompactionPolicy, _pick_missing_version_cumulative_compaction) {
- std::vector<RowsetMetaSharedPtr> rs_metas;
- init_rs_meta_missing_version(&rs_metas);
-
- for (auto& rowset : rs_metas) {
- static_cast<void>(_tablet_meta->add_rs_meta(rowset));
- }
-
- TabletSharedPtr _tablet(
- new Tablet(_engine, _tablet_meta, nullptr, CUMULATIVE_SIZE_BASED_POLICY));
- static_cast<void>(_tablet->init());
- ;
-
- // has miss version
- std::vector<RowsetSharedPtr> rowsets;
- rowsets.push_back(_tablet->get_rowset_by_version({0, 0}));
- rowsets.push_back(_tablet->get_rowset_by_version({1, 1}));
- rowsets.push_back(_tablet->get_rowset_by_version({2, 2}));
- rowsets.push_back(_tablet->get_rowset_by_version({4, 4}));
- CumulativeCompaction compaction(_tablet);
- static_cast<void>(compaction.find_longest_consecutive_version(&rowsets, nullptr));
- EXPECT_EQ(3, rowsets.size());
- EXPECT_EQ(2, rowsets[2]->end_version());
-
- // no miss version
- std::vector<RowsetSharedPtr> rowsets2;
- rowsets2.push_back(_tablet->get_rowset_by_version({0, 0}));
- static_cast<void>(compaction.find_longest_consecutive_version(&rowsets2, nullptr));
- EXPECT_EQ(1, rowsets2.size());
- EXPECT_EQ(0, rowsets[0]->end_version());
-
- // no version
- std::vector<RowsetSharedPtr> rowsets3;
- static_cast<void>(compaction.find_longest_consecutive_version(&rowsets3, nullptr));
- EXPECT_EQ(0, rowsets3.size());
-}
} // namespace doris
// @brief Test Stub
diff --git a/be/test/olap/cumulative_compaction_time_series_policy_test.cpp b/be/test/olap/cumulative_compaction_time_series_policy_test.cpp
index 1390f1deb14..98169426d20 100644
--- a/be/test/olap/cumulative_compaction_time_series_policy_test.cpp
+++ b/be/test/olap/cumulative_compaction_time_series_policy_test.cpp
@@ -562,41 +562,6 @@ TEST_F(TestTimeSeriesCumulativeCompactionPolicy, pick_input_rowsets_delete) {
EXPECT_EQ(5, last_delete_version.second);
}
-TEST_F(TestTimeSeriesCumulativeCompactionPolicy, _pick_missing_version_cumulative_compaction) {
- std::vector<RowsetMetaSharedPtr> rs_metas;
- init_rs_meta_missing_version(&rs_metas);
-
- for (auto& rowset : rs_metas) {
- static_cast<void>(_tablet_meta->add_rs_meta(rowset));
- }
-
- TabletSharedPtr _tablet(
- new Tablet(_engine, _tablet_meta, nullptr, CUMULATIVE_TIME_SERIES_POLICY));
- static_cast<void>(_tablet->init());
-
- // has miss version
- std::vector<RowsetSharedPtr> rowsets;
- rowsets.push_back(_tablet->get_rowset_by_version({0, 0}));
- rowsets.push_back(_tablet->get_rowset_by_version({1, 1}));
- rowsets.push_back(_tablet->get_rowset_by_version({2, 2}));
- rowsets.push_back(_tablet->get_rowset_by_version({4, 4}));
- CumulativeCompaction compaction(_tablet);
- static_cast<void>(compaction.find_longest_consecutive_version(&rowsets, nullptr));
- EXPECT_EQ(3, rowsets.size());
- EXPECT_EQ(2, rowsets[2]->end_version());
-
- // no miss version
- std::vector<RowsetSharedPtr> rowsets2;
- rowsets2.push_back(_tablet->get_rowset_by_version({0, 0}));
- static_cast<void>(compaction.find_longest_consecutive_version(&rowsets2, nullptr));
- EXPECT_EQ(1, rowsets2.size());
- EXPECT_EQ(0, rowsets[0]->end_version());
-
- // no version
- std::vector<RowsetSharedPtr> rowsets3;
- static_cast<void>(compaction.find_longest_consecutive_version(&rowsets3, nullptr));
- EXPECT_EQ(0, rowsets3.size());
-}
} // namespace doris
// @brief Test Stub
diff --git a/be/test/olap/delta_writer_test.cpp b/be/test/olap/delta_writer_test.cpp
index 8e4ea186134..f33c3f31bf8 100644
--- a/be/test/olap/delta_writer_test.cpp
+++ b/be/test/olap/delta_writer_test.cpp
@@ -86,7 +86,6 @@ static void set_up() {
engine_ref = engine.get();
Status s = engine->open();
ASSERT_TRUE(s.ok()) << s;
- s = engine->start_bg_threads();
ASSERT_TRUE(s.ok()) << s;
ExecEnv* exec_env = doris::ExecEnv::GetInstance();
diff --git a/be/test/olap/engine_storage_migration_task_test.cpp b/be/test/olap/engine_storage_migration_task_test.cpp
index e74cfb54be0..6d87413e5d3 100644
--- a/be/test/olap/engine_storage_migration_task_test.cpp
+++ b/be/test/olap/engine_storage_migration_task_test.cpp
@@ -87,8 +87,6 @@ static void set_up() {
engine_ref = engine.get();
Status s = engine->open();
EXPECT_TRUE(s.ok()) << s;
- s = engine->start_bg_threads();
- EXPECT_TRUE(s.ok()) << s;
ExecEnv* exec_env = doris::ExecEnv::GetInstance();
exec_env->set_memtable_memory_limiter(new MemTableMemoryLimiter());
exec_env->set_storage_engine(std::move(engine));
diff --git a/be/test/olap/memtable_memory_limiter_test.cpp b/be/test/olap/memtable_memory_limiter_test.cpp
index bb29f91f750..14eae847163 100644
--- a/be/test/olap/memtable_memory_limiter_test.cpp
+++ b/be/test/olap/memtable_memory_limiter_test.cpp
@@ -100,7 +100,6 @@ protected:
// So we must do this before storage engine's other operation.
exec_env->set_storage_engine(std::move(engine));
exec_env->set_memtable_memory_limiter(new MemTableMemoryLimiter());
- static_cast<void>(_engine_ref->start_bg_threads());
}
void TearDown() override {
diff --git a/be/test/olap/ordered_data_compaction_test.cpp b/be/test/olap/ordered_data_compaction_test.cpp
index cb6f9518aa0..08df161153e 100644
--- a/be/test/olap/ordered_data_compaction_test.cpp
+++ b/be/test/olap/ordered_data_compaction_test.cpp
@@ -432,14 +432,11 @@ TEST_F(OrderedDataCompactionTest, test_01) {
input_rowsets.push_back(rowset);
}
//auto end_version = input_rowsets.back()->end_version();
- CumulativeCompaction cu_compaction(tablet);
- cu_compaction.set_input_rowset(input_rowsets);
+ CumulativeCompaction cu_compaction(*engine_ref, tablet);
+ cu_compaction._input_rowsets = std::move(input_rowsets);
EXPECT_EQ(cu_compaction.handle_ordered_data_compaction(), true);
- for (int i = 0; i < 100; ++i) {
- LOG(INFO) << "stop";
- }
- RowsetSharedPtr out_rowset = cu_compaction.output_rowset();
+ auto& out_rowset = cu_compaction._output_rowset;
// create output rowset reader
RowsetReaderContext reader_context;
diff --git a/be/test/olap/rowid_conversion_test.cpp b/be/test/olap/rowid_conversion_test.cpp
index 3257fdf52dd..b87507228cf 100644
--- a/be/test/olap/rowid_conversion_test.cpp
+++ b/be/test/olap/rowid_conversion_test.cpp
@@ -349,10 +349,10 @@ protected:
Status s;
if (is_vertical_merger) {
s = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION,
- tablet_schema, input_rs_readers,
+ *tablet_schema, input_rs_readers,
output_rs_writer.get(), 10000000, &stats);
} else {
- s = Merger::vmerge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION, tablet_schema,
+ s = Merger::vmerge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION, *tablet_schema,
input_rs_readers, output_rs_writer.get(), &stats);
}
EXPECT_TRUE(s.ok());
diff --git a/be/test/runtime/load_stream_test.cpp b/be/test/runtime/load_stream_test.cpp
index 89535642436..1d9ff6b347c 100644
--- a/be/test/runtime/load_stream_test.cpp
+++ b/be/test/runtime/load_stream_test.cpp
@@ -603,8 +603,6 @@ public:
EXPECT_TRUE(io::global_local_filesystem()->create_directory(zTestDir).ok());
- static_cast<void>(engine_ref->start_bg_threads());
-
_load_stream_mgr = std::make_unique<LoadStreamMgr>(4, &_heavy_work_pool, &_light_work_pool);
_stream_service = new StreamService(_load_stream_mgr.get());
CHECK_EQ(0, _server->AddService(_stream_service, brpc::SERVER_OWNS_SERVICE));
diff --git a/be/test/vec/olap/vertical_compaction_test.cpp b/be/test/vec/olap/vertical_compaction_test.cpp
index e1dd1440e71..33051678d2b 100644
--- a/be/test/vec/olap/vertical_compaction_test.cpp
+++ b/be/test/vec/olap/vertical_compaction_test.cpp
@@ -488,8 +488,8 @@ TEST_F(VerticalCompactionTest, TestDupKeyVerticalMerge) {
RowIdConversion rowid_conversion;
stats.rowid_conversion = &rowid_conversion;
auto s = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION,
- tablet_schema, input_rs_readers, output_rs_writer.get(),
- 100, &stats);
+ *tablet_schema, input_rs_readers,
+ output_rs_writer.get(), 100, &stats);
ASSERT_TRUE(s.ok()) << s;
RowsetSharedPtr out_rowset;
EXPECT_EQ(Status::OK(), output_rs_writer->build(out_rowset));
@@ -596,8 +596,8 @@ TEST_F(VerticalCompactionTest, TestDupWithoutKeyVerticalMerge) {
RowIdConversion rowid_conversion;
stats.rowid_conversion = &rowid_conversion;
auto s = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION,
- tablet_schema, input_rs_readers, output_rs_writer.get(),
- 100, &stats);
+ *tablet_schema, input_rs_readers,
+ output_rs_writer.get(), 100, &stats);
ASSERT_TRUE(s.ok()) << s;
RowsetSharedPtr out_rowset;
EXPECT_EQ(Status::OK(), output_rs_writer->build(out_rowset));
@@ -704,8 +704,8 @@ TEST_F(VerticalCompactionTest, TestUniqueKeyVerticalMerge) {
RowIdConversion rowid_conversion;
stats.rowid_conversion = &rowid_conversion;
auto s = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION,
- tablet_schema, input_rs_readers, output_rs_writer.get(),
- 10000, &stats);
+ *tablet_schema, input_rs_readers,
+ output_rs_writer.get(), 10000, &stats);
EXPECT_TRUE(s.ok());
RowsetSharedPtr out_rowset;
EXPECT_EQ(Status::OK(), output_rs_writer->build(out_rowset));
@@ -813,7 +813,7 @@ TEST_F(VerticalCompactionTest, TestDupKeyVerticalMergeWithDelete) {
Merger::Statistics stats;
RowIdConversion rowid_conversion;
stats.rowid_conversion = &rowid_conversion;
- st = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION, tablet_schema,
+ st = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION, *tablet_schema,
input_rs_readers, output_rs_writer.get(), 100, &stats);
ASSERT_TRUE(st.ok()) << st;
RowsetSharedPtr out_rowset;
@@ -915,7 +915,7 @@ TEST_F(VerticalCompactionTest, TestDupWithoutKeyVerticalMergeWithDelete) {
Merger::Statistics stats;
RowIdConversion rowid_conversion;
stats.rowid_conversion = &rowid_conversion;
- st = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION, tablet_schema,
+ st = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION, *tablet_schema,
input_rs_readers, output_rs_writer.get(), 100, &stats);
ASSERT_TRUE(st.ok()) << st;
RowsetSharedPtr out_rowset;
@@ -1008,8 +1008,8 @@ TEST_F(VerticalCompactionTest, TestAggKeyVerticalMerge) {
RowIdConversion rowid_conversion;
stats.rowid_conversion = &rowid_conversion;
auto s = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION,
- tablet_schema, input_rs_readers, output_rs_writer.get(),
- 100, &stats);
+ *tablet_schema, input_rs_readers,
+ output_rs_writer.get(), 100, &stats);
EXPECT_TRUE(s.ok());
RowsetSharedPtr out_rowset;
EXPECT_EQ(Status::OK(), output_rs_writer->build(out_rowset));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org