Method LoadSSTableReader

storage/sstable/reader/merged_sstable_reader.cpp:233–291

}

bool MergedSSTableReader::LoadSSTableReader(const std::string &path, ReadMode type) {
    SSTableReader *sstable = SSTableReader::Open(path, type);
    if (sstable == NULL) {
        LOG(ERROR)<< "Failed to open sstable:" << path;
        return false;
    }
    if (sstable->EntryCount() == 0) {
        LOG(WARNING)<< "sstable " << path << " is empty.";
        delete sstable;
        return false;
    }

    // now fill it into proper sstable set.
    std::string set_id = sstable->GetMetaData(kSSTableSetID);
    if (set_id.empty()) {
        LOG(WARNING)<< "sstable with empty set id";
    }
    std::string policy = sstable->GetMetaData(kShardPolicy);
    std::string str_num_shard = sstable->GetMetaData(kShardTotalNum);
    std::string str_shard_id = sstable->GetMetaData(kShardID);
    VLOG(2) << "set_id: " << set_id << ", policy: " << policy << ", num_shard: "
            << str_num_shard << ", shard_id: " << str_shard_id;

    int num_shard = 0;
    if (!StringToNumber(str_num_shard, &num_shard)) {
        if (!set_id.empty()) {
            LOG(WARNING)<< "bad num shard: " << str_num_shard << ", path: "
            << sstable->GetPath();
            delete sstable;
            return false;
        }
    }
    int shard_id = -1;
    if (!StringToNumber(str_shard_id, &shard_id)) {
        if (!set_id.empty()) {
            LOG(WARNING)<< "bad shard id: " << str_shard_id;
            delete sstable;
            return false;
        }
    }

    std::map<std::string, SSTableReaderSet*>::iterator it = impl_->sets_.find(set_id);
    if (it == impl_->sets_.end()) {
        if (set_id.empty()) {
            // Ignore sharding policy for sstable files that has no set id
            if (FLAGS_ignore_sstable_setid) {
                impl_->sets_[""] = new SSTableReaderSet("", policy, num_shard);
            } else {
                impl_->sets_[""] = new SSTableReaderSet("", "", 0);
            }
        } else {
            impl_->sets_[set_id] = new SSTableReaderSet(set_id, policy, num_shard);
        }
        it = impl_->sets_.find(set_id);
    }

    impl_->tables_.push_back(sstable);
    return it->second->AddSSTableReader(sstable, set_id, policy, num_shard, shard_id);