diff --git a/db/column_family.cc b/db/column_family.cc index 2c1ad930ab..6b54454f8f 100644 --- a/db/column_family.cc +++ b/db/column_family.cc @@ -200,6 +200,7 @@ const uint64_t kDefaultPeriodicCompSecs = 0xfffffffffffffffe; } // anonymous namespace ColumnFamilyOptions SanitizeOptions(const ImmutableDBOptions& db_options, + bool read_only, const ColumnFamilyOptions& src) { ColumnFamilyOptions result = src; size_t clamp_max = std::conditional< @@ -435,6 +436,18 @@ ColumnFamilyOptions SanitizeOptions(const ImmutableDBOptions& db_options, result.periodic_compaction_seconds = 0; } + if (read_only && (result.preserve_internal_time_seconds > 0 || + result.preclude_last_level_data_seconds > 0)) { + // With no writes coming in, we don't need periodic SeqnoToTime entries. + // Existing SST files may or may not have that info associated with them. + ROCKS_LOG_WARN( + db_options.info_log.get(), + "preserve_internal_time_seconds and preclude_last_level_data_seconds " + "are ignored in read-only DB"); + result.preserve_internal_time_seconds = 0; + result.preclude_last_level_data_seconds = 0; + } + return result; } @@ -492,6 +505,17 @@ void SuperVersion::Init( imm->Ref(); current->Ref(); refs.store(1, std::memory_order_relaxed); + + // There should be at least one mapping entry iff time tracking is enabled. +#ifndef NDEBUG + MinAndMaxPreserveSeconds preserve_info{mutable_cf_options}; + if (preserve_info.IsEnabled()) { + assert(seqno_to_time_mapping); + assert(!seqno_to_time_mapping->Empty()); + } else { + assert(seqno_to_time_mapping == nullptr); + } +#endif // NDEBUG } namespace { @@ -530,7 +554,7 @@ ColumnFamilyData::ColumnFamilyData( const FileOptions* file_options, ColumnFamilySet* column_family_set, BlockCacheTracer* const block_cache_tracer, const std::shared_ptr& io_tracer, const std::string& db_id, - const std::string& db_session_id) + const std::string& db_session_id, bool read_only) : id_(id), name_(name), dummy_versions_(_dummy_versions), @@ -540,7 +564,7 @@ ColumnFamilyData::ColumnFamilyData( dropped_(false), flush_skip_reschedule_(false), internal_comparator_(cf_options.comparator), - initial_cf_options_(SanitizeOptions(db_options, cf_options)), + initial_cf_options_(SanitizeOptions(db_options, read_only, cf_options)), ioptions_(db_options, initial_cf_options_), mutable_cf_options_(initial_cf_options_), is_delete_range_supported_( @@ -1339,20 +1363,17 @@ bool ColumnFamilyData::ReturnThreadLocalSuperVersion(SuperVersion* sv) { return false; } -void ColumnFamilyData::InstallSuperVersion(SuperVersionContext* sv_context, - InstrumentedMutex* db_mutex) { - db_mutex->AssertHeld(); - return InstallSuperVersion(sv_context, mutable_cf_options_); -} - void ColumnFamilyData::InstallSuperVersion( - SuperVersionContext* sv_context, - const MutableCFOptions& mutable_cf_options) { + SuperVersionContext* sv_context, InstrumentedMutex* db_mutex, + std::optional> + new_seqno_to_time_mapping) { + db_mutex->AssertHeld(); + SuperVersion* new_superversion = sv_context->new_superversion.release(); - new_superversion->mutable_cf_options = mutable_cf_options; + new_superversion->mutable_cf_options = GetLatestMutableCFOptions(); new_superversion->Init(this, mem_, imm_.current(), current_, - sv_context->new_seqno_to_time_mapping - ? std::move(sv_context->new_seqno_to_time_mapping) + new_seqno_to_time_mapping.has_value() + ? std::move(new_seqno_to_time_mapping.value()) : super_version_ ? super_version_->ShareSeqnoToTimeMapping() : nullptr); @@ -1365,7 +1386,7 @@ void ColumnFamilyData::InstallSuperVersion( // currently RecalculateWriteStallConditions() treats it as further slowing // down is needed. super_version_->write_stall_condition = - RecalculateWriteStallConditions(mutable_cf_options); + RecalculateWriteStallConditions(new_superversion->mutable_cf_options); } else { super_version_->write_stall_condition = old_superversion->write_stall_condition; @@ -1378,8 +1399,9 @@ void ColumnFamilyData::InstallSuperVersion( ResetThreadLocalSuperVersions(); if (old_superversion->mutable_cf_options.write_buffer_size != - mutable_cf_options.write_buffer_size) { - mem_->UpdateWriteBufferSize(mutable_cf_options.write_buffer_size); + new_superversion->mutable_cf_options.write_buffer_size) { + mem_->UpdateWriteBufferSize( + new_superversion->mutable_cf_options.write_buffer_size); } if (old_superversion->write_stall_condition != new_superversion->write_stall_condition) { @@ -1680,7 +1702,8 @@ ColumnFamilySet::ColumnFamilySet(const std::string& dbname, dummy_cfd_(new ColumnFamilyData( ColumnFamilyData::kDummyColumnFamilyDataId, "", nullptr, nullptr, nullptr, ColumnFamilyOptions(), *db_options, &file_options_, nullptr, - block_cache_tracer, io_tracer, db_id, db_session_id)), + block_cache_tracer, io_tracer, db_id, db_session_id, + /*read_only*/ true)), default_cfd_cache_(nullptr), db_name_(dbname), db_options_(db_options), @@ -1752,12 +1775,12 @@ size_t ColumnFamilySet::NumberOfColumnFamilies() const { // under a DB mutex AND write thread ColumnFamilyData* ColumnFamilySet::CreateColumnFamily( const std::string& name, uint32_t id, Version* dummy_versions, - const ColumnFamilyOptions& options) { + const ColumnFamilyOptions& options, bool read_only) { assert(column_families_.find(name) == column_families_.end()); ColumnFamilyData* new_cfd = new ColumnFamilyData( id, name, dummy_versions, table_cache_, write_buffer_manager_, options, *db_options_, &file_options_, this, block_cache_tracer_, io_tracer_, - db_id_, db_session_id_); + db_id_, db_session_id_, read_only); column_families_.insert({name, id}); column_family_data_.insert({id, new_cfd}); auto ucmp = new_cfd->user_comparator(); diff --git a/db/column_family.h b/db/column_family.h index 51ad803b90..89ededb502 100644 --- a/db/column_family.h +++ b/db/column_family.h @@ -282,6 +282,7 @@ Status CheckCFPathsSupported(const DBOptions& db_options, const ColumnFamilyOptions& cf_options); ColumnFamilyOptions SanitizeOptions(const ImmutableDBOptions& db_options, + bool read_only, const ColumnFamilyOptions& src); // Wrap user defined table properties collector factories `from cf_options` // into internal ones in internal_tbl_prop_coll_factories. Add a system internal @@ -384,10 +385,10 @@ class ColumnFamilyData { Version* dummy_versions() { return dummy_versions_; } Version* current() { return current_; } // REQUIRE: DB mutex held void SetCurrent(Version* _current); // REQUIRE: DB mutex held - uint64_t GetNumLiveVersions() const; // REQUIRE: DB mutex held - uint64_t GetTotalSstFilesSize() const; // REQUIRE: DB mutex held - uint64_t GetLiveSstFilesSize() const; // REQUIRE: DB mutex held - uint64_t GetTotalBlobFileSize() const; // REQUIRE: DB mutex held + uint64_t GetNumLiveVersions() const; // REQUIRE: DB mutex held + uint64_t GetTotalSstFilesSize() const; // REQUIRE: DB mutex held + uint64_t GetLiveSstFilesSize() const; // REQUIRE: DB mutex held + uint64_t GetTotalBlobFileSize() const; // REQUIRE: DB mutex held // REQUIRE: DB mutex held void SetMemtable(MemTable* new_mem) { AssignMemtableID(new_mem); @@ -487,15 +488,11 @@ class ColumnFamilyData { uint64_t GetSuperVersionNumberRelaxed() const { return super_version_number_.load(std::memory_order_relaxed); } - // will return a pointer to SuperVersion* if previous SuperVersion - // if its reference count is zero and needs deletion or nullptr if not - // As argument takes a pointer to allocated SuperVersion to enable - // the clients to allocate SuperVersion outside of mutex. - // IMPORTANT: Only call this from DBImpl::InstallSuperVersion() + // Only intended for use by DBImpl::InstallSuperVersion() and variants void InstallSuperVersion(SuperVersionContext* sv_context, - const MutableCFOptions& mutable_cf_options); - void InstallSuperVersion(SuperVersionContext* sv_context, - InstrumentedMutex* db_mutex); + InstrumentedMutex* db_mutex, + std::optional> + new_seqno_to_time_mapping = {}); void ResetThreadLocalSuperVersions(); @@ -590,16 +587,14 @@ class ColumnFamilyData { private: friend class ColumnFamilySet; - ColumnFamilyData(uint32_t id, const std::string& name, - Version* dummy_versions, Cache* table_cache, - WriteBufferManager* write_buffer_manager, - const ColumnFamilyOptions& options, - const ImmutableDBOptions& db_options, - const FileOptions* file_options, - ColumnFamilySet* column_family_set, - BlockCacheTracer* const block_cache_tracer, - const std::shared_ptr& io_tracer, - const std::string& db_id, const std::string& db_session_id); + ColumnFamilyData( + uint32_t id, const std::string& name, Version* dummy_versions, + Cache* table_cache, WriteBufferManager* write_buffer_manager, + const ColumnFamilyOptions& options, const ImmutableDBOptions& db_options, + const FileOptions* file_options, ColumnFamilySet* column_family_set, + BlockCacheTracer* const block_cache_tracer, + const std::shared_ptr& io_tracer, const std::string& db_id, + const std::string& db_session_id, bool read_only); std::vector GetDbPaths() const; @@ -761,7 +756,8 @@ class ColumnFamilySet { ColumnFamilyData* CreateColumnFamily(const std::string& name, uint32_t id, Version* dummy_version, - const ColumnFamilyOptions& options); + const ColumnFamilyOptions& options, + bool read_only); const UnorderedMap& GetRunningColumnFamiliesTimestampSize() const { diff --git a/db/column_family_test.cc b/db/column_family_test.cc index 29ff2d15ad..224257df49 100644 --- a/db/column_family_test.cc +++ b/db/column_family_test.cc @@ -271,7 +271,8 @@ class ColumnFamilyTestBase : public testing::Test { // them. ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions( ConfigOptions(), desc.options, - SanitizeOptions(dbfull()->immutable_db_options(), current_cf_opt))); + SanitizeOptions(dbfull()->immutable_db_options(), /*read_only*/ false, + current_cf_opt))); cfi++; } } @@ -2248,8 +2249,8 @@ TEST_P(ColumnFamilyTest, SanitizeOptions) { original.write_buffer_size = l * 4 * 1024 * 1024 + i * 1024 * 1024 + j * 1024 + k; - ColumnFamilyOptions result = - SanitizeOptions(ImmutableDBOptions(db_options), original); + ColumnFamilyOptions result = SanitizeOptions( + ImmutableDBOptions(db_options), /*read_only*/ false, original); ASSERT_TRUE(result.level0_stop_writes_trigger >= result.level0_slowdown_writes_trigger); ASSERT_TRUE(result.level0_slowdown_writes_trigger >= diff --git a/db/compaction/compaction_job.cc b/db/compaction/compaction_job.cc index 0ea74891e4..4fba2e6d9b 100644 --- a/db/compaction/compaction_job.cc +++ b/db/compaction/compaction_job.cc @@ -301,8 +301,7 @@ void CompactionJob::Prepare( SequenceNumber preserve_time_min_seqno = kMaxSequenceNumber; SequenceNumber preclude_last_level_min_seqno = kMaxSequenceNumber; uint64_t preserve_time_duration = - std::max(c->mutable_cf_options().preserve_internal_time_seconds, - c->mutable_cf_options().preclude_last_level_data_seconds); + MinAndMaxPreserveSeconds(c->mutable_cf_options()).max_preserve_seconds; if (preserve_time_duration > 0) { const ReadOptions read_options(Env::IOActivity::kCompaction); diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index be175614d8..a3c107e0ab 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -284,9 +284,8 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname, periodic_task_functions_.emplace(PeriodicTaskType::kFlushInfoLog, [this]() { this->FlushInfoLog(); }); periodic_task_functions_.emplace( - PeriodicTaskType::kRecordSeqnoTime, [this]() { - this->RecordSeqnoToTimeMapping(/*populate_historical_seconds=*/0); - }); + PeriodicTaskType::kRecordSeqnoTime, + [this]() { this->RecordSeqnoToTimeMapping(); }); versions_.reset(new VersionSet( dbname_, &immutable_db_options_, file_options_, table_cache_.get(), @@ -843,57 +842,40 @@ Status DBImpl::StartPeriodicTaskScheduler() { return s; } -Status DBImpl::RegisterRecordSeqnoTimeWorker(const ReadOptions& read_options, - const WriteOptions& write_options, - bool is_new_db) { +Status DBImpl::RegisterRecordSeqnoTimeWorker() { options_mutex_.AssertHeld(); - uint64_t min_preserve_seconds = std::numeric_limits::max(); - uint64_t max_preserve_seconds = std::numeric_limits::min(); - std::vector sv_contexts; + // We assume InstallSuperVersionForConfigChange has already ensured suitable + // mappings are present for each relevant CF. We just need to be sure the DB's + // seqno_to_time_mapping_ and worker scheduler are appropriate for the + // combination of CF settings. + + MinAndMaxPreserveSeconds preserve_info; + uint64_t seqno_time_cadence; { InstrumentedMutexLock l(&mutex_); for (auto cfd : *versions_->GetColumnFamilySet()) { auto& mopts = cfd->GetLatestMutableCFOptions(); - // preserve time is the max of 2 options. - uint64_t preserve_seconds = - std::max(mopts.preserve_internal_time_seconds, - mopts.preclude_last_level_data_seconds); - if (!cfd->IsDropped() && preserve_seconds > 0) { - min_preserve_seconds = std::min(preserve_seconds, min_preserve_seconds); - max_preserve_seconds = std::max(preserve_seconds, max_preserve_seconds); + if (!cfd->IsDropped()) { + preserve_info.Combine(mopts); } } - size_t old_mapping_size = seqno_to_time_mapping_.Size(); - if (min_preserve_seconds == std::numeric_limits::max()) { - // Don't track + seqno_time_cadence = preserve_info.GetRecodingCadence(); + if (seqno_time_cadence == 0) { + // To return as much as possible to the feature being disabled, + // clear the existing mapping seqno_to_time_mapping_.SetCapacity(0); seqno_to_time_mapping_.SetMaxTimeSpan(UINT64_MAX); + assert(seqno_to_time_mapping_.Empty()); } else { uint64_t cap = std::min(kMaxSeqnoToTimeEntries, - max_preserve_seconds * kMaxSeqnoTimePairsPerCF / - min_preserve_seconds); + preserve_info.max_preserve_seconds * + kMaxSeqnoTimePairsPerCF / + preserve_info.min_preserve_seconds); seqno_to_time_mapping_.SetCapacity(cap); - seqno_to_time_mapping_.SetMaxTimeSpan(max_preserve_seconds); + seqno_to_time_mapping_.SetMaxTimeSpan(preserve_info.max_preserve_seconds); } - if (old_mapping_size != seqno_to_time_mapping_.Size()) { - InstallSeqnoToTimeMappingInSV(&sv_contexts); - } - } - - // clean up outside db mutex - for (SuperVersionContext& sv_context : sv_contexts) { - sv_context.Clean(); - } - sv_contexts.clear(); - - uint64_t seqno_time_cadence = 0; - if (min_preserve_seconds != std::numeric_limits::max()) { - // round up to 1 when the time_duration is smaller than - // kMaxSeqnoTimePairsPerCF - seqno_time_cadence = (min_preserve_seconds + kMaxSeqnoTimePairsPerCF - 1) / - kMaxSeqnoTimePairsPerCF; } TEST_SYNC_POINT_CALLBACK( @@ -903,64 +885,6 @@ Status DBImpl::RegisterRecordSeqnoTimeWorker(const ReadOptions& read_options, if (seqno_time_cadence == 0) { s = periodic_task_scheduler_.Unregister(PeriodicTaskType::kRecordSeqnoTime); } else { - // Before registering the periodic task, we need to be sure to fulfill two - // promises: - // 1) Any DB created with preserve/preclude options set from the beginning - // will get pre-allocated seqnos with pre-populated time mappings back to - // the times we are interested in. (This will enable future import of data - // while preserving rough write time. We can only do this reliably from - // DB::Open, as otherwise there could be a race between CreateColumnFamily - // and the first Write to the DB, and seqno-to-time mappings need to be - // monotonic. - // 2) In any DB, any data written after setting preserve/preclude options - // must have a reasonable time estimate (so that we can accurately place - // the data), which means at least one entry in seqno_to_time_mapping_. - // - // FIXME: We don't currently guarantee that if the first column family with - // that setting is added or configured after initial DB::Open but before - // the first user Write. Fixing this causes complications with the crash - // test because if DB starts without preserve/preclude option, does some - // user writes but all those writes are lost in crash, then re-opens with - // preserve/preclude option, it sees seqno==1 which looks like one of the - // user writes was recovered, when actually it was not. - bool last_seqno_zero = GetLatestSequenceNumber() == 0; - assert(!is_new_db || last_seqno_zero); - if (is_new_db && last_seqno_zero) { - // Pre-allocate seqnos and pre-populate historical mapping - // We can simply modify these, before writes are allowed - constexpr uint64_t kMax = kMaxSeqnoTimePairsPerSST; - versions_->SetLastAllocatedSequence(kMax); - versions_->SetLastPublishedSequence(kMax); - versions_->SetLastSequence(kMax); - - // And record in manifest, to avoid going backwards in seqno on re-open - // (potentially with different options). Concurrency is simple because we - // are in DB::Open - { - InstrumentedMutexLock l(&mutex_); - VersionEdit edit; - edit.SetLastSequence(kMax); - s = versions_->LogAndApplyToDefaultColumnFamily( - read_options, write_options, &edit, &mutex_, - directories_.GetDbDir()); - if (!s.ok() && versions_->io_status().IsIOError()) { - error_handler_.SetBGError(versions_->io_status(), - BackgroundErrorReason::kManifestWrite); - } - } - - // Pre-populate mappings for reserved sequence numbers. - RecordSeqnoToTimeMapping(max_preserve_seconds); - } else { - if (!last_seqno_zero) { - // Ensure at least one mapping (or log a warning), and - // an updated entry whenever relevant SetOptions is called - RecordSeqnoToTimeMapping(/*populate_historical_seconds=*/0); - } else { - // FIXME (see limitation described above) - } - } - s = periodic_task_scheduler_.Register( PeriodicTaskType::kRecordSeqnoTime, periodic_task_functions_.at(PeriodicTaskType::kRecordSeqnoTime), @@ -1309,8 +1233,9 @@ Status DBImpl::SetOptions( // // (b) Append a new Version without manifest write nor DB mutex release // - // Thus aren't releasing the DB mutex again until the end of this block, - // after installing the new SuperVersion. + // Thus aren't releasing the DB mutex from LogAndApply calling pre_cb, + // through installing the new Version until the end of this block, after + // installing the new SuperVersion. auto pre_cb = [&]() -> Status { Status cb_s = cfd->SetOptions(db_options, options_map); if (cb_s.ok()) { @@ -1335,7 +1260,7 @@ Status DBImpl::SetOptions( // Trigger possible flush/compactions. This has to be before we persist // options to file, otherwise there will be a deadlock with writer // thread. - InstallSuperVersionAndScheduleWork(cfd, &sv_context); + InstallSuperVersionForConfigChange(cfd, &sv_context); persist_options_status = WriteOptionsFile(write_options, true /*db_mutex_already_held*/); bg_cv_.SignalAll(); @@ -1353,8 +1278,7 @@ Status DBImpl::SetOptions( if (s.ok() && (options_map.count("preserve_internal_time_seconds") > 0 || options_map.count("preclude_last_level_data_seconds") > 0)) { - s = RegisterRecordSeqnoTimeWorker(read_options, write_options, - false /* is_new_db*/); + s = RegisterRecordSeqnoTimeWorker(); } ROCKS_LOG_INFO( @@ -3542,7 +3466,7 @@ void DBImpl::MultiGetEntityWithCallback( } Status DBImpl::WrapUpCreateColumnFamilies( - const ReadOptions& read_options, const WriteOptions& write_options, + const WriteOptions& write_options, const std::vector& cf_options) { options_mutex_.AssertHeld(); @@ -3559,8 +3483,7 @@ Status DBImpl::WrapUpCreateColumnFamilies( // Attempt both follow-up actions even if one fails Status s = WriteOptionsFile(write_options, false /*db_mutex_already_held*/); if (register_worker) { - s.UpdateIfOk(RegisterRecordSeqnoTimeWorker(read_options, write_options, - /* is_new_db */ false)); + s.UpdateIfOk(RegisterRecordSeqnoTimeWorker()); } return s; } @@ -3575,8 +3498,7 @@ Status DBImpl::CreateColumnFamily(const ReadOptions& read_options, Status s = CreateColumnFamilyImpl(read_options, write_options, cf_options, column_family, handle); if (s.ok()) { - s.UpdateIfOk( - WrapUpCreateColumnFamilies(read_options, write_options, {&cf_options})); + s.UpdateIfOk(WrapUpCreateColumnFamilies(write_options, {&cf_options})); } return s; } @@ -3603,8 +3525,7 @@ Status DBImpl::CreateColumnFamilies( success_once = true; } if (success_once) { - s.UpdateIfOk( - WrapUpCreateColumnFamilies(read_options, write_options, {&cf_options})); + s.UpdateIfOk(WrapUpCreateColumnFamilies(write_options, {&cf_options})); } return s; } @@ -3634,8 +3555,7 @@ Status DBImpl::CreateColumnFamilies( cf_opts.push_back(&column_families[i].options); } if (success_once) { - s.UpdateIfOk( - WrapUpCreateColumnFamilies(read_options, write_options, cf_opts)); + s.UpdateIfOk(WrapUpCreateColumnFamilies(write_options, cf_opts)); } return s; } @@ -3704,7 +3624,7 @@ Status DBImpl::CreateColumnFamilyImpl(const ReadOptions& read_options, auto* cfd = versions_->GetColumnFamilySet()->GetColumnFamily(column_family_name); assert(cfd != nullptr); - InstallSuperVersionAndScheduleWork(cfd, &sv_context); + InstallSuperVersionForConfigChange(cfd, &sv_context); if (!cfd->mem()->IsSnapshotSupported()) { is_snapshot_supported_ = false; @@ -3788,7 +3708,7 @@ Status DBImpl::DropColumnFamilyImpl(ColumnFamilyHandle* column_family) { Status s; // Save re-aquiring lock for RegisterRecordSeqnoTimeWorker when not // applicable - bool used_preserve_preclude = false; + MinAndMaxPreserveSeconds preserve_info; { InstrumentedMutexLock l(&mutex_); if (cfd->IsDropped()) { @@ -3806,8 +3726,7 @@ Status DBImpl::DropColumnFamilyImpl(ColumnFamilyHandle* column_family) { auto& moptions = cfd->GetLatestMutableCFOptions(); max_total_in_memory_state_ -= moptions.write_buffer_size * moptions.max_write_buffer_number; - used_preserve_preclude = moptions.preserve_internal_time_seconds > 0 || - moptions.preclude_last_level_data_seconds > 0; + preserve_info.Combine(moptions); } if (!cf_support_snapshot) { @@ -3825,9 +3744,8 @@ Status DBImpl::DropColumnFamilyImpl(ColumnFamilyHandle* column_family) { bg_cv_.SignalAll(); } - if (used_preserve_preclude) { - s = RegisterRecordSeqnoTimeWorker(read_options, write_options, - /* is_new_db */ false); + if (preserve_info.IsEnabled()) { + s = RegisterRecordSeqnoTimeWorker(); } if (s.ok()) { @@ -6239,7 +6157,7 @@ Status DBImpl::CreateColumnFamilyWithImport( versions_->LogAndApply(cfd, read_options, write_options, &dummy_edit, &mutex_, directories_.GetDbDir()); if (status.ok()) { - InstallSuperVersionAndScheduleWork(cfd, &dummy_sv_ctx); + InstallSuperVersionForConfigChange(cfd, &dummy_sv_ctx); } } } @@ -6276,7 +6194,7 @@ Status DBImpl::CreateColumnFamilyWithImport( import_job.edit(), &mutex_, directories_.GetDbDir()); if (status.ok()) { - InstallSuperVersionAndScheduleWork(cfd, &sv_context); + InstallSuperVersionForConfigChange(cfd, &sv_context); } } @@ -6736,7 +6654,7 @@ Status DBImpl::GetCreationTimeOfOldestFile(uint64_t* creation_time) { } } -void DBImpl::RecordSeqnoToTimeMapping(uint64_t populate_historical_seconds) { +std::pair DBImpl::GetSeqnoToTimeSample() const { // TECHNICALITY: Sample last sequence number *before* time, as prescribed // for SeqnoToTimeMapping. We don't know how long it has been since the last // sequence number was written, so we at least have a one-sided bound by @@ -6745,63 +6663,162 @@ void DBImpl::RecordSeqnoToTimeMapping(uint64_t populate_historical_seconds) { // while holding the DB mutex. (This is really to make testing happy because // it's fine to throw out extra close-but-not-quite-consistent mappings in // production.) - std::vector sv_contexts; - bool success = true; - SequenceNumber seqno; - uint64_t unix_time; + mutex_.AssertHeld(); + SequenceNumber seqno = GetLatestSequenceNumber(); + // HACK/TODO: seqno might be zero but we can't record a mapping for that. + // Start with 1, which should be close enough. + seqno = std::max(seqno, SequenceNumber{1}); + int64_t unix_time_signed = 0; + immutable_db_options_.clock->GetCurrentTime(&unix_time_signed) + .PermitUncheckedError(); // Ignore error + return {seqno, static_cast(unix_time_signed)}; +} + +void DBImpl::EnsureSeqnoToTimeMapping( + const MinAndMaxPreserveSeconds& preserve_info) { + mutex_.AssertHeld(); + assert(preserve_info.IsEnabled()); + + // Atomically with CF creation or mutable option change (see + // InstallSuperVersionForConfigChange()), we need to be sure any data written + // after setting preserve/preclude options must have a reasonable time + // estimate (so that we can accurately place the data), which means at least + // one entry in seqno_to_time_mapping_. It's not critical that `preserve_info` + // take into account all CFs, as that's mostly relevant to how we add + // recurring entries and purge old ones. + + auto [seqno, unix_time_now] = GetSeqnoToTimeSample(); + // Ensure at least one sample that is sufficiently recent + uint64_t unix_time_last_sample = 0; + if (seqno_to_time_mapping_.Empty()) { + // The exact best settings will be found and applied in + // RegisterRecordSeqnoTimeWorker() + seqno_to_time_mapping_.SetCapacity(kMaxSeqnoToTimeEntries); + } else { + unix_time_last_sample = + seqno_to_time_mapping_.GetProximalTimeBeforeSeqno(kMaxSequenceNumber); + } + uint64_t cadence = preserve_info.GetRecodingCadence(); + // Extend cadence so as to avoid stepping on toes of recorder job, which + // could lag a bit. + cadence += 3 + cadence / 100; + if (unix_time_now >= cadence && + unix_time_last_sample <= unix_time_now - cadence) { + assert(seqno > 0); // See GetSeqnoToTimeSample() + // Always successful assuming seqno never go backwards + seqno_to_time_mapping_.Append(seqno, unix_time_now); + } +} + +void DBImpl::PrepopulateSeqnoToTimeMapping( + const MinAndMaxPreserveSeconds& preserve_info) { + // Only for opening a new DB, with preserve/preclude options set + if (!preserve_info.IsEnabled()) { + assert(false); + return; + } + if (GetLatestSequenceNumber() != 0) { + assert(false); + return; + } + + // Here we fulfill the following promise: + // + // Any DB/CF created with preserve/preclude options set from the beginning + // will get pre-allocated seqnos with pre-populated time mappings back to + // the times we are interested in. (This will enable future import of data + // while preserving rough write time. We can only do this reliably from + // DB::Open, as otherwise there could be a race between CreateColumnFamily + // and the first Write to the DB, and seqno-to-time mappings need to be + // monotonic. + // + // FIXME: We don't currently guarantee that if the first column family with + // that setting is added or configured after initial DB::Open but before + // the first user Write. Fixing this causes complications with the crash + // test because if DB starts without preserve/preclude option, does some + // user writes but all those writes are lost in crash, then re-opens with + // preserve/preclude option, it sees seqno==1 which looks like one of the + // user writes was recovered, when actually it was not. + + // Pre-allocate seqnos and pre-populate historical mapping + // We can simply modify these, before writes are allowed + constexpr uint64_t kMax = kMaxSeqnoTimePairsPerSST; + versions_->SetLastAllocatedSequence(kMax); + versions_->SetLastPublishedSequence(kMax); + versions_->SetLastSequence(kMax); + + // And record in manifest, to avoid going backwards in seqno on re-open + // (potentially with different options). Concurrency is simple because we + // are in DB::Open + const WriteOptions write_options(Env::IOActivity::kDBOpen); + const ReadOptions read_options(Env::IOActivity::kDBOpen); + VersionEdit edit; + edit.SetLastSequence(kMax); + Status s = versions_->LogAndApplyToDefaultColumnFamily( + read_options, write_options, &edit, &mutex_, directories_.GetDbDir()); + if (!s.ok() && versions_->io_status().IsIOError()) { + error_handler_.SetBGError(versions_->io_status(), + BackgroundErrorReason::kManifestWrite); + } + + auto [seqno, unix_time_now] = GetSeqnoToTimeSample(); + uint64_t populate_historical_seconds = preserve_info.max_preserve_seconds; + if (seqno > 1 && unix_time_now > populate_historical_seconds) { + // seqno=0 is reserved + SequenceNumber from_seqno = 1; + seqno_to_time_mapping_.PrePopulate( + from_seqno, seqno, unix_time_now - populate_historical_seconds, + unix_time_now); + } else { + // One of these will fail + assert(seqno > 1); + assert(unix_time_now > populate_historical_seconds); + } +} + +void DBImpl::InstallSuperVersionForConfigChange( + ColumnFamilyData* cfd, SuperVersionContext* sv_context) { + MinAndMaxPreserveSeconds preserve_info{cfd->GetLatestCFOptions()}; + std::shared_ptr new_seqno_to_time_mapping; + if (preserve_info.IsEnabled()) { + // TODO: detect & optimize if mapping hasn't changed from previous + // SuperVersion + EnsureSeqnoToTimeMapping(preserve_info); + new_seqno_to_time_mapping = std::make_shared(); + new_seqno_to_time_mapping->CopyFrom(seqno_to_time_mapping_); + } + InstallSuperVersionAndScheduleWork(cfd, sv_context, + std::move(new_seqno_to_time_mapping)); +} + +void DBImpl::RecordSeqnoToTimeMapping() { + SuperVersionContext sv_context; { InstrumentedMutexLock l(&mutex_); + // Record next sample + seqno_to_time_mapping_.Append(GetSeqnoToTimeSample()); + // Create an immutable snapshot for sharing across CFs + std::shared_ptr new_seqno_to_time_mapping = + std::make_shared(); + new_seqno_to_time_mapping->CopyFrom(seqno_to_time_mapping_); - seqno = GetLatestSequenceNumber(); - int64_t unix_time_signed = 0; - immutable_db_options_.clock->GetCurrentTime(&unix_time_signed) - .PermitUncheckedError(); // Ignore error - unix_time = static_cast(unix_time_signed); - - if (populate_historical_seconds > 0) { - if (seqno > 1 && unix_time > populate_historical_seconds) { - // seqno=0 is reserved - SequenceNumber from_seqno = 1; - success = seqno_to_time_mapping_.PrePopulate( - from_seqno, seqno, unix_time - populate_historical_seconds, - unix_time); - InstallSeqnoToTimeMappingInSV(&sv_contexts); - } else { - // One of these will fail - assert(seqno > 1); - assert(unix_time > populate_historical_seconds); - success = false; + // Update in SV of all applicable CFs + for (ColumnFamilyData* cfd : *versions_->GetColumnFamilySet()) { + if (cfd->IsDropped()) { + continue; + } + MinAndMaxPreserveSeconds preserve_info{cfd->GetLatestCFOptions()}; + if (preserve_info.IsEnabled()) { + sv_context.NewSuperVersion(); + cfd->InstallSuperVersion(&sv_context, &mutex_, + new_seqno_to_time_mapping); } - } else { - // FIXME: assert(seqno > 0); - // Always successful assuming seqno never go backwards - seqno_to_time_mapping_.Append(seqno, unix_time); - InstallSeqnoToTimeMappingInSV(&sv_contexts); } + bg_cv_.SignalAll(); } // clean up & report outside db mutex - for (SuperVersionContext& sv_context : sv_contexts) { - sv_context.Clean(); - } - - if (populate_historical_seconds > 0) { - if (success) { - ROCKS_LOG_INFO( - immutable_db_options_.info_log, - "Pre-populated sequence number to time entries: [1,%" PRIu64 - "] -> [%" PRIu64 ",%" PRIu64 "]", - seqno, unix_time - populate_historical_seconds, unix_time); - } else { - ROCKS_LOG_WARN( - immutable_db_options_.info_log, - "Failed to pre-populate sequence number to time entries: [1,%" PRIu64 - "] -> [%" PRIu64 ",%" PRIu64 "]", - seqno, unix_time - populate_historical_seconds, unix_time); - } - } else { - assert(success); - } + sv_context.Clean(); } void DBImpl::TrackOrUntrackFiles( @@ -6860,22 +6877,4 @@ void DBImpl::TrackOrUntrackFiles( } } -void DBImpl::InstallSeqnoToTimeMappingInSV( - std::vector* sv_contexts) { - mutex_.AssertHeld(); - std::shared_ptr new_seqno_to_time_mapping = - std::make_shared(); - new_seqno_to_time_mapping->CopyFrom(seqno_to_time_mapping_); - for (ColumnFamilyData* cfd : *versions_->GetColumnFamilySet()) { - if (cfd->IsDropped()) { - continue; - } - sv_contexts->emplace_back(/*create_superversion=*/true); - sv_contexts->back().new_seqno_to_time_mapping = new_seqno_to_time_mapping; - cfd->InstallSuperVersion(&sv_contexts->back(), - cfd->GetLatestMutableCFOptions()); - } - bg_cv_.SignalAll(); -} - } // namespace ROCKSDB_NAMESPACE diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index 9c3f4dbd7c..b746a7f390 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -1267,27 +1267,18 @@ class DBImpl : public DB { // flush LOG out of application buffer void FlushInfoLog(); - // record current sequence number to time mapping. If - // populate_historical_seconds > 0 then pre-populate all the - // sequence numbers from [1, last] to map to [now minus - // populate_historical_seconds, now]. - void RecordSeqnoToTimeMapping(uint64_t populate_historical_seconds); + // For the background timer job + void RecordSeqnoToTimeMapping(); - // Everytime DB's seqno to time mapping changed (which already hold the db - // mutex), we install a new SuperVersion in each column family with a shared - // copy of the new mapping while holding the db mutex. - // This is done for all column families even though the column family does not - // explicitly enabled the - // `preclude_last_level_data_seconds` or `preserve_internal_time_seconds` - // features. - // This mapping supports iterators to fulfill the - // "rocksdb.iterator.write-time" iterator property for entries in memtables. - // - // Since this new SuperVersion doesn't involve an LSM tree shape change, we - // don't schedule work after installing this SuperVersion. It returns the used - // `SuperVersionContext` for clean up after release mutex. - void InstallSeqnoToTimeMappingInSV( - std::vector* sv_contexts); + // REQUIRES: DB mutex held + std::pair GetSeqnoToTimeSample() const; + + // REQUIRES: DB mutex held or during open + void EnsureSeqnoToTimeMapping(const MinAndMaxPreserveSeconds& preserve_secs); + + // Only called during open + void PrepopulateSeqnoToTimeMapping( + const MinAndMaxPreserveSeconds& preserve_secs); // Interface to block and signal the DB in case of stalling writes by // WriteBufferManager. Each DBImpl object contains ptr to WBMStallInterface. @@ -1979,7 +1970,7 @@ class DBImpl : public DB { // Follow-up work to user creating a column family or (families) Status WrapUpCreateColumnFamilies( - const ReadOptions& read_options, const WriteOptions& write_options, + const WriteOptions& write_options, const std::vector& cf_options); Status DropColumnFamilyImpl(ColumnFamilyHandle* column_family); @@ -2450,9 +2441,7 @@ class DBImpl : public DB { // Cancel scheduled periodic tasks Status CancelPeriodicTaskScheduler(); - Status RegisterRecordSeqnoTimeWorker(const ReadOptions& read_options, - const WriteOptions& write_options, - bool is_new_db); + Status RegisterRecordSeqnoTimeWorker(); void PrintStatistics(); @@ -2518,12 +2507,21 @@ class DBImpl : public DB { // Background threads call this function, which is just a wrapper around // the InstallSuperVersion() function. Background threads carry - // sv_context which can have new_superversion already - // allocated. + // sv_context to allow allocation of SuperVersion object outside of holding + // the DB mutex. // All ColumnFamily state changes go through this function. Here we analyze // the new state and we schedule background work if we detect that the new // state needs flush or compaction. - void InstallSuperVersionAndScheduleWork(ColumnFamilyData* cfd, + // See also InstallSuperVersionForConfigChange(). + void InstallSuperVersionAndScheduleWork( + ColumnFamilyData* cfd, SuperVersionContext* sv_context, + std::optional> + new_seqno_to_time_mapping = {}); + + // A variant of InstallSuperVersionAndScheduleWork() that must be used for + // new CFs or for changes to mutable_cf_options. This is so that it can + // update seqno_to_time_mapping cached for the new SuperVersion as relevant. + void InstallSuperVersionForConfigChange(ColumnFamilyData* cfd, SuperVersionContext* sv_context); bool GetIntPropertyInternal(ColumnFamilyData* cfd, diff --git a/db/db_impl/db_impl_compaction_flush.cc b/db/db_impl/db_impl_compaction_flush.cc index 0cbb6c79e3..f623511d73 100644 --- a/db/db_impl/db_impl_compaction_flush.cc +++ b/db/db_impl/db_impl_compaction_flush.cc @@ -4273,9 +4273,10 @@ void DBImpl::BuildCompactionJobInfo( // for superversion_to_free void DBImpl::InstallSuperVersionAndScheduleWork( - ColumnFamilyData* cfd, SuperVersionContext* sv_context) { + ColumnFamilyData* cfd, SuperVersionContext* sv_context, + std::optional> + new_seqno_to_time_mapping) { mutex_.AssertHeld(); - const auto& mutable_cf_options = cfd->GetLatestMutableCFOptions(); // Update max_total_in_memory_state_ size_t old_memtable_size = 0; @@ -4289,7 +4290,8 @@ void DBImpl::InstallSuperVersionAndScheduleWork( if (UNLIKELY(sv_context->new_superversion == nullptr)) { sv_context->NewSuperVersion(); } - cfd->InstallSuperVersion(sv_context, mutable_cf_options); + cfd->InstallSuperVersion(sv_context, &mutex_, + std::move(new_seqno_to_time_mapping)); // There may be a small data race here. The snapshot tricking bottommost // compaction may already be released here. But assuming there will always be @@ -4316,9 +4318,10 @@ void DBImpl::InstallSuperVersionAndScheduleWork( MaybeScheduleFlushOrCompaction(); // Update max_total_in_memory_state_ - max_total_in_memory_state_ = max_total_in_memory_state_ - old_memtable_size + - mutable_cf_options.write_buffer_size * - mutable_cf_options.max_write_buffer_number; + max_total_in_memory_state_ = + max_total_in_memory_state_ - old_memtable_size + + cfd->GetLatestMutableCFOptions().write_buffer_size * + cfd->GetLatestMutableCFOptions().max_write_buffer_number; } // ShouldPurge is called by FindObsoleteFiles when doing a full scan, diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc index 577a861dcc..85a491066e 100644 --- a/db/db_impl/db_impl_open.cc +++ b/db/db_impl/db_impl_open.cc @@ -35,8 +35,8 @@ Options SanitizeOptions(const std::string& dbname, const Options& src, auto db_options = SanitizeOptions(dbname, DBOptions(src), read_only, logger_creation_s); ImmutableDBOptions immutable_db_options(db_options); - auto cf_options = - SanitizeOptions(immutable_db_options, ColumnFamilyOptions(src)); + auto cf_options = SanitizeOptions(immutable_db_options, read_only, + ColumnFamilyOptions(src)); return Options(db_options, cf_options); } @@ -2334,9 +2334,11 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, handles->clear(); size_t max_write_buffer_size = 0; + MinAndMaxPreserveSeconds preserve_info; for (const auto& cf : column_families) { max_write_buffer_size = std::max(max_write_buffer_size, cf.options.write_buffer_size); + preserve_info.Combine(cf.options); } auto impl = std::make_unique(db_options, dbname, seq_per_batch, @@ -2469,6 +2471,12 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, s = impl->InitPersistStatsColumnFamily(); } + // After reaching the post-recovery seqno but before creating SuperVersions + // ensure seqno to time mapping is pre-populated as needed. + if (s.ok() && recovery_ctx.is_new_db_ && preserve_info.IsEnabled()) { + impl->PrepopulateSeqnoToTimeMapping(preserve_info); + } + if (s.ok()) { // set column family handles for (const auto& cf : column_families) { @@ -2478,6 +2486,9 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, handles->push_back( new ColumnFamilyHandleImpl(cfd, impl.get(), &impl->mutex_)); impl->NewThreadStatusCfInfo(cfd); + SuperVersionContext sv_context(/* create_superversion */ true); + impl->InstallSuperVersionForConfigChange(cfd, &sv_context); + sv_context.Clean(); } else { if (db_options.create_missing_column_families) { // missing column family, create it @@ -2485,6 +2496,7 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, impl->mutex_.Unlock(); // NOTE: the work normally done in WrapUpCreateColumnFamilies will // be done separately below. + // This includes InstallSuperVersionForConfigChange. s = impl->CreateColumnFamilyImpl(read_options, write_options, cf.options, cf.name, &handle); impl->mutex_.Lock(); @@ -2501,15 +2513,14 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, } } - if (s.ok()) { - SuperVersionContext sv_context(/* create_superversion */ true); - for (auto cfd : *impl->versions_->GetColumnFamilySet()) { - impl->InstallSuperVersionAndScheduleWork(cfd, &sv_context); - } - sv_context.Clean(); - } - if (s.ok() && impl->immutable_db_options_.persist_stats_to_disk) { + // Install SuperVersion for hidden column family + assert(impl->persist_stats_cf_handle_); + assert(impl->persist_stats_cf_handle_->cfd()); + SuperVersionContext sv_context(/* create_superversion */ true); + impl->InstallSuperVersionForConfigChange( + impl->persist_stats_cf_handle_->cfd(), &sv_context); + sv_context.Clean(); // try to read format version s = impl->PersistentStatsProcessFormatVersion(); } @@ -2618,8 +2629,7 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname, s = impl->StartPeriodicTaskScheduler(); } if (s.ok()) { - s = impl->RegisterRecordSeqnoTimeWorker(read_options, write_options, - recovery_ctx.is_new_db_); + s = impl->RegisterRecordSeqnoTimeWorker(); } impl->options_mutex_.Unlock(); if (s.ok()) { diff --git a/db/db_options_test.cc b/db/db_options_test.cc index 99d390db23..d619e8604e 100644 --- a/db/db_options_test.cc +++ b/db/db_options_test.cc @@ -70,7 +70,8 @@ class DBOptionsTest : public DBTestBase { options.env = env_; ImmutableDBOptions db_options(options); test::RandomInitCFOptions(&options, options, rnd); - auto sanitized_options = SanitizeOptions(db_options, options); + auto sanitized_options = + SanitizeOptions(db_options, /*read_only*/ false, options); auto opt_map = GetMutableCFOptionsMap(sanitized_options); delete options.compaction_filter; return opt_map; diff --git a/db/flush_job.cc b/db/flush_job.cc index ac2eaeb6c5..552c122656 100644 --- a/db/flush_job.cc +++ b/db/flush_job.cc @@ -1193,13 +1193,12 @@ void FlushJob::GetEffectiveCutoffUDTForPickedMemTables() { } void FlushJob::GetPrecludeLastLevelMinSeqno() { - if (mutable_cf_options_.preclude_last_level_data_seconds == 0 || - // FIXME: create FlushJob and build SuperVersions such that - // preclude_last_level_data_seconds > 0 implies - // seqno_to_time_mapping_ != nullptr - seqno_to_time_mapping_ == nullptr) { + if (mutable_cf_options_.preclude_last_level_data_seconds == 0) { return; } + // SuperVersion should guarantee this + assert(seqno_to_time_mapping_); + assert(!seqno_to_time_mapping_->Empty()); int64_t current_time = 0; Status s = db_options_.clock->GetCurrentTime(¤t_time); if (!s.ok()) { diff --git a/db/job_context.h b/db/job_context.h index 83e9f5faca..766502ca46 100644 --- a/db/job_context.h +++ b/db/job_context.h @@ -22,6 +22,9 @@ namespace ROCKSDB_NAMESPACE { class MemTable; struct SuperVersion; +// The purpose of this struct is to simplify pushing work such as +// allocation/construction, de-allocation/destruction, and notifications to +// outside of holding the DB mutex. struct SuperVersionContext { struct WriteStallNotification { WriteStallInfo write_stall_info; @@ -35,12 +38,6 @@ struct SuperVersionContext { std::unique_ptr new_superversion; // if nullptr no new superversion - // If not nullptr, a new seqno to time mapping is available to be installed. - // Otherwise, make a shared copy of the one in the existing SuperVersion and - // carry it over to the new SuperVersion. This is moved to the SuperVersion - // during installation. - std::shared_ptr new_seqno_to_time_mapping{nullptr}; - explicit SuperVersionContext(bool create_superversion = false) : new_superversion(create_superversion ? new SuperVersion() : nullptr) {} diff --git a/db/repair.cc b/db/repair.cc index 73671154ba..3918940293 100644 --- a/db/repair.cc +++ b/db/repair.cc @@ -100,13 +100,13 @@ class Repairer { db_options_(SanitizeOptions(dbname_, db_options)), immutable_db_options_(ImmutableDBOptions(db_options_)), icmp_(default_cf_opts.comparator), - default_cf_opts_( - SanitizeOptions(immutable_db_options_, default_cf_opts)), + default_cf_opts_(SanitizeOptions(immutable_db_options_, + /*read_only*/ false, default_cf_opts)), default_iopts_( ImmutableOptions(immutable_db_options_, default_cf_opts_)), default_mopts_(MutableCFOptions(default_cf_opts_)), - unknown_cf_opts_( - SanitizeOptions(immutable_db_options_, unknown_cf_opts)), + unknown_cf_opts_(SanitizeOptions(immutable_db_options_, + /*read_only*/ false, unknown_cf_opts)), create_unknown_cfs_(create_unknown_cfs), raw_table_cache_( // TableCache can be small since we expect each table to be opened diff --git a/db/seqno_time_test.cc b/db/seqno_time_test.cc index cb247edfb7..a23f9ce667 100644 --- a/db/seqno_time_test.cc +++ b/db/seqno_time_test.cc @@ -919,10 +919,11 @@ TEST_P(SeqnoTimeTablePropTest, PrePopulateInDB) { ASSERT_EQ(db_->GetLatestSequenceNumber(), 0); // And even if we re-open read-write, we do not get pre-population, - // because that's only for new DBs. + // because that's only for new DBs. We just get a single bootstrap + // entry as a lower bound on write times of future writes. Reopen(track_options); sttm = dbfull()->TEST_GetSeqnoToTimeMapping(); - ASSERT_EQ(sttm.Size(), 0); + ASSERT_EQ(sttm.Size(), 1); ASSERT_EQ(db_->GetLatestSequenceNumber(), 0); } } diff --git a/db/seqno_to_time_mapping.cc b/db/seqno_to_time_mapping.cc index b540fd9196..36da27c5bf 100644 --- a/db/seqno_to_time_mapping.cc +++ b/db/seqno_to_time_mapping.cc @@ -490,7 +490,7 @@ bool SeqnoToTimeMapping::Append(SequenceNumber seqno, uint64_t time) { return added; } -bool SeqnoToTimeMapping::PrePopulate(SequenceNumber from_seqno, +void SeqnoToTimeMapping::PrePopulate(SequenceNumber from_seqno, SequenceNumber to_seqno, uint64_t from_time, uint64_t to_time) { assert(Empty()); @@ -505,8 +505,6 @@ bool SeqnoToTimeMapping::PrePopulate(SequenceNumber from_seqno, (to_seqno - from_seqno); pairs_.emplace_back(i, t); } - - return /*success*/ true; } std::string SeqnoToTimeMapping::ToHumanString() const { diff --git a/db/seqno_to_time_mapping.h b/db/seqno_to_time_mapping.h index 741e643694..a74041fd9a 100644 --- a/db/seqno_to_time_mapping.h +++ b/db/seqno_to_time_mapping.h @@ -138,7 +138,7 @@ class SeqnoToTimeMapping { // Adds a series of mappings interpolating from from_seqno->from_time to // to_seqno->to_time. This can only be called on an empty object and both // seqno range and time range are inclusive. - bool PrePopulate(SequenceNumber from_seqno, SequenceNumber to_seqno, + void PrePopulate(SequenceNumber from_seqno, SequenceNumber to_seqno, uint64_t from_time, uint64_t to_time); // Append a new entry to the list. The `seqno` should be >= all previous @@ -148,6 +148,10 @@ class SeqnoToTimeMapping { // rather than creating a new entry. bool Append(SequenceNumber seqno, uint64_t time); + bool Append(std::pair seqno_time_pair) { + return Append(seqno_time_pair.first, seqno_time_pair.second); + } + // Clear all entries and (re-)enter enforced mode if not already in that // state. Enforced limits are unchanged. void Clear() { @@ -274,6 +278,48 @@ class SeqnoToTimeMapping { pair_const_iterator FindGreaterEqSeqno(SequenceNumber seqno) const; }; +// A struct to help combining settings across column families +struct MinAndMaxPreserveSeconds { + uint64_t min_preserve_seconds = std::numeric_limits::max(); + uint64_t max_preserve_seconds = std::numeric_limits::min(); + + MinAndMaxPreserveSeconds() = default; + + template + explicit MinAndMaxPreserveSeconds(const CFOpts& opts) { + Combine(opts); + } + + bool IsEnabled() const { + return min_preserve_seconds != std::numeric_limits::max(); + } + + // Incorporate another CF's settings into the result. If preserve/preclude are + // disabled for this CF, they are excluded from the result. + template + void Combine(const CFOpts& opts) { + uint64_t preserve_seconds = std::max(opts.preserve_internal_time_seconds, + opts.preclude_last_level_data_seconds); + if (preserve_seconds > 0) { + min_preserve_seconds = std::min(preserve_seconds, min_preserve_seconds); + max_preserve_seconds = std::max(preserve_seconds, max_preserve_seconds); + } + } + + // Choose how many seconds between mapping samples + uint64_t GetRecodingCadence() const { + if (IsEnabled()) { + // round up to 1 when the time_duration is smaller than + // kMaxSeqnoTimePairsPerCF + return (min_preserve_seconds + kMaxSeqnoTimePairsPerCF - 1) / + kMaxSeqnoTimePairsPerCF; + } else { + // disabled + return 0; + } + } +}; + // === Utility methods used for TimedPut === // // Pack a value Slice and a unix write time into buffer `buf` and return a Slice diff --git a/db/version_edit_handler.cc b/db/version_edit_handler.cc index 52947c484c..d1b5ee68ce 100644 --- a/db/version_edit_handler.cc +++ b/db/version_edit_handler.cc @@ -471,8 +471,8 @@ void VersionEditHandler::CheckIterationResult(const log::Reader& reader, ColumnFamilyData* VersionEditHandler::CreateCfAndInit( const ColumnFamilyOptions& cf_options, const VersionEdit& edit) { uint32_t cf_id = edit.GetColumnFamily(); - ColumnFamilyData* cfd = - version_set_->CreateColumnFamily(cf_options, read_options_, &edit); + ColumnFamilyData* cfd = version_set_->CreateColumnFamily( + cf_options, read_options_, &edit, read_only_); assert(cfd != nullptr); cfd->set_initialized(); assert(builders_.find(cf_id) == builders_.end()); diff --git a/db/version_set.cc b/db/version_set.cc index b560713cbb..da1ad3ea87 100644 --- a/db/version_set.cc +++ b/db/version_set.cc @@ -5741,7 +5741,8 @@ Status VersionSet::ProcessManifestWrites( assert(new_cf_options != nullptr); assert(max_last_sequence == descriptor_last_sequence_); CreateColumnFamily(*new_cf_options, read_options, - first_writer.edit_list.front()); + first_writer.edit_list.front(), + /*read_only*/ false); } else if (first_writer.edit_list.front()->IsColumnFamilyDrop()) { assert(batch_edits.size() == 1); assert(max_last_sequence == descriptor_last_sequence_); @@ -7294,7 +7295,7 @@ uint64_t VersionSet::GetObsoleteSstFilesSize() const { ColumnFamilyData* VersionSet::CreateColumnFamily( const ColumnFamilyOptions& cf_options, const ReadOptions& read_options, - const VersionEdit* edit) { + const VersionEdit* edit, bool read_only) { assert(edit->IsColumnFamilyAdd()); MutableCFOptions dummy_cf_options; @@ -7305,7 +7306,7 @@ ColumnFamilyData* VersionSet::CreateColumnFamily( dummy_versions->Ref(); auto new_cfd = column_family_set_->CreateColumnFamily( edit->GetColumnFamilyName(), edit->GetColumnFamily(), dummy_versions, - cf_options); + cf_options, read_only); Version* v = new Version(new_cfd, this, file_options_, new_cfd->GetLatestMutableCFOptions(), io_tracer_, diff --git a/db/version_set.h b/db/version_set.h index 72ae58f162..d9cc5a8e07 100644 --- a/db/version_set.h +++ b/db/version_set.h @@ -1625,7 +1625,7 @@ class VersionSet { ColumnFamilyData* CreateColumnFamily(const ColumnFamilyOptions& cf_options, const ReadOptions& read_options, - const VersionEdit* edit); + const VersionEdit* edit, bool read_only); Status VerifyFileMetadata(const ReadOptions& read_options, ColumnFamilyData* cfd, const std::string& fpath, diff --git a/tools/db_crashtest.py b/tools/db_crashtest.py index 1f618c8232..831de21fd9 100644 --- a/tools/db_crashtest.py +++ b/tools/db_crashtest.py @@ -455,7 +455,7 @@ blackbox_default_params = { # since we will be killing anyway, use large value for ops_per_thread "ops_per_thread": 100000000, "reopen": 0, - "set_options_one_in": 2000, + "set_options_one_in": 1000, } whitebox_default_params = {