Skip to content

[fix](cloud) Delete versioned tablet indexes and metadata atomically #68322

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wyxxxcat wants to merge 1 commit into
base: master
Choose a base branch
from
+106 −30
Open

[fix](cloud) Delete versioned tablet indexes and metadata atomically #68322

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 32 additions & 30 deletions cloud/src/recycler/recycler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3571,9 +3571,6 @@ int InstanceRecycler::recycle_orphan_partitions() {
int InstanceRecycler::recycle_tablets(int64_t table_id, int64_t index_id,
RecyclerMetricsContext& metrics_context,
int64_t partition_id) {
bool is_multi_version =
instance_info_.has_multi_version_status() &&
instance_info_.multi_version_status() != MultiVersionStatus::MULTI_VERSION_DISABLED;
int64_t num_scanned = 0;
std::atomic_long num_recycled = 0;

Expand Down Expand Up @@ -3710,7 +3707,37 @@ int InstanceRecycler::recycle_tablets(int64_t table_id, int64_t index_id,
}
}
}
if (is_multi_version) {
if (should_recycle_versioned_keys()) {
// Remove tablet indexes in the same transaction as tablet metadata.
for (const auto& tablet_info : tablets_info) {
std::string versioned_idx_key =
versioned::tablet_index_key({instance_id_, tablet_info.tablet_id});
std::string tablet_index_val;
TxnErrorCode err = txn->get(versioned_idx_key, &tablet_index_val);
if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
continue;
}
if (err != TxnErrorCode::TXN_OK) {
LOG_WARNING("failed to get tablet index kv")
.tag("instance_id", instance_id_)
.tag("tablet_id", tablet_info.tablet_id)
.tag("err", err);
return -1;
}
TabletIndexPB tablet_index_pb;
if (!tablet_index_pb.ParseFromString(tablet_index_val)) {
LOG_WARNING("failed to parse tablet index pb")
.tag("instance_id", instance_id_)
.tag("tablet_id", tablet_info.tablet_id);
return -1;
}
std::string versioned_inverted_idx_key = versioned::tablet_inverted_index_key(
{instance_id_, tablet_index_pb.db_id(), tablet_index_pb.table_id(),
tablet_index_pb.index_id(), tablet_index_pb.partition_id(),
tablet_info.tablet_id});
txn->remove(versioned_inverted_idx_key);
txn->remove(versioned_idx_key);
}
for (auto& tablet_info : tablets_info) {
// Remove all versions of tablet compact stats for recycled tablet
auto k = versioned::tablet_compact_stats_key({instance_id_, tablet_info.tablet_id});
Expand Down Expand Up @@ -3744,6 +3771,7 @@ int InstanceRecycler::recycle_tablets(int64_t table_id, int64_t index_id,
for (auto& k : init_rs_keys) {
txn->remove(k);
}
TEST_SYNC_POINT_CALLBACK("InstanceRecycler::recycle_tablets.before_commit", txn.get());
if (TxnErrorCode err = txn->commit(); err != TxnErrorCode::TXN_OK) {
LOG(WARNING) << "failed to delete kvs related to tablets, instance_id=" << instance_id_
<< ", err=" << err;
Expand Down Expand Up @@ -5699,32 +5727,6 @@ int InstanceRecycler::recycle_versioned_tablet(int64_t tablet_id,
LOG(INFO) << "remove delete bitmap kv, tablet=" << tablet_id << ", begin=" << hex(dbm_start_key)
<< " end=" << hex(dbm_end_key);

std::string versioned_idx_key = versioned::tablet_index_key({instance_id_, tablet_id});
std::string tablet_index_val;
err = txn->get(versioned_idx_key, &tablet_index_val);
if (err != TxnErrorCode::TXN_KEY_NOT_FOUND && err != TxnErrorCode::TXN_OK) {
LOG_WARNING("failed to get tablet index kv")
.tag("instance_id", instance_id_)
.tag("tablet_id", tablet_id)
.tag("err", err);
ret = -1;
} else if (err == TxnErrorCode::TXN_OK) {
// If the tablet index kv exists, we need to delete it
TabletIndexPB tablet_index_pb;
if (!tablet_index_pb.ParseFromString(tablet_index_val)) {
LOG_WARNING("failed to parse tablet index pb")
.tag("instance_id", instance_id_)
.tag("tablet_id", tablet_id);
ret = -1;
} else {
std::string versioned_inverted_idx_key = versioned::tablet_inverted_index_key(
{instance_id_, tablet_index_pb.db_id(), tablet_index_pb.table_id(),
tablet_index_pb.index_id(), tablet_index_pb.partition_id(), tablet_id});
txn->remove(versioned_inverted_idx_key);
txn->remove(versioned_idx_key);
}
}

err = txn->commit();
if (err != TxnErrorCode::TXN_OK) {
LOG(WARNING) << "failed to delete rowset kv of tablet " << tablet_id << ", err=" << err;
Expand Down
74 changes: 74 additions & 0 deletions cloud/test/recycle_versioned_keys_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@

#include "common/defer.h"
#include "common/util.h"
#include "cpp/sync_point.h"
#include "meta-service/meta_service.h"
#include "meta-store/codec.h"
#include "meta-store/document_message.h"
Expand Down Expand Up @@ -1537,6 +1538,79 @@ TEST(RecycleVersionedKeysTest, RecycleTabletWithRowsetRefCountConcurrent) {
}
}

TEST(RecycleVersionedKeysTest, RecycleTabletMetadataAndIndexesAtomically) {
auto meta_service = get_meta_service();
auto txn_kv = meta_service->txn_kv();
std::string instance_id = "recycle_tablet_metadata_and_indexes";
std::string cloud_unique_id = fmt::format("1:{}:0", instance_id);
ASSERT_NO_FATAL_FAILURE(create_and_refresh_instance(meta_service.get(), instance_id));

int64_t db_id = 1, table_id = 2, index_id = 3, partition_id = 4, tablet_id = 5;
ASSERT_NO_FATAL_FAILURE(prepare_and_commit_index(meta_service.get(), cloud_unique_id, db_id,
table_id, index_id));
ASSERT_NO_FATAL_FAILURE(prepare_and_commit_partition(meta_service.get(), cloud_unique_id, db_id,
table_id, partition_id, index_id));
ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), cloud_unique_id, db_id, table_id,
index_id, partition_id, tablet_id));

auto check_tablet_keys = [&](bool exists) {
std::unique_ptr<Transaction> txn;
ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
for (const auto& key :
{meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}),
meta_tablet_idx_key({instance_id, tablet_id}),
versioned::tablet_index_key({instance_id, tablet_id}),
versioned::tablet_inverted_index_key(
{instance_id, db_id, table_id, index_id, partition_id, tablet_id})}) {
std::string value;
EXPECT_EQ(txn->get(key, &value),
exists ? TxnErrorCode::TXN_OK : TxnErrorCode::TXN_KEY_NOT_FOUND)
<< hex(key);
}
for (const auto& key : {versioned::meta_tablet_key({instance_id, tablet_id}),
versioned::tablet_load_stats_key({instance_id, tablet_id}),
versioned::tablet_compact_stats_key({instance_id, tablet_id})}) {
std::vector<std::pair<std::string, Versionstamp>> values;
ASSERT_NO_FATAL_FAILURE(versioned_get_all(txn_kv.get(), key, values));
EXPECT_EQ(values.size(), exists ? 1 : 0) << hex(key);
}
};
ASSERT_NO_FATAL_FAILURE(check_tablet_keys(true));

InstanceInfoPB instance_info;
ASSERT_NO_FATAL_FAILURE(get_instance(meta_service.get(), cloud_unique_id, instance_info));
auto recycler = get_instance_recycler(meta_service.get(), instance_info);
auto* sp = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
sp->clear_all_call_backs();
sp->disable_processing();
};
bool commit_attempted = false;
sp->set_call_back("InstanceRecycler::recycle_tablets.before_commit", [&](auto&& args) {
commit_attempted = true;
// Force a real commit conflict after data recycling without changing tablet keys.
auto* txn = try_any_cast<Transaction*>(args[0]);
std::string key = instance_key(instance_id);
std::string value;
ASSERT_EQ(txn->get(key, &value), TxnErrorCode::TXN_OK);
std::unique_ptr<Transaction> conflicting_txn;
ASSERT_EQ(txn_kv->create_txn(&conflicting_txn), TxnErrorCode::TXN_OK);
conflicting_txn->put(key, value);
ASSERT_EQ(conflicting_txn->commit(), TxnErrorCode::TXN_OK);
});
sp->enable_processing();

RecyclerMetricsContext ctx;
ASSERT_EQ(recycler->recycle_tablets(table_id, index_id, ctx), -1);
ASSERT_TRUE(commit_attempted);
ASSERT_NO_FATAL_FAILURE(check_tablet_keys(true));

sp->disable_processing();
recycler = get_instance_recycler(meta_service.get(), instance_info);
ASSERT_EQ(recycler->recycle_tablets(table_id, index_id, ctx), 0);
ASSERT_NO_FATAL_FAILURE(check_tablet_keys(false));
}

// A test that simulates a drop index operation.
TEST(RecycleVersionedKeysTest, RecycleIndex) {
auto meta_service = get_meta_service();
Expand Down
Loading