From 8e1569b351ae8ffbded7796027cce4095adeb100 Mon Sep 17 00:00:00 2001 From: Peter Dillinger Date: Fri, 7 Aug 2026 12:08:03 -0700 Subject: [PATCH] Lazy wide-column byte-range (partial) blob reads (#15067) Summary: Builds on the lazy wide-column read API (DB::GetEntityLazy / DB::MultiGetEntityLazy) to serve true byte-range reads: resolving a sub-range of an uncompressed blob column now reads only the requested bytes from storage instead of fetching and slicing the whole column. This covers blobs in separate blob files and embedded (same-file) blobs in the SST, so a caller that knows a large value's layout can pull just the parts it needs. A partial read skips whole-record checksum verification (a strict sub-range cannot cover the checksum) and does not populate the blob cache (the cache entry holds the whole record). A compressed column, a whole-column read, a cache hit, or a read with LazyColumnReadRequest::force_verify still resolves (and verifies) the whole column; force_verify forces verification even when ReadOptions::verify_checksums is off. Mechanics: new range primitives mirror the whole-value read paths -- BlobFileReader::GetBlobRange, BlobSource::GetBlobRange, Version::GetBlobRange, and VersionBlobFetcherBase::FetchBlobRange for separate-file blobs; ReadSimpleGen2BlobRange, BlobSource::GetSimpleGen2BlobRange, and SameFileBlobReader::GetSameFileBlobRange (implemented by BlockBasedTable) for embedded blobs. ReadPathBlobResolver::ResolveColumnRange owns the partial-vs-whole decision and routes same-file references through the current SST's reader; the whole-column path also gained VersionBlobFetcherBase::FetchBlobForceVerify to honor force_verify. Observability: deferred resolve reads are attributed to a new Env::IOActivity::kLazyResolve (distinct from the kGetEntity / kMultiGetEntity of the initial entity read), and lazy resolution's storage reads are counted by new statistics rocksdb.blobdb.lazy.read.count / .read.bytes (all lazy reads) and rocksdb.blobdb.lazy.partial.read.count / .partial.bytes.saved (the byte-range subset and the bytes it avoided). The whole-column read sites gate recording on kLazyResolve so non-lazy reads are excluded. Also fixes a latent bug in LazyWideColumnsBatch::MultiResolve where an empty/default-constructed batch (null rep_) would accept a foreign standalone column (null owning_batch_rep_) through a nullptr == nullptr comparison instead of returning InvalidArgument. Follow-ups (in a local plan): unifying the blob-read API surface -- folding the whole/range method twins and the verify-policy flags (verify_checksums x force_verify) into one internal BlobReadRequest primitive (a range descriptor plus a 3-valued verify policy), which retires FetchBlobForceVerify and the whole/range method pairs; this is deferred for feature velocity and is not public-API affecting (the public force_verify bool stays). Marked in code with TODO(lazy-blob-resolution-cleanup) in db/blob/blob_fetcher.h and db/wide/read_path_blob_resolver.cc. Also cross-key coalescing (Phase 3) and async execution (Phase 4). Pull Request resolved: https://github.com/facebook/rocksdb/pull/15067 Test Plan: Extended db/wide/db_lazy_entity_test.cc: separate-file and embedded partial reads return the exact sub-span and read only the requested bytes; a partial read does not fill the blob cache and a cache hit slices with no I/O; a compressed column resolves whole and slices; force_verify does a full verified read and is detected via a tampered record even with verify_checksums off; the lazy statistics distinguish whole vs partial reads and are not touched by non-lazy reads; deferred reads carry kLazyResolve (verified with a FileSystem wrapper); and a regression test for the empty-batch ownership check. Added an IOActivityToString case in env/env_test.cc and registered the new tickers in the statistics name map. Crash test: extended the db_stress lazy-vs-eager differential (db_stress_tool/db_stress_test_base.cc) to fuzz per-column byte ranges. Each resolved column now randomly picks a whole-column read or a partial (offset, length) sub-range -- occasionally with an offset past the end or a length beyond the remainder to exercise clamping -- for both GetEntityLazy and MultiGetEntityLazy; when an eager reference is available the resolved bytes are compared against the same clamped sub-range of it. No new flags (reuses --lazy_entity_read_one_in); force_verify fuzzing remains a follow-up. Reviewed By: xingbowang Differential Revision: D115130388 Pulled By: pdillinger fbshipit-source-id: 454368dea8fae069d9de649c2c51611469d3c809 --- CLAUDE.md | 5 + db/blob/blob_fetcher.cc | 39 + db/blob/blob_fetcher.h | 38 + db/blob/blob_file_reader.cc | 71 ++ db/blob/blob_file_reader.h | 17 + db/blob/blob_gen2_format.cc | 48 ++ db/blob/blob_gen2_format.h | 18 + db/blob/blob_source.cc | 331 ++++++-- db/blob/blob_source.h | 50 ++ db/blob/same_file_blob_reader.h | 18 + db/db_impl/db_impl_files.cc | 2 + db/version_set.cc | 34 + db/version_set.h | 13 + db/wide/db_lazy_entity_test.cc | 728 ++++++++++++++++++ db/wide/lazy_wide_columns.cc | 127 +-- db/wide/read_path_blob_resolver.cc | 159 +++- db/wide/read_path_blob_resolver.h | 30 + db_stress_tool/db_stress_test_base.cc | 96 ++- env/env.cc | 2 + env/env_test.cc | 3 + include/rocksdb/env.h | 6 + include/rocksdb/lazy_wide_columns.h | 37 +- include/rocksdb/statistics.h | 20 + include/rocksdb/thread_status.h | 1 + monitoring/statistics.cc | 5 + monitoring/thread_status_util_debug.cc | 2 + table/block_based/block_based_table_reader.cc | 52 ++ table/block_based/block_based_table_reader.h | 22 + .../new_features/lazy_wide_columns.md | 2 +- util/thread_operation.h | 1 + 30 files changed, 1828 insertions(+), 149 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index a82bcf7102..5a991ea711 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -393,6 +393,11 @@ rather than relying on libstdc++ transitive includes. ### Adding release note * Release note should be kept short at high level for external user consumption. + Release notes identify what users might care about most in a release. They + are not exhaustive and are not a guide. PLEASE learn from past agents who + ried to build elaborate release notes with implementation details and + elsewhere-documented nuance. That wastes time. Fight the bias that + "my change" is important so must be worthy of release note mention. * If more than single markdown line, consider how their formatting will be integrated into HISTORY.md. diff --git a/db/blob/blob_fetcher.cc b/db/blob/blob_fetcher.cc index 47c5960058..912a7688cd 100644 --- a/db/blob/blob_fetcher.cc +++ b/db/blob/blob_fetcher.cc @@ -44,6 +44,45 @@ 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. + 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); +} + Status EmbeddedAwareBlobFetcher::FetchBlob(const Slice& user_key, const BlobIndex& blob_index, FilePrefetchBuffer* prefetch_buffer, diff --git a/db/blob/blob_fetcher.h b/db/blob/blob_fetcher.h index a49457a124..68699b64e3 100644 --- a/db/blob/blob_fetcher.h +++ b/db/blob/blob_fetcher.h @@ -63,6 +63,37 @@ 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. + Status 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; + + // 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 @@ -73,6 +104,13 @@ class VersionBlobFetcherBase : public BlobFetcher { (allow_write_path_fallback_ && blob_file_cache_ != nullptr); } + // True if this fetcher can serve an I/O-saving byte-range read: it reads + // through a Version (FetchBlobRange goes to Version::GetBlobRange) and is not + // in direct-write fallback mode (whose reads resolve whole records instead). + bool SupportsRangeRead() const { + return version_ != nullptr && !allow_write_path_fallback_; + } + protected: VersionBlobFetcherBase(const Version* version, BlobFileCache* blob_file_cache, bool allow_write_path_fallback) diff --git a/db/blob/blob_file_reader.cc b/db/blob/blob_file_reader.cc index d8d39a2190..2e052c13ab 100644 --- a/db/blob/blob_file_reader.cc +++ b/db/blob/blob_file_reader.cc @@ -416,6 +416,77 @@ Status BlobFileReader::GetBlob( return Status::OK(); } +Status BlobFileReader::GetBlobRange(const ReadOptions& read_options, + const Slice& user_key, uint64_t offset, + uint64_t value_size, uint64_t range_offset, + size_t range_length, + MemoryAllocator* allocator, + std::unique_ptr* result, + uint64_t* bytes_read) const { + assert(result); + + // Partial reads are only meaningful for uncompressed blobs: a strict + // sub-range of a compressed record cannot be decompressed in isolation. + // Callers (BlobSource::GetBlobRange) must ensure this; enforce it here too. + if (compression_type_ != kNoCompression) { + return Status::Corruption("Cannot range-read a compressed blob"); + } + + const uint64_t key_size = user_key.size(); + + // Validate that the full value region is within the file (same check GetBlob + // performs); this guards the sub-range read below. + if (!IsValidBlobOffset(offset, key_size, value_size, file_size_, + has_footer_)) { + return Status::Corruption("Invalid blob offset"); + } + + // The requested sub-range must lie within the value. + if (range_offset > value_size || range_length > value_size - range_offset) { + return Status::InvalidArgument("Blob range out of bounds"); + } + + // Read only the requested bytes, at the value's file position plus the range + // offset. Unlike GetBlob there is no record-header adjustment (we do not read + // the key/header) and no whole-record checksum verification (a strict + // sub-range cannot cover it -- callers that require verification take the + // full-read path instead). + const uint64_t read_offset = offset + range_offset; + const size_t read_size = range_length; + + Slice range_slice; + Buffer buf; + AlignedBuffer direct_io_buffer; + + TEST_SYNC_POINT("BlobFileReader::GetBlobRange:ReadFromFile"); + PERF_COUNTER_ADD(blob_read_count, 1); + PERF_COUNTER_ADD(blob_read_byte, read_size); + PERF_TIMER_GUARD(blob_read_time); + + { + const Status s = + ReadFromFile(file_reader_.get(), read_options, read_offset, read_size, + statistics_, &range_slice, &buf, &direct_io_buffer); + if (!s.ok()) { + return s; + } + } + + TEST_SYNC_POINT_CALLBACK("BlobFileReader::GetBlobRange:TamperWithResult", + &range_slice); + + // The blob is uncompressed, so the bytes read are the requested value bytes; + // copy them into an owned BlobContents (no decompression). + BlobContentsCreator::Create(result, /*out_charge=*/nullptr, range_slice, + kNoCompression, allocator); + + if (bytes_read) { + *bytes_read = read_size; + } + + return Status::OK(); +} + void BlobFileReader::MultiGetBlob( const ReadOptions& read_options, MemoryAllocator* allocator, autovector>>& diff --git a/db/blob/blob_file_reader.h b/db/blob/blob_file_reader.h index 12b18e86e3..58ed1a89e4 100644 --- a/db/blob/blob_file_reader.h +++ b/db/blob/blob_file_reader.h @@ -64,6 +64,23 @@ class BlobFileReader { std::unique_ptr* result, uint64_t* bytes_read) const; + // Reads a byte sub-range [range_offset, range_offset + range_length) of an + // *uncompressed* blob's value directly from the file, reading only those + // bytes. Unlike GetBlob it does not read the record header/key, does not + // verify the whole-record checksum (a strict sub-range cannot cover it), and + // never decompresses -- so it is only valid when this reader's blob file is + // uncompressed (checked). `offset` and `value_size` are the blob value's file + // offset and full size (from the BlobIndex); the caller must ensure + // range_offset + range_length <= value_size. On success `*result` owns a copy + // of the requested bytes and `*bytes_read` (when non-null) is the number of + // bytes read from the file. + Status GetBlobRange(const ReadOptions& read_options, const Slice& user_key, + uint64_t offset, uint64_t value_size, + uint64_t range_offset, size_t range_length, + MemoryAllocator* allocator, + std::unique_ptr* result, + uint64_t* bytes_read) const; + // offsets must be sorted in ascending order by caller. void MultiGetBlob( const ReadOptions& read_options, MemoryAllocator* allocator, diff --git a/db/blob/blob_gen2_format.cc b/db/blob/blob_gen2_format.cc index 6f307a2b1e..61b5eb7eca 100644 --- a/db/blob/blob_gen2_format.cc +++ b/db/blob/blob_gen2_format.cc @@ -76,6 +76,54 @@ Status ReadAndVerifySimpleGen2BlobRecord( return Status::OK(); } +Status ReadSimpleGen2BlobRange(const ReadOptions& read_options, + RandomAccessFileReader* file, + uint64_t record_offset, size_t payload_size, + uint64_t range_offset, size_t range_length, + CompressionType expected_compression, + char* buf) { + assert(file != nullptr); + assert(buf != nullptr); + + // A strict sub-range of a compressed payload cannot be decompressed in + // isolation; callers resolve such payloads whole and slice instead. + if (expected_compression != kNoCompression) { + return Status::Corruption("Cannot range-read a compressed blob"); + } + + // The requested sub-range must lie within the payload (the trailer that + // follows the payload is never part of a range read). + if (range_offset > payload_size || + range_length > payload_size - range_offset) { + return Status::InvalidArgument("Blob range out of bounds"); + } + + Slice result; + IOOptions opts; + IODebugContext dbg; + Status s = file->PrepareIOOptions(read_options, opts, &dbg); + if (s.ok()) { + s = file->Read(opts, record_offset + range_offset, range_length, &result, + buf, nullptr, &dbg); + } + if (!s.ok()) { + return s; + } + if (result.size() != range_length) { + return Status::Corruption("Could not read complete blob range"); + } + // With mmap reads the data lands outside `buf`; copy it in so the caller can + // rely on `buf` owning the bytes (the only copy on the mmap path). + if (result.data() != buf) { + memcpy(buf, result.data(), range_length); + } + + // No trailer read and no checksum verification: a strict sub-range cannot + // cover the record's checksum (callers that require it read the whole + // record). + return Status::OK(); +} + IOStatus WriteSimpleGen2BlobRecord(WritableFileWriter* file, const WriteOptions& write_options, ChecksumType checksum_type, diff --git a/db/blob/blob_gen2_format.h b/db/blob/blob_gen2_format.h index 2eaed5ebfe..e811efb098 100644 --- a/db/blob/blob_gen2_format.h +++ b/db/blob/blob_gen2_format.h @@ -73,6 +73,24 @@ Status ReadAndVerifySimpleGen2BlobRecord( ChecksumType checksum_type, uint32_t base_context_checksum, CompressionType expected_compression, char* buf); +// Reads a byte sub-range [range_offset, range_offset + range_length) of an +// *uncompressed* SimpleGen2Blob payload directly into `buf` (capacity >= +// range_length), reading only those bytes. Unlike +// ReadAndVerifySimpleGen2BlobRecord it does not read the trailer and does not +// verify the record's checksum (a strict sub-range cannot cover it); callers +// that require verification read the whole record instead. `payload_size` is +// the full payload size (from the blob index); the caller must ensure +// range_offset + range_length <= payload_size. `expected_compression` must be +// kNoCompression (a strict sub-range of a compressed payload cannot be +// decompressed in isolation). +// +// On success, buf[0, range_length) holds the requested payload bytes. +Status ReadSimpleGen2BlobRange(const ReadOptions& read_options, + RandomAccessFileReader* file, + uint64_t record_offset, size_t payload_size, + uint64_t range_offset, size_t range_length, + CompressionType expected_compression, char* buf); + // Writes a SimpleGen2Blob record for `payload` at the current end of `file`, // which the caller asserts is byte offset `record_offset`. Appends the payload // bytes followed by the 5-byte trailer (compression marker + context-modified diff --git a/db/blob/blob_source.cc b/db/blob/blob_source.cc index 4758829f9b..37fc74aac1 100644 --- a/db/blob/blob_source.cc +++ b/db/blob/blob_source.cc @@ -36,6 +36,102 @@ Status AppendBlobRefreshRetryFailure(const Status& stale_status, stale_status, "; refresh retry failed: ", retry_status.ToString()); } +// Runs read(reader) against the blob-file reader for `file_number`. On a +// Corruption -- which a stale cached reader (e.g. the physical file was +// replaced) can surface -- evicts the cached reader, reopens it uncached, +// retries read() once, and refreshes the cache on success. Shared by the +// single-blob read paths (GetBlob / GetBlobRange) so the stale-file recovery +// lives in one place. +// +// `read` is invoked with a BlobFileReader* and returns a Status; because it may +// run twice, it must re-initialize any output it populates on each call (and is +// taken by const reference rather than forwarded, since it is not moved-from). +template +Status ReadBlobWithReaderRetry(BlobFileCache* blob_file_cache, + const ReadOptions& read_options, + uint64_t file_number, const ReadFn& read) { + CacheHandleGuard blob_file_reader; + Status s = blob_file_cache->GetBlobFileReader(read_options, file_number, + &blob_file_reader); + if (!s.ok()) { + return s; + } + assert(blob_file_reader.GetValue()); + + s = read(blob_file_reader.GetValue()); + if (!s.IsCorruption()) { + return s; + } + + const Status stale_status = s; + blob_file_reader.Reset(); + blob_file_cache->Evict(file_number); + + std::unique_ptr fresh_reader; + s = blob_file_cache->OpenBlobFileReaderUncached( + read_options, file_number, &fresh_reader, + /*allow_footer_skip_retry=*/false); + if (!s.ok()) { + return AppendBlobRefreshRetryFailure(stale_status, s); + } + + s = read(fresh_reader.get()); + if (!s.ok()) { + return AppendBlobRefreshRetryFailure(stale_status, s); + } + + CacheHandleGuard ignored_reader; + blob_file_cache + ->RefreshBlobFileReader(file_number, &fresh_reader, &ignored_reader) + .PermitUncheckedError(); + return s; +} + +// Records lazy wide-column read metrics for one actual storage read issued +// while resolving a lazy result (attributed via Env::IOActivity::kLazyResolve): +// `bytes_read` bytes were read from storage. Covers both whole-column and +// partial reads; not called for cache hits (no storage read). +void RecordLazyRead(Statistics* statistics, uint64_t bytes_read) { + RecordTick(statistics, BLOB_DB_LAZY_READ_COUNT); + RecordTick(statistics, BLOB_DB_LAZY_READ_BYTES, bytes_read); +} + +// Additionally records the partial-read metrics for one actual partial +// (byte-range) read: `bytes_read` bytes were fetched instead of the column's +// full `value_size`. Call alongside RecordLazyRead (a partial read is also a +// lazy read). +void RecordLazyPartialRead(Statistics* statistics, uint64_t bytes_read, + uint64_t value_size) { + RecordTick(statistics, BLOB_DB_LAZY_PARTIAL_READ_COUNT); + if (value_size > bytes_read) { + RecordTick(statistics, BLOB_DB_LAZY_PARTIAL_BYTES_SAVED, + value_size - bytes_read); + } +} + +// Pins the sub-range [range_offset, range_offset + range_length) of a cache-hit +// whole blob value into *value (clamped: an offset at/past the end yields +// empty, length is clamped to the remainder), keeping the cached bytes alive by +// transferring the cache handle into *value. Shared by the range-read cache-hit +// paths (GetBlobRange / GetSimpleGen2BlobRange). +void PinCacheHitSubRange(CacheHandleGuard* blob_handle, + uint64_t range_offset, size_t range_length, + PinnableSlice* value) { + const Slice full = blob_handle->GetValue()->data(); + Slice sub; + if (range_offset < full.size()) { + const size_t off = static_cast(range_offset); + const size_t avail = full.size() - off; + const size_t len = range_length > avail ? avail : range_length; + sub = Slice(full.data() + off, len); + } // else: offset at/past end -> empty (not an error) + + value->Reset(); + constexpr Cleanable* cleanable = nullptr; + value->PinSlice(sub, cleanable); + blob_handle->TransferTo(value); +} + } // namespace BlobSource::BlobSource(const ImmutableOptions& immutable_options, @@ -228,66 +324,37 @@ Status BlobSource::GetBlob(const ReadOptions& read_options, std::unique_ptr blob_contents; { - CacheHandleGuard blob_file_reader; - s = blob_file_cache_->GetBlobFileReader(read_options, file_number, - &blob_file_reader); - if (!s.ok()) { - return s; - } - - assert(blob_file_reader.GetValue()); - - if (compression_type != blob_file_reader.GetValue()->GetCompressionType()) { - return Status::Corruption("Compression type mismatch when reading blob"); - } - MemoryAllocator* const allocator = (blob_cache_ && read_options.fill_cache) ? blob_cache_.get()->memory_allocator() : nullptr; uint64_t read_size = 0; - s = blob_file_reader.GetValue()->GetBlob( - read_options, user_key, offset, value_size, compression_type, - prefetch_buffer, allocator, &blob_contents, &read_size); - if (s.IsCorruption()) { - const Status stale_status = s; - blob_file_reader.Reset(); - blob_file_cache_->Evict(file_number); - - std::unique_ptr fresh_reader; - s = blob_file_cache_->OpenBlobFileReaderUncached( - read_options, file_number, &fresh_reader, - /*allow_footer_skip_retry=*/false); - if (!s.ok()) { - return AppendBlobRefreshRetryFailure(stale_status, s); - } - - if (compression_type != fresh_reader->GetCompressionType()) { - return Status::Corruption( - "Compression type mismatch when reading blob"); - } - - blob_contents.reset(); - read_size = 0; - s = fresh_reader->GetBlob(read_options, user_key, offset, value_size, - compression_type, prefetch_buffer, allocator, - &blob_contents, &read_size); - if (!s.ok()) { - s = AppendBlobRefreshRetryFailure(stale_status, s); - } else { - CacheHandleGuard ignored_reader; - blob_file_cache_ - ->RefreshBlobFileReader(file_number, &fresh_reader, &ignored_reader) - .PermitUncheckedError(); - } - } + s = ReadBlobWithReaderRetry( + blob_file_cache_, read_options, file_number, + [&](BlobFileReader* reader) { + if (compression_type != reader->GetCompressionType()) { + return Status::Corruption( + "Compression type mismatch when reading blob"); + } + blob_contents.reset(); + read_size = 0; + return reader->GetBlob(read_options, user_key, offset, value_size, + compression_type, prefetch_buffer, allocator, + &blob_contents, &read_size); + }); if (!s.ok()) { return s; } if (bytes_read) { *bytes_read = read_size; } + // Whole-column read on the lazy resolve path (partial reads go through + // GetBlobRange). Counts the storage read; partial reads are not counted + // here. + if (read_options.io_activity == Env::IOActivity::kLazyResolve) { + RecordLazyRead(statistics_, read_size); + } } if (blob_cache_ && read_options.fill_cache) { @@ -308,6 +375,92 @@ Status BlobSource::GetBlob(const ReadOptions& read_options, return s; } +Status BlobSource::GetBlobRange(const ReadOptions& read_options, + const Slice& user_key, uint64_t file_number, + uint64_t offset, uint64_t file_size, + uint64_t value_size, + CompressionType compression_type, + uint64_t range_offset, size_t range_length, + PinnableSlice* value, uint64_t* bytes_read) { + assert(value); + // Partial reads are for uncompressed blobs only; the caller decides this (a + // compressed column takes the whole-value GetBlob path and slices). + assert(compression_type == kNoCompression); + // Range reads are (currently) only issued while resolving a lazy result, so + // the lazy read stats below are recorded unconditionally (unlike GetBlob, + // which gates on the activity). Enforce that invariant here. + assert(read_options.io_activity == Env::IOActivity::kLazyResolve); + + Status s; + + const CacheKey cache_key = GetCacheKey(file_number, file_size, offset); + + CacheHandleGuard blob_handle; + + // First, probe the blob cache for the whole value. On a hit, slice the + // requested sub-range out of the cached value while pinning the cache handle + // (zero-copy, no disk read). A partial read never inserts into the cache. + if (blob_cache_) { + Slice key = cache_key.AsSlice(); + s = GetBlobFromCache(key, &blob_handle); + if (s.ok()) { + PinCacheHitSubRange(&blob_handle, range_offset, range_length, value); + if (bytes_read) { + *bytes_read = 0; // served from cache; no disk read + } + return s; + } + } + + assert(blob_handle.IsEmpty()); + + const bool no_io = read_options.read_tier == kBlockCacheTier; + if (no_io) { + return Status::Incomplete("Cannot read blob(s): no disk I/O allowed"); + } + + // Cache miss: read only the requested sub-range from the file. The result is + // never inserted into the blob cache (see the header comment). + std::unique_ptr blob_contents; + { + // No cache-fill allocator: a partial value is never inserted into the + // cache. + constexpr MemoryAllocator* allocator = nullptr; + + uint64_t read_size = 0; + // Reuses the stale-reader retry shared with GetBlob. A range read skips + // whole-record checksum verification, so the retry handles the stale-file + // case (offset/size mismatch surfacing as Corruption), not a payload + // checksum failure. + s = ReadBlobWithReaderRetry( + blob_file_cache_, read_options, file_number, + [&](BlobFileReader* reader) { + if (compression_type != reader->GetCompressionType()) { + return Status::Corruption( + "Compression type mismatch when reading blob"); + } + blob_contents.reset(); + read_size = 0; + return reader->GetBlobRange(read_options, user_key, offset, + value_size, range_offset, range_length, + allocator, &blob_contents, &read_size); + }); + if (!s.ok()) { + return s; + } + if (bytes_read) { + *bytes_read = read_size; + } + RecordLazyRead(statistics_, read_size); + RecordLazyPartialRead(statistics_, read_size, value_size); + } + + PinOwnedBlob(&blob_contents, value); + + assert(s.ok()); + return s; +} + Status BlobSource::GetSimpleGen2Blob( const ReadOptions& read_options, const OffsetableCacheKey& base_cache_key, RandomAccessFileReader* file, uint64_t record_offset, uint64_t payload_size, @@ -379,6 +532,11 @@ Status BlobSource::GetSimpleGen2Blob( if (bytes_read) { *bytes_read = record_size; } + // Whole-column read on the lazy resolve path (partial reads go through + // GetSimpleGen2BlobRange). + if (read_options.io_activity == Env::IOActivity::kLazyResolve) { + RecordLazyRead(statistics_, record_size); + } if (blob_cache_ && read_options.fill_cache) { // If filling cache is allowed and a cache is configured, try to put the @@ -398,6 +556,83 @@ Status BlobSource::GetSimpleGen2Blob( return s; } +Status BlobSource::GetSimpleGen2BlobRange( + const ReadOptions& read_options, const OffsetableCacheKey& base_cache_key, + RandomAccessFileReader* file, uint64_t record_offset, uint64_t payload_size, + ChecksumType /*checksum_type*/, uint32_t /*base_context_checksum*/, + CompressionType expected_compression, uint64_t range_offset, + size_t range_length, PinnableSlice* value, uint64_t* bytes_read) { + assert(value); + assert(file); + // Partial reads are for uncompressed payloads only; the caller decides this + // (a compressed column takes the whole-payload GetSimpleGen2Blob path + + // slice). + assert(expected_compression == kNoCompression); + // Range reads are (currently) only issued while resolving a lazy result, so + // the lazy read stats below are recorded unconditionally (unlike + // GetSimpleGen2Blob, which gates on the activity). Enforce that invariant + // here. + assert(read_options.io_activity == Env::IOActivity::kLazyResolve); + + const CacheKey cache_key = + GetSimpleGen2BlobCacheKey(base_cache_key, record_offset); + + Status s; + + CacheHandleGuard blob_handle; + + // First, probe the blob cache for the whole payload. On a hit, slice the + // requested sub-range out of the cached payload while pinning the cache + // handle (zero-copy, no disk read). A partial read never inserts into the + // cache. + if (blob_cache_) { + Slice key = cache_key.AsSlice(); + s = GetBlobFromCache(key, &blob_handle); + if (s.ok()) { + PinCacheHitSubRange(&blob_handle, range_offset, range_length, value); + if (bytes_read) { + *bytes_read = 0; // served from cache; no disk read + } + return s; + } + } + + assert(blob_handle.IsEmpty()); + + const bool no_io = read_options.read_tier == kBlockCacheTier; + if (no_io) { + return Status::Incomplete("Cannot read blob(s): no disk I/O allowed"); + } + + // Cache miss: read only the requested sub-range from the file. The result is + // never inserted into the blob cache (a partial payload cannot represent the + // whole-record cache entry). + CacheAllocationPtr buf = AllocateBlock(range_length, /*allocator=*/nullptr); + s = ReadSimpleGen2BlobRange(read_options, file, record_offset, + static_cast(payload_size), range_offset, + range_length, expected_compression, buf.get()); + if (!s.ok()) { + return s; + } + + std::unique_ptr blob_contents( + new BlobContents(std::move(buf), range_length)); + + RecordTick(statistics_, BLOB_DB_BLOB_FILE_BYTES_READ, range_length); + PERF_COUNTER_ADD(blob_read_count, 1); + PERF_COUNTER_ADD(blob_read_byte, range_length); + RecordLazyRead(statistics_, range_length); + RecordLazyPartialRead(statistics_, range_length, payload_size); + if (bytes_read) { + *bytes_read = range_length; + } + + PinOwnedBlob(&blob_contents, value); + + assert(s.ok()); + return s; +} + void BlobSource::MultiGetBlob(const ReadOptions& read_options, autovector& blob_reqs, uint64_t* bytes_read) { diff --git a/db/blob/blob_source.h b/db/blob/blob_source.h index c81b1de7a7..8168126b1c 100644 --- a/db/blob/blob_source.h +++ b/db/blob/blob_source.h @@ -61,6 +61,31 @@ class BlobSource { FilePrefetchBuffer* prefetch_buffer, PinnableSlice* value, uint64_t* bytes_read); + // Reads a byte sub-range [range_offset, range_offset + range_length) of an + // *uncompressed* blob's value, reading only those bytes on a cache miss. + // + // First probes the blob cache for the whole value. On a hit, pins the cache + // handle into *value and points it at the requested sub-range (zero-copy, no + // disk I/O). On a miss, reads only the requested bytes from the blob file + // (via BlobFileReader::GetBlobRange -- no record-header/key read and no + // whole-record checksum verification) and pins the owned buffer into *value. + // + // Unlike GetBlob, a partial read never inserts into the blob cache: the cache + // entry is keyed per (file, offset) and holds the *whole* BlobContents, so a + // partial value would violate that invariant. `compression_type` must be + // kNoCompression (a strict sub-range of a compressed blob cannot be + // decompressed in isolation); callers that need a compressed column, a + // whole-column read, or checksum verification use GetBlob and slice instead. + // + // The caller must ensure range_offset + range_length <= value_size. On a miss + // *bytes_read (when non-null) is the number of bytes read from the file; on a + // hit it is 0. + Status GetBlobRange(const ReadOptions& read_options, const Slice& user_key, + uint64_t file_number, uint64_t offset, uint64_t file_size, + uint64_t value_size, CompressionType compression_type, + uint64_t range_offset, size_t range_length, + PinnableSlice* value, uint64_t* bytes_read); + // Reads a SimpleGen2Blob payload (see db/blob/blob_gen2_format.h) through the // blob value cache and BLOB_DB_* statistics. This is the counterpart to // GetBlob() for the second-generation blob record format, which is read @@ -96,6 +121,31 @@ class BlobSource { CompressionType expected_compression, PinnableSlice* value, uint64_t* bytes_read); + // Reads a byte sub-range [range_offset, range_offset + range_length) of an + // *uncompressed* SimpleGen2Blob payload, reading only those bytes on a cache + // miss. This is the embedded (same-file) counterpart of GetBlobRange(). + // + // First probes the blob cache for the whole payload (keyed exactly as + // GetSimpleGen2Blob). On a hit, pins the cache handle into *value and points + // it at the requested sub-range (zero-copy, no disk I/O). On a miss, reads + // only the requested bytes from the file (via ReadSimpleGen2BlobRange -- no + // trailer read and no checksum verification) and pins the owned buffer. + // + // Like GetBlobRange, a partial read never inserts into the blob cache (the + // cache entry holds the whole payload). `expected_compression` must be + // kNoCompression. The caller must ensure range_offset + range_length <= + // payload_size. On a miss *bytes_read (when non-null) is the number of bytes + // read from the file; on a hit it is 0. + Status GetSimpleGen2BlobRange(const ReadOptions& read_options, + const OffsetableCacheKey& base_cache_key, + RandomAccessFileReader* file, + uint64_t record_offset, uint64_t payload_size, + ChecksumType checksum_type, + uint32_t base_context_checksum, + CompressionType expected_compression, + uint64_t range_offset, size_t range_length, + PinnableSlice* value, uint64_t* bytes_read); + // Read multiple blobs from the underlying cache or blob file(s). // // If successful, returns ok and sets "result" in the elements of "blob_reqs" diff --git a/db/blob/same_file_blob_reader.h b/db/blob/same_file_blob_reader.h index db3c41cbd8..2e565b7914 100644 --- a/db/blob/same_file_blob_reader.h +++ b/db/blob/same_file_blob_reader.h @@ -5,6 +5,9 @@ #pragma once +#include +#include + #include "rocksdb/rocksdb_namespace.h" #include "rocksdb/status.h" @@ -34,6 +37,21 @@ class SameFileBlobReader { virtual Status GetSameFileBlob(const ReadOptions& read_options, const BlobIndex& blob_index, 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 diff --git a/db/db_impl/db_impl_files.cc b/db/db_impl/db_impl_files.cc index 4884b0cd7d..4e9ea52315 100644 --- a/db/db_impl/db_impl_files.cc +++ b/db/db_impl/db_impl_files.cc @@ -55,6 +55,8 @@ Env::IOActivity GetCurrentThreadIOActivityForMetadataRead() { case ThreadStatus::OperationType:: OP_GET_FILE_CHECKSUMS_FROM_CURRENT_MANIFEST: return Env::IOActivity::kGetFileChecksumsFromCurrentManifest; + case ThreadStatus::OperationType::OP_LAZY_RESOLVE: + return Env::IOActivity::kLazyResolve; default: return Env::IOActivity::kUnknown; } diff --git a/db/version_set.cc b/db/version_set.cc index 13be3d7d24..ca4ee9203e 100644 --- a/db/version_set.cc +++ b/db/version_set.cc @@ -2803,6 +2803,40 @@ Status Version::GetBlob(const ReadOptions& read_options, const Slice& user_key, return s; } +Status Version::GetBlobRange(const ReadOptions& read_options, + const Slice& user_key, const BlobIndex& blob_index, + uint64_t range_offset, size_t range_length, + PinnableSlice* value, uint64_t* bytes_read) const { + assert(value); + + if (blob_index.HasTTL() || blob_index.IsInlined()) { + return Status::Corruption("Unexpected TTL/inlined blob index"); + } + + // A strict sub-range of a compressed blob cannot be decompressed in + // isolation; the caller resolves such columns whole and slices instead. + if (blob_index.compression() != kNoCompression) { + return Status::Corruption("Cannot range-read a compressed blob"); + } + + const uint64_t blob_file_number = blob_index.file_number(); + + auto blob_file_meta = storage_info_.GetBlobFileMetaData(blob_file_number); + if (!blob_file_meta) { + // INTEGRITY CHECK -- see Version::GetBlob. No metadata (including + // file_number 0, the same-file "embedded" sentinel) means a corrupt index + // or a same-file reference that the lazy caller must not route here. + return Status::Corruption("Invalid blob file number"); + } + + assert(blob_source_); + value->Reset(); + return blob_source_->GetBlobRange( + read_options, user_key, blob_file_number, blob_index.offset(), + blob_file_meta->GetBlobFileSize(), blob_index.size(), + blob_index.compression(), range_offset, range_length, value, bytes_read); +} + void Version::MultiGetBlob( const ReadOptions& read_options, MultiGetRange& range, std::unordered_map& blob_ctxs) { diff --git a/db/version_set.h b/db/version_set.h index d4c17133a4..b7ae799594 100644 --- a/db/version_set.h +++ b/db/version_set.h @@ -984,6 +984,19 @@ class Version { FilePrefetchBuffer* prefetch_buffer, PinnableSlice* value, uint64_t* bytes_read) const; + // Retrieves a byte sub-range [range_offset, range_offset + range_length) of + // an *uncompressed*, separate-file blob's value into *value, reading only + // those bytes on a cache miss (see BlobSource::GetBlobRange). The blob file + // must be part of this Version. Returns Corruption for a + // TTL/inlined/same-file or compressed blob index (the lazy caller only routes + // uncompressed separate-file references here; other cases take the + // whole-value GetBlob path). The caller must ensure range_offset + + // range_length <= blob_index.size(). + Status GetBlobRange(const ReadOptions& read_options, const Slice& user_key, + const BlobIndex& blob_index, uint64_t range_offset, + size_t range_length, PinnableSlice* value, + uint64_t* bytes_read) const; + struct BlobReadContext { BlobReadContext(const BlobIndex& blob_idx, KeyContext* key_ctx) : blob_index(blob_idx), key_context(key_ctx) {} diff --git a/db/wide/db_lazy_entity_test.cc b/db/wide/db_lazy_entity_test.cc index 34cb488df3..fae88db0b2 100644 --- a/db/wide/db_lazy_entity_test.cc +++ b/db/wide/db_lazy_entity_test.cc @@ -17,15 +17,88 @@ #include "db/db_test_util.h" #include "db/wide/wide_column_test_util.h" #include "port/stack_trace.h" +#include "rocksdb/cache.h" #include "rocksdb/comparator.h" +#include "rocksdb/env.h" +#include "rocksdb/file_system.h" #include "rocksdb/lazy_wide_columns.h" +#include "rocksdb/sst_file_writer.h" #include "rocksdb/statistics.h" #include "rocksdb/utilities/transaction_db.h" #include "test_util/testutil.h" #include "util/coding.h" +#include "util/compression.h" +#include "util/defer.h" namespace ROCKSDB_NAMESPACE { +namespace { +// FileSystem wrapper that records the ReadOptions::io_activity of the most +// recent read to a blob file (*.blob), so tests can assert how deferred lazy +// resolve reads are attributed. Separate-file blob payloads live in .blob +// files, so this isolates a deferred ResolveColumn read from the SST reads done +// by GetEntityLazy itself. +class BlobReadIOActivityFS : public FileSystemWrapper { + public: + explicit BlobReadIOActivityFS(std::shared_ptr base) + : FileSystemWrapper(base) {} + + static const char* kClassName() { return "BlobReadIOActivityFS"; } + const char* Name() const override { return kClassName(); } + + IOStatus NewRandomAccessFile(const std::string& fname, + const FileOptions& file_opts, + std::unique_ptr* result, + IODebugContext* dbg) override { + std::unique_ptr file; + const IOStatus s = + target()->NewRandomAccessFile(fname, file_opts, &file, dbg); + if (!s.ok()) { + return s; + } + const bool is_blob_file = + fname.size() >= 5 && fname.compare(fname.size() - 5, 5, ".blob") == 0; + if (is_blob_file) { + *result = std::make_unique(std::move(file), this); + } else { + *result = std::move(file); + } + return s; + } + + void ResetLastBlobReadIOActivity() { + last_blob_read_io_activity_.store( + static_cast(Env::IOActivity::kUnknown)); + } + Env::IOActivity last_blob_read_io_activity() const { + return static_cast(last_blob_read_io_activity_.load()); + } + + private: + class RecordingFile : public FSRandomAccessFileOwnerWrapper { + public: + RecordingFile(std::unique_ptr&& file, + BlobReadIOActivityFS* fs) + : FSRandomAccessFileOwnerWrapper(std::move(file)), fs_(fs) {} + + IOStatus Read(uint64_t offset, size_t n, const IOOptions& options, + Slice* result, char* scratch, + IODebugContext* dbg) const override { + fs_->last_blob_read_io_activity_.store( + static_cast(options.io_activity)); + return FSRandomAccessFileOwnerWrapper::Read(offset, n, options, result, + scratch, dbg); + } + + private: + BlobReadIOActivityFS* fs_; + }; + + std::atomic last_blob_read_io_activity_{ + static_cast(Env::IOActivity::kUnknown)}; +}; +} // namespace + class DBLazyEntityTest : public DBTestBase { protected: DBLazyEntityTest() @@ -43,9 +116,63 @@ class DBLazyEntityTest : public DBTestBase { return options; } + // Like GetLazyTestOptions but with a blob cache configured, so tests can + // observe blob-cache hits / (non-)insertions on the partial-read path. + Options GetLazyTestOptionsWithBlobCache() { + Options options = GetLazyTestOptions(); + LRUCacheOptions co; + co.capacity = 8 << 20; // 8MB + options.blob_cache = NewLRUCache(co); + return options; + } + uint64_t BlobBytesRead(const Options& options) { return options.statistics->getTickerCount(BLOB_DB_BLOB_FILE_BYTES_READ); } + + uint64_t BlobCacheAdds(const Options& options) { + return options.statistics->getTickerCount(BLOB_DB_CACHE_ADD); + } + + uint64_t LazyReadCount(const Options& options) { + return options.statistics->getTickerCount(BLOB_DB_LAZY_READ_COUNT); + } + uint64_t LazyReadBytes(const Options& options) { + return options.statistics->getTickerCount(BLOB_DB_LAZY_READ_BYTES); + } + uint64_t LazyPartialReadCount(const Options& options) { + return options.statistics->getTickerCount(BLOB_DB_LAZY_PARTIAL_READ_COUNT); + } + uint64_t LazyPartialBytesSaved(const Options& options) { + return options.statistics->getTickerCount(BLOB_DB_LAZY_PARTIAL_BYTES_SAVED); + } + + // Warms the blob-file reader for a column with a 1-byte partial read: opening + // the blob file reads its header + footer, a one-time cost. Tests that assert + // an exact BLOB_DB_BLOB_FILE_BYTES_READ delta call this and then reset stats, + // so only the bytes of the measured read are counted. + void WarmBlobFileReader(LazyWideColumns& lazy, size_t column_index) { + PinnableSlice warm; + ASSERT_OK(lazy.ResolveColumnRange(lazy[column_index], /*offset=*/0, + /*length=*/1, &warm)); + } + + // Writes an SST containing a single embedded (same-file blob) wide-column + // entity and ingests it into the default column family. Columns at least + // `min_blob_size` bytes become same-file blob references; smaller ones stay + // inline. Used by the Phase 2 (embedded range read) tests. + void IngestEmbeddedEntity(const Options& options, const std::string& key, + const WideColumns& columns, + uint64_t min_blob_size = 64) { + const std::string sst_path = dbname_ + "/embedded_" + key + ".sst"; + SstFileWriterEmbeddedBlobOptions embedded_blob_options; + embedded_blob_options.min_blob_size = min_blob_size; + SstFileWriter writer(EnvOptions(), options); + ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_path, embedded_blob_options)); + ASSERT_OK(writer.PutEntity(key, columns)); + ASSERT_OK(writer.Finish()); + ASSERT_OK(db_->IngestExternalFile({sst_path}, IngestExternalFileOptions())); + } }; // Enumerate a lazily-read entity without triggering any blob I/O: inline @@ -130,6 +257,570 @@ TEST_F(DBLazyEntityTest, ResolveSingleColumnRange) { ASSERT_TRUE(past_end.empty()); } +// Phase 1: a byte-range read of an uncompressed separate-file blob reads only +// the requested bytes from storage (not the whole column, and not the record +// header/key), and returns exactly the requested sub-span. +TEST_F(DBLazyEntityTest, PartialReadReadsOnlyRequestedBytes) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + std::string big(4000, '\0'); + for (size_t i = 0; i < big.size(); ++i) { + big[i] = static_cast('0' + (i % 10)); + } + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + + // Warm the blob-file reader (opening it reads the file header + footer, a + // one-time cost that would otherwise be counted below). Partial reads are not + // cached, so the measured read still hits the file. + WarmBlobFileReader(lazy, /*column_index=*/1); + ASSERT_OK(options.statistics->Reset()); + + constexpr uint64_t kOffset = 1000; + constexpr size_t kLength = 100; + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], kOffset, kLength, &range)); + ASSERT_EQ(range, big.substr(kOffset, kLength)); + + // Only the requested bytes were read from the blob file -- not the whole + // 4000-byte column, and not the record header/key. + ASSERT_EQ(BlobBytesRead(options), kLength); +} + +// Phase 1: length == kLazyWholeColumn with a nonzero offset is a partial read +// from the offset to the end -- only the remaining bytes are read. +TEST_F(DBLazyEntityTest, PartialReadToEndFromOffset) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + std::string big(4000, '\0'); + for (size_t i = 0; i < big.size(); ++i) { + big[i] = static_cast('0' + (i % 10)); + } + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + + // Warm the blob-file reader (see PartialReadReadsOnlyRequestedBytes). + WarmBlobFileReader(lazy, /*column_index=*/1); + ASSERT_OK(options.statistics->Reset()); + + constexpr uint64_t kOffset = 3000; + PinnableSlice range; + ASSERT_OK( + lazy.ResolveColumnRange(lazy[1], kOffset, kLazyWholeColumn, &range)); + ASSERT_EQ(range, big.substr(kOffset)); + ASSERT_EQ(BlobBytesRead(options), big.size() - kOffset); +} + +// Phase 1: a partial read never inserts into the blob cache (which is keyed by +// (file, offset) and holds the whole record). A subsequent whole-column read +// does populate the cache, confirming the cache is otherwise functional. +TEST_F(DBLazyEntityTest, PartialReadDoesNotFillBlobCache) { + Options options = GetLazyTestOptionsWithBlobCache(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(options.statistics->Reset()); + + // A partial read reads only the range and adds nothing to the cache. + { + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], /*offset=*/1000, + /*length=*/100, &range)); + ASSERT_EQ(range, big.substr(1000, 100)); + } + ASSERT_EQ(BlobCacheAdds(options), 0U); + + // A whole-column read of the same column does populate the cache. + { + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + PinnableSlice whole; + ASSERT_OK(lazy.ResolveColumn(lazy[1], &whole)); + ASSERT_EQ(whole, big); + } + ASSERT_EQ(BlobCacheAdds(options), 1U); +} + +// Phase 1: when the whole value is already in the blob cache, a partial read is +// served by slicing the cached value -- no blob-file I/O. +TEST_F(DBLazyEntityTest, PartialReadServedFromCachedWholeValueDoesNoIO) { + Options options = GetLazyTestOptionsWithBlobCache(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + + // Warm the blob cache with a whole-column read. + { + 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()); + + // A fresh lazy read + partial read: the whole value is cached, so the range + // is sliced out of the cache with no blob-file I/O. + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], /*offset=*/1000, /*length=*/100, + &range)); + ASSERT_EQ(range, big.substr(1000, 100)); + ASSERT_EQ(BlobBytesRead(options), 0U); +} + +// Phase 1: a byte-range read of a *compressed* blob cannot read a sub-range in +// isolation, so it resolves (and decompresses) the whole column and slices the +// range. Its logical size is unknown before resolution. +TEST_F(DBLazyEntityTest, CompressedColumnRangeResolvesWholeAndSlices) { + if (!Snappy_Supported()) { + ROCKSDB_GTEST_SKIP("Snappy compression not supported"); + return; + } + Options options = GetLazyTestOptions(); + options.blob_compression_type = kSnappyCompression; + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); // highly compressible + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + + // A compressed reference reports an unknown logical size before resolution. + ASSERT_TRUE(lazy[1].is_reference()); + ASSERT_FALSE(lazy[1].logical_size().has_value()); + + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], /*offset=*/1000, /*length=*/100, + &range)); + ASSERT_EQ(range, big.substr(1000, 100)); +} + +// Phase 1: force_verify on a partial read prioritizes checksum verification +// over I/O efficiency -- it reads (and checks) the whole record even though +// only a sub-range was requested. +TEST_F(DBLazyEntityTest, ForceVerifyPartialReadDoesFullVerifiedRead) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(options.statistics->Reset()); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + + PinnableSlice range; + Status status; + LazyColumnReadRequest read; + read.column = &lazy[1]; + read.offset = 1000; + read.length = 100; + read.force_verify = true; + read.result = ⦥ + read.status = &status; + ASSERT_OK(lazy.MultiResolve(/*num_reads=*/1, &read)); + ASSERT_OK(status); + ASSERT_EQ(range, big.substr(1000, 100)); + + // A verified read covers the whole record (value + key + header), so strictly + // more than the requested 100 bytes are read from the blob file. + ASSERT_GT(BlobBytesRead(options), big.size()); +} + +// Phase 1: force_verify verifies the record's checksum even when +// ReadOptions::verify_checksums is off. With verification forced on, a +// corrupted blob record surfaces as Corruption; without force_verify the same +// partial read skips verification and succeeds. +TEST_F(DBLazyEntityTest, ForceVerifyVerifiesEvenWhenVerifyChecksumsOff) { + Options options = GetLazyTestOptions(); // no blob cache: reads hit the file + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + + // Corrupt the value bytes after they are read from the file, so a checksum + // verification (if it runs) fails. Flip a byte in place to keep the record + // size intact (a strict sub-range read of the requested bytes would not + // trigger this on the whole-record path, which is the point). + SyncPoint::GetInstance()->SetCallBack( + "BlobFileReader::GetBlob:TamperWithResult", [](void* arg) { + Slice* const record_slice = static_cast(arg); + ASSERT_NE(record_slice, nullptr); + ASSERT_FALSE(record_slice->empty()); + char* const data = const_cast(record_slice->data()); + data[record_slice->size() - 1] ^= 0xff; + }); + SyncPoint::GetInstance()->EnableProcessing(); + + ReadOptions ro; + ro.verify_checksums = false; // globally off + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ro, db_->DefaultColumnFamily(), key, &lazy)); + + // With force_verify, the whole record is read and its checksum verified, so + // the corruption is detected. + PinnableSlice range; + Status status; + LazyColumnReadRequest read; + read.column = &lazy[1]; + read.offset = 1000; + read.length = 100; + read.force_verify = true; + read.result = ⦥ + read.status = &status; + ASSERT_OK(lazy.MultiResolve(/*num_reads=*/1, &read)); + ASSERT_TRUE(status.IsCorruption()) << status.ToString(); + + SyncPoint::GetInstance()->DisableProcessing(); + SyncPoint::GetInstance()->ClearAllCallBacks(); +} + +// Phase 2: a byte-range read of an uncompressed embedded (same-file) blob reads +// only the requested bytes from the SST -- the embedded counterpart of +// PartialReadReadsOnlyRequestedBytes. +TEST_F(DBLazyEntityTest, EmbeddedPartialReadReadsOnlyRequestedBytes) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + const std::string key = "entity"; + std::string big(4000, '\0'); + for (size_t i = 0; i < big.size(); ++i) { + big[i] = static_cast('0' + (i % 10)); + } + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + IngestEmbeddedEntity(options, key, columns); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + + // The blob column is an unresolved same-file reference with a known + // (uncompressed) logical size. + ASSERT_EQ(lazy.size(), 2U); + ASSERT_TRUE(lazy[1].is_reference()); + ASSERT_TRUE(lazy[1].logical_size().has_value()); + ASSERT_EQ(*lazy[1].logical_size(), big.size()); + + ASSERT_OK(options.statistics->Reset()); + + constexpr uint64_t kOffset = 1000; + constexpr size_t kLength = 100; + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], kOffset, kLength, &range)); + ASSERT_EQ(range, big.substr(kOffset, kLength)); + + // Only the requested bytes were read from the SST's embedded blob region -- + // not the whole 4000-byte payload, and not the record trailer. + ASSERT_EQ(BlobBytesRead(options), kLength); +} + +// Phase 2: with the whole embedded payload already in the blob cache, a partial +// read slices the cached value with no disk I/O. +TEST_F(DBLazyEntityTest, + EmbeddedPartialReadServedFromCachedWholeValueDoesNoIO) { + 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. + { + 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()); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], /*offset=*/1000, /*length=*/100, + &range)); + ASSERT_EQ(range, big.substr(1000, 100)); + ASSERT_EQ(BlobBytesRead(options), 0U); +} + +// Phase 2: force_verify on an embedded partial read forces the whole-record +// (verifying) path even when ReadOptions::verify_checksums is off -- so more +// than the requested bytes are read from the SST. +TEST_F(DBLazyEntityTest, EmbeddedForceVerifyReadsWholeRecordWithVerifyOff) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + const std::string key = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + IngestEmbeddedEntity(options, key, columns); + + ReadOptions ro; + ro.verify_checksums = false; // globally off + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ro, db_->DefaultColumnFamily(), key, &lazy)); + + ASSERT_OK(options.statistics->Reset()); + + PinnableSlice range; + Status status; + LazyColumnReadRequest read; + read.column = &lazy[1]; + read.offset = 1000; + read.length = 100; + read.force_verify = true; + read.result = ⦥ + read.status = &status; + ASSERT_OK(lazy.MultiResolve(/*num_reads=*/1, &read)); + ASSERT_OK(status); + ASSERT_EQ(range, big.substr(1000, 100)); + + // force_verify forces the whole embedded record (payload + trailer) to be + // read and verified even though verify_checksums is off, so strictly more + // than the requested 100 bytes are read. + ASSERT_GT(BlobBytesRead(options), big.size()); +} + +// Phase 1/2 statistics: an actual partial read (separate-file) bumps the lazy +// partial-read tickers by the requested length and the bytes it avoided. +TEST_F(DBLazyEntityTest, PartialReadUpdatesLazyStats) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(options.statistics->Reset()); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + constexpr uint64_t kOffset = 1000; + constexpr size_t kLength = 100; + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], kOffset, kLength, &range)); + ASSERT_EQ(range, big.substr(kOffset, kLength)); + + // A partial read is both a lazy read and a partial read. + ASSERT_EQ(LazyReadCount(options), 1U); + ASSERT_EQ(LazyReadBytes(options), kLength); + ASSERT_EQ(LazyPartialReadCount(options), 1U); + ASSERT_EQ(LazyPartialBytesSaved(options), big.size() - kLength); +} + +// The same lazy stats are recorded for an embedded (same-file) partial read. +TEST_F(DBLazyEntityTest, EmbeddedPartialReadUpdatesLazyStats) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + const std::string key = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + IngestEmbeddedEntity(options, key, columns); + ASSERT_OK(options.statistics->Reset()); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + constexpr uint64_t kOffset = 1000; + constexpr size_t kLength = 100; + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], kOffset, kLength, &range)); + ASSERT_EQ(range, big.substr(kOffset, kLength)); + + ASSERT_EQ(LazyReadCount(options), 1U); + ASSERT_EQ(LazyReadBytes(options), kLength); + ASSERT_EQ(LazyPartialReadCount(options), 1U); + ASSERT_EQ(LazyPartialBytesSaved(options), big.size() - kLength); +} + +// Lazy read stats count every lazy storage read (whole-column and partial), but +// the partial tickers count only actual partial reads: a whole-column read and +// a force_verify (full verified) read bump the lazy-read tickers without the +// partial ones, and a cache-hit slice bumps neither (no storage read). +TEST_F(DBLazyEntityTest, LazyStatsCountReadsAndPartialsSeparately) { + Options options = GetLazyTestOptionsWithBlobCache(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(options.statistics->Reset()); + + // A whole-column read is a lazy read, but not a partial read. + { + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + PinnableSlice whole; + ASSERT_OK(lazy.ResolveColumn(lazy[1], &whole)); + } + ASSERT_EQ(LazyReadCount(options), 1U); + ASSERT_GE(LazyReadBytes(options), big.size()); // whole record (+ header) + ASSERT_EQ(LazyPartialReadCount(options), 0U); + ASSERT_EQ(LazyPartialBytesSaved(options), 0U); + + // A force_verify partial request becomes a full verified read: a lazy read, + // not a partial read. (Fresh column so the blob cache does not serve it.) + ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), "e2", + columns)); + ASSERT_OK(Flush()); + ASSERT_OK(options.statistics->Reset()); + { + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), + "e2", &lazy)); + PinnableSlice v; + Status s; + LazyColumnReadRequest read; + read.column = &lazy[1]; + read.offset = 1000; + read.length = 100; + read.force_verify = true; + read.result = &v; + read.status = &s; + ASSERT_OK(lazy.MultiResolve(/*num_reads=*/1, &read)); + ASSERT_OK(s); + } + ASSERT_EQ(LazyReadCount(options), 1U); + ASSERT_EQ(LazyPartialReadCount(options), 0U); + + // A partial read served from the blob cache does no storage read, so it + // counts as neither a lazy read nor a partial read. + ASSERT_OK(options.statistics->Reset()); + { + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + PinnableSlice range; + ASSERT_OK(lazy.ResolveColumnRange(lazy[1], /*offset=*/1000, /*length=*/100, + &range)); + ASSERT_EQ(range, big.substr(1000, 100)); + } + ASSERT_EQ(LazyReadCount(options), 0U); + ASSERT_EQ(LazyReadBytes(options), 0U); + ASSERT_EQ(LazyPartialReadCount(options), 0U); +} + +// A non-lazy read (ordinary GetEntity) does not touch the lazy read stats: the +// lazy tickers are gated on Env::IOActivity::kLazyResolve. +TEST_F(DBLazyEntityTest, NonLazyReadsDoNotUpdateLazyStats) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + ASSERT_OK(options.statistics->Reset()); + + // An ordinary (eager) GetEntity resolves the blob column via the normal path. + PinnableWideColumns eager; + ASSERT_OK( + db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(), key, &eager)); + ASSERT_GT(BlobBytesRead(options), 0U); // it did read the blob + ASSERT_EQ(LazyReadCount(options), 0U); + ASSERT_EQ(LazyReadBytes(options), 0U); + ASSERT_EQ(LazyPartialReadCount(options), 0U); +} + +// Deferred lazy resolve reads are attributed to Env::IOActivity::kLazyResolve, +// distinct from the kGetEntity of the initial entity read. +TEST_F(DBLazyEntityTest, LazyResolveReadsUseLazyResolveIOActivity) { + auto fs = std::make_shared(FileSystem::Default()); + std::unique_ptr env(NewCompositeEnv(fs)); + // Close the DB before the local Env is destroyed (even on an early ASSERT). + Defer close_db_on_exit([this]() { Close(); }); + + Options options = GetLazyTestOptions(); + options.env = env.get(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(4000, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + + LazyWideColumns lazy; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &lazy)); + + // The initial entity read touched only the SST, not the .blob file. + fs->ResetLastBlobReadIOActivity(); + + PinnableSlice value; + ASSERT_OK(lazy.ResolveColumn(lazy[1], &value)); + ASSERT_EQ(value, big); + + // The deferred blob-file read carried the lazy-resolve activity. + ASSERT_EQ(fs->last_blob_read_io_activity(), Env::IOActivity::kLazyResolve); +} + // The headline example: resolve *many blobs in a single call*. One MultiResolve // submits reads for several blob columns at once; the engine coalesces them // (and, later, issues them asynchronously) under the hood, filling each @@ -631,6 +1322,43 @@ TEST_F(DBLazyEntityTest, BatchRejectsForeignAndNullColumns) { ASSERT_EQ(Slice(v_ok), big); } +// Regression test: an empty (default-constructed) LazyWideColumnsBatch has a +// null rep_, and a standalone column (from GetEntityLazy, not part of any +// batch) has a null owning_batch_rep_. MultiResolve must still reject the +// standalone column with InvalidArgument -- it must not fall through the +// nullptr == nullptr comparison and resolve a column that does not belong to +// this batch. +TEST_F(DBLazyEntityTest, EmptyBatchRejectsStandaloneColumn) { + Options options = GetLazyTestOptions(); + DestroyAndReopen(options); + + constexpr char key[] = "entity"; + const std::string big(200, 'a'); + const WideColumns columns{{kDefaultWideColumnName, "inline"}, {"data", big}}; + ASSERT_OK( + db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key, columns)); + ASSERT_OK(Flush()); + + LazyWideColumns standalone; + ASSERT_OK(db_->GetEntityLazy(ReadOptions(), db_->DefaultColumnFamily(), key, + &standalone)); + + // An empty batch owns no columns, so resolving any column through it must + // fail rather than resolve the standalone column. + LazyWideColumnsBatch empty_batch; + ASSERT_TRUE(empty_batch.empty()); + + PinnableSlice value; + Status status; + LazyColumnReadRequest read; + read.column = &standalone[1]; + read.result = &value; + read.status = &status; + ASSERT_OK(empty_batch.MultiResolve(/*num_reads=*/1, &read)); + ASSERT_TRUE(status.IsInvalidArgument()) << status.ToString(); + ASSERT_TRUE(value.empty()); +} + // Wide-column entities do not support user-defined timestamps: PutEntity() // rejects a UDT column family (see DBWideBasicTest.PutEntityTimestampError), so // no entity can exist there. GetEntityLazy() inherits this feature-wide diff --git a/db/wide/lazy_wide_columns.cc b/db/wide/lazy_wide_columns.cc index ed9ce0ed68..ec627d3499 100644 --- a/db/wide/lazy_wide_columns.cc +++ b/db/wide/lazy_wide_columns.cc @@ -25,15 +25,50 @@ #include "rocksdb/wide_columns.h" // Current implementation of the lazy blob resolution API. Enumeration and -// whole-column resolution are functional; a byte range is served by resolving -// the whole column and slicing it. A future phase reads only the requested -// bytes from storage (skipping checksum verification and cache-fill for -// uncompressed blobs) instead of the whole column. Cross-key coalescing and -// async execution are also future work; here LazyWideColumnsBatch::MultiResolve -// simply routes each read to its owning per-key result, and MultiGetEntityLazy -// fills the batch key-by-key. +// whole-column resolution are functional. Byte-range reads of an uncompressed +// blob -- whether it lives in a separate blob file or is embedded (same-file) +// in the SST -- are served by reading only the requested bytes from storage +// (skipping checksum verification and cache-fill); other cases (compressed, +// whole-column, already-cached, or force_verify) resolve the whole column and +// slice it. Cross-key coalescing and async execution are future work; here +// LazyWideColumnsBatch::MultiResolve simply routes each read to its owning +// per-key result, and MultiGetEntityLazy fills the batch key-by-key. namespace ROCKSDB_NAMESPACE { + +namespace { +// Scope guard used while resolving a lazy result: sets the thread's +// ThreadStatus operation to OP_LAZY_RESOLVE for the duration, then returns it +// to OP_UNKNOWN on exit. A resolve is a self-contained top-level operation, so +// it starts and ends with no active op rather than restoring a prior one -- +// thread operations do not stack, and any op lingering from the originating +// GetEntity/MultiGetEntity call is stale. (This mirrors the SetThreadOperation +// + ResetThreadStatus pattern other top-level ops such as compaction/DBOpen +// use.) +// +// Setting the parallel operation type keeps the io_activity and thread +// operation consistent while the deferred reads (which carry +// Env::IOActivity::kLazyResolve) run: db_stress asserts every file read's +// io_activity matches the one implied by the current thread operation (see +// db_stress_env_wrapper.h / TEST_GetExpectedIOActivity), and ThreadStatus +// reporting then attributes these reads to the lazy-resolve operation. +// +// TODO: unified RAII wrappers for thread status updates +class LazyResolveThreadOpScope { + public: + LazyResolveThreadOpScope() { + ThreadStatusUtil::SetThreadOperation( + ThreadStatus::OperationType::OP_LAZY_RESOLVE); + } + ~LazyResolveThreadOpScope() { + ThreadStatusUtil::SetThreadOperation( + ThreadStatus::OperationType::OP_UNKNOWN); + } + LazyResolveThreadOpScope(const LazyResolveThreadOpScope&) = delete; + LazyResolveThreadOpScope& operator=(const LazyResolveThreadOpScope&) = delete; +}; +} // namespace + // Internal representation. Owns the serialized-entity backing buffer + inline // columns (via `entity_`), the decoded blob references, the per-column views, // the on-demand blob resolver, and the SuperVersion pin that keeps all of it @@ -127,52 +162,23 @@ class LazyWideColumns::Rep { // writing the per-read outcome to read.status / read.result. Shared by the // single-entity and batch MultiResolve paths. void ResolveOneRead(size_t index, LazyColumnReadRequest& read) { + assert(resolver_); + + // Delegate to the resolver, which owns the partial-vs-whole decision: for a + // strict sub-range of an uncompressed blob reference (separate-file or + // embedded, and not force_verify) it reads only the requested bytes, + // skipping checksum and cache-fill; every other case resolves the whole + // column (caching it) and slices. Repeated whole-column reads therefore + // read the blob at most once; partial reads own their own bytes and are not + // cached. A null read.result still resolves (to surface I/O/integrity + // errors on read.status and honor force_verify), just without producing + // output -- ResolveColumnRange handles that case. + const Status s = resolver_->ResolveColumnRange( + index, read.offset, read.length, read.force_verify, read.result); + if (read.status) { - *read.status = Status::OK(); + *read.status = s; } - - // Resolve the whole column (inline value directly, blob reference via the - // resolver, which caches so repeated reads of one column read the blob at - // most once), then slice out the requested byte range. - // - // TODO(lazy-blob-resolution-phase1): for a strict sub-range of an - // uncompressed separate-file blob on a cache miss, read only the requested - // bytes from storage (skipping checksum and cache-fill) instead of the - // whole column; a further phase (TODO(lazy-blob-resolution-phase2)) does - // the same for embedded (same-file) blobs. - const LazyWideColumn& info = columns_[index]; - Slice whole; - Status s; - if (!info.is_reference()) { - whole = *info.inline_value(); - } else { - assert(resolver_); - s = resolver_->ResolveColumn(index, &whole); - } - - if (!s.ok()) { - if (read.status) { - *read.status = s; - } - return; - } - - if (read.result != nullptr) { - read.result->Reset(); - if (read.offset >= whole.size()) { - // Offset at/past the end clamps to empty (not an error). - read.result->PinSlice(Slice(), nullptr); - } else { - const size_t offset = static_cast(read.offset); - const size_t avail = whole.size() - offset; - const size_t len = - (read.length == kLazyWholeColumn || read.length > avail) - ? avail - : read.length; - read.result->PinSlice(Slice(whole.data() + offset, len), nullptr); - } - } - // read.status already OK. } }; @@ -194,6 +200,7 @@ const LazyWideColumn& LazyWideColumns::operator[](size_t i) const { Status LazyWideColumns::MultiResolve(size_t num_reads, LazyColumnReadRequest* reads) { + LazyResolveThreadOpScope thread_op_scope; for (size_t i = 0; i < num_reads; ++i) { LazyColumnReadRequest& read = reads[i]; if (read.status) { @@ -285,6 +292,7 @@ LazyWideColumns& LazyWideColumnsBatch::operator[](size_t i) { Status LazyWideColumnsBatch::MultiResolve(size_t num_reads, LazyColumnReadRequest* reads) { + LazyResolveThreadOpScope thread_op_scope; const void* this_rep = rep_.get(); for (size_t i = 0; i < num_reads; ++i) { LazyColumnReadRequest& read = reads[i]; @@ -301,13 +309,18 @@ Status LazyWideColumnsBatch::MultiResolve(size_t num_reads, continue; } // Route the read to the entity that owns its column, and require that - // entity to belong to this batch. + // entity to belong to this batch. The owning_batch_rep_ == nullptr check is + // essential: a standalone column (from GetEntityLazy) has + // owning_batch_rep_ == nullptr, and an empty/default-constructed batch has + // this_rep == nullptr, so without it a foreign standalone column would slip + // through the nullptr == nullptr comparison instead of being rejected. // TODO(lazy-blob-resolution-phase3): group reads per (CF, Version, blob // file) across entities and coalesce them; the request shape is unchanged. LazyWideColumns::Rep* entity_rep = static_cast(read.column) ->parent_rep_; - if (entity_rep == nullptr || entity_rep->owning_batch_rep_ != this_rep) { + if (entity_rep == nullptr || entity_rep->owning_batch_rep_ == nullptr || + entity_rep->owning_batch_rep_ != this_rep) { if (read.status) { *read.status = Status::InvalidArgument("Column does not belong to this batch"); @@ -385,10 +398,14 @@ Status LazyWideColumnsHelper::Finalize( // Take ownership of the SuperVersion pin and stand up the resolver bound to // the (address-stable, since Rep is heap-allocated) entity columns + blob - // references. + // references. The resolver's deferred blob-byte reads are attributed to + // Env::IOActivity::kLazyResolve (distinct from the kGetEntity/kMultiGetEntity + // of the initial entity read that already completed via GetImpl). rep.pin_ = std::move(pin); rep.user_key_.assign(user_key.data(), user_key.size()); - rep.resolver_.emplace(version, read_options, blob_file_cache, + ReadOptions resolve_read_options(read_options); + resolve_read_options.io_activity = Env::IOActivity::kLazyResolve; + rep.resolver_.emplace(version, resolve_read_options, blob_file_cache, allow_write_path_fallback); rep.resolver_->Reset(Slice(rep.user_key_), &rep.entity_.columns(), &rep.blob_columns_, same_file_reader); diff --git a/db/wide/read_path_blob_resolver.cc b/db/wide/read_path_blob_resolver.cc index a986cad5c3..281051eac0 100644 --- a/db/wide/read_path_blob_resolver.cc +++ b/db/wide/read_path_blob_resolver.cc @@ -34,6 +34,13 @@ void ReadPathBlobResolver::Reset( Status ReadPathBlobResolver::ResolveColumn(size_t column_index, Slice* resolved_value) { + return ResolveColumnInternal(column_index, /*force_verify=*/false, + resolved_value); +} + +Status ReadPathBlobResolver::ResolveColumnInternal(size_t column_index, + bool force_verify, + Slice* resolved_value) { assert(columns_); assert(resolved_value); @@ -68,16 +75,50 @@ Status ReadPathBlobResolver::ResolveColumn(size_t column_index, constexpr FilePrefetchBuffer* prefetch_buffer = nullptr; constexpr uint64_t* bytes_read = nullptr; - // 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); + // 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); + } if (!status.ok()) { resolved_cache_.pop_back(); } else { @@ -90,6 +131,104 @@ Status ReadPathBlobResolver::ResolveColumn(size_t column_index, return status; } +namespace { + +// Pin [range_offset, range_offset + range_length) of `whole` into *result, +// clamped to `whole`'s size (offset at/past end -> empty). The bytes are not +// copied or owned here: *result points into `whole`, whose storage must outlive +// *result (the resolver's whole-column cache / entity buffer, which lives as +// long as this resolver). +void PinClampedSubRange(const Slice& whole, uint64_t range_offset, + size_t range_length, PinnableSlice* result) { + result->Reset(); + if (range_offset >= whole.size()) { + result->PinSlice(Slice(), nullptr); + return; + } + const size_t off = static_cast(range_offset); + const size_t avail = whole.size() - off; + const size_t len = range_length > avail ? avail : range_length; + result->PinSlice(Slice(whole.data() + off, len), nullptr); +} + +} // namespace + +Status ReadPathBlobResolver::ResolveColumnRange(size_t column_index, + uint64_t range_offset, + size_t range_length, + bool force_verify, + PinnableSlice* result) { + assert(columns_); + + if (column_index >= columns_->size()) { + return Status::InvalidArgument("Column index out of bounds"); + } + + // 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); + } + + const BlobIndex* blob_index_ptr = + blob_resolver_util::FindBlobColumn(blob_columns_, column_index); + + // Decide whether the I/O-saving partial path applies. It requires a blob + // 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 + // 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 && + blob_resolver_util::FindInCache(resolved_cache_, column_index) == + nullptr) { + const BlobIndex& blob_index = *blob_index_ptr; + const bool separate_file_ok = + !blob_index.IsSameFile() && blob_fetcher_.SupportsRangeRead(); + const bool same_file_ok = + blob_index.IsSameFile() && same_file_reader_ != nullptr; + if (!blob_index.IsInlined() && blob_index.compression() == kNoCompression && + (separate_file_ok || same_file_ok)) { + const uint64_t value_size = blob_index.size(); + const bool strict_subrange = + range_offset > 0 || range_length < value_size; + if (strict_subrange) { + result->Reset(); + if (range_offset >= value_size || range_length == 0) { + // Nothing to read; empty (not an error), no I/O. + result->PinSlice(Slice(), nullptr); + return Status::OK(); + } + const size_t avail = static_cast(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); + } + } + } + + // 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. + Slice whole; + Status s = ResolveColumnInternal(column_index, force_verify, &whole); + if (!s.ok()) { + return s; + } + PinClampedSubRange(whole, range_offset, range_length, result); + return Status::OK(); +} + Status ReadPathBlobResolver::ResolveColumns( const std::vector& column_indices, std::vector* resolved_values) { diff --git a/db/wide/read_path_blob_resolver.h b/db/wide/read_path_blob_resolver.h index afaa246ae1..ebe6847b98 100644 --- a/db/wide/read_path_blob_resolver.h +++ b/db/wide/read_path_blob_resolver.h @@ -72,6 +72,29 @@ class ReadPathBlobResolver { // - I/O error occurred while fetching the blob Status ResolveColumn(size_t column_index, Slice* resolved_value); + // Resolve a byte sub-range [range_offset, range_offset + range_length) of the + // column at `column_index` into *result (zero-copy). `range_length` may be + // larger than the remaining bytes (or SIZE_MAX for "to the end"); it is + // clamped to the column's logical size. An offset at/past the end yields an + // empty result (not an error). + // + // Reads only the requested bytes -- skipping whole-record checksum + // verification and blob-cache population -- only for a strict sub-range of an + // uncompressed blob reference (in a separate blob file, or + // embedded/same-file) that is not already resolved, when !force_verify and + // the read is servable (a Version-backed range read for separate-file refs, + // or a SameFileBlobReader for embedded refs). Every other case (inline + // column, already-resolved column, inlined blob, compressed reference, a + // whole-column read, or force_verify) resolves the whole column (caching it, + // verifying under ReadOptions::verify_checksums) and slices the requested + // range out of it. + // + // Unlike ResolveColumn, a partial read is NOT cached: *result owns (or pins) + // its own bytes, independent of this resolver's whole-column cache. + Status ResolveColumnRange(size_t column_index, uint64_t range_offset, + size_t range_length, bool force_verify, + PinnableSlice* result); + // Resolve multiple columns in the order provided by `column_indices`. // Resolved blob values are cached exactly as if ResolveColumn() were called // repeatedly. @@ -101,6 +124,13 @@ class ReadPathBlobResolver { } private: + // 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, + Slice* resolved_value); + // Owns its ReadOptions: a resolver's lifetime is independent of the caller // that created it (e.g. it may outlive the originating ReadOptions once // returned as part of a lazy result). diff --git a/db_stress_tool/db_stress_test_base.cc b/db_stress_tool/db_stress_test_base.cc index 9864fd000b..ccfff8a3a3 100644 --- a/db_stress_tool/db_stress_test_base.cc +++ b/db_stress_tool/db_stress_test_base.cc @@ -551,6 +551,51 @@ void StressTest::MaybeTestGetEntityLazy( ResolveLazyEntity(thread, key.ToString(), reference, lazy); } +namespace { +// Chooses a byte range for a lazy column read in db_stress fuzzing. Half the +// time (or when the size is unknown or zero) reads the whole column; otherwise +// a random sub-range, occasionally with an offset past the end or a length +// beyond the remainder to exercise the API's clamp-to-empty / +// clamp-to-remainder behavior. `known_size` is the column's logical size when +// known without I/O. +void PickLazyReadRange(Random* rand, std::optional known_size, + uint64_t* offset, size_t* length) { + if (!known_size.has_value() || *known_size == 0 || rand->OneIn(2)) { + *offset = 0; + *length = kLazyWholeColumn; + return; + } + // A small over-range so ~1/16 of picks land past the end or beyond the + // remainder, exercising the API's clamp-to-empty / clamp-to-remainder + // behavior. db_stress column values fit comfortably in int. + const int size_plus_some = static_cast(*known_size + (*known_size >> 4)); + // Offset in [0, size_plus_some); occasionally past the end to exercise + // clamping. + *offset = rand->Uniform(size_plus_some); + // Length: usually a bounded sub-range (which may exceed the remainder), + // sometimes to the end of the column. + if (rand->OneIn(4)) { + *length = kLazyWholeColumn; + } else { + *length = rand->Uniform(size_plus_some); + } +} + +// The sub-range of `value` that a partial read with (offset, length) should +// return, applying the API's documented clamping: an offset at/past the end +// yields empty, length is clamped to the remainder, and kLazyWholeColumn reads +// to the end. +Slice ExpectedLazySlice(const Slice& value, uint64_t offset, size_t length) { + if (offset >= value.size()) { + return Slice(); + } + const size_t off = static_cast(offset); + const size_t avail = value.size() - off; + const size_t len = length > avail ? avail : length; + return Slice(value.data() + off, len); +} +} // namespace + void StressTest::MaybeTestMultiGetEntityLazy( ThreadState* thread, const ReadOptions& read_opts, ColumnFamilyHandle* cfh, size_t num_keys, const Slice* keys, @@ -699,9 +744,17 @@ void StressTest::MaybeTestMultiGetEntityLazy( std::vector statuses(chosen.size()); std::vector reads(chosen.size()); for (size_t j = 0; j < chosen.size(); ++j) { - reads[j].column = &batch[chosen[j].first][chosen[j].second]; - reads[j].offset = 0; - reads[j].length = kLazyWholeColumn; + 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. + const WideColumns* reference = entity_ref[i]; + const std::optional size_hint = + reference ? std::optional((*reference)[c].value().size()) + : batch[i][c].logical_size(); + reads[j].column = &batch[i][c]; + PickLazyReadRange(&thread->rand, size_hint, &reads[j].offset, + &reads[j].length); reads[j].result = &results[j]; reads[j].status = &statuses[j]; } @@ -732,14 +785,17 @@ void StressTest::MaybeTestMultiGetEntityLazy( } continue; } - const Slice& expected = (*reference)[chosen[j].second].value(); + const Slice expected = + ExpectedLazySlice((*reference)[chosen[j].second].value(), + reads[j].offset, reads[j].length); if (Slice(results[j]) != expected) { shared->SetVerificationFailure(); fprintf(stderr, - "MultiGetEntityLazy value mismatch for key %s column %zu: lazy " - "%s, reference %s\n", + "MultiGetEntityLazy value mismatch for key %s column %zu " + "[offset %" PRIu64 " len %zu]: lazy %s, reference %s\n", StringToHex(keys[chosen[j].first].ToString()).c_str(), - chosen[j].second, Slice(results[j]).ToString(true).c_str(), + chosen[j].second, reads[j].offset, reads[j].length, + Slice(results[j]).ToString(true).c_str(), expected.ToString(true).c_str()); } } @@ -814,9 +870,15 @@ void StressTest::ResolveLazyEntity(ThreadState* thread, const std::string& key, std::vector statuses(chosen.size()); std::vector reads(chosen.size()); for (size_t j = 0; j < chosen.size(); ++j) { - reads[j].column = &lazy[chosen[j]]; - reads[j].offset = 0; - reads[j].length = kLazyWholeColumn; + 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. + const std::optional size_hint = + reference ? std::optional((*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].result = &results[j]; reads[j].status = &statuses[j]; } @@ -846,14 +908,16 @@ void StressTest::ResolveLazyEntity(ThreadState* thread, const std::string& key, } continue; } - if (Slice(results[j]) != (*reference)[chosen[j]].value()) { + const Slice expected = ExpectedLazySlice((*reference)[chosen[j]].value(), + reads[j].offset, reads[j].length); + if (Slice(results[j]) != expected) { shared->SetVerificationFailure(); fprintf(stderr, - "GetEntityLazy value mismatch for key %s column %zu: lazy %s, " - "reference %s\n", - StringToHex(key).c_str(), chosen[j], - Slice(results[j]).ToString(true).c_str(), - (*reference)[chosen[j]].value().ToString(true).c_str()); + "GetEntityLazy value mismatch for key %s column %zu " + "[offset %" PRIu64 " len %zu]: lazy %s, reference %s\n", + StringToHex(key).c_str(), chosen[j], reads[j].offset, + reads[j].length, Slice(results[j]).ToString(true).c_str(), + expected.ToString(true).c_str()); } } } diff --git a/env/env.cc b/env/env.cc index 3a2efd5a85..d858eb881b 100644 --- a/env/env.cc +++ b/env/env.cc @@ -771,6 +771,8 @@ std::string Env::IOActivityToString(IOActivity activity) { return "MultiGetEntity"; case Env::IOActivity::kGetFileChecksumsFromCurrentManifest: return "GetFileChecksumsFromCurrentManifest"; + case Env::IOActivity::kLazyResolve: + return "LazyResolve"; case Env::IOActivity::kUnknown: return "Unknown"; default: diff --git a/env/env_test.cc b/env/env_test.cc index 9f1928e1e4..0a7abfea35 100644 --- a/env/env_test.cc +++ b/env/env_test.cc @@ -4625,6 +4625,9 @@ class TestIOActivity : public testing::Test { TEST_F(TestIOActivity, IOActivityToString) { ASSERT_EQ(Env::IOActivityToString(Env::IOActivity::kMultiGet), "MultiGet"); + ASSERT_EQ(Env::IOActivityToString(Env::IOActivity::kLazyResolve), + "LazyResolve"); + ASSERT_EQ(Env::IOActivityToString(Env::IOActivity::kCustomIOActivity80), "CustomIOActivity80"); ASSERT_EQ(Env::IOActivityToString(Env::IOActivity::kCustomIOActivityA9), diff --git a/include/rocksdb/env.h b/include/rocksdb/env.h index 8ba4198714..bd875a392b 100644 --- a/include/rocksdb/env.h +++ b/include/rocksdb/env.h @@ -464,6 +464,12 @@ class Env : public Customizable { kGetEntity = 8, kMultiGetEntity = 9, kGetFileChecksumsFromCurrentManifest = 10, + // Deferred blob-byte reads performed while resolving a lazy wide-column + // result (DB::GetEntityLazy / DB::MultiGetEntityLazy), i.e. reads issued by + // LazyWideColumns::MultiResolve after the originating call returned. + // Distinct from kGetEntity/kMultiGetEntity, which cover the initial entity + // read. + kLazyResolve = 11, // Enums after this, up to 0x7F, are reserved for future use for the public // RocksDB API (i.e. they should be "non-custom" IO activities). Make sure // to also update IOActivityToString when adding new values. diff --git a/include/rocksdb/lazy_wide_columns.h b/include/rocksdb/lazy_wide_columns.h index b0baa72690..66f25de312 100644 --- a/include/rocksdb/lazy_wide_columns.h +++ b/include/rocksdb/lazy_wide_columns.h @@ -40,29 +40,31 @@ inline constexpr size_t kLazyWholeColumn = std::numeric_limits::max(); // does not belong to the target (InvalidArgument) -- there are no untyped // indices to mix up. `result` and `status` are out-params filled by the resolve // call, so both pointed-to objects must outlive it. Zero-copy: on success -// `*result` is a view into a stable backing buffer owned by the owning -// LazyWideColumns. +// `*result` is a view into stable backing storage -- either a buffer owned or +// pinned by the owning LazyWideColumns (for whole-column / cached reads) or, +// for a partial read that reads only the requested bytes, storage owned by +// `*result` itself. struct LazyColumnReadRequest { // The column to read (from the target result's operator[] / iteration). Must // belong to the result being resolved. const LazyWideColumn* column = nullptr; - // Starting byte offset within the column's logical value. An offset at or - // past the end of the value yields an empty result (not an error). A nonzero - // offset makes this a partial read; see `force_verify` for checksums. + // `offset` and `length` select the byte range [offset, offset + length) of + // the column's logical, post-decompression value to put in the result. length + // == kLazyWholeColumn reads to the end; otherwise length is clamped to the + // bytes remaining after offset, and an offset at or past the end yields an + // empty result rather than an error. // - // Not yet optimized: a partial read currently resolves the whole column and - // slices it, so for now it saves no I/O and skips no checksum. + // A strict sub-range -- a nonzero offset, or a length short of the remainder + // -- makes this a partial read request. It becomes an actual partial read + // only for an uncompressed blob reference, whether in a separate blob file or + // embedded in the same SST. Such a read fetches only the requested bytes, + // does not populate the blob cache, and skips whole-record checksum + // verification, which a sub-range cannot cover; set force_verify to override. + // Any other request, including a compressed reference or a whole-column read, + // resolves the whole column, verifying under ReadOptions::verify_checksums, + // and slices the requested range out of it. uint64_t offset = 0; - - // Number of bytes to read starting at `offset`, clamped to the end of the - // value. kLazyWholeColumn reads the entire remainder from `offset`. Anything - // less (or a nonzero `offset`) is a partial read: it returns only the - // requested bytes and by default skips whole-record checksum verification - // (see `force_verify`). - // - // Not yet optimized: a partial read currently resolves the whole column and - // slices it, so for now it saves no I/O and skips no checksum. size_t length = kLazyWholeColumn; // Extra lever to prioritize checksum verification over I/O efficiency, mainly @@ -72,9 +74,6 @@ struct LazyColumnReadRequest { // when that exceeds the requested range or ReadOptions::verify_checksums is // off. No effect where verification already happens (e.g. a whole-column read // under verify_checksums) or where the format has no usable checksum. - // - // Not yet honored: because reads currently resolve the whole column, this - // flag has no effect for now. bool force_verify = false; // Output: on OK status, a zero-copy view of the requested bytes. diff --git a/include/rocksdb/statistics.h b/include/rocksdb/statistics.h index 4f194a4908..e21068b176 100644 --- a/include/rocksdb/statistics.h +++ b/include/rocksdb/statistics.h @@ -657,6 +657,26 @@ enum Tickers : uint32_t { // Number of SubmitReadAsync calls that fell back to a synchronous read FILE_SUBMIT_ASYNC_READ_FALLBACK, + // Lazy wide-column resolution (DB::GetEntityLazy / DB::MultiGetEntityLazy): + // metrics for the blob reads issued from storage while resolving a lazy + // result (attributed via Env::IOActivity::kLazyResolve). Cache hits, which do + // no storage read, are not counted. + // + // Number of storage reads issued by lazy resolution (both whole-column and + // partial byte-range reads). + BLOB_DB_LAZY_READ_COUNT, + // Bytes actually read from storage by those reads (on-disk record bytes for + // whole-column reads; the requested sub-range length for partial reads). + BLOB_DB_LAZY_READ_BYTES, + // Of the above, the number that were actual partial (byte-range) reads: only + // a requested sub-range of an uncompressed blob's value was read (skipping + // the whole-column read). A whole-column read or a force_verify read is not + // counted here. + BLOB_DB_LAZY_PARTIAL_READ_COUNT, + // Bytes not read thanks to those partial reads: for each, the column's + // logical value size minus the bytes actually read. + BLOB_DB_LAZY_PARTIAL_BYTES_SAVED, + TICKER_ENUM_MAX }; diff --git a/include/rocksdb/thread_status.h b/include/rocksdb/thread_status.h index 07c872c0e9..bd8d5c0586 100644 --- a/include/rocksdb/thread_status.h +++ b/include/rocksdb/thread_status.h @@ -57,6 +57,7 @@ struct ThreadStatus { OP_GETENTITY, OP_MULTIGETENTITY, OP_GET_FILE_CHECKSUMS_FROM_CURRENT_MANIFEST, + OP_LAZY_RESOLVE, NUM_OP_TYPES }; diff --git a/monitoring/statistics.cc b/monitoring/statistics.cc index f1059da725..20828a9748 100644 --- a/monitoring/statistics.cc +++ b/monitoring/statistics.cc @@ -338,6 +338,11 @@ const std::vector> TickersNameMap = { {CHECKPOINT_WRITE_BYTES, "rocksdb.checkpoint.write.bytes"}, {FILE_SUBMIT_ASYNC_READ_FALLBACK, "rocksdb.file.submit.async.read.fallback"}, + {BLOB_DB_LAZY_READ_COUNT, "rocksdb.blobdb.lazy.read.count"}, + {BLOB_DB_LAZY_READ_BYTES, "rocksdb.blobdb.lazy.read.bytes"}, + {BLOB_DB_LAZY_PARTIAL_READ_COUNT, "rocksdb.blobdb.lazy.partial.read.count"}, + {BLOB_DB_LAZY_PARTIAL_BYTES_SAVED, + "rocksdb.blobdb.lazy.partial.bytes.saved"}, }; const std::vector> HistogramsNameMap = { diff --git a/monitoring/thread_status_util_debug.cc b/monitoring/thread_status_util_debug.cc index 7b6211bb54..a0f99e1c7f 100644 --- a/monitoring/thread_status_util_debug.cc +++ b/monitoring/thread_status_util_debug.cc @@ -53,6 +53,8 @@ Env::IOActivity ThreadStatusUtil::TEST_GetExpectedIOActivity( case ThreadStatus::OperationType:: OP_GET_FILE_CHECKSUMS_FROM_CURRENT_MANIFEST: return Env::IOActivity::kGetFileChecksumsFromCurrentManifest; + case ThreadStatus::OperationType::OP_LAZY_RESOLVE: + return Env::IOActivity::kLazyResolve; default: return Env::IOActivity::kUnknown; } diff --git a/table/block_based/block_based_table_reader.cc b/table/block_based/block_based_table_reader.cc index 700ad0cbec..d28d2d0212 100644 --- a/table/block_based/block_based_table_reader.cc +++ b/table/block_based/block_based_table_reader.cc @@ -1374,6 +1374,58 @@ Status BlockBasedTable::GetSameFileBlob(const ReadOptions& read_options, return ResolveEmbeddedBlobPinned(read_options, blob_index, value); } +Status BlockBasedTable::ResolveEmbeddedBlobRangeCached( + const ReadOptions& read_options, const BlobIndex& blob_index, + uint64_t range_offset, size_t range_length, PinnableSlice* value) const { + assert(value != nullptr); + 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); + if (!s.ok()) { + return s; + } + + return rep_->blob_source_->GetSimpleGen2BlobRange( + read_options, rep_->base_cache_key, rep_->file.get(), blob_index.offset(), + payload_size, rep_->footer.checksum_type(), + rep_->footer.base_context_checksum(), blob_index.compression(), + 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, diff --git a/table/block_based/block_based_table_reader.h b/table/block_based/block_based_table_reader.h index f32126fe52..7ea7089abb 100644 --- a/table/block_based/block_based_table_reader.h +++ b/table/block_based/block_based_table_reader.h @@ -237,6 +237,19 @@ class BlockBasedTable : public TableReader, public SameFileBlobReader { const BlobIndex& blob_index, PinnableSlice* value) const; + // Range-read counterpart of ResolveEmbeddedBlobCached: resolve only + // [range_offset, range_offset + range_length) of an uncompressed same-file + // payload (skipping the trailer + checksum verification, and never inserting + // into the blob cache). The caller must ensure range_offset + range_length <= + // blob_index.size(). Requires a BlobSource (rep_->blob_source_ != nullptr): + // byte-range embedded reads are only issued on the lazy read path, which + // always has one, so there is no direct-pinned range variant. + Status ResolveEmbeddedBlobRangeCached(const ReadOptions& read_options, + const BlobIndex& blob_index, + uint64_t range_offset, + 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 // through the blob cache (ResolveEmbeddedBlobCached) when a BlobSource is @@ -245,6 +258,15 @@ class BlockBasedTable : public TableReader, public SameFileBlobReader { const BlobIndex& blob_index, 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 diff --git a/unreleased_history/new_features/lazy_wide_columns.md b/unreleased_history/new_features/lazy_wide_columns.md index 617d4f7f49..448dcebf43 100644 --- a/unreleased_history/new_features/lazy_wide_columns.md +++ b/unreleased_history/new_features/lazy_wide_columns.md @@ -1 +1 @@ -Added an experimental lazy wide-column read API, `DB::GetEntityLazy()` / `DB::MultiGetEntityLazy()` (see `include/rocksdb/lazy_wide_columns.h`): the returned `LazyWideColumns` leaves blob-backed columns as unresolved references so the caller resolves only the columns it needs, and (like an iterator) it can be used after the call returns. Requires `max_open_files == -1`. +Added an experimental lazy wide-column read API, `DB::GetEntityLazy()` / `DB::MultiGetEntityLazy()` (see `include/rocksdb/lazy_wide_columns.h`). The returned `LazyWideColumns` leaves blob-backed columns as unresolved references and, like an iterator, stays valid after the call returns, so the caller resolves only the columns -- and only the byte ranges -- it needs; a byte-range read of an uncompressed blob (separate-file or embedded) reads just those bytes from storage. Adds `Env::IOActivity::kLazyResolve` and `rocksdb.blobdb.lazy.*` statistics. Requires `max_open_files == -1`. diff --git a/util/thread_operation.h b/util/thread_operation.h index 91c26f9907..e2a9d6b1f1 100644 --- a/util/thread_operation.h +++ b/util/thread_operation.h @@ -49,6 +49,7 @@ static OperationInfo global_operation_table[] = { {ThreadStatus::OP_MULTIGETENTITY, "MultiGetEntity"}, {ThreadStatus::OP_GET_FILE_CHECKSUMS_FROM_CURRENT_MANIFEST, "GetFileChecksumsFromCurrentManifest"}, + {ThreadStatus::OP_LAZY_RESOLVE, "LazyResolve"}, };