NextGraph
/
rocksdb
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[CF] Options -> DBOptions

Browse Source 
Summary: Replaced most of occurrences of Options with more specific DBOptions. This brings us very close to supporting different configuration options for each column family.

Test Plan: make check

Reviewers: dhruba, haobo, kailiu, sdong

CC: leveldb

Differential Revision: https://reviews.facebook.net/D15933
main
Igor Canadi 11 years ago
parent 6e56ab5702
commit f276e0e59d
12 changed files with 124 additions and 116 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 6
      db/column_family.cc
  2. 7
      db/column_family.h
  3. 147
      db/db_impl.cc
  4. 7
      db/db_impl.h
  5. 17
      db/db_impl_readonly.cc
  6. 2
      db/db_impl_readonly.h
  7. 31
      db/transaction_log_impl.cc
  8. 7
      db/transaction_log_impl.h
  9. 2
      db/version_set.cc
  10. 4
      db/version_set.h
  11. 4
      include/rocksdb/env.h
  12. 6
      util/env.cc

6
db/column_family.cc
Unescape View File

@ -132,7 +132,7 @@ ColumnFamilyData::ColumnFamilyData(const std::string& dbname, uint32_t id,
const std::string& name, const std::string& name,
Version* dummy_versions, Cache* table_cache, Version* dummy_versions, Cache* table_cache,
const ColumnFamilyOptions& options, const ColumnFamilyOptions& options,
const Options* db_options, const DBOptions* db_options,
const EnvOptions& storage_options) const EnvOptions& storage_options)
: id_(id), : id_(id),
name_(name), name_(name),
@ -142,7 +142,7 @@ ColumnFamilyData::ColumnFamilyData(const std::string& dbname, uint32_t id,
internal_filter_policy_(options.filter_policy), internal_filter_policy_(options.filter_policy),
options_(SanitizeOptions(&internal_comparator_, &internal_filter_policy_, options_(SanitizeOptions(&internal_comparator_, &internal_filter_policy_,
options)), options)),
full_options_(DBOptions(*db_options), options_), full_options_(*db_options, options_),
mem_(nullptr), mem_(nullptr),
imm_(options.min_write_buffer_number_to_merge), imm_(options.min_write_buffer_number_to_merge),
super_version_(nullptr), super_version_(nullptr),
@ -237,7 +237,7 @@ SuperVersion* ColumnFamilyData::InstallSuperVersion(
} }
ColumnFamilySet::ColumnFamilySet(const std::string& dbname, ColumnFamilySet::ColumnFamilySet(const std::string& dbname,
const Options* db_options, const DBOptions* db_options,
const EnvOptions& storage_options, const EnvOptions& storage_options,
Cache* table_cache) Cache* table_cache)
: max_column_family_(0), : max_column_family_(0),

7
db/column_family.h
Unescape View File

@ -124,7 +124,7 @@ class ColumnFamilyData {
ColumnFamilyData(const std::string& dbname, uint32_t id, ColumnFamilyData(const std::string& dbname, uint32_t id,
const std::string& name, Version* dummy_versions, const std::string& name, Version* dummy_versions,
Cache* table_cache, const ColumnFamilyOptions& options, Cache* table_cache, const ColumnFamilyOptions& options,
const Options* db_options, const DBOptions* db_options,
const EnvOptions& storage_options); const EnvOptions& storage_options);
~ColumnFamilyData(); ~ColumnFamilyData();
@ -194,7 +194,7 @@ class ColumnFamilySet {
ColumnFamilyData* current_; ColumnFamilyData* current_;
}; };
ColumnFamilySet(const std::string& dbname, const Options* db_options_, ColumnFamilySet(const std::string& dbname, const DBOptions* db_options,
const EnvOptions& storage_options, Cache* table_cache); const EnvOptions& storage_options, Cache* table_cache);
~ColumnFamilySet(); ~ColumnFamilySet();
@ -230,8 +230,7 @@ class ColumnFamilySet {
ColumnFamilyData* dummy_cfd_; ColumnFamilyData* dummy_cfd_;
const std::string db_name_; const std::string db_name_;
// TODO(icanadi) change to DBOptions const DBOptions* const db_options_;
const Options* const db_options_;
const EnvOptions storage_options_; const EnvOptions storage_options_;
Cache* table_cache_; Cache* table_cache_;
}; };

147
db/db_impl.cc
Unescape View File

@ -203,13 +203,10 @@ CompressionType GetCompressionFlush(const Options& options) {
} }
} }
DBImpl::DBImpl(const Options& options, const std::string& dbname) DBImpl::DBImpl(const DBOptions& options, const std::string& dbname)
: env_(options.env), : env_(options.env),
dbname_(dbname), dbname_(dbname),
internal_comparator_(options.comparator), options_(SanitizeOptions(dbname, options)),
options_(SanitizeOptions(dbname, &internal_comparator_,
&internal_filter_policy_, options)),
internal_filter_policy_(options.filter_policy),
// Reserve ten files or so for other uses and give the rest to TableCache. // Reserve ten files or so for other uses and give the rest to TableCache.
table_cache_(NewLRUCache(options_.max_open_files - 10, table_cache_(NewLRUCache(options_.max_open_files - 10,
options_.table_cache_numshardbits, options_.table_cache_numshardbits,
@ -245,7 +242,8 @@ DBImpl::DBImpl(const Options& options, const std::string& dbname)
new ColumnFamilyMemTablesImpl(versions_->GetColumnFamilySet())); new ColumnFamilyMemTablesImpl(versions_->GetColumnFamilySet()));
dumpLeveldbBuildVersion(options_.info_log.get()); dumpLeveldbBuildVersion(options_.info_log.get());
options_.Dump(options_.info_log.get()); // TODO(icanadi) dump DBOptions and ColumnFamilyOptions separately
// options_.Dump(options_.info_log.get());
char name[100]; char name[100];
Status st = env_->GetHostName(name, 100L); Status st = env_->GetHostName(name, 100L);
@ -322,7 +320,6 @@ uint64_t DBImpl::TEST_Current_Manifest_FileNo() {
Status DBImpl::NewDB() { Status DBImpl::NewDB() {
VersionEdit new_db; VersionEdit new_db;
new_db.SetComparatorName(internal_comparator_.user_comparator()->Name());
new_db.SetLogNumber(0); new_db.SetLogNumber(0);
new_db.SetNextFile(2); new_db.SetNextFile(2);
new_db.SetLastSequence(0); new_db.SetLastSequence(0);
@ -909,8 +906,9 @@ Status DBImpl::RecoverLogFile(uint64_t log_number, SequenceNumber* max_sequence,
// filter out all the column families that have already // filter out all the column families that have already
// flushed memtables with log_number // flushed memtables with log_number
column_family_memtables_->SetLogNumber(log_number); column_family_memtables_->SetLogNumber(log_number);
// TODO(icanadi) options_
status = WriteBatchInternal::InsertInto( status = WriteBatchInternal::InsertInto(
&batch, column_family_memtables_.get(), &options_); &batch, column_family_memtables_.get(), default_cfd_->full_options());
column_family_memtables_->SetLogNumber(0); column_family_memtables_->SetLogNumber(0);
MaybeIgnoreError(&status); MaybeIgnoreError(&status);
@ -1013,10 +1011,10 @@ Status DBImpl::WriteLevel0TableForRecovery(ColumnFamilyData* cfd, MemTable* mem,
Status s; Status s;
{ {
mutex_.Unlock(); mutex_.Unlock();
s = BuildTable(dbname_, env_, options_, storage_options_, s = BuildTable(dbname_, env_, *cfd->full_options(), storage_options_,
cfd->table_cache(), iter, &meta, cfd->user_comparator(), cfd->table_cache(), iter, &meta, cfd->user_comparator(),
newest_snapshot, earliest_seqno_in_memtable, newest_snapshot, earliest_seqno_in_memtable,
GetCompressionFlush(options_)); GetCompressionFlush(*cfd->full_options()));
LogFlush(options_.info_log); LogFlush(options_.info_log);
mutex_.Lock(); mutex_.Lock();
} }
@ -1078,10 +1076,10 @@ Status DBImpl::WriteLevel0Table(ColumnFamilyData* cfd,
"Level-0 flush table #%lu: started", "Level-0 flush table #%lu: started",
(unsigned long)meta.number); (unsigned long)meta.number);
s = BuildTable(dbname_, env_, options_, storage_options_, s = BuildTable(dbname_, env_, *cfd->full_options(), storage_options_,
cfd->table_cache(), iter, &meta, cfd->user_comparator(), cfd->table_cache(), iter, &meta, cfd->user_comparator(),
newest_snapshot, earliest_seqno_in_memtable, newest_snapshot, earliest_seqno_in_memtable,
GetCompressionFlush(options_)); GetCompressionFlush(*cfd->full_options()));
LogFlush(options_.info_log); LogFlush(options_.info_log);
delete iter; delete iter;
Log(options_.info_log, "Level-0 flush table #%lu: %lu bytes %s", Log(options_.info_log, "Level-0 flush table #%lu: %lu bytes %s",
@ -1643,9 +1641,10 @@ Status DBImpl::RunManualCompaction(ColumnFamilyData* cfd, int input_level,
Status DBImpl::TEST_CompactRange(int level, Status DBImpl::TEST_CompactRange(int level,
const Slice* begin, const Slice* begin,
const Slice* end) { const Slice* end) {
int output_level = (options_.compaction_style == kCompactionStyleUniversal) int output_level =
? level (default_cfd_->options()->compaction_style == kCompactionStyleUniversal)
: level + 1; ? level
: level + 1;
return RunManualCompaction(default_cfd_, level, output_level, begin, end); return RunManualCompaction(default_cfd_, level, output_level, begin, end);
} }
@ -1763,7 +1762,8 @@ Status DBImpl::BackgroundFlush(bool* madeProgress,
void DBImpl::BackgroundCallFlush() { void DBImpl::BackgroundCallFlush() {
bool madeProgress = false; bool madeProgress = false;
DeletionState deletion_state(options_.max_write_buffer_number, true); DeletionState deletion_state(default_cfd_->options()->max_write_buffer_number,
true);
assert(bg_flush_scheduled_); assert(bg_flush_scheduled_);
MutexLock l(&mutex_); MutexLock l(&mutex_);
@ -1814,7 +1814,8 @@ uint64_t DBImpl::TEST_GetLevel0TotalSize() {
void DBImpl::BackgroundCallCompaction() { void DBImpl::BackgroundCallCompaction() {
bool madeProgress = false; bool madeProgress = false;
DeletionState deletion_state(options_.max_write_buffer_number, true); DeletionState deletion_state(default_cfd_->options()->max_write_buffer_number,
true);
MaybeDumpStats(); MaybeDumpStats();
@ -1921,14 +1922,16 @@ Status DBImpl::BackgroundCompaction(bool* madeProgress,
((m->done || manual_end == nullptr) ((m->done || manual_end == nullptr)
? "(end)" ? "(end)"
: manual_end->DebugString().c_str())); : manual_end->DebugString().c_str()));
} else if (!options_.disable_auto_compactions) { } else {
for (auto cfd : *versions_->GetColumnFamilySet()) { for (auto cfd : *versions_->GetColumnFamilySet()) {
c.reset(cfd->PickCompaction()); if (!cfd->options()->disable_auto_compactions) {
if (c != nullptr) { c.reset(cfd->PickCompaction());
// update statistics if (c != nullptr) {
MeasureTime(options_.statistics.get(), NUM_FILES_IN_SINGLE_COMPACTION, // update statistics
c->inputs(0)->size()); MeasureTime(options_.statistics.get(), NUM_FILES_IN_SINGLE_COMPACTION,
break; c->inputs(0)->size());
break;
}
} }
} }
} }
@ -2005,7 +2008,7 @@ Status DBImpl::BackgroundCompaction(bool* madeProgress,
// We only compacted part of the requested range. Update *m // We only compacted part of the requested range. Update *m
// to the range that is left to be compacted. // to the range that is left to be compacted.
// Universal compaction should always compact the whole range // Universal compaction should always compact the whole range
assert(options_.compaction_style != kCompactionStyleUniversal); assert(m->cfd->options()->compaction_style != kCompactionStyleUniversal);
m->tmp_storage = *manual_end; m->tmp_storage = *manual_end;
m->begin = &m->tmp_storage; m->begin = &m->tmp_storage;
} }
@ -2097,11 +2100,11 @@ Status DBImpl::OpenCompactionOutputFile(CompactionState* compact) {
compact->compaction->output_level())); compact->compaction->output_level()));
CompressionType compression_type = GetCompressionType( CompressionType compression_type = GetCompressionType(
options_, compact->compaction->output_level(), *cfd->full_options(), compact->compaction->output_level(),
compact->compaction->enable_compression()); compact->compaction->enable_compression());
compact->builder.reset( compact->builder.reset(GetTableBuilder(
GetTableBuilder(options_, compact->outfile.get(), compression_type)); *cfd->full_options(), compact->outfile.get(), compression_type));
} }
LogFlush(options_.info_log); LogFlush(options_.info_log);
return s; return s;
@ -2286,15 +2289,16 @@ Status DBImpl::DoCompactionWork(CompactionState* compact,
SequenceNumber visible_in_snapshot = kMaxSequenceNumber; SequenceNumber visible_in_snapshot = kMaxSequenceNumber;
std::string compaction_filter_value; std::string compaction_filter_value;
std::vector<char> delete_key; // for compaction filter std::vector<char> delete_key; // for compaction filter
MergeHelper merge(cfd->user_comparator(), options_.merge_operator.get(), MergeHelper merge(
options_.info_log.get(), cfd->user_comparator(), cfd->options()->merge_operator.get(),
false /* internal key corruption is expected */); options_.info_log.get(), false /* internal key corruption is expected */);
auto compaction_filter = cfd->options()->compaction_filter; auto compaction_filter = cfd->options()->compaction_filter;
std::unique_ptr<CompactionFilter> compaction_filter_from_factory = nullptr; std::unique_ptr<CompactionFilter> compaction_filter_from_factory = nullptr;
if (!compaction_filter) { if (!compaction_filter) {
auto context = compact->GetFilterContext(); auto context = compact->GetFilterContext();
compaction_filter_from_factory = compaction_filter_from_factory =
options_.compaction_filter_factory->CreateCompactionFilter(context); cfd->options()->compaction_filter_factory->CreateCompactionFilter(
context);
compaction_filter = compaction_filter_from_factory.get(); compaction_filter = compaction_filter_from_factory.get();
} }
@ -2706,8 +2710,9 @@ std::pair<Iterator*, Iterator*> DBImpl::GetTailingIteratorPair(
Iterator* mutable_iter = super_version->mem->NewIterator(options); Iterator* mutable_iter = super_version->mem->NewIterator(options);
// create a DBIter that only uses memtable content; see NewIterator() // create a DBIter that only uses memtable content; see NewIterator()
mutable_iter = NewDBIterator(&dbname_, env_, options_, cfd->user_comparator(), mutable_iter =
mutable_iter, kMaxSequenceNumber); NewDBIterator(&dbname_, env_, *cfd->full_options(),
cfd->user_comparator(), mutable_iter, kMaxSequenceNumber);
std::vector<Iterator*> list; std::vector<Iterator*> list;
super_version->imm->AddIterators(options, &list); super_version->imm->AddIterators(options, &list);
@ -2717,8 +2722,8 @@ std::pair<Iterator*, Iterator*> DBImpl::GetTailingIteratorPair(
// create a DBIter that only uses memtable content; see NewIterator() // create a DBIter that only uses memtable content; see NewIterator()
immutable_iter = immutable_iter =
NewDBIterator(&dbname_, env_, options_, cfd->user_comparator(), NewDBIterator(&dbname_, env_, *cfd->full_options(),
immutable_iter, kMaxSequenceNumber); cfd->user_comparator(), immutable_iter, kMaxSequenceNumber);
// register cleanups // register cleanups
mutable_iter->RegisterCleanup(CleanupIteratorState, mutable_iter->RegisterCleanup(CleanupIteratorState,
@ -2801,21 +2806,23 @@ Status DBImpl::GetImpl(const ReadOptions& options,
// s is both in/out. When in, s could either be OK or MergeInProgress. // s is both in/out. When in, s could either be OK or MergeInProgress.
// merge_operands will contain the sequence of merges in the latter case. // merge_operands will contain the sequence of merges in the latter case.
LookupKey lkey(key, snapshot); LookupKey lkey(key, snapshot);
if (get_version->mem->Get(lkey, value, &s, merge_context, options_)) { if (get_version->mem->Get(lkey, value, &s, merge_context,
*cfd->full_options())) {
// Done // Done
RecordTick(options_.statistics.get(), MEMTABLE_HIT); RecordTick(options_.statistics.get(), MEMTABLE_HIT);
} else if (get_version->imm->Get(lkey, value, &s, merge_context, options_)) { } else if (get_version->imm->Get(lkey, value, &s, merge_context,
*cfd->full_options())) {
// Done // Done
RecordTick(options_.statistics.get(), MEMTABLE_HIT); RecordTick(options_.statistics.get(), MEMTABLE_HIT);
} else { } else {
get_version->current->Get(options, lkey, value, &s, &merge_context, &stats, get_version->current->Get(options, lkey, value, &s, &merge_context, &stats,
options_, value_found); *cfd->full_options(), value_found);
have_stat_update = true; have_stat_update = true;
RecordTick(options_.statistics.get(), MEMTABLE_MISS); RecordTick(options_.statistics.get(), MEMTABLE_MISS);
} }
bool delete_get_version = false; bool delete_get_version = false;
if (!options_.disable_seek_compaction && have_stat_update) { if (!cfd->options()->disable_seek_compaction && have_stat_update) {
mutex_.Lock(); mutex_.Lock();
if (get_version->current->UpdateStats(stats)) { if (get_version->current->UpdateStats(stats)) {
MaybeScheduleFlushOrCompaction(); MaybeScheduleFlushOrCompaction();
@ -2852,6 +2859,7 @@ std::vector<Status> DBImpl::MultiGet(
SequenceNumber snapshot; SequenceNumber snapshot;
struct MultiGetColumnFamilyData { struct MultiGetColumnFamilyData {
ColumnFamilyData* cfd;
SuperVersion* super_version; SuperVersion* super_version;
Version::GetStats stats; Version::GetStats stats;
bool have_stat_update = false; bool have_stat_update = false;
@ -2873,6 +2881,7 @@ std::vector<Status> DBImpl::MultiGet(
for (auto mgd_iter : multiget_cf_data) { for (auto mgd_iter : multiget_cf_data) {
auto cfd = versions_->GetColumnFamilySet()->GetColumnFamily(mgd_iter.first); auto cfd = versions_->GetColumnFamilySet()->GetColumnFamily(mgd_iter.first);
assert(cfd != nullptr); assert(cfd != nullptr);
mgd_iter.second->cfd = cfd;
mgd_iter.second->super_version = cfd->GetSuperVersion()->Ref(); mgd_iter.second->super_version = cfd->GetSuperVersion()->Ref();
} }
mutex_.Unlock(); mutex_.Unlock();
@ -2902,14 +2911,16 @@ std::vector<Status> DBImpl::MultiGet(
assert(mgd_iter != multiget_cf_data.end()); assert(mgd_iter != multiget_cf_data.end());
auto mgd = mgd_iter->second; auto mgd = mgd_iter->second;
auto super_version = mgd->super_version; auto super_version = mgd->super_version;
if (super_version->mem->Get(lkey, value, &s, merge_context, options_)) { auto cfd = mgd->cfd;
if (super_version->mem->Get(lkey, value, &s, merge_context,
*cfd->full_options())) {
// Done // Done
} else if (super_version->imm->Get(lkey, value, &s, merge_context, } else if (super_version->imm->Get(lkey, value, &s, merge_context,
options_)) { *cfd->full_options())) {
// Done // Done
} else { } else {
super_version->current->Get(options, lkey, value, &s, &merge_context, super_version->current->Get(options, lkey, value, &s, &merge_context,
&mgd->stats, options_); &mgd->stats, *cfd->full_options());
mgd->have_stat_update = true; mgd->have_stat_update = true;
} }
@ -2924,7 +2935,8 @@ std::vector<Status> DBImpl::MultiGet(
mutex_.Lock(); mutex_.Lock();
for (auto mgd_iter : multiget_cf_data) { for (auto mgd_iter : multiget_cf_data) {
auto mgd = mgd_iter.second; auto mgd = mgd_iter.second;
if (!options_.disable_seek_compaction && mgd->have_stat_update) { auto cfd = mgd->cfd;
if (!cfd->options()->disable_seek_compaction && mgd->have_stat_update) {
if (mgd->super_version->current->UpdateStats(mgd->stats)) { if (mgd->super_version->current->UpdateStats(mgd->stats)) {
schedule_flush_or_compaction = true; schedule_flush_or_compaction = true;
} }
@ -3037,7 +3049,7 @@ Iterator* DBImpl::NewIterator(const ReadOptions& options,
iter = NewInternalIterator(options, cfd, super_version); iter = NewInternalIterator(options, cfd, super_version);
iter = NewDBIterator( iter = NewDBIterator(
&dbname_, env_, options_, cfd->user_comparator(), iter, &dbname_, env_, *cfd->full_options(), cfd->user_comparator(), iter,
(options.snapshot != nullptr (options.snapshot != nullptr
? reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_ ? reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_
: latest_snapshot)); : latest_snapshot));
@ -3047,7 +3059,7 @@ Iterator* DBImpl::NewIterator(const ReadOptions& options,
// use extra wrapper to exclude any keys from the results which // use extra wrapper to exclude any keys from the results which
// don't begin with the prefix // don't begin with the prefix
iter = new PrefixFilterIterator(iter, *options.prefix, iter = new PrefixFilterIterator(iter, *options.prefix,
options_.prefix_extractor); cfd->options()->prefix_extractor);
} }
return iter; return iter;
} }
@ -3080,7 +3092,11 @@ Status DBImpl::Put(const WriteOptions& o,
Status DBImpl::Merge(const WriteOptions& o, Status DBImpl::Merge(const WriteOptions& o,
const ColumnFamilyHandle& column_family, const Slice& key, const ColumnFamilyHandle& column_family, const Slice& key,
const Slice& val) { const Slice& val) {
if (!options_.merge_operator) { mutex_.Lock();
auto cfd = versions_->GetColumnFamilySet()->GetColumnFamily(column_family.id);
mutex_.Unlock();
assert(cfd != nullptr);
if (!cfd->options()->merge_operator) {
return Status::NotSupported("Provide a merge_operator when opening DB"); return Status::NotSupported("Provide a merge_operator when opening DB");
} else { } else {
return DB::Merge(o, column_family, key, val); return DB::Merge(o, column_family, key, val);
@ -3186,9 +3202,11 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* my_batch) {
// We'll need to add a spinlock for reading that we also lock when we // We'll need to add a spinlock for reading that we also lock when we
// write to a column family (only on column family add/drop, which is // write to a column family (only on column family add/drop, which is
// a very rare action) // a very rare action)
// TODO(icanadi) options_
status = WriteBatchInternal::InsertInto( status = WriteBatchInternal::InsertInto(
updates, column_family_memtables_.get(), &options_, this, updates, column_family_memtables_.get(),
options_.filter_deletes); default_cfd_->full_options(), this,
default_cfd_->options()->filter_deletes);
if (!status.ok()) { if (!status.ok()) {
// Panic for in-memory corruptions // Panic for in-memory corruptions
@ -3390,7 +3408,8 @@ Status DBImpl::MakeRoomForWrite(ColumnFamilyData* cfd, bool force) {
RecordTick(options_.statistics.get(), STALL_L0_NUM_FILES_MICROS, stall); RecordTick(options_.statistics.get(), STALL_L0_NUM_FILES_MICROS, stall);
cfd->internal_stats()->RecordWriteStall(InternalStats::LEVEL0_NUM_FILES, cfd->internal_stats()->RecordWriteStall(InternalStats::LEVEL0_NUM_FILES,
stall); stall);
} else if (allow_hard_rate_limit_delay && options_.hard_rate_limit > 1.0 && } else if (allow_hard_rate_limit_delay &&
cfd->options()->hard_rate_limit > 1.0 &&
(score = cfd->current()->MaxCompactionScore()) > (score = cfd->current()->MaxCompactionScore()) >
cfd->options()->hard_rate_limit) { cfd->options()->hard_rate_limit) {
// Delay a write when the compaction score for any level is too large. // Delay a write when the compaction score for any level is too large.
@ -3495,7 +3514,7 @@ Env* DBImpl::GetEnv() const {
const Options& DBImpl::GetOptions(const ColumnFamilyHandle& column_family) const Options& DBImpl::GetOptions(const ColumnFamilyHandle& column_family)
const { const {
return options_; return *default_cfd_->full_options();
} }
bool DBImpl::GetProperty(const ColumnFamilyHandle& column_family, bool DBImpl::GetProperty(const ColumnFamilyHandle& column_family,
@ -3699,24 +3718,18 @@ Status DB::OpenWithColumnFamilies(
std::vector<ColumnFamilyHandle>* handles, DB** dbptr) { std::vector<ColumnFamilyHandle>* handles, DB** dbptr) {
*dbptr = nullptr; *dbptr = nullptr;
EnvOptions soptions; EnvOptions soptions;
// TODO temporary until we change DBImpl to accept
// DBOptions instead of Options
ColumnFamilyOptions default_column_family_options;
for (auto cfd : column_families) {
if (cfd.name == default_column_family_name) {
default_column_family_options = cfd.options;
break;
}
}
// default options
Options options(db_options, default_column_family_options);
if (options.block_cache != nullptr && options.no_block_cache) { size_t max_write_buffer_size = 0;
return Status::InvalidArgument( for (auto cf : column_families) {
"no_block_cache is true while block_cache is not nullptr"); max_write_buffer_size =
std::max(max_write_buffer_size, cf.options.write_buffer_size);
if (cf.options.block_cache != nullptr && cf.options.no_block_cache) {
return Status::InvalidArgument(
"no_block_cache is true while block_cache is not nullptr");
}
} }
DBImpl* impl = new DBImpl(options, dbname); DBImpl* impl = new DBImpl(db_options, dbname);
Status s = impl->env_->CreateDirIfMissing(impl->options_.wal_dir); Status s = impl->env_->CreateDirIfMissing(impl->options_.wal_dir);
if (!s.ok()) { if (!s.ok()) {
delete impl; delete impl;
@ -3732,6 +3745,7 @@ Status DB::OpenWithColumnFamilies(
// Handles create_if_missing, error_if_exists // Handles create_if_missing, error_if_exists
s = impl->Recover(column_families); s = impl->Recover(column_families);
if (s.ok()) { if (s.ok()) {
lfile->SetPreallocationBlockSize(1.1 * max_write_buffer_size);
uint64_t new_log_number = impl->versions_->NewFileNumber(); uint64_t new_log_number = impl->versions_->NewFileNumber();
unique_ptr<WritableFile> lfile; unique_ptr<WritableFile> lfile;
soptions.use_mmap_writes = false; soptions.use_mmap_writes = false;
@ -3741,7 +3755,6 @@ Status DB::OpenWithColumnFamilies(
soptions soptions
); );
if (s.ok()) { if (s.ok()) {
lfile->SetPreallocationBlockSize(1.1 * impl->options_.write_buffer_size);
VersionEdit edit; VersionEdit edit;
impl->logfile_number_ = new_log_number; impl->logfile_number_ = new_log_number;
impl->log_.reset(new log::Writer(std::move(lfile))); impl->log_.reset(new log::Writer(std::move(lfile)));

7
db/db_impl.h
Unescape View File

@ -37,7 +37,7 @@ class VersionSet;
class DBImpl : public DB { class DBImpl : public DB {
public: public:
DBImpl(const Options& options, const std::string& dbname); DBImpl(const DBOptions& options, const std::string& dbname);
virtual ~DBImpl(); virtual ~DBImpl();
// Implementations of the DB interface // Implementations of the DB interface
@ -250,8 +250,7 @@ class DBImpl : public DB {
Env* const env_; Env* const env_;
const std::string dbname_; const std::string dbname_;
unique_ptr<VersionSet> versions_; unique_ptr<VersionSet> versions_;
const InternalKeyComparator internal_comparator_; const DBOptions options_;
const Options options_;
Iterator* NewInternalIterator(const ReadOptions&, ColumnFamilyData* cfd, Iterator* NewInternalIterator(const ReadOptions&, ColumnFamilyData* cfd,
SuperVersion* super_version); SuperVersion* super_version);
@ -370,8 +369,6 @@ class DBImpl : public DB {
const ReadOptions& options, ColumnFamilyData* cfd, const ReadOptions& options, ColumnFamilyData* cfd,
uint64_t* superversion_number); uint64_t* superversion_number);
const InternalFilterPolicy internal_filter_policy_;
// table_cache_ provides its own synchronization // table_cache_ provides its own synchronization
std::shared_ptr<Cache> table_cache_; std::shared_ptr<Cache> table_cache_;

17
db/db_impl_readonly.cc
Unescape View File

@ -43,8 +43,8 @@
namespace rocksdb { namespace rocksdb {
DBImplReadOnly::DBImplReadOnly(const Options& options, DBImplReadOnly::DBImplReadOnly(const DBOptions& options,
const std::string& dbname) const std::string& dbname)
: DBImpl(options, dbname) { : DBImpl(options, dbname) {
Log(options_.info_log, "Opening the db in read only mode"); Log(options_.info_log, "Opening the db in read only mode");
} }
@ -62,11 +62,12 @@ Status DBImplReadOnly::Get(const ReadOptions& options,
SuperVersion* super_version = cfd->GetSuperVersion(); SuperVersion* super_version = cfd->GetSuperVersion();
MergeContext merge_context; MergeContext merge_context;
LookupKey lkey(key, snapshot); LookupKey lkey(key, snapshot);
if (super_version->mem->Get(lkey, value, &s, merge_context, options_)) { if (super_version->mem->Get(lkey, value, &s, merge_context,
*cfd->full_options())) {
} else { } else {
Version::GetStats stats; Version::GetStats stats;
super_version->current->Get(options, lkey, value, &s, &merge_context, super_version->current->Get(options, lkey, value, &s, &merge_context,
&stats, options_); &stats, *cfd->full_options());
} }
return s; return s;
} }
@ -79,7 +80,8 @@ Iterator* DBImplReadOnly::NewIterator(const ReadOptions& options,
SequenceNumber latest_snapshot = versions_->LastSequence(); SequenceNumber latest_snapshot = versions_->LastSequence();
Iterator* internal_iter = NewInternalIterator(options, cfd, super_version); Iterator* internal_iter = NewInternalIterator(options, cfd, super_version);
return NewDBIterator( return NewDBIterator(
&dbname_, env_, options_, cfd->user_comparator(), internal_iter, &dbname_, env_, *cfd->full_options(), cfd->user_comparator(),
internal_iter,
(options.snapshot != nullptr (options.snapshot != nullptr
? reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_ ? reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_
: latest_snapshot)); : latest_snapshot));
@ -89,13 +91,14 @@ Status DB::OpenForReadOnly(const Options& options, const std::string& dbname,
DB** dbptr, bool error_if_log_file_exist) { DB** dbptr, bool error_if_log_file_exist) {
*dbptr = nullptr; *dbptr = nullptr;
DBImplReadOnly* impl = new DBImplReadOnly(options, dbname);
impl->mutex_.Lock();
DBOptions db_options(options); DBOptions db_options(options);
ColumnFamilyOptions cf_options(options); ColumnFamilyOptions cf_options(options);
std::vector<ColumnFamilyDescriptor> column_families; std::vector<ColumnFamilyDescriptor> column_families;
column_families.push_back( column_families.push_back(
ColumnFamilyDescriptor(default_column_family_name, cf_options)); ColumnFamilyDescriptor(default_column_family_name, cf_options));
DBImplReadOnly* impl = new DBImplReadOnly(db_options, dbname);
impl->mutex_.Lock();
Status s = impl->Recover(column_families, true /* read only */, Status s = impl->Recover(column_families, true /* read only */,
error_if_log_file_exist); error_if_log_file_exist);
if (s.ok()) { if (s.ok()) {

2
db/db_impl_readonly.h
Unescape View File

@ -26,7 +26,7 @@ namespace rocksdb {
class DBImplReadOnly : public DBImpl { class DBImplReadOnly : public DBImpl {
public: public:
DBImplReadOnly(const Options& options, const std::string& dbname); DBImplReadOnly(const DBOptions& options, const std::string& dbname);
virtual ~DBImplReadOnly(); virtual ~DBImplReadOnly();
// Implementations of the DB interface // Implementations of the DB interface

31
db/transaction_log_impl.cc
Unescape View File

@ -9,23 +9,20 @@
namespace rocksdb { namespace rocksdb {
TransactionLogIteratorImpl::TransactionLogIteratorImpl( TransactionLogIteratorImpl::TransactionLogIteratorImpl(
const std::string& dir, const std::string& dir, const DBOptions* options,
const Options* options, const EnvOptions& soptions, const SequenceNumber seq,
const EnvOptions& soptions, std::unique_ptr<VectorLogPtr> files, DBImpl const* const dbimpl)
const SequenceNumber seq, : dir_(dir),
std::unique_ptr<VectorLogPtr> files, options_(options),
DBImpl const * const dbimpl) : soptions_(soptions),
dir_(dir), startingSequenceNumber_(seq),
options_(options), files_(std::move(files)),
soptions_(soptions), started_(false),
startingSequenceNumber_(seq), isValid_(false),
files_(std::move(files)), currentFileIndex_(0),
started_(false), currentBatchSeq_(0),
isValid_(false), currentLastSeq_(0),
currentFileIndex_(0), dbimpl_(dbimpl) {
currentBatchSeq_(0),
currentLastSeq_(0),
dbimpl_(dbimpl) {
assert(files_ != nullptr); assert(files_ != nullptr);
assert(dbimpl_ != nullptr); assert(dbimpl_ != nullptr);

7
db/transaction_log_impl.h
Unescape View File

@ -66,12 +66,11 @@ class LogFileImpl : public LogFile {
class TransactionLogIteratorImpl : public TransactionLogIterator { class TransactionLogIteratorImpl : public TransactionLogIterator {
public: public:
TransactionLogIteratorImpl(const std::string& dir, TransactionLogIteratorImpl(const std::string& dir, const DBOptions* options,
const Options* options,
const EnvOptions& soptions, const EnvOptions& soptions,
const SequenceNumber seqNum, const SequenceNumber seqNum,
std::unique_ptr<VectorLogPtr> files, std::unique_ptr<VectorLogPtr> files,
DBImpl const * const dbimpl); DBImpl const* const dbimpl);
virtual bool Valid(); virtual bool Valid();
@ -83,7 +82,7 @@ class TransactionLogIteratorImpl : public TransactionLogIterator {
private: private:
const std::string& dir_; const std::string& dir_;
const Options* options_; const DBOptions* options_;
const EnvOptions& soptions_; const EnvOptions& soptions_;
SequenceNumber startingSequenceNumber_; SequenceNumber startingSequenceNumber_;
std::unique_ptr<VectorLogPtr> files_; std::unique_ptr<VectorLogPtr> files_;

2
db/version_set.cc
Unescape View File

@ -1369,7 +1369,7 @@ class VersionSet::Builder {
} }
}; };
VersionSet::VersionSet(const std::string& dbname, const Options* options, VersionSet::VersionSet(const std::string& dbname, const DBOptions* options,
const EnvOptions& storage_options, Cache* table_cache) const EnvOptions& storage_options, Cache* table_cache)
: column_family_set_(new ColumnFamilySet(dbname, options, storage_options, : column_family_set_(new ColumnFamilySet(dbname, options, storage_options,
table_cache)), table_cache)),

4
db/version_set.h
Unescape View File

@ -279,7 +279,7 @@ class Version {
class VersionSet { class VersionSet {
public: public:
VersionSet(const std::string& dbname, const Options* options, VersionSet(const std::string& dbname, const DBOptions* options,
const EnvOptions& storage_options, Cache* table_cache); const EnvOptions& storage_options, Cache* table_cache);
~VersionSet(); ~VersionSet();
@ -422,7 +422,7 @@ class VersionSet {
Env* const env_; Env* const env_;
const std::string dbname_; const std::string dbname_;
const Options* const options_; const DBOptions* const options_;
uint64_t next_file_number_; uint64_t next_file_number_;
uint64_t manifest_file_number_; uint64_t manifest_file_number_;
std::atomic<uint64_t> last_sequence_; std::atomic<uint64_t> last_sequence_;

4
include/rocksdb/env.h
Unescape View File

@ -34,7 +34,7 @@ class Slice;
class WritableFile; class WritableFile;
class RandomRWFile; class RandomRWFile;
class Directory; class Directory;
struct Options; struct DBOptions;
using std::unique_ptr; using std::unique_ptr;
using std::shared_ptr; using std::shared_ptr;
@ -47,7 +47,7 @@ struct EnvOptions {
EnvOptions(); EnvOptions();
// construct from Options // construct from Options
explicit EnvOptions(const Options& options); explicit EnvOptions(const DBOptions& options);
// If true, then allow caching of data in environment buffers // If true, then allow caching of data in environment buffers
bool use_os_buffer = true; bool use_os_buffer = true;

6
util/env.cc
Unescape View File

@ -119,7 +119,7 @@ EnvWrapper::~EnvWrapper() {
namespace { // anonymous namespace namespace { // anonymous namespace
void AssignEnvOptions(EnvOptions* env_options, const Options& options) { void AssignEnvOptions(EnvOptions* env_options, const DBOptions& options) {
env_options->use_os_buffer = options.allow_os_buffer; env_options->use_os_buffer = options.allow_os_buffer;
env_options->use_mmap_reads = options.allow_mmap_reads; env_options->use_mmap_reads = options.allow_mmap_reads;
env_options->use_mmap_writes = options.allow_mmap_writes; env_options->use_mmap_writes = options.allow_mmap_writes;
@ -129,12 +129,12 @@ void AssignEnvOptions(EnvOptions* env_options, const Options& options) {
} }
EnvOptions::EnvOptions(const Options& options) { EnvOptions::EnvOptions(const DBOptions& options) {
AssignEnvOptions(this, options); AssignEnvOptions(this, options);
} }
EnvOptions::EnvOptions() { EnvOptions::EnvOptions() {
Options options; DBOptions options;
AssignEnvOptions(this, options); AssignEnvOptions(this, options);
} }

Write Preview
Loading…
Cancel
Save