diff --git a/cloud/src/recycler/recycler.cpp b/cloud/src/recycler/recycler.cpp index 19c438c73b9b9d..130540406db4e9 100644 --- a/cloud/src/recycler/recycler.cpp +++ b/cloud/src/recycler/recycler.cpp @@ -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; @@ -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}); @@ -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; @@ -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; diff --git a/cloud/test/recycle_versioned_keys_test.cpp b/cloud/test/recycle_versioned_keys_test.cpp index 1551f895c30605..763749dafb4ff8 100644 --- a/cloud/test/recycle_versioned_keys_test.cpp +++ b/cloud/test/recycle_versioned_keys_test.cpp @@ -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" @@ -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 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> 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(args[0]); + std::string key = instance_key(instance_id); + std::string value; + ASSERT_EQ(txn->get(key, &value), TxnErrorCode::TXN_OK); + std::unique_ptr 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();