Fix rocksdb.estimate-num-keys DB property underflow

Summary:
rocksdb.estimate-num-keys is compute from `estimate_num_keys - 2 * estimate_num_deletes`. If  `2 * estimate_num_deletes > estimate_num_keys` it will underflow. Fixing it.
Closes https://github.com/facebook/rocksdb/pull/2348

Differential Revision: D5109272

Pulled By: yiwu-arbug

fbshipit-source-id: e1bfb91346a59b7282a282b615002507e9d7c246

CMakeLists.txt

@@ -432,6 +432,12 @@ set(SOURCES
         util/xxhash.cc
         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
+        utilities/blob_db/blob_log_format.cc
         utilities/checkpoint/checkpoint.cc
         utilities/col_buf_decoder.cc
         utilities/col_buf_encoder.cc
@@ -645,10 +651,10 @@ set(TESTS
         util/heap_test.cc
         util/rate_limiter_test.cc
         util/slice_transform_test.cc
+        util/timer_queue_test.cc
         util/thread_list_test.cc
         util/thread_local_test.cc
         utilities/backupable/backupable_db_test.cc
-        utilities/blob_db/blob_db_test.cc
         utilities/checkpoint/checkpoint_test.cc
         utilities/column_aware_encoding_test.cc
         utilities/date_tiered/date_tiered_test.cc

HISTORY.md

@@ -1,5 +1,36 @@
 # Rocksdb Change Log
-## Unreleased
+## 5.4.5 (05/19/2017)
+### Bug Fixes
+* Facility for cross-building RocksJava using Docker.
+* Build and link with ZStd when creating the static RocksJava build.
+* Fix snprintf's usage to be cross-platform.
+* Fix build error with blob DB.
+* Fix readamp test type inconsistency.
+
+## 5.4.4 (05/11/2017)
+### New Features
+* Added DB::CreateColumnFamilie() and DB::DropColumnFamilies() to bulk create/drop column families.
+
+### Bug Fixes
+* Fix a manual / auto compaction data race.
+* Fix CentOS 5 cross-building of RocksJava.
+
+## 5.4.3 (05/10/2017)
+### Bug Fixes
+* Fix memory alignment with logical sector size.
+* Fix alignment in ReadaheadRandomAccessFile.
+* Fix bias with read amplification stats (READ_AMP_ESTIMATE_USEFUL_BYTES and READ_AMP_TOTAL_READ_BYTES).
+
+## 5.4.2 (05/08/2017)
+### Bug Fixes
+* Fix huge fallocate calls fail and make XFS unhappy.
+
+## 5.4.1 (05/02/2017)
+### Bug Fxies
+* Fix WriteBatchWithIndex address use after scope error.
+* Fix WritableFile buffer size in direct IO.
+* Add prefetch to PosixRandomAccessFile in buffered io.
+* Fix PinnableSlice access invalid address when row cache is enabled.
 
 ## 5.4.0 (04/11/2017)
 ### Public API Change
@@ -8,11 +39,14 @@
 * DB::Get in place of std::string accepts PinnableSlice, which avoids the extra memcpy of value to std::string in most of cases.
     * PinnableSlice releases the pinned resources that contain the value when it is destructed or when ::Reset() is called on it.
     * The old API that accepts std::string, although discouraged, is still supported.
+* Replace Options::use_direct_writes with Options::use_direct_io_for_flush_and_compaction. Read Direct IO wiki for details.
+* Added CompactionEventListener and EventListener::OnFlushBegin interfaces.
 
 ### New Features
 * Memtable flush can be avoided during checkpoint creation if total log file size is smaller than a threshold specified by the user.
 * Introduce level-based L0->L0 compactions to reduce file count, so write delays are incurred less often.
 * (Experimental) Partitioning filters which creates an index on the partitions. The feature can be enabled by setting partition_filters when using kFullFilter. Currently the feature also requires two-level indexing to be enabled. Number of partitions is the same as the number of partitions for indexes, which is controlled by metadata_block_size.
+* DB::ResetStats() to reset internal stats.
 
 ## 5.3.0 (03/08/2017)
 ### Public API Change

Makefile

@@ -379,7 +379,6 @@ TESTS = \
 	ttl_test \
 	date_tiered_test \
 	backupable_db_test \
-	blob_db_test \
 	document_db_test \
 	json_document_test \
 	sim_cache_test \
@@ -400,6 +399,7 @@ TESTS = \
 	options_settable_test \
 	options_util_test \
 	event_logger_test \
+	timer_queue_test \
 	cuckoo_table_builder_test \
 	cuckoo_table_reader_test \
 	cuckoo_table_db_test \
@@ -1075,9 +1075,6 @@ prefix_test: db/prefix_test.o $(LIBOBJECTS) $(TESTHARNESS)
 backupable_db_test: utilities/backupable/backupable_db_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
-blob_db_test: utilities/blob_db/blob_db_test.o $(LIBOBJECTS) $(TESTHARNESS)
-	$(AM_LINK)
-
 checkpoint_test: utilities/checkpoint/checkpoint_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
@@ -1272,6 +1269,9 @@ db_bench_tool_test: tools/db_bench_tool_test.o $(BENCHTOOLOBJECTS) $(TESTHARNESS
 event_logger_test: util/event_logger_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
+timer_queue_test: util/timer_queue_test.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(AM_LINK)
+
 sst_dump_test: tools/sst_dump_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
@@ -1405,6 +1405,9 @@ SNAPPY_VER = 1.1.4
 SNAPPY_SHA256 = 134bfe122fd25599bb807bb8130e7ba6d9bdb851e0b16efcb83ac4f5d0b70057
 LZ4_VER = 1.7.5
 LZ4_SHA256 = 0190cacd63022ccb86f44fa5041dc6c3804407ad61550ca21c382827319e7e7e
+ZSTD_VER ?= 1.2.0
+ZSTD_SHA256 ?= 4a7e4593a3638276ca7f2a09dc4f38e674d8317bbea51626393ca73fc047cbfb
+ZSTD_DOWNLOAD_BASE ?= https://github.com/facebook/zstd/archive
 
 ifeq ($(PLATFORM), OS_MACOSX)
 	ROCKSDBJNILIB = librocksdbjni-osx.jnilib
@@ -1478,16 +1481,29 @@ liblz4.a:
 	cd lz4-$(LZ4_VER)/lib && make CFLAGS='-fPIC' all
 	cp lz4-$(LZ4_VER)/lib/liblz4.a .
 
+libzstd.a:
+	-rm -rf zstd-$(ZSTD_VER)
+	curl -O -L ${ZSTD_DOWNLOAD_BASE}/v$(ZSTD_VER).tar.gz
+	mv v$(ZSTD_VER).tar.gz zstd-$(ZSTD_VER).tar.gz
+	ZSTD_SHA256_ACTUAL=`$(SHA256_CMD) zstd-$(ZSTD_VER).tar.gz | cut -d ' ' -f 1`; \
+	if [ "$(ZSTD_SHA256)" != "$$ZSTD_SHA256_ACTUAL" ]; then \
+		echo zstd-$(ZSTD_VER).tar.gz checksum mismatch, expected=\"$(ZSTD_SHA256)\" actual=\"$$ZSTD_SHA256_ACTUAL\"; \
+		exit 1; \
+	fi
+	tar xvzf zstd-$(ZSTD_VER).tar.gz
+	cd zstd-$(ZSTD_VER)/lib && make CFLAGS='-fPIC -O2 ${EXTRA_CFLAGS}' all
+	cp zstd-$(ZSTD_VER)/lib/libzstd.a .
+
 # A version of each $(LIBOBJECTS) compiled with -fPIC and a fixed set of static compression libraries
 java_static_libobjects = $(patsubst %,jls/%,$(LIBOBJECTS))
 CLEAN_FILES += jls
 
 ifneq ($(ROCKSDB_JAVA_NO_COMPRESSION), 1)
-JAVA_COMPRESSIONS = libz.a libbz2.a libsnappy.a liblz4.a
+JAVA_COMPRESSIONS = libz.a libbz2.a libsnappy.a liblz4.a libzstd.a
 endif
 
-JAVA_STATIC_FLAGS = -DZLIB -DBZIP2 -DSNAPPY -DLZ4
-JAVA_STATIC_INCLUDES = -I./zlib-$(ZLIB_VER) -I./bzip2-$(BZIP2_VER) -I./snappy-$(SNAPPY_VER) -I./lz4-$(LZ4_VER)/lib
+JAVA_STATIC_FLAGS = -DZLIB -DBZIP2 -DSNAPPY -DLZ4 -DZSTD
+JAVA_STATIC_INCLUDES = -I./zlib-$(ZLIB_VER) -I./bzip2-$(BZIP2_VER) -I./snappy-$(SNAPPY_VER) -I./lz4-$(LZ4_VER)/lib -I./zstd-$(ZSTD_VER)/lib
 
 $(java_static_libobjects): jls/%.o: %.cc $(JAVA_COMPRESSIONS)
 	$(AM_V_CC)mkdir -p $(@D) && $(CXX) $(CXXFLAGS) $(JAVA_STATIC_FLAGS) $(JAVA_STATIC_INCLUDES) -fPIC -c $< -o $@ $(COVERAGEFLAGS)
@@ -1512,8 +1528,25 @@ rocksdbjavastaticrelease: rocksdbjavastatic
 	cd java/target;jar -uf $(ROCKSDB_JAR_ALL) librocksdbjni-*.so librocksdbjni-*.jnilib
 	cd java/target/classes;jar -uf ../$(ROCKSDB_JAR_ALL) org/rocksdb/*.class org/rocksdb/util/*.class
 
+rocksdbjavastaticreleasedocker: rocksdbjavastatic
+	DOCKER_LINUX_X64_CONTAINER=`docker ps -aqf name=rocksdb_linux_x64-be`; \
+	if [ -z "$$DOCKER_LINUX_X64_CONTAINER" ]; then \
+		docker container create --attach stdin --attach stdout --attach stderr --volume `pwd`:/rocksdb-host --name rocksdb_linux_x64-be evolvedbinary/rocksjava:centos5_x64-be /rocksdb-host/java/crossbuild/docker-build-linux-centos.sh; \
+	fi
+	docker start -a rocksdb_linux_x64-be
+	DOCKER_LINUX_X86_CONTAINER=`docker ps -aqf name=rocksdb_linux_x86-be`; \
+	if [ -z "$$DOCKER_LINUX_X86_CONTAINER" ]; then \
+		docker container create --attach stdin --attach stdout --attach stderr --volume `pwd`:/rocksdb-host --name rocksdb_linux_x86-be evolvedbinary/rocksjava:centos5_x86-be /rocksdb-host/java/crossbuild/docker-build-linux-centos.sh; \
+	fi
+	docker start -a rocksdb_linux_x86-be
+	cd java;jar -cf target/$(ROCKSDB_JAR_ALL) HISTORY*.md
+	cd java/target;jar -uf $(ROCKSDB_JAR_ALL) librocksdbjni-*.so librocksdbjni-*.jnilib
+	cd java/target/classes;jar -uf ../$(ROCKSDB_JAR_ALL) org/rocksdb/*.class org/rocksdb/util/*.class
+
 rocksdbjavastaticpublish: rocksdbjavastaticrelease rocksdbjavastaticpublishcentral
 
+rocksdbjavastaticpublishdocker: rocksdbjavastaticreleasedocker rocksdbjavastaticpublishcentral
+
 rocksdbjavastaticpublishcentral:
 	mvn gpg:sign-and-deploy-file -Durl=https://oss.sonatype.org/service/local/staging/deploy/maven2/ -DrepositoryId=sonatype-nexus-staging -DpomFile=java/rocksjni.pom -Dfile=java/target/rocksdbjni-$(ROCKSDB_MAJOR).$(ROCKSDB_MINOR).$(ROCKSDB_PATCH)-javadoc.jar -Dclassifier=javadoc
 	mvn gpg:sign-and-deploy-file -Durl=https://oss.sonatype.org/service/local/staging/deploy/maven2/ -DrepositoryId=sonatype-nexus-staging -DpomFile=java/rocksjni.pom -Dfile=java/target/rocksdbjni-$(ROCKSDB_MAJOR).$(ROCKSDB_MINOR).$(ROCKSDB_PATCH)-sources.jar -Dclassifier=sources

TARGETS

@@ -195,6 +195,12 @@ cpp_library(
       "util/xxhash.cc",
       "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",
+      "utilities/blob_db/blob_log_format.cc",
       "utilities/checkpoint/checkpoint.cc",
       "utilities/compaction_filters/remove_emptyvalue_compactionfilter.cc",
       "utilities/convenience/info_log_finder.cc",
@@ -319,7 +325,6 @@ ROCKS_TESTS = [['merger_test', 'table/merger_test.cc', 'serial'],
  ['compact_files_test', 'db/compact_files_test.cc', 'serial'],
  ['db_options_test', 'db/db_options_test.cc', 'serial'],
  ['object_registry_test', 'utilities/object_registry_test.cc', 'serial'],
- ['blob_db_test', 'utilities/blob_db/blob_db_test.cc', 'serial'],
  ['auto_roll_logger_test', 'util/auto_roll_logger_test.cc', 'serial'],
  ['dbformat_test', 'db/dbformat_test.cc', 'serial'],
  ['write_batch_with_index_test',

db/builder.cc

@@ -32,6 +32,7 @@
 #include "util/file_reader_writer.h"
 #include "util/filename.h"
 #include "util/stop_watch.h"
+#include "util/sync_point.h"
 
 namespace rocksdb {
 
@@ -103,6 +104,10 @@ Status BuildTable(
     unique_ptr<WritableFileWriter> file_writer;
     {
       unique_ptr<WritableFile> file;
+#ifndef NDEBUG
+      bool use_direct_writes = env_options.use_direct_writes;
+      TEST_SYNC_POINT_CALLBACK("BuildTable:create_file", &use_direct_writes);
+#endif  // !NDEBUG
       s = NewWritableFile(env, fname, &file, env_options);
       if (!s.ok()) {
         EventHelpers::LogAndNotifyTableFileCreationFinished(
@@ -180,6 +185,11 @@ Status BuildTable(
 
     if (s.ok() && !empty) {
       // Verify that the table is usable
+      // We set for_compaction to false and don't OptimizeForCompactionTableRead
+      // here because this is a special case after we finish the table building
+      // No matter whether use_direct_io_for_flush_and_compaction is true,
+      // we will regrad this verification as user reads since the goal is
+      // to cache it here for further user reads
       std::unique_ptr<InternalIterator> it(table_cache->NewIterator(
           ReadOptions(), env_options, internal_comparator, meta->fd,
           nullptr /* range_del_agg */, nullptr,

db/c.cc

@@ -2121,9 +2121,9 @@ void rocksdb_options_set_use_direct_reads(rocksdb_options_t* opt,
   opt->rep.use_direct_reads = v;
 }
 
-void rocksdb_options_set_use_direct_writes(rocksdb_options_t* opt,
-                                           unsigned char v) {
-  opt->rep.use_direct_writes = v;
+void rocksdb_options_set_use_direct_io_for_flush_and_compaction(
+    rocksdb_options_t* opt, unsigned char v) {
+  opt->rep.use_direct_io_for_flush_and_compaction = v;
 }
 
 void rocksdb_options_set_allow_mmap_reads(

db/column_family_test.cc

@@ -676,6 +676,44 @@ TEST_F(ColumnFamilyTest, AddDrop) {
               std::vector<std::string>({"default", "four", "three"}));
 }
 
+TEST_F(ColumnFamilyTest, BulkAddDrop) {
+  constexpr int kNumCF = 1000;
+  ColumnFamilyOptions cf_options;
+  WriteOptions write_options;
+  Open();
+  std::vector<std::string> cf_names;
+  std::vector<ColumnFamilyHandle*> cf_handles;
+  for (int i = 1; i <= kNumCF; i++) {
+    cf_names.push_back("cf1-" + ToString(i));
+  }
+  ASSERT_OK(db_->CreateColumnFamilies(cf_options, cf_names, &cf_handles));
+  for (int i = 1; i <= kNumCF; i++) {
+    ASSERT_OK(db_->Put(write_options, cf_handles[i - 1], "foo", "bar"));
+  }
+  ASSERT_OK(db_->DropColumnFamilies(cf_handles));
+  std::vector<ColumnFamilyDescriptor> cf_descriptors;
+  for (auto* handle : cf_handles) {
+    delete handle;
+  }
+  cf_handles.clear();
+  for (int i = 1; i <= kNumCF; i++) {
+    cf_descriptors.emplace_back("cf2-" + ToString(i), ColumnFamilyOptions());
+  }
+  ASSERT_OK(db_->CreateColumnFamilies(cf_descriptors, &cf_handles));
+  for (int i = 1; i <= kNumCF; i++) {
+    ASSERT_OK(db_->Put(write_options, cf_handles[i - 1], "foo", "bar"));
+  }
+  ASSERT_OK(db_->DropColumnFamilies(cf_handles));
+  for (auto* handle : cf_handles) {
+    delete handle;
+  }
+  Close();
+  std::vector<std::string> families;
+  ASSERT_OK(DB::ListColumnFamilies(db_options_, dbname_, &families));
+  std::sort(families.begin(), families.end());
+  ASSERT_TRUE(families == std::vector<std::string>({"default"}));
+}
+
 TEST_F(ColumnFamilyTest, DropTest) {
   // first iteration - dont reopen DB before dropping
   // second iteration - reopen DB before dropping

db/compaction.cc

@@ -413,14 +413,15 @@ void Compaction::Summary(char* output, int len) {
 uint64_t Compaction::OutputFilePreallocationSize() const {
   uint64_t preallocation_size = 0;
 
-  if (cfd_->ioptions()->compaction_style == kCompactionStyleLevel ||
-      output_level() > 0) {
+  if (max_output_file_size_ != port::kMaxUint64 &&
+      (cfd_->ioptions()->compaction_style == kCompactionStyleLevel ||
+       output_level() > 0)) {
     preallocation_size = max_output_file_size_;
   } else {
-    // output_level() == 0
-    assert(num_input_levels() > 0);
-    for (const auto& f : inputs_[0].files) {
-      preallocation_size += f->fd.GetFileSize();
+    for (const auto& level_files : inputs_) {
+      for (const auto& file : level_files.files) {
+        preallocation_size += file->fd.GetFileSize();
+      }
     }
   }
   // Over-estimate slightly so we don't end up just barely crossing

db/compaction_iterator.cc

@@ -10,12 +10,36 @@
 
 namespace rocksdb {
 
+#ifndef ROCKSDB_LITE
+CompactionEventListener::CompactionListenerValueType fromInternalValueType(
+    ValueType vt) {
+  switch (vt) {
+    case kTypeDeletion:
+      return CompactionEventListener::CompactionListenerValueType::kDelete;
+    case kTypeValue:
+      return CompactionEventListener::CompactionListenerValueType::kValue;
+    case kTypeMerge:
+      return CompactionEventListener::CompactionListenerValueType::
+          kMergeOperand;
+    case kTypeSingleDeletion:
+      return CompactionEventListener::CompactionListenerValueType::
+          kSingleDelete;
+    case kTypeRangeDeletion:
+      return CompactionEventListener::CompactionListenerValueType::kRangeDelete;
+    default:
+      assert(false);
+      return CompactionEventListener::CompactionListenerValueType::kInvalid;
+  }
+}
+#endif  // ROCKSDB_LITE
+
 CompactionIterator::CompactionIterator(
     InternalIterator* input, const Comparator* cmp, MergeHelper* merge_helper,
     SequenceNumber last_sequence, std::vector<SequenceNumber>* snapshots,
     SequenceNumber earliest_write_conflict_snapshot, Env* env,
     bool expect_valid_internal_key, RangeDelAggregator* range_del_agg,
     const Compaction* compaction, const CompactionFilter* compaction_filter,
+    CompactionEventListener* compaction_listener,
     const std::atomic<bool>* shutting_down)
     : CompactionIterator(
           input, cmp, merge_helper, last_sequence, snapshots,
@@ -23,7 +47,7 @@ CompactionIterator::CompactionIterator(
           range_del_agg,
           std::unique_ptr<CompactionProxy>(
               compaction ? new CompactionProxy(compaction) : nullptr),
-          compaction_filter, shutting_down) {}
+          compaction_filter, compaction_listener, shutting_down) {}
 
 CompactionIterator::CompactionIterator(
     InternalIterator* input, const Comparator* cmp, MergeHelper* merge_helper,
@@ -32,6 +56,7 @@ CompactionIterator::CompactionIterator(
     bool expect_valid_internal_key, RangeDelAggregator* range_del_agg,
     std::unique_ptr<CompactionProxy> compaction,
     const CompactionFilter* compaction_filter,
+    CompactionEventListener* compaction_listener,
     const std::atomic<bool>* shutting_down)
     : input_(input),
       cmp_(cmp),
@@ -43,7 +68,9 @@ CompactionIterator::CompactionIterator(
       range_del_agg_(range_del_agg),
       compaction_(std::move(compaction)),
       compaction_filter_(compaction_filter),
+      compaction_listener_(compaction_listener),
       shutting_down_(shutting_down),
+      ignore_snapshots_(false),
       merge_out_iter_(merge_helper_) {
   assert(compaction_filter_ == nullptr || compaction_ != nullptr);
   bottommost_level_ =
@@ -62,8 +89,10 @@ CompactionIterator::CompactionIterator(
     earliest_snapshot_ = snapshots_->at(0);
     latest_snapshot_ = snapshots_->back();
   }
-  if (compaction_filter_ != nullptr && compaction_filter_->IgnoreSnapshots()) {
-    ignore_snapshots_ = true;
+  if (compaction_filter_ != nullptr) {
+    if (compaction_filter_->IgnoreSnapshots()) {
+      ignore_snapshots_ = true;
+    }
   } else {
     ignore_snapshots_ = false;
   }
@@ -188,6 +217,14 @@ void CompactionIterator::NextFromInput() {
       current_user_key_sequence_ = kMaxSequenceNumber;
       current_user_key_snapshot_ = 0;
 
+#ifndef ROCKSDB_LITE
+      if (compaction_listener_) {
+        compaction_listener_->OnCompaction(compaction_->level(), ikey_.user_key,
+                                           fromInternalValueType(ikey_.type),
+                                           value_, ikey_.sequence, true);
+      }
+#endif  // ROCKSDB_LITE
+
       // apply the compaction filter to the first occurrence of the user key
       if (compaction_filter_ != nullptr && ikey_.type == kTypeValue &&
           (visible_at_tip_ || ikey_.sequence > latest_snapshot_ ||
@@ -235,6 +272,14 @@ void CompactionIterator::NextFromInput() {
         }
       }
     } else {
+#ifndef ROCKSDB_LITE
+      if (compaction_listener_) {
+        compaction_listener_->OnCompaction(compaction_->level(), ikey_.user_key,
+                                           fromInternalValueType(ikey_.type),
+                                           value_, ikey_.sequence, false);
+      }
+#endif  // ROCKSDB_LITE
+
       // Update the current key to reflect the new sequence number/type without
       // copying the user key.
       // TODO(rven): Compaction filter does not process keys in this path
@@ -394,7 +439,7 @@ void CompactionIterator::NextFromInput() {
       // is the same as the visibility of a previous instance of the
       // same key, then this kv is not visible in any snapshot.
       // Hidden by an newer entry for same user key
-      // TODO: why not > ?
+      // TODO(noetzli): why not > ?
       //
       // Note: Dropping this key will not affect TransactionDB write-conflict
       // checking since there has already been a record returned for this key

db/compaction_iterator.h

@@ -20,6 +20,8 @@
 
 namespace rocksdb {
 
+class CompactionEventListener;
+
 class CompactionIterator {
  public:
   // A wrapper around Compaction. Has a much smaller interface, only what
@@ -60,6 +62,7 @@ class CompactionIterator {
                      RangeDelAggregator* range_del_agg,
                      const Compaction* compaction = nullptr,
                      const CompactionFilter* compaction_filter = nullptr,
+                     CompactionEventListener* compaction_listener = nullptr,
                      const std::atomic<bool>* shutting_down = nullptr);
 
   // Constructor with custom CompactionProxy, used for tests.
@@ -71,6 +74,7 @@ class CompactionIterator {
                      RangeDelAggregator* range_del_agg,
                      std::unique_ptr<CompactionProxy> compaction,
                      const CompactionFilter* compaction_filter = nullptr,
+                     CompactionEventListener* compaction_listener = nullptr,
                      const std::atomic<bool>* shutting_down = nullptr);
 
   ~CompactionIterator();
@@ -124,6 +128,7 @@ class CompactionIterator {
   RangeDelAggregator* range_del_agg_;
   std::unique_ptr<CompactionProxy> compaction_;
   const CompactionFilter* compaction_filter_;
+  CompactionEventListener* compaction_listener_;
   const std::atomic<bool>* shutting_down_;
   bool bottommost_level_;
   bool valid_ = false;

db/compaction_iterator_test.cc

@@ -189,7 +189,7 @@ class CompactionIteratorTest : public testing::Test {
     c_iter_.reset(new CompactionIterator(
         iter_.get(), cmp_, merge_helper_.get(), last_sequence, &snapshots_,
         kMaxSequenceNumber, Env::Default(), false, range_del_agg_.get(),
-        std::move(compaction), filter, &shutting_down_));
+        std::move(compaction), filter, nullptr, &shutting_down_));
   }
 
   void AddSnapshot(SequenceNumber snapshot) { snapshots_.push_back(snapshot); }

db/compaction_job.cc

@@ -739,12 +739,23 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
     input->SeekToFirst();
   }
 
+  // we allow only 1 compaction event listener. Used by blob storage
+  CompactionEventListener* comp_event_listener = nullptr;
+#ifndef ROCKSDB_LITE
+  for (auto& celitr : cfd->ioptions()->listeners) {
+    comp_event_listener = celitr->GetCompactionEventListener();
+    if (comp_event_listener != nullptr) {
+      break;
+    }
+  }
+#endif  // ROCKSDB_LITE
+
   Status status;
   sub_compact->c_iter.reset(new CompactionIterator(
       input.get(), cfd->user_comparator(), &merge, versions_->LastSequence(),
       &existing_snapshots_, earliest_write_conflict_snapshot_, env_, false,
       range_del_agg.get(), sub_compact->compaction, compaction_filter,
-      shutting_down_));
+      comp_event_listener, shutting_down_));
   auto c_iter = sub_compact->c_iter.get();
   c_iter->SeekToFirst();
   if (c_iter->Valid() &&
@@ -1072,6 +1083,11 @@ Status CompactionJob::FinishCompactionOutputFile(
   TableProperties tp;
   if (s.ok() && current_entries > 0) {
     // Verify that the table is usable
+    // We set for_compaction to false and don't OptimizeForCompactionTableRead
+    // here because this is a special case after we finish the table building
+    // No matter whether use_direct_io_for_flush_and_compaction is true,
+    // we will regrad this verification as user reads since the goal is
+    // to cache it here for further user reads
     InternalIterator* iter = cfd->table_cache()->NewIterator(
         ReadOptions(), env_options_, cfd->internal_comparator(), meta->fd,
         nullptr /* range_del_agg */, nullptr,
@@ -1198,7 +1214,11 @@ Status CompactionJob::OpenCompactionOutputFile(
 #endif  // !ROCKSDB_LITE
   // Make the output file
   unique_ptr<WritableFile> writable_file;
-  Status s = NewWritableFile(env_, fname, &writable_file, env_options_);
+  EnvOptions opt_env_opts =
+      env_->OptimizeForCompactionTableWrite(env_options_, db_options_);
+  TEST_SYNC_POINT_CALLBACK("CompactionJob::OpenCompactionOutputFile",
+                           &opt_env_opts.use_direct_writes);
+  Status s = NewWritableFile(env_, fname, &writable_file, opt_env_opts);
   if (!s.ok()) {
     ROCKS_LOG_ERROR(
         db_options_.info_log,

db/db_compaction_test.cc

@@ -40,6 +40,12 @@ class DBCompactionTestWithParam
   bool exclusive_manual_compaction_;
 };
 
+class DBCompactionDirectIOTest : public DBCompactionTest,
+                                 public ::testing::WithParamInterface<bool> {
+ public:
+  DBCompactionDirectIOTest() : DBCompactionTest() {}
+};
+
 namespace {
 
 class FlushedFileCollector : public EventListener {
@@ -772,6 +778,41 @@ TEST_F(DBCompactionTest, ZeroSeqIdCompaction) {
   ASSERT_OK(Put("", ""));
 }
 
+TEST_F(DBCompactionTest, ManualCompactionUnknownOutputSize) {
+  // github issue #2249
+  Options options = CurrentOptions();
+  options.compaction_style = kCompactionStyleLevel;
+  options.level0_file_num_compaction_trigger = 3;
+  DestroyAndReopen(options);
+
+  // create two files in l1 that we can compact
+  for (int i = 0; i < 2; ++i) {
+    for (int j = 0; j < options.level0_file_num_compaction_trigger; j++) {
+      // make l0 files' ranges overlap to avoid trivial move
+      Put(std::to_string(2 * i), std::string(1, 'A'));
+      Put(std::to_string(2 * i + 1), std::string(1, 'A'));
+      Flush();
+      dbfull()->TEST_WaitForFlushMemTable();
+    }
+    dbfull()->TEST_WaitForCompact();
+    ASSERT_EQ(NumTableFilesAtLevel(0, 0), 0);
+    ASSERT_EQ(NumTableFilesAtLevel(1, 0), i + 1);
+  }
+
+  ColumnFamilyMetaData cf_meta;
+  dbfull()->GetColumnFamilyMetaData(dbfull()->DefaultColumnFamily(), &cf_meta);
+  ASSERT_EQ(2, cf_meta.levels[1].files.size());
+  std::vector<std::string> input_filenames;
+  for (const auto& sst_file : cf_meta.levels[1].files) {
+    input_filenames.push_back(sst_file.name);
+  }
+
+  // note CompactionOptions::output_file_size_limit is unset.
+  CompactionOptions compact_opt;
+  compact_opt.compression = kNoCompression;
+  dbfull()->CompactFiles(compact_opt, input_filenames, 1);
+}
+
 // Check that writes done during a memtable compaction are recovered
 // if the database is shutdown during the memtable compaction.
 TEST_F(DBCompactionTest, RecoverDuringMemtableCompaction) {
@@ -1843,6 +1884,47 @@ TEST_F(DBCompactionTest, L0_CompactionBug_Issue44_b) {
   } while (ChangeCompactOptions());
 }
 
+TEST_F(DBCompactionTest, ManualAutoRace) {
+  CreateAndReopenWithCF({"pikachu"}, CurrentOptions());
+  rocksdb::SyncPoint::GetInstance()->LoadDependency(
+      {{"DBImpl::BGWorkCompaction", "DBCompactionTest::ManualAutoRace:1"},
+       {"DBImpl::RunManualCompaction:WaitScheduled",
+        "BackgroundCallCompaction:0"}});
+
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  Put(1, "foo", "");
+  Put(1, "bar", "");
+  Flush(1);
+  Put(1, "foo", "");
+  Put(1, "bar", "");
+  // Generate four files in CF 0, which should trigger an auto compaction
+  Put("foo", "");
+  Put("bar", "");
+  Flush();
+  Put("foo", "");
+  Put("bar", "");
+  Flush();
+  Put("foo", "");
+  Put("bar", "");
+  Flush();
+  Put("foo", "");
+  Put("bar", "");
+  Flush();
+
+  // The auto compaction is scheduled but waited until here
+  TEST_SYNC_POINT("DBCompactionTest::ManualAutoRace:1");
+  // The auto compaction will wait until the the manual compaction is registerd
+  // before processing so that it will be cancelled.
+  dbfull()->CompactRange(CompactRangeOptions(), handles_[1], nullptr, nullptr);
+  ASSERT_EQ("0,1", FilesPerLevel(1));
+
+  // Eventually the cancelled compaction will be rescheduled and executed.
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ("0,1", FilesPerLevel(0));
+  rocksdb::SyncPoint::GetInstance()->DisableProcessing();
+}
+
 TEST_P(DBCompactionTestWithParam, ManualCompaction) {
   Options options = CurrentOptions();
   options.max_subcompactions = max_subcompactions_;
@@ -2552,6 +2634,47 @@ INSTANTIATE_TEST_CASE_P(DBCompactionTestWithParam, DBCompactionTestWithParam,
                                           std::make_tuple(4, true),
                                           std::make_tuple(4, false)));
 
+TEST_P(DBCompactionDirectIOTest, DirectIO) {
+  Options options = CurrentOptions();
+  Destroy(options);
+  options.create_if_missing = true;
+  options.disable_auto_compactions = true;
+  options.use_direct_io_for_flush_and_compaction = GetParam();
+  options.env = new MockEnv(Env::Default());
+  Reopen(options);
+  bool readahead = false;
+  SyncPoint::GetInstance()->SetCallBack(
+      "TableCache::NewIterator:for_compaction", [&](void* arg) {
+        bool* use_direct_reads = static_cast<bool*>(arg);
+        ASSERT_EQ(*use_direct_reads,
+                  options.use_direct_io_for_flush_and_compaction);
+      });
+  SyncPoint::GetInstance()->SetCallBack(
+      "CompactionJob::OpenCompactionOutputFile", [&](void* arg) {
+        bool* use_direct_writes = static_cast<bool*>(arg);
+        ASSERT_EQ(*use_direct_writes,
+                  options.use_direct_io_for_flush_and_compaction);
+      });
+  if (options.use_direct_io_for_flush_and_compaction) {
+    SyncPoint::GetInstance()->SetCallBack(
+        "SanitizeOptions:direct_io", [&](void* arg) {
+          readahead = true;
+        });
+  }
+  SyncPoint::GetInstance()->EnableProcessing();
+  CreateAndReopenWithCF({"pikachu"}, options);
+  MakeTables(3, "p", "q", 1);
+  ASSERT_EQ("1,1,1", FilesPerLevel(1));
+  Compact(1, "p1", "p9");
+  ASSERT_FALSE(readahead ^ options.use_direct_io_for_flush_and_compaction);
+  ASSERT_EQ("0,0,1", FilesPerLevel(1));
+  Destroy(options);
+  delete options.env;
+}
+
+INSTANTIATE_TEST_CASE_P(DBCompactionDirectIOTest, DBCompactionDirectIOTest,
+                        testing::Bool());
+
 class CompactionPriTest : public DBTestBase,
                           public testing::WithParamInterface<uint32_t> {
  public:

db/db_flush_test.cc

@@ -19,6 +19,12 @@ class DBFlushTest : public DBTestBase {
   DBFlushTest() : DBTestBase("/db_flush_test") {}
 };
 
+class DBFlushDirectIOTest : public DBFlushTest,
+                            public ::testing::WithParamInterface<bool> {
+ public:
+  DBFlushDirectIOTest() : DBFlushTest() {}
+};
+
 // We had issue when two background threads trying to flush at the same time,
 // only one of them get committed. The test verifies the issue is fixed.
 TEST_F(DBFlushTest, FlushWhileWritingManifest) {
@@ -83,6 +89,33 @@ TEST_F(DBFlushTest, SyncFail) {
   Destroy(options);
 }
 
+TEST_P(DBFlushDirectIOTest, DirectIO) {
+  Options options;
+  options.create_if_missing = true;
+  options.disable_auto_compactions = true;
+  options.max_background_flushes = 2;
+  options.use_direct_io_for_flush_and_compaction = GetParam();
+  options.env = new MockEnv(Env::Default());
+  SyncPoint::GetInstance()->SetCallBack(
+      "BuildTable:create_file", [&](void* arg) {
+        bool* use_direct_writes = static_cast<bool*>(arg);
+        ASSERT_EQ(*use_direct_writes,
+                  options.use_direct_io_for_flush_and_compaction);
+      });
+
+  SyncPoint::GetInstance()->EnableProcessing();
+  Reopen(options);
+  ASSERT_OK(Put("foo", "v"));
+  FlushOptions flush_options;
+  flush_options.wait = true;
+  ASSERT_OK(dbfull()->Flush(flush_options));
+  Destroy(options);
+  delete options.env;
+}
+
+INSTANTIATE_TEST_CASE_P(DBFlushDirectIOTest, DBFlushDirectIOTest,
+                        testing::Bool());
+
 }  // namespace rocksdb
 
 int main(int argc, char** argv) {

db/db_impl.cc

@@ -514,9 +514,8 @@ Status DBImpl::SetOptions(ColumnFamilyHandle* column_family,
           InstallSuperVersionAndScheduleWork(cfd, nullptr, new_options);
       delete old_sv;
 
-      write_thread_.EnterUnbatched(&w, &mutex_);
-      persist_options_status = WriteOptionsFile();
-      write_thread_.ExitUnbatched(&w);
+      persist_options_status = WriteOptionsFile(
+          false /*need_mutex_lock*/, true /*need_enter_write_thread*/);
     }
   }
 
@@ -532,14 +531,7 @@ Status DBImpl::SetOptions(ColumnFamilyHandle* column_family,
                    "[%s] SetOptions() succeeded", cfd->GetName().c_str());
     new_options.Dump(immutable_db_options_.info_log.get());
     if (!persist_options_status.ok()) {
-      if (immutable_db_options_.fail_if_options_file_error) {
-        s = Status::IOError(
-            "SetOptions() succeeded, but unable to persist options",
-            persist_options_status.ToString());
-      }
-      ROCKS_LOG_WARN(immutable_db_options_.info_log,
-                     "Unable to persist options in SetOptions() -- %s",
-                     persist_options_status.ToString().c_str());
+      s = persist_options_status;
     }
   } else {
     ROCKS_LOG_WARN(immutable_db_options_.info_log, "[%s] SetOptions() failed",
@@ -591,7 +583,8 @@ Status DBImpl::SetDBOptions(
                          purge_wal_status.ToString().c_str());
         }
       }
-      persist_options_status = WriteOptionsFile();
+      persist_options_status = WriteOptionsFile(
+          false /*need_mutex_lock*/, false /*need_enter_write_thread*/);
       write_thread_.ExitUnbatched(&w);
     }
   }
@@ -1121,8 +1114,76 @@ std::vector<Status> DBImpl::MultiGet(
 }
 
 Status DBImpl::CreateColumnFamily(const ColumnFamilyOptions& cf_options,
-                                  const std::string& column_family_name,
+                                  const std::string& column_family,
                                   ColumnFamilyHandle** handle) {
+  assert(handle != nullptr);
+  Status s = CreateColumnFamilyImpl(cf_options, column_family, handle);
+  if (s.ok()) {
+    s = WriteOptionsFile(true /*need_mutex_lock*/,
+                         true /*need_enter_write_thread*/);
+  }
+  return s;
+}
+
+Status DBImpl::CreateColumnFamilies(
+    const ColumnFamilyOptions& cf_options,
+    const std::vector<std::string>& column_family_names,
+    std::vector<ColumnFamilyHandle*>* handles) {
+  assert(handles != nullptr);
+  handles->clear();
+  size_t num_cf = column_family_names.size();
+  Status s;
+  bool success_once = false;
+  for (size_t i = 0; i < num_cf; i++) {
+    ColumnFamilyHandle* handle;
+    s = CreateColumnFamilyImpl(cf_options, column_family_names[i], &handle);
+    if (!s.ok()) {
+      break;
+    }
+    handles->push_back(handle);
+    success_once = true;
+  }
+  if (success_once) {
+    Status persist_options_status = WriteOptionsFile(
+        true /*need_mutex_lock*/, true /*need_enter_write_thread*/);
+    if (s.ok() && !persist_options_status.ok()) {
+      s = persist_options_status;
+    }
+  }
+  return s;
+}
+
+Status DBImpl::CreateColumnFamilies(
+    const std::vector<ColumnFamilyDescriptor>& column_families,
+    std::vector<ColumnFamilyHandle*>* handles) {
+  assert(handles != nullptr);
+  handles->clear();
+  size_t num_cf = column_families.size();
+  Status s;
+  bool success_once = false;
+  for (size_t i = 0; i < num_cf; i++) {
+    ColumnFamilyHandle* handle;
+    s = CreateColumnFamilyImpl(column_families[i].options,
+                               column_families[i].name, &handle);
+    if (!s.ok()) {
+      break;
+    }
+    handles->push_back(handle);
+    success_once = true;
+  }
+  if (success_once) {
+    Status persist_options_status = WriteOptionsFile(
+        true /*need_mutex_lock*/, true /*need_enter_write_thread*/);
+    if (s.ok() && !persist_options_status.ok()) {
+      s = persist_options_status;
+    }
+  }
+  return s;
+}
+
+Status DBImpl::CreateColumnFamilyImpl(const ColumnFamilyOptions& cf_options,
+                                      const std::string& column_family_name,
+                                      ColumnFamilyHandle** handle) {
   Status s;
   Status persist_options_status;
   *handle = nullptr;
@@ -1159,12 +1220,6 @@ Status DBImpl::CreateColumnFamily(const ColumnFamilyOptions& cf_options,
       s = versions_->LogAndApply(nullptr, MutableCFOptions(cf_options), &edit,
                                  &mutex_, directories_.GetDbDir(), false,
                                  &cf_options);
-
-      if (s.ok()) {
-        // If the column family was created successfully, we then persist
-        // the updated RocksDB options under the same single write thread
-        persist_options_status = WriteOptionsFile();
-      }
       write_thread_.ExitUnbatched(&w);
     }
     if (s.ok()) {
@@ -1194,22 +1249,42 @@ Status DBImpl::CreateColumnFamily(const ColumnFamilyOptions& cf_options,
   if (s.ok()) {
     NewThreadStatusCfInfo(
         reinterpret_cast<ColumnFamilyHandleImpl*>(*handle)->cfd());
-    if (!persist_options_status.ok()) {
-      if (immutable_db_options_.fail_if_options_file_error) {
-        s = Status::IOError(
-            "ColumnFamily has been created, but unable to persist"
-            "options in CreateColumnFamily()",
-            persist_options_status.ToString().c_str());
-      }
-      ROCKS_LOG_WARN(immutable_db_options_.info_log,
-                     "Unable to persist options in CreateColumnFamily() -- %s",
-                     persist_options_status.ToString().c_str());
-    }
   }
   return s;
 }
 
 Status DBImpl::DropColumnFamily(ColumnFamilyHandle* column_family) {
+  assert(column_family != nullptr);
+  Status s = DropColumnFamilyImpl(column_family);
+  if (s.ok()) {
+    s = WriteOptionsFile(true /*need_mutex_lock*/,
+                         true /*need_enter_write_thread*/);
+  }
+  return s;
+}
+
+Status DBImpl::DropColumnFamilies(
+    const std::vector<ColumnFamilyHandle*>& column_families) {
+  Status s;
+  bool success_once = false;
+  for (auto* handle : column_families) {
+    s = DropColumnFamilyImpl(handle);
+    if (!s.ok()) {
+      break;
+    }
+    success_once = true;
+  }
+  if (success_once) {
+    Status persist_options_status = WriteOptionsFile(
+        true /*need_mutex_lock*/, true /*need_enter_write_thread*/);
+    if (s.ok() && !persist_options_status.ok()) {
+      s = persist_options_status;
+    }
+  }
+  return s;
+}
+
+Status DBImpl::DropColumnFamilyImpl(ColumnFamilyHandle* column_family) {
   auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
   auto cfd = cfh->cfd();
   if (cfd->GetID() == 0) {
@@ -1223,7 +1298,6 @@ Status DBImpl::DropColumnFamily(ColumnFamilyHandle* column_family) {
   edit.SetColumnFamily(cfd->GetID());
 
   Status s;
-  Status options_persist_status;
   {
     InstrumentedMutexLock l(&mutex_);
     if (cfd->IsDropped()) {
@@ -1235,11 +1309,6 @@ Status DBImpl::DropColumnFamily(ColumnFamilyHandle* column_family) {
       write_thread_.EnterUnbatched(&w, &mutex_);
       s = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
                                  &edit, &mutex_);
-      if (s.ok()) {
-        // If the column family was dropped successfully, we then persist
-        // the updated RocksDB options under the same single write thread
-        options_persist_status = WriteOptionsFile();
-      }
       write_thread_.ExitUnbatched(&w);
     }
 
@@ -1268,18 +1337,6 @@ Status DBImpl::DropColumnFamily(ColumnFamilyHandle* column_family) {
                                   mutable_cf_options->max_write_buffer_number;
     ROCKS_LOG_INFO(immutable_db_options_.info_log,
                    "Dropped column family with id %u\n", cfd->GetID());
-
-    if (!options_persist_status.ok()) {
-      if (immutable_db_options_.fail_if_options_file_error) {
-        s = Status::IOError(
-            "ColumnFamily has been dropped, but unable to persist "
-            "options in DropColumnFamily()",
-            options_persist_status.ToString().c_str());
-      }
-      ROCKS_LOG_WARN(immutable_db_options_.info_log,
-                     "Unable to persist options in DropColumnFamily() -- %s",
-                     options_persist_status.ToString().c_str());
-    }
   } else {
     ROCKS_LOG_ERROR(immutable_db_options_.info_log,
                     "Dropping column family with id %u FAILED -- %s\n",
@@ -1515,6 +1572,14 @@ void DBImpl::ReleaseSnapshot(const Snapshot* s) {
   delete casted_s;
 }
 
+bool DBImpl::HasActiveSnapshotLaterThanSN(SequenceNumber sn) {
+  InstrumentedMutexLock l(&mutex_);
+  if (snapshots_.empty()) {
+    return false;
+  }
+  return (snapshots_.newest()->GetSequenceNumber() > sn);
+}
+
 #ifndef ROCKSDB_LITE
 Status DBImpl::GetPropertiesOfAllTables(ColumnFamilyHandle* column_family,
                                         TablePropertiesCollection* props) {
@@ -1666,6 +1731,16 @@ bool DBImpl::GetIntPropertyInternal(ColumnFamilyData* cfd,
   }
 }
 
+#ifndef ROCKSDB_LITE
+Status DBImpl::ResetStats() {
+  InstrumentedMutexLock l(&mutex_);
+  for (auto* cfd : *versions_->GetColumnFamilySet()) {
+    cfd->internal_stats()->Clear();
+  }
+  return Status::OK();
+}
+#endif  // ROCKSDB_LITE
+
 bool DBImpl::GetAggregatedIntProperty(const Slice& property,
                                       uint64_t* aggregated_value) {
   const DBPropertyInfo* property_info = GetPropertyInfo(property);
@@ -1749,6 +1824,20 @@ ColumnFamilyHandle* DBImpl::GetColumnFamilyHandle(uint32_t column_family_id) {
   return cf_memtables->GetColumnFamilyHandle();
 }
 
+// REQUIRED: mutex is NOT held.
+ColumnFamilyHandle* DBImpl::GetColumnFamilyHandleUnlocked(
+    uint32_t column_family_id) {
+  ColumnFamilyMemTables* cf_memtables = column_family_memtables_.get();
+
+  InstrumentedMutexLock l(&mutex_);
+
+  if (!cf_memtables->Seek(column_family_id)) {
+    return nullptr;
+  }
+
+  return cf_memtables->GetColumnFamilyHandle();
+}
+
 void DBImpl::GetApproximateMemTableStats(ColumnFamilyHandle* column_family,
                                          const Range& range,
                                          uint64_t* const count,
@@ -2097,9 +2186,29 @@ Status DB::CreateColumnFamily(const ColumnFamilyOptions& cf_options,
                               ColumnFamilyHandle** handle) {
   return Status::NotSupported("");
 }
+
+Status DB::CreateColumnFamilies(
+    const ColumnFamilyOptions& cf_options,
+    const std::vector<std::string>& column_family_names,
+    std::vector<ColumnFamilyHandle*>* handles) {
+  return Status::NotSupported("");
+}
+
+Status DB::CreateColumnFamilies(
+    const std::vector<ColumnFamilyDescriptor>& column_families,
+    std::vector<ColumnFamilyHandle*>* handles) {
+  return Status::NotSupported("");
+}
+
 Status DB::DropColumnFamily(ColumnFamilyHandle* column_family) {
   return Status::NotSupported("");
 }
+
+Status DB::DropColumnFamilies(
+    const std::vector<ColumnFamilyHandle*>& column_families) {
+  return Status::NotSupported("");
+}
+
 Status DB::DestroyColumnFamilyHandle(ColumnFamilyHandle* column_family) {
   delete column_family;
   return Status::OK();
@@ -2206,9 +2315,18 @@ Status DestroyDB(const std::string& dbname, const Options& options) {
   return result;
 }
 
-Status DBImpl::WriteOptionsFile() {
+Status DBImpl::WriteOptionsFile(bool need_mutex_lock,
+                                bool need_enter_write_thread) {
 #ifndef ROCKSDB_LITE
-  mutex_.AssertHeld();
+  WriteThread::Writer w;
+  if (need_mutex_lock) {
+    mutex_.Lock();
+  } else {
+    mutex_.AssertHeld();
+  }
+  if (need_enter_write_thread) {
+    write_thread_.EnterUnbatched(&w, &mutex_);
+  }
 
   std::vector<std::string> cf_names;
   std::vector<ColumnFamilyOptions> cf_opts;
@@ -2236,11 +2354,23 @@ Status DBImpl::WriteOptionsFile() {
   if (s.ok()) {
     s = RenameTempFileToOptionsFile(file_name);
   }
-  mutex_.Lock();
-  return s;
-#else
-  return Status::OK();
+  // restore lock
+  if (!need_mutex_lock) {
+    mutex_.Lock();
+  }
+  if (need_enter_write_thread) {
+    write_thread_.ExitUnbatched(&w);
+  }
+  if (!s.ok()) {
+    ROCKS_LOG_WARN(immutable_db_options_.info_log,
+                   "Unnable to persist options -- %s", s.ToString().c_str());
+    if (immutable_db_options_.fail_if_options_file_error) {
+      return Status::IOError("Unable to persist options.",
+                             s.ToString().c_str());
+    }
+  }
 #endif  // !ROCKSDB_LITE
+  return Status::OK();
 }
 
 #ifndef ROCKSDB_LITE

db/db_impl.h

@@ -100,10 +100,19 @@ class DBImpl : public DB {
       const std::vector<Slice>& keys,
       std::vector<std::string>* values) override;
 
-  virtual Status CreateColumnFamily(const ColumnFamilyOptions& options,
+  virtual Status CreateColumnFamily(const ColumnFamilyOptions& cf_options,
                                     const std::string& column_family,
                                     ColumnFamilyHandle** handle) override;
+  virtual Status CreateColumnFamilies(
+      const ColumnFamilyOptions& cf_options,
+      const std::vector<std::string>& column_family_names,
+      std::vector<ColumnFamilyHandle*>* handles) override;
+  virtual Status CreateColumnFamilies(
+      const std::vector<ColumnFamilyDescriptor>& column_families,
+      std::vector<ColumnFamilyHandle*>* handles) override;
   virtual Status DropColumnFamily(ColumnFamilyHandle* column_family) override;
+  virtual Status DropColumnFamilies(
+      const std::vector<ColumnFamilyHandle*>& column_families) override;
 
   // Returns false if key doesn't exist in the database and true if it may.
   // If value_found is not passed in as null, then return the value if found in
@@ -192,7 +201,11 @@ class DBImpl : public DB {
 
   virtual SequenceNumber GetLatestSequenceNumber() const override;
 
+  bool HasActiveSnapshotLaterThanSN(SequenceNumber sn);
+
 #ifndef ROCKSDB_LITE
+  using DB::ResetStats;
+  virtual Status ResetStats() override;
   virtual Status DisableFileDeletions() override;
   virtual Status EnableFileDeletions(bool force) override;
   virtual int IsFileDeletionsEnabled() const;
@@ -452,6 +465,9 @@ class DBImpl : public DB {
   // mutex is released.
   ColumnFamilyHandle* GetColumnFamilyHandle(uint32_t column_family_id);
 
+  // Same as above, should called without mutex held and not on write thread.
+  ColumnFamilyHandle* GetColumnFamilyHandleUnlocked(uint32_t column_family_id);
+
   // Returns the number of currently running flushes.
   // REQUIREMENT: mutex_ must be held when calling this function.
   int num_running_flushes() {
@@ -547,9 +563,10 @@ class DBImpl : public DB {
                                         RangeDelAggregator* range_del_agg);
 
   // Except in DB::Open(), WriteOptionsFile can only be called when:
-  // 1. WriteThread::Writer::EnterUnbatched() is used.
-  // 2. db_mutex is held
-  Status WriteOptionsFile();
+  // Persist options to options file.
+  // If need_mutex_lock = false, the method will lock DB mutex.
+  // If need_enter_write_thread = false, the method will enter write thread.
+  Status WriteOptionsFile(bool need_mutex_lock, bool need_enter_write_thread);
 
   // The following two functions can only be called when:
   // 1. WriteThread::Writer::EnterUnbatched() is used.
@@ -557,6 +574,10 @@ class DBImpl : public DB {
   Status RenameTempFileToOptionsFile(const std::string& file_name);
   Status DeleteObsoleteOptionsFiles();
 
+  void NotifyOnFlushBegin(ColumnFamilyData* cfd, FileMetaData* file_meta,
+                          const MutableCFOptions& mutable_cf_options,
+                          int job_id, TableProperties prop);
+
   void NotifyOnFlushCompleted(ColumnFamilyData* cfd, FileMetaData* file_meta,
                               const MutableCFOptions& mutable_cf_options,
                               int job_id, TableProperties prop);
@@ -628,6 +649,12 @@ class DBImpl : public DB {
 
   const Status CreateArchivalDirectory();
 
+  Status CreateColumnFamilyImpl(const ColumnFamilyOptions& cf_options,
+                                const std::string& cf_name,
+                                ColumnFamilyHandle** handle);
+
+  Status DropColumnFamilyImpl(ColumnFamilyHandle* column_family);
+
   // Delete any unneeded files and stale in-memory entries.
   void DeleteObsoleteFiles();
   // Delete obsolete files and log status and information of file deletion
@@ -1091,13 +1118,6 @@ class DBImpl : public DB {
   DBImpl(const DBImpl&);
   void operator=(const DBImpl&);
 
-  // Return the earliest snapshot where seqno is visible.
-  // Store the snapshot right before that, if any, in prev_snapshot
-  inline SequenceNumber findEarliestVisibleSnapshot(
-    SequenceNumber in,
-    std::vector<SequenceNumber>& snapshots,
-    SequenceNumber* prev_snapshot);
-
   // Background threads call this function, which is just a wrapper around
   // the InstallSuperVersion() function. Background threads carry
   // job_context which can have new_superversion already

db/db_impl_compaction_flush.cc

@@ -90,6 +90,12 @@ Status DBImpl::FlushMemTableToOutputFile(
 
   flush_job.PickMemTable();
 
+#ifndef ROCKSDB_LITE
+  // may temporarily unlock and lock the mutex.
+  NotifyOnFlushBegin(cfd, &file_meta, mutable_cf_options, job_context->job_id,
+                     flush_job.GetTableProperties());
+#endif  // ROCKSDB_LITE
+
   Status s;
   if (logfile_number_ > 0 &&
       versions_->GetColumnFamilySet()->NumberOfColumnFamilies() > 0) {
@@ -156,6 +162,49 @@ Status DBImpl::FlushMemTableToOutputFile(
   return s;
 }
 
+void DBImpl::NotifyOnFlushBegin(ColumnFamilyData* cfd, FileMetaData* file_meta,
+                                const MutableCFOptions& mutable_cf_options,
+                                int job_id, TableProperties prop) {
+#ifndef ROCKSDB_LITE
+  if (immutable_db_options_.listeners.size() == 0U) {
+    return;
+  }
+  mutex_.AssertHeld();
+  if (shutting_down_.load(std::memory_order_acquire)) {
+    return;
+  }
+  bool triggered_writes_slowdown =
+      (cfd->current()->storage_info()->NumLevelFiles(0) >=
+       mutable_cf_options.level0_slowdown_writes_trigger);
+  bool triggered_writes_stop =
+      (cfd->current()->storage_info()->NumLevelFiles(0) >=
+       mutable_cf_options.level0_stop_writes_trigger);
+  // release lock while notifying events
+  mutex_.Unlock();
+  {
+    FlushJobInfo info;
+    info.cf_name = cfd->GetName();
+    // TODO(yhchiang): make db_paths dynamic in case flush does not
+    //                 go to L0 in the future.
+    info.file_path = MakeTableFileName(immutable_db_options_.db_paths[0].path,
+                                       file_meta->fd.GetNumber());
+    info.thread_id = env_->GetThreadID();
+    info.job_id = job_id;
+    info.triggered_writes_slowdown = triggered_writes_slowdown;
+    info.triggered_writes_stop = triggered_writes_stop;
+    info.smallest_seqno = file_meta->smallest_seqno;
+    info.largest_seqno = file_meta->largest_seqno;
+    info.table_properties = prop;
+    for (auto listener : immutable_db_options_.listeners) {
+      listener->OnFlushBegin(this, info);
+    }
+  }
+  mutex_.Lock();
+// no need to signal bg_cv_ as it will be signaled at the end of the
+// flush process.
+#endif  // ROCKSDB_LITE
+}
+
 void DBImpl::NotifyOnFlushCompleted(ColumnFamilyData* cfd,
                                     FileMetaData* file_meta,
                                     const MutableCFOptions& mutable_cf_options,
@@ -777,6 +826,7 @@ Status DBImpl::RunManualCompaction(ColumnFamilyData* cfd, int input_level,
   TEST_SYNC_POINT_CALLBACK("DBImpl::RunManualCompaction:NotScheduled", &mutex_);
   if (exclusive) {
     while (bg_compaction_scheduled_ > 0) {
+      TEST_SYNC_POINT("DBImpl::RunManualCompaction:WaitScheduled");
       ROCKS_LOG_INFO(
           immutable_db_options_.info_log,
           "[%s] Manual compaction waiting for all other scheduled background "
@@ -1343,6 +1393,16 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
                : m->manual_end->DebugString().c_str()));
     }
   } else if (!compaction_queue_.empty()) {
+    if (HaveManualCompaction(compaction_queue_.front())) {
+      // Can't compact right now, but try again later
+      TEST_SYNC_POINT("DBImpl::BackgroundCompaction()::Conflict");
+
+      // Stay in the compaciton queue.
+      unscheduled_compactions_++;
+
+      return Status::OK();
+    }
+
     // cfd is referenced here
     auto cfd = PopFirstFromCompactionQueue();
     // We unreference here because the following code will take a Ref() on
@@ -1357,12 +1417,6 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
       return Status::OK();
     }
 
-    if (HaveManualCompaction(cfd)) {
-      // Can't compact right now, but try again later
-      TEST_SYNC_POINT("DBImpl::BackgroundCompaction()::Conflict");
-      return Status::OK();
-    }
-
     // Pick up latest mutable CF Options and use it throughout the
     // compaction job
     // Compaction makes a copy of the latest MutableCFOptions. It should be used

db/db_impl_files.cc

@@ -444,6 +444,7 @@ void DBImpl::PurgeObsoleteFiles(const JobContext& state, bool schedule_only) {
       case kIdentityFile:
       case kMetaDatabase:
       case kOptionsFile:
+      case kBlobFile:
         keep = true;
         break;
     }

db/db_impl_open.cc

@@ -97,11 +97,14 @@ DBOptions SanitizeOptions(const std::string& dbname, const DBOptions& src) {
     result.db_paths.emplace_back(dbname, std::numeric_limits<uint64_t>::max());
   }
 
-  if (result.use_direct_reads && result.compaction_readahead_size == 0) {
+  if (result.use_direct_io_for_flush_and_compaction &&
+      result.compaction_readahead_size == 0) {
+    TEST_SYNC_POINT_CALLBACK("SanitizeOptions:direct_io", nullptr);
     result.compaction_readahead_size = 1024 * 1024 * 2;
   }
 
-  if (result.compaction_readahead_size > 0) {
+  if (result.compaction_readahead_size > 0 ||
+      result.use_direct_io_for_flush_and_compaction) {
     result.new_table_reader_for_compaction_inputs = true;
   }
 
@@ -165,10 +168,12 @@ static Status ValidateOptions(
         "then direct I/O reads (use_direct_reads) must be disabled. ");
   }
 
-  if (db_options.allow_mmap_writes && db_options.use_direct_writes) {
+  if (db_options.allow_mmap_writes &&
+      db_options.use_direct_io_for_flush_and_compaction) {
     return Status::NotSupported(
         "If memory mapped writes (allow_mmap_writes) are enabled "
-        "then direct I/O writes (use_direct_writes) must be disabled. ");
+        "then direct I/O writes (use_direct_io_for_flush_and_compaction) must "
+        "be disabled. ");
   }
 
   if (db_options.keep_log_file_num == 0) {
@@ -823,9 +828,11 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
       std::vector<SequenceNumber> snapshot_seqs =
           snapshots_.GetAll(&earliest_write_conflict_snapshot);
 
+      EnvOptions optimized_env_options =
+          env_->OptimizeForCompactionTableWrite(env_options_, immutable_db_options_);
       s = BuildTable(
-          dbname_, env_, *cfd->ioptions(), mutable_cf_options, env_options_,
-          cfd->table_cache(), iter.get(),
+          dbname_, env_, *cfd->ioptions(), mutable_cf_options,
+          optimized_env_options, cfd->table_cache(), iter.get(),
           std::unique_ptr<InternalIterator>(mem->NewRangeTombstoneIterator(ro)),
           &meta, cfd->internal_comparator(),
           cfd->int_tbl_prop_collector_factories(), cfd->GetID(), cfd->GetName(),
@@ -1025,7 +1032,8 @@ Status DB::Open(const DBOptions& db_options, const std::string& dbname,
   if (s.ok()) {
     // Persist RocksDB Options before scheduling the compaction.
     // The WriteOptionsFile() will release and lock the mutex internally.
-    persist_options_status = impl->WriteOptionsFile();
+    persist_options_status = impl->WriteOptionsFile(
+        false /*need_mutex_lock*/, false /*need_enter_write_thread*/);
 
     *dbptr = impl;
     impl->opened_successfully_ = true;
@@ -1058,14 +1066,9 @@ Status DB::Open(const DBOptions& db_options, const std::string& dbname,
     ROCKS_LOG_INFO(impl->immutable_db_options_.info_log, "DB pointer %p", impl);
     LogFlush(impl->immutable_db_options_.info_log);
     if (!persist_options_status.ok()) {
-      if (db_options.fail_if_options_file_error) {
-        s = Status::IOError(
-            "DB::Open() failed --- Unable to persist Options file",
-            persist_options_status.ToString());
-      }
-      ROCKS_LOG_WARN(impl->immutable_db_options_.info_log,
-                     "Unable to persist options in DB::Open() -- %s",
-                     persist_options_status.ToString().c_str());
+      s = Status::IOError(
+          "DB::Open() failed --- Unable to persist Options file",
+          persist_options_status.ToString());
     }
   }
   if (!s.ok()) {

db/db_iterator_test.cc

@@ -22,6 +22,34 @@ class DBIteratorTest : public DBTestBase {
   DBIteratorTest() : DBTestBase("/db_iterator_test") {}
 };
 
+class FlushBlockEveryKeyPolicy : public FlushBlockPolicy {
+ public:
+  virtual bool Update(const Slice& key, const Slice& value) override {
+    if (!start_) {
+      start_ = true;
+      return false;
+    }
+    return true;
+  }
+ private:
+  bool start_ = false;
+};
+
+class FlushBlockEveryKeyPolicyFactory : public FlushBlockPolicyFactory {
+ public:
+  explicit FlushBlockEveryKeyPolicyFactory() {}
+
+  const char* Name() const override {
+    return "FlushBlockEveryKeyPolicyFactory";
+  }
+
+  FlushBlockPolicy* NewFlushBlockPolicy(
+    const BlockBasedTableOptions& table_options,
+    const BlockBuilder& data_block_builder) const override {
+    return new FlushBlockEveryKeyPolicy;
+  }
+};
+
 TEST_F(DBIteratorTest, IteratorProperty) {
   // The test needs to be changed if kPersistedTier is supported in iterator.
   Options options = CurrentOptions();
@@ -975,6 +1003,50 @@ TEST_F(DBIteratorTest, DBIteratorBoundTest) {
   }
 }
 
+TEST_F(DBIteratorTest, DBIteratorBoundOptimizationTest) {
+  int upper_bound_hits = 0;
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "BlockBasedTable::BlockEntryIteratorState::KeyReachedUpperBound",
+      [&upper_bound_hits](void* arg) {
+        assert(arg != nullptr);
+        upper_bound_hits += (*static_cast<bool*>(arg) ? 1 : 0);
+      });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+  Options options = CurrentOptions();
+  options.env = env_;
+  options.create_if_missing = true;
+  options.prefix_extractor = nullptr;
+  BlockBasedTableOptions table_options;
+  table_options.flush_block_policy_factory =
+    std::make_shared<FlushBlockEveryKeyPolicyFactory>();
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("foo1", "bar1"));
+  ASSERT_OK(Put("foo2", "bar2"));
+  ASSERT_OK(Put("foo4", "bar4"));
+  ASSERT_OK(Flush());
+
+  Slice ub("foo3");
+  ReadOptions ro;
+  ro.iterate_upper_bound = &ub;
+
+  std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+
+  iter->Seek("foo");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ(iter->key().compare(Slice("foo1")), 0);
+  ASSERT_EQ(upper_bound_hits, 0);
+
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ(iter->key().compare(Slice("foo2")), 0);
+  ASSERT_EQ(upper_bound_hits, 0);
+
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_EQ(upper_bound_hits, 1);
+}
 // TODO(3.13): fix the issue of Seek() + Prev() which might not necessary
 //             return the biggest key which is smaller than the seek key.
 TEST_F(DBIteratorTest, PrevAfterAndNextAfterMerge) {

db/db_properties_test.cc

@@ -425,6 +425,13 @@ TEST_F(DBPropertiesTest, ReadLatencyHistogramByLevel) {
   ASSERT_NE(std::string::npos, prop.find("** Level 0 read latency histogram"));
   ASSERT_NE(std::string::npos, prop.find("** Level 1 read latency histogram"));
   ASSERT_EQ(std::string::npos, prop.find("** Level 2 read latency histogram"));
+
+  // Clear internal stats
+  dbfull()->ResetStats();
+  ASSERT_TRUE(dbfull()->GetProperty("rocksdb.cfstats", &prop));
+  ASSERT_EQ(std::string::npos, prop.find("** Level 0 read latency histogram"));
+  ASSERT_EQ(std::string::npos, prop.find("** Level 1 read latency histogram"));
+  ASSERT_EQ(std::string::npos, prop.find("** Level 2 read latency histogram"));
 }
 
 TEST_F(DBPropertiesTest, AggregatedTablePropertiesAtLevel) {
@@ -1293,6 +1300,18 @@ TEST_F(DBPropertiesTest, NeedCompactHintPersistentTest) {
     SetPerfLevel(kDisable);
   }
 }
+
+TEST_F(DBPropertiesTest, EstimateNumKeysUnderflow) {
+  Options options;
+  Reopen(options);
+  Put("foo", "bar");
+  Delete("foo");
+  Delete("foo");
+  uint64_t num_keys = 0;
+  ASSERT_TRUE(dbfull()->GetIntProperty("rocksdb.estimate-num-keys", &num_keys));
+  ASSERT_EQ(0, num_keys);
+}
+
 #endif  // ROCKSDB_LITE
 }  // namespace rocksdb
 

db/db_test2.cc

@@ -1640,66 +1640,80 @@ size_t GetEncodedEntrySize(size_t key_size, size_t value_size) {
 TEST_F(DBTest2, ReadAmpBitmap) {
   Options options = CurrentOptions();
   BlockBasedTableOptions bbto;
-  // Disable delta encoding to make it easier to calculate read amplification
-  bbto.use_delta_encoding = false;
-  // Huge block cache to make it easier to calculate read amplification
-  bbto.block_cache = NewLRUCache(1024 * 1024 * 1024);
-  bbto.read_amp_bytes_per_bit = 16;
-  options.table_factory.reset(NewBlockBasedTableFactory(bbto));
-  options.statistics = rocksdb::CreateDBStatistics();
-  DestroyAndReopen(options);
+  uint32_t bytes_per_bit[2] = {1, 16};
+  for (size_t k = 0; k < 2; k++) {
+    // Disable delta encoding to make it easier to calculate read amplification
+    bbto.use_delta_encoding = false;
+    // Huge block cache to make it easier to calculate read amplification
+    bbto.block_cache = NewLRUCache(1024 * 1024 * 1024);
+    bbto.read_amp_bytes_per_bit = bytes_per_bit[k];
+    options.table_factory.reset(NewBlockBasedTableFactory(bbto));
+    options.statistics = rocksdb::CreateDBStatistics();
+    DestroyAndReopen(options);
 
-  const size_t kNumEntries = 10000;
+    const size_t kNumEntries = 10000;
 
-  Random rnd(301);
-  for (size_t i = 0; i < kNumEntries; i++) {
-    ASSERT_OK(Put(Key(static_cast<int>(i)), RandomString(&rnd, 100)));
-  }
-  ASSERT_OK(Flush());
+    Random rnd(301);
+    for (size_t i = 0; i < kNumEntries; i++) {
+      ASSERT_OK(Put(Key(static_cast<int>(i)), RandomString(&rnd, 100)));
+    }
+    ASSERT_OK(Flush());
 
-  Close();
-  Reopen(options);
+    Close();
+    Reopen(options);
 
-  // Read keys/values randomly and verify that reported read amp error
-  // is less than 2%
-  uint64_t total_useful_bytes = 0;
-  std::set<int> read_keys;
-  std::string value;
-  for (size_t i = 0; i < kNumEntries * 5; i++) {
-    int key_idx = rnd.Next() % kNumEntries;
-    std::string k = Key(key_idx);
-    ASSERT_OK(db_->Get(ReadOptions(), k, &value));
+    // Read keys/values randomly and verify that reported read amp error
+    // is less than 2%
+    uint64_t total_useful_bytes = 0;
+    std::set<int> read_keys;
+    std::string value;
+    for (size_t i = 0; i < kNumEntries * 5; i++) {
+      int key_idx = rnd.Next() % kNumEntries;
+      std::string key = Key(key_idx);
+      ASSERT_OK(db_->Get(ReadOptions(), key, &value));
 
-    if (read_keys.find(key_idx) == read_keys.end()) {
-      auto ik = InternalKey(k, 0, ValueType::kTypeValue);
-      total_useful_bytes += GetEncodedEntrySize(ik.size(), value.size());
-      read_keys.insert(key_idx);
+      if (read_keys.find(key_idx) == read_keys.end()) {
+        auto internal_key = InternalKey(key, 0, ValueType::kTypeValue);
+        total_useful_bytes +=
+            GetEncodedEntrySize(internal_key.size(), value.size());
+        read_keys.insert(key_idx);
+      }
+
+      double expected_read_amp =
+          static_cast<double>(total_useful_bytes) /
+          options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES);
+
+      double read_amp =
+          static_cast<double>(options.statistics->getTickerCount(
+              READ_AMP_ESTIMATE_USEFUL_BYTES)) /
+          options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES);
+
+      double error_pct = fabs(expected_read_amp - read_amp) * 100;
+      // Error between reported read amp and real read amp should be less than
+      // 2%
+      EXPECT_LE(error_pct, 2);
     }
 
-    double expected_read_amp =
-        static_cast<double>(total_useful_bytes) /
-        options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES);
+    // Make sure we read every thing in the DB (which is smaller than our cache)
+    Iterator* iter = db_->NewIterator(ReadOptions());
+    for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+      ASSERT_EQ(iter->value().ToString(), Get(iter->key().ToString()));
+    }
+    delete iter;
 
-    double read_amp =
-        static_cast<double>(options.statistics->getTickerCount(
-            READ_AMP_ESTIMATE_USEFUL_BYTES)) /
-        options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES);
-
-    double error_pct = fabs(expected_read_amp - read_amp) * 100;
-    // Error between reported read amp and real read amp should be less than 2%
-    EXPECT_LE(error_pct, 2);
+    // Read amp is on average 100% since we read all what we loaded in memory
+    if (k == 0) {
+      ASSERT_EQ(
+          options.statistics->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES),
+          options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES));
+    } else {
+      ASSERT_NEAR(
+          options.statistics->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES) *
+              1.0f /
+              options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES),
+          1, .01);
+    }
   }
-
-  // Make sure we read every thing in the DB (which is smaller than our cache)
-  Iterator* iter = db_->NewIterator(ReadOptions());
-  for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
-    ASSERT_EQ(iter->value().ToString(), Get(iter->key().ToString()));
-  }
-  delete iter;
-
-  // Read amp is 100% since we read all what we loaded in memory
-  ASSERT_EQ(options.statistics->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES),
-            options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES));
 }
 
 TEST_F(DBTest2, ReadAmpBitmapLiveInCacheAfterDBClose) {
@@ -1709,82 +1723,93 @@ TEST_F(DBTest2, ReadAmpBitmapLiveInCacheAfterDBClose) {
     // the blocks again regardless of them being already in the cache
     return;
   }
+  uint32_t bytes_per_bit[2] = {1, 16};
+  for (size_t k = 0; k < 2; k++) {
+    std::shared_ptr<Cache> lru_cache = NewLRUCache(1024 * 1024 * 1024);
+    std::shared_ptr<Statistics> stats = rocksdb::CreateDBStatistics();
 
-  std::shared_ptr<Cache> lru_cache = NewLRUCache(1024 * 1024 * 1024);
-  std::shared_ptr<Statistics> stats = rocksdb::CreateDBStatistics();
+    Options options = CurrentOptions();
+    BlockBasedTableOptions bbto;
+    // Disable delta encoding to make it easier to calculate read amplification
+    bbto.use_delta_encoding = false;
+    // Huge block cache to make it easier to calculate read amplification
+    bbto.block_cache = lru_cache;
+    bbto.read_amp_bytes_per_bit = bytes_per_bit[k];
+    options.table_factory.reset(NewBlockBasedTableFactory(bbto));
+    options.statistics = stats;
+    DestroyAndReopen(options);
 
-  Options options = CurrentOptions();
-  BlockBasedTableOptions bbto;
-  // Disable delta encoding to make it easier to calculate read amplification
-  bbto.use_delta_encoding = false;
-  // Huge block cache to make it easier to calculate read amplification
-  bbto.block_cache = lru_cache;
-  bbto.read_amp_bytes_per_bit = 16;
-  options.table_factory.reset(NewBlockBasedTableFactory(bbto));
-  options.statistics = stats;
-  DestroyAndReopen(options);
+    const int kNumEntries = 10000;
 
-  const int kNumEntries = 10000;
+    Random rnd(301);
+    for (int i = 0; i < kNumEntries; i++) {
+      ASSERT_OK(Put(Key(i), RandomString(&rnd, 100)));
+    }
+    ASSERT_OK(Flush());
 
-  Random rnd(301);
-  for (int i = 0; i < kNumEntries; i++) {
-    ASSERT_OK(Put(Key(i), RandomString(&rnd, 100)));
-  }
-  ASSERT_OK(Flush());
+    Close();
+    Reopen(options);
 
-  Close();
-  Reopen(options);
+    uint64_t total_useful_bytes = 0;
+    std::set<int> read_keys;
+    std::string value;
+    // Iter1: Read half the DB, Read even keys
+    // Key(0), Key(2), Key(4), Key(6), Key(8), ...
+    for (int i = 0; i < kNumEntries; i += 2) {
+      std::string key = Key(i);
+      ASSERT_OK(db_->Get(ReadOptions(), key, &value));
 
-  uint64_t total_useful_bytes = 0;
-  std::set<int> read_keys;
-  std::string value;
-  // Iter1: Read half the DB, Read even keys
-  // Key(0), Key(2), Key(4), Key(6), Key(8), ...
-  for (int i = 0; i < kNumEntries; i += 2) {
-    std::string k = Key(i);
-    ASSERT_OK(db_->Get(ReadOptions(), k, &value));
+      if (read_keys.find(i) == read_keys.end()) {
+        auto internal_key = InternalKey(key, 0, ValueType::kTypeValue);
+        total_useful_bytes +=
+            GetEncodedEntrySize(internal_key.size(), value.size());
+        read_keys.insert(i);
+      }
+    }
 
-    if (read_keys.find(i) == read_keys.end()) {
-      auto ik = InternalKey(k, 0, ValueType::kTypeValue);
-      total_useful_bytes += GetEncodedEntrySize(ik.size(), value.size());
-      read_keys.insert(i);
+    size_t total_useful_bytes_iter1 =
+        options.statistics->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES);
+    size_t total_loaded_bytes_iter1 =
+        options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES);
+
+    Close();
+    std::shared_ptr<Statistics> new_statistics = rocksdb::CreateDBStatistics();
+    // Destroy old statistics obj that the blocks in lru_cache are pointing to
+    options.statistics.reset();
+    // Use the statistics object that we just created
+    options.statistics = new_statistics;
+    Reopen(options);
+
+    // Iter2: Read half the DB, Read odd keys
+    // Key(1), Key(3), Key(5), Key(7), Key(9), ...
+    for (int i = 1; i < kNumEntries; i += 2) {
+      std::string key = Key(i);
+      ASSERT_OK(db_->Get(ReadOptions(), key, &value));
+
+      if (read_keys.find(i) == read_keys.end()) {
+        auto internal_key = InternalKey(key, 0, ValueType::kTypeValue);
+        total_useful_bytes +=
+            GetEncodedEntrySize(internal_key.size(), value.size());
+        read_keys.insert(i);
+      }
+    }
+
+    size_t total_useful_bytes_iter2 =
+        options.statistics->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES);
+    size_t total_loaded_bytes_iter2 =
+        options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES);
+
+
+    // Read amp is on average 100% since we read all what we loaded in memory
+    if (k == 0) {
+      ASSERT_EQ(total_useful_bytes_iter1 + total_useful_bytes_iter2,
+                total_loaded_bytes_iter1 + total_loaded_bytes_iter2);
+    } else {
+      ASSERT_NEAR((total_useful_bytes_iter1 + total_useful_bytes_iter2) * 1.0f /
+                      (total_loaded_bytes_iter1 + total_loaded_bytes_iter2),
+                  1, .01);
     }
   }
-
-  size_t total_useful_bytes_iter1 =
-      options.statistics->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES);
-  size_t total_loaded_bytes_iter1 =
-      options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES);
-
-  Close();
-  std::shared_ptr<Statistics> new_statistics = rocksdb::CreateDBStatistics();
-  // Destroy old statistics obj that the blocks in lru_cache are pointing to
-  options.statistics.reset();
-  // Use the statistics object that we just created
-  options.statistics = new_statistics;
-  Reopen(options);
-
-  // Iter2: Read half the DB, Read odd keys
-  // Key(1), Key(3), Key(5), Key(7), Key(9), ...
-  for (int i = 1; i < kNumEntries; i += 2) {
-    std::string k = Key(i);
-    ASSERT_OK(db_->Get(ReadOptions(), k, &value));
-
-    if (read_keys.find(i) == read_keys.end()) {
-      auto ik = InternalKey(k, 0, ValueType::kTypeValue);
-      total_useful_bytes += GetEncodedEntrySize(ik.size(), value.size());
-      read_keys.insert(i);
-    }
-  }
-
-  size_t total_useful_bytes_iter2 =
-      options.statistics->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES);
-  size_t total_loaded_bytes_iter2 =
-      options.statistics->getTickerCount(READ_AMP_TOTAL_READ_BYTES);
-
-  // We reached read_amp of 100% because we read all the keys in the DB
-  ASSERT_EQ(total_useful_bytes_iter1 + total_useful_bytes_iter2,
-            total_loaded_bytes_iter1 + total_loaded_bytes_iter2);
 }
 
 #ifndef ROCKSDB_LITE
@@ -1995,7 +2020,8 @@ TEST_F(DBTest2, DirectIO) {
     return;
   }
   Options options = CurrentOptions();
-  options.use_direct_reads = options.use_direct_writes = true;
+  options.use_direct_reads = options.use_direct_io_for_flush_and_compaction =
+      true;
   options.allow_mmap_reads = options.allow_mmap_writes = false;
   DestroyAndReopen(options);
 

db/db_test_util.h

@@ -277,6 +277,9 @@ class SpecialEnv : public EnvWrapper {
       bool use_direct_io() const override {
         return base_->use_direct_io();
       }
+      Status Allocate(uint64_t offset, uint64_t len) override {
+        return base_->Allocate(offset, len);
+      }
     };
     class ManifestFile : public WritableFile {
      public:

db/dbformat.cc

@@ -129,7 +129,7 @@ void InternalKeyComparator::FindShortestSeparator(
   Slice user_limit = ExtractUserKey(limit);
   std::string tmp(user_start.data(), user_start.size());
   user_comparator_->FindShortestSeparator(&tmp, user_limit);
-  if (tmp.size() < user_start.size() &&
+  if (tmp.size() <= user_start.size() &&
       user_comparator_->Compare(user_start, tmp) < 0) {
     // User key has become shorter physically, but larger logically.
     // Tack on the earliest possible number to the shortened user key.
@@ -144,7 +144,7 @@ void InternalKeyComparator::FindShortSuccessor(std::string* key) const {
   Slice user_key = ExtractUserKey(*key);
   std::string tmp(user_key.data(), user_key.size());
   user_comparator_->FindShortSuccessor(&tmp);
-  if (tmp.size() < user_key.size() &&
+  if (tmp.size() <= user_key.size() &&
       user_comparator_->Compare(user_key, tmp) < 0) {
     // User key has become shorter physically, but larger logically.
     // Tack on the earliest possible number to the shortened user key.

db/flush_job.cc

@@ -294,9 +294,11 @@ Status FlushJob::WriteLevel0Table() {
 
       TEST_SYNC_POINT_CALLBACK("FlushJob::WriteLevel0Table:output_compression",
                                &output_compression_);
+      EnvOptions optimized_env_options =
+          db_options_.env->OptimizeForCompactionTableWrite(env_options_, db_options_);
       s = BuildTable(
           dbname_, db_options_.env, *cfd_->ioptions(), mutable_cf_options_,
-          env_options_, cfd_->table_cache(), iter.get(),
+          optimized_env_options, cfd_->table_cache(), iter.get(),
           std::move(range_del_iter), &meta_, cfd_->internal_comparator(),
           cfd_->int_tbl_prop_collector_factories(), cfd_->GetID(),
           cfd_->GetName(), existing_snapshots_,

db/forward_iterator.cc

@@ -73,7 +73,8 @@ class LevelIterator : public InternalIterator {
         cfd_->internal_comparator(), {} /* snapshots */);
     file_iter_ = cfd_->table_cache()->NewIterator(
         read_options_, *(cfd_->soptions()), cfd_->internal_comparator(),
-        files_[file_index_]->fd, read_options_.ignore_range_deletions ? nullptr : &range_del_agg,
+        files_[file_index_]->fd,
+        read_options_.ignore_range_deletions ? nullptr : &range_del_agg,
         nullptr /* table_reader_ptr */, nullptr, false);
     file_iter_->SetPinnedItersMgr(pinned_iters_mgr_);
     if (!range_del_agg.IsEmpty()) {

db/forward_iterator_bench.cc

@@ -331,7 +331,7 @@ int main(int argc, char** argv) {
   options.compaction_style = rocksdb::CompactionStyle::kCompactionStyleNone;
   options.level0_slowdown_writes_trigger = 99999;
   options.level0_stop_writes_trigger = 99999;
-  options.use_direct_writes = true;
+  options.use_direct_io_for_flush_and_compaction = true;
   options.write_buffer_size = FLAGS_memtable_size;
   rocksdb::BlockBasedTableOptions table_options;
   table_options.block_cache = rocksdb::NewLRUCache(FLAGS_block_cache_size);

db/internal_stats.cc

@@ -677,12 +677,14 @@ bool InternalStats::HandleEstimateNumKeys(uint64_t* value, DBImpl* db,
   // Estimate number of entries in the column family:
   // Use estimated entries in tables + total entries in memtables.
   const auto* vstorage = cfd_->current()->storage_info();
-  *value = cfd_->mem()->num_entries() +
-           cfd_->imm()->current()->GetTotalNumEntries() -
-           (cfd_->mem()->num_deletes() +
-            cfd_->imm()->current()->GetTotalNumDeletes()) *
-               2 +
-           vstorage->GetEstimatedActiveKeys();
+  uint64_t estimate_keys = cfd_->mem()->num_entries() +
+                           cfd_->imm()->current()->GetTotalNumEntries() +
+                           vstorage->GetEstimatedActiveKeys();
+  uint64_t estimate_deletes =
+      cfd_->mem()->num_deletes() + cfd_->imm()->current()->GetTotalNumDeletes();
+  *value = estimate_keys > estimate_deletes * 2
+               ? estimate_keys - (estimate_deletes * 2)
+               : 0;
   return true;
 }
 

db/internal_stats.h

@@ -189,6 +189,20 @@ class InternalStats {
           num_dropped_records(c.num_dropped_records),
           count(c.count) {}
 
+    void Clear() {
+      this->micros = 0;
+      this->bytes_read_non_output_levels = 0;
+      this->bytes_read_output_level = 0;
+      this->bytes_written = 0;
+      this->bytes_moved = 0;
+      this->num_input_files_in_non_output_levels = 0;
+      this->num_input_files_in_output_level = 0;
+      this->num_output_files = 0;
+      this->num_input_records = 0;
+      this->num_dropped_records = 0;
+      this->count = 0;
+    }
+
     void Add(const CompactionStats& c) {
       this->micros += c.micros;
       this->bytes_read_non_output_levels += c.bytes_read_non_output_levels;
@@ -222,6 +236,26 @@ class InternalStats {
     }
   };
 
+  void Clear() {
+    for (int i = 0; i < INTERNAL_DB_STATS_ENUM_MAX; i++) {
+      db_stats_[i].store(0);
+    }
+    for (int i = 0; i < INTERNAL_CF_STATS_ENUM_MAX; i++) {
+      cf_stats_count_[i] = 0;
+      cf_stats_value_[i] = 0;
+    }
+    for (auto& comp_stat : comp_stats_) {
+      comp_stat.Clear();
+    }
+    for (auto& h : file_read_latency_) {
+      h.Clear();
+    }
+    cf_stats_snapshot_.Clear();
+    db_stats_snapshot_.Clear();
+    bg_error_count_ = 0;
+    started_at_ = env_->NowMicros();
+  }
+
   void AddCompactionStats(int level, const CompactionStats& stats) {
     comp_stats_[level].Add(stats);
   }
@@ -319,6 +353,20 @@ class InternalStats {
           ingest_files_addfile(0),
           ingest_l0_files_addfile(0),
           ingest_keys_addfile(0) {}
+
+    void Clear() {
+      comp_stats.Clear();
+      ingest_bytes_flush = 0;
+      stall_count = 0;
+      compact_bytes_write = 0;
+      compact_bytes_read = 0;
+      compact_micros = 0;
+      seconds_up = 0;
+      ingest_bytes_addfile = 0;
+      ingest_files_addfile = 0;
+      ingest_l0_files_addfile = 0;
+      ingest_keys_addfile = 0;
+    }
   } cf_stats_snapshot_;
 
   struct DBStatsSnapshot {
@@ -350,6 +398,18 @@ class InternalStats {
           num_keys_written(0),
           write_stall_micros(0),
           seconds_up(0) {}
+
+    void Clear() {
+      ingest_bytes = 0;
+      wal_bytes = 0;
+      wal_synced = 0;
+      write_with_wal = 0;
+      write_other = 0;
+      write_self = 0;
+      num_keys_written = 0;
+      write_stall_micros = 0;
+      seconds_up = 0;
+    }
   } db_stats_snapshot_;
 
   // Handler functions for getting property values. They use "value" as a value-
@@ -420,7 +480,7 @@ class InternalStats {
   const int number_levels_;
   Env* env_;
   ColumnFamilyData* cfd_;
-  const uint64_t started_at_;
+  uint64_t started_at_;
 };
 
 #else

db/log_format.h

@@ -34,8 +34,8 @@ static const int kMaxRecordType = kRecyclableLastType;
 
 static const unsigned int kBlockSize = 32768;
 
-// Header is checksum (4 bytes), type (1 byte), length (2 bytes).
-static const int kHeaderSize = 4 + 1 + 2;
+// Header is checksum (4 bytes), length (2 bytes), type (1 byte)
+static const int kHeaderSize = 4 + 2 + 1;
 
 // Recyclable header is checksum (4 bytes), type (1 byte), log number
 // (4 bytes), length (2 bytes).

db/memtable.cc

@@ -724,9 +724,9 @@ void MemTable::Update(SequenceNumber seq,
         uint32_t new_size = static_cast<uint32_t>(value.size());
 
         // Update value, if new value size  <= previous value size
-        if (new_size <= prev_size ) {
-          char* p = EncodeVarint32(const_cast<char*>(key_ptr) + key_length,
-                                   new_size);
+        if (new_size <= prev_size) {
+          char* p =
+              EncodeVarint32(const_cast<char*>(key_ptr) + key_length, new_size);
           WriteLock wl(GetLock(lkey.user_key()));
           memcpy(p, value.data(), value.size());
           assert((unsigned)((p + value.size()) - entry) ==

db/repair.cc

@@ -377,9 +377,11 @@ class Repairer {
       ro.total_order_seek = true;
       Arena arena;
       ScopedArenaIterator iter(mem->NewIterator(ro, &arena));
+      EnvOptions optimized_env_options =
+          env_->OptimizeForCompactionTableWrite(env_options_, immutable_db_options_);
       status = BuildTable(
           dbname_, env_, *cfd->ioptions(), *cfd->GetLatestMutableCFOptions(),
-          env_options_, table_cache_, iter.get(),
+          optimized_env_options, table_cache_, iter.get(),
           std::unique_ptr<InternalIterator>(mem->NewRangeTombstoneIterator(ro)),
           &meta, cfd->internal_comparator(),
           cfd->int_tbl_prop_collector_factories(), cfd->GetID(), cfd->GetName(),

db/table_cache.cc

@@ -184,6 +184,11 @@ InternalIterator* TableCache::NewIterator(
     }
     size_t readahead = 0;
     if (for_compaction) {
+#ifndef NDEBUG
+      bool use_direct_reads_for_compaction = env_options.use_direct_reads;
+      TEST_SYNC_POINT_CALLBACK("TableCache::NewIterator:for_compaction",
+                               &use_direct_reads_for_compaction);
+#endif  // !NDEBUG
       if (ioptions_.new_table_reader_for_compaction_inputs) {
         readahead = ioptions_.compaction_readahead_size;
         create_new_table_reader = true;
@@ -217,7 +222,8 @@ InternalIterator* TableCache::NewIterator(
   }
   InternalIterator* result = nullptr;
   if (s.ok()) {
-    result = table_reader->NewIterator(options, arena, skip_filters);
+    result =
+      table_reader->NewIterator(options, arena, &icomparator, skip_filters);
     if (create_new_table_reader) {
       assert(handle == nullptr);
       result->RegisterCleanup(&DeleteTableReader, table_reader, nullptr);

db/version_set.cc

@@ -526,6 +526,13 @@ class LevelFileIteratorState : public TwoLevelIteratorState {
     return true;
   }
 
+  bool KeyReachedUpperBound(const Slice& internal_key) override {
+    return read_options_.iterate_upper_bound != nullptr &&
+           icomparator_.user_comparator()->Compare(
+               ExtractUserKey(internal_key),
+               *read_options_.iterate_upper_bound) >= 0;
+  }
+
  private:
   TableCache* table_cache_;
   const ReadOptions read_options_;
@@ -1314,9 +1321,9 @@ void VersionStorageInfo::ComputeCompactionScore(
           // Level-based involves L0->L0 compactions that can lead to oversized
           // L0 files. Take into account size as well to avoid later giant
           // compactions to the base level.
-          uint64_t base_level_max_bytes = MaxBytesForLevel(base_level());
           score = std::max(
-              score, static_cast<double>(total_size) / base_level_max_bytes);
+              score, static_cast<double>(total_size) /
+                     mutable_cf_options.max_bytes_for_level_base);
         }
       }
     } else {
@@ -2218,7 +2225,8 @@ VersionSet::VersionSet(const std::string& dbname,
       current_version_number_(0),
       manifest_file_size_(0),
       env_options_(storage_options),
-      env_options_compactions_(env_options_) {}
+      env_options_compactions_(
+          env_->OptimizeForCompactionTableRead(env_options_, *db_options_)) {}
 
 void CloseTables(void* ptr, size_t) {
   TableReader* table_reader = reinterpret_cast<TableReader*>(ptr);
@@ -3477,7 +3485,7 @@ InternalIterator* VersionSet::MakeInputIterator(
         // Create concatenating iterator for the files from this level
         list[num++] = NewTwoLevelIterator(
             new LevelFileIteratorState(
-                cfd->table_cache(), read_options, env_options_,
+                cfd->table_cache(), read_options, env_options_compactions_,
                 cfd->internal_comparator(),
                 nullptr /* no per level latency histogram */,
                 true /* for_compaction */, false /* prefix enabled */,

env/env.cc

@@ -10,9 +10,9 @@
 #include "rocksdb/env.h"
 
 #include <thread>
+#include "options/db_options.h"
 #include "port/port.h"
 #include "port/sys_time.h"
-
 #include "rocksdb/options.h"
 #include "util/arena.h"
 #include "util/autovector.h"
@@ -316,7 +316,6 @@ void AssignEnvOptions(EnvOptions* env_options, const DBOptions& options) {
   env_options->use_mmap_reads = options.allow_mmap_reads;
   env_options->use_mmap_writes = options.allow_mmap_writes;
   env_options->use_direct_reads = options.use_direct_reads;
-  env_options->use_direct_writes = options.use_direct_writes;
   env_options->set_fd_cloexec = options.is_fd_close_on_exec;
   env_options->bytes_per_sync = options.bytes_per_sync;
   env_options->compaction_readahead_size = options.compaction_readahead_size;
@@ -341,6 +340,22 @@ EnvOptions Env::OptimizeForManifestWrite(const EnvOptions& env_options) const {
   return env_options;
 }
 
+EnvOptions Env::OptimizeForCompactionTableWrite(
+    const EnvOptions& env_options, const ImmutableDBOptions& db_options) const {
+  EnvOptions optimized_env_options(env_options);
+  optimized_env_options.use_direct_writes =
+      db_options.use_direct_io_for_flush_and_compaction;
+  return optimized_env_options;
+}
+
+EnvOptions Env::OptimizeForCompactionTableRead(
+    const EnvOptions& env_options, const ImmutableDBOptions& db_options) const {
+  EnvOptions optimized_env_options(env_options);
+  optimized_env_options.use_direct_reads =
+      db_options.use_direct_io_for_flush_and_compaction;
+  return optimized_env_options;
+}
+
 EnvOptions::EnvOptions(const DBOptions& options) {
   AssignEnvOptions(this, options);
 }

env/env_posix.cc

@@ -251,13 +251,14 @@ class PosixEnv : public Env {
     return s;
   }
 
-  virtual Status NewWritableFile(const std::string& fname,
-                                 unique_ptr<WritableFile>* result,
-                                 const EnvOptions& options) override {
+  virtual Status OpenWritableFile(const std::string& fname,
+                                  unique_ptr<WritableFile>* result,
+                                  const EnvOptions& options,
+                                  bool reopen = false) {
     result->reset();
     Status s;
     int fd = -1;
-    int flags = O_CREAT | O_TRUNC;
+    int flags = (reopen) ? (O_CREAT | O_APPEND) : (O_CREAT | O_TRUNC);
     // Direct IO mode with O_DIRECT flag or F_NOCAHCE (MAC OSX)
     if (options.use_direct_writes && !options.use_mmap_writes) {
       // Note: we should avoid O_APPEND here due to ta the following bug:
@@ -323,6 +324,18 @@ class PosixEnv : public Env {
     return s;
   }
 
+  virtual Status NewWritableFile(const std::string& fname,
+                                 unique_ptr<WritableFile>* result,
+                                 const EnvOptions& options) override {
+    return OpenWritableFile(fname, result, options, false);
+  }
+
+  virtual Status ReopenWritableFile(const std::string& fname,
+                                    unique_ptr<WritableFile>* result,
+                                    const EnvOptions& options) override {
+    return OpenWritableFile(fname, result, options, true);
+  }
+
   virtual Status ReuseWritableFile(const std::string& fname,
                                    const std::string& old_fname,
                                    unique_ptr<WritableFile>* result,

env/io_posix.cc

@@ -37,7 +37,7 @@
 namespace rocksdb {
 
 // A wrapper for fadvise, if the platform doesn't support fadvise,
-// it will simply return Status::NotSupport.
+// it will simply return 0.
 int Fadvise(int fd, off_t offset, size_t len, int advice) {
 #ifdef OS_LINUX
   return posix_fadvise(fd, offset, len, advice);
@@ -115,18 +115,13 @@ size_t GetLogicalBufferSize(int __attribute__((__unused__)) fd) {
  */
 #ifndef NDEBUG
 namespace {
-#ifdef OS_LINUX
-const size_t kPageSize = sysconf(_SC_PAGESIZE);
-#else
-const size_t kPageSize = 4 * 1024;
-#endif
 
 bool IsSectorAligned(const size_t off, size_t sector_size) {
   return off % sector_size == 0;
 }
 
-static bool IsPageAligned(const void* ptr) {
-  return uintptr_t(ptr) % (kPageSize) == 0;
+bool IsSectorAligned(const void* ptr, size_t sector_size) {
+  return uintptr_t(ptr) % sector_size == 0;
 }
 
 }
@@ -182,6 +177,11 @@ Status PosixSequentialFile::Read(size_t n, Slice* result, char* scratch) {
 
 Status PosixSequentialFile::PositionedRead(uint64_t offset, size_t n,
                                            Slice* result, char* scratch) {
+  if (use_direct_io()) {
+    assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
+    assert(IsSectorAligned(n, GetRequiredBufferAlignment()));
+    assert(IsSectorAligned(scratch, GetRequiredBufferAlignment()));
+  }
   Status s;
   ssize_t r = -1;
   size_t left = n;
@@ -306,6 +306,11 @@ PosixRandomAccessFile::~PosixRandomAccessFile() { close(fd_); }
 
 Status PosixRandomAccessFile::Read(uint64_t offset, size_t n, Slice* result,
                                    char* scratch) const {
+  if (use_direct_io()) {
+    assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
+    assert(IsSectorAligned(n, GetRequiredBufferAlignment()));
+    assert(IsSectorAligned(scratch, GetRequiredBufferAlignment()));
+  }
   Status s;
   ssize_t r = -1;
   size_t left = n;
@@ -336,7 +341,27 @@ Status PosixRandomAccessFile::Read(uint64_t offset, size_t n, Slice* result,
   return s;
 }
 
-#if defined(OS_LINUX) || defined(OS_MACOSX)
+Status PosixRandomAccessFile::Prefetch(uint64_t offset, size_t n) {
+  Status s;
+  if (!use_direct_io()) {
+    ssize_t r = 0;
+#ifdef OS_LINUX
+    r = readahead(fd_, offset, n);
+#endif
+#ifdef OS_MACOSX
+    radvisory advice;
+    advice.ra_offset = static_cast<off_t>(offset);
+    advice.ra_count = static_cast<int>(n);
+    r = fcntl(fd_, F_RDADVISE, &advice);
+#endif
+    if (r == -1) {
+      s = IOError(filename_, errno);
+    }
+  }
+  return s;
+}
+
+#if defined(OS_LINUX) || defined(OS_MACOSX) || defined(OS_AIX)
 size_t PosixRandomAccessFile::GetUniqueId(char* id, size_t max_size) const {
   return PosixHelper::GetUniqueIdFromFile(fd_, id, max_size);
 }
@@ -684,9 +709,10 @@ PosixWritableFile::~PosixWritableFile() {
 }
 
 Status PosixWritableFile::Append(const Slice& data) {
-  assert(!use_direct_io() ||
-         (IsSectorAligned(data.size(), GetRequiredBufferAlignment()) &&
-          IsPageAligned(data.data())));
+  if (use_direct_io()) {
+    assert(IsSectorAligned(data.size(), GetRequiredBufferAlignment()));
+    assert(IsSectorAligned(data.data(), GetRequiredBufferAlignment()));
+  }
   const char* src = data.data();
   size_t left = data.size();
   while (left != 0) {
@@ -705,10 +731,11 @@ Status PosixWritableFile::Append(const Slice& data) {
 }
 
 Status PosixWritableFile::PositionedAppend(const Slice& data, uint64_t offset) {
-  assert(use_direct_io() &&
-         IsSectorAligned(offset, GetRequiredBufferAlignment()) &&
-         IsSectorAligned(data.size(), GetRequiredBufferAlignment()) &&
-         IsPageAligned(data.data()));
+  if (use_direct_io()) {
+    assert(IsSectorAligned(offset, GetRequiredBufferAlignment()));
+    assert(IsSectorAligned(data.size(), GetRequiredBufferAlignment()));
+    assert(IsSectorAligned(data.data(), GetRequiredBufferAlignment()));
+  }
   assert(offset <= std::numeric_limits<off_t>::max());
   const char* src = data.data();
   size_t left = data.size();
@@ -772,17 +799,9 @@ Status PosixWritableFile::Close() {
     fstat(fd_, &file_stats);
     // After ftruncate, we check whether ftruncate has the correct behavior.
     // If not, we should hack it with FALLOC_FL_PUNCH_HOLE
-    static bool buggy = false;
-    if (!buggy && (file_stats.st_size + file_stats.st_blksize - 1) /
+    if ((file_stats.st_size + file_stats.st_blksize - 1) /
             file_stats.st_blksize !=
         file_stats.st_blocks / (file_stats.st_blksize / 512)) {
-      fprintf(stderr,
-              "WARNING: Your kernel is buggy (<= 4.0.x) and does not free"
-              "preallocated blocks on truncate. Hacking around it, but you"
-              "should upgrade!\n");
-      buggy = true;
-    }
-    if (buggy) {
       IOSTATS_TIMER_GUARD(allocate_nanos);
       if (allow_fallocate_) {
         fallocate(fd_, FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE, filesize_,

env/io_posix.h

@@ -79,7 +79,10 @@ class PosixRandomAccessFile : public RandomAccessFile {
 
   virtual Status Read(uint64_t offset, size_t n, Slice* result,
                       char* scratch) const override;
-#if defined(OS_LINUX) || defined(OS_MACOSX)
+
+  virtual Status Prefetch(uint64_t offset, size_t n) override;
+
+#if defined(OS_LINUX) || defined(OS_MACOSX) || defined(OS_AIX)
   virtual size_t GetUniqueId(char* id, size_t max_size) const override;
 #endif
   virtual void Hint(AccessPattern pattern) override;

include/rocksdb/c.h

@@ -804,8 +804,9 @@ extern ROCKSDB_LIBRARY_API void rocksdb_options_set_allow_mmap_writes(
     rocksdb_options_t*, unsigned char);
 extern ROCKSDB_LIBRARY_API void rocksdb_options_set_use_direct_reads(
     rocksdb_options_t*, unsigned char);
-extern ROCKSDB_LIBRARY_API void rocksdb_options_set_use_direct_writes(
-    rocksdb_options_t*, unsigned char);
+extern ROCKSDB_LIBRARY_API void
+rocksdb_options_set_use_direct_io_for_flush_and_compaction(rocksdb_options_t*,
+                                                           unsigned char);
 extern ROCKSDB_LIBRARY_API void rocksdb_options_set_is_fd_close_on_exec(
     rocksdb_options_t*, unsigned char);
 extern ROCKSDB_LIBRARY_API void rocksdb_options_set_skip_log_error_on_recovery(

include/rocksdb/db.h

@@ -179,10 +179,37 @@ class DB {
                                     const std::string& column_family_name,
                                     ColumnFamilyHandle** handle);
 
+  // Bulk create column families with the same column family options.
+  // Return the handles of the column families through the argument handles.
+  // In case of error, the request may succeed partially, and handles will
+  // contain column family handles that it managed to create, and have size
+  // equal to the number of created column families.
+  virtual Status CreateColumnFamilies(
+      const ColumnFamilyOptions& options,
+      const std::vector<std::string>& column_family_names,
+      std::vector<ColumnFamilyHandle*>* handles);
+
+  // Bulk create column families.
+  // Return the handles of the column families through the argument handles.
+  // In case of error, the request may succeed partially, and handles will
+  // contain column family handles that it managed to create, and have size
+  // equal to the number of created column families.
+  virtual Status CreateColumnFamilies(
+      const std::vector<ColumnFamilyDescriptor>& column_families,
+      std::vector<ColumnFamilyHandle*>* handles);
+
   // Drop a column family specified by column_family handle. This call
   // only records a drop record in the manifest and prevents the column
   // family from flushing and compacting.
   virtual Status DropColumnFamily(ColumnFamilyHandle* column_family);
+
+  // Bulk drop column families. This call only records drop records in the
+  // manifest and prevents the column families from flushing and compacting.
+  // In case of error, the request may succeed partially. User may call
+  // ListColumnFamilies to check the result.
+  virtual Status DropColumnFamilies(
+      const std::vector<ColumnFamilyHandle*>& column_families);
+
   // Close a column family specified by column_family handle and destroy
   // the column family handle specified to avoid double deletion. This call
   // deletes the column family handle by default. Use this method to
@@ -611,6 +638,13 @@ class DB {
     return GetIntProperty(DefaultColumnFamily(), property, value);
   }
 
+  // Reset internal stats for DB and all column families.
+  // Note this doesn't reset options.statistics as it is not owned by
+  // DB.
+  virtual Status ResetStats() {
+    return Status::NotSupported("Not implemented");
+  }
+
   // Same as GetIntProperty(), but this one returns the aggregated int
   // property from all column families.
   virtual bool GetAggregatedIntProperty(const Slice& property,

include/rocksdb/env.h

@@ -44,6 +44,7 @@ class WritableFile;
 class RandomRWFile;
 class Directory;
 struct DBOptions;
+struct ImmutableDBOptions;
 class RateLimiter;
 class ThreadStatusUpdater;
 struct ThreadStatus;
@@ -162,6 +163,20 @@ class Env {
                                  unique_ptr<WritableFile>* result,
                                  const EnvOptions& options) = 0;
 
+  // Create an object that writes to a new file with the specified
+  // name.  Deletes any existing file with the same name and creates a
+  // new file.  On success, stores a pointer to the new file in
+  // *result and returns OK.  On failure stores nullptr in *result and
+  // returns non-OK.
+  //
+  // The returned file will only be accessed by one thread at a time.
+  virtual Status ReopenWritableFile(const std::string& fname,
+                                    unique_ptr<WritableFile>* result,
+                                    const EnvOptions& options) {
+    Status s;
+    return s;
+  }
+
   // Reuse an existing file by renaming it and opening it as writable.
   virtual Status ReuseWritableFile(const std::string& fname,
                                    const std::string& old_fname,
@@ -375,6 +390,20 @@ class Env {
   virtual EnvOptions OptimizeForManifestWrite(
       const EnvOptions& env_options) const;
 
+  // OptimizeForCompactionTableWrite will create a new EnvOptions object that is a copy
+  // of the EnvOptions in the parameters, but is optimized for writing table
+  // files. Default implementation returns the copy of the same object.
+  virtual EnvOptions OptimizeForCompactionTableWrite(
+      const EnvOptions& env_options,
+      const ImmutableDBOptions& db_options) const;
+
+  // OptimizeForCompactionTableWrite will create a new EnvOptions object that is a copy
+  // of the EnvOptions in the parameters, but is optimized for reading table
+  // files. Default implementation returns the copy of the same object.
+  virtual EnvOptions OptimizeForCompactionTableRead(
+      const EnvOptions& env_options,
+      const ImmutableDBOptions& db_options) const;
+
   // Returns the status of all threads that belong to the current Env.
   virtual Status GetThreadList(std::vector<ThreadStatus>* thread_list) {
     return Status::NotSupported("Not supported.");
@@ -457,6 +486,7 @@ class SequentialFile {
 // A file abstraction for randomly reading the contents of a file.
 class RandomAccessFile {
  public:
+
   RandomAccessFile() { }
   virtual ~RandomAccessFile();
 
@@ -473,6 +503,11 @@ class RandomAccessFile {
   virtual Status Read(uint64_t offset, size_t n, Slice* result,
                       char* scratch) const = 0;
 
+  // Readahead the file starting from offset by n bytes for caching.
+  virtual Status Prefetch(uint64_t offset, size_t n) {
+    return Status::OK();
+  }
+
   // Used by the file_reader_writer to decide if the ReadAhead wrapper
   // should simply forward the call and do not enact buffering or locking.
   virtual bool ShouldForwardRawRequest() const {
@@ -676,14 +711,12 @@ class WritableFile {
     }
   }
 
- protected:
-  /*
-   * Pre-allocate space for a file.
-   */
+  // Pre-allocates space for a file.
   virtual Status Allocate(uint64_t offset, uint64_t len) {
     return Status::OK();
   }
 
+ protected:
   size_t preallocation_block_size() { return preallocation_block_size_; }
 
  private:
@@ -897,6 +930,11 @@ class EnvWrapper : public Env {
                          const EnvOptions& options) override {
     return target_->NewWritableFile(f, r, options);
   }
+  Status ReopenWritableFile(const std::string& fname,
+                            unique_ptr<WritableFile>* result,
+                            const EnvOptions& options) override {
+    return target_->ReopenWritableFile(fname, result, options);
+  }
   Status ReuseWritableFile(const std::string& fname,
                            const std::string& old_fname,
                            unique_ptr<WritableFile>* r,

include/rocksdb/listener.h

@@ -183,6 +183,28 @@ struct ExternalFileIngestionInfo {
   TableProperties table_properties;
 };
 
+// A call-back function to RocksDB which will be called when the compaction
+// iterator is compacting values. It is mean to be returned from
+// EventListner::GetCompactionEventListner() at the beginning of compaction
+// job.
+class CompactionEventListener {
+ public:
+  enum CompactionListenerValueType {
+    kValue,
+    kMergeOperand,
+    kDelete,
+    kSingleDelete,
+    kRangeDelete,
+    kInvalid,
+  };
+
+  virtual void OnCompaction(int level, const Slice& key,
+                            CompactionListenerValueType value_type,
+                            const Slice& existing_value,
+                            const SequenceNumber& sn, bool is_new) = 0;
+
+  virtual ~CompactionEventListener() = default;
+};
 
 // EventListener class contains a set of call-back functions that will
 // be called when specific RocksDB event happens such as flush.  It can
@@ -225,6 +247,16 @@ class EventListener {
   virtual void OnFlushCompleted(DB* /*db*/,
                                 const FlushJobInfo& /*flush_job_info*/) {}
 
+  // A call-back function to RocksDB which will be called before a
+  // RocksDB starts to flush memtables.  The default implementation is
+  // no-op.
+  //
+  // Note that the this function must be implemented in a way such that
+  // it should not run for an extended period of time before the function
+  // returns.  Otherwise, RocksDB may be blocked.
+  virtual void OnFlushBegin(DB* /*db*/,
+                            const FlushJobInfo& /*flush_job_info*/) {}
+
   // A call-back function for RocksDB which will be called whenever
   // a SST file is deleted.  Different from OnCompactionCompleted and
   // OnFlushCompleted, this call-back is designed for external logging
@@ -314,6 +346,12 @@ class EventListener {
   virtual void OnExternalFileIngested(
       DB* /*db*/, const ExternalFileIngestionInfo& /*info*/) {}
 
+  // Factory method to return CompactionEventListener. If multiple listeners
+  // provides CompactionEventListner, only the first one will be used.
+  virtual CompactionEventListener* GetCompactionEventListener() {
+    return nullptr;
+  }
+
   virtual ~EventListener() {}
 };
 

include/rocksdb/options.h

@@ -581,15 +581,16 @@ struct DBOptions {
   // bufferized. The hardware buffer of the devices may however still
   // be used. Memory mapped files are not impacted by these parameters.
 
-  // Use O_DIRECT for reading file
+  // Use O_DIRECT for user reads
   // Default: false
   // Not supported in ROCKSDB_LITE mode!
   bool use_direct_reads = false;
 
-  // Use O_DIRECT for writing file
+  // Use O_DIRECT for both reads and writes in background flush and compactions
+  // When true, we also force new_table_reader_for_compaction_inputs to true.
   // Default: false
   // Not supported in ROCKSDB_LITE mode!
-  bool use_direct_writes = false;
+  bool use_direct_io_for_flush_and_compaction = false;
 
   // If false, fallocate() calls are bypassed
   bool allow_fallocate = true;
@@ -689,7 +690,9 @@ struct DBOptions {
 
   // This is the maximum buffer size that is used by WritableFileWriter.
   // On Windows, we need to maintain an aligned buffer for writes.
-  // We allow the buffer to grow until it's size hits the limit.
+  // We allow the buffer to grow until it's size hits the limit in buffered
+  // IO and fix the buffer size when using direct IO to ensure alignment of
+  // write requests if the logical sector size is unusual
   //
   // Default: 1024 * 1024 (1 MB)
   size_t writable_file_max_buffer_size = 1024 * 1024;

include/rocksdb/utilities/stackable_db.h

@@ -37,10 +37,28 @@ class StackableDB : public DB {
     return db_->CreateColumnFamily(options, column_family_name, handle);
   }
 
+  virtual Status CreateColumnFamilies(
+      const ColumnFamilyOptions& options,
+      const std::vector<std::string>& column_family_names,
+      std::vector<ColumnFamilyHandle*>* handles) override {
+    return db_->CreateColumnFamilies(options, column_family_names, handles);
+  }
+
+  virtual Status CreateColumnFamilies(
+      const std::vector<ColumnFamilyDescriptor>& column_families,
+      std::vector<ColumnFamilyHandle*>* handles) override {
+    return db_->CreateColumnFamilies(column_families, handles);
+  }
+
   virtual Status DropColumnFamily(ColumnFamilyHandle* column_family) override {
     return db_->DropColumnFamily(column_family);
   }
 
+  virtual Status DropColumnFamilies(
+      const std::vector<ColumnFamilyHandle*>& column_families) override {
+    return db_->DropColumnFamilies(column_families);
+  }
+
   virtual Status DestroyColumnFamilyHandle(
       ColumnFamilyHandle* column_family) override {
     return db_->DestroyColumnFamilyHandle(column_family);

include/rocksdb/version.h

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

java/crossbuild/Vagrantfile

@@ -7,11 +7,11 @@ VAGRANTFILE_API_VERSION = "2"
 Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
 
   config.vm.define "linux32" do |linux32|
-    linux32.vm.box = "hansode/centos-5.6-i386"
+    linux32.vm.box = "hansode/centos-5.11-i386"
   end
 
   config.vm.define "linux64" do |linux64|
-    linux64.vm.box = "hansode/centos-5.6-x86_64"
+    linux64.vm.box = "hansode/centos-5.11-x86_64"
   end
 
   config.vm.provider "virtualbox" do |v|

java/crossbuild/build-linux-centos.sh

@@ -1,17 +1,35 @@
 #!/usr/bin/env bash
+
+set -e
+
+# CentOS 5 is now end of life so we need to switch its repos over to the vault
+sed -i -e '/^mirrorlist=/d' -e 's%^#baseurl=http://mirror.centos.org/centos/$releasever/%baseurl=http://vault.centos.org/5.11/%g' /etc/yum.repos.d/CentOS-Base.repo
+sed -i -e '/^mirrorlist=/d' -e 's%^#baseurl=http://mirror.centos.org/centos/$releasever/%baseurl=http://vault.centos.org/5.11/%g' /etc/yum.repos.d/CentOS-fasttrack.repo
+if [ -f /etc/yum.repos.d/libselinux.repo ]; then
+	sed -i -e '/^mirrorlist=/d' -e 's%^#baseurl=http://mirror.centos.org/centos/$releasever/%baseurl=http://vault.centos.org/5.11/%g' /etc/yum.repos.d/libselinux.repo
+fi
+
+# enable EPEL
+sudo yum -y install epel-release
+
 # install all required packages for rocksdb that are available through yum
 ARCH=$(uname -i)
-sudo yum -y install openssl java-1.7.0-openjdk-devel.$ARCH 
+sudo yum -y install openssl java-1.7.0-openjdk-devel.$ARCH zlib-devel bzip2-devel lz4-devel snappy-devel
 
 # install gcc/g++ 4.8.2 via CERN (http://linux.web.cern.ch/linux/devtoolset/)
 sudo wget -O /etc/yum.repos.d/slc5-devtoolset.repo http://linuxsoft.cern.ch/cern/devtoolset/slc5-devtoolset.repo
-sudo wget -O /etc/pki/rpm-gpg/RPM-GPG-KEY-cern http://ftp.mirrorservice.org/sites/ftp.scientificlinux.org/linux/scientific/51/i386/RPM-GPG-KEYs/RPM-GPG-KEY-cern
+sudo wget -O /etc/pki/rpm-gpg/RPM-GPG-KEY-cern http://ftp.mirrorservice.org/sites/ftp.scientificlinux.org/linux/scientific/obsolete/51/i386/RPM-GPG-KEYs/RPM-GPG-KEY-cern
 sudo yum -y install devtoolset-2
 
+# install gflags
 wget https://github.com/gflags/gflags/archive/v2.0.tar.gz -O gflags-2.0.tar.gz
 tar xvfz gflags-2.0.tar.gz; cd gflags-2.0; scl enable devtoolset-2 ./configure; scl enable devtoolset-2 make; sudo make install
 export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/local/lib
 
+# install zstandard
+wget https://github.com/facebook/zstd/archive/v1.1.3.tar.gz -O zstd-1.1.3.tar.gz
+tar zxvf zstd-1.1.3.tar.gz; cd zstd-1.1.3; scl enable devtoolset-2 make; sudo make install
+
 # set java home so we can build rocksdb jars
 export JAVA_HOME=/usr/lib/jvm/java-1.7.0
 

java/crossbuild/docker-build-linux-centos.sh

@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+
+set -e
+
+rm -rf /rocksdb-local
+cp -r /rocksdb-host /rocksdb-local
+cd /rocksdb-local
+scl enable devtoolset-2 'make jclean clean'
+scl enable devtoolset-2 'PORTABLE=1 make rocksdbjavastatic'
+cp java/target/librocksdbjni-linux*.so java/target/rocksdbjni-*-linux*.jar /rocksdb-host/java/target
+

java/rocksjni/options.cc

@@ -1052,24 +1052,26 @@ void Java_org_rocksdb_Options_setUseDirectReads(JNIEnv* env, jobject jobj,
 
 /*
  * Class:     org_rocksdb_Options
- * Method:    useDirectWrites
+ * Method:    useDirectIoForFlushAndCompaction
  * Signature: (J)Z
  */
-jboolean Java_org_rocksdb_Options_useDirectWrites(JNIEnv* env, jobject jobj,
-                                                  jlong jhandle) {
-  return reinterpret_cast<rocksdb::Options*>(jhandle)->use_direct_writes;
+jboolean Java_org_rocksdb_Options_useDirectIoForFlushAndCompaction(
+    JNIEnv* env, jobject jobj, jlong jhandle) {
+  return reinterpret_cast<rocksdb::Options*>(jhandle)
+      ->use_direct_io_for_flush_and_compaction;
 }
 
 /*
  * Class:     org_rocksdb_Options
- * Method:    setUseDirectReads
+ * Method:    setUseDirectIoForFlushAndCompaction
  * Signature: (JZ)V
  */
-void Java_org_rocksdb_Options_setUseDirectWrites(JNIEnv* env, jobject jobj,
-                                                 jlong jhandle,
-                                                 jboolean use_direct_writes) {
-  reinterpret_cast<rocksdb::Options*>(jhandle)->use_direct_writes =
-      static_cast<bool>(use_direct_writes);
+void Java_org_rocksdb_Options_setUseDirectIoForFlushAndCompaction(
+    JNIEnv* env, jobject jobj, jlong jhandle,
+    jboolean use_direct_io_for_flush_and_compaction) {
+  reinterpret_cast<rocksdb::Options*>(jhandle)
+      ->use_direct_io_for_flush_and_compaction =
+      static_cast<bool>(use_direct_io_for_flush_and_compaction);
 }
 
 /*
@@ -4920,12 +4922,13 @@ void Java_org_rocksdb_DBOptions_setUseDirectReads(JNIEnv* env, jobject jobj,
 
 /*
  * Class:     org_rocksdb_DBOptions
- * Method:    useDirectWrites
+ * Method:    useDirectIoForFlushAndCompaction
  * Signature: (J)Z
  */
-jboolean Java_org_rocksdb_DBOptions_useDirectWrites(JNIEnv* env, jobject jobj,
-                                                    jlong jhandle) {
-  return reinterpret_cast<rocksdb::DBOptions*>(jhandle)->use_direct_writes;
+jboolean Java_org_rocksdb_DBOptions_useDirectIoForFlushAndCompaction(
+    JNIEnv* env, jobject jobj, jlong jhandle) {
+  return reinterpret_cast<rocksdb::DBOptions*>(jhandle)
+      ->use_direct_io_for_flush_and_compaction;
 }
 
 /*
@@ -4933,11 +4936,12 @@ jboolean Java_org_rocksdb_DBOptions_useDirectWrites(JNIEnv* env, jobject jobj,
  * Method:    setUseDirectReads
  * Signature: (JZ)V
  */
-void Java_org_rocksdb_DBOptions_setUseDirectWrites(JNIEnv* env, jobject jobj,
-                                                   jlong jhandle,
-                                                   jboolean use_direct_writes) {
-  reinterpret_cast<rocksdb::DBOptions*>(jhandle)->use_direct_writes =
-      static_cast<bool>(use_direct_writes);
+void Java_org_rocksdb_DBOptions_setUseDirectIoForFlushAndCompaction(
+    JNIEnv* env, jobject jobj, jlong jhandle,
+    jboolean use_direct_io_for_flush_and_compaction) {
+  reinterpret_cast<rocksdb::DBOptions*>(jhandle)
+      ->use_direct_io_for_flush_and_compaction =
+      static_cast<bool>(use_direct_io_for_flush_and_compaction);
 }
 
 /*

java/src/main/java/org/rocksdb/DBOptions.java

@@ -531,17 +531,18 @@ public class DBOptions
   }
 
   @Override
-  public DBOptions setUseDirectWrites(
-      final boolean useDirectWrites) {
+  public DBOptions setUseDirectIoForFlushAndCompaction(
+      final boolean useDirectIoForFlushAndCompaction) {
     assert(isOwningHandle());
-    setUseDirectWrites(nativeHandle_, useDirectWrites);
+    setUseDirectIoForFlushAndCompaction(nativeHandle_,
+        useDirectIoForFlushAndCompaction);
     return this;
   }
 
   @Override
-  public boolean useDirectWrites() {
+  public boolean useDirectIoForFlushAndCompaction() {
     assert(isOwningHandle());
-    return useDirectWrites(nativeHandle_);
+    return useDirectIoForFlushAndCompaction(nativeHandle_);
   }
 
   @Override
@@ -1025,8 +1026,9 @@ public class DBOptions
   private native long manifestPreallocationSize(long handle);
   private native void setUseDirectReads(long handle, boolean useDirectReads);
   private native boolean useDirectReads(long handle);
-  private native void setUseDirectWrites(long handle, boolean useDirectWrites);
-  private native boolean useDirectWrites(long handle);
+  private native void setUseDirectIoForFlushAndCompaction(
+      long handle, boolean useDirectIoForFlushAndCompaction);
+  private native boolean useDirectIoForFlushAndCompaction(long handle);
   private native void setAllowFAllocate(final long handle,
       final boolean allowFAllocate);
   private native boolean allowFAllocate(final long handle);

java/src/main/java/org/rocksdb/DBOptionsInterface.java

@@ -804,21 +804,24 @@ public interface DBOptionsInterface<T extends DBOptionsInterface> {
   boolean useDirectReads();
 
   /**
-   * Enable the OS to use direct I/O for writing sst tables.
+   * Enable the OS to use direct reads and writes in flush and
+   * compaction
    * Default: false
    *
-   * @param useDirectWrites if true, then direct write is enabled
+   * @param useDirectIoForFlushAndCompaction if true, then direct
+   *        I/O will be enabled for background flush and compactions
    * @return the instance of the current object.
    */
-  T setUseDirectWrites(boolean useDirectWrites);
+  T setUseDirectIoForFlushAndCompaction(boolean useDirectIoForFlushAndCompaction);
 
   /**
-   * Enable the OS to use direct I/O for writing sst tables.
-   * Default: false
+   * Enable the OS to use direct reads and writes in flush and
+   * compaction
    *
-   * @return if true, then direct writes are enabled
+   * @return if true, then direct I/O is enabled for flush and
+   *         compaction
    */
-  boolean useDirectWrites();
+  boolean useDirectIoForFlushAndCompaction();
 
   /**
    * Whether fallocate calls are allowed

java/src/main/java/org/rocksdb/Options.java

@@ -593,16 +593,17 @@ public class Options extends RocksObject
   }
 
   @Override
-  public Options setUseDirectWrites(final boolean useDirectWrites) {
+  public Options setUseDirectIoForFlushAndCompaction(
+      final boolean useDirectIoForFlushAndCompaction) {
     assert(isOwningHandle());
-    setUseDirectWrites(nativeHandle_, useDirectWrites);
+    setUseDirectIoForFlushAndCompaction(nativeHandle_, useDirectIoForFlushAndCompaction);
     return this;
   }
 
   @Override
-  public boolean useDirectWrites() {
+  public boolean useDirectIoForFlushAndCompaction() {
     assert(isOwningHandle());
-    return useDirectWrites(nativeHandle_);
+    return useDirectIoForFlushAndCompaction(nativeHandle_);
   }
 
   @Override
@@ -1621,8 +1622,9 @@ public class Options extends RocksObject
   private native long manifestPreallocationSize(long handle);
   private native void setUseDirectReads(long handle, boolean useDirectReads);
   private native boolean useDirectReads(long handle);
-  private native void setUseDirectWrites(long handle, boolean useDirectWrites);
-  private native boolean useDirectWrites(long handle);
+  private native void setUseDirectIoForFlushAndCompaction(
+      long handle, boolean useDirectIoForFlushAndCompaction);
+  private native boolean useDirectIoForFlushAndCompaction(long handle);
   private native void setAllowFAllocate(final long handle,
       final boolean allowFAllocate);
   private native boolean allowFAllocate(final long handle);

java/src/test/java/org/rocksdb/DBOptionsTest.java

@@ -331,11 +331,11 @@ public class DBOptionsTest {
   }
 
   @Test
-  public void useDirectWrites() {
+  public void useDirectIoForFlushAndCompaction() {
     try(final DBOptions opt = new DBOptions()) {
       final boolean boolValue = rand.nextBoolean();
-      opt.setUseDirectWrites(boolValue);
-      assertThat(opt.useDirectWrites()).isEqualTo(boolValue);
+      opt.setUseDirectIoForFlushAndCompaction(boolValue);
+      assertThat(opt.useDirectIoForFlushAndCompaction()).isEqualTo(boolValue);
     }
   }
 

java/src/test/java/org/rocksdb/OptionsTest.java

@@ -553,11 +553,11 @@ public class OptionsTest {
   }
 
   @Test
-  public void useDirectWrites() {
+  public void useDirectIoForFlushAndCompaction() {
     try(final Options opt = new Options()) {
       final boolean boolValue = rand.nextBoolean();
-      opt.setUseDirectWrites(boolValue);
-      assertThat(opt.useDirectWrites()).isEqualTo(boolValue);
+      opt.setUseDirectIoForFlushAndCompaction(boolValue);
+      assertThat(opt.useDirectIoForFlushAndCompaction()).isEqualTo(boolValue);
     }
   }
 

options/db_options.cc

@@ -53,7 +53,8 @@ ImmutableDBOptions::ImmutableDBOptions(const DBOptions& options)
       allow_mmap_reads(options.allow_mmap_reads),
       allow_mmap_writes(options.allow_mmap_writes),
       use_direct_reads(options.use_direct_reads),
-      use_direct_writes(options.use_direct_writes),
+      use_direct_io_for_flush_and_compaction(
+          options.use_direct_io_for_flush_and_compaction),
       allow_fallocate(options.allow_fallocate),
       is_fd_close_on_exec(options.is_fd_close_on_exec),
       advise_random_on_open(options.advise_random_on_open),
@@ -127,8 +128,10 @@ void ImmutableDBOptions::Dump(Logger* log) const {
                    allow_mmap_writes);
   ROCKS_LOG_HEADER(log, "                       Options.use_direct_reads: %d",
                    use_direct_reads);
-  ROCKS_LOG_HEADER(log, "                       Options.use_direct_writes: %d",
-                   use_direct_writes);
+  ROCKS_LOG_HEADER(log,
+                   "                       "
+                   "Options.use_direct_io_for_flush_and_compaction: %d",
+                   use_direct_io_for_flush_and_compaction);
   ROCKS_LOG_HEADER(log, "         Options.create_missing_column_families: %d",
                    create_missing_column_families);
   ROCKS_LOG_HEADER(log, "                             Options.db_log_dir: %s",

options/db_options.h

@@ -48,7 +48,7 @@ struct ImmutableDBOptions {
   bool allow_mmap_reads;
   bool allow_mmap_writes;
   bool use_direct_reads;
-  bool use_direct_writes;
+  bool use_direct_io_for_flush_and_compaction;
   bool allow_fallocate;
   bool is_fd_close_on_exec;
   bool advise_random_on_open;

options/options.cc

@@ -153,7 +153,8 @@ DBOptions::DBOptions(const Options& options)
       allow_mmap_reads(options.allow_mmap_reads),
       allow_mmap_writes(options.allow_mmap_writes),
       use_direct_reads(options.use_direct_reads),
-      use_direct_writes(options.use_direct_writes),
+      use_direct_io_for_flush_and_compaction(
+          options.use_direct_io_for_flush_and_compaction),
       allow_fallocate(options.allow_fallocate),
       is_fd_close_on_exec(options.is_fd_close_on_exec),
       skip_log_error_on_recovery(options.skip_log_error_on_recovery),

options/options_helper.cc

@@ -70,7 +70,8 @@ DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.allow_mmap_reads = immutable_db_options.allow_mmap_reads;
   options.allow_mmap_writes = immutable_db_options.allow_mmap_writes;
   options.use_direct_reads = immutable_db_options.use_direct_reads;
-  options.use_direct_writes = immutable_db_options.use_direct_writes;
+  options.use_direct_io_for_flush_and_compaction =
+      immutable_db_options.use_direct_io_for_flush_and_compaction;
   options.allow_fallocate = immutable_db_options.allow_fallocate;
   options.is_fd_close_on_exec = immutable_db_options.is_fd_close_on_exec;
   options.stats_dump_period_sec = mutable_db_options.stats_dump_period_sec;
@@ -1031,7 +1032,8 @@ std::string ParseBlockBasedTableOption(const std::string& name,
     return "Unrecognized option";
   }
   const auto& opt_info = iter->second;
-  if (!ParseOptionHelper(reinterpret_cast<char*>(new_options) + opt_info.offset,
+  if (opt_info.verification != OptionVerificationType::kDeprecated &&
+      !ParseOptionHelper(reinterpret_cast<char*>(new_options) + opt_info.offset,
                          opt_info.type, value)) {
     return "Invalid value";
   }
@@ -1040,7 +1042,7 @@ std::string ParseBlockBasedTableOption(const std::string& name,
 
 std::string ParsePlainTableOptions(const std::string& name,
                                    const std::string& org_value,
-                                   PlainTableOptions* new_option,
+                                   PlainTableOptions* new_options,
                                    bool input_strings_escaped = false) {
   const std::string& value =
       input_strings_escaped ? UnescapeOptionString(org_value) : org_value;
@@ -1049,7 +1051,8 @@ std::string ParsePlainTableOptions(const std::string& name,
     return "Unrecognized option";
   }
   const auto& opt_info = iter->second;
-  if (!ParseOptionHelper(reinterpret_cast<char*>(new_option) + opt_info.offset,
+  if (opt_info.verification != OptionVerificationType::kDeprecated &&
+      !ParseOptionHelper(reinterpret_cast<char*>(new_options) + opt_info.offset,
                          opt_info.type, value)) {
     return "Invalid value";
   }

options/options_helper.h

@@ -163,8 +163,10 @@ static std::unordered_map<std::string, OptionTypeInfo> db_options_type_info = {
      {offsetof(struct DBOptions, use_direct_reads), OptionType::kBoolean,
       OptionVerificationType::kNormal, false, 0}},
     {"use_direct_writes",
-     {offsetof(struct DBOptions, use_direct_writes), OptionType::kBoolean,
-      OptionVerificationType::kNormal, false, 0}},
+     {0, OptionType::kBoolean, OptionVerificationType::kDeprecated, false, 0}},
+    {"use_direct_io_for_flush_and_compaction",
+     {offsetof(struct DBOptions, use_direct_io_for_flush_and_compaction),
+      OptionType::kBoolean, OptionVerificationType::kNormal, false, 0}},
     {"allow_2pc",
      {offsetof(struct DBOptions, allow_2pc), OptionType::kBoolean,
       OptionVerificationType::kNormal, false, 0}},
@@ -598,6 +600,9 @@ static std::unordered_map<std::string, OptionTypeInfo>
         {"index_block_restart_interval",
          {offsetof(struct BlockBasedTableOptions, index_block_restart_interval),
           OptionType::kInt, OptionVerificationType::kNormal, false, 0}},
+        {"index_per_partition",
+         {0, OptionType::kUInt64T, OptionVerificationType::kDeprecated, false,
+          0}},
         {"metadata_block_size",
          {offsetof(struct BlockBasedTableOptions, metadata_block_size),
           OptionType::kUInt64T, OptionVerificationType::kNormal, false, 0}},

options/options_settable_test.cc

@@ -274,7 +274,7 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
                              "allow_fallocate=true;"
                              "allow_mmap_reads=false;"
                              "use_direct_reads=false;"
-                             "use_direct_writes=false;"
+                             "use_direct_io_for_flush_and_compaction=false;"
                              "max_log_file_size=4607;"
                              "random_access_max_buffer_size=1048576;"
                              "advise_random_on_open=true;"

options/options_test.cc

@@ -120,7 +120,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
       {"allow_mmap_reads", "true"},
       {"allow_mmap_writes", "false"},
       {"use_direct_reads", "false"},
-      {"use_direct_writes", "false"},
+      {"use_direct_io_for_flush_and_compaction", "false"},
       {"is_fd_close_on_exec", "true"},
       {"skip_log_error_on_recovery", "false"},
       {"stats_dump_period_sec", "46"},
@@ -234,7 +234,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_db_opt.allow_mmap_reads, true);
   ASSERT_EQ(new_db_opt.allow_mmap_writes, false);
   ASSERT_EQ(new_db_opt.use_direct_reads, false);
-  ASSERT_EQ(new_db_opt.use_direct_writes, false);
+  ASSERT_EQ(new_db_opt.use_direct_io_for_flush_and_compaction, false);
   ASSERT_EQ(new_db_opt.is_fd_close_on_exec, true);
   ASSERT_EQ(new_db_opt.skip_log_error_on_recovery, false);
   ASSERT_EQ(new_db_opt.stats_dump_period_sec, 46U);

src.mk

@@ -150,6 +150,12 @@ LIB_SOURCES =                                                   \
   util/xxhash.cc                                                \
   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                          \
+  utilities/blob_db/blob_log_format.cc                          \
   utilities/checkpoint/checkpoint.cc                            \
   utilities/compaction_filters/remove_emptyvalue_compactionfilter.cc    \
   utilities/convenience/info_log_finder.cc                      \
@@ -307,10 +313,10 @@ MAIN_SOURCES =                                                    \
   util/log_write_bench.cc                                               \
   util/rate_limiter_test.cc                                             \
   util/slice_transform_test.cc                                          \
+  util/timer_queue_test.cc                                             \
   util/thread_list_test.cc                                              \
   util/thread_local_test.cc                                             \
   utilities/backupable/backupable_db_test.cc                            \
-  utilities/blob_db/blob_db_test.cc                                     \
   utilities/checkpoint/checkpoint_test.cc                               \
   utilities/column_aware_encoding_exp.cc                                \
   utilities/column_aware_encoding_test.cc                               \

table/block.h

@@ -27,7 +27,8 @@
 #include "rocksdb/statistics.h"
 #include "table/block_prefix_index.h"
 #include "table/internal_iterator.h"
-
+#include "util/random.h"
+#include "util/sync_point.h"
 #include "format.h"
 
 namespace rocksdb {
@@ -44,7 +45,12 @@ class BlockReadAmpBitmap {
  public:
   explicit BlockReadAmpBitmap(size_t block_size, size_t bytes_per_bit,
                               Statistics* statistics)
-      : bitmap_(nullptr), bytes_per_bit_pow_(0), statistics_(statistics) {
+      : bitmap_(nullptr),
+        bytes_per_bit_pow_(0),
+        statistics_(statistics),
+        rnd_(
+            Random::GetTLSInstance()->Uniform(static_cast<int>(bytes_per_bit))) {
+    TEST_SYNC_POINT_CALLBACK("BlockReadAmpBitmap:rnd", &rnd_);
     assert(block_size > 0 && bytes_per_bit > 0);
 
     // convert bytes_per_bit to be a power of 2
@@ -54,62 +60,38 @@ class BlockReadAmpBitmap {
 
     // num_bits_needed = ceil(block_size / bytes_per_bit)
     size_t num_bits_needed =
-        (block_size >> static_cast<size_t>(bytes_per_bit_pow_)) +
-        (block_size % (static_cast<size_t>(1)
-                       << static_cast<size_t>(bytes_per_bit_pow_)) !=
-         0);
+      ((block_size - 1) >> bytes_per_bit_pow_) + 1;
+    assert(num_bits_needed > 0);
 
     // bitmap_size = ceil(num_bits_needed / kBitsPerEntry)
-    size_t bitmap_size = (num_bits_needed / kBitsPerEntry) +
-                         (num_bits_needed % kBitsPerEntry != 0);
+    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);
 
-    RecordTick(GetStatistics(), READ_AMP_TOTAL_READ_BYTES,
-               num_bits_needed << bytes_per_bit_pow_);
+    RecordTick(GetStatistics(), READ_AMP_TOTAL_READ_BYTES, block_size);
   }
 
   ~BlockReadAmpBitmap() { delete[] bitmap_; }
 
   void Mark(uint32_t start_offset, uint32_t end_offset) {
     assert(end_offset >= start_offset);
-
-    // Every new bit we set will bump this counter
-    uint32_t new_useful_bytes = 0;
-    // Index of first bit in mask (start_offset / bytes_per_bit)
-    uint32_t start_bit = start_offset >> bytes_per_bit_pow_;
-    // Index of last bit in mask (end_offset / bytes_per_bit)
-    uint32_t end_bit = end_offset >> bytes_per_bit_pow_;
-    // Index of middle bit (unique to this range)
-    uint32_t mid_bit = start_bit + 1;
-
-    // It's guaranteed that ranges sent to Mark() wont overlap, this mean that
-    // we dont need to set the middle bits, we can simply set only one bit of
-    // the middle bits, and check this bit if we want to know if the whole
-    // range is set or not.
-    if (mid_bit < end_bit) {
-      if (GetAndSet(mid_bit) == 0) {
-        new_useful_bytes += (end_bit - mid_bit) << bytes_per_bit_pow_;
-      } else {
-        // If the middle bit is set, it's guaranteed that start and end bits
-        // are also set
-        return;
-      }
-    } else {
-      // This range dont have a middle bit, the whole range fall in 1 or 2 bits
+    // Index of first bit in mask
+    uint32_t start_bit =
+        (start_offset + (1 << bytes_per_bit_pow_) - rnd_ - 1) >>
+        bytes_per_bit_pow_;
+    // Index of last bit in mask + 1
+    uint32_t exclusive_end_bit =
+        (end_offset + (1 << bytes_per_bit_pow_) - rnd_) >> bytes_per_bit_pow_;
+    if (start_bit >= exclusive_end_bit) {
+      return;
     }
+    assert(exclusive_end_bit > 0);
 
     if (GetAndSet(start_bit) == 0) {
-      new_useful_bytes += (1 << bytes_per_bit_pow_);
-    }
-
-    if (GetAndSet(end_bit) == 0) {
-      new_useful_bytes += (1 << bytes_per_bit_pow_);
-    }
-
-    if (new_useful_bytes > 0) {
+      uint32_t new_useful_bytes = (exclusive_end_bit - start_bit)
+                                  << bytes_per_bit_pow_;
       RecordTick(GetStatistics(), READ_AMP_ESTIMATE_USEFUL_BYTES,
                  new_useful_bytes);
     }
@@ -146,6 +128,7 @@ class BlockReadAmpBitmap {
   // this pointer maybe invalid, but the DB will update it to a valid pointer
   // by using SetStatistics() before calling Mark()
   std::atomic<Statistics*> statistics_;
+  uint32_t rnd_;
 };
 
 class Block {

table/block_based_table_reader.cc

@@ -159,7 +159,8 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
   static Status Create(BlockBasedTable* table, RandomAccessFileReader* file,
                        const Footer& footer, const BlockHandle& index_handle,
                        const ImmutableCFOptions& ioptions,
-                       const Comparator* comparator, IndexReader** index_reader,
+                       const InternalKeyComparator* icomparator,
+                       IndexReader** index_reader,
                        const PersistentCacheOptions& cache_options,
                        const int level) {
     std::unique_ptr<Block> index_block;
@@ -170,7 +171,7 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
 
     if (s.ok()) {
       *index_reader =
-          new PartitionIndexReader(table, comparator, std::move(index_block),
+          new PartitionIndexReader(table, icomparator, std::move(index_block),
                                    ioptions.statistics, level);
     }
 
@@ -194,8 +195,9 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
     }
     return NewTwoLevelIterator(
         new BlockBasedTable::BlockEntryIteratorState(
-            table_, ReadOptions(), skip_filters, is_index, block_cache_cleaner),
-        index_block_->NewIterator(comparator_, nullptr, true));
+            table_, ReadOptions(), icomparator_, skip_filters, is_index,
+            block_cache_cleaner),
+        index_block_->NewIterator(icomparator_, nullptr, true));
     // TODO(myabandeh): Update TwoLevelIterator to be able to make use of
     // on-stack
     // BlockIter while the state is on heap
@@ -212,10 +214,11 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
   }
 
  private:
-  PartitionIndexReader(BlockBasedTable* table, const Comparator* comparator,
+  PartitionIndexReader(BlockBasedTable* table,
+                       const InternalKeyComparator* icomparator,
                        std::unique_ptr<Block>&& index_block, Statistics* stats,
                        const int level)
-      : IndexReader(comparator, stats),
+      : IndexReader(icomparator, stats),
         table_(table),
         index_block_(std::move(index_block)),
         level_(level) {
@@ -237,8 +240,9 @@ class BinarySearchIndexReader : public IndexReader {
   // unmodified.
   static Status Create(RandomAccessFileReader* file, const Footer& footer,
                        const BlockHandle& index_handle,
-                       const ImmutableCFOptions &ioptions,
-                       const Comparator* comparator, IndexReader** index_reader,
+                       const ImmutableCFOptions& ioptions,
+                       const InternalKeyComparator* icomparator,
+                       IndexReader** index_reader,
                        const PersistentCacheOptions& cache_options) {
     std::unique_ptr<Block> index_block;
     auto s = ReadBlockFromFile(
@@ -248,7 +252,7 @@ class BinarySearchIndexReader : public IndexReader {
 
     if (s.ok()) {
       *index_reader = new BinarySearchIndexReader(
-          comparator, std::move(index_block), ioptions.statistics);
+          icomparator, std::move(index_block), ioptions.statistics);
     }
 
     return s;
@@ -256,7 +260,7 @@ class BinarySearchIndexReader : public IndexReader {
 
   virtual InternalIterator* NewIterator(BlockIter* iter = nullptr,
                                         bool dont_care = true) override {
-    return index_block_->NewIterator(comparator_, iter, true);
+    return index_block_->NewIterator(icomparator_, iter, true);
   }
 
   virtual size_t size() const override { return index_block_->size(); }
@@ -270,10 +274,10 @@ class BinarySearchIndexReader : public IndexReader {
   }
 
  private:
-  BinarySearchIndexReader(const Comparator* comparator,
+  BinarySearchIndexReader(const InternalKeyComparator* icomparator,
                           std::unique_ptr<Block>&& index_block,
                           Statistics* stats)
-      : IndexReader(comparator, stats), index_block_(std::move(index_block)) {
+      : IndexReader(icomparator, stats), index_block_(std::move(index_block)) {
     assert(index_block_ != nullptr);
   }
   std::unique_ptr<Block> index_block_;
@@ -286,7 +290,7 @@ class HashIndexReader : public IndexReader {
   static Status Create(const SliceTransform* hash_key_extractor,
                        const Footer& footer, RandomAccessFileReader* file,
                        const ImmutableCFOptions& ioptions,
-                       const Comparator* comparator,
+                       const InternalKeyComparator* icomparator,
                        const BlockHandle& index_handle,
                        InternalIterator* meta_index_iter,
                        IndexReader** index_reader,
@@ -307,7 +311,7 @@ class HashIndexReader : public IndexReader {
     // So, Create will succeed regardless, from this point on.
 
     auto new_index_reader =
-        new HashIndexReader(comparator, std::move(index_block),
+        new HashIndexReader(icomparator, std::move(index_block),
           ioptions.statistics);
     *index_reader = new_index_reader;
 
@@ -359,7 +363,7 @@ class HashIndexReader : public IndexReader {
 
   virtual InternalIterator* NewIterator(BlockIter* iter = nullptr,
                                         bool total_order_seek = true) override {
-    return index_block_->NewIterator(comparator_, iter, total_order_seek);
+    return index_block_->NewIterator(icomparator_, iter, total_order_seek);
   }
 
   virtual size_t size() const override { return index_block_->size(); }
@@ -374,9 +378,9 @@ class HashIndexReader : public IndexReader {
   }
 
  private:
-  HashIndexReader(const Comparator* comparator,
+  HashIndexReader(const InternalKeyComparator* icomparator,
                   std::unique_ptr<Block>&& index_block, Statistics* stats)
-      : IndexReader(comparator, stats), index_block_(std::move(index_block)) {
+      : IndexReader(icomparator, stats), index_block_(std::move(index_block)) {
     assert(index_block_ != nullptr);
   }
 
@@ -530,7 +534,12 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
   table_reader->reset();
 
   Footer footer;
-  auto s = ReadFooterFromFile(file.get(), file_size, &footer,
+
+  // Before read footer, readahead backwards to prefetch data
+  Status s =
+      file->Prefetch((file_size < 512 * 1024 ? 0 : file_size - 512 * 1024),
+                     512 * 1024 /* 512 KB prefetching */);
+  s = ReadFooterFromFile(file.get(), file_size, &footer,
                               kBlockBasedTableMagicNumber);
   if (!s.ok()) {
     return s;
@@ -541,8 +550,7 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
         "version of RocksDB?");
   }
 
-  // We've successfully read the footer and the index block: we're
-  // ready to serve requests.
+  // We've successfully read the footer. We are ready to serve requests.
   // Better not mutate rep_ after the creation. eg. internal_prefix_transform
   // raw pointer will be used to create HashIndexReader, whose reset may
   // access a dangling pointer.
@@ -1375,11 +1383,13 @@ Status BlockBasedTable::MaybeLoadDataBlockToCache(
 }
 
 BlockBasedTable::BlockEntryIteratorState::BlockEntryIteratorState(
-    BlockBasedTable* table, const ReadOptions& read_options, bool skip_filters,
-    bool is_index, Cleanable* block_cache_cleaner)
+    BlockBasedTable* table, const ReadOptions& read_options,
+    const InternalKeyComparator* icomparator, bool skip_filters, bool is_index,
+    Cleanable* block_cache_cleaner)
     : TwoLevelIteratorState(table->rep_->ioptions.prefix_extractor != nullptr),
       table_(table),
       read_options_(read_options),
+      icomparator_(icomparator),
       skip_filters_(skip_filters),
       is_index_(is_index),
       block_cache_cleaner_(block_cache_cleaner) {}
@@ -1418,6 +1428,19 @@ bool BlockBasedTable::BlockEntryIteratorState::PrefixMayMatch(
   return table_->PrefixMayMatch(internal_key);
 }
 
+bool BlockBasedTable::BlockEntryIteratorState::KeyReachedUpperBound(
+    const Slice& internal_key) {
+  bool reached_upper_bound = read_options_.iterate_upper_bound != nullptr &&
+                             icomparator_ != nullptr &&
+                             icomparator_->user_comparator()->Compare(
+                                 ExtractUserKey(internal_key),
+                                 *read_options_.iterate_upper_bound) >= 0;
+  TEST_SYNC_POINT_CALLBACK(
+      "BlockBasedTable::BlockEntryIteratorState::KeyReachedUpperBound",
+      &reached_upper_bound);
+  return reached_upper_bound;
+}
+
 // This will be broken if the user specifies an unusual implementation
 // of Options.comparator, or if the user specifies an unusual
 // definition of prefixes in BlockBasedTableOptions.filter_policy.
@@ -1520,11 +1543,11 @@ bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key) {
   return may_match;
 }
 
-InternalIterator* BlockBasedTable::NewIterator(const ReadOptions& read_options,
-                                               Arena* arena,
-                                               bool skip_filters) {
+InternalIterator* BlockBasedTable::NewIterator(
+    const ReadOptions& read_options, Arena* arena,
+    const InternalKeyComparator* icomp, bool skip_filters) {
   return NewTwoLevelIterator(
-      new BlockEntryIteratorState(this, read_options, skip_filters),
+      new BlockEntryIteratorState(this, read_options, icomp, skip_filters),
       NewIndexIterator(read_options), arena);
 }
 
@@ -1776,7 +1799,7 @@ Status BlockBasedTable::CreateIndexReader(
   }
 
   auto file = rep_->file.get();
-  auto comparator = &rep_->internal_comparator;
+  const InternalKeyComparator* icomparator = &rep_->internal_comparator;
   const Footer& footer = rep_->footer;
   if (index_type_on_file == BlockBasedTableOptions::kHashSearch &&
       rep_->ioptions.prefix_extractor == nullptr) {
@@ -1790,12 +1813,12 @@ Status BlockBasedTable::CreateIndexReader(
   switch (index_type_on_file) {
     case BlockBasedTableOptions::kTwoLevelIndexSearch: {
       return PartitionIndexReader::Create(
-          this, file, footer, footer.index_handle(), rep_->ioptions, comparator,
-          index_reader, rep_->persistent_cache_options, level);
+          this, file, footer, footer.index_handle(), rep_->ioptions,
+          icomparator, index_reader, rep_->persistent_cache_options, level);
     }
     case BlockBasedTableOptions::kBinarySearch: {
       return BinarySearchIndexReader::Create(
-          file, footer, footer.index_handle(), rep_->ioptions, comparator,
+          file, footer, footer.index_handle(), rep_->ioptions, icomparator,
           index_reader, rep_->persistent_cache_options);
     }
     case BlockBasedTableOptions::kHashSearch: {
@@ -1811,7 +1834,7 @@ Status BlockBasedTable::CreateIndexReader(
                          "Unable to read the metaindex block."
                          " Fall back to binary search index.");
           return BinarySearchIndexReader::Create(
-              file, footer, footer.index_handle(), rep_->ioptions, comparator,
+              file, footer, footer.index_handle(), rep_->ioptions, icomparator,
               index_reader, rep_->persistent_cache_options);
         }
         meta_index_iter = meta_iter_guard.get();
@@ -1819,7 +1842,7 @@ Status BlockBasedTable::CreateIndexReader(
 
       return HashIndexReader::Create(
           rep_->internal_prefix_transform.get(), footer, file, rep_->ioptions,
-          comparator, footer.index_handle(), meta_index_iter, index_reader,
+          icomparator, footer.index_handle(), meta_index_iter, index_reader,
           rep_->hash_index_allow_collision, rep_->persistent_cache_options);
     }
     default: {

table/block_based_table_reader.h

@@ -100,8 +100,10 @@ class BlockBasedTable : public TableReader {
   // The result of NewIterator() is initially invalid (caller must
   // call one of the Seek methods on the iterator before using it).
   // @param skip_filters Disables loading/accessing the filter block
-  InternalIterator* NewIterator(const ReadOptions&, Arena* arena = nullptr,
-                                bool skip_filters = false) override;
+  InternalIterator* NewIterator(
+      const ReadOptions&, Arena* arena = nullptr,
+      const InternalKeyComparator* icomparator = nullptr,
+      bool skip_filters = false) override;
 
   InternalIterator* NewRangeTombstoneIterator(
       const ReadOptions& read_options) override;
@@ -149,8 +151,9 @@ class BlockBasedTable : public TableReader {
   // access.
   class IndexReader {
    public:
-    explicit IndexReader(const Comparator* comparator, Statistics* stats)
-        : comparator_(comparator), statistics_(stats) {}
+    explicit IndexReader(const InternalKeyComparator* icomparator,
+                         Statistics* stats)
+        : icomparator_(icomparator), statistics_(stats) {}
 
     virtual ~IndexReader() {}
 
@@ -178,7 +181,7 @@ class BlockBasedTable : public TableReader {
     virtual size_t ApproximateMemoryUsage() const = 0;
 
    protected:
-    const Comparator* comparator_;
+    const InternalKeyComparator* icomparator_;
 
    private:
     Statistics* statistics_;
@@ -341,16 +344,19 @@ class BlockBasedTable : public TableReader {
 class BlockBasedTable::BlockEntryIteratorState : public TwoLevelIteratorState {
  public:
   BlockEntryIteratorState(BlockBasedTable* table,
-                          const ReadOptions& read_options, bool skip_filters,
-                          bool is_index = false,
+                          const ReadOptions& read_options,
+                          const InternalKeyComparator* icomparator,
+                          bool skip_filters, bool is_index = false,
                           Cleanable* block_cache_cleaner = nullptr);
   InternalIterator* NewSecondaryIterator(const Slice& index_value) override;
   bool PrefixMayMatch(const Slice& internal_key) override;
+  bool KeyReachedUpperBound(const Slice& internal_key) override;
 
  private:
   // Don't own table_
   BlockBasedTable* table_;
   const ReadOptions read_options_;
+  const InternalKeyComparator* icomparator_;
   bool skip_filters_;
   // true if the 2nd level iterator is on indexes instead of on user data.
   bool is_index_;

table/block_test.cc

@@ -226,18 +226,17 @@ class BlockReadAmpBitmapSlowAndAccurate {
  public:
   void Mark(size_t start_offset, size_t end_offset) {
     assert(end_offset >= start_offset);
-
     marked_ranges_.emplace(end_offset, start_offset);
   }
 
   // Return true if any byte in this range was Marked
-  bool IsAnyInRangeMarked(size_t start_offset, size_t end_offset) {
+  bool IsPinMarked(size_t offset) {
     auto it = marked_ranges_.lower_bound(
-        std::make_pair(start_offset, static_cast<size_t>(0)));
+        std::make_pair(offset, static_cast<size_t>(0)));
     if (it == marked_ranges_.end()) {
       return false;
     }
-    return start_offset <= it->first && end_offset >= it->second;
+    return offset <= it->first && offset >= it->second;
   }
 
  private:
@@ -245,6 +244,12 @@ class BlockReadAmpBitmapSlowAndAccurate {
 };
 
 TEST_F(BlockTest, BlockReadAmpBitmap) {
+  uint32_t pin_offset = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+    "BlockReadAmpBitmap:rnd", [&pin_offset](void* arg) {
+      pin_offset = *(static_cast<uint32_t*>(arg));
+    });
+  SyncPoint::GetInstance()->EnableProcessing();
   std::vector<size_t> block_sizes = {
       1,                 // 1 byte
       32,                // 32 bytes
@@ -277,10 +282,8 @@ TEST_F(BlockTest, BlockReadAmpBitmap) {
     if (needed_bits % 32 != 0) {
       bitmap_size++;
     }
-    size_t bits_in_bitmap = bitmap_size * 32;
 
-    ASSERT_EQ(stats->getTickerCount(READ_AMP_TOTAL_READ_BYTES),
-              needed_bits * kBytesPerBit);
+    ASSERT_EQ(stats->getTickerCount(READ_AMP_TOTAL_READ_BYTES), block_size);
 
     // Generate some random entries
     std::vector<size_t> random_entry_offsets;
@@ -314,20 +317,18 @@ TEST_F(BlockTest, BlockReadAmpBitmap) {
                                       current_entry.second);
 
       size_t total_bits = 0;
-      for (size_t bit_idx = 0; bit_idx < bits_in_bitmap; bit_idx++) {
-        size_t start_rng = bit_idx * kBytesPerBit;
-        size_t end_rng = (start_rng + kBytesPerBit) - 1;
-
-        total_bits +=
-            read_amp_slow_and_accurate.IsAnyInRangeMarked(start_rng, end_rng);
+      for (size_t bit_idx = 0; bit_idx < needed_bits; bit_idx++) {
+        total_bits += read_amp_slow_and_accurate.IsPinMarked(
+          bit_idx * kBytesPerBit + pin_offset);
       }
       size_t expected_estimate_useful = total_bits * kBytesPerBit;
       size_t got_estimate_useful =
-          stats->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES);
-
+        stats->getTickerCount(READ_AMP_ESTIMATE_USEFUL_BYTES);
       ASSERT_EQ(expected_estimate_useful, got_estimate_useful);
     }
   }
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
 }
 
 TEST_F(BlockTest, BlockWithReadAmpBitmap) {

table/cuckoo_table_reader.cc

@@ -364,7 +364,8 @@ extern InternalIterator* NewErrorInternalIterator(const Status& status,
                                                   Arena* arena);
 
 InternalIterator* CuckooTableReader::NewIterator(
-    const ReadOptions& read_options, Arena* arena, bool skip_filters) {
+    const ReadOptions& read_options, Arena* arena,
+    const InternalKeyComparator* icomp, bool skip_filters) {
   if (!status().ok()) {
     return NewErrorInternalIterator(
         Status::Corruption("CuckooTableReader status is not okay."), arena);

table/cuckoo_table_reader.h

@@ -45,8 +45,10 @@ class CuckooTableReader: public TableReader {
   Status Get(const ReadOptions& read_options, const Slice& key,
              GetContext* get_context, bool skip_filters = false) override;
 
-  InternalIterator* NewIterator(const ReadOptions&, Arena* arena = nullptr,
-                                bool skip_filters = false) override;
+  InternalIterator* NewIterator(
+      const ReadOptions&, Arena* arena = nullptr,
+      const InternalKeyComparator* icomparator = nullptr,
+      bool skip_filters = false) override;
   void Prepare(const Slice& target) override;
 
   // Report an approximation of how much memory has been used.

table/get_context.cc

@@ -180,7 +180,6 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
 void replayGetContextLog(const Slice& replay_log, const Slice& user_key,
                          GetContext* get_context) {
 #ifndef ROCKSDB_LITE
-  static Cleanable nonToClean;
   Slice s = replay_log;
   while (s.size()) {
     auto type = static_cast<ValueType>(*s.data());
@@ -193,8 +192,7 @@ void replayGetContextLog(const Slice& replay_log, const Slice& user_key,
     // Since SequenceNumber is not stored and unknown, we will use
     // kMaxSequenceNumber.
     get_context->SaveValue(
-        ParsedInternalKey(user_key, kMaxSequenceNumber, type), value,
-        &nonToClean);
+        ParsedInternalKey(user_key, kMaxSequenceNumber, type), value, nullptr);
   }
 #else   // ROCKSDB_LITE
   assert(false);

table/mock_table.cc

@@ -28,7 +28,9 @@ stl_wrappers::KVMap MakeMockFile(
   return stl_wrappers::KVMap(l, stl_wrappers::LessOfComparator(&icmp_));
 }
 
-InternalIterator* MockTableReader::NewIterator(const ReadOptions&, Arena* arena,
+InternalIterator* MockTableReader::NewIterator(const ReadOptions&,
+                                               Arena* arena,
+                                               const InternalKeyComparator*,
                                                bool skip_filters) {
   return new MockTableIterator(table_);
 }

table/mock_table.h

@@ -40,7 +40,9 @@ class MockTableReader : public TableReader {
  public:
   explicit MockTableReader(const stl_wrappers::KVMap& table) : table_(table) {}
 
-  InternalIterator* NewIterator(const ReadOptions&, Arena* arena,
+  InternalIterator* NewIterator(const ReadOptions&,
+                                Arena* arena,
+                                const InternalKeyComparator* = nullptr,
                                 bool skip_filters = false) override;
 
   Status Get(const ReadOptions&, const Slice& key, GetContext* get_context,

table/plain_table_reader.cc

@@ -191,6 +191,7 @@ void PlainTableReader::SetupForCompaction() {
 
 InternalIterator* PlainTableReader::NewIterator(const ReadOptions& options,
                                                 Arena* arena,
+                                                const InternalKeyComparator*,
                                                 bool skip_filters) {
   if (options.total_order_seek && !IsTotalOrderMode()) {
     return NewErrorInternalIterator(

table/plain_table_reader.h

@@ -71,14 +71,16 @@ class PlainTableReader: public TableReader {
  public:
   static Status Open(const ImmutableCFOptions& ioptions,
                      const EnvOptions& env_options,
-                     const InternalKeyComparator& internal_comparator,
+                     const InternalKeyComparator& icomparator,
                      unique_ptr<RandomAccessFileReader>&& file,
                      uint64_t file_size, unique_ptr<TableReader>* table,
                      const int bloom_bits_per_key, double hash_table_ratio,
                      size_t index_sparseness, size_t huge_page_tlb_size,
                      bool full_scan_mode);
 
-  InternalIterator* NewIterator(const ReadOptions&, Arena* arena = nullptr,
+  InternalIterator* NewIterator(const ReadOptions&,
+                                Arena* arena = nullptr,
+                                const InternalKeyComparator* = nullptr,
                                 bool skip_filters = false) override;
 
   void Prepare(const Slice& target) override;

table/table_reader.h

@@ -40,6 +40,7 @@ class TableReader {
   //               option is effective only for block-based table format.
   virtual InternalIterator* NewIterator(const ReadOptions&,
                                         Arena* arena = nullptr,
+                                        const InternalKeyComparator* = nullptr,
                                         bool skip_filters = false) = 0;
 
   virtual InternalIterator* NewRangeTombstoneIterator(

table/two_level_iterator.cc

@@ -178,13 +178,13 @@ void TwoLevelIterator::Prev() {
   SkipEmptyDataBlocksBackward();
 }
 
-
 void TwoLevelIterator::SkipEmptyDataBlocksForward() {
   while (second_level_iter_.iter() == nullptr ||
          (!second_level_iter_.Valid() &&
-         !second_level_iter_.status().IsIncomplete())) {
+          !second_level_iter_.status().IsIncomplete())) {
     // Move to next block
-    if (!first_level_iter_.Valid()) {
+    if (!first_level_iter_.Valid() ||
+        state_->KeyReachedUpperBound(first_level_iter_.key())) {
       SetSecondLevelIterator(nullptr);
       return;
     }
@@ -199,7 +199,7 @@ void TwoLevelIterator::SkipEmptyDataBlocksForward() {
 void TwoLevelIterator::SkipEmptyDataBlocksBackward() {
   while (second_level_iter_.iter() == nullptr ||
          (!second_level_iter_.Valid() &&
-         !second_level_iter_.status().IsIncomplete())) {
+          !second_level_iter_.status().IsIncomplete())) {
     // Move to next block
     if (!first_level_iter_.Valid()) {
       SetSecondLevelIterator(nullptr);

table/two_level_iterator.h

@@ -25,6 +25,7 @@ struct TwoLevelIteratorState {
   virtual ~TwoLevelIteratorState() {}
   virtual InternalIterator* NewSecondaryIterator(const Slice& handle) = 0;
   virtual bool PrefixMayMatch(const Slice& internal_key) = 0;
+  virtual bool KeyReachedUpperBound(const Slice& internal_key) = 0;
 
   // If call PrefixMayMatch()
   bool check_prefix_may_match;

tools/db_bench_tool.cc

@@ -176,6 +176,7 @@ DEFINE_string(
     "Meta operations:\n"
     "\tcompact     -- Compact the entire DB\n"
     "\tstats       -- Print DB stats\n"
+    "\tresetstats  -- Reset DB stats\n"
     "\tlevelstats  -- Print the number of files and bytes per level\n"
     "\tsstables    -- Print sstable info\n"
     "\theapprofile -- Dump a heap profile (if supported by this"
@@ -580,6 +581,10 @@ DEFINE_bool(optimistic_transaction_db, false,
             "Open a OptimisticTransactionDB instance. "
             "Required for randomtransaction benchmark.");
 
+DEFINE_bool(use_blob_db, false,
+            "Open a BlobDB instance. "
+            "Required for largevalue benchmark.");
+
 DEFINE_bool(transaction_db, false,
             "Open a TransactionDB instance. "
             "Required for randomtransaction benchmark.");
@@ -625,8 +630,6 @@ DEFINE_bool(report_bg_io_stats, false,
 DEFINE_bool(use_stderr_info_logger, false,
             "Write info logs to stderr instead of to LOG file. ");
 
-DEFINE_bool(use_blob_db, false, "Whether to use BlobDB. ");
-
 static enum rocksdb::CompressionType StringToCompressionType(const char* ctype) {
   assert(ctype);
 
@@ -800,17 +803,18 @@ DEFINE_uint64(wal_size_limit_MB, 0, "Set the size limit for the WAL Files"
               " in MB.");
 DEFINE_uint64(max_total_wal_size, 0, "Set total max WAL size");
 
-DEFINE_bool(mmap_read, rocksdb::EnvOptions().use_mmap_reads,
+DEFINE_bool(mmap_read, rocksdb::Options().allow_mmap_reads,
             "Allow reads to occur via mmap-ing files");
 
-DEFINE_bool(mmap_write, rocksdb::EnvOptions().use_mmap_writes,
+DEFINE_bool(mmap_write, rocksdb::Options().allow_mmap_writes,
             "Allow writes to occur via mmap-ing files");
 
-DEFINE_bool(use_direct_reads, rocksdb::EnvOptions().use_direct_reads,
+DEFINE_bool(use_direct_reads, rocksdb::Options().use_direct_reads,
             "Use O_DIRECT for reading data");
 
-DEFINE_bool(use_direct_writes, rocksdb::EnvOptions().use_direct_writes,
-            "Use O_DIRECT for writing data");
+DEFINE_bool(use_direct_io_for_flush_and_compaction,
+            rocksdb::Options().use_direct_io_for_flush_and_compaction,
+            "Use O_DIRECT for background flush and compaction I/O");
 
 DEFINE_bool(advise_random_on_open, rocksdb::Options().advise_random_on_open,
             "Advise random access on table file open");
@@ -1122,6 +1126,15 @@ class RandomGenerator {
     pos_ += len;
     return Slice(data_.data() + pos_ - len, len);
   }
+
+  Slice GenerateWithTTL(unsigned int len) {
+    assert(len <= data_.size());
+    if (pos_ + len > data_.size()) {
+      pos_ = 0;
+    }
+    pos_ += len;
+    return Slice(data_.data() + pos_ - len, len);
+  }
 };
 
 static void AppendWithSpace(std::string* str, Slice msg) {
@@ -2429,6 +2442,8 @@ void VerifyDBFromDB(std::string& truth_db_name) {
         method = &Benchmark::TimeSeries;
       } else if (name == "stats") {
         PrintStats("rocksdb.stats");
+      } else if (name == "resetstats") {
+        ResetStats();
       } else if (name == "verify") {
         VerifyDBFromDB(FLAGS_truth_db);
       } else if (name == "levelstats") {
@@ -2813,7 +2828,8 @@ void VerifyDBFromDB(std::string& truth_db_name) {
     options.allow_mmap_reads = FLAGS_mmap_read;
     options.allow_mmap_writes = FLAGS_mmap_write;
     options.use_direct_reads = FLAGS_use_direct_reads;
-    options.use_direct_writes = FLAGS_use_direct_writes;
+    options.use_direct_io_for_flush_and_compaction =
+        FLAGS_use_direct_io_for_flush_and_compaction;
 #ifndef ROCKSDB_LITE
     options.compaction_options_fifo = CompactionOptionsFIFO(
        FLAGS_fifo_compaction_max_table_files_size_mb * 1024 * 1024);
@@ -3217,9 +3233,14 @@ void VerifyDBFromDB(std::string& truth_db_name) {
       if (s.ok()) {
         db->db = ptr;
       }
-#endif  // ROCKSDB_LITE
     } else if (FLAGS_use_blob_db) {
-      s = NewBlobDB(options, db_name, &db->db);
+      blob_db::BlobDBOptions blob_db_options;
+      blob_db::BlobDB* ptr;
+      s = blob_db::BlobDB::Open(options, blob_db_options, db_name, &ptr);
+      if (s.ok()) {
+        db->db = ptr;
+      }
+#endif  // ROCKSDB_LITE
     } else {
       s = DB::Open(options, db_name, &db->db);
     }
@@ -3396,8 +3417,12 @@ void VerifyDBFromDB(std::string& truth_db_name) {
         int64_t rand_num = key_gens[id]->Next();
         GenerateKeyFromInt(rand_num, FLAGS_num, &key);
         if (FLAGS_use_blob_db) {
-          s = db_with_cfh->db->Put(write_options_, key,
-                                   gen.Generate(value_size_));
+          Slice val = gen.Generate(value_size_);
+          int ttl = rand() % 86400;
+          blob_db::BlobDB* blobdb =
+              static_cast<blob_db::BlobDB*>(db_with_cfh->db);
+          s = blobdb->PutWithTTL(write_options_, key, val, ttl);
+
         } else if (FLAGS_num_column_families <= 1) {
           batch.Put(key, gen.Generate(value_size_));
         } else {
@@ -5010,6 +5035,15 @@ void VerifyDBFromDB(std::string& truth_db_name) {
     db->CompactRange(CompactRangeOptions(), nullptr, nullptr);
   }
 
+  void ResetStats() {
+    if (db_.db != nullptr) {
+      db_.db->ResetStats();
+    }
+    for (const auto& db_with_cfh : multi_dbs_) {
+      db_with_cfh.db->ResetStats();
+    }
+  }
+
   void PrintStats(const char* key) {
     if (db_.db != nullptr) {
       PrintStats(db_.db, key, false);

tools/db_bench_tool_test.cc

@@ -211,7 +211,7 @@ const std::string options_file_content = R"OPTIONS_FILE(
   allow_mmap_reads=false
   allow_mmap_writes=false
   use_direct_reads=false
-  use_direct_writes=false
+  use_direct_io_for_flush_and_compaction=false
   stats_dump_period_sec=600
   allow_fallocate=true
   max_log_file_size=83886080

tools/db_stress.cc

@@ -269,16 +269,17 @@ DEFINE_string(db, "", "Use the db with the following name.");
 DEFINE_bool(verify_checksum, false,
             "Verify checksum for every block read from storage");
 
-DEFINE_bool(mmap_read, rocksdb::EnvOptions().use_mmap_reads,
+DEFINE_bool(mmap_read, rocksdb::Options().allow_mmap_reads,
             "Allow reads to occur via mmap-ing files");
 
-DEFINE_bool(mmap_write, rocksdb::EnvOptions().use_mmap_writes,
+DEFINE_bool(mmap_write, rocksdb::Options().allow_mmap_writes,
             "Allow writes to occur via mmap-ing files");
 
-DEFINE_bool(use_direct_reads, rocksdb::EnvOptions().use_direct_reads,
+DEFINE_bool(use_direct_reads, rocksdb::Options().use_direct_reads,
             "Use O_DIRECT for reading data");
 
-DEFINE_bool(use_direct_writes, rocksdb::EnvOptions().use_direct_writes,
+DEFINE_bool(use_direct_io_for_flush_and_compaction,
+            rocksdb::Options().use_direct_io_for_flush_and_compaction,
             "Use O_DIRECT for writing data");
 
 // Database statistics
@@ -2156,7 +2157,8 @@ class StressTest {
     options_.allow_mmap_reads = FLAGS_mmap_read;
     options_.allow_mmap_writes = FLAGS_mmap_write;
     options_.use_direct_reads = FLAGS_use_direct_reads;
-    options_.use_direct_writes = FLAGS_use_direct_writes;
+    options_.use_direct_io_for_flush_and_compaction =
+        FLAGS_use_direct_io_for_flush_and_compaction;
     options_.target_file_size_base = FLAGS_target_file_size_base;
     options_.target_file_size_multiplier = FLAGS_target_file_size_multiplier;
     options_.max_bytes_for_level_base = FLAGS_max_bytes_for_level_base;

util/coding.h

@@ -48,6 +48,7 @@ extern void PutLengthPrefixedSliceParts(std::string* dst,
 // Standard Get... routines parse a value from the beginning of a Slice
 // and advance the slice past the parsed value.
 extern bool GetFixed64(Slice* input, uint64_t* value);
+extern bool GetFixed32(Slice* input, uint32_t* value);
 extern bool GetVarint32(Slice* input, uint32_t* value);
 extern bool GetVarint64(Slice* input, uint64_t* value);
 extern bool GetLengthPrefixedSlice(Slice* input, Slice* result);
@@ -271,6 +272,15 @@ inline bool GetFixed64(Slice* input, uint64_t* value) {
   return true;
 }
 
+inline bool GetFixed32(Slice* input, uint32_t* value) {
+  if (input->size() < sizeof(uint32_t)) {
+    return false;
+  }
+  *value = DecodeFixed32(input->data());
+  input->remove_prefix(sizeof(uint32_t));
+  return true;
+}
+
 inline bool GetVarint32(Slice* input, uint32_t* value) {
   const char* p = input->data();
   const char* limit = p + input->size();

util/comparator.cc

@@ -51,7 +51,7 @@ class BytewiseComparatorImpl : public Comparator {
     } else {
       uint8_t start_byte = static_cast<uint8_t>((*start)[diff_index]);
       uint8_t limit_byte = static_cast<uint8_t>(limit[diff_index]);
-      if (start_byte >= limit_byte || (diff_index == start->size() - 1)) {
+      if (start_byte >= limit_byte) {
         // Cannot shorten since limit is smaller than start or start is
         // already the shortest possible.
         return;

util/file_reader_writer.cc

@@ -21,6 +21,16 @@
 
 namespace rocksdb {
 
+#ifndef NDEBUG
+namespace {
+
+bool IsSectorAligned(const size_t off, size_t sector_size) {
+  return off % sector_size == 0;
+}
+
+}
+#endif
+
 Status SequentialFileReader::Read(size_t n, Slice* result, char* scratch) {
   Status s;
   if (use_direct_io()) {
@@ -472,7 +482,7 @@ class ReadaheadRandomAccessFile : public RandomAccessFile {
     // complitely or partially in the buffer
     // If it's completely cached, including end of file case when offset + n is
     // greater than EOF, return
-    if (TryReadFromCache_(offset, n, &cached_len, scratch) &&
+    if (TryReadFromCache(offset, n, &cached_len, scratch) &&
         (cached_len == n ||
          // End of file
          buffer_len_ < readahead_size_)) {
@@ -484,34 +494,35 @@ class ReadaheadRandomAccessFile : public RandomAccessFile {
     // chunk_offset equals to advanced_offset
     size_t chunk_offset = TruncateToPageBoundary(alignment_, advanced_offset);
     Slice readahead_result;
-    Status s = file_->Read(chunk_offset, readahead_size_, &readahead_result,
-                           buffer_.BufferStart());
-    if (!s.ok()) {
-      return s;
-    }
-    // In the case of cache miss, i.e. when cached_len equals 0, an offset can
-    // exceed the file end position, so the following check is required
-    if (advanced_offset < chunk_offset + readahead_result.size()) {
-      // In the case of cache miss, the first chunk_padding bytes in buffer_ are
-      // stored for alignment only and must be skipped
-      size_t chunk_padding = advanced_offset - chunk_offset;
-      auto remaining_len =
-          std::min(readahead_result.size() - chunk_padding, n - cached_len);
-      memcpy(scratch + cached_len, readahead_result.data() + chunk_padding,
-             remaining_len);
-      *result = Slice(scratch, cached_len + remaining_len);
-    } else {
-      *result = Slice(scratch, cached_len);
-    }
 
-    if (readahead_result.data() == buffer_.BufferStart()) {
-      buffer_offset_ = chunk_offset;
-      buffer_len_ = readahead_result.size();
-    } else {
-      buffer_len_ = 0;
+    Status s = ReadIntoBuffer(chunk_offset, readahead_size_);
+    if (s.ok()) {
+      // In the case of cache miss, i.e. when cached_len equals 0, an offset can
+      // exceed the file end position, so the following check is required
+      if (advanced_offset < chunk_offset + buffer_len_) {
+        // In the case of cache miss, the first chunk_padding bytes in buffer_
+        // are
+        // stored for alignment only and must be skipped
+        size_t chunk_padding = advanced_offset - chunk_offset;
+        auto remaining_len =
+            std::min(buffer_len_ - chunk_padding, n - cached_len);
+        memcpy(scratch + cached_len, buffer_.BufferStart() + chunk_padding,
+               remaining_len);
+        *result = Slice(scratch, cached_len + remaining_len);
+      } else {
+        *result = Slice(scratch, cached_len);
+      }
     }
+    return s;
+  }
 
-    return Status::OK();
+  virtual Status Prefetch(uint64_t offset, size_t n) override {
+    size_t prefetch_offset = TruncateToPageBoundary(alignment_, offset);
+    if (prefetch_offset == buffer_offset_) {
+      return Status::OK();
+    }
+    return ReadIntoBuffer(prefetch_offset,
+                          Roundup(offset + n, alignment_) - prefetch_offset);
   }
 
   virtual size_t GetUniqueId(char* id, size_t max_size) const override {
@@ -529,7 +540,7 @@ class ReadaheadRandomAccessFile : public RandomAccessFile {
   }
 
  private:
-  bool TryReadFromCache_(uint64_t offset, size_t n, size_t* cached_len,
+  bool TryReadFromCache(uint64_t offset, size_t n, size_t* cached_len,
                          char* scratch) const {
     if (offset < buffer_offset_ || offset >= buffer_offset_ + buffer_len_) {
       *cached_len = 0;
@@ -542,15 +553,30 @@ class ReadaheadRandomAccessFile : public RandomAccessFile {
     return true;
   }
 
+  Status ReadIntoBuffer(uint64_t offset, size_t n) const {
+    if (n > buffer_.Capacity()) {
+      n = buffer_.Capacity();
+    }
+    assert(IsSectorAligned(offset, alignment_));
+    assert(IsSectorAligned(n, alignment_));
+    Slice result;
+    Status s = file_->Read(offset, n, &result, buffer_.BufferStart());
+    if (s.ok()) {
+      buffer_offset_ = offset;
+      buffer_len_ = result.size();
+    }
+    return s;
+  }
+
   std::unique_ptr<RandomAccessFile> file_;
   const size_t alignment_;
   size_t               readahead_size_;
   const bool           forward_calls_;
 
-  mutable std::mutex   lock_;
+  mutable std::mutex lock_;
   mutable AlignedBuffer buffer_;
-  mutable uint64_t     buffer_offset_;
-  mutable size_t       buffer_len_;
+  mutable uint64_t buffer_offset_;
+  mutable size_t buffer_len_;
 };
 }  // namespace
 

util/file_reader_writer.h

@@ -46,6 +46,8 @@ class SequentialFileReader {
 
   Status Skip(uint64_t n);
 
+  void Rewind();
+
   SequentialFile* file() { return file_.get(); }
 
   bool use_direct_io() const { return file_->use_direct_io(); }
@@ -92,6 +94,10 @@ class RandomAccessFileReader {
 
   Status Read(uint64_t offset, size_t n, Slice* result, char* scratch) const;
 
+  Status Prefetch(uint64_t offset, size_t n) const {
+    return file_->Prefetch(offset, n);
+  }
+
   RandomAccessFile* file() { return file_.get(); }
 
   bool use_direct_io() const { return file_->use_direct_io(); }
@@ -134,7 +140,9 @@ class WritableFileWriter {
         rate_limiter_(options.rate_limiter),
         stats_(stats) {
     buf_.Alignment(writable_file_->GetRequiredBufferAlignment());
-    buf_.AllocateNewBuffer(std::min((size_t)65536, max_buffer_size_));
+    buf_.AllocateNewBuffer(use_direct_io()
+                               ? max_buffer_size_
+                               : std::min((size_t)65536, max_buffer_size_));
   }
 
   WritableFileWriter(const WritableFileWriter&) = delete;

util/filename.cc

@@ -27,6 +27,7 @@ namespace rocksdb {
 
 static const std::string kRocksDbTFileExt = "sst";
 static const std::string kLevelDbTFileExt = "ldb";
+static const std::string kRocksDBBlobFileExt = "blob";
 
 // Given a path, flatten the path name by replacing all chars not in
 // {[0-9,a-z,A-Z,-,_,.]} with _. And append '_LOG\0' at the end.
@@ -74,6 +75,11 @@ std::string LogFileName(const std::string& name, uint64_t number) {
   return MakeFileName(name, number, "log");
 }
 
+std::string BlobFileName(const std::string& blobdirname, uint64_t number) {
+  assert(number > 0);
+  return MakeFileName(blobdirname, number, kRocksDBBlobFileExt.c_str());
+}
+
 std::string ArchivalDirectory(const std::string& dir) {
   return dir + "/" + ARCHIVAL_DIR;
 }
@@ -220,7 +226,7 @@ std::string IdentityFileName(const std::string& dbname) {
 //    dbname/<info_log_name_prefix>
 //    dbname/<info_log_name_prefix>.old.[0-9]+
 //    dbname/MANIFEST-[0-9]+
-//    dbname/[0-9]+.(log|sst)
+//    dbname/[0-9]+.(log|sst|blob)
 //    dbname/METADB-[0-9]+
 //    dbname/OPTIONS-[0-9]+
 //    dbname/OPTIONS-[0-9]+.dbtmp
@@ -335,6 +341,8 @@ bool ParseFileName(const std::string& fname, uint64_t* number,
     } else if (suffix == Slice(kRocksDbTFileExt) ||
                suffix == Slice(kLevelDbTFileExt)) {
       *type = kTableFile;
+    } else if (suffix == Slice(kRocksDBBlobFileExt)) {
+      *type = kBlobFile;
     } else if (suffix == Slice(kTempFileNameSuffix)) {
       *type = kTempFile;
     } else {

util/filename.h

@@ -38,7 +38,8 @@ enum FileType {
   kInfoLogFile,  // Either the current one, or an old one
   kMetaDatabase,
   kIdentityFile,
-  kOptionsFile
+  kOptionsFile,
+  kBlobFile
 };
 
 // Return the name of the log file with the specified number
@@ -46,6 +47,8 @@ enum FileType {
 // "dbname".
 extern std::string LogFileName(const std::string& dbname, uint64_t number);
 
+extern std::string BlobFileName(const std::string& bdirname, uint64_t number);
+
 static const std::string ARCHIVAL_DIR = "archive";
 
 extern std::string ArchivalDirectory(const std::string& dbname);

util/mpsc.h

@@ -0,0 +1,158 @@
+//  Portions Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+//
+// Large parts of this file is borrowed from the public domain code below.
+// from https://github.com/mstump/queues
+
+// C++ implementation of Dmitry Vyukov's non-intrusive
+// lock free unbound MPSC queue
+// http://www.1024cores.net/home/
+// lock-free-algorithms/queues/non-intrusive-mpsc-node-based-queue
+
+// License from mstump/queues
+// This is free and unencumbered software released into the public domain.
+//
+// Anyone is free to copy, modify, publish, use, compile, sell, or
+// distribute this software, either in source code form or as a compiled
+// binary, for any purpose, commercial or non-commercial, and by any
+// means.
+//
+// In jurisdictions that recognize copyright laws, the author or authors
+// of this software dedicate any and all copyright interest in the
+// software to the public domain. We make this dedication for the benefit
+// of the public at large and to the detriment of our heirs and
+// successors. We intend this dedication to be an overt act of
+// relinquishment in perpetuity of all present and future rights to this
+// software under copyright law.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+// IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+// OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+// ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+// OTHER DEALINGS IN THE SOFTWARE.
+//
+// For more information, please refer to <http://unlicense.org>
+
+// License from http://www.1024cores.net/home/
+// lock-free-algorithms/queues/non-intrusive-mpsc-node-based-queue
+// Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
+// Redistribution and use in source and binary forms, with or
+// without modification, are permitted provided that the following
+// conditions are met:
+// 1. Redistributions of source code must retain the above copyright notice,
+// this list of conditions and the following disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+// this list of conditions and the following disclaimer in the documentation
+// and/or other materials provided with the distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR
+// IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+// OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
+// EVENT SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+// INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+// BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
+// USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+// The views and conclusions contained in the software and documentation
+// are those of the authors and should not be interpreted as representing
+// official policies, either expressed or implied, of Dmitry Vyukov.
+//
+
+#ifndef UTIL_MPSC_H_
+#define UTIL_MPSC_H_
+
+#include <atomic>
+#include <cassert>
+#include <type_traits>
+
+/**
+ * Multiple Producer Single Consumer Lockless Q
+ */
+template <typename T>
+class mpsc_queue_t {
+ public:
+  struct buffer_node_t {
+    T data;
+    std::atomic<buffer_node_t*> next;
+  };
+
+  mpsc_queue_t() {
+    buffer_node_aligned_t* al_st = new buffer_node_aligned_t;
+    buffer_node_t* node = new (al_st) buffer_node_t();
+    _head.store(node);
+    _tail.store(node);
+
+    node->next.store(nullptr, std::memory_order_relaxed);
+  }
+
+  ~mpsc_queue_t() {
+    T output;
+    while (this->dequeue(&output)) {
+    }
+    buffer_node_t* front = _head.load(std::memory_order_relaxed);
+    front->~buffer_node_t();
+
+    ::operator delete(front);
+  }
+
+  void enqueue(const T& input) {
+    buffer_node_aligned_t* al_st = new buffer_node_aligned_t;
+    buffer_node_t* node = new (al_st) buffer_node_t();
+
+    node->data = input;
+    node->next.store(nullptr, std::memory_order_relaxed);
+
+    buffer_node_t* prev_head = _head.exchange(node, std::memory_order_acq_rel);
+    prev_head->next.store(node, std::memory_order_release);
+  }
+
+  bool dequeue(T* output) {
+    buffer_node_t* tail = _tail.load(std::memory_order_relaxed);
+    buffer_node_t* next = tail->next.load(std::memory_order_acquire);
+
+    if (next == nullptr) {
+      return false;
+    }
+
+    *output = next->data;
+    _tail.store(next, std::memory_order_release);
+
+    tail->~buffer_node_t();
+
+    ::operator delete(tail);
+    return true;
+  }
+
+  // you can only use pop_all if the queue is SPSC
+  buffer_node_t* pop_all() {
+    // nobody else can move the tail pointer.
+    buffer_node_t* tptr = _tail.load(std::memory_order_relaxed);
+    buffer_node_t* next =
+        tptr->next.exchange(nullptr, std::memory_order_acquire);
+    _head.exchange(tptr, std::memory_order_acquire);
+
+    // there is a race condition here
+    return next;
+  }
+
+ private:
+  typedef typename std::aligned_storage<
+      sizeof(buffer_node_t), std::alignment_of<buffer_node_t>::value>::type
+      buffer_node_aligned_t;
+
+  std::atomic<buffer_node_t*> _head;
+  std::atomic<buffer_node_t*> _tail;
+
+  mpsc_queue_t(const mpsc_queue_t&) = delete;
+  mpsc_queue_t& operator=(const mpsc_queue_t&) = delete;
+};
+
+#endif  // UTIL_MPSC_H_

util/testutil.cc

@@ -242,7 +242,7 @@ void RandomInitDBOptions(DBOptions* db_opt, Random* rnd) {
   db_opt->allow_mmap_reads = rnd->Uniform(2);
   db_opt->allow_mmap_writes = rnd->Uniform(2);
   db_opt->use_direct_reads = rnd->Uniform(2);
-  db_opt->use_direct_writes = rnd->Uniform(2);
+  db_opt->use_direct_io_for_flush_and_compaction = rnd->Uniform(2);
   db_opt->create_if_missing = rnd->Uniform(2);
   db_opt->create_missing_column_families = rnd->Uniform(2);
   db_opt->enable_thread_tracking = rnd->Uniform(2);

util/timer_queue.h

@@ -0,0 +1,217 @@
+//  Portions Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+//
+// Borrowed from
+// http://www.crazygaze.com/blog/2016/03/24/portable-c-timer-queue/
+// Timer Queue
+//
+// License
+//
+// The source code in this article is licensed under the CC0 license, so feel
+// free to copy, modify, share, do whatever you want with it.
+// No attribution is required, but Ill be happy if you do.
+// CC0 license
+
+// The person who associated a work with this deed has dedicated the work to the
+// public domain by waiving all of his or her rights to the work worldwide
+// under copyright law, including all related and neighboring rights, to the
+// extent allowed by law.  You can copy, modify, distribute and perform the
+// work, even for commercial purposes, all without asking permission.
+
+#pragma once
+#include <assert.h>
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <queue>
+#include <thread>
+#include <utility>
+#include <vector>
+
+// Allows execution of handlers at a specified time in the future
+// Guarantees:
+//  - All handlers are executed ONCE, even if cancelled (aborted parameter will
+// be set to true)
+//      - If TimerQueue is destroyed, it will cancel all handlers.
+//  - Handlers are ALWAYS executed in the Timer Queue worker thread.
+//  - Handlers execution order is NOT guaranteed
+//
+////////////////////////////////////////////////////////////////////////////////
+// borrowed from
+// http://www.crazygaze.com/blog/2016/03/24/portable-c-timer-queue/
+class TimerQueue {
+ public:
+  TimerQueue() : m_th(&TimerQueue::run, this) {}
+
+  ~TimerQueue() {
+    cancelAll();
+    // Abusing the timer queue to trigger the shutdown.
+    add(0, [this](bool) {
+      m_finish = true;
+      return std::make_pair(false, 0);
+    });
+    m_th.join();
+  }
+
+  // Adds a new timer
+  // \return
+  //  Returns the ID of the new timer. You can use this ID to cancel the
+  // timer
+  uint64_t add(int64_t milliseconds,
+               std::function<std::pair<bool, int64_t>(bool)> handler) {
+    WorkItem item;
+    Clock::time_point tp = Clock::now();
+    item.end = tp + std::chrono::milliseconds(milliseconds);
+    item.period = milliseconds;
+    item.handler = std::move(handler);
+
+    std::unique_lock<std::mutex> lk(m_mtx);
+    uint64_t id = ++m_idcounter;
+    item.id = id;
+    m_items.push(std::move(item));
+
+    // Something changed, so wake up timer thread
+    m_checkWork.notify_one();
+    return id;
+  }
+
+  // Cancels the specified timer
+  // \return
+  //  1 if the timer was cancelled.
+  //  0 if you were too late to cancel (or the timer ID was never valid to
+  // start with)
+  size_t cancel(uint64_t id) {
+    // Instead of removing the item from the container (thus breaking the
+    // heap integrity), we set the item as having no handler, and put
+    // that handler on a new item at the top for immediate execution
+    // The timer thread will then ignore the original item, since it has no
+    // handler.
+    std::unique_lock<std::mutex> lk(m_mtx);
+    for (auto&& item : m_items.getContainer()) {
+      if (item.id == id && item.handler) {
+        WorkItem newItem;
+        // Zero time, so it stays at the top for immediate execution
+        newItem.end = Clock::time_point();
+        newItem.id = 0;  // Means it is a canceled item
+        // Move the handler from item to newitem (thus clearing item)
+        newItem.handler = std::move(item.handler);
+        m_items.push(std::move(newItem));
+
+        // Something changed, so wake up timer thread
+        m_checkWork.notify_one();
+        return 1;
+      }
+    }
+    return 0;
+  }
+
+  // Cancels all timers
+  // \return
+  //  The number of timers cancelled
+  size_t cancelAll() {
+    // Setting all "end" to 0 (for immediate execution) is ok,
+    // since it maintains the heap integrity
+    std::unique_lock<std::mutex> lk(m_mtx);
+    m_cancel = true;
+    for (auto&& item : m_items.getContainer()) {
+      if (item.id && item.handler) {
+        item.end = Clock::time_point();
+        item.id = 0;
+      }
+    }
+    auto ret = m_items.size();
+
+    m_checkWork.notify_one();
+    return ret;
+  }
+
+ private:
+  using Clock = std::chrono::steady_clock;
+  TimerQueue(const TimerQueue&) = delete;
+  TimerQueue& operator=(const TimerQueue&) = delete;
+
+  void run() {
+    std::unique_lock<std::mutex> lk(m_mtx);
+    while (!m_finish) {
+      auto end = calcWaitTime_lock();
+      if (end.first) {
+        // Timers found, so wait until it expires (or something else
+        // changes)
+        m_checkWork.wait_until(lk, end.second);
+      } else {
+        // No timers exist, so wait forever until something changes
+        m_checkWork.wait(lk);
+      }
+
+      // Check and execute as much work as possible, such as, all expired
+      // timers
+      checkWork(&lk);
+    }
+
+    // If we are shutting down, we should not have any items left,
+    // since the shutdown cancels all items
+    assert(m_items.size() == 0);
+  }
+
+  std::pair<bool, Clock::time_point> calcWaitTime_lock() {
+    while (m_items.size()) {
+      if (m_items.top().handler) {
+        // Item present, so return the new wait time
+        return std::make_pair(true, m_items.top().end);
+      } else {
+        // Discard empty handlers (they were cancelled)
+        m_items.pop();
+      }
+    }
+
+    // No items found, so return no wait time (causes the thread to wait
+    // indefinitely)
+    return std::make_pair(false, Clock::time_point());
+  }
+
+  void checkWork(std::unique_lock<std::mutex>* lk) {
+    while (m_items.size() && m_items.top().end <= Clock::now()) {
+      WorkItem item(m_items.top());
+      m_items.pop();
+
+      if (item.handler) {
+        (*lk).unlock();
+        auto reschedule_pair = item.handler(item.id == 0);
+        (*lk).lock();
+        if (!m_cancel && reschedule_pair.first) {
+          int64_t new_period = (reschedule_pair.second == -1)
+                                   ? item.period
+                                   : reschedule_pair.second;
+
+          item.period = new_period;
+          item.end = Clock::now() + std::chrono::milliseconds(new_period);
+          m_items.push(std::move(item));
+        }
+      }
+    }
+  }
+
+  bool m_finish = false;
+  bool m_cancel = false;
+  uint64_t m_idcounter = 0;
+  std::condition_variable m_checkWork;
+
+  struct WorkItem {
+    Clock::time_point end;
+    int64_t period;
+    uint64_t id;  // id==0 means it was cancelled
+    std::function<std::pair<bool, int64_t>(bool)> handler;
+    bool operator>(const WorkItem& other) const { return end > other.end; }
+  };
+
+  std::mutex m_mtx;
+  // Inheriting from priority_queue, so we can access the internal container
+  class Queue : public std::priority_queue<WorkItem, std::vector<WorkItem>,
+                                           std::greater<WorkItem>> {
+   public:
+    std::vector<WorkItem>& getContainer() { return this->c; }
+  } m_items;
+  std::thread m_th;
+};

util/timer_queue_test.cc

@@ -0,0 +1,72 @@
+//  Portions Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+
+// borrowed from
+// http://www.crazygaze.com/blog/2016/03/24/portable-c-timer-queue/
+// Timer Queue
+//
+// License
+//
+// The source code in this article is licensed under the CC0 license, so feel
+// free
+// to copy, modify, share, do whatever you want with it.
+// No attribution is required, but Ill be happy if you do.
+// CC0 license
+
+// The person who associated a work with this deed has dedicated the work to the
+// public domain by waiving all of his or her rights to the work worldwide
+// under copyright law, including all related and neighboring rights, to the
+// extent allowed by law.  You can copy, modify, distribute and perform the
+// work, even for
+// commercial purposes, all without asking permission. See Other Information
+// below.
+//
+
+#include "util/timer_queue.h"
+#include <future>
+
+namespace Timing {
+
+using Clock = std::chrono::high_resolution_clock;
+double now() {
+  static auto start = Clock::now();
+  return std::chrono::duration<double, std::milli>(Clock::now() - start)
+      .count();
+}
+
+}  // namespace Timing
+
+int main() {
+  TimerQueue q;
+
+  double tnow = Timing::now();
+
+  q.add(10000, [tnow](bool aborted) mutable {
+    printf("T 1: %d, Elapsed %4.2fms\n", aborted, Timing::now() - tnow);
+    return std::make_pair(false, 0);
+  });
+  q.add(10001, [tnow](bool aborted) mutable {
+    printf("T 2: %d, Elapsed %4.2fms\n", aborted, Timing::now() - tnow);
+    return std::make_pair(false, 0);
+  });
+
+  q.add(1000, [tnow](bool aborted) mutable {
+    printf("T 3: %d, Elapsed %4.2fms\n", aborted, Timing::now() - tnow);
+    return std::make_pair(!aborted, 1000);
+  });
+
+  auto id = q.add(2000, [tnow](bool aborted) mutable {
+    printf("T 4: %d, Elapsed %4.2fms\n", aborted, Timing::now() - tnow);
+    return std::make_pair(!aborted, 2000);
+  });
+
+  (void)id;
+  // auto ret = q.cancel(id);
+  // assert(ret == 1);
+  // q.cancelAll();
+
+  return 0;
+}
+//////////////////////////////////////////

utilities/blob_db/blob_db.cc

@@ -1,12 +1,15 @@
-// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file. See the AUTHORS file for names of contributors.
-#include "utilities/blob_db/blob_db.h"
-
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+//
 #ifndef ROCKSDB_LITE
+
+#include "utilities/blob_db/blob_db.h"
 #include "db/write_batch_internal.h"
 #include "monitoring/instrumented_mutex.h"
 #include "options/cf_options.h"
+#include "rocksdb/compaction_filter.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/env.h"
 #include "rocksdb/iterator.h"
@@ -17,194 +20,152 @@
 #include "util/crc32c.h"
 #include "util/file_reader_writer.h"
 #include "util/filename.h"
+#include "utilities/blob_db/blob_db_impl.h"
 
 namespace rocksdb {
 
-namespace {
-int kBlockBasedTableVersionFormat = 2;
-}  // namespace
+namespace blob_db {
+port::Mutex listener_mutex;
+typedef std::shared_ptr<BlobDBFlushBeginListener> FlushBeginListener_t;
+typedef std::shared_ptr<BlobReconcileWalFilter> ReconcileWalFilter_t;
+typedef std::shared_ptr<EvictAllVersionsCompactionListener>
+    CompactionListener_t;
 
-class BlobDB : public StackableDB {
- public:
-  using rocksdb::StackableDB::Put;
-  Status Put(const WriteOptions& options, const Slice& key,
-             const Slice& value) override;
+// to ensure the lifetime of the listeners
+std::vector<std::shared_ptr<EventListener>> all_blobdb_listeners;
+std::vector<ReconcileWalFilter_t> all_wal_filters;
 
-  using rocksdb::StackableDB::Get;
-  Status Get(const ReadOptions& options, const Slice& key,
-             std::string* value) override;
+Status BlobDB::OpenAndLoad(const Options& options,
+                           const BlobDBOptions& bdb_options,
+                           const std::string& dbname, BlobDB** blob_db,
+                           Options* changed_options) {
+  *changed_options = options;
+  *blob_db = nullptr;
 
-  Status Open();
+  FlushBeginListener_t fblistener =
+      std::make_shared<BlobDBFlushBeginListener>();
+  ReconcileWalFilter_t rw_filter = std::make_shared<BlobReconcileWalFilter>();
+  CompactionListener_t ce_listener =
+      std::make_shared<EvictAllVersionsCompactionListener>();
 
-  explicit BlobDB(DB* db);
-
- private:
-  std::string dbname_;
-  ImmutableCFOptions ioptions_;
-  InstrumentedMutex mutex_;
-  std::unique_ptr<RandomAccessFileReader> file_reader_;
-  std::unique_ptr<WritableFileWriter> file_writer_;
-  size_t writer_offset_;
-  size_t next_sync_offset_;
-
-  static const std::string kFileName;
-  static const size_t kBlockHeaderSize;
-  static const size_t kBytesPerSync;
-};
-
-Status NewBlobDB(Options options, std::string dbname, DB** blob_db) {
-  DB* db;
-  Status s = DB::Open(options, dbname, &db);
-  if (!s.ok()) {
-    return s;
+  {
+    MutexLock l(&listener_mutex);
+    all_blobdb_listeners.push_back(fblistener);
+    all_blobdb_listeners.push_back(ce_listener);
+    all_wal_filters.push_back(rw_filter);
   }
-  BlobDB* bdb = new BlobDB(db);
-  s = bdb->Open();
+
+  changed_options->listeners.emplace_back(fblistener);
+  changed_options->listeners.emplace_back(ce_listener);
+  changed_options->wal_filter = rw_filter.get();
+
+  DBOptions db_options(*changed_options);
+
+  // we need to open blob db first so that recovery can happen
+  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;
+
+  *blob_db = bdb;
+  return s;
+}
+
+Status BlobDB::Open(const Options& options, const BlobDBOptions& bdb_options,
+                    const std::string& dbname, BlobDB** blob_db) {
+  *blob_db = nullptr;
+  DBOptions db_options(options);
+  ColumnFamilyOptions cf_options(options);
+  std::vector<ColumnFamilyDescriptor> column_families;
+  column_families.push_back(
+      ColumnFamilyDescriptor(kDefaultColumnFamilyName, cf_options));
+  std::vector<ColumnFamilyHandle*> handles;
+  Status s = BlobDB::Open(db_options, bdb_options, dbname, column_families,
+                          &handles, blob_db);
+  if (s.ok()) {
+    assert(handles.size() == 1);
+    // i can delete the handle since DBImpl is always holding a reference to
+    // default column family
+    delete handles[0];
+  }
+  return s;
+}
+
+Status BlobDB::Open(const DBOptions& db_options,
+                    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;
+
+  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();
+
+  {
+    MutexLock l(&listener_mutex);
+    all_blobdb_listeners.push_back(fblistener);
+    all_blobdb_listeners.push_back(ce_listener);
+    all_wal_filters.push_back(rw_filter);
+  }
+
+  // we need to open blob db first so that recovery can happen
+  BlobDBImpl* bdb = new BlobDBImpl(dbname, bdb_options, my_db_options);
+  fblistener->SetImplPtr(bdb);
+  ce_listener->SetImplPtr(bdb);
+  rw_filter->SetImplPtr(bdb);
+
+  Status s = bdb->OpenPhase1();
+  if (!s.ok()) 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;
+
+  // set the implementation pointer
+  s = bdb->LinkToBaseDB(db);
   if (!s.ok()) {
     delete bdb;
+    bdb = nullptr;
   }
   *blob_db = bdb;
   return s;
 }
 
-const std::string BlobDB::kFileName = "blob_log";
-const size_t BlobDB::kBlockHeaderSize = 8;
-const size_t BlobDB::kBytesPerSync = 1024 * 1024 * 128;
+BlobDB::BlobDB(DB* db) : StackableDB(db) {}
 
-BlobDB::BlobDB(DB* db)
-    : StackableDB(db),
-      ioptions_(db->GetOptions()),
-      writer_offset_(0),
-      next_sync_offset_(kBytesPerSync) {}
+////////////////////////////////////////////////////////////////////////////////
+//
+//
+// 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),
+      default_ttl_extractor(false),
+      compression(kNoCompression) {}
 
-Status BlobDB::Open() {
-  unique_ptr<WritableFile> wfile;
-  EnvOptions env_options(db_->GetOptions());
-  Status s = ioptions_.env->NewWritableFile(db_->GetName() + "/" + kFileName,
-                                            &wfile, env_options);
-  if (!s.ok()) {
-    return s;
-  }
-  file_writer_.reset(new WritableFileWriter(std::move(wfile), env_options));
-
-  // Write version
-  std::string version;
-  PutFixed64(&version, 0);
-  s = file_writer_->Append(Slice(version));
-  if (!s.ok()) {
-    return s;
-  }
-  writer_offset_ += version.size();
-
-  std::unique_ptr<RandomAccessFile> rfile;
-  s = ioptions_.env->NewRandomAccessFile(db_->GetName() + "/" + kFileName,
-                                         &rfile, env_options);
-  if (!s.ok()) {
-    return s;
-  }
-  file_reader_.reset(new RandomAccessFileReader(std::move(rfile)));
-  return s;
-}
-
-Status BlobDB::Put(const WriteOptions& options, const Slice& key,
-                   const Slice& value) {
-  BlockBuilder block_builder(1, false);
-  block_builder.Add(key, value);
-
-  CompressionType compression = CompressionType::kLZ4Compression;
-  CompressionOptions compression_opts;
-
-  Slice block_contents;
-  std::string compression_output;
-
-  block_contents = CompressBlock(block_builder.Finish(), compression_opts,
-                                 &compression, kBlockBasedTableVersionFormat,
-                                 Slice() /* dictionary */, &compression_output);
-
-  char header[kBlockHeaderSize];
-  char trailer[kBlockTrailerSize];
-  trailer[0] = compression;
-  auto crc = crc32c::Value(block_contents.data(), block_contents.size());
-  crc = crc32c::Extend(crc, trailer, 1);  // Extend to cover block type
-  EncodeFixed32(trailer + 1, crc32c::Mask(crc));
-
-  BlockHandle handle;
-  std::string index_entry;
-  Status s;
-  {
-    InstrumentedMutexLock l(&mutex_);
-    auto raw_block_size = block_contents.size();
-    EncodeFixed64(header, raw_block_size);
-    s = file_writer_->Append(Slice(header, kBlockHeaderSize));
-    writer_offset_ += kBlockHeaderSize;
-    if (s.ok()) {
-      handle.set_offset(writer_offset_);
-      handle.set_size(raw_block_size);
-      s = file_writer_->Append(block_contents);
-    }
-    if (s.ok()) {
-      s = file_writer_->Append(Slice(trailer, kBlockTrailerSize));
-    }
-    if (s.ok()) {
-      s = file_writer_->Flush();
-    }
-    if (s.ok() && writer_offset_ > next_sync_offset_) {
-      // Sync every kBytesPerSync. This is a hacky way to limit unsynced data.
-      next_sync_offset_ += kBytesPerSync;
-      s = file_writer_->Sync(db_->GetOptions().use_fsync);
-    }
-    if (s.ok()) {
-      writer_offset_ += block_contents.size() + kBlockTrailerSize;
-      // Put file number
-      PutVarint64(&index_entry, 0);
-      handle.EncodeTo(&index_entry);
-      s = db_->Put(options, key, index_entry);
-    }
-  }
-  return s;
-}
-
-Status BlobDB::Get(const ReadOptions& options, const Slice& key,
-                   std::string* value) {
-  Status s;
-  std::string index_entry;
-  s = db_->Get(options, key, &index_entry);
-  if (!s.ok()) {
-    return s;
-  }
-  BlockHandle handle;
-  Slice index_entry_slice(index_entry);
-  uint64_t file_number;
-  if (!GetVarint64(&index_entry_slice, &file_number)) {
-    return Status::Corruption();
-  }
-  assert(file_number == 0);
-  s = handle.DecodeFrom(&index_entry_slice);
-  if (!s.ok()) {
-    return s;
-  }
-  Footer footer(0, kBlockBasedTableVersionFormat);
-  BlockContents contents;
-  s = ReadBlockContents(file_reader_.get(), footer, options, handle, &contents,
-                        ioptions_);
-  if (!s.ok()) {
-    return s;
-  }
-  Block block(std::move(contents), kDisableGlobalSequenceNumber);
-  BlockIter bit;
-  InternalIterator* it = block.NewIterator(nullptr, &bit);
-  it->SeekToFirst();
-  if (!it->status().ok()) {
-    return it->status();
-  }
-  *value = it->value().ToString();
-  return s;
-}
+}  // namespace blob_db
 }  // namespace rocksdb
-#else
-namespace rocksdb {
-Status NewBlobDB(Options options, std::string dbname, DB** blob_db) {
-  return Status::NotSupported();
-}
-}  // namespace rocksdb
-#endif  // ROCKSDB_LITE
+#endif

utilities/blob_db/blob_db.h

@@ -5,12 +5,19 @@
 
 #pragma once
 
+#ifndef ROCKSDB_LITE
+
+#include <functional>
 #include <string>
+#include <vector>
 #include "rocksdb/db.h"
 #include "rocksdb/status.h"
+#include "rocksdb/utilities/stackable_db.h"
 
 namespace rocksdb {
-// EXPERIMENAL ONLY
+
+namespace blob_db {
+
 // A wrapped database which puts values of KV pairs in a separate log
 // and store location to the log in the underlying DB.
 // It lacks lots of importatant functionalities, e.g. DB restarts,
@@ -18,5 +25,177 @@ namespace rocksdb {
 //
 // The factory needs to be moved to include/rocksdb/utilities to allow
 // users to use blob DB.
-extern Status NewBlobDB(Options options, std::string dbname, DB** blob_db);
+
+struct BlobDBOptions {
+  // name of the directory under main db, where blobs will be stored.
+  // default is "blob_dir"
+  std::string blob_dir;
+
+  // whether the blob_dir path is relative or absolute.
+  bool path_relative;
+
+  // is the eviction strategy fifo based
+  bool is_fifo;
+
+  // 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;
+
+  // a new bucket is opened, for ttl_range. So if ttl_range is 600seconds
+  // (10 minutes), and the first bucket starts at 1471542000
+  // then the blob buckets will be
+  // 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;
+
+  // at what bytes will the blob files be synced to blob log.
+  uint64_t bytes_per_sync;
+
+  // the target size of each blob file. File will become immutable
+  // after it exceeds that size
+  uint64_t blob_file_size;
+
+  // how many files to use for simple blobs at one time
+  uint32_t num_concurrent_simple_blobs;
+
+  // this function is to be provided by client if they intend to
+  // use Put API to provide TTL.
+  // the first argument is the value in the Put API
+  // in case you want to do some modifications to the value,
+  // return a new Slice in the second.
+  // otherwise just copy the input value into output.
+  // the ttl should be extracted and returned in last pointer.
+  // otherwise assign it to -1
+  std::function<bool(const Slice&, Slice*, int32_t*)> extract_ttl_fn;
+
+  // eviction callback.
+  // this function will be called for every blob that is getting
+  // evicted.
+  std::function<void(const ColumnFamilyHandle*, const Slice&, const Slice&)>
+      gc_evict_cb_fn;
+
+  // default ttl extactor
+  bool default_ttl_extractor;
+
+  // what compression to use for Blob's
+  CompressionType compression;
+
+  // default constructor
+  BlobDBOptions();
+
+  BlobDBOptions(const BlobDBOptions& in) = default;
+
+  virtual ~BlobDBOptions() = default;
+};
+
+class BlobDB : public StackableDB {
+ public:
+  // the suffix to a blob value to represent "ttl:TTLVAL"
+  static const uint64_t kTTLSuffixLength = 8;
+
+ public:
+  using rocksdb::StackableDB::Put;
+
+  // This function needs to be called before destroying
+  // the base DB
+  static Status DestroyBlobDB(const std::string& dbname, const Options& options,
+                              const BlobDBOptions& bdb_options);
+
+  virtual Status Put(const WriteOptions& options,
+                     ColumnFamilyHandle* column_family, const Slice& key,
+                     const Slice& value) override = 0;
+
+  using rocksdb::StackableDB::Delete;
+  virtual Status Delete(const WriteOptions& options,
+                        ColumnFamilyHandle* column_family,
+                        const Slice& key) override = 0;
+
+  virtual Status PutWithTTL(const WriteOptions& options,
+                            ColumnFamilyHandle* column_family, const Slice& key,
+                            const Slice& value, int32_t ttl) = 0;
+
+  virtual Status PutWithTTL(const WriteOptions& options, const Slice& key,
+                            const Slice& value, int32_t ttl) {
+    return PutWithTTL(options, DefaultColumnFamily(), key, value, ttl);
+  }
+
+  virtual Status PutUntil(const WriteOptions& options,
+                          ColumnFamilyHandle* column_family, const Slice& key,
+                          const Slice& value, int32_t expiration) = 0;
+
+  virtual Status PutUntil(const WriteOptions& options, const Slice& key,
+                          const Slice& value, int32_t expiration) {
+    return PutUntil(options, DefaultColumnFamily(), key, value, expiration);
+  }
+
+  using rocksdb::StackableDB::Get;
+  virtual Status Get(const ReadOptions& options,
+                     ColumnFamilyHandle* column_family, const Slice& key,
+                     std::string* value) override = 0;
+
+  using rocksdb::StackableDB::MultiGet;
+  virtual std::vector<Status> MultiGet(
+      const ReadOptions& options,
+      const std::vector<ColumnFamilyHandle*>& column_family,
+      const std::vector<Slice>& keys,
+      std::vector<std::string>* values) override = 0;
+
+  using rocksdb::StackableDB::SingleDelete;
+  virtual Status SingleDelete(const WriteOptions& wopts,
+                              ColumnFamilyHandle* column_family,
+                              const Slice& key) override = 0;
+
+  using rocksdb::StackableDB::Merge;
+  virtual Status Merge(const WriteOptions& options,
+                       ColumnFamilyHandle* column_family, const Slice& key,
+                       const Slice& value) override {
+    return Status::NotSupported("Not supported operation in blob db.");
+  }
+
+  virtual Status Write(const WriteOptions& opts,
+                       WriteBatch* updates) override = 0;
+
+  // Starting point for opening a Blob DB.
+  // changed_options - critical. Blob DB loads and inserts listeners
+  // into options which are necessary for recovery and atomicity
+  // Use this pattern if you need control on step 2, i.e. your
+  // BaseDB is not just a simple rocksdb but a stacked DB
+  // 1. ::OpenAndLoad
+  // 2. Open Base DB with the changed_options
+  // 3. ::LinkToBaseDB
+  static Status OpenAndLoad(const Options& options,
+                            const BlobDBOptions& bdb_options,
+                            const std::string& dbname, BlobDB** blob_db,
+                            Options* changed_options);
+
+  // This is another way to open BLOB DB which do not have other
+  // Stackable DB's in play
+  // Steps.
+  // 1. ::Open
+  static Status Open(const Options& options, const BlobDBOptions& bdb_options,
+                     const std::string& dbname, BlobDB** blob_db);
+
+  static Status Open(const DBOptions& db_options,
+                     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 = false);
+
+  virtual ~BlobDB() {}
+
+  virtual Status LinkToBaseDB(DB* db_base) = 0;
+
+ protected:
+  explicit BlobDB(DB* db);
+};
+
+}  // namespace blob_db
 }  // namespace rocksdb
+#endif  // ROCKSDB_LITE

utilities/blob_db/blob_db_impl.h

@@ -0,0 +1,657 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+
+#pragma once
+
+#ifndef ROCKSDB_LITE
+
+#include <atomic>
+#include <condition_variable>
+#include <ctime>
+#include <list>
+#include <memory>
+#include <set>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "rocksdb/compaction_filter.h"
+#include "rocksdb/db.h"
+#include "rocksdb/listener.h"
+#include "rocksdb/options.h"
+#include "rocksdb/wal_filter.h"
+#include "util/file_reader_writer.h"
+#include "util/mpsc.h"
+#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"
+
+namespace rocksdb {
+
+class DBImpl;
+class ColumnFamilyHandle;
+class ColumnFamilyData;
+class OptimisticTransactionDBImpl;
+struct FlushJobInfo;
+
+namespace blob_db {
+
+class BlobFile;
+class BlobDBImpl;
+struct GCStats;
+
+class BlobDBFlushBeginListener : public EventListener {
+ public:
+  explicit BlobDBFlushBeginListener() : impl_(nullptr) {}
+
+  void OnFlushBegin(DB* db, const FlushJobInfo& info) override;
+
+  void SetImplPtr(BlobDBImpl* p) { impl_ = p; }
+
+ protected:
+  BlobDBImpl* impl_;
+};
+
+// this implements the callback from the WAL which ensures that the
+// blob record is present in the blob log. If fsync/fdatasync in not
+// happening on every write, there is the probability that keys in the
+// blob log can lag the keys in blobs
+class BlobReconcileWalFilter : public WalFilter {
+ public:
+  virtual WalFilter::WalProcessingOption LogRecordFound(
+      unsigned long long log_number, const std::string& log_file_name,
+      const WriteBatch& batch, WriteBatch* new_batch,
+      bool* batch_changed) override;
+
+  virtual const char* Name() const override { return "BlobDBWalReconciler"; }
+
+  void SetImplPtr(BlobDBImpl* p) { impl_ = p; }
+
+ protected:
+  BlobDBImpl* impl_;
+};
+
+class EvictAllVersionsCompactionListener : public EventListener {
+ public:
+  class InternalListener : public CompactionEventListener {
+    friend class BlobDBImpl;
+
+   public:
+    virtual void OnCompaction(int level, const Slice& key,
+                              CompactionListenerValueType value_type,
+                              const Slice& existing_value,
+                              const SequenceNumber& sn, bool is_new) override;
+
+    void SetImplPtr(BlobDBImpl* p) { impl_ = p; }
+
+   private:
+    BlobDBImpl* impl_;
+  };
+
+  explicit EvictAllVersionsCompactionListener()
+      : internal_listener_(new InternalListener()) {}
+
+  virtual CompactionEventListener* GetCompactionEventListener() override {
+    return internal_listener_.get();
+  }
+
+  void SetImplPtr(BlobDBImpl* p) { internal_listener_->SetImplPtr(p); }
+
+ private:
+  std::unique_ptr<InternalListener> internal_listener_;
+};
+
+#if 0
+class EvictAllVersionsFilterFactory : public CompactionFilterFactory {
+ private:
+  BlobDBImpl* impl_;
+
+ public:
+  EvictAllVersionsFilterFactory() : impl_(nullptr) {}
+
+  void SetImplPtr(BlobDBImpl* p) { impl_ = p; }
+
+  virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+      const CompactionFilter::Context& context) override;
+
+  virtual const char* Name() const override {
+    return "EvictAllVersionsFilterFactory";
+  }
+};
+#endif
+
+// Comparator to sort "TTL" aware Blob files based on the lower value of
+// TTL range.
+struct blobf_compare_ttl {
+  bool operator()(const std::shared_ptr<BlobFile>& lhs,
+                  const std::shared_ptr<BlobFile>& rhs) const;
+};
+
+/**
+ * The implementation class for BlobDB. This manages the value
+ * part in TTL aware sequentially written files. These files are
+ * Garbage Collected.
+ */
+class BlobDBImpl : public BlobDB {
+  friend class BlobDBFlushBeginListener;
+  friend class EvictAllVersionsCompactionListener;
+  friend class BlobDB;
+  friend class BlobFile;
+  friend class BlobDBIterator;
+
+ public:
+  using rocksdb::StackableDB::Put;
+  Status Put(const WriteOptions& options, ColumnFamilyHandle* column_family,
+             const Slice& key, const Slice& value) override;
+
+  using rocksdb::StackableDB::Delete;
+  Status Delete(const WriteOptions& options, ColumnFamilyHandle* column_family,
+                const Slice& key) override;
+
+  using rocksdb::StackableDB::SingleDelete;
+  virtual Status SingleDelete(const WriteOptions& wopts,
+                              ColumnFamilyHandle* column_family,
+                              const Slice& key) override;
+
+  using rocksdb::StackableDB::Get;
+  Status Get(const ReadOptions& options, ColumnFamilyHandle* column_family,
+             const Slice& key, std::string* value) override;
+
+  using rocksdb::StackableDB::NewIterator;
+  virtual Iterator* NewIterator(const ReadOptions& opts,
+                                ColumnFamilyHandle* column_family) override;
+
+  using rocksdb::StackableDB::MultiGet;
+  virtual std::vector<Status> MultiGet(
+      const ReadOptions& options,
+      const std::vector<ColumnFamilyHandle*>& column_family,
+      const std::vector<Slice>& keys,
+      std::vector<std::string>* values) override;
+
+  virtual Status Write(const WriteOptions& opts, WriteBatch* updates) override;
+
+  using BlobDB::PutWithTTL;
+  Status PutWithTTL(const WriteOptions& options,
+                    ColumnFamilyHandle* column_family, const Slice& key,
+                    const Slice& value, int32_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;
+
+  Status LinkToBaseDB(DB* db) override;
+
+  BlobDBImpl(DB* db, const BlobDBOptions& bdb_options);
+
+  BlobDBImpl(const std::string& dbname, const BlobDBOptions& bdb_options,
+             const DBOptions& db_options);
+
+  ~BlobDBImpl();
+
+ private:
+  static bool ExtractTTLFromBlob(const Slice& value, Slice* newval,
+                                 int32_t* ttl_val);
+
+  Status OpenPhase1();
+
+  Status CommonGet(const ColumnFamilyData* cfd, const Slice& key,
+                   const std::string& index_entry, std::string* value);
+
+  // Just before flush starts acting on memtable files,
+  // 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);
+
+  // 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);
+
+  Status AppendBlob(const std::shared_ptr<BlobFile>& bfile,
+                    const std::string& headerbuf, const Slice& key,
+                    const Slice& value, std::string* index_entry);
+
+  Status AppendSN(const std::shared_ptr<BlobFile>& bfile,
+                  const SequenceNumber& sn);
+
+  // 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);
+
+  // 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;
+
+  void UpdateWriteOptions(const WriteOptions& options);
+
+  void Shutdown();
+
+  // periodic sanity check. Bunch of checks
+  std::pair<bool, int64_t> SanityCheck(bool aborted);
+
+  // 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);
+
+  // Major task to garbage collect expired and deleted blobs
+  std::pair<bool, int64_t> RunGC(bool aborted);
+
+  // asynchronous task to fsync/fdatasync the open blob files
+  std::pair<bool, int64_t> FsyncFiles(bool aborted);
+
+  // periodically check if open blob files and their TTL's has expired
+  // if expired, close the sequential writer and make the file immutable
+  std::pair<bool, int64_t> CheckSeqFiles(bool aborted);
+
+  // if the number of open files, approaches ULIMIT's this
+  // task will close random readers, which are kept around for
+  // efficiency
+  std::pair<bool, int64_t> ReclaimOpenFiles(bool aborted);
+
+  // periodically print write amplification statistics
+  std::pair<bool, int64_t> WaStats(bool aborted);
+
+  // background task to do book-keeping of deleted keys
+  std::pair<bool, int64_t> EvictDeletions(bool aborted);
+
+  std::pair<bool, int64_t> EvictCompacted(bool aborted);
+
+  bool CallbackEvictsImpl(std::shared_ptr<BlobFile> bfile);
+
+  std::pair<bool, int64_t> RemoveTimerQ(TimerQueue* tq, bool aborted);
+
+  std::pair<bool, int64_t> CallbackEvicts(TimerQueue* tq,
+                                          std::shared_ptr<BlobFile> bfile,
+                                          bool aborted);
+
+  // Adds the background tasks to the timer queue
+  void StartBackgroundTasks();
+
+  // add a new Blob File
+  std::shared_ptr<BlobFile> NewBlobFile(const std::string& reason);
+
+  Status OpenAllFiles();
+
+  // hold write mutex on file and call
+  // creates a Random Access reader for GET call
+  std::shared_ptr<RandomAccessFileReader> GetOrOpenRandomAccessReader(
+      const std::shared_ptr<BlobFile>& bfile, Env* env,
+      const EnvOptions& env_options);
+
+  // hold write mutex on file and call.
+  // Close the above Random Access reader
+  void CloseRandomAccessLocked(const std::shared_ptr<BlobFile>& bfile);
+
+  // hold write mutex on file and call
+  // creates a sequential (append) writer for this blobfile
+  Status CreateWriterLocked(const std::shared_ptr<BlobFile>& bfile);
+
+  // 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);
+
+  // Iterate through keys and values on Blob and write into
+  // separate file the remaining blobs and delete/update pointers
+  // in LSM atomically
+  Status GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
+                            GCStats* gcstats);
+
+  // checks if there is no snapshot which is referencing the
+  // blobs
+  bool FileDeleteOk_SnapshotCheckLocked(const std::shared_ptr<BlobFile>& bfile);
+
+  bool MarkBlobDeleted(const Slice& key, const Slice& lsmValue);
+
+  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 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);
+
+ private:
+  // the base DB
+  DBImpl* db_impl_;
+
+  Env* myenv_;
+
+  // Optimistic Transaction DB used during Garbage collection
+  // for atomicity
+  std::unique_ptr<OptimisticTransactionDBImpl> opt_db_;
+
+  // a boolean to capture whether write_options has been set
+  std::atomic<bool> wo_set_;
+  WriteOptions write_options_;
+
+  // the options that govern the behavior of Blob Storage
+  BlobDBOptionsImpl bdb_options_;
+  DBOptions db_options_;
+  EnvOptions env_options_;
+
+  // name of the database directory
+  std::string dbname_;
+
+  // by default this is "blob_dir" under dbname_
+  // but can be configured
+  std::string blob_dir_;
+
+  // pointer to directory
+  std::unique_ptr<Directory> dir_ent_;
+
+  std::atomic<bool> dir_change_;
+
+  // Read Write Mutex, which protects all the data structures
+  // HEAVILY TRAFFICKED
+  port::RWMutex 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_;
+
+  // 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)
+  std::vector<std::shared_ptr<BlobFile>> open_simple_files_;
+
+  // all the blob files which are currently being appended to based
+  // on variety of incoming TTL's
+  std::multiset<std::shared_ptr<BlobFile>, blobf_compare_ttl> open_blob_files_;
+
+  // packet of information to put in lockess delete(s) queue
+  struct delete_packet_t {
+    ColumnFamilyHandle* cfh_;
+    std::string key_;
+    SequenceNumber dsn_;
+  };
+
+  struct override_packet_t {
+    uint64_t file_number_;
+    uint64_t key_size_;
+    uint64_t blob_offset_;
+    uint64_t blob_size_;
+    SequenceNumber dsn_;
+  };
+
+  // LOCKLESS multiple producer single consumer queue to quickly append
+  // deletes without taking lock. Can rapidly grow in size!!
+  // deletes happen in LSM, but minor book-keeping needs to happen on
+  // BLOB side (for triggering eviction)
+  mpsc_queue_t<delete_packet_t> delete_keys_q_;
+
+  // LOCKLESS multiple producer single consumer queue for values
+  // that are being compacted
+  mpsc_queue_t<override_packet_t> override_vals_q_;
+
+  // atomic bool to represent shutdown
+  std::atomic<bool> shutdown_;
+
+  // timer based queue to execute tasks
+  TimerQueue tqueue_;
+
+  // timer queues to call eviction callbacks.
+  std::vector<std::shared_ptr<TimerQueue>> cb_threads_;
+
+  // only accessed in GC thread, hence not atomic. The epoch of the
+  // GC task. Each execution is one epoch. Helps us in allocating
+  // files to one execution
+  uint64_t current_epoch_;
+
+  // number of files opened for random access/GET
+  // counter is used to monitor and close excess RA files.
+  std::atomic<uint32_t> open_file_count_;
+
+  // should hold mutex to modify
+  // STATISTICS for WA of Blob Files due to GC
+  // collect by default 24 hourly periods
+  std::list<uint64_t> all_periods_write_;
+  std::list<uint64_t> all_periods_ampl_;
+
+  std::atomic<uint64_t> last_period_write_;
+  std::atomic<uint64_t> last_period_ampl_;
+
+  uint64_t total_periods_write_;
+  uint64_t total_periods_ampl_;
+
+  // total size of all blob files at a given time
+  std::atomic<uint64_t> total_blob_space_;
+  std::list<std::shared_ptr<BlobFile>> obsolete_files_;
+  bool open_p1_done_;
+
+  uint32_t debug_level_;
+};
+
+class BlobFile {
+  friend class BlobDBImpl;
+  friend struct blobf_compare_ttl;
+
+ private:
+  // access to parent
+  const BlobDBImpl* parent_;
+
+  // path to blob directory
+  std::string path_to_dir_;
+
+  // the id of the file.
+  // the above 2 are created during file creation and never changed
+  // after that
+  uint64_t file_number_;
+
+  // number of blobs in the file
+  std::atomic<uint64_t> blob_count_;
+
+  // the file will be selected for GC in this future epoch
+  std::atomic<int64_t> gc_epoch_;
+
+  // size of the file
+  std::atomic<uint64_t> file_size_;
+
+  // number of blobs in this particular file which have been evicted
+  uint64_t deleted_count_;
+
+  // size of deleted blobs (used by heuristic to select file for GC)
+  uint64_t deleted_size_;
+
+  BlobLogHeader header_;
+
+  // closed_ = true implies the file is no more mutable
+  // no more blobs will be appended and the footer has been written out
+  std::atomic<bool> closed_;
+
+  // has a pass of garbage collection successfully finished on this file
+  // can_be_deleted_ still needs to do iterator/snapshot checks
+  std::atomic<bool> can_be_deleted_;
+
+  // should this file been gc'd once to reconcile lost deletes/compactions
+  std::atomic<bool> gc_once_after_open_;
+
+  // et - lt of the blobs
+  ttlrange_t ttl_range_;
+
+  // et - lt of the timestamp of the KV pairs.
+  tsrange_t time_range_;
+
+  // ESN - LSN of the blobs
+  snrange_t sn_range_;
+
+  // Sequential/Append writer for blobs
+  std::shared_ptr<Writer> log_writer_;
+
+  // random access file reader for GET calls
+  std::shared_ptr<RandomAccessFileReader> ra_file_reader_;
+
+  // This Read-Write mutex is per file specific and protects
+  // all the datastructures
+  port::RWMutex mutex_;
+
+  // time when the random access reader was last created.
+  std::atomic<std::time_t> last_access_;
+
+  // last time file was fsync'd/fdatasyncd
+  std::atomic<uint64_t> last_fsync_;
+
+  bool header_valid_;
+
+ public:
+  BlobFile();
+
+  BlobFile(const BlobDBImpl* parent, const std::string& bdir, uint64_t fnum);
+
+  ~BlobFile();
+
+  ColumnFamilyHandle* GetColumnFamily(DB* db);
+
+  // Returns log file's pathname relative to the main db dir
+  // Eg. For a live-log-file = blob_dir/000003.blob
+  std::string PathName() const;
+
+  // Primary identifier for blob file.
+  // once the file is created, this never changes
+  uint64_t BlobFileNumber() const { return file_number_; }
+
+  // the following functions are atomic, and don't need
+  // read lock
+  uint64_t BlobCount() const {
+    return blob_count_.load(std::memory_order_acquire);
+  }
+
+  std::string DumpState() const;
+
+  // if the file has gone through GC and blobs have been relocated
+  bool Obsolete() const { return can_be_deleted_.load(); }
+
+  // if the file is not taking any more appends.
+  bool Immutable() const { return closed_.load(); }
+
+  // we will assume this is atomic
+  bool NeedsFsync(bool hard, uint64_t bytes_per_sync) const;
+
+  uint64_t GetFileSize() const {
+    return file_size_.load(std::memory_order_acquire);
+  }
+
+  // All Get functions which are not atomic, will need ReadLock on the mutex
+  tsrange_t GetTimeRange() const {
+    assert(HasTimestamp());
+    return time_range_;
+  }
+
+  ttlrange_t GetTTLRange() const { return ttl_range_; }
+
+  snrange_t GetSNRange() const { return sn_range_; }
+
+  bool HasTTL() const {
+    assert(header_valid_);
+    return header_.HasTTL();
+  }
+
+  bool HasTimestamp() const {
+    assert(header_valid_);
+    return header_.HasTimestamp();
+  }
+
+  std::shared_ptr<Writer> GetWriter() const { return log_writer_; }
+
+  void Fsync();
+
+ private:
+  std::shared_ptr<Reader> OpenSequentialReader(
+      Env* env, const DBOptions& db_options,
+      const EnvOptions& env_options) const;
+
+  Status ReadFooter(BlobLogFooter* footer);
+
+  Status WriteFooterAndCloseLocked();
+
+  std::shared_ptr<RandomAccessFileReader> GetOrOpenRandomAccessReader(
+      Env* env, const EnvOptions& env_options, bool* fresh_open);
+
+  void CloseRandomAccessLocked();
+
+  // this is used, when you are reading only the footer of a
+  // previously closed file
+  Status SetFromFooterLocked(const BlobLogFooter& footer);
+
+  void set_time_range(const tsrange_t& tr) { time_range_ = tr; }
+
+  void set_ttl_range(const ttlrange_t& ttl) { ttl_range_ = ttl; }
+
+  void SetSNRange(const snrange_t& snr) { sn_range_ = snr; }
+
+  // The following functions are atomic, and don't need locks
+  void SetFileSize(uint64_t fs) { file_size_ = fs; }
+
+  void SetBlobCount(uint64_t bc) { blob_count_ = bc; }
+
+  void SetCanBeDeleted() { can_be_deleted_ = true; }
+};
+
+class BlobDBIterator : public Iterator {
+ public:
+  explicit BlobDBIterator(Iterator* iter, ColumnFamilyHandle* column_family,
+                          BlobDBImpl* impl)
+      : iter_(iter), cfh_(column_family), db_impl_(impl) {
+    assert(iter_);
+  }
+
+  ~BlobDBIterator() { delete iter_; }
+
+  bool Valid() const override { return iter_->Valid(); }
+
+  void SeekToFirst() override { iter_->SeekToFirst(); }
+
+  void SeekToLast() override { iter_->SeekToLast(); }
+
+  void Seek(const Slice& target) override { iter_->Seek(target); }
+
+  void SeekForPrev(const Slice& target) override { iter_->SeekForPrev(target); }
+
+  void Next() override { iter_->Next(); }
+
+  void Prev() override { iter_->Prev(); }
+
+  Slice key() const override { return iter_->key(); }
+
+  Slice value() const override;
+
+  Status status() const override { return iter_->status(); }
+
+ private:
+  Iterator* iter_;
+  ColumnFamilyHandle* cfh_;
+  BlobDBImpl* db_impl_;
+  mutable std::string vpart_;
+};
+
+}  // namespace blob_db
+}  // namespace rocksdb
+#endif  // ROCKSDB_LITE

utilities/blob_db/blob_db_options_impl.cc

@@ -0,0 +1,66 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same 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) {}
+
+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) {}
+
+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;
+  }
+  return *this;
+}
+
+}  // namespace blob_db
+}  // namespace rocksdb
+#endif  // ROCKSDB_LITE

utilities/blob_db/blob_db_options_impl.h

@@ -0,0 +1,73 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same 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;
+
+  // 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

@@ -1,66 +1,567 @@
-// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file. See the AUTHORS file for names of contributors.
-
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+//
 #ifndef ROCKSDB_LITE
 
 #include "utilities/blob_db/blob_db.h"
+#include <cstdlib>
+#include "db/db_test_util.h"
 #include "util/random.h"
 #include "util/testharness.h"
 #include "util/testutil.h"
+#include "utilities/blob_db/blob_db_options_impl.h"
 
 namespace rocksdb {
-class BlobDBTest : public testing::Test {
- public:
-  BlobDBTest() {
-    dbname_ = test::TmpDir() + "/blob_db_test";
-    Options options;
-    options.create_if_missing = true;
-    EXPECT_TRUE(NewBlobDB(options, dbname_, &db_).ok());
+
+namespace blob_db {
+Random s_rnd(301);
+
+void gen_random(char *s, const int len) {
+  static const char alphanum[] =
+      "0123456789"
+      "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+      "abcdefghijklmnopqrstuvwxyz";
+
+  for (int i = 0; i < len; ++i) {
+    s[i] = alphanum[s_rnd.Next() % (sizeof(alphanum) - 1)];
   }
 
-  ~BlobDBTest() { delete db_; }
+  s[len] = 0;
+}
 
-  DB* db_;
+class BlobDBTest : public testing::Test {
+ public:
+  BlobDBTest() : blobdb_(nullptr) {
+    dbname_ = test::TmpDir() + "/blob_db_test";
+    // Reopen1(BlobDBOptionsImpl());
+  }
+
+  ~BlobDBTest() {
+    if (blobdb_) {
+      delete blobdb_;
+      blobdb_ = nullptr;
+    }
+  }
+
+  void Reopen1(const BlobDBOptionsImpl &bdboptions,
+               const Options &options = Options()) {
+    if (blobdb_) {
+      delete blobdb_;
+      blobdb_ = nullptr;
+    }
+
+    BlobDBOptionsImpl bblobdb_options = bdboptions;
+    Options myoptions = options;
+    BlobDB::DestroyBlobDB(dbname_, myoptions, bblobdb_options);
+
+    DestroyDB(dbname_, myoptions);
+
+    myoptions.create_if_missing = true;
+    EXPECT_TRUE(
+        BlobDB::Open(myoptions, bblobdb_options, dbname_, &blobdb_).ok());
+  }
+
+  void insert_blobs() {
+    WriteOptions wo;
+    ReadOptions ro;
+    std::string value;
+
+    ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
+
+    Random rnd(301);
+    for (size_t i = 0; i < 100000; i++) {
+      int len = rnd.Next() % 16384;
+      if (!len) continue;
+
+      char *val = new char[len + 1];
+      gen_random(val, len);
+
+      std::string key("key");
+      key += std::to_string(i % 500);
+
+      Slice keyslice(key);
+      Slice valslice(val, len + 1);
+
+      int ttl = rnd.Next() % 86400;
+
+      ASSERT_OK(blobdb_->PutWithTTL(wo, dcfh, keyslice, valslice, ttl));
+      delete[] val;
+    }
+
+    for (size_t i = 0; i < 10; i++) {
+      std::string key("key");
+      key += std::to_string(i % 500);
+      Slice keyslice(key);
+      blobdb_->Delete(wo, dcfh, keyslice);
+    }
+  }
+
+  BlobDB *blobdb_;
   std::string dbname_;
 };  // class BlobDBTest
 
-TEST_F(BlobDBTest, Basic) {
+TEST_F(BlobDBTest, DeleteComplex) {
+  BlobDBOptionsImpl bdboptions;
+  bdboptions.partial_expiration_pct = 75;
+  bdboptions.gc_check_period_millisecs = 20 * 1000;
+  bdboptions.blob_file_size = 219 * 1024;
+
+  Reopen1(bdboptions);
+
   WriteOptions wo;
   ReadOptions ro;
   std::string value;
 
-  ASSERT_OK(db_->Put(wo, "foo", "v1"));
-  ASSERT_OK(db_->Put(wo, "bar", "v2"));
+  ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
 
-  ASSERT_OK(db_->Get(ro, "foo", &value));
-  ASSERT_EQ("v1", value);
-  ASSERT_OK(db_->Get(ro, "bar", &value));
-  ASSERT_EQ("v2", value);
+  Random rnd(301);
+  for (size_t i = 0; i < 100; i++) {
+    int len = rnd.Next() % 16384;
+    if (!len) continue;
+
+    char *val = new char[len + 1];
+    gen_random(val, len);
+
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + 1);
+
+    ASSERT_OK(blobdb_->Put(wo, dcfh, keyslice, valslice));
+    delete[] val;
+  }
+
+  for (size_t i = 0; i < 99; i++) {
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    blobdb_->Delete(wo, dcfh, keyslice);
+  }
+
+  Env::Default()->SleepForMicroseconds(60 * 1000 * 1000);
 }
 
+TEST_F(BlobDBTest, OverrideTest) {
+  BlobDBOptionsImpl bdboptions;
+  bdboptions.ttl_range_secs = 30;
+  bdboptions.gc_file_pct = 100;
+  bdboptions.gc_check_period_millisecs = 20 * 1000;
+  bdboptions.num_concurrent_simple_blobs = 2;
+  bdboptions.blob_file_size = 876 * 1024 * 10;
+
+  Options options;
+  options.write_buffer_size = 256 * 1024;
+  options.info_log_level = INFO_LEVEL;
+
+  Reopen1(bdboptions, options);
+
+  WriteOptions wo;
+  ReadOptions ro;
+  std::string value;
+
+  Random rnd(301);
+  ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
+
+  for (int i = 0; i < 10000; i++) {
+    int len = rnd.Next() % 16384;
+    if (!len) continue;
+
+    char *val = new char[len + 1];
+    gen_random(val, len);
+
+    std::string key("key");
+    char x[10];
+    std::sprintf(x, "%04d", i);
+    key += std::string(x);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + 1);
+
+    ASSERT_OK(blobdb_->Put(wo, dcfh, keyslice, valslice));
+    delete[] val;
+  }
+
+  // override all the keys
+  for (int i = 0; i < 10000; i++) {
+    int len = rnd.Next() % 16384;
+    if (!len) continue;
+
+    char *val = new char[len + 1];
+    gen_random(val, len);
+
+    std::string key("key");
+    char x[10];
+    std::sprintf(x, "%04d", i);
+    key += std::string(x);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + 1);
+
+    ASSERT_OK(blobdb_->Put(wo, dcfh, keyslice, valslice));
+    delete[] val;
+  }
+
+  blobdb_->Flush(FlushOptions());
+
+#if 1
+  blobdb_->GetBaseDB()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  reinterpret_cast<DBImpl *>(blobdb_->GetBaseDB())->TEST_WaitForFlushMemTable();
+  reinterpret_cast<DBImpl *>(blobdb_->GetBaseDB())->TEST_WaitForCompact();
+#endif
+
+  Env::Default()->SleepForMicroseconds(120 * 1000 * 1000);
+}
+
+TEST_F(BlobDBTest, DeleteTest) {
+  BlobDBOptionsImpl bdboptions;
+  bdboptions.ttl_range_secs = 30;
+  bdboptions.gc_file_pct = 100;
+  bdboptions.partial_expiration_pct = 18;
+  bdboptions.gc_check_period_millisecs = 20 * 1000;
+  bdboptions.num_concurrent_simple_blobs = 1;
+  bdboptions.blob_file_size = 876 * 1024;
+
+  Reopen1(bdboptions);
+
+  WriteOptions wo;
+  ReadOptions ro;
+  std::string value;
+
+  Random rnd(301);
+  ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
+
+  for (size_t i = 0; i < 100; i++) {
+    int len = rnd.Next() % 16384;
+    if (!len) continue;
+
+    char *val = new char[len + 1];
+    gen_random(val, len);
+
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + 1);
+
+    ASSERT_OK(blobdb_->Put(wo, dcfh, keyslice, valslice));
+    delete[] val;
+  }
+
+  for (size_t i = 0; i < 100; i += 5) {
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    blobdb_->Delete(wo, dcfh, keyslice);
+  }
+
+  Env::Default()->SleepForMicroseconds(60 * 1000 * 1000);
+}
+
+TEST_F(BlobDBTest, GCTestWithWrite) {
+  BlobDBOptionsImpl bdboptions;
+  bdboptions.ttl_range_secs = 30;
+  bdboptions.gc_file_pct = 100;
+  bdboptions.gc_check_period_millisecs = 20 * 1000;
+  bdboptions.default_ttl_extractor = true;
+
+  Reopen1(bdboptions);
+
+  WriteOptions wo;
+  ReadOptions ro;
+  std::string value;
+
+  ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
+
+  WriteBatch WB;
+
+  Random rnd(301);
+  for (size_t i = 0; i < 100; i++) {
+    int len = rnd.Next() % 16384;
+    if (!len) continue;
+
+    int ttl = 30;
+
+    char *val = new char[len + BlobDB::kTTLSuffixLength];
+    gen_random(val, len);
+    strncpy(val + len, "ttl:", 4);
+    EncodeFixed32(val + len + 4, ttl);
+
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + BlobDB::kTTLSuffixLength);
+
+    WB.Put(dcfh, keyslice, valslice);
+    delete[] val;
+  }
+
+  ASSERT_OK(blobdb_->Write(wo, &WB));
+
+  Env::Default()->SleepForMicroseconds(120 * 1000 * 1000);
+}
+
+void cb_evict(const ColumnFamilyHandle *cfh, const Slice &key,
+              const Slice &val) {
+  fprintf(stderr, "key evicted: %s\n", key.ToString().c_str());
+}
+
+static const char *LONG_STRING =
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFJFJFJTWFNLLFKFFMFMFMFMFMFMFMFMFMFMFMFMFMMF "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJAJFJFJFJFJTWBFNMFLLWMFMFMFMWKWMFMFMFMFMFMFM "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH "
+    "AJFJFJFFFFFFFFFFFFFFFFFFFJFHFHFHFHFHFHFHHFHHFHHFH ";
+
+TEST_F(BlobDBTest, GetWithCompression) {
+  BlobDBOptionsImpl bdboptions;
+  bdboptions.gc_file_pct = 100;
+  bdboptions.gc_check_period_millisecs = 20 * 1000;
+  bdboptions.default_ttl_extractor = true;
+  bdboptions.gc_evict_cb_fn = &cb_evict;
+  bdboptions.compression = CompressionType::kLZ4Compression;
+
+  Reopen1(bdboptions);
+
+  WriteOptions wo;
+  ReadOptions ro;
+  std::string value;
+  Random rnd(301);
+
+  ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
+
+  std::string orig(LONG_STRING);
+
+  for (size_t i = 0; i < 10000; i++) {
+    int len = orig.length();
+    int ttl = 3000 * (rnd.Next() % 10);
+
+    char *val = new char[len + BlobDB::kTTLSuffixLength];
+    strncpy(val, LONG_STRING, len);
+    strncpy(val + len, "ttl:", 4);
+    EncodeFixed32(val + len + 4, ttl);
+
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + BlobDB::kTTLSuffixLength);
+
+    ASSERT_OK(blobdb_->Put(wo, dcfh, keyslice, valslice));
+    delete[] val;
+  }
+
+  for (size_t i = 0; i < 10000; i++) {
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    std::string val;
+    Status s = blobdb_->Get(ro, dcfh, keyslice, &val);
+    ASSERT_TRUE(orig == val);
+  }
+
+  Env::Default()->SleepForMicroseconds(120 * 1000 * 1000);
+}
+
+TEST_F(BlobDBTest, GCTestWithPutAndCompression) {
+  BlobDBOptionsImpl bdboptions;
+  bdboptions.ttl_range_secs = 30;
+  bdboptions.gc_file_pct = 100;
+  bdboptions.gc_check_period_millisecs = 20 * 1000;
+  bdboptions.default_ttl_extractor = true;
+  bdboptions.gc_evict_cb_fn = &cb_evict;
+  bdboptions.compression = CompressionType::kLZ4Compression;
+
+  Reopen1(bdboptions);
+
+  WriteOptions wo;
+  ReadOptions ro;
+  std::string value;
+  Random rnd(301);
+
+  ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
+
+  for (size_t i = 0; i < 100; i++) {
+    int len = rnd.Next() % 16384;
+    if (!len) continue;
+
+    int ttl = 30;
+
+    char *val = new char[len + BlobDB::kTTLSuffixLength];
+    gen_random(val, len);
+    strncpy(val + len, "ttl:", 4);
+    EncodeFixed32(val + len + 4, ttl);
+
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + BlobDB::kTTLSuffixLength);
+
+    ASSERT_OK(blobdb_->Put(wo, dcfh, keyslice, valslice));
+    delete[] val;
+  }
+
+  Env::Default()->SleepForMicroseconds(120 * 1000 * 1000);
+}
+
+TEST_F(BlobDBTest, GCTestWithPut) {
+  BlobDBOptionsImpl bdboptions;
+  bdboptions.ttl_range_secs = 30;
+  bdboptions.gc_file_pct = 100;
+  bdboptions.gc_check_period_millisecs = 20 * 1000;
+  bdboptions.default_ttl_extractor = true;
+  bdboptions.gc_evict_cb_fn = &cb_evict;
+
+  Reopen1(bdboptions);
+
+  WriteOptions wo;
+  ReadOptions ro;
+  std::string value;
+  Random rnd(301);
+
+  ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
+
+  for (size_t i = 0; i < 100; i++) {
+    int len = rnd.Next() % 16384;
+    if (!len) continue;
+
+    int ttl = 30;
+
+    char *val = new char[len + BlobDB::kTTLSuffixLength];
+    gen_random(val, len);
+    strncpy(val + len, "ttl:", 4);
+    EncodeFixed32(val + len + 4, ttl);
+
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + BlobDB::kTTLSuffixLength);
+
+    ASSERT_OK(blobdb_->Put(wo, dcfh, keyslice, valslice));
+    delete[] val;
+  }
+
+  Env::Default()->SleepForMicroseconds(120 * 1000 * 1000);
+}
+
+TEST_F(BlobDBTest, GCTest) {
+  BlobDBOptionsImpl bdboptions;
+  bdboptions.ttl_range_secs = 30;
+  bdboptions.gc_file_pct = 100;
+
+  Reopen1(bdboptions);
+
+  WriteOptions wo;
+  ReadOptions ro;
+  std::string value;
+  Random rnd(301);
+
+  ColumnFamilyHandle *dcfh = blobdb_->DefaultColumnFamily();
+
+  for (size_t i = 0; i < 100; i++) {
+    int len = rnd.Next() % 16384;
+    if (!len) continue;
+
+    char *val = new char[len + 1];
+    gen_random(val, len);
+
+    std::string key("key");
+    key += std::to_string(i);
+
+    Slice keyslice(key);
+    Slice valslice(val, len + 1);
+
+    int ttl = 30;
+
+    ASSERT_OK(blobdb_->PutWithTTL(wo, dcfh, keyslice, valslice, ttl));
+    delete[] val;
+  }
+
+  Env::Default()->SleepForMicroseconds(240 * 1000 * 1000);
+}
+
+TEST_F(BlobDBTest, DISABLED_MultipleWriters) {
+  BlobDBOptionsImpl bdboptions;
+  Reopen1(bdboptions);
+
+  ASSERT_TRUE(blobdb_ != nullptr);
+
+  std::vector<std::thread> workers;
+  for (size_t ii = 0; ii < 10; ii++)
+    workers.push_back(std::thread(&BlobDBTest::insert_blobs, this));
+
+  for (std::thread &t : workers) {
+    if (t.joinable()) {
+      t.join();
+    }
+  }
+
+  Env::Default()->SleepForMicroseconds(180 * 1000 * 1000);
+  // ASSERT_OK(blobdb_->PutWithTTL(wo, dcfh, "bar", "v2", 60));
+  // ASSERT_OK(blobdb_->Get(ro, dcfh, "foo", &value));
+  // ASSERT_EQ("v1", value);
+  // ASSERT_OK(blobdb_->Get(ro, dcfh, "bar", &value));
+  // ASSERT_EQ("v2", value);
+}
+
+#if 0
 TEST_F(BlobDBTest, Large) {
+  ASSERT_TRUE(blobdb_ != nullptr);
+
   WriteOptions wo;
   ReadOptions ro;
   std::string value1, value2, value3;
   Random rnd(301);
+  ColumnFamilyHandle* dcfh = blobdb_->DefaultColumnFamily();
 
   value1.assign(8999, '1');
-  ASSERT_OK(db_->Put(wo, "foo", value1));
+  ASSERT_OK(blobdb_->PutWithTTL(wo, dcfh, "foo", value1, 3600));
   value2.assign(9001, '2');
-  ASSERT_OK(db_->Put(wo, "bar", value2));
+  ASSERT_OK(blobdb_->PutWithTTL(wo, dcfh, "bar", value2, 3600));
   test::RandomString(&rnd, 13333, &value3);
-  ASSERT_OK(db_->Put(wo, "barfoo", value3));
+  ASSERT_OK(blobdb_->PutWithTTL(wo, dcfh, "barfoo", value3, 3600));
 
   std::string value;
-  ASSERT_OK(db_->Get(ro, "foo", &value));
+  ASSERT_OK(blobdb_->Get(ro, dcfh, "foo", &value));
   ASSERT_EQ(value1, value);
-  ASSERT_OK(db_->Get(ro, "bar", &value));
+  ASSERT_OK(blobdb_->Get(ro, dcfh, "bar", &value));
   ASSERT_EQ(value2, value);
-  ASSERT_OK(db_->Get(ro, "barfoo", &value));
+  ASSERT_OK(blobdb_->Get(ro, dcfh, "barfoo", &value));
   ASSERT_EQ(value3, value);
 }
+#endif
 
+}  //  namespace blob_db
 }  //  namespace rocksdb
 
 // A black-box test for the ttl wrapper around rocksdb

utilities/blob_db/blob_file.cc

@@ -0,0 +1,226 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+#ifndef ROCKSDB_LITE
+
+#include <stdio.h>
+#include <chrono>
+#include <cinttypes>
+#include <memory>
+#include "utilities/blob_db/blob_db_impl.h"
+
+#include "util/filename.h"
+
+namespace rocksdb {
+
+namespace blob_db {
+
+BlobFile::BlobFile()
+    : parent_(nullptr),
+      file_number_(0),
+      blob_count_(0),
+      gc_epoch_(-1),
+      file_size_(0),
+      deleted_count_(0),
+      deleted_size_(0),
+      closed_(false),
+      can_be_deleted_(false),
+      gc_once_after_open_(false),
+      ttl_range_(std::make_pair(0, 0)),
+      time_range_(std::make_pair(0, 0)),
+      sn_range_(std::make_pair(0, 0)),
+      last_access_(-1),
+      last_fsync_(0),
+      header_valid_(false) {}
+
+BlobFile::BlobFile(const BlobDBImpl* p, const std::string& bdir, uint64_t fn)
+    : parent_(p),
+      path_to_dir_(bdir),
+      file_number_(fn),
+      blob_count_(0),
+      gc_epoch_(-1),
+      file_size_(0),
+      deleted_count_(0),
+      deleted_size_(0),
+      closed_(false),
+      can_be_deleted_(false),
+      gc_once_after_open_(false),
+      ttl_range_(std::make_pair(0, 0)),
+      time_range_(std::make_pair(0, 0)),
+      sn_range_(std::make_pair(0, 0)),
+      last_access_(-1),
+      last_fsync_(0),
+      header_valid_(false) {}
+
+BlobFile::~BlobFile() {
+  if (can_be_deleted_) {
+    std::string pn(PathName());
+    Status s = Env::Default()->DeleteFile(PathName());
+    if (!s.ok()) {
+      // Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
+      // "File could not be deleted %s", pn.c_str());
+    }
+  }
+}
+
+std::string BlobFile::PathName() const {
+  return BlobFileName(path_to_dir_, file_number_);
+}
+
+std::shared_ptr<Reader> BlobFile::OpenSequentialReader(
+    Env* env, const DBOptions& db_options,
+    const EnvOptions& env_options) const {
+  std::unique_ptr<SequentialFile> sfile;
+  Status s = env->NewSequentialFile(PathName(), &sfile, env_options);
+  if (!s.ok()) {
+    // report something here.
+    return nullptr;
+  }
+
+  std::unique_ptr<SequentialFileReader> sfile_reader;
+  sfile_reader.reset(new SequentialFileReader(std::move(sfile)));
+
+  std::shared_ptr<Reader> log_reader =
+      std::make_shared<Reader>(db_options.info_log, std::move(sfile_reader));
+
+  return log_reader;
+}
+
+std::string BlobFile::DumpState() const {
+  char str[1000];
+  snprintf(str, sizeof(str),
+           "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",
+           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,
+           ttl_range_.second, sn_range_.first, sn_range_.second,
+           (!!log_writer_), (!!ra_file_reader_));
+  return str;
+}
+
+bool BlobFile::NeedsFsync(bool hard, uint64_t bytes_per_sync) const {
+  assert(last_fsync_ <= file_size_);
+  return (hard) ? file_size_ > last_fsync_
+                : (file_size_ - last_fsync_) >= bytes_per_sync;
+}
+
+Status BlobFile::WriteFooterAndCloseLocked() {
+  Log(InfoLogLevel::INFO_LEVEL, parent_->db_options_.info_log,
+      "File is being closed after footer %s", PathName().c_str());
+
+  BlobLogFooter footer;
+  footer.blob_count_ = blob_count_;
+  if (HasTTL()) footer.set_ttl_range(ttl_range_);
+
+  footer.sn_range_ = sn_range_;
+  if (HasTimestamp()) footer.set_time_range(time_range_);
+
+  // this will close the file and reset the Writable File Pointer.
+  Status s = log_writer_->AppendFooter(footer);
+  if (s.ok()) {
+    closed_ = true;
+    file_size_ += BlobLogFooter::kFooterSize;
+  } else {
+    Log(InfoLogLevel::ERROR_LEVEL, parent_->db_options_.info_log,
+        "Failure to read Header for blob-file %s", PathName().c_str());
+  }
+  // delete the sequential writer
+  log_writer_.reset();
+  return s;
+}
+
+Status BlobFile::ReadFooter(BlobLogFooter* bf) {
+  if (file_size_ < (BlobLogHeader::kHeaderSize + BlobLogFooter::kFooterSize)) {
+    return Status::IOError("File does not have footer", PathName());
+  }
+
+  uint64_t footer_offset = file_size_ - BlobLogFooter::kFooterSize;
+  // assume that ra_file_reader_ is valid before we enter this
+  assert(ra_file_reader_);
+
+  Slice result;
+  char scratch[BlobLogFooter::kFooterSize + 10];
+  Status s = ra_file_reader_->Read(footer_offset, BlobLogFooter::kFooterSize,
+                                   &result, scratch);
+  if (!s.ok()) return s;
+  if (result.size() != BlobLogFooter::kFooterSize) {
+    // should not happen
+    return Status::IOError("EOF reached before footer");
+  }
+
+  s = bf->DecodeFrom(&result);
+  return s;
+}
+
+Status BlobFile::SetFromFooterLocked(const BlobLogFooter& footer) {
+  if (footer.HasTTL() != header_.HasTTL()) {
+    return Status::Corruption("has_ttl mismatch");
+  }
+  if (footer.HasTimestamp() != header_.HasTimestamp()) {
+    return Status::Corruption("has_ts mismatch");
+  }
+
+  // assume that file has been fully fsync'd
+  last_fsync_.store(file_size_);
+  blob_count_ = footer.GetBlobCount();
+  ttl_range_ = footer.GetTTLRange();
+  time_range_ = footer.GetTimeRange();
+  sn_range_ = footer.GetSNRange();
+  closed_ = true;
+
+  return Status::OK();
+}
+
+void BlobFile::Fsync() {
+  if (log_writer_.get()) {
+    log_writer_->Sync();
+    last_fsync_.store(file_size_.load());
+  }
+}
+
+void BlobFile::CloseRandomAccessLocked() {
+  ra_file_reader_.reset();
+  last_access_ = -1;
+}
+
+std::shared_ptr<RandomAccessFileReader> BlobFile::GetOrOpenRandomAccessReader(
+    Env* env, const EnvOptions& env_options, bool* fresh_open) {
+  *fresh_open = false;
+  last_access_ =
+      std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
+  {
+    ReadLock lockbfile_r(&mutex_);
+    if (ra_file_reader_) return ra_file_reader_;
+  }
+
+  WriteLock lockbfile_w(&mutex_);
+  if (ra_file_reader_) return ra_file_reader_;
+
+  std::unique_ptr<RandomAccessFile> rfile;
+  Status s = env->NewRandomAccessFile(PathName(), &rfile, env_options);
+  if (!s.ok()) {
+    Log(InfoLogLevel::ERROR_LEVEL, parent_->db_options_.info_log,
+        "Failed to open blob file for random-read: %s status: '%s'"
+        " exists: '%s'",
+        PathName().c_str(), s.ToString().c_str(),
+        env->FileExists(PathName()).ToString().c_str());
+    return nullptr;
+  }
+
+  ra_file_reader_ = std::make_shared<RandomAccessFileReader>(std::move(rfile));
+  *fresh_open = true;
+  return ra_file_reader_;
+}
+
+ColumnFamilyHandle* BlobFile::GetColumnFamily(DB* db) {
+  return db->DefaultColumnFamily();
+}
+
+}  // namespace blob_db
+}  // namespace rocksdb
+#endif  // ROCKSDB_LITE

utilities/blob_db/blob_log_format.cc

@@ -0,0 +1,313 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+//
+#ifndef ROCKSDB_LITE
+
+#include "utilities/blob_db/blob_log_format.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+
+namespace rocksdb {
+namespace blob_db {
+
+const uint32_t kMagicNumber = 2395959;
+const uint32_t kVersion1 = 1;
+const size_t kBlockSize = 32768;
+
+BlobLogHeader::BlobLogHeader()
+    : magic_number_(kMagicNumber), compression_(kNoCompression) {}
+
+BlobLogHeader& BlobLogHeader::operator=(BlobLogHeader&& in) noexcept {
+  if (this != &in) {
+    magic_number_ = in.magic_number_;
+    version_ = in.version_;
+    ttl_guess_ = std::move(in.ttl_guess_);
+    ts_guess_ = std::move(in.ts_guess_);
+    compression_ = in.compression_;
+  }
+  return *this;
+}
+
+BlobLogFooter::BlobLogFooter() : magic_number_(kMagicNumber), blob_count_(0) {}
+
+Status BlobLogFooter::DecodeFrom(Slice* input) {
+  uint32_t val;
+  if (!GetFixed32(input, &val)) {
+    return Status::Corruption("Invalid Blob Footer: flags");
+  }
+
+  bool has_ttl = false;
+  bool has_ts = false;
+  val >>= 8;
+  RecordSubType st = static_cast<RecordSubType>(val);
+  switch (st) {
+    case kRegularType:
+      break;
+    case kTTLType:
+      has_ttl = true;
+      break;
+    case kTimestampType:
+      has_ts = true;
+      break;
+    default:
+      return Status::Corruption("Invalid Blob Footer: flags_val");
+  }
+
+  if (!GetFixed64(input, &blob_count_)) {
+    return Status::Corruption("Invalid Blob Footer: blob_count");
+  }
+
+  ttlrange_t temp_ttl;
+  if (!GetFixed32(input, &temp_ttl.first) ||
+      !GetFixed32(input, &temp_ttl.second)) {
+    return Status::Corruption("Invalid Blob Footer: ttl_range");
+  }
+  if (has_ttl) {
+    printf("has ttl\n");
+    ttl_range_.reset(new ttlrange_t(temp_ttl));
+  }
+
+  if (!GetFixed64(input, &sn_range_.first) ||
+      !GetFixed64(input, &sn_range_.second)) {
+    return Status::Corruption("Invalid Blob Footer: sn_range");
+  }
+
+  tsrange_t temp_ts;
+  if (!GetFixed64(input, &temp_ts.first) ||
+      !GetFixed64(input, &temp_ts.second)) {
+    return Status::Corruption("Invalid Blob Footer: ts_range");
+  }
+  if (has_ts) ts_range_.reset(new tsrange_t(temp_ts));
+
+  if (!GetFixed32(input, &magic_number_) || magic_number_ != kMagicNumber) {
+    return Status::Corruption("Invalid Blob Footer: magic");
+  }
+
+  return Status::OK();
+}
+
+void BlobLogFooter::EncodeTo(std::string* dst) const {
+  dst->reserve(kFooterSize);
+
+  RecordType rt = kFullType;
+  RecordSubType st = kRegularType;
+  if (HasTTL()) {
+    st = kTTLType;
+  } else if (HasTimestamp()) {
+    st = kTimestampType;
+  }
+  uint32_t val = static_cast<uint32_t>(rt) | (static_cast<uint32_t>(st) << 8);
+  PutFixed32(dst, val);
+
+  PutFixed64(dst, blob_count_);
+  bool has_ttl = HasTTL();
+  bool has_ts = HasTimestamp();
+
+  if (has_ttl) {
+    PutFixed32(dst, ttl_range_.get()->first);
+    PutFixed32(dst, ttl_range_.get()->second);
+  } else {
+    PutFixed32(dst, 0);
+    PutFixed32(dst, 0);
+  }
+  PutFixed64(dst, sn_range_.first);
+  PutFixed64(dst, sn_range_.second);
+
+  if (has_ts) {
+    PutFixed64(dst, ts_range_.get()->first);
+    PutFixed64(dst, ts_range_.get()->second);
+  } else {
+    PutFixed64(dst, 0);
+    PutFixed64(dst, 0);
+  }
+
+  PutFixed32(dst, magic_number_);
+}
+
+void BlobLogHeader::EncodeTo(std::string* dst) const {
+  dst->reserve(kHeaderSize);
+
+  PutFixed32(dst, magic_number_);
+
+  PutFixed32(dst, version_);
+
+  RecordSubType st = kRegularType;
+  bool has_ttl = HasTTL();
+  bool has_ts = HasTimestamp();
+
+  if (has_ttl) {
+    st = kTTLType;
+  } else if (has_ts) {
+    st = kTimestampType;
+  }
+  uint32_t val =
+      static_cast<uint32_t>(st) | (static_cast<uint32_t>(compression_) << 8);
+  PutFixed32(dst, val);
+
+  if (has_ttl) {
+    PutFixed32(dst, ttl_guess_.get()->first);
+    PutFixed32(dst, ttl_guess_.get()->second);
+  } else {
+    PutFixed32(dst, 0);
+    PutFixed32(dst, 0);
+  }
+
+  if (has_ts) {
+    PutFixed64(dst, ts_guess_.get()->first);
+    PutFixed64(dst, ts_guess_.get()->second);
+  } else {
+    PutFixed64(dst, 0);
+    PutFixed64(dst, 0);
+  }
+}
+
+Status BlobLogHeader::DecodeFrom(Slice* input) {
+  if (!GetFixed32(input, &magic_number_) || magic_number_ != kMagicNumber) {
+    return Status::Corruption("Invalid Blob Log Header: magic");
+  }
+
+  // as of today, we only support 1 version
+  if (!GetFixed32(input, &version_) || version_ != kVersion1) {
+    return Status::Corruption("Invalid Blob Log Header: version");
+  }
+
+  uint32_t val;
+  if (!GetFixed32(input, &val)) {
+    return Status::Corruption("Invalid Blob Log Header: subtype");
+  }
+
+  bool has_ttl = false;
+  bool has_ts = false;
+  RecordSubType st = static_cast<RecordSubType>(val & 0xff);
+  compression_ = static_cast<CompressionType>((val >> 8) & 0xff);
+  switch (st) {
+    case kRegularType:
+      break;
+    case kTTLType:
+      has_ttl = true;
+      break;
+    case kTimestampType:
+      has_ts = true;
+      break;
+    default:
+      return Status::Corruption("Invalid Blob Log Header: subtype_2");
+  }
+
+  ttlrange_t temp_ttl;
+  if (!GetFixed32(input, &temp_ttl.first) ||
+      !GetFixed32(input, &temp_ttl.second)) {
+    return Status::Corruption("Invalid Blob Log Header: ttl");
+  }
+  if (has_ttl) set_ttl_guess(temp_ttl);
+
+  tsrange_t temp_ts;
+  if (!GetFixed64(input, &temp_ts.first) ||
+      !GetFixed64(input, &temp_ts.second)) {
+    return Status::Corruption("Invalid Blob Log Header: timestamp");
+  }
+  if (has_ts) set_ts_guess(temp_ts);
+
+  return Status::OK();
+}
+
+BlobLogRecord::BlobLogRecord()
+    : checksum_(0),
+      header_cksum_(0),
+      key_size_(0),
+      blob_size_(0),
+      time_val_(0),
+      ttl_val_(0),
+      sn_(0),
+      type_(0),
+      subtype_(0) {}
+
+BlobLogRecord::~BlobLogRecord() {}
+
+void BlobLogRecord::ResizeKeyBuffer(size_t kbs) {
+  if (kbs > key_buffer_.size()) {
+    key_buffer_.resize(kbs);
+  }
+}
+
+void BlobLogRecord::ResizeBlobBuffer(size_t bbs) {
+  if (bbs > blob_buffer_.size()) {
+    blob_buffer_.resize(bbs);
+  }
+}
+
+void BlobLogRecord::Clear() {
+  checksum_ = 0;
+  header_cksum_ = 0;
+  key_size_ = 0;
+  blob_size_ = 0;
+  time_val_ = 0;
+  ttl_val_ = 0;
+  sn_ = 0;
+  type_ = subtype_ = 0;
+  key_.clear();
+  blob_.clear();
+}
+
+Status BlobLogRecord::DecodeHeaderFrom(const Slice& hdrslice) {
+  Slice input = hdrslice;
+  if (input.size() < kHeaderSize) {
+    return Status::Corruption("Invalid Blob Record Header: size");
+  }
+
+  if (!GetFixed32(&input, &key_size_)) {
+    return Status::Corruption("Invalid Blob Record Header: key_size");
+  }
+  if (!GetFixed64(&input, &blob_size_)) {
+    return Status::Corruption("Invalid Blob Record Header: blob_size");
+  }
+  if (!GetFixed32(&input, &ttl_val_)) {
+    return Status::Corruption("Invalid Blob Record Header: ttl_val");
+  }
+  if (!GetFixed64(&input, &time_val_)) {
+    return Status::Corruption("Invalid Blob Record Header: time_val");
+  }
+
+  type_ = *(input.data());
+  input.remove_prefix(1);
+  subtype_ = *(input.data());
+  input.remove_prefix(1);
+
+  if (!GetFixed32(&input, &header_cksum_)) {
+    return Status::Corruption("Invalid Blob Record Header: header_cksum");
+  }
+  if (!GetFixed32(&input, &checksum_)) {
+    return Status::Corruption("Invalid Blob Record Header: checksum");
+  }
+
+  return Status::OK();
+}
+
+Status BlobLogRecord::DecodeFooterFrom(const Slice& footerslice) {
+  Slice input = footerslice;
+  if (input.size() < kFooterSize) {
+    return Status::Corruption("Invalid Blob Record Footer: size");
+  }
+
+  uint32_t f_crc = crc32c::Extend(0, input.data(), 8);
+  f_crc = crc32c::Mask(f_crc);
+
+  if (!GetFixed64(&input, &sn_)) {
+    return Status::Corruption("Invalid Blob Record Footer: sn");
+  }
+
+  if (!GetFixed32(&input, &footer_cksum_)) {
+    return Status::Corruption("Invalid Blob Record Footer: cksum");
+  }
+
+  if (f_crc != footer_cksum_) {
+    return Status::Corruption("Record Checksum mismatch: footer_cksum");
+  }
+
+  return Status::OK();
+}
+
+}  // namespace blob_db
+}  // namespace rocksdb
+#endif  // ROCKSDB_LITE

utilities/blob_db/blob_log_format.h

@@ -0,0 +1,226 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+//
+// Log format information shared by reader and writer.
+
+#pragma once
+
+#ifndef ROCKSDB_LITE
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+#include "rocksdb/options.h"
+#include "rocksdb/status.h"
+#include "rocksdb/types.h"
+
+namespace rocksdb {
+
+namespace blob_db {
+class BlobFile;
+class BlobDBImpl;
+
+enum RecordType : uint8_t {
+  // Zero is reserved for preallocated files
+  kFullType = 0,
+
+  // For fragments
+  kFirstType = 1,
+  kMiddleType = 2,
+  kLastType = 3,
+  kMaxRecordType = kLastType
+};
+
+enum RecordSubType : uint8_t {
+  kRegularType = 0,
+  kTTLType = 1,
+  kTimestampType = 2,
+};
+
+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;
+
+class BlobLogHeader {
+  friend class BlobFile;
+  friend class BlobDBImpl;
+
+ private:
+  uint32_t magic_number_ = 0;
+  uint32_t version_ = 1;
+  CompressionType compression_;
+  std::unique_ptr<ttlrange_t> ttl_guess_;
+  std::unique_ptr<tsrange_t> ts_guess_;
+
+ private:
+  void set_ttl_guess(const ttlrange_t& ttl) {
+    ttl_guess_.reset(new ttlrange_t(ttl));
+  }
+
+  void set_version(uint32_t v) { version_ = v; }
+
+  void set_ts_guess(const tsrange_t& ts) { ts_guess_.reset(new tsrange_t(ts)); }
+
+ public:
+  // magic number + version + flags + ttl guess + timestamp range
+  static const size_t kHeaderSize = 4 + 4 + 4 + 4 * 2 + 8 * 2;
+  // 32
+
+  void EncodeTo(std::string* dst) const;
+
+  Status DecodeFrom(Slice* input);
+
+  BlobLogHeader();
+
+  bool HasTTL() const { return !!ttl_guess_; }
+
+  bool HasTimestamp() const { return !!ts_guess_; }
+
+  BlobLogHeader& operator=(BlobLogHeader&& in) noexcept;
+};
+
+// Footer encapsulates the fixed information stored at the tail
+// end of every blob log file.
+class BlobLogFooter {
+  friend class BlobFile;
+
+ public:
+  // Use this constructor when you plan to write out the footer using
+  // EncodeTo(). Never use this constructor with DecodeFrom().
+  BlobLogFooter();
+
+  uint64_t magic_number() const { return magic_number_; }
+
+  void EncodeTo(std::string* dst) const;
+
+  Status DecodeFrom(Slice* input);
+
+  // convert this object to a human readable form
+  std::string ToString() const;
+
+  // footer size = 4 byte magic number
+  // 8 bytes count
+  // 4, 4 - 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);
+
+  bool HasTTL() const { return !!ttl_range_; }
+
+  bool HasTimestamp() const { return !!ts_range_; }
+
+  uint64_t GetBlobCount() const { return blob_count_; }
+
+  ttlrange_t GetTTLRange() const {
+    if (ttl_range_) {
+      *ttl_range_;
+    }
+    return {0, 0};
+  }
+
+  tsrange_t GetTimeRange() const {
+    if (ts_range_) {
+      return *ts_range_;
+    }
+    return {0, 0};
+  }
+
+  const snrange_t& GetSNRange() const { return sn_range_; }
+
+ private:
+  uint32_t magic_number_ = 0;
+  uint64_t blob_count_ = 0;
+
+  std::unique_ptr<ttlrange_t> ttl_range_;
+  std::unique_ptr<tsrange_t> ts_range_;
+  snrange_t sn_range_;
+
+ private:
+  void set_ttl_range(const ttlrange_t& ttl) {
+    ttl_range_.reset(new ttlrange_t(ttl));
+  }
+  void set_time_range(const tsrange_t& ts) {
+    ts_range_.reset(new tsrange_t(ts));
+  }
+};
+
+extern const size_t kBlockSize;
+
+class BlobLogRecord {
+  friend class Reader;
+
+ private:
+  // this might not be set.
+  uint32_t checksum_;
+  uint32_t header_cksum_;
+  uint32_t key_size_;
+  uint64_t blob_size_;
+  uint64_t time_val_;
+  uint32_t ttl_val_;
+  SequenceNumber sn_;
+  uint32_t footer_cksum_;
+  char type_;
+  char subtype_;
+  Slice key_;
+  Slice blob_;
+  std::string key_buffer_;
+  std::string blob_buffer_;
+
+ private:
+  void Clear();
+
+  char* GetKeyBuffer() { return &(key_buffer_[0]); }
+
+  char* GetBlobBuffer() { return &(blob_buffer_[0]); }
+
+  void ResizeKeyBuffer(size_t kbs);
+
+  void ResizeBlobBuffer(size_t bbs);
+
+ public:
+  // Header is
+  // Key Length ( 4 bytes ),
+  // Blob Length ( 8 bytes), timestamp/ttl (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;
+
+  static const size_t kFooterSize = 8 + 4;
+
+ public:
+  BlobLogRecord();
+
+  ~BlobLogRecord();
+
+  const Slice& Key() const { return key_; }
+
+  const Slice& Blob() const { return blob_; }
+
+  uint32_t GetKeySize() const { return key_size_; }
+
+  uint64_t GetBlobSize() const { return blob_size_; }
+
+  uint32_t GetTTL() const { return ttl_val_; }
+
+  uint64_t GetTimeVal() const { return time_val_; }
+
+  SequenceNumber GetSN() const { return sn_; }
+
+  Status DecodeHeaderFrom(const Slice& hdrslice);
+
+  Status DecodeFooterFrom(const Slice& footerslice);
+};
+
+}  // namespace blob_db
+}  // namespace rocksdb
+#endif  // ROCKSDB_LITE

utilities/blob_db/blob_log_reader.cc

@@ -0,0 +1,163 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+//
+#ifndef ROCKSDB_LITE
+
+#include "utilities/blob_db/blob_log_reader.h"
+
+#include <cstdio>
+#include "rocksdb/env.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/file_reader_writer.h"
+
+namespace rocksdb {
+namespace blob_db {
+
+Reader::Reader(std::shared_ptr<Logger> info_log,
+               unique_ptr<SequentialFileReader>&& _file)
+    : info_log_(info_log), file_(std::move(_file)), buffer_(), next_byte_(0) {
+  backing_store_.resize(kBlockSize);
+}
+
+Reader::~Reader() {}
+
+Status Reader::ReadHeader(BlobLogHeader* header) {
+  assert(file_.get() != nullptr);
+  assert(next_byte_ == 0);
+  Status status =
+      file_->Read(BlobLogHeader::kHeaderSize, &buffer_, GetReadBuffer());
+  next_byte_ += buffer_.size();
+  if (!status.ok()) return status;
+
+  if (buffer_.size() != BlobLogHeader::kHeaderSize) {
+    return Status::IOError("EOF reached before file header");
+  }
+
+  status = header->DecodeFrom(&buffer_);
+  return status;
+}
+
+Status Reader::ReadRecord(BlobLogRecord* record, ReadLevel level,
+                          WALRecoveryMode wal_recovery_mode) {
+  record->Clear();
+  buffer_.clear();
+  backing_store_[0] = '\0';
+
+  Status status =
+      file_->Read(BlobLogRecord::kHeaderSize, &buffer_, GetReadBuffer());
+  next_byte_ += buffer_.size();
+  if (!status.ok()) return status;
+  if (buffer_.size() != BlobLogRecord::kHeaderSize) {
+    return Status::IOError("EOF reached before record header");
+  }
+
+  status = record->DecodeHeaderFrom(buffer_);
+  if (!status.ok()) return status;
+
+  uint32_t header_crc = 0;
+  uint32_t blob_crc = 0;
+  size_t crc_data_size = BlobLogRecord::kHeaderSize - 2 * sizeof(uint32_t);
+  header_crc = crc32c::Extend(header_crc, buffer_.data(), crc_data_size);
+
+  uint64_t kb_size = record->GetKeySize() + record->GetBlobSize();
+  switch (level) {
+    case kReadHdrFooter:
+      file_->Skip(kb_size);
+      next_byte_ += kb_size;
+      status =
+          file_->Read(BlobLogRecord::kFooterSize, &buffer_, GetReadBuffer());
+      next_byte_ += buffer_.size();
+      if (!status.ok()) return status;
+      if (buffer_.size() != BlobLogRecord::kFooterSize) {
+        return Status::IOError("EOF reached before record footer");
+      }
+
+      status = record->DecodeFooterFrom(buffer_);
+      return status;
+
+    case kReadHdrKeyFooter:
+      record->ResizeKeyBuffer(record->GetKeySize());
+      status = file_->Read(record->GetKeySize(), &record->key_,
+                           record->GetKeyBuffer());
+      next_byte_ += record->key_.size();
+      if (!status.ok()) return status;
+      if (record->key_.size() != record->GetKeySize()) {
+        return Status::IOError("EOF reached before key read");
+      }
+
+      header_crc =
+          crc32c::Extend(header_crc, record->key_.data(), record->GetKeySize());
+      header_crc = crc32c::Mask(header_crc);
+      if (header_crc != record->header_cksum_) {
+        return Status::Corruption("Record Checksum mismatch: header_cksum");
+      }
+
+      file_->Skip(record->GetBlobSize());
+      next_byte_ += record->GetBlobSize();
+
+      status =
+          file_->Read(BlobLogRecord::kFooterSize, &buffer_, GetReadBuffer());
+      next_byte_ += buffer_.size();
+      if (!status.ok()) return status;
+      if (buffer_.size() != BlobLogRecord::kFooterSize) {
+        return Status::IOError("EOF reached during footer read");
+      }
+
+      status = record->DecodeFooterFrom(buffer_);
+      return status;
+
+    case kReadHdrKeyBlobFooter:
+      record->ResizeKeyBuffer(record->GetKeySize());
+      status = file_->Read(record->GetKeySize(), &record->key_,
+                           record->GetKeyBuffer());
+      next_byte_ += record->key_.size();
+      if (!status.ok()) return status;
+      if (record->key_.size() != record->GetKeySize()) {
+        return Status::IOError("EOF reached before key read");
+      }
+
+      header_crc =
+          crc32c::Extend(header_crc, record->key_.data(), record->GetKeySize());
+      header_crc = crc32c::Mask(header_crc);
+      if (header_crc != record->header_cksum_) {
+        return Status::Corruption("Record Checksum mismatch: header_cksum");
+      }
+
+      record->ResizeBlobBuffer(record->GetBlobSize());
+      status = file_->Read(record->GetBlobSize(), &record->blob_,
+                           record->GetBlobBuffer());
+      next_byte_ += record->blob_.size();
+      if (!status.ok()) return status;
+      if (record->blob_.size() != record->GetBlobSize()) {
+        return Status::IOError("EOF reached during blob read");
+      }
+
+      blob_crc =
+          crc32c::Extend(blob_crc, record->blob_.data(), record->blob_.size());
+      blob_crc = crc32c::Mask(blob_crc);
+      if (blob_crc != record->checksum_) {
+        return Status::Corruption("Blob Checksum mismatch");
+      }
+
+      status =
+          file_->Read(BlobLogRecord::kFooterSize, &buffer_, GetReadBuffer());
+      next_byte_ += buffer_.size();
+      if (!status.ok()) return status;
+      if (buffer_.size() != BlobLogRecord::kFooterSize) {
+        return Status::IOError("EOF reached during blob footer read");
+      }
+
+      status = record->DecodeFooterFrom(buffer_);
+      return status;
+    default:
+      assert(0);
+      return status;
+  }
+}
+
+}  // namespace blob_db
+}  // namespace rocksdb
+#endif  // ROCKSDB_LITE