Add experimental embedded blob SST support (#14851)

Summary:
Add EXPERIMENTAL embedded blob SST support for SstFileWriter through
OpenWithEmbeddedBlobs(). Eligible large values are written as same-file blob
records inline in a block-based SST as values are added (interleaved with data
blocks), while table entries store same-file BlobIndex references that readers
resolve for Get, MultiGet, and iteration, including mixed embedded and
non-embedded wide-column values.

Embedded-blob handling is folded directly into BlockBasedTableBuilder rather than
living in SstFileWriter or a separate table-builder wrapper: SstFileWriter only
selects the mode (via TableBuilderOptions::embedded_blob_options), and the
builder writes blob records inline using its own file writer and running offset.
This is enabled by disabling index-value delta encoding for these SSTs — delta
encoding reconstructs a data block's offset from the previous block and so
requires byte-contiguous data blocks, which interleaved blob records would break.
With full (non-delta) index values, blob records can sit between data blocks, so
no entry buffering/replay is needed. To keep inline blob appends correctly
ordered with data-block writes, these SSTs use single-threaded (non-parallel)
compression. The mode is the only entry point today but the placement keeps it
open to generalization beyond SstFileWriter; regardless, this experimental
feature is expected only to have niche applications.

(Previous revisions of this change allowed delta-encoded index blocks by
putting all blobs at the beginning of the file, but that was a more awkward
and memory-hungry implementation due to buffering all the data blocks before
writing.)

The on-disk record format (SimpleGen2Blob: payload bytes followed by a 5-byte
trailer of a compression marker plus a builtin checksum that is context-modified
by the record's absolute file offset) lives in db/blob/blob_gen2_format.{h,cc},
which now owns both the read (ReadAndVerifySimpleGen2BlobRecord) and write
(WriteSimpleGen2BlobRecord) sides so the format is defined in one place. This is
expected to be reused for upcoming "blog file" support.

Readers need no record-layout metadata: same-file blob resolution is purely
absolute-offset keyed, and the per-record offset-modified checksum (plus a cheap
"record fits within the file" bound) is the corruption guard. The reader's only
embedded-blob metadata is presence: a best-effort auxiliary table property
(blob count and payload-byte totals, for diagnostics) whose mere presence signals
that the SST contains embedded blobs.

Reads route through the column family's BlobSource when a DB is attached, so
embedded payloads are served from / inserted into the blob value cache and
recorded in BLOB_DB_* statistics; the cache key is derived from the
SimpleGen2Blob offset scheme (the same scheme block-based SST blocks use), so
embedded blob records stay collision-free with data blocks even when the blob
cache and block cache are the same cache. Non-DB openers (SstFileReader,
sst_dump, repair, ingestion prevalidation) have no BlobSource and fall back to a
direct, uncached read.

Same-file BlobIndex references use blob file number 0 as the marker. That value
also serves as the invalid blob-file-number sentinel in broader metadata code,
but the meanings do not conflict when used carefully: only the embedded-SST
reader/writer path interprets 0 as same-file, while generic file-metadata paths
continue to reject it as invalid. Using 1 would be worse because legacy
"stackable" BlobDB can use low blob file numbers, including 1, so reserving it
would collide with real blob files.

Compression options remain in the public API as placeholders, but embedded blob
compression support is deferred. Integrating compression with
BlockBasedTableBuilder while avoiding copied CompressAndVerifyBlock-style logic
is tricky enough to deserve a separate, focused PR.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/14851

Test Plan:
- Added basic feature coverage to the crash test.
- Added BlobIndexTest.SameFileBlobIndex and BlobGarbageMeterTest.SameFileBlobIndex
  coverage for same-file BlobIndex encoding, display, recognition, and ignoring
  same-file references in blob-garbage accounting.
- Extended FileMetaDataTest.UpdateBoundariesBlobIndex to preserve the generic
  zero-file-number corruption check while keeping same-file embedded blob
  semantics at the table-reader/writer layers.
- SstFileReader embedded blob coverage: round-trip Get, MultiGet, and iterator
  reads; format_version gating; ignored placeholder compression options; the
  2048-byte default min_blob_size; wide-column mixed embedded/non-embedded
  values; and early append-error surfacing.
- Added an interleaved-layout test (small block_size with alternating
  small/large values) asserting the SST property index_value_is_delta_encoded==0,
  more than one data block, and that blob records are interspersed with data
  blocks (not a strict front prefix), with all values read back correctly via
  Get and iteration; replaces the old "ignored bytes before/after the blob record
  prefix" test.
- Added an embedded-record corruption test: flipping a byte inside a blob
  record's payload yields Corruption on read with verify_checksums (the
  offset-keyed record checksum is the guard now that the range pre-check is gone).
- Exercised the cached read path through BlobSource, including blob-cache
  hit/miss behavior and the shared blob_cache == block_cache configuration, in
  db_blob_index_test.
- Normal-path CPU regression check: release (DEBUG_LEVEL=0) db_bench on a
  non-embedded DB comparing this change vs upstream main, 3 interleaved reps of
  fillseq, fillrandom, readrandom, and readseq (5M keys, value_size=100,
  compression none, DB on /dev/shm). All deltas were within run-to-run noise
  (~1%), i.e. no measurable regression from adding the embedded-mode branch to
  the builder hot path.

Reviewed By: xingbowang

Differential Revision: D108564468

Pulled By: pdillinger

fbshipit-source-id: 5f01ffb1d40c6fd5b82d2451ec3342abb5040ca6

BUCK

@@ -35,6 +35,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
         "db/blob/blob_file_partition_manager.cc",
         "db/blob/blob_file_reader.cc",
         "db/blob/blob_garbage_meter.cc",
+        "db/blob/blob_gen2_format.cc",
         "db/blob/blob_log_format.cc",
         "db/blob/blob_log_sequential_reader.cc",
         "db/blob/blob_log_writer.cc",

CLAUDE.md

@@ -38,7 +38,7 @@ This document provides guidance for generating and reviewing code in the RocksDB
 
 **SIMD and Vectorization:** Leverage SIMD instructions (SSE, AVX) for data-parallel operations when appropriate. Structure data to enable auto-vectorization by the compiler. Consider explicit SIMD intrinsics for critical hot paths like checksum computation, encoding/decoding, and bulk data processing.
 
-**Branch Prediction:** Minimize unpredictable branches in hot paths. Use `LIKELY`/`UNLIKELY` macros to hint branch prediction. Consider branchless alternatives for simple conditionals. Order switch cases and if-else chains by frequency.
+**Branch Prediction:** Minimize unpredictable branches in hot paths. Use `LIKELY`/`UNLIKELY` macros to hint branch prediction, e.g. for error cases and other rare or otherwise costly cases, but NOT for predicting popular configurations. Consider branchless alternatives for simple conditionals. Order switch cases and if-else chains by frequency.
 
 **Memory and Resource Management:** Be mindful of memory allocations, especially in hot paths. Use RAII patterns, smart pointers, and RocksDB's memory management utilities appropriately.
 

CMakeLists.txt

@@ -775,6 +775,7 @@ set(SOURCES
         db/blob/blob_file_meta.cc
         db/blob/blob_file_reader.cc
         db/blob/blob_garbage_meter.cc
+        db/blob/blob_gen2_format.cc
         db/blob/blob_log_format.cc
         db/blob/blob_log_sequential_reader.cc
         db/blob/blob_log_writer.cc

cache/cache_key.h

@@ -127,6 +127,24 @@ class OffsetableCacheKey : private CacheKey {
     return CacheKey(file_num_etc64_, offset_etc64_ ^ offset);
   }
 
+  // Construct a CacheKey for a record located at byte `offset` within a file in
+  // which every distinct cached region is at least 5 bytes long. The low two
+  // offset bits are dropped, which is collision-free under that >= 5-byte (>= 4
+  // would suffice) guarantee and lets different kinds of records in the same
+  // file share a single keyspace without colliding.
+  //
+  // This is the canonical cache-key scheme for byte-offset records that carry
+  // the block-based "min 5 byte" guarantee: block-based SST blocks (see
+  // BlockBasedTable::GetCacheKey) and SimpleGen2Blob records (which always
+  // include a >= 5-byte trailer; see db/blob/blob_gen2_format.h). Because both
+  // use this same function, an SST's data blocks and its embedded blob records
+  // never collide even when the block cache and blob cache are the same cache.
+  // Keeping a single implementation here avoids a divergence bug that would
+  // silently alias the two keyspaces.
+  inline CacheKey WithOffsetForMinSizeRecord(uint64_t offset) const {
+    return WithOffset(offset >> 2);
+  }
+
   // The "common prefix" is a shared prefix for all the returned CacheKeys.
   // It is specific to the file but the same for all offsets within the file.
   static constexpr size_t kCommonPrefixSize = 8;

db/blob/blob_constants.h

@@ -11,6 +11,17 @@
 
 namespace ROCKSDB_NAMESPACE {
 
+// WARNING: This value is == kCurrentFileBlobIndexFileNumber.
+// Use this name only where file number zero means "no valid blob file" or
+// "current file" is not understood/supported.
 constexpr uint64_t kInvalidBlobFileNumber = 0;
 
+// WARNING: This value is == kInvalidBlobFileNumber.
+// Use this name only for BlobIndex references to the same physical file as what
+// is currently being read; generic blob-file metadata must treat zero as
+// invalid. (Using a distinct value like 1 was found to be more problematic,
+// e.g. because of legacy "stackable" blob implementation.)
+constexpr uint64_t kCurrentFileBlobIndexFileNumber = kInvalidBlobFileNumber;
+static_assert(kCurrentFileBlobIndexFileNumber == kInvalidBlobFileNumber);
+
 }  // namespace ROCKSDB_NAMESPACE

db/blob/blob_garbage_meter.cc

@@ -32,6 +32,9 @@ Status BlobGarbageMeter::GetBlobReferenceDetails(const ParsedInternalKey& ikey,
   if (blob_index.IsInlined() || blob_index.HasTTL()) {
     return Status::Corruption("Unexpected TTL/inlined blob index");
   }
+  if (blob_index.IsSameFile()) {
+    return Status::OK();
+  }
 
   *blob_file_number = blob_index.file_number();
   *bytes =
@@ -125,7 +128,9 @@ Status BlobGarbageMeter::ProcessEntityBlobReferences(
             !s.ok()) {
           return s;
         }
+        if (file_number != kInvalidBlobFileNumber) {
           AddFlow(file_number, blob_bytes, is_inflow);
+        }
         return Status::OK();
       });
 }

db/blob/blob_garbage_meter_test.cc

@@ -151,6 +151,24 @@ TEST(BlobGarbageMeterTest, PlainValue) {
   ASSERT_TRUE(blob_garbage_meter.flows().empty());
 }
 
+TEST(BlobGarbageMeterTest, SameFileBlobIndex) {
+  constexpr char user_key[] = "user_key";
+  constexpr SequenceNumber seq = 123;
+
+  const InternalKey key(user_key, seq, kTypeBlobIndex);
+  const Slice key_slice = key.Encode();
+
+  std::string value;
+  BlobIndex::EncodeBlob(&value, kCurrentFileBlobIndexFileNumber,
+                        /*offset=*/123, /*size=*/456, kNoCompression);
+
+  BlobGarbageMeter blob_garbage_meter;
+
+  ASSERT_OK(blob_garbage_meter.ProcessInFlow(key_slice, value));
+  ASSERT_OK(blob_garbage_meter.ProcessOutFlow(key_slice, value));
+  ASSERT_TRUE(blob_garbage_meter.flows().empty());
+}
+
 TEST(BlobGarbageMeterTest, CorruptInternalKey) {
   constexpr char corrupt_key[] = "i_am_corrupt";
   const Slice key_slice(corrupt_key);

db/blob/blob_gen2_format.cc

@@ -0,0 +1,111 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "db/blob/blob_gen2_format.h"
+
+#include <array>
+#include <cstring>
+#include <string>
+
+#include "file/random_access_file_reader.h"
+#include "file/writable_file_writer.h"
+#include "rocksdb/options.h"
+#include "rocksdb/slice.h"
+#include "table/format.h"
+#include "util/cast_util.h"
+#include "util/coding.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+Status ReadAndVerifySimpleGen2BlobRecord(
+    const ReadOptions& read_options, RandomAccessFileReader* file,
+    uint64_t record_offset, size_t payload_size, size_t record_size,
+    ChecksumType checksum_type, uint32_t base_context_checksum,
+    CompressionType expected_compression, char* buf) {
+  assert(file != nullptr);
+  assert(buf != nullptr);
+  assert(record_size == payload_size + kSimpleGen2BlobTrailerSize);
+
+  Slice result;
+  IOOptions opts;
+  IODebugContext dbg;
+  Status s = file->PrepareIOOptions(read_options, opts, &dbg);
+  if (s.ok()) {
+    s = file->Read(opts, record_offset, record_size, &result, buf, nullptr,
+                   &dbg);
+  }
+  if (!s.ok()) {
+    return s;
+  }
+  if (result.size() != record_size) {
+    return Status::Corruption("Could not read complete blob record");
+  }
+  // With mmap reads the data lands outside `buf`; copy it in so the caller can
+  // rely on `buf` owning the bytes (this is the only copy on the mmap path).
+  // TODO: fix this extra memcpy in the mmap case
+  if (result.data() != buf) {
+    memcpy(buf, result.data(), record_size);
+  }
+
+  const char* record = buf;
+  const CompressionType compression =
+      static_cast<CompressionType>(record[payload_size]);
+  if (compression != expected_compression) {
+    return Status::Corruption(
+        "Blob record compression does not match blob index");
+  }
+
+  if (read_options.verify_checksums) {
+    uint32_t stored = DecodeFixed32(record + payload_size + 1);
+    stored -= ChecksumModifierForContext(base_context_checksum, record_offset);
+    const uint32_t computed = ComputeBuiltinChecksumWithLastByte(
+        checksum_type, record, payload_size, record[payload_size]);
+    if (stored != computed) {
+      return Status::Corruption("Blob record checksum mismatch in " +
+                                file->file_name() + " offset " +
+                                std::to_string(record_offset) + " size " +
+                                std::to_string(payload_size));
+    }
+  }
+
+  if (compression != kNoCompression) {
+    return Status::Corruption("Blob record compression is not supported");
+  }
+  return Status::OK();
+}
+
+IOStatus WriteSimpleGen2BlobRecord(WritableFileWriter* file,
+                                   const WriteOptions& write_options,
+                                   ChecksumType checksum_type,
+                                   uint32_t base_context_checksum,
+                                   uint64_t record_offset, const Slice& payload,
+                                   CompressionType compression) {
+  assert(file != nullptr);
+  // Placeholder for future embedded blob compression support; only
+  // uncompressed payloads are currently written.
+  assert(compression == kNoCompression);
+
+  std::array<char, kSimpleGen2BlobTrailerSize> trailer;
+  trailer[0] = lossless_cast<char>(compression);
+  uint32_t checksum = ComputeBuiltinChecksumWithLastByte(
+      checksum_type, payload.data(), payload.size(), /*last_byte=*/trailer[0]);
+  checksum += ChecksumModifierForContext(base_context_checksum, record_offset);
+  EncodeFixed32(trailer.data() + 1, checksum);
+
+  IOOptions opts;
+  IOStatus io_s = WritableFileWriter::PrepareIOOptions(write_options, opts);
+  if (!io_s.ok()) {
+    return io_s;
+  }
+  if (!payload.empty()) {
+    io_s = file->Append(opts, payload);
+    if (!io_s.ok()) {
+      return io_s;
+    }
+  }
+  return file->Append(opts, Slice(trailer.data(), trailer.size()));
+}
+
+}  // namespace ROCKSDB_NAMESPACE

db/blob/blob_gen2_format.h

@@ -0,0 +1,92 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+
+#include "cache/cache_key.h"
+#include "rocksdb/compression_type.h"
+#include "rocksdb/io_status.h"
+#include "rocksdb/rocksdb_namespace.h"
+#include "rocksdb/status.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+struct ReadOptions;
+struct WriteOptions;
+class RandomAccessFileReader;
+class WritableFileWriter;
+class Slice;
+enum ChecksumType : char;
+
+// "SimpleGen2Blob" is the second-generation on-disk format for a blob payload
+// referenced by a same-file / context-relative blob index. A record is just the
+// payload bytes followed by a small block-style trailer:
+//
+//   <payload bytes> <1-byte compression marker> <4-byte checksum>
+//
+// The checksum is a builtin block checksum over (payload + compression marker),
+// context-modified by the record's absolute file offset (see
+// ChecksumModifierForContext / ComputeBuiltinChecksumWithLastByte in
+// table/format.h) so that an identical payload at a different offset gets a
+// different stored checksum. Payloads are currently uncompressed only.
+//
+// This format is deliberately decoupled from any particular file type: the same
+// record shape is read out of SST files (embedded blobs) and is intended to be
+// reused by other blob-bearing files. The only file-type-specific inputs are
+// the byte source (a RandomAccessFileReader) and the checksum context (checksum
+// type + base context checksum), both passed in by the caller.
+inline constexpr size_t kSimpleGen2BlobTrailerSize = 5;
+
+// Cache key for the SimpleGen2Blob record at byte `record_offset` within a file
+// whose base cache key is `base_cache_key`. Delegates to the shared
+// OffsetableCacheKey::WithOffsetForMinSizeRecord scheme -- the exact same
+// scheme block-based SST data blocks use via BlockBasedTable::GetCacheKey --
+// which is valid because a SimpleGen2Blob record always carries a >= 5-byte
+// trailer. As a result the cache key is collision-free with the file's data
+// blocks even when the blob cache and block cache are the same cache, with no
+// separately maintained algorithm to drift out of sync.
+inline CacheKey GetSimpleGen2BlobCacheKey(
+    const OffsetableCacheKey& base_cache_key, uint64_t record_offset) {
+  return base_cache_key.WithOffsetForMinSizeRecord(record_offset);
+}
+
+// Reads a SimpleGen2Blob record of `record_size` bytes (which must equal
+// `payload_size + kSimpleGen2BlobTrailerSize`) from `file` at `record_offset`
+// into `buf` (capacity >= record_size) and verifies its compression marker and,
+// when read_options.verify_checksums is set, its context-modified checksum.
+//
+// `checksum_type` and `base_context_checksum` are the file's checksum context
+// (e.g. from the SST footer). `expected_compression` is the compression type
+// the caller expects the record to carry (from the blob index); it must match
+// the record's marker, and currently must be kNoCompression.
+//
+// On success, buf[0, payload_size) holds the verified, uncompressed payload
+// (the trailer remains at the tail of `buf` and can be ignored).
+Status ReadAndVerifySimpleGen2BlobRecord(
+    const ReadOptions& read_options, RandomAccessFileReader* file,
+    uint64_t record_offset, size_t payload_size, size_t record_size,
+    ChecksumType checksum_type, uint32_t base_context_checksum,
+    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
+// builtin checksum), mirroring ReadAndVerifySimpleGen2BlobRecord so the on-disk
+// record format lives in one module.
+//
+// `checksum_type` and `base_context_checksum` are the file's checksum context
+// (e.g. from the SST footer). `compression` is the payload's compression type,
+// which currently must be kNoCompression.
+IOStatus WriteSimpleGen2BlobRecord(WritableFileWriter* file,
+                                   const WriteOptions& write_options,
+                                   ChecksumType checksum_type,
+                                   uint32_t base_context_checksum,
+                                   uint64_t record_offset, const Slice& payload,
+                                   CompressionType compression);
+
+}  // namespace ROCKSDB_NAMESPACE

db/blob/blob_index.h

@@ -7,6 +7,7 @@
 #include <sstream>
 #include <string>
 
+#include "db/blob/blob_constants.h"
 #include "rocksdb/compression_type.h"
 #include "util/coding.h"
 #include "util/compression.h"
@@ -57,6 +58,14 @@ class BlobIndex {
 
   bool IsInlined() const { return type_ == Type::kInlinedTTL; }
 
+  // True for a blob record stored in the same physical file as the table entry.
+  // Only embedded-blob SST reader/writer paths should interpret file number
+  // zero this way; generic blob metadata treats it as invalid.
+  bool IsSameFile() const {
+    return (type_ == Type::kBlob || type_ == Type::kBlobTTL) &&
+           file_number_ == kCurrentFileBlobIndexFileNumber;
+  }
+
   bool HasTTL() const {
     return type_ == Type::kInlinedTTL || type_ == Type::kBlobTTL;
   }
@@ -125,8 +134,13 @@ class BlobIndex {
     if (IsInlined()) {
       oss << "[inlined blob] value:" << value_.ToString(output_hex);
     } else {
-      oss << "[blob ref] file:" << file_number_ << " offset:" << offset_
+      oss << "[blob ref] file:";
-          << " size:" << size_
+      if (IsSameFile()) {
+        oss << "same";
+      } else {
+        oss << file_number_;
+      }
+      oss << " offset:" << offset_ << " size:" << size_
           << " compression: " << CompressionTypeToString(compression_);
     }
 

db/blob/blob_source.cc

@@ -12,7 +12,10 @@
 #include "cache/charged_cache.h"
 #include "db/blob/blob_contents.h"
 #include "db/blob/blob_file_reader.h"
+#include "db/blob/blob_gen2_format.h"
 #include "db/blob/blob_log_format.h"
+#include "file/random_access_file_reader.h"
+#include "memory/memory_allocator_impl.h"
 #include "monitoring/statistics_impl.h"
 #include "options/cf_options.h"
 #include "table/get_context.h"
@@ -305,6 +308,96 @@ Status BlobSource::GetBlob(const ReadOptions& read_options,
   return s;
 }
 
+Status BlobSource::GetSimpleGen2Blob(
+    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, PinnableSlice* value,
+    uint64_t* bytes_read) {
+  assert(value);
+  assert(file);
+
+  const uint64_t record_size = payload_size + kSimpleGen2BlobTrailerSize;
+
+  // The cache key is derived from the SimpleGen2Blob format (shared scheme with
+  // block-based SST blocks); see GetSimpleGen2BlobCacheKey.
+  const CacheKey cache_key =
+      GetSimpleGen2BlobCacheKey(base_cache_key, record_offset);
+
+  Status s;
+
+  CacheHandleGuard<BlobContents> blob_handle;
+
+  // First, try to get the blob from the cache.
+  if (blob_cache_) {
+    Slice key = cache_key.AsSlice();
+    s = GetBlobFromCache(key, &blob_handle);
+    if (s.ok()) {
+      PinCachedBlob(&blob_handle, value);
+
+      // For consistency, the on-disk record size is assigned to bytes_read on
+      // both cache hits and misses.
+      if (bytes_read) {
+        *bytes_read = record_size;
+      }
+      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 (or no cache configured). Read the record into a buffer
+  // allocated from the blob cache's memory allocator when we intend to insert
+  // it, exposing the uncompressed payload as BlobContents (the trailer just
+  // sits unused at the tail of the buffer).
+  MemoryAllocator* const allocator = (blob_cache_ && read_options.fill_cache)
+                                         ? blob_cache_.get()->memory_allocator()
+                                         : nullptr;
+
+  CacheAllocationPtr buf =
+      AllocateBlock(static_cast<size_t>(record_size), allocator);
+  s = ReadAndVerifySimpleGen2BlobRecord(
+      read_options, file, record_offset, static_cast<size_t>(payload_size),
+      static_cast<size_t>(record_size), checksum_type, base_context_checksum,
+      expected_compression, buf.get());
+  if (!s.ok()) {
+    return s;
+  }
+
+  std::unique_ptr<BlobContents> blob_contents(
+      new BlobContents(std::move(buf), static_cast<size_t>(payload_size)));
+
+  // Record the per-read statistics (mirrors BlobFileReader::GetBlob).
+  RecordTick(statistics_, BLOB_DB_BLOB_FILE_BYTES_READ, record_size);
+  PERF_COUNTER_ADD(blob_read_count, 1);
+  PERF_COUNTER_ADD(blob_read_byte, record_size);
+  if (bytes_read) {
+    *bytes_read = record_size;
+  }
+
+  if (blob_cache_ && read_options.fill_cache) {
+    // If filling cache is allowed and a cache is configured, try to put the
+    // blob into the cache.
+    Slice key = cache_key.AsSlice();
+    s = PutBlobIntoCache(key, &blob_contents, &blob_handle);
+    if (!s.ok()) {
+      return s;
+    }
+
+    PinCachedBlob(&blob_handle, value);
+  } else {
+    PinOwnedBlob(&blob_contents, value);
+  }
+
+  assert(s.ok());
+  return s;
+}
+
 void BlobSource::MultiGetBlob(const ReadOptions& read_options,
                               autovector<BlobFileReadRequests>& blob_reqs,
                               uint64_t* bytes_read) {

db/blob/blob_source.h

@@ -25,6 +25,8 @@ struct MutableCFOptions;
 class Status;
 class FilePrefetchBuffer;
 class Slice;
+class RandomAccessFileReader;
+enum ChecksumType : char;
 
 // BlobSource is a class that provides universal access to blobs, regardless of
 // whether they are in the blob cache, secondary cache, or (remote) storage.
@@ -59,6 +61,41 @@ class BlobSource {
                  FilePrefetchBuffer* prefetch_buffer, 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
+  // directly from a RandomAccessFileReader rather than from a traditional blob
+  // file.
+  //
+  // The cache key is derived from the SimpleGen2Blob format itself, not chosen
+  // by the caller: it is GetSimpleGen2BlobCacheKey(base_cache_key,
+  // record_offset), the same offset scheme block-based SST blocks use. The
+  // caller supplies only its file's `base_cache_key` (db_id / db_session_id /
+  // file_number). This keeps blob records collision-free with the file's data
+  // blocks even when the blob cache and block cache are the same cache.
+  //
+  // `file`, `record_offset`, `payload_size`, `checksum_type`,
+  // `base_context_checksum`, and `expected_compression` are the inputs to the
+  // SimpleGen2Blob reader used on a cache miss (see
+  // ReadAndVerifySimpleGen2BlobRecord). The on-disk record size (payload +
+  // trailer) is reported via `*bytes_read` (when non-null) and the
+  // BLOB_DB_BLOB_FILE_BYTES_READ / blob_read_byte counters, consistently on
+  // both cache hits and misses.
+  //
+  // On a cache hit, pins the cached value into `*value` (no copy). On a miss,
+  // reads + verifies the record into a cache-allocator buffer, records the
+  // per-read stats, inserts it into the cache (when configured and fill_cache
+  // is set), and pins it into `*value`. If blob_cache_ is not configured, the
+  // record is still read and read stats recorded, just without a cache
+  // lookup/insert.
+  Status GetSimpleGen2Blob(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,
+                           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"

db/blob/db_blob_index_test.cc

@@ -25,6 +25,9 @@
 #include "file/filename.h"
 #include "port/port.h"
 #include "port/stack_trace.h"
+#include "rocksdb/perf_context.h"
+#include "rocksdb/sst_file_writer.h"
+#include "rocksdb/table.h"
 #include "util/string_util.h"
 #include "utilities/merge_operators.h"
 
@@ -41,6 +44,18 @@ void CorruptPinnedBlobIndexOnCleanup(void* arg1, void* /*arg2*/) {
 
 }  // namespace
 
+TEST(BlobIndexTest, SameFileBlobIndex) {
+  std::string encoded;
+  BlobIndex::EncodeBlob(&encoded, kCurrentFileBlobIndexFileNumber,
+                        /*offset=*/123, /*size=*/456, kNoCompression);
+
+  BlobIndex blob_index;
+  ASSERT_OK(blob_index.DecodeFrom(encoded));
+  EXPECT_TRUE(blob_index.IsSameFile());
+  EXPECT_EQ(blob_index.file_number(), kCurrentFileBlobIndexFileNumber);
+  EXPECT_NE(blob_index.DebugString(false).find("file:same"), std::string::npos);
+}
+
 // kTypeBlobIndex is a value type used by BlobDB only. The base rocksdb
 // should accept the value type on write, and report not supported value
 // for reads, unless caller request for it explicitly. The base rocksdb
@@ -330,6 +345,232 @@ TEST_F(DBBlobIndexTest, ReadOnlyGetImplReturnsBlobIndexWhenRequested) {
   ASSERT_TRUE(is_blob_index);
 }
 
+// Regression test for a same-file (embedded) BlobIndex being exposed
+// unresolved through an iterator. With index_type=kBinarySearchWithFirstKey
+// and allow_unprepared_value (the default for DB iterators), the block-based
+// table iterator can sit in the is_at_first_key_from_index_ state and return
+// the raw kTypeBlobIndex internal key from the index, while value() resolves
+// the same-file blob to its plain payload. DBIter then reads the stale
+// kTypeBlobIndex type and routes the already-resolved plain value through the
+// blob-index path, corrupting/misreading it.
+//
+// Forcing one entry per data block makes every embedded blob the first key of
+// a block, reliably triggering the deferred-first-key state on both SeekToFirst
+// and forward block transitions.
+TEST_F(DBBlobIndexTest, EmbeddedBlobIteratorWithFirstKeyIndex) {
+  Options options = GetTestOptions();
+  options.create_if_missing = true;
+  BlockBasedTableOptions bbto;
+  bbto.index_type =
+      BlockBasedTableOptions::IndexType::kBinarySearchWithFirstKey;
+  // Soft limit of 1 byte starts a new data block after every entry.
+  bbto.block_size = 1;
+  options.table_factory.reset(NewBlockBasedTableFactory(bbto));
+  DestroyAndReopen(options);
+
+  // Build an embedded-blob SST whose large values are stored as same-file blob
+  // records (table entries become same-file BlobIndex references).
+  const std::string sst_path = dbname_ + "/embedded_first_key.sst";
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 8;
+
+  const std::string big1(1024, 'x');
+  const std::string big2(1024, 'y');
+  const std::string big3(1024, 'z');
+
+  SstFileWriter writer(EnvOptions(), options);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_path, embedded_blob_options));
+  ASSERT_OK(writer.Put("k1", big1));
+  ASSERT_OK(writer.Put("k2", big2));
+  ASSERT_OK(writer.Put("k3", big3));
+  ASSERT_OK(writer.Finish());
+
+  ASSERT_OK(db_->IngestExternalFile({sst_path}, IngestExternalFileOptions()));
+
+  // Point lookups resolve same-file blobs before exposing them, so Get works.
+  ASSERT_EQ(Get("k1"), big1);
+  ASSERT_EQ(Get("k2"), big2);
+  ASSERT_EQ(Get("k3"), big3);
+
+  // Iterator path: the bug manifests here.
+  std::unique_ptr<Iterator> iter(db_->NewIterator(ReadOptions()));
+  iter->SeekToFirst();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  EXPECT_EQ(iter->key(), "k1");
+  EXPECT_EQ(iter->value(), big1);
+
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  EXPECT_EQ(iter->key(), "k2");
+  EXPECT_EQ(iter->value(), big2);
+
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  EXPECT_EQ(iter->key(), "k3");
+  EXPECT_EQ(iter->value(), big3);
+
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_OK(iter->status());
+
+  // Seek directly to a block whose first key is an embedded blob.
+  iter->Seek("k2");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  EXPECT_EQ(iter->key(), "k2");
+  EXPECT_EQ(iter->value(), big2);
+}
+
+// When a blob cache is configured, same-file ("embedded") blob reads should go
+// through BlobSource: the first read misses + inserts + does a disk read, and
+// the second read of the same blob is served from the blob cache. This holds on
+// both the Get() and the iterator paths.
+TEST_F(DBBlobIndexTest, EmbeddedBlobBlobCacheHitMiss) {
+  Options options = GetTestOptions();
+  options.create_if_missing = true;
+  options.statistics = CreateDBStatistics();
+
+  LRUCacheOptions co;
+  co.capacity = 8 << 20;
+  options.blob_cache = NewLRUCache(co);
+
+  DestroyAndReopen(options);
+
+  const std::string sst_path = dbname_ + "/embedded_cache.sst";
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 8;
+
+  const std::string big1(1024, 'x');
+  const std::string big2(1024, 'y');
+
+  SstFileWriter writer(EnvOptions(), options);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_path, embedded_blob_options));
+  ASSERT_OK(writer.Put("k1", big1));
+  ASSERT_OK(writer.Put("k2", big2));
+  ASSERT_OK(writer.Finish());
+
+  ASSERT_OK(db_->IngestExternalFile({sst_path}, IngestExternalFileOptions()));
+
+  Statistics* const stats = options.statistics.get();
+
+  // First Get: blob cache miss -> disk read -> cache insert.
+  ASSERT_OK(stats->Reset());
+  get_perf_context()->Reset();
+  SetPerfLevel(kEnableCount);
+  ASSERT_EQ(Get("k1"), big1);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_MISS), 1);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_HIT), 0);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_ADD), 1);
+  EXPECT_GT(stats->getTickerCount(BLOB_DB_BLOB_FILE_BYTES_READ), 0);
+  EXPECT_EQ(get_perf_context()->blob_read_count, 1);
+
+  // Second Get of the same key: blob cache hit, no disk read, no insert.
+  ASSERT_OK(stats->Reset());
+  get_perf_context()->Reset();
+  ASSERT_EQ(Get("k1"), big1);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_HIT), 1);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_MISS), 0);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_ADD), 0);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_BLOB_FILE_BYTES_READ), 0);
+  EXPECT_EQ(get_perf_context()->blob_read_count, 0);
+  EXPECT_EQ(get_perf_context()->blob_cache_hit_count, 1);
+
+  // Iterator path: first read of a fresh key misses + inserts.
+  ASSERT_OK(stats->Reset());
+  {
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ReadOptions()));
+    iter->Seek("k2");
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    EXPECT_EQ(iter->value(), big2);
+  }
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_MISS), 1);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_ADD), 1);
+
+  // Iterator path: second read of the same key hits the blob cache.
+  ASSERT_OK(stats->Reset());
+  {
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ReadOptions()));
+    iter->Seek("k2");
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    EXPECT_EQ(iter->value(), big2);
+  }
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_HIT), 1);
+  EXPECT_EQ(stats->getTickerCount(BLOB_DB_CACHE_ADD), 0);
+
+  SetPerfLevel(kDisable);
+}
+
+// With the blob cache and block cache pointing at the same Cache, the SST-like
+// cache-key scheme (offset >> 2 on the SST's base key) keeps embedded blob
+// records disjoint from the SST's data blocks. Reading embedded blobs must not
+// alias or evict data blocks, and values must remain correct.
+TEST_F(DBBlobIndexTest, EmbeddedBlobSharedBlockAndBlobCache) {
+  Options options = GetTestOptions();
+  options.create_if_missing = true;
+  options.statistics = CreateDBStatistics();
+
+  LRUCacheOptions co;
+  co.capacity = 16 << 20;
+  auto shared_cache = NewLRUCache(co);
+  options.blob_cache = shared_cache;
+
+  BlockBasedTableOptions bbto;
+  bbto.block_cache = shared_cache;
+  bbto.cache_index_and_filter_blocks = true;
+  options.table_factory.reset(NewBlockBasedTableFactory(bbto));
+
+  DestroyAndReopen(options);
+
+  const std::string sst_path = dbname_ + "/embedded_shared.sst";
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 8;
+
+  const std::string big1(1024, 'x');
+  const std::string big2(1024, 'y');
+
+  SstFileWriter writer(EnvOptions(), options);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_path, embedded_blob_options));
+  ASSERT_OK(writer.Put("k1", big1));
+  ASSERT_OK(writer.Put("k2", big2));
+  ASSERT_OK(writer.Finish());
+
+  ASSERT_OK(db_->IngestExternalFile({sst_path}, IngestExternalFileOptions()));
+
+  Statistics* const stats = options.statistics.get();
+
+  // Values are correct with a shared cache, and re-reads hit both the block
+  // cache (data blocks) and the blob cache (embedded payloads) independently.
+  ASSERT_EQ(Get("k1"), big1);
+  ASSERT_EQ(Get("k2"), big2);
+  ASSERT_EQ(Get("k1"), big1);
+  ASSERT_EQ(Get("k2"), big2);
+  EXPECT_GT(stats->getTickerCount(BLOB_DB_CACHE_HIT), 0);
+  EXPECT_GT(stats->getTickerCount(BLOCK_CACHE_HIT), 0);
+
+  // Iterator yields correct values too.
+  std::unique_ptr<Iterator> iter(db_->NewIterator(ReadOptions()));
+  iter->SeekToFirst();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  EXPECT_EQ(iter->key(), "k1");
+  EXPECT_EQ(iter->value(), big1);
+
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  EXPECT_EQ(iter->key(), "k2");
+  EXPECT_EQ(iter->value(), big2);
+
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_OK(iter->status());
+}
+
 class PlainBlobValueFilterV3 : public CompactionFilter {
  public:
   PlainBlobValueFilterV3(std::atomic<int>* filter_call_count,

db/column_family.cc

@@ -678,6 +678,9 @@ ColumnFamilyData::ColumnFamilyData(
                           internal_stats_->GetBlobFileReadHist(), io_tracer));
     blob_source_.reset(new BlobSource(ioptions_, mutable_cf_options_, db_id,
                                       db_session_id, blob_file_cache_.get()));
+    // Let the table cache route same-file ("embedded") blob reads through the
+    // blob value cache + stats. Both objects share this CFD's lifetime.
+    table_cache_->SetBlobSource(blob_source_.get());
 
     if (ioptions_.compaction_style == kCompactionStyleLevel) {
       compaction_picker_.reset(

db/table_cache.cc

@@ -179,20 +179,22 @@ Status TableCache::GetTableReader(
     } else {
       expected_unique_id = kNullUniqueId64x2;  // null ID == no verification
     }
-    s = mutable_cf_options.table_factory->NewTableReader(
+    TableReaderOptions table_reader_options(
-        ro,
-        TableReaderOptions(
         ioptions_, mutable_cf_options.prefix_extractor,
         mutable_cf_options.compression_manager.get(), file_options,
-            internal_comparator,
+        internal_comparator, mutable_cf_options.block_protection_bytes_per_key,
-            mutable_cf_options.block_protection_bytes_per_key, skip_filters,
+        skip_filters, immortal_tables_, false /* force_direct_prefetch */,
-            immortal_tables_, false /* force_direct_prefetch */, level,
+        level, block_cache_tracer_, max_file_size_for_l0_meta_pin,
-            block_cache_tracer_, max_file_size_for_l0_meta_pin, db_session_id_,
+        db_session_id_, file_meta.fd.GetNumber(), expected_unique_id,
-            file_meta.fd.GetNumber(), expected_unique_id,
         file_meta.fd.largest_seqno, file_meta.tail_size,
         file_meta.user_defined_timestamps_persisted,
-            avoid_shared_metadata_cache),
+        avoid_shared_metadata_cache);
-        std::move(file_reader), file_meta.fd.GetFileSize(), table_reader,
+    // Route same-file ("embedded") blob reads through the CFD's BlobSource for
+    // caching + stats. nullptr in non-DB contexts (e.g. repair).
+    table_reader_options.blob_source = blob_source_;
+    s = mutable_cf_options.table_factory->NewTableReader(
+        ro, table_reader_options, std::move(file_reader),
+        file_meta.fd.GetFileSize(), table_reader,
         prefetch_index_and_filter_in_cache);
     TEST_SYNC_POINT("TableCache::GetTableReader:0");
   }

db/table_cache.h

@@ -34,6 +34,7 @@ class Arena;
 struct FileDescriptor;
 class GetContext;
 class HistogramImpl;
+class BlobSource;
 
 struct TableCacheOpenOptions {
   // Open a new TableReader owned by the returned iterator instead of reusing
@@ -69,6 +70,15 @@ class TableCache {
              const std::string& db_session_id, bool fast_sst_open = false);
   ~TableCache();
 
+  // Sets the BlobSource used to route same-file ("embedded") blob reads of SSTs
+  // opened through this cache via the blob value cache + BLOB_DB_* statistics.
+  // Both the TableCache and the BlobSource are owned by the same
+  // ColumnFamilyData and share its lifetime, so the raw pointer is safe.
+  // Called once at CFD setup, after the BlobSource is constructed. Left unset
+  // (nullptr) in non-DB contexts (e.g. repair), where embedded reads fall back
+  // to a direct (uncached) read.
+  void SetBlobSource(BlobSource* blob_source) { blob_source_ = blob_source; }
+
   // Cache interface for table cache
   using CacheInterface =
       BasicTypedCacheInterface<TableReader, CacheEntryRole::kMisc>;
@@ -353,6 +363,9 @@ class TableCache {
   Striped<CacheAlignedWrapper<port::Mutex>> loader_mutex_;
   std::shared_ptr<IOTracer> io_tracer_;
   std::string db_session_id_;
+  // Owned by the same ColumnFamilyData; see SetBlobSource(). nullptr in non-DB
+  // contexts (e.g. repair).
+  BlobSource* blob_source_ = nullptr;
 };
 
 }  // namespace ROCKSDB_NAMESPACE

db_stress_tool/db_stress_common.h

@@ -231,6 +231,7 @@ DECLARE_int32(ingest_external_file_one_in);
 DECLARE_int32(ingest_external_file_width);
 DECLARE_int32(ingest_external_file_prepare_commit_one_in);
 DECLARE_int32(ingest_external_file_use_file_info_one_in);
+DECLARE_bool(ingest_external_file_with_embedded_blobs);
 DECLARE_int32(compact_files_one_in);
 DECLARE_int32(compact_range_one_in);
 DECLARE_int32(promote_l0_one_in);

db_stress_tool/db_stress_gflags.cc

@@ -866,6 +866,14 @@ DEFINE_int32(ingest_external_file_use_file_info_one_in, 0,
              "SstFileWriter::Finish) once every N ingestions on average, so "
              "ingestion skips re-opening and scanning the files.");
 
+DEFINE_bool(
+    ingest_external_file_with_embedded_blobs, false,
+    "If true, external files for ingestion are written with embedded "
+    "blobs via SstFileWriter::OpenWithEmbeddedBlobs(). min_blob_size is "
+    "set so that only the largest generated values are embedded. "
+    "Requires block-based table format_version >= 7 and "
+    "ingest_external_file_one_in > 0.");
+
 DEFINE_int32(compact_files_one_in, 0,
              "If non-zero, then CompactFiles() will be called once for every N "
              "operations on average.  0 indicates CompactFiles() is disabled.");

db_stress_tool/no_batched_ops_stress.cc

@@ -2436,12 +2436,28 @@ class NonBatchedOpsStressTest : public StressTest {
     }
 
     std::vector<ExternalSstFileInfo> file_infos(data_file_count);
+    // Embedded blobs are only supported by block-based table format_version >=
+    // 7 (the default db_stress table factory is block-based).
+    const bool use_embedded_blobs =
+        FLAGS_ingest_external_file_with_embedded_blobs &&
+        FLAGS_format_version >= 7;
+    // Set the embedded-blob threshold so that only the largest values generated
+    // by GenerateValue() (size kRandomValueMaxFactor * value_size_mult) are
+    // written as same-file blob records; smaller values stay inline.
+    SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+    embedded_blob_options.min_blob_size =
+        static_cast<uint64_t>(kRandomValueMaxFactor) * FLAGS_value_size_mult;
     std::deque<SstFileWriter> sst_file_writers;
     for (size_t file_idx = 0; s.ok() && file_idx < data_file_count;
          ++file_idx) {
       sst_file_writers.emplace_back(EnvOptions(options_), options_);
+      if (use_embedded_blobs) {
+        s = sst_file_writers.back().OpenWithEmbeddedBlobs(
+            data_filenames[file_idx], embedded_blob_options);
+      } else {
         s = sst_file_writers.back().Open(data_filenames[file_idx]);
       }
+    }
     SstFileWriter standalone_rangedel_sst_file_writer(EnvOptions(options_),
                                                       options_);
     if (s.ok() && test_standalone_range_deletion) {

include/rocksdb/sst_file_writer.h

@@ -19,6 +19,33 @@ namespace ROCKSDB_NAMESPACE {
 
 class Comparator;
 
+// Options for SstFileWriter::OpenWithEmbeddedBlobs().
+//
+// Embedded blobs are large values stored in a leading segment of the same
+// block-based SST file. The table entry stores a same-file BlobIndex reference
+// instead of the value bytes. Readers that understand the embedded blob
+// metadata return the original value to callers.
+//
+// This mode is EXPERIMENTAL. It is intended for block-based table files with
+// format_version >= 7 only. Eligible blob records are written before Finish(),
+// so Put() and PutEntity() can surface write or checksum errors that would
+// normally be deferred until Finish().
+struct SstFileWriterEmbeddedBlobOptions {
+  // Minimum value size, in bytes, to store as an embedded blob.
+  //
+  // Put() values with size >= min_blob_size are embedded. For PutEntity(), each
+  // wide-column value is considered independently. Merge operands, deletions,
+  // and range deletions are not embedded. The default embeds eligible values
+  // of at least 2KB.
+  uint64_t min_blob_size = 2048;
+
+  // PLACEHOLDER: Not yet implemented (ignored).
+  CompressionType compression_type = kDisableCompressionOption;
+
+  // PLACEHOLDER: Not yet implemented (ignored).
+  CompressionOptions compression_options;
+};
+
 // ExternalSstFileInfo include information about sst files created
 // using SstFileWriter.
 struct ExternalSstFileInfo {
@@ -107,6 +134,20 @@ class SstFileWriter {
   Status Open(const std::string& file_path,
               Temperature temp = Temperature::kUnknown);
 
+  // EXPERIMENTAL: opens an SST writer mode that stores eligible blob payloads
+  // inside the same block-based SST file and stores same-file BlobIndex
+  // references in table entries. This is a special-purpose mode that doesn't
+  // offer write-amp improvements under compaction like standard blob support
+  // does. Blob records are written inline (interleaved with data blocks) as
+  // eligible values are added.
+  //
+  // Requires the writer to use block-based table format_version >= 7. The
+  // resulting SST can be consumed by RocksDB >= 11.6.0.
+  Status OpenWithEmbeddedBlobs(
+      const std::string& file_path,
+      const SstFileWriterEmbeddedBlobOptions& embedded_blob_options,
+      Temperature temp = Temperature::kUnknown);
+
   // Add a Put key with value to currently opened file
   // REQUIRES: user_key is after any previously added point (Put/Merge/Delete)
   //           key according to the comparator.

include/rocksdb/status.h

@@ -66,6 +66,15 @@ class Status {
   // error swallowing occurs.
   inline void PermitUncheckedError() const { MarkChecked(); }
 
+  // Like PermitUncheckedError(), but documents that this Status is expected to
+  // be OK and is permitted to go unread (e.g. a default-constructed member that
+  // may legitimately never be checked when a feature is unused). Asserts the
+  // status is OK in debug builds.
+  inline void PermitUncheckedOk() const {
+    assert(code_ == kOk);
+    MarkChecked();
+  }
+
   inline void MustCheck() const {
 #ifdef ROCKSDB_ASSERT_STATUS_CHECKED
     checked_ = false;

src.mk

@@ -25,6 +25,7 @@ LIB_SOURCES =                                                   \
   db/blob/blob_file_meta.cc                                     \
   db/blob/blob_file_reader.cc                                   \
   db/blob/blob_garbage_meter.cc                                 \
+  db/blob/blob_gen2_format.cc                                   \
   db/blob/blob_log_format.cc                                    \
   db/blob/blob_log_sequential_reader.cc                         \
   db/blob/blob_log_writer.cc                                    \

table/block_based/block.cc

@@ -636,6 +636,7 @@ bool DataBlockIter::ParseNextDataKey(bool* is_shared) {
              value_type == ValueType::kTypeDeletion ||
              value_type == ValueType::kTypeDeletionWithTimestamp ||
              value_type == ValueType::kTypeRangeDeletion ||
+             value_type == ValueType::kTypeBlobIndex ||
              value_type == ValueType::kTypeWideColumnEntity);
       assert(seqno == 0);
     }
@@ -699,6 +700,7 @@ void IndexBlockIter::DecodeCurrentValue(bool is_shared) {
            value_type == ValueType::kTypeMerge ||
            value_type == ValueType::kTypeDeletion ||
            value_type == ValueType::kTypeRangeDeletion ||
+           value_type == ValueType::kTypeBlobIndex ||
            value_type == ValueType::kTypeWideColumnEntity);
 
     first_internal_key.UpdateInternalKey(global_seqno_state_->global_seqno,

table/block_based/block_based_table_builder.cc

@@ -12,6 +12,8 @@
 #include <atomic>
 #include <cassert>
 #include <cstdio>
+#include <cstring>
+#include <deque>
 #include <list>
 #include <map>
 #include <memory>
@@ -25,9 +27,14 @@
 #include "cache/cache_helpers.h"
 #include "cache/cache_key.h"
 #include "cache/cache_reservation_manager.h"
+#include "db/blob/blob_constants.h"
+#include "db/blob/blob_gen2_format.h"
+#include "db/blob/blob_index.h"
 #include "db/dbformat.h"
+#include "db/wide/wide_column_serialization.h"
 #include "index_builder.h"
 #include "logging/logging.h"
+#include "memory/arena.h"
 #include "memory/memory_allocator_impl.h"
 #include "options/options_helper.h"
 #include "rocksdb/cache.h"
@@ -36,8 +43,10 @@
 #include "rocksdb/filter_policy.h"
 #include "rocksdb/flush_block_policy.h"
 #include "rocksdb/merge_operator.h"
+#include "rocksdb/sst_file_writer.h"
 #include "rocksdb/table.h"
 #include "rocksdb/types.h"
+#include "rocksdb/wide_columns.h"
 #include "table/block_based/block.h"
 #include "table/block_based/block_based_table_factory.h"
 #include "table/block_based/block_based_table_reader.h"
@@ -50,6 +59,7 @@
 #include "table/format.h"
 #include "table/meta_blocks.h"
 #include "table/table_builder.h"
+#include "test_util/sync_point.h"
 #include "util/bit_fields.h"
 #include "util/coding.h"
 #include "util/compression.h"
@@ -962,6 +972,28 @@ struct BlockBasedTableBuilder::Rep {
   // This is used for logging compaction stats.
   uint64_t pre_compression_size = 0;
 
+  // Embedded-blob SST support. When `embedded_blob_options` is set the builder
+  // is in embedded mode: index value delta encoding is disabled and eligible
+  // large values are written inline as same-file blob records as values are
+  // added (so they may be interleaved with data blocks), with table entries
+  // rewritten to same-file BlobIndex references. This is a private owned copy:
+  // the caller (e.g. SstFileWriter::OpenWithEmbeddedBlobs) may free its own
+  // copy as soon as the builder is constructed, so we must not alias it.
+  std::unique_ptr<const EmbeddedBlobSstBuilderOptions> embedded_blob_options;
+
+  // Mutable state for embedded blob writing, lazily allocated when the first
+  // blob record is written. Its presence is the signal that the file contains
+  // embedded blobs (driving the presence property). Holds the diagnostic
+  // counters plus scratch buffers reused across Add() calls to avoid per-entry
+  // allocation.
+  struct EmbeddedBlobState {
+    EmbeddedBlobStats stats;
+    std::string key_buf;
+    std::string blob_index_buf;
+    std::string entity_buf;
+  };
+  std::unique_ptr<EmbeddedBlobState> embedded_blob_state;
+
   // See class Footer
   uint32_t base_context_checksum;
 
@@ -1084,8 +1116,10 @@ struct BlockBasedTableBuilder::Rep {
         compression_parallel_threads(tbo.compression_opts.parallel_threads),
         max_compressed_bytes_per_kb(
             tbo.compression_opts.max_compressed_bytes_per_kb),
-        use_delta_encoding_for_index_values(table_opt.format_version >= 4 &&
+        use_delta_encoding_for_index_values(
-                                            !table_opt.block_align),
+            table_opt.format_version >= 4 && !table_opt.block_align &&
+            /* surely no embedded blobs */ tbo.embedded_blob_options ==
+                nullptr),
         reason(tbo.reason),
         target_file_size_is_upper_bound(
             tbo.moptions.target_file_size_is_upper_bound),
@@ -1103,7 +1137,12 @@ struct BlockBasedTableBuilder::Rep {
                            BlockBasedTableOptions::kBinarySearchWithFirstKey,
                        table_options.block_restart_interval,
                        table_options.index_block_restart_interval),
-        tail_size(0) {
+        tail_size(0),
+        embedded_blob_options(
+            tbo.embedded_blob_options
+                ? std::make_unique<EmbeddedBlobSstBuilderOptions>(
+                      *tbo.embedded_blob_options)
+                : nullptr) {
     FilterBuildingContext filter_context(table_options);
 
     filter_context.info_log = ioptions.logger;
@@ -1133,25 +1172,26 @@ struct BlockBasedTableBuilder::Rep {
         std::to_string(static_cast<int>(tbo.compression_type)));
     props.compression_options.append("; ");
 
-    auto* mgr = tbo.moptions.compression_manager.get();
+    auto* compression_manager = tbo.moptions.compression_manager.get();
-    if (mgr == nullptr) {
+    if (compression_manager == nullptr) {
       uses_explicit_compression_manager = false;
-      mgr = GetBuiltinV2CompressionManager().get();
+      compression_manager = GetBuiltinV2CompressionManager().get();
     } else {
       uses_explicit_compression_manager = true;
 
       // Stuff some extra debugging info as extra pseudo-options. Using
       // underscore prefix to indicate they are special.
       props.compression_options.append("_compression_manager=");
-      props.compression_options.append(mgr->GetId());
+      props.compression_options.append(compression_manager->GetId());
       props.compression_options.append("; ");
     }
+    assert(compression_manager);
 
     // Sanitize to only allowing compression when it saves space.
     max_compressed_bytes_per_kb =
         std::min(int{1023}, tbo.compression_opts.max_compressed_bytes_per_kb);
 
-    basic_compressor = mgr->GetCompressorForSST(
+    basic_compressor = compression_manager->GetCompressorForSST(
         filter_context, tbo.compression_opts, tbo.compression_type);
     if (basic_compressor) {
       if (table_options.enable_index_compression) {
@@ -1201,7 +1241,7 @@ struct BlockBasedTableBuilder::Rep {
       basic_decompressor = basic_compressor->GetOptimizedDecompressor();
       if (basic_decompressor == nullptr) {
         // Optimized version not available
-        basic_decompressor = mgr->GetDecompressor();
+        basic_decompressor = compression_manager->GetDecompressor();
       }
       create_context.decompressor = basic_decompressor.get();
 
@@ -1232,7 +1272,10 @@ struct BlockBasedTableBuilder::Rep {
     // Hard structural constraints override any recommendation
     if ((table_opt.partition_filters &&
          !table_opt.decouple_partitioned_filters) ||
-        table_options.user_defined_index_factory) {
+        table_options.user_defined_index_factory || embedded_blob_options) {
+      // Embedded-blob SSTs write blob records inline on the emit thread, so
+      // they require single-threaded writes for correct, race-free ordering of
+      // blob appends and data-block writes.
       compression_parallel_threads = 1;
     }
 
@@ -1390,7 +1433,7 @@ struct BlockBasedTableBuilder::Rep {
     props.key_largest_seqno = 0;
     // Default is UINT64_MAX for unknown.
     props.key_smallest_seqno = UINT64_MAX;
-    PrePopulateCompressionProperties(mgr);
+    PrePopulateCompressionProperties(compression_manager);
 
     if (FormatVersionUsesContextChecksum(table_options.format_version)) {
       // Must be non-zero and semi- or quasi-random
@@ -1431,14 +1474,15 @@ struct BlockBasedTableBuilder::Rep {
   Rep(const Rep&) = delete;
   Rep& operator=(const Rep&) = delete;
 
-  void PrePopulateCompressionProperties(UnownedPtr<CompressionManager> mgr) {
+  void PrePopulateCompressionProperties(
+      UnownedPtr<CompressionManager> compression_manager) {
     if (FormatVersionUsesCompressionManagerName(table_options.format_version)) {
-      assert(mgr);
+      assert(compression_manager);
       // Use newer compression_name property
       props.compression_name.reserve(32);
       // If compression is disabled, use empty manager name
       if (basic_compressor) {
-        props.compression_name.append(mgr->CompatibilityName());
+        props.compression_name.append(compression_manager->CompatibilityName());
       }
       props.compression_name.push_back(';');
       // Rest of property to be filled out at the end of building the file
@@ -1447,7 +1491,7 @@ struct BlockBasedTableBuilder::Rep {
       // building the file. Not compatible with compression managers using
       // custom algorithms / compression types.
       assert(
-          Slice(mgr->CompatibilityName())
+          Slice(compression_manager->CompatibilityName())
               .compare(GetBuiltinV2CompressionManager()->CompatibilityName()) ==
           0);
     }
@@ -1578,10 +1622,25 @@ void BlockBasedTableBuilder::Add(const Slice& ikey, const Slice& value) {
   UnPackSequenceAndType(ExtractInternalKeyFooter(ikey), &seq, &value_type);
   r->props.key_largest_seqno = std::max(r->props.key_largest_seqno, seq);
   r->props.key_smallest_seqno = std::min(r->props.key_smallest_seqno, seq);
+
+  // For embedded-blob SSTs, large value payloads are written inline as
+  // same-file blob records and the table entry is rewritten to reference them
+  // before the normal flush/add logic. entry_ikey / entry_value alias the
+  // inputs unless rewriting occurs.
+  Slice entry_ikey = ikey;
+  Slice entry_value = value;
+
   if (IsValueType(value_type)) {
+    if (r->embedded_blob_options) {
+      if (!MaybeExtractEmbeddedBlobs(seq, value_type, &entry_ikey,
+                                     &entry_value)) {
+        return;  // failed status already recorded
+      }
+    }
 #ifndef NDEBUG
     if (r->props.num_entries > r->props.num_range_deletions) {
-      assert(r->internal_comparator.Compare(ikey, Slice(r->last_ikey)) > 0);
+      assert(r->internal_comparator.Compare(entry_ikey, Slice(r->last_ikey)) >
+             0);
     }
     bool skip = false;
     TEST_SYNC_POINT_CALLBACK("BlockBasedTableBuilder::Add::skip", (void*)&skip);
@@ -1590,10 +1649,10 @@ void BlockBasedTableBuilder::Add(const Slice& ikey, const Slice& value) {
     }
 #endif  // !NDEBUG
 
-    auto should_flush = r->flush_block_policy->Update(ikey, value);
+    auto should_flush = r->flush_block_policy->Update(entry_ikey, entry_value);
     if (should_flush) {
       assert(!r->data_block.empty());
-      Flush(/*first_key_in_next_block=*/&ikey);
+      Flush(/*first_key_in_next_block=*/&entry_ikey);
     }
 
     // Note: PartitionedFilterBlockBuilder with
@@ -1603,7 +1662,7 @@ void BlockBasedTableBuilder::Add(const Slice& ikey, const Slice& value) {
     if (r->state == Rep::State::kUnbuffered) {
       if (r->filter_builder != nullptr) {
         r->filter_builder->AddWithPrevKey(
-            ExtractUserKeyAndStripTimestamp(ikey, r->ts_sz),
+            ExtractUserKeyAndStripTimestamp(entry_ikey, r->ts_sz),
             r->last_ikey.empty()
                 ? Slice{}
                 : ExtractUserKeyAndStripTimestamp(r->last_ikey, r->ts_sz));
@@ -1611,17 +1670,17 @@ void BlockBasedTableBuilder::Add(const Slice& ikey, const Slice& value) {
     }
 
     // NOTE: WriteBatch guarantees keys < 4GB; value size checked above
-    r->data_block.AddWithLastKey(ikey, value, r->last_ikey);
+    r->data_block.AddWithLastKey(entry_ikey, entry_value, r->last_ikey);
-    r->last_ikey.assign(ikey.data(), ikey.size());
+    r->last_ikey.assign(entry_ikey.data(), entry_ikey.size());
     assert(!r->last_ikey.empty());
     if (r->state == Rep::State::kBuffered) {
       // Buffered keys will be replayed from data_block_buffers during
       // `Finish()` once compression dictionary has been finalized.
     } else {
-      r->index_builder->OnKeyAdded(ikey, value);
+      r->index_builder->OnKeyAdded(entry_ikey, entry_value);
     }
     // TODO offset passed in is not accurate for parallel compression case
-    NotifyCollectTableCollectorsOnAdd(ikey, value, r->get_offset(),
+    NotifyCollectTableCollectorsOnAdd(entry_ikey, entry_value, r->get_offset(),
                                       r->table_properties_collectors,
                                       r->ioptions.logger);
 
@@ -1650,11 +1709,11 @@ void BlockBasedTableBuilder::Add(const Slice& ikey, const Slice& value) {
   }
 
   r->props.num_entries++;
-  r->props.raw_key_size += ikey.size();
+  r->props.raw_key_size += entry_ikey.size();
   if (!r->persist_user_defined_timestamps) {
     r->props.raw_key_size -= r->ts_sz;
   }
-  r->props.raw_value_size += value.size();
+  r->props.raw_value_size += entry_value.size();
   if (value_type == kTypeDeletion || value_type == kTypeSingleDeletion ||
       value_type == kTypeDeletionWithTimestamp) {
     r->props.num_deletions++;
@@ -2335,6 +2394,143 @@ IOStatus BlockBasedTableBuilder::io_status() const {
 
 bool BlockBasedTableBuilder::ok() const { return rep_->StatusOk(); }
 
+Status BlockBasedTableBuilder::WriteEmbeddedBlobRecord(const Slice& payload) {
+  Rep* r = rep_.get();
+  assert(r->embedded_blob_options);
+  // Inline blob writing requires single-threaded writes so the running table
+  // offset matches the file position (enforced in the Rep constructor).
+  assert(!r->IsParallelCompressionActive());
+
+  // Embedded blob payloads share the 32-bit value-size limit with the rest of
+  // the table format; enforcing it here keeps record sizes in range and gives
+  // an early, cheap corruption check.
+  if (payload.size() > std::numeric_limits<uint32_t>::max()) {
+    return Status::InvalidArgument("Embedded blob value is too large");
+  }
+
+  if (!r->embedded_blob_state) {
+    r->embedded_blob_state = std::make_unique<Rep::EmbeddedBlobState>();
+  }
+
+  // The builder's running table offset equals the current file size, so the
+  // record begins exactly here.
+  const uint64_t offset = r->get_offset();
+  // Placeholder for future embedded blob compression support.
+  const CompressionType compression = kNoCompression;
+  IOStatus io_s = WriteSimpleGen2BlobRecord(
+      r->file, r->write_options, r->table_options.checksum,
+      r->base_context_checksum, offset, payload, compression);
+  if (!io_s.ok()) {
+    r->SetIOStatus(io_s);
+    return io_s;
+  }
+
+  const uint64_t record_size = payload.size() + kSimpleGen2BlobTrailerSize;
+  r->set_offset(offset + record_size);
+  r->embedded_blob_state->stats.blob_count++;
+  r->embedded_blob_state->stats.payload_bytes += payload.size();
+
+  BlobIndex::EncodeBlob(&r->embedded_blob_state->blob_index_buf,
+                        kCurrentFileBlobIndexFileNumber, offset, payload.size(),
+                        compression);
+  return Status::OK();
+}
+
+Status BlockBasedTableBuilder::BuildEmbeddedWideColumnEntity(const Slice& value,
+                                                             bool* rewritten) {
+  Rep* r = rep_.get();
+  assert(rewritten != nullptr);
+  *rewritten = false;
+
+  WideColumns columns;
+  Slice input = value;
+  Status s = WideColumnSerialization::Deserialize(input, columns);
+  if (!s.ok()) {
+    return s;
+  }
+
+  std::vector<std::pair<size_t, BlobIndex>> blob_columns;
+  blob_columns.reserve(columns.size());
+
+  const uint64_t min_blob_size = r->embedded_blob_options->min_blob_size;
+  for (size_t column_idx = 0; column_idx < columns.size(); ++column_idx) {
+    const Slice& column_value = columns[column_idx].value();
+    if (column_value.size() < min_blob_size) {
+      continue;
+    }
+    s = WriteEmbeddedBlobRecord(column_value);
+    if (!s.ok()) {
+      return s;
+    }
+    BlobIndex blob_index;
+    s = blob_index.DecodeFrom(r->embedded_blob_state->blob_index_buf);
+    if (!s.ok()) {
+      return s;
+    }
+    blob_columns.emplace_back(column_idx, blob_index);
+  }
+
+  if (blob_columns.empty()) {
+    return Status::OK();
+  }
+
+  // At least one record was written above, so the state exists.
+  r->embedded_blob_state->entity_buf.clear();
+  s = WideColumnSerialization::SerializeV2(columns, blob_columns,
+                                           r->embedded_blob_state->entity_buf);
+  if (!s.ok()) {
+    return s;
+  }
+  *rewritten = true;
+  return Status::OK();
+}
+
+bool BlockBasedTableBuilder::MaybeExtractEmbeddedBlobs(SequenceNumber seq,
+                                                       ValueType value_type,
+                                                       Slice* key,
+                                                       Slice* value) {
+  Rep* r = rep_.get();
+  assert(r->embedded_blob_options);
+  assert(key != nullptr);
+  assert(value != nullptr);
+
+  if (value_type == kTypeValue) {
+    if (value->size() < r->embedded_blob_options->min_blob_size) {
+      return true;  // leave small values inline
+    }
+    Status s = WriteEmbeddedBlobRecord(*value);
+    if (!s.ok()) {
+      r->SetStatus(std::move(s));
+      return false;
+    }
+    // Rewrite the internal key footer type to kTypeBlobIndex in scratch (the
+    // input key is not mutated in place).
+    std::string& key_buf = r->embedded_blob_state->key_buf;
+    key_buf.assign(key->data(), key->size());
+    assert(key_buf.size() >= kNumInternalBytes);
+    EncodeFixed64(&key_buf[key_buf.size() - kNumInternalBytes],
+                  PackSequenceAndType(seq, kTypeBlobIndex));
+    *key = Slice(key_buf);
+    *value = Slice(r->embedded_blob_state->blob_index_buf);
+    return true;
+  }
+
+  if (value_type == kTypeWideColumnEntity) {
+    bool rewritten = false;
+    Status s = BuildEmbeddedWideColumnEntity(*value, &rewritten);
+    if (!s.ok()) {
+      r->SetStatus(std::move(s));
+      return false;
+    }
+    if (rewritten) {
+      *value = Slice(r->embedded_blob_state->entity_buf);
+    }
+    return true;
+  }
+
+  return true;
+}
+
 Status BlockBasedTableBuilder::InsertBlockInCacheHelper(
     const Slice& block_contents, const BlockHandle* handle,
     BlockType block_type) {
@@ -2582,6 +2778,13 @@ void BlockBasedTableBuilder::WritePropertiesBlock(
     rep_->props.user_defined_timestamps_persisted =
         rep_->persist_user_defined_timestamps;
 
+    if (rep_->embedded_blob_state) {
+      std::string encoded_stats;
+      EncodeEmbeddedBlobStats(rep_->embedded_blob_state->stats, &encoded_stats);
+      rep_->props.user_collected_properties[kEmbeddedBlobSstStatsPropertyName] =
+          std::move(encoded_stats);
+    }
+
     rep_->props.num_data_blocks_compression_rejected =
         rep_->num_data_blocks_compression_rejected.LoadRelaxed();
     rep_->props.num_data_blocks_compression_bypassed =

table/block_based/block_based_table_builder.h

@@ -12,6 +12,7 @@
 
 #include <array>
 #include <limits>
+#include <memory>
 #include <string>
 #include <utility>
 #include <vector>
@@ -179,6 +180,31 @@ class BlockBasedTableBuilder : public TableBuilder {
   void WriteFooter(BlockHandle& metaindex_block_handle,
                    BlockHandle& index_block_handle);
 
+  // Embedded-blob SST support. These are only exercised when the builder is in
+  // embedded mode (rep_->embedded_blob_options is set), in which delta encoding
+  // of index values is disabled so blob records can be written inline as values
+  // are added (possibly interleaved with data blocks).
+
+  // For an embedded-mode value entry, possibly extracts large value payload(s)
+  // into inline same-file blob records and rewrites *key / *value to reference
+  // them (a kTypeBlobIndex entry for whole values, or a rebuilt wide-column
+  // entity with per-column BlobIndex refs). On no-op, *key / *value are left
+  // unchanged. Returns false and records a failed status on error.
+  bool MaybeExtractEmbeddedBlobs(SequenceNumber seq, ValueType value_type,
+                                 Slice* key, Slice* value);
+
+  // Rewrites eligible wide-column values as inline same-file blob records,
+  // building the rebuilt entity into rep_->embedded_blob_state->entity_buf.
+  // Sets *rewritten when at least one column was extracted (otherwise the
+  // original value should be used as-is).
+  Status BuildEmbeddedWideColumnEntity(const Slice& value, bool* rewritten);
+
+  // Appends one uncompressed same-file blob record for `payload` at the current
+  // table offset, advances the offset, lazily allocates and updates the
+  // embedded-blob state (counters), and encodes a same-file BlobIndex into
+  // rep_->embedded_blob_state->blob_index_buf.
+  Status WriteEmbeddedBlobRecord(const Slice& payload);
+
   struct Rep;
   class BlockBasedTablePropertiesCollectorFactory;
   class BlockBasedTablePropertiesCollector;

table/block_based/block_based_table_factory.cc

@@ -626,12 +626,15 @@ Status BlockBasedTableFactory::NewTableReader(
       table_reader_options.cur_db_session_id, table_reader_options.cur_file_num,
       table_reader_options.unique_id,
       table_reader_options.user_defined_timestamps_persisted,
-      table_reader_options.avoid_shared_metadata_cache);
+      table_reader_options.avoid_shared_metadata_cache,
+      table_reader_options.blob_source);
 }
 
 TableBuilder* BlockBasedTableFactory::NewTableBuilder(
     const TableBuilderOptions& table_builder_options,
     WritableFileWriter* file) const {
+  // BlockBasedTableBuilder self-detects embedded-blob mode from
+  // table_builder_options.embedded_blob_options.
   return new BlockBasedTableBuilder(table_options_, table_builder_options,
                                     file);
 }

table/block_based/block_based_table_reader.cc

@@ -14,6 +14,7 @@
 #include <cstdint>
 #include <limits>
 #include <memory>
+#include <optional>
 #include <string>
 #include <unordered_set>
 #include <utility>
@@ -22,13 +23,18 @@
 #include "block_cache.h"
 #include "cache/cache_entry_roles.h"
 #include "cache/cache_key.h"
+#include "db/blob/blob_gen2_format.h"
+#include "db/blob/blob_index.h"
+#include "db/blob/blob_source.h"
 #include "db/compaction/compaction_picker.h"
 #include "db/dbformat.h"
 #include "db/pinned_iterators_manager.h"
+#include "db/wide/wide_column_serialization.h"
 #include "file/file_prefetch_buffer.h"
 #include "file/file_util.h"
 #include "file/random_access_file_reader.h"
 #include "logging/logging.h"
+#include "memory/memory_allocator_impl.h"
 #include "monitoring/perf_context_imp.h"
 #include "parsed_full_filter_block.h"
 #include "port/lang.h"
@@ -53,6 +59,7 @@
 #include "table/block_based/block_based_table_iterator.h"
 #include "table/block_based/block_prefix_index.h"
 #include "table/block_based/block_type.h"
+#include "table/block_based/embedded_blob_resolving_iterator.h"
 #include "table/block_based/filter_block.h"
 #include "table/block_based/filter_policy_internal.h"
 #include "table/block_based/full_filter_block.h"
@@ -61,6 +68,7 @@
 #include "table/block_based/partitioned_index_reader.h"
 #include "table/block_based/user_defined_index_wrapper.h"
 #include "table/block_fetcher.h"
+#include "table/embedded_blob_sst.h"
 #include "table/format.h"
 #include "table/get_context.h"
 #include "table/internal_iterator.h"
@@ -85,6 +93,12 @@ CacheAllocationPtr CopyBufferToHeap(MemoryAllocator* allocator, Slice& buf) {
   memcpy(heap_buf.get(), buf.data(), buf.size());
   return heap_buf;
 }
+
+// Cleanup function for a Cleanable/PinnableSlice that owns a `new char[]`
+// buffer (used to pin embedded blob payloads without copying).
+void DeleteCharArray(void* arg1, void* /*arg2*/) {
+  delete[] static_cast<char*>(arg1);
+}
 }  // namespace
 
 Status AllocateReadScopedBlockBuffer(
@@ -847,8 +861,10 @@ void BlockBasedTable::SetupBaseCacheKey(const TableProperties* properties,
 CacheKey BlockBasedTable::GetCacheKey(const OffsetableCacheKey& base_cache_key,
                                       const BlockHandle& handle) {
   // Minimum block size is 5 bytes; therefore we can trim off two lower bits
-  // from offet.
+  // from offset. This is the same scheme SimpleGen2Blob records use (see
-  return base_cache_key.WithOffset(handle.offset() >> 2);
+  // OffsetableCacheKey::WithOffsetForMinSizeRecord), so blocks and embedded
+  // blob records in the same file share one collision-free keyspace.
+  return base_cache_key.WithOffsetForMinSizeRecord(handle.offset());
 }
 
 Status BlockBasedTable::Open(
@@ -869,7 +885,7 @@ Status BlockBasedTable::Open(
     size_t max_file_size_for_l0_meta_pin, const std::string& cur_db_session_id,
     uint64_t cur_file_num, UniqueId64x2 expected_unique_id,
     const bool user_defined_timestamps_persisted,
-    const bool avoid_shared_metadata_cache) {
+    const bool avoid_shared_metadata_cache, BlobSource* blob_source) {
   table_reader->reset();
 
   Status s;
@@ -945,6 +961,7 @@ Status BlockBasedTable::Open(
       file_size, level, immortal_table, user_defined_timestamps_persisted);
   rep->file = std::move(file);
   rep->footer = footer;
+  rep->blob_source_ = blob_source;
 
   // For fully portable/stable cache keys, we need to read the properties
   // block before setting up cache keys. TODO: consider setting up a bootstrap
@@ -1263,6 +1280,11 @@ Status BlockBasedTable::ReadPropertiesBlock(
 
   // Read index_type from properties (required for format_version >= 2)
   auto& props = rep_->table_properties->user_collected_properties;
+  // The presence of the embedded blob stats property signals that this SST
+  // contains same-file ("embedded") blob records. The counter values are
+  // diagnostic only; only presence matters for correctness.
+  rep_->has_embedded_blobs =
+      props.find(kEmbeddedBlobSstStatsPropertyName) != props.end();
   auto index_type_pos = props.find(BlockBasedTablePropertyNames::kIndexType);
   if (index_type_pos == props.end()) {
     return Status::Corruption("Missing index type property");
@@ -1277,7 +1299,6 @@ Status BlockBasedTable::ReadPropertiesBlock(
   if (max_ts_pos != props.end()) {
     rep_->max_timestamp = Slice(max_ts_pos->second);
   }
-
   rep_->index_has_first_key =
       rep_->index_type == BlockBasedTableOptions::kBinarySearchWithFirstKey;
 
@@ -1289,6 +1310,314 @@ Status BlockBasedTable::ReadPropertiesBlock(
   return s;
 }
 
+bool BlockBasedTable::HasEmbeddedBlobRecords() const {
+  return rep_->has_embedded_blobs;
+}
+
+Status BlockBasedTable::ValidateEmbeddedBlobIndex(
+    const ReadOptions& read_options, const BlobIndex& blob_index,
+    size_t* payload_size, size_t* record_size) const {
+  assert(payload_size != nullptr);
+  assert(record_size != nullptr);
+  if (!blob_index.IsSameFile()) {
+    return Status::InvalidArgument("Blob index does not reference this file");
+  }
+  if (blob_index.HasTTL()) {
+    return Status::Corruption(
+        "Embedded blob index must not contain TTL metadata");
+  }
+  if (!rep_->has_embedded_blobs) {
+    return Status::Corruption(
+        "Same-file blob index found without embedded blob records");
+  }
+  if (read_options.read_tier == kBlockCacheTier) {
+    return Status::Incomplete(
+        "Cannot read embedded blob in block-cache-only mode");
+  }
+
+  const uint64_t payload_size_u64 = blob_index.size();
+  if (payload_size_u64 >
+      static_cast<uint64_t>(std::numeric_limits<size_t>::max()) -
+          kSimpleGen2BlobTrailerSize) {
+    return Status::Corruption("Embedded blob record is too large");
+  }
+  const uint64_t record_size_u64 =
+      payload_size_u64 + kSimpleGen2BlobTrailerSize;
+  // The per-record context checksum is the correctness guard for same-file
+  // references; this cheap bound just prevents a wild read past EOF (the range
+  // pre-check no longer exists now that records are interleaved with blocks).
+  const uint64_t record_offset = blob_index.offset();
+  if (record_offset > rep_->file_size ||
+      record_size_u64 > rep_->file_size - record_offset) {
+    return Status::Corruption("Embedded blob record extends past file size");
+  }
+
+  *payload_size = static_cast<size_t>(payload_size_u64);
+  *record_size = static_cast<size_t>(record_size_u64);
+  return Status::OK();
+}
+
+Status BlockBasedTable::ResolveEmbeddedBlob(const ReadOptions& read_options,
+                                            const BlobIndex& blob_index,
+                                            std::string* value) const {
+  assert(value != nullptr);
+  size_t payload_size = 0;
+  size_t record_size = 0;
+  Status s = ValidateEmbeddedBlobIndex(read_options, blob_index, &payload_size,
+                                       &record_size);
+  if (!s.ok()) {
+    return s;
+  }
+  // Read the record (payload + trailer) directly into `value`, then drop the
+  // trailer with a shrink (no extra copy on the common non-mmap path).
+  value->resize(record_size);
+  s = ReadAndVerifySimpleGen2BlobRecord(
+      read_options, rep_->file.get(), blob_index.offset(), payload_size,
+      record_size, rep_->footer.checksum_type(),
+      rep_->footer.base_context_checksum(), blob_index.compression(),
+      &(*value)[0]);
+  if (!s.ok()) {
+    return s;
+  }
+  value->resize(payload_size);
+  return Status::OK();
+}
+
+Status BlockBasedTable::ResolveEmbeddedBlobPinned(
+    const ReadOptions& read_options, const BlobIndex& blob_index,
+    PinnableSlice* value) const {
+  assert(value != nullptr);
+  size_t payload_size = 0;
+  size_t record_size = 0;
+  Status s = ValidateEmbeddedBlobIndex(read_options, blob_index, &payload_size,
+                                       &record_size);
+  if (!s.ok()) {
+    return s;
+  }
+  // Read into a heap buffer owned by `value` via a registered cleanup, so the
+  // payload can be pinned (no copy) and outlive this reader. The trailer just
+  // sits unused at the tail of the buffer.
+  std::unique_ptr<char[]> buf(new char[record_size]);
+  s = ReadAndVerifySimpleGen2BlobRecord(
+      read_options, rep_->file.get(), blob_index.offset(), payload_size,
+      record_size, rep_->footer.checksum_type(),
+      rep_->footer.base_context_checksum(), blob_index.compression(),
+      buf.get());
+  if (!s.ok()) {
+    return s;
+  }
+  char* raw = buf.release();
+  value->PinSlice(Slice(raw, payload_size), &DeleteCharArray, raw, nullptr);
+  return Status::OK();
+}
+
+Status BlockBasedTable::ResolveEmbeddedBlobCached(
+    const ReadOptions& read_options, const BlobIndex& blob_index,
+    PinnableSlice* value) const {
+  assert(value != nullptr);
+  assert(rep_->blob_source_ != nullptr);
+  size_t payload_size = 0;
+  size_t record_size = 0;
+  Status s = ValidateEmbeddedBlobIndex(read_options, blob_index, &payload_size,
+                                       &record_size);
+  if (!s.ok()) {
+    return s;
+  }
+
+  // BlobSource derives the cache key from the SimpleGen2Blob format (the same
+  // offset scheme as the SST's data blocks; see GetSimpleGen2BlobCacheKey), so
+  // we only pass the SST's base cache key. This keeps embedded blob records
+  // collision-free with data blocks even when blob_cache == block_cache.
+  return rep_->blob_source_->GetSimpleGen2Blob(
+      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(), value,
+      /*bytes_read=*/nullptr);
+}
+
+Status BlockBasedTable::MaybeResolveEmbeddedValue(
+    const ReadOptions& read_options, const Slice& internal_key,
+    const Slice& value, std::string* resolved_internal_key,
+    std::string* resolved_value, bool* resolved, PinnableSlice* pinned_value,
+    bool* value_pinned) const {
+  assert(resolved_internal_key != nullptr);
+  assert(resolved_value != nullptr);
+  assert(resolved != nullptr);
+
+  *resolved = false;
+  if (value_pinned != nullptr) {
+    *value_pinned = false;
+  }
+  if (!rep_->has_embedded_blobs) {
+    return Status::OK();
+  }
+
+  ParsedInternalKey parsed_key;
+  // FIXME: de-dup ParseInternalKey() work from callers
+  Status s =
+      ParseInternalKey(internal_key, &parsed_key, false /* log_err_key */);
+  if (!s.ok()) {
+    return s;
+  }
+
+  if (parsed_key.type == kTypeBlobIndex) {
+    BlobIndex blob_index;
+    s = blob_index.DecodeFrom(value);
+    if (!s.ok() || !blob_index.IsSameFile()) {
+      return s;
+    }
+
+    // Whole-value same-file blob. Route through the BlobSource (blob value
+    // cache + BLOB_DB_* stats) when available, pinning the cached payload (no
+    // copy). When no BlobSource is wired (SstFileReader/sst_dump/repair), fall
+    // back to a direct read. When the caller provides no PinnableSlice, use a
+    // local one so we still benefit from caching, then copy into
+    // `resolved_value`.
+    if (rep_->blob_source_ != nullptr) {
+      if (pinned_value != nullptr) {
+        s = ResolveEmbeddedBlobCached(read_options, blob_index, pinned_value);
+        if (s.ok() && value_pinned != nullptr) {
+          *value_pinned = true;
+        }
+      } else {
+        PinnableSlice local_value;
+        s = ResolveEmbeddedBlobCached(read_options, blob_index, &local_value);
+        if (s.ok()) {
+          resolved_value->assign(local_value.data(), local_value.size());
+        }
+      }
+    } else if (pinned_value != nullptr) {
+      s = ResolveEmbeddedBlobPinned(read_options, blob_index, pinned_value);
+      if (s.ok() && value_pinned != nullptr) {
+        *value_pinned = true;
+      }
+    } else {
+      s = ResolveEmbeddedBlob(read_options, blob_index, resolved_value);
+    }
+    if (!s.ok()) {
+      return s;
+    }
+
+    InternalKey resolved_key(parsed_key.user_key, parsed_key.sequence,
+                             kTypeValue);
+    const Slice encoded_resolved_key = resolved_key.Encode();
+    resolved_internal_key->assign(encoded_resolved_key.data(),
+                                  encoded_resolved_key.size());
+    *resolved = true;
+    return Status::OK();
+  }
+
+  if (parsed_key.type != kTypeWideColumnEntity) {
+    return Status::OK();
+  }
+
+  // FIXME: Reading embedded blob wide columns is very heavyweight with memcpy
+  // and serialization/deserialization
+  bool has_blob_columns = false;
+  s = WideColumnSerialization::HasBlobColumns(value, has_blob_columns);
+  if (!s.ok() || !has_blob_columns) {
+    return s;
+  }
+
+  Slice entity(value);
+  std::vector<WideColumn> columns;
+  std::vector<std::pair<size_t, BlobIndex>> blob_columns;
+  s = WideColumnSerialization::DeserializeV2(entity, columns, blob_columns);
+  if (!s.ok()) {
+    return s;
+  }
+
+  std::vector<std::pair<std::string, std::string>> rewritten_columns;
+  rewritten_columns.reserve(columns.size());
+  for (const WideColumn& column : columns) {
+    rewritten_columns.emplace_back(column.name().ToString(),
+                                   column.value().ToString());
+  }
+
+  std::vector<std::pair<size_t, BlobIndex>> remaining_blob_columns;
+  remaining_blob_columns.reserve(blob_columns.size());
+  for (const auto& blob_column : blob_columns) {
+    const size_t column_index = blob_column.first;
+    const BlobIndex& blob_index = blob_column.second;
+    if (!blob_index.IsSameFile()) {
+      remaining_blob_columns.emplace_back(blob_column);
+      continue;
+    }
+    if (column_index >= rewritten_columns.size()) {
+      return Status::Corruption("Wide-column blob index out of range");
+    }
+
+    // Per-column same-file blob: cache via BlobSource when available (same
+    // SST-like key, keyed by this column's record offset), copying into the
+    // rebuilt column value; otherwise read directly.
+    if (rep_->blob_source_ != nullptr) {
+      PinnableSlice column_value;
+      s = ResolveEmbeddedBlobCached(read_options, blob_index, &column_value);
+      if (s.ok()) {
+        rewritten_columns[column_index].second.assign(column_value.data(),
+                                                      column_value.size());
+      }
+    } else {
+      s = ResolveEmbeddedBlob(read_options, blob_index,
+                              &rewritten_columns[column_index].second);
+    }
+    if (!s.ok()) {
+      return s;
+    }
+    *resolved = true;
+  }
+
+  if (!*resolved) {
+    return Status::OK();
+  }
+
+  resolved_internal_key->assign(internal_key.data(), internal_key.size());
+  resolved_value->clear();
+  if (!remaining_blob_columns.empty()) {
+    return WideColumnSerialization::SerializeV2(
+        rewritten_columns, remaining_blob_columns, *resolved_value);
+  }
+
+  WideColumns resolved_columns;
+  resolved_columns.reserve(rewritten_columns.size());
+  for (const auto& column : rewritten_columns) {
+    resolved_columns.emplace_back(column.first, column.second);
+  }
+  return WideColumnSerialization::Serialize(resolved_columns, *resolved_value);
+}
+
+Status BlockBasedTable::ResolveEmbeddedValueForGet(
+    const ReadOptions& read_options, const Slice& key, const Slice& value,
+    EmbeddedValueGetScratch* scratch, Cleanable* block_pinner,
+    Slice* key_to_save, Slice* value_to_save, Cleanable** value_pinner) const {
+  assert(scratch != nullptr);
+  *key_to_save = key;
+  *value_to_save = value;
+  // When nothing is resolved, save the original entry pinned by the data block.
+  *value_pinner = block_pinner;
+  bool resolved = false;
+  bool value_pinned = false;
+  // The pinned buffer (if any) from a prior entry has been delegated to that
+  // entry's output; reset before reuse so PinSlice can pin again.
+  scratch->pinned_value.Reset();
+  Status s = MaybeResolveEmbeddedValue(
+      read_options, key, value, &scratch->key_buf, &scratch->value_buf,
+      &resolved, &scratch->pinned_value, &value_pinned);
+  if (s.ok() && resolved) {
+    *key_to_save = Slice(scratch->key_buf);
+    if (value_pinned) {
+      // Whole-value blob: pin the owned buffer into the output (no copy).
+      *value_to_save = Slice(scratch->pinned_value);
+      *value_pinner = &scratch->pinned_value;
+    } else {
+      // Built (wide-column) value lives in reused scratch; it must be copied.
+      *value_to_save = Slice(scratch->value_buf);
+      *value_pinner = nullptr;
+    }
+  }
+  return s;
+}
+
 Status BlockBasedTable::ReadRangeDelBlock(
     const ReadOptions& read_options, FilePrefetchBuffer* prefetch_buffer,
     InternalIterator* meta_iter,
@@ -2423,25 +2752,45 @@ InternalIterator* BlockBasedTable::NewIterator(
       /*disable_prefix_seek=*/need_upper_bound_check &&
           rep_->index_type == BlockBasedTableOptions::kHashSearch,
       /*input_iter=*/nullptr, /*get_context=*/nullptr, &lookup_context));
-  if (arena == nullptr) {
+  bool check_filter =
-    return new BlockBasedTableIterator(
-        this, read_options, rep_->internal_comparator, std::move(index_iter),
       !skip_filters &&
       (!read_options.total_order_seek || read_options.auto_prefix_mode ||
        read_options.prefix_same_as_start) &&
-            prefix_extractor != nullptr,
+      prefix_extractor != nullptr;
-        need_upper_bound_check, prefix_extractor, caller,
+  // Same-file ("embedded") blob references are resolved by a wrapper iterator
-        compaction_readahead_size, allow_unprepared_value);
+  // so the block-based iterator never exposes an unresolved BlobIndex. The SST
+  // dump tool must keep seeing raw BlobIndex values, so it is excluded.
+  const bool resolve_embedded_values =
+      caller != TableReaderCaller::kSSTDumpTool && HasEmbeddedBlobRecords();
+  // When wrapping, the inner iterator must materialize each data block (it must
+  // not defer to the index's first key), so force allow_unprepared_value off;
+  // the wrapper re-introduces value laziness above it.
+  const bool inner_allow_unprepared_value =
+      allow_unprepared_value && !resolve_embedded_values;
+
+  if (arena == nullptr) {
+    auto* iter = new BlockBasedTableIterator(
+        this, read_options, rep_->internal_comparator, std::move(index_iter),
+        check_filter, need_upper_bound_check, prefix_extractor, caller,
+        compaction_readahead_size, inner_allow_unprepared_value);
+    if (!resolve_embedded_values) {
+      return iter;
+    }
+    return new EmbeddedBlobResolvingIterator(this, read_options, iter,
+                                             /*arena_mode=*/false);
   } else {
     auto* mem = arena->AllocateAligned(sizeof(BlockBasedTableIterator));
-    return new (mem) BlockBasedTableIterator(
+    auto* iter = new (mem) BlockBasedTableIterator(
         this, read_options, rep_->internal_comparator, std::move(index_iter),
-        !skip_filters &&
+        check_filter, need_upper_bound_check, prefix_extractor, caller,
-            (!read_options.total_order_seek || read_options.auto_prefix_mode ||
+        compaction_readahead_size, inner_allow_unprepared_value);
-             read_options.prefix_same_as_start) &&
+    if (!resolve_embedded_values) {
-            prefix_extractor != nullptr,
+      return iter;
-        need_upper_bound_check, prefix_extractor, caller,
+    }
-        compaction_readahead_size, allow_unprepared_value);
+    auto* wrap_mem =
+        arena->AllocateAligned(sizeof(EmbeddedBlobResolvingIterator));
+    return new (wrap_mem) EmbeddedBlobResolvingIterator(
+        this, read_options, iter, /*arena_mode=*/true);
   }
 }
 
@@ -2686,6 +3035,14 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
         rep_->internal_comparator.user_comparator()->timestamp_size();
     bool matched = false;  // if such user key matched a key in SST
     bool done = false;
+    // Embedded blob resolution is gated on this table actually having embedded
+    // blob records, so the common (non-embedded) Get path is unchanged. The
+    // scratch (reused across entries) is only constructed for embedded tables
+    // and avoids per-entry allocation.
+    std::optional<EmbeddedValueGetScratch> embedded_scratch;
+    if (rep_->has_embedded_blobs) {
+      embedded_scratch.emplace();
+    }
     for (iiter->Seek(key); iiter->Valid() && !done; iiter->Next()) {
       IndexValue v = iiter->value();
 
@@ -2739,18 +3096,29 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
       } else {
         // Call the *saver function on each entry/block until it returns false
         for (; biter.Valid(); biter.Next()) {
+          Slice key_to_save = biter.key();
+          Slice value_to_save = biter.value();
+          Cleanable* value_pinner = biter.IsValuePinned() ? &biter : nullptr;
+          if (embedded_scratch) {
+            s = ResolveEmbeddedValueForGet(
+                read_options, biter.key(), biter.value(), &*embedded_scratch,
+                value_pinner, &key_to_save, &value_to_save, &value_pinner);
+            if (!s.ok()) {
+              break;
+            }
+          }
+
           ParsedInternalKey parsed_key;
           Status pik_status = ParseInternalKey(
-              biter.key(), &parsed_key, false /* log_err_key */);  // TODO
+              key_to_save, &parsed_key, false /* log_err_key */);  // TODO
           if (!pik_status.ok()) {
             s = pik_status;
             break;
           }
 
           Status read_status;
-          bool ret = get_context->SaveValue(
+          bool ret = get_context->SaveValue(parsed_key, value_to_save, &matched,
-              parsed_key, biter.value(), &matched, &read_status,
+                                            &read_status, value_pinner);
-              biter.IsValuePinned() ? &biter : nullptr);
           if (!read_status.ok()) {
             s = read_status;
             break;

table/block_based/block_based_table_reader.h

@@ -11,6 +11,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <string>
 
 #include "cache/cache_entry_roles.h"
 #include "cache/cache_key.h"
@@ -27,6 +28,7 @@
 #include "table/block_based/cachable_entry.h"
 #include "table/block_based/filter_block.h"
 #include "table/block_based/uncompression_dict_reader.h"
+#include "table/embedded_blob_sst.h"
 #include "table/format.h"
 #include "table/persistent_cache_options.h"
 #include "table/table_properties_internal.h"
@@ -51,6 +53,8 @@ class FSRandomAccessFile;
 class TableCache;
 class TableReader;
 class WritableFile;
+class BlobIndex;
+class BlobSource;
 struct BlockBasedTableOptions;
 struct EnvOptions;
 struct ReadOptions;
@@ -169,7 +173,8 @@ class BlockBasedTable : public TableReader {
       const std::string& cur_db_session_id = "", uint64_t cur_file_num = 0,
       UniqueId64x2 expected_unique_id = {},
       const bool user_defined_timestamps_persisted = true,
-      bool avoid_shared_metadata_cache = false);
+      bool avoid_shared_metadata_cache = false,
+      BlobSource* blob_source = nullptr);
 
   bool PrefixRangeMayMatch(const Slice& internal_key,
                            const ReadOptions& read_options,
@@ -203,6 +208,80 @@ class BlockBasedTable : public TableReader {
              GetContext* get_context, const SliceTransform* prefix_extractor,
              bool skip_filters = false) override;
 
+  // Whether this SST contains embedded (same-file) blob records, as advertised
+  // by the embedded blob stats property.
+  bool HasEmbeddedBlobRecords() const;
+
+  // Reads a single same-file blob record referenced by `blob_index`.
+  Status ResolveEmbeddedBlob(const ReadOptions& read_options,
+                             const BlobIndex& blob_index,
+                             std::string* value) const;
+
+  // Like ResolveEmbeddedBlob, but reads the payload into a heap buffer owned by
+  // `value` (via a registered cleanup) and pins it, avoiding a copy. Use on the
+  // Get()/MultiGet() path so the resolved value can be pinned into the output.
+  Status ResolveEmbeddedBlobPinned(const ReadOptions& read_options,
+                                   const BlobIndex& blob_index,
+                                   PinnableSlice* value) const;
+
+  // Like ResolveEmbeddedBlobPinned, but routes the read through the CFD's
+  // BlobSource so the payload is served from / inserted into the blob value
+  // cache and BLOB_DB_* statistics are recorded. BlobSource derives the cache
+  // key from the SimpleGen2Blob format (the same offset scheme as the SST's
+  // data blocks; see GetSimpleGen2BlobCacheKey), so embedded blob records stay
+  // collision-free with data blocks even when the blob cache and block cache
+  // are shared. Must only be called when rep_->blob_source_ is non-null.
+  Status ResolveEmbeddedBlobCached(const ReadOptions& read_options,
+                                   const BlobIndex& blob_index,
+                                   PinnableSlice* value) const;
+
+  // If `value` is a same-file BlobIndex, materializes the referenced payload
+  // and updates `resolved_internal_key` to the corresponding value type. Leaves
+  // `resolved` false when no embedded resolution is needed. When `pinned_value`
+  // is non-null, a whole-value same-file blob is pinned into it without a copy
+  // (and `*value_pinned` is set); otherwise the value is built into
+  // `resolved_value`. Wide-column entities are always built into
+  // `resolved_value`.
+  Status MaybeResolveEmbeddedValue(const ReadOptions& read_options,
+                                   const Slice& internal_key,
+                                   const Slice& value,
+                                   std::string* resolved_internal_key,
+                                   std::string* resolved_value, bool* resolved,
+                                   PinnableSlice* pinned_value = nullptr,
+                                   bool* value_pinned = nullptr) const;
+
+  // Reusable scratch for resolving embedded values across a Get()/MultiGet()
+  // loop, so the hot loop performs no per-entry allocation. Construct one per
+  // Get()/MultiGet() call and pass it to ResolveEmbeddedValueForGet().
+  struct EmbeddedValueGetScratch {
+    std::string key_buf;
+    std::string value_buf;
+    PinnableSlice pinned_value;
+  };
+
+  // Resolves the embedded blob / wide-column value for one Get()/MultiGet()
+  // entry and selects the Cleanable to hand to GetContext::SaveValue.
+  // `block_pinner` is the data block iterator's value pinner (or nullptr). On
+  // return, *key_to_save / *value_to_save are the slices to save and
+  // *value_pinner is the pinner to use: the zero-copy blob payload pinner (in
+  // `scratch`), nullptr for a copied wide-column value, or `block_pinner` when
+  // nothing needed resolving. Only call this when the table actually has
+  // embedded blob records (see Rep::has_embedded_blobs); the common
+  // non-embedded path should skip it entirely.
+  Status ResolveEmbeddedValueForGet(const ReadOptions& read_options,
+                                    const Slice& key, const Slice& value,
+                                    EmbeddedValueGetScratch* scratch,
+                                    Cleanable* block_pinner, Slice* key_to_save,
+                                    Slice* value_to_save,
+                                    Cleanable** value_pinner) const;
+
+  // Validates a same-file `blob_index` and returns the payload and full record
+  // (payload + trailer) sizes. Shared by the ResolveEmbeddedBlob* variants.
+  Status ValidateEmbeddedBlobIndex(const ReadOptions& read_options,
+                                   const BlobIndex& blob_index,
+                                   size_t* payload_size,
+                                   size_t* record_size) const;
+
   Status MultiGetFilter(const ReadOptions& read_options,
                         const SliceTransform* prefix_extractor,
                         MultiGetRange* mget_range) override;
@@ -756,6 +835,18 @@ struct BlockBasedTable::Rep {
 
   // If true, then data blocks have keys and values separated.
   bool separate_key_value_in_data_block = false;
+  // Whether this SST contains embedded (same-file) blob records, detected from
+  // the presence of the embedded blob stats property. Same-file BlobIndex
+  // references are only valid when this is true.
+  bool has_embedded_blobs = false;
+
+  // BlobSource for routing same-file ("embedded") blob reads through the blob
+  // value cache + BLOB_DB_* statistics. Owned by the ColumnFamilyData; this
+  // reader is owned (via TableCache) by the same CFD, so the raw pointer is
+  // lifetime-safe. nullptr for non-DB openers (SstFileReader, sst_dump,
+  // repair, external-file ingestion prevalidation, etc.); in that case
+  // embedded reads fall back to a direct (uncached) read.
+  BlobSource* blob_source_ = nullptr;
 
   // Whether block checksums in metadata blocks were verified on open.
   // This is only to mostly maintain current dubious behavior of VerifyChecksum

table/block_based/block_based_table_reader_sync_and_async.h

@@ -4,6 +4,8 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 
+#include <optional>
+
 #include "util/aligned_buffer.h"
 #include "util/async_file_reader.h"
 #include "util/coro_utils.h"
@@ -542,6 +544,14 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::MultiGet)
     DataBlockIter next_biter;
     size_t idx_in_batch = 0;
     SharedCleanablePtr shared_cleanable;
+    // Embedded blob resolution is gated on this table actually having embedded
+    // blob records, so the common (non-embedded) MultiGet path is unchanged.
+    // The scratch (reused across entries) is only constructed for embedded
+    // tables and avoids per-entry allocation.
+    std::optional<EmbeddedValueGetScratch> embedded_scratch;
+    if (rep_->has_embedded_blobs) {
+      embedded_scratch.emplace();
+    }
     for (auto miter = sst_file_range.begin(); miter != sst_file_range.end();
          ++miter) {
       Status s;
@@ -671,17 +681,28 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::MultiGet)
 
         // Call the *saver function on each entry/block until it returns false
         for (; biter->status().ok() && biter->Valid(); biter->Next()) {
+          Slice key_to_save = biter->key();
+          Slice value_to_save = biter->value();
+          Cleanable* effective_pinner = value_pinner;
+          if (embedded_scratch) {
+            s = ResolveEmbeddedValueForGet(
+                read_options, biter->key(), biter->value(), &*embedded_scratch,
+                value_pinner, &key_to_save, &value_to_save, &effective_pinner);
+            if (!s.ok()) {
+              break;
+            }
+          }
+
           ParsedInternalKey parsed_key;
           Status pik_status = ParseInternalKey(
-              biter->key(), &parsed_key, false /* log_err_key */);  // TODO
+              key_to_save, &parsed_key, false /* log_err_key */);  // TODO
           if (!pik_status.ok()) {
             s = pik_status;
             break;
           }
           Status read_status;
-          bool ret = get_context->SaveValue(
+          bool ret = get_context->SaveValue(parsed_key, value_to_save, &matched,
-              parsed_key, biter->value(), &matched, &read_status,
+                                            &read_status, effective_pinner);
-              value_pinner ? value_pinner : nullptr);
           if (!read_status.ok()) {
             s = read_status;
             break;

table/block_based/embedded_blob_resolving_iterator.h

@@ -0,0 +1,367 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+#pragma once
+
+#include <limits>
+#include <string>
+
+#include "db/blob/blob_index.h"
+#include "db/dbformat.h"
+#include "db/pinned_iterators_manager.h"
+#include "rocksdb/slice.h"
+#include "table/block_based/block_based_table_reader.h"
+#include "table/internal_iterator.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+// Wraps a block-based table iterator and resolves same-file ("embedded") blob
+// references so that callers never see an unresolved same-file BlobIndex. It
+// makes an embedded entry look like an ordinary key/value entry:
+//   - key(): a whole-value same-file BlobIndex's internal key has its value
+//     type rewritten from kTypeBlobIndex back to kTypeValue.
+//   - value(): the referenced same-file blob payload is materialized (a wide-
+//     column entity's same-file blob columns are rewritten inline).
+// Traditional (separate-file) BlobIndex entries and plain entries pass through
+// untouched.
+//
+// Why a wrapper instead of putting this in BlockBasedTableIterator:
+//   - The wrapped iterator is created with allow_unprepared_value=false, so it
+//     always materializes the data block and never sits in the "first key from
+//     index" deferred state. That keeps the "never expose an unresolved
+//     same-file blob index" invariant structural rather than relying on every
+//     downstream consumer re-checking the key type after PrepareValue().
+//   - The wrapper preserves value laziness on its own: key() does only the
+//     cheap key-type rewrite (no blob-region I/O), while the payload is read
+//     lazily in value()/PrepareValue().
+//   - It keeps the embedded-blob concern out of the hot, complex
+//     BlockBasedTableIterator, and is only instantiated for SSTs that actually
+//     carry an embedded blob segment.
+//
+// Only created when the table advertises an embedded blob segment and the
+// caller is not the SST dump tool (which must keep seeing raw BlobIndex
+// values).
+class EmbeddedBlobResolvingIterator : public InternalIterator {
+ public:
+  // `iter` is owned by this wrapper. `arena_mode` must match how `iter` (and
+  // this wrapper) were allocated so the destructor frees `iter` correctly.
+  EmbeddedBlobResolvingIterator(const BlockBasedTable* table,
+                                const ReadOptions& read_options,
+                                InternalIterator* iter, bool arena_mode)
+      : table_(table),
+        read_options_(read_options),
+        iter_(iter),
+        arena_mode_(arena_mode) {
+    status_.PermitUncheckedError();
+  }
+
+  ~EmbeddedBlobResolvingIterator() override {
+    if (arena_mode_) {
+      iter_->~InternalIteratorBase<Slice>();
+    } else {
+      delete iter_;
+    }
+  }
+
+  // No copying allowed
+  EmbeddedBlobResolvingIterator(const EmbeddedBlobResolvingIterator&) = delete;
+  EmbeddedBlobResolvingIterator& operator=(
+      const EmbeddedBlobResolvingIterator&) = delete;
+
+  bool Valid() const override { return iter_->Valid() && status_.ok(); }
+
+  void SeekToFirst() override {
+    ResetState();
+    iter_->SeekToFirst();
+  }
+  void SeekToLast() override {
+    ResetState();
+    iter_->SeekToLast();
+  }
+  void Seek(const Slice& target) override {
+    ResetState();
+    iter_->Seek(target);
+  }
+  void SeekForPrev(const Slice& target) override {
+    ResetState();
+    iter_->SeekForPrev(target);
+  }
+  void Next() override {
+    ResetState();
+    iter_->Next();
+  }
+  bool NextAndGetResult(IterateResult* result) override {
+    ResetState();
+    iter_->Next();
+    if (!Valid()) {
+      return false;
+    }
+    result->key = key();
+    result->bound_check_result = iter_->UpperBoundCheckResult();
+    result->value_prepared = false;
+    // key() may have set a (corruption) status while resolving the key type.
+    return Valid();
+  }
+  void Prev() override {
+    ResetState();
+    iter_->Prev();
+  }
+
+  Slice key() const override {
+    assert(Valid());
+    if (ResolveKeyType() && key_resolved_) {
+      return Slice(resolved_internal_key_);
+    }
+    // ResolveKeyType() may have set a (corruption-only, no I/O) error. This
+    // const accessor falls back to the raw key; the error is still surfaced
+    // through status()/Valid(), so permit it from being unchecked here.
+    status_.PermitUncheckedError();
+    return iter_->key();
+  }
+
+  Slice user_key() const override { return ExtractUserKey(key()); }
+
+  Slice value() const override {
+    assert(Valid());
+    if (MaterializeValue() && value_resolved_) {
+      if (value_is_pinned_) {
+        // Whole-value blob payload pinned in the blob cache (or an owned
+        // buffer); returned without a copy.
+        return Slice(resolved_pinned_value_);
+      }
+      return Slice(resolved_value_);
+    }
+    // MaterializeValue() may have set an error (corruption or blob-region I/O).
+    // Fall back to the raw value; the error is surfaced via status()/Valid().
+    status_.PermitUncheckedError();
+    return iter_->value();
+  }
+
+  bool PrepareValue() override {
+    assert(Valid());
+    if (!iter_->PrepareValue()) {
+      return false;
+    }
+    return MaterializeValue();
+  }
+
+  Status status() const override {
+    if (!status_.ok()) {
+      return status_;
+    }
+    return iter_->status();
+  }
+
+  uint64_t write_unix_time() const override { return iter_->write_unix_time(); }
+
+  bool IsKeyPinned() const override {
+    // The rewritten key is owned by this wrapper, so it is not pinned.
+    if (key_resolved_) {
+      return false;
+    }
+    return iter_->IsKeyPinned();
+  }
+  bool IsValuePinned() const override {
+    if (value_resolved_) {
+      // A resolved whole-value blob payload is pinned in the blob cache (or an
+      // owned buffer). The IsValuePinned() contract requires the value to stay
+      // valid until the iterator is deleted / ReleasePinnedData is called, so
+      // only advertise it as pinned when a PinnedIteratorsManager is active to
+      // take over the pin's cleanup across repositioning (see ResetState). A
+      // built (wide-column) value lives in `resolved_value_` and is never
+      // pinned.
+      return value_is_pinned_ && pinned_iters_mgr_ != nullptr &&
+             pinned_iters_mgr_->PinningEnabled();
+    }
+    return iter_->IsValuePinned();
+  }
+
+  void SetPinnedItersMgr(PinnedIteratorsManager* pinned_iters_mgr) override {
+    pinned_iters_mgr_ = pinned_iters_mgr;
+    iter_->SetPinnedItersMgr(pinned_iters_mgr);
+  }
+
+  IterBoundCheck UpperBoundCheckResult() override {
+    return iter_->UpperBoundCheckResult();
+  }
+  bool MayBeOutOfLowerBound() override { return iter_->MayBeOutOfLowerBound(); }
+
+  void SetRangeDelReadSeqno(SequenceNumber read_seqno) override {
+    iter_->SetRangeDelReadSeqno(read_seqno);
+  }
+  bool IsDeleteRangeSentinelKey() const override {
+    return iter_->IsDeleteRangeSentinelKey();
+  }
+
+  void GetReadaheadState(ReadaheadFileInfo* readahead_file_info) override {
+    iter_->GetReadaheadState(readahead_file_info);
+  }
+  void SetReadaheadState(ReadaheadFileInfo* readahead_file_info) override {
+    iter_->SetReadaheadState(readahead_file_info);
+  }
+
+  Status GetProperty(std::string prop_name, std::string* prop) override {
+    return iter_->GetProperty(std::move(prop_name), prop);
+  }
+
+  void Prepare(const MultiScanArgs* scan_opts) override {
+    iter_->Prepare(scan_opts);
+  }
+
+ private:
+  void ResetState() {
+    status_.PermitUncheckedError();
+    status_ = Status::OK();
+    key_prepared_ = false;
+    value_prepared_ = false;
+    key_resolved_ = false;
+    value_resolved_ = false;
+    resolved_internal_key_.clear();
+    resolved_value_.clear();
+    if (value_is_pinned_) {
+      // If a PinnedIteratorsManager is active it has been told (via
+      // IsValuePinned) that the previous entry's pinned blob stays valid until
+      // ReleasePinnedData; hand the cache pin's cleanup to it so the slice
+      // outlives this reposition. Otherwise just release the pin now.
+      if (pinned_iters_mgr_ != nullptr && pinned_iters_mgr_->PinningEnabled()) {
+        resolved_pinned_value_.DelegateCleanupsTo(pinned_iters_mgr_);
+      }
+      value_is_pinned_ = false;
+    }
+    resolved_pinned_value_.Reset();
+  }
+
+  // Cheap, key-only resolution: for a whole-value same-file BlobIndex entry,
+  // computes the logical internal key with its value type rewritten from
+  // kTypeBlobIndex to kTypeValue. Only touches data already in the data block
+  // (the key and the inline serialized BlobIndex) -- it does NOT read the blob
+  // region. No-op (returns true) for entries that are not a same-file
+  // BlobIndex. Returns false and sets `status_` (corruption only) on a
+  // malformed key/BlobIndex.
+  bool ResolveKeyType() const {
+    if (!iter_->Valid()) {
+      return true;
+    }
+    if (key_prepared_) {
+      return status_.ok();
+    }
+    key_prepared_ = true;
+
+    const Slice current_key = iter_->key();
+    if (ExtractValueType(current_key) != kTypeBlobIndex) {
+      return true;
+    }
+
+    ParsedInternalKey parsed_key;
+    // Only data-corruption errors are possible here (no I/O).
+    status_ = ParseInternalKey(current_key, &parsed_key, false /* log_err */);
+    if (!status_.ok()) {
+      return false;
+    }
+
+    BlobIndex blob_index;
+    // Decodes the inline BlobIndex from the data block; still no blob-region
+    // IO.
+    status_ = blob_index.DecodeFrom(iter_->value());
+    if (!status_.ok()) {
+      return false;
+    }
+    if (!blob_index.IsSameFile()) {
+      // Traditional (separate-file) BlobDB index; leave the key type as
+      // kTypeBlobIndex for the higher layers to resolve.
+      return true;
+    }
+
+    InternalKey resolved_key(parsed_key.user_key, parsed_key.sequence,
+                             kTypeValue);
+    const Slice encoded_resolved_key = resolved_key.Encode();
+    resolved_internal_key_.assign(encoded_resolved_key.data(),
+                                  encoded_resolved_key.size());
+    key_resolved_ = true;
+    return true;
+  }
+
+  // Heavyweight value resolution: materializes the logical value for the
+  // current entry. For a same-file BlobIndex entry this reads the blob payload
+  // (may do I/O); for a wide-column entity it rewrites same-file blob columns
+  // inline. For a BlobIndex entry it first resolves the key type so key() and
+  // value() agree. No-op (returns true) for plain entries. Returns false and
+  // sets `status_` (corruption or I/O error) on failure.
+  bool MaterializeValue() const {
+    if (!iter_->Valid()) {
+      return true;
+    }
+
+    const Slice current_key = iter_->key();
+    const ValueType value_type = ExtractValueType(current_key);
+    if (value_type != kTypeBlobIndex && value_type != kTypeWideColumnEntity) {
+      return true;
+    }
+
+    if (value_type == kTypeBlobIndex && !ResolveKeyType()) {
+      return false;
+    }
+    if (value_prepared_) {
+      return status_.ok();
+    }
+    value_prepared_ = true;
+
+    std::string resolved_key;
+    std::string resolved_value;
+    bool resolved = false;
+    bool value_pinned = false;
+    status_ = table_->MaybeResolveEmbeddedValue(
+        read_options_, current_key, iter_->value(), &resolved_key,
+        &resolved_value, &resolved, &resolved_pinned_value_, &value_pinned);
+    if (!status_.ok()) {
+      return false;
+    }
+    if (!resolved) {
+      return true;
+    }
+
+    if (!resolved_key.empty()) {
+      resolved_internal_key_ = std::move(resolved_key);
+      key_resolved_ = true;
+    }
+    if (value_pinned) {
+      // Whole-value blob: payload pinned into resolved_pinned_value_ (no copy).
+      value_is_pinned_ = true;
+    } else {
+      // Built (wide-column) value: owned by this wrapper.
+      resolved_value_ = std::move(resolved_value);
+    }
+    value_resolved_ = true;
+    return true;
+  }
+
+  const BlockBasedTable* table_;
+  const ReadOptions& read_options_;
+  InternalIterator* iter_;
+  const bool arena_mode_;
+  PinnedIteratorsManager* pinned_iters_mgr_ = nullptr;
+
+  // Cached resolution state for the current entry, reset on every reposition.
+  // Mutable because resolution is driven lazily from the const key()/value()
+  // accessors.
+  mutable Status status_;
+  mutable std::string resolved_internal_key_;
+  mutable std::string resolved_value_;
+  // Holds a whole-value same-file blob payload pinned in the blob cache (or an
+  // owned buffer when no cache is configured), avoiding a copy. Used only when
+  // value_is_pinned_ is true; wide-column values use resolved_value_ instead.
+  mutable PinnableSlice resolved_pinned_value_;
+  mutable bool key_prepared_ = false;
+  mutable bool value_prepared_ = false;
+  mutable bool key_resolved_ = false;
+  mutable bool value_resolved_ = false;
+  mutable bool value_is_pinned_ = false;
+};
+
+}  // namespace ROCKSDB_NAMESPACE

table/embedded_blob_sst.h

@@ -0,0 +1,80 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <string>
+
+#include "db/blob/blob_gen2_format.h"
+#include "rocksdb/slice.h"
+#include "rocksdb/status.h"
+#include "util/coding.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+struct SstFileWriterEmbeddedBlobOptions;
+
+// Internal builder-side spelling for the public SstFileWriter options. Keeping
+// this as an alias makes the table builder's contract explicit without forking
+// the option definition.
+using EmbeddedBlobSstBuilderOptions = SstFileWriterEmbeddedBlobOptions;
+
+// User property with best-effort diagnostic counters for the embedded blob
+// records. The presence of this property is also the reader's signal that the
+// SST contains embedded blob records; readers must not depend on the counter
+// values for correctness.
+inline constexpr char kEmbeddedBlobSstStatsPropertyName[] =
+    "rocksdb.embedded.blob.stats";
+
+// Embedded blob records use the SimpleGen2Blob record format (payload bytes
+// followed by a compression-marker byte and a four-byte checksum); see
+// db/blob/blob_gen2_format.h.
+
+// Diagnostic counters for embedded blob records. Stored as a user property and
+// intentionally ignored (other than for presence detection) by readers.
+struct EmbeddedBlobStats {
+  uint64_t blob_count = 0;
+  uint64_t payload_bytes = 0;
+
+  // Whether any embedded blob records contributed to the counters.
+  bool HasRecords() const { return blob_count > 0; }
+};
+
+// Encodes EmbeddedBlobStats for kEmbeddedBlobSstStatsPropertyName.
+inline void EncodeEmbeddedBlobStats(const EmbeddedBlobStats& stats,
+                                    std::string* dst) {
+  dst->clear();
+  PutVarint64(dst, stats.blob_count);
+  PutVarint64(dst, stats.payload_bytes);
+}
+
+// Decodes EmbeddedBlobStats from kEmbeddedBlobSstStatsPropertyName.
+inline Status DecodeEmbeddedBlobStats(Slice input, EmbeddedBlobStats* stats) {
+  if (stats == nullptr) {
+    return Status::InvalidArgument("Missing embedded blob stats output");
+  }
+
+  EmbeddedBlobStats decoded;
+  if (!GetVarint64(&input, &decoded.blob_count) ||
+      !GetVarint64(&input, &decoded.payload_bytes)) {
+    return Status::Corruption("Error decoding embedded blob stats");
+  }
+
+  if (decoded.blob_count == 0 && decoded.payload_bytes != 0) {
+    return Status::Corruption("Embedded blob stats have bytes but no records");
+  }
+
+  *stats = decoded;
+  return Status::OK();
+}
+
+}  // namespace ROCKSDB_NAMESPACE

table/sst_file_reader_test.cc

@@ -5,17 +5,32 @@
 
 #include "rocksdb/sst_file_reader.h"
 
+#include <atomic>
 #include <cinttypes>
 
 #include "db/db_test_util.h"
+#include "db/dbformat.h"
+#include "env/composite_env_wrapper.h"
+#include "file/random_access_file_reader.h"
+#include "options/cf_options.h"
 #include "port/stack_trace.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
+#include "rocksdb/file_system.h"
 #include "rocksdb/sst_file_writer.h"
 #include "rocksdb/utilities/types_util.h"
+#include "rocksdb/wide_columns.h"
+#include "table/block_based/block.h"
+#include "table/embedded_blob_sst.h"
+#include "table/format.h"
+#include "table/internal_iterator.h"
+#include "table/meta_blocks.h"
 #include "table/sst_file_writer_collectors.h"
+#include "table/table_reader.h"
+#include "test_util/sync_point.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
+#include "util/compression.h"
 #include "utilities/merge_operators.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -101,6 +116,37 @@ class SstFileReaderTest : public testing::Test {
     CheckFile(sst_name_, keys);
   }
 
+  void GetEmbeddedBlobStats(SstFileReader* reader, EmbeddedBlobStats* stats) {
+    ASSERT_NE(reader, nullptr);
+    ASSERT_NE(stats, nullptr);
+
+    std::shared_ptr<const TableProperties> properties =
+        reader->GetTableProperties();
+    ASSERT_NE(properties, nullptr);
+    const auto& user_properties = properties->user_collected_properties;
+    auto stats_iter = user_properties.find(kEmbeddedBlobSstStatsPropertyName);
+    ASSERT_NE(stats_iter, user_properties.end());
+    ASSERT_OK(DecodeEmbeddedBlobStats(Slice(stats_iter->second), stats));
+  }
+
+  void ReadFileBytes(const std::string& file_name, uint64_t offset, size_t size,
+                     std::string* bytes) {
+    ASSERT_NE(bytes, nullptr);
+
+    std::unique_ptr<FSRandomAccessFile> file;
+    ASSERT_OK(env_->GetFileSystem()->NewRandomAccessFile(
+        file_name, FileOptions(), &file, nullptr));
+    std::unique_ptr<RandomAccessFileReader> file_reader;
+    file_reader.reset(new RandomAccessFileReader(std::move(file), file_name));
+
+    std::string scratch(size, '\0');
+    Slice result;
+    ASSERT_OK(
+        file_reader->Read(IOOptions(), offset, size, &result, scratch.data()));
+    ASSERT_EQ(result.size(), size);
+    bytes->assign(result.data(), result.size());
+  }
+
  protected:
   Options options_;
   EnvOptions soptions_;
@@ -109,6 +155,60 @@ class SstFileReaderTest : public testing::Test {
   Env* env_;
 };
 
+class FailingAppendWritableFile : public FSWritableFileOwnerWrapper {
+ public:
+  FailingAppendWritableFile(std::unique_ptr<FSWritableFile>&& target,
+                            std::atomic<bool>* fail_writes)
+      : FSWritableFileOwnerWrapper(std::move(target)),
+        fail_writes_(fail_writes) {}
+
+  IOStatus Append(const Slice& data, const IOOptions& options,
+                  IODebugContext* dbg) override {
+    if (fail_writes_->load()) {
+      return IOStatus::IOError("Injected embedded blob append failure");
+    }
+    return FSWritableFileOwnerWrapper::Append(data, options, dbg);
+  }
+
+  IOStatus Append(const Slice& data, const IOOptions& options,
+                  const DataVerificationInfo& verification_info,
+                  IODebugContext* dbg) override {
+    if (fail_writes_->load()) {
+      return IOStatus::IOError("Injected embedded blob append failure");
+    }
+    return FSWritableFileOwnerWrapper::Append(data, options, verification_info,
+                                              dbg);
+  }
+
+ private:
+  std::atomic<bool>* fail_writes_;
+};
+
+class FailingAppendFileSystem : public FileSystemWrapper {
+ public:
+  explicit FailingAppendFileSystem(const std::shared_ptr<FileSystem>& target)
+      : FileSystemWrapper(target) {}
+
+  const char* Name() const override { return "FailingAppendFileSystem"; }
+
+  IOStatus NewWritableFile(const std::string& fname,
+                           const FileOptions& file_opts,
+                           std::unique_ptr<FSWritableFile>* result,
+                           IODebugContext* dbg) override {
+    IOStatus s = target()->NewWritableFile(fname, file_opts, result, dbg);
+    if (s.ok()) {
+      result->reset(
+          new FailingAppendWritableFile(std::move(*result), &fail_writes_));
+    }
+    return s;
+  }
+
+  void SetFailWrites(bool fail_writes) { fail_writes_.store(fail_writes); }
+
+ private:
+  std::atomic<bool> fail_writes_{false};
+};
+
 const uint64_t kNumKeys = 100;
 
 TEST_F(SstFileReaderTest, Basic) {
@@ -119,6 +219,394 @@ TEST_F(SstFileReaderTest, Basic) {
   CreateFileAndCheck(keys);
 }
 
+TEST_F(SstFileReaderTest, EmbeddedBlobRoundTrip) {
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 6;
+  const std::string embedded_value(4096, 'v');
+  const std::string embedded_default_column_value(4096, 'd');
+
+  SstFileWriter writer(soptions_, options_);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_name_, embedded_blob_options));
+  ASSERT_OK(writer.Put("a", "tiny"));
+  ASSERT_OK(writer.Put("b", embedded_value));
+  ASSERT_OK(writer.PutEntity(
+      "c", {{kDefaultWideColumnName, embedded_default_column_value},
+            {"meta", "x"}}));
+
+  ExternalSstFileInfo file_info;
+  ASSERT_OK(writer.Finish(&file_info));
+  ASSERT_GT(file_info.file_size, 0);
+
+  SstFileReader reader(options_);
+  ASSERT_OK(reader.Open(sst_name_));
+  ASSERT_OK(reader.VerifyChecksum());
+
+  std::string value;
+  ASSERT_OK(reader.Get(ReadOptions(), "a", &value));
+  ASSERT_EQ(value, "tiny");
+  ASSERT_OK(reader.Get(ReadOptions(), "b", &value));
+  ASSERT_EQ(value, embedded_value);
+  ASSERT_OK(reader.Get(ReadOptions(), "c", &value));
+  ASSERT_EQ(value, embedded_default_column_value);
+
+  std::vector<std::string> key_storage = {"a", "b", "c"};
+  std::vector<Slice> keys;
+  keys.reserve(key_storage.size());
+  for (const std::string& key : key_storage) {
+    keys.emplace_back(key);
+  }
+  std::vector<std::string> values;
+  std::vector<Status> statuses = reader.MultiGet(ReadOptions(), keys, &values);
+  ASSERT_EQ(statuses.size(), keys.size());
+  ASSERT_EQ(values.size(), keys.size());
+  for (const Status& status : statuses) {
+    ASSERT_OK(status);
+  }
+  EXPECT_EQ(values[0], "tiny");
+  EXPECT_EQ(values[1], embedded_value);
+  EXPECT_EQ(values[2], embedded_default_column_value);
+
+  std::unique_ptr<Iterator> iter(reader.NewIterator(ReadOptions()));
+  iter->SeekToFirst();
+  ASSERT_TRUE(iter->Valid());
+  EXPECT_EQ(iter->key(), "a");
+  EXPECT_EQ(iter->value(), "tiny");
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  EXPECT_EQ(iter->key(), "b");
+  EXPECT_EQ(iter->value(), embedded_value);
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  EXPECT_EQ(iter->key(), "c");
+  EXPECT_EQ(iter->value(), embedded_default_column_value);
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_OK(iter->status());
+
+  // Embedded-blob SSTs disable index value delta encoding so blob records can
+  // be written interleaved with data blocks.
+  std::shared_ptr<const TableProperties> props = reader.GetTableProperties();
+  ASSERT_NE(props, nullptr);
+  EXPECT_EQ(props->index_value_is_delta_encoded, 0);
+
+  EmbeddedBlobStats stats;
+  GetEmbeddedBlobStats(&reader, &stats);
+  EXPECT_EQ(stats.blob_count, 2);
+  EXPECT_GT(stats.payload_bytes, 0);
+}
+
+// Invariant: the internal block-based table iterator must never expose an
+// unresolved same-file kTypeBlobIndex internal key through key(). With
+// index_type=kBinarySearchWithFirstKey and allow_unprepared_value, the iterator
+// can sit in the is_at_first_key_from_index_ state and return the raw first key
+// from the index (type kTypeBlobIndex) while value() resolves the same-file
+// blob to its plain payload. DBIter happens to tolerate this (it re-parses the
+// key after PrepareValue), but other internal-iterator consumers must not see
+// an unresolved same-file blob index. Forcing one entry per data block makes
+// every embedded blob the first key of a block, exercising the deferred path on
+// SeekToFirst, forward block transitions, and Seek.
+TEST_F(SstFileReaderTest, EmbeddedBlobIteratorKeyTypeWithFirstKeyIndex) {
+  BlockBasedTableOptions bbto;
+  bbto.index_type =
+      BlockBasedTableOptions::IndexType::kBinarySearchWithFirstKey;
+  // Soft limit of 1 byte starts a new data block after every entry.
+  bbto.block_size = 1;
+  options_.table_factory.reset(NewBlockBasedTableFactory(bbto));
+
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 8;
+  const std::string big1(1024, 'x');
+  const std::string big2(1024, 'y');
+  const std::string big3(1024, 'z');
+
+  SstFileWriter writer(soptions_, options_);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_name_, embedded_blob_options));
+  ASSERT_OK(writer.Put("k1", big1));
+  ASSERT_OK(writer.Put("k2", big2));
+  ASSERT_OK(writer.Put("k3", big3));
+  ASSERT_OK(writer.Finish());
+
+  // Open a BlockBasedTable reader directly so we can request an internal
+  // iterator with allow_unprepared_value=true. The public SstFileReader
+  // iterator uses allow_unprepared_value=false, which never defers.
+  ImmutableOptions ioptions(options_);
+  MutableCFOptions moptions(options_);
+  InternalKeyComparator icomp(options_.comparator);
+
+  uint64_t file_size = 0;
+  ASSERT_OK(env_->GetFileSize(sst_name_, &file_size));
+
+  std::unique_ptr<FSRandomAccessFile> file;
+  ASSERT_OK(env_->GetFileSystem()->NewRandomAccessFile(sst_name_, FileOptions(),
+                                                       &file, nullptr));
+  std::unique_ptr<RandomAccessFileReader> file_reader(
+      new RandomAccessFileReader(std::move(file), sst_name_));
+
+  ReadOptions read_opts;
+  read_opts.verify_checksums = true;
+  TableReaderOptions table_reader_options(
+      ioptions, moptions.prefix_extractor, moptions.compression_manager.get(),
+      soptions_, icomp, /*block_protection_bytes_per_key=*/0);
+  std::unique_ptr<TableReader> table_reader;
+  ASSERT_OK(options_.table_factory->NewTableReader(
+      read_opts, table_reader_options, std::move(file_reader), file_size,
+      &table_reader, /*prefetch_index_and_filter_in_cache=*/true));
+
+  std::unique_ptr<InternalIterator> iter(table_reader->NewIterator(
+      read_opts, moptions.prefix_extractor.get(), /*arena=*/nullptr,
+      /*skip_filters=*/false, TableReaderCaller::kUncategorized,
+      /*compaction_readahead_size=*/0, /*allow_unprepared_value=*/true));
+
+  auto check_current = [&](const std::string& user_key,
+                           const std::string& value) {
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    // key() must already present the resolved value type, never the raw
+    // same-file blob index type.
+    ParsedInternalKey parsed;
+    ASSERT_OK(ParseInternalKey(iter->key(), &parsed, /*log_err_key=*/true));
+    EXPECT_EQ(parsed.user_key, user_key);
+    EXPECT_NE(parsed.type, kTypeBlobIndex);
+    EXPECT_EQ(parsed.type, kTypeValue);
+    ASSERT_TRUE(iter->PrepareValue());
+    EXPECT_EQ(iter->value(), value);
+  };
+
+  const std::vector<std::pair<std::string, std::string>> expected = {
+      {"k1", big1}, {"k2", big2}, {"k3", big3}};
+
+  size_t idx = 0;
+  for (iter->SeekToFirst(); iter->Valid(); iter->Next(), ++idx) {
+    ASSERT_LT(idx, expected.size());
+    check_current(expected[idx].first, expected[idx].second);
+  }
+  ASSERT_OK(iter->status());
+  EXPECT_EQ(idx, expected.size());
+
+  // Seek directly onto a block whose first key is an embedded blob.
+  InternalKey seek_target("k2", kMaxSequenceNumber, kValueTypeForSeek);
+  iter->Seek(seek_target.Encode());
+  check_current("k2", big2);
+}
+
+TEST_F(SstFileReaderTest, EmbeddedBlobRequiresFormatVersion7) {
+  Options options = options_;
+  BlockBasedTableOptions table_options;
+  table_options.format_version = 6;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  SstFileWriter writer(soptions_, options);
+  Status s = writer.OpenWithEmbeddedBlobs(sst_name_, embedded_blob_options);
+  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+}
+
+TEST_F(SstFileReaderTest, EmbeddedBlobCompressionOptionsIgnored) {
+  const std::string value(16 * 1024, 'x');
+
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 1;
+  embedded_blob_options.compression_type = kSnappyCompression;
+  embedded_blob_options.compression_options.SetMinRatio(100.0);
+
+  SstFileWriter writer(soptions_, options_);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_name_, embedded_blob_options));
+  ASSERT_OK(writer.Put("a", value));
+  ASSERT_OK(writer.Finish());
+
+  SstFileReader reader(options_);
+  ASSERT_OK(reader.Open(sst_name_));
+
+  std::string read_value;
+  ASSERT_OK(reader.Get(ReadOptions(), "a", &read_value));
+  ASSERT_EQ(read_value, value);
+
+  EmbeddedBlobStats stats;
+  GetEmbeddedBlobStats(&reader, &stats);
+  EXPECT_EQ(stats.blob_count, 1);
+  EXPECT_EQ(stats.payload_bytes, value.size());
+}
+
+TEST_F(SstFileReaderTest, EmbeddedBlobDefaultMinBlobSize) {
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  const std::string small_value(2047, 's');
+  const std::string large_value(2048, 'l');
+
+  SstFileWriter writer(soptions_, options_);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_name_, embedded_blob_options));
+  ASSERT_OK(writer.Put("a", small_value));
+  ASSERT_OK(writer.Put("b", large_value));
+  ASSERT_OK(writer.Finish());
+
+  SstFileReader reader(options_);
+  ASSERT_OK(reader.Open(sst_name_));
+
+  std::string read_value;
+  ASSERT_OK(reader.Get(ReadOptions(), "a", &read_value));
+  ASSERT_EQ(read_value, small_value);
+  ASSERT_OK(reader.Get(ReadOptions(), "b", &read_value));
+  ASSERT_EQ(read_value, large_value);
+
+  EmbeddedBlobStats stats;
+  GetEmbeddedBlobStats(&reader, &stats);
+  EXPECT_EQ(stats.blob_count, 1);
+  EXPECT_EQ(stats.payload_bytes, large_value.size());
+}
+
+TEST_F(SstFileReaderTest, EmbeddedBlobInterleavedLayout) {
+  // Force one entry per data block so blob records (written inline as values
+  // are added) end up interleaved between data blocks rather than buffered in a
+  // strict front prefix.
+  BlockBasedTableOptions bbto;
+  bbto.block_size = 1;
+  options_.table_factory.reset(NewBlockBasedTableFactory(bbto));
+
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 64;
+  const std::string large0(4096, '0');
+  const std::string large1(4096, '1');
+  const std::string large2(4096, '2');
+
+  SstFileWriter writer(soptions_, options_);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_name_, embedded_blob_options));
+  // The first entry is small, so data block 0 is flushed (at file offset 0)
+  // before any blob record is written.
+  ASSERT_OK(writer.Put("a", "tiny"));
+  ASSERT_OK(writer.Put("b", large0));
+  ASSERT_OK(writer.Put("c", "tiny"));
+  ASSERT_OK(writer.Put("d", large1));
+  ASSERT_OK(writer.Put("e", "tiny"));
+  ASSERT_OK(writer.Put("f", large2));
+  ASSERT_OK(writer.Finish());
+
+  SstFileReader reader(options_);
+  ASSERT_OK(reader.Open(sst_name_));
+  ASSERT_OK(reader.VerifyChecksum());
+
+  // Index value delta encoding is off for embedded-blob SSTs (the mechanism
+  // that allows interleaving blob records with data blocks), and there is more
+  // than one data block.
+  std::shared_ptr<const TableProperties> props = reader.GetTableProperties();
+  ASSERT_NE(props, nullptr);
+  EXPECT_EQ(props->index_value_is_delta_encoded, 0);
+  EXPECT_GT(props->num_data_blocks, 1);
+
+  // The blob records are interleaved with data blocks rather than buffered as a
+  // contiguous front prefix: immediately after the first blob record sits a
+  // data block, not the second blob's payload (which is what a strict prefix
+  // layout would place there).
+  const size_t first_record_size = large0.size() + kSimpleGen2BlobTrailerSize;
+  std::string after_first_record;
+  ReadFileBytes(sst_name_, first_record_size, 32, &after_first_record);
+  EXPECT_NE(after_first_record, std::string(32, '1'));
+
+  EmbeddedBlobStats stats;
+  GetEmbeddedBlobStats(&reader, &stats);
+  EXPECT_EQ(stats.blob_count, 3);
+  EXPECT_EQ(stats.payload_bytes, large0.size() + large1.size() + large2.size());
+
+  // All values, large and small, read back correctly.
+  const std::vector<std::pair<std::string, std::string>> expected = {
+      {"a", "tiny"}, {"b", large0}, {"c", "tiny"},
+      {"d", large1}, {"e", "tiny"}, {"f", large2}};
+  std::string value;
+  for (const auto& kv : expected) {
+    ASSERT_OK(reader.Get(ReadOptions(), kv.first, &value));
+    EXPECT_EQ(value, kv.second);
+  }
+
+  std::unique_ptr<Iterator> iter(reader.NewIterator(ReadOptions()));
+  size_t idx = 0;
+  for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+    ASSERT_LT(idx, expected.size());
+    EXPECT_EQ(iter->key(), expected[idx].first);
+    EXPECT_EQ(iter->value(), expected[idx].second);
+    ++idx;
+  }
+  ASSERT_OK(iter->status());
+  EXPECT_EQ(idx, expected.size());
+}
+
+TEST_F(SstFileReaderTest, EmbeddedBlobRecordCorruptionDetected) {
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 1;
+  const std::string value(4096, 'v');
+
+  SstFileWriter writer(soptions_, options_);
+  ASSERT_OK(writer.OpenWithEmbeddedBlobs(sst_name_, embedded_blob_options));
+  // A single large value: its blob record is written first, at file offset 0,
+  // so byte 100 falls within the blob payload.
+  ASSERT_OK(writer.Put("a", value));
+  ASSERT_OK(writer.Finish());
+
+  // Flip a byte inside the embedded blob record's payload. The offset-keyed
+  // record checksum is now the only guard against this (no range pre-check).
+  {
+    std::unique_ptr<RandomRWFile> rw_file;
+    ASSERT_OK(env_->NewRandomRWFile(sst_name_, &rw_file, EnvOptions()));
+    char scratch = 0;
+    Slice chunk;
+    ASSERT_OK(rw_file->Read(100, 1, &chunk, &scratch));
+    ASSERT_EQ(chunk.size(), 1);
+    char corrupted = static_cast<char>(chunk[0] ^ 0xff);
+    ASSERT_OK(rw_file->Write(100, Slice(&corrupted, 1)));
+    ASSERT_OK(rw_file->Close());
+  }
+
+  SstFileReader reader(options_);
+  ASSERT_OK(reader.Open(sst_name_));
+
+  std::string read_value;
+  Status s = reader.Get(ReadOptions(), "a", &read_value);
+  EXPECT_TRUE(s.IsCorruption()) << s.ToString();
+}
+
+TEST_F(SstFileReaderTest, EmbeddedBlobAppendErrorsSurfaceEarly) {
+  std::shared_ptr<FailingAppendFileSystem> fs(
+      new FailingAppendFileSystem(env_->GetFileSystem()));
+  std::unique_ptr<Env> failing_env(new CompositeEnvWrapper(env_, fs));
+
+  Options options = options_;
+  options.env = failing_env.get();
+  EnvOptions env_options = soptions_;
+  env_options.writable_file_max_buffer_size = 1;
+
+  SstFileWriterEmbeddedBlobOptions embedded_blob_options;
+  embedded_blob_options.min_blob_size = 1;
+
+  const std::string put_file = sst_name_ + "_put";
+  const std::string entity_file = sst_name_ + "_entity";
+  const std::string value(4096, 'v');
+
+  {
+    SstFileWriter writer(env_options, options);
+    ASSERT_OK(writer.OpenWithEmbeddedBlobs(put_file, embedded_blob_options));
+    fs->SetFailWrites(true);
+    Status s = writer.Put("a", value);
+    ASSERT_TRUE(s.IsIOError()) << s.ToString();
+    s = writer.Finish();
+    ASSERT_TRUE(s.IsIOError()) << s.ToString();
+    fs->SetFailWrites(false);
+  }
+
+  {
+    SstFileWriter writer(env_options, options);
+    ASSERT_OK(writer.OpenWithEmbeddedBlobs(entity_file, embedded_blob_options));
+    fs->SetFailWrites(true);
+    Status s =
+        writer.PutEntity("a", {{kDefaultWideColumnName, value}, {"meta", "m"}});
+    ASSERT_TRUE(s.IsIOError()) << s.ToString();
+    s = writer.Finish();
+    ASSERT_TRUE(s.IsIOError()) << s.ToString();
+    fs->SetFailWrites(false);
+  }
+
+  Status s = env_->DeleteFile(put_file);
+  s.PermitUncheckedError();
+  s = env_->DeleteFile(entity_file);
+  s.PermitUncheckedError();
+}
+
 TEST_F(SstFileReaderTest, ParseTableIteratorKey) {
   // Verify that callers of the raw table iterator can decode key metadata
   // through the public SstFileReader API instead of duplicating dbformat.h.

table/sst_file_writer.cc

@@ -16,6 +16,8 @@
 #include "rocksdb/file_system.h"
 #include "rocksdb/table.h"
 #include "table/block_based/block_based_table_builder.h"
+#include "table/embedded_blob_sst.h"
+#include "table/format.h"
 #include "table/prepared_file_info.h"
 #include "table/sst_file_writer_collectors.h"
 #include "test_util/sync_point.h"
@@ -75,6 +77,7 @@ struct SstFileWriter::Rep {
   uint64_t next_file_number = 1;
   size_t ts_sz;
   bool strip_timestamp;
+  std::unique_ptr<EmbeddedBlobSstBuilderOptions> embedded_blob_options;
 
   Status AddImpl(const Slice& user_key, const Slice& value,
                  ValueType value_type) {
@@ -416,7 +419,8 @@ Status SstFileWriter::Open(const std::string& file_path, Temperature temp) {
       unknown_level, kUnknownNewestKeyTime, false /* is_bottommost */,
       TableFileCreationReason::kMisc, 0 /* oldest_key_time */,
       0 /* file_creation_time */, "SST Writer" /* db_id */, r->db_session_id,
-      0 /* target_file_size */, r->next_file_number);
+      0 /* target_file_size */, r->next_file_number, kMaxSequenceNumber,
+      r->embedded_blob_options.get());
   // External SST files used to each get a unique session id. Now for
   // slightly better uniqueness probability in constructing cache keys, we
   // assign fake file numbers to each file (into table properties) and keep
@@ -441,6 +445,37 @@ Status SstFileWriter::Open(const std::string& file_path, Temperature temp) {
   return s;
 }
 
+Status SstFileWriter::OpenWithEmbeddedBlobs(
+    const std::string& file_path,
+    const SstFileWriterEmbeddedBlobOptions& embedded_blob_options,
+    Temperature temp) {
+  Rep* r = rep_.get();
+  if (r->builder) {
+    return Status::InvalidArgument("File is already opened");
+  }
+
+  const BlockBasedTableOptions* const table_options =
+      r->mutable_cf_options.table_factory == nullptr
+          ? nullptr
+          : r->mutable_cf_options.table_factory
+                ->GetOptions<BlockBasedTableOptions>();
+  if (table_options == nullptr) {
+    return Status::InvalidArgument(
+        "Embedded blob SSTs require block-based table format");
+  }
+  if (!FormatVersionUsesCompressionManagerName(table_options->format_version)) {
+    return Status::InvalidArgument(
+        "Embedded blob SSTs require block-based table format_version >= 7");
+  }
+
+  r->embedded_blob_options.reset(
+      new EmbeddedBlobSstBuilderOptions(embedded_blob_options));
+
+  Status s = Open(file_path, temp);
+  r->embedded_blob_options.reset();
+  return s;
+}
+
 Status SstFileWriter::Put(const Slice& user_key, const Slice& value) {
   return rep_->Add(user_key, value, ValueType::kTypeValue);
 }

table/table_builder.h

@@ -22,6 +22,7 @@
 #include "options/cf_options.h"
 #include "rocksdb/options.h"
 #include "rocksdb/table_properties.h"
+#include "table/embedded_blob_sst.h"
 #include "table/unique_id_impl.h"
 #include "trace_replay/block_cache_tracer.h"
 #include "util/cast_util.h"
@@ -30,6 +31,7 @@ namespace ROCKSDB_NAMESPACE {
 
 class Slice;
 class Status;
+class BlobSource;
 
 struct TableReaderOptions {
   // @param skip_filters Disables loading/accessing the filter block
@@ -109,6 +111,14 @@ struct TableReaderOptions {
   // Open-time metadata reads should not insert index/filter/dictionary blocks
   // into the shared block cache.
   bool avoid_shared_metadata_cache;
+
+  // Blob source for routing same-file ("embedded") blob reads through the blob
+  // value cache + BLOB_DB_* statistics. Owned by the ColumnFamilyData; the
+  // table reader's lifetime is a subset of the CFD's (it is owned via the CFD's
+  // TableCache), so a raw pointer is lifetime-safe. nullptr for non-DB openers
+  // (SstFileReader, sst_dump, repair, external-file ingestion prevalidation,
+  // etc.), in which case embedded reads fall back to a direct (uncached) read.
+  BlobSource* blob_source = nullptr;
 };
 
 struct TableBuilderOptions : public TablePropertiesCollectorFactory::Context {
@@ -127,7 +137,8 @@ struct TableBuilderOptions : public TablePropertiesCollectorFactory::Context {
       const std::string& _db_session_id = "",
       const uint64_t _target_file_size = 0, const uint64_t _cur_file_num = 0,
       const SequenceNumber _last_level_inclusive_max_seqno_threshold =
-          kMaxSequenceNumber)
+          kMaxSequenceNumber,
+      const EmbeddedBlobSstBuilderOptions* _embedded_blob_options = nullptr)
       : TablePropertiesCollectorFactory::Context(
             _column_family_id, _level, _ioptions.num_levels,
             _last_level_inclusive_max_seqno_threshold),
@@ -148,7 +159,8 @@ struct TableBuilderOptions : public TablePropertiesCollectorFactory::Context {
         db_session_id(_db_session_id),
         is_bottommost(_is_bottommost),
         reason(_reason),
-        cur_file_num(_cur_file_num) {}
+        cur_file_num(_cur_file_num),
+        embedded_blob_options(_embedded_blob_options) {}
 
   const ImmutableOptions& ioptions;
   const MutableCFOptions& moptions;
@@ -171,6 +183,10 @@ struct TableBuilderOptions : public TablePropertiesCollectorFactory::Context {
   // END for FilterBuildingContext
 
   const uint64_t cur_file_num;
+  // Non-null only for table builders that should write eligible large values as
+  // same-file ("embedded") blob records. Currently only honored by
+  // BlockBasedTableBuilder.
+  const EmbeddedBlobSstBuilderOptions* embedded_blob_options;
 };
 
 // TableBuilder provides the interface used to build a Table

tools/db_crashtest.py

@@ -221,6 +221,7 @@ default_params = {
     "ingest_external_file_one_in": lambda: random.choice([1000, 1000000]),
     "ingest_external_file_prepare_commit_one_in": lambda: random.choice([0, 1, 2]),
     "ingest_external_file_use_file_info_one_in": lambda: random.choice([0, 1, 2]),
+    "ingest_external_file_with_embedded_blobs": lambda: random.choice([0, 1]),
     "test_ingest_standalone_range_deletion_one_in": lambda: random.choice([0, 5, 10]),
     "iterpercent": 10,
     "lock_wal_one_in": lambda: random.choice([10000, 1000000]),
@@ -1503,6 +1504,13 @@ def finalize_and_sanitize(src_params):
         or dest_params.get("delrangepercent") == 0
     ):
         dest_params["test_ingest_standalone_range_deletion_one_in"] = 0
+    # Embedded blobs in ingested files require ingestion to be enabled and
+    # block-based table format_version >= 7.
+    if (
+        dest_params.get("ingest_external_file_one_in") == 0
+        or dest_params.get("format_version", 2) < 7
+    ):
+        dest_params["ingest_external_file_with_embedded_blobs"] = 0
     if (
         dest_params.get("use_txn", 0) == 1
         and dest_params.get("commit_bypass_memtable_one_in", 0) > 0

unreleased_history/new_features/embedded_blob_sst_writer.md

@@ -0,0 +1 @@
+Added EXPERIMENTAL embedded blob SST support through `SstFileWriter::OpenWithEmbeddedBlobs()`, storing eligible large values as same-file blob records in block-based SST files and resolving them transparently for reads. This niche feature currently supports uncompressed embedded blobs only; compression options are placeholders and compression support is deferred to follow-up work.