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

Summary: Making platform 007 (gcc 7) default in build_detect_platform.sh. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5947 Differential Revision: D18038837 Pulled By: vjnadimpalli fbshipit-source-id: 9ac2ddaa93bf328a416faec028970e039886378e
Add latest toolchain (gcc-8, etc.) build support for fbcode users (#4923 )
2026-07-07 14:47:40 +08:00 · 2019-10-30 10:05:36 -07:00 · 2019-10-30 10:05:26 -07:00 · 2018-12-05 19:02:18 -08:00 · 2018-12-05 19:01:06 -08:00 · 2018-12-05 18:17:12 -08:00
53 changed files with 1118 additions and 524 deletions
@@ -1,5 +1,22 @@
 # Rocksdb Change Log
-## Unreleased
+## 5.17.4 (12/05/2018)
+### Bug Fixes
+* Fix a deadlock caused by compaction and file ingestion waiting for each other in the event of write stalls.
+
+# 5.17.2 (10/24/2018)
+### Bug Fixes
+* Fix the bug that WriteBatchWithIndex's SeekForPrev() doesn't see the entries with the same key.
+
+# 5.17.1 (10/16/2018)
+### Bug Fixes
+* Fix slow flush/compaction when DB contains many snapshots. The problem became noticeable to us in DBs with 100,000+ snapshots, though it will affect others at different thresholds.
+* Properly set the stop key for a truncated manual CompactRange
+* Fix corner case where a write group leader blocked due to write stall blocks other writers in queue with WriteOptions::no_slowdown set.
+
+### New Features
+* Introduced CacheAllocator, which lets the user specify custom allocator for memory in block cache.
+
+## 5.17.0 (10/05/2018)
 ### Public API Change
 * `OnTableFileCreated` will now be called for empty files generated during compaction. In that case, `TableFileCreationInfo::file_path` will be "(nil)" and `TableFileCreationInfo::file_size` will be zero.
 * Add `FlushOptions::allow_write_stall`, which controls whether Flush calls start working immediately, even if it causes user writes to stall, or will wait until flush can be performed without causing write stall (similar to `CompactRangeOptions::allow_write_stall`). Note that the default value is false, meaning we add delay to Flush calls until stalling can be avoided when possible. This is behavior change compared to previous RocksDB versions, where Flush calls didn't check if they might cause stall or not.
@@ -7,14 +24,25 @@

 ### New Features
 * TransactionOptions::skip_concurrency_control allows pessimistic transactions to skip the overhead of concurrency control. Could be used for optimizing certain transactions or during recovery.
-* Introduced CacheAllocator, which lets the user specify custom allocator for memory in block cache.

 ### Bug Fixes
 * Avoid creating empty SSTs and subsequently deleting them in certain cases during compaction.
 * Sync CURRENT file contents during checkpoint.
-* Fix format_version 4 bug with partitioned filters
+
+## 5.16.3 (10/1/2018)
+### Bug Fixes
 * Fix crash caused when `CompactFiles` run with `CompactionOptions::compression == CompressionType::kDisableCompressionOption`. Now that setting causes the compression type to be chosen according to the column family-wide compression options.

+## 5.16.2 (9/21/2018)
+### Bug Fixes
+* Fix bug in partition filters with format_version=4.
+
+## 5.16.1 (9/17/2018)
+### Bug Fixes
+* Remove trace_analyzer_tool from rocksdb_lib target in TARGETS file.
+* Fix RocksDB Java build and tests.
+* Remove sync point in Block destructor.
+
 ## 5.16.0 (8/21/2018)
 ### Public API Change
 * The merge operands are passed to `MergeOperator::ShouldMerge` in the reversed order relative to how they were merged (passed to FullMerge or FullMergeV2) for performance reasons
@@ -292,7 +292,7 @@ cpp_library(
        "utilities/write_batch_with_index/write_batch_with_index.cc",
        "utilities/write_batch_with_index/write_batch_with_index_internal.cc",
    ],
-    headers = AutoHeaders.RECURSIVE_GLOB,
+    auto_headers = AutoHeaders.RECURSIVE_GLOB,
    arch_preprocessor_flags = rocksdb_arch_preprocessor_flags,
    compiler_flags = rocksdb_compiler_flags,
    preprocessor_flags = rocksdb_preprocessor_flags,
@@ -314,7 +314,7 @@ cpp_library(
        "utilities/col_buf_encoder.cc",
        "utilities/column_aware_encoding_util.cc",
    ],
-    headers = AutoHeaders.RECURSIVE_GLOB,
+    auto_headers = AutoHeaders.RECURSIVE_GLOB,
    arch_preprocessor_flags = rocksdb_arch_preprocessor_flags,
    compiler_flags = rocksdb_compiler_flags,
    preprocessor_flags = rocksdb_preprocessor_flags,
@@ -329,7 +329,7 @@ cpp_library(
        "tools/trace_analyzer_tool.cc",
        "util/testutil.cc",
    ],
-    headers = AutoHeaders.RECURSIVE_GLOB,
+    auto_headers = AutoHeaders.RECURSIVE_GLOB,
    arch_preprocessor_flags = rocksdb_arch_preprocessor_flags,
    compiler_flags = rocksdb_compiler_flags,
    preprocessor_flags = rocksdb_preprocessor_flags,
@@ -340,7 +340,7 @@ cpp_library(
 cpp_library(
    name = "env_basic_test_lib",
    srcs = ["env/env_basic_test.cc"],
-    headers = AutoHeaders.RECURSIVE_GLOB,
+    auto_headers = AutoHeaders.RECURSIVE_GLOB,
    arch_preprocessor_flags = rocksdb_arch_preprocessor_flags,
    compiler_flags = rocksdb_compiler_flags,
    preprocessor_flags = rocksdb_preprocessor_flags,
@@ -10,7 +10,7 @@ def pretty_list(lst, indent=8):

    if len(lst) == 1:
        return "\"%s\"" % lst[0]
-    
+
    separator = "\",\n%s\"" % (" " * indent)
    res = separator.join(sorted(lst))
    res = "\n" + (" " * indent) + "\"" + res + "\",\n" + (" " * (indent - 4))
@@ -31,13 +31,16 @@ class TARGETSBuilder:
        self.targets_file.close()

    def add_library(self, name, srcs, deps=None, headers=None):
+        headers_attr_prefix = ""
        if headers is None:
+            headers_attr_prefix = "auto_"
            headers = "AutoHeaders.RECURSIVE_GLOB"
-        self.targets_file.write(targets_cfg.library_template % (
-            name,
-            pretty_list(srcs),
-            headers,
-            pretty_list(deps)))
+        self.targets_file.write(targets_cfg.library_template.format(
+            name=name,
+            srcs=pretty_list(srcs),
+            headers_attr_prefix=headers_attr_prefix,
+            headers=headers,
+            deps=pretty_list(deps)))
        self.total_lib = self.total_lib + 1

    def add_binary(self, name, srcs, deps=None):
@@ -75,13 +75,13 @@ if is_opt_mode:

 library_template = """
 cpp_library(
-    name = "%s",
-    srcs = [%s],
-    headers = %s,
+    name = "{name}",
+    srcs = [{srcs}],
+    {headers_attr_prefix}headers = {headers},
    arch_preprocessor_flags = rocksdb_arch_preprocessor_flags,
    compiler_flags = rocksdb_compiler_flags,
    preprocessor_flags = rocksdb_preprocessor_flags,
-    deps = [%s],
+    deps = [{deps}],
    external_deps = rocksdb_external_deps,
 )
 """
@@ -53,11 +53,13 @@ if [ -z "$ROCKSDB_NO_FBCODE" -a -d /mnt/gvfs/third-party ]; then
    FBCODE_BUILD="true"
    # If we're compiling with TSAN we need pic build
    PIC_BUILD=$COMPILE_WITH_TSAN
-    if [ -z "$ROCKSDB_FBCODE_BUILD_WITH_481" ]; then
-      source "$PWD/build_tools/fbcode_config.sh"
-    else
+    if [ -n "$ROCKSDB_FBCODE_BUILD_WITH_481" ]; then
      # we need this to build with MySQL. Don't use for other purposes.
      source "$PWD/build_tools/fbcode_config4.8.1.sh"
+    elif [ -n "$ROCKSDB_FBCODE_BUILD_WITH_5xx" ]; then
+      source "$PWD/build_tools/fbcode_config.sh"
+    else
+      source "$PWD/build_tools/fbcode_config_platform007.sh"
    fi
 fi

@@ -0,0 +1,18 @@
+GCC_BASE=/mnt/gvfs/third-party2/gcc/6e8e715624fd15256a7970073387793dfcf79b46/7.x/centos7-native/b2ef2b6
+CLANG_BASE=/mnt/gvfs/third-party2/llvm-fb/ef37e1faa1c29782abfac1ae65a291b9b7966f6d/stable/centos7-native/c9f9104
+LIBGCC_BASE=/mnt/gvfs/third-party2/libgcc/c67031f0f739ac61575a061518d6ef5038f99f90/7.x/platform007/5620abc
+GLIBC_BASE=/mnt/gvfs/third-party2/glibc/60d6f124a78798b73944f5ba87c2306ae3460153/2.26/platform007/f259413
+SNAPPY_BASE=/mnt/gvfs/third-party2/snappy/7f9bdaada18f59bc27ec2b0871eb8a6144343aef/1.1.3/platform007/ca4da3d
+ZLIB_BASE=/mnt/gvfs/third-party2/zlib/22c2d65676fb7c23cfa797c4f6937f38b026f3cf/1.2.8/platform007/ca4da3d
+BZIP2_BASE=/mnt/gvfs/third-party2/bzip2/dc49a21c5fceec6456a7a28a94dcd16690af1337/1.0.6/platform007/ca4da3d
+LZ4_BASE=/mnt/gvfs/third-party2/lz4/907b498203d297947f3bb70b9466f47e100f1873/r131/platform007/ca4da3d
+ZSTD_BASE=/mnt/gvfs/third-party2/zstd/3ee276cbacfad3074e3f07bf826ac47f06970f4e/1.3.5/platform007/15a3614
+GFLAGS_BASE=/mnt/gvfs/third-party2/gflags/0b9929d2588991c65a57168bf88aff2db87c5d48/2.2.0/platform007/ca4da3d
+JEMALLOC_BASE=/mnt/gvfs/third-party2/jemalloc/9c910d36d6235cc40e8ff559358f1833452300ca/master/platform007/5b0f53e
+NUMA_BASE=/mnt/gvfs/third-party2/numa/9cbf2460284c669ed19c3ccb200a71f7dd7e53c7/2.0.11/platform007/ca4da3d
+LIBUNWIND_BASE=/mnt/gvfs/third-party2/libunwind/bf3d7497fe4e6d007354f0adffa16ce3003f8338/1.3/platform007/6f3e0a9
+TBB_BASE=/mnt/gvfs/third-party2/tbb/ff4e0b093534704d8abab678a4fd7f5ea7b094c7/2018_U5/platform007/ca4da3d
+KERNEL_HEADERS_BASE=/mnt/gvfs/third-party2/kernel-headers/b5c4a61a5c483ba24722005ae07895971a2ac707/fb/platform007/da39a3e
+BINUTILS_BASE=/mnt/gvfs/third-party2/binutils/92ff90349e2f43ea0a8246d8b1cf17b6869013e3/2.29.1/centos7-native/da39a3e
+VALGRIND_BASE=/mnt/gvfs/third-party2/valgrind/f3f697a28122e6bcd513273dd9c1ff23852fc59f/3.13.0/platform007/ca4da3d
+LUA_BASE=/mnt/gvfs/third-party2/lua/f0cd714433206d5139df61659eb7b28b1dea6683/5.3.4/platform007/5007832
@@ -0,0 +1,157 @@
+#!/bin/sh
+#
+# Set environment variables so that we can compile rocksdb using
+# fbcode settings.  It uses the latest g++ and clang compilers and also
+# uses jemalloc
+# Environment variables that change the behavior of this script:
+# PIC_BUILD -- if true, it will only take pic versions of libraries from fbcode. libraries that don't have pic variant will not be included
+
+
+BASEDIR=`dirname $BASH_SOURCE`
+source "$BASEDIR/dependencies_platform007.sh"
+
+CFLAGS=""
+
+# libgcc
+LIBGCC_INCLUDE="$LIBGCC_BASE/include/c++/7.3.0"
+LIBGCC_LIBS=" -L $LIBGCC_BASE/lib"
+
+# glibc
+GLIBC_INCLUDE="$GLIBC_BASE/include"
+GLIBC_LIBS=" -L $GLIBC_BASE/lib"
+
+# snappy
+SNAPPY_INCLUDE=" -I $SNAPPY_BASE/include/"
+if test -z $PIC_BUILD; then
+  SNAPPY_LIBS=" $SNAPPY_BASE/lib/libsnappy.a"
+else
+  SNAPPY_LIBS=" $SNAPPY_BASE/lib/libsnappy_pic.a"
+fi
+CFLAGS+=" -DSNAPPY"
+
+if test -z $PIC_BUILD; then
+  # location of zlib headers and libraries
+  ZLIB_INCLUDE=" -I $ZLIB_BASE/include/"
+  ZLIB_LIBS=" $ZLIB_BASE/lib/libz.a"
+  CFLAGS+=" -DZLIB"
+
+  # location of bzip headers and libraries
+  BZIP_INCLUDE=" -I $BZIP2_BASE/include/"
+  BZIP_LIBS=" $BZIP2_BASE/lib/libbz2.a"
+  CFLAGS+=" -DBZIP2"
+
+  LZ4_INCLUDE=" -I $LZ4_BASE/include/"
+  LZ4_LIBS=" $LZ4_BASE/lib/liblz4.a"
+  CFLAGS+=" -DLZ4"
+fi
+
+ZSTD_INCLUDE=" -I $ZSTD_BASE/include/"
+if test -z $PIC_BUILD; then
+  ZSTD_LIBS=" $ZSTD_BASE/lib/libzstd.a"
+else
+  ZSTD_LIBS=" $ZSTD_BASE/lib/libzstd_pic.a"
+fi
+CFLAGS+=" -DZSTD"
+
+# location of gflags headers and libraries
+GFLAGS_INCLUDE=" -I $GFLAGS_BASE/include/"
+if test -z $PIC_BUILD; then
+  GFLAGS_LIBS=" $GFLAGS_BASE/lib/libgflags.a"
+else
+  GFLAGS_LIBS=" $GFLAGS_BASE/lib/libgflags_pic.a"
+fi
+CFLAGS+=" -DGFLAGS=gflags"
+
+# location of jemalloc
+JEMALLOC_INCLUDE=" -I $JEMALLOC_BASE/include/"
+JEMALLOC_LIB=" $JEMALLOC_BASE/lib/libjemalloc.a"
+
+if test -z $PIC_BUILD; then
+  # location of numa
+  NUMA_INCLUDE=" -I $NUMA_BASE/include/"
+  NUMA_LIB=" $NUMA_BASE/lib/libnuma.a"
+  CFLAGS+=" -DNUMA"
+
+  # location of libunwind
+  LIBUNWIND="$LIBUNWIND_BASE/lib/libunwind.a"
+fi
+
+# location of TBB
+TBB_INCLUDE=" -isystem $TBB_BASE/include/"
+if test -z $PIC_BUILD; then
+  TBB_LIBS="$TBB_BASE/lib/libtbb.a"
+else
+  TBB_LIBS="$TBB_BASE/lib/libtbb_pic.a"
+fi
+CFLAGS+=" -DTBB"
+
+# use Intel SSE support for checksum calculations
+export USE_SSE=1
+export PORTABLE=1
+
+BINUTILS="$BINUTILS_BASE/bin"
+AR="$BINUTILS/ar"
+
+DEPS_INCLUDE="$SNAPPY_INCLUDE $ZLIB_INCLUDE $BZIP_INCLUDE $LZ4_INCLUDE $ZSTD_INCLUDE $GFLAGS_INCLUDE $NUMA_INCLUDE $TBB_INCLUDE"
+
+STDLIBS="-L $GCC_BASE/lib64"
+
+CLANG_BIN="$CLANG_BASE/bin"
+CLANG_LIB="$CLANG_BASE/lib"
+CLANG_SRC="$CLANG_BASE/../../src"
+
+CLANG_ANALYZER="$CLANG_BIN/clang++"
+CLANG_SCAN_BUILD="$CLANG_SRC/llvm/tools/clang/tools/scan-build/bin/scan-build"
+
+if [ -z "$USE_CLANG" ]; then
+  # gcc
+  CC="$GCC_BASE/bin/gcc"
+  CXX="$GCC_BASE/bin/g++"
+
+  CFLAGS+=" -B$BINUTILS/gold"
+  CFLAGS+=" -isystem $LIBGCC_INCLUDE"
+  CFLAGS+=" -isystem $GLIBC_INCLUDE"
+  JEMALLOC=1
+else
+  # clang
+  CLANG_INCLUDE="$CLANG_LIB/clang/stable/include"
+  CC="$CLANG_BIN/clang"
+  CXX="$CLANG_BIN/clang++"
+
+  KERNEL_HEADERS_INCLUDE="$KERNEL_HEADERS_BASE/include"
+
+  CFLAGS+=" -B$BINUTILS/gold -nostdinc -nostdlib"
+  CFLAGS+=" -isystem $LIBGCC_BASE/include/c++/7.x "
+  CFLAGS+=" -isystem $LIBGCC_BASE/include/c++/7.x/x86_64-facebook-linux "
+  CFLAGS+=" -isystem $GLIBC_INCLUDE"
+  CFLAGS+=" -isystem $LIBGCC_INCLUDE"
+  CFLAGS+=" -isystem $CLANG_INCLUDE"
+  CFLAGS+=" -isystem $KERNEL_HEADERS_INCLUDE/linux "
+  CFLAGS+=" -isystem $KERNEL_HEADERS_INCLUDE "
+  CFLAGS+=" -Wno-expansion-to-defined "
+  CXXFLAGS="-nostdinc++"
+fi
+
+CFLAGS+=" $DEPS_INCLUDE"
+CFLAGS+=" -DROCKSDB_PLATFORM_POSIX -DROCKSDB_LIB_IO_POSIX -DROCKSDB_FALLOCATE_PRESENT -DROCKSDB_MALLOC_USABLE_SIZE -DROCKSDB_RANGESYNC_PRESENT -DROCKSDB_SCHED_GETCPU_PRESENT -DROCKSDB_SUPPORT_THREAD_LOCAL -DHAVE_SSE42"
+CXXFLAGS+=" $CFLAGS"
+
+EXEC_LDFLAGS=" $SNAPPY_LIBS $ZLIB_LIBS $BZIP_LIBS $LZ4_LIBS $ZSTD_LIBS $GFLAGS_LIBS $NUMA_LIB $TBB_LIBS"
+EXEC_LDFLAGS+=" -B$BINUTILS/gold"
+EXEC_LDFLAGS+=" -Wl,--dynamic-linker,/usr/local/fbcode/platform007/lib/ld.so"
+EXEC_LDFLAGS+=" $LIBUNWIND"
+EXEC_LDFLAGS+=" -Wl,-rpath=/usr/local/fbcode/platform007/lib"
+# required by libtbb
+EXEC_LDFLAGS+=" -ldl"
+
+PLATFORM_LDFLAGS="$LIBGCC_LIBS $GLIBC_LIBS $STDLIBS -lgcc -lstdc++"
+
+EXEC_LDFLAGS_SHARED="$SNAPPY_LIBS $ZLIB_LIBS $BZIP_LIBS $LZ4_LIBS $ZSTD_LIBS $GFLAGS_LIBS $TBB_LIBS"
+
+VALGRIND_VER="$VALGRIND_BASE/bin/"
+
+# lua not supported because it's on track for deprecation, I think
+LUA_PATH=
+LUA_LIB=
+
+export CC CXX AR CFLAGS CXXFLAGS EXEC_LDFLAGS EXEC_LDFLAGS_SHARED VALGRIND_VER JEMALLOC_LIB JEMALLOC_INCLUDE CLANG_ANALYZER CLANG_SCAN_BUILD LUA_PATH LUA_LIB
@@ -85,8 +85,9 @@ NON_SHM="TMPD=/tmp/rocksdb_test_tmp"
 GCC_481="ROCKSDB_FBCODE_BUILD_WITH_481=1"
 ASAN="COMPILE_WITH_ASAN=1"
 CLANG="USE_CLANG=1"
-LITE="OPT=\"-DROCKSDB_LITE -g\""
-TSAN="COMPILE_WITH_TSAN=1"
+# in gcc-5 there are known problems with TSAN like https://gcc.gnu.org/bugzilla/show_bug.cgi?id=71090.
+# using platform007 gives us gcc-8 or higher which has that bug fixed.
+TSAN="ROCKSDB_FBCODE_BUILD_WITH_PLATFORM007=1 COMPILE_WITH_TSAN=1"
 UBSAN="COMPILE_WITH_UBSAN=1"
 TSAN_CRASH='CRASH_TEST_EXT_ARGS="--compression_type=zstd --log2_keys_per_lock=22"'
 NON_TSAN_CRASH="CRASH_TEST_EXT_ARGS=--compression_type=zstd"
@@ -53,6 +53,45 @@ function get_lib_base()
  log_variable $__res_var
 }

+###########################################################
+#                platform007 dependencies                 #
+###########################################################
+
+OUTPUT="$BASEDIR/dependencies_platform007.sh"
+
+rm -f "$OUTPUT"
+touch "$OUTPUT"
+
+echo "Writing dependencies to $OUTPUT"
+
+# Compilers locations
+GCC_BASE=`readlink -f $TP2_LATEST/gcc/7.x/centos7-native/*/`
+CLANG_BASE=`readlink -f $TP2_LATEST/llvm-fb/stable/centos7-native/*/`
+
+log_variable GCC_BASE
+log_variable CLANG_BASE
+
+# Libraries locations
+get_lib_base libgcc     7.x     platform007
+get_lib_base glibc      2.26    platform007
+get_lib_base snappy     LATEST  platform007
+get_lib_base zlib       LATEST  platform007
+get_lib_base bzip2      LATEST  platform007
+get_lib_base lz4        LATEST  platform007
+get_lib_base zstd       LATEST  platform007
+get_lib_base gflags     LATEST  platform007
+get_lib_base jemalloc   LATEST  platform007
+get_lib_base numa       LATEST  platform007
+get_lib_base libunwind  LATEST  platform007
+get_lib_base tbb        LATEST  platform007
+
+get_lib_base kernel-headers fb platform007
+get_lib_base binutils   LATEST centos7-native
+get_lib_base valgrind   LATEST platform007
+get_lib_base lua        5.3.4  platform007
+
+git diff $OUTPUT
+
 ###########################################################
 #                   5.x dependencies                      #
 ###########################################################
@@ -461,10 +461,8 @@ std::string LRUCacheShard::GetPrintableOptions() const {
 }

 LRUCache::LRUCache(size_t capacity, int num_shard_bits,
-                   bool strict_capacity_limit, double high_pri_pool_ratio,
-                   std::shared_ptr<CacheAllocator> allocator)
-    : ShardedCache(capacity, num_shard_bits, strict_capacity_limit,
-                   std::move(allocator)) {
+                   bool strict_capacity_limit, double high_pri_pool_ratio)
+    : ShardedCache(capacity, num_shard_bits, strict_capacity_limit) {
  num_shards_ = 1 << num_shard_bits;
  shards_ = reinterpret_cast<LRUCacheShard*>(
      port::cacheline_aligned_alloc(sizeof(LRUCacheShard) * num_shards_));
@@ -539,14 +537,12 @@ double LRUCache::GetHighPriPoolRatio() {
 std::shared_ptr<Cache> NewLRUCache(const LRUCacheOptions& cache_opts) {
  return NewLRUCache(cache_opts.capacity, cache_opts.num_shard_bits,
                     cache_opts.strict_capacity_limit,
-                     cache_opts.high_pri_pool_ratio,
-                     cache_opts.cache_allocator);
+                     cache_opts.high_pri_pool_ratio);
 }

-std::shared_ptr<Cache> NewLRUCache(
-    size_t capacity, int num_shard_bits, bool strict_capacity_limit,
-    double high_pri_pool_ratio,
-    std::shared_ptr<CacheAllocator> cache_allocator) {
+std::shared_ptr<Cache> NewLRUCache(size_t capacity, int num_shard_bits,
+                                   bool strict_capacity_limit,
+                                   double high_pri_pool_ratio) {
  if (num_shard_bits >= 20) {
    return nullptr;  // the cache cannot be sharded into too many fine pieces
  }
@@ -558,8 +554,7 @@ std::shared_ptr<Cache> NewLRUCache(
    num_shard_bits = GetDefaultCacheShardBits(capacity);
  }
  return std::make_shared<LRUCache>(capacity, num_shard_bits,
-                                    strict_capacity_limit, high_pri_pool_ratio,
-                                    std::move(cache_allocator));
+                                    strict_capacity_limit, high_pri_pool_ratio);
 }

 }  // namespace rocksdb
@@ -279,8 +279,7 @@ class ALIGN_AS(CACHE_LINE_SIZE) LRUCacheShard : public CacheShard {
 class LRUCache : public ShardedCache {
 public:
  LRUCache(size_t capacity, int num_shard_bits, bool strict_capacity_limit,
-           double high_pri_pool_ratio,
-           std::shared_ptr<CacheAllocator> cache_allocator = nullptr);
+           double high_pri_pool_ratio);
  virtual ~LRUCache();
  virtual const char* Name() const override { return "LRUCache"; }
  virtual CacheShard* GetShard(int shard) override;
@@ -20,10 +20,8 @@
 namespace rocksdb {

 ShardedCache::ShardedCache(size_t capacity, int num_shard_bits,
-                           bool strict_capacity_limit,
-                           std::shared_ptr<CacheAllocator> allocator)
-    : Cache(std::move(allocator)),
-      num_shard_bits_(num_shard_bits),
+                           bool strict_capacity_limit)
+    : num_shard_bits_(num_shard_bits),
      capacity_(capacity),
      strict_capacity_limit_(strict_capacity_limit),
      last_id_(1) {}
@@ -144,9 +142,6 @@ std::string ShardedCache::GetPrintableOptions() const {
             strict_capacity_limit_);
    ret.append(buffer);
  }
-  snprintf(buffer, kBufferSize, "    cache_allocator : %s\n",
-           cache_allocator() ? cache_allocator()->Name() : "None");
-  ret.append(buffer);
  ret.append(GetShard(0)->GetPrintableOptions());
  return ret;
 }
@@ -47,8 +47,7 @@ class CacheShard {
 // Keys are sharded by the highest num_shard_bits bits of hash value.
 class ShardedCache : public Cache {
 public:
-  ShardedCache(size_t capacity, int num_shard_bits, bool strict_capacity_limit,
-               std::shared_ptr<CacheAllocator> cache_allocator = nullptr);
+  ShardedCache(size_t capacity, int num_shard_bits, bool strict_capacity_limit);
  virtual ~ShardedCache() = default;
  virtual const char* Name() const override = 0;
  virtual CacheShard* GetShard(int shard) = 0;
@@ -77,6 +77,12 @@ CompactionIterator::CompactionIterator(
    earliest_snapshot_ = snapshots_->at(0);
    latest_snapshot_ = snapshots_->back();
  }
+#ifndef NDEBUG
+  // findEarliestVisibleSnapshot assumes this ordering.
+  for (size_t i = 1; i < snapshots_->size(); ++i) {
+    assert(snapshots_->at(i - 1) <= snapshots_->at(i));
+  }
+#endif
  if (compaction_filter_ != nullptr) {
    if (compaction_filter_->IgnoreSnapshots()) {
      ignore_snapshots_ = true;
@@ -628,18 +634,23 @@ void CompactionIterator::PrepareOutput() {
 inline SequenceNumber CompactionIterator::findEarliestVisibleSnapshot(
    SequenceNumber in, SequenceNumber* prev_snapshot) {
  assert(snapshots_->size());
-  SequenceNumber prev = kMaxSequenceNumber;
-  for (const auto cur : *snapshots_) {
-    assert(prev == kMaxSequenceNumber || prev <= cur);
-    if (cur >= in && (snapshot_checker_ == nullptr ||
-                      snapshot_checker_->IsInSnapshot(in, cur))) {
-      *prev_snapshot = prev == kMaxSequenceNumber ? 0 : prev;
+  auto snapshots_iter = std::lower_bound(
+      snapshots_->begin(), snapshots_->end(), in);
+  if (snapshots_iter == snapshots_->begin()) {
+    *prev_snapshot = 0;
+  } else {
+    *prev_snapshot = *std::prev(snapshots_iter);
+    assert(*prev_snapshot < in);
+  }
+  for (; snapshots_iter != snapshots_->end(); ++snapshots_iter) {
+    auto cur = *snapshots_iter;
+    assert(in <= cur);
+    if (snapshot_checker_ == nullptr ||
+        snapshot_checker_->IsInSnapshot(in, cur)) {
      return cur;
    }
-    prev = cur;
-    assert(prev < kMaxSequenceNumber);
+    *prev_snapshot = cur;
  }
-  *prev_snapshot = prev;
  return kMaxSequenceNumber;
 }

@@ -219,7 +219,8 @@ void CompactionPicker::GetRange(const std::vector<CompactionInputFiles>& inputs,

 bool CompactionPicker::ExpandInputsToCleanCut(const std::string& /*cf_name*/,
                                              VersionStorageInfo* vstorage,
-                                              CompactionInputFiles* inputs) {
+                                              CompactionInputFiles* inputs,
+                                              InternalKey** next_smallest) {
  // This isn't good compaction
  assert(!inputs->empty());

@@ -242,7 +243,8 @@ bool CompactionPicker::ExpandInputsToCleanCut(const std::string& /*cf_name*/,
    GetRange(*inputs, &smallest, &largest);
    inputs->clear();
    vstorage->GetOverlappingInputs(level, &smallest, &largest, &inputs->files,
-                                   hint_index, &hint_index);
+                                   hint_index, &hint_index, true,
+                                   next_smallest);
  } while (inputs->size() > old_size);

  // we started off with inputs non-empty and the previous loop only grew
@@ -649,7 +651,6 @@ Compaction* CompactionPicker::CompactRange(
      uint64_t s = inputs[i]->compensated_file_size;
      total += s;
      if (total >= limit) {
-        **compaction_end = inputs[i + 1]->smallest;
        covering_the_whole_range = false;
        inputs.files.resize(i + 1);
        break;
@@ -658,7 +659,10 @@ Compaction* CompactionPicker::CompactRange(
  }
  assert(output_path_id < static_cast<uint32_t>(ioptions_.cf_paths.size()));

-  if (ExpandInputsToCleanCut(cf_name, vstorage, &inputs) == false) {
+  InternalKey key_storage;
+  InternalKey* next_smallest = &key_storage;
+  if (ExpandInputsToCleanCut(cf_name, vstorage, &inputs, &next_smallest) ==
+      false) {
    // manual compaction is now multi-threaded, so it can
    // happen that ExpandWhileOverlapping fails
    // we handle it higher in RunManualCompaction
@@ -666,8 +670,10 @@ Compaction* CompactionPicker::CompactRange(
    return nullptr;
  }

-  if (covering_the_whole_range) {
+  if (covering_the_whole_range || !next_smallest) {
    *compaction_end = nullptr;
+  } else {
+    **compaction_end = *next_smallest;
  }

  CompactionInputFiles output_level_inputs;
@@ -151,7 +151,8 @@ class CompactionPicker {
  // Will return false if it is impossible to apply this compaction.
  bool ExpandInputsToCleanCut(const std::string& cf_name,
                              VersionStorageInfo* vstorage,
-                              CompactionInputFiles* inputs);
+                              CompactionInputFiles* inputs,
+                              InternalKey** next_smallest = nullptr);

  // Returns true if any one of the parent files are being compacted
  bool IsRangeInCompaction(VersionStorageInfo* vstorage,
@@ -3961,6 +3961,127 @@ INSTANTIATE_TEST_CASE_P(
                      CompactionPri::kOldestSmallestSeqFirst,
                      CompactionPri::kMinOverlappingRatio));

+class NoopMergeOperator : public MergeOperator {
+ public:
+  NoopMergeOperator() {}
+
+  virtual bool FullMergeV2(const MergeOperationInput& /*merge_in*/,
+                           MergeOperationOutput* merge_out) const override {
+    std::string val("bar");
+    merge_out->new_value = val;
+    return true;
+  }
+
+  virtual const char* Name() const override { return "Noop"; }
+};
+
+TEST_F(DBCompactionTest, PartialManualCompaction) {
+  Options opts = CurrentOptions();
+  opts.num_levels = 3;
+  opts.level0_file_num_compaction_trigger = 10;
+  opts.compression = kNoCompression;
+  opts.merge_operator.reset(new NoopMergeOperator());
+  opts.target_file_size_base = 10240;
+  DestroyAndReopen(opts);
+
+  Random rnd(301);
+  for (auto i = 0; i < 8; ++i) {
+    for (auto j = 0; j < 10; ++j) {
+      Merge("foo", RandomString(&rnd, 1024));
+    }
+    Flush();
+  }
+
+  MoveFilesToLevel(2);
+
+  std::string prop;
+  EXPECT_TRUE(dbfull()->GetProperty(DB::Properties::kLiveSstFilesSize, &prop));
+  uint64_t max_compaction_bytes = atoi(prop.c_str()) / 2;
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"max_compaction_bytes", std::to_string(max_compaction_bytes)}}));
+
+  CompactRangeOptions cro;
+  cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+  dbfull()->CompactRange(cro, nullptr, nullptr);
+}
+
+// FixFileIngestionCompactionDeadlock tests and verifies that compaction and
+// file ingestion do not cause deadlock in the event of write stall triggered
+// by number of L0 files reaching level0_stop_writes_trigger.
+TEST_P(DBCompactionTestWithParam, FixFileIngestionCompactionDeadlock) {
+  const int kNumKeysPerFile = 100;
+  // Generate SST files.
+  Options options = CurrentOptions();
+
+  // Generate an external SST file containing a single key, i.e. 99
+  std::string sst_files_dir = dbname_ + "/sst_files/";
+  test::DestroyDir(env_, sst_files_dir);
+  ASSERT_OK(env_->CreateDir(sst_files_dir));
+  SstFileWriter sst_writer(EnvOptions(), options);
+  const std::string sst_file_path = sst_files_dir + "test.sst";
+  ASSERT_OK(sst_writer.Open(sst_file_path));
+  ASSERT_OK(sst_writer.Put(Key(kNumKeysPerFile - 1), "value"));
+  ASSERT_OK(sst_writer.Finish());
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->LoadDependency({
+      {"DBImpl::IngestExternalFile:AfterIncIngestFileCounter",
+       "BackgroundCallCompaction:0"},
+  });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  options.write_buffer_size = 110 << 10;  // 110KB
+  options.level0_file_num_compaction_trigger =
+      options.level0_stop_writes_trigger;
+  options.max_subcompactions = max_subcompactions_;
+  options.memtable_factory.reset(new SpecialSkipListFactory(kNumKeysPerFile));
+  DestroyAndReopen(options);
+  Random rnd(301);
+
+  // Generate level0_stop_writes_trigger L0 files to trigger write stop
+  for (int i = 0; i != options.level0_file_num_compaction_trigger; ++i) {
+    for (int j = 0; j != kNumKeysPerFile; ++j) {
+      ASSERT_OK(Put(Key(j), RandomString(&rnd, 990)));
+    }
+    if (0 == i) {
+      // When we reach here, the memtables have kNumKeysPerFile keys. Note that
+      // flush is not yet triggered. We need to write an extra key so that the
+      // write path will call PreprocessWrite and flush the previous key-value
+      // pairs to e flushed. After that, there will be the newest key in the
+      // memtable, and a bunch of L0 files. Since there is already one key in
+      // the memtable, then for i = 1, 2, ..., we do not have to write this
+      // extra key to trigger flush.
+      ASSERT_OK(Put("", ""));
+    }
+    dbfull()->TEST_WaitForFlushMemTable();
+    ASSERT_EQ(NumTableFilesAtLevel(0 /*level*/, 0 /*cf*/), i + 1);
+  }
+  // When we reach this point, there will be level0_stop_writes_trigger L0
+  // files and one extra key (99) in memory, which overlaps with the external
+  // SST file. Write stall triggers, and can be cleared only after compaction
+  // reduces the number of L0 files.
+
+  // Compaction will also be triggered since we have reached the threshold for
+  // auto compaction. Note that compaction may begin after the following file
+  // ingestion thread and waits for ingestion to finish.
+
+  // Thread to ingest file with overlapping key range with the current
+  // memtable. Consequently ingestion will trigger a flush. The flush MUST
+  // proceed without waiting for the write stall condition to clear, otherwise
+  // deadlock can happen.
+  port::Thread ingestion_thr([&]() {
+    IngestExternalFileOptions ifo;
+    Status s = db_->IngestExternalFile({sst_file_path}, ifo);
+    ASSERT_OK(s);
+  });
+
+  // More write to trigger write stop
+  ingestion_thr.join();
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  Close();
+}
+
 #endif // !defined(ROCKSDB_LITE)
 }  // namespace rocksdb

@@ -3059,6 +3059,7 @@ Status DBImpl::IngestExternalFile(
    }

    num_running_ingest_file_++;
+    TEST_SYNC_POINT("DBImpl::IngestExternalFile:AfterIncIngestFileCounter");

    // We cannot ingest a file into a dropped CF
    if (cfd->IsDropped()) {
@@ -3074,7 +3075,9 @@ Status DBImpl::IngestExternalFile(
                               &need_flush);
      if (status.ok() && need_flush) {
        mutex_.Unlock();
-        status = FlushMemTable(cfd, FlushOptions(),
+        FlushOptions flush_opts;
+        flush_opts.allow_write_stall = true;
+        status = FlushMemTable(cfd, flush_opts,
                               FlushReason::kExternalFileIngestion,
                               true /* writes_stopped */);
        mutex_.Lock();
@@ -813,6 +813,7 @@ class DBImpl : public DB {
  friend struct SuperVersion;
  friend class CompactedDBImpl;
  friend class DBTest_ConcurrentFlushWAL_Test;
+  friend class DBTest_MixedSlowdownOptionsStop_Test;
 #ifndef NDEBUG
  friend class DBTest2_ReadCallbackTest_Test;
  friend class WriteCallbackTest_WriteWithCallbackTest_Test;
@@ -1162,10 +1162,14 @@ Status DBImpl::DelayWrite(uint64_t num_bytes,
    uint64_t delay = write_controller_.GetDelay(env_, num_bytes);
    if (delay > 0) {
      if (write_options.no_slowdown) {
-        return Status::Incomplete();
+        return Status::Incomplete("Write stall");
      }
      TEST_SYNC_POINT("DBImpl::DelayWrite:Sleep");

+      // Notify write_thread_ about the stall so it can setup a barrier and
+      // fail any pending writers with no_slowdown
+      write_thread_.BeginWriteStall();
+      TEST_SYNC_POINT("DBImpl::DelayWrite:BeginWriteStallDone");
      mutex_.Unlock();
      // We will delay the write until we have slept for delay ms or
      // we don't need a delay anymore
@@ -1182,6 +1186,7 @@ Status DBImpl::DelayWrite(uint64_t num_bytes,
        env_->SleepForMicroseconds(kDelayInterval);
      }
      mutex_.Lock();
+      write_thread_.EndWriteStall();
    }

    // Don't wait if there's a background error, even if its a soft error. We
@@ -1190,11 +1195,16 @@ Status DBImpl::DelayWrite(uint64_t num_bytes,
    // indefinitely
    while (error_handler_.GetBGError().ok() && write_controller_.IsStopped()) {
      if (write_options.no_slowdown) {
-        return Status::Incomplete();
+        return Status::Incomplete("Write stall");
      }
      delayed = true;
+
+      // Notify write_thread_ about the stall so it can setup a barrier and
+      // fail any pending writers with no_slowdown
+      write_thread_.BeginWriteStall();
      TEST_SYNC_POINT("DBImpl::DelayWrite:Wait");
      bg_cv_.Wait();
+      write_thread_.EndWriteStall();
    }
  }
  assert(!delayed || !write_options.no_slowdown);
@@ -262,6 +262,196 @@ TEST_F(DBTest, SkipDelay) {
  }
 }

+TEST_F(DBTest, MixedSlowdownOptions) {
+  Options options = CurrentOptions();
+  options.env = env_;
+  options.write_buffer_size = 100000;
+  CreateAndReopenWithCF({"pikachu"}, options);
+  std::vector<port::Thread> threads;
+  std::atomic<int> thread_num(0);
+
+  std::function<void()> write_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = false;
+    ASSERT_OK(dbfull()->Put(wo, key, "bar"));
+  };
+  std::function<void()> write_no_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = true;
+    ASSERT_NOK(dbfull()->Put(wo, key, "bar"));
+  };
+  // Use a small number to ensure a large delay that is still effective
+  // when we do Put
+  // TODO(myabandeh): this is time dependent and could potentially make
+  // the test flaky
+  auto token = dbfull()->TEST_write_controler().GetDelayToken(1);
+  std::atomic<int> sleep_count(0);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::DelayWrite:BeginWriteStallDone",
+      [&](void* /*arg*/) {
+        sleep_count.fetch_add(1);
+        if (threads.empty()) {
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_slowdown_func);
+          }
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_no_slowdown_func);
+          }
+        }
+      });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  WriteOptions wo;
+  wo.sync = false;
+  wo.disableWAL = false;
+  wo.no_slowdown = false;
+  dbfull()->Put(wo, "foo", "bar");
+  // We need the 2nd write to trigger delay. This is because delay is
+  // estimated based on the last write size which is 0 for the first write.
+  ASSERT_OK(dbfull()->Put(wo, "foo2", "bar2"));
+          token.reset();
+
+  for (auto& t : threads) {
+    t.join();
+  }
+  ASSERT_GE(sleep_count.load(), 1);
+
+  wo.no_slowdown = true;
+  ASSERT_OK(dbfull()->Put(wo, "foo3", "bar"));
+}
+
+TEST_F(DBTest, MixedSlowdownOptionsInQueue) {
+  Options options = CurrentOptions();
+  options.env = env_;
+  options.write_buffer_size = 100000;
+  CreateAndReopenWithCF({"pikachu"}, options);
+  std::vector<port::Thread> threads;
+  std::atomic<int> thread_num(0);
+
+  std::function<void()> write_no_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = true;
+    ASSERT_NOK(dbfull()->Put(wo, key, "bar"));
+  };
+  // Use a small number to ensure a large delay that is still effective
+  // when we do Put
+  // TODO(myabandeh): this is time dependent and could potentially make
+  // the test flaky
+  auto token = dbfull()->TEST_write_controler().GetDelayToken(1);
+  std::atomic<int> sleep_count(0);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::DelayWrite:Sleep",
+      [&](void* /*arg*/) {
+        sleep_count.fetch_add(1);
+        if (threads.empty()) {
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_no_slowdown_func);
+          }
+          // Sleep for 2s to allow the threads to insert themselves into the
+          // write queue
+          env_->SleepForMicroseconds(3000000ULL);
+        }
+      });
+  std::atomic<int> wait_count(0);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::DelayWrite:Wait",
+      [&](void* /*arg*/) { wait_count.fetch_add(1); });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  WriteOptions wo;
+  wo.sync = false;
+  wo.disableWAL = false;
+  wo.no_slowdown = false;
+  dbfull()->Put(wo, "foo", "bar");
+  // We need the 2nd write to trigger delay. This is because delay is
+  // estimated based on the last write size which is 0 for the first write.
+  ASSERT_OK(dbfull()->Put(wo, "foo2", "bar2"));
+          token.reset();
+
+  for (auto& t : threads) {
+    t.join();
+  }
+  ASSERT_EQ(sleep_count.load(), 1);
+  ASSERT_GE(wait_count.load(), 0);
+}
+
+TEST_F(DBTest, MixedSlowdownOptionsStop) {
+  Options options = CurrentOptions();
+  options.env = env_;
+  options.write_buffer_size = 100000;
+  CreateAndReopenWithCF({"pikachu"}, options);
+  std::vector<port::Thread> threads;
+  std::atomic<int> thread_num(0);
+
+  std::function<void()> write_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = false;
+    ASSERT_OK(dbfull()->Put(wo, key, "bar"));
+  };
+  std::function<void()> write_no_slowdown_func = [&]() {
+    int a = thread_num.fetch_add(1);
+    std::string key = "foo" + std::to_string(a);
+    WriteOptions wo;
+    wo.no_slowdown = true;
+    ASSERT_NOK(dbfull()->Put(wo, key, "bar"));
+  };
+  std::function<void()> wakeup_writer = [&]() {
+    dbfull()->mutex_.Lock();
+    dbfull()->bg_cv_.SignalAll();
+    dbfull()->mutex_.Unlock();
+  };
+  // Use a small number to ensure a large delay that is still effective
+  // when we do Put
+  // TODO(myabandeh): this is time dependent and could potentially make
+  // the test flaky
+  auto token = dbfull()->TEST_write_controler().GetStopToken();
+  std::atomic<int> wait_count(0);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::DelayWrite:Wait",
+      [&](void* /*arg*/) {
+        wait_count.fetch_add(1);
+        if (threads.empty()) {
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_slowdown_func);
+          }
+          for (int i = 0; i < 2; ++i) {
+            threads.emplace_back(write_no_slowdown_func);
+          }
+          // Sleep for 2s to allow the threads to insert themselves into the
+          // write queue
+          env_->SleepForMicroseconds(3000000ULL);
+        }
+        token.reset();
+        threads.emplace_back(wakeup_writer);
+      });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  WriteOptions wo;
+  wo.sync = false;
+  wo.disableWAL = false;
+  wo.no_slowdown = false;
+  dbfull()->Put(wo, "foo", "bar");
+  // We need the 2nd write to trigger delay. This is because delay is
+  // estimated based on the last write size which is 0 for the first write.
+  ASSERT_OK(dbfull()->Put(wo, "foo2", "bar2"));
+          token.reset();
+
+  for (auto& t : threads) {
+    t.join();
+  }
+  ASSERT_GE(wait_count.load(), 1);
+
+  wo.no_slowdown = true;
+  ASSERT_OK(dbfull()->Put(wo, "foo3", "bar"));
+}
 #ifndef ROCKSDB_LITE

 TEST_F(DBTest, LevelLimitReopen) {
@@ -18,9 +18,11 @@ namespace rocksdb {
 #ifdef ROCKSDB_JEMALLOC
 #ifdef __FreeBSD__
 #include <malloc_np.h>
-#define je_malloc_stats_print malloc_stats_print
 #else
 #include "jemalloc/jemalloc.h"
+#ifdef JEMALLOC_NO_RENAME
+#define malloc_stats_print je_malloc_stats_print
+#endif
 #endif

 typedef struct {
@@ -48,7 +50,7 @@ void DumpMallocStats(std::string* stats) {
  std::unique_ptr<char[]> buf{new char[kMallocStatusLen + 1]};
  mstat.cur = buf.get();
  mstat.end = buf.get() + kMallocStatusLen;
-  je_malloc_stats_print(GetJemallocStatus, &mstat, "");
+  malloc_stats_print(GetJemallocStatus, &mstat, "");
  stats->append(buf.get());
 }
 #else
@@ -2028,7 +2028,7 @@ bool VersionStorageInfo::OverlapInLevel(int level,
 void VersionStorageInfo::GetOverlappingInputs(
    int level, const InternalKey* begin, const InternalKey* end,
    std::vector<FileMetaData*>* inputs, int hint_index, int* file_index,
-    bool expand_range) const {
+    bool expand_range, InternalKey** next_smallest) const {
  if (level >= num_non_empty_levels_) {
    // this level is empty, no overlapping inputs
    return;
@@ -2040,11 +2040,17 @@ void VersionStorageInfo::GetOverlappingInputs(
  }
  const Comparator* user_cmp = user_comparator_;
  if (level > 0) {
-    GetOverlappingInputsRangeBinarySearch(level, begin, end, inputs,
-                                          hint_index, file_index);
+    GetOverlappingInputsRangeBinarySearch(level, begin, end, inputs, hint_index,
+                                          file_index, false, next_smallest);
    return;
  }

+  if (next_smallest) {
+    // next_smallest key only makes sense for non-level 0, where files are
+    // non-overlapping
+    *next_smallest = nullptr;
+  }
+
  Slice user_begin, user_end;
  if (begin != nullptr) {
    user_begin = begin->user_key();
@@ -2198,7 +2204,7 @@ int sstableKeyCompare(const Comparator* user_cmp,
 void VersionStorageInfo::GetOverlappingInputsRangeBinarySearch(
    int level, const InternalKey* begin, const InternalKey* end,
    std::vector<FileMetaData*>* inputs, int hint_index, int* file_index,
-    bool within_interval) const {
+    bool within_interval, InternalKey** next_smallest) const {
  assert(level > 0);
  int min = 0;
  int mid = 0;
@@ -2234,6 +2240,9 @@ void VersionStorageInfo::GetOverlappingInputsRangeBinarySearch(

  // If there were no overlapping files, return immediately.
  if (!foundOverlap) {
+    if (next_smallest) {
+      next_smallest = nullptr;
+    }
    return;
  }
  // returns the index where an overlap is found
@@ -2254,6 +2263,15 @@ void VersionStorageInfo::GetOverlappingInputsRangeBinarySearch(
  for (int i = start_index; i <= end_index; i++) {
    inputs->push_back(files_[level][i]);
  }
+
+  if (next_smallest != nullptr) {
+    // Provide the next key outside the range covered by inputs
+    if (++end_index < static_cast<int>(files_[level].size())) {
+      **next_smallest = files_[level][end_index]->smallest;
+    } else {
+      *next_smallest = nullptr;
+    }
+  }
 }

 // Store in *start_index and *end_index the range of all files in
@@ -188,9 +188,11 @@ class VersionStorageInfo {
      std::vector<FileMetaData*>* inputs,
      int hint_index = -1,        // index of overlap file
      int* file_index = nullptr,  // return index of overlap file
-      bool expand_range = true)   // if set, returns files which overlap the
-      const;                      // range and overlap each other. If false,
+      bool expand_range = true,   // if set, returns files which overlap the
+                                  // range and overlap each other. If false,
                                  // then just files intersecting the range
+      InternalKey** next_smallest = nullptr)  // if non-null, returns the
+      const;  // smallest key of next file not included
  void GetCleanInputsWithinInterval(
      int level, const InternalKey* begin,  // nullptr means before all keys
      const InternalKey* end,               // nullptr means after all keys
@@ -200,14 +202,15 @@ class VersionStorageInfo {
      const;

  void GetOverlappingInputsRangeBinarySearch(
-      int level,           // level > 0
+      int level,                 // level > 0
      const InternalKey* begin,  // nullptr means before all keys
      const InternalKey* end,    // nullptr means after all keys
      std::vector<FileMetaData*>* inputs,
      int hint_index,                // index of overlap file
      int* file_index,               // return index of overlap file
-      bool within_interval = false)  // if set, force the inputs within interval
-      const;
+      bool within_interval = false,  // if set, force the inputs within interval
+      InternalKey** next_smallest = nullptr)  // if non-null, returns the
+      const;  // smallest key of next file not included

  void ExtendFileRangeOverlappingInterval(
      int level,
@@ -24,7 +24,10 @@ WriteThread::WriteThread(const ImmutableDBOptions& db_options)
      enable_pipelined_write_(db_options.enable_pipelined_write),
      newest_writer_(nullptr),
      newest_memtable_writer_(nullptr),
-      last_sequence_(0) {}
+      last_sequence_(0),
+      write_stall_dummy_(),
+      stall_mu_(),
+      stall_cv_(&stall_mu_) {}

 uint8_t WriteThread::BlockingAwaitState(Writer* w, uint8_t goal_mask) {
  // We're going to block.  Lazily create the mutex.  We guarantee
@@ -219,6 +222,28 @@ bool WriteThread::LinkOne(Writer* w, std::atomic<Writer*>* newest_writer) {
  assert(w->state == STATE_INIT);
  Writer* writers = newest_writer->load(std::memory_order_relaxed);
  while (true) {
+    // If write stall in effect, and w->no_slowdown is not true,
+    // block here until stall is cleared. If its true, then return
+    // immediately
+    if (writers == &write_stall_dummy_) {
+      if (w->no_slowdown) {
+        w->status = Status::Incomplete("Write stall");
+        SetState(w, STATE_COMPLETED);
+        return false;
+      }
+      // Since no_slowdown is false, wait here to be notified of the write
+      // stall clearing
+      {
+        MutexLock lock(&stall_mu_);
+        writers = newest_writer->load(std::memory_order_relaxed);
+        if (writers == &write_stall_dummy_) {
+          stall_cv_.Wait();
+          // Load newest_writers_ again since it may have changed
+          writers = newest_writer->load(std::memory_order_relaxed);
+          continue;
+        }
+      }
+    }
    w->link_older = writers;
    if (newest_writer->compare_exchange_weak(writers, w)) {
      return (writers == nullptr);
@@ -303,12 +328,44 @@ void WriteThread::CompleteFollower(Writer* w, WriteGroup& write_group) {
  SetState(w, STATE_COMPLETED);
 }

+void WriteThread::BeginWriteStall() {
+  LinkOne(&write_stall_dummy_, &newest_writer_);
+
+  // Walk writer list until w->write_group != nullptr. The current write group
+  // will not have a mix of slowdown/no_slowdown, so its ok to stop at that
+  // point
+  Writer* w = write_stall_dummy_.link_older;
+  Writer* prev = &write_stall_dummy_;
+  while (w != nullptr && w->write_group == nullptr) {
+    if (w->no_slowdown) {
+      prev->link_older = w->link_older;
+      w->status = Status::Incomplete("Write stall");
+      SetState(w, STATE_COMPLETED);
+      w = prev->link_older;
+    } else {
+      prev = w;
+      w = w->link_older;
+    }
+  }
+}
+
+void WriteThread::EndWriteStall() {
+  MutexLock lock(&stall_mu_);
+
+  assert(newest_writer_.load(std::memory_order_relaxed) == &write_stall_dummy_);
+  newest_writer_.exchange(write_stall_dummy_.link_older);
+
+  // Wake up writers
+  stall_cv_.SignalAll();
+}
+
 static WriteThread::AdaptationContext jbg_ctx("JoinBatchGroup");
 void WriteThread::JoinBatchGroup(Writer* w) {
  TEST_SYNC_POINT_CALLBACK("WriteThread::JoinBatchGroup:Start", w);
  assert(w->batch != nullptr);

  bool linked_as_leader = LinkOne(w, &newest_writer_);
+
  if (linked_as_leader) {
    SetState(w, STATE_GROUP_LEADER);
  }
@@ -342,6 +342,13 @@ class WriteThread {
    return last_sequence_;
  }

+  // Insert a dummy writer at the tail of the write queue to indicate a write
+  // stall, and fail any writers in the queue with no_slowdown set to true
+  void BeginWriteStall();
+
+  // Remove the dummy writer and wake up waiting writers
+  void EndWriteStall();
+
 private:
  // See AwaitState.
  const uint64_t max_yield_usec_;
@@ -365,6 +372,17 @@ class WriteThread {
  // is not necessary visible to reads because the writer can be ongoing.
  SequenceNumber last_sequence_;

+  // A dummy writer to indicate a write stall condition. This will be inserted
+  // at the tail of the writer queue by the leader, so newer writers can just
+  // check for this and bail
+  Writer write_stall_dummy_;
+
+  // Mutex and condvar for writers to block on a write stall. During a write
+  // stall, writers with no_slowdown set to false will wait on this rather
+  // on the writer queue
+  port::Mutex stall_mu_;
+  port::CondVar stall_cv_;
+
  // Waits for w->state & goal_mask using w->StateMutex().  Returns
  // the state that satisfies goal_mask.
  uint8_t BlockingAwaitState(Writer* w, uint8_t goal_mask);
@@ -25,7 +25,6 @@
 #include <stdint.h>
 #include <memory>
 #include <string>
-#include "rocksdb/cache_allocator.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/statistics.h"
 #include "rocksdb/status.h"
@@ -59,20 +58,13 @@ struct LRUCacheOptions {
  // BlockBasedTableOptions::cache_index_and_filter_blocks_with_high_priority.
  double high_pri_pool_ratio = 0.0;

-  // If non-nullptr will use this allocator instead of system allocator when
-  // allocating memory for cache blocks. Call this method before you start using
-  // the cache!
-  std::shared_ptr<CacheAllocator> cache_allocator;
-
  LRUCacheOptions() {}
  LRUCacheOptions(size_t _capacity, int _num_shard_bits,
-                  bool _strict_capacity_limit, double _high_pri_pool_ratio,
-                  std::shared_ptr<CacheAllocator> _cache_allocator = nullptr)
+                  bool _strict_capacity_limit, double _high_pri_pool_ratio)
      : capacity(_capacity),
        num_shard_bits(_num_shard_bits),
        strict_capacity_limit(_strict_capacity_limit),
-        high_pri_pool_ratio(_high_pri_pool_ratio),
-        cache_allocator(std::move(_cache_allocator)) {}
+        high_pri_pool_ratio(_high_pri_pool_ratio) {}
 };

 // Create a new cache with a fixed size capacity. The cache is sharded
@@ -83,10 +75,10 @@ struct LRUCacheOptions {
 // high_pri_pool_pct.
 // num_shard_bits = -1 means it is automatically determined: every shard
 // will be at least 512KB and number of shard bits will not exceed 6.
-extern std::shared_ptr<Cache> NewLRUCache(
-    size_t capacity, int num_shard_bits = -1,
-    bool strict_capacity_limit = false, double high_pri_pool_ratio = 0.0,
-    std::shared_ptr<CacheAllocator> cache_allocator = nullptr);
+extern std::shared_ptr<Cache> NewLRUCache(size_t capacity,
+                                          int num_shard_bits = -1,
+                                          bool strict_capacity_limit = false,
+                                          double high_pri_pool_ratio = 0.0);

 extern std::shared_ptr<Cache> NewLRUCache(const LRUCacheOptions& cache_opts);

@@ -99,15 +91,13 @@ extern std::shared_ptr<Cache> NewClockCache(size_t capacity,
                                            int num_shard_bits = -1,
                                            bool strict_capacity_limit = false);

-
 class Cache {
 public:
  // Depending on implementation, cache entries with high priority could be less
  // likely to get evicted than low priority entries.
  enum class Priority { HIGH, LOW };

-  Cache(std::shared_ptr<CacheAllocator> allocator = nullptr)
-      : cache_allocator_(std::move(allocator)) {}
+  Cache() {}

  // Destroys all existing entries by calling the "deleter"
  // function that was passed via the Insert() function.
@@ -238,14 +228,10 @@ class Cache {
  virtual void TEST_mark_as_data_block(const Slice& /*key*/,
                                       size_t /*charge*/) {}

-  CacheAllocator* cache_allocator() const { return cache_allocator_.get(); }
-
 private:
  // No copying allowed
  Cache(const Cache&);
  Cache& operator=(const Cache&);
-
-  std::shared_ptr<CacheAllocator> cache_allocator_;
 };

 }  // namespace rocksdb
@@ -1,29 +0,0 @@
-// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-
-#pragma once
-
-// CacheAllocator is an interface that a client can implement to supply custom
-// cache allocation and deallocation methods. See rocksdb/cache.h for more
-// information.
-// All methods should be thread-safe.
-class CacheAllocator {
- public:
-  virtual ~CacheAllocator() = default;
-
-  // Name of the cache allocator, printed in the log
-  virtual const char* Name() const = 0;
-
-  // Allocate a block of at least size size
-  virtual void* Allocate(size_t size) = 0;
-  // Deallocate previously allocated block
-  virtual void Deallocate(void* p) = 0;
-  // Returns the memory size of the block allocated at p. The default
-  // implementation that just returns the original allocation_size is fine.
-  virtual size_t UsableSize(void* /*p*/, size_t allocation_size) const {
-    // default implementation just returns the allocation size
-    return allocation_size;
-  }
-};
@@ -2,8 +2,8 @@
 //  This source code is licensed under both the GPLv2 (found in the
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
-#ifndef ROCKSDB_LITE
 #pragma once
+#ifndef ROCKSDB_LITE
 #include <string>
 #include <vector>
 #include "rocksdb/db.h"
@@ -3,10 +3,10 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).

-#ifndef ROCKSDB_LITE
-
 #pragma once

+#ifndef ROCKSDB_LITE
+
 #include <memory>
 #include <string>

@@ -5,8 +5,8 @@
 #pragma once

 #define ROCKSDB_MAJOR 5
-#define ROCKSDB_MINOR 16
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_MINOR 17
+#define ROCKSDB_PATCH 4

 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these
@@ -39,7 +39,6 @@
 #include "table/full_filter_block.h"
 #include "table/table_builder.h"

-#include "util/cache_allocator.h"
 #include "util/coding.h"
 #include "util/compression.h"
 #include "util/crc32c.h"
@@ -655,8 +654,7 @@ Status BlockBasedTableBuilder::InsertBlockInCache(const Slice& block_contents,

    size_t size = block_contents.size();

-    auto ubuf =
-        AllocateBlock(size + 1, block_cache_compressed->cache_allocator());
+    std::unique_ptr<char[]> ubuf(new char[size + 1]);
    memcpy(ubuf.get(), block_contents.data(), size);
    ubuf[size] = type;

@@ -80,12 +80,11 @@ Status ReadBlockFromFile(
    std::unique_ptr<Block>* result, const ImmutableCFOptions& ioptions,
    bool do_uncompress, const Slice& compression_dict,
    const PersistentCacheOptions& cache_options, SequenceNumber global_seqno,
-    size_t read_amp_bytes_per_bit, CacheAllocator* allocator = nullptr,
-    const bool immortal_file = false) {
+    size_t read_amp_bytes_per_bit, const bool immortal_file = false) {
  BlockContents contents;
-  BlockFetcher block_fetcher(
-      file, prefetch_buffer, footer, options, handle, &contents, ioptions,
-      do_uncompress, compression_dict, cache_options, allocator, immortal_file);
+  BlockFetcher block_fetcher(file, prefetch_buffer, footer, options, handle,
+                             &contents, ioptions, do_uncompress,
+                             compression_dict, cache_options, immortal_file);
  Status s = block_fetcher.ReadBlockContents();
  if (s.ok()) {
    result->reset(new Block(std::move(contents), global_seqno,
@@ -95,13 +94,6 @@ Status ReadBlockFromFile(
  return s;
 }

-inline CacheAllocator* GetCacheAllocator(
-    const BlockBasedTableOptions& table_options) {
-  return table_options.block_cache.get()
-             ? table_options.block_cache->cache_allocator()
-             : nullptr;
-}
-
 // Delete the resource that is held by the iterator.
 template <class ResourceType>
 void DeleteHeldResource(void* arg, void* /*ignored*/) {
@@ -1158,8 +1150,7 @@ Status BlockBasedTable::ReadMetaBlock(Rep* rep,
      rep->footer.metaindex_handle(), &meta, rep->ioptions,
      true /* decompress */, Slice() /*compression dict*/,
      rep->persistent_cache_options, kDisableGlobalSequenceNumber,
-      0 /* read_amp_bytes_per_bit */,
-      GetCacheAllocator(rep->table_options));
+      0 /* read_amp_bytes_per_bit */);

  if (!s.ok()) {
    ROCKS_LOG_ERROR(rep->ioptions.info_log,
@@ -1182,7 +1173,7 @@ Status BlockBasedTable::GetDataBlockFromCache(
    const ImmutableCFOptions& ioptions, const ReadOptions& read_options,
    BlockBasedTable::CachableEntry<Block>* block, uint32_t format_version,
    const Slice& compression_dict, size_t read_amp_bytes_per_bit, bool is_index,
-    GetContext* get_context, CacheAllocator* allocator) {
+    GetContext* get_context) {
  Status s;
  Block* compressed_block = nullptr;
  Cache::Handle* block_cache_compressed_handle = nullptr;
@@ -1239,7 +1230,7 @@ Status BlockBasedTable::GetDataBlockFromCache(
                                          compression_dict);
  s = UncompressBlockContents(uncompresssion_ctx, compressed_block->data(),
                              compressed_block->size(), &contents,
-                              format_version, ioptions, allocator);
+                              format_version, ioptions);

  // Insert uncompressed block into block cache
  if (s.ok()) {
@@ -1301,8 +1292,7 @@ Status BlockBasedTable::PutDataBlockToCache(
    const ReadOptions& /*read_options*/, const ImmutableCFOptions& ioptions,
    CachableEntry<Block>* block, Block* raw_block, uint32_t format_version,
    const Slice& compression_dict, size_t read_amp_bytes_per_bit, bool is_index,
-    Cache::Priority priority, GetContext* get_context,
-    CacheAllocator* allocator) {
+    Cache::Priority priority, GetContext* get_context) {
  assert(raw_block->compression_type() == kNoCompression ||
         block_cache_compressed != nullptr);

@@ -1315,7 +1305,7 @@ Status BlockBasedTable::PutDataBlockToCache(
                                           compression_dict);
    s = UncompressBlockContents(uncompression_ctx, raw_block->data(),
                                raw_block->size(), &contents, format_version,
-                                ioptions, allocator);
+                                ioptions);
  }
  if (!s.ok()) {
    delete raw_block;
@@ -1412,8 +1402,7 @@ FilterBlockReader* BlockBasedTable::ReadFilter(
  BlockFetcher block_fetcher(rep->file.get(), prefetch_buffer, rep->footer,
                             ReadOptions(), filter_handle, &block,
                             rep->ioptions, false /* decompress */,
-                             dummy_comp_dict, rep->persistent_cache_options,
-                             GetCacheAllocator(rep->table_options));
+                             dummy_comp_dict, rep->persistent_cache_options);
  Status s = block_fetcher.ReadBlockContents();

  if (!s.ok()) {
@@ -1711,9 +1700,7 @@ TBlockIter* BlockBasedTable::NewDataBlockIterator(
          &block_value, rep->ioptions, rep->blocks_maybe_compressed,
          compression_dict, rep->persistent_cache_options,
          is_index ? kDisableGlobalSequenceNumber : rep->global_seqno,
-          rep->table_options.read_amp_bytes_per_bit,
-          GetCacheAllocator(rep->table_options),
-          rep->immortal_table);
+          rep->table_options.read_amp_bytes_per_bit, rep->immortal_table);
    }
    if (s.ok()) {
      block.value = block_value.release();
@@ -1805,8 +1792,7 @@ Status BlockBasedTable::MaybeLoadDataBlockToCache(
    s = GetDataBlockFromCache(
        key, ckey, block_cache, block_cache_compressed, rep->ioptions, ro,
        block_entry, rep->table_options.format_version, compression_dict,
-        rep->table_options.read_amp_bytes_per_bit, is_index, get_context,
-        GetCacheAllocator(rep->table_options));
+        rep->table_options.read_amp_bytes_per_bit, is_index, get_context);

    if (block_entry->value == nullptr && !no_io && ro.fill_cache) {
      std::unique_ptr<Block> raw_block;
@@ -1818,9 +1804,7 @@ Status BlockBasedTable::MaybeLoadDataBlockToCache(
            block_cache_compressed == nullptr && rep->blocks_maybe_compressed,
            compression_dict, rep->persistent_cache_options,
            is_index ? kDisableGlobalSequenceNumber : rep->global_seqno,
-            rep->table_options.read_amp_bytes_per_bit,
-            GetCacheAllocator(rep->table_options),
-            rep->immortal_table);
+            rep->table_options.read_amp_bytes_per_bit, rep->immortal_table);
      }

      if (s.ok()) {
@@ -1833,7 +1817,7 @@ Status BlockBasedTable::MaybeLoadDataBlockToCache(
                            .cache_index_and_filter_blocks_with_high_priority
                ? Cache::Priority::HIGH
                : Cache::Priority::LOW,
-            get_context, GetCacheAllocator(rep->table_options));
+            get_context);
      }
    }
  }
@@ -2540,12 +2524,11 @@ Status BlockBasedTable::VerifyChecksumInBlocks(
    BlockHandle handle = index_iter->value();
    BlockContents contents;
    Slice dummy_comp_dict;
-    BlockFetcher block_fetcher(
-        rep_->file.get(), nullptr /* prefetch buffer */, rep_->footer,
-        ReadOptions(), handle, &contents, rep_->ioptions,
-        false /* decompress */, dummy_comp_dict /*compression dict*/,
-        rep_->persistent_cache_options,
-        GetCacheAllocator(rep_->table_options));
+    BlockFetcher block_fetcher(rep_->file.get(), nullptr /* prefetch buffer */,
+                               rep_->footer, ReadOptions(), handle, &contents,
+                               rep_->ioptions, false /* decompress */,
+                               dummy_comp_dict /*compression dict*/,
+                               rep_->persistent_cache_options);
    s = block_fetcher.ReadBlockContents();
    if (!s.ok()) {
      break;
@@ -2567,12 +2550,11 @@ Status BlockBasedTable::VerifyChecksumInBlocks(
    s = handle.DecodeFrom(&input);
    BlockContents contents;
    Slice dummy_comp_dict;
-    BlockFetcher block_fetcher(
-        rep_->file.get(), nullptr /* prefetch buffer */, rep_->footer,
-        ReadOptions(), handle, &contents, rep_->ioptions,
-        false /* decompress */, dummy_comp_dict /*compression dict*/,
-        rep_->persistent_cache_options,
-        GetCacheAllocator(rep_->table_options));
+    BlockFetcher block_fetcher(rep_->file.get(), nullptr /* prefetch buffer */,
+                               rep_->footer, ReadOptions(), handle, &contents,
+                               rep_->ioptions, false /* decompress */,
+                               dummy_comp_dict /*compression dict*/,
+                               rep_->persistent_cache_options);
    s = block_fetcher.ReadBlockContents();
    if (!s.ok()) {
      break;
@@ -2876,8 +2858,7 @@ Status BlockBasedTable::DumpTable(WritableFile* out_file,
              rep_->file.get(), nullptr /* prefetch_buffer */, rep_->footer,
              ReadOptions(), handle, &block, rep_->ioptions,
              false /*decompress*/, dummy_comp_dict /*compression dict*/,
-              rep_->persistent_cache_options,
-              GetCacheAllocator(rep_->table_options));
+              rep_->persistent_cache_options);
          s = block_fetcher.ReadBlockContents();
          if (!s.ok()) {
            rep_->filter.reset(new BlockBasedFilterBlockReader(
@@ -303,8 +303,7 @@ class BlockBasedTable : public TableReader {
      const ImmutableCFOptions& ioptions, const ReadOptions& read_options,
      BlockBasedTable::CachableEntry<Block>* block, uint32_t format_version,
      const Slice& compression_dict, size_t read_amp_bytes_per_bit,
-      bool is_index = false, GetContext* get_context = nullptr,
-      CacheAllocator* allocator = nullptr);
+      bool is_index = false, GetContext* get_context = nullptr);

  // Put a raw block (maybe compressed) to the corresponding block caches.
  // This method will perform decompression against raw_block if needed and then
@@ -323,7 +322,7 @@ class BlockBasedTable : public TableReader {
      CachableEntry<Block>* block, Block* raw_block, uint32_t format_version,
      const Slice& compression_dict, size_t read_amp_bytes_per_bit,
      bool is_index = false, Cache::Priority pri = Cache::Priority::LOW,
-      GetContext* get_context = nullptr, CacheAllocator* allocator = nullptr);
+      GetContext* get_context = nullptr);

  // Calls (*handle_result)(arg, ...) repeatedly, starting with the entry found
  // after a call to Seek(key), until handle_result returns false.
@@ -17,9 +17,8 @@
 #include "rocksdb/env.h"
 #include "table/block.h"
 #include "table/block_based_table_reader.h"
-#include "table/format.h"
 #include "table/persistent_cache_helper.h"
-#include "util/cache_allocator.h"
+#include "table/format.h"
 #include "util/coding.h"
 #include "util/compression.h"
 #include "util/crc32c.h"
@@ -108,11 +107,9 @@ bool BlockFetcher::TryGetCompressedBlockFromPersistentCache() {
  if (cache_options_.persistent_cache &&
      cache_options_.persistent_cache->IsCompressed()) {
    // lookup uncompressed cache mode p-cache
-    std::unique_ptr<char[]> raw_data;
    status_ = PersistentCacheHelper::LookupRawPage(
-        cache_options_, handle_, &raw_data, block_size_ + kBlockTrailerSize);
+        cache_options_, handle_, &heap_buf_, block_size_ + kBlockTrailerSize);
    if (status_.ok()) {
-      heap_buf_ = CacheAllocationPtr(raw_data.release());
      used_buf_ = heap_buf_.get();
      slice_ = Slice(heap_buf_.get(), block_size_);
      return true;
@@ -135,7 +132,7 @@ void BlockFetcher::PrepareBufferForBlockFromFile() {
    // trivially allocated stack buffer instead of needing a full malloc()
    used_buf_ = &stack_buf_[0];
  } else {
-    heap_buf_ = AllocateBlock(block_size_ + kBlockTrailerSize, allocator_);
+    heap_buf_.reset(new char[block_size_ + kBlockTrailerSize]);
    used_buf_ = heap_buf_.get();
  }
 }
@@ -173,7 +170,7 @@ void BlockFetcher::GetBlockContents() {
    // or heap provided. Refer to https://github.com/facebook/rocksdb/pull/4096
    if (got_from_prefetch_buffer_ || used_buf_ == &stack_buf_[0]) {
      assert(used_buf_ != heap_buf_.get());
-      heap_buf_ = AllocateBlock(block_size_ + kBlockTrailerSize, allocator_);
+      heap_buf_.reset(new char[block_size_ + kBlockTrailerSize]);
      memcpy(heap_buf_.get(), used_buf_, block_size_ + kBlockTrailerSize);
    }
    *contents_ = BlockContents(std::move(heap_buf_), block_size_, true,
@@ -231,9 +228,9 @@ Status BlockFetcher::ReadBlockContents() {
  if (do_uncompress_ && compression_type != kNoCompression) {
    // compressed page, uncompress, update cache
    UncompressionContext uncompression_ctx(compression_type, compression_dict_);
-    status_ = UncompressBlockContents(uncompression_ctx, slice_.data(),
-                                      block_size_, contents_, footer_.version(),
-                                      ioptions_, allocator_);
+    status_ =
+        UncompressBlockContents(uncompression_ctx, slice_.data(), block_size_,
+                                contents_, footer_.version(), ioptions_);
  } else {
    GetBlockContents();
  }
@@ -11,8 +11,6 @@
 #include "table/block.h"
 #include "table/format.h"

-#include "util/cache_allocator.h"
-
 namespace rocksdb {
 class BlockFetcher {
 public:
@@ -28,7 +26,6 @@ class BlockFetcher {
               BlockContents* contents, const ImmutableCFOptions& ioptions,
               bool do_uncompress, const Slice& compression_dict,
               const PersistentCacheOptions& cache_options,
-               CacheAllocator* allocator = nullptr,
               const bool immortal_source = false)
      : file_(file),
        prefetch_buffer_(prefetch_buffer),
@@ -40,8 +37,7 @@ class BlockFetcher {
        do_uncompress_(do_uncompress),
        immortal_source_(immortal_source),
        compression_dict_(compression_dict),
-        cache_options_(cache_options),
-        allocator_(allocator) {}
+        cache_options_(cache_options) {}
  Status ReadBlockContents();

 private:
@@ -58,12 +54,11 @@ class BlockFetcher {
  const bool immortal_source_;
  const Slice& compression_dict_;
  const PersistentCacheOptions& cache_options_;
-  CacheAllocator* allocator_;
  Status status_;
  Slice slice_;
  char* used_buf_ = nullptr;
  size_t block_size_;
-  CacheAllocationPtr heap_buf_;
+  std::unique_ptr<char[]> heap_buf_;
  char stack_buf_[kDefaultStackBufferSize];
  bool got_from_prefetch_buffer_ = false;
  rocksdb::CompressionType compression_type;
@@ -19,7 +19,6 @@
 #include "table/block_based_table_reader.h"
 #include "table/block_fetcher.h"
 #include "table/persistent_cache_helper.h"
-#include "util/cache_allocator.h"
 #include "util/coding.h"
 #include "util/compression.h"
 #include "util/crc32c.h"
@@ -280,9 +279,8 @@ Status ReadFooterFromFile(RandomAccessFileReader* file,
 Status UncompressBlockContentsForCompressionType(
    const UncompressionContext& uncompression_ctx, const char* data, size_t n,
    BlockContents* contents, uint32_t format_version,
-    const ImmutableCFOptions& ioptions,
-    CacheAllocator* allocator) {
-  CacheAllocationPtr ubuf;
+    const ImmutableCFOptions& ioptions) {
+  std::unique_ptr<char[]> ubuf;

  assert(uncompression_ctx.type() != kNoCompression &&
         "Invalid compression type");
@@ -298,7 +296,7 @@ Status UncompressBlockContentsForCompressionType(
      if (!Snappy_GetUncompressedLength(data, n, &ulength)) {
        return Status::Corruption(snappy_corrupt_msg);
      }
-      ubuf = AllocateBlock(ulength, allocator);
+      ubuf.reset(new char[ulength]);
      if (!Snappy_Uncompress(data, n, ubuf.get())) {
        return Status::Corruption(snappy_corrupt_msg);
      }
@@ -306,10 +304,9 @@ Status UncompressBlockContentsForCompressionType(
      break;
    }
    case kZlibCompression:
-      ubuf = Zlib_Uncompress(
+      ubuf.reset(Zlib_Uncompress(
          uncompression_ctx, data, n, &decompress_size,
-          GetCompressFormatForVersion(kZlibCompression, format_version),
-          allocator);
+          GetCompressFormatForVersion(kZlibCompression, format_version)));
      if (!ubuf) {
        static char zlib_corrupt_msg[] =
          "Zlib not supported or corrupted Zlib compressed block contents";
@@ -319,10 +316,9 @@ Status UncompressBlockContentsForCompressionType(
          BlockContents(std::move(ubuf), decompress_size, true, kNoCompression);
      break;
    case kBZip2Compression:
-      ubuf = BZip2_Uncompress(
+      ubuf.reset(BZip2_Uncompress(
          data, n, &decompress_size,
-          GetCompressFormatForVersion(kBZip2Compression, format_version),
-          allocator);
+          GetCompressFormatForVersion(kBZip2Compression, format_version)));
      if (!ubuf) {
        static char bzip2_corrupt_msg[] =
          "Bzip2 not supported or corrupted Bzip2 compressed block contents";
@@ -332,10 +328,9 @@ Status UncompressBlockContentsForCompressionType(
          BlockContents(std::move(ubuf), decompress_size, true, kNoCompression);
      break;
    case kLZ4Compression:
-      ubuf = LZ4_Uncompress(
+      ubuf.reset(LZ4_Uncompress(
          uncompression_ctx, data, n, &decompress_size,
-          GetCompressFormatForVersion(kLZ4Compression, format_version),
-          allocator);
+          GetCompressFormatForVersion(kLZ4Compression, format_version)));
      if (!ubuf) {
        static char lz4_corrupt_msg[] =
          "LZ4 not supported or corrupted LZ4 compressed block contents";
@@ -345,10 +340,9 @@ Status UncompressBlockContentsForCompressionType(
          BlockContents(std::move(ubuf), decompress_size, true, kNoCompression);
      break;
    case kLZ4HCCompression:
-      ubuf = LZ4_Uncompress(
+      ubuf.reset(LZ4_Uncompress(
          uncompression_ctx, data, n, &decompress_size,
-          GetCompressFormatForVersion(kLZ4HCCompression, format_version),
-          allocator);
+          GetCompressFormatForVersion(kLZ4HCCompression, format_version)));
      if (!ubuf) {
        static char lz4hc_corrupt_msg[] =
          "LZ4HC not supported or corrupted LZ4HC compressed block contents";
@@ -358,8 +352,6 @@ Status UncompressBlockContentsForCompressionType(
          BlockContents(std::move(ubuf), decompress_size, true, kNoCompression);
      break;
    case kXpressCompression:
-      // XPRESS allocates memory internally, thus no support for custom
-      // allocator.
      ubuf.reset(XPRESS_Uncompress(data, n, &decompress_size));
      if (!ubuf) {
        static char xpress_corrupt_msg[] =
@@ -371,8 +363,7 @@ Status UncompressBlockContentsForCompressionType(
      break;
    case kZSTD:
    case kZSTDNotFinalCompression:
-      ubuf = ZSTD_Uncompress(uncompression_ctx, data, n, &decompress_size,
-                             allocator);
+      ubuf.reset(ZSTD_Uncompress(uncompression_ctx, data, n, &decompress_size));
      if (!ubuf) {
        static char zstd_corrupt_msg[] =
            "ZSTD not supported or corrupted ZSTD compressed block contents";
@@ -405,13 +396,11 @@ Status UncompressBlockContentsForCompressionType(
 Status UncompressBlockContents(const UncompressionContext& uncompression_ctx,
                               const char* data, size_t n,
                               BlockContents* contents, uint32_t format_version,
-                               const ImmutableCFOptions& ioptions,
-                               CacheAllocator* allocator) {
+                               const ImmutableCFOptions& ioptions) {
  assert(data[n] != kNoCompression);
  assert(data[n] == uncompression_ctx.type());
-  return UncompressBlockContentsForCompressionType(uncompression_ctx, data, n,
-                                                   contents, format_version,
-                                                   ioptions, allocator);
+  return UncompressBlockContentsForCompressionType(
+      uncompression_ctx, data, n, contents, format_version, ioptions);
 }

 }  // namespace rocksdb
@@ -26,7 +26,6 @@
 #include "port/port.h"  // noexcept
 #include "table/persistent_cache_options.h"
 #include "util/file_reader_writer.h"
-#include "util/cache_allocator.h"

 namespace rocksdb {

@@ -193,7 +192,7 @@ struct BlockContents {
  Slice data;     // Actual contents of data
  bool cachable;  // True iff data can be cached
  CompressionType compression_type;
-  CacheAllocationPtr allocation;
+  std::unique_ptr<char[]> allocation;

  BlockContents() : cachable(false), compression_type(kNoCompression) {}

@@ -201,28 +200,16 @@ struct BlockContents {
                CompressionType _compression_type)
      : data(_data), cachable(_cachable), compression_type(_compression_type) {}

-  BlockContents(CacheAllocationPtr&& _data, size_t _size, bool _cachable,
+  BlockContents(std::unique_ptr<char[]>&& _data, size_t _size, bool _cachable,
                CompressionType _compression_type)
      : data(_data.get(), _size),
        cachable(_cachable),
        compression_type(_compression_type),
        allocation(std::move(_data)) {}

-  BlockContents(std::unique_ptr<char[]>&& _data, size_t _size, bool _cachable,
-                CompressionType _compression_type)
-      : data(_data.get(), _size),
-        cachable(_cachable),
-        compression_type(_compression_type) {
-    allocation.reset(_data.release());
-  }
-
  // The additional memory space taken by the block data.
  size_t usable_size() const {
    if (allocation.get() != nullptr) {
-      auto allocator = allocation.get_deleter().allocator;
-      if (allocator) {
-        return allocator->UsableSize(allocation.get(), data.size());
-      }
 #ifdef ROCKSDB_MALLOC_USABLE_SIZE
      return malloc_usable_size(allocation.get());
 #else
@@ -265,7 +252,7 @@ extern Status ReadBlockContents(
 extern Status UncompressBlockContents(
    const UncompressionContext& uncompression_ctx, const char* data, size_t n,
    BlockContents* contents, uint32_t compress_format_version,
-    const ImmutableCFOptions& ioptions, CacheAllocator* allocator = nullptr);
+    const ImmutableCFOptions& ioptions);

 // This is an extension to UncompressBlockContents that accepts
 // a specific compression type. This is used by un-wrapped blocks
@@ -273,7 +260,7 @@ extern Status UncompressBlockContents(
 extern Status UncompressBlockContentsForCompressionType(
    const UncompressionContext& uncompression_ctx, const char* data, size_t n,
    BlockContents* contents, uint32_t compress_format_version,
-    const ImmutableCFOptions& ioptions, CacheAllocator* allocator = nullptr);
+    const ImmutableCFOptions& ioptions);

 // Implementation details follow.  Clients should ignore,

@@ -153,8 +153,8 @@ class PlainTableReader: public TableReader {
  DynamicBloom bloom_;
  PlainTableReaderFileInfo file_info_;
  Arena arena_;
-  CacheAllocationPtr index_block_alloc_;
-  CacheAllocationPtr bloom_block_alloc_;
+  std::unique_ptr<char[]> index_block_alloc_;
+  std::unique_ptr<char[]> bloom_block_alloc_;

  const ImmutableCFOptions& ioptions_;
  uint64_t file_size_;
@@ -2477,78 +2477,6 @@ TEST_P(BlockBasedTableTest, BlockCacheLeak) {
  c.ResetTableReader();
 }

-namespace {
-class CustomCacheAllocator : public CacheAllocator {
- public:
-  virtual const char* Name() const override { return "CustomCacheAllocator"; }
-
-  void* Allocate(size_t size) override {
-    ++numAllocations;
-    auto ptr = new char[size + 16];
-    memcpy(ptr, "cache_allocator_", 16);  // mangle first 16 bytes
-    return reinterpret_cast<void*>(ptr + 16);
-  }
-  void Deallocate(void* p) override {
-    ++numDeallocations;
-    char* ptr = reinterpret_cast<char*>(p) - 16;
-    delete[] ptr;
-  }
-
-  std::atomic<int> numAllocations;
-  std::atomic<int> numDeallocations;
-};
-}  // namespace
-
-TEST_P(BlockBasedTableTest, CacheAllocator) {
-  auto custom_cache_allocator = std::make_shared<CustomCacheAllocator>();
-  {
-    Options opt;
-    unique_ptr<InternalKeyComparator> ikc;
-    ikc.reset(new test::PlainInternalKeyComparator(opt.comparator));
-    opt.compression = kNoCompression;
-    BlockBasedTableOptions table_options;
-    table_options.block_size = 1024;
-    LRUCacheOptions lruOptions;
-    lruOptions.cache_allocator = custom_cache_allocator;
-    lruOptions.capacity = 16 * 1024 * 1024;
-    lruOptions.num_shard_bits = 4;
-    table_options.block_cache = NewLRUCache(std::move(lruOptions));
-    opt.table_factory.reset(NewBlockBasedTableFactory(table_options));
-
-    TableConstructor c(BytewiseComparator(),
-                       true /* convert_to_internal_key_ */);
-    c.Add("k01", "hello");
-    c.Add("k02", "hello2");
-    c.Add("k03", std::string(10000, 'x'));
-    c.Add("k04", std::string(200000, 'x'));
-    c.Add("k05", std::string(300000, 'x'));
-    c.Add("k06", "hello3");
-    c.Add("k07", std::string(100000, 'x'));
-    std::vector<std::string> keys;
-    stl_wrappers::KVMap kvmap;
-    const ImmutableCFOptions ioptions(opt);
-    const MutableCFOptions moptions(opt);
-    c.Finish(opt, ioptions, moptions, table_options, *ikc, &keys, &kvmap);
-
-    unique_ptr<InternalIterator> iter(
-        c.NewIterator(moptions.prefix_extractor.get()));
-    iter->SeekToFirst();
-    while (iter->Valid()) {
-      iter->key();
-      iter->value();
-      iter->Next();
-    }
-    ASSERT_OK(iter->status());
-  }
-
-  // out of scope, block cache should have been deleted, all allocations
-  // deallocated
-  EXPECT_EQ(custom_cache_allocator->numAllocations.load(),
-            custom_cache_allocator->numDeallocations.load());
-  // make sure that allocations actually happened through the cache allocator
-  EXPECT_GT(custom_cache_allocator->numAllocations.load(), 0);
-}
-
 TEST_P(BlockBasedTableTest, NewIndexIteratorLeak) {
  // A regression test to avoid data race described in
  // https://github.com/facebook/rocksdb/issues/1267
@@ -3040,7 +3040,7 @@ void VerifyDBFromDB(std::string& truth_db_name) {
    int64_t bytes = 0;
    int decompress_size;
    while (ok && bytes < 1024 * 1048576) {
-      CacheAllocationPtr uncompressed;
+      char *uncompressed = nullptr;
      switch (FLAGS_compression_type_e) {
        case rocksdb::kSnappyCompression: {
          // get size and allocate here to make comparison fair
@@ -3050,44 +3050,45 @@ void VerifyDBFromDB(std::string& truth_db_name) {
            ok = false;
            break;
          }
-          uncompressed = AllocateBlock(ulength, nullptr);
+          uncompressed = new char[ulength];
          ok = Snappy_Uncompress(compressed.data(), compressed.size(),
-                                 uncompressed.get());
+                                 uncompressed);
          break;
        }
      case rocksdb::kZlibCompression:
        uncompressed = Zlib_Uncompress(uncompression_ctx, compressed.data(),
                                       compressed.size(), &decompress_size, 2);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kBZip2Compression:
        uncompressed = BZip2_Uncompress(compressed.data(), compressed.size(),
                                        &decompress_size, 2);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kLZ4Compression:
        uncompressed = LZ4_Uncompress(uncompression_ctx, compressed.data(),
                                      compressed.size(), &decompress_size, 2);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kLZ4HCCompression:
        uncompressed = LZ4_Uncompress(uncompression_ctx, compressed.data(),
                                      compressed.size(), &decompress_size, 2);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kXpressCompression:
-        uncompressed.reset(XPRESS_Uncompress(
-            compressed.data(), compressed.size(), &decompress_size));
-        ok = uncompressed.get() != nullptr;
+        uncompressed = XPRESS_Uncompress(compressed.data(), compressed.size(),
+          &decompress_size);
+        ok = uncompressed != nullptr;
        break;
      case rocksdb::kZSTD:
        uncompressed = ZSTD_Uncompress(uncompression_ctx, compressed.data(),
                                       compressed.size(), &decompress_size);
-        ok = uncompressed.get() != nullptr;
+        ok = uncompressed != nullptr;
        break;
      default:
        ok = false;
      }
+      delete[] uncompressed;
      bytes += input.size();
      thread->stats.FinishedOps(nullptr, nullptr, 1, kUncompress);
    }
@@ -1,38 +0,0 @@
-//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-//
-
-#pragma once
-
-#include "rocksdb/cache_allocator.h"
-
-namespace rocksdb {
-
-struct CustomDeleter {
-  CustomDeleter(CacheAllocator* a = nullptr) : allocator(a) {}
-
-  void operator()(char* ptr) const {
-    if (allocator) {
-      allocator->Deallocate(reinterpret_cast<void*>(ptr));
-    } else {
-      delete[] ptr;
-    }
-  }
-
-  CacheAllocator* allocator;
-};
-
-using CacheAllocationPtr = std::unique_ptr<char[], CustomDeleter>;
-
-inline CacheAllocationPtr AllocateBlock(size_t size,
-                                        CacheAllocator* allocator) {
-  if (allocator) {
-    auto block = reinterpret_cast<char*>(allocator->Allocate(size));
-    return CacheAllocationPtr(block, allocator);
-  }
-  return CacheAllocationPtr(new char[size]);
-}
-
-}  // namespace rocksdb
@@ -14,8 +14,6 @@
 #include <string>

 #include "rocksdb/options.h"
-#include "rocksdb/table.h"
-#include "util/cache_allocator.h"
 #include "util/coding.h"
 #include "util/compression_context_cache.h"

@@ -497,10 +495,11 @@ inline bool Zlib_Compress(const CompressionContext& ctx,
 // header in varint32 format
 // @param compression_dict Data for presetting the compression library's
 //    dictionary.
-inline CacheAllocationPtr Zlib_Uncompress(
-    const UncompressionContext& ctx, const char* input_data,
-    size_t input_length, int* decompress_size, uint32_t compress_format_version,
-    CacheAllocator* allocator = nullptr, int windowBits = -14) {
+inline char* Zlib_Uncompress(const UncompressionContext& ctx,
+                             const char* input_data, size_t input_length,
+                             int* decompress_size,
+                             uint32_t compress_format_version,
+                             int windowBits = -14) {
 #ifdef ZLIB
  uint32_t output_len = 0;
  if (compress_format_version == 2) {
@@ -542,9 +541,9 @@ inline CacheAllocationPtr Zlib_Uncompress(
  _stream.next_in = (Bytef*)input_data;
  _stream.avail_in = static_cast<unsigned int>(input_length);

-  auto output = AllocateBlock(output_len, allocator);
+  char* output = new char[output_len];

-  _stream.next_out = (Bytef*)output.get();
+  _stream.next_out = (Bytef*)output;
  _stream.avail_out = static_cast<unsigned int>(output_len);

  bool done = false;
@@ -562,17 +561,19 @@ inline CacheAllocationPtr Zlib_Uncompress(
        size_t old_sz = output_len;
        uint32_t output_len_delta = output_len / 5;
        output_len += output_len_delta < 10 ? 10 : output_len_delta;
-        auto tmp = AllocateBlock(output_len, allocator);
-        memcpy(tmp.get(), output.get(), old_sz);
-        output = std::move(tmp);
+        char* tmp = new char[output_len];
+        memcpy(tmp, output, old_sz);
+        delete[] output;
+        output = tmp;

        // Set more output.
-        _stream.next_out = (Bytef*)(output.get() + old_sz);
+        _stream.next_out = (Bytef*)(output + old_sz);
        _stream.avail_out = static_cast<unsigned int>(output_len - old_sz);
        break;
      }
      case Z_BUF_ERROR:
      default:
+        delete[] output;
        inflateEnd(&_stream);
        return nullptr;
    }
@@ -659,9 +660,9 @@ inline bool BZip2_Compress(const CompressionContext& /*ctx*/,
 // block header
 // compress_format_version == 2 -- decompressed size is included in the block
 // header in varint32 format
-inline CacheAllocationPtr BZip2_Uncompress(
-    const char* input_data, size_t input_length, int* decompress_size,
-    uint32_t compress_format_version, CacheAllocator* allocator = nullptr) {
+inline char* BZip2_Uncompress(const char* input_data, size_t input_length,
+                              int* decompress_size,
+                              uint32_t compress_format_version) {
 #ifdef BZIP2
  uint32_t output_len = 0;
  if (compress_format_version == 2) {
@@ -689,9 +690,9 @@ inline CacheAllocationPtr BZip2_Uncompress(
  _stream.next_in = (char*)input_data;
  _stream.avail_in = static_cast<unsigned int>(input_length);

-  auto output = AllocateBlock(output_len, allocator);
+  char* output = new char[output_len];

-  _stream.next_out = (char*)output.get();
+  _stream.next_out = (char*)output;
  _stream.avail_out = static_cast<unsigned int>(output_len);

  bool done = false;
@@ -708,16 +709,18 @@ inline CacheAllocationPtr BZip2_Uncompress(
        assert(compress_format_version != 2);
        uint32_t old_sz = output_len;
        output_len = output_len * 1.2;
-        auto tmp = AllocateBlock(output_len, allocator);
-        memcpy(tmp.get(), output.get(), old_sz);
-        output = std::move(tmp);
+        char* tmp = new char[output_len];
+        memcpy(tmp, output, old_sz);
+        delete[] output;
+        output = tmp;

        // Set more output.
-        _stream.next_out = (char*)(output.get() + old_sz);
+        _stream.next_out = (char*)(output + old_sz);
        _stream.avail_out = static_cast<unsigned int>(output_len - old_sz);
        break;
      }
      default:
+        delete[] output;
        BZ2_bzDecompressEnd(&_stream);
        return nullptr;
    }
@@ -811,12 +814,10 @@ inline bool LZ4_Compress(const CompressionContext& ctx,
 // header in varint32 format
 // @param compression_dict Data for presetting the compression library's
 //    dictionary.
-inline CacheAllocationPtr LZ4_Uncompress(const UncompressionContext& ctx,
-                                         const char* input_data,
-                                         size_t input_length,
-                                         int* decompress_size,
-                                         uint32_t compress_format_version,
-                                         CacheAllocator* allocator = nullptr) {
+inline char* LZ4_Uncompress(const UncompressionContext& ctx,
+                            const char* input_data, size_t input_length,
+                            int* decompress_size,
+                            uint32_t compress_format_version) {
 #ifdef LZ4
  uint32_t output_len = 0;
  if (compress_format_version == 2) {
@@ -836,7 +837,7 @@ inline CacheAllocationPtr LZ4_Uncompress(const UncompressionContext& ctx,
    input_data += 8;
  }

-  auto output = AllocateBlock(output_len, allocator);
+  char* output = new char[output_len];
 #if LZ4_VERSION_NUMBER >= 10400  // r124+
  LZ4_streamDecode_t* stream = LZ4_createStreamDecode();
  if (ctx.dict().size()) {
@@ -844,16 +845,17 @@ inline CacheAllocationPtr LZ4_Uncompress(const UncompressionContext& ctx,
                        static_cast<int>(ctx.dict().size()));
  }
  *decompress_size = LZ4_decompress_safe_continue(
-      stream, input_data, output.get(), static_cast<int>(input_length),
+      stream, input_data, output, static_cast<int>(input_length),
      static_cast<int>(output_len));
  LZ4_freeStreamDecode(stream);
 #else   // up to r123
-  *decompress_size = LZ4_decompress_safe(input_data, output.get(),
-                                         static_cast<int>(input_length),
-                                         static_cast<int>(output_len));
+  *decompress_size =
+      LZ4_decompress_safe(input_data, output, static_cast<int>(input_length),
+                          static_cast<int>(output_len));
 #endif  // LZ4_VERSION_NUMBER >= 10400

  if (*decompress_size < 0) {
+    delete[] output;
    return nullptr;
  }
  assert(*decompress_size == static_cast<int>(output_len));
@@ -864,7 +866,6 @@ inline CacheAllocationPtr LZ4_Uncompress(const UncompressionContext& ctx,
  (void)input_length;
  (void)decompress_size;
  (void)compress_format_version;
-  (void)allocator;
  return nullptr;
 #endif
 }
@@ -1027,11 +1028,9 @@ inline bool ZSTD_Compress(const CompressionContext& ctx, const char* input,

 // @param compression_dict Data for presetting the compression library's
 //    dictionary.
-inline CacheAllocationPtr ZSTD_Uncompress(const UncompressionContext& ctx,
-                                          const char* input_data,
-                                          size_t input_length,
-                                          int* decompress_size,
-                                          CacheAllocator* allocator = nullptr) {
+inline char* ZSTD_Uncompress(const UncompressionContext& ctx,
+                             const char* input_data, size_t input_length,
+                             int* decompress_size) {
 #ifdef ZSTD
  uint32_t output_len = 0;
  if (!compression::GetDecompressedSizeInfo(&input_data, &input_length,
@@ -1039,17 +1038,17 @@ inline CacheAllocationPtr ZSTD_Uncompress(const UncompressionContext& ctx,
    return nullptr;
  }

-  auto output = AllocateBlock(output_len, allocator);
+  char* output = new char[output_len];
  size_t actual_output_length;
 #if ZSTD_VERSION_NUMBER >= 500  // v0.5.0+
  ZSTD_DCtx* context = ctx.GetZSTDContext();
  assert(context != nullptr);
  actual_output_length = ZSTD_decompress_usingDict(
-      context, output.get(), output_len, input_data, input_length,
-      ctx.dict().data(), ctx.dict().size());
+      context, output, output_len, input_data, input_length, ctx.dict().data(),
+      ctx.dict().size());
 #else   // up to v0.4.x
  actual_output_length =
-      ZSTD_decompress(output.get(), output_len, input_data, input_length);
+      ZSTD_decompress(output, output_len, input_data, input_length);
 #endif  // ZSTD_VERSION_NUMBER >= 500
  assert(actual_output_length == output_len);
  *decompress_size = static_cast<int>(actual_output_length);
@@ -1059,7 +1058,6 @@ inline CacheAllocationPtr ZSTD_Uncompress(const UncompressionContext& ctx,
  (void)input_data;
  (void)input_length;
  (void)decompress_size;
-  (void)allocator;
  return nullptr;
 #endif
 }
@@ -197,6 +197,27 @@ Status FaultInjectionTestEnv::NewWritableFile(const std::string& fname,
  return s;
 }

+Status FaultInjectionTestEnv::ReopenWritableFile(
+    const std::string& fname, unique_ptr<WritableFile>* result,
+    const EnvOptions& soptions) {
+  if (!IsFilesystemActive()) {
+    return GetError();
+  }
+  Status s = target()->ReopenWritableFile(fname, result, soptions);
+  if (s.ok()) {
+    result->reset(new TestWritableFile(fname, std::move(*result), this));
+    // WritableFileWriter* file is opened
+    // again then it will be truncated - so forget our saved state.
+    UntrackFile(fname);
+    MutexLock l(&mutex_);
+    open_files_.insert(fname);
+    auto dir_and_name = GetDirAndName(fname);
+    auto& list = dir_to_new_files_since_last_sync_[dir_and_name.first];
+    list.insert(dir_and_name.second);
+  }
+  return s;
+}
+
 Status FaultInjectionTestEnv::NewRandomAccessFile(
    const std::string& fname, std::unique_ptr<RandomAccessFile>* result,
    const EnvOptions& soptions) {
@@ -110,6 +110,10 @@ class FaultInjectionTestEnv : public EnvWrapper {
                         unique_ptr<WritableFile>* result,
                         const EnvOptions& soptions) override;

+  Status ReopenWritableFile(const std::string& fname,
+                            unique_ptr<WritableFile>* result,
+                            const EnvOptions& soptions) override;
+
  Status NewRandomAccessFile(const std::string& fname,
                             std::unique_ptr<RandomAccessFile>* result,
                             const EnvOptions& soptions) override;
@@ -404,82 +404,91 @@ std::shared_ptr<BlobFile> BlobDBImpl::FindBlobFileLocked(
  return (b1 || b2) ? nullptr : (*finditr);
 }

-std::shared_ptr<Writer> BlobDBImpl::CheckOrCreateWriterLocked(
-    const std::shared_ptr<BlobFile>& bfile) {
-  std::shared_ptr<Writer> writer = bfile->GetWriter();
-  if (writer) return writer;
-
-  Status s = CreateWriterLocked(bfile);
-  if (!s.ok()) return nullptr;
-
-  writer = bfile->GetWriter();
-  return writer;
+Status BlobDBImpl::CheckOrCreateWriterLocked(
+    const std::shared_ptr<BlobFile>& blob_file,
+    std::shared_ptr<Writer>* writer) {
+  assert(writer != nullptr);
+  *writer = blob_file->GetWriter();
+  if (*writer != nullptr) {
+    return Status::OK();
+  }
+  Status s = CreateWriterLocked(blob_file);
+  if (s.ok()) {
+    *writer = blob_file->GetWriter();
+  }
+  return s;
 }

-std::shared_ptr<BlobFile> BlobDBImpl::SelectBlobFile() {
+Status BlobDBImpl::SelectBlobFile(std::shared_ptr<BlobFile>* blob_file) {
+  assert(blob_file != nullptr);
  {
    ReadLock rl(&mutex_);
    if (open_non_ttl_file_ != nullptr) {
-      return open_non_ttl_file_;
+      *blob_file = open_non_ttl_file_;
+      return Status::OK();
    }
  }

  // CHECK again
  WriteLock wl(&mutex_);
  if (open_non_ttl_file_ != nullptr) {
-    return open_non_ttl_file_;
+    *blob_file = open_non_ttl_file_;
+    return Status::OK();
  }

-  std::shared_ptr<BlobFile> bfile = NewBlobFile("SelectBlobFile");
-  assert(bfile);
+  *blob_file = NewBlobFile("SelectBlobFile");
+  assert(*blob_file != nullptr);

  // file not visible, hence no lock
-  std::shared_ptr<Writer> writer = CheckOrCreateWriterLocked(bfile);
-  if (!writer) {
+  std::shared_ptr<Writer> writer;
+  Status s = CheckOrCreateWriterLocked(*blob_file, &writer);
+  if (!s.ok()) {
    ROCKS_LOG_ERROR(db_options_.info_log,
-                    "Failed to get writer from blob file: %s",
-                    bfile->PathName().c_str());
-    return nullptr;
+                    "Failed to get writer from blob file: %s, error: %s",
+                    (*blob_file)->PathName().c_str(), s.ToString().c_str());
+    return s;
  }

-  bfile->file_size_ = BlobLogHeader::kSize;
-  bfile->header_.compression = bdb_options_.compression;
-  bfile->header_.has_ttl = false;
-  bfile->header_.column_family_id =
+  (*blob_file)->file_size_ = BlobLogHeader::kSize;
+  (*blob_file)->header_.compression = bdb_options_.compression;
+  (*blob_file)->header_.has_ttl = false;
+  (*blob_file)->header_.column_family_id =
      reinterpret_cast<ColumnFamilyHandleImpl*>(DefaultColumnFamily())->GetID();
-  bfile->header_valid_ = true;
-  bfile->SetColumnFamilyId(bfile->header_.column_family_id);
-  bfile->SetHasTTL(false);
-  bfile->SetCompression(bdb_options_.compression);
+  (*blob_file)->header_valid_ = true;
+  (*blob_file)->SetColumnFamilyId((*blob_file)->header_.column_family_id);
+  (*blob_file)->SetHasTTL(false);
+  (*blob_file)->SetCompression(bdb_options_.compression);

-  Status s = writer->WriteHeader(bfile->header_);
+  s = writer->WriteHeader((*blob_file)->header_);
  if (!s.ok()) {
    ROCKS_LOG_ERROR(db_options_.info_log,
                    "Failed to write header to new blob file: %s"
                    " status: '%s'",
-                    bfile->PathName().c_str(), s.ToString().c_str());
-    return nullptr;
+                    (*blob_file)->PathName().c_str(), s.ToString().c_str());
+    return s;
  }

-  blob_files_.insert(std::make_pair(bfile->BlobFileNumber(), bfile));
-  open_non_ttl_file_ = bfile;
+  blob_files_.insert(
+      std::make_pair((*blob_file)->BlobFileNumber(), *blob_file));
+  open_non_ttl_file_ = *blob_file;
  total_blob_size_ += BlobLogHeader::kSize;
-  return bfile;
+  return s;
 }

-std::shared_ptr<BlobFile> BlobDBImpl::SelectBlobFileTTL(uint64_t expiration) {
+Status BlobDBImpl::SelectBlobFileTTL(uint64_t expiration,
+                                     std::shared_ptr<BlobFile>* blob_file) {
+  assert(blob_file != nullptr);
  assert(expiration != kNoExpiration);
  uint64_t epoch_read = 0;
-  std::shared_ptr<BlobFile> bfile;
  {
    ReadLock rl(&mutex_);
-    bfile = FindBlobFileLocked(expiration);
+    *blob_file = FindBlobFileLocked(expiration);
    epoch_read = epoch_of_.load();
  }

-  if (bfile) {
-    assert(!bfile->Immutable());
-    return bfile;
+  if (*blob_file != nullptr) {
+    assert(!(*blob_file)->Immutable());
+    return Status::OK();
  }

  uint64_t exp_low =
@@ -487,61 +496,66 @@ std::shared_ptr<BlobFile> BlobDBImpl::SelectBlobFileTTL(uint64_t expiration) {
  uint64_t exp_high = exp_low + bdb_options_.ttl_range_secs;
  ExpirationRange expiration_range = std::make_pair(exp_low, exp_high);

-  bfile = NewBlobFile("SelectBlobFileTTL");
-  assert(bfile);
+  *blob_file = NewBlobFile("SelectBlobFileTTL");
+  assert(*blob_file != nullptr);

  ROCKS_LOG_INFO(db_options_.info_log, "New blob file TTL range: %s %d %d",
-                 bfile->PathName().c_str(), exp_low, exp_high);
+                 (*blob_file)->PathName().c_str(), exp_low, exp_high);
  LogFlush(db_options_.info_log);

  // we don't need to take lock as no other thread is seeing bfile yet
-  std::shared_ptr<Writer> writer = CheckOrCreateWriterLocked(bfile);
-  if (!writer) {
-    ROCKS_LOG_ERROR(db_options_.info_log,
-                    "Failed to get writer from blob file with TTL: %s",
-                    bfile->PathName().c_str());
-    return nullptr;
+  std::shared_ptr<Writer> writer;
+  Status s = CheckOrCreateWriterLocked(*blob_file, &writer);
+  if (!s.ok()) {
+    ROCKS_LOG_ERROR(
+        db_options_.info_log,
+        "Failed to get writer from blob file with TTL: %s, error: %s",
+        (*blob_file)->PathName().c_str(), s.ToString().c_str());
+    return s;
  }

-  bfile->header_.expiration_range = expiration_range;
-  bfile->header_.compression = bdb_options_.compression;
-  bfile->header_.has_ttl = true;
-  bfile->header_.column_family_id =
+  (*blob_file)->header_.expiration_range = expiration_range;
+  (*blob_file)->header_.compression = bdb_options_.compression;
+  (*blob_file)->header_.has_ttl = true;
+  (*blob_file)->header_.column_family_id =
      reinterpret_cast<ColumnFamilyHandleImpl*>(DefaultColumnFamily())->GetID();
-  ;
-  bfile->header_valid_ = true;
-  bfile->SetColumnFamilyId(bfile->header_.column_family_id);
-  bfile->SetHasTTL(true);
-  bfile->SetCompression(bdb_options_.compression);
-  bfile->file_size_ = BlobLogHeader::kSize;
+  (*blob_file)->header_valid_ = true;
+  (*blob_file)->SetColumnFamilyId((*blob_file)->header_.column_family_id);
+  (*blob_file)->SetHasTTL(true);
+  (*blob_file)->SetCompression(bdb_options_.compression);
+  (*blob_file)->file_size_ = BlobLogHeader::kSize;

  // set the first value of the range, since that is
  // concrete at this time.  also necessary to add to open_ttl_files_
-  bfile->expiration_range_ = expiration_range;
+  (*blob_file)->expiration_range_ = expiration_range;

  WriteLock wl(&mutex_);
  // in case the epoch has shifted in the interim, then check
  // check condition again - should be rare.
  if (epoch_of_.load() != epoch_read) {
-    auto bfile2 = FindBlobFileLocked(expiration);
-    if (bfile2) return bfile2;
+    std::shared_ptr<BlobFile> blob_file2 = FindBlobFileLocked(expiration);
+    if (blob_file2 != nullptr) {
+      *blob_file = std::move(blob_file2);
+      return Status::OK();
+    }
  }

-  Status s = writer->WriteHeader(bfile->header_);
+  s = writer->WriteHeader((*blob_file)->header_);
  if (!s.ok()) {
    ROCKS_LOG_ERROR(db_options_.info_log,
                    "Failed to write header to new blob file: %s"
                    " status: '%s'",
-                    bfile->PathName().c_str(), s.ToString().c_str());
-    return nullptr;
+                    (*blob_file)->PathName().c_str(), s.ToString().c_str());
+    return s;
  }

-  blob_files_.insert(std::make_pair(bfile->BlobFileNumber(), bfile));
-  open_ttl_files_.insert(bfile);
+  blob_files_.insert(
+      std::make_pair((*blob_file)->BlobFileNumber(), *blob_file));
+  open_ttl_files_.insert(*blob_file);
  total_blob_size_ += BlobLogHeader::kSize;
  epoch_of_++;

-  return bfile;
+  return s;
 }

 class BlobDBImpl::BlobInserter : public WriteBatch::Handler {
@@ -695,36 +709,41 @@ Status BlobDBImpl::PutBlobValue(const WriteOptions& /*options*/,
      return s;
    }

-    std::shared_ptr<BlobFile> bfile = (expiration != kNoExpiration)
-                                          ? SelectBlobFileTTL(expiration)
-                                          : SelectBlobFile();
-    assert(bfile != nullptr);
-    assert(bfile->compression() == bdb_options_.compression);
-
-    s = AppendBlob(bfile, headerbuf, key, value_compressed, expiration,
-                   &index_entry);
-    if (expiration == kNoExpiration) {
-      RecordTick(statistics_, BLOB_DB_WRITE_BLOB);
+    std::shared_ptr<BlobFile> blob_file;
+    if (expiration != kNoExpiration) {
+      s = SelectBlobFileTTL(expiration, &blob_file);
    } else {
-      RecordTick(statistics_, BLOB_DB_WRITE_BLOB_TTL);
+      s = SelectBlobFile(&blob_file);
+    }
+    if (s.ok()) {
+      assert(blob_file != nullptr);
+      assert(blob_file->compression() == bdb_options_.compression);
+      s = AppendBlob(blob_file, headerbuf, key, value_compressed, expiration,
+                     &index_entry);
    }
-
    if (s.ok()) {
      if (expiration != kNoExpiration) {
-        bfile->ExtendExpirationRange(expiration);
+        blob_file->ExtendExpirationRange(expiration);
      }
-      s = CloseBlobFileIfNeeded(bfile);
-      if (s.ok()) {
-        s = WriteBatchInternal::PutBlobIndex(batch, column_family_id, key,
-                                             index_entry);
+      s = CloseBlobFileIfNeeded(blob_file);
+    }
+    if (s.ok()) {
+      s = WriteBatchInternal::PutBlobIndex(batch, column_family_id, key,
+                                           index_entry);
+    }
+    if (s.ok()) {
+      if (expiration == kNoExpiration) {
+        RecordTick(statistics_, BLOB_DB_WRITE_BLOB);
+      } else {
+        RecordTick(statistics_, BLOB_DB_WRITE_BLOB_TTL);
      }
    } else {
      ROCKS_LOG_ERROR(db_options_.info_log,
                      "Failed to append blob to FILE: %s: KEY: %s VALSZ: %d"
                      " status: '%s' blob_file: '%s'",
-                      bfile->PathName().c_str(), key.ToString().c_str(),
+                      blob_file->PathName().c_str(), key.ToString().c_str(),
                      value.size(), s.ToString().c_str(),
-                      bfile->DumpState().c_str());
+                      blob_file->DumpState().c_str());
    }
  }

@@ -867,9 +886,10 @@ Status BlobDBImpl::AppendBlob(const std::shared_ptr<BlobFile>& bfile,
  uint64_t key_offset = 0;
  {
    WriteLock lockbfile_w(&bfile->mutex_);
-    std::shared_ptr<Writer> writer = CheckOrCreateWriterLocked(bfile);
-    if (!writer) {
-      return Status::IOError("Failed to create blob writer");
+    std::shared_ptr<Writer> writer;
+    s = CheckOrCreateWriterLocked(bfile, &writer);
+    if (!s.ok()) {
+      return s;
    }

    // write the blob to the blob log.
@@ -1575,7 +1595,13 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
      reason += bfptr->PathName();
      newfile = NewBlobFile(reason);

-      new_writer = CheckOrCreateWriterLocked(newfile);
+      s = CheckOrCreateWriterLocked(newfile, &new_writer);
+      if (!s.ok()) {
+        ROCKS_LOG_ERROR(db_options_.info_log,
+                        "Failed to open file %s for writer, error: %s",
+                        newfile->PathName().c_str(), s.ToString().c_str());
+        break;
+      }
      // Can't use header beyond this point
      newfile->header_ = std::move(header);
      newfile->header_valid_ = true;
@@ -255,10 +255,11 @@ class BlobDBImpl : public BlobDB {

  // find an existing blob log file based on the expiration unix epoch
  // if such a file does not exist, return nullptr
-  std::shared_ptr<BlobFile> SelectBlobFileTTL(uint64_t expiration);
+  Status SelectBlobFileTTL(uint64_t expiration,
+                           std::shared_ptr<BlobFile>* blob_file);

  // find an existing blob log file to append the value to
-  std::shared_ptr<BlobFile> SelectBlobFile();
+  Status SelectBlobFile(std::shared_ptr<BlobFile>* blob_file);

  std::shared_ptr<BlobFile> FindBlobFileLocked(uint64_t expiration) const;

@@ -309,8 +310,8 @@ class BlobDBImpl : public BlobDB {

  // returns a Writer object for the file. If writer is not
  // already present, creates one. Needs Write Mutex to be held
-  std::shared_ptr<Writer> CheckOrCreateWriterLocked(
-      const std::shared_ptr<BlobFile>& bfile);
+  Status CheckOrCreateWriterLocked(const std::shared_ptr<BlobFile>& blob_file,
+                                   std::shared_ptr<Writer>* writer);

  // Iterate through keys and values on Blob and write into
  // separate file the remaining blobs and delete/update pointers
@@ -374,6 +374,19 @@ TEST_F(BlobDBTest, GetIOError) {
  fault_injection_env_->SetFilesystemActive(true);
 }

+TEST_F(BlobDBTest, PutIOError) {
+  Options options;
+  options.env = fault_injection_env_.get();
+  BlobDBOptions bdb_options;
+  bdb_options.min_blob_size = 0;  // Make sure value write to blob file
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options, options);
+  fault_injection_env_->SetFilesystemActive(false, Status::IOError());
+  ASSERT_TRUE(Put("foo", "v1").IsIOError());
+  fault_injection_env_->SetFilesystemActive(true, Status::IOError());
+  ASSERT_OK(Put("bar", "v1"));
+}
+
 TEST_F(BlobDBTest, WriteBatch) {
  Random rnd(301);
  BlobDBOptions bdb_options;
@@ -5,8 +5,8 @@
 * Copyright 2013 Facebook
 */

-#ifndef ROCKSDB_LITE
 #pragma once
+#ifndef ROCKSDB_LITE
 #include <exception>

 namespace rocksdb {
@@ -352,14 +352,16 @@ class WBWIIteratorImpl : public WBWIIterator {
  }

  virtual void SeekToFirst() override {
-    WriteBatchIndexEntry search_entry(WriteBatchIndexEntry::kFlagMin,
-                                      column_family_id_, 0, 0);
+    WriteBatchIndexEntry search_entry(
+        nullptr /* search_key */, column_family_id_,
+        true /* is_forward_direction */, true /* is_seek_to_first */);
    skip_list_iter_.Seek(&search_entry);
  }

  virtual void SeekToLast() override {
-    WriteBatchIndexEntry search_entry(WriteBatchIndexEntry::kFlagMin,
-                                      column_family_id_ + 1, 0, 0);
+    WriteBatchIndexEntry search_entry(
+        nullptr /* search_key */, column_family_id_ + 1,
+        true /* is_forward_direction */, true /* is_seek_to_first */);
    skip_list_iter_.Seek(&search_entry);
    if (!skip_list_iter_.Valid()) {
      skip_list_iter_.SeekToLast();
@@ -369,12 +371,16 @@ class WBWIIteratorImpl : public WBWIIterator {
  }

  virtual void Seek(const Slice& key) override {
-    WriteBatchIndexEntry search_entry(&key, column_family_id_);
+    WriteBatchIndexEntry search_entry(&key, column_family_id_,
+                                      true /* is_forward_direction */,
+                                      false /* is_seek_to_first */);
    skip_list_iter_.Seek(&search_entry);
  }

  virtual void SeekForPrev(const Slice& key) override {
-    WriteBatchIndexEntry search_entry(&key, column_family_id_);
+    WriteBatchIndexEntry search_entry(&key, column_family_id_,
+                                      false /* is_forward_direction */,
+                                      false /* is_seek_to_first */);
    skip_list_iter_.SeekForPrev(&search_entry);
  }

@@ -85,6 +85,20 @@ Status ReadableWriteBatch::GetEntryFromDataOffset(size_t data_offset,
  return Status::OK();
 }

+// If both of `entry1` and `entry2` point to real entry in write batch, we
+// compare the entries as following:
+// 1. first compare the column family, the one with larger CF will be larger;
+// 2. Inside the same CF, we first decode the entry to find the key of the entry
+//    and the entry with larger key will be larger;
+// 3. If two entries are of the same CF and offset, the one with larger offset
+//    will be larger.
+// Some times either `entry1` or `entry2` is dummy entry, which is actually
+// a search key. In this case, in step 2, we don't go ahead and decode the
+// entry but use the value in WriteBatchIndexEntry::search_key.
+// One special case is WriteBatchIndexEntry::key_size is kFlagMinInCf.
+// This indicate that we are going to seek to the first of the column family.
+// Once we see this, this entry will be smaller than all the real entries of
+// the column family.
 int WriteBatchEntryComparator::operator()(
    const WriteBatchIndexEntry* entry1,
    const WriteBatchIndexEntry* entry2) const {
@@ -94,9 +108,10 @@ int WriteBatchEntryComparator::operator()(
    return -1;
  }

-  if (entry1->offset == WriteBatchIndexEntry::kFlagMin) {
+  // Deal with special case of seeking to the beginning of a column family
+  if (entry1->is_min_in_cf()) {
    return -1;
-  } else if (entry2->offset == WriteBatchIndexEntry::kFlagMin) {
+  } else if (entry2->is_min_in_cf()) {
    return 1;
  }

@@ -31,21 +31,52 @@ struct WriteBatchIndexEntry {
        key_offset(ko),
        key_size(ksz),
        search_key(nullptr) {}
-  WriteBatchIndexEntry(const Slice* sk, uint32_t c)
-      : offset(0),
-        column_family(c),
+  // Create a dummy entry as the search key. This index entry won't be backed
+  // by an entry from the write batch, but a pointer to the search key. Or a
+  // special flag of offset can indicate we are seek to first.
+  // @_search_key: the search key
+  // @_column_family: column family
+  // @is_forward_direction: true for Seek(). False for SeekForPrev()
+  // @is_seek_to_first: true if we seek to the beginning of the column family
+  //                    _search_key should be null in this case.
+  WriteBatchIndexEntry(const Slice* _search_key, uint32_t _column_family,
+                       bool is_forward_direction, bool is_seek_to_first)
+      // For SeekForPrev(), we need to make the dummy entry larger than any
+      // entry who has the same search key. Otherwise, we'll miss those entries.
+      : offset(is_forward_direction ? 0 : port::kMaxSizet),
+        column_family(_column_family),
        key_offset(0),
-        key_size(0),
-        search_key(sk) {}
+        key_size(is_seek_to_first ? kFlagMinInCf : 0),
+        search_key(_search_key) {
+    assert(_search_key != nullptr || is_seek_to_first);
+  }

-  // If this flag appears in the offset, it indicates a key that is smaller
-  // than any other entry for the same column family
-  static const size_t kFlagMin = port::kMaxSizet;
+  // If this flag appears in the key_size, it indicates a
+  // key that is smaller than any other entry for the same column family.
+  static const size_t kFlagMinInCf = port::kMaxSizet;

-  size_t offset;           // offset of an entry in write batch's string buffer.
-  uint32_t column_family;  // column family of the entry.
+  bool is_min_in_cf() const {
+    assert(key_size != kFlagMinInCf ||
+           (key_offset == 0 && search_key == nullptr));
+    return key_size == kFlagMinInCf;
+  }
+
+  // offset of an entry in write batch's string buffer. If this is a dummy
+  // lookup key, in which case search_key != nullptr, offset is set to either
+  // 0 or max, only for comparison purpose. Because when entries have the same
+  // key, the entry with larger offset is larger, offset = 0 will make a seek
+  // key small or equal than all the entries with the seek key, so that Seek()
+  // will find all the entries of the same key. Similarly, offset = MAX will
+  // make the entry just larger than all entries with the search key so
+  // SeekForPrev() will see all the keys with the same key.
+  size_t offset;
+  uint32_t column_family;  // c1olumn family of the entry.
  size_t key_offset;       // offset of the key in write batch's string buffer.
-  size_t key_size;         // size of the key.
+  size_t key_size;         // size of the key. kFlagMinInCf indicates
+                           // that this is a dummy look up entry for
+                           // SeekToFirst() to the beginning of the column
+                           // family. We use the flag here to save a boolean
+                           // in the struct.

  const Slice* search_key;  // if not null, instead of reading keys from
                            // write batch, use it to compare. This is used
@@ -621,7 +621,7 @@ TEST_F(WriteBatchWithIndexTest, TestRandomIteraratorWithBase) {
    for (int i = 0; i < 128; i++) {
      // Random walk and make sure iter and result_iter returns the
      // same key and value
-      int type = rnd.Uniform(5);
+      int type = rnd.Uniform(6);
      ASSERT_OK(iter->status());
      switch (type) {
        case 0:
@@ -642,7 +642,15 @@ TEST_F(WriteBatchWithIndexTest, TestRandomIteraratorWithBase) {
          result_iter->Seek(key);
          break;
        }
-        case 3:
+        case 3: {
+          // SeekForPrev to random key
+          auto key_idx = rnd.Uniform(static_cast<int>(source_strings.size()));
+          auto key = source_strings[key_idx];
+          iter->SeekForPrev(key);
+          result_iter->SeekForPrev(key);
+          break;
+        }
+        case 4:
          // Next
          if (is_valid) {
            iter->Next();
@@ -652,7 +660,7 @@ TEST_F(WriteBatchWithIndexTest, TestRandomIteraratorWithBase) {
          }
          break;
        default:
-          assert(type == 4);
+          assert(type == 5);
          // Prev
          if (is_valid) {
            iter->Prev();
Author	SHA1	Message	Date
Vijay Nadimpalli	764bfcd19b	Making platform 007 (gcc 7) default in build_detect_platform.sh (#5947 ) Summary: Making platform 007 (gcc 7) default in build_detect_platform.sh. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5947 Differential Revision: D18038837 Pulled By: vjnadimpalli fbshipit-source-id: 9ac2ddaa93bf328a416faec028970e039886378e	2019-10-30 10:05:36 -07:00
Andrew Kryczka	1fdb34fe99	Add latest toolchain (gcc-8, etc.) build support for fbcode users (#4923 ) Summary: - When building with internal dependencies, specify this toolchain by setting `ROCKSDB_FBCODE_BUILD_WITH_PLATFORM007=1` - It is not enabled by default. However, it is enabled for TSAN builds in CI since there is a known problem with TSAN in gcc-5: https://gcc.gnu.org/bugzilla/show_bug.cgi?id=71090 - I did not add support for Lua since (1) we agreed to deprecate it, and (2) we only have an internal build for v5.3 with this toolchain while that has breaking changes compared to our current version (v5.2). Pull Request resolved: https://github.com/facebook/rocksdb/pull/4923 Differential Revision: D13827226 Pulled By: ajkr fbshipit-source-id: 9aa3388ed3679777cfb15ef8cbcb83c07f62f947	2019-10-30 10:05:26 -07:00
Yanqin Jin	5d2b9971e8	Bump version to 5.17.4	2018-12-05 19:02:18 -08:00
Yanqin Jin	e8830cab4b	Update HISTORY.md (#4753 ) Summary: As titled. Update history to include a recent bug fix in `9be3e6b488`. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4753 Differential Revision: D13350286 Pulled By: riversand963 fbshipit-source-id: b6324780dee4cb1757bc2209403a08531c150c08	2018-12-05 19:01:06 -08:00
Yanqin Jin	73d59f235e	Allow file-ingest-triggered flush to skip waiting for write-stall clear (#4751 ) Summary: When write stall has already been triggered due to number of L0 files reaching threshold, file ingestion must proceed with its flush without waiting for the write stall condition to cleared by the compaction because compaction can wait for ingestion to finish (circular wait). In order to avoid this wait, we can set `FlushOptions.allow_write_stall` to be true (default is false). Setting it to false can cause deadlock. This can happen when the number of compaction threads is low. Considere the following ``` Time compaction_thread ingestion_thread \| num_running_ingest_file_++ \| while(num_running_ingest_file_>0){wait} \| flush V ``` Pull Request resolved: https://github.com/facebook/rocksdb/pull/4751 Differential Revision: D13343037 Pulled By: riversand963 fbshipit-source-id: d3b95938814af46ec4c463feff0b50c70bd8b23f	2018-12-05 18:17:12 -08:00
Yi Wu	91b2c6d1f4	Bump version to 5.17.3	2018-11-19 22:17:46 -08:00
Yi Wu	f445eadba6	BlobDB: handle IO error on write (#4580 ) Summary: A fix similar to #4410 but on the write path. On IO error on `SelectBlobFile()` we didn't return error code properly, but simply a nullptr of `BlobFile`. The `AppendBlob()` method didn't have null check for the pointer and caused crash. The fix make sure we properly return error code in this case. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4580 Differential Revision: D10513849 Pulled By: yiwu-arbug fbshipit-source-id: 80bca920d1d7a3541149de981015ad83e0aa14b5	2018-11-19 22:17:12 -08:00
Fosco Marotto	f438b98e55	Update history for 5.17.2 release	2018-11-12 11:57:32 -08:00
Yanqin Jin	fb4063749b	Bump version	2018-11-02 15:23:50 -07:00
Yanqin Jin	05c9d53a40	Update TARGET file after changing buck template	2018-11-02 15:21:10 -07:00
Philip Jameson	832ab1424d	Change BUCK template files (#4624 ) Summary: Slightly changes the format of generated BUCK files for Facebook consumption. Generated targets end up looking like this: ``` cpp_library( name = "rocksdb_tools_lib", srcs = [ "tools/db_bench_tool.cc", "tools/trace_analyzer_tool.cc", "util/testutil.cc", ], auto_headers = AutoHeaders.RECURSIVE_GLOB, arch_preprocessor_flags = rocksdb_arch_preprocessor_flags, compiler_flags = rocksdb_compiler_flags, preprocessor_flags = rocksdb_preprocessor_flags, deps = [":rocksdb_lib"], external_deps = rocksdb_external_deps, ) ``` Instead of ``` cpp_library( name = "rocksdb_tools_lib", srcs = [ "tools/db_bench_tool.cc", "tools/trace_analyzer_tool.cc", "util/testutil.cc", ], headers = AutoHeaders.RECURSIVE_GLOB, arch_preprocessor_flags = rocksdb_arch_preprocessor_flags, compiler_flags = rocksdb_compiler_flags, preprocessor_flags = rocksdb_preprocessor_flags, deps = [":rocksdb_lib"], external_deps = rocksdb_external_deps, ) ``` Pull Request resolved: https://github.com/facebook/rocksdb/pull/4624 Reviewed By: riversand963 Differential Revision: D12906711 Pulled By: philipjameson fbshipit-source-id: 32ab64a3390cdcf2c4043ff77517ac1ad58a5e2b	2018-11-02 15:20:35 -07:00
Fosco Marotto	516c7df05d	Revert "Introduce CacheAllocator, a custom allocator for cache blocks (#4437 )" This reverts commit `bb3b2eb960`.	2018-10-31 14:55:11 -07:00
Igor Canadi	bb3b2eb960	Introduce CacheAllocator, a custom allocator for cache blocks (#4437 ) Summary: This is a conceptually simple change, but it touches many files to pass the allocator through function calls. We introduce CacheAllocator, which can be used by clients to configure custom allocator for cache blocks. Our motivation is to hook this up with folly's `JemallocNodumpAllocator` (https://github.com/facebook/folly/blob/f43ce6d6866b7b994b3019df561109afae050ebc/folly/experimental/JemallocNodumpAllocator.h), but there are many other possible use cases. Additionally, this commit cleans up memory allocation in `util/compression.h`, making sure that all allocations are wrapped in a unique_ptr as soon as possible. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4437 Differential Revision: D10132814 Pulled By: yiwu-arbug fbshipit-source-id: be1343a4b69f6048df127939fea9bbc96969f564	2018-10-31 13:02:53 -07:00
Yi Wu	ae5305ee70	Fix compile error with jemalloc (#4488 ) Summary: The "je_" prefix of jemalloc APIs presents only when the macro `JEMALLOC_NO_RENAME` from jemalloc.h presents. With the patch I'm also adding -DROCKSDB_JEMALLOC flag in buck TARGETS. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4488 Differential Revision: D10355971 Pulled By: yiwu-arbug fbshipit-source-id: 03a2d69790a44ac89219c7525763fa937a63d95a	2018-10-30 12:37:06 -07:00
Anand Ananthabhotla	f37ea82191	Update HISTORY.md Summary: Test Plan: Reviewers: Subscribers: Tasks: Tags:	2018-10-19 16:38:03 -07:00
Anand Ananthabhotla	c954445ac8	Handle mixed slowdown/no_slowdown writer properly (#4475 ) Summary: There is a bug when the write queue leader is blocked on a write delay/stop, and the queue has writers with WriteOptions::no_slowdown set to true. They are not woken up until the write stall is cleared. The fix introduces a dummy writer inserted at the tail to indicate a write stall and prevent further inserts into the queue, and a condition variable that writers who can tolerate slowdown wait on before adding themselves to the queue. The leader calls WriteThread::BeginWriteStall() to add the dummy writer and then walk the queue to fail any writers with no_slowdown set. Once the stall clears, the leader calls WriteThread::EndWriteStall() to remove the dummy writer and signal the condition variable. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4475 Differential Revision: D10285827 Pulled By: anand1976 fbshipit-source-id: 747465e5e7f07a829b1fb0bc1afcd7b93f4ab1a9	2018-10-19 16:29:57 -07:00
Siying Dong	619f754816	Fix WriteBatchWithIndex's SeekForPrev() (#4559 ) Summary: WriteBatchWithIndex's SeekForPrev() has a bug that we internally place the position just before the seek key rather than after. This makes the iterator to miss the result that is the same as the seek key. Fix it by position the iterator equal or smaller. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4559 Differential Revision: D10468534 Pulled By: siying fbshipit-source-id: 2fb371ae809c561b60a1c11cef71e1c66fea1f19	2018-10-19 15:31:49 -07:00
Andrew Kryczka	f81fe96eea	update HISTORY.md and version number	2018-10-16 15:43:44 -07:00
anand1976	7d3eb006ff	Properly determine a truncated CompactRange stop key (#4496 ) Summary: When a CompactRange() call for a level is truncated before the end key is reached, because it exceeds max_compaction_bytes, we need to properly set the compaction_end parameter to indicate the stop key. The next CompactRange will use that as the begin key. We set it to the smallest key of the next file in the level after expanding inputs to get a clean cut. Previously, we were setting it before expanding inputs. So we could end up recompacting some files. In a pathological case, where a single key has many entries spanning all the files in the level (possibly due to merge operands without a partial merge operator, thus resulting in compaction output identical to the input), this would result in an endless loop over the same set of files. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4496 Differential Revision: D10395026 Pulled By: anand1976 fbshipit-source-id: f0c2f89fee29b4b3be53b6467b53abba8e9146a9	2018-10-16 15:42:06 -07:00
Andrew Kryczka	02d47c1ff4	Avoid per-key linear scan over snapshots in compaction (#4495 ) Summary: `CompactionIterator::snapshots_` is ordered by ascending seqnum, just like `DBImpl`'s linked list of snapshots from which it was copied. This PR exploits this ordering to make `findEarliestVisibleSnapshot` do binary search rather than linear scan. This can make flush/compaction significantly faster when many snapshots exist since that function is called on every single key. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4495 Differential Revision: D10386470 Pulled By: ajkr fbshipit-source-id: 29734991631227b6b7b677e156ac567690118a8b	2018-10-16 13:11:32 -07:00
Fosco Marotto	0f17fb9622	Update version macro for 5.17 (#4472 ) Summary: Forgot this in previous commit. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4472 Differential Revision: D10244227 Pulled By: gfosco fbshipit-source-id: ba0cf7a2f5271f0d9f9443004e2620887cd5fd11	2018-10-08 16:26:12 -07:00
Fosco Marotto	9552660ecc	Update HISTORY.md to current status (#4471 ) Summary: 5.16.x status wasn't tracked, and also updated for pending 5.17 release. Pull Request resolved: https://github.com/facebook/rocksdb/pull/4471 Differential Revision: D10240925 Pulled By: gfosco fbshipit-source-id: 95ab368a04a65b201d2518097af69edf2402f544	2018-10-08 11:21:55 -07:00
Andrew Gallagher	016437d1cd	rocksdb: put `#pragma once` before `#ifdef` Summary: Work around upstream bug with modules: https://bugs.llvm.org/show_bug.cgi?id=39184. Reviewed By: yiwu-arbug Differential Revision: D10209569 fbshipit-source-id: 696853a02a3869e9c33d0e61168ad4b0436fa3c0	2018-10-05 10:30:11 -07:00
Fosco Marotto	b6b72687c6	Revert "Introduce CacheAllocator, a custom allocator for cache blocks (#4437 )" This reverts commit `1cf5deb8fd`.	2018-10-04 15:17:30 -07:00