This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a60f34a296e [fix](cloud) Skip stale delete bitmap aggregation (#68084)
a60f34a296e is described below
commit a60f34a296ecff4d720a6d68faab6f4b68679a42
Author: meiyi <[email protected]>
AuthorDate: Tue Oct 20 09:45:31 2026 +0800
[fix](cloud) Skip stale delete bitmap aggregation (#68084)
Problem Summary: Delayed compaction aggregation can overwrite newer
delete bitmaps and make deleted rows visible. Validate the compaction
output range in every MS transaction, stopping stale requests before
writes or cleanup. Preserve non-OK error checks and add unit tests and a
two-BE Docker reproducer.
---
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 8a5e2953c99..b12569a9afc 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -3935,7 +3935,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,
@@ -3969,7 +4013,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();
@@ -3983,7 +4027,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});
@@ -3994,7 +4041,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;
}
}
}
@@ -4003,7 +4050,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
@@ -4015,7 +4062,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;
@@ -4031,7 +4078,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())) {
@@ -4039,7 +4086,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);
@@ -4050,7 +4097,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) {
@@ -4058,14 +4105,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
@@ -4075,7 +4122,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) {
@@ -4111,12 +4158,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="
@@ -4149,7 +4199,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(
@@ -4213,7 +4263,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;
}
@@ -4313,6 +4363,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);
@@ -4439,10 +4493,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) {
@@ -4458,10 +4515,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 e40d79998f2..f1b6a5734d2 100644
--- a/cloud/test/meta_service_test.cpp
+++ b/cloud/test/meta_service_test.cpp
@@ -7738,23 +7738,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;
@@ -8007,6 +8163,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";
@@ -8051,6 +8208,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');
@@ -8133,6 +8291,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');
@@ -8240,6 +8399,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]
