This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 2134b43d9c0 branch-4.2: [fix](cloud) Skip stale delete bitmap
aggregation #68084 (#69306)
2134b43d9c0 is described below
commit 2134b43d9c0f2129823a131ebd7499d659ccf6c2
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Oct 21 10:49:55 2026 +0800
branch-4.2: [fix](cloud) Skip stale delete bitmap aggregation #68084
(#69306)
Cherry-picked from #68084
Co-authored-by: meiyi <[email protected]>
---
cloud/src/meta-service/meta_service.cpp | 106 +++++++++++---
cloud/test/meta_service_test.cpp | 162 ++++++++++++++++++++-
...t_cloud_mow_compaction_agg_bitmap_two_be.groovy | 145 ++++++++++++++++++
3 files changed, 389 insertions(+), 24 deletions(-)
diff --git a/cloud/src/meta-service/meta_service.cpp
b/cloud/src/meta-service/meta_service.cpp
index 2a6e71fe35d..b5171c53bf2 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -3905,7 +3905,51 @@ static bool
check_delete_bitmap_point_delete(MetaServiceCode& code, std::string&
return true;
}
-void _write_delete_bitmap_kvs(MetaServiceCode& code, std::string& msg,
std::stringstream& ss,
+// False stops the whole request, including cleanup. An obsolete aggregation
is a successful no-op.
+static bool check_pre_rowset_aggregation(MetaServiceCode& code, std::string&
msg, Transaction* txn,
+ const UpdateDeleteBitmapRequest*
request,
+ const std::string& instance_id) {
+ if (!request->without_lock() ||
+ request->lock_id() != COMPACTION_WITHOUT_LOCK_DELETE_BITMAP_LOCK_ID ||
+ !request->has_pre_rowset_agg_start_version() ||
+ !request->has_pre_rowset_agg_end_version()) {
+ return true;
+ }
+
+ auto tablet_id = request->tablet_id();
+ auto start = request->pre_rowset_agg_start_version();
+ auto end = request->pre_rowset_agg_end_version();
+ auto key = meta_rowset_key({instance_id, tablet_id, end});
+ std::string value;
+ // Compaction removes/replaces this live key even with versioned metadata
enabled.
+ // A non-snapshot read conflicts with a compaction that invalidates the
output rowset
+ // before this transaction commits. Historical versioned rowsets cannot
establish liveness.
+ auto err = txn->get(key, &value);
+ if (err != TxnErrorCode::TXN_OK && err != TxnErrorCode::TXN_KEY_NOT_FOUND)
{
+ code = cast_as<ErrCategory::READ>(err);
+ msg = fmt::format("failed to get aggregation output rowset,
tablet_id={}, key={}, err={}",
+ tablet_id, hex(key), err);
+ return false;
+ }
+ RowsetMetaCloudPB rowset;
+ if (err == TxnErrorCode::TXN_OK && !rowset.ParseFromString(value)) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = fmt::format("malformed aggregation output rowset, tablet_id={},
key={}", tablet_id,
+ hex(key));
+ return false;
+ }
+ if (err == TxnErrorCode::TXN_KEY_NOT_FOUND || rowset.start_version() !=
start ||
+ rowset.end_version() != end) {
+ LOG(INFO) << "skip obsolete pre rowset delete bitmap aggregation,
instance_id="
+ << instance_id << " tablet_id=" << tablet_id << "
start_version=" << start
+ << " end_version=" << end;
+ return false;
+ }
+
TEST_SYNC_POINT_CALLBACK("update_delete_bitmap:check_pre_rowset_aggregation",
request);
+ return true;
+}
+
+bool _write_delete_bitmap_kvs(MetaServiceCode& code, std::string& msg,
std::stringstream& ss,
std::shared_ptr<TxnKv> txn_kv,
std::unique_ptr<Transaction>& txn,
std::string& use_version, std::set<std::string>&
non_exist_rowset_ids,
size_t i, std::string& log, KVStats& stats,
@@ -3939,7 +3983,7 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
<< " txn_size=" << txn->approximate_bytes();
msg = ss.str();
g_bvar_update_delete_bitmap_fail_counter << 1;
- return;
+ return false;
}
stats.get_bytes += txn->get_bytes();
stats.put_bytes += txn->put_bytes();
@@ -3953,7 +3997,10 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
if (err != TxnErrorCode::TXN_OK) {
code = cast_as<ErrCategory::CREATE>(err);
msg = "failed to init txn";
- return;
+ return false;
+ }
+ if (!check_pre_rowset_aggregation(code, msg, txn.get(), request,
instance_id)) {
+ return false;
}
if (!without_lock) {
std::string lock_key =
meta_delete_bitmap_update_lock_key({instance_id, table_id, -1});
@@ -3964,7 +4011,7 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
LOG(WARNING) << "failed to check delete bitmap lock,
table_id=" << table_id
<< " request lock_id=" << request->lock_id()
<< " request initiator=" << request->initiator()
<< " msg " << msg;
- return;
+ return false;
}
}
}
@@ -3973,7 +4020,7 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
if (config::enable_update_delete_bitmap_kv_check) {
if (!check_delete_bitmap_kv_exists(code, msg, txn, key, instance_id,
tablet_id,
request->lock_id())) {
- return;
+ return false;
}
}
#endif
@@ -3985,7 +4032,7 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
LOG(INFO) << "skip update delete bitmap, rowset_id=" <<
request->rowset_ids(i)
<< " version=" << request->pre_rowset_versions(i)
<< " tablet_id=" << tablet_id << " because the rowset
does not exist";
- return;
+ return true;
}
TxnErrorCode err = TxnErrorCode::TXN_OK;
@@ -4001,7 +4048,7 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
<< " err=" << err;
msg = ss.str();
code = cast_as<ErrCategory::READ>(err);
- return;
+ return false;
}
if (err == TxnErrorCode::TXN_OK &&
!rs.ParseFromArray(rowset_val.data(), rowset_val.size())) {
@@ -4009,7 +4056,7 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
ss << "malformed rowset meta, unable to deserialize,
tablet_id=" << tablet_id
<< " key=" << hex(rowset_key);
msg = ss.str();
- return;
+ return false;
}
} else {
CloneChainReader reader(instance_id, resource_mgr);
@@ -4020,7 +4067,7 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
<< " version=" << request->pre_rowset_versions(i);
msg = ss.str();
code = cast_as<ErrCategory::READ>(err);
- return;
+ return false;
}
}
if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
@@ -4028,14 +4075,14 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
LOG(INFO) << "skip update delete bitmap, rowset_id=" <<
request->rowset_ids(i)
<< " version=" << request->pre_rowset_versions(i)
<< " tablet_id=" << tablet_id << " because the rowset is
not exist";
- return;
+ return true;
}
if (rs.rowset_id_v2() != request->rowset_ids(i)) {
LOG(INFO) << "skip update delete bitmap, rowset_id=" <<
request->rowset_ids(i)
<< " version=" << request->pre_rowset_versions(i)
<< " tablet_id=" << tablet_id << " because the rowset is
not exist";
non_exist_rowset_ids.emplace(request->rowset_ids(i));
- return;
+ return true;
}
}
// remove first
@@ -4045,7 +4092,7 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
std::optional<uint16_t> max_blob_sequence;
if (!check_delete_bitmap_point_delete(code, msg, txn, key, request, i,
point_delete,
delete_bitmap_exists,
max_blob_sequence)) {
- return;
+ return false;
}
if (point_delete) {
if (delete_bitmap_exists) {
@@ -4081,12 +4128,15 @@ void _write_delete_bitmap_kvs(MetaServiceCode& code,
std::string& msg, std::stri
VLOG_DEBUG << "xxx update delete bitmap put delete_bitmap_key=" << hex(key)
<< " lock_id=" << request->lock_id() << " initiator=" <<
request->initiator()
<< " key_size: " << key.size() << " value_size: " << val.size();
+ return true;
}
static bool commit_pre_rowset_delete_bitmap_removal(
MetaServiceCode& code, std::string& msg, std::stringstream& ss,
const std::shared_ptr<TxnKv>& txn_kv, std::unique_ptr<Transaction>&
txn, KVStats& stats,
- UpdateDeleteBitmapTxnStats& txn_stats, int64_t tablet_id, const
std::string& rowset_id) {
+ UpdateDeleteBitmapTxnStats& txn_stats, const
UpdateDeleteBitmapRequest* request,
+ const std::string& instance_id, const std::string& rowset_id) {
+ auto tablet_id = request->tablet_id();
auto txn_size = txn->approximate_bytes();
LOG(INFO) << "commit delete bitmap point deletes before transaction size
exceeds limit, "
"tablet_id="
@@ -4119,7 +4169,7 @@ static bool commit_pre_rowset_delete_bitmap_removal(
msg = "failed to init txn when removing pre rowsets delete bitmap";
return false;
}
- return true;
+ return check_pre_rowset_aggregation(code, msg, txn.get(), request,
instance_id);
}
static bool remove_pre_rowset_delete_bitmap(
@@ -4183,7 +4233,7 @@ static bool remove_pre_rowset_delete_bitmap(
if (txn_size > 0 &&
txn_size + key.size() * 3 >
static_cast<size_t>(config::max_txn_commit_byte) &&
!commit_pre_rowset_delete_bitmap_removal(code, msg, ss,
txn_kv, txn, stats,
- txn_stats,
tablet_id,
+ txn_stats,
request, instance_id,
rowset_stats.rowset_id())) {
return false;
}
@@ -4283,6 +4333,10 @@ void
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
return;
}
+ if (!check_pre_rowset_aggregation(code, msg, txn.get(), request,
instance_id)) {
+ return;
+ }
+
bool is_explicit_txn = (request->has_is_explicit_txn() &&
request->is_explicit_txn());
bool is_first_sub_txn = (is_explicit_txn && request->txn_id() ==
request->lock_id());
std::string log = ", update delete bitmap for tablet " +
std::to_string(tablet_id);
@@ -4409,10 +4463,13 @@ void
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
auto& key = delete_bitmap_keys_v1[i];
auto& val = request->segment_delete_bitmaps(i);
bool is_versioned_read = is_version_read_enabled(instance_id);
- _write_delete_bitmap_kvs(code, msg, ss, txn_kv_, txn, use_version,
non_exist_rowset_ids, i,
- log, stats, request, instance_id, key, val,
txn_stats,
- is_versioned_read, resource_mgr_.get());
- if (code != MetaServiceCode::OK) return;
+ if (!_write_delete_bitmap_kvs(code, msg, ss, txn_kv_, txn, use_version,
+ non_exist_rowset_ids, i, log, stats,
request, instance_id,
+ key, val, txn_stats, is_versioned_read,
+ resource_mgr_.get()) ||
+ code != MetaServiceCode::OK) {
+ return;
+ }
}
// write v2 kvs
for (size_t i = 0; i < delete_bitmap_keys_v2.size(); ++i) {
@@ -4428,10 +4485,13 @@ void
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
<<
request->delete_bitmap_storages(i).delete_bitmap().rowset_ids_size()
<< ". key=" << hex(key) << ", value_size=" << val.size();
bool is_versioned_read = is_version_read_enabled(instance_id);
- _write_delete_bitmap_kvs(code, msg, ss, txn_kv_, txn, use_version,
non_exist_rowset_ids, i,
- log, stats, request, instance_id, key, val,
txn_stats,
- is_versioned_read, resource_mgr_.get());
- if (code != MetaServiceCode::OK) return;
+ if (!_write_delete_bitmap_kvs(code, msg, ss, txn_kv_, txn, use_version,
+ non_exist_rowset_ids, i, log, stats,
request, instance_id,
+ key, val, txn_stats, is_versioned_read,
+ resource_mgr_.get()) ||
+ code != MetaServiceCode::OK) {
+ return;
+ }
}
if (!remove_pre_rowset_delete_bitmap(code, msg, ss, txn_kv_, txn, stats,
request, instance_id,
diff --git a/cloud/test/meta_service_test.cpp b/cloud/test/meta_service_test.cpp
index 6738b498cc1..1c5fc0c06c1 100644
--- a/cloud/test/meta_service_test.cpp
+++ b/cloud/test/meta_service_test.cpp
@@ -7664,23 +7664,179 @@ TEST(MetaServiceTest, UpdateDeleteBitmapWithException)
{
static void put_delete_bitmap_test_rowset(MetaServiceProxy* meta_service,
const std::string& instance_id,
int64_t tablet_id,
- int64_t version, const std::string&
rowset_id) {
+ int64_t version, const std::string&
rowset_id,
+ int64_t start_version = -1) {
std::unique_ptr<Transaction> txn;
ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
std::string rowset_key = meta_rowset_key({instance_id, tablet_id,
version});
doris::RowsetMetaCloudPB rowset_meta;
rowset_meta.set_rowset_id(0);
rowset_meta.set_rowset_id_v2(rowset_id);
+ rowset_meta.set_start_version(start_version < 0 ? version : start_version);
+ rowset_meta.set_end_version(version);
std::string rowset_value;
ASSERT_TRUE(rowset_meta.SerializeToString(&rowset_value));
txn->put(rowset_key, rowset_value);
ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
}
+TEST(MetaServiceTest, PreRowsetAggregationChecksOutputInEveryTransaction) {
+ auto sp = SyncPoint::get_instance();
+ auto old_max_txn_commit_byte = config::max_txn_commit_byte;
+ DORIS_CLOUD_DEFER {
+ config::max_txn_commit_byte = old_max_txn_commit_byte;
+ sp->clear_all_call_backs();
+ sp->disable_processing();
+ };
+
+ for (bool point_delete : {false, true}) {
+ for (std::string_view scenario :
+ {"valid", "same_range", "replaced", "removed", "malformed",
"after_check",
+ "same_range_after_check", "write_split", "cleanup_split"}) {
+ // Range cleanup does not split transactions.
+ if (!point_delete && scenario == "cleanup_split") {
+ continue;
+ }
+ SCOPED_TRACE(fmt::format("scenario={}, point_delete={}", scenario,
point_delete));
+ const bool same_range =
+ scenario == "same_range" || scenario ==
"same_range_after_check";
+ const bool after_check =
+ scenario == "after_check" || scenario ==
"same_range_after_check";
+ const bool valid = scenario == "valid" || scenario == "same_range";
+ sp->clear_all_call_backs();
+ sp->disable_processing();
+ config::max_txn_commit_byte = old_max_txn_commit_byte;
+ auto meta_service = get_meta_service();
+ const std::string instance_id = "test_instance";
+ constexpr int64_t tablet_id = 652;
+ put_delete_bitmap_test_rowset(meta_service.get(), instance_id,
tablet_id, 2, "r2");
+ put_delete_bitmap_test_rowset(meta_service.get(), instance_id,
tablet_id, 5, "r4-5", 4);
+ auto output_key = meta_rowset_key({instance_id, tablet_id, 5});
+ auto old_key = meta_delete_bitmap_key({instance_id, tablet_id,
"r2", 4, 0});
+ const std::string old_bitmap(
+ scenario == "cleanup_split" ? DEFAULT_BLOB_SPLIT_SIZE * 20
: 10, 'o');
+ const std::string complete_bitmap = "D3|D4|D5";
+ const std::string stale_bitmap(128, 'a');
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ blob_put(txn.get(), old_key, old_bitmap, 0);
+ for (int segment = 0; segment < 2; ++segment) {
+ auto key = meta_delete_bitmap_key({instance_id, tablet_id,
"r2", 5, segment});
+ blob_put(txn.get(), key, complete_bitmap, 0);
+ }
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ UpdateDeleteBitmapRequest req;
+ req.set_cloud_unique_id("test_cloud_unique_id");
+ req.set_table_id(650);
+ req.set_partition_id(651);
+ req.set_tablet_id(tablet_id);
+ req.set_lock_id(-3);
+ req.set_without_lock(true);
+ req.set_initiator(tablet_id);
+ req.set_pre_rowset_agg_start_version(4);
+ req.set_pre_rowset_agg_end_version(5);
+
req.set_enable_remove_agg_pre_rowsets_delete_bitmap_by_keys(point_delete);
+
req.set_enable_remove_pre_rowsets_delete_bitmap_by_keys(point_delete);
+ for (int segment = 0; segment < 2; ++segment) {
+ req.add_rowset_ids("r2");
+ req.add_segment_ids(segment);
+ req.add_versions(5);
+ req.add_segment_delete_bitmaps(stale_bitmap);
+ req.add_pre_rowset_versions(2);
+ }
+ auto* rowset_stats = req.add_pre_rowset_delete_bitmap_stats();
+ rowset_stats->set_rowset_id("r2");
+ auto* bitmap_stats = rowset_stats->add_delete_bitmap_stats();
+ bitmap_stats->set_segment_id(0);
+ bitmap_stats->set_version(4);
+ bitmap_stats->set_delete_bitmap_size(old_bitmap.size());
+
+ auto replace_output = [&] {
+ // Rewriting the same range changes the output ID, but not the
aggregation range
+ // for the unchanged pre-rowset. Replacing the key still
conflicts with a reader.
+ put_delete_bitmap_test_rowset(meta_service.get(), instance_id,
tablet_id, 5,
+ "replacement", same_range ? 4 :
3);
+ };
+ bool injected = false;
+ auto after_split = [&](auto&&) {
+ ASSERT_FALSE(injected);
+ injected = true;
+ // A newer compaction and its aggregation finish between two
batches of A.
+ replace_output();
+ std::unique_ptr<Transaction> newer_txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&newer_txn),
TxnErrorCode::TXN_OK);
+ for (int segment = 0; segment < 2; ++segment) {
+ auto key = meta_delete_bitmap_key({instance_id, tablet_id,
"r2", 5, segment});
+ blob_put(newer_txn.get(), key, complete_bitmap, 0);
+ }
+ ASSERT_EQ(newer_txn->commit(), TxnErrorCode::TXN_OK);
+ };
+ if (scenario == "replaced" || scenario == "same_range") {
+ replace_output();
+ } else if (scenario == "removed" || scenario == "malformed") {
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ if (scenario == "removed") {
+ txn->remove(output_key);
+ } else {
+ txn->put(output_key, "invalid protobuf");
+ }
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ } else if (after_check) {
+
sp->set_call_back("update_delete_bitmap:check_pre_rowset_aggregation",
[&](auto&&) {
+ injected = true;
+ // Change only the rowset key: the conflict must come from
+ // the validity read, not a concurrent bitmap write.
+ replace_output();
+ });
+ } else if (scenario == "write_split") {
+ config::max_txn_commit_byte = 600;
+ sp->set_call_back("update_delete_bitmap:commit:err",
[&](auto&& args) {
+ ASSERT_EQ(try_any_cast<size_t>(args[1]), 1);
+ after_split(args);
+ });
+ } else if (scenario == "cleanup_split") {
+ config::max_txn_commit_byte = 1000;
+
sp->set_call_back("update_delete_bitmap:remove_pre_rowsets:commit",
after_split);
+ }
+ sp->enable_processing();
+ UpdateDeleteBitmapResponse res;
+ brpc::Controller cntl;
+ meta_service->update_delete_bitmap(&cntl, &req, &res, nullptr);
+ auto expected_code = after_check ?
MetaServiceCode::KV_TXN_CONFLICT
+ : scenario == "malformed" ?
MetaServiceCode::PROTOBUF_PARSE_ERR
+ :
MetaServiceCode::OK;
+ ASSERT_EQ(res.status().code(), expected_code) <<
res.status().msg();
+ EXPECT_EQ(injected,
+ after_check || scenario == "write_split" || scenario ==
"cleanup_split");
+ if (valid) {
+ // Repeating an aggregation whose output is still live remains
valid.
+ meta_service->update_delete_bitmap(&cntl, &req, &res, nullptr);
+ ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+ }
+
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ for (int segment = 0; segment < 2; ++segment) {
+ auto key = meta_delete_bitmap_key({instance_id, tablet_id,
"r2", 5, segment});
+ ValueBuf value;
+ ASSERT_EQ(blob_get(txn.get(), key, &value),
TxnErrorCode::TXN_OK);
+ EXPECT_EQ(value.value(), valid ? stale_bitmap :
complete_bitmap);
+ }
+ // Cleanup may have committed a prefix before becoming stale, but
must stop before
+ // removing the remaining keys. Other stale requests must preserve
the entire source.
+ auto remaining_key = encode_blob_key(old_key, 0, scenario ==
"cleanup_split" ? 19 : 0);
+ std::string remaining_value;
+ EXPECT_EQ(txn->get(remaining_key, &remaining_value),
+ valid ? TxnErrorCode::TXN_KEY_NOT_FOUND :
TxnErrorCode::TXN_OK);
+ }
+ }
+}
+
void update_delete_bitmap_with_remove_pre(MetaServiceProxy* meta_service,
int64_t table_id,
int64_t tablet_id, bool inject =
false,
bool rowset_non_exist = false,
bool use_delete_bitmap_stats = true)
{
+ put_delete_bitmap_test_rowset(meta_service, "test_instance", tablet_id, 6,
"r4-6", 4);
// create rowset, if `rowset_non_exist` enabled, only r4 exists
{
std::unique_ptr<Transaction> txn;
@@ -7933,6 +8089,7 @@ TEST(MetaServiceTest, EmptyPreRowsetStatsUsesKeyRemoval) {
constexpr int64_t tablet_id = 622;
const std::string rowset_id = "empty_pre_rowset_stats";
put_delete_bitmap_test_rowset(meta_service.get(), instance_id, tablet_id,
1, rowset_id);
+ put_delete_bitmap_test_rowset(meta_service.get(), instance_id, tablet_id,
2, "r1-2", 1);
auto old_delete_bitmap_key = meta_delete_bitmap_key({instance_id,
tablet_id, rowset_id, 1, 0});
const std::string old_delete_bitmap = "old_delete_bitmap";
@@ -7977,6 +8134,7 @@ TEST(MetaServiceTest,
UnderestimatedPreRowsetDeleteBitmapSizeLeavesSkippedTail)
constexpr int64_t tablet_id = 632;
const std::string rowset_id = "underestimated_delete_bitmap";
put_delete_bitmap_test_rowset(meta_service.get(), instance_id, tablet_id,
1, rowset_id);
+ put_delete_bitmap_test_rowset(meta_service.get(), instance_id, tablet_id,
2, "r1-2", 1);
auto old_delete_bitmap_key = meta_delete_bitmap_key({instance_id,
tablet_id, rowset_id, 1, 0});
const std::string old_delete_bitmap(DEFAULT_BLOB_SPLIT_SIZE * 3 + 1, 'a');
@@ -8059,6 +8217,7 @@ static void
test_get_delete_bitmap_during_point_cleanup(size_t delete_bitmap_siz
constexpr int64_t tablet_id = 642;
const std::string rowset_id = "point_cleanup_during_paginated_read";
put_delete_bitmap_test_rowset(meta_service.get(), instance_id, tablet_id,
1, rowset_id);
+ put_delete_bitmap_test_rowset(meta_service.get(), instance_id, tablet_id,
2, "r1-2", 1);
auto old_delete_bitmap_key = meta_delete_bitmap_key({instance_id,
tablet_id, rowset_id, 1, 0});
std::string old_delete_bitmap(DEFAULT_BLOB_SPLIT_SIZE, 'a');
@@ -8502,6 +8661,7 @@ TEST(MetaServiceTest,
RemovePreDeleteBitmapBatchesEachBlobKey) {
const std::string& cloud_unique_id);
auto instance_id = get_instance_id(meta_service->resource_mgr(),
"test_cloud_unique_id");
constexpr int64_t tablet_id = 612;
+ put_delete_bitmap_test_rowset(meta_service.get(), instance_id, tablet_id,
2, "r1-2", 1);
const std::string rowset_id = "batch_blob_keys_rowset";
const std::string value(DEFAULT_BLOB_SPLIT_SIZE * 20, 'a');
auto delete_bitmap_key = meta_delete_bitmap_key({instance_id, tablet_id,
rowset_id, 1, 0});
diff --git
a/regression-test/suites/compaction/test_cloud_mow_compaction_agg_bitmap_two_be.groovy
b/regression-test/suites/compaction/test_cloud_mow_compaction_agg_bitmap_two_be.groovy
new file mode 100644
index 00000000000..fcfd96582c2
--- /dev/null
+++
b/regression-test/suites/compaction/test_cloud_mow_compaction_agg_bitmap_two_be.groovy
@@ -0,0 +1,145 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+// Use an existing cloud Docker image with the debug points below enabled.
+// Set image and excludeDockerTest=false in regression-conf-custom.groovy,
then run:
+// bash run-regression-test.sh --run -d compaction \
+// -s test_cloud_mow_compaction_agg_bitmap_two_be -noKillDocker
+// The final comparison fails on affected builds: 18 rows instead of 12, with
+// stale (k, 0) rows for k=0,2,4,6,8,10. No generated .out file is required.
+suite("test_cloud_mow_compaction_agg_bitmap_two_be", "docker") {
+ def options = new ClusterOptions()
+ options.cloudMode = true
+ options.setFeNum(1)
+ options.setBeNum(2)
+ options.enableDebugPoints()
+ options.beConfigs += [
+ 'disable_auto_compaction=true',
+ 'enable_parallel_cumu_compaction=false',
+ 'cumulative_compaction_min_deltas=2',
+ 'delete_bitmap_store_read_version=1',
+ 'delete_bitmap_store_write_version=1',
+ 'enable_sync_tablet_delete_bitmap_by_cache=false',
+ 'enable_agg_and_remove_pre_rowsets_delete_bitmap=true',
+ 'enable_remove_agg_pre_rowsets_delete_bitmap_by_keys=true',
+ 'enable_remove_pre_rowsets_delete_bitmap_by_keys=true',
+ ]
+
+ docker(options) {
+ sql "DROP TABLE IF EXISTS test_cloud_mow_compaction_agg_bitmap_two_be"
+ sql """
+ CREATE TABLE test_cloud_mow_compaction_agg_bitmap_two_be (k
BIGINT, v INT)
+ UNIQUE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true"
+ )
+ """
+ // R2 contains all old rows; R3, R4 and R5 overwrite disjoint sets of
keys.
+ sql """INSERT INTO test_cloud_mow_compaction_agg_bitmap_two_be
+ SELECT number, 0 FROM numbers("number" = "12")"""
+ sql """INSERT INTO test_cloud_mow_compaction_agg_bitmap_two_be
+ SELECT number, 10 FROM numbers("number" = "12") WHERE number %
2 = 0"""
+ sql """INSERT INTO test_cloud_mow_compaction_agg_bitmap_two_be
+ SELECT number, 20 FROM numbers("number" = "12") WHERE number %
4 = 1"""
+ sql """INSERT INTO test_cloud_mow_compaction_agg_bitmap_two_be
+ SELECT number, 30 FROM numbers("number" = "12") WHERE number %
4 = 3"""
+
+ def actualSql = "SELECT k, v FROM
test_cloud_mow_compaction_agg_bitmap_two_be ORDER BY k, v"
+ def expectedSql = """SELECT number AS k,
+ CASE WHEN number % 2 = 0 THEN 10 WHEN number % 4 = 1 THEN 20
ELSE 30 END AS v
+ FROM numbers("number" = "12") ORDER BY k, v"""
+ quickTest("before_compaction", actualSql, expectedSql)
+
+ def tabletId = sql_return_maparray("SHOW TABLETS FROM
test_cloud_mow_compaction_agg_bitmap_two_be")[0].TabletId
+ def beA = cluster.getBeByIndex(1)
+ def beB = cluster.getBeByIndex(2)
+ def selector =
"CloudSizeBasedCumulativeCompactionPolicy::pick_input_rowsets.set_input_rowsets"
+ def barrier =
"CumulativeCompaction.modify_rowsets.cloud_update_delete_bitmap_without_lock.block"
+ def request = { be, method, path ->
+ def (code, out, err) = curl(method,
"http://${be.host}:${be.httpPort}${path}";)
+ assertEquals(0, code, "${method} ${path}: ${err}")
+ return out
+ }
+ def rowsets = { be ->
+ parseJson(request(be, "GET",
"/api/compaction/show?tablet_id=${tabletId}")).rowsets
+ }
+ def running = { be ->
+ def status = parseJson(request(be, "GET",
"/api/compaction/run_status?tablet_id=${tabletId}"))
+ assertEquals("success", status.status.toLowerCase())
+ return status.run_status
+ }
+ def submit = { be, start ->
+ be.enableDebugPoint(selector, [tablet_id: tabletId, start_version:
"${start}", end_version: "5"])
+ def result = parseJson(request(be, "POST",
+
"/api/compaction/run?tablet_id=${tabletId}&compact_type=cumulative"))
+ assertEquals("Success", result.status, result.toString())
+ }
+ def waitForAggregation = { be, start ->
+ awaitUntil(120) {
+ !running(be) && new File(be.getLogFilePath()).readLines().any {
+ it.contains("agg pre rowsets delete bitmap to ms.
tablet_id=${tabletId},") &&
+ it.contains("output version=[${start}-5],
cost(us)=")
+ }
+ }
+ }
+
+ try {
+ // A publishes R45, computes D4 | D5 for R2, then pauses before
the MS write.
+ beA.enableDebugPoint(barrier, [timeout: "600"])
+ // Cache warmup may have loaded A before all four inserts were
published.
+ assertTrue(request(beA, "GET",
"/api/clear_cache/CloudTabletCache").contains("prune win"))
+ def initialRowsets = rowsets(beA)
+ assertTrue(initialRowsets.any { it.contains("[4-4]") },
initialRowsets.toString())
+ assertTrue(initialRowsets.any { it.contains("[5-5]") },
initialRowsets.toString())
+ submit(beA, 4)
+ def beALog = new File(beA.getLogFilePath())
+ awaitUntil(120) {
+ beALog.text.contains("start debug block ${barrier}")
+ }
+ assertTrue(running(beA))
+ assertTrue(rowsets(beA).any { it.contains("[4-5]") })
+
+ // B must reload the published R45 instead of using its
pre-compaction tablet.
+ assertTrue(request(beB, "GET",
"/api/clear_cache/CloudTabletCache").contains("prune win"))
+ submit(beB, 3)
+ waitForAggregation(beB, 3)
+ assertTrue(rowsets(beB).any { it.contains("[3-5]") })
+ assertTrue(running(beA))
+ // B stores D3 | D4 | D5 at version 5 and removes source versions
3 and 4.
+ quickTest("after_B_while_A_blocked", actualSql, expectedSql)
+
+ // A's delayed D4 | D5 overwrites B's version-5 bitmap on affected
builds.
+ beA.disableDebugPoint(barrier)
+ waitForAggregation(beA, 4)
+ quickTest("after_A_released", actualSql, expectedSql)
+ } finally {
+ beA.disableDebugPoint(barrier)
+ beA.disableDebugPoint(selector)
+ beB.disableDebugPoint(selector)
+ }
+
+ // Both BEs can retain the complete bitmap in memory; restart both for
a cold read.
+ cluster.restartBackends()
+ logger.info("Rows after restarting both BEs: ${sql(actualSql)}")
+ quickTest("after_restart", actualSql, expectedSql)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
