This is an automated email from the ASF dual-hosted git repository.
mymeiyi 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 9f2491c7680 [fix](cloud) Correct cumulative point advancement when
enable parallel cumulative compaction (#65915)
9f2491c7680 is described below
commit 9f2491c76806e7e797b02b14cfd96bfeab4e130b
Author: meiyi <[email protected]>
AuthorDate: Tue Aug 18 14:38:27 2026 +0800
[fix](cloud) Correct cumulative point advancement when enable parallel
cumulative compaction (#65915)
When enable parallel cumulative compaction, the cu compaction with higher
version range may finish before the cu compaction with lower range.
For example, cu1: v10-v20, cu2: v21-v30. The cu2 may finish before cu1.
When cu2 finish, it may set cu point to v31, does not consider if cu1 finish or
fail.
This pr correct the cu point advancement.
related to #64619
---
be/src/cloud/cloud_base_compaction.cpp | 3 +
be/src/cloud/cloud_cumulative_compaction.cpp | 281 ++++++++++---
be/src/cloud/cloud_cumulative_compaction.h | 11 +-
be/src/cloud/cloud_index_change_compaction.cpp | 24 +-
be/src/storage/compaction/compaction.cpp | 20 +
be/src/storage/compaction/compaction.h | 3 +
be/test/cloud/cloud_compaction_test.cpp | 460 +++++++++++++++++++++
.../cloud_index_change_compaction_test.cpp | 20 +
cloud/src/meta-service/meta_service_job.cpp | 113 ++++-
cloud/test/meta_service_job_test.cpp | 348 +++++++++++++++-
10 files changed, 1185 insertions(+), 98 deletions(-)
diff --git a/be/src/cloud/cloud_base_compaction.cpp
b/be/src/cloud/cloud_base_compaction.cpp
index 33fca48daed..87e2378fdba 100644
--- a/be/src/cloud/cloud_base_compaction.cpp
+++ b/be/src/cloud/cloud_base_compaction.cpp
@@ -439,6 +439,9 @@ Status CloudBaseCompaction::modify_rowsets() {
// ATTN: MUST NOT update `cumu_compaction_cnt` or `cumu_point` which
are used when sync rowsets, otherwise may cause
// the tablet to be unable to synchronize the rowset meta changes
generated by cumu compaction.
cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
+ if (stats.cumulative_point() >
cloud_tablet()->cumulative_layer_point()) {
+ cloud_tablet()->last_sync_time_s = 0;
+ }
if (output_rowset_delete_bitmap) {
_tablet->tablet_meta()->delete_bitmap().merge(*output_rowset_delete_bitmap);
}
diff --git a/be/src/cloud/cloud_cumulative_compaction.cpp
b/be/src/cloud/cloud_cumulative_compaction.cpp
index d136944ae4b..259c909e2d9 100644
--- a/be/src/cloud/cloud_cumulative_compaction.cpp
+++ b/be/src/cloud/cloud_cumulative_compaction.cpp
@@ -19,6 +19,8 @@
#include <gen_cpp/cloud.pb.h>
+#include <random>
+
#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_tablet_mgr.h"
#include "cloud/config.h"
@@ -29,6 +31,7 @@
#include "service/backend_options.h"
#include "storage/compaction/compaction.h"
#include "storage/compaction/cumulative_compaction_policy.h"
+#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "util/debug_points.h"
#include "util/trace.h"
#include "util/uuid_generator.h"
@@ -43,7 +46,8 @@ bvar::LatencyRecorder
g_cu_compaction_hold_delete_bitmap_lock_time_ms(
CloudCumulativeCompaction::CloudCumulativeCompaction(CloudStorageEngine&
engine,
CloudTabletSPtr tablet)
: CloudCompactionMixin(engine, tablet,
- "BaseCompaction:" +
std::to_string(tablet->tablet_id())) {}
+ "BaseCompaction:" +
std::to_string(tablet->tablet_id())),
+
_enable_parallel_cumu_compaction(config::enable_parallel_cumu_compaction) {}
CloudCumulativeCompaction::~CloudCumulativeCompaction() = default;
@@ -64,11 +68,11 @@ Status CloudCumulativeCompaction::prepare_compact() {
std::vector<std::shared_ptr<CloudCumulativeCompaction>> cumu_compactions;
_engine.get_cumu_compaction(_tablet->tablet_id(), cumu_compactions);
- if (!cumu_compactions.empty()) {
- for (auto& cumu : cumu_compactions) {
- _max_conflict_version =
- std::max(_max_conflict_version,
cumu->_input_rowsets.back()->end_version());
- }
+ for (const auto& cumu : cumu_compactions) {
+ _min_conflict_version =
+ std::min(_min_conflict_version,
cumu->_input_rowsets.front()->start_version());
+ _max_conflict_version =
+ std::max(_max_conflict_version,
cumu->_input_rowsets.back()->end_version());
}
bool need_sync_tablet = true;
@@ -94,7 +98,10 @@ Status CloudCumulativeCompaction::prepare_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.
- update_cumulative_point();
+ // The picker only preserves a delete version reached continuously
from the
+ // cumulative point.
+ DORIS_CHECK_LE(_picked_cumulative_point,
_last_delete_version.first);
+ update_cumulative_point(_picked_cumulative_point,
_last_delete_version.first + 1);
if (!config::enable_sleep_between_delete_cumu_compaction) {
st = Status::Error<CUMULATIVE_MEET_DELETE_VERSION>(
"cumulative compaction meet delete version");
@@ -150,7 +157,7 @@ Status CloudCumulativeCompaction::request_global_lock() {
compaction_job->add_input_versions(_input_rowsets.front()->start_version());
compaction_job->add_input_versions(_input_rowsets.back()->end_version());
// Set input version range to let meta-service check version range conflict
-
compaction_job->set_check_input_versions_range(config::enable_parallel_cumu_compaction);
+
compaction_job->set_check_input_versions_range(_enable_parallel_cumu_compaction);
cloud::StartTabletJobResponse resp;
Status st = _engine.meta_mgr().prepare_tablet_job(job, &resp);
if (!st.ok()) {
@@ -244,12 +251,33 @@ Status CloudCumulativeCompaction::execute_compact() {
Status CloudCumulativeCompaction::modify_rowsets() {
// calculate new cumulative point
- int64_t input_cumulative_point = cloud_tablet()->cumulative_layer_point();
+ int64_t input_cumulative_point;
+ int64_t proposal_base_compaction_cnt;
+ int64_t proposal_cumulative_compaction_cnt;
+ TabletState input_tablet_state;
+ int64_t input_alter_version;
+ {
+ std::shared_lock rlock(_tablet->get_header_lock());
+ input_cumulative_point = cloud_tablet()->cumulative_layer_point();
+ proposal_base_compaction_cnt = cloud_tablet()->base_compaction_cnt();
+ proposal_cumulative_compaction_cnt =
cloud_tablet()->cumulative_compaction_cnt();
+ input_tablet_state = _tablet->tablet_state();
+ input_alter_version = cloud_tablet()->alter_version();
+ }
auto compaction_policy =
cloud_tablet()->tablet_meta()->compaction_policy();
- int64_t new_cumulative_point =
- _engine.cumu_compaction_policy(compaction_policy)
- ->new_cumulative_point(cloud_tablet(), _output_rowset,
_last_delete_version,
- input_cumulative_point);
+ int64_t new_cumulative_point = input_cumulative_point;
+ if (!_enable_parallel_cumu_compaction && input_tablet_state ==
TABLET_NOTREADY &&
+ _output_rowset->start_version() > input_cumulative_point) {
+ // Historical rowsets are absent from a schema-change target until
conversion finishes.
+ DORIS_CHECK_LE(input_cumulative_point, input_alter_version);
+ DORIS_CHECK_GT(_output_rowset->start_version(), input_alter_version);
+ } else if (!_enable_parallel_cumu_compaction ||
+ _output_rowset->start_version() == input_cumulative_point) {
+ new_cumulative_point =
+ _engine.cumu_compaction_policy(compaction_policy)
+ ->new_cumulative_point(cloud_tablet(), _output_rowset,
_last_delete_version,
+ input_cumulative_point);
+ }
// commit compaction job
cloud::TabletJobInfoPB job;
auto idx = job.mutable_idx();
@@ -262,6 +290,8 @@ Status CloudCumulativeCompaction::modify_rowsets() {
compaction_job->set_initiator(BackendOptions::get_localhost() + ':' +
std::to_string(config::heartbeat_service_port));
compaction_job->set_type(cloud::TabletCompactionJobPB::CUMULATIVE);
+ compaction_job->set_base_compaction_cnt(proposal_base_compaction_cnt);
+
compaction_job->set_cumulative_compaction_cnt(proposal_cumulative_compaction_cnt);
compaction_job->set_input_cumulative_point(input_cumulative_point);
compaction_job->set_output_cumulative_point(new_cumulative_point);
compaction_job->set_num_input_rows(_input_row_num);
@@ -307,6 +337,39 @@ Status CloudCumulativeCompaction::modify_rowsets() {
}
});
+ DBUG_EXECUTE_IF("CloudCumulativeCompaction::modify_rowsets.random_sleep", {
+ auto probability = dp->param("probability", dp->param("percent", 0.0));
+ DORIS_CHECK(probability >= 0.0 && probability <= 1.0);
+ static thread_local std::mt19937 gen(std::random_device {}());
+ std::bernoulli_distribution inject_sleep {probability};
+ if (inject_sleep(gen)) {
+ auto max_sleep_ms = dp->param<int64_t>(
+ "max_sleep_ms", dp->param<int64_t>("max_sleep_time_ms",
+
dp->param<int64_t>("max_sleep_time", 0)));
+ DORIS_CHECK(max_sleep_ms >= 0);
+ std::uniform_int_distribution<int64_t> sleep_dist(0, max_sleep_ms);
+ auto sleep_ms = sleep_dist(gen);
+ LOG(INFO) <<
"CloudCumulativeCompaction::modify_rowsets.random_sleep"
+ << ", tablet_id=" << _tablet->tablet_id() << ",
sleep_ms=" << sleep_ms
+ << ", probability=" << probability;
+ std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms));
+ }
+ });
+
+ DBUG_EXECUTE_IF("CloudCumulativeCompaction::modify_rowsets.random_fail", {
+ auto probability = dp->param("probability", dp->param("percent", 0.0));
+ DORIS_CHECK(probability >= 0.0 && probability <= 1.0);
+ static thread_local std::mt19937 gen(std::random_device {}());
+ std::bernoulli_distribution inject_fail {probability};
+ if (inject_fail(gen)) {
+ LOG(WARNING) <<
"CloudCumulativeCompaction::modify_rowsets.random_fail"
+ << ", tablet_id=" << _tablet->tablet_id()
+ << ", probability=" << probability;
+ return Status::InternalError(
+ "debug cloud cumulative compaction modify rowsets random
failed");
+ }
+ });
+
DeleteBitmapPtr output_rowset_delete_bitmap = nullptr;
int64_t initiator = this->initiator();
int64_t get_delete_bitmap_lock_start_time = 0;
@@ -374,20 +437,10 @@ Status CloudCumulativeCompaction::modify_rowsets() {
cloud_tablet()->set_last_cumu_compaction_success_time(std::max(cloud_tablet()->last_cumu_compaction_success_time(),
stats.last_cumu_compaction_time_ms()));
cloud_tablet()->set_last_full_compaction_success_time(std::max(cloud_tablet()->last_full_compaction_success_time(),
stats.last_full_compaction_time_ms()));
// clang-format on
- if (cloud_tablet()->cumulative_compaction_cnt() >=
stats.cumulative_compaction_cnt()) {
- // This could happen while calling `sync_tablet_rowsets` during
`commit_tablet_job`, or parallel cumu compactions which are
- // committed later increase tablet.cumulative_compaction_cnt (see
CloudCompactionTest.parallel_cumu_compaction)
+ if
(!should_apply_cumulative_compaction_result(stats.cumulative_compaction_cnt()))
{
return Status::OK();
}
// Try to make output rowset visible immediately in tablet cache,
instead of waiting for next synchronization from meta-service.
- if (stats.cumulative_point() >
cloud_tablet()->cumulative_layer_point() &&
- stats.cumulative_compaction_cnt() !=
cloud_tablet()->cumulative_compaction_cnt() + 1) {
- // This could happen when there are multiple parallel cumu
compaction committed, tablet cache lags several
- // cumu compactions behind meta-service
(stats.cumulative_compaction_cnt > tablet.cumulative_compaction_cnt + 1).
- // If `cumu_point` of the tablet cache also falls behind, MUST
ONLY synchronize tablet cache from meta-service,
- // otherwise may cause the tablet to be unable to synchronize the
rowset meta changes generated by other cumu compaction.
- return Status::OK();
- }
if (_input_rowsets.size() == 1) {
DCHECK_EQ(_output_rowset->version(), _input_rowsets[0]->version());
// MUST NOT move input rowset to stale path
@@ -491,34 +544,73 @@ Status CloudCumulativeCompaction::garbage_collection() {
return st;
}
-Status CloudCumulativeCompaction::pick_rowsets_to_compact() {
- _input_rowsets.clear();
+Status CloudCumulativeCompaction::advance_cumulative_point_before_pick(
+ int64_t min_conflict_version) {
+ if (!_enable_parallel_cumu_compaction) {
+ return Status::OK();
+ }
- std::vector<RowsetSharedPtr> candidate_rowsets;
+ auto compaction_policy =
+
_engine.cumu_compaction_policy(cloud_tablet()->tablet_meta()->compaction_policy());
+ int64_t input_cumulative_point;
+ int64_t output_cumulative_point;
+ std::vector<RowsetSharedPtr> candidates;
{
std::shared_lock rlock(_tablet->get_header_lock());
+ input_cumulative_point = cloud_tablet()->cumulative_layer_point();
+ output_cumulative_point = input_cumulative_point;
_base_compaction_cnt = cloud_tablet()->base_compaction_cnt();
_cumulative_compaction_cnt =
cloud_tablet()->cumulative_compaction_cnt();
- int64_t candidate_version = std::max(
- std::max(cloud_tablet()->cumulative_layer_point(),
_max_conflict_version + 1),
- cloud_tablet()->alter_version() + 1);
- // Get all rowsets whose version >= `candidate_version` as candidate
rowsets
cloud_tablet()->traverse_rowsets_unlocked(
- [&candidate_rowsets, candidate_version](const RowsetSharedPtr&
rs) {
- if (rs->start_version() >= candidate_version) {
- candidate_rowsets.push_back(rs);
+ [&candidates, input_cumulative_point,
+ min_conflict_version](const RowsetSharedPtr& rs) {
+ if (rs->start_version() >= input_cumulative_point &&
+ rs->end_version() < min_conflict_version) {
+ candidates.push_back(rs);
}
});
}
- if (candidate_rowsets.empty()) {
- return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
- "no suitable versions: candidate rowsets empty");
+ std::sort(candidates.begin(), candidates.end(), Rowset::comparator);
+ const bool is_time_series_policy = compaction_policy->name() ==
CUMULATIVE_TIME_SERIES_POLICY;
+ Version no_delete_version {-1, -1};
+ for (const auto& rowset : candidates) {
+ if (rowset->start_version() != output_cumulative_point) {
+ break;
+ }
+
+ auto rowset_meta = rowset->rowset_meta();
+ if (rowset_meta->has_delete_predicate()) {
+ output_cumulative_point = rowset->end_version() + 1;
+ continue;
+ }
+ // A time-series singleton is a raw delta. Its post-compaction point
rule is not safe here.
+ if (rowset_meta->is_segments_overlapping() ||
+ (is_time_series_policy && rowset_meta->is_singleton_delta())) {
+ break;
+ }
+
+ int64_t new_cumulative_point = compaction_policy->new_cumulative_point(
+ cloud_tablet(), rowset, no_delete_version,
output_cumulative_point);
+ if (new_cumulative_point == output_cumulative_point) {
+ break;
+ }
+ DORIS_CHECK_EQ(new_cumulative_point, rowset->end_version() + 1);
+ output_cumulative_point = new_cumulative_point;
}
- std::sort(candidate_rowsets.begin(), candidate_rowsets.end(),
Rowset::comparator);
- if (auto st = check_version_continuity(candidate_rowsets); !st.ok()) {
- DCHECK(false) << st;
- return st;
+ if (output_cumulative_point == input_cumulative_point) {
+ return Status::OK();
}
+ update_cumulative_point(input_cumulative_point, output_cumulative_point);
+ return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
+ "cumulative point advanced before picking rowsets");
+}
+
+Status CloudCumulativeCompaction::pick_rowsets_to_compact() {
+ _input_rowsets.clear();
+
+ int64_t min_conflict_version = _min_conflict_version;
+ int64_t max_conflict_version = _max_conflict_version;
+
RETURN_IF_ERROR(advance_cumulative_point_before_pick(min_conflict_version));
int64_t max_score = config::cumulative_compaction_max_deltas;
double process_memory_usage =
@@ -532,29 +624,93 @@ Status
CloudCumulativeCompaction::pick_rowsets_to_compact() {
config::cumulative_compaction_min_deltas + 1);
}
- size_t compaction_score = 0;
auto compaction_policy =
cloud_tablet()->tablet_meta()->compaction_policy();
- _engine.cumu_compaction_policy(compaction_policy)
- ->pick_input_rowsets(cloud_tablet(), candidate_rowsets, max_score,
- config::cumulative_compaction_min_deltas,
&_input_rowsets,
- &_last_delete_version, &compaction_score);
-
- if (_input_rowsets.empty()) {
- return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
- "no suitable versions: input rowsets empty");
- } else if (_input_rowsets.size() == 1 &&
-
!_input_rowsets.front()->rowset_meta()->is_segments_overlapping()) {
- VLOG_DEBUG << "there is only one rowset and not overlapping.
tablet_id="
- << _tablet->tablet_id() << ", version=" <<
_input_rowsets.front()->version();
- return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
- "no suitable versions: only one rowset and not overlapping");
+ auto pick_from_candidates = [&](std::vector<RowsetSharedPtr>& candidates) {
+ _input_rowsets.clear();
+ _last_delete_version = Version {-1, -1};
+
+ if (candidates.empty()) {
+ return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
+ "no suitable versions: candidate rowsets empty");
+ }
+ std::sort(candidates.begin(), candidates.end(), Rowset::comparator);
+ if (auto st = check_version_continuity(candidates); !st.ok()) {
+ DCHECK(false) << st;
+ return st;
+ }
+
+ size_t compaction_score = 0;
+ _engine.cumu_compaction_policy(compaction_policy)
+ ->pick_input_rowsets(cloud_tablet(), candidates, max_score,
+ config::cumulative_compaction_min_deltas,
&_input_rowsets,
+ &_last_delete_version, &compaction_score);
+
+ if (_input_rowsets.empty()) {
+ return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
+ "no suitable versions: input rowsets empty");
+ }
+ if (_input_rowsets.size() == 1 &&
+ !_input_rowsets.front()->rowset_meta()->is_segments_overlapping())
{
+ VLOG_DEBUG << "there is only one rowset and not overlapping.
tablet_id="
+ << _tablet->tablet_id() << ", version=" <<
_input_rowsets.front()->version();
+ _input_rowsets.clear();
+ return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
+ "no suitable versions: only one rowset and not
overlapping");
+ }
+ return Status::OK();
+ };
+
+ auto pick_from_version_range = [&](int64_t start_version, int64_t
end_version,
+ bool preserve_delete_version) {
+ std::vector<RowsetSharedPtr> candidates;
+ {
+ std::shared_lock rlock(_tablet->get_header_lock());
+ _base_compaction_cnt = cloud_tablet()->base_compaction_cnt();
+ _cumulative_compaction_cnt =
cloud_tablet()->cumulative_compaction_cnt();
+ _picked_cumulative_point =
cloud_tablet()->cumulative_layer_point();
+ int64_t range_start_version =
+ std::max(std::max(_picked_cumulative_point, start_version),
+ cloud_tablet()->alter_version() + 1);
+ cloud_tablet()->traverse_rowsets_unlocked(
+ [&candidates, range_start_version, end_version](const
RowsetSharedPtr& rs) {
+ if (rs->start_version() >= range_start_version &&
+ rs->start_version() < end_version) {
+ candidates.push_back(rs);
+ }
+ });
+ }
+ auto st = pick_from_candidates(candidates);
+ if (_last_delete_version.first != -1) {
+ DORIS_CHECK(!candidates.empty());
+ if (!preserve_delete_version ||
+ candidates.front()->start_version() !=
_picked_cumulative_point) {
+ _last_delete_version = Version {-1, -1};
+ }
+ }
+ return st;
+ };
+
+ auto st = pick_from_version_range(0, min_conflict_version, true);
+ if (!st.ok()) {
+ if (_last_delete_version.first != -1) {
+ return st;
+ }
+ if (!st.is<CUMULATIVE_NO_SUITABLE_VERSION>() ||
+ min_conflict_version == std::numeric_limits<int64_t>::max()) {
+ return st;
+ }
+ st = pick_from_version_range(max_conflict_version + 1,
std::numeric_limits<int64_t>::max(),
+ false);
+ RETURN_IF_ERROR(st);
}
apply_txn_size_truncation_and_log("CloudCumulativeCompaction");
return Status::OK();
}
-void CloudCumulativeCompaction::update_cumulative_point() {
+void CloudCumulativeCompaction::update_cumulative_point(int64_t
input_cumulative_point,
+ int64_t
output_cumulative_point) {
+ DORIS_CHECK_LT(input_cumulative_point, output_cumulative_point);
cloud::TabletJobInfoPB job;
auto idx = job.mutable_idx();
idx->set_tablet_id(_tablet->tablet_id());
@@ -568,6 +724,9 @@ void CloudCumulativeCompaction::update_cumulative_point() {
compaction_job->set_type(cloud::TabletCompactionJobPB::EMPTY_CUMULATIVE);
compaction_job->set_base_compaction_cnt(_base_compaction_cnt);
compaction_job->set_cumulative_compaction_cnt(_cumulative_compaction_cnt);
+ compaction_job->add_input_versions(input_cumulative_point);
+ compaction_job->add_input_versions(output_cumulative_point - 1);
+
compaction_job->set_check_input_versions_range(_enable_parallel_cumu_compaction);
int64_t now = time(nullptr);
compaction_job->set_lease(now + config::lease_compaction_interval_seconds);
// No need to set expiration time, since there is no output rowset
@@ -587,8 +746,6 @@ void CloudCumulativeCompaction::update_cumulative_point() {
.error(st);
return;
}
- int64_t input_cumulative_point = cloud_tablet()->cumulative_layer_point();
- int64_t output_cumulative_point = _last_delete_version.first + 1;
compaction_job->set_input_cumulative_point(input_cumulative_point);
compaction_job->set_output_cumulative_point(output_cumulative_point);
cloud::FinishTabletJobResponse finish_resp;
@@ -616,14 +773,12 @@ void CloudCumulativeCompaction::update_cumulative_point()
{
cloud_tablet()->set_last_base_compaction_success_time(std::max(cloud_tablet()->last_base_compaction_success_time(),
stats.last_base_compaction_time_ms()));
cloud_tablet()->set_last_cumu_compaction_success_time(std::max(cloud_tablet()->last_cumu_compaction_success_time(),
stats.last_cumu_compaction_time_ms()));
// clang-format on
- if (cloud_tablet()->cumulative_compaction_cnt() >=
stats.cumulative_compaction_cnt()) {
- // This could happen while calling `sync_tablet_rowsets` during
`commit_tablet_job`
+ if
(!should_apply_cumulative_compaction_result(stats.cumulative_compaction_cnt()))
{
return;
}
// ATTN: MUST NOT update `base_compaction_cnt` which are used when
sync rowsets, otherwise may cause
// the tablet to be unable to synchronize the rowset meta changes
generated by base compaction.
-
cloud_tablet()->set_cumulative_compaction_cnt(cloud_tablet()->cumulative_compaction_cnt()
+
- 1);
+
cloud_tablet()->set_cumulative_compaction_cnt(stats.cumulative_compaction_cnt());
cloud_tablet()->set_cumulative_layer_point(stats.cumulative_point());
if (stats.base_compaction_cnt() >=
cloud_tablet()->base_compaction_cnt()) {
cloud_tablet()->reset_approximate_stats(stats.num_rowsets(),
stats.num_segments(),
diff --git a/be/src/cloud/cloud_cumulative_compaction.h
b/be/src/cloud/cloud_cumulative_compaction.h
index 6c0570faa4f..ddccbc52405 100644
--- a/be/src/cloud/cloud_cumulative_compaction.h
+++ b/be/src/cloud/cloud_cumulative_compaction.h
@@ -17,6 +17,7 @@
#pragma once
+#include <limits>
#include <memory>
#include <optional>
@@ -48,23 +49,31 @@ public:
int64_t get_input_num_rows() const { return _input_row_num; }
private:
+ Status advance_cumulative_point_before_pick(int64_t min_conflict_version);
+
Status pick_rowsets_to_compact();
std::string_view compaction_name() const override { return
"CloudCumulativeCompaction"; }
+protected:
Status modify_rowsets() override;
+private:
Status garbage_collection() override;
- void update_cumulative_point();
+ void update_cumulative_point(int64_t input_cumulative_point, int64_t
output_cumulative_point);
ReaderType compaction_type() const override { return
ReaderType::READER_CUMULATIVE_COMPACTION; }
int64_t _input_segments = 0;
+ // A task must use one execution mode even if the dynamic config changes
while it is running.
+ const bool _enable_parallel_cumu_compaction;
+ int64_t _min_conflict_version = std::numeric_limits<int64_t>::max();
int64_t _max_conflict_version = 0;
// Snapshot values when pick input rowsets
int64_t _base_compaction_cnt = 0;
int64_t _cumulative_compaction_cnt = 0;
+ int64_t _picked_cumulative_point = 0;
Version _last_delete_version {-1, -1};
};
diff --git a/be/src/cloud/cloud_index_change_compaction.cpp
b/be/src/cloud/cloud_index_change_compaction.cpp
index 03a0e02aec6..3bb55f46ce8 100644
--- a/be/src/cloud/cloud_index_change_compaction.cpp
+++ b/be/src/cloud/cloud_index_change_compaction.cpp
@@ -263,8 +263,13 @@ Status CloudIndexChangeCompaction::modify_rowsets() {
compaction_job->set_id(_uuid);
compaction_job->set_initiator(BackendOptions::get_localhost() + ':' +
std::to_string(config::heartbeat_service_port));
-
compaction_job->set_input_cumulative_point(cloud_tablet()->cumulative_layer_point());
-
compaction_job->set_output_cumulative_point(cloud_tablet()->cumulative_layer_point());
+ {
+ std::shared_lock rlock(_tablet->get_header_lock());
+
compaction_job->set_input_cumulative_point(cloud_tablet()->cumulative_layer_point());
+
compaction_job->set_output_cumulative_point(cloud_tablet()->cumulative_layer_point());
+
compaction_job->set_base_compaction_cnt(cloud_tablet()->base_compaction_cnt());
+
compaction_job->set_cumulative_compaction_cnt(cloud_tablet()->cumulative_compaction_cnt());
+ }
compaction_job->set_num_input_rows(_input_row_num);
compaction_job->set_num_output_rows(_output_rowset->num_rows());
compaction_job->set_size_input_rowsets(_input_rowsets_total_size);
@@ -374,6 +379,9 @@ void
CloudIndexChangeCompaction::_update_tablet_for_base_compaction(
// ATTN: MUST NOT update `cumu_compaction_cnt` or `cumu_point` which
are used when sync rowsets, otherwise may cause
// the tablet to be unable to synchronize the rowset meta changes
generated by cumu compaction.
cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
+ if (stats.cumulative_point() >
cloud_tablet()->cumulative_layer_point()) {
+ cloud_tablet()->last_sync_time_s = 0;
+ }
if (output_rowset_delete_bitmap) {
_tablet->tablet_meta()->delete_bitmap().merge(*output_rowset_delete_bitmap);
}
@@ -395,20 +403,10 @@ void
CloudIndexChangeCompaction::_update_tablet_for_cumu_compaction(
cloud_tablet()->set_last_cumu_compaction_success_time(std::max(cloud_tablet()->last_cumu_compaction_success_time(),
stats.last_cumu_compaction_time_ms()));
cloud_tablet()->set_last_full_compaction_success_time(std::max(cloud_tablet()->last_full_compaction_success_time(),
stats.last_full_compaction_time_ms()));
// clang-format on
- if (cloud_tablet()->cumulative_compaction_cnt() >=
stats.cumulative_compaction_cnt()) {
- // This could happen while calling `sync_tablet_rowsets` during
`commit_tablet_job`, or parallel cumu compactions which are
- // committed later increase tablet.cumulative_compaction_cnt (see
CloudCompactionTest.parallel_cumu_compaction)
+ if
(!should_apply_cumulative_compaction_result(stats.cumulative_compaction_cnt()))
{
return;
}
// Try to make output rowset visible immediately in tablet cache,
instead of waiting for next synchronization from meta-service.
- if (stats.cumulative_point() >
cloud_tablet()->cumulative_layer_point() &&
- stats.cumulative_compaction_cnt() !=
cloud_tablet()->cumulative_compaction_cnt() + 1) {
- // This could happen when there are multiple parallel cumu
compaction committed, tablet cache lags several
- // cumu compactions behind meta-service
(stats.cumulative_compaction_cnt > tablet.cumulative_compaction_cnt + 1).
- // If `cumu_point` of the tablet cache also falls behind, MUST
ONLY synchronize tablet cache from meta-service,
- // otherwise may cause the tablet to be unable to synchronize the
rowset meta changes generated by other cumu compaction.
- return;
- }
if (_input_rowsets.size() == 1) {
DCHECK_EQ(_output_rowset->version(), _input_rowsets[0]->version());
// MUST NOT move input rowset to stale path.
diff --git a/be/src/storage/compaction/compaction.cpp
b/be/src/storage/compaction/compaction.cpp
index 1b9a110e6e2..3e411f1d388 100644
--- a/be/src/storage/compaction/compaction.cpp
+++ b/be/src/storage/compaction/compaction.cpp
@@ -2089,6 +2089,26 @@
CloudCompactionMixin::CloudCompactionMixin(CloudStorageEngine& engine, CloudTabl
_uuid = ss.str();
}
+bool CloudCompactionMixin::should_apply_cumulative_compaction_result(
+ int64_t response_cumulative_compaction_cnt) {
+ int64_t local_cumulative_compaction_cnt =
cloud_tablet()->cumulative_compaction_cnt();
+ if (local_cumulative_compaction_cnt >= response_cumulative_compaction_cnt)
{
+ // sync_rowsets or another compaction has already installed this
result.
+ return false;
+ }
+ if (response_cumulative_compaction_cnt != local_cumulative_compaction_cnt
+ 1) {
+ // Only the current task's output is available locally. Sync all
missing outputs instead.
+ cloud_tablet()->last_sync_time_s = 0;
+ LOG_INFO("defer applying cumulative compaction result until tablet
sync")
+ .tag("tablet_id", _tablet->tablet_id())
+ .tag("job_id", _uuid)
+ .tag("local_cumulative_compaction_cnt",
local_cumulative_compaction_cnt)
+ .tag("response_cumulative_compaction_cnt",
response_cumulative_compaction_cnt);
+ return false;
+ }
+ return true;
+}
+
Status CloudCompactionMixin::execute_compact_impl(int64_t permits) {
OlapStopWatch watch;
diff --git a/be/src/storage/compaction/compaction.h
b/be/src/storage/compaction/compaction.h
index 4c8556e4203..212039e77e2 100644
--- a/be/src/storage/compaction/compaction.h
+++ b/be/src/storage/compaction/compaction.h
@@ -259,6 +259,9 @@ protected:
// Returns the number of rowsets that were truncated
size_t apply_txn_size_truncation_and_log(const std::string&
compaction_name);
+ // Caller must hold the tablet header lock.
+ bool should_apply_cumulative_compaction_result(int64_t
response_cumulative_compaction_cnt);
+
CloudStorageEngine& _engine;
std::string _uuid;
diff --git a/be/test/cloud/cloud_compaction_test.cpp
b/be/test/cloud/cloud_compaction_test.cpp
index 9ebf0c8d01e..f435997e5de 100644
--- a/be/test/cloud/cloud_compaction_test.cpp
+++ b/be/test/cloud/cloud_compaction_test.cpp
@@ -22,16 +22,20 @@
#include <gtest/gtest.h>
#include <memory>
+#include <mutex>
#include <string_view>
+#include <unordered_map>
#include "cloud/cloud_base_compaction.h"
#include "cloud/cloud_cluster_info.h"
+#include "cloud/cloud_cumulative_compaction.h"
#include "cloud/cloud_rowset_builder.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
#include "cloud/config.h"
#include "common/metrics/doris_metrics.h"
+#include "cpp/sync_point.h"
#include "io/fs/s3_file_system.h"
#include "json2pb/json_to_pb.h"
#include "storage/compaction/cumulative_compaction_time_series_policy.h"
@@ -42,6 +46,7 @@
#include "storage/tablet/tablet_meta.h"
#include "util/defer_op.h"
#include "util/s3_util.h"
+#include "util/time.h"
#include "util/uid_util.h"
namespace doris {
@@ -406,6 +411,15 @@ static RowsetSharedPtr create_rowset(Version version, int
num_segments, bool ove
return rowset;
}
+static RowsetSharedPtr create_delete_rowset(Version version) {
+ auto rowset = create_rowset(version, 0, false, 0);
+ DORIS_CHECK(rowset != nullptr);
+ DeletePredicatePB delete_predicate;
+ delete_predicate.set_version(version.second);
+ rowset->rowset_meta()->set_delete_predicate(std::move(delete_predicate));
+ return rowset;
+}
+
static RowsetSharedPtr create_prepared_rowset(int num_segments, int data_size)
{
auto rs_meta = std::make_shared<RowsetMeta>();
rs_meta->set_rowset_type(BETA_ROWSET);
@@ -475,6 +489,11 @@ public:
return apply_txn_size_truncation_and_log(compaction_name);
}
+ bool test_should_apply_cumulative_compaction_result(
+ int64_t response_cumulative_compaction_cnt) {
+ return
should_apply_cumulative_compaction_result(response_cumulative_compaction_cnt);
+ }
+
Status prepare_compact() override { return Status::OK(); }
ReaderType compaction_type() const override { return
ReaderType::READER_CUMULATIVE_COMPACTION; }
@@ -482,6 +501,447 @@ public:
std::string_view compaction_name() const override { return
"test_compaction"; }
};
+TEST_F(CloudCompactionTest, cumulative_result_requires_next_counter) {
+ auto tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
+ tablet->set_cumulative_compaction_cnt(1);
+ tablet->last_sync_time_s = 1;
+ TestableCloudCompaction compaction(_engine, tablet);
+
+ std::unique_lock lock(tablet->get_header_lock());
+ EXPECT_FALSE(compaction.test_should_apply_cumulative_compaction_result(1));
+ EXPECT_EQ(tablet->last_sync_time_s, 1);
+ EXPECT_TRUE(compaction.test_should_apply_cumulative_compaction_result(2));
+ EXPECT_EQ(tablet->last_sync_time_s, 1);
+ EXPECT_FALSE(compaction.test_should_apply_cumulative_compaction_result(3));
+ EXPECT_EQ(tablet->last_sync_time_s, 0);
+}
+
+class TestableCloudCumulativeCompaction : public CloudCumulativeCompaction {
+public:
+ TestableCloudCumulativeCompaction(CloudStorageEngine& engine,
CloudTabletSPtr tablet)
+ : CloudCumulativeCompaction(engine, tablet) {}
+
+ void set_input_rowsets(const std::vector<RowsetSharedPtr>& rowsets) {
+ _input_rowsets = rowsets;
+ }
+
+ const std::vector<RowsetSharedPtr>& input_rowsets() const { return
_input_rowsets; }
+
+ void set_output_rowset(RowsetSharedPtr rowset) { _output_rowset =
std::move(rowset); }
+
+ Status test_modify_rowsets() { return modify_rowsets(); }
+};
+
+static TabletMetaSharedPtr create_cloud_compaction_test_tablet_meta(int64_t
tablet_id) {
+ return std::make_shared<TabletMeta>(1, 2, tablet_id, 15674, 4, 5,
TTabletSchema(), 6,
+ std::unordered_map<uint32_t, uint32_t>
{{7, 8}},
+ UniqueId(9, 10),
TTabletType::TABLET_TYPE_DISK,
+ TCompressionType::LZ4F);
+}
+
+static CloudTabletSPtr create_cloud_tablet_with_rowsets(CloudStorageEngine&
engine,
+ const
TabletMetaSharedPtr& tablet_meta,
+ int64_t
cumulative_point,
+
std::vector<RowsetSharedPtr> rowsets) {
+ auto tablet = std::make_shared<CloudTablet>(engine, tablet_meta);
+ auto num_rowsets = rowsets.size();
+ {
+ std::unique_lock wlock(tablet->get_header_lock());
+ tablet->add_rowsets(std::move(rowsets), false, wlock, false);
+ }
+ tablet->set_cumulative_layer_point(cumulative_point);
+
tablet->fetch_add_approximate_num_rowsets(static_cast<int64_t>(num_rowsets) -
+
tablet->fetch_add_approximate_num_rowsets(0));
+ tablet->last_sync_time_s = 1;
+ return tablet;
+}
+
+static CloudTabletSPtr create_cloud_tablet_with_rowsets(CloudStorageEngine&
engine,
+ const
TabletMetaSharedPtr& tablet_meta,
+ int64_t
cumulative_point,
+ const
std::vector<int64_t>& versions,
+ int64_t data_size =
1024 * 1024) {
+ std::vector<RowsetSharedPtr> rowsets;
+ rowsets.reserve(versions.size());
+ for (int64_t version : versions) {
+ auto rowset = create_rowset(Version(version, version), 1, false,
data_size);
+ DORIS_CHECK(rowset != nullptr);
+ rowsets.push_back(rowset);
+ }
+ return create_cloud_tablet_with_rowsets(engine, tablet_meta,
cumulative_point,
+ std::move(rowsets));
+}
+
+static std::shared_ptr<TestableCloudCumulativeCompaction>
create_inflight_cumu_compaction(
+ CloudStorageEngine& engine, const CloudTabletSPtr& tablet, int64_t
start, int64_t end) {
+ std::vector<RowsetSharedPtr> input_rowsets;
+ input_rowsets.reserve(end - start + 1);
+ for (int64_t version = start; version <= end; ++version) {
+ auto rowset = create_rowset(Version(version, version), 1, false, 1024
* 1024);
+ DORIS_CHECK(rowset != nullptr);
+ input_rowsets.push_back(rowset);
+ }
+ auto compaction =
std::make_shared<TestableCloudCumulativeCompaction>(engine, tablet);
+ compaction->set_input_rowsets(input_rowsets);
+ return compaction;
+}
+
+TEST_F(CloudCompactionTest,
base_result_with_newer_cumulative_point_forces_sync) {
+ auto* sync_point = SyncPoint::get_instance();
+ Defer clear_sync_points([&] {
+ sync_point->disable_processing();
+ sync_point->clear_all_call_backs();
+ });
+ int64_t response_cumulative_point = 6;
+ sync_point->set_call_back("CloudMetaMgr::commit_tablet_job", [&](auto&&
outcome) {
+ auto* result = try_any_cast_ret<Status>(outcome);
+ result->first = Status::OK();
+ result->second = true;
+ auto* response =
try_any_cast<cloud::FinishTabletJobResponse*>(outcome[1]);
+ response->mutable_status()->set_code(cloud::MetaServiceCode::OK);
+ auto* stats = response->mutable_stats();
+ stats->set_base_compaction_cnt(1);
+ stats->set_cumulative_compaction_cnt(0);
+ stats->set_cumulative_point(response_cumulative_point);
+ stats->set_num_rowsets(1);
+ });
+ sync_point->enable_processing();
+
+ auto run_case = [&](int64_t tablet_id, int64_t response_point, int64_t
expected_sync_time) {
+ auto input = create_rowset(Version(2, 7), 1, false, 1024);
+ auto tablet = create_cloud_tablet_with_rowsets(
+ _engine, create_cloud_compaction_test_tablet_meta(tablet_id),
6, {input});
+ auto output = create_rowset(Version(2, 7), 1, false, 1024);
+ CloudBaseCompaction compaction(_engine, tablet);
+ compaction._input_rowsets = {input};
+ compaction._output_rowset = output;
+ response_cumulative_point = response_point;
+
+ ASSERT_TRUE(compaction.modify_rowsets().ok());
+ EXPECT_EQ(tablet->cumulative_layer_point(), 6);
+ EXPECT_EQ(tablet->last_sync_time_s, expected_sync_time);
+ };
+
+ run_case(10008, 8, 0);
+ run_case(10009, 6, 1);
+}
+
+TEST_F(CloudCompactionTest, cumulative_pick_uses_local_conflict_window) {
+ auto old_min_deltas = config::cumulative_compaction_min_deltas;
+ auto old_parallel_cumu_compaction =
config::enable_parallel_cumu_compaction;
+ Defer restore_config([&] {
+ config::cumulative_compaction_min_deltas = old_min_deltas;
+ config::enable_parallel_cumu_compaction = old_parallel_cumu_compaction;
+ });
+ config::cumulative_compaction_min_deltas = 2;
+ config::enable_parallel_cumu_compaction = true;
+
+ {
+ auto tablet_meta = create_cloud_compaction_test_tablet_meta(10001);
+ auto tablet =
+ create_cloud_tablet_with_rowsets(_engine, tablet_meta, 114,
+ {114, 115, 116, 117, 118,
119, 120, 121, 122, 123,
+ 124, 125, 126, 127, 128,
129, 130, 131, 132});
+ _engine._submitted_cumu_compactions[tablet->tablet_id()] = {
+ create_inflight_cumu_compaction(_engine, tablet, 117, 119),
+ create_inflight_cumu_compaction(_engine, tablet, 126, 130)};
+
+ TestableCloudCumulativeCompaction compaction(_engine, tablet);
+ auto st = compaction.prepare_compact();
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(compaction.input_rowsets().size(), 3);
+ EXPECT_EQ(compaction.input_rowsets().front()->start_version(), 114);
+ EXPECT_EQ(compaction.input_rowsets().back()->end_version(), 116);
+ _engine._submitted_cumu_compactions.clear();
+ }
+
+ {
+ auto tablet_meta = create_cloud_compaction_test_tablet_meta(10002);
+ auto tablet = create_cloud_tablet_with_rowsets(_engine, tablet_meta,
1, {1, 2, 3, 4});
+ _engine._submitted_cumu_compactions[tablet->tablet_id()] = {
+ create_inflight_cumu_compaction(_engine, tablet, 1, 2)};
+
+ TestableCloudCumulativeCompaction compaction(_engine, tablet);
+ auto st = compaction.prepare_compact();
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(compaction.input_rowsets().size(), 2);
+ EXPECT_EQ(compaction.input_rowsets().front()->start_version(), 3);
+ EXPECT_EQ(compaction.input_rowsets().back()->end_version(), 4);
+ _engine._submitted_cumu_compactions.clear();
+ }
+
+ {
+ auto tablet_meta = create_cloud_compaction_test_tablet_meta(10003);
+ auto tablet = create_cloud_tablet_with_rowsets(_engine, tablet_meta,
114,
+ {114, 115, 116, 117,
118, 119});
+ _engine._submitted_cumu_compactions[tablet->tablet_id()] = {
+ create_inflight_cumu_compaction(_engine, tablet, 115, 116)};
+
+ TestableCloudCumulativeCompaction compaction(_engine, tablet);
+ auto st = compaction.prepare_compact();
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(compaction.input_rowsets().size(), 3);
+ EXPECT_EQ(compaction.input_rowsets().front()->start_version(), 117);
+ EXPECT_EQ(compaction.input_rowsets().back()->end_version(), 119);
+ _engine._submitted_cumu_compactions.clear();
+ }
+}
+
+TEST_F(CloudCompactionTest,
serial_suffix_compaction_on_running_tablet_keeps_point) {
+ auto old_parallel_cumu_compaction =
config::enable_parallel_cumu_compaction;
+ Defer restore_config(
+ [&] { config::enable_parallel_cumu_compaction =
old_parallel_cumu_compaction; });
+ config::enable_parallel_cumu_compaction = false;
+
+ auto* sync_point = SyncPoint::get_instance();
+ Defer clear_sync_points([&] {
+ sync_point->disable_processing();
+ sync_point->clear_all_call_backs();
+ });
+ sync_point->set_call_back("CloudMetaMgr::commit_tablet_job", [&](auto&&
outcome) {
+ auto job = try_any_cast<cloud::TabletJobInfoPB>(outcome[0]);
+ ASSERT_EQ(job.compaction_size(), 1);
+ EXPECT_EQ(job.compaction(0).input_cumulative_point(), 2);
+ EXPECT_EQ(job.compaction(0).output_cumulative_point(), 2);
+
+ auto* result = try_any_cast_ret<Status>(outcome);
+ result->first = Status::InternalError("stop after checking cumulative
point");
+ result->second = true;
+ });
+ sync_point->enable_processing();
+
+ auto tablet_meta = create_cloud_compaction_test_tablet_meta(10007);
+ auto tablet = create_cloud_tablet_with_rowsets(_engine, tablet_meta, 2,
{2});
+ TestableCloudCumulativeCompaction compaction(_engine, tablet);
+ compaction.set_input_rowsets({create_rowset(Version(3, 22), 1, true, 20 *
1024 * 1024)});
+ compaction.set_output_rowset(create_rowset(Version(3, 22), 1, false, 20 *
1024 * 1024));
+
+ EXPECT_FALSE(compaction.test_modify_rowsets().ok());
+}
+
+TEST_F(CloudCompactionTest,
parallel_time_series_pick_preserves_raw_singletons) {
+ auto old_parallel_cumu_compaction =
config::enable_parallel_cumu_compaction;
+ Defer restore_config(
+ [&] { config::enable_parallel_cumu_compaction =
old_parallel_cumu_compaction; });
+ config::enable_parallel_cumu_compaction = true;
+
+ auto* sync_point = SyncPoint::get_instance();
+ Defer clear_sync_points([&] {
+ sync_point->disable_processing();
+ sync_point->clear_all_call_backs();
+ });
+ bool point_update_called = false;
+ sync_point->set_call_back("CloudMetaMgr::prepare_tablet_job", [&](auto&&
outcome) {
+ point_update_called = true;
+ auto* result = try_any_cast_ret<Status>(outcome);
+ result->first = Status::InternalError("unexpected cumulative point
update");
+ result->second = true;
+ });
+ sync_point->enable_processing();
+
+ auto tablet_meta = create_cloud_compaction_test_tablet_meta(10008);
+
tablet_meta->set_compaction_policy(std::string(CUMULATIVE_TIME_SERIES_POLICY));
+ tablet_meta->set_time_series_compaction_level_threshold(1);
+ std::vector<RowsetSharedPtr> rowsets {
+ create_rowset(Version(2, 2), 1, false, 1024 * 1024),
+ create_rowset(Version(3, 3), 1, false, 1024 * 1024),
+ };
+ for (const auto& rowset : rowsets) {
+ rowset->rowset_meta()->set_creation_time(UnixSeconds());
+ }
+ auto tablet = create_cloud_tablet_with_rowsets(_engine, tablet_meta, 2,
std::move(rowsets));
+ TestableCloudCumulativeCompaction compaction(_engine, tablet);
+
+ auto st = compaction.prepare_compact();
+ EXPECT_TRUE(st.is<ErrorCode::CUMULATIVE_NO_SUITABLE_VERSION>()) << st;
+ EXPECT_FALSE(point_update_called);
+ EXPECT_EQ(tablet->cumulative_layer_point(), 2);
+}
+
+TEST_F(CloudCompactionTest,
parallel_pick_keeps_mode_after_dynamic_config_change) {
+ auto old_parallel_cumu_compaction =
config::enable_parallel_cumu_compaction;
+ Defer restore_config(
+ [&] { config::enable_parallel_cumu_compaction =
old_parallel_cumu_compaction; });
+ config::enable_parallel_cumu_compaction = true;
+
+ auto* sync_point = SyncPoint::get_instance();
+ Defer clear_sync_points([&] {
+ sync_point->disable_processing();
+ sync_point->clear_all_call_backs();
+ });
+
+ bool prepare_called = false;
+ sync_point->set_call_back("CloudMetaMgr::prepare_tablet_job", [&](auto&&
outcome) {
+ prepare_called = true;
+ auto job = try_any_cast<cloud::TabletJobInfoPB>(outcome[0]);
+ ASSERT_EQ(job.compaction_size(), 1);
+ const auto& compaction = job.compaction(0);
+ EXPECT_EQ(compaction.type(),
cloud::TabletCompactionJobPB::EMPTY_CUMULATIVE);
+ ASSERT_EQ(compaction.input_versions_size(), 2);
+ EXPECT_EQ(compaction.input_versions(0), 2);
+ EXPECT_EQ(compaction.input_versions(1), 4);
+ EXPECT_TRUE(compaction.check_input_versions_range());
+ EXPECT_EQ(compaction.base_compaction_cnt(), 0);
+ EXPECT_EQ(compaction.cumulative_compaction_cnt(), 0);
+
+ auto* result = try_any_cast_ret<Status>(outcome);
+ result->first = Status::OK();
+ result->second = true;
+ auto* response =
try_any_cast<cloud::StartTabletJobResponse*>(outcome[1]);
+ response->mutable_status()->set_code(cloud::MetaServiceCode::OK);
+ });
+
+ bool commit_called = false;
+ sync_point->set_call_back("CloudMetaMgr::commit_tablet_job", [&](auto&&
outcome) {
+ commit_called = true;
+ auto job = try_any_cast<cloud::TabletJobInfoPB>(outcome[0]);
+ ASSERT_EQ(job.compaction_size(), 1);
+ const auto& compaction = job.compaction(0);
+ EXPECT_EQ(compaction.input_cumulative_point(), 2);
+ EXPECT_EQ(compaction.output_cumulative_point(), 5);
+
+ auto* result = try_any_cast_ret<Status>(outcome);
+ result->first = Status::OK();
+ result->second = true;
+ auto* response =
try_any_cast<cloud::FinishTabletJobResponse*>(outcome[1]);
+ response->mutable_status()->set_code(cloud::MetaServiceCode::OK);
+ auto* stats = response->mutable_stats();
+ stats->set_base_compaction_cnt(0);
+ stats->set_cumulative_compaction_cnt(2);
+ stats->set_cumulative_point(5);
+ stats->set_num_rowsets(3);
+ stats->set_num_segments(3);
+ stats->set_num_rows(0);
+ stats->set_data_size(300 * 1024 * 1024);
+ });
+ sync_point->enable_processing();
+
+ auto tablet_meta = create_cloud_compaction_test_tablet_meta(10004);
+ auto tablet =
+ create_cloud_tablet_with_rowsets(_engine, tablet_meta, 2, {2, 3,
4}, 100 * 1024 * 1024);
+ TestableCloudCumulativeCompaction compaction(_engine, tablet);
+ config::enable_parallel_cumu_compaction = false;
+ auto st = compaction.prepare_compact();
+
+ EXPECT_TRUE(st.is<ErrorCode::CUMULATIVE_NO_SUITABLE_VERSION>()) << st;
+ EXPECT_TRUE(prepare_called);
+ EXPECT_TRUE(commit_called);
+ EXPECT_EQ(tablet->cumulative_compaction_cnt(), 0);
+ EXPECT_EQ(tablet->cumulative_layer_point(), 2);
+ EXPECT_EQ(tablet->last_sync_time_s, 0);
+}
+
+TEST_F(CloudCompactionTest,
parallel_pick_advances_continuous_low_prefix_through_delete) {
+ auto old_parallel_cumu_compaction =
config::enable_parallel_cumu_compaction;
+ Defer restore_config(
+ [&] { config::enable_parallel_cumu_compaction =
old_parallel_cumu_compaction; });
+ config::enable_parallel_cumu_compaction = true;
+
+ auto* sync_point = SyncPoint::get_instance();
+ Defer clear_sync_points([&] {
+ sync_point->disable_processing();
+ sync_point->clear_all_call_backs();
+ });
+
+ bool prepare_called = false;
+ sync_point->set_call_back("CloudMetaMgr::prepare_tablet_job", [&](auto&&
outcome) {
+ prepare_called = true;
+ auto job = try_any_cast<cloud::TabletJobInfoPB>(outcome[0]);
+ ASSERT_EQ(job.compaction_size(), 1);
+ const auto& compaction = job.compaction(0);
+ EXPECT_EQ(compaction.type(),
cloud::TabletCompactionJobPB::EMPTY_CUMULATIVE);
+ ASSERT_EQ(compaction.input_versions_size(), 2);
+ EXPECT_EQ(compaction.input_versions(0), 160);
+ EXPECT_EQ(compaction.input_versions(1), 162);
+ EXPECT_TRUE(compaction.check_input_versions_range());
+
+ auto* result = try_any_cast_ret<Status>(outcome);
+ result->first = Status::OK();
+ result->second = true;
+ auto* response =
try_any_cast<cloud::StartTabletJobResponse*>(outcome[1]);
+ response->mutable_status()->set_code(cloud::MetaServiceCode::OK);
+ });
+
+ bool commit_called = false;
+ sync_point->set_call_back("CloudMetaMgr::commit_tablet_job", [&](auto&&
outcome) {
+ commit_called = true;
+ auto job = try_any_cast<cloud::TabletJobInfoPB>(outcome[0]);
+ ASSERT_EQ(job.compaction_size(), 1);
+ const auto& compaction = job.compaction(0);
+ EXPECT_EQ(compaction.input_cumulative_point(), 160);
+ EXPECT_EQ(compaction.output_cumulative_point(), 163);
+
+ auto* result = try_any_cast_ret<Status>(outcome);
+ result->first = Status::OK();
+ result->second = true;
+ auto* response =
try_any_cast<cloud::FinishTabletJobResponse*>(outcome[1]);
+ response->mutable_status()->set_code(cloud::MetaServiceCode::OK);
+ auto* stats = response->mutable_stats();
+ stats->set_base_compaction_cnt(0);
+ stats->set_cumulative_compaction_cnt(1);
+ stats->set_cumulative_point(163);
+ stats->set_num_rowsets(2);
+ stats->set_num_segments(1);
+ stats->set_num_rows(0);
+ stats->set_data_size(1024);
+ });
+ sync_point->enable_processing();
+
+ auto tablet_meta = create_cloud_compaction_test_tablet_meta(10005);
+ std::vector<RowsetSharedPtr> rowsets {
+ create_rowset(Version(160, 161), 1, false, 1024),
+ create_delete_rowset(Version(162, 162)),
+ };
+ auto tablet = create_cloud_tablet_with_rowsets(_engine, tablet_meta, 160,
std::move(rowsets));
+ TestableCloudCumulativeCompaction compaction(_engine, tablet);
+ auto st = compaction.prepare_compact();
+
+ EXPECT_FALSE(st.ok()) << st;
+ EXPECT_TRUE(prepare_called);
+ EXPECT_TRUE(commit_called);
+ EXPECT_EQ(tablet->cumulative_layer_point(), 163);
+}
+
+TEST_F(CloudCompactionTest,
parallel_pick_does_not_advance_from_high_range_delete) {
+ auto old_parallel_cumu_compaction =
config::enable_parallel_cumu_compaction;
+ Defer restore_config(
+ [&] { config::enable_parallel_cumu_compaction =
old_parallel_cumu_compaction; });
+ config::enable_parallel_cumu_compaction = true;
+
+ auto* sync_point = SyncPoint::get_instance();
+ Defer clear_sync_points([&] {
+ sync_point->disable_processing();
+ sync_point->clear_all_call_backs();
+ });
+ bool prepare_called = false;
+ sync_point->set_call_back("CloudMetaMgr::prepare_tablet_job", [&](auto&&
outcome) {
+ prepare_called = true;
+ auto* result = try_any_cast_ret<Status>(outcome);
+ result->first = Status::InternalError("unexpected cumulative point
update");
+ result->second = true;
+ });
+ sync_point->enable_processing();
+
+ auto tablet_meta = create_cloud_compaction_test_tablet_meta(10006);
+ std::vector<RowsetSharedPtr> rowsets {
+ create_rowset(Version(100, 199), 1, false, 1024),
+ create_rowset(Version(200, 300), 1, false, 1024),
+ create_rowset(Version(301, 301), 1, false, 1024),
+ create_delete_rowset(Version(302, 302)),
+ };
+ auto tablet = create_cloud_tablet_with_rowsets(_engine, tablet_meta, 100,
std::move(rowsets));
+ _engine._submitted_cumu_compactions[tablet->tablet_id()] = {
+ create_inflight_cumu_compaction(_engine, tablet, 200, 300)};
+ Defer clear_compactions([&] { _engine._submitted_cumu_compactions.clear();
});
+
+ TestableCloudCumulativeCompaction compaction(_engine, tablet);
+ auto st = compaction.prepare_compact();
+
+ EXPECT_TRUE(st.is<ErrorCode::CUMULATIVE_NO_SUITABLE_VERSION>()) << st;
+ EXPECT_FALSE(prepare_called);
+ EXPECT_EQ(tablet->cumulative_layer_point(), 100);
+}
+
TEST_F(CloudCompactionTest, test_set_storage_resource_from_input_rowsets) {
S3Conf s3_conf {.bucket = "bucket",
.prefix = "prefix",
diff --git a/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
b/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
index 6788f317f8d..62e5f27d7a4 100644
--- a/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
+++ b/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
@@ -417,6 +417,26 @@ TEST_F(CloudIndexChangeCompactionTest, ms_ret_status_test)
{
ASSERT_TRUE(!ret.ok());
ASSERT_TRUE(contains_str(ret.to_string(), "failed in schema change"));
}
+
+ {
+ tablet->set_base_compaction_cnt(0);
+ tablet->set_cumulative_layer_point(6);
+ tablet->last_sync_time_s = 1;
+ auto index_change_compact =
std::make_shared<CloudIndexChangeCompaction>(
+ *_engine, tablet, 0, index_list, columns);
+ index_change_compact->_input_rowsets.push_back(rowset_ptr);
+ index_change_compact->_output_rowset = rowset_ptr;
+ index_change_compact->_compact_type =
cloud::TabletCompactionJobPB::BASE;
+ cloud::FinishTabletJobResponse response;
+ response.mutable_stats()->set_base_compaction_cnt(1);
+ response.mutable_stats()->set_cumulative_compaction_cnt(0);
+ response.mutable_stats()->set_cumulative_point(8);
+
+ index_change_compact->_update_tablet_for_base_compaction(response,
nullptr);
+
+ EXPECT_EQ(tablet->cumulative_layer_point(), 6);
+ EXPECT_EQ(tablet->last_sync_time_s, 0);
+ }
}
TEST_F(CloudIndexChangeCompactionTest, basic_compaction_test) {
diff --git a/cloud/src/meta-service/meta_service_job.cpp
b/cloud/src/meta-service/meta_service_job.cpp
index 35aedd81852..6cc587a41ae 100644
--- a/cloud/src/meta-service/meta_service_job.cpp
+++ b/cloud/src/meta-service/meta_service_job.cpp
@@ -807,20 +807,96 @@ static void
remove_delete_bitmap_update_lock(std::unique_ptr<Transaction>& txn,
}
}
+static bool should_accept_cumulative_point(const std::string& instance_id,
int64_t tablet_id,
+ const TabletCompactionJobPB&
compaction,
+ const TabletCompactionJobPB&
recorded_compaction,
+ const TabletStatsPB& stats) {
+ if (compaction.type() != TabletCompactionJobPB::CUMULATIVE &&
+ compaction.type() != TabletCompactionJobPB::EMPTY_CUMULATIVE) {
+ return true;
+ }
+ // Safe because tablet stats keep max(current, proposal).
+ if (compaction.output_cumulative_point() <= stats.cumulative_point()) {
+ return true;
+ }
+ // The committed output covers [current point, proposal - 1].
+ if (compaction.type() == TabletCompactionJobPB::CUMULATIVE &&
+ compaction.input_versions_size() == 2 &&
+ compaction.input_versions(0) == stats.cumulative_point() &&
+ compaction.output_cumulative_point() == compaction.input_versions(1) +
1) {
+ return true;
+ }
+
+ // For legacy BEs:
+ // 1. The snapshot comes from START.
+ // 2. Parallel jobs may share it, so it cannot prove that a FINISH
proposal is current.
+ const bool finish_has_counters = compaction.has_base_compaction_cnt();
+ const auto& snapshot = finish_has_counters ? compaction :
recorded_compaction;
+ const int64_t snapshot_base_cnt = snapshot.base_compaction_cnt();
+ const int64_t snapshot_cumu_cnt = snapshot.cumulative_compaction_cnt();
+ // FULL also increments base_compaction_cnt
+ bool accept = snapshot_base_cnt == stats.base_compaction_cnt() &&
+ snapshot_cumu_cnt == stats.cumulative_compaction_cnt();
+ if (!finish_has_counters) {
+ if (compaction.type() == TabletCompactionJobPB::CUMULATIVE) {
+ // An advancing legacy CUMULATIVE proposal is safe only when:
+ // With point=2, accept [2-4] -> 5 but reject [5-7] -> 8.
+ // 1. The BASE/FULL layout is unchanged.
+ // 2. The current point is inside its input range.
+ // 3. The proposal is exactly input_end + 1.
+ accept = compaction.input_versions_size() == 2 &&
+ recorded_compaction.base_compaction_cnt() ==
stats.base_compaction_cnt() &&
+ compaction.input_versions(0) <= stats.cumulative_point()
&&
+ stats.cumulative_point() <= compaction.input_versions(1)
&&
+ compaction.output_cumulative_point() ==
compaction.input_versions(1) + 1;
+ }
+ // Legacy EMPTY has no output range, so its START snapshot must match
current stats.
+ }
+ if (accept) {
+ return true;
+ }
+
+ INSTANCE_LOG(INFO) << "ignore stale cumulative point=" <<
compaction.output_cumulative_point()
+ << ", tablet_id=" << tablet_id << ", job_id=" <<
compaction.id()
+ << ", base_cnt=" <<
recorded_compaction.base_compaction_cnt() << "(start),"
+ << snapshot_base_cnt << "(finish)," <<
stats.base_compaction_cnt()
+ << "(stats); cumu_cnt=" <<
recorded_compaction.cumulative_compaction_cnt()
+ << "(start)," << snapshot_cumu_cnt << "(finish),"
+ << stats.cumulative_compaction_cnt()
+ << "(stats); full_cnt=" << stats.full_compaction_cnt()
+ << "; finish_has_counters=" << finish_has_counters;
+ return false;
+}
+
int compaction_update_tablet_stats(const TabletCompactionJobPB& compaction,
TabletStatsPB* stats,
- MetaServiceCode& code, std::string& msg,
int64_t now) {
+ bool accept_cumulative_point_proposal,
MetaServiceCode& code,
+ std::string& msg, int64_t now) {
if (compaction.type() == TabletCompactionJobPB::EMPTY_CUMULATIVE) {
stats->set_cumulative_compaction_cnt(stats->cumulative_compaction_cnt() + 1);
- stats->set_cumulative_point(compaction.output_cumulative_point());
+ if (accept_cumulative_point_proposal) {
+ stats->set_cumulative_point(
+ std::max(stats->cumulative_point(),
compaction.output_cumulative_point()));
+ }
stats->set_last_cumu_compaction_time_ms(now * 1000);
} else if (compaction.type() == TabletCompactionJobPB::CUMULATIVE) {
// clang-format off
stats->set_cumulative_compaction_cnt(stats->cumulative_compaction_cnt() + 1);
- if (compaction.output_cumulative_point() > stats->cumulative_point()) {
- // After supporting parallel cumu compaction, compaction with
older cumu point may be committed after
- // new cumu point has been set, MUST NOT set cumu point back to
old value
- stats->set_cumulative_point(compaction.output_cumulative_point());
- }
+ int64_t output_cumulative_point = stats->cumulative_point();
+ if (accept_cumulative_point_proposal) {
+ output_cumulative_point =
+ std::max(compaction.output_cumulative_point(),
output_cumulative_point);
+ }
+ if (compaction.input_versions_size() == 2 &&
+ output_cumulative_point > compaction.input_versions(0) &&
+ output_cumulative_point <= compaction.input_versions(1)) {
+ LOG_WARNING("cumulative point falls inside cumulative compaction
input range")
+ .tag("job_id", compaction.id())
+ .tag("cumulative_point", output_cumulative_point)
+ .tag("input_start_version", compaction.input_versions(0))
+ .tag("input_end_version", compaction.input_versions(1));
+ output_cumulative_point = compaction.input_versions(1) + 1;
+ }
+ stats->set_cumulative_point(output_cumulative_point);
stats->set_num_rows(stats->num_rows() + (compaction.num_output_rows()
- compaction.num_input_rows()));
stats->set_data_size(stats->data_size() +
(compaction.size_output_rowsets() - compaction.size_input_rowsets()));
stats->set_num_rowsets(stats->num_rowsets() +
(compaction.num_output_rowsets() - compaction.num_input_rowsets()));
@@ -832,6 +908,16 @@ int compaction_update_tablet_stats(const
TabletCompactionJobPB& compaction, Tabl
} else if (compaction.type() == TabletCompactionJobPB::BASE) {
// clang-format off
stats->set_base_compaction_cnt(stats->base_compaction_cnt() + 1);
+ if (compaction.input_versions_size() == 2 &&
+ stats->cumulative_point() > compaction.input_versions(0) &&
+ stats->cumulative_point() <= compaction.input_versions(1)) {
+ LOG_WARNING("cumulative point falls inside base compaction input
range")
+ .tag("job_id", compaction.id())
+ .tag("cumulative_point", stats->cumulative_point())
+ .tag("input_start_version", compaction.input_versions(0))
+ .tag("input_end_version", compaction.input_versions(1));
+ stats->set_cumulative_point(compaction.input_versions(1) + 1);
+ }
stats->set_num_rows(stats->num_rows() + (compaction.num_output_rows()
- compaction.num_input_rows()));
stats->set_data_size(stats->data_size() +
(compaction.size_output_rowsets() - compaction.size_input_rowsets()));
stats->set_num_rowsets(stats->num_rowsets() +
(compaction.num_output_rowsets() - compaction.num_input_rowsets()));
@@ -1024,6 +1110,14 @@ void process_compaction_job(MetaServiceCode& code,
std::string& msg, std::string
// 4. remove compaction job
//
//==========================================================================
+ if ((compaction.type() == TabletCompactionJobPB::CUMULATIVE ||
+ compaction.type() == TabletCompactionJobPB::EMPTY_CUMULATIVE) &&
+ compaction.has_base_compaction_cnt() !=
compaction.has_cumulative_compaction_cnt()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "incomplete compaction counters for cumulative point proposal";
+ return;
+ }
+
// Update tablet stats
//==========================================================================
auto stats = response->mutable_stats();
@@ -1068,7 +1162,10 @@ void process_compaction_job(MetaServiceCode& code,
std::string& msg, std::string
}
}
- if (compaction_update_tablet_stats(compaction, stats, code, msg, now) ==
-1) {
+ const bool accept_cumulative_point_proposal =
should_accept_cumulative_point(
+ instance_id, tablet_id, compaction, *recorded_compaction, *stats);
+ if (compaction_update_tablet_stats(compaction, stats,
accept_cumulative_point_proposal, code,
+ msg, now) == -1) {
return;
}
diff --git a/cloud/test/meta_service_job_test.cpp
b/cloud/test/meta_service_job_test.cpp
index b2b37084040..58ce61cfce1 100644
--- a/cloud/test/meta_service_job_test.cpp
+++ b/cloud/test/meta_service_job_test.cpp
@@ -120,7 +120,8 @@ void start_compaction_job(MetaService* meta_service,
int64_t tablet_id, const st
const std::string& initiator, int
base_compaction_cnt,
int cumu_compaction_cnt,
TabletCompactionJobPB::CompactionType type,
StartTabletJobResponse& res,
- std::pair<int64_t, int64_t> input_version = {0, 0}) {
+ std::pair<int64_t, int64_t> input_version = {0, 0},
+ bool check_input_versions_range = true) {
brpc::Controller cntl;
StartTabletJobRequest req;
req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);
@@ -136,7 +137,7 @@ void start_compaction_job(MetaService* meta_service,
int64_t tablet_id, const st
if (input_version.second > 0) {
compaction->add_input_versions(input_version.first);
compaction->add_input_versions(input_version.second);
- compaction->set_check_input_versions_range(true);
+ compaction->set_check_input_versions_range(check_input_versions_range);
}
meta_service->start_tablet_job(&cntl, &req, &res, nullptr);
};
@@ -175,6 +176,45 @@ void finish_compaction_job(MetaService* meta_service,
int64_t tablet_id, const s
meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
}
+void finish_rowset_compaction_job(
+ MetaService* meta_service, int64_t tablet_id, const std::string&
job_id,
+ TabletCompactionJobPB::CompactionType type, const
doris::RowsetMetaCloudPB& output_rowset,
+ int num_input_rowsets, int64_t output_cumulative_point,
FinishTabletJobResponse& res,
+ int64_t proposal_base_compaction_cnt = 0, int64_t
proposal_cumulative_compaction_cnt = 0,
+ bool include_proposal_snapshot = true) {
+ brpc::Controller cntl;
+ FinishTabletJobRequest req;
+ req.set_action(FinishTabletJobRequest::COMMIT);
+ req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);
+ auto* compaction = req.mutable_job()->add_compaction();
+ compaction->set_id(job_id);
+ compaction->set_initiator("BE1");
+ compaction->set_type(type);
+ if (include_proposal_snapshot) {
+ compaction->set_base_compaction_cnt(proposal_base_compaction_cnt);
+
compaction->set_cumulative_compaction_cnt(proposal_cumulative_compaction_cnt);
+ }
+ compaction->add_input_versions(output_rowset.start_version());
+ compaction->add_input_versions(output_rowset.end_version());
+ compaction->add_output_versions(output_rowset.end_version());
+ compaction->add_txn_id(output_rowset.txn_id());
+ compaction->add_output_rowset_ids(output_rowset.rowset_id_v2());
+ compaction->set_output_cumulative_point(output_cumulative_point);
+ compaction->set_num_input_rows(num_input_rowsets * 100);
+ compaction->set_num_output_rows(output_rowset.num_rows());
+ compaction->set_size_input_rowsets(num_input_rowsets * 10000);
+ compaction->set_size_output_rowsets(output_rowset.total_disk_size());
+ compaction->set_num_input_segments(num_input_rowsets);
+ compaction->set_num_output_segments(output_rowset.num_segments());
+ compaction->set_num_input_rowsets(num_input_rowsets);
+ compaction->set_num_output_rowsets(1);
+ compaction->set_index_size_input_rowsets(num_input_rowsets * 5000);
+ compaction->set_segment_size_input_rowsets(num_input_rowsets * 5000);
+ compaction->set_index_size_output_rowsets(output_rowset.index_disk_size());
+
compaction->set_segment_size_output_rowsets(output_rowset.data_disk_size());
+ meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
+}
+
void get_tablet_stats(MetaService* meta_service, int64_t tablet_id,
TabletStatsPB& stats) {
brpc::Controller cntl;
GetTabletStatsRequest req;
@@ -728,6 +768,8 @@ TEST(MetaServiceJobTest, ProcessCompactionArguments) {
// Prepare job kv
recorded_compaction->set_expiration(::time(nullptr) + 10);
+ recorded_compaction->set_base_compaction_cnt(0);
+ recorded_compaction->set_cumulative_compaction_cnt(0);
job_val = recorded_job.SerializeAsString();
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
txn->put(job_key, job_val);
@@ -754,6 +796,12 @@ TEST(MetaServiceJobTest, ProcessCompactionArguments) {
<< res.status().msg();
compaction->set_type(TabletCompactionJobPB::EMPTY_CUMULATIVE);
+ compaction->set_base_compaction_cnt(0);
+ meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
+ ASSERT_EQ(res.status().code(), MetaServiceCode::INVALID_ARGUMENT) <<
res.status().msg();
+ EXPECT_NE(res.status().msg().find("incomplete compaction counters"),
std::string::npos)
+ << res.status().msg();
+ compaction->clear_base_compaction_cnt();
meta_service->finish_tablet_job(&cntl, &req, &res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg();
}
@@ -988,8 +1036,8 @@ TEST(MetaServiceJobTest, CompactionJobTest) {
req.mutable_job()->mutable_idx()->set_index_id(index_id);
req.mutable_job()->mutable_idx()->set_partition_id(partition_id);
req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);
- compaction->set_base_compaction_cnt(10);
- compaction->set_cumulative_compaction_cnt(20);
+ compaction->set_base_compaction_cnt(9);
+ compaction->set_cumulative_compaction_cnt(19);
// Action is not set
meta_service->finish_tablet_job(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
&req, &res, nullptr);
@@ -1045,8 +1093,8 @@ TEST(MetaServiceJobTest, CompactionJobTest) {
compaction->set_segment_size_output_rowsets(dist(rng));
compaction->set_type(type);
- tablet_stats_pb.set_cumulative_compaction_cnt(dist(rng));
- tablet_stats_pb.set_base_compaction_cnt(dist(rng));
+ tablet_stats_pb.set_cumulative_compaction_cnt(19);
+ tablet_stats_pb.set_base_compaction_cnt(9);
tablet_stats_pb.set_cumulative_point(tablet_meta_pb.cumulative_layer_point());
// MUST let data stats be larger than input data size
tablet_stats_pb.set_num_rows(dist(rng) + compaction->num_input_rows());
@@ -1068,7 +1116,9 @@ TEST(MetaServiceJobTest, CompactionJobTest) {
ASSERT_NE(res.status().msg().find("invalid input"), std::string::npos);
// Provide input and output rowset info
- int64_t input_version_start = dist(rng);
+ int64_t input_version_start = type == TabletCompactionJobPB::BASE
+ ?
tablet_meta_pb.cumulative_layer_point()
+ : dist(rng);
int64_t input_version_end = input_version_start + 100;
compaction->add_input_versions(input_version_start);
compaction->add_input_versions(input_version_end);
@@ -1833,8 +1883,8 @@ TEST(MetaServiceJobTest,
DeleteBitmapUpdateLockCompatibilityTest) {
compaction->set_size_output_rowsets(dist(rng));
compaction->set_type(type);
- tablet_stats_pb.set_cumulative_compaction_cnt(dist(rng));
- tablet_stats_pb.set_base_compaction_cnt(dist(rng));
+ tablet_stats_pb.set_cumulative_compaction_cnt(19);
+ tablet_stats_pb.set_base_compaction_cnt(9);
tablet_stats_pb.set_cumulative_point(tablet_meta_pb.cumulative_layer_point());
// MUST let data stats be larger than input data size
tablet_stats_pb.set_num_rows(dist(rng) + compaction->num_input_rows());
@@ -3398,8 +3448,8 @@ void testCompactionJobWithMoWTest(int lock_version) {
compaction->set_size_output_rowsets(dist(rng));
compaction->set_type(type);
- tablet_stats_pb.set_cumulative_compaction_cnt(dist(rng));
- tablet_stats_pb.set_base_compaction_cnt(dist(rng));
+ tablet_stats_pb.set_cumulative_compaction_cnt(19);
+ tablet_stats_pb.set_base_compaction_cnt(9);
tablet_stats_pb.set_cumulative_point(tablet_meta_pb.cumulative_layer_point());
// MUST let data stats be larger than input data size
tablet_stats_pb.set_num_rows(dist(rng) + compaction->num_input_rows());
@@ -4304,6 +4354,8 @@ TEST(MetaServiceJobTest, ConcurrentCompactionTest) {
compaction->set_id("job5");
compaction->set_initiator("BE2");
compaction->set_type(TabletCompactionJobPB::CUMULATIVE);
+ compaction->set_base_compaction_cnt(0);
+ compaction->set_cumulative_compaction_cnt(0);
compaction->add_input_versions(5);
compaction->add_input_versions(10);
compaction->add_txn_id(output_rowset.txn_id());
@@ -4631,6 +4683,276 @@ TEST(MetaServiceJobTest, ParallelCumuCompactionTest) {
ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
}
+TEST(MetaServiceJobTest, LegacyParallelCumuFinishValidatesCurrentInputRange) {
+ auto meta_service = get_meta_service();
+
+ auto sp = SyncPoint::get_instance();
+ DORIS_CLOUD_DEFER {
+ SyncPoint::get_instance()->clear_all_call_backs();
+ };
+ sp->set_call_back("get_instance_id", [&](auto&& args) {
+ auto* ret = try_any_cast_ret<std::string>(args);
+ ret->first = instance_id;
+ ret->second = true;
+ });
+ sp->enable_processing();
+
+ auto run_case = [&](int64_t tablet_id, bool higher_first, int64_t
lower_proposal,
+ int64_t expected_final_point) {
+ constexpr int64_t table_id = 1;
+ constexpr int64_t index_id = 2;
+ constexpr int64_t partition_id = 3;
+ ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), table_id,
index_id, partition_id,
+ tablet_id, false));
+
+ std::vector<doris::RowsetMetaCloudPB> input_rowsets;
+ for (int64_t version = 2; version <= 7; ++version) {
+ input_rowsets.push_back(create_rowset(tablet_id, version,
version));
+ }
+ insert_rowsets(meta_service->txn_kv().get(), table_id, index_id,
partition_id, tablet_id,
+ input_rowsets);
+
+ StartTabletJobResponse start_res;
+ start_compaction_job(meta_service.get(), tablet_id, "lower", "BE1", 0,
0,
+ TabletCompactionJobPB::CUMULATIVE, start_res, {2,
4});
+ ASSERT_EQ(start_res.status().code(), MetaServiceCode::OK);
+ start_res.Clear();
+ start_compaction_job(meta_service.get(), tablet_id, "higher", "BE1",
0, 0,
+ TabletCompactionJobPB::CUMULATIVE, start_res, {5,
7});
+ ASSERT_EQ(start_res.status().code(), MetaServiceCode::OK);
+
+ auto lower_output = create_rowset(tablet_id, 2, 4);
+ auto higher_output = create_rowset(tablet_id, 5, 7);
+ for (const auto* output : {&lower_output, &higher_output}) {
+ CreateRowsetResponse rowset_res;
+ prepare_rowset(meta_service.get(), *output, rowset_res);
+ ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
+ commit_rowset(meta_service.get(), *output, rowset_res);
+ ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
+ }
+
+ // Legacy regular and index-change cumulative compactions omit
counters from FINISH.
+ FinishTabletJobResponse finish_res;
+ auto finish = [&](const std::string& job_id, const auto& output,
int64_t proposal,
+ int64_t expected_point) {
+ finish_res.Clear();
+ finish_rowset_compaction_job(meta_service.get(), tablet_id, job_id,
+ TabletCompactionJobPB::CUMULATIVE,
output, 3, proposal,
+ finish_res, 0, 0, false);
+ ASSERT_EQ(finish_res.status().code(), MetaServiceCode::OK);
+ EXPECT_EQ(finish_res.stats().cumulative_point(), expected_point);
+ };
+ if (higher_first) {
+ finish("higher", higher_output, 8, 2);
+ finish("lower", lower_output, lower_proposal,
expected_final_point);
+ } else {
+ finish("lower", lower_output, lower_proposal, 5);
+ finish("higher", higher_output, 8, expected_final_point);
+ }
+
+ TabletStatsPB stats;
+ get_tablet_stats(meta_service.get(), tablet_id, stats);
+ EXPECT_EQ(stats.cumulative_point(), expected_final_point);
+ };
+
+ run_case(40001, false, 5, 8);
+ run_case(40004, true, 5, 5);
+ run_case(40005, true, 6, 2);
+}
+
+TEST(MetaServiceJobTest, ParallelCumuCompactionUsesPointProposalSnapshot) {
+ auto meta_service = get_meta_service();
+
+ auto sp = SyncPoint::get_instance();
+ DORIS_CLOUD_DEFER {
+ SyncPoint::get_instance()->clear_all_call_backs();
+ };
+ sp->set_call_back("get_instance_id", [&](auto&& args) {
+ auto* ret = try_any_cast_ret<std::string>(args);
+ ret->first = instance_id;
+ ret->second = true;
+ });
+ sp->enable_processing();
+
+ constexpr int64_t table_id = 1;
+ constexpr int64_t index_id = 2;
+ constexpr int64_t partition_id = 3;
+ auto run_case = [&](int64_t tablet_id, int64_t lower_proposal_cumu_cnt,
+ int64_t expected_cumulative_point) {
+ ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), table_id,
index_id, partition_id,
+ tablet_id, false));
+
+ std::vector<doris::RowsetMetaCloudPB> input_rowsets;
+ for (int64_t version = 2; version <= 7; ++version) {
+ input_rowsets.push_back(create_rowset(tablet_id, version,
version));
+ }
+ insert_rowsets(meta_service->txn_kv().get(), table_id, index_id,
partition_id, tablet_id,
+ input_rowsets);
+
+ StartTabletJobResponse start_res;
+ start_compaction_job(meta_service.get(), tablet_id, "lower", "BE1", 0,
0,
+ TabletCompactionJobPB::CUMULATIVE, start_res, {2,
4});
+ ASSERT_EQ(start_res.status().code(), MetaServiceCode::OK);
+ start_res.Clear();
+ start_compaction_job(meta_service.get(), tablet_id, "higher", "BE1",
0, 0,
+ TabletCompactionJobPB::CUMULATIVE, start_res, {5,
7});
+ ASSERT_EQ(start_res.status().code(), MetaServiceCode::OK);
+
+ auto lower_output = create_rowset(tablet_id, 2, 4);
+ auto higher_output = create_rowset(tablet_id, 5, 7);
+ for (const auto* output : {&lower_output, &higher_output}) {
+ CreateRowsetResponse rowset_res;
+ prepare_rowset(meta_service.get(), *output, rowset_res);
+ ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
+ commit_rowset(meta_service.get(), *output, rowset_res);
+ ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
+ }
+
+ FinishTabletJobResponse finish_res;
+ finish_rowset_compaction_job(meta_service.get(), tablet_id, "higher",
+ TabletCompactionJobPB::CUMULATIVE,
higher_output, 3, 2,
+ finish_res);
+ ASSERT_EQ(finish_res.status().code(), MetaServiceCode::OK);
+ EXPECT_EQ(finish_res.stats().cumulative_point(), 2);
+
+ finish_res.Clear();
+ finish_rowset_compaction_job(meta_service.get(), tablet_id, "lower",
+ TabletCompactionJobPB::CUMULATIVE,
lower_output, 3, 6,
+ finish_res, 0, lower_proposal_cumu_cnt);
+ ASSERT_EQ(finish_res.status().code(), MetaServiceCode::OK);
+ EXPECT_EQ(finish_res.stats().cumulative_point(),
expected_cumulative_point);
+
+ TabletStatsPB stats;
+ get_tablet_stats(meta_service.get(), tablet_id, stats);
+ EXPECT_EQ(stats.cumulative_point(), expected_cumulative_point);
+ };
+
+ // Both jobs calculated from the START layout: reject the lower job's
stale proposal.
+ run_case(40002, 0, 2);
+ // The lower job recalculated after applying the higher result: accept its
fresh proposal.
+ run_case(40003, 1, 6);
+}
+
+TEST(MetaServiceJobTest,
SerialCumuPointAdvanceIgnoresUnrelatedBaseCounterChange) {
+ auto meta_service = get_meta_service();
+
+ auto sp = SyncPoint::get_instance();
+ DORIS_CLOUD_DEFER {
+ SyncPoint::get_instance()->clear_all_call_backs();
+ };
+ sp->set_call_back("get_instance_id", [&](auto&& args) {
+ auto* ret = try_any_cast_ret<std::string>(args);
+ ret->first = instance_id;
+ ret->second = true;
+ });
+ sp->enable_processing();
+
+ constexpr int64_t table_id = 1;
+ constexpr int64_t index_id = 2;
+ constexpr int64_t partition_id = 3;
+ constexpr int64_t tablet_id = 40006;
+ ASSERT_NO_FATAL_FAILURE(
+ create_tablet(meta_service.get(), table_id, index_id,
partition_id, tablet_id, false));
+
+ std::vector<doris::RowsetMetaCloudPB> input_rowsets;
+ for (int64_t version = 2; version <= 4; ++version) {
+ input_rowsets.push_back(create_rowset(tablet_id, version, version));
+ }
+ insert_rowsets(meta_service->txn_kv().get(), table_id, index_id,
partition_id, tablet_id,
+ input_rowsets);
+
+ StartTabletJobResponse start_res;
+ start_compaction_job(meta_service.get(), tablet_id, "cumu", "BE1", 0, 0,
+ TabletCompactionJobPB::CUMULATIVE, start_res, {2, 4},
false);
+ ASSERT_EQ(start_res.status().code(), MetaServiceCode::OK);
+ start_res.Clear();
+ start_compaction_job(meta_service.get(), tablet_id, "base", "BE1", 0, 0,
+ TabletCompactionJobPB::BASE, start_res, {0, 1},
false);
+ ASSERT_EQ(start_res.status().code(), MetaServiceCode::OK);
+
+ auto cumu_output = create_rowset(tablet_id, 2, 4);
+ auto base_output = create_rowset(tablet_id, 0, 1);
+ for (const auto* output : {&cumu_output, &base_output}) {
+ CreateRowsetResponse rowset_res;
+ prepare_rowset(meta_service.get(), *output, rowset_res);
+ ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
+ commit_rowset(meta_service.get(), *output, rowset_res);
+ ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
+ }
+
+ FinishTabletJobResponse finish_res;
+ finish_rowset_compaction_job(meta_service.get(), tablet_id, "base",
TabletCompactionJobPB::BASE,
+ base_output, 1, 2, finish_res);
+ ASSERT_EQ(finish_res.status().code(), MetaServiceCode::OK);
+ ASSERT_EQ(finish_res.stats().base_compaction_cnt(), 1);
+ ASSERT_EQ(finish_res.stats().cumulative_point(), 2);
+
+ finish_res.Clear();
+ finish_rowset_compaction_job(meta_service.get(), tablet_id, "cumu",
+ TabletCompactionJobPB::CUMULATIVE,
cumu_output, 3, 5, finish_res);
+ ASSERT_EQ(finish_res.status().code(), MetaServiceCode::OK);
+ EXPECT_EQ(finish_res.stats().cumulative_point(), 5);
+}
+
+TEST(MetaServiceJobTest, BaseCompactionAdvancesPointPastOutput) {
+ auto meta_service = get_meta_service();
+
+ auto sp = SyncPoint::get_instance();
+ DORIS_CLOUD_DEFER {
+ SyncPoint::get_instance()->clear_all_call_backs();
+ };
+ sp->set_call_back("get_instance_id", [&](auto&& args) {
+ auto* ret = try_any_cast_ret<std::string>(args);
+ ret->first = instance_id;
+ ret->second = true;
+ });
+ sp->enable_processing();
+
+ constexpr int64_t table_id = 1;
+ constexpr int64_t index_id = 2;
+ constexpr int64_t partition_id = 3;
+ constexpr int64_t tablet_id = 40003;
+ ASSERT_NO_FATAL_FAILURE(
+ create_tablet(meta_service.get(), table_id, index_id,
partition_id, tablet_id, false));
+
+ std::vector<doris::RowsetMetaCloudPB> input_rowsets =
{create_rowset(tablet_id, 2, 4),
+
create_rowset(tablet_id, 5, 7)};
+ insert_rowsets(meta_service->txn_kv().get(), table_id, index_id,
partition_id, tablet_id,
+ input_rowsets);
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ auto stats_key = stats_tablet_key({instance_id, table_id, index_id,
partition_id, tablet_id});
+ std::string stats_value;
+ ASSERT_EQ(txn->get(stats_key, &stats_value), TxnErrorCode::TXN_OK);
+ TabletStatsPB stats;
+ ASSERT_TRUE(stats.ParseFromString(stats_value));
+ stats.set_cumulative_point(6);
+ txn->put(stats_key, stats.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ StartTabletJobResponse start_res;
+ start_compaction_job(meta_service.get(), tablet_id, "base", "BE1", 0, 0,
+ TabletCompactionJobPB::BASE, start_res, {2, 7});
+ ASSERT_EQ(start_res.status().code(), MetaServiceCode::OK);
+
+ auto output_rowset = create_rowset(tablet_id, 2, 7);
+ CreateRowsetResponse rowset_res;
+ prepare_rowset(meta_service.get(), output_rowset, rowset_res);
+ ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
+ commit_rowset(meta_service.get(), output_rowset, rowset_res);
+ ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
+
+ FinishTabletJobResponse finish_res;
+ finish_rowset_compaction_job(meta_service.get(), tablet_id, "base",
TabletCompactionJobPB::BASE,
+ output_rowset, 2, 6, finish_res);
+ ASSERT_EQ(finish_res.status().code(), MetaServiceCode::OK);
+ EXPECT_EQ(finish_res.stats().cumulative_point(), 8);
+
+ get_tablet_stats(meta_service.get(), tablet_id, stats);
+ EXPECT_EQ(stats.cumulative_point(), 8);
+}
+
// Plan A regression test: EMPTY_CUMULATIVE must be considered the same
conflict family as
// CUMULATIVE so that an EMPTY_CUMULATIVE submitted while a real CUMULATIVE is
still active on the
// same tablet is rejected with JOB_TABLET_BUSY. Otherwise EMPTY_CUMULATIVE
could advance
@@ -5198,8 +5520,8 @@ TEST(MetaServiceJobTest, IdempotentCompactionJob) {
compaction->set_segment_size_output_rowsets(dist(rng));
compaction->set_type(type);
- tablet_stats_pb.set_cumulative_compaction_cnt(dist(rng));
- tablet_stats_pb.set_base_compaction_cnt(dist(rng));
+ tablet_stats_pb.set_cumulative_compaction_cnt(19);
+ tablet_stats_pb.set_base_compaction_cnt(9);
tablet_stats_pb.set_cumulative_point(tablet_meta_pb.cumulative_layer_point());
// MUST let data stats be larger than input data size
tablet_stats_pb.set_num_rows(dist(rng) + compaction->num_input_rows());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
