Merge remote-tracking branch 'upstream/main' into 2026_04_07_ws_bdw

# Conflicts:
#	db/blob/db_blob_direct_write_test.cc

Author: xingbowang

.gitignore

@@ -105,3 +105,5 @@ third-party/folly/
 
 # Claude Code local settings
 .claude/settings.local.json
+
+tools/__pycache__/

BUCK

@@ -111,6 +111,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
         "db/version_util.cc",
         "db/wal_edit.cc",
         "db/wal_manager.cc",
+        "db/wide/read_path_blob_resolver.cc",
         "db/wide/wide_column_serialization.cc",
         "db/wide/wide_columns.cc",
         "db/wide/wide_columns_helper.cc",
@@ -4951,6 +4952,12 @@ cpp_unittest_wrapper(name="db_merge_operator_test",
             extra_compiler_flags=[])
 
 
+cpp_unittest_wrapper(name="db_open_with_config_test",
+            srcs=["db/db_open_with_config_test.cc"],
+            deps=[":rocksdb_test_lib"],
+            extra_compiler_flags=[])
+
+
 cpp_unittest_wrapper(name="db_options_test",
             srcs=["db/db_options_test.cc"],
             deps=[":rocksdb_test_lib"],
@@ -5041,6 +5048,12 @@ cpp_unittest_wrapper(name="db_wide_basic_test",
             extra_compiler_flags=[])
 
 
+cpp_unittest_wrapper(name="db_wide_blob_direct_write_test",
+            srcs=["db/wide/db_wide_blob_direct_write_test.cc"],
+            deps=[":rocksdb_test_lib"],
+            extra_compiler_flags=[])
+
+
 cpp_unittest_wrapper(name="db_with_timestamp_basic_test",
             srcs=["db/db_with_timestamp_basic_test.cc"],
             deps=[":rocksdb_test_lib"],

CMakeLists.txt

@@ -788,6 +788,7 @@ set(SOURCES
         db/version_util.cc
         db/wal_edit.cc
         db/wal_manager.cc
+        db/wide/read_path_blob_resolver.cc
         db/wide/wide_column_serialization.cc
         db/wide/wide_columns.cc
         db/wide/wide_columns_helper.cc
@@ -1427,6 +1428,7 @@ if(WITH_TESTS)
         db/db_memtable_test.cc
         db/db_merge_operator_test.cc
         db/db_merge_operand_test.cc
+        db/db_open_with_config_test.cc
         db/db_options_test.cc
         db/db_properties_test.cc
         db/db_range_del_test.cc
@@ -1479,6 +1481,7 @@ if(WITH_TESTS)
         db/wal_manager_test.cc
         db/wal_edit_test.cc
         db/wide/db_wide_basic_test.cc
+        db/wide/db_wide_blob_direct_write_test.cc
         db/wide/wide_column_serialization_test.cc
         db/wide/wide_columns_helper_test.cc
         db/write_batch_test.cc

Makefile

@@ -1496,6 +1496,9 @@ db_readonly_with_timestamp_test: $(OBJ_DIR)/db/db_readonly_with_timestamp_test.o
 db_wide_basic_test: $(OBJ_DIR)/db/wide/db_wide_basic_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
+db_wide_blob_direct_write_test: $(OBJ_DIR)/db/wide/db_wide_blob_direct_write_test.o $(TEST_LIBRARY) $(LIBRARY)
+	$(AM_LINK)
+
 db_with_timestamp_basic_test: $(OBJ_DIR)/db/db_with_timestamp_basic_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
@@ -1505,6 +1508,9 @@ db_with_timestamp_compaction_test: db/db_with_timestamp_compaction_test.o $(TEST
 db_encryption_test: $(OBJ_DIR)/db/db_encryption_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
+db_open_with_config_test: $(OBJ_DIR)/db/db_open_with_config_test.o $(TEST_LIBRARY) $(LIBRARY)
+	$(AM_LINK)
+
 db_test: $(OBJ_DIR)/db/db_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 

db/blob/blob_fetcher.cc

@@ -5,6 +5,8 @@
 
 #include "db/blob/blob_fetcher.h"
 
+#include "db/blob/blob_file_partition_manager.h"
+#include "db/blob/blob_index.h"
 #include "db/version_set.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -14,21 +16,30 @@ Status BlobFetcher::FetchBlob(const Slice& user_key,
                               FilePrefetchBuffer* prefetch_buffer,
                               PinnableSlice* blob_value,
                               uint64_t* bytes_read) const {
-  assert(version_);
-
-  return version_->GetBlob(read_options_, user_key, blob_index_slice,
-                           prefetch_buffer, blob_value, bytes_read);
+  BlobIndex blob_index;
+  Status status = blob_index.DecodeFrom(blob_index_slice);
+  if (status.ok()) {
+    status = FetchBlob(user_key, blob_index, prefetch_buffer, blob_value,
+                       bytes_read);
+  }
+  return status;
 }
 
 Status BlobFetcher::FetchBlob(const Slice& user_key,
                               const BlobIndex& blob_index,
                               FilePrefetchBuffer* prefetch_buffer,
                               PinnableSlice* blob_value,
                               uint64_t* bytes_read) const {
-  assert(version_);
+  if (!allow_write_path_fallback_) {
+    assert(version_);
+
+    return version_->GetBlob(read_options_, user_key, blob_index,
+                             prefetch_buffer, blob_value, bytes_read);
+  }
 
-  return version_->GetBlob(read_options_, user_key, blob_index, prefetch_buffer,
-                           blob_value, bytes_read);
+  return BlobFilePartitionManager::ResolveBlobDirectWriteIndex(
+      read_options_, user_key, blob_index, version_, blob_file_cache_,
+      prefetch_buffer, blob_value, bytes_read);
 }
 
 }  // namespace ROCKSDB_NAMESPACE

db/blob/blob_fetcher.h

@@ -10,17 +10,25 @@
 
 namespace ROCKSDB_NAMESPACE {
 
+class BlobFileCache;
 class Version;
 class Slice;
 class FilePrefetchBuffer;
 class PinnableSlice;
 class BlobIndex;
 
-// A thin wrapper around the blob retrieval functionality of Version.
+// A thin wrapper around blob retrieval. By default it reads through Version,
+// and it can optionally fall back to direct-write blob files that are not yet
+// manifest-visible.
 class BlobFetcher {
  public:
-  BlobFetcher(const Version* version, const ReadOptions& read_options)
-      : version_(version), read_options_(read_options) {}
+  BlobFetcher(const Version* version, const ReadOptions& read_options,
+              BlobFileCache* blob_file_cache = nullptr,
+              bool allow_write_path_fallback = false)
+      : version_(version),
+        read_options_(read_options),
+        blob_file_cache_(blob_file_cache),
+        allow_write_path_fallback_(allow_write_path_fallback) {}
 
   Status FetchBlob(const Slice& user_key, const Slice& blob_index_slice,
                    FilePrefetchBuffer* prefetch_buffer,
@@ -33,5 +41,7 @@ class BlobFetcher {
  private:
   const Version* version_;
   ReadOptions read_options_;
+  BlobFileCache* blob_file_cache_;
+  bool allow_write_path_fallback_;
 };
 }  // namespace ROCKSDB_NAMESPACE

db/blob/blob_file_partition_manager.cc

@@ -35,6 +35,8 @@ namespace {
 
 class RoundRobinBlobFilePartitionStrategy : public BlobFilePartitionStrategy {
  public:
+  using BlobFilePartitionStrategy::SelectPartition;
+
   const char* Name() const override {
     return "RoundRobinBlobFilePartitionStrategy";
   }
@@ -408,11 +410,19 @@ Status BlobFilePartitionManager::MaybePrepopulateBlobCache(
                                 CacheTier::kVolatileTier);
 }
 
+uint32_t BlobFilePartitionManager::SelectWideColumnPartition(
+    uint32_t column_family_id, const Slice& key,
+    const WideColumns& columns) const {
+  return strategy_->SelectPartition(num_partitions_, column_family_id, key,
+                                    columns) %
+         num_partitions_;
+}
+
 Status BlobFilePartitionManager::WriteBlob(
     const WriteOptions& write_options, uint32_t column_family_id,
     CompressionType compression, const Slice& key, const Slice& value,
     uint64_t* blob_file_number, uint64_t* blob_offset, uint64_t* blob_size,
-    const BlobDirectWriteSettings* settings) {
+    const BlobDirectWriteSettings* settings, const uint32_t* partition_idx) {
   assert(blob_file_number != nullptr);
   assert(blob_offset != nullptr);
   assert(blob_size != nullptr);
@@ -443,14 +453,16 @@ Status BlobFilePartitionManager::WriteBlob(
   // uncompressed value rather than `write_value`. The modulo here is
   // intentional so custom strategies can return arbitrary hashed or sentinel
   // values without violating the partition bounds.
-  const uint32_t partition_idx =
-      strategy_->SelectPartition(num_partitions_, column_family_id, key,
-                                 value) %
-      num_partitions_;
+  const uint32_t selected_partition_idx =
+      partition_idx != nullptr
+          ? (*partition_idx % num_partitions_)
+          : (strategy_->SelectPartition(num_partitions_, column_family_id, key,
+                                        value) %
+             num_partitions_);
 
   {
     MutexLock lock(&mutex_);
-    Partition* partition = partitions_[partition_idx].get();
+    Partition* partition = partitions_[selected_partition_idx].get();
 
     auto seal_current_file = [&]() -> Status {
       if (!partition->writer) {
@@ -481,7 +493,7 @@ Status BlobFilePartitionManager::WriteBlob(
 
     if (!partition->writer) {
       Status s = OpenNewBlobFile(partition, column_family_id, compression,
-                                 partition_idx);
+                                 selected_partition_idx);
       if (!s.ok()) {
         return s;
       }
@@ -681,6 +693,14 @@ void BlobFilePartitionManager::GetProtectedBlobFileNumbers(
   }
 }
 
+bool BlobFilePartitionManager::IsTrackedBlobFileNumber(
+    uint64_t file_number) const {
+  ReadLock lock(&file_partition_mutex_);
+  return file_to_partition_.find(file_number) != file_to_partition_.end() ||
+         protected_blob_file_refs_.find(file_number) !=
+             protected_blob_file_refs_.end();
+}
+
 void BlobFilePartitionManager::ProtectSealedBlobFileNumbers(
     const std::vector<uint64_t>& file_numbers) {
   if (file_numbers.empty()) {
@@ -747,12 +767,10 @@ void BlobFilePartitionManager::RemoveFilePartitionMappings(
 Status BlobFilePartitionManager::ResolveBlobDirectWriteIndex(
     const ReadOptions& read_options, const Slice& user_key,
     const BlobIndex& blob_idx, const Version* version,
-    BlobFileCache* blob_file_cache, PinnableSlice* blob_value) {
+    BlobFileCache* blob_file_cache, FilePrefetchBuffer* prefetch_buffer,
+    PinnableSlice* blob_value, uint64_t* bytes_read) {
   assert(blob_value != nullptr);
 
-  constexpr FilePrefetchBuffer* prefetch_buffer = nullptr;
-  constexpr uint64_t* bytes_read = nullptr;
-
   if (version != nullptr) {
     // Only fall back when the blob file is still owned exclusively by the
     // write path and therefore absent from Version metadata. Once Version

db/blob/blob_file_partition_manager.h

@@ -31,6 +31,7 @@ class BlobFileCache;
 class BlobFileCompletionCallback;
 class BlobIndex;
 class BlobLogWriter;
+class FilePrefetchBuffer;
 class IOTracer;
 class Logger;
 class PinnableSlice;
@@ -76,7 +77,14 @@ class BlobFilePartitionManager {
                    CompressionType compression, const Slice& key,
                    const Slice& value, uint64_t* blob_file_number,
                    uint64_t* blob_offset, uint64_t* blob_size,
-                   const BlobDirectWriteSettings* settings = nullptr);
+                   const BlobDirectWriteSettings* settings = nullptr,
+                   const uint32_t* partition_idx = nullptr);
+
+  // Selects the partition to use for all blob-backed columns of one PutEntity
+  // operation. The return value is already normalized to [0, num_partitions_).
+  uint32_t SelectWideColumnPartition(uint32_t column_family_id,
+                                     const Slice& key,
+                                     const WideColumns& columns) const;
 
   // Move the current active partition files into the next immutable
   // memtable-generation batch. Called from SwitchMemtable() while DB mutex is
@@ -117,6 +125,10 @@ class BlobFilePartitionManager {
   // or old SuperVersions and therefore must not be purged yet.
   void GetProtectedBlobFileNumbers(UnorderedSet<uint64_t>* file_numbers) const;
 
+  // Returns true when the blob file is still owned by the write path or
+  // protected by a live memtable / old SuperVersion.
+  bool IsTrackedBlobFileNumber(uint64_t file_number) const;
+
   // Increments / decrements memtable-held protection on sealed blob files.
   void ProtectSealedBlobFileNumbers(const std::vector<uint64_t>& file_numbers);
   void UnprotectSealedBlobFileNumbers(
@@ -136,12 +148,11 @@ class BlobFilePartitionManager {
   // blob file is still write-path-owned and therefore not yet tracked by
   // Version. Existing manifest-visible read results, including I/O failures,
   // are returned directly rather than masked by fallback logic.
-  static Status ResolveBlobDirectWriteIndex(const ReadOptions& read_options,
-                                            const Slice& user_key,
-                                            const BlobIndex& blob_idx,
-                                            const Version* version,
-                                            BlobFileCache* blob_file_cache,
-                                            PinnableSlice* blob_value);
+  static Status ResolveBlobDirectWriteIndex(
+      const ReadOptions& read_options, const Slice& user_key,
+      const BlobIndex& blob_idx, const Version* version,
+      BlobFileCache* blob_file_cache, FilePrefetchBuffer* prefetch_buffer,
+      PinnableSlice* blob_value, uint64_t* bytes_read);
 
  private:
   struct Partition {