Update HISTORY and version for 10.7.6

Summary:
* Address feedback from https://github.com/facebook/rocksdb/issues/14040
* Add additional test for MultiScan
* Fix a bug when del range and data are in same file for multi-scan
* Rewrite the cases need to be handled in SeekMultiScan

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

Test Plan: Unit test

Reviewed By: cbi42, anand1976

Differential Revision: D84851788

Pulled By: xingbowang

fbshipit-source-id: 0f69632733afb99685f6341badbf239681010c38

.github/actions/setup-folly/action.yml

@@ -3,5 +3,7 @@ runs:
   using: composite
   steps:
   - name: Checkout folly sources
-    run: make checkout_folly
+    run: |
+      make checkout_folly
+      echo "GETDEPS_USE_WGET=1" >> "$GITHUB_ENV"
     shell: bash

HISTORY.md

@@ -1,6 +1,57 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 10.7.6 (10/23/2025)
+* Fix a bug when del range and data are in same file for multi-scan
+
+## 10.7.5 (10/20/2025)
+### Bug Fixes
+* Fix a bug in Page unpinning in MultiScan
+
+## 10.7.4 (10/14/2025)
+### Public API Changes
+* The MultiScan API contract is updated. After a multi scan range got prepared with Prepare API call, the following seeks must seek the start of each prepared scan range in order. In addition, when limit is set, upper bound must be set to the same value of limit before each seek
+
+## 10.7.3 (10/06/2025)
+### Bug Fixes
+* Fix a few bugs in MultiScan
+
+## 10.7.2 (09/30/2025)
+### Bug Fixes
+* Fix incorrect MultiScan seek error status due to bugs in handling range limit falling between adjacent SST files key range.
+
+### Performance Improvements
+* Fixed a performance regression in LZ4 compression that started in version 10.6.0
+
+## 10.7.0 (09/24/2025)
+### New Features
+* Add the fail_if_no_udi_on_open flag in BlockBasedTableOption to control whether a missing user defined index block in a SST is a hard error or not.
+* A new flag memtable_veirfy_per_key_checksum_on_seek is added to AdvancedColumnFamilyOptions. When it is enabled, it will validate key checksum along the binary search path on skiplist based memtable during seek operation.
+* Introduce option MultiScanArgs::use_async_io to enable asynchronous I/O during MultiScan, instead of waiting for I/O to be done in Prepare().
+* Add new option `MultiScanArgs::max_prefetch_size` that limits the memory usage of per file pinning of prefetched blocks.
+* Improved `sst_dump` by allowing standalone file and directory arguments without `--file=`. Also added new options and better output for `sst_dump --command=recompress`. See `sst_dump --help`
+
+### Public API Changes
+* HyperClockCache with no `estimated_entry_charge` is now production-ready and is the preferred block cache implementation vs. LRUCache. Please consider updating your code to minimize the risk of hitting performance bottlenecks or anomalies from LRUCache. See cache.h for more detail.
+* RocksDB now requires a C++20 compatible compiler (GCC >= 11, Clang >= 10, Visual Studio >= 2019), including for any code using RocksDB headers.
+* MultiScanArgs used to have a default constructor with default parameter of BytewiseComparator. Now it always requires Comparator in its constructor.
+
+### Behavior Changes
+* The default provided block cache implementation is now HyperClockCache instead of LRUCache, when `block_cache` is nullptr (default) and `no_block_cache==false` (default). We recommend explicitly creating a HyperClockCache block cache based on memory budget and sharing it across all column families and even DB instances. This change could expose previously hidden memory or resource leaks.
+* Allow UDIs with a non BytewiseComparator
+
+### Bug Fixes
+* Reported numbers for compaction and flush CPU usage now include time spent by parallel compression worker threads. This now means compaction/flush CPU usage could exceed the wall clock time.
+* Fix a race condition in FIFO size-based compaction where concurrent threads could select the same non-L0 file, causing assertion failures in debug builds or "Cannot delete table file from LSM tree" errors in release builds.
+* Fix a bug in RocksDB MultiScan with UDI when one of the scan ranges is determined to be empty by the UDI, which causes incorrect results.
+
+### Performance Improvements
+* Add a new table property "rocksdb.key.smallest.seqno" which records the smallest sequence number of all keys in file. It makes ingesting DB generated files faster by
+avoiding scanning the whole file to find the smallest sequence number.
+* Add a new experimental PerKeyPointLockManager to improve efficiency under high lock contention. PointLockManager was not efficient when there is high write contention on same key, as it uses a single conditional variable per lock stripe. PerKeyPointLockManager uses per thread conditional variable supporting fifo order. Although this is an experimental feature. By default, it is disabled. A new boolean flag TransactionDBOptions::use_per_key_point_lock_mgr is added to optionally enable it. Search the flag in code for more info.
+Together, a new configuration TransactionOptions::deadlock_timeout_us is added, which allows the transaction to wait for a short period before perform deadlock detection. When the workload has low lock contention, the deadlock_timeout_us can be configured to be slightly higher than average transaction execution time, so that transaction would likely be able to take the lock before deadlock detection is performed when it is waiting for a lock. This allows transaction to reduce CPU cost on performing deadlock detection, which could be expensive in CPU time. When the workload has high lock contention, the deadlock_timeout_us can be configured to 0, so that transaction would perform deadlock detection immediately. By default the value is 0 to keep the behavior same as before.
+* Majorly improved CPU efficiency and scalability of parallel compression (`CompressionOptions::parallel_threads` > 1), though this efficiency improvement makes parallel compression currently incompatible with UserDefinedIndex and with old setting of `decouple_partitioned_filters=false`. Parallel compression is now considered a production-ready feature. Maximum performance is available with `-DROCKSDB_USE_STD_SEMAPHORES` at compile time, but this is not currently recommended because of reported bugs in implementations of `std::counting_semaphore`/`binary_semaphore`.
+
 ## 10.6.0 (08/22/2025)
 ### New Features
 * Introduce column family option `cf_allow_ingest_behind`. This option aims to replace `DBOptions::allow_ingest_behind` to enable ingest behind at the per-CF level. `DBOptions::allow_ingest_behind` is deprecated.

cache/clock_cache.cc

@@ -2062,9 +2062,10 @@ AutoHyperClockTable::~AutoHyperClockTable() {
   }
   // This check can be extra expensive for a cache that is just created,
   // maybe used for a small number of entries, as in a unit test, and then
-  // destroyed. Only do this in rare modes.
+  // destroyed. Only do this in rare modes. REVISED: Don't scan the whole mmap,
+  // just a reasonable frontier past what we expect to have written.
 #ifdef MUST_FREE_HEAP_ALLOCATIONS
-  for (size_t i = used_end; i < array_.Count(); i++) {
+  for (size_t i = used_end; i < array_.Count() && i < used_end + 64U; i++) {
     assert(array_[i].head_next_with_shift.LoadRelaxed() == 0);
     assert(array_[i].chain_next_with_shift.LoadRelaxed() == 0);
     assert(array_[i].meta.LoadRelaxed() == 0);

db/compaction/tiered_compaction_test.cc

@@ -1764,9 +1764,10 @@ TEST_P(PrecludeLastLevelTest, SmallPrecludeTime) {
   options.env = mock_env_.get();
   options.level0_file_num_compaction_trigger = kNumTrigger;
   options.num_levels = kNumLevels;
-  // This existing test selected to also check the kIce case, which should not
-  // be interesting enough to exercise across all the test cases
-  options.last_level_temperature = Temperature::kIce;
+  // This existing test selected to also check the case of various temperatures
+  // for last_level_temperature, which should not be interesting enough to
+  // exercise across many/all test cases
+  options.last_level_temperature = RandomKnownTemperature();
   DestroyAndReopen(options);
 
   Random rnd(301);
@@ -1794,8 +1795,9 @@ TEST_P(PrecludeLastLevelTest, SmallPrecludeTime) {
   auto seqs = tp_mapping.TEST_GetInternalMapping();
   ASSERT_FALSE(seqs.empty());
   ASSERT_GE(GetSstSizeHelper(Temperature::kUnknown), 1);
-  ASSERT_EQ(GetSstSizeHelper(Temperature::kCold), 0);
-  ASSERT_EQ(GetSstSizeHelper(Temperature::kIce), 0);
+  for (auto t : kKnownTemperatures) {
+    ASSERT_EQ(GetSstSizeHelper(t), 0);
+  }
 
   // Wait more than preclude_last_level time, then make sure all the data is
   // compacted to the last level even there's no write (no seqno -> time
@@ -1804,9 +1806,14 @@ TEST_P(PrecludeLastLevelTest, SmallPrecludeTime) {
 
   ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
   ASSERT_EQ("0,0,0,0,0,0,1", FilesPerLevel());
-  ASSERT_EQ(GetSstSizeHelper(Temperature::kUnknown), 0);
-  ASSERT_EQ(GetSstSizeHelper(Temperature::kCold), 0);
-  ASSERT_GE(GetSstSizeHelper(Temperature::kIce), 1);
+
+  for (auto t : kKnownTemperatures) {
+    if (t == options.last_level_temperature) {
+      ASSERT_GT(GetSstSizeHelper(t), 0);
+    } else {
+      ASSERT_EQ(GetSstSizeHelper(t), 0);
+    }
+  }
 
   Close();
 }

db/db_compaction_test.cc

@@ -9915,6 +9915,20 @@ static void VerifyTemperatureFileReadStats(const Statistics& st,
     EXPECT_EQ(iostats->file_io_stats_by_temperature.warm_file_read_count, 0);
   }
 
+  if (temps.Contains(Temperature::kCool)) {
+    EXPECT_GE(st.getTickerCount(COOL_FILE_READ_BYTES), min_bytes);
+    EXPECT_GE(st.getTickerCount(COOL_FILE_READ_COUNT), min_count);
+    EXPECT_GE(iostats->file_io_stats_by_temperature.cool_file_bytes_read,
+              min_bytes);
+    EXPECT_GE(iostats->file_io_stats_by_temperature.cool_file_read_count,
+              min_count);
+  } else {
+    EXPECT_EQ(st.getTickerCount(COOL_FILE_READ_BYTES), 0);
+    EXPECT_EQ(st.getTickerCount(COOL_FILE_READ_COUNT), 0);
+    EXPECT_EQ(iostats->file_io_stats_by_temperature.cool_file_bytes_read, 0);
+    EXPECT_EQ(iostats->file_io_stats_by_temperature.cool_file_read_count, 0);
+  }
+
   if (temps.Contains(Temperature::kCold)) {
     EXPECT_GE(st.getTickerCount(COLD_FILE_READ_BYTES), min_bytes);
     EXPECT_GE(st.getTickerCount(COLD_FILE_READ_COUNT), min_count);
@@ -9945,7 +9959,7 @@ static void VerifyTemperatureFileReadStats(const Statistics& st,
 }
 
 TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
-  // Test multi-tier aging: Hot -> Warm -> Cold -> Ice
+  // Test multi-tier aging: Hot -> Warm -> Cool -> Cold -> Ice
   Options options = CurrentOptions();
   options.compaction_style = kCompactionStyleFIFO;
   options.num_levels = 1;
@@ -9961,8 +9975,9 @@ TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
   // Multi-tier aging: files age through multiple temperatures
   fifo_options.file_temperature_age_thresholds = {
       {Temperature::kWarm, 500},   // Hot -> Warm after 500s
-      {Temperature::kCold, 1000},  // Warm -> Cold after 1000s
-      {Temperature::kIce, 1500}    // Cold -> Ice after 1500s
+      {Temperature::kCool, 1000},  // Warm -> Cool
+      {Temperature::kCold, 1500},  // Cool -> Cold
+      {Temperature::kIce, 2000}    // Cold -> Ice
   };
   fifo_options.max_table_files_size = 100000000;
   fifo_options.allow_trivial_copy_when_change_temperature = true;
@@ -9973,8 +9988,8 @@ TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
   env_->SetMockSleep();
 
   // Track all temperature file creations
-  int total_hot = 0, total_warm = 0, total_cold = 0, total_ice = 0,
-      total_unknown = 0;
+  int total_hot = 0, total_warm = 0, total_cool = 0, total_cold = 0,
+      total_ice = 0, total_unknown = 0;
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
       "NewWritableFile::FileOptions.temperature", [&](void* arg) {
         Temperature temperature = *(static_cast<Temperature*>(arg));
@@ -9985,6 +10000,9 @@ TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
           case Temperature::kWarm:
             total_warm++;
             break;
+          case Temperature::kCool:
+            total_cool++;
+            break;
           case Temperature::kCold:
             total_cold++;
             break;
@@ -10016,8 +10034,11 @@ TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
 
   VerifyTemperatureFileReadStats(*options.statistics, Temperature::kHot);
 
+  // Land well into each time interval
+  env_->MockSleepForSeconds(100);
+
   // Age initial files to warm
-  env_->MockSleepForSeconds(600);
+  env_->MockSleepForSeconds(500);
   ASSERT_OK(Put(Key(1), Random::GetTLSInstance()->RandomBinaryString(101)));
   ASSERT_OK(Flush());
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
@@ -10031,12 +10052,26 @@ TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
   // Verify Warm file statistics
   VerifyTemperatureFileReadStats(*options.statistics, Temperature::kWarm);
 
-  // Age initial files to cold
-  env_->MockSleepForSeconds(600);
+  // Age initial files to cool
+  env_->MockSleepForSeconds(500);
   ASSERT_OK(Put(Key(2), Random::GetTLSInstance()->RandomBinaryString(102)));
   ASSERT_OK(Flush());
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
 
+  // Test reading from Cool temperature file (the aged file)
+  ASSERT_OK(options.statistics->Reset());
+  get_iostats_context()->Reset();
+
+  ASSERT_EQ(100U, Get(Key(0)).size());
+
+  VerifyTemperatureFileReadStats(*options.statistics, Temperature::kCool);
+
+  // Age initial files to cold
+  env_->MockSleepForSeconds(500);
+  ASSERT_OK(Put(Key(3), Random::GetTLSInstance()->RandomBinaryString(103)));
+  ASSERT_OK(Flush());
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+
   // Test reading from Cold temperature file (the aged file)
   ASSERT_OK(options.statistics->Reset());
   get_iostats_context()->Reset();
@@ -10046,8 +10081,8 @@ TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
   VerifyTemperatureFileReadStats(*options.statistics, Temperature::kCold);
 
   // Age initial files to ice
-  env_->MockSleepForSeconds(600);
-  ASSERT_OK(Put(Key(3), Random::GetTLSInstance()->RandomBinaryString(103)));
+  env_->MockSleepForSeconds(500);
+  ASSERT_OK(Put(Key(4), Random::GetTLSInstance()->RandomBinaryString(104)));
   ASSERT_OK(Flush());
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
 
@@ -10072,12 +10107,14 @@ TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
   // Verify current files temperatures
   EXPECT_EQ(temp_counts[Temperature::kHot], 1);
   EXPECT_EQ(temp_counts[Temperature::kWarm], 1);
+  EXPECT_EQ(temp_counts[Temperature::kCool], 1);
   EXPECT_EQ(temp_counts[Temperature::kCold], 1);
   EXPECT_EQ(temp_counts[Temperature::kIce], 3);
 
   // Verify historical (and current) file temperatures
-  EXPECT_EQ(total_hot, 6);
-  EXPECT_EQ(total_warm, 5);
+  EXPECT_EQ(total_hot, 7);
+  EXPECT_EQ(total_warm, 6);
+  EXPECT_EQ(total_cool, 5);
   EXPECT_EQ(total_cold, 4);
   EXPECT_EQ(total_ice, 3);
 
@@ -10087,7 +10124,7 @@ TEST_F(DBCompactionTest, FIFOMultiTierTemperatureAging) {
   get_iostats_context()->Reset();
 
   // Read from all files to verify cumulative statistics
-  for (int i = 0; i < 4; i++) {
+  for (int i = 0; i < 5; i++) {
     ASSERT_EQ(static_cast<unsigned>(100 + i), Get(Key(i)).size());
   }
 

db/db_iter.cc

@@ -1565,11 +1565,115 @@ void DBIter::SetSavedKeyToSeekForPrevTarget(const Slice& target) {
   }
 }
 
+Status DBIter::ValidateScanOptions(const MultiScanArgs& multiscan_opts) const {
+  if (multiscan_opts.empty()) {
+    return Status::InvalidArgument("Empty MultiScanArgs");
+  }
+
+  const std::vector<ScanOptions>& scan_opts = multiscan_opts.GetScanRanges();
+  const bool has_limit = scan_opts.front().range.limit.has_value();
+  if (!has_limit && scan_opts.size() > 1) {
+    return Status::InvalidArgument("Scan has no upper bound");
+  }
+
+  for (size_t i = 0; i < scan_opts.size(); ++i) {
+    const auto& scan_range = scan_opts[i].range;
+    if (!scan_range.start.has_value()) {
+      return Status::InvalidArgument("Scan has no start key at index " +
+                                     std::to_string(i));
+    }
+
+    if (scan_range.limit.has_value()) {
+      if (user_comparator_.CompareWithoutTimestamp(
+              scan_range.start.value(), /*a_has_ts=*/false,
+              scan_range.limit.value(), /*b_has_ts=*/false) >= 0) {
+        return Status::InvalidArgument(
+            "Scan start key is large or equal than limit at index " +
+            std::to_string(i));
+      }
+    }
+
+    if (i > 0) {
+      if (!scan_range.limit.has_value()) {
+        // multiple scan without limit scan ranges
+        return Status::InvalidArgument("Scan has no upper bound at index " +
+                                       std::to_string(i));
+      }
+
+      const auto& last_end_key = scan_opts[i - 1].range.limit.value();
+      if (user_comparator_.CompareWithoutTimestamp(
+              scan_range.start.value(), /*a_has_ts=*/false, last_end_key,
+              /*b_has_ts=*/false) < 0) {
+        return Status::InvalidArgument("Overlapping ranges at index " +
+                                       std::to_string(i));
+      }
+    }
+  }
+  return Status::OK();
+}
+
+void DBIter::Prepare(const MultiScanArgs& scan_opts) {
+  status_ = ValidateScanOptions(scan_opts);
+  if (!status_.ok()) {
+    return;
+  }
+  std::optional<MultiScanArgs> new_scan_opts;
+  new_scan_opts.emplace(scan_opts);
+  scan_opts_.swap(new_scan_opts);
+  scan_index_ = 0;
+  if (!scan_opts.empty()) {
+    iter_.Prepare(&scan_opts_.value());
+  } else {
+    iter_.Prepare(nullptr);
+  }
+}
+
 void DBIter::Seek(const Slice& target) {
   PERF_COUNTER_ADD(iter_seek_count, 1);
   PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, clock_);
   StopWatch sw(clock_, statistics_, DB_SEEK);
 
+  if (scan_opts_.has_value()) {
+    // Validate the seek target is as expected in the previously prepared range
+    auto const& scan_ranges = scan_opts_.value().GetScanRanges();
+    if (scan_index_ >= scan_ranges.size()) {
+      status_ = Status::InvalidArgument(
+          "Seek called after exhausting all of the scan ranges");
+      valid_ = false;
+      return;
+    }
+
+    // Validate start key of next prepare range matches the seek target
+    auto const& range = scan_ranges[scan_index_];
+    auto const& start = range.range.start;
+    assert(start.has_value());
+    if (user_comparator_.CompareWithoutTimestamp(target, *start) != 0) {
+      status_ = Status::InvalidArgument(
+          "Seek target does not match the start of the next prepared range at "
+          "index " +
+          std::to_string(scan_index_));
+      valid_ = false;
+      return;
+    }
+
+    // validate the upper bound is set to the same value of limit, if limit
+    // exists
+    auto const& limit = range.range.limit;
+    if (limit.has_value()) {
+      if (iterate_upper_bound_ == nullptr ||
+          user_comparator_.CompareWithoutTimestamp(
+              limit.value(), *iterate_upper_bound_) != 0) {
+        status_ = Status::InvalidArgument(
+            "Upper bound is not set to the same limit value of the next "
+            "prepared range at index " +
+            std::to_string(scan_index_));
+        valid_ = false;
+        return;
+      }
+    }
+    scan_index_++;
+  }
+
   if (cfh_ != nullptr) {
     // TODO: What do we do if this returns an error?
     Slice lower_bound, upper_bound;

db/db_iter.h

@@ -240,16 +240,8 @@ class DBIter final : public Iterator {
 
   bool PrepareValue() override;
 
-  void Prepare(const MultiScanArgs& scan_opts) override {
-    std::optional<MultiScanArgs> 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);
-    }
-  }
+  void Prepare(const MultiScanArgs& scan_opts) override;
+  Status ValidateScanOptions(const MultiScanArgs& multiscan_opts) const;
 
  private:
   DBIter(Env* _env, const ReadOptions& read_options,
@@ -506,6 +498,7 @@ class DBIter final : public Iterator {
   const size_t timestamp_size_;
   std::string saved_timestamp_;
   std::optional<MultiScanArgs> scan_opts_;
+  size_t scan_index_{0};
   ReadOnlyMemTable* const active_mem_;
   SequenceNumber memtable_seqno_lb_;
   uint32_t memtable_op_scan_flush_trigger_;

db/db_iterator_test.cc

@@ -4149,6 +4149,7 @@ class DBMultiScanIteratorTest : public DBTestBase,
       : DBTestBase("db_multi_scan_iterator_test", /*env_do_fsync=*/true) {}
 };
 
+// Param 0: ReadOptions::fill_cache
 INSTANTIATE_TEST_CASE_P(DBMultiScanIteratorTest, DBMultiScanIteratorTest,
                         ::testing::Bool());
 
@@ -4192,66 +4193,6 @@ TEST_P(DBMultiScanIteratorTest, BasicTest) {
     abort();
   }
   iter.reset();
-
-  // Test the overlapping scan case
-  key_ranges[1] = "k30";
-  scan_options = MultiScanArgs(BytewiseComparator());
-  scan_options.insert(key_ranges[0], key_ranges[1]);
-  scan_options.insert(key_ranges[2], key_ranges[3]);
-
-  iter = dbfull()->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 (MultiScanException& ex) {
-    // Make sure exception contains the status
-    ASSERT_NOK(ex.status());
-    std::cerr << "Iterator returned status " << ex.what();
-    abort();
-  } catch (std::logic_error& ex) {
-    std::cerr << "Iterator returned logic error " << ex.what();
-    abort();
-  }
-  iter.reset();
-
-  // Test the no limit scan case
-  scan_options = MultiScanArgs(BytewiseComparator());
-  scan_options.insert(key_ranges[0]);
-  scan_options.insert(key_ranges[2]);
-  iter = dbfull()->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 (MultiScanException& ex) {
-    // Make sure exception contains the status
-    ASSERT_NOK(ex.status());
-    std::cerr << "Iterator returned status " << ex.what();
-    abort();
-  } catch (std::logic_error& ex) {
-    std::cerr << "Iterator returned logic error " << ex.what();
-    abort();
-  }
-  iter.reset();
 }
 
 TEST_P(DBMultiScanIteratorTest, MixedBoundsTest) {
@@ -4366,15 +4307,297 @@ TEST_P(DBMultiScanIteratorTest, RangeAcrossFiles) {
   ColumnFamilyHandle* cfh = dbfull()->DefaultColumnFamily();
   std::unique_ptr<MultiScan> iter =
       dbfull()->NewMultiScan(ro, cfh, scan_options);
-  int i = 10;
-  for (auto range : *iter) {
-    for (auto it : range) {
-      ASSERT_EQ(it.first.ToString(), Key(i));
-      ++i;
+  try {
+    int i = 10;
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_EQ(it.first.ToString(), Key(i));
+        ++i;
+      }
     }
+    ASSERT_EQ(i, 90);
+  } catch (MultiScanException& ex) {
+    // Make sure exception contains the status
+    ASSERT_NOK(ex.status());
+    std::cerr << "Iterator returned status " << ex.what();
+    abort();
+  } catch (std::logic_error& ex) {
+    std::cerr << "Iterator returned logic error " << ex.what();
+    abort();
   }
-  ASSERT_EQ(i, 90);
+  iter.reset();
 }
+
+TEST_P(DBMultiScanIteratorTest, FailureTest) {
+  auto options = CurrentOptions();
+  options.compression = kNoCompression;
+  DestroyAndReopen(options);
+
+  Random rnd(301);
+  // Create a file
+  for (int i = 0; i < 100; ++i) {
+    std::stringstream ss;
+    ss << std::setw(2) << std::setfill('0') << i;
+    ASSERT_OK(Put("k" + ss.str(), rnd.RandomString(1024)));
+  }
+  ASSERT_OK(Flush());
+
+  std::vector<std::string> key_ranges({"k04", "k06", "k12", "k14"});
+  ReadOptions ro;
+  Slice ub;
+  ro.iterate_upper_bound = &ub;
+  ro.fill_cache = GetParam();
+  MultiScanArgs scan_options(BytewiseComparator());
+  scan_options.insert(key_ranges[0], key_ranges[1]);
+  scan_options.insert(key_ranges[2], key_ranges[3]);
+  scan_options.max_prefetch_size = 4500;
+  ColumnFamilyHandle* cfh = dbfull()->DefaultColumnFamily();
+  std::unique_ptr<Iterator> iter(dbfull()->NewIterator(ro, cfh));
+  ASSERT_NE(iter, nullptr);
+  iter->Prepare(scan_options);
+  int count = 0;
+  ub = key_ranges[1];
+  iter->Seek(key_ranges[0]);
+  while (iter->status().ok() && iter->Valid()) {
+    ASSERT_GE(iter->key().compare(key_ranges[0]), 0);
+    ASSERT_LT(iter->key().compare(key_ranges[1]), 0);
+    count++;
+    iter->Next();
+  }
+  ASSERT_OK(iter->status()) << iter->status().ToString();
+  ASSERT_EQ(count, 2);
+
+  // Second seek should hit the max_prefetch_size limit
+  ub = key_ranges[3];
+  iter->Seek(key_ranges[2]);
+  ASSERT_NOK(iter->status());
+  iter.reset();
+
+  // Test the case of unexpected Seek key
+  iter.reset(dbfull()->NewIterator(ro, cfh));
+  ASSERT_NE(iter, nullptr);
+  scan_options.max_prefetch_size = 0;
+  iter->Prepare(scan_options);
+  ub = key_ranges[3];
+  iter->Seek(key_ranges[2]);
+  ASSERT_NOK(iter->status());
+  iter.reset();
+}
+
+TEST_P(DBMultiScanIteratorTest, RangeBetweenFiles) {
+  auto options = CurrentOptions();
+  options.target_file_size_base = 100 << 10;  // 20KB
+  options.compaction_style = kCompactionStyleUniversal;
+  options.num_levels = 50;
+  options.compression = kNoCompression;
+  DestroyAndReopen(options);
+
+  auto rnd = Random::GetTLSInstance();
+  // Write ~200KB data
+  for (int i = 0; i < 100; ++i) {
+    ASSERT_OK(Put(Key(i), rnd->RandomString(2 << 10)));
+  }
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(db_->CompactRange({}, nullptr, nullptr));
+  ASSERT_EQ(2, NumTableFilesAtLevel(49));
+
+  // Test with a scan range that overlaps an entire file, with upper bound
+  // between 2 files
+  std::vector<LiveFileMetaData> file_meta;
+  dbfull()->GetLiveFilesMetaData(&file_meta);
+  ASSERT_EQ(file_meta.size(), 2);
+  std::vector<std::string> key_ranges(4);
+  key_ranges[0] = file_meta[0].smallestkey;
+  key_ranges[1] = file_meta[0].largestkey + "0";
+  key_ranges[2] = file_meta[1].smallestkey + "0";
+  key_ranges[3] = file_meta[1].largestkey;
+  ReadOptions ro;
+  ro.fill_cache = GetParam();
+  MultiScanArgs scan_options(BytewiseComparator());
+  scan_options.insert(key_ranges[0], key_ranges[1]);
+  scan_options.insert(key_ranges[2], key_ranges[3]);
+  ColumnFamilyHandle* cfh = dbfull()->DefaultColumnFamily();
+  std::unique_ptr<MultiScan> iter =
+      dbfull()->NewMultiScan(ro, cfh, scan_options);
+  try {
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_GE(it.first.ToString(), key_ranges[0]);
+      }
+    }
+  } catch (MultiScanException& ex) {
+    // Make sure exception contains the status
+    ASSERT_NOK(ex.status());
+    std::cerr << "Iterator returned status " << ex.what();
+    abort();
+  } catch (std::logic_error& ex) {
+    std::cerr << "Iterator returned logic error " << ex.what();
+    abort();
+  }
+  iter.reset();
+
+  // Test multiscan with a range entirely between adjacent files
+  key_ranges[0] = file_meta[0].largestkey + "0";
+  key_ranges[1] = file_meta[0].largestkey + "1";
+  key_ranges[2] = file_meta[1].smallestkey + "0";
+  key_ranges[3] = file_meta[1].largestkey;
+  (*scan_options).clear();
+  scan_options.insert(key_ranges[0], key_ranges[1]);
+  scan_options.insert(key_ranges[2], key_ranges[3]);
+  iter = dbfull()->NewMultiScan(ro, cfh, scan_options);
+  try {
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_GE(it.first.ToString(), key_ranges[0]);
+      }
+    }
+  } catch (MultiScanException& ex) {
+    // Make sure exception contains the status
+    ASSERT_NOK(ex.status());
+    std::cerr << "Iterator returned status " << ex.what();
+    abort();
+  } catch (std::logic_error& ex) {
+    std::cerr << "Iterator returned logic error " << ex.what();
+    abort();
+  }
+  iter.reset();
+}
+
+// This test case tests multiscan in the presence of fragmented range
+// tombstones in the LSM.
+TEST_P(DBMultiScanIteratorTest, FragmentedRangeTombstones) {
+  auto options = CurrentOptions();
+  // Compaction may create files 2x the target_file_size_base,
+  // so set this to 50KB so we atleast end up with 2 files of
+  // 100KB
+  options.target_file_size_base = 50 << 10;  // 50KB
+  options.compaction_style = kCompactionStyleUniversal;
+  options.num_levels = 50;
+  options.compression = kNoCompression;
+  DestroyAndReopen(options);
+
+  // Setup the LSM as follows -
+  // 1. Ingest a file with 100 keys
+  // 2. Ingest a file with one overlapping key
+  // 3. Do a Put and flush a file to L0 with one overlapping key
+  // 4. Ingest a standalone delete range file that covers the full key space
+  //    and a file with the same 100 keys with new values. This will ingest
+  //    into L0 due to the presence of an existing file in L0
+  // The final LSM will have an SST in Lmax with 100 keys, and 2 SST files
+  // in Lmax-1 with half the keys each and completely overlapping delete ranges
+  std::unordered_map<std::string, std::string> kvs;
+  auto rnd = Random::GetTLSInstance();
+  auto create_ingestion_data_file_and_update_key_value =
+      [&](const std::string& filename, int start_key, int end_key) {
+        std::unique_ptr<SstFileWriter> writer;
+        writer.reset(new SstFileWriter(EnvOptions(), options));
+        ASSERT_OK(writer->Open(filename));
+        for (int i = start_key; i < end_key; ++i) {
+          auto kiter = kvs.find(Key(i));
+          if (kiter != kvs.end()) {
+            kvs.erase(kiter);
+          }
+          auto res =
+              kvs.emplace(std::make_pair(Key(i), rnd->RandomString(2 << 10)));
+          ASSERT_OK(writer->Put(res.first->first, res.first->second));
+        }
+        ASSERT_OK(writer->Finish());
+        writer.reset();
+      };
+
+  CreateColumnFamilies({"new_cf"}, options);
+  std::string ingest_file = dbname_ + "test.sst";
+  // Write ~200KB data
+  create_ingestion_data_file_and_update_key_value(ingest_file + "_0", 0, 100);
+  create_ingestion_data_file_and_update_key_value(ingest_file + "_1", 50, 51);
+  ColumnFamilyHandle* cfh = handles_[0];
+  IngestExternalFileOptions ifo;
+  Status s = dbfull()->IngestExternalFile(
+      cfh, {ingest_file + "_0", ingest_file + "_1"}, ifo);
+  ASSERT_OK(s);
+
+  ASSERT_OK(Put(0, Key(50), rnd->RandomString(2 << 10)));
+  ASSERT_OK(Flush());
+
+  {
+    std::unique_ptr<SstFileWriter> writer;
+    writer.reset(new SstFileWriter(EnvOptions(), options));
+    ASSERT_OK(writer->Open(ingest_file + "_2"));
+    ASSERT_OK(writer->DeleteRange("a", "z"));
+    ASSERT_OK(writer->Finish());
+    writer.reset();
+  }
+  create_ingestion_data_file_and_update_key_value(ingest_file + "_3", 0, 100);
+  s = dbfull()->IngestExternalFile(
+      cfh, {ingest_file + "_2", ingest_file + "_3"}, ifo);
+  ASSERT_OK(s);
+
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+
+  // The first scan range overlaps the DB key range, while the second extends
+  // beyond but overlaps the delete range
+  std::vector<std::string> key_ranges({"key000085", "key000090", "l", "n"});
+  ReadOptions ro;
+  ro.fill_cache = GetParam();
+  MultiScanArgs scan_options(BytewiseComparator());
+  scan_options.insert(key_ranges[0], key_ranges[1]);
+  scan_options.insert(key_ranges[2], key_ranges[3]);
+  std::unique_ptr<MultiScan> iter =
+      dbfull()->NewMultiScan(ro, cfh, scan_options);
+  try {
+    int i = 0;
+    int count = 0;
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_GE(it.first.ToString(), key_ranges[i]);
+        ASSERT_LT(it.first.ToString(), key_ranges[i + 1]);
+        auto kiter = kvs.find(it.first.ToString());
+        ASSERT_NE(kiter, kvs.end());
+        ASSERT_EQ(kiter->second, it.second.ToString());
+        count++;
+      }
+      i += 2;
+    }
+    ASSERT_EQ(i, 4);
+    ASSERT_EQ(count, 5);
+  } catch (MultiScanException& ex) {
+    ASSERT_OK(ex.status());
+  }
+  iter.reset();
+
+  // The second scan range start overlaps the delete range in the first file
+  // in Lmax-1, while the end overlaps the keys in the second file
+  (*scan_options).clear();
+  key_ranges[0] = "key000010";
+  key_ranges[1] = "key000020";
+  key_ranges[2] = "key0000500";
+  key_ranges[3] = "key000060";
+  scan_options.insert(key_ranges[0], key_ranges[1]);
+  scan_options.insert(key_ranges[2], key_ranges[3]);
+  iter = dbfull()->NewMultiScan(ro, cfh, scan_options);
+  try {
+    int i = 0;
+    int count = 0;
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_GE(it.first.ToString(), key_ranges[i]);
+        ASSERT_LT(it.first.ToString(), key_ranges[i + 1]);
+        auto kiter = kvs.find(it.first.ToString());
+        ASSERT_NE(kiter, kvs.end());
+        ASSERT_EQ(kiter->second, it.second.ToString());
+        count++;
+      }
+      i += 2;
+    }
+    ASSERT_EQ(i, 4);
+    ASSERT_EQ(count, 19);
+  } catch (MultiScanException& ex) {
+    ASSERT_OK(ex.status());
+  }
+  iter.reset();
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

db/db_test2.cc

@@ -6063,16 +6063,9 @@ TEST_F(DBTest2, VariousFileTemperatures) {
   };
 
   // We don't have enough non-unknown temps to confidently distinguish that
-  // a specific setting caused a specific outcome, in a single run. This is a
-  // reasonable work-around without blowing up test time. Only returns
-  // non-unknown temperatures.
-  auto RandomTemp = [] {
-    static std::vector<Temperature> temps = {
-        Temperature::kHot, Temperature::kWarm, Temperature::kCold,
-        Temperature::kIce};
-    return temps[Random::GetTLSInstance()->Uniform(
-        static_cast<int>(temps.size()))];
-  };
+  // a specific setting caused a specific outcome, in a single run. Using
+  // RandomKnownTemperature() is a reasonable work-around without blowing up
+  // test time.
 
   auto test_fs = std::make_shared<MyTestFS>(env_->GetFileSystem());
   std::unique_ptr<Env> env(new CompositeEnvWrapper(env_, test_fs));
@@ -6088,22 +6081,22 @@ TEST_F(DBTest2, VariousFileTemperatures) {
       options.env = env.get();
       test_fs->Reset();
       if (use_optimize) {
-        test_fs->optimize_manifest_temperature = RandomTemp();
+        test_fs->optimize_manifest_temperature = RandomKnownTemperature();
         test_fs->expected_manifest_temperature =
             test_fs->optimize_manifest_temperature;
-        test_fs->optimize_wal_temperature = RandomTemp();
+        test_fs->optimize_wal_temperature = RandomKnownTemperature();
         test_fs->expected_wal_temperature = test_fs->optimize_wal_temperature;
       }
       if (use_temp_options) {
-        options.metadata_write_temperature = RandomTemp();
+        options.metadata_write_temperature = RandomKnownTemperature();
         test_fs->expected_manifest_temperature =
             options.metadata_write_temperature;
         test_fs->expected_other_metadata_temperature =
             options.metadata_write_temperature;
-        options.wal_write_temperature = RandomTemp();
+        options.wal_write_temperature = RandomKnownTemperature();
         test_fs->expected_wal_temperature = options.wal_write_temperature;
-        options.last_level_temperature = RandomTemp();
-        options.default_write_temperature = RandomTemp();
+        options.last_level_temperature = RandomKnownTemperature();
+        options.default_write_temperature = RandomKnownTemperature();
       }
 
       DestroyAndReopen(options);

db/db_test_util.cc

@@ -1872,4 +1872,13 @@ template class TargetCacheChargeTrackingCache<
     CacheEntryRole::kBlockBasedTableReader>;
 template class TargetCacheChargeTrackingCache<CacheEntryRole::kFileMetadata>;
 
+const std::vector<Temperature> kKnownTemperatures = {
+    Temperature::kHot, Temperature::kWarm, Temperature::kCool,
+    Temperature::kCold, Temperature::kIce};
+
+Temperature RandomKnownTemperature() {
+  return kKnownTemperatures[Random::GetTLSInstance()->Uniform(
+      static_cast<int>(kKnownTemperatures.size()))];
+}
+
 }  // namespace ROCKSDB_NAMESPACE

db/db_test_util.h

@@ -1467,4 +1467,8 @@ class DBTestBase : public testing::Test {
 // unique ids.
 void VerifySstUniqueIds(const TablePropertiesCollection& props);
 
+// Excludes kUnknown
+extern const std::vector<Temperature> kKnownTemperatures;
+Temperature RandomKnownTemperature();
+
 }  // namespace ROCKSDB_NAMESPACE

db/external_sst_file_basic_test.cc

@@ -1311,7 +1311,7 @@ TEST_F(ExternalSSTFileBasicTest, SyncFailure) {
     });
     if (i == 0) {
       SyncPoint::GetInstance()->SetCallBack(
-          "ExternalSstFileIngestionJob::CheckSyncReturnCode", [&](void* s) {
+          "ExternalSstFileIngestionJob::Prepare:Reopen", [&](void* s) {
             Status* status = static_cast<Status*>(s);
             if (status->IsNotSupported()) {
               no_sync = true;
@@ -1372,11 +1372,11 @@ TEST_F(ExternalSSTFileBasicTest, ReopenNotSupported) {
   options.create_if_missing = true;
   options.env = env_;
 
-  SyncPoint::GetInstance()->SetCallBack("FileSystem::SyncFile:Open",
-                                        [&](void* arg) {
-                                          Status* s = static_cast<Status*>(arg);
-                                          *s = Status::NotSupported();
-                                        });
+  SyncPoint::GetInstance()->SetCallBack(
+      "ExternalSstFileIngestionJob::Prepare:Reopen", [&](void* arg) {
+        Status* s = static_cast<Status*>(arg);
+        *s = Status::NotSupported();
+      });
   SyncPoint::GetInstance()->EnableProcessing();
 
   DestroyAndReopen(options);

db/external_sst_file_ingestion_job.cc

@@ -163,26 +163,35 @@ Status ExternalSstFileIngestionJob::Prepare(
         // It is unsafe to assume application had sync the file and file
         // directory before ingest the file. For integrity of RocksDB we need
         // to sync the file.
-        TEST_SYNC_POINT("ExternalSstFileIngestionJob::BeforeSyncIngestedFile");
-        auto s = fs_->SyncFile(path_inside_db, env_options_, IOOptions(),
-                               db_options_.use_fsync, nullptr);
-        TEST_SYNC_POINT("ExternalSstFileIngestionJob::AfterSyncIngestedFile");
-        TEST_SYNC_POINT_CALLBACK(
-            "ExternalSstFileIngestionJob::CheckSyncReturnCode", &s);
-        if (!s.ok()) {
-          if (s.IsNotSupported()) {
-            // Some file systems (especially remote/distributed) don't support
-            // SyncFile API. Ignore the NotSupported error in that case.
-            ROCKS_LOG_WARN(db_options_.info_log,
-                           "After link the file, SyncFile API is not supported "
-                           "for file %s: %s",
-                           path_inside_db.c_str(), status.ToString().c_str());
-          } else {
-            // for other errors, propagate the error
-            status = s;
-            ROCKS_LOG_WARN(db_options_.info_log,
-                           "Failed to sync ingested file %s: %s",
-                           path_inside_db.c_str(), status.ToString().c_str());
+
+        // TODO(xingbo), We should in general be moving away from production
+        // uses of ReuseWritableFile (except explicitly for WAL recycling),
+        // ReopenWritableFile, and NewRandomRWFile. We should create a
+        // FileSystem::SyncFile/FsyncFile API that by default does the
+        // re-open+sync+close combo but can (a) be reused easily, and (b) be
+        // overridden to do that more cleanly, e.g. in EncryptedEnv.
+        // https://github.com/facebook/rocksdb/issues/13741
+        std::unique_ptr<FSWritableFile> file_to_sync;
+        Status s = fs_->ReopenWritableFile(path_inside_db, env_options_,
+                                           &file_to_sync, nullptr);
+        TEST_SYNC_POINT_CALLBACK("ExternalSstFileIngestionJob::Prepare:Reopen",
+                                 &s);
+        // Some file systems (especially remote/distributed) don't support
+        // reopening a file for writing and don't require reopening and
+        // syncing the file. Ignore the NotSupported error in that case.
+        if (!s.IsNotSupported()) {
+          status = s;
+          if (status.ok()) {
+            TEST_SYNC_POINT(
+                "ExternalSstFileIngestionJob::BeforeSyncIngestedFile");
+            status = SyncIngestedFile(file_to_sync.get());
+            TEST_SYNC_POINT(
+                "ExternalSstFileIngestionJob::AfterSyncIngestedFile");
+            if (!status.ok()) {
+              ROCKS_LOG_WARN(db_options_.info_log,
+                             "Failed to sync ingested file %s: %s",
+                             path_inside_db.c_str(), status.ToString().c_str());
+            }
           }
         }
       } else if (status.IsNotSupported() &&

db/multi_scan.cc

@@ -40,6 +40,11 @@ MultiScan::MultiScan(const ReadOptions& read_options,
 }
 
 MultiScanIterator& MultiScanIterator::operator++() {
+  status_ = db_iter_->status();
+  if (!status_.ok()) {
+    throw MultiScanException(status_);
+  }
+
   if (idx_ >= scan_opts_.size()) {
     throw std::logic_error("Index out of range");
   }

db/version_set.cc

@@ -1181,15 +1181,25 @@ class LevelIterator final : public InternalIterator {
       // 3. [  S  ] ...... [  E  ]
       for (auto i = fstart; i <= fend; i++) {
         if (i < flevel_->num_files) {
-          auto args = GetMultiScanArgForFile(i);
+          // FindFile only compares against the largest_key, so we need this
+          // additional check to ensure the scan range overlaps the file
+          if (icomparator_.InternalKeyComparator::Compare(
+                  iend.Encode(), flevel_->files[i].smallest_key) < 0) {
+            continue;
+          }
+          auto const metadata = flevel_->files[i].file_metadata;
+          if (metadata->FileIsStandAloneRangeTombstone()) {
+            // Skip stand alone range deletion files.
+            continue;
+          }
+          auto& args = GetMultiScanArgForFile(i);
           args.insert(start.value(), end.value(), opt.property_bag);
         }
       }
     }
-    // Propagate io colaescing threshold
+    // Propagate multiscan configs
     for (auto& file_to_arg : *file_to_scan_opts_) {
-      file_to_arg.second.io_coalesce_threshold = so->io_coalesce_threshold;
-      file_to_arg.second.use_async_io = so->use_async_io;
+      file_to_arg.second.CopyConfigFrom(*so);
     }
   }
 
@@ -1265,6 +1275,10 @@ class LevelIterator final : public InternalIterator {
     }
   }
 
+#ifndef NDEBUG
+  bool OverlapRange(const ScanOptions& opts);
+#endif
+
   TableCache* table_cache_;
   const ReadOptions& read_options_;
   const FileOptions& file_options_;
@@ -1362,6 +1376,14 @@ void LevelIterator::Seek(const Slice& target) {
   }
 
   if (file_iter_.iter() != nullptr) {
+    if (scan_opts_) {
+      // At this point, we only know that the seek target is < largest_key
+      // in the file. We need to check whether there is actual overlap.
+      const FdWithKeyRange& cur_file = flevel_->files[file_index_];
+      if (KeyReachedUpperBound(cur_file.smallest_key)) {
+        return;
+      }
+    }
     file_iter_.Seek(target);
     // Status::TryAgain indicates asynchronous request for retrieval of data
     // blocks has been submitted. So it should return at this point and Seek
@@ -1639,6 +1661,19 @@ void LevelIterator::SkipEmptyFileBackward() {
   }
 }
 
+#ifndef NDEBUG
+bool LevelIterator::OverlapRange(const ScanOptions& opts) {
+  return (user_comparator_.CompareWithoutTimestamp(
+              opts.range.start.value(), /*a_has_ts=*/false,
+              ExtractUserKey(flevel_->files[file_index_].largest_key),
+              /*b_has_ts=*/true) <= 0 &&
+          user_comparator_.CompareWithoutTimestamp(
+              opts.range.limit.value(), /*a_has_ts=*/false,
+              ExtractUserKey(flevel_->files[file_index_].smallest_key),
+              /*b_has_ts=*/true) > 0);
+}
+#endif
+
 void LevelIterator::SetFileIterator(InternalIterator* iter) {
   if (pinned_iters_mgr_ && iter) {
     iter->SetPinnedItersMgr(pinned_iters_mgr_);
@@ -1648,9 +1683,9 @@ void LevelIterator::SetFileIterator(InternalIterator* iter) {
   if (iter && scan_opts_) {
     if (FileHasMultiScanArg(file_index_)) {
       const MultiScanArgs& new_opts = GetMultiScanArgForFile(file_index_);
+      assert(OverlapRange(*new_opts.GetScanRanges().begin()) &&
+             OverlapRange(*new_opts.GetScanRanges().rbegin()));
       file_iter_.Prepare(&new_opts);
-    } else {
-      file_iter_.Prepare(scan_opts_);
     }
   }
 

db_stress_tool/db_stress_test_base.cc

@@ -428,8 +428,9 @@ bool StressTest::BuildOptionsTable() {
     options_tbl.emplace(
         "file_temperature_age_thresholds",
         std::vector<std::string>{
-            "{{temperature=kWarm;age=10}:{temperature=kCold;age=50}:{"
-            "temperature=kIce;age=250}}",
+            "{{temperature=kWarm;age=10}:{temperature=kCool;age=30}:{"
+            "temperature=kCold;age=100}:{"
+            "temperature=kIce;age=300}}",
             "{{temperature=kWarm;age=30}:{temperature=kCold;age=300}}",
             "{{temperature=kCold;age=100}}", "{}"});
     options_tbl.emplace(
@@ -1693,7 +1694,7 @@ Status StressTest::TestMultiScan(ThreadState* thread,
   std::vector<std::string> start_key_strs;
   std::vector<std::string> end_key_strs;
   // TODO support reverse BytewiseComparator in the stress test
-  MultiScanArgs scan_opts(BytewiseComparator());
+  MultiScanArgs scan_opts(options_.comparator);
   scan_opts.use_async_io = FLAGS_multiscan_use_async_io;
   start_key_strs.reserve(num_scans);
   end_key_strs.reserve(num_scans);

env/composite_env_wrapper.h

@@ -142,12 +142,6 @@ class CompositeEnv : public Env {
     return file_system_->LinkFile(s, t, io_opts, &dbg);
   }
 
-  Status SyncFile(const std::string& fname, const EnvOptions& env_options,
-                  bool use_fsync) override {
-    return file_system_->SyncFile(fname, env_options, IOOptions(), use_fsync,
-                                  nullptr);
-  }
-
   Status NumFileLinks(const std::string& fname, uint64_t* count) override {
     IOOptions io_opts;
     IODebugContext dbg;

env/env.cc

@@ -529,13 +529,6 @@ class LegacyFileSystemWrapper : public FileSystem {
     return status_to_io_status(target_->LinkFile(s, t));
   }
 
-  IOStatus SyncFile(const std::string& fname, const FileOptions& file_options,
-                    const IOOptions& /*io_options*/, bool use_fsync,
-                    IODebugContext* /*dbg*/) override {
-    return status_to_io_status(
-        target_->SyncFile(fname, file_options, use_fsync));
-  }
-
   IOStatus NumFileLinks(const std::string& fname, const IOOptions& /*options*/,
                         uint64_t* count, IODebugContext* /*dbg*/) override {
     return status_to_io_status(target_->NumFileLinks(fname, count));
@@ -878,43 +871,6 @@ std::string Env::GenerateUniqueId() {
   return result;
 }
 
-// This API Env::SyncFile is used for testing for 2 reasons:
-//
-// 1. The default implementation of SyncFile API is essentially a wrapper of
-// other FileSystem APIs. FaultInjectionTestEnv uses this default
-// implementation to call other FileSystem APIs defined at
-// FaultInjectionTestEnv class to inject failurses. See
-// FaultInjectionTestEnv::SyncFile for more details
-//
-// 2. Some of old tests are using LegacyFileSystemWrapper.
-// LegacyFileSystemWrapper forwards the API call to EnvWrapper, which forwards
-// to CompositeEnv, and then forwards to the actual FileSystem implemention.
-// Without this API in Env, LegacyFileSystemWrapper will not be able to
-// forward the API call to EnvWrapper, causing the default FileSystem API to
-// be called.
-//
-// Due to the above reason, adding a new API in FileSystem, would very likely
-// require the same API to be added to Env.
-//
-// TODO xingbo. Getting rid of FileSystem functions from Env.
-// We need to simplify the relationship between Env and FileSystem. At least
-// for internal test, we should stop using Env and switch to FileSystem, if
-// possible. Related github issue #9274
-Status Env::SyncFile(const std::string& fname, const EnvOptions& env_options,
-                     bool use_fsync) {
-  std::unique_ptr<WritableFile> file_to_sync;
-  auto status = ReopenWritableFile(fname, &file_to_sync, env_options);
-  TEST_SYNC_POINT_CALLBACK("FileSystem::SyncFile:Open", &status);
-  if (status.ok()) {
-    if (use_fsync) {
-      status = file_to_sync->Fsync();
-    } else {
-      status = file_to_sync->Sync();
-    }
-  }
-  return status;
-}
-
 SequentialFile::~SequentialFile() = default;
 
 RandomAccessFile::~RandomAccessFile() = default;

env/env_encryption.cc

@@ -664,8 +664,6 @@ class EncryptedFileSystemImpl : public EncryptedFileSystem {
                               const FileOptions& options,
                               std::unique_ptr<FSWritableFile>* result,
                               IODebugContext* dbg) override {
-    // TODO xingbo Add unit test for the new implementation of
-    // EncryptedFileSysmteImpl::ReopenWritableFile.
     result->reset();
     if (options.use_mmap_reads || options.use_mmap_writes) {
       return IOStatus::InvalidArgument();
@@ -816,15 +814,6 @@ class EncryptedFileSystemImpl : public EncryptedFileSystem {
     return status;
   }
 
-  IOStatus SyncFile(const std::string& fname, const FileOptions& file_options,
-                    const IOOptions& io_options, bool use_fsync,
-                    IODebugContext* dbg) override {
-    // Use the underlying file system to sync the file, as we don't need to
-    // read/write the file.
-    return FileSystemWrapper::SyncFile(fname, file_options, io_options,
-                                       use_fsync, dbg);
-  }
-
  private:
   std::shared_ptr<EncryptionProvider> provider_;
 };

env/file_system.cc

@@ -107,23 +107,6 @@ IOStatus FileSystem::ReuseWritableFile(const std::string& fname,
   return NewWritableFile(fname, opts, result, dbg);
 }
 
-IOStatus FileSystem::SyncFile(const std::string& fname,
-                              const FileOptions& file_options,
-                              const IOOptions& io_options, bool use_fsync,
-                              IODebugContext* dbg) {
-  std::unique_ptr<FSWritableFile> file_to_sync;
-  auto status = ReopenWritableFile(fname, file_options, &file_to_sync, dbg);
-  TEST_SYNC_POINT_CALLBACK("FileSystem::SyncFile:Open", &status);
-  if (status.ok()) {
-    if (use_fsync) {
-      status = file_to_sync->Fsync(io_options, dbg);
-    } else {
-      status = file_to_sync->Sync(io_options, dbg);
-    }
-  }
-  return status;
-}
-
 IOStatus FileSystem::NewLogger(const std::string& fname,
                                const IOOptions& io_opts,
                                std::shared_ptr<Logger>* result,

env/mock_env.cc

@@ -957,14 +957,6 @@ IOStatus MockFileSystem::LinkFile(const std::string& src,
   return IOStatus::OK();
 }
 
-IOStatus MockFileSystem::SyncFile(const std::string& /*fname*/,
-                                  const FileOptions& /*file_options*/,
-                                  const IOOptions& /*io_options*/,
-                                  bool /*use_fsync*/, IODebugContext* /*dbg*/) {
-  // Noop
-  return IOStatus::OK();
-}
-
 IOStatus MockFileSystem::NewLogger(const std::string& fname,
                                    const IOOptions& io_opts,
                                    std::shared_ptr<Logger>* result,

env/mock_env.h

@@ -86,10 +86,6 @@ class MockFileSystem : public FileSystem {
   IOStatus LinkFile(const std::string& /*src*/, const std::string& /*target*/,
                     const IOOptions& /*options*/,
                     IODebugContext* /*dbg*/) override;
-  IOStatus SyncFile(const std::string& /*fname*/,
-                    const FileOptions& /*file_options*/,
-                    const IOOptions& /*io_options*/, bool /*use_fsync*/,
-                    IODebugContext* /*dbg*/) override;
   IOStatus LockFile(const std::string& fname, const IOOptions& options,
                     FileLock** lock, IODebugContext* dbg) override;
   IOStatus UnlockFile(FileLock* lock, const IOOptions& options,

file/random_access_file_reader.cc

@@ -80,6 +80,12 @@ inline void RecordIOStats(Statistics* stats, Temperature file_temperature,
         RecordTick(stats, WARM_FILE_READ_BYTES, size);
         RecordTick(stats, WARM_FILE_READ_COUNT, 1);
         break;
+      case Temperature::kCool:
+        IOSTATS_ADD(file_io_stats_by_temperature.cool_file_bytes_read, size);
+        IOSTATS_ADD(file_io_stats_by_temperature.cool_file_read_count, 1);
+        RecordTick(stats, COOL_FILE_READ_BYTES, size);
+        RecordTick(stats, COOL_FILE_READ_COUNT, 1);
+        break;
       case Temperature::kCold:
         IOSTATS_ADD(file_io_stats_by_temperature.cold_file_bytes_read, size);
         IOSTATS_ADD(file_io_stats_by_temperature.cold_file_read_count, 1);

include/rocksdb/env.h

@@ -385,13 +385,6 @@ class Env : public Customizable {
     return Status::NotSupported("LinkFile is not supported for this Env");
   }
 
-  // Sync the file content to file system.
-  // This API is only used for testing.
-  // See FileSystem::SyncFile comment for details
-  virtual Status SyncFile(const std::string& /*fname*/,
-                          const EnvOptions& /*env_options*/,
-                          bool /*use_fsync*/);
-
   virtual Status NumFileLinks(const std::string& /*fname*/,
                               uint64_t* /*count*/) {
     return Status::NotSupported(
@@ -1684,11 +1677,6 @@ class EnvWrapper : public Env {
     return target_.env->LinkFile(s, t);
   }
 
-  Status SyncFile(const std::string& fname, const EnvOptions& env_options,
-                  bool use_fsync) override {
-    return target_.env->SyncFile(fname, env_options, use_fsync);
-  }
-
   Status NumFileLinks(const std::string& fname, uint64_t* count) override {
     return target_.env->NumFileLinks(fname, count);
   }

include/rocksdb/file_system.h

@@ -24,6 +24,7 @@
 #include <functional>
 #include <limits>
 #include <memory>
+#include <shared_mutex>
 #include <sstream>
 #include <string>
 #include <unordered_map>
@@ -254,6 +255,10 @@ struct IODebugContext {
   // Arbitrary structure containing cost information about the IO request
   std::any cost_info;
 
+  // FileSystem implementations can use this mutex to synchronize concurrent
+  // reads/writes as needed (e.g. to update the counters or cost_info field)
+  std::shared_mutex mutex;
+
   IODebugContext() {}
 
   // Copy constructor
@@ -606,18 +611,6 @@ class FileSystem : public Customizable {
         "LinkFile is not supported for this FileSystem");
   }
 
-  // Sync the file content to file system.
-  // The default implementation would open, sync and close the file.
-  // This function could be overridden with no-op, if the file system
-  // automatically sync the data when file is closed.
-  // This is used when a user-provided file, probably unsynced, is pulled into a
-  // context where power-outage-proof persistence is required (e.g.
-  // IngestExternalFile without copy).
-  virtual IOStatus SyncFile(const std::string& fname,
-                            const FileOptions& file_options,
-                            const IOOptions& io_options, bool use_fsync,
-                            IODebugContext* dbg);
-
   virtual IOStatus NumFileLinks(const std::string& /*fname*/,
                                 const IOOptions& /*options*/,
                                 uint64_t* /*count*/, IODebugContext* /*dbg*/) {
@@ -1604,12 +1597,6 @@ class FileSystemWrapper : public FileSystem {
     return target_->LinkFile(s, t, options, dbg);
   }
 
-  IOStatus SyncFile(const std::string& fname, const FileOptions& file_options,
-                    const IOOptions& io_options, bool use_fsync,
-                    IODebugContext* dbg) override {
-    return target_->SyncFile(fname, file_options, io_options, use_fsync, dbg);
-  }
-
   IOStatus NumFileLinks(const std::string& fname, const IOOptions& options,
                         uint64_t* count, IODebugContext* dbg) override {
     return target_->NumFileLinks(fname, options, count, dbg);

include/rocksdb/iostats_context.h

@@ -32,6 +32,8 @@ struct FileIOByTemperature {
   uint64_t hot_file_bytes_read;
   // the number of bytes read to Temperature::kWarm file
   uint64_t warm_file_bytes_read;
+  // the number of bytes read to Temperature::kCool file
+  uint64_t cool_file_bytes_read;
   // the number of bytes read to Temperature::kCold file
   uint64_t cold_file_bytes_read;
   // the number of bytes read to Temperature::kIce file
@@ -40,6 +42,8 @@ struct FileIOByTemperature {
   uint64_t hot_file_read_count;
   // total number of reads to Temperature::kWarm file
   uint64_t warm_file_read_count;
+  // total number of reads to Temperature::kCool file
+  uint64_t cool_file_read_count;
   // total number of reads to Temperature::kCold file
   uint64_t cold_file_read_count;
   // total number of reads to Temperature::kIce file
@@ -48,10 +52,12 @@ struct FileIOByTemperature {
   void Reset() {
     hot_file_bytes_read = 0;
     warm_file_bytes_read = 0;
+    cool_file_bytes_read = 0;
     cold_file_bytes_read = 0;
     ice_file_bytes_read = 0;
     hot_file_read_count = 0;
     warm_file_read_count = 0;
+    cool_file_read_count = 0;
     cold_file_read_count = 0;
     ice_file_read_count = 0;
   }

include/rocksdb/multi_scan.h

@@ -97,7 +97,13 @@ class Scan {
 
     ScanIterator() : db_iter_(nullptr), valid_(false) {}
 
-    ~ScanIterator() { assert(status_.ok()); }
+    ~ScanIterator() {
+      if (!status_.ok()) {
+        fprintf(stderr, "ScanIterator status: %s\n",
+                status_.ToString().c_str());
+        assert(false);
+      }
+    }
 
     ScanIterator& operator++() {
       if (!valid_) {
@@ -185,6 +191,10 @@ class MultiScan {
       if (scan_opts_.empty()) {
         throw std::logic_error("Zero scans in multi-scan");
       }
+      status_ = db_iter_->status();
+      if (!status_.ok()) {
+        throw MultiScanException(status_);
+      }
       db_iter_->Seek(*scan_opts_[idx_].range.start);
       status_ = db_iter_->status();
       if (!status_.ok()) {

include/rocksdb/options.h

@@ -1848,7 +1848,7 @@ class MultiScanArgs {
   operator std::vector<ScanOptions>*() { return &original_ranges_; }
 
   operator const std::vector<ScanOptions>*() const { return &original_ranges_; }
-  // Destructor
+
   ~MultiScanArgs() {}
 
   const std::vector<ScanOptions>& GetScanRanges() const {
@@ -1857,6 +1857,14 @@ class MultiScanArgs {
 
   const Comparator* GetComparator() const { return comp_; }
 
+  // Copies the configurations (excluding actual scan ranges) from another
+  // MultiScanArgs.
+  void CopyConfigFrom(const MultiScanArgs& other) {
+    io_coalesce_threshold = other.io_coalesce_threshold;
+    max_prefetch_size = other.max_prefetch_size;
+    use_async_io = other.use_async_io;
+  }
+
   uint64_t io_coalesce_threshold = 16 << 10;  // 16KB by default
 
   // Maximum size (in bytes) for the data blocks loaded by a MultiScan.

include/rocksdb/statistics.h

@@ -443,10 +443,12 @@ enum Tickers : uint32_t {
   // Tiered storage related statistics
   HOT_FILE_READ_BYTES,
   WARM_FILE_READ_BYTES,
+  COOL_FILE_READ_BYTES,
   COLD_FILE_READ_BYTES,
   ICE_FILE_READ_BYTES,
   HOT_FILE_READ_COUNT,
   WARM_FILE_READ_COUNT,
+  COOL_FILE_READ_COUNT,
   COLD_FILE_READ_COUNT,
   ICE_FILE_READ_COUNT,
 

include/rocksdb/types.h

@@ -118,6 +118,7 @@ enum class Temperature : uint8_t {
   kUnknown = 0,
   kHot = 0x04,
   kWarm = 0x08,
+  kCool = 0x0A,
   kCold = 0x0C,
   kIce = 0x10,
   // XXX: this is mis-named. It is instead an invalid temperature beyond the

include/rocksdb/user_defined_index.h

@@ -30,8 +30,7 @@ inline const std::string kUserDefinedIndexPrefix =
 //
 // This is currently supported only for a restricted set of use cases. The
 // CF must be ingest only, and only files containing Puts generated by
-// SstFileWriter are supported. The user_comparator used for the CF must
-// be BytewiseComparator.
+// SstFileWriter are supported.
 
 // The interface for building user-defined index.
 class UserDefinedIndexBuilder {
@@ -145,6 +144,11 @@ class UserDefinedIndexReader {
   virtual size_t ApproximateMemoryUsage() const = 0;
 };
 
+// Options for user defined index
+struct UserDefinedIndexOption {
+  const Comparator* comparator = BytewiseComparator();
+};
+
 // Factory for creating user-defined index builders.
 class UserDefinedIndexFactory : public Customizable {
  public:
@@ -163,6 +167,21 @@ class UserDefinedIndexFactory : public Customizable {
   // block
   virtual std::unique_ptr<UserDefinedIndexReader> NewReader(
       Slice& index_block) const = 0;
+
+  // New API for allowing customized comparator
+  virtual Status NewBuilder(
+      const UserDefinedIndexOption& /*option*/,
+      std::unique_ptr<UserDefinedIndexBuilder>& builder) const {
+    builder.reset(NewBuilder());
+    return Status::OK();
+  };
+
+  virtual Status NewReader(
+      const UserDefinedIndexOption& /*option*/, Slice& index_block,
+      std::unique_ptr<UserDefinedIndexReader>& reader) const {
+    reader = NewReader(index_block);
+    return Status::OK();
+  };
 };
 
 }  // namespace ROCKSDB_NAMESPACE

include/rocksdb/version.h

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

java/rocksjni/portal.h

@@ -5279,6 +5279,14 @@ class TickerTypeJni {
         return -0x59;
       case ROCKSDB_NAMESPACE::Tickers::ICE_FILE_READ_COUNT:
         return -0x5A;
+      case ROCKSDB_NAMESPACE::Tickers::COOL_FILE_READ_BYTES:
+        return -0x5B;
+      case ROCKSDB_NAMESPACE::Tickers::COOL_FILE_READ_COUNT:
+        return -0x5C;
+      case ROCKSDB_NAMESPACE::Tickers::NUMBER_WBWI_INGEST:
+        return -0x5D;
+      case ROCKSDB_NAMESPACE::Tickers::SST_USER_DEFINED_INDEX_LOAD_FAIL_COUNT:
+        return -0x5E;
       case ROCKSDB_NAMESPACE::Tickers::TICKER_ENUM_MAX:
         // -0x54 is the max value at this time. Since these values are exposed
         // directly to Java clients, we'll keep the value the same till the next
@@ -5747,6 +5755,15 @@ class TickerTypeJni {
         return ROCKSDB_NAMESPACE::Tickers::ICE_FILE_READ_BYTES;
       case -0x5A:
         return ROCKSDB_NAMESPACE::Tickers::ICE_FILE_READ_COUNT;
+      case -0x5B:
+        return ROCKSDB_NAMESPACE::Tickers::COOL_FILE_READ_BYTES;
+      case -0x5C:
+        return ROCKSDB_NAMESPACE::Tickers::COOL_FILE_READ_COUNT;
+      case -0x5D:
+        return ROCKSDB_NAMESPACE::Tickers::NUMBER_WBWI_INGEST;
+      case -0x5E:
+        return ROCKSDB_NAMESPACE::Tickers::
+            SST_USER_DEFINED_INDEX_LOAD_FAIL_COUNT;
       case -0x54:
         // -0x54 is the max value at this time. Since these values are exposed
         // directly to Java clients, we'll keep the value the same till the next

java/src/main/java/org/rocksdb/TickerType.java

@@ -764,12 +764,14 @@ public enum TickerType {
      */
     HOT_FILE_READ_BYTES((byte) -0x31),
     WARM_FILE_READ_BYTES((byte) -0x32),
+    COOL_FILE_READ_BYTES((byte) -0x5B),
     COLD_FILE_READ_BYTES((byte) -0x33),
-    ICE_FILE_READ_BYTES((byte) -0x58),
+    ICE_FILE_READ_BYTES((byte) -0x59),
     HOT_FILE_READ_COUNT((byte) -0x34),
     WARM_FILE_READ_COUNT((byte) -0x35),
+    COOL_FILE_READ_COUNT((byte) -0x5C),
     COLD_FILE_READ_COUNT((byte) -0x36),
-    ICE_FILE_READ_COUNT((byte) -0x59),
+    ICE_FILE_READ_COUNT((byte) -0x5A),
 
     /**
      * (non-)last level read statistics
@@ -872,6 +874,8 @@ public enum TickerType {
 
     FIFO_TTL_COMPACTIONS((byte) -0x50),
 
+    FIFO_CHANGE_TEMPERATURE_COMPACTIONS((byte) -0x58),
+
     PREFETCH_BYTES((byte) -0x51),
 
     PREFETCH_BYTES_USEFUL((byte) -0x52),
@@ -884,6 +888,19 @@ public enum TickerType {
 
     FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT((byte) -0x57),
 
+    /**
+     * Counter for the number of times a WBWI is ingested into the DB. This
+     * happens when IngestWriteBatchWithIndex() is used and when large
+     * transaction optimization is enabled through
+     * TransactionOptions::large_txn_commit_optimize_threshold.
+     */
+    NUMBER_WBWI_INGEST((byte) -0x5D),
+
+    /**
+     * Failure to load the UDI during SST table open
+     */
+    SST_USER_DEFINED_INDEX_LOAD_FAIL_COUNT((byte) -0x5E),
+
     TICKER_ENUM_MAX((byte) -0x54);
 
     private final byte value;

monitoring/iostats_context.cc

@@ -65,9 +65,11 @@ std::string IOStatsContext::ToString(bool exclude_zero_counters) const {
   IOSTATS_CONTEXT_OUTPUT(cpu_read_nanos);
   IOSTATS_CONTEXT_OUTPUT(file_io_stats_by_temperature.hot_file_bytes_read);
   IOSTATS_CONTEXT_OUTPUT(file_io_stats_by_temperature.warm_file_bytes_read);
+  IOSTATS_CONTEXT_OUTPUT(file_io_stats_by_temperature.cool_file_bytes_read);
   IOSTATS_CONTEXT_OUTPUT(file_io_stats_by_temperature.cold_file_bytes_read);
   IOSTATS_CONTEXT_OUTPUT(file_io_stats_by_temperature.hot_file_read_count);
   IOSTATS_CONTEXT_OUTPUT(file_io_stats_by_temperature.warm_file_read_count);
+  IOSTATS_CONTEXT_OUTPUT(file_io_stats_by_temperature.cool_file_read_count);
   IOSTATS_CONTEXT_OUTPUT(file_io_stats_by_temperature.cold_file_read_count);
   std::string str = ss.str();
   str.erase(str.find_last_not_of(", ") + 1);

monitoring/statistics.cc

@@ -226,10 +226,12 @@ const std::vector<std::pair<Tickers, std::string>> TickersNameMap = {
     {REMOTE_COMPACT_WRITE_BYTES, "rocksdb.remote.compact.write.bytes"},
     {HOT_FILE_READ_BYTES, "rocksdb.hot.file.read.bytes"},
     {WARM_FILE_READ_BYTES, "rocksdb.warm.file.read.bytes"},
+    {COOL_FILE_READ_BYTES, "rocksdb.cool.file.read.bytes"},
     {COLD_FILE_READ_BYTES, "rocksdb.cold.file.read.bytes"},
     {ICE_FILE_READ_BYTES, "rocksdb.ice.file.read.bytes"},
     {HOT_FILE_READ_COUNT, "rocksdb.hot.file.read.count"},
     {WARM_FILE_READ_COUNT, "rocksdb.warm.file.read.count"},
+    {COOL_FILE_READ_COUNT, "rocksdb.cool.file.read.count"},
     {COLD_FILE_READ_COUNT, "rocksdb.cold.file.read.count"},
     {ICE_FILE_READ_COUNT, "rocksdb.ice.file.read.count"},
     {LAST_LEVEL_READ_BYTES, "rocksdb.last.level.read.bytes"},

options/options_helper.cc

@@ -367,11 +367,9 @@ std::map<CompactionStopStyle, std::string>
         {kCompactionStopStyleTotalSize, "kCompactionStopStyleTotalSize"}};
 
 std::map<Temperature, std::string> OptionsHelper::temperature_to_string = {
-    {Temperature::kUnknown, "kUnknown"},
-    {Temperature::kHot, "kHot"},
-    {Temperature::kWarm, "kWarm"},
-    {Temperature::kCold, "kCold"},
-    {Temperature::kIce, "kIce"}};
+    {Temperature::kUnknown, "kUnknown"}, {Temperature::kHot, "kHot"},
+    {Temperature::kWarm, "kWarm"},       {Temperature::kCool, "kCool"},
+    {Temperature::kCold, "kCold"},       {Temperature::kIce, "kIce"}};
 
 std::unordered_map<std::string, ChecksumType>
     OptionsHelper::checksum_type_string_map = {{"kNoChecksum", kNoChecksum},
@@ -966,11 +964,9 @@ std::unordered_map<std::string, CompactionStopStyle>
 
 std::unordered_map<std::string, Temperature>
     OptionsHelper::temperature_string_map = {
-        {"kUnknown", Temperature::kUnknown},
-        {"kHot", Temperature::kHot},
-        {"kWarm", Temperature::kWarm},
-        {"kCold", Temperature::kCold},
-        {"kIce", Temperature::kIce}};
+        {"kUnknown", Temperature::kUnknown}, {"kHot", Temperature::kHot},
+        {"kWarm", Temperature::kWarm},       {"kCool", Temperature::kCool},
+        {"kCold", Temperature::kCold},       {"kIce", Temperature::kIce}};
 
 std::unordered_map<std::string, PrepopulateBlobCache>
     OptionsHelper::prepopulate_blob_cache_string_map = {

table/block_based/block_based_table_builder.cc

@@ -1190,20 +1190,21 @@ struct BlockBasedTableBuilder::Rep {
         SetStatus(
             Status::InvalidArgument("user_defined_index_factory not supported "
                                     "with parallel compression"));
-      } else if (ioptions.user_comparator != BytewiseComparator()) {
-        // TODO: Pass the user_comparator to the UDI and let it validate. Do
-        // it in a major release.
-        SetStatus(
-            Status::InvalidArgument("user_defined_index_factory only supported "
-                                    "with bytewise comparator"));
       } else {
-        std::unique_ptr<UserDefinedIndexBuilder> user_defined_index_builder(
-            table_options.user_defined_index_factory->NewBuilder());
-        if (user_defined_index_builder != nullptr) {
-          index_builder = std::make_unique<UserDefinedIndexBuilderWrapper>(
-              std::string(table_options.user_defined_index_factory->Name()),
-              std::move(index_builder), std::move(user_defined_index_builder),
-              &internal_comparator, ts_sz, persist_user_defined_timestamps);
+        std::unique_ptr<UserDefinedIndexBuilder> user_defined_index_builder;
+        UserDefinedIndexOption udi_options;
+        udi_options.comparator = internal_comparator.user_comparator();
+        auto s = table_options.user_defined_index_factory->NewBuilder(
+            udi_options, user_defined_index_builder);
+        if (!s.ok()) {
+          SetStatus(s);
+        } else {
+          if (user_defined_index_builder != nullptr) {
+            index_builder = std::make_unique<UserDefinedIndexBuilderWrapper>(
+                std::string(table_options.user_defined_index_factory->Name()),
+                std::move(index_builder), std::move(user_defined_index_builder),
+                &internal_comparator, ts_sz, persist_user_defined_timestamps);
+          }
         }
       }
     }

table/block_based/block_based_table_iterator.cc

@@ -37,13 +37,13 @@ void BlockBasedTableIterator::SeekImpl(const Slice* target,
                                        bool async_prefetch) {
   // TODO(hx235): set `seek_key_prefix_for_readahead_trimming_`
   // even when `target == nullptr` that is when `SeekToFirst()` is called
-  if (multi_scan_) {
-    if (SeekMultiScan(target)) {
-      return;
-    }
+  if (!multi_scan_status_.ok()) {
+    return;
+  }
+  if (multi_scan_) {
+    SeekMultiScan(target);
+    return;
   }
-
-  assert(!multi_scan_);
 
   if (target != nullptr && prefix_extractor_ &&
       read_options_.prefix_same_as_start) {
@@ -961,30 +961,43 @@ BlockBasedTableIterator::MultiScanState::~MultiScanState() {
 // - scan ranges should be non-overlapping, and have increasing start keys.
 // If a scan range's limit is not set, then there should only be one scan range.
 // - After Prepare(), the iterator expects Seek to be called on the start key
-// of each ScanOption in order. If any other seek is done, the optimization here
-// is aborted and fall back to vanilla iterator.
+// of each ScanOption in order. If any other Seek is done, an error status is
+// returned
+// - Whenever all blocks of a scan opt are exhausted, the iterator will become
+// invalid and UpperBoundCheckResult() will return kOutOfBound. So that the
+// upper layer (LevelIterator) will stop scanning instead thinking EOF is
+// reached and continue into the next file. The only exception is for the last
+// scan opt. If we reach the end of the last scan opt, UpperBoundCheckResult()
+// will return kUnknown instead of kOutOfBound. This mechanism requires that
+// scan opts are properly pruned such that there is no scan opt that is after
+// this file's key range.
 // FIXME: DBIter and MergingIterator may
 // internally do Seek() on child iterators, e.g. due to
 // ReadOptions::max_skippable_internal_keys or reseeking into range deletion
-// end key. So these Seeks can cause iterator to fall back to normal
-// (non-prepared) iterator and ignore the optimizations done in Prepare().
+// end key. These Seeks will be handled properly, as long as the target is
+// moving forward.
 void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
-  index_iter_->Prepare(multiscan_opts);
-
   assert(!multi_scan_);
+  if (!index_iter_->status().ok()) {
+    multi_scan_status_ = index_iter_->status();
+    return;
+  }
   if (multi_scan_) {
     multi_scan_.reset();
-    return;
-  }
-  if (!ValidateScanOptions(multiscan_opts)) {
+    multi_scan_status_ = Status::InvalidArgument("Prepare already called");
     return;
   }
 
+  index_iter_->Prepare(multiscan_opts);
+
   std::vector<BlockHandle> scan_block_handles;
+  std::vector<std::string> data_block_separators;
   std::vector<std::tuple<size_t, size_t>> block_index_ranges_per_scan;
   const std::vector<ScanOptions>& scan_opts = multiscan_opts->GetScanRanges();
-  if (!CollectBlockHandles(scan_opts, &scan_block_handles,
-                           &block_index_ranges_per_scan)) {
+  multi_scan_status_ =
+      CollectBlockHandles(scan_opts, &scan_block_handles,
+                          &block_index_ranges_per_scan, &data_block_separators);
+  if (!multi_scan_status_.ok()) {
     return;
   }
 
@@ -993,9 +1006,10 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
   std::vector<CachableEntry<Block>> pinned_data_blocks_guard(
       scan_block_handles.size());
   size_t prefetched_max_idx;
-  if (!FilterAndPinCachedBlocks(
-          scan_block_handles, multiscan_opts, &block_indices_to_read,
-          &pinned_data_blocks_guard, &prefetched_max_idx)) {
+  multi_scan_status_ = FilterAndPinCachedBlocks(
+      scan_block_handles, multiscan_opts, &block_indices_to_read,
+      &pinned_data_blocks_guard, &prefetched_max_idx);
+  if (!multi_scan_status_.ok()) {
     return;
   }
 
@@ -1009,8 +1023,10 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
                       &read_reqs, &block_idx_to_readreq_idx,
                       &coalesced_block_indices);
 
-    if (!ExecuteIO(scan_block_handles, multiscan_opts, coalesced_block_indices,
-                   &read_reqs, &async_states, &pinned_data_blocks_guard)) {
+    multi_scan_status_ =
+        ExecuteIO(scan_block_handles, multiscan_opts, coalesced_block_indices,
+                  &read_reqs, &async_states, &pinned_data_blocks_guard);
+    if (!multi_scan_status_.ok()) {
       return;
     }
   }
@@ -1019,7 +1035,7 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
   // blocks.
   multi_scan_ = std::make_unique<MultiScanState>(
       table_->get_rep()->ioptions.env->GetFileSystem(), multiscan_opts,
-      std::move(pinned_data_blocks_guard),
+      std::move(pinned_data_blocks_guard), std::move(data_block_separators),
       std::move(block_index_ranges_per_scan),
       std::move(block_idx_to_readreq_idx), std::move(async_states),
       prefetched_max_idx);
@@ -1028,93 +1044,273 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
   block_iter_points_to_real_block_ = false;
 }
 
-bool BlockBasedTableIterator::SeekMultiScan(const Slice* target) {
-  assert(multi_scan_);
+void BlockBasedTableIterator::SeekMultiScan(const Slice* seek_target) {
+  assert(multi_scan_ && multi_scan_status_.ok());
   // This is a MultiScan and Preapre() has been called.
-  //
+
+  // Reset out of bound on seek, if it is out of bound again, it will be set
+  // properly later in the code path
+  is_out_of_bound_ = false;
+
   // Validate seek key with scan options
-  if (multi_scan_->next_scan_idx >= multi_scan_->scan_opts->size()) {
-    multi_scan_.reset();
-  } else if (!target) {
+  if (!seek_target) {
     // start key must be set for multi-scan
-    multi_scan_.reset();
-  } else if (user_comparator_.CompareWithoutTimestamp(
-                 ExtractUserKey(*target), /*a_has_ts=*/true,
-                 multi_scan_->scan_opts
-                     ->GetScanRanges()[multi_scan_->next_scan_idx]
-                     .range.start.value(),
-                 /*b_has_ts=*/false) != 0) {
-    // Unexpected seek key
-    multi_scan_.reset();
+    multi_scan_status_ = Status::InvalidArgument("No seek key for MultiScan");
+    return;
+  }
+
+  // Check the case where there is no range prepared on this table
+  if (multi_scan_->scan_opts->size() == 0) {
+    // out of bound
+    MarkPreparedRangeExhausted();
+    return;
+  }
+
+  // Check whether seek key is moving forward.
+  if (multi_scan_->prev_seek_key_.empty() ||
+      icomp_.Compare(*seek_target, multi_scan_->prev_seek_key_) > 0) {
+    // If seek key is empty or is larger than previous seek key, update the
+    // previous seek key. Otherwise use the previous seek key as the adjusted
+    // seek target moving forward. This prevents seek target going backward,
+    // which would visit pages that have been unpinned.
+    // This issue is caused by sub-optimal range delete handling inside merge
+    // iterator.
+    // TODO xingbo issues:14068 : Optimize the handling of range delete iterator
+    // inside merge iterator, so that it doesn't move seek key backward. After
+    // that we could return error if the key moves backward here.
+    multi_scan_->prev_seek_key_ = seek_target->ToString();
   } else {
-    if (multi_scan_->next_scan_idx > 0) {
-      UnpinPreviousScanBlocks(multi_scan_->next_scan_idx);
+    // Seek key is adjusted to previous one, we can return here directly.
+    return;
+  }
+
+  // There are 3 different Cases we need to handle:
+  // The following diagram explain different seek targets seeking at various
+  // position on the table, while the next_scan_idx points to the PreparedRange
+  // 2.
+  //
+  // next_scan_idx: -------------------┐
+  //                                   ▼
+  // table:     : __[PreparedRange 1]__[PreparedRange 2]__[PreparedRange 3]__
+  // Seek target: <----- Case 1 ------>▲<------------- Case 2 -------------->
+  //                                   │
+  //                                 Case 3
+  //
+  // Case 1: seek before the start of next prepared ranges. This could happen
+  //    due to too many delete tomestone triggered reseek or delete range.
+  // Case 2: seek after the start of next prepared range.
+  //    This could happen due to seek key adjustment from delete range file.
+  //    E.g. LSM has 3 levels, each level has only 1 file:
+  //    L1 : key :              0---10
+  //    L2 : Delete range key : 0-5
+  //    L3 : key :              0---10
+  //    When a range 2-8 was prepared, the prepared key would be 2 on L3 file,
+  //    but the seek key would be 5, as the seek key was updated by the largest
+  //    key of delete range. This causes all of the cases above to be possible,
+  //    when the ranges are adjusted in the above examples.
+  // Case 3: seek at the beginning of a prepared range (expected case)
+
+  // Allow reseek on the start of the last prepared range due to too many
+  // tombstone
+  multi_scan_->next_scan_idx =
+      std::min(multi_scan_->next_scan_idx,
+               multi_scan_->block_index_ranges_per_scan.size() - 1);
+
+  auto user_seek_target = ExtractUserKey(*seek_target);
+
+  auto compare_next_scan_start_result =
+      user_comparator_.CompareWithoutTimestamp(
+          user_seek_target, /*a_has_ts=*/true,
+          multi_scan_->scan_opts->GetScanRanges()[multi_scan_->next_scan_idx]
+              .range.start.value(),
+          /*b_has_ts=*/false);
+
+  if (compare_next_scan_start_result != 0) {
+    // The seek target is not exactly same as what was prepared.
+    if (compare_next_scan_start_result < 0) {
+      // Case 1:
+      if (multi_scan_->next_scan_idx == 0) {
+        // This should not happen, even when seek target is adjusted by delete
+        // range. The reason is that if the seek target is before the start key
+        // of the first prepared range, its end key needs to be >= the smallest
+        // key of this file, otherwise it is skipped in level iterator. If its
+        // end key is >= the smallest key of this file, then this range will be
+        // prepared for this file. As delete range could only adjust seek
+        // target forward, so it would never be before the start key of the
+        // first prepared range.
+        assert(false && "Seek target before the first prepared range");
+        MarkPreparedRangeExhausted();
+        return;
+      }
+      auto seek_target_before_previous_prepared_range =
+          user_comparator_.CompareWithoutTimestamp(
+              user_seek_target, /*a_has_ts=*/true,
+              multi_scan_->scan_opts
+                  ->GetScanRanges()[multi_scan_->next_scan_idx - 1]
+                  .range.start.value(),
+              /*b_has_ts=*/false) < 0;
+      // Not expected to happen
+      // This should never happen, the reason is that the
+      // multi_scan_->next_scan_idx is set to a non zero value is due to a seek
+      // target larger or equal to the start key of multi_scan_->next_scan_idx-1
+      // happended earlier. If a seek happens before the start key of
+      // multi_scan_->next_scan_idx-1, it would seek a key that is less than
+      // what was seeked before.
+      assert(!seek_target_before_previous_prepared_range);
+      if (seek_target_before_previous_prepared_range) {
+        multi_scan_status_ = Status::InvalidArgument(
+            "Seek target is before the previous prepared range at index " +
+            std::to_string(multi_scan_->next_scan_idx));
+        return;
+      }
+      // It should only be possible to seek a key between the start of current
+      // prepared scan and start of next prepared range.
+      MultiScanUnexpectedSeekTarget(
+          seek_target, &user_seek_target,
+          std::get<0>(multi_scan_->block_index_ranges_per_scan
+                          [multi_scan_->next_scan_idx - 1]));
+    } else {
+      // Case 2:
+      MultiScanUnexpectedSeekTarget(
+          seek_target, &user_seek_target,
+          std::get<0>(
+              multi_scan_
+                  ->block_index_ranges_per_scan[multi_scan_->next_scan_idx]));
     }
+  } else {
+    // Case 2:
+    assert(multi_scan_->next_scan_idx <
+           multi_scan_->block_index_ranges_per_scan.size());
 
     auto [cur_scan_start_idx, cur_scan_end_idx] =
         multi_scan_->block_index_ranges_per_scan[multi_scan_->next_scan_idx];
     // We should have the data block already loaded
     ++multi_scan_->next_scan_idx;
     if (cur_scan_start_idx >= cur_scan_end_idx) {
-      is_out_of_bound_ = true;
-      assert(!Valid());
-      return true;
-    } else {
-      is_out_of_bound_ = false;
+      // No blocks are prepared for this range at current file.
+      MarkPreparedRangeExhausted();
+      return;
     }
 
-    if (!block_iter_points_to_real_block_ ||
-        multi_scan_->cur_data_block_idx != cur_scan_start_idx) {
-      if (block_iter_points_to_real_block_) {
-        // Should be scan in increasing key range.
-        // All blocks before cur_data_block_idx_ are not pinned anymore.
-        assert(multi_scan_->cur_data_block_idx < cur_scan_start_idx);
-      }
-
-      ResetDataIter();
-
-      multi_scan_->cur_data_block_idx = cur_scan_start_idx;
-      multi_scan_->status = MultiScanLoadDataBlock(cur_scan_start_idx);
-      if (!multi_scan_->status.ok()) {
-        assert(!Valid());
-        assert(status() == multi_scan_->status);
-        return true;
-      }
-    }
-    multi_scan_->cur_data_block_idx = cur_scan_start_idx;
-    block_iter_points_to_real_block_ = true;
-    block_iter_.Seek(*target);
-    FindKeyForward();
-    return true;
+    MultiScanSeekTargetFromBlock(seek_target, cur_scan_start_idx);
   }
-
-  // We are aborting MultiScan.
-  ResetDataIter();
-  assert(!is_index_at_curr_block_);
-  assert(!block_iter_points_to_real_block_);
-  return false;
 }
 
-void BlockBasedTableIterator::UnpinPreviousScanBlocks(size_t current_scan_idx) {
-  // TODO: support aborting and clearn up async IO requests, currently
-  // only unpins already initialized blocks
-  assert(multi_scan_);
-  assert(current_scan_idx < multi_scan_->block_index_ranges_per_scan.size());
-  if (current_scan_idx == 0) return;
+void BlockBasedTableIterator::MultiScanUnexpectedSeekTarget(
+    const Slice* seek_target, const Slice* user_seek_target, size_t block_idx) {
+  // linear search the block that contains the seek target, and unpin blocks
+  // that are before it.
 
-  auto [prev_start_block_idx, prev_end_block_idx] =
-      multi_scan_->block_index_ranges_per_scan[current_scan_idx - 1];
-  // Since a block can be shared between consecutive scans, we need
-  // curr_start_block_idx here instead of just release blocks
-  // up to prev_end_block_idx.
-  auto [curr_start_block_idx, curr_end_block_idx] =
-      multi_scan_->block_index_ranges_per_scan[current_scan_idx];
-  for (size_t block_idx = prev_start_block_idx;
-       block_idx < curr_start_block_idx; ++block_idx) {
+  // The logic here could be confusing when there is a delete range involved.
+  // E.g. we have an LSM with 3 levels, each level has only 1 file:
+  // L1: data file :    0---10
+  // L2: Delete range : 0-5
+  // L3: data file :    0---10
+  //
+  // MultiScan on ranges 1-2, 3-4, and 5-6.
+  // When user first do Seek(1), on level 2, due to delete range 0-5, the seek
+  // key is adjusted to 5 at level 3. Therefore, we will internally do Seek(5)
+  // and unpins all blocks until 5 at level 3. Then the next scan's blocks from
+  // 3-4 are unpinned at level 3. It is confusing that maybe block 3-4 should
+  // not be unpinned, as next scan would need it. But Seek(5) implies that these
+  // keys are all covered by some range deletion, so the next Seek(3) will also
+  // do Seek(5) internally, so the blocks from 3-4 could be safely unpinned.
+
+  // advance to the right prepared range
+  while (
+      multi_scan_->next_scan_idx <
+          multi_scan_->block_index_ranges_per_scan.size() &&
+      (user_comparator_.CompareWithoutTimestamp(
+           *user_seek_target, /*a_has_ts=*/true,
+           multi_scan_->scan_opts->GetScanRanges()[multi_scan_->next_scan_idx]
+               .range.start.value(),
+           /*b_has_ts=*/false) >= 0)) {
+    multi_scan_->next_scan_idx++;
+  }
+
+  // next_scan_idx is guaranteed to be higher than 0. If the seek key is before
+  // the start key of first prepared range, it is already handled by caller
+  // SeekMultiScan. It is equal, it would not call this funciton. If it is
+  // after, next_scan_idx would be advanced by the loop above.
+  assert(multi_scan_->next_scan_idx > 0);
+  // Get the current range
+  auto cur_scan_idx = multi_scan_->next_scan_idx - 1;
+  auto [cur_scan_start_idx, cur_scan_end_idx] =
+      multi_scan_->block_index_ranges_per_scan[cur_scan_idx];
+
+  if (cur_scan_start_idx >= cur_scan_end_idx) {
+    // No blocks are prepared for this range at current file.
+    MarkPreparedRangeExhausted();
+    return;
+  }
+
+  // Unpin all the blocks from multi_scan_->cur_data_block_idx to
+  // cur_scan_start_idx
+  for (auto unpin_block_idx = multi_scan_->cur_data_block_idx;
+       unpin_block_idx < cur_scan_start_idx; unpin_block_idx++) {
+    if (!multi_scan_->pinned_data_blocks[unpin_block_idx].IsEmpty()) {
+      multi_scan_->pinned_data_blocks[unpin_block_idx].Reset();
+    }
+  }
+
+  // Find the right block_idx;
+  block_idx = cur_scan_start_idx;
+  auto const& data_block_separators = multi_scan_->data_block_separators;
+  while (block_idx < data_block_separators.size() &&
+         (user_comparator_.CompareWithoutTimestamp(
+              *user_seek_target, /*a_has_ts=*/true,
+              data_block_separators[block_idx],
+              /*b_has_ts=*/false) > 0)) {
+    // Unpin the blocks that are passed
     if (!multi_scan_->pinned_data_blocks[block_idx].IsEmpty()) {
       multi_scan_->pinned_data_blocks[block_idx].Reset();
     }
+    block_idx++;
   }
+
+  if (block_idx >= data_block_separators.size()) {
+    // All of the prepared blocks for this file is exhausted.
+    MarkPreparedRangeExhausted();
+    return;
+  }
+
+  // The current block may contain the data for the target key
+  MultiScanSeekTargetFromBlock(seek_target, block_idx);
+}
+
+void BlockBasedTableIterator::MultiScanSeekTargetFromBlock(
+    const Slice* seek_target, size_t block_idx) {
+  assert(multi_scan_->cur_data_block_idx <= block_idx);
+
+  if (!block_iter_points_to_real_block_ ||
+      multi_scan_->cur_data_block_idx != block_idx) {
+    if (block_iter_points_to_real_block_) {
+      // Should be scan in increasing key range.
+      // All blocks before cur_data_block_idx_ are not pinned anymore.
+      assert(multi_scan_->cur_data_block_idx < block_idx);
+    }
+
+    ResetDataIter();
+
+    if (MultiScanLoadDataBlock(block_idx)) {
+      return;
+    }
+  }
+
+  // Move current data block index forward until block_idx, meantime, unpin all
+  // the blocks in between
+  while (multi_scan_->cur_data_block_idx < block_idx) {
+    // unpin block
+    if (!multi_scan_->pinned_data_blocks[multi_scan_->cur_data_block_idx]
+             .IsEmpty()) {
+      multi_scan_->pinned_data_blocks[multi_scan_->cur_data_block_idx].Reset();
+    }
+    multi_scan_->cur_data_block_idx++;
+  }
+  block_iter_points_to_real_block_ = true;
+  block_iter_.Seek(*seek_target);
+  FindKeyForward();
+  CheckOutOfBound();
 }
 
 void BlockBasedTableIterator::FindBlockForwardInMultiScan() {
@@ -1133,31 +1329,19 @@ void BlockBasedTableIterator::FindBlockForwardInMultiScan() {
     // for this file, it may need to continue to scan into the next file, so
     // we do not set is_out_of_bound_ in this case.
     if (multi_scan_->cur_data_block_idx + 1 >= cur_scan_end_idx) {
-      if (multi_scan_->next_scan_idx >=
-          multi_scan_->block_index_ranges_per_scan.size()) {
-        // We are done with this file, should let LevelIter advance to the next
-        // file instead of ending the scan
-        ResetDataIter();
-        assert(!is_out_of_bound_);
-        assert(!Valid());
-        return;
-      }
-      // We don't ResetDataIter() here since next scan might be reading from
-      // the same block. ResetDataIter() will free the underlying block cache
-      // handle and we don't want the block to be unpinned.
-      is_out_of_bound_ = true;
-      assert(!Valid());
+      MarkPreparedRangeExhausted();
       return;
     }
     // Move to the next pinned data block
     ResetDataIter();
+    // Unpin previous block if it is not reset by data iterator
+    if (!multi_scan_->pinned_data_blocks[multi_scan_->cur_data_block_idx]
+             .IsEmpty()) {
+      multi_scan_->pinned_data_blocks[multi_scan_->cur_data_block_idx].Reset();
+    }
     ++multi_scan_->cur_data_block_idx;
 
-    multi_scan_->status =
-        MultiScanLoadDataBlock(multi_scan_->cur_data_block_idx);
-    if (!multi_scan_->status.ok()) {
-      assert(!Valid());
-      assert(status() == multi_scan_->status);
+    if (MultiScanLoadDataBlock(multi_scan_->cur_data_block_idx)) {
       return;
     }
 
@@ -1268,54 +1452,19 @@ Status BlockBasedTableIterator::CreateAndPinBlockFromBuffer(
       &pinned_block_entry.As<Block_kData>());
 }
 
-bool BlockBasedTableIterator::ValidateScanOptions(
-    const MultiScanArgs* multiscan_opts) {
-  if (multiscan_opts == nullptr || multiscan_opts->empty()) {
-    return false;
-  }
+constexpr auto kVerbose = false;
 
-  const std::vector<ScanOptions>& scan_opts = multiscan_opts->GetScanRanges();
-  const bool has_limit = scan_opts.front().range.limit.has_value();
-  if (!has_limit && scan_opts.size() > 1) {
-    // Abort: overlapping ranges
-    return false;
-  }
-
-  for (size_t i = 0; i < scan_opts.size(); ++i) {
-    const auto& scan_range = scan_opts[i].range;
-    if (!scan_range.start.has_value()) {
-      // Abort: no start key
-      return false;
-    }
-
-    if (scan_range.limit.has_value()) {
-      assert(user_comparator_.CompareWithoutTimestamp(
-                 scan_range.start.value(), /*a_has_ts=*/false,
-                 scan_range.limit.value(), /*b_has_ts=*/false) <= 0);
-    }
-
-    if (i > 0) {
-      if (!scan_range.limit.has_value()) {
-        // multiple no limit scan ranges
-        return false;
-      }
-
-      const auto& last_end_key = scan_opts[i - 1].range.limit.value();
-      if (user_comparator_.CompareWithoutTimestamp(
-              scan_range.start.value(), /*a_has_ts=*/false, last_end_key,
-              /*b_has_ts=*/false) < 0) {
-        // Abort: overlapping ranges
-        return false;
-      }
-    }
-  }
-  return true;
-}
-
-bool BlockBasedTableIterator::CollectBlockHandles(
+Status BlockBasedTableIterator::CollectBlockHandles(
     const std::vector<ScanOptions>& scan_opts,
     std::vector<BlockHandle>* scan_block_handles,
-    std::vector<std::tuple<size_t, size_t>>* block_index_ranges_per_scan) {
+    std::vector<std::tuple<size_t, size_t>>* block_index_ranges_per_scan,
+    std::vector<std::string>* data_block_separators) {
+  // print file name and level
+  if (UNLIKELY(kVerbose)) {
+    auto file_name = table_->get_rep()->file->file_name();
+    auto level = table_->get_rep()->level;
+    printf("file name : %s, level %d\n", file_name.c_str(), level);
+  }
   for (const auto& scan_opt : scan_opts) {
     size_t num_blocks = 0;
     bool check_overlap = !scan_block_handles->empty();
@@ -1335,15 +1484,19 @@ bool BlockBasedTableIterator::CollectBlockHandles(
     index_iter_->Seek(start_key.Encode());
     while (index_iter_->status().ok() && index_iter_->Valid() &&
            (!scan_opt.range.limit.has_value() ||
-            user_comparator_.CompareWithoutTimestamp(
-                index_iter_->user_key(),
-                /*a_has_ts*/ true, *scan_opt.range.limit,
-                /*b_has_ts=*/false) <= 0)) {
+            user_comparator_.CompareWithoutTimestamp(index_iter_->user_key(),
+                                                     /*a_has_ts*/ true,
+                                                     *scan_opt.range.limit,
+                                                     /*b_has_ts=*/false) < 0)) {
+      // Only add the block if the index separator is smaller than limit. When
+      // they are equal or larger, it will be handled later below.
       if (check_overlap &&
           scan_block_handles->back() == index_iter_->value().handle) {
         // Skip the current block since it's already in the list
       } else {
         scan_block_handles->push_back(index_iter_->value().handle);
+        // clone the Slice to avoid the lifetime issue
+        data_block_separators->push_back(index_iter_->user_key().ToString());
       }
       ++num_blocks;
       index_iter_->Next();
@@ -1352,32 +1505,39 @@ bool BlockBasedTableIterator::CollectBlockHandles(
 
     if (!index_iter_->status().ok()) {
       // Abort: index iterator error
-      return false;
+      return index_iter_->status();
     }
 
     if (index_iter_->Valid()) {
+      // Handle the last block when its separator is equal or larger than limit
       if (check_overlap &&
           scan_block_handles->back() == index_iter_->value().handle) {
         // Skip adding the current block since it's already in the list
       } else {
         scan_block_handles->push_back(index_iter_->value().handle);
+        data_block_separators->push_back(index_iter_->user_key().ToString());
       }
       ++num_blocks;
-    } else if (num_blocks == 0 && index_iter_->UpperBoundCheckResult() !=
-                                      IterBoundCheck::kOutOfBound) {
-      // We should not have scan ranges that are completely after the file's
-      // range. This is important for FindBlockForwardInMultiScan() which only
-      // lets the upper layer (LevelIterator) advance to the next SST file when
-      // the last scan range is exhausted.
-      return false;
     }
     block_index_ranges_per_scan->emplace_back(
         scan_block_handles->size() - num_blocks, scan_block_handles->size());
+    if (UNLIKELY(kVerbose)) {
+      printf("separators :");
+      for (const auto& separator : *data_block_separators) {
+        printf("%s, ", separator.c_str());
+      }
+      printf("\nblock_index_ranges_per_scan :");
+      for (auto const& block_index_range : *block_index_ranges_per_scan) {
+        printf("[%zu, %zu], ", std::get<0>(block_index_range),
+               std::get<1>(block_index_range));
+      }
+      printf("\n");
+    }
   }
-  return true;
+  return Status::OK();
 }
 
-bool BlockBasedTableIterator::FilterAndPinCachedBlocks(
+Status BlockBasedTableIterator::FilterAndPinCachedBlocks(
     const std::vector<BlockHandle>& scan_block_handles,
     const MultiScanArgs* multiscan_opts,
     std::vector<size_t>* block_indices_to_read,
@@ -1406,14 +1566,14 @@ bool BlockBasedTableIterator::FilterAndPinCachedBlocks(
 
     if (!s.ok()) {
       // Abort: block cache look up failed.
-      return false;
+      return s;
     }
     if (!(*pinned_data_blocks_guard)[i].GetValue()) {
       // Block not in cache
       block_indices_to_read->emplace_back(i);
     }
   }
-  return true;
+  return Status::OK();
 }
 
 void BlockBasedTableIterator::PrepareIORequests(
@@ -1500,7 +1660,7 @@ void BlockBasedTableIterator::PrepareIORequests(
   }
 }
 
-bool BlockBasedTableIterator::ExecuteIO(
+Status BlockBasedTableIterator::ExecuteIO(
     const std::vector<BlockHandle>& scan_block_handles,
     const MultiScanArgs* multiscan_opts,
     const std::vector<std::vector<size_t>>& coalesced_block_indices,
@@ -1508,9 +1668,11 @@ bool BlockBasedTableIterator::ExecuteIO(
     std::vector<AsyncReadState>* async_states,
     std::vector<CachableEntry<Block>>* pinned_data_blocks_guard) {
   IOOptions io_opts;
-  if (!table_->get_rep()->file->PrepareIOOptions(read_options_, io_opts).ok()) {
+  Status s;
+  s = table_->get_rep()->file->PrepareIOOptions(read_options_, io_opts);
+  if (!s.ok()) {
     // Abort: PrepareIOOptions failed
-    return false;
+    return s;
   }
   const bool direct_io = table_->get_rep()->file->use_direct_io();
 
@@ -1538,7 +1700,7 @@ bool BlockBasedTableIterator::ExecuteIO(
                           this, std::placeholders::_1, std::placeholders::_2);
       // TODO: for mmap, io_handle will not be set but callback will already
       // be called.
-      Status s = table_->get_rep()->file.get()->ReadAsync(
+      s = table_->get_rep()->file.get()->ReadAsync(
           read_req, io_opts, cb, &async_read, &async_read.io_handle,
           &async_read.del_fn, direct_io ? &async_read.aligned_buf : nullptr);
       if (!s.ok()) {
@@ -1546,13 +1708,15 @@ bool BlockBasedTableIterator::ExecuteIO(
         fprintf(stderr, "ReadAsync failed with %s\n", s.ToString().c_str());
 #endif
         assert(false);
-        return false;
+        return s;
       }
       assert(async_read.io_handle);
       for (auto& req : *read_reqs) {
         if (!req.status.ok()) {
           assert(false);
-          return false;
+          // Silence compiler warning about NRVO
+          s = req.status;
+          return s;
         }
       }
     }
@@ -1579,15 +1743,17 @@ bool BlockBasedTableIterator::ExecuteIO(
     }
 
     AlignedBuf aligned_buf;
-    Status s = table_->get_rep()->file->MultiRead(
-        io_opts, read_reqs->data(), read_reqs->size(),
-        direct_io ? &aligned_buf : nullptr);
+    s = table_->get_rep()->file->MultiRead(io_opts, read_reqs->data(),
+                                           read_reqs->size(),
+                                           direct_io ? &aligned_buf : nullptr);
     if (!s.ok()) {
-      return false;
+      return s;
     }
     for (auto& req : *read_reqs) {
       if (!req.status.ok()) {
-        return false;
+        // Silence compiler warning about NRVO
+        s = req.status;
+        return s;
       }
     }
 
@@ -1604,13 +1770,13 @@ bool BlockBasedTableIterator::ExecuteIO(
         if (!s.ok()) {
           assert(false);
           // Abort: failed to create and pin block in cache
-          return false;
+          return s;
         }
         assert((*pinned_data_blocks_guard)[block_idx].GetValue());
       }
     }
   }
-  return true;
+  return s;
 }
 
 }  // namespace ROCKSDB_NAMESPACE

table/block_based/block_based_table_iterator.h

@@ -45,7 +45,9 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
         need_upper_bound_check_(need_upper_bound_check),
         async_read_in_progress_(false),
         is_last_level_(table->IsLastLevel()),
-        block_iter_points_to_real_block_(false) {}
+        block_iter_points_to_real_block_(false) {
+    multi_scan_status_.PermitUncheckedError();
+  }
 
   ~BlockBasedTableIterator() override { ClearBlockHandles(); }
 
@@ -57,7 +59,7 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
   bool NextAndGetResult(IterateResult* result) override;
   void Prev() override;
   bool Valid() const override {
-    return !is_out_of_bound_ &&
+    return !is_out_of_bound_ && multi_scan_status_.ok() &&
            (is_at_first_key_from_index_ ||
             (block_iter_points_to_real_block_ && block_iter_.Valid()));
   }
@@ -136,6 +138,9 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
     return block_iter_.value();
   }
   Status status() const override {
+    if (!multi_scan_status_.ok()) {
+      return multi_scan_status_;
+    }
     // In case of block cache readahead lookup, it won't add the block to
     // block_handles if it's index is invalid. So index_iter_->status check can
     // be skipped.
@@ -151,7 +156,7 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
       assert(!multi_scan_);
       return Status::TryAgain("Async read in progress");
     } else if (multi_scan_) {
-      return multi_scan_->status;
+      return multi_scan_status_;
     } else {
       return Status::OK();
     }
@@ -376,6 +381,27 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
   bool block_iter_points_to_real_block_;
   // See InternalIteratorBase::IsOutOfBound().
   bool is_out_of_bound_ = false;
+
+  // Mark prepared ranges as exhausted for multiscan.
+  void MarkPreparedRangeExhausted() {
+    assert(multi_scan_ != nullptr);
+    if (multi_scan_->next_scan_idx <
+        multi_scan_->block_index_ranges_per_scan.size()) {
+      // If there are more prepared ranges, we don't ResetDataIter() here,
+      // because next scan might be reading from the same block. ResetDataIter()
+      // will free the underlying block cache handle and we don't want the
+      // block to be unpinned.
+      // Set out of bound to mark the current prepared range as exhausted.
+      is_out_of_bound_ = true;
+    } else {
+      // This is the last prepared range of this file, there might be more
+      // data on next file. Reset data iterator to indicate the iterator is
+      // no longer valid on this file. Let LevelIter advance to the next file
+      // instead of ending the scan.
+      ResetDataIter();
+    }
+  }
+
   // During cache lookup to find readahead size, index_iter_ is iterated and it
   // can point to a different block.
   // If Prepare() is called, index_iter_ is used to prefetch data blocks for the
@@ -433,6 +459,15 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
     const std::shared_ptr<FileSystem> fs;
     const MultiScanArgs* scan_opts;
     std::vector<CachableEntry<Block>> pinned_data_blocks;
+    // The separator of each data block in above pinned_data_blocks vector.
+    // Its size is same as pinned_data_blocks.
+    // The value of separator is larger than or equal to the last key in the
+    // corresponding data block.
+    std::vector<std::string> data_block_separators;
+    // Track previously seeked key in multi-scan.
+    // This is used to ensure that the seek key is keep moving forward, as
+    // blocks that are smaller than the seek key are unpinned from memory.
+    std::string prev_seek_key_;
 
     // Indicies into pinned_data_blocks for data blocks for each scan range.
     // inclusive start, exclusive end
@@ -454,31 +489,30 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
     // async_states[j].
     std::vector<AsyncReadState> async_states;
     UnorderedMap<size_t, size_t> block_idx_to_readreq_idx;
-    Status status;
     size_t prefetch_max_idx;
 
     MultiScanState(
         const std::shared_ptr<FileSystem>& _fs, const MultiScanArgs* _scan_opts,
         std::vector<CachableEntry<Block>>&& _pinned_data_blocks,
+        std::vector<std::string>&& _data_block_separators,
         std::vector<std::tuple<size_t, size_t>>&& _block_index_ranges_per_scan,
         UnorderedMap<size_t, size_t>&& _block_idx_to_readreq_idx,
         std::vector<AsyncReadState>&& _async_states, size_t _prefetch_max_idx)
         : fs(_fs),
           scan_opts(_scan_opts),
           pinned_data_blocks(std::move(_pinned_data_blocks)),
+          data_block_separators(std::move(_data_block_separators)),
           block_index_ranges_per_scan(std::move(_block_index_ranges_per_scan)),
           next_scan_idx(0),
           cur_data_block_idx(0),
           async_states(std::move(_async_states)),
           block_idx_to_readreq_idx(std::move(_block_idx_to_readreq_idx)),
-          status(Status::OK()),
-          prefetch_max_idx(_prefetch_max_idx) {
-      status.PermitUncheckedError();
-    }
+          prefetch_max_idx(_prefetch_max_idx) {}
 
     ~MultiScanState();
   };
 
+  Status multi_scan_status_;
   std::unique_ptr<MultiScanState> multi_scan_;
   // *** END MultiScan related APIs and states ***
 
@@ -599,14 +633,10 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
 
   // *** BEGIN APIs relevant to multiscan ***
 
-  // Returns true iff we should fallback to regular scan.
-  bool SeekMultiScan(const Slice* target);
+  void SeekMultiScan(const Slice* target);
 
   void FindBlockForwardInMultiScan();
 
-  // Unpins blocks from the immediately previous scan range.
-  void UnpinPreviousScanBlocks(size_t current_scan_idx);
-
   void PrepareReadAsyncCallBack(FSReadRequest& req, void* cb_arg) {
     // Record status, result and sanity check offset from `req`.
     AsyncReadState* async_state = static_cast<AsyncReadState*>(cb_arg);
@@ -627,15 +657,33 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
     }
   }
 
-  Status MultiScanLoadDataBlock(size_t idx) {
+  void MultiScanSeekTargetFromBlock(const Slice* seek_target, size_t block_idx);
+  void MultiScanUnexpectedSeekTarget(const Slice* seek_target,
+                                     const Slice* user_seek_target,
+                                     size_t block_idx);
+
+  // Return true, if there is an error, or end of file
+  bool MultiScanLoadDataBlock(size_t idx) {
     if (idx >= multi_scan_->prefetch_max_idx) {
-      return Status::PrefetchLimitReached();
+      // TODO: Fix the max_prefetch_size support for multiple files.
+      // The goal is to limit the memory usage, prefetch could be done
+      // incrementally.
+      if (multi_scan_->scan_opts->max_prefetch_size == 0) {
+        // If max_prefetch_size is not set, treat this as end of file.
+        ResetDataIter();
+        assert(!is_out_of_bound_);
+        assert(!Valid());
+      } else {
+        // If max_prefetch_size is set, treat this as error.
+        multi_scan_status_ = Status::PrefetchLimitReached();
+      }
+      return true;
     }
 
     if (!multi_scan_->async_states.empty()) {
-      Status s = PollForBlock(idx);
-      if (!s.ok()) {
-        return s;
+      multi_scan_status_ = PollForBlock(idx);
+      if (!multi_scan_status_.ok()) {
+        return true;
       }
     }
     // This block should have been initialized
@@ -646,7 +694,7 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
     table_->NewDataBlockIterator<DataBlockIter>(
         read_options_, multi_scan_->pinned_data_blocks[idx], &block_iter_,
         Status::OK());
-    return Status::OK();
+    return false;
   }
 
   // After PollForBlock(idx), the async request that contains
@@ -664,15 +712,13 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
                                      const Slice& buffer_data,
                                      CachableEntry<Block>& pinned_block_entry);
 
-  // Helper functions for Prepare():
-  bool ValidateScanOptions(const MultiScanArgs* multiscan_opts);
-
-  bool CollectBlockHandles(
+  Status CollectBlockHandles(
       const std::vector<ScanOptions>& scan_opts,
       std::vector<BlockHandle>* scan_block_handles,
-      std::vector<std::tuple<size_t, size_t>>* block_index_ranges_per_scan);
+      std::vector<std::tuple<size_t, size_t>>* block_index_ranges_per_scan,
+      std::vector<std::string>* data_block_boundary_keys);
 
-  bool FilterAndPinCachedBlocks(
+  Status FilterAndPinCachedBlocks(
       const std::vector<BlockHandle>& scan_block_handles,
       const MultiScanArgs* multiscan_opts,
       std::vector<size_t>* block_indices_to_read,
@@ -687,7 +733,7 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
       UnorderedMap<size_t, size_t>* block_idx_to_readreq_idx,
       std::vector<std::vector<size_t>>* coalesced_block_indices);
 
-  bool ExecuteIO(
+  Status ExecuteIO(
       const std::vector<BlockHandle>& scan_block_handles,
       const MultiScanArgs* multiscan_opts,
       const std::vector<std::vector<size_t>>& coalesced_block_indices,

table/block_based/block_based_table_reader.cc

@@ -1369,15 +1369,20 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks(
       if (s.ok()) {
         assert(!rep_->udi_block.IsEmpty());
 
-        std::unique_ptr<UserDefinedIndexReader> udi_reader =
-            table_options.user_defined_index_factory->NewReader(
-                rep_->udi_block.GetValue()->data);
-        if (udi_reader) {
-          index_reader = std::make_unique<UserDefinedIndexReaderWrapper>(
-              udi_name, std::move(index_reader), std::move(udi_reader));
-        } else {
-          s = Status::Corruption("Failed to create UDI reader for " + udi_name +
-                                 " in file " + rep_->file->file_name());
+        std::unique_ptr<UserDefinedIndexReader> udi_reader;
+        UserDefinedIndexOption udi_option;
+        udi_option.comparator = rep_->internal_comparator.user_comparator();
+        s = table_options.user_defined_index_factory->NewReader(
+            udi_option, rep_->udi_block.GetValue()->data, udi_reader);
+        if (s.ok()) {
+          if (udi_reader) {
+            index_reader = std::make_unique<UserDefinedIndexReaderWrapper>(
+                udi_name, std::move(index_reader), std::move(udi_reader));
+          } else {
+            s = Status::Corruption("Failed to create UDI reader for " +
+                                   udi_name + " in file " +
+                                   rep_->file->file_name());
+          }
         }
       }
     }

table/block_based/block_based_table_reader_test.cc

@@ -1173,61 +1173,6 @@ TEST_P(BlockBasedTableReaderTest, MultiScanPrepare) {
       } else {
         ASSERT_EQ(read_count_before + 3, read_count_after);
       }
-
-      // 4. Check cases when Seek key does not match start key in ScanOptions
-      iter.reset(table->NewIterator(
-          read_opts, options_.prefix_extractor.get(), /*arena=*/nullptr,
-          /*skip_filters=*/false, TableReaderCaller::kUncategorized));
-      scan_options = MultiScanArgs(comparator_);
-      scan_options.use_async_io = use_async_io;
-      scan_options.insert(ExtractUserKey(kv[10 * kEntriesPerBlock].first),
-                          ExtractUserKey(kv[20 * kEntriesPerBlock].first));
-      scan_options.insert(ExtractUserKey(kv[30 * kEntriesPerBlock].first),
-                          ExtractUserKey(kv[40 * kEntriesPerBlock].first));
-      iter->Prepare(&scan_options);
-      // Match start key
-      iter->Seek(kv[10 * kEntriesPerBlock].first);
-      for (size_t i = 10 * kEntriesPerBlock; i < 20 * kEntriesPerBlock; ++i) {
-        ASSERT_TRUE(iter->Valid());
-        ASSERT_EQ(iter->key().ToString(), kv[i].first);
-        iter->Next();
-      }
-      ASSERT_OK(iter->status());
-
-      // Does not match start key of the second ScanOptions.
-      iter->Seek(kv[50 * kEntriesPerBlock + 1].first);
-      for (size_t i = 50 * kEntriesPerBlock + 1; i < 100 * kEntriesPerBlock;
-           ++i) {
-        ASSERT_TRUE(iter->Valid());
-        ASSERT_EQ(iter->key().ToString(), kv[i].first);
-        iter->Next();
-      }
-      ASSERT_FALSE(iter->Valid());
-      ASSERT_OK(iter->status());
-
-      iter.reset(table->NewIterator(
-          read_opts, options_.prefix_extractor.get(), /*arena=*/nullptr,
-          /*skip_filters=*/false, TableReaderCaller::kUncategorized));
-      scan_options = MultiScanArgs(comparator_);
-      scan_options.use_async_io = use_async_io;
-      scan_options.insert(ExtractUserKey(kv[10 * kEntriesPerBlock].first));
-      scan_options.insert(ExtractUserKey(kv[11 * kEntriesPerBlock].first));
-      iter->Prepare(&scan_options);
-      // Does not match the first ScanOptions.
-      iter->SeekToFirst();
-      for (size_t i = 0; i < kEntriesPerBlock; ++i) {
-        ASSERT_TRUE(iter->Valid());
-        ASSERT_EQ(iter->key().ToString(), kv[i].first);
-        iter->Next();
-      }
-      ASSERT_OK(iter->status());
-      iter->Seek(kv[10 * kEntriesPerBlock].first);
-      for (size_t i = 10 * kEntriesPerBlock; i < 12 * kEntriesPerBlock; ++i) {
-        ASSERT_TRUE(iter->Valid());
-        ASSERT_EQ(iter->key().ToString(), kv[i].first);
-        iter->Next();
-      }
-      ASSERT_OK(iter->status());
     }
   }
 }
@@ -1509,17 +1454,6 @@ TEST_P(BlockBasedTableReaderTest, MultiScanUnpinPreviousBlocks) {
   }
 }
 
-// Param 1: compression type
-// Param 2: whether to use direct reads
-// Param 3: Block Based Table Index type, partitioned filters are also enabled
-//          when index type is kTwoLevelIndexSearch
-// Param 4: BBTO no_block_cache option
-// Param 5: test mode for the user-defined timestamp feature
-// Param 6: number of parallel compression threads
-// Param 7: CompressionOptions.max_dict_bytes and
-//          CompressionOptions.max_dict_buffer_bytes. This enable/disables
-//          compression dictionary.
-// Param 8: test mode to specify the pattern for generating key / value pairs.
 INSTANTIATE_TEST_CASE_P(
     BlockBasedTableReaderTest, BlockBasedTableReaderTest,
     ::testing::Combine(

table/block_based/user_defined_index_wrapper.h

@@ -289,7 +289,13 @@ class UserDefinedIndexReaderWrapper : public BlockBasedTable::IndexReader {
     }
     std::unique_ptr<UserDefinedIndexIterator> udi_iter =
         udi_reader_->NewIterator(read_options);
-    return new UserDefinedIndexIteratorWrapper(std::move(udi_iter));
+    if (udi_iter) {
+      InternalIteratorBase<IndexValue>* wrap_iter =
+          new UserDefinedIndexIteratorWrapper(std::move(udi_iter));
+      return wrap_iter;
+    }
+    return NewErrorInternalIterator<IndexValue>(
+        Status::NotFound("COuld not create UDI iterator"));
   }
 
   virtual Status CacheDependencies(

table/sst_file_dumper.cc

@@ -231,7 +231,9 @@ Status SstFileDumper::DumpTable(const std::string& out_filename) {
 }
 
 Status SstFileDumper::CalculateCompressedTableSize(
-    const TableBuilderOptions& tb_options, TableProperties* props) {
+    const TableBuilderOptions& tb_options, TableProperties* props,
+    std::chrono::microseconds* write_time,
+    std::chrono::microseconds* read_time) {
   std::unique_ptr<Env> env(NewMemEnv(options_.env));
   std::unique_ptr<WritableFileWriter> dest_writer;
   Status s =
@@ -240,6 +242,8 @@ Status SstFileDumper::CalculateCompressedTableSize(
   if (!s.ok()) {
     return s;
   }
+  std::chrono::steady_clock::time_point start =
+      std::chrono::steady_clock::now();
   std::unique_ptr<TableBuilder> table_builder{
       tb_options.moptions.table_factory->NewTableBuilder(tb_options,
                                                          dest_writer.get())};
@@ -253,17 +257,69 @@ Status SstFileDumper::CalculateCompressedTableSize(
   if (!s.ok()) {
     return s;
   }
+  iter.reset();
   s = table_builder->Finish();
+  *write_time = std::chrono::duration_cast<std::chrono::microseconds>(
+      std::chrono::steady_clock::now() - start);
   if (!s.ok()) {
     return s;
   }
+  s = dest_writer->Close({});
+  if (!s.ok()) {
+    return s;
+  }
+  dest_writer.reset();
   *props = table_builder->GetTableProperties();
+  start = std::chrono::steady_clock::now();
+  TableReaderOptions reader_options(ioptions_, moptions_.prefix_extractor,
+                                    moptions_.compression_manager.get(),
+                                    soptions_, internal_comparator_,
+                                    0 /* block_protection_bytes_per_key */);
+  std::unique_ptr<RandomAccessFileReader> file_reader;
+  s = RandomAccessFileReader::Create(env->GetFileSystem(), testFileName,
+                                     soptions_, &file_reader, /*dbg=*/nullptr);
+  if (!s.ok()) {
+    return s;
+  }
+  std::unique_ptr<TableReader> table_reader;
+  s = tb_options.moptions.table_factory->NewTableReader(
+      reader_options, std::move(file_reader), table_builder->FileSize(),
+      &table_reader);
+  if (!s.ok()) {
+    return s;
+  }
+  iter.reset(table_reader->NewIterator(
+      read_options_, moptions_.prefix_extractor.get(), /*arena=*/nullptr,
+      /*skip_filters=*/false, TableReaderCaller::kSSTDumpTool));
+  for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+  }
+  s = iter->status();
+  if (!s.ok()) {
+    return s;
+  }
+  iter.reset();
+  table_reader.reset();
+  file_reader.reset();
+  *read_time = std::chrono::duration_cast<std::chrono::microseconds>(
+      std::chrono::steady_clock::now() - start);
   return env->DeleteFile(testFileName);
 }
 
 Status SstFileDumper::ShowAllCompressionSizes(
     const std::vector<CompressionType>& compression_types,
     int32_t compress_level_from, int32_t compress_level_to) {
+#ifndef NDEBUG
+  fprintf(stdout,
+          "WARNING: Assertions are enabled; benchmarks unnecessarily slow\n");
+#endif
+  BlockBasedTableOptions bbto;
+  if (options_.table_factory->IsInstanceOf(
+          TableFactory::kBlockBasedTableName())) {
+    bbto = *(static_cast_with_check<BlockBasedTableFactory>(
+                 options_.table_factory.get()))
+                ->GetOptions<BlockBasedTableOptions>();
+  }
+
   for (CompressionType ctype : compression_types) {
     std::string cname;
     if (!GetStringFromCompressionType(&cname, ctype).ok()) {
@@ -273,10 +329,12 @@ Status SstFileDumper::ShowAllCompressionSizes(
     if (options_.compression_manager
             ? options_.compression_manager->SupportsCompressionType(ctype)
             : CompressionTypeSupported(ctype)) {
-      fprintf(stdout, "Compression: %-24s\n", cname.c_str());
       CompressionOptions compress_opt = options_.compression_opts;
+      fprintf(stdout,
+              "Compression: %-24s Block Size: %" PRIu64 "  Threads: %u\n",
+              cname.c_str(), bbto.block_size, compress_opt.parallel_threads);
       for (int32_t j = compress_level_from; j <= compress_level_to; j++) {
-        fprintf(stdout, "Compression level: %d", j);
+        fprintf(stdout, "Cx level: %d", j);
         compress_opt.level = j;
         Status s = ShowCompressionSize(ctype, compress_opt);
         if (!s.ok()) {
@@ -320,27 +378,26 @@ Status SstFileDumper::ShowCompressionSize(
       TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
       column_family_name, unknown_level, kUnknownNewestKeyTime);
   TableProperties props;
-  std::chrono::steady_clock::time_point start =
-      std::chrono::steady_clock::now();
-  Status s = CalculateCompressedTableSize(tb_opts, &props);
+  std::chrono::microseconds write_time;
+  std::chrono::microseconds read_time;
+  Status s =
+      CalculateCompressedTableSize(tb_opts, &props, &write_time, &read_time);
   if (!s.ok()) {
     return s;
   }
 
   uint64_t num_data_blocks = props.num_data_blocks;
 
-  std::chrono::steady_clock::time_point end = std::chrono::steady_clock::now();
-  fprintf(stdout, " Comp size: %10" PRIu64, props.data_size);
-  fprintf(stdout, " Uncompressed: %10" PRIu64, props.uncompressed_data_size);
+  fprintf(stdout, " Cx size: %10" PRIu64, props.data_size);
+  fprintf(stdout, " Uncx size: %10" PRIu64, props.uncompressed_data_size);
   fprintf(stdout, " Ratio: %10s",
           std::to_string(static_cast<double>(props.uncompressed_data_size) /
                          static_cast<double>(props.data_size))
               .c_str());
-  fprintf(stdout, " Microsecs: %10s ",
-          std::to_string(
-              std::chrono::duration_cast<std::chrono::microseconds>(end - start)
-                  .count())
-              .c_str());
+  fprintf(stdout, " Write usec: %10s ",
+          std::to_string(write_time.count()).c_str());
+  fprintf(stdout, " Read usec: %10s ",
+          std::to_string(read_time.count()).c_str());
   const uint64_t compressed_blocks =
       opts.statistics->getAndResetTickerCount(NUMBER_BLOCK_COMPRESSED);
   const uint64_t not_compressed_blocks =
@@ -370,11 +427,11 @@ Status SstFileDumper::ShowCompressionSize(
                              : ((static_cast<double>(not_compressed_blocks) /
                                  static_cast<double>(num_data_blocks)) *
                                 100.0);
-  fprintf(stdout, " Comp count: %6" PRIu64 " (%5.1f%%)", compressed_blocks,
+  fprintf(stdout, " Cx count: %6" PRIu64 " (%5.1f%%)", compressed_blocks,
           compressed_pcnt);
-  fprintf(stdout, " Not compressed (ratio): %6" PRIu64 " (%5.1f%%)",
+  fprintf(stdout, " Not cx for ratio: %6" PRIu64 " (%5.1f%%)",
           ratio_not_compressed_blocks, ratio_not_compressed_pcnt);
-  fprintf(stdout, " Not compressed (abort): %6" PRIu64 " (%5.1f%%)\n",
+  fprintf(stdout, " Not cx otherwise: %6" PRIu64 " (%5.1f%%)\n",
           not_compressed_blocks, not_compressed_pcnt);
   return Status::OK();
 }

table/sst_file_dumper.h

@@ -59,7 +59,9 @@ class SstFileDumper {
                              FilePrefetchBuffer* prefetch_buffer);
 
   Status CalculateCompressedTableSize(const TableBuilderOptions& tb_options,
-                                      TableProperties* props);
+                                      TableProperties* props,
+                                      std::chrono::microseconds* write_time,
+                                      std::chrono::microseconds* read_time);
 
   Status SetTableOptionsByMagicNumber(uint64_t table_magic_number);
   Status SetOldTableOptions();

tools/db_crashtest.py

@@ -348,13 +348,13 @@ default_params = {
     "enable_custom_split_merge": lambda: random.choice([0, 1]),
     "adm_policy": lambda: random.choice([0, 1, 2, 3]),
     "last_level_temperature": lambda: random.choice(
-        ["kUnknown", "kHot", "kWarm", "kCold", "kIce"]
+        ["kUnknown", "kHot", "kWarm", "kCool", "kCold", "kIce"]
     ),
     "default_write_temperature": lambda: random.choice(
-        ["kUnknown", "kHot", "kWarm", "kCold", "kIce"]
+        ["kUnknown", "kHot", "kWarm", "kCool", "kCold", "kIce"]
     ),
     "default_temperature": lambda: random.choice(
-        ["kUnknown", "kHot", "kWarm", "kCold", "kIce"]
+        ["kUnknown", "kHot", "kWarm", "kCool", "kCold", "kIce"]
     ),
     # TODO(hx235): enable `enable_memtable_insert_with_hint_prefix_extractor`
     # after fixing the surfaced issue with delete range
@@ -367,7 +367,7 @@ default_params = {
     "memtable_veirfy_per_key_checksum_on_seek": lambda: random.choice([0] * 7 + [1]),
     "allow_unprepared_value": lambda: random.choice([0, 1]),
     # TODO(hx235): enable `track_and_verify_wals` after stabalizing the stress test
-    "track_and_verify_wals": lambda: random.choice([0]),    
+    "track_and_verify_wals": lambda: random.choice([0]),
     "remote_compaction_worker_threads": lambda: random.choice([0, 8]),
     # TODO(jaykorean): Change to lambda: random.choice([0, 1]) after addressing all remote compaction failures
     "remote_compaction_failure_fall_back_to_local": 1,
@@ -667,7 +667,7 @@ tiered_params = {
     # For FIFO compaction (ignored otherwise)
     "file_temperature_age_thresholds": lambda: random.choice(
         [
-            "{{temperature=kWarm;age=10}:{temperature=kCold;age=50}:{temperature=kIce;age=250}}",
+            "{{temperature=kWarm;age=10}:{temperature=kCool;age=30}:{temperature=kCold;age=100}:{temperature=kIce;age=300}}",
             "{{temperature=kWarm;age=30}:{temperature=kCold;age=300}}",
             "{{temperature=kCold;age=100}}",
         ]
@@ -1155,13 +1155,15 @@ def finalize_and_sanitize(src_params):
     # Continuous verification fails with secondaries inside NonBatchedOpsStressTest
     if dest_params.get("test_secondary") == 1:
         dest_params["continuous_verification_interval"] = 0
-    if (
-        dest_params.get("prefix_size", 0) > 0
-        or dest_params.get("read_fault_one_in", 0) > 0
-    ):
-        dest_params["use_multiscan"] = 0
     if dest_params.get("use_multiscan") == 1:
         dest_params["async_io"] = 0
+        dest_params["delpercent"] += dest_params["delrangepercent"]
+        dest_params["delrangepercent"] = 0
+        dest_params["prefix_size"] = -1
+        dest_params["iterpercent"] += dest_params["prefixpercent"]
+        dest_params["prefixpercent"] = 0
+        dest_params["read_fault_one_in"] = 0
+        dest_params["memtable_prefix_bloom_size_ratio"] = 0
     return dest_params
 
 

tools/sst_dump_tool.cc

@@ -181,7 +181,6 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
   bool list_meta_blocks = false;
   bool has_compression_level_from = false;
   bool has_compression_level_to = false;
-  bool has_specified_compression_types = false;
   std::string from_key;
   std::string to_key;
   std::string block_size_str;
@@ -258,7 +257,6 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
       std::string compression_types_csv = argv[i] + 20;
       std::istringstream iss(compression_types_csv);
       std::string compression_type;
-      has_specified_compression_types = true;
 
       while (std::getline(iss, compression_type, ',')) {
         auto iter =
@@ -392,12 +390,7 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
     }
   }
 
-  if (has_compression_level_from && has_compression_level_to) {
-    if (!has_specified_compression_types || compression_types.size() != 1) {
-      fprintf(stderr, "Specify one compression type.\n\n");
-      exit(1);
-    }
-  } else if (has_compression_level_from || has_compression_level_to) {
+  if (has_compression_level_from ^ has_compression_level_to) {
     fprintf(stderr,
             "Specify both --compression_level_from and "
             "--compression_level_to.\n\n");
@@ -536,14 +529,20 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
     }
 
     if (command == "recompress") {
-      fprintf(stdout, "Block Size: %zu  Threads: %u\n", block_size,
-              (unsigned)compression_parallel_threads);
-      // TODO: consider getting supported compressions from the compression
-      // manager
+      if (compression_types.empty()) {
+        if (options.compression_manager != nullptr) {
+          for (int c = 0; c < kDisableCompressionOption; ++c) {
+            if (options.compression_manager->SupportsCompressionType(
+                    static_cast<CompressionType>(c))) {
+              compression_types.emplace_back(static_cast<CompressionType>(c));
+            }
+          }
+        } else {
+          compression_types = GetSupportedCompressions();
+        }
+      }
       st = dumper.ShowAllCompressionSizes(
-          compression_types.empty() ? GetSupportedCompressions()
-                                    : compression_types,
-          compress_level_from, compress_level_to);
+          compression_types, compress_level_from, compress_level_to);
       if (!st.ok()) {
         fprintf(stderr, "Failed to recompress: %s\n", st.ToString().c_str());
         exit(1);

unreleased_history/behavior_changes/autohcc.md

@@ -1 +0,0 @@
-* The default provided block cache implementation is now HyperClockCache instead of LRUCache, when `block_cache` is nullptr (default) and `no_block_cache==false` (default). We recommend explicitly creating a HyperClockCache block cache based on memory budget and sharing it across all column families and even DB instances. This change could expose previously hidden memory or resource leaks.

unreleased_history/bug_fixes/compaction_cpu.md

@@ -1 +0,0 @@
-* Reported numbers for compaction and flush CPU usage now include time spent by parallel compression worker threads. This now means compaction/flush CPU usage could exceed the wall clock time.

unreleased_history/bug_fixes/fix_multi_level_fifo_double_picking_bug.md

@@ -1 +0,0 @@
-Fix a race condition in FIFO size-based compaction where concurrent threads could select the same non-L0 file, causing assertion failures in debug builds or "Cannot delete table file from LSM tree" errors in release builds.

unreleased_history/bug_fixes/udi_empty_scan_range_fix.md

@@ -1 +0,0 @@
-Fix a bug in RocksDB MultiScan with UDI when one of the scan ranges is determined to be empty by the UDI, which causes incorrect results.

unreleased_history/new_features/fail_if_no_udi_on_open.md

@@ -1 +0,0 @@
-Add the fail_if_no_udi_on_open flag in BlockBasedTableOption to control whether a missing user defined index block in a SST is a hard error or not.

unreleased_history/new_features/improve_data_integrity_check_on_seek.md

@@ -1 +0,0 @@
-A new flag memtable_veirfy_per_key_checksum_on_seek is added to AdvancedColumnFamilyOptions. When it is enabled, it will validate key checksum along the binary search path on skiplist based memtable during seek operation.

unreleased_history/new_features/multi-scan-async-io.md

@@ -1 +0,0 @@
-* Introduce option MultiScanArgs::use_async_io to enable asynchronous I/O during MultiScan, instead of waiting for I/O to be done in Prepare().

unreleased_history/new_features/multi-scan-max-prefetch.md

@@ -1 +0,0 @@
-* Add new option `MultiScanArgs::max_prefetch_size` that limits the memory usage of per file pinning of prefetched blocks.

unreleased_history/new_features/sst_dump_recompress.md

@@ -1 +0,0 @@
-* Improved `sst_dump` by allowing standalone file and directory arguments without `--file=`. Also added new options and better output for `sst_dump --command=recompress`. See `sst_dump --help`

unreleased_history/performance_improvements/external_sst_ingestion_seqno_optimization.md

@@ -1,2 +0,0 @@
-* Add a new table property "rocksdb.key.smallest.seqno" which records the smallest sequence number of all keys in file. It makes ingesting DB generated files faster by
-avoiding scanning the whole file to find the smallest sequence number.

unreleased_history/performance_improvements/improve_point_lock_manager_performance.md

@@ -1,2 +0,0 @@
-Add a new experimental PerKeyPointLockManager to improve efficiency under high lock contention. PointLockManager was not efficient when there is high write contention on same key, as it uses a single conditional variable per lock stripe. PerKeyPointLockManager uses per thread conditional variable supporting fifo order. Although this is an experimental feature. By default, it is disabled. A new boolean flag TransactionDBOptions::use_per_key_point_lock_mgr is added to optionally enable it. Search the flag in code for more info.
-Together, a new configuration TransactionOptions::deadlock_timeout_us is added, which allows the transaction to wait for a short period before perform deadlock detection. When the workload has low lock contention, the deadlock_timeout_us can be configured to be slightly higher than average transaction execution time, so that transaction would likely be able to take the lock before deadlock detection is performed when it is waiting for a lock. This allows transaction to reduce CPU cost on performing deadlock detection, which could be expensive in CPU time. When the workload has high lock contention, the deadlock_timeout_us can be configured to 0, so that transaction would perform deadlock detection immediately. By default the value is 0 to keep the behavior same as before.

unreleased_history/performance_improvements/parallel_compression.md

@@ -1 +0,0 @@
-* Majorly improved CPU efficiency and scalability of parallel compression (`CompressionOptions::parallel_threads` > 1), though this efficiency improvement makes parallel compression currently incompatible with UserDefinedIndex and with old setting of `decouple_partitioned_filters=false`. Parallel compression is now considered a production-ready feature. Maximum performance is available with `-DROCKSDB_USE_STD_SEMAPHORES` at compile time, but this is not currently recommended because of reported bugs in implementations of `std::counting_semaphore`/`binary_semaphore`.

unreleased_history/public_api_changes/MultiScanArgs_contructor_parameter_change.md

@@ -1 +0,0 @@
-MultiScanArgs used to have a default constructor with default parameter of BytewiseComparator. Now it always requires Comparator in its constructor.

unreleased_history/public_api_changes/autohcc.md

@@ -1 +0,0 @@
-* HyperClockCache with no `estimated_entry_charge` is now production-ready and is the preferred block cache implementation vs. LRUCache. Please consider updating your code to minimize the risk of hitting performance bottlenecks or anomalies from LRUCache. See cache.h for more detail.

unreleased_history/public_api_changes/cplusplus20.md

@@ -1 +0,0 @@
-* RocksDB now requires a C++20 compatible compiler (GCC >= 11, Clang >= 10, Visual Studio >= 2019), including for any code using RocksDB headers.

util/compression.cc

@@ -516,7 +516,7 @@ class BuiltinBZip2CompressorV2 : public CompressorWithSimpleDictBase {
   }
 };
 
-class BuiltinLZ4CompressorV2 : public CompressorWithSimpleDictBase {
+class BuiltinLZ4CompressorV2WithDict : public CompressorWithSimpleDictBase {
  public:
   using CompressorWithSimpleDictBase::CompressorWithSimpleDictBase;
 
@@ -527,8 +527,8 @@ class BuiltinLZ4CompressorV2 : public CompressorWithSimpleDictBase {
   }
 
   std::unique_ptr<Compressor> CloneForDict(std::string&& dict_data) override {
-    return std::make_unique<BuiltinLZ4CompressorV2>(opts_,
-                                                    std::move(dict_data));
+    return std::make_unique<BuiltinLZ4CompressorV2WithDict>(
+        opts_, std::move(dict_data));
   }
 
   ManagedWorkingArea ObtainWorkingArea() override {
@@ -611,6 +611,72 @@ class BuiltinLZ4CompressorV2 : public CompressorWithSimpleDictBase {
   }
 };
 
+class BuiltinLZ4CompressorV2NoDict : public BuiltinLZ4CompressorV2WithDict {
+ public:
+  BuiltinLZ4CompressorV2NoDict(const CompressionOptions& opts)
+      : BuiltinLZ4CompressorV2WithDict(opts, /*dict_data=*/{}) {}
+
+  ManagedWorkingArea ObtainWorkingArea() override {
+    // Using an LZ4_stream_t between compressions and resetting with
+    // LZ4_resetStream_fast is actually slower than using a fresh LZ4_stream_t
+    // each time, or not involving a stream at all. Similarly, using an extState
+    // does not seem to offer a performance boost, perhaps a small regression.
+    return {};
+  }
+
+  void ReleaseWorkingArea(WorkingArea* wa) override {
+    // Should not be called
+    (void)wa;
+    assert(wa == nullptr);
+  }
+
+  Status CompressBlock(Slice uncompressed_data, char* compressed_output,
+                       size_t* compressed_output_size,
+                       CompressionType* out_compression_type,
+                       ManagedWorkingArea* wa) override {
+#ifdef LZ4
+    (void)wa;
+    auto [alg_output, alg_max_output_size] = StartCompressBlockV2(
+        uncompressed_data, compressed_output, *compressed_output_size);
+    if (alg_max_output_size == 0) {
+      // Compression bypassed
+      *compressed_output_size = 0;
+      *out_compression_type = kNoCompression;
+      return Status::OK();
+    }
+    int acceleration;
+    if (opts_.level < 0) {
+      acceleration = -opts_.level;
+    } else {
+      acceleration = 1;
+    }
+    auto outlen =
+        LZ4_compress_fast(uncompressed_data.data(), alg_output,
+                          static_cast<int>(uncompressed_data.size()),
+                          static_cast<int>(alg_max_output_size), acceleration);
+    if (outlen > 0) {
+      // Compression kept/successful
+      size_t output_size = static_cast<size_t>(
+          outlen + /*header size*/ (alg_output - compressed_output));
+      assert(output_size <= *compressed_output_size);
+      *compressed_output_size = output_size;
+      *out_compression_type = kLZ4Compression;
+      return Status::OK();
+    }
+    // Compression rejected
+    *compressed_output_size = 1;
+#else
+    (void)uncompressed_data;
+    (void)compressed_output;
+    (void)wa;
+    // Compression bypassed (not supported)
+    *compressed_output_size = 0;
+#endif
+    *out_compression_type = kNoCompression;
+    return Status::OK();
+  }
+};
+
 class BuiltinLZ4HCCompressorV2 : public CompressorWithSimpleDictBase {
  public:
   using CompressorWithSimpleDictBase::CompressorWithSimpleDictBase;
@@ -1508,7 +1574,7 @@ class BuiltinCompressionManagerV2 : public CompressionManager {
       case kBZip2Compression:
         return std::make_unique<BuiltinBZip2CompressorV2>(opts);
       case kLZ4Compression:
-        return std::make_unique<BuiltinLZ4CompressorV2>(opts);
+        return std::make_unique<BuiltinLZ4CompressorV2NoDict>(opts);
       case kLZ4HCCompression:
         return std::make_unique<BuiltinLZ4HCCompressorV2>(opts);
       case kXpressCompression:

utilities/fault_injection_env.cc

@@ -464,17 +464,6 @@ Status FaultInjectionTestEnv::LinkFile(const std::string& s,
   return ret;
 }
 
-Status FaultInjectionTestEnv::SyncFile(const std::string& fname,
-                                       const EnvOptions& env_options,
-                                       bool use_fsync) {
-  // Call the default implement of SyncFile API in Env, so that it would call
-  // other FileSystem API at FaultInjectionTestEnv layer for failure injection.
-  // Otherwise, the default behavior is WrapperEnv::SyncFile, which forward the
-  // call to the underlying FileSystem, instead of the ones in
-  // FaultInjectionTestEnv.
-  return Env::SyncFile(fname, env_options, use_fsync);
-}
-
 void FaultInjectionTestEnv::WritableFileClosed(const FileState& state) {
   MutexLock l(&mutex_);
   if (open_managed_files_.find(state.filename_) != open_managed_files_.end()) {

utilities/fault_injection_env.h

@@ -177,9 +177,6 @@ class FaultInjectionTestEnv : public EnvWrapper {
 
   Status LinkFile(const std::string& s, const std::string& t) override;
 
-  Status SyncFile(const std::string& fname, const EnvOptions& env_options,
-                  bool use_fsync) override;
-
 // Undef to eliminate clash on Windows
 #undef GetFreeSpace
   Status GetFreeSpace(const std::string& path, uint64_t* disk_free) override {

utilities/fault_injection_fs.cc

@@ -1200,17 +1200,6 @@ IOStatus FaultInjectionTestFS::LinkFile(const std::string& s,
   }
   return io_s;
 }
-IOStatus FaultInjectionTestFS::SyncFile(const std::string& fname,
-                                        const FileOptions& file_options,
-                                        const IOOptions& io_options,
-                                        bool use_fsync, IODebugContext* dbg) {
-  // Call the default implement of SyncFile API in FileSystem, so that it would
-  // call other FileSystem API at FaultInjectionTestFS layer for failure
-  // injection. Otherwise, the default behavior is calling target()->SyncFile,
-  // which forward the call to the underlying FileSystem, instead of the ones in
-  // FaultInjectionTestFS.
-  return FileSystem::SyncFile(fname, file_options, io_options, use_fsync, dbg);
-}
 
 IOStatus FaultInjectionTestFS::NumFileLinks(const std::string& fname,
                                             const IOOptions& options,

utilities/fault_injection_fs.h

@@ -312,10 +312,6 @@ class FaultInjectionTestFS : public FileSystemWrapper {
   IOStatus LinkFile(const std::string& src, const std::string& target,
                     const IOOptions& options, IODebugContext* dbg) override;
 
-  IOStatus SyncFile(const std::string& fname, const FileOptions& file_options,
-                    const IOOptions& io_options, bool use_fsync,
-                    IODebugContext* dbg) override;
-
   IOStatus NumFileLinks(const std::string& fname, const IOOptions& options,
                         uint64_t* count, IODebugContext* dbg) override;