diff --git a/cloud/src/meta-service/meta_service_job.cpp b/cloud/src/meta-service/meta_service_job.cpp index 35aedd81852c28..f357418b4fcdfa 100644 --- a/cloud/src/meta-service/meta_service_job.cpp +++ b/cloud/src/meta-service/meta_service_job.cpp @@ -946,7 +946,8 @@ void process_compaction_job(MetaServiceCode& code, std::string& msg, std::string using namespace std::chrono; int64_t now = duration_cast(system_clock::now().time_since_epoch()).count(); - if (recorded_compaction->expiration() > 0 && recorded_compaction->expiration() < now) { + if (request->action() != FinishTabletJobRequest::ABORT && + recorded_compaction->expiration() > 0 && recorded_compaction->expiration() < now) { code = MetaServiceCode::JOB_EXPIRED; SS << "expired compaction job, tablet_id=" << tablet_id << " job=" << proto_to_json(*recorded_compaction); @@ -1544,7 +1545,8 @@ void process_schema_change_job(MetaServiceCode& code, std::string& msg, std::str auto& recorded_schema_change = recorded_job.schema_change(); using namespace std::chrono; int64_t now = duration_cast(system_clock::now().time_since_epoch()).count(); - if (recorded_schema_change.expiration() > 0 && recorded_schema_change.expiration() < now) { + if (request->action() != FinishTabletJobRequest::ABORT && + recorded_schema_change.expiration() > 0 && recorded_schema_change.expiration() < now) { code = MetaServiceCode::JOB_EXPIRED; SS << "expired schema_change job, tablet_id=" << tablet_id << " job=" << proto_to_json(recorded_schema_change); diff --git a/cloud/src/recycler/recycler.cpp b/cloud/src/recycler/recycler.cpp index c2b0e6b7c176eb..ccb0a1fd9320d4 100644 --- a/cloud/src/recycler/recycler.cpp +++ b/cloud/src/recycler/recycler.cpp @@ -85,6 +85,8 @@ using namespace std::chrono; namespace { +constexpr size_t kRowsetBatchGetSize = 256; + int64_t packed_file_retry_sleep_ms() { const int64_t min_ms = std::max(0, config::packed_file_txn_retry_sleep_min_ms); const int64_t max_ms = std::max(min_ms, config::packed_file_txn_retry_sleep_max_ms); @@ -1750,7 +1752,7 @@ int InstanceRecycler::abort_txn_for_related_rowset(int64_t txn_id) { if (err != TxnErrorCode::TXN_OK) { if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { // maybe recycled - LOG_INFO("txn index not found, txn_id={} instance_id={}", txn_id, instance_id_) + LOG_WARNING("txn index not found, txn_id={} instance_id={}", txn_id, instance_id_) .tag("key", hex(index_key)) .tag("txn_id", txn_id); return 0; @@ -1775,7 +1777,7 @@ int InstanceRecycler::abort_txn_for_related_rowset(int64_t txn_id) { if (err != TxnErrorCode::TXN_OK) { if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { // maybe recycled - LOG_INFO("txn info not found, txn_id={} instance_id={}", txn_id, instance_id_) + LOG_WARNING("txn info not found, txn_id={} instance_id={}", txn_id, instance_id_) .tag("key", hex(info_key)) .tag("txn_id", txn_id); return 0; @@ -1794,19 +1796,31 @@ int InstanceRecycler::abort_txn_for_related_rowset(int64_t txn_id) { return -1; } - if (txn_info.status() != TxnStatusPB::TXN_STATUS_PREPARED) { - LOG_INFO("txn is not prepared status, txn_id={} status={}", txn_id, txn_info.status()) + if (txn_info.status() == TxnStatusPB::TXN_STATUS_ABORTED) { + LOG_INFO("txn has already been aborted, txn_id={}", txn_id) .tag("key", hex(info_key)) .tag("txn_id", txn_id); return 0; } + if (txn_info.status() != TxnStatusPB::TXN_STATUS_PREPARED) { + LOG_WARNING("txn cannot be aborted, txn_id={} status={}", txn_id, txn_info.status()) + .tag("key", hex(info_key)) + .tag("txn_id", txn_id); + return -1; + } req.set_txn_id(txn_id); - LOG(INFO) << "begin abort txn for related rowset, txn_id=" << txn_id - << " instance_id=" << instance_id_ << " txn_info=" << txn_info.ShortDebugString(); + LOG(WARNING) << "begin abort txn for related rowset, txn_id=" << txn_id + << " instance_id=" << instance_id_ << " txn_info=" << txn_info.ShortDebugString(); _abort_txn(instance_id_, &req, txn.get(), txn_info, ss, code, msg); + if (code != MetaServiceCode::OK) { + LOG(WARNING) << "failed to abort txn for related rowset, txn_id=" << txn_id + << " instance_id=" << instance_id_ << " code=" << code << " msg=" << msg; + return -1; + } + err = txn->commit(); if (err != TxnErrorCode::TXN_OK) { code = cast_as(err); @@ -1815,14 +1829,15 @@ int InstanceRecycler::abort_txn_for_related_rowset(int64_t txn_id) { return -1; } - LOG(INFO) << "finish abort txn for related rowset, txn_id=" << txn_id - << " instance_id=" << instance_id_ << " txn_info=" << txn_info.ShortDebugString() - << " code=" << code << " msg=" << msg; + LOG(WARNING) << "finish abort txn for related rowset, txn_id=" << txn_id + << " instance_id=" << instance_id_ << " txn_info=" << txn_info.ShortDebugString() + << " code=" << code << " msg=" << msg; return 0; } -int InstanceRecycler::abort_job_for_related_rowset(const RowsetMetaCloudPB& rowset_meta) { +int InstanceRecycler::abort_job_for_related_rowset(int64_t tablet_id, const std::string& rowset_id, + const std::string& job_id) { FinishTabletJobRequest req; FinishTabletJobResponse res; req.set_action(FinishTabletJobRequest::ABORT); @@ -1831,12 +1846,12 @@ int InstanceRecycler::abort_job_for_related_rowset(const RowsetMetaCloudPB& rows std::stringstream ss; TabletIndexPB tablet_idx; - int ret = get_tablet_idx(txn_kv_.get(), instance_id_, rowset_meta.tablet_id(), tablet_idx); + int ret = get_tablet_idx(txn_kv_.get(), instance_id_, tablet_id, tablet_idx); if (ret == 1) { // tablet maybe recycled, directly return 0 return 0; } else if (ret != 0) { - LOG(WARNING) << "failed to get tablet index, tablet_id=" << rowset_meta.tablet_id() + LOG(WARNING) << "failed to get tablet index, tablet_id=" << tablet_id << " instance_id=" << instance_id_ << " ret=" << ret; return ret; } @@ -1857,6 +1872,8 @@ int InstanceRecycler::abort_job_for_related_rowset(const RowsetMetaCloudPB& rows if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { LOG(INFO) << "job not exists, instance_id=" << instance_id_ << " tablet_id=" << tablet_idx.tablet_id() << " key=" << hex(job_key); + // A previous recycler round may have aborted the job. The caller must recheck the + // recycle key before deleting object data. return 0; } LOG(WARNING) << "failed to get job, instance_id=" << instance_id_ @@ -1872,42 +1889,107 @@ int InstanceRecycler::abort_job_for_related_rowset(const RowsetMetaCloudPB& rows return -1; } - std::string job_id {}; - if (!job_pb.compaction().empty()) { - for (const auto& c : job_pb.compaction()) { - if (c.id() == rowset_meta.job_id()) { - job_id = c.id(); - break; - } + const TabletCompactionJobPB* matched_compaction = nullptr; + for (const auto& compaction : job_pb.compaction()) { + if (compaction.id() == job_id) { + matched_compaction = &compaction; + break; } - } else if (job_pb.has_schema_change()) { - job_id = job_pb.schema_change().id(); + } + const TabletSchemaChangeJobPB* matched_schema_change = nullptr; + if (matched_compaction == nullptr && job_pb.has_schema_change() && + job_pb.schema_change().id() == job_id) { + matched_schema_change = &job_pb.schema_change(); } - if (!job_id.empty() && rowset_meta.job_id() == job_id) { - LOG(INFO) << "begin to abort job for related rowset, job_id=" << rowset_meta.job_id() - << " instance_id=" << instance_id_ << " tablet_id=" << tablet_idx.tablet_id(); - req.mutable_job()->CopyFrom(job_pb); + if (matched_compaction != nullptr || matched_schema_change != nullptr) { + LOG(WARNING) << "begin to abort job for related rowset, job_id=" << job_id + << " instance_id=" << instance_id_ << " tablet_id=" << tablet_idx.tablet_id(); + // Compaction jobs belong to the rowset's tablet. A schema-change job is mirrored under + // the new tablet key, but its recorded index remains the base tablet so ABORT can remove + // both the base and mirrored schema-change records. + if (matched_compaction != nullptr) { + req.mutable_job()->mutable_idx()->CopyFrom(tablet_idx); + req.mutable_job()->add_compaction()->CopyFrom(*matched_compaction); + } else { + req.mutable_job()->mutable_idx()->CopyFrom(job_pb.idx()); + req.mutable_job()->mutable_schema_change()->CopyFrom(*matched_schema_change); + } req.set_action(FinishTabletJobRequest::ABORT); _finish_tablet_job(&req, &res, instance_id_, txn, txn_kv_.get(), delete_bitmap_lock_white_list_.get(), resource_mgr_.get(), code, msg, ss); if (code != MetaServiceCode::OK) { + // These two codes mean that the job cannot be aborted anymore. + // TABLET_NOT_FOUND means that the tablet has been dropped or recycled. + // There is no running job that still owns the PREPARE rowset, so the caller may + // recheck the recycle key. + bool no_job_to_abort = code == MetaServiceCode::TABLET_NOT_FOUND; + // INVALID_ARGUMENT has more than one meaning. It may mean that this job was removed or + // replaced by a newer job, but it may also report invalid metadata. Read the current + // job again and continue recycling only when the target job id no longer exists. + if (!no_job_to_abort && code == MetaServiceCode::INVALID_ARGUMENT) { + std::unique_ptr check_txn; + err = txn_kv_->create_txn(&check_txn); + if (err != TxnErrorCode::TXN_OK) { + LOG(WARNING) << "failed to create txn to check related job, instance_id=" + << instance_id_ << " err=" << err; + return -1; + } + + const auto& idx = req.job().idx(); + std::string current_job_key = + job_tablet_key({instance_id_, idx.table_id(), idx.index_id(), + idx.partition_id(), idx.tablet_id()}); + std::string current_job_val; + err = check_txn->get(current_job_key, ¤t_job_val); + if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { + no_job_to_abort = true; + } else if (err != TxnErrorCode::TXN_OK) { + LOG(WARNING) << "failed to check related job, instance_id=" << instance_id_ + << " tablet_id=" << idx.tablet_id() << " err=" << err + << " key=" << hex(current_job_key); + return -1; + } else { + TabletJobInfoPB current_job; + if (!current_job.ParseFromString(current_job_val)) { + LOG(WARNING) << "failed to parse related job, instance_id=" << instance_id_ + << " tablet_id=" << idx.tablet_id() + << " key=" << hex(current_job_key); + return -1; + } + if (matched_compaction != nullptr) { + no_job_to_abort = true; + for (const auto& compaction : current_job.compaction()) { + if (compaction.id() == job_id) { + no_job_to_abort = false; + break; + } + } + } else { + no_job_to_abort = !current_job.has_schema_change() || + current_job.schema_change().id() != job_id; + } + } + } + if (no_job_to_abort) { + return 0; + } LOG(WARNING) << "failed to abort job, instance_id=" << instance_id_ << " tablet_id=" << tablet_idx.tablet_id() << " code=" << code << " msg=" << msg; return -1; } - LOG(INFO) << "finish abort job for related rowset, job_id=" << rowset_meta.job_id() - << " instance_id=" << instance_id_ << " tablet_id=" << tablet_idx.tablet_id() - << " code=" << code << " msg=" << msg; + LOG(WARNING) << "finish abort job for related rowset, job_id=" << job_id + << " instance_id=" << instance_id_ << " tablet_id=" << tablet_idx.tablet_id() + << " code=" << code << " msg=" << msg; } else { // clang-format off - LOG(INFO) << "there is no job for related rowset, directly recycle rowset data" + LOG(INFO) << "there is no job for related rowset, recheck recycle rowset before deletion" << ", instance_id=" << instance_id_ << ", tablet_id=" << tablet_idx.tablet_id() << ", job_id=" << job_id - << ", rowset_id=" << rowset_meta.rowset_id_v2(); + << ", rowset_id=" << rowset_id; // clang-format on } @@ -1932,30 +2014,9 @@ const RowsetMetaCloudPB& rowset_meta(const T& rowset_meta_pb) { } } -struct DeferredRecycleAbortTask { - enum class Type : uint8_t { - TXN, - JOB, - }; - - Type type = Type::TXN; - int64_t txn_id = 0; - int64_t tablet_id = 0; - int64_t start_version = 0; - int64_t end_version = 0; - std::string rowset_id; - std::string job_id; -}; - -struct DeferredRecyclePrepareDeleteTask { - std::string key; - std::string resource_id; - std::string rowset_id; - int64_t tablet_id = 0; -}; - template -std::optional make_deferred_abort_task(const T& rowset_meta_pb) { +std::optional make_related_txn_or_job_abort_task( + const T& rowset_meta_pb) { if constexpr (std::is_same_v) { if (rowset_meta_pb.type() != RecycleRowsetPB::PREPARE) { return std::nullopt; @@ -1963,18 +2024,18 @@ std::optional make_deferred_abort_task(const T& rowset } const auto& rs_meta = rowset_meta(rowset_meta_pb); - DeferredRecycleAbortTask task; + RelatedTxnOrJobAbortTask task; task.tablet_id = rs_meta.tablet_id(); task.start_version = rs_meta.start_version(); task.end_version = rs_meta.end_version(); + task.rowset_id = rs_meta.rowset_id_v2(); if (rs_meta.has_load_id()) { - task.type = DeferredRecycleAbortTask::Type::TXN; + task.type = RelatedTxnOrJobAbortTask::Type::TXN; task.txn_id = rs_meta.txn_id(); return task; } if (rs_meta.has_job_id()) { - task.type = DeferredRecycleAbortTask::Type::JOB; - task.rowset_id = rs_meta.rowset_id_v2(); + task.type = RelatedTxnOrJobAbortTask::Type::JOB; task.job_id = rs_meta.job_id(); return task; } @@ -1982,96 +2043,131 @@ std::optional make_deferred_abort_task(const T& rowset } template -bool need_mark_rowset_as_recycled(const T& rowset_meta_pb) { +bool matches_related_txn_or_job_abort_task(const T& rowset_meta_pb, + const RelatedTxnOrJobAbortTask& abort_task) { + if constexpr (std::is_same_v) { + if (rowset_meta_pb.type() != RecycleRowsetPB::PREPARE) { + return false; + } + } + const auto& rs_meta = rowset_meta(rowset_meta_pb); - return !rs_meta.has_is_recycled() || !rs_meta.is_recycled(); + if (rs_meta.tablet_id() != abort_task.tablet_id || + rs_meta.rowset_id_v2() != abort_task.rowset_id) { + return false; + } + if (abort_task.type == RelatedTxnOrJobAbortTask::Type::TXN) { + return rs_meta.has_load_id() && rs_meta.txn_id() == abort_task.txn_id; + } + return rs_meta.has_job_id() && rs_meta.job_id() == abort_task.job_id; +} + +bool need_mark_rowset_as_recycled(const RowsetMetaCloudPB& rowset_meta_pb) { + return !rowset_meta_pb.has_is_recycled() || !rowset_meta_pb.is_recycled(); } template int batch_mark_rowsets_as_recycled(TxnKv* txn_kv, const std::string& instance_id, const std::vector& keys) { - std::unique_ptr txn; - TxnErrorCode err = txn_kv->create_txn(&txn); - if (err != TxnErrorCode::TXN_OK) { - LOG(WARNING) << "failed to create txn, instance_id=" << instance_id; - return -1; - } - std::vector> values; - err = txn->batch_get(&values, keys); - if (err != TxnErrorCode::TXN_OK) { - LOG(WARNING) << "failed to batch get rowset meta, instance_id=" << instance_id << ' ' - << "keys size=" << keys.size() << ' ' << "err=" << err; - return -1; - } - size_t total_keys = keys.size(); - for (size_t i = 0; i < total_keys; i++) { - if (!values[i].has_value()) { - // has already been removed by commit_rowset - continue; + for (size_t offset = 0; offset < keys.size(); offset += kRowsetBatchGetSize) { + size_t limit = std::min(keys.size(), offset + kRowsetBatchGetSize); + std::vector batch_keys(keys.begin() + offset, keys.begin() + limit); + std::unique_ptr txn; + TxnErrorCode err = txn_kv->create_txn(&txn); + if (err != TxnErrorCode::TXN_OK) { + LOG(WARNING) << "failed to create txn, instance_id=" << instance_id; + return -1; } - auto key = keys[i]; - auto val = values[i].value(); - T rowset_meta_pb; - if (!rowset_meta_pb.ParseFromString(val)) { - LOG(WARNING) << "failed to parse rowset meta, instance_id=" << instance_id - << " key=" << hex(key); + std::vector> values; + err = txn->batch_get(&values, batch_keys); + if (err != TxnErrorCode::TXN_OK) { + LOG(WARNING) << "failed to batch get rowset meta, instance_id=" << instance_id << ' ' + << "keys size=" << batch_keys.size() << ' ' << "err=" << err; return -1; } - if (!need_mark_rowset_as_recycled(rowset_meta_pb)) { - continue; + DCHECK_EQ(values.size(), batch_keys.size()); + for (size_t i = 0; i < batch_keys.size(); i++) { + if (!values[i].has_value()) { + // has already been removed by commit_rowset + continue; + } + const auto& key = batch_keys[i]; + auto val = values[i].value(); + T rowset_meta_pb; + if (!rowset_meta_pb.ParseFromString(val)) { + LOG(WARNING) << "failed to parse rowset meta, instance_id=" << instance_id + << " key=" << hex(key); + return -1; + } + if (!need_mark_rowset_as_recycled(rowset_meta(rowset_meta_pb))) { + continue; + } + mutable_rowset_meta(rowset_meta_pb)->set_is_recycled(true); + if constexpr (std::is_same_v) { + [[maybe_unused]] auto type = rowset_meta_pb.type(); + TEST_SYNC_POINT_CALLBACK("InstanceRecycler::batch_mark_rowsets_as_recycled", &type); + } + val.clear(); + rowset_meta_pb.SerializeToString(&val); + txn->put(key, val); + } + err = txn->commit(); + if (err != TxnErrorCode::TXN_OK) { + LOG(WARNING) << "failed to commit txn, instance_id=" << instance_id; + return -1; } - mutable_rowset_meta(rowset_meta_pb)->set_is_recycled(true); - val.clear(); - rowset_meta_pb.SerializeToString(&val); - txn->put(key, val); - } - err = txn->commit(); - if (err != TxnErrorCode::TXN_OK) { - LOG(WARNING) << "failed to commit txn, instance_id=" << instance_id; - return -1; } return 0; } +template +void InstanceRecycler::submit_batch_mark_rowsets_as_recycled_job( + SimpleThreadPool& worker_pool, std::vector rowset_keys_to_mark) { + worker_pool.submit([this, mark_keys = std::move(rowset_keys_to_mark)]() { + if (batch_mark_rowsets_as_recycled(txn_kv_.get(), instance_id_, mark_keys) != 0) { + LOG(WARNING) << "failed to batch mark rowsets as recycled, instance_id=" << instance_id_ + << ' ' << "rowset_keys_to_mark.size()=" << mark_keys.size(); + } + }); +} + template int collect_deferred_abort_tasks(TxnKv* txn_kv, const std::string& instance_id, const std::vector& keys, - std::vector* abort_tasks, - bool skip_base_version) { - constexpr size_t kAbortCheckBatchSize = 256; - for (size_t offset = 0; offset < keys.size(); offset += kAbortCheckBatchSize) { - size_t limit = std::min(keys.size(), offset + kAbortCheckBatchSize); + std::vector* abort_tasks) { + for (size_t offset = 0; offset < keys.size(); offset += kRowsetBatchGetSize) { + size_t limit = std::min(keys.size(), offset + kRowsetBatchGetSize); + std::vector batch_keys(keys.begin() + offset, keys.begin() + limit); std::unique_ptr txn; TxnErrorCode err = txn_kv->create_txn(&txn); if (err != TxnErrorCode::TXN_OK) { LOG(WARNING) << "failed to create txn, instance_id=" << instance_id; return -1; } - for (size_t idx = offset; idx < limit; ++idx) { - const std::string& key = keys[idx]; - std::string val; - err = txn->get(key, &val); - if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { + std::vector> values; + err = txn->batch_get(&values, batch_keys, Transaction::BatchGetOptions(true)); + if (err != TxnErrorCode::TXN_OK) { + LOG(WARNING) << "failed to batch get rowset meta, instance_id=" << instance_id + << " keys_size=" << batch_keys.size() << " err=" << err; + return -1; + } + DCHECK_EQ(values.size(), batch_keys.size()); + for (size_t idx = 0; idx < batch_keys.size(); ++idx) { + const std::string& key = batch_keys[idx]; + if (!values[idx].has_value()) { // has already been removed continue; } - if (err != TxnErrorCode::TXN_OK) { - LOG(WARNING) << "failed to get rowset meta, instance_id=" << instance_id - << " key=" << hex(key); - return -1; - } T rowset_meta_pb; - if (!rowset_meta_pb.ParseFromString(val)) { + if (!rowset_meta_pb.ParseFromString(*values[idx])) { LOG(WARNING) << "failed to parse rowset meta, instance_id=" << instance_id << " key=" << hex(key); return -1; } - if (skip_base_version && rowset_meta(rowset_meta_pb).end_version() == 1) { - continue; - } - if (auto abort_task = make_deferred_abort_task(rowset_meta_pb); + if (auto abort_task = make_related_txn_or_job_abort_task(rowset_meta_pb); abort_task.has_value()) { + abort_task->key = key; abort_tasks->emplace_back(std::move(*abort_task)); } } @@ -2080,78 +2176,213 @@ int collect_deferred_abort_tasks(TxnKv* txn_kv, const std::string& instance_id, } template -int InstanceRecycler::batch_abort_txn_or_job_for_recycle(const std::vector& keys, - bool skip_base_version) { - std::vector abort_tasks; - if (collect_deferred_abort_tasks(txn_kv_.get(), instance_id_, keys, &abort_tasks, - skip_base_version) != 0) { +int batch_recheck_rowsets_after_abort( + TxnKv* txn_kv, const std::string& instance_id, + const std::vector>& keys_to_recheck, + std::vector>* rowsets_to_recycle) { + int ret = 0; + for (size_t offset = 0; offset < keys_to_recheck.size(); offset += kRowsetBatchGetSize) { + size_t limit = std::min(keys_to_recheck.size(), offset + kRowsetBatchGetSize); + std::vector batch_keys; + batch_keys.reserve(limit - offset); + for (size_t idx = offset; idx < limit; ++idx) { + batch_keys.push_back(keys_to_recheck[idx].first); + } + + std::unique_ptr txn; + TxnErrorCode err = txn_kv->create_txn(&txn); + if (err != TxnErrorCode::TXN_OK) { + LOG(WARNING) << "failed to create txn for rowset recheck, instance_id=" << instance_id + << " keys_size=" << batch_keys.size() << " err=" << err; + ret = -1; + continue; + } + + std::vector> values; + err = txn->batch_get(&values, batch_keys, Transaction::BatchGetOptions(true)); + if (err != TxnErrorCode::TXN_OK) { + LOG(WARNING) << "failed to batch recheck rowset meta, instance_id=" << instance_id + << " keys_size=" << batch_keys.size() << " err=" << err; + ret = -1; + continue; + } + DCHECK_EQ(values.size(), batch_keys.size()); + [[maybe_unused]] size_t batch_size = batch_keys.size(); + TEST_SYNC_POINT_CALLBACK("InstanceRecycler::batch_recheck_rowsets_after_abort", + &batch_size); + + for (size_t idx = 0; idx < batch_keys.size(); ++idx) { + const auto& [key, abort_task] = keys_to_recheck[offset + idx]; + if (!values[idx].has_value()) { + continue; + } + + T rowset_meta_pb; + if (!rowset_meta_pb.ParseFromString(*values[idx])) { + LOG(WARNING) << "failed to parse rowset meta during abort recheck, instance_id=" + << instance_id << " key=" << hex(key); + ret = -1; + continue; + } + if (config::enable_mark_delete_rowset_before_recycle && + need_mark_rowset_as_recycled(rowset_meta(rowset_meta_pb))) { + LOG(WARNING) << "skip unmarked rowset during abort recheck, instance_id=" + << instance_id << " key=" << hex(key); + continue; + } + if (!matches_related_txn_or_job_abort_task(rowset_meta_pb, abort_task)) { + LOG(WARNING) << "skip changed rowset during abort recheck, instance_id=" + << instance_id << " key=" << hex(key); + continue; + } + rowsets_to_recycle->emplace_back(key, std::move(rowset_meta_pb)); + } + } + return ret; +} + +template +int InstanceRecycler::batch_abort_txn_or_job_for_recycle( + const std::vector& keys, + std::vector>& keys_to_recheck) { + std::vector abort_tasks; + if (collect_deferred_abort_tasks(txn_kv_.get(), instance_id_, keys, &abort_tasks) != 0) { LOG(WARNING) << "failed to collect rowset abort tasks, instance_id=" << instance_id_; return -1; } + TEST_SYNC_POINT_CALLBACK("InstanceRecycler::batch_abort_txn_or_job_for_recycle::after_collect"); + int ret = 0; for (const auto& abort_task : abort_tasks) { - LOG(INFO) << "begin to abort txn or job for related rowset, instance_id=" << instance_id_ - << " tablet_id=" << abort_task.tablet_id << " version=[" - << abort_task.start_version << '-' << abort_task.end_version << "]"; int abort_ret = 0; - if (abort_task.type == DeferredRecycleAbortTask::Type::TXN) { + if (abort_task.type == RelatedTxnOrJobAbortTask::Type::TXN) { abort_ret = abort_txn_for_related_rowset(abort_task.txn_id); } else { - RowsetMetaCloudPB rowset_meta; - rowset_meta.set_tablet_id(abort_task.tablet_id); - rowset_meta.set_rowset_id_v2(abort_task.rowset_id); - rowset_meta.set_job_id(abort_task.job_id); - abort_ret = abort_job_for_related_rowset(rowset_meta); + abort_ret = abort_job_for_related_rowset(abort_task.tablet_id, abort_task.rowset_id, + abort_task.job_id); } if (abort_ret != 0) { LOG(WARNING) << "failed to abort txn or job for related rowset, instance_id=" << instance_id_ << " tablet_id=" << abort_task.tablet_id << " version=[" << abort_task.start_version << '-' << abort_task.end_version << "]"; - return abort_ret; + ret = abort_ret; + continue; } + keys_to_recheck.emplace_back(abort_task.key, abort_task); } - return 0; + return ret; } -int collect_prepare_delete_tasks(TxnKv* txn_kv, const std::string& instance_id, - const std::vector& keys, - std::vector* delete_tasks) { - constexpr size_t kPrepareCheckBatchSize = 256; - for (size_t offset = 0; offset < keys.size(); offset += kPrepareCheckBatchSize) { - size_t limit = std::min(keys.size(), offset + kPrepareCheckBatchSize); - std::unique_ptr txn; - TxnErrorCode err = txn_kv->create_txn(&txn); - if (err != TxnErrorCode::TXN_OK) { - LOG(WARNING) << "failed to create txn, instance_id=" << instance_id; - return -1; +void InstanceRecycler::submit_recycle_prepare_rowsets_job( + SimpleThreadPool& worker_pool, std::vector rowset_keys_to_abort, + std::atomic_long* num_recycled) { + int ret = worker_pool.submit([this, rowset_keys_to_abort = std::move(rowset_keys_to_abort), + num_recycled]() mutable { + std::vector> keys_to_recheck; + int abort_ret = batch_abort_txn_or_job_for_recycle(rowset_keys_to_abort, + keys_to_recheck); + if (abort_ret != 0) { + LOG(WARNING) << "failed to abort some txn or job for related rowset, " + "instance_id=" + << instance_id_; } - for (size_t idx = offset; idx < limit; ++idx) { - const std::string& key = keys[idx]; - std::string val; - err = txn->get(key, &val); - if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { - // has already been removed + if (keys_to_recheck.empty()) { + return; + } + + std::vector> rowsets_to_recycle; + int recheck_ret = batch_recheck_rowsets_after_abort(txn_kv_.get(), instance_id_, + keys_to_recheck, &rowsets_to_recycle); + if (recheck_ret != 0) { + LOG(WARNING) << "failed to recheck some recycle rowsets after abort, instance_id=" + << instance_id_; + } + for (const auto& [key, current_rowset] : rowsets_to_recycle) { + const auto& current_meta = current_rowset.rowset_meta(); + if (delete_rowset_data(current_meta.resource_id(), current_meta.tablet_id(), + current_meta.rowset_id_v2()) != 0) { + LOG(WARNING) << "failed to delete rowset data, key=" << hex(key); continue; } - if (err != TxnErrorCode::TXN_OK) { - LOG(WARNING) << "failed to get recycle rowset, instance_id=" << instance_id - << " key=" << hex(key); - return -1; - } - RecycleRowsetPB rowset; - if (!rowset.ParseFromString(val)) { - LOG(WARNING) << "failed to parse recycle rowset, instance_id=" << instance_id - << " key=" << hex(key); - return -1; + if (delete_versioned_delete_bitmap_kvs(current_meta.tablet_id(), + current_meta.rowset_id_v2()) != 0) { + continue; } - if (rowset.type() != RecycleRowsetPB::PREPARE) { + std::vector keys {key}; + if (txn_remove(txn_kv_.get(), keys) != 0) { + LOG(WARNING) << "failed to delete recycle rowset kv, instance_id=" << instance_id_; continue; } - const auto& rs_meta = rowset.rowset_meta(); - delete_tasks->push_back( - {key, rs_meta.resource_id(), rs_meta.rowset_id_v2(), rs_meta.tablet_id()}); + num_recycled->fetch_add(1, std::memory_order_relaxed); } + }); + if (ret != 0) { + LOG(WARNING) << "failed to submit recycle prepare rowset job, instance_id=" << instance_id_; } - return 0; +} + +void InstanceRecycler::submit_recycle_tmp_rowsets_job(SimpleThreadPool& worker_pool, + std::vector rowset_keys_to_abort, + std::atomic_long* num_recycled, + RecyclerMetricsContext* metrics_context) { + if (rowset_keys_to_abort.empty()) { + return; + } + worker_pool.submit([this, rowset_keys_to_abort = std::move(rowset_keys_to_abort), num_recycled, + metrics_context]() mutable { + std::vector> abort_keys_to_recheck; + int abort_ret = batch_abort_txn_or_job_for_recycle( + rowset_keys_to_abort, abort_keys_to_recheck); + if (abort_ret != 0) { + LOG(WARNING) << "failed to abort some txn or job for related tmp rowset, " + "instance_id=" + << instance_id_; + } + + std::map rowsets_to_delete; + std::vector keys_to_delete; + std::vector ref_count_keys_to_delete; + std::vector> rowsets_to_recycle; + int recheck_ret = batch_recheck_rowsets_after_abort( + txn_kv_.get(), instance_id_, abort_keys_to_recheck, &rowsets_to_recycle); + if (recheck_ret != 0) { + LOG(WARNING) << "failed to recheck some tmp rowsets after abort, instance_id=" + << instance_id_; + } + for (auto& [key, rowset] : rowsets_to_recycle) { + keys_to_delete.push_back(key); + ref_count_keys_to_delete.push_back(versioned::data_rowset_ref_count_key( + {instance_id_, rowset.tablet_id(), rowset.rowset_id_v2()})); + rowsets_to_delete.emplace(rowset.rowset_id_v2(), std::move(rowset)); + } + + if (keys_to_delete.empty()) { + return; + } + if (delete_rowset_data(rowsets_to_delete, RowsetRecyclingState::TMP_ROWSET, + *metrics_context) != 0) { + LOG(WARNING) << "failed to delete tmp rowset data, instance_id=" << instance_id_; + return; + } + for (const auto& [_, rowset] : rowsets_to_delete) { + if (delete_versioned_delete_bitmap_kvs(rowset.tablet_id(), rowset.rowset_id_v2()) != + 0) { + return; + } + if (delete_delete_bitmap_kvs(rowset.tablet_id(), rowset.rowset_id_v2()) != 0) { + return; + } + } + if (txn_remove(txn_kv_.get(), keys_to_delete) != 0) { + LOG(WARNING) << "failed to delete tmp rowset kv, instance_id=" << instance_id_; + return; + } + if (txn_remove(txn_kv_.get(), ref_count_keys_to_delete) != 0) { + LOG(WARNING) << "failed to delete tmp rowset ref count kv, instance_id=" + << instance_id_; + return; + } + num_recycled->fetch_add(keys_to_delete.size(), std::memory_order_relaxed); + }); } int InstanceRecycler::recycle_ref_rowsets(bool* has_unrecycled_rowsets) { @@ -4073,6 +4304,33 @@ int InstanceRecycler::delete_rowset_data(const std::string& resource_id, int64_t return accessor->delete_prefix(rowset_path_prefix(tablet_id, rowset_id)); } +int InstanceRecycler::delete_versioned_delete_bitmap_kvs(int64_t tablet_id, + const std::string& rowset_id) { + std::string dbm_start_key = + versioned::meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id}); + std::string dbm_end_key = dbm_start_key; + encode_int64(INT64_MAX, &dbm_end_key); + int ret = txn_remove(txn_kv_.get(), dbm_start_key, dbm_end_key); + if (ret != 0) { + LOG(WARNING) << "failed to delete versioned delete bitmap kv, instance_id=" << instance_id_ + << " tablet_id=" << tablet_id << " rowset_id=" << rowset_id; + } + return ret; +} + +int InstanceRecycler::delete_delete_bitmap_kvs(int64_t tablet_id, const std::string& rowset_id) { + std::string delete_bitmap_start = + meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id, 0, 0}); + std::string delete_bitmap_end = + meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id, INT64_MAX, INT64_MAX}); + int ret = txn_remove(txn_kv_.get(), delete_bitmap_start, delete_bitmap_end); + if (ret != 0) { + LOG(WARNING) << "failed to delete delete bitmap kv, instance_id=" << instance_id_ + << " tablet_id=" << tablet_id << " rowset_id=" << rowset_id; + } + return ret; +} + bool InstanceRecycler::decode_packed_file_key(std::string_view key, std::string* packed_path) { if (key.empty()) { return false; @@ -4999,32 +5257,14 @@ int InstanceRecycler::recycle_rowsets() { }; std::vector rowset_keys_to_mark_recycled; - std::vector rowset_keys_to_abort; - std::vector prepare_rowset_keys_to_delete; + std::vector rowset_keys_to_abort_job; - // Store keys of rowset recycled by background workers std::mutex async_recycled_rowset_keys_mutex; std::vector async_recycled_rowset_keys; std::vector rowset_keys_without_data; auto worker_pool = std::make_unique( config::instance_recycler_worker_pool_size, "recycle_rowsets"); worker_pool->start(); - auto mark_abort_worker_pool = std::make_unique( - config::instance_recycler_worker_pool_size, "recycle_rs_mark_abort"); - mark_abort_worker_pool->start(); - // TODO bacth delete - auto delete_versioned_delete_bitmap_kvs = [&](int64_t tablet_id, const std::string& rowset_id) { - std::string dbm_start_key = - versioned::meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id}); - std::string dbm_end_key = dbm_start_key; - encode_int64(INT64_MAX, &dbm_end_key); - auto ret = txn_remove(txn_kv_.get(), dbm_start_key, dbm_end_key); - if (ret != 0) { - LOG(WARNING) << "failed to delete versioned delete bitmap kv, instance_id=" - << instance_id_; - } - return ret; - }; auto delete_rowset_data_by_prefix = [&](std::string key, const std::string& resource_id, int64_t tablet_id, const std::string& rowset_id) { // Try to delete rowset data in background thread @@ -5131,31 +5371,7 @@ int InstanceRecycler::recycle_rowsets() { segment_metrics_context_.total_recycled_num += rowset.rowset_meta().num_segments(); return 0; } - auto* rowset_meta = rowset.mutable_rowset_meta(); - if (config::enable_mark_delete_rowset_before_recycle) { - if (need_mark_rowset_as_recycled(rowset)) { - rowset_keys_to_mark_recycled.emplace_back(k); - LOG(INFO) << "rowset queued to mark as recycled, recycler will delete data and kv " - "at next turn, instance_id=" - << instance_id_ << " tablet_id=" << rowset_meta->tablet_id() - << " version=[" << rowset_meta->start_version() << '-' - << rowset_meta->end_version() << "]"; - return 0; - } - } - - if (config::enable_abort_txn_and_job_for_delete_rowset_before_recycle && - rowset_meta->end_version() != 1) { - if (make_deferred_abort_task(rowset).has_value()) { - LOG(INFO) << "rowset queued to abort related txn or job after current scan batch, " - "instance_id=" - << instance_id_ << " tablet_id=" << rowset_meta->tablet_id() - << " version=[" << rowset_meta->start_version() << '-' - << rowset_meta->end_version() << "]"; - rowset_keys_to_abort.emplace_back(k); - } - } // TODO(plat1ko): check rowset not referenced if (!rowset_meta->has_resource_id()) [[unlikely]] { // impossible @@ -5177,8 +5393,32 @@ int InstanceRecycler::recycle_rowsets() { << " creation_time=" << rowset_meta->creation_time() << " task_type=" << metrics_context.operation_type; if (rowset.type() == RecycleRowsetPB::PREPARE) { - // unable to calculate file path, can only be deleted by rowset id prefix + if (config::enable_mark_delete_rowset_before_recycle) { + if (need_mark_rowset_as_recycled(rowset.rowset_meta())) { + rowset_keys_to_mark_recycled.emplace_back(k); + LOG(INFO) << "rowset queued to mark as recycled, recycler will delete data and " + "kv " + "at next turn, instance_id=" + << instance_id_ << " tablet_id=" << rowset_meta->tablet_id() + << " version=[" << rowset_meta->start_version() << '-' + << rowset_meta->end_version() << "]"; + return 0; + } + } + num_prepare += 1; + if (config::enable_abort_txn_and_job_for_delete_rowset_before_recycle && + rowset_meta->end_version() != 1) { + if (make_related_txn_or_job_abort_task(rowset).has_value()) { + LOG(INFO) << "rowset queued to abort related txn or job before recycling, " + "instance_id=" + << instance_id_ << " tablet_id=" << rowset_meta->tablet_id() + << " version=[" << rowset_meta->start_version() << '-' + << rowset_meta->end_version() << "]"; + rowset_keys_to_abort_job.emplace_back(k); + return 0; + } + } if (delete_rowset_data_by_prefix(std::string(k), rowset_meta->resource_id(), rowset_meta->tablet_id(), rowset_meta->rowset_id_v2()) != 0) { @@ -5220,50 +5460,20 @@ int InstanceRecycler::recycle_rowsets() { }); }; - auto submit_mark_abort_rowset_job = [&](std::vector rowset_keys_to_mark, - std::vector rowset_keys_to_abort) { - if (rowset_keys_to_mark.empty() && rowset_keys_to_abort.empty()) { - return; - } - mark_abort_worker_pool->submit([&, rowset_keys_to_mark = std::move(rowset_keys_to_mark), - rowset_keys_to_abort = - std::move(rowset_keys_to_abort)]() mutable { - auto start = steady_clock::now(); - DORIS_CLOUD_DEFER { - auto cost = duration_cast(steady_clock::now() - start).count(); - LOG(INFO) << "finish mark and abort rowset job, instance_id=" << instance_id_ << ' ' - << "cost_ms=" << cost << ' ' - << "rowset_keys_to_mark.size()=" << rowset_keys_to_mark.size() << ' ' - << "rowset_keys_to_abort.size()=" << rowset_keys_to_abort.size(); - }; - if (!rowset_keys_to_mark.empty() && - batch_mark_rowsets_as_recycled(txn_kv_.get(), instance_id_, - rowset_keys_to_mark) != 0) { - LOG(WARNING) << "failed to batch mark rowsets as recycled, instance_id=" - << instance_id_ << ' ' - << "rowset_keys_to_mark.size()=" << rowset_keys_to_mark.size(); - return; - } - if (!rowset_keys_to_abort.empty() && - batch_abort_txn_or_job_for_recycle(rowset_keys_to_abort, true) != - 0) { - LOG(WARNING) << "failed to batch abort txn or job for related rowset, " - "instance_id=" - << instance_id_ << ' ' - << "rowset_keys_to_abort.size()=" << rowset_keys_to_abort.size(); - return; - } - }); - }; - bool scan_finished = false; auto loop_done = [&]() -> int { std::vector mark_keys_to_process; - std::vector abort_keys_to_process; + std::vector abort_job_keys_to_process; mark_keys_to_process.swap(rowset_keys_to_mark_recycled); - abort_keys_to_process.swap(rowset_keys_to_abort); - submit_mark_abort_rowset_job(std::move(mark_keys_to_process), - std::move(abort_keys_to_process)); + abort_job_keys_to_process.swap(rowset_keys_to_abort_job); + if (!mark_keys_to_process.empty()) { + submit_batch_mark_rowsets_as_recycled_job( + *worker_pool, std::move(mark_keys_to_process)); + } + if (!abort_job_keys_to_process.empty()) { + submit_recycle_prepare_rowsets_job(*worker_pool, std::move(abort_job_keys_to_process), + &num_recycled); + } if (!scan_finished && rowsets.size() < delete_rowset_batch_size) { return 0; } @@ -5331,7 +5541,6 @@ int InstanceRecycler::recycle_rowsets() { ret = -1; } - mark_abort_worker_pool->stop(); worker_pool->stop(); if (!async_recycled_rowset_keys.empty()) { @@ -5939,8 +6148,7 @@ int InstanceRecycler::recycle_tmp_rowsets() { .tag("expired_rowset_meta_size_recycled", expired_rowset_size); }; - // Elements in `tmp_rowset_keys` has the same lifetime as `it` - + // Store direct-delete keys separately from keys whose related txn or job must be aborted. std::vector tmp_rowset_keys; std::vector tmp_rowset_ref_count_keys; std::vector tmp_rowset_keys_to_mark_recycled; @@ -5955,11 +6163,11 @@ int InstanceRecycler::recycle_tmp_rowsets() { int64_t earlest_ts = std::numeric_limits::max(); - auto handle_rowset_kv = [&num_scanned, &num_expired, &tmp_rowset_keys, &tmp_rowsets, - &expired_rowset_size, &total_rowset_key_size, &total_rowset_value_size, - &earlest_ts, &tmp_rowset_ref_count_keys, + auto handle_rowset_kv = [&num_scanned, &num_expired, &tmp_rowset_keys, &expired_rowset_size, + &total_rowset_key_size, &total_rowset_value_size, &earlest_ts, &tmp_rowset_keys_to_mark_recycled, &tmp_rowset_keys_to_abort, this, - &metrics_context](std::string_view k, std::string_view v) -> int { + &metrics_context, &tmp_rowsets, &tmp_rowset_ref_count_keys]( + std::string_view k, std::string_view v) -> int { ++num_scanned; total_rowset_key_size += k.size(); total_rowset_value_size += v.size(); @@ -5977,6 +6185,26 @@ int InstanceRecycler::recycle_tmp_rowsets() { if (current_time < expiration) { // not expired return 0; } + ++num_expired; + expired_rowset_size += v.size(); + if (!rowset.has_resource_id()) { + if (rowset.num_segments() > 0) [[unlikely]] { // impossible + LOG_WARNING("rowset meta has empty resource id").tag("key", k); + return -1; + } + // might be a delete pred rowset + tmp_rowset_keys.emplace_back(k); + return 0; + } + + // TODO(plat1ko): check rowset not referenced + LOG(INFO) << "delete rowset data, instance_id=" << instance_id_ + << " tablet_id=" << rowset.tablet_id() << " rowset_id=" << rowset.rowset_id_v2() + << " version=[" << rowset.start_version() << '-' << rowset.end_version() + << "] txn_id=" << rowset.txn_id() << " rowset_meta_size=" << v.size() + << " creation_time=" << rowset.creation_time() << " num_scanned=" << num_scanned + << " num_expired=" << num_expired + << " task_type=" << metrics_context.operation_type; if (config::enable_mark_delete_rowset_before_recycle) { if (need_mark_rowset_as_recycled(rowset)) { @@ -5990,35 +6218,16 @@ int InstanceRecycler::recycle_tmp_rowsets() { } if (config::enable_abort_txn_and_job_for_delete_rowset_before_recycle) { - if (make_deferred_abort_task(rowset).has_value()) { + if (make_related_txn_or_job_abort_task(rowset).has_value()) { LOG(INFO) << "rowset queued to abort related txn or job after current scan batch, " "instance_id=" << instance_id_ << " tablet_id=" << rowset.tablet_id() << " version=[" << rowset.start_version() << '-' << rowset.end_version() << "]"; tmp_rowset_keys_to_abort.emplace_back(k); + return 0; } } - ++num_expired; - expired_rowset_size += v.size(); - if (!rowset.has_resource_id()) { - if (rowset.num_segments() > 0) [[unlikely]] { // impossible - LOG_WARNING("rowset meta has empty resource id").tag("key", k); - return -1; - } - // might be a delete pred rowset - tmp_rowset_keys.emplace_back(k); - return 0; - } - // TODO(plat1ko): check rowset not referenced - LOG(INFO) << "delete rowset data, instance_id=" << instance_id_ - << " tablet_id=" << rowset.tablet_id() << " rowset_id=" << rowset.rowset_id_v2() - << " version=[" << rowset.start_version() << '-' << rowset.end_version() - << "] txn_id=" << rowset.txn_id() << " rowset_meta_size=" << v.size() - << " creation_time=" << rowset.creation_time() << " num_scanned=" << num_scanned - << " num_expired=" << num_expired - << " task_type=" << metrics_context.operation_type; - tmp_rowset_keys.emplace_back(k.data(), k.size()); // Remove the rowset ref count key directly since it has not been used. std::string rowset_ref_count_key = versioned::data_rowset_ref_count_key( @@ -6031,34 +6240,6 @@ int InstanceRecycler::recycle_tmp_rowsets() { return 0; }; - // TODO bacth delete - auto delete_versioned_delete_bitmap_kvs = [&](int64_t tablet_id, const std::string& rowset_id) { - std::string dbm_start_key = - versioned::meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id}); - std::string dbm_end_key = dbm_start_key; - encode_int64(INT64_MAX, &dbm_end_key); - auto ret = txn_remove(txn_kv_.get(), dbm_start_key, dbm_end_key); - if (ret != 0) { - LOG(WARNING) << "failed to delete versioned delete bitmap kv, instance_id=" - << instance_id_ << ", tablet_id=" << tablet_id - << ", rowset_id=" << rowset_id; - } - return ret; - }; - - auto delete_delete_bitmap_kvs = [&](int64_t tablet_id, const std::string& rowset_id) { - auto delete_bitmap_start = - meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id, 0, 0}); - auto delete_bitmap_end = - meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id, INT64_MAX, INT64_MAX}); - auto ret = txn_remove(txn_kv_.get(), delete_bitmap_start, delete_bitmap_end); - if (ret != 0) { - LOG(WARNING) << "failed to delete delete bitmap kv, instance_id=" << instance_id_ - << ", tablet_id=" << tablet_id << ", rowset_id=" << rowset_id; - } - return ret; - }; - auto loop_done = [&]() -> int { std::vector tmp_rowset_keys_to_delete; std::vector tmp_rowset_ref_count_keys_to_delete; @@ -6070,26 +6251,20 @@ int InstanceRecycler::recycle_tmp_rowsets() { tmp_rowset_ref_count_keys_to_delete.swap(tmp_rowset_ref_count_keys); mark_keys_to_process.swap(tmp_rowset_keys_to_mark_recycled); abort_keys_to_process.swap(tmp_rowset_keys_to_abort); + if (!mark_keys_to_process.empty()) { + submit_batch_mark_rowsets_as_recycled_job( + *worker_pool, std::move(mark_keys_to_process)); + } + if (!abort_keys_to_process.empty()) { + submit_recycle_tmp_rowsets_job(*worker_pool, std::move(abort_keys_to_process), + &num_recycled, &metrics_context); + } worker_pool->submit([&, tmp_rowset_keys_to_delete = std::move(tmp_rowset_keys_to_delete), tmp_rowsets_to_delete = std::move(tmp_rowsets_to_delete), tmp_rowset_ref_count_keys_to_delete = std::move(tmp_rowset_ref_count_keys_to_delete), mark_keys_to_process = std::move(mark_keys_to_process), abort_keys_to_process = std::move(abort_keys_to_process)]() mutable { - if (!mark_keys_to_process.empty() && - batch_mark_rowsets_as_recycled(txn_kv_.get(), instance_id_, - mark_keys_to_process) != 0) { - LOG(WARNING) << "failed to batch mark tmp rowsets as recycled, instance_id=" - << instance_id_; - return; - } - if (!abort_keys_to_process.empty() && - batch_abort_txn_or_job_for_recycle(abort_keys_to_process, - false) != 0) { - LOG(WARNING) << "failed to batch abort txn or job for releated rowset, instance_id=" - << instance_id_; - return; - } if (delete_rowset_data(tmp_rowsets_to_delete, RowsetRecyclingState::TMP_ROWSET, metrics_context) != 0) { LOG(WARNING) << "failed to delete tmp rowset data, instance_id=" << instance_id_; @@ -7350,7 +7525,9 @@ int InstanceRecycler::scan_and_statistics_rowsets() { return 0; } - if(!rowset_meta->has_is_recycled() || !rowset_meta->is_recycled()) { + if (config::enable_mark_delete_rowset_before_recycle && + rowset.type() == RecycleRowsetPB::PREPARE && + (!rowset_meta->has_is_recycled() || !rowset_meta->is_recycled())) { return 0; } diff --git a/cloud/src/recycler/recycler.h b/cloud/src/recycler/recycler.h index 2e8b6d57f7f3ae..e3e36669803b8e 100644 --- a/cloud/src/recycler/recycler.h +++ b/cloud/src/recycler/recycler.h @@ -22,6 +22,7 @@ #include #include +#include #include #include #include @@ -61,7 +62,6 @@ class SegmentRecyclerMetricsContext; int64_t calculate_tmp_rowset_expired_time( const std::string& instance_id_, const doris::RowsetMetaCloudPB& tmp_rowset_meta_pb, int64_t* earlest_ts /* tmp_rowset earliest expiration ts */); - struct RecyclerThreadPoolGroup { RecyclerThreadPoolGroup() = default; RecyclerThreadPoolGroup(std::shared_ptr s3_producer_pool, @@ -254,6 +254,21 @@ class SegmentRecyclerMetricsContext : public RecyclerMetricsContext { struct OplogRecycleStats; +struct RelatedTxnOrJobAbortTask { + enum class Type : uint8_t { + TXN, + JOB, + }; + + Type type = Type::TXN; + std::string key; + int64_t txn_id = 0; + int64_t tablet_id = 0; + int64_t start_version = 0; + int64_t end_version = 0; + std::string rowset_id; + std::string job_id; +}; class InstanceRecycler { public: struct PackedFileRecycleStats { @@ -469,6 +484,10 @@ class InstanceRecycler { int delete_rowset_data(const std::string& resource_id, int64_t tablet_id, const std::string& rowset_id); + int delete_versioned_delete_bitmap_kvs(int64_t tablet_id, const std::string& rowset_id); + + int delete_delete_bitmap_kvs(int64_t tablet_id, const std::string& rowset_id); + // return 0 for success otherwise error int delete_rowset_data(const std::map& rowsets, RowsetRecyclingState type, RecyclerMetricsContext& metrics_context); @@ -588,23 +607,38 @@ class InstanceRecycler { // reference non-existent data. // // Solution: - // Before recycling the rowset data, this function aborts the associated transaction/job to ensure - // it cannot be committed. This guarantees that: - // 1. The transaction/job state is marked as ABORTED - // 2. Any subsequent commit_rowset/commit_txn attempts will fail - // 3. The rowset data can be safely deleted without risk of data loss + // Before recycling rowset data, try to abort the associated transaction/job. A zero return only + // permits the caller to recheck the recycle key. Object data can be deleted only when the key + // still exists, still describes the same PREPARE rowset, and still belongs to the same owner. // // Parameters: // txn_id: The transaction/job ID associated with the rowset to be recycled // // Returns: - // 0 on success, -1 on failure + // 0 if the recycle key may be rechecked before deletion. The caller must never delete + // object data directly from the scan snapshot. + // Non-zero if object data and the recycle key must be retained for a later retry. int abort_txn_for_related_rowset(int64_t txn_id); - int abort_job_for_related_rowset(const RowsetMetaCloudPB& rowset_meta); + int abort_job_for_related_rowset(int64_t tablet_id, const std::string& rowset_id, + const std::string& job_id); + + template + int batch_abort_txn_or_job_for_recycle( + const std::vector& keys, + std::vector>& keys_to_recheck); template - int batch_abort_txn_or_job_for_recycle(const std::vector& keys, - bool skip_base_version); + void submit_batch_mark_rowsets_as_recycled_job(SimpleThreadPool& worker_pool, + std::vector rowset_keys_to_mark); + + void submit_recycle_prepare_rowsets_job(SimpleThreadPool& worker_pool, + std::vector rowset_keys_to_abort, + std::atomic_long* num_recycled); + + void submit_recycle_tmp_rowsets_job(SimpleThreadPool& worker_pool, + std::vector rowset_keys_to_abort, + std::atomic_long* num_recycled, + RecyclerMetricsContext* metrics_context); private: std::atomic_bool stopped_ {false}; diff --git a/cloud/test/recycler_test.cpp b/cloud/test/recycler_test.cpp index 28ce6b0f716f4a..b795a2cb3c37c6 100644 --- a/cloud/test/recycler_test.cpp +++ b/cloud/test/recycler_test.cpp @@ -1053,7 +1053,7 @@ static int get_txn_info(std::shared_ptr txn_kv, std::string instance_id, LOG(WARNING) << "ParseFromString failed"; return -3; } - LOG(INFO) << "txn_info_pb" << txn_info_pb.DebugString(); + LOG(INFO) << "txn_info_pb" << ' ' << txn_info_pb.DebugString(); err = txn->commit(); if (err != TxnErrorCode::TXN_OK) { LOG(WARNING) << "txn->commit failed, err=" << err; @@ -1062,6 +1062,98 @@ static int get_txn_info(std::shared_ptr txn_kv, std::string instance_id, return 0; } +static int create_prepared_txn(TxnKv* txn_kv, int64_t db_id, int64_t tablet_id, int64_t txn_id) { + TxnIndexPB txn_index_pb; + txn_index_pb.mutable_tablet_index()->set_db_id(db_id); + txn_index_pb.mutable_tablet_index()->set_tablet_id(tablet_id); + + TxnInfoPB txn_info_pb; + txn_info_pb.set_txn_id(txn_id); + txn_info_pb.set_status(TxnStatusPB::TXN_STATUS_PREPARED); + + std::unique_ptr txn; + if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) { + return -1; + } + txn->put(txn_index_key({instance_id, txn_id}), txn_index_pb.SerializeAsString()); + txn->put(txn_info_key({instance_id, db_id, txn_id}), txn_info_pb.SerializeAsString()); + return txn->commit() == TxnErrorCode::TXN_OK ? 0 : -1; +} + +static InstanceInfoPB create_recycler_test_instance(const std::string& resource_id) { + InstanceInfoPB instance; + instance.set_instance_id(instance_id); + auto obj_info = instance.add_obj_info(); + obj_info->set_id(resource_id); + obj_info->set_ak(config::test_s3_ak); + obj_info->set_sk(config::test_s3_sk); + obj_info->set_endpoint(config::test_s3_endpoint); + obj_info->set_region(config::test_s3_region); + obj_info->set_bucket(config::test_s3_bucket); + obj_info->set_prefix(resource_id); + return instance; +} + +static bool get_recycle_rowset(TxnKv* txn_kv, int64_t tablet_id, const std::string& rowset_id, + RecycleRowsetPB* rowset) { + std::unique_ptr txn; + if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) { + return false; + } + std::string val; + if (txn->get(recycle_rowset_key({instance_id, tablet_id, rowset_id}), &val) != + TxnErrorCode::TXN_OK) { + return false; + } + return rowset->ParseFromString(val) && txn->commit() == TxnErrorCode::TXN_OK; +} + +static int put_tablet_job(TxnKv* txn_kv, const TabletIndexPB& key_idx, const TabletJobInfoPB& job) { + std::unique_ptr txn; + if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) { + return -1; + } + txn->put(job_tablet_key({instance_id, key_idx.table_id(), key_idx.index_id(), + key_idx.partition_id(), key_idx.tablet_id()}), + job.SerializeAsString()); + return txn->commit() == TxnErrorCode::TXN_OK ? 0 : -1; +} + +static bool get_tablet_job(TxnKv* txn_kv, const TabletIndexPB& key_idx, TabletJobInfoPB* job) { + std::unique_ptr txn; + if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) { + return false; + } + std::string val; + if (txn->get(job_tablet_key({instance_id, key_idx.table_id(), key_idx.index_id(), + key_idx.partition_id(), key_idx.tablet_id()}), + &val) != TxnErrorCode::TXN_OK) { + return false; + } + return job->ParseFromString(val) && txn->commit() == TxnErrorCode::TXN_OK; +} + +struct RecyclePrepareRowsetConfigGuard { + explicit RecyclePrepareRowsetConfigGuard(bool enable_mark) + : old_enable_force_recycle(config::force_immediate_recycle), + old_enable_mark(config::enable_mark_delete_rowset_before_recycle), + old_enable_abort(config::enable_abort_txn_and_job_for_delete_rowset_before_recycle) { + config::force_immediate_recycle = true; + config::enable_mark_delete_rowset_before_recycle = enable_mark; + config::enable_abort_txn_and_job_for_delete_rowset_before_recycle = true; + } + + ~RecyclePrepareRowsetConfigGuard() { + config::force_immediate_recycle = old_enable_force_recycle; + config::enable_mark_delete_rowset_before_recycle = old_enable_mark; + config::enable_abort_txn_and_job_for_delete_rowset_before_recycle = old_enable_abort; + } + + bool old_enable_force_recycle; + bool old_enable_mark; + bool old_enable_abort; +}; + static int check_recycle_txn_keys(std::shared_ptr txn_kv, std::string instance_id, int64_t db_id, int64_t txn_id, const std::string& label) { std::string txn_inf_key; @@ -1267,167 +1359,1081 @@ static int get_copy_file_num(TxnKv* txn_kv, const std::string& stage_id, int64_t return 0; } -static void check_delete_bitmap_keys_size(TxnKv* txn_kv, int64_t tablet_id, int expected_size, - int version = 2) { +static void check_delete_bitmap_keys_size(TxnKv* txn_kv, int64_t tablet_id, int expected_size, + int version = 2) { + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + std::unique_ptr it; + std::string dbm_start_key, dbm_end_key; + if (version == 2) { + dbm_start_key = versioned::meta_delete_bitmap_key({instance_id, tablet_id, ""}); + dbm_end_key = versioned::meta_delete_bitmap_key({instance_id, tablet_id + 1, ""}); + } else if (version == 1) { + dbm_start_key = meta_delete_bitmap_key({instance_id, tablet_id, "", 0, 0}); + dbm_end_key = meta_delete_bitmap_key({instance_id, tablet_id + 1, "", 0, 0}); + } + int size = 0; + while (it == nullptr /* may be not init */ || it->more()) { + ASSERT_EQ(txn->get(dbm_start_key, dbm_end_key, &it), TxnErrorCode::TXN_OK); + while (it->has_next()) { + it->next(); + size++; + } + dbm_start_key = it->next_begin_key(); + } + EXPECT_EQ(size, expected_size); +} + +static void check_delete_bitmap_file_size(std::shared_ptr accessor, + int64_t tablet_id, int expected_size) { + int size = 0; + std::unique_ptr list_iter; + ASSERT_EQ(0, accessor->list_directory(tablet_path_prefix(tablet_id), &list_iter)); + while (list_iter->has_next()) { + auto file_info = list_iter->next(); + ASSERT_TRUE(file_info.has_value()); + std::string filename = file_info.value().path; + if (filename.ends_with("_delete_bitmap.db")) { + size++; + } + } + EXPECT_EQ(size, expected_size); +} + +TEST(RecyclerTest, recycle_empty) { + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + InstanceInfoPB instance; + instance.set_instance_id(instance_id); + auto obj_info = instance.add_obj_info(); + obj_info->set_id("recycle_empty"); + obj_info->set_ak(config::test_s3_ak); + obj_info->set_sk(config::test_s3_sk); + obj_info->set_endpoint(config::test_s3_endpoint); + obj_info->set_region(config::test_s3_region); + obj_info->set_bucket(config::test_s3_bucket); + obj_info->set_prefix("recycle_empty"); + + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); +} + +TEST(RecyclerTest, recycle_rowsets) { + config::retention_seconds = 0; + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + InstanceInfoPB instance; + instance.set_instance_id(instance_id); + auto obj_info = instance.add_obj_info(); + obj_info->set_id("recycle_rowsets"); + obj_info->set_ak(config::test_s3_ak); + obj_info->set_sk(config::test_s3_sk); + obj_info->set_endpoint(config::test_s3_endpoint); + obj_info->set_region(config::test_s3_region); + obj_info->set_bucket(config::test_s3_bucket); + obj_info->set_prefix("recycle_rowsets"); + + config::instance_recycler_worker_pool_size = 1; + + int insert_no_inverted_index = 0; + int insert_inverted_index = 0; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + SyncPoint::get_instance()->clear_all_call_backs(); + }; + sp->set_call_back("InvertedIndexIdCache::insert1", [&](auto&&) { ++insert_no_inverted_index; }); + sp->set_call_back("InvertedIndexIdCache::insert2", [&](auto&&) { ++insert_inverted_index; }); + sp->enable_processing(); + + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + + std::vector schemas; + for (int i = 0; i < 5; ++i) { + auto& schema = schemas.emplace_back(); + schema.set_schema_version(i); + schema.set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V1); + for (int j = 0; j < i; ++j) { + auto index = schema.add_index(); + index->set_index_id(j); + index->set_index_type(IndexType::INVERTED); + } + } + + auto accessor = recycler.accessor_map_.begin()->second; + constexpr int index_id = 10001, tablet_id = 10002; + for (int i = 0; i < 1000; ++i) { + auto rowset = create_rowset("recycle_rowsets", tablet_id, index_id, 5, schemas[i % 5]); + create_recycle_rowset( + txn_kv.get(), accessor.get(), rowset, + static_cast(i % (RecycleRowsetPB::Type_MAX + 1)), i & 1, + i < 500); + } + for (int i = 0; i < 1000; ++i) { + auto rowset = create_rowset("recycle_rowsets", tablet_id, index_id, 5, schemas[i % 5]); + create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::COMPACT, true, + i < 500); + } + check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 1000); + check_delete_bitmap_file_size(accessor, tablet_id, 1000); + + for (size_t i = 0; i < 10; i++) { + ASSERT_EQ(recycler.recycle_rowsets(), 0); + } + + // check rowset does not exist on obj store + std::unique_ptr list_iter; + ASSERT_EQ(0, accessor->list_directory(tablet_path_prefix(tablet_id), &list_iter)); + for (auto file = list_iter->next(); file.has_value(); file = list_iter->next()) { + LOG(INFO) << "file: " << file->path; + } + EXPECT_FALSE(list_iter->has_next()); + // check all recycle rowset kv have been deleted + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + std::unique_ptr it; + auto begin_key = recycle_key_prefix(instance_id); + auto end_key = recycle_key_prefix(instance_id + '\xff'); + ASSERT_EQ(txn->get(begin_key, end_key, &it), TxnErrorCode::TXN_OK); + EXPECT_EQ(it->size(), 0); + // Check InvertedIndexIdCache + EXPECT_EQ(insert_inverted_index, 4); + EXPECT_EQ(insert_no_inverted_index, 1); + // check all versioned delete bitmap kv have been deleted + check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 0); + check_delete_bitmap_file_size(accessor, tablet_id, 0); +} + +TEST(RecyclerTest, next_recycle_rowset_tablet_key_overwrites_existing_buffer) { + std::string next_key = recycle_rowset_key({instance_id, 10002, "rowset"}); + ASSERT_EQ(InstanceRecycler::next_recycle_rowset_tablet_key(instance_id, 10002, &next_key), 0); + + std::string_view k1 = next_key; + k1.remove_prefix(1); + std::vector, int, int>> out; + ASSERT_EQ(decode_key(&k1, &out), 0); + EXPECT_EQ(std::get(std::get<0>(out[3])), 10003); + EXPECT_TRUE(std::get(std::get<0>(out[4])).empty()); +} + +TEST(RecyclerTest, recycle_rowsets_only_marks_prepare_rowsets_as_recycled) { + config::retention_seconds = 0; + auto old_enable_mark = config::enable_mark_delete_rowset_before_recycle; + config::enable_mark_delete_rowset_before_recycle = true; + DORIS_CLOUD_DEFER { + config::enable_mark_delete_rowset_before_recycle = old_enable_mark; + }; + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + InstanceInfoPB instance; + instance.set_instance_id(instance_id); + auto obj_info = instance.add_obj_info(); + obj_info->set_id("recycle_rowsets_only_marks_prepare"); + obj_info->set_ak(config::test_s3_ak); + obj_info->set_sk(config::test_s3_sk); + obj_info->set_endpoint(config::test_s3_endpoint); + obj_info->set_region(config::test_s3_region); + obj_info->set_bucket(config::test_s3_bucket); + obj_info->set_prefix("recycle_rowsets_only_marks_prepare"); + + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + constexpr int64_t index_id = 10001; + constexpr int64_t tablet_id = 10002; + auto prepare_rowset = + create_rowset("recycle_rowsets_only_marks_prepare", tablet_id, index_id, 1, schema); + auto compact_rowset = + create_rowset("recycle_rowsets_only_marks_prepare", tablet_id, index_id, 1, schema); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), prepare_rowset, + RecycleRowsetPB::PREPARE, true), + 0); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), compact_rowset, + RecycleRowsetPB::COMPACT, true), + 0); + + std::atomic marked_prepare_rowset_count = 0; + std::atomic marked_non_prepare_rowset_count = 0; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + }; + sp->set_call_back("InstanceRecycler::batch_mark_rowsets_as_recycled", [&](auto&& args) { + auto* type = try_any_cast(args[0]); + if (*type == RecycleRowsetPB::PREPARE) { + ++marked_prepare_rowset_count; + } else { + ++marked_non_prepare_rowset_count; + } + }); + sp->enable_processing(); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(marked_prepare_rowset_count.load(), 1); + EXPECT_EQ(marked_non_prepare_rowset_count.load(), 0); +} + +TEST(RecyclerTest, recycle_prepare_rowset_aborts_before_delete) { + RecyclePrepareRowsetConfigGuard config_guard(true); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_prepare_rowset_abort_before_delete"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t txn_db_id = 1000; + constexpr int64_t txn_id = 70001; + constexpr int64_t table_id = 10000; + constexpr int64_t index_id = 10001; + constexpr int64_t tablet_id = 10002; + constexpr int64_t partition_id = 10003; + ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id), 0); + ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id, txn_id), 0); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + auto rowset = create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema, txn_id); + rowset.set_start_version(2); + rowset.set_end_version(2); + rowset.mutable_load_id()->set_hi(123); + rowset.mutable_load_id()->set_lo(456); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::PREPARE, + true, true), + 0); + + std::atomic delete_saw_aborted = false; + std::atomic delete_prefix_calls = 0; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("MockAccessor::delete_prefix", [&](auto&&) { + ++delete_prefix_calls; + TxnInfoPB txn_info; + if (get_txn_info(txn_kv, instance_id, txn_db_id, txn_id, txn_info) == 0) { + delete_saw_aborted.store(txn_info.status() == TxnStatusPB::TXN_STATUS_ABORTED, + std::memory_order_relaxed); + } + }); + sp->enable_processing(); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + RecycleRowsetPB marked_rowset; + ASSERT_TRUE(get_recycle_rowset(txn_kv.get(), tablet_id, rowset.rowset_id_v2(), &marked_rowset)); + EXPECT_TRUE(marked_rowset.rowset_meta().is_recycled()); + TxnInfoPB txn_info; + ASSERT_EQ(get_txn_info(txn_kv, instance_id, txn_db_id, txn_id, txn_info), 0); + EXPECT_EQ(txn_info.status(), TxnStatusPB::TXN_STATUS_PREPARED); + EXPECT_EQ(delete_prefix_calls.load(), 0); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 0); + check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 1); + check_delete_bitmap_file_size(accessor, tablet_id, 1); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_TRUE(delete_saw_aborted.load(std::memory_order_relaxed)); + EXPECT_EQ(delete_prefix_calls.load(), 1); + EXPECT_EQ(count_recycle_rowsets(txn_kv.get(), tablet_id), 0); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 1); + check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 0); + check_delete_bitmap_file_size(accessor, tablet_id, 0); + ASSERT_EQ(get_txn_info(txn_kv, instance_id, txn_db_id, txn_id, txn_info), 0); + EXPECT_EQ(txn_info.status(), TxnStatusPB::TXN_STATUS_ABORTED); +} + +TEST(RecyclerTest, recycle_prepare_compaction_job_aborts_before_delete) { + RecyclePrepareRowsetConfigGuard config_guard(true); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_prepare_compaction_job"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t table_id = 11000; + constexpr int64_t index_id = 11001; + constexpr int64_t partition_id = 11002; + constexpr int64_t tablet_id = 11003; + ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id), 0); + + TabletIndexPB tablet_idx; + ASSERT_EQ(get_tablet_idx(txn_kv.get(), instance_id, tablet_id, tablet_idx), 0); + TabletJobInfoPB job; + job.mutable_idx()->CopyFrom(tablet_idx); + auto* other_compaction = job.add_compaction(); + other_compaction->set_id("other_compaction"); + other_compaction->set_expiration(current_time + 1000); + auto* target_compaction = job.add_compaction(); + target_compaction->set_id("target_compaction"); + target_compaction->set_expiration(current_time + 1000); + ASSERT_EQ(put_tablet_job(txn_kv.get(), tablet_idx, job), 0); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + auto rowset = create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema); + rowset.set_start_version(2); + rowset.set_end_version(2); + rowset.set_job_id(target_compaction->id()); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::PREPARE, + true, true), + 0); + + std::atomic delete_saw_job_aborted = false; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("MockAccessor::delete_prefix", [&](auto&&) { + TabletJobInfoPB current_job; + if (!get_tablet_job(txn_kv.get(), tablet_idx, ¤t_job)) { + return; + } + bool target_exists = false; + for (const auto& compaction : current_job.compaction()) { + target_exists |= compaction.id() == "target_compaction"; + } + delete_saw_job_aborted.store(!target_exists, std::memory_order_relaxed); + }); + sp->enable_processing(); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + RecycleRowsetPB marked_rowset; + ASSERT_TRUE(get_recycle_rowset(txn_kv.get(), tablet_id, rowset.rowset_id_v2(), &marked_rowset)); + EXPECT_TRUE(marked_rowset.rowset_meta().is_recycled()); + TabletJobInfoPB current_job; + ASSERT_TRUE(get_tablet_job(txn_kv.get(), tablet_idx, ¤t_job)); + EXPECT_EQ(current_job.compaction_size(), 2); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 0); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_TRUE(delete_saw_job_aborted.load(std::memory_order_relaxed)); + ASSERT_TRUE(get_tablet_job(txn_kv.get(), tablet_idx, ¤t_job)); + ASSERT_EQ(current_job.compaction_size(), 1); + EXPECT_EQ(current_job.compaction(0).id(), "other_compaction"); + EXPECT_EQ(count_recycle_rowsets(txn_kv.get(), tablet_id), 0); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 1); + check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 0); + check_delete_bitmap_file_size(accessor, tablet_id, 0); +} + +TEST(RecyclerTest, recycle_prepare_schema_change_job_aborts_before_delete) { + RecyclePrepareRowsetConfigGuard config_guard(true); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_prepare_schema_change_job"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t table_id = 12000; + constexpr int64_t base_index_id = 12001; + constexpr int64_t new_index_id = 12002; + constexpr int64_t partition_id = 12003; + constexpr int64_t base_tablet_id = 12004; + constexpr int64_t new_tablet_id = 12005; + ASSERT_EQ(create_tablet(txn_kv.get(), table_id, base_index_id, partition_id, base_tablet_id), + 0); + ASSERT_EQ(create_tablet(txn_kv.get(), table_id, new_index_id, partition_id, new_tablet_id), 0); + + TabletIndexPB base_tablet_idx; + TabletIndexPB new_tablet_idx; + ASSERT_EQ(get_tablet_idx(txn_kv.get(), instance_id, base_tablet_id, base_tablet_idx), 0); + ASSERT_EQ(get_tablet_idx(txn_kv.get(), instance_id, new_tablet_id, new_tablet_idx), 0); + + TabletJobInfoPB job; + job.mutable_idx()->CopyFrom(base_tablet_idx); + auto* schema_change = job.mutable_schema_change(); + schema_change->set_id("schema_change_job"); + schema_change->set_initiator("BE1"); + schema_change->set_expiration(current_time + 1000); + schema_change->mutable_new_tablet_idx()->CopyFrom(new_tablet_idx); + ASSERT_EQ(put_tablet_job(txn_kv.get(), base_tablet_idx, job), 0); + ASSERT_EQ(put_tablet_job(txn_kv.get(), new_tablet_idx, job), 0); + + doris::TabletMetaCloudPB new_tablet_meta; + new_tablet_meta.set_tablet_id(new_tablet_id); + new_tablet_meta.set_tablet_state(doris::TabletStatePB::PB_NOTREADY); + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->put(meta_tablet_key({instance_id, table_id, new_index_id, partition_id, new_tablet_id}), + new_tablet_meta.SerializeAsString()); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + auto rowset = create_rowset(std::string(resource_id), new_tablet_id, new_index_id, 1, schema); + rowset.set_start_version(2); + rowset.set_end_version(2); + rowset.set_job_id(schema_change->id()); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::PREPARE, + true, true), + 0); + + std::atomic delete_saw_both_jobs_aborted = false; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("MockAccessor::delete_prefix", [&](auto&&) { + TabletJobInfoPB base_job; + TabletJobInfoPB new_job; + if (get_tablet_job(txn_kv.get(), base_tablet_idx, &base_job) && + get_tablet_job(txn_kv.get(), new_tablet_idx, &new_job)) { + delete_saw_both_jobs_aborted.store( + !base_job.has_schema_change() && !new_job.has_schema_change(), + std::memory_order_relaxed); + } + }); + sp->enable_processing(); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + RecycleRowsetPB marked_rowset; + ASSERT_TRUE( + get_recycle_rowset(txn_kv.get(), new_tablet_id, rowset.rowset_id_v2(), &marked_rowset)); + EXPECT_TRUE(marked_rowset.rowset_meta().is_recycled()); + TabletJobInfoPB current_job; + ASSERT_TRUE(get_tablet_job(txn_kv.get(), base_tablet_idx, ¤t_job)); + EXPECT_TRUE(current_job.has_schema_change()); + ASSERT_TRUE(get_tablet_job(txn_kv.get(), new_tablet_idx, ¤t_job)); + EXPECT_TRUE(current_job.has_schema_change()); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_TRUE(delete_saw_both_jobs_aborted.load(std::memory_order_relaxed)); + ASSERT_TRUE(get_tablet_job(txn_kv.get(), base_tablet_idx, ¤t_job)); + EXPECT_FALSE(current_job.has_schema_change()); + ASSERT_TRUE(get_tablet_job(txn_kv.get(), new_tablet_idx, ¤t_job)); + EXPECT_FALSE(current_job.has_schema_change()); + EXPECT_EQ(count_recycle_rowsets(txn_kv.get(), new_tablet_id), 0); + EXPECT_EQ(accessor->exists(segment_path(new_tablet_id, rowset.rowset_id_v2(), 0)), 1); + check_delete_bitmap_keys_size(txn_kv.get(), new_tablet_id, 0); + check_delete_bitmap_file_size(accessor, new_tablet_id, 0); +} + +TEST(RecyclerTest, recycle_prepare_rowset_compatibility_paths) { + RecyclePrepareRowsetConfigGuard config_guard(true); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_prepare_compatibility"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t txn_db_id = 13000; + constexpr int64_t base_txn_id = 13001; + constexpr int64_t missing_txn_id = 13002; + constexpr int64_t index_id = 13003; + constexpr int64_t tablet_id = 13004; + ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id, base_txn_id), 0); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + std::vector rowsets; + auto& no_association = rowsets.emplace_back( + create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema)); + no_association.set_start_version(2); + no_association.set_end_version(2); + + auto& base_version = rowsets.emplace_back( + create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema, base_txn_id)); + base_version.set_start_version(0); + base_version.set_end_version(1); + base_version.mutable_load_id()->set_hi(1); + base_version.mutable_load_id()->set_lo(1); + + auto& missing_txn = rowsets.emplace_back(create_rowset(std::string(resource_id), tablet_id, + index_id, 1, schema, missing_txn_id)); + missing_txn.set_start_version(2); + missing_txn.set_end_version(2); + missing_txn.mutable_load_id()->set_hi(2); + missing_txn.mutable_load_id()->set_lo(2); + + for (const auto& rowset : rowsets) { + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, + RecycleRowsetPB::PREPARE, true), + 0); + } + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + for (const auto& rowset : rowsets) { + RecycleRowsetPB marked_rowset; + ASSERT_TRUE( + get_recycle_rowset(txn_kv.get(), tablet_id, rowset.rowset_id_v2(), &marked_rowset)); + EXPECT_TRUE(marked_rowset.rowset_meta().is_recycled()); + } + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(count_recycle_rowsets(txn_kv.get(), tablet_id), 0); + for (const auto& rowset : rowsets) { + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 1); + } + TxnInfoPB txn_info; + ASSERT_EQ(get_txn_info(txn_kv, instance_id, txn_db_id, base_txn_id, txn_info), 0); + EXPECT_EQ(txn_info.status(), TxnStatusPB::TXN_STATUS_PREPARED); +} + +TEST(RecyclerTest, recycle_prepare_rowset_abort_failure_is_retryable) { + RecyclePrepareRowsetConfigGuard config_guard(true); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_prepare_abort_retry"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t txn_db_id = 14000; + constexpr int64_t txn_id = 14001; + constexpr int64_t index_id = 14002; + constexpr int64_t tablet_id = 14003; + ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id, txn_id), 0); + + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->put(txn_info_key({instance_id, txn_db_id, txn_id}), "invalid_txn_info"); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + auto rowset = create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema, txn_id); + rowset.set_start_version(2); + rowset.set_end_version(2); + rowset.mutable_load_id()->set_hi(3); + rowset.mutable_load_id()->set_lo(3); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::PREPARE, + true), + 0); + + std::atomic delete_prefix_calls = 0; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("MockAccessor::delete_prefix", [&](auto&&) { ++delete_prefix_calls; }); + sp->enable_processing(); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(delete_prefix_calls.load(), 0); + RecycleRowsetPB retained_rowset; + ASSERT_TRUE( + get_recycle_rowset(txn_kv.get(), tablet_id, rowset.rowset_id_v2(), &retained_rowset)); + EXPECT_TRUE(retained_rowset.rowset_meta().is_recycled()); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 0); + + TxnInfoPB repaired_txn_info; + repaired_txn_info.set_txn_id(txn_id); + repaired_txn_info.set_status(TxnStatusPB::TXN_STATUS_PREPARED); + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->put(txn_info_key({instance_id, txn_db_id, txn_id}), repaired_txn_info.SerializeAsString()); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(delete_prefix_calls.load(), 1); + EXPECT_EQ(count_recycle_rowsets(txn_kv.get(), tablet_id), 0); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 1); + ASSERT_EQ(get_txn_info(txn_kv, instance_id, txn_db_id, txn_id, repaired_txn_info), 0); + EXPECT_EQ(repaired_txn_info.status(), TxnStatusPB::TXN_STATUS_ABORTED); +} + +TEST(RecyclerTest, recycle_prepare_rowset_delete_failure_is_retryable) { + RecyclePrepareRowsetConfigGuard config_guard(false); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_prepare_delete_retry"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t txn_db_id = 15000; + constexpr int64_t txn_id = 15001; + constexpr int64_t index_id = 15002; + constexpr int64_t tablet_id = 15003; + ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id, txn_id), 0); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + auto rowset = create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema, txn_id); + rowset.set_start_version(2); + rowset.set_end_version(2); + rowset.mutable_load_id()->set_hi(4); + rowset.mutable_load_id()->set_lo(4); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::PREPARE, + true), + 0); + + std::atomic delete_prefix_calls = 0; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("MockAccessor::delete_prefix", [&](auto&& args) { + ++delete_prefix_calls; + auto* ret = try_any_cast_ret(args); + ret->first = -1; + ret->second = true; + }); + sp->enable_processing(); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(delete_prefix_calls.load(), 1); + RecycleRowsetPB retained_rowset; + EXPECT_TRUE( + get_recycle_rowset(txn_kv.get(), tablet_id, rowset.rowset_id_v2(), &retained_rowset)); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 0); + TxnInfoPB txn_info; + ASSERT_EQ(get_txn_info(txn_kv, instance_id, txn_db_id, txn_id, txn_info), 0); + EXPECT_EQ(txn_info.status(), TxnStatusPB::TXN_STATUS_ABORTED); + + sp->clear_all_call_backs(); + sp->disable_processing(); + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(count_recycle_rowsets(txn_kv.get(), tablet_id), 0); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 1); +} + +TEST(RecyclerTest, recycle_prepare_rowset_commit_wins_mark_race) { + RecyclePrepareRowsetConfigGuard config_guard(true); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_prepare_commit_wins"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t txn_db_id = 16000; + constexpr int64_t txn_id = 16001; + constexpr int64_t index_id = 16002; + constexpr int64_t tablet_id = 16003; + ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id, txn_id), 0); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + auto rowset = create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema, txn_id); + rowset.set_start_version(2); + rowset.set_end_version(2); + rowset.mutable_load_id()->set_hi(5); + rowset.mutable_load_id()->set_lo(5); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::PREPARE, + true), + 0); + + auto mark_started = std::make_shared>(); + auto mark_started_future = mark_started->get_future(); + auto release_mark = std::make_shared>(); + auto release_mark_future = release_mark->get_future().share(); + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("InstanceRecycler::batch_mark_rowsets_as_recycled", [&](auto&&) { + mark_started->set_value(); + release_mark_future.wait(); + }); + sp->enable_processing(); + + auto recycle_future = + std::async(std::launch::async, [&]() { return recycler.recycle_rowsets(); }); + mark_started_future.wait(); + + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->remove(recycle_rowset_key({instance_id, tablet_id, rowset.rowset_id_v2()})); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + release_mark->set_value(); + + EXPECT_EQ(recycle_future.get(), 0); + RecycleRowsetPB removed_rowset; + EXPECT_FALSE( + get_recycle_rowset(txn_kv.get(), tablet_id, rowset.rowset_id_v2(), &removed_rowset)); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 0); + TxnInfoPB txn_info; + ASSERT_EQ(get_txn_info(txn_kv, instance_id, txn_db_id, txn_id, txn_info), 0); + EXPECT_EQ(txn_info.status(), TxnStatusPB::TXN_STATUS_PREPARED); +} + +TEST(RecyclerTest, abort_txn_for_related_rowset_terminal_status_contract) { + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "abort_txn_terminal_status"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + + constexpr int64_t txn_db_id = 16500; + constexpr int64_t txn_id = 16501; + constexpr int64_t tablet_id = 16502; + ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id, txn_id), 0); + + auto set_txn_status = [&](TxnStatusPB status) { + TxnInfoPB txn_info; + txn_info.set_txn_id(txn_id); + txn_info.set_status(status); + std::unique_ptr txn; + if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) { + return -1; + } + txn->put(txn_info_key({instance_id, txn_db_id, txn_id}), txn_info.SerializeAsString()); + return txn->commit() == TxnErrorCode::TXN_OK ? 0 : -1; + }; + + ASSERT_EQ(set_txn_status(TxnStatusPB::TXN_STATUS_ABORTED), 0); + EXPECT_EQ(recycler.abort_txn_for_related_rowset(txn_id), 0); + ASSERT_EQ(set_txn_status(TxnStatusPB::TXN_STATUS_COMMITTED), 0); + EXPECT_NE(recycler.abort_txn_for_related_rowset(txn_id), 0); + ASSERT_EQ(set_txn_status(TxnStatusPB::TXN_STATUS_VISIBLE), 0); + EXPECT_NE(recycler.abort_txn_for_related_rowset(txn_id), 0); + + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->remove(txn_info_key({instance_id, txn_db_id, txn_id})); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + EXPECT_EQ(recycler.abort_txn_for_related_rowset(txn_id), 0); + + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->remove(txn_index_key({instance_id, txn_id})); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + EXPECT_EQ(recycler.abort_txn_for_related_rowset(txn_id), 0); +} + +TEST(RecyclerTest, recycle_prepare_job_finish_wins_abort_delete_race) { + RecyclePrepareRowsetConfigGuard config_guard(false); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_prepare_job_finish_wins"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t table_id = 16600; + constexpr int64_t index_id = 16601; + constexpr int64_t partition_id = 16602; + constexpr int64_t tablet_id = 16603; + ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id), 0); + + TabletIndexPB tablet_idx; + ASSERT_EQ(get_tablet_idx(txn_kv.get(), instance_id, tablet_id, tablet_idx), 0); + TabletJobInfoPB job; + job.mutable_idx()->CopyFrom(tablet_idx); + auto* compaction = job.add_compaction(); + compaction->set_id("finished_compaction"); + compaction->set_expiration(current_time + 1000); + ASSERT_EQ(put_tablet_job(txn_kv.get(), tablet_idx, job), 0); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + auto rowset = create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema); + rowset.set_start_version(2); + rowset.set_end_version(2); + rowset.set_job_id(compaction->id()); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::PREPARE, + true), + 0); + + const auto recycle_key = recycle_rowset_key({instance_id, tablet_id, rowset.rowset_id_v2()}); + const auto job_key = job_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}); + const auto published_key = meta_rowset_key({instance_id, tablet_id, rowset.start_version()}); + std::atomic finish_ret = -1; + std::atomic delete_prefix_calls = 0; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("InstanceRecycler::batch_abort_txn_or_job_for_recycle::after_collect", + [&](auto&&) { + std::unique_ptr txn; + if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) { + finish_ret.store(1, std::memory_order_relaxed); + return; + } + txn->remove(recycle_key); + txn->remove(job_key); + txn->put(published_key, rowset.SerializeAsString()); + finish_ret.store(txn->commit() == TxnErrorCode::TXN_OK ? 0 : 2, + std::memory_order_relaxed); + }); + sp->set_call_back("MockAccessor::delete_prefix", [&](auto&&) { ++delete_prefix_calls; }); + sp->enable_processing(); + + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(finish_ret.load(std::memory_order_relaxed), 0); + EXPECT_EQ(delete_prefix_calls.load(), 0); + RecycleRowsetPB removed_rowset; + EXPECT_FALSE( + get_recycle_rowset(txn_kv.get(), tablet_id, rowset.rowset_id_v2(), &removed_rowset)); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 0); std::unique_ptr txn; ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); - std::unique_ptr it; - std::string dbm_start_key, dbm_end_key; - if (version == 2) { - dbm_start_key = versioned::meta_delete_bitmap_key({instance_id, tablet_id, ""}); - dbm_end_key = versioned::meta_delete_bitmap_key({instance_id, tablet_id + 1, ""}); - } else if (version == 1) { - dbm_start_key = meta_delete_bitmap_key({instance_id, tablet_id, "", 0, 0}); - dbm_end_key = meta_delete_bitmap_key({instance_id, tablet_id + 1, "", 0, 0}); - } - int size = 0; - while (it == nullptr /* may be not init */ || it->more()) { - ASSERT_EQ(txn->get(dbm_start_key, dbm_end_key, &it), TxnErrorCode::TXN_OK); - while (it->has_next()) { - it->next(); - size++; - } - dbm_start_key = it->next_begin_key(); - } - EXPECT_EQ(size, expected_size); + std::string published_value; + ASSERT_EQ(txn->get(published_key, &published_value), TxnErrorCode::TXN_OK); + RowsetMetaCloudPB published_rowset; + ASSERT_TRUE(published_rowset.ParseFromString(published_value)); + EXPECT_EQ(published_rowset.rowset_id_v2(), rowset.rowset_id_v2()); } -static void check_delete_bitmap_file_size(std::shared_ptr accessor, - int64_t tablet_id, int expected_size) { - int size = 0; - std::unique_ptr list_iter; - ASSERT_EQ(0, accessor->list_directory(tablet_path_prefix(tablet_id), &list_iter)); - while (list_iter->has_next()) { - auto file_info = list_iter->next(); - ASSERT_TRUE(file_info.has_value()); - std::string filename = file_info.value().path; - if (filename.ends_with("_delete_bitmap.db")) { - size++; - } - } - EXPECT_EQ(size, expected_size); -} +TEST(RecyclerTest, recycle_prepare_rowset_owner_change_skips_delete) { + RecyclePrepareRowsetConfigGuard config_guard(false); -TEST(RecyclerTest, recycle_empty) { auto txn_kv = std::make_shared(); ASSERT_EQ(txn_kv->init(), 0); - InstanceInfoPB instance; - instance.set_instance_id(instance_id); - auto obj_info = instance.add_obj_info(); - obj_info->set_id("recycle_empty"); - obj_info->set_ak(config::test_s3_ak); - obj_info->set_sk(config::test_s3_sk); - obj_info->set_endpoint(config::test_s3_endpoint); - obj_info->set_region(config::test_s3_region); - obj_info->set_bucket(config::test_s3_bucket); - obj_info->set_prefix("recycle_empty"); - + constexpr std::string_view resource_id = "recycle_prepare_owner_change"; + auto instance = create_recycler_test_instance(std::string(resource_id)); InstanceRecycler recycler(txn_kv, instance, thread_group, std::make_shared(txn_kv)); ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; + + constexpr int64_t table_id = 16700; + constexpr int64_t index_id = 16701; + constexpr int64_t partition_id = 16702; + constexpr int64_t tablet_id = 16703; + ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id), 0); + + TabletIndexPB tablet_idx; + ASSERT_EQ(get_tablet_idx(txn_kv.get(), instance_id, tablet_id, tablet_idx), 0); + TabletJobInfoPB job; + job.mutable_idx()->CopyFrom(tablet_idx); + auto* original_compaction = job.add_compaction(); + original_compaction->set_id("original_compaction"); + original_compaction->set_expiration(current_time + 1000); + ASSERT_EQ(put_tablet_job(txn_kv.get(), tablet_idx, job), 0); + + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + auto rowset = create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema); + rowset.set_start_version(2); + rowset.set_end_version(2); + rowset.set_job_id(original_compaction->id()); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::PREPARE, + true), + 0); + + RecycleRowsetPB replacement_rowset; + replacement_rowset.set_creation_time(current_time); + replacement_rowset.set_type(RecycleRowsetPB::PREPARE); + replacement_rowset.mutable_rowset_meta()->CopyFrom(rowset); + replacement_rowset.mutable_rowset_meta()->set_job_id("replacement_compaction"); + TabletJobInfoPB replacement_job; + replacement_job.mutable_idx()->CopyFrom(tablet_idx); + auto* replacement_compaction = replacement_job.add_compaction(); + replacement_compaction->set_id("replacement_compaction"); + replacement_compaction->set_expiration(current_time + 1000); + + const auto recycle_key = recycle_rowset_key({instance_id, tablet_id, rowset.rowset_id_v2()}); + const auto job_key = job_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}); + std::atomic replace_ret = -1; + std::atomic delete_prefix_calls = 0; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("InstanceRecycler::batch_abort_txn_or_job_for_recycle::after_collect", + [&](auto&&) { + std::unique_ptr txn; + if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) { + replace_ret.store(1, std::memory_order_relaxed); + return; + } + txn->put(recycle_key, replacement_rowset.SerializeAsString()); + txn->put(job_key, replacement_job.SerializeAsString()); + replace_ret.store(txn->commit() == TxnErrorCode::TXN_OK ? 0 : 2, + std::memory_order_relaxed); + }); + sp->set_call_back("MockAccessor::delete_prefix", [&](auto&&) { ++delete_prefix_calls; }); + sp->enable_processing(); ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(replace_ret.load(std::memory_order_relaxed), 0); + EXPECT_EQ(delete_prefix_calls.load(), 0); + RecycleRowsetPB retained_rowset; + ASSERT_TRUE( + get_recycle_rowset(txn_kv.get(), tablet_id, rowset.rowset_id_v2(), &retained_rowset)); + EXPECT_EQ(retained_rowset.rowset_meta().job_id(), "replacement_compaction"); + EXPECT_EQ(accessor->exists(segment_path(tablet_id, rowset.rowset_id_v2(), 0)), 0); } -TEST(RecyclerTest, recycle_rowsets) { - config::retention_seconds = 0; +TEST(RecyclerTest, recycle_prepare_rowsets_cross_abort_batch_boundary) { + RecyclePrepareRowsetConfigGuard config_guard(false); + auto txn_kv = std::make_shared(); ASSERT_EQ(txn_kv->init(), 0); - InstanceInfoPB instance; - instance.set_instance_id(instance_id); - auto obj_info = instance.add_obj_info(); - obj_info->set_id("recycle_rowsets"); - obj_info->set_ak(config::test_s3_ak); - obj_info->set_sk(config::test_s3_sk); - obj_info->set_endpoint(config::test_s3_endpoint); - obj_info->set_region(config::test_s3_region); - obj_info->set_bucket(config::test_s3_bucket); - obj_info->set_prefix("recycle_rowsets"); + constexpr std::string_view resource_id = "recycle_prepare_abort_batch"; + auto instance = create_recycler_test_instance(std::string(resource_id)); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; - config::instance_recycler_worker_pool_size = 1; + constexpr int64_t txn_db_id = 17000; + constexpr int64_t txn_id = 17001; + constexpr int64_t index_id = 17002; + constexpr int64_t tablet_id = 17003; + constexpr int rowset_count = 257; + ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id, txn_id), 0); - int insert_no_inverted_index = 0; - int insert_inverted_index = 0; + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + for (int i = 0; i < rowset_count; ++i) { + auto rowset = + create_rowset(std::string(resource_id), tablet_id, index_id, 1, schema, txn_id); + rowset.set_start_version(i + 2); + rowset.set_end_version(i + 2); + rowset.mutable_load_id()->set_hi(6); + rowset.mutable_load_id()->set_lo(i); + ASSERT_EQ(create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, + RecycleRowsetPB::PREPARE, true), + 0); + } + + std::atomic delete_prefix_calls = 0; + std::atomic recheck_batch_count = 0; + std::atomic rechecked_key_count = 0; + std::atomic all_deletes_saw_aborted = true; auto sp = SyncPoint::get_instance(); DORIS_CLOUD_DEFER { - SyncPoint::get_instance()->clear_all_call_backs(); + sp->clear_all_call_backs(); + sp->disable_processing(); }; - sp->set_call_back("InvertedIndexIdCache::insert1", [&](auto&&) { ++insert_no_inverted_index; }); - sp->set_call_back("InvertedIndexIdCache::insert2", [&](auto&&) { ++insert_inverted_index; }); + sp->set_call_back("MockAccessor::delete_prefix", [&](auto&&) { + ++delete_prefix_calls; + TxnInfoPB txn_info; + if (get_txn_info(txn_kv, instance_id, txn_db_id, txn_id, txn_info) != 0 || + txn_info.status() != TxnStatusPB::TXN_STATUS_ABORTED) { + all_deletes_saw_aborted.store(false, std::memory_order_relaxed); + } + }); + sp->set_call_back("InstanceRecycler::batch_recheck_rowsets_after_abort", [&](auto&& args) { + auto* batch_size = try_any_cast(args[0]); + recheck_batch_count.fetch_add(1, std::memory_order_relaxed); + rechecked_key_count.fetch_add(*batch_size, std::memory_order_relaxed); + }); sp->enable_processing(); + ASSERT_EQ(recycler.recycle_rowsets(), 0); + EXPECT_EQ(delete_prefix_calls.load(), rowset_count); + EXPECT_EQ(recheck_batch_count.load(), 2); + EXPECT_EQ(rechecked_key_count.load(), rowset_count); + EXPECT_TRUE(all_deletes_saw_aborted.load(std::memory_order_relaxed)); + EXPECT_EQ(count_recycle_rowsets(txn_kv.get(), tablet_id), 0); + std::unique_ptr list_iter; + ASSERT_EQ(accessor->list_directory(tablet_path_prefix(tablet_id), &list_iter), 0); + EXPECT_FALSE(list_iter->has_next()); +} + +TEST(RecyclerTest, recycle_tmp_rowsets_cross_abort_recheck_batch_boundary) { + RecyclePrepareRowsetConfigGuard config_guard(false); + + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + constexpr std::string_view resource_id = "recycle_tmp_abort_recheck_batch"; + auto instance = create_recycler_test_instance(std::string(resource_id)); InstanceRecycler recycler(txn_kv, instance, thread_group, std::make_shared(txn_kv)); ASSERT_EQ(recycler.init(), 0); + auto accessor = recycler.accessor_map_.begin()->second; - std::vector schemas; - for (int i = 0; i < 5; ++i) { - auto& schema = schemas.emplace_back(); - schema.set_schema_version(i); - schema.set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V1); - for (int j = 0; j < i; ++j) { - auto index = schema.add_index(); - index->set_index_id(j); - index->set_index_type(IndexType::INVERTED); - } - } + constexpr int64_t txn_db_id = 17100; + constexpr int64_t txn_id = 17101; + constexpr int64_t index_id = 17102; + constexpr int64_t tablet_id_base = 17103; + constexpr int rowset_count = 257; + ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id_base, txn_id), 0); - auto accessor = recycler.accessor_map_.begin()->second; - constexpr int index_id = 10001, tablet_id = 10002; - for (int i = 0; i < 1000; ++i) { - auto rowset = create_rowset("recycle_rowsets", tablet_id, index_id, 5, schemas[i % 5]); - create_recycle_rowset( - txn_kv.get(), accessor.get(), rowset, - static_cast(i % (RecycleRowsetPB::Type_MAX + 1)), i & 1, - i < 500); - } - for (int i = 0; i < 1000; ++i) { - auto rowset = create_rowset("recycle_rowsets", tablet_id, index_id, 5, schemas[i % 5]); - create_recycle_rowset(txn_kv.get(), accessor.get(), rowset, RecycleRowsetPB::COMPACT, true, - i < 500); + doris::TabletSchemaCloudPB schema; + schema.set_schema_version(1); + for (int i = 0; i < rowset_count; ++i) { + auto rowset = create_rowset(std::string(resource_id), tablet_id_base + i, index_id, 1, + schema, txn_id); + rowset.mutable_load_id()->set_hi(7); + rowset.mutable_load_id()->set_lo(i); + ASSERT_EQ(create_tmp_rowset(txn_kv.get(), accessor.get(), rowset, false), 0); } - check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 1000); - check_delete_bitmap_file_size(accessor, tablet_id, 1000); - for (size_t i = 0; i < 10; i++) { - ASSERT_EQ(recycler.recycle_rowsets(), 0); - } + std::atomic recheck_batch_count = 0; + std::atomic rechecked_key_count = 0; + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + sp->clear_all_call_backs(); + sp->disable_processing(); + }; + sp->set_call_back("InstanceRecycler::batch_recheck_rowsets_after_abort", [&](auto&& args) { + auto* batch_size = try_any_cast(args[0]); + recheck_batch_count.fetch_add(1, std::memory_order_relaxed); + rechecked_key_count.fetch_add(*batch_size, std::memory_order_relaxed); + }); + sp->enable_processing(); + + ASSERT_EQ(recycler.recycle_tmp_rowsets(), 0); + EXPECT_EQ(recheck_batch_count.load(), 2); + EXPECT_EQ(rechecked_key_count.load(), rowset_count); - // check rowset does not exist on obj store - std::unique_ptr list_iter; - ASSERT_EQ(0, accessor->list_directory(tablet_path_prefix(tablet_id), &list_iter)); - for (auto file = list_iter->next(); file.has_value(); file = list_iter->next()) { - LOG(INFO) << "file: " << file->path; - } - EXPECT_FALSE(list_iter->has_next()); - // check all recycle rowset kv have been deleted std::unique_ptr txn; ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); std::unique_ptr it; - auto begin_key = recycle_key_prefix(instance_id); - auto end_key = recycle_key_prefix(instance_id + '\xff'); + auto begin_key = meta_rowset_tmp_key({instance_id, 0, 0}); + auto end_key = meta_rowset_tmp_key({instance_id, INT64_MAX, 0}); ASSERT_EQ(txn->get(begin_key, end_key, &it), TxnErrorCode::TXN_OK); EXPECT_EQ(it->size(), 0); - // Check InvertedIndexIdCache - EXPECT_EQ(insert_inverted_index, 4); - EXPECT_EQ(insert_no_inverted_index, 1); - // check all versioned delete bitmap kv have been deleted - check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 0); - check_delete_bitmap_file_size(accessor, tablet_id, 0); -} - -TEST(RecyclerTest, next_recycle_rowset_tablet_key_overwrites_existing_buffer) { - std::string next_key = recycle_rowset_key({instance_id, 10002, "rowset"}); - ASSERT_EQ(InstanceRecycler::next_recycle_rowset_tablet_key(instance_id, 10002, &next_key), 0); - std::string_view k1 = next_key; - k1.remove_prefix(1); - std::vector, int, int>> out; - ASSERT_EQ(decode_key(&k1, &out), 0); - EXPECT_EQ(std::get(std::get<0>(out[3])), 10003); - EXPECT_TRUE(std::get(std::get<0>(out[4])).empty()); + std::unique_ptr list_iter; + ASSERT_EQ(accessor->list_directory("data/", &list_iter), 0); + EXPECT_FALSE(list_iter->has_next()); } TEST(RecyclerTest, recycle_rowsets_tablet_batch_limit_recycles_remaining_in_next_round) { @@ -1649,11 +2655,11 @@ TEST(RecyclerTest, recycle_rowsets_limit_per_tablet_batch) { }; ASSERT_EQ(recycler.recycle_rowsets(), 0); + ASSERT_EQ(recycler.recycle_rowsets(), 0); EXPECT_EQ(count_recycle_rowsets(tablet_id0), 1); EXPECT_EQ(count_recycle_rowsets(tablet_id1), 1); - ASSERT_EQ(recycler.recycle_rowsets(), 0); ASSERT_EQ(recycler.recycle_rowsets(), 0); EXPECT_EQ(count_recycle_rowsets(tablet_id0), 0); EXPECT_EQ(count_recycle_rowsets(tablet_id1), 0); @@ -9176,13 +10182,185 @@ TEST(RecyclerTest, abort_job_for_related_rowset_when_tablet_recycled) { rowset.set_job_id("test_job_id"); // Call abort_job_for_related_rowset with a rowset whose tablet has been recycled - int ret = recycler.abort_job_for_related_rowset(rowset); + int ret = recycler.abort_job_for_related_rowset(rowset.tablet_id(), rowset.rowset_id_v2(), + rowset.job_id()); // Should return 0 (success) because tablet is already recycled ASSERT_EQ(ret, 0) << "Should return 0 when tablet is already recycled (parallel recycle scenario)"; } +TEST(RecyclerTest, abort_exact_job_and_allow_expired_job_for_related_rowset) { + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + const std::string test_instance_id = "instance_id_recycle_test"; + InstanceInfoPB instance; + instance.set_instance_id(test_instance_id); + auto obj_info = instance.add_obj_info(); + obj_info->set_id("abort_exact_job_and_allow_expired_job_for_related_rowset"); + obj_info->set_ak(config::test_s3_ak); + obj_info->set_sk(config::test_s3_sk); + obj_info->set_endpoint(config::test_s3_endpoint); + obj_info->set_region(config::test_s3_region); + obj_info->set_bucket(config::test_s3_bucket); + obj_info->set_prefix("abort_exact_job_and_allow_expired_job_for_related_rowset"); + + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + + constexpr int64_t table_id = 22000; + constexpr int64_t index_id = 22001; + constexpr int64_t partition_id = 22002; + constexpr int64_t tablet_id = 22003; + ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id), 0); + + TabletIndexPB tablet_idx; + ASSERT_EQ(get_tablet_idx(txn_kv.get(), test_instance_id, tablet_id, tablet_idx), 0); + + TabletJobInfoPB job; + job.mutable_idx()->CopyFrom(tablet_idx); + auto* first_compaction = job.add_compaction(); + first_compaction->set_id("job_a"); + first_compaction->set_expiration(current_time + 1000); + auto* target_compaction = job.add_compaction(); + target_compaction->set_id("job_b"); + target_compaction->set_expiration(current_time + 1000); + + auto job_key = job_tablet_key({test_instance_id, tablet_idx.table_id(), tablet_idx.index_id(), + tablet_idx.partition_id(), tablet_idx.tablet_id()}); + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->put(job_key, job.SerializeAsString()); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + RowsetMetaCloudPB rowset_meta; + rowset_meta.set_tablet_id(tablet_id); + rowset_meta.set_job_id("job_b"); + rowset_meta.set_rowset_id_v2("rowset_b"); + ASSERT_EQ(recycler.abort_job_for_related_rowset( + rowset_meta.tablet_id(), rowset_meta.rowset_id_v2(), rowset_meta.job_id()), + 0); + + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + std::string job_val; + ASSERT_EQ(txn->get(job_key, &job_val), TxnErrorCode::TXN_OK); + TabletJobInfoPB remaining_job; + ASSERT_TRUE(remaining_job.ParseFromString(job_val)); + ASSERT_EQ(remaining_job.compaction_size(), 1); + EXPECT_EQ(remaining_job.compaction(0).id(), "job_a"); + + job.Clear(); + first_compaction = job.add_compaction(); + first_compaction->set_id("job_a"); + first_compaction->set_expiration(current_time + 1000); + target_compaction = job.add_compaction(); + target_compaction->set_id("job_b"); + target_compaction->set_expiration(current_time - 1000); + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->put(job_key, job.SerializeAsString()); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + ASSERT_EQ(recycler.abort_job_for_related_rowset( + rowset_meta.tablet_id(), rowset_meta.rowset_id_v2(), rowset_meta.job_id()), + 0); + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + ASSERT_EQ(txn->get(job_key, &job_val), TxnErrorCode::TXN_OK); + ASSERT_TRUE(remaining_job.ParseFromString(job_val)); + ASSERT_EQ(remaining_job.compaction_size(), 1); + EXPECT_EQ(remaining_job.compaction(0).id(), "job_a"); +} + +TEST(RecyclerTest, abort_schema_change_from_new_tablet_rowset) { + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + + const std::string test_instance_id = "instance_id_recycle_test"; + InstanceInfoPB instance; + instance.set_instance_id(test_instance_id); + auto* obj_info = instance.add_obj_info(); + obj_info->set_id("abort_schema_change_from_new_tablet_rowset"); + obj_info->set_ak(config::test_s3_ak); + obj_info->set_sk(config::test_s3_sk); + obj_info->set_endpoint(config::test_s3_endpoint); + obj_info->set_region(config::test_s3_region); + obj_info->set_bucket(config::test_s3_bucket); + obj_info->set_prefix("abort_schema_change_from_new_tablet_rowset"); + + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + + constexpr int64_t table_id = 23000; + constexpr int64_t base_index_id = 23001; + constexpr int64_t new_index_id = 23002; + constexpr int64_t partition_id = 23003; + constexpr int64_t base_tablet_id = 23004; + constexpr int64_t new_tablet_id = 23005; + + TabletIndexPB base_tablet_idx; + base_tablet_idx.set_table_id(table_id); + base_tablet_idx.set_index_id(base_index_id); + base_tablet_idx.set_partition_id(partition_id); + base_tablet_idx.set_tablet_id(base_tablet_id); + + TabletIndexPB new_tablet_idx; + new_tablet_idx.set_table_id(table_id); + new_tablet_idx.set_index_id(new_index_id); + new_tablet_idx.set_partition_id(partition_id); + new_tablet_idx.set_tablet_id(new_tablet_id); + + TabletJobInfoPB job; + job.mutable_idx()->CopyFrom(base_tablet_idx); + auto* schema_change = job.mutable_schema_change(); + schema_change->set_id("schema_change_job"); + schema_change->set_initiator("BE1"); + schema_change->set_expiration(current_time - 1000); + schema_change->mutable_new_tablet_idx()->CopyFrom(new_tablet_idx); + + auto base_job_key = job_tablet_key( + {test_instance_id, table_id, base_index_id, partition_id, base_tablet_id}); + auto new_job_key = + job_tablet_key({test_instance_id, table_id, new_index_id, partition_id, new_tablet_id}); + + doris::TabletMetaCloudPB new_tablet_meta; + new_tablet_meta.set_tablet_id(new_tablet_id); + new_tablet_meta.set_tablet_state(doris::TabletStatePB::PB_NOTREADY); + + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + txn->put(meta_tablet_idx_key({test_instance_id, new_tablet_id}), + new_tablet_idx.SerializeAsString()); + txn->put(meta_tablet_key( + {test_instance_id, table_id, new_index_id, partition_id, new_tablet_id}), + new_tablet_meta.SerializeAsString()); + txn->put(base_job_key, job.SerializeAsString()); + txn->put(new_job_key, job.SerializeAsString()); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); + + RowsetMetaCloudPB rowset_meta; + rowset_meta.set_tablet_id(new_tablet_id); + rowset_meta.set_job_id(schema_change->id()); + rowset_meta.set_rowset_id_v2("new_tablet_output_rowset"); + ASSERT_EQ(recycler.abort_job_for_related_rowset( + rowset_meta.tablet_id(), rowset_meta.rowset_id_v2(), rowset_meta.job_id()), + 0); + + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + std::string base_job_val; + ASSERT_EQ(txn->get(base_job_key, &base_job_val), TxnErrorCode::TXN_OK); + TabletJobInfoPB base_job; + ASSERT_TRUE(base_job.ParseFromString(base_job_val)); + EXPECT_FALSE(base_job.has_schema_change()); + + std::string new_job_val; + ASSERT_EQ(txn->get(new_job_key, &new_job_val), TxnErrorCode::TXN_OK); + TabletJobInfoPB new_job; + ASSERT_TRUE(new_job.ParseFromString(new_job_val)); + EXPECT_FALSE(new_job.has_schema_change()); +} + TEST(RecyclerTest, recycle_tablet_with_delete_file_failure) { // If object deletion fails, recycle_tablets should report failure and keep the tablet KV.