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 cb505369752 [fix](cloud) Prevent stale delete rowset replay from
recreating tmp metadata (#69007)
cb505369752 is described below
commit cb50536975217fb1bd26cf98250b3e0887689077
Author: Yixuan Wang <[email protected]>
AuthorDate: Fri Oct 23 00:19:57 2026 +0800
[fix](cloud) Prevent stale delete rowset replay from recreating tmp
metadata (#69007)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary: When recycle rowset key checking is enabled,
commit_rowset skips the load transaction status check.
For a delete-predicate rowset, the recycle key check treats the key as
present for compatibility with older BEs.
A replay after the transaction becomes visible can therefore recreate a
tmp rowset key that the Recycler cannot safely remove.
Run the existing transaction status check for delete-predicate rowsets
before writing tmp metadata. Add a unit test covering the post-commit
replay.
### Release note
Prevent stale DELETE rowset requests from recreating temporary rowset
metadata after transaction commit.
---
cloud/src/meta-service/meta_service.cpp | 128 ++++++++++++++------------------
cloud/test/meta_service_job_test.cpp | 4 +-
cloud/test/meta_service_test.cpp | 50 +++++++++++++
3 files changed, 109 insertions(+), 73 deletions(-)
diff --git a/cloud/src/meta-service/meta_service.cpp
b/cloud/src/meta-service/meta_service.cpp
index 19e1ed1ccd3..3b691083730 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -40,6 +40,7 @@
#include <limits>
#include <memory>
#include <numeric>
+#include <optional>
#include <ostream>
#include <sstream>
#include <string>
@@ -2655,44 +2656,33 @@ void
MetaServiceImpl::prepare_rowset(::google::protobuf::RpcController* controll
}
}
-// Returns false only when reading or parsing the recycle rowset key fails, or
when the rowset has
-// already been marked as recycled. recycle_rs_key_exists tells the caller
whether the key exists,
-bool check_recycle_rowset_key(Transaction* txn, const std::string&
recycle_rs_key,
- const doris::RowsetMetaCloudPB& rowset_meta,
- bool* recycle_rs_key_exists, MetaServiceCode&
code,
- std::string& msg) {
- *recycle_rs_key_exists = false;
- if (!rowset_meta.has_delete_predicate()) {
- std::string recycle_rs_val;
- TxnErrorCode err = txn->get(recycle_rs_key, &recycle_rs_val);
- if (err == TxnErrorCode::TXN_OK) {
- *recycle_rs_key_exists = true;
- RecycleRowsetPB recycle_rowset;
- if (!recycle_rowset.ParseFromString(recycle_rs_val)) {
- code = MetaServiceCode::PROTOBUF_PARSE_ERR;
- msg = fmt::format("malformed recycle rowset value. key={}",
hex(recycle_rs_key));
- return false;
- }
- auto rs_meta = recycle_rowset.rowset_meta();
- if (rs_meta.has_is_recycled() && rs_meta.is_recycled()) {
- code = MetaServiceCode::INVALID_ARGUMENT;
- msg = fmt::format("rowset has already been marked as recycled,
key={}, rs_meta={}",
- hex(recycle_rs_key),
rs_meta.ShortDebugString());
- return false;
- }
- } else if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
- return true;
- } else {
- code = cast_as<ErrCategory::READ>(err);
- msg = fmt::format("failed to get recycle rowset, err={}, key={}",
err,
- hex(recycle_rs_key));
- return false;
- }
- } else {
- // Old BE does not write recycle rowset keys for delete predicate
rowsets during
- // commit_rowset, so treat the key as existing for compatibility.
- *recycle_rs_key_exists = true;
+bool check_recycle_rowset_key_exists(Transaction* txn, const std::string&
recycle_rs_key,
+ std::optional<bool>&
recycle_rs_key_exists,
+ MetaServiceCode& code, std::string& msg) {
+ std::string value;
+ auto err = txn->get(recycle_rs_key, &value);
+
+ if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
+ recycle_rs_key_exists = false;
+ return true;
+ }
+
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::READ>(err);
+ msg = fmt::format("failed to get recycle rowset, err={}, key={}", err,
hex(recycle_rs_key));
+ return false;
}
+
+ RecycleRowsetPB recycle_rowset;
+ if (!recycle_rowset.ParseFromString(value)) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = fmt::format("malformed recycle rowset value. key={}",
hex(recycle_rs_key));
+ return false;
+ }
+
+ recycle_rs_key_exists = recycle_rowset.rowset_meta().has_is_recycled()
+ ?
!recycle_rowset.rowset_meta().is_recycled()
+ : true;
return true;
}
@@ -2744,8 +2734,18 @@ void
MetaServiceImpl::commit_rowset(::google::protobuf::RpcController* controlle
bool is_versioned_read = is_version_read_enabled(instance_id);
auto recycle_rs_key = recycle_rowset_key({instance_id, tablet_id,
rowset_id});
+ std::optional<bool> recycle_rs_key_exists;
- if (!config::enable_recycle_delete_rowset_key_check) {
+ if (config::enable_recycle_delete_rowset_key_check &&
!rowset_meta.has_delete_predicate()) {
+ if (!check_recycle_rowset_key_exists(txn.get(), recycle_rs_key,
recycle_rs_key_exists, code,
+ msg)) {
+ return;
+ }
+ } else {
+ // Fallback path:
+ // Use the old job and transaction checks when the recycle key check
is off
+ // or the rowset has a delete predicate, because older BEs do not call
prepare_rowset for these rowsets.
+ // enable_tablet_job_check and enable_load_txn_status_check are both
true by default.
std::string tablet_job_id = request->tablet_job_id();
if (config::enable_tablet_job_check && !tablet_job_id.empty()) {
if (!check_job_existed(txn.get(), code, msg, instance_id,
tablet_id, rowset_id,
@@ -2780,25 +2780,18 @@ void
MetaServiceImpl::commit_rowset(::google::protobuf::RpcController* controlle
return;
}
if (existed_rowset_meta->rowset_id_v2() == rowset_meta.rowset_id_v2())
{
- if (config::enable_recycle_delete_rowset_key_check) {
- bool recycle_rs_key_exists = false;
- if (!check_recycle_rowset_key(txn.get(), recycle_rs_key,
rowset_meta,
- &recycle_rs_key_exists, code,
msg)) {
- return;
- }
- if (recycle_rs_key_exists) {
- code = MetaServiceCode::INVALID_ARGUMENT;
- msg = fmt::format(
- "tmp rowset key and recycle rowset key are
mutually exclusive, "
- "tmp_rs_key={}, recycle_rs_key={}",
- hex(tmp_rs_key), hex(recycle_rs_key));
- LOG(INFO) << "skip commit rowset because tmp rowset key
and recycle rowset key "
- "are mutually exclusive, txn_id="
- << rowset_meta.txn_id() << ", tablet_id=" <<
tablet_id
- << ", rowset_id=" << rowset_id << ",
tmp_rs_key=" << hex(tmp_rs_key)
- << ", recycle_rs_key=" << hex(recycle_rs_key) <<
", msg=" << msg;
- return;
- }
+ if (recycle_rs_key_exists.has_value() &&
recycle_rs_key_exists.value()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = fmt::format(
+ "tmp rowset key and recycle rowset key are mutually
exclusive, "
+ "tmp_rs_key={}, recycle_rs_key={}",
+ hex(tmp_rs_key), hex(recycle_rs_key));
+ LOG(INFO) << "skip commit rowset because tmp rowset key and
recycle rowset key "
+ "are mutually exclusive, txn_id="
+ << rowset_meta.txn_id() << ", tablet_id=" <<
tablet_id
+ << ", rowset_id=" << rowset_id << ", tmp_rs_key=" <<
hex(tmp_rs_key)
+ << ", recycle_rs_key=" << hex(recycle_rs_key) << ",
msg=" << msg;
+ return;
}
// Same request, return OK
response->set_allocated_existed_rowset_meta(nullptr);
@@ -2851,20 +2844,13 @@ void
MetaServiceImpl::commit_rowset(::google::protobuf::RpcController* controlle
msg = fmt::format("failed to check whether rowset exists, err={}",
err);
return;
}
- if (config::enable_recycle_delete_rowset_key_check) {
- bool recycle_rs_key_exists = false;
- if (!check_recycle_rowset_key(txn.get(), recycle_rs_key, rowset_meta,
- &recycle_rs_key_exists, code, msg)) {
- return;
- }
- if (!recycle_rs_key_exists) {
- code = MetaServiceCode::INVALID_ARGUMENT;
- msg = fmt::format("recycle rowset key not found, key={}",
hex(recycle_rs_key));
- LOG(INFO) << "skip commit rowset because recycle rowset key does
not exist, txn_id="
- << rowset_meta.txn_id() << ", tablet_id=" << tablet_id
- << ", rowset_id=" << rowset_id << ", recycle_rs_key=" <<
hex(recycle_rs_key);
- return;
- }
+ if (recycle_rs_key_exists.has_value() && !recycle_rs_key_exists.value()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = fmt::format("recycle rowset key not found, key={}",
hex(recycle_rs_key));
+ LOG(INFO) << "skip commit rowset because recycle rowset key does not
exist, txn_id="
+ << rowset_meta.txn_id() << ", tablet_id=" << tablet_id
+ << ", rowset_id=" << rowset_id << ", recycle_rs_key=" <<
hex(recycle_rs_key);
+ return;
}
// write schema kv if rowset_meta has schema
if (config::write_schema_kv && rowset_meta.has_tablet_schema()) {
diff --git a/cloud/test/meta_service_job_test.cpp
b/cloud/test/meta_service_job_test.cpp
index 87a07325294..eddb734b503 100644
--- a/cloud/test/meta_service_job_test.cpp
+++ b/cloud/test/meta_service_job_test.cpp
@@ -7159,7 +7159,7 @@ TEST(MetaServiceJobTest, AbortTxnForRelatedRowsetTest3) {
if (!arena) delete req;
// Should fail because transaction is aborted
ASSERT_NE(res.status().code(), MetaServiceCode::OK);
- ASSERT_TRUE(res.status().msg().find("rowset has already been
marked as recycled") !=
+ ASSERT_TRUE(res.status().msg().find("recycle rowset key not
found") !=
std::string::npos)
<< res.status().msg();
LOG(INFO) << "Step 5: commit_rowset correctly failed for aborted
txn";
@@ -7271,7 +7271,7 @@ TEST(MetaServiceJobTest, AbortJobForRelatedRowsetTest3) {
if (!arena) delete req;
// Should fail because recycle_rowset_key is marked recycled
ASSERT_NE(res.status().code(), MetaServiceCode::OK);
- ASSERT_TRUE(res.status().msg().find("rowset has already been
marked as recycled") !=
+ ASSERT_TRUE(res.status().msg().find("recycle rowset key not
found") !=
std::string::npos)
<< res.status().msg();
LOG(INFO) << "Step 5: commit_rowset executed with code=" <<
res.status().code();
diff --git a/cloud/test/meta_service_test.cpp b/cloud/test/meta_service_test.cpp
index 6655c3da64d..ec393bae1de 100644
--- a/cloud/test/meta_service_test.cpp
+++ b/cloud/test/meta_service_test.cpp
@@ -12869,6 +12869,56 @@ TEST(MetaServiceTest, StaleCommitRowset) {
ASSERT_EQ(res.status().code(), MetaServiceCode::INVALID_ARGUMENT) <<
res.status().code();
}
+TEST(MetaServiceTest, StaleDeletePredicateCommitRowset) {
+ auto meta_service = get_meta_service();
+
+ const bool old_recycle_key_check =
config::enable_recycle_delete_rowset_key_check;
+ const bool old_load_txn_status_check =
config::enable_load_txn_status_check;
+ config::enable_recycle_delete_rowset_key_check = true;
+ config::enable_load_txn_status_check = true;
+ DORIS_CLOUD_DEFER {
+ config::enable_recycle_delete_rowset_key_check = old_recycle_key_check;
+ config::enable_load_txn_status_check = old_load_txn_status_check;
+ };
+
+ constexpr int64_t db_id = 100201;
+ constexpr int64_t table_id = 1;
+ constexpr int64_t partition_id = 1;
+ constexpr int64_t tablet_id = 1;
+ const std::string label = "test_stale_delete_predicate_commit_rowset";
+ ASSERT_NO_FATAL_FAILURE(
+ create_tablet(meta_service.get(), table_id, 1, partition_id,
tablet_id));
+
+ int64_t txn_id = 0;
+ ASSERT_NO_FATAL_FAILURE(begin_txn(meta_service.get(), db_id, label,
table_id, txn_id));
+ auto rowset = create_rowset(txn_id, tablet_id, partition_id, -1, 0);
+ rowset.set_num_segments(0);
+ rowset.set_resource_id("1");
+ rowset.mutable_load_id()->set_hi(0);
+ rowset.mutable_load_id()->set_lo(0);
+ rowset.mutable_delete_predicate()->set_version(1);
+
+ // Old BE commits delete-predicate rowsets without a recycle rowset key.
+ CreateRowsetResponse res;
+ ASSERT_NO_FATAL_FAILURE(commit_rowset(meta_service.get(), rowset, res));
+ ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg();
+ ASSERT_NO_FATAL_FAILURE(commit_txn(meta_service.get(), db_id, txn_id,
label));
+
+ const auto tmp_rs_key = meta_rowset_tmp_key({mock_instance, txn_id,
tablet_id});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ std::string val;
+ ASSERT_EQ(txn->get(tmp_rs_key, &val), TxnErrorCode::TXN_KEY_NOT_FOUND);
+
+ res.Clear();
+ ASSERT_NO_FATAL_FAILURE(commit_rowset(meta_service.get(), rowset, res));
+ ASSERT_EQ(res.status().code(), MetaServiceCode::INVALID_ARGUMENT) <<
res.status().msg();
+ ASSERT_NE(res.status().msg().find("txn is not in"), std::string::npos) <<
res.status().msg();
+
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ ASSERT_EQ(txn->get(tmp_rs_key, &val), TxnErrorCode::TXN_KEY_NOT_FOUND);
+}
+
TEST(MetaServiceTest, CommitRowsetCheckTmpAndRecycleKeyExclusion) {
auto meta_service = get_meta_service();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
