mirror of
https://github.com/facebook/rocksdb.git
synced 2026-10-23 08:09:02 +03:00
Unify lazy wide-column blob read verify policy; tool coverage (#15115)
Summary:
Follow-ups to the lazy wide-column read API (DB::GetEntityLazy / MultiGetEntityLazy). Three related, low-risk changes under one theme; no public API change.
1. Internal read-API cleanup (retires TODO(lazy-blob-resolution-cleanup)). The lazy byte-range work had left a whole/range/force-verify method fan-out on the blob read path: VersionBlobFetcherBase::FetchBlobRange plus a separate FetchBlobForceVerify, and SameFileBlobReader::GetSameFileBlob paired with GetSameFileBlobRange, with the (verify_checksums, force_verify) decision duplicated across branches. This folds them into a single 3-valued BlobVerifyPolicy (kVerifyIfPresent / kVerifyIfNoAmplification / kSkip, added to db/blob/blob_constants.h) derived once at the resolver boundary and threaded down. FetchBlobRange and GetSameFileBlob now take a range descriptor (range_length == kWholeBlobLength means the whole value) plus the policy; FetchBlobForceVerify and the *Range twins are removed. The public LazyColumnReadRequest::force_verify bool is unchanged. The widely-called lower-layer whole-value primitives (BlobFileReader / BlobSource / Version GetBlob) are intentionally left as delegated primitives to bound blast radius; folding those is a later cleanup.
2. Embedded block-cache-tier quirk fix (retires TODO(lazy-blob-resolution-quirks)). ValidateEmbeddedBlobIndex rejected read_tier == kBlockCacheTier before the blob-cache probe, so a same-file (embedded) column already in the blob cache returned Incomplete under kBlockCacheTier -- unlike the separate-file path, which probes the cache first. The up-front rejection moves to the two no-cache paths (ResolveEmbeddedBlob / ResolveEmbeddedBlobPinned); the cache-probing paths now serve a cached record under kBlockCacheTier, matching separate-file behavior. This is a shared lower-layer change: the non-lazy Get / GetEntity and iterator embedded read paths (get_context / MaybeResolveEmbeddedValue -> ResolveEmbeddedBlobCached) get the same corrected behavior, not just the lazy resolver.
3. Coverage. db_bench gains a readrandomentitylazy benchmark (GetEntityLazy + MultiResolve of --lazy_entity_read_length bytes per column) for measuring byte-range (partial) blob read savings; it requires open_files == -1. The crash test now also fuzzes LazyColumnReadRequest::force_verify in the lazy-vs-eager differential (force_verify affects only verification, not the returned bytes, so the comparison is unchanged).
This also sets up the next planned step -- cross-key coalescing for MultiGetEntityLazy / LazyWideColumnsBatch::MultiResolve. The internal BlobReadRequest already carries a {offset, len} range descriptor, and with the verify decision now a single BlobVerifyPolicy rather than a bool plus a separate force-verify method, a batched resolve can group a vector of range+policy requests per (CF, Version, blob file) and coalesce them onto the existing MultiGetBlob machinery -- without re-introducing whole/range/force-verify method twins in the multi-read path.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/15115
Test Plan:
- New unit test DBLazyEntityTest.EmbeddedBlockCacheTierServedFromCachedValue: a cached embedded column serves whole- and sub-range reads under kBlockCacheTier instead of returning Incomplete. Existing force_verify / BlockCacheTier tests are unchanged (public behavior preserved).
- New unit test DBBlobIndexTest.EmbeddedBlobWideColumnGetEntityBlockCacheTierServesCached: covers the non-lazy GetEntity path -- a cached embedded column is served under kBlockCacheTier; the existing ...GetEntityBlockCacheTier test still asserts a genuine miss yields Incomplete.
- Crash test: extended the existing lazy differential to fuzz force_verify (reuses --lazy_entity_read_one_in).
- db_bench: readrandomentitylazy over 2000 entities x 2 uncompressed embedded blob columns (4000 B each), reading 100 B per column:
| metric | value |
|---|---|
| rocksdb.blobdb.lazy.partial.read.count | 4000 |
| rocksdb.blobdb.lazy.read.bytes | 400000 |
| rocksdb.blobdb.lazy.partial.bytes.saved | 15600000 |
i.e. 100 B read per column instead of the 4000 B whole-column value.
Reviewed By: xingbowang
Differential Revision: D116334969
Pulled By: pdillinger
fbshipit-source-id: cebe90456b23be0f5be60cdc4bff85c8a37b5442
This commit is contained in:
committed by
meta-codesync[bot]
parent
575b6df1b8
commit
495c21dc5c
@@ -5,7 +5,9 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
#include <limits>
|
||||
|
||||
#include "rocksdb/rocksdb_namespace.h"
|
||||
|
||||
@@ -31,4 +33,43 @@ constexpr uint64_t kInvalidBlobFileNumber = 0;
|
||||
constexpr uint64_t kCurrentFileBlobIndexFileNumber = kInvalidBlobFileNumber;
|
||||
static_assert(kCurrentFileBlobIndexFileNumber == kInvalidBlobFileNumber);
|
||||
|
||||
// Sentinel range length meaning "the whole blob value" for the unified
|
||||
// blob-read primitives, which take a range descriptor {range_offset,
|
||||
// range_length}; range_length == kWholeBlobLength selects the whole value and
|
||||
// any other value selects the strict sub-range
|
||||
// [range_offset, range_offset + range_length).
|
||||
//
|
||||
// This is the internal counterpart of the public kLazyWholeColumn
|
||||
// (include/rocksdb/lazy_wide_columns.h); both are the maximum size_t. The lazy
|
||||
// resolver does not forward the public value as a range descriptor -- it
|
||||
// normalizes a whole-column request to kWholeBlobLength -- so the two are not
|
||||
// required to be equal, but both rely on the sentinel exceeding any real
|
||||
// blob/column size so a "whole" request is never mistaken for a strict
|
||||
// sub-range. Keep them in sync.
|
||||
inline constexpr size_t kWholeBlobLength = std::numeric_limits<size_t>::max();
|
||||
|
||||
// How aggressively a blob read verifies the record's checksum. Derived once at
|
||||
// the lazy read-path boundary (ReadPathBlobResolver) from
|
||||
// (ReadOptions::verify_checksums, LazyColumnReadRequest::force_verify) and
|
||||
// threaded down through the blob-read primitives, replacing the earlier mix of
|
||||
// a verify_checksums bool and a separate force-verify method. (Not literally a
|
||||
// constant, but kept here as a small shared blob-read enum rather than its own
|
||||
// header.)
|
||||
//
|
||||
// - kVerifyIfPresent: verify the whole-record checksum whenever the format
|
||||
// provides one, even if ReadOptions::verify_checksums is off. A byte-range
|
||||
// request under this policy is escalated to a whole-record read (a strict
|
||||
// sub-range cannot cover the checksum). This is the public force_verify.
|
||||
// - kVerifyIfNoAmplification: verify when the checksummable unit is already
|
||||
// being read -- a whole-value read verifies (the check is "free"), while a
|
||||
// strict sub-range read skips verification rather than amplify the read to
|
||||
// the whole record. Corresponds to verify_checksums on, force_verify off.
|
||||
// - kSkip: never verify. Corresponds to verify_checksums off, force_verify
|
||||
// off.
|
||||
enum class BlobVerifyPolicy : uint8_t {
|
||||
kVerifyIfPresent,
|
||||
kVerifyIfNoAmplification,
|
||||
kSkip,
|
||||
};
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -44,43 +44,41 @@ Status VersionBlobFetcherBase::FetchBlob(const Slice& user_key,
|
||||
prefetch_buffer, blob_value, bytes_read);
|
||||
}
|
||||
|
||||
Status VersionBlobFetcherBase::FetchBlobRange(const Slice& user_key,
|
||||
const BlobIndex& blob_index,
|
||||
uint64_t range_offset,
|
||||
size_t range_length,
|
||||
PinnableSlice* blob_value,
|
||||
uint64_t* bytes_read) const {
|
||||
// Only ever called when SupportsRangeRead() holds (a Version to read through,
|
||||
// no direct-write fallback); the direct-write path resolves whole records and
|
||||
// has no range variant.
|
||||
assert(SupportsRangeRead());
|
||||
assert(version_);
|
||||
return version_->GetBlobRange(read_options(), user_key, blob_index,
|
||||
range_offset, range_length, blob_value,
|
||||
bytes_read);
|
||||
}
|
||||
|
||||
Status VersionBlobFetcherBase::FetchBlobForceVerify(
|
||||
const Slice& user_key, const BlobIndex& blob_index,
|
||||
FilePrefetchBuffer* prefetch_buffer, PinnableSlice* blob_value,
|
||||
uint64_t* bytes_read) const {
|
||||
if (read_options().verify_checksums) {
|
||||
// Verification already happens on the normal path.
|
||||
Status VersionBlobFetcherBase::FetchBlobRange(
|
||||
const Slice& user_key, const BlobIndex& blob_index, uint64_t range_offset,
|
||||
size_t range_length, BlobVerifyPolicy verify_policy,
|
||||
PinnableSlice* blob_value, uint64_t* bytes_read) const {
|
||||
if (range_length == kWholeBlobLength) {
|
||||
// Whole-value read. Delegate to FetchBlob (which handles the direct-write
|
||||
// fallback). For kVerifyIfPresent, force whole-record checksum verification
|
||||
// even when ReadOptions::verify_checksums is off, via a transient fetcher
|
||||
// over a verify-enabled copy of the read options (both locals outlive the
|
||||
// borrowed-options VersionBlobFetcher).
|
||||
constexpr FilePrefetchBuffer* prefetch_buffer = nullptr;
|
||||
if (verify_policy == BlobVerifyPolicy::kVerifyIfPresent &&
|
||||
!read_options().verify_checksums) {
|
||||
ReadOptions verify_read_options = read_options();
|
||||
verify_read_options.verify_checksums = true;
|
||||
VersionBlobFetcher verifying_fetcher(version_, verify_read_options,
|
||||
blob_file_cache_,
|
||||
allow_write_path_fallback_);
|
||||
return verifying_fetcher.FetchBlob(user_key, blob_index, prefetch_buffer,
|
||||
blob_value, bytes_read);
|
||||
}
|
||||
return FetchBlob(user_key, blob_index, prefetch_buffer, blob_value,
|
||||
bytes_read);
|
||||
}
|
||||
|
||||
// Force verification via a transient fetcher over a verify-enabled copy of
|
||||
// the read options. `verify_read_options` outlives `verifying_fetcher` (both
|
||||
// are local to this call), so the borrowed-options VersionBlobFetcher is
|
||||
// safe.
|
||||
ReadOptions verify_read_options = read_options();
|
||||
verify_read_options.verify_checksums = true;
|
||||
VersionBlobFetcher verifying_fetcher(version_, verify_read_options,
|
||||
blob_file_cache_,
|
||||
allow_write_path_fallback_);
|
||||
return verifying_fetcher.FetchBlob(user_key, blob_index, prefetch_buffer,
|
||||
blob_value, bytes_read);
|
||||
// Strict sub-range read: value bytes only, no whole-record checksum, no cache
|
||||
// fill. Only ever chosen when SupportsRangeRead() holds (a Version to read
|
||||
// through, no direct-write fallback) and verification is not forced (a strict
|
||||
// sub-range cannot cover the whole-record checksum).
|
||||
assert(SupportsRangeRead());
|
||||
assert(version_);
|
||||
assert(verify_policy != BlobVerifyPolicy::kVerifyIfPresent);
|
||||
return version_->GetBlobRange(read_options(), user_key, blob_index,
|
||||
range_offset, range_length, blob_value,
|
||||
bytes_read);
|
||||
}
|
||||
|
||||
Status EmbeddedAwareBlobFetcher::FetchBlob(const Slice& user_key,
|
||||
@@ -94,9 +92,16 @@ Status EmbeddedAwareBlobFetcher::FetchBlob(const Slice& user_key,
|
||||
// Same-file ("embedded") blob records live in the current SST and are read
|
||||
// by the SameFileBlobReader (the BlockBasedTable). A separate blob-file
|
||||
// prefetch buffer and bytes_read counter do not apply here (embedded reads
|
||||
// account their own BLOB_DB_* stats via BlobSource).
|
||||
return same_file_reader_->GetSameFileBlob(read_options(), blob_index,
|
||||
blob_value);
|
||||
// account their own BLOB_DB_* stats via BlobSource). This is a whole-value
|
||||
// read; the verify policy follows read_options().verify_checksums (the
|
||||
// decorator carries no separate force-verify signal).
|
||||
const BlobVerifyPolicy verify_policy =
|
||||
read_options().verify_checksums
|
||||
? BlobVerifyPolicy::kVerifyIfNoAmplification
|
||||
: BlobVerifyPolicy::kSkip;
|
||||
return same_file_reader_->GetSameFileBlob(
|
||||
read_options(), blob_index,
|
||||
/*range_offset=*/0, kWholeBlobLength, verify_policy, blob_value);
|
||||
}
|
||||
|
||||
if (base_ == nullptr) {
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
|
||||
#include <utility>
|
||||
|
||||
#include "db/blob/blob_constants.h"
|
||||
#include "rocksdb/options.h"
|
||||
#include "rocksdb/status.h"
|
||||
|
||||
@@ -63,37 +64,26 @@ class VersionBlobFetcherBase : public BlobFetcher {
|
||||
PinnableSlice* blob_value,
|
||||
uint64_t* bytes_read) const override;
|
||||
|
||||
// Fetches a byte sub-range [range_offset, range_offset + range_length) of an
|
||||
// *uncompressed*, separate-file blob's value into *blob_value, reading only
|
||||
// those bytes on a cache miss (see Version::GetBlobRange /
|
||||
// BlobSource::GetBlobRange). Only valid when SupportsRangeRead() is true (a
|
||||
// Version to read through and no direct-write fallback); callers check that
|
||||
// first and take the whole-value FetchBlob path otherwise. Not routed through
|
||||
// the same-file (embedded) decorator -- same-file range reads are a later
|
||||
// phase.
|
||||
// Fetches the value of a separate-file blob reference into *blob_value,
|
||||
// applying `verify_policy` (see BlobVerifyPolicy). range_length ==
|
||||
// kWholeBlobLength reads the whole value (delegating to FetchBlob, which also
|
||||
// handles the direct-write fallback, and forcing whole-record checksum
|
||||
// verification for kVerifyIfPresent even when ReadOptions::verify_checksums
|
||||
// is off). Any other length reads only the sub-range [range_offset,
|
||||
// range_offset + range_length) directly from the blob file, reading just
|
||||
// those bytes on a cache miss -- no record-header/key read, no whole-record
|
||||
// checksum, no cache fill (see Version::GetBlobRange /
|
||||
// BlobSource::GetBlobRange). A strict sub-range is only valid when
|
||||
// SupportsRangeRead() is true and verify_policy != kVerifyIfPresent
|
||||
// (verifying a strict sub-range would have to amplify the read to the whole
|
||||
// record); callers check that first and take the whole-value path otherwise.
|
||||
// Not routed through the same-file (embedded) decorator -- same-file
|
||||
// references are resolved by the caller through a SameFileBlobReader.
|
||||
Status FetchBlobRange(const Slice& user_key, const BlobIndex& blob_index,
|
||||
uint64_t range_offset, size_t range_length,
|
||||
BlobVerifyPolicy verify_policy,
|
||||
PinnableSlice* blob_value, uint64_t* bytes_read) const;
|
||||
|
||||
// Fetches the whole blob with checksum verification forced on, regardless of
|
||||
// this fetcher's ReadOptions::verify_checksums. Used to honor a lazy read's
|
||||
// force_verify when verify_checksums is off: the caller wants the record's
|
||||
// checksum checked even though the global option would skip it. When
|
||||
// verify_checksums is already on this is just FetchBlob.
|
||||
//
|
||||
// TODO(lazy-blob-resolution-cleanup): fold FetchBlobRange /
|
||||
// FetchBlobForceVerify (and the whole-vs-range GetBlob twins throughout the
|
||||
// blob stack) into one parameterized primitive taking a BlobReadRequest -- a
|
||||
// range descriptor ({offset, len}; whole == [0, npos)) plus a 3-valued verify
|
||||
// policy (must-verify-if-present / verify-if-no-amplification / ignore). That
|
||||
// retires this method and the range/whole method pairs. See the lazy blob
|
||||
// resolution plan (deferred for feature velocity).
|
||||
Status FetchBlobForceVerify(const Slice& user_key,
|
||||
const BlobIndex& blob_index,
|
||||
FilePrefetchBuffer* prefetch_buffer,
|
||||
PinnableSlice* blob_value,
|
||||
uint64_t* bytes_read) const;
|
||||
|
||||
// True if this fetcher has enough context to resolve a (non-inlined,
|
||||
// non-same-file) blob reference: either a Version to read through, or the
|
||||
// direct-write fallback (a blob file cache) enabled. Callers that may hold a
|
||||
|
||||
@@ -416,7 +416,8 @@ Status BlobSource::GetBlobRange(const ReadOptions& read_options,
|
||||
|
||||
const bool no_io = read_options.read_tier == kBlockCacheTier;
|
||||
if (no_io) {
|
||||
return Status::Incomplete("Cannot read blob(s): no disk I/O allowed");
|
||||
s = Status::Incomplete("Cannot read blob(s): no disk I/O allowed");
|
||||
return s;
|
||||
}
|
||||
|
||||
// Cache miss: read only the requested sub-range from the file. The result is
|
||||
@@ -501,7 +502,8 @@ Status BlobSource::GetSimpleGen2Blob(
|
||||
|
||||
const bool no_io = read_options.read_tier == kBlockCacheTier;
|
||||
if (no_io) {
|
||||
return Status::Incomplete("Cannot read blob(s): no disk I/O allowed");
|
||||
s = Status::Incomplete("Cannot read blob(s): no disk I/O allowed");
|
||||
return s;
|
||||
}
|
||||
|
||||
// Cache miss (or no cache configured). Read the record into a buffer
|
||||
@@ -601,7 +603,8 @@ Status BlobSource::GetSimpleGen2BlobRange(
|
||||
|
||||
const bool no_io = read_options.read_tier == kBlockCacheTier;
|
||||
if (no_io) {
|
||||
return Status::Incomplete("Cannot read blob(s): no disk I/O allowed");
|
||||
s = Status::Incomplete("Cannot read blob(s): no disk I/O allowed");
|
||||
return s;
|
||||
}
|
||||
|
||||
// Cache miss: read only the requested sub-range from the file. The result is
|
||||
|
||||
@@ -1032,6 +1032,53 @@ TEST_F(DBBlobIndexTest, EmbeddedBlobWideColumnGetEntityBlockCacheTier) {
|
||||
ASSERT_TRUE(s.IsIncomplete()) << s.ToString();
|
||||
}
|
||||
|
||||
// Companion to the test above (genuine miss -> Incomplete): with a blob cache
|
||||
// configured, a warm read populates the embedded record into the blob cache, so
|
||||
// a subsequent kBlockCacheTier GetEntity serves it from cache instead of
|
||||
// returning Incomplete. This exercises the non-lazy Get/GetEntity embedded path
|
||||
// (get_context -> EmbeddedAwareBlobFetcher -> GetSameFileBlob whole ->
|
||||
// ResolveEmbeddedBlobCached), which shares the block-cache-tier behavior with
|
||||
// the lazy resolver (a same-file column already in the blob cache is served).
|
||||
TEST_F(DBBlobIndexTest,
|
||||
EmbeddedBlobWideColumnGetEntityBlockCacheTierServesCached) {
|
||||
Options options = GetTestOptions();
|
||||
options.create_if_missing = true;
|
||||
LRUCacheOptions co;
|
||||
co.capacity = 8 << 20;
|
||||
options.blob_cache = NewLRUCache(co);
|
||||
DestroyAndReopen(options);
|
||||
|
||||
const std::string sst_path = dbname_ + "/embedded_wc_blockcache_hit.sst";
|
||||
SstFileWriterEmbeddedBlobOptions embedded_blob_options;
|
||||
embedded_blob_options.min_blob_size = 8;
|
||||
|
||||
const std::string big1(1024, 'a');
|
||||
const WideColumns columns{{kDefaultWideColumnName, "d"}, {"big", big1}};
|
||||
|
||||
SstFileWriter writer(EnvOptions(), options);
|
||||
ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_path, embedded_blob_options));
|
||||
ASSERT_OK(writer.PutEntity("k", columns));
|
||||
ASSERT_OK(writer.Finish());
|
||||
ASSERT_OK(db_->IngestExternalFile({sst_path}, IngestExternalFileOptions()));
|
||||
|
||||
// Warm both the block cache (data block) and the blob cache (embedded record)
|
||||
// with a full read.
|
||||
{
|
||||
PinnableWideColumns result;
|
||||
ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(), "k",
|
||||
&result));
|
||||
ASSERT_EQ(result.columns(), columns);
|
||||
}
|
||||
|
||||
// The embedded record is now cached, so a block-cache-only read serves it.
|
||||
ReadOptions read_options;
|
||||
read_options.read_tier = kBlockCacheTier;
|
||||
PinnableWideColumns result;
|
||||
ASSERT_OK(
|
||||
db_->GetEntity(read_options, db_->DefaultColumnFamily(), "k", &result));
|
||||
ASSERT_EQ(result.columns(), columns);
|
||||
}
|
||||
|
||||
// A corrupt embedded blob record backing a wide-column column must surface an
|
||||
// error on GetEntity, not expose the raw same-file BlobIndex.
|
||||
// A failure resolving an embedded blob record backing a wide-column column
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
|
||||
#include "db/blob/blob_constants.h"
|
||||
#include "rocksdb/rocksdb_namespace.h"
|
||||
#include "rocksdb/status.h"
|
||||
|
||||
@@ -30,28 +31,28 @@ class SameFileBlobReader {
|
||||
public:
|
||||
virtual ~SameFileBlobReader() = default;
|
||||
|
||||
// Resolves a same-file blob reference (one for which blob_index.IsSameFile()
|
||||
// is true) into `value`, pinning the payload zero-copy (from the blob cache
|
||||
// or an owned buffer). read_tier == kBlockCacheTier without a cached record
|
||||
// yields Status::Incomplete; a bad record yields Status::Corruption.
|
||||
// Resolves a same-file ("embedded") blob reference (one for which
|
||||
// blob_index.IsSameFile() is true) into `value`, pinning the payload
|
||||
// zero-copy (from the blob cache or an owned buffer).
|
||||
//
|
||||
// range_length == kWholeBlobLength resolves the whole value. Any other length
|
||||
// resolves only the sub-range [range_offset, range_offset + range_length),
|
||||
// reading just those bytes on a cache miss -- skipping whole-record checksum
|
||||
// verification (a strict sub-range cannot cover it) and never populating the
|
||||
// blob cache; valid only for an uncompressed record and a verify_policy other
|
||||
// than kVerifyIfPresent, with range_offset + range_length <=
|
||||
// blob_index.size().
|
||||
//
|
||||
// verify_policy controls whole-record checksum verification (see
|
||||
// BlobVerifyPolicy); a whole read under kVerifyIfPresent verifies even when
|
||||
// ReadOptions::verify_checksums is off. read_tier == kBlockCacheTier without
|
||||
// a cached record yields Status::Incomplete; a bad record yields
|
||||
// Status::Corruption.
|
||||
virtual Status GetSameFileBlob(const ReadOptions& read_options,
|
||||
const BlobIndex& blob_index,
|
||||
uint64_t range_offset, size_t range_length,
|
||||
BlobVerifyPolicy verify_policy,
|
||||
PinnableSlice* value) const = 0;
|
||||
|
||||
// Resolves a byte sub-range [range_offset, range_offset + range_length) of an
|
||||
// *uncompressed* same-file blob reference into `value`, reading only those
|
||||
// bytes on a cache miss (the embedded counterpart of a separate-file blob
|
||||
// range read). Pins the payload zero-copy (a sliced cached record, or an
|
||||
// owned sub-range buffer). Unlike GetSameFileBlob it skips whole-record
|
||||
// checksum verification (a strict sub-range cannot cover it) and never
|
||||
// populates the blob cache. The caller must ensure range_offset +
|
||||
// range_length <= blob_index.size(). read_tier == kBlockCacheTier without a
|
||||
// cached record yields Status::Incomplete.
|
||||
virtual Status GetSameFileBlobRange(const ReadOptions& read_options,
|
||||
const BlobIndex& blob_index,
|
||||
uint64_t range_offset,
|
||||
size_t range_length,
|
||||
PinnableSlice* value) const = 0;
|
||||
};
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -1149,6 +1149,62 @@ TEST_F(DBLazyEntityTest, BlockCacheTierYieldsIncompleteOnMiss) {
|
||||
ASSERT_TRUE(s.IsIncomplete()) << s.ToString();
|
||||
}
|
||||
|
||||
// The embedded (same-file) read path probes the blob cache before honoring
|
||||
// read_tier == kBlockCacheTier, so a cached embedded record serves a
|
||||
// block-cache-only read -- whole column or sub-range -- instead of returning
|
||||
// Incomplete. Regression test for the former quirk where
|
||||
// ValidateEmbeddedBlobIndex rejected kBlockCacheTier up front, before the cache
|
||||
// probe (unlike the separate-file range path, which always probed first).
|
||||
TEST_F(DBLazyEntityTest, EmbeddedBlockCacheTierServedFromCachedValue) {
|
||||
Options options = GetLazyTestOptionsWithBlobCache();
|
||||
DestroyAndReopen(options);
|
||||
|
||||
const std::string key = "entity";
|
||||
const std::string big(4000, 'a');
|
||||
const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}};
|
||||
IngestEmbeddedEntity(options, key, columns);
|
||||
|
||||
// Warm the blob cache with a whole-column read of the embedded record.
|
||||
{
|
||||
LazyWideColumns warm;
|
||||
ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key,
|
||||
&warm));
|
||||
PinnableSlice whole;
|
||||
ASSERT_OK(warm.ResolveColumn(warm[1], &whole));
|
||||
ASSERT_EQ(whole, big);
|
||||
}
|
||||
|
||||
ASSERT_OK(options.statistics->Reset());
|
||||
|
||||
ReadOptions block_cache_only;
|
||||
block_cache_only.read_tier = kBlockCacheTier;
|
||||
|
||||
// Sub-range read on a fresh result (resolver cache empty) exercises the
|
||||
// embedded range path's blob-cache probe under kBlockCacheTier.
|
||||
{
|
||||
LazyWideColumns lazy;
|
||||
ASSERT_OK(db_->GetEntityLazy(block_cache_only, db_->DefaultColumnFamily(),
|
||||
key, &lazy));
|
||||
PinnableSlice range;
|
||||
ASSERT_OK(lazy.ResolveColumnRange(lazy[1], /*offset=*/1000, /*length=*/100,
|
||||
&range));
|
||||
ASSERT_EQ(range, big.substr(1000, 100));
|
||||
}
|
||||
|
||||
// Whole-column read on a fresh result, also served from the cache.
|
||||
{
|
||||
LazyWideColumns lazy;
|
||||
ASSERT_OK(db_->GetEntityLazy(block_cache_only, db_->DefaultColumnFamily(),
|
||||
key, &lazy));
|
||||
PinnableSlice whole;
|
||||
ASSERT_OK(lazy.ResolveColumn(lazy[1], &whole));
|
||||
ASSERT_EQ(whole, big);
|
||||
}
|
||||
|
||||
// Everything came from the blob cache: no disk blob bytes were read.
|
||||
ASSERT_EQ(BlobBytesRead(options), 0U);
|
||||
}
|
||||
|
||||
// A column that belongs to a different result is rejected with InvalidArgument
|
||||
// on the per-read status (columns identify their owning result, so there is no
|
||||
// untyped index to run out of range).
|
||||
|
||||
@@ -34,12 +34,46 @@ void ReadPathBlobResolver::Reset(
|
||||
|
||||
Status ReadPathBlobResolver::ResolveColumn(size_t column_index,
|
||||
Slice* resolved_value) {
|
||||
return ResolveColumnInternal(column_index, /*force_verify=*/false,
|
||||
resolved_value);
|
||||
// Whole-column resolution uses the captured ReadOptions' verify_checksums
|
||||
// (no force_verify).
|
||||
return ResolveColumnInternal(
|
||||
column_index,
|
||||
DeriveVerifyPolicy(blob_fetcher_.read_options().verify_checksums,
|
||||
/*force_verify=*/false),
|
||||
resolved_value);
|
||||
}
|
||||
|
||||
BlobVerifyPolicy ReadPathBlobResolver::DeriveVerifyPolicy(bool verify_checksums,
|
||||
bool force_verify) {
|
||||
if (force_verify) {
|
||||
return BlobVerifyPolicy::kVerifyIfPresent;
|
||||
}
|
||||
return verify_checksums ? BlobVerifyPolicy::kVerifyIfNoAmplification
|
||||
: BlobVerifyPolicy::kSkip;
|
||||
}
|
||||
|
||||
Status ReadPathBlobResolver::FetchBlobRef(const BlobIndex& blob_index,
|
||||
uint64_t range_offset,
|
||||
size_t range_length,
|
||||
BlobVerifyPolicy policy,
|
||||
PinnableSlice* out) {
|
||||
if (blob_index.IsSameFile()) {
|
||||
// Same-file ("embedded") references need the originating SST's reader.
|
||||
if (same_file_reader_ == nullptr) {
|
||||
return Status::Corruption(
|
||||
"Cannot resolve same-file blob reference: no same-file reader");
|
||||
}
|
||||
return same_file_reader_->GetSameFileBlob(blob_fetcher_.read_options(),
|
||||
blob_index, range_offset,
|
||||
range_length, policy, out);
|
||||
}
|
||||
return blob_fetcher_.FetchBlobRange(user_key_, blob_index, range_offset,
|
||||
range_length, policy, out,
|
||||
/*bytes_read=*/nullptr);
|
||||
}
|
||||
|
||||
Status ReadPathBlobResolver::ResolveColumnInternal(size_t column_index,
|
||||
bool force_verify,
|
||||
BlobVerifyPolicy policy,
|
||||
Slice* resolved_value) {
|
||||
assert(columns_);
|
||||
assert(resolved_value);
|
||||
@@ -72,53 +106,13 @@ Status ReadPathBlobResolver::ResolveColumnInternal(size_t column_index,
|
||||
std::make_unique<PinnableSlice>());
|
||||
auto& new_entry = resolved_cache_.back();
|
||||
|
||||
constexpr FilePrefetchBuffer* prefetch_buffer = nullptr;
|
||||
constexpr uint64_t* bytes_read = nullptr;
|
||||
|
||||
// TODO(lazy-blob-resolution-cleanup): the force_verify branching here
|
||||
// (and the whole-vs-range split in ResolveColumnRange) is really a
|
||||
// single 3-valued verify policy -- must-verify-if-present (today's
|
||||
// force_verify), verify-if-no-amplification (verify_checksums on: a
|
||||
// whole read verifies, a partial read skips), and ignore. Map
|
||||
// (verify_checksums, force_verify) to that internal enum once at this
|
||||
// boundary and thread it down, collapsing these branches. Deferred
|
||||
// for feature velocity; see the lazy blob resolution plan.
|
||||
if (force_verify && !blob_index.IsSameFile()) {
|
||||
// Honor force_verify for separate-file references: read and verify
|
||||
// the whole record even if ReadOptions::verify_checksums is off.
|
||||
status = blob_fetcher_.FetchBlobForceVerify(
|
||||
user_key_, blob_index, prefetch_buffer, new_entry.second.get(),
|
||||
bytes_read);
|
||||
} else if (force_verify && blob_index.IsSameFile()) {
|
||||
// Honor force_verify for same-file (embedded) references the same
|
||||
// way: force checksum verification on via a verify-enabled copy of
|
||||
// the read options. A same-file reference is only produced when the
|
||||
// entity was read from an SST that supplied its SameFileBlobReader,
|
||||
// so it is expected to be set; surface a clean error rather than
|
||||
// silently dropping force_verify (and taking the unverified path
|
||||
// below) if it somehow is not.
|
||||
assert(same_file_reader_ != nullptr);
|
||||
if (same_file_reader_ == nullptr) {
|
||||
status = Status::Corruption(
|
||||
"Cannot force-verify same-file blob: no same-file reader");
|
||||
} else {
|
||||
ReadOptions verify_read_options = blob_fetcher_.read_options();
|
||||
verify_read_options.verify_checksums = true;
|
||||
status = same_file_reader_->GetSameFileBlob(
|
||||
verify_read_options, blob_index, new_entry.second.get());
|
||||
}
|
||||
} else {
|
||||
// Route same-file ("embedded") references through the current SST's
|
||||
// SameFileBlobReader when one is set; separate-file references (and
|
||||
// the no-same-file-reader case) go straight to the Version-backed
|
||||
// fetcher. EffectiveFetcher() returns the plain base fetcher when
|
||||
// there is no same-file reader, adding no indirection.
|
||||
EmbeddedAwareBlobFetcher embedded_fetcher(&blob_fetcher_,
|
||||
same_file_reader_);
|
||||
status = embedded_fetcher.EffectiveFetcher()->FetchBlob(
|
||||
user_key_, blob_index, prefetch_buffer, new_entry.second.get(),
|
||||
bytes_read);
|
||||
}
|
||||
// Whole-column read of a blob reference, cached. FetchBlobRef routes
|
||||
// same-file vs separate-file references and applies the verify policy
|
||||
// (which folds in force_verify: kVerifyIfPresent verifies the whole
|
||||
// record even when ReadOptions::verify_checksums is off).
|
||||
status =
|
||||
FetchBlobRef(blob_index, /*range_offset=*/0, kWholeBlobLength,
|
||||
policy, new_entry.second.get());
|
||||
if (!status.ok()) {
|
||||
resolved_cache_.pop_back();
|
||||
} else {
|
||||
@@ -164,12 +158,17 @@ Status ReadPathBlobResolver::ResolveColumnRange(size_t column_index,
|
||||
return Status::InvalidArgument("Column index out of bounds");
|
||||
}
|
||||
|
||||
// Derive the verify policy once from (verify_checksums, force_verify) and let
|
||||
// it drive both the partial-vs-whole decision below and the downstream read.
|
||||
const BlobVerifyPolicy policy = DeriveVerifyPolicy(
|
||||
blob_fetcher_.read_options().verify_checksums, force_verify);
|
||||
|
||||
// No output buffer: the caller only wants to surface any I/O / integrity
|
||||
// error (and honor force_verify). Resolve the whole column and return; there
|
||||
// is nothing to slice into.
|
||||
if (result == nullptr) {
|
||||
Slice ignored;
|
||||
return ResolveColumnInternal(column_index, force_verify, &ignored);
|
||||
return ResolveColumnInternal(column_index, policy, &ignored);
|
||||
}
|
||||
|
||||
const BlobIndex* blob_index_ptr =
|
||||
@@ -179,10 +178,12 @@ Status ReadPathBlobResolver::ResolveColumnRange(size_t column_index,
|
||||
// reference that is: not already resolved (else we slice the cached whole
|
||||
// value), uncompressed (a strict sub-range of a compressed record can't be
|
||||
// decompressed in isolation), a strict sub-range (a whole-column read takes
|
||||
// the verifying + cache-filling path), and not force_verify. The read then
|
||||
// the verifying + cache-filling path), and a policy that permits skipping
|
||||
// verification (kVerifyIfPresent forces a whole verified read). The read then
|
||||
// goes to either the separate-file range fetcher (needs a Version) or, for a
|
||||
// same-file / embedded reference, the current SST's SameFileBlobReader.
|
||||
if (blob_index_ptr != nullptr && !force_verify &&
|
||||
if (blob_index_ptr != nullptr &&
|
||||
policy != BlobVerifyPolicy::kVerifyIfPresent &&
|
||||
blob_resolver_util::FindInCache(resolved_cache_, column_index) ==
|
||||
nullptr) {
|
||||
const BlobIndex& blob_index = *blob_index_ptr;
|
||||
@@ -204,24 +205,17 @@ Status ReadPathBlobResolver::ResolveColumnRange(size_t column_index,
|
||||
}
|
||||
const size_t avail = static_cast<size_t>(value_size - range_offset);
|
||||
const size_t actual_len = range_length > avail ? avail : range_length;
|
||||
if (blob_index.IsSameFile()) {
|
||||
return same_file_reader_->GetSameFileBlobRange(
|
||||
blob_fetcher_.read_options(), blob_index, range_offset,
|
||||
actual_len, result);
|
||||
}
|
||||
return blob_fetcher_.FetchBlobRange(user_key_, blob_index, range_offset,
|
||||
actual_len, result,
|
||||
/*bytes_read=*/nullptr);
|
||||
return FetchBlobRef(blob_index, range_offset, actual_len, policy,
|
||||
result);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Full path: resolve the whole column (inline value directly, blob reference
|
||||
// via the resolver's cache, filling it on a miss and verifying under
|
||||
// ReadOptions::verify_checksums or force_verify), then slice out the
|
||||
// requested range.
|
||||
// via the resolver's cache, filling it on a miss and verifying per policy),
|
||||
// then slice out the requested range.
|
||||
Slice whole;
|
||||
Status s = ResolveColumnInternal(column_index, force_verify, &whole);
|
||||
Status s = ResolveColumnInternal(column_index, policy, &whole);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include "db/blob/blob_constants.h"
|
||||
#include "db/blob/blob_fetcher.h"
|
||||
#include "db/blob/blob_index.h"
|
||||
#include "rocksdb/cleanable.h"
|
||||
@@ -124,11 +125,27 @@ class ReadPathBlobResolver {
|
||||
}
|
||||
|
||||
private:
|
||||
// Maps the public (ReadOptions::verify_checksums, force_verify) pair to the
|
||||
// internal 3-valued verify policy, once at this boundary (see
|
||||
// BlobVerifyPolicy). Every downstream blob read is driven by the resulting
|
||||
// policy rather than by re-deriving verification from two bools.
|
||||
static BlobVerifyPolicy DeriveVerifyPolicy(bool verify_checksums,
|
||||
bool force_verify);
|
||||
|
||||
// Reads a blob reference into *out, applying `policy`. range_length ==
|
||||
// kWholeBlobLength selects the whole value; any other length selects the
|
||||
// strict sub-range [range_offset, range_offset + range_length). Routes
|
||||
// same-file ("embedded") references to the current SST's SameFileBlobReader
|
||||
// and separate-file references to the Version-backed fetcher.
|
||||
Status FetchBlobRef(const BlobIndex& blob_index, uint64_t range_offset,
|
||||
size_t range_length, BlobVerifyPolicy policy,
|
||||
PinnableSlice* out);
|
||||
|
||||
// Shared implementation of ResolveColumn / the whole-column path of
|
||||
// ResolveColumnRange. When force_verify is set, a separate-file blob
|
||||
// reference is read with checksum verification forced on even if
|
||||
// ReadOptions::verify_checksums is off (see LazyColumnReadRequest).
|
||||
Status ResolveColumnInternal(size_t column_index, bool force_verify,
|
||||
// ResolveColumnRange: resolves and caches the whole column value under the
|
||||
// given verify policy (kVerifyIfPresent forces whole-record verification even
|
||||
// when ReadOptions::verify_checksums is off; see LazyColumnReadRequest).
|
||||
Status ResolveColumnInternal(size_t column_index, BlobVerifyPolicy policy,
|
||||
Slice* resolved_value);
|
||||
|
||||
// Owns its ReadOptions: a resolver's lifetime is independent of the caller
|
||||
|
||||
@@ -785,7 +785,10 @@ void StressTest::MaybeTestMultiGetEntityLazy(
|
||||
const size_t i = chosen[j].first;
|
||||
const size_t c = chosen[j].second;
|
||||
// Randomly fuzz a partial (byte-range) read vs a whole-column read, sized
|
||||
// against the reference when available, else the known logical size.
|
||||
// against the reference when available, else the known logical size; and
|
||||
// independently fuzz force_verify (which forces a whole, checksum-verified
|
||||
// read even for a sub-range) -- the resolved bytes are unchanged either
|
||||
// way.
|
||||
const WideColumns* reference = entity_ref[i];
|
||||
const std::optional<uint64_t> size_hint =
|
||||
reference ? std::optional<uint64_t>((*reference)[c].value().size())
|
||||
@@ -793,6 +796,7 @@ void StressTest::MaybeTestMultiGetEntityLazy(
|
||||
reads[j].column = &batch[i][c];
|
||||
PickLazyReadRange(&thread->rand, size_hint, &reads[j].offset,
|
||||
&reads[j].length);
|
||||
reads[j].force_verify = thread->rand.OneIn(4);
|
||||
reads[j].result = &results[j];
|
||||
reads[j].status = &statuses[j];
|
||||
}
|
||||
@@ -910,13 +914,17 @@ void StressTest::ResolveLazyEntity(ThreadState* thread, const std::string& key,
|
||||
for (size_t j = 0; j < chosen.size(); ++j) {
|
||||
const size_t c = chosen[j];
|
||||
// Randomly fuzz a partial (byte-range) read vs a whole-column read, sized
|
||||
// against the reference when available, else the known logical size.
|
||||
// against the reference when available, else the known logical size; and
|
||||
// independently fuzz force_verify (which forces a whole, checksum-verified
|
||||
// read even for a sub-range) -- the resolved bytes are unchanged either
|
||||
// way.
|
||||
const std::optional<uint64_t> size_hint =
|
||||
reference ? std::optional<uint64_t>((*reference)[c].value().size())
|
||||
: lazy[c].logical_size();
|
||||
reads[j].column = &lazy[c];
|
||||
PickLazyReadRange(&thread->rand, size_hint, &reads[j].offset,
|
||||
&reads[j].length);
|
||||
reads[j].force_verify = thread->rand.OneIn(4);
|
||||
reads[j].result = &results[j];
|
||||
reads[j].status = &statuses[j];
|
||||
}
|
||||
|
||||
@@ -1249,9 +1249,9 @@ bool BlockBasedTable::HasEmbeddedBlobRecords() const {
|
||||
return rep_->has_embedded_blobs;
|
||||
}
|
||||
|
||||
Status BlockBasedTable::ValidateEmbeddedBlobIndex(
|
||||
const ReadOptions& read_options, const BlobIndex& blob_index,
|
||||
size_t* payload_size, size_t* record_size) const {
|
||||
Status BlockBasedTable::ValidateEmbeddedBlobIndex(const BlobIndex& blob_index,
|
||||
size_t* payload_size,
|
||||
size_t* record_size) const {
|
||||
assert(payload_size != nullptr);
|
||||
assert(record_size != nullptr);
|
||||
if (!blob_index.IsSameFile()) {
|
||||
@@ -1265,10 +1265,6 @@ Status BlockBasedTable::ValidateEmbeddedBlobIndex(
|
||||
return Status::Corruption(
|
||||
"Same-file blob index found without embedded blob records");
|
||||
}
|
||||
if (read_options.read_tier == kBlockCacheTier) {
|
||||
return Status::Incomplete(
|
||||
"Cannot read embedded blob in block-cache-only mode");
|
||||
}
|
||||
|
||||
const uint64_t payload_size_u64 = blob_index.size();
|
||||
if (payload_size_u64 >
|
||||
@@ -1298,11 +1294,15 @@ Status BlockBasedTable::ResolveEmbeddedBlob(const ReadOptions& read_options,
|
||||
assert(value != nullptr);
|
||||
size_t payload_size = 0;
|
||||
size_t record_size = 0;
|
||||
Status s = ValidateEmbeddedBlobIndex(read_options, blob_index, &payload_size,
|
||||
&record_size);
|
||||
Status s = ValidateEmbeddedBlobIndex(blob_index, &payload_size, &record_size);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
// Direct file read (no blob cache): a block-cache-only read cannot be served.
|
||||
if (read_options.read_tier == kBlockCacheTier) {
|
||||
return Status::Incomplete(
|
||||
"Cannot read embedded blob in block-cache-only mode");
|
||||
}
|
||||
// Read the record (payload + trailer) directly into `value`, then drop the
|
||||
// trailer with a shrink (no extra copy on the common non-mmap path).
|
||||
value->resize(record_size);
|
||||
@@ -1324,11 +1324,15 @@ Status BlockBasedTable::ResolveEmbeddedBlobPinned(
|
||||
assert(value != nullptr);
|
||||
size_t payload_size = 0;
|
||||
size_t record_size = 0;
|
||||
Status s = ValidateEmbeddedBlobIndex(read_options, blob_index, &payload_size,
|
||||
&record_size);
|
||||
Status s = ValidateEmbeddedBlobIndex(blob_index, &payload_size, &record_size);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
// Direct file read (no blob cache): a block-cache-only read cannot be served.
|
||||
if (read_options.read_tier == kBlockCacheTier) {
|
||||
return Status::Incomplete(
|
||||
"Cannot read embedded blob in block-cache-only mode");
|
||||
}
|
||||
// Read into a heap buffer owned by `value` via a registered cleanup, so the
|
||||
// payload can be pinned (no copy) and outlive this reader. The trailer just
|
||||
// sits unused at the tail of the buffer.
|
||||
@@ -1353,8 +1357,7 @@ Status BlockBasedTable::ResolveEmbeddedBlobCached(
|
||||
assert(rep_->blob_source_ != nullptr);
|
||||
size_t payload_size = 0;
|
||||
size_t record_size = 0;
|
||||
Status s = ValidateEmbeddedBlobIndex(read_options, blob_index, &payload_size,
|
||||
&record_size);
|
||||
Status s = ValidateEmbeddedBlobIndex(blob_index, &payload_size, &record_size);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
@@ -1372,16 +1375,46 @@ Status BlockBasedTable::ResolveEmbeddedBlobCached(
|
||||
|
||||
Status BlockBasedTable::GetSameFileBlob(const ReadOptions& read_options,
|
||||
const BlobIndex& blob_index,
|
||||
uint64_t range_offset,
|
||||
size_t range_length,
|
||||
BlobVerifyPolicy verify_policy,
|
||||
PinnableSlice* value) const {
|
||||
// Mirror the whole-value same-file path in MaybeResolveEmbeddedValue: prefer
|
||||
// the blob-cache-backed read (BLOB_DB_* stats) when a BlobSource is wired,
|
||||
// otherwise a direct pinned read (SstFileReader/sst_dump/repair). Both pin
|
||||
// the payload zero-copy and run ValidateEmbeddedBlobIndex (which enforces
|
||||
// read_tier == kBlockCacheTier -> Incomplete and record bounds).
|
||||
if (rep_->blob_source_ != nullptr) {
|
||||
return ResolveEmbeddedBlobCached(read_options, blob_index, value);
|
||||
if (range_length != kWholeBlobLength) {
|
||||
// Byte-range embedded read: value bytes only, no whole-record checksum, no
|
||||
// cache fill (see ResolveEmbeddedBlobRangeCached). Only issued while
|
||||
// resolving a lazy result, which always runs against an open column family
|
||||
// (so a BlobSource is wired) and never forces verification of a strict
|
||||
// sub-range. Reaching here without a BlobSource means a same-file reference
|
||||
// was routed from an unexpected context (a corrupt index can produce the
|
||||
// file_number 0 same-file sentinel), so report corruption -- mirroring
|
||||
// Version::GetBlobRange's "Invalid blob file number".
|
||||
assert(verify_policy != BlobVerifyPolicy::kVerifyIfPresent);
|
||||
assert(rep_->blob_source_ != nullptr);
|
||||
if (rep_->blob_source_ == nullptr) {
|
||||
return Status::Corruption(
|
||||
"Invalid same-file blob reference (no BlobSource for range read)");
|
||||
}
|
||||
return ResolveEmbeddedBlobRangeCached(read_options, blob_index,
|
||||
range_offset, range_length, value);
|
||||
}
|
||||
return ResolveEmbeddedBlobPinned(read_options, blob_index, value);
|
||||
|
||||
// Whole-value read: prefer the blob-cache-backed read (BLOB_DB_* stats) when
|
||||
// a BlobSource is wired, otherwise a direct pinned read
|
||||
// (SstFileReader/sst_dump/repair). For kVerifyIfPresent, force whole-record
|
||||
// checksum verification even when ReadOptions::verify_checksums is off.
|
||||
if (verify_policy == BlobVerifyPolicy::kVerifyIfPresent &&
|
||||
!read_options.verify_checksums) {
|
||||
ReadOptions verify_read_options = read_options;
|
||||
verify_read_options.verify_checksums = true;
|
||||
return rep_->blob_source_ != nullptr
|
||||
? ResolveEmbeddedBlobCached(verify_read_options, blob_index,
|
||||
value)
|
||||
: ResolveEmbeddedBlobPinned(verify_read_options, blob_index,
|
||||
value);
|
||||
}
|
||||
return rep_->blob_source_ != nullptr
|
||||
? ResolveEmbeddedBlobCached(read_options, blob_index, value)
|
||||
: ResolveEmbeddedBlobPinned(read_options, blob_index, value);
|
||||
}
|
||||
|
||||
Status BlockBasedTable::ResolveEmbeddedBlobRangeCached(
|
||||
@@ -1391,17 +1424,11 @@ Status BlockBasedTable::ResolveEmbeddedBlobRangeCached(
|
||||
assert(rep_->blob_source_ != nullptr);
|
||||
size_t payload_size = 0;
|
||||
size_t record_size = 0;
|
||||
// TODO(lazy-blob-resolution-quirks): ValidateEmbeddedBlobIndex rejects
|
||||
// read_tier == kBlockCacheTier up front (before the blob-cache probe in
|
||||
// GetSimpleGen2BlobRange), so a block-cache-only read of an embedded column
|
||||
// returns Incomplete even when the whole payload is already cached. This
|
||||
// mirrors the whole-value embedded path (ResolveEmbeddedBlobCached) but
|
||||
// differs from the separate-file GetBlobRange, which probes the cache first
|
||||
// and can serve a kBlockCacheTier read from cache. Unify the two (probe cache
|
||||
// before rejecting kBlockCacheTier) in a later quirk-cleanup phase; see the
|
||||
// lazy blob resolution plan.
|
||||
Status s = ValidateEmbeddedBlobIndex(read_options, blob_index, &payload_size,
|
||||
&record_size);
|
||||
// A block-cache-only read is handled after the blob-cache probe inside
|
||||
// GetSimpleGen2BlobRange (a cached whole payload can still serve a sub-range;
|
||||
// only a genuine miss yields Incomplete), matching the separate-file
|
||||
// GetBlobRange path.
|
||||
Status s = ValidateEmbeddedBlobIndex(blob_index, &payload_size, &record_size);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
@@ -1413,29 +1440,6 @@ Status BlockBasedTable::ResolveEmbeddedBlobRangeCached(
|
||||
range_offset, range_length, value, /*bytes_read=*/nullptr);
|
||||
}
|
||||
|
||||
Status BlockBasedTable::GetSameFileBlobRange(const ReadOptions& read_options,
|
||||
const BlobIndex& blob_index,
|
||||
uint64_t range_offset,
|
||||
size_t range_length,
|
||||
PinnableSlice* value) const {
|
||||
// Byte-range embedded reads are only issued while resolving a lazy result,
|
||||
// which always runs against an open DB column family (so a BlobSource is
|
||||
// wired); unlike the whole-value GetSameFileBlob there is no direct-pinned
|
||||
// (no-BlobSource, e.g. SstFileReader) range path. Reaching here without a
|
||||
// BlobSource therefore means a same-file/embedded reference was routed from
|
||||
// an unexpected context -- which a corrupt index can cause in production
|
||||
// (e.g. an entry decoding to the file_number 0 same-file sentinel) -- so
|
||||
// report it as corruption, mirroring Version::GetBlobRange's "Invalid blob
|
||||
// file number".
|
||||
assert(rep_->blob_source_ != nullptr);
|
||||
if (rep_->blob_source_ == nullptr) {
|
||||
return Status::Corruption(
|
||||
"Invalid same-file blob reference (no BlobSource for range read)");
|
||||
}
|
||||
return ResolveEmbeddedBlobRangeCached(read_options, blob_index, range_offset,
|
||||
range_length, value);
|
||||
}
|
||||
|
||||
Status BlockBasedTable::MaybeResolveEmbeddedValue(
|
||||
const ReadOptions& read_options, const Slice& internal_key,
|
||||
const Slice& value, std::string* resolved_internal_key,
|
||||
|
||||
@@ -250,23 +250,18 @@ class BlockBasedTable : public TableReader, public SameFileBlobReader {
|
||||
size_t range_length,
|
||||
PinnableSlice* value) const;
|
||||
|
||||
// SameFileBlobReader: resolve a same-file blob reference for GetContext's
|
||||
// zero-copy wide-column resolution on the Get()/MultiGet() path. Routes
|
||||
// SameFileBlobReader: resolve a same-file blob reference (whole value or a
|
||||
// sub-range) for GetContext's zero-copy wide-column resolution on the
|
||||
// Get()/MultiGet() path and for the lazy read path. A whole-value read routes
|
||||
// through the blob cache (ResolveEmbeddedBlobCached) when a BlobSource is
|
||||
// wired, else a direct pinned read (ResolveEmbeddedBlobPinned).
|
||||
// wired, else a direct pinned read (ResolveEmbeddedBlobPinned); a sub-range
|
||||
// read routes through ResolveEmbeddedBlobRangeCached (requires a BlobSource).
|
||||
// See BlobVerifyPolicy / kWholeBlobLength for the range + verify semantics.
|
||||
Status GetSameFileBlob(const ReadOptions& read_options,
|
||||
const BlobIndex& blob_index,
|
||||
const BlobIndex& blob_index, uint64_t range_offset,
|
||||
size_t range_length, BlobVerifyPolicy verify_policy,
|
||||
PinnableSlice* value) const override;
|
||||
|
||||
// SameFileBlobReader: byte-range counterpart of GetSameFileBlob, used by the
|
||||
// lazy read path for partial (sub-range) reads of an uncompressed embedded
|
||||
// blob. Routes through ResolveEmbeddedBlobRangeCached (requires a
|
||||
// BlobSource).
|
||||
Status GetSameFileBlobRange(const ReadOptions& read_options,
|
||||
const BlobIndex& blob_index,
|
||||
uint64_t range_offset, size_t range_length,
|
||||
PinnableSlice* value) const override;
|
||||
|
||||
// If `value` is a same-file BlobIndex, materializes the referenced payload
|
||||
// and updates `resolved_internal_key` to the corresponding value type. Leaves
|
||||
// `resolved`. When `pinned_value` is non-null, a whole-value same-file blob
|
||||
@@ -319,8 +314,7 @@ class BlockBasedTable : public TableReader, public SameFileBlobReader {
|
||||
|
||||
// Validates a same-file `blob_index` and returns the payload and full record
|
||||
// (payload + trailer) sizes. Shared by the ResolveEmbeddedBlob* variants.
|
||||
Status ValidateEmbeddedBlobIndex(const ReadOptions& read_options,
|
||||
const BlobIndex& blob_index,
|
||||
Status ValidateEmbeddedBlobIndex(const BlobIndex& blob_index,
|
||||
size_t* payload_size,
|
||||
size_t* record_size) const;
|
||||
|
||||
|
||||
@@ -57,6 +57,7 @@
|
||||
#include "rocksdb/filter_policy.h"
|
||||
#include "rocksdb/io_dispatcher.h"
|
||||
#include "rocksdb/io_status.h"
|
||||
#include "rocksdb/lazy_wide_columns.h"
|
||||
#include "rocksdb/memtablerep.h"
|
||||
#include "rocksdb/options.h"
|
||||
#include "rocksdb/perf_context.h"
|
||||
@@ -265,6 +266,10 @@ DEFINE_string(
|
||||
"\tfillembeddedblob -- Create and ingest an SST of whole-value embedded "
|
||||
"(same-file) blobs; read with readrandom. Requires format_version>=7\n"
|
||||
"\treadrandomentity -- read N times in random order via GetEntity\n"
|
||||
"\treadrandomentitylazy -- read N times in random order via GetEntityLazy, "
|
||||
"resolving lazy_entity_read_length bytes of each column "
|
||||
"(byte-range/partial "
|
||||
"blob reads); requires open_files=-1\n"
|
||||
"\tmultireadentity -- read N in random batches via MultiGetEntity\n"
|
||||
"\topenandcompact -- Open DB and compact all files to bottommost level, "
|
||||
"writing output to separate directory without modifying source DB. "
|
||||
@@ -1289,6 +1294,14 @@ DEFINE_int32(num_short_wide_columns, 1,
|
||||
"default column is one of these inline columns; if 0, the default "
|
||||
"column is an embedded blob.");
|
||||
|
||||
DEFINE_int64(
|
||||
lazy_entity_read_length, -1,
|
||||
"For the readrandomentitylazy benchmark: number of bytes to resolve from "
|
||||
"the start of each column via LazyWideColumns::MultiResolve. -1 resolves "
|
||||
"the whole column; a smaller value exercises byte-range (partial) blob "
|
||||
"reads, which read only the requested bytes of an uncompressed blob column "
|
||||
"(see rocksdb.blobdb.lazy.* statistics). Requires -open_files=-1.");
|
||||
|
||||
// Secondary DB instance Options
|
||||
DEFINE_bool(use_secondary_db, false,
|
||||
"Open a RocksDB secondary instance. A primary instance can be "
|
||||
@@ -3444,6 +3457,7 @@ class Benchmark {
|
||||
bool read_operands_; // read via GetMergeOperands()
|
||||
bool read_entity_; // read via GetEntity() (readrandomentity)
|
||||
bool multiread_entity_; // read via MultiGetEntity() (multireadentity)
|
||||
bool read_entity_lazy_; // read via GetEntityLazy() (readrandomentitylazy)
|
||||
std::vector<std::string> keys_;
|
||||
|
||||
class ErrorHandlerListener : public EventListener {
|
||||
@@ -3962,7 +3976,8 @@ class Benchmark {
|
||||
use_blob_db_(FLAGS_use_blob_db), // Stacked BlobDB
|
||||
read_operands_(false),
|
||||
read_entity_(false),
|
||||
multiread_entity_(false) {
|
||||
multiread_entity_(false),
|
||||
read_entity_lazy_(false) {
|
||||
// use simcache instead of cache
|
||||
if (FLAGS_simcache_size >= 0) {
|
||||
if (FLAGS_cache_numshardbits >= 1) {
|
||||
@@ -4248,6 +4263,7 @@ class Benchmark {
|
||||
read_operands_ = false;
|
||||
read_entity_ = false;
|
||||
multiread_entity_ = false;
|
||||
read_entity_lazy_ = false;
|
||||
|
||||
int num_repeat = 1;
|
||||
int num_warmup = 0;
|
||||
@@ -4381,6 +4397,9 @@ class Benchmark {
|
||||
} else if (name == "readrandomentity") {
|
||||
method = &Benchmark::ReadRandom;
|
||||
read_entity_ = true;
|
||||
} else if (name == "readrandomentitylazy") {
|
||||
method = &Benchmark::ReadRandom;
|
||||
read_entity_lazy_ = true;
|
||||
} else if (name == "multireadentity") {
|
||||
fprintf(stderr, "entries_per_batch = %" PRIi64 "\n",
|
||||
entries_per_batch_);
|
||||
@@ -7461,6 +7480,29 @@ class Benchmark {
|
||||
fprintf(stderr, "readrandomentity does not support user timestamps\n");
|
||||
db_bench_exit(1);
|
||||
}
|
||||
if (read_entity_lazy_) {
|
||||
if (user_timestamp_size_ > 0) {
|
||||
fprintf(stderr,
|
||||
"readrandomentitylazy does not support user timestamps\n");
|
||||
db_bench_exit(1);
|
||||
}
|
||||
if (open_options_.max_open_files != -1) {
|
||||
// The lazy API pins table readers via the immortal-table-cache mode.
|
||||
fprintf(stderr, "readrandomentitylazy requires max_open_files == -1\n");
|
||||
db_bench_exit(1);
|
||||
}
|
||||
}
|
||||
// Reusable buffers for the lazy (readrandomentitylazy) path; empty and
|
||||
// unused otherwise. lazy_read_length == kLazyWholeColumn resolves whole
|
||||
// columns; a smaller value drives byte-range (partial) blob reads.
|
||||
LazyWideColumns lazy_columns;
|
||||
std::vector<PinnableSlice> lazy_results;
|
||||
std::vector<Status> lazy_statuses;
|
||||
std::vector<LazyColumnReadRequest> lazy_reads;
|
||||
const size_t lazy_read_length =
|
||||
FLAGS_lazy_entity_read_length < 0
|
||||
? kLazyWholeColumn
|
||||
: static_cast<size_t>(FLAGS_lazy_entity_read_length);
|
||||
std::unique_ptr<char[]> ts_guard;
|
||||
Slice ts;
|
||||
if (user_timestamp_size_ > 0) {
|
||||
@@ -7499,6 +7541,9 @@ class Benchmark {
|
||||
Status s;
|
||||
pinnable_val.Reset();
|
||||
pinnable_columns.Reset();
|
||||
// Release the previous iteration's resolved slices before GetEntityLazy
|
||||
// resets lazy_columns (whose resolver backs those slices).
|
||||
lazy_results.clear();
|
||||
for (size_t i = 0; i < pinnable_vals.size(); ++i) {
|
||||
pinnable_vals[i].Reset();
|
||||
}
|
||||
@@ -7530,6 +7575,8 @@ class Benchmark {
|
||||
}
|
||||
} else if (read_entity_) {
|
||||
s = db_with_cfh->db->GetEntity(options, cfh, key, &pinnable_columns);
|
||||
} else if (read_entity_lazy_) {
|
||||
s = db_with_cfh->db->GetEntityLazy(options, cfh, key, &lazy_columns);
|
||||
} else {
|
||||
s = db_with_cfh->db->Get(options, cfh, key, &pinnable_val, ts_ptr);
|
||||
}
|
||||
@@ -7546,6 +7593,37 @@ class Benchmark {
|
||||
bytes += column.name().size() + column.value().size();
|
||||
}
|
||||
}
|
||||
if (read_entity_lazy_) {
|
||||
// Resolve lazy_read_length bytes of each column in one MultiResolve
|
||||
// call. For an uncompressed blob column a strict sub-range reads only
|
||||
// the requested bytes from storage (see rocksdb.blobdb.lazy.*);
|
||||
// inline columns and whole-column reads resolve as usual.
|
||||
const size_t n = lazy_columns.size();
|
||||
lazy_results.resize(n);
|
||||
lazy_statuses.assign(n, Status::OK());
|
||||
lazy_reads.resize(n);
|
||||
for (size_t c = 0; c < n; ++c) {
|
||||
lazy_reads[c].column = &lazy_columns[c];
|
||||
lazy_reads[c].offset = 0;
|
||||
lazy_reads[c].length = lazy_read_length;
|
||||
lazy_reads[c].result = &lazy_results[c];
|
||||
lazy_reads[c].status = &lazy_statuses[c];
|
||||
}
|
||||
const Status rs = lazy_columns.MultiResolve(lazy_reads);
|
||||
if (rs.ok()) {
|
||||
for (size_t c = 0; c < n; ++c) {
|
||||
if (lazy_statuses[c].ok()) {
|
||||
bytes += lazy_columns[c].name().size() + lazy_results[c].size();
|
||||
} else if (!lazy_statuses[c].IsNotFound()) {
|
||||
HandleBenchmarkIOError(lazy_statuses[c],
|
||||
"MultiResolve column read returned "
|
||||
"an error");
|
||||
}
|
||||
}
|
||||
} else if (!rs.IsNotFound()) {
|
||||
HandleBenchmarkIOError(rs, "MultiResolve returned an error");
|
||||
}
|
||||
}
|
||||
} else if (!s.IsNotFound()) {
|
||||
// Tolerated IO error: skip this read and continue.
|
||||
HandleBenchmarkIOError(s, "Get returned an error");
|
||||
|
||||
Reference in New Issue
Block a user