Bump version to 10.1.3

Summary:
Public APIs like `DB::GetFullHistoryTsLow` and `DB::IncreaseFullHistoryTsLow` have such safeguarding, allowing them to only be invoked when user defined timestamp is enabled. This PR adds safeguarding into related internal APIs in `ColumnFamilyData` to properly handle the case when the UDT feature are toggled.

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

Test Plan: ./db_with_timestamp_basic_test --gtest_filter="*EnableDisableUDT*"

Reviewed By: cbi42

Differential Revision: D72475234

Pulled By: jowlyzhang

fbshipit-source-id: 194c07287e3100da95450b04c76552c9d4a86c2d

HISTORY.md

@@ -1,6 +1,36 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 10.1.3 (04/09/2025)
+### Bug Fixes
+* Fix a bug where resurrected full_history_ts_low from a previous session that enables UDT is used by this session that disables UDT.
+
+## 10.1.2 (04/07/2025)
+### Bug Fixes
+* Fix a bug where tail size of remote compaction output is not persisted in primary db's manifest
+
+## 10.1.0 (03/24/2025)
+### New Features
+* Added a new `DBOptions.calculate_sst_write_lifetime_hint_set` setting that allows to customize which compaction styles SST write lifetime hint calculation is allowed on. Today RocksDB supports only two modes `kCompactionStyleLevel` and `kCompactionStyleUniversal`.
+* Add a new field `num_l0_files` in `CompactionJobInfo` about the number of L0 files in the CF right before and after the compaction
+* Added per-key-placement feature in Remote Compaction
+* Implemented API DB::GetPropertiesOfTablesByLevel that retrieves table properties for files in each LSM tree level
+
+### Public API Changes
+* `GetAllKeyVersions()` now interprets empty slices literally, as valid keys, and uses new `OptSlice` type default value for extreme upper and lower range limits.
+* `DeleteFilesInRanges()` now takes `RangeOpt` which is based on `OptSlice`. The overload taking `RangePtr` is deprecated.
+* Add an unordered map of name/value pairs, ReadOptions::property_bag, to pass opaque options through to an external table when creating an Iterator.
+* Introduced CompactionServiceJobStatus::kAborted to allow handling aborted scenario in Schedule(), Wait() or OnInstallation() APIs in Remote Compactions.
+* format\_version < 2 in BlockBasedTableOptions is no longer supported for writing new files. Support for reading such files is deprecated and might be removed in the future. `CompressedSecondaryCacheOptions::compress_format_version == 1` is also deprecated.
+
+### Behavior Changes
+* `ldb` now returns an error if the specified `--compression_type` is not supported in the build.
+* MultiGet with snapshot and ReadOptions::read_tier = kPersistedTier will now read a consistent view across CFs (instead of potentially reading some CF before and some CF after a flush).
+* CreateColumnFamily() is no longer allowed on a read-only DB (OpenForReadOnly())
+
+### Bug Fixes
+* Fixed stats for Tiered Storage with preclude_last_level feature
+
 ## 10.0.0 (02/21/2025)
 ### New Features
 * Introduced new `auto_refresh_iterator_with_snapshot` opt-in knob that (when enabled) will periodically release obsolete memory and storage resources for as long as the iterator is making progress and its supplied `read_options.snapshot` was initialized with non-nullptr value.

db/arena_wrapped_db_iter.h

@@ -99,6 +99,10 @@ class ArenaWrappedDBIter : public Iterator {
 
   bool PrepareValue() override { return db_iter_->PrepareValue(); }
 
+  void Prepare(const std::vector<ScanOptions>& scan_opts) override {
+    db_iter_->Prepare(scan_opts);
+  }
+
   void Init(Env* env, const ReadOptions& read_options,
             const ImmutableOptions& ioptions,
             const MutableCFOptions& mutable_cf_options, const Version* version,

db/column_family.h

@@ -537,6 +537,12 @@ class ColumnFamilyData {
     assert(!ts_low.empty());
     const Comparator* ucmp = user_comparator();
     assert(ucmp);
+    // Guard against resurrected full_history_ts_low persisted in MANIFEST
+    // from previous DB sessions. This could happen if UDT was enabled and then
+    // disabled.
+    if (ucmp->timestamp_size() == 0) {
+      return;
+    }
     if (full_history_ts_low_.empty() ||
         ucmp->CompareTimestamp(ts_low, full_history_ts_low_) > 0) {
       full_history_ts_low_ = std::move(ts_low);
@@ -544,6 +550,11 @@ class ColumnFamilyData {
   }
 
   const std::string& GetFullHistoryTsLow() const {
+    const Comparator* ucmp = user_comparator();
+    assert(ucmp);
+    if (ucmp->timestamp_size() == 0) {
+      assert(full_history_ts_low_.empty());
+    }
     return full_history_ts_low_;
   }
 

db/compaction/compaction_service_job.cc

@@ -240,7 +240,8 @@ CompactionJob::ProcessKeyValueCompactionWithCompactionService(
     meta.marked_for_compaction = file.marked_for_compaction;
     meta.unique_id = file.unique_id;
     meta.temperature = file.file_temperature;
-
+    meta.tail_size =
+        FileMetaData::CalculateTailSize(file_size, file.table_properties);
     auto cfd = compaction->column_family_data();
     CompactionOutputs* compaction_outputs =
         sub_compact->Outputs(file.is_proximal_level_output);

db/compaction/compaction_service_test.cc

@@ -277,8 +277,17 @@ TEST_F(CompactionServiceTest, BasicCompactions) {
   Statistics* primary_statistics = GetPrimaryStatistics();
   Statistics* compactor_statistics = GetCompactorStatistics();
 
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "BlockBasedTable::PrefetchTail::TaiSizeNotRecorded",
+      [&](void* /* arg */) {
+        // Trigger assertion to verify precise tail prefetch size calculation
+        assert(false);
+      });
+
+  SyncPoint::GetInstance()->EnableProcessing();
   GenerateTestData();
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  SyncPoint::GetInstance()->DisableProcessing();
   VerifyTestData();
 
   auto my_cs = GetCompactionService();
@@ -380,6 +389,7 @@ TEST_F(CompactionServiceTest, BasicCompactions) {
   ASSERT_FALSE(result.stats.is_full_compaction);
 
   Close();
+  SyncPoint::GetInstance()->DisableProcessing();
 }
 
 TEST_F(CompactionServiceTest, ManualCompaction) {
@@ -890,6 +900,12 @@ TEST_F(CompactionServiceTest, TruncatedOutput) {
   Slice end(end_str);
   uint64_t comp_num = my_cs->GetCompactionNum();
 
+  // Skip calculating tail size to avoid crashing due to truncated file size
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "FileMetaData::CalculateTailSize", [&](void* arg) {
+        bool* skip = static_cast<bool*>(arg);
+        *skip = true;
+      });
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
       "CompactionServiceCompactionJob::Run:0", [&](void* arg) {
         CompactionServiceResult* compaction_result =
@@ -906,7 +922,7 @@ TEST_F(CompactionServiceTest, TruncatedOutput) {
           ASSERT_OK(s);
           ASSERT_GT(file_size, 0);
 
-          ASSERT_OK(test::TruncateFile(env_, file_name, file_size / 2));
+          ASSERT_OK(test::TruncateFile(env_, file_name, file_size / 4));
         }
       });
   SyncPoint::GetInstance()->EnableProcessing();

db/db_impl/db_impl.cc

@@ -3796,6 +3796,16 @@ bool DBImpl::KeyMayExist(const ReadOptions& read_options,
   return s.ok() || s.IsIncomplete();
 }
 
+std::unique_ptr<MultiScan> DBImpl::NewMultiScan(
+    const ReadOptions& _read_options, ColumnFamilyHandle* column_family,
+    const std::vector<ScanOptions>& scan_opts) {
+  std::unique_ptr<Iterator> iter(NewIterator(_read_options, column_family));
+  iter->Prepare(scan_opts);
+  std::unique_ptr<MultiScan> ms_iter =
+      std::make_unique<MultiScan>(scan_opts, std::move(iter));
+  return ms_iter;
+}
+
 Iterator* DBImpl::NewIterator(const ReadOptions& _read_options,
                               ColumnFamilyHandle* column_family) {
   if (_read_options.io_activity != Env::IOActivity::kUnknown &&

db/db_impl/db_impl.h

@@ -379,6 +379,11 @@ class DBImpl : public DB {
                       const std::vector<ColumnFamilyHandle*>& column_families,
                       std::vector<Iterator*>* iterators) override;
 
+  using DB::NewMultiScan;
+  std::unique_ptr<MultiScan> NewMultiScan(
+      const ReadOptions& _read_options, ColumnFamilyHandle* column_family,
+      const std::vector<ScanOptions>& scan_opts) override;
+
   const Snapshot* GetSnapshot() override;
   void ReleaseSnapshot(const Snapshot* snapshot) override;
 

db/db_iter.h

@@ -220,6 +220,17 @@ class DBIter final : public Iterator {
 
   bool PrepareValue() override;
 
+  void Prepare(const std::vector<ScanOptions>& scan_opts) override {
+    std::optional<std::vector<ScanOptions>> new_scan_opts;
+    new_scan_opts.emplace(scan_opts);
+    scan_opts_.swap(new_scan_opts);
+    if (!scan_opts.empty()) {
+      iter_.Prepare(&scan_opts_.value());
+    } else {
+      iter_.Prepare(nullptr);
+    }
+  }
+
  private:
   class BlobReader {
    public:
@@ -455,6 +466,7 @@ class DBIter final : public Iterator {
   const Slice* const timestamp_lb_;
   const size_t timestamp_size_;
   std::string saved_timestamp_;
+  std::optional<std::vector<ScanOptions>> scan_opts_;
 };
 
 // Return a new iterator that converts internal keys (yielded by

db/db_with_timestamp_basic_test.cc

@@ -19,6 +19,13 @@
 #include "utilities/merge_operators/string_append/stringappend2.h"
 
 namespace ROCKSDB_NAMESPACE {
+namespace {
+std::string EncodeAsUint64(uint64_t v) {
+  std::string dst;
+  PutFixed64(&dst, v);
+  return dst;
+}
+}  // namespace
 class DBBasicTestWithTimestamp : public DBBasicTestWithTimestampBase {
  public:
   DBBasicTestWithTimestamp()
@@ -3746,17 +3753,42 @@ INSTANTIATE_TEST_CASE_P(
         test::UserDefinedTimestampTestMode::kStripUserDefinedTimestamp,
         test::UserDefinedTimestampTestMode::kNormal));
 
-TEST_F(DBBasicTestWithTimestamp, EnableDisableUDT) {
+// Test params:
+// 1) whether to flush before close
+class EnableDisableUDTTest : public DBBasicTestWithTimestampBase,
+                             public testing::WithParamInterface<bool> {
+ public:
+  EnableDisableUDTTest()
+      : DBBasicTestWithTimestampBase("/enable_disable_udt") {}
+};
+
+INSTANTIATE_TEST_CASE_P(EnableDisableUDTTest, EnableDisableUDTTest,
+                        ::testing::Values(true, false));
+
+TEST_P(EnableDisableUDTTest, Basic) {
   Options options = CurrentOptions();
+  // Un-flushed data before close will involve a WAL replay on DB reopen.
+  bool flush_before_close = GetParam();
   options.env = env_;
-  // Create a column family without user-defined timestamps.
   options.comparator = BytewiseComparator();
   options.persist_user_defined_timestamps = true;
   DestroyAndReopen(options);
 
+  ReadOptions ropts;
+  std::string read_ts;
+  std::string value;
+  std::string key_ts;
+
   // Create one SST file, its user keys have no user-defined timestamps.
-  ASSERT_OK(db_->Put(WriteOptions(), "foo", "val1"));
-  ASSERT_OK(Flush(0));
+  ASSERT_OK(db_->Put(WriteOptions(), "foo", "val0"));
+  ASSERT_OK(db_->Put(WriteOptions(), "bar", "val0"));
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), "bar", "baz"));
+  ASSERT_OK(db_->Get(ReadOptions(), "foo", &value));
+  ASSERT_EQ("val0", value);
+  ASSERT_TRUE(db_->Get(ReadOptions(), "bar", &value).IsNotFound());
+  if (flush_before_close) {
+    ASSERT_OK(Flush(0));
+  }
   Close();
 
   // Reopen the existing column family and enable user-defined timestamps
@@ -3765,47 +3797,63 @@ TEST_F(DBBasicTestWithTimestamp, EnableDisableUDT) {
   options.persist_user_defined_timestamps = false;
   options.allow_concurrent_memtable_write = false;
   Reopen(options);
-
-  std::string value;
-  ASSERT_TRUE(db_->Get(ReadOptions(), "foo", &value).IsInvalidArgument());
-  std::string read_ts;
-  PutFixed64(&read_ts, 0);
-  ReadOptions ropts;
+  // Read data from previous session before and after compaction.
+  read_ts = EncodeAsUint64(1);
   Slice read_ts_slice = read_ts;
   ropts.timestamp = &read_ts_slice;
-  std::string key_ts;
-  // Entries in pre-existing SST files are treated as if they have minimum
-  // user-defined timestamps.
-  ASSERT_OK(db_->Get(ropts, "foo", &value, &key_ts));
-  ASSERT_EQ("val1", value);
-  ASSERT_EQ(read_ts, key_ts);
+  for (int i = 0; i < 2; i++) {
+    ASSERT_TRUE(db_->Get(ReadOptions(), "foo", &value).IsInvalidArgument());
+    // Entries in pre-existing SST files are treated as if they have minimum
+    // user-defined timestamps.
+    ASSERT_OK(db_->Get(ropts, "foo", &value, &key_ts));
+    ASSERT_EQ("val0", value);
+    ASSERT_EQ(EncodeAsUint64(0), key_ts);
+    ASSERT_TRUE(db_->Get(ropts, "bar", &value, &key_ts).IsNotFound());
+    ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+  }
 
   // Do timestamped read / write.
-  std::string write_ts;
-  PutFixed64(&write_ts, 1);
-  ASSERT_OK(db_->Put(WriteOptions(), "foo", write_ts, "val2"));
-  read_ts.clear();
-  PutFixed64(&read_ts, 1);
+  ASSERT_OK(db_->Put(WriteOptions(), "foo", EncodeAsUint64(1), "val1"));
+  ASSERT_OK(db_->Put(WriteOptions(), "bar", EncodeAsUint64(1), "val1"));
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), "bar", "baz", EncodeAsUint64(2)));
   ASSERT_OK(db_->Get(ropts, "foo", &value, &key_ts));
-  ASSERT_EQ("val2", value);
-  ASSERT_EQ(write_ts, key_ts);
+  ASSERT_EQ("val1", value);
+  ASSERT_EQ(EncodeAsUint64(1), key_ts);
+  ASSERT_OK(db_->Get(ropts, "bar", &value, &key_ts));
+  ASSERT_EQ("val1", value);
+  ASSERT_EQ(EncodeAsUint64(1), key_ts);
+  read_ts = EncodeAsUint64(2);
+  ASSERT_TRUE(db_->Get(ropts, "bar", &value, &key_ts).IsNotFound());
   // The user keys in this SST file don't have user-defined timestamps either,
   // because `persist_user_defined_timestamps` flag is set to false.
-  ASSERT_OK(Flush(0));
+  if (flush_before_close) {
+    ASSERT_OK(Flush(0));
+  }
   Close();
 
   // Reopen the existing column family while disabling user-defined timestamps.
   options.comparator = BytewiseComparator();
   Reopen(options);
 
-  ASSERT_TRUE(db_->Get(ropts, "foo", &value).IsInvalidArgument());
-  ASSERT_OK(db_->Get(ReadOptions(), "foo", &value));
-  ASSERT_EQ("val2", value);
+  // Read data from previous session before and after compaction.
+  for (int i = 0; i < 2; i++) {
+    ASSERT_TRUE(db_->Get(ropts, "foo", &value).IsInvalidArgument());
+    ASSERT_OK(db_->Get(ReadOptions(), "foo", &value));
+    ASSERT_EQ("val1", value);
+    ASSERT_TRUE(db_->Get(ReadOptions(), "bar", &value).IsNotFound());
+    ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+  }
 
   // Continue to write / read the column family without user-defined timestamps.
-  ASSERT_OK(db_->Put(WriteOptions(), "foo", "val3"));
+  ASSERT_OK(db_->Put(WriteOptions(), "foo", "val2"));
+  ASSERT_OK(db_->Put(WriteOptions(), "bar", "val2"));
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), "bar", "baz"));
   ASSERT_OK(db_->Get(ReadOptions(), "foo", &value));
-  ASSERT_EQ("val3", value);
+  ASSERT_EQ("val2", value);
+  ASSERT_TRUE(db_->Get(ReadOptions(), "bar", &value).IsNotFound());
+  if (flush_before_close) {
+    ASSERT_OK(Flush(0));
+  }
   Close();
 }
 

db/version_edit.h

@@ -380,6 +380,33 @@ struct FileMetaData {
     assert(!res || fd.smallest_seqno == fd.largest_seqno);
     return res;
   }
+
+  static uint64_t CalculateTailSize(uint64_t file_size,
+                                    const TableProperties& props) {
+#ifndef NDEBUG
+    bool skip = false;
+    TEST_SYNC_POINT_CALLBACK("FileMetaData::CalculateTailSize", &skip);
+    if (skip) {
+      return 0;
+    }
+#endif  // NDEBUG
+    uint64_t tail_size = 0;
+
+    // Differentiate between a file with no data blocks (tail_start_offset = 0)
+    // and a file with unknown tail_start_offset (also set to 0 due to
+    // non-negative integer storage limitation)
+    bool contain_no_data_blocks =
+        props.num_entries == 0 ||
+        (props.num_entries > 0 &&
+         (props.num_entries == props.num_range_deletions));
+
+    if (props.tail_start_offset > 0 || contain_no_data_blocks) {
+      assert(props.tail_start_offset <= file_size);
+      tail_size = file_size - props.tail_start_offset;
+    }
+
+    return tail_size;
+  }
 };
 
 // A compressed copy of file meta data that just contain minimum data needed

db/version_set.cc

@@ -1002,7 +1002,8 @@ class LevelIterator final : public InternalIterator {
         skip_filters_(skip_filters),
         allow_unprepared_value_(allow_unprepared_value),
         is_next_read_sequential_(false),
-        to_return_sentinel_(false) {
+        to_return_sentinel_(false),
+        scan_opts_(nullptr) {
     // Empty level is not supported.
     assert(flevel_ != nullptr && flevel_->num_files > 0);
     if (range_tombstone_iter_ptr_) {
@@ -1098,6 +1099,13 @@ class LevelIterator final : public InternalIterator {
     read_seq_ = read_seq;
   }
 
+  void Prepare(const std::vector<ScanOptions>* scan_opts) override {
+    scan_opts_ = scan_opts;
+    if (file_iter_.iter()) {
+      file_iter_.Prepare(scan_opts_);
+    }
+  }
+
  private:
   // Return true if at least one invalid file is seen and skipped.
   bool SkipEmptyFileForward();
@@ -1223,6 +1231,7 @@ class LevelIterator final : public InternalIterator {
   bool prefix_exhausted_ = false;
   // Whether next/prev key is a sentinel key.
   bool to_return_sentinel_ = false;
+  const std::vector<ScanOptions>* scan_opts_;
 
   // Sets flags for if we should return the sentinel key next.
   // The condition for returning sentinel is reaching the end of current
@@ -1533,6 +1542,11 @@ void LevelIterator::SetFileIterator(InternalIterator* iter) {
   }
 
   InternalIterator* old_iter = file_iter_.Set(iter);
+  // Since this is a new table iterator, no need to call Prepare() if
+  // scan_opts_ is null
+  if (iter && scan_opts_) {
+    file_iter_.Prepare(scan_opts_);
+  }
 
   // Update the read pattern for PrefetchBuffer.
   if (is_next_read_sequential_) {

include/rocksdb/advanced_iterator.h

@@ -0,0 +1,36 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include "rocksdb/slice.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+enum class IterBoundCheck : char {
+  kUnknown = 0,
+  kOutOfBound,
+  kInbound,
+};
+
+// This structure encapsulates the result of NextAndGetResult()
+struct IterateResult {
+  // The lifetime of key is guaranteed until Next()/NextAndGetResult() is
+  // called.
+  Slice key;
+  // If the iterator becomes invalid after a NextAndGetResult(), the table
+  // iterator should set this to indicate whether it became invalid due
+  // to the next key being out of bound (kOutOfBound) or it reached end
+  // of file (kUnknown). If the iiterator is still valid, this should
+  // be set to kInbound.
+  IterBoundCheck bound_check_result = IterBoundCheck::kUnknown;
+  // If false, PrepareValue() needs to be called before value()
+  // This is useful if the table reader wants to materialize the value in a
+  // lazy manner. In that case, it can set this to false and RocksDB
+  // guarantees that it'll call PrepareValue() before calling value().
+  bool value_prepared = true;
+};
+
+}  // namespace ROCKSDB_NAMESPACE

include/rocksdb/db.h

@@ -22,6 +22,7 @@
 #include "rocksdb/iterator.h"
 #include "rocksdb/listener.h"
 #include "rocksdb/metadata.h"
+#include "rocksdb/multi_scan.h"
 #include "rocksdb/options.h"
 #include "rocksdb/snapshot.h"
 #include "rocksdb/sst_file_writer.h"
@@ -1073,6 +1074,18 @@ class DB {
       const ReadOptions& options,
       const std::vector<ColumnFamilyHandle*>& column_families) = 0;
 
+  // Get an iterator that scans multiple key ranges. The scan ranges should
+  // be in increasing order of start key. See multi_scan_iterator.h for more
+  // details.
+  virtual std::unique_ptr<MultiScan> NewMultiScan(
+      const ReadOptions& /*options*/, ColumnFamilyHandle* /*column_family*/,
+      const std::vector<ScanOptions>& /*scan_opts*/) {
+    std::unique_ptr<Iterator> iter(NewErrorIterator(Status::NotSupported()));
+    std::unique_ptr<MultiScan> ms_iter =
+        std::make_unique<MultiScan>(std::move(iter));
+    return ms_iter;
+  }
+
   // Return a handle to the current DB state.  Iterators created with
   // this handle will all observe a stable snapshot of the current DB
   // state.  The caller must call ReleaseSnapshot(result) when the

include/rocksdb/external_table.h

@@ -5,9 +5,10 @@
 
 #pragma once
 
+#include "rocksdb/advanced_iterator.h"
 #include "rocksdb/customizable.h"
 #include "rocksdb/file_checksum.h"
-#include "rocksdb/iterator.h"
+#include "rocksdb/iterator_base.h"
 #include "rocksdb/options.h"
 #include "rocksdb/status.h"
 
@@ -58,6 +59,47 @@ class ExternalTableFactory;
 // the external table implementation.
 // TODO: Specify which options are relevant
 
+class ExternalTableIterator : public IteratorBase {
+ public:
+  virtual ~ExternalTableIterator() {}
+
+  // This can optionally be called to prepare the iterator for a series
+  // of scans. The scan_opts parameter specifies the order of scans to
+  // follow, as well as the limits for those scans. After calling this,
+  // the caller will Seek() the iterator to successive start keys in scan_opts.
+  //
+  // If Prepare() is called again with a different scan_opts pointer, it
+  // means the iterator will be reused for a new multi scan. If scan_opts
+  // is null, then the previous Prepare() can be discarded.
+  //
+  // The caller guarantees the lifetime of scan_opts until its either cleared
+  // or replaced by another Prepare().
+  // TODO: Update the contract to trim the scan_opts range to only include
+  // scans that potentially intersect the file key range.
+  //
+  // If the sequence of Seeks is interrupted by seeking to some other target
+  // key, then the iterator is free to discard anything done during Prepare.
+  virtual void Prepare(const ScanOptions scan_opts[], size_t num_opts) = 0;
+
+  // Similar to Next(), except it also fills the result and returns whether
+  // the iterator is on a valid key or not
+  virtual bool NextAndGetResult(IterateResult* result) = 0;
+
+  // Prepares the value if its lazily materialized. The implementation can
+  // request that this be called by setting value_prepared to false in
+  // IterateResult. Next() should always implicitly materialize the
+  // value.
+  virtual bool PrepareValue() = 0;
+
+  // Return the current key's value
+  virtual Slice value() const = 0;
+
+  // Return the current position bounds check result - kInbound if the
+  // position is a valid key, kOutOfBound if the key is out of bound (i.e
+  // scan has terminated), or kUnknown if end of file.
+  virtual IterBoundCheck UpperBoundCheckResult() = 0;
+};
+
 class ExternalTableReader {
  public:
   virtual ~ExternalTableReader() {}
@@ -65,8 +107,9 @@ class ExternalTableReader {
   // Return an Iterator that can be used to scan the table file.
   // The read_options can optionally contain the upper bound
   // key (exclusive) of the scan in iterate_upper_bound.
-  virtual Iterator* NewIterator(const ReadOptions& read_options,
-                                const SliceTransform* prefix_extractor) = 0;
+  virtual ExternalTableIterator* NewIterator(
+      const ReadOptions& read_options,
+      const SliceTransform* prefix_extractor) = 0;
 
   // Point lookup the given key and return its value
   virtual Status Get(const ReadOptions& read_options, const Slice& key,

include/rocksdb/iterator.h

@@ -21,6 +21,7 @@
 #include <string>
 
 #include "rocksdb/iterator_base.h"
+#include "rocksdb/options.h"
 #include "rocksdb/wide_columns.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -93,6 +94,18 @@ class Iterator : public IteratorBase {
     assert(false);
     return Slice();
   }
+
+  // RocksDB Internal - DO NOT USE
+  // Prepare the iterator to scan the ranges specified in scan_opts. The
+  // upper bound and other table specific limits may be specified. This will
+  // typically be followed by Seeks to the start keys in the order they're
+  // specified in scan_opts. If the user does a Seek to some other target key,
+  // the iterator should disregard the scan_opts from that point onwards and
+  // behave like a normal iterator. Its the user's responsibility to again
+  // call Prepare().
+  // If Prepare() is called, it overrides the iterate_upper_bound in
+  // ReadOptions
+  virtual void Prepare(const std::vector<ScanOptions>& /*scan_opts*/) {}
 };
 
 // Return an empty iterator (yields nothing).

include/rocksdb/multi_scan.h

@@ -0,0 +1,223 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include "rocksdb/iterator.h"
+#include "rocksdb/options.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+// EXPERIMENTAL
+//
+// An iterator that returns results from multiple scan ranges. The ranges are
+// expected to be in increasing sorted order.
+// The results are returned in nested container objects that can be iterated
+// using an std::input_iterator.
+//
+// MultiScan
+//     |
+//     ---
+//       |
+//  MultiScanIterator  <-- std::input_iterator (returns a Scan object for each
+//         |                                    scan range)
+//         ---
+//           |
+//          Scan
+//            |
+//            ---
+//              |
+//          ScanIterator <-- std::input_iterator (returns the KVs of a single
+//                                                scan range)
+//
+// The application on top of RocksDB
+// would use this as follows -
+//
+//  std::vector<ScanOptions> scans{{.start = Slice("bar")},
+//                              {.start = Slice("foo")}};
+//  std::unique_ptr<MultiScanIterator> iter.reset(
+//                                      db->NewMultiScanIterator());
+//  try {
+//    for (auto scan : *iter) {
+//      for (auto it : scan) {
+//        // Do something with key - it.first
+//        // Do something with value - it.second
+//      }
+//    }
+//  } catch (Status s) {
+//  }
+
+// A container object encapsulating a single scan range. It supports an
+// std::input_iterator for a single pass iteration of the KVs in the range.
+// A Status exception is thrown if there is an error in scanning the range.
+class Scan {
+ public:
+  class ScanIterator;
+
+  Scan(Iterator* db_iter) : db_iter_(db_iter) {}
+
+  ScanIterator begin() { return ScanIterator(db_iter_); }
+
+  std::nullptr_t end() { return nullptr; }
+
+  class ScanIterator {
+   public:
+    using self_type = ScanIterator;
+    using value_type = std::pair<Slice, Slice>;
+    using reference = std::pair<Slice, Slice>&;
+    using pointer = std::pair<Slice, Slice>*;
+    using difference_type = int;
+    using iterator_category = std::input_iterator_tag;
+
+    explicit ScanIterator(Iterator* db_iter) : db_iter_(db_iter) {
+      valid_ = db_iter_->Valid();
+      if (valid_) {
+        result_ = value_type(db_iter_->key(), db_iter_->value());
+      }
+    }
+
+    ScanIterator() : db_iter_(nullptr), valid_(false) {}
+
+    ~ScanIterator() { assert(status_.ok()); }
+
+    ScanIterator& operator++() {
+      if (!valid_) {
+        throw Status::InvalidArgument("Trying to advance invalid iterator");
+      } else {
+        db_iter_->Next();
+        status_ = db_iter_->status();
+        if (!status_.ok()) {
+          throw status_;
+        } else {
+          valid_ = db_iter_->Valid();
+          if (valid_) {
+            result_ = value_type(db_iter_->key(), db_iter_->value());
+          }
+        }
+      }
+      return *this;
+    }
+
+    bool operator==(std::nullptr_t /*other*/) const { return !valid_; }
+
+    bool operator!=(std::nullptr_t /*other*/) const { return valid_; }
+
+    reference operator*() {
+      if (!valid_) {
+        throw Status::InvalidArgument("Trying to deref invalid iterator");
+      }
+      return result_;
+    }
+    reference operator->() {
+      if (!valid_) {
+        throw Status::InvalidArgument("Trying to deref invalid iterator");
+      }
+      return result_;
+    }
+
+   private:
+    Iterator* db_iter_;
+    bool valid_;
+    Status status_;
+    value_type result_;
+  };
+
+ private:
+  Iterator* db_iter_;
+};
+
+// A container object encapsulating the scan ranges for a multi scan.
+// It supports an std::input_iterator for a single pass iteration of the
+// ScanOptions in scan_opts, which can be dereferenced to get the container
+// (Scan) for a single range.
+// A Status exception is thrown if there is an error.
+class MultiScan {
+ public:
+  MultiScan(const std::vector<ScanOptions>& scan_opts,
+            std::unique_ptr<Iterator>&& db_iter)
+      : scan_opts_(scan_opts), db_iter_(std::move(db_iter)) {}
+
+  explicit MultiScan(std::unique_ptr<Iterator>&& db_iter)
+      : db_iter_(std::move(db_iter)) {}
+
+  class MultiScanIterator {
+   public:
+    MultiScanIterator(MultiScanIterator&) = delete;
+    MultiScanIterator operator=(MultiScanIterator&) = delete;
+
+    using self_type = MultiScanIterator;
+    using value_type = Scan;
+    using reference = Scan&;
+    using pointer = Scan*;
+    using difference_type = int;
+    using iterator_category = std::input_iterator_tag;
+
+    MultiScanIterator(const std::vector<ScanOptions>& scan_opts,
+                      Iterator* db_iter)
+        : scan_opts_(scan_opts), idx_(0), db_iter_(db_iter), scan_(db_iter_) {
+      if (scan_opts_.empty()) {
+        throw Status::InvalidArgument("Zero scans in multi-scan");
+      }
+      db_iter_->Seek(*scan_opts_[idx_].range.start);
+      status_ = db_iter_->status();
+      if (!status_.ok()) {
+        throw status_;
+      }
+    }
+
+    MultiScanIterator(const std::vector<ScanOptions>& scan_opts)
+        : scan_opts_(scan_opts),
+          idx_(scan_opts_.size()),
+          db_iter_(nullptr),
+          scan_(nullptr) {}
+
+    ~MultiScanIterator() { assert(status_.ok()); }
+
+    MultiScanIterator& operator++() {
+      if (idx_ >= scan_opts_.size()) {
+        throw Status::InvalidArgument("Index out of range");
+      }
+      idx_++;
+      if (idx_ < scan_opts_.size()) {
+        db_iter_->Seek(*scan_opts_[idx_].range.start);
+        status_ = db_iter_->status();
+        if (!status_.ok()) {
+          throw status_;
+        }
+      }
+      return *this;
+    }
+
+    bool operator==(std::nullptr_t /*other*/) const {
+      return idx_ >= scan_opts_.size();
+    }
+
+    bool operator!=(std::nullptr_t /*other*/) const {
+      return idx_ < scan_opts_.size();
+    }
+
+    reference operator*() { return scan_; }
+    reference operator->() { return scan_; }
+
+   private:
+    const std::vector<ScanOptions>& scan_opts_;
+    size_t idx_;
+    Iterator* db_iter_;
+    Status status_;
+    Scan scan_;
+  };
+
+  MultiScanIterator begin() {
+    return MultiScanIterator(scan_opts_, db_iter_.get());
+  }
+
+  std::nullptr_t end() { return nullptr; }
+
+ private:
+  const std::vector<ScanOptions> scan_opts_;
+  std::unique_ptr<Iterator> db_iter_;
+};
+
+}  // namespace ROCKSDB_NAMESPACE

include/rocksdb/options.h

@@ -1703,6 +1703,50 @@ enum ReadTier {
   kMemtableTier = 0x3     // data in memtable. used for memtable-only iterators.
 };
 
+// A range of keys. In case of user_defined timestamp, if enabled, `start` and
+// `limit` should point to key without timestamp part.
+struct Range {
+  Slice start;
+  Slice limit;
+
+  Range() {}
+  Range(const Slice& s, const Slice& l) : start(s), limit(l) {}
+};
+
+// A key range with optional endpoints. In case of user_defined timestamp, if
+// enabled, `start` and `limit` should point to key without timestamp part.
+struct RangeOpt {
+  // When start.has_value() == false, refers to starting before every key
+  OptSlice start;
+  // When limit.has_value() == false, refers to ending after every key
+  OptSlice limit;
+
+  RangeOpt() {}
+  RangeOpt(const OptSlice& s, const OptSlice& l) : start(s), limit(l) {}
+};
+
+// EXPERIMENTAL
+//
+// Options for a RocksDB scan request. Only forward scans for now.
+// We may add other options such as prefix scan in the future.
+struct ScanOptions {
+  // The scan range. Mandatory for start to be set, limit is optional
+  RangeOpt range;
+
+  // A map of name,value pairs that can be passed by the user to an
+  // external table reader. This is completely opaque to RocksDB and is
+  // ignored by the natively supported table readers like block based and plain
+  // table. This is only useful for Iterator.
+  std::optional<std::unordered_map<std::string, std::string>> property_bag;
+
+  // An unbounded scan with a start key
+  ScanOptions(const Slice& _start) : range(_start, OptSlice()) {}
+
+  // A bounded scan with a start key and upper bound
+  ScanOptions(const Slice& _start, const Slice& _upper_bound)
+      : range(_start, _upper_bound) {}
+};
+
 // Options that control read operations
 struct ReadOptions {
   // *** BEGIN options relevant to point lookups as well as scans ***
@@ -1996,22 +2040,6 @@ struct ReadOptions {
   // EXPERIMENTAL
   Env::IOActivity io_activity = Env::IOActivity::kUnknown;
 
-  // EXPERIMENTAL
-  // An optional weight of values to be returned by a scan. Once the
-  // weight is reached or exceeded the scan is terminated (i.e Next()
-  // invalidates the iterator). In the case of a DB with one of the built-in
-  // table formats, such as BlockBasedTable, the weight is simply the number
-  // of key-value pairs. In the case of an ExternalTableReader, the weight is
-  // passed through to the table reader and the interpretation is upto the
-  // reader implementation.
-  uint64_t weight = 0;
-
-  // A map of name,value pairs that can be passed by the user to an
-  // external table reader. This is completely opaque to RocksDB and is
-  // ignored by the natively supported table readers like block based and plain
-  // table. This is only useful for Iterator.
-  std::optional<std::unordered_map<std::string, std::string>> property_bag;
-
   // *** END options for RocksDB internal use only ***
 
   ReadOptions() {}
@@ -2245,29 +2273,6 @@ struct CompactRangeOptions {
   double blob_garbage_collection_age_cutoff = -1;
 };
 
-// A range of keys. In case of user_defined timestamp, if enabled, `start` and
-// `limit` should point to key without timestamp part.
-struct Range {
-  Slice start;
-  Slice limit;
-
-  Range() {}
-  Range(const Slice& s, const Slice& l) : start(s), limit(l) {}
-};
-
-// A key range with optional endpoints. In case of user_defined timestamp, if
-// enabled, `start` and `limit` should point to key without timestamp part.
-struct RangeOpt {
-  // When start.has_value() == false, refers to starting before every key
-  OptSlice start;
-  // When limit.has_value() == false, refers to ending after every key
-  OptSlice limit;
-
-  RangeOpt() {}
-  RangeOpt(const OptSlice& s, const OptSlice& l) : start(s), limit(l) {}
-  // RangeOpt(const Slice& s, const Slice& l) : start(s), limit(l) {}
-};
-
 // IngestExternalFileOptions is used by IngestExternalFile()
 struct IngestExternalFileOptions {
   // Can be set to true to move the files instead of copying them.

include/rocksdb/utilities/stackable_db.h

@@ -289,6 +289,13 @@ class StackableDB : public DB {
     return db_->NewAttributeGroupIterator(options, column_families);
   }
 
+  using DB::NewMultiScan;
+  std::unique_ptr<MultiScan> NewMultiScan(
+      const ReadOptions& opts, ColumnFamilyHandle* column_family,
+      const std::vector<ScanOptions>& scan_opts) override {
+    return db_->NewMultiScan(opts, column_family, scan_opts);
+  }
+
   const Snapshot* GetSnapshot() override { return db_->GetSnapshot(); }
 
   void ReleaseSnapshot(const Snapshot* snapshot) override {

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 10
 #define ROCKSDB_MINOR 1
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 3
 
 // 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

table/block_based/block_based_table_reader.cc

@@ -914,6 +914,7 @@ Status BlockBasedTable::PrefetchTail(
                      "TailPrefetchStats.",
                      file->file_name().c_str(), tail_prefetch_size);
     }
+    TEST_SYNC_POINT("BlockBasedTable::PrefetchTail::TaiSizeNotRecorded");
   }
   size_t prefetch_off;
   size_t prefetch_len;

table/external_table.cc

@@ -14,16 +14,17 @@ namespace ROCKSDB_NAMESPACE {
 
 namespace {
 
-class ExternalTableIterator : public InternalIterator {
+class ExternalTableIteratorAdapter : public InternalIterator {
  public:
-  explicit ExternalTableIterator(Iterator* iterator)
+  explicit ExternalTableIteratorAdapter(ExternalTableIterator* iterator)
       : iterator_(iterator), valid_(false) {}
 
   // No copying allowed
-  ExternalTableIterator(const ExternalTableIterator&) = delete;
-  ExternalTableIterator& operator=(const ExternalTableIterator&) = delete;
+  ExternalTableIteratorAdapter(const ExternalTableIteratorAdapter&) = delete;
+  ExternalTableIteratorAdapter& operator=(const ExternalTableIteratorAdapter&) =
+      delete;
 
-  ~ExternalTableIterator() override {}
+  ~ExternalTableIteratorAdapter() override {}
 
   bool Valid() const override { return valid_; }
 
@@ -31,7 +32,7 @@ class ExternalTableIterator : public InternalIterator {
     status_ = Status::OK();
     if (iterator_) {
       iterator_->SeekToFirst();
-      UpdateKey();
+      UpdateKey(OptSlice());
     }
   }
 
@@ -39,7 +40,7 @@ class ExternalTableIterator : public InternalIterator {
     status_ = Status::OK();
     if (iterator_) {
       iterator_->SeekToLast();
-      UpdateKey();
+      UpdateKey(OptSlice());
     }
   }
 
@@ -50,7 +51,7 @@ class ExternalTableIterator : public InternalIterator {
       status_ = ParseInternalKey(target, &pkey, /*log_err_key=*/false);
       if (status_.ok()) {
         iterator_->Seek(pkey.user_key);
-        UpdateKey();
+        UpdateKey(OptSlice());
       }
     }
   }
@@ -62,7 +63,7 @@ class ExternalTableIterator : public InternalIterator {
       status_ = ParseInternalKey(target, &pkey, /*log_err_key=*/false);
       if (status_.ok()) {
         iterator_->SeekForPrev(pkey.user_key);
-        UpdateKey();
+        UpdateKey(OptSlice());
       }
     }
   }
@@ -70,14 +71,44 @@ class ExternalTableIterator : public InternalIterator {
   void Next() override {
     if (iterator_) {
       iterator_->Next();
-      UpdateKey();
+      UpdateKey(OptSlice());
     }
   }
 
+  bool NextAndGetResult(IterateResult* result) override {
+    if (iterator_) {
+      valid_ = iterator_->NextAndGetResult(&result_);
+      result->value_prepared = result_.value_prepared;
+      result->bound_check_result = result_.bound_check_result;
+      if (valid_) {
+        UpdateKey(result_.key);
+        result->key = key();
+      }
+    } else {
+      valid_ = false;
+    }
+    return valid_;
+  }
+
+  bool PrepareValue() override {
+    if (iterator_ && !result_.value_prepared) {
+      valid_ = iterator_->PrepareValue();
+      result_.value_prepared = true;
+    }
+    return valid_;
+  }
+
+  IterBoundCheck UpperBoundCheckResult() override {
+    if (iterator_) {
+      result_.bound_check_result = iterator_->UpperBoundCheckResult();
+    }
+    return result_.bound_check_result;
+  }
+
   void Prev() override {
     if (iterator_) {
       iterator_->Prev();
-      UpdateKey();
+      UpdateKey(OptSlice());
     }
   }
 
@@ -97,18 +128,26 @@ class ExternalTableIterator : public InternalIterator {
 
   Status status() const override { return status_; }
 
+  void Prepare(const std::vector<ScanOptions>* scan_opts) override {
+    if (iterator_) {
+      iterator_->Prepare(scan_opts->data(), scan_opts->size());
+    }
+  }
+
  private:
-  std::unique_ptr<Iterator> iterator_;
+  std::unique_ptr<ExternalTableIterator> iterator_;
   InternalKey key_;
   bool valid_;
   Status status_;
+  IterateResult result_;
 
-  void UpdateKey() {
+  void UpdateKey(OptSlice res) {
     if (iterator_) {
       valid_ = iterator_->Valid();
       status_ = iterator_->status();
       if (valid_ && status_.ok()) {
-        key_.Set(iterator_->key(), 0, ValueType::kTypeValue);
+        key_.Set(res.has_value() ? res.value() : iterator_->key(), 0,
+                 ValueType::kTypeValue);
       }
     }
   }
@@ -134,10 +173,10 @@ class ExternalTableReaderAdapter : public TableReader {
       bool /* allow_unprepared_value */ = false) override {
     auto iterator = reader_->NewIterator(read_options, prefix_extractor);
     if (arena == nullptr) {
-      return new ExternalTableIterator(iterator);
+      return new ExternalTableIteratorAdapter(iterator);
     } else {
-      auto* mem = arena->AllocateAligned(sizeof(ExternalTableIterator));
-      return new (mem) ExternalTableIterator(iterator);
+      auto* mem = arena->AllocateAligned(sizeof(ExternalTableIteratorAdapter));
+      return new (mem) ExternalTableIteratorAdapter(iterator);
     }
   }
 

table/internal_iterator.h

@@ -10,6 +10,7 @@
 
 #include "db/dbformat.h"
 #include "file/readahead_file_info.h"
+#include "rocksdb/advanced_iterator.h"
 #include "rocksdb/comparator.h"
 #include "rocksdb/iterator.h"
 #include "rocksdb/status.h"
@@ -19,19 +20,6 @@ namespace ROCKSDB_NAMESPACE {
 
 class PinnedIteratorsManager;
 
-enum class IterBoundCheck : char {
-  kUnknown = 0,
-  kOutOfBound,
-  kInbound,
-};
-
-struct IterateResult {
-  Slice key;
-  IterBoundCheck bound_check_result = IterBoundCheck::kUnknown;
-  // If false, PrepareValue() needs to be called before value().
-  bool value_prepared = true;
-};
-
 template <class TValue>
 class InternalIteratorBase : public Cleanable {
  public:
@@ -212,6 +200,8 @@ class InternalIteratorBase : public Cleanable {
   // used by MergingIterator and LevelIterator for now.
   virtual bool IsDeleteRangeSentinelKey() const { return false; }
 
+  virtual void Prepare(const std::vector<ScanOptions>* /*scan_opts*/) {}
+
  protected:
   void SeekForPrevImpl(const Slice& target, const CompareInterface* cmp) {
     Seek(target);

table/iterator_wrapper.h

@@ -195,6 +195,14 @@ class IteratorWrapperBase {
     return iter_->IsDeleteRangeSentinelKey();
   }
 
+  // scan_opts lifetime is guaranteed until the iterator is destructed, or
+  // Prepare() is called with a new scan_opts
+  void Prepare(const std::vector<ScanOptions>* scan_opts) {
+    if (iter_) {
+      iter_->Prepare(scan_opts);
+    }
+  }
+
  private:
   void Update() {
     valid_ = iter_->Valid();

table/merging_iterator.cc

@@ -482,6 +482,12 @@ class MergingIterator : public InternalIterator {
            current_->IsValuePinned();
   }
 
+  void Prepare(const std::vector<ScanOptions>* scan_opts) override {
+    for (auto& child : children_) {
+      child.iter.Prepare(scan_opts);
+    }
+  }
+
  private:
   // Represents an element in the min/max heap. Each HeapItem corresponds to a
   // point iterator or a range tombstone iterator, differentiated by

table/table_test.cc

@@ -6615,29 +6615,66 @@ class ExternalTableReaderTest : public DBTestBase {
     uint64_t file_size_;
   };
 
-  class DummyExternalTableIterator : public Iterator {
+  class DummyExternalTableIterator : public ExternalTableIterator {
    public:
     explicit DummyExternalTableIterator(
-        const ReadOptions& ro, const std::map<std::string, std::string>& kv_map)
-        : weight_(ro.weight), kv_map_(kv_map), valid_(false) {}
+        const ReadOptions& /*ro*/,
+        const std::map<std::string, std::string>& kv_map)
+        : scan_options_(nullptr),
+          num_opts_(0),
+          scan_idx_(0),
+          kv_map_(kv_map),
+          valid_(false) {}
 
     bool Valid() const override { return valid_; }
 
     void SeekToFirst() override {
-      iter_ = kv_map_.begin();
-      valid_ = iter_ != kv_map_.end();
-      status_ = Status::OK();
+      if (scan_options_) {
+        status_ = Status::InvalidArgument();
+      } else {
+        iter_ = kv_map_.begin();
+        valid_ = iter_ != kv_map_.end();
+        status_ = Status::OK();
+      }
     }
 
     void SeekToLast() override {
-      valid_ = false;
-      status_ = Status::NotSupported();
+      if (scan_options_) {
+        status_ = Status::InvalidArgument();
+      } else {
+        if (!kv_map_.empty()) {
+          iter_ = kv_map_.begin();
+          for (uint64_t i = 0; i < kv_map_.size() - 1; ++i) {
+            iter_++;
+          }
+          valid_ = true;
+        } else {
+          valid_ = false;
+        }
+        status_ = Status::OK();
+      }
     }
 
     void Seek(const Slice& target) override {
-      iter_ = kv_map_.find(target.ToString());
-      valid_ = iter_ != kv_map_.end();
-      status_ = Status::OK();
+      if (status_.ok()) {
+        iter_ = kv_map_.find(target.ToString());
+        valid_ = iter_ != kv_map_.end();
+        eof_ = iter_ == kv_map_.end();
+      }
+      if (scan_options_) {
+        if (scan_idx_ >= num_opts_ ||
+            target != scan_options_[scan_idx_].range.start.value().ToString()) {
+          status_ = Status::InvalidArgument();
+        } else {
+          if (valid_ && scan_options_[scan_idx_].range.limit.has_value() &&
+              iter_->first.compare(
+                  scan_options_[scan_idx_].range.limit.value().ToString()) >=
+                  0) {
+            valid_ = false;
+          }
+          scan_idx_++;
+        }
+      }
     }
 
     void SeekForPrev(const Slice& /*target*/) override {
@@ -6647,9 +6684,37 @@ class ExternalTableReaderTest : public DBTestBase {
 
     void Next() override {
       iter_++;
-      weight_--;
-      valid_ = iter_ != kv_map_.end() && weight_ > 0;
-      // status_ is still ok. valid_ indicates end of scan
+      valid_ = iter_ != kv_map_.end();
+      eof_ = iter_ == kv_map_.end();
+      if (valid_ && scan_options_ &&
+          scan_options_[scan_idx_ - 1].range.limit.has_value() &&
+          iter_->first.compare(
+              scan_options_[scan_idx_ - 1].range.limit.value().ToString()) >=
+              0) {
+        valid_ = false;
+      }
+      // status_ is still ok. !valid_ indicates end of scan
+    }
+
+    bool NextAndGetResult(IterateResult* result) override {
+      Next();
+      if (valid_) {
+        result->key = key();
+        result->bound_check_result = IterBoundCheck::kInbound;
+        result->value_prepared = true;
+      } else {
+        result->key = Slice();
+        result->bound_check_result =
+            eof_ ? IterBoundCheck::kUnknown : IterBoundCheck::kOutOfBound;
+        result->value_prepared = false;
+      }
+      return valid_;
+    }
+
+    bool PrepareValue() override { return valid_ ? true : false; }
+
+    IterBoundCheck UpperBoundCheckResult() override {
+      return eof_ ? IterBoundCheck::kUnknown : IterBoundCheck::kOutOfBound;
     }
 
     void Prev() override {
@@ -6672,10 +6737,18 @@ class ExternalTableReaderTest : public DBTestBase {
       return Slice(iter_->second);
     }
 
+    void Prepare(const ScanOptions scan_opts[], size_t num_opts) override {
+      scan_options_ = scan_opts;
+      num_opts_ = num_opts;
+    }
+
    private:
-    uint64_t weight_;
+    const ScanOptions* scan_options_;
+    size_t num_opts_;
+    size_t scan_idx_;
     std::map<std::string, std::string> kv_map_;
     bool valid_ = false;
+    bool eof_ = false;
     Status status_ = Status::OK();
     std::map<std::string, std::string>::iterator iter_;
   };
@@ -6688,8 +6761,9 @@ class ExternalTableReaderTest : public DBTestBase {
       EXPECT_OK(s);
     }
 
-    Iterator* NewIterator(const ReadOptions& read_options,
-                          const SliceTransform* /*prefix_extractor*/) override {
+    ExternalTableIterator* NewIterator(
+        const ReadOptions& read_options,
+        const SliceTransform* /*prefix_extractor*/) override {
       return new DummyExternalTableIterator(read_options, kv_map_);
     }
 
@@ -6808,8 +6882,7 @@ TEST_F(ExternalTableReaderTest, BasicTest) {
       {}, file_path, ExternalTableOptions(prefix_extractor, nullptr), &reader));
 
   ReadOptions ro;
-  ro.weight = 1;
-  std::unique_ptr<Iterator> iter(reader->NewIterator(ro, nullptr));
+  std::unique_ptr<ExternalTableIterator> iter(reader->NewIterator(ro, nullptr));
   ASSERT_NE(iter, nullptr);
   iter->Seek("foo");
   ASSERT_TRUE(iter->Valid() && iter->status().ok());
@@ -6854,7 +6927,6 @@ TEST_F(ExternalTableReaderTest, SstReaderTest) {
   ASSERT_OK(reader->Open(ingest_file));
 
   ReadOptions ro;
-  ro.weight = 1;
   std::unique_ptr<Iterator> iter(reader->NewIterator(ro));
   ASSERT_NE(iter, nullptr);
   iter->Seek("foo");
@@ -6865,6 +6937,172 @@ TEST_F(ExternalTableReaderTest, SstReaderTest) {
   ASSERT_TRUE(iter->status().ok());
 }
 
+TEST_F(ExternalTableReaderTest, DBIterTest) {
+  Options options = GetDefaultOptions();
+  std::string dbname = test::PerThreadDBPath("external_table_reader_test");
+  std::string ingest_file = dbname + "test.immutable";
+  dbname += "_db";
+  // This test doesn't work with some custom Envs, like EncryptedEnv
+  options.env = Env::Default();
+  ASSERT_OK(DestroyDB(dbname, options));
+
+  std::shared_ptr<ExternalTableFactory> factory(
+      new DummyExternalTableFactory());
+  options.table_factory = NewExternalTableFactory(factory);
+
+  // Create a file
+  std::unique_ptr<SstFileWriter> writer;
+  writer.reset(new SstFileWriter(EnvOptions(), options));
+  ASSERT_OK(writer->Open(ingest_file));
+  ASSERT_OK(writer->Put("foo", "bar"));
+  ASSERT_OK(writer->Put("foo2", "bar2"));
+  ASSERT_OK(writer->Finish());
+  writer.reset();
+
+  std::unique_ptr<DB> db;
+  options.create_if_missing = true;
+  Status s = DB::Open(options, dbname, &db);
+  ASSERT_OK(s);
+  ASSERT_TRUE(db != nullptr);
+  ColumnFamilyHandle* cfh = nullptr;
+  ASSERT_OK(db->CreateColumnFamily(options, "new_cf", &cfh));
+
+  IngestExternalFileOptions ifo;
+  ifo.allow_db_generated_files = true;
+  ifo.fill_cache = false;
+  s = db->IngestExternalFile(cfh, {ingest_file}, ifo);
+  ASSERT_OK(s);
+
+  std::unique_ptr<Iterator> iter(db->NewIterator({}, cfh));
+  ASSERT_NE(iter, nullptr);
+  iter->Seek("foo");
+  ASSERT_TRUE(iter->Valid() && iter->status().ok());
+  ASSERT_EQ(iter->value(), "bar");
+  iter->Next();
+  ASSERT_TRUE(iter->Valid() && iter->status().ok());
+  ASSERT_EQ(iter->key(), "foo2");
+  ASSERT_EQ(iter->value(), "bar2");
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_OK(iter->status());
+  iter.reset();
+
+  ASSERT_OK(db->DestroyColumnFamilyHandle(cfh));
+  ASSERT_OK(db->Close());
+}
+
+TEST_F(ExternalTableReaderTest, DBMultiScanTest) {
+  Options options = GetDefaultOptions();
+  std::string dbname = test::PerThreadDBPath("external_table_reader_test");
+  std::string ingest_file = dbname + "test.immutable";
+  dbname += "_db";
+  // This test doesn't work with some custom Envs, like EncryptedEnv
+  options.env = Env::Default();
+  ASSERT_OK(DestroyDB(dbname, options));
+
+  std::shared_ptr<ExternalTableFactory> factory(
+      new DummyExternalTableFactory());
+  options.table_factory = NewExternalTableFactory(factory);
+
+  // Create a file
+  std::unique_ptr<SstFileWriter> writer;
+  writer.reset(new SstFileWriter(EnvOptions(), options));
+  ASSERT_OK(writer->Open(ingest_file));
+  for (int i = 0; i < 100; ++i) {
+    std::stringstream ss;
+    ss << std::setw(2) << std::setfill('0') << i;
+    ASSERT_OK(writer->Put("k" + ss.str(), "val" + ss.str()));
+  }
+  ASSERT_OK(writer->Finish());
+  writer.reset();
+
+  std::unique_ptr<DB> db;
+  options.create_if_missing = true;
+  Status s = DB::Open(options, dbname, &db);
+  ASSERT_OK(s);
+  ASSERT_TRUE(db != nullptr);
+  ColumnFamilyHandle* cfh = nullptr;
+  ASSERT_OK(db->CreateColumnFamily(options, "new_cf", &cfh));
+
+  IngestExternalFileOptions ifo;
+  ifo.allow_db_generated_files = true;
+  ifo.fill_cache = false;
+  s = db->IngestExternalFile(cfh, {ingest_file}, ifo);
+  ASSERT_OK(s);
+
+  std::vector<std::string> key_ranges({"k03", "k10", "k25", "k50"});
+  ReadOptions ro;
+  std::vector<ScanOptions> scan_options(
+      {ScanOptions(key_ranges[0], key_ranges[1]),
+       ScanOptions(key_ranges[2], key_ranges[3])});
+  std::unique_ptr<MultiScan> iter = db->NewMultiScan(ro, cfh, scan_options);
+  try {
+    int idx = 0;
+    int count = 0;
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_GE(it.first.ToString().compare(key_ranges[idx]), 0);
+        ASSERT_LT(it.first.ToString().compare(key_ranges[idx + 1]), 0);
+        count++;
+      }
+      idx += 2;
+    }
+    ASSERT_EQ(count, 32);
+  } catch (Status status) {
+    std::cerr << "Iterator returned status " << status.ToString();
+    abort();
+  }
+  iter.reset();
+
+  // Test the overlapping scan case
+  key_ranges[1] = "k30";
+  scan_options[0] = ScanOptions(key_ranges[0], key_ranges[1]);
+  iter = db->NewMultiScan(ro, cfh, scan_options);
+  try {
+    int idx = 0;
+    int count = 0;
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_GE(it.first.ToString().compare(key_ranges[idx]), 0);
+        ASSERT_LT(it.first.ToString().compare(key_ranges[idx + 1]), 0);
+        count++;
+      }
+      idx += 2;
+    }
+    ASSERT_EQ(count, 52);
+  } catch (Status status) {
+    std::cerr << "Iterator returned status " << status.ToString();
+    abort();
+  }
+  iter.reset();
+
+  // Test the no limit scan case
+  scan_options[0] = ScanOptions(key_ranges[0]);
+  scan_options[1] = ScanOptions(key_ranges[2]);
+  iter = db->NewMultiScan(ro, cfh, scan_options);
+  try {
+    int idx = 0;
+    int count = 0;
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_GE(it.first.ToString().compare(key_ranges[idx]), 0);
+        if (it.first.ToString().compare(key_ranges[idx + 1]) == 0) {
+          break;
+        }
+        count++;
+      }
+      idx += 2;
+    }
+    ASSERT_EQ(count, 52);
+  } catch (Status status) {
+    std::cerr << "Iterator returned status " << status.ToString();
+    abort();
+  }
+  iter.reset();
+
+  ASSERT_OK(db->DestroyColumnFamilyHandle(cfh));
+  ASSERT_OK(db->Close());
+}
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

unreleased_history/behavior_changes/ldb_comp.md

@@ -1 +0,0 @@
-* `ldb` now returns an error if the specified `--compression_type` is not supported in the build.

unreleased_history/behavior_changes/persisted-tier-multiget.md

@@ -1 +0,0 @@
-* MultiGet with snapshot and ReadOptions::read_tier = kPersistedTier will now read a consistent view across CFs (instead of potentially reading some CF before and some CF after a flush).

unreleased_history/behavior_changes/read_only_create_cf.md

@@ -1 +0,0 @@
-* CreateColumnFamily() is no longer allowed on a read-only DB (OpenForReadOnly())

unreleased_history/bug_fixes/stats_fix_for_tiered_storage.md

@@ -1 +0,0 @@
-Fixed stats for Tiered Storage with preclude_last_level feature

unreleased_history/new_features/calculate_sst_write_lifetime_hint_set.md

@@ -1 +0,0 @@
-Added a new `DBOptions.calculate_sst_write_lifetime_hint_set` setting that allows to customize which compaction styles SST write lifetime hint calculation is allowed on. Today RocksDB supports only two modes `kCompactionStyleLevel` and `kCompactionStyleUniversal`.

unreleased_history/new_features/l0_file.md

@@ -1 +0,0 @@
-Add a new field `num_l0_files` in `CompactionJobInfo` about the number of L0 files in the CF right before and after the compaction

unreleased_history/new_features/per_key_placement_remote_compaction.md

@@ -1 +0,0 @@
-Added per-key-placement feature in Remote Compaction

unreleased_history/new_features/table_properties_for_levels.md

@@ -1 +0,0 @@
-Implemented API DB::GetPropertiesOfTablesByLevel that retrieves table properties for files in each LSM tree level

unreleased_history/public_api_changes/optslice.md

@@ -1,2 +0,0 @@
-* `GetAllKeyVersions()` now interprets empty slices literally, as valid keys, and uses new `OptSlice` type default value for extreme upper and lower range limits.
-* `DeleteFilesInRanges()` now takes `RangeOpt` which is based on `OptSlice`. The overload taking `RangePtr` is deprecated.

unreleased_history/public_api_changes/read_options_property_bag.md

@@ -1 +0,0 @@
-Add an unordered map of name/value pairs, ReadOptions::property_bag, to pass opaque options through to an external table when creating an Iterator.

unreleased_history/public_api_changes/remote_compaction_aborted_status.md

@@ -1 +0,0 @@
-Introduced CompactionServiceJobStatus::kAborted to allow handling aborted scenario in Schedule(), Wait() or OnInstallation() APIs in Remote Compactions.

unreleased_history/public_api_changes/unsupport_fv1.md

@@ -1 +0,0 @@
-* format\_version < 2 in BlockBasedTableOptions is no longer supported for writing new files. Support for reading such files is deprecated and might be removed in the future. `CompressedSecondaryCacheOptions::compress_format_version == 1` is also deprecated.