Update HISTORY and version to 9.8.4

Summary:
In PR https://github.com/facebook/rocksdb/issues/13074 , we added a logic to prevent stale OPTIONS file from getting deleted by `PurgeObsoleteFiles()` if the OPTIONS file is being referenced by any of the scheduled the remote compactions.

`PurgeObsoleteFiles()` was not the only place that we were cleaning up the old OPTIONS file. We've been also directly cleaning up the old OPTIONS file as part of `SetOptions()`: `RenameTempFileToOptionsFile()` -> `DeleteObsoleteOptionsFiles()` unless FileDeletion is disabled.

This was not caught by the UnitTest because we always preserve the last two OPTIONS file. A single call of `SetOptions()` was not enough to surface this issue in the previous PR.

To keep things simple, we are just skipping the old OPTIONS file clean up in `RenameTempFileToOptionsFile()` if remote compaction is enabled. We let `PurgeObsoleteFiles()` clean up the old options file later after the compaction is done.

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

Test Plan:
Updated UnitTest to reproduce the scenario. It's now passing with the fix.

```
./compaction_service_test --gtest_filter="*PreservedOptionsRemoteCompaction*"
```

Reviewed By: cbi42

Differential Revision: D65974726

Pulled By: jaykorean

fbshipit-source-id: 1907e8450d2ccbb42a93084f275e666648ef5b8c

HISTORY.md

@@ -1,6 +1,51 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 9.8.4 (11/18/2024)
+### Behavior Changes
+* When Remote Compaction is enabled, do not purge OPTIONS file immediately by DeleteObsoleteOptionsFiles() after SetOptions(). Rely on PurgeObsoleteFiles() to clean up obsolete OPTIONS file after each compaction.
+
+## 9.8.3 (11/12/2024)
+### Bug Fixes
+* Fix missing cases of corruption retry during DB open and read API processing.
+
+## 9.8.2 (11/06/2024)
+### Public API Changes
+* Added a new API `Transaction::GetAttributeGroupIterator` that can be used to create a multi-column-family attribute group iterator over the specified column families, including the data from both the transaction and the underlying database. This API is currently supported for optimistic and write-committed pessimistic transactions.
+
+### Behavior Changes
+* `BaseDeltaIterator` now honors the read option `allow_unprepared_value`.
+
+### Bug Fixes
+* `BaseDeltaIterator` now calls `PrepareValue` on the base iterator in case it has been created with the `allow_unprepared_value` read option set. Earlier, such base iterators could lead to incorrect values being exposed from `BaseDeltaIterator`.
+* Fix a bug for replaying WALs for WriteCommitted transaction DB when its user-defined timestamps setting is toggled on/off between DB sessions.
+
+## 9.8.1 (10/31/2024)
+### Bug Fixes
+* Fix a leak of obsolete blob files left open until DB::Close(). This bug was introduced in version 9.4.0.
+
+## 9.8.0 (10/25/2024)
+### New Features
+* All non-`block_cache` options in `BlockBasedTableOptions` are now mutable with `DB::SetOptions()`. See also Bug Fixes below.
+* When using iterators with BlobDB, it is now possible to load large values on an on-demand basis, i.e. only if they are actually needed by the application. This can save I/O in use cases where the values associated with certain keys are not needed. For more details, see the new read option `allow_unprepared_value` and the iterator API `PrepareValue`.
+* Add a new file ingestion option `IngestExternalFileOptions::fill_cache` to support not adding blocks from ingested files into block cache during file ingestion.
+* The option `allow_unprepared_value` is now also supported for multi-column-family iterators (i.e. `CoalescingIterator` and `AttributeGroupIterator`).
+* When a file with just one range deletion (standalone range deletion file) is ingested via bulk loading, it will be marked for compaction. During compaction, this type of files can be used to directly filter out some input files that are not protected by any snapshots and completely deleted by the standalone range deletion file.
+
+### Behavior Changes
+* During file ingestion, overlapping files level assignment are done in multiple batches, so that they can potentially be assigned to lower levels other than always land on L0.
+* OPTIONS file to be loaded by remote worker is now preserved so that it does not get purged by the primary host. A similar technique as how we are preserving new SST files from getting purged is used for this. min_options_file_numbers_ is tracked like pending_outputs_ is tracked.
+* Trim readahead_size during scans so data blocks containing keys that are not in the same prefix as the seek key in `Seek()` are not prefetched when `ReadOptions::auto_readahead_size=true` (default value) and `ReadOptions::prefix_same_as_start = true`
+* Assigning levels for external files are done in the same way for universal compaction and leveled compaction. The old behavior tends to assign files to L0 while the new behavior will assign the files to the lowest level possible.
+
+### Bug Fixes
+* Fix a longstanding race condition in SetOptions for `block_based_table_factory` options. The fix has some subtle behavior changes because of copying and replacing the TableFactory on a change with SetOptions, including requiring an Iterator::Refresh() for an existing Iterator to use the latest options.
+* Fix under counting of allocated memory in the compressed secondary cache due to looking at the compressed block size rather than the actual memory allocated, which could be larger due to internal fragmentation.
+* `GetApproximateMemTableStats()` could return disastrously bad estimates 5-25% of the time. The function has been re-engineered to return much better estimates with similar CPU cost.
+* Skip insertion of compressed blocks in the secondary cache if the lowest_used_cache_tier DB option is kVolatileTier.
+* Fix an issue in level compaction where a small CF with small compaction debt can cause the DB to allow parallel compactions. (#13054)
+* Several DB option settings could be lost through `GetOptionsFromString()`, possibly elsewhere as well. Affected options, now fixed:`background_close_inactive_wals`, `write_dbid_to_manifest`, `write_identity_file`, `prefix_seek_opt_in_only`
+
 ## 9.7.0 (09/20/2024)
 ### New Features
 * Make Cache a customizable class that can be instantiated by the object registry.

db/attribute_group_iterator_impl.h

@@ -13,11 +13,11 @@ namespace ROCKSDB_NAMESPACE {
 class AttributeGroupIteratorImpl : public AttributeGroupIterator {
  public:
   AttributeGroupIteratorImpl(
-      const Comparator* comparator, bool allow_unprepared_value,
-      const std::vector<ColumnFamilyHandle*>& column_families,
-      const std::vector<Iterator*>& child_iterators)
-      : impl_(comparator, allow_unprepared_value, column_families,
-              child_iterators, ResetFunc(this), PopulateFunc(this)) {}
+      const ReadOptions& read_options, const Comparator* comparator,
+      std::vector<std::pair<ColumnFamilyHandle*, std::unique_ptr<Iterator>>>&&
+          cfh_iter_pairs)
+      : impl_(read_options, comparator, std::move(cfh_iter_pairs),
+              ResetFunc(this), PopulateFunc(this)) {}
   ~AttributeGroupIteratorImpl() override {}
 
   // No copy allowed

db/blob/blob_file_cache.cc

@@ -42,6 +42,7 @@ Status BlobFileCache::GetBlobFileReader(
   assert(blob_file_reader);
   assert(blob_file_reader->IsEmpty());
 
+  // NOTE: sharing same Cache with table_cache
   const Slice key = GetSliceForKey(&blob_file_number);
 
   assert(cache_);
@@ -98,4 +99,13 @@ Status BlobFileCache::GetBlobFileReader(
   return Status::OK();
 }
 
+void BlobFileCache::Evict(uint64_t blob_file_number) {
+  // NOTE: sharing same Cache with table_cache
+  const Slice key = GetSliceForKey(&blob_file_number);
+
+  assert(cache_);
+
+  cache_.get()->Erase(key);
+}
+
 }  // namespace ROCKSDB_NAMESPACE

db/blob/blob_file_cache.h

@@ -36,6 +36,15 @@ class BlobFileCache {
                            uint64_t blob_file_number,
                            CacheHandleGuard<BlobFileReader>* blob_file_reader);
 
+  // Called when a blob file is obsolete to ensure it is removed from the cache
+  // to avoid effectively leaking the open file and assicated memory
+  void Evict(uint64_t blob_file_number);
+
+  // Used to identify cache entries for blob files (not normally useful)
+  static const Cache::CacheItemHelper* GetHelper() {
+    return CacheInterface::GetBasicHelper();
+  }
+
  private:
   using CacheInterface =
       BasicTypedCacheInterface<BlobFileReader, CacheEntryRole::kMisc>;

db/coalescing_iterator.h

@@ -12,11 +12,12 @@ namespace ROCKSDB_NAMESPACE {
 // EXPERIMENTAL
 class CoalescingIterator : public Iterator {
  public:
-  CoalescingIterator(const Comparator* comparator, bool allow_unprepared_value,
-                     const std::vector<ColumnFamilyHandle*>& column_families,
-                     const std::vector<Iterator*>& child_iterators)
-      : impl_(comparator, allow_unprepared_value, column_families,
-              child_iterators, ResetFunc(this), PopulateFunc(this)) {}
+  CoalescingIterator(
+      const ReadOptions& read_options, const Comparator* comparator,
+      std::vector<std::pair<ColumnFamilyHandle*, std::unique_ptr<Iterator>>>&&
+          cfh_iter_pairs)
+      : impl_(read_options, comparator, std::move(cfh_iter_pairs),
+              ResetFunc(this), PopulateFunc(this)) {}
   ~CoalescingIterator() override {}
 
   // No copy allowed

db/column_family.h

@@ -402,6 +402,7 @@ class ColumnFamilyData {
                          SequenceNumber earliest_seq);
 
   TableCache* table_cache() const { return table_cache_.get(); }
+  BlobFileCache* blob_file_cache() const { return blob_file_cache_.get(); }
   BlobSource* blob_source() const { return blob_source_.get(); }
 
   // See documentation in compaction_picker.h

db/compaction/compaction_service_test.cc

@@ -483,21 +483,29 @@ TEST_F(CompactionServiceTest, PreservedOptionsRemoteCompaction) {
     ASSERT_OK(Flush());
   }
 
-  bool is_primary_called = false;
-  // This will be called twice. One from primary and one from remote.
-  // Try changing the option when called from remote. Otherwise, the new option
-  // will be used
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency(
+      {{"CompactionServiceTest::OptionsFileChanged",
+        "DBImplSecondary::OpenAndCompact::BeforeLoadingOptions:1"}});
+
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-      "DBImpl::BackgroundCompaction:NonTrivial:BeforeRun", [&](void* /*arg*/) {
-        if (!is_primary_called) {
-          is_primary_called = true;
-          return;
-        }
-        // Change the option right before the compaction run
+      "DBImplSecondary::OpenAndCompact::BeforeLoadingOptions:0",
+      [&](void* arg) {
+        auto options_file_number = static_cast<uint64_t*>(arg);
+        // Change the option twice before the compaction run
         ASSERT_OK(dbfull()->SetOptions(
             {{"level0_file_num_compaction_trigger", "4"}}));
         ASSERT_EQ(4, dbfull()->GetOptions().level0_file_num_compaction_trigger);
-        dbfull()->TEST_DeleteObsoleteFiles();
+        ASSERT_TRUE(dbfull()->versions_->options_file_number() >
+                    *options_file_number);
+
+        // Change the option twice before the compaction run
+        ASSERT_OK(dbfull()->SetOptions(
+            {{"level0_file_num_compaction_trigger", "5"}}));
+        ASSERT_EQ(5, dbfull()->GetOptions().level0_file_num_compaction_trigger);
+        ASSERT_TRUE(dbfull()->versions_->options_file_number() >
+                    *options_file_number);
+
+        TEST_SYNC_POINT("CompactionServiceTest::OptionsFileChanged");
       });
 
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(

db/db_impl/db_impl.cc

@@ -659,8 +659,9 @@ Status DBImpl::CloseHelper() {
   // We need to release them before the block cache is destroyed. The block
   // cache may be destroyed inside versions_.reset(), when column family data
   // list is destroyed, so leaving handles in table cache after
-  // versions_.reset() may cause issues.
-  // Here we clean all unreferenced handles in table cache.
+  // versions_.reset() may cause issues. Here we clean all unreferenced handles
+  // in table cache, and (for certain builds/conditions) assert that no obsolete
+  // files are hanging around unreferenced (leak) in the table/blob file cache.
   // Now we assume all user queries have finished, so only version set itself
   // can possibly hold the blocks from block cache. After releasing unreferenced
   // handles here, only handles held by version set left and inside
@@ -668,6 +669,9 @@ Status DBImpl::CloseHelper() {
   // time a handle is released, we erase it from the cache too. By doing that,
   // we can guarantee that after versions_.reset(), table cache is empty
   // so the cache can be safely destroyed.
+#ifndef NDEBUG
+  TEST_VerifyNoObsoleteFilesCached(/*db_mutex_already_held=*/true);
+#endif  // !NDEBUG
   table_cache_->EraseUnRefEntries();
 
   for (auto& txn_entry : recovered_transactions_) {
@@ -3230,6 +3234,8 @@ Status DBImpl::MultiGetImpl(
       s = Status::Aborted();
       break;
     }
+    // This could be a long-running operation
+    ROCKSDB_THREAD_YIELD_HOOK();
   }
 
   // Post processing (decrement reference counts and record statistics)
@@ -3987,14 +3993,25 @@ std::unique_ptr<IterType> DBImpl::NewMultiCfIterator(
           "Different comparators are being used across CFs"));
     }
   }
+
   std::vector<Iterator*> child_iterators;
   Status s = NewIterators(_read_options, column_families, &child_iterators);
   if (!s.ok()) {
     return error_iterator_func(s);
   }
-  return std::make_unique<ImplType>(
-      column_families[0]->GetComparator(), _read_options.allow_unprepared_value,
-      column_families, std::move(child_iterators));
+
+  assert(column_families.size() == child_iterators.size());
+
+  std::vector<std::pair<ColumnFamilyHandle*, std::unique_ptr<Iterator>>>
+      cfh_iter_pairs;
+  cfh_iter_pairs.reserve(column_families.size());
+  for (size_t i = 0; i < column_families.size(); ++i) {
+    cfh_iter_pairs.emplace_back(column_families[i], child_iterators[i]);
+  }
+
+  return std::make_unique<ImplType>(_read_options,
+                                    column_families[0]->GetComparator(),
+                                    std::move(cfh_iter_pairs));
 }
 
 Status DBImpl::NewIterators(
@@ -5181,11 +5198,12 @@ Status DBImpl::GetDbIdentity(std::string& identity) const {
   return Status::OK();
 }
 
-Status DBImpl::GetDbIdentityFromIdentityFile(std::string* identity) const {
+Status DBImpl::GetDbIdentityFromIdentityFile(const IOOptions& opts,
+                                             std::string* identity) const {
   std::string idfilename = IdentityFileName(dbname_);
   const FileOptions soptions;
 
-  Status s = ReadFileToString(fs_.get(), idfilename, identity);
+  Status s = ReadFileToString(fs_.get(), idfilename, opts, identity);
   if (!s.ok()) {
     return s;
   }
@@ -5496,7 +5514,8 @@ Status DBImpl::WriteOptionsFile(const WriteOptions& write_options,
                                    file_name, fs_.get());
 
   if (s.ok()) {
-    s = RenameTempFileToOptionsFile(file_name);
+    s = RenameTempFileToOptionsFile(file_name,
+                                    db_options.compaction_service != nullptr);
   }
 
   if (!s.ok() && GetEnv()->FileExists(file_name).ok()) {
@@ -5573,7 +5592,8 @@ Status DBImpl::DeleteObsoleteOptionsFiles() {
   return Status::OK();
 }
 
-Status DBImpl::RenameTempFileToOptionsFile(const std::string& file_name) {
+Status DBImpl::RenameTempFileToOptionsFile(const std::string& file_name,
+                                           bool is_remote_compaction_enabled) {
   Status s;
 
   uint64_t options_file_number = versions_->NewFileNumber();
@@ -5617,7 +5637,7 @@ Status DBImpl::RenameTempFileToOptionsFile(const std::string& file_name) {
       my_disable_delete_obsolete_files = disable_delete_obsolete_files_;
     }
 
-    if (!my_disable_delete_obsolete_files) {
+    if (!my_disable_delete_obsolete_files && !is_remote_compaction_enabled) {
       // TODO: Should we check for errors here?
       DeleteObsoleteOptionsFiles().PermitUncheckedError();
     }

db/db_impl/db_impl.h

@@ -482,7 +482,8 @@ class DBImpl : public DB {
 
   Status GetDbIdentity(std::string& identity) const override;
 
-  virtual Status GetDbIdentityFromIdentityFile(std::string* identity) const;
+  virtual Status GetDbIdentityFromIdentityFile(const IOOptions& opts,
+                                               std::string* identity) const;
 
   Status GetDbSessionId(std::string& session_id) const override;
 
@@ -1239,9 +1240,14 @@ class DBImpl : public DB {
   static Status TEST_ValidateOptions(const DBOptions& db_options) {
     return ValidateOptions(db_options);
   }
-
 #endif  // NDEBUG
 
+  // In certain configurations, verify that the table/blob file cache only
+  // contains entries for live files, to check for effective leaks of open
+  // files. This can only be called when purging of obsolete files has
+  // "settled," such as during parts of DB Close().
+  void TEST_VerifyNoObsoleteFilesCached(bool db_mutex_already_held) const;
+
   // persist stats to column family "_persistent_stats"
   void PersistStats();
 
@@ -1463,7 +1469,8 @@ class DBImpl : public DB {
   // The following two functions can only be called when:
   // 1. WriteThread::Writer::EnterUnbatched() is used.
   // 2. db_mutex is NOT held
-  Status RenameTempFileToOptionsFile(const std::string& file_name);
+  Status RenameTempFileToOptionsFile(const std::string& file_name,
+                                     bool is_remote_compaction_enabled);
   Status DeleteObsoleteOptionsFiles();
 
   void NotifyOnManualFlushScheduled(autovector<ColumnFamilyData*> cfds,
@@ -1587,7 +1594,7 @@ class DBImpl : public DB {
   // Read/create DB identity file (as appropriate), and write DB ID to
   // version_edit if provided.
   Status SetupDBId(const WriteOptions& write_options, bool read_only,
-                   bool is_new_db, VersionEdit* version_edit);
+                   bool is_new_db, bool is_retry, VersionEdit* version_edit);
   // Assign db_id_ and write DB ID to version_edit if provided.
   void SetDBId(std::string&& id, bool read_only, VersionEdit* version_edit);
 

db/db_impl/db_impl_debug.cc

@@ -9,6 +9,7 @@
 
 #ifndef NDEBUG
 
+#include "db/blob/blob_file_cache.h"
 #include "db/column_family.h"
 #include "db/db_impl/db_impl.h"
 #include "db/error_handler.h"
@@ -321,5 +322,49 @@ size_t DBImpl::TEST_EstimateInMemoryStatsHistorySize() const {
   InstrumentedMutexLock l(&const_cast<DBImpl*>(this)->stats_history_mutex_);
   return EstimateInMemoryStatsHistorySize();
 }
+
+void DBImpl::TEST_VerifyNoObsoleteFilesCached(
+    bool db_mutex_already_held) const {
+  // This check is somewhat expensive and obscure to make a part of every
+  // unit test in every build variety. Thus, we only enable it for ASAN builds.
+  if (!kMustFreeHeapAllocations) {
+    return;
+  }
+
+  std::optional<InstrumentedMutexLock> l;
+  if (db_mutex_already_held) {
+    mutex_.AssertHeld();
+  } else {
+    l.emplace(&mutex_);
+  }
+
+  std::vector<uint64_t> live_files;
+  for (auto cfd : *versions_->GetColumnFamilySet()) {
+    if (cfd->IsDropped()) {
+      continue;
+    }
+    // Sneakily add both SST and blob files to the same list
+    cfd->current()->AddLiveFiles(&live_files, &live_files);
+  }
+  std::sort(live_files.begin(), live_files.end());
+
+  auto fn = [&live_files](const Slice& key, Cache::ObjectPtr, size_t,
+                          const Cache::CacheItemHelper* helper) {
+    if (helper != BlobFileCache::GetHelper()) {
+      // Skip non-blob files for now
+      // FIXME: diagnose and fix the leaks of obsolete SST files revealed in
+      // unit tests.
+      return;
+    }
+    // See TableCache and BlobFileCache
+    assert(key.size() == sizeof(uint64_t));
+    uint64_t file_number;
+    GetUnaligned(reinterpret_cast<const uint64_t*>(key.data()), &file_number);
+    // Assert file is in sorted live_files
+    assert(
+        std::binary_search(live_files.begin(), live_files.end(), file_number));
+  };
+  table_cache_->ApplyToAllEntries(fn, {});
+}
 }  // namespace ROCKSDB_NAMESPACE
 #endif  // NDEBUG

db/db_impl/db_impl_files.cc

@@ -980,7 +980,8 @@ void DBImpl::SetDBId(std::string&& id, bool read_only,
 }
 
 Status DBImpl::SetupDBId(const WriteOptions& write_options, bool read_only,
-                         bool is_new_db, VersionEdit* version_edit) {
+                         bool is_new_db, bool is_retry,
+                         VersionEdit* version_edit) {
   Status s;
   if (!is_new_db) {
     // Check for the IDENTITY file and create it if not there or
@@ -988,7 +989,11 @@ Status DBImpl::SetupDBId(const WriteOptions& write_options, bool read_only,
     std::string db_id_in_file;
     s = fs_->FileExists(IdentityFileName(dbname_), IOOptions(), nullptr);
     if (s.ok()) {
-      s = GetDbIdentityFromIdentityFile(&db_id_in_file);
+      IOOptions opts;
+      if (is_retry) {
+        opts.verify_and_reconstruct_read = true;
+      }
+      s = GetDbIdentityFromIdentityFile(opts, &db_id_in_file);
       if (s.ok() && !db_id_in_file.empty()) {
         if (db_id_.empty()) {
           // Loaded from file and wasn't already known from manifest

db/db_impl/db_impl_open.cc

@@ -301,7 +301,7 @@ Status DBImpl::NewDB(std::vector<std::string>* new_filenames) {
   VersionEdit new_db_edit;
   const WriteOptions write_options(Env::IOActivity::kDBOpen);
   Status s = SetupDBId(write_options, /*read_only=*/false, /*is_new_db=*/true,
-                       &new_db_edit);
+                       /*is_retry=*/false, &new_db_edit);
   if (!s.ok()) {
     return s;
   }
@@ -675,11 +675,11 @@ Status DBImpl::Recover(
     // Already set up DB ID in NewDB
   } else if (immutable_db_options_.write_dbid_to_manifest && recovery_ctx) {
     VersionEdit edit;
-    s = SetupDBId(write_options, read_only, is_new_db, &edit);
+    s = SetupDBId(write_options, read_only, is_new_db, is_retry, &edit);
     recovery_ctx->UpdateVersionEdits(
         versions_->GetColumnFamilySet()->GetDefault(), edit);
   } else {
-    s = SetupDBId(write_options, read_only, is_new_db, nullptr);
+    s = SetupDBId(write_options, read_only, is_new_db, is_retry, nullptr);
   }
   assert(!s.ok() || !db_id_.empty());
   ROCKS_LOG_INFO(immutable_db_options_.info_log, "DB ID: %s\n", db_id_.c_str());
@@ -1274,7 +1274,8 @@ Status DBImpl::RecoverLogFiles(const std::vector<uint64_t>& wal_numbers,
           reader.GetRecordedTimestampSize();
       status = HandleWriteBatchTimestampSizeDifference(
           &batch, running_ts_sz, record_ts_sz,
-          TimestampSizeConsistencyMode::kReconcileInconsistency, &new_batch);
+          TimestampSizeConsistencyMode::kReconcileInconsistency, seq_per_batch_,
+          batch_per_txn_, &new_batch);
       if (!status.ok()) {
         return status;
       }

db/db_impl/db_impl_readonly.cc

@@ -265,7 +265,8 @@ Status OpenForReadOnlyCheckExistence(const DBOptions& db_options,
     const std::shared_ptr<FileSystem>& fs = db_options.env->GetFileSystem();
     std::string manifest_path;
     uint64_t manifest_file_number;
-    s = VersionSet::GetCurrentManifestPath(dbname, fs.get(), &manifest_path,
+    s = VersionSet::GetCurrentManifestPath(dbname, fs.get(), /*is_retry=*/false,
+                                           &manifest_path,
                                            &manifest_file_number);
   } else {
     // Historic behavior that doesn't necessarily make sense

db/db_impl/db_impl_secondary.cc

@@ -233,7 +233,8 @@ Status DBImplSecondary::RecoverLogFiles(
           reader->GetRecordedTimestampSize();
       status = HandleWriteBatchTimestampSizeDifference(
           &batch, running_ts_sz, record_ts_sz,
-          TimestampSizeConsistencyMode::kVerifyConsistency);
+          TimestampSizeConsistencyMode::kVerifyConsistency, seq_per_batch_,
+          batch_per_txn_);
       if (!status.ok()) {
         break;
       }
@@ -957,6 +958,10 @@ Status DB::OpenAndCompact(
   config_options.env = override_options.env;
   std::vector<ColumnFamilyDescriptor> all_column_families;
 
+  TEST_SYNC_POINT_CALLBACK(
+      "DBImplSecondary::OpenAndCompact::BeforeLoadingOptions:0",
+      &compaction_input.options_file_number);
+  TEST_SYNC_POINT("DBImplSecondary::OpenAndCompact::BeforeLoadingOptions:1");
   std::string options_file_name =
       OptionsFileName(name, compaction_input.options_file_number);
 

db/db_io_failure_test.cc

@@ -27,12 +27,14 @@ class CorruptionFS : public FileSystemWrapper {
         num_writable_file_errors_(0),
         corruption_trigger_(INT_MAX),
         read_count_(0),
+        corrupt_offset_(0),
+        corrupt_len_(0),
         rnd_(300),
         fs_buffer_(fs_buffer),
         verify_read_(verify_read) {}
   ~CorruptionFS() override {
     // Assert that the corruption was reset, which means it got triggered
-    assert(corruption_trigger_ == INT_MAX);
+    assert(corruption_trigger_ == INT_MAX || corrupt_len_ > 0);
   }
   const char* Name() const override { return "ErrorEnv"; }
 
@@ -48,8 +50,10 @@ class CorruptionFS : public FileSystemWrapper {
   }
 
   void SetCorruptionTrigger(const int trigger) {
+    MutexLock l(&mutex_);
     corruption_trigger_ = trigger;
     read_count_ = 0;
+    corrupt_fname_.clear();
   }
 
   IOStatus NewRandomAccessFile(const std::string& fname,
@@ -58,25 +62,31 @@ class CorruptionFS : public FileSystemWrapper {
                                IODebugContext* dbg) override {
     class CorruptionRandomAccessFile : public FSRandomAccessFileOwnerWrapper {
      public:
-      CorruptionRandomAccessFile(CorruptionFS& fs,
+      CorruptionRandomAccessFile(CorruptionFS& fs, const std::string& fname,
                                  std::unique_ptr<FSRandomAccessFile>& file)
-          : FSRandomAccessFileOwnerWrapper(std::move(file)), fs_(fs) {}
+          : FSRandomAccessFileOwnerWrapper(std::move(file)),
+            fs_(fs),
+            fname_(fname) {}
 
       IOStatus Read(uint64_t offset, size_t len, const IOOptions& opts,
                     Slice* result, char* scratch,
                     IODebugContext* dbg) const override {
         IOStatus s = target()->Read(offset, len, opts, result, scratch, dbg);
         if (opts.verify_and_reconstruct_read) {
+          fs_.MaybeResetOverlapWithCorruptedChunk(fname_, offset,
+                                                  result->size());
           return s;
         }
+
+        MutexLock l(&fs_.mutex_);
         if (s.ok() && ++fs_.read_count_ >= fs_.corruption_trigger_) {
-          fs_.read_count_ = 0;
           fs_.corruption_trigger_ = INT_MAX;
           char* data = const_cast<char*>(result->data());
           std::memcpy(
               data,
               fs_.rnd_.RandomString(static_cast<int>(result->size())).c_str(),
               result->size());
+          fs_.SetCorruptedChunk(fname_, offset, result->size());
         }
         return s;
       }
@@ -101,14 +111,76 @@ class CorruptionFS : public FileSystemWrapper {
         return IOStatus::OK();
       }
 
+      IOStatus Prefetch(uint64_t /*offset*/, size_t /*n*/,
+                        const IOOptions& /*options*/,
+                        IODebugContext* /*dbg*/) override {
+        return IOStatus::NotSupported("Prefetch");
+      }
+
      private:
       CorruptionFS& fs_;
+      std::string fname_;
     };
 
     std::unique_ptr<FSRandomAccessFile> file;
     IOStatus s = target()->NewRandomAccessFile(fname, opts, &file, dbg);
     EXPECT_OK(s);
-    result->reset(new CorruptionRandomAccessFile(*this, file));
+    result->reset(new CorruptionRandomAccessFile(*this, fname, file));
+
+    return s;
+  }
+
+  IOStatus NewSequentialFile(const std::string& fname,
+                             const FileOptions& file_opts,
+                             std::unique_ptr<FSSequentialFile>* result,
+                             IODebugContext* dbg) override {
+    class CorruptionSequentialFile : public FSSequentialFileOwnerWrapper {
+     public:
+      CorruptionSequentialFile(CorruptionFS& fs, const std::string& fname,
+                               std::unique_ptr<FSSequentialFile>& file)
+          : FSSequentialFileOwnerWrapper(std::move(file)),
+            fs_(fs),
+            fname_(fname),
+            offset_(0) {}
+
+      IOStatus Read(size_t len, const IOOptions& opts, Slice* result,
+                    char* scratch, IODebugContext* dbg) override {
+        IOStatus s = target()->Read(len, opts, result, scratch, dbg);
+        if (result->size() == 0 ||
+            fname_.find("IDENTITY") != std::string::npos) {
+          return s;
+        }
+
+        if (opts.verify_and_reconstruct_read) {
+          fs_.MaybeResetOverlapWithCorruptedChunk(fname_, offset_,
+                                                  result->size());
+          return s;
+        }
+
+        MutexLock l(&fs_.mutex_);
+        if (s.ok() && ++fs_.read_count_ >= fs_.corruption_trigger_) {
+          fs_.corruption_trigger_ = INT_MAX;
+          char* data = const_cast<char*>(result->data());
+          std::memcpy(
+              data,
+              fs_.rnd_.RandomString(static_cast<int>(result->size())).c_str(),
+              result->size());
+          fs_.SetCorruptedChunk(fname_, offset_, result->size());
+        }
+        offset_ += result->size();
+        return s;
+      }
+
+     private:
+      CorruptionFS& fs_;
+      std::string fname_;
+      size_t offset_;
+    };
+
+    std::unique_ptr<FSSequentialFile> file;
+    IOStatus s = target()->NewSequentialFile(fname, file_opts, &file, dbg);
+    EXPECT_OK(s);
+    result->reset(new CorruptionSequentialFile(*this, fname, file));
 
     return s;
   }
@@ -123,12 +195,40 @@ class CorruptionFS : public FileSystemWrapper {
     }
   }
 
+  void SetCorruptedChunk(const std::string& fname, size_t offset, size_t len) {
+    assert(corrupt_fname_.empty());
+
+    corrupt_fname_ = fname;
+    corrupt_offset_ = offset;
+    corrupt_len_ = len;
+  }
+
+  void MaybeResetOverlapWithCorruptedChunk(const std::string& fname,
+                                           size_t offset, size_t len) {
+    if (fname == corrupt_fname_ &&
+        ((offset <= corrupt_offset_ && (offset + len) > corrupt_offset_) ||
+         (offset >= corrupt_offset_ &&
+          offset < (corrupt_offset_ + corrupt_len_)))) {
+      corrupt_fname_.clear();
+    }
+  }
+
+  bool VerifyRetry() { return corrupt_len_ > 0 && corrupt_fname_.empty(); }
+
+  int read_count() { return read_count_; }
+
+  int corruption_trigger() { return corruption_trigger_; }
+
  private:
   int corruption_trigger_;
   int read_count_;
+  std::string corrupt_fname_;
+  size_t corrupt_offset_;
+  size_t corrupt_len_;
   Random rnd_;
   bool fs_buffer_;
   bool verify_read_;
+  port::Mutex mutex_;
 };
 }  // anonymous namespace
 
@@ -717,6 +817,7 @@ class DBIOCorruptionTest
     bbto.num_file_reads_for_auto_readahead = 0;
     options_.table_factory.reset(NewBlockBasedTableFactory(bbto));
     options_.disable_auto_compactions = true;
+    options_.max_file_opening_threads = 0;
 
     Reopen(options_);
   }
@@ -857,8 +958,8 @@ TEST_P(DBIOCorruptionTest, FlushReadCorruptionRetry) {
   Status s = Flush();
   if (std::get<2>(GetParam())) {
     ASSERT_OK(s);
-    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
-    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
+    ASSERT_GT(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
+    ASSERT_GT(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
               1);
 
     std::string val;
@@ -885,8 +986,8 @@ TEST_P(DBIOCorruptionTest, ManifestCorruptionRetry) {
 
   if (std::get<2>(GetParam())) {
     ASSERT_OK(ReopenDB());
-    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
-    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
+    ASSERT_GT(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
+    ASSERT_GT(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
               1);
   } else {
     ASSERT_EQ(ReopenDB(), Status::Corruption());
@@ -970,6 +1071,57 @@ TEST_P(DBIOCorruptionTest, TablePropertiesCorruptionRetry) {
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
 }
 
+TEST_P(DBIOCorruptionTest, DBOpenReadCorruptionRetry) {
+  if (!std::get<2>(GetParam())) {
+    return;
+  }
+  CorruptionFS* fs =
+      static_cast<CorruptionFS*>(env_guard_->GetFileSystem().get());
+
+  for (int sst = 0; sst < 3; ++sst) {
+    for (int key = 0; key < 100; ++key) {
+      std::stringstream ss;
+      ss << std::setw(3) << 100 * sst + key;
+      ASSERT_OK(Put("key" + ss.str(), "val" + ss.str()));
+    }
+    ASSERT_OK(Flush());
+  }
+  Close();
+
+  // DB open will create table readers unless we reduce the table cache
+  // capacity.
+  // SanitizeOptions will set max_open_files to minimum of 20. Table cache
+  // is allocated with max_open_files - 10 as capacity. So override
+  // max_open_files to 11 so table cache capacity will become 1. This will
+  // prevent file open during DB open and force the file to be opened
+  // during MultiGet
+  SyncPoint::GetInstance()->SetCallBack(
+      "SanitizeOptions::AfterChangeMaxOpenFiles", [&](void* arg) {
+        int* max_open_files = (int*)arg;
+        *max_open_files = 11;
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // Progressively increase the IO count trigger for corruption, and verify
+  // that it was retried
+  int corruption_trigger = 1;
+  fs->SetCorruptionTrigger(corruption_trigger);
+  do {
+    fs->SetCorruptionTrigger(corruption_trigger);
+    ASSERT_OK(ReopenDB());
+    for (int sst = 0; sst < 3; ++sst) {
+      for (int key = 0; key < 100; ++key) {
+        std::stringstream ss;
+        ss << std::setw(3) << 100 * sst + key;
+        ASSERT_EQ(Get("key" + ss.str()), "val" + ss.str());
+      }
+    }
+    // Verify that the injected corruption was repaired
+    ASSERT_TRUE(fs->VerifyRetry());
+    corruption_trigger++;
+  } while (fs->corruption_trigger() == INT_MAX);
+}
+
 // The parameters are - 1. Use FS provided buffer, 2. Use async IO ReadOption,
 // 3. Retry with verify_and_reconstruct_read IOOption
 INSTANTIATE_TEST_CASE_P(DBIOCorruptionTest, DBIOCorruptionTest,

db/db_iter.cc

@@ -581,6 +581,8 @@ bool DBIter::FindNextUserEntryInternal(bool skipping_saved_key,
     } else {
       iter_.Next();
     }
+    // This could be a long-running operation due to tombstones, etc.
+    ROCKSDB_THREAD_YIELD_HOOK();
   } while (iter_.Valid());
 
   valid_ = false;

db/multi_cf_iterator_impl.h

@@ -24,24 +24,18 @@ struct MultiCfIteratorInfo {
 template <typename ResetFunc, typename PopulateFunc>
 class MultiCfIteratorImpl {
  public:
-  MultiCfIteratorImpl(const Comparator* comparator, bool allow_unprepared_value,
-                      const std::vector<ColumnFamilyHandle*>& column_families,
-                      const std::vector<Iterator*>& child_iterators,
-                      ResetFunc reset_func, PopulateFunc populate_func)
-      : comparator_(comparator),
-        allow_unprepared_value_(allow_unprepared_value),
-        heap_(MultiCfMinHeap(
-            MultiCfHeapItemComparator<std::greater<int>>(comparator_))),
+  MultiCfIteratorImpl(
+      const ReadOptions& read_options, const Comparator* comparator,
+      std::vector<std::pair<ColumnFamilyHandle*, std::unique_ptr<Iterator>>>&&
+          cfh_iter_pairs,
+      ResetFunc reset_func, PopulateFunc populate_func)
+      : allow_unprepared_value_(read_options.allow_unprepared_value),
+        comparator_(comparator),
+        cfh_iter_pairs_(std::move(cfh_iter_pairs)),
         reset_func_(std::move(reset_func)),
-        populate_func_(std::move(populate_func)) {
-    assert(column_families.size() > 0 &&
-           column_families.size() == child_iterators.size());
-    cfh_iter_pairs_.reserve(column_families.size());
-    for (size_t i = 0; i < column_families.size(); ++i) {
-      cfh_iter_pairs_.emplace_back(
-          column_families[i], std::unique_ptr<Iterator>(child_iterators[i]));
-    }
-  }
+        populate_func_(std::move(populate_func)),
+        heap_(MultiCfMinHeap(
+            MultiCfHeapItemComparator<std::greater<int>>(comparator_))) {}
   ~MultiCfIteratorImpl() { status_.PermitUncheckedError(); }
 
   // No copy allowed
@@ -108,38 +102,21 @@ class MultiCfIteratorImpl {
       return true;
     }
 
-    auto prepare_value_func = [this](auto& heap, Iterator* iterator) {
-      assert(iterator);
-      assert(iterator->Valid());
-      assert(iterator->status().ok());
-
-      if (!iterator->PrepareValue()) {
-        assert(!iterator->Valid());
-        assert(!iterator->status().ok());
-
-        considerStatus(iterator->status());
-        assert(!status_.ok());
-
-        heap.clear();
-        return false;
-      }
-
-      return true;
-    };
-
     if (std::holds_alternative<MultiCfMaxHeap>(heap_)) {
-      return PopulateIterator(std::get<MultiCfMaxHeap>(heap_),
-                              prepare_value_func);
+      return PopulateIterator(std::get<MultiCfMaxHeap>(heap_));
     }
 
-    return PopulateIterator(std::get<MultiCfMinHeap>(heap_),
-                            prepare_value_func);
+    return PopulateIterator(std::get<MultiCfMinHeap>(heap_));
   }
 
  private:
+  Status status_;
+  bool allow_unprepared_value_;
+  const Comparator* comparator_;
   std::vector<std::pair<ColumnFamilyHandle*, std::unique_ptr<Iterator>>>
       cfh_iter_pairs_;
-  Status status_;
+  ResetFunc reset_func_;
+  PopulateFunc populate_func_;
 
   template <typename CompareOp>
   class MultiCfHeapItemComparator {
@@ -161,9 +138,6 @@ class MultiCfIteratorImpl {
     const Comparator* comparator_;
   };
 
-  const Comparator* comparator_;
-  bool allow_unprepared_value_;
-
   using MultiCfMinHeap =
       BinaryHeap<MultiCfIteratorInfo,
                  MultiCfHeapItemComparator<std::greater<int>>>;
@@ -174,9 +148,6 @@ class MultiCfIteratorImpl {
 
   MultiCfIterHeap heap_;
 
-  ResetFunc reset_func_;
-  PopulateFunc populate_func_;
-
   Iterator* current() const {
     if (std::holds_alternative<MultiCfMaxHeap>(heap_)) {
       auto& max_heap = std::get<MultiCfMaxHeap>(heap_);
@@ -230,10 +201,8 @@ class MultiCfIteratorImpl {
       ++i;
     }
     if (!allow_unprepared_value_ && !heap.empty()) {
-      [[maybe_unused]] const bool result = PopulateIterator(
-          heap,
-          [](auto& /* heap */, Iterator* /* iterator */) { return true; });
-      assert(result);
+      [[maybe_unused]] const bool result = PopulateIterator(heap);
+      assert(result || (!Valid() && !status_.ok()));
     }
   }
 
@@ -300,15 +269,13 @@ class MultiCfIteratorImpl {
     }
 
     if (!allow_unprepared_value_ && !heap.empty()) {
-      [[maybe_unused]] const bool result = PopulateIterator(
-          heap,
-          [](auto& /* heap */, Iterator* /* iterator */) { return true; });
-      assert(result);
+      [[maybe_unused]] const bool result = PopulateIterator(heap);
+      assert(result || (!Valid() && !status_.ok()));
     }
   }
 
-  template <typename BinaryHeap, typename PrepareValueFunc>
-  bool PopulateIterator(BinaryHeap& heap, PrepareValueFunc prepare_value_func) {
+  template <typename BinaryHeap>
+  bool PopulateIterator(BinaryHeap& heap) {
     // 1. Keep the top iterator (by popping it from the heap) and add it to list
     //    to populate
     // 2. For all non-top iterators having the same key as top iter popped
@@ -319,12 +286,33 @@ class MultiCfIteratorImpl {
     //    collected in step 1 and 2 and add all the iters back to the heap
     assert(!heap.empty());
 
+    auto prepare_value = [this, &heap](Iterator* iterator) {
+      assert(iterator);
+      assert(iterator->Valid());
+      assert(iterator->status().ok());
+
+      if (!iterator->PrepareValue()) {
+        assert(!iterator->Valid());
+        assert(!iterator->status().ok());
+
+        considerStatus(iterator->status());
+        heap.clear();
+
+        assert(!Valid());
+        assert(!status_.ok());
+
+        return false;
+      }
+
+      return true;
+    };
+
     auto top = heap.top();
     assert(top.iterator);
     assert(top.iterator->Valid());
     assert(top.iterator->status().ok());
 
-    if (!prepare_value_func(heap, top.iterator)) {
+    if (!prepare_value(top.iterator)) {
       return false;
     }
 
@@ -344,7 +332,7 @@ class MultiCfIteratorImpl {
         break;
       }
 
-      if (!prepare_value_func(heap, current.iterator)) {
+      if (!prepare_value(current.iterator)) {
         return false;
       }
 

db/repair.cc

@@ -413,9 +413,12 @@ class Repairer {
       if (record_status.ok()) {
         const UnorderedMap<uint32_t, size_t>& record_ts_sz =
             reader.GetRecordedTimestampSize();
+        // Use same value for `seq_per_batch` and `batch_per_txn` as
+        // WriteBatchInternal::InsertInto does below.
         record_status = HandleWriteBatchTimestampSizeDifference(
             &batch, running_ts_sz, record_ts_sz,
-            TimestampSizeConsistencyMode::kVerifyConsistency);
+            TimestampSizeConsistencyMode::kVerifyConsistency,
+            /* seq_per_batch */ false, /* batch_per_txn */ true);
         if (record_status.ok()) {
           record_status =
               WriteBatchInternal::InsertInto(&batch, cf_mems, nullptr, nullptr);

db/table_cache.cc

@@ -164,6 +164,7 @@ Status TableCache::GetTableReader(
 }
 
 Cache::Handle* TableCache::Lookup(Cache* cache, uint64_t file_number) {
+  // NOTE: sharing same Cache with BlobFileCache
   Slice key = GetSliceForFileNumber(&file_number);
   return cache->Lookup(key);
 }
@@ -178,6 +179,7 @@ Status TableCache::FindTable(
     size_t max_file_size_for_l0_meta_pin, Temperature file_temperature) {
   PERF_TIMER_GUARD_WITH_CLOCK(find_table_nanos, ioptions_.clock);
   uint64_t number = file_meta.fd.GetNumber();
+  // NOTE: sharing same Cache with BlobFileCache
   Slice key = GetSliceForFileNumber(&number);
   *handle = cache_.Lookup(key);
   TEST_SYNC_POINT_CALLBACK("TableCache::FindTable:0",

db/version_builder.cc

@@ -24,6 +24,7 @@
 #include <vector>
 
 #include "cache/cache_reservation_manager.h"
+#include "db/blob/blob_file_cache.h"
 #include "db/blob/blob_file_meta.h"
 #include "db/dbformat.h"
 #include "db/internal_stats.h"
@@ -744,12 +745,9 @@ class VersionBuilder::Rep {
       return Status::Corruption("VersionBuilder", oss.str());
     }
 
-    // Note: we use C++11 for now but in C++14, this could be done in a more
-    // elegant way using generalized lambda capture.
-    VersionSet* const vs = version_set_;
-    const ImmutableCFOptions* const ioptions = ioptions_;
-
-    auto deleter = [vs, ioptions](SharedBlobFileMetaData* shared_meta) {
+    auto deleter = [vs = version_set_, ioptions = ioptions_,
+                    bc = cfd_ ? cfd_->blob_file_cache()
+                              : nullptr](SharedBlobFileMetaData* shared_meta) {
       if (vs) {
         assert(ioptions);
         assert(!ioptions->cf_paths.empty());
@@ -758,6 +756,9 @@ class VersionBuilder::Rep {
         vs->AddObsoleteBlobFile(shared_meta->GetBlobFileNumber(),
                                 ioptions->cf_paths.front().path);
       }
+      if (bc) {
+        bc->Evict(shared_meta->GetBlobFileNumber());
+      }
 
       delete shared_meta;
     };
@@ -766,7 +767,7 @@ class VersionBuilder::Rep {
         blob_file_number, blob_file_addition.GetTotalBlobCount(),
         blob_file_addition.GetTotalBlobBytes(),
         blob_file_addition.GetChecksumMethod(),
-        blob_file_addition.GetChecksumValue(), deleter);
+        blob_file_addition.GetChecksumValue(), std::move(deleter));
 
     mutable_blob_file_metas_.emplace(
         blob_file_number, MutableBlobFileMetaData(std::move(shared_meta)));

db/version_set.cc

@@ -6014,15 +6014,19 @@ Status VersionSet::LogAndApplyHelper(ColumnFamilyData* cfd,
 }
 
 Status VersionSet::GetCurrentManifestPath(const std::string& dbname,
-                                          FileSystem* fs,
+                                          FileSystem* fs, bool is_retry,
                                           std::string* manifest_path,
                                           uint64_t* manifest_file_number) {
   assert(fs != nullptr);
   assert(manifest_path != nullptr);
   assert(manifest_file_number != nullptr);
 
+  IOOptions opts;
   std::string fname;
-  Status s = ReadFileToString(fs, CurrentFileName(dbname), &fname);
+  if (is_retry) {
+    opts.verify_and_reconstruct_read = true;
+  }
+  Status s = ReadFileToString(fs, CurrentFileName(dbname), opts, &fname);
   if (!s.ok()) {
     return s;
   }
@@ -6052,8 +6056,8 @@ Status VersionSet::Recover(
   // Read "CURRENT" file, which contains a pointer to the current manifest
   // file
   std::string manifest_path;
-  Status s = GetCurrentManifestPath(dbname_, fs_.get(), &manifest_path,
-                                    &manifest_file_number_);
+  Status s = GetCurrentManifestPath(dbname_, fs_.get(), is_retry,
+                                    &manifest_path, &manifest_file_number_);
   if (!s.ok()) {
     return s;
   }
@@ -6298,8 +6302,8 @@ Status VersionSet::ListColumnFamilies(std::vector<std::string>* column_families,
   // Read "CURRENT" file, which contains a pointer to the current manifest file
   std::string manifest_path;
   uint64_t manifest_file_number;
-  Status s =
-      GetCurrentManifestPath(dbname, fs, &manifest_path, &manifest_file_number);
+  Status s = GetCurrentManifestPath(dbname, fs, /*is_retry=*/false,
+                                    &manifest_path, &manifest_file_number);
   if (!s.ok()) {
     return s;
   }
@@ -7497,8 +7501,8 @@ Status ReactiveVersionSet::MaybeSwitchManifest(
   assert(manifest_reader != nullptr);
   Status s;
   std::string manifest_path;
-  s = GetCurrentManifestPath(dbname_, fs_.get(), &manifest_path,
-                             &manifest_file_number_);
+  s = GetCurrentManifestPath(dbname_, fs_.get(), /*is_retry=*/false,
+                             &manifest_path, &manifest_file_number_);
   if (!s.ok()) {
     return s;
   }

db/version_set.h

@@ -1278,7 +1278,7 @@ class VersionSet {
           {});
 
   static Status GetCurrentManifestPath(const std::string& dbname,
-                                       FileSystem* fs,
+                                       FileSystem* fs, bool is_retry,
                                        std::string* manifest_filename,
                                        uint64_t* manifest_file_number);
   void WakeUpWaitingManifestWriters();
@@ -1526,7 +1526,6 @@ class VersionSet {
   void GetLiveFilesMetaData(std::vector<LiveFileMetaData>* metadata);
 
   void AddObsoleteBlobFile(uint64_t blob_file_number, std::string path) {
-    // TODO: Erase file from BlobFileCache?
     obsolete_blob_files_.emplace_back(blob_file_number, std::move(path));
   }
 

db/version_set_test.cc

@@ -1223,7 +1223,7 @@ class VersionSetTestBase {
       tmp_db_options.env = env_;
       std::unique_ptr<DBImpl> impl(new DBImpl(tmp_db_options, dbname_));
       std::string db_id;
-      ASSERT_OK(impl->GetDbIdentityFromIdentityFile(&db_id));
+      ASSERT_OK(impl->GetDbIdentityFromIdentityFile(IOOptions(), &db_id));
       new_db.SetDBId(db_id);
     }
     new_db.SetLogNumber(0);
@@ -1391,7 +1391,8 @@ class VersionSetTestBase {
     assert(manifest_path != nullptr);
     uint64_t manifest_file_number = 0;
     Status s = versions_->GetCurrentManifestPath(
-        dbname_, fs_.get(), manifest_path, &manifest_file_number);
+        dbname_, fs_.get(), /*is_retry=*/false, manifest_path,
+        &manifest_file_number);
     ASSERT_OK(s);
   }
 
@@ -1399,7 +1400,8 @@ class VersionSetTestBase {
     assert(manifest_path != nullptr);
     uint64_t manifest_file_number = 0;
     Status s = versions_->GetCurrentManifestPath(
-        dbname_, fs_.get(), manifest_path, &manifest_file_number);
+        dbname_, fs_.get(), /*is_retry=*/false, manifest_path,
+        &manifest_file_number);
     ASSERT_OK(s);
     ASSERT_EQ(1, manifest_file_number);
   }
@@ -3515,7 +3517,7 @@ class VersionSetTestEmptyDb
       tmp_db_options.env = env_;
       std::unique_ptr<DBImpl> impl(new DBImpl(tmp_db_options, dbname_));
       std::string db_id;
-      ASSERT_OK(impl->GetDbIdentityFromIdentityFile(&db_id));
+      ASSERT_OK(impl->GetDbIdentityFromIdentityFile(IOOptions(), &db_id));
       new_db.SetDBId(db_id);
     }
     const std::string manifest_path = DescriptorFileName(dbname_, 1);
@@ -3839,7 +3841,7 @@ class VersionSetTestMissingFiles : public VersionSetTestBase,
       tmp_db_options.env = env_;
       std::unique_ptr<DBImpl> impl(new DBImpl(tmp_db_options, dbname_));
       std::string db_id;
-      ASSERT_OK(impl->GetDbIdentityFromIdentityFile(&db_id));
+      ASSERT_OK(impl->GetDbIdentityFromIdentityFile(IOOptions(), &db_id));
       new_db.SetDBId(db_id);
     }
     {

db/write_batch.cc

@@ -1160,6 +1160,34 @@ Status WriteBatchInternal::InsertNoop(WriteBatch* b) {
   return Status::OK();
 }
 
+ValueType WriteBatchInternal::GetBeginPrepareType(bool write_after_commit,
+                                                  bool unprepared_batch) {
+  return write_after_commit
+             ? kTypeBeginPrepareXID
+             : (unprepared_batch ? kTypeBeginUnprepareXID
+                                 : kTypeBeginPersistedPrepareXID);
+}
+
+Status WriteBatchInternal::InsertBeginPrepare(WriteBatch* b,
+                                              bool write_after_commit,
+                                              bool unprepared_batch) {
+  b->rep_.push_back(static_cast<char>(
+      GetBeginPrepareType(write_after_commit, unprepared_batch)));
+  b->content_flags_.store(b->content_flags_.load(std::memory_order_relaxed) |
+                              ContentFlags::HAS_BEGIN_PREPARE,
+                          std::memory_order_relaxed);
+  return Status::OK();
+}
+
+Status WriteBatchInternal::InsertEndPrepare(WriteBatch* b, const Slice& xid) {
+  b->rep_.push_back(static_cast<char>(kTypeEndPrepareXID));
+  PutLengthPrefixedSlice(&b->rep_, xid);
+  b->content_flags_.store(b->content_flags_.load(std::memory_order_relaxed) |
+                              ContentFlags::HAS_END_PREPARE,
+                          std::memory_order_relaxed);
+  return Status::OK();
+}
+
 Status WriteBatchInternal::MarkEndPrepare(WriteBatch* b, const Slice& xid,
                                           bool write_after_commit,
                                           bool unprepared_batch) {
@@ -1175,13 +1203,8 @@ Status WriteBatchInternal::MarkEndPrepare(WriteBatch* b, const Slice& xid,
 
   // rewrite noop as begin marker
   b->rep_[12] = static_cast<char>(
-      write_after_commit ? kTypeBeginPrepareXID
-                         : (unprepared_batch ? kTypeBeginUnprepareXID
-                                             : kTypeBeginPersistedPrepareXID));
-  b->rep_.push_back(static_cast<char>(kTypeEndPrepareXID));
-  PutLengthPrefixedSlice(&b->rep_, xid);
+      GetBeginPrepareType(write_after_commit, unprepared_batch));
   b->content_flags_.store(b->content_flags_.load(std::memory_order_relaxed) |
-                              ContentFlags::HAS_END_PREPARE |
                               ContentFlags::HAS_BEGIN_PREPARE,
                           std::memory_order_relaxed);
   if (unprepared_batch) {
@@ -1189,7 +1212,7 @@ Status WriteBatchInternal::MarkEndPrepare(WriteBatch* b, const Slice& xid,
                                 ContentFlags::HAS_BEGIN_UNPREPARE,
                             std::memory_order_relaxed);
   }
-  return Status::OK();
+  return WriteBatchInternal::InsertEndPrepare(b, xid);
 }
 
 Status WriteBatchInternal::MarkCommit(WriteBatch* b, const Slice& xid) {

db/write_batch_internal.h

@@ -122,6 +122,15 @@ class WriteBatchInternal {
   static Status PutBlobIndex(WriteBatch* batch, uint32_t column_family_id,
                              const Slice& key, const Slice& value);
 
+  static ValueType GetBeginPrepareType(bool write_after_commit,
+                                       bool unprepared_batch);
+
+  static Status InsertBeginPrepare(WriteBatch* batch,
+                                   const bool write_after_commit = true,
+                                   bool unprepared_batch = false);
+
+  static Status InsertEndPrepare(WriteBatch* batch, const Slice& xid);
+
   static Status MarkEndPrepare(WriteBatch* batch, const Slice& xid,
                                const bool write_after_commit = true,
                                const bool unprepared_batch = false);

env/file_system.cc

@@ -200,6 +200,11 @@ IOStatus WriteStringToFile(FileSystem* fs, const Slice& data,
 
 IOStatus ReadFileToString(FileSystem* fs, const std::string& fname,
                           std::string* data) {
+  return ReadFileToString(fs, fname, IOOptions(), data);
+}
+
+IOStatus ReadFileToString(FileSystem* fs, const std::string& fname,
+                          const IOOptions& opts, std::string* data) {
   FileOptions soptions;
   data->clear();
   std::unique_ptr<FSSequentialFile> file;
@@ -212,7 +217,7 @@ IOStatus ReadFileToString(FileSystem* fs, const std::string& fname,
   char* space = new char[kBufferSize];
   while (true) {
     Slice fragment;
-    s = file->Read(kBufferSize, IOOptions(), &fragment, space, nullptr);
+    s = file->Read(kBufferSize, opts, &fragment, space, nullptr);
     if (!s.ok()) {
       break;
     }

include/rocksdb/file_system.h

@@ -1961,4 +1961,8 @@ IOStatus WriteStringToFile(FileSystem* fs, const Slice& data,
 IOStatus ReadFileToString(FileSystem* fs, const std::string& fname,
                           std::string* data);
 
+// A utility routine: read contents of named file into *data
+IOStatus ReadFileToString(FileSystem* fs, const std::string& fname,
+                          const IOOptions& opts, std::string* data);
+
 }  // namespace ROCKSDB_NAMESPACE

include/rocksdb/utilities/transaction.h

@@ -477,6 +477,21 @@ class Transaction {
   virtual Iterator* GetIterator(const ReadOptions& read_options,
                                 ColumnFamilyHandle* column_family) = 0;
 
+  // Returns a multi-column-family attribute group iterator for the given column
+  // families that includes both keys in the DB and uncommitted keys in this
+  // transaction.
+  //
+  // Setting read_options.snapshot will affect what is read from the
+  // DB but will NOT change which keys are read from this transaction (the keys
+  // in this transaction do not yet belong to any snapshot and will be fetched
+  // regardless).
+  //
+  // The returned iterator is only valid until Commit(), Rollback(), or
+  // RollbackToSavePoint() is called.
+  virtual std::unique_ptr<AttributeGroupIterator> GetAttributeGroupIterator(
+      const ReadOptions& read_options,
+      const std::vector<ColumnFamilyHandle*>& column_families) = 0;
+
   // Put, PutEntity, Merge, Delete, and SingleDelete behave similarly to the
   // corresponding functions in WriteBatch, but will also do conflict checking
   // on the keys being written.

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 9
 #define ROCKSDB_MINOR 8
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 4
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

options/options_helper.cc

@@ -201,6 +201,7 @@ void BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.metadata_write_temperature =
       immutable_db_options.metadata_write_temperature;
   options.wal_write_temperature = immutable_db_options.wal_write_temperature;
+  options.compaction_service = immutable_db_options.compaction_service;
 }
 
 ColumnFamilyOptions BuildColumnFamilyOptions(

options/options_parser.cc

@@ -12,6 +12,7 @@
 #include <utility>
 #include <vector>
 
+#include "file/file_util.h"
 #include "file/line_file_reader.h"
 #include "file/writable_file_writer.h"
 #include "options/cf_options.h"
@@ -268,70 +269,89 @@ Status RocksDBOptionsParser::Parse(const ConfigOptions& config_options_in,
   Reset();
   ConfigOptions config_options = config_options_in;
 
-  std::unique_ptr<FSSequentialFile> seq_file;
-  Status s = fs->NewSequentialFile(file_name, FileOptions(), &seq_file,
-                                   nullptr);
-  if (!s.ok()) {
-    return s;
-  }
-  LineFileReader lf_reader(std::move(seq_file), file_name,
-                           config_options.file_readahead_size);
-
-  OptionSection section = kOptionSectionUnknown;
-  std::string title;
-  std::string argument;
-  std::unordered_map<std::string, std::string> opt_map;
-  std::string line;
-  // we only support single-lined statement.
-  while (lf_reader.ReadLine(&line, Env::IO_TOTAL /* rate_limiter_priority */)) {
-    int line_num = static_cast<int>(lf_reader.GetLineNumber());
-    line = TrimAndRemoveComment(line);
-    if (line.empty()) {
-      continue;
+  Status s;
+  bool retry = false;
+  do {
+    std::unique_ptr<FSSequentialFile> seq_file;
+    s = fs->NewSequentialFile(file_name, FileOptions(), &seq_file, nullptr);
+    if (!s.ok()) {
+      return s;
     }
-    if (IsSection(line)) {
+
+    LineFileReader lf_reader(
+        std::move(seq_file), file_name, config_options.file_readahead_size,
+        nullptr, std::vector<std::shared_ptr<EventListener>>{}, nullptr, retry);
+
+    OptionSection section = kOptionSectionUnknown;
+    std::string title;
+    std::string argument;
+    std::unordered_map<std::string, std::string> opt_map;
+    std::string line;
+    // we only support single-lined statement.
+    while (
+        lf_reader.ReadLine(&line, Env::IO_TOTAL /* rate_limiter_priority */)) {
+      int line_num = static_cast<int>(lf_reader.GetLineNumber());
+      line = TrimAndRemoveComment(line);
+      if (line.empty()) {
+        continue;
+      }
+      if (IsSection(line)) {
+        s = EndSection(config_options, section, title, argument, opt_map);
+        opt_map.clear();
+        if (!s.ok()) {
+          break;
+        }
+
+        // If the option file is not generated by a higher version, unknown
+        // option should only mean corruption.
+        if (config_options.ignore_unknown_options &&
+            section == kOptionSectionVersion) {
+          using VTuple = std::tuple<int, int, int>;
+          if (VTuple(db_version[0], db_version[1], db_version[2]) <=
+              VTuple(ROCKSDB_MAJOR, ROCKSDB_MINOR, ROCKSDB_PATCH)) {
+            config_options.ignore_unknown_options = false;
+          }
+        }
+
+        s = ParseSection(&section, &title, &argument, line, line_num);
+        if (!s.ok()) {
+          break;
+        }
+      } else {
+        std::string name;
+        std::string value;
+        s = ParseStatement(&name, &value, line, line_num);
+        if (!s.ok()) {
+          break;
+        }
+        opt_map.insert({name, value});
+      }
+    }
+    if (s.ok()) {
+      s = lf_reader.GetStatus();
+    }
+    if (s.ok()) {
       s = EndSection(config_options, section, title, argument, opt_map);
       opt_map.clear();
-      if (!s.ok()) {
-        return s;
-      }
-
-      // If the option file is not generated by a higher version, unknown
-      // option should only mean corruption.
-      if (config_options.ignore_unknown_options &&
-          section == kOptionSectionVersion) {
-        using VTuple = std::tuple<int, int, int>;
-        if (VTuple(db_version[0], db_version[1], db_version[2]) <=
-            VTuple(ROCKSDB_MAJOR, ROCKSDB_MINOR, ROCKSDB_PATCH)) {
-          config_options.ignore_unknown_options = false;
-        }
-      }
-
-      s = ParseSection(&section, &title, &argument, line, line_num);
-      if (!s.ok()) {
+    }
+    if (s.ok()) {
+      s = ValidityCheck();
+    }
+    if (!s.ok()) {
+      if ((s.IsCorruption() || s.IsInvalidArgument()) && !retry &&
+          CheckFSFeatureSupport(fs,
+                                FSSupportedOps::kVerifyAndReconstructRead)) {
+        retry = true;
+        Reset();
+      } else {
         return s;
       }
     } else {
-      std::string name;
-      std::string value;
-      s = ParseStatement(&name, &value, line, line_num);
-      if (!s.ok()) {
-        return s;
-      }
-      opt_map.insert({name, value});
+      return s;
     }
-  }
-  s = lf_reader.GetStatus();
-  if (!s.ok()) {
-    return s;
-  }
+  } while (retry);
 
-  s = EndSection(config_options, section, title, argument, opt_map);
-  opt_map.clear();
-  if (!s.ok()) {
-    return s;
-  }
-  return ValidityCheck();
+  return s;
 }
 
 Status RocksDBOptionsParser::CheckSection(const OptionSection section,

port/port.h

@@ -19,3 +19,19 @@
 #elif defined(OS_WIN)
 #include "port/win/port_win.h"
 #endif
+
+#ifdef OS_LINUX
+// A temporary hook into long-running RocksDB threads to support modifying their
+// priority etc. This should become a public API hook once the requirements
+// are better understood.
+extern "C" void RocksDbThreadYield() __attribute__((__weak__));
+#define ROCKSDB_THREAD_YIELD_HOOK() \
+  {                                 \
+    if (RocksDbThreadYield) {       \
+      RocksDbThreadYield();         \
+    }                               \
+  }
+#else
+#define ROCKSDB_THREAD_YIELD_HOOK() \
+  {}
+#endif

table/format.cc

@@ -560,9 +560,9 @@ Status ReadFooterFromFile(const IOOptions& opts, RandomAccessFileReader* file,
     IOOptions new_opts = opts;
     new_opts.verify_and_reconstruct_read = true;
     footer->Reset();
-    s = ReadFooterFromFileInternal(new_opts, file, fs, prefetch_buffer,
-                                   file_size, footer,
-                                   enforce_table_magic_number);
+    s = ReadFooterFromFileInternal(new_opts, file, fs,
+                                   /*prefetch_buffer=*/nullptr, file_size,
+                                   footer, enforce_table_magic_number);
     RecordTick(stats, FILE_READ_CORRUPTION_RETRY_COUNT);
     if (s.ok()) {
       RecordTick(stats, FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT);

table/meta_blocks.cc

@@ -570,7 +570,7 @@ Status ReadMetaIndexBlockInFile(RandomAccessFileReader* file,
     return s;
   }
   s = ReadFooterFromFile(opts, file, *ioptions.fs, prefetch_buffer, file_size,
-                         &footer, table_magic_number);
+                         &footer, table_magic_number, ioptions.stats);
   if (!s.ok()) {
     return s;
   }

tools/ldb_cmd.cc

@@ -3053,6 +3053,7 @@ void DumpWalFile(Options options, const std::string& wal_file,
           status = HandleWriteBatchTimestampSizeDifference(
               &batch, running_ts_sz, recorded_ts_sz,
               TimestampSizeConsistencyMode::kVerifyConsistency,
+              /* seq_per_batch */ false, /* batch_per_txn */ true,
               /*new_batch=*/nullptr);
           if (!status.ok()) {
             std::stringstream oss;

unreleased_history/behavior_changes/overlapping_files_ingestion.md

@@ -1 +0,0 @@
-*During file ingestion, overlapping files level assignment are done in multiple batches, so that they can potentially be assigned to lower levels other than always land on L0.

unreleased_history/behavior_changes/parallel-compaction.md

@@ -1 +0,0 @@
-* Fix an issue in level compaction where a small CF with small compaction debt can cause the DB to allow parallel compactions. (#13054) 

unreleased_history/behavior_changes/remote_compaction_preserve_options.md

@@ -1 +0,0 @@
-OPTIONS file to be loaded by remote worker is now preserved so that it does not get purged by the primary host. A similar technique as how we are preserving new SST files from getting purged is used for this. min_options_file_numbers_ is tracked like pending_outputs_ is tracked.

unreleased_history/behavior_changes/trim_readahead_by_prefix.md

@@ -1 +0,0 @@
-*  Trim readahead_size during scans so data blocks containing keys that are not in the same prefix as the seek key in `Seek()` are not prefetched when `ReadOptions::auto_readahead_size=true` (default value) and `ReadOptions::prefix_same_as_start = true`

unreleased_history/behavior_changes/universal_level_assgiment.md

@@ -1 +0,0 @@
-*Assigning levels for external files are done in the same way for universal compaction and leveled compaction. The old behavior tends to assign files to L0 while the new behavior will assign the files to the lowest level possible.

unreleased_history/bug_fixes/build_db_options.md

@@ -1 +0,0 @@
-* Several DB option settings could be lost through `GetOptionsFromString()`, possibly elsewhere as well. Affected options, now fixed:`background_close_inactive_wals`, `write_dbid_to_manifest`, `write_identity_file`, `prefix_seek_opt_in_only`

unreleased_history/bug_fixes/compressed_secondary_cache_account.md

@@ -1 +0,0 @@
-Fix under counting of allocated memory in the compressed secondary cache due to looking at the compressed block size rather than the actual memory allocated, which could be larger due to internal fragmentation.

unreleased_history/bug_fixes/memtable_stats.md

@@ -1 +0,0 @@
-* `GetApproximateMemTableStats()` could return disastrously bad estimates 5-25% of the time. The function has been re-engineered to return much better estimates with similar CPU cost.

unreleased_history/bug_fixes/set_options_race.md

@@ -1 +0,0 @@
-Fix a longstanding race condition in SetOptions for `block_based_table_factory` options. The fix has some subtle behavior changes because of copying and replacing the TableFactory on a change with SetOptions, including requiring an Iterator::Refresh() for an existing Iterator to use the latest options.

unreleased_history/bug_fixes/skip_insertion_tiered_sec_cache.md

@@ -1 +0,0 @@
-Skip insertion of compressed blocks in the secondary cache if the lowest_used_cache_tier DB option is kVolatileTier.

unreleased_history/new_features/blob_db_allow_unprepared_value.md

@@ -1 +0,0 @@
-When using iterators with BlobDB, it is now possible to load large values on an on-demand basis, i.e. only if they are actually needed by the application. This can save I/O in use cases where the values associated with certain keys are not needed. For more details, see the new read option `allow_unprepared_value` and the iterator API `PrepareValue`.

unreleased_history/new_features/ingest-fill-cache.md

@@ -1 +0,0 @@
-* Add a new file ingestion option `IngestExternalFileOptions::fill_cache` to support not adding blocks from ingested files into block cache during file ingestion.    

unreleased_history/new_features/multi_cf_allow_unprepared_value.md

@@ -1 +0,0 @@
-The option `allow_unprepared_value` is now also supported for multi-column-family iterators (i.e. `CoalescingIterator` and `AttributeGroupIterator`).

unreleased_history/new_features/mutable_table_options.md

@@ -1 +0,0 @@
-* All "simple" options in `BlockBasedTableOptions` are now mutable with `DB::SetOptions()`. For now, "simple" only includes non-pointer options that are 64 bits or less.

unreleased_history/new_features/standalone_range_deletion_file.md

@@ -1 +0,0 @@
-*When a file with just one range deletion (standalone range deletion file) is ingested via bulk loading, it will be marked for compaction. During compaction, this type of files can be used to directly filter out some input files that are not protected by any snapshots and completely deleted by the standalone range deletion file. 

util/udt_util.cc

@@ -139,9 +139,17 @@ ToggleUDT CompareComparator(const Comparator* new_comparator,
 
 TimestampRecoveryHandler::TimestampRecoveryHandler(
     const UnorderedMap<uint32_t, size_t>& running_ts_sz,
-    const UnorderedMap<uint32_t, size_t>& record_ts_sz)
+    const UnorderedMap<uint32_t, size_t>& record_ts_sz, bool seq_per_batch,
+    bool batch_per_txn)
     : running_ts_sz_(running_ts_sz),
       record_ts_sz_(record_ts_sz),
+      // Write after commit currently uses one seq per key (instead of per
+      // batch). So seq_per_batch being false indicates write_after_commit
+      // approach.
+      write_after_commit_(!seq_per_batch),
+      // WriteUnprepared can write multiple WriteBatches per transaction, so
+      // batch_per_txn being false indicates write_before_prepare.
+      write_before_prepare_(!batch_per_txn),
       new_batch_(new WriteBatch()),
       handler_valid_(true),
       new_batch_diff_from_orig_batch_(false) {}
@@ -258,6 +266,43 @@ Status TimestampRecoveryHandler::PutBlobIndexCF(uint32_t cf, const Slice& key,
   return WriteBatchInternal::PutBlobIndex(new_batch_.get(), cf, new_key, value);
 }
 
+Status TimestampRecoveryHandler::MarkBeginPrepare(bool unprepare) {
+  // Transaction policy change requires empty WAL and User-defined timestamp is
+  // only supported for write committed txns.
+  // WriteBatch::Iterate has will handle this based on
+  // handler->WriteAfterCommit() and handler->WriteBeforePrepare().
+  if (unprepare) {
+    return Status::InvalidArgument(
+        "Handle user defined timestamp setting change is not supported for"
+        "write unprepared policy. The WAL must be emptied.");
+  }
+  return WriteBatchInternal::InsertBeginPrepare(new_batch_.get(),
+                                                write_after_commit_,
+                                                /* unprepared_batch */ false);
+}
+
+Status TimestampRecoveryHandler::MarkEndPrepare(const Slice& name) {
+  return WriteBatchInternal::InsertEndPrepare(new_batch_.get(), name);
+}
+
+Status TimestampRecoveryHandler::MarkCommit(const Slice& name) {
+  return WriteBatchInternal::MarkCommit(new_batch_.get(), name);
+}
+
+Status TimestampRecoveryHandler::MarkCommitWithTimestamp(
+    const Slice& name, const Slice& commit_ts) {
+  return WriteBatchInternal::MarkCommitWithTimestamp(new_batch_.get(), name,
+                                                     commit_ts);
+}
+
+Status TimestampRecoveryHandler::MarkRollback(const Slice& name) {
+  return WriteBatchInternal::MarkRollback(new_batch_.get(), name);
+}
+
+Status TimestampRecoveryHandler::MarkNoop(bool /*empty_batch*/) {
+  return WriteBatchInternal::InsertNoop(new_batch_.get());
+}
+
 Status TimestampRecoveryHandler::ReconcileTimestampDiscrepancy(
     uint32_t cf, const Slice& key, std::string* new_key_buf, Slice* new_key) {
   assert(handler_valid_);
@@ -304,8 +349,8 @@ Status HandleWriteBatchTimestampSizeDifference(
     const WriteBatch* batch,
     const UnorderedMap<uint32_t, size_t>& running_ts_sz,
     const UnorderedMap<uint32_t, size_t>& record_ts_sz,
-    TimestampSizeConsistencyMode check_mode,
-    std::unique_ptr<WriteBatch>* new_batch) {
+    TimestampSizeConsistencyMode check_mode, bool seq_per_batch,
+    bool batch_per_txn, std::unique_ptr<WriteBatch>* new_batch) {
   // Quick path to bypass checking the WriteBatch.
   if (AllRunningColumnFamiliesConsistent(running_ts_sz, record_ts_sz)) {
     return Status::OK();
@@ -318,7 +363,8 @@ Status HandleWriteBatchTimestampSizeDifference(
   } else if (need_recovery) {
     assert(new_batch);
     SequenceNumber sequence = WriteBatchInternal::Sequence(batch);
-    TimestampRecoveryHandler recovery_handler(running_ts_sz, record_ts_sz);
+    TimestampRecoveryHandler recovery_handler(running_ts_sz, record_ts_sz,
+                                              seq_per_batch, batch_per_txn);
     status = batch->Iterate(&recovery_handler);
     if (!status.ok()) {
       return status;

util/udt_util.h

@@ -105,7 +105,8 @@ class UserDefinedTimestampSizeRecord {
 class TimestampRecoveryHandler : public WriteBatch::Handler {
  public:
   TimestampRecoveryHandler(const UnorderedMap<uint32_t, size_t>& running_ts_sz,
-                           const UnorderedMap<uint32_t, size_t>& record_ts_sz);
+                           const UnorderedMap<uint32_t, size_t>& record_ts_sz,
+                           bool seq_per_batch, bool batch_per_txn);
 
   ~TimestampRecoveryHandler() override {}
 
@@ -135,19 +136,18 @@ class TimestampRecoveryHandler : public WriteBatch::Handler {
   Status PutBlobIndexCF(uint32_t cf, const Slice& key,
                         const Slice& value) override;
 
-  Status MarkBeginPrepare(bool) override { return Status::OK(); }
+  Status MarkBeginPrepare(bool unprepare) override;
 
-  Status MarkEndPrepare(const Slice&) override { return Status::OK(); }
+  Status MarkEndPrepare(const Slice& name) override;
 
-  Status MarkCommit(const Slice&) override { return Status::OK(); }
+  Status MarkCommit(const Slice& name) override;
 
-  Status MarkCommitWithTimestamp(const Slice&, const Slice&) override {
-    return Status::OK();
-  }
+  Status MarkCommitWithTimestamp(const Slice& name,
+                                 const Slice& commit_ts) override;
 
-  Status MarkRollback(const Slice&) override { return Status::OK(); }
+  Status MarkRollback(const Slice& name) override;
 
-  Status MarkNoop(bool /*empty_batch*/) override { return Status::OK(); }
+  Status MarkNoop(bool empty_batch) override;
 
   std::unique_ptr<WriteBatch>&& TransferNewBatch() {
     assert(new_batch_diff_from_orig_batch_);
@@ -155,6 +155,16 @@ class TimestampRecoveryHandler : public WriteBatch::Handler {
     return std::move(new_batch_);
   }
 
+ protected:
+  Handler::OptionState WriteBeforePrepare() const override {
+    return write_before_prepare_ ? Handler::OptionState::kEnabled
+                                 : Handler::OptionState::kDisabled;
+  }
+  Handler::OptionState WriteAfterCommit() const override {
+    return write_after_commit_ ? Handler::OptionState::kEnabled
+                               : Handler::OptionState::kDisabled;
+  }
+
  private:
   Status ReconcileTimestampDiscrepancy(uint32_t cf, const Slice& key,
                                        std::string* new_key_buf,
@@ -168,6 +178,9 @@ class TimestampRecoveryHandler : public WriteBatch::Handler {
   // in the WAL. This only contains non-zero user-defined timestamp size.
   const UnorderedMap<uint32_t, size_t>& record_ts_sz_;
 
+  bool write_after_commit_;
+  bool write_before_prepare_;
+
   std::unique_ptr<WriteBatch> new_batch_;
   // Handler is valid upon creation and becomes invalid after its `new_batch_`
   // is transferred.
@@ -220,8 +233,8 @@ Status HandleWriteBatchTimestampSizeDifference(
     const WriteBatch* batch,
     const UnorderedMap<uint32_t, size_t>& running_ts_sz,
     const UnorderedMap<uint32_t, size_t>& record_ts_sz,
-    TimestampSizeConsistencyMode check_mode,
-    std::unique_ptr<WriteBatch>* new_batch = nullptr);
+    TimestampSizeConsistencyMode check_mode, bool seq_per_batch,
+    bool batch_per_txn, std::unique_ptr<WriteBatch>* new_batch = nullptr);
 
 // This util function is used when opening an existing column family and
 // processing its VersionEdit. It does a sanity check for the column family's

util/udt_util_test.cc

@@ -226,11 +226,13 @@ TEST_F(HandleTimestampSizeDifferenceTest, AllColumnFamiliesConsistent) {
   // All `check_mode` pass with OK status and `batch` not checked or updated.
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kVerifyConsistency));
+      TimestampSizeConsistencyMode::kVerifyConsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true));
   std::unique_ptr<WriteBatch> new_batch(nullptr);
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kReconcileInconsistency, &new_batch));
+      TimestampSizeConsistencyMode::kReconcileInconsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true, &new_batch));
   ASSERT_TRUE(new_batch.get() == nullptr);
 }
 
@@ -245,11 +247,13 @@ TEST_F(HandleTimestampSizeDifferenceTest,
   // All `check_mode` pass with OK status and `batch` not checked or updated.
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kVerifyConsistency));
+      TimestampSizeConsistencyMode::kVerifyConsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true));
   std::unique_ptr<WriteBatch> new_batch(nullptr);
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kReconcileInconsistency, &new_batch));
+      TimestampSizeConsistencyMode::kReconcileInconsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true, &new_batch));
   ASSERT_TRUE(new_batch.get() == nullptr);
 }
 
@@ -263,11 +267,13 @@ TEST_F(HandleTimestampSizeDifferenceTest, InvolvedColumnFamiliesConsistent) {
   // All `check_mode` pass with OK status and `batch` not updated.
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kVerifyConsistency));
+      TimestampSizeConsistencyMode::kVerifyConsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true));
   std::unique_ptr<WriteBatch> new_batch(nullptr);
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kReconcileInconsistency, &new_batch));
+      TimestampSizeConsistencyMode::kReconcileInconsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true, &new_batch));
   ASSERT_TRUE(new_batch.get() == nullptr);
 }
 
@@ -282,13 +288,15 @@ TEST_F(HandleTimestampSizeDifferenceTest,
   // families.
   ASSERT_TRUE(HandleWriteBatchTimestampSizeDifference(
                   &batch, running_ts_sz, record_ts_sz,
-                  TimestampSizeConsistencyMode::kVerifyConsistency)
+                  TimestampSizeConsistencyMode::kVerifyConsistency,
+                  /* seq_per_batch */ false, /* batch_per_txn */ true)
                   .IsInvalidArgument());
 
   std::unique_ptr<WriteBatch> new_batch(nullptr);
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kReconcileInconsistency, &new_batch));
+      TimestampSizeConsistencyMode::kReconcileInconsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true, &new_batch));
   ASSERT_TRUE(new_batch.get() != nullptr);
   CheckContentsWithTimestampStripping(batch, *new_batch, sizeof(uint64_t),
                                       std::nullopt /* dropped_cf */);
@@ -307,13 +315,15 @@ TEST_F(HandleTimestampSizeDifferenceTest,
   // families.
   ASSERT_TRUE(HandleWriteBatchTimestampSizeDifference(
                   &batch, running_ts_sz, record_ts_sz,
-                  TimestampSizeConsistencyMode::kVerifyConsistency)
+                  TimestampSizeConsistencyMode::kVerifyConsistency,
+                  /* seq_per_batch */ false, /* batch_per_txn */ true)
                   .IsInvalidArgument());
 
   std::unique_ptr<WriteBatch> new_batch(nullptr);
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kReconcileInconsistency, &new_batch));
+      TimestampSizeConsistencyMode::kReconcileInconsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true, &new_batch));
   ASSERT_TRUE(new_batch.get() != nullptr);
   CheckContentsWithTimestampPadding(batch, *new_batch, sizeof(uint64_t));
 }
@@ -331,7 +341,8 @@ TEST_F(HandleTimestampSizeDifferenceTest,
   // and all related entries copied over to the new WriteBatch.
   ASSERT_OK(HandleWriteBatchTimestampSizeDifference(
       &batch, running_ts_sz, record_ts_sz,
-      TimestampSizeConsistencyMode::kReconcileInconsistency, &new_batch));
+      TimestampSizeConsistencyMode::kReconcileInconsistency,
+      /* seq_per_batch */ false, /* batch_per_txn */ true, &new_batch));
 
   ASSERT_TRUE(new_batch.get() != nullptr);
   CheckContentsWithTimestampStripping(batch, *new_batch, sizeof(uint64_t),
@@ -346,12 +357,14 @@ TEST_F(HandleTimestampSizeDifferenceTest, UnrecoverableInconsistency) {
 
   ASSERT_TRUE(HandleWriteBatchTimestampSizeDifference(
                   &batch, running_ts_sz, record_ts_sz,
-                  TimestampSizeConsistencyMode::kVerifyConsistency)
+                  TimestampSizeConsistencyMode::kVerifyConsistency,
+                  /* seq_per_batch */ false, /* batch_per_txn */ true)
                   .IsInvalidArgument());
 
   ASSERT_TRUE(HandleWriteBatchTimestampSizeDifference(
                   &batch, running_ts_sz, record_ts_sz,
-                  TimestampSizeConsistencyMode::kReconcileInconsistency)
+                  TimestampSizeConsistencyMode::kReconcileInconsistency,
+                  /* seq_per_batch */ false, /* batch_per_txn */ true)
                   .IsInvalidArgument());
 }
 

utilities/transactions/optimistic_transaction_test.cc

@@ -2222,6 +2222,241 @@ TEST_P(OptimisticTransactionTest, EntityReadSanityChecks) {
   }
 }
 
+TEST_P(OptimisticTransactionTest, AttributeGroupIterator) {
+  ColumnFamilyOptions cf_opts;
+  cf_opts.enable_blob_files = true;
+
+  ColumnFamilyHandle* cfh1 = nullptr;
+  ASSERT_OK(txn_db->CreateColumnFamily(cf_opts, "cf1", &cfh1));
+  std::unique_ptr<ColumnFamilyHandle> cfh1_guard(cfh1);
+
+  ColumnFamilyHandle* cfh2 = nullptr;
+  ASSERT_OK(txn_db->CreateColumnFamily(cf_opts, "cf2", &cfh2));
+  std::unique_ptr<ColumnFamilyHandle> cfh2_guard(cfh2);
+
+  // Note: "cf1" keys are present only in CF1; "cf2" keys are only present in
+  // CF2; "cf12" keys are present in both CFs. "a" keys are present only in the
+  // database; "b" keys are present only in the transaction; "c" keys are
+  // present in both the database and the transaction. The values indicate the
+  // column family as well as whether the entry came from the database or the
+  // transaction.
+
+  ASSERT_OK(txn_db->Put(WriteOptions(), cfh1, "cf1_a", "cf1_a_db_cf1"));
+  ASSERT_OK(txn_db->Put(WriteOptions(), cfh1, "cf1_c", "cf1_c_db_cf1"));
+
+  ASSERT_OK(txn_db->Put(WriteOptions(), cfh2, "cf2_a", "cf2_a_db_cf2"));
+  ASSERT_OK(txn_db->Put(WriteOptions(), cfh2, "cf2_c", "cf2_c_db_cf2"));
+
+  ASSERT_OK(txn_db->Put(WriteOptions(), cfh1, "cf12_a", "cf12_a_db_cf1"));
+  ASSERT_OK(txn_db->Put(WriteOptions(), cfh2, "cf12_a", "cf12_a_db_cf2"));
+  ASSERT_OK(txn_db->Put(WriteOptions(), cfh1, "cf12_c", "cf12_c_db_cf1"));
+  ASSERT_OK(txn_db->Put(WriteOptions(), cfh2, "cf12_c", "cf12_c_db_cf2"));
+
+  ASSERT_OK(txn_db->Flush(FlushOptions(), cfh1));
+  ASSERT_OK(txn_db->Flush(FlushOptions(), cfh2));
+
+  std::unique_ptr<Transaction> txn(txn_db->BeginTransaction(WriteOptions()));
+
+  ASSERT_OK(txn->Put(cfh1, "cf1_b", "cf1_b_txn_cf1"));
+  ASSERT_OK(txn->Put(cfh1, "cf1_c", "cf1_c_txn_cf1"));
+
+  ASSERT_OK(txn->Put(cfh2, "cf2_b", "cf2_b_txn_cf2"));
+  ASSERT_OK(txn->Put(cfh2, "cf2_c", "cf2_c_txn_cf2"));
+
+  ASSERT_OK(txn->Put(cfh1, "cf12_b", "cf12_b_txn_cf1"));
+  ASSERT_OK(txn->Put(cfh2, "cf12_b", "cf12_b_txn_cf2"));
+  ASSERT_OK(txn->Put(cfh1, "cf12_c", "cf12_c_txn_cf1"));
+  ASSERT_OK(txn->Put(cfh2, "cf12_c", "cf12_c_txn_cf2"));
+
+  auto verify = [&](bool allow_unprepared_value, auto prepare_if_needed) {
+    ReadOptions read_options;
+    read_options.allow_unprepared_value = allow_unprepared_value;
+
+    std::unique_ptr<AttributeGroupIterator> iter(
+        txn->GetAttributeGroupIterator(read_options, {cfh1, cfh2}));
+
+    {
+      iter->SeekToFirst();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf12_a");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf12_a_db_cf1"}};
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf12_a_db_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns},
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf12_b");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf12_b_txn_cf1"}};
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf12_b_txn_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns},
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf12_c");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf12_c_txn_cf1"}};
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf12_c_txn_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns},
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf1_a");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf1_a_db_cf1"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf1_b");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf1_b_txn_cf1"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf1_c");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf1_c_txn_cf1"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf2_a");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf2_a_db_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf2_b");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf2_b_txn_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf2_c");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf2_c_txn_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_FALSE(iter->Valid());
+      ASSERT_OK(iter->status());
+    }
+  };
+
+  verify(/* allow_unprepared_value */ false, [](AttributeGroupIterator*) {});
+  verify(/* allow_unprepared_value */ true, [](AttributeGroupIterator* iter) {
+    ASSERT_TRUE(iter->attribute_groups().empty());
+    ASSERT_TRUE(iter->PrepareValue());
+  });
+}
+
+TEST_P(OptimisticTransactionTest, AttributeGroupIteratorSanityChecks) {
+  ColumnFamilyOptions cf1_opts;
+  ColumnFamilyHandle* cfh1 = nullptr;
+  ASSERT_OK(txn_db->CreateColumnFamily(cf1_opts, "cf1", &cfh1));
+  std::unique_ptr<ColumnFamilyHandle> cfh1_guard(cfh1);
+
+  ColumnFamilyOptions cf2_opts;
+  cf2_opts.comparator = ReverseBytewiseComparator();
+  ColumnFamilyHandle* cfh2 = nullptr;
+  ASSERT_OK(txn_db->CreateColumnFamily(cf2_opts, "cf2", &cfh2));
+  std::unique_ptr<ColumnFamilyHandle> cfh2_guard(cfh2);
+
+  std::unique_ptr<Transaction> txn(txn_db->BeginTransaction(WriteOptions()));
+
+  {
+    std::unique_ptr<AttributeGroupIterator> iter(
+        txn->GetAttributeGroupIterator(ReadOptions(), {}));
+    ASSERT_TRUE(iter->status().IsInvalidArgument());
+  }
+
+  {
+    std::unique_ptr<AttributeGroupIterator> iter(
+        txn->GetAttributeGroupIterator(ReadOptions(), {cfh1, cfh2}));
+    ASSERT_TRUE(iter->status().IsInvalidArgument());
+  }
+
+  {
+    ReadOptions read_options;
+    read_options.io_activity = Env::IOActivity::kCompaction;
+
+    std::unique_ptr<AttributeGroupIterator> iter(
+        txn->GetAttributeGroupIterator(read_options, {cfh1}));
+    ASSERT_TRUE(iter->status().IsInvalidArgument());
+  }
+}
+
 INSTANTIATE_TEST_CASE_P(
     InstanceOccGroup, OptimisticTransactionTest,
     testing::Values(OccValidationPolicy::kValidateSerial,

utilities/transactions/transaction_base.cc

@@ -7,6 +7,7 @@
 
 #include <cinttypes>
 
+#include "db/attribute_group_iterator_impl.h"
 #include "db/column_family.h"
 #include "db/db_impl/db_impl.h"
 #include "logging/logging.h"
@@ -486,6 +487,64 @@ Iterator* TransactionBaseImpl::GetIterator(const ReadOptions& read_options,
                                           &read_options);
 }
 
+template <typename IterType, typename ImplType, typename ErrorIteratorFuncType>
+std::unique_ptr<IterType> TransactionBaseImpl::NewMultiCfIterator(
+    const ReadOptions& read_options,
+    const std::vector<ColumnFamilyHandle*>& column_families,
+    ErrorIteratorFuncType error_iterator_func) {
+  if (column_families.empty()) {
+    return error_iterator_func(
+        Status::InvalidArgument("No Column Family was provided"));
+  }
+
+  const Comparator* const first_comparator =
+      column_families[0]->GetComparator();
+  assert(first_comparator);
+
+  for (size_t i = 1; i < column_families.size(); ++i) {
+    const Comparator* cf_comparator = column_families[i]->GetComparator();
+    assert(cf_comparator);
+
+    if (first_comparator != cf_comparator &&
+        first_comparator->GetId() != cf_comparator->GetId()) {
+      return error_iterator_func(Status::InvalidArgument(
+          "Different comparators are being used across CFs"));
+    }
+  }
+
+  std::vector<Iterator*> child_iterators;
+  const Status s =
+      db_->NewIterators(read_options, column_families, &child_iterators);
+  if (!s.ok()) {
+    return error_iterator_func(s);
+  }
+
+  assert(column_families.size() == child_iterators.size());
+
+  std::vector<std::pair<ColumnFamilyHandle*, std::unique_ptr<Iterator>>>
+      cfh_iter_pairs;
+  cfh_iter_pairs.reserve(column_families.size());
+  for (size_t i = 0; i < column_families.size(); ++i) {
+    cfh_iter_pairs.emplace_back(
+        column_families[i],
+        write_batch_.NewIteratorWithBase(column_families[i], child_iterators[i],
+                                         &read_options));
+  }
+
+  return std::make_unique<ImplType>(read_options,
+                                    column_families[0]->GetComparator(),
+                                    std::move(cfh_iter_pairs));
+}
+
+std::unique_ptr<AttributeGroupIterator>
+TransactionBaseImpl::GetAttributeGroupIterator(
+    const ReadOptions& read_options,
+    const std::vector<ColumnFamilyHandle*>& column_families) {
+  return NewMultiCfIterator<AttributeGroupIterator, AttributeGroupIteratorImpl>(
+      read_options, column_families,
+      [](const Status& s) { return NewAttributeGroupErrorIterator(s); });
+}
+
 Status TransactionBaseImpl::PutEntityImpl(ColumnFamilyHandle* column_family,
                                           const Slice& key,
                                           const WideColumns& columns,

utilities/transactions/transaction_base.h

@@ -146,6 +146,10 @@ class TransactionBaseImpl : public Transaction {
   Iterator* GetIterator(const ReadOptions& read_options,
                         ColumnFamilyHandle* column_family) override;
 
+  std::unique_ptr<AttributeGroupIterator> GetAttributeGroupIterator(
+      const ReadOptions& read_options,
+      const std::vector<ColumnFamilyHandle*>& column_families) override;
+
   Status Put(ColumnFamilyHandle* column_family, const Slice& key,
              const Slice& value, const bool assume_tracked = false) override;
   Status Put(const Slice& key, const Slice& value) override {
@@ -304,6 +308,13 @@ class TransactionBaseImpl : public Transaction {
   LockTracker& GetTrackedLocks() { return *tracked_locks_; }
 
  protected:
+  template <typename IterType, typename ImplType,
+            typename ErrorIteratorFuncType>
+  std::unique_ptr<IterType> NewMultiCfIterator(
+      const ReadOptions& read_options,
+      const std::vector<ColumnFamilyHandle*>& column_families,
+      ErrorIteratorFuncType error_iterator_func);
+
   Status GetImpl(const ReadOptions& options, ColumnFamilyHandle* column_family,
                  const Slice& key, std::string* value) override;
 

utilities/transactions/transaction_test.cc

@@ -13,6 +13,7 @@
 
 #include "db/db_impl/db_impl.h"
 #include "port/port.h"
+#include "rocksdb/attribute_groups.h"
 #include "rocksdb/db.h"
 #include "rocksdb/options.h"
 #include "rocksdb/perf_context.h"
@@ -7464,6 +7465,273 @@ TEST_P(TransactionTest, PutEntityRecovery) {
   }
 }
 
+TEST_P(TransactionTest, AttributeGroupIterator) {
+  const TxnDBWritePolicy write_policy = std::get<2>(GetParam());
+  if (write_policy != TxnDBWritePolicy::WRITE_COMMITTED) {
+    ROCKSDB_GTEST_BYPASS("Test only WriteCommitted for now");
+    return;
+  }
+
+  ColumnFamilyOptions cf_opts;
+  cf_opts.enable_blob_files = true;
+
+  ColumnFamilyHandle* cfh1 = nullptr;
+  ASSERT_OK(db->CreateColumnFamily(cf_opts, "cf1", &cfh1));
+  std::unique_ptr<ColumnFamilyHandle> cfh1_guard(cfh1);
+
+  ColumnFamilyHandle* cfh2 = nullptr;
+  ASSERT_OK(db->CreateColumnFamily(cf_opts, "cf2", &cfh2));
+  std::unique_ptr<ColumnFamilyHandle> cfh2_guard(cfh2);
+
+  // Note: "cf1" keys are present only in CF1; "cf2" keys are only present in
+  // CF2; "cf12" keys are present in both CFs. "a" keys are present only in the
+  // database; "b" keys are present only in the transaction; "c" keys are
+  // present in both the database and the transaction. The values indicate the
+  // column family as well as whether the entry came from the database or the
+  // transaction.
+
+  ASSERT_OK(db->Put(WriteOptions(), cfh1, "cf1_a", "cf1_a_db_cf1"));
+  ASSERT_OK(db->Put(WriteOptions(), cfh1, "cf1_c", "cf1_c_db_cf1"));
+
+  ASSERT_OK(db->Put(WriteOptions(), cfh2, "cf2_a", "cf2_a_db_cf2"));
+  ASSERT_OK(db->Put(WriteOptions(), cfh2, "cf2_c", "cf2_c_db_cf2"));
+
+  ASSERT_OK(db->Put(WriteOptions(), cfh1, "cf12_a", "cf12_a_db_cf1"));
+  ASSERT_OK(db->Put(WriteOptions(), cfh2, "cf12_a", "cf12_a_db_cf2"));
+  ASSERT_OK(db->Put(WriteOptions(), cfh1, "cf12_c", "cf12_c_db_cf1"));
+  ASSERT_OK(db->Put(WriteOptions(), cfh2, "cf12_c", "cf12_c_db_cf2"));
+
+  ASSERT_OK(db->Flush(FlushOptions(), cfh1));
+  ASSERT_OK(db->Flush(FlushOptions(), cfh2));
+
+  std::unique_ptr<Transaction> txn(db->BeginTransaction(WriteOptions()));
+
+  ASSERT_OK(txn->Put(cfh1, "cf1_b", "cf1_b_txn_cf1"));
+  ASSERT_OK(txn->Put(cfh1, "cf1_c", "cf1_c_txn_cf1"));
+
+  ASSERT_OK(txn->Put(cfh2, "cf2_b", "cf2_b_txn_cf2"));
+  ASSERT_OK(txn->Put(cfh2, "cf2_c", "cf2_c_txn_cf2"));
+
+  ASSERT_OK(txn->Put(cfh1, "cf12_b", "cf12_b_txn_cf1"));
+  ASSERT_OK(txn->Put(cfh2, "cf12_b", "cf12_b_txn_cf2"));
+  ASSERT_OK(txn->Put(cfh1, "cf12_c", "cf12_c_txn_cf1"));
+  ASSERT_OK(txn->Put(cfh2, "cf12_c", "cf12_c_txn_cf2"));
+
+  auto verify = [&](bool allow_unprepared_value, auto prepare_if_needed) {
+    ReadOptions read_options;
+    read_options.allow_unprepared_value = allow_unprepared_value;
+
+    std::unique_ptr<AttributeGroupIterator> iter(
+        txn->GetAttributeGroupIterator(read_options, {cfh1, cfh2}));
+
+    {
+      iter->SeekToFirst();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf12_a");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf12_a_db_cf1"}};
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf12_a_db_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns},
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf12_b");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf12_b_txn_cf1"}};
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf12_b_txn_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns},
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf12_c");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf12_c_txn_cf1"}};
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf12_c_txn_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns},
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf1_a");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf1_a_db_cf1"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf1_b");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf1_b_txn_cf1"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf1_c");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf1_columns{{kDefaultWideColumnName, "cf1_c_txn_cf1"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh1, &cf1_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf2_a");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf2_a_db_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf2_b");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf2_b_txn_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), "cf2_c");
+
+      prepare_if_needed(iter.get());
+
+      WideColumns cf2_columns{{kDefaultWideColumnName, "cf2_c_txn_cf2"}};
+      IteratorAttributeGroups expected{
+          IteratorAttributeGroup{cfh2, &cf2_columns}};
+      ASSERT_EQ(iter->attribute_groups(), expected);
+    }
+
+    {
+      iter->Next();
+      ASSERT_FALSE(iter->Valid());
+      ASSERT_OK(iter->status());
+    }
+  };
+
+  verify(/* allow_unprepared_value */ false, [](AttributeGroupIterator*) {});
+  verify(/* allow_unprepared_value */ true, [](AttributeGroupIterator* iter) {
+    ASSERT_TRUE(iter->attribute_groups().empty());
+    ASSERT_TRUE(iter->PrepareValue());
+  });
+}
+
+TEST_P(TransactionTest, AttributeGroupIteratorSanityChecks) {
+  ColumnFamilyOptions cf1_opts;
+  ColumnFamilyHandle* cfh1 = nullptr;
+  ASSERT_OK(db->CreateColumnFamily(cf1_opts, "cf1", &cfh1));
+  std::unique_ptr<ColumnFamilyHandle> cfh1_guard(cfh1);
+
+  ColumnFamilyOptions cf2_opts;
+  cf2_opts.comparator = ReverseBytewiseComparator();
+  ColumnFamilyHandle* cfh2 = nullptr;
+  ASSERT_OK(db->CreateColumnFamily(cf2_opts, "cf2", &cfh2));
+  std::unique_ptr<ColumnFamilyHandle> cfh2_guard(cfh2);
+
+  std::unique_ptr<Transaction> txn(db->BeginTransaction(WriteOptions()));
+
+  const TxnDBWritePolicy write_policy = std::get<2>(GetParam());
+  if (write_policy != TxnDBWritePolicy::WRITE_COMMITTED) {
+    {
+      std::unique_ptr<AttributeGroupIterator> iter(
+          txn->GetAttributeGroupIterator(ReadOptions(), {}));
+      ASSERT_TRUE(iter->status().IsNotSupported());
+    }
+
+    {
+      std::unique_ptr<AttributeGroupIterator> iter(
+          txn->GetAttributeGroupIterator(ReadOptions(), {cfh1, cfh2}));
+      ASSERT_TRUE(iter->status().IsNotSupported());
+    }
+
+    {
+      ReadOptions read_options;
+      read_options.io_activity = Env::IOActivity::kCompaction;
+
+      std::unique_ptr<AttributeGroupIterator> iter(
+          txn->GetAttributeGroupIterator(read_options, {cfh1}));
+      ASSERT_TRUE(iter->status().IsNotSupported());
+    }
+
+    return;
+  }
+
+  {
+    std::unique_ptr<AttributeGroupIterator> iter(
+        txn->GetAttributeGroupIterator(ReadOptions(), {}));
+    ASSERT_TRUE(iter->status().IsInvalidArgument());
+  }
+
+  {
+    std::unique_ptr<AttributeGroupIterator> iter(
+        txn->GetAttributeGroupIterator(ReadOptions(), {cfh1, cfh2}));
+    ASSERT_TRUE(iter->status().IsInvalidArgument());
+  }
+
+  {
+    ReadOptions read_options;
+    read_options.io_activity = Env::IOActivity::kCompaction;
+
+    std::unique_ptr<AttributeGroupIterator> iter(
+        txn->GetAttributeGroupIterator(read_options, {cfh1}));
+    ASSERT_TRUE(iter->status().IsInvalidArgument());
+  }
+}
+
 TEST_F(TransactionDBTest, CollapseKey) {
   ASSERT_OK(ReOpen());
   ASSERT_OK(db->Put({}, "hello", "world"));

utilities/transactions/write_committed_transaction_ts_test.cc

@@ -418,6 +418,21 @@ TEST_P(WriteCommittedTxnWithTsTest, RecoverFromWal) {
   ASSERT_OK(txn3->Prepare());
   txn3.reset();
 
+  std::unique_ptr<Transaction> txn4(
+      NewTxn(WriteOptions(), TransactionOptions()));
+  assert(txn4);
+  ASSERT_OK(txn4->SetName("no_op_txn"));
+  txn4.reset();
+
+  std::unique_ptr<Transaction> rolled_back_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  ASSERT_NE(nullptr, rolled_back_txn);
+  ASSERT_OK(rolled_back_txn->Put("non_exist0", "donotcare"));
+  ASSERT_OK(rolled_back_txn->Put(handles_[1], "non_exist1", "donotcare"));
+  ASSERT_OK(rolled_back_txn->SetName("rolled_back_txn"));
+  ASSERT_OK(rolled_back_txn->Rollback());
+  rolled_back_txn.reset();
+
   ASSERT_OK(ReOpenNoDelete(cf_descs, &handles_));
 
   {
@@ -452,9 +467,318 @@ TEST_P(WriteCommittedTxnWithTsTest, RecoverFromWal) {
 
     s = GetFromDb(ReadOptions(), handles_[1], "baz", /*ts=*/24, &value);
     ASSERT_TRUE(s.IsNotFound());
+
+    Transaction* no_op_txn = db->GetTransactionByName("no_op_txn");
+    ASSERT_EQ(nullptr, no_op_txn);
+
+    s = db->Get(ReadOptions(), handles_[0], "non_exist0", &value);
+    ASSERT_TRUE(s.IsNotFound());
+
+    s = GetFromDb(ReadOptions(), handles_[1], "non_exist1", /*ts=*/24, &value);
+    ASSERT_TRUE(s.IsNotFound());
   }
 }
 
+TEST_P(WriteCommittedTxnWithTsTest, EnabledUDTDisabledRecoverFromWal) {
+  // This feature is not compatible with UDT in memtable only.
+  options.allow_concurrent_memtable_write = false;
+  ASSERT_OK(ReOpenNoDelete());
+
+  ColumnFamilyOptions cf_opts;
+  cf_opts.comparator = test::BytewiseComparatorWithU64TsWrapper();
+  cf_opts.persist_user_defined_timestamps = false;
+  const std::string test_cf_name = "test_cf";
+  ColumnFamilyHandle* cfh = nullptr;
+  assert(db);
+  ASSERT_OK(db->CreateColumnFamily(cf_opts, test_cf_name, &cfh));
+  delete cfh;
+  cfh = nullptr;
+
+  std::vector<ColumnFamilyDescriptor> cf_descs;
+  cf_descs.emplace_back(kDefaultColumnFamilyName, options);
+  cf_descs.emplace_back(test_cf_name, cf_opts);
+  options.avoid_flush_during_shutdown = true;
+  ASSERT_OK(ReOpenNoDelete(cf_descs, &handles_));
+
+  std::unique_ptr<Transaction> no_op_txn(
+      NewTxn(WriteOptions(), TransactionOptions()));
+  ASSERT_NE(nullptr, no_op_txn);
+  ASSERT_OK(no_op_txn->SetName("no_op_txn"));
+  no_op_txn.reset();
+
+  std::unique_ptr<Transaction> prepared_but_uncommitted_txn(
+      NewTxn(WriteOptions(), TransactionOptions()));
+  ASSERT_NE(nullptr, prepared_but_uncommitted_txn);
+  ASSERT_OK(prepared_but_uncommitted_txn->Put("foo0", "foo_value_0"));
+  ASSERT_OK(
+      prepared_but_uncommitted_txn->Put(handles_[1], "foo1", "foo_value_1"));
+  ASSERT_OK(
+      prepared_but_uncommitted_txn->SetName("prepared_but_uncommitted_txn"));
+  ASSERT_OK(prepared_but_uncommitted_txn->Prepare());
+
+  prepared_but_uncommitted_txn.reset();
+
+  WriteOptions write_opts;
+  write_opts.sync = true;
+  std::unique_ptr<Transaction> committed_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  ASSERT_NE(nullptr, committed_txn);
+  ASSERT_OK(committed_txn->Put("bar0", "bar_value_0"));
+  ASSERT_OK(committed_txn->Put(handles_[1], "bar1", "bar_value_1"));
+  ASSERT_OK(committed_txn->SetName("committed_txn"));
+  ASSERT_OK(committed_txn->Prepare());
+  ASSERT_OK(committed_txn->SetCommitTimestamp(/*ts=*/23));
+  ASSERT_OK(committed_txn->Commit());
+  committed_txn.reset();
+
+  std::unique_ptr<Transaction> committed_without_prepare_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  ASSERT_NE(nullptr, committed_without_prepare_txn);
+  ASSERT_OK(committed_without_prepare_txn->Put("baz0", "baz_value_0"));
+  ASSERT_OK(
+      committed_without_prepare_txn->Put(handles_[1], "baz1", "baz_value_1"));
+  ASSERT_OK(
+      committed_without_prepare_txn->SetName("committed_without_prepare_txn"));
+  ASSERT_OK(committed_without_prepare_txn->SetCommitTimestamp(/*ts=*/24));
+  ASSERT_OK(committed_without_prepare_txn->Commit());
+  committed_without_prepare_txn.reset();
+
+  std::unique_ptr<Transaction> rolled_back_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  assert(rolled_back_txn);
+  ASSERT_OK(rolled_back_txn->Put("non_exist0", "donotcare"));
+  ASSERT_OK(rolled_back_txn->Put(handles_[1], "non_exist1", "donotcare"));
+  ASSERT_OK(rolled_back_txn->SetName("rolled_back_txn"));
+  ASSERT_OK(rolled_back_txn->Rollback());
+  rolled_back_txn.reset();
+
+  // Reopen and disable UDT to replay WAL entries.
+  cf_descs[1].options.comparator = BytewiseComparator();
+  ASSERT_OK(ReOpenNoDelete(cf_descs, &handles_));
+
+  {
+    Transaction* recovered_txn0 = db->GetTransactionByName("no_op_txn");
+    ASSERT_EQ(nullptr, recovered_txn0);
+
+    Transaction* recovered_txn1 =
+        db->GetTransactionByName("prepared_but_uncommitted_txn");
+    ASSERT_NE(nullptr, recovered_txn1);
+    std::string value;
+    ASSERT_OK(recovered_txn1->Commit());
+    Status s = db->Get(ReadOptions(), handles_[0], "foo0", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("foo_value_0", value);
+    s = db->Get(ReadOptions(), handles_[1], "foo1", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("foo_value_1", value);
+    delete recovered_txn1;
+
+    ASSERT_EQ(nullptr, db->GetTransactionByName("committed_txn"));
+    s = db->Get(ReadOptions(), handles_[0], "bar0", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("bar_value_0", value);
+    s = db->Get(ReadOptions(), handles_[1], "bar1", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("bar_value_1", value);
+
+    ASSERT_EQ(nullptr,
+              db->GetTransactionByName("committed_without_prepare_txn"));
+    s = db->Get(ReadOptions(), handles_[0], "baz0", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("baz_value_0", value);
+    s = db->Get(ReadOptions(), handles_[1], "baz1", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("baz_value_1", value);
+
+    ASSERT_EQ(nullptr, db->GetTransactionByName("rolled_back_txn"));
+    s = db->Get(ReadOptions(), handles_[0], "non_exist0", &value);
+    ASSERT_TRUE(s.IsNotFound());
+    s = db->Get(ReadOptions(), handles_[1], "non_exist1", &value);
+    ASSERT_TRUE(s.IsNotFound());
+  }
+}
+
+TEST_P(WriteCommittedTxnWithTsTest, UDTNewlyEnabledRecoverFromWal) {
+  ASSERT_OK(ReOpenNoDelete());
+
+  ColumnFamilyOptions cf_opts;
+  const std::string test_cf_name = "test_cf";
+  ColumnFamilyHandle* cfh = nullptr;
+  assert(db);
+  ASSERT_OK(db->CreateColumnFamily(cf_opts, test_cf_name, &cfh));
+  delete cfh;
+  cfh = nullptr;
+
+  std::vector<ColumnFamilyDescriptor> cf_descs;
+  cf_descs.emplace_back(kDefaultColumnFamilyName, options);
+  cf_descs.emplace_back(test_cf_name, cf_opts);
+  options.avoid_flush_during_shutdown = true;
+  ASSERT_OK(ReOpenNoDelete(cf_descs, &handles_));
+
+  std::unique_ptr<Transaction> no_op_txn(
+      NewTxn(WriteOptions(), TransactionOptions()));
+  ASSERT_NE(nullptr, no_op_txn);
+  ASSERT_OK(no_op_txn->SetName("no_op_txn"));
+  no_op_txn.reset();
+
+  std::unique_ptr<Transaction> prepared_but_uncommitted_txn(
+      NewTxn(WriteOptions(), TransactionOptions()));
+  ASSERT_NE(nullptr, prepared_but_uncommitted_txn);
+  ASSERT_OK(
+      prepared_but_uncommitted_txn->Put(handles_[0], "foo0", "foo_value_0"));
+  ASSERT_OK(
+      prepared_but_uncommitted_txn->Put(handles_[1], "foo1", "foo_value_1"));
+  ASSERT_OK(
+      prepared_but_uncommitted_txn->SetName("prepared_but_uncommitted_txn"));
+  ASSERT_OK(prepared_but_uncommitted_txn->Prepare());
+
+  prepared_but_uncommitted_txn.reset();
+
+  WriteOptions write_opts;
+  write_opts.sync = true;
+  std::unique_ptr<Transaction> committed_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  ASSERT_NE(nullptr, committed_txn);
+  ASSERT_OK(committed_txn->Put("bar0", "bar_value_0"));
+  ASSERT_OK(committed_txn->Put(handles_[1], "bar1", "bar_value_1"));
+  ASSERT_OK(committed_txn->SetName("committed_txn"));
+  ASSERT_OK(committed_txn->Prepare());
+  ASSERT_OK(committed_txn->Commit());
+  committed_txn.reset();
+
+  std::unique_ptr<Transaction> committed_without_prepare_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  assert(committed_without_prepare_txn);
+  ASSERT_OK(committed_without_prepare_txn->Put("baz0", "baz_value_0"));
+  ASSERT_OK(
+      committed_without_prepare_txn->Put(handles_[1], "baz1", "baz_value_1"));
+  ASSERT_OK(
+      committed_without_prepare_txn->SetName("committed_without_prepare_txn"));
+  ASSERT_OK(committed_without_prepare_txn->Commit());
+  committed_without_prepare_txn.reset();
+
+  std::unique_ptr<Transaction> rolled_back_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  ASSERT_NE(nullptr, rolled_back_txn);
+  ASSERT_OK(rolled_back_txn->Put("non_exist0", "donotcare"));
+  ASSERT_OK(rolled_back_txn->Put(handles_[1], "non_exist1", "donotcare"));
+  ASSERT_OK(rolled_back_txn->SetName("rolled_back_txn"));
+  ASSERT_OK(rolled_back_txn->Rollback());
+  rolled_back_txn.reset();
+
+  // Reopen and enable UDT to replay WAL entries.
+  options.allow_concurrent_memtable_write = false;
+  cf_descs[1].options.comparator = test::BytewiseComparatorWithU64TsWrapper();
+  cf_descs[1].options.persist_user_defined_timestamps = false;
+  ASSERT_OK(ReOpenNoDelete(cf_descs, &handles_));
+
+  {
+    Transaction* recovered_txn1 =
+        db->GetTransactionByName("prepared_but_uncommitted_txn");
+    ASSERT_NE(nullptr, recovered_txn1);
+    std::string value;
+    ASSERT_OK(recovered_txn1->SetCommitTimestamp(23));
+    ASSERT_OK(recovered_txn1->Commit());
+    Status s = db->Get(ReadOptions(), handles_[0], "foo0", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("foo_value_0", value);
+    s = GetFromDb(ReadOptions(), handles_[1], "foo1", /*ts=*/23, &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("foo_value_1", value);
+    delete recovered_txn1;
+
+    ASSERT_EQ(nullptr, db->GetTransactionByName("committed_txn"));
+    s = db->Get(ReadOptions(), handles_[0], "bar0", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("bar_value_0", value);
+    s = GetFromDb(ReadOptions(), handles_[1], "bar1", /*ts=*/23, &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("bar_value_1", value);
+
+    ASSERT_EQ(nullptr,
+              db->GetTransactionByName("committed_without_prepare_txn"));
+    s = db->Get(ReadOptions(), handles_[0], "baz0", &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("baz_value_0", value);
+    s = GetFromDb(ReadOptions(), handles_[1], "baz1", /*ts=*/23, &value);
+    ASSERT_OK(s);
+    ASSERT_EQ("baz_value_1", value);
+
+    ASSERT_EQ(nullptr, db->GetTransactionByName("rolled_back_txn"));
+    s = db->Get(ReadOptions(), handles_[0], "non_exist0", &value);
+    ASSERT_TRUE(s.IsNotFound());
+    s = GetFromDb(ReadOptions(), handles_[1], "non_exist1", /*ts=*/23, &value);
+    ASSERT_TRUE(s.IsNotFound());
+  }
+}
+
+TEST_P(WriteCommittedTxnWithTsTest, ChangeFromWriteCommittedAndDisableUDT) {
+  // This feature is not compatible with UDT in memtable only.
+  options.allow_concurrent_memtable_write = false;
+  ASSERT_OK(ReOpenNoDelete());
+
+  ColumnFamilyOptions cf_opts;
+  cf_opts.comparator = test::BytewiseComparatorWithU64TsWrapper();
+  cf_opts.persist_user_defined_timestamps = false;
+  const std::string test_cf_name = "test_cf";
+  ColumnFamilyHandle* cfh = nullptr;
+  assert(db);
+  ASSERT_OK(db->CreateColumnFamily(cf_opts, test_cf_name, &cfh));
+  delete cfh;
+  cfh = nullptr;
+
+  std::vector<ColumnFamilyDescriptor> cf_descs;
+  cf_descs.emplace_back(kDefaultColumnFamilyName, options);
+  cf_descs.emplace_back(test_cf_name, cf_opts);
+  options.avoid_flush_during_shutdown = true;
+  ASSERT_OK(ReOpenNoDelete(cf_descs, &handles_));
+
+  std::unique_ptr<Transaction> prepared_but_uncommitted_txn(
+      NewTxn(WriteOptions(), TransactionOptions()));
+  assert(prepared_but_uncommitted_txn);
+  ASSERT_OK(prepared_but_uncommitted_txn->Put("foo0", "foo_value_0"));
+  ASSERT_OK(
+      prepared_but_uncommitted_txn->Put(handles_[1], "foo1", "foo_value_1"));
+  ASSERT_OK(
+      prepared_but_uncommitted_txn->SetName("prepared_but_uncommitted_txn"));
+  ASSERT_OK(prepared_but_uncommitted_txn->Prepare());
+
+  prepared_but_uncommitted_txn.reset();
+
+  WriteOptions write_opts;
+  write_opts.sync = true;
+  std::unique_ptr<Transaction> committed_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  assert(committed_txn);
+  ASSERT_OK(committed_txn->Put("bar0", "bar_value_0"));
+  ASSERT_OK(committed_txn->Put(handles_[1], "bar1", "bar_value_1"));
+  ASSERT_OK(committed_txn->SetName("committed_txn"));
+  ASSERT_OK(committed_txn->Prepare());
+  ASSERT_OK(committed_txn->SetCommitTimestamp(/*ts=*/23));
+  ASSERT_OK(committed_txn->Commit());
+  committed_txn.reset();
+
+  std::unique_ptr<Transaction> committed_without_prepare_txn(
+      NewTxn(write_opts, TransactionOptions()));
+  assert(committed_without_prepare_txn);
+  ASSERT_OK(committed_without_prepare_txn->Put("baz0", "baz_value_0"));
+  ASSERT_OK(
+      committed_without_prepare_txn->Put(handles_[1], "baz1", "baz_value_1"));
+  ASSERT_OK(
+      committed_without_prepare_txn->SetName("committed_without_prepare_txn"));
+  ASSERT_OK(committed_without_prepare_txn->SetCommitTimestamp(/*ts=*/24));
+  ASSERT_OK(committed_without_prepare_txn->Commit());
+  committed_without_prepare_txn.reset();
+
+  // Disable UDT and change write policy.
+  cf_descs[1].options.comparator = BytewiseComparator();
+  txn_db_options.write_policy = TxnDBWritePolicy::WRITE_PREPARED;
+  ASSERT_NOK(ReOpenNoDelete(cf_descs, &handles_));
+
+  txn_db_options.write_policy = TxnDBWritePolicy::WRITE_UNPREPARED;
+  ASSERT_NOK(ReOpenNoDelete(cf_descs, &handles_));
+}
+
 TEST_P(WriteCommittedTxnWithTsTest, TransactionDbLevelApi) {
   ASSERT_OK(ReOpenNoDelete());
 

utilities/transactions/write_prepared_txn.cc

@@ -3,13 +3,13 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 
-
 #include "utilities/transactions/write_prepared_txn.h"
 
 #include <cinttypes>
 #include <map>
 #include <set>
 
+#include "db/attribute_group_iterator_impl.h"
 #include "db/column_family.h"
 #include "db/db_impl/db_impl.h"
 #include "rocksdb/db.h"
@@ -135,6 +135,15 @@ Iterator* WritePreparedTxn::GetIterator(const ReadOptions& options,
   return write_batch_.NewIteratorWithBase(column_family, db_iter, &options);
 }
 
+std::unique_ptr<AttributeGroupIterator>
+WritePreparedTxn::GetAttributeGroupIterator(
+    const ReadOptions& /* read_options */,
+    const std::vector<ColumnFamilyHandle*>& /* column_families */) {
+  return NewAttributeGroupErrorIterator(
+      Status::NotSupported("GetAttributeGroupIterator not supported for "
+                           "write-prepared/write-unprepared transactions"));
+}
+
 Status WritePreparedTxn::PrepareInternal() {
   WriteOptions write_options = write_options_;
   write_options.disableWAL = false;

utilities/transactions/write_prepared_txn.h

@@ -5,7 +5,6 @@
 
 #pragma once
 
-
 #include <algorithm>
 #include <atomic>
 #include <mutex>
@@ -71,6 +70,10 @@ class WritePreparedTxn : public PessimisticTransaction {
   Iterator* GetIterator(const ReadOptions& options,
                         ColumnFamilyHandle* column_family) override;
 
+  std::unique_ptr<AttributeGroupIterator> GetAttributeGroupIterator(
+      const ReadOptions& read_options,
+      const std::vector<ColumnFamilyHandle*>& column_families) override;
+
   void SetSnapshot() override;
 
  protected:

utilities/write_batch_with_index/write_batch_with_index.cc

@@ -329,7 +329,8 @@ Iterator* WriteBatchWithIndex::NewIteratorWithBase(
   }
 
   return new BaseDeltaIterator(column_family, base_iterator, wbwiii,
-                               GetColumnFamilyUserComparator(column_family));
+                               GetColumnFamilyUserComparator(column_family),
+                               read_options);
 }
 
 Iterator* WriteBatchWithIndex::NewIteratorWithBase(Iterator* base_iterator) {
@@ -337,7 +338,8 @@ Iterator* WriteBatchWithIndex::NewIteratorWithBase(Iterator* base_iterator) {
   auto wbwiii = new WBWIIteratorImpl(0, &(rep->skip_list), &rep->write_batch,
                                      &rep->comparator);
   return new BaseDeltaIterator(nullptr, base_iterator, wbwiii,
-                               rep->comparator.default_comparator());
+                               rep->comparator.default_comparator(),
+                               /* read_options */ nullptr);
 }
 
 Status WriteBatchWithIndex::Put(ColumnFamilyHandle* column_family,

utilities/write_batch_with_index/write_batch_with_index_internal.cc

@@ -22,10 +22,13 @@ namespace ROCKSDB_NAMESPACE {
 BaseDeltaIterator::BaseDeltaIterator(ColumnFamilyHandle* column_family,
                                      Iterator* base_iterator,
                                      WBWIIteratorImpl* delta_iterator,
-                                     const Comparator* comparator)
+                                     const Comparator* comparator,
+                                     const ReadOptions* read_options)
     : forward_(true),
       current_at_base_(true),
       equal_keys_(false),
+      allow_unprepared_value_(
+          read_options ? read_options->allow_unprepared_value : false),
       status_(Status::OK()),
       column_family_(column_family),
       base_iterator_(base_iterator),
@@ -262,6 +265,17 @@ void BaseDeltaIterator::SetValueAndColumnsFromBase() {
   assert(value_.empty());
   assert(columns_.empty());
 
+  if (!base_iterator_->PrepareValue()) {
+    assert(!BaseValid());
+    assert(!base_iterator_->status().ok());
+
+    Invalidate(base_iterator_->status());
+
+    assert(!Valid());
+    assert(!status().ok());
+    return;
+  }
+
   value_ = base_iterator_->value();
   columns_ = base_iterator_->columns();
 }
@@ -314,6 +328,17 @@ void BaseDeltaIterator::SetValueAndColumnsFromDelta() {
         /* result_operand */ nullptr, &result_type);
   } else if (delta_entry.type == kMergeRecord) {
     if (equal_keys_) {
+      if (!base_iterator_->PrepareValue()) {
+        assert(!BaseValid());
+        assert(!base_iterator_->status().ok());
+
+        Invalidate(base_iterator_->status());
+
+        assert(!Valid());
+        assert(!status().ok());
+        return;
+      }
+
       if (WideColumnsHelper::HasDefaultColumnOnly(base_iterator_->columns())) {
         status_ = WriteBatchWithIndexInternal::MergeKeyWithBaseValue(
             column_family_, delta_entry.key, MergeHelper::kPlainBaseValue,
@@ -402,7 +427,9 @@ void BaseDeltaIterator::UpdateCurrent() {
     } else if (!DeltaValid()) {
       // Delta has finished.
       current_at_base_ = true;
-      SetValueAndColumnsFromBase();
+      if (!allow_unprepared_value_) {
+        SetValueAndColumnsFromBase();
+      }
       return;
     } else {
       int compare =
@@ -426,7 +453,9 @@ void BaseDeltaIterator::UpdateCurrent() {
         }
       } else {
         current_at_base_ = true;
-        SetValueAndColumnsFromBase();
+        if (!allow_unprepared_value_) {
+          SetValueAndColumnsFromBase();
+        }
         return;
       }
     }

utilities/write_batch_with_index/write_batch_with_index_internal.h

@@ -36,7 +36,8 @@ class BaseDeltaIterator : public Iterator {
  public:
   BaseDeltaIterator(ColumnFamilyHandle* column_family, Iterator* base_iterator,
                     WBWIIteratorImpl* delta_iterator,
-                    const Comparator* comparator);
+                    const Comparator* comparator,
+                    const ReadOptions* read_options);
 
   ~BaseDeltaIterator() override;
 
@@ -47,6 +48,23 @@ class BaseDeltaIterator : public Iterator {
   void SeekForPrev(const Slice& k) override;
   void Next() override;
   void Prev() override;
+
+  bool PrepareValue() override {
+    assert(Valid());
+
+    if (!allow_unprepared_value_) {
+      return true;
+    }
+
+    if (!current_at_base_) {
+      return true;
+    }
+
+    SetValueAndColumnsFromBase();
+
+    return Valid();
+  }
+
   Slice key() const override;
   Slice value() const override { return value_; }
   const WideColumns& columns() const override { return columns_; }
@@ -69,6 +87,7 @@ class BaseDeltaIterator : public Iterator {
   bool forward_;
   bool current_at_base_;
   bool equal_keys_;
+  bool allow_unprepared_value_;
   Status status_;
   ColumnFamilyHandle* column_family_;
   std::unique_ptr<Iterator> base_iterator_;

utilities/write_batch_with_index/write_batch_with_index_test.cc

@@ -81,73 +81,116 @@ using KVMap = std::map<std::string, std::string>;
 
 class KVIter : public Iterator {
  public:
-  explicit KVIter(const KVMap* map) : map_(map), iter_(map_->end()) {}
+  explicit KVIter(const KVMap* map, bool allow_unprepared_value = false,
+                  bool fail_prepare_value = false)
+      : map_(map),
+        iter_(map_->end()),
+        allow_unprepared_value_(allow_unprepared_value),
+        fail_prepare_value_(fail_prepare_value) {}
 
-  bool Valid() const override { return iter_ != map_->end(); }
+  bool Valid() const override { return status_.ok() && iter_ != map_->end(); }
 
   void SeekToFirst() override {
+    status_ = Status::OK();
+    Reset();
+
     iter_ = map_->begin();
 
-    if (Valid()) {
+    if (Valid() && !allow_unprepared_value_) {
       Update();
     }
   }
 
   void SeekToLast() override {
+    status_ = Status::OK();
+    Reset();
+
     if (map_->empty()) {
       iter_ = map_->end();
     } else {
       iter_ = map_->find(map_->rbegin()->first);
     }
 
-    if (Valid()) {
+    if (Valid() && !allow_unprepared_value_) {
       Update();
     }
   }
 
   void Seek(const Slice& k) override {
+    status_ = Status::OK();
+    Reset();
+
     iter_ = map_->lower_bound(k.ToString());
 
-    if (Valid()) {
+    if (Valid() && !allow_unprepared_value_) {
       Update();
     }
   }
 
   void SeekForPrev(const Slice& k) override {
+    status_ = Status::OK();
+    Reset();
+
     iter_ = map_->upper_bound(k.ToString());
     Prev();
 
-    if (Valid()) {
+    if (Valid() && !allow_unprepared_value_) {
       Update();
     }
   }
 
   void Next() override {
+    Reset();
+
     ++iter_;
 
-    if (Valid()) {
+    if (Valid() && !allow_unprepared_value_) {
       Update();
     }
   }
 
   void Prev() override {
+    Reset();
+
     if (iter_ == map_->begin()) {
       iter_ = map_->end();
       return;
     }
     --iter_;
 
-    if (Valid()) {
+    if (Valid() && !allow_unprepared_value_) {
       Update();
     }
   }
 
+  bool PrepareValue() override {
+    assert(Valid());
+
+    if (!allow_unprepared_value_) {
+      return true;
+    }
+
+    if (fail_prepare_value_) {
+      status_ = Status::Corruption("PrepareValue() failed");
+      return false;
+    }
+
+    Update();
+
+    return true;
+  }
+
   Slice key() const override { return iter_->first; }
   Slice value() const override { return value_; }
   const WideColumns& columns() const override { return columns_; }
-  Status status() const override { return Status::OK(); }
+  Status status() const override { return status_; }
 
  private:
+  void Reset() {
+    value_.clear();
+    columns_.clear();
+  }
+
   void Update() {
     assert(Valid());
 
@@ -157,8 +200,11 @@ class KVIter : public Iterator {
 
   const KVMap* const map_;
   KVMap::const_iterator iter_;
+  Status status_;
   Slice value_;
   WideColumns columns_;
+  bool allow_unprepared_value_;
+  bool fail_prepare_value_;
 };
 
 static std::string PrintContents(WriteBatchWithIndex* batch,
@@ -1693,6 +1739,279 @@ TEST_P(WriteBatchWithIndexTest, TestNewIteratorWithBaseFromWbwi) {
   ASSERT_OK(iter->status());
 }
 
+TEST_P(WriteBatchWithIndexTest, NewIteratorWithBasePrepareValue) {
+  // BaseDeltaIterator by default should call PrepareValue if it lands on the
+  // base iterator in case it was created with allow_unprepared_value=true.
+  ColumnFamilyHandleImplDummy cf1(1, BytewiseComparator());
+  KVMap map{{"a", "aa"}, {"c", "cc"}, {"e", "ee"}};
+
+  ASSERT_OK(batch_->Put(&cf1, "c", "cc1"));
+
+  {
+    std::unique_ptr<Iterator> iter(batch_->NewIteratorWithBase(
+        &cf1, new KVIter(&map, /* allow_unprepared_value */ true)));
+
+    iter->SeekToFirst();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "a");
+    ASSERT_EQ(iter->value(), "aa");
+
+    iter->Next();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "c");
+    ASSERT_EQ(iter->value(), "cc1");
+
+    iter->Next();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "e");
+    ASSERT_EQ(iter->value(), "ee");
+
+    iter->Next();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_OK(iter->status());
+
+    iter->SeekToLast();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "e");
+    ASSERT_EQ(iter->value(), "ee");
+
+    iter->Prev();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "c");
+    ASSERT_EQ(iter->value(), "cc1");
+
+    iter->Prev();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "a");
+    ASSERT_EQ(iter->value(), "aa");
+
+    iter->Prev();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_OK(iter->status());
+  }
+
+  // PrepareValue failures from the base iterator should be propagated
+  {
+    std::unique_ptr<Iterator> iter(batch_->NewIteratorWithBase(
+        &cf1, new KVIter(&map, /* allow_unprepared_value */ true,
+                         /* fail_prepare_value */ true)));
+
+    iter->SeekToFirst();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsCorruption());
+
+    iter->SeekToLast();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsCorruption());
+  }
+}
+
+TEST_P(WriteBatchWithIndexTest, NewIteratorWithBaseAllowUnpreparedValue) {
+  ColumnFamilyHandleImplDummy cf1(1, BytewiseComparator());
+  KVMap map{{"a", "aa"}, {"c", "cc"}, {"e", "ee"}};
+
+  ASSERT_OK(batch_->Put(&cf1, "c", "cc1"));
+
+  ReadOptions read_options = read_opts_;
+  read_options.allow_unprepared_value = true;
+
+  {
+    std::unique_ptr<Iterator> iter(batch_->NewIteratorWithBase(
+        &cf1, new KVIter(&map, /* allow_unprepared_value */ true),
+        &read_options));
+
+    iter->SeekToFirst();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "a");
+    ASSERT_TRUE(iter->value().empty());
+    ASSERT_TRUE(iter->PrepareValue());
+    ASSERT_EQ(iter->value(), "aa");
+
+    iter->Next();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "c");
+    ASSERT_EQ(iter->value(), "cc1");
+    // This key is served out of the delta iterator so this PrepareValue() is a
+    // no-op
+    ASSERT_TRUE(iter->PrepareValue());
+    ASSERT_EQ(iter->value(), "cc1");
+
+    iter->Next();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "e");
+    ASSERT_TRUE(iter->value().empty());
+    ASSERT_TRUE(iter->PrepareValue());
+    ASSERT_EQ(iter->value(), "ee");
+
+    iter->Next();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_OK(iter->status());
+
+    iter->SeekToLast();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "e");
+    ASSERT_TRUE(iter->value().empty());
+    ASSERT_TRUE(iter->PrepareValue());
+    ASSERT_EQ(iter->value(), "ee");
+
+    iter->Prev();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "c");
+    ASSERT_EQ(iter->value(), "cc1");
+    // This key is served out of the delta iterator so this PrepareValue() is a
+    // no-op
+    ASSERT_TRUE(iter->PrepareValue());
+    ASSERT_EQ(iter->value(), "cc1");
+
+    iter->Prev();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "a");
+    ASSERT_TRUE(iter->value().empty());
+    ASSERT_TRUE(iter->PrepareValue());
+    ASSERT_EQ(iter->value(), "aa");
+
+    iter->Prev();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_OK(iter->status());
+  }
+
+  // PrepareValue failures from the base iterator should be propagated
+  {
+    std::unique_ptr<Iterator> iter(batch_->NewIteratorWithBase(
+        &cf1,
+        new KVIter(&map, /* allow_unprepared_value */ true,
+                   /* fail_prepare_value */ true),
+        &read_options));
+
+    iter->SeekToFirst();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_FALSE(iter->PrepareValue());
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsCorruption());
+
+    iter->SeekToLast();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_FALSE(iter->PrepareValue());
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsCorruption());
+  }
+}
+
+TEST_P(WriteBatchWithIndexTest, NewIteratorWithBaseMergePrepareValue) {
+  // When performing a merge across the base and delta iterators,
+  // BaseDeltaIterator should call PrepareValue on the base iterator in case it
+  // was created with allow_unprepared_value=true. (Note: we use BlobDB here
+  // to ensure PrepareValue is not a no-op.)
+  options_.enable_blob_files = true;
+
+  ASSERT_OK(OpenDB());
+
+  ASSERT_OK(db_->Put(write_opts_, db_->DefaultColumnFamily(), "a", "aa"));
+  ASSERT_OK(db_->Put(write_opts_, db_->DefaultColumnFamily(), "c", "cc"));
+  ASSERT_OK(db_->Put(write_opts_, db_->DefaultColumnFamily(), "e", "ee"));
+  ASSERT_OK(db_->Flush(FlushOptions(), db_->DefaultColumnFamily()));
+
+  ASSERT_OK(batch_->Merge(db_->DefaultColumnFamily(), "a", "aa1"));
+  ASSERT_OK(batch_->Merge(db_->DefaultColumnFamily(), "c", "cc1"));
+  ASSERT_OK(batch_->Merge(db_->DefaultColumnFamily(), "e", "ee1"));
+
+  ReadOptions db_read_options = read_opts_;
+  db_read_options.allow_unprepared_value = true;
+
+  {
+    std::unique_ptr<Iterator> iter(batch_->NewIteratorWithBase(
+        db_->DefaultColumnFamily(),
+        db_->NewIterator(db_read_options, db_->DefaultColumnFamily()),
+        &read_opts_));
+
+    iter->SeekToFirst();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "a");
+    ASSERT_EQ(iter->value(), "aa,aa1");
+
+    iter->Next();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "c");
+    ASSERT_EQ(iter->value(), "cc,cc1");
+
+    iter->Next();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "e");
+    ASSERT_EQ(iter->value(), "ee,ee1");
+
+    iter->Next();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_OK(iter->status());
+
+    iter->SeekToLast();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "e");
+    ASSERT_EQ(iter->value(), "ee,ee1");
+
+    iter->Prev();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "c");
+    ASSERT_EQ(iter->value(), "cc,cc1");
+
+    iter->Prev();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), "a");
+    ASSERT_EQ(iter->value(), "aa,aa1");
+
+    iter->Prev();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_OK(iter->status());
+  }
+
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobFileReader::GetBlob:TamperWithResult", [](void* arg) {
+        Slice* const blob_index = static_cast<Slice*>(arg);
+        assert(blob_index);
+        assert(!blob_index->empty());
+        blob_index->remove_prefix(1);
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // PrepareValue failures from the base iterator should be propagated
+  {
+    std::unique_ptr<Iterator> iter(batch_->NewIteratorWithBase(
+        db_->DefaultColumnFamily(),
+        db_->NewIterator(db_read_options, db_->DefaultColumnFamily()),
+        &read_opts_));
+
+    iter->SeekToFirst();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsCorruption());
+
+    iter->SeekToLast();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsCorruption());
+  }
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
 TEST_P(WriteBatchWithIndexTest, TestBoundsCheckingInDeltaIterator) {
   Status s = OpenDB();
   ASSERT_OK(s);