This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new cf1288086ae branch-4.1: [fix](cloud) Correct cumulative point
advancement when enable parallel cumulative compaction (#65915) (#67425)
cf1288086ae is described below
commit cf1288086ae5b75397ac4b370f911f5552ff00a1
Author: meiyi <[email protected]>
AuthorDate: Mon Sep 7 22:40:59 2026 +0800
branch-4.1: [fix](cloud) Correct cumulative point advancement when enable
parallel cumulative compaction (#65915) (#67425)
pick https://github.com/apache/doris/pull/65915
---
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 | 461 +++++++++++++++++++++
.../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, 1186 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 ee8784fb8c1..cd61ea9c426 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"
@@ -44,7 +47,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;
@@ -65,11 +69,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;
@@ -95,7 +99,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");
@@ -151,7 +158,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()) {
@@ -245,12 +252,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();
@@ -263,6 +291,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);
@@ -308,6 +338,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;
@@ -375,20 +438,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
@@ -492,34 +545,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 =
@@ -533,29 +625,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());
@@ -569,6 +725,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
@@ -588,8 +747,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;
@@ -617,14 +774,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 473e5a916b2..28f88c6432f 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>
@@ -49,23 +50,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 001b9d64f4b..66e008dba84 100644
--- a/be/src/cloud/cloud_index_change_compaction.cpp
+++ b/be/src/cloud/cloud_index_change_compaction.cpp
@@ -257,8 +257,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);
@@ -368,6 +373,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);
}
@@ -389,20 +397,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 37222b29b59..0112fc2d279 100644
--- a/be/src/storage/compaction/compaction.cpp
+++ b/be/src/storage/compaction/compaction.cpp
@@ -1610,6 +1610,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 b4d609a3f00..4bc0b4e3ad2 100644
--- a/be/src/storage/compaction/compaction.h
+++ b/be/src/storage/compaction/compaction.h
@@ -254,6 +254,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 f7916ab511e..6c83d12f254 100644
--- a/be/test/cloud/cloud_compaction_test.cpp
+++ b/be/test/cloud/cloud_compaction_test.cpp
@@ -22,20 +22,26 @@
#include <gtest/gtest.h>
#include <memory>
+#include <mutex>
+#include <unordered_map>
#include "cloud/cloud_base_compaction.h"
#include "cloud/cloud_cluster_info.h"
+#include "cloud/cloud_cumulative_compaction.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
#include "cloud/config.h"
+#include "cpp/sync_point.h"
#include "json2pb/json_to_pb.h"
+#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "storage/olap_common.h"
#include "storage/rowset/rowset_factory.h"
#include "storage/rowset/rowset_meta.h"
#include "storage/storage_policy.h"
#include "storage/tablet/tablet_meta.h"
#include "util/defer_op.h"
+#include "util/time.h"
#include "util/uid_util.h"
namespace doris {
@@ -218,6 +224,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;
+}
+
class TestableCloudCompaction : public CloudCompactionMixin {
public:
TestableCloudCompaction(CloudStorageEngine& engine, CloudTabletSPtr tablet)
@@ -236,6 +251,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; }
@@ -243,6 +263,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 663dd23aebb..c1654a96914 100644
--- a/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
+++ b/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
@@ -326,6 +326,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 f357418b4fc..1553624851f 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()));
@@ -1025,6 +1111,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();
@@ -1069,7 +1163,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]
