ceph/src/rocksdb/db/db_impl_readonly.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
   7 #include "db/db_impl_readonly.h"
   8
   9 #include "db/compacted_db_impl.h"
  10 #include "db/db_impl.h"
  11 #include "db/db_iter.h"
  12 #include "db/merge_context.h"
  13 #include "db/range_del_aggregator.h"
  14 #include "monitoring/perf_context_imp.h"
  15
  16 namespace rocksdb {
  17
  18 #ifndef ROCKSDB_LITE
  19
  20 DBImplReadOnly::DBImplReadOnly(const DBOptions& db_options,
  21                                const std::string& dbname)
  22     : DBImpl(db_options, dbname) {
  23   ROCKS_LOG_INFO(immutable_db_options_.info_log,
  24                  "Opening the db in read only mode");
  25   LogFlush(immutable_db_options_.info_log);
  26 }
  27
  28 DBImplReadOnly::~DBImplReadOnly() {
  29 }
  30
  31 // Implementations of the DB interface
  32 Status DBImplReadOnly::Get(const ReadOptions& read_options,
  33                            ColumnFamilyHandle* column_family, const Slice& key,
  34                            PinnableSlice* pinnable_val) {
  35   assert(pinnable_val != nullptr);
  36   Status s;
  37   SequenceNumber snapshot = versions_->LastSequence();
  38   auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  39   auto cfd = cfh->cfd();
  40   SuperVersion* super_version = cfd->GetSuperVersion();
  41   MergeContext merge_context;
  42   RangeDelAggregator range_del_agg(cfd->internal_comparator(), snapshot);
  43   LookupKey lkey(key, snapshot);
  44   if (super_version->mem->Get(lkey, pinnable_val->GetSelf(), &s, &merge_context,
  45                               &range_del_agg, read_options)) {
  46     pinnable_val->PinSelf();
  47   } else {
  48     PERF_TIMER_GUARD(get_from_output_files_time);
  49     super_version->current->Get(read_options, lkey, pinnable_val, &s,
  50                                 &merge_context, &range_del_agg);
  51   }
  52   return s;
  53 }
  54
  55 Iterator* DBImplReadOnly::NewIterator(const ReadOptions& read_options,
  56                                       ColumnFamilyHandle* column_family) {
  57   auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  58   auto cfd = cfh->cfd();
  59   SuperVersion* super_version = cfd->GetSuperVersion()->Ref();
  60   SequenceNumber latest_snapshot = versions_->LastSequence();
  61   auto db_iter = NewArenaWrappedDbIterator(
  62       env_, read_options, *cfd->ioptions(), cfd->user_comparator(),
  63       (read_options.snapshot != nullptr
  64            ? reinterpret_cast<const SnapshotImpl*>(read_options.snapshot)
  65                  ->number_
  66            : latest_snapshot),
  67       super_version->mutable_cf_options.max_sequential_skip_in_iterations,
  68       super_version->version_number);
  69   auto internal_iter =
  70       NewInternalIterator(read_options, cfd, super_version, db_iter->GetArena(),
  71                           db_iter->GetRangeDelAggregator());
  72   db_iter->SetIterUnderDBIter(internal_iter);
  73   return db_iter;
  74 }
  75
  76 Status DBImplReadOnly::NewIterators(
  77     const ReadOptions& read_options,
  78     const std::vector<ColumnFamilyHandle*>& column_families,
  79     std::vector<Iterator*>* iterators) {
  80   if (iterators == nullptr) {
  81     return Status::InvalidArgument("iterators not allowed to be nullptr");
  82   }
  83   iterators->clear();
  84   iterators->reserve(column_families.size());
  85   SequenceNumber latest_snapshot = versions_->LastSequence();
  86
  87   for (auto cfh : column_families) {
  88     auto* cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(cfh)->cfd();
  89     auto* sv = cfd->GetSuperVersion()->Ref();
  90     auto* db_iter = NewArenaWrappedDbIterator(
  91         env_, read_options, *cfd->ioptions(), cfd->user_comparator(),
  92         (read_options.snapshot != nullptr
  93              ? reinterpret_cast<const SnapshotImpl*>(read_options.snapshot)
  94                    ->number_
  95              : latest_snapshot),
  96         sv->mutable_cf_options.max_sequential_skip_in_iterations,
  97         sv->version_number);
  98     auto* internal_iter =
  99         NewInternalIterator(read_options, cfd, sv, db_iter->GetArena(),
 100                             db_iter->GetRangeDelAggregator());
 101     db_iter->SetIterUnderDBIter(internal_iter);
 102     iterators->push_back(db_iter);
 103   }
 104
 105   return Status::OK();
 106 }
 107
 108 Status DB::OpenForReadOnly(const Options& options, const std::string& dbname,
 109                            DB** dbptr, bool error_if_log_file_exist) {
 110   *dbptr = nullptr;
 111
 112   // Try to first open DB as fully compacted DB
 113   Status s;
 114   s = CompactedDBImpl::Open(options, dbname, dbptr);
 115   if (s.ok()) {
 116     return s;
 117   }
 118
 119   DBOptions db_options(options);
 120   ColumnFamilyOptions cf_options(options);
 121   std::vector<ColumnFamilyDescriptor> column_families;
 122   column_families.push_back(
 123       ColumnFamilyDescriptor(kDefaultColumnFamilyName, cf_options));
 124   std::vector<ColumnFamilyHandle*> handles;
 125
 126   s = DB::OpenForReadOnly(db_options, dbname, column_families, &handles, dbptr);
 127   if (s.ok()) {
 128     assert(handles.size() == 1);
 129     // i can delete the handle since DBImpl is always holding a
 130     // reference to default column family
 131     delete handles[0];
 132   }
 133   return s;
 134 }
 135
 136 Status DB::OpenForReadOnly(
 137     const DBOptions& db_options, const std::string& dbname,
 138     const std::vector<ColumnFamilyDescriptor>& column_families,
 139     std::vector<ColumnFamilyHandle*>* handles, DB** dbptr,
 140     bool error_if_log_file_exist) {
 141   *dbptr = nullptr;
 142   handles->clear();
 143
 144   DBImplReadOnly* impl = new DBImplReadOnly(db_options, dbname);
 145   impl->mutex_.Lock();
 146   Status s = impl->Recover(column_families, true /* read only */,
 147                            error_if_log_file_exist);
 148   if (s.ok()) {
 149     // set column family handles
 150     for (auto cf : column_families) {
 151       auto cfd =
 152           impl->versions_->GetColumnFamilySet()->GetColumnFamily(cf.name);
 153       if (cfd == nullptr) {
 154         s = Status::InvalidArgument("Column family not found: ", cf.name);
 155         break;
 156       }
 157       handles->push_back(new ColumnFamilyHandleImpl(cfd, impl, &impl->mutex_));
 158     }
 159   }
 160   if (s.ok()) {
 161     for (auto cfd : *impl->versions_->GetColumnFamilySet()) {
 162       delete cfd->InstallSuperVersion(new SuperVersion(), &impl->mutex_);
 163     }
 164   }
 165   impl->mutex_.Unlock();
 166   if (s.ok()) {
 167     *dbptr = impl;
 168     for (auto* h : *handles) {
 169       impl->NewThreadStatusCfInfo(
 170           reinterpret_cast<ColumnFamilyHandleImpl*>(h)->cfd());
 171     }
 172   } else {
 173     for (auto h : *handles) {
 174       delete h;
 175     }
 176     handles->clear();
 177     delete impl;
 178   }
 179   return s;
 180 }
 181
 182 #else  // !ROCKSDB_LITE
 183
 184 Status DB::OpenForReadOnly(const Options& options, const std::string& dbname,
 185                            DB** dbptr, bool error_if_log_file_exist) {
 186   return Status::NotSupported("Not supported in ROCKSDB_LITE.");
 187 }
 188
 189 Status DB::OpenForReadOnly(
 190     const DBOptions& db_options, const std::string& dbname,
 191     const std::vector<ColumnFamilyDescriptor>& column_families,
 192     std::vector<ColumnFamilyHandle*>* handles, DB** dbptr,
 193     bool error_if_log_file_exist) {
 194   return Status::NotSupported("Not supported in ROCKSDB_LITE.");
 195 }
 196 #endif  // !ROCKSDB_LITE
 197
 198 }   // namespace rocksdb