diff --git a/db/column_family.cc b/db/column_family.cc index 51cd597b55..8983c1f0e0 100644 --- a/db/column_family.cc +++ b/db/column_family.cc @@ -1208,7 +1208,7 @@ Status ColumnFamilyData::RangesOverlapWithMemtables( MergeIteratorBuilder merge_iter_builder(&internal_comparator_, &arena); merge_iter_builder.AddIterator(super_version->mem->NewIterator( read_opts, /*seqno_to_time_mapping=*/nullptr, &arena, - /*prefix_extractor=*/nullptr)); + /*prefix_extractor=*/nullptr, /*for_flush=*/false)); super_version->imm->AddIterators(read_opts, /*seqno_to_time_mapping=*/nullptr, /*prefix_extractor=*/nullptr, &merge_iter_builder, diff --git a/db/column_family.h b/db/column_family.h index baaa9b9042..4cb6ab5965 100644 --- a/db/column_family.h +++ b/db/column_family.h @@ -388,10 +388,14 @@ class ColumnFamilyData { uint64_t GetTotalBlobFileSize() const; // REQUIRE: DB mutex held // REQUIRE: DB mutex held void SetMemtable(MemTable* new_mem) { - new_mem->SetID(++last_memtable_id_); + AssignMemtableID(new_mem); mem_ = new_mem; } + void AssignMemtableID(ReadOnlyMemTable* new_imm) { + new_imm->SetID(++last_memtable_id_); + } + // calculate the oldest log needed for the durability of this column family uint64_t OldestLogToKeep(); diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 05d1c04153..3bc2be1f04 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -2098,7 +2098,8 @@ InternalIterator* DBImpl::NewInternalIterator( // Collect iterator for mutable memtable auto mem_iter = super_version->mem->NewIterator( read_options, super_version->GetSeqnoToTimeMapping(), arena, - super_version->mutable_cf_options.prefix_extractor.get()); + super_version->mutable_cf_options.prefix_extractor.get(), + /*for_flush=*/false); Status s; if (!read_options.ignore_range_deletions) { std::unique_ptr mem_tombstone_iter; diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index 1492fe7a4a..a66ad12e78 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -48,6 +48,7 @@ #include "db/write_controller.h" #include "db/write_thread.h" #include "logging/event_logger.h" +#include "memtable/wbwi_memtable.h" #include "monitoring/instrumented_mutex.h" #include "options/db_options.h" #include "port/port.h" @@ -60,6 +61,7 @@ #include "rocksdb/transaction_log.h" #include "rocksdb/user_write_callback.h" #include "rocksdb/utilities/replayer.h" +#include "rocksdb/utilities/write_batch_with_index.h" #include "rocksdb/write_buffer_manager.h" #include "table/merging_iterator.h" #include "util/autovector.h" @@ -1507,6 +1509,23 @@ class DBImpl : public DB { void EraseThreadStatusDbInfo() const; + // For CFs that has updates in `wbwi`, their memtable will be switched, + // and `wbwi` will be added as the latest immutable memtable. + // + // REQUIRES: this thread is currently at the front of the main writer queue. + // @param prep_log refers to the WAL that contains prepare record + // for the transaction based on wbwi. + // @param assigned_seqno Sequence numbers for the ingested memtable. + // @param last_seqno the value of versions_->LastSequence() after the write + // ingests `wbwi` is done. + // @param memtable_updated Whether the same write that ingests wbwi has + // updated memtable. This is useful for determining whether to set bg + // error when IngestWBWI fails. + Status IngestWBWI(std::shared_ptr wbwi, + const WBWIMemTable::SeqnoRange& assigned_seqno, + uint64_t min_prep_log, SequenceNumber last_seqno, + bool memtable_updated, bool ignore_missing_cf); + // If disable_memtable is set the application logic must guarantee that the // batch will still be skipped from memtable during the recovery. An excption // to this is seq_per_batch_ mode, in which since each batch already takes one @@ -1522,6 +1541,16 @@ class DBImpl : public DB { // batch_cnt is expected to be non-zero in seq_per_batch mode and // indicates the number of sub-patches. A sub-patch is a subset of the write // batch that does not have duplicate keys. + // `callback` is called before WAL write. + // See more in comment above WriteCallback::Callback(). + // pre_release_callback is called after WAL write and before memtable write. + // See more in comment above PreReleaseCallback::Callback(). + // post_memtable_callback is called after memtable write but before publishing + // the sequence number to readers. + // + // The main write queue. This is the only write queue that updates + // LastSequence. When using one write queue, the same sequence also indicates + // the last published sequence. Status WriteImpl(const WriteOptions& options, WriteBatch* updates, WriteCallback* callback = nullptr, UserWriteCallback* user_write_cb = nullptr, @@ -1529,7 +1558,9 @@ class DBImpl : public DB { bool disable_memtable = false, uint64_t* seq_used = nullptr, size_t batch_cnt = 0, PreReleaseCallback* pre_release_callback = nullptr, - PostMemTableCallback* post_memtable_callback = nullptr); + PostMemTableCallback* post_memtable_callback = nullptr, + std::shared_ptr wbwi = nullptr, + uint64_t min_prep_log = 0); Status PipelinedWriteImpl(const WriteOptions& options, WriteBatch* updates, WriteCallback* callback = nullptr, @@ -2053,7 +2084,19 @@ class DBImpl : public DB { // Switches the current live memtable to immutable/read-only memtable. // A new WAL is created if the current WAL is not empty. - Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context); + // If `new_imm` is not nullptr, it will be added as the newest immutable + // memtable, if and only if OK status is returned. + // `last_seqno` needs to be provided if `new_imm` is not nullptr. It is + // the value of versions_->LastSequence() after the write that ingests new_imm + // is done. + // + // REQUIRES: mutex_ is held + // REQUIRES: this thread is currently at the front of the writer queue + // REQUIRES: this thread is currently at the front of the 2nd writer queue if + // two_write_queues_ is true (This is to simplify the reasoning.) + Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, + ReadOnlyMemTable* new_imm = nullptr, + SequenceNumber last_seqno = 0); // Select and output column families qualified for atomic flush in // `selected_cfds`. If `provided_candidate_cfds` is non-empty, it will be used diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc index e902968249..4601c40a6e 100644 --- a/db/db_impl/db_impl_open.cc +++ b/db/db_impl/db_impl_open.cc @@ -1373,6 +1373,9 @@ Status DBImpl::RecoverLogFiles(const std::vector& wal_numbers, } } } + ROCKS_LOG_INFO(immutable_db_options_.info_log, + "Recovered to log #%" PRIu64 " seq #%" PRIu64, wal_number, + *next_sequence); if (!status.ok() || old_log_record) { if (status.IsNotSupported()) { @@ -1405,10 +1408,6 @@ Status DBImpl::RecoverLogFiles(const std::vector& wal_numbers, if (corrupted_wal_found != nullptr) { *corrupted_wal_found = true; } - ROCKS_LOG_INFO(immutable_db_options_.info_log, - "Point in time recovered to log #%" PRIu64 - " seq #%" PRIu64, - wal_number, *next_sequence); } else { assert(immutable_db_options_.wal_recovery_mode == WALRecoveryMode::kTolerateCorruptedTailRecords || @@ -1681,7 +1680,8 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd, ro, /*seqno_to_time_mapping=*/nullptr, &arena, /*prefix_extractor=*/nullptr, ts_sz) : mem->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena, - /*prefix_extractor=*/nullptr)); + /*prefix_extractor=*/nullptr, + /*for_flush=*/true)); ROCKS_LOG_DEBUG(immutable_db_options_.info_log, "[%s] [WriteLevel0TableForRecovery]" " Level-0 table #%" PRIu64 ": started", diff --git a/db/db_impl/db_impl_write.cc b/db/db_impl/db_impl_write.cc index 91a2f0f4c6..0d14c24a4b 100644 --- a/db/db_impl/db_impl_write.cc +++ b/db/db_impl/db_impl_write.cc @@ -12,6 +12,7 @@ #include "db/error_handler.h" #include "db/event_helpers.h" #include "logging/logging.h" +#include "memtable/wbwi_memtable.h" #include "monitoring/perf_context_imp.h" #include "options/options_helper.h" #include "test_util/sync_point.h" @@ -189,16 +190,137 @@ Status DBImpl::WriteWithCallback(const WriteOptions& write_options, return s; } -// The main write queue. This is the only write queue that updates LastSequence. -// When using one write queue, the same sequence also indicates the last -// published sequence. +Status DBImpl::IngestWBWI(std::shared_ptr wbwi, + const WBWIMemTable::SeqnoRange& assigned_seqno, + uint64_t prep_log, + SequenceNumber last_seqno_after_ingest, + bool memtable_updated, bool ignore_missing_cf) { + // Keys in new memtable have seqno > last_seqno_after_ingest >= keys in wbwi. + assert(assigned_seqno.upper_bound <= last_seqno_after_ingest); + // Keys in the current memtable have seqno <= LastSequence() < keys in wbwi. + assert(assigned_seqno.lower_bound > versions_->LastSequence()); + autovector memtables; + autovector cfds; + InstrumentedMutexLock lock(&mutex_); + ColumnFamilySet* cf_set = versions_->GetColumnFamilySet(); + + // Create WBWIMemTables + for (const auto [cf_id, stat] : wbwi->GetCFStats()) { + ColumnFamilyData* cfd = cf_set->GetColumnFamily(cf_id); + if (!cfd) { + if (ignore_missing_cf) { + continue; + } + for (auto mem : memtables) { + mem->Unref(); + delete mem; + } + for (auto cfd_ptr : cfds) { + cfd_ptr->UnrefAndTryDelete(); + } + Status s = Status::InvalidArgument( + "Invalid column family id from WriteBatchWithIndex: " + + std::to_string(cf_id)); + if (memtable_updated) { + s = Status::Corruption( + "Part of the write batch is applied. Memtable is in a inconsistent " + "state. " + + s.ToString()); + error_handler_.SetBGError(s, BackgroundErrorReason::kMemTable); + } + + return s; + } + WBWIMemTable* wbwi_memtable = + new WBWIMemTable(wbwi, cfd->user_comparator(), cf_id, cfd->ioptions(), + cfd->GetLatestMutableCFOptions(), stat); + wbwi_memtable->Ref(); + wbwi_memtable->AssignSequenceNumbers(assigned_seqno); + // This is needed to keep the WAL that contains Prepare alive until + // committed data in this memtable is persisted. + wbwi_memtable->SetMinPrepLog(prep_log); + memtables.push_back(wbwi_memtable); + cfd->Ref(); + cfds.push_back(cfd); + } + + // Stop writes to the DB by entering both write threads + WriteThread::Writer nonmem_w; + if (two_write_queues_) { + nonmem_write_thread_.EnterUnbatched(&nonmem_w, &mutex_); + } + WaitForPendingWrites(); + + // Switch memtable and add WBWIMemTables + Status s; + for (size_t i = 0; i < memtables.size(); ++i) { + assert(!immutable_db_options_.atomic_flush); + // NOTE: to support atomic flush, need to call + // SelectColumnFamiliesForAtomicFlush() + WriteContext write_context; + // TODO: not switch on empty memtable, may need to update metadata + // like NextLogNumber(), earliest_seqno and memtable id. + s = SwitchMemtable(cfds[i], &write_context, memtables[i], + last_seqno_after_ingest); + if (!s.ok()) { + // SwitchMemtable() can only fail if a new WAL is to be created, this + // should only happen for the first call to SwitchMemtable(). log will + // be empty and no new WAL is created for the rest of the calls. + assert(i == 0); + if (i != 0 || memtable_updated) { + // escalate error to non-recoverable + s = Status::Corruption( + "Part of the write batch is applied. Memtable is in a inconsistent " + "state. " + + s.ToString()); + error_handler_.SetBGError(s, BackgroundErrorReason::kMemTable); + } else { + // SwitchMemtable() already sets appropriate bg error + } + for (size_t j = i; j < memtables.size(); j++) { + memtables[j]->Unref(); + delete memtables[j]; + } + break; + } + } + for (size_t i = 0; i < cfds.size(); ++i) { + if (cfds[i]->UnrefAndTryDelete()) { + cfds[i] = nullptr; + } + } + + // exit the second queue before returning + if (two_write_queues_) { + nonmem_write_thread_.ExitUnbatched(&nonmem_w); + } + if (s.ok()) { + // Trigger flushes for the new immutable memtables. + for (const auto cfd : cfds) { + if (cfd == nullptr) { + continue; + } + cfd->imm()->FlushRequested(); + FlushRequest flush_req; + // TODO: a new flush reason for ingesting memtable + GenerateFlushRequest({cfd}, FlushReason::kExternalFileIngestion, + &flush_req); + EnqueuePendingFlush(flush_req); + } + MaybeScheduleFlushOrCompaction(); + } + return s; +} + Status DBImpl::WriteImpl(const WriteOptions& write_options, WriteBatch* my_batch, WriteCallback* callback, UserWriteCallback* user_write_cb, uint64_t* log_used, uint64_t log_ref, bool disable_memtable, uint64_t* seq_used, size_t batch_cnt, PreReleaseCallback* pre_release_callback, - PostMemTableCallback* post_memtable_callback) { + PostMemTableCallback* post_memtable_callback, + std::shared_ptr wbwi, + uint64_t prep_log) { assert(!seq_per_batch_ || batch_cnt != 0); assert(my_batch == nullptr || my_batch->Count() == 0 || write_options.protection_bytes_per_key == 0 || @@ -287,6 +409,23 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, return Status::NotSupported( "DeleteRange is not compatible with row cache."); } + if (wbwi) { + assert(prep_log > 0); + // Used only in WriteCommittedTxn::CommitInternal() with no `callback`. + assert(!callback); + if (immutable_db_options_.unordered_write) { + return Status::NotSupported( + "Ingesting WriteBatch does not support unordered_write"); + } + if (immutable_db_options_.enable_pipelined_write) { + return Status::NotSupported( + "Ingesting WriteBatch does not support pipelined_write"); + } + if (immutable_db_options_.atomic_flush) { + return Status::NotSupported( + "Ingesting WriteBatch does not support atomic_flush"); + } + } // Otherwise IsLatestPersistentState optimization does not make sense assert(!WriteBatchInternal::IsLatestPersistentState(my_batch) || disable_memtable); @@ -344,7 +483,8 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, PERF_TIMER_GUARD(write_pre_and_post_process_time); WriteThread::Writer w(write_options, my_batch, callback, user_write_cb, log_ref, disable_memtable, batch_cnt, - pre_release_callback, post_memtable_callback); + pre_release_callback, post_memtable_callback, + /*_ingest_wbwi=*/wbwi != nullptr); StopWatch write_sw(immutable_db_options_.clock, stats_, DB_WRITE); write_thread_.JoinBatchGroup(&w); @@ -441,6 +581,9 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, TEST_SYNC_POINT("DBImpl::WriteImpl:BeforeLeaderEnters"); last_batch_group_size_ = write_thread_.EnterAsBatchGroupLeader(&w, &write_group); + if (wbwi) { + assert(write_group.size == 1); + } IOStatus io_s; Status pre_release_cb_status; @@ -494,10 +637,25 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, // Note about seq_per_batch_: either disableWAL is set for the entire write // group or not. In either case we inc seq for each write batch with no // failed callback. This means that there could be a batch with - // disalbe_memtable in between; although we do not write this batch to + // disable_memtable in between; although we do not write this batch to // memtable it still consumes a seq. Otherwise, if !seq_per_batch_, we inc // the seq per valid written key to mem. size_t seq_inc = seq_per_batch_ ? valid_batches : total_count; + if (wbwi) { + // Reserve sequence numbers for the ingested memtable. We need to reserve + // at lease this amount for recovery. During recovery, + // transactions do not commit by ingesting WBWI. The sequence number + // associated with the commit entry in WAL is used as the starting + // sequence number for inserting into memtable. We need to reserve + // enough sequence numbers here (at least the number of operations + // in write batch) to assign to memtable entries for this transaction. + // This prevents updates in different transactions from using out-of-order + // sequence numbers or the same key+seqno. + // + // WBWI ingestion requires not grouping writes, so we don't need to + // consider incrementing sequence number for WBWI from other writers. + seq_inc += wbwi->GetWriteBatch()->Count(); + } const bool concurrent_update = two_write_queues_; // Update stats while we are an exclusive group leader, so we know @@ -674,6 +832,27 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options, // handle exit, false means somebody else did should_exit_batch_group = write_thread_.CompleteParallelMemTableWriter(&w); } + if (wbwi) { + if (status.ok() && w.status.ok()) { + // w.batch contains (potentially empty) commit time batch updates, + // only ingest wbwi if w.batch is applied to memtable successfully + assert(wbwi->GetWriteBatch()->Count() > 0); + + uint32_t memtable_update_count = w.batch->Count(); + SequenceNumber lb = versions_->LastSequence() + memtable_update_count + 1; + SequenceNumber ub = versions_->LastSequence() + memtable_update_count + + wbwi->GetWriteBatch()->Count(); + assert(ub == last_sequence); + if (two_write_queues_) { + assert(ub <= versions_->LastAllocatedSequence()); + } + status = IngestWBWI(wbwi, {/*lower_bound=*/lb, /*upper_bound=*/ub}, + prep_log, last_sequence, + /*memtable_updated=*/memtable_update_count > 0, + write_options.ignore_missing_column_families); + } + } + if (should_exit_batch_group) { if (status.ok()) { for (auto* tmp_w : write_group) { @@ -2204,16 +2383,13 @@ void DBImpl::NotifyOnMemTableSealed(ColumnFamilyData* /*cfd*/, mutex_.Lock(); } -// REQUIRES: mutex_ is held -// REQUIRES: this thread is currently at the front of the writer queue -// REQUIRES: this thread is currently at the front of the 2nd writer queue if -// two_write_queues_ is true (This is to simplify the reasoning.) -Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) { +Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context, + ReadOnlyMemTable* new_imm, + SequenceNumber last_seqno) { mutex_.AssertHeld(); assert(lock_wal_count_ == 0); // TODO: plumb Env::IOActivity, Env::IOPriority - const ReadOptions read_options; const WriteOptions write_options; log::Writer* new_log = nullptr; @@ -2249,6 +2425,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) { const MutableCFOptions mutable_cf_options = *cfd->GetLatestMutableCFOptions(); // Set memtable_info for memtable sealed callback + // TODO: memtable_info for `new_imm` MemTableInfo memtable_info; memtable_info.cf_name = cfd->GetName(); memtable_info.first_seqno = cfd->mem()->GetFirstSequenceNumber(); @@ -2276,8 +2453,20 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) { } } if (s.ok()) { - SequenceNumber seq = versions_->LastSequence(); - new_mem = cfd->ConstructNewMemtable(mutable_cf_options, seq); + // FIXME: from the comment for GetEarliestSequenceNumber(), any key with + // seqno >= earliest_seqno should be in this or later memtable. This means + // we should use LastSequence() + 1 or last_seqno + 1 here. And it needs to + // be incremented with file ingestion and other operations that consumes + // sequence number. + SequenceNumber seq; + if (new_imm) { + assert(last_seqno > versions_->LastSequence()); + seq = last_seqno; + } else { + seq = versions_->LastSequence(); + } + new_mem = + cfd->ConstructNewMemtable(mutable_cf_options, /*earliest_seq=*/seq); context->superversion_context.NewSuperVersion(); ROCKS_LOG_INFO(immutable_db_options_.info_log, @@ -2359,6 +2548,8 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) { versions_->PreComputeMinLogNumberWithUnflushedData(logfile_number_); if (min_wal_number_to_keep > versions_->GetWalSet().GetMinWalNumberToKeep()) { + // TODO: plumb Env::IOActivity, Env::IOPriority + const ReadOptions read_options; // Get a snapshot of the empty column families. // LogAndApply may release and reacquire db // mutex, during that period, column family may become empty (e.g. its @@ -2416,6 +2607,18 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) { cfd->mem()->SetNextLogNumber(logfile_number_); assert(new_mem != nullptr); cfd->imm()->Add(cfd->mem(), &context->memtables_to_free_); + if (new_imm) { + // Need to assign memtable id here before SetMemtable() below assigns id to + // the new live memtable + cfd->AssignMemtableID(new_imm); + // NOTE: new_imm and cfd->mem() references the same WAL and has the same + // NextLogNumber(). They should be flushed together. For non-atomic-flush, + // we always try to flush all immutable memtable. For atomic flush, these + // two memtables will be marked eligible for flush in the same call to + // AssignAtomicFlushSeq(). + new_imm->SetNextLogNumber(logfile_number_); + cfd->imm()->Add(new_imm, &context->memtables_to_free_); + } new_mem->Ref(); cfd->SetMemtable(new_mem); InstallSuperVersionAndScheduleWork(cfd, &context->superversion_context, @@ -2428,6 +2631,9 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) { // that is okay. If we did, it most likely means that s was already an error. // In any case, ignore any unchecked error for i_os here. io_s.PermitUncheckedError(); + // We guarantee that if a non-ok status is returned, `new_imm` was not added + // to the db. + assert(s.ok()); return s; } diff --git a/db/db_test_util.cc b/db/db_test_util.cc index 09d485a0b6..13957e2144 100644 --- a/db/db_test_util.cc +++ b/db/db_test_util.cc @@ -1600,42 +1600,74 @@ std::vector DBTestBase::ListTableFiles(Env* env, return file_numbers; } -void DBTestBase::VerifyDBFromMap(std::map true_data, - size_t* total_reads_res, bool tailing_iter, - std::map status) { - size_t total_reads = 0; +void DBTestBase::VerifyDBFromMap( + std::map true_data, size_t* total_reads_res, + bool tailing_iter, ReadOptions* ro, ColumnFamilyHandle* cf, + std::unordered_set* not_found) const { + ReadOptions temp_ro; + if (!ro) { + ro = &temp_ro; + ro->verify_checksums = true; + } + if (!cf) { + cf = db_->DefaultColumnFamily(); + } - for (auto& kv : true_data) { - Status s = status[kv.first]; - if (s.ok()) { - ASSERT_EQ(Get(kv.first), kv.second); - } else { - std::string value; - ASSERT_EQ(s, db_->Get(ReadOptions(), kv.first, &value)); - } + // Get + size_t total_reads = 0; + std::string result; + for (auto& [k, v] : true_data) { + ASSERT_OK(db_->Get(*ro, cf, k, &result)) << "key is " << k; + ASSERT_EQ(v, result); total_reads++; } + if (not_found) { + for (const auto& k : *not_found) { + ASSERT_TRUE(db_->Get(*ro, cf, k, &result).IsNotFound()) + << "key is " << k << " val is " << result; + } + } + + // MultiGet + std::vector key_slice; + for (const auto& [k, _] : true_data) { + key_slice.emplace_back(k); + } + std::vector values; + std::vector cfs(key_slice.size(), cf); + std::vector status = db_->MultiGet(*ro, cfs, key_slice, &values); + total_reads += key_slice.size(); + auto data_iter = true_data.begin(); + for (size_t i = 0; i < key_slice.size(); ++i, ++data_iter) { + ASSERT_OK(status[i]); + ASSERT_EQ(values[i], data_iter->second); + } + // MultiGet - not found + if (not_found) { + key_slice.clear(); + for (const auto& k : *not_found) { + key_slice.emplace_back(k); + } + cfs = std::vector(key_slice.size(), cf); + values.clear(); + status = db_->MultiGet(*ro, cfs, key_slice, &values); + for (const auto& s : status) { + ASSERT_TRUE(s.IsNotFound()); + } + } // Normal Iterator { int iter_cnt = 0; - ReadOptions ro; - ro.total_order_seek = true; - Iterator* iter = db_->NewIterator(ro); + ReadOptions ro_ = *ro; + ro_.total_order_seek = true; + Iterator* iter = db_->NewIterator(ro_, cf); // Verify Iterator::Next() iter_cnt = 0; - auto data_iter = true_data.begin(); - Status s; - for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) { + data_iter = true_data.begin(); + for (iter->SeekToFirst(); iter->Valid(); iter->Next(), ++data_iter) { ASSERT_EQ(iter->key().ToString(), data_iter->first); - Status current_status = status[data_iter->first]; - if (!current_status.ok()) { - s = current_status; - } - ASSERT_EQ(iter->status(), s); - if (current_status.ok()) { ASSERT_EQ(iter->value().ToString(), data_iter->second); - } iter_cnt++; total_reads++; } @@ -1646,20 +1678,12 @@ void DBTestBase::VerifyDBFromMap(std::map true_data, // Verify Iterator::Prev() // Use a new iterator to make sure its status is clean. - iter = db_->NewIterator(ro); + iter = db_->NewIterator(ro_, cf); iter_cnt = 0; - s = Status::OK(); auto data_rev = true_data.rbegin(); for (iter->SeekToLast(); iter->Valid(); iter->Prev(), data_rev++) { ASSERT_EQ(iter->key().ToString(), data_rev->first); - Status current_status = status[data_rev->first]; - if (!current_status.ok()) { - s = current_status; - } - ASSERT_EQ(iter->status(), s); - if (current_status.ok()) { ASSERT_EQ(iter->value().ToString(), data_rev->second); - } iter_cnt++; total_reads++; } @@ -1667,12 +1691,20 @@ void DBTestBase::VerifyDBFromMap(std::map true_data, ASSERT_EQ(data_rev, true_data.rend()) << iter_cnt << " / " << true_data.size(); - // Verify Iterator::Seek() - for (const auto& kv : true_data) { - iter->Seek(kv.first); - ASSERT_EQ(kv.first, iter->key().ToString()); - ASSERT_EQ(kv.second, iter->value().ToString()); - total_reads++; + // Verify Iterator::Seek() and SeekForPrev() + for (const auto& [k, v] : true_data) { + for (bool prev : {false, true}) { + if (prev) { + iter->SeekForPrev(k); + } else { + iter->Seek(k); + } + ASSERT_TRUE(iter->Valid()); + ASSERT_OK(iter->status()); + ASSERT_EQ(iter->key(), k); + ASSERT_EQ(iter->value(), v); + ++total_reads; + } } delete iter; } @@ -1680,14 +1712,14 @@ void DBTestBase::VerifyDBFromMap(std::map true_data, if (tailing_iter) { // Tailing iterator int iter_cnt = 0; - ReadOptions ro; - ro.tailing = true; - ro.total_order_seek = true; - Iterator* iter = db_->NewIterator(ro); + ReadOptions ro_ = *ro; + ro_.tailing = true; + ro_.total_order_seek = true; + Iterator* iter = db_->NewIterator(ro_, cf); // Verify ForwardIterator::Next() iter_cnt = 0; - auto data_iter = true_data.begin(); + data_iter = true_data.begin(); for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) { ASSERT_EQ(iter->key().ToString(), data_iter->first); ASSERT_EQ(iter->value().ToString(), data_iter->second); diff --git a/db/db_test_util.h b/db/db_test_util.h index 820c09c769..4b5bc48450 100644 --- a/db/db_test_util.h +++ b/db/db_test_util.h @@ -1367,7 +1367,8 @@ class DBTestBase : public testing::Test { void VerifyDBFromMap( std::map true_data, size_t* total_reads_res = nullptr, bool tailing_iter = false, - std::map status = std::map()); + ReadOptions* ro = nullptr, ColumnFamilyHandle* cf = nullptr, + std::unordered_set* not_found = nullptr) const; void VerifyDBInternal( std::vector> true_data); diff --git a/db/flush_job.cc b/db/flush_job.cc index 69ce103022..161afc06e0 100644 --- a/db/flush_job.cc +++ b/db/flush_job.cc @@ -422,7 +422,8 @@ Status FlushJob::MemPurge() { range_del_iters; for (ReadOnlyMemTable* m : mems_) { memtables.push_back(m->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, - &arena, /*prefix_extractor=*/nullptr)); + &arena, /*prefix_extractor=*/nullptr, + /*for_flush=*/true)); auto* range_del_iter = m->NewRangeTombstoneIterator( ro, kMaxSequenceNumber, true /* immutable_memtable */); if (range_del_iter != nullptr) { @@ -624,10 +625,13 @@ Status FlushJob::MemPurge() { // Construct fragmented memtable range tombstones without mutex new_mem->ConstructFragmentedRangeTombstones(); db_mutex_->Lock(); - uint64_t new_mem_id = mems_[0]->GetID(); + // Take the newest id, so that memtables in MemtableList don't have + // out-of-order memtable ids. + uint64_t new_mem_id = mems_.back()->GetID(); new_mem->SetID(new_mem_id); - new_mem->SetNextLogNumber(mems_[0]->GetNextLogNumber()); + // Take the latest memtable's next log number. + new_mem->SetNextLogNumber(mems_.back()->GetNextLogNumber()); // This addition will not trigger another flush, because // we do not call EnqueuePendingFlush(). @@ -907,7 +911,7 @@ Status FlushJob::WriteLevel0Table() { } else { memtables.push_back( m->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena, - /*prefix_extractor=*/nullptr)); + /*prefix_extractor=*/nullptr, /*for_flush=*/true)); } auto* range_del_iter = logical_strip_timestamp diff --git a/db/forward_iterator.cc b/db/forward_iterator.cc index f111e2fc28..6e7e1e6de5 100644 --- a/db/forward_iterator.cc +++ b/db/forward_iterator.cc @@ -714,7 +714,8 @@ void ForwardIterator::RebuildIterators(bool refresh_sv) { sv_->GetSeqnoToTimeMapping(); mutable_iter_ = sv_->mem->NewIterator(read_options_, seqno_to_time_mapping, &arena_, - sv_->mutable_cf_options.prefix_extractor.get()); + sv_->mutable_cf_options.prefix_extractor.get(), + /*for_flush=*/false); sv_->imm->AddIterators(read_options_, seqno_to_time_mapping, sv_->mutable_cf_options.prefix_extractor.get(), &imm_iters_, &arena_); @@ -784,7 +785,8 @@ void ForwardIterator::RenewIterators() { svnew->GetSeqnoToTimeMapping(); mutable_iter_ = svnew->mem->NewIterator(read_options_, seqno_to_time_mapping, &arena_, - svnew->mutable_cf_options.prefix_extractor.get()); + svnew->mutable_cf_options.prefix_extractor.get(), + /*for_flush=*/false); svnew->imm->AddIterators(read_options_, seqno_to_time_mapping, svnew->mutable_cf_options.prefix_extractor.get(), &imm_iters_, &arena_); diff --git a/db/memtable.cc b/db/memtable.cc index 1d3ee7eef4..d51a85350c 100644 --- a/db/memtable.cc +++ b/db/memtable.cc @@ -599,7 +599,7 @@ class MemTableIterator : public InternalIterator { InternalIterator* MemTable::NewIterator( const ReadOptions& read_options, UnownedPtr seqno_to_time_mapping, Arena* arena, - const SliceTransform* prefix_extractor) { + const SliceTransform* prefix_extractor, bool /*for_flush*/) { assert(arena != nullptr); auto mem = arena->AllocateAligned(sizeof(MemTableIterator)); return new (mem) diff --git a/db/memtable.h b/db/memtable.h index 897a3d079d..f53c614fd7 100644 --- a/db/memtable.h +++ b/db/memtable.h @@ -153,7 +153,7 @@ class ReadOnlyMemTable { virtual InternalIterator* NewIterator( const ReadOptions& read_options, UnownedPtr seqno_to_time_mapping, Arena* arena, - const SliceTransform* prefix_extractor) = 0; + const SliceTransform* prefix_extractor, bool for_flush) = 0; // Returns an iterator that wraps a MemTableIterator and logically strips the // user-defined timestamp of each key. This API is only used by flush when @@ -541,7 +541,7 @@ class MemTable final : public ReadOnlyMemTable { InternalIterator* NewIterator( const ReadOptions& read_options, UnownedPtr seqno_to_time_mapping, Arena* arena, - const SliceTransform* prefix_extractor) override; + const SliceTransform* prefix_extractor, bool for_flush) override; InternalIterator* NewTimestampStrippingIterator( const ReadOptions& read_options, diff --git a/db/memtable_list.cc b/db/memtable_list.cc index 05b085f878..af1183a7ef 100644 --- a/db/memtable_list.cc +++ b/db/memtable_list.cc @@ -32,6 +32,10 @@ class Mutex; class VersionSet; void MemTableListVersion::AddMemTable(ReadOnlyMemTable* m) { + if (!memlist_.empty()) { + // ID can be equal for MemPurge + assert(m->GetID() >= memlist_.front()->GetID()); + } memlist_.push_front(m); *parent_memtable_list_memory_usage_ += m->ApproximateMemoryUsage(); } @@ -216,7 +220,8 @@ void MemTableListVersion::AddIterators( std::vector* iterator_list, Arena* arena) { for (auto& m : memlist_) { iterator_list->push_back(m->NewIterator(options, seqno_to_time_mapping, - arena, prefix_extractor)); + arena, prefix_extractor, + /*for_flush=*/false)); } } @@ -228,7 +233,8 @@ void MemTableListVersion::AddIterators( for (auto& m : memlist_) { auto mem_iter = m->NewIterator(options, seqno_to_time_mapping, - merge_iter_builder->GetArena(), prefix_extractor); + merge_iter_builder->GetArena(), prefix_extractor, + /*for_flush=*/false); if (!add_range_tombstone_iter || options.ignore_range_deletions) { merge_iter_builder->AddIterator(mem_iter); } else { @@ -410,7 +416,8 @@ void MemTableList::PickMemtablesToFlush(uint64_t max_memtable_id, // ret is filled with memtables already sorted in increasing MemTable ID. // However, when the mempurge feature is activated, new memtables with older // IDs will be added to the memlist. - for (auto it = memlist.rbegin(); it != memlist.rend(); ++it) { + auto it = memlist.rbegin(); + for (; it != memlist.rend(); ++it) { ReadOnlyMemTable* m = *it; if (!atomic_flush && m->atomic_flush_seqno_ != kMaxSequenceNumber) { atomic_flush = true; @@ -439,6 +446,12 @@ void MemTableList::PickMemtablesToFlush(uint64_t max_memtable_id, break; } } + if (!ret->empty() && it != memlist.rend()) { + // checks that the first memtable not picked to flush is not ingested wbwi. + // Ingested memtable should be flushed together with the memtable before it + // since they map to the same WAL and have the same NextLogNumber(). + assert(strcmp((*it)->Name(), "WBWIMemTable") != 0); + } if (!atomic_flush || num_flush_not_started_ == 0) { flush_requested_ = false; // start-flush request is complete } diff --git a/db/memtable_list_test.cc b/db/memtable_list_test.cc index 812c176d27..2de350ae8d 100644 --- a/db/memtable_list_test.cc +++ b/db/memtable_list_test.cc @@ -330,6 +330,7 @@ TEST_F(MemTableListTest, GetTest) { // This is to make assert(memtable->IsFragmentedRangeTombstonesConstructed()) // in MemTableListVersion::GetFromList work. mem->ConstructFragmentedRangeTombstones(); + mem->SetID(1); list.Add(mem, &to_delete); SequenceNumber saved_seq = seq; @@ -338,6 +339,7 @@ TEST_F(MemTableListTest, GetTest) { WriteBufferManager wb2(options.db_write_buffer_size); MemTable* mem2 = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb2, kMaxSequenceNumber, 0 /* column_family_id */); + mem2->SetID(2); mem2->Ref(); ASSERT_OK( diff --git a/db/repair.cc b/db/repair.cc index 8c0f36d991..3b2996eef4 100644 --- a/db/repair.cc +++ b/db/repair.cc @@ -449,7 +449,7 @@ class Repairer { Arena arena; ScopedArenaPtr iter( mem->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena, - /*prefix_extractor=*/nullptr)); + /*prefix_extractor=*/nullptr, /*for_flush=*/true)); int64_t _current_time = 0; immutable_db_options_.clock->GetCurrentTime(&_current_time) .PermitUncheckedError(); // ignore error diff --git a/db/write_batch_test.cc b/db/write_batch_test.cc index d09dd0167d..e1fc9f41cd 100644 --- a/db/write_batch_test.cc +++ b/db/write_batch_test.cc @@ -59,7 +59,8 @@ static std::string PrintContents(WriteBatch* b, InternalIterator* iter; if (i == 0) { iter = mem->NewIterator(ReadOptions(), /*seqno_to_time_mapping=*/nullptr, - &arena, /*prefix_extractor=*/nullptr); + &arena, /*prefix_extractor=*/nullptr, + /*for_flush=*/false); arena_iter_guard.reset(iter); } else { iter = mem->NewRangeTombstoneIterator(ReadOptions(), diff --git a/db/write_thread.cc b/db/write_thread.cc index 87c8262724..bc4cc3c380 100644 --- a/db/write_thread.cc +++ b/db/write_thread.cc @@ -523,8 +523,10 @@ size_t WriteThread::EnterAsBatchGroupLeader(Writer* leader, // those are something else. They want to be alone (w->callback != nullptr && !w->callback->AllowWriteBatching()) || // dont batch writes that don't want to be batched - (size + WriteBatchInternal::ByteSize(w->batch) > max_size) + (size + WriteBatchInternal::ByteSize(w->batch) > max_size) || // Do not make batch too big + (leader->ingest_wbwi || w->ingest_wbwi) + // ingesting WBWI needs to be its own group ) { // remove from list w->link_older->link_newer = w->link_newer; diff --git a/db/write_thread.h b/db/write_thread.h index 16af946b12..42256970f4 100644 --- a/db/write_thread.h +++ b/db/write_thread.h @@ -148,6 +148,8 @@ class WriteThread { Writer* link_older; // read/write only before linking, or as leader Writer* link_newer; // lazy, read/write only before linking, or as leader + bool ingest_wbwi; + Writer() : batch(nullptr), sync(false), @@ -174,7 +176,8 @@ class WriteThread { WriteCallback* _callback, UserWriteCallback* _user_write_cb, uint64_t _log_ref, bool _disable_memtable, size_t _batch_cnt = 0, PreReleaseCallback* _pre_release_callback = nullptr, - PostMemTableCallback* _post_memtable_callback = nullptr) + PostMemTableCallback* _post_memtable_callback = nullptr, + bool _ingest_wbwi = false) : batch(_batch), // TODO: store a copy of WriteOptions instead of its seperated data // members @@ -196,7 +199,8 @@ class WriteThread { write_group(nullptr), sequence(kMaxSequenceNumber), link_older(nullptr), - link_newer(nullptr) {} + link_newer(nullptr), + ingest_wbwi(_ingest_wbwi) {} ~Writer() { if (made_waitable) { diff --git a/db_stress_tool/db_stress_common.h b/db_stress_tool/db_stress_common.h index d2eb620ed1..62c79ac0e5 100644 --- a/db_stress_tool/db_stress_common.h +++ b/db_stress_tool/db_stress_common.h @@ -420,6 +420,7 @@ DECLARE_uint32(uncache_aggressiveness); DECLARE_int32(test_ingest_standalone_range_deletion_one_in); DECLARE_bool(allow_unprepared_value); DECLARE_string(file_temperature_age_thresholds); +DECLARE_uint32(commit_bypass_memtable_one_in); constexpr long KB = 1024; constexpr int kRandomValueMaxFactor = 3; diff --git a/db_stress_tool/db_stress_gflags.cc b/db_stress_tool/db_stress_gflags.cc index ac7422c8fc..c49846b9f1 100644 --- a/db_stress_tool/db_stress_gflags.cc +++ b/db_stress_tool/db_stress_gflags.cc @@ -1469,4 +1469,7 @@ DEFINE_bool(paranoid_memory_checks, ROCKSDB_NAMESPACE::Options().paranoid_memory_checks, "Sets CF option paranoid_memory_checks."); +DEFINE_uint32(commit_bypass_memtable_one_in, 0, + "If greater than zero, transaction option will set " + "commit_bypass_memtable to per every N transactions on average."); #endif // GFLAGS diff --git a/db_stress_tool/db_stress_test_base.cc b/db_stress_tool/db_stress_test_base.cc index e5b7f344b7..1d6c7de35c 100644 --- a/db_stress_tool/db_stress_test_base.cc +++ b/db_stress_tool/db_stress_test_base.cc @@ -805,7 +805,7 @@ void StressTest::ProcessRecoveredPreparedTxnsHelper(Transaction* txn, } } -Status StressTest::NewTxn(WriteOptions& write_opts, +Status StressTest::NewTxn(WriteOptions& write_opts, ThreadState* thread, std::unique_ptr* out_txn) { if (!FLAGS_use_txn) { return Status::InvalidArgument("NewTxn when FLAGS_use_txn is not set"); @@ -821,6 +821,12 @@ Status StressTest::NewTxn(WriteOptions& write_opts, FLAGS_use_only_the_last_commit_time_batch_for_recovery; txn_options.lock_timeout = 600000; // 10 min txn_options.deadlock_detect = true; + if (FLAGS_commit_bypass_memtable_one_in > 0) { + assert(FLAGS_txn_write_policy == 0); + assert(FLAGS_user_timestamp_size == 0); + txn_options.commit_bypass_memtable = + thread->rand.OneIn(FLAGS_commit_bypass_memtable_one_in); + } out_txn->reset(txn_db_->BeginTransaction(write_opts, txn_options)); auto istr = std::to_string(txn_id.fetch_add(1)); Status s = (*out_txn)->SetName("xid" + istr); @@ -880,7 +886,7 @@ Status StressTest::ExecuteTransaction( WriteOptions& write_opts, ThreadState* thread, std::function&& ops) { std::unique_ptr txn; - Status s = NewTxn(write_opts, &txn); + Status s = NewTxn(write_opts, thread, &txn); std::string try_again_messages; if (s.ok()) { for (int tries = 1;; ++tries) { diff --git a/db_stress_tool/db_stress_test_base.h b/db_stress_tool/db_stress_test_base.h index cf6f174b8d..1b338e8940 100644 --- a/db_stress_tool/db_stress_test_base.h +++ b/db_stress_tool/db_stress_test_base.h @@ -138,7 +138,7 @@ class StressTest { SharedState* shared); // ExecuteTransaction is recommended instead - Status NewTxn(WriteOptions& write_opts, + Status NewTxn(WriteOptions& write_opts, ThreadState* thread, std::unique_ptr* out_txn); Status CommitTxn(Transaction& txn, ThreadState* thread = nullptr); diff --git a/db_stress_tool/multi_ops_txns_stress.cc b/db_stress_tool/multi_ops_txns_stress.cc index 574552c52e..6ec42e4e7f 100644 --- a/db_stress_tool/multi_ops_txns_stress.cc +++ b/db_stress_tool/multi_ops_txns_stress.cc @@ -544,6 +544,11 @@ Status MultiOpsTxnsStressTest::TestCustomOperations( // Should never reach here. assert(false); } + if (!s.ok()) { + fprintf(stderr, "Transaction failed %s\n", s.ToString().c_str()); + fflush(stderr); + thread->shared->SafeTerminate(); + } return s; } @@ -579,7 +584,7 @@ Status MultiOpsTxnsStressTest::PrimaryKeyUpdateTxn(ThreadState* thread, std::string new_pk = Record::EncodePrimaryKey(new_a); std::unique_ptr txn; WriteOptions wopts; - Status s = NewTxn(wopts, &txn); + Status s = NewTxn(wopts, thread, &txn); if (!s.ok()) { assert(!txn); thread->stats.AddErrors(1); @@ -611,7 +616,12 @@ Status MultiOpsTxnsStressTest::PrimaryKeyUpdateTxn(ThreadState* thread, } auto& key_gen = key_gen_for_a_[thread->tid]; key_gen->UndoAllocation(new_a); - txn->Rollback().PermitUncheckedError(); + s = txn->Rollback(); + if (!s.ok()) { + fprintf(stderr, "Transaction rollback failed %s\n", s.ToString().c_str()); + fflush(stderr); + assert(false); + } }); ReadOptions ropts; @@ -699,7 +709,7 @@ Status MultiOpsTxnsStressTest::SecondaryKeyUpdateTxn(ThreadState* thread, uint32_t new_c) { std::unique_ptr txn; WriteOptions wopts; - Status s = NewTxn(wopts, &txn); + Status s = NewTxn(wopts, thread, &txn); if (!s.ok()) { assert(!txn); thread->stats.AddErrors(1); @@ -735,7 +745,12 @@ Status MultiOpsTxnsStressTest::SecondaryKeyUpdateTxn(ThreadState* thread, } auto& key_gen = key_gen_for_c_[thread->tid]; key_gen->UndoAllocation(new_c); - txn->Rollback().PermitUncheckedError(); + s = txn->Rollback(); + if (!s.ok()) { + fprintf(stderr, "Transaction rollback failed %s\n", s.ToString().c_str()); + fflush(stderr); + assert(false); + } }); // TODO (yanqin) try SetSnapshotOnNextOperation(). We currently need to take @@ -901,7 +916,7 @@ Status MultiOpsTxnsStressTest::UpdatePrimaryIndexValueTxn(ThreadState* thread, std::string pk_str = Record::EncodePrimaryKey(a); std::unique_ptr txn; WriteOptions wopts; - Status s = NewTxn(wopts, &txn); + Status s = NewTxn(wopts, thread, &txn); if (!s.ok()) { assert(!txn); thread->stats.AddErrors(1); @@ -927,7 +942,12 @@ Status MultiOpsTxnsStressTest::UpdatePrimaryIndexValueTxn(ThreadState* thread, } else { thread->stats.AddErrors(1); } - txn->Rollback().PermitUncheckedError(); + s = txn->Rollback(); + if (!s.ok()) { + fprintf(stderr, "Transaction rollback failed %s\n", s.ToString().c_str()); + fflush(stderr); + assert(false); + } }); ReadOptions ropts; ropts.rate_limiter_priority = @@ -982,7 +1002,7 @@ Status MultiOpsTxnsStressTest::PointLookupTxn(ThreadState* thread, std::unique_ptr txn; WriteOptions wopts; - Status s = NewTxn(wopts, &txn); + Status s = NewTxn(wopts, thread, &txn); if (!s.ok()) { assert(!txn); thread->stats.AddErrors(1); @@ -1026,7 +1046,7 @@ Status MultiOpsTxnsStressTest::RangeScanTxn(ThreadState* thread, std::unique_ptr txn; WriteOptions wopts; - Status s = NewTxn(wopts, &txn); + Status s = NewTxn(wopts, thread, &txn); if (!s.ok()) { assert(!txn); thread->stats.AddErrors(1); @@ -1384,6 +1404,12 @@ Status MultiOpsTxnsStressTest::CommitAndCreateTimestampedSnapshotIfNeeded( } else { s = txn.Commit(); } + if (!s.ok()) { + fprintf(stderr, "Txn %s commit failed with %s\n", txn.GetName().c_str(), + s.ToString().c_str()); + fflush(stderr); + } + assert(txn_db_); if (FLAGS_create_timestamped_snapshot_one_in > 0 && thread->rand.OneInOpt(50000)) { diff --git a/db_stress_tool/no_batched_ops_stress.cc b/db_stress_tool/no_batched_ops_stress.cc index cd14bac68a..3b5341392e 100644 --- a/db_stress_tool/no_batched_ops_stress.cc +++ b/db_stress_tool/no_batched_ops_stress.cc @@ -666,7 +666,7 @@ class NonBatchedOpsStressTest : public StressTest { if (FLAGS_rate_limit_auto_wal_flush) { wo.rate_limiter_priority = Env::IO_USER; } - Status s = NewTxn(wo, &txn); + Status s = NewTxn(wo, thread, &txn); if (!s.ok()) { fprintf(stderr, "NewTxn error: %s\n", s.ToString().c_str()); shared->SafeTerminate(); @@ -1127,7 +1127,7 @@ class NonBatchedOpsStressTest : public StressTest { write_options.rate_limiter_priority = Env::IO_USER; } - const Status s = NewTxn(write_options, &txn); + const Status s = NewTxn(write_options, thread, &txn); if (!s.ok()) { fprintf(stderr, "NewTxn error: %s\n", s.ToString().c_str()); thread->shared->SafeTerminate(); diff --git a/include/rocksdb/utilities/transaction.h b/include/rocksdb/utilities/transaction.h index 6c444ac26d..e2b1a5d007 100644 --- a/include/rocksdb/utilities/transaction.h +++ b/include/rocksdb/utilities/transaction.h @@ -750,6 +750,8 @@ class Transaction { virtual TxnTimestamp GetCommitTimestamp() const { return kMaxTxnTimestamp; } + virtual bool GetCommitBypassMemTable() const { return false; } + protected: explicit Transaction(const TransactionDB* /*db*/) {} Transaction() : log_number_(0), txn_state_(STARTED) {} diff --git a/include/rocksdb/utilities/transaction_db.h b/include/rocksdb/utilities/transaction_db.h index 78ff0816ff..c09e4e8267 100644 --- a/include/rocksdb/utilities/transaction_db.h +++ b/include/rocksdb/utilities/transaction_db.h @@ -339,6 +339,23 @@ struct TransactionOptions { // size in APIs that MyRocks currently are using, including Put, Merge, Delete // DeleteRange, SingleDelete. bool write_batch_track_timestamp_size = false; + + // EXPERIMENTAL + // Only supports write-committed policy. If set to true, the transaction will + // skip memtable write and ingest into the DB directly during Commit(). This + // makes Commit() much faster for transactions with many operations. + // Transactions with Merge() or PutEntity() is not supported yet. + // + // Note that the transaction will be ingested as an immutable memtable for + // CFs it updates, and the current memtable will be switched to a new one. + // So ingesting many transactions in a short period of time may cause stall + // due to too many memtables. + // Note that the ingestion relies on the transaction's underlying index, + // (WriteBatchWithIndex), so updates that are added to the transaction + // without indexing (e.g. added directly to the transaction underlying + // write batch through Transaction::GetWriteBatch()->GetWriteBatch()) + // are not supported. They will not be applied to the DB. + bool commit_bypass_memtable = false; }; // The per-write optimizations that do not involve transactions. TransactionDB diff --git a/include/rocksdb/utilities/write_batch_with_index.h b/include/rocksdb/utilities/write_batch_with_index.h index 59bb4228df..7bb0653274 100644 --- a/include/rocksdb/utilities/write_batch_with_index.h +++ b/include/rocksdb/utilities/write_batch_with_index.h @@ -74,6 +74,10 @@ class WBWIIterator { // WriteBatchWithIndex virtual WriteEntry Entry() const = 0; + // For this user key, there is a single delete in this write batch, + // and it was overwritten by another update. + virtual bool HasOverWrittenSingleDel() const { return false; } + virtual Status status() const = 0; }; @@ -350,6 +354,17 @@ class WriteBatchWithIndex : public WriteBatchBase { void SetMaxBytes(size_t max_bytes) override; size_t GetDataSize() const; + struct CFStat { + uint32_t entry_count = 0; + uint32_t overwritten_sd_count = 0; + }; + // Will track CF ID, per CF entry count and overwritten sd count. + // Should be enabled when WBWI is empty for correct tracking. + void SetTrackPerCFStat(bool track); + const std::unordered_map& GetCFStats() const; + + bool GetOverwriteKey() const; + private: friend class PessimisticTransactionDB; friend class WritePreparedTxn; diff --git a/java/rocksjni/write_batch_test.cc b/java/rocksjni/write_batch_test.cc index ed456acc3e..fc45daaab4 100644 --- a/java/rocksjni/write_batch_test.cc +++ b/java/rocksjni/write_batch_test.cc @@ -61,7 +61,7 @@ jbyteArray Java_org_rocksdb_WriteBatchTest_getContents(JNIEnv* env, ROCKSDB_NAMESPACE::ScopedArenaPtr iter( mem->NewIterator(ROCKSDB_NAMESPACE::ReadOptions(), /*seqno_to_time_mapping=*/nullptr, &arena, - /*prefix_extractor=*/nullptr)); + /*prefix_extractor=*/nullptr, /*for_flush=*/false)); for (iter->SeekToFirst(); iter->Valid(); iter->Next()) { ROCKSDB_NAMESPACE::ParsedInternalKey ikey; ikey.clear(); diff --git a/memtable/wbwi_memtable.cc b/memtable/wbwi_memtable.cc index dc06c1736c..9ebd1b8d78 100644 --- a/memtable/wbwi_memtable.cc +++ b/memtable/wbwi_memtable.cc @@ -21,6 +21,27 @@ const std::unordered_map // kLogDataRecord, kXIDRecord, kUnknownRecord }; +InternalIterator* WBWIMemTable::NewIterator( + const ReadOptions&, UnownedPtr, Arena* arena, + const SliceTransform* /* prefix_extractor */, bool for_flush) { + // Ingested WBWIMemTable should have an assigned seqno + assert(assigned_seqno_.upper_bound != kMaxSequenceNumber); + assert(assigned_seqno_.lower_bound != kMaxSequenceNumber); + assert(arena); + auto mem = arena->AllocateAligned(sizeof(WBWIMemTableIterator)); + return new (mem) WBWIMemTableIterator( + std::unique_ptr(wbwi_->NewIterator(cf_id_)), + assigned_seqno_, comparator_, for_flush); +} + +inline InternalIterator* WBWIMemTable::NewIterator() const { + assert(assigned_seqno_.upper_bound != kMaxSequenceNumber); + assert(assigned_seqno_.lower_bound != kMaxSequenceNumber); + return new WBWIMemTableIterator( + std::unique_ptr(wbwi_->NewIterator(cf_id_)), + assigned_seqno_, comparator_, /*for_flush=*/false); +} + bool WBWIMemTable::Get(const LookupKey& key, std::string* value, PinnableWideColumns* columns, std::string* timestamp, Status* s, MergeContext* merge_context, @@ -34,7 +55,8 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value, assert(immutable_memtable); assert(!timestamp); // TODO: support UDT assert(!columns); // TODO: support WideColumn - assert(global_seqno_ != kMaxSequenceNumber); + assert(assigned_seqno_.upper_bound != kMaxSequenceNumber); + assert(assigned_seqno_.lower_bound != kMaxSequenceNumber); // WBWI does not support DeleteRange yet. assert(!wbwi_->GetWriteBatch()->HasDeleteRange()); diff --git a/memtable/wbwi_memtable.h b/memtable/wbwi_memtable.h index 088f4931ba..fb3a9cd228 100644 --- a/memtable/wbwi_memtable.h +++ b/memtable/wbwi_memtable.h @@ -8,157 +8,56 @@ #include "rocksdb/utilities/write_batch_with_index.h" namespace ROCKSDB_NAMESPACE { - -class WBWIMemTableIterator final : public InternalIterator { - public: - WBWIMemTableIterator(std::unique_ptr&& it, SequenceNumber seqno, - const Comparator* comparator) - : it_(std::move(it)), global_seqno_(seqno), comparator_(comparator) { - assert(seqno != kMaxSequenceNumber); - s_.PermitUncheckedError(); - } - - // No copying allowed - WBWIMemTableIterator(const WBWIMemTableIterator&) = delete; - WBWIMemTableIterator& operator=(const WBWIMemTableIterator&) = delete; - - bool Valid() const override { return valid_; } - - void SeekToFirst() override { - it_->SeekToFirst(); - UpdateKey(); - } - - void SeekToLast() override { - it_->SeekToLast(); - UpdateKey(); - } - - void Seek(const Slice& target) override { - Slice target_user_key = ExtractUserKey(target); - it_->Seek(target_user_key); - if (it_->Valid()) { - // compare seqno - SequenceNumber seqno = GetInternalKeySeqno(target); - if (seqno < global_seqno_ && - comparator_->Compare(it_->Entry().key, target_user_key) == 0) { - it_->Next(); - // TODO: cannot assume distinct keys once Merge is supported - if (it_->Valid()) { - assert(comparator_->Compare(it_->Entry().key, target_user_key) > 0); - } - } - } - UpdateKey(); - } - - void SeekForPrev(const Slice& target) override { - Slice target_user_key = ExtractUserKey(target); - it_->SeekForPrev(target_user_key); - if (it_->Valid()) { - SequenceNumber seqno = GetInternalKeySeqno(target); - if (seqno > global_seqno_ && - comparator_->Compare(it_->Entry().key, target_user_key) == 0) { - it_->Prev(); - if (it_->Valid()) { - // TODO: cannot assume distinct keys once Merge is supported - assert(comparator_->Compare(it_->Entry().key, target_user_key) < 0); - } - } - } - UpdateKey(); - } - - void Next() override { - assert(Valid()); - it_->Next(); - UpdateKey(); - } - - bool NextAndGetResult(IterateResult* result) override { - assert(Valid()); - Next(); - bool is_valid = Valid(); - if (is_valid) { - result->key = key(); - result->bound_check_result = IterBoundCheck::kUnknown; - result->value_prepared = true; - } - return is_valid; - } - - void Prev() override { - assert(Valid()); - it_->Prev(); - UpdateKey(); - } - - Slice key() const override { - assert(Valid()); - return key_; - } - - Slice value() const override { - assert(Valid()); - return it_->Entry().value; - } - - Status status() const override { - assert(it_->status().ok()); - return s_; - } - - private: - static const std::unordered_map WriteTypeToValueTypeMap; - - void UpdateKey() { - valid_ = it_->Valid(); - if (!Valid()) { - key_.clear(); - return; - } - auto t = WriteTypeToValueTypeMap.find(it_->Entry().type); - assert(t != WriteTypeToValueTypeMap.end()); - if (t == WriteTypeToValueTypeMap.end()) { - key_.clear(); - valid_ = false; - s_ = Status::Corruption("Unexpected write_batch_with_index entry type " + - std::to_string(it_->Entry().type)); - return; - } - key_buf_.SetInternalKey(it_->Entry().key, global_seqno_, t->second); - key_ = key_buf_.GetInternalKey(); - } - - std::unique_ptr it_; - // The sequence number of entries in this write batch. - SequenceNumber global_seqno_; - const Comparator* comparator_; - IterKey key_buf_; - // The current internal key. - Slice key_; - Status s_; - bool valid_ = false; -}; - +// An implementation of the ReadOnlyMemTable interface based on the content +// of the given write batch with index (WBWI) object. This can be used to ingest +// a transaction (which is based on WBWI) into the DB as an immutable memtable. +// +// REQUIRE overwrite_key to be true for the WBWI +// Since the keys in WBWI do not have sequence number, the memtable needs to be +// assigned a range of sequence numbers through AssignSequenceNumbers(seqno) +// before being available for reads. +// With overwrite_key = true, WBWI keeps track of the most recent update for +// each key, and each such key will be assigned seqno.upper_bound during reads. +// One exception is during flush, consider the following scenario: +// - WBWI has SD(k) then PUT(k, v1) +// - DB has PUT(k, v2) in L1 +// - flush WBWI adds PUT(k, v1) into L0 +// - live memtable contains SD(k) +// - flush live memtable and compact it with L0 will drop SD(k) and PUT(k, v1) +// - the PUT(k, v2) in L1 incorrectly becomes visible +// So during flush, iterator from this memtable will need emit overwritten +// single deletion. These single deletion entries will be +// assigned seqno.upper_bound - 1. class WBWIMemTable final : public ReadOnlyMemTable { public: + struct SeqnoRange { + SequenceNumber lower_bound = kMaxSequenceNumber; + SequenceNumber upper_bound = kMaxSequenceNumber; + }; WBWIMemTable(const std::shared_ptr& wbwi, const Comparator* cmp, uint32_t cf_id, const ImmutableOptions* immutable_options, - const MutableCFOptions* cf_options) + const MutableCFOptions* cf_options, + const WriteBatchWithIndex::CFStat& stat) : wbwi_(wbwi), comparator_(cmp), ikey_comparator_(comparator_), moptions_(*immutable_options, *cf_options), clock_(immutable_options->clock), - cf_id_(cf_id) {} + // We need to include overwritten_sd_count in num_entries_ since flush + // verifies number of entries processed and that iterator for this + // memtable will emit overwritten SingleDelete entries during flush, See + // comment above WBWIMemTableIterator for more detail. + num_entries_(stat.entry_count + stat.overwritten_sd_count), + cf_id_(cf_id) { + assert(wbwi->GetOverwriteKey()); + } // No copying allowed WBWIMemTable(const WBWIMemTable&) = delete; WBWIMemTable& operator=(const WBWIMemTable&) = delete; - ~WBWIMemTable() override = default; + ~WBWIMemTable() override { assert(refs_ == 0); } const char* Name() const override { return "WBWIMemTable"; } @@ -182,17 +81,11 @@ class WBWIMemTable final : public ReadOnlyMemTable { assert(false); } - InternalIterator* NewIterator( - const ReadOptions&, UnownedPtr, Arena* arena, - const SliceTransform* /* prefix_extractor */) override { - // Ingested WBWIMemTable should have an assigned seqno - assert(global_seqno_ != kMaxSequenceNumber); - assert(arena); - auto mem = arena->AllocateAligned(sizeof(WBWIMemTableIterator)); - return new (mem) WBWIMemTableIterator( - std::unique_ptr(wbwi_->NewIterator(cf_id_)), - global_seqno_, comparator_); - } + InternalIterator* NewIterator(const ReadOptions&, + UnownedPtr, + Arena* arena, + const SliceTransform* /* prefix_extractor */, + bool for_flush) override; // Returns an iterator that wraps a MemTableIterator and logically strips the // user-defined timestamp of each key. This API is only used by flush when @@ -235,13 +128,7 @@ class WBWIMemTable final : public ReadOnlyMemTable { void MultiGet(const ReadOptions& read_options, MultiGetRange* range, ReadCallback* callback, bool immutable_memtable) override; - uint64_t NumEntries() const override { - // FIXME: used in - // - verify number of entries processed during flush - // - stats for estimate num entries and num entries in immutable memtables - // - MemPurgeDecider - return 0; - } + uint64_t NumEntries() const override { return num_entries_; } uint64_t NumDeletion() const override { // FIXME: this is used for stats and event logging @@ -259,7 +146,9 @@ class WBWIMemTable final : public ReadOnlyMemTable { return 0; } - SequenceNumber GetEarliestSequenceNumber() override { return global_seqno_; } + SequenceNumber GetEarliestSequenceNumber() override { + return assigned_seqno_.lower_bound; + } bool IsEmpty() const override { // Ideally also check that wbwi contains updates from this CF. For now, we @@ -267,7 +156,9 @@ class WBWIMemTable final : public ReadOnlyMemTable { return wbwi_->GetWriteBatch()->Count() == 0; } - SequenceNumber GetFirstSequenceNumber() override { return global_seqno_; } + SequenceNumber GetFirstSequenceNumber() override { + return assigned_seqno_.lower_bound; + } uint64_t GetMinLogContainingPrepSection() override { // FIXME: used to retain WAL with pending Prepare @@ -304,31 +195,206 @@ class WBWIMemTable final : public ReadOnlyMemTable { } // Assign a sequence number to the entries in this memtable. - void SetGlobalSequenceNumber(SequenceNumber global_seqno) { + void AssignSequenceNumbers(const SeqnoRange& seqno_range) { // Not expecting to assign seqno multiple times. - assert(global_seqno_ == kMaxSequenceNumber); - global_seqno_ = global_seqno; + assert(assigned_seqno_.lower_bound == kMaxSequenceNumber); + assert(assigned_seqno_.upper_bound == kMaxSequenceNumber); + + assigned_seqno_ = seqno_range; + + assert(assigned_seqno_.lower_bound <= assigned_seqno_.upper_bound); + assert(assigned_seqno_.upper_bound != kMaxSequenceNumber); + } + + void SetMinPrepLog(uint64_t min_prep_log) { + min_prep_log_referenced_ = min_prep_log; } private: - InternalIterator* NewIterator() const { - assert(global_seqno_ != kMaxSequenceNumber); - return new WBWIMemTableIterator( - std::unique_ptr(wbwi_->NewIterator(cf_id_)), - global_seqno_, comparator_); - } + inline InternalIterator* NewIterator() const; Slice newest_udt_; std::shared_ptr wbwi_; const Comparator* comparator_; InternalKeyComparator ikey_comparator_; - SequenceNumber global_seqno_ = kMaxSequenceNumber; + SeqnoRange assigned_seqno_; const ImmutableMemTableOptions moptions_; SystemClock* clock_; uint64_t min_prep_log_referenced_{0}; + uint64_t num_entries_; // WBWI can contains updates to multiple CFs. `cf_id_` determines which CF // this memtable is for. uint32_t cf_id_; }; +class WBWIMemTableIterator final : public InternalIterator { + public: + WBWIMemTableIterator(std::unique_ptr&& it, + const WBWIMemTable::SeqnoRange& assigned_seqno, + const Comparator* comparator, bool for_flush) + : it_(std::move(it)), + assigned_seqno_(assigned_seqno), + comparator_(comparator), + emit_overwritten_single_del_(for_flush) { + assert(assigned_seqno_.lower_bound <= assigned_seqno_.upper_bound); + assert(assigned_seqno_.upper_bound < kMaxSequenceNumber); + s_.PermitUncheckedError(); + } + + // No copying allowed + WBWIMemTableIterator(const WBWIMemTableIterator&) = delete; + WBWIMemTableIterator& operator=(const WBWIMemTableIterator&) = delete; + + bool Valid() const override { return valid_; } + + void SeekToFirst() override { + it_->SeekToFirst(); + UpdateKey(); + } + + void SeekToLast() override { + it_->SeekToLast(); + UpdateKey(); + } + + void Seek(const Slice& target) override { + Slice target_user_key = ExtractUserKey(target); + it_->Seek(target_user_key); + if (it_->Valid()) { + // compare seqno + SequenceNumber seqno = GetInternalKeySeqno(target); + assert(!emit_overwritten_single_del_); + // For now all keys are assigned seqno_ub_, this may change after merge + // is supported. + assert(seqno <= assigned_seqno_.lower_bound || + seqno >= assigned_seqno_.upper_bound); + if (seqno < assigned_seqno_.upper_bound && + comparator_->Compare(it_->Entry().key, target_user_key) == 0) { + it_->Next(); + // TODO: cannot assume distinct keys once Merge is supported + if (it_->Valid()) { + assert(comparator_->Compare(it_->Entry().key, target_user_key) > 0); + } + } + } + UpdateKey(); + } + + void SeekForPrev(const Slice& target) override { + Slice target_user_key = ExtractUserKey(target); + it_->SeekForPrev(target_user_key); + if (it_->Valid()) { + SequenceNumber seqno = GetInternalKeySeqno(target); + assert(seqno <= assigned_seqno_.lower_bound || + seqno >= assigned_seqno_.upper_bound); + if (seqno > assigned_seqno_.upper_bound && + comparator_->Compare(it_->Entry().key, target_user_key) == 0) { + it_->Prev(); + if (it_->Valid()) { + // TODO: cannot assume distinct keys once Merge is supported + assert(comparator_->Compare(it_->Entry().key, target_user_key) < 0); + } + } + } + UpdateKey(); + } + + void Next() override { + // Only need to emit single deletion during flush. Since Flush does + // sequential forward scan, we only need to emit single deletion in Next(), + // and do not need to consider iterator direction change. + assert(Valid()); + if (emit_overwritten_single_del_) { + if (it_->HasOverWrittenSingleDel() && !at_overwritten_single_del_) { + UpdateSingleDeleteKey(); + return; + } + at_overwritten_single_del_ = false; + } + + it_->Next(); + UpdateKey(); + } + + bool NextAndGetResult(IterateResult* result) override { + assert(Valid()); + Next(); + bool is_valid = Valid(); + if (is_valid) { + result->key = key(); + result->bound_check_result = IterBoundCheck::kUnknown; + result->value_prepared = true; + } + return is_valid; + } + + void Prev() override { + assert(Valid()); + it_->Prev(); + UpdateKey(); + } + + Slice key() const override { + assert(Valid()); + return key_; + } + + Slice value() const override { + assert(Valid()); + // TODO: it_->Entry() is not trivial, cache it + return it_->Entry().value; + } + + Status status() const override { + assert(it_->status().ok()); + return s_; + } + + bool IsValuePinned() const override { return true; } + + private: + static const std::unordered_map WriteTypeToValueTypeMap; + + void UpdateKey() { + valid_ = it_->Valid(); + if (!Valid()) { + key_.clear(); + return; + } + auto t = WriteTypeToValueTypeMap.find(it_->Entry().type); + assert(t != WriteTypeToValueTypeMap.end()); + if (t == WriteTypeToValueTypeMap.end()) { + key_.clear(); + valid_ = false; + s_ = Status::Corruption("Unexpected write_batch_with_index entry type " + + std::to_string(t->second)); + return; + } + key_buf_.SetInternalKey(it_->Entry().key, assigned_seqno_.upper_bound, + t->second); + key_ = key_buf_.GetInternalKey(); + } + + void UpdateSingleDeleteKey() { + assert(it_->Valid()); + assert(Valid()); + assert(assigned_seqno_.lower_bound < assigned_seqno_.upper_bound); + key_buf_.SetInternalKey(it_->Entry().key, assigned_seqno_.upper_bound - 1, + kTypeSingleDeletion); + key_ = key_buf_.GetInternalKey(); + at_overwritten_single_del_ = true; + } + + std::unique_ptr it_; + const WBWIMemTable::SeqnoRange assigned_seqno_; + const Comparator* comparator_; + IterKey key_buf_; + // The current internal key. + Slice key_; + Status s_; + bool valid_ = false; + bool at_overwritten_single_del_ = false; + bool emit_overwritten_single_del_ = false; +}; + } // namespace ROCKSDB_NAMESPACE diff --git a/table/table_test.cc b/table/table_test.cc index b198baada4..6207a40ce2 100644 --- a/table/table_test.cc +++ b/table/table_test.cc @@ -534,7 +534,8 @@ class MemTableConstructor : public Constructor { const SliceTransform* /*prefix_extractor*/) const override { return new KeyConvertingIterator( memtable_->NewIterator(ReadOptions(), /*seqno_to_time_mapping=*/nullptr, - &arena_, /*prefix_extractor=*/nullptr), + &arena_, /*prefix_extractor=*/nullptr, + /*for_flush=*/false), true); } @@ -4911,7 +4912,8 @@ TEST_F(MemTableTest, Simple) { if (i == 0) { iter = GetMemTable()->NewIterator(ReadOptions(), /*seqno_to_time_mapping=*/nullptr, - &arena, /*prefix_extractor=*/nullptr); + &arena, /*prefix_extractor=*/nullptr, + /*for_flush=*/false); arena_iter_guard.reset(iter); } else { iter = GetMemTable()->NewRangeTombstoneIterator( diff --git a/tools/db_crashtest.py b/tools/db_crashtest.py index f61dc9f7b8..aaa3b645b2 100644 --- a/tools/db_crashtest.py +++ b/tools/db_crashtest.py @@ -674,6 +674,7 @@ multiops_wc_txn_params = { "txn_write_policy": 0, # TODO re-enable pipelined write. Not well tested atm "enable_pipelined_write": 0, + "commit_bypass_memtable_one_in": random.choice([0] * 4 + [100]), } multiops_wp_txn_params = { @@ -988,6 +989,18 @@ def finalize_and_sanitize(src_params): or dest_params.get("delrangepercent") == 0 ): dest_params["test_ingest_standalone_range_deletion_one_in"] = 0 + if dest_params.get("commit_bypass_memtable_one_in", 0) > 0: + dest_params["enable_blob_files"] = 0 + dest_params["allow_setting_blob_options_dynamically"] = 0 + dest_params["atomic_flush"] = 0 + dest_params["allow_concurrent_memtable_write"] = 0 + dest_params["use_put_entity_one_in"] = 0 + dest_params["use_get_entity"] = 0 + dest_params["use_multi_get_entity"] = 0 + dest_params["use_merge"] = 0 + dest_params["use_full_merge_v1"] = 0 + dest_params["enable_pipelined_write"] = 0 + return dest_params diff --git a/unreleased_history/new_features/txn-bypass-memtable.md b/unreleased_history/new_features/txn-bypass-memtable.md new file mode 100644 index 0000000000..9c8720306f --- /dev/null +++ b/unreleased_history/new_features/txn-bypass-memtable.md @@ -0,0 +1 @@ +* Introduce `TransactionOptions::commit_bypass_memtable` to enable transaction commit to bypass memtable insertions. This can be beneficial for transactions with many operations, as it reduces commit time that is mostly spent on memtable insertion. \ No newline at end of file diff --git a/utilities/transactions/pessimistic_transaction.cc b/utilities/transactions/pessimistic_transaction.cc index ab7ea4f62b..ef3180f04e 100644 --- a/utilities/transactions/pessimistic_transaction.cc +++ b/utilities/transactions/pessimistic_transaction.cc @@ -103,6 +103,9 @@ void PessimisticTransaction::Initialize(const TransactionOptions& txn_options) { read_timestamp_ = kMaxTxnTimestamp; commit_timestamp_ = kMaxTxnTimestamp; + + commit_bypass_memtable_ = txn_options.commit_bypass_memtable; + write_batch_.SetTrackPerCFStat(txn_options.commit_bypass_memtable); } PessimisticTransaction::~PessimisticTransaction() { @@ -843,6 +846,7 @@ Status WriteCommittedTxn::CommitInternal() { if (!needs_ts) { s = WriteBatchInternal::MarkCommit(working_batch, name_); } else { + assert(!commit_bypass_memtable_); assert(commit_timestamp_ != kMaxTxnTimestamp); char commit_ts_buf[sizeof(kMaxTxnTimestamp)]; EncodeFixed64(commit_ts_buf, commit_timestamp_); @@ -878,11 +882,14 @@ Status WriteCommittedTxn::CommitInternal() { // any operations appended to this working_batch will be ignored from WAL working_batch->MarkWalTerminationPoint(); - // insert prepared batch into Memtable only skipping WAL. - // Memtable will ignore BeginPrepare/EndPrepare markers - // in non recovery mode and simply insert the values - s = WriteBatchInternal::Append(working_batch, wb); - assert(s.ok()); + const bool bypass_memtable = commit_bypass_memtable_ && wb->Count() > 0; + if (!bypass_memtable) { + // insert prepared batch into Memtable only skipping WAL. + // Memtable will ignore BeginPrepare/EndPrepare markers + // in non recovery mode and simply insert the values + s = WriteBatchInternal::Append(working_batch, wb); + assert(s.ok()); + } uint64_t seq_used = kMaxSequenceNumber; SnapshotCreationCallback snapshot_creation_cb(db_impl_, commit_timestamp_, @@ -896,12 +903,28 @@ Status WriteCommittedTxn::CommitInternal() { post_mem_cb = &snapshot_creation_cb; } } - s = db_impl_->WriteImpl(write_options_, working_batch, /*callback*/ nullptr, - /*user_write_cb=*/nullptr, - /*log_used*/ nullptr, /*log_ref*/ log_number_, - /*disable_memtable*/ false, &seq_used, - /*batch_cnt=*/0, /*pre_release_callback=*/nullptr, - post_mem_cb); + assert(log_number_ > 0); + if (bypass_memtable) { + s = db_impl_->WriteImpl( + write_options_, working_batch, /*callback*/ nullptr, + /*user_write_cb=*/nullptr, + /*log_used*/ nullptr, /*log_ref*/ log_number_, + /*disable_memtable*/ false, &seq_used, + /*batch_cnt=*/0, /*pre_release_callback=*/nullptr, post_mem_cb, + /*wbwi=*/std::make_shared(std::move(write_batch_)), + /*min_prep_log=*/log_number_); + // Reset write_batch_ since it's accessed in transaction clean up and + // might be used for transaction reuse. + write_batch_ = WriteBatchWithIndex(cmp_, 0, true, 0, + write_options_.protection_bytes_per_key); + } else { + s = db_impl_->WriteImpl(write_options_, working_batch, /*callback*/ nullptr, + /*user_write_cb=*/nullptr, + /*log_used*/ nullptr, /*log_ref*/ log_number_, + /*disable_memtable*/ false, &seq_used, + /*batch_cnt=*/0, /*pre_release_callback=*/nullptr, + post_mem_cb); + } assert(!s.ok() || seq_used != kMaxSequenceNumber); if (s.ok()) { SetId(seq_used); diff --git a/utilities/transactions/pessimistic_transaction.h b/utilities/transactions/pessimistic_transaction.h index c4f5a22a0e..ecf8ccb24f 100644 --- a/utilities/transactions/pessimistic_transaction.h +++ b/utilities/transactions/pessimistic_transaction.h @@ -122,13 +122,6 @@ class PessimisticTransaction : public TransactionBaseImpl { ColumnFamilyHandle* column_family = nullptr) override; protected: - // Refer to - // TransactionOptions::use_only_the_last_commit_time_batch_for_recovery - bool use_only_the_last_commit_time_batch_for_recovery_ = false; - // Refer to - // TransactionOptions::skip_prepare - bool skip_prepare_ = false; - virtual Status PrepareInternal() = 0; virtual Status CommitWithoutPrepareInternal() = 0; @@ -167,6 +160,16 @@ class PessimisticTransaction : public TransactionBaseImpl { TxnTimestamp read_timestamp_{kMaxTxnTimestamp}; TxnTimestamp commit_timestamp_{kMaxTxnTimestamp}; + // Refer to + // TransactionOptions::use_only_the_last_commit_time_batch_for_recovery + bool use_only_the_last_commit_time_batch_for_recovery_ = false; + // Refer to + // TransactionOptions::skip_prepare + bool skip_prepare_ = false; + // Refer to + // TransactionOptions::commit_bypass_memtable + bool commit_bypass_memtable_ = false; + private: friend class TransactionTest_ValidateSnapshotTest_Test; // Used to create unique ids for transactions. @@ -304,6 +307,10 @@ class WriteCommittedTxn : public PessimisticTransaction { Status SetCommitTimestamp(TxnTimestamp ts) override; TxnTimestamp GetCommitTimestamp() const override { return commit_timestamp_; } + bool GetCommitBypassMemTable() const override { + return commit_bypass_memtable_; + } + private: template Status GetForUpdateImpl(const ReadOptions& read_options, diff --git a/utilities/transactions/transaction_test.cc b/utilities/transactions/transaction_test.cc index a089605d37..08bd14360c 100644 --- a/utilities/transactions/transaction_test.cc +++ b/utilities/transactions/transaction_test.cc @@ -8,6 +8,7 @@ #include #include #include +#include #include #include @@ -8100,6 +8101,485 @@ TEST_F(TransactionDBTest, FlushedLogWithPendingPrepareIsSynced) { } } +class CommitBypassMemtableTest : public DBTestBase, + public ::testing::WithParamInterface { + public: + CommitBypassMemtableTest() : DBTestBase("commit_bypass_memtable_test", true) { + SetUpTransactionDB(); + } + + protected: + TransactionDB* txn_db = nullptr; + Options options; + TransactionDBOptions txn_db_opts; + + void SetUpTransactionDB() { + options = CurrentOptions(); + options.create_if_missing = true; + options.allow_2pc = true; + options.two_write_queues = GetParam(); + // Avoid write stall + options.max_write_buffer_number = 8; + // Destroy the DB to recreate as a TransactionDB. + Close(); + Destroy(options, true); + + txn_db_opts.write_policy = TxnDBWritePolicy::WRITE_COMMITTED; + ASSERT_OK(TransactionDB::Open(options, txn_db_opts, dbname_, &txn_db)); + ASSERT_NE(txn_db, nullptr); + db_ = txn_db; + } +}; + +INSTANTIATE_TEST_CASE_P(, CommitBypassMemtableTest, testing::Bool()); + +// TODO: parameterize other tests in the file with commit_bypass_memtable +TEST_P(CommitBypassMemtableTest, SingleCFUpdate) { + // 10000 updates for one CF in a single transaction. + // Tests basic read before and after flush, with and w/o snapshot. + for (bool disable_flush : {false, true}) { + SetUpTransactionDB(); + if (disable_flush) { + ASSERT_OK(db_->PauseBackgroundWork()); + } + + WriteOptions wopts; + std::unordered_set not_found_at_snapshot; + std::map snapshot_map; + std::map expected_map; + + for (int i = 0; i < 10000; i += 3) { + ASSERT_OK(db_->Put(wopts, Key(i), "foo")); + not_found_at_snapshot.insert(Key(i + 1)); + snapshot_map[Key(i)] = "foo"; + } + const Snapshot* snapshot = db_->GetSnapshot(); + + TransactionOptions txn_opts; + txn_opts.commit_bypass_memtable = true; + Transaction* txn1 = txn_db->BeginTransaction(wopts, txn_opts, nullptr); + std::unordered_set expected_not_found; + for (int i = 0; i < 10000; i += 2) { + std::string v = "val" + std::to_string(i); + ASSERT_OK(txn1->Put(Key(i), v)); + expected_map[Key(i)] = v; + ASSERT_OK(txn1->Delete(Key(i + 1))); + expected_not_found.insert(Key(i + 1)); + } + ASSERT_OK(txn1->SetName("xid1")); + ASSERT_OK(txn1->Prepare()); + ASSERT_OK(txn1->Commit()); + delete txn1; + + ReadOptions ro_snapshot; + ro_snapshot.snapshot = snapshot; + // Verify at snapshot + VerifyDBFromMap(snapshot_map, nullptr, false, &ro_snapshot, nullptr, + ¬_found_at_snapshot); + // Verify latest state + VerifyDBFromMap(expected_map, nullptr, false, nullptr, nullptr, + &expected_not_found); + + db_->ReleaseSnapshot(snapshot); + if (disable_flush) { + ASSERT_OK(db_->ContinueBackgroundWork()); + } + ASSERT_OK(db_->Flush({})); + VerifyDBFromMap(expected_map, nullptr, false, nullptr, nullptr, + &expected_not_found); + } +} + +TEST_P(CommitBypassMemtableTest, SingleCFUpdateWithOverWrite) { + // Test the case where DB has base data and there are overwrites + // over the data in WBWI for one CF. + // + // live mem has k2 + // ingested wbwi has k2 del k4, del k5, k6 + // older imm mem has k3, k4, k5, k6 + // SST has k1, k2, k4 (if not using single del) + for (bool single_del : {false, true}) { + SCOPED_TRACE("use single_del " + std::to_string(single_del)); + for (bool disable_flush : {false, true}) { + SCOPED_TRACE("disable_flush: " + std::to_string(disable_flush)); + SetUpTransactionDB(); + + std::map expected = {{"k1", "sst"}, + {"k2", "sst"}}; + WriteOptions wopts; + ASSERT_OK(txn_db->Put(wopts, "k1", "sst")); + ASSERT_OK(txn_db->Put(wopts, "k2", "sst")); + if (!single_del) { + // single del does not allow overwrite. We write to this key below. + ASSERT_OK(txn_db->Put(wopts, "k4", "sst")); + expected["k4"] = "sst"; + } + ASSERT_OK(txn_db->Flush({})); + std::unordered_set not_found; + VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, ¬_found); + + if (disable_flush) { + ASSERT_OK(db_->PauseBackgroundWork()); + } + + // immutable mem + TransactionOptions topts; + Transaction* txn = txn_db->BeginTransaction(wopts, topts); + ASSERT_OK(txn->Put("k3", "imm")); + ASSERT_OK(txn->Put("k4", "imm")); + ASSERT_OK(txn->Put("k5", "imm")); + ASSERT_OK(txn->Put("k6", "imm")); + ASSERT_OK(txn->SetName("xid1")); + ASSERT_OK(txn->Prepare()); + ASSERT_OK(txn->Commit()); + auto dbimpl = static_cast_with_check(txn_db->GetRootDB()); + ASSERT_OK(dbimpl->TEST_SwitchMemtable()); + for (const auto k : {"k3", "k4", "k5", "k6"}) { + expected[k] = "imm"; + } + VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, ¬_found); + + uint64_t num_imm_mems; + if (disable_flush) { + ASSERT_TRUE(txn_db->GetIntProperty( + DB::Properties::kNumImmutableMemTable, &num_imm_mems)); + ASSERT_EQ(num_imm_mems, 1); + } + + // ingest wbwi + const Snapshot* snapshot = txn_db->GetSnapshot(); + TransactionOptions topts_ingest; + topts_ingest.commit_bypass_memtable = true; + // reuse txn1 + txn = txn_db->BeginTransaction(wopts, topts_ingest, txn); + ASSERT_OK(txn->Put("k2", "wbwi")); + if (single_del) { + ASSERT_OK(txn->SingleDelete("k4")); + ASSERT_OK(txn->SingleDelete("k5")); + } else { + ASSERT_OK(txn->Delete("k4")); + ASSERT_OK(txn->Delete("k5")); + } + ASSERT_OK(txn->Put("k6", "wbwi")); + ASSERT_OK(txn->SetName("xid2")); + ASSERT_OK(txn->Prepare()); + VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, ¬_found); + + ASSERT_OK(txn->Commit()); + if (disable_flush) { + ASSERT_TRUE(txn_db->GetIntProperty( + DB::Properties::kNumImmutableMemTable, &num_imm_mems)); + // During Commit(), current empty memtable becomes a new immutable + // memtable. 3 here includes wbwi for this transaction, the empty + // immutable memtable, and the immutable memtable created + // by TEST_SwitchMemtable() above. + ASSERT_EQ(num_imm_mems, 3); + } + + ReadOptions ro_snapshot; + ro_snapshot.snapshot = snapshot; + auto expected_at_snapshot = expected; + auto not_found_at_snapshot = not_found; + VerifyDBFromMap(expected, nullptr, false, &ro_snapshot, nullptr, + ¬_found); + not_found = {"k4", "k5"}; + expected.erase("k4"); + expected.erase("k5"); + expected["k2"] = "wbwi"; + expected["k6"] = "wbwi"; + VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, ¬_found); + + // live mem + txn = txn_db->BeginTransaction(wopts, topts, txn); + ASSERT_OK(txn->Put("k2", "live_mem")); + ASSERT_OK(txn->SetName("xid3")); + ASSERT_OK(txn->Prepare()); + ASSERT_OK(txn->Commit()); + delete txn; + expected["k2"] = "live_mem"; + VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, ¬_found); + + if (disable_flush) { + ASSERT_OK(db_->ContinueBackgroundWork()); + ASSERT_OK(db_->Flush({})); + } else { + ASSERT_OK(db_->WaitForCompact({})); + } + ASSERT_TRUE(txn_db->GetIntProperty(DB::Properties::kNumImmutableMemTable, + &num_imm_mems)); + ASSERT_EQ(num_imm_mems, 0); + + VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, ¬_found); + VerifyDBFromMap(expected_at_snapshot, nullptr, false, &ro_snapshot, + nullptr, ¬_found_at_snapshot); + txn_db->ReleaseSnapshot(snapshot); + } + } +} + +TEST_P(CommitBypassMemtableTest, MultiCFOverwrite) { + // mini-stress test multi CF update + const int kNumKeys = 10000; + for (bool disable_flush : {false, true}) { + SCOPED_TRACE("disable_flush: " + std::to_string(disable_flush)); + SetUpTransactionDB(); + if (disable_flush) { + ASSERT_OK(txn_db->PauseBackgroundWork()); + } + std::vector cfs = {"puppy", "kitty", "meta"}; + CreateColumnFamilies(cfs, options); + ASSERT_EQ(handles_.size(), 3); + ASSERT_EQ(handles_[0]->GetName(), cfs[0]); + ASSERT_EQ(handles_[1]->GetName(), cfs[1]); + ASSERT_EQ(handles_[2]->GetName(), cfs[2]); + + std::map expected_cf0; + std::unordered_set not_found_cf0; + std::map expected_cf1; + std::unordered_set not_found_cf1; + WriteOptions wopts; + Random* rnd = Random::GetTLSInstance(); + // Some base data + for (int i = 0; i < kNumKeys; ++i) { + std::string val_cf0 = "cf0_sst" + std::to_string(i); + if (rnd->OneIn(2)) { + ASSERT_OK(txn_db->Put(wopts, handles_[0], Key(i), val_cf0)); + expected_cf0[Key(i)] = val_cf0; + } else { + not_found_cf0.insert(Key(i)); + } + std::string val_cf1 = "cf1_sst" + std::to_string(i); + ASSERT_OK(txn_db->Put(wopts, handles_[1], Key(i), val_cf1)); + expected_cf1[Key(i)] = val_cf1; + } + SCOPED_TRACE("Verify after SST"); + ASSERT_OK(txn_db->Put(wopts, handles_[2], "id", "0")); + std::map expected_cf2 = {{"id", "0"}}; + VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0], + ¬_found_cf0); + VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1], + ¬_found_cf1); + + if (!disable_flush) { + ASSERT_OK(txn_db->Flush({}, handles_[0])); + ASSERT_OK(txn_db->Flush({}, handles_[1])); + } + + const Snapshot* snapshot = nullptr; + std::map expected_cf0_snapshot; + std::unordered_set not_found_cf0_snapshot; + std::map expected_cf1_snapshot; + std::unordered_set not_found_cf1_snapshot; + std::map expected_cf2_snapshot; + auto init_snapshot_states = [&]() { + snapshot = txn_db->GetSnapshot(); + expected_cf0_snapshot = expected_cf0; + not_found_cf0_snapshot = not_found_cf0; + expected_cf1_snapshot = expected_cf1; + not_found_cf1_snapshot = not_found_cf1; + expected_cf2_snapshot = expected_cf2; + }; + if (rnd->OneIn(4)) { + init_snapshot_states(); + } + + // Transaction 1 + TransactionOptions topts; + topts.commit_bypass_memtable = true; + Transaction* txn1 = txn_db->BeginTransaction(wopts, topts); + auto fill_txn = [&](Transaction* txn, const std::string& v) { + std::vector indices(kNumKeys); + std::iota(indices.begin(), indices.end(), 0); + RandomShuffle(indices.begin(), indices.end()); + for (int i : indices) { + // CF0 update + if (rnd->OneIn(4)) { + std::string val_cf0 = "cf0_" + v + std::to_string(i); + ASSERT_OK(txn->Put(handles_[0], Key(i), val_cf0)); + expected_cf0[Key(i)] = val_cf0; + not_found_cf0.erase(Key(i)); + } else if (rnd->OneIn(4)) { + ASSERT_OK(txn->Delete(handles_[0], Key(i))); + expected_cf0.erase(Key(i)); + not_found_cf0.insert(Key(i)); + } + + // CF1 update + if (rnd->OneIn(4)) { + ASSERT_OK(txn->SingleDelete(handles_[1], Key(i))); + expected_cf1.erase(Key(i)); + not_found_cf1.insert(Key(i)); + } + if (rnd->OneIn(4) && + not_found_cf1.find(Key(i)) != not_found_cf1.end()) { + // Can not overwrite when using SD. + std::string val_cf1 = "cf1_" + v + std::to_string(i); + ASSERT_OK(txn->Put(handles_[1], Key(i), val_cf1)); + expected_cf1[Key(i)] = val_cf1; + not_found_cf1.erase(Key(i)); + } + } + }; + fill_txn(txn1, "txn1"); + ASSERT_OK(txn1->SetName("xid1")); + ASSERT_OK(txn1->Prepare()); + if (rnd->OneIn(2)) { + ASSERT_OK(txn1->GetCommitTimeWriteBatch()->Put(handles_[2], "id", "1")); + expected_cf2["id"] = "1"; + } + ASSERT_OK(txn1->Commit()); + delete txn1; + + uint64_t num_imm_mems = 0; + if (disable_flush) { + ASSERT_TRUE(txn_db->GetIntProperty( + handles_[0], DB::Properties::kNumImmutableMemTable, &num_imm_mems)); + ASSERT_EQ(num_imm_mems, 2); + + ASSERT_TRUE(txn_db->GetIntProperty( + handles_[1], DB::Properties::kNumImmutableMemTable, &num_imm_mems)); + ASSERT_EQ(num_imm_mems, 2); + ASSERT_TRUE(txn_db->GetIntProperty( + handles_[2], DB::Properties::kNumImmutableMemTable, &num_imm_mems)); + ASSERT_EQ(num_imm_mems, 0); + } + SCOPED_TRACE("Verify cf0 after txn1"); + VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0], + ¬_found_cf0); + SCOPED_TRACE("Verify cf1 after txn1"); + SCOPED_TRACE("db is " + txn_db->GetName()); + VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1], + ¬_found_cf1); + VerifyDBFromMap(expected_cf2, nullptr, false, nullptr, handles_[2]); + auto verify_at_snapshot = [&](const std::string& m) { + if (snapshot) { + SCOPED_TRACE("Verify at snapshot " + m); + ReadOptions ro; + ro.snapshot = snapshot; + SCOPED_TRACE("Verify cf0 after txn1"); + VerifyDBFromMap(expected_cf0_snapshot, nullptr, false, &ro, handles_[0], + ¬_found_cf0_snapshot); + SCOPED_TRACE("Verify cf1 after txn1"); + VerifyDBFromMap(expected_cf1_snapshot, nullptr, false, &ro, handles_[1], + ¬_found_cf1_snapshot); + VerifyDBFromMap(expected_cf2_snapshot, nullptr, false, &ro, + handles_[2]); + if (rnd->OneIn(2)) { + txn_db->ReleaseSnapshot(snapshot); + snapshot = nullptr; + } + } + }; + verify_at_snapshot("txn1"); + if (!snapshot && rnd->OneIn(4)) { + init_snapshot_states(); + } + + // Live memtable or another transaction with commit_bypass_memtable + TransactionOptions topts2; + topts2.commit_bypass_memtable = rnd->OneIn(2); + SCOPED_TRACE("txn2 commit_bypass_memtable = " + + std::to_string(topts2.commit_bypass_memtable)); + Transaction* txn2 = txn_db->BeginTransaction(wopts, topts2); + fill_txn(txn2, "txn2"); + ASSERT_OK(txn2->SetName("xid2")); + ASSERT_OK(txn2->Prepare()); + if (rnd->OneIn(2)) { + ASSERT_OK(txn2->GetCommitTimeWriteBatch()->Put(handles_[2], "id", "2")); + expected_cf2["id"] = "2"; + } + ASSERT_OK(txn2->Commit()); + delete txn2; + if (disable_flush) { + ASSERT_TRUE(txn_db->GetIntProperty( + handles_[0], DB::Properties::kNumImmutableMemTable, &num_imm_mems)); + ASSERT_EQ(num_imm_mems, topts2.commit_bypass_memtable ? 4 : 2); + + ASSERT_TRUE(txn_db->GetIntProperty( + handles_[1], DB::Properties::kNumImmutableMemTable, &num_imm_mems)); + ASSERT_EQ(num_imm_mems, topts2.commit_bypass_memtable ? 4 : 2); + ASSERT_TRUE(txn_db->GetIntProperty( + handles_[2], DB::Properties::kNumImmutableMemTable, &num_imm_mems)); + ASSERT_EQ(num_imm_mems, 0); + } + + SCOPED_TRACE("Verify cf0 after txn2"); + VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0], + ¬_found_cf0); + SCOPED_TRACE("Verify cf1 after txn2"); + VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1], + ¬_found_cf1); + // cf2 should not have keys in not_found_cf1 either + VerifyDBFromMap(expected_cf2, nullptr, false, nullptr, handles_[2], + ¬_found_cf1); + verify_at_snapshot("txn2"); + if (!snapshot && rnd->OneIn(4)) { + init_snapshot_states(); + } + + // Verify all data is flushed + if (disable_flush) { + ASSERT_OK(db_->ContinueBackgroundWork()); + ASSERT_OK(db_->Flush({}, handles_[0])); + ASSERT_OK(db_->Flush({}, handles_[1])); + } else { + ASSERT_OK(db_->WaitForCompact({})); + } + + VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0], + ¬_found_cf0); + VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1], + ¬_found_cf1); + VerifyDBFromMap(expected_cf2, nullptr, false, nullptr, handles_[2], + ¬_found_cf1); + verify_at_snapshot("flush"); + if (snapshot) { + db_->ReleaseSnapshot(snapshot); + } + } +} + +TEST_P(CommitBypassMemtableTest, Recovery) { + // Test that ingested txn can be recovered. + // Implementation detail: this tests that write path reserves enough sequence + // number for the ingested memtables (see seq_inc in WriteImpl()). For + // example, suppose that we only assign one sequence number per ingested + // memtable. Then txn1 will be assigned seqno 1 and txn2 will be assigned 2. + // During recovery, we will not use memtable ingestion and will fall back + // to memtable insertion. The sequence number for the first key in a + // transaction is the commit sequence number. So for txn1, we will insert into + // memtable k2@1 and k1@2. And for txn2, we will insert k1@2 and k2@3. This + // results in duplicated key@seqno. + WriteOptions wopts; + TransactionOptions txn_opts; + txn_opts.commit_bypass_memtable = true; + Transaction* txn1 = txn_db->BeginTransaction(wopts, txn_opts, nullptr); + ASSERT_OK(txn1->SetName("xid1")); + ASSERT_OK(txn1->Put("k2", "v2")); + ASSERT_OK(txn1->Put("k1", "v1")); + ASSERT_OK(txn1->Prepare()); + ASSERT_OK(txn1->Commit()); + + // Test txn reuse code path + txn1 = txn_db->BeginTransaction(wopts, txn_opts, txn1); + ASSERT_OK(txn1->SetName("xid2")); + ASSERT_OK(txn1->Put("k1", "v3")); + ASSERT_OK(txn1->Put("k2", "v4")); + ASSERT_OK(txn1->Prepare()); + ASSERT_OK(txn1->Commit()); + delete txn1; + + std::map expected = {{"k1", "v3"}, {"k2", "v4"}}; + VerifyDBFromMap(expected); + + ASSERT_OK(txn_db->Close()); + delete txn_db; + ASSERT_OK(TransactionDB::Open(options, txn_db_opts, dbname_, &txn_db)); + db_ = txn_db; + + VerifyDBFromMap(expected); +} + } // namespace ROCKSDB_NAMESPACE int main(int argc, char** argv) { diff --git a/utilities/write_batch_with_index/write_batch_with_index.cc b/utilities/write_batch_with_index/write_batch_with_index.cc index db4c53847b..98d7a95b4d 100644 --- a/utilities/write_batch_with_index/write_batch_with_index.cc +++ b/utilities/write_batch_with_index/write_batch_with_index.cc @@ -35,7 +35,8 @@ struct WriteBatchWithIndex::Rep { overwrite_key(_overwrite_key), last_entry_offset(0), last_sub_batch_offset(0), - sub_batch_cnt(1) {} + sub_batch_cnt(1), + track_cf_stat(false) {} ReadableWriteBatch write_batch; WriteBatchEntryComparator comparator; Arena arena; @@ -49,6 +50,11 @@ struct WriteBatchWithIndex::Rep { // Total number of sub-batches in the write batch. Default is 1. size_t sub_batch_cnt; + bool track_cf_stat; + // Tracks ids of CFs that have updates in this WBWI, number of updates and + // number of overwritten single deletions per cf. + std::unordered_map cf_id_to_stat; + // Remember current offset of internal write batch, which is used as // the starting offset of the next record. void SetLastEntryOffset() { last_entry_offset = write_batch.GetDataSize(); } @@ -69,7 +75,7 @@ struct WriteBatchWithIndex::Rep { // Allocate an index entry pointing to the last entry in the write batch and // put it to skip list. - void AddNewEntry(uint32_t column_family_id); + void AddNewEntry(uint32_t column_family_id, WriteType type); // Clear all updates buffered in this batch. void Clear(); @@ -114,6 +120,16 @@ bool WriteBatchWithIndex::Rep::UpdateExistingEntryWithCfId( last_sub_batch_offset = last_entry_offset; sub_batch_cnt++; } + if (track_cf_stat) { + if (non_const_entry->has_single_del && + !non_const_entry->has_overwritten_single_del) { + cf_id_to_stat[column_family_id].overwritten_sd_count++; + non_const_entry->has_overwritten_single_del = true; + } + if (type == kSingleDeleteRecord) { + non_const_entry->has_single_del = true; + } + } if (type == kMergeRecord) { return false; } else { @@ -130,18 +146,19 @@ void WriteBatchWithIndex::Rep::AddOrUpdateIndex( if (cf_cmp != nullptr) { comparator.SetComparatorForCF(cf_id, cf_cmp); } - AddNewEntry(cf_id); + AddNewEntry(cf_id, type); } } void WriteBatchWithIndex::Rep::AddOrUpdateIndex(const Slice& key, WriteType type) { if (!UpdateExistingEntryWithCfId(0, key, type)) { - AddNewEntry(0); + AddNewEntry(0, type); } } -void WriteBatchWithIndex::Rep::AddNewEntry(uint32_t column_family_id) { +void WriteBatchWithIndex::Rep::AddNewEntry(uint32_t column_family_id, + WriteType type) { const std::string& wb_data = write_batch.Data(); Slice entry_ptr = Slice(wb_data.data() + last_entry_offset, wb_data.size() - last_entry_offset); @@ -166,10 +183,18 @@ void WriteBatchWithIndex::Rep::AddNewEntry(uint32_t column_family_id) { new (mem) WriteBatchIndexEntry(last_entry_offset, column_family_id, key.data() - wb_data.data(), key.size()); skip_list.Insert(index_entry); + + if (track_cf_stat) { + if (type == kSingleDeleteRecord) { + index_entry->has_single_del = true; + } + cf_id_to_stat[column_family_id].entry_count++; + } } void WriteBatchWithIndex::Rep::Clear() { write_batch.Clear(); + cf_id_to_stat.clear(); ClearIndex(); } @@ -220,7 +245,7 @@ Status WriteBatchWithIndex::Rep::ReBuildIndex() { case kTypeValue: found++; if (!UpdateExistingEntryWithCfId(column_family_id, key, kPutRecord)) { - AddNewEntry(column_family_id); + AddNewEntry(column_family_id, kPutRecord); } break; case kTypeColumnFamilyDeletion: @@ -228,7 +253,7 @@ Status WriteBatchWithIndex::Rep::ReBuildIndex() { found++; if (!UpdateExistingEntryWithCfId(column_family_id, key, kDeleteRecord)) { - AddNewEntry(column_family_id); + AddNewEntry(column_family_id, kDeleteRecord); } break; case kTypeColumnFamilySingleDeletion: @@ -236,14 +261,14 @@ Status WriteBatchWithIndex::Rep::ReBuildIndex() { found++; if (!UpdateExistingEntryWithCfId(column_family_id, key, kSingleDeleteRecord)) { - AddNewEntry(column_family_id); + AddNewEntry(column_family_id, kSingleDeleteRecord); } break; case kTypeColumnFamilyMerge: case kTypeMerge: found++; if (!UpdateExistingEntryWithCfId(column_family_id, key, kMergeRecord)) { - AddNewEntry(column_family_id); + AddNewEntry(column_family_id, kMergeRecord); } break; case kTypeLogData: @@ -261,7 +286,7 @@ Status WriteBatchWithIndex::Rep::ReBuildIndex() { found++; if (!UpdateExistingEntryWithCfId(column_family_id, key, kPutEntityRecord)) { - AddNewEntry(column_family_id); + AddNewEntry(column_family_id, kPutEntityRecord); } break; case kTypeColumnFamilyValuePreferredSeqno: @@ -1138,4 +1163,17 @@ const Comparator* WriteBatchWithIndexInternal::GetUserComparator( return ucmps.GetComparator(cf_id); } +void WriteBatchWithIndex::SetTrackPerCFStat(bool track) { + // Should be set when the wbwi contains no update. + assert(GetWriteBatch()->Count() == 0); + rep->track_cf_stat = track; +} + +const std::unordered_map& +WriteBatchWithIndex::GetCFStats() const { + assert(rep->track_cf_stat); + return rep->cf_id_to_stat; +} + +bool WriteBatchWithIndex::GetOverwriteKey() const { return rep->overwrite_key; } } // namespace ROCKSDB_NAMESPACE diff --git a/utilities/write_batch_with_index/write_batch_with_index_internal.h b/utilities/write_batch_with_index/write_batch_with_index_internal.h index c60ed93868..b06ea2dd34 100644 --- a/utilities/write_batch_with_index/write_batch_with_index_internal.h +++ b/utilities/write_batch_with_index/write_batch_with_index_internal.h @@ -104,6 +104,8 @@ struct WriteBatchIndexEntry { WriteBatchIndexEntry(size_t o, uint32_t c, size_t ko, size_t ksz) : offset(o), column_family(c), + has_single_del(false), + has_overwritten_single_del(false), key_offset(ko), key_size(ksz), search_key(nullptr) {} @@ -121,6 +123,8 @@ struct WriteBatchIndexEntry { // entry who has the same search key. Otherwise, we'll miss those entries. : offset(is_forward_direction ? 0 : std::numeric_limits::max()), column_family(_column_family), + has_single_del(false), + has_overwritten_single_del(false), key_offset(0), key_size(is_seek_to_first ? kFlagMinInCf : 0), search_key(_search_key) { @@ -147,6 +151,9 @@ struct WriteBatchIndexEntry { // SeekForPrev() will see all the keys with the same key. size_t offset; uint32_t column_family; // column family of the entry. + bool has_single_del; // whether single del was issued for this key + bool has_overwritten_single_del; // whether a single del for this key was + // overwritten by another key size_t key_offset; // offset of the key in write batch's string buffer. size_t key_size; // size of the key. kFlagMinInCf indicates // that this is a dummy look up entry for @@ -209,7 +216,7 @@ class WriteBatchEntryComparator { using WriteBatchEntrySkipList = SkipList; -class WBWIIteratorImpl : public WBWIIterator { +class WBWIIteratorImpl final : public WBWIIterator { public: enum Result : uint8_t { kFound, @@ -325,6 +332,11 @@ class WBWIIteratorImpl : public WBWIIterator { WriteEntry Entry() const override; + bool HasOverWrittenSingleDel() const override { + assert(Valid()); + return skip_list_iter_.key()->has_overwritten_single_del; + } + Status status() const override { // this is in-memory data structure, so the only way status can be non-ok is // through memory corruption diff --git a/utilities/write_batch_with_index/write_batch_with_index_test.cc b/utilities/write_batch_with_index/write_batch_with_index_test.cc index 113f84d9fe..e779b268cf 100644 --- a/utilities/write_batch_with_index/write_batch_with_index_test.cc +++ b/utilities/write_batch_with_index/write_batch_with_index_test.cc @@ -3417,6 +3417,52 @@ TEST_P(WriteBatchWithIndexTest, EntityReadSanityChecks) { } } +TEST_P(WriteBatchWithIndexTest, TrackAndClearCFStats) { + std::string value; + batch_->SetTrackPerCFStat(true); + ASSERT_OK(batch_->Put("A", "val")); + ASSERT_OK(batch_->SingleDelete("B")); + + ColumnFamilyHandleImplDummy cf1(/*id=*/1, BytewiseComparator()); + ASSERT_OK(batch_->Put(&cf1, "bar", "foo")); + + { + auto& cf_id_to_count = batch_->GetCFStats(); + ASSERT_EQ(2, cf_id_to_count.size()); + for (const auto [cf_id, stat] : cf_id_to_count) { + if (cf_id == 0) { + ASSERT_EQ(2, stat.entry_count); + ASSERT_EQ(0, stat.overwritten_sd_count); + } else { + ASSERT_EQ(cf_id, 1); + ASSERT_EQ(1, stat.entry_count); + ASSERT_EQ(0, stat.overwritten_sd_count); + } + } + } + + batch_->Clear(); + ASSERT_TRUE(batch_->GetCFStats().empty()); + + // Now do a version with overwritten SD + ASSERT_OK(batch_->Put("A", "val")); + ASSERT_OK(batch_->SingleDelete("A")); + bool overwrite = GetParam(); + { + auto& cf_id_to_count = batch_->GetCFStats(); + ASSERT_EQ(1, cf_id_to_count.size()); + ASSERT_EQ(overwrite ? 1 : 2, cf_id_to_count.at(0).entry_count); + ASSERT_EQ(0, cf_id_to_count.at(0).overwritten_sd_count); + } + ASSERT_OK(batch_->Put("A", "new_val")); + { + auto& cf_id_to_count = batch_->GetCFStats(); + ASSERT_EQ(1, cf_id_to_count.size()); + ASSERT_EQ(overwrite ? 1 : 3, cf_id_to_count.at(0).entry_count); + ASSERT_EQ(overwrite ? 1 : 0, cf_id_to_count.at(0).overwritten_sd_count); + } +} + INSTANTIATE_TEST_CASE_P(WBWI, WriteBatchWithIndexTest, testing::Bool()); std::string Get(const std::string& k, std::unique_ptr& wbwi_mem, @@ -3453,8 +3499,10 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { ImmutableOptions immutable_opts(opts); MutableCFOptions mutable_cf_options(opts); - Random rnd(301); - auto wbwi = std::make_shared(cmp, 0, true, 0, 0); + Random& rnd = *Random::GetTLSInstance(); + auto wbwi = std::make_shared( + cmp, 0, /*overwrite_key=*/true, 0, 0); + wbwi->SetTrackPerCFStat(true); std::vector> expected; expected.resize(10000); for (int i = 0; i < 10000; ++i) { @@ -3468,12 +3516,14 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { RandomShuffle(expected.begin(), expected.end()); std::unique_ptr wbwi_mem{ new WBWIMemTable(wbwi, cmp, - /*cf_id=*/0, &immutable_opts, &mutable_cf_options)}; + /*cf_id=*/0, &immutable_opts, &mutable_cf_options, + // stats is inaccurate but read path should still work + /*stat=*/{})}; ASSERT_TRUE(wbwi_mem->IsEmpty()); - constexpr SequenceNumber visible_seq = 3; + constexpr SequenceNumber visible_seq = 10002; constexpr SequenceNumber non_visible_seq = 1; - constexpr SequenceNumber assigned_seq = 2; - wbwi_mem->SetGlobalSequenceNumber(assigned_seq); + constexpr WBWIMemTable::SeqnoRange assigned_seq = {2, 10001}; + wbwi_mem->AssignSequenceNumbers(assigned_seq); bool found_final_value = false; for (const auto& [key, val] : expected) { @@ -3501,7 +3551,7 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { RandomShuffle(expected.begin(), expected.end()); // overwrites for (size_t i = 0; i < 2000; ++i) { - // We don't expect mixing SD and DEL, or issue multiple SD consecutively in + // We don't expect to mix SD and DEL, or issue multiple SD consecutively in // a DB. Read from WBWI should still work so we do it here to keep the test // simple. if (rnd.OneIn(2)) { @@ -3540,7 +3590,6 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { ASSERT_TRUE(val == Get(key, wbwi_mem, visible_seq, &found_final_value)); ASSERT_TRUE(found_final_value); } - // MultiGet int batch_size = 30; for (int i = 0; i < 10000; i += batch_size) { @@ -3601,9 +3650,9 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { return a.first < b.first; }); Arena arena; - InternalIterator* iter = - wbwi_mem->NewIterator(ReadOptions(), /*seqno_to_time_mapping=*/nullptr, - &arena, /*prefix_extractor=*/nullptr); + InternalIterator* iter = wbwi_mem->NewIterator( + ReadOptions(), /*seqno_to_time_mapping=*/nullptr, &arena, + /*prefix_extractor=*/nullptr, /*for_flush=*/false); ASSERT_OK(iter->status()); auto verify_iter_at = [&](size_t idx) { @@ -3615,7 +3664,7 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { ValueType val_type; UnPackSequenceAndType(ExtractInternalKeyFooter(iter->key()), &seq, &val_type); - ASSERT_EQ(seq, assigned_seq); + ASSERT_EQ(seq, assigned_seq.upper_bound); if (expected[idx].second == "NOT_FOUND") { ASSERT_TRUE(val_type == kTypeDeletion || val_type == kTypeSingleDeletion); } else { @@ -3639,14 +3688,20 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { iter->Seek(seek_key.GetInternalKey()); } verify_iter_at(key_idx); - uint32_t j = key_idx + 1; - for (; j < std::min(10000u, key_idx + 5); ++j) { + + // verify next/prev 5 times + for (int j = 0; j < 5; ++j) { + if (++key_idx >= 10000) { + --key_idx; + break; + } iter->Next(); - verify_iter_at(j); + verify_iter_at(key_idx); } - for (j -= 2; j >= key_idx; --j) { + for (int j = 0; j < 5; ++j) { iter->Prev(); - verify_iter_at(j); + assert(key_idx >= 1); + verify_iter_at(--key_idx); } } } @@ -3668,8 +3723,9 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { // Read from another CF std::unique_ptr meta_wbwi_mem{new WBWIMemTable( - wbwi, cmp, /*cf_id=*/1, &immutable_opts, &mutable_cf_options)}; - meta_wbwi_mem->SetGlobalSequenceNumber(assigned_seq); + wbwi, cmp, /*cf_id=*/1, &immutable_opts, &mutable_cf_options, + /*stat=*/{1, 0})}; + meta_wbwi_mem->AssignSequenceNumbers(assigned_seq); found_final_value = false; ASSERT_TRUE("foo" == Get(DBTestBase::Key(0), meta_wbwi_mem, visible_seq, &found_final_value)); @@ -3682,7 +3738,8 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { wbwi_mem.reset(); // allocated by arena iter->~InternalIterator(); - iter = meta_wbwi_mem->NewIterator(ReadOptions(), nullptr, &arena, nullptr); + iter = meta_wbwi_mem->NewIterator(ReadOptions(), nullptr, &arena, nullptr, + /*for_flush=*/false); iter->SeekToFirst(); ASSERT_OK(iter->status()); ASSERT_TRUE(iter->Valid()); @@ -3693,6 +3750,64 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) { ASSERT_FALSE(iter->Valid()); iter->~InternalIterator(); } + +TEST_F(WBWIMemTableTest, IterEmitSingleDelete) { + const Comparator* cmp = BytewiseComparator(); + Options opts; + ImmutableOptions immutable_opts(opts); + MutableCFOptions mutable_cf_options(opts); + + auto wbwi = std::make_shared( + cmp, 0, /*overwrite_key=*/true, 0, 0); + wbwi->SetTrackPerCFStat(true); + + ASSERT_OK(wbwi->Put(DBTestBase::Key(0), "val0")); + ASSERT_OK(wbwi->SingleDelete(DBTestBase::Key(0))); + ASSERT_OK(wbwi->SingleDelete(DBTestBase::Key(1))); + ASSERT_OK(wbwi->SingleDelete(DBTestBase::Key(2))); + ASSERT_OK(wbwi->Put(DBTestBase::Key(3), "val3")); + // SD at key1 overwritten + ASSERT_OK(wbwi->Put(DBTestBase::Key(1), "val1")); + + std::unique_ptr wbwi_mem{ + new WBWIMemTable(wbwi, cmp, + /*cf_id=*/0, &immutable_opts, &mutable_cf_options, + /*stat=*/wbwi->GetCFStats().at(0))}; + WBWIMemTable::SeqnoRange assigned_seqno = { + 1, 1 + wbwi->GetWriteBatch()->Count()}; + wbwi_mem->AssignSequenceNumbers(assigned_seqno); + Arena arena; + InternalIterator* iter_for_flush = wbwi_mem->NewIterator( + ReadOptions(), /*seqno_to_time_mapping=*/nullptr, &arena, + /*prefix_extractor=*/nullptr, /*for_flush=*/true); + InternalIterator* iter = wbwi_mem->NewIterator( + ReadOptions(), /*seqno_to_time_mapping=*/nullptr, &arena, + /*prefix_extractor=*/nullptr, /*for_flush=*/false); + iter->SeekToFirst(); + iter_for_flush->SeekToFirst(); + for (int i = 0; i < 4; ++i) { + ASSERT_TRUE(iter->Valid()); + ASSERT_TRUE(iter_for_flush->Valid()); + ASSERT_EQ(iter->key(), iter_for_flush->key()); + + iter->Next(); + iter_for_flush->Next(); + if (i == 1) { + // overwritten SD at key1 + // See WBWIMemTableIterator::UpdateSingleDeleteKey() for seqno assignment + InternalKey ikey(DBTestBase::Key(1), assigned_seqno.upper_bound - 1, + kTypeSingleDeletion); + ASSERT_EQ(ikey.Encode(), iter_for_flush->key()); + iter_for_flush->Next(); + } + } + ASSERT_FALSE(iter->Valid()); + ASSERT_FALSE(iter_for_flush->Valid()); + ASSERT_OK(iter->status()); + ASSERT_OK(iter_for_flush->status()); + iter->~InternalIterator(); + iter_for_flush->~InternalIteratorBase(); +} } // namespace ROCKSDB_NAMESPACE int main(int argc, char** argv) {