Add a unit test for PR 11049

Bump version to 7.9.2
Fix async prefetch heap use after free (#11049 )
2026-07-07 14:47:40 +08:00 · 2022-12-22 09:30:39 -08:00 · 2022-12-21 09:51:53 -08:00 · 2022-12-21 09:46:30 -08:00 · 2022-12-08 19:54:29 -08:00 · 2022-12-08 19:25:26 -08:00
43 changed files with 726 additions and 726 deletions
@@ -76,7 +76,7 @@ commands:
          name: "Test RocksDB"
          shell: powershell.exe
          command: |
-            build_tools\run_ci_db_test.ps1 -SuiteRun arena_test,db_basic_test,db_test,db_test2,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test -Concurrency 16
+            build_tools\run_ci_db_test.ps1 -SuiteRun db_basic_test,db_test,db_test2,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test -Concurrency 16
  pre-steps-macos:
      steps:
        - pre-steps
@@ -796,7 +796,6 @@ set(SOURCES
        options/options.cc
        options/options_helper.cc
        options/options_parser.cc
-        port/mmap.cc
        port/stack_trace.cc
        table/adaptive/adaptive_table_factory.cc
        table/block_based/binary_search_index_reader.cc
@@ -1,10 +1,21 @@
 # Rocksdb Change Log
+## 7.9.2 (12/21/2022)
+### Bug Fixes
+* Fixed a heap use after free bug in async scan prefetching when the scan thread and another thread try to read and load the same seek block into cache.
+
+## 7.9.1 (12/8/2022)
+### Bug Fixes
+* Fixed a regression in iterator where range tombstones after `iterate_upper_bound` is processed.
+* Fixed a memory leak in MultiGet with async_io read option, caused by IO errors during table file open
+
+### Behavior changes
+* Make best-efforts recovery verify SST unique ID before Version construction (#10962)
+
 ## 7.9.0 (11/21/2022)
 ### Performance Improvements
 * Fixed an iterator performance regression for delete range users when scanning through a consecutive sequence of range tombstones (#10877).

 ### Bug Fixes
-* Fix FIFO compaction causing corruption of overlapping seqnos in L0 files due to ingesting files of overlapping seqnos with memtable's under `CompactionOptionsFIFO::allow_compaction=true` or `CompactionOptionsFIFO::age_for_warm>0` or `CompactRange()/CompactFiles()` is used. Before the fix, `force_consistency_checks=true` may catch the corruption before it's exposed to readers, in which case writes returning `Status::Corruption` would be expected.
 * Fix memory corruption error in scans if async_io is enabled. Memory corruption happened if there is IOError while reading the data leading to empty buffer and other buffer already in progress of async read goes again for reading.
 * Fix failed memtable flush retry bug that could cause wrongly ordered updates, which would surface to writers as `Status::Corruption` in case of `force_consistency_checks=true` (default). It affects use cases that enable both parallel flush (`max_background_flushes > 1` or `max_background_jobs >= 8`) and non-default memtable count (`max_write_buffer_number > 2`).
 * Fixed an issue where the `READ_NUM_MERGE_OPERANDS` ticker was not updated when the base key-value or tombstone was read from an SST file.
@@ -163,7 +163,6 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
        "options/options.cc",
        "options/options_helper.cc",
        "options/options_parser.cc",
-        "port/mmap.cc",
        "port/port_posix.cc",
        "port/stack_trace.cc",
        "port/win/env_default.cc",
@@ -503,7 +502,6 @@ cpp_library_wrapper(name="rocksdb_whole_archive_lib", srcs=[
        "options/options.cc",
        "options/options_helper.cc",
        "options/options_parser.cc",
-        "port/mmap.cc",
        "port/port_posix.cc",
        "port/stack_trace.cc",
        "port/win/env_default.cc",
@@ -1212,14 +1212,11 @@ Compaction* ColumnFamilyData::CompactRange(
    const InternalKey* begin, const InternalKey* end,
    InternalKey** compaction_end, bool* conflict,
    uint64_t max_file_num_to_ignore, const std::string& trim_ts) {
-  SequenceNumber earliest_mem_seqno =
-      std::min(mem_->GetEarliestSequenceNumber(),
-               imm_.current()->GetEarliestSequenceNumber(false));
  auto* result = compaction_picker_->CompactRange(
      GetName(), mutable_cf_options, mutable_db_options,
      current_->storage_info(), input_level, output_level,
      compact_range_options, begin, end, compaction_end, conflict,
-      max_file_num_to_ignore, trim_ts, earliest_mem_seqno);
+      max_file_num_to_ignore, trim_ts);
  if (result != nullptr) {
    result->SetInputVersion(current_);
  }
@@ -32,7 +32,7 @@ bool FindIntraL0Compaction(const std::vector<FileMetaData*>& level_files,
                           uint64_t max_compact_bytes_per_del_file,
                           uint64_t max_compaction_bytes,
                           CompactionInputFiles* comp_inputs,
-                           const SequenceNumber earliest_mem_seqno) {
+                           SequenceNumber earliest_mem_seqno) {
  // Do not pick ingested file when there is at least one memtable not flushed
  // which of seqno is overlap with the sst.
  TEST_SYNC_POINT("FindIntraL0Compaction");
@@ -613,8 +613,7 @@ Compaction* CompactionPicker::CompactRange(
    int input_level, int output_level,
    const CompactRangeOptions& compact_range_options, const InternalKey* begin,
    const InternalKey* end, InternalKey** compaction_end, bool* manual_conflict,
-    uint64_t max_file_num_to_ignore, const std::string& trim_ts,
-    const SequenceNumber /*earliest_mem_seqno*/) {
+    uint64_t max_file_num_to_ignore, const std::string& trim_ts) {
  // CompactionPickerFIFO has its own implementation of compact range
  assert(ioptions_.compaction_style != kCompactionStyleFIFO);

@@ -919,8 +918,7 @@ bool HaveOverlappingKeyRanges(const Comparator* c, const SstFileMetaData& a,

 Status CompactionPicker::SanitizeCompactionInputFilesForAllLevels(
    std::unordered_set<uint64_t>* input_files,
-    const ColumnFamilyMetaData& cf_meta, const int output_level,
-    const SequenceNumber earliest_mem_seqno) const {
+    const ColumnFamilyMetaData& cf_meta, const int output_level) const {
  auto& levels = cf_meta.levels;
  auto comparator = icmp_->user_comparator();

@@ -997,13 +995,6 @@ Status CompactionPicker::SanitizeCompactionInputFilesForAllLevels(
                               current_files[f].name +
                               " is currently being compacted.");
      }
-      if (output_level == 0 &&
-          current_files[f].largest_seqno > earliest_mem_seqno) {
-        return Status::Aborted(
-            "Necessary compaction input file " + current_files[f].name +
-            " has overlapping seqnos with earliest memtable seqnos.");
-      }
-
      input_files->insert(TableFileNameToNumber(current_files[f].name));
    }

@@ -1060,14 +1051,12 @@ Status CompactionPicker::SanitizeCompactionInputFilesForAllLevels(
        "A running compaction is writing to the same output level in an "
        "overlapping key range");
  }
-
  return Status::OK();
 }

 Status CompactionPicker::SanitizeCompactionInputFiles(
    std::unordered_set<uint64_t>* input_files,
-    const ColumnFamilyMetaData& cf_meta, const int output_level,
-    const SequenceNumber earliest_mem_seqno) const {
+    const ColumnFamilyMetaData& cf_meta, const int output_level) const {
  assert(static_cast<int>(cf_meta.levels.size()) - 1 ==
         cf_meta.levels[cf_meta.levels.size() - 1].level);
  if (output_level >= static_cast<int>(cf_meta.levels.size())) {
@@ -1093,8 +1082,8 @@ Status CompactionPicker::SanitizeCompactionInputFiles(
        "A compaction must contain at least one file.");
  }

-  Status s = SanitizeCompactionInputFilesForAllLevels(
-      input_files, cf_meta, output_level, earliest_mem_seqno);
+  Status s = SanitizeCompactionInputFilesForAllLevels(input_files, cf_meta,
+                                                      output_level);

  if (!s.ok()) {
    return s;
@@ -51,24 +51,15 @@ class CompactionPicker {
  virtual ~CompactionPicker();

  // Pick level and inputs for a new compaction.
-  //
-  // `earliest_mem_seqno` is the earliest seqno of unflushed memtables.
-  // It is needed to compare with compaction input SST files' largest seqnos
-  // in order to exclude those of seqnos potentially overlap with memtables'
-  // seqnos when doing compaction to L0. This will avoid creating a SST files in
-  // L0 newer than a unflushed memtable. Such SST file can exist in the first
-  // place when it's ingested or resulted from compaction involving files
-  // ingested.
-  //
  // Returns nullptr if there is no compaction to be done.
  // Otherwise returns a pointer to a heap-allocated object that
  // describes the compaction.  Caller should delete the result.
  virtual Compaction* PickCompaction(
      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
      const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-      LogBuffer* log_buffer, const SequenceNumber earliest_mem_seqno) = 0;
+      LogBuffer* log_buffer,
+      SequenceNumber earliest_memtable_seqno = kMaxSequenceNumber) = 0;

-  // `earliest_mem_seqno`: see PickCompaction() API
  // Return a compaction object for compacting the range [begin,end] in
  // the specified level.  Returns nullptr if there is nothing in that
  // level that overlaps the specified range.  Caller should delete
@@ -87,8 +78,7 @@ class CompactionPicker {
      const CompactRangeOptions& compact_range_options,
      const InternalKey* begin, const InternalKey* end,
      InternalKey** compaction_end, bool* manual_conflict,
-      uint64_t max_file_num_to_ignore, const std::string& trim_ts,
-      const SequenceNumber earliest_mem_seqno);
+      uint64_t max_file_num_to_ignore, const std::string& trim_ts);

  // The maximum allowed output level.  Default value is NumberLevels() - 1.
  virtual int MaxOutputLevel() const { return NumberLevels() - 1; }
@@ -101,18 +91,10 @@ class CompactionPicker {
 // files.  If it's not possible to conver an invalid input_files
 // into a valid one by adding more files, the function will return a
 // non-ok status with specific reason.
-//
-// Cases of returning non-ok status include but not limited to:
-// - When output_level == 0 and input_files contains sst files
-// of largest seqno greater than `earliest_mem_seqno`. This will
-// avoid creating a SST files in L0 newer than a unflushed memtable.
-// Such SST file can exist in the first place when it's ingested or
-// resulted from compaction involving files ingested.
 #ifndef ROCKSDB_LITE
-  Status SanitizeCompactionInputFiles(
-      std::unordered_set<uint64_t>* input_files,
-      const ColumnFamilyMetaData& cf_meta, const int output_level,
-      const SequenceNumber earliest_mem_seqno) const;
+  Status SanitizeCompactionInputFiles(std::unordered_set<uint64_t>* input_files,
+                                      const ColumnFamilyMetaData& cf_meta,
+                                      const int output_level) const;
 #endif  // ROCKSDB_LITE

  // Free up the files that participated in a compaction
@@ -248,8 +230,7 @@ class CompactionPicker {
 #ifndef ROCKSDB_LITE
  virtual Status SanitizeCompactionInputFilesForAllLevels(
      std::unordered_set<uint64_t>* input_files,
-      const ColumnFamilyMetaData& cf_meta, const int output_level,
-      const SequenceNumber earliest_mem_seqno) const;
+      const ColumnFamilyMetaData& cf_meta, const int output_level) const;
 #endif  // ROCKSDB_LITE

  // Keeps track of all compactions that are running on Level0.
@@ -279,22 +260,23 @@ class NullCompactionPicker : public CompactionPicker {
      const MutableCFOptions& /*mutable_cf_options*/,
      const MutableDBOptions& /*mutable_db_options*/,
      VersionStorageInfo* /*vstorage*/, LogBuffer* /* log_buffer */,
-      const SequenceNumber /* earliest_mem_seqno */) override {
+      SequenceNumber /* earliest_memtable_seqno */) override {
    return nullptr;
  }

  // Always return "nullptr"
-  Compaction* CompactRange(
-      const std::string& /*cf_name*/,
-      const MutableCFOptions& /*mutable_cf_options*/,
-      const MutableDBOptions& /*mutable_db_options*/,
-      VersionStorageInfo* /*vstorage*/, int /*input_level*/,
-      int /*output_level*/,
-      const CompactRangeOptions& /*compact_range_options*/,
-      const InternalKey* /*begin*/, const InternalKey* /*end*/,
-      InternalKey** /*compaction_end*/, bool* /*manual_conflict*/,
-      uint64_t /*max_file_num_to_ignore*/, const std::string& /*trim_ts*/,
-      const SequenceNumber /* earliest_mem_seqno */) override {
+  Compaction* CompactRange(const std::string& /*cf_name*/,
+                           const MutableCFOptions& /*mutable_cf_options*/,
+                           const MutableDBOptions& /*mutable_db_options*/,
+                           VersionStorageInfo* /*vstorage*/,
+                           int /*input_level*/, int /*output_level*/,
+                           const CompactRangeOptions& /*compact_range_options*/,
+                           const InternalKey* /*begin*/,
+                           const InternalKey* /*end*/,
+                           InternalKey** /*compaction_end*/,
+                           bool* /*manual_conflict*/,
+                           uint64_t /*max_file_num_to_ignore*/,
+                           const std::string& /*trim_ts*/) override {
    return nullptr;
  }

@@ -321,14 +303,12 @@ class NullCompactionPicker : public CompactionPicker {
 //                                        initialized with corresponding input
 //                                        files. Cannot be nullptr.
 //
-// @param earliest_mem_seqno              See PickCompaction() API
 // @return                                true iff compaction was found.
-bool FindIntraL0Compaction(const std::vector<FileMetaData*>& level_files,
-                           size_t min_files_to_compact,
-                           uint64_t max_compact_bytes_per_del_file,
-                           uint64_t max_compaction_bytes,
-                           CompactionInputFiles* comp_inputs,
-                           const SequenceNumber earliest_mem_seqno);
+bool FindIntraL0Compaction(
+    const std::vector<FileMetaData*>& level_files, size_t min_files_to_compact,
+    uint64_t max_compact_bytes_per_del_file, uint64_t max_compaction_bytes,
+    CompactionInputFiles* comp_inputs,
+    SequenceNumber earliest_mem_seqno = kMaxSequenceNumber);

 CompressionType GetCompressionType(const VersionStorageInfo* vstorage,
                                   const MutableCFOptions& mutable_cf_options,
@@ -139,7 +139,7 @@ Compaction* FIFOCompactionPicker::PickTTLCompaction(
 Compaction* FIFOCompactionPicker::PickSizeCompaction(
    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
    const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-    LogBuffer* log_buffer, const SequenceNumber earliest_mem_seqno) {
+    LogBuffer* log_buffer) {
  // compute the total size and identify the last non-empty level
  int last_level = 0;
  uint64_t total_size = 0;
@@ -176,8 +176,7 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
                  .level0_file_num_compaction_trigger /* min_files_to_compact */
              ,
              max_compact_bytes_per_del_file,
-              mutable_cf_options.max_compaction_bytes, &comp_inputs,
-              earliest_mem_seqno)) {
+              mutable_cf_options.max_compaction_bytes, &comp_inputs)) {
        Compaction* c = new Compaction(
            vstorage, ioptions_, mutable_cf_options, mutable_db_options,
            {comp_inputs}, 0, 16 * 1024 * 1024 /* output file size limit */,
@@ -276,8 +275,7 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
 Compaction* FIFOCompactionPicker::PickCompactionToWarm(
    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
    const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-    LogBuffer* log_buffer, const SequenceNumber earliest_mem_seqno) {
-  TEST_SYNC_POINT("PickCompactionToWarm");
+    LogBuffer* log_buffer) {
  if (mutable_cf_options.compaction_options_fifo.age_for_warm == 0) {
    return nullptr;
  }
@@ -301,8 +299,6 @@ Compaction* FIFOCompactionPicker::PickCompactionToWarm(
                     cf_name.c_str(), status.ToString().c_str());
    return nullptr;
  }
-  TEST_SYNC_POINT_CALLBACK("PickCompactionToWarm::BeforeGetCurrentTime",
-                           &_current_time);
  const uint64_t current_time = static_cast<uint64_t>(_current_time);

  if (!level0_compactions_in_progress_.empty()) {
@@ -349,8 +345,7 @@ Compaction* FIFOCompactionPicker::PickCompactionToWarm(
        // for warm tier.
        break;
      }
-      if (prev_file != nullptr &&
-          prev_file->fd.largest_seqno <= earliest_mem_seqno) {
+      if (prev_file != nullptr) {
        compaction_size += prev_file->fd.GetFileSize();
        if (compaction_size > mutable_cf_options.max_compaction_bytes) {
          break;
@@ -394,7 +389,7 @@ Compaction* FIFOCompactionPicker::PickCompactionToWarm(
 Compaction* FIFOCompactionPicker::PickCompaction(
    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
    const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-    LogBuffer* log_buffer, const SequenceNumber earliest_mem_seqno) {
+    LogBuffer* log_buffer, SequenceNumber /*earliest_memtable_seqno*/) {
  Compaction* c = nullptr;
  if (mutable_cf_options.ttl > 0) {
    c = PickTTLCompaction(cf_name, mutable_cf_options, mutable_db_options,
@@ -402,11 +397,11 @@ Compaction* FIFOCompactionPicker::PickCompaction(
  }
  if (c == nullptr) {
    c = PickSizeCompaction(cf_name, mutable_cf_options, mutable_db_options,
-                           vstorage, log_buffer, earliest_mem_seqno);
+                           vstorage, log_buffer);
  }
  if (c == nullptr) {
    c = PickCompactionToWarm(cf_name, mutable_cf_options, mutable_db_options,
-                             vstorage, log_buffer, earliest_mem_seqno);
+                             vstorage, log_buffer);
  }
  RegisterCompaction(c);
  return c;
@@ -419,8 +414,7 @@ Compaction* FIFOCompactionPicker::CompactRange(
    const CompactRangeOptions& /*compact_range_options*/,
    const InternalKey* /*begin*/, const InternalKey* /*end*/,
    InternalKey** compaction_end, bool* /*manual_conflict*/,
-    uint64_t /*max_file_num_to_ignore*/, const std::string& /*trim_ts*/,
-    const SequenceNumber earliest_mem_seqno) {
+    uint64_t /*max_file_num_to_ignore*/, const std::string& /*trim_ts*/) {
 #ifdef NDEBUG
  (void)input_level;
  (void)output_level;
@@ -429,9 +423,8 @@ Compaction* FIFOCompactionPicker::CompactRange(
  assert(output_level == 0);
  *compaction_end = nullptr;
  LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL, ioptions_.logger);
-  Compaction* c =
-      PickCompaction(cf_name, mutable_cf_options, mutable_db_options, vstorage,
-                     &log_buffer, earliest_mem_seqno);
+  Compaction* c = PickCompaction(cf_name, mutable_cf_options,
+                                 mutable_db_options, vstorage, &log_buffer);
  log_buffer.FlushBufferToLog();
  return c;
 }
@@ -22,12 +22,9 @@ class FIFOCompactionPicker : public CompactionPicker {
  virtual Compaction* PickCompaction(
      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
      const MutableDBOptions& mutable_db_options, VersionStorageInfo* version,
-      LogBuffer* log_buffer, const SequenceNumber earliest_mem_seqno) override;
+      LogBuffer* log_buffer,
+      SequenceNumber earliest_memtable_seqno = kMaxSequenceNumber) override;

-  // `earliest_mem_seqno`: see PickCompaction() API for more. In FIFO's
-  // implementation of CompactRange(), different from others, we will not return
-  // `nullptr` right away when intput files of compaction to L0 has seqnos
-  // potentially overlapping with memtable's but exlucde those files.
  virtual Compaction* CompactRange(
      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
      const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
@@ -35,8 +32,7 @@ class FIFOCompactionPicker : public CompactionPicker {
      const CompactRangeOptions& compact_range_options,
      const InternalKey* begin, const InternalKey* end,
      InternalKey** compaction_end, bool* manual_conflict,
-      uint64_t max_file_num_to_ignore, const std::string& trim_ts,
-      const SequenceNumber earliest_mem_seqno) override;
+      uint64_t max_file_num_to_ignore, const std::string& trim_ts) override;

  // The maximum allowed output level.  Always returns 0.
  virtual int MaxOutputLevel() const override { return 0; }
@@ -55,15 +51,13 @@ class FIFOCompactionPicker : public CompactionPicker {
                                 const MutableCFOptions& mutable_cf_options,
                                 const MutableDBOptions& mutable_db_options,
                                 VersionStorageInfo* version,
-                                 LogBuffer* log_buffer,
-                                 SequenceNumber earliest_mem_seqno);
+                                 LogBuffer* log_buffer);

  Compaction* PickCompactionToWarm(const std::string& cf_name,
                                   const MutableCFOptions& mutable_cf_options,
                                   const MutableDBOptions& mutable_db_options,
                                   VersionStorageInfo* version,
-                                   LogBuffer* log_buffer,
-                                   const SequenceNumber earliest_mem_seqno);
+                                   LogBuffer* log_buffer);
 };
 }  // namespace ROCKSDB_NAMESPACE
 #endif  // !ROCKSDB_LITE
@@ -50,7 +50,7 @@ class LevelCompactionBuilder {
 public:
  LevelCompactionBuilder(const std::string& cf_name,
                         VersionStorageInfo* vstorage,
-                         const SequenceNumber earliest_mem_seqno,
+                         SequenceNumber earliest_mem_seqno,
                         CompactionPicker* compaction_picker,
                         LogBuffer* log_buffer,
                         const MutableCFOptions& mutable_cf_options,
@@ -122,7 +122,7 @@ class LevelCompactionBuilder {

  const std::string& cf_name_;
  VersionStorageInfo* vstorage_;
-  const SequenceNumber earliest_mem_seqno_;
+  SequenceNumber earliest_mem_seqno_;
  CompactionPicker* compaction_picker_;
  LogBuffer* log_buffer_;
  int start_level_ = -1;
@@ -832,7 +832,7 @@ bool LevelCompactionBuilder::PickIntraL0Compaction() {
 Compaction* LevelCompactionPicker::PickCompaction(
    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
    const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-    LogBuffer* log_buffer, const SequenceNumber earliest_mem_seqno) {
+    LogBuffer* log_buffer, SequenceNumber earliest_mem_seqno) {
  LevelCompactionBuilder builder(cf_name, vstorage, earliest_mem_seqno, this,
                                 log_buffer, mutable_cf_options, ioptions_,
                                 mutable_db_options);
@@ -23,7 +23,8 @@ class LevelCompactionPicker : public CompactionPicker {
  virtual Compaction* PickCompaction(
      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
      const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-      LogBuffer* log_buffer, const SequenceNumber earliest_mem_seqno) override;
+      LogBuffer* log_buffer,
+      SequenceNumber earliest_memtable_seqno = kMaxSequenceNumber) override;

  virtual bool NeedsCompaction(
      const VersionStorageInfo* vstorage) const override;
@@ -218,7 +218,7 @@ TEST_F(CompactionPickerTest, Empty) {
  UpdateVersionStorageInfo();
  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() == nullptr);
 }

@@ -230,7 +230,7 @@ TEST_F(CompactionPickerTest, Single) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() == nullptr);
 }

@@ -244,7 +244,7 @@ TEST_F(CompactionPickerTest, Level0Trigger) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_files(0));
  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());
@@ -258,7 +258,7 @@ TEST_F(CompactionPickerTest, Level1Trigger) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  ASSERT_EQ(66U, compaction->input(0, 0)->fd.GetNumber());
@@ -277,7 +277,7 @@ TEST_F(CompactionPickerTest, Level1Trigger2) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  ASSERT_EQ(2U, compaction->num_input_files(1));
@@ -309,7 +309,7 @@ TEST_F(CompactionPickerTest, LevelMaxScore) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  ASSERT_EQ(7U, compaction->input(0, 0)->fd.GetNumber());
@@ -357,7 +357,7 @@ TEST_F(CompactionPickerTest, Level0TriggerDynamic) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_files(0));
  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());
@@ -382,7 +382,7 @@ TEST_F(CompactionPickerTest, Level0TriggerDynamic2) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_files(0));
  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());
@@ -408,7 +408,7 @@ TEST_F(CompactionPickerTest, Level0TriggerDynamic3) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_files(0));
  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());
@@ -438,7 +438,7 @@ TEST_F(CompactionPickerTest, Level0TriggerDynamic4) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_files(0));
  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());
@@ -471,7 +471,7 @@ TEST_F(CompactionPickerTest, LevelTriggerDynamic4) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  ASSERT_EQ(5U, compaction->input(0, 0)->fd.GetNumber());
@@ -528,7 +528,7 @@ TEST_F(CompactionPickerTest, CompactionUniversalIngestBehindReservedLevel) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  // output level should be the one above the bottom-most
  ASSERT_EQ(1, compaction->output_level());
@@ -563,7 +563,7 @@ TEST_F(CompactionPickerTest, CannotTrivialMoveUniversal) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_TRUE(!compaction->is_trivial_move());
 }
@@ -590,7 +590,7 @@ TEST_F(CompactionPickerTest, AllowsTrivialMoveUniversal) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_TRUE(compaction->is_trivial_move());
 }
@@ -619,7 +619,7 @@ TEST_F(CompactionPickerTest, UniversalPeriodicCompaction1) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_TRUE(compaction);
  ASSERT_EQ(4, compaction->output_level());
@@ -650,7 +650,7 @@ TEST_F(CompactionPickerTest, UniversalPeriodicCompaction2) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_FALSE(compaction);
 }
@@ -677,7 +677,7 @@ TEST_F(CompactionPickerTest, UniversalPeriodicCompaction3) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_FALSE(compaction);
 }
@@ -708,7 +708,7 @@ TEST_F(CompactionPickerTest, UniversalPeriodicCompaction4) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(!compaction ||
              compaction->start_level() != compaction->output_level());
 }
@@ -729,7 +729,7 @@ TEST_F(CompactionPickerTest, UniversalPeriodicCompaction5) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction);
  ASSERT_EQ(0, compaction->start_level());
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -754,7 +754,7 @@ TEST_F(CompactionPickerTest, UniversalPeriodicCompaction6) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction);
  ASSERT_EQ(4, compaction->start_level());
  ASSERT_EQ(2U, compaction->num_input_files(0));
@@ -792,7 +792,7 @@ TEST_F(CompactionPickerTest, UniversalIncrementalSpace1) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction);
  ASSERT_EQ(4, compaction->output_level());
  ASSERT_EQ(3, compaction->start_level());
@@ -834,7 +834,7 @@ TEST_F(CompactionPickerTest, UniversalIncrementalSpace2) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction);
  ASSERT_EQ(4, compaction->output_level());
  ASSERT_EQ(2, compaction->start_level());
@@ -876,7 +876,7 @@ TEST_F(CompactionPickerTest, UniversalIncrementalSpace3) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction);
  ASSERT_EQ(4, compaction->output_level());
  ASSERT_EQ(2, compaction->start_level());
@@ -924,7 +924,7 @@ TEST_F(CompactionPickerTest, UniversalIncrementalSpace4) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction);
  ASSERT_EQ(4, compaction->output_level());
  ASSERT_EQ(3, compaction->start_level());
@@ -968,7 +968,7 @@ TEST_F(CompactionPickerTest, UniversalIncrementalSpace5) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction);
  ASSERT_EQ(4, compaction->output_level());
  ASSERT_EQ(3, compaction->start_level());
@@ -1035,7 +1035,7 @@ TEST_F(CompactionPickerTest, FIFOToWarm1) {
  ASSERT_EQ(fifo_compaction_picker.NeedsCompaction(vstorage_.get()), true);
  std::unique_ptr<Compaction> compaction(fifo_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  ASSERT_EQ(3U, compaction->input(0, 0)->fd.GetNumber());
@@ -1073,7 +1073,7 @@ TEST_F(CompactionPickerTest, FIFOToWarm2) {
  ASSERT_EQ(fifo_compaction_picker.NeedsCompaction(vstorage_.get()), true);
  std::unique_ptr<Compaction> compaction(fifo_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_files(0));
  ASSERT_EQ(2U, compaction->input(0, 0)->fd.GetNumber());
@@ -1114,7 +1114,7 @@ TEST_F(CompactionPickerTest, FIFOToWarmMaxSize) {
  ASSERT_EQ(fifo_compaction_picker.NeedsCompaction(vstorage_.get()), true);
  std::unique_ptr<Compaction> compaction(fifo_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_files(0));
  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());
@@ -1155,7 +1155,7 @@ TEST_F(CompactionPickerTest, FIFOToWarmWithExistingWarm) {
  ASSERT_EQ(fifo_compaction_picker.NeedsCompaction(vstorage_.get()), true);
  std::unique_ptr<Compaction> compaction(fifo_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_files(0));
  ASSERT_EQ(2U, compaction->input(0, 0)->fd.GetNumber());
@@ -1197,7 +1197,7 @@ TEST_F(CompactionPickerTest, FIFOToWarmWithOngoing) {
  ASSERT_EQ(fifo_compaction_picker.NeedsCompaction(vstorage_.get()), true);
  std::unique_ptr<Compaction> compaction(fifo_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  // Stop if a file is being compacted
  ASSERT_TRUE(compaction.get() == nullptr);
 }
@@ -1236,7 +1236,7 @@ TEST_F(CompactionPickerTest, FIFOToWarmWithHotBetweenWarms) {
  ASSERT_EQ(fifo_compaction_picker.NeedsCompaction(vstorage_.get()), true);
  std::unique_ptr<Compaction> compaction(fifo_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  // Stop if a file is being compacted
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -1267,7 +1267,7 @@ TEST_F(CompactionPickerTest, CompactionPriMinOverlapping1) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  // Pick file 8 because it overlaps with 0 files on level 3.
@@ -1300,7 +1300,7 @@ TEST_F(CompactionPickerTest, CompactionPriMinOverlapping2) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  // Picking file 7 because overlapping ratio is the biggest.
@@ -1328,7 +1328,7 @@ TEST_F(CompactionPickerTest, CompactionPriMinOverlapping3) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  // Picking file 8 because overlapping ratio is the biggest.
@@ -1359,7 +1359,7 @@ TEST_F(CompactionPickerTest, CompactionPriMinOverlapping4) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  // Picking file 8 because overlapping ratio is the biggest.
@@ -1395,7 +1395,7 @@ TEST_F(CompactionPickerTest, CompactionPriRoundRobin) {
    std::unique_ptr<Compaction> compaction(
        local_level_compaction_picker.PickCompaction(
            cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-            &log_buffer_, kMaxSequenceNumber));
+            &log_buffer_));
    ASSERT_TRUE(compaction.get() != nullptr);
    // Since the max bytes for level 2 is 120M, picking one file to compact
    // makes the post-compaction level size less than 120M, there is exactly one
@@ -1435,7 +1435,7 @@ TEST_F(CompactionPickerTest, CompactionPriMultipleFilesRoundRobin1) {
  std::unique_ptr<Compaction> compaction(
      local_level_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);

  // The maximum compaction bytes is very large in this case so we can igore its
@@ -1478,7 +1478,7 @@ TEST_F(CompactionPickerTest, CompactionPriMultipleFilesRoundRobin2) {
  std::unique_ptr<Compaction> compaction(
      local_level_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);

  // The maximum compaction bytes is only 2500 bytes now. Even though we are
@@ -1522,7 +1522,7 @@ TEST_F(CompactionPickerTest, CompactionPriMultipleFilesRoundRobin3) {
  std::unique_ptr<Compaction> compaction(
      local_level_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);

  // Cannot pick more files since we reach the last file in level 2
@@ -1581,7 +1581,7 @@ TEST_F(CompactionPickerTest, CompactionPriMinOverlappingManyFiles) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  // Picking file 8 because overlapping ratio is the biggest.
@@ -1609,7 +1609,7 @@ TEST_F(CompactionPickerTest, ParentIndexResetBug) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
 }

 // This test checks ExpandWhileOverlapping() by having overlapping user keys
@@ -1627,7 +1627,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_levels());
  ASSERT_EQ(2U, compaction->num_input_files(0));
@@ -1647,7 +1647,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys2) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(2U, compaction->num_input_files(0));
@@ -1675,7 +1675,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys3) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(5U, compaction->num_input_files(0));
@@ -1706,7 +1706,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys4) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -1730,7 +1730,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys5) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() == nullptr);
 }

@@ -1752,7 +1752,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys6) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -1773,7 +1773,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys7) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_GE(1U, compaction->num_input_files(0));
@@ -1802,7 +1802,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys8) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(3U, compaction->num_input_files(0));
@@ -1835,7 +1835,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys9) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(5U, compaction->num_input_files(0));
@@ -1876,7 +1876,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys10) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -1915,7 +1915,7 @@ TEST_F(CompactionPickerTest, OverlappingUserKeys11) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -2013,7 +2013,7 @@ TEST_F(CompactionPickerTest, NotScheduleL1IfL0WithHigherPri1) {
  ASSERT_EQ(1, vstorage_->CompactionScoreLevel(1));
  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() == nullptr);
 }

@@ -2044,7 +2044,7 @@ TEST_F(CompactionPickerTest, NotScheduleL1IfL0WithHigherPri2) {
  ASSERT_EQ(1, vstorage_->CompactionScoreLevel(1));
  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
 }

@@ -2078,7 +2078,7 @@ TEST_F(CompactionPickerTest, NotScheduleL1IfL0WithHigherPri3) {
  ASSERT_EQ(0, vstorage_->CompactionScoreLevel(1));
  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
 }

@@ -2374,7 +2374,7 @@ TEST_F(CompactionPickerTest, MaxCompactionBytesHit) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -2400,7 +2400,7 @@ TEST_F(CompactionPickerTest, MaxCompactionBytesNotHit) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(3U, compaction->num_input_files(0));
@@ -2430,7 +2430,7 @@ TEST_F(CompactionPickerTest, IsTrivialMoveOn) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_TRUE(compaction->IsTrivialMove());
 }
@@ -2455,7 +2455,7 @@ TEST_F(CompactionPickerTest, L0TrivialMove1) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1, compaction->num_input_levels());
  ASSERT_EQ(2, compaction->num_input_files(0));
@@ -2484,7 +2484,7 @@ TEST_F(CompactionPickerTest, L0TrivialMoveOneFile) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1, compaction->num_input_levels());
  ASSERT_EQ(1, compaction->num_input_files(0));
@@ -2510,7 +2510,7 @@ TEST_F(CompactionPickerTest, L0TrivialMoveWholeL0) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1, compaction->num_input_levels());
  ASSERT_EQ(4, compaction->num_input_files(0));
@@ -2541,7 +2541,7 @@ TEST_F(CompactionPickerTest, IsTrivialMoveOffSstPartitioned) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  // No trivial move, because partitioning is applied
  ASSERT_TRUE(!compaction->IsTrivialMove());
@@ -2564,7 +2564,7 @@ TEST_F(CompactionPickerTest, IsTrivialMoveOff) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_FALSE(compaction->IsTrivialMove());
 }
@@ -2593,7 +2593,7 @@ TEST_F(CompactionPickerTest, TrivialMoveMultipleFiles1) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_TRUE(compaction->IsTrivialMove());
  ASSERT_EQ(1, compaction->num_input_levels());
@@ -2627,7 +2627,7 @@ TEST_F(CompactionPickerTest, TrivialMoveMultipleFiles2) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_TRUE(compaction->IsTrivialMove());
  ASSERT_EQ(1, compaction->num_input_levels());
@@ -2660,7 +2660,7 @@ TEST_F(CompactionPickerTest, TrivialMoveMultipleFiles3) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_TRUE(compaction->IsTrivialMove());
  ASSERT_EQ(1, compaction->num_input_levels());
@@ -2686,7 +2686,7 @@ TEST_F(CompactionPickerTest, TrivialMoveMultipleFiles4) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_TRUE(compaction->IsTrivialMove());
  ASSERT_EQ(1, compaction->num_input_levels());
@@ -2716,7 +2716,7 @@ TEST_F(CompactionPickerTest, TrivialMoveMultipleFiles5) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_TRUE(compaction->IsTrivialMove());
  ASSERT_EQ(1, compaction->num_input_levels());
@@ -2750,7 +2750,7 @@ TEST_F(CompactionPickerTest, TrivialMoveMultipleFiles6) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_TRUE(compaction->IsTrivialMove());
  ASSERT_EQ(1, compaction->num_input_levels());
@@ -2785,7 +2785,7 @@ TEST_F(CompactionPickerTest, CacheNextCompactionIndex) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -2795,7 +2795,7 @@ TEST_F(CompactionPickerTest, CacheNextCompactionIndex) {

  compaction.reset(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(2U, compaction->num_input_levels());
  ASSERT_EQ(1U, compaction->num_input_files(0));
@@ -2805,7 +2805,7 @@ TEST_F(CompactionPickerTest, CacheNextCompactionIndex) {

  compaction.reset(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() == nullptr);
  ASSERT_EQ(4, vstorage_->NextCompactionIndex(1 /* level */));
 }
@@ -2831,7 +2831,7 @@ TEST_F(CompactionPickerTest, IntraL0MaxCompactionBytesNotHit) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_levels());
  ASSERT_EQ(5U, compaction->num_input_files(0));
@@ -2862,7 +2862,7 @@ TEST_F(CompactionPickerTest, IntraL0MaxCompactionBytesHit) {

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_, kMaxSequenceNumber));
+      &log_buffer_));
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_levels());
  ASSERT_EQ(4U, compaction->num_input_files(0));
@@ -2928,7 +2928,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedCompactionFullOverlap) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_TRUE(compaction);
  // Validate that its a compaction to reduce sorted runs
@@ -2946,7 +2946,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedCompactionFullOverlap) {
  std::unique_ptr<Compaction> compaction2(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_FALSE(compaction2);
 }

@@ -2971,7 +2971,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedCompactionFullOverlap2) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_TRUE(compaction);
  // Validate that its a delete triggered compaction
@@ -2990,7 +2990,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedCompactionFullOverlap2) {
  std::unique_ptr<Compaction> compaction2(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_FALSE(compaction2);
 }

@@ -3031,7 +3031,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedCompactionStartOutputOverlap) {
    std::unique_ptr<Compaction> compaction(
        universal_compaction_picker.PickCompaction(
            cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-            &log_buffer_, kMaxSequenceNumber));
+            &log_buffer_));

    ASSERT_TRUE(compaction);
    // Validate that its a delete triggered compaction
@@ -3062,7 +3062,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedCompactionStartOutputOverlap) {
    std::unique_ptr<Compaction> compaction2(
        universal_compaction_picker.PickCompaction(
            cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-            &log_buffer_, kMaxSequenceNumber));
+            &log_buffer_));
    ASSERT_FALSE(compaction2);
    DeleteVersionStorage();
  }
@@ -3088,7 +3088,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedL0NoOverlap) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_TRUE(compaction);
  // Validate that its a delete triggered compaction
@@ -3125,7 +3125,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedL0WithOverlap) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_TRUE(compaction);
  // Validate that its a delete triggered compaction
@@ -3159,7 +3159,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedL0Overlap2) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  ASSERT_TRUE(compaction);
  // Validate that its a delete triggered compaction
@@ -3180,7 +3180,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedL0Overlap2) {
  std::unique_ptr<Compaction> compaction2(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  ASSERT_TRUE(compaction2);
  ASSERT_EQ(3U, compaction->num_input_files(0));
  ASSERT_TRUE(file_map_[1].first->being_compacted);
@@ -3215,7 +3215,7 @@ TEST_F(CompactionPickerTest, UniversalMarkedManualCompaction) {
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
          ColumnFamilyData::kCompactAllLevels, 6, CompactRangeOptions(),
          nullptr, nullptr, &manual_end, &manual_conflict,
-          std::numeric_limits<uint64_t>::max(), "", kMaxSequenceNumber));
+          std::numeric_limits<uint64_t>::max(), ""));

  ASSERT_TRUE(compaction);

@@ -3256,7 +3256,7 @@ TEST_F(CompactionPickerTest, UniversalSizeAmpTierCompactionNonLastLevel) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  // Make sure it's a size amp compaction and includes all files
  ASSERT_EQ(compaction->compaction_reason(),
@@ -3292,7 +3292,7 @@ TEST_F(CompactionPickerTest, UniversalSizeRatioTierCompactionLastLevel) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  // Internally, size amp compaction is evaluated before size ratio compaction.
  // Here to make sure it's size ratio compaction instead of size amp
@@ -3329,7 +3329,7 @@ TEST_F(CompactionPickerTest, UniversalSizeAmpTierCompactionNotSuport) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  // size amp compaction is still triggered even preclude_last_level is set
  ASSERT_EQ(compaction->compaction_reason(),
@@ -3363,7 +3363,7 @@ TEST_F(CompactionPickerTest, UniversalSizeAmpTierCompactionLastLevel) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));

  // It's a Size Amp compaction, but doesn't include the last level file and
  // output to the penultimate level.
@@ -3471,7 +3471,7 @@ TEST_F(CompactionPickerU64TsTest, CannotTrivialMoveUniversal) {
  std::unique_ptr<Compaction> compaction(
      universal_compaction_picker.PickCompaction(
          cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-          &log_buffer_, kMaxSequenceNumber));
+          &log_buffer_));
  assert(compaction);
  ASSERT_TRUE(!compaction->is_trivial_move());
 }
@@ -293,7 +293,7 @@ bool UniversalCompactionPicker::NeedsCompaction(
 Compaction* UniversalCompactionPicker::PickCompaction(
    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
    const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-    LogBuffer* log_buffer, const SequenceNumber /* earliest_mem_seqno */) {
+    LogBuffer* log_buffer, SequenceNumber /* earliest_memtable_seqno */) {
  UniversalCompactionBuilder builder(ioptions_, icmp_, cf_name,
                                     mutable_cf_options, mutable_db_options,
                                     vstorage, this, log_buffer);
@@ -21,7 +21,8 @@ class UniversalCompactionPicker : public CompactionPicker {
  virtual Compaction* PickCompaction(
      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
      const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-      LogBuffer* log_buffer, const SequenceNumber earliest_mem_seqno) override;
+      LogBuffer* log_buffer,
+      SequenceNumber earliest_memtable_seqno = kMaxSequenceNumber) override;
  virtual int MaxOutputLevel() const override { return NumberLevels() - 1; }

  virtual bool NeedsCompaction(
@@ -2158,11 +2158,11 @@ class DBMultiGetAsyncIOTest : public DBBasicTest,
      : DBBasicTest(), statistics_(ROCKSDB_NAMESPACE::CreateDBStatistics()) {
    BlockBasedTableOptions bbto;
    bbto.filter_policy.reset(NewBloomFilterPolicy(10));
-    Options options = CurrentOptions();
-    options.disable_auto_compactions = true;
-    options.statistics = statistics_;
-    options.table_factory.reset(NewBlockBasedTableFactory(bbto));
-    Reopen(options);
+    options_ = CurrentOptions();
+    options_.disable_auto_compactions = true;
+    options_.statistics = statistics_;
+    options_.table_factory.reset(NewBlockBasedTableFactory(bbto));
+    Reopen(options_);
    int num_keys = 0;

    // Put all keys in the bottommost level, and overwrite some keys
@@ -2227,8 +2227,12 @@ class DBMultiGetAsyncIOTest : public DBBasicTest,

  const std::shared_ptr<Statistics>& statistics() { return statistics_; }

+ protected:
+  void ReopenDB() { Reopen(options_); }
+
 private:
  std::shared_ptr<Statistics> statistics_;
+  Options options_;
 };

 TEST_P(DBMultiGetAsyncIOTest, GetFromL0) {
@@ -2305,6 +2309,69 @@ TEST_P(DBMultiGetAsyncIOTest, GetFromL1) {
  ASSERT_EQ(statistics()->getTickerCount(MULTIGET_COROUTINE_COUNT), 3);
 }

+TEST_P(DBMultiGetAsyncIOTest, GetFromL1Error) {
+  std::vector<std::string> key_strs;
+  std::vector<Slice> keys;
+  std::vector<PinnableSlice> values;
+  std::vector<Status> statuses;
+
+  key_strs.push_back(Key(33));
+  key_strs.push_back(Key(54));
+  key_strs.push_back(Key(102));
+  keys.push_back(key_strs[0]);
+  keys.push_back(key_strs[1]);
+  keys.push_back(key_strs[2]);
+  values.resize(keys.size());
+  statuses.resize(keys.size());
+
+  SyncPoint::GetInstance()->SetCallBack(
+      "TableCache::GetTableReader:BeforeOpenFile", [&](void* status) {
+        static int count = 0;
+        count++;
+        // Fail the last table reader open, which is the 6th SST file
+        // since 3 overlapping L0 files + 3 L1 files containing the keys
+        if (count == 6) {
+          Status* s = static_cast<Status*>(status);
+          *s = Status::IOError();
+        }
+      });
+  // DB open will create table readers unless we reduce the table cache
+  // capacity.
+  // SanitizeOptions will set max_open_files to minimum of 20. Table cache
+  // is allocated with max_open_files - 10 as capacity. So override
+  // max_open_files to 11 so table cache capacity will become 1. This will
+  // prevent file open during DB open and force the file to be opened
+  // during MultiGet
+  SyncPoint::GetInstance()->SetCallBack(
+      "SanitizeOptions::AfterChangeMaxOpenFiles", [&](void* arg) {
+        int* max_open_files = (int*)arg;
+        *max_open_files = 11;
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ReopenDB();
+
+  ReadOptions ro;
+  ro.async_io = true;
+  ro.optimize_multiget_for_io = GetParam();
+  dbfull()->MultiGet(ro, dbfull()->DefaultColumnFamily(), keys.size(),
+                     keys.data(), values.data(), statuses.data());
+  SyncPoint::GetInstance()->DisableProcessing();
+  ASSERT_EQ(values.size(), 3);
+  ASSERT_EQ(statuses[0], Status::OK());
+  ASSERT_EQ(statuses[1], Status::OK());
+  ASSERT_EQ(statuses[2], Status::IOError());
+
+  HistogramData multiget_io_batch_size;
+
+  statistics()->histogramData(MULTIGET_IO_BATCH_SIZE, &multiget_io_batch_size);
+
+  // A batch of 3 async IOs is expected, one for each overlapping file in L1
+  ASSERT_EQ(multiget_io_batch_size.count, 1);
+  ASSERT_EQ(multiget_io_batch_size.max, 2);
+  ASSERT_EQ(statistics()->getTickerCount(MULTIGET_COROUTINE_COUNT), 2);
+}
+
 TEST_P(DBMultiGetAsyncIOTest, LastKeyInFile) {
  std::vector<std::string> key_strs;
  std::vector<Slice> keys;
@@ -186,13 +186,6 @@ class RoundRobinSubcompactionsAgainstResources
  int max_compaction_limits_;
 };

-class DBCompactionTestFIFOCheckConsistencyWithParam
-    : public DBCompactionTest,
-      public testing::WithParamInterface<std::string> {
- public:
-  DBCompactionTestFIFOCheckConsistencyWithParam() : DBCompactionTest() {}
-};
-
 namespace {
 class FlushedFileCollector : public EventListener {
 public:
@@ -6444,163 +6437,6 @@ TEST_P(DBCompactionTestWithParam,
  }
 }

-INSTANTIATE_TEST_CASE_P(DBCompactionTestFIFOCheckConsistencyWithParam,
-                        DBCompactionTestFIFOCheckConsistencyWithParam,
-                        ::testing::Values("FindIntraL0Compaction",
-                                          "PickCompactionToWarm",
-                                          "CompactRange", "CompactFile"));
-
-TEST_P(DBCompactionTestFIFOCheckConsistencyWithParam,
-       FlushAfterIntraL0CompactionWithIngestedFile) {
-  Options options = CurrentOptions();
-  options.create_if_missing = true;
-  options.compression = kNoCompression;
-
-  options.force_consistency_checks = true;
-  options.compaction_style = kCompactionStyleFIFO;
-  options.max_open_files = -1;
-  options.num_levels = 1;
-  options.level0_file_num_compaction_trigger = 3;
-
-  CompactionOptionsFIFO fifo_options;
-  const std::string compaction_path_to_test = GetParam();
-  if (compaction_path_to_test == "FindIntraL0Compaction") {
-    fifo_options.allow_compaction = true;
-    fifo_options.age_for_warm = 0;
-  } else if (compaction_path_to_test == "PickCompactionToWarm") {
-    fifo_options.allow_compaction = false;
-    fifo_options.age_for_warm = 2;
-  } else if (compaction_path_to_test == "CompactRange") {
-    // FIFOCompactionPicker::CompactRange() implementes
-    // on top of regular compaction paths. Here we choose
-    // to trigger FIFOCompactionPicker::PickCompactionToWarm()
-    // for simplicity
-    fifo_options.allow_compaction = false;
-    fifo_options.age_for_warm = 2;
-    options.disable_auto_compactions = true;
-  } else if (compaction_path_to_test == "CompactFile") {
-    fifo_options.allow_compaction = false;
-    fifo_options.age_for_warm = 0;
-    options.disable_auto_compactions = true;
-  } else {
-    assert(false);
-  }
-  options.compaction_options_fifo = fifo_options;
-
-  DestroyAndReopen(options);
-
-  // To force assigning the global seqno to ingested file
-  // for our test purpose
-  const Snapshot* snapshot = db_->GetSnapshot();
-
-  std::atomic<bool> compaction_path_sync_point_called(false);
-  if (compaction_path_to_test == "FindIntraL0Compaction") {
-    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-        "FindIntraL0Compaction",
-        [&](void* /*arg*/) { compaction_path_sync_point_called.store(true); });
-  } else if (compaction_path_to_test == "PickCompactionToWarm" ||
-             compaction_path_to_test == "CompactRange") {
-    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-        "PickCompactionToWarm",
-        [&](void* /*arg*/) { compaction_path_sync_point_called.store(true); });
-    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-        "PickCompactionToWarm::BeforeGetCurrentTime",
-        [&fifo_options](void* current_time_arg) -> void {
-          // The unit test goes so quickly that there is almost no time
-          // elapsed after we ingest a file and before we check whether ingested
-          // files can compact to warm.
-          // Therefore we need this trick to simulate elapsed
-          // time in reality.
-          int64_t* current_time = (int64_t*)current_time_arg;
-          *current_time = *current_time + fifo_options.age_for_warm + 1;
-        });
-  } else if (compaction_path_to_test == "CompactFile") {
-    // Sync point is not needed in this case
-    compaction_path_sync_point_called.store(true);
-  }
-
-  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
-
-  // Create an existing SST file s0 of key range [key1,key4] and seqno range
-  // [1,2]
-  ASSERT_OK(Put("key1", "seq1"));
-  ASSERT_OK(Put("key4", "seq2"));
-  ASSERT_OK(Flush());
-  ASSERT_EQ(1, NumTableFilesAtLevel(0));
-
-  // Accumulate entries in a memtable m1 of key range [key1,key2] and seqno
-  // range [3,4] Noted that it contains a overlaped key with s0
-  ASSERT_OK(Put("key1", "seq3"));  // overlapped key
-  ASSERT_OK(Put("key2", "seq4"));
-
-  ASSERT_TRUE(compaction_path_to_test == "CompactFile" ||
-              !compaction_path_sync_point_called.load());
-
-  // Stop background compaction job to obtain accurate
-  // `NumTableFilesAtLevel(0)` after file ingestion
-  test::SleepingBackgroundTask sleeping_tasks;
-  if (!options.disable_auto_compactions) {
-    env_->SetBackgroundThreads(1, Env::LOW);
-    env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask, &sleeping_tasks,
-                   Env::Priority::LOW);
-    sleeping_tasks.WaitUntilSleeping();
-  }
-
-  // Ingested two SST files, s1 of key range [key5,key5] and seqno range [5,5]
-  // and s2 of key range [key6,key6] and seqno range [6,6]
-  IngestOneKeyValue(dbfull(), "key5", "seq5", options);
-  IngestOneKeyValue(dbfull(), "key6", "seq6", options);
-  // Up to now, L0 contains s0, s1, s2
-  ASSERT_EQ(3, NumTableFilesAtLevel(0));
-
-  // Resume background compaction job so that Intra level0 compaction can be
-  // triggered
-  if (!options.disable_auto_compactions) {
-    sleeping_tasks.WakeUp();
-    sleeping_tasks.WaitUntilDone();
-  }
-
-  if (compaction_path_to_test == "CompactRange") {
-    // `start` and `end` is carefully chosen so that compact range:
-    // (1) doesn't overlap with memtable therefore the memtable won't be flushed
-    // (2) should target at compacting s0 with s1 and s2
-    Slice start("key4"), end("key6");
-    ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &end));
-  } else if (compaction_path_to_test == "CompactFile") {
-    ColumnFamilyMetaData cf_meta_data;
-    db_->GetColumnFamilyMetaData(&cf_meta_data);
-    assert(cf_meta_data.levels[0].files.size() == 3);
-    std::vector<std::string> input_files;
-    for (const auto& file : cf_meta_data.levels[0].files) {
-      input_files.push_back(file.name);
-    }
-    Status s = db_->CompactFiles(CompactionOptions(), input_files, 0);
-    EXPECT_TRUE(s.IsAborted());
-    EXPECT_TRUE(s.ToString().find(
-                    "has overlapping seqnos with earliest memtable seqnos") !=
-                std::string::npos);
-  } else {
-    ASSERT_OK(dbfull()->TEST_WaitForCompact());
-  }
-  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
-
-  ASSERT_TRUE(compaction_path_to_test == "CompactFile" ||
-              compaction_path_sync_point_called.load());
-
-  // To verify compaction of s0, s1 and s2 (leading to new SST s4) didn't
-  // happen.
-  //
-  // Otherwise, when m1 flushes in the next step and become s3,
-  // we will have s3 of seqnos [3, 4], s4 of seqnos [1, 6], which is a
-  // corruption because s3 is older than s4 based on largest seqno while s2
-  // contains a value of Key(1) newer than the value of Key(1) contained in s4.
-  // And in this case, Flush() will return Status::Corruption() caught by
-  // `force_consistency_checks=1`
-  EXPECT_EQ(3, NumTableFilesAtLevel(0));
-  EXPECT_OK(Flush());
-  db_->ReleaseSnapshot(snapshot);
-}
-
 TEST_P(DBCompactionTestWithBottommostParam, SequenceKeysManualCompaction) {
  constexpr int kSstNum = 10;
  Options options = CurrentOptions();
@@ -1823,7 +1823,8 @@ InternalIterator* DBImpl::NewInternalIterator(
  MergeIteratorBuilder merge_iter_builder(
      &cfd->internal_comparator(), arena,
      !read_options.total_order_seek &&
-          super_version->mutable_cf_options.prefix_extractor != nullptr);
+          super_version->mutable_cf_options.prefix_extractor != nullptr,
+      read_options.iterate_upper_bound);
  // Collect iterator for mutable memtable
  auto mem_iter = super_version->mem->NewIterator(read_options, arena);
  Status s;
@@ -1344,18 +1344,8 @@ Status DBImpl::CompactFilesImpl(
    }
  }

-  SequenceNumber earliest_mem_seqno = kMaxSequenceNumber;
-  if (cfd->mem() != nullptr) {
-    earliest_mem_seqno =
-        std::min(cfd->mem()->GetEarliestSequenceNumber(), earliest_mem_seqno);
-  }
-  if (cfd->imm() != nullptr && cfd->imm()->current() != nullptr) {
-    earliest_mem_seqno =
-        std::min(cfd->imm()->current()->GetEarliestSequenceNumber(false),
-                 earliest_mem_seqno);
-  }
  Status s = cfd->compaction_picker()->SanitizeCompactionInputFiles(
-      &input_set, cf_meta, output_level, earliest_mem_seqno);
+      &input_set, cf_meta, output_level);
  if (!s.ok()) {
    return s;
  }
@@ -2756,6 +2756,46 @@ TEST_F(DBRangeDelTest, RefreshMemtableIter) {
  ASSERT_OK(iter->Refresh());
 }

+TEST_F(DBRangeDelTest, RangeTombstoneRespectIterateUpperBound) {
+  // Memtable: a, [b, bz)
+  // Do a Seek on `a` with iterate_upper_bound being az
+  // range tombstone [b, bz) should not be processed (added to and
+  // popped from the min_heap in MergingIterator).
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+
+  ASSERT_OK(Put("a", "bar"));
+  ASSERT_OK(
+      db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), "b", "bz"));
+
+  // I could not find a cleaner way to test this without relying on
+  // implementation detail. Tried to test the value of
+  // `internal_range_del_reseek_count` but that did not work
+  // since BlockBasedTable iterator becomes !Valid() when point key
+  // is out of bound and that reseek only happens when a point key
+  // is covered by some range tombstone.
+  SyncPoint::GetInstance()->SetCallBack("MergeIterator::PopDeleteRangeStart",
+                                        [](void*) {
+                                          // there should not be any range
+                                          // tombstone in the heap.
+                                          FAIL();
+                                        });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ReadOptions read_opts;
+  std::string upper_bound = "az";
+  Slice upper_bound_slice = upper_bound;
+  read_opts.iterate_upper_bound = &upper_bound_slice;
+  std::unique_ptr<Iterator> iter{db_->NewIterator(read_opts)};
+  iter->Seek("a");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ(iter->key(), "a");
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_OK(iter->status());
+}
+
 #endif  // ROCKSDB_LITE

 }  // namespace ROCKSDB_NAMESPACE
@@ -7509,6 +7509,80 @@ TEST_F(DBTest2, SstUniqueIdVerifyMultiCFs) {
  ASSERT_TRUE(s.IsCorruption());
 }

+TEST_F(DBTest2, BestEffortsRecoveryWithSstUniqueIdVerification) {
+  const auto tamper_with_uniq_id = [&](void* arg) {
+    auto props = static_cast<TableProperties*>(arg);
+    assert(props);
+    // update table property session_id to a different one
+    props->db_session_id = DBImpl::GenerateDbSessionId(nullptr);
+  };
+
+  const auto assert_db = [&](size_t expected_count,
+                             const std::string& expected_v) {
+    std::unique_ptr<Iterator> it(db_->NewIterator(ReadOptions()));
+    size_t cnt = 0;
+    for (it->SeekToFirst(); it->Valid(); it->Next(), ++cnt) {
+      ASSERT_EQ(std::to_string(cnt), it->key());
+      ASSERT_EQ(expected_v, it->value());
+    }
+    ASSERT_EQ(expected_count, cnt);
+  };
+
+  const int num_l0_compaction_trigger = 8;
+  const int num_l0 = num_l0_compaction_trigger - 1;
+  Options options = CurrentOptions();
+  options.level0_file_num_compaction_trigger = num_l0_compaction_trigger;
+
+  for (int k = 0; k < num_l0; ++k) {
+    // Allow mismatch for now
+    options.verify_sst_unique_id_in_manifest = false;
+
+    DestroyAndReopen(options);
+
+    constexpr size_t num_keys_per_file = 10;
+    for (int i = 0; i < num_l0; ++i) {
+      for (size_t j = 0; j < num_keys_per_file; ++j) {
+        ASSERT_OK(Put(std::to_string(j), "v" + std::to_string(i)));
+      }
+      if (i == k) {
+        SyncPoint::GetInstance()->DisableProcessing();
+        SyncPoint::GetInstance()->SetCallBack(
+            "PropertyBlockBuilder::AddTableProperty:Start",
+            tamper_with_uniq_id);
+        SyncPoint::GetInstance()->EnableProcessing();
+      }
+      ASSERT_OK(Flush());
+    }
+
+    options.verify_sst_unique_id_in_manifest = true;
+    Status s = TryReopen(options);
+    ASSERT_TRUE(s.IsCorruption());
+
+    options.best_efforts_recovery = true;
+    Reopen(options);
+    assert_db(k == 0 ? 0 : num_keys_per_file, "v" + std::to_string(k - 1));
+
+    // Reopen with regular recovery
+    options.best_efforts_recovery = false;
+    Reopen(options);
+    assert_db(k == 0 ? 0 : num_keys_per_file, "v" + std::to_string(k - 1));
+
+    SyncPoint::GetInstance()->DisableProcessing();
+    SyncPoint::GetInstance()->ClearAllCallBacks();
+
+    for (size_t i = 0; i < num_keys_per_file; ++i) {
+      ASSERT_OK(Put(std::to_string(i), "v"));
+    }
+    ASSERT_OK(Flush());
+    Reopen(options);
+    {
+      for (size_t i = 0; i < num_keys_per_file; ++i) {
+        ASSERT_EQ("v", Get(std::to_string(i)));
+      }
+    }
+  }
+}
+
 #ifndef ROCKSDB_LITE
 TEST_F(DBTest2, GetLatestSeqAndTsForKey) {
  Destroy(last_options_);
@@ -49,9 +49,6 @@
 #include "util/string_util.h"
 #include "utilities/merge_operators.h"

-// In case defined by Windows headers
-#undef small
-
 namespace ROCKSDB_NAMESPACE {
 class MockEnv;

@@ -76,7 +76,7 @@ MemTable::MemTable(const InternalKeyComparator& cmp,
    : comparator_(cmp),
      moptions_(ioptions, mutable_cf_options),
      refs_(0),
-      kArenaBlockSize(Arena::OptimizeBlockSize(moptions_.arena_block_size)),
+      kArenaBlockSize(OptimizeBlockSize(moptions_.arena_block_size)),
      mem_tracker_(write_buffer_manager),
      arena_(moptions_.arena_block_size,
             (write_buffer_manager != nullptr &&
@@ -127,6 +127,8 @@ Status TableCache::GetTableReader(
  FileOptions fopts = file_options;
  fopts.temperature = file_temperature;
  Status s = PrepareIOFromReadOptions(ro, ioptions_.clock, fopts.io_options);
+  TEST_SYNC_POINT_CALLBACK("TableCache::GetTableReader:BeforeOpenFile",
+                           const_cast<Status*>(&s));
  if (s.ok()) {
    s = ioptions_.fs->NewRandomAccessFile(fname, fopts, &file, nullptr);
  }
@@ -734,12 +734,13 @@ Status VersionEditHandlerPointInTime::MaybeCreateVersion(
  assert(!cfd->ioptions()->cf_paths.empty());
  Status s;
  for (const auto& elem : edit.GetNewFiles()) {
+    int level = elem.first;
    const FileMetaData& meta = elem.second;
    const FileDescriptor& fd = meta.fd;
    uint64_t file_num = fd.GetNumber();
    const std::string fpath =
        MakeTableFileName(cfd->ioptions()->cf_paths[0].path, file_num);
-    s = VerifyFile(fpath, meta);
+    s = VerifyFile(cfd, fpath, level, meta);
    if (s.IsPathNotFound() || s.IsNotFound() || s.IsCorruption()) {
      missing_files.insert(file_num);
      s = Status::OK();
@@ -804,6 +805,18 @@ Status VersionEditHandlerPointInTime::MaybeCreateVersion(
    auto* version = new Version(cfd, version_set_, version_set_->file_options_,
                                *cfd->GetLatestMutableCFOptions(), io_tracer_,
                                version_set_->current_version_number_++);
+    s = builder->LoadTableHandlers(
+        cfd->internal_stats(),
+        version_set_->db_options_->max_file_opening_threads, false, true,
+        cfd->GetLatestMutableCFOptions()->prefix_extractor,
+        MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
+    if (!s.ok()) {
+      delete version;
+      if (s.IsCorruption()) {
+        s = Status::OK();
+      }
+      return s;
+    }
    s = builder->SaveTo(version->storage_info());
    if (s.ok()) {
      version->PrepareAppend(
@@ -823,9 +836,11 @@ Status VersionEditHandlerPointInTime::MaybeCreateVersion(
  return s;
 }

-Status VersionEditHandlerPointInTime::VerifyFile(const std::string& fpath,
+Status VersionEditHandlerPointInTime::VerifyFile(ColumnFamilyData* cfd,
+                                                 const std::string& fpath,
+                                                 int level,
                                                 const FileMetaData& fmeta) {
-  return version_set_->VerifyFileMetadata(fpath, fmeta);
+  return version_set_->VerifyFileMetadata(cfd, fpath, level, fmeta);
 }

 Status VersionEditHandlerPointInTime::VerifyBlobFile(
@@ -843,6 +858,12 @@ Status VersionEditHandlerPointInTime::VerifyBlobFile(
  return s;
 }

+Status VersionEditHandlerPointInTime::LoadTables(
+    ColumnFamilyData* /*cfd*/, bool /*prefetch_index_and_filter_in_cache*/,
+    bool /*is_initial_load*/) {
+  return Status::OK();
+}
+
 Status ManifestTailer::Initialize() {
  if (Mode::kRecovery == mode_) {
    return VersionEditHandler::Initialize();
@@ -930,9 +951,11 @@ void ManifestTailer::CheckIterationResult(const log::Reader& reader,
  }
 }

-Status ManifestTailer::VerifyFile(const std::string& fpath,
+Status ManifestTailer::VerifyFile(ColumnFamilyData* cfd,
+                                  const std::string& fpath, int level,
                                  const FileMetaData& fmeta) {
-  Status s = VersionEditHandlerPointInTime::VerifyFile(fpath, fmeta);
+  Status s =
+      VersionEditHandlerPointInTime::VerifyFile(cfd, fpath, level, fmeta);
  // TODO: Open file or create hard link to prevent the file from being
  // deleted.
  return s;
@@ -164,9 +164,9 @@ class VersionEditHandler : public VersionEditHandlerBase {
                                    ColumnFamilyData* cfd,
                                    bool force_create_version);

-  Status LoadTables(ColumnFamilyData* cfd,
-                    bool prefetch_index_and_filter_in_cache,
-                    bool is_initial_load);
+  virtual Status LoadTables(ColumnFamilyData* cfd,
+                            bool prefetch_index_and_filter_in_cache,
+                            bool is_initial_load);

  virtual bool MustOpenAllColumnFamilies() const { return !read_only_; }

@@ -213,11 +213,15 @@ class VersionEditHandlerPointInTime : public VersionEditHandler {
  ColumnFamilyData* DestroyCfAndCleanup(const VersionEdit& edit) override;
  Status MaybeCreateVersion(const VersionEdit& edit, ColumnFamilyData* cfd,
                            bool force_create_version) override;
-  virtual Status VerifyFile(const std::string& fpath,
-                            const FileMetaData& fmeta);
+  virtual Status VerifyFile(ColumnFamilyData* cfd, const std::string& fpath,
+                            int level, const FileMetaData& fmeta);
  virtual Status VerifyBlobFile(ColumnFamilyData* cfd, uint64_t blob_file_num,
                                const BlobFileAddition& blob_addition);

+  Status LoadTables(ColumnFamilyData* cfd,
+                    bool prefetch_index_and_filter_in_cache,
+                    bool is_initial_load) override;
+
  std::unordered_map<uint32_t, Version*> versions_;
 };

@@ -250,7 +254,7 @@ class ManifestTailer : public VersionEditHandlerPointInTime {

  void CheckIterationResult(const log::Reader& reader, Status* s) override;

-  Status VerifyFile(const std::string& fpath,
+  Status VerifyFile(ColumnFamilyData* cfd, const std::string& fpath, int level,
                    const FileMetaData& fmeta) override;

  enum Mode : uint8_t {
@@ -2539,16 +2539,19 @@ void Version::MultiGet(const ReadOptions& read_options, MultiGetRange* range,
          }
          f = fp.GetNextFileInLevel();
        }
-        if (s.ok() && mget_tasks.size() > 0) {
+        if (mget_tasks.size() > 0) {
          RecordTick(db_statistics_, MULTIGET_COROUTINE_COUNT,
                     mget_tasks.size());
          // Collect all results so far
          std::vector<Status> statuses = folly::coro::blockingWait(
              folly::coro::collectAllRange(std::move(mget_tasks))
                  .scheduleOn(&range->context()->executor()));
-          for (Status stat : statuses) {
-            if (!stat.ok()) {
-              s = stat;
+          if (s.ok()) {
+            for (Status stat : statuses) {
+              if (!stat.ok()) {
+                s = std::move(stat);
+                break;
+              }
            }
          }

@@ -2794,6 +2797,9 @@ Status Version::MultiGetAsync(
    unsigned int num_tasks_queued = 0;
    to_process.pop_front();
    if (batch->IsSearchEnded() || batch->GetRange().empty()) {
+      // If to_process is empty, i.e no more batches to look at, then we need
+      // schedule the enqueued coroutines and wait for them. Otherwise, we
+      // skip this batch and move to the next one in to_process.
      if (!to_process.empty()) {
        continue;
      }
@@ -2802,9 +2808,6 @@ Status Version::MultiGetAsync(
      // to_process
      s = ProcessBatch(options, batch, mget_tasks, blob_ctxs, batches, waiting,
                       to_process, num_tasks_queued, mget_stats);
-      if (!s.ok()) {
-        break;
-      }
      // If ProcessBatch didn't enqueue any coroutine tasks, it means all
      // keys were filtered out. So put the batch back in to_process to
      // lookup in the next level
@@ -2815,8 +2818,10 @@ Status Version::MultiGetAsync(
        waiting.emplace_back(idx);
      }
    }
-    if (to_process.empty()) {
-      if (s.ok() && mget_tasks.size() > 0) {
+    // If ProcessBatch() returned an error, then schedule the enqueued
+    // coroutines and wait for them, then abort the MultiGet.
+    if (to_process.empty() || !s.ok()) {
+      if (mget_tasks.size() > 0) {
        assert(waiting.size());
        RecordTick(db_statistics_, MULTIGET_COROUTINE_COUNT, mget_tasks.size());
        // Collect all results so far
@@ -2824,10 +2829,12 @@ Status Version::MultiGetAsync(
            folly::coro::collectAllRange(std::move(mget_tasks))
                .scheduleOn(&range->context()->executor()));
        mget_tasks.clear();
-        for (Status stat : statuses) {
-          if (!stat.ok()) {
-            s = stat;
-            break;
+        if (s.ok()) {
+          for (Status stat : statuses) {
+            if (!stat.ok()) {
+              s = std::move(stat);
+              break;
+            }
          }
        }

@@ -2850,6 +2857,9 @@ Status Version::MultiGetAsync(
        assert(!s.ok() || waiting.size() == 0);
      }
    }
+    if (!s.ok()) {
+      break;
+    }
  }

  uint64_t num_levels = 0;
@@ -6704,8 +6714,9 @@ uint64_t VersionSet::GetTotalBlobFileSize(Version* dummy_versions) {
  return all_versions_blob_file_size;
 }

-Status VersionSet::VerifyFileMetadata(const std::string& fpath,
-                                      const FileMetaData& meta) const {
+Status VersionSet::VerifyFileMetadata(ColumnFamilyData* cfd,
+                                      const std::string& fpath, int level,
+                                      const FileMetaData& meta) {
  uint64_t fsize = 0;
  Status status = fs_->GetFileSize(fpath, IOOptions(), &fsize, nullptr);
  if (status.ok()) {
@@ -6713,6 +6724,38 @@ Status VersionSet::VerifyFileMetadata(const std::string& fpath,
      status = Status::Corruption("File size mismatch: " + fpath);
    }
  }
+  if (status.ok() && db_options_->verify_sst_unique_id_in_manifest) {
+    assert(cfd);
+    TableCache* table_cache = cfd->table_cache();
+    assert(table_cache);
+
+    const MutableCFOptions* const cf_opts = cfd->GetLatestMutableCFOptions();
+    assert(cf_opts);
+    std::shared_ptr<const SliceTransform> pe = cf_opts->prefix_extractor;
+    size_t max_sz_for_l0_meta_pin = MaxFileSizeForL0MetaPin(*cf_opts);
+
+    const FileOptions& file_opts = file_options();
+
+    Version* version = cfd->current();
+    assert(version);
+    VersionStorageInfo& storage_info = version->storage_info_;
+    const InternalKeyComparator* icmp = storage_info.InternalComparator();
+    assert(icmp);
+
+    InternalStats* internal_stats = cfd->internal_stats();
+
+    FileMetaData meta_copy = meta;
+    status = table_cache->FindTable(
+        ReadOptions(), file_opts, *icmp, meta_copy,
+        &(meta_copy.table_reader_handle), pe,
+        /*no_io=*/false, /*record_read_stats=*/true,
+        internal_stats->GetFileReadHist(level), false, level,
+        /*prefetch_index_and_filter_in_cache*/ false, max_sz_for_l0_meta_pin,
+        meta_copy.temperature);
+    if (meta_copy.table_reader_handle) {
+      table_cache->ReleaseHandle(meta_copy.table_reader_handle);
+    }
+  }
  return status;
 }

@@ -1501,8 +1501,8 @@ class VersionSet {
  ColumnFamilyData* CreateColumnFamily(const ColumnFamilyOptions& cf_options,
                                       const VersionEdit* edit);

-  Status VerifyFileMetadata(const std::string& fpath,
-                            const FileMetaData& meta) const;
+  Status VerifyFileMetadata(ColumnFamilyData* cfd, const std::string& fpath,
+                            int level, const FileMetaData& meta);

  // Protected by DB mutex.
  WalSet wals_;
@@ -150,7 +150,6 @@ DECLARE_string(cache_type);
 DECLARE_uint64(subcompactions);
 DECLARE_uint64(periodic_compaction_seconds);
 DECLARE_uint64(compaction_ttl);
-DECLARE_bool(fifo_allow_compaction);
 DECLARE_bool(allow_concurrent_memtable_write);
 DECLARE_double(experimental_mempurge_threshold);
 DECLARE_bool(enable_write_thread_adaptive_yield);
@@ -379,10 +379,6 @@ DEFINE_uint64(compaction_ttl, 1000,
 DEFINE_bool(allow_concurrent_memtable_write, false,
            "Allow multi-writers to update mem tables in parallel.");

-DEFINE_bool(fifo_allow_compaction, false,
-            "If true, set `Options::compaction_options_fifo.allow_compaction = "
-            "true`. It only take effect when FIFO compaction is used.");
-
 DEFINE_double(experimental_mempurge_threshold, 0.0,
              "Maximum estimated useful payload that triggers a "
              "mempurge process to collect memtable garbage bytes.");
@@ -3121,11 +3121,6 @@ void InitializeOptionsFromFlags(
  options.max_background_flushes = FLAGS_max_background_flushes;
  options.compaction_style =
      static_cast<ROCKSDB_NAMESPACE::CompactionStyle>(FLAGS_compaction_style);
-  if (options.compaction_style ==
-      ROCKSDB_NAMESPACE::CompactionStyle::kCompactionStyleFIFO) {
-    options.compaction_options_fifo.allow_compaction =
-        FLAGS_fifo_allow_compaction;
-  }
  options.compaction_pri =
      static_cast<ROCKSDB_NAMESPACE::CompactionPri>(FLAGS_compaction_pri);
  options.num_levels = FLAGS_num_levels;
@@ -247,10 +247,14 @@ void FilePrefetchBuffer::AbortAllIOs() {
  // Release io_handles.
  if (bufs_[curr_].io_handle_ != nullptr && bufs_[curr_].del_fn_ != nullptr) {
    DestroyAndClearIOHandle(curr_);
+  } else {
+    bufs_[curr_].async_read_in_progress_ = false;
  }

  if (bufs_[second].io_handle_ != nullptr && bufs_[second].del_fn_ != nullptr) {
    DestroyAndClearIOHandle(second);
+  } else {
+    bufs_[second].async_read_in_progress_ = false;
  }
 }

@@ -325,7 +329,16 @@ Status FilePrefetchBuffer::HandleOverlappingData(
    uint64_t& tmp_offset, size_t& tmp_length) {
  Status s;
  size_t alignment = reader->file()->GetRequiredBufferAlignment();
-  uint32_t second = curr_ ^ 1;
+  uint32_t second;
+
+  // Check if the first buffer has the required offset and the async read is
+  // still in progress. This should only happen if a prefetch was initiated
+  // by Seek, but the next access is at another offset.
+  if (bufs_[curr_].async_read_in_progress_ &&
+      IsOffsetInBufferWithAsyncProgress(offset, curr_)) {
+    PollAndUpdateBuffersIfNeeded(offset);
+  }
+  second = curr_ ^ 1;

  // If data is overlapping over two buffers, copy the data from curr_ and
  // call ReadAsync on curr_.
@@ -4,10 +4,14 @@
 //  (found in the LICENSE.Apache file in the root directory).

 #include "db/db_test_util.h"
+#include "file/file_prefetch_buffer.h"
+#include "file/file_util.h"
+#include "rocksdb/file_system.h"
 #include "test_util/sync_point.h"
 #ifdef GFLAGS
 #include "tools/io_tracer_parser_tool.h"
 #endif
+#include "util/random.h"

 namespace ROCKSDB_NAMESPACE {

@@ -2023,6 +2027,77 @@ TEST_P(PrefetchTest, TraceReadAsyncWithCallbackWrapper) {
  Close();
 }
 #endif  // GFLAGS
+
+class FilePrefetchBufferTest : public testing::Test {
+ public:
+  void SetUp() override {
+    SetupSyncPointsToMockDirectIO();
+    env_ = Env::Default();
+    fs_ = FileSystem::Default();
+    test_dir_ = test::PerThreadDBPath("file_prefetch_buffer_test");
+    ASSERT_OK(fs_->CreateDir(test_dir_, IOOptions(), nullptr));
+  }
+
+  void TearDown() override { EXPECT_OK(DestroyDir(env_, test_dir_)); }
+
+  void Write(const std::string& fname, const std::string& content) {
+    std::unique_ptr<FSWritableFile> f;
+    ASSERT_OK(fs_->NewWritableFile(Path(fname), FileOptions(), &f, nullptr));
+    ASSERT_OK(f->Append(content, IOOptions(), nullptr));
+    ASSERT_OK(f->Close(IOOptions(), nullptr));
+  }
+
+  void Read(const std::string& fname, const FileOptions& opts,
+            std::unique_ptr<RandomAccessFileReader>* reader) {
+    std::string fpath = Path(fname);
+    std::unique_ptr<FSRandomAccessFile> f;
+    ASSERT_OK(fs_->NewRandomAccessFile(fpath, opts, &f, nullptr));
+    reader->reset(new RandomAccessFileReader(std::move(f), fpath,
+                                             env_->GetSystemClock().get()));
+  }
+
+  void AssertResult(const std::string& content,
+                    const std::vector<FSReadRequest>& reqs) {
+    for (const auto& r : reqs) {
+      ASSERT_OK(r.status);
+      ASSERT_EQ(r.len, r.result.size());
+      ASSERT_EQ(content.substr(r.offset, r.len), r.result.ToString());
+    }
+  }
+
+  FileSystem* fs() { return fs_.get(); }
+
+ private:
+  Env* env_;
+  std::shared_ptr<FileSystem> fs_;
+  std::string test_dir_;
+
+  std::string Path(const std::string& fname) { return test_dir_ + "/" + fname; }
+};
+
+TEST_F(FilePrefetchBufferTest, SeekWithBlockCacheHit) {
+  std::string fname = "seek-with-block-cache-hit";
+  Random rand(0);
+  std::string content = rand.RandomString(32768);
+  Write(fname, content);
+
+  FileOptions opts;
+  std::unique_ptr<RandomAccessFileReader> r;
+  Read(fname, opts, &r);
+
+  FilePrefetchBuffer fpb(16384, 16384, true, false, false, 0, 0, fs());
+  Slice result;
+  // Simulate a seek of 4096 bytes at offset 0. Due to the readahead settings,
+  // it will do two reads of 4096+8192 and 8192
+  Status s = fpb.PrefetchAsync(IOOptions(), r.get(), 0, 4096, &result);
+  ASSERT_EQ(s, Status::TryAgain());
+  // Simulate a block cache hit
+  fpb.UpdateReadPattern(0, 4096, false);
+  // Now read some data that straddles the two prefetch buffers - offset 8192 to
+  // 16384
+  ASSERT_TRUE(fpb.TryReadFromCacheAsync(IOOptions(), r.get(), 8192, 8192,
+                                        &result, &s, Env::IOPriority::IO_LOW));
+}
 #endif  // ROCKSDB_LITE
 }  // namespace ROCKSDB_NAMESPACE

@@ -1285,30 +1285,37 @@ struct DBOptions {
  // Default: nullptr
  std::shared_ptr<FileChecksumGenFactory> file_checksum_gen_factory = nullptr;

-  // By default, RocksDB recovery fails if any table/blob file referenced in
+  // By default, RocksDB recovery fails if any table/blob file referenced in the
+  // final version reconstructed from the
  // MANIFEST are missing after scanning the MANIFEST pointed to by the
-  // CURRENT file.
-  // Best-efforts recovery is another recovery mode that tolerates missing or
-  // corrupted table or blob files.
+  // CURRENT file. It can also fail if verification of unique SST id fails.
+  // Best-efforts recovery is another recovery mode that does not necessarily
+  // fail when certain table/blob files are missing/corrupted or have mismatched
+  // unique id table property. Instead, best-efforts recovery recovers each
+  // column family to a point in the MANIFEST that corresponds to a version. In
+  // such a version, all valid table/blob files referenced have the expected
+  // file size. For table files, their unique id table property match the
+  // MANIFEST.
+  //
  // Best-efforts recovery does not need a valid CURRENT file, and tries to
  // recover the database using one of the available MANIFEST files in the db
  // directory.
-  // Best-efforts recovery recovers database to a state in which the database
-  // includes only table and blob files whose actual sizes match the
-  // information in the chosen MANIFEST without holes in the history.
  // Best-efforts recovery tries the available MANIFEST files from high file
  // numbers (newer) to low file numbers (older), and stops after finding the
  // first MANIFEST file from which the db can be recovered to a state without
-  // invalid (missing/file-mismatch) table and blob files.
-  // It is possible that the database can be restored to an empty state with no
-  // table or blob files.
-  // Regardless of this option, the IDENTITY file is updated if needed during
-  // recovery to match the DB ID in the MANIFEST (if previously using
-  // write_dbid_to_manifest) or to be in some valid state (non-empty DB ID).
-  // Currently, not compatible with atomic flush. Furthermore, WAL files will
-  // not be used for recovery if best_efforts_recovery is true.
-  // Also requires either 1) LOCK file exists or 2) underlying env's LockFile()
-  // call returns ok even for non-existing LOCK file.
+  // invalid (missing/filesize-mismatch/unique-id-mismatch) table and blob
+  // files. It is possible that the database can be restored to an empty state
+  // with no table or blob files.
+  //
+  // Regardless of this option, the IDENTITY file
+  // is updated if needed during recovery to match the DB ID in the MANIFEST (if
+  // previously using write_dbid_to_manifest) or to be in some valid state
+  // (non-empty DB ID). Currently, not compatible with atomic flush.
+  // Furthermore, WAL files will not be used for recovery if
+  // best_efforts_recovery is true. Also requires either 1) LOCK file exists or
+  // 2) underlying env's LockFile() call returns ok even for non-existing LOCK
+  // file.
+  //
  // Default: false
  bool best_efforts_recovery = false;

@@ -1480,7 +1487,7 @@ struct ReadOptions {
  const Slice* iterate_lower_bound;

  // "iterate_upper_bound" defines the extent up to which the forward iterator
-  // can returns entries. Once the bound is reached, Valid() will be false.
+  // can return entries. Once the bound is reached, Valid() will be false.
  // "iterate_upper_bound" is exclusive ie the bound value is
  // not a valid entry. If prefix_extractor is not null:
  // 1. If options.auto_prefix_mode = true, iterate_upper_bound will be used
@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 7
 #define ROCKSDB_MINOR 9
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 2

 // 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
@@ -8,7 +8,9 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.

 #include "memory/arena.h"
-
+#ifndef OS_WIN
+#include <sys/mman.h>
+#endif
 #include <algorithm>

 #include "logging/logging.h"
@@ -20,7 +22,16 @@

 namespace ROCKSDB_NAMESPACE {

-size_t Arena::OptimizeBlockSize(size_t block_size) {
+// MSVC complains that it is already defined since it is static in the header.
+#ifndef _MSC_VER
+const size_t Arena::kInlineSize;
+#endif
+
+const size_t Arena::kMinBlockSize = 4096;
+const size_t Arena::kMaxBlockSize = 2u << 30;
+static const int kAlignUnit = alignof(max_align_t);
+
+size_t OptimizeBlockSize(size_t block_size) {
  // Make sure block_size is in optimal range
  block_size = std::max(Arena::kMinBlockSize, block_size);
  block_size = std::min(Arena::kMaxBlockSize, block_size);
@@ -42,12 +53,14 @@ Arena::Arena(size_t block_size, AllocTracker* tracker, size_t huge_page_size)
  blocks_memory_ += alloc_bytes_remaining_;
  aligned_alloc_ptr_ = inline_block_;
  unaligned_alloc_ptr_ = inline_block_ + alloc_bytes_remaining_;
-  if (MemMapping::kHugePageSupported) {
-    hugetlb_size_ = huge_page_size;
-    if (hugetlb_size_ && kBlockSize > hugetlb_size_) {
-      hugetlb_size_ = ((kBlockSize - 1U) / hugetlb_size_ + 1U) * hugetlb_size_;
-    }
+#ifdef MAP_HUGETLB
+  hugetlb_size_ = huge_page_size;
+  if (hugetlb_size_ && kBlockSize > hugetlb_size_) {
+    hugetlb_size_ = ((kBlockSize - 1U) / hugetlb_size_ + 1U) * hugetlb_size_;
  }
+#else
+  (void)huge_page_size;
+#endif
  if (tracker_ != nullptr) {
    tracker_->Allocate(kInlineSize);
  }
@@ -58,6 +71,21 @@ Arena::~Arena() {
    assert(tracker_->is_freed());
    tracker_->FreeMem();
  }
+  for (const auto& block : blocks_) {
+    delete[] block;
+  }
+
+#ifdef MAP_HUGETLB
+  for (const auto& mmap_info : huge_blocks_) {
+    if (mmap_info.addr_ == nullptr) {
+      continue;
+    }
+    auto ret = munmap(mmap_info.addr_, mmap_info.length_);
+    if (ret != 0) {
+      // TODO(sdong): Better handling
+    }
+  }
+#endif
 }

 char* Arena::AllocateFallback(size_t bytes, bool aligned) {
@@ -71,10 +99,12 @@ char* Arena::AllocateFallback(size_t bytes, bool aligned) {
  // We waste the remaining space in the current block.
  size_t size = 0;
  char* block_head = nullptr;
-  if (MemMapping::kHugePageSupported && hugetlb_size_ > 0) {
+#ifdef MAP_HUGETLB
+  if (hugetlb_size_) {
    size = hugetlb_size_;
    block_head = AllocateFromHugePage(size);
  }
+#endif
  if (!block_head) {
    size = kBlockSize;
    block_head = AllocateNewBlock(size);
@@ -93,22 +123,45 @@ char* Arena::AllocateFallback(size_t bytes, bool aligned) {
 }

 char* Arena::AllocateFromHugePage(size_t bytes) {
-  MemMapping mm = MemMapping::AllocateHuge(bytes);
-  auto addr = static_cast<char*>(mm.Get());
-  if (addr) {
-    huge_blocks_.push_back(std::move(mm));
-    blocks_memory_ += bytes;
-    if (tracker_ != nullptr) {
-      tracker_->Allocate(bytes);
-    }
+#ifdef MAP_HUGETLB
+  if (hugetlb_size_ == 0) {
+    return nullptr;
  }
-  return addr;
+  // Reserve space in `huge_blocks_` before calling `mmap`.
+  // Use `emplace_back()` instead of `reserve()` to let std::vector manage its
+  // own memory and do fewer reallocations.
+  //
+  // - If `emplace_back` throws, no memory leaks because we haven't called
+  //   `mmap` yet.
+  // - If `mmap` throws, no memory leaks because the vector will be cleaned up
+  //   via RAII.
+  huge_blocks_.emplace_back(nullptr /* addr */, 0 /* length */);
+
+  void* addr = mmap(nullptr, bytes, (PROT_READ | PROT_WRITE),
+                    (MAP_PRIVATE | MAP_ANONYMOUS | MAP_HUGETLB), -1, 0);
+
+  if (addr == MAP_FAILED) {
+    return nullptr;
+  }
+  huge_blocks_.back() = MmapInfo(addr, bytes);
+  blocks_memory_ += bytes;
+  if (tracker_ != nullptr) {
+    tracker_->Allocate(bytes);
+  }
+  return reinterpret_cast<char*>(addr);
+#else
+  (void)bytes;
+  return nullptr;
+#endif
 }

 char* Arena::AllocateAligned(size_t bytes, size_t huge_page_size,
                             Logger* logger) {
-  if (MemMapping::kHugePageSupported && hugetlb_size_ > 0 &&
-      huge_page_size > 0 && bytes > 0) {
+  assert((kAlignUnit & (kAlignUnit - 1)) ==
+         0);  // Pointer size should be a power of 2
+
+#ifdef MAP_HUGETLB
+  if (huge_page_size > 0 && bytes > 0) {
    // Allocate from a huge page TLB table.
    size_t reserved_size =
        ((bytes - 1U) / huge_page_size + 1U) * huge_page_size;
@@ -124,6 +177,10 @@ char* Arena::AllocateAligned(size_t bytes, size_t huge_page_size,
      return addr;
    }
  }
+#else
+  (void)huge_page_size;
+  (void)logger;
+#endif

  size_t current_mod =
      reinterpret_cast<uintptr_t>(aligned_alloc_ptr_) & (kAlignUnit - 1);
@@ -143,10 +200,17 @@ char* Arena::AllocateAligned(size_t bytes, size_t huge_page_size,
 }

 char* Arena::AllocateNewBlock(size_t block_bytes) {
-  auto uniq = std::make_unique<char[]>(block_bytes);
-  char* block = uniq.get();
-  blocks_.push_back(std::move(uniq));
+  // Reserve space in `blocks_` before allocating memory via new.
+  // Use `emplace_back()` instead of `reserve()` to let std::vector manage its
+  // own memory and do fewer reallocations.
+  //
+  // - If `emplace_back` throws, no memory leaks because we haven't called `new`
+  //   yet.
+  // - If `new` throws, no memory leaks because the vector will be cleaned up
+  //   via RAII.
+  blocks_.emplace_back(nullptr);

+  char* block = new char[block_bytes];
  size_t allocated_size;
 #ifdef ROCKSDB_MALLOC_USABLE_SIZE
  allocated_size = malloc_usable_size(block);
@@ -163,6 +227,7 @@ char* Arena::AllocateNewBlock(size_t block_bytes) {
  if (tracker_ != nullptr) {
    tracker_->Allocate(allocated_size);
  }
+  blocks_.back() = block;
  return block;
 }

@@ -12,13 +12,16 @@
 // size, it uses malloc to directly get the requested size.

 #pragma once
-
+#ifndef OS_WIN
+#include <sys/mman.h>
+#endif
+#include <assert.h>
+#include <stdint.h>
+#include <cerrno>
 #include <cstddef>
-#include <deque>
-
+#include <vector>
 #include "memory/allocator.h"
-#include "port/mmap.h"
-#include "rocksdb/env.h"
+#include "util/mutexlock.h"

 namespace ROCKSDB_NAMESPACE {

@@ -28,13 +31,9 @@ class Arena : public Allocator {
  Arena(const Arena&) = delete;
  void operator=(const Arena&) = delete;

-  static constexpr size_t kInlineSize = 2048;
-  static constexpr size_t kMinBlockSize = 4096;
-  static constexpr size_t kMaxBlockSize = 2u << 30;
-
-  static constexpr unsigned kAlignUnit = alignof(std::max_align_t);
-  static_assert((kAlignUnit & (kAlignUnit - 1)) == 0,
-                "Pointer size should be power of 2");
+  static const size_t kInlineSize = 2048;
+  static const size_t kMinBlockSize;
+  static const size_t kMaxBlockSize;

  // huge_page_size: if 0, don't use huge page TLB. If > 0 (should set to the
  // supported hugepage size of the system), block allocation will try huge
@@ -64,7 +63,7 @@ class Arena : public Allocator {
  // by the arena (exclude the space allocated but not yet used for future
  // allocations).
  size_t ApproximateMemoryUsage() const {
-    return blocks_memory_ + blocks_.size() * sizeof(char*) -
+    return blocks_memory_ + blocks_.capacity() * sizeof(char*) -
           alloc_bytes_remaining_;
  }

@@ -82,19 +81,21 @@ class Arena : public Allocator {
    return blocks_.empty() && huge_blocks_.empty();
  }

-  // check and adjust the block_size so that the return value is
-  //  1. in the range of [kMinBlockSize, kMaxBlockSize].
-  //  2. the multiple of align unit.
-  static size_t OptimizeBlockSize(size_t block_size);
-
 private:
-  alignas(std::max_align_t) char inline_block_[kInlineSize];
+  char inline_block_[kInlineSize] __attribute__((__aligned__(alignof(max_align_t))));
  // Number of bytes allocated in one block
  const size_t kBlockSize;
-  // Allocated memory blocks
-  std::deque<std::unique_ptr<char[]>> blocks_;
-  // Huge page allocations
-  std::deque<MemMapping> huge_blocks_;
+  // Array of new[] allocated memory blocks
+  using Blocks = std::vector<char*>;
+  Blocks blocks_;
+
+  struct MmapInfo {
+    void* addr_;
+    size_t length_;
+
+    MmapInfo(void* addr, size_t length) : addr_(addr), length_(length) {}
+  };
+  std::vector<MmapInfo> huge_blocks_;
  size_t irregular_block_num = 0;

  // Stats for current active block.
@@ -107,15 +108,15 @@ class Arena : public Allocator {
  // How many bytes left in currently active block?
  size_t alloc_bytes_remaining_ = 0;

+#ifdef MAP_HUGETLB
  size_t hugetlb_size_ = 0;
-
+#endif  // MAP_HUGETLB
  char* AllocateFromHugePage(size_t bytes);
  char* AllocateFallback(size_t bytes, bool aligned);
  char* AllocateNewBlock(size_t block_bytes);

  // Bytes of memory in blocks allocated so far
  size_t blocks_memory_ = 0;
-  // Non-owned
  AllocTracker* tracker_;
 };

@@ -132,4 +133,9 @@ inline char* Arena::Allocate(size_t bytes) {
  return AllocateFallback(bytes, false /* unaligned */);
 }

+// check and adjust the block_size so that the return value is
+//  1. in the range of [kMinBlockSize, kMaxBlockSize].
+//  2. the multiple of align unit.
+extern size_t OptimizeBlockSize(size_t block_size);
+
 }  // namespace ROCKSDB_NAMESPACE
@@ -8,11 +8,6 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.

 #include "memory/arena.h"
-
-#ifndef OS_WIN
-#include <sys/resource.h>
-#endif
-#include "port/port.h"
 #include "test_util/testharness.h"
 #include "util/random.h"

@@ -201,61 +196,6 @@ TEST_F(ArenaTest, Simple) {
  SimpleTest(0);
  SimpleTest(kHugePageSize);
 }
-
-// Number of minor page faults since last call
-size_t PopMinorPageFaultCount() {
-#ifdef RUSAGE_SELF
-  static long prev = 0;
-  struct rusage usage;
-  EXPECT_EQ(getrusage(RUSAGE_SELF, &usage), 0);
-  size_t rv = usage.ru_minflt - prev;
-  prev = usage.ru_minflt;
-  return rv;
-#else
-  // Conservative
-  return SIZE_MAX;
-#endif  // RUSAGE_SELF
-}
-
-TEST(MmapTest, AllocateLazyZeroed) {
-  // Doesn't have to be page aligned
-  constexpr size_t len = 1234567;
-  MemMapping m = MemMapping::AllocateLazyZeroed(len);
-  auto arr = static_cast<char*>(m.Get());
-
-  // Should generally work
-  ASSERT_NE(arr, nullptr);
-
-  // Start counting page faults
-  PopMinorPageFaultCount();
-
-  // Access half of the allocation
-  size_t i = 0;
-  for (; i < len / 2; ++i) {
-    ASSERT_EQ(arr[i], 0);
-    arr[i] = static_cast<char>(i & 255);
-  }
-
-  // Appropriate page faults (maybe more)
-  size_t faults = PopMinorPageFaultCount();
-  ASSERT_GE(faults, len / 2 / port::kPageSize);
-
-  // Access rest of the allocation
-  for (; i < len; ++i) {
-    ASSERT_EQ(arr[i], 0);
-    arr[i] = static_cast<char>(i & 255);
-  }
-
-  // Appropriate page faults (maybe more)
-  faults = PopMinorPageFaultCount();
-  ASSERT_GE(faults, len / 2 / port::kPageSize);
-
-  // Verify data
-  for (i = 0; i < len; ++i) {
-    ASSERT_EQ(arr[i], static_cast<char>(i & 255));
-  }
-}
-
 }  // namespace ROCKSDB_NAMESPACE

 int main(int argc, char** argv) {
@@ -1,98 +0,0 @@
-//  Copyright (c) Meta Platforms, Inc. and affiliates.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-
-#include "port/mmap.h"
-
-#include <cassert>
-#include <cstdio>
-#include <cstring>
-#include <new>
-#include <utility>
-
-#include "util/hash.h"
-
-namespace ROCKSDB_NAMESPACE {
-
-MemMapping::~MemMapping() {
-#ifdef OS_WIN
-  if (addr_ != nullptr) {
-    (void)::UnmapViewOfFile(addr_);
-  }
-  if (page_file_handle_ != NULL) {
-    (void)::CloseHandle(page_file_handle_);
-  }
-#else   // OS_WIN -> !OS_WIN
-  if (addr_ != nullptr) {
-    auto status = munmap(addr_, length_);
-    assert(status == 0);
-    if (status != 0) {
-      // TODO: handle error?
-    }
-  }
-#endif  // OS_WIN
-}
-
-MemMapping::MemMapping(MemMapping&& other) noexcept {
-  *this = std::move(other);
-}
-
-MemMapping& MemMapping::operator=(MemMapping&& other) noexcept {
-  if (&other == this) {
-    return *this;
-  }
-  this->~MemMapping();
-  std::memcpy(this, &other, sizeof(*this));
-  new (&other) MemMapping();
-  return *this;
-}
-
-MemMapping MemMapping::AllocateAnonymous(size_t length, bool huge) {
-  MemMapping mm;
-  mm.length_ = length;
-  assert(mm.addr_ == nullptr);
-  if (length == 0) {
-    // OK to leave addr as nullptr
-    return mm;
-  }
-  int huge_flag = 0;
-#ifdef OS_WIN
-  if (huge) {
-#ifdef FILE_MAP_LARGE_PAGES
-    huge_flag = FILE_MAP_LARGE_PAGES;
-#endif  // FILE_MAP_LARGE_PAGES
-  }
-  mm.page_file_handle_ = ::CreateFileMapping(
-      INVALID_HANDLE_VALUE, nullptr, PAGE_READWRITE | SEC_COMMIT,
-      Upper32of64(length), Lower32of64(length), nullptr);
-  if (mm.page_file_handle_ == NULL) {
-    // Failure
-    return mm;
-  }
-  mm.addr_ = ::MapViewOfFile(mm.page_file_handle_, FILE_MAP_WRITE | huge_flag,
-                             0, 0, length);
-#else  // OS_WIN -> !OS_WIN
-  if (huge) {
-#ifdef MAP_HUGETLB
-    huge_flag = MAP_HUGETLB;
-#endif  // MAP_HUGE_TLB
-  }
-  mm.addr_ = mmap(nullptr, length, PROT_READ | PROT_WRITE,
-                  MAP_PRIVATE | MAP_ANONYMOUS | huge_flag, -1, 0);
-  if (mm.addr_ == MAP_FAILED) {
-    mm.addr_ = nullptr;
-  }
-#endif  // OS_WIN
-  return mm;
-}
-
-MemMapping MemMapping::AllocateHuge(size_t length) {
-  return AllocateAnonymous(length, /*huge*/ true);
-}
-
-MemMapping MemMapping::AllocateLazyZeroed(size_t length) {
-  return AllocateAnonymous(length, /*huge*/ false);
-}
-
-}  // namespace ROCKSDB_NAMESPACE
@@ -1,70 +0,0 @@
-//  Copyright (c) Meta Platforms, Inc. and affiliates.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-
-#pragma once
-
-#ifdef OS_WIN
-#include "port/win/port_win.h"
-// ^^^ For proper/safe inclusion of windows.h. Must come first.
-#include <memoryapi.h>
-#else
-#include <sys/mman.h>
-#endif  // OS_WIN
-
-#include <cstdint>
-
-#include "rocksdb/rocksdb_namespace.h"
-
-namespace ROCKSDB_NAMESPACE {
-
-// An RAII wrapper for mmaped memory
-class MemMapping {
- public:
-  static constexpr bool kHugePageSupported =
-#if defined(MAP_HUGETLB) || defined(FILE_MAP_LARGE_PAGES)
-      true;
-#else
-      false;
-#endif
-
-  // Allocate memory requesting to be backed by huge pages
-  static MemMapping AllocateHuge(size_t length);
-
-  // Allocate memory that is only lazily mapped to resident memory and
-  // guaranteed to be zero-initialized. Note that some platforms like
-  // Linux allow memory over-commit, where only the used portion of memory
-  // matters, while other platforms require enough swap space (page file) to
-  // back the full mapping.
-  static MemMapping AllocateLazyZeroed(size_t length);
-
-  // No copies
-  MemMapping(const MemMapping&) = delete;
-  MemMapping& operator=(const MemMapping&) = delete;
-  // Move
-  MemMapping(MemMapping&&) noexcept;
-  MemMapping& operator=(MemMapping&&) noexcept;
-
-  // Releases the mapping
-  ~MemMapping();
-
-  inline void* Get() const { return addr_; }
-  inline size_t Length() const { return length_; }
-
- private:
-  MemMapping() {}
-
-  // The mapped memory, or nullptr on failure / not supported
-  void* addr_ = nullptr;
-  // The known usable number of bytes starting at that address
-  size_t length_ = 0;
-
-#ifdef OS_WIN
-  HANDLE page_file_handle_ = NULL;
-#endif  // OS_WIN
-
-  static MemMapping AllocateAnonymous(size_t length, bool huge);
-};
-
-}  // namespace ROCKSDB_NAMESPACE
@@ -154,7 +154,6 @@ LIB_SOURCES =                                                   \
  options/options.cc                                            \
  options/options_helper.cc                                     \
  options/options_parser.cc                                     \
-  port/mmap.cc                                                  \
  port/port_posix.cc                                            \
  port/win/env_default.cc                                       \
  port/win/env_win.cc                                           \
@@ -117,14 +117,16 @@ class MergingIterator : public InternalIterator {
 public:
  MergingIterator(const InternalKeyComparator* comparator,
                  InternalIterator** children, int n, bool is_arena_mode,
-                  bool prefix_seek_mode)
+                  bool prefix_seek_mode,
+                  const Slice* iterate_upper_bound = nullptr)
      : is_arena_mode_(is_arena_mode),
        prefix_seek_mode_(prefix_seek_mode),
        direction_(kForward),
        comparator_(comparator),
        current_(nullptr),
        minHeap_(comparator_),
-        pinned_iters_mgr_(nullptr) {
+        pinned_iters_mgr_(nullptr),
+        iterate_upper_bound_(iterate_upper_bound) {
    children_.resize(n);
    for (int i = 0; i < n; i++) {
      children_[i].level = i;
@@ -202,11 +204,26 @@ class MergingIterator : public InternalIterator {
    assert(!range_tombstone_iters_.empty() &&
           range_tombstone_iters_[level]->Valid());
    if (start_key) {
-      pinned_heap_item_[level].SetTombstoneKey(
-          range_tombstone_iters_[level]->start_key());
+      ParsedInternalKey pik = range_tombstone_iters_[level]->start_key();
+      // iterate_upper_bound does not have timestamp
+      if (iterate_upper_bound_ &&
+          comparator_->user_comparator()->CompareWithoutTimestamp(
+              pik.user_key, true /* a_has_ts */, *iterate_upper_bound_,
+              false /* b_has_ts */) >= 0) {
+        if (replace_top) {
+          // replace_top implies this range tombstone iterator is still in
+          // minHeap_ and at the top.
+          minHeap_.pop();
+        }
+        return;
+      }
+      pinned_heap_item_[level].SetTombstoneKey(std::move(pik));
      pinned_heap_item_[level].type = HeapItem::DELETE_RANGE_START;
      assert(active_.count(level) == 0);
    } else {
+      // allow end key to go over upper bound (if present) since start key is
+      // before upper bound and the range tombstone could still cover a
+      // range before upper bound.
      pinned_heap_item_[level].SetTombstoneKey(
          range_tombstone_iters_[level]->end_key());
      pinned_heap_item_[level].type = HeapItem::DELETE_RANGE_END;
@@ -251,6 +268,7 @@ class MergingIterator : public InternalIterator {
  void PopDeleteRangeStart() {
    while (!minHeap_.empty() &&
           minHeap_.top()->type == HeapItem::DELETE_RANGE_START) {
+      TEST_SYNC_POINT_CALLBACK("MergeIterator::PopDeleteRangeStart", nullptr);
      // insert end key of this range tombstone and updates active_
      InsertRangeTombstoneToMinHeap(
          minHeap_.top()->level, false /* start_key */, true /* replace_top */);
@@ -573,6 +591,10 @@ class MergingIterator : public InternalIterator {
  std::unique_ptr<MergerMaxIterHeap> maxHeap_;
  PinnedIteratorsManager* pinned_iters_mgr_;

+  // Used to bound range tombstones. For point keys, DBIter and SSTable iterator
+  // take care of boundary checking.
+  const Slice* iterate_upper_bound_;
+
  // In forward direction, process a child that is not in the min heap.
  // If valid, add to the min heap. Otherwise, check status.
  void AddToMinHeapOrCheckStatus(HeapItem*);
@@ -634,9 +656,19 @@ void MergingIterator::SeekImpl(const Slice& target, size_t starting_level,
    for (size_t level = 0; level < starting_level; ++level) {
      if (range_tombstone_iters_[level] &&
          range_tombstone_iters_[level]->Valid()) {
-        assert(static_cast<bool>(active_.count(level)) ==
-               (pinned_heap_item_[level].type == HeapItem::DELETE_RANGE_END));
-        minHeap_.push(&pinned_heap_item_[level]);
+        // use an iterator on active_ if performance becomes an issue here
+        if (active_.count(level) > 0) {
+          assert(pinned_heap_item_[level].type == HeapItem::DELETE_RANGE_END);
+          // if it was active, then start key must be within upper_bound,
+          // so we can add to minHeap_ directly.
+          minHeap_.push(&pinned_heap_item_[level]);
+        } else {
+          // this takes care of checking iterate_upper_bound, but with an extra
+          // key comparison if range_tombstone_iters_[level] was already out of
+          // bound. Consider using a new HeapItem type or some flag to remember
+          // boundary checking result.
+          InsertRangeTombstoneToMinHeap(level);
+        }
      } else {
        assert(!active_.count(level));
      }
@@ -1280,11 +1312,12 @@ InternalIterator* NewMergingIterator(const InternalKeyComparator* cmp,
 }

 MergeIteratorBuilder::MergeIteratorBuilder(
-    const InternalKeyComparator* comparator, Arena* a, bool prefix_seek_mode)
+    const InternalKeyComparator* comparator, Arena* a, bool prefix_seek_mode,
+    const Slice* iterate_upper_bound)
    : first_iter(nullptr), use_merging_iter(false), arena(a) {
  auto mem = arena->AllocateAligned(sizeof(MergingIterator));
-  merge_iter =
-      new (mem) MergingIterator(comparator, nullptr, 0, true, prefix_seek_mode);
+  merge_iter = new (mem) MergingIterator(comparator, nullptr, 0, true,
+                                         prefix_seek_mode, iterate_upper_bound);
 }

 MergeIteratorBuilder::~MergeIteratorBuilder() {
@@ -45,7 +45,8 @@ class MergeIteratorBuilder {
  // comparator: the comparator used in merging comparator
  // arena: where the merging iterator needs to be allocated from.
  explicit MergeIteratorBuilder(const InternalKeyComparator* comparator,
-                                Arena* arena, bool prefix_seek_mode = false);
+                                Arena* arena, bool prefix_seek_mode = false,
+                                const Slice* iterate_upper_bound = nullptr);
  ~MergeIteratorBuilder();

  // Add iter to the merging iterator.
@@ -139,7 +139,6 @@ default_params = {
    # 0 = never (used by some), 10 = often (for threading bugs), 600 = default
    "stats_dump_period_sec": lambda: random.choice([0, 10, 600]),
    "compaction_ttl": lambda: random.choice([0, 0, 1, 2, 10, 100, 1000]),
-    "fifo_allow_compaction": lambda: random.randint(0, 1),
    # Test small max_manifest_file_size in a smaller chance, as most of the
    # time we wnat manifest history to be preserved to help debug
    "max_manifest_file_size": lambda: random.choice(
Author	SHA1	Message	Date
anand76	444b3f4845	Add a unit test for PR 11049	2022-12-22 09:30:39 -08:00
anand76	23bec22a81	Bump version to 7.9.2	2022-12-21 09:51:53 -08:00
anand76	208310ad99	Fix async prefetch heap use after free (#11049 ) Summary: This PR fixes a heap use after free bug in the async prefetch code that happens in the following scenario - 1. Scan thread starts 2 async reads for Seek, one for the seek block and one for prefetching 2. Before the first read in https://github.com/facebook/rocksdb/issues/1 completes, another thread reads and loads the block in cache 3. The first scan thread finds the block in cache, continues and the next block cache miss is for a block that spans the boundary of the 2 prefetch buffers, and the 1st read is complete but the 2nd one is not complete yet 4. The scan thread will reallocate (i.e free the old buffer and allocate a new one) the 2nd prefetch buffer, and the in-progress prefetch is orphaned 5. The orphaned prefetch finally completes, resulting in a use after free Also add a few asserts to surface bugs earlier in the crash tests. Pull Request resolved: https://github.com/facebook/rocksdb/pull/11049 Test Plan: Repro with db_stress and verify the fix Reviewed By: akankshamahajan15 Differential Revision: D42181118 Pulled By: anand1976 fbshipit-source-id: 1ac55d2f64a89ce128c1c574262b8aa7d82eb8cc	2022-12-21 09:46:30 -08:00
anand76	afa92034d9	Update version.h to 7.9.1	2022-12-08 19:54:29 -08:00
anand76	a5e9481295	Update HISTORY.md for 7.9.1	2022-12-08 19:25:26 -08:00
anand76	742035d21f	Fix table cache leak in MultiGet with async_io (#10997 ) Summary: When MultiGet with the async_io option encounters an IO error in TableCache::GetTableReader, it may result in leakage of table cache handles due to queued coroutines being abandoned. This PR fixes it by ensuring any queued coroutines are run before aborting the MultiGet. Pull Request resolved: https://github.com/facebook/rocksdb/pull/10997 Test Plan: 1. New unit test in db_basic_test 2. asan_crash Reviewed By: pdillinger Differential Revision: D41587244 Pulled By: anand1976 fbshipit-source-id: 900920cd3fba47cb0fc744a62facc5ffe2eccb64	2022-12-08 17:51:33 -08:00
Peter Dillinger	62f3e58936	Revert "Improve / refactor anonymous mmap capabilities (#10810 )" This reverts commit `8367f0d2d7`.	2022-11-30 17:14:01 -08:00
Peter Dillinger	be728f83e8	Revert "Fix include of windows.h in mmap.h (#10885 )" This reverts commit `49b7f219de`.	2022-11-30 17:12:09 -08:00
Hui Xiao	3ff86f39ae	Revert PR 10777 "Fix FIFO causing overlapping seqnos in L0 files due to overla…" (#10999 ) Summary: Context/Summary: This reverts commit `fc74abb436` and related HISTORY record. The issue with PR 10777 or general approach using earliest_mem_seqno like https://github.com/facebook/rocksdb/pull/5958#issue-511150930 is that the earliest seqno of memtable of each CFs does not get persisted and will always start with 0 upon Recover(). Later when creating a new memtable in certain CF, we use the last seqno of the whole DB (but not of that CF from previous DB session) for this CF. This will lead to false positive overlapping seqno and PR 10777 will throw something like https://github.com/facebook/rocksdb/blob/main/db/compaction/compaction_picker.cc#L1002-L1004 Luckily a more elegant and complete solution to the overlapping seqno problem these PR aim to solve does not have above problem, see https://github.com/facebook/rocksdb/pull/10922. It is already being pursued and in the process of review. So we can just revert this PR and focus on getting PR10922 to land. Pull Request resolved: https://github.com/facebook/rocksdb/pull/10999 Test Plan: make check Reviewed By: anand1976 Differential Revision: D41572604 Pulled By: hx235 fbshipit-source-id: 9d9bdf594abd235e2137045cef513ca0b14e0a3a	2022-11-29 11:39:13 -08:00
Changyu Bi	ab9741a6a9	Merge pull request #10984 from cbi42/7.9-iterate-upperbound [7.9] Prevent iterating over range tombstones beyond `iterate_upper_bound`	2022-11-23 16:28:01 -08:00
Changyu Bi	9399bb844b	Prevent iterating over range tombstones beyond `iterate_upper_bound` (#10966 ) Summary: Currently, `iterate_upper_bound` is not checked for range tombstone keys in MergingIterator. This may impact performance when there is a large number of range tombstones right after `iterate_upper_bound`. This PR fixes this issue by checking `iterate_upper_bound` in MergingIterator for range tombstone keys. Pull Request resolved: https://github.com/facebook/rocksdb/pull/10966 Test Plan: - added unit test - stress test: `python3 tools/db_crashtest.py whitebox --simple --verify_iterator_with_expected_state_one_in=5 --delrangepercent=5 --prefixpercent=18 --writepercent=48 --readpercen=15 --duration=36000 --range_deletion_width=100` - ran different stress tests over sandcastle - Falcon team ran some test traffic and saw reduced CPU usage on processing range tombstones. Reviewed By: ajkr Differential Revision: D41414172 Pulled By: cbi42 fbshipit-source-id: 9b2c29eb3abb99327c6a649bdc412e70d863f981	2022-11-23 15:46:51 -08:00
Yanqin Jin	ed91ed0a50	Make best-efforts recovery verify SST unique ID before Version construction (#10962 ) Summary: The check for SST unique IDs added to best-efforts recovery (`Options::best_efforts_recovery` is true). With best_efforts_recovery being true, RocksDB will recover to the latest point in MANIFEST such that all valid SST files included up to this point pass unique ID checks as well. Pull Request resolved: https://github.com/facebook/rocksdb/pull/10962 Test Plan: make check Reviewed By: pdillinger Differential Revision: D41378241 Pulled By: riversand963 fbshipit-source-id: a036064e2c17dec13d080a24ef2a9f85d607b16c	2022-11-23 09:01:51 -08:00