github-actions[bot] commented on code in PR #69492:
URL: https://github.com/apache/doris/pull/69492#discussion_r4375817896
##########
cloud/src/recycler/checker.cpp:
##########
@@ -519,16 +531,156 @@ int key_exist(TxnKv* txn_kv, std::string_view key) {
}
}
-InstanceChecker::InstanceChecker(std::shared_ptr<TxnKv> txn_kv, const
std::string& instance_id)
- : txn_kv_(txn_kv), instance_id_(instance_id) {
+SharedStorageIndex::SharedStorageIndex(std::shared_ptr<TxnKv> txn_kv,
+
std::optional<std::vector<InstanceInfoPB>> instances)
+ : txn_kv_(std::move(txn_kv)),
preloaded_instances_(std::move(instances)) {}
+
+std::optional<SharedStorageIndex::StorageLocation>
SharedStorageIndex::s3_location(
+ const ObjectStoreInfoPB& obj_info) {
+ // Missing connection info must not make unrelated resources match.
+ if (obj_info.endpoint().empty() || obj_info.bucket().empty()) {
+ return std::nullopt;
+ }
+ return StorageLocation {"s3", obj_info.endpoint(), obj_info.bucket(),
obj_info.prefix()};
+}
+
+std::optional<SharedStorageIndex::StorageLocation>
SharedStorageIndex::hdfs_location(
+ const HdfsVaultInfo& hdfs_info) {
+ if (hdfs_info.build_conf().fs_name().empty()) {
+ return std::nullopt;
+ }
+ return StorageLocation {"hdfs", hdfs_info.build_conf().fs_name(), "",
hdfs_info.prefix()};
+}
+
+void SharedStorageIndex::ensure_loaded() {
+ std::call_once(load_once_, [this] {
+ load_status_ = load_all_instances();
+ if (load_status_ != 0) {
+ // Never expose a partially built index.
+ instance_resource_locations_.clear();
+ location_to_owners_.clear();
+ }
+ });
+}
+
+int SharedStorageIndex::load_all_instances() {
+ std::vector<InstanceInfoPB> instances;
+ if (preloaded_instances_.has_value()) {
+ instances = std::move(*preloaded_instances_);
+ preloaded_instances_.reset();
+ } else if (get_all_instances(txn_kv_.get(), instances) != 0) {
+ return -1;
+ }
+ // Load all instances, even those not scheduled for checking: decoupled
+ // instances can still share storage with the checked one.
+ for (const auto& instance : instances) {
+ if (load_instance(instance) != 0) {
+ return -1;
+ }
+ }
+ return 0;
+}
+
+int SharedStorageIndex::load_instance(const InstanceInfoPB& instance) {
+ const std::string& instance_id = instance.instance_id();
+ // Register the instance even if it has no resources, so that it can be
told
+ // apart from an unknown instance.
+ instance_resource_locations_[instance_id];
+ // Locations of the resources in `resource_ids` are filled by the vault
scan below.
+ for (const auto& resource_id : instance.resource_ids()) {
+ add_resource(instance_id, resource_id, std::nullopt);
+ }
+ for (const auto& obj_info : instance.obj_info()) {
+ add_resource(instance_id, obj_info.id(), s3_location(obj_info));
+ }
+ if (instance.resource_ids().empty()) {
+ return 0;
+ }
+ return load_storage_vaults(instance_id);
+}
+
+int SharedStorageIndex::load_storage_vaults(const std::string& instance_id) {
+ std::unique_ptr<Transaction> txn;
+ if (txn_kv_->create_txn(&txn) != TxnErrorCode::TXN_OK) {
+ return -1;
+ }
+ auto iter = txn->full_range_get(storage_vault_key({instance_id, ""}),
+ storage_vault_key({instance_id, "\xff"}));
+ for (auto entry = iter->next(); entry.has_value(); entry = iter->next()) {
+ StorageVaultPB vault;
+ if (!vault.ParseFromArray(entry->second.data(), entry->second.size()))
{
+ LOG_WARNING("malformed storage vault").tag("instance_id",
instance_id);
+ return -1;
+ }
+ std::optional<StorageLocation> location;
+ if (vault.has_obj_info()) {
+ location = s3_location(vault.obj_info());
+ } else if (vault.has_hdfs_info()) {
+ location = hdfs_location(vault.hdfs_info());
+ }
+ add_resource(instance_id, vault.id(), std::move(location));
+ }
+ return iter->is_valid() ? 0 : -1;
+}
+
+void SharedStorageIndex::add_resource(const std::string& instance_id,
+ const std::string& resource_id,
+ std::optional<StorageLocation> location)
{
+ auto& resource_location =
instance_resource_locations_[instance_id][resource_id];
+ if (!location.has_value()) {
+ return;
+ }
+ if (!resource_location.has_value()) {
+ resource_location = location;
+ }
+ location_to_owners_[std::move(*location)].push_back({instance_id,
resource_id});
+}
+
+int SharedStorageIndex::get_shared_resource_owners(std::string_view
instance_id,
+ SharedResourceOwners*
owners) {
+ ensure_loaded();
+ owners->clear();
+ if (load_status_ != 0) {
+ return -1;
+ }
+ auto instance =
instance_resource_locations_.find(std::string(instance_id));
+ if (instance == instance_resource_locations_.end()) {
+ LOG_WARNING("instance not found while checking shared resources")
+ .tag("instance_id", instance_id);
+ return -1;
+ }
+ for (const auto& [resource_id, location] : instance->second) {
+ if (location.has_value()) {
+ (*owners)[resource_id] = location_to_owners_.at(*location);
+ } else {
+ // Without a known location, the resource is only owned by its own
instance.
+ (*owners)[resource_id] = {{instance->first, resource_id}};
+ }
+ }
+ return 0;
+}
+
+InstanceChecker::InstanceChecker(std::shared_ptr<TxnKv> txn_kv, const
std::string& instance_id,
+ std::shared_ptr<ResourceManager> resource_mgr,
+ std::shared_ptr<SharedStorageIndex>
shared_storage_index)
+ : txn_kv_(txn_kv),
+ instance_id_(instance_id),
+ resource_mgr_(std::move(resource_mgr)),
+ shared_storage_index_(std::move(shared_storage_index)) {
snapshot_manager_ = create_snapshot_manager(txn_kv);
- resource_mgr_ = std::make_shared<ResourceManager>(std::move(txn_kv));
- resource_mgr_->init();
+ if (!shared_storage_index_) {
+ shared_storage_index_ = std::make_shared<SharedStorageIndex>(txn_kv);
+ }
+ if (!resource_mgr_) {
+ resource_mgr_ = std::make_shared<ResourceManager>(std::move(txn_kv));
+ resource_mgr_->init();
+ }
}
int InstanceChecker::init(const InstanceInfoPB& instance) {
table_stream_versioned_write_ = instance.multi_version_status() ==
MULTI_VERSION_WRITE_ONLY ||
instance.multi_version_status() ==
MULTI_VERSION_READ_WRITE;
+ resource_mgr_->refresh_instance(instance_id_, instance);
Review Comment:
[P2] Populate intermediate ancestor links before clone-chain lookup
The new fallback still fails for supported multilevel clones because this
refresh registers only the checked instance. For `C -> B -> A`, let C commit a
nonempty rowset for an inherited tablet whose versioned tablet index remains in
A. After a checker restart, `CloneChainReader::get_tablet_index()` searches C
and B, then returns `TXN_KEY_NOT_FOUND` because B's parent link is absent from
this manager. This persists if B is filtered out or a DELETED tombstone: both
scheduling and inspection skip it, so healthy C checks keep failing. Populate
the full lineage before filtering/scheduling, or fetch missing ancestors on
demand, including the HTTP check path. Calling `ResourceManager::init()` alone
is insufficient because it does not populate source-snapshot links. Add a
checker case with C's local rowset, the index only in A, and an unscheduled B.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
