Making platform 007 (gcc 7) default in build_detect_platform.sh (#5947 )

Summary: Making platform 007 (gcc 7) default in build_detect_platform.sh. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5947 Differential Revision: D18038837 Pulled By: vjnadimpalli fbshipit-source-id: 9ac2ddaa93bf328a416faec028970e039886378e
Bump version to 6.2.4
2026-11-07 03:22:34 -05:00 · 2019-10-30 10:29:05 -07:00 · 2019-09-18 14:42:43 -07:00 · 2019-09-18 14:39:45 -07:00 · 2019-09-18 14:37:22 -07:00 · 2019-09-18 14:37:16 -07:00
53 changed files with 1083 additions and 219 deletions
@@ -92,7 +92,7 @@ else()
  endif()

  if(WITH_ZLIB)
-    find_package(zlib REQUIRED)
+    find_package(ZLIB REQUIRED)
    add_definitions(-DZLIB)
    if(ZLIB_INCLUDE_DIRS)
      # CMake 3
@@ -317,6 +317,10 @@ if(DISABLE_STALL_NOTIF)
  add_definitions(-DROCKSDB_DISABLE_STALL_NOTIFICATION)
 endif()

+option(WITH_DYNAMIC_EXTENSION "build with dynamic extension support" OFF)
+if(NOT WITH_DYNAMIC_EXTENSION)
+  add_definitions(-DROCKSDB_NO_DYNAMIC_EXTENSION)
+endif()

 if(DEFINED USE_RTTI)
  if(USE_RTTI)
@@ -461,8 +465,6 @@ include_directories(${PROJECT_SOURCE_DIR}/include)
 include_directories(SYSTEM ${PROJECT_SOURCE_DIR}/third-party/gtest-1.7.0/fused-src)
 find_package(Threads REQUIRED)

-add_subdirectory(third-party/gtest-1.7.0/fused-src/gtest)
-
 # Main library source code

 set(SOURCES
@@ -843,6 +845,7 @@ endif()

 option(WITH_TESTS "build with tests" ON)
 if(WITH_TESTS)
+  add_subdirectory(third-party/gtest-1.7.0/fused-src/gtest)
  set(TESTS
        cache/cache_test.cc
        cache/lru_cache_test.cc
@@ -1,13 +1,30 @@
 # Rocksdb Change Log
-## Unreleased
+## 6.2.4 (9/18/2019)
+### Bug Fixes
+* Disable snap_refresh_nanos by default. The feature is to be deprecated in the next release.
+
+## 6.2.3 (9/3/2019)
+### Bug Fixes
+* Fix a bug in file ingestion caused by incorrect file number allocation when the number of column families involved in the ingestion exceeds 2.
+
+## 6.2.2 (6/7/2019)
+### Bug Fixes
+* Disable dynamic extension support by default for CMake.
+
+## 6.2.1 (6/4/2019)
+### Bug Fixes
+* Fix flush's/compaction's merge processing logic which allowed `Put`s covered by range tombstones to reappear. Note `Put`s may exist even if the user only ever called `Merge()` due to an internal conversion during compaction to the bottommost level.
+
+## 6.2.0 (4/30/2019)
 ### New Features
 * Add an option `strict_bytes_per_sync` that causes a file-writing thread to block rather than exceed the limit on bytes pending writeback specified by `bytes_per_sync` or `wal_bytes_per_sync`.
-
-## Unreleased
-### New Features
 * Improve range scan performance by avoiding per-key upper bound check in BlockBasedTableIterator.
 * Introduce Periodic Compaction for Level style compaction. Files are re-compacted periodically and put in the same level.
 * Block-based table index now contains exact highest key in the file, rather than an upper bound. This may improve Get() and iterator Seek() performance in some situations, especially when direct IO is enabled and block cache is disabled. A setting BlockBasedTableOptions::index_shortening is introduced to control this behavior. Set it to kShortenSeparatorsAndSuccessor to get the old behavior.
+* When reading from option file/string/map, customized envs can be filled according to object registry.
+* Add an option `snap_refresh_nanos` (default to 0.5s) to periodically refresh the snapshot list in compaction jobs. Assign to 0 to disable the feature.
+* Improve range scan performance when using explicit user readahead by not creating new table readers for every iterator.
+
 ### Public API Change
 * Change the behavior of OptimizeForPointLookup(): move away from hash-based block-based-table index, and use whole key memtable filtering.
 * Change the behavior of OptimizeForSmallDb(): use a 16MB block cache, put index and filter blocks into it, and cost the memtable size to it. DBOptions.OptimizeForSmallDb() and ColumnFamilyOptions.OptimizeForSmallDb() start to take an optional cache object.
@@ -17,6 +34,8 @@
 * Adjust WriteBufferManager's dummy entry size to block cache from 1MB to 256KB.
 * Fix a race condition between WritePrepared::Get and ::Put with duplicate keys.
 * Fix crash when memtable prefix bloom is enabled and read/write a key out of domain of prefix extractor.
+* Close a WAL file before another thread deletes it.
+* Fix an assertion failure `IsFlushPending() == true` caused by one bg thread releasing the db mutex in ~ColumnFamilyData and another thread clearing `flush_requested_` flag.

 ## 6.1.1 (4/9/2019)
 ### New Features
@@ -137,6 +137,12 @@ CFLAGS +=  -DHAVE_POWER8
 HAVE_POWER8=1
 endif

+ifeq (,$(shell $(CXX) -fsyntax-only -march=armv8-a+crc -xc /dev/null 2>&1))
+CXXFLAGS += -march=armv8-a+crc
+CFLAGS += -march=armv8-a+crc
+ARMCRC_SOURCE=1
+endif
+
 # if we're compiling for release, compile without debug code (-DNDEBUG)
 ifeq ($(DEBUG_LEVEL),0)
 OPT += -DNDEBUG
@@ -56,10 +56,10 @@ if [ -z "$ROCKSDB_NO_FBCODE" -a -d /mnt/gvfs/third-party ]; then
    if [ -n "$ROCKSDB_FBCODE_BUILD_WITH_481" ]; then
      # we need this to build with MySQL. Don't use for other purposes.
      source "$PWD/build_tools/fbcode_config4.8.1.sh"
-    elif [ -n "$ROCKSDB_FBCODE_BUILD_WITH_PLATFORM007" ]; then
-      source "$PWD/build_tools/fbcode_config_platform007.sh"
-    else
+    elif [ -n "$ROCKSDB_FBCODE_BUILD_WITH_5xx" ]; then
      source "$PWD/build_tools/fbcode_config.sh"
+    else
+      source "$PWD/build_tools/fbcode_config_platform007.sh"
    fi
 fi

@@ -1070,6 +1070,18 @@ int rocksdb_property_int(
  }
 }

+int rocksdb_property_int_cf(
+    rocksdb_t* db,
+    rocksdb_column_family_handle_t* column_family,
+    const char* propname,
+    uint64_t *out_val) {
+  if (db->rep->GetIntProperty(column_family->rep, Slice(propname), out_val)) {
+    return 0;
+  } else {
+    return -1;
+  }
+}
+
 char* rocksdb_property_value_cf(
    rocksdb_t* db,
    rocksdb_column_family_handle_t* column_family,
@@ -1183,6 +1195,14 @@ void rocksdb_flush(
  SaveError(errptr, db->rep->Flush(options->rep));
 }

+void rocksdb_flush_cf(
+    rocksdb_t* db,
+    const rocksdb_flushoptions_t* options,
+    rocksdb_column_family_handle_t* column_family,
+    char** errptr) {
+  SaveError(errptr, db->rep->Flush(options->rep, column_family->rep));
+}
+
 void rocksdb_disable_file_deletions(
    rocksdb_t* db,
    char** errptr) {
@@ -2206,6 +2226,11 @@ void rocksdb_options_set_max_bytes_for_level_base(
  opt->rep.max_bytes_for_level_base = n;
 }

+void rocksdb_options_set_snap_refresh_nanos(rocksdb_options_t* opt,
+                                            uint64_t n) {
+  opt->rep.snap_refresh_nanos = n;
+}
+
 void rocksdb_options_set_level_compaction_dynamic_level_bytes(
    rocksdb_options_t* opt, unsigned char v) {
  opt->rep.level_compaction_dynamic_level_bytes = v;
@@ -1188,6 +1188,12 @@ int main(int argc, char** argv) {
    rocksdb_put_cf(db, woptions, handles[1], "foo", 3, "hello", 5, &err);
    CheckNoError(err);

+    rocksdb_flushoptions_t *flush_options = rocksdb_flushoptions_create();
+    rocksdb_flushoptions_set_wait(flush_options, 1);
+    rocksdb_flush_cf(db, flush_options, handles[1], &err);
+    CheckNoError(err)
+    rocksdb_flushoptions_destroy(flush_options);
+
    CheckGetCF(db, roptions, handles[1], "foo", "hello");
    CheckPinGetCF(db, roptions, handles[1], "foo", "hello");

@@ -63,9 +63,14 @@ ColumnFamilyHandleImpl::~ColumnFamilyHandleImpl() {
    JobContext job_context(0);
    mutex_->Lock();
    if (cfd_->Unref()) {
+      bool dropped = cfd_->IsDropped();
+
      delete cfd_;
+
+      if (dropped) {
+        db_->FindObsoleteFiles(&job_context, false, true);
+      }
    }
-    db_->FindObsoleteFiles(&job_context, false, true);
    mutex_->Unlock();
    if (job_context.HaveSomethingToDelete()) {
      bool defer_purge =
@@ -38,14 +38,16 @@ CompactionIterator::CompactionIterator(
    CompactionRangeDelAggregator* range_del_agg, const Compaction* compaction,
    const CompactionFilter* compaction_filter,
    const std::atomic<bool>* shutting_down,
-    const SequenceNumber preserve_deletes_seqnum)
+    const SequenceNumber preserve_deletes_seqnum,
+    SnapshotListFetchCallback* snap_list_callback)
    : CompactionIterator(
          input, cmp, merge_helper, last_sequence, snapshots,
          earliest_write_conflict_snapshot, snapshot_checker, env,
          report_detailed_time, expect_valid_internal_key, range_del_agg,
          std::unique_ptr<CompactionProxy>(
              compaction ? new CompactionProxy(compaction) : nullptr),
-          compaction_filter, shutting_down, preserve_deletes_seqnum) {}
+          compaction_filter, shutting_down, preserve_deletes_seqnum,
+          snap_list_callback) {}

 CompactionIterator::CompactionIterator(
    InternalIterator* input, const Comparator* cmp, MergeHelper* merge_helper,
@@ -57,7 +59,8 @@ CompactionIterator::CompactionIterator(
    std::unique_ptr<CompactionProxy> compaction,
    const CompactionFilter* compaction_filter,
    const std::atomic<bool>* shutting_down,
-    const SequenceNumber preserve_deletes_seqnum)
+    const SequenceNumber preserve_deletes_seqnum,
+    SnapshotListFetchCallback* snap_list_callback)
    : input_(input),
      cmp_(cmp),
      merge_helper_(merge_helper),
@@ -75,7 +78,8 @@ CompactionIterator::CompactionIterator(
      current_user_key_sequence_(0),
      current_user_key_snapshot_(0),
      merge_out_iter_(merge_helper_),
-      current_key_committed_(false) {
+      current_key_committed_(false),
+      snap_list_callback_(snap_list_callback) {
  assert(compaction_filter_ == nullptr || compaction_ != nullptr);
  assert(snapshots_ != nullptr);
  bottommost_level_ =
@@ -83,24 +87,7 @@ CompactionIterator::CompactionIterator(
  if (compaction_ != nullptr) {
    level_ptrs_ = std::vector<size_t>(compaction_->number_levels(), 0);
  }
-  if (snapshots_->size() == 0) {
-    // optimize for fast path if there are no snapshots
-    visible_at_tip_ = true;
-    earliest_snapshot_iter_ = snapshots_->end();
-    earliest_snapshot_ = kMaxSequenceNumber;
-    latest_snapshot_ = 0;
-  } else {
-    visible_at_tip_ = false;
-    earliest_snapshot_iter_ = snapshots_->begin();
-    earliest_snapshot_ = snapshots_->at(0);
-    latest_snapshot_ = snapshots_->back();
-  }
-#ifndef NDEBUG
-  // findEarliestVisibleSnapshot assumes this ordering.
-  for (size_t i = 1; i < snapshots_->size(); ++i) {
-    assert(snapshots_->at(i - 1) < snapshots_->at(i));
-  }
-#endif
+  ProcessSnapshotList();
  input_->SetPinnedItersMgr(&pinned_iters_mgr_);
  TEST_SYNC_POINT_CALLBACK("CompactionIterator:AfterInit", compaction_.get());
 }
@@ -222,6 +209,28 @@ void CompactionIterator::InvokeFilterIfNeeded(bool* need_skip,
  }
 }

+void CompactionIterator::ProcessSnapshotList() {
+#ifndef NDEBUG
+  // findEarliestVisibleSnapshot assumes this ordering.
+  for (size_t i = 1; i < snapshots_->size(); ++i) {
+    assert(snapshots_->at(i - 1) < snapshots_->at(i));
+  }
+#endif
+  if (snapshots_->size() == 0) {
+    // optimize for fast path if there are no snapshots
+    visible_at_tip_ = true;
+    earliest_snapshot_iter_ = snapshots_->end();
+    earliest_snapshot_ = kMaxSequenceNumber;
+    latest_snapshot_ = 0;
+  } else {
+    visible_at_tip_ = false;
+    earliest_snapshot_iter_ = snapshots_->begin();
+    earliest_snapshot_ = snapshots_->at(0);
+    latest_snapshot_ = snapshots_->back();
+  }
+  released_snapshots_.clear();
+}
+
 void CompactionIterator::NextFromInput() {
  at_next_ = false;
  valid_ = false;
@@ -269,6 +278,13 @@ void CompactionIterator::NextFromInput() {
    // compaction filter). ikey_.user_key is pointing to the copy.
    if (!has_current_user_key_ ||
        !cmp_->Equal(ikey_.user_key, current_user_key_)) {
+      num_keys_++;
+      // Use num_keys_ to reduce the overhead of reading current time
+      if (snap_list_callback_ && snapshots_->size() &&
+          snap_list_callback_->TimeToRefresh(num_keys_)) {
+        snap_list_callback_->Refresh(snapshots_, latest_snapshot_);
+        ProcessSnapshotList();
+      }
      // First occurrence of this user key
      // Copy key for output
      key_ = current_key_.SetInternalKey(key_, &ikey_);
@@ -21,6 +21,54 @@

 namespace rocksdb {

+// This callback can be used to refresh the snapshot list from the db. It
+// includes logics to exponentially decrease the refresh rate to limit the
+// overhead of refresh.
+class SnapshotListFetchCallback {
+ public:
+  SnapshotListFetchCallback(Env* env, uint64_t snap_refresh_nanos,
+                            size_t every_nth_key = 1024)
+      : timer_(env, /*auto restart*/ true),
+        snap_refresh_nanos_(snap_refresh_nanos),
+        every_nth_key_minus_one_(every_nth_key - 1) {
+    assert(every_nth_key > 0);
+    assert((ceil(log2(every_nth_key)) == floor(log2(every_nth_key))));
+  }
+  // Refresh the snapshot list. snapshots will bre replacted with the new list.
+  // max is the upper bound. Note: this function will acquire the db_mutex_.
+  virtual void Refresh(std::vector<SequenceNumber>* snapshots,
+                       SequenceNumber max) = 0;
+  inline bool TimeToRefresh(const size_t key_index) {
+    assert(snap_refresh_nanos_ != 0);
+    // skip the key if key_index % every_nth_key (which is of power 2) is not 0.
+    if ((key_index & every_nth_key_minus_one_) != 0) {
+      return false;
+    }
+    const uint64_t elapsed = timer_.ElapsedNanos();
+    auto ret = elapsed > snap_refresh_nanos_;
+    // pre-compute the next time threshold
+    if (ret) {
+      // inc next refresh period exponentially (by x4)
+      auto next_refresh_threshold = snap_refresh_nanos_ << 2;
+      // make sure the shift has not overflown the highest 1 bit
+      snap_refresh_nanos_ =
+          std::max(snap_refresh_nanos_, next_refresh_threshold);
+    }
+    return ret;
+  }
+  static constexpr SnapshotListFetchCallback* kDisabled = nullptr;
+
+  virtual ~SnapshotListFetchCallback() {}
+
+ private:
+  // Time since the callback was created
+  StopWatchNano timer_;
+  // The delay before calling ::Refresh. To be increased exponentially.
+  uint64_t snap_refresh_nanos_;
+  // Skip evey nth key. Number n if of power 2. The math will require n-1.
+  const uint64_t every_nth_key_minus_one_;
+};
+
 class CompactionIterator {
 public:
  // A wrapper around Compaction. Has a much smaller interface, only what
@@ -69,7 +117,8 @@ class CompactionIterator {
                     const Compaction* compaction = nullptr,
                     const CompactionFilter* compaction_filter = nullptr,
                     const std::atomic<bool>* shutting_down = nullptr,
-                     const SequenceNumber preserve_deletes_seqnum = 0);
+                     const SequenceNumber preserve_deletes_seqnum = 0,
+                     SnapshotListFetchCallback* snap_list_callback = nullptr);

  // Constructor with custom CompactionProxy, used for tests.
  CompactionIterator(InternalIterator* input, const Comparator* cmp,
@@ -82,7 +131,8 @@ class CompactionIterator {
                     std::unique_ptr<CompactionProxy> compaction,
                     const CompactionFilter* compaction_filter = nullptr,
                     const std::atomic<bool>* shutting_down = nullptr,
-                     const SequenceNumber preserve_deletes_seqnum = 0);
+                     const SequenceNumber preserve_deletes_seqnum = 0,
+                     SnapshotListFetchCallback* snap_list_callback = nullptr);

  ~CompactionIterator();

@@ -110,6 +160,8 @@ class CompactionIterator {
 private:
  // Processes the input stream to find the next output
  void NextFromInput();
+  // Process snapshots_ and assign related variables
+  void ProcessSnapshotList();

  // Do last preparations before presenting the output to the callee. At this
  // point this only zeroes out the sequence number if possible for better
@@ -144,7 +196,7 @@ class CompactionIterator {
  InternalIterator* input_;
  const Comparator* cmp_;
  MergeHelper* merge_helper_;
-  const std::vector<SequenceNumber>* snapshots_;
+  std::vector<SequenceNumber>* snapshots_;
  // List of snapshots released during compaction.
  // findEarliestVisibleSnapshot() find them out from return of
  // snapshot_checker, and make sure they will not be returned as
@@ -219,6 +271,9 @@ class CompactionIterator {
  // Used to avoid purging uncommitted values. The application can specify
  // uncommitted values by providing a SnapshotChecker object.
  bool current_key_committed_;
+  SnapshotListFetchCallback* snap_list_callback_;
+  // number of distinct keys processed
+  size_t num_keys_ = 0;

  bool IsShuttingDown() {
    // This is a best-effort facility, so memory_order_relaxed is sufficient.
@@ -315,7 +315,7 @@ CompactionJob::CompactionJob(
    const SnapshotChecker* snapshot_checker, std::shared_ptr<Cache> table_cache,
    EventLogger* event_logger, bool paranoid_file_checks, bool measure_io_stats,
    const std::string& dbname, CompactionJobStats* compaction_job_stats,
-    Env::Priority thread_pri)
+    Env::Priority thread_pri, SnapshotListFetchCallback* snap_list_callback)
    : job_id_(job_id),
      compact_(new CompactionState(compaction)),
      compaction_job_stats_(compaction_job_stats),
@@ -336,6 +336,7 @@ CompactionJob::CompactionJob(
      db_mutex_(db_mutex),
      db_error_handler_(db_error_handler),
      existing_snapshots_(std::move(existing_snapshots)),
+      snap_list_callback_(snap_list_callback),
      earliest_write_conflict_snapshot_(earliest_write_conflict_snapshot),
      snapshot_checker_(snapshot_checker),
      table_cache_(std::move(table_cache)),
@@ -892,7 +893,9 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
      &existing_snapshots_, earliest_write_conflict_snapshot_,
      snapshot_checker_, env_, ShouldReportDetailedTime(env_, stats_), false,
      &range_del_agg, sub_compact->compaction, compaction_filter,
-      shutting_down_, preserve_deletes_seqnum_));
+      shutting_down_, preserve_deletes_seqnum_,
+      // Currently range_del_agg is incompatible with snapshot refresh feature.
+      range_del_agg.IsEmpty() ? snap_list_callback_ : nullptr));
  auto c_iter = sub_compact->c_iter.get();
  c_iter->SeekToFirst();
  if (c_iter->Valid() && sub_compact->compaction->output_level() != 0) {
@@ -57,22 +57,20 @@ class VersionSet;

 class CompactionJob {
 public:
-  CompactionJob(int job_id, Compaction* compaction,
-                const ImmutableDBOptions& db_options,
-                const EnvOptions env_options, VersionSet* versions,
-                const std::atomic<bool>* shutting_down,
-                const SequenceNumber preserve_deletes_seqnum,
-                LogBuffer* log_buffer, Directory* db_directory,
-                Directory* output_directory, Statistics* stats,
-                InstrumentedMutex* db_mutex, ErrorHandler* db_error_handler,
-                std::vector<SequenceNumber> existing_snapshots,
-                SequenceNumber earliest_write_conflict_snapshot,
-                const SnapshotChecker* snapshot_checker,
-                std::shared_ptr<Cache> table_cache, EventLogger* event_logger,
-                bool paranoid_file_checks, bool measure_io_stats,
-                const std::string& dbname,
-                CompactionJobStats* compaction_job_stats,
-                Env::Priority thread_pri);
+  CompactionJob(
+      int job_id, Compaction* compaction, const ImmutableDBOptions& db_options,
+      const EnvOptions env_options, VersionSet* versions,
+      const std::atomic<bool>* shutting_down,
+      const SequenceNumber preserve_deletes_seqnum, LogBuffer* log_buffer,
+      Directory* db_directory, Directory* output_directory, Statistics* stats,
+      InstrumentedMutex* db_mutex, ErrorHandler* db_error_handler,
+      std::vector<SequenceNumber> existing_snapshots,
+      SequenceNumber earliest_write_conflict_snapshot,
+      const SnapshotChecker* snapshot_checker,
+      std::shared_ptr<Cache> table_cache, EventLogger* event_logger,
+      bool paranoid_file_checks, bool measure_io_stats,
+      const std::string& dbname, CompactionJobStats* compaction_job_stats,
+      Env::Priority thread_pri, SnapshotListFetchCallback* snap_list_callback);

  ~CompactionJob();

@@ -152,6 +150,7 @@ class CompactionJob {
  // entirely within s1 and s2, then the earlier version of k1 can be safely
  // deleted because that version is not visible in any snapshot.
  std::vector<SequenceNumber> existing_snapshots_;
+  SnapshotListFetchCallback* snap_list_callback_;

  // This is the earliest snapshot that could be used for write-conflict
  // checking by a transaction.  For any user-key newer than this snapshot, we
@@ -5,7 +5,13 @@

 #ifndef ROCKSDB_LITE

+#ifndef __STDC_FORMAT_MACROS
+#define __STDC_FORMAT_MACROS
+#endif
+
+#include <inttypes.h>
 #include <algorithm>
+#include <array>
 #include <map>
 #include <string>
 #include <tuple>
@@ -194,6 +200,13 @@ class CompactionJobTest : public testing::Test {
  }

  void NewDB() {
+    DestroyDB(dbname_, Options());
+    EXPECT_OK(env_->CreateDirIfMissing(dbname_));
+    versions_.reset(new VersionSet(dbname_, &db_options_, env_options_,
+                                   table_cache_.get(), &write_buffer_manager_,
+                                   &write_controller_));
+    compaction_job_stats_.Reset();
+
    VersionEdit new_db;
    new_db.SetLogNumber(0);
    new_db.SetNextFile(2);
@@ -230,7 +243,10 @@ class CompactionJobTest : public testing::Test {
      const std::vector<std::vector<FileMetaData*>>& input_files,
      const stl_wrappers::KVMap& expected_results,
      const std::vector<SequenceNumber>& snapshots = {},
-      SequenceNumber earliest_write_conflict_snapshot = kMaxSequenceNumber) {
+      SequenceNumber earliest_write_conflict_snapshot = kMaxSequenceNumber,
+      int output_level = 1, bool verify = true,
+      SnapshotListFetchCallback* snapshot_fetcher =
+          SnapshotListFetchCallback::kDisabled) {
    auto cfd = versions_->GetColumnFamilySet()->GetDefault();

    size_t num_input_files = 0;
@@ -247,7 +263,7 @@ class CompactionJobTest : public testing::Test {

    Compaction compaction(cfd->current()->storage_info(), *cfd->ioptions(),
                          *cfd->GetLatestMutableCFOptions(),
-                          compaction_input_files, 1, 1024 * 1024,
+                          compaction_input_files, output_level, 1024 * 1024,
                          10 * 1024 * 1024, 0, kNoCompression,
                          cfd->ioptions()->compression_opts, 0, {}, true);
    compaction.SetInputVersion(cfd->current());
@@ -263,7 +279,7 @@ class CompactionJobTest : public testing::Test {
        nullptr, nullptr, &mutex_, &error_handler_, snapshots,
        earliest_write_conflict_snapshot, snapshot_checker, table_cache_,
        &event_logger, false, false, dbname_, &compaction_job_stats_,
-        Env::Priority::USER);
+        Env::Priority::USER, snapshot_fetcher);
    VerifyInitializationOfCompactionJobStats(compaction_job_stats_);

    compaction_job.Prepare();
@@ -275,15 +291,17 @@ class CompactionJobTest : public testing::Test {
    ASSERT_OK(compaction_job.Install(*cfd->GetLatestMutableCFOptions()));
    mutex_.Unlock();

-    if (expected_results.size() == 0) {
-      ASSERT_GE(compaction_job_stats_.elapsed_micros, 0U);
-      ASSERT_EQ(compaction_job_stats_.num_input_files, num_input_files);
-      ASSERT_EQ(compaction_job_stats_.num_output_files, 0U);
-    } else {
-      ASSERT_GE(compaction_job_stats_.elapsed_micros, 0U);
-      ASSERT_EQ(compaction_job_stats_.num_input_files, num_input_files);
-      ASSERT_EQ(compaction_job_stats_.num_output_files, 1U);
-      mock_table_factory_->AssertLatestFile(expected_results);
+    if (verify) {
+      if (expected_results.size() == 0) {
+        ASSERT_GE(compaction_job_stats_.elapsed_micros, 0U);
+        ASSERT_EQ(compaction_job_stats_.num_input_files, num_input_files);
+        ASSERT_EQ(compaction_job_stats_.num_output_files, 0U);
+      } else {
+        ASSERT_GE(compaction_job_stats_.elapsed_micros, 0U);
+        ASSERT_EQ(compaction_job_stats_.num_input_files, num_input_files);
+        ASSERT_EQ(compaction_job_stats_.num_output_files, 1U);
+        mock_table_factory_->AssertLatestFile(expected_results);
+      }
    }
  }

@@ -938,6 +956,105 @@ TEST_F(CompactionJobTest, CorruptionAfterDeletion) {
  RunCompaction({files}, expected_results);
 }

+// Test the snapshot fetcher in compaction
+TEST_F(CompactionJobTest, SnapshotRefresh) {
+  uint64_t time_seed = env_->NowMicros();
+  printf("time_seed is %" PRIu64 "\n", time_seed);  // would help to reproduce
+  Random64 rand(time_seed);
+  std::vector<SequenceNumber> db_snapshots;
+  class SnapshotListFetchCallbackTest : public SnapshotListFetchCallback {
+   public:
+    SnapshotListFetchCallbackTest(Env* env, Random64& rand,
+                                  std::vector<SequenceNumber>* snapshots)
+        : SnapshotListFetchCallback(env, 1 /*short time delay*/,
+                                    1 /*fetch after each key*/),
+          rand_(rand),
+          snapshots_(snapshots) {}
+    virtual void Refresh(std::vector<SequenceNumber>* snapshots,
+                         SequenceNumber) override {
+      assert(snapshots->size());
+      assert(snapshots_->size());
+      assert(snapshots_->size() == snapshots->size());
+      if (rand_.OneIn(2)) {
+        uint64_t release_index = rand_.Uniform(snapshots_->size());
+        snapshots_->erase(snapshots_->begin() + release_index);
+        *snapshots = *snapshots_;
+      }
+    }
+
+   private:
+    Random64 rand_;
+    std::vector<SequenceNumber>* snapshots_;
+  } snapshot_fetcher(env_, rand, &db_snapshots);
+
+  std::vector<std::pair<const std::string, std::string>> file1_kvs, file2_kvs;
+  std::array<ValueType, 4> types = {kTypeValue, kTypeDeletion,
+                                    kTypeSingleDeletion};
+  SequenceNumber last_seq = 0;
+  for (int i = 1; i < 100; i++) {
+    SequenceNumber seq = last_seq + 1;
+    last_seq = seq;
+    if (rand.OneIn(2)) {
+      auto type = types[rand.Uniform(types.size())];
+      file1_kvs.push_back(
+          {test::KeyStr("k" + ToString(i), seq, type), "v" + ToString(i)});
+    }
+  }
+  auto file1 = mock::MakeMockFile(file1_kvs);
+  for (int i = 1; i < 100; i++) {
+    SequenceNumber seq = last_seq + 1;
+    last_seq++;
+    if (rand.OneIn(2)) {
+      auto type = types[rand.Uniform(types.size())];
+      file2_kvs.push_back(
+          {test::KeyStr("k" + ToString(i), seq, type), "v" + ToString(i)});
+    }
+  }
+  auto file2 = mock::MakeMockFile(file2_kvs);
+  for (SequenceNumber i = 1; i < last_seq + 1; i++) {
+    if (rand.OneIn(5)) {
+      db_snapshots.push_back(i);
+    }
+  }
+
+  const bool kVerify = true;
+  const int output_level_0 = 0;
+  NewDB();
+  AddMockFile(file1);
+  AddMockFile(file2);
+  SetLastSequence(last_seq);
+  auto files = cfd_->current()->storage_info()->LevelFiles(0);
+  // put the output on L0 since it is easier to feed them again to the 2nd
+  // compaction
+  RunCompaction({files}, file1, db_snapshots, kMaxSequenceNumber,
+                output_level_0, !kVerify, &snapshot_fetcher);
+
+  // Now db_snapshots are changed. Run the compaction again without snapshot
+  // fetcher but with the updated snapshot list.
+  compaction_job_stats_.Reset();
+  files = cfd_->current()->storage_info()->LevelFiles(0);
+  RunCompaction({files}, file1, db_snapshots, kMaxSequenceNumber,
+                output_level_0 + 1, !kVerify);
+  // The result should be what we get if we run compaction without snapshot
+  // fetcher on the updated list of snapshots
+  auto expected = mock_table_factory_->output();
+
+  NewDB();
+  AddMockFile(file1);
+  AddMockFile(file2);
+  SetLastSequence(last_seq);
+  files = cfd_->current()->storage_info()->LevelFiles(0);
+  RunCompaction({files}, expected, db_snapshots, kMaxSequenceNumber,
+                output_level_0, !kVerify);
+  // The 2nd compaction above would get rid of useless delete markers. To get
+  // the output here exactly as what we got above after two compactions, we also
+  // run the compaction for 2nd time.
+  compaction_job_stats_.Reset();
+  files = cfd_->current()->storage_info()->LevelFiles(0);
+  RunCompaction({files}, expected, db_snapshots, kMaxSequenceNumber,
+                output_level_0 + 1, !kVerify);
+}
+
 }  // namespace rocksdb

 int main(int argc, char** argv) {
@@ -488,6 +488,63 @@ TEST_P(DBAtomicFlushTest,
  Destroy(options);
 }

+TEST_P(DBAtomicFlushTest, TriggerFlushAndClose) {
+  bool atomic_flush = GetParam();
+  if (!atomic_flush) {
+    return;
+  }
+  const int kNumKeysTriggerFlush = 4;
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.atomic_flush = atomic_flush;
+  options.memtable_factory.reset(
+      new SpecialSkipListFactory(kNumKeysTriggerFlush));
+  CreateAndReopenWithCF({"pikachu"}, options);
+
+  for (int i = 0; i != kNumKeysTriggerFlush; ++i) {
+    ASSERT_OK(Put(0, "key" + std::to_string(i), "value" + std::to_string(i)));
+  }
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->EnableProcessing();
+  ASSERT_OK(Put(0, "key", "value"));
+  Close();
+
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu"}, options);
+  ASSERT_EQ("value", Get(0, "key"));
+}
+
+TEST_P(DBAtomicFlushTest, PickMemtablesRaceWithBackgroundFlush) {
+  bool atomic_flush = GetParam();
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.atomic_flush = atomic_flush;
+  options.max_write_buffer_number = 4;
+  // Set min_write_buffer_number_to_merge to be greater than 1, so that
+  // a column family with one memtable in the imm will not cause IsFlushPending
+  // to return true when flush_requested_ is false.
+  options.min_write_buffer_number_to_merge = 2;
+  CreateAndReopenWithCF({"pikachu"}, options);
+  ASSERT_EQ(2, handles_.size());
+  ASSERT_OK(dbfull()->PauseBackgroundWork());
+  ASSERT_OK(Put(0, "key00", "value00"));
+  ASSERT_OK(Put(1, "key10", "value10"));
+  FlushOptions flush_opts;
+  flush_opts.wait = false;
+  ASSERT_OK(dbfull()->Flush(flush_opts, handles_));
+  ASSERT_OK(Put(0, "key01", "value01"));
+  // Since max_write_buffer_number is 4, the following flush won't cause write
+  // stall.
+  ASSERT_OK(dbfull()->Flush(flush_opts));
+  ASSERT_OK(dbfull()->DropColumnFamily(handles_[1]));
+  ASSERT_OK(dbfull()->DestroyColumnFamilyHandle(handles_[1]));
+  handles_[1] = nullptr;
+  ASSERT_OK(dbfull()->ContinueBackgroundWork());
+  ASSERT_OK(dbfull()->TEST_WaitForFlushMemTable(handles_[0]));
+  delete handles_[0];
+  handles_.clear();
+}
+
 INSTANTIATE_TEST_CASE_P(DBFlushDirectIOTest, DBFlushDirectIOTest,
                        testing::Bool());

@@ -1209,7 +1209,15 @@ void DBImpl::BackgroundCallPurge() {
  // both queues are empty. This is stricter than what is needed, but can make
  // it easier for us to reason the correctness.
  while (!purge_queue_.empty() || !logs_to_free_queue_.empty()) {
-    if (!purge_queue_.empty()) {
+    // Check logs_to_free_queue_ first and close log writers.
+    if (!logs_to_free_queue_.empty()) {
+      assert(!logs_to_free_queue_.empty());
+      log::Writer* log_writer = *(logs_to_free_queue_.begin());
+      logs_to_free_queue_.pop_front();
+      mutex_.Unlock();
+      delete log_writer;
+      mutex_.Lock();
+    } else {
      auto purge_file = purge_queue_.begin();
      auto fname = purge_file->fname;
      auto dir_to_sync = purge_file->dir_to_sync;
@@ -1221,13 +1229,6 @@ void DBImpl::BackgroundCallPurge() {
      mutex_.Unlock();
      DeleteObsoleteFileImpl(job_id, fname, dir_to_sync, type, number);
      mutex_.Lock();
-    } else {
-      assert(!logs_to_free_queue_.empty());
-      log::Writer* log_writer = *(logs_to_free_queue_.begin());
-      logs_to_free_queue_.pop_front();
-      mutex_.Unlock();
-      delete log_writer;
-      mutex_.Lock();
    }
  }
  bg_purge_scheduled_--;
@@ -3537,9 +3538,9 @@ Status DBImpl::IngestExternalFiles(
    exec_results.emplace_back(false, Status::OK());
  }
  // TODO(yanqin) maybe make jobs run in parallel
+  uint64_t start_file_number = next_file_number;
  for (size_t i = 1; i != num_cfs; ++i) {
-    uint64_t start_file_number =
-        next_file_number + args[i - 1].external_files.size();
+    start_file_number += args[i - 1].external_files.size();
    auto* cfd =
        static_cast<ColumnFamilyHandleImpl*>(args[i].column_family)->cfd();
    SuperVersion* super_version = cfd->GetReferencedSuperVersion(&mutex_);
@@ -561,6 +561,13 @@ class DBImpl : public DB {

  const SnapshotList& snapshots() const { return snapshots_; }

+  void LoadSnapshots(std::vector<SequenceNumber>* snap_vector,
+                     SequenceNumber* oldest_write_conflict_snapshot,
+                     const SequenceNumber& max_seq) const {
+    InstrumentedMutexLock l(mutex());
+    snapshots().GetAll(snap_vector, oldest_write_conflict_snapshot, max_seq);
+  }
+
  const ImmutableDBOptions& immutable_db_options() const {
    return immutable_db_options_;
  }
@@ -739,7 +746,7 @@ class DBImpl : public DB {
  // Not thread-safe.
  void SetRecoverableStatePreReleaseCallback(PreReleaseCallback* callback);

-  InstrumentedMutex* mutex() { return &mutex_; }
+  InstrumentedMutex* mutex() const { return &mutex_; }

  Status NewDB();

@@ -397,12 +397,21 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
    s = error_status.ok() ? s : error_status;
  }

+  // If db is NOT shutting down, and one or more column families have been
+  // dropped.
+  // TODO: use separate status code for db shutdown and column family dropped.
+  if (s.IsShutdownInProgress() &&
+      !shutting_down_.load(std::memory_order_acquire)) {
+    s = Status::OK();
+  }
+
  if (s.ok() || s.IsShutdownInProgress()) {
    // Sync on all distinct output directories.
    for (auto dir : distinct_output_dirs) {
      if (dir != nullptr) {
-        s = dir->Fsync();
-        if (!s.ok()) {
+        Status error_status = dir->Fsync();
+        if (!error_status.ok()) {
+          s = error_status;
          break;
        }
      }
@@ -469,7 +478,7 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
        &job_context->memtables_to_free, directories_.GetDbDir(), log_buffer);
  }

-  if (s.ok() || s.IsShutdownInProgress()) {
+  if (s.ok()) {
    assert(num_cfs ==
           static_cast<int>(job_context->superversion_contexts.size()));
    for (int i = 0; i != num_cfs; ++i) {
@@ -789,6 +798,31 @@ Status DBImpl::CompactRange(const CompactRangeOptions& options,
  return s;
 }

+namespace {
+class SnapshotListFetchCallbackImpl : public SnapshotListFetchCallback {
+ public:
+  SnapshotListFetchCallbackImpl(DBImpl* db_impl, Env* env,
+                                uint64_t snap_refresh_nanos, Logger* info_log)
+      : SnapshotListFetchCallback(env, snap_refresh_nanos),
+        db_impl_(db_impl),
+        info_log_(info_log) {}
+  virtual void Refresh(std::vector<SequenceNumber>* snapshots,
+                       SequenceNumber max) override {
+    size_t prev = snapshots->size();
+    snapshots->clear();
+    db_impl_->LoadSnapshots(snapshots, nullptr, max);
+    size_t now = snapshots->size();
+    ROCKS_LOG_DEBUG(info_log_,
+                    "Compaction snapshot count refreshed from %zu to %zu", prev,
+                    now);
+  }
+
+ private:
+  DBImpl* db_impl_;
+  Logger* info_log_;
+};
+}  // namespace
+
 Status DBImpl::CompactFiles(const CompactionOptions& compact_options,
                            ColumnFamilyHandle* column_family,
                            const std::vector<std::string>& input_file_names,
@@ -960,6 +994,9 @@ Status DBImpl::CompactFilesImpl(

  assert(is_snapshot_supported_ || snapshots_.empty());
  CompactionJobStats compaction_job_stats;
+  SnapshotListFetchCallbackImpl fetch_callback(
+      this, env_, c->mutable_cf_options()->snap_refresh_nanos,
+      immutable_db_options_.info_log.get());
  CompactionJob compaction_job(
      job_context->job_id, c.get(), immutable_db_options_,
      env_options_for_compaction_, versions_.get(), &shutting_down_,
@@ -969,7 +1006,11 @@ Status DBImpl::CompactFilesImpl(
      snapshot_checker, table_cache_, &event_logger_,
      c->mutable_cf_options()->paranoid_file_checks,
      c->mutable_cf_options()->report_bg_io_stats, dbname_,
-      &compaction_job_stats, Env::Priority::USER);
+      &compaction_job_stats, Env::Priority::USER,
+      immutable_db_options_.max_subcompactions <= 1 &&
+              c->mutable_cf_options()->snap_refresh_nanos > 0
+          ? &fetch_callback
+          : nullptr);

  // Creating a compaction influences the compaction score because the score
  // takes running compactions into account (by skipping files that are already
@@ -2043,6 +2084,7 @@ Status DBImpl::BackgroundFlush(bool* made_progress, JobContext* job_context,
  autovector<BGFlushArg> bg_flush_args;
  std::vector<SuperVersionContext>& superversion_contexts =
      job_context->superversion_contexts;
+  autovector<ColumnFamilyData*> column_families_not_to_flush;
  while (!flush_queue_.empty()) {
    // This cfd is already referenced
    const FlushRequest& flush_req = PopFirstFromFlushQueue();
@@ -2053,9 +2095,7 @@ Status DBImpl::BackgroundFlush(bool* made_progress, JobContext* job_context,
      ColumnFamilyData* cfd = iter.first;
      if (cfd->IsDropped() || !cfd->imm()->IsFlushPending()) {
        // can't flush this CF, try next one
-        if (cfd->Unref()) {
-          delete cfd;
-        }
+        column_families_not_to_flush.push_back(cfd);
        continue;
      }
      superversion_contexts.emplace_back(SuperVersionContext(true));
@@ -2094,6 +2134,11 @@ Status DBImpl::BackgroundFlush(bool* made_progress, JobContext* job_context,
      }
    }
  }
+  for (auto cfd : column_families_not_to_flush) {
+    if (cfd->Unref()) {
+      delete cfd;
+    }
+  }
  return status;
 }

@@ -2613,6 +2658,9 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
    GetSnapshotContext(job_context, &snapshot_seqs,
                       &earliest_write_conflict_snapshot, &snapshot_checker);
    assert(is_snapshot_supported_ || snapshots_.empty());
+    SnapshotListFetchCallbackImpl fetch_callback(
+        this, env_, c->mutable_cf_options()->snap_refresh_nanos,
+        immutable_db_options_.info_log.get());
    CompactionJob compaction_job(
        job_context->job_id, c.get(), immutable_db_options_,
        env_options_for_compaction_, versions_.get(), &shutting_down_,
@@ -2622,7 +2670,11 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
        earliest_write_conflict_snapshot, snapshot_checker, table_cache_,
        &event_logger_, c->mutable_cf_options()->paranoid_file_checks,
        c->mutable_cf_options()->report_bg_io_stats, dbname_,
-        &compaction_job_stats, thread_pri);
+        &compaction_job_stats, thread_pri,
+        immutable_db_options_.max_subcompactions <= 1 &&
+                c->mutable_cf_options()->snap_refresh_nanos > 0
+            ? &fetch_callback
+            : nullptr);
    compaction_job.Prepare();

    NotifyOnCompactionBegin(c->column_family_data(), c.get(), status,
@@ -481,6 +481,11 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) {
    }
 #endif  // !ROCKSDB_LITE

+    for (const auto w : state.logs_to_free) {
+      // TODO: maybe check the return value of Close.
+      w->Close();
+    }
+
    Status file_deletion_status;
    if (schedule_only) {
      InstrumentedMutexLock guard_lock(&mutex_);
@@ -1943,8 +1943,8 @@ TEST_P(DBIteratorTest, ReadAhead) {
  delete iter;
  int64_t num_file_closes_readahead =
      TestGetTickerCount(options, NO_FILE_CLOSES);
-  ASSERT_EQ(num_file_opens + 3, num_file_opens_readahead);
-  ASSERT_EQ(num_file_closes + 3, num_file_closes_readahead);
+  ASSERT_EQ(num_file_opens, num_file_opens_readahead);
+  ASSERT_EQ(num_file_closes, num_file_closes_readahead);
  ASSERT_GT(bytes_read_readahead, bytes_read);
  ASSERT_GT(bytes_read_readahead, read_options.readahead_size * 3);

@@ -491,6 +491,30 @@ TEST_F(DBRangeDelTest, CompactionRemovesCoveredMergeOperands) {
  ASSERT_EQ(expected, actual);
 }

+TEST_F(DBRangeDelTest, PutDeleteRangeMergeFlush) {
+  // Test the sequence of operations: (1) Put, (2) DeleteRange, (3) Merge, (4)
+  // Flush. The `CompactionIterator` previously had a bug where we forgot to
+  // check for covering range tombstones when processing the (1) Put, causing
+  // it to reappear after the flush.
+  Options opts = CurrentOptions();
+  opts.merge_operator = MergeOperators::CreateUInt64AddOperator();
+  Reopen(opts);
+
+  std::string val;
+  PutFixed64(&val, 1);
+  ASSERT_OK(db_->Put(WriteOptions(), "key", val));
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(),
+                             "key", "key_"));
+  ASSERT_OK(db_->Merge(WriteOptions(), "key", val));
+  ASSERT_OK(db_->Flush(FlushOptions()));
+
+  ReadOptions read_opts;
+  std::string expected, actual;
+  ASSERT_OK(db_->Get(read_opts, "key", &actual));
+  PutFixed64(&expected, 1);
+  ASSERT_EQ(expected, actual);
+}
+
 // NumTableFilesAtLevel() is not supported in ROCKSDB_LITE
 #ifndef ROCKSDB_LITE
 TEST_F(DBRangeDelTest, ObsoleteTombstoneCleanup) {
@@ -438,6 +438,12 @@ class SpecialEnv : public EnvWrapper {
        return s;
      }

+      virtual Status Prefetch(uint64_t offset, size_t n) override {
+        Status s = target_->Prefetch(offset, n);
+        *bytes_read_ += n;
+        return s;
+      }
+
     private:
      std::unique_ptr<RandomAccessFile> target_;
      anon::AtomicCounter* counter_;
@@ -124,7 +124,7 @@ Status ExternalSstFileIngestionJob::Prepare(
    // We failed, remove all files that we copied into the db
    for (IngestedFileInfo& f : files_to_ingest_) {
      if (f.internal_file_path.empty()) {
-        break;
+        continue;
      }
      Status s = env_->DeleteFile(f.internal_file_path);
      if (!s.ok()) {
@@ -255,6 +255,9 @@ void ExternalSstFileIngestionJob::Cleanup(const Status& status) {
    // We failed to add the files to the database
    // remove all the files we copied
    for (IngestedFileInfo& f : files_to_ingest_) {
+      if (f.internal_file_path.empty()) {
+        continue;
+      }
      Status s = env_->DeleteFile(f.internal_file_path);
      if (!s.ok()) {
        ROCKS_LOG_WARN(db_options_.info_log,
@@ -2300,10 +2300,11 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_Success) {
      new FaultInjectionTestEnv(env_));
  Options options = CurrentOptions();
  options.env = fault_injection_env.get();
-  CreateAndReopenWithCF({"pikachu"}, options);
+  CreateAndReopenWithCF({"pikachu", "eevee"}, options);
  std::vector<ColumnFamilyHandle*> column_families;
  column_families.push_back(handles_[0]);
  column_families.push_back(handles_[1]);
+  column_families.push_back(handles_[2]);
  std::vector<IngestExternalFileOptions> ifos(column_families.size());
  for (auto& ifo : ifos) {
    ifo.allow_global_seqno = true;  // Always allow global_seqno
@@ -2317,6 +2318,9 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_Success) {
      {std::make_pair("foo1", "fv1"), std::make_pair("foo2", "fv2")});
  data.push_back(
      {std::make_pair("bar1", "bv1"), std::make_pair("bar2", "bv2")});
+  data.push_back(
+      {std::make_pair("bar3", "bv3"), std::make_pair("bar4", "bv4")});
+
  // Resize the true_data vector upon construction to avoid re-alloc
  std::vector<std::map<std::string, std::string>> true_data(
      column_families.size());
@@ -2324,8 +2328,9 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_Success) {
                                         -1, true, true_data);
  ASSERT_OK(s);
  Close();
-  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu"}, options);
-  ASSERT_EQ(2, handles_.size());
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu", "eevee"},
+                           options);
+  ASSERT_EQ(3, handles_.size());
  int cf = 0;
  for (const auto& verify_map : true_data) {
    for (const auto& elem : verify_map) {
@@ -2357,10 +2362,11 @@ TEST_P(ExternalSSTFileTest,

  Options options = CurrentOptions();
  options.env = fault_injection_env.get();
-  CreateAndReopenWithCF({"pikachu"}, options);
+  CreateAndReopenWithCF({"pikachu", "eevee"}, options);
  const std::vector<std::map<std::string, std::string>> data_before_ingestion =
      {{{"foo1", "fv1_0"}, {"foo2", "fv2_0"}, {"foo3", "fv3_0"}},
-       {{"bar1", "bv1_0"}, {"bar2", "bv2_0"}, {"bar3", "bv3_0"}}};
+       {{"bar1", "bv1_0"}, {"bar2", "bv2_0"}, {"bar3", "bv3_0"}},
+       {{"bar4", "bv4_0"}, {"bar5", "bv5_0"}, {"bar6", "bv6_0"}}};
  for (size_t i = 0; i != handles_.size(); ++i) {
    int cf = static_cast<int>(i);
    const auto& orig_data = data_before_ingestion[i];
@@ -2373,6 +2379,7 @@ TEST_P(ExternalSSTFileTest,
  std::vector<ColumnFamilyHandle*> column_families;
  column_families.push_back(handles_[0]);
  column_families.push_back(handles_[1]);
+  column_families.push_back(handles_[2]);
  std::vector<IngestExternalFileOptions> ifos(column_families.size());
  for (auto& ifo : ifos) {
    ifo.allow_global_seqno = true;  // Always allow global_seqno
@@ -2386,6 +2393,8 @@ TEST_P(ExternalSSTFileTest,
      {std::make_pair("foo1", "fv1"), std::make_pair("foo2", "fv2")});
  data.push_back(
      {std::make_pair("bar1", "bv1"), std::make_pair("bar2", "bv2")});
+  data.push_back(
+      {std::make_pair("bar3", "bv3"), std::make_pair("bar4", "bv4")});
  // Resize the true_data vector upon construction to avoid re-alloc
  std::vector<std::map<std::string, std::string>> true_data(
      column_families.size());
@@ -2439,10 +2448,11 @@ TEST_P(ExternalSSTFileTest,
  dbfull()->ReleaseSnapshot(read_opts.snapshot);

  Close();
-  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu"}, options);
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu", "eevee"},
+                           options);
  // Should see consistent state after ingestion for all column families even
  // without snapshot.
-  ASSERT_EQ(2, handles_.size());
+  ASSERT_EQ(3, handles_.size());
  int cf = 0;
  for (const auto& verify_map : true_data) {
    for (const auto& elem : verify_map) {
@@ -2472,10 +2482,11 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_PrepareFail) {
       "DBImpl::IngestExternalFiles:BeforeLastJobPrepare:1"},
  });
  SyncPoint::GetInstance()->EnableProcessing();
-  CreateAndReopenWithCF({"pikachu"}, options);
+  CreateAndReopenWithCF({"pikachu", "eevee"}, options);
  std::vector<ColumnFamilyHandle*> column_families;
  column_families.push_back(handles_[0]);
  column_families.push_back(handles_[1]);
+  column_families.push_back(handles_[2]);
  std::vector<IngestExternalFileOptions> ifos(column_families.size());
  for (auto& ifo : ifos) {
    ifo.allow_global_seqno = true;  // Always allow global_seqno
@@ -2489,6 +2500,9 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_PrepareFail) {
      {std::make_pair("foo1", "fv1"), std::make_pair("foo2", "fv2")});
  data.push_back(
      {std::make_pair("bar1", "bv1"), std::make_pair("bar2", "bv2")});
+  data.push_back(
+      {std::make_pair("bar3", "bv3"), std::make_pair("bar4", "bv4")});
+
  // Resize the true_data vector upon construction to avoid re-alloc
  std::vector<std::map<std::string, std::string>> true_data(
      column_families.size());
@@ -2508,8 +2522,9 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_PrepareFail) {

  fault_injection_env->SetFilesystemActive(true);
  Close();
-  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu"}, options);
-  ASSERT_EQ(2, handles_.size());
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu", "eevee"},
+                           options);
+  ASSERT_EQ(3, handles_.size());
  int cf = 0;
  for (const auto& verify_map : true_data) {
    for (const auto& elem : verify_map) {
@@ -2538,10 +2553,11 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_CommitFail) {
       "DBImpl::IngestExternalFiles:BeforeJobsRun:1"},
  });
  SyncPoint::GetInstance()->EnableProcessing();
-  CreateAndReopenWithCF({"pikachu"}, options);
+  CreateAndReopenWithCF({"pikachu", "eevee"}, options);
  std::vector<ColumnFamilyHandle*> column_families;
  column_families.push_back(handles_[0]);
  column_families.push_back(handles_[1]);
+  column_families.push_back(handles_[2]);
  std::vector<IngestExternalFileOptions> ifos(column_families.size());
  for (auto& ifo : ifos) {
    ifo.allow_global_seqno = true;  // Always allow global_seqno
@@ -2555,6 +2571,8 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_CommitFail) {
      {std::make_pair("foo1", "fv1"), std::make_pair("foo2", "fv2")});
  data.push_back(
      {std::make_pair("bar1", "bv1"), std::make_pair("bar2", "bv2")});
+  data.push_back(
+      {std::make_pair("bar3", "bv3"), std::make_pair("bar4", "bv4")});
  // Resize the true_data vector upon construction to avoid re-alloc
  std::vector<std::map<std::string, std::string>> true_data(
      column_families.size());
@@ -2574,8 +2592,9 @@ TEST_P(ExternalSSTFileTest, IngestFilesIntoMultipleColumnFamilies_CommitFail) {

  fault_injection_env->SetFilesystemActive(true);
  Close();
-  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu"}, options);
-  ASSERT_EQ(2, handles_.size());
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu", "eevee"},
+                           options);
+  ASSERT_EQ(3, handles_.size());
  int cf = 0;
  for (const auto& verify_map : true_data) {
    for (const auto& elem : verify_map) {
@@ -2595,7 +2614,7 @@ TEST_P(ExternalSSTFileTest,
  Options options = CurrentOptions();
  options.env = fault_injection_env.get();

-  CreateAndReopenWithCF({"pikachu"}, options);
+  CreateAndReopenWithCF({"pikachu", "eevee"}, options);

  SyncPoint::GetInstance()->ClearTrace();
  SyncPoint::GetInstance()->DisableProcessing();
@@ -2613,6 +2632,7 @@ TEST_P(ExternalSSTFileTest,
  std::vector<ColumnFamilyHandle*> column_families;
  column_families.push_back(handles_[0]);
  column_families.push_back(handles_[1]);
+  column_families.push_back(handles_[2]);
  std::vector<IngestExternalFileOptions> ifos(column_families.size());
  for (auto& ifo : ifos) {
    ifo.allow_global_seqno = true;  // Always allow global_seqno
@@ -2626,6 +2646,8 @@ TEST_P(ExternalSSTFileTest,
      {std::make_pair("foo1", "fv1"), std::make_pair("foo2", "fv2")});
  data.push_back(
      {std::make_pair("bar1", "bv1"), std::make_pair("bar2", "bv2")});
+  data.push_back(
+      {std::make_pair("bar3", "bv3"), std::make_pair("bar4", "bv4")});
  // Resize the true_data vector upon construction to avoid re-alloc
  std::vector<std::map<std::string, std::string>> true_data(
      column_families.size());
@@ -2646,8 +2668,9 @@ TEST_P(ExternalSSTFileTest,
  fault_injection_env->DropUnsyncedFileData();
  fault_injection_env->SetFilesystemActive(true);
  Close();
-  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu"}, options);
-  ASSERT_EQ(2, handles_.size());
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "pikachu", "eevee"},
+                           options);
+  ASSERT_EQ(3, handles_.size());
  int cf = 0;
  for (const auto& verify_map : true_data) {
    for (const auto& elem : verify_map) {
@@ -31,10 +31,23 @@ Writer::Writer(std::unique_ptr<WritableFileWriter>&& dest, uint64_t log_number,
  }
 }

-Writer::~Writer() { WriteBuffer(); }
+Writer::~Writer() {
+  if (dest_) {
+    WriteBuffer();
+  }
+}

 Status Writer::WriteBuffer() { return dest_->Flush(); }

+Status Writer::Close() {
+  Status s;
+  if (dest_) {
+    s = dest_->Close();
+    dest_.reset();
+  }
+  return s;
+}
+
 Status Writer::AddRecord(const Slice& slice) {
  const char* ptr = slice.data();
  size_t left = slice.size();
@@ -84,6 +84,8 @@ class Writer {

  Status WriteBuffer();

+  Status Close();
+
  bool TEST_BufferIsEmpty();

 private:
@@ -277,8 +277,12 @@ void MemTableList::PickMemtablesToFlush(const uint64_t* max_memtable_id,
  AutoThreadOperationStageUpdater stage_updater(
      ThreadStatus::STAGE_PICK_MEMTABLES_TO_FLUSH);
  const auto& memlist = current_->memlist_;
+  bool atomic_flush = false;
  for (auto it = memlist.rbegin(); it != memlist.rend(); ++it) {
    MemTable* m = *it;
+    if (!atomic_flush && m->atomic_flush_seqno_ != kMaxSequenceNumber) {
+      atomic_flush = true;
+    }
    if (max_memtable_id != nullptr && m->GetID() > *max_memtable_id) {
      break;
    }
@@ -292,7 +296,9 @@ void MemTableList::PickMemtablesToFlush(const uint64_t* max_memtable_id,
      ret->push_back(m);
    }
  }
-  flush_requested_ = false;  // start-flush request is complete
+  if (!atomic_flush || num_flush_not_started_ == 0) {
+    flush_requested_ = false;  // start-flush request is complete
+  }
 }

 void MemTableList::RollbackMemtableFlush(const autovector<MemTable*>& mems,
@@ -201,7 +201,15 @@ Status MergeHelper::MergeUntil(InternalIterator* iter,
      // want. Also if we're in compaction and it's a put, it would be nice to
      // run compaction filter on it.
      const Slice val = iter->value();
-      const Slice* val_ptr = (kTypeValue == ikey.type) ? &val : nullptr;
+      const Slice* val_ptr;
+      if (kTypeValue == ikey.type &&
+          (range_del_agg == nullptr ||
+           !range_del_agg->ShouldDelete(
+               ikey, RangeDelPositioningMode::kForwardTraversal))) {
+        val_ptr = &val;
+      } else {
+        val_ptr = nullptr;
+      }
      std::string merge_result;
      s = TimedFullMerge(user_merge_operator_, ikey.user_key, val_ptr,
                         merge_context_.GetOperands(), &merge_result, logger_,
@@ -91,13 +91,23 @@ class SnapshotList {
      SequenceNumber* oldest_write_conflict_snapshot = nullptr,
      const SequenceNumber& max_seq = kMaxSequenceNumber) const {
    std::vector<SequenceNumber> ret;
+    GetAll(&ret, oldest_write_conflict_snapshot, max_seq);
+    return ret;
+  }
+
+  void GetAll(std::vector<SequenceNumber>* snap_vector,
+              SequenceNumber* oldest_write_conflict_snapshot = nullptr,
+              const SequenceNumber& max_seq = kMaxSequenceNumber) const {
+    std::vector<SequenceNumber>& ret = *snap_vector;
+    // So far we have no use case that would pass a non-empty vector
+    assert(ret.size() == 0);

    if (oldest_write_conflict_snapshot != nullptr) {
      *oldest_write_conflict_snapshot = kMaxSequenceNumber;
    }

    if (empty()) {
-      return ret;
+      return;
    }
    const SnapshotImpl* s = &list_;
    while (s->next_ != &list_) {
@@ -119,7 +129,7 @@ class SnapshotList {

      s = s->next_;
    }
-    return ret;
+    return;
  }

  // get the sequence number of the most recent snapshot
@@ -213,9 +213,6 @@ InternalIterator* TableCache::NewIterator(
      readahead = env_options.compaction_readahead_size;
      create_new_table_reader = true;
    }
-  } else {
-    readahead = options.readahead_size;
-    create_new_table_reader = readahead > 0;
  }

  auto& fd = file_meta.fd;
@@ -353,7 +353,7 @@ class FilePickerMultiGet {
  struct FilePickerContext;

 public:
-  FilePickerMultiGet(std::vector<FileMetaData*>* files, MultiGetRange* range,
+  FilePickerMultiGet(MultiGetRange* range,
                     autovector<LevelFilesBrief>* file_levels,
                     unsigned int num_levels, FileIndexer* file_indexer,
                     const Comparator* user_comparator,
@@ -368,18 +368,12 @@ class FilePickerMultiGet {
        maybe_repeat_key_(false),
        current_level_range_(*range, range->begin(), range->end()),
        current_file_range_(*range, range->begin(), range->end()),
-#ifndef NDEBUG
-        files_(files),
-#endif
        level_files_brief_(file_levels),
        is_hit_file_last_in_level_(false),
        curr_file_level_(nullptr),
        file_indexer_(file_indexer),
        user_comparator_(user_comparator),
        internal_comparator_(internal_comparator) {
-#ifdef NDEBUG
-    (void)files;
-#endif
    for (auto iter = range_->begin(); iter != range_->end(); ++iter) {
      fp_ctx_array_[iter.index()] =
          FilePickerContext(0, FileIndexer::kLevelMaxIndex);
@@ -416,6 +410,18 @@ class FilePickerMultiGet {
    bool file_hit = false;
    int cmp_largest = -1;
    if (curr_file_index >= curr_file_level_->num_files) {
+      // In the unlikely case the next key is a duplicate of the current key,
+      // and the current key is the last in the level and the internal key
+      // was not found, we need to skip lookup for the remaining keys and
+      // reset the search bounds
+      if (batch_iter_ != current_level_range_.end()) {
+        ++batch_iter_;
+        for (; batch_iter_ != current_level_range_.end(); ++batch_iter_) {
+          struct FilePickerContext& fp_ctx = fp_ctx_array_[batch_iter_.index()];
+          fp_ctx.search_left_bound = 0;
+          fp_ctx.search_right_bound = FileIndexer::kLevelMaxIndex;
+        }
+      }
      return false;
    }
    // Loops over keys in the MultiGet batch until it finds a file with
@@ -473,25 +479,6 @@ class FilePickerMultiGet {
      } else {
        file_hit = true;
      }
-#ifndef NDEBUG
-      // Sanity check to make sure that the files are correctly sorted
-      if (f != prev_file_) {
-        if (prev_file_) {
-          if (curr_level_ != 0) {
-            int comp_sign = internal_comparator_->Compare(
-                prev_file_->largest_key, f->smallest_key);
-            assert(comp_sign < 0);
-          } else if (fp_ctx.curr_index_in_curr_level > 0) {
-            // level == 0, the current file cannot be newer than the previous
-            // one. Use compressed data structure, has no attribute seqNo
-            assert(!NewestFirstBySeqNo(
-                files_[0][fp_ctx.curr_index_in_curr_level],
-                files_[0][fp_ctx.curr_index_in_curr_level - 1]));
-          }
-        }
-        prev_file_ = f;
-      }
-#endif
      if (cmp_largest == 0) {
        // cmp_largest is 0, which means the next key will not be in this
        // file, so stop looking further. Also don't increment megt_iter_
@@ -533,7 +520,10 @@ class FilePickerMultiGet {
          // any further for that key, so advance batch_iter_. Else, keep
          // batch_iter_ positioned on that key so we look it up again in
          // the next file
-          if (current_level_range_.CheckKeyDone(batch_iter_)) {
+          // For L0, always advance the key because we will look in the next
+          // file regardless for all keys not found yet
+          if (current_level_range_.CheckKeyDone(batch_iter_) ||
+              curr_level_ == 0) {
            ++batch_iter_;
          }
        }
@@ -601,7 +591,8 @@ class FilePickerMultiGet {
    unsigned int start_index_in_curr_level;

    FilePickerContext(int32_t left, int32_t right)
-        : search_left_bound(left), search_right_bound(right) {}
+        : search_left_bound(left), search_right_bound(right),
+          curr_index_in_curr_level(0), start_index_in_curr_level(0) {}

    FilePickerContext() = default;
  };
@@ -619,9 +610,6 @@ class FilePickerMultiGet {
  bool maybe_repeat_key_;
  MultiGetRange current_level_range_;
  MultiGetRange current_file_range_;
-#ifndef NDEBUG
-  std::vector<FileMetaData*>* files_;
-#endif
  autovector<LevelFilesBrief>* level_files_brief_;
  bool search_ended_;
  bool is_hit_file_last_in_level_;
@@ -629,9 +617,6 @@ class FilePickerMultiGet {
  FileIndexer* file_indexer_;
  const Comparator* user_comparator_;
  const InternalKeyComparator* internal_comparator_;
-#ifndef NDEBUG
-  FdWithKeyRange* prev_file_;
-#endif

  // Setup local variables to search next level.
  // Returns false if there are no more levels to search.
@@ -640,9 +625,6 @@ class FilePickerMultiGet {
      MultiGetRange::Iterator mget_iter = current_level_range_.begin();
      if (fp_ctx_array_[mget_iter.index()].curr_index_in_curr_level <
          curr_file_level_->num_files) {
-#ifndef NDEBUG
-        prev_file_ = nullptr;
-#endif
        batch_iter_prev_ = current_level_range_.begin();
        batch_iter_ = current_level_range_.begin();
        return true;
@@ -738,9 +720,6 @@ class FilePickerMultiGet {
        fp_ctx.curr_index_in_curr_level = start_index;
      }
      if (level_contains_keys) {
-#ifndef NDEBUG
-        prev_file_ = nullptr;
-#endif
        batch_iter_prev_ = current_level_range_.begin();
        batch_iter_ = current_level_range_.begin();
        return true;
@@ -1758,12 +1737,16 @@ void Version::MultiGet(const ReadOptions& read_options, MultiGetRange* range,
        iter->value, nullptr, &(iter->merge_context),
        &iter->max_covering_tombstone_seq, this->env_, &iter->seq,
        merge_operator_ ? &pinned_iters_mgr : nullptr, callback, is_blob);
-    iter->get_context = &get_ctx.back();
+  }
+  int get_ctx_index = 0;
+  for (auto iter = range->begin(); iter != range->end();
+       ++iter, get_ctx_index++) {
+    iter->get_context = &(get_ctx[get_ctx_index]);
  }

  MultiGetRange file_picker_range(*range, range->begin(), range->end());
  FilePickerMultiGet fp(
-      storage_info_.files_, &file_picker_range,
+      &file_picker_range,
      &storage_info_.level_files_brief_, storage_info_.num_non_empty_levels_,
      &storage_info_.file_indexer_, user_comparator(), internal_comparator());
  FdWithKeyRange* f = fp.GetNextFile();
@@ -336,6 +336,11 @@ int rocksdb_property_int(
    rocksdb_t* db,
    const char* propname, uint64_t *out_val);

+/* returns 0 on success, -1 otherwise */
+int rocksdb_property_int_cf(
+    rocksdb_t* db, rocksdb_column_family_handle_t* column_family,
+    const char* propname, uint64_t *out_val);
+
 extern ROCKSDB_LIBRARY_API char* rocksdb_property_value_cf(
    rocksdb_t* db, rocksdb_column_family_handle_t* column_family,
    const char* propname);
@@ -380,6 +385,10 @@ extern ROCKSDB_LIBRARY_API const rocksdb_livefiles_t* rocksdb_livefiles(
 extern ROCKSDB_LIBRARY_API void rocksdb_flush(
    rocksdb_t* db, const rocksdb_flushoptions_t* options, char** errptr);

+extern ROCKSDB_LIBRARY_API void rocksdb_flush_cf(
+    rocksdb_t* db, const rocksdb_flushoptions_t* options,
+    rocksdb_column_family_handle_t* column_family, char** errptr);
+
 extern ROCKSDB_LIBRARY_API void rocksdb_disable_file_deletions(rocksdb_t* db,
                                                               char** errptr);

@@ -807,6 +816,8 @@ extern ROCKSDB_LIBRARY_API void rocksdb_options_set_target_file_size_multiplier(
    rocksdb_options_t*, int);
 extern ROCKSDB_LIBRARY_API void rocksdb_options_set_max_bytes_for_level_base(
    rocksdb_options_t*, uint64_t);
+extern ROCKSDB_LIBRARY_API void rocksdb_options_set_snap_refresh_nanos(
+    rocksdb_options_t*, uint64_t);
 extern ROCKSDB_LIBRARY_API void
 rocksdb_options_set_level_compaction_dynamic_level_bytes(rocksdb_options_t*,
                                                         unsigned char);
@@ -269,6 +269,19 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
  // Dynamically changeable through SetOptions() API
  uint64_t max_bytes_for_level_base = 256 * 1048576;

+  // If non-zero, compactions will periodically refresh the snapshot list. The
+  // delay for the first refresh is snap_refresh_nanos nano seconds and
+  // exponentially increases afterwards. When having many short-lived snapshots,
+  // this option helps reducing the cpu usage of long-running compactions. The
+  // feature is disabled when max_subcompactions is greater than one.
+  //
+  // NOTE: This feautre is currently incompatible with RangeDeletes.
+  //
+  // Default: 0
+  //
+  // Dynamically changeable through SetOptions() API
+  uint64_t snap_refresh_nanos = 0;
+
  // Disable automatic compactions. Manual compactions can still
  // be issued on this column family
  //
@@ -1120,9 +1133,14 @@ struct ReadOptions {
  // Default: nullptr
  const Slice* iterate_upper_bound;

-  // If non-zero, NewIterator will create a new table reader which
-  // performs reads of the given size. Using a large size (> 2MB) can
-  // improve the performance of forward iteration on spinning disks.
+  // RocksDB does auto-readahead for iterators on noticing more than two reads
+  // for a table file. The readahead starts at 8KB and doubles on every
+  // additional read upto 256KB.
+  // This option can help if most of the range scans are large, and if it is
+  // determined that a larger readahead than that enabled by auto-readahead is
+  // needed.
+  // Using a large readahead size (> 2MB) can typically improve the performance
+  // of forward iteration on spinning disks.
  // Default: 0
  size_t readahead_size;

@@ -33,9 +33,10 @@ namespace rocksdb {
 // * merge_operator
 // * compaction_filter
 //
-// User can also choose to load customized comparator and/or merge_operator
-// through object registry:
+// User can also choose to load customized comparator, env, and/or
+// merge_operator through object registry:
 // * comparator needs to be registered through Registrar<const Comparator>
+// * env needs to be registered through Registrar<Env>
 // * merge operator needs to be registered through
 //     Registrar<std::shared_ptr<MergeOperator>>.
 //
@@ -5,8 +5,8 @@
 #pragma once

 #define ROCKSDB_MAJOR 6
-#define ROCKSDB_MINOR 1
-#define ROCKSDB_PATCH 1
+#define ROCKSDB_MINOR 2
+#define ROCKSDB_PATCH 4

 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these
@@ -169,6 +169,8 @@ void MutableCFOptions::Dump(Logger* log) const {
                 target_file_size_multiplier);
  ROCKS_LOG_INFO(log, "                 max_bytes_for_level_base: %" PRIu64,
                 max_bytes_for_level_base);
+  ROCKS_LOG_INFO(log, "                       snap_refresh_nanos: %" PRIu64,
+                 snap_refresh_nanos);
  ROCKS_LOG_INFO(log, "           max_bytes_for_level_multiplier: %f",
                 max_bytes_for_level_multiplier);
  ROCKS_LOG_INFO(log, "                                      ttl: %" PRIu64,
@@ -149,6 +149,7 @@ struct MutableCFOptions {
        target_file_size_base(options.target_file_size_base),
        target_file_size_multiplier(options.target_file_size_multiplier),
        max_bytes_for_level_base(options.max_bytes_for_level_base),
+        snap_refresh_nanos(options.snap_refresh_nanos),
        max_bytes_for_level_multiplier(options.max_bytes_for_level_multiplier),
        ttl(options.ttl),
        periodic_compaction_seconds(options.periodic_compaction_seconds),
@@ -185,6 +186,7 @@ struct MutableCFOptions {
        target_file_size_base(0),
        target_file_size_multiplier(0),
        max_bytes_for_level_base(0),
+        snap_refresh_nanos(0),
        max_bytes_for_level_multiplier(0),
        ttl(0),
        periodic_compaction_seconds(0),
@@ -236,6 +238,7 @@ struct MutableCFOptions {
  uint64_t target_file_size_base;
  int target_file_size_multiplier;
  uint64_t max_bytes_for_level_base;
+  uint64_t snap_refresh_nanos;
  double max_bytes_for_level_multiplier;
  uint64_t ttl;
  uint64_t periodic_compaction_seconds;
@@ -215,6 +215,9 @@ void ColumnFamilyOptions::Dump(Logger* log) const {
    ROCKS_LOG_HEADER(
        log, "               Options.max_bytes_for_level_base: %" PRIu64,
        max_bytes_for_level_base);
+    ROCKS_LOG_HEADER(
+        log, "                     Options.snap_refresh_nanos: %" PRIu64,
+        snap_refresh_nanos);
    ROCKS_LOG_HEADER(log, "Options.level_compaction_dynamic_level_bytes: %d",
                     level_compaction_dynamic_level_bytes);
    ROCKS_LOG_HEADER(log, "         Options.max_bytes_for_level_multiplier: %f",
@@ -490,6 +493,7 @@ ColumnFamilyOptions* ColumnFamilyOptions::OptimizeForSmallDb(
  write_buffer_size = 2 << 20;
  target_file_size_base = 2 * 1048576;
  max_bytes_for_level_base = 10 * 1048576;
+  snap_refresh_nanos = 0;
  soft_pending_compaction_bytes_limit = 256 * 1048576;
  hard_pending_compaction_bytes_limit = 1073741824ul;

@@ -177,6 +177,7 @@ ColumnFamilyOptions BuildColumnFamilyOptions(
      mutable_cf_options.target_file_size_multiplier;
  cf_opts.max_bytes_for_level_base =
      mutable_cf_options.max_bytes_for_level_base;
+  cf_opts.snap_refresh_nanos = mutable_cf_options.snap_refresh_nanos;
  cf_opts.max_bytes_for_level_multiplier =
      mutable_cf_options.max_bytes_for_level_multiplier;
  cf_opts.ttl = mutable_cf_options.ttl;
@@ -247,6 +248,7 @@ std::unordered_map<std::string, CompressionType>
 #ifndef ROCKSDB_LITE

 const std::string kNameComparator = "comparator";
+const std::string kNameEnv = "env";
 const std::string kNameMergeOperator = "merge_operator";

 template <typename T>
@@ -525,9 +527,9 @@ bool ParseOptionHelper(char* opt_address, const OptionType& opt_type,
              opt_address));
    case OptionType::kBlockBasedTableIndexShorteningMode:
      return ParseEnum<BlockBasedTableOptions::IndexShorteningMode>(
-        block_base_table_index_shortening_mode_string_map, value,
-        reinterpret_cast<BlockBasedTableOptions::IndexShorteningMode*>(
-            opt_address));
+          block_base_table_index_shortening_mode_string_map, value,
+          reinterpret_cast<BlockBasedTableOptions::IndexShorteningMode*>(
+              opt_address));
    case OptionType::kEncodingType:
      return ParseEnum<EncodingType>(
          encoding_type_string_map, value,
@@ -1179,6 +1181,14 @@ Status ParseDBOption(const std::string& name,
    if (name == "rate_limiter_bytes_per_sec") {
      new_options->rate_limiter.reset(
          NewGenericRateLimiter(static_cast<int64_t>(ParseUint64(value))));
+    } else if (name == kNameEnv) {
+      // Currently `Env` can be deserialized from object registry only.
+      std::unique_ptr<Env> env_guard;
+      Env* env = NewCustomObject<Env>(value, &env_guard);
+      // Only support static env for now.
+      if (env != nullptr && !env_guard) {
+        new_options->env = env;
+      }
    } else {
      auto iter = db_options_type_info.find(name);
      if (iter == db_options_type_info.end()) {
@@ -1388,7 +1398,6 @@ std::unordered_map<std::string, OptionTypeInfo>
    OptionsHelper::db_options_type_info = {
        /*
         // not yet supported
-          Env* env;
          std::shared_ptr<Cache> row_cache;
          std::shared_ptr<DeleteScheduler> delete_scheduler;
          std::shared_ptr<Logger> info_log;
@@ -1553,8 +1562,8 @@ std::unordered_map<std::string, OptionTypeInfo>
          OptionVerificationType::kNormal, true,
          offsetof(struct MutableDBOptions, wal_bytes_per_sync)}},
        {"strict_bytes_per_sync",
-         {offsetof(struct DBOptions, strict_bytes_per_sync), OptionType::kBoolean,
-          OptionVerificationType::kNormal, true,
+         {offsetof(struct DBOptions, strict_bytes_per_sync),
+          OptionType::kBoolean, OptionVerificationType::kNormal, true,
          offsetof(struct MutableDBOptions, strict_bytes_per_sync)}},
        {"stats_dump_period_sec",
         {offsetof(struct DBOptions, stats_dump_period_sec), OptionType::kUInt,
@@ -1639,8 +1648,8 @@ std::unordered_map<std::string, OptionTypeInfo>
        {"avoid_unnecessary_blocking_io",
         {offsetof(struct DBOptions, avoid_unnecessary_blocking_io),
          OptionType::kBoolean, OptionVerificationType::kNormal, false,
-          offsetof(struct ImmutableDBOptions, avoid_unnecessary_blocking_io)}}
-      };
+          offsetof(struct ImmutableDBOptions, avoid_unnecessary_blocking_io)}},
+};

 std::unordered_map<std::string, BlockBasedTableOptions::IndexType>
    OptionsHelper::block_base_table_index_type_string_map = {
@@ -1658,13 +1667,13 @@ std::unordered_map<std::string, BlockBasedTableOptions::DataBlockIndexType>

 std::unordered_map<std::string, BlockBasedTableOptions::IndexShorteningMode>
    OptionsHelper::block_base_table_index_shortening_mode_string_map = {
-      {"kNoShortening",
-       BlockBasedTableOptions::IndexShorteningMode::kNoShortening},
-      {"kShortenSeparators",
-       BlockBasedTableOptions::IndexShorteningMode::kShortenSeparators},
-      {"kShortenSeparatorsAndSuccessor",
-       BlockBasedTableOptions::IndexShorteningMode::
-           kShortenSeparatorsAndSuccessor}};
+        {"kNoShortening",
+         BlockBasedTableOptions::IndexShorteningMode::kNoShortening},
+        {"kShortenSeparators",
+         BlockBasedTableOptions::IndexShorteningMode::kShortenSeparators},
+        {"kShortenSeparatorsAndSuccessor",
+         BlockBasedTableOptions::IndexShorteningMode::
+             kShortenSeparatorsAndSuccessor}};

 std::unordered_map<std::string, EncodingType>
    OptionsHelper::encoding_type_string_map = {{"kPlain", kPlain},
@@ -1902,6 +1911,10 @@ std::unordered_map<std::string, OptionTypeInfo>
         {offset_of(&ColumnFamilyOptions::max_bytes_for_level_base),
          OptionType::kUInt64T, OptionVerificationType::kNormal, true,
          offsetof(struct MutableCFOptions, max_bytes_for_level_base)}},
+        {"snap_refresh_nanos",
+         {offset_of(&ColumnFamilyOptions::snap_refresh_nanos),
+          OptionType::kUInt64T, OptionVerificationType::kNormal, true,
+          offsetof(struct MutableCFOptions, snap_refresh_nanos)}},
        {"max_bytes_for_level_multiplier",
         {offset_of(&ColumnFamilyOptions::max_bytes_for_level_multiplier),
          OptionType::kDouble, OptionVerificationType::kNormal, true,
@@ -81,7 +81,8 @@ enum class OptionType {
  kAccessHint,
  kInfoLogLevel,
  kLRUCacheOptions,
-  kUnknown
+  kEnv,
+  kUnknown,
 };

 enum class OptionVerificationType {
@@ -415,6 +415,7 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
      "kBZip2Compression:kNoCompression:kZlibCompression:kBZip2Compression:"
      "kSnappyCompression;"
      "max_bytes_for_level_base=986;"
+      "snap_refresh_nanos=1000000000;"
      "bloom_locality=8016;"
      "target_file_size_base=4294976376;"
      "memtable_huge_page_size=2557;"
@@ -74,6 +74,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
      {"target_file_size_base", "12"},
      {"target_file_size_multiplier", "13"},
      {"max_bytes_for_level_base", "14"},
+      {"snap_refresh_nanos", "1000000000"},
      {"level_compaction_dynamic_level_bytes", "true"},
      {"max_bytes_for_level_multiplier", "15.0"},
      {"max_bytes_for_level_multiplier_additional", "16:17:18"},
@@ -183,6 +184,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
  ASSERT_EQ(new_cf_opt.target_file_size_base, static_cast<uint64_t>(12));
  ASSERT_EQ(new_cf_opt.target_file_size_multiplier, 13);
  ASSERT_EQ(new_cf_opt.max_bytes_for_level_base, 14U);
+  ASSERT_EQ(new_cf_opt.snap_refresh_nanos, 1000000000U);
  ASSERT_EQ(new_cf_opt.level_compaction_dynamic_level_bytes, true);
  ASSERT_EQ(new_cf_opt.max_bytes_for_level_multiplier, 15.0);
  ASSERT_EQ(new_cf_opt.max_bytes_for_level_multiplier_additional.size(), 3U);
@@ -759,6 +761,21 @@ TEST_F(OptionsTest, GetOptionsFromStringTest) {
  block_based_table_options.cache_index_and_filter_blocks = true;
  base_options.table_factory.reset(
      NewBlockBasedTableFactory(block_based_table_options));
+
+  // Register an Env with object registry.
+  const static char* kCustomEnvName = "CustomEnv";
+  class CustomEnv : public EnvWrapper {
+   public:
+    explicit CustomEnv(Env* _target) : EnvWrapper(_target) {}
+  };
+
+  static Registrar<Env> test_reg_env(
+      kCustomEnvName,
+      [](const std::string& /*name*/, std::unique_ptr<Env>* /*env_guard*/) {
+        static CustomEnv env(Env::Default());
+        return &env;
+      });
+
  ASSERT_OK(GetOptionsFromString(
      base_options,
      "write_buffer_size=10;max_write_buffer_number=16;"
@@ -766,7 +783,7 @@ TEST_F(OptionsTest, GetOptionsFromStringTest) {
      "compression_opts=4:5:6;create_if_missing=true;max_open_files=1;"
      "bottommost_compression_opts=5:6:7;create_if_missing=true;max_open_files="
      "1;"
-      "rate_limiter_bytes_per_sec=1024",
+      "rate_limiter_bytes_per_sec=1024;env=CustomEnv",
      &new_options));

  ASSERT_EQ(new_options.compression_opts.window_bits, 4);
@@ -795,6 +812,8 @@ TEST_F(OptionsTest, GetOptionsFromStringTest) {
  ASSERT_EQ(new_options.create_if_missing, true);
  ASSERT_EQ(new_options.max_open_files, 1);
  ASSERT_TRUE(new_options.rate_limiter.get() != nullptr);
+  std::unique_ptr<Env> env_guard;
+  ASSERT_EQ(NewCustomObject<Env>(kCustomEnvName, &env_guard), new_options.env);
 }

 TEST_F(OptionsTest, DBOptionsSerialization) {
@@ -216,6 +216,11 @@ LIB_SOURCES =                                                   \
  utilities/write_batch_with_index/write_batch_with_index.cc    \
  utilities/write_batch_with_index/write_batch_with_index_internal.cc    \

+ifeq ($(ARMCRC_SOURCE),1)
+LIB_SOURCES +=\
+  util/crc32c_arm64.cc
+endif
+
 ifeq (,$(shell $(CXX) -fsyntax-only -maltivec -xc /dev/null 2>&1))
 LIB_SOURCES_ASM =\
  util/crc32c_ppc_asm.S
@@ -838,6 +838,11 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
                                         rep->persistent_cache_key_prefix_size),
                             rep->ioptions.statistics);

+  // Meta-blocks are not dictionary compressed. Explicitly set the dictionary
+  // handle to null, otherwise it may be seen as uninitialized during the below
+  // meta-block reads.
+  rep->compression_dict_handle = BlockHandle::NullBlockHandle();
+
  // Read metaindex
  std::unique_ptr<Block> meta;
  std::unique_ptr<InternalIterator> meta_iter;
@@ -2167,10 +2172,6 @@ BlockBasedTable::PartitionedIndexIteratorState::PartitionedIndexIteratorState(
      index_key_includes_seq_(index_key_includes_seq),
      index_key_is_full_(index_key_is_full) {}

-template <class TBlockIter, typename TValue>
-const size_t BlockBasedTableIterator<TBlockIter, TValue>::kMaxReadaheadSize =
-    256 * 1024;
-
 InternalIteratorBase<BlockHandle>*
 BlockBasedTable::PartitionedIndexIteratorState::NewSecondaryIterator(
    const BlockHandle& handle) {
@@ -2453,6 +2454,13 @@ void BlockBasedTableIterator<TBlockIter, TValue>::Prev() {
  FindKeyBackward();
 }

+// Found that 256 KB readahead size provides the best performance, based on
+// experiments, for auto readahead. Experiment data is in PR #3282.
+template <class TBlockIter, typename TValue>
+const size_t
+    BlockBasedTableIterator<TBlockIter, TValue>::kMaxAutoReadaheadSize =
+        256 * 1024;
+
 template <class TBlockIter, typename TValue>
 void BlockBasedTableIterator<TBlockIter, TValue>::InitDataBlock() {
  BlockHandle data_block_handle = index_iter_->value();
@@ -2465,32 +2473,47 @@ void BlockBasedTableIterator<TBlockIter, TValue>::InitDataBlock() {
    }
    auto* rep = table_->get_rep();

-    // Automatically prefetch additional data when a range scan (iterator) does
-    // more than 2 sequential IOs. This is enabled only for user reads and when
-    // ReadOptions.readahead_size is 0.
-    if (!for_compaction_ && read_options_.readahead_size == 0) {
-      num_file_reads_++;
-      if (num_file_reads_ > 2) {
-        if (!rep->file->use_direct_io() &&
-            (data_block_handle.offset() +
-                 static_cast<size_t>(data_block_handle.size()) +
-                 kBlockTrailerSize >
-             readahead_limit_)) {
-          // Buffered I/O
-          // Discarding the return status of Prefetch calls intentionally, as we
-          // can fallback to reading from disk if Prefetch fails.
-          rep->file->Prefetch(data_block_handle.offset(), readahead_size_);
-          readahead_limit_ =
-              static_cast<size_t>(data_block_handle.offset() + readahead_size_);
-          // Keep exponentially increasing readahead size until
-          // kMaxReadaheadSize.
-          readahead_size_ = std::min(kMaxReadaheadSize, readahead_size_ * 2);
-        } else if (rep->file->use_direct_io() && !prefetch_buffer_) {
-          // Direct I/O
-          // Let FilePrefetchBuffer take care of the readahead.
-          prefetch_buffer_.reset(new FilePrefetchBuffer(
-              rep->file.get(), kInitReadaheadSize, kMaxReadaheadSize));
+    // Prefetch additional data for range scans (iterators). Enabled only for
+    // user reads.
+    // Implicit auto readahead:
+    //   Enabled after 2 sequential IOs when ReadOptions.readahead_size == 0.
+    // Explicit user requested readahead:
+    //   Enabled from the very first IO when ReadOptions.readahead_size is set.
+    if (!for_compaction_) {
+      if (read_options_.readahead_size == 0) {
+        // Implicit auto readahead
+        num_file_reads_++;
+        if (num_file_reads_ > kMinNumFileReadsToStartAutoReadahead) {
+          if (!rep->file->use_direct_io() &&
+              (data_block_handle.offset() +
+                   static_cast<size_t>(data_block_handle.size()) +
+                   kBlockTrailerSize >
+               readahead_limit_)) {
+            // Buffered I/O
+            // Discarding the return status of Prefetch calls intentionally, as
+            // we can fallback to reading from disk if Prefetch fails.
+            rep->file->Prefetch(data_block_handle.offset(), readahead_size_);
+            readahead_limit_ = static_cast<size_t>(data_block_handle.offset() +
+                                                   readahead_size_);
+            // Keep exponentially increasing readahead size until
+            // kMaxAutoReadaheadSize.
+            readahead_size_ =
+                std::min(kMaxAutoReadaheadSize, readahead_size_ * 2);
+          } else if (rep->file->use_direct_io() && !prefetch_buffer_) {
+            // Direct I/O
+            // Let FilePrefetchBuffer take care of the readahead.
+            prefetch_buffer_.reset(
+                new FilePrefetchBuffer(rep->file.get(), kInitAutoReadaheadSize,
+                                       kMaxAutoReadaheadSize));
+          }
        }
+      } else if (!prefetch_buffer_) {
+        // Explicit user requested readahead
+        // The actual condition is:
+        // if (read_options_.readahead_size != 0 && !prefetch_buffer_)
+        prefetch_buffer_.reset(new FilePrefetchBuffer(
+            rep->file.get(), read_options_.readahead_size,
+            read_options_.readahead_size));
      }
    }

@@ -717,13 +717,15 @@ class BlockBasedTableIterator : public InternalIteratorBase<TValue> {
  bool for_compaction_;
  BlockHandle prev_index_value_;

-  static const size_t kInitReadaheadSize = 8 * 1024;
+  // All the below fields control iterator readahead
+  static const size_t kInitAutoReadaheadSize = 8 * 1024;
  // Found that 256 KB readahead size provides the best performance, based on
-  // experiments.
-  static const size_t kMaxReadaheadSize;
-  size_t readahead_size_ = kInitReadaheadSize;
+  // experiments, for auto readahead. Experiment data is in PR #3282.
+  static const size_t kMaxAutoReadaheadSize;
+  static const int kMinNumFileReadsToStartAutoReadahead = 2;
+  size_t readahead_size_ = kInitAutoReadaheadSize;
  size_t readahead_limit_ = 0;
-  int num_file_reads_ = 0;
+  int64_t num_file_reads_ = 0;
  std::unique_ptr<FilePrefetchBuffer> prefetch_buffer_;
 };

@@ -21,6 +21,12 @@ const InternalKeyComparator icmp_(BytewiseComparator());

 }  // namespace

+stl_wrappers::KVMap MakeMockFile(
+    std::vector<std::pair<const std::string, std::string>> l) {
+  return stl_wrappers::KVMap(l.begin(), l.end(),
+                             stl_wrappers::LessOfComparator(&icmp_));
+}
+
 stl_wrappers::KVMap MakeMockFile(
    std::initializer_list<std::pair<const std::string, std::string>> l) {
  return stl_wrappers::KVMap(l, stl_wrappers::LessOfComparator(&icmp_));
@@ -137,6 +143,14 @@ void MockTableFactory::AssertLatestFile(
      ParseInternalKey(Slice(key), &ikey);
      std::cout << ikey.DebugString(false) << " -> " << value << std::endl;
    }
+    std::cout << "Expected:" << std::endl;
+    for (const auto& kv : file_contents) {
+      ParsedInternalKey ikey;
+      std::string key, value;
+      std::tie(key, value) = kv;
+      ParseInternalKey(Slice(key), &ikey);
+      std::cout << ikey.DebugString(false) << " -> " << value << std::endl;
+    }
    FAIL();
  }
 }
@@ -28,6 +28,8 @@ namespace mock {

 stl_wrappers::KVMap MakeMockFile(
    std::initializer_list<std::pair<const std::string, std::string>> l = {});
+stl_wrappers::KVMap MakeMockFile(
+    std::vector<std::pair<const std::string, std::string>> l);

 struct MockTableFileSystem {
  port::Mutex mutex;
@@ -184,6 +186,12 @@ class MockTableFactory : public TableFactory {
  // contents are equal to file_contents
  void AssertSingleFile(const stl_wrappers::KVMap& file_contents);
  void AssertLatestFile(const stl_wrappers::KVMap& file_contents);
+  stl_wrappers::KVMap output() {
+    assert(!file_system_.files.empty());
+    auto latest = file_system_.files.end();
+    --latest;
+    return latest->second;
+  }

 private:
  uint32_t GetAndWriteNextID(WritableFileWriter* file) const;
@@ -123,7 +123,7 @@ class MultiGetContext {
  KeyContext** sorted_keys_;
  size_t num_keys_;
  uint64_t value_mask_;
-  std::unique_ptr<char> lookup_key_heap_buf;
+  std::unique_ptr<char[]> lookup_key_heap_buf;
  LookupKey* lookup_key_ptr_;

 public:
@@ -1172,6 +1172,7 @@ DEFINE_int32(skip_list_lookahead, 0, "Used with skip_list memtablerep; try "
             "position");
 DEFINE_bool(report_file_operations, false, "if report number of file "
            "operations");
+DEFINE_int32(readahead_size, 0, "Iterator readahead size");

 static const bool FLAGS_soft_rate_limit_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&FLAGS_soft_rate_limit, &ValidateRateLimit);
@@ -4987,6 +4988,7 @@ void VerifyDBFromDB(std::string& truth_db_name) {
    options.total_order_seek = FLAGS_total_order_seek;
    options.prefix_same_as_start = FLAGS_prefix_same_as_start;
    options.tailing = FLAGS_use_tailing_iterator;
+    options.readahead_size = FLAGS_readahead_size;

    Iterator* single_iter = nullptr;
    std::vector<Iterator*> multi_iters;
@@ -25,6 +25,7 @@ expected_values_file = tempfile.NamedTemporaryFile()
 default_params = {
    "acquire_snapshot_one_in": 10000,
    "block_size": 16384,
+    "cache_index_and_filter_blocks": lambda: random.randint(0, 1),
    "cache_size": 1048576,
    "checkpoint_one_in": 1000000,
    "compression_type": "snappy",
@@ -64,6 +65,7 @@ default_params = {
    "writepercent": 35,
    "format_version": lambda: random.randint(2, 4),
    "index_block_restart_interval": lambda: random.choice(range(1, 16)),
+    "use_multiget" : lambda: random.randint(0, 1),
 }

 _TEST_DIR_ENV_VAR = 'TEST_TMPDIR'
@@ -136,6 +138,8 @@ atomic_flush_params = {
    # use small value for write_buffer_size so that RocksDB triggers flush
    # more frequently
    "write_buffer_size": 1024 * 1024,
+    # disable pipelined write when test_atomic_flush is true
+    "enable_pipelined_write": 0,
 }


@@ -339,8 +343,9 @@ def whitebox_crash_main(args, unknown_args):
        if additional_opts['kill_random_test'] is None and (retncode == 0):
            # we expect zero retncode if no kill option
            expected = True
-        elif additional_opts['kill_random_test'] is not None and retncode < 0:
-            # we expect negative retncode if kill option was given
+        elif additional_opts['kill_random_test'] is not None and retncode <= 0:
+            # When kill option is given, the test MIGHT kill itself.
+            # If it does, negative retncode is expected. Otherwise 0.
            expected = True

        if not expected:
@@ -297,6 +297,9 @@ DEFINE_int32(set_in_place_one_in, 0,
 DEFINE_int64(cache_size, 2LL * KB * KB * KB,
             "Number of bytes to use as a cache of uncompressed data.");

+DEFINE_bool(cache_index_and_filter_blocks, false,
+            "True if indexes/filters should be cached in block cache.");
+
 DEFINE_bool(use_clock_cache, false,
            "Replace default LRU block cache with clock cache.");

@@ -452,6 +455,9 @@ DEFINE_uint64(snapshot_hold_ops, 0,
              "If non-zero, then releases snapshots N operations after they're "
              "acquired.");

+DEFINE_bool(use_multiget, false,
+            "If set, use the batched MultiGet API for reads");
+
 static bool ValidateInt32Percent(const char* flagname, int32_t value) {
  if (value < 0 || value>100) {
    fprintf(stderr, "Invalid value for --%s: %d, 0<= pct <=100 \n",
@@ -1722,6 +1728,27 @@ class StressTest {
    return base_key + thread->rand.Next() % FLAGS_active_width;
  }

+  static std::vector<int64_t> GenerateNKeys(
+      ThreadState* thread,
+      int num_keys,
+      uint64_t iteration) {
+    const double completed_ratio =
+        static_cast<double>(iteration) / FLAGS_ops_per_thread;
+    const int64_t base_key = static_cast<int64_t>(
+        completed_ratio * (FLAGS_max_key - FLAGS_active_width));
+    std::vector<int64_t> keys;
+    keys.reserve(num_keys);
+    int64_t next_key = base_key + thread->rand.Next() % FLAGS_active_width;
+    keys.push_back(next_key);
+    for (int i = 1; i < num_keys; ++i) {
+      // This may result in some duplicate keys
+      next_key = next_key + thread->rand.Next() %
+        (FLAGS_active_width - (next_key - base_key));
+      keys.push_back(next_key);
+    }
+    return keys;
+  }
+
  static size_t GenerateValue(uint32_t rand, char *v, size_t max_sz) {
    size_t value_sz =
        ((rand % kRandomValueMaxFactor) + 1) * FLAGS_value_size_mult;
@@ -2159,7 +2186,14 @@ class StressTest {
      int prob_op = thread->rand.Uniform(100);
      if (prob_op >= 0 && prob_op < (int)FLAGS_readpercent) {
        // OPERATION read
-        TestGet(thread, read_opts, rand_column_families, rand_keys);
+        if (FLAGS_use_multiget) {
+          int num_keys = thread->rand.Uniform(64);
+          rand_keys = GenerateNKeys(thread, num_keys, i);
+          TestMultiGet(thread, read_opts, rand_column_families, rand_keys);
+          i += num_keys - 1;
+        } else {
+          TestGet(thread, read_opts, rand_column_families, rand_keys);
+        }
      } else if ((int)FLAGS_readpercent <= prob_op && prob_op < prefixBound) {
        // OPERATION prefix scan
        // keys are 8 bytes long, prefix size is FLAGS_prefix_size. There are
@@ -2208,6 +2242,11 @@ class StressTest {
      const std::vector<int>& rand_column_families,
      const std::vector<int64_t>& rand_keys) = 0;

+  virtual std::vector<Status> TestMultiGet(ThreadState* thread,
+      const ReadOptions& read_opts,
+      const std::vector<int>& rand_column_families,
+      const std::vector<int64_t>& rand_keys) = 0;
+
  virtual Status TestPrefixScan(ThreadState* thread,
      const ReadOptions& read_opts,
      const std::vector<int>& rand_column_families,
@@ -2543,6 +2582,8 @@ class StressTest {
    fprintf(stdout, "Checksum type             : %s\n", checksum.c_str());
    fprintf(stdout, "Max subcompactions        : %" PRIu64 "\n",
            FLAGS_subcompactions);
+    fprintf(stdout, "Use MultiGet              : %s\n",
+            FLAGS_use_multiget ? "true" : "false");

    const char* memtablerep = "";
    switch (FLAGS_rep_factory) {
@@ -2578,6 +2619,8 @@ class StressTest {
    if (FLAGS_options_file.empty()) {
      BlockBasedTableOptions block_based_options;
      block_based_options.block_cache = cache_;
+      block_based_options.cache_index_and_filter_blocks =
+          FLAGS_cache_index_and_filter_blocks;
      block_based_options.block_cache_compressed = compressed_cache_;
      block_based_options.checksum = FLAGS_checksum_type_e;
      block_based_options.block_size = FLAGS_block_size;
@@ -3007,6 +3050,40 @@ class NonBatchedOpsStressTest : public StressTest {
    return s;
  }

+  virtual std::vector<Status> TestMultiGet(ThreadState* thread,
+      const ReadOptions& read_opts,
+      const std::vector<int>& rand_column_families,
+      const std::vector<int64_t>& rand_keys) {
+    size_t num_keys = rand_keys.size();
+    std::vector<std::string> key_str;
+    std::vector<Slice> keys;
+    key_str.reserve(num_keys);
+    keys.reserve(num_keys);
+    std::vector<PinnableSlice> values(num_keys);
+    std::vector<Status> statuses(num_keys);
+    ColumnFamilyHandle* cfh = column_families_[rand_column_families[0]];
+
+    for (size_t i = 0; i < num_keys; ++i) {
+      key_str.emplace_back(Key(rand_keys[i]));
+      keys.emplace_back(key_str.back());
+    }
+    db_->MultiGet(read_opts, cfh, num_keys, keys.data(), values.data(),
+                  statuses.data());
+    for (const auto& s : statuses) {
+      if (s.ok()) {
+        // found case
+        thread->stats.AddGets(1, 1);
+      } else if (s.IsNotFound()) {
+        // not found case
+        thread->stats.AddGets(1, 0);
+      } else {
+        // errors case
+        thread->stats.AddErrors(1);
+      }
+    }
+    return statuses;
+  }
+
  virtual Status TestPrefixScan(ThreadState* thread,
      const ReadOptions& read_opts,
      const std::vector<int>& rand_column_families,
@@ -3527,6 +3604,76 @@ class BatchedOpsStressTest : public StressTest {
    return s;
  }

+  virtual std::vector<Status> TestMultiGet(ThreadState* thread,
+      const ReadOptions& readoptions,
+      const std::vector<int>& rand_column_families,
+      const std::vector<int64_t>& rand_keys) {
+    size_t num_keys = rand_keys.size();
+    std::vector<Status> ret_status(num_keys);
+    std::array<std::string, 10> keys = {"0", "1", "2", "3", "4", "5", "6", "7", "8", "9"};
+    size_t num_prefixes = keys.size();
+    for (size_t rand_key = 0; rand_key < num_keys; ++rand_key) {
+      std::vector<Slice> key_slices;
+      std::vector<PinnableSlice> values(num_prefixes);
+      std::vector<Status> statuses(num_prefixes);
+      ReadOptions readoptionscopy = readoptions;
+      readoptionscopy.snapshot = db_->GetSnapshot();
+      std::vector<std::string> key_str;
+      key_str.reserve(num_prefixes);
+      key_slices.reserve(num_prefixes);
+      std::string from_db;
+      ColumnFamilyHandle* cfh = column_families_[rand_column_families[0]];
+
+      for (size_t key = 0; key < num_prefixes; ++key) {
+        key_str.emplace_back(keys[key] + Key(rand_keys[rand_key]));
+        key_slices.emplace_back(key_str.back());
+      }
+      db_->MultiGet(readoptionscopy, cfh, num_prefixes, key_slices.data(),
+          values.data(), statuses.data());
+      for (size_t i = 0; i < num_prefixes; i++) {
+        Status s = statuses[i];
+        if (!s.ok() && !s.IsNotFound()) {
+          fprintf(stderr, "get error: %s\n", s.ToString().c_str());
+          thread->stats.AddErrors(1);
+          ret_status[rand_key] = s;
+          // we continue after error rather than exiting so that we can
+          // find more errors if any
+        } else if (s.IsNotFound()) {
+          thread->stats.AddGets(1, 0);
+          ret_status[rand_key] = s;
+        } else {
+          char expected_prefix = (keys[i])[0];
+          char actual_prefix = (values[i])[0];
+          if (actual_prefix != expected_prefix) {
+            fprintf(stderr, "error expected prefix = %c actual = %c\n",
+                    expected_prefix, actual_prefix);
+          }
+          std::string str;
+          str.assign(values[i].data(), values[i].size());
+          values[i].Reset();
+          str[0] = ' '; // blank out the differing character
+          values[i].PinSelf(str);
+          thread->stats.AddGets(1, 1);
+        }
+      }
+      db_->ReleaseSnapshot(readoptionscopy.snapshot);
+
+      // Now that we retrieved all values, check that they all match
+      for (size_t i = 1; i < num_prefixes; i++) {
+        if (values[i] != values[0]) {
+          fprintf(stderr, "error : inconsistent values for key %s: %s, %s\n",
+                  key_str[i].c_str(),
+                  StringToHex(values[0].ToString()).c_str(),
+                  StringToHex(values[i].ToString()).c_str());
+        // we continue after error rather than exiting so that we can
+        // find more errors if any
+        }
+      }
+    }
+
+    return ret_status;
+  }
+
  // Given a key, this does prefix scans for "0"+P, "1"+P,..."9"+P
  // in the same snapshot where P is the first FLAGS_prefix_size - 1 bytes
  // of the key. Each of these 10 scans returns a series of values;
@@ -3742,6 +3889,39 @@ class AtomicFlushStressTest : public StressTest {
    return s;
  }

+  virtual std::vector<Status> TestMultiGet(ThreadState* thread,
+      const ReadOptions& read_opts,
+      const std::vector<int>& rand_column_families,
+      const std::vector<int64_t>& rand_keys) {
+    int num_keys = rand_keys.size();
+    std::vector<std::string> key_str;
+    std::vector<Slice> keys;
+    keys.reserve(num_keys);
+    key_str.reserve(num_keys);
+    std::vector<PinnableSlice> values(num_keys);
+    std::vector<Status> statuses(num_keys);
+    ColumnFamilyHandle* cfh = column_families_[rand_column_families[0]];
+
+    for (int i = 0; i < num_keys; ++i) {
+      key_str.emplace_back(Key(rand_keys[i]));
+      keys.emplace_back(key_str.back());
+    }
+    db_->MultiGet(read_opts, cfh, num_keys, keys.data(), values.data(), statuses.data());
+    for (auto s : statuses) {
+      if (s.ok()) {
+        // found case
+        thread->stats.AddGets(1, 1);
+      } else if (s.IsNotFound()) {
+        // not found case
+        thread->stats.AddGets(1, 0);
+      } else {
+        // errors case
+        thread->stats.AddErrors(1);
+      }
+    }
+    return statuses;
+  }
+
  virtual Status TestPrefixScan(ThreadState* thread,
                                const ReadOptions& readoptions,
                                const std::vector<int>& rand_column_families,
@@ -40,6 +40,9 @@ void CompactionJobStats::Reset() {
  file_fsync_nanos = 0;
  file_prepare_write_nanos = 0;

+  smallest_output_key_prefix.clear();
+  largest_output_key_prefix.clear();
+
  num_single_del_fallthru = 0;
  num_single_del_mismatch = 0;
 }
@@ -18,6 +18,8 @@
 #include "util/coding.h"
 #include "util/util.h"

+#include "util/crc32c_arm64.h"
+
 #ifdef __powerpc64__
 #include "util/crc32c_ppc.h"
 #include "util/crc32c_ppc_constants.h"
@@ -463,6 +465,11 @@ static bool isAltiVec() {
 }
 #endif

+#if defined(__linux__) && defined(HAVE_ARM64_CRC)
+uint32_t ExtendARMImpl(uint32_t crc, const char *buf, size_t size) {
+  return crc32c_arm64(crc, (const unsigned char *)buf, size);
+}
+#endif

 std::string IsFastCrc32Supported() {
  bool has_fast_crc = false;
@@ -478,6 +485,14 @@ std::string IsFastCrc32Supported() {
  has_fast_crc = false;
  arch = "PPC";
 #endif
+#elif defined(__linux__) && defined(HAVE_ARM64_CRC)
+  if (crc32c_runtime_check()) {
+    has_fast_crc = true;
+    arch = "Arm64";
+  } else {
+    has_fast_crc = false;
+    arch = "Arm64";
+  }
 #else
  has_fast_crc = isSSE42();
  arch = "x86";
@@ -1200,7 +1215,15 @@ uint32_t crc32c_3way(uint32_t crc, const char* buf, size_t len) {
 #endif //HAVE_SSE42 && HAVE_PCLMUL

 static inline Function Choose_Extend() {
-#ifndef HAVE_POWER8
+#ifdef HAVE_POWER8
+  return isAltiVec() ? ExtendPPCImpl : ExtendImpl<Slow_CRC32>;
+#elif defined(__linux__) && defined(HAVE_ARM64_CRC)
+  if(crc32c_runtime_check()) {
+    return ExtendARMImpl;
+  } else {
+    return ExtendImpl<Slow_CRC32>;
+  }
+#else
  if (isSSE42()) {
    if (isPCLMULQDQ()) {
 #if defined HAVE_SSE42  && defined HAVE_PCLMUL && !defined NO_THREEWAY_CRC32C
@@ -1216,8 +1239,6 @@ static inline Function Choose_Extend() {
  else {
    return ExtendImpl<Slow_CRC32>;
  }
-#else  //HAVE_POWER8
-  return isAltiVec() ? ExtendPPCImpl : ExtendImpl<Slow_CRC32>;
 #endif
 }

@@ -0,0 +1,56 @@
+//  Copyright (c) 2018, Arm Limited and affiliates. All rights reserved.
+//  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 "util/crc32c_arm64.h"
+
+#if defined(__linux__) && defined(HAVE_ARM64_CRC)
+
+#include <asm/hwcap.h>
+#include <sys/auxv.h>
+#ifndef HWCAP_CRC32
+#define HWCAP_CRC32 (1 << 7)
+#endif
+uint32_t crc32c_runtime_check(void) {
+  uint64_t auxv = getauxval(AT_HWCAP);
+  return (auxv & HWCAP_CRC32) != 0;
+}
+
+uint32_t crc32c_arm64(uint32_t crc, unsigned char const *data,
+                             unsigned len) {
+  const uint8_t *buf1;
+  const uint16_t *buf2;
+  const uint32_t *buf4;
+  const uint64_t *buf8;
+
+  int64_t length = (int64_t)len;
+
+  crc ^= 0xffffffff;
+  buf8 = (const uint64_t *)data;
+  while ((length -= sizeof(uint64_t)) >= 0) {
+    crc = __crc32cd(crc, *buf8++);
+  }
+
+  /* The following is more efficient than the straight loop */
+  buf4 = (const uint32_t *)buf8;
+  if (length & sizeof(uint32_t)) {
+    crc = __crc32cw(crc, *buf4++);
+    length -= 4;
+  }
+
+  buf2 = (const uint16_t *)buf4;
+  if (length & sizeof(uint16_t)) {
+    crc = __crc32ch(crc, *buf2++);
+    length -= 2;
+  }
+
+  buf1 = (const uint8_t *)buf2;
+  if (length & sizeof(uint8_t))
+    crc = __crc32cb(crc, *buf1);
+
+  crc ^= 0xffffffff;
+  return crc;
+}
+
+#endif
@@ -0,0 +1,21 @@
+//  Copyright (c) 2018, Arm Limited and affiliates. All rights reserved.
+//  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).
+
+#ifndef UTIL_CRC32C_ARM64_H
+#define UTIL_CRC32C_ARM64_H
+
+#include <inttypes.h>
+
+#if defined(__aarch64__) || defined(__AARCH64__)
+#ifdef __ARM_FEATURE_CRC32
+#define HAVE_ARM64_CRC
+#include <arm_acle.h>
+extern uint32_t crc32c_arm64(uint32_t crc, unsigned char const *data, unsigned len);
+extern uint32_t crc32c_runtime_check(void);
+#endif
+#endif
+
+
+#endif