From 4707775ae9d46363157ca7b45a85155712f950d5 Mon Sep 17 00:00:00 2001 From: Xingbo Wang Date: Tue, 12 May 2026 15:29:27 -0700 Subject: [PATCH] Fix GetContext status propagation and blob-backed wide-column merge operands (#14640) Summary: - propagate lower-level read and merge failures through `GetContext` via `read_status`, so `Get` and `GetEntity` preserve the original error instead of synthesizing `Corruption` when blob-backed reads or merge resolution fail - teach `GetMergeOperands` to resolve blob-backed default columns from wide-column entities, covering both the direct base-value path and the merge-plus-base path - add regression coverage for blob-read IO errors during `Get`/`GetEntity` merge resolution and for `GetMergeOperands` on blob-backed wide-column entities - fix the `DBFlushTest.MemPurgeCorrectLogNumberAndSSTFileCreation` test race by waiting for flush callbacks and cleaning up sync points ## Testing - `make db_blob_basic_test -j14` - `/usr/bin/perl -e 'alarm shift; exec ARGV' 60 ./db_blob_basic_test --gtest_filter='DBBlobBasicTest/DBBlobBasicIOErrorTest.GetBlob_IOError/*:DBBlobBasicTest/DBBlobBasicIOErrorTest.GetEntityMergeWithBlobBaseIOError/*'` ## Task T265824017, T265415808 Pull Request resolved: https://github.com/facebook/rocksdb/pull/14640 Reviewed By: anand1976 Differential Revision: D101690700 Pulled By: xingbowang fbshipit-source-id: 2b6fc357b37a01efa72a2d54dcff55be8992f42a --- db/blob/blob_file_partition_manager.cc | 14 +- db/blob/db_blob_basic_test.cc | 41 ++ db/blob/db_blob_index_test.cc | 59 +++ db/db_flush_test.cc | 7 + db/db_impl/db_impl.cc | 208 ++++++++- db/db_impl/db_impl.h | 24 ++ db/db_impl/db_impl_follower.cc | 3 +- db/db_impl/db_impl_readonly.cc | 42 +- db/db_impl/db_impl_secondary.cc | 67 ++- db/db_secondary_test.cc | 197 ++++++++- db/db_test2.cc | 59 +++ db/memtable.cc | 102 ++++- db/memtable.h | 20 +- db/memtable_list.cc | 32 +- db/memtable_list.h | 25 +- db/version_set.cc | 7 +- db/version_set.h | 4 +- db/version_set_test.cc | 2 +- db/wide/db_wide_basic_test.cc | 396 ++++++++++++++++++ db/wide/wide_column_serialization.h | 4 + memtable/wbwi_memtable.cc | 14 +- memtable/wbwi_memtable.h | 6 +- table/get_context.cc | 179 +++++--- table/get_context.h | 17 +- table/sst_file_reader.cc | 4 + table/sst_file_reader_test.cc | 41 ++ .../blob_backed_wide_column_merge_reads.md | 3 + .../merge_reads_preserve_precise_statuses.md | 1 + 28 files changed, 1414 insertions(+), 164 deletions(-) create mode 100644 unreleased_history/bug_fixes/blob_backed_wide_column_merge_reads.md create mode 100644 unreleased_history/bug_fixes/merge_reads_preserve_precise_statuses.md diff --git a/db/blob/blob_file_partition_manager.cc b/db/blob/blob_file_partition_manager.cc index 55bdc2276d..e1e1f071ff 100644 --- a/db/blob/blob_file_partition_manager.cc +++ b/db/blob/blob_file_partition_manager.cc @@ -786,13 +786,19 @@ Status BlobFilePartitionManager::ResolveBlobDirectWriteIndex( } } - Status s = version != nullptr ? Status::Corruption("Invalid blob file number") - : Status::NotFound(); - if (blob_file_cache == nullptr) { - return s; + return version != nullptr ? Status::Corruption("Invalid blob file number") + : Status::NotFound(); } + if (read_options.read_tier == kBlockCacheTier) { + // The direct-write fallback below may need to open the blob file reader, + // which `kBlockCacheTier` forbids. Keep the normal Version-backed path + // above eligible for cache-only hits. + return Status::Incomplete("Cannot read blob(s): no disk I/O allowed"); + } + + Status s; CacheHandleGuard reader; s = blob_file_cache->GetBlobFileReader(read_options, blob_idx.file_number(), &reader, diff --git a/db/blob/db_blob_basic_test.cc b/db/blob/db_blob_basic_test.cc index 57cdbdb817..5f8ecbdb10 100644 --- a/db/blob/db_blob_basic_test.cc +++ b/db/blob/db_blob_basic_test.cc @@ -1620,6 +1620,47 @@ TEST_P(DBBlobBasicIOErrorTest, GetBlob_IOError) { SyncPoint::GetInstance()->ClearAllCallBacks(); } +TEST_P(DBBlobBasicIOErrorTest, GetEntityMergeWithBlobBaseIOError) { + // Goal: verify GetEntity preserves injected blob-read IOErrors when merge + // reads a blob-backed base value, instead of laundering them into Corruption. + // The test writes a blob-backed base value plus a merge operand, then injects + // an IOError at blob read time and checks both GetEntity and Get see it. + Options options; + options.env = fault_injection_env_.get(); + options.enable_blob_files = true; + options.min_blob_size = 0; + options.merge_operator = MergeOperators::CreateStringAppendOperator(); + + Reopen(options); + + constexpr char key[] = "key"; + constexpr char base_value[] = "base_value"; + + ASSERT_OK(Put(key, base_value)); + ASSERT_OK(Flush()); + + ASSERT_OK(Merge(key, "merge_operand")); + ASSERT_OK(Flush()); + + SyncPoint::GetInstance()->SetCallBack(sync_point_, [this](void* /* arg */) { + fault_injection_env_->SetFilesystemActive(false, + Status::IOError(sync_point_)); + }); + SyncPoint::GetInstance()->EnableProcessing(); + + PinnableWideColumns entity_result; + Status s = db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(), key, + &entity_result); + ASSERT_TRUE(s.IsIOError()) << "Expected IOError but got: " << s.ToString(); + + PinnableSlice get_result; + s = db_->Get(ReadOptions(), db_->DefaultColumnFamily(), key, &get_result); + ASSERT_TRUE(s.IsIOError()) << "Expected IOError but got: " << s.ToString(); + + SyncPoint::GetInstance()->DisableProcessing(); + SyncPoint::GetInstance()->ClearAllCallBacks(); +} + TEST_P(DBBlobBasicIOErrorMultiGetTest, MultiGetBlobs_IOError) { Options options = GetDefaultOptions(); options.env = fault_injection_env_.get(); diff --git a/db/blob/db_blob_index_test.cc b/db/blob/db_blob_index_test.cc index ae8827851c..316b9555b8 100644 --- a/db/blob/db_blob_index_test.cc +++ b/db/blob/db_blob_index_test.cc @@ -128,6 +128,16 @@ class DBBlobIndexTest : public DBTestBase { columns, s, is_blob_index, value_found); } + bool MaybeResolveMemtableBlobValueForTest(const Slice& key, + const BlobFetcher* blob_fetcher, + PinnableSlice* value, + PinnableWideColumns* columns, + Status* s, bool* is_blob_index, + bool* value_found = nullptr) { + return DBImpl::MaybeResolveMemtableBlobValue( + key, blob_fetcher, value, columns, s, is_blob_index, value_found); + } + Options GetTestOptions() { Options options; options.env = CurrentOptions().env; @@ -271,6 +281,55 @@ TEST_F(DBBlobIndexTest, ASSERT_EQ(static_cast(BlobIndex::Type::kUnknown), blob_index.front()); } +TEST_F(DBBlobIndexTest, + MaybeResolveMemtableBlobValueWithoutFetcherFailsClosed) { + // Goal: if a readonly/secondary memtable hit produces a blob-backed payload + // but no BlobFetcher is available, the helper must fail closed instead of + // handing raw blob-index bytes back to the caller as if they were the value. + std::string blob_index; + BlobIndex::EncodeBlob(&blob_index, /*file_number=*/123, /*offset=*/456, + /*size=*/789, kNoCompression); + + PinnableSlice value; + value.GetSelf()->assign(blob_index.data(), blob_index.size()); + value.PinSelf(); + + Status s = Status::OK(); + bool is_blob_index = true; + ASSERT_TRUE(MaybeResolveMemtableBlobValueForTest( + Slice("key"), /*blob_fetcher=*/nullptr, &value, /*columns=*/nullptr, &s, + &is_blob_index)); + + ASSERT_TRUE(s.IsNotSupported()) << s.ToString(); + ASSERT_TRUE(value.empty()); + ASSERT_FALSE(is_blob_index); +} + +TEST_F(DBBlobIndexTest, ReadOnlyGetImplReturnsBlobIndexWhenRequested) { + // Goal: cover the internal read-only GetImpl contract when the caller + // explicitly asks for raw blob-index bytes via `is_blob_index`. Recovery + // keeps the blob index in the memtable, and the read-only path must preserve + // the encoded index instead of eagerly resolving or rejecting it. + Options options = GetTestOptions(); + + DestroyAndReopen(options); + + std::string blob_index; + BlobIndex::EncodeInlinedTTL(&blob_index, /*expiration=*/9876543210, "blob"); + + WriteBatch batch; + ASSERT_OK(PutBlobIndex(&batch, "blob_key", blob_index)); + ASSERT_OK(Write(&batch)); + + Close(); + options.avoid_flush_during_recovery = true; + ASSERT_OK(ReadOnlyReopen(options)); + + bool is_blob_index = false; + ASSERT_EQ(blob_index, GetImpl("blob_key", &is_blob_index)); + ASSERT_TRUE(is_blob_index); +} + class PlainBlobValueFilterV3 : public CompactionFilter { public: PlainBlobValueFilterV3(std::atomic* filter_call_count, diff --git a/db/db_flush_test.cc b/db/db_flush_test.cc index 146f982282..cc46236c3f 100644 --- a/db/db_flush_test.cc +++ b/db/db_flush_test.cc @@ -1852,6 +1852,8 @@ TEST_F(DBFlushTest, MemPurgeCorrectLogNumberAndSSTFileCreation) { } } + ASSERT_OK(WaitForFlushCallbacks()); + // Check that there was at least one mempurge uint32_t expected_min_mempurge_count = 1; // Check that there was no SST files created during flush. @@ -1872,6 +1874,8 @@ TEST_F(DBFlushTest, MemPurgeCorrectLogNumberAndSSTFileCreation) { ASSERT_EQ(Get(key), value); } + ASSERT_OK(WaitForFlushCallbacks()); + // Check that there was at least one SST files created during flush. expected_sst_count = 1; EXPECT_GE(sst_count.load(), expected_sst_count); @@ -1891,6 +1895,9 @@ TEST_F(DBFlushTest, MemPurgeCorrectLogNumberAndSSTFileCreation) { // Extra check of database consistency. ASSERT_EQ(Get(key), value); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Close(); } diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 5a881b68e6..cb148bc698 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -30,6 +30,7 @@ #include "db/arena_wrapped_db_iter.h" #include "db/attribute_group_iterator_impl.h" +#include "db/blob/blob_fetcher.h" #include "db/blob/blob_file_partition_manager.h" #include "db/blob/blob_index.h" #include "db/builder.h" @@ -2871,6 +2872,148 @@ Status DBImpl::ResolveDirectWriteWideColumns(const ReadOptions& read_options, return status; } +bool DBImpl::MaybeResolveMemtableBlobValue(const Slice& key, + const BlobFetcher* blob_fetcher, + PinnableSlice* value, + PinnableWideColumns* columns, + Status* s, bool* is_blob_index, + bool* value_found) { + if (!s->ok() || (!value && !columns)) { + return false; + } + + auto reset_outputs = [&]() { + if (value != nullptr) { + value->Reset(); + } + if (columns != nullptr) { + columns->Reset(); + } + }; + auto clear_blob_state = [&]() { + if (is_blob_index != nullptr) { + *is_blob_index = false; + } + }; + + const bool needs_plain_value_resolution = + is_blob_index != nullptr && *is_blob_index; + const bool needs_wide_column_resolution = + columns != nullptr && !columns->unresolved_blob_column_indices_.empty(); + if (!needs_plain_value_resolution && !needs_wide_column_resolution) { + return false; + } + + if (blob_fetcher == nullptr) { + reset_outputs(); + *s = Status::NotSupported( + "Encountered blob-backed memtable value without blob fetcher."); + clear_blob_state(); + return true; + } + + if (needs_plain_value_resolution) { + Slice blob_index_slice; + std::string blob_index_storage; + if (value != nullptr) { + if (value->size() > 0) { + blob_index_slice = Slice(value->data(), value->size()); + } else { + blob_index_slice = Slice(*(value->GetSelf())); + } + } else { + assert(columns != nullptr); + + const WideColumns& plain_value_columns = columns->columns(); + assert(plain_value_columns.size() == 1); + assert(plain_value_columns.front().name() == kDefaultWideColumnName); + blob_index_slice = plain_value_columns.front().value(); + } + + PinnableSlice resolved_value; + PinnableSlice* target = value != nullptr ? value : &resolved_value; + if (value != nullptr) { + // BlobIndex::DecodeFrom can retain Slices into the encoded bytes for + // inlined blob indices, so take an owned copy before resetting `value` + // and reusing the same PinnableSlice as the output target. + blob_index_storage.assign(blob_index_slice.data(), + blob_index_slice.size()); + blob_index_slice = Slice(blob_index_storage); + value->Reset(); + } + + *s = blob_fetcher->FetchBlob(key, blob_index_slice, + nullptr /* prefetch_buffer */, target, + nullptr /* bytes_read */); + if (s->ok() && columns != nullptr) { + columns->SetPlainValue(std::move(*target)); + } else if (!s->ok()) { + reset_outputs(); + if (s->IsIncomplete() && value_found != nullptr) { + *value_found = false; + } + } + + clear_blob_state(); + return true; + } + + assert(columns != nullptr); + + std::string resolved_entity; + bool resolved = false; + *s = WideColumnSerialization::ResolveEntityBlobColumns( + columns->value_, key, blob_fetcher, nullptr /* prefetch_buffers */, + resolved_entity, resolved, nullptr /* total_bytes_read */, + nullptr /* num_blobs_resolved */); + if (s->ok()) { + assert(resolved); + if (resolved) { + *s = columns->SetWideColumnValue(std::move(resolved_entity)); + } + } + if (!s->ok()) { + reset_outputs(); + if (s->IsIncomplete() && value_found != nullptr) { + *value_found = false; + } + } + + clear_blob_state(); + return true; +} + +void DBImpl::PostprocessMemtableValueRead( + const Slice& key, const std::string* timestamp, + bool resolve_blob_backed_memtable_value, + const BlobFetcher* memtable_blob_fetcher, PinnableSlice* value, + PinnableWideColumns* columns, Status* s, bool* is_blob_index, + bool* value_found) { + if (resolve_blob_backed_memtable_value) { + std::string blob_lookup_key_storage; + const bool value_resolved = MaybeResolveMemtableBlobValue( + GetBlobLookupUserKey(key, timestamp, &blob_lookup_key_storage), + memtable_blob_fetcher, value, columns, s, is_blob_index, value_found); + if (!value_resolved && value != nullptr && s->ok()) { + value->PinSelf(); + } + return; + } + + if (s->ok()) { + if (value != nullptr) { + value->PinSelf(); + } + } else { + if (value != nullptr) { + value->Reset(); + } + if (columns != nullptr) { + columns->Reset(); + } + } +} + bool DBImpl::MaybeResolveDirectWriteValue( const ReadOptions& read_options, const Slice& key, bool resolve_direct_write_value, const Version* current, @@ -3086,6 +3229,14 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key, } const bool resolve_direct_write_value = partition_mgr != nullptr && (is_blob_ptr == &is_blob_index); + std::optional memtable_blob_fetcher; + if (partition_mgr != nullptr) { + memtable_blob_fetcher.emplace(sv->current, read_options, + cfd->blob_file_cache(), + /*allow_write_path_fallback=*/true); + } + const BlobFetcher* memtable_blob_fetcher_ptr = + memtable_blob_fetcher ? &*memtable_blob_fetcher : nullptr; std::string blob_lookup_key_storage; auto get_blob_lookup_key = [&]() -> Slice { return GetBlobLookupUserKey(key, timestamp, &blob_lookup_key_storage); @@ -3113,19 +3264,21 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key, get_impl_options.columns, timestamp, &s, &merge_context, &max_covering_tombstone_seq, read_options, false /* immutable_memtable */, - get_impl_options.callback, is_blob_ptr)) { + get_impl_options.callback, is_blob_ptr, + /*do_merge=*/true, memtable_blob_fetcher_ptr)) { done = true; maybe_resolve_memtable_value(); RecordTick(stats_, MEMTABLE_HIT); } else if ((s.ok() || s.IsMergeInProgress()) && - sv->imm->Get( - lkey, - get_impl_options.value ? get_impl_options.value->GetSelf() - : nullptr, - get_impl_options.columns, timestamp, &s, &merge_context, - &max_covering_tombstone_seq, read_options, - get_impl_options.callback, is_blob_ptr)) { + sv->imm->Get(lkey, + get_impl_options.value + ? get_impl_options.value->GetSelf() + : nullptr, + get_impl_options.columns, timestamp, &s, + &merge_context, &max_covering_tombstone_seq, + read_options, get_impl_options.callback, + is_blob_ptr, memtable_blob_fetcher_ptr)) { done = true; maybe_resolve_memtable_value(); @@ -3137,14 +3290,14 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key, if (sv->mem->Get(lkey, /*value=*/nullptr, /*columns=*/nullptr, /*timestamp=*/nullptr, &s, &merge_context, &max_covering_tombstone_seq, read_options, - false /* immutable_memtable */, nullptr, nullptr, - false)) { + false /* immutable_memtable */, nullptr, nullptr, false, + memtable_blob_fetcher_ptr)) { done = true; RecordTick(stats_, MEMTABLE_HIT); } else if ((s.ok() || s.IsMergeInProgress()) && - sv->imm->GetMergeOperands(lkey, &s, &merge_context, - &max_covering_tombstone_seq, - read_options)) { + sv->imm->GetMergeOperands( + lkey, &s, &merge_context, &max_covering_tombstone_seq, + read_options, memtable_blob_fetcher_ptr)) { done = true; RecordTick(stats_, MEMTABLE_HIT); } @@ -3840,6 +3993,14 @@ Status DBImpl::MultiGetImpl( (*sorted_keys)[start_key]->column_family); ColumnFamilyData* cfd = cfh->cfd(); auto* partition_mgr = cfd->blob_partition_manager(); + std::optional memtable_blob_fetcher; + if (partition_mgr != nullptr) { + memtable_blob_fetcher.emplace(super_version->current, read_options, + cfd->blob_file_cache(), + /*allow_write_path_fallback=*/true); + } + const BlobFetcher* memtable_blob_fetcher_ptr = + memtable_blob_fetcher ? &*memtable_blob_fetcher : nullptr; // Clear the timestamps for returning results so that we can distinguish // between tombstone or key that has never been written for (size_t i = start_key; i < start_key + num_keys; ++i) { @@ -3886,9 +4047,11 @@ Status DBImpl::MultiGetImpl( has_unpersisted_data_.load(std::memory_order_relaxed)); if (!skip_memtable) { super_version->mem->MultiGet(read_options, &range, callback, - false /* immutable_memtable */); + false /* immutable_memtable */, + memtable_blob_fetcher_ptr); if (!range.empty()) { - super_version->imm->MultiGet(read_options, &range, callback); + super_version->imm->MultiGet(read_options, &range, callback, + memtable_blob_fetcher_ptr); } if (!range.empty()) { uint64_t left = range.KeysLeft(); @@ -6323,12 +6486,20 @@ Status DBImpl::GetLatestSequenceForKey( *seq = kMaxSequenceNumber; *found_record_for_key = false; + std::optional memtable_blob_fetcher; + if (cfd->blob_partition_manager() != nullptr) { + memtable_blob_fetcher.emplace(sv->current, read_options, + cfd->blob_file_cache(), + /*allow_write_path_fallback=*/true); + } + const BlobFetcher* memtable_blob_fetcher_ptr = + memtable_blob_fetcher ? &*memtable_blob_fetcher : nullptr; // Check if there is a record for this key in the latest memtable sv->mem->Get(lkey, /*value=*/nullptr, /*columns=*/nullptr, timestamp, &s, &merge_context, &max_covering_tombstone_seq, seq, read_options, false /* immutable_memtable */, nullptr /*read_callback*/, - is_blob_index); + is_blob_index, /*do_merge=*/true, memtable_blob_fetcher_ptr); if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) { // unexpected error reading memtable. @@ -6361,7 +6532,8 @@ Status DBImpl::GetLatestSequenceForKey( // Check if there is a record for this key in the immutable memtables sv->imm->Get(lkey, /*value=*/nullptr, /*columns=*/nullptr, timestamp, &s, &merge_context, &max_covering_tombstone_seq, seq, read_options, - nullptr /*read_callback*/, is_blob_index); + nullptr /*read_callback*/, is_blob_index, + memtable_blob_fetcher_ptr); if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) { // unexpected error reading memtable. @@ -6394,7 +6566,7 @@ Status DBImpl::GetLatestSequenceForKey( sv->imm->GetFromHistory(lkey, /*value=*/nullptr, /*columns=*/nullptr, timestamp, &s, &merge_context, &max_covering_tombstone_seq, seq, read_options, - is_blob_index); + is_blob_index, memtable_blob_fetcher_ptr); if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) { // unexpected error reading memtable. diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index da2039c462..3e693076be 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -1875,6 +1875,7 @@ class DBImpl : public DB { private: friend class DB; + friend class DBImplReadOnly; friend class DBImplSecondary; friend class ErrorHandler; friend class InternalStats; @@ -2952,6 +2953,29 @@ class DBImpl : public DB { bool resolve_direct_write_value, const Version* current, ColumnFamilyData* cfd, PinnableSlice* value, PinnableWideColumns* columns, Status* s, bool* is_blob_index, bool* value_found = nullptr); + // Resolves memtable read results that still carry blob references through + // either a raw blob-index payload in `value` or unresolved blob columns in + // `columns`. Unlike the direct-write helper above, this path only depends on + // a BlobFetcher and therefore works for read-only/secondary DBs. + static bool MaybeResolveMemtableBlobValue(const Slice& key, + const BlobFetcher* blob_fetcher, + PinnableSlice* value, + PinnableWideColumns* columns, + Status* s, bool* is_blob_index, + bool* value_found = nullptr); + // Completes read-only/secondary memtable Get()/GetEntity() hits by resolving + // blob-backed payloads when `resolve_blob_backed_memtable_value` is true, + // pinning plain values on success, and clearing outputs on error. When the + // caller explicitly requested raw blob indices via + // `GetImplOptions::is_blob_index`, this helper leaves that payload + // untouched. `memtable_blob_fetcher` may be null when blob support is + // disabled for the column family. + static void PostprocessMemtableValueRead( + const Slice& key, const std::string* timestamp, + bool resolve_blob_backed_memtable_value, + const BlobFetcher* memtable_blob_fetcher, PinnableSlice* value, + PinnableWideColumns* columns, Status* s, bool* is_blob_index, + bool* value_found = nullptr); template diff --git a/db/db_impl/db_impl_follower.cc b/db/db_impl/db_impl_follower.cc index 11ba6070e7..51cbde19a4 100644 --- a/db/db_impl/db_impl_follower.cc +++ b/db/db_impl/db_impl_follower.cc @@ -295,7 +295,8 @@ Status DB::OpenAsFollower( impl->versions_.reset(new ReactiveVersionSet( dbname, &impl->immutable_db_options_, impl->mutable_db_options_, impl->file_options_, impl->table_cache_.get(), - impl->write_buffer_manager_, &impl->write_controller_, impl->io_tracer_)); + impl->write_buffer_manager_, &impl->write_controller_, impl->io_tracer_, + impl->db_id_, impl->db_session_id_)); impl->column_family_memtables_.reset( new ColumnFamilyMemTablesImpl(impl->versions_->GetColumnFamilySet())); impl->wal_in_db_path_ = impl->immutable_db_options_.IsWalDirSameAsDBPath(); diff --git a/db/db_impl/db_impl_readonly.cc b/db/db_impl/db_impl_readonly.cc index 6ff0a48ba9..09df0bd6b7 100644 --- a/db/db_impl/db_impl_readonly.cc +++ b/db/db_impl/db_impl_readonly.cc @@ -5,7 +5,10 @@ #include "db/db_impl/db_impl_readonly.h" +#include + #include "db/arena_wrapped_db_iter.h" +#include "db/blob/blob_fetcher.h" #include "db/db_impl/compacted_db_impl.h" #include "db/db_impl/db_impl.h" #include "db/manifest_ops.h" @@ -63,13 +66,25 @@ Status DBImplReadOnly::GetImpl(const ReadOptions& read_options, const Comparator* ucmp = get_impl_options.column_family->GetComparator(); assert(ucmp); - std::string* ts = - ucmp->timestamp_size() > 0 ? get_impl_options.timestamp : nullptr; SequenceNumber snapshot = versions_->LastSequence(); GetWithTimestampReadCallback read_cb(snapshot); auto cfh = static_cast_with_check( get_impl_options.column_family); auto cfd = cfh->cfd(); + bool is_blob_index = false; + bool* is_blob_ptr = get_impl_options.is_blob_index; + std::string timestamp_storage; + std::string* ts = nullptr; + if (ucmp->timestamp_size() > 0) { + ts = get_impl_options.timestamp != nullptr + ? get_impl_options.timestamp + : (get_impl_options.get_value ? ×tamp_storage : nullptr); + } + if (!is_blob_ptr && get_impl_options.get_value) { + is_blob_ptr = &is_blob_index; + } + const bool resolve_blob_backed_memtable_value = + get_impl_options.get_value && (is_blob_ptr == &is_blob_index); if (tracer_) { InstrumentedMutexLock lock(&trace_mutex_); if (tracer_) { @@ -95,6 +110,18 @@ Status DBImplReadOnly::GetImpl(const ReadOptions& read_options, SequenceNumber max_covering_tombstone_seq = 0; LookupKey lkey(key, snapshot, read_options.timestamp); PERF_TIMER_STOP(get_snapshot_time); + std::optional memtable_blob_fetcher; + if (cfd->ioptions().enable_blob_direct_write || + cfd->GetLatestMutableCFOptions().enable_blob_files) { + // Recovered memtables can still contain older blob references after + // mutable blob-file settings change, so keep blob resolution available + // whenever either blob knob indicates it may be needed. + memtable_blob_fetcher.emplace(super_version->current, read_options, + cfd->blob_file_cache(), + /*allow_write_path_fallback=*/true); + } + const BlobFetcher* memtable_blob_fetcher_ptr = + memtable_blob_fetcher ? &*memtable_blob_fetcher : nullptr; // Look up starts here if (super_version->mem->Get( @@ -102,11 +129,12 @@ Status DBImplReadOnly::GetImpl(const ReadOptions& read_options, get_impl_options.value ? get_impl_options.value->GetSelf() : nullptr, get_impl_options.columns, ts, &s, &merge_context, &max_covering_tombstone_seq, read_options, - false /* immutable_memtable */, &read_cb, - /*is_blob_index=*/nullptr, /*do_merge=*/get_impl_options.get_value)) { - if (get_impl_options.value) { - get_impl_options.value->PinSelf(); - } + false /* immutable_memtable */, &read_cb, is_blob_ptr, + /*do_merge=*/get_impl_options.get_value, memtable_blob_fetcher_ptr)) { + DBImpl::PostprocessMemtableValueRead( + key, ts, resolve_blob_backed_memtable_value, memtable_blob_fetcher_ptr, + get_impl_options.value, get_impl_options.columns, &s, &is_blob_index, + get_impl_options.value_found); RecordTick(stats_, MEMTABLE_HIT); } else { PERF_TIMER_GUARD(get_from_output_files_time); diff --git a/db/db_impl/db_impl_secondary.cc b/db/db_impl/db_impl_secondary.cc index 46758e82a7..56b41e4cf7 100644 --- a/db/db_impl/db_impl_secondary.cc +++ b/db/db_impl/db_impl_secondary.cc @@ -6,8 +6,10 @@ #include "db/db_impl/db_impl_secondary.h" #include +#include #include "db/arena_wrapped_db_iter.h" +#include "db/blob/blob_fetcher.h" #include "db/log_reader.h" #include "db/log_writer.h" #include "db/merge_context.h" @@ -363,13 +365,25 @@ Status DBImplSecondary::GetImpl(const ReadOptions& read_options, const Comparator* ucmp = get_impl_options.column_family->GetComparator(); assert(ucmp); - std::string* ts = - ucmp->timestamp_size() > 0 ? get_impl_options.timestamp : nullptr; SequenceNumber snapshot = versions_->LastSequence(); GetWithTimestampReadCallback read_cb(snapshot); auto cfh = static_cast_with_check( get_impl_options.column_family); auto cfd = cfh->cfd(); + bool is_blob_index = false; + bool* is_blob_ptr = get_impl_options.is_blob_index; + std::string timestamp_storage; + std::string* ts = nullptr; + if (ucmp->timestamp_size() > 0) { + ts = get_impl_options.timestamp != nullptr + ? get_impl_options.timestamp + : (get_impl_options.get_value ? ×tamp_storage : nullptr); + } + if (!is_blob_ptr && get_impl_options.get_value) { + is_blob_ptr = &is_blob_index; + } + const bool resolve_blob_backed_memtable_value = + get_impl_options.get_value && (is_blob_ptr == &is_blob_index); if (tracer_) { InstrumentedMutexLock lock(&trace_mutex_); if (tracer_) { @@ -395,6 +409,18 @@ Status DBImplSecondary::GetImpl(const ReadOptions& read_options, LookupKey lkey(key, snapshot, read_options.timestamp); PERF_TIMER_STOP(get_snapshot_time); bool done = false; + std::optional memtable_blob_fetcher; + if (cfd->ioptions().enable_blob_direct_write || + cfd->GetLatestMutableCFOptions().enable_blob_files) { + // Catch-up can rebuild older blob references into memtables after mutable + // blob-file settings change, so keep blob resolution available whenever + // either blob knob indicates it may be needed. + memtable_blob_fetcher.emplace(super_version->current, read_options, + cfd->blob_file_cache(), + /*allow_write_path_fallback=*/true); + } + const BlobFetcher* memtable_blob_fetcher_ptr = + memtable_blob_fetcher ? &*memtable_blob_fetcher : nullptr; // Look up starts here if (get_impl_options.get_value) { @@ -404,12 +430,14 @@ Status DBImplSecondary::GetImpl(const ReadOptions& read_options, : nullptr, get_impl_options.columns, ts, &s, &merge_context, &max_covering_tombstone_seq, read_options, - false /* immutable_memtable */, &read_cb, - /*is_blob_index=*/nullptr, /*do_merge=*/true)) { + false /* immutable_memtable */, &read_cb, is_blob_ptr, + /*do_merge=*/true, memtable_blob_fetcher_ptr)) { done = true; - if (get_impl_options.value) { - get_impl_options.value->PinSelf(); - } + DBImpl::PostprocessMemtableValueRead( + key, ts, resolve_blob_backed_memtable_value, + memtable_blob_fetcher_ptr, get_impl_options.value, + get_impl_options.columns, &s, &is_blob_index, + get_impl_options.value_found); RecordTick(stats_, MEMTABLE_HIT); } else if ((s.ok() || s.IsMergeInProgress()) && super_version->imm->Get( @@ -417,11 +445,14 @@ Status DBImplSecondary::GetImpl(const ReadOptions& read_options, get_impl_options.value ? get_impl_options.value->GetSelf() : nullptr, get_impl_options.columns, ts, &s, &merge_context, - &max_covering_tombstone_seq, read_options, &read_cb)) { + &max_covering_tombstone_seq, read_options, &read_cb, + is_blob_ptr, memtable_blob_fetcher_ptr)) { done = true; - if (get_impl_options.value) { - get_impl_options.value->PinSelf(); - } + DBImpl::PostprocessMemtableValueRead( + key, ts, resolve_blob_backed_memtable_value, + memtable_blob_fetcher_ptr, get_impl_options.value, + get_impl_options.columns, &s, &is_blob_index, + get_impl_options.value_found); RecordTick(stats_, MEMTABLE_HIT); } } else { @@ -433,13 +464,14 @@ Status DBImplSecondary::GetImpl(const ReadOptions& read_options, get_impl_options.columns, ts, &s, &merge_context, &max_covering_tombstone_seq, read_options, false /* immutable_memtable */, &read_cb, - /*is_blob_index=*/nullptr, /*do_merge=*/false)) { + /*is_blob_index=*/nullptr, /*do_merge=*/false, + memtable_blob_fetcher_ptr)) { done = true; RecordTick(stats_, MEMTABLE_HIT); } else if ((s.ok() || s.IsMergeInProgress()) && - super_version->imm->GetMergeOperands(lkey, &s, &merge_context, - &max_covering_tombstone_seq, - read_options)) { + super_version->imm->GetMergeOperands( + lkey, &s, &merge_context, &max_covering_tombstone_seq, + read_options, memtable_blob_fetcher_ptr)) { done = true; RecordTick(stats_, MEMTABLE_HIT); } @@ -457,7 +489,7 @@ Status DBImplSecondary::GetImpl(const ReadOptions& read_options, ts, &s, &merge_context, &max_covering_tombstone_seq, &pinned_iters_mgr, /*value_found*/ nullptr, /*key_exists*/ nullptr, /*seq*/ nullptr, &read_cb, /*is_blob*/ nullptr, - /*do_merge*/ true); + /*do_merge=*/get_impl_options.get_value); RecordTick(stats_, MEMTABLE_MISS); } { @@ -777,7 +809,8 @@ Status DBImplSecondary::OpenAsSecondaryImpl( impl->versions_.reset(new ReactiveVersionSet( dbname, &impl->immutable_db_options_, impl->mutable_db_options_, impl->file_options_, impl->table_cache_.get(), - impl->write_buffer_manager_, &impl->write_controller_, impl->io_tracer_)); + impl->write_buffer_manager_, &impl->write_controller_, impl->io_tracer_, + impl->db_id_, impl->db_session_id_)); impl->column_family_memtables_.reset( new ColumnFamilyMemTablesImpl(impl->versions_->GetColumnFamilySet())); impl->wal_in_db_path_ = impl->immutable_db_options_.IsWalDirSameAsDBPath(); diff --git a/db/db_secondary_test.cc b/db/db_secondary_test.cc index 0acdf36a22..be4a17274a 100644 --- a/db/db_secondary_test.cc +++ b/db/db_secondary_test.cc @@ -7,9 +7,12 @@ // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. See the AUTHORS file for names of contributors. +#include "db/blob/blob_index.h" #include "db/db_impl/db_impl_secondary.h" #include "db/db_test_util.h" #include "db/db_with_timestamp_test_util.h" +#include "db/wide/wide_column_test_util.h" +#include "db/write_batch_internal.h" #include "port/stack_trace.h" #include "rocksdb/utilities/transaction_db.h" #include "test_util/sync_point.h" @@ -357,8 +360,7 @@ TEST_F(DBSecondaryTest, GetMergeOperands) { const Status s = db_secondary_->GetMergeOperands( ReadOptions(), cfh, "k1", values.data(), &merge_operands_info, &number_of_operands); - ASSERT_NOK(s); - ASSERT_TRUE(s.IsMergeInProgress()); + ASSERT_OK(s); ASSERT_EQ(number_of_operands, 4); ASSERT_EQ(values[0].ToString(), "v1"); @@ -367,6 +369,197 @@ TEST_F(DBSecondaryTest, GetMergeOperands) { ASSERT_EQ(values[3].ToString(), "v4"); } +TEST_F(DBSecondaryTest, GetMergeOperandsWithBlobBackedEntityDefaultColumn) { + // Goal: exercise the secondary read path with a blob-backed V2 entity base + // in SST and a newer merge operand applied through catch-up. The secondary + // DB must resolve the blob-backed default column for both Get() and + // GetMergeOperands() while combining the newer memtable merge with the older + // SST-backed base entity. + Options options = GetDefaultOptions(); + options.create_if_missing = true; + options.enable_blob_files = true; + options.min_blob_size = 50; + options.disable_auto_compactions = true; + options.merge_operator = MergeOperators::CreateStringAppendOperator("|"); + options.env = env_; + Reopen(options); + + const std::string key = "secondary_blob_entity"; + const std::string default_value(100, 'd'); + const std::string large_value(120, 'l'); + const std::string merge_operand = "suffix"; + const std::string expected_merged = default_value + "|" + merge_operand; + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"meta", "inline"}}; + + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + + options.max_open_files = -1; + OpenSecondary(options); + ASSERT_OK(db_->Merge(WriteOptions(), db_->DefaultColumnFamily(), key, + merge_operand)); + ASSERT_OK(db_secondary_->TryCatchUpWithPrimary()); + + auto* cfh = db_secondary_->DefaultColumnFamily(); + + { + PinnableSlice result; + ASSERT_OK(db_secondary_->Get(ReadOptions(), cfh, key, &result)); + ASSERT_EQ(result, expected_merged); + } + + { + GetMergeOperandsOptions get_merge_opts; + get_merge_opts.expected_max_number_of_operands = 2; + + std::array merge_operands; + int number_of_operands = 0; + + ASSERT_OK(db_secondary_->GetMergeOperands( + ReadOptions(), cfh, key, merge_operands.data(), &get_merge_opts, + &number_of_operands)); + ASSERT_EQ(number_of_operands, 2); + ASSERT_EQ(merge_operands[0], default_value); + ASSERT_EQ(merge_operands[1], merge_operand); + } +} + +TEST_F(DBSecondaryTest, GetImplReturnsBlobIndexWhenRequested) { + // Goal: cover the internal secondary GetImpl contract when the caller + // explicitly asks for raw blob-index bytes via `is_blob_index`. Catch-up + // rebuilds the blob index into the secondary memtable, and this path must + // preserve the encoded index instead of eagerly resolving or rejecting it. + Options options = GetDefaultOptions(); + options.create_if_missing = true; + options.env = env_; + + Reopen(options); + + options.max_open_files = -1; + OpenSecondary(options); + + std::string blob_index; + BlobIndex::EncodeInlinedTTL(&blob_index, /*expiration=*/9876543210, "blob"); + + WriteBatch batch; + ASSERT_OK(WriteBatchInternal::PutBlobIndex( + &batch, db_->DefaultColumnFamily()->GetID(), "blob_key", blob_index)); + ASSERT_OK(db_->Write(WriteOptions(), &batch)); + ASSERT_OK(db_secondary_->TryCatchUpWithPrimary()); + + PinnableSlice value; + bool is_blob_index = false; + DBImpl::GetImplOptions get_impl_options; + get_impl_options.column_family = db_secondary_->DefaultColumnFamily(); + get_impl_options.value = &value; + get_impl_options.is_blob_index = &is_blob_index; + + ASSERT_OK(db_secondary_full()->GetImpl(ReadOptions(), Slice("blob_key"), + get_impl_options)); + ASSERT_TRUE(is_blob_index); + ASSERT_EQ(value, blob_index); +} + +TEST_F(DBSecondaryTest, + GetAndGetEntityWithBlobBackedDefaultColumnDirectWriteMemtable) { + // Goal: cover the secondary memtable path after catch-up replays a blob + // direct-write entity from the primary WAL. The test checks both `Get()`, + // which must resolve the blob-backed default column, and `GetEntity()`, + // which must eagerly resolve all unresolved blob columns instead of exposing + // encoded blob indices to the caller. + Options options = + wide_column_test_util::GetDirectWriteOptions(GetDefaultOptions()); + options.create_if_missing = true; + options.min_blob_size = 50; + options.env = env_; + + Reopen(options); + + options.max_open_files = -1; + OpenSecondary(options); + + const std::string key = "secondary_direct_write_memtable_entity"; + const std::string default_value = + wide_column_test_util::GenerateLargeValue(100, 'd'); + const std::string large_value = + wide_column_test_util::GenerateLargeValue(120, 'l'); + const std::string small_value = wide_column_test_util::GenerateSmallValue(); + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(db_secondary_->TryCatchUpWithPrimary()); + + auto* cfh = db_secondary_->DefaultColumnFamily(); + + { + PinnableSlice result; + ASSERT_OK(db_secondary_->Get(ReadOptions(), cfh, key, &result)); + ASSERT_EQ(result, default_value); + } + + { + PinnableWideColumns result; + ASSERT_OK(db_secondary_->GetEntity(ReadOptions(), cfh, key, &result)); + ASSERT_EQ(result.columns(), columns); + } +} + +TEST_F(DBSecondaryTest, SecondaryDirectWriteMemtableBlobBlockCacheTier) { + // Goal: cover the secondary memtable path under kBlockCacheTier. Catch-up + // rebuilds the blob-backed direct-write entity from WAL, so serving Get() or + // GetEntity() would require blob I/O through the memtable resolution path, + // which must return Incomplete instead of issuing that I/O. + Options options = + wide_column_test_util::GetDirectWriteOptions(GetDefaultOptions()); + options.create_if_missing = true; + options.min_blob_size = 50; + options.env = env_; + + Reopen(options); + + options.max_open_files = -1; + OpenSecondary(options); + + const std::string key = "secondary_direct_write_memtable_block_cache_tier"; + const std::string default_value = + wide_column_test_util::GenerateLargeValue(100, 'd'); + const std::string large_value = + wide_column_test_util::GenerateLargeValue(120, 'l'); + const std::string small_value = wide_column_test_util::GenerateSmallValue(); + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(db_secondary_->TryCatchUpWithPrimary()); + + auto* cfh = db_secondary_->DefaultColumnFamily(); + ReadOptions read_opts; + read_opts.read_tier = kBlockCacheTier; + + { + PinnableSlice result; + const Status s = db_secondary_->Get(read_opts, cfh, key, &result); + ASSERT_TRUE(s.IsIncomplete()) << s.ToString(); + ASSERT_TRUE(result.empty()); + } + + { + PinnableWideColumns result; + const Status s = db_secondary_->GetEntity(read_opts, cfh, key, &result); + ASSERT_TRUE(s.IsIncomplete()) << s.ToString(); + ASSERT_TRUE(result.columns().empty()); + } +} + TEST_F(DBSecondaryTest, InternalCompactionCompactedFiles) { Options options; options.env = env_; diff --git a/db/db_test2.cc b/db/db_test2.cc index 5267eac218..16a54f0208 100644 --- a/db/db_test2.cc +++ b/db/db_test2.cc @@ -7510,6 +7510,65 @@ TEST_F(DBTest2, GetLatestSeqAndTsForKey) { ASSERT_EQ(0, options.statistics->getTickerCount(GET_HIT_L0)); } +TEST_F(DBTest2, + GetLatestSequenceForKeyFromHistoryWithBlobBackedWideColumnEntity) { + // Goal: exercise the memtable history lookup in GetLatestSequenceForKey() + // after blob direct write stores a blob-backed V2 entity in a flushed + // memtable. Using cache_only=true ensures the lookup succeeds only if the + // history path can resolve the wide-column base value under the newer merge. + Destroy(last_options_); + + Options options = CurrentOptions(); + options.create_if_missing = true; + options.enable_blob_files = true; + options.enable_blob_direct_write = true; + options.min_blob_size = 50; + options.allow_concurrent_memtable_write = false; + options.blob_direct_write_partitions = 1; + options.max_write_buffer_size_to_maintain = 64 << 10; + options.merge_operator = MergeOperators::CreateStringAppendOperator("|"); + + Reopen(options); + + const std::string key = "history_blob_entity"; + const std::string default_value(100, 'd'); + const std::string merge_operand = "suffix"; + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"meta", "inline"}}; + + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(db_->Merge(WriteOptions(), db_->DefaultColumnFamily(), key, + merge_operand)); + const SequenceNumber expected_seq = dbfull()->GetLatestSequenceNumber(); + + ASSERT_OK(Flush()); + ASSERT_FALSE(GetBlobFileNumbers().empty()); + + uint64_t num_immutable_memtables = 0; + ASSERT_TRUE(db_->GetIntProperty(DB::Properties::kNumImmutableMemTable, + &num_immutable_memtables)); + ASSERT_EQ(num_immutable_memtables, 0); + + auto* cfhi = static_cast_with_check( + dbfull()->DefaultColumnFamily()); + assert(cfhi); + assert(cfhi->cfd()); + SuperVersion* sv = cfhi->cfd()->GetSuperVersion(); + + SequenceNumber seq = kMaxSequenceNumber; + bool found_record_for_key = false; + bool is_blob_index = false; + const Status s = dbfull()->GetLatestSequenceForKey( + sv, key, /*cache_only=*/true, /*lower_bound_seq=*/0, &seq, + /*timestamp=*/nullptr, &found_record_for_key, &is_blob_index); + + ASSERT_OK(s); + ASSERT_TRUE(found_record_for_key); + ASSERT_EQ(expected_seq, seq); + ASSERT_FALSE(is_blob_index); +} + #if defined(ZSTD) TEST_F(DBTest2, ZSTDChecksum) { // Verify that corruption during decompression is caught. diff --git a/db/memtable.cc b/db/memtable.cc index d672abda93..a971e031ad 100644 --- a/db/memtable.cc +++ b/db/memtable.cc @@ -15,6 +15,7 @@ #include #include +#include "db/blob/blob_fetcher.h" #include "db/blob/blob_file_partition_manager.h" #include "db/blob/blob_index.h" #include "db/dbformat.h" @@ -70,6 +71,67 @@ Status GetDefaultColumnBlobIndexSlice(Slice entity, Slice* blob_index_slice) { return status; } +Status PushWideColumnEntityDefaultOperand(const Slice& user_key, + const Slice& entity, + MergeContext* merge_context, + bool operand_pinned, + const BlobFetcher* blob_fetcher) { + assert(merge_context != nullptr); + + Slice entity_ref = entity; + Slice value_of_default; + Status status = WideColumnSerialization::GetValueOfDefaultColumn( + entity_ref, value_of_default); + if (status.ok()) { + merge_context->PushOperand(value_of_default, operand_pinned); + return status; + } + if (!status.IsNotSupported()) { + return status; + } + if (blob_fetcher == nullptr) { + return Status::Corruption( + "Cannot resolve blob-backed default column without a blob fetcher"); + } + + PinnableSlice resolved_default; + bool resolved = false; + status = WideColumnSerialization::GetValueOfDefaultColumnResolvingBlobs( + entity, user_key, blob_fetcher, resolved_default, resolved); + if (status.ok()) { + // Resolved blob values are backed by this stack-local PinnableSlice, so + // copy them into MergeContext instead of pinning their storage. + merge_context->PushOperand(Slice(resolved_default), false); + } + return status; +} + +Status MergeWithWideColumnEntityBaseValue( + const Slice& user_key, const Slice& entity, + const MergeOperator* merge_operator, MergeContext* merge_context, + Logger* logger, Statistics* statistics, SystemClock* clock, + std::string* value, PinnableWideColumns* columns, + const BlobFetcher* blob_fetcher) { + assert(merge_context != nullptr); + + std::string resolved_entity; + Slice effective_entity; + Status status = WideColumnSerialization::ResolveEntityForMerge( + entity, user_key, blob_fetcher, nullptr /* prefetch_buffers */, + resolved_entity, effective_entity); + if (!status.ok()) { + return status; + } + + // `op_failure_scope` (an output parameter) is not provided (set to nullptr) + // since a failure must be propagated regardless of its value. + return MergeHelper::TimedFullMerge( + merge_operator, user_key, MergeHelper::kWideBaseValue, effective_entity, + merge_context->GetOperands(), logger, statistics, clock, + /* update_num_ops_stats */ true, /* op_failure_scope */ nullptr, value, + columns); +} + } // namespace ImmutableMemTableOptions::ImmutableMemTableOptions( @@ -1254,6 +1316,7 @@ struct Saver { bool inplace_update_support; bool do_merge; SystemClock* clock; + const BlobFetcher* blob_fetcher; ReadCallback* callback_; bool* is_blob_index; @@ -1424,27 +1487,17 @@ static bool SaveValue(void* arg, const char* entry) { // Preserve the value with the goal of returning it as part of // raw merge operands to the user - Slice value_of_default; - *(s->status) = WideColumnSerialization::GetValueOfDefaultColumn( - v, value_of_default); - - if (s->status->ok()) { - merge_context->PushOperand( - value_of_default, - s->inplace_update_support == false /* operand_pinned */); - } + *(s->status) = PushWideColumnEntityDefaultOperand( + s->key->user_key(), v, merge_context, + s->inplace_update_support == false /* operand_pinned */, + s->blob_fetcher); } else if (*(s->merge_in_progress)) { assert(s->do_merge); if (s->value || s->columns) { - // `op_failure_scope` (an output parameter) is not provided (set - // to nullptr) since a failure must be propagated regardless of - // its value. - *(s->status) = MergeHelper::TimedFullMerge( - merge_operator, s->key->user_key(), MergeHelper::kWideBaseValue, - v, merge_context->GetOperands(), s->logger, s->statistics, - s->clock, /* update_num_ops_stats */ true, - /* op_failure_scope */ nullptr, s->value, s->columns); + *(s->status) = MergeWithWideColumnEntityBaseValue( + s->key->user_key(), v, merge_operator, merge_context, s->logger, + s->statistics, s->clock, s->value, s->columns, s->blob_fetcher); } } else if (s->value) { Slice value_of_default; @@ -1520,7 +1573,8 @@ bool MemTable::Get(const LookupKey& key, std::string* value, SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, const ReadOptions& read_opts, bool immutable_memtable, ReadCallback* callback, - bool* is_blob_index, bool do_merge) { + bool* is_blob_index, bool do_merge, + const BlobFetcher* blob_fetcher) { // The sequence number is updated synchronously in version_set.h if (IsEmpty()) { // Avoiding recording stats for speed. @@ -1578,7 +1632,7 @@ bool MemTable::Get(const LookupKey& key, std::string* value, } GetFromTable(key, *max_covering_tombstone_seq, do_merge, callback, is_blob_index, value, columns, timestamp, s, merge_context, - seq, &found_final_value, &merge_in_progress); + seq, &found_final_value, &merge_in_progress, blob_fetcher); } // No change to value, since we have not yet found a Put/Delete @@ -1601,7 +1655,8 @@ void MemTable::GetFromTable(const LookupKey& key, PinnableWideColumns* columns, std::string* timestamp, Status* s, MergeContext* merge_context, SequenceNumber* seq, - bool* found_final_value, bool* merge_in_progress) { + bool* found_final_value, bool* merge_in_progress, + const BlobFetcher* blob_fetcher) { Saver saver; saver.status = s; saver.found_final_value = found_final_value; @@ -1619,6 +1674,7 @@ void MemTable::GetFromTable(const LookupKey& key, saver.inplace_update_support = moptions_.inplace_update_support; saver.statistics = moptions_.statistics; saver.clock = clock_; + saver.blob_fetcher = blob_fetcher; saver.callback_ = callback; saver.is_blob_index = is_blob_index; saver.do_merge = do_merge; @@ -1648,7 +1704,8 @@ Status MemTable::ValidateKey(const char* key, bool allow_data_in_errors) { } void MemTable::MultiGet(const ReadOptions& read_options, MultiGetRange* range, - ReadCallback* callback, bool immutable_memtable) { + ReadCallback* callback, bool immutable_memtable, + const BlobFetcher* blob_fetcher) { // The sequence number is updated synchronously in version_set.h if (IsEmpty()) { // Avoiding recording stats for speed. @@ -1740,6 +1797,7 @@ void MemTable::MultiGet(const ReadOptions& read_options, MultiGetRange* range, saver.inplace_update_support = moptions_.inplace_update_support; saver.statistics = moptions_.statistics; saver.clock = clock_; + saver.blob_fetcher = blob_fetcher; saver.callback_ = callback; saver.is_blob_index = &iter->is_blob_index; saver.do_merge = true; @@ -1831,7 +1889,7 @@ void MemTable::MultiGet(const ReadOptions& read_options, MultiGetRange* range, *(iter->lkey), iter->max_covering_tombstone_seq, true, callback, &iter->is_blob_index, iter->value ? iter->value->GetSelf() : nullptr, iter->columns, iter->timestamp, iter->s, &(iter->merge_context), - &dummy_seq, &found_final_value, &merge_in_progress); + &dummy_seq, &found_final_value, &merge_in_progress, blob_fetcher); if (!found_final_value && merge_in_progress) { if (iter->s->ok()) { diff --git a/db/memtable.h b/db/memtable.h index 9c145e0471..24cc83330f 100644 --- a/db/memtable.h +++ b/db/memtable.h @@ -39,6 +39,7 @@ namespace ROCKSDB_NAMESPACE { +class BlobFetcher; class BlobFilePartitionManager; struct FlushJobInfo; class Mutex; @@ -236,25 +237,27 @@ class ReadOnlyMemTable { SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, const ReadOptions& read_opts, bool immutable_memtable, ReadCallback* callback = nullptr, - bool* is_blob_index = nullptr, bool do_merge = true) = 0; + bool* is_blob_index = nullptr, bool do_merge = true, + const BlobFetcher* blob_fetcher = nullptr) = 0; bool Get(const LookupKey& key, std::string* value, PinnableWideColumns* columns, std::string* timestamp, Status* s, MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, const ReadOptions& read_opts, bool immutable_memtable, ReadCallback* callback = nullptr, bool* is_blob_index = nullptr, - bool do_merge = true) { + bool do_merge = true, const BlobFetcher* blob_fetcher = nullptr) { SequenceNumber seq; return Get(key, value, columns, timestamp, s, merge_context, max_covering_tombstone_seq, &seq, read_opts, immutable_memtable, - callback, is_blob_index, do_merge); + callback, is_blob_index, do_merge, blob_fetcher); } // @param immutable_memtable Whether this memtable is immutable. Used // internally by NewRangeTombstoneIterator(). See comment above // NewRangeTombstoneIterator() for more detail. virtual void MultiGet(const ReadOptions& read_options, MultiGetRange* range, - ReadCallback* callback, bool immutable_memtable) = 0; + ReadCallback* callback, bool immutable_memtable, + const BlobFetcher* blob_fetcher = nullptr) = 0; // Get total number of entries in the mem table. // REQUIRES: external synchronization to prevent simultaneous @@ -688,10 +691,12 @@ class MemTable final : public ReadOnlyMemTable { SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, const ReadOptions& read_opts, bool immutable_memtable, ReadCallback* callback = nullptr, bool* is_blob_index = nullptr, - bool do_merge = true) override; + bool do_merge = true, + const BlobFetcher* blob_fetcher = nullptr) override; void MultiGet(const ReadOptions& read_options, MultiGetRange* range, - ReadCallback* callback, bool immutable_memtable) override; + ReadCallback* callback, bool immutable_memtable, + const BlobFetcher* blob_fetcher = nullptr) override; // If `key` exists in current memtable with type value_type and the existing // value is at least as large as the new value, updates it in-place. Otherwise @@ -991,7 +996,8 @@ class MemTable final : public ReadOnlyMemTable { std::string* value, PinnableWideColumns* columns, std::string* timestamp, Status* s, MergeContext* merge_context, SequenceNumber* seq, - bool* found_final_value, bool* merge_in_progress); + bool* found_final_value, bool* merge_in_progress, + const BlobFetcher* blob_fetcher); // Always returns non-null and assumes certain pre-checks (e.g., // is_range_del_table_empty_) are done. This is only valid during the lifetime diff --git a/db/memtable_list.cc b/db/memtable_list.cc index 2309025389..9e99c74db6 100644 --- a/db/memtable_list.cc +++ b/db/memtable_list.cc @@ -108,18 +108,19 @@ bool MemTableListVersion::Get(const LookupKey& key, std::string* value, MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, const ReadOptions& read_opts, - ReadCallback* callback, bool* is_blob_index) { + ReadCallback* callback, bool* is_blob_index, + const BlobFetcher* blob_fetcher) { return GetFromList(&memlist_, key, value, columns, timestamp, s, merge_context, max_covering_tombstone_seq, seq, read_opts, - callback, is_blob_index); + callback, is_blob_index, blob_fetcher); } void MemTableListVersion::MultiGet(const ReadOptions& read_options, - MultiGetRange* range, - ReadCallback* callback) { + MultiGetRange* range, ReadCallback* callback, + const BlobFetcher* blob_fetcher) { for (auto memtable : memlist_) { memtable->MultiGet(read_options, range, callback, - true /* immutable_memtable */); + true /* immutable_memtable */, blob_fetcher); if (range->empty()) { return; } @@ -128,12 +129,13 @@ void MemTableListVersion::MultiGet(const ReadOptions& read_options, bool MemTableListVersion::GetMergeOperands( const LookupKey& key, Status* s, MergeContext* merge_context, - SequenceNumber* max_covering_tombstone_seq, const ReadOptions& read_opts) { + SequenceNumber* max_covering_tombstone_seq, const ReadOptions& read_opts, + const BlobFetcher* blob_fetcher) { for (ReadOnlyMemTable* memtable : memlist_) { bool done = memtable->Get( key, /*value=*/nullptr, /*columns=*/nullptr, /*timestamp=*/nullptr, s, merge_context, max_covering_tombstone_seq, read_opts, - true /* immutable_memtable */, nullptr, nullptr, false); + true /* immutable_memtable */, nullptr, nullptr, false, blob_fetcher); if (done) { return true; } @@ -145,10 +147,11 @@ bool MemTableListVersion::GetFromHistory( const LookupKey& key, std::string* value, PinnableWideColumns* columns, std::string* timestamp, Status* s, MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, - const ReadOptions& read_opts, bool* is_blob_index) { + const ReadOptions& read_opts, bool* is_blob_index, + const BlobFetcher* blob_fetcher) { return GetFromList(&memlist_history_, key, value, columns, timestamp, s, merge_context, max_covering_tombstone_seq, seq, read_opts, - nullptr /*read_callback*/, is_blob_index); + nullptr /*read_callback*/, is_blob_index, blob_fetcher); } bool MemTableListVersion::GetFromList( @@ -156,17 +159,18 @@ bool MemTableListVersion::GetFromList( std::string* value, PinnableWideColumns* columns, std::string* timestamp, Status* s, MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, - const ReadOptions& read_opts, ReadCallback* callback, bool* is_blob_index) { + const ReadOptions& read_opts, ReadCallback* callback, bool* is_blob_index, + const BlobFetcher* blob_fetcher) { *seq = kMaxSequenceNumber; for (auto& memtable : *list) { assert(memtable->IsFragmentedRangeTombstonesConstructed()); SequenceNumber current_seq = kMaxSequenceNumber; - bool done = - memtable->Get(key, value, columns, timestamp, s, merge_context, - max_covering_tombstone_seq, ¤t_seq, read_opts, - true /* immutable_memtable */, callback, is_blob_index); + bool done = memtable->Get(key, value, columns, timestamp, s, merge_context, + max_covering_tombstone_seq, ¤t_seq, + read_opts, true /* immutable_memtable */, + callback, is_blob_index, true, blob_fetcher); if (*seq == kMaxSequenceNumber) { // Store the most recent sequence number of any operation on this key. // Since we only care about the most recent change, we only need to diff --git a/db/memtable_list.h b/db/memtable_list.h index b5a7be6a28..a8e36550a5 100644 --- a/db/memtable_list.h +++ b/db/memtable_list.h @@ -61,29 +61,33 @@ class MemTableListVersion { MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, const ReadOptions& read_opts, ReadCallback* callback = nullptr, - bool* is_blob_index = nullptr); + bool* is_blob_index = nullptr, + const BlobFetcher* blob_fetcher = nullptr); bool Get(const LookupKey& key, std::string* value, PinnableWideColumns* columns, std::string* timestamp, Status* s, MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, const ReadOptions& read_opts, ReadCallback* callback = nullptr, - bool* is_blob_index = nullptr) { + bool* is_blob_index = nullptr, + const BlobFetcher* blob_fetcher = nullptr) { SequenceNumber seq; return Get(key, value, columns, timestamp, s, merge_context, max_covering_tombstone_seq, &seq, read_opts, callback, - is_blob_index); + is_blob_index, blob_fetcher); } void MultiGet(const ReadOptions& read_options, MultiGetRange* range, - ReadCallback* callback); + ReadCallback* callback, + const BlobFetcher* blob_fetcher = nullptr); // Returns all the merge operands corresponding to the key by searching all // memtables starting from the most recent one. bool GetMergeOperands(const LookupKey& key, Status* s, MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, - const ReadOptions& read_opts); + const ReadOptions& read_opts, + const BlobFetcher* blob_fetcher = nullptr); // Similar to Get(), but searches the Memtable history of memtables that // have already been flushed. Should only be used from in-memory only @@ -94,17 +98,19 @@ class MemTableListVersion { Status* s, MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, const ReadOptions& read_opts, - bool* is_blob_index = nullptr); + bool* is_blob_index = nullptr, + const BlobFetcher* blob_fetcher = nullptr); bool GetFromHistory(const LookupKey& key, std::string* value, PinnableWideColumns* columns, std::string* timestamp, Status* s, MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq, const ReadOptions& read_opts, - bool* is_blob_index = nullptr) { + bool* is_blob_index = nullptr, + const BlobFetcher* blob_fetcher = nullptr) { SequenceNumber seq; return GetFromHistory(key, value, columns, timestamp, s, merge_context, max_covering_tombstone_seq, &seq, read_opts, - is_blob_index); + is_blob_index, blob_fetcher); } Status AddRangeTombstoneIterators(const ReadOptions& read_opts, Arena* arena, @@ -188,7 +194,8 @@ class MemTableListVersion { SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, const ReadOptions& read_opts, ReadCallback* callback = nullptr, - bool* is_blob_index = nullptr); + bool* is_blob_index = nullptr, + const BlobFetcher* blob_fetcher = nullptr); void AddMemTable(ReadOnlyMemTable* m); diff --git a/db/version_set.cc b/db/version_set.cc index 8de073c8ef..00450e2f9d 100644 --- a/db/version_set.cc +++ b/db/version_set.cc @@ -8163,11 +8163,12 @@ ReactiveVersionSet::ReactiveVersionSet( const MutableDBOptions& mutable_db_options, const FileOptions& _file_options, Cache* table_cache, WriteBufferManager* write_buffer_manager, WriteController* write_controller, - const std::shared_ptr& io_tracer) + const std::shared_ptr& io_tracer, const std::string& db_id, + const std::string& db_session_id) : VersionSet(dbname, imm_db_options, mutable_db_options, _file_options, table_cache, write_buffer_manager, write_controller, - /*block_cache_tracer=*/nullptr, io_tracer, /*db_id*/ "", - /*db_session_id*/ "", /*daily_offpeak_time_utc*/ "", + /*block_cache_tracer=*/nullptr, io_tracer, db_id, + db_session_id, /*daily_offpeak_time_utc*/ "", /*error_handler=*/nullptr, /*unchanging=*/false) {} ReactiveVersionSet::~ReactiveVersionSet() = default; diff --git a/db/version_set.h b/db/version_set.h index a01987002e..6a1d654714 100644 --- a/db/version_set.h +++ b/db/version_set.h @@ -1898,7 +1898,9 @@ class ReactiveVersionSet : public VersionSet { const FileOptions& _file_options, Cache* table_cache, WriteBufferManager* write_buffer_manager, WriteController* write_controller, - const std::shared_ptr& io_tracer); + const std::shared_ptr& io_tracer, + const std::string& db_id, + const std::string& db_session_id); ~ReactiveVersionSet() override; diff --git a/db/version_set_test.cc b/db/version_set_test.cc index 2c0bccf9c5..70bcce2649 100644 --- a/db/version_set_test.cc +++ b/db/version_set_test.cc @@ -1203,7 +1203,7 @@ class VersionSetTestBase { reactive_versions_ = std::make_shared( dbname_, &imm_db_options_, mutable_db_options_, env_options_, table_cache_.get(), &write_buffer_manager_, &write_controller_, - nullptr); + /*io_tracer=*/nullptr, /*db_id=*/"", /*db_session_id=*/""); imm_db_options_.db_paths.emplace_back(dbname_, std::numeric_limits::max()); } diff --git a/db/wide/db_wide_basic_test.cc b/db/wide/db_wide_basic_test.cc index d2645b2e53..94addc3c9e 100644 --- a/db/wide/db_wide_basic_test.cc +++ b/db/wide/db_wide_basic_test.cc @@ -30,6 +30,10 @@ class DBWideBasicTest : public DBTestBase { return wide_column_test_util::GetOptionsForBlobTest(GetDefaultOptions()); } + Options GetDirectWriteOptions() { + return wide_column_test_util::GetDirectWriteOptions(GetDefaultOptions()); + } + // Helper: runs the EntityBlobAfterFlush test logic with the given options. void RunEntityBlobAfterFlush(const Options& options); @@ -2159,6 +2163,57 @@ TEST_F(DBWideBasicTest, MultiGetEntityWithBlobResolution) { } } +TEST_F(DBWideBasicTest, + MultiGetBlobBackedEntityDirectWriteMemtableBatchLookup) { + // Goal: force the memtable batch MultiGet optimization to read a direct-write + // blob-backed entity from the active memtable. One key exercises merge + // resolution against a blob-backed entity base, which requires Saver to carry + // the BlobFetcher through the batched callback path. The second key proves + // the non-merge default-column path still resolves correctly in the same + // batched request. + Options options = GetDirectWriteOptions(); + options.min_blob_size = 50; + options.memtable_batch_lookup_optimization = true; + options.merge_operator = MergeOperators::CreateStringAppendOperator("|"); + + DestroyAndReopen(options); + + const std::string merge_key = "multiget_direct_write_merge_key"; + const std::string plain_key = "multiget_direct_write_plain_key"; + const std::string default_value = GenerateLargeValue(100, 'd'); + const std::string other_default_value = GenerateLargeValue(110, 'p'); + const std::string large_value = GenerateLargeValue(120, 'l'); + const std::string small_value = GenerateSmallValue(); + const std::string merge_operand = "tail"; + + WideColumns merge_columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + WideColumns plain_columns{{kDefaultWideColumnName, other_default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + + ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), + merge_key, merge_columns)); + ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), + plain_key, plain_columns)); + ASSERT_OK(db_->Merge(WriteOptions(), db_->DefaultColumnFamily(), merge_key, + merge_operand)); + + std::array keys{{merge_key, plain_key}}; + std::array values; + std::array statuses; + + db_->MultiGet(ReadOptions(), db_->DefaultColumnFamily(), keys.size(), + keys.data(), values.data(), statuses.data()); + + ASSERT_OK(statuses[0]); + ASSERT_EQ(values[0], default_value + "|" + merge_operand); + + ASSERT_OK(statuses[1]); + ASSERT_EQ(values[1], other_default_value); +} + void DBWideBasicTest::RunEntityBlobAfterFlush(const Options& options) { Reopen(options); @@ -2909,6 +2964,347 @@ TEST_F(DBWideBasicTest, MergeEntityWithBlobColumns) { } } +TEST_F(DBWideBasicTest, GetMergeOperandsWithBlobBackedEntityDefaultColumn) { + // Goal: cover both GetMergeOperands code paths that read a compacted V2 + // wide-column entity whose default column was moved to a blob file. The + // first read exercises the base-value path directly, then a merge operand is + // added so the second read exercises the merge-plus-base path. + Options options = GetBlobTestOptions(); + options.min_blob_size = 50; + options.merge_operator = MergeOperators::CreateStringAppendOperator("|"); + + DestroyAndReopen(options); + + const std::string key = "merge_operands_blob_entity"; + const std::string default_value = GenerateLargeValue(100, 'd'); + const std::string large_value = GenerateLargeValue(120, 'l'); + const std::string small_value = GenerateSmallValue(); + const std::string merge_operand = "suffix"; + + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + ASSERT_FALSE(GetBlobFileNumbers().empty()); + + { + // The compacted V2 entity becomes the first operand when there are no + // newer merge operands above it. + GetMergeOperandsOptions get_merge_opts; + get_merge_opts.expected_max_number_of_operands = 1; + + std::array merge_operands; + int number_of_operands = 0; + + ASSERT_OK(db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), + key, merge_operands.data(), &get_merge_opts, + &number_of_operands)); + ASSERT_EQ(number_of_operands, 1); + ASSERT_EQ(merge_operands[0], default_value); + } + + ASSERT_OK(db_->Merge(WriteOptions(), db_->DefaultColumnFamily(), key, + merge_operand)); + + { + // After a merge is added, GetMergeOperands must still resolve the blob + // backed base default column while traversing the older base entry below + // the newer merge operand. + GetMergeOperandsOptions get_merge_opts; + get_merge_opts.expected_max_number_of_operands = 2; + + std::array merge_operands; + int number_of_operands = 0; + + ASSERT_OK(db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), + key, merge_operands.data(), &get_merge_opts, + &number_of_operands)); + ASSERT_EQ(number_of_operands, 2); + ASSERT_EQ(merge_operands[0], default_value); + ASSERT_EQ(merge_operands[1], merge_operand); + } +} + +TEST_F(DBWideBasicTest, + GetMergeOperandsWithBlobBackedEntityDefaultColumnDirectWriteMemtable) { + // Goal: cover the pre-flush memtable path when blob direct write stores a + // blob-backed V2 entity in memory. The first GetMergeOperands call exercises + // the no-merge base-operand path. After adding a merge, DB::Get exercises + // merge resolution against the same memtable entity, and a second + // GetMergeOperands call exercises the merge-plus-base path. + Options options = GetDirectWriteOptions(); + options.min_blob_size = 50; + options.merge_operator = MergeOperators::CreateStringAppendOperator("|"); + + DestroyAndReopen(options); + + const std::string key = "direct_write_merge_operands_blob_entity"; + const std::string default_value = GenerateLargeValue(100, 'd'); + const std::string large_value = GenerateLargeValue(120, 'l'); + const std::string small_value = GenerateSmallValue(); + const std::string merge_operand = "suffix"; + const std::string expected_merged = default_value + "|" + merge_operand; + + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + + { + GetMergeOperandsOptions get_merge_opts; + get_merge_opts.expected_max_number_of_operands = 1; + + std::array merge_operands; + int number_of_operands = 0; + + ASSERT_OK(db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), + key, merge_operands.data(), &get_merge_opts, + &number_of_operands)); + ASSERT_EQ(number_of_operands, 1); + ASSERT_EQ(merge_operands[0], default_value); + } + + ASSERT_OK(db_->Merge(WriteOptions(), db_->DefaultColumnFamily(), key, + merge_operand)); + + { + PinnableSlice result; + ASSERT_OK( + db_->Get(ReadOptions(), db_->DefaultColumnFamily(), key, &result)); + ASSERT_EQ(result, expected_merged); + } + + { + GetMergeOperandsOptions get_merge_opts; + get_merge_opts.expected_max_number_of_operands = 2; + + std::array merge_operands; + int number_of_operands = 0; + + ASSERT_OK(db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), + key, merge_operands.data(), &get_merge_opts, + &number_of_operands)); + ASSERT_EQ(number_of_operands, 2); + ASSERT_EQ(merge_operands[0], default_value); + ASSERT_EQ(merge_operands[1], merge_operand); + } +} + +TEST_F(DBWideBasicTest, + GetAndGetEntityWithBlobBackedDefaultColumnDirectWriteMemtableReadOnly) { + // Goal: cover the read-only reopen path after writing a blob-backed + // direct-write entity without manually flushing it first. The test checks + // both `Get()`, which must resolve the default-column blob reference, and + // `GetEntity()`, which must eagerly resolve the entity's unresolved blob + // columns before returning them to the caller. + Options options = GetDirectWriteOptions(); + options.min_blob_size = 50; + + DestroyAndReopen(options); + + const std::string key = "readonly_direct_write_memtable_entity"; + const std::string default_value = GenerateLargeValue(100, 'd'); + const std::string large_value = GenerateLargeValue(120, 'l'); + const std::string small_value = GenerateSmallValue(); + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + + Close(); + options.avoid_flush_during_recovery = true; + ASSERT_OK(ReadOnlyReopen(options)); + + { + PinnableSlice result; + ASSERT_OK( + db_->Get(ReadOptions(), db_->DefaultColumnFamily(), key, &result)); + ASSERT_EQ(result, default_value); + } + + { + PinnableWideColumns result; + ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(), key, + &result)); + ASSERT_EQ(result.columns(), columns); + } +} + +TEST_F(DBWideBasicTest, ReadOnlyDirectWriteMemtableBlobBlockCacheTier) { + // Goal: cover the read-only memtable path under kBlockCacheTier. The test + // keeps a blob-backed direct-write entity in WAL-backed recovery state so + // resolving either Get() or GetEntity() would require blob I/O, which must + // surface as Incomplete rather than silently reading the blob file. + Options options = GetDirectWriteOptions(); + options.min_blob_size = 50; + + DestroyAndReopen(options); + + const std::string key = "readonly_direct_write_memtable_block_cache_tier"; + const std::string default_value = GenerateLargeValue(100, 'd'); + const std::string large_value = GenerateLargeValue(120, 'l'); + const std::string small_value = GenerateSmallValue(); + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + + Close(); + options.avoid_flush_during_recovery = true; + ASSERT_OK(ReadOnlyReopen(options)); + + ReadOptions read_opts; + read_opts.read_tier = kBlockCacheTier; + + { + PinnableSlice result; + const Status s = + db_->Get(read_opts, db_->DefaultColumnFamily(), key, &result); + ASSERT_TRUE(s.IsIncomplete()) << s.ToString(); + ASSERT_TRUE(result.empty()); + } + + { + PinnableWideColumns result; + const Status s = + db_->GetEntity(read_opts, db_->DefaultColumnFamily(), key, &result); + ASSERT_TRUE(s.IsIncomplete()) << s.ToString(); + ASSERT_TRUE(result.columns().empty()); + } +} + +TEST_F(DBWideBasicTest, + GetMergeOperandsWithBlobBackedEntityDefaultColumnReadOnly) { + // Goal: exercise OpenForReadOnly on a blob-backed V2 entity base in SST with + // a newer merge operand in a separate SST. The read-only DB must resolve the + // blob-backed default column for both Get() and GetMergeOperands() without + // relying on mutable memtable state. + Options options = GetBlobTestOptions(); + options.min_blob_size = 50; + options.merge_operator = MergeOperators::CreateStringAppendOperator("|"); + + DestroyAndReopen(options); + + const std::string key = "readonly_merge_operands_blob_entity"; + const std::string default_value = GenerateLargeValue(100, 'd'); + const std::string large_value = GenerateLargeValue(120, 'l'); + const std::string small_value = GenerateSmallValue(); + const std::string merge_operand = "suffix"; + const std::string expected_merged = default_value + "|" + merge_operand; + + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + ASSERT_FALSE(GetBlobFileNumbers().empty()); + + ASSERT_OK(db_->Merge(WriteOptions(), db_->DefaultColumnFamily(), key, + merge_operand)); + ASSERT_OK(Flush()); + + Close(); + ASSERT_OK(ReadOnlyReopen(options)); + + { + PinnableSlice result; + ASSERT_OK( + db_->Get(ReadOptions(), db_->DefaultColumnFamily(), key, &result)); + ASSERT_EQ(result, expected_merged); + } + + { + GetMergeOperandsOptions get_merge_opts; + get_merge_opts.expected_max_number_of_operands = 2; + + std::array merge_operands; + int number_of_operands = 0; + + ASSERT_OK(db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), + key, merge_operands.data(), &get_merge_opts, + &number_of_operands)); + ASSERT_EQ(number_of_operands, 2); + ASSERT_EQ(merge_operands[0], default_value); + ASSERT_EQ(merge_operands[1], merge_operand); + } +} + +TEST_F(DBWideBasicTest, + GetMergeOperandsWithBlobBackedEntityDefaultColumnBlockCacheTier) { + // Goal: prove the SST-backed GetMergeOperands path propagates Incomplete + // when resolving a blob-backed default column would require I/O. One key + // covers the direct base-entity path, and the other covers the merge-plus- + // base path with the merge operand flushed to a newer SST. + Options options = GetBlobTestOptions(); + options.min_blob_size = 50; + options.merge_operator = MergeOperators::CreateStringAppendOperator("|"); + + DestroyAndReopen(options); + + const std::string base_key = "merge_operands_blob_entity_cache_base"; + const std::string merge_key = "merge_operands_blob_entity_cache_merge"; + const std::string default_value = GenerateLargeValue(100, 'd'); + const std::string large_value = GenerateLargeValue(120, 'l'); + const std::string small_value = GenerateSmallValue(); + + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"col_large", large_value}, + {"col_small", small_value}}; + ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), base_key, + columns)); + ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), + merge_key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + + ASSERT_OK(db_->Merge(WriteOptions(), db_->DefaultColumnFamily(), merge_key, + "tail")); + ASSERT_OK(Flush()); + + // Reopen to clear any cached blob readers/values so kBlockCacheTier must + // report Incomplete instead of succeeding through a warm cache. + Reopen(options); + + ReadOptions read_opts; + read_opts.read_tier = kBlockCacheTier; + + { + GetMergeOperandsOptions get_merge_opts; + get_merge_opts.expected_max_number_of_operands = 1; + + std::array merge_operands; + int number_of_operands = 0; + + const Status s = db_->GetMergeOperands( + read_opts, db_->DefaultColumnFamily(), base_key, merge_operands.data(), + &get_merge_opts, &number_of_operands); + ASSERT_TRUE(s.IsIncomplete()) << s.ToString(); + } + + { + GetMergeOperandsOptions get_merge_opts; + get_merge_opts.expected_max_number_of_operands = 2; + + std::array merge_operands; + int number_of_operands = 0; + + const Status s = db_->GetMergeOperands( + read_opts, db_->DefaultColumnFamily(), merge_key, merge_operands.data(), + &get_merge_opts, &number_of_operands); + ASSERT_TRUE(s.IsIncomplete()) << s.ToString(); + } +} + TEST_F(DBWideBasicTest, MergeEntityWithBlobColumnsNoDefault) { // Test: Merge on a V2 entity without a default column. The merge result // should produce a valid entity with all original columns plus a default diff --git a/db/wide/wide_column_serialization.h b/db/wide/wide_column_serialization.h index 1b2852f6b3..ab8d412884 100644 --- a/db/wide/wide_column_serialization.h +++ b/db/wide/wide_column_serialization.h @@ -153,6 +153,8 @@ class WideColumnSerialization { // value() is initially the serialized BlobIndex bytes from the entity. // blob_columns: receives (column_index, blob_index) pairs identifying which // entries in `columns` are blob references and their decoded BlobIndex data. + // For valid input, this must agree with HasBlobColumns() about whether blob + // column references are present. static Status DeserializeV2( Slice& input, std::vector& columns, std::vector>& blob_columns); @@ -160,6 +162,8 @@ class WideColumnSerialization { // Check if the serialized entity has any blob column references. // Sets *has_blob_columns to true if version >= 2 and at least one column // has blob type; false otherwise. + // For valid input, this must agree with DeserializeV2() about whether blob + // column references are present. // Returns Status::Corruption on decode errors. static Status HasBlobColumns(const Slice& input, bool& has_blob_columns); diff --git a/memtable/wbwi_memtable.cc b/memtable/wbwi_memtable.cc index 9686eac502..0c8c298541 100644 --- a/memtable/wbwi_memtable.cc +++ b/memtable/wbwi_memtable.cc @@ -48,13 +48,18 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value, SequenceNumber* max_covering_tombstone_seq, SequenceNumber* out_seq, const ReadOptions&, bool immutable_memtable, ReadCallback* callback, - bool* is_blob_index, bool do_merge) { + bool* is_blob_index, bool do_merge, + const BlobFetcher* blob_fetcher) { assert(s->ok() || s->IsMergeInProgress()); (void)immutable_memtable; (void)timestamp; (void)columns; + (void)blob_fetcher; assert(immutable_memtable); assert(!timestamp); // TODO: support UDT + // IngestWriteBatchWithIndex() is rejected while any live column family has + // blob direct write enabled, so WBWI should never need blob resolution. + assert(blob_fetcher == nullptr); assert(assigned_seqno_.upper_bound != kMaxSequenceNumber); assert(assigned_seqno_.lower_bound != kMaxSequenceNumber); // WBWI does not support DeleteRange yet. @@ -154,10 +159,15 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value, void WBWIMemTable::MultiGet(const ReadOptions& read_options, MultiGetRange* range, ReadCallback* callback, - bool immutable_memtable) { + bool immutable_memtable, + const BlobFetcher* blob_fetcher) { (void)immutable_memtable; + (void)blob_fetcher; // Should only be used as immutable memtable. assert(immutable_memtable); + // IngestWriteBatchWithIndex() is rejected while any live column family has + // blob direct write enabled, so WBWI should never need blob resolution. + assert(blob_fetcher == nullptr); // TODO: reuse the InternalIterator created in Get(). for (auto iter = range->begin(); iter != range->end(); ++iter) { SequenceNumber dummy_seq = 0; diff --git a/memtable/wbwi_memtable.h b/memtable/wbwi_memtable.h index b1239f73de..5838a85f9a 100644 --- a/memtable/wbwi_memtable.h +++ b/memtable/wbwi_memtable.h @@ -134,10 +134,12 @@ class WBWIMemTable final : public ReadOnlyMemTable { SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq, const ReadOptions& read_opts, bool immutable_memtable, ReadCallback* callback = nullptr, bool* is_blob_index = nullptr, - bool do_merge = true) override; + bool do_merge = true, + const BlobFetcher* blob_fetcher = nullptr) override; void MultiGet(const ReadOptions& read_options, MultiGetRange* range, - ReadCallback* callback, bool immutable_memtable) override; + ReadCallback* callback, bool immutable_memtable, + const BlobFetcher* blob_fetcher = nullptr) override; uint64_t NumEntries() const override { return num_entries_; } diff --git a/table/get_context.cc b/table/get_context.cc index d51f303af2..b63758e42d 100644 --- a/table/get_context.cc +++ b/table/get_context.cc @@ -127,6 +127,10 @@ Status GetContext::SaveWideColumnEntityToPinnable(const Slice& user_key, pinnable_val_->PinSelf(value_of_default); } } else if (status.IsNotSupported()) { + if (blob_fetcher_ == nullptr) { + return Status::Corruption( + "Cannot resolve blob-backed default column without a blob fetcher"); + } // Default column is a blob reference, so resolve it into the output value. bool resolved = false; status = WideColumnSerialization::GetValueOfDefaultColumnResolvingBlobs( @@ -140,14 +144,28 @@ Status GetContext::SaveWideColumnEntityToColumns(const Slice& user_key, Cleanable* value_pinner) { assert(columns_ != nullptr); + bool has_blob_columns = false; + Status status = + WideColumnSerialization::HasBlobColumns(entity, has_blob_columns); + if (!status.ok()) { + return status; + } + if (!has_blob_columns) { + return columns_->SetWideColumnValue(entity, value_pinner); + } std::vector entity_columns; std::vector> blob_cols; Slice entity_ref = entity; - Status status = WideColumnSerialization::DeserializeV2( - entity_ref, entity_columns, blob_cols); + status = WideColumnSerialization::DeserializeV2(entity_ref, entity_columns, + blob_cols); if (status.ok()) { - if (LIKELY(blob_cols.empty())) { - return columns_->SetWideColumnValue(entity, value_pinner); + // HasBlobColumns() and DeserializeV2() must agree on whether this entity + // contains blob-valued columns. + assert(!blob_cols.empty()); + if (blob_fetcher_ == nullptr) { + return Status::Corruption( + "Cannot resolve blob-backed wide-column entity without a blob " + "fetcher"); } // TODO: Add lazy resolution support for GetEntity point lookups. This @@ -201,6 +219,37 @@ Status GetContext::SaveWideColumnEntityToColumns(const Slice& user_key, return status; } +Status GetContext::PushWideColumnEntityDefaultOperand(const Slice& user_key, + const Slice& entity, + Cleanable* value_pinner) { + Slice value_of_default; + Slice entity_ref = entity; + Status status = WideColumnSerialization::GetValueOfDefaultColumn( + entity_ref, value_of_default); + if (status.ok()) { + push_operand(value_of_default, value_pinner); + return status; + } + if (!status.IsNotSupported()) { + return status; + } + if (blob_fetcher_ == nullptr) { + return Status::Corruption( + "Cannot resolve blob-backed default column without a blob fetcher"); + } + + PinnableSlice resolved_default; + bool resolved = false; + status = WideColumnSerialization::GetValueOfDefaultColumnResolvingBlobs( + entity, user_key, blob_fetcher_, resolved_default, resolved); + if (status.ok()) { + // Resolved blob values are backed by this stack-local PinnableSlice, so + // copy them into MergeContext instead of pinning their storage. + push_operand(Slice(resolved_default), nullptr); + } + return status; +} + void GetContext::SaveValue(const Slice& value, SequenceNumber /*seq*/, Cleanable* value_pinner) { assert(state_ == kNotFound); @@ -329,7 +378,7 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, const Slice& value, bool* matched, Status* read_status, Cleanable* value_pinner) { assert(matched); - assert((state_ != kMerge && parsed_key.type != kTypeMerge) || + assert((State() != kMerge && parsed_key.type != kTypeMerge) || merge_context_ != nullptr); if (ucmp_->EqualWithoutTimestamp(parsed_key.user_key, user_key_)) { *matched = true; @@ -400,7 +449,7 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, case kTypeValuePreferredSeqno: case kTypeBlobIndex: case kTypeWideColumnEntity: - assert(state_ == kNotFound || state_ == kMerge); + assert(State() == kNotFound || State() == kMerge); if (type == kTypeValuePreferredSeqno) { unpacked_value = ParsePackedValueForValue(value); } @@ -416,7 +465,7 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, *is_blob_index_ = (type == kTypeBlobIndex); } - if (kNotFound == state_) { + if (State() == kNotFound) { state_ = kFound; if (do_merge_) { if (type == kTypeBlobIndex && ucmp_->timestamp_size() != 0) { @@ -431,8 +480,10 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, if (!s.ok()) { if (s.IsIncomplete()) { MarkKeyMayExist(); + *read_status = s; } else { state_ = kCorrupt; + *read_status = s; } return false; } @@ -456,8 +507,10 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, if (!s.ok()) { if (s.IsIncomplete()) { MarkKeyMayExist(); + *read_status = s; } else { state_ = kCorrupt; + *read_status = s; } return false; } @@ -478,23 +531,24 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, Slice blob_value(pin_val); push_operand(blob_value, nullptr); } else if (type == kTypeWideColumnEntity) { - Slice value_copy = unpacked_value; - Slice value_of_default; - - if (!WideColumnSerialization::GetValueOfDefaultColumn( - value_copy, value_of_default) - .ok()) { - state_ = kCorrupt; + const Status s = PushWideColumnEntityDefaultOperand( + parsed_key.user_key, unpacked_value, value_pinner); + if (!s.ok()) { + if (s.IsIncomplete()) { + MarkKeyMayExist(); + *read_status = s; + } else { + state_ = kCorrupt; + *read_status = s; + } return false; } - - push_operand(value_of_default, value_pinner); } else { assert(type == kTypeValue || type == kTypeValuePreferredSeqno); push_operand(unpacked_value, value_pinner); } } - } else if (kMerge == state_) { + } else if (State() == kMerge) { assert(merge_operator_ != nullptr); if (type == kTypeBlobIndex) { PinnableSlice pin_val; @@ -505,7 +559,11 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, Slice blob_value(pin_val); state_ = kFound; if (do_merge_) { - MergeWithPlainBaseValue(blob_value); + const Status s = MergeWithPlainBaseValue(blob_value); + if (!s.ok()) { + *read_status = s; + return false; + } } else { // It means this function is called as part of DB GetMergeOperands // API and the current value should be part of @@ -516,29 +574,38 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, state_ = kFound; if (do_merge_) { - MergeWithWideColumnBaseValue(unpacked_value); + const Status s = MergeWithWideColumnBaseValue(unpacked_value); + if (!s.ok()) { + *read_status = s; + return false; + } } else { // It means this function is called as part of DB GetMergeOperands // API and the current value should be part of // merge_context_->operand_list - Slice value_copy = unpacked_value; - Slice value_of_default; - - if (!WideColumnSerialization::GetValueOfDefaultColumn( - value_copy, value_of_default) - .ok()) { - state_ = kCorrupt; + const Status s = PushWideColumnEntityDefaultOperand( + parsed_key.user_key, unpacked_value, value_pinner); + if (!s.ok()) { + if (s.IsIncomplete()) { + MarkKeyMayExist(); + *read_status = s; + } else { + state_ = kCorrupt; + *read_status = s; + } return false; } - - push_operand(value_of_default, value_pinner); } } else { assert(type == kTypeValue || type == kTypeValuePreferredSeqno); state_ = kFound; if (do_merge_) { - MergeWithPlainBaseValue(unpacked_value); + const Status s = MergeWithPlainBaseValue(unpacked_value); + if (!s.ok()) { + *read_status = s; + return false; + } } else { // It means this function is called as part of DB GetMergeOperands // API and the current value should be part of @@ -555,13 +622,17 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, case kTypeRangeDeletion: // TODO(noetzli): Verify correctness once merge of single-deletes // is supported - assert(state_ == kNotFound || state_ == kMerge); - if (kNotFound == state_) { + assert(State() == kNotFound || State() == kMerge); + if (State() == kNotFound) { state_ = kDeleted; - } else if (kMerge == state_) { + } else if (State() == kMerge) { state_ = kFound; if (do_merge_) { - MergeWithNoBaseValue(); + const Status s = MergeWithNoBaseValue(); + if (!s.ok()) { + *read_status = s; + return false; + } } // If do_merge_ = false then the current value shouldn't be part of // merge_context_->operand_list @@ -569,7 +640,7 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, return false; case kTypeMerge: - assert(state_ == kNotFound || state_ == kMerge); + assert(State() == kNotFound || State() == kMerge); state_ = kMerge; // value_pinner is not set from plain_table_reader.cc for example. push_operand(value, value_pinner); @@ -579,7 +650,11 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, merge_operator_->ShouldMerge( merge_context_->GetOperandsDirectionBackward())) { state_ = kFound; - MergeWithNoBaseValue(); + const Status s = MergeWithNoBaseValue(); + if (!s.ok()) { + *read_status = s; + return false; + } return false; } if (merge_context_->get_merge_operands_options != nullptr && @@ -597,26 +672,27 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key, } } - // state_ could be Corrupt, merge or notfound + // State() could be Corrupt, merge or notfound return false; } -void GetContext::PostprocessMerge(const Status& merge_status) { +Status GetContext::PostprocessMerge(const Status& merge_status) { if (!merge_status.ok()) { if (merge_status.subcode() == Status::SubCode::kMergeOperatorFailed) { state_ = kMergeOperatorFailed; } else { state_ = kCorrupt; } - return; + return merge_status; } if (LIKELY(pinnable_val_ != nullptr)) { pinnable_val_->PinSelf(); } + return Status::OK(); } -void GetContext::MergeWithNoBaseValue() { +Status GetContext::MergeWithNoBaseValue() { assert(do_merge_); assert(pinnable_val_ || columns_); assert(!pinnable_val_ || !columns_); @@ -628,10 +704,10 @@ void GetContext::MergeWithNoBaseValue() { merge_context_->GetOperands(), logger_, statistics_, clock_, /* update_num_ops_stats */ true, /* op_failure_scope */ nullptr, pinnable_val_ ? pinnable_val_->GetSelf() : nullptr, columns_); - PostprocessMerge(s); + return PostprocessMerge(s); } -void GetContext::MergeWithPlainBaseValue(const Slice& value) { +Status GetContext::MergeWithPlainBaseValue(const Slice& value) { assert(do_merge_); assert(pinnable_val_ || columns_); assert(!pinnable_val_ || !columns_); @@ -643,10 +719,10 @@ void GetContext::MergeWithPlainBaseValue(const Slice& value) { merge_context_->GetOperands(), logger_, statistics_, clock_, /* update_num_ops_stats */ true, /* op_failure_scope */ nullptr, pinnable_val_ ? pinnable_val_->GetSelf() : nullptr, columns_); - PostprocessMerge(s); + return PostprocessMerge(s); } -void GetContext::MergeWithWideColumnBaseValue(const Slice& entity) { +Status GetContext::MergeWithWideColumnBaseValue(const Slice& entity) { assert(do_merge_); assert(pinnable_val_ || columns_); assert(!pinnable_val_ || !columns_); @@ -655,16 +731,16 @@ void GetContext::MergeWithWideColumnBaseValue(const Slice& entity) { // supports V1 format. std::string resolved_entity; Slice effective_entity; - const Status s_resolve = WideColumnSerialization::ResolveEntityForMerge( + Status s_resolve = WideColumnSerialization::ResolveEntityForMerge( entity, user_key_, blob_fetcher_, nullptr /* prefetch_buffers */, resolved_entity, effective_entity); if (!s_resolve.ok()) { if (s_resolve.IsIncomplete()) { MarkKeyMayExist(); - return; + return s_resolve; } state_ = kCorrupt; - return; + return s_resolve; } // `op_failure_scope` (an output parameter) is not provided (set to nullptr) @@ -674,7 +750,7 @@ void GetContext::MergeWithWideColumnBaseValue(const Slice& entity) { merge_context_->GetOperands(), logger_, statistics_, clock_, /* update_num_ops_stats */ true, /* op_failure_scope */ nullptr, pinnable_val_ ? pinnable_val_->GetSelf() : nullptr, columns_); - PostprocessMerge(s); + return PostprocessMerge(s); } bool GetContext::GetBlobValue(const Slice& user_key, const Slice& blob_index, @@ -741,11 +817,16 @@ Status replayGetContextLog(const Slice& replay_log, const Slice& user_key, (void)ret; - Status read_status; - get_context->SaveValue(ikey, value, &dont_care, &read_status, value_pinner); + Status read_status = Status::OK(); + const bool keep_replaying = get_context->SaveValue( + ikey, value, &dont_care, &read_status, value_pinner); if (!read_status.ok()) { return read_status; } + if (!keep_replaying) { + // SaveValue() reached a terminal state for this row-cache replay. + break; + } } return Status::OK(); } diff --git a/table/get_context.h b/table/get_context.h index ac9fe1dcfb..8a033a964d 100644 --- a/table/get_context.h +++ b/table/get_context.h @@ -7,6 +7,7 @@ #include #include "db/read_callback.h" +#include "rocksdb/status.h" #include "rocksdb/types.h" namespace ROCKSDB_NAMESPACE { @@ -19,7 +20,6 @@ class MergeOperator; class PinnableWideColumns; class PinnedIteratorsManager; class Statistics; -class Status; class SystemClock; struct ParsedInternalKey; @@ -121,6 +121,10 @@ class GetContext { PinnedIteratorsManager* _pinned_iters_mgr = nullptr, ReadCallback* callback = nullptr, bool* is_blob_index = nullptr, uint64_t tracing_get_id = 0, BlobFetcher* blob_fetcher = nullptr); + // emplace-only; default construction and move assignment are intentionally + // disabled. + GetContext(GetContext&&) noexcept = default; + GetContext& operator=(GetContext&&) noexcept = delete; GetContext() = delete; @@ -207,16 +211,19 @@ class GetContext { Status SaveWideColumnEntityToColumns(const Slice& user_key, const Slice& entity, Cleanable* value_pinner); + Status PushWideColumnEntityDefaultOperand(const Slice& user_key, + const Slice& entity, + Cleanable* value_pinner); // Helper method that postprocesses the results of merge operations, e.g. it // sets the state correctly upon merge errors. - void PostprocessMerge(const Status& merge_status); + Status PostprocessMerge(const Status& merge_status); // The following methods perform the actual merge operation for the // no base value/plain base value/wide-column base value cases. - void MergeWithNoBaseValue(); - void MergeWithPlainBaseValue(const Slice& value); - void MergeWithWideColumnBaseValue(const Slice& entity); + Status MergeWithNoBaseValue(); + Status MergeWithPlainBaseValue(const Slice& value); + Status MergeWithWideColumnBaseValue(const Slice& entity); bool GetBlobValue(const Slice& user_key, const Slice& blob_index, PinnableSlice* blob_value, Status* read_status); diff --git a/table/sst_file_reader.cc b/table/sst_file_reader.cc index ec3993c7ed..fc7dee82cf 100644 --- a/table/sst_file_reader.cc +++ b/table/sst_file_reader.cc @@ -157,6 +157,8 @@ std::vector SstFileReader::MultiGet( statuses[i] = Status::MergeInProgress(); break; case GetContext::kCorrupt: + statuses[i] = Status::Corruption(); + break; case GetContext::kUnexpectedBlobIndex: case GetContext::kMergeOperatorFailed: statuses[i] = Status::Corruption(); @@ -218,6 +220,8 @@ Status SstFileReader::Get(const ReadOptions& roptions, const Slice& key, status = Status::MergeInProgress(); break; case GetContext::kCorrupt: + status = Status::Corruption(); + break; case GetContext::kUnexpectedBlobIndex: case GetContext::kMergeOperatorFailed: status = Status::Corruption(); diff --git a/table/sst_file_reader_test.cc b/table/sst_file_reader_test.cc index eba5184830..2d8f8df7b1 100644 --- a/table/sst_file_reader_test.cc +++ b/table/sst_file_reader_test.cc @@ -864,6 +864,47 @@ TEST_P(SstFileReaderTableGetTest, Basic) { Close(); } +TEST_P(SstFileReaderTableGetTest, BlobBackedWideColumnDefaultWithoutFetcher) { + // Goal: cover the SstFileReader path where GetContext has no BlobFetcher. + // The test writes a blob-backed wide-column entity, compacts it so the SST + // stores blob references, and then verifies SstFileReader reports a clean + // Corruption status instead of dereferencing a null BlobFetcher. + Options options = CurrentOptions(); + options.disable_auto_compactions = true; + options.enable_blob_files = true; + options.min_blob_size = 50; + + DestroyAndReopen(options); + + const std::string key = "blob_backed_entity"; + const std::string default_value(100, 'd'); + WideColumns columns{{kDefaultWideColumnName, default_value}, + {"meta", "inline"}}; + + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + + std::vector files; + dbfull()->GetLiveFilesMetaData(&files); + ASSERT_EQ(files.size(), 1); + std::string file_name = files[0].directory + "/" + files[0].relative_filename; + + SstFileReader reader(options); + ASSERT_OK(reader.Open(file_name)); + + std::vector keys = {key}; + std::vector values; + auto statuses = DoGet(reader, keys, &values); + + ASSERT_TRUE(statuses[0].IsCorruption()) << statuses[0].ToString(); + ASSERT_NE(statuses[0].ToString().find("blob fetcher"), std::string::npos) + << statuses[0].ToString(); + + Close(); +} + INSTANTIATE_TEST_CASE_P(SingleAndMulti, SstFileReaderTableGetTest, testing::Bool()); diff --git a/unreleased_history/bug_fixes/blob_backed_wide_column_merge_reads.md b/unreleased_history/bug_fixes/blob_backed_wide_column_merge_reads.md new file mode 100644 index 0000000000..c691137ce5 --- /dev/null +++ b/unreleased_history/bug_fixes/blob_backed_wide_column_merge_reads.md @@ -0,0 +1,3 @@ +Fixed blob-backed wide-column merge reads to preserve correct status +propagation and resolution across memtable, read-only, and secondary DB +paths. diff --git a/unreleased_history/bug_fixes/merge_reads_preserve_precise_statuses.md b/unreleased_history/bug_fixes/merge_reads_preserve_precise_statuses.md new file mode 100644 index 0000000000..f795c2ac67 --- /dev/null +++ b/unreleased_history/bug_fixes/merge_reads_preserve_precise_statuses.md @@ -0,0 +1 @@ +Fixed merge reads against wide-column/blob-backed base values to preserve precise failure statuses, including `GetMergeOperands()` and direct-write memtable reads.