ceph/src/rocksdb/db/compacted_db_impl.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under the BSD-style license found in the
   3 //  LICENSE file in the root directory of this source tree. An additional grant
   4 //  of patent rights can be found in the PATENTS file in the same directory.
   5
   6 #ifndef ROCKSDB_LITE
   7 #include "db/compacted_db_impl.h"
   8 #include "db/db_impl.h"
   9 #include "db/version_set.h"
  10 #include "table/get_context.h"
  11
  12 namespace rocksdb {
  13
  14 extern void MarkKeyMayExist(void* arg);
  15 extern bool SaveValue(void* arg, const ParsedInternalKey& parsed_key,
  16                       const Slice& v, bool hit_and_return);
  17
  18 CompactedDBImpl::CompactedDBImpl(
  19   const DBOptions& options, const std::string& dbname)
  20   : DBImpl(options, dbname) {
  21 }
  22
  23 CompactedDBImpl::~CompactedDBImpl() {
  24 }
  25
  26 size_t CompactedDBImpl::FindFile(const Slice& key) {
  27   size_t left = 0;
  28   size_t right = files_.num_files - 1;
  29   while (left < right) {
  30     size_t mid = (left + right) >> 1;
  31     const FdWithKeyRange& f = files_.files[mid];
  32     if (user_comparator_->Compare(ExtractUserKey(f.largest_key), key) < 0) {
  33       // Key at "mid.largest" is < "target".  Therefore all
  34       // files at or before "mid" are uninteresting.
  35       left = mid + 1;
  36     } else {
  37       // Key at "mid.largest" is >= "target".  Therefore all files
  38       // after "mid" are uninteresting.
  39       right = mid;
  40     }
  41   }
  42   return right;
  43 }
  44
  45 Status CompactedDBImpl::Get(const ReadOptions& options, ColumnFamilyHandle*,
  46                             const Slice& key, PinnableSlice* value) {
  47   GetContext get_context(user_comparator_, nullptr, nullptr, nullptr,
  48                          GetContext::kNotFound, key, value, nullptr, nullptr,
  49                          nullptr, nullptr);
  50   LookupKey lkey(key, kMaxSequenceNumber);
  51   files_.files[FindFile(key)].fd.table_reader->Get(
  52       options, lkey.internal_key(), &get_context);
  53   if (get_context.State() == GetContext::kFound) {
  54     return Status::OK();
  55   }
  56   return Status::NotFound();
  57 }
  58
  59 std::vector<Status> CompactedDBImpl::MultiGet(const ReadOptions& options,
  60     const std::vector<ColumnFamilyHandle*>&,
  61     const std::vector<Slice>& keys, std::vector<std::string>* values) {
  62   autovector<TableReader*, 16> reader_list;
  63   for (const auto& key : keys) {
  64     const FdWithKeyRange& f = files_.files[FindFile(key)];
  65     if (user_comparator_->Compare(key, ExtractUserKey(f.smallest_key)) < 0) {
  66       reader_list.push_back(nullptr);
  67     } else {
  68       LookupKey lkey(key, kMaxSequenceNumber);
  69       f.fd.table_reader->Prepare(lkey.internal_key());
  70       reader_list.push_back(f.fd.table_reader);
  71     }
  72   }
  73   std::vector<Status> statuses(keys.size(), Status::NotFound());
  74   values->resize(keys.size());
  75   int idx = 0;
  76   for (auto* r : reader_list) {
  77     if (r != nullptr) {
  78       PinnableSlice pinnable_val;
  79       std::string& value = (*values)[idx];
  80       GetContext get_context(user_comparator_, nullptr, nullptr, nullptr,
  81                              GetContext::kNotFound, keys[idx], &pinnable_val,
  82                              nullptr, nullptr, nullptr, nullptr);
  83       LookupKey lkey(keys[idx], kMaxSequenceNumber);
  84       r->Get(options, lkey.internal_key(), &get_context);
  85       value.assign(pinnable_val.data(), pinnable_val.size());
  86       if (get_context.State() == GetContext::kFound) {
  87         statuses[idx] = Status::OK();
  88       }
  89     }
  90     ++idx;
  91   }
  92   return statuses;
  93 }
  94
  95 Status CompactedDBImpl::Init(const Options& options) {
  96   mutex_.Lock();
  97   ColumnFamilyDescriptor cf(kDefaultColumnFamilyName,
  98                             ColumnFamilyOptions(options));
  99   Status s = Recover({cf}, true /* read only */, false, true);
 100   if (s.ok()) {
 101     cfd_ = reinterpret_cast<ColumnFamilyHandleImpl*>(
 102               DefaultColumnFamily())->cfd();
 103     delete cfd_->InstallSuperVersion(new SuperVersion(), &mutex_);
 104   }
 105   mutex_.Unlock();
 106   if (!s.ok()) {
 107     return s;
 108   }
 109   NewThreadStatusCfInfo(cfd_);
 110   version_ = cfd_->GetSuperVersion()->current;
 111   user_comparator_ = cfd_->user_comparator();
 112   auto* vstorage = version_->storage_info();
 113   if (vstorage->num_non_empty_levels() == 0) {
 114     return Status::NotSupported("no file exists");
 115   }
 116   const LevelFilesBrief& l0 = vstorage->LevelFilesBrief(0);
 117   // L0 should not have files
 118   if (l0.num_files > 1) {
 119     return Status::NotSupported("L0 contain more than 1 file");
 120   }
 121   if (l0.num_files == 1) {
 122     if (vstorage->num_non_empty_levels() > 1) {
 123       return Status::NotSupported("Both L0 and other level contain files");
 124     }
 125     files_ = l0;
 126     return Status::OK();
 127   }
 128
 129   for (int i = 1; i < vstorage->num_non_empty_levels() - 1; ++i) {
 130     if (vstorage->LevelFilesBrief(i).num_files > 0) {
 131       return Status::NotSupported("Other levels also contain files");
 132     }
 133   }
 134
 135   int level = vstorage->num_non_empty_levels() - 1;
 136   if (vstorage->LevelFilesBrief(level).num_files > 0) {
 137     files_ = vstorage->LevelFilesBrief(level);
 138     return Status::OK();
 139   }
 140   return Status::NotSupported("no file exists");
 141 }
 142
 143 Status CompactedDBImpl::Open(const Options& options,
 144                              const std::string& dbname, DB** dbptr) {
 145   *dbptr = nullptr;
 146
 147   if (options.max_open_files != -1) {
 148     return Status::InvalidArgument("require max_open_files = -1");
 149   }
 150   if (options.merge_operator.get() != nullptr) {
 151     return Status::InvalidArgument("merge operator is not supported");
 152   }
 153   DBOptions db_options(options);
 154   std::unique_ptr<CompactedDBImpl> db(new CompactedDBImpl(db_options, dbname));
 155   Status s = db->Init(options);
 156   if (s.ok()) {
 157     ROCKS_LOG_INFO(db->immutable_db_options_.info_log,
 158                    "Opened the db as fully compacted mode");
 159     LogFlush(db->immutable_db_options_.info_log);
 160     *dbptr = db.release();
 161   }
 162   return s;
 163 }
 164
 165 }   // namespace rocksdb
 166 #endif  // ROCKSDB_LITE