| 607 | } |
| 608 | |
| 609 | rocksdb::Status init(rocksdb::Options options) { |
| 610 | // Open instance. |
| 611 | TraceEvent(SevInfo, "ShardedRocksShardManagerInitBegin", this->logId).detail("DataPath", path); |
| 612 | std::vector<std::string> columnFamilies; |
| 613 | rocksdb::Status status = rocksdb::DB::ListColumnFamilies(options, path, &columnFamilies); |
| 614 | |
| 615 | rocksdb::ColumnFamilyOptions cfOptions = getCFOptions(); |
| 616 | std::vector<rocksdb::ColumnFamilyDescriptor> descriptors; |
| 617 | bool foundMetadata = false; |
| 618 | for (const auto& name : columnFamilies) { |
| 619 | if (name == "kvs-metadata") { |
| 620 | foundMetadata = true; |
| 621 | } |
| 622 | descriptors.push_back(rocksdb::ColumnFamilyDescriptor{ name, cfOptions }); |
| 623 | } |
| 624 | |
| 625 | ASSERT(foundMetadata || descriptors.size() == 0); |
| 626 | |
| 627 | // Add default column family if it's a newly opened database. |
| 628 | if (descriptors.size() == 0) { |
| 629 | descriptors.push_back(rocksdb::ColumnFamilyDescriptor{ "default", cfOptions }); |
| 630 | } |
| 631 | |
| 632 | std::vector<rocksdb::ColumnFamilyHandle*> handles; |
| 633 | status = rocksdb::DB::Open(options, path, descriptors, &handles, &db); |
| 634 | if (!status.ok()) { |
| 635 | logRocksDBError(status, "Open"); |
| 636 | return status; |
| 637 | } |
| 638 | |
| 639 | if (foundMetadata) { |
| 640 | TraceEvent(SevInfo, "ShardedRocksInitLoadPhysicalShards", this->logId) |
| 641 | .detail("PhysicalShardCount", handles.size()); |
| 642 | for (auto handle : handles) { |
| 643 | if (handle->GetName() == "kvs-metadata") { |
| 644 | metadataShard = std::make_shared<PhysicalShard>(db, "kvs-metadata", handle); |
| 645 | } else { |
| 646 | physicalShards[handle->GetName()] = std::make_shared<PhysicalShard>(db, handle->GetName(), handle); |
| 647 | } |
| 648 | columnFamilyMap[handle->GetID()] = handle; |
| 649 | TraceEvent(SevVerbose, "ShardedRocksInitPhysicalShard", this->logId) |
| 650 | .detail("PhysicalShard", handle->GetName()); |
| 651 | } |
| 652 | RangeResult metadata; |
| 653 | readRangeInDb(metadataShard.get(), prefixRange(shardMappingPrefix), UINT16_MAX, UINT16_MAX, &metadata); |
| 654 | |
| 655 | std::vector<std::pair<KeyRange, std::string>> mapping = decodeShardMapping(metadata, shardMappingPrefix); |
| 656 | |
| 657 | for (const auto& [range, name] : mapping) { |
| 658 | TraceEvent(SevVerbose, "ShardedRocksLoadRange", this->logId) |
| 659 | .detail("Range", range) |
| 660 | .detail("PhysicalShard", name); |
| 661 | auto it = physicalShards.find(name); |
| 662 | // Raise error if physical shard is missing. |
| 663 | if (it == physicalShards.end()) { |
| 664 | TraceEvent(SevError, "ShardedRocksDB").detail("MissingShard", name); |
| 665 | return rocksdb::Status::NotFound(); |
| 666 | } |
no test coverage detected