StackableDB optionally take shared ownership of the underlying DB

BlockBasedTable::NewDataBlockIterator to always return BlockIter
Summary: This is a pre-cleaning up before a major block based table iterator refactoring. BlockBasedTable::NewDataBlockIterator() will always return BlockIter. This simplifies the logic and code and enable further refactoring and optimization. Closes https://github.com/facebook/rocksdb/pull/3398 Differential Revision: D6780165 Pulled By: siying fbshipit-source-id: 273f7dc896724f682c0118fb69a359d9cc4418b4
2026-11-06 11:02:36 -05:00 · 2018-01-25 15:34:12 -08:00 · 2018-01-25 14:57:18 -08:00 · 2018-01-23 16:57:11 -08:00 · 2018-01-23 16:57:11 -08:00 · 2018-01-23 16:12:19 -08:00
76 changed files with 1119 additions and 409 deletions
@@ -34,12 +34,15 @@

 cmake_minimum_required(VERSION 2.8.12)
 project(rocksdb)
+enable_language(CXX)
+enable_language(C)
+enable_language(ASM)

 if(POLICY CMP0042)
  cmake_policy(SET CMP0042 NEW)
 endif()

-list(APPEND CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/cmake/modules/")
+list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_LIST_DIR}/cmake/modules/")

 option(WITH_JEMALLOC "build with JeMalloc" OFF)
 option(WITH_SNAPPY "build with SNAPPY" OFF)
@@ -129,20 +132,19 @@ endif()

 string(REGEX REPLACE "[^0-9a-f]+" "" GIT_SHA "${GIT_SHA}")

-if(NOT WIN32)
-  execute_process(COMMAND
-      "./build_tools/version.sh" "full"
-      WORKING_DIRECTORY ${PROJECT_SOURCE_DIR}
-      OUTPUT_VARIABLE ROCKSDB_VERSION
-  )
-  string(STRIP "${ROCKSDB_VERSION}" ROCKSDB_VERSION)
-  execute_process(COMMAND
-      "./build_tools/version.sh" "major"
-      WORKING_DIRECTORY ${PROJECT_SOURCE_DIR}
-      OUTPUT_VARIABLE ROCKSDB_VERSION_MAJOR
-  )
-  string(STRIP "${ROCKSDB_VERSION_MAJOR}" ROCKSDB_VERSION_MAJOR)
-endif()
+set(SH_CMD "sh")
+execute_process(COMMAND
+  ${SH_CMD} -c "build_tools/version.sh full"
+  WORKING_DIRECTORY ${PROJECT_SOURCE_DIR}
+  OUTPUT_VARIABLE ROCKSDB_VERSION
+)
+string(STRIP "${ROCKSDB_VERSION}" ROCKSDB_VERSION)
+execute_process(COMMAND
+  ${SH_CMD} -c "build_tools/version.sh major"
+  WORKING_DIRECTORY ${PROJECT_SOURCE_DIR}
+  OUTPUT_VARIABLE ROCKSDB_VERSION_MAJOR
+)
+string(STRIP "${ROCKSDB_VERSION_MAJOR}" ROCKSDB_VERSION_MAJOR)

 option(WITH_MD_LIBRARY "build with MD" ON)
 if(WIN32 AND MSVC)
@@ -178,6 +180,18 @@ else()
  endif()
 endif()

+include(CheckCCompilerFlag)
+if(CMAKE_SYSTEM_PROCESSOR MATCHES "ppc64le")
+  CHECK_C_COMPILER_FLAG("-maltivec" HAS_ALTIVEC)
+  if(HAS_ALTIVEC)
+    message(STATUS " HAS_ALTIVEC yes")
+    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -maltivec")
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -maltivec")
+    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mcpu=power8")
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mcpu=power8")
+  endif(HAS_ALTIVEC)
+endif(CMAKE_SYSTEM_PROCESSOR MATCHES "ppc64le")
+
 option(PORTABLE "build a portable binary" OFF)
 option(FORCE_SSE42 "force building with SSE4.2, even when PORTABLE=ON" OFF)
 if(PORTABLE)
@@ -190,7 +204,9 @@ else()
  if(MSVC)
    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /arch:AVX2")
  else()
-    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -march=native")
+    if(NOT HAVE_POWER8)
+      set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -march=native")
+    endif()
  endif()
 endif()

@@ -626,6 +642,12 @@ set_source_files_properties(
  endif()
 endif()

+if(HAVE_POWER8)
+  list(APPEND SOURCES
+    util/crc32c_ppc.c
+    util/crc32c_ppc_asm.S)
+endif(HAVE_POWER8)
+
 if(WIN32)
  list(APPEND SOURCES
    port/win/io_win.cc
@@ -720,7 +742,7 @@ if(NOT WIN32 OR ROCKSDB_INSTALL_ON_WINDOWS)
  set(package_config_destination ${CMAKE_INSTALL_LIBDIR}/cmake/rocksdb)

  configure_package_config_file(
-    ${CMAKE_SOURCE_DIR}/cmake/RocksDBConfig.cmake.in RocksDBConfig.cmake
+    ${CMAKE_CURRENT_LIST_DIR}/cmake/RocksDBConfig.cmake.in RocksDBConfig.cmake
    INSTALL_DESTINATION ${package_config_destination}
  )

@@ -950,17 +972,18 @@ if(WITH_TESTS)

  foreach(sourcefile ${TEST_EXES})
      get_filename_component(exename ${sourcefile} NAME_WE)
-      add_executable(${exename}${ARTIFACT_SUFFIX} ${sourcefile}
+      add_executable(${CMAKE_PROJECT_NAME}_${exename}${ARTIFACT_SUFFIX} ${sourcefile}
        $<TARGET_OBJECTS:testharness>)
-      set_target_properties(${exename}${ARTIFACT_SUFFIX}
+      set_target_properties(${CMAKE_PROJECT_NAME}_${exename}${ARTIFACT_SUFFIX}
        PROPERTIES EXCLUDE_FROM_DEFAULT_BUILD_RELEASE 1
        EXCLUDE_FROM_DEFAULT_BUILD_MINRELEASE 1
        EXCLUDE_FROM_DEFAULT_BUILD_RELWITHDEBINFO 1
+        OUTPUT_NAME ${exename}${ARTIFACT_SUFFIX}
        )
-      target_link_libraries(${exename}${ARTIFACT_SUFFIX} testutillib${ARTIFACT_SUFFIX} gtest ${LIBS})
+      target_link_libraries(${CMAKE_PROJECT_NAME}_${exename}${ARTIFACT_SUFFIX} testutillib${ARTIFACT_SUFFIX} gtest ${LIBS})
      if(NOT "${exename}" MATCHES "db_sanity_test")
        add_test(NAME ${exename} COMMAND ${exename}${ARTIFACT_SUFFIX})
-        add_dependencies(check ${exename}${ARTIFACT_SUFFIX})
+        add_dependencies(check ${CMAKE_PROJECT_NAME}_${exename}${ARTIFACT_SUFFIX})
      endif()
  endforeach(sourcefile ${TEST_EXES})

@@ -1,5 +1,10 @@
 # Rocksdb Change Log
 ## Unreleased
+### Public API Change
+* Iterator::SeekForPrev is now a pure virtual method. This is to prevent user who implement the Iterator interface fail to implement SeekForPrev by mistake.
+### Bug Fixes
+* Fix `DisableFileDeletions()` followed by `GetSortedWalFiles()` to not return obsolete WAL files that `PurgeObsoleteFiles()` is going to delete.
+* Fix DB::Flush() keep waiting after flush finish under certain condition.

 ## 5.10.0 (12/11/2017)
 ### Public API Change
@@ -107,6 +107,41 @@ to build a portable binary, add `PORTABLE=1` before your make commands, like thi
            * run `brew tap homebrew/versions; brew install gcc48 --use-llvm` to install gcc 4.8 (or higher).
    * run `brew install rocksdb`

+* **FreeBSD** (11.01):
+
+    * You can either install RocksDB from the Ports system using `cd /usr/ports/databases/rocksdb && make install`, or you can follow the details below to install dependencies and compile from source code:
+
+    * Install the dependencies for RocksDB:
+
+        export BATCH=YES
+        cd /usr/ports/devel/gmake && make install
+        cd /usr/ports/devel/gflags && make install
+
+        cd /usr/ports/archivers/snappy && make install
+        cd /usr/ports/archivers/bzip2 && make install
+        cd /usr/ports/archivers/liblz4 && make install
+        cd /usr/ports/archivesrs/zstd && make install
+
+        cd /usr/ports/devel/git && make install
+
+
+    * Install the dependencies for RocksJava (optional):
+
+        export BATCH=yes
+        cd /usr/ports/java/openjdk7 && make install
+
+    * Build RocksDB from source:
+        cd ~
+        git clone https://github.com/facebook/rocksdb.git
+        cd rocksdb
+        gmake static_lib
+
+    * Build RocksJava from source (optional):
+        cd rocksdb
+        export JAVA_HOME=/usr/local/openjdk7
+        gmake rocksdbjava
+
+
 * **iOS**:
  * Run: `TARGET_OS=IOS make static_lib`. When building the project which uses rocksdb iOS library, make sure to define two important pre-processing macros: `ROCKSDB_LITE` and `IOS_CROSS_COMPILE`.

@@ -1553,8 +1553,8 @@ ZLIB_DOWNLOAD_BASE ?= http://zlib.net
 BZIP2_VER ?= 1.0.6
 BZIP2_SHA256 ?= a2848f34fcd5d6cf47def00461fcb528a0484d8edef8208d6d2e2909dc61d9cd
 BZIP2_DOWNLOAD_BASE ?= http://www.bzip.org
-SNAPPY_VER ?= 1.1.7
-SNAPPY_SHA256 ?= 3dfa02e873ff51a11ee02b9ca391807f0c8ea0529a4924afa645fbf97163f9d4
+SNAPPY_VER ?= 1.1.4
+SNAPPY_SHA256 ?= 134bfe122fd25599bb807bb8130e7ba6d9bdb851e0b16efcb83ac4f5d0b70057
 SNAPPY_DOWNLOAD_BASE ?= https://github.com/google/snappy/releases/download
 LZ4_VER ?= 1.8.0
 LZ4_SHA256 ?= 2ca482ea7a9bb103603108b5a7510b7592b90158c151ff50a28f1ca8389fccf6
@@ -1574,7 +1574,7 @@ else
 endif
 endif
 ifeq ($(PLATFORM), OS_FREEBSD)
-	JAVA_INCLUDE += -I$(JAVA_HOME)/include/freebsd
+	JAVA_INCLUDE = -I$(JAVA_HOME)/include -I$(JAVA_HOME)/include/freebsd
 	ROCKSDBJNILIB = librocksdbjni-freebsd$(ARCH).so
 	ROCKSDB_JAR = rocksdbjni-$(ROCKSDB_MAJOR).$(ROCKSDB_MINOR).$(ROCKSDB_PATCH)-freebsd$(ARCH).jar
 endif
@@ -1600,7 +1600,7 @@ libz.a:
 		exit 1; \
 	fi
 	tar xvzf zlib-$(ZLIB_VER).tar.gz
-	cd zlib-$(ZLIB_VER) && CFLAGS='-fPIC ${EXTRA_CFLAGS}' LDFLAGS='${EXTRA_LDFLAGS}' ./configure --static && make
+	cd zlib-$(ZLIB_VER) && CFLAGS='-fPIC ${EXTRA_CFLAGS}' LDFLAGS='${EXTRA_LDFLAGS}' ./configure --static && $(MAKE)
 	cp zlib-$(ZLIB_VER)/libz.a .

 libbz2.a:
@@ -1612,7 +1612,7 @@ libbz2.a:
 		exit 1; \
 	fi
 	tar xvzf bzip2-$(BZIP2_VER).tar.gz
-	cd bzip2-$(BZIP2_VER) && make CFLAGS='-fPIC -O2 -g -D_FILE_OFFSET_BITS=64 ${EXTRA_CFLAGS}' AR='ar ${EXTRA_ARFLAGS}'
+	cd bzip2-$(BZIP2_VER) && $(MAKE) CFLAGS='-fPIC -O2 -g -D_FILE_OFFSET_BITS=64 ${EXTRA_CFLAGS}' AR='ar ${EXTRA_ARFLAGS}'
 	cp bzip2-$(BZIP2_VER)/libbz2.a .

 libsnappy.a:
@@ -1625,7 +1625,7 @@ libsnappy.a:
 	fi
 	tar xvzf snappy-$(SNAPPY_VER).tar.gz
 	cd snappy-$(SNAPPY_VER) && CFLAGS='${EXTRA_CFLAGS}' CXXFLAGS='${EXTRA_CXXFLAGS}' LDFLAGS='${EXTRA_LDFLAGS}' ./configure --with-pic --enable-static --disable-shared
-	cd snappy-$(SNAPPY_VER) && make ${SNAPPY_MAKE_TARGET}
+	cd snappy-$(SNAPPY_VER) && $(MAKE) ${SNAPPY_MAKE_TARGET}
 	cp snappy-$(SNAPPY_VER)/.libs/libsnappy.a .

 liblz4.a:
@@ -1638,7 +1638,7 @@ liblz4.a:
 		exit 1; \
 	fi
 	tar xvzf lz4-$(LZ4_VER).tar.gz
-	cd lz4-$(LZ4_VER)/lib && make CFLAGS='-fPIC -O2 ${EXTRA_CFLAGS}' all
+	cd lz4-$(LZ4_VER)/lib && $(MAKE) CFLAGS='-fPIC -O2 ${EXTRA_CFLAGS}' all
 	cp lz4-$(LZ4_VER)/lib/liblz4.a .

 libzstd.a:
@@ -1651,7 +1651,7 @@ libzstd.a:
 		exit 1; \
 	fi
 	tar xvzf zstd-$(ZSTD_VER).tar.gz
-	cd zstd-$(ZSTD_VER)/lib && DESTDIR=. PREFIX= make CFLAGS='-fPIC -O2 ${EXTRA_CFLAGS}' install
+	cd zstd-$(ZSTD_VER)/lib && DESTDIR=. PREFIX= $(MAKE) CFLAGS='-fPIC -O2 ${EXTRA_CFLAGS}' install
 	cp zstd-$(ZSTD_VER)/lib/libzstd.a .

 # A version of each $(LIBOBJECTS) compiled with -fPIC and a fixed set of static compression libraries
@@ -1750,7 +1750,7 @@ JAVA_C_LIBOBJECTS = $(patsubst %.c.o,jl/%.c.o,$(JAVA_C_OBJECTS))
 JAVA_ASM_LIBOBJECTS = $(patsubst %.S.o,jl/%.S.o,$(JAVA_ASM_OBJECTS))
 endif

-java_libobjects = $(patsubst %,jl/%,$(LIBOBJECTS))
+java_libobjects = $(patsubst %,jl/%,$(LIB_CC_OBJECTS))
 CLEAN_FILES += jl
 java_all_libobjects = $(java_libobjects)

@@ -1,7 +1,8 @@
 ## RocksDB: A Persistent Key-Value Store for Flash and RAM Storage

-[![Build Status](https://travis-ci.org/facebook/rocksdb.svg?branch=master)](https://travis-ci.org/facebook/rocksdb)
-[![Build status](https://ci.appveyor.com/api/projects/status/fbgfu0so3afcno78/branch/master?svg=true)](https://ci.appveyor.com/project/Facebook/rocksdb/branch/master)
+[![Linux/Mac Build Status](https://travis-ci.org/facebook/rocksdb.svg?branch=master)](https://travis-ci.org/facebook/rocksdb)
+[![Windows Build status](https://ci.appveyor.com/api/projects/status/fbgfu0so3afcno78/branch/master?svg=true)](https://ci.appveyor.com/project/Facebook/rocksdb/branch/master)
+[![PPC64le Build Status](http://140.211.168.68:8080/buildStatus/icon?job=Rocksdb)](http://140.211.168.68:8080/job/Rocksdb)


 RocksDB is developed and maintained by Facebook Database Engineering Team.
@@ -141,6 +141,7 @@ case "$TARGET_OS" in
        ;;
    FreeBSD)
        PLATFORM=OS_FREEBSD
+        CXX=clang++
        COMMON_FLAGS="$COMMON_FLAGS -fno-builtin-memcmp -D_REENTRANT -DOS_FREEBSD"
        PLATFORM_LDFLAGS="$PLATFORM_LDFLAGS -lpthread"
        # PORT_FILES=port/freebsd/freebsd_specific.cc
@@ -481,6 +482,9 @@ if test -z "$PORTABLE"; then
    COMMON_FLAGS="$COMMON_FLAGS -mcpu=$POWER -mtune=$POWER "
  elif test -n "`echo $TARGET_ARCHITECTURE | grep ^s390x`"; then
    COMMON_FLAGS="$COMMON_FLAGS -march=z10 "
+  elif test -n "`echo $TARGET_ARCHITECTURE | grep ^arm`"; then
+    # TODO: Handle this with approprite options.
+    COMMON_FLAGS="$COMMON_FLAGS"
  elif [ "$TARGET_OS" != AIX ] && [ "$TARGET_OS" != SunOS ]; then
    COMMON_FLAGS="$COMMON_FLAGS -march=native "
  elif test "$USE_SSE"; then
@@ -385,7 +385,8 @@ ColumnFamilyData::ColumnFamilyData(
      pending_flush_(false),
      pending_compaction_(false),
      prev_compaction_needed_bytes_(0),
-      allow_2pc_(db_options.allow_2pc) {
+      allow_2pc_(db_options.allow_2pc),
+      last_memtable_id_(0) {
  Ref();

  // Convert user defined table properties collector factories to internal ones.
@@ -1059,10 +1060,12 @@ ColumnFamilySet::~ColumnFamilySet() {
  while (column_family_data_.size() > 0) {
    // cfd destructor will delete itself from column_family_data_
    auto cfd = column_family_data_.begin()->second;
-    cfd->Unref();
+    bool last_ref __attribute__((__unused__)) = cfd->Unref();
+    assert(last_ref);
    delete cfd;
  }
-  dummy_cfd_->Unref();
+  bool dummy_last_ref __attribute__((__unused__)) = dummy_cfd_->Unref();
+  assert(dummy_last_ref);
  delete dummy_cfd_;
 }

@@ -239,7 +239,11 @@ class ColumnFamilyData {
  void SetCurrent(Version* _current);
  uint64_t GetNumLiveVersions() const;  // REQUIRE: DB mutex held
  uint64_t GetTotalSstFilesSize() const;  // REQUIRE: DB mutex held
-  void SetMemtable(MemTable* new_mem) { mem_ = new_mem; }
+  void SetMemtable(MemTable* new_mem) {
+    uint64_t memtable_id = last_memtable_id_.fetch_add(1) + 1;
+    new_mem->SetID(memtable_id);
+    mem_ = new_mem;
+  }

  // calculate the oldest log needed for the durability of this column family
  uint64_t OldestLogToKeep();
@@ -419,6 +423,9 @@ class ColumnFamilyData {

  // if the database was opened with 2pc enabled
  bool allow_2pc_;
+
+  // Memtable id to track flush.
+  std::atomic<uint64_t> last_memtable_id_;
 };

 // ColumnFamilySet has interesting thread-safety requirements
@@ -1868,126 +1868,6 @@ TEST_F(ColumnFamilyTest, SameCFManualAutomaticCompactionsLevel) {
  }
 }

-// This test checks for automatic getting a conflict if there is a
-// manual which has not yet been scheduled.
-// The manual compaction waits in NotScheduled
-// We generate more files and then trigger an automatic compaction
-// This will wait because there is an unscheduled manual compaction.
-// Once the conflict is hit, the manual compaction starts and ends
-// Then another automatic will start and end.
-TEST_F(ColumnFamilyTest, SameCFManualAutomaticConflict) {
-  Open();
-  CreateColumnFamilies({"one"});
-  ColumnFamilyOptions default_cf, one;
-  db_options_.max_open_files = 20;  // only 10 files in file cache
-  db_options_.max_background_compactions = 3;
-
-  default_cf.compaction_style = kCompactionStyleLevel;
-  default_cf.num_levels = 3;
-  default_cf.write_buffer_size = 64 << 10;  // 64KB
-  default_cf.target_file_size_base = 30 << 10;
-  default_cf.max_compaction_bytes = default_cf.target_file_size_base * 1100;
-  BlockBasedTableOptions table_options;
-  table_options.no_block_cache = true;
-  default_cf.table_factory.reset(NewBlockBasedTableFactory(table_options));
-
-  one.compaction_style = kCompactionStyleUniversal;
-
-  one.num_levels = 1;
-  // trigger compaction if there are >= 4 files
-  one.level0_file_num_compaction_trigger = 4;
-  one.write_buffer_size = 120000;
-
-  Reopen({default_cf, one});
-  // make sure all background compaction jobs can be scheduled
-  auto stop_token =
-      dbfull()->TEST_write_controler().GetCompactionPressureToken();
-
-  // SETUP column family "one" -- universal style
-  for (int i = 0; i < one.level0_file_num_compaction_trigger - 2; ++i) {
-    PutRandomData(1, 10, 12000, true);
-    PutRandomData(1, 1, 10, true);
-    WaitForFlush(1);
-    AssertFilesPerLevel(ToString(i + 1), 1);
-  }
-  bool cf_1_1 = true;
-  bool cf_1_2 = true;
-  rocksdb::SyncPoint::GetInstance()->LoadDependency(
-      {{"DBImpl::BackgroundCompaction()::Conflict",
-        "ColumnFamilyTest::ManualAutoCon:7"},
-       {"ColumnFamilyTest::ManualAutoCon:9",
-        "ColumnFamilyTest::ManualAutoCon:8"},
-       {"ColumnFamilyTest::ManualAutoCon:2",
-        "ColumnFamilyTest::ManualAutoCon:6"},
-       {"ColumnFamilyTest::ManualAutoCon:4",
-        "ColumnFamilyTest::ManualAutoCon:5"},
-       {"ColumnFamilyTest::ManualAutoCon:1",
-        "ColumnFamilyTest::ManualAutoCon:2"},
-       {"ColumnFamilyTest::ManualAutoCon:1",
-        "ColumnFamilyTest::ManualAutoCon:3"}});
-  rocksdb::SyncPoint::GetInstance()->SetCallBack(
-      "DBImpl::BackgroundCompaction:NonTrivial:AfterRun", [&](void* arg) {
-        if (cf_1_1) {
-          TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:4");
-          cf_1_1 = false;
-          TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:3");
-        } else if (cf_1_2) {
-          cf_1_2 = false;
-          TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:2");
-        }
-      });
-  rocksdb::SyncPoint::GetInstance()->SetCallBack(
-      "DBImpl::RunManualCompaction:NotScheduled", [&](void* arg) {
-        InstrumentedMutex* mutex = static_cast<InstrumentedMutex*>(arg);
-        mutex->Unlock();
-        TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:9");
-        TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:7");
-        mutex->Lock();
-      });
-
-  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
-  rocksdb::port::Thread threads([&] {
-    CompactRangeOptions compact_options;
-    compact_options.exclusive_manual_compaction = false;
-    ASSERT_OK(
-        db_->CompactRange(compact_options, handles_[1], nullptr, nullptr));
-    TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:6");
-  });
-
-  TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:8");
-  WaitForFlush(1);
-
-  // Add more L0 files and force automatic compaction
-  for (int i = 0; i < one.level0_file_num_compaction_trigger; ++i) {
-    PutRandomData(1, 10, 12000, true);
-    PutRandomData(1, 1, 10, true);
-    WaitForFlush(1);
-    AssertFilesPerLevel(ToString(one.level0_file_num_compaction_trigger + i),
-                        1);
-  }
-
-  TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:5");
-  // Add more L0 files and force automatic compaction
-  for (int i = 0; i < one.level0_file_num_compaction_trigger; ++i) {
-    PutRandomData(1, 10, 12000, true);
-    PutRandomData(1, 1, 10, true);
-    WaitForFlush(1);
-  }
-  TEST_SYNC_POINT("ColumnFamilyTest::ManualAutoCon:1");
-
-  threads.join();
-  WaitForCompaction();
-  // VERIFY compaction "one"
-  ASSERT_LE(NumTableFilesAtLevel(0, 1), 3);
-
-  // Compare against saved keys
-  std::set<std::string>::iterator key_iter = keys_.begin();
-  while (key_iter != keys_.end()) {
-    ASSERT_NE("NOT_FOUND", Get(1, *key_iter));
-    key_iter++;
-  }
-}
-
 // In this test, we generate enough files to trigger automatic compactions.
 // The automatic compaction waits in NonTrivial:AfterRun
 // We generate more files and then trigger an automatic compaction
@@ -405,20 +405,23 @@ void Compaction::Summary(char* output, int len) {
 uint64_t Compaction::OutputFilePreallocationSize() const {
  uint64_t preallocation_size = 0;

-  if (max_output_file_size_ != port::kMaxUint64 &&
-      (cfd_->ioptions()->compaction_style == kCompactionStyleLevel ||
-       output_level() > 0)) {
-    preallocation_size = max_output_file_size_;
-  } else {
-    for (const auto& level_files : inputs_) {
-      for (const auto& file : level_files.files) {
-        preallocation_size += file->fd.GetFileSize();
-      }
+  for (const auto& level_files : inputs_) {
+    for (const auto& file : level_files.files) {
+      preallocation_size += file->fd.GetFileSize();
    }
  }
+
+  if (max_output_file_size_ != port::kMaxUint64 &&
+      (immutable_cf_options_.compaction_style == kCompactionStyleLevel ||
+       output_level() > 0)) {
+    preallocation_size = std::min(max_output_file_size_, preallocation_size);
+  }
+
  // Over-estimate slightly so we don't end up just barely crossing
  // the threshold
-  return preallocation_size + (preallocation_size / 10);
+  // No point to prellocate more than 1GB.
+  return std::min(uint64_t{1073741824},
+                  preallocation_size + (preallocation_size / 10));
 }

 std::unique_ptr<CompactionFilter> Compaction::CreateCompactionFilter() const {
@@ -175,6 +175,8 @@ TEST_F(CompactionPickerTest, Level1Trigger) {
 }

 TEST_F(CompactionPickerTest, Level1Trigger2) {
+  mutable_cf_options_.target_file_size_base = 10000000000;
+  mutable_cf_options_.RefreshDerivedOptions(ioptions_);
  NewVersionStorage(6, kCompactionStyleLevel);
  Add(1, 66U, "150", "200", 1000000001U);
  Add(1, 88U, "201", "300", 1000000000U);
@@ -191,13 +193,14 @@ TEST_F(CompactionPickerTest, Level1Trigger2) {
  ASSERT_EQ(66U, compaction->input(0, 0)->fd.GetNumber());
  ASSERT_EQ(6U, compaction->input(1, 0)->fd.GetNumber());
  ASSERT_EQ(7U, compaction->input(1, 1)->fd.GetNumber());
+  ASSERT_EQ(uint64_t{1073741824}, compaction->OutputFilePreallocationSize());
 }

 TEST_F(CompactionPickerTest, LevelMaxScore) {
  NewVersionStorage(6, kCompactionStyleLevel);
  mutable_cf_options_.target_file_size_base = 10000000;
-  mutable_cf_options_.target_file_size_multiplier = 10;
  mutable_cf_options_.max_bytes_for_level_base = 10 * 1024 * 1024;
+  mutable_cf_options_.RefreshDerivedOptions(ioptions_);
  Add(0, 1U, "150", "200", 1000000U);
  // Level 1 score 1.2
  Add(1, 66U, "150", "200", 6000000U);
@@ -218,6 +221,9 @@ TEST_F(CompactionPickerTest, LevelMaxScore) {
  ASSERT_TRUE(compaction.get() != nullptr);
  ASSERT_EQ(1U, compaction->num_input_files(0));
  ASSERT_EQ(7U, compaction->input(0, 0)->fd.GetNumber());
+  ASSERT_EQ(mutable_cf_options_.target_file_size_base +
+                mutable_cf_options_.target_file_size_base / 10,
+            compaction->OutputFilePreallocationSize());
 }

 TEST_F(CompactionPickerTest, NeedsCompactionLevel) {
@@ -521,9 +527,10 @@ TEST_F(CompactionPickerTest, NeedsCompactionFIFO) {
 TEST_F(CompactionPickerTest, CompactionPriMinOverlapping1) {
  NewVersionStorage(6, kCompactionStyleLevel);
  ioptions_.compaction_pri = kMinOverlappingRatio;
-  mutable_cf_options_.target_file_size_base = 10000000;
+  mutable_cf_options_.target_file_size_base = 100000000000;
  mutable_cf_options_.target_file_size_multiplier = 10;
  mutable_cf_options_.max_bytes_for_level_base = 10 * 1024 * 1024;
+  mutable_cf_options_.RefreshDerivedOptions(ioptions_);

  Add(2, 6U, "150", "179", 50000000U);
  Add(2, 7U, "180", "220", 50000000U);
@@ -543,6 +550,8 @@ TEST_F(CompactionPickerTest, CompactionPriMinOverlapping1) {
  ASSERT_EQ(1U, compaction->num_input_files(0));
  // Pick file 8 because it overlaps with 0 files on level 3.
  ASSERT_EQ(8U, compaction->input(0, 0)->fd.GetNumber());
+  // Compaction input size * 1.1
+  ASSERT_GE(uint64_t{55000000}, compaction->OutputFilePreallocationSize());
 }

 TEST_F(CompactionPickerTest, CompactionPriMinOverlapping2) {
@@ -570,19 +570,19 @@ TEST_F(DBBasicTest, CompactBetweenSnapshots) {

 TEST_F(DBBasicTest, DBOpen_Options) {
  Options options = CurrentOptions();
-  std::string dbname = test::TmpDir(env_) + "/db_options_test";
-  ASSERT_OK(DestroyDB(dbname, options));
+  Close();
+  Destroy(options);

  // Does not exist, and create_if_missing == false: error
  DB* db = nullptr;
  options.create_if_missing = false;
-  Status s = DB::Open(options, dbname, &db);
+  Status s = DB::Open(options, dbname_, &db);
  ASSERT_TRUE(strstr(s.ToString().c_str(), "does not exist") != nullptr);
  ASSERT_TRUE(db == nullptr);

  // Does not exist, and create_if_missing == true: OK
  options.create_if_missing = true;
-  s = DB::Open(options, dbname, &db);
+  s = DB::Open(options, dbname_, &db);
  ASSERT_OK(s);
  ASSERT_TRUE(db != nullptr);

@@ -592,14 +592,14 @@ TEST_F(DBBasicTest, DBOpen_Options) {
  // Does exist, and error_if_exists == true: error
  options.create_if_missing = false;
  options.error_if_exists = true;
-  s = DB::Open(options, dbname, &db);
+  s = DB::Open(options, dbname_, &db);
  ASSERT_TRUE(strstr(s.ToString().c_str(), "exists") != nullptr);
  ASSERT_TRUE(db == nullptr);

  // Does exist, and error_if_exists == false: OK
  options.create_if_missing = true;
  options.error_if_exists = false;
-  s = DB::Open(options, dbname, &db);
+  s = DB::Open(options, dbname_, &db);
  ASSERT_OK(s);
  ASSERT_TRUE(db != nullptr);

@@ -847,6 +847,57 @@ TEST_F(DBBasicTest, MmapAndBufferOptions) {
 }
 #endif

+class TestEnv : public EnvWrapper {
+  public:
+    explicit TestEnv(Env* base) : EnvWrapper(base) { };
+
+    class TestLogger : public Logger {
+      public:
+        using Logger::Logv;
+        virtual void Logv(const char *format, va_list ap) override { };
+      private:
+        virtual Status CloseImpl() override {
+          return Status::NotSupported();
+        }
+    };
+
+    virtual Status NewLogger(const std::string& fname,
+                             shared_ptr<Logger>* result) {
+      result->reset(new TestLogger());
+      return Status::OK();
+    }
+};
+
+TEST_F(DBBasicTest, DBClose) {
+  Options options = GetDefaultOptions();
+  std::string dbname = test::TmpDir(env_) + "/db_close_test";
+  ASSERT_OK(DestroyDB(dbname, options));
+
+  DB* db = nullptr;
+  options.create_if_missing = true;
+  options.env = new TestEnv(Env::Default());
+  Status s = DB::Open(options, dbname, &db);
+  ASSERT_OK(s);
+  ASSERT_TRUE(db != nullptr);
+
+  s = db->Close();
+  ASSERT_EQ(s, Status::NotSupported());
+
+  delete db;
+
+  // Provide our own logger and ensure DB::Close() does not close it
+  options.info_log.reset(new TestEnv::TestLogger());
+  options.create_if_missing = false;
+  s = DB::Open(options, dbname, &db);
+  ASSERT_OK(s);
+  ASSERT_TRUE(db != nullptr);
+
+  s = db->Close();
+  ASSERT_EQ(s, Status::OK());
+  delete db;
+  delete options.env;
+}
+
 }  // namespace rocksdb

 int main(int argc, char** argv) {
@@ -57,6 +57,7 @@ Status DBImpl::EnableFileDeletions(bool force) {
      ROCKS_LOG_INFO(immutable_db_options_.info_log, "File Deletions Enabled");
      should_purge_files = true;
      FindObsoleteFiles(&job_context, true);
+      bg_cv_.SignalAll();
    } else {
      ROCKS_LOG_WARN(
          immutable_db_options_.info_log,
@@ -141,6 +142,18 @@ Status DBImpl::GetLiveFiles(std::vector<std::string>& ret,
 }

 Status DBImpl::GetSortedWalFiles(VectorLogPtr& files) {
+  {
+    // If caller disabled deletions, this function should return files that are
+    // guaranteed not to be deleted until deletions are re-enabled. We need to
+    // wait for pending purges to finish since WalManager doesn't know which
+    // files are going to be purged. Additional purges won't be scheduled as
+    // long as deletions are disabled (so the below loop must terminate).
+    InstrumentedMutexLock l(&mutex_);
+    while (disable_delete_obsolete_files_ > 0 &&
+           pending_purge_obsolete_files_ > 0) {
+      bg_cv_.Wait();
+    }
+  }
  return wal_manager_.GetSortedWalFiles(files);
 }

@@ -126,6 +126,41 @@ TEST_F(DBFlushTest, FlushInLowPriThreadPool) {
  ASSERT_EQ(1, num_compactions);
 }

+TEST_F(DBFlushTest, ManualFlushWithMinWriteBufferNumberToMerge) {
+  Options options = CurrentOptions();
+  options.write_buffer_size = 100;
+  options.max_write_buffer_number = 4;
+  options.min_write_buffer_number_to_merge = 3;
+  Reopen(options);
+
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"DBImpl::BGWorkFlush",
+        "DBFlushTest::ManualFlushWithMinWriteBufferNumberToMerge:1"},
+       {"DBFlushTest::ManualFlushWithMinWriteBufferNumberToMerge:2",
+        "DBImpl::FlushMemTableToOutputFile:BeforeInstallSV"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ASSERT_OK(Put("key1", "value1"));
+
+  port::Thread t([&]() {
+    // The call wait for flush to finish, i.e. with flush_options.wait = true.
+    ASSERT_OK(Flush());
+  });
+
+  // Wait for flush start.
+  TEST_SYNC_POINT("DBFlushTest::ManualFlushWithMinWriteBufferNumberToMerge:1");
+  // Insert a second memtable before the manual flush finish.
+  // At the end of the manual flush job, it will check if further flush
+  // is needed, but it will not trigger flush of the second memtable because
+  // min_write_buffer_number_to_merge is not reached.
+  ASSERT_OK(Put("key2", "value2"));
+  ASSERT_OK(dbfull()->TEST_SwitchMemtable());
+  TEST_SYNC_POINT("DBFlushTest::ManualFlushWithMinWriteBufferNumberToMerge:2");
+
+  // Manual flush should return, without waiting for flush indefinitely.
+  t.join();
+}
+
 TEST_P(DBFlushDirectIOTest, DirectIO) {
  Options options;
  options.create_if_missing = true;
@@ -123,13 +123,13 @@ CompressionType GetCompressionFlush(
 namespace {
 void DumpSupportInfo(Logger* logger) {
  ROCKS_LOG_HEADER(logger, "Compression algorithms supported:");
-  ROCKS_LOG_HEADER(logger, "\tSnappy supported: %d", Snappy_Supported());
-  ROCKS_LOG_HEADER(logger, "\tZlib supported: %d", Zlib_Supported());
-  ROCKS_LOG_HEADER(logger, "\tBzip supported: %d", BZip2_Supported());
-  ROCKS_LOG_HEADER(logger, "\tLZ4 supported: %d", LZ4_Supported());
-  ROCKS_LOG_HEADER(logger, "\tZSTDNotFinal supported: %d",
-                   ZSTDNotFinal_Supported());
-  ROCKS_LOG_HEADER(logger, "\tZSTD supported: %d", ZSTD_Supported());
+  for (auto& compression : OptionsHelper::compression_type_string_map) {
+    if (compression.second != kNoCompression &&
+        compression.second != kDisableCompressionOption) {
+      ROCKS_LOG_HEADER(logger, "\t%s supported: %d", compression.first.c_str(),
+                       CompressionTypeSupported(compression.second));
+    }
+  }
  ROCKS_LOG_HEADER(logger, "Fast CRC32 supported: %s",
                   crc32c::IsFastCrc32Supported().c_str());
 }
@@ -141,6 +141,7 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
               const bool seq_per_batch)
    : env_(options.env),
      dbname_(dbname),
+      own_info_log_(options.info_log == nullptr),
      initial_db_options_(SanitizeOptions(dbname, options)),
      immutable_db_options_(initial_db_options_),
      mutable_db_options_(initial_db_options_),
@@ -177,6 +178,7 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
      num_running_flushes_(0),
      bg_purge_scheduled_(0),
      disable_delete_obsolete_files_(0),
+      pending_purge_obsolete_files_(0),
      delete_obsolete_files_last_run_(env_->NowMicros()),
      last_stats_dump_time_microsec_(0),
      next_job_id_(1),
@@ -212,7 +214,8 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
      // requires a custom gc for compaction, we use that to set use_custom_gc_
      // as well.
      use_custom_gc_(seq_per_batch),
-      preserve_deletes_(options.preserve_deletes) {
+      preserve_deletes_(options.preserve_deletes),
+      closed_(false) {
  env_->GetAbsolutePath(dbname, &db_absolute_path_);

  // Reserve ten files or so for other uses and give the rest to TableCache.
@@ -275,7 +278,7 @@ void DBImpl::CancelAllBackgroundWork(bool wait) {
  }
 }

-DBImpl::~DBImpl() {
+Status DBImpl::CloseImpl() {
  // CancelAllBackgroundWork called with false means we just set the shutdown
  // marker. After this we do a variant of the waiting and unschedule work
  // (to consider: moving all the waiting into CancelAllBackgroundWork(true))
@@ -291,7 +294,8 @@ DBImpl::~DBImpl() {

  // Wait for background work to finish
  while (bg_bottom_compaction_scheduled_ || bg_compaction_scheduled_ ||
-         bg_flush_scheduled_ || bg_purge_scheduled_) {
+         bg_flush_scheduled_ || bg_purge_scheduled_ ||
+         pending_purge_obsolete_files_) {
    TEST_SYNC_POINT("DBImpl::~DBImpl:WaitJob");
    bg_cv_.Wait();
  }
@@ -378,8 +382,16 @@ DBImpl::~DBImpl() {

  ROCKS_LOG_INFO(immutable_db_options_.info_log, "Shutdown complete");
  LogFlush(immutable_db_options_.info_log);
+
+  Status s = Status::OK();
+  if (immutable_db_options_.info_log && own_info_log_) {
+    s = immutable_db_options_.info_log->Close();
+  }
+  return s;
 }

+DBImpl::~DBImpl() { Close(); }
+
 void DBImpl::MaybeIgnoreError(Status* s) const {
  if (s->ok() || immutable_db_options_.paranoid_checks) {
    // No change needed
@@ -2047,8 +2059,7 @@ Status DBImpl::DeleteFile(std::string name) {
                      name.c_str());
      return Status::NotSupported("Delete only supported for archived logs");
    }
-    status =
-        env_->DeleteFile(immutable_db_options_.wal_dir + "/" + name.c_str());
+    status = wal_manager_.DeleteFile(name, number);
    if (!status.ok()) {
      ROCKS_LOG_ERROR(immutable_db_options_.info_log,
                      "DeleteFile %s failed -- %s.\n", name.c_str(),
@@ -2321,7 +2332,15 @@ Status DB::DestroyColumnFamilyHandle(ColumnFamilyHandle* column_family) {
  return Status::OK();
 }

-DB::~DB() { }
+DB::~DB() {}
+
+Status DBImpl::Close() {
+  if (!closed_) {
+    closed_ = true;
+    return CloseImpl();
+  }
+  return Status::OK();
+}

 Status DB::ListColumnFamilies(const DBOptions& db_options,
                              const std::string& name,
@@ -468,6 +468,8 @@ class DBImpl : public DB {
  // belong to live files are posibly removed. Also, removes all the
  // files in sst_delete_files and log_delete_files.
  // It is not necessary to hold the mutex when invoking this method.
+  // If FindObsoleteFiles() was run, we need to also run
+  // PurgeObsoleteFiles(), even if disable_delete_obsolete_files_ is true
  void PurgeObsoleteFiles(const JobContext& background_contet,
                          bool schedule_only = false);

@@ -611,10 +613,14 @@ class DBImpl : public DB {
                     std::vector<ColumnFamilyHandle*>* handles, DB** dbptr,
                     const bool seq_per_batch);

+  virtual Status Close() override;
+
 protected:
  Env* const env_;
  const std::string dbname_;
  unique_ptr<VersionSet> versions_;
+  // Flag to check whether we allocated and own the info log file
+  bool own_info_log_;
  const DBOptions initial_db_options_;
  const ImmutableDBOptions immutable_db_options_;
  MutableDBOptions mutable_db_options_;
@@ -807,8 +813,12 @@ class DBImpl : public DB {
  Status FlushMemTable(ColumnFamilyData* cfd, const FlushOptions& options,
                       bool writes_stopped = false);

-  // Wait for memtable flushed
-  Status WaitForFlushMemTable(ColumnFamilyData* cfd);
+  // Wait for memtable flushed.
+  // If flush_memtable_id is non-null, wait until the memtable with the ID
+  // gets flush. Otherwise, wait until the column family don't have any
+  // memtable pending flush.
+  Status WaitForFlushMemTable(ColumnFamilyData* cfd,
+                              const uint64_t* flush_memtable_id = nullptr);

  // REQUIRES: mutex locked
  Status SwitchWAL(WriteContext* write_context);
@@ -912,6 +922,9 @@ class DBImpl : public DB {

  uint64_t GetMaxTotalWalSize() const;

+  // Actual implementation of Close()
+  virtual Status CloseImpl();
+
  // table_cache_ provides its own synchronization
  std::shared_ptr<Cache> table_cache_;

@@ -940,6 +953,8 @@ class DBImpl : public DB {
  // (i.e. whenever a flush is done, even if it didn't make any progress)
  // * whenever there is an error in background purge, flush or compaction
  // * whenever num_running_ingest_file_ goes to 0.
+  // * whenever pending_purge_obsolete_files_ goes to 0.
+  // * whenever disable_delete_obsolete_files_ goes to 0.
  InstrumentedCondVar bg_cv_;
  // Writes are protected by locking both mutex_ and log_write_mutex_, and reads
  // must be under either mutex_ or log_write_mutex_. Since after ::Open,
@@ -1206,6 +1221,10 @@ class DBImpl : public DB {
  // without any synchronization
  int disable_delete_obsolete_files_;

+  // Number of times FindObsoleteFiles has found deletable files and the
+  // corresponding call to PurgeObsoleteFiles has not yet finished.
+  int pending_purge_obsolete_files_;
+
  // last time when DeleteObsoleteFiles with full scan was executed. Originaly
  // initialized with startup time.
  uint64_t delete_obsolete_files_last_run_;
@@ -1359,6 +1378,9 @@ class DBImpl : public DB {
  // is set to false.
  std::atomic<SequenceNumber> preserve_deletes_seqnum_;
  const bool preserve_deletes_;
+
+  // Flag to check whether Close() has been called on this DB
+  bool closed_;
 };

 extern Options SanitizeOptions(const std::string& db,
@@ -134,6 +134,7 @@ Status DBImpl::FlushMemTableToOutputFile(
  }

  if (s.ok()) {
+    TEST_SYNC_POINT("DBImpl::FlushMemTableToOutputFile:BeforeInstallSV");
    InstallSuperVersionAndScheduleWork(cfd, &job_context->superversion_context,
                                       mutable_cf_options);
    if (made_progress) {
@@ -441,7 +442,8 @@ Status DBImpl::CompactFiles(
  }  // release the mutex

  // delete unnecessary files if any, this is done outside the mutex
-  if (job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
+  if (job_context.HaveSomethingToClean() ||
+      job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
    // Have to flush the info logs before bg_compaction_scheduled_--
    // because if bg_flush_scheduled_ becomes 0 and the lock is
    // released, the deconstructor of DB can kick in and destroy all the
@@ -809,7 +811,13 @@ int DBImpl::Level0StopWriteTrigger(ColumnFamilyHandle* column_family) {
 Status DBImpl::Flush(const FlushOptions& flush_options,
                     ColumnFamilyHandle* column_family) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
-  return FlushMemTable(cfh->cfd(), flush_options);
+  ROCKS_LOG_INFO(immutable_db_options_.info_log, "[%s] Manual flush start.",
+                 cfh->GetName().c_str());
+  Status s = FlushMemTable(cfh->cfd(), flush_options);
+  ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                 "[%s] Manual flush finished, status: %s\n",
+                 cfh->GetName().c_str(), s.ToString().c_str());
+  return s;
 }

 Status DBImpl::RunManualCompaction(ColumnFamilyData* cfd, int input_level,
@@ -944,6 +952,7 @@ Status DBImpl::FlushMemTable(ColumnFamilyData* cfd,
                             const FlushOptions& flush_options,
                             bool writes_stopped) {
  Status s;
+  uint64_t flush_memtable_id = 0;
  {
    WriteContext context;
    InstrumentedMutexLock guard_lock(&mutex_);
@@ -961,6 +970,7 @@ Status DBImpl::FlushMemTable(ColumnFamilyData* cfd,

    // SwitchMemtable() will release and reacquire mutex during execution
    s = SwitchMemtable(cfd, &context);
+    flush_memtable_id = cfd->imm()->GetLatestMemTableID();

    if (!writes_stopped) {
      write_thread_.ExitUnbatched(&w);
@@ -975,16 +985,19 @@ Status DBImpl::FlushMemTable(ColumnFamilyData* cfd,

  if (s.ok() && flush_options.wait) {
    // Wait until the compaction completes
-    s = WaitForFlushMemTable(cfd);
+    s = WaitForFlushMemTable(cfd, &flush_memtable_id);
  }
  return s;
 }

-Status DBImpl::WaitForFlushMemTable(ColumnFamilyData* cfd) {
+Status DBImpl::WaitForFlushMemTable(ColumnFamilyData* cfd,
+                                    const uint64_t* flush_memtable_id) {
  Status s;
  // Wait until the compaction completes
  InstrumentedMutexLock l(&mutex_);
-  while (cfd->imm()->NumNotFlushed() > 0 && bg_error_.ok()) {
+  while (cfd->imm()->NumNotFlushed() > 0 && bg_error_.ok() &&
+         (flush_memtable_id == nullptr ||
+          cfd->imm()->GetEarliestMemTableID() <= *flush_memtable_id)) {
    if (shutting_down_.load(std::memory_order_acquire)) {
      return Status::ShutdownInProgress();
    }
@@ -1059,6 +1072,7 @@ void DBImpl::MaybeScheduleFlushOrCompaction() {
  if (HasExclusiveManualCompaction()) {
    // only manual compactions are allowed to run. don't schedule automatic
    // compactions
+    TEST_SYNC_POINT("DBImpl::MaybeScheduleFlushOrCompaction:Conflict");
    return;
  }

@@ -1303,8 +1317,10 @@ void DBImpl::BackgroundCallFlush() {
    // created. Thus, we force full scan in FindObsoleteFiles()
    FindObsoleteFiles(&job_context, !s.ok() && !s.IsShutdownInProgress());
    // delete unnecessary files if any, this is done outside the mutex
-    if (job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
+    if (job_context.HaveSomethingToClean() ||
+        job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
      mutex_.Unlock();
+      TEST_SYNC_POINT("DBImpl::BackgroundCallFlush:FilesFound");
      // Have to flush the info logs before bg_flush_scheduled_--
      // because if bg_flush_scheduled_ becomes 0 and the lock is
      // released, the deconstructor of DB can kick in and destroy all the
@@ -1384,7 +1400,8 @@ void DBImpl::BackgroundCallCompaction(PrepickedCompaction* prepicked_compaction,
    FindObsoleteFiles(&job_context, !s.ok() && !s.IsShutdownInProgress());

    // delete unnecessary files if any, this is done outside the mutex
-    if (job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
+    if (job_context.HaveSomethingToClean() ||
+        job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
      mutex_.Unlock();
      // Have to flush the info logs before bg_compaction_scheduled_--
      // because if bg_flush_scheduled_ becomes 0 and the lock is
@@ -1503,7 +1520,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
               : m->manual_end->DebugString().c_str()));
    }
  } else if (!is_prepicked && !compaction_queue_.empty()) {
-    if (HaveManualCompaction(compaction_queue_.front())) {
+    if (HasExclusiveManualCompaction()) {
      // Can't compact right now, but try again later
      TEST_SYNC_POINT("DBImpl::BackgroundCompaction()::Conflict");

@@ -285,6 +285,9 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
  job_context->logs_to_free = logs_to_free_;
  job_context->log_recycle_files.assign(log_recycle_files_.begin(),
                                        log_recycle_files_.end());
+  if (job_context->HaveSomethingToDelete()) {
+    ++pending_purge_obsolete_files_;
+  }
  logs_to_free_.clear();
 }

@@ -342,15 +345,12 @@ void DBImpl::DeleteObsoleteFileImpl(int job_id, const std::string& fname,
 // files in sst_delete_files and log_delete_files.
 // It is not necessary to hold the mutex when invoking this method.
 void DBImpl::PurgeObsoleteFiles(const JobContext& state, bool schedule_only) {
+  TEST_SYNC_POINT("DBImpl::PurgeObsoleteFiles:Begin");
  // we'd better have sth to delete
  assert(state.HaveSomethingToDelete());

-  // this checks if FindObsoleteFiles() was run before. If not, don't do
-  // PurgeObsoleteFiles(). If FindObsoleteFiles() was run, we need to also
-  // run PurgeObsoleteFiles(), even if disable_delete_obsolete_files_ is true
-  if (state.manifest_file_number == 0) {
-    return;
-  }
+  // FindObsoleteFiles() should've populated this so nonzero
+  assert(state.manifest_file_number != 0);

  // Now, convert live list to an unordered map, WITHOUT mutex held;
  // set is slow.
@@ -533,6 +533,13 @@ void DBImpl::PurgeObsoleteFiles(const JobContext& state, bool schedule_only) {
  wal_manager_.PurgeObsoleteWALFiles();
 #endif  // ROCKSDB_LITE
  LogFlush(immutable_db_options_.info_log);
+  InstrumentedMutexLock l(&mutex_);
+  --pending_purge_obsolete_files_;
+  assert(pending_purge_obsolete_files_ >= 0);
+  if (pending_purge_obsolete_files_ == 0) {
+    bg_cv_.SignalAll();
+  }
+  TEST_SYNC_POINT("DBImpl::PurgeObsoleteFiles:End");
 }

 void DBImpl::DeleteObsoleteFiles() {
@@ -972,6 +972,17 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
  // assume there is at least one parseable key for this user key
  ParsedInternalKey ikey;
  FindParseableKey(&ikey, kForward);
+  assert(iter_->Valid());
+  assert(user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()));
+
+  // In case read_callback presents, the value we seek to may not be visible.
+  // Seek for the next value that's visible.
+  while (!IsVisible(ikey.sequence)) {
+    iter_->Next();
+    FindParseableKey(&ikey, kForward);
+    assert(iter_->Valid());
+    assert(user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()));
+  }

  if (ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
      range_del_agg_.ShouldDelete(
@@ -9,6 +9,7 @@

 #include <functional>

+#include "db/db_iter.h"
 #include "db/db_test_util.h"
 #include "port/port.h"
 #include "port/stack_trace.h"
@@ -2157,6 +2158,141 @@ TEST_F(DBIteratorTest, SkipStatistics) {
  ASSERT_EQ(skip_count, TestGetTickerCount(options, NUMBER_ITER_SKIP));
 }

+TEST_F(DBIteratorTest, ReadCallback) {
+  class TestReadCallback : public ReadCallback {
+   public:
+    explicit TestReadCallback(SequenceNumber last_visible_seq)
+        : last_visible_seq_(last_visible_seq) {}
+
+    bool IsCommitted(SequenceNumber seq) override {
+      return seq <= last_visible_seq_;
+    }
+
+   private:
+    SequenceNumber last_visible_seq_;
+  };
+
+  ASSERT_OK(Put("foo", "v1"));
+  ASSERT_OK(Put("foo", "v2"));
+  ASSERT_OK(Put("foo", "v3"));
+  ASSERT_OK(Put("a", "va"));
+  ASSERT_OK(Put("z", "vz"));
+  SequenceNumber seq1 = db_->GetLatestSequenceNumber();
+  TestReadCallback callback1(seq1);
+  ASSERT_OK(Put("foo", "v4"));
+  ASSERT_OK(Put("foo", "v5"));
+  ASSERT_OK(Put("bar", "v7"));
+
+  SequenceNumber seq2 = db_->GetLatestSequenceNumber();
+  auto* cfd =
+      reinterpret_cast<ColumnFamilyHandleImpl*>(db_->DefaultColumnFamily())
+          ->cfd();
+  // The iterator are suppose to see data before seq1.
+  Iterator* iter =
+      dbfull()->NewIteratorImpl(ReadOptions(), cfd, seq2, &callback1);
+
+  // Seek
+  // The latest value of "foo" before seq1 is "v3"
+  iter->Seek("foo");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("foo", iter->key());
+  ASSERT_EQ("v3", iter->value());
+  // "bar" is not visible to the iterator. It will move on to the next key
+  // "foo".
+  iter->Seek("bar");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("foo", iter->key());
+  ASSERT_EQ("v3", iter->value());
+
+  // Next
+  // Seek to "a"
+  iter->Seek("a");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("va", iter->value());
+  // "bar" is not visible to the iterator. It will move on to the next key
+  // "foo".
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("foo", iter->key());
+  ASSERT_EQ("v3", iter->value());
+
+  // Prev
+  // Seek to "z"
+  iter->Seek("z");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("vz", iter->value());
+  // The previous key is "foo", which is visible to the iterator.
+  iter->Prev();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("foo", iter->key());
+  ASSERT_EQ("v3", iter->value());
+  // "bar" is not visible to the iterator. It will move on to the next key "a".
+  iter->Prev();  // skipping "bar"
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("a", iter->key());
+  ASSERT_EQ("va", iter->value());
+
+  // SeekForPrev
+  // The previous key is "foo", which is visible to the iterator.
+  iter->SeekForPrev("y");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("foo", iter->key());
+  ASSERT_EQ("v3", iter->value());
+  // "bar" is not visible to the iterator. It will move on to the next key "a".
+  iter->SeekForPrev("bar");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("a", iter->key());
+  ASSERT_EQ("va", iter->value());
+
+  delete iter;
+
+  // Prev beyond max_sequential_skip_in_iterations
+  uint64_t num_versions =
+      CurrentOptions().max_sequential_skip_in_iterations + 10;
+  for (uint64_t i = 0; i < num_versions; i++) {
+    ASSERT_OK(Put("bar", ToString(i)));
+  }
+  SequenceNumber seq3 = db_->GetLatestSequenceNumber();
+  TestReadCallback callback2(seq3);
+  ASSERT_OK(Put("bar", "v8"));
+  SequenceNumber seq4 = db_->GetLatestSequenceNumber();
+
+  // The iterator is suppose to see data before seq3.
+  iter = dbfull()->NewIteratorImpl(ReadOptions(), cfd, seq4, &callback2);
+  // Seek to "z", which is visible.
+  iter->Seek("z");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("vz", iter->value());
+  // Previous key is "foo" and the last value "v5" is visible.
+  iter->Prev();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("foo", iter->key());
+  ASSERT_EQ("v5", iter->value());
+  // Since the number of values of "bar" is more than
+  // max_sequential_skip_in_iterations, Prev() will ultimately fallback to
+  // seek in forward direction. Here we test the fallback seek is correct.
+  // The last visible value should be (num_versions - 1), as "v8" is not
+  // visible.
+  iter->Prev();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("bar", iter->key());
+  ASSERT_EQ(ToString(num_versions - 1), iter->value());
+
+  delete iter;
+}
+
 }  // namespace rocksdb

 int main(int argc, char** argv) {
@@ -518,8 +518,13 @@ static void assert_candidate_files_empty(DBImpl* dbfull, const bool empty) {
  JobContext job_context(0);
  dbfull->FindObsoleteFiles(&job_context, false);
  ASSERT_EQ(empty, job_context.full_scan_candidate_files.empty());
-  job_context.Clean();
  dbfull->TEST_UnlockMutex();
+  if (job_context.HaveSomethingToDelete()) {
+    // fulfill the contract of FindObsoleteFiles by calling PurgeObsoleteFiles
+    // afterwards; otherwise the test may hang on shutdown
+    dbfull->PurgeObsoleteFiles(job_context);
+  }
+  job_context.Clean();
 }

 TEST_F(DBOptionsTest, DeleteObsoleteFilesPeriodChange) {
@@ -503,7 +503,9 @@ TEST_F(DBTest, DISABLED_VeryLargeValue) {
  ASSERT_OK(Put(key2, raw));
  dbfull()->TEST_WaitForFlushMemTable();

+#ifndef ROCKSDB_LITE
  ASSERT_EQ(1, NumTableFilesAtLevel(0));
+#endif  // !ROCKSDB_LITE

  std::string value;
  Status s = db_->Get(ReadOptions(), key1, &value);
@@ -2191,6 +2193,8 @@ class ModelDB : public DB {
    batch.Put(cf, k, v);
    return Write(o, &batch);
  }
+  using DB::Close;
+  virtual Status Close() override { return Status::OK(); }
  using DB::Delete;
  virtual Status Delete(const WriteOptions& o, ColumnFamilyHandle* cf,
                        const Slice& key) override {
@@ -5054,55 +5058,84 @@ TEST_F(DBTest, PromoteL0Failure) {
  status = experimental::PromoteL0(db_, db_->DefaultColumnFamily());
  ASSERT_TRUE(status.IsInvalidArgument());
 }
-#endif  // ROCKSDB_LITE

 // Github issue #596
-TEST_F(DBTest, HugeNumberOfLevels) {
+TEST_F(DBTest, CompactRangeWithEmptyBottomLevel) {
+  const int kNumLevels = 2;
+  const int kNumL0Files = 2;
  Options options = CurrentOptions();
-  options.write_buffer_size = 2 * 1024 * 1024;         // 2MB
-  options.max_bytes_for_level_base = 2 * 1024 * 1024;  // 2MB
-  options.num_levels = 12;
-  options.max_background_compactions = 10;
-  options.max_bytes_for_level_multiplier = 2;
-  options.level_compaction_dynamic_level_bytes = true;
+  options.disable_auto_compactions = true;
+  options.num_levels = kNumLevels;
  DestroyAndReopen(options);

  Random rnd(301);
-  for (int i = 0; i < 300000; ++i) {
-    ASSERT_OK(Put(Key(i), RandomString(&rnd, 1024)));
+  for (int i = 0; i < kNumL0Files; ++i) {
+    ASSERT_OK(Put(Key(0), RandomString(&rnd, 1024)));
+    Flush();
  }
+  ASSERT_EQ(NumTableFilesAtLevel(0), kNumL0Files);
+  ASSERT_EQ(NumTableFilesAtLevel(1), 0);

  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
+  ASSERT_EQ(NumTableFilesAtLevel(1), kNumL0Files);
 }
+#endif  // ROCKSDB_LITE

 TEST_F(DBTest, AutomaticConflictsWithManualCompaction) {
+  const int kNumL0Files = 50;
  Options options = CurrentOptions();
-  options.write_buffer_size = 2 * 1024 * 1024;         // 2MB
-  options.max_bytes_for_level_base = 2 * 1024 * 1024;  // 2MB
-  options.num_levels = 12;
+  options.level0_file_num_compaction_trigger = 4;
+  // never slowdown / stop
+  options.level0_slowdown_writes_trigger = 999999;
+  options.level0_stop_writes_trigger = 999999;
  options.max_background_compactions = 10;
-  options.max_bytes_for_level_multiplier = 2;
-  options.level_compaction_dynamic_level_bytes = true;
  DestroyAndReopen(options);

-  Random rnd(301);
-  for (int i = 0; i < 300000; ++i) {
-    ASSERT_OK(Put(Key(i), RandomString(&rnd, 1024)));
-  }
-
+  // schedule automatic compactions after the manual one starts, but before it
+  // finishes to ensure conflict.
+  rocksdb::SyncPoint::GetInstance()->LoadDependency(
+      {{"DBImpl::BackgroundCompaction:Start",
+        "DBTest::AutomaticConflictsWithManualCompaction:PrePuts"},
+       {"DBTest::AutomaticConflictsWithManualCompaction:PostPuts",
+        "DBImpl::BackgroundCompaction:NonTrivial:AfterRun"}});
  std::atomic<int> callback_count(0);
  rocksdb::SyncPoint::GetInstance()->SetCallBack(
-      "DBImpl::BackgroundCompaction()::Conflict",
+      "DBImpl::MaybeScheduleFlushOrCompaction:Conflict",
      [&](void* arg) { callback_count.fetch_add(1); });
  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
-  CompactRangeOptions croptions;
-  croptions.exclusive_manual_compaction = false;
-  ASSERT_OK(db_->CompactRange(croptions, nullptr, nullptr));
+
+  Random rnd(301);
+  for (int i = 0; i < 2; ++i) {
+    // put two keys to ensure no trivial move
+    for (int j = 0; j < 2; ++j) {
+      ASSERT_OK(Put(Key(j), RandomString(&rnd, 1024)));
+    }
+    ASSERT_OK(Flush());
+  }
+  std::thread manual_compaction_thread([this]() {
+      CompactRangeOptions croptions;
+      croptions.exclusive_manual_compaction = true;
+      ASSERT_OK(db_->CompactRange(croptions, nullptr, nullptr));
+    });
+
+  TEST_SYNC_POINT("DBTest::AutomaticConflictsWithManualCompaction:PrePuts");
+  for (int i = 0; i < kNumL0Files; ++i) {
+    // put two keys to ensure no trivial move
+    for (int j = 0; j < 2; ++j) {
+      ASSERT_OK(Put(Key(j), RandomString(&rnd, 1024)));
+    }
+    ASSERT_OK(Flush());
+  }
+  TEST_SYNC_POINT("DBTest::AutomaticConflictsWithManualCompaction:PostPuts");
+
  ASSERT_GE(callback_count.load(), 1);
-  rocksdb::SyncPoint::GetInstance()->DisableProcessing();
-  for (int i = 0; i < 300000; ++i) {
+  for (int i = 0; i < 2; ++i) {
    ASSERT_NE("NOT_FOUND", Get(Key(i)));
  }
+  rocksdb::SyncPoint::GetInstance()->DisableProcessing();
+  manual_compaction_thread.join();
+  dbfull()->TEST_WaitForCompact();
 }

 // Github issue #595
@@ -5322,12 +5355,15 @@ class WriteStallListener : public EventListener {
 public:
  WriteStallListener() : condition_(WriteStallCondition::kNormal) {}
  void OnStallConditionsChanged(const WriteStallInfo& info) override {
+    MutexLock l(&mutex_);
    condition_ = info.condition.cur;
  }
  bool CheckCondition(WriteStallCondition expected) {
+    MutexLock l(&mutex_);
    return expected == condition_;
  }
 private:
+  port::Mutex mutex_;
  WriteStallCondition condition_;
 };

@@ -2430,6 +2430,60 @@ TEST_F(DBTest2, ReadCallbackTest) {
  }
 }

+#ifndef ROCKSDB_LITE
+
+TEST_F(DBTest2, LiveFilesOmitObsoleteFiles) {
+  // Regression test for race condition where an obsolete file is returned to
+  // user as a "live file" but then deleted, all while file deletions are
+  // disabled.
+  //
+  // It happened like this:
+  //
+  // 1. [flush thread] Log file "x.log" found by FindObsoleteFiles
+  // 2. [user thread] DisableFileDeletions, GetSortedWalFiles are called and the
+  //    latter returned "x.log"
+  // 3. [flush thread] PurgeObsoleteFiles deleted "x.log"
+  // 4. [user thread] Reading "x.log" failed
+  //
+  // Unfortunately the only regression test I can come up with involves sleep.
+  // We cannot set SyncPoints to repro since, once the fix is applied, the
+  // SyncPoints would cause a deadlock as the repro's sequence of events is now
+  // prohibited.
+  //
+  // Instead, if we sleep for a second between Find and Purge, and ensure the
+  // read attempt happens after purge, then the sequence of events will almost
+  // certainly happen on the old code.
+  rocksdb::SyncPoint::GetInstance()->LoadDependency({
+      {"DBImpl::BackgroundCallFlush:FilesFound",
+       "DBTest2::LiveFilesOmitObsoleteFiles:FlushTriggered"},
+      {"DBImpl::PurgeObsoleteFiles:End",
+       "DBTest2::LiveFilesOmitObsoleteFiles:LiveFilesCaptured"},
+  });
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::PurgeObsoleteFiles:Begin",
+      [&](void* arg) { env_->SleepForMicroseconds(1000000); });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  Put("key", "val");
+  FlushOptions flush_opts;
+  flush_opts.wait = false;
+  db_->Flush(flush_opts);
+  TEST_SYNC_POINT("DBTest2::LiveFilesOmitObsoleteFiles:FlushTriggered");
+
+  db_->DisableFileDeletions();
+  VectorLogPtr log_files;
+  db_->GetSortedWalFiles(log_files);
+  TEST_SYNC_POINT("DBTest2::LiveFilesOmitObsoleteFiles:LiveFilesCaptured");
+  for (const auto& log_file : log_files) {
+    ASSERT_OK(env_->FileExists(LogFileName(dbname_, log_file->LogNumber())));
+  }
+
+  db_->EnableFileDeletions();
+  rocksdb::SyncPoint::GetInstance()->DisableProcessing();
+}
+
+#endif  // ROCKSDB_LITE
+
 }  // namespace rocksdb

 int main(int argc, char** argv) {
@@ -867,7 +867,6 @@ size_t DBTestBase::CountLiveFiles() {
  db_->GetLiveFilesMetaData(&metadata);
  return metadata.size();
 }
-#endif  // ROCKSDB_LITE

 int DBTestBase::NumTableFilesAtLevel(int level, int cf) {
  std::string property;
@@ -928,6 +927,7 @@ std::string DBTestBase::FilesPerLevel(int cf) {
  result.resize(last_non_zero_offset);
  return result;
 }
+#endif  // !ROCKSDB_LITE

 size_t DBTestBase::CountFiles() {
  std::vector<std::string> files;
@@ -997,6 +997,7 @@ void DBTestBase::MoveFilesToLevel(int level, int cf) {
  }
 }

+#ifndef ROCKSDB_LITE
 void DBTestBase::DumpFileCounts(const char* label) {
  fprintf(stderr, "---\n%s:\n", label);
  fprintf(stderr, "maxoverlap: %" PRIu64 "\n",
@@ -1008,6 +1009,7 @@ void DBTestBase::DumpFileCounts(const char* label) {
    }
  }
 }
+#endif  // !ROCKSDB_LITE

 std::string DBTestBase::DumpSSTableList() {
  std::string property;
@@ -863,13 +863,13 @@ class DBTestBase : public testing::Test {
  size_t TotalLiveFiles(int cf = 0);

  size_t CountLiveFiles();
-#endif  // ROCKSDB_LITE

  int NumTableFilesAtLevel(int level, int cf = 0);

  double CompressionRatioAtLevel(int level, int cf = 0);

  int TotalTableFiles(int cf = 0, int levels = -1);
+#endif  // ROCKSDB_LITE

  // Return spread of files per level
  std::string FilesPerLevel(int cf = 0);
@@ -897,7 +897,9 @@ class DBTestBase : public testing::Test {

  void MoveFilesToLevel(int level, int cf = 0);

+#ifndef ROCKSDB_LITE
  void DumpFileCounts(const char* label);
+#endif  // ROCKSDB_LITE

  std::string DumpSSTableList();

@@ -1940,6 +1940,55 @@ TEST_F(ExternalSSTFileTest, IngestBehind) {
  size_t kcnt = 0;
  VerifyDBFromMap(true_data, &kcnt, false);
 }
+
+TEST_F(ExternalSSTFileTest, SkipBloomFilter) {
+  Options options = CurrentOptions();
+
+  BlockBasedTableOptions table_options;
+  table_options.filter_policy.reset(NewBloomFilterPolicy(10));
+  table_options.cache_index_and_filter_blocks = true;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+
+  // Create external SST file and include bloom filters
+  options.statistics = rocksdb::CreateDBStatistics();
+  DestroyAndReopen(options);
+  {
+    std::string file_path = sst_files_dir_ + "sst_with_bloom.sst";
+    SstFileWriter sst_file_writer(EnvOptions(), options);
+    ASSERT_OK(sst_file_writer.Open(file_path));
+    ASSERT_OK(sst_file_writer.Put("Key1", "Value1"));
+    ASSERT_OK(sst_file_writer.Finish());
+
+    ASSERT_OK(
+        db_->IngestExternalFile({file_path}, IngestExternalFileOptions()));
+
+    ASSERT_EQ(Get("Key1"), "Value1");
+    ASSERT_GE(
+        options.statistics->getTickerCount(Tickers::BLOCK_CACHE_FILTER_ADD), 1);
+  }
+
+  // Create external SST file but skip bloom filters
+  options.statistics = rocksdb::CreateDBStatistics();
+  DestroyAndReopen(options);
+  {
+    std::string file_path = sst_files_dir_ + "sst_with_no_bloom.sst";
+    SstFileWriter sst_file_writer(EnvOptions(), options, nullptr, true,
+                                  Env::IOPriority::IO_TOTAL,
+                                  true /* skip_filters */);
+    ASSERT_OK(sst_file_writer.Open(file_path));
+    ASSERT_OK(sst_file_writer.Put("Key1", "Value1"));
+    ASSERT_OK(sst_file_writer.Finish());
+
+    ASSERT_OK(
+        db_->IngestExternalFile({file_path}, IngestExternalFileOptions()));
+
+    ASSERT_EQ(Get("Key1"), "Value1");
+    ASSERT_EQ(
+        options.statistics->getTickerCount(Tickers::BLOCK_CACHE_FILTER_ADD), 0);
+  }
+}
+
 }  // namespace rocksdb

 int main(int argc, char** argv) {
@@ -81,8 +81,11 @@ struct SuperVersionContext {
 struct JobContext {
  inline bool HaveSomethingToDelete() const {
    return full_scan_candidate_files.size() || sst_delete_files.size() ||
-           log_delete_files.size() || manifest_delete_files.size() ||
-           memtables_to_free.size() > 0 || logs_to_free.size() > 0 ||
+           log_delete_files.size() || manifest_delete_files.size();
+  }
+
+  inline bool HaveSomethingToClean() const {
+    return memtables_to_free.size() > 0 || logs_to_free.size() > 0 ||
           superversion_context.HaveSomethingToDelete();
  }

@@ -16,7 +16,12 @@
 namespace rocksdb {

 #ifdef ROCKSDB_JEMALLOC
+#ifdef __FreeBSD__
+#include <malloc_np.h>
+#define je_malloc_stats_print malloc_stats_print
+#else
 #include "jemalloc/jemalloc.h"
+#endif

 typedef struct {
  char* cur;
@@ -236,6 +236,14 @@ int MemTable::KeyComparator::operator()(const char* prefix_len_key,
  return comparator.Compare(a, key);
 }

+void MemTableRep::InsertConcurrently(KeyHandle handle) {
+#ifndef ROCKSDB_LITE
+    throw std::runtime_error("concurrent insert not supported");
+#else
+    abort();
+#endif
+  }
+
 Slice MemTableRep::UserKey(const char* key) const {
  Slice slice = GetLengthPrefixedSlice(key);
  return Slice(slice.data(), slice.size() - 8);
@@ -368,6 +368,11 @@ class MemTable {
    return oldest_key_time_.load(std::memory_order_relaxed);
  }

+  // REQUIRES: db_mutex held.
+  void SetID(uint64_t id) { id_ = id; }
+
+  uint64_t GetID() const { return id_; }
+
 private:
  enum FlushStateEnum { FLUSH_NOT_REQUESTED, FLUSH_REQUESTED, FLUSH_SCHEDULED };

@@ -437,6 +442,9 @@ class MemTable {
  // Timestamp of oldest key
  std::atomic<uint64_t> oldest_key_time_;

+  // Memtable id to track flush.
+  uint64_t id_ = 0;
+
  // Returns a heuristic flush decision
  bool ShouldFlushNow() const;

@@ -5,11 +5,12 @@
 //
 #pragma once

-#include <string>
-#include <list>
-#include <vector>
-#include <set>
 #include <deque>
+#include <limits>
+#include <list>
+#include <set>
+#include <string>
+#include <vector>

 #include "db/dbformat.h"
 #include "db/memtable.h"
@@ -244,6 +245,22 @@ class MemTableList {

  uint64_t GetMinLogContainingPrepSection();

+  uint64_t GetEarliestMemTableID() const {
+    auto& memlist = current_->memlist_;
+    if (memlist.empty()) {
+      return std::numeric_limits<uint64_t>::max();
+    }
+    return memlist.back()->GetID();
+  }
+
+  uint64_t GetLatestMemTableID() const {
+    auto& memlist = current_->memlist_;
+    if (memlist.empty()) {
+      return 0;
+    }
+    return memlist.front()->GetID();
+  }
+
 private:
  // DB mutex held
  void InstallNewVersion();
@@ -39,6 +39,15 @@ namespace rocksdb {

 #ifndef ROCKSDB_LITE

+Status WalManager::DeleteFile(const std::string& fname, uint64_t number) {
+  auto s = env_->DeleteFile(db_options_.wal_dir + "/" + fname);
+  if (s.ok()) {
+    MutexLock l(&read_first_record_cache_mutex_);
+    read_first_record_cache_.erase(number);
+  }
+  return s;
+}
+
 Status WalManager::GetSortedWalFiles(VectorLogPtr& files) {
  // First get sorted files in db dir, then get sorted files from archived
  // dir, to avoid a race condition where a log file is moved to archived
@@ -49,6 +49,8 @@ class WalManager {

  void ArchiveWALFile(const std::string& fname, uint64_t number);

+  Status DeleteFile(const std::string& fname, uint64_t number);
+
  Status TEST_ReadFirstRecord(const WalFileType type, const uint64_t number,
                              SequenceNumber* sequence) {
    return ReadFirstRecord(type, number, sequence);
@@ -15,10 +15,12 @@
 //    kTypeValue varstring varstring
 //    kTypeDeletion varstring
 //    kTypeSingleDeletion varstring
+//    kTypeRangeDeletion varstring varstring
 //    kTypeMerge varstring varstring
 //    kTypeColumnFamilyValue varint32 varstring varstring
-//    kTypeColumnFamilyDeletion varint32 varstring varstring
-//    kTypeColumnFamilySingleDeletion varint32 varstring varstring
+//    kTypeColumnFamilyDeletion varint32 varstring
+//    kTypeColumnFamilySingleDeletion varint32 varstring
+//    kTypeColumnFamilyRangeDeletion varint32 varstring varstring
 //    kTypeColumnFamilyMerge varint32 varstring varstring
 //    kTypeBeginPrepareXID varstring
 //    kTypeEndPrepareXID
@@ -144,6 +146,12 @@ WriteBatch::WriteBatch(const std::string& rep)
      max_bytes_(0),
      rep_(rep) {}

+WriteBatch::WriteBatch(std::string&& rep)
+    : save_points_(nullptr),
+      content_flags_(ContentFlags::DEFERRED),
+      max_bytes_(0),
+      rep_(std::move(rep)) {}
+
 WriteBatch::WriteBatch(const WriteBatch& src)
    : save_points_(src.save_points_),
      wal_term_point_(src.wal_term_point_),
@@ -105,13 +105,13 @@ GEM
      rb-fsevent (>= 0.9.3)
      rb-inotify (>= 0.9.7)
    mercenary (0.3.6)
-    mini_portile2 (2.1.0)
+    mini_portile2 (2.3.0)
    minima (2.0.0)
    minitest (5.9.1)
    multipart-post (2.0.0)
    net-dns (0.8.0)
-    nokogiri (1.6.8.1)
-      mini_portile2 (~> 2.1.0)
+    nokogiri (1.8.1)
+      mini_portile2 (~> 2.3.0)
    octokit (4.4.1)
      sawyer (~> 0.7.0, >= 0.5.3)
    pathutil (0.14.0)
@@ -73,9 +73,18 @@ RandomAccessFile::~RandomAccessFile() {
 WritableFile::~WritableFile() {
 }

-Logger::~Logger() {
+Logger::~Logger() { Close(); }
+
+Status Logger::Close() {
+  if (!closed_) {
+    closed_ = true;
+    return CloseImpl();
+  }
+  return Status::OK();
 }

+Status Logger::CloseImpl() { return Status::OK(); }
+
 FileLock::~FileLock() {
 }

@@ -277,6 +277,16 @@ class HdfsLogger : public Logger {
  HdfsWritableFile* file_;
  uint64_t (*gettid_)();  // Return the thread id for the current thread

+  virtual Status CloseImpl() {
+    ROCKS_LOG_DEBUG(mylog, "[hdfs] HdfsLogger closed %s\n",
+                    file_->getName().c_str());
+    Status s = file_->Close();
+    if (mylog != nullptr && mylog == this) {
+      mylog = nullptr;
+    }
+    return s;
+  }
+
 public:
  HdfsLogger(HdfsWritableFile* f, uint64_t (*gettid)())
      : file_(f), gettid_(gettid) {
@@ -285,12 +295,6 @@ class HdfsLogger : public Logger {
  }

  virtual ~HdfsLogger() {
-    ROCKS_LOG_DEBUG(mylog, "[hdfs] HdfsLogger closed %s\n",
-                    file_->getName().c_str());
-    delete file_;
-    if (mylog != nullptr && mylog == this) {
-      mylog = nullptr;
-    }
  }

  virtual void Logv(const char* format, va_list ap) {
@@ -24,6 +24,7 @@
 #endif

 #include <atomic>
+#include "env/io_posix.h"
 #include "monitoring/iostats_context_imp.h"
 #include "rocksdb/env.h"
 #include "util/sync_point.h"
@@ -32,6 +33,15 @@ namespace rocksdb {

 class PosixLogger : public Logger {
 private:
+  virtual Status CloseImpl() override {
+    int ret;
+
+    ret = fclose(file_);
+    if (ret) {
+      return IOError("Unable to close log file", "", ret);
+    }
+    return Status::OK();
+  }
  FILE* file_;
  uint64_t (*gettid_)();  // Return the thread id for the current thread
  std::atomic_size_t log_size_;
@@ -51,9 +61,7 @@ class PosixLogger : public Logger {
        last_flush_micros_(0),
        env_(env),
        flush_pending_(false) {}
-  virtual ~PosixLogger() {
-    fclose(file_);
-  }
+  virtual ~PosixLogger() { Close(); }
  virtual void Flush() override {
    TEST_SYNC_POINT("PosixLogger::Flush:Begin1");
    TEST_SYNC_POINT("PosixLogger::Flush:Begin2");
@@ -163,6 +163,12 @@ class DB {
                     const std::vector<ColumnFamilyDescriptor>& column_families,
                     std::vector<ColumnFamilyHandle*>* handles, DB** dbptr);

+  // Close the DB by releasing resources, closing files etc. This should be
+  // called before calling the desctructor so that the caller can get back a
+  // status in case there are any errors. Regardless of the return status, the
+  // DB must be freed
+  virtual Status Close() { return Status::OK(); }
+
  // ListColumnFamilies will open the DB specified by argument name
  // and return the list of all column families in that DB
  // through column_families argument. The ordering of
@@ -973,7 +979,7 @@ class DB {
  // the file can fit in, and ingest the file into this level (2). A file that
  // have a key range that overlap with the memtable key range will require us
  // to Flush the memtable first before ingesting the file.
-  // In the second mode we will always ingest in the bottom mode level (see
+  // In the second mode we will always ingest in the bottom most level (see
  // docs to IngestExternalFileOptions::ingest_behind).
  //
  // (1) External SST files can be created using SstFileWriter
@@ -819,9 +819,12 @@ class Logger {
  size_t kDoNotSupportGetLogFileSize = (std::numeric_limits<size_t>::max)();

  explicit Logger(const InfoLogLevel log_level = InfoLogLevel::INFO_LEVEL)
-      : log_level_(log_level) {}
+      : closed_(false), log_level_(log_level) {}
  virtual ~Logger();

+  // Close the log file. Must be called before destructor
+  virtual Status Close();
+
  // Write a header to the log file with the specified format
  // It is recommended that you log all header information at the start of the
  // application. But it is not enforced.
@@ -852,6 +855,8 @@ class Logger {
  // No copying allowed
  Logger(const Logger&);
  void operator=(const Logger&);
+  virtual Status CloseImpl();
+  bool closed_;
  InfoLogLevel log_level_;
 };

@@ -51,7 +51,7 @@ class Iterator : public Cleanable {
  // Position at the last key in the source that at or before target
  // The iterator is Valid() after this call iff the source contains
  // an entry that comes at or before target.
-  virtual void SeekForPrev(const Slice& target) {}
+  virtual void SeekForPrev(const Slice& target) = 0;

  // Moves to the next entry in the source.  After this call, Valid() is
  // true iff the iterator was not positioned at the last entry in the source.
@@ -96,13 +96,7 @@ class MemTableRep {

  // Like Insert(handle), but may be called concurrent with other calls
  // to InsertConcurrently for other handles
-  virtual void InsertConcurrently(KeyHandle handle) {
-#ifndef ROCKSDB_LITE
-    throw std::runtime_error("concurrent insert not supported");
-#else
-    abort();
-#endif
-  }
+  virtual void InsertConcurrently(KeyHandle handle);

  // Returns true iff an entry that compares equal to key is in the collection.
  virtual bool Contains(const char* key) const = 0;
@@ -72,16 +72,18 @@ class SstFileWriter {
  SstFileWriter(const EnvOptions& env_options, const Options& options,
                ColumnFamilyHandle* column_family = nullptr,
                bool invalidate_page_cache = true,
-                Env::IOPriority io_priority = Env::IOPriority::IO_TOTAL)
+                Env::IOPriority io_priority = Env::IOPriority::IO_TOTAL,
+                bool skip_filters = false)
      : SstFileWriter(env_options, options, options.comparator, column_family,
-                      invalidate_page_cache, io_priority) {}
+                      invalidate_page_cache, io_priority, skip_filters) {}

  // Deprecated API
  SstFileWriter(const EnvOptions& env_options, const Options& options,
                const Comparator* user_comparator,
                ColumnFamilyHandle* column_family = nullptr,
                bool invalidate_page_cache = true,
-                Env::IOPriority io_priority = Env::IOPriority::IO_TOTAL);
+                Env::IOPriority io_priority = Env::IOPriority::IO_TOTAL,
+                bool skip_filters = false);

  ~SstFileWriter();

@@ -217,6 +217,11 @@ struct BlockBasedTableOptions {
  // This option only affects newly written tables. When reading exising tables,
  // the information about version is read from the footer.
  uint32_t format_version = 2;
+
+  // Store index blocks on disk in compressed format. Changing this option to
+  // false  will avoid the overhead of decompression if index blocks are evicted
+  // and read back
+  bool enable_index_compression = true;
 };

 // Table Properties that are specific to block-based table properties.
@@ -25,7 +25,7 @@ enum EntryType {
  kEntryOther,
 };

-// <user key, seqeence number and entry type> tuple.
+// <user key, sequence number, and entry type> tuple.
 struct FullKey {
  Slice user_key;
  SequenceNumber sequence;
@@ -60,6 +60,10 @@ class DBWithTTL : public StackableDB {
                     DBWithTTL** dbptr, std::vector<int32_t> ttls,
                     bool read_only = false);

+ virtual void SetTtl(int32_t ttl) = 0;
+
+ virtual void SetTtl(ColumnFamilyHandle *h, int32_t ttl) = 0;
+
 protected:
  explicit DBWithTTL(DB* db) : StackableDB(db) {}
 };
@@ -4,6 +4,7 @@

 #pragma once
 #include <map>
+#include <memory>
 #include <string>
 #include "rocksdb/db.h"

@@ -18,13 +19,23 @@ namespace rocksdb {
 // This class contains APIs to stack rocksdb wrappers.Eg. Stack TTL over base d
 class StackableDB : public DB {
 public:
-  // StackableDB is the owner of db now!
+  // StackableDB take sole ownership of the underlying db.
  explicit StackableDB(DB* db) : db_(db) {}

+  // StackableDB take shared ownership of the underlying db.
+  explicit StackableDB(std::shared_ptr<DB>& db)
+      : db_(db.get()), shared_db_(db) {}
+
  ~StackableDB() {
-    delete db_;
+    if (shared_db_ == nullptr) {
+      delete db_;
+    } else {
+      assert(shared_db_.get() == db_);
+    }
  }

+  virtual Status Close() override { return db_->Close(); }
+
  virtual DB* GetBaseDB() {
    return db_;
  }
@@ -373,6 +384,7 @@ class StackableDB : public DB {

 protected:
  DB* db_;
+  std::shared_ptr<DB> shared_db_;
 };

 } //  namespace rocksdb
@@ -315,6 +315,7 @@ class WriteBatch : public WriteBatchBase {

  // Constructor with a serialized string object
  explicit WriteBatch(const std::string& rep);
+  explicit WriteBatch(std::string&& rep);

  WriteBatch(const WriteBatch& src);
  WriteBatch(WriteBatch&& src) noexcept;
@@ -12,6 +12,10 @@ public class Environment {
    return (OS.contains("win"));
  }

+  public static boolean isFreeBSD() {
+    return (OS.contains("freebsd"));
+  }
+
  public static boolean isMac() {
    return (OS.contains("mac"));
  }
@@ -54,6 +58,8 @@ public class Environment {
      }
    } else if (isMac()) {
      return String.format("%sjni-osx", name);
+    } else if (isFreeBSD()) {
+      return String.format("%sjni-freebsd%s", name, is64Bit() ? "64" : "32");
    } else if (isAix() && is64Bit()) {
      return String.format("%sjni-aix64", name);
    } else if (isSolaris()) {
@@ -71,7 +77,7 @@ public class Environment {
  }

  private static String appendLibOsSuffix(final String libraryFileName, final boolean shared) {
-    if (isUnix() || isAix() || isSolaris()) {
+    if (isUnix() || isAix() || isSolaris() || isFreeBSD()) {
      return libraryFileName + ".so";
    } else if (isMac()) {
      return libraryFileName + (shared ? ".dylib" : ".jnilib");
@@ -208,6 +208,18 @@ std::unordered_map<std::string, ChecksumType>
                                               {"kCRC32c", kCRC32c},
                                               {"kxxHash", kxxHash}};

+std::unordered_map<std::string, CompressionType>
+    OptionsHelper::compression_type_string_map = {
+        {"kNoCompression", kNoCompression},
+        {"kSnappyCompression", kSnappyCompression},
+        {"kZlibCompression", kZlibCompression},
+        {"kBZip2Compression", kBZip2Compression},
+        {"kLZ4Compression", kLZ4Compression},
+        {"kLZ4HCCompression", kLZ4HCCompression},
+        {"kXpressCompression", kXpressCompression},
+        {"kZSTD", kZSTD},
+        {"kZSTDNotFinalCompression", kZSTDNotFinalCompression},
+        {"kDisableCompressionOption", kDisableCompressionOption}};
 #ifndef ROCKSDB_LITE

 template <typename T>
@@ -1476,19 +1488,6 @@ std::unordered_map<std::string, OptionTypeInfo>
         {0, OptionType::kBoolean, OptionVerificationType::kDeprecated, false,
          0}}};

-std::unordered_map<std::string, CompressionType>
-    OptionsHelper::compression_type_string_map = {
-        {"kNoCompression", kNoCompression},
-        {"kSnappyCompression", kSnappyCompression},
-        {"kZlibCompression", kZlibCompression},
-        {"kBZip2Compression", kBZip2Compression},
-        {"kLZ4Compression", kLZ4Compression},
-        {"kLZ4HCCompression", kLZ4HCCompression},
-        {"kXpressCompression", kXpressCompression},
-        {"kZSTD", kZSTD},
-        {"kZSTDNotFinalCompression", kZSTDNotFinalCompression},
-        {"kDisableCompressionOption", kDisableCompressionOption}};
-
 std::unordered_map<std::string, BlockBasedTableOptions::IndexType>
    OptionsHelper::block_base_table_index_type_string_map = {
        {"kBinarySearch", BlockBasedTableOptions::IndexType::kBinarySearch},
@@ -1582,8 +1581,6 @@ int offset_of(T1 CompactionOptionsUniversal::*member) {
 std::unordered_map<std::string, OptionTypeInfo>
    OptionsHelper::cf_options_type_info = {
        /* not yet supported
-        CompactionOptionsFIFO compaction_options_fifo;
-        CompactionOptionsUniversal compaction_options_universal;
        CompressionOptions compression_opts;
        TablePropertiesCollectorFactories table_properties_collector_factories;
        typedef std::vector<std::shared_ptr<TablePropertiesCollectorFactory>>
@@ -144,6 +144,8 @@ struct OptionsHelper {
  static std::map<CompactionStopStyle, std::string>
      compaction_stop_style_to_string;
  static std::unordered_map<std::string, ChecksumType> checksum_type_string_map;
+  static std::unordered_map<std::string, CompressionType>
+      compression_type_string_map;
 #ifndef ROCKSDB_LITE
  static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info;
  static std::unordered_map<std::string, OptionTypeInfo>
@@ -155,8 +157,6 @@ struct OptionsHelper {
  static std::unordered_map<std::string, OptionTypeInfo> db_options_type_info;
  static std::unordered_map<std::string, OptionTypeInfo>
      lru_cache_options_type_info;
-  static std::unordered_map<std::string, CompressionType>
-      compression_type_string_map;
  static std::unordered_map<std::string, BlockBasedTableOptions::IndexType>
      block_base_table_index_type_string_map;
  static std::unordered_map<std::string, EncodingType> encoding_type_string_map;
@@ -199,7 +199,7 @@ Status RocksDBOptionsParser::ParseStatement(std::string* name,
 namespace {
 bool ReadOneLine(std::istringstream* iss, SequentialFile* seq_file,
                 std::string* output, bool* has_data, Status* result) {
-  const int kBufferSize = 4096;
+  const int kBufferSize = 8192;
  char buffer[kBufferSize + 1];
  Slice input_slice;

@@ -150,7 +150,8 @@ TEST_F(OptionsSettableTest, BlockBasedTableOptionsAllFieldsSettable) {
      "filter_policy=bloomfilter:4:true;whole_key_filtering=1;"
      "format_version=1;"
      "hash_index_allow_collision=false;"
-      "verify_compression=true;read_amp_bytes_per_bit=0",
+      "verify_compression=true;read_amp_bytes_per_bit=0;"
+      "enable_index_compression=false",
      new_bbto));

  ASSERT_EQ(unset_bytes_base,
@@ -32,7 +32,7 @@ namespace port {

 namespace {

-#ifdef OS_LINUX
+#if defined(OS_LINUX) || defined(OS_FREEBSD)
 const char* GetExecutableName() {
  static char name[1024];

@@ -421,36 +421,28 @@ Block::Block(BlockContents&& contents, SequenceNumber _global_seqno,
  }
 }

-InternalIterator* Block::NewIterator(const Comparator* cmp, BlockIter* iter,
-                                     bool total_order_seek, Statistics* stats) {
+BlockIter* Block::NewIterator(const Comparator* cmp, BlockIter* iter,
+                              bool total_order_seek, Statistics* stats) {
+  BlockIter* ret_iter;
+  if (iter != nullptr) {
+    ret_iter = iter;
+  } else {
+    ret_iter = new BlockIter;
+  }
  if (size_ < 2*sizeof(uint32_t)) {
-    if (iter != nullptr) {
-      iter->SetStatus(Status::Corruption("bad block contents"));
-      return iter;
-    } else {
-      return NewErrorInternalIterator(Status::Corruption("bad block contents"));
-    }
+    ret_iter->SetStatus(Status::Corruption("bad block contents"));
+    return ret_iter;
  }
  const uint32_t num_restarts = NumRestarts();
  if (num_restarts == 0) {
-    if (iter != nullptr) {
-      iter->SetStatus(Status::OK());
-      return iter;
-    } else {
-      return NewEmptyInternalIterator();
-    }
+    ret_iter->SetStatus(Status::OK());
+    return ret_iter;
  } else {
    BlockPrefixIndex* prefix_index_ptr =
        total_order_seek ? nullptr : prefix_index_.get();
-
-    if (iter != nullptr) {
-      iter->Initialize(cmp, data_, restart_offset_, num_restarts,
-                       prefix_index_ptr, global_seqno_, read_amp_bitmap_.get());
-    } else {
-      iter = new BlockIter(cmp, data_, restart_offset_, num_restarts,
-                           prefix_index_ptr, global_seqno_,
-                           read_amp_bitmap_.get());
-    }
+    ret_iter->Initialize(cmp, data_, restart_offset_, num_restarts,
+                         prefix_index_ptr, global_seqno_,
+                         read_amp_bitmap_.get());

    if (read_amp_bitmap_) {
      if (read_amp_bitmap_->GetStatistics() != stats) {
@@ -460,7 +452,7 @@ InternalIterator* Block::NewIterator(const Comparator* cmp, BlockIter* iter,
    }
  }

-  return iter;
+  return ret_iter;
 }

 void Block::SetBlockPrefixIndex(BlockPrefixIndex* prefix_index) {
@@ -168,10 +168,10 @@ class Block {
  // If total_order_seek is true, hash_index_ and prefix_index_ are ignored.
  // This option only applies for index block. For data block, hash_index_
  // and prefix_index_ are null, so this option does not matter.
-  InternalIterator* NewIterator(const Comparator* comparator,
-                                BlockIter* iter = nullptr,
-                                bool total_order_seek = true,
-                                Statistics* stats = nullptr);
+  BlockIter* NewIterator(const Comparator* comparator,
+                         BlockIter* iter = nullptr,
+                         bool total_order_seek = true,
+                         Statistics* stats = nullptr);
  void SetBlockPrefixIndex(BlockPrefixIndex* prefix_index);

  // Report an approximation of how much memory has been used.
@@ -191,12 +191,15 @@ class Block {
  const SequenceNumber global_seqno_;

  // No copying allowed
-  Block(const Block&);
-  void operator=(const Block&);
+  Block(const Block&) = delete;
+  void operator=(const Block&) = delete;
 };

 class BlockIter : public InternalIterator {
 public:
+  // Object created using this constructor will behave like an iterator
+  // against an empty block. The state after the creation: Valid()=false
+  // and status() is OK.
  BlockIter()
      : comparator_(nullptr),
        data_(nullptr),
@@ -783,9 +783,12 @@ Status BlockBasedTableBuilder::Finish() {
    WriteRawBlock(meta_index_builder.Finish(), kNoCompression,
                  &metaindex_block_handle);

-    const bool is_data_block = true;
-    WriteBlock(index_blocks.index_block_contents, &index_block_handle,
-               !is_data_block);
+    if (r->table_options.enable_index_compression) {
+      WriteBlock(index_blocks.index_block_contents, &index_block_handle, false);
+    } else {
+      WriteRawBlock(index_blocks.index_block_contents, kNoCompression,
+                    &index_block_handle);
+    }
    // If there are more index partitions, finish them and write them out
    Status& s = index_builder_status;
    while (s.IsIncomplete()) {
@@ -793,8 +796,13 @@ Status BlockBasedTableBuilder::Finish() {
      if (!s.ok() && !s.IsIncomplete()) {
        return s;
      }
-      WriteBlock(index_blocks.index_block_contents, &index_block_handle,
-                 !is_data_block);
+      if (r->table_options.enable_index_compression) {
+        WriteBlock(index_blocks.index_block_contents, &index_block_handle,
+                   false);
+      } else {
+        WriteRawBlock(index_blocks.index_block_contents, kNoCompression,
+                      &index_block_handle);
+      }
      // The last index_block_handle will be for the partition index block
    }
  }
@@ -223,6 +223,9 @@ std::string BlockBasedTableFactory::GetPrintableTableOptions() const {
  snprintf(buffer, kBufferSize, "  format_version: %d\n",
           table_options_.format_version);
  ret.append(buffer);
+  snprintf(buffer, kBufferSize, "  enable_index_compression: %d\n",
+           table_options_.enable_index_compression);
+  ret.append(buffer);
  return ret;
 }

@@ -152,6 +152,9 @@ static std::unordered_map<std::string, OptionTypeInfo>
          OptionType::kBoolean, OptionVerificationType::kNormal, false, 0}},
        {"read_amp_bytes_per_bit",
         {offsetof(struct BlockBasedTableOptions, read_amp_bytes_per_bit),
-          OptionType::kSizeT, OptionVerificationType::kNormal, false, 0}}};
+          OptionType::kSizeT, OptionVerificationType::kNormal, false, 0}},
+        {"enable_index_compression",
+         {offsetof(struct BlockBasedTableOptions, enable_index_compression),
+          OptionType::kBoolean, OptionVerificationType::kNormal, false, 0}}};
 #endif  // !ROCKSDB_LITE
 }  // namespace rocksdb
@@ -759,23 +759,27 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,

  // Read the compression dictionary meta block
  bool found_compression_dict;
-  s = SeekToCompressionDictBlock(meta_iter.get(), &found_compression_dict);
+  BlockHandle compression_dict_handle;
+  s = SeekToCompressionDictBlock(meta_iter.get(), &found_compression_dict, 
+    &compression_dict_handle);
  if (!s.ok()) {
    ROCKS_LOG_WARN(
        rep->ioptions.info_log,
        "Error when seeking to compression dictionary block from file: %s",
        s.ToString().c_str());
-  } else if (found_compression_dict) {
+  } else if (found_compression_dict && !compression_dict_handle.IsNull()) {
    // TODO(andrewkr): Add to block cache if cache_index_and_filter_blocks is
    // true.
-    unique_ptr<BlockContents> compression_dict_block{new BlockContents()};
-    // TODO(andrewkr): ReadMetaBlock repeats SeekToCompressionDictBlock().
-    // maybe decode a handle from meta_iter
-    // and do ReadBlockContents(handle) instead
-    s = rocksdb::ReadMetaBlock(rep->file.get(), prefetch_buffer.get(),
-                               file_size, kBlockBasedTableMagicNumber,
-                               rep->ioptions, rocksdb::kCompressionDictBlock,
-                               compression_dict_block.get());
+    std::unique_ptr<BlockContents> compression_dict_cont{new BlockContents()};
+    PersistentCacheOptions cache_options;
+    ReadOptions read_options;
+    read_options.verify_checksums = false;
+    BlockFetcher compression_block_fetcher(
+      rep->file.get(), prefetch_buffer.get(), rep->footer, read_options,
+      compression_dict_handle, compression_dict_cont.get(), rep->ioptions, false /* decompress */,
+      Slice() /*compression dict*/, cache_options);
+    s = compression_block_fetcher.ReadBlockContents();
+
    if (!s.ok()) {
      ROCKS_LOG_WARN(
          rep->ioptions.info_log,
@@ -783,7 +787,7 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
          "block %s",
          s.ToString().c_str());
    } else {
-      rep->compression_dict_block = std::move(compression_dict_block);
+      rep->compression_dict_block = std::move(compression_dict_cont);
    }
  }

@@ -1438,7 +1442,7 @@ InternalIterator* BlockBasedTable::NewIndexIterator(
  return iter;
 }

-InternalIterator* BlockBasedTable::NewDataBlockIterator(
+BlockIter* BlockBasedTable::NewDataBlockIterator(
    Rep* rep, const ReadOptions& ro, const Slice& index_value,
    BlockIter* input_iter, bool is_index, GetContext* get_context) {
  BlockHandle handle;
@@ -1454,7 +1458,7 @@ InternalIterator* BlockBasedTable::NewDataBlockIterator(
 // into an iterator over the contents of the corresponding block.
 // If input_iter is null, new a iterator
 // If input_iter is not null, update this iter and return it
-InternalIterator* BlockBasedTable::NewDataBlockIterator(
+BlockIter* BlockBasedTable::NewDataBlockIterator(
    Rep* rep, const ReadOptions& ro, const BlockHandle& handle,
    BlockIter* input_iter, bool is_index, GetContext* get_context, Status s) {
  PERF_TIMER_GUARD(new_table_block_iter_nanos);
@@ -1472,16 +1476,18 @@ InternalIterator* BlockBasedTable::NewDataBlockIterator(
                                  get_context);
  }

+  BlockIter* iter;
+  if (input_iter != nullptr) {
+    iter = input_iter;
+  } else {
+    iter = new BlockIter;
+  }
  // Didn't get any data from block caches.
  if (s.ok() && block.value == nullptr) {
    if (no_io) {
      // Could not read from block_cache and can't do IO
-      if (input_iter != nullptr) {
-        input_iter->SetStatus(Status::Incomplete("no blocking io"));
-        return input_iter;
-      } else {
-        return NewErrorInternalIterator(Status::Incomplete("no blocking io"));
-      }
+      iter->SetStatus(Status::Incomplete("no blocking io"));
+      return iter;
    }
    std::unique_ptr<Block> block_value;
    s = ReadBlockFromFile(rep->file.get(), nullptr /* prefetch_buffer */,
@@ -1494,10 +1500,9 @@ InternalIterator* BlockBasedTable::NewDataBlockIterator(
    }
  }

-  InternalIterator* iter;
  if (s.ok()) {
    assert(block.value != nullptr);
-    iter = block.value->NewIterator(&rep->internal_comparator, input_iter, true,
+    iter = block.value->NewIterator(&rep->internal_comparator, iter, true,
                                    rep->ioptions.statistics);
    if (block.cache_handle != nullptr) {
      iter->RegisterCleanup(&ReleaseCachedEntry, block_cache,
@@ -1507,12 +1512,7 @@ InternalIterator* BlockBasedTable::NewDataBlockIterator(
    }
  } else {
    assert(block.value == nullptr);
-    if (input_iter != nullptr) {
-      input_iter->SetStatus(s);
-      iter = input_iter;
-    } else {
-      iter = NewErrorInternalIterator(s);
-    }
+    iter->SetStatus(s);
  }
  return iter;
 }
@@ -215,14 +215,17 @@ class BlockBasedTable : public TableReader {
 private:
  friend class MockedBlockBasedTable;
  // input_iter: if it is not null, update this one and return it as Iterator
-  static InternalIterator* NewDataBlockIterator(
-      Rep* rep, const ReadOptions& ro, const Slice& index_value,
-      BlockIter* input_iter = nullptr, bool is_index = false,
-      GetContext* get_context = nullptr);
-  static InternalIterator* NewDataBlockIterator(
-      Rep* rep, const ReadOptions& ro, const BlockHandle& block_hanlde,
-      BlockIter* input_iter = nullptr, bool is_index = false,
-      GetContext* get_context = nullptr, Status s = Status());
+  static BlockIter* NewDataBlockIterator(Rep* rep, const ReadOptions& ro,
+                                         const Slice& index_value,
+                                         BlockIter* input_iter = nullptr,
+                                         bool is_index = false,
+                                         GetContext* get_context = nullptr);
+  static BlockIter* NewDataBlockIterator(Rep* rep, const ReadOptions& ro,
+                                         const BlockHandle& block_hanlde,
+                                         BlockIter* input_iter = nullptr,
+                                         bool is_index = false,
+                                         GetContext* get_context = nullptr,
+                                         Status s = Status());
  // If block cache enabled (compressed or uncompressed), looks for the block
  // identified by handle in (1) uncompressed cache, (2) compressed cache, and
  // then (3) file. If found, inserts into the cache(s) that were searched
@@ -27,7 +27,7 @@ const size_t kFadviseTrigger = 1024 * 1024; // 1MB
 struct SstFileWriter::Rep {
  Rep(const EnvOptions& _env_options, const Options& options,
      Env::IOPriority _io_priority, const Comparator* _user_comparator,
-      ColumnFamilyHandle* _cfh, bool _invalidate_page_cache)
+      ColumnFamilyHandle* _cfh, bool _invalidate_page_cache, bool _skip_filters)
      : env_options(_env_options),
        ioptions(options),
        mutable_cf_options(options),
@@ -35,7 +35,8 @@ struct SstFileWriter::Rep {
        internal_comparator(_user_comparator),
        cfh(_cfh),
        invalidate_page_cache(_invalidate_page_cache),
-        last_fadvise_size(0) {}
+        last_fadvise_size(0),
+        skip_filters(_skip_filters) {}

  std::unique_ptr<WritableFileWriter> file_writer;
  std::unique_ptr<TableBuilder> builder;
@@ -54,6 +55,7 @@ struct SstFileWriter::Rep {
  // The size of the file during the last time we called Fadvise to remove
  // cached pages from page cache.
  uint64_t last_fadvise_size;
+  bool skip_filters;
  Status Add(const Slice& user_key, const Slice& value,
             const ValueType value_type) {
    if (!builder) {
@@ -122,9 +124,9 @@ SstFileWriter::SstFileWriter(const EnvOptions& env_options,
                             const Comparator* user_comparator,
                             ColumnFamilyHandle* column_family,
                             bool invalidate_page_cache,
-                             Env::IOPriority io_priority)
+                             Env::IOPriority io_priority, bool skip_filters)
    : rep_(new Rep(env_options, options, io_priority, user_comparator,
-                   column_family, invalidate_page_cache)) {
+                   column_family, invalidate_page_cache, skip_filters)) {
  rep_->file_info.file_size = 0;
 }

@@ -189,8 +191,8 @@ Status SstFileWriter::Open(const std::string& file_path) {
  TableBuilderOptions table_builder_options(
      r->ioptions, r->internal_comparator, &int_tbl_prop_collector_factories,
      compression_type, r->ioptions.compression_opts,
-      nullptr /* compression_dict */, false /* skip_filters */,
-      r->column_family_name, unknown_level);
+      nullptr /* compression_dict */, r->skip_filters, r->column_family_name,
+      unknown_level);
  r->file_writer.reset(
      new WritableFileWriter(std::move(sst_file), r->env_options));

@@ -215,8 +215,9 @@ Status SeekToPropertiesBlock(InternalIterator* meta_iter, bool* is_found) {

 // Seek to the compression dictionary block.
 // Return true if it successfully seeks to that block.
-Status SeekToCompressionDictBlock(InternalIterator* meta_iter, bool* is_found) {
-  return SeekToMetaBlock(meta_iter, kCompressionDictBlock, is_found);
+Status SeekToCompressionDictBlock(InternalIterator* meta_iter, bool* is_found,
+                                  BlockHandle* block_handle) {
+  return SeekToMetaBlock(meta_iter, kCompressionDictBlock, is_found, block_handle);
 }

 Status SeekToRangeDelBlock(InternalIterator* meta_iter, bool* is_found,
@@ -21,7 +21,8 @@ Status SeekToPropertiesBlock(InternalIterator* meta_iter, bool* is_found);
 // Seek to the compression dictionary block.
 // If it successfully seeks to the properties block, "is_found" will be
 // set to true.
-Status SeekToCompressionDictBlock(InternalIterator* meta_iter, bool* is_found);
+Status SeekToCompressionDictBlock(InternalIterator* meta_iter, bool* is_found,
+                          BlockHandle* block_handle);

 // TODO(andrewkr) should not put all meta block in table_properties.h/cc
 Status SeekToRangeDelBlock(InternalIterator* meta_iter, bool* is_found,
@@ -1003,7 +1003,10 @@ class TableTest : public testing::Test {
 };

 class GeneralTableTest : public TableTest {};
-class BlockBasedTableTest : public TableTest {};
+class BlockBasedTableTest : public TableTest {
+ protected:
+  uint64_t IndexUncompressedHelper(bool indexCompress);
+};
 class PlainTableTest : public TableTest {};
 class TablePropertyTest : public testing::Test {};

@@ -1064,13 +1067,17 @@ TEST_F(BlockBasedTableTest, BasicBlockBasedTableProperties) {
  stl_wrappers::KVMap kvmap;
  Options options;
  options.compression = kNoCompression;
+  options.statistics = CreateDBStatistics();
+  options.statistics->stats_level_ = StatsLevel::kAll;
  BlockBasedTableOptions table_options;
  table_options.block_restart_interval = 1;
  options.table_factory.reset(NewBlockBasedTableFactory(table_options));

-  const ImmutableCFOptions ioptions(options);
+  ImmutableCFOptions ioptions(options);
+  ioptions.statistics = options.statistics.get();
  c.Finish(options, ioptions, table_options,
           GetPlainInternalComparator(options.comparator), &keys, &kvmap);
+  ASSERT_EQ(options.statistics->getTickerCount(NUMBER_BLOCK_NOT_COMPRESSED), 0);

  auto& props = *c.GetTableReader()->GetTableProperties();
  ASSERT_EQ(kvmap.size(), props.num_entries);
@@ -1094,6 +1101,41 @@ TEST_F(BlockBasedTableTest, BasicBlockBasedTableProperties) {
  c.ResetTableReader();
 }

+#ifdef SNAPPY
+uint64_t BlockBasedTableTest::IndexUncompressedHelper(bool compressed) {
+  TableConstructor c(BytewiseComparator(), true /* convert_to_internal_key_ */);
+  constexpr size_t kNumKeys = 10000;
+
+  for (size_t k = 0; k < kNumKeys; ++k) {
+    c.Add("key" + ToString(k), "val" + ToString(k));
+  }
+
+  std::vector<std::string> keys;
+  stl_wrappers::KVMap kvmap;
+  Options options;
+  options.compression = kSnappyCompression;
+  options.statistics = CreateDBStatistics();
+  options.statistics->stats_level_ = StatsLevel::kAll;
+  BlockBasedTableOptions table_options;
+  table_options.block_restart_interval = 1;
+  table_options.enable_index_compression = compressed;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+  ImmutableCFOptions ioptions(options);
+  ioptions.statistics = options.statistics.get();
+  c.Finish(options, ioptions, table_options,
+           GetPlainInternalComparator(options.comparator), &keys, &kvmap);
+  c.ResetTableReader();
+  return options.statistics->getTickerCount(NUMBER_BLOCK_COMPRESSED);
+}
+TEST_F(BlockBasedTableTest, IndexUncompressed) {
+  uint64_t tbl1_compressed_cnt = IndexUncompressedHelper(true);
+  uint64_t tbl2_compressed_cnt = IndexUncompressedHelper(false);
+  // tbl1_compressed_cnt should include 1 index block
+  EXPECT_EQ(tbl2_compressed_cnt + 1, tbl1_compressed_cnt);
+}
+#endif  // SNAPPY
+
 TEST_F(BlockBasedTableTest, BlockBasedTableProperties2) {
  TableConstructor c(&reverse_key_comparator);
  std::vector<std::string> keys;
@@ -42,7 +42,7 @@ EOF
 done

 declare -a backward_compatible_checkout_objs=("2.2.fb.branch" "2.3.fb.branch" "2.4.fb.branch" "2.5.fb.branch" "2.6.fb.branch" "2.7.fb.branch" "2.8.1.fb" "3.0.fb.branch" "3.1.fb" "3.2.fb" "3.3.fb" "3.4.fb" "3.5.fb" "3.6.fb" "3.7.fb" "3.8.fb" "3.9.fb")
-declare -a forward_compatible_checkout_objs=("3.10.fb" "3.11.fb" "3.12.fb" "3.13.fb" "4.0.fb" "4.1.fb" "4.2.fb" "4.3.fb" "4.4.fb" "4.5.fb" "4.6.fb" "4.7.fb" "4.8.fb" "4.9.fb" "4.10.fb" "4.11.fb" "4.12.fb" "4.13.fb" "5.0.fb" "5.1.fb" "5.2.fb" "5.3.fb" "5.4.fb" "5.5.fb" "5.6.fb" "5.7.fb" "5.8.fb" "5.9.fb")
+declare -a forward_compatible_checkout_objs=("3.10.fb" "3.11.fb" "3.12.fb" "3.13.fb" "4.0.fb" "4.1.fb" "4.2.fb" "4.3.fb" "4.4.fb" "4.5.fb" "4.6.fb" "4.7.fb" "4.8.fb" "4.9.fb" "4.10.fb" "4.11.fb" "4.12.fb" "4.13.fb" "5.0.fb" "5.1.fb" "5.2.fb" "5.3.fb" "5.4.fb" "5.5.fb" "5.6.fb" "5.7.fb" "5.8.fb" "5.9.fb" "5.10.fb")
 declare -a checkout_objs=(${backward_compatible_checkout_objs[@]} ${forward_compatible_checkout_objs[@]})

 generate_db()
@@ -442,6 +442,10 @@ DEFINE_int32(read_amp_bytes_per_bit,
             rocksdb::BlockBasedTableOptions().read_amp_bytes_per_bit,
             "Number of bytes per bit to be used in block read-amp bitmap");

+DEFINE_bool(enable_index_compression,
+            rocksdb::BlockBasedTableOptions().enable_index_compression,
+            "Compress the index block");
+
 DEFINE_int64(compressed_cache_size, -1,
             "Number of bytes to use as a cache of compressed data.");

@@ -3110,6 +3114,8 @@ void VerifyDBFromDB(std::string& truth_db_name) {
      block_based_options.filter_policy = filter_policy_;
      block_based_options.format_version = 2;
      block_based_options.read_amp_bytes_per_bit = FLAGS_read_amp_bytes_per_bit;
+      block_based_options.enable_index_compression =
+          FLAGS_enable_index_compression;
      if (FLAGS_read_cache_path != "") {
 #ifndef ROCKSDB_LITE
        Status rc_status;
@@ -103,6 +103,14 @@ class AutoRollLogger : public Logger {
  std::string ValistToString(const char* format, va_list args) const;
  // Write the logs marked as headers to the new log file
  void WriteHeaderInfo();
+  // Implementation of Close()
+  virtual Status CloseImpl() override {
+    if (logger_) {
+      return logger_->Close();
+    } else {
+      return Status::OK();
+    }
+  }

  std::string log_fname_; // Current active info log's file name.
  std::string dbname_;
@@ -354,6 +354,45 @@ TEST_F(AutoRollLoggerTest, InfoLogLevel) {
  inFile.close();
 }

+TEST_F(AutoRollLoggerTest, Close) {
+  InitTestDb();
+
+  size_t log_size = 8192;
+  size_t log_lines = 0;
+  AutoRollLogger logger(Env::Default(), kTestDir, "", log_size, 0);
+  for (int log_level = InfoLogLevel::HEADER_LEVEL;
+       log_level >= InfoLogLevel::DEBUG_LEVEL; log_level--) {
+    logger.SetInfoLogLevel((InfoLogLevel)log_level);
+    for (int log_type = InfoLogLevel::DEBUG_LEVEL;
+         log_type <= InfoLogLevel::HEADER_LEVEL; log_type++) {
+      // log messages with log level smaller than log_level will not be
+      // logged.
+      LogMessage((InfoLogLevel)log_type, &logger, kSampleMessage.c_str());
+    }
+    log_lines += InfoLogLevel::HEADER_LEVEL - log_level + 1;
+  }
+  for (int log_level = InfoLogLevel::HEADER_LEVEL;
+       log_level >= InfoLogLevel::DEBUG_LEVEL; log_level--) {
+    logger.SetInfoLogLevel((InfoLogLevel)log_level);
+
+    // again, messages with level smaller than log_level will not be logged.
+    ROCKS_LOG_HEADER(&logger, "%s", kSampleMessage.c_str());
+    ROCKS_LOG_DEBUG(&logger, "%s", kSampleMessage.c_str());
+    ROCKS_LOG_INFO(&logger, "%s", kSampleMessage.c_str());
+    ROCKS_LOG_WARN(&logger, "%s", kSampleMessage.c_str());
+    ROCKS_LOG_ERROR(&logger, "%s", kSampleMessage.c_str());
+    ROCKS_LOG_FATAL(&logger, "%s", kSampleMessage.c_str());
+    log_lines += InfoLogLevel::HEADER_LEVEL - log_level + 1;
+  }
+  ASSERT_EQ(logger.Close(), Status::OK());
+
+  std::ifstream inFile(AutoRollLoggerTest::kLogFile.c_str());
+  size_t lines = std::count(std::istreambuf_iterator<char>(inFile),
+                         std::istreambuf_iterator<char>(), '\n');
+  ASSERT_EQ(log_lines, lines);
+  inFile.close();
+}
+
 // Test the logger Header function for roll over logs
 // We expect the new logs creates as roll over to carry the headers specified
 static std::vector<std::string> GetOldFileNames(const std::string& path) {
@@ -73,6 +73,8 @@ void BlobDBOptions::Dump(Logger* log) const {
                   blob_dir_size);
  ROCKS_LOG_HEADER(log, "           blob_db_options.ttl_range_secs: %" PRIu32,
                   ttl_range_secs);
+  ROCKS_LOG_HEADER(log, "            blob_db_options.min_blob_size: %" PRIu64,
+                   min_blob_size);
  ROCKS_LOG_HEADER(log, "           blob_db_options.bytes_per_sync: %" PRIu64,
                   bytes_per_sync);
  ROCKS_LOG_HEADER(log, "           blob_db_options.blob_file_size: %" PRIu64,
@@ -859,6 +859,9 @@ std::shared_ptr<BlobFile> BlobDBImpl::GetOldestBlobFile() {
  CopyBlobFiles(&blob_files, [](const std::shared_ptr<BlobFile>& f) {
    return !f->Obsolete() && f->Immutable();
  });
+  if (blob_files.empty()) {
+    return nullptr;
+  }
  blobf_compare_ttl compare;
  return *std::min_element(blob_files.begin(), blob_files.end(), compare);
 }
@@ -889,6 +892,7 @@ bool BlobDBImpl::EvictOldestBlobFile() {
               oldest_file->BlobCount());
    RecordTick(statistics_, BLOB_DB_FIFO_BYTES_EVICTED,
               oldest_file->GetFileSize());
+    TEST_SYNC_POINT("BlobDBImpl::EvictOldestBlobFile:Evicted");
    return true;
  }

@@ -1061,15 +1061,19 @@ TEST_F(BlobDBTest, OutOfSpace) {
 }

 TEST_F(BlobDBTest, EvictOldestFileWhenCloseToSpaceLimit) {
-  // Use mock env to stop wall clock.
-  Options options;
  BlobDBOptions bdb_options;
  bdb_options.blob_dir_size = 270;
  bdb_options.blob_file_size = 100;
-  bdb_options.disable_background_tasks = true;
  bdb_options.is_fifo = true;
+  bdb_options.disable_background_tasks = true;
  Open(bdb_options);

+  std::atomic<int> evict_count{0};
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobDBImpl::EvictOldestBlobFile:Evicted",
+      [&](void *) { evict_count++; });
+  SyncPoint::GetInstance()->EnableProcessing();
+
  // Each stored blob has an overhead of 32 bytes currently.
  // So a 100 byte blob should take up 132 bytes.
  std::string value(100, 'v');
@@ -1093,6 +1097,28 @@ TEST_F(BlobDBTest, EvictOldestFileWhenCloseToSpaceLimit) {
  bdb_impl->TEST_DeleteObsoleteFiles();
  obsolete_files = bdb_impl->TEST_GetObsoleteFiles();
  ASSERT_TRUE(obsolete_files.empty());
+  ASSERT_EQ(1, evict_count);
+}
+
+TEST_F(BlobDBTest, NoOldestFileToEvict) {
+  Options options;
+  BlobDBOptions bdb_options;
+  bdb_options.blob_dir_size = 1000;
+  bdb_options.blob_file_size = 5000;
+  bdb_options.is_fifo = true;
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options);
+
+  std::atomic<int> evict_count{0};
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobDBImpl::EvictOldestBlobFile:Evicted",
+      [&](void *) { evict_count++; });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  std::string value(2000, 'v');
+  ASSERT_OK(Put("foo", std::string(2000, 'v')));
+  ASSERT_OK(Put("bar", std::string(2000, 'v')));
+  ASSERT_EQ(0, evict_count);
 }

 TEST_F(BlobDBTest, InlineSmallValues) {
@@ -42,11 +42,12 @@ class CheckpointTest : public testing::Test {
  DB* db_;
  Options last_options_;
  std::vector<ColumnFamilyHandle*> handles_;
+  std::string snapshot_name_;

  CheckpointTest() : env_(Env::Default()) {
    env_->SetBackgroundThreads(1, Env::LOW);
    env_->SetBackgroundThreads(1, Env::HIGH);
-    dbname_ = test::TmpDir(env_) + "/db_test";
+    dbname_ = test::TmpDir(env_) + "/checkpoint_test";
    alternative_wal_dir_ = dbname_ + "/wal";
    auto options = CurrentOptions();
    auto delete_options = options;
@@ -55,6 +56,12 @@ class CheckpointTest : public testing::Test {
    // Destroy it for not alternative WAL dir is used.
    EXPECT_OK(DestroyDB(dbname_, options));
    db_ = nullptr;
+    snapshot_name_ = test::TmpDir(env_) + "/snapshot";
+    std::string snapshot_tmp_name = snapshot_name_ + ".tmp";
+    EXPECT_OK(DestroyDB(snapshot_name_, options));
+    env_->DeleteDir(snapshot_name_);
+    EXPECT_OK(DestroyDB(snapshot_tmp_name, options));
+    env_->DeleteDir(snapshot_tmp_name);
    Reopen(options);
  }

@@ -69,6 +76,7 @@ class CheckpointTest : public testing::Test {
    options.db_paths.emplace_back(dbname_ + "_3", 0);
    options.db_paths.emplace_back(dbname_ + "_4", 0);
    EXPECT_OK(DestroyDB(dbname_, options));
+    EXPECT_OK(DestroyDB(snapshot_name_, options));
  }

  // Return the current option configuration.
@@ -219,7 +227,6 @@ class CheckpointTest : public testing::Test {
 TEST_F(CheckpointTest, GetSnapshotLink) {
  for (uint64_t log_size_for_flush : {0, 1000000}) {
    Options options;
-    const std::string snapshot_name = test::TmpDir(env_) + "/snapshot";
    DB* snapshotDB;
    ReadOptions roptions;
    std::string result;
@@ -229,8 +236,6 @@ TEST_F(CheckpointTest, GetSnapshotLink) {
    delete db_;
    db_ = nullptr;
    ASSERT_OK(DestroyDB(dbname_, options));
-    ASSERT_OK(DestroyDB(snapshot_name, options));
-    env_->DeleteDir(snapshot_name);

    // Create a database
    Status s;
@@ -240,14 +245,14 @@ TEST_F(CheckpointTest, GetSnapshotLink) {
    ASSERT_OK(Put(key, "v1"));
    // Take a snapshot
    ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
-    ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name, log_size_for_flush));
+    ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name_, log_size_for_flush));
    ASSERT_OK(Put(key, "v2"));
    ASSERT_EQ("v2", Get(key));
    ASSERT_OK(Flush());
    ASSERT_EQ("v2", Get(key));
    // Open snapshot and verify contents while DB is running
    options.create_if_missing = false;
-    ASSERT_OK(DB::Open(options, snapshot_name, &snapshotDB));
+    ASSERT_OK(DB::Open(options, snapshot_name_, &snapshotDB));
    ASSERT_OK(snapshotDB->Get(roptions, key, &result));
    ASSERT_EQ("v1", result);
    delete snapshotDB;
@@ -260,7 +265,7 @@ TEST_F(CheckpointTest, GetSnapshotLink) {

    // Open snapshot and verify contents
    options.create_if_missing = false;
-    dbname_ = snapshot_name;
+    dbname_ = snapshot_name_;
    ASSERT_OK(DB::Open(options, dbname_, &db_));
    ASSERT_EQ("v1", Get(key));
    delete db_;
@@ -289,21 +294,17 @@ TEST_F(CheckpointTest, CheckpointCF) {
  ASSERT_OK(Put(4, "four", "four"));
  ASSERT_OK(Put(5, "five", "five"));

-  const std::string snapshot_name = test::TmpDir(env_) + "/snapshot";
  DB* snapshotDB;
  ReadOptions roptions;
  std::string result;
  std::vector<ColumnFamilyHandle*> cphandles;

-  ASSERT_OK(DestroyDB(snapshot_name, options));
-  env_->DeleteDir(snapshot_name);
-
  Status s;
  // Take a snapshot
  rocksdb::port::Thread t([&]() {
    Checkpoint* checkpoint;
    ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
-    ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name));
+    ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name_));
    delete checkpoint;
  });
  TEST_SYNC_POINT("CheckpointTest::CheckpointCF:1");
@@ -331,7 +332,7 @@ TEST_F(CheckpointTest, CheckpointCF) {
    for (size_t i = 0; i < cfs.size(); ++i) {
      column_families.push_back(ColumnFamilyDescriptor(cfs[i], options));
    }
-  ASSERT_OK(DB::Open(options, snapshot_name,
+  ASSERT_OK(DB::Open(options, snapshot_name_,
        column_families, &cphandles, &snapshotDB));
  ASSERT_OK(snapshotDB->Get(roptions, cphandles[0], "Default", &result));
  ASSERT_EQ("Default1", result);
@@ -344,7 +345,6 @@ TEST_F(CheckpointTest, CheckpointCF) {
  cphandles.clear();
  delete snapshotDB;
  snapshotDB = nullptr;
-  ASSERT_OK(DestroyDB(snapshot_name, options));
 }

 TEST_F(CheckpointTest, CheckpointCFNoFlush) {
@@ -358,15 +358,11 @@ TEST_F(CheckpointTest, CheckpointCFNoFlush) {
  Flush();
  ASSERT_OK(Put(2, "two", "two"));

-  const std::string snapshot_name = test::TmpDir(env_) + "/snapshot";
  DB* snapshotDB;
  ReadOptions roptions;
  std::string result;
  std::vector<ColumnFamilyHandle*> cphandles;

-  ASSERT_OK(DestroyDB(snapshot_name, options));
-  env_->DeleteDir(snapshot_name);
-
  Status s;
  // Take a snapshot
  rocksdb::SyncPoint::GetInstance()->SetCallBack(
@@ -377,7 +373,7 @@ TEST_F(CheckpointTest, CheckpointCFNoFlush) {
  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
  Checkpoint* checkpoint;
  ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
-  ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name, 1000000));
+  ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name_, 1000000));
  rocksdb::SyncPoint::GetInstance()->DisableProcessing();

  delete checkpoint;
@@ -395,7 +391,7 @@ TEST_F(CheckpointTest, CheckpointCFNoFlush) {
  for (size_t i = 0; i < cfs.size(); ++i) {
    column_families.push_back(ColumnFamilyDescriptor(cfs[i], options));
  }
-  ASSERT_OK(DB::Open(options, snapshot_name, column_families, &cphandles,
+  ASSERT_OK(DB::Open(options, snapshot_name_, column_families, &cphandles,
                     &snapshotDB));
  ASSERT_OK(snapshotDB->Get(roptions, cphandles[0], "Default", &result));
  ASSERT_EQ("Default", result);
@@ -409,14 +405,9 @@ TEST_F(CheckpointTest, CheckpointCFNoFlush) {
  cphandles.clear();
  delete snapshotDB;
  snapshotDB = nullptr;
-  ASSERT_OK(DestroyDB(snapshot_name, options));
 }

 TEST_F(CheckpointTest, CurrentFileModifiedWhileCheckpointing) {
-  const std::string kSnapshotName = test::TmpDir(env_) + "/snapshot";
-  ASSERT_OK(DestroyDB(kSnapshotName, CurrentOptions()));
-  env_->DeleteDir(kSnapshotName);
-
  Options options = CurrentOptions();
  options.max_manifest_file_size = 0;  // always rollover manifest for file add
  Reopen(options);
@@ -438,7 +429,7 @@ TEST_F(CheckpointTest, CurrentFileModifiedWhileCheckpointing) {
  rocksdb::port::Thread t([&]() {
    Checkpoint* checkpoint;
    ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
-    ASSERT_OK(checkpoint->CreateCheckpoint(kSnapshotName));
+    ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name_));
    delete checkpoint;
  });
  TEST_SYNC_POINT(
@@ -452,18 +443,15 @@ TEST_F(CheckpointTest, CurrentFileModifiedWhileCheckpointing) {
  DB* snapshotDB;
  // Successful Open() implies that CURRENT pointed to the manifest in the
  // checkpoint.
-  ASSERT_OK(DB::Open(options, kSnapshotName, &snapshotDB));
+  ASSERT_OK(DB::Open(options, snapshot_name_, &snapshotDB));
  delete snapshotDB;
  snapshotDB = nullptr;
 }

 TEST_F(CheckpointTest, CurrentFileModifiedWhileCheckpointing2PC) {
  Close();
-  const std::string kSnapshotName = test::TmpDir(env_) + "/snapshot";
  const std::string dbname = test::TmpDir() + "/transaction_testdb";
-  ASSERT_OK(DestroyDB(kSnapshotName, CurrentOptions()));
  ASSERT_OK(DestroyDB(dbname, CurrentOptions()));
-  env_->DeleteDir(kSnapshotName);
  env_->DeleteDir(dbname);

  Options options = CurrentOptions();
@@ -521,7 +509,7 @@ TEST_F(CheckpointTest, CurrentFileModifiedWhileCheckpointing2PC) {
  rocksdb::port::Thread t([&]() {
    Checkpoint* checkpoint;
    ASSERT_OK(Checkpoint::Create(txdb, &checkpoint));
-    ASSERT_OK(checkpoint->CreateCheckpoint(kSnapshotName));
+    ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name_));
    delete checkpoint;
  });
  TEST_SYNC_POINT(
@@ -536,7 +524,7 @@ TEST_F(CheckpointTest, CurrentFileModifiedWhileCheckpointing2PC) {

  // No more than two logs files should exist.
  std::vector<std::string> files;
-  env_->GetChildren(kSnapshotName, &files);
+  env_->GetChildren(snapshot_name_, &files);
  int num_log_files = 0;
  for (auto& file : files) {
    uint64_t num;
@@ -559,7 +547,7 @@ TEST_F(CheckpointTest, CurrentFileModifiedWhileCheckpointing2PC) {
  column_families.push_back(
      ColumnFamilyDescriptor("CFB", ColumnFamilyOptions()));
  std::vector<rocksdb::ColumnFamilyHandle*> cf_handles;
-  ASSERT_OK(TransactionDB::Open(options, txn_db_options, kSnapshotName,
+  ASSERT_OK(TransactionDB::Open(options, txn_db_options, snapshot_name_,
                                column_families, &cf_handles, &snapshotDB));
  ASSERT_OK(snapshotDB->Get(read_options, "foo", &value));
  ASSERT_EQ(value, "bar");
@@ -309,5 +309,16 @@ Iterator* DBWithTTLImpl::NewIterator(const ReadOptions& opts,
  return new TtlIterator(db_->NewIterator(opts, column_family));
 }

+void DBWithTTLImpl::SetTtl(ColumnFamilyHandle *h, int32_t ttl) {
+  std::shared_ptr<TtlCompactionFilterFactory> filter;
+  Options opts;
+  opts = GetOptions(h);
+  filter = std::static_pointer_cast<TtlCompactionFilterFactory>(
+                                       opts.compaction_filter_factory);
+  if (!filter)
+    return;
+  filter->SetTtl(ttl);
+}
+
 }  // namespace rocksdb
 #endif  // ROCKSDB_LITE
@@ -94,6 +94,10 @@ class DBWithTTLImpl : public DBWithTTL {
  static const int32_t kMinTimestamp = 1368146402;  // 05/09/2013:5:40PM GMT-8

  static const int32_t kMaxTimestamp = 2147483647;  // 01/18/2038:7:14PM GMT-8
+
+  void SetTtl(int32_t ttl) override { SetTtl(DefaultColumnFamily(), ttl); }
+
+  void SetTtl(ColumnFamilyHandle *h, int32_t ttl) override;
 };

 class TtlIterator : public Iterator {
@@ -209,6 +213,10 @@ class TtlCompactionFilterFactory : public CompactionFilterFactory {
        ttl_, env_, nullptr, std::move(user_comp_filter_from_factory)));
  }

+  void SetTtl(int32_t ttl) {
+    ttl_ = ttl;
+  }
+
  virtual const char* Name() const override {
    return "TtlCompactionFilterFactory";
  }
@@ -283,6 +283,12 @@ class TtlTest : public testing::Test {
    delete dbiter;
  }

+  // Set ttl on open db
+  void SetTtl(int32_t ttl, ColumnFamilyHandle* cf = nullptr) {
+    ASSERT_TRUE(db_ttl_);
+    cf == nullptr ? db_ttl_->SetTtl(ttl) : db_ttl_->SetTtl(cf, ttl);
+  }
+
  class TestFilter : public CompactionFilter {
   public:
    TestFilter(const int64_t kSampleSize, const std::string& kNewValue)
@@ -626,6 +632,17 @@ TEST_F(TtlTest, ColumnFamiliesTest) {
  db_ttl_ = nullptr;
 }

+// Puts a set of values and checks its absence using Get after ttl
+TEST_F(TtlTest, ChangeTtlOnOpenDb) {
+  MakeKVMap(kSampleSize_);
+
+  OpenTtl(1);                                  // T=0:Open the db with ttl = 2
+  SetTtl(3);
+  PutValues(0, kSampleSize_);		       // T=0:Insert Set1. Delete at t=2
+  SleepCompactCheck(2, 0, kSampleSize_, true); // T=2:Set1 should be there
+  CloseTtl();
+}
+
 } //  namespace rocksdb

 // A black-box test for the ttl wrapper around rocksdb