table/block.h: change memset

Summary:
In gcc-7 the following is an error identified by -Werror=class-memaccess

In file included from ./table/get_context.h:14:0,
                 from db/version_set.cc:43:
./table/block.h: In constructor ‘rocksdb::BlockReadAmpBitmap::BlockReadAmpBitmap(size_t, size_t, rocksdb::Statistics*)’:
./table/block.h:73:53: error: ‘void* memset(void*, int, size_t)’ clearing an object of type ‘struct std::atomic<unsigned int>’ with no trivial copy-assignment; use value-initialization instead [-Werror=class-memaccess]
     memset(bitmap_, 0, bitmap_size * kBytesPersEntry);
                                                     ^
In file included from ./db/version_set.h:23:0,
                 from db/version_set.cc:12:
/toolchain/include/c++/8.0.0/atomic:684:12: note: ‘struct std::atomic<unsigned int>’ declared here
     struct atomic<unsigned int> : __atomic_base<unsigned int>
            ^~~~~~~~~~~~~~~~~~~~

As a solution the default initializer can be applied in list context.

Signed-off-by: Daniel Black <daniel.black@au.ibm.com>
Closes https://github.com/facebook/rocksdb/pull/2561

Differential Revision: D5398714

Pulled By: siying

fbshipit-source-id: d883fb88ec7535eee60d551038fe91f14488be36

CMakeLists.txt

@@ -477,7 +477,6 @@ set(SOURCES
         utilities/backupable/backupable_db.cc
         utilities/blob_db/blob_db.cc
         utilities/blob_db/blob_db_impl.cc
-        utilities/blob_db/blob_db_options_impl.cc
         utilities/blob_db/blob_dump_tool.cc
         utilities/blob_db/blob_file.cc
         utilities/blob_db/blob_log_reader.cc

HISTORY.md

@@ -1,4 +1,16 @@
 # Rocksdb Change Log
+## 5.7.5 (11/28/2017)
+### Bug Fixes
+* Fix IOError on WAL write doesn't propagate to write group follower
+
+## 5.7.4 (08/31/2017)
+No significant changes.
+
+## 5.7.3 (08/29/2017)
+### Bug Fixes
+* Fix transient reappearance of keys covered by range deletions when memtable prefix bloom filter is enabled.
+* Fix potentially wrong file smallest key when range deletions separated by snapshot are written together.
+
 ## 5.7.2 (08/15/2017)
 ### Bug Fixes
 * Fix incorrect dropping of deletions issue with FIFO compaction.

TARGETS

@@ -207,7 +207,6 @@ cpp_library(
       "utilities/backupable/backupable_db.cc",
       "utilities/blob_db/blob_db.cc",
       "utilities/blob_db/blob_db_impl.cc",
-      "utilities/blob_db/blob_db_options_impl.cc",
       "utilities/blob_db/blob_file.cc",
       "utilities/blob_db/blob_log_reader.cc",
       "utilities/blob_db/blob_log_writer.cc",

build_tools/build_detect_platform

@@ -51,11 +51,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
 

build_tools/dependencies_platform007.sh

@@ -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

build_tools/fbcode_config_platform007.sh

@@ -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

build_tools/rocksdb-lego-determinator

@@ -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"
 DISABLE_JEMALLOC="DISABLE_JEMALLOC=1"
 HTTP_PROXY="https_proxy=http://fwdproxy.29.prn1:8080 http_proxy=http://fwdproxy.29.prn1:8080 ftp_proxy=http://fwdproxy.29.prn1:8080"

build_tools/update_dependencies.sh

@@ -52,6 +52,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                      #
 ###########################################################

db/c.cc

@@ -1372,23 +1372,24 @@ void rocksdb_writebatch_put_log_data(
   b->rep.PutLogData(Slice(blob, len));
 }
 
+class H : public WriteBatch::Handler {
+ public:
+  void* state_;
+  void (*put_)(void*, const char* k, size_t klen, const char* v, size_t vlen);
+  void (*deleted_)(void*, const char* k, size_t klen);
+  virtual void Put(const Slice& key, const Slice& value) override {
+    (*put_)(state_, key.data(), key.size(), value.data(), value.size());
+  }
+  virtual void Delete(const Slice& key) override {
+    (*deleted_)(state_, key.data(), key.size());
+  }
+};
+
 void rocksdb_writebatch_iterate(
     rocksdb_writebatch_t* b,
     void* state,
     void (*put)(void*, const char* k, size_t klen, const char* v, size_t vlen),
     void (*deleted)(void*, const char* k, size_t klen)) {
-  class H : public WriteBatch::Handler {
-   public:
-    void* state_;
-    void (*put_)(void*, const char* k, size_t klen, const char* v, size_t vlen);
-    void (*deleted_)(void*, const char* k, size_t klen);
-    virtual void Put(const Slice& key, const Slice& value) override {
-      (*put_)(state_, key.data(), key.size(), value.data(), value.size());
-    }
-    virtual void Delete(const Slice& key) override {
-      (*deleted_)(state_, key.data(), key.size());
-    }
-  };
   H handler;
   handler.state_ = state;
   handler.put_ = put;
@@ -1633,18 +1634,6 @@ void rocksdb_writebatch_wi_iterate(
     void* state,
     void (*put)(void*, const char* k, size_t klen, const char* v, size_t vlen),
     void (*deleted)(void*, const char* k, size_t klen)) {
-  class H : public WriteBatch::Handler {
-   public:
-    void* state_;
-    void (*put_)(void*, const char* k, size_t klen, const char* v, size_t vlen);
-    void (*deleted_)(void*, const char* k, size_t klen);
-    virtual void Put(const Slice& key, const Slice& value) override {
-      (*put_)(state_, key.data(), key.size(), value.data(), value.size());
-    }
-    virtual void Delete(const Slice& key) override {
-      (*deleted_)(state_, key.data(), key.size());
-    }
-  };
   H handler;
   handler.state_ = state;
   handler.put_ = put;
@@ -2966,20 +2955,21 @@ void rocksdb_slicetransform_destroy(rocksdb_slicetransform_t* st) {
   delete st;
 }
 
+struct Wrapper : public rocksdb_slicetransform_t {
+  const SliceTransform* rep_;
+  ~Wrapper() { delete rep_; }
+  const char* Name() const override { return rep_->Name(); }
+  Slice Transform(const Slice& src) const override {
+    return rep_->Transform(src);
+  }
+  bool InDomain(const Slice& src) const override {
+    return rep_->InDomain(src);
+  }
+  bool InRange(const Slice& src) const override { return rep_->InRange(src); }
+  static void DoNothing(void*) { }
+};
+
 rocksdb_slicetransform_t* rocksdb_slicetransform_create_fixed_prefix(size_t prefixLen) {
-  struct Wrapper : public rocksdb_slicetransform_t {
-    const SliceTransform* rep_;
-    ~Wrapper() { delete rep_; }
-    const char* Name() const override { return rep_->Name(); }
-    Slice Transform(const Slice& src) const override {
-      return rep_->Transform(src);
-    }
-    bool InDomain(const Slice& src) const override {
-      return rep_->InDomain(src);
-    }
-    bool InRange(const Slice& src) const override { return rep_->InRange(src); }
-    static void DoNothing(void*) { }
-  };
   Wrapper* wrapper = new Wrapper;
   wrapper->rep_ = rocksdb::NewFixedPrefixTransform(prefixLen);
   wrapper->state_ = nullptr;
@@ -2988,19 +2978,6 @@ rocksdb_slicetransform_t* rocksdb_slicetransform_create_fixed_prefix(size_t pref
 }
 
 rocksdb_slicetransform_t* rocksdb_slicetransform_create_noop() {
-  struct Wrapper : public rocksdb_slicetransform_t {
-    const SliceTransform* rep_;
-    ~Wrapper() { delete rep_; }
-    const char* Name() const override { return rep_->Name(); }
-    Slice Transform(const Slice& src) const override {
-      return rep_->Transform(src);
-    }
-    bool InDomain(const Slice& src) const override {
-      return rep_->InDomain(src);
-    }
-    bool InRange(const Slice& src) const override { return rep_->InRange(src); }
-    static void DoNothing(void*) { }
-  };
   Wrapper* wrapper = new Wrapper;
   wrapper->rep_ = rocksdb::NewNoopTransform();
   wrapper->state_ = nullptr;

db/db_impl_write.cc

@@ -313,7 +313,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
       versions_->SetLastSequence(last_sequence);
     }
     MemTableInsertStatusCheck(w.status);
-    write_thread_.ExitAsBatchGroupLeader(write_group, w.status);
+    write_thread_.ExitAsBatchGroupLeader(write_group, status);
   }
 
   if (status.ok()) {
@@ -523,7 +523,7 @@ Status DBImpl::WriteImplWALOnly(const WriteOptions& write_options,
   if (!w.CallbackFailed()) {
     WriteCallbackStatusCheck(status);
   }
-  nonmem_write_thread_.ExitAsBatchGroupLeader(write_group, w.status);
+  nonmem_write_thread_.ExitAsBatchGroupLeader(write_group, status);
   if (status.ok()) {
     status = w.FinalStatus();
   }

db/db_range_del_test.cc

@@ -868,6 +868,66 @@ TEST_F(DBRangeDelTest, SubcompactionHasEmptyDedicatedRangeDelFile) {
   db_->ReleaseSnapshot(snapshot);
 }
 
+TEST_F(DBRangeDelTest, MemtableBloomFilter) {
+  // regression test for #2743. the range delete tombstones in memtable should
+  // be added even when Get() skips searching due to its prefix bloom filter
+  const int kMemtableSize = 1 << 20;              // 1MB
+  const int kMemtablePrefixFilterSize = 1 << 13;  // 8KB
+  const int kNumKeys = 1000;
+  const int kPrefixLen = 8;
+  Options options = CurrentOptions();
+  options.memtable_prefix_bloom_size_ratio =
+      static_cast<double>(kMemtablePrefixFilterSize) / kMemtableSize;
+  options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(kPrefixLen));
+  options.write_buffer_size = kMemtableSize;
+  Reopen(options);
+
+  for (int i = 0; i < kNumKeys; ++i) {
+    ASSERT_OK(Put(Key(i), "val"));
+  }
+  Flush();
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), Key(0),
+                             Key(kNumKeys)));
+  for (int i = 0; i < kNumKeys; ++i) {
+    std::string value;
+    ASSERT_TRUE(db_->Get(ReadOptions(), Key(i), &value).IsNotFound());
+  }
+}
+
+TEST_F(DBRangeDelTest, UnorderedTombstones) {
+  // Regression test for #2752. Range delete tombstones between
+  // different snapshot stripes are not stored in order, so the first
+  // tombstone of each snapshot stripe should be checked as a smallest
+  // candidate.
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+
+  auto cf = db_->DefaultColumnFamily();
+
+  ASSERT_OK(db_->Put(WriteOptions(), cf, "a", "a"));
+  ASSERT_OK(db_->Flush(FlushOptions(), cf));
+  ASSERT_EQ(1, NumTableFilesAtLevel(0));
+  ASSERT_OK(dbfull()->TEST_CompactRange(0, nullptr, nullptr));
+  ASSERT_EQ(1, NumTableFilesAtLevel(1));
+
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), cf, "b", "c"));
+  // Hold a snapshot to separate these two delete ranges.
+  auto snapshot = db_->GetSnapshot();
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), cf, "a", "b"));
+  ASSERT_OK(db_->Flush(FlushOptions(), cf));
+  db_->ReleaseSnapshot(snapshot);
+
+  std::vector<std::vector<FileMetaData>> files;
+  dbfull()->TEST_GetFilesMetaData(cf, &files);
+  ASSERT_EQ(1, files[0].size());
+  ASSERT_EQ("a", files[0][0].smallest.user_key());
+  ASSERT_EQ("c", files[0][0].largest.user_key());
+
+  std::string v;
+  auto s = db_->Get(ReadOptions(), "a", &v);
+  ASSERT_TRUE(s.IsNotFound());
+}
+
 #endif  // ROCKSDB_LITE
 
 }  // namespace rocksdb

db/db_write_test.cc

@@ -3,12 +3,17 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 
+#include <atomic>
 #include <memory>
 #include <thread>
 #include <vector>
 #include "db/db_test_util.h"
 #include "db/write_batch_internal.h"
+#include "db/write_thread.h"
+#include "port/port.h"
 #include "port/stack_trace.h"
+#include "util/fault_injection_test_env.h"
+#include "util/string_util.h"
 #include "util/sync_point.h"
 
 namespace rocksdb {
@@ -18,7 +23,9 @@ class DBWriteTest : public DBTestBase, public testing::WithParamInterface<int> {
  public:
   DBWriteTest() : DBTestBase("/db_write_test") {}
 
-  void Open() { DBTestBase::Reopen(GetOptions(GetParam())); }
+  Options GetOptions() { return DBTestBase::GetOptions(GetParam()); }
+
+  void Open() { DBTestBase::Reopen(GetOptions()); }
 };
 
 // Sequence number should be return through input write batch.
@@ -67,6 +74,47 @@ TEST_P(DBWriteTest, ReturnSeuqneceNumberMultiThreaded) {
   }
 }
 
+TEST_P(DBWriteTest, IOErrorOnWALWritePropagateToWriteThreadFollower) {
+  constexpr int kNumThreads = 5;
+  std::unique_ptr<FaultInjectionTestEnv> mock_env(
+      new FaultInjectionTestEnv(Env::Default()));
+  Options options = GetOptions();
+  options.env = mock_env.get();
+  Reopen(options);
+  std::atomic<int> ready_count{0};
+  std::atomic<int> leader_count{0};
+  std::vector<port::Thread> threads;
+  mock_env->SetFilesystemActive(false);
+  // Wait until all threads linked to write threads, to make sure
+  // all threads join the same batch group.
+  SyncPoint::GetInstance()->SetCallBack(
+      "WriteThread::JoinBatchGroup:Wait", [&](void* arg) {
+        ready_count++;
+        auto* w = reinterpret_cast<WriteThread::Writer*>(arg);
+        if (w->state == WriteThread::STATE_GROUP_LEADER) {
+          leader_count++;
+          while (ready_count < kNumThreads) {
+            // busy waiting
+          }
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+  for (int i = 0; i < kNumThreads; i++) {
+    threads.push_back(port::Thread(
+        [&](int index) {
+          // All threads should fail.
+          ASSERT_FALSE(Put("key" + ToString(index), "value").ok());
+        },
+        i));
+  }
+  for (int i = 0; i < kNumThreads; i++) {
+    threads[i].join();
+  }
+  ASSERT_EQ(1, leader_count);
+  // Close before mock_env destruct.
+  Close();
+}
+
 INSTANTIATE_TEST_CASE_P(DBWriteTestInstance, DBWriteTest,
                         testing::Values(DBTestBase::kDefault,
                                         DBTestBase::kConcurrentWALWrites,

db/memtable.cc

@@ -643,6 +643,14 @@ bool MemTable::Get(const LookupKey& key, std::string* value, Status* s,
   }
   PERF_TIMER_GUARD(get_from_memtable_time);
 
+  std::unique_ptr<InternalIterator> range_del_iter(
+      NewRangeTombstoneIterator(read_opts));
+  Status status = range_del_agg->AddTombstones(std::move(range_del_iter));
+  if (!status.ok()) {
+    *s = status;
+    return false;
+  }
+
   Slice user_key = key.user_key();
   bool found_final_value = false;
   bool merge_in_progress = s->IsMergeInProgress();
@@ -658,13 +666,6 @@ bool MemTable::Get(const LookupKey& key, std::string* value, Status* s,
     if (prefix_bloom_) {
       PERF_COUNTER_ADD(bloom_memtable_hit_count, 1);
     }
-    std::unique_ptr<InternalIterator> range_del_iter(
-        NewRangeTombstoneIterator(read_opts));
-    Status status = range_del_agg->AddTombstones(std::move(range_del_iter));
-    if (!status.ok()) {
-      *s = status;
-      return false;
-    }
     Saver saver;
     saver.status = s;
     saver.found_final_value = &found_final_value;

db/range_del_aggregator.cc

@@ -412,8 +412,8 @@ void RangeDelAggregator::AddToBuilder(
 
   // Note the order in which tombstones are stored is insignificant since we
   // insert them into a std::map on the read path.
-  bool first_added = false;
   while (stripe_map_iter != rep_->stripe_map_.end()) {
+    bool first_added = false;
     for (auto tombstone_map_iter = stripe_map_iter->second.raw_map.begin();
          tombstone_map_iter != stripe_map_iter->second.raw_map.end();
          ++tombstone_map_iter) {
@@ -452,7 +452,7 @@ void RangeDelAggregator::AddToBuilder(
       builder->Add(ikey_and_end_key.first.Encode(), ikey_and_end_key.second);
       if (!first_added) {
         first_added = true;
-        InternalKey smallest_candidate = std::move(ikey_and_end_key.first);;
+        InternalKey smallest_candidate = std::move(ikey_and_end_key.first);
         if (lower_bound != nullptr &&
             icmp_.user_comparator()->Compare(smallest_candidate.user_key(),
                                              *lower_bound) <= 0) {

db/write_callback_test.cc

@@ -134,6 +134,11 @@ TEST_F(WriteCallbackTest, WriteWithCallbackTest) {
               options.allow_concurrent_memtable_write = allow_parallel;
               options.enable_pipelined_write = enable_pipelined_write;
               options.concurrent_prepare = two_queues;
+              if (options.enable_pipelined_write &&
+                  options.concurrent_prepare) {
+                // This combination is not supported
+                continue;
+              }
 
               ReadOptions read_options;
               DB* db;

db/write_thread.cc

@@ -511,6 +511,11 @@ void WriteThread::ExitAsBatchGroupLeader(WriteGroup& write_group,
   Writer* last_writer = write_group.last_writer;
   assert(leader->link_older == nullptr);
 
+  // Propagate memtable write error to the whole group.
+  if (status.ok() && !write_group.status.ok()) {
+    status = write_group.status;
+  }
+
   if (enable_pipelined_write_) {
     // Notify writers don't write to memtable to exit.
     for (Writer* w = last_writer; w != leader;) {

include/rocksdb/version.h

@@ -6,7 +6,7 @@
 
 #define ROCKSDB_MAJOR 5
 #define ROCKSDB_MINOR 7
-#define ROCKSDB_PATCH 2
+#define ROCKSDB_PATCH 5
 
 // 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

memtable/inlineskiplist.h

@@ -278,7 +278,7 @@ struct InlineSkipList<Comparator>::Node {
   // next_[0].  This is used for passing data from AllocateKey to Insert.
   void StashHeight(const int height) {
     assert(sizeof(int) <= sizeof(next_[0]));
-    memcpy(&next_[0], &height, sizeof(int));
+    memcpy(static_cast<void*>(&next_[0]), &height, sizeof(int));
   }
 
   // Retrieves the value passed to StashHeight.  Undefined after a call
@@ -298,30 +298,30 @@ struct InlineSkipList<Comparator>::Node {
     assert(n >= 0);
     // Use an 'acquire load' so that we observe a fully initialized
     // version of the returned Node.
-    return (next_[-n].load(std::memory_order_acquire));
+    return ((&next_[0] - n)->load(std::memory_order_acquire));
   }
 
   void SetNext(int n, Node* x) {
     assert(n >= 0);
     // Use a 'release store' so that anybody who reads through this
     // pointer observes a fully initialized version of the inserted node.
-    next_[-n].store(x, std::memory_order_release);
+    (&next_[0] - n)->store(x, std::memory_order_release);
   }
 
   bool CASNext(int n, Node* expected, Node* x) {
     assert(n >= 0);
-    return next_[-n].compare_exchange_strong(expected, x);
+    return (&next_[0] - n)->compare_exchange_strong(expected, x);
   }
 
   // No-barrier variants that can be safely used in a few locations.
   Node* NoBarrier_Next(int n) {
     assert(n >= 0);
-    return next_[-n].load(std::memory_order_relaxed);
+    return (&next_[0] - n)->load(std::memory_order_relaxed);
   }
 
   void NoBarrier_SetNext(int n, Node* x) {
     assert(n >= 0);
-    next_[-n].store(x, std::memory_order_relaxed);
+    (&next_[0] - n)->store(x, std::memory_order_relaxed);
   }
 
   // Insert node after prev on specific level.

options/options.cc

@@ -191,7 +191,9 @@ DBOptions::DBOptions(const Options& options)
       dump_malloc_stats(options.dump_malloc_stats),
       avoid_flush_during_recovery(options.avoid_flush_during_recovery),
       avoid_flush_during_shutdown(options.avoid_flush_during_shutdown),
-      allow_ingest_behind(options.allow_ingest_behind) {
+      allow_ingest_behind(options.allow_ingest_behind),
+      concurrent_prepare(options.concurrent_prepare),
+      manual_wal_flush(options.manual_wal_flush) {
 }
 
 void DBOptions::Dump(Logger* log) const {

src.mk

@@ -154,7 +154,6 @@ LIB_SOURCES =                                                   \
   utilities/backupable/backupable_db.cc                         \
   utilities/blob_db/blob_db.cc                                  \
   utilities/blob_db/blob_db_impl.cc                             \
-  utilities/blob_db/blob_db_options_impl.cc                     \
   utilities/blob_db/blob_file.cc                                \
   utilities/blob_db/blob_log_reader.cc                          \
   utilities/blob_db/blob_log_writer.cc                          \

table/block.h

@@ -67,8 +67,7 @@ class BlockReadAmpBitmap {
     size_t bitmap_size = (num_bits_needed - 1) / kBitsPerEntry + 1;
 
     // Create bitmap and set all the bits to 0
-    bitmap_ = new std::atomic<uint32_t>[bitmap_size];
-    memset(bitmap_, 0, bitmap_size * kBytesPersEntry);
+    bitmap_ = new std::atomic<uint32_t>[bitmap_size]();
 
     RecordTick(GetStatistics(), READ_AMP_TOTAL_READ_BYTES, block_size);
   }

tools/db_bench_tool.cc

@@ -3307,10 +3307,7 @@ void VerifyDBFromDB(std::string& truth_db_name) {
     } else if (FLAGS_use_blob_db) {
       blob_db::BlobDBOptions blob_db_options;
       blob_db::BlobDB* ptr;
-      s = CreateLoggerFromOptions(db_name, options, &options.info_log);
-      if (s.ok()) {
-        s = blob_db::BlobDB::Open(options, blob_db_options, db_name, &ptr);
-      }
+      s = blob_db::BlobDB::Open(options, blob_db_options, db_name, &ptr);
       if (s.ok()) {
         db->db = ptr;
       }

utilities/blob_db/blob_db.cc

@@ -5,7 +5,14 @@
 //
 #ifndef ROCKSDB_LITE
 
+#ifndef __STDC_FORMAT_MACROS
+#define __STDC_FORMAT_MACROS
+#endif
+
 #include "utilities/blob_db/blob_db.h"
+
+#include <inttypes.h>
+
 #include "db/write_batch_internal.h"
 #include "monitoring/instrumented_mutex.h"
 #include "options/cf_options.h"
@@ -94,23 +101,31 @@ Status BlobDB::Open(const Options& options, const BlobDBOptions& bdb_options,
   return s;
 }
 
-Status BlobDB::Open(const DBOptions& db_options,
+Status BlobDB::Open(const DBOptions& db_options_input,
                     const BlobDBOptions& bdb_options, const std::string& dbname,
                     const std::vector<ColumnFamilyDescriptor>& column_families,
                     std::vector<ColumnFamilyHandle*>* handles, BlobDB** blob_db,
                     bool no_base_db) {
   *blob_db = nullptr;
+  Status s;
+
+  DBOptions db_options(db_options_input);
+  if (db_options.info_log == nullptr) {
+    s = CreateLoggerFromOptions(dbname, db_options, &db_options.info_log);
+    if (!s.ok()) {
+      return s;
+    }
+  }
 
-  DBOptions my_db_options(db_options);
   FlushBeginListener_t fblistener =
       std::make_shared<BlobDBFlushBeginListener>();
   CompactionListener_t ce_listener =
       std::make_shared<EvictAllVersionsCompactionListener>();
   ReconcileWalFilter_t rw_filter = std::make_shared<BlobReconcileWalFilter>();
 
-  my_db_options.listeners.emplace_back(fblistener);
-  my_db_options.listeners.emplace_back(ce_listener);
-  my_db_options.wal_filter = rw_filter.get();
+  db_options.listeners.emplace_back(fblistener);
+  db_options.listeners.emplace_back(ce_listener);
+  db_options.wal_filter = rw_filter.get();
 
   {
     MutexLock l(&listener_mutex);
@@ -120,19 +135,25 @@ Status BlobDB::Open(const DBOptions& db_options,
   }
 
   // we need to open blob db first so that recovery can happen
-  BlobDBImpl* bdb = new BlobDBImpl(dbname, bdb_options, my_db_options);
+  BlobDBImpl* bdb = new BlobDBImpl(dbname, bdb_options, db_options);
   fblistener->SetImplPtr(bdb);
   ce_listener->SetImplPtr(bdb);
   rw_filter->SetImplPtr(bdb);
 
-  Status s = bdb->OpenPhase1();
-  if (!s.ok()) return s;
+  s = bdb->OpenPhase1();
+  if (!s.ok()) {
+    return s;
+  }
 
-  if (no_base_db) return s;
+  if (no_base_db) {
+    return s;
+  }
 
   DB* db = nullptr;
-  s = DB::Open(my_db_options, dbname, column_families, handles, &db);
-  if (!s.ok()) return s;
+  s = DB::Open(db_options, dbname, column_families, handles, &db);
+  if (!s.ok()) {
+    return s;
+  }
 
   // set the implementation pointer
   s = bdb->LinkToBaseDB(db);
@@ -141,28 +162,36 @@ Status BlobDB::Open(const DBOptions& db_options,
     bdb = nullptr;
   }
   *blob_db = bdb;
+  bdb_options.Dump(db_options.info_log.get());
   return s;
 }
 
 BlobDB::BlobDB(DB* db) : StackableDB(db) {}
 
-////////////////////////////////////////////////////////////////////////////////
-//
-//
-// std::function<int(double)> fnCaller =
-//     std::bind(&A::fn, &anInstance, std::placeholders::_1);
-////////////////////////////////////////////////////////////////////////////////
-BlobDBOptions::BlobDBOptions()
-    : blob_dir("blob_dir"),
-      path_relative(true),
-      is_fifo(false),
-      blob_dir_size(1000ULL * 1024ULL * 1024ULL * 1024ULL),
-      ttl_range_secs(3600),
-      min_blob_size(512),
-      bytes_per_sync(0),
-      blob_file_size(256 * 1024 * 1024),
-      num_concurrent_simple_blobs(4),
-      compression(kNoCompression) {}
+void BlobDBOptions::Dump(Logger* log) const {
+  ROCKS_LOG_HEADER(log, "                   blob_db_options.blob_dir: %s",
+                   blob_dir.c_str());
+  ROCKS_LOG_HEADER(log, "              blob_db_options.path_relative: %d",
+                   path_relative);
+  ROCKS_LOG_HEADER(log, "                    blob_db_options.is_fifo: %d",
+                   is_fifo);
+  ROCKS_LOG_HEADER(log, "              blob_db_options.blob_dir_size: %" PRIu64,
+                   blob_dir_size);
+  ROCKS_LOG_HEADER(log, "             blob_db_options.ttl_range_secs: %" PRIu32,
+                   ttl_range_secs);
+  ROCKS_LOG_HEADER(log, "             blob_db_options.bytes_per_sync: %" PRIu64,
+                   bytes_per_sync);
+  ROCKS_LOG_HEADER(log, "             blob_db_options.blob_file_size: %" PRIu64,
+                   blob_file_size);
+  ROCKS_LOG_HEADER(log, "blob_db_options.num_concurrent_simple_blobs: %" PRIu32,
+                   num_concurrent_simple_blobs);
+  ROCKS_LOG_HEADER(log, "              blob_db_options.ttl_extractor: %p",
+                   ttl_extractor.get());
+  ROCKS_LOG_HEADER(log, "                blob_db_options.compression: %d",
+                   static_cast<int>(compression));
+  ROCKS_LOG_HEADER(log, "   blob_db_options.disable_background_tasks: %d",
+                   disable_background_tasks);
+}
 
 }  // namespace blob_db
 }  // namespace rocksdb

utilities/blob_db/blob_db.h

@@ -31,18 +31,18 @@ class TTLExtractor;
 struct BlobDBOptions {
   // name of the directory under main db, where blobs will be stored.
   // default is "blob_dir"
-  std::string blob_dir;
+  std::string blob_dir = "blob_dir";
 
   // whether the blob_dir path is relative or absolute.
-  bool path_relative;
+  bool path_relative = true;
 
   // is the eviction strategy fifo based
-  bool is_fifo;
+  bool is_fifo = false;
 
   // maximum size of the blob dir. Once this gets used, up
   // evict the blob file which is oldest (is_fifo )
   // 0 means no limits
-  uint64_t blob_dir_size;
+  uint64_t blob_dir_size = 0;
 
   // a new bucket is opened, for ttl_range. So if ttl_range is 600seconds
   // (10 minutes), and the first bucket starts at 1471542000
@@ -50,26 +50,22 @@ struct BlobDBOptions {
   // first bucket is 1471542000 - 1471542600
   // second bucket is 1471542600 - 1471543200
   // and so on
-  uint32_t ttl_range_secs;
-
-  // at what size will the blobs be stored in separate log rather than
-  // inline
-  uint64_t min_blob_size;
+  uint64_t ttl_range_secs = 3600;
 
   // at what bytes will the blob files be synced to blob log.
-  uint64_t bytes_per_sync;
+  uint64_t bytes_per_sync = 0;
 
   // the target size of each blob file. File will become immutable
   // after it exceeds that size
-  uint64_t blob_file_size;
+  uint64_t blob_file_size = 256 * 1024 * 1024;
 
   // how many files to use for simple blobs at one time
-  uint32_t num_concurrent_simple_blobs;
+  uint32_t num_concurrent_simple_blobs = 1;
 
   // Instead of setting TTL explicitly by calling PutWithTTL or PutUntil,
   // applications can set a TTLExtractor which can extract TTL from key-value
   // pairs.
-  std::shared_ptr<TTLExtractor> ttl_extractor;
+  std::shared_ptr<TTLExtractor> ttl_extractor = nullptr;
 
   // eviction callback.
   // this function will be called for every blob that is getting
@@ -78,14 +74,12 @@ struct BlobDBOptions {
       gc_evict_cb_fn;
 
   // what compression to use for Blob's
-  CompressionType compression;
+  CompressionType compression = kNoCompression;
 
-  // default constructor
-  BlobDBOptions();
+  // Disable all background job.
+  bool disable_background_tasks = false;
 
-  BlobDBOptions(const BlobDBOptions& in) = default;
-
-  virtual ~BlobDBOptions() = default;
+  void Dump(Logger* log) const;
 };
 
 class BlobDB : public StackableDB {
@@ -103,21 +97,21 @@ class BlobDB : public StackableDB {
 
   virtual Status PutWithTTL(const WriteOptions& options,
                             ColumnFamilyHandle* column_family, const Slice& key,
-                            const Slice& value, int32_t ttl) = 0;
+                            const Slice& value, uint64_t ttl) = 0;
 
   virtual Status PutWithTTL(const WriteOptions& options, const Slice& key,
-                            const Slice& value, int32_t ttl) {
+                            const Slice& value, uint64_t ttl) {
     return PutWithTTL(options, DefaultColumnFamily(), key, value, ttl);
   }
 
-  // Put with expiration. Key with expiration time equal to -1
-  // means the key don't expire.
+  // Put with expiration. Key with expiration time equal to
+  // std::numeric_limits<uint64_t>::max() means the key don't expire.
   virtual Status PutUntil(const WriteOptions& options,
                           ColumnFamilyHandle* column_family, const Slice& key,
-                          const Slice& value, int32_t expiration) = 0;
+                          const Slice& value, uint64_t expiration) = 0;
 
   virtual Status PutUntil(const WriteOptions& options, const Slice& key,
-                          const Slice& value, int32_t expiration) {
+                          const Slice& value, uint64_t expiration) {
     return PutUntil(options, DefaultColumnFamily(), key, value, expiration);
   }
 

utilities/blob_db/blob_db_impl.h

@@ -29,7 +29,6 @@
 #include "util/mutexlock.h"
 #include "util/timer_queue.h"
 #include "utilities/blob_db/blob_db.h"
-#include "utilities/blob_db/blob_db_options_impl.h"
 #include "utilities/blob_db/blob_log_format.h"
 #include "utilities/blob_db/blob_log_reader.h"
 #include "utilities/blob_db/blob_log_writer.h"
@@ -138,10 +137,13 @@ struct GCStats {
   uint64_t blob_count = 0;
   uint64_t num_deletes = 0;
   uint64_t deleted_size = 0;
-  uint64_t num_relocs = 0;
-  uint64_t succ_deletes_lsm = 0;
-  uint64_t overrided_while_delete = 0;
-  uint64_t succ_relocs = 0;
+  uint64_t retry_delete = 0;
+  uint64_t delete_succeeded = 0;
+  uint64_t overwritten_while_delete = 0;
+  uint64_t num_relocate = 0;
+  uint64_t retry_relocate = 0;
+  uint64_t relocate_succeeded = 0;
+  uint64_t overwritten_while_relocate = 0;
   std::shared_ptr<BlobFile> newfile = nullptr;
 };
 
@@ -158,8 +160,50 @@ class BlobDBImpl : public BlobDB {
   friend class BlobDBIterator;
 
  public:
-  static constexpr uint64_t kNoExpiration =
-      std::numeric_limits<uint64_t>::max();
+  // deletions check period
+  static constexpr uint32_t kDeleteCheckPeriodMillisecs = 2 * 1000;
+
+  // gc percentage each check period
+  static constexpr uint32_t kGCFilePercentage = 100;
+
+  // gc period
+  static constexpr uint32_t kGCCheckPeriodMillisecs = 60 * 1000;
+
+  // sanity check task
+  static constexpr uint32_t kSanityCheckPeriodMillisecs = 20 * 60 * 1000;
+
+  // how many random access open files can we tolerate
+  static constexpr uint32_t kOpenFilesTrigger = 100;
+
+  // how many periods of stats do we keep.
+  static constexpr uint32_t kWriteAmplificationStatsPeriods = 24;
+
+  // what is the length of any period
+  static constexpr uint32_t kWriteAmplificationStatsPeriodMillisecs =
+      3600 * 1000;
+
+  // we will garbage collect blob files in
+  // which entire files have expired. However if the
+  // ttl_range of files is very large say a day, we
+  // would have to wait for the entire day, before we
+  // recover most of the space.
+  static constexpr uint32_t kPartialExpirationGCRangeSecs = 4 * 3600;
+
+  // this should be based on allowed Write Amplification
+  // if 50% of the space of a blob file has been deleted/expired,
+  static constexpr uint32_t kPartialExpirationPercentage = 75;
+
+  // how often should we schedule a job to fsync open files
+  static constexpr uint32_t kFSyncFilesPeriodMillisecs = 10 * 1000;
+
+  // how often to schedule reclaim open files.
+  static constexpr uint32_t kReclaimOpenFilesPeriodMillisecs = 1 * 1000;
+
+  // how often to schedule delete obs files periods
+  static constexpr uint32_t kDeleteObsoleteFilesPeriodMillisecs = 10 * 1000;
+
+  // how often to schedule check seq files period
+  static constexpr uint32_t kCheckSeqFilesPeriodMillisecs = 10 * 1000;
 
   using rocksdb::StackableDB::Put;
   Status Put(const WriteOptions& options, ColumnFamilyHandle* column_family,
@@ -175,16 +219,16 @@ class BlobDBImpl : public BlobDB {
                               const Slice& key) override;
 
   using rocksdb::StackableDB::Get;
-  Status Get(const ReadOptions& options, ColumnFamilyHandle* column_family,
+  Status Get(const ReadOptions& read_options, ColumnFamilyHandle* column_family,
              const Slice& key, PinnableSlice* value) override;
 
   using rocksdb::StackableDB::NewIterator;
-  virtual Iterator* NewIterator(const ReadOptions& opts,
+  virtual Iterator* NewIterator(const ReadOptions& read_options,
                                 ColumnFamilyHandle* column_family) override;
 
   using rocksdb::StackableDB::MultiGet;
   virtual std::vector<Status> MultiGet(
-      const ReadOptions& options,
+      const ReadOptions& read_options,
       const std::vector<ColumnFamilyHandle*>& column_family,
       const std::vector<Slice>& keys,
       std::vector<std::string>* values) override;
@@ -194,12 +238,12 @@ class BlobDBImpl : public BlobDB {
   using BlobDB::PutWithTTL;
   Status PutWithTTL(const WriteOptions& options,
                     ColumnFamilyHandle* column_family, const Slice& key,
-                    const Slice& value, int32_t ttl) override;
+                    const Slice& value, uint64_t ttl) override;
 
   using BlobDB::PutUntil;
   Status PutUntil(const WriteOptions& options,
                   ColumnFamilyHandle* column_family, const Slice& key,
-                  const Slice& value_unc, int32_t expiration) override;
+                  const Slice& value_unc, uint64_t expiration) override;
 
   Status LinkToBaseDB(DB* db) override;
 
@@ -217,15 +261,27 @@ class BlobDBImpl : public BlobDB {
 
   std::vector<std::shared_ptr<BlobFile>> TEST_GetBlobFiles() const;
 
-  void TEST_CloseBlobFile(std::shared_ptr<BlobFile>& bfile);
+  std::vector<std::shared_ptr<BlobFile>> TEST_GetObsoleteFiles() const;
+
+  Status TEST_CloseBlobFile(std::shared_ptr<BlobFile>& bfile);
 
   Status TEST_GCFileAndUpdateLSM(std::shared_ptr<BlobFile>& bfile,
                                  GCStats* gc_stats);
+
+  void TEST_RunGC();
+
+  void TEST_ObsoleteFile(std::shared_ptr<BlobFile>& bfile);
+
+  void TEST_DeleteObsoleteFiles();
 #endif  //  !NDEBUG
 
  private:
   Status OpenPhase1();
 
+  // Create a snapshot if there isn't one in read options.
+  // Return true if a snapshot is created.
+  bool SetSnapshotIfNeeded(ReadOptions* read_options);
+
   Status CommonGet(const ColumnFamilyData* cfd, const Slice& key,
                    const std::string& index_entry, std::string* value,
                    SequenceNumber* sequence = nullptr);
@@ -237,26 +293,24 @@ class BlobDBImpl : public BlobDB {
   // this handler is called.
   void OnFlushBeginHandler(DB* db, const FlushJobInfo& info);
 
-  // timer queue callback to close a file by appending a footer
-  // removes file from open files list
-  std::pair<bool, int64_t> CloseSeqWrite(std::shared_ptr<BlobFile> bfile,
-                                         bool aborted);
-
   // is this file ready for Garbage collection. if the TTL of the file
   // has expired or if threshold of the file has been evicted
   // tt - current time
   // last_id - the id of the non-TTL file to evict
-  bool ShouldGCFile(std::shared_ptr<BlobFile> bfile, std::time_t tt,
-                    uint64_t last_id, std::string* reason);
+  bool ShouldGCFile(std::shared_ptr<BlobFile> bfile, uint64_t now,
+                    bool is_oldest_simple_blob_file, std::string* reason);
 
   // collect all the blob log files from the blob directory
   Status GetAllLogFiles(std::set<std::pair<uint64_t, std::string>>* file_nums);
 
-  // appends a task into timer queue to close the file
-  void CloseIf(const std::shared_ptr<BlobFile>& bfile);
+  // Close a file by appending a footer, and removes file from open files list.
+  Status CloseBlobFile(std::shared_ptr<BlobFile> bfile);
 
-  int32_t ExtractExpiration(const Slice& key, const Slice& value,
-                            Slice* value_slice, std::string* new_value);
+  // Close a file if its size exceeds blob_file_size
+  Status CloseBlobFileIfNeeded(std::shared_ptr<BlobFile>& bfile);
+
+  uint64_t ExtractExpiration(const Slice& key, const Slice& value,
+                             Slice* value_slice, std::string* new_value);
 
   Status AppendBlob(const std::shared_ptr<BlobFile>& bfile,
                     const std::string& headerbuf, const Slice& key,
@@ -267,12 +321,12 @@ 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(uint32_t expiration);
+  std::shared_ptr<BlobFile> SelectBlobFileTTL(uint64_t expiration);
 
   // find an existing blob log file to append the value to
   std::shared_ptr<BlobFile> SelectBlobFile();
 
-  std::shared_ptr<BlobFile> FindBlobFileLocked(uint32_t expiration) const;
+  std::shared_ptr<BlobFile> FindBlobFileLocked(uint64_t expiration) const;
 
   void UpdateWriteOptions(const WriteOptions& options);
 
@@ -284,7 +338,7 @@ class BlobDBImpl : public BlobDB {
   // delete files which have been garbage collected and marked
   // obsolete. Check whether any snapshots exist which refer to
   // the same
-  std::pair<bool, int64_t> DeleteObsFiles(bool aborted);
+  std::pair<bool, int64_t> DeleteObsoleteFiles(bool aborted);
 
   // Major task to garbage collect expired and deleted blobs
   std::pair<bool, int64_t> RunGC(bool aborted);
@@ -359,13 +413,12 @@ class BlobDBImpl : public BlobDB {
   bool FindFileAndEvictABlob(uint64_t file_number, uint64_t key_size,
                              uint64_t blob_offset, uint64_t blob_size);
 
-  void CopyBlobFiles(std::vector<std::shared_ptr<BlobFile>>* bfiles_copy,
-                     uint64_t* last_id);
+  void CopyBlobFiles(std::vector<std::shared_ptr<BlobFile>>* bfiles_copy);
 
   void FilterSubsetOfFiles(
       const std::vector<std::shared_ptr<BlobFile>>& blob_files,
       std::vector<std::shared_ptr<BlobFile>>* to_process, uint64_t epoch,
-      uint64_t last_id, size_t files_to_collect);
+      size_t files_to_collect);
 
   uint64_t EpochNow() { return env_->NowMicros() / 1000000; }
 
@@ -383,7 +436,7 @@ class BlobDBImpl : public BlobDB {
   WriteOptions write_options_;
 
   // the options that govern the behavior of Blob Storage
-  BlobDBOptionsImpl bdb_options_;
+  BlobDBOptions bdb_options_;
   DBOptions db_options_;
   EnvOptions env_options_;
 
@@ -401,18 +454,21 @@ class BlobDBImpl : public BlobDB {
 
   // Read Write Mutex, which protects all the data structures
   // HEAVILY TRAFFICKED
-  port::RWMutex mutex_;
+  mutable port::RWMutex mutex_;
+
+  // Writers has to hold write_mutex_ before writing.
+  mutable port::Mutex write_mutex_;
 
   // counter for blob file number
   std::atomic<uint64_t> next_file_number_;
 
   // entire metadata of all the BLOB files memory
-  std::unordered_map<uint64_t, std::shared_ptr<BlobFile>> blob_files_;
+  std::map<uint64_t, std::shared_ptr<BlobFile>> blob_files_;
 
   // epoch or version of the open files.
   std::atomic<uint64_t> epoch_of_;
 
-  // typically we keep 4 open blob files (simple i.e. no TTL)
+  // All opened non-TTL blob files.
   std::vector<std::shared_ptr<BlobFile>> open_simple_files_;
 
   // all the blob files which are currently being appended to based
@@ -543,7 +599,7 @@ class BlobFile {
 
   // This Read-Write mutex is per file specific and protects
   // all the datastructures
-  port::RWMutex mutex_;
+  mutable port::RWMutex mutex_;
 
   // time when the random access reader was last created.
   std::atomic<std::time_t> last_access_;
@@ -650,12 +706,23 @@ class BlobFile {
 class BlobDBIterator : public Iterator {
  public:
   explicit BlobDBIterator(Iterator* iter, ColumnFamilyHandle* column_family,
-                          BlobDBImpl* impl)
-      : iter_(iter), cfh_(column_family), db_impl_(impl) {
-    assert(iter_);
+                          BlobDBImpl* impl, bool own_snapshot,
+                          const Snapshot* snapshot)
+      : iter_(iter),
+        cfh_(column_family),
+        db_impl_(impl),
+        own_snapshot_(own_snapshot),
+        snapshot_(snapshot) {
+    assert(iter != nullptr);
+    assert(snapshot != nullptr);
   }
 
-  ~BlobDBIterator() { delete iter_; }
+  ~BlobDBIterator() {
+    if (own_snapshot_) {
+      db_impl_->ReleaseSnapshot(snapshot_);
+    }
+    delete iter_;
+  }
 
   bool Valid() const override { return iter_->Valid(); }
 
@@ -677,10 +744,14 @@ class BlobDBIterator : public Iterator {
 
   Status status() const override { return iter_->status(); }
 
+  // Iterator::Refresh() not supported.
+
  private:
   Iterator* iter_;
   ColumnFamilyHandle* cfh_;
   BlobDBImpl* db_impl_;
+  bool own_snapshot_;
+  const Snapshot* snapshot_;
   mutable std::string vpart_;
 };
 

utilities/blob_db/blob_db_options_impl.cc

@@ -1,69 +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).
-#ifndef ROCKSDB_LITE
-
-#include "utilities/blob_db/blob_db_options_impl.h"
-
-namespace rocksdb {
-
-namespace blob_db {
-
-BlobDBOptionsImpl::BlobDBOptionsImpl(const BlobDBOptions& in)
-    : BlobDBOptions(in),
-      deletion_check_period_millisecs(2 * 1000),
-      gc_file_pct(20),
-      gc_check_period_millisecs(60 * 1000),
-      sanity_check_period_millisecs(20 * 60 * 1000),
-      open_files_trigger(100),
-      wa_num_stats_periods(24),
-      wa_stats_period_millisecs(3600 * 1000),
-      partial_expiration_gc_range_secs(4 * 3600),
-      partial_expiration_pct(75),
-      fsync_files_period_millisecs(10 * 1000),
-      reclaim_of_period_millisecs(1 * 1000),
-      delete_obsf_period_millisecs(10 * 1000),
-      check_seqf_period_millisecs(10 * 1000),
-      disable_background_tasks(false) {}
-
-BlobDBOptionsImpl::BlobDBOptionsImpl()
-    : deletion_check_period_millisecs(2 * 1000),
-      gc_file_pct(20),
-      gc_check_period_millisecs(60 * 1000),
-      sanity_check_period_millisecs(20 * 60 * 1000),
-      open_files_trigger(100),
-      wa_num_stats_periods(24),
-      wa_stats_period_millisecs(3600 * 1000),
-      partial_expiration_gc_range_secs(4 * 3600),
-      partial_expiration_pct(75),
-      fsync_files_period_millisecs(10 * 1000),
-      reclaim_of_period_millisecs(1 * 1000),
-      delete_obsf_period_millisecs(10 * 1000),
-      check_seqf_period_millisecs(10 * 1000),
-      disable_background_tasks(false) {}
-
-BlobDBOptionsImpl& BlobDBOptionsImpl::operator=(const BlobDBOptionsImpl& in) {
-  BlobDBOptions::operator=(in);
-  if (this != &in) {
-    deletion_check_period_millisecs = in.deletion_check_period_millisecs;
-    gc_file_pct = in.gc_file_pct;
-    gc_check_period_millisecs = in.gc_check_period_millisecs;
-    sanity_check_period_millisecs = in.sanity_check_period_millisecs;
-    open_files_trigger = in.open_files_trigger;
-    wa_num_stats_periods = in.wa_num_stats_periods;
-    wa_stats_period_millisecs = in.wa_stats_period_millisecs;
-    partial_expiration_gc_range_secs = in.partial_expiration_gc_range_secs;
-    partial_expiration_pct = in.partial_expiration_pct;
-    fsync_files_period_millisecs = in.fsync_files_period_millisecs;
-    reclaim_of_period_millisecs = in.reclaim_of_period_millisecs;
-    delete_obsf_period_millisecs = in.delete_obsf_period_millisecs;
-    check_seqf_period_millisecs = in.check_seqf_period_millisecs;
-    disable_background_tasks = in.disable_background_tasks;
-  }
-  return *this;
-}
-
-}  // namespace blob_db
-}  // namespace rocksdb
-#endif  // ROCKSDB_LITE

utilities/blob_db/blob_db_options_impl.h

@@ -1,76 +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
-
-#ifndef ROCKSDB_LITE
-
-#include "utilities/blob_db/blob_db.h"
-
-namespace rocksdb {
-
-namespace blob_db {
-
-struct BlobDBOptionsImpl : public BlobDBOptions {
-  // deletions check period
-  uint32_t deletion_check_period_millisecs;
-
-  // gc percentage each check period
-  uint32_t gc_file_pct;
-
-  // gc period
-  uint32_t gc_check_period_millisecs;
-
-  // sanity check task
-  uint32_t sanity_check_period_millisecs;
-
-  // how many random access open files can we tolerate
-  uint32_t open_files_trigger;
-
-  // how many periods of stats do we keep.
-  uint32_t wa_num_stats_periods;
-
-  // what is the length of any period
-  uint32_t wa_stats_period_millisecs;
-
-  // we will garbage collect blob files in
-  // which entire files have expired. However if the
-  // ttl_range of files is very large say a day, we
-  // would have to wait for the entire day, before we
-  // recover most of the space.
-  uint32_t partial_expiration_gc_range_secs;
-
-  // this should be based on allowed Write Amplification
-  // if 50% of the space of a blob file has been deleted/expired,
-  uint32_t partial_expiration_pct;
-
-  // how often should we schedule a job to fsync open files
-  uint32_t fsync_files_period_millisecs;
-
-  // how often to schedule reclaim open files.
-  uint32_t reclaim_of_period_millisecs;
-
-  // how often to schedule delete obs files periods
-  uint32_t delete_obsf_period_millisecs;
-
-  // how often to schedule check seq files period
-  uint32_t check_seqf_period_millisecs;
-
-  // Disable all background job.
-  bool disable_background_tasks;
-
-  // default constructor
-  BlobDBOptionsImpl();
-
-  explicit BlobDBOptionsImpl(const BlobDBOptions& in);
-
-  BlobDBOptionsImpl& operator=(const BlobDBOptionsImpl& in);
-};
-
-}  // namespace blob_db
-
-}  // namespace rocksdb
-
-#endif  // endif ROCKSDB

utilities/blob_db/blob_db_test.cc

@@ -14,9 +14,9 @@
 #include "port/port.h"
 #include "util/random.h"
 #include "util/string_util.h"
+#include "util/sync_point.h"
 #include "util/testharness.h"
 #include "utilities/blob_db/blob_db_impl.h"
-#include "utilities/blob_db/blob_db_options_impl.h"
 
 namespace rocksdb {
 namespace blob_db {
@@ -47,7 +47,7 @@ class BlobDBTest : public testing::Test {
 
   ~BlobDBTest() { Destroy(); }
 
-  void Open(BlobDBOptionsImpl bdb_options = BlobDBOptionsImpl(),
+  void Open(BlobDBOptions bdb_options = BlobDBOptions(),
             Options options = Options()) {
     options.create_if_missing = true;
     ASSERT_OK(BlobDB::Open(options, bdb_options, dbname_, &blob_db_));
@@ -63,7 +63,7 @@ class BlobDBTest : public testing::Test {
     }
   }
 
-  void PutRandomWithTTL(const std::string &key, int32_t ttl, Random *rnd,
+  void PutRandomWithTTL(const std::string &key, uint64_t ttl, Random *rnd,
                         std::map<std::string, std::string> *data = nullptr) {
     int len = rnd->Next() % kMaxBlobSize + 1;
     std::string value = test::RandomHumanReadableString(rnd, len);
@@ -74,7 +74,7 @@ class BlobDBTest : public testing::Test {
     }
   }
 
-  void PutRandomUntil(const std::string &key, int32_t expiration, Random *rnd,
+  void PutRandomUntil(const std::string &key, uint64_t expiration, Random *rnd,
                       std::map<std::string, std::string> *data = nullptr) {
     int len = rnd->Next() % kMaxBlobSize + 1;
     std::string value = test::RandomHumanReadableString(rnd, len);
@@ -136,7 +136,7 @@ class BlobDBTest : public testing::Test {
 
     Random rnd(301);
     for (size_t i = 0; i < 100000; i++) {
-      int32_t ttl = rnd.Next() % 86400;
+      uint64_t ttl = rnd.Next() % 86400;
       PutRandomWithTTL("key" + ToString(i % 500), ttl, &rnd, nullptr);
     }
 
@@ -153,7 +153,7 @@ class BlobDBTest : public testing::Test {
 
 TEST_F(BlobDBTest, Put) {
   Random rnd(301);
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.disable_background_tasks = true;
   Open(bdb_options);
   std::map<std::string, std::string> data;
@@ -167,7 +167,7 @@ TEST_F(BlobDBTest, PutWithTTL) {
   Random rnd(301);
   Options options;
   options.env = mock_env_.get();
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.ttl_range_secs = 1000;
   bdb_options.blob_file_size = 256 * 1000 * 1000;
   bdb_options.disable_background_tasks = true;
@@ -175,20 +175,20 @@ TEST_F(BlobDBTest, PutWithTTL) {
   std::map<std::string, std::string> data;
   mock_env_->set_now_micros(50 * 1000000);
   for (size_t i = 0; i < 100; i++) {
-    int32_t ttl = rnd.Next() % 100;
+    uint64_t ttl = rnd.Next() % 100;
     PutRandomWithTTL("key" + ToString(i), ttl, &rnd,
-                     (ttl < 50 ? nullptr : &data));
+                     (ttl <= 50 ? nullptr : &data));
   }
   mock_env_->set_now_micros(100 * 1000000);
   auto *bdb_impl = static_cast<BlobDBImpl *>(blob_db_);
   auto blob_files = bdb_impl->TEST_GetBlobFiles();
   ASSERT_EQ(1, blob_files.size());
   ASSERT_TRUE(blob_files[0]->HasTTL());
-  bdb_impl->TEST_CloseBlobFile(blob_files[0]);
+  ASSERT_OK(bdb_impl->TEST_CloseBlobFile(blob_files[0]));
   GCStats gc_stats;
   ASSERT_OK(bdb_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
   ASSERT_EQ(100 - data.size(), gc_stats.num_deletes);
-  ASSERT_EQ(data.size(), gc_stats.num_relocs);
+  ASSERT_EQ(data.size(), gc_stats.num_relocate);
   VerifyDB(data);
 }
 
@@ -196,7 +196,7 @@ TEST_F(BlobDBTest, PutUntil) {
   Random rnd(301);
   Options options;
   options.env = mock_env_.get();
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.ttl_range_secs = 1000;
   bdb_options.blob_file_size = 256 * 1000 * 1000;
   bdb_options.disable_background_tasks = true;
@@ -204,20 +204,20 @@ TEST_F(BlobDBTest, PutUntil) {
   std::map<std::string, std::string> data;
   mock_env_->set_now_micros(50 * 1000000);
   for (size_t i = 0; i < 100; i++) {
-    int32_t expiration = rnd.Next() % 100 + 50;
+    uint64_t expiration = rnd.Next() % 100 + 50;
     PutRandomUntil("key" + ToString(i), expiration, &rnd,
-                   (expiration < 100 ? nullptr : &data));
+                   (expiration <= 100 ? nullptr : &data));
   }
   mock_env_->set_now_micros(100 * 1000000);
   auto *bdb_impl = static_cast<BlobDBImpl *>(blob_db_);
   auto blob_files = bdb_impl->TEST_GetBlobFiles();
   ASSERT_EQ(1, blob_files.size());
   ASSERT_TRUE(blob_files[0]->HasTTL());
-  bdb_impl->TEST_CloseBlobFile(blob_files[0]);
+  ASSERT_OK(bdb_impl->TEST_CloseBlobFile(blob_files[0]));
   GCStats gc_stats;
   ASSERT_OK(bdb_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
   ASSERT_EQ(100 - data.size(), gc_stats.num_deletes);
-  ASSERT_EQ(data.size(), gc_stats.num_relocs);
+  ASSERT_EQ(data.size(), gc_stats.num_relocate);
   VerifyDB(data);
 }
 
@@ -227,7 +227,7 @@ TEST_F(BlobDBTest, TTLExtrator_NoTTL) {
   Random rnd(301);
   Options options;
   options.env = mock_env_.get();
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.ttl_range_secs = 1000;
   bdb_options.blob_file_size = 256 * 1000 * 1000;
   bdb_options.num_concurrent_simple_blobs = 1;
@@ -245,11 +245,11 @@ TEST_F(BlobDBTest, TTLExtrator_NoTTL) {
   auto blob_files = bdb_impl->TEST_GetBlobFiles();
   ASSERT_EQ(1, blob_files.size());
   ASSERT_FALSE(blob_files[0]->HasTTL());
-  bdb_impl->TEST_CloseBlobFile(blob_files[0]);
+  ASSERT_OK(bdb_impl->TEST_CloseBlobFile(blob_files[0]));
   GCStats gc_stats;
   ASSERT_OK(bdb_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
   ASSERT_EQ(0, gc_stats.num_deletes);
-  ASSERT_EQ(100, gc_stats.num_relocs);
+  ASSERT_EQ(100, gc_stats.num_relocate);
   VerifyDB(data);
 }
 
@@ -263,7 +263,7 @@ TEST_F(BlobDBTest, TTLExtractor_ExtractTTL) {
                             std::string * /*new_value*/,
                             bool * /*value_changed*/) override {
       *ttl = rnd->Next() % 100;
-      if (*ttl >= 50) {
+      if (*ttl > 50) {
         data[key.ToString()] = value.ToString();
       }
       return true;
@@ -275,7 +275,7 @@ TEST_F(BlobDBTest, TTLExtractor_ExtractTTL) {
   ttl_extractor_.reset(new TestTTLExtractor(&rnd));
   Options options;
   options.env = mock_env_.get();
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.ttl_range_secs = 1000;
   bdb_options.blob_file_size = 256 * 1000 * 1000;
   bdb_options.ttl_extractor = ttl_extractor_;
@@ -290,12 +290,12 @@ TEST_F(BlobDBTest, TTLExtractor_ExtractTTL) {
   auto blob_files = bdb_impl->TEST_GetBlobFiles();
   ASSERT_EQ(1, blob_files.size());
   ASSERT_TRUE(blob_files[0]->HasTTL());
-  bdb_impl->TEST_CloseBlobFile(blob_files[0]);
+  ASSERT_OK(bdb_impl->TEST_CloseBlobFile(blob_files[0]));
   GCStats gc_stats;
   ASSERT_OK(bdb_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
   auto &data = static_cast<TestTTLExtractor *>(ttl_extractor_.get())->data;
   ASSERT_EQ(100 - data.size(), gc_stats.num_deletes);
-  ASSERT_EQ(data.size(), gc_stats.num_relocs);
+  ASSERT_EQ(data.size(), gc_stats.num_relocate);
   VerifyDB(data);
 }
 
@@ -310,7 +310,7 @@ TEST_F(BlobDBTest, TTLExtractor_ExtractExpiration) {
                                    std::string * /*new_value*/,
                                    bool * /*value_changed*/) override {
       *expiration = rnd->Next() % 100 + 50;
-      if (*expiration >= 100) {
+      if (*expiration > 100) {
         data[key.ToString()] = value.ToString();
       }
       return true;
@@ -322,7 +322,7 @@ TEST_F(BlobDBTest, TTLExtractor_ExtractExpiration) {
   ttl_extractor_.reset(new TestTTLExtractor(&rnd));
   Options options;
   options.env = mock_env_.get();
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.ttl_range_secs = 1000;
   bdb_options.blob_file_size = 256 * 1000 * 1000;
   bdb_options.ttl_extractor = ttl_extractor_;
@@ -337,12 +337,12 @@ TEST_F(BlobDBTest, TTLExtractor_ExtractExpiration) {
   auto blob_files = bdb_impl->TEST_GetBlobFiles();
   ASSERT_EQ(1, blob_files.size());
   ASSERT_TRUE(blob_files[0]->HasTTL());
-  bdb_impl->TEST_CloseBlobFile(blob_files[0]);
+  ASSERT_OK(bdb_impl->TEST_CloseBlobFile(blob_files[0]));
   GCStats gc_stats;
   ASSERT_OK(bdb_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
   auto &data = static_cast<TestTTLExtractor *>(ttl_extractor_.get())->data;
   ASSERT_EQ(100 - data.size(), gc_stats.num_deletes);
-  ASSERT_EQ(data.size(), gc_stats.num_relocs);
+  ASSERT_EQ(data.size(), gc_stats.num_relocate);
   VerifyDB(data);
 }
 
@@ -369,7 +369,7 @@ TEST_F(BlobDBTest, TTLExtractor_ChangeValue) {
   Random rnd(301);
   Options options;
   options.env = mock_env_.get();
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.ttl_range_secs = 1000;
   bdb_options.blob_file_size = 256 * 1000 * 1000;
   bdb_options.ttl_extractor = std::make_shared<TestTTLExtractor>();
@@ -385,7 +385,7 @@ TEST_F(BlobDBTest, TTLExtractor_ChangeValue) {
     std::string value_ttl = value + "ttl:";
     PutFixed64(&value_ttl, ttl);
     ASSERT_OK(blob_db_->Put(WriteOptions(), Slice(key), Slice(value_ttl)));
-    if (ttl >= 50) {
+    if (ttl > 50) {
       data[key] = value;
     }
   }
@@ -394,17 +394,17 @@ TEST_F(BlobDBTest, TTLExtractor_ChangeValue) {
   auto blob_files = bdb_impl->TEST_GetBlobFiles();
   ASSERT_EQ(1, blob_files.size());
   ASSERT_TRUE(blob_files[0]->HasTTL());
-  bdb_impl->TEST_CloseBlobFile(blob_files[0]);
+  ASSERT_OK(bdb_impl->TEST_CloseBlobFile(blob_files[0]));
   GCStats gc_stats;
   ASSERT_OK(bdb_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
   ASSERT_EQ(100 - data.size(), gc_stats.num_deletes);
-  ASSERT_EQ(data.size(), gc_stats.num_relocs);
+  ASSERT_EQ(data.size(), gc_stats.num_relocate);
   VerifyDB(data);
 }
 
 TEST_F(BlobDBTest, StackableDBGet) {
   Random rnd(301);
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.disable_background_tasks = true;
   Open(bdb_options);
   std::map<std::string, std::string> data;
@@ -426,7 +426,7 @@ TEST_F(BlobDBTest, StackableDBGet) {
 
 TEST_F(BlobDBTest, WriteBatch) {
   Random rnd(301);
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.disable_background_tasks = true;
   Open(bdb_options);
   std::map<std::string, std::string> data;
@@ -442,7 +442,7 @@ TEST_F(BlobDBTest, WriteBatch) {
 
 TEST_F(BlobDBTest, Delete) {
   Random rnd(301);
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.disable_background_tasks = true;
   Open(bdb_options);
   std::map<std::string, std::string> data;
@@ -457,7 +457,7 @@ TEST_F(BlobDBTest, Delete) {
 
 TEST_F(BlobDBTest, DeleteBatch) {
   Random rnd(301);
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.disable_background_tasks = true;
   Open(bdb_options);
   for (size_t i = 0; i < 100; i++) {
@@ -474,7 +474,7 @@ TEST_F(BlobDBTest, DeleteBatch) {
 
 TEST_F(BlobDBTest, Override) {
   Random rnd(301);
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.disable_background_tasks = true;
   Open(bdb_options);
   std::map<std::string, std::string> data;
@@ -491,7 +491,7 @@ TEST_F(BlobDBTest, Override) {
 #ifdef SNAPPY
 TEST_F(BlobDBTest, Compression) {
   Random rnd(301);
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.disable_background_tasks = true;
   bdb_options.compression = CompressionType::kSnappyCompression;
   Open(bdb_options);
@@ -511,28 +511,43 @@ TEST_F(BlobDBTest, Compression) {
 }
 #endif
 
-TEST_F(BlobDBTest, DISABLED_MultipleWriters) {
-  Open();
+TEST_F(BlobDBTest, MultipleWriters) {
+  Open(BlobDBOptions());
 
   std::vector<port::Thread> workers;
-  for (size_t ii = 0; ii < 10; ii++)
-    workers.push_back(port::Thread(&BlobDBTest::InsertBlobs, this));
-
-  for (auto& t : workers) {
-    if (t.joinable()) {
-      t.join();
-    }
+  std::vector<std::map<std::string, std::string>> data_set(10);
+  for (uint32_t i = 0; i < 10; i++)
+    workers.push_back(port::Thread(
+        [&](uint32_t id) {
+          Random rnd(301 + id);
+          for (int j = 0; j < 100; j++) {
+            std::string key = "key" + ToString(id) + "_" + ToString(j);
+            if (id < 5) {
+              PutRandom(key, &rnd, &data_set[id]);
+            } else {
+              WriteBatch batch;
+              PutRandomToWriteBatch(key, &rnd, &batch, &data_set[id]);
+              blob_db_->Write(WriteOptions(), &batch);
+            }
+          }
+        },
+        i));
+  std::map<std::string, std::string> data;
+  for (size_t i = 0; i < 10; i++) {
+    workers[i].join();
+    data.insert(data_set[i].begin(), data_set[i].end());
   }
+  VerifyDB(data);
 }
 
 // Test sequence number store in blob file is correct.
 TEST_F(BlobDBTest, SequenceNumber) {
   Random rnd(301);
-  BlobDBOptionsImpl bdb_options;
+  BlobDBOptions bdb_options;
   bdb_options.disable_background_tasks = true;
   Open(bdb_options);
   SequenceNumber sequence = blob_db_->GetLatestSequenceNumber();
-  BlobDBImpl *blob_db_impl = reinterpret_cast<BlobDBImpl *>(blob_db_);
+  BlobDBImpl *blob_db_impl = dynamic_cast<BlobDBImpl*>(blob_db_);
   for (int i = 0; i < 100; i++) {
     std::string key = "key" + ToString(i);
     PutRandom(key, &rnd);
@@ -561,6 +576,237 @@ TEST_F(BlobDBTest, SequenceNumber) {
   }
 }
 
+TEST_F(BlobDBTest, GCAfterOverwriteKeys) {
+  Random rnd(301);
+  BlobDBOptions bdb_options;
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options);
+  BlobDBImpl *blob_db_impl = dynamic_cast<BlobDBImpl*>(blob_db_);
+  DBImpl *db_impl = dynamic_cast<DBImpl*>(blob_db_->GetBaseDB());
+  std::map<std::string, std::string> data;
+  for (int i = 0; i < 200; i++) {
+    PutRandom("key" + ToString(i), &rnd, &data);
+  }
+  auto blob_files = blob_db_impl->TEST_GetBlobFiles();
+  ASSERT_EQ(1, blob_files.size());
+  ASSERT_OK(blob_db_impl->TEST_CloseBlobFile(blob_files[0]));
+  // Test for data in SST
+  size_t new_keys = 0;
+  for (int i = 0; i < 100; i++) {
+    if (rnd.Next() % 2 == 1) {
+      new_keys++;
+      PutRandom("key" + ToString(i), &rnd, &data);
+    }
+  }
+  db_impl->TEST_FlushMemTable(true /*wait*/);
+  // Test for data in memtable
+  for (int i = 100; i < 200; i++) {
+    if (rnd.Next() % 2 == 1) {
+      new_keys++;
+      PutRandom("key" + ToString(i), &rnd, &data);
+    }
+  }
+  GCStats gc_stats;
+  ASSERT_OK(blob_db_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
+  ASSERT_EQ(200, gc_stats.blob_count);
+  ASSERT_EQ(0, gc_stats.num_deletes);
+  ASSERT_EQ(200 - new_keys, gc_stats.num_relocate);
+  VerifyDB(data);
+}
+
+TEST_F(BlobDBTest, GCRelocateKeyWhileOverwritting) {
+  Random rnd(301);
+  BlobDBOptions bdb_options;
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options);
+  ASSERT_OK(blob_db_->Put(WriteOptions(), "foo", "v1"));
+  BlobDBImpl *blob_db_impl = dynamic_cast<BlobDBImpl*>(blob_db_);
+  auto blob_files = blob_db_impl->TEST_GetBlobFiles();
+  ASSERT_EQ(1, blob_files.size());
+  ASSERT_OK(blob_db_impl->TEST_CloseBlobFile(blob_files[0]));
+
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"BlobDBImpl::GCFileAndUpdateLSM:AfterGetForUpdate",
+        "BlobDBImpl::PutUntil:Start"},
+       {"BlobDBImpl::PutUntil:Finish",
+        "BlobDBImpl::GCFileAndUpdateLSM:BeforeRelocate"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  auto writer = port::Thread(
+      [this]() { ASSERT_OK(blob_db_->Put(WriteOptions(), "foo", "v2")); });
+
+  GCStats gc_stats;
+  ASSERT_OK(blob_db_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
+  ASSERT_EQ(1, gc_stats.blob_count);
+  ASSERT_EQ(0, gc_stats.num_deletes);
+  ASSERT_EQ(1, gc_stats.num_relocate);
+  ASSERT_EQ(0, gc_stats.relocate_succeeded);
+  ASSERT_EQ(1, gc_stats.overwritten_while_relocate);
+  writer.join();
+  VerifyDB({{"foo", "v2"}});
+}
+
+TEST_F(BlobDBTest, GCExpiredKeyWhileOverwritting) {
+  Random rnd(301);
+  Options options;
+  options.env = mock_env_.get();
+  BlobDBOptions bdb_options;
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options, options);
+  mock_env_->set_now_micros(100 * 1000000);
+  ASSERT_OK(blob_db_->PutUntil(WriteOptions(), "foo", "v1", 200));
+  BlobDBImpl *blob_db_impl = dynamic_cast<BlobDBImpl*>(blob_db_);
+  auto blob_files = blob_db_impl->TEST_GetBlobFiles();
+  ASSERT_EQ(1, blob_files.size());
+  ASSERT_OK(blob_db_impl->TEST_CloseBlobFile(blob_files[0]));
+  mock_env_->set_now_micros(300 * 1000000);
+
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"BlobDBImpl::GCFileAndUpdateLSM:AfterGetForUpdate",
+        "BlobDBImpl::PutUntil:Start"},
+       {"BlobDBImpl::PutUntil:Finish",
+        "BlobDBImpl::GCFileAndUpdateLSM:BeforeDelete"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  auto writer = port::Thread([this]() {
+    ASSERT_OK(blob_db_->PutUntil(WriteOptions(), "foo", "v2", 400));
+  });
+
+  GCStats gc_stats;
+  ASSERT_OK(blob_db_impl->TEST_GCFileAndUpdateLSM(blob_files[0], &gc_stats));
+  ASSERT_EQ(1, gc_stats.blob_count);
+  ASSERT_EQ(1, gc_stats.num_deletes);
+  ASSERT_EQ(0, gc_stats.delete_succeeded);
+  ASSERT_EQ(1, gc_stats.overwritten_while_delete);
+  ASSERT_EQ(0, gc_stats.num_relocate);
+  writer.join();
+  VerifyDB({{"foo", "v2"}});
+}
+
+TEST_F(BlobDBTest, GCOldestSimpleBlobFileWhenOutOfSpace) {
+  // Use mock env to stop wall clock.
+  Options options;
+  options.env = mock_env_.get();
+  BlobDBOptions bdb_options;
+  bdb_options.blob_dir_size = 100;
+  bdb_options.blob_file_size = 100;
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options);
+  std::string value(100, 'v');
+  ASSERT_OK(blob_db_->PutWithTTL(WriteOptions(), "key_with_ttl", value, 60));
+  for (int i = 0; i < 10; i++) {
+    ASSERT_OK(blob_db_->Put(WriteOptions(), "key" + ToString(i), value));
+  }
+  BlobDBImpl *blob_db_impl = dynamic_cast<BlobDBImpl*>(blob_db_);
+  auto blob_files = blob_db_impl->TEST_GetBlobFiles();
+  ASSERT_EQ(11, blob_files.size());
+  ASSERT_TRUE(blob_files[0]->HasTTL());
+  ASSERT_TRUE(blob_files[0]->Immutable());
+  for (int i = 1; i <= 10; i++) {
+    ASSERT_FALSE(blob_files[i]->HasTTL());
+    if (i < 10) {
+      ASSERT_TRUE(blob_files[i]->Immutable());
+    }
+  }
+  blob_db_impl->TEST_RunGC();
+  // The oldest simple blob file (i.e. blob_files[1]) has been selected for GC.
+  auto obsolete_files = blob_db_impl->TEST_GetObsoleteFiles();
+  ASSERT_EQ(1, obsolete_files.size());
+  ASSERT_EQ(blob_files[1]->BlobFileNumber(),
+            obsolete_files[0]->BlobFileNumber());
+}
+
+TEST_F(BlobDBTest, ReadWhileGC) {
+  // run the same test for Get(), MultiGet() and Iterator each.
+  for (int i = 0; i < 3; i++) {
+    BlobDBOptions bdb_options;
+    bdb_options.disable_background_tasks = true;
+    Open(bdb_options);
+    blob_db_->Put(WriteOptions(), "foo", "bar");
+    BlobDBImpl *blob_db_impl = dynamic_cast<BlobDBImpl*>(blob_db_);
+    auto blob_files = blob_db_impl->TEST_GetBlobFiles();
+    ASSERT_EQ(1, blob_files.size());
+    std::shared_ptr<BlobFile> bfile = blob_files[0];
+    uint64_t bfile_number = bfile->BlobFileNumber();
+    ASSERT_OK(blob_db_impl->TEST_CloseBlobFile(bfile));
+
+    switch (i) {
+      case 0:
+        SyncPoint::GetInstance()->LoadDependency(
+            {{"BlobDBImpl::Get:AfterIndexEntryGet:1",
+              "BlobDBTest::ReadWhileGC:1"},
+             {"BlobDBTest::ReadWhileGC:2",
+              "BlobDBImpl::Get:AfterIndexEntryGet:2"}});
+        break;
+      case 1:
+        SyncPoint::GetInstance()->LoadDependency(
+            {{"BlobDBImpl::MultiGet:AfterIndexEntryGet:1",
+              "BlobDBTest::ReadWhileGC:1"},
+             {"BlobDBTest::ReadWhileGC:2",
+              "BlobDBImpl::MultiGet:AfterIndexEntryGet:2"}});
+        break;
+      case 2:
+        SyncPoint::GetInstance()->LoadDependency(
+            {{"BlobDBIterator::value:BeforeGetBlob:1",
+              "BlobDBTest::ReadWhileGC:1"},
+             {"BlobDBTest::ReadWhileGC:2",
+              "BlobDBIterator::value:BeforeGetBlob:2"}});
+        break;
+    }
+    SyncPoint::GetInstance()->EnableProcessing();
+
+    auto reader = port::Thread([this, i]() {
+      std::string value;
+      std::vector<std::string> values;
+      std::vector<Status> statuses;
+      switch (i) {
+        case 0:
+          ASSERT_OK(blob_db_->Get(ReadOptions(), "foo", &value));
+          ASSERT_EQ("bar", value);
+          break;
+        case 1:
+          statuses = blob_db_->MultiGet(ReadOptions(), {"foo"}, &values);
+          ASSERT_EQ(1, statuses.size());
+          ASSERT_EQ(1, values.size());
+          ASSERT_EQ("bar", values[0]);
+          break;
+        case 2:
+          // VerifyDB use iterator to scan the DB.
+          VerifyDB({{"foo", "bar"}});
+          break;
+      }
+    });
+
+    TEST_SYNC_POINT("BlobDBTest::ReadWhileGC:1");
+    GCStats gc_stats;
+    ASSERT_OK(blob_db_impl->TEST_GCFileAndUpdateLSM(bfile, &gc_stats));
+    ASSERT_EQ(1, gc_stats.blob_count);
+    ASSERT_EQ(1, gc_stats.num_relocate);
+    ASSERT_EQ(1, gc_stats.relocate_succeeded);
+    blob_db_impl->TEST_ObsoleteFile(blob_files[0]);
+    blob_db_impl->TEST_DeleteObsoleteFiles();
+    // The file shouln't be deleted
+    blob_files = blob_db_impl->TEST_GetBlobFiles();
+    ASSERT_EQ(2, blob_files.size());
+    ASSERT_EQ(bfile_number, blob_files[0]->BlobFileNumber());
+    auto obsolete_files = blob_db_impl->TEST_GetObsoleteFiles();
+    ASSERT_EQ(1, obsolete_files.size());
+    ASSERT_EQ(bfile_number, obsolete_files[0]->BlobFileNumber());
+    TEST_SYNC_POINT("BlobDBTest::ReadWhileGC:2");
+    reader.join();
+    SyncPoint::GetInstance()->DisableProcessing();
+
+    // The file is deleted this time
+    blob_db_impl->TEST_DeleteObsoleteFiles();
+    blob_files = blob_db_impl->TEST_GetBlobFiles();
+    ASSERT_EQ(1, blob_files.size());
+    ASSERT_NE(bfile_number, blob_files[0]->BlobFileNumber());
+    ASSERT_EQ(0, blob_db_impl->TEST_GetObsoleteFiles().size());
+    VerifyDB({{"foo", "bar"}});
+    Destroy();
+  }
+}
+
 }  //  namespace blob_db
 }  //  namespace rocksdb
 

utilities/blob_db/blob_dump_tool.cc

@@ -102,8 +102,8 @@ Status BlobDumpTool::DumpBlobLogHeader(uint64_t* offset) {
     return s;
   }
   fprintf(stdout, "Blob log header:\n");
-  fprintf(stdout, "  Magic Number   : %u\n", header.magic_number());
-  fprintf(stdout, "  Version        : %d\n", header.version());
+  fprintf(stdout, "  Magic Number   : %" PRIu32 "\n", header.magic_number());
+  fprintf(stdout, "  Version        : %" PRIu32 "\n", header.version());
   CompressionType compression = header.compression();
   std::string compression_str;
   if (!GetStringFromCompressionType(&compression_str, compression).ok()) {
@@ -175,13 +175,13 @@ Status BlobDumpTool::DumpRecord(DisplayType show_key, DisplayType show_blob,
   }
   uint32_t key_size = record.GetKeySize();
   uint64_t blob_size = record.GetBlobSize();
-  fprintf(stdout, "  key size   : %d\n", key_size);
+  fprintf(stdout, "  key size   : %" PRIu32 "\n", key_size);
   fprintf(stdout, "  blob size  : %" PRIu64 "\n", record.GetBlobSize());
-  fprintf(stdout, "  TTL        : %u\n", record.GetTTL());
+  fprintf(stdout, "  TTL        : %" PRIu64 "\n", record.GetTTL());
   fprintf(stdout, "  time       : %" PRIu64 "\n", record.GetTimeVal());
   fprintf(stdout, "  type       : %d, %d\n", record.type(), record.subtype());
-  fprintf(stdout, "  header CRC : %u\n", record.header_checksum());
-  fprintf(stdout, "  CRC        : %u\n", record.checksum());
+  fprintf(stdout, "  header CRC : %" PRIu32 "\n", record.header_checksum());
+  fprintf(stdout, "  CRC        : %" PRIu32 "\n", record.checksum());
   uint32_t header_crc =
       crc32c::Extend(0, slice.data(), slice.size() - 2 * sizeof(uint32_t));
   *offset += BlobLogRecord::kHeaderSize;
@@ -213,7 +213,7 @@ Status BlobDumpTool::DumpRecord(DisplayType show_key, DisplayType show_blob,
   if (!s.ok()) {
     return s;
   }
-  fprintf(stdout, "  footer CRC : %u\n", record.footer_checksum());
+  fprintf(stdout, "  footer CRC : %" PRIu32 "\n", record.footer_checksum());
   fprintf(stdout, "  sequence   : %" PRIu64 "\n", record.GetSN());
   *offset += key_size + blob_size + BlobLogRecord::kFooterSize;
   return s;

utilities/blob_db/blob_file.cc

@@ -94,8 +94,8 @@ std::string BlobFile::DumpState() const {
            "path: %s fn: %" PRIu64 " blob_count: %" PRIu64 " gc_epoch: %" PRIu64
            " file_size: %" PRIu64 " deleted_count: %" PRIu64
            " deleted_size: %" PRIu64
-           " closed: %d can_be_deleted: %d ttl_range: (%d, %d)"
-           " sn_range: (%" PRIu64 " %" PRIu64 "), writer: %d reader: %d",
+           " closed: %d can_be_deleted: %d ttl_range: (%" PRIu64 ", %" PRIu64
+           ") sn_range: (%" PRIu64 " %" PRIu64 "), writer: %d reader: %d",
            path_to_dir_.c_str(), file_number_, blob_count_.load(),
            gc_epoch_.load(), file_size_.load(), deleted_count_, deleted_size_,
            closed_.load(), can_be_deleted_.load(), ttl_range_.first,

utilities/blob_db/blob_log_format.cc

@@ -61,8 +61,8 @@ Status BlobLogFooter::DecodeFrom(const Slice& input) {
   }
 
   ttlrange_t temp_ttl;
-  if (!GetFixed32(&slice, &temp_ttl.first) ||
-      !GetFixed32(&slice, &temp_ttl.second)) {
+  if (!GetFixed64(&slice, &temp_ttl.first) ||
+      !GetFixed64(&slice, &temp_ttl.second)) {
     return Status::Corruption("Invalid Blob Footer: ttl_range");
   }
   if (has_ttl) {
@@ -108,11 +108,11 @@ void BlobLogFooter::EncodeTo(std::string* dst) const {
   bool has_ts = HasTimestamp();
 
   if (has_ttl) {
-    PutFixed32(dst, ttl_range_.get()->first);
-    PutFixed32(dst, ttl_range_.get()->second);
+    PutFixed64(dst, ttl_range_.get()->first);
+    PutFixed64(dst, ttl_range_.get()->second);
   } else {
-    PutFixed32(dst, 0);
-    PutFixed32(dst, 0);
+    PutFixed64(dst, 0);
+    PutFixed64(dst, 0);
   }
   PutFixed64(dst, sn_range_.first);
   PutFixed64(dst, sn_range_.second);
@@ -149,11 +149,11 @@ void BlobLogHeader::EncodeTo(std::string* dst) const {
   PutFixed32(dst, val);
 
   if (has_ttl) {
-    PutFixed32(dst, ttl_guess_.get()->first);
-    PutFixed32(dst, ttl_guess_.get()->second);
+    PutFixed64(dst, ttl_guess_.get()->first);
+    PutFixed64(dst, ttl_guess_.get()->second);
   } else {
-    PutFixed32(dst, 0);
-    PutFixed32(dst, 0);
+    PutFixed64(dst, 0);
+    PutFixed64(dst, 0);
   }
 
   if (has_ts) {
@@ -199,11 +199,13 @@ Status BlobLogHeader::DecodeFrom(const Slice& input) {
   }
 
   ttlrange_t temp_ttl;
-  if (!GetFixed32(&slice, &temp_ttl.first) ||
-      !GetFixed32(&slice, &temp_ttl.second)) {
+  if (!GetFixed64(&slice, &temp_ttl.first) ||
+      !GetFixed64(&slice, &temp_ttl.second)) {
     return Status::Corruption("Invalid Blob Log Header: ttl");
   }
-  if (has_ttl) set_ttl_guess(temp_ttl);
+  if (has_ttl) {
+    set_ttl_guess(temp_ttl);
+  }
 
   tsrange_t temp_ts;
   if (!GetFixed64(&slice, &temp_ts.first) ||
@@ -265,7 +267,7 @@ Status BlobLogRecord::DecodeHeaderFrom(const Slice& hdrslice) {
   if (!GetFixed64(&input, &blob_size_)) {
     return Status::Corruption("Invalid Blob Record Header: blob_size");
   }
-  if (!GetFixed32(&input, &ttl_val_)) {
+  if (!GetFixed64(&input, &ttl_val_)) {
     return Status::Corruption("Invalid Blob Record Header: ttl_val");
   }
   if (!GetFixed64(&input, &time_val_)) {

utilities/blob_db/blob_log_format.h

@@ -25,6 +25,8 @@ namespace blob_db {
 class BlobFile;
 class BlobDBImpl;
 
+constexpr uint64_t kNoExpiration = std::numeric_limits<uint64_t>::max();
+
 enum RecordType : uint8_t {
   // Zero is reserved for preallocated files
   kFullType = 0,
@@ -46,9 +48,9 @@ extern const uint32_t kMagicNumber;
 
 class Reader;
 
-typedef std::pair<uint32_t, uint32_t> ttlrange_t;
-typedef std::pair<uint64_t, uint64_t> tsrange_t;
-typedef std::pair<rocksdb::SequenceNumber, rocksdb::SequenceNumber> snrange_t;
+using ttlrange_t = std::pair<uint64_t, uint64_t>;
+using tsrange_t = std::pair<uint64_t, uint64_t>;
+using snrange_t = std::pair<rocksdb::SequenceNumber, rocksdb::SequenceNumber>;
 
 class BlobLogHeader {
   friend class BlobFile;
@@ -71,8 +73,8 @@ class BlobLogHeader {
   void set_ts_guess(const tsrange_t& ts) { ts_guess_.reset(new tsrange_t(ts)); }
 
  public:
-  // magic number + version + flags + ttl guess + timestamp range = 36
-  static const size_t kHeaderSize = 4 + 4 + 4 + 4 * 2 + 8 * 2;
+  // magic number + version + flags + ttl guess + timestamp range = 44
+  static const size_t kHeaderSize = 4 + 4 + 4 + 8 * 2 + 8 * 2;
 
   void EncodeTo(std::string* dst) const;
 
@@ -100,9 +102,9 @@ class BlobLogHeader {
     return *ts_guess_;
   }
 
-  bool HasTTL() const { return !!ttl_guess_; }
+  bool HasTTL() const { return ttl_guess_ != nullptr; }
 
-  bool HasTimestamp() const { return !!ts_guess_; }
+  bool HasTimestamp() const { return ts_guess_ != nullptr; }
 
   BlobLogHeader& operator=(BlobLogHeader&& in) noexcept;
 };
@@ -128,11 +130,11 @@ class BlobLogFooter {
 
   // footer size = 4 byte magic number
   // 8 bytes count
-  // 4, 4 - ttl range
+  // 8, 8 - ttl range
   // 8, 8 - sn range
   // 8, 8 - ts range
-  // = 56
-  static const size_t kFooterSize = 4 + 4 + 8 + (4 * 2) + (8 * 2) + (8 * 2);
+  // = 64
+  static const size_t kFooterSize = 4 + 4 + 8 + (8 * 2) + (8 * 2) + (8 * 2);
 
   bool HasTTL() const { return !!ttl_range_; }
 
@@ -185,7 +187,7 @@ class BlobLogRecord {
   uint32_t key_size_;
   uint64_t blob_size_;
   uint64_t time_val_;
-  uint32_t ttl_val_;
+  uint64_t ttl_val_;
   SequenceNumber sn_;
   uint32_t footer_cksum_;
   char type_;
@@ -209,11 +211,12 @@ class BlobLogRecord {
  public:
   // Header is
   // Key Length ( 4 bytes ),
-  // Blob Length ( 8 bytes), timestamp/ttl (8 bytes),
+  // Blob Length ( 8 bytes),
+  // ttl (8 bytes), timestamp (8 bytes),
   // type (1 byte), subtype (1 byte)
   // header checksum (4 bytes), blob checksum (4 bytes),
-  // = 34
-  static const size_t kHeaderSize = 4 + 4 + 4 + 8 + 4 + 8 + 1 + 1;
+  // = 42
+  static const size_t kHeaderSize = 4 + 4 + 8 + 8 + 4 + 8 + 1 + 1;
 
   static const size_t kFooterSize = 8 + 4;
 
@@ -234,7 +237,7 @@ class BlobLogRecord {
     return ttl_val_ != std::numeric_limits<uint32_t>::max();
   }
 
-  uint32_t GetTTL() const { return ttl_val_; }
+  uint64_t GetTTL() const { return ttl_val_; }
 
   uint64_t GetTimeVal() const { return time_val_; }
 

utilities/blob_db/blob_log_reader.cc

@@ -41,7 +41,7 @@ Status Reader::ReadHeader(BlobLogHeader* header) {
 }
 
 Status Reader::ReadRecord(BlobLogRecord* record, ReadLevel level,
-                          WALRecoveryMode wal_recovery_mode) {
+                          uint64_t* blob_offset) {
   record->Clear();
   buffer_.clear();
   backing_store_[0] = '\0';
@@ -65,6 +65,9 @@ Status Reader::ReadRecord(BlobLogRecord* record, ReadLevel level,
   header_crc = crc32c::Extend(header_crc, buffer_.data(), crc_data_size);
 
   uint64_t kb_size = record->GetKeySize() + record->GetBlobSize();
+  if (blob_offset != nullptr) {
+    *blob_offset = next_byte_ + record->GetKeySize();
+  }
   switch (level) {
     case kReadHdrFooter:
       file_->Skip(kb_size);

utilities/blob_db/blob_log_reader.h

@@ -60,9 +60,9 @@ class Reader {
   // "*scratch" as temporary storage.  The contents filled in *record
   // will only be valid until the next mutating operation on this
   // reader or the next mutation to *scratch.
+  // If blob_offset is non-null, return offset of the blob through it.
   Status ReadRecord(BlobLogRecord* record, ReadLevel level = kReadHdrFooter,
-                    WALRecoveryMode wal_recovery_mode =
-                        WALRecoveryMode::kTolerateCorruptedTailRecords);
+                    uint64_t* blob_offset = nullptr);
 
   SequentialFileReader* file() { return file_.get(); }
 

utilities/blob_db/blob_log_writer.cc

@@ -8,7 +8,6 @@
 #include "utilities/blob_db/blob_log_writer.h"
 
 #include <cstdint>
-#include <limits>
 #include <string>
 #include "rocksdb/env.h"
 #include "util/coding.h"
@@ -72,7 +71,7 @@ Status Writer::AppendFooter(const BlobLogFooter& footer) {
 
 Status Writer::AddRecord(const Slice& key, const Slice& val,
                          uint64_t* key_offset, uint64_t* blob_offset,
-                         uint32_t ttl) {
+                         uint64_t ttl) {
   assert(block_offset_ != 0);
   assert(last_elem_type_ == kEtFileHdr || last_elem_type_ == kEtFooter);
 
@@ -96,26 +95,23 @@ Status Writer::AddRecord(const Slice& key, const Slice& val,
 }
 
 void Writer::ConstructBlobHeader(std::string* headerbuf, const Slice& key,
-                                 const Slice& val, int32_t ttl, int64_t ts) {
+                                 const Slice& val, uint64_t ttl, int64_t ts) {
   headerbuf->reserve(BlobLogRecord::kHeaderSize);
 
   uint32_t key_size = static_cast<uint32_t>(key.size());
   PutFixed32(headerbuf, key_size);
   PutFixed64(headerbuf, val.size());
 
-  uint32_t ttl_write = (ttl != -1) ? static_cast<uint32_t>(ttl)
-                                   : std::numeric_limits<uint32_t>::max();
-  PutFixed32(headerbuf, ttl_write);
-
-  uint64_t ts_write = (ts != -1) ? static_cast<uint64_t>(ts)
-                                 : std::numeric_limits<uint64_t>::max();
-  PutFixed64(headerbuf, ts_write);
+  PutFixed64(headerbuf, ttl);
+  PutFixed64(headerbuf, ts);
 
   RecordType t = kFullType;
   headerbuf->push_back(static_cast<char>(t));
 
   RecordSubType st = kRegularType;
-  if (ttl != -1) st = kTTLType;
+  if (ttl != kNoExpiration) {
+    st = kTTLType;
+  }
   headerbuf->push_back(static_cast<char>(st));
 
   uint32_t header_crc = 0;

utilities/blob_db/blob_log_writer.h

@@ -41,13 +41,13 @@ class Writer {
   ~Writer();
 
   static void ConstructBlobHeader(std::string* headerbuf, const Slice& key,
-                                  const Slice& val, int32_t ttl, int64_t ts);
+                                  const Slice& val, uint64_t ttl, int64_t ts);
 
   Status AddRecord(const Slice& key, const Slice& val, uint64_t* key_offset,
                    uint64_t* blob_offset);
 
   Status AddRecord(const Slice& key, const Slice& val, uint64_t* key_offset,
-                   uint64_t* blob_offset, uint32_t ttl);
+                   uint64_t* blob_offset, uint64_t ttl);
 
   Status EmitPhysicalRecord(const std::string& headerbuf, const Slice& key,
                             const Slice& val, uint64_t* key_offset,