Some small improvements to the build_tools

WritePrepared Txn: AddPrepared for all sub-batches
Summary: Currently AddPrepared is performed only on the first sub-batch if there are duplicate keys in the write batch. This could cause a problem if the transaction takes too long to commit and the seq number of the first sub-patch moved to old_prepared_ but not the seq of the later ones. The patch fixes this by calling AddPrepared for all sub-patches. Closes https://github.com/facebook/rocksdb/pull/3651 Differential Revision: D7388635 Pulled By: maysamyabandeh fbshipit-source-id: 0ccd80c150d9bc42fe955e49ddb9d7ca353067b4
2026-11-07 03:22:34 -05:00 · 2018-03-24 21:33:43 +00:00 · 2018-03-23 17:30:04 -07:00 · 2018-03-23 13:26:55 -07:00 · 2018-03-23 12:56:52 -07:00 · 2018-03-23 12:43:06 -07:00
70 changed files with 1210 additions and 370 deletions
@@ -569,6 +569,7 @@ set(SOURCES
        util/status_message.cc
        util/string_util.cc
        util/sync_point.cc
+        util/sync_point_impl.cc
        util/testutil.cc
        util/thread_local.cc
        util/threadpool_imp.cc
@@ -1,6 +1,17 @@
 # Rocksdb Change Log
 ## Unreleased
 ### Public API Change
+
+### New Features
+
+### Bug Fixes
+* Fsync after writing global seq number to the ingestion file in ExternalSstFileIngestionJob.
+
+### Java API Changes
+* Add `BlockBasedTableConfig.setBlockCache` to allow sharing a block cache across DB instances.
+
+## 5.13.0 (3/20/2018)
+### Public API Change
 * RocksDBOptionsParser::Parse()'s `ignore_unknown_options` argument will only be effective if the option file shows it is generated using a higher version of RocksDB than the current version.
 * Remove CompactionEventListener.

@@ -9,6 +20,7 @@
 * Avoid unnecessarily flushing in `CompactRange()` when the range specified by the user does not overlap unflushed memtables.
 * If `ColumnFamilyOptions::max_subcompactions` is set greater than one, we now parallelize large manual level-based compactions.
 * Add "rocksdb.live-sst-files-size" DB property to return total bytes of all SST files belong to the latest LSM tree.
+* NewSstFileManager to add an argument bytes_max_delete_chunk with default 64MB. With this argument, a file larger than 64MB will be ftruncated multiple times based on this size.

 ### Bug Fixes
 * Fix a leak in prepared_section_completed_ where the zeroed entries would not removed from the map.
@@ -215,6 +215,7 @@ cpp_library(
        "util/status_message.cc",
        "util/string_util.cc",
        "util/sync_point.cc",
+        "util/sync_point_impl.cc",
        "util/thread_local.cc",
        "util/threadpool_imp.cc",
        "util/transaction_test_util.cc",
@@ -1,4 +1,4 @@
-#!/bin/sh
+#!/usr/bin/env bash
 #
 # Detects OS we're compiling on and outputs a file specified by the first
 # argument, which in turn gets read while processing Makefile.
@@ -66,11 +66,19 @@ rm -f "$OUTPUT"
 touch "$OUTPUT"

 if test -z "$CC"; then
-   CC=cc
+    if [ -x "$(command -v clang)" ]; then
+        CC=clang
+    else
+        CC=cc
+    fi
 fi

 if test -z "$CXX"; then
-    CXX=g++
+    if [ -x "$(command -v clang++)" ]; then
+        CXX=clang++
+    else
+        CXX=g++
+    fi
 fi

 # Detect OS
@@ -87,7 +95,7 @@ if test -z "$CLANG_SCAN_BUILD"; then
 fi

 if test -z "$CLANG_ANALYZER"; then
-    CLANG_ANALYZER=$(which clang++ 2> /dev/null)
+    CLANG_ANALYZER=$(command -v clang++ 2> /dev/null)
 fi

 if test -z "$FIND"; then
@@ -498,9 +506,9 @@ if test -z "$PORTABLE"; then
  elif test -n "`echo $TARGET_ARCHITECTURE | grep ^arm`"; then
    # TODO: Handle this with approprite options.
    COMMON_FLAGS="$COMMON_FLAGS"
-  elif [ "$TARGET_OS" == IOS ]; then
+  elif [ "$TARGET_OS" == "IOS" ]; then
    COMMON_FLAGS="$COMMON_FLAGS"
-  elif [ "$TARGET_OS" != AIX ] && [ "$TARGET_OS" != SunOS ]; then
+  elif [ "$TARGET_OS" != "AIX" ] && [ "$TARGET_OS" != "SunOS" ]; then
    COMMON_FLAGS="$COMMON_FLAGS -march=native "
  elif test "$USE_SSE"; then
    COMMON_FLAGS="$COMMON_FLAGS -msse4.2 -mpclmul"
@@ -1,4 +1,4 @@
-#!/bin/sh
+#/usr/bin/env bash
 if [ "$#" = "0" ]; then
  echo "Usage: $0 major|minor|patch|full"
  exit 1
@@ -38,6 +38,7 @@
 #include "rocksdb/utilities/transaction_db.h"
 #include "rocksdb/utilities/write_batch_with_index.h"
 #include "rocksdb/write_batch.h"
+#include "rocksdb/perf_context.h"
 #include "utilities/merge_operators.h"

 using rocksdb::BytewiseComparator;
@@ -101,6 +102,8 @@ using rocksdb::OptimisticTransactionDB;
 using rocksdb::OptimisticTransactionOptions;
 using rocksdb::Transaction;
 using rocksdb::Checkpoint;
+using rocksdb::PerfLevel;
+using rocksdb::PerfContext;

 using std::shared_ptr;

@@ -141,6 +144,7 @@ struct rocksdb_envoptions_t      { EnvOptions        rep; };
 struct rocksdb_ingestexternalfileoptions_t  { IngestExternalFileOptions rep; };
 struct rocksdb_sstfilewriter_t   { SstFileWriter*    rep; };
 struct rocksdb_ratelimiter_t     { RateLimiter*      rep; };
+struct rocksdb_perfcontext_t     { PerfContext*      rep; };
 struct rocksdb_pinnableslice_t {
  PinnableSlice rep;
 };
@@ -2552,6 +2556,176 @@ void rocksdb_ratelimiter_destroy(rocksdb_ratelimiter_t *limiter) {
  delete limiter;
 }

+void rocksdb_set_perf_level(int v) {
+  PerfLevel level = static_cast<PerfLevel>(v);
+  SetPerfLevel(level);
+}
+
+rocksdb_perfcontext_t* rocksdb_perfcontext_create() {
+  rocksdb_perfcontext_t* context = new rocksdb_perfcontext_t;
+  context->rep = rocksdb::get_perf_context();
+  return context;
+}
+
+void rocksdb_perfcontext_reset(rocksdb_perfcontext_t* context) {
+  context->rep->Reset();
+}
+
+char* rocksdb_perfcontext_report(rocksdb_perfcontext_t* context,
+    unsigned char exclude_zero_counters) {
+  return strdup(context->rep->ToString(exclude_zero_counters).c_str());
+}
+
+uint64_t rocksdb_perfcontext_metric(rocksdb_perfcontext_t* context,
+    int metric) {
+  PerfContext* rep = context->rep;
+  switch (metric) {
+    case rocksdb_user_key_comparison_count:
+      return rep->user_key_comparison_count;
+    case rocksdb_block_cache_hit_count:
+      return rep->block_cache_hit_count;
+    case rocksdb_block_read_count:
+      return rep->block_read_count;
+    case rocksdb_block_read_byte:
+      return rep->block_read_byte;
+    case rocksdb_block_read_time:
+      return rep->block_read_time;
+    case rocksdb_block_checksum_time:
+      return rep->block_checksum_time;
+    case rocksdb_block_decompress_time:
+      return rep->block_decompress_time;
+    case rocksdb_get_read_bytes:
+      return rep->get_read_bytes;
+    case rocksdb_multiget_read_bytes:
+      return rep->multiget_read_bytes;
+    case rocksdb_iter_read_bytes:
+      return rep->iter_read_bytes;
+    case rocksdb_internal_key_skipped_count:
+      return rep->internal_key_skipped_count;
+    case rocksdb_internal_delete_skipped_count:
+      return rep->internal_delete_skipped_count;
+    case rocksdb_internal_recent_skipped_count:
+      return rep->internal_recent_skipped_count;
+    case rocksdb_internal_merge_count:
+      return rep->internal_merge_count;
+    case rocksdb_get_snapshot_time:
+      return rep->get_snapshot_time;
+    case rocksdb_get_from_memtable_time:
+      return rep->get_from_memtable_time;
+    case rocksdb_get_from_memtable_count:
+      return rep->get_from_memtable_count;
+    case rocksdb_get_post_process_time:
+      return rep->get_post_process_time;
+    case rocksdb_get_from_output_files_time:
+      return rep->get_from_output_files_time;
+    case rocksdb_seek_on_memtable_time:
+      return rep->seek_on_memtable_time;
+    case rocksdb_seek_on_memtable_count:
+      return rep->seek_on_memtable_count;
+    case rocksdb_next_on_memtable_count:
+      return rep->next_on_memtable_count;
+    case rocksdb_prev_on_memtable_count:
+      return rep->prev_on_memtable_count;
+    case rocksdb_seek_child_seek_time:
+      return rep->seek_child_seek_time;
+    case rocksdb_seek_child_seek_count:
+      return rep->seek_child_seek_count;
+    case rocksdb_seek_min_heap_time:
+      return rep->seek_min_heap_time;
+    case rocksdb_seek_max_heap_time:
+      return rep->seek_max_heap_time;
+    case rocksdb_seek_internal_seek_time:
+      return rep->seek_internal_seek_time;
+    case rocksdb_find_next_user_entry_time:
+      return rep->find_next_user_entry_time;
+    case rocksdb_write_wal_time:
+      return rep->write_wal_time;
+    case rocksdb_write_memtable_time:
+      return rep->write_memtable_time;
+    case rocksdb_write_delay_time:
+      return rep->write_delay_time;
+    case rocksdb_write_pre_and_post_process_time:
+      return rep->write_pre_and_post_process_time;
+    case rocksdb_db_mutex_lock_nanos:
+      return rep->db_mutex_lock_nanos;
+    case rocksdb_db_condition_wait_nanos:
+      return rep->db_condition_wait_nanos;
+    case rocksdb_merge_operator_time_nanos:
+      return rep->merge_operator_time_nanos;
+    case rocksdb_read_index_block_nanos:
+      return rep->read_index_block_nanos;
+    case rocksdb_read_filter_block_nanos:
+      return rep->read_filter_block_nanos;
+    case rocksdb_new_table_block_iter_nanos:
+      return rep->new_table_block_iter_nanos;
+    case rocksdb_new_table_iterator_nanos:
+      return rep->new_table_iterator_nanos;
+    case rocksdb_block_seek_nanos:
+      return rep->block_seek_nanos;
+    case rocksdb_find_table_nanos:
+      return rep->find_table_nanos;
+    case rocksdb_bloom_memtable_hit_count:
+      return rep->bloom_memtable_hit_count;
+    case rocksdb_bloom_memtable_miss_count:
+      return rep->bloom_memtable_miss_count;
+    case rocksdb_bloom_sst_hit_count:
+      return rep->bloom_sst_hit_count;
+    case rocksdb_bloom_sst_miss_count:
+      return rep->bloom_sst_miss_count;
+    case rocksdb_key_lock_wait_time:
+      return rep->key_lock_wait_time;
+    case rocksdb_key_lock_wait_count:
+      return rep->key_lock_wait_count;
+    case rocksdb_env_new_sequential_file_nanos:
+      return rep->env_new_sequential_file_nanos;
+    case rocksdb_env_new_random_access_file_nanos:
+      return rep->env_new_random_access_file_nanos;
+    case rocksdb_env_new_writable_file_nanos:
+      return rep->env_new_writable_file_nanos;
+    case rocksdb_env_reuse_writable_file_nanos:
+      return rep->env_reuse_writable_file_nanos;
+    case rocksdb_env_new_random_rw_file_nanos:
+      return rep->env_new_random_rw_file_nanos;
+    case rocksdb_env_new_directory_nanos:
+      return rep->env_new_directory_nanos;
+    case rocksdb_env_file_exists_nanos:
+      return rep->env_file_exists_nanos;
+    case rocksdb_env_get_children_nanos:
+      return rep->env_get_children_nanos;
+    case rocksdb_env_get_children_file_attributes_nanos:
+      return rep->env_get_children_file_attributes_nanos;
+    case rocksdb_env_delete_file_nanos:
+      return rep->env_delete_file_nanos;
+    case rocksdb_env_create_dir_nanos:
+      return rep->env_create_dir_nanos;
+    case rocksdb_env_create_dir_if_missing_nanos:
+      return rep->env_create_dir_if_missing_nanos;
+    case rocksdb_env_delete_dir_nanos:
+      return rep->env_delete_dir_nanos;
+    case rocksdb_env_get_file_size_nanos:
+      return rep->env_get_file_size_nanos;
+    case rocksdb_env_get_file_modification_time_nanos:
+      return rep->env_get_file_modification_time_nanos;
+    case rocksdb_env_rename_file_nanos:
+      return rep->env_rename_file_nanos;
+    case rocksdb_env_link_file_nanos:
+      return rep->env_link_file_nanos;
+    case rocksdb_env_lock_file_nanos:
+      return rep->env_lock_file_nanos;
+    case rocksdb_env_unlock_file_nanos:
+      return rep->env_unlock_file_nanos;
+    case rocksdb_env_new_logger_nanos:
+      return rep->env_new_logger_nanos;
+    default:
+      break;
+  }
+  return 0;
+}
+
+void rocksdb_perfcontext_destroy(rocksdb_perfcontext_t* context) {
+  delete context;
+}
+
 /*
 TODO:
 DB::OpenForReadOnly
@@ -54,6 +54,9 @@ ColumnFamilyHandleImpl::~ColumnFamilyHandleImpl() {
 #endif  // ROCKSDB_LITE
    // Job id == 0 means that this is not our background process, but rather
    // user thread
+    // Need to hold some shared pointers owned by the initial_cf_options
+    // before final cleaning up finishes.
+    ColumnFamilyOptions initial_cf_options_copy = cfd_->initial_cf_options();
    JobContext job_context(0);
    mutex_->Lock();
    if (cfd_->Unref()) {
@@ -383,7 +386,7 @@ ColumnFamilyData::ColumnFamilyData(
      next_(nullptr),
      prev_(nullptr),
      log_number_(0),
-      flush_reason_(FlushReason::kUnknown),
+      flush_reason_(FlushReason::kOthers),
      column_family_set_(column_family_set),
      pending_flush_(false),
      pending_compaction_(false),
@@ -370,6 +370,10 @@ class ColumnFamilyData {

  bool initialized() const { return initialized_.load(); }

+  const ColumnFamilyOptions& initial_cf_options() {
+    return initial_cf_options_;
+  }
+
  Env::WriteLifeTimeHint CalculateSSTWriteHint(int level);

 private:
@@ -2791,6 +2791,18 @@ TEST_F(ColumnFamilyTest, CompactionSpeedupTwoColumnFamilies) {
  ASSERT_EQ(1, dbfull()->TEST_BGCompactionsAllowed());
 }

+TEST_F(ColumnFamilyTest, CreateAndDestoryOptions) {
+  std::unique_ptr<ColumnFamilyOptions> cfo(new ColumnFamilyOptions());
+  ColumnFamilyHandle* cfh;
+  Open();
+  ASSERT_OK(db_->CreateColumnFamily(*(cfo.get()), "yoyo", &cfh));
+  cfo.reset();
+  ASSERT_OK(db_->Put(WriteOptions(), cfh, "foo", "bar"));
+  ASSERT_OK(db_->Flush(FlushOptions(), cfh));
+  ASSERT_OK(db_->DropColumnFamily(cfh));
+  ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
+}
+
 #ifndef ROCKSDB_LITE
 TEST_F(ColumnFamilyTest, FlushCloseWALFiles) {
  SpecialEnv env(Env::Default());
@@ -74,7 +74,7 @@ Status DBImpl::EnableFileDeletions(bool force) {
 }

 int DBImpl::IsFileDeletionsEnabled() const {
-  return disable_delete_obsolete_files_;
+  return !disable_delete_obsolete_files_;
 }

 Status DBImpl::GetLiveFiles(std::vector<std::string>& ret,
@@ -2145,9 +2145,9 @@ Status DBImpl::DeleteFile(std::string name) {
    status = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
                                    &edit, &mutex_, directories_.GetDbDir());
    if (status.ok()) {
-      InstallSuperVersionAndScheduleWork(
-          cfd, &job_context.superversion_context,
-          *cfd->GetLatestMutableCFOptions());
+      InstallSuperVersionAndScheduleWork(cfd, &job_context.superversion_context,
+                                         *cfd->GetLatestMutableCFOptions(),
+                                         FlushReason::kDeleteFiles);
    }
    FindObsoleteFiles(&job_context, false);
  }  // lock released here
@@ -2229,9 +2229,9 @@ Status DBImpl::DeleteFilesInRanges(ColumnFamilyHandle* column_family,
    status = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
                                    &edit, &mutex_, directories_.GetDbDir());
    if (status.ok()) {
-      InstallSuperVersionAndScheduleWork(
-          cfd, &job_context.superversion_context,
-          *cfd->GetLatestMutableCFOptions());
+      InstallSuperVersionAndScheduleWork(cfd, &job_context.superversion_context,
+                                         *cfd->GetLatestMutableCFOptions(),
+                                         FlushReason::kDeleteFiles);
    }
    for (auto* deleted_file : deleted_files) {
      deleted_file->being_compacted = false;
@@ -2859,8 +2859,8 @@ Status DBImpl::IngestExternalFile(
                                 &mutex_, directories_.GetDbDir());
    }
    if (status.ok()) {
-      InstallSuperVersionAndScheduleWork(cfd, &sv_context,
-                                         *mutable_cf_options);
+      InstallSuperVersionAndScheduleWork(cfd, &sv_context, *mutable_cf_options,
+                                         FlushReason::kExternalFileIngestion);
    }

    // Resume writes to the DB
@@ -223,6 +223,8 @@ class DBImpl : public DB {
  virtual Status SyncWAL() override;

  virtual SequenceNumber GetLatestSequenceNumber() const override;
+  // REQUIRES: joined the main write queue if two_write_queues is disabled, and
+  // the second write queue otherwise.
  virtual void SetLastPublishedSequence(SequenceNumber seq);
  // Returns LastSequence in last_seq_same_as_publish_seq_
  // mode and LastAllocatedSequence otherwise. This is useful when visiblility
@@ -833,7 +835,8 @@ class DBImpl : public DB {

  Status ScheduleFlushes(WriteContext* context);

-  Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context);
+  Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
+                        FlushReason flush_reason = FlushReason::kOthers);

  // Force current memtable contents to be flushed.
  Status FlushMemTable(ColumnFamilyData* cfd, const FlushOptions& options,
@@ -873,7 +876,7 @@ class DBImpl : public DB {
                              size_t seq_inc);

  // Used by WriteImpl to update bg_error_ if paranoid check is enabled.
-  void WriteCallbackStatusCheck(const Status& status);
+  void WriteStatusCheck(const Status& status);

  // Used by WriteImpl to update bg_error_ in case of memtable insert error.
  void MemTableInsertStatusCheck(const Status& memtable_insert_status);
@@ -1359,7 +1362,8 @@ class DBImpl : public DB {
  // state needs flush or compaction.
  void InstallSuperVersionAndScheduleWork(
      ColumnFamilyData* cfd, SuperVersionContext* sv_context,
-      const MutableCFOptions& mutable_cf_options);
+      const MutableCFOptions& mutable_cf_options,
+      FlushReason flush_reason = FlushReason::kOthers);

 #ifndef ROCKSDB_LITE
  using DB::GetPropertiesOfAllTables;
@@ -655,7 +655,7 @@ Status DBImpl::CompactFilesImpl(
  if (status.ok()) {
    InstallSuperVersionAndScheduleWork(
        c->column_family_data(), &job_context->superversion_context,
-       *c->mutable_cf_options());
+        *c->mutable_cf_options(), FlushReason::kManualCompaction);
  }
  c->ReleaseCompactionFiles(s);

@@ -891,7 +891,7 @@ Status DBImpl::Flush(const FlushOptions& flush_options,
  ROCKS_LOG_INFO(immutable_db_options_.info_log, "[%s] Manual flush start.",
                 cfh->GetName().c_str());
  Status s =
-      FlushMemTable(cfh->cfd(), flush_options, FlushReason::kManualCompaction);
+      FlushMemTable(cfh->cfd(), flush_options, FlushReason::kManualFlush);
  ROCKS_LOG_INFO(immutable_db_options_.info_log,
                 "[%s] Manual flush finished, status: %s\n",
                 cfh->GetName().c_str(), s.ToString().c_str());
@@ -1727,7 +1727,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
                                    &mutex_, directories_.GetDbDir());
    InstallSuperVersionAndScheduleWork(
        c->column_family_data(), &job_context->superversion_context,
-        *c->mutable_cf_options());
+        *c->mutable_cf_options(), FlushReason::kAutoCompaction);
    ROCKS_LOG_BUFFER(log_buffer, "[%s] Deleted %d files\n",
                     c->column_family_data()->GetName().c_str(),
                     c->num_input_files(0));
@@ -1774,7 +1774,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
    // Use latest MutableCFOptions
    InstallSuperVersionAndScheduleWork(
        c->column_family_data(), &job_context->superversion_context,
-        *c->mutable_cf_options());
+        *c->mutable_cf_options(), FlushReason::kAutoCompaction);

    VersionStorageInfo::LevelSummaryStorage tmp;
    c->column_family_data()->internal_stats()->IncBytesMoved(c->output_level(),
@@ -1853,7 +1853,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
    if (status.ok()) {
      InstallSuperVersionAndScheduleWork(
          c->column_family_data(), &job_context->superversion_context,
-          *c->mutable_cf_options());
+          *c->mutable_cf_options(), FlushReason::kAutoCompaction);
    }
    *made_progress = true;
  }
@@ -2044,7 +2044,7 @@ bool DBImpl::MCOverlap(ManualCompactionState* m, ManualCompactionState* m1) {

 void DBImpl::InstallSuperVersionAndScheduleWork(
    ColumnFamilyData* cfd, SuperVersionContext* sv_context,
-    const MutableCFOptions& mutable_cf_options) {
+    const MutableCFOptions& mutable_cf_options, FlushReason flush_reason) {
  mutex_.AssertHeld();

  // Update max_total_in_memory_state_
@@ -2063,7 +2063,7 @@ void DBImpl::InstallSuperVersionAndScheduleWork(

  // Whenever we install new SuperVersion, we might need to issue new flushes or
  // compactions.
-  SchedulePendingFlush(cfd, FlushReason::kSuperVersionChange);
+  SchedulePendingFlush(cfd, flush_reason);
  SchedulePendingCompaction(cfd);
  MaybeScheduleFlushOrCompaction();

@@ -133,7 +133,8 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
      for (auto* writer : *(w.write_group)) {
        if (!writer->CallbackFailed() && writer->pre_release_callback) {
          assert(writer->sequence != kMaxSequenceNumber);
-          Status ws = writer->pre_release_callback->Callback(writer->sequence);
+          Status ws = writer->pre_release_callback->Callback(writer->sequence,
+                                                             disable_memtable);
          if (!ws.ok()) {
            status = ws;
            break;
@@ -339,7 +340,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
  PERF_TIMER_START(write_pre_and_post_process_time);

  if (!w.CallbackFailed()) {
-    WriteCallbackStatusCheck(status);
+    WriteStatusCheck(status);
  }

  if (need_log_sync) {
@@ -368,7 +369,8 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
      for (auto* writer : write_group) {
        if (!writer->CallbackFailed() && writer->pre_release_callback) {
          assert(writer->sequence != kMaxSequenceNumber);
-          Status ws = writer->pre_release_callback->Callback(writer->sequence);
+          Status ws = writer->pre_release_callback->Callback(writer->sequence,
+                                                             disable_memtable);
          if (!ws.ok()) {
            status = ws;
            break;
@@ -462,7 +464,7 @@ Status DBImpl::PipelinedWriteImpl(const WriteOptions& write_options,
    }

    if (!w.CallbackFailed()) {
-      WriteCallbackStatusCheck(w.status);
+      WriteStatusCheck(w.status);
    }

    if (need_log_sync) {
@@ -623,13 +625,15 @@ Status DBImpl::WriteImplWALOnly(const WriteOptions& write_options,
  PERF_TIMER_START(write_pre_and_post_process_time);

  if (!w.CallbackFailed()) {
-    WriteCallbackStatusCheck(status);
+    WriteStatusCheck(status);
  }
  if (status.ok()) {
    for (auto* writer : write_group) {
      if (!writer->CallbackFailed() && writer->pre_release_callback) {
        assert(writer->sequence != kMaxSequenceNumber);
-        Status ws = writer->pre_release_callback->Callback(writer->sequence);
+        const bool DISABLE_MEMTABLE = true;
+        Status ws = writer->pre_release_callback->Callback(writer->sequence,
+                                                           DISABLE_MEMTABLE);
        if (!ws.ok()) {
          status = ws;
          break;
@@ -647,7 +651,7 @@ Status DBImpl::WriteImplWALOnly(const WriteOptions& write_options,
  return status;
 }

-void DBImpl::WriteCallbackStatusCheck(const Status& status) {
+void DBImpl::WriteStatusCheck(const Status& status) {
  // Is setting bg_error_ enough here?  This will at least stop
  // compaction and fail any further writes.
  if (immutable_db_options_.paranoid_checks && !status.ok() &&
@@ -1062,7 +1066,8 @@ Status DBImpl::HandleWriteBufferFull(WriteContext* write_context) {
    }
  }
  if (cfd_picked != nullptr) {
-    status = SwitchMemtable(cfd_picked, write_context);
+    status = SwitchMemtable(cfd_picked, write_context,
+                            FlushReason::kWriteBufferFull);
    if (status.ok()) {
      cfd_picked->imm()->FlushRequested();
      SchedulePendingFlush(cfd_picked, FlushReason::kWriteBufferFull);
@@ -1163,7 +1168,7 @@ Status DBImpl::ThrottleLowPriWritesIfNeeded(const WriteOptions& write_options,
 Status DBImpl::ScheduleFlushes(WriteContext* context) {
  ColumnFamilyData* cfd;
  while ((cfd = flush_scheduler_.TakeNextColumnFamily()) != nullptr) {
-    auto status = SwitchMemtable(cfd, context);
+    auto status = SwitchMemtable(cfd, context, FlushReason::kWriteBufferFull);
    if (cfd->Unref()) {
      delete cfd;
    }
@@ -1192,7 +1197,8 @@ void DBImpl::NotifyOnMemTableSealed(ColumnFamilyData* /*cfd*/,

 // REQUIRES: mutex_ is held
 // REQUIRES: this thread is currently at the front of the writer queue
-Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
+Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
+                              FlushReason flush_reason) {
  mutex_.AssertHeld();
  WriteThread::Writer nonmem_w;
  if (two_write_queues_) {
@@ -1360,7 +1366,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
  new_mem->Ref();
  cfd->SetMemtable(new_mem);
  InstallSuperVersionAndScheduleWork(cfd, &context->superversion_context,
-                                     mutable_cf_options);
+                                     mutable_cf_options, flush_reason);
  if (two_write_queues_) {
    nonmem_write_thread_.ExitUnbatched(&nonmem_w);
  }
@@ -69,27 +69,27 @@ TEST_F(DBPropertiesTest, Empty) {
    ASSERT_OK(db_->DisableFileDeletions());
    ASSERT_TRUE(
        dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
-    ASSERT_EQ("1", num);
+    ASSERT_EQ("0", num);

    ASSERT_OK(db_->DisableFileDeletions());
    ASSERT_TRUE(
        dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
-    ASSERT_EQ("2", num);
+    ASSERT_EQ("0", num);

    ASSERT_OK(db_->DisableFileDeletions());
    ASSERT_TRUE(
        dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
-    ASSERT_EQ("3", num);
+    ASSERT_EQ("0", num);

    ASSERT_OK(db_->EnableFileDeletions(false));
    ASSERT_TRUE(
        dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
-    ASSERT_EQ("2", num);
+    ASSERT_EQ("0", num);

    ASSERT_OK(db_->EnableFileDeletions());
    ASSERT_TRUE(
        dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
-    ASSERT_EQ("0", num);
+    ASSERT_EQ("1", num);
  } while (ChangeOptions());
 }

@@ -80,6 +80,22 @@ TEST_P(DBWriteTest, IOErrorOnWALWritePropagateToWriteThreadFollower) {
  Close();
 }

+TEST_P(DBWriteTest, IOErrorOnWALWriteTriggersReadOnlyMode) {
+  std::unique_ptr<FaultInjectionTestEnv> mock_env(
+      new FaultInjectionTestEnv(Env::Default()));
+  Options options = GetOptions();
+  options.env = mock_env.get();
+  Reopen(options);
+  for (int i = 0; i < 2; i++) {
+    // Forcibly fail WAL write for the first Put only. Subsequent Puts should
+    // fail due to read-only mode
+    mock_env->SetFilesystemActive(i != 0);
+    ASSERT_FALSE(Put("key" + ToString(i), "value").ok());
+  }
+  // Close before mock_env destruct.
+  Close();
+}
+
 INSTANTIATE_TEST_CASE_P(DBWriteTestInstance, DBWriteTest,
                        testing::Values(DBTestBase::kDefault,
                                        DBTestBase::kConcurrentWALWrites,
@@ -106,45 +106,6 @@ const char* InternalKeyComparator::Name() const {
  return name_.c_str();
 }

-int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
-  // Order by:
-  //    increasing user key (according to user-supplied comparator)
-  //    decreasing sequence number
-  //    decreasing type (though sequence# should be enough to disambiguate)
-  int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
-  PERF_COUNTER_ADD(user_key_comparison_count, 1);
-  if (r == 0) {
-    const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);
-    const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);
-    if (anum > bnum) {
-      r = -1;
-    } else if (anum < bnum) {
-      r = +1;
-    }
-  }
-  return r;
-}
-
-int InternalKeyComparator::CompareKeySeq(const Slice& akey,
-                                         const Slice& bkey) const {
-  // Order by:
-  //    increasing user key (according to user-supplied comparator)
-  //    decreasing sequence number
-  int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
-  PERF_COUNTER_ADD(user_key_comparison_count, 1);
-  if (r == 0) {
-    // Shift the number to exclude the last byte which contains the value type
-    const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8) >> 8;
-    const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8) >> 8;
-    if (anum > bnum) {
-      r = -1;
-    } else if (anum < bnum) {
-      r = +1;
-    }
-  }
-  return r;
-}
-
 int InternalKeyComparator::Compare(const ParsedInternalKey& a,
                                   const ParsedInternalKey& b) const {
  // Order by:
@@ -11,6 +11,7 @@
 #include <stdio.h>
 #include <string>
 #include <utility>
+#include "monitoring/perf_context_imp.h"
 #include "rocksdb/comparator.h"
 #include "rocksdb/db.h"
 #include "rocksdb/filter_policy.h"
@@ -607,4 +608,46 @@ struct RangeTombstone {
  }
 };

+inline
+int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
+  // Order by:
+  //    increasing user key (according to user-supplied comparator)
+  //    decreasing sequence number
+  //    decreasing type (though sequence# should be enough to disambiguate)
+  int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
+  PERF_COUNTER_ADD(user_key_comparison_count, 1);
+  if (r == 0) {
+    const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);
+    const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);
+    if (anum > bnum) {
+      r = -1;
+    } else if (anum < bnum) {
+      r = +1;
+    }
+  }
+  return r;
+}
+
+inline
+int InternalKeyComparator::CompareKeySeq(const Slice& akey,
+                                         const Slice& bkey) const {
+  // Order by:
+  //    increasing user key (according to user-supplied comparator)
+  //    decreasing sequence number
+  int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
+  PERF_COUNTER_ADD(user_key_comparison_count, 1);
+  if (r == 0) {
+    // Shift the number to exclude the last byte which contains the value type
+    const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8) >> 8;
+    const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8) >> 8;
+    if (anum > bnum) {
+      r = -1;
+    } else if (anum < bnum) {
+      r = +1;
+    }
+  }
+  return r;
+}
+
+
 }  // namespace rocksdb
@@ -510,6 +510,9 @@ Status ExternalSstFileIngestionJob::AssignGlobalSeqnoForIngestedFile(
  std::string seqno_val;
  PutFixed64(&seqno_val, seqno);
  status = rwfile->Write(file_to_ingest->global_seqno_offset, seqno_val);
+  if (status.ok()) {
+    status = rwfile->Fsync();
+  }
  if (status.ok()) {
    file_to_ingest->assigned_seqno = seqno;
  }
@@ -56,8 +56,8 @@ namespace rocksdb {

 const char* GetFlushReasonString (FlushReason flush_reason) {
  switch (flush_reason) {
-    case FlushReason::kUnknown:
-      return "Unknown";
+    case FlushReason::kOthers:
+      return "Other Reasons";
    case FlushReason::kGetLiveFiles:
      return "Get Live Files";
    case FlushReason::kShutDown:
@@ -72,8 +72,12 @@ const char* GetFlushReasonString (FlushReason flush_reason) {
      return "Write Buffer Full";
    case FlushReason::kTest:
      return "Test";
-    case FlushReason::kSuperVersionChange:
-      return "SuperVersion Change";
+    case FlushReason::kDeleteFiles:
+      return "Delete Files";
+    case FlushReason::kAutoCompaction:
+      return "Auto Compaction";
+    case FlushReason::kManualFlush:
+      return "Manual Flush";
    default:
      return "Invalid";
  }
@@ -229,7 +229,7 @@ int MemTable::KeyComparator::operator()(const char* prefix_len_key1,
 }

 int MemTable::KeyComparator::operator()(const char* prefix_len_key,
-                                        const Slice& key)
+                                        const KeyComparator::DecodedType& key)
    const {
  // Internal keys are encoded as length-prefixed strings.
  Slice a = GetLengthPrefixedSlice(prefix_len_key);
@@ -64,7 +64,7 @@ struct MemTablePostProcessInfo {
 };

 // Note:  Many of the methods in this class have comments indicating that
-// external synchromization is required as these methods are not thread-safe.
+// external synchronization is required as these methods are not thread-safe.
 // It is up to higher layers of code to decide how to prevent concurrent
 // invokation of these methods.  This is usually done by acquiring either
 // the db mutex or the single writer thread.
@@ -84,7 +84,7 @@ class MemTable {
    virtual int operator()(const char* prefix_len_key1,
                           const char* prefix_len_key2) const override;
    virtual int operator()(const char* prefix_len_key,
-                           const Slice& key) const override;
+                           const DecodedType& key) const override;
  };

  // MemTables are reference counted.  The initial reference count
@@ -24,7 +24,9 @@ class PreReleaseCallback {
  // propagated to all the writers in the write group.
  // seq is the sequence number that is used for this write and will be
  // released.
-  virtual Status Callback(SequenceNumber seq) = 0;
+  // is_mem_disabled is currently used for debugging purposes to assert that
+  // the callback is done from the right write queue.
+  virtual Status Callback(SequenceNumber seq, const bool is_mem_disabled) = 0;
 };

 }  //  namespace rocksdb
@@ -536,4 +536,11 @@ bool RangeDelAggregator::IsEmpty() {
  return true;
 }

+bool RangeDelAggregator::AddFile(uint64_t file_number) {
+  if (added_files_ == nullptr) {
+    added_files_.reset(new std::set<uint64_t>());
+  }
+  return added_files_->emplace(file_number).second;
+}
+
 }  // namespace rocksdb
@@ -6,6 +6,7 @@
 #pragma once

 #include <map>
+#include <set>
 #include <string>
 #include <vector>

@@ -140,6 +141,7 @@ class RangeDelAggregator {
                    CompactionIterationStats* range_del_out_stats = nullptr,
                    bool bottommost_level = false);
  bool IsEmpty();
+  bool AddFile(uint64_t file_number);

 private:
  // Maps tombstone user start key -> tombstone object
@@ -180,6 +182,10 @@ class RangeDelAggregator {
  const InternalKeyComparator& icmp_;
  // collapse range deletions so they're binary searchable
  const bool collapse_deletions_;
+
+  // Record files whose tombstones have been added, to avoid duplicate adding.
+  // Same as rep_, we initializes it lazily.
+  std::unique_ptr<std::set<uint64_t>> added_files_;
 };

 }  // namespace rocksdb
@@ -247,13 +247,15 @@ InternalIterator* TableCache::NewIterator(
    }
  }
  if (s.ok() && range_del_agg != nullptr && !options.ignore_range_deletions) {
-    std::unique_ptr<InternalIterator> range_del_iter(
-        table_reader->NewRangeTombstoneIterator(options));
-    if (range_del_iter != nullptr) {
-      s = range_del_iter->status();
-    }
-    if (s.ok()) {
-      s = range_del_agg->AddTombstones(std::move(range_del_iter));
+    if (range_del_agg->AddFile(fd.GetNumber())) {
+      std::unique_ptr<InternalIterator> range_del_iter(
+          table_reader->NewRangeTombstoneIterator(options));
+      if (range_del_iter != nullptr) {
+        s = range_del_iter->status();
+      }
+      if (s.ok()) {
+        s = range_del_agg->AddTombstones(std::move(range_del_iter));
+      }
    }
  }

@@ -294,7 +294,8 @@ TEST_F(WriteCallbackTest, WriteWithCallbackTest) {
                   public:
                    PublishSeqCallback(DBImpl* db_impl_in)
                        : db_impl_(db_impl_in) {}
-                    virtual Status Callback(SequenceNumber last_seq) {
+                    virtual Status Callback(SequenceNumber last_seq,
+                                            const bool /*not used*/) override {
                      db_impl_->SetLastPublishedSequence(last_seq);
                      return Status::OK();
                    }
@@ -113,6 +113,7 @@ typedef struct rocksdb_envoptions_t      rocksdb_envoptions_t;
 typedef struct rocksdb_ingestexternalfileoptions_t rocksdb_ingestexternalfileoptions_t;
 typedef struct rocksdb_sstfilewriter_t   rocksdb_sstfilewriter_t;
 typedef struct rocksdb_ratelimiter_t     rocksdb_ratelimiter_t;
+typedef struct rocksdb_perfcontext_t     rocksdb_perfcontext_t;
 typedef struct rocksdb_pinnableslice_t rocksdb_pinnableslice_t;
 typedef struct rocksdb_transactiondb_options_t rocksdb_transactiondb_options_t;
 typedef struct rocksdb_transactiondb_t rocksdb_transactiondb_t;
@@ -969,6 +970,97 @@ extern ROCKSDB_LIBRARY_API rocksdb_ratelimiter_t* rocksdb_ratelimiter_create(
    int64_t rate_bytes_per_sec, int64_t refill_period_us, int32_t fairness);
 extern ROCKSDB_LIBRARY_API void rocksdb_ratelimiter_destroy(rocksdb_ratelimiter_t*);

+/* PerfContext */
+enum {
+  rocksdb_uninitialized = 0,
+  rocksdb_disable = 1,
+  rocksdb_enable_count = 2,
+  rocksdb_enable_time_except_for_mutex = 3,
+  rocksdb_enable_time = 4,
+  rocksdb_out_of_bounds = 5
+};
+
+enum {
+  rocksdb_user_key_comparison_count = 0,
+  rocksdb_block_cache_hit_count,
+  rocksdb_block_read_count,
+  rocksdb_block_read_byte,
+  rocksdb_block_read_time,
+  rocksdb_block_checksum_time,
+  rocksdb_block_decompress_time,
+  rocksdb_get_read_bytes,
+  rocksdb_multiget_read_bytes,
+  rocksdb_iter_read_bytes,
+  rocksdb_internal_key_skipped_count,
+  rocksdb_internal_delete_skipped_count,
+  rocksdb_internal_recent_skipped_count,
+  rocksdb_internal_merge_count,
+  rocksdb_get_snapshot_time,
+  rocksdb_get_from_memtable_time,
+  rocksdb_get_from_memtable_count,
+  rocksdb_get_post_process_time,
+  rocksdb_get_from_output_files_time,
+  rocksdb_seek_on_memtable_time,
+  rocksdb_seek_on_memtable_count,
+  rocksdb_next_on_memtable_count,
+  rocksdb_prev_on_memtable_count,
+  rocksdb_seek_child_seek_time,
+  rocksdb_seek_child_seek_count,
+  rocksdb_seek_min_heap_time,
+  rocksdb_seek_max_heap_time,
+  rocksdb_seek_internal_seek_time,
+  rocksdb_find_next_user_entry_time,
+  rocksdb_write_wal_time,
+  rocksdb_write_memtable_time,
+  rocksdb_write_delay_time,
+  rocksdb_write_pre_and_post_process_time,
+  rocksdb_db_mutex_lock_nanos,
+  rocksdb_db_condition_wait_nanos,
+  rocksdb_merge_operator_time_nanos,
+  rocksdb_read_index_block_nanos,
+  rocksdb_read_filter_block_nanos,
+  rocksdb_new_table_block_iter_nanos,
+  rocksdb_new_table_iterator_nanos,
+  rocksdb_block_seek_nanos,
+  rocksdb_find_table_nanos,
+  rocksdb_bloom_memtable_hit_count,
+  rocksdb_bloom_memtable_miss_count,
+  rocksdb_bloom_sst_hit_count,
+  rocksdb_bloom_sst_miss_count,
+  rocksdb_key_lock_wait_time,
+  rocksdb_key_lock_wait_count,
+  rocksdb_env_new_sequential_file_nanos,
+  rocksdb_env_new_random_access_file_nanos,
+  rocksdb_env_new_writable_file_nanos,
+  rocksdb_env_reuse_writable_file_nanos,
+  rocksdb_env_new_random_rw_file_nanos,
+  rocksdb_env_new_directory_nanos,
+  rocksdb_env_file_exists_nanos,
+  rocksdb_env_get_children_nanos,
+  rocksdb_env_get_children_file_attributes_nanos,
+  rocksdb_env_delete_file_nanos,
+  rocksdb_env_create_dir_nanos,
+  rocksdb_env_create_dir_if_missing_nanos,
+  rocksdb_env_delete_dir_nanos,
+  rocksdb_env_get_file_size_nanos,
+  rocksdb_env_get_file_modification_time_nanos,
+  rocksdb_env_rename_file_nanos,
+  rocksdb_env_link_file_nanos,
+  rocksdb_env_lock_file_nanos,
+  rocksdb_env_unlock_file_nanos,
+  rocksdb_env_new_logger_nanos,
+  rocksdb_total_metric_count = 68
+};
+
+extern ROCKSDB_LIBRARY_API void rocksdb_set_perf_level(int);
+extern ROCKSDB_LIBRARY_API rocksdb_perfcontext_t* rocksdb_perfcontext_create();
+extern ROCKSDB_LIBRARY_API void rocksdb_perfcontext_reset(
+    rocksdb_perfcontext_t* context);
+extern ROCKSDB_LIBRARY_API char* rocksdb_perfcontext_report(
+    rocksdb_perfcontext_t* context, unsigned char exclude_zero_counters);
+extern ROCKSDB_LIBRARY_API uint64_t rocksdb_perfcontext_metric(
+    rocksdb_perfcontext_t* context, int metric);
+
 /* Compaction Filter */

 extern ROCKSDB_LIBRARY_API rocksdb_compactionfilter_t*
@@ -83,7 +83,7 @@ enum class CompactionReason {
 };

 enum class FlushReason : int {
-  kUnknown = 0x00,
+  kOthers = 0x00,
  kGetLiveFiles = 0x01,
  kShutDown = 0x02,
  kExternalFileIngestion = 0x03,
@@ -91,7 +91,9 @@ enum class FlushReason : int {
  kWriteBufferManager = 0x05,
  kWriteBufferFull = 0x06,
  kTest = 0x07,
-  kSuperVersionChange = 0x08,
+  kDeleteFiles = 0x08,
+  kAutoCompaction = 0x09,
+  kManualFlush = 0x0a,
 };

 enum class BackgroundErrorReason {
@@ -39,24 +39,34 @@
 #include <stdexcept>
 #include <stdint.h>
 #include <stdlib.h>
+#include <rocksdb/slice.h>

 namespace rocksdb {

 class Arena;
 class Allocator;
 class LookupKey;
-class Slice;
 class SliceTransform;
 class Logger;

 typedef void* KeyHandle;

+extern Slice GetLengthPrefixedSlice(const char* data);
+
 class MemTableRep {
 public:
  // KeyComparator provides a means to compare keys, which are internal keys
  // concatenated with values.
  class KeyComparator {
   public:
+    typedef rocksdb::Slice DecodedType;
+
+    virtual DecodedType decode_key(const char* key) const {
+      // The format of key is frozen and can be terated as a part of the API
+      // contract. Refer to MemTable::Add for details.
+      return GetLengthPrefixedSlice(key);
+    }
+
    // Compare a and b. Return a negative value if a is less than b, 0 if they
    // are equal, and a positive value if a is greater than b
    virtual int operator()(const char* prefix_len_key1,
@@ -96,10 +96,14 @@ class SstFileManager {
 // @param max_trash_db_ratio: If the trash size constitutes for more than this
 //    fraction of the total DB size we will start deleting new files passed to
 //    DeleteScheduler immediately
+// @param bytes_max_delete_chunk: if a single file is larger than delete chunk,
+//    ftruncate the file by this size each time, rather than dropping the whole
+//    file. 0 means to always delete the whole file.
 extern SstFileManager* NewSstFileManager(
    Env* env, std::shared_ptr<Logger> info_log = nullptr,
    std::string trash_dir = "", int64_t rate_bytes_per_sec = 0,
    bool delete_existing_trash = true, Status* status = nullptr,
-    double max_trash_db_ratio = 0.25);
+    double max_trash_db_ratio = 0.25,
+    uint64_t bytes_max_delete_chunk = 64 * 1024 * 1024);

 }  // namespace rocksdb
@@ -5,7 +5,7 @@
 #pragma once

 #define ROCKSDB_MAJOR 5
-#define ROCKSDB_MINOR 12
+#define ROCKSDB_MINOR 13
 #define ROCKSDB_PATCH 0

 // Do not use these. We made the mistake of declaring macros starting with
@@ -38,13 +38,14 @@ jlong Java_org_rocksdb_PlainTableConfig_newTableFactoryHandle(
 /*
 * Class:     org_rocksdb_BlockBasedTableConfig
 * Method:    newTableFactoryHandle
- * Signature: (ZJIJIIZIZZZJIBBI)J
+ * Signature: (ZJIJJIIZIZZZJIBBI)J
 */
 jlong Java_org_rocksdb_BlockBasedTableConfig_newTableFactoryHandle(
-    JNIEnv* env, jobject jobj, jboolean no_block_cache, jlong block_cache_size,
-    jint block_cache_num_shardbits, jlong block_size, jint block_size_deviation,
-    jint block_restart_interval, jboolean whole_key_filtering,
-    jlong jfilterPolicy, jboolean cache_index_and_filter_blocks,
+    JNIEnv *env, jobject jobj, jboolean no_block_cache, jlong block_cache_size,
+    jint block_cache_num_shardbits, jlong jblock_cache, jlong block_size,
+    jint block_size_deviation, jint block_restart_interval,
+    jboolean whole_key_filtering, jlong jfilter_policy,
+    jboolean cache_index_and_filter_blocks,
    jboolean pin_l0_filter_and_index_blocks_in_cache,
    jboolean hash_index_allow_collision, jlong block_cache_compressed_size,
    jint block_cache_compressd_num_shard_bits, jbyte jchecksum_type,
@@ -52,22 +53,28 @@ jlong Java_org_rocksdb_BlockBasedTableConfig_newTableFactoryHandle(
  rocksdb::BlockBasedTableOptions options;
  options.no_block_cache = no_block_cache;

-  if (!no_block_cache && block_cache_size > 0) {
-    if (block_cache_num_shardbits > 0) {
-      options.block_cache =
-          rocksdb::NewLRUCache(block_cache_size, block_cache_num_shardbits);
-    } else {
-      options.block_cache = rocksdb::NewLRUCache(block_cache_size);
+  if (!no_block_cache) {
+    if (jblock_cache > 0) {
+      std::shared_ptr<rocksdb::Cache> *pCache =
+          reinterpret_cast<std::shared_ptr<rocksdb::Cache> *>(jblock_cache);
+      options.block_cache = *pCache;
+    } else if (block_cache_size > 0) {
+      if (block_cache_num_shardbits > 0) {
+        options.block_cache =
+            rocksdb::NewLRUCache(block_cache_size, block_cache_num_shardbits);
+      } else {
+        options.block_cache = rocksdb::NewLRUCache(block_cache_size);
+      }
    }
  }
  options.block_size = block_size;
  options.block_size_deviation = block_size_deviation;
  options.block_restart_interval = block_restart_interval;
  options.whole_key_filtering = whole_key_filtering;
-  if (jfilterPolicy > 0) {
+  if (jfilter_policy > 0) {
    std::shared_ptr<rocksdb::FilterPolicy> *pFilterPolicy =
        reinterpret_cast<std::shared_ptr<rocksdb::FilterPolicy> *>(
-            jfilterPolicy);
+            jfilter_policy);
    options.filter_policy = *pFilterPolicy;
  }
  options.cache_index_and_filter_blocks = cache_index_and_filter_blocks;
@@ -15,6 +15,7 @@ public class BlockBasedTableConfig extends TableFormatConfig {
    noBlockCache_ = false;
    blockCacheSize_ = 8 * 1024 * 1024;
    blockCacheNumShardBits_ = 0;
+    blockCache_ = null;
    blockSize_ = 4 * 1024;
    blockSizeDeviation_ = 10;
    blockRestartInterval_ = 16;
@@ -71,6 +72,24 @@ public class BlockBasedTableConfig extends TableFormatConfig {
    return blockCacheSize_;
  }

+  /**
+   * Use the specified cache for blocks.
+   * When not null this take precedence even if the user sets a block cache size.
+   *
+   * {@link org.rocksdb.Cache} should not be disposed before options instances
+   * using this cache is disposed.
+   *
+   * {@link org.rocksdb.Cache} instance can be re-used in multiple options
+   * instances.
+   *
+   * @param cache {@link org.rocksdb.Cache} Cache java instance (e.g. LRUCache).
+   * @return the reference to the current config.
+   */
+  public BlockBasedTableConfig setBlockCache(final Cache cache) {
+    blockCache_ = cache;
+    return this;
+  }
+
  /**
   * Controls the number of shards for the block cache.
   * This is applied only if cacheSize is set to non-negative.
@@ -413,25 +432,25 @@ public class BlockBasedTableConfig extends TableFormatConfig {
      filterHandle = filter_.nativeHandle_;
    }

-    return newTableFactoryHandle(noBlockCache_, blockCacheSize_,
-        blockCacheNumShardBits_, blockSize_, blockSizeDeviation_,
-        blockRestartInterval_, wholeKeyFiltering_,
-        filterHandle, cacheIndexAndFilterBlocks_,
-        pinL0FilterAndIndexBlocksInCache_,
-        hashIndexAllowCollision_, blockCacheCompressedSize_,
-        blockCacheCompressedNumShardBits_,
-        checksumType_.getValue(), indexType_.getValue(),
+    long blockCacheHandle = 0;
+    if (blockCache_ != null) {
+      blockCacheHandle = blockCache_.nativeHandle_;
+    }
+
+    return newTableFactoryHandle(noBlockCache_, blockCacheSize_, blockCacheNumShardBits_,
+        blockCacheHandle, blockSize_, blockSizeDeviation_, blockRestartInterval_,
+        wholeKeyFiltering_, filterHandle, cacheIndexAndFilterBlocks_,
+        pinL0FilterAndIndexBlocksInCache_, hashIndexAllowCollision_, blockCacheCompressedSize_,
+        blockCacheCompressedNumShardBits_, checksumType_.getValue(), indexType_.getValue(),
        formatVersion_);
  }

-  private native long newTableFactoryHandle(
-      boolean noBlockCache, long blockCacheSize, int blockCacheNumShardBits,
-      long blockSize, int blockSizeDeviation, int blockRestartInterval,
-      boolean wholeKeyFiltering, long filterPolicyHandle,
+  private native long newTableFactoryHandle(boolean noBlockCache, long blockCacheSize,
+      int blockCacheNumShardBits, long blockCacheHandle, long blockSize, int blockSizeDeviation,
+      int blockRestartInterval, boolean wholeKeyFiltering, long filterPolicyHandle,
      boolean cacheIndexAndFilterBlocks, boolean pinL0FilterAndIndexBlocksInCache,
      boolean hashIndexAllowCollision, long blockCacheCompressedSize,
-      int blockCacheCompressedNumShardBits, byte checkSumType,
-      byte indexType, int formatVersion);
+      int blockCacheCompressedNumShardBits, byte checkSumType, byte indexType, int formatVersion);

  private boolean cacheIndexAndFilterBlocks_;
  private boolean pinL0FilterAndIndexBlocksInCache_;
@@ -442,6 +461,7 @@ public class BlockBasedTableConfig extends TableFormatConfig {
  private long blockSize_;
  private long blockCacheSize_;
  private int blockCacheNumShardBits_;
+  private Cache blockCache_;
  private long blockCacheCompressedSize_;
  private int blockCacheCompressedNumShardBits_;
  private int blockSizeDeviation_;
@@ -6,7 +6,11 @@
 package org.rocksdb;

 import org.junit.ClassRule;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.nio.charset.StandardCharsets;

 import static org.assertj.core.api.Assertions.assertThat;

@@ -16,6 +20,8 @@ public class BlockBasedTableConfigTest {
  public static final RocksMemoryResource rocksMemoryResource =
      new RocksMemoryResource();

+  @Rule public TemporaryFolder dbFolder = new TemporaryFolder();
+
  @Test
  public void noBlockCache() {
    BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig();
@@ -31,6 +37,31 @@ public class BlockBasedTableConfigTest {
        isEqualTo(8 * 1024);
  }

+  @Test
+  public void sharedBlockCache() throws RocksDBException {
+    try (final Cache cache = new LRUCache(8 * 1024 * 1024);
+         final Statistics statistics = new Statistics()) {
+      for (int shard = 0; shard < 8; shard++) {
+        try (final Options options =
+                 new Options()
+                     .setCreateIfMissing(true)
+                     .setStatistics(statistics)
+                     .setTableFormatConfig(new BlockBasedTableConfig().setBlockCache(cache));
+             final RocksDB db =
+                 RocksDB.open(options, dbFolder.getRoot().getAbsolutePath() + "/" + shard)) {
+          final byte[] key = "some-key".getBytes(StandardCharsets.UTF_8);
+          final byte[] value = "some-value".getBytes(StandardCharsets.UTF_8);
+
+          db.put(key, value);
+          db.flush(new FlushOptions());
+          db.get(key);
+
+          assertThat(statistics.getTickerCount(TickerType.BLOCK_CACHE_ADD)).isEqualTo(shard + 1);
+        }
+      }
+    }
+  }
+
  @Test
  public void blockSizeDeviation() {
    BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig();
@@ -148,6 +179,14 @@ public class BlockBasedTableConfigTest {
    }
  }

+  @Test
+  public void blockBasedTableWithBlockCache() {
+    try (final Options options = new Options().setTableFormatConfig(
+             new BlockBasedTableConfig().setBlockCache(new LRUCache(17 * 1024 * 1024)))) {
+      assertThat(options.tableFactoryName()).isEqualTo("BlockBasedTable");
+    }
+  }
+
  @Test
  public void blockBasedTableFormatVersion() {
    BlockBasedTableConfig config = new BlockBasedTableConfig();
@@ -45,9 +45,12 @@
 #include <stdlib.h>
 #include <algorithm>
 #include <atomic>
+#include <type_traits>
 #include "port/likely.h"
 #include "port/port.h"
+#include "rocksdb/slice.h"
 #include "util/allocator.h"
+#include "util/coding.h"
 #include "util/random.h"

 namespace rocksdb {
@@ -59,6 +62,9 @@ class InlineSkipList {
  struct Splice;

 public:
+  using DecodedKey = \
+    typename std::remove_reference<Comparator>::type::DecodedType;
+
  static const uint16_t kMaxPossibleHeight = 32;

  // Create a new InlineSkipList object that will use "cmp" for comparing
@@ -212,6 +218,7 @@ class InlineSkipList {
  // Return true if key is greater than the data stored in "n".  Null n
  // is considered infinite.  n should not be head_.
  bool KeyIsAfterNode(const char* key, Node* n) const;
+  bool KeyIsAfterNode(const DecodedKey& key, Node* n) const;

  // Returns the earliest node with a key >= key.
  // Return nullptr if there is no such node.
@@ -241,12 +248,12 @@ class InlineSkipList {
  // a node that is after the key.  after should be nullptr if a good after
  // node isn't conveniently available.
  template<bool prefetch_before>
-  void FindSpliceForLevel(const char* key, Node* before, Node* after, int level,
+  void FindSpliceForLevel(const DecodedKey& key, Node* before, Node* after, int level,
                          Node** out_prev, Node** out_next);

  // Recomputes Splice levels from highest_level (inclusive) down to
  // lowest_level (inclusive).
-  void RecomputeSpliceLevels(const char* key, Splice* splice,
+  void RecomputeSpliceLevels(const DecodedKey& key, Splice* splice,
                             int recompute_level);

  // No copying allowed
@@ -435,6 +442,14 @@ bool InlineSkipList<Comparator>::KeyIsAfterNode(const char* key,
  return (n != nullptr) && (compare_(n->Key(), key) < 0);
 }

+template <class Comparator>
+bool InlineSkipList<Comparator>::KeyIsAfterNode(const DecodedKey& key,
+                                                Node* n) const {
+  // nullptr n is considered infinite
+  assert(n != head_);
+  return (n != nullptr) && (compare_(n->Key(), key) < 0);
+}
+
 template <class Comparator>
 typename InlineSkipList<Comparator>::Node*
 InlineSkipList<Comparator>::FindGreaterOrEqual(const char* key) const {
@@ -446,6 +461,7 @@ InlineSkipList<Comparator>::FindGreaterOrEqual(const char* key) const {
  Node* x = head_;
  int level = GetMaxHeight() - 1;
  Node* last_bigger = nullptr;
+  const DecodedKey key_decoded = compare_.decode_key(key);
  while (true) {
    Node* next = x->Next(level);
    if (next != nullptr) {
@@ -454,10 +470,10 @@ InlineSkipList<Comparator>::FindGreaterOrEqual(const char* key) const {
    // Make sure the lists are sorted
    assert(x == head_ || next == nullptr || KeyIsAfterNode(next->Key(), x));
    // Make sure we haven't overshot during our search
-    assert(x == head_ || KeyIsAfterNode(key, x));
+    assert(x == head_ || KeyIsAfterNode(key_decoded, x));
    int cmp = (next == nullptr || next == last_bigger)
                  ? 1
-                  : compare_(next->Key(), key);
+                  : compare_(next->Key(), key_decoded);
    if (cmp == 0 || (cmp > 0 && level == 0)) {
      return next;
    } else if (cmp < 0) {
@@ -487,6 +503,7 @@ InlineSkipList<Comparator>::FindLessThan(const char* key, Node** prev,
  Node* x = root;
  // KeyIsAfter(key, last_not_after) is definitely false
  Node* last_not_after = nullptr;
+  const DecodedKey key_decoded = compare_.decode_key(key);
  while (true) {
    assert(x != nullptr);
    Node* next = x->Next(level);
@@ -494,8 +511,8 @@ InlineSkipList<Comparator>::FindLessThan(const char* key, Node** prev,
      PREFETCH(next->Next(level), 0, 1);
    }
    assert(x == head_ || next == nullptr || KeyIsAfterNode(next->Key(), x));
-    assert(x == head_ || KeyIsAfterNode(key, x));
-    if (next != last_not_after && KeyIsAfterNode(key, next)) {
+    assert(x == head_ || KeyIsAfterNode(key_decoded, x));
+    if (next != last_not_after && KeyIsAfterNode(key_decoded, next)) {
      // Keep searching in this list
      assert(next != nullptr);
      x = next;
@@ -540,13 +557,14 @@ uint64_t InlineSkipList<Comparator>::EstimateCount(const char* key) const {

  Node* x = head_;
  int level = GetMaxHeight() - 1;
+  const DecodedKey key_decoded = compare_.decode_key(key);
  while (true) {
-    assert(x == head_ || compare_(x->Key(), key) < 0);
+    assert(x == head_ || compare_(x->Key(), key_decoded) < 0);
    Node* next = x->Next(level);
    if (next != nullptr) {
      PREFETCH(next->Next(level), 0, 1);
    }
-    if (next == nullptr || compare_(next->Key(), key) >= 0) {
+    if (next == nullptr || compare_(next->Key(), key_decoded) >= 0) {
      if (level == 0) {
        return count;
      } else {
@@ -654,7 +672,7 @@ bool InlineSkipList<Comparator>::InsertWithHint(const char* key, void** hint) {

 template <class Comparator>
 template <bool prefetch_before>
-void InlineSkipList<Comparator>::FindSpliceForLevel(const char* key,
+void InlineSkipList<Comparator>::FindSpliceForLevel(const DecodedKey& key,
                                                    Node* before, Node* after,
                                                    int level, Node** out_prev,
                                                    Node** out_next) {
@@ -682,7 +700,7 @@ void InlineSkipList<Comparator>::FindSpliceForLevel(const char* key,
 }

 template <class Comparator>
-void InlineSkipList<Comparator>::RecomputeSpliceLevels(const char* key,
+void InlineSkipList<Comparator>::RecomputeSpliceLevels(const DecodedKey& key,
                                                       Splice* splice,
                                                       int recompute_level) {
  assert(recompute_level > 0);
@@ -698,6 +716,7 @@ template <bool UseCAS>
 bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
                                        bool allow_partial_splice_fix) {
  Node* x = reinterpret_cast<Node*>(const_cast<char*>(key)) - 1;
+  const DecodedKey key_decoded = compare_.decode_key(key);
  int height = x->UnstashHeight();
  assert(height >= 1 && height <= kMaxHeight_);

@@ -765,7 +784,8 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
        // our chances of success.
        ++recompute_height;
      } else if (splice->prev_[recompute_height] != head_ &&
-                 !KeyIsAfterNode(key, splice->prev_[recompute_height])) {
+                 !KeyIsAfterNode(key_decoded,
+                                 splice->prev_[recompute_height])) {
        // key is from before splice
        if (allow_partial_splice_fix) {
          // skip all levels with the same node without more comparisons
@@ -777,7 +797,8 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
          // we're pessimistic, recompute everything
          recompute_height = max_height;
        }
-      } else if (KeyIsAfterNode(key, splice->next_[recompute_height])) {
+      } else if (KeyIsAfterNode(key_decoded,
+                                splice->next_[recompute_height])) {
        // key is from after splice
        if (allow_partial_splice_fix) {
          Node* bad = splice->next_[recompute_height];
@@ -795,7 +816,7 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
  }
  assert(recompute_height <= max_height);
  if (recompute_height > 0) {
-    RecomputeSpliceLevels(key, splice, recompute_height);
+    RecomputeSpliceLevels(key_decoded, splice, recompute_height);
  }

  bool splice_is_valid = true;
@@ -827,8 +848,8 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
        // search, because it should be unlikely that lots of nodes have
        // been inserted between prev[i] and next[i]. No point in using
        // next[i] as the after hint, because we know it is stale.
-        FindSpliceForLevel<false>(key, splice->prev_[i], nullptr, i, &splice->prev_[i],
-                           &splice->next_[i]);
+        FindSpliceForLevel<false>(key_decoded, splice->prev_[i], nullptr, i,
+                                  &splice->prev_[i], &splice->next_[i]);

        // Since we've narrowed the bracket for level i, we might have
        // violated the Splice constraint between i and i-1.  Make sure
@@ -842,8 +863,8 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
    for (int i = 0; i < height; ++i) {
      if (i >= recompute_height &&
          splice->prev_[i]->Next(i) != splice->next_[i]) {
-        FindSpliceForLevel<false>(key, splice->prev_[i], nullptr, i, &splice->prev_[i],
-                           &splice->next_[i]);
+        FindSpliceForLevel<false>(key_decoded, splice->prev_[i], nullptr, i,
+                                  &splice->prev_[i], &splice->next_[i]);
      }
      // Checking for duplicate keys on the level 0 is sufficient
      if (UNLIKELY(i == 0 && splice->next_[i] != nullptr &&
@@ -32,6 +32,12 @@ static Key Decode(const char* key) {
 }

 struct TestComparator {
+  typedef Key DecodedType;
+
+  static DecodedType decode_key(const char* b) {
+    return Decode(b);
+  }
+
  int operator()(const char* a, const char* b) const {
    if (Decode(a) < Decode(b)) {
      return -1;
@@ -41,6 +47,16 @@ struct TestComparator {
      return 0;
    }
  }
+
+  int operator()(const char* a, const DecodedType b) const {
+    if (Decode(a) < b) {
+      return -1;
+    } else if (Decode(a) > b) {
+      return +1;
+    } else {
+      return 0;
+    }
+  }
 };

 typedef InlineSkipList<TestComparator> TestInlineSkipList;
@@ -168,6 +168,10 @@ void ColumnFamilyOptions::Dump(Logger* log) const {
        log,
        "        Options.compression_opts.max_dict_bytes: %" ROCKSDB_PRIszt,
        compression_opts.max_dict_bytes);
+    ROCKS_LOG_HEADER(log,
+                     "        Options.compression_opts.zstd_max_train_bytes: "
+                     "%" ROCKSDB_PRIszt,
+                     compression_opts.zstd_max_train_bytes);
    ROCKS_LOG_HEADER(log, "     Options.level0_file_num_compaction_trigger: %d",
                     level0_file_num_compaction_trigger);
    ROCKS_LOG_HEADER(log, "         Options.level0_slowdown_writes_trigger: %d",
@@ -919,6 +919,17 @@ Status ParseColumnFamilyOption(const std::string& name,
        }
        new_options->compression_opts.max_dict_bytes =
            ParseInt(value.substr(start, value.size() - start));
+        end = value.find(':', start);
+      }
+      // zstd_max_train_bytes is optional for backwards compatibility
+      if (end != std::string::npos) {
+        start = end + 1;
+        if (start >= value.size()) {
+          return Status::InvalidArgument(
+              "unable to parse the specified CF option " + name);
+        }
+        new_options->compression_opts.zstd_max_train_bytes =
+            ParseInt(value.substr(start, value.size() - start));
      }
    } else {
      auto iter = cf_options_type_info.find(name);
@@ -138,7 +138,9 @@ void WindowsThread::join() {
      "WaitForSingleObjectFailed: thread join");
  }

-  CloseHandle(reinterpret_cast<HANDLE>(data_->handle_));
+  BOOL rc;
+  rc = CloseHandle(reinterpret_cast<HANDLE>(data_->handle_));
+  assert(rc != 0);
  data_->handle_ = 0;
 }

@@ -154,7 +156,7 @@ bool WindowsThread::detach() {
  BOOL ret = CloseHandle(reinterpret_cast<HANDLE>(data_->handle_));
  data_->handle_ = 0;

-  return (ret == TRUE);
+  return (ret != 0);
 }

 void  WindowsThread::swap(WindowsThread& o) {
@@ -148,6 +148,7 @@ LIB_SOURCES =                                                   \
  util/status_message.cc                                        \
  util/string_util.cc                                           \
  util/sync_point.cc                                            \
+  util/sync_point_impl.cc                                       \
  util/thread_local.cc                                          \
  util/threadpool_imp.cc                                        \
  util/transaction_test_util.cc                                 \
@@ -67,7 +67,8 @@ BlockBasedFilterBlockBuilder::BlockBasedFilterBlockBuilder(
      prefix_extractor_(prefix_extractor),
      whole_key_filtering_(table_opt.whole_key_filtering),
      prev_prefix_start_(0),
-      prev_prefix_size_(0) {
+      prev_prefix_size_(0),
+      num_added_(0) {
  assert(policy_);
 }

@@ -91,6 +92,7 @@ void BlockBasedFilterBlockBuilder::Add(const Slice& key) {

 // Add key to filter if needed
 inline void BlockBasedFilterBlockBuilder::AddKey(const Slice& key) {
+  num_added_++;
  start_.push_back(entries_.size());
  entries_.append(key.data(), key.size());
 }
@@ -106,10 +108,9 @@ inline void BlockBasedFilterBlockBuilder::AddPrefix(const Slice& key) {
  Slice prefix = prefix_extractor_->Transform(key);
  // insert prefix only when it's different from the previous prefix.
  if (prev.size() == 0 || prefix != prev) {
-    start_.push_back(entries_.size());
+    AddKey(prefix);
    prev_prefix_start_ = entries_.size();
    prev_prefix_size_ = prefix.size();
-    entries_.append(prefix.data(), prefix.size());
  }
 }

@@ -41,6 +41,7 @@ class BlockBasedFilterBlockBuilder : public FilterBlockBuilder {
  virtual bool IsBlockBased() override { return true; }
  virtual void StartBlock(uint64_t block_offset) override;
  virtual void Add(const Slice& key) override;
+  virtual size_t NumAdded() const override { return num_added_; }
  virtual Slice Finish(const BlockHandle& tmp, Status* status) override;
  using FilterBlockBuilder::Finish;

@@ -65,6 +66,7 @@ class BlockBasedFilterBlockBuilder : public FilterBlockBuilder {
  std::string result_;              // Filter data computed so far
  std::vector<Slice> tmp_entries_;  // policy_->CreateFilter() argument
  std::vector<uint32_t> filter_offsets_;
+  size_t num_added_;                // Number of keys added

  // No copying allowed
  BlockBasedFilterBlockBuilder(const BlockBasedFilterBlockBuilder&);
@@ -65,6 +65,7 @@ TEST_F(FilterBlockTest, EmptyBuilder) {

 TEST_F(FilterBlockTest, SingleChunk) {
  BlockBasedFilterBlockBuilder builder(nullptr, table_options_);
+  ASSERT_EQ(0, builder.NumAdded());
  builder.StartBlock(100);
  builder.Add("foo");
  builder.Add("bar");
@@ -73,6 +74,7 @@ TEST_F(FilterBlockTest, SingleChunk) {
  builder.Add("box");
  builder.StartBlock(300);
  builder.Add("hello");
+  ASSERT_EQ(5, builder.NumAdded());
  BlockContents block(builder.Finish(), false, kNoCompression);
  BlockBasedFilterBlockReader reader(nullptr, table_options_, true,
                                     std::move(block), nullptr);
@@ -650,8 +650,11 @@ Status BlockBasedTableBuilder::Finish() {

  BlockHandle filter_block_handle, metaindex_block_handle, index_block_handle,
      compression_dict_block_handle, range_del_block_handle;
+
  // Write filter block
-  if (ok() && r->filter_builder != nullptr) {
+  bool empty_filter_block = (r->filter_builder == nullptr ||
+                             r->filter_builder->NumAdded() == 0);
+  if (ok() && !empty_filter_block) {
    Status s = Status::Incomplete();
    while (s.IsIncomplete()) {
      Slice filter_content = r->filter_builder->Finish(filter_block_handle, &s);
@@ -687,7 +690,7 @@ Status BlockBasedTableBuilder::Finish() {
  }

  if (ok()) {
-    if (r->filter_builder != nullptr) {
+    if (!empty_filter_block) {
      // Add mapping from "<filter_block_prefix>.Name" to location
      // of filter data.
      std::string key;
@@ -30,6 +30,7 @@

 namespace rocksdb {

+inline
 void BlockFetcher::CheckBlockChecksum() {
  // Check the crc of the type and the block contents
  if (read_options_.verify_checksums) {
@@ -62,6 +63,7 @@ void BlockFetcher::CheckBlockChecksum() {
  }
 }

+inline
 bool BlockFetcher::TryGetUncompressBlockFromPersistentCache() {
  if (cache_options_.persistent_cache &&
      !cache_options_.persistent_cache->IsCompressed()) {
@@ -83,6 +85,7 @@ bool BlockFetcher::TryGetUncompressBlockFromPersistentCache() {
  return false;
 }

+inline
 bool BlockFetcher::TryGetFromPrefetchBuffer() {
  if (prefetch_buffer_ != nullptr &&
      prefetch_buffer_->TryReadFromCache(
@@ -99,6 +102,7 @@ bool BlockFetcher::TryGetFromPrefetchBuffer() {
  return got_from_prefetch_buffer_;
 }

+inline
 bool BlockFetcher::TryGetCompressedBlockFromPersistentCache() {
  if (cache_options_.persistent_cache &&
      cache_options_.persistent_cache->IsCompressed()) {
@@ -119,6 +123,7 @@ bool BlockFetcher::TryGetCompressedBlockFromPersistentCache() {
  return false;
 }

+inline
 void BlockFetcher::PrepareBufferForBlockFromFile() {
  // cache miss read from device
  if (do_uncompress_ &&
@@ -127,12 +132,12 @@ void BlockFetcher::PrepareBufferForBlockFromFile() {
    // trivially allocated stack buffer instead of needing a full malloc()
    used_buf_ = &stack_buf_[0];
  } else {
-    heap_buf_ =
-        std::unique_ptr<char[]>(new char[block_size_ + kBlockTrailerSize]);
+    heap_buf_.reset(new char[block_size_ + kBlockTrailerSize]);
    used_buf_ = heap_buf_.get();
  }
 }

+inline
 void BlockFetcher::InsertCompressedBlockToPersistentCacheIfNeeded() {
  if (status_.ok() && read_options_.fill_cache &&
      cache_options_.persistent_cache &&
@@ -143,6 +148,7 @@ void BlockFetcher::InsertCompressedBlockToPersistentCacheIfNeeded() {
  }
 }

+inline
 void BlockFetcher::InsertUncompressedBlockToPersistentCacheIfNeeded() {
  if (status_.ok() && !got_from_prefetch_buffer_ && read_options_.fill_cache &&
      cache_options_.persistent_cache &&
@@ -153,6 +159,7 @@ void BlockFetcher::InsertUncompressedBlockToPersistentCacheIfNeeded() {
  }
 }

+inline
 void BlockFetcher::GetBlockContents() {
  if (slice_.data() != used_buf_) {
    // the slice content is not the buffer provided
@@ -161,7 +168,7 @@ void BlockFetcher::GetBlockContents() {
  } else {
    // page is uncompressed, the buffer either stack or heap provided
    if (got_from_prefetch_buffer_ || used_buf_ == &stack_buf_[0]) {
-      heap_buf_ = std::unique_ptr<char[]>(new char[block_size_]);
+      heap_buf_.reset(new char[block_size_]);
      memcpy(heap_buf_.get(), used_buf_, block_size_);
    }
    *contents_ = BlockContents(std::move(heap_buf_), block_size_, true,
@@ -51,6 +51,7 @@ class FilterBlockBuilder {
  virtual bool IsBlockBased() = 0;                    // If is blockbased filter
  virtual void StartBlock(uint64_t block_offset) = 0;  // Start new block filter
  virtual void Add(const Slice& key) = 0;      // Add a key to current filter
+  virtual size_t NumAdded() const = 0;         // Number of keys added
  Slice Finish() {                             // Generate Filter
    const BlockHandle empty_handle;
    Status dont_care_status;
@@ -45,6 +45,7 @@ class FullFilterBlockBuilder : public FilterBlockBuilder {
  virtual bool IsBlockBased() override { return false; }
  virtual void StartBlock(uint64_t /*block_offset*/) override {}
  virtual void Add(const Slice& key) override;
+  virtual size_t NumAdded() const override { return num_added_; }
  virtual Slice Finish(const BlockHandle& tmp, Status* status) override;
  using FilterBlockBuilder::Finish;

@@ -163,11 +163,13 @@ TEST_F(FullFilterBlockTest, EmptyBuilder) {
 TEST_F(FullFilterBlockTest, SingleChunk) {
  FullFilterBlockBuilder builder(
      nullptr, true, table_options_.filter_policy->GetFilterBitsBuilder());
+  ASSERT_EQ(0, builder.NumAdded());
  builder.Add("foo");
  builder.Add("bar");
  builder.Add("box");
  builder.Add("box");
  builder.Add("hello");
+  ASSERT_EQ(5, builder.NumAdded());
  Slice block = builder.Finish();
  FullFilterBlockReader reader(
      nullptr, true, block,
@@ -25,7 +25,8 @@ PartitionedFilterBlockBuilder::PartitionedFilterBlockBuilder(
                             filter_bits_builder),
      index_on_filter_block_builder_(index_block_restart_interval),
      p_index_builder_(p_index_builder),
-      filters_in_partition_(0) {
+      filters_in_partition_(0),
+      num_added_(0) {
  filters_per_partition_ =
      filter_bits_builder_->CalculateNumEntry(partition_size);
 }
@@ -53,6 +54,7 @@ void PartitionedFilterBlockBuilder::AddKey(const Slice& key) {
  MaybeCutAFilterBlock();
  filter_bits_builder_->AddKey(key);
  filters_in_partition_++;
+  num_added_++;
 }

 Slice PartitionedFilterBlockBuilder::Finish(
@@ -33,6 +33,8 @@ class PartitionedFilterBlockBuilder : public FullFilterBlockBuilder {

  void AddKey(const Slice& key) override;

+  size_t NumAdded() const override { return num_added_; }
+
  virtual Slice Finish(const BlockHandle& last_partition_block_handle,
                       Status* status) override;

@@ -59,6 +61,8 @@ class PartitionedFilterBlockBuilder : public FullFilterBlockBuilder {
  uint32_t filters_per_partition_;
  // The current number of filters in the last partition
  uint32_t filters_in_partition_;
+  // Number of keys added
+  size_t num_added_;
 };

 class PartitionedFilterBlockReader : public FilterBlockReader {
@@ -108,14 +108,12 @@ fi

 params_w="$const_params \
          $l0_config \
-          --max_background_compactions=16 \
-          --max_write_buffer_number=8 \
-          --max_background_flushes=7"
+          --max_background_jobs=20 \
+          --max_write_buffer_number=8"

 params_bulkload="$const_params \
-                 --max_background_compactions=16 \
+                 --max_background_jobs=20 \
                 --max_write_buffer_number=8 \
-                 --max_background_flushes=7 \
                 --level0_file_num_compaction_trigger=$((10 * M)) \
                 --level0_slowdown_writes_trigger=$((10 * M)) \
                 --level0_stop_writes_trigger=$((10 * M))"
@@ -126,14 +124,14 @@ params_bulkload="$const_params \
 # LSM. In level-based compaction, it means number of L0 files.
 #
 params_level_compact="$const_params \
-                --max_background_flushes=4 \
+                --max_background_jobs=16 \
                --max_write_buffer_number=4 \
                --level0_file_num_compaction_trigger=4 \
                --level0_slowdown_writes_trigger=16 \
                --level0_stop_writes_trigger=20"

 params_univ_compact="$const_params \
-                --max_background_flushes=4 \
+                --max_background_jobs=20 \
                --max_write_buffer_number=4 \
                --level0_file_num_compaction_trigger=8 \
                --level0_slowdown_writes_trigger=16 \
@@ -232,7 +230,7 @@ function run_manual_compaction_worker {
       --subcompactions=$3 \
       --memtablerep=vector \
       --disable_wal=1 \
-       --max_background_compactions=$4 \
+       --max_background_jobs=$4 \
       --seed=$( date +%s ) \
       2>&1 | tee -a $fillrandom_output_file"

@@ -276,7 +274,7 @@ function run_univ_compaction {

  # Define a set of benchmarks.
  subcompactions=(1 2 4 8 16)
-  max_background_compactions=(16 16 8 4 2)
+  max_background_jobs=(20 20 10 5 4)

  i=0
  total=${#subcompactions[@]}
@@ -285,7 +283,7 @@ function run_univ_compaction {
  while [ "$i" -lt "$total" ]
  do
    run_manual_compaction_worker $io_stats $compaction_style ${subcompactions[$i]} \
-      ${max_background_compactions[$i]}
+      ${max_background_jobs[$i]}
    ((i++))
  done
 }
@@ -42,7 +42,7 @@ EOF
 done

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

 generate_db()
@@ -22,11 +22,13 @@ namespace rocksdb {
 DeleteScheduler::DeleteScheduler(Env* env, int64_t rate_bytes_per_sec,
                                 Logger* info_log,
                                 SstFileManagerImpl* sst_file_manager,
-                                 double max_trash_db_ratio)
+                                 double max_trash_db_ratio,
+                                 uint64_t bytes_max_delete_chunk)
    : env_(env),
      total_trash_size_(0),
      rate_bytes_per_sec_(rate_bytes_per_sec),
      pending_files_(0),
+      bytes_max_delete_chunk_(bytes_max_delete_chunk),
      closing_(false),
      cv_(&mu_),
      info_log_(info_log),
@@ -208,15 +210,18 @@ void DeleteScheduler::BackgroundEmptyTrash() {

      // Get new file to delete
      std::string path_in_trash = queue_.front();
-      queue_.pop();

      // We dont need to hold the lock while deleting the file
      mu_.Unlock();
      uint64_t deleted_bytes = 0;
+      bool is_complete = true;
      // Delete file from trash and update total_penlty value
-      Status s = DeleteTrashFile(path_in_trash,  &deleted_bytes);
+      Status s = DeleteTrashFile(path_in_trash, &deleted_bytes, &is_complete);
      total_deleted_bytes += deleted_bytes;
      mu_.Lock();
+      if (is_complete) {
+        queue_.pop();
+      }

      if (!s.ok()) {
        bg_errors_[path_in_trash] = s;
@@ -236,7 +241,9 @@ void DeleteScheduler::BackgroundEmptyTrash() {
      TEST_SYNC_POINT_CALLBACK("DeleteScheduler::BackgroundEmptyTrash:Wait",
                               &total_penlty);

-      pending_files_--;
+      if (is_complete) {
+        pending_files_--;
+      }
      if (pending_files_ == 0) {
        // Unblock WaitForEmptyTrash since there are no more files waiting
        // to be deleted
@@ -247,23 +254,49 @@ void DeleteScheduler::BackgroundEmptyTrash() {
 }

 Status DeleteScheduler::DeleteTrashFile(const std::string& path_in_trash,
-                                        uint64_t* deleted_bytes) {
+                                        uint64_t* deleted_bytes,
+                                        bool* is_complete) {
  uint64_t file_size;
  Status s = env_->GetFileSize(path_in_trash, &file_size);
+  *is_complete = true;
+  TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:DeleteFile");
  if (s.ok()) {
-    TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:DeleteFile");
-    s = env_->DeleteFile(path_in_trash);
-  }
+    bool need_full_delete = true;
+    if (bytes_max_delete_chunk_ != 0 && file_size > bytes_max_delete_chunk_) {
+      unique_ptr<WritableFile> wf;
+      Status my_status =
+          env_->ReopenWritableFile(path_in_trash, &wf, EnvOptions());
+      if (my_status.ok()) {
+        my_status = wf->Truncate(file_size - bytes_max_delete_chunk_);
+        if (my_status.ok()) {
+          TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:Fsync");
+          my_status = wf->Fsync();
+        }
+      }
+      if (my_status.ok()) {
+        *deleted_bytes = bytes_max_delete_chunk_;
+        need_full_delete = false;
+        *is_complete = false;
+      } else {
+        ROCKS_LOG_WARN(info_log_,
+                       "Failed to partially delete %s from trash -- %s",
+                       path_in_trash.c_str(), my_status.ToString().c_str());
+      }
+    }

+    if (need_full_delete) {
+      s = env_->DeleteFile(path_in_trash);
+      *deleted_bytes = file_size;
+      sst_file_manager_->OnDeleteFile(path_in_trash);
+    }
+  }
  if (!s.ok()) {
    // Error while getting file size or while deleting
    ROCKS_LOG_ERROR(info_log_, "Failed to delete %s from trash -- %s",
                    path_in_trash.c_str(), s.ToString().c_str());
    *deleted_bytes = 0;
  } else {
-    *deleted_bytes = file_size;
-    total_trash_size_.fetch_sub(file_size);
-    sst_file_manager_->OnDeleteFile(path_in_trash);
+    total_trash_size_.fetch_sub(*deleted_bytes);
  }

  return s;
@@ -34,7 +34,7 @@ class DeleteScheduler {
 public:
  DeleteScheduler(Env* env, int64_t rate_bytes_per_sec, Logger* info_log,
                  SstFileManagerImpl* sst_file_manager,
-                  double max_trash_db_ratio);
+                  double max_trash_db_ratio, uint64_t bytes_max_delete_chunk);

  ~DeleteScheduler();

@@ -82,7 +82,7 @@ class DeleteScheduler {
  Status MarkAsTrash(const std::string& file_path, std::string* path_in_trash);

  Status DeleteTrashFile(const std::string& path_in_trash,
-                         uint64_t* deleted_bytes);
+                         uint64_t* deleted_bytes, bool* is_complete);

  void BackgroundEmptyTrash();

@@ -97,6 +97,7 @@ class DeleteScheduler {
  std::queue<std::string> queue_;
  // Number of trash files that are waiting to be deleted
  int32_t pending_files_;
+  uint64_t bytes_max_delete_chunk_;
  // Errors that happened in BackgroundEmptyTrash (file_path => error)
  std::map<std::string, Status> bg_errors_;
  // Set to true in ~DeleteScheduler() to force BackgroundEmptyTrash to stop
@@ -99,7 +99,7 @@ class DeleteSchedulerTest : public testing::Test {
    // 25%)
    sst_file_mgr_.reset(
        new SstFileManagerImpl(env_, nullptr, rate_bytes_per_sec_,
-                               /* max_trash_db_ratio= */ 1.1));
+                               /* max_trash_db_ratio= */ 1.1, 128 * 1024));
    delete_scheduler_ = sst_file_mgr_->delete_scheduler();
  }

@@ -436,6 +436,34 @@ TEST_F(DeleteSchedulerTest, StartBGEmptyTrashMultipleTimes) {
  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
 }

+TEST_F(DeleteSchedulerTest, DeletePartialFile) {
+  int bg_delete_file = 0;
+  int bg_fsync = 0;
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteTrashFile:DeleteFile",
+      [&](void*) { bg_delete_file++; });
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteTrashFile:Fsync", [&](void*) { bg_fsync++; });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  rate_bytes_per_sec_ = 1024 * 1024;  // 1 MB / sec
+  NewDeleteScheduler();
+
+  // Should delete in 4 batch
+  ASSERT_OK(delete_scheduler_->DeleteFile(NewDummyFile("data_1", 500 * 1024)));
+  ASSERT_OK(delete_scheduler_->DeleteFile(NewDummyFile("data_2", 100 * 1024)));
+  // Should delete in 2 batch
+  ASSERT_OK(delete_scheduler_->DeleteFile(NewDummyFile("data_2", 200 * 1024)));
+
+  delete_scheduler_->WaitForEmptyTrash();
+
+  auto bg_errors = delete_scheduler_->GetBackgroundErrors();
+  ASSERT_EQ(bg_errors.size(), 0);
+  ASSERT_EQ(7, bg_delete_file);
+  ASSERT_EQ(4, bg_fsync);
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+}
+
 // 1- Create a DeleteScheduler with very slow rate limit (1 Byte / sec)
 // 2- Delete 100 files using DeleteScheduler
 // 3- Delete the DeleteScheduler (call the destructor while queue is not empty)
@@ -18,7 +18,8 @@ namespace rocksdb {
 #ifndef ROCKSDB_LITE
 SstFileManagerImpl::SstFileManagerImpl(Env* env, std::shared_ptr<Logger> logger,
                                       int64_t rate_bytes_per_sec,
-                                       double max_trash_db_ratio)
+                                       double max_trash_db_ratio,
+                                       uint64_t bytes_max_delete_chunk)
    : env_(env),
      logger_(logger),
      total_files_size_(0),
@@ -26,7 +27,7 @@ SstFileManagerImpl::SstFileManagerImpl(Env* env, std::shared_ptr<Logger> logger,
      cur_compactions_reserved_size_(0),
      max_allowed_space_(0),
      delete_scheduler_(env, rate_bytes_per_sec, logger.get(), this,
-                        max_trash_db_ratio) {}
+                        max_trash_db_ratio, bytes_max_delete_chunk) {}

 SstFileManagerImpl::~SstFileManagerImpl() {}

@@ -196,10 +197,11 @@ SstFileManager* NewSstFileManager(Env* env, std::shared_ptr<Logger> info_log,
                                  std::string trash_dir,
                                  int64_t rate_bytes_per_sec,
                                  bool delete_existing_trash, Status* status,
-                                  double max_trash_db_ratio) {
+                                  double max_trash_db_ratio,
+                                  uint64_t bytes_max_delete_chunk) {
  SstFileManagerImpl* res =
      new SstFileManagerImpl(env, info_log, rate_bytes_per_sec,
-                             max_trash_db_ratio);
+                             max_trash_db_ratio, bytes_max_delete_chunk);

  // trash_dir is deprecated and not needed anymore, but if user passed it
  // we will still remove files in it.
@@ -236,7 +238,8 @@ SstFileManager* NewSstFileManager(Env* env, std::shared_ptr<Logger> info_log,
                                  std::string trash_dir,
                                  int64_t rate_bytes_per_sec,
                                  bool delete_existing_trash, Status* status,
-                                  double max_trash_db_ratio) {
+                                  double max_trash_db_ratio,
+                                  uint64_t bytes_max_delete_chunk) {
  if (status) {
    *status =
      Status::NotSupported("SstFileManager is not supported in ROCKSDB_LITE");
@@ -27,7 +27,8 @@ class SstFileManagerImpl : public SstFileManager {
 public:
  explicit SstFileManagerImpl(Env* env, std::shared_ptr<Logger> logger,
                              int64_t rate_bytes_per_sec,
-                              double max_trash_db_ratio);
+                              double max_trash_db_ratio,
+                              uint64_t bytes_max_delete_chunk);

  ~SstFileManagerImpl();

@@ -4,10 +4,7 @@
 //  (found in the LICENSE.Apache file in the root directory).

 #include "util/sync_point.h"
-#include <functional>
-#include <thread>
-#include "port/port.h"
-#include "util/random.h"
+#include "util/sync_point_impl.h"

 int rocksdb_kill_odds = 0;
 std::vector<std::string> rocksdb_kill_prefix_blacklist;
@@ -15,156 +12,57 @@ std::vector<std::string> rocksdb_kill_prefix_blacklist;
 #ifndef NDEBUG
 namespace rocksdb {

-void TestKillRandom(std::string kill_point, int odds,
-                    const std::string& srcfile, int srcline) {
-  for (auto& p : rocksdb_kill_prefix_blacklist) {
-    if (kill_point.substr(0, p.length()) == p) {
-      return;
-    }
-  }
-
-  assert(odds > 0);
-  if (odds % 7 == 0) {
-    // class Random uses multiplier 16807, which is 7^5. If odds are
-    // multiplier of 7, there might be limited values generated.
-    odds++;
-  }
-  auto* r = Random::GetTLSInstance();
-  bool crash = r->OneIn(odds);
-  if (crash) {
-    port::Crash(srcfile, srcline);
-  }
-}
-
 SyncPoint* SyncPoint::GetInstance() {
  static SyncPoint sync_point;
  return &sync_point;
 }

+SyncPoint::SyncPoint() : 
+  impl_(new Data) {
+}
+
+SyncPoint:: ~SyncPoint() {
+  delete impl_;
+}
+
 void SyncPoint::LoadDependency(const std::vector<SyncPointPair>& dependencies) {
-  std::unique_lock<std::mutex> lock(mutex_);
-  successors_.clear();
-  predecessors_.clear();
-  cleared_points_.clear();
-  for (const auto& dependency : dependencies) {
-    successors_[dependency.predecessor].push_back(dependency.successor);
-    predecessors_[dependency.successor].push_back(dependency.predecessor);
-  }
-  cv_.notify_all();
+  impl_->LoadDependency(dependencies);
 }

 void SyncPoint::LoadDependencyAndMarkers(
-    const std::vector<SyncPointPair>& dependencies,
-    const std::vector<SyncPointPair>& markers) {
-  std::unique_lock<std::mutex> lock(mutex_);
-  successors_.clear();
-  predecessors_.clear();
-  cleared_points_.clear();
-  markers_.clear();
-  marked_thread_id_.clear();
-  for (const auto& dependency : dependencies) {
-    successors_[dependency.predecessor].push_back(dependency.successor);
-    predecessors_[dependency.successor].push_back(dependency.predecessor);
-  }
-  for (const auto& marker : markers) {
-    successors_[marker.predecessor].push_back(marker.successor);
-    predecessors_[marker.successor].push_back(marker.predecessor);
-    markers_[marker.predecessor].push_back(marker.successor);
-  }
-  cv_.notify_all();
+  const std::vector<SyncPointPair>& dependencies,
+  const std::vector<SyncPointPair>& markers) {
+  impl_->LoadDependencyAndMarkers(dependencies, markers);
 }

-bool SyncPoint::PredecessorsAllCleared(const std::string& point) {
-  for (const auto& pred : predecessors_[point]) {
-    if (cleared_points_.count(pred) == 0) {
-      return false;
-    }
-  }
-  return true;
+void SyncPoint::SetCallBack(const std::string& point,
+  const std::function<void(void*)>& callback) {
+  impl_->SetCallBack(point, callback);
 }

-void SyncPoint::SetCallBack(const std::string point,
-                            std::function<void(void*)> callback) {
-  std::unique_lock<std::mutex> lock(mutex_);
-  callbacks_[point] = callback;
-}
-
-void SyncPoint::ClearCallBack(const std::string point) {
-  std::unique_lock<std::mutex> lock(mutex_);
-  while (num_callbacks_running_ > 0) {
-    cv_.wait(lock);
-  }
-  callbacks_.erase(point);
+void SyncPoint::ClearCallBack(const std::string& point) {
+  impl_->ClearCallBack(point);
 }

 void SyncPoint::ClearAllCallBacks() {
-  std::unique_lock<std::mutex> lock(mutex_);
-  while (num_callbacks_running_ > 0) {
-    cv_.wait(lock);
-  }
-  callbacks_.clear();
+  impl_->ClearAllCallBacks();
 }

 void SyncPoint::EnableProcessing() {
-  std::unique_lock<std::mutex> lock(mutex_);
-  enabled_ = true;
+  impl_->EnableProcessing();
 }

 void SyncPoint::DisableProcessing() {
-  std::unique_lock<std::mutex> lock(mutex_);
-  enabled_ = false;
+  impl_->DisableProcessing();
 }

 void SyncPoint::ClearTrace() {
-  std::unique_lock<std::mutex> lock(mutex_);
-  cleared_points_.clear();
-}
-
-bool SyncPoint::DisabledByMarker(const std::string& point,
-                                 std::thread::id thread_id) {
-  auto marked_point_iter = marked_thread_id_.find(point);
-  return marked_point_iter != marked_thread_id_.end() &&
-         thread_id != marked_point_iter->second;
+  impl_->ClearTrace();
 }

 void SyncPoint::Process(const std::string& point, void* cb_arg) {
-  std::unique_lock<std::mutex> lock(mutex_);
-  if (!enabled_) {
-    return;
-  }
-
-  auto thread_id = std::this_thread::get_id();
-
-  auto marker_iter = markers_.find(point);
-  if (marker_iter != markers_.end()) {
-    for (auto marked_point : marker_iter->second) {
-      marked_thread_id_.insert(std::make_pair(marked_point, thread_id));
-    }
-  }
-
-  if (DisabledByMarker(point, thread_id)) {
-    return;
-  }
-
-  while (!PredecessorsAllCleared(point)) {
-    cv_.wait(lock);
-    if (DisabledByMarker(point, thread_id)) {
-      return;
-    }
-  }
-
-  auto callback_pair = callbacks_.find(point);
-  if (callback_pair != callbacks_.end()) {
-    num_callbacks_running_++;
-    mutex_.unlock();
-    callback_pair->second(cb_arg);
-    mutex_.lock();
-    num_callbacks_running_--;
-    cv_.notify_all();
-  }
-
-  cleared_points_.insert(point);
-  cv_.notify_all();
+  impl_->Process(point, cb_arg);
 }
+
 }  // namespace rocksdb
 #endif  // NDEBUG
@@ -5,13 +5,10 @@
 #pragma once

 #include <assert.h>
-#include <condition_variable>
 #include <functional>
 #include <mutex>
 #include <string>
 #include <thread>
-#include <unordered_map>
-#include <unordered_set>
 #include <vector>

 // This is only set from db_stress.cc and for testing only.
@@ -65,6 +62,10 @@ class SyncPoint {
 public:
  static SyncPoint* GetInstance();

+  SyncPoint(const SyncPoint&) = delete;
+  SyncPoint& operator=(const SyncPoint&) = delete;
+  ~SyncPoint();
+
  struct SyncPointPair {
    std::string predecessor;
    std::string successor;
@@ -81,15 +82,14 @@ class SyncPoint {
  void LoadDependencyAndMarkers(const std::vector<SyncPointPair>& dependencies,
                                const std::vector<SyncPointPair>& markers);

-  // Set up a call back function in sync point.
  // The argument to the callback is passed through from
  // TEST_SYNC_POINT_CALLBACK(); nullptr if TEST_SYNC_POINT or
  // TEST_IDX_SYNC_POINT was used.
-  void SetCallBack(const std::string point,
-                   std::function<void(void*)> callback);
+  void SetCallBack(const std::string& point,
+                   const std::function<void(void*)>& callback);

  // Clear callback function by point
-  void ClearCallBack(const std::string point);
+  void ClearCallBack(const std::string& point);

  // Clear all call back functions.
  void ClearAllCallBacks();
@@ -111,29 +111,20 @@ class SyncPoint {
  // TODO: it might be useful to provide a function that blocks until all
  // sync points are cleared.

+  // We want this to be public so we can
+  // subclass the implementation
+  struct Data;
+
 private:
-  bool PredecessorsAllCleared(const std::string& point);
-  bool DisabledByMarker(const std::string& point, std::thread::id thread_id);
-
-  // successor/predecessor map loaded from LoadDependency
-  std::unordered_map<std::string, std::vector<std::string>> successors_;
-  std::unordered_map<std::string, std::vector<std::string>> predecessors_;
-  std::unordered_map<std::string, std::function<void(void*)> > callbacks_;
-  std::unordered_map<std::string, std::vector<std::string> > markers_;
-  std::unordered_map<std::string, std::thread::id> marked_thread_id_;
-
-  std::mutex mutex_;
-  std::condition_variable cv_;
-  // sync points that have been passed through
-  std::unordered_set<std::string> cleared_points_;
-  bool enabled_ = false;
-  int num_callbacks_running_ = 0;
+   // Singleton
+  SyncPoint();
+  Data*  impl_;
 };

 }  // namespace rocksdb

 // Use TEST_SYNC_POINT to specify sync points inside code base.
-// Sync points can have happens-after dependency on other sync points,
+// Sync points can have happens-after depedency on other sync points,
 // configured at runtime via SyncPoint::LoadDependency. This could be
 // utilized to re-produce race conditions between threads.
 // See TransactionLogIteratorRace in db_test.cc for an example use case.
@@ -0,0 +1,129 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "util/sync_point_impl.h"
+
+#ifndef NDEBUG
+namespace rocksdb {
+
+void TestKillRandom(std::string kill_point, int odds,
+                    const std::string& srcfile, int srcline) {
+  for (auto& p : rocksdb_kill_prefix_blacklist) {
+    if (kill_point.substr(0, p.length()) == p) {
+      return;
+    }
+  }
+
+  assert(odds > 0);
+  if (odds % 7 == 0) {
+    // class Random uses multiplier 16807, which is 7^5. If odds are
+    // multiplier of 7, there might be limited values generated.
+    odds++;
+  }
+  auto* r = Random::GetTLSInstance();
+  bool crash = r->OneIn(odds);
+  if (crash) {
+    port::Crash(srcfile, srcline);
+  }
+}
+
+
+void SyncPoint::Data::LoadDependency(const std::vector<SyncPointPair>& dependencies) {
+  std::lock_guard<std::mutex> lock(mutex_);
+  successors_.clear();
+  predecessors_.clear();
+  cleared_points_.clear();
+  for (const auto& dependency : dependencies) {
+    successors_[dependency.predecessor].push_back(dependency.successor);
+    predecessors_[dependency.successor].push_back(dependency.predecessor);
+  }
+  cv_.notify_all();
+}
+
+void SyncPoint::Data::LoadDependencyAndMarkers(
+  const std::vector<SyncPointPair>& dependencies,
+  const std::vector<SyncPointPair>& markers) {
+  std::lock_guard<std::mutex> lock(mutex_);
+  successors_.clear();
+  predecessors_.clear();
+  cleared_points_.clear();
+  markers_.clear();
+  marked_thread_id_.clear();
+  for (const auto& dependency : dependencies) {
+    successors_[dependency.predecessor].push_back(dependency.successor);
+    predecessors_[dependency.successor].push_back(dependency.predecessor);
+  }
+  for (const auto& marker : markers) {
+    successors_[marker.predecessor].push_back(marker.successor);
+    predecessors_[marker.successor].push_back(marker.predecessor);
+    markers_[marker.predecessor].push_back(marker.successor);
+  }
+  cv_.notify_all();
+}
+
+bool SyncPoint::Data::PredecessorsAllCleared(const std::string& point) {
+  for (const auto& pred : predecessors_[point]) {
+    if (cleared_points_.count(pred) == 0) {
+      return false;
+    }
+  }
+  return true;
+}
+
+void SyncPoint::Data::ClearCallBack(const std::string& point) {
+  std::unique_lock<std::mutex> lock(mutex_);
+  while (num_callbacks_running_ > 0) {
+    cv_.wait(lock);
+  }
+  callbacks_.erase(point);
+}
+
+void SyncPoint::Data::ClearAllCallBacks() {
+  std::unique_lock<std::mutex> lock(mutex_);
+  while (num_callbacks_running_ > 0) {
+    cv_.wait(lock);
+  }
+  callbacks_.clear();
+}
+
+void SyncPoint::Data::Process(const std::string& point, void* cb_arg) {
+  std::unique_lock<std::mutex> lock(mutex_);
+  if (!enabled_) {
+    return;
+  }
+
+  auto thread_id = std::this_thread::get_id();
+
+  auto marker_iter = markers_.find(point);
+  if (marker_iter != markers_.end()) {
+    for (auto& marked_point : marker_iter->second) {
+      marked_thread_id_.emplace(marked_point, thread_id);
+    }
+  }
+
+  if (DisabledByMarker(point, thread_id)) {
+    return;
+  }
+
+  while (!PredecessorsAllCleared(point)) {
+    cv_.wait(lock);
+    if (DisabledByMarker(point, thread_id)) {
+      return;
+    }
+  }
+
+  auto callback_pair = callbacks_.find(point);
+  if (callback_pair != callbacks_.end()) {
+    num_callbacks_running_++;
+    mutex_.unlock();
+    callback_pair->second(cb_arg);
+    mutex_.lock();
+    num_callbacks_running_--;
+  }
+  cleared_points_.insert(point);
+  cv_.notify_all();
+}
+} // rocksdb
+#endif
@@ -0,0 +1,74 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "util/sync_point.h"
+
+#include <assert.h>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <unordered_map>
+#include <unordered_set>
+
+#include "port/port.h"
+#include "util/random.h"
+
+#pragma once
+
+#ifndef NDEBUG
+namespace rocksdb {
+struct SyncPoint::Data {
+  // Enable proper deletion by subclasses
+  virtual ~Data() {}
+  // successor/predecessor map loaded from LoadDependency
+  std::unordered_map<std::string, std::vector<std::string>> successors_;
+  std::unordered_map<std::string, std::vector<std::string>> predecessors_;
+  std::unordered_map<std::string, std::function<void(void*)> > callbacks_;
+  std::unordered_map<std::string, std::vector<std::string> > markers_;
+  std::unordered_map<std::string, std::thread::id> marked_thread_id_;
+
+  std::mutex              mutex_;
+  std::condition_variable cv_;
+  // sync points that have been passed through
+  std::unordered_set<std::string> cleared_points_;
+  bool enabled_ = false;
+  int num_callbacks_running_ = 0;
+
+  void LoadDependency(const std::vector<SyncPointPair>& dependencies);
+  void LoadDependencyAndMarkers(const std::vector<SyncPointPair>& dependencies,
+    const std::vector<SyncPointPair>& markers);
+  bool PredecessorsAllCleared(const std::string& point);
+  void SetCallBack(const std::string& point,
+    const std::function<void(void*)>& callback) {
+  std::lock_guard<std::mutex> lock(mutex_);
+  callbacks_[point] = callback;
+}
+
+  void ClearCallBack(const std::string& point);
+  void ClearAllCallBacks();
+  void EnableProcessing() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    enabled_ = true;
+  }
+  void DisableProcessing() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    enabled_ = false;
+  }
+  void ClearTrace() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    cleared_points_.clear();
+  }
+  bool DisabledByMarker(const std::string& point,
+                        std::thread::id thread_id) {
+    auto marked_point_iter = marked_thread_id_.find(point);
+    return marked_point_iter != marked_thread_id_.end() &&
+           thread_id != marked_point_iter->second;
+  }
+  void Process(const std::string& point, void* cb_arg);
+};
+}
+#endif // NDEBUG
@@ -1733,25 +1733,51 @@ Status BackupEngineImpl::BackupMeta::StoreToFile(bool sync) {
  if (!app_metadata_.empty()) {
    std::string hex_encoded_metadata =
        Slice(app_metadata_).ToString(/* hex */ true);
+
+    // +1 to accomodate newline character
+    size_t hex_meta_strlen = kMetaDataPrefix.ToString().length() + hex_encoded_metadata.length() + 1;
+    if (hex_meta_strlen >= buf_size) {
+      return Status::Corruption("Buffer too small to fit backup metadata");
+    }
+    else if (len + hex_meta_strlen >= buf_size) {
+      backup_meta_file->Append(Slice(buf.get(), len));
+      buf.reset();
+      unique_ptr<char[]> new_reset_buf(new char[max_backup_meta_file_size_]);
+      buf.swap(new_reset_buf);
+      len = 0;
+    }
    len += snprintf(buf.get() + len, buf_size - len, "%s%s\n",
                    kMetaDataPrefix.ToString().c_str(),
                    hex_encoded_metadata.c_str());
-    if (len >= buf_size) {
-      return Status::Corruption("Buffer too small to fit backup metadata");
-    }
  }
-  len += snprintf(buf.get() + len, buf_size - len, "%" ROCKSDB_PRIszt "\n",
-                  files_.size());
-  if (len >= buf_size) {
-    return Status::Corruption("Buffer too small to fit backup metadata");
+
+  char writelen_temp[19];
+  if (len + sprintf(writelen_temp, "%" ROCKSDB_PRIszt "\n", files_.size()) >= buf_size) {
+    backup_meta_file->Append(Slice(buf.get(), len));
+    buf.reset();
+    unique_ptr<char[]> new_reset_buf(new char[max_backup_meta_file_size_]);
+    buf.swap(new_reset_buf);
+    len = 0;
  }
+  {
+    const char *const_write = writelen_temp;
+    len += snprintf(buf.get() + len, buf_size - len, "%s", const_write);
+  }
+
  for (const auto& file : files_) {
    // use crc32 for now, switch to something else if needed
-    len += snprintf(buf.get() + len, buf_size - len, "%s crc32 %u\n",
-                    file->filename.c_str(), file->checksum_value);
-    if (len >= buf_size) {
-      return Status::Corruption("Buffer too small to fit backup metadata");
+
+    size_t newlen = len + file->filename.length() + sprintf(writelen_temp, " crc32 %u\n", file->checksum_value);
+    const char *const_write = writelen_temp;
+    if (newlen >= buf_size) {
+      backup_meta_file->Append(Slice(buf.get(), len));
+      buf.reset();
+      unique_ptr<char[]> new_reset_buf(new char[max_backup_meta_file_size_]);
+      buf.swap(new_reset_buf);
+      len = 0;
    }
+    len += snprintf(buf.get() + len, buf_size - len, "%s%s",
+                    file->filename.c_str(), const_write);
  }

  s = backup_meta_file->Append(Slice(buf.get(), len));
@@ -1218,9 +1218,7 @@ Status BlobDBImpl::CloseBlobFile(std::shared_ptr<BlobFile> bfile,
    if (bfile->HasTTL()) {
      size_t erased __attribute__((__unused__));
      erased = open_ttl_files_.erase(bfile);
-      assert(erased == 1);
-    } else {
-      assert(bfile == open_non_ttl_file_);
+    } else if (bfile == open_non_ttl_file_) {
      open_non_ttl_file_ = nullptr;
    }
  }
@@ -1301,7 +1299,9 @@ std::pair<bool, int64_t> BlobDBImpl::CheckSeqFiles(bool aborted) {
      {
        ReadLock lockbfile_r(&bfile->mutex_);

-        if (bfile->expiration_range_.second > epoch_now) continue;
+        if (bfile->expiration_range_.second > epoch_now) {
+          continue;
+        }
        process_files.push_back(bfile);
      }
    }
@@ -1568,12 +1568,16 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
      newfile = NewBlobFile(reason);

      new_writer = CheckOrCreateWriterLocked(newfile);
-      newfile->header_ = std::move(header);
      // Can't use header beyond this point
+      newfile->header_ = std::move(header);
      newfile->header_valid_ = true;
      newfile->file_size_ = BlobLogHeader::kSize;
-      s = new_writer->WriteHeader(newfile->header_);
+      newfile->SetColumnFamilyId(bfptr->column_family_id());
+      newfile->SetHasTTL(bfptr->HasTTL());
+      newfile->SetCompression(bfptr->compression());
+      newfile->expiration_range_ = bfptr->expiration_range_;

+      s = new_writer->WriteHeader(newfile->header_);
      if (!s.ok()) {
        ROCKS_LOG_ERROR(db_options_.info_log,
                        "File: %s - header writing failed",
@@ -1581,8 +1585,10 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
        break;
      }

+      // We don't add the file to open_ttl_files_ or open_non_ttl_files_, to
+      // avoid user writes writing to the file, and avoid CheckSeqFiles close
+      // the file by mistake.
      WriteLock wl(&mutex_);
-
      blob_files_.insert(std::make_pair(newfile->BlobFileNumber(), newfile));
    }

@@ -1650,6 +1656,10 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
             gc_stats->bytes_overwritten);
  RecordTick(statistics_, BLOB_DB_GC_BYTES_EXPIRED, gc_stats->bytes_expired);
  if (newfile != nullptr) {
+    {
+      MutexLock l(&write_mutex_);
+      CloseBlobFile(newfile);
+    }
    total_blob_size_ += newfile->file_size_;
    ROCKS_LOG_INFO(db_options_.info_log, "New blob file %" PRIu64 ".",
                   newfile->BlobFileNumber());
@@ -763,6 +763,26 @@ TEST_F(BlobDBTest, GCExpiredKeyWhileOverwriting) {
  VerifyDB({{"foo", "v2"}});
 }

+TEST_F(BlobDBTest, NewFileGeneratedFromGCShouldMarkAsImmutable) {
+  BlobDBOptions bdb_options;
+  bdb_options.min_blob_size = 0;
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options);
+  ASSERT_OK(Put("foo", "bar"));
+  auto blob_files = blob_db_impl()->TEST_GetBlobFiles();
+  auto blob_file1 = blob_files[0];
+  ASSERT_EQ(1, blob_files.size());
+  ASSERT_OK(blob_db_impl()->TEST_CloseBlobFile(blob_file1));
+  GCStats gc_stats;
+  ASSERT_OK(blob_db_impl()->TEST_GCFileAndUpdateLSM(blob_file1, &gc_stats));
+  ASSERT_EQ(1, gc_stats.blob_count);
+  ASSERT_EQ(1, gc_stats.num_keys_relocated);
+  blob_files = blob_db_impl()->TEST_GetBlobFiles();
+  ASSERT_EQ(2, blob_files.size());
+  ASSERT_EQ(blob_file1, blob_files[0]);
+  ASSERT_TRUE(blob_files[1]->Immutable());
+}
+
 // This test is no longer valid since we now return an error when we go
 // over the configured max_db_size.
 // The test needs to be re-written later in such a way that writes continue
@@ -573,6 +573,18 @@ TEST_F(CheckpointTest, CheckpointInvalidDirectoryName) {
  }
 }

+TEST_F(CheckpointTest, CheckpointWithParallelWrites) {
+  // When run with TSAN, this exposes the data race fixed in
+  // https://github.com/facebook/rocksdb/pull/3603
+  ASSERT_OK(Put("key1", "val1"));
+  port::Thread thread([this]() { ASSERT_OK(Put("key2", "val2")); });
+  Checkpoint* checkpoint;
+  ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
+  ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name_));
+  delete checkpoint;
+  thread.join();
+}
+
 }  // namespace rocksdb

 int main(int argc, char** argv) {
@@ -930,6 +930,44 @@ TEST_P(WritePreparedTransactionTest, AdvanceMaxEvictedSeqBasicTest) {
  }
 }

+// This tests that transactions with duplicate keys perform correctly after max
+// is advancing their prepared sequence numbers. This will not be the case if
+// for example the txn does not add the prepared seq for the second sub-batch to
+// the PrepareHeap structure.
+TEST_P(WritePreparedTransactionTest, AdvanceMaxEvictedSeqWithDuplicatesTest) {
+  WriteOptions write_options;
+  TransactionOptions txn_options;
+  Transaction* txn0 = db->BeginTransaction(write_options, txn_options);
+  ASSERT_OK(txn0->SetName("xid"));
+  ASSERT_OK(txn0->Put(Slice("key"), Slice("value1")));
+  ASSERT_OK(txn0->Put(Slice("key"), Slice("value2")));
+  ASSERT_OK(txn0->Prepare());
+
+  WritePreparedTxnDB* wp_db = dynamic_cast<WritePreparedTxnDB*>(db);
+  // Ensure that all the prepared sequence numbers will be removed from the
+  // PrepareHeap.
+  SequenceNumber new_max = wp_db->COMMIT_CACHE_SIZE;
+  wp_db->AdvanceMaxEvictedSeq(0, new_max);
+
+  ReadOptions ropt;
+  PinnableSlice pinnable_val;
+  auto s = db->Get(ropt, db->DefaultColumnFamily(), "key", &pinnable_val);
+  ASSERT_TRUE(s.IsNotFound());
+  delete txn0;
+
+  wp_db->db_impl_->FlushWAL(true);
+  wp_db->TEST_Crash();
+  ReOpenNoDelete();
+  wp_db = dynamic_cast<WritePreparedTxnDB*>(db);
+  wp_db->AdvanceMaxEvictedSeq(0, new_max);
+  s = db->Get(ropt, db->DefaultColumnFamily(), "key", &pinnable_val);
+  ASSERT_TRUE(s.IsNotFound());
+
+  txn0 = db->GetTransactionByName("xid");
+  ASSERT_OK(txn0->Rollback());
+  delete txn0;
+}
+
 TEST_P(WritePreparedTransactionTest, SeqAdvanceConcurrentTest) {
  // Given the sequential run of txns, with this timeout we should never see a
  // deadlock nor a timeout unless we have a key conflict, which should be
@@ -83,7 +83,9 @@ Status WritePreparedTxn::PrepareInternal() {
  // callback otherwise there is a non-zero chance of max dvancing prepare_seq
  // and readers assume the data as committed.
  if (s.ok()) {
-    wpt_db_->AddPrepared(prepare_seq);
+    for (size_t i = 0; i < prepare_batch_cnt_; i++) {
+      wpt_db_->AddPrepared(prepare_seq + i);
+    }
  }
  return s;
 }
@@ -128,9 +130,14 @@ Status WritePreparedTxn::CommitInternal() {
    assert(s.ok());
    commit_batch_cnt = counter.BatchCount();
  }
-  WritePreparedCommitEntryPreReleaseCallback update_commit_map(
-      wpt_db_, db_impl_, prepare_seq, prepare_batch_cnt_, commit_batch_cnt);
+  const bool PREP_HEAP_SKIPPED = true;
  const bool disable_memtable = !includes_data;
+  const bool do_one_write =
+      !db_impl_->immutable_db_options().two_write_queues || disable_memtable;
+  const bool publish_seq = do_one_write;
+  WritePreparedCommitEntryPreReleaseCallback update_commit_map(
+      wpt_db_, db_impl_, prepare_seq, prepare_batch_cnt_, commit_batch_cnt,
+      !PREP_HEAP_SKIPPED, publish_seq);
  uint64_t seq_used = kMaxSequenceNumber;
  // Since the prepared batch is directly written to memtable, there is already
  // a connection between the memtable and its WAL, so there is no need to
@@ -141,6 +148,38 @@ Status WritePreparedTxn::CommitInternal() {
                               zero_log_number, disable_memtable, &seq_used,
                               batch_cnt, &update_commit_map);
  assert(!s.ok() || seq_used != kMaxSequenceNumber);
+  if (LIKELY(do_one_write || !s.ok())) {
+    return s;
+  }  // else do the 2nd write to publish seq
+  // Note: the 2nd write comes with a performance penality. So if we have too
+  // many of commits accompanied with ComitTimeWriteBatch and yet we cannot
+  // enable use_only_the_last_commit_time_batch_for_recovery_ optimization,
+  // two_write_queues should be disabled to avoid many additional writes here.
+  class PublishSeqPreReleaseCallback : public PreReleaseCallback {
+   public:
+    explicit PublishSeqPreReleaseCallback(DBImpl* db_impl)
+        : db_impl_(db_impl) {}
+    virtual Status Callback(SequenceNumber seq, bool is_mem_disabled) override {
+      assert(is_mem_disabled);
+      assert(db_impl_->immutable_db_options().two_write_queues);
+      db_impl_->SetLastPublishedSequence(seq);
+      return Status::OK();
+    }
+
+   private:
+    DBImpl* db_impl_;
+  } publish_seq_callback(db_impl_);
+  WriteBatch empty_batch;
+  empty_batch.PutLogData(Slice());
+  // In the absence of Prepare markers, use Noop as a batch separator
+  WriteBatchInternal::InsertNoop(&empty_batch);
+  const bool DISABLE_MEMTABLE = true;
+  const size_t ONE_BATCH = 1;
+  const uint64_t NO_REF_LOG = 0;
+  s = db_impl_->WriteImpl(write_options_, &empty_batch, nullptr, nullptr,
+                          NO_REF_LOG, DISABLE_MEMTABLE, &seq_used, ONE_BATCH,
+                          &publish_seq_callback);
+  assert(!s.ok() || seq_used != kMaxSequenceNumber);
  return s;
 }

@@ -240,14 +279,14 @@ Status WritePreparedTxn::RollbackInternal() {
  WriteBatchInternal::MarkRollback(&rollback_batch, name_);
  bool do_one_write = !db_impl_->immutable_db_options().two_write_queues;
  const bool DISABLE_MEMTABLE = true;
-  const uint64_t no_log_ref = 0;
+  const uint64_t NO_REF_LOG = 0;
  uint64_t seq_used = kMaxSequenceNumber;
  const size_t ZERO_PREPARES = 0;
  const size_t ONE_BATCH = 1;
  WritePreparedCommitEntryPreReleaseCallback update_commit_map(
      wpt_db_, db_impl_, kMaxSequenceNumber, ZERO_PREPARES, ONE_BATCH);
  s = db_impl_->WriteImpl(write_options_, &rollback_batch, nullptr, nullptr,
-                          no_log_ref, !DISABLE_MEMTABLE, &seq_used, ONE_BATCH,
+                          NO_REF_LOG, !DISABLE_MEMTABLE, &seq_used, ONE_BATCH,
                          do_one_write ? &update_commit_map : nullptr);
  assert(!s.ok() || seq_used != kMaxSequenceNumber);
  if (!s.ok()) {
@@ -256,7 +295,9 @@ Status WritePreparedTxn::RollbackInternal() {
  if (do_one_write) {
    // Mark the txn as rolled back
    uint64_t& rollback_seq = seq_used;
-    wpt_db_->RollbackPrepared(GetId(), rollback_seq);
+    for (size_t i = 0; i < prepare_batch_cnt_; i++) {
+      wpt_db_->RollbackPrepared(GetId() + i, rollback_seq);
+    }
    return s;
  }  // else do the 2nd write for commit
  uint64_t& prepare_seq = seq_used;
@@ -275,13 +316,15 @@ Status WritePreparedTxn::RollbackInternal() {
  // In the absence of Prepare markers, use Noop as a batch separator
  WriteBatchInternal::InsertNoop(&empty_batch);
  s = db_impl_->WriteImpl(write_options_, &empty_batch, nullptr, nullptr,
-                          no_log_ref, DISABLE_MEMTABLE, &seq_used, ONE_BATCH,
+                          NO_REF_LOG, DISABLE_MEMTABLE, &seq_used, ONE_BATCH,
                          &update_commit_map_with_prepare);
  assert(!s.ok() || seq_used != kMaxSequenceNumber);
  // Mark the txn as rolled back
  uint64_t& rollback_seq = seq_used;
  if (s.ok()) {
-    wpt_db_->RollbackPrepared(GetId(), rollback_seq);
+    for (size_t i = 0; i < prepare_batch_cnt_; i++) {
+      wpt_db_->RollbackPrepared(GetId() + i, rollback_seq);
+    }
  }

  return s;
@@ -36,7 +36,10 @@ Status WritePreparedTxnDB::Initialize(
  assert(dbimpl != nullptr);
  auto rtxns = dbimpl->recovered_transactions();
  for (auto rtxn : rtxns) {
-    AddPrepared(rtxn.second->seq_);
+    auto cnt = rtxn.second->batch_cnt_ ? rtxn.second->batch_cnt_ : 1;
+    for (size_t i = 0; i < cnt; i++) {
+      AddPrepared(rtxn.second->seq_ + i);
+    }
  }
  SequenceNumber prev_max = max_evicted_seq_;
  SequenceNumber last_seq = db_impl_->GetLatestSequenceNumber();
@@ -241,6 +241,8 @@ class WritePreparedTxnDB : public PessimisticTransactionDB {
  friend class PreparedHeap_Concurrent_Test;
  friend class WritePreparedTxnDBMock;
  friend class WritePreparedTransactionTest_AdvanceMaxEvictedSeqBasicTest_Test;
+  friend class
+      WritePreparedTransactionTest_AdvanceMaxEvictedSeqWithDuplicatesTest_Test;
  friend class WritePreparedTransactionTest_BasicRecoveryTest_Test;
  friend class WritePreparedTransactionTest_IsInSnapshotEmptyMapTest_Test;
  friend class WritePreparedTransactionTest_OldCommitMapGC_Test;
@@ -459,19 +461,22 @@ class WritePreparedCommitEntryPreReleaseCallback : public PreReleaseCallback {
                                             SequenceNumber prep_seq,
                                             size_t prep_batch_cnt,
                                             size_t data_batch_cnt = 0,
-                                             bool prep_heap_skipped = false)
+                                             bool prep_heap_skipped = false,
+                                             bool publish_seq = true)
      : db_(db),
        db_impl_(db_impl),
        prep_seq_(prep_seq),
        prep_batch_cnt_(prep_batch_cnt),
        data_batch_cnt_(data_batch_cnt),
        prep_heap_skipped_(prep_heap_skipped),
-        includes_data_(data_batch_cnt_ > 0) {
+        includes_data_(data_batch_cnt_ > 0),
+        publish_seq_(publish_seq) {
    assert((prep_batch_cnt_ > 0) != (prep_seq == kMaxSequenceNumber));  // xor
    assert(prep_batch_cnt_ > 0 || data_batch_cnt_ > 0);
  }

-  virtual Status Callback(SequenceNumber commit_seq) override {
+  virtual Status Callback(SequenceNumber commit_seq,
+                          bool is_mem_disabled) override {
    assert(includes_data_ || prep_seq_ != kMaxSequenceNumber);
    const uint64_t last_commit_seq = LIKELY(data_batch_cnt_ <= 1)
                                         ? commit_seq
@@ -492,7 +497,8 @@ class WritePreparedCommitEntryPreReleaseCallback : public PreReleaseCallback {
        db_->AddCommitted(commit_seq + i, last_commit_seq, PREPARE_SKIPPED);
      }
    }
-    if (db_impl_->immutable_db_options().two_write_queues) {
+    if (db_impl_->immutable_db_options().two_write_queues && publish_seq_) {
+      assert(is_mem_disabled);  // implies the 2nd queue
      // Publish the sequence number. We can do that here assuming the callback
      // is invoked only from one write queue, which would guarantee that the
      // publish sequence numbers will be in order, i.e., once a seq is
@@ -517,6 +523,8 @@ class WritePreparedCommitEntryPreReleaseCallback : public PreReleaseCallback {
  // Either because it is commit without prepare or it has a
  // CommitTimeWriteBatch
  bool includes_data_;
+  // Should the callback also publishes the commit seq number
+  bool publish_seq_;
 };

 // Count the number of sub-batches inside a batch. A sub-batch does not have
Author	SHA1	Message	Date
Adam Retter	51e7cb6d39	Some small improvements to the build_tools	2018-03-24 21:33:43 +00:00
Maysam Yabandeh	3e417a6607	WritePrepared Txn: AddPrepared for all sub-batches Summary: Currently AddPrepared is performed only on the first sub-batch if there are duplicate keys in the write batch. This could cause a problem if the transaction takes too long to commit and the seq number of the first sub-patch moved to old_prepared_ but not the seq of the later ones. The patch fixes this by calling AddPrepared for all sub-patches. Closes https://github.com/facebook/rocksdb/pull/3651 Differential Revision: D7388635 Pulled By: maysamyabandeh fbshipit-source-id: 0ccd80c150d9bc42fe955e49ddb9d7ca353067b4	2018-03-23 17:30:04 -07:00
Dmitri Smirnov	d382ae7de6	Imporve perf of random read and insert compare by suggesting inlining to the compiler Summary: Results from 2015 compiler. This improve sequential insert. Random Read results are inconclusive but I hope 2017 will do a better job at inlining. Before: fillseq : 3.638 micros/op 274866 ops/sec; 213.9 MB/s After: fillseq : 3.379 micros/op 295979 ops/sec; 230.3 MB/s Closes https://github.com/facebook/rocksdb/pull/3645 Differential Revision: D7382711 Pulled By: siying fbshipit-source-id: 092a07ffe8a6e598d1226ceff0f11b35e6c5c8e4	2018-03-23 13:26:55 -07:00
Dmitri Smirnov	53d66df0c4	Refactor sync_point to make implementation either customizable or replaceable Summary: Closes https://github.com/facebook/rocksdb/pull/3637 Differential Revision: D7354373 Pulled By: ajkr fbshipit-source-id: 6816c7bbc192ed0fb944942b11c7074bf24eddf1	2018-03-23 12:56:52 -07:00
Sagar Vemuri	a993c0139d	Add 5.11 and 5.12 to tools/check_format_compatible.sh Summary: Closes https://github.com/facebook/rocksdb/pull/3646 Differential Revision: D7384727 Pulled By: sagar0 fbshipit-source-id: f713af7adb2ffea5303bbf0fac8a8a1630af7b38	2018-03-23 12:43:06 -07:00
LingBin	e80709a33a	Avoid adding tombstones of the same file to RangeDelAggregator multiple times Summary: RangeDelAggregator will remember the files whose range tombstones have been added, so the caller can check whether the file has been added before call AddTombstones. Closes https://github.com/facebook/rocksdb/pull/3635 Differential Revision: D7354604 Pulled By: ajkr fbshipit-source-id: 9b9f7ec130556028df417e650711554b46d8d107	2018-03-23 12:43:06 -07:00
Sagar Vemuri	7ffce2805b	Add Java-API-Changes section to History Summary: We have not been updating our HISTORY.md change log with the RocksJava changes. Going forward, lets add Java changes also to HISTORY.md. There is an old java/HISTORY-JAVA.md, but it hasn't been updated in years. It is much easier to remember to update the change log in a single file, HISTORY.md. I added information about shared block cache here, which was introduced in #3623. Closes https://github.com/facebook/rocksdb/pull/3647 Differential Revision: D7384448 Pulled By: sagar0 fbshipit-source-id: 9b6e569f44e6df5cb7ba06413d9975df0b517d20	2018-03-23 12:28:51 -07:00
Radoslaw Zarzynski	09b6bf828a	InlineSkiplist: don't decode keys unnecessarily during comparisons Summary: Summary ======== `InlineSkipList<>::Insert` takes the `key` parameter as a C-string. Then, it performs multiple comparisons with it requiring the `GetLengthPrefixedSlice()` to be spawn in `MemTable::KeyComparator::operator()(const char* prefix_len_key1, const char* prefix_len_key2)` on the same data over and over. The patch tries to optimize that. Rough performance comparison ===== Big keys, no compression. ``` $ ./db_bench --writes 20000000 --benchmarks="fillrandom" --compression_type none -key_size 256 (...) fillrandom : 4.222 micros/op 236836 ops/sec; 80.4 MB/s ``` ``` $ ./db_bench --writes 20000000 --benchmarks="fillrandom" --compression_type none -key_size 256 (...) fillrandom : 4.064 micros/op 246059 ops/sec; 83.5 MB/s ``` TODO ====== In ~~a separated~~ this PR: - [x] Go outside the write path. Maybe even eradicate the C-string-taking variant of `KeyIsAfterNode` entirely. - [x] Try to cache the transformations applied by `KeyComparator` & friends in situations where we havy many comparisons with the same key. Closes https://github.com/facebook/rocksdb/pull/3516 Differential Revision: D7059300 Pulled By: ajkr fbshipit-source-id: 6f027dbb619a488129f79f79b5f7dbe566fb2dbb	2018-03-23 12:14:30 -07:00
Zhongyi Xie	1cbc96d236	FlushReason improvement Summary: Right now flush reason "SuperVersion Change" covers a few different scenarios which is a bit vague. For example, the following db_bench job should trigger "Write Buffer Full" > $ TEST_TMPDIR=/dev/shm ./db_bench -benchmarks=fillrandom -write_buffer_size=1048576 -target_file_size_base=1048576 -max_bytes_for_level_base=4194304 $ grep 'flush_reason' /dev/shm/dbbench/LOG ... 2018/03/06-17:30:42.543638 7f2773b99700 EVENT_LOG_v1 {"time_micros": 1520386242543634, "job": 192, "event": "flush_started", "num_memtables": 1, "num_entries": 7006, "num_deletes": 0, "memory_usage": 1018024, "flush_reason": "SuperVersion Change"} 2018/03/06-17:30:42.569541 7f2773b99700 EVENT_LOG_v1 {"time_micros": 1520386242569536, "job": 193, "event": "flush_started", "num_memtables": 1, "num_entries": 7006, "num_deletes": 0, "memory_usage": 1018104, "flush_reason": "SuperVersion Change"} 2018/03/06-17:30:42.596396 7f2773b99700 EVENT_LOG_v1 {"time_micros": 1520386242596392, "job": 194, "event": "flush_started", "num_memtables": 1, "num_entries": 7008, "num_deletes": 0, "memory_usage": 1018048, "flush_reason": "SuperVersion Change"} 2018/03/06-17:30:42.622444 7f2773b99700 EVENT_LOG_v1 {"time_micros": 1520386242622440, "job": 195, "event": "flush_started", "num_memtables": 1, "num_entries": 7006, "num_deletes": 0, "memory_usage": 1018104, "flush_reason": "SuperVersion Change"} With the fix: > 2018/03/19-14:40:02.341451 7f11dc257700 EVENT_LOG_v1 {"time_micros": 1521495602341444, "job": 98, "event": "flush_started", "num_memtables": 1, "num_entries": 7009, "num_deletes": 0, "memory_usage": 1018008, "flush_reason": "Write Buffer Full"} 2018/03/19-14:40:02.379655 7f11dc257700 EVENT_LOG_v1 {"time_micros": 1521495602379642, "job": 100, "event": "flush_started", "num_memtables": 1, "num_entries": 7006, "num_deletes": 0, "memory_usage": 1018016, "flush_reason": "Write Buffer Full"} 2018/03/19-14:40:02.418479 7f11dc257700 EVENT_LOG_v1 {"time_micros": 1521495602418474, "job": 101, "event": "flush_started", "num_memtables": 1, "num_entries": 7009, "num_deletes": 0, "memory_usage": 1018104, "flush_reason": "Write Buffer Full"} 2018/03/19-14:40:02.455084 7f11dc257700 EVENT_LOG_v1 {"time_micros": 1521495602455079, "job": 102, "event": "flush_started", "num_memtables": 1, "num_entries": 7009, "num_deletes": 0, "memory_usage": 1018048, "flush_reason": "Write Buffer Full"} 2018/03/19-14:40:02.492293 7f11dc257700 EVENT_LOG_v1 {"time_micros": 1521495602492288, "job": 104, "event": "flush_started", "num_memtables": 1, "num_entries": 7007, "num_deletes": 0, "memory_usage": 1018056, "flush_reason": "Write Buffer Full"} 2018/03/19-14:40:02.528720 7f11dc257700 EVENT_LOG_v1 {"time_micros": 1521495602528715, "job": 105, "event": "flush_started", "num_memtables": 1, "num_entries": 7006, "num_deletes": 0, "memory_usage": 1018104, "flush_reason": "Write Buffer Full"} 2018/03/19-14:40:02.566255 7f11dc257700 EVENT_LOG_v1 {"time_micros": 1521495602566238, "job": 107, "event": "flush_started", "num_memtables": 1, "num_entries": 7009, "num_deletes": 0, "memory_usage": 1018112, "flush_reason": "Write Buffer Full"} Closes https://github.com/facebook/rocksdb/pull/3627 Differential Revision: D7328772 Pulled By: miasantreble fbshipit-source-id: 67c94065fbdd36930f09930aad0aaa6d2c152bb8	2018-03-22 18:42:18 -07:00
Andrew Kryczka	82137f0ce8	Add unit test for WAL corruption Summary: Closes https://github.com/facebook/rocksdb/pull/3618 Differential Revision: D7301053 Pulled By: ajkr fbshipit-source-id: a9dde90caa548c294d03d6386f78428c8536ca14	2018-03-22 18:28:01 -07:00
Sagar Vemuri	2e3d407778	Fsync after writing global seq number in ExternalSstFileIngestionJob Summary: Fsync after writing global sequence number to the ingestion file in ExternalSstFileIngestionJob. Otherwise the file metadata could be incorrect. Closes https://github.com/facebook/rocksdb/pull/3644 Differential Revision: D7373813 Pulled By: sagar0 fbshipit-source-id: 4da2c9e71a8beb5c08b4ac955f288ee1576358b8	2018-03-22 17:42:56 -07:00
Andrew Kryczka	4d51feab0b	Rename function for handling WAL write error Summary: It was misnamed. It actually updates `bg_error_` if `PreprocessWrite()` or `WriteToWAL()` fail, not related to the user callback. Closes https://github.com/facebook/rocksdb/pull/3485 Differential Revision: D6955787 Pulled By: ajkr fbshipit-source-id: bd7afc3fdb7a52830c021cbfc25fcbc3ab7d5e10	2018-03-22 15:58:39 -07:00
Siying Dong	118058ba69	SstFileManager: add bytes_max_delete_chunk Summary: Add `bytes_max_delete_chunk` in SstFileManager so that we can drop a large file in multiple batches. Closes https://github.com/facebook/rocksdb/pull/3640 Differential Revision: D7358679 Pulled By: siying fbshipit-source-id: ef17f0da2f5723dbece2669485a9b91b3edc0bb7	2018-03-22 15:58:37 -07:00
Andrew Kryczka	88c3e26cc0	log value of CompressionOptions::zstd_max_train_bytes Summary: Closes https://github.com/facebook/rocksdb/pull/3587 Differential Revision: D7206901 Pulled By: ajkr fbshipit-source-id: 5d4b1a2653627b44aa3c22db7d98c9cd5dcdb67a	2018-03-22 15:13:28 -07:00
Andrew Kryczka	620823f88b	parse CompressionOptions::zstd_max_train_bytes in options string Summary: Closes https://github.com/facebook/rocksdb/pull/3588 Differential Revision: D7208087 Pulled By: ajkr fbshipit-source-id: 688f7a7c447cb17bee1b410d1fd891c0bf966617	2018-03-22 15:13:27 -07:00
Fosco Marotto	de6cf95a53	Update history for future 5.13 release Summary: Closes https://github.com/facebook/rocksdb/pull/3631 Differential Revision: D7367519 Pulled By: gfosco fbshipit-source-id: 57826cc1c9ffc9f2b351075567b8ad929809cb74	2018-03-22 14:59:27 -07:00
Maysam Yabandeh	7429b20e39	WritePrepared Txn: fix race condition on publishing seq Summary: This commit fixes a race condition on calling SetLastPublishedSequence. The function must be called only from the 2nd write queue when two_write_queues is enabled. However there was a bug that would also call it from the main write queue if CommitTimeWriteBatch is provided to the commit request and yet use_only_the_last_commit_time_batch_for_recovery optimization is not enabled. To fix that we penalize the commit request in such cases by doing an additional write solely to publish the seq number from the 2nd queue. Closes https://github.com/facebook/rocksdb/pull/3641 Differential Revision: D7361508 Pulled By: maysamyabandeh fbshipit-source-id: bf8f7a27e5cccf5425dccbce25eb0032e8e5a4d7	2018-03-22 14:43:36 -07:00
Rohan Rathi	fa8c050e9f	Fixed buffer overrun in BackupEngineImpl::BackupMeta::StoreToFile Summary: The 10MB buffer in BackupEngineImpl::BackupMeta::StoreToFile can be corrupted with a large number of files. Added a check to determine current buffer length and append data to file if buffer becomes full. Resolves https://github.com/facebook/rocksdb/issues/3228 Closes https://github.com/facebook/rocksdb/pull/3636 Differential Revision: D7354160 Pulled By: ajkr fbshipit-source-id: eec12d38095a0d17551a4aaee52b99d30a555722	2018-03-22 14:08:10 -07:00
Huachao Huang	7a6353bd1c	Ignore empty filter block when data block is empty Summary: Close https://github.com/facebook/rocksdb/issues/3592 Closes https://github.com/facebook/rocksdb/pull/3614 Differential Revision: D7291706 Pulled By: ajkr fbshipit-source-id: 9dd8f40bd7716588e1e3fd6be0c2bc2766861f8c	2018-03-21 23:13:05 -07:00
QingpingWang	70282cf876	fix behavior does not match name for "IsFileDeletionsEnabled" Summary: for PR https://github.com/facebook/rocksdb/pull/3598 I deleted the original repo for some reason. Sorry for the inconvenience. Closes https://github.com/facebook/rocksdb/pull/3612 Differential Revision: D7291671 Pulled By: ajkr fbshipit-source-id: 918490ba86b13fe450d232af436cbe259d847c64	2018-03-21 22:13:34 -07:00
QingpingWang	2ce8f63f81	C API for PerfContext Summary: This pull request exposes the interface of PerfContext as C API Closes https://github.com/facebook/rocksdb/pull/3607 Differential Revision: D7294225 Pulled By: ajkr fbshipit-source-id: eddcfbc13538f379950b2c8b299486695ffb5e2c	2018-03-21 22:13:34 -07:00
Sagar Vemuri	d5585bb605	Shared block cache in RocksJava Summary: Changes to support sharing block cache using the Java API. Previously DB instances could share the block cache only when the same Options instance is passed to all the DB instances. But now, with this change, it is possible to explicitly create a cache and pass it to multiple options instances, to share the block cache. Implementing this for [Rocksandra](https://github.com/instagram/cassandra/tree/rocks_3.0), but this feature has been requested by many java api users over the years. Closes https://github.com/facebook/rocksdb/pull/3623 Differential Revision: D7305794 Pulled By: sagar0 fbshipit-source-id: 03e4e8ed7aeee6f88bada4a8365d4279ede2ad71	2018-03-21 18:43:05 -07:00
Yi Wu	f1b7b790c9	BlobDB: Fix BlobDBImpl::GCFileAndUpdateLSM issues Summary: * Fix BlobDBImpl::GCFileAndUpdateLSM doesn't close the new file, and the new file will not be able to be garbage collected later. * Fix BlobDBImpl::GCFileAndUpdateLSM doesn't copy over metadata from old file to new file. Closes https://github.com/facebook/rocksdb/pull/3639 Differential Revision: D7355092 Pulled By: yiwu-arbug fbshipit-source-id: 4fa3594ac5ce376bed1af04a545c532cfc0088c4	2018-03-21 16:30:09 -07:00
Jingguo Yao	8823487ff7	doc: fix a typo Summary: s/synchromization/synchronization/ Closes https://github.com/facebook/rocksdb/pull/3583 Differential Revision: D7276596 Pulled By: sagar0 fbshipit-source-id: 552ec6d6935f642e1a3a7c552de6c94441ac50e0	2018-03-21 15:58:58 -07:00
Siying Dong	6383e42362	benchmark.sh to use --max_background_job Summary: Closes https://github.com/facebook/rocksdb/pull/3632 Differential Revision: D7347012 Pulled By: siying fbshipit-source-id: 46230ec4a917ccf4c478825b07e92b4665a4820b	2018-03-20 18:57:55 -07:00
Dmitri Smirnov	61785c73ed	CloseHandle docs says that the return is non-zero, does not say TRUE(1) Summary: say it is TRUE(1). Add assert. Closes https://github.com/facebook/rocksdb/pull/3630 Differential Revision: D7346895 Pulled By: ajkr fbshipit-source-id: a46075aa4dd89f32520230606adecccecc874cdf	2018-03-20 18:43:02 -07:00
Siying Dong	93d52696bf	Memory Problem Of Destorying ColumnFamilyHandle after deleting the CF Summary: When destorying column family handle after the column family has been deleted, the handle may hold share pointers of some objects in ColumnFamilyOptions, but in the destructor, the destructing order may cause some of the objects to be destoryed before being used by the following steps. Fix it by making a copy of the option object and destory it as the last step. Closes https://github.com/facebook/rocksdb/pull/3610 Differential Revision: D7281025 Pulled By: siying fbshipit-source-id: ac18f3b2841788cba4ccfa1abd8d59158c1113bc	2018-03-20 17:13:12 -07:00