From e3426d30e2d007819a2a2721b1b208fc436818ff Mon Sep 17 00:00:00 2001 From: gengliqi Date: Wed, 5 Aug 2026 13:44:41 +0800 Subject: [PATCH 01/10] fix ingest allow write Signed-off-by: gengliqi --- db/db_impl/db_impl.cc | 85 ++++++++++++++++++++------- db/external_sst_file_ingestion_job.cc | 7 +-- db/external_sst_file_ingestion_job.h | 10 +++- db/external_sst_file_test.cc | 57 ++++++++++++++---- 4 files changed, 119 insertions(+), 40 deletions(-) diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 2041aeac9870..8caf029c6115 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -5928,9 +5928,45 @@ Status DBImpl::IngestExternalFiles( } // Run ingestion jobs. if (status.ok()) { + if (allow_write) { + // Briefly stop writes while reserving sequence numbers for ingestion. + write_thread_.EnterUnbatched(&w, &mutex_); + if (two_write_queues_) { + nonmem_write_thread_.EnterUnbatched(&nonmem_w, &mutex_); + } + WaitForPendingWrites(); + } + + SequenceNumber last_seqno = versions_->LastSequence(); + if (allow_write) { + SequenceNumber reserved_seqno_count = 0; + for (size_t i = 0; i != num_cfs; ++i) { + reserved_seqno_count = + std::max(reserved_seqno_count, + static_cast( + ingestion_jobs[i].files_to_ingest().size())); + } + assert(reserved_seqno_count > 0); + assert(last_seqno <= kMaxSequenceNumber - reserved_seqno_count); + const SequenceNumber reserved_last_seqno = + last_seqno + reserved_seqno_count; + versions_->SetLastAllocatedSequence(reserved_last_seqno); + versions_->SetLastPublishedSequence(reserved_last_seqno); + versions_->SetLastSequence(reserved_last_seqno); + + if (two_write_queues_) { + nonmem_write_thread_.ExitUnbatched(&nonmem_w); + } + write_thread_.ExitUnbatched(&w); + + // The reservation cannot be rolled back if ingestion fails because a + // foreground write may have already consumed a later sequence number. + TEST_SYNC_POINT("DBImpl::IngestExternalFiles:AfterReserveSeqno"); + } + for (size_t i = 0; i != num_cfs; ++i) { mutex_.AssertHeld(); - status = ingestion_jobs[i].Run(); + status = ingestion_jobs[i].Run(last_seqno); if (!status.ok()) { break; } @@ -5964,29 +6000,36 @@ Status DBImpl::IngestExternalFiles( } assert(0 == num_entries); } + // With allow_write, foreground flushes may advance the MANIFEST sequence + // after sequence numbers are reserved. LogAndApplyHelper keeps + // VersionEdit sequence numbers non-decreasing in the MANIFEST. status = versions_->LogAndApply(cfds_to_commit, mutable_cf_options_list, read_options, edit_lists, &mutex_, directories_.GetDbDir()); - // It is safe to update VersionSet last seqno here after LogAndApply since - // LogAndApply persists last sequence number from VersionEdits, - // which are from file's largest seqno and not from VersionSet. - // - // It is necessary to update last seqno here since LogAndApply releases - // mutex when persisting MANIFEST file, and the snapshots taken during - // that period will not be stable if VersionSet last seqno is updated - // before LogAndApply. - int consumed_seqno_count = - ingestion_jobs[0].ConsumedSequenceNumbersCount(); - for (size_t i = 1; i != num_cfs; ++i) { - consumed_seqno_count = - std::max(consumed_seqno_count, - ingestion_jobs[i].ConsumedSequenceNumbersCount()); - } - if (consumed_seqno_count > 0) { - const SequenceNumber last_seqno = versions_->LastSequence(); - versions_->SetLastAllocatedSequence(last_seqno + consumed_seqno_count); - versions_->SetLastPublishedSequence(last_seqno + consumed_seqno_count); - versions_->SetLastSequence(last_seqno + consumed_seqno_count); + if (!allow_write) { + // It is safe to update VersionSet last seqno here after LogAndApply + // since LogAndApply persists last sequence number from VersionEdits, + // which are from file's largest seqno and not from VersionSet. + // + // It is necessary to update last seqno here since LogAndApply releases + // mutex when persisting MANIFEST file, and the snapshots taken during + // that period will not be stable if VersionSet last seqno is updated + // before LogAndApply. + int consumed_seqno_count = + ingestion_jobs[0].ConsumedSequenceNumbersCount(); + for (size_t i = 1; i != num_cfs; ++i) { + consumed_seqno_count = + std::max(consumed_seqno_count, + ingestion_jobs[i].ConsumedSequenceNumbersCount()); + } + if (consumed_seqno_count > 0) { + const SequenceNumber last_seqno = versions_->LastSequence(); + versions_->SetLastAllocatedSequence(last_seqno + + consumed_seqno_count); + versions_->SetLastPublishedSequence(last_seqno + + consumed_seqno_count); + versions_->SetLastSequence(last_seqno + consumed_seqno_count); + } } } diff --git a/db/external_sst_file_ingestion_job.cc b/db/external_sst_file_ingestion_job.cc index 3487de8b3f95..2abbf4f09228 100644 --- a/db/external_sst_file_ingestion_job.cc +++ b/db/external_sst_file_ingestion_job.cc @@ -372,9 +372,7 @@ Status ExternalSstFileIngestionJob::NeedsFlush(bool* flush_needed, return status; } -// REQUIRES: we have become the only writer by entering both write_thread_ and -// nonmem_write_thread_ -Status ExternalSstFileIngestionJob::Run() { +Status ExternalSstFileIngestionJob::Run(SequenceNumber last_seqno) { Status status; SuperVersion* super_version = cfd_->GetSuperVersion(); #ifndef NDEBUG @@ -398,9 +396,6 @@ Status ExternalSstFileIngestionJob::Run() { // if the don't overlap with any ranges since we have snapshots force_global_seqno = true; } - // It is safe to use this instead of LastAllocatedSequence since we are - // the only active writer, and hence they are equal - SequenceNumber last_seqno = versions_->LastSequence(); edit_.SetColumnFamily(cfd_->GetID()); // The levels that the files will be ingested into diff --git a/db/external_sst_file_ingestion_job.h b/db/external_sst_file_ingestion_job.h index 49bb1e31e59f..6488d23a7f83 100644 --- a/db/external_sst_file_ingestion_job.h +++ b/db/external_sst_file_ingestion_job.h @@ -131,8 +131,16 @@ class ExternalSstFileIngestionJob { Status NeedsFlush(bool* flush_needed, SuperVersion* super_version); // Will execute the ingestion job and prepare edit() to be applied. + // + // If `allow_write` is false, foreground writes must remain blocked while + // this job runs. Otherwise, the caller must ensure that concurrent writes do + // not overlap the ingested key ranges. + // + // `last_seqno` must immediately precede the sequence numbers available to + // this job. If foreground writes are allowed, the caller must reserve enough + // sequence numbers before resuming them. // REQUIRES: Mutex held - Status Run(); + Status Run(SequenceNumber last_seqno); // Register key range involved in this ingestion job // to prevent key range conflict with other ongoing compaction/file ingestion diff --git a/db/external_sst_file_test.cc b/db/external_sst_file_test.cc index f86af51e2a41..ac30de213dfa 100644 --- a/db/external_sst_file_test.cc +++ b/db/external_sst_file_test.cc @@ -2394,24 +2394,57 @@ TEST_P(ExternalSSTFileTest, IngestBehind) { TEST_P(ExternalSSTFileTest, WriteDuringIngest) { SyncPoint::GetInstance()->DisableProcessing(); - // Set callback to simulate concurrent write during ingestion - SyncPoint::GetInstance()->SetCallBack( - "DBImpl::IngestExternalFile:AfterAllowWriteCheck", [&](void*) { - // Write a non-overlapping key - ASSERT_OK(Put("foo", "v1")); - }); - Options options = CurrentOptions(); + options.enable_multi_batch_write = true; DestroyAndReopen(options); + std::vector external_files; + for (const std::string& key : {"foo1", "foo2"}) { + std::string file_path = sst_files_dir_ + env_->GenerateUniqueId(); + SstFileWriter writer(EnvOptions(), options); + ASSERT_OK(writer.Open(file_path)); + ASSERT_OK(writer.Put(key, "v1")); + ASSERT_OK(writer.Finish()); + external_files.push_back(std::move(file_path)); + } + + const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); + const Snapshot* snapshot = db_->GetSnapshot(); + std::vector assigned_seqnos; + + // Write after the ingestion sequence number has been reserved but before it + // is assigned to the ingested file. + SyncPoint::GetInstance()->SetCallBack( + "DBImpl::IngestExternalFiles:AfterReserveSeqno", + [&](void*) { ASSERT_OK(Put("bar", "v1")); }); + SyncPoint::GetInstance()->SetCallBack( + "ExternalSstFileIngestionJob::Run", [&](void* arg) { + ASSERT_NE(arg, nullptr); + assigned_seqnos.push_back(*static_cast(arg)); + }); + SyncPoint::GetInstance()->EnableProcessing(); - ASSERT_OK(GenerateAndAddExternalFile(options, {{"foo", "v1"}}, -1, true, - false, true, false, false, - true /* allow_write */)); - ASSERT_OK(Put("bar", "v1")); - ASSERT_EQ(Get("foo"), "v1"); + IngestExternalFileOptions ifo; + ifo.allow_global_seqno = true; + ifo.write_global_seqno = std::get<0>(GetParam()); + ifo.verify_checksums_before_ingest = std::get<1>(GetParam()); + ifo.allow_write = true; + ASSERT_OK(db_->IngestExternalFile(external_files, ifo)); + + ASSERT_EQ((std::vector{last_seqno + 1, last_seqno + 2}), + assigned_seqnos); + ASSERT_EQ(last_seqno + 3, db_->GetLatestSequenceNumber()); + ASSERT_EQ(Get("foo1"), "v1"); + ASSERT_EQ(Get("foo2"), "v1"); ASSERT_EQ(Get("bar"), "v1"); + ReadOptions read_options; + read_options.snapshot = snapshot; + std::string value; + ASSERT_TRUE(db_->Get(read_options, "foo1", &value).IsNotFound()); + ASSERT_TRUE(db_->Get(read_options, "foo2", &value).IsNotFound()); + db_->ReleaseSnapshot(snapshot); + SyncPoint::GetInstance()->DisableProcessing(); SyncPoint::GetInstance()->ClearAllCallBacks(); } From 394c101c09b7fd69f5f3653d84e14242834c9399 Mon Sep 17 00:00:00 2001 From: gengliqi Date: Wed, 5 Aug 2026 16:37:09 +0800 Subject: [PATCH 02/10] add more comments Signed-off-by: gengliqi --- db/db_impl/db_impl.cc | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 8caf029c6115..05cc039d784e 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -5938,7 +5938,11 @@ Status DBImpl::IngestExternalFiles( } SequenceNumber last_seqno = versions_->LastSequence(); + if (allow_write) { + // Each file consumes at most one sequence number. Jobs for different + // column families share the same sequence range, so reserve the maximum + // file count. Unused sequence numbers are harmless gaps. SequenceNumber reserved_seqno_count = 0; for (size_t i = 0; i != num_cfs; ++i) { reserved_seqno_count = @@ -5947,7 +5951,6 @@ Status DBImpl::IngestExternalFiles( ingestion_jobs[i].files_to_ingest().size())); } assert(reserved_seqno_count > 0); - assert(last_seqno <= kMaxSequenceNumber - reserved_seqno_count); const SequenceNumber reserved_last_seqno = last_seqno + reserved_seqno_count; versions_->SetLastAllocatedSequence(reserved_last_seqno); From 02a4f9a79a8e45f262cbecb55eab9416563d8df7 Mon Sep 17 00:00:00 2001 From: gengliqi Date: Wed, 5 Aug 2026 17:02:30 +0800 Subject: [PATCH 03/10] add a regression unit test Signed-off-by: gengliqi --- db/external_sst_file_ingestion_job.h | 8 ---- db/external_sst_file_test.cc | 63 ++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 8 deletions(-) diff --git a/db/external_sst_file_ingestion_job.h b/db/external_sst_file_ingestion_job.h index 6488d23a7f83..d915ed25d566 100644 --- a/db/external_sst_file_ingestion_job.h +++ b/db/external_sst_file_ingestion_job.h @@ -131,14 +131,6 @@ class ExternalSstFileIngestionJob { Status NeedsFlush(bool* flush_needed, SuperVersion* super_version); // Will execute the ingestion job and prepare edit() to be applied. - // - // If `allow_write` is false, foreground writes must remain blocked while - // this job runs. Otherwise, the caller must ensure that concurrent writes do - // not overlap the ingested key ranges. - // - // `last_seqno` must immediately precede the sequence numbers available to - // this job. If foreground writes are allowed, the caller must reserve enough - // sequence numbers before resuming them. // REQUIRES: Mutex held Status Run(SequenceNumber last_seqno); diff --git a/db/external_sst_file_test.cc b/db/external_sst_file_test.cc index ac30de213dfa..95819bc88577 100644 --- a/db/external_sst_file_test.cc +++ b/db/external_sst_file_test.cc @@ -2449,6 +2449,69 @@ TEST_P(ExternalSSTFileTest, WriteDuringIngest) { SyncPoint::GetInstance()->ClearAllCallBacks(); } +TEST_P(ExternalSSTFileTest, AllowWriteIngestWaitsForPendingWriter) { + constexpr auto kReleaseWriter = + "ExternalSSTFileTest::AllowWriteIngestWaitsForPendingWriter:" + "ReleaseWriter"; + constexpr auto kWriterPrepared = + "ExternalSSTFileTest::AllowWriteIngestWaitsForPendingWriter:" + "WriterPrepared"; + constexpr auto kStartIngest = + "ExternalSSTFileTest::AllowWriteIngestWaitsForPendingWriter:" + "StartIngest"; + constexpr auto kContinueWriter = + "ExternalSSTFileTest::AllowWriteIngestWaitsForPendingWriter:" + "ContinueWriter"; + + auto* sync_point = SyncPoint::GetInstance(); + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); + sync_point->LoadDependency( + {{kWriterPrepared, kStartIngest}, {kReleaseWriter, kContinueWriter}}); + + sync_point->SetCallBack("DBImpl::WriteImpl:BeforeLeaderEnters", [&](void*) { + TEST_SYNC_POINT(kWriterPrepared); + TEST_SYNC_POINT(kContinueWriter); + }); + + const auto release_writer = [&](void*) { TEST_SYNC_POINT(kReleaseWriter); }; + sync_point->SetCallBack("WriteThread::EnterUnbatched:Wait", release_writer); + + SequenceNumber assigned_seqno = 0; + sync_point->SetCallBack("ExternalSstFileIngestionJob::Run", [&](void* arg) { + release_writer(nullptr); + ASSERT_NE(arg, nullptr); + assigned_seqno = *static_cast(arg); + }); + + Options options = CurrentOptions(); + options.enable_multi_batch_write = false; + DestroyAndReopen(options); + + const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); + const Snapshot* snapshot = db_->GetSnapshot(); + Status write_status; + + sync_point->EnableProcessing(); + port::Thread writer([&]() { write_status = Put("bar", "v1"); }); + + TEST_SYNC_POINT(kStartIngest); + ASSERT_OK(GenerateAndAddExternalFile( + options, {{"foo", "v"}}, -1, true, std::get<0>(GetParam()), + std::get<1>(GetParam()), false, false, true /* allow_write */)); + writer.join(); + + ASSERT_OK(write_status); + ASSERT_EQ(last_seqno + 2, assigned_seqno); + ASSERT_EQ(last_seqno + 2, db_->GetLatestSequenceNumber()); + ASSERT_EQ("v1", Get("bar")); + ASSERT_EQ("v", Get("foo")); + db_->ReleaseSnapshot(snapshot); + + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); +} + TEST_P(ExternalSSTFileTest, InconsistentAllowWriteArguments) { Options options = CurrentOptions(); CreateAndReopenWithCF({"koko"}, options); From 3d9a6b9cebd61cb74c3da8d881e7bcc8ed0d6e82 Mon Sep 17 00:00:00 2001 From: gengliqi Date: Wed, 5 Aug 2026 17:06:48 +0800 Subject: [PATCH 04/10] update comment Signed-off-by: gengliqi --- db/db_impl/db_impl.cc | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 05cc039d784e..74bca9974c02 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -6003,9 +6003,10 @@ Status DBImpl::IngestExternalFiles( } assert(0 == num_entries); } - // With allow_write, foreground flushes may advance the MANIFEST sequence - // after sequence numbers are reserved. LogAndApplyHelper keeps - // VersionEdit sequence numbers non-decreasing in the MANIFEST. + // With allow_write, a concurrent flush may persist a higher last sequence + // before this ingestion edit is applied. LogAndApplyHelper raises this edit's + // last sequence as needed to keep VersionEdit::last_sequence values + // non-decreasing in the MANIFEST. status = versions_->LogAndApply(cfds_to_commit, mutable_cf_options_list, read_options, edit_lists, &mutex_, directories_.GetDbDir()); From 391be8a98bbc2b601b01b6a3b18206c702dc6208 Mon Sep 17 00:00:00 2001 From: gengliqi Date: Wed, 5 Aug 2026 17:38:12 +0800 Subject: [PATCH 05/10] format Signed-off-by: gengliqi --- db/db_impl/db_impl.cc | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 74bca9974c02..2aba360f2a69 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -6004,9 +6004,9 @@ Status DBImpl::IngestExternalFiles( assert(0 == num_entries); } // With allow_write, a concurrent flush may persist a higher last sequence - // before this ingestion edit is applied. LogAndApplyHelper raises this edit's - // last sequence as needed to keep VersionEdit::last_sequence values - // non-decreasing in the MANIFEST. + // before this ingestion edit is applied. LogAndApplyHelper raises this + // edit's last sequence as needed to keep VersionEdit::last_sequence + // values non-decreasing in the MANIFEST. status = versions_->LogAndApply(cfds_to_commit, mutable_cf_options_list, read_options, edit_lists, &mutex_, directories_.GetDbDir()); From 3235185420dfd6a165576e0c72f691c5227d7b45 Mon Sep 17 00:00:00 2001 From: gengliqi Date: Wed, 5 Aug 2026 18:57:19 +0800 Subject: [PATCH 06/10] add more comments Signed-off-by: gengliqi --- db/db_impl/db_impl.cc | 8 ++++---- include/rocksdb/options.h | 6 +++++- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 2aba360f2a69..e3d93d8e161b 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -5956,7 +5956,7 @@ Status DBImpl::IngestExternalFiles( versions_->SetLastAllocatedSequence(reserved_last_seqno); versions_->SetLastPublishedSequence(reserved_last_seqno); versions_->SetLastSequence(reserved_last_seqno); - + // Resume writes if (two_write_queues_) { nonmem_write_thread_.ExitUnbatched(&nonmem_w); } @@ -6004,9 +6004,9 @@ Status DBImpl::IngestExternalFiles( assert(0 == num_entries); } // With allow_write, a concurrent flush may persist a higher last sequence - // before this ingestion edit is applied. LogAndApplyHelper raises this - // edit's last sequence as needed to keep VersionEdit::last_sequence - // values non-decreasing in the MANIFEST. + // before this ingestion edit is applied. LogAndApply updates the edit as + // needed to keep VersionEdit::last_sequence non-decreasing in the + // MANIFEST. status = versions_->LogAndApply(cfds_to_commit, mutable_cf_options_list, read_options, edit_lists, &mutex_, directories_.GetDbDir()); diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index 312c65fc2566..2fb4fa18e96a 100644 --- a/include/rocksdb/options.h +++ b/include/rocksdb/options.h @@ -2101,7 +2101,11 @@ struct IngestExternalFileOptions { // ingest_behind takes precedence over fail_if_not_bottommost_level. bool fail_if_not_bottommost_level = false; // Set to TRUE if user wants to allow writes to the DB during ingestion. - // User must ensure no writes overlap with the ingested data. + // User must ensure that concurrent writes do not overlap the ingested key + // ranges. + // Reads using snapshots created before ingestion are allowed. A snapshot + // created while ingestion is in progress must not read the ingested key + // ranges before ingestion completes. bool allow_write = false; }; From 30f72f11a73b395b9cd81a823e5fd5076ffd977a Mon Sep 17 00:00:00 2001 From: gengliqi Date: Wed, 5 Aug 2026 23:25:12 +0800 Subject: [PATCH 07/10] update test Signed-off-by: gengliqi --- db/external_sst_file_test.cc | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/db/external_sst_file_test.cc b/db/external_sst_file_test.cc index 95819bc88577..e2367fdd15af 100644 --- a/db/external_sst_file_test.cc +++ b/db/external_sst_file_test.cc @@ -2411,12 +2411,19 @@ TEST_P(ExternalSSTFileTest, WriteDuringIngest) { const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); const Snapshot* snapshot = db_->GetSnapshot(); std::vector assigned_seqnos; + Status write_status; + std::unique_ptr write_thread; - // Write after the ingestion sequence number has been reserved but before it - // is assigned to the ingested file. + // Start a foreground write after the ingestion sequence numbers have been + // reserved. The callback runs with the DB mutex held, so run Put in another + // thread and join it after ingestion to avoid a mutex deadlock. SyncPoint::GetInstance()->SetCallBack( - "DBImpl::IngestExternalFiles:AfterReserveSeqno", - [&](void*) { ASSERT_OK(Put("bar", "v1")); }); + "DBImpl::IngestExternalFiles:AfterReserveSeqno", [&](void*) { + ASSERT_EQ(last_seqno + external_files.size(), + db_->GetLatestSequenceNumber()); + write_thread = std::make_unique( + [&] { write_status = Put("bar", "v1"); }); + }); SyncPoint::GetInstance()->SetCallBack( "ExternalSstFileIngestionJob::Run", [&](void* arg) { ASSERT_NE(arg, nullptr); @@ -2429,7 +2436,12 @@ TEST_P(ExternalSSTFileTest, WriteDuringIngest) { ifo.write_global_seqno = std::get<0>(GetParam()); ifo.verify_checksums_before_ingest = std::get<1>(GetParam()); ifo.allow_write = true; - ASSERT_OK(db_->IngestExternalFile(external_files, ifo)); + Status ingest_status = db_->IngestExternalFile(external_files, ifo); + + ASSERT_NE(write_thread, nullptr); + write_thread->join(); + ASSERT_OK(ingest_status); + ASSERT_OK(write_status); ASSERT_EQ((std::vector{last_seqno + 1, last_seqno + 2}), assigned_seqnos); From 7528f73c69875ba38ef8a807c828fcff218719d8 Mon Sep 17 00:00:00 2001 From: gengliqi Date: Thu, 6 Aug 2026 11:43:22 +0800 Subject: [PATCH 08/10] address comments Signed-off-by: gengliqi --- db/db_impl/db_impl.cc | 7 +++++-- include/rocksdb/options.h | 7 ++++--- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index e3d93d8e161b..a318c11b2ff2 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -5938,12 +5938,11 @@ Status DBImpl::IngestExternalFiles( } SequenceNumber last_seqno = versions_->LastSequence(); - + SequenceNumber reserved_seqno_count = 0; if (allow_write) { // Each file consumes at most one sequence number. Jobs for different // column families share the same sequence range, so reserve the maximum // file count. Unused sequence numbers are harmless gaps. - SequenceNumber reserved_seqno_count = 0; for (size_t i = 0; i != num_cfs; ++i) { reserved_seqno_count = std::max(reserved_seqno_count, @@ -5973,6 +5972,10 @@ Status DBImpl::IngestExternalFiles( if (!status.ok()) { break; } + assert(!allow_write || + static_cast( + ingestion_jobs[i].ConsumedSequenceNumbersCount()) <= + reserved_seqno_count); ingestion_jobs[i].RegisterRange(); } } diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index 2fb4fa18e96a..0c35cb9424cd 100644 --- a/include/rocksdb/options.h +++ b/include/rocksdb/options.h @@ -2103,9 +2103,10 @@ struct IngestExternalFileOptions { // Set to TRUE if user wants to allow writes to the DB during ingestion. // User must ensure that concurrent writes do not overlap the ingested key // ranges. - // Reads using snapshots created before ingestion are allowed. A snapshot - // created while ingestion is in progress must not read the ingested key - // ranges before ingestion completes. + // Snapshot consistency is not guaranteed for snapshots created during + // ingestion because ingestion sequence numbers are published before the + // ingested files become visible. Such snapshots must not be used to read the + // ingested key ranges. bool allow_write = false; }; From 89d09be144480d66d5fee74b18e178f65fda94b5 Mon Sep 17 00:00:00 2001 From: gengliqi Date: Fri, 7 Aug 2026 01:15:43 +0800 Subject: [PATCH 09/10] add a unit test Signed-off-by: gengliqi --- db/external_sst_file_test.cc | 75 +++++++++++++++++++++++++++++++++--- 1 file changed, 69 insertions(+), 6 deletions(-) diff --git a/db/external_sst_file_test.cc b/db/external_sst_file_test.cc index e2367fdd15af..eb8da49fde9f 100644 --- a/db/external_sst_file_test.cc +++ b/db/external_sst_file_test.cc @@ -2461,18 +2461,18 @@ TEST_P(ExternalSSTFileTest, WriteDuringIngest) { SyncPoint::GetInstance()->ClearAllCallBacks(); } -TEST_P(ExternalSSTFileTest, AllowWriteIngestWaitsForPendingWriter) { +TEST_P(ExternalSSTFileTest, AllowWriteIngestWaitsForActiveWriter) { constexpr auto kReleaseWriter = - "ExternalSSTFileTest::AllowWriteIngestWaitsForPendingWriter:" + "ExternalSSTFileTest::AllowWriteIngestWaitsForActiveWriter:" "ReleaseWriter"; constexpr auto kWriterPrepared = - "ExternalSSTFileTest::AllowWriteIngestWaitsForPendingWriter:" + "ExternalSSTFileTest::AllowWriteIngestWaitsForActiveWriter:" "WriterPrepared"; constexpr auto kStartIngest = - "ExternalSSTFileTest::AllowWriteIngestWaitsForPendingWriter:" + "ExternalSSTFileTest::AllowWriteIngestWaitsForActiveWriter:" "StartIngest"; constexpr auto kContinueWriter = - "ExternalSSTFileTest::AllowWriteIngestWaitsForPendingWriter:" + "ExternalSSTFileTest::AllowWriteIngestWaitsForActiveWriter:" "ContinueWriter"; auto* sync_point = SyncPoint::GetInstance(); @@ -2491,8 +2491,9 @@ TEST_P(ExternalSSTFileTest, AllowWriteIngestWaitsForPendingWriter) { SequenceNumber assigned_seqno = 0; sync_point->SetCallBack("ExternalSstFileIngestionJob::Run", [&](void* arg) { + // Release the writer here as a fallback so that a failure to wait in + // EnterUnbatched causes an assertion failure instead of a deadlock. release_writer(nullptr); - ASSERT_NE(arg, nullptr); assigned_seqno = *static_cast(arg); }); @@ -2524,6 +2525,68 @@ TEST_P(ExternalSSTFileTest, AllowWriteIngestWaitsForPendingWriter) { sync_point->ClearAllCallBacks(); } +TEST_F(ExternalSSTFileTest, AllowWriteIngestWaitsForPendingMultiBatchWrite) { + constexpr auto kStartIngest = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForPendingMultiBatchWrite:StartIngest"; + constexpr auto kReleaseWriter = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForPendingMultiBatchWrite:ReleaseWriter"; + + auto* sync_point = SyncPoint::GetInstance(); + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); + sync_point->LoadDependency( + {{"DBImpl::WriteImpl:CommitAfterWriteWAL", kStartIngest}, + {kReleaseWriter, "DBImpl::WriteImpl:BeforePipelineWriteMemtable"}}); + + const auto release_writer = [&](void*) { TEST_SYNC_POINT(kReleaseWriter); }; + sync_point->SetCallBack("DBImpl::WaitForPendingWrites:BeforeBlock", + release_writer); + + SequenceNumber assigned_seqno = 0; + sync_point->SetCallBack("ExternalSstFileIngestionJob::Run", [&](void* arg) { + assigned_seqno = *static_cast(arg); + }); + + Options options = CurrentOptions(); + options.enable_pipelined_write = false; + options.unordered_write = false; + options.enable_multi_batch_write = true; + DestroyAndReopen(options); + + const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); + // Force the ingested file to consume a global sequence number. + const Snapshot* snapshot = db_->GetSnapshot(); + Status write_status; + + sync_point->EnableProcessing(); + port::Thread writer([&]() { write_status = Put("bar", "v1"); }); + + // The writer has allocated a sequence and released write_thread_, but has + // not inserted into the memtable or published the sequence yet. + TEST_SYNC_POINT(kStartIngest); + Status ingest_status = + GenerateAndAddExternalFile(options, {{"foo", "v"}}, -1, true, false, true, + false, false, true /* allow_write */); + + // If ingestion does not wait for the pending writer, release it here so the + // sequence assertions fail instead of hanging in writer.join(). + release_writer(nullptr); + writer.join(); + + ASSERT_OK(ingest_status); + ASSERT_OK(write_status); + ASSERT_EQ(last_seqno + 2, assigned_seqno); + ASSERT_EQ(last_seqno + 2, db_->GetLatestSequenceNumber()); + ASSERT_EQ("v1", Get("bar")); + ASSERT_EQ("v", Get("foo")); + db_->ReleaseSnapshot(snapshot); + + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); +} + TEST_P(ExternalSSTFileTest, InconsistentAllowWriteArguments) { Options options = CurrentOptions(); CreateAndReopenWithCF({"koko"}, options); From 42ebac30e2196ecacbf58609062de34b65444ac6 Mon Sep 17 00:00:00 2001 From: gengliqi Date: Mon, 10 Aug 2026 16:56:20 +0800 Subject: [PATCH 10/10] fix multi-batch pending_memtable_writes bug Signed-off-by: gengliqi --- db/db_impl/db_impl.h | 7 ++-- db/db_impl/db_impl_write.cc | 7 ++-- db/external_sst_file_test.cc | 73 ++++++++++++++++++++++++++++++++++++ 3 files changed, 81 insertions(+), 6 deletions(-) diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index c75c8c33a774..fed4cef867c1 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -2045,9 +2045,10 @@ class DBImpl : public DB { return; } - // Wait for the ones who already wrote to the WAL to finish their - // memtable write. + // Wait for writers that have allocated sequence numbers to finish their + // memtable writes and publish their sequences. if (pending_memtable_writes_.load() != 0) { + TEST_SYNC_POINT("DBImpl::WaitForPendingWrites:PendingWrites"); std::unique_lock guard(switch_mutex_); switch_cv_.wait(guard, [&] { return pending_memtable_writes_.load() == 0; }); @@ -2719,7 +2720,7 @@ class DBImpl : public DB { std::condition_variable switch_cv_; // The mutex used by switch_cv_. mutex_ should be acquired beforehand. std::mutex switch_mutex_; - // Number of threads intending to write to memtable + // Number of writers with pending memtable writes or sequence publication. std::atomic pending_memtable_writes_ = {}; // A flag indicating whether the current rocksdb database has any diff --git a/db/db_impl/db_impl_write.cc b/db/db_impl/db_impl_write.cc index c74ec8dba933..762462a8b425 100644 --- a/db/db_impl/db_impl_write.cc +++ b/db/db_impl/db_impl_write.cc @@ -366,13 +366,14 @@ Status DBImpl::MultiBatchWriteImpl(const WriteOptions& write_options, const ReadOptions read_options; writer.status = ApplyWALToManifest(read_options, &synced_wals); } - if (writer.status.ok()) { - pending_memtable_writes_ += memtable_write_cnt; - } else { + if (!writer.status.ok()) { // The `pending_wb_cnt` must be reset to avoid other writers helping // the front writer write its WBs after it failed to write the WAL. writer.ResetPendingWBCnt(); } + // Every writer in the commit queue calls MultiBatchWriteCommit, including + // when the write fails before its memtable write. + pending_memtable_writes_ += memtable_write_cnt; write_thread_.ExitAsBatchGroupLeader(wal_write_group, writer.status); } diff --git a/db/external_sst_file_test.cc b/db/external_sst_file_test.cc index eb8da49fde9f..411dc678c325 100644 --- a/db/external_sst_file_test.cc +++ b/db/external_sst_file_test.cc @@ -2587,6 +2587,79 @@ TEST_F(ExternalSSTFileTest, AllowWriteIngestWaitsForPendingMultiBatchWrite) { sync_point->ClearAllCallBacks(); } +TEST_F(ExternalSSTFileTest, AllowWriteIngestWaitsForFailedMultiBatchWrite) { + constexpr auto kWriterPrepared = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForFailedMultiBatchWrite:WriterPrepared"; + constexpr auto kStartIngest = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForFailedMultiBatchWrite:StartIngest"; + constexpr auto kReleaseWriter = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForFailedMultiBatchWrite:ReleaseWriter"; + constexpr auto kContinueWriter = + "ExternalSSTFileTest::" + "AllowWriteIngestWaitsForFailedMultiBatchWrite:ContinueWriter"; + + auto* sync_point = SyncPoint::GetInstance(); + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); + sync_point->LoadDependency( + {{kWriterPrepared, kStartIngest}, {kReleaseWriter, kContinueWriter}}); + + sync_point->SetCallBack("DBImpl::WriteImpl:CommitAfterWriteWAL", [&](void*) { + TEST_SYNC_POINT(kWriterPrepared); + TEST_SYNC_POINT(kContinueWriter); + }); + + const auto release_writer = [&](void*) { TEST_SYNC_POINT(kReleaseWriter); }; + sync_point->SetCallBack("DBImpl::WaitForPendingWrites:PendingWrites", + release_writer); + + SequenceNumber assigned_seqno = 0; + sync_point->SetCallBack("ExternalSstFileIngestionJob::Run", [&](void* arg) { + // Release the writer here as a fallback so that a failure to wait for it + // causes the sequence assertions to fail instead of a deadlock. + release_writer(nullptr); + assigned_seqno = *static_cast(arg); + }); + + Options options = CurrentOptions(); + options.enable_pipelined_write = false; + options.unordered_write = false; + options.enable_multi_batch_write = true; + // Keep the DB usable after the injected WAL error so ingestion can proceed. + options.paranoid_checks = false; + DestroyAndReopen(options); + + const SequenceNumber last_seqno = db_->GetLatestSequenceNumber(); + // Force the ingested file to consume a global sequence number. + const Snapshot* snapshot = db_->GetSnapshot(); + Status write_status; + + env_->log_write_error_.store(true, std::memory_order_release); + sync_point->EnableProcessing(); + port::Thread writer([&]() { write_status = Put("bar", "v1"); }); + + TEST_SYNC_POINT(kStartIngest); + env_->log_write_error_.store(false, std::memory_order_release); + Status ingest_status = + GenerateAndAddExternalFile(options, {{"foo", "v"}}, -1, true, false, true, + false, false, true /* allow_write */); + writer.join(); + + ASSERT_OK(ingest_status); + ASSERT_NOK(write_status); + ASSERT_EQ(last_seqno + 2, assigned_seqno); + ASSERT_EQ(last_seqno + 2, db_->GetLatestSequenceNumber()); + ASSERT_EQ("NOT_FOUND", Get("bar")); + ASSERT_EQ("v", Get("foo")); + db_->ReleaseSnapshot(snapshot); + + sync_point->DisableProcessing(); + sync_point->ClearAllCallBacks(); +} + TEST_P(ExternalSSTFileTest, InconsistentAllowWriteArguments) { Options options = CurrentOptions(); CreateAndReopenWithCF({"koko"}, options);