e45673dece
Summary: Context: Index type `kBinarySearchWithFirstKey` added the ability for sst file iterator to sometimes report a key from index without reading the corresponding data block. This is useful when sst blocks are cut at some meaningful boundaries (e.g. one block per key prefix), and many seeks land between blocks (e.g. for each prefix, the ranges of keys in different sst files are nearly disjoint, so a typical seek needs to read a data block from only one file even if all files have the prefix). But this added a new error condition, which rocksdb code was really not equipped to deal with: `InternalIterator::value()` may fail with an IO error or Status::Incomplete, but it's just a method returning a Slice, with no way to report error instead. Before this PR, this type of error wasn't handled at all (an empty slice was returned), and kBinarySearchWithFirstKey implementation was considered a prototype. Now that we (LogDevice) have experimented with kBinarySearchWithFirstKey for a while and confirmed that it's really useful, this PR is adding the missing error handling. It's a pretty inconvenient situation implementation-wise. The error needs to be reported from InternalIterator when trying to access value. But there are ~700 call sites of `InternalIterator::value()`, most of which either can't hit the error condition (because the iterator is reading from memtable or from index or something) or wouldn't benefit from the deferred loading of the value (e.g. compaction iterator that reads all values anyway). Adding error handling to all these call sites would needlessly bloat the code. So instead I made the deferred value loading optional: only the call sites that may use deferred loading have to call the new method `PrepareValue()` before calling `value()`. The feature is enabled with a new bool argument `allow_unprepared_value` to a bunch of methods that create iterators (it wouldn't make sense to put it in ReadOptions because it's completely internal to iterators, with virtually no user-visible effect). Lmk if you have better ideas. Note that the deferred value loading only happens for *internal* iterators. The user-visible iterator (DBIter) always prepares the value before returning from Seek/Next/etc. We could go further and add an API to defer that value loading too, but that's most likely not useful for LogDevice, so it doesn't seem worth the complexity for now. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6621 Test Plan: make -j5 check . Will also deploy to some logdevice test clusters and look at stats. Reviewed By: siying Differential Revision: D20786930 Pulled By: al13n321 fbshipit-source-id: 6da77d918bad3780522e918f17f4d5513d3e99ee
675 lines
25 KiB
C++
675 lines
25 KiB
C++
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
|
|
// This source code is licensed under both the GPLv2 (found in the
|
|
// COPYING file in the root directory) and Apache 2.0 License
|
|
// (found in the LICENSE.Apache file in the root directory).
|
|
|
|
#include "db/db_impl/db_impl_secondary.h"
|
|
|
|
#include <cinttypes>
|
|
|
|
#include "db/arena_wrapped_db_iter.h"
|
|
#include "db/merge_context.h"
|
|
#include "logging/auto_roll_logger.h"
|
|
#include "monitoring/perf_context_imp.h"
|
|
#include "util/cast_util.h"
|
|
|
|
namespace ROCKSDB_NAMESPACE {
|
|
|
|
#ifndef ROCKSDB_LITE
|
|
DBImplSecondary::DBImplSecondary(const DBOptions& db_options,
|
|
const std::string& dbname)
|
|
: DBImpl(db_options, dbname) {
|
|
ROCKS_LOG_INFO(immutable_db_options_.info_log,
|
|
"Opening the db in secondary mode");
|
|
LogFlush(immutable_db_options_.info_log);
|
|
}
|
|
|
|
DBImplSecondary::~DBImplSecondary() {}
|
|
|
|
Status DBImplSecondary::Recover(
|
|
const std::vector<ColumnFamilyDescriptor>& column_families,
|
|
bool /*readonly*/, bool /*error_if_log_file_exist*/,
|
|
bool /*error_if_data_exists_in_logs*/, uint64_t*) {
|
|
mutex_.AssertHeld();
|
|
|
|
JobContext job_context(0);
|
|
Status s;
|
|
s = static_cast<ReactiveVersionSet*>(versions_.get())
|
|
->Recover(column_families, &manifest_reader_, &manifest_reporter_,
|
|
&manifest_reader_status_);
|
|
if (!s.ok()) {
|
|
return s;
|
|
}
|
|
if (immutable_db_options_.paranoid_checks && s.ok()) {
|
|
s = CheckConsistency();
|
|
}
|
|
// Initial max_total_in_memory_state_ before recovery logs.
|
|
max_total_in_memory_state_ = 0;
|
|
for (auto cfd : *versions_->GetColumnFamilySet()) {
|
|
auto* mutable_cf_options = cfd->GetLatestMutableCFOptions();
|
|
max_total_in_memory_state_ += mutable_cf_options->write_buffer_size *
|
|
mutable_cf_options->max_write_buffer_number;
|
|
}
|
|
if (s.ok()) {
|
|
default_cf_handle_ = new ColumnFamilyHandleImpl(
|
|
versions_->GetColumnFamilySet()->GetDefault(), this, &mutex_);
|
|
default_cf_internal_stats_ = default_cf_handle_->cfd()->internal_stats();
|
|
single_column_family_mode_ =
|
|
versions_->GetColumnFamilySet()->NumberOfColumnFamilies() == 1;
|
|
|
|
std::unordered_set<ColumnFamilyData*> cfds_changed;
|
|
s = FindAndRecoverLogFiles(&cfds_changed, &job_context);
|
|
}
|
|
|
|
if (s.IsPathNotFound()) {
|
|
ROCKS_LOG_INFO(immutable_db_options_.info_log,
|
|
"Secondary tries to read WAL, but WAL file(s) have already "
|
|
"been purged by primary.");
|
|
s = Status::OK();
|
|
}
|
|
// TODO: update options_file_number_ needed?
|
|
|
|
job_context.Clean();
|
|
return s;
|
|
}
|
|
|
|
// find new WAL and apply them in order to the secondary instance
|
|
Status DBImplSecondary::FindAndRecoverLogFiles(
|
|
std::unordered_set<ColumnFamilyData*>* cfds_changed,
|
|
JobContext* job_context) {
|
|
assert(nullptr != cfds_changed);
|
|
assert(nullptr != job_context);
|
|
Status s;
|
|
std::vector<uint64_t> logs;
|
|
s = FindNewLogNumbers(&logs);
|
|
if (s.ok() && !logs.empty()) {
|
|
SequenceNumber next_sequence(kMaxSequenceNumber);
|
|
s = RecoverLogFiles(logs, &next_sequence, cfds_changed, job_context);
|
|
}
|
|
return s;
|
|
}
|
|
|
|
// List wal_dir and find all new WALs, return these log numbers
|
|
Status DBImplSecondary::FindNewLogNumbers(std::vector<uint64_t>* logs) {
|
|
assert(logs != nullptr);
|
|
std::vector<std::string> filenames;
|
|
Status s;
|
|
s = env_->GetChildren(immutable_db_options_.wal_dir, &filenames);
|
|
if (s.IsNotFound()) {
|
|
return Status::InvalidArgument("Failed to open wal_dir",
|
|
immutable_db_options_.wal_dir);
|
|
} else if (!s.ok()) {
|
|
return s;
|
|
}
|
|
|
|
// if log_readers_ is non-empty, it means we have applied all logs with log
|
|
// numbers smaller than the smallest log in log_readers_, so there is no
|
|
// need to pass these logs to RecoverLogFiles
|
|
uint64_t log_number_min = 0;
|
|
if (!log_readers_.empty()) {
|
|
log_number_min = log_readers_.begin()->first;
|
|
}
|
|
for (size_t i = 0; i < filenames.size(); i++) {
|
|
uint64_t number;
|
|
FileType type;
|
|
if (ParseFileName(filenames[i], &number, &type) && type == kLogFile &&
|
|
number >= log_number_min) {
|
|
logs->push_back(number);
|
|
}
|
|
}
|
|
// Recover logs in the order that they were generated
|
|
if (!logs->empty()) {
|
|
std::sort(logs->begin(), logs->end());
|
|
}
|
|
return s;
|
|
}
|
|
|
|
Status DBImplSecondary::MaybeInitLogReader(
|
|
uint64_t log_number, log::FragmentBufferedReader** log_reader) {
|
|
auto iter = log_readers_.find(log_number);
|
|
// make sure the log file is still present
|
|
if (iter == log_readers_.end() ||
|
|
iter->second->reader_->GetLogNumber() != log_number) {
|
|
// delete the obsolete log reader if log number mismatch
|
|
if (iter != log_readers_.end()) {
|
|
log_readers_.erase(iter);
|
|
}
|
|
// initialize log reader from log_number
|
|
// TODO: min_log_number_to_keep_2pc check needed?
|
|
// Open the log file
|
|
std::string fname = LogFileName(immutable_db_options_.wal_dir, log_number);
|
|
ROCKS_LOG_INFO(immutable_db_options_.info_log,
|
|
"Recovering log #%" PRIu64 " mode %d", log_number,
|
|
static_cast<int>(immutable_db_options_.wal_recovery_mode));
|
|
|
|
std::unique_ptr<SequentialFileReader> file_reader;
|
|
{
|
|
std::unique_ptr<FSSequentialFile> file;
|
|
Status status = fs_->NewSequentialFile(
|
|
fname, fs_->OptimizeForLogRead(file_options_), &file,
|
|
nullptr);
|
|
if (!status.ok()) {
|
|
*log_reader = nullptr;
|
|
return status;
|
|
}
|
|
file_reader.reset(new SequentialFileReader(
|
|
std::move(file), fname, immutable_db_options_.log_readahead_size));
|
|
}
|
|
|
|
// Create the log reader.
|
|
LogReaderContainer* log_reader_container = new LogReaderContainer(
|
|
env_, immutable_db_options_.info_log, std::move(fname),
|
|
std::move(file_reader), log_number);
|
|
log_readers_.insert(std::make_pair(
|
|
log_number, std::unique_ptr<LogReaderContainer>(log_reader_container)));
|
|
}
|
|
iter = log_readers_.find(log_number);
|
|
assert(iter != log_readers_.end());
|
|
*log_reader = iter->second->reader_;
|
|
return Status::OK();
|
|
}
|
|
|
|
// After manifest recovery, replay WALs and refresh log_readers_ if necessary
|
|
// REQUIRES: log_numbers are sorted in ascending order
|
|
Status DBImplSecondary::RecoverLogFiles(
|
|
const std::vector<uint64_t>& log_numbers, SequenceNumber* next_sequence,
|
|
std::unordered_set<ColumnFamilyData*>* cfds_changed,
|
|
JobContext* job_context) {
|
|
assert(nullptr != cfds_changed);
|
|
assert(nullptr != job_context);
|
|
mutex_.AssertHeld();
|
|
Status status;
|
|
for (auto log_number : log_numbers) {
|
|
log::FragmentBufferedReader* reader = nullptr;
|
|
status = MaybeInitLogReader(log_number, &reader);
|
|
if (!status.ok()) {
|
|
return status;
|
|
}
|
|
assert(reader != nullptr);
|
|
}
|
|
for (auto log_number : log_numbers) {
|
|
auto it = log_readers_.find(log_number);
|
|
assert(it != log_readers_.end());
|
|
log::FragmentBufferedReader* reader = it->second->reader_;
|
|
// Manually update the file number allocation counter in VersionSet.
|
|
versions_->MarkFileNumberUsed(log_number);
|
|
|
|
// Determine if we should tolerate incomplete records at the tail end of the
|
|
// Read all the records and add to a memtable
|
|
std::string scratch;
|
|
Slice record;
|
|
WriteBatch batch;
|
|
|
|
while (reader->ReadRecord(&record, &scratch,
|
|
immutable_db_options_.wal_recovery_mode) &&
|
|
status.ok()) {
|
|
if (record.size() < WriteBatchInternal::kHeader) {
|
|
reader->GetReporter()->Corruption(
|
|
record.size(), Status::Corruption("log record too small"));
|
|
continue;
|
|
}
|
|
WriteBatchInternal::SetContents(&batch, record);
|
|
SequenceNumber seq_of_batch = WriteBatchInternal::Sequence(&batch);
|
|
std::vector<uint32_t> column_family_ids;
|
|
status = CollectColumnFamilyIdsFromWriteBatch(batch, &column_family_ids);
|
|
if (status.ok()) {
|
|
for (const auto id : column_family_ids) {
|
|
ColumnFamilyData* cfd =
|
|
versions_->GetColumnFamilySet()->GetColumnFamily(id);
|
|
if (cfd == nullptr) {
|
|
continue;
|
|
}
|
|
if (cfds_changed->count(cfd) == 0) {
|
|
cfds_changed->insert(cfd);
|
|
}
|
|
const std::vector<FileMetaData*>& l0_files =
|
|
cfd->current()->storage_info()->LevelFiles(0);
|
|
SequenceNumber seq =
|
|
l0_files.empty() ? 0 : l0_files.back()->fd.largest_seqno;
|
|
// If the write batch's sequence number is smaller than the last
|
|
// sequence number of the largest sequence persisted for this column
|
|
// family, then its data must reside in an SST that has already been
|
|
// added in the prior MANIFEST replay.
|
|
if (seq_of_batch <= seq) {
|
|
continue;
|
|
}
|
|
auto curr_log_num = port::kMaxUint64;
|
|
if (cfd_to_current_log_.count(cfd) > 0) {
|
|
curr_log_num = cfd_to_current_log_[cfd];
|
|
}
|
|
// If the active memtable contains records added by replaying an
|
|
// earlier WAL, then we need to seal the memtable, add it to the
|
|
// immutable memtable list and create a new active memtable.
|
|
if (!cfd->mem()->IsEmpty() && (curr_log_num == port::kMaxUint64 ||
|
|
curr_log_num != log_number)) {
|
|
const MutableCFOptions mutable_cf_options =
|
|
*cfd->GetLatestMutableCFOptions();
|
|
MemTable* new_mem =
|
|
cfd->ConstructNewMemtable(mutable_cf_options, seq_of_batch);
|
|
cfd->mem()->SetNextLogNumber(log_number);
|
|
cfd->imm()->Add(cfd->mem(), &job_context->memtables_to_free);
|
|
new_mem->Ref();
|
|
cfd->SetMemtable(new_mem);
|
|
}
|
|
}
|
|
bool has_valid_writes = false;
|
|
status = WriteBatchInternal::InsertInto(
|
|
&batch, column_family_memtables_.get(),
|
|
nullptr /* flush_scheduler */, nullptr /* trim_history_scheduler*/,
|
|
true, log_number, this, false /* concurrent_memtable_writes */,
|
|
next_sequence, &has_valid_writes, seq_per_batch_, batch_per_txn_);
|
|
}
|
|
// If column family was not found, it might mean that the WAL write
|
|
// batch references to the column family that was dropped after the
|
|
// insert. We don't want to fail the whole write batch in that case --
|
|
// we just ignore the update.
|
|
// That's why we set ignore missing column families to true
|
|
// passing null flush_scheduler will disable memtable flushing which is
|
|
// needed for secondary instances
|
|
if (status.ok()) {
|
|
for (const auto id : column_family_ids) {
|
|
ColumnFamilyData* cfd =
|
|
versions_->GetColumnFamilySet()->GetColumnFamily(id);
|
|
if (cfd == nullptr) {
|
|
continue;
|
|
}
|
|
std::unordered_map<ColumnFamilyData*, uint64_t>::iterator iter =
|
|
cfd_to_current_log_.find(cfd);
|
|
if (iter == cfd_to_current_log_.end()) {
|
|
cfd_to_current_log_.insert({cfd, log_number});
|
|
} else if (log_number > iter->second) {
|
|
iter->second = log_number;
|
|
}
|
|
}
|
|
auto last_sequence = *next_sequence - 1;
|
|
if ((*next_sequence != kMaxSequenceNumber) &&
|
|
(versions_->LastSequence() <= last_sequence)) {
|
|
versions_->SetLastAllocatedSequence(last_sequence);
|
|
versions_->SetLastPublishedSequence(last_sequence);
|
|
versions_->SetLastSequence(last_sequence);
|
|
}
|
|
} else {
|
|
// We are treating this as a failure while reading since we read valid
|
|
// blocks that do not form coherent data
|
|
reader->GetReporter()->Corruption(record.size(), status);
|
|
}
|
|
}
|
|
if (!status.ok()) {
|
|
return status;
|
|
}
|
|
}
|
|
// remove logreaders from map after successfully recovering the WAL
|
|
if (log_readers_.size() > 1) {
|
|
auto erase_iter = log_readers_.begin();
|
|
std::advance(erase_iter, log_readers_.size() - 1);
|
|
log_readers_.erase(log_readers_.begin(), erase_iter);
|
|
}
|
|
return status;
|
|
}
|
|
|
|
// Implementation of the DB interface
|
|
Status DBImplSecondary::Get(const ReadOptions& read_options,
|
|
ColumnFamilyHandle* column_family, const Slice& key,
|
|
PinnableSlice* value) {
|
|
return GetImpl(read_options, column_family, key, value);
|
|
}
|
|
|
|
Status DBImplSecondary::GetImpl(const ReadOptions& read_options,
|
|
ColumnFamilyHandle* column_family,
|
|
const Slice& key, PinnableSlice* pinnable_val) {
|
|
assert(pinnable_val != nullptr);
|
|
PERF_CPU_TIMER_GUARD(get_cpu_nanos, env_);
|
|
StopWatch sw(env_, stats_, DB_GET);
|
|
PERF_TIMER_GUARD(get_snapshot_time);
|
|
|
|
auto cfh = static_cast<ColumnFamilyHandleImpl*>(column_family);
|
|
ColumnFamilyData* cfd = cfh->cfd();
|
|
if (tracer_) {
|
|
InstrumentedMutexLock lock(&trace_mutex_);
|
|
if (tracer_) {
|
|
tracer_->Get(column_family, key);
|
|
}
|
|
}
|
|
// Acquire SuperVersion
|
|
SuperVersion* super_version = GetAndRefSuperVersion(cfd);
|
|
SequenceNumber snapshot = versions_->LastSequence();
|
|
MergeContext merge_context;
|
|
SequenceNumber max_covering_tombstone_seq = 0;
|
|
Status s;
|
|
LookupKey lkey(key, snapshot);
|
|
PERF_TIMER_STOP(get_snapshot_time);
|
|
|
|
bool done = false;
|
|
if (super_version->mem->Get(lkey, pinnable_val->GetSelf(),
|
|
/*timestamp=*/nullptr, &s, &merge_context,
|
|
&max_covering_tombstone_seq, read_options)) {
|
|
done = true;
|
|
pinnable_val->PinSelf();
|
|
RecordTick(stats_, MEMTABLE_HIT);
|
|
} else if ((s.ok() || s.IsMergeInProgress()) &&
|
|
super_version->imm->Get(
|
|
lkey, pinnable_val->GetSelf(), /*timestamp=*/nullptr, &s,
|
|
&merge_context, &max_covering_tombstone_seq, read_options)) {
|
|
done = true;
|
|
pinnable_val->PinSelf();
|
|
RecordTick(stats_, MEMTABLE_HIT);
|
|
}
|
|
if (!done && !s.ok() && !s.IsMergeInProgress()) {
|
|
ReturnAndCleanupSuperVersion(cfd, super_version);
|
|
return s;
|
|
}
|
|
if (!done) {
|
|
PERF_TIMER_GUARD(get_from_output_files_time);
|
|
super_version->current->Get(read_options, lkey, pinnable_val,
|
|
/*timestamp=*/nullptr, &s, &merge_context,
|
|
&max_covering_tombstone_seq);
|
|
RecordTick(stats_, MEMTABLE_MISS);
|
|
}
|
|
{
|
|
PERF_TIMER_GUARD(get_post_process_time);
|
|
ReturnAndCleanupSuperVersion(cfd, super_version);
|
|
RecordTick(stats_, NUMBER_KEYS_READ);
|
|
size_t size = pinnable_val->size();
|
|
RecordTick(stats_, BYTES_READ, size);
|
|
RecordTimeToHistogram(stats_, BYTES_PER_READ, size);
|
|
PERF_COUNTER_ADD(get_read_bytes, size);
|
|
}
|
|
return s;
|
|
}
|
|
|
|
Iterator* DBImplSecondary::NewIterator(const ReadOptions& read_options,
|
|
ColumnFamilyHandle* column_family) {
|
|
if (read_options.managed) {
|
|
return NewErrorIterator(
|
|
Status::NotSupported("Managed iterator is not supported anymore."));
|
|
}
|
|
if (read_options.read_tier == kPersistedTier) {
|
|
return NewErrorIterator(Status::NotSupported(
|
|
"ReadTier::kPersistedData is not yet supported in iterators."));
|
|
}
|
|
Iterator* result = nullptr;
|
|
auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
|
|
auto cfd = cfh->cfd();
|
|
ReadCallback* read_callback = nullptr; // No read callback provided.
|
|
if (read_options.tailing) {
|
|
return NewErrorIterator(Status::NotSupported(
|
|
"tailing iterator not supported in secondary mode"));
|
|
} else if (read_options.snapshot != nullptr) {
|
|
// TODO (yanqin) support snapshot.
|
|
return NewErrorIterator(
|
|
Status::NotSupported("snapshot not supported in secondary mode"));
|
|
} else {
|
|
auto snapshot = versions_->LastSequence();
|
|
result = NewIteratorImpl(read_options, cfd, snapshot, read_callback);
|
|
}
|
|
return result;
|
|
}
|
|
|
|
ArenaWrappedDBIter* DBImplSecondary::NewIteratorImpl(
|
|
const ReadOptions& read_options, ColumnFamilyData* cfd,
|
|
SequenceNumber snapshot, ReadCallback* read_callback) {
|
|
assert(nullptr != cfd);
|
|
SuperVersion* super_version = cfd->GetReferencedSuperVersion(this);
|
|
auto db_iter = NewArenaWrappedDbIterator(
|
|
env_, read_options, *cfd->ioptions(), super_version->mutable_cf_options,
|
|
snapshot,
|
|
super_version->mutable_cf_options.max_sequential_skip_in_iterations,
|
|
super_version->version_number, read_callback);
|
|
auto internal_iter =
|
|
NewInternalIterator(read_options, cfd, super_version, db_iter->GetArena(),
|
|
db_iter->GetRangeDelAggregator(), snapshot,
|
|
/* allow_unprepared_value */ true);
|
|
db_iter->SetIterUnderDBIter(internal_iter);
|
|
return db_iter;
|
|
}
|
|
|
|
Status DBImplSecondary::NewIterators(
|
|
const ReadOptions& read_options,
|
|
const std::vector<ColumnFamilyHandle*>& column_families,
|
|
std::vector<Iterator*>* iterators) {
|
|
if (read_options.managed) {
|
|
return Status::NotSupported("Managed iterator is not supported anymore.");
|
|
}
|
|
if (read_options.read_tier == kPersistedTier) {
|
|
return Status::NotSupported(
|
|
"ReadTier::kPersistedData is not yet supported in iterators.");
|
|
}
|
|
ReadCallback* read_callback = nullptr; // No read callback provided.
|
|
if (iterators == nullptr) {
|
|
return Status::InvalidArgument("iterators not allowed to be nullptr");
|
|
}
|
|
iterators->clear();
|
|
iterators->reserve(column_families.size());
|
|
if (read_options.tailing) {
|
|
return Status::NotSupported(
|
|
"tailing iterator not supported in secondary mode");
|
|
} else if (read_options.snapshot != nullptr) {
|
|
// TODO (yanqin) support snapshot.
|
|
return Status::NotSupported("snapshot not supported in secondary mode");
|
|
} else {
|
|
SequenceNumber read_seq = versions_->LastSequence();
|
|
for (auto cfh : column_families) {
|
|
ColumnFamilyData* cfd = static_cast<ColumnFamilyHandleImpl*>(cfh)->cfd();
|
|
iterators->push_back(
|
|
NewIteratorImpl(read_options, cfd, read_seq, read_callback));
|
|
}
|
|
}
|
|
return Status::OK();
|
|
}
|
|
|
|
Status DBImplSecondary::CheckConsistency() {
|
|
mutex_.AssertHeld();
|
|
Status s = DBImpl::CheckConsistency();
|
|
// If DBImpl::CheckConsistency() which is stricter returns success, then we
|
|
// do not need to give a second chance.
|
|
if (s.ok()) {
|
|
return s;
|
|
}
|
|
// It's possible that DBImpl::CheckConssitency() can fail because the primary
|
|
// may have removed certain files, causing the GetFileSize(name) call to
|
|
// fail and returning a PathNotFound. In this case, we take a best-effort
|
|
// approach and just proceed.
|
|
TEST_SYNC_POINT_CALLBACK(
|
|
"DBImplSecondary::CheckConsistency:AfterFirstAttempt", &s);
|
|
|
|
if (immutable_db_options_.skip_checking_sst_file_sizes_on_db_open) {
|
|
return Status::OK();
|
|
}
|
|
|
|
std::vector<LiveFileMetaData> metadata;
|
|
versions_->GetLiveFilesMetaData(&metadata);
|
|
|
|
std::string corruption_messages;
|
|
for (const auto& md : metadata) {
|
|
// md.name has a leading "/".
|
|
std::string file_path = md.db_path + md.name;
|
|
|
|
uint64_t fsize = 0;
|
|
s = env_->GetFileSize(file_path, &fsize);
|
|
if (!s.ok() &&
|
|
(env_->GetFileSize(Rocks2LevelTableFileName(file_path), &fsize).ok() ||
|
|
s.IsPathNotFound())) {
|
|
s = Status::OK();
|
|
}
|
|
if (!s.ok()) {
|
|
corruption_messages +=
|
|
"Can't access " + md.name + ": " + s.ToString() + "\n";
|
|
}
|
|
}
|
|
return corruption_messages.empty() ? Status::OK()
|
|
: Status::Corruption(corruption_messages);
|
|
}
|
|
|
|
Status DBImplSecondary::TryCatchUpWithPrimary() {
|
|
assert(versions_.get() != nullptr);
|
|
assert(manifest_reader_.get() != nullptr);
|
|
Status s;
|
|
// read the manifest and apply new changes to the secondary instance
|
|
std::unordered_set<ColumnFamilyData*> cfds_changed;
|
|
JobContext job_context(0, true /*create_superversion*/);
|
|
{
|
|
InstrumentedMutexLock lock_guard(&mutex_);
|
|
s = static_cast_with_check<ReactiveVersionSet>(versions_.get())
|
|
->ReadAndApply(&mutex_, &manifest_reader_, &cfds_changed);
|
|
|
|
ROCKS_LOG_INFO(immutable_db_options_.info_log, "Last sequence is %" PRIu64,
|
|
static_cast<uint64_t>(versions_->LastSequence()));
|
|
for (ColumnFamilyData* cfd : cfds_changed) {
|
|
if (cfd->IsDropped()) {
|
|
ROCKS_LOG_DEBUG(immutable_db_options_.info_log, "[%s] is dropped\n",
|
|
cfd->GetName().c_str());
|
|
continue;
|
|
}
|
|
VersionStorageInfo::LevelSummaryStorage tmp;
|
|
ROCKS_LOG_DEBUG(immutable_db_options_.info_log,
|
|
"[%s] Level summary: %s\n", cfd->GetName().c_str(),
|
|
cfd->current()->storage_info()->LevelSummary(&tmp));
|
|
}
|
|
|
|
// list wal_dir to discover new WALs and apply new changes to the secondary
|
|
// instance
|
|
if (s.ok()) {
|
|
s = FindAndRecoverLogFiles(&cfds_changed, &job_context);
|
|
}
|
|
if (s.IsPathNotFound()) {
|
|
ROCKS_LOG_INFO(
|
|
immutable_db_options_.info_log,
|
|
"Secondary tries to read WAL, but WAL file(s) have already "
|
|
"been purged by primary.");
|
|
s = Status::OK();
|
|
}
|
|
if (s.ok()) {
|
|
for (auto cfd : cfds_changed) {
|
|
cfd->imm()->RemoveOldMemTables(cfd->GetLogNumber(),
|
|
&job_context.memtables_to_free);
|
|
auto& sv_context = job_context.superversion_contexts.back();
|
|
cfd->InstallSuperVersion(&sv_context, &mutex_);
|
|
sv_context.NewSuperVersion();
|
|
}
|
|
}
|
|
}
|
|
job_context.Clean();
|
|
|
|
// Cleanup unused, obsolete files.
|
|
JobContext purge_files_job_context(0);
|
|
{
|
|
InstrumentedMutexLock lock_guard(&mutex_);
|
|
// Currently, secondary instance does not own the database files, thus it
|
|
// is unnecessary for the secondary to force full scan.
|
|
FindObsoleteFiles(&purge_files_job_context, /*force=*/false);
|
|
}
|
|
if (purge_files_job_context.HaveSomethingToDelete()) {
|
|
PurgeObsoleteFiles(purge_files_job_context);
|
|
}
|
|
purge_files_job_context.Clean();
|
|
return s;
|
|
}
|
|
|
|
Status DB::OpenAsSecondary(const Options& options, const std::string& dbname,
|
|
const std::string& secondary_path, DB** dbptr) {
|
|
*dbptr = nullptr;
|
|
|
|
DBOptions db_options(options);
|
|
ColumnFamilyOptions cf_options(options);
|
|
std::vector<ColumnFamilyDescriptor> column_families;
|
|
column_families.emplace_back(kDefaultColumnFamilyName, cf_options);
|
|
std::vector<ColumnFamilyHandle*> handles;
|
|
|
|
Status s = DB::OpenAsSecondary(db_options, dbname, secondary_path,
|
|
column_families, &handles, dbptr);
|
|
if (s.ok()) {
|
|
assert(handles.size() == 1);
|
|
delete handles[0];
|
|
}
|
|
return s;
|
|
}
|
|
|
|
Status DB::OpenAsSecondary(
|
|
const DBOptions& db_options, const std::string& dbname,
|
|
const std::string& secondary_path,
|
|
const std::vector<ColumnFamilyDescriptor>& column_families,
|
|
std::vector<ColumnFamilyHandle*>* handles, DB** dbptr) {
|
|
*dbptr = nullptr;
|
|
if (db_options.max_open_files != -1) {
|
|
// TODO (yanqin) maybe support max_open_files != -1 by creating hard links
|
|
// on SST files so that db secondary can still have access to old SSTs
|
|
// while primary instance may delete original.
|
|
return Status::InvalidArgument("require max_open_files to be -1");
|
|
}
|
|
|
|
DBOptions tmp_opts(db_options);
|
|
Status s;
|
|
if (nullptr == tmp_opts.info_log) {
|
|
s = CreateLoggerFromOptions(secondary_path, tmp_opts, &tmp_opts.info_log);
|
|
if (!s.ok()) {
|
|
tmp_opts.info_log = nullptr;
|
|
}
|
|
}
|
|
|
|
handles->clear();
|
|
DBImplSecondary* impl = new DBImplSecondary(tmp_opts, dbname);
|
|
impl->versions_.reset(new ReactiveVersionSet(
|
|
dbname, &impl->immutable_db_options_, impl->file_options_,
|
|
impl->table_cache_.get(), impl->write_buffer_manager_,
|
|
&impl->write_controller_));
|
|
impl->column_family_memtables_.reset(
|
|
new ColumnFamilyMemTablesImpl(impl->versions_->GetColumnFamilySet()));
|
|
impl->wal_in_db_path_ = IsWalDirSameAsDBPath(&impl->immutable_db_options_);
|
|
|
|
impl->mutex_.Lock();
|
|
s = impl->Recover(column_families, true, false, false);
|
|
if (s.ok()) {
|
|
for (auto cf : column_families) {
|
|
auto cfd =
|
|
impl->versions_->GetColumnFamilySet()->GetColumnFamily(cf.name);
|
|
if (nullptr == cfd) {
|
|
s = Status::InvalidArgument("Column family not found: ", cf.name);
|
|
break;
|
|
}
|
|
handles->push_back(new ColumnFamilyHandleImpl(cfd, impl, &impl->mutex_));
|
|
}
|
|
}
|
|
SuperVersionContext sv_context(true /* create_superversion */);
|
|
if (s.ok()) {
|
|
for (auto cfd : *impl->versions_->GetColumnFamilySet()) {
|
|
sv_context.NewSuperVersion();
|
|
cfd->InstallSuperVersion(&sv_context, &impl->mutex_);
|
|
}
|
|
}
|
|
impl->mutex_.Unlock();
|
|
sv_context.Clean();
|
|
if (s.ok()) {
|
|
*dbptr = impl;
|
|
for (auto h : *handles) {
|
|
impl->NewThreadStatusCfInfo(
|
|
reinterpret_cast<ColumnFamilyHandleImpl*>(h)->cfd());
|
|
}
|
|
} else {
|
|
for (auto h : *handles) {
|
|
delete h;
|
|
}
|
|
handles->clear();
|
|
delete impl;
|
|
}
|
|
return s;
|
|
}
|
|
#else // !ROCKSDB_LITE
|
|
|
|
Status DB::OpenAsSecondary(const Options& /*options*/,
|
|
const std::string& /*name*/,
|
|
const std::string& /*secondary_path*/,
|
|
DB** /*dbptr*/) {
|
|
return Status::NotSupported("Not supported in ROCKSDB_LITE.");
|
|
}
|
|
|
|
Status DB::OpenAsSecondary(
|
|
const DBOptions& /*db_options*/, const std::string& /*dbname*/,
|
|
const std::string& /*secondary_path*/,
|
|
const std::vector<ColumnFamilyDescriptor>& /*column_families*/,
|
|
std::vector<ColumnFamilyHandle*>* /*handles*/, DB** /*dbptr*/) {
|
|
return Status::NotSupported("Not supported in ROCKSDB_LITE.");
|
|
}
|
|
#endif // !ROCKSDB_LITE
|
|
|
|
} // namespace ROCKSDB_NAMESPACE
|