Bump version to 5.14.3 and update HISTORY

Summary:
Followup for #4266. There is one more place in **get_context.cc** where **MergeOperator::ShouldMerge** should be called with reversed list of operands.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/4284

Differential Revision: D9380008

Pulled By: sagar0

fbshipit-source-id: 70ec26e607e5b88465e1acbdcd6c6171bd76b9f2

HISTORY.md

@@ -1,7 +1,20 @@
 # Rocksdb Change Log
-## Unreleased
+
+## 5.14.3 (8/21/2018)
 ### Public API Change
-* For users of `Statistics` objects created via `CreateDBStatistics()`, the format of the string returned by its `ToString()` method has changed.
+* The merge operands are passed to `MergeOperator::ShouldMerge` in the reversed order relative to how they were merged (passed to FullMerge or FullMergeV2) for performance reasons
+### Bug Fixes
+* Fixes DBImpl::FindObsoleteFiles() calling GetChildren() on the same path
+
+## 5.14.2 (7/3/2018)
+### Bug Fixes
+* Change default value of `bytes_max_delete_chunk` to 0 in NewSstFileManager() as it doesn't work well with checkpoints.
+* Set DEBUG_LEVEL=0 for RocksJava Mac Release build.
+
+## 5.14.1 (6/20/2018)
+### Bug Fixes
+* Fix block-based table reader pinning blocks throughout its lifetime, causing memory usage increase.
+* Fix bug with prefix search in partition filters where a shared prefix would be ignored from the later partitions. The bug could report an eixstent key as missing. The bug could be triggered if prefix_extractor is set and partition filters is enabled.
 
 ## 5.14.0 (5/16/2018)
 ### Public API Change
@@ -13,7 +26,6 @@
 * Now, `DBOptions::use_direct_io_for_flush_and_compaction` only applies to background writes, and `DBOptions::use_direct_reads` applies to both user reads and background reads. This conforms with Linux's `open(2)` manpage, which advises against simultaneously reading a file in buffered and direct modes, due to possibly undefined behavior and degraded performance.
 * Iterator::Valid() always returns false if !status().ok(). So, now when doing a Seek() followed by some Next()s, there's no need to check status() after every operation.
 * Iterator::Seek()/SeekForPrev()/SeekToFirst()/SeekToLast() always resets status().
-* Introduced `CompressionOptions::kDefaultCompressionLevel`, which is a generic way to tell RocksDB to use the compression library's default level. It is now the default value for `CompressionOptions::level`. Previously the level defaulted to -1, which gave poor compression ratios in ZSTD.
 
 ### New Features
 * Introduce TTL for level compaction so that all files older than ttl go through the compaction process to get rid of old data.
@@ -22,7 +34,6 @@
 * Add `Env::LowerThreadPoolCPUPriority(Priority)` method, which lowers the CPU priority of background (esp. compaction) threads to minimize interference with foreground tasks.
 * Fsync parent directory after deleting a file in delete scheduler.
 * In level-based compaction, if bottom-pri thread pool was setup via `Env::SetBackgroundThreads()`, compactions to the bottom level will be delegated to that thread pool.
-* `prefix_extractor` has been moved from ImmutableCFOptions to MutableCFOptions, meaning it can be dynamically changed without a DB restart.
 
 ### Bug Fixes
 * Fsync after writing global seq number to the ingestion file in ExternalSstFileIngestionJob.
@@ -30,7 +41,7 @@
 * Fix `BackupableDBOptions::max_valid_backups_to_open` to not delete backup files when refcount cannot be accurately determined.
 * Fix memory leak when pin_l0_filter_and_index_blocks_in_cache is used with partitioned filters
 * Disable rollback of merge operands in WritePrepared transactions to work around an issue in MyRocks. It can be enabled back by setting TransactionDBOptions::rollback_merge_operands to true.
-* Fix wrong results by ReverseBytewiseComparator::FindShortSuccessor()
+* Fix bug with prefix search in partition filters where a shared prefix would be ignored from the later partitions. The bug could report an eixstent key as missing. The bug could be triggered if prefix_extractor is set and partition filters is enabled.
 
 ### Java API Changes
 * Add `BlockBasedTableConfig.setBlockCache` to allow sharing a block cache across DB instances.

Makefile

@@ -85,6 +85,10 @@ ifeq ($(MAKECMDGOALS),rocksdbjavastaticrelease)
 	DEBUG_LEVEL=0
 endif
 
+ifeq ($(MAKECMDGOALS),rocksdbjavastaticreleasedocker)
+        DEBUG_LEVEL=0
+endif
+
 ifeq ($(MAKECMDGOALS),rocksdbjavastaticpublish)
 	DEBUG_LEVEL=0
 endif

TARGETS

@@ -28,7 +28,7 @@ rocksdb_compiler_flags = [
     # Added missing flags from output of build_detect_platform
     "-DROCKSDB_PTHREAD_ADAPTIVE_MUTEX",
     "-DROCKSDB_BACKTRACE",
-    "-Wshorten-64-to-32",
+    "-Wnarrowing",
 ]
 
 rocksdb_external_deps = [
@@ -556,13 +556,13 @@ ROCKS_TESTS = [
         "serial",
     ],
     [
-        "db_iter_test",
-        "db/db_iter_test.cc",
+        "db_iter_stress_test",
+        "db/db_iter_stress_test.cc",
         "serial",
     ],
     [
-        "db_iter_stress_test",
-        "db/db_iter_stress_test.cc",
+        "db_iter_test",
+        "db/db_iter_test.cc",
         "serial",
     ],
     [
@@ -1069,14 +1069,3 @@ if not is_opt_mode:
           command = [TEST_RUNNER, BUCK_BINS + test_bin]
         )
 
-custom_unittest(
-    name = "make_rocksdbjavastatic",
-    command = ["internal_repo_rocksdb/make_rocksdbjavastatic.sh"],
-    type = "simple",
-)
-
-custom_unittest(
-    name = "make_rocksdb_lite_release",
-    command = ["internal_repo_rocksdb/make_rocksdb_lite_release.sh"],
-    type = "simple",
-)

buckifier/targets_cfg.py

@@ -32,7 +32,7 @@ rocksdb_compiler_flags = [
     # Added missing flags from output of build_detect_platform
     "-DROCKSDB_PTHREAD_ADAPTIVE_MUTEX",
     "-DROCKSDB_BACKTRACE",
-    "-Wshorten-64-to-32",
+    "-Wnarrowing",
 ]
 
 rocksdb_external_deps = [
@@ -135,15 +135,4 @@ if not is_opt_mode:
           command = [TEST_RUNNER, BUCK_BINS + test_bin]
         )
 
-custom_unittest(
-    name = "make_rocksdbjavastatic",
-    command = ["internal_repo_rocksdb/make_rocksdbjavastatic.sh"],
-    type = "simple",
-)
-
-custom_unittest(
-    name = "make_rocksdb_lite_release",
-    command = ["internal_repo_rocksdb/make_rocksdb_lite_release.sh"],
-    type = "simple",
-)
 """

cache/lru_cache.cc

@@ -99,20 +99,14 @@ void LRUHandleTable::Resize() {
   length_ = new_length;
 }
 
-LRUCacheShard::LRUCacheShard(size_t capacity, bool strict_capacity_limit,
-                             double high_pri_pool_ratio)
-    : capacity_(0),
-      high_pri_pool_usage_(0),
-      strict_capacity_limit_(strict_capacity_limit),
-      high_pri_pool_ratio_(high_pri_pool_ratio),
-      high_pri_pool_capacity_(0),
-      usage_(0),
+LRUCacheShard::LRUCacheShard()
+    : capacity_(0), high_pri_pool_usage_(0), strict_capacity_limit_(false),
+      high_pri_pool_ratio_(0), high_pri_pool_capacity_(0), usage_(0),
       lru_usage_(0) {
   // Make empty circular linked list
   lru_.next = &lru_;
   lru_.prev = &lru_;
   lru_low_pri_ = &lru_;
-  SetCapacity(capacity);
 }
 
 LRUCacheShard::~LRUCacheShard() {}
@@ -250,13 +244,17 @@ void* LRUCacheShard::operator new(size_t size) {
   return port::cacheline_aligned_alloc(size);
 }
 
-void* LRUCacheShard::operator new(size_t /*size*/, void* ptr) { return ptr; }
+void* LRUCacheShard::operator new[](size_t size) {
+  return port::cacheline_aligned_alloc(size);
+}
 
 void LRUCacheShard::operator delete(void *memblock) {
   port::cacheline_aligned_free(memblock);
 }
 
-void LRUCacheShard::operator delete(void* /*memblock*/, void* /*ptr*/) {}
+void LRUCacheShard::operator delete[](void* memblock) {
+  port::cacheline_aligned_free(memblock);
+}
 
 void LRUCacheShard::SetCapacity(size_t capacity) {
   autovector<LRUHandle*> last_reference_list;
@@ -475,21 +473,15 @@ LRUCache::LRUCache(size_t capacity, int num_shard_bits,
                    bool strict_capacity_limit, double high_pri_pool_ratio)
     : ShardedCache(capacity, num_shard_bits, strict_capacity_limit) {
   num_shards_ = 1 << num_shard_bits;
-  shards_ = reinterpret_cast<LRUCacheShard*>(
-      port::cacheline_aligned_alloc(sizeof(LRUCacheShard) * num_shards_));
-  size_t per_shard = (capacity + (num_shards_ - 1)) / num_shards_;
+  shards_ = new LRUCacheShard[num_shards_];
+  SetCapacity(capacity);
+  SetStrictCapacityLimit(strict_capacity_limit);
   for (int i = 0; i < num_shards_; i++) {
-    new (&shards_[i])
-        LRUCacheShard(per_shard, strict_capacity_limit, high_pri_pool_ratio);
+    shards_[i].SetHighPriorityPoolRatio(high_pri_pool_ratio);
   }
 }
 
-LRUCache::~LRUCache() {
-  for (int i = 0; i < num_shards_; i++) {
-    shards_[i].~LRUCacheShard();
-  }
-  port::cacheline_aligned_free(shards_);
-}
+LRUCache::~LRUCache() { delete[] shards_; }
 
 CacheShard* LRUCache::GetShard(int shard) {
   return reinterpret_cast<CacheShard*>(&shards_[shard]);

cache/lru_cache.h

@@ -156,8 +156,7 @@ class LRUHandleTable {
 // A single shard of sharded cache.
 class ALIGN_AS(CACHE_LINE_SIZE) LRUCacheShard : public CacheShard {
  public:
-  LRUCacheShard(size_t capacity, bool strict_capacity_limit,
-                double high_pri_pool_ratio);
+  LRUCacheShard();
   virtual ~LRUCacheShard();
 
   // Separate from constructor so caller can easily make an array of LRUCache
@@ -207,16 +206,13 @@ class ALIGN_AS(CACHE_LINE_SIZE) LRUCacheShard : public CacheShard {
   double GetHighPriPoolRatio();
 
   // Overloading to aligned it to cache line size
-  // They are used by tests.
   void* operator new(size_t);
 
-  // placement new
-  void* operator new(size_t, void*);
+  void* operator new[](size_t);
 
   void operator delete(void *);
 
-  // placement delete, does nothing.
-  void operator delete(void*, void*);
+  void operator delete[](void*);
 
  private:
   void LRU_Remove(LRUHandle* e);

cache/lru_cache_test.cc

@@ -7,7 +7,6 @@
 
 #include <string>
 #include <vector>
-#include "port/port.h"
 #include "util/testharness.h"
 
 namespace rocksdb {
@@ -18,8 +17,19 @@ class LRUCacheTest : public testing::Test {
   ~LRUCacheTest() {}
 
   void NewCache(size_t capacity, double high_pri_pool_ratio = 0.0) {
-    cache_.reset(new LRUCacheShard(capacity, false /*strict_capcity_limit*/,
-                                   high_pri_pool_ratio));
+    cache_.reset(
+#if defined(_MSC_VER)
+#pragma warning(push)
+#pragma warning(disable: 4316) // We've validated the alignment with the new operators
+#endif
+      new LRUCacheShard()
+#if defined(_MSC_VER)
+#pragma warning(pop)
+#endif
+    );
+    cache_->SetCapacity(capacity);
+    cache_->SetStrictCapacityLimit(false);
+    cache_->SetHighPriorityPoolRatio(high_pri_pool_ratio);
   }
 
   void Insert(const std::string& key,
@@ -104,7 +114,7 @@ TEST_F(LRUCacheTest, BasicLRU) {
   ValidateLRUList({"e", "z", "d", "u", "v"});
 }
 
-TEST_F(LRUCacheTest, EntriesWithPriority) {
+TEST_F(LRUCacheTest, MidPointInsertion) {
   // Allocate 2 cache entries to high-pri pool.
   NewCache(5, 0.45);
 

db/builder.cc

@@ -39,7 +39,7 @@ namespace rocksdb {
 class TableFactory;
 
 TableBuilder* NewTableBuilder(
-    const ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
+    const ImmutableCFOptions& ioptions,
     const InternalKeyComparator& internal_comparator,
     const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
         int_tbl_prop_collector_factories,
@@ -52,20 +52,19 @@ TableBuilder* NewTableBuilder(
           TablePropertiesCollectorFactory::Context::kUnknownColumnFamily) ==
          column_family_name.empty());
   return ioptions.table_factory->NewTableBuilder(
-      TableBuilderOptions(ioptions, moptions, internal_comparator,
-                          int_tbl_prop_collector_factories, compression_type,
-                          compression_opts, compression_dict, skip_filters,
-                          column_family_name, level, creation_time,
-                          oldest_key_time),
+      TableBuilderOptions(
+          ioptions, internal_comparator, int_tbl_prop_collector_factories,
+          compression_type, compression_opts, compression_dict, skip_filters,
+          column_family_name, level, creation_time, oldest_key_time),
       column_family_id, file);
 }
 
 Status BuildTable(
     const std::string& dbname, Env* env, const ImmutableCFOptions& ioptions,
-    const MutableCFOptions& mutable_cf_options, const EnvOptions& env_options,
-    TableCache* table_cache, InternalIterator* iter,
-    std::unique_ptr<InternalIterator> range_del_iter, FileMetaData* meta,
-    const InternalKeyComparator& internal_comparator,
+    const MutableCFOptions& /*mutable_cf_options*/,
+    const EnvOptions& env_options, TableCache* table_cache,
+    InternalIterator* iter, std::unique_ptr<InternalIterator> range_del_iter,
+    FileMetaData* meta, const InternalKeyComparator& internal_comparator,
     const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
         int_tbl_prop_collector_factories,
     uint32_t column_family_id, const std::string& column_family_name,
@@ -123,11 +122,10 @@ Status BuildTable(
       file_writer.reset(new WritableFileWriter(std::move(file), env_options,
                                                ioptions.statistics));
       builder = NewTableBuilder(
-          ioptions, mutable_cf_options, internal_comparator,
-          int_tbl_prop_collector_factories, column_family_id,
-          column_family_name, file_writer.get(), compression, compression_opts,
-          level, nullptr /* compression_dict */, false /* skip_filters */,
-          creation_time, oldest_key_time);
+          ioptions, internal_comparator, int_tbl_prop_collector_factories,
+          column_family_id, column_family_name, file_writer.get(), compression,
+          compression_opts, level, nullptr /* compression_dict */,
+          false /* skip_filters */, creation_time, oldest_key_time);
     }
 
     MergeHelper merge(env, internal_comparator.user_comparator(),
@@ -197,8 +195,7 @@ Status BuildTable(
       // to cache it here for further user reads
       std::unique_ptr<InternalIterator> it(table_cache->NewIterator(
           ReadOptions(), env_options, internal_comparator, meta->fd,
-          nullptr /* range_del_agg */,
-          mutable_cf_options.prefix_extractor.get(), nullptr,
+          nullptr /* range_del_agg */, nullptr,
           (internal_stats == nullptr) ? nullptr
                                       : internal_stats->GetFileReadHist(0),
           false /* for_compaction */, nullptr /* arena */,

db/builder.h

@@ -43,7 +43,7 @@ class InternalIterator;
 // @param compression_dict Data for presetting the compression library's
 //    dictionary, or nullptr.
 TableBuilder* NewTableBuilder(
-    const ImmutableCFOptions& options, const MutableCFOptions& moptions,
+    const ImmutableCFOptions& options,
     const InternalKeyComparator& internal_comparator,
     const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
         int_tbl_prop_collector_factories,

db/column_family_test.cc

@@ -299,9 +299,6 @@ class ColumnFamilyTest : public testing::Test {
   }
 
   void PutRandomData(int cf, int num, int key_value_size, bool save = false) {
-    if (cf >= static_cast<int>(keys_.size())) {
-      keys_.resize(cf + 1);
-    }
     for (int i = 0; i < num; ++i) {
       // 10 bytes for key, rest is value
       if (!save) {
@@ -309,7 +306,7 @@ class ColumnFamilyTest : public testing::Test {
                       RandomString(&rnd_, key_value_size - 10)));
       } else {
         std::string key = test::RandomKey(&rnd_, 11);
-        keys_[cf].insert(key);
+        keys_.insert(key);
         ASSERT_OK(Put(cf, key, RandomString(&rnd_, key_value_size - 10)));
       }
     }
@@ -509,7 +506,7 @@ class ColumnFamilyTest : public testing::Test {
 
   std::vector<ColumnFamilyHandle*> handles_;
   std::vector<std::string> names_;
-  std::vector<std::set<std::string>> keys_;
+  std::set<std::string> keys_;
   ColumnFamilyOptions column_family_options_;
   DBOptions db_options_;
   std::string dbname_;
@@ -1417,8 +1414,8 @@ TEST_F(ColumnFamilyTest, MultipleManualCompactions) {
   CompactAll(2);
   AssertFilesPerLevel("0,1", 2);
   // Compare against saved keys
-  std::set<std::string>::iterator key_iter = keys_[1].begin();
-  while (key_iter != keys_[1].end()) {
+  std::set<std::string>::iterator key_iter = keys_.begin();
+  while (key_iter != keys_.end()) {
     ASSERT_NE("NOT_FOUND", Get(1, *key_iter));
     key_iter++;
   }
@@ -1511,8 +1508,8 @@ TEST_F(ColumnFamilyTest, AutomaticAndManualCompactions) {
   CompactAll(2);
   AssertFilesPerLevel("0,1", 2);
   // Compare against saved keys
-  std::set<std::string>::iterator key_iter = keys_[1].begin();
-  while (key_iter != keys_[1].end()) {
+  std::set<std::string>::iterator key_iter = keys_.begin();
+  while (key_iter != keys_.end()) {
     ASSERT_NE("NOT_FOUND", Get(1, *key_iter));
     key_iter++;
   }
@@ -1607,8 +1604,8 @@ TEST_F(ColumnFamilyTest, ManualAndAutomaticCompactions) {
   CompactAll(2);
   AssertFilesPerLevel("0,1", 2);
   // Compare against saved keys
-  std::set<std::string>::iterator key_iter = keys_[1].begin();
-  while (key_iter != keys_[1].end()) {
+  std::set<std::string>::iterator key_iter = keys_.begin();
+  while (key_iter != keys_.end()) {
     ASSERT_NE("NOT_FOUND", Get(1, *key_iter));
     key_iter++;
   }
@@ -1706,8 +1703,8 @@ TEST_F(ColumnFamilyTest, SameCFManualManualCompactions) {
   ASSERT_LE(NumTableFilesAtLevel(0, 1), 2);
 
   // Compare against saved keys
-  std::set<std::string>::iterator key_iter = keys_[1].begin();
-  while (key_iter != keys_[1].end()) {
+  std::set<std::string>::iterator key_iter = keys_.begin();
+  while (key_iter != keys_.end()) {
     ASSERT_NE("NOT_FOUND", Get(1, *key_iter));
     key_iter++;
   }
@@ -1796,8 +1793,8 @@ TEST_F(ColumnFamilyTest, SameCFManualAutomaticCompactions) {
   ASSERT_LE(NumTableFilesAtLevel(0, 1), 2);
 
   // Compare against saved keys
-  std::set<std::string>::iterator key_iter = keys_[1].begin();
-  while (key_iter != keys_[1].end()) {
+  std::set<std::string>::iterator key_iter = keys_.begin();
+  while (key_iter != keys_.end()) {
     ASSERT_NE("NOT_FOUND", Get(1, *key_iter));
     key_iter++;
   }
@@ -1886,8 +1883,8 @@ TEST_F(ColumnFamilyTest, SameCFManualAutomaticCompactionsLevel) {
   AssertFilesPerLevel("0,1", 1);
 
   // Compare against saved keys
-  std::set<std::string>::iterator key_iter = keys_[1].begin();
-  while (key_iter != keys_[1].end()) {
+  std::set<std::string>::iterator key_iter = keys_.begin();
+  while (key_iter != keys_.end()) {
     ASSERT_NE("NOT_FOUND", Get(1, *key_iter));
     key_iter++;
   }
@@ -1975,8 +1972,8 @@ TEST_F(ColumnFamilyTest, SameCFAutomaticManualCompactions) {
   // VERIFY compaction "one"
   AssertFilesPerLevel("1", 1);
   // Compare against saved keys
-  std::set<std::string>::iterator key_iter = keys_[1].begin();
-  while (key_iter != keys_[1].end()) {
+  std::set<std::string>::iterator key_iter = keys_.begin();
+  while (key_iter != keys_.end()) {
     ASSERT_NE("NOT_FOUND", Get(1, *key_iter));
     key_iter++;
   }
@@ -2827,6 +2824,27 @@ TEST_F(ColumnFamilyTest, CreateAndDestoryOptions) {
   ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
 }
 
+TEST_F(ColumnFamilyTest, CreateDropAndDestroy) {
+  ColumnFamilyHandle* cfh;
+  Open();
+  ASSERT_OK(db_->CreateColumnFamily(ColumnFamilyOptions(), "yoyo", &cfh));
+  ASSERT_OK(db_->Put(WriteOptions(), cfh, "foo", "bar"));
+  ASSERT_OK(db_->Flush(FlushOptions(), cfh));
+  ASSERT_OK(db_->DropColumnFamily(cfh));
+  ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
+}
+
+TEST_F(ColumnFamilyTest, CreateDropAndDestroyWithoutFileDeletion) {
+  ColumnFamilyHandle* cfh;
+  Open();
+  ASSERT_OK(db_->CreateColumnFamily(ColumnFamilyOptions(), "yoyo", &cfh));
+  ASSERT_OK(db_->Put(WriteOptions(), cfh, "foo", "bar"));
+  ASSERT_OK(db_->Flush(FlushOptions(), cfh));
+  ASSERT_OK(db_->DisableFileDeletions());
+  ASSERT_OK(db_->DropColumnFamily(cfh));
+  ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
+}
+
 #ifndef ROCKSDB_LITE
 TEST_F(ColumnFamilyTest, FlushCloseWALFiles) {
   SpecialEnv env(Env::Default());
@@ -3192,38 +3210,19 @@ TEST_F(ColumnFamilyTest, MultipleCFPathsTest) {
   CreateColumnFamilies({"one", "two"}, {cf_opt1, cf_opt2});
   Reopen({ColumnFamilyOptions(), cf_opt1, cf_opt2});
 
-  PutRandomData(1, 100, 100, true /* save */);
+  PutRandomData(1, 100, 100);
   Flush(1);
 
   // Check that files are generated in appropriate paths.
   ASSERT_EQ(1, GetSstFileCount(cf_opt1.cf_paths[0].path));
   ASSERT_EQ(0, GetSstFileCount(dbname_));
 
-  PutRandomData(2, 100, 100, true /* save */);
+  PutRandomData(2, 100, 100);
   Flush(2);
 
   ASSERT_EQ(1, GetSstFileCount(cf_opt2.cf_paths[0].path));
   ASSERT_EQ(0, GetSstFileCount(dbname_));
-
-  // Re-open and verify the keys.
-  Reopen({ColumnFamilyOptions(), cf_opt1, cf_opt2});
-  DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
-  for (int cf = 1; cf != 3; ++cf) {
-    ReadOptions read_options;
-    read_options.readahead_size = 0;
-    auto it = dbi->NewIterator(read_options, handles_[cf]);
-    for (it->SeekToFirst(); it->Valid(); it->Next()) {
-      Slice key(it->key());
-      ASSERT_NE(keys_[cf].end(), keys_[cf].find(key.ToString()));
-    }
-    delete it;
-
-    for (const auto& key : keys_[cf]) {
-      ASSERT_NE("NOT_FOUND", Get(cf, key));
-    }
-  }
 }
-
 }  // namespace rocksdb
 
 int main(int argc, char** argv) {

db/compacted_db_impl.cc

@@ -49,8 +49,8 @@ Status CompactedDBImpl::Get(const ReadOptions& options, ColumnFamilyHandle*,
                          GetContext::kNotFound, key, value, nullptr, nullptr,
                          nullptr, nullptr);
   LookupKey lkey(key, kMaxSequenceNumber);
-  files_.files[FindFile(key)].fd.table_reader->Get(options, lkey.internal_key(),
-                                                   &get_context, nullptr);
+  files_.files[FindFile(key)].fd.table_reader->Get(
+      options, lkey.internal_key(), &get_context);
   if (get_context.State() == GetContext::kFound) {
     return Status::OK();
   }
@@ -82,7 +82,7 @@ std::vector<Status> CompactedDBImpl::MultiGet(const ReadOptions& options,
                              GetContext::kNotFound, keys[idx], &pinnable_val,
                              nullptr, nullptr, nullptr, nullptr);
       LookupKey lkey(keys[idx], kMaxSequenceNumber);
-      r->Get(options, lkey.internal_key(), &get_context, nullptr);
+      r->Get(options, lkey.internal_key(), &get_context);
       value.assign(pinnable_val.data(), pinnable_val.size());
       if (get_context.State() == GetContext::kFound) {
         statuses[idx] = Status::OK();

db/compaction_job.cc

@@ -1183,9 +1183,7 @@ Status CompactionJob::FinishCompactionOutputFile(
     // to cache it here for further user reads
     InternalIterator* iter = cfd->table_cache()->NewIterator(
         ReadOptions(), env_options_, cfd->internal_comparator(), meta->fd,
-        nullptr /* range_del_agg */,
-        sub_compact->compaction->mutable_cf_options()->prefix_extractor.get(),
-        nullptr,
+        nullptr /* range_del_agg */, nullptr,
         cfd->internal_stats()->GetFileReadHist(
             compact_->compaction->output_level()),
         false, nullptr /* arena */, false /* skip_filters */,
@@ -1388,10 +1386,9 @@ Status CompactionJob::OpenCompactionOutputFile(
   }
 
   sub_compact->builder.reset(NewTableBuilder(
-      *cfd->ioptions(), *(sub_compact->compaction->mutable_cf_options()),
-      cfd->internal_comparator(), cfd->int_tbl_prop_collector_factories(),
-      cfd->GetID(), cfd->GetName(), sub_compact->outfile.get(),
-      sub_compact->compaction->output_compression(),
+      *cfd->ioptions(), cfd->internal_comparator(),
+      cfd->int_tbl_prop_collector_factories(), cfd->GetID(), cfd->GetName(),
+      sub_compact->outfile.get(), sub_compact->compaction->output_compression(),
       cfd->ioptions()->compression_opts,
       sub_compact->compaction->output_level(), &sub_compact->compression_dict,
       skip_filters, output_file_creation_time));

db/comparator_db_test.cc

@@ -2,7 +2,6 @@
 //  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 <array>
 #include <map>
 #include <string>
 
@@ -434,145 +433,6 @@ TEST_F(ComparatorDBTest, TwoStrComparator) {
   }
 }
 
-TEST_F(ComparatorDBTest, FindShortestSeparator) {
-  std::string s1 = "abc1xyz";
-  std::string s2 = "abc3xy";
-
-  BytewiseComparator()->FindShortestSeparator(&s1, s2);
-  ASSERT_EQ("abc2", s1);
-
-  s1 = "abc5xyztt";
-
-  ReverseBytewiseComparator()->FindShortestSeparator(&s1, s2);
-  ASSERT_EQ("abc5", s1);
-
-  s1 = "abc3";
-  s2 = "abc2xy";
-  ReverseBytewiseComparator()->FindShortestSeparator(&s1, s2);
-  ASSERT_EQ("abc3", s1);
-
-  s1 = "abc3xyz";
-  s2 = "abc2xy";
-  ReverseBytewiseComparator()->FindShortestSeparator(&s1, s2);
-  ASSERT_EQ("abc3", s1);
-
-  s1 = "abc3xyz";
-  s2 = "abc2";
-  ReverseBytewiseComparator()->FindShortestSeparator(&s1, s2);
-  ASSERT_EQ("abc3", s1);
-
-  std::string old_s1 = s1 = "abc2xy";
-  s2 = "abc2";
-  ReverseBytewiseComparator()->FindShortestSeparator(&s1, s2);
-  ASSERT_TRUE(old_s1 >= s1);
-  ASSERT_TRUE(s1 > s2);
-}
-
-TEST_F(ComparatorDBTest, SeparatorSuccessorRandomizeTest) {
-  // Char list for boundary cases.
-  std::array<unsigned char, 6> char_list{{0, 1, 2, 253, 254, 255}};
-  Random rnd(301);
-
-  for (int attempts = 0; attempts < 1000; attempts++) {
-    uint32_t size1 = rnd.Skewed(4);
-    uint32_t size2;
-
-    if (rnd.OneIn(2)) {
-      // size2 to be random size
-      size2 = rnd.Skewed(4);
-    } else {
-      // size1 is within [-2, +2] of size1
-      int diff = static_cast<int>(rnd.Uniform(5)) - 2;
-      int tmp_size2 = static_cast<int>(size1) + diff;
-      if (tmp_size2 < 0) {
-        tmp_size2 = 0;
-      }
-      size2 = static_cast<uint32_t>(tmp_size2);
-    }
-
-    std::string s1;
-    std::string s2;
-    for (uint32_t i = 0; i < size1; i++) {
-      if (rnd.OneIn(2)) {
-        // Use random byte
-        s1 += static_cast<char>(rnd.Uniform(256));
-      } else {
-        // Use one byte in char_list
-        char c = static_cast<char>(char_list[rnd.Uniform(sizeof(char_list))]);
-        s1 += c;
-      }
-    }
-
-    // First set s2 to be the same as s1, and then modify s2.
-    s2 = s1;
-    s2.resize(size2);
-    // We start from the back of the string
-    if (size2 > 0) {
-      uint32_t pos = size2 - 1;
-      do {
-        if (pos >= size1 || rnd.OneIn(4)) {
-          // For 1/4 chance, use random byte
-          s2[pos] = static_cast<char>(rnd.Uniform(256));
-        } else if (rnd.OneIn(4)) {
-          // In 1/4 chance, stop here.
-          break;
-        } else {
-          // Create a char within [-2, +2] of the matching char of s1.
-          int diff = static_cast<int>(rnd.Uniform(5)) - 2;
-          // char may be signed or unsigned based on platform.
-          int s1_char = static_cast<int>(static_cast<unsigned char>(s1[pos]));
-          int s2_char = s1_char + diff;
-          if (s2_char < 0) {
-            s2_char = 0;
-          }
-          if (s2_char > 255) {
-            s2_char = 255;
-          }
-          s2[pos] = static_cast<char>(s2_char);
-        }
-      } while (pos-- != 0);
-    }
-
-    // Test separators
-    for (int rev = 0; rev < 2; rev++) {
-      if (rev == 1) {
-        // switch s1 and s2
-        std::string t = s1;
-        s1 = s2;
-        s2 = t;
-      }
-      std::string separator = s1;
-      BytewiseComparator()->FindShortestSeparator(&separator, s2);
-      std::string rev_separator = s1;
-      ReverseBytewiseComparator()->FindShortestSeparator(&rev_separator, s2);
-
-      if (s1 == s2) {
-        ASSERT_EQ(s1, separator);
-        ASSERT_EQ(s2, rev_separator);
-      } else if (s1 < s2) {
-        ASSERT_TRUE(s1 <= separator);
-        ASSERT_TRUE(s2 > separator);
-        ASSERT_LE(separator.size(), std::max(s1.size(), s2.size()));
-        ASSERT_EQ(s1, rev_separator);
-      } else {
-        ASSERT_TRUE(s1 >= rev_separator);
-        ASSERT_TRUE(s2 < rev_separator);
-        ASSERT_LE(rev_separator.size(), std::max(s1.size(), s2.size()));
-        ASSERT_EQ(s1, separator);
-      }
-    }
-
-    // Test successors
-    std::string succ = s1;
-    BytewiseComparator()->FindShortSuccessor(&succ);
-    ASSERT_TRUE(succ >= s1);
-
-    succ = s1;
-    ReverseBytewiseComparator()->FindShortSuccessor(&succ);
-    ASSERT_TRUE(succ <= s1);
-  }
-}
-
 }  // namespace rocksdb
 
 int main(int argc, char** argv) {

db/convenience.cc

@@ -50,9 +50,8 @@ Status VerifySstFileChecksum(const Options& options,
   std::unique_ptr<RandomAccessFileReader> file_reader(
       new RandomAccessFileReader(std::move(file), file_path));
   s = ioptions.table_factory->NewTableReader(
-      TableReaderOptions(ioptions, options.prefix_extractor.get(), env_options,
-                         internal_comparator, false /* skip_filters */,
-                         -1 /* level */),
+      TableReaderOptions(ioptions, env_options, internal_comparator,
+                         false /* skip_filters */, -1 /* level */),
       std::move(file_reader), file_size, &table_reader,
       false /* prefetch_index_and_filter_in_cache */);
   if (!s.ok()) {

db/db_impl.cc

@@ -969,7 +969,7 @@ InternalIterator* DBImpl::NewInternalIterator(
   MergeIteratorBuilder merge_iter_builder(
       &cfd->internal_comparator(), arena,
       !read_options.total_order_seek &&
-          super_version->mutable_cf_options.prefix_extractor != nullptr);
+          cfd->ioptions()->prefix_extractor != nullptr);
   // Collect iterator for mutable mem
   merge_iter_builder.AddIterator(
       super_version->mem->NewIterator(read_options, arena));
@@ -1568,11 +1568,11 @@ Iterator* DBImpl::NewIterator(const ReadOptions& read_options,
 #else
     SuperVersion* sv = cfd->GetReferencedSuperVersion(&mutex_);
     auto iter = new ForwardIterator(this, read_options, cfd, sv);
-    result = NewDBIterator(
-        env_, read_options, *cfd->ioptions(), sv->mutable_cf_options,
-        cfd->user_comparator(), iter, kMaxSequenceNumber,
-        sv->mutable_cf_options.max_sequential_skip_in_iterations,
-        read_callback);
+    result =
+        NewDBIterator(env_, read_options, *cfd->ioptions(),
+                      cfd->user_comparator(), iter, kMaxSequenceNumber,
+                      sv->mutable_cf_options.max_sequential_skip_in_iterations,
+                      read_callback);
 #endif
   } else {
     // Note: no need to consider the special case of
@@ -1637,7 +1637,7 @@ ArenaWrappedDBIter* DBImpl::NewIteratorImpl(const ReadOptions& read_options,
   // likely that any iterator pointer is close to the iterator it points to so
   // that they are likely to be in the same cache line and/or page.
   ArenaWrappedDBIter* db_iter = NewArenaWrappedDbIterator(
-      env_, read_options, *cfd->ioptions(), sv->mutable_cf_options, snapshot,
+      env_, read_options, *cfd->ioptions(), snapshot,
       sv->mutable_cf_options.max_sequential_skip_in_iterations,
       sv->version_number, read_callback,
       ((read_options.snapshot != nullptr) ? nullptr : this), cfd, allow_blob,
@@ -1688,8 +1688,8 @@ Status DBImpl::NewIterators(
       SuperVersion* sv = cfd->GetReferencedSuperVersion(&mutex_);
       auto iter = new ForwardIterator(this, read_options, cfd, sv);
       iterators->push_back(NewDBIterator(
-          env_, read_options, *cfd->ioptions(), sv->mutable_cf_options,
-          cfd->user_comparator(), iter, kMaxSequenceNumber,
+          env_, read_options, *cfd->ioptions(), cfd->user_comparator(), iter,
+          kMaxSequenceNumber,
           sv->mutable_cf_options.max_sequential_skip_in_iterations,
           read_callback));
     }
@@ -2863,9 +2863,7 @@ Status DBImpl::IngestExternalFile(
     pending_output_elem = CaptureCurrentFileNumberInPendingOutputs();
   }
 
-  SuperVersion* super_version = cfd->GetReferencedSuperVersion(&mutex_);
-  status = ingestion_job.Prepare(external_files, super_version);
-  CleanupSuperVersion(super_version);
+  status = ingestion_job.Prepare(external_files);
   if (!status.ok()) {
     return status;
   }

db/db_impl_files.cc

@@ -12,6 +12,7 @@
 #define __STDC_FORMAT_MACROS
 #endif
 #include <inttypes.h>
+#include <set>
 #include <unordered_set>
 #include "db/event_helpers.h"
 #include "db/memtable_list.h"
@@ -99,10 +100,10 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
   if (doing_the_full_scan) {
     InfoLogPrefix info_log_prefix(!immutable_db_options_.db_log_dir.empty(),
         dbname_);
-    std::vector<std::string> paths;
+    std::set<std::string> paths;
     for (size_t path_id = 0; path_id < immutable_db_options_.db_paths.size();
          path_id++) {
-      paths.emplace_back(immutable_db_options_.db_paths[path_id].path);
+      paths.insert(immutable_db_options_.db_paths[path_id].path);
     }
 
     // Note that if cf_paths is not specified in the ColumnFamilyOptions
@@ -113,7 +114,11 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
     for (auto cfd : *versions_->GetColumnFamilySet()) {
       for (size_t path_id = 0; path_id < cfd->ioptions()->cf_paths.size();
            path_id++) {
-        paths.emplace_back(cfd->ioptions()->cf_paths[path_id].path);
+        auto& path = cfd->ioptions()->cf_paths[path_id].path;
+
+        if (paths.find(path) == paths.end()) {
+          paths.insert(path);
+        }
       }
     }
 

db/db_impl_readonly.cc

@@ -58,7 +58,7 @@ Iterator* DBImplReadOnly::NewIterator(const ReadOptions& read_options,
   SequenceNumber latest_snapshot = versions_->LastSequence();
   ReadCallback* read_callback = nullptr;  // No read callback provided.
   auto db_iter = NewArenaWrappedDbIterator(
-      env_, read_options, *cfd->ioptions(), super_version->mutable_cf_options,
+      env_, read_options, *cfd->ioptions(),
       (read_options.snapshot != nullptr
            ? reinterpret_cast<const SnapshotImpl*>(read_options.snapshot)
                  ->number_
@@ -88,7 +88,7 @@ Status DBImplReadOnly::NewIterators(
     auto* cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(cfh)->cfd();
     auto* sv = cfd->GetSuperVersion()->Ref();
     auto* db_iter = NewArenaWrappedDbIterator(
-        env_, read_options, *cfd->ioptions(), sv->mutable_cf_options,
+        env_, read_options, *cfd->ioptions(),
         (read_options.snapshot != nullptr
              ? reinterpret_cast<const SnapshotImpl*>(read_options.snapshot)
                    ->number_

db/db_impl_write.cc

@@ -211,7 +211,6 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
   // and protects against concurrent loggers and concurrent writes
   // into memtables
 
-  TEST_SYNC_POINT("DBImpl::WriteImpl:BeforeLeaderEnters");
   last_batch_group_size_ =
       write_thread_.EnterAsBatchGroupLeader(&w, &write_group);
 

db/db_iter.cc

@@ -110,8 +110,7 @@ class DBIter final: public Iterator {
   };
 
   DBIter(Env* _env, const ReadOptions& read_options,
-         const ImmutableCFOptions& cf_options,
-         const MutableCFOptions& mutable_cf_options, const Comparator* cmp,
+         const ImmutableCFOptions& cf_options, const Comparator* cmp,
          InternalIterator* iter, SequenceNumber s, bool arena_mode,
          uint64_t max_sequential_skip_in_iterations,
          ReadCallback* read_callback, bool allow_blob)
@@ -139,7 +138,7 @@ class DBIter final: public Iterator {
         is_blob_(false),
         start_seqnum_(read_options.iter_start_seqnum) {
     RecordTick(statistics_, NO_ITERATORS);
-    prefix_extractor_ = mutable_cf_options.prefix_extractor.get();
+    prefix_extractor_ = cf_options.prefix_extractor;
     max_skip_ = max_sequential_skip_in_iterations;
     max_skippable_internal_keys_ = read_options.max_skippable_internal_keys;
     if (pin_thru_lifetime_) {
@@ -1427,15 +1426,14 @@ void DBIter::SeekToLast() {
 
 Iterator* NewDBIterator(Env* env, const ReadOptions& read_options,
                         const ImmutableCFOptions& cf_options,
-                        const MutableCFOptions& mutable_cf_options,
                         const Comparator* user_key_comparator,
                         InternalIterator* internal_iter,
                         const SequenceNumber& sequence,
                         uint64_t max_sequential_skip_in_iterations,
                         ReadCallback* read_callback, bool allow_blob) {
   DBIter* db_iter =
-      new DBIter(env, read_options, cf_options, mutable_cf_options,
-                 user_key_comparator, internal_iter, sequence, false,
+      new DBIter(env, read_options, cf_options, user_key_comparator,
+                 internal_iter, sequence, false,
                  max_sequential_skip_in_iterations, read_callback, allow_blob);
   return db_iter;
 }
@@ -1479,7 +1477,6 @@ inline Status ArenaWrappedDBIter::GetProperty(std::string prop_name,
 
 void ArenaWrappedDBIter::Init(Env* env, const ReadOptions& read_options,
                               const ImmutableCFOptions& cf_options,
-                              const MutableCFOptions& mutable_cf_options,
                               const SequenceNumber& sequence,
                               uint64_t max_sequential_skip_in_iteration,
                               uint64_t version_number,
@@ -1487,9 +1484,9 @@ void ArenaWrappedDBIter::Init(Env* env, const ReadOptions& read_options,
                               bool allow_refresh) {
   auto mem = arena_.AllocateAligned(sizeof(DBIter));
   db_iter_ = new (mem)
-      DBIter(env, read_options, cf_options, mutable_cf_options,
-             cf_options.user_comparator, nullptr, sequence, true,
-             max_sequential_skip_in_iteration, read_callback, allow_blob);
+      DBIter(env, read_options, cf_options, cf_options.user_comparator, nullptr,
+             sequence, true, max_sequential_skip_in_iteration, read_callback,
+             allow_blob);
   sv_number_ = version_number;
   allow_refresh_ = allow_refresh;
 }
@@ -1511,8 +1508,8 @@ Status ArenaWrappedDBIter::Refresh() {
     new (&arena_) Arena();
 
     SuperVersion* sv = cfd_->GetReferencedSuperVersion(db_impl_->mutex());
-    Init(env, read_options_, *(cfd_->ioptions()), sv->mutable_cf_options,
-         latest_seq, sv->mutable_cf_options.max_sequential_skip_in_iterations,
+    Init(env, read_options_, *(cfd_->ioptions()), latest_seq,
+         sv->mutable_cf_options.max_sequential_skip_in_iterations,
          cur_sv_number, read_callback_, allow_blob_, allow_refresh_);
 
     InternalIterator* internal_iter = db_impl_->NewInternalIterator(
@@ -1527,13 +1524,12 @@ Status ArenaWrappedDBIter::Refresh() {
 
 ArenaWrappedDBIter* NewArenaWrappedDbIterator(
     Env* env, const ReadOptions& read_options,
-    const ImmutableCFOptions& cf_options,
-    const MutableCFOptions& mutable_cf_options, const SequenceNumber& sequence,
+    const ImmutableCFOptions& cf_options, const SequenceNumber& sequence,
     uint64_t max_sequential_skip_in_iterations, uint64_t version_number,
     ReadCallback* read_callback, DBImpl* db_impl, ColumnFamilyData* cfd,
     bool allow_blob, bool allow_refresh) {
   ArenaWrappedDBIter* iter = new ArenaWrappedDBIter();
-  iter->Init(env, read_options, cf_options, mutable_cf_options, sequence,
+  iter->Init(env, read_options, cf_options, sequence,
              max_sequential_skip_in_iterations, version_number, read_callback,
              allow_blob, allow_refresh);
   if (db_impl != nullptr && cfd != nullptr && allow_refresh) {

db/db_iter.h

@@ -30,7 +30,6 @@ class InternalIterator;
 // into appropriate user keys.
 extern Iterator* NewDBIterator(Env* env, const ReadOptions& read_options,
                                const ImmutableCFOptions& cf_options,
-                               const MutableCFOptions& mutable_cf_options,
                                const Comparator* user_key_comparator,
                                InternalIterator* internal_iter,
                                const SequenceNumber& sequence,
@@ -72,7 +71,6 @@ class ArenaWrappedDBIter : public Iterator {
 
   void Init(Env* env, const ReadOptions& read_options,
             const ImmutableCFOptions& cf_options,
-            const MutableCFOptions& mutable_cf_options,
             const SequenceNumber& sequence,
             uint64_t max_sequential_skip_in_iterations, uint64_t version_number,
             ReadCallback* read_callback, bool allow_blob, bool allow_refresh);
@@ -104,8 +102,7 @@ class ArenaWrappedDBIter : public Iterator {
 // be supported.
 extern ArenaWrappedDBIter* NewArenaWrappedDbIterator(
     Env* env, const ReadOptions& read_options,
-    const ImmutableCFOptions& cf_options,
-    const MutableCFOptions& mutable_cf_options, const SequenceNumber& sequence,
+    const ImmutableCFOptions& cf_options, const SequenceNumber& sequence,
     uint64_t max_sequential_skip_in_iterations, uint64_t version_number,
     ReadCallback* read_callback, DBImpl* db_impl = nullptr,
     ColumnFamilyData* cfd = nullptr, bool allow_blob = false,

db/db_iter_stress_test.cc

@@ -173,7 +173,7 @@ struct StressTestIterator : public InternalIterator {
     }
   }
 
-  void SeekToFirst() override {
+  void SeekToFirst() {
     if (MaybeFail()) return;
     MaybeMutate();
 
@@ -181,7 +181,7 @@ struct StressTestIterator : public InternalIterator {
     iter = 0;
     SkipForward();
   }
-  void SeekToLast() override {
+  void SeekToLast() {
     if (MaybeFail()) return;
     MaybeMutate();
 
@@ -190,7 +190,7 @@ struct StressTestIterator : public InternalIterator {
     SkipBackward();
   }
 
-  void Seek(const Slice& target) override {
+  void Seek(const Slice& target) {
     if (MaybeFail()) return;
     MaybeMutate();
 
@@ -202,7 +202,7 @@ struct StressTestIterator : public InternalIterator {
     iter = (int)(it - data->entries.begin());
     SkipForward();
   }
-  void SeekForPrev(const Slice& target) override {
+  void SeekForPrev(const Slice& target) {
     if (MaybeFail()) return;
     MaybeMutate();
 
@@ -216,14 +216,14 @@ struct StressTestIterator : public InternalIterator {
     SkipBackward();
   }
 
-  void Next() override {
+  void Next() {
     assert(Valid());
     if (MaybeFail()) return;
     MaybeMutate();
     ++iter;
     SkipForward();
   }
-  void Prev() override {
+  void Prev() {
     assert(Valid());
     if (MaybeFail()) return;
     MaybeMutate();
@@ -231,11 +231,11 @@ struct StressTestIterator : public InternalIterator {
     SkipBackward();
   }
 
-  Slice key() const override {
+  Slice key() const {
     assert(Valid());
     return data->entries[iter].ikey;
   }
-  Slice value() const override {
+  Slice value() const {
     assert(Valid());
     return data->entries[iter].value;
   }
@@ -404,14 +404,13 @@ TEST_F(DBIteratorStressTest, StressTest) {
   Random64 rnd(826909345792864532ll);
 
   auto gen_key = [&](int max_key) {
-    assert(max_key > 0);
     int len = 0;
     int a = max_key;
     while (a) {
       a /= 10;
       ++len;
     }
-    std::string s = ToString(rnd.Next() % static_cast<uint64_t>(max_key));
+    std::string s = ToString(rnd.Next() % (uint64_t)max_key);
     s.insert(0, len - (int)s.size(), '0');
     return s;
   };
@@ -509,8 +508,7 @@ TEST_F(DBIteratorStressTest, StressTest) {
                   internal_iter->trace = trace;
                   db_iter.reset(NewDBIterator(
                       env_, ropt, ImmutableCFOptions(options),
-                      MutableCFOptions(options), BytewiseComparator(),
-                      internal_iter, sequence,
+                      BytewiseComparator(), internal_iter, sequence,
                       options.max_sequential_skip_in_iterations,
                       nullptr /*read_callback*/));
                 }

db/db_iter_test.cc

@@ -36,7 +36,9 @@ class TestIterator : public InternalIterator {
         valid_(false),
         sequence_number_(0),
         iter_(0),
-        cmp(comparator) {}
+        cmp(comparator) {
+    data_.reserve(16);
+  }
 
   void AddPut(std::string argkey, std::string argvalue) {
     Add(argkey, kTypeValue, argvalue);
@@ -234,7 +236,7 @@ class DBIteratorTest : public testing::Test {
 TEST_F(DBIteratorTest, DBIteratorPrevNext) {
   Options options;
   ImmutableCFOptions cf_options = ImmutableCFOptions(options);
-  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
+
   {
     TestIterator* internal_iter = new TestIterator(BytewiseComparator());
     internal_iter->AddDeletion("a");
@@ -248,9 +250,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -281,9 +282,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -308,9 +308,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -341,9 +340,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -377,9 +375,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(!db_iter->Valid());
@@ -407,9 +404,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 7, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 7,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     SetPerfLevel(kEnableCount);
     ASSERT_TRUE(GetPerfLevel() == kEnableCount);
@@ -445,9 +441,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 4, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 4,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -471,9 +466,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(!db_iter->Valid());
@@ -494,9 +488,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -530,9 +523,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 7, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 7,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     SetPerfLevel(kEnableCount);
     ASSERT_TRUE(GetPerfLevel() == kEnableCount);
@@ -560,9 +552,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -603,9 +594,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 2, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 2,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "b");
@@ -635,9 +625,8 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "c");
@@ -658,7 +647,6 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 TEST_F(DBIteratorTest, DBIteratorEmpty) {
   Options options;
   ImmutableCFOptions cf_options = ImmutableCFOptions(options);
-  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
   ReadOptions ro;
 
   {
@@ -666,9 +654,8 @@ TEST_F(DBIteratorTest, DBIteratorEmpty) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 0, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(!db_iter->Valid());
   }
@@ -678,9 +665,8 @@ TEST_F(DBIteratorTest, DBIteratorEmpty) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 0, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToFirst();
     ASSERT_TRUE(!db_iter->Valid());
   }
@@ -700,10 +686,10 @@ TEST_F(DBIteratorTest, DBIteratorUseSkipCountSkips) {
   }
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 2,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 2, options.max_sequential_skip_in_iterations,
+                    nullptr /*read_callback*/));
   db_iter->SeekToLast();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "c");
@@ -732,7 +718,6 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
   Options options;
   options.merge_operator = MergeOperators::CreateFromStringId("stringappend");
   ImmutableCFOptions cf_options = ImmutableCFOptions(options);
-  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
 
   {
     for (size_t i = 0; i < 200; ++i) {
@@ -746,8 +731,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
 
       options.statistics = rocksdb::CreateDBStatistics();
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-          internal_iter, i + 2, options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, BytewiseComparator(), internal_iter, i + 2,
+          options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -782,8 +767,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->Finish();
 
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-          internal_iter, i + 2, options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, BytewiseComparator(), internal_iter, i + 2,
+          options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -811,8 +796,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->Finish();
 
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-          internal_iter, 202, options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, BytewiseComparator(), internal_iter, 202,
+          options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -844,8 +829,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->AddPut("c", "200");
       internal_iter->Finish();
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-          internal_iter, i, options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, BytewiseComparator(), internal_iter, i,
+          options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(!db_iter->Valid());
@@ -861,9 +846,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
     internal_iter->AddPut("c", "200");
     internal_iter->Finish();
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 200, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 200,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "c");
@@ -896,8 +880,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->Finish();
 
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-          internal_iter, i + 2, options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, BytewiseComparator(), internal_iter, i + 2,
+          options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -931,8 +915,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->Finish();
 
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-          internal_iter, i + 2, options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, BytewiseComparator(), internal_iter, i + 2,
+          options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -963,7 +947,6 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
 TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
   Options options;
   ImmutableCFOptions cf_options = ImmutableCFOptions(options);
-  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
   ReadOptions ro;
 
   // Basic test case ... Make sure explicityly passing the default value works.
@@ -981,9 +964,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 0;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1028,9 +1010,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1073,9 +1054,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1112,9 +1092,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1148,9 +1127,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -1179,9 +1157,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1217,9 +1194,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1255,8 +1231,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
       ro.max_skippable_internal_keys = i;
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-          internal_iter, 2 * i + 1, options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, BytewiseComparator(), internal_iter, 2 * i + 1,
+          options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
 
       db_iter->SeekToFirst();
@@ -1309,8 +1285,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
       options.max_sequential_skip_in_iterations = 1000;
       ro.max_skippable_internal_keys = i;
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-          internal_iter, 2 * i + 1, options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, BytewiseComparator(), internal_iter, 2 * i + 1,
+          options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
 
       db_iter->SeekToFirst();
@@ -1347,10 +1323,10 @@ TEST_F(DBIteratorTest, DBIterator1) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 1,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 1, options.max_sequential_skip_in_iterations,
+                    nullptr /*read_callback*/));
   db_iter->SeekToFirst();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1375,10 +1351,10 @@ TEST_F(DBIteratorTest, DBIterator2) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 0,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 0, options.max_sequential_skip_in_iterations,
+                    nullptr /*read_callback*/));
   db_iter->SeekToFirst();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1400,10 +1376,10 @@ TEST_F(DBIteratorTest, DBIterator3) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 2,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 2, options.max_sequential_skip_in_iterations,
+                    nullptr /*read_callback*/));
   db_iter->SeekToFirst();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1425,10 +1401,10 @@ TEST_F(DBIteratorTest, DBIterator4) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 4,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 4, options.max_sequential_skip_in_iterations,
+                    nullptr /*read_callback*/));
   db_iter->SeekToFirst();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1446,7 +1422,6 @@ TEST_F(DBIteratorTest, DBIterator5) {
   Options options;
   options.merge_operator = MergeOperators::CreateFromStringId("stringappend");
   ImmutableCFOptions cf_options = ImmutableCFOptions(options);
-  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
 
   {
     TestIterator* internal_iter = new TestIterator(BytewiseComparator());
@@ -1460,9 +1435,8 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 0, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1483,9 +1457,8 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 1, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 1,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1506,9 +1479,8 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 2, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 2,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1529,9 +1501,8 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 3, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 3,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1552,9 +1523,8 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 4, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 4,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1575,9 +1545,8 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 5, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 5,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1598,9 +1567,8 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 6, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 6,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1619,9 +1587,8 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->AddPut("b", "val_b");
     internal_iter->Finish();
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->Seek("b");
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "b");
@@ -1636,7 +1603,6 @@ TEST_F(DBIteratorTest, DBIterator6) {
   Options options;
   options.merge_operator = MergeOperators::CreateFromStringId("stringappend");
   ImmutableCFOptions cf_options = ImmutableCFOptions(options);
-  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
 
   {
     TestIterator* internal_iter = new TestIterator(BytewiseComparator());
@@ -1650,9 +1616,8 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 0, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1673,9 +1638,8 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 1, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 1,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1696,9 +1660,8 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 2, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 2,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1719,9 +1682,8 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 3, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 3,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(!db_iter->Valid());
   }
@@ -1738,9 +1700,8 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 4, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 4,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1761,9 +1722,8 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 5, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 5,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1784,9 +1744,8 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 6, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 6,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1801,7 +1760,6 @@ TEST_F(DBIteratorTest, DBIterator7) {
   Options options;
   options.merge_operator = MergeOperators::CreateFromStringId("stringappend");
   ImmutableCFOptions cf_options = ImmutableCFOptions(options);
-  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
 
   {
     TestIterator* internal_iter = new TestIterator(BytewiseComparator());
@@ -1827,9 +1785,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 0, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
     ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -1862,9 +1819,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 2, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 2,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -1903,9 +1859,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 4, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 4,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -1944,9 +1899,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 5, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 5,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -1990,9 +1944,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 6, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 6,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2037,9 +1990,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 7, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 7,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2078,9 +2030,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 9, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 9,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2125,9 +2076,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 13, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 13,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2173,9 +2123,8 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
-        internal_iter, 14, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, cf_options, BytewiseComparator(), internal_iter, 14,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2204,9 +2153,9 @@ TEST_F(DBIteratorTest, DBIterator8) {
   internal_iter->Finish();
 
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 10,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+      internal_iter, 10, options.max_sequential_skip_in_iterations,
+      nullptr /*read_callback*/));
   db_iter->SeekToLast();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "b");
@@ -2235,9 +2184,9 @@ TEST_F(DBIteratorTest, DBIterator9) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-        BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -2302,9 +2251,9 @@ TEST_F(DBIteratorTest, DBIterator10) {
   internal_iter->Finish();
 
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 10,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+      internal_iter, 10, options.max_sequential_skip_in_iterations,
+      nullptr /*read_callback*/));
 
   db_iter->Seek("c");
   ASSERT_TRUE(db_iter->Valid());
@@ -2342,9 +2291,8 @@ TEST_F(DBIteratorTest, SeekToLastOccurrenceSeq0) {
   internal_iter->Finish();
 
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 10, 0 /* force seek */,
-      nullptr /*read_callback*/));
+      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+      internal_iter, 10, 0 /* force seek */, nullptr /*read_callback*/));
   db_iter->SeekToFirst();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -2370,10 +2318,10 @@ TEST_F(DBIteratorTest, DBIterator11) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 1,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 1, options.max_sequential_skip_in_iterations,
+                    nullptr /*read_callback*/));
   db_iter->SeekToFirst();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "a");
@@ -2397,9 +2345,9 @@ TEST_F(DBIteratorTest, DBIterator12) {
   internal_iter->AddSingleDeletion("b");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 10, 0, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 10, 0, nullptr /*read_callback*/));
   db_iter->SeekToLast();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "c");
@@ -2434,9 +2382,9 @@ TEST_F(DBIteratorTest, DBIterator13) {
   internal_iter->AddPut(key, "8");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 2, 3, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 2, 3, nullptr /*read_callback*/));
   db_iter->Seek("b");
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), key);
@@ -2462,9 +2410,9 @@ TEST_F(DBIteratorTest, DBIterator14) {
   internal_iter->AddPut("c", "9");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 4, 1, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 4, 1, nullptr /*read_callback*/));
   db_iter->Seek("b");
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "b");
@@ -2489,10 +2437,10 @@ TEST_F(DBIteratorTest, DBIteratorTestDifferentialSnapshots) {
     }
     internal_iter->Finish();
 
-    std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-        BytewiseComparator(), internal_iter, 13,
-        options.max_sequential_skip_in_iterations, nullptr));
+    std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 13,
+                    options.max_sequential_skip_in_iterations, nullptr));
     // Expecting InternalKeys in [5,8] range with correct type
     int seqnums[4] = {5,8,11,13};
     std::string user_keys[4] = {"1","2","3","4"};
@@ -2524,10 +2472,10 @@ TEST_F(DBIteratorTest, DBIteratorTestDifferentialSnapshots) {
     }
     internal_iter->Finish();
 
-    std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-        BytewiseComparator(), internal_iter, 13,
-        options.max_sequential_skip_in_iterations, nullptr));
+    std::unique_ptr<Iterator> db_iter(
+      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+                    internal_iter, 13,
+                    options.max_sequential_skip_in_iterations, nullptr));
     // Expecting InternalKeys in [5,8] range with correct type
     int seqnums[4] = {5,8,11,13};
     EntryType key_types[4] = {EntryType::kEntryDelete,EntryType::kEntryDelete,
@@ -2575,9 +2523,8 @@ class DBIterWithMergeIterTest : public testing::Test {
         NewMergingIterator(&icomp_, &child_iters[0], 2u);
 
     db_iter_.reset(NewDBIterator(
-        env_, ro_, ImmutableCFOptions(options_), MutableCFOptions(options_),
-        BytewiseComparator(), merge_iter,
-        8 /* read data earlier than seqId 8 */,
+        env_, ro_, ImmutableCFOptions(options_), BytewiseComparator(),
+        merge_iter, 8 /* read data earlier than seqId 8 */,
         3 /* max iterators before reseek */, nullptr /*read_callback*/));
   }
 
@@ -2660,7 +2607,7 @@ TEST_F(DBIterWithMergeIterTest, InnerMergeIteratorDataRace1) {
   // MergeIterator::Prev() realized the mem table iterator is at its end
   // and before an SeekToLast() is called.
   rocksdb::SyncPoint::GetInstance()->SetCallBack(
-      "MergeIterator::Prev:BeforeSeekToLast",
+      "MergeIterator::Prev:BeforePrev",
       [&](void* /*arg*/) { internal_iter2_->Add("z", kTypeValue, "7", 12u); });
   rocksdb::SyncPoint::GetInstance()->EnableProcessing();
 
@@ -2696,7 +2643,7 @@ TEST_F(DBIterWithMergeIterTest, InnerMergeIteratorDataRace2) {
   // mem table after MergeIterator::Prev() realized the mem tableiterator is at
   // its end and before an SeekToLast() is called.
   rocksdb::SyncPoint::GetInstance()->SetCallBack(
-      "MergeIterator::Prev:BeforeSeekToLast", [&](void* /*arg*/) {
+      "MergeIterator::Prev:BeforePrev", [&](void* /*arg*/) {
         internal_iter2_->Add("z", kTypeValue, "7", 12u);
         internal_iter2_->Add("z", kTypeValue, "7", 11u);
       });
@@ -2734,7 +2681,7 @@ TEST_F(DBIterWithMergeIterTest, InnerMergeIteratorDataRace3) {
   // mem table after MergeIterator::Prev() realized the mem table iterator is at
   // its end and before an SeekToLast() is called.
   rocksdb::SyncPoint::GetInstance()->SetCallBack(
-      "MergeIterator::Prev:BeforeSeekToLast", [&](void* /*arg*/) {
+      "MergeIterator::Prev:BeforePrev", [&](void* /*arg*/) {
         internal_iter2_->Add("z", kTypeValue, "7", 16u, true);
         internal_iter2_->Add("z", kTypeValue, "7", 15u, true);
         internal_iter2_->Add("z", kTypeValue, "7", 14u, true);
@@ -3015,9 +2962,9 @@ TEST_F(DBIteratorTest, SeekPrefixTombstones) {
 
   ro.prefix_same_as_start = true;
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 10,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+      internal_iter, 10, options.max_sequential_skip_in_iterations,
+      nullptr /*read_callback*/));
 
   int skipped_keys = 0;
 
@@ -3051,10 +2998,9 @@ TEST_F(DBIteratorTest, SeekToFirstLowerBound) {
     ro.iterate_lower_bound = &lower_bound;
     Options options;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-        BytewiseComparator(), internal_iter, 10 /* sequence */,
-        options.max_sequential_skip_in_iterations,
-        nullptr /* read_callback */));
+        env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+        internal_iter, 10 /* sequence */,
+        options.max_sequential_skip_in_iterations, nullptr /* read_callback */));
 
     db_iter->SeekToFirst();
     if (i == kNumKeys + 1) {
@@ -3090,8 +3036,8 @@ TEST_F(DBIteratorTest, PrevLowerBound) {
   ro.iterate_lower_bound = &lower_bound;
   Options options;
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 10 /* sequence */,
+      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+      internal_iter, 10 /* sequence */,
       options.max_sequential_skip_in_iterations, nullptr /* read_callback */));
 
   db_iter->SeekToLast();
@@ -3118,8 +3064,8 @@ TEST_F(DBIteratorTest, SeekLessLowerBound) {
   ro.iterate_lower_bound = &lower_bound;
   Options options;
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
-      BytewiseComparator(), internal_iter, 10 /* sequence */,
+      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
+      internal_iter, 10 /* sequence */,
       options.max_sequential_skip_in_iterations, nullptr /* read_callback */));
 
   auto before_lower_bound_str = std::to_string(kLowerBound - 1);
@@ -3143,9 +3089,9 @@ TEST_F(DBIteratorTest, ReverseToForwardWithDisappearingKeys) {
   internal_iter->Finish();
 
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ReadOptions(), ImmutableCFOptions(options),
-      MutableCFOptions(options), BytewiseComparator(), internal_iter, 10,
-      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+      env_, ReadOptions(), ImmutableCFOptions(options), BytewiseComparator(),
+      internal_iter, 10, options.max_sequential_skip_in_iterations,
+      nullptr /*read_callback*/));
 
   db_iter->SeekForPrev("a");
   ASSERT_TRUE(db_iter->Valid());

db/db_iterator_test.cc

@@ -862,8 +862,6 @@ TEST_P(DBIteratorTest, IteratorPinsRef) {
   } while (ChangeCompactOptions());
 }
 
-// SetOptions not defined in ROCKSDB LITE
-#ifndef ROCKSDB_LITE
 TEST_P(DBIteratorTest, DBIteratorBoundTest) {
   Options options = CurrentOptions();
   options.env = env_;
@@ -948,7 +946,9 @@ TEST_P(DBIteratorTest, DBIteratorBoundTest) {
   }
 
   // prefix is the first letter of the key
-  ASSERT_OK(dbfull()->SetOptions({{"prefix_extractor", "fixed:1"}}));
+  options.prefix_extractor.reset(NewFixedPrefixTransform(1));
+
+  DestroyAndReopen(options);
   ASSERT_OK(Put("a", "0"));
   ASSERT_OK(Put("foo", "bar"));
   ASSERT_OK(Put("foo1", "bar1"));
@@ -1035,7 +1035,6 @@ TEST_P(DBIteratorTest, DBIteratorBoundTest) {
     ASSERT_EQ(static_cast<int>(get_perf_context()->internal_delete_skipped_count), 0);
   }
 }
-#endif
 
 TEST_P(DBIteratorTest, DBIteratorBoundOptimizationTest) {
   int upper_bound_hits = 0;
@@ -2044,6 +2043,43 @@ TEST_P(DBIteratorTest, CreationFailure) {
   delete iter;
 }
 
+TEST_P(DBIteratorTest, UpperBoundWithChangeDirection) {
+  Options options = CurrentOptions();
+  options.max_sequential_skip_in_iterations = 3;
+  DestroyAndReopen(options);
+
+  // write a bunch of kvs to the database.
+  ASSERT_OK(Put("a", "1"));
+  ASSERT_OK(Put("y", "1"));
+  ASSERT_OK(Put("y1", "1"));
+  ASSERT_OK(Put("y2", "1"));
+  ASSERT_OK(Put("y3", "1"));
+  ASSERT_OK(Put("z", "1"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(Put("a", "1"));
+  ASSERT_OK(Put("z", "1"));
+  ASSERT_OK(Put("bar", "1"));
+  ASSERT_OK(Put("foo", "1"));
+
+  std::string upper_bound = "x";
+  Slice ub_slice(upper_bound);
+  ReadOptions ro;
+  ro.iterate_upper_bound = &ub_slice;
+  ro.max_skippable_internal_keys = 1000;
+
+  Iterator* iter = NewIterator(ro);
+  iter->Seek("foo");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ("foo", iter->key().ToString());
+
+  iter->Prev();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_EQ("bar", iter->key().ToString());
+
+  delete iter;
+}
+
 TEST_P(DBIteratorTest, TableFilter) {
   ASSERT_OK(Put("a", "1"));
   dbfull()->Flush(FlushOptions());
@@ -2110,6 +2146,47 @@ TEST_P(DBIteratorTest, TableFilter) {
   }
 }
 
+TEST_P(DBIteratorTest, UpperBoundWithPrevReseek) {
+  Options options = CurrentOptions();
+  options.max_sequential_skip_in_iterations = 3;
+  DestroyAndReopen(options);
+
+  // write a bunch of kvs to the database.
+  ASSERT_OK(Put("a", "1"));
+  ASSERT_OK(Put("y", "1"));
+  ASSERT_OK(Put("z", "1"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(Put("a", "1"));
+  ASSERT_OK(Put("z", "1"));
+  ASSERT_OK(Put("bar", "1"));
+  ASSERT_OK(Put("foo", "1"));
+  ASSERT_OK(Put("foo", "2"));
+
+  ASSERT_OK(Put("foo", "3"));
+  ASSERT_OK(Put("foo", "4"));
+  ASSERT_OK(Put("foo", "5"));
+  const Snapshot* snapshot = db_->GetSnapshot();
+  ASSERT_OK(Put("foo", "6"));
+
+  std::string upper_bound = "x";
+  Slice ub_slice(upper_bound);
+  ReadOptions ro;
+  ro.snapshot = snapshot;
+  ro.iterate_upper_bound = &ub_slice;
+
+  Iterator* iter = NewIterator(ro);
+  iter->SeekForPrev("goo");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ("foo", iter->key().ToString());
+  iter->Prev();
+
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ("bar", iter->key().ToString());
+
+  delete iter;
+  db_->ReleaseSnapshot(snapshot);
+}
+
 TEST_P(DBIteratorTest, SkipStatistics) {
   Options options = CurrentOptions();
   options.statistics = rocksdb::CreateDBStatistics();

db/db_options_test.cc

@@ -242,12 +242,11 @@ TEST_F(DBOptionsTest, WritableFileMaxBufferSize) {
   ASSERT_EQ(unmatch_cnt, 0);
   ASSERT_GE(match_cnt, 11);
 
-  ASSERT_OK(
-      dbfull()->SetDBOptions({{"writable_file_max_buffer_size", "524288"}}));
   buffer_size = 512 * 1024;
   match_cnt = 0;
-  unmatch_cnt = 0;  // SetDBOptions() will create a WriteableFileWriter
-
+  unmatch_cnt = 0;
+  ASSERT_OK(
+      dbfull()->SetDBOptions({{"writable_file_max_buffer_size", "524288"}}));
   ASSERT_EQ(buffer_size,
             dbfull()->GetDBOptions().writable_file_max_buffer_size);
   i = 0;

db/db_test.cc

@@ -2153,14 +2153,10 @@ TEST_F(DBTest, GroupCommitTest) {
   do {
     Options options = CurrentOptions();
     options.env = env_;
+    env_->log_write_slowdown_.store(100);
     options.statistics = rocksdb::CreateDBStatistics();
     Reopen(options);
 
-    rocksdb::SyncPoint::GetInstance()->LoadDependency(
-        {{"WriteThread::JoinBatchGroup:BeganWaiting",
-          "DBImpl::WriteImpl:BeforeLeaderEnters"}});
-    rocksdb::SyncPoint::GetInstance()->EnableProcessing();
-
     // Start threads
     GCThread thread[kGCNumThreads];
     for (int id = 0; id < kGCNumThreads; id++) {
@@ -2169,7 +2165,13 @@ TEST_F(DBTest, GroupCommitTest) {
       thread[id].done = false;
       env_->StartThread(GCThreadBody, &thread[id]);
     }
-    env_->WaitForJoin();
+
+    for (int id = 0; id < kGCNumThreads; id++) {
+      while (thread[id].done == false) {
+        env_->SleepForMicroseconds(100000);
+      }
+    }
+    env_->log_write_slowdown_.store(0);
 
     ASSERT_GT(TestGetTickerCount(options, WRITE_DONE_BY_OTHER), 0);
 
@@ -4605,181 +4607,6 @@ TEST_F(DBTest, FileCreationRandomFailure) {
 }
 
 #ifndef ROCKSDB_LITE
-int CountIter(Iterator* iter, const Slice& key) {
-  int count = 0;
-  for (iter->Seek(key); iter->Valid() && iter->status() == Status::OK();
-       iter->Next()) {
-    count++;
-  }
-  return count;
-}
-
-// Create multiple SST files each with a different prefix_extractor config,
-// verify iterators can read all SST files using the latest config.
-TEST_F(DBTest, DynamicBloomFilterMultipleSST) {
-  Options options;
-  options.create_if_missing = true;
-  options.prefix_extractor.reset(NewFixedPrefixTransform(1));
-  options.disable_auto_compactions = true;
-  // Enable prefix bloom for SST files
-  BlockBasedTableOptions table_options;
-  table_options.filter_policy.reset(NewBloomFilterPolicy(10, true));
-  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
-  DestroyAndReopen(options);
-
-  ReadOptions read_options;
-  read_options.prefix_same_as_start = true;
-
-  // first SST with fixed:1 BF
-  ASSERT_OK(Put("foo2", "bar2"));
-  ASSERT_OK(Put("foo", "bar"));
-  ASSERT_OK(Put("foq1", "bar1"));
-  ASSERT_OK(Put("fpa", "0"));
-  dbfull()->Flush(FlushOptions());
-  Iterator* iter_old = db_->NewIterator(read_options);
-  ASSERT_EQ(CountIter(iter_old, "foo"), 4);
-
-  ASSERT_OK(dbfull()->SetOptions({{"prefix_extractor", "capped:3"}}));
-  ASSERT_EQ(0, strcmp(dbfull()->GetOptions().prefix_extractor->Name(),
-                      "rocksdb.CappedPrefix.3"));
-  Iterator* iter = db_->NewIterator(read_options);
-  ASSERT_EQ(CountIter(iter, "foo"), 2);
-
-  // second SST with capped:3 BF
-  ASSERT_OK(Put("foo3", "bar3"));
-  ASSERT_OK(Put("foo4", "bar4"));
-  ASSERT_OK(Put("foq5", "bar5"));
-  ASSERT_OK(Put("fpb", "1"));
-  dbfull()->Flush(FlushOptions());
-  // BF is cappped:3 now
-  Iterator* iter_tmp = db_->NewIterator(read_options);
-  ASSERT_EQ(CountIter(iter_tmp, "foo"), 4);
-  delete iter_tmp;
-
-  ASSERT_OK(dbfull()->SetOptions({{"prefix_extractor", "fixed:2"}}));
-  ASSERT_EQ(0, strcmp(dbfull()->GetOptions().prefix_extractor->Name(),
-                      "rocksdb.FixedPrefix.2"));
-  // third SST with fixed:2 BF
-  ASSERT_OK(Put("foo6", "bar6"));
-  ASSERT_OK(Put("foo7", "bar7"));
-  ASSERT_OK(Put("foq8", "bar8"));
-  ASSERT_OK(Put("fpc", "2"));
-  dbfull()->Flush(FlushOptions());
-  // BF is fixed:2 now
-  iter_tmp = db_->NewIterator(read_options);
-  ASSERT_EQ(CountIter(iter_tmp, "foo"), 9);
-  delete iter_tmp;
-
-  // TODO(Zhongyi): verify existing iterator cannot see newly inserted keys
-  ASSERT_EQ(CountIter(iter_old, "foo"), 4);
-  ASSERT_EQ(CountIter(iter, "foo"), 2);
-  delete iter;
-  delete iter_old;
-
-  // keys in all three SSTs are visible to iterator
-  Iterator* iter_all = db_->NewIterator(read_options);
-  ASSERT_EQ(CountIter(iter_all, "foo"), 9);
-  delete iter_all;
-  ASSERT_OK(dbfull()->SetOptions({{"prefix_extractor", "capped:3"}}));
-  ASSERT_EQ(0, strcmp(dbfull()->GetOptions().prefix_extractor->Name(),
-                      "rocksdb.CappedPrefix.3"));
-  iter_all = db_->NewIterator(read_options);
-  ASSERT_EQ(CountIter(iter_all, "foo"), 6);
-  delete iter_all;
-  // TODO(Zhongyi): add test for cases where certain SST are skipped
-  // Also verify BF related counters like BLOOM_FILTER_USEFUL
-}
-
-// Create a new column family in a running DB, change prefix_extractor
-// dynamically, verify the iterator created on the new column family behaves
-// as expected
-TEST_F(DBTest, DynamicBloomFilterNewColumnFamily) {
-  Options options = CurrentOptions();
-  options.create_if_missing = true;
-  options.prefix_extractor.reset(NewFixedPrefixTransform(1));
-  options.disable_auto_compactions = true;
-  // Enable prefix bloom for SST files
-  BlockBasedTableOptions table_options;
-  table_options.filter_policy.reset(NewBloomFilterPolicy(10, true));
-  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
-  CreateAndReopenWithCF({"pikachu"}, options);
-  ReadOptions read_options;
-  read_options.prefix_same_as_start = true;
-  // create a new CF and set prefix_extractor dynamically
-  options.prefix_extractor.reset(NewCappedPrefixTransform(3));
-  CreateColumnFamilies({"ramen_dojo"}, options);
-  ASSERT_EQ(0,
-            strcmp(dbfull()->GetOptions(handles_[2]).prefix_extractor->Name(),
-                   "rocksdb.CappedPrefix.3"));
-  ASSERT_OK(Put(2, "foo3", "bar3"));
-  ASSERT_OK(Put(2, "foo4", "bar4"));
-  ASSERT_OK(Put(2, "foo5", "bar5"));
-  ASSERT_OK(Put(2, "foq6", "bar6"));
-  ASSERT_OK(Put(2, "fpq7", "bar7"));
-  dbfull()->Flush(FlushOptions());
-  Iterator* iter = db_->NewIterator(read_options, handles_[2]);
-  ASSERT_EQ(CountIter(iter, "foo"), 3);
-  delete iter;
-  ASSERT_OK(
-      dbfull()->SetOptions(handles_[2], {{"prefix_extractor", "fixed:2"}}));
-  ASSERT_EQ(0,
-            strcmp(dbfull()->GetOptions(handles_[2]).prefix_extractor->Name(),
-                   "rocksdb.FixedPrefix.2"));
-  iter = db_->NewIterator(read_options, handles_[2]);
-  ASSERT_EQ(CountIter(iter, "foo"), 4);
-  delete iter;
-}
-
-// Verify it's possible to change prefix_extractor at runtime and iterators
-// behaves as expected
-TEST_F(DBTest, DynamicBloomFilterOptions) {
-  Options options;
-  options.create_if_missing = true;
-  options.prefix_extractor.reset(NewFixedPrefixTransform(1));
-  options.disable_auto_compactions = true;
-  // Enable prefix bloom for SST files
-  BlockBasedTableOptions table_options;
-  table_options.filter_policy.reset(NewBloomFilterPolicy(10, true));
-  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
-  DestroyAndReopen(options);
-
-  ASSERT_OK(Put("foo2", "bar2"));
-  ASSERT_OK(Put("foo", "bar"));
-  ASSERT_OK(Put("foo1", "bar1"));
-  ASSERT_OK(Put("fpa", "0"));
-  dbfull()->Flush(FlushOptions());
-  ASSERT_OK(Put("foo3", "bar3"));
-  ASSERT_OK(Put("foo4", "bar4"));
-  ASSERT_OK(Put("foo5", "bar5"));
-  ASSERT_OK(Put("fpb", "1"));
-  dbfull()->Flush(FlushOptions());
-  ASSERT_OK(Put("foo6", "bar6"));
-  ASSERT_OK(Put("foo7", "bar7"));
-  ASSERT_OK(Put("foo8", "bar8"));
-  ASSERT_OK(Put("fpc", "2"));
-  dbfull()->Flush(FlushOptions());
-
-  ReadOptions read_options;
-  read_options.prefix_same_as_start = true;
-  Iterator* iter = db_->NewIterator(read_options);
-  ASSERT_EQ(CountIter(iter, "foo"), 12);
-  delete iter;
-  Iterator* iter_old = db_->NewIterator(read_options);
-  ASSERT_EQ(CountIter(iter_old, "foo"), 12);
-
-  ASSERT_OK(dbfull()->SetOptions({{"prefix_extractor", "capped:3"}}));
-  ASSERT_EQ(0, strcmp(dbfull()->GetOptions().prefix_extractor->Name(),
-                      "rocksdb.CappedPrefix.3"));
-  iter = db_->NewIterator(read_options);
-  // "fp*" should be skipped
-  ASSERT_EQ(CountIter(iter, "foo"), 9);
-  delete iter;
-
-  // iterator created before should not be affected and see all keys
-  ASSERT_EQ(CountIter(iter_old, "foo"), 12);
-  delete iter_old;
-}
-
 TEST_F(DBTest, DynamicMiscOptions) {
   // Test max_sequential_skip_in_iterations
   Options options;
@@ -5618,17 +5445,18 @@ TEST_F(DBTest, HardLimit) {
 #if !defined(ROCKSDB_LITE) && !defined(ROCKSDB_DISABLE_STALL_NOTIFICATION)
 class WriteStallListener : public EventListener {
  public:
-  WriteStallListener()
-      : cond_(&mutex_),
-        condition_(WriteStallCondition::kNormal),
-        expected_(WriteStallCondition::kNormal),
-        expected_set_(false) {}
+  WriteStallListener() : cond_(&mutex_), 
+    condition_(WriteStallCondition::kNormal),
+    expected_(WriteStallCondition::kNormal),
+    expected_set_(false) 
+  {}
   void OnStallConditionsChanged(const WriteStallInfo& info) override {
     MutexLock l(&mutex_);
     condition_ = info.condition.cur;
-    if (expected_set_ && condition_ == expected_) {
-      cond_.Signal();
-      expected_set_ = false;
+    if (expected_set_ && 
+      condition_ == expected_) {
+        cond_.Signal();
+        expected_set_ = false;
     }
   }
   bool CheckCondition(WriteStallCondition expected) {

db/external_sst_file_ingestion_job.cc

@@ -29,13 +29,13 @@
 namespace rocksdb {
 
 Status ExternalSstFileIngestionJob::Prepare(
-    const std::vector<std::string>& external_files_paths, SuperVersion* sv) {
+    const std::vector<std::string>& external_files_paths) {
   Status status;
 
   // Read the information of files we are ingesting
   for (const std::string& file_path : external_files_paths) {
     IngestedFileInfo file_to_ingest;
-    status = GetIngestedFileInfo(file_path, &file_to_ingest, sv);
+    status = GetIngestedFileInfo(file_path, &file_to_ingest);
     if (!status.ok()) {
       return status;
     }
@@ -284,8 +284,7 @@ void ExternalSstFileIngestionJob::Cleanup(const Status& status) {
 }
 
 Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
-    const std::string& external_file, IngestedFileInfo* file_to_ingest,
-    SuperVersion* sv) {
+    const std::string& external_file, IngestedFileInfo* file_to_ingest) {
   file_to_ingest->external_file_path = external_file;
 
   // Get external file size
@@ -307,9 +306,8 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
                                                    external_file));
 
   status = cfd_->ioptions()->table_factory->NewTableReader(
-      TableReaderOptions(*cfd_->ioptions(),
-                         sv->mutable_cf_options.prefix_extractor.get(),
-                         env_options_, cfd_->internal_comparator()),
+      TableReaderOptions(*cfd_->ioptions(), env_options_,
+                         cfd_->internal_comparator()),
       std::move(sst_file_reader), file_to_ingest->file_size, &table_reader);
   if (!status.ok()) {
     return status;
@@ -365,8 +363,7 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
   // We need to disable fill_cache so that we read from the file without
   // updating the block cache.
   ro.fill_cache = false;
-  std::unique_ptr<InternalIterator> iter(table_reader->NewIterator(
-      ro, sv->mutable_cf_options.prefix_extractor.get()));
+  std::unique_ptr<InternalIterator> iter(table_reader->NewIterator(ro));
 
   // Get first (smallest) key from file
   iter->SeekToFirst();

db/external_sst_file_ingestion_job.h

@@ -86,8 +86,7 @@ class ExternalSstFileIngestionJob {
         job_start_time_(env_->NowMicros()) {}
 
   // Prepare the job by copying external files into the DB.
-  Status Prepare(const std::vector<std::string>& external_files_paths,
-                 SuperVersion* sv);
+  Status Prepare(const std::vector<std::string>& external_files_paths);
 
   // Check if we need to flush the memtable before running the ingestion job
   // This will be true if the files we are ingesting are overlapping with any
@@ -120,8 +119,7 @@ class ExternalSstFileIngestionJob {
   // Open the external file and populate `file_to_ingest` with all the
   // external information we need to ingest this file.
   Status GetIngestedFileInfo(const std::string& external_file,
-                             IngestedFileInfo* file_to_ingest,
-                             SuperVersion* sv);
+                             IngestedFileInfo* file_to_ingest);
 
   // Assign `file_to_ingest` the appropriate sequence number and  the lowest
   // possible level that it can be ingested to according to compaction_style.

db/forward_iterator.cc

@@ -33,16 +33,14 @@ class ForwardLevelIterator : public InternalIterator {
  public:
   ForwardLevelIterator(const ColumnFamilyData* const cfd,
                        const ReadOptions& read_options,
-                       const std::vector<FileMetaData*>& files,
-                       const SliceTransform* prefix_extractor)
+                       const std::vector<FileMetaData*>& files)
       : cfd_(cfd),
         read_options_(read_options),
         files_(files),
         valid_(false),
         file_index_(std::numeric_limits<uint32_t>::max()),
         file_iter_(nullptr),
-        pinned_iters_mgr_(nullptr),
-        prefix_extractor_(prefix_extractor) {}
+        pinned_iters_mgr_(nullptr) {}
 
   ~ForwardLevelIterator() {
     // Reset current pointer
@@ -77,7 +75,7 @@ class ForwardLevelIterator : public InternalIterator {
         read_options_, *(cfd_->soptions()), cfd_->internal_comparator(),
         files_[file_index_]->fd,
         read_options_.ignore_range_deletions ? nullptr : &range_del_agg,
-        prefix_extractor_, nullptr /* table_reader_ptr */, nullptr, false);
+        nullptr /* table_reader_ptr */, nullptr, false);
     file_iter_->SetPinnedItersMgr(pinned_iters_mgr_);
     valid_ = false;
     if (!range_del_agg.IsEmpty()) {
@@ -190,7 +188,6 @@ class ForwardLevelIterator : public InternalIterator {
   Status status_;
   InternalIterator* file_iter_;
   PinnedIteratorsManager* pinned_iters_mgr_;
-  const SliceTransform* prefix_extractor_;
 };
 
 ForwardIterator::ForwardIterator(DBImpl* db, const ReadOptions& read_options,
@@ -199,7 +196,7 @@ ForwardIterator::ForwardIterator(DBImpl* db, const ReadOptions& read_options,
     : db_(db),
       read_options_(read_options),
       cfd_(cfd),
-      prefix_extractor_(current_sv->mutable_cf_options.prefix_extractor.get()),
+      prefix_extractor_(cfd->ioptions()->prefix_extractor),
       user_comparator_(cfd->user_comparator()),
       immutable_min_heap_(MinIterComparator(&cfd_->internal_comparator())),
       sv_(current_sv),
@@ -636,8 +633,7 @@ void ForwardIterator::RebuildIterators(bool refresh_sv) {
     }
     l0_iters_.push_back(cfd_->table_cache()->NewIterator(
         read_options_, *cfd_->soptions(), cfd_->internal_comparator(), l0->fd,
-        read_options_.ignore_range_deletions ? nullptr : &range_del_agg,
-        sv_->mutable_cf_options.prefix_extractor.get()));
+        read_options_.ignore_range_deletions ? nullptr : &range_del_agg));
   }
   BuildLevelIterators(vstorage);
   current_ = nullptr;
@@ -707,8 +703,7 @@ void ForwardIterator::RenewIterators() {
     l0_iters_new.push_back(cfd_->table_cache()->NewIterator(
         read_options_, *cfd_->soptions(), cfd_->internal_comparator(),
         l0_files_new[inew]->fd,
-        read_options_.ignore_range_deletions ? nullptr : &range_del_agg,
-        svnew->mutable_cf_options.prefix_extractor.get()));
+        read_options_.ignore_range_deletions ? nullptr : &range_del_agg));
   }
 
   for (auto* f : l0_iters_) {
@@ -749,9 +744,8 @@ void ForwardIterator::BuildLevelIterators(const VersionStorageInfo* vstorage) {
         has_iter_trimmed_for_upper_bound_ = true;
       }
     } else {
-      level_iters_.push_back(new ForwardLevelIterator(
-          cfd_, read_options_, level_files,
-          sv_->mutable_cf_options.prefix_extractor.get()));
+      level_iters_.push_back(
+          new ForwardLevelIterator(cfd_, read_options_, level_files));
     }
   }
 }
@@ -766,8 +760,7 @@ void ForwardIterator::ResetIncompleteIterators() {
     DeleteIterator(l0_iters_[i]);
     l0_iters_[i] = cfd_->table_cache()->NewIterator(
         read_options_, *cfd_->soptions(), cfd_->internal_comparator(),
-        l0_files[i]->fd, nullptr /* range_del_agg */,
-        sv_->mutable_cf_options.prefix_extractor.get());
+        l0_files[i]->fd, nullptr /* range_del_agg */);
     l0_iters_[i]->SetPinnedItersMgr(pinned_iters_mgr_);
   }
 

db/memtable.cc

@@ -74,8 +74,8 @@ MemTable::MemTable(const InternalKeyComparator& cmp,
               : nullptr,
           mutable_cf_options.memtable_huge_page_size),
       table_(ioptions.memtable_factory->CreateMemTableRep(
-          comparator_, &arena_, mutable_cf_options.prefix_extractor.get(),
-          ioptions.info_log, column_family_id)),
+          comparator_, &arena_, ioptions.prefix_extractor, ioptions.info_log,
+          column_family_id)),
       range_del_table_(SkipListFactory().CreateMemTableRep(
           comparator_, &arena_, nullptr /* transform */, ioptions.info_log,
           column_family_id)),
@@ -95,7 +95,7 @@ MemTable::MemTable(const InternalKeyComparator& cmp,
       locks_(moptions_.inplace_update_support
                  ? moptions_.inplace_update_num_locks
                  : 0),
-      prefix_extractor_(mutable_cf_options.prefix_extractor.get()),
+      prefix_extractor_(ioptions.prefix_extractor),
       flush_state_(FLUSH_NOT_REQUESTED),
       env_(ioptions.env),
       insert_with_hint_prefix_extractor_(
@@ -696,7 +696,7 @@ static bool SaveValue(void* arg, const char* entry) {
         *(s->merge_in_progress) = true;
         merge_context->PushOperand(
             v, s->inplace_update_support == false /* operand_pinned */);
-        if (merge_operator->ShouldMerge(merge_context->GetOperands())) {
+        if (merge_operator->ShouldMerge(merge_context->GetOperandsDirectionBackward())) {
           *(s->status) = MergeHelper::TimedFullMerge(
               merge_operator, s->key->user_key(), nullptr,
               merge_context->GetOperands(), s->value, s->logger, s->statistics,

db/merge_context.h

@@ -74,8 +74,13 @@ class MergeContext {
     return (*operand_list_)[index];
   }
 
-  // Return all the operands.
+  // Same as GetOperandsDirectionForward
   const std::vector<Slice>& GetOperands() {
+    return GetOperandsDirectionForward();
+  }
+
+  // Return all the operands in the order as they were merged (passed to FullMerge or FullMergeV2)
+  const std::vector<Slice>& GetOperandsDirectionForward() {
     if (!operand_list_) {
       return empty_operand_list;
     }
@@ -84,6 +89,16 @@ class MergeContext {
     return *operand_list_;
   }
 
+  // Return all the operands in the reversed order relative to how they were merged (passed to FullMerge or FullMergeV2)
+  const std::vector<Slice>& GetOperandsDirectionBackward() {
+    if (!operand_list_) {
+      return empty_operand_list;
+    }
+
+    SetDirectionBackward();
+    return *operand_list_;
+  }
+
  private:
   void Initialize() {
     if (!operand_list_) {

db/plain_table_db_test.cc

@@ -262,13 +262,11 @@ class TestPlainTableReader : public PlainTableReader {
                        const TableProperties* table_properties,
                        unique_ptr<RandomAccessFileReader>&& file,
                        const ImmutableCFOptions& ioptions,
-                       const SliceTransform* prefix_extractor,
                        bool* expect_bloom_not_match, bool store_index_in_file,
                        uint32_t column_family_id,
                        const std::string& column_family_name)
       : PlainTableReader(ioptions, std::move(file), env_options, icomparator,
-                         encoding_type, file_size, table_properties,
-                         prefix_extractor),
+                         encoding_type, file_size, table_properties),
         expect_bloom_not_match_(expect_bloom_not_match) {
     Status s = MmapDataIfNeeded();
     EXPECT_TRUE(s.ok());
@@ -362,8 +360,7 @@ class TestPlainTableFactory : public PlainTableFactory {
         table_reader_options.env_options,
         table_reader_options.internal_comparator, encoding_type, file_size,
         bloom_bits_per_key_, hash_table_ratio_, index_sparseness_, props,
-        std::move(file), table_reader_options.ioptions,
-        table_reader_options.prefix_extractor, expect_bloom_not_match_,
+        std::move(file), table_reader_options.ioptions, expect_bloom_not_match_,
         store_index_in_file_, column_family_id_, column_family_name_));
 
     *table = std::move(new_reader);

db/range_del_aggregator.cc

@@ -195,10 +195,6 @@ Status RangeDelAggregator::AddTombstones(
   input->SeekToFirst();
   bool first_iter = true;
   while (input->Valid()) {
-    // The tombstone map holds slices into the iterator's memory. This assert
-    // ensures pinning the iterator also pins the keys/values.
-    assert(input->IsKeyPinned() && input->IsValuePinned());
-
     if (first_iter) {
       if (rep_ == nullptr) {
         InitRep({upper_bound_});

db/repair.cc

@@ -501,8 +501,7 @@ class Repairer {
     if (status.ok()) {
       InternalIterator* iter = table_cache_->NewIterator(
           ReadOptions(), env_options_, cfd->internal_comparator(), t->meta.fd,
-          nullptr /* range_del_agg */,
-          cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
+          nullptr /* range_del_agg */);
       bool empty = true;
       ParsedInternalKey parsed;
       t->min_sequence = 0;

db/table_cache.cc

@@ -89,8 +89,8 @@ Status TableCache::GetTableReader(
     const InternalKeyComparator& internal_comparator, const FileDescriptor& fd,
     bool sequential_mode, size_t readahead, bool record_read_stats,
     HistogramImpl* file_read_hist, unique_ptr<TableReader>* table_reader,
-    const SliceTransform* prefix_extractor, bool skip_filters, int level,
-    bool prefetch_index_and_filter_in_cache, bool for_compaction) {
+    bool skip_filters, int level, bool prefetch_index_and_filter_in_cache,
+    bool for_compaction) {
   std::string fname =
       TableFileName(ioptions_.cf_paths, fd.GetNumber(), fd.GetPathId());
   unique_ptr<RandomAccessFile> file;
@@ -115,8 +115,8 @@ Status TableCache::GetTableReader(
             record_read_stats ? ioptions_.statistics : nullptr, SST_READ_MICROS,
             file_read_hist, ioptions_.rate_limiter, for_compaction));
     s = ioptions_.table_factory->NewTableReader(
-        TableReaderOptions(ioptions_, prefix_extractor, env_options,
-                           internal_comparator, skip_filters, level),
+        TableReaderOptions(ioptions_, env_options, internal_comparator,
+                           skip_filters, level),
         std::move(file_reader), fd.GetFileSize(), table_reader,
         prefetch_index_and_filter_in_cache);
     TEST_SYNC_POINT("TableCache::GetTableReader:0");
@@ -134,7 +134,6 @@ void TableCache::EraseHandle(const FileDescriptor& fd, Cache::Handle* handle) {
 Status TableCache::FindTable(const EnvOptions& env_options,
                              const InternalKeyComparator& internal_comparator,
                              const FileDescriptor& fd, Cache::Handle** handle,
-                             const SliceTransform* prefix_extractor,
                              const bool no_io, bool record_read_stats,
                              HistogramImpl* file_read_hist, bool skip_filters,
                              int level,
@@ -155,8 +154,7 @@ Status TableCache::FindTable(const EnvOptions& env_options,
     s = GetTableReader(env_options, internal_comparator, fd,
                        false /* sequential mode */, 0 /* readahead */,
                        record_read_stats, file_read_hist, &table_reader,
-                       prefix_extractor, skip_filters, level,
-                       prefetch_index_and_filter_in_cache);
+                       skip_filters, level, prefetch_index_and_filter_in_cache);
     if (!s.ok()) {
       assert(table_reader == nullptr);
       RecordTick(ioptions_.statistics, NO_FILE_ERRORS);
@@ -177,9 +175,9 @@ Status TableCache::FindTable(const EnvOptions& env_options,
 InternalIterator* TableCache::NewIterator(
     const ReadOptions& options, const EnvOptions& env_options,
     const InternalKeyComparator& icomparator, const FileDescriptor& fd,
-    RangeDelAggregator* range_del_agg, const SliceTransform* prefix_extractor,
-    TableReader** table_reader_ptr, HistogramImpl* file_read_hist,
-    bool for_compaction, Arena* arena, bool skip_filters, int level) {
+    RangeDelAggregator* range_del_agg, TableReader** table_reader_ptr,
+    HistogramImpl* file_read_hist, bool for_compaction, Arena* arena,
+    bool skip_filters, int level) {
   PERF_TIMER_GUARD(new_table_iterator_nanos);
 
   Status s;
@@ -212,7 +210,7 @@ InternalIterator* TableCache::NewIterator(
     s = GetTableReader(
         env_options, icomparator, fd, true /* sequential_mode */, readahead,
         !for_compaction /* record stats */, nullptr, &table_reader_unique_ptr,
-        prefix_extractor, false /* skip_filters */, level,
+        false /* skip_filters */, level,
         true /* prefetch_index_and_filter_in_cache */, for_compaction);
     if (s.ok()) {
       table_reader = table_reader_unique_ptr.release();
@@ -220,7 +218,7 @@ InternalIterator* TableCache::NewIterator(
   } else {
     table_reader = fd.table_reader;
     if (table_reader == nullptr) {
-      s = FindTable(env_options, icomparator, fd, &handle, prefix_extractor,
+      s = FindTable(env_options, icomparator, fd, &handle,
                     options.read_tier == kBlockCacheTier /* no_io */,
                     !for_compaction /* record read_stats */, file_read_hist,
                     skip_filters, level);
@@ -235,8 +233,7 @@ InternalIterator* TableCache::NewIterator(
         !options.table_filter(*table_reader->GetTableProperties())) {
       result = NewEmptyInternalIterator(arena);
     } else {
-      result = table_reader->NewIterator(options, prefix_extractor, arena,
-                                         skip_filters);
+      result = table_reader->NewIterator(options, arena, skip_filters);
     }
     if (create_new_table_reader) {
       assert(handle == nullptr);
@@ -279,13 +276,12 @@ InternalIterator* TableCache::NewIterator(
 InternalIterator* TableCache::NewRangeTombstoneIterator(
     const ReadOptions& options, const EnvOptions& env_options,
     const InternalKeyComparator& icomparator, const FileDescriptor& fd,
-    HistogramImpl* file_read_hist, bool skip_filters, int level,
-    const SliceTransform* prefix_extractor) {
+    HistogramImpl* file_read_hist, bool skip_filters, int level) {
   Status s;
   Cache::Handle* handle = nullptr;
   TableReader* table_reader = fd.table_reader;
   if (table_reader == nullptr) {
-    s = FindTable(env_options, icomparator, fd, &handle, prefix_extractor,
+    s = FindTable(env_options, icomparator, fd, &handle,
                   options.read_tier == kBlockCacheTier /* no_io */,
                   true /* record read_stats */, file_read_hist, skip_filters,
                   level);
@@ -317,10 +313,8 @@ InternalIterator* TableCache::NewRangeTombstoneIterator(
 Status TableCache::Get(const ReadOptions& options,
                        const InternalKeyComparator& internal_comparator,
                        const FileDescriptor& fd, const Slice& k,
-                       GetContext* get_context,
-                       const SliceTransform* prefix_extractor,
-                       HistogramImpl* file_read_hist, bool skip_filters,
-                       int level) {
+                       GetContext* get_context, HistogramImpl* file_read_hist,
+                       bool skip_filters, int level) {
   std::string* row_cache_entry = nullptr;
   bool done = false;
 #ifndef ROCKSDB_LITE
@@ -384,10 +378,10 @@ Status TableCache::Get(const ReadOptions& options,
   Cache::Handle* handle = nullptr;
   if (!done && s.ok()) {
     if (t == nullptr) {
-      s = FindTable(
-          env_options_, internal_comparator, fd, &handle, prefix_extractor,
-          options.read_tier == kBlockCacheTier /* no_io */,
-          true /* record_read_stats */, file_read_hist, skip_filters, level);
+      s = FindTable(env_options_, internal_comparator, fd, &handle,
+                    options.read_tier == kBlockCacheTier /* no_io */,
+                    true /* record_read_stats */, file_read_hist, skip_filters,
+                    level);
       if (s.ok()) {
         t = GetTableReaderFromHandle(handle);
       }
@@ -406,7 +400,7 @@ Status TableCache::Get(const ReadOptions& options,
     }
     if (s.ok()) {
       get_context->SetReplayLog(row_cache_entry);  // nullptr if no cache.
-      s = t->Get(options, k, get_context, prefix_extractor, skip_filters);
+      s = t->Get(options, k, get_context, skip_filters);
       get_context->SetReplayLog(nullptr);
     } else if (options.read_tier == kBlockCacheTier && s.IsIncomplete()) {
       // Couldn't find Table in cache but treat as kFound if no_io set
@@ -436,8 +430,7 @@ Status TableCache::Get(const ReadOptions& options,
 Status TableCache::GetTableProperties(
     const EnvOptions& env_options,
     const InternalKeyComparator& internal_comparator, const FileDescriptor& fd,
-    std::shared_ptr<const TableProperties>* properties,
-    const SliceTransform* prefix_extractor, bool no_io) {
+    std::shared_ptr<const TableProperties>* properties, bool no_io) {
   Status s;
   auto table_reader = fd.table_reader;
   // table already been pre-loaded?
@@ -448,8 +441,7 @@ Status TableCache::GetTableProperties(
   }
 
   Cache::Handle* table_handle = nullptr;
-  s = FindTable(env_options, internal_comparator, fd, &table_handle,
-                prefix_extractor, no_io);
+  s = FindTable(env_options, internal_comparator, fd, &table_handle, no_io);
   if (!s.ok()) {
     return s;
   }
@@ -462,8 +454,8 @@ Status TableCache::GetTableProperties(
 
 size_t TableCache::GetMemoryUsageByTableReader(
     const EnvOptions& env_options,
-    const InternalKeyComparator& internal_comparator, const FileDescriptor& fd,
-    const SliceTransform* prefix_extractor) {
+    const InternalKeyComparator& internal_comparator,
+    const FileDescriptor& fd) {
   Status s;
   auto table_reader = fd.table_reader;
   // table already been pre-loaded?
@@ -472,8 +464,7 @@ size_t TableCache::GetMemoryUsageByTableReader(
   }
 
   Cache::Handle* table_handle = nullptr;
-  s = FindTable(env_options, internal_comparator, fd, &table_handle,
-                prefix_extractor, true);
+  s = FindTable(env_options, internal_comparator, fd, &table_handle, true);
   if (!s.ok()) {
     return 0;
   }

db/table_cache.h

@@ -54,7 +54,6 @@ class TableCache {
       const ReadOptions& options, const EnvOptions& toptions,
       const InternalKeyComparator& internal_comparator,
       const FileDescriptor& file_fd, RangeDelAggregator* range_del_agg,
-      const SliceTransform* prefix_extractor = nullptr,
       TableReader** table_reader_ptr = nullptr,
       HistogramImpl* file_read_hist = nullptr, bool for_compaction = false,
       Arena* arena = nullptr, bool skip_filters = false, int level = -1);
@@ -63,8 +62,7 @@ class TableCache {
       const ReadOptions& options, const EnvOptions& toptions,
       const InternalKeyComparator& internal_comparator,
       const FileDescriptor& file_fd, HistogramImpl* file_read_hist,
-      bool skip_filters, int level,
-      const SliceTransform* prefix_extractor = nullptr);
+      bool skip_filters, int level);
 
   // If a seek to internal key "k" in specified file finds an entry,
   // call (*handle_result)(arg, found_key, found_value) repeatedly until
@@ -77,10 +75,8 @@ class TableCache {
   Status Get(const ReadOptions& options,
              const InternalKeyComparator& internal_comparator,
              const FileDescriptor& file_fd, const Slice& k,
-             GetContext* get_context,
-             const SliceTransform* prefix_extractor = nullptr,
-             HistogramImpl* file_read_hist = nullptr, bool skip_filters = false,
-             int level = -1);
+             GetContext* get_context, HistogramImpl* file_read_hist = nullptr,
+             bool skip_filters = false, int level = -1);
 
   // Evict any entry for the specified file number
   static void Evict(Cache* cache, uint64_t file_number);
@@ -95,7 +91,6 @@ class TableCache {
   Status FindTable(const EnvOptions& toptions,
                    const InternalKeyComparator& internal_comparator,
                    const FileDescriptor& file_fd, Cache::Handle**,
-                   const SliceTransform* prefix_extractor = nullptr,
                    const bool no_io = false, bool record_read_stats = true,
                    HistogramImpl* file_read_hist = nullptr,
                    bool skip_filters = false, int level = -1,
@@ -114,7 +109,6 @@ class TableCache {
                             const InternalKeyComparator& internal_comparator,
                             const FileDescriptor& file_meta,
                             std::shared_ptr<const TableProperties>* properties,
-                            const SliceTransform* prefix_extractor = nullptr,
                             bool no_io = false);
 
   // Return total memory usage of the table reader of the file.
@@ -122,8 +116,7 @@ class TableCache {
   size_t GetMemoryUsageByTableReader(
       const EnvOptions& toptions,
       const InternalKeyComparator& internal_comparator,
-      const FileDescriptor& fd,
-      const SliceTransform* prefix_extractor = nullptr);
+      const FileDescriptor& fd);
 
   // Release the handle from a cache
   void ReleaseHandle(Cache::Handle* handle);
@@ -140,7 +133,6 @@ class TableCache {
                         size_t readahead, bool record_read_stats,
                         HistogramImpl* file_read_hist,
                         unique_ptr<TableReader>* table_reader,
-                        const SliceTransform* prefix_extractor = nullptr,
                         bool skip_filters = false, int level = -1,
                         bool prefetch_index_and_filter_in_cache = true,
                         bool for_compaction = false);

db/table_properties_collector_test.cc

@@ -39,7 +39,6 @@ static const uint32_t kTestColumnFamilyId = 66;
 static const std::string kTestColumnFamilyName = "test_column_fam";
 
 void MakeBuilder(const Options& options, const ImmutableCFOptions& ioptions,
-                 const MutableCFOptions& moptions,
                  const InternalKeyComparator& internal_comparator,
                  const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
                      int_tbl_prop_collector_factories,
@@ -49,9 +48,10 @@ void MakeBuilder(const Options& options, const ImmutableCFOptions& ioptions,
   writable->reset(new WritableFileWriter(std::move(wf), EnvOptions()));
   int unknown_level = -1;
   builder->reset(NewTableBuilder(
-      ioptions, moptions, internal_comparator, int_tbl_prop_collector_factories,
-      kTestColumnFamilyId, kTestColumnFamilyName, writable->get(),
-      options.compression, options.compression_opts, unknown_level));
+      ioptions, internal_comparator, int_tbl_prop_collector_factories,
+      kTestColumnFamilyId, kTestColumnFamilyName,
+      writable->get(), options.compression, options.compression_opts,
+      unknown_level));
 }
 }  // namespace
 
@@ -251,7 +251,6 @@ void TestCustomizedTablePropertiesCollector(
   std::unique_ptr<TableBuilder> builder;
   std::unique_ptr<WritableFileWriter> writer;
   const ImmutableCFOptions ioptions(options);
-  const MutableCFOptions moptions(options);
   std::vector<std::unique_ptr<IntTblPropCollectorFactory>>
       int_tbl_prop_collector_factories;
   if (test_int_tbl_prop_collector) {
@@ -260,7 +259,7 @@ void TestCustomizedTablePropertiesCollector(
   } else {
     GetIntTblPropCollectorFactory(ioptions, &int_tbl_prop_collector_factories);
   }
-  MakeBuilder(options, ioptions, moptions, internal_comparator,
+  MakeBuilder(options, ioptions, internal_comparator,
               &int_tbl_prop_collector_factories, &writer, &builder);
 
   SequenceNumber seqNum = 0U;
@@ -402,11 +401,10 @@ void TestInternalKeyPropertiesCollector(
         new InternalKeyPropertiesCollectorFactory);
   }
   const ImmutableCFOptions ioptions(options);
-  MutableCFOptions moptions(options);
 
   for (int iter = 0; iter < 2; ++iter) {
-    MakeBuilder(options, ioptions, moptions, pikc,
-                &int_tbl_prop_collector_factories, &writable, &builder);
+    MakeBuilder(options, ioptions, pikc, &int_tbl_prop_collector_factories,
+                &writable, &builder);
     for (const auto& k : keys) {
       builder->Add(k.Encode(), "val");
     }

db/version_builder.cc

@@ -368,8 +368,7 @@ class VersionBuilder::Rep {
   }
 
   void LoadTableHandlers(InternalStats* internal_stats, int max_threads,
-                         bool prefetch_index_and_filter_in_cache,
-                         const SliceTransform* prefix_extractor) {
+                         bool prefetch_index_and_filter_in_cache) {
     assert(table_cache_ != nullptr);
     // <file metadata, level>
     std::vector<std::pair<FileMetaData*, int>> files_meta;
@@ -391,12 +390,12 @@ class VersionBuilder::Rep {
 
         auto* file_meta = files_meta[file_idx].first;
         int level = files_meta[file_idx].second;
-        table_cache_->FindTable(
-            env_options_, *(base_vstorage_->InternalComparator()),
-            file_meta->fd, &file_meta->table_reader_handle, prefix_extractor,
-            false /*no_io */, true /* record_read_stats */,
-            internal_stats->GetFileReadHist(level), false, level,
-            prefetch_index_and_filter_in_cache);
+        table_cache_->FindTable(env_options_,
+                                *(base_vstorage_->InternalComparator()),
+                                file_meta->fd, &file_meta->table_reader_handle,
+                                false /*no_io */, true /* record_read_stats */,
+                                internal_stats->GetFileReadHist(level), false,
+                                level, prefetch_index_and_filter_in_cache);
         if (file_meta->table_reader_handle != nullptr) {
           // Load table_reader
           file_meta->fd.table_reader = table_cache_->GetTableReaderFromHandle(
@@ -456,12 +455,11 @@ void VersionBuilder::SaveTo(VersionStorageInfo* vstorage) {
   rep_->SaveTo(vstorage);
 }
 
-void VersionBuilder::LoadTableHandlers(InternalStats* internal_stats,
-                                       int max_threads,
-                                       bool prefetch_index_and_filter_in_cache,
-                                       const SliceTransform* prefix_extractor) {
+void VersionBuilder::LoadTableHandlers(
+    InternalStats* internal_stats, int max_threads,
+    bool prefetch_index_and_filter_in_cache) {
   rep_->LoadTableHandlers(internal_stats, max_threads,
-                          prefetch_index_and_filter_in_cache, prefix_extractor);
+                          prefetch_index_and_filter_in_cache);
 }
 
 void VersionBuilder::MaybeAddFile(VersionStorageInfo* vstorage, int level,

db/version_builder.h

@@ -9,7 +9,6 @@
 //
 #pragma once
 #include "rocksdb/env.h"
-#include "rocksdb/slice_transform.h"
 
 namespace rocksdb {
 
@@ -34,8 +33,7 @@ class VersionBuilder {
   void Apply(VersionEdit* edit);
   void SaveTo(VersionStorageInfo* vstorage);
   void LoadTableHandlers(InternalStats* internal_stats, int max_threads,
-                         bool prefetch_index_and_filter_in_cache,
-                         const SliceTransform* prefix_extractor);
+                         bool prefetch_index_and_filter_in_cache);
   void MaybeAddFile(VersionStorageInfo* vstorage, int level, FileMetaData* f);
 
  private:

db/version_edit.cc

@@ -20,7 +20,7 @@ namespace rocksdb {
 
 // Tag numbers for serialized VersionEdit.  These numbers are written to
 // disk and should not be changed.
-enum Tag : uint32_t {
+enum Tag {
   kComparator = 1,
   kLogNumber = 2,
   kNextFileNumber = 3,
@@ -42,7 +42,7 @@ enum Tag : uint32_t {
   kMaxColumnFamily = 203,
 };
 
-enum CustomTag : uint32_t {
+enum CustomTag {
   kTerminate = 1,  // The end of customized fields
   kNeedCompaction = 2,
   // Since Manifest is not entirely currently forward-compatible, and the only

db/version_set.cc

@@ -463,8 +463,7 @@ class LevelIterator final : public InternalIterator {
   LevelIterator(TableCache* table_cache, const ReadOptions& read_options,
                 const EnvOptions& env_options,
                 const InternalKeyComparator& icomparator,
-                const LevelFilesBrief* flevel,
-                const SliceTransform* prefix_extractor, bool should_sample,
+                const LevelFilesBrief* flevel, bool should_sample,
                 HistogramImpl* file_read_hist, bool for_compaction,
                 bool skip_filters, int level, RangeDelAggregator* range_del_agg)
       : table_cache_(table_cache),
@@ -472,7 +471,6 @@ class LevelIterator final : public InternalIterator {
         env_options_(env_options),
         icomparator_(icomparator),
         flevel_(flevel),
-        prefix_extractor_(prefix_extractor),
         file_read_hist_(file_read_hist),
         should_sample_(should_sample),
         for_compaction_(for_compaction),
@@ -549,9 +547,8 @@ class LevelIterator final : public InternalIterator {
 
     return table_cache_->NewIterator(
         read_options_, env_options_, icomparator_, file_meta.fd, range_del_agg_,
-        prefix_extractor_, nullptr /* don't need reference to table */,
-        file_read_hist_, for_compaction_, nullptr /* arena */, skip_filters_,
-        level_);
+        nullptr /* don't need reference to table */, file_read_hist_,
+        for_compaction_, nullptr /* arena */, skip_filters_, level_);
   }
 
   TableCache* table_cache_;
@@ -560,7 +557,6 @@ class LevelIterator final : public InternalIterator {
   const InternalKeyComparator& icomparator_;
   const LevelFilesBrief* flevel_;
   mutable FileDescriptor current_value_;
-  const SliceTransform* prefix_extractor_;
 
   HistogramImpl* file_read_hist_;
   bool should_sample_;
@@ -726,8 +722,8 @@ Status Version::GetTableProperties(std::shared_ptr<const TableProperties>* tp,
   auto table_cache = cfd_->table_cache();
   auto ioptions = cfd_->ioptions();
   Status s = table_cache->GetTableProperties(
-      env_options_, cfd_->internal_comparator(), file_meta->fd, tp,
-      mutable_cf_options_.prefix_extractor.get(), true /* no io */);
+      env_options_, cfd_->internal_comparator(), file_meta->fd,
+      tp, true /* no io */);
   if (s.ok()) {
     return s;
   }
@@ -861,8 +857,8 @@ size_t Version::GetMemoryUsageByTableReaders() {
   for (auto& file_level : storage_info_.level_files_brief_) {
     for (size_t i = 0; i < file_level.num_files; i++) {
       total_usage += cfd_->table_cache()->GetMemoryUsageByTableReader(
-          env_options_, cfd_->internal_comparator(), file_level.files[i].fd,
-          mutable_cf_options_.prefix_extractor.get());
+          env_options_, cfd_->internal_comparator(),
+          file_level.files[i].fd);
     }
   }
   return total_usage;
@@ -1000,9 +996,8 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options,
       const auto& file = storage_info_.LevelFilesBrief(0).files[i];
       merge_iter_builder->AddIterator(cfd_->table_cache()->NewIterator(
           read_options, soptions, cfd_->internal_comparator(), file.fd,
-          range_del_agg, mutable_cf_options_.prefix_extractor.get(), nullptr,
-          cfd_->internal_stats()->GetFileReadHist(0), false, arena,
-          false /* skip_filters */, 0 /* level */));
+          range_del_agg, nullptr, cfd_->internal_stats()->GetFileReadHist(0),
+          false, arena, false /* skip_filters */, 0 /* level */));
     }
     if (should_sample) {
       // Count ones for every L0 files. This is done per iterator creation
@@ -1021,7 +1016,7 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options,
     merge_iter_builder->AddIterator(new (mem) LevelIterator(
         cfd_->table_cache(), read_options, soptions,
         cfd_->internal_comparator(), &storage_info_.LevelFilesBrief(level),
-        mutable_cf_options_.prefix_extractor.get(), should_sample_file_read(),
+        should_sample_file_read(),
         cfd_->internal_stats()->GetFileReadHist(level),
         false /* for_compaction */, IsFilterSkipped(level), level,
         range_del_agg));
@@ -1053,9 +1048,8 @@ Status Version::OverlapWithLevelIterator(const ReadOptions& read_options,
       }
       ScopedArenaIterator iter(cfd_->table_cache()->NewIterator(
           read_options, env_options, cfd_->internal_comparator(), file->fd,
-          &range_del_agg, mutable_cf_options_.prefix_extractor.get(), nullptr,
-          cfd_->internal_stats()->GetFileReadHist(0), false, &arena,
-          false /* skip_filters */, 0 /* level */));
+          &range_del_agg, nullptr, cfd_->internal_stats()->GetFileReadHist(0),
+          false, &arena, false /* skip_filters */, 0 /* level */));
       status = OverlapWithIterator(
           ucmp, smallest_user_key, largest_user_key, iter.get(), overlap);
       if (!status.ok() || *overlap) {
@@ -1067,7 +1061,7 @@ Status Version::OverlapWithLevelIterator(const ReadOptions& read_options,
     ScopedArenaIterator iter(new (mem) LevelIterator(
         cfd_->table_cache(), read_options, env_options,
         cfd_->internal_comparator(), &storage_info_.LevelFilesBrief(level),
-        mutable_cf_options_.prefix_extractor.get(), should_sample_file_read(),
+        should_sample_file_read(),
         cfd_->internal_stats()->GetFileReadHist(level),
         false /* for_compaction */, IsFilterSkipped(level), level,
         &range_del_agg));
@@ -1128,9 +1122,7 @@ VersionStorageInfo::VersionStorageInfo(
 }
 
 Version::Version(ColumnFamilyData* column_family_data, VersionSet* vset,
-                 const EnvOptions& env_opt,
-                 const MutableCFOptions mutable_cf_options,
-                 uint64_t version_number)
+                 const EnvOptions& env_opt, uint64_t version_number)
     : env_(vset->env_),
       cfd_(column_family_data),
       info_log_((cfd_ == nullptr) ? nullptr : cfd_->ioptions()->info_log),
@@ -1154,7 +1146,6 @@ Version::Version(ColumnFamilyData* column_family_data, VersionSet* vset,
       prev_(this),
       refs_(0),
       env_options_(env_opt),
-      mutable_cf_options_(mutable_cf_options),
       version_number_(version_number) {}
 
 void Version::Get(const ReadOptions& read_options, const LookupKey& k,
@@ -1198,7 +1189,6 @@ void Version::Get(const ReadOptions& read_options, const LookupKey& k,
 
     *status = table_cache_->Get(
         read_options, *internal_comparator(), f->fd, ikey, &get_context,
-        mutable_cf_options_.prefix_extractor.get(),
         cfd_->internal_stats()->GetFileReadHist(fp.GetHitFileLevel()),
         IsFilterSkipped(static_cast<int>(fp.GetHitFileLevel()),
                         fp.IsHitFileLastInLevel()),
@@ -2785,7 +2775,7 @@ Status VersionSet::LogAndApply(ColumnFamilyData* column_family_data,
     LogAndApplyCFHelper(w.edit_list.front());
     batch_edits.push_back(w.edit_list.front());
   } else {
-    v = new Version(column_family_data, this, env_options_, mutable_cf_options,
+    v = new Version(column_family_data, this, env_options_,
                     current_version_number_++);
     builder_guard.reset(new BaseReferencedVersionBuilder(column_family_data));
     auto* builder = builder_guard->version_builder();
@@ -2846,8 +2836,7 @@ Status VersionSet::LogAndApply(ColumnFamilyData* column_family_data,
       builder_guard->version_builder()->LoadTableHandlers(
           column_family_data->internal_stats(),
           column_family_data->ioptions()->optimize_filters_for_hits,
-          true /* prefetch_index_and_filter_in_cache */,
-          mutable_cf_options.prefix_extractor.get());
+          true /* prefetch_index_and_filter_in_cache */);
     }
 
     // This is fine because everything inside of this block is serialized --
@@ -3338,13 +3327,11 @@ Status VersionSet::Recover(
         // Need to do it out of the mutex.
         builder->LoadTableHandlers(
             cfd->internal_stats(), db_options_->max_file_opening_threads,
-            false /* prefetch_index_and_filter_in_cache */,
-            cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
+            false /* prefetch_index_and_filter_in_cache */);
       }
 
-      Version* v = new Version(cfd, this, env_options_,
-                               *cfd->GetLatestMutableCFOptions(),
-                               current_version_number_++);
+      Version* v =
+          new Version(cfd, this, env_options_, current_version_number_++);
       builder->SaveTo(v->storage_info());
 
       // Install recovered version
@@ -3709,9 +3696,8 @@ Status VersionSet::DumpManifest(Options& options, std::string& dscname,
       assert(builders_iter != builders.end());
       auto builder = builders_iter->second->version_builder();
 
-      Version* v = new Version(cfd, this, env_options_,
-                               *cfd->GetLatestMutableCFOptions(),
-                               current_version_number_++);
+      Version* v =
+          new Version(cfd, this, env_options_, current_version_number_++);
       builder->SaveTo(v->storage_info());
       v->PrepareApply(*cfd->GetLatestMutableCFOptions(), false);
 
@@ -3934,8 +3920,7 @@ uint64_t VersionSet::ApproximateSize(Version* v, const FdWithKeyRange& f,
     TableReader* table_reader_ptr;
     InternalIterator* iter = v->cfd_->table_cache()->NewIterator(
         ReadOptions(), v->env_options_, v->cfd_->internal_comparator(), f.fd,
-        nullptr /* range_del_agg */,
-        v->GetMutableCFOptions().prefix_extractor.get(), &table_reader_ptr);
+        nullptr /* range_del_agg */, &table_reader_ptr);
     if (table_reader_ptr != nullptr) {
       result = table_reader_ptr->ApproximateOffsetOf(key);
     }
@@ -4015,7 +4000,6 @@ InternalIterator* VersionSet::MakeInputIterator(
           list[num++] = cfd->table_cache()->NewIterator(
               read_options, env_options_compactions, cfd->internal_comparator(),
               flevel->files[i].fd, range_del_agg,
-              c->mutable_cf_options()->prefix_extractor.get(),
               nullptr /* table_reader_ptr */,
               nullptr /* no per level latency histogram */,
               true /* for_compaction */, nullptr /* arena */,
@@ -4026,7 +4010,6 @@ InternalIterator* VersionSet::MakeInputIterator(
         list[num++] = new LevelIterator(
             cfd->table_cache(), read_options, env_options_compactions,
             cfd->internal_comparator(), c->input_levels(which),
-            c->mutable_cf_options()->prefix_extractor.get(),
             false /* should_sample */,
             nullptr /* no per level latency histogram */,
             true /* for_compaction */, false /* skip_filters */,
@@ -4166,9 +4149,7 @@ ColumnFamilyData* VersionSet::CreateColumnFamily(
     const ColumnFamilyOptions& cf_options, VersionEdit* edit) {
   assert(edit->is_column_family_add_);
 
-  MutableCFOptions dummy_cf_options;
-  Version* dummy_versions =
-      new Version(nullptr, this, env_options_, dummy_cf_options);
+  Version* dummy_versions = new Version(nullptr, this, env_options_);
   // Ref() dummy version once so that later we can call Unref() to delete it
   // by avoiding calling "delete" explicitly (~Version is private)
   dummy_versions->Ref();
@@ -4176,9 +4157,8 @@ ColumnFamilyData* VersionSet::CreateColumnFamily(
       edit->column_family_name_, edit->column_family_, dummy_versions,
       cf_options);
 
-  Version* v = new Version(new_cfd, this, env_options_,
-                           *new_cfd->GetLatestMutableCFOptions(),
-                           current_version_number_++);
+  Version* v =
+      new Version(new_cfd, this, env_options_, current_version_number_++);
 
   // Fill level target base information.
   v->storage_info()->CalculateBaseBytes(*new_cfd->ioptions(),

db/version_set.h

@@ -633,8 +633,6 @@ class Version {
 
   uint64_t GetSstFilesSize();
 
-  MutableCFOptions GetMutableCFOptions() { return mutable_cf_options_; }
-
  private:
   Env* env_;
   friend class VersionSet;
@@ -682,14 +680,13 @@ class Version {
   Version* prev_;               // Previous version in linked list
   int refs_;                    // Number of live refs to this version
   const EnvOptions env_options_;
-  const MutableCFOptions mutable_cf_options_;
 
   // A version number that uniquely represents this version. This is
   // used for debugging and logging purposes only.
   uint64_t version_number_;
 
   Version(ColumnFamilyData* cfd, VersionSet* vset, const EnvOptions& env_opt,
-          MutableCFOptions mutable_cf_options, uint64_t version_number = 0);
+          uint64_t version_number = 0);
 
   ~Version();
 

db/write_thread.cc

@@ -318,7 +318,6 @@ void WriteThread::JoinBatchGroup(Writer* w) {
      * 3.2) an existing memtable writer group leader tell us to finish memtable
      *      writes in parallel.
      */
-    TEST_SYNC_POINT_CALLBACK("WriteThread::JoinBatchGroup:BeganWaiting", w);
     AwaitState(w, STATE_GROUP_LEADER | STATE_MEMTABLE_WRITER_LEADER |
                       STATE_PARALLEL_MEMTABLE_WRITER | STATE_COMPLETED,
                &jbg_ctx);

include/rocksdb/advanced_options.h

@@ -87,14 +87,6 @@ struct CompactionOptionsFIFO {
 
 // Compression options for different compression algorithms like Zlib
 struct CompressionOptions {
-  // RocksDB's generic default compression level. Internally it'll be translated
-  // to the default compression level specific to the library being used (see
-  // comment above `ColumnFamilyOptions::compression`).
-  //
-  // The default value is the max 16-bit int as it'll be written out in OPTIONS
-  // file, which should be portable.
-  const static int kDefaultCompressionLevel = 32767;
-
   int window_bits;
   int level;
   int strategy;
@@ -128,7 +120,7 @@ struct CompressionOptions {
 
   CompressionOptions()
       : window_bits(-14),
-        level(kDefaultCompressionLevel),
+        level(-1),
         strategy(0),
         max_dict_bytes(0),
         zstd_max_train_bytes(0) {}

include/rocksdb/merge_operator.h

@@ -195,6 +195,11 @@ class MergeOperator {
   // during a point lookup, thereby helping in limiting the number of levels to
   // read from.
   // Doesn't help with iterators.
+  //
+  // Note: the merge operands are passed to this function in the reversed order
+  // relative to how they were merged (passed to FullMerge or FullMergeV2)
+  // for performance reasons, see also:
+  // https://github.com/facebook/rocksdb/issues/3865
   virtual bool ShouldMerge(const std::vector<Slice>& /*operands*/) const {
     return false;
   }

include/rocksdb/options.h

@@ -198,21 +198,11 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
   // Typical speeds of kSnappyCompression on an Intel(R) Core(TM)2 2.4GHz:
   //    ~200-500MB/s compression
   //    ~400-800MB/s decompression
-  //
   // Note that these speeds are significantly faster than most
   // persistent storage speeds, and therefore it is typically never
   // worth switching to kNoCompression.  Even if the input data is
   // incompressible, the kSnappyCompression implementation will
   // efficiently detect that and will switch to uncompressed mode.
-  //
-  // If you do not set `compression_opts.level`, or set it to
-  // `CompressionOptions::kDefaultCompressionLevel`, we will attempt to pick the
-  // default corresponding to `compression` as follows:
-  //
-  // - kZSTD: 3
-  // - kZlibCompression: Z_DEFAULT_COMPRESSION (currently -1)
-  // - kLZ4HCCompression: 0
-  // - For all others, we do not specify a compression level
   CompressionType compression;
 
   // Compression algorithm that will be used for the bottommost level that
@@ -281,7 +271,7 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
   // later in the vector.
   // Note that, if a path is supplied to multiple column
   // families, it would have files and total size from all
-  // the column families combined. User should provision for the
+  // the column families combined. User should privision for the
   // total size(from all the column families) in such cases.
   //
   // If left empty, db_paths will be used.
@@ -568,9 +558,8 @@ struct DBOptions {
 
   // manifest file is rolled over on reaching this limit.
   // The older manifest file be deleted.
-  // The default value is 1GB so that the manifest file can grow, but not
-  // reach the limit of storage capacity.
-  uint64_t max_manifest_file_size = 1024 * 1024 * 1024;
+  // The default value is MAX_INT so that roll-over does not take place.
+  uint64_t max_manifest_file_size = std::numeric_limits<uint64_t>::max();
 
   // Number of shards used for table cache.
   int table_cache_numshardbits = 6;

include/rocksdb/slice.h

@@ -43,9 +43,7 @@ class Slice {
 
   // Create a slice that refers to s[0,strlen(s)-1]
   /* implicit */
-  Slice(const char* s) : data_(s) {
-    size_ = (s == nullptr) ? 0 : strlen(s);
-  }
+  Slice(const char* s) : data_(s), size_(strlen(s)) { }
 
   // Create a single slice from SliceParts using buf as storage.
   // buf must exist as long as the returned Slice exists.

include/rocksdb/sst_file_manager.h

@@ -98,12 +98,13 @@ class SstFileManager {
 //    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.
+//    file. 0 means to always delete the whole file. NOTE this options may not
+//    work well with checkpoints, which relies on file system hard links.
 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,
-    uint64_t bytes_max_delete_chunk = 64 * 1024 * 1024);
+    uint64_t bytes_max_delete_chunk = 0);
 
 }  // namespace rocksdb

include/rocksdb/statistics.h

@@ -619,8 +619,6 @@ struct HistogramData {
   // zero-initialize new members since old Statistics::histogramData()
   // implementations won't write them.
   double max = 0.0;
-  uint64_t count = 0;
-  uint64_t sum = 0;
 };
 
 enum StatsLevel {

include/rocksdb/utilities/ldb_cmd.h

@@ -210,12 +210,20 @@ class LDBCommand {
   bool ParseStringOption(const std::map<std::string, std::string>& options,
                          const std::string& option, std::string* value);
 
+  /**
+   * Returns the value of the specified option as a boolean.
+   * default_val is used if the option is not found in options.
+   * Throws an exception if the value of the option is not
+   * "true" or "false" (case insensitive).
+   */
+  bool ParseBooleanOption(const std::map<std::string, std::string>& options,
+                          const std::string& option, bool default_val);
+
   Options options_;
   std::vector<ColumnFamilyDescriptor> column_families_;
   LDBOptions ldb_options_;
 
  private:
-  friend class WALDumperCommand;
   /**
    * Interpret command line options and flags to determine if the key
    * should be input/output in hex.
@@ -230,15 +238,6 @@ class LDBCommand {
   bool IsValueHex(const std::map<std::string, std::string>& options,
                   const std::vector<std::string>& flags);
 
-  /**
-   * Returns the value of the specified option as a boolean.
-   * default_val is used if the option is not found in options.
-   * Throws an exception if the value of the option is not
-   * "true" or "false" (case insensitive).
-   */
-  bool ParseBooleanOption(const std::map<std::string, std::string>& options,
-                          const std::string& option, bool default_val);
-
   /**
    * Converts val to a boolean.
    * val must be either true or false (case insensitive).

include/rocksdb/version.h

@@ -6,7 +6,7 @@
 
 #define ROCKSDB_MAJOR 5
 #define ROCKSDB_MINOR 14
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 3
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

java/rocksjni/options.cc

@@ -3230,7 +3230,7 @@ Java_org_rocksdb_ColumnFamilyOptions_setCompactionFilterFactoryHandle(
     JNIEnv* /*env*/, jobject /*jobj*/, jlong jopt_handle,
     jlong jcompactionfilterfactory_handle) {
   auto* cff_factory =
-      reinterpret_cast<std::shared_ptr<rocksdb::CompactionFilterFactory>*>(
+      reinterpret_cast<std::shared_ptr<rocksdb::CompactionFilterFactoryJniCallback>*>(
           jcompactionfilterfactory_handle);
   reinterpret_cast<rocksdb::ColumnFamilyOptions*>(jopt_handle)
       ->compaction_filter_factory = *cff_factory;

java/rocksjni/portal.h

@@ -4292,25 +4292,12 @@ class JniUtil {
      * @param bytes The bytes to copy
      *
      * @return the Java byte[] or nullptr if an exception occurs
+     * 
+     * @throws RocksDBException thrown 
+     *   if memory size to copy exceeds general java specific array size limitation.
      */
     static jbyteArray copyBytes(JNIEnv* env, std::string bytes) {
-      const jsize jlen = static_cast<jsize>(bytes.size());
-
-      jbyteArray jbytes = env->NewByteArray(jlen);
-      if(jbytes == nullptr) {
-        // exception thrown: OutOfMemoryError
-        return nullptr;
-      }
-
-      env->SetByteArrayRegion(jbytes, 0, jlen,
-        const_cast<jbyte*>(reinterpret_cast<const jbyte*>(bytes.c_str())));
-      if(env->ExceptionCheck()) {
-        // exception thrown: ArrayIndexOutOfBoundsException
-        env->DeleteLocalRef(jbytes);
-        return nullptr;
-      }
-
-      return jbytes;
+      return createJavaByteArrayWithSizeCheck(env, bytes.c_str(), bytes.size());
     }
 
     /**
@@ -4473,6 +4460,47 @@ class JniUtil {
 
       return jbyte_strings;
     }
+    
+    /**
+      * Copies bytes to a new jByteArray with the check of java array size limitation.
+      *
+      * @param bytes pointer to memory to copy to a new jByteArray
+      * @param size number of bytes to copy
+      *
+      * @return the Java byte[] or nullptr if an exception occurs
+      * 
+      * @throws RocksDBException thrown 
+      *   if memory size to copy exceeds general java array size limitation to avoid overflow.
+      */
+    static jbyteArray createJavaByteArrayWithSizeCheck(JNIEnv* env, const char* bytes, const size_t size) {
+      // Limitation for java array size is vm specific
+      // In general it cannot exceed Integer.MAX_VALUE (2^31 - 1)
+      // Current HotSpot VM limitation for array size is Integer.MAX_VALUE - 5 (2^31 - 1 - 5)
+      // It means that the next call to env->NewByteArray can still end with 
+      // OutOfMemoryError("Requested array size exceeds VM limit") coming from VM
+      static const size_t MAX_JARRAY_SIZE = (static_cast<size_t>(1)) << 31;
+      if(size > MAX_JARRAY_SIZE) {
+        rocksdb::RocksDBExceptionJni::ThrowNew(env, "Requested array size exceeds VM limit");
+        return nullptr;
+      }
+      
+      const jsize jlen = static_cast<jsize>(size);
+      jbyteArray jbytes = env->NewByteArray(jlen);
+      if(jbytes == nullptr) {
+        // exception thrown: OutOfMemoryError	
+        return nullptr;
+      }
+      
+      env->SetByteArrayRegion(jbytes, 0, jlen,
+        const_cast<jbyte*>(reinterpret_cast<const jbyte*>(bytes)));
+      if(env->ExceptionCheck()) {
+        // exception thrown: ArrayIndexOutOfBoundsException
+        env->DeleteLocalRef(jbytes);
+        return nullptr;
+      }
+
+      return jbytes;
+    }
 
     /**
      * Copies bytes from a rocksdb::Slice to a jByteArray
@@ -4481,25 +4509,12 @@ class JniUtil {
      * @param bytes The bytes to copy
      *
      * @return the Java byte[] or nullptr if an exception occurs
+     * 
+     * @throws RocksDBException thrown 
+     *   if memory size to copy exceeds general java specific array size limitation.
      */
     static jbyteArray copyBytes(JNIEnv* env, const Slice& bytes) {
-      const jsize jlen = static_cast<jsize>(bytes.size());
-
-      jbyteArray jbytes = env->NewByteArray(jlen);
-      if(jbytes == nullptr) {
-        // exception thrown: OutOfMemoryError
-        return nullptr;
-      }
-
-      env->SetByteArrayRegion(jbytes, 0, jlen,
-        const_cast<jbyte*>(reinterpret_cast<const jbyte*>(bytes.data())));
-      if(env->ExceptionCheck()) {
-        // exception thrown: ArrayIndexOutOfBoundsException
-        env->DeleteLocalRef(jbytes);
-        return nullptr;
-      }
-
-      return jbytes;
+      return createJavaByteArrayWithSizeCheck(env, bytes.data(), bytes.size());
     }
 
     /*

java/rocksjni/sst_file_manager.cc

@@ -138,7 +138,7 @@ jobject Java_org_rocksdb_SstFileManager_getTrackedFiles(JNIEnv* env,
 
   const rocksdb::HashMapJni::FnMapKV<const std::string, const uint64_t>
       fn_map_kv =
-          [env, &tracked_files](
+          [env](
               const std::pair<const std::string, const uint64_t>& pair) {
             const jstring jtracked_file_path =
                 env->NewStringUTF(pair.first.c_str());

java/rocksjni/transaction.cc

@@ -48,7 +48,7 @@ void Java_org_rocksdb_Transaction_setSnapshotOnNextOperation__JJ(
     jlong jtxn_notifier_handle) {
   auto* txn = reinterpret_cast<rocksdb::Transaction*>(jhandle);
   auto* txn_notifier =
-      reinterpret_cast<std::shared_ptr<rocksdb::TransactionNotifier>*>(
+      reinterpret_cast<std::shared_ptr<rocksdb::TransactionNotifierJniCallback>*>(
           jtxn_notifier_handle);
   txn->SetSnapshotOnNextOperation(*txn_notifier);
 }

java/rocksjni/transaction_db.cc

@@ -307,22 +307,14 @@ jobject Java_org_rocksdb_TransactionDB_getLockStatusData(JNIEnv* env,
     return nullptr;
   }
 
-<<<<<<< Updated upstream
   const rocksdb::HashMapJni::FnMapKV<const int32_t, const rocksdb::KeyLockInfo>
       fn_map_kv =
-          [env, txn_db, &lock_status_data](
+          [env](
               const std::pair<const int32_t, const rocksdb::KeyLockInfo>&
                   pair) {
             const jobject jlong_column_family_id =
                 rocksdb::LongJni::valueOf(env, pair.first);
             if (jlong_column_family_id == nullptr) {
-=======
-  const rocksdb::HashMapJni::FnMapKV<const int32_t, const rocksdb::KeyLockInfo> fn_map_kv =
-      [env](const std::pair<const int32_t, const rocksdb::KeyLockInfo>& pair) {
-          const jobject jlong_column_family_id =
-              rocksdb::LongJni::valueOf(env, pair.first);
-          if (jlong_column_family_id == nullptr) {
->>>>>>> Stashed changes
               // an error occurred
               return std::unique_ptr<std::pair<jobject, jobject>>(nullptr);
             }

java/src/main/java/org/rocksdb/ColumnFamilyHandle.java

@@ -28,8 +28,10 @@ public class ColumnFamilyHandle extends RocksObject {
    * Gets the name of the Column Family.
    *
    * @return The name of the Column Family.
+   *
+   * @throws RocksDBException if an error occurs whilst retrieving the name.
    */
-  public byte[] getName() {
+  public byte[] getName() throws RocksDBException {
     return getName(nativeHandle_);
   }
 
@@ -71,14 +73,22 @@ public class ColumnFamilyHandle extends RocksObject {
     }
 
     final ColumnFamilyHandle that = (ColumnFamilyHandle) o;
-    return rocksDB_.nativeHandle_ == that.rocksDB_.nativeHandle_ &&
-        getID() == that.getID() &&
-        Arrays.equals(getName(), that.getName());
+    try {
+      return rocksDB_.nativeHandle_ == that.rocksDB_.nativeHandle_ &&
+          getID() == that.getID() &&
+          Arrays.equals(getName(), that.getName());
+    } catch (RocksDBException e) {
+      throw new RuntimeException("Cannot compare column family handles", e);
+    }
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(getName(), getID(), rocksDB_.nativeHandle_);
+    try {
+      return Objects.hash(getName(), getID(), rocksDB_.nativeHandle_);
+    } catch (RocksDBException e) {
+      throw new RuntimeException("Cannot calculate hash code of column family handle", e);
+    }
   }
 
   /**
@@ -96,7 +106,7 @@ public class ColumnFamilyHandle extends RocksObject {
     }
   }
 
-  private native byte[] getName(final long handle);
+  private native byte[] getName(final long handle) throws RocksDBException;
   private native int getID(final long handle);
   private native ColumnFamilyDescriptor getDescriptor(final long handle) throws RocksDBException;
   @Override protected final native void disposeInternal(final long handle);

java/src/main/java/org/rocksdb/WriteBatch.java

@@ -65,8 +65,11 @@ public class WriteBatch extends AbstractWriteBatch {
    * Retrieve the serialized version of this batch.
    *
    * @return the serialized representation of this write batch.
+   *
+   * @throws RocksDBException if an error occurs whilst retrieving
+   *   the serialized batch data.
    */
-  public byte[] data() {
+  public byte[] data() throws RocksDBException {
     return data(nativeHandle_);
   }
 
@@ -253,7 +256,7 @@ public class WriteBatch extends AbstractWriteBatch {
       final int serializedLength);
   private native void iterate(final long handle, final long handlerHandle)
       throws RocksDBException;
-  private native byte[] data(final long nativeHandle);
+  private native byte[] data(final long nativeHandle) throws RocksDBException;
   private native long getDataSize(final long nativeHandle);
   private native boolean hasPut(final long nativeHandle);
   private native boolean hasDelete(final long nativeHandle);

java/src/test/java/org/rocksdb/RocksDBTest.java

@@ -4,9 +4,11 @@
 //  (found in the LICENSE.Apache file in the root directory).
 package org.rocksdb;
 
+import org.junit.Assume;
 import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 import org.junit.rules.TemporaryFolder;
 
 import java.util.*;
@@ -143,6 +145,39 @@ public class RocksDBTest {
     }
   }
 
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  @Test
+  public void getOutOfArrayMaxSizeValue() throws RocksDBException {
+    final int numberOfValueSplits = 10;
+    final int splitSize = Integer.MAX_VALUE / numberOfValueSplits;
+
+    Runtime runtime = Runtime.getRuntime();
+    long neededMemory = ((long)(splitSize)) * (((long)numberOfValueSplits) + 3);
+    boolean isEnoughMemory = runtime.maxMemory() - runtime.totalMemory() > neededMemory;
+    Assume.assumeTrue(isEnoughMemory);
+
+    final byte[] valueSplit = new byte[splitSize];
+    final byte[] key = "key".getBytes();
+
+    thrown.expect(RocksDBException.class);
+    thrown.expectMessage("Requested array size exceeds VM limit");
+
+    // merge (numberOfValueSplits + 1) valueSplit's to get value size exceeding Integer.MAX_VALUE
+    try (final StringAppendOperator stringAppendOperator = new StringAppendOperator();
+         final Options opt = new Options()
+                 .setCreateIfMissing(true)
+                 .setMergeOperator(stringAppendOperator);
+         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
+      db.put(key, valueSplit);
+      for (int i = 0; i < numberOfValueSplits; i++) {
+        db.merge(key, valueSplit);
+      }
+      db.get(key);
+    }
+  }
+
   @Test
   public void multiGet() throws RocksDBException, InterruptedException {
     try (final RocksDB db = RocksDB.open(dbFolder.getRoot().getAbsolutePath());

memtable/inlineskiplist.h

@@ -184,9 +184,10 @@ class InlineSkipList {
   const uint16_t kBranching_;
   const uint32_t kScaledInverseBranching_;
 
-  Allocator* const allocator_;  // Allocator used for allocations of nodes
   // Immutable after construction
   Comparator const compare_;
+  Allocator* const allocator_;  // Allocator used for allocations of nodes
+
   Node* const head_;
 
   // Modified only by Insert().  Read racily by readers, but stale
@@ -586,8 +587,8 @@ InlineSkipList<Comparator>::InlineSkipList(const Comparator cmp,
     : kMaxHeight_(static_cast<uint16_t>(max_height)),
       kBranching_(static_cast<uint16_t>(branching_factor)),
       kScaledInverseBranching_((Random::kMaxNext + 1) / kBranching_),
-      allocator_(allocator),
       compare_(cmp),
+      allocator_(allocator),
       head_(AllocateNode(0, max_height)),
       max_height_(1),
       seq_splice_(AllocateSplice()) {

monitoring/histogram.cc

@@ -235,8 +235,6 @@ void HistogramStat::Data(HistogramData * const data) const {
   data->max = static_cast<double>(max());
   data->average = Average();
   data->standard_deviation = StandardDeviation();
-  data->count = num();
-  data->sum = sum();
 }
 
 void HistogramImpl::Clear() {

monitoring/statistics.cc

@@ -169,17 +169,11 @@ std::string StatisticsImpl::ToString() const {
       char buffer[kTmpStrBufferSize];
       HistogramData hData;
       getHistogramImplLocked(h.first)->Data(&hData);
-      // don't handle failures - buffer should always be big enough and arguments
-      // should be provided correctly
-      int ret = snprintf(
+      snprintf(
           buffer, kTmpStrBufferSize,
-          "%s P50 : %f P95 : %f P99 : %f P100 : %f COUNT : %" PRIu64 " SUM : %"
-          PRIu64 "\n", h.second.c_str(), hData.median, hData.percentile95,
-          hData.percentile99, hData.max, hData.count, hData.sum);
-      if (ret < 0 || ret >= kTmpStrBufferSize) {
-        assert(false);
-        continue;
-      }
+          "%s statistics Percentiles :=> 50 : %f 95 : %f 99 : %f 100 : %f\n",
+          h.second.c_str(), hData.median, hData.percentile95,
+          hData.percentile99, hData.max);
       res.append(buffer);
     }
   }

options/cf_options.cc

@@ -27,6 +27,7 @@ ImmutableCFOptions::ImmutableCFOptions(const ImmutableDBOptions& db_options,
                                        const ColumnFamilyOptions& cf_options)
     : compaction_style(cf_options.compaction_style),
       compaction_pri(cf_options.compaction_pri),
+      prefix_extractor(cf_options.prefix_extractor.get()),
       user_comparator(cf_options.comparator),
       internal_comparator(InternalKeyComparator(cf_options.comparator)),
       merge_operator(cf_options.merge_operator.get()),
@@ -142,9 +143,6 @@ void MutableCFOptions::Dump(Logger* log) const {
   ROCKS_LOG_INFO(log,
                  "                 inplace_update_num_locks: %" ROCKSDB_PRIszt,
                  inplace_update_num_locks);
-  ROCKS_LOG_INFO(
-      log, "                         prefix_extractor: %s",
-      prefix_extractor == nullptr ? "nullptr" : prefix_extractor->Name());
   ROCKS_LOG_INFO(log, "                 disable_auto_compactions: %d",
                  disable_auto_compactions);
   ROCKS_LOG_INFO(log, "      soft_pending_compaction_bytes_limit: %" PRIu64,
@@ -191,7 +189,4 @@ void MutableCFOptions::Dump(Logger* log) const {
                  static_cast<int>(compression));
 }
 
-MutableCFOptions::MutableCFOptions(const Options& options)
-    : MutableCFOptions(ColumnFamilyOptions(options)) {}
-
 }  // namespace rocksdb

options/cf_options.h

@@ -30,6 +30,8 @@ struct ImmutableCFOptions {
 
   CompactionPri compaction_pri;
 
+  const SliceTransform* prefix_extractor;
+
   const Comparator* user_comparator;
   InternalKeyComparator internal_comparator;
 
@@ -132,7 +134,6 @@ struct MutableCFOptions {
         memtable_huge_page_size(options.memtable_huge_page_size),
         max_successive_merges(options.max_successive_merges),
         inplace_update_num_locks(options.inplace_update_num_locks),
-        prefix_extractor(options.prefix_extractor),
         disable_auto_compactions(options.disable_auto_compactions),
         soft_pending_compaction_bytes_limit(
             options.soft_pending_compaction_bytes_limit),
@@ -167,7 +168,6 @@ struct MutableCFOptions {
         memtable_huge_page_size(0),
         max_successive_merges(0),
         inplace_update_num_locks(0),
-        prefix_extractor(nullptr),
         disable_auto_compactions(false),
         soft_pending_compaction_bytes_limit(0),
         hard_pending_compaction_bytes_limit(0),
@@ -185,8 +185,6 @@ struct MutableCFOptions {
         report_bg_io_stats(false),
         compression(Snappy_Supported() ? kSnappyCompression : kNoCompression) {}
 
-  explicit MutableCFOptions(const Options& options);
-
   // Must be called after any change to MutableCFOptions
   void RefreshDerivedOptions(int num_levels, CompactionStyle compaction_style);
 
@@ -212,7 +210,6 @@ struct MutableCFOptions {
   size_t memtable_huge_page_size;
   size_t max_successive_merges;
   size_t inplace_update_num_locks;
-  std::shared_ptr<const SliceTransform> prefix_extractor;
 
   // Compaction related options
   bool disable_auto_compactions;

options/options_helper.cc

@@ -145,7 +145,6 @@ ColumnFamilyOptions BuildColumnFamilyOptions(
   cf_opts.max_successive_merges = mutable_cf_options.max_successive_merges;
   cf_opts.inplace_update_num_locks =
       mutable_cf_options.inplace_update_num_locks;
-  cf_opts.prefix_extractor = mutable_cf_options.prefix_extractor;
 
   // Compaction related options
   cf_opts.disable_auto_compactions =
@@ -384,8 +383,7 @@ bool ParseSliceTransformHelper(
     const std::string& kFixedPrefixName, const std::string& kCappedPrefixName,
     const std::string& value,
     std::shared_ptr<const SliceTransform>* slice_transform) {
-  const char* no_op_name = "rocksdb.Noop";
-  size_t no_op_length = strlen(no_op_name);
+
   auto& pe_value = value;
   if (pe_value.size() > kFixedPrefixName.size() &&
       pe_value.compare(0, kFixedPrefixName.size(), kFixedPrefixName) == 0) {
@@ -397,10 +395,6 @@ bool ParseSliceTransformHelper(
     int prefix_length =
         ParseInt(trim(pe_value.substr(kCappedPrefixName.size())));
     slice_transform->reset(NewCappedPrefixTransform(prefix_length));
-  } else if (pe_value.size() == no_op_length &&
-             pe_value.compare(0, no_op_length, no_op_name) == 0) {
-    const SliceTransform* no_op_transform = NewNoopTransform();
-    slice_transform->reset(no_op_transform);
   } else if (value == kNullptrString) {
     slice_transform->reset();
   } else {
@@ -1797,7 +1791,7 @@ std::unordered_map<std::string, OptionTypeInfo>
         {"prefix_extractor",
          {offset_of(&ColumnFamilyOptions::prefix_extractor),
           OptionType::kSliceTransform, OptionVerificationType::kByNameAllowNull,
-          true, offsetof(struct MutableCFOptions, prefix_extractor)}},
+          false, 0}},
         {"memtable_insert_with_hint_prefix_extractor",
          {offset_of(
               &ColumnFamilyOptions::memtable_insert_with_hint_prefix_extractor),

options/options_parser.cc

@@ -17,7 +17,6 @@
 #include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
 #include "util/cast_util.h"
-#include "util/file_reader_writer.h"
 #include "util/string_util.h"
 #include "util/sync_point.h"
 
@@ -42,16 +41,12 @@ Status PersistRocksDBOptions(const DBOptions& db_opt,
     return Status::InvalidArgument(
         "cf_names.size() and cf_opts.size() must be the same");
   }
-  std::unique_ptr<WritableFile> wf;
+  std::unique_ptr<WritableFile> writable;
 
-  Status s = env->NewWritableFile(file_name, &wf, EnvOptions());
+  Status s = env->NewWritableFile(file_name, &writable, EnvOptions());
   if (!s.ok()) {
     return s;
   }
-  unique_ptr<WritableFileWriter> writable;
-  writable.reset(new WritableFileWriter(std::move(wf), EnvOptions(),
-                                        nullptr /* statistics */));
-
   std::string options_file_content;
 
   writable->Append(option_file_header + "[" +
@@ -98,7 +93,8 @@ Status PersistRocksDBOptions(const DBOptions& db_opt,
       writable->Append(options_file_content + "\n");
     }
   }
-  writable->Sync(true /* use_fsync */);
+  writable->Flush();
+  writable->Fsync();
   writable->Close();
 
   return RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(

src.mk

@@ -250,8 +250,6 @@ MAIN_SOURCES =                                                          \
   cache/cache_bench.cc                                                  \
   cache/cache_test.cc                                                   \
   db/column_family_test.cc                                              \
-  db/compact_files_test.cc                                              \
-  db/compaction_iterator_test.cc                                        \
   db/compaction_job_stats_test.cc                                       \
   db/compaction_job_test.cc                                             \
   db/compaction_picker_test.cc                                          \
@@ -259,7 +257,6 @@ MAIN_SOURCES =                                                          \
   db/corruption_test.cc                                                 \
   db/cuckoo_table_db_test.cc                                            \
   db/db_basic_test.cc                                                   \
-  db/db_blob_index_test.cc                                              \
   db/db_block_cache_test.cc                                             \
   db/db_bloom_filter_test.cc                                            \
   db/db_compaction_filter_test.cc                                       \
@@ -276,50 +273,33 @@ MAIN_SOURCES =                                                          \
   db/db_memtable_test.cc                                                \
   db/db_merge_operator_test.cc                                          \
   db/db_options_test.cc                                                 \
-  db/db_properties_test.cc                                              \
   db/db_range_del_test.cc                                               \
   db/db_sst_test.cc                                                     \
   db/db_statistics_test.cc                                              \
   db/db_table_properties_test.cc                                        \
   db/db_tailing_iter_test.cc                                            \
   db/db_test.cc                                                         \
-  db/db_test2.cc                                                        \
   db/db_universal_compaction_test.cc                                    \
   db/db_wal_test.cc                                                     \
   db/db_write_test.cc                                                   \
   db/dbformat_test.cc                                                   \
   db/deletefile_test.cc                                                 \
-  db/env_timed_test.cc                                                  \
+  db/obsolete_files_test.cc						\
   db/external_sst_file_basic_test.cc                                    \
   db/external_sst_file_test.cc                                          \
   db/fault_injection_test.cc                                            \
   db/file_indexer_test.cc                                               \
-  db/file_reader_writer_test.cc                                         \
   db/filename_test.cc                                                   \
   db/flush_job_test.cc                                                  \
-  db/hash_table_test.cc                                                 \
-  db/hash_test.cc                                                       \
-  db/heap_test.cc                                                       \
   db/listener_test.cc                                                   \
   db/log_test.cc                                                        \
-  db/lru_cache_test.cc                                                  \
   db/manual_compaction_test.cc                                          \
-  db/memtable_list_test.cc                                              \
-  db/merge_helper_test.cc                                               \
   db/merge_test.cc                                                      \
-  db/obsolete_files_test.cc						\
-  db/options_settable_test.cc                                           \
   db/options_file_test.cc                                               \
-  db/partitioned_filter_block_test.cc                                   \
   db/perf_context_test.cc                                               \
-  db/persistent_cache_test.cc                                           \
   db/plain_table_db_test.cc                                             \
   db/prefix_test.cc                                                     \
-  db/redis_test.cc                                                      \
-  db/repair_test.cc                                                     \
-  db/range_del_aggregator_test.cc                                       \
   db/table_properties_collector_test.cc                                 \
-  db/util_merge_operators_test.cc                                       \
   db/version_builder_test.cc                                            \
   db/version_edit_test.cc                                               \
   db/version_set_test.cc                                                \

table/block.cc

@@ -402,14 +402,12 @@ Block::Block(BlockContents&& contents, SequenceNumber _global_seqno,
       data_(contents_.data.data()),
       size_(contents_.data.size()),
       restart_offset_(0),
-      num_restarts_(0),
       global_seqno_(_global_seqno) {
   if (size_ < sizeof(uint32_t)) {
     size_ = 0;  // Error marker
   } else {
-    num_restarts_ = NumRestarts();
     restart_offset_ =
-        static_cast<uint32_t>(size_) - (1 + num_restarts_) * sizeof(uint32_t);
+        static_cast<uint32_t>(size_) - (1 + NumRestarts()) * sizeof(uint32_t);
     if (restart_offset_ > size_ - sizeof(uint32_t)) {
       // The size is too small for NumRestarts() and therefore
       // restart_offset_ wrapped around.
@@ -434,14 +432,15 @@ BlockIter* Block::NewIterator(const Comparator* cmp, BlockIter* iter,
     ret_iter->Invalidate(Status::Corruption("bad block contents"));
     return ret_iter;
   }
-  if (num_restarts_ == 0) {
+  const uint32_t num_restarts = NumRestarts();
+  if (num_restarts == 0) {
     // Empty block.
     ret_iter->Invalidate(Status::OK());
     return ret_iter;
   } else {
     BlockPrefixIndex* prefix_index_ptr =
         total_order_seek ? nullptr : prefix_index_.get();
-    ret_iter->Initialize(cmp, data_, restart_offset_, num_restarts_,
+    ret_iter->Initialize(cmp, data_, restart_offset_, num_restarts,
                          prefix_index_ptr, global_seqno_,
                          read_amp_bitmap_.get());
 

table/block.h

@@ -184,7 +184,6 @@ class Block {
   const char* data_;            // contents_.data.data()
   size_t size_;                 // contents_.data.size()
   uint32_t restart_offset_;     // Offset in data_ of restart array
-  uint32_t num_restarts_;
   std::unique_ptr<BlockPrefixIndex> prefix_index_;
   std::unique_ptr<BlockReadAmpBitmap> read_amp_bitmap_;
   // All keys in the block will have seqno = global_seqno_, regardless of
@@ -243,7 +242,7 @@ class BlockIter final : public InternalIterator {
   }
 
   // Makes Valid() return false, status() return `s`, and Seek()/Prev()/etc do
-  // nothing.
+  // nothing. Calls cleanup functions.
   void Invalidate(Status s) {
     // Assert that the BlockIter is never deleted while Pinning is Enabled.
     assert(!pinned_iters_mgr_ ||
@@ -253,6 +252,9 @@ class BlockIter final : public InternalIterator {
     current_ = restarts_;
     status_ = s;
 
+    // Call cleanup callbacks.
+    Cleanable::Reset();
+
     // Clear prev entries cache.
     prev_entries_keys_buff_.clear();
     prev_entries_.clear();

table/block_based_filter_block.cc

@@ -186,8 +186,7 @@ BlockBasedFilterBlockReader::BlockBasedFilterBlockReader(
 }
 
 bool BlockBasedFilterBlockReader::KeyMayMatch(
-    const Slice& key, const SliceTransform* /* prefix_extractor */,
-    uint64_t block_offset, const bool /*no_io*/,
+    const Slice& key, uint64_t block_offset, const bool /*no_io*/,
     const Slice* const /*const_ikey_ptr*/) {
   assert(block_offset != kNotValid);
   if (!whole_key_filtering_) {
@@ -197,8 +196,7 @@ bool BlockBasedFilterBlockReader::KeyMayMatch(
 }
 
 bool BlockBasedFilterBlockReader::PrefixMayMatch(
-    const Slice& prefix, const SliceTransform* /* prefix_extractor */,
-    uint64_t block_offset, const bool /*no_io*/,
+    const Slice& prefix, uint64_t block_offset, const bool /*no_io*/,
     const Slice* const /*const_ikey_ptr*/) {
   assert(block_offset != kNotValid);
   if (!prefix_extractor_) {

table/block_based_filter_block.h

@@ -83,14 +83,13 @@ class BlockBasedFilterBlockReader : public FilterBlockReader {
                               bool whole_key_filtering,
                               BlockContents&& contents, Statistics* statistics);
   virtual bool IsBlockBased() override { return true; }
-
   virtual bool KeyMayMatch(
-      const Slice& key, const SliceTransform* prefix_extractor,
-      uint64_t block_offset = kNotValid, const bool no_io = false,
+      const Slice& key, uint64_t block_offset = kNotValid,
+      const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual bool PrefixMayMatch(
-      const Slice& prefix, const SliceTransform* prefix_extractor,
-      uint64_t block_offset = kNotValid, const bool no_io = false,
+      const Slice& prefix, uint64_t block_offset = kNotValid,
+      const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual size_t ApproximateMemoryUsage() const override;
 

table/block_based_filter_block_test.cc

@@ -59,8 +59,8 @@ TEST_F(FilterBlockTest, EmptyBuilder) {
   ASSERT_EQ("\\x00\\x00\\x00\\x00\\x0b", EscapeString(block.data));
   BlockBasedFilterBlockReader reader(nullptr, table_options_, true,
                                      std::move(block), nullptr);
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr, uint64_t{0}));
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr, 100000));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", 0));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", 100000));
 }
 
 TEST_F(FilterBlockTest, SingleChunk) {
@@ -78,13 +78,13 @@ TEST_F(FilterBlockTest, SingleChunk) {
   BlockContents block(builder.Finish(), false, kNoCompression);
   BlockBasedFilterBlockReader reader(nullptr, table_options_, true,
                                      std::move(block), nullptr);
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr, 100));
-  ASSERT_TRUE(reader.KeyMayMatch("bar", nullptr, 100));
-  ASSERT_TRUE(reader.KeyMayMatch("box", nullptr, 100));
-  ASSERT_TRUE(reader.KeyMayMatch("hello", nullptr, 100));
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr, 100));
-  ASSERT_TRUE(!reader.KeyMayMatch("missing", nullptr, 100));
-  ASSERT_TRUE(!reader.KeyMayMatch("other", nullptr, 100));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", 100));
+  ASSERT_TRUE(reader.KeyMayMatch("bar", 100));
+  ASSERT_TRUE(reader.KeyMayMatch("box", 100));
+  ASSERT_TRUE(reader.KeyMayMatch("hello", 100));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", 100));
+  ASSERT_TRUE(!reader.KeyMayMatch("missing", 100));
+  ASSERT_TRUE(!reader.KeyMayMatch("other", 100));
 }
 
 TEST_F(FilterBlockTest, MultiChunk) {
@@ -112,28 +112,28 @@ TEST_F(FilterBlockTest, MultiChunk) {
                                      std::move(block), nullptr);
 
   // Check first filter
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr, uint64_t{0}));
-  ASSERT_TRUE(reader.KeyMayMatch("bar", nullptr, 2000));
-  ASSERT_TRUE(!reader.KeyMayMatch("box", nullptr, uint64_t{0}));
-  ASSERT_TRUE(!reader.KeyMayMatch("hello", nullptr, uint64_t{0}));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", 0));
+  ASSERT_TRUE(reader.KeyMayMatch("bar", 2000));
+  ASSERT_TRUE(!reader.KeyMayMatch("box", 0));
+  ASSERT_TRUE(!reader.KeyMayMatch("hello", 0));
 
   // Check second filter
-  ASSERT_TRUE(reader.KeyMayMatch("box", nullptr, 3100));
-  ASSERT_TRUE(!reader.KeyMayMatch("foo", nullptr, 3100));
-  ASSERT_TRUE(!reader.KeyMayMatch("bar", nullptr, 3100));
-  ASSERT_TRUE(!reader.KeyMayMatch("hello", nullptr, 3100));
+  ASSERT_TRUE(reader.KeyMayMatch("box", 3100));
+  ASSERT_TRUE(!reader.KeyMayMatch("foo", 3100));
+  ASSERT_TRUE(!reader.KeyMayMatch("bar", 3100));
+  ASSERT_TRUE(!reader.KeyMayMatch("hello", 3100));
 
   // Check third filter (empty)
-  ASSERT_TRUE(!reader.KeyMayMatch("foo", nullptr, 4100));
-  ASSERT_TRUE(!reader.KeyMayMatch("bar", nullptr, 4100));
-  ASSERT_TRUE(!reader.KeyMayMatch("box", nullptr, 4100));
-  ASSERT_TRUE(!reader.KeyMayMatch("hello", nullptr, 4100));
+  ASSERT_TRUE(!reader.KeyMayMatch("foo", 4100));
+  ASSERT_TRUE(!reader.KeyMayMatch("bar", 4100));
+  ASSERT_TRUE(!reader.KeyMayMatch("box", 4100));
+  ASSERT_TRUE(!reader.KeyMayMatch("hello", 4100));
 
   // Check last filter
-  ASSERT_TRUE(reader.KeyMayMatch("box", nullptr, 9000));
-  ASSERT_TRUE(reader.KeyMayMatch("hello", nullptr, 9000));
-  ASSERT_TRUE(!reader.KeyMayMatch("foo", nullptr, 9000));
-  ASSERT_TRUE(!reader.KeyMayMatch("bar", nullptr, 9000));
+  ASSERT_TRUE(reader.KeyMayMatch("box", 9000));
+  ASSERT_TRUE(reader.KeyMayMatch("hello", 9000));
+  ASSERT_TRUE(!reader.KeyMayMatch("foo", 9000));
+  ASSERT_TRUE(!reader.KeyMayMatch("bar", 9000));
 }
 
 // Test for block based filter block
@@ -156,8 +156,8 @@ TEST_F(BlockBasedFilterBlockTest, BlockBasedEmptyBuilder) {
   ASSERT_EQ("\\x00\\x00\\x00\\x00\\x0b", EscapeString(block.data));
   FilterBlockReader* reader = new BlockBasedFilterBlockReader(
       nullptr, table_options_, true, std::move(block), nullptr);
-  ASSERT_TRUE(reader->KeyMayMatch("foo", nullptr, uint64_t{0}));
-  ASSERT_TRUE(reader->KeyMayMatch("foo", nullptr, 100000));
+  ASSERT_TRUE(reader->KeyMayMatch("foo", 0));
+  ASSERT_TRUE(reader->KeyMayMatch("foo", 100000));
 
   delete builder;
   delete reader;
@@ -177,13 +177,13 @@ TEST_F(BlockBasedFilterBlockTest, BlockBasedSingleChunk) {
   BlockContents block(builder->Finish(), false, kNoCompression);
   FilterBlockReader* reader = new BlockBasedFilterBlockReader(
       nullptr, table_options_, true, std::move(block), nullptr);
-  ASSERT_TRUE(reader->KeyMayMatch("foo", nullptr, 100));
-  ASSERT_TRUE(reader->KeyMayMatch("bar", nullptr, 100));
-  ASSERT_TRUE(reader->KeyMayMatch("box", nullptr, 100));
-  ASSERT_TRUE(reader->KeyMayMatch("hello", nullptr, 100));
-  ASSERT_TRUE(reader->KeyMayMatch("foo", nullptr, 100));
-  ASSERT_TRUE(!reader->KeyMayMatch("missing", nullptr, 100));
-  ASSERT_TRUE(!reader->KeyMayMatch("other", nullptr, 100));
+  ASSERT_TRUE(reader->KeyMayMatch("foo", 100));
+  ASSERT_TRUE(reader->KeyMayMatch("bar", 100));
+  ASSERT_TRUE(reader->KeyMayMatch("box", 100));
+  ASSERT_TRUE(reader->KeyMayMatch("hello", 100));
+  ASSERT_TRUE(reader->KeyMayMatch("foo", 100));
+  ASSERT_TRUE(!reader->KeyMayMatch("missing", 100));
+  ASSERT_TRUE(!reader->KeyMayMatch("other", 100));
 
   delete builder;
   delete reader;
@@ -215,28 +215,28 @@ TEST_F(BlockBasedFilterBlockTest, BlockBasedMultiChunk) {
       nullptr, table_options_, true, std::move(block), nullptr);
 
   // Check first filter
-  ASSERT_TRUE(reader->KeyMayMatch("foo", nullptr, uint64_t{0}));
-  ASSERT_TRUE(reader->KeyMayMatch("bar", nullptr, 2000));
-  ASSERT_TRUE(!reader->KeyMayMatch("box", nullptr, uint64_t{0}));
-  ASSERT_TRUE(!reader->KeyMayMatch("hello", nullptr, uint64_t{0}));
+  ASSERT_TRUE(reader->KeyMayMatch("foo", 0));
+  ASSERT_TRUE(reader->KeyMayMatch("bar", 2000));
+  ASSERT_TRUE(!reader->KeyMayMatch("box", 0));
+  ASSERT_TRUE(!reader->KeyMayMatch("hello", 0));
 
   // Check second filter
-  ASSERT_TRUE(reader->KeyMayMatch("box", nullptr, 3100));
-  ASSERT_TRUE(!reader->KeyMayMatch("foo", nullptr, 3100));
-  ASSERT_TRUE(!reader->KeyMayMatch("bar", nullptr, 3100));
-  ASSERT_TRUE(!reader->KeyMayMatch("hello", nullptr, 3100));
+  ASSERT_TRUE(reader->KeyMayMatch("box", 3100));
+  ASSERT_TRUE(!reader->KeyMayMatch("foo", 3100));
+  ASSERT_TRUE(!reader->KeyMayMatch("bar", 3100));
+  ASSERT_TRUE(!reader->KeyMayMatch("hello", 3100));
 
   // Check third filter (empty)
-  ASSERT_TRUE(!reader->KeyMayMatch("foo", nullptr, 4100));
-  ASSERT_TRUE(!reader->KeyMayMatch("bar", nullptr, 4100));
-  ASSERT_TRUE(!reader->KeyMayMatch("box", nullptr, 4100));
-  ASSERT_TRUE(!reader->KeyMayMatch("hello", nullptr, 4100));
+  ASSERT_TRUE(!reader->KeyMayMatch("foo", 4100));
+  ASSERT_TRUE(!reader->KeyMayMatch("bar", 4100));
+  ASSERT_TRUE(!reader->KeyMayMatch("box", 4100));
+  ASSERT_TRUE(!reader->KeyMayMatch("hello", 4100));
 
   // Check last filter
-  ASSERT_TRUE(reader->KeyMayMatch("box", nullptr, 9000));
-  ASSERT_TRUE(reader->KeyMayMatch("hello", nullptr, 9000));
-  ASSERT_TRUE(!reader->KeyMayMatch("foo", nullptr, 9000));
-  ASSERT_TRUE(!reader->KeyMayMatch("bar", nullptr, 9000));
+  ASSERT_TRUE(reader->KeyMayMatch("box", 9000));
+  ASSERT_TRUE(reader->KeyMayMatch("hello", 9000));
+  ASSERT_TRUE(!reader->KeyMayMatch("foo", 9000));
+  ASSERT_TRUE(!reader->KeyMayMatch("bar", 9000));
 
   delete builder;
   delete reader;

table/block_based_table_builder.cc

@@ -62,16 +62,14 @@ namespace {
 
 // Create a filter block builder based on its type.
 FilterBlockBuilder* CreateFilterBlockBuilder(
-    const ImmutableCFOptions& /*opt*/, const MutableCFOptions& mopt,
-    const BlockBasedTableOptions& table_opt,
+    const ImmutableCFOptions& opt, const BlockBasedTableOptions& table_opt,
     PartitionedIndexBuilder* const p_index_builder) {
   if (table_opt.filter_policy == nullptr) return nullptr;
 
   FilterBitsBuilder* filter_bits_builder =
       table_opt.filter_policy->GetFilterBitsBuilder();
   if (filter_bits_builder == nullptr) {
-    return new BlockBasedFilterBlockBuilder(mopt.prefix_extractor.get(),
-                                            table_opt);
+    return new BlockBasedFilterBlockBuilder(opt.prefix_extractor, table_opt);
   } else {
     if (table_opt.partition_filters) {
       assert(p_index_builder != nullptr);
@@ -84,11 +82,11 @@ FilterBlockBuilder* CreateFilterBlockBuilder(
           (100 - table_opt.block_size_deviation)) + 99) / 100);
       partition_size = std::max(partition_size, static_cast<uint32_t>(1));
       return new PartitionedFilterBlockBuilder(
-          mopt.prefix_extractor.get(), table_opt.whole_key_filtering,
+          opt.prefix_extractor, table_opt.whole_key_filtering,
           filter_bits_builder, table_opt.index_block_restart_interval,
           p_index_builder, partition_size);
     } else {
-      return new FullFilterBlockBuilder(mopt.prefix_extractor.get(),
+      return new FullFilterBlockBuilder(opt.prefix_extractor,
                                         table_opt.whole_key_filtering,
                                         filter_bits_builder);
     }
@@ -246,7 +244,6 @@ class BlockBasedTableBuilder::BlockBasedTablePropertiesCollector
 
 struct BlockBasedTableBuilder::Rep {
   const ImmutableCFOptions ioptions;
-  const MutableCFOptions moptions;
   const BlockBasedTableOptions table_options;
   const InternalKeyComparator& internal_comparator;
   WritableFileWriter* file;
@@ -283,7 +280,7 @@ struct BlockBasedTableBuilder::Rep {
 
   std::vector<std::unique_ptr<IntTblPropCollector>> table_properties_collectors;
 
-  Rep(const ImmutableCFOptions& _ioptions, const MutableCFOptions& _moptions,
+  Rep(const ImmutableCFOptions& _ioptions,
       const BlockBasedTableOptions& table_opt,
       const InternalKeyComparator& icomparator,
       const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
@@ -295,7 +292,6 @@ struct BlockBasedTableBuilder::Rep {
       const std::string& _column_family_name, const uint64_t _creation_time,
       const uint64_t _oldest_key_time)
       : ioptions(_ioptions),
-        moptions(_moptions),
         table_options(table_opt),
         internal_comparator(icomparator),
         file(f),
@@ -304,8 +300,8 @@ struct BlockBasedTableBuilder::Rep {
                       : 0),
         data_block(table_options.block_restart_interval,
                    table_options.use_delta_encoding),
-        range_del_block(1 /* block_restart_interval */),
-        internal_prefix_transform(_moptions.prefix_extractor.get()),
+        range_del_block(1),  // TODO(andrewkr): restart_interval unnecessary
+        internal_prefix_transform(_ioptions.prefix_extractor),
         compression_type(_compression_type),
         compression_opts(_compression_opts),
         compression_dict(_compression_dict),
@@ -330,8 +326,8 @@ struct BlockBasedTableBuilder::Rep {
     if (skip_filters) {
       filter_builder = nullptr;
     } else {
-      filter_builder.reset(CreateFilterBlockBuilder(
-          _ioptions, _moptions, table_options, p_index_builder_));
+      filter_builder.reset(
+          CreateFilterBlockBuilder(_ioptions, table_options, p_index_builder_));
     }
 
     for (auto& collector_factories : *int_tbl_prop_collector_factories) {
@@ -341,12 +337,12 @@ struct BlockBasedTableBuilder::Rep {
     table_properties_collectors.emplace_back(
         new BlockBasedTablePropertiesCollector(
             table_options.index_type, table_options.whole_key_filtering,
-            _moptions.prefix_extractor != nullptr));
+            _ioptions.prefix_extractor != nullptr));
   }
 };
 
 BlockBasedTableBuilder::BlockBasedTableBuilder(
-    const ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
+    const ImmutableCFOptions& ioptions,
     const BlockBasedTableOptions& table_options,
     const InternalKeyComparator& internal_comparator,
     const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
@@ -369,11 +365,11 @@ BlockBasedTableBuilder::BlockBasedTableBuilder(
     sanitized_table_options.format_version = 1;
   }
 
-  rep_ =
-      new Rep(ioptions, moptions, sanitized_table_options, internal_comparator,
-              int_tbl_prop_collector_factories, column_family_id, file,
-              compression_type, compression_opts, compression_dict,
-              skip_filters, column_family_name, creation_time, oldest_key_time);
+  rep_ = new Rep(ioptions, sanitized_table_options, internal_comparator,
+                 int_tbl_prop_collector_factories, column_family_id, file,
+                 compression_type, compression_opts, compression_dict,
+                 skip_filters, column_family_name, creation_time,
+                 oldest_key_time);
 
   if (rep_->filter_builder != nullptr) {
     rep_->filter_builder->StartBlock(0);
@@ -741,8 +737,8 @@ Status BlockBasedTableBuilder::Finish() {
                                          : "nullptr";
       r->props.compression_name = CompressionTypeToString(r->compression_type);
       r->props.prefix_extractor_name =
-          r->moptions.prefix_extractor != nullptr
-              ? r->moptions.prefix_extractor->Name()
+          r->ioptions.prefix_extractor != nullptr
+              ? r->ioptions.prefix_extractor->Name()
               : "nullptr";
 
       std::string property_collectors_names = "[";

table/block_based_table_builder.h

@@ -38,7 +38,7 @@ class BlockBasedTableBuilder : public TableBuilder {
   // @param compression_dict Data for presetting the compression library's
   //    dictionary, or nullptr.
   BlockBasedTableBuilder(
-      const ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
+      const ImmutableCFOptions& ioptions,
       const BlockBasedTableOptions& table_options,
       const InternalKeyComparator& internal_comparator,
       const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*

table/block_based_table_factory.cc

@@ -69,17 +69,16 @@ Status BlockBasedTableFactory::NewTableReader(
   return BlockBasedTable::Open(
       table_reader_options.ioptions, table_reader_options.env_options,
       table_options_, table_reader_options.internal_comparator, std::move(file),
-      file_size, table_reader, table_reader_options.prefix_extractor,
-      prefetch_index_and_filter_in_cache, table_reader_options.skip_filters,
-      table_reader_options.level);
+      file_size, table_reader, prefetch_index_and_filter_in_cache,
+      table_reader_options.skip_filters, table_reader_options.level);
 }
 
 TableBuilder* BlockBasedTableFactory::NewTableBuilder(
     const TableBuilderOptions& table_builder_options, uint32_t column_family_id,
     WritableFileWriter* file) const {
   auto table_builder = new BlockBasedTableBuilder(
-      table_builder_options.ioptions, table_builder_options.moptions,
-      table_options_, table_builder_options.internal_comparator,
+      table_builder_options.ioptions, table_options_,
+      table_builder_options.internal_comparator,
       table_builder_options.int_tbl_prop_collector_factories, column_family_id,
       file, table_builder_options.compression_type,
       table_builder_options.compression_opts,

table/block_based_table_reader.cc

@@ -173,29 +173,6 @@ Cache::Handle* GetEntryFromCache(Cache* block_cache, const Slice& key,
   return cache_handle;
 }
 
-// For hash based index, return true if prefix_extractor and
-// prefix_extractor_block mismatch, false otherwise. This flag will be used
-// as total_order_seek via NewIndexIterator
-bool PrefixExtractorChanged(
-    std::shared_ptr<const TableProperties> table_properties,
-    const SliceTransform* prefix_extractor) {
-  // BlockBasedTableOptions::kHashSearch requires prefix_extractor to be set.
-  // Turn off hash index in prefix_extractor is not set; if  prefix_extractor
-  // is set but prefix_extractor_block is not set, also disable hash index
-  if (prefix_extractor == nullptr || table_properties == nullptr ||
-      table_properties->prefix_extractor_name.empty()) {
-    return true;
-  }
-
-  // prefix_extractor and prefix_extractor_block are both non-empty
-  if (table_properties->prefix_extractor_name.compare(
-          prefix_extractor->Name()) != 0) {
-    return true;
-  } else {
-    return false;
-  }
-}
-
 }  // namespace
 
 // Index that allows binary search lookup in a two-level index structure.
@@ -244,8 +221,7 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
       ro.fill_cache = fill_cache;
       return new BlockBasedTableIterator(
           table_, ro, *icomparator_,
-          index_block_->NewIterator(icomparator_, nullptr, true), false,
-          /* prefix_extractor */ nullptr);
+          index_block_->NewIterator(icomparator_, nullptr, true), false);
     }
     // TODO(myabandeh): Update TwoLevelIterator to be able to make use of
     // on-stack BlockIter while the state is on heap. Currentlly it assumes
@@ -670,7 +646,6 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
                              unique_ptr<RandomAccessFileReader>&& file,
                              uint64_t file_size,
                              unique_ptr<TableReader>* table_reader,
-                             const SliceTransform* prefix_extractor,
                              const bool prefetch_index_and_filter_in_cache,
                              const bool skip_filters, const int level) {
   table_reader->reset();
@@ -722,7 +697,7 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
   // We need to wrap data with internal_prefix_transform to make sure it can
   // handle prefix correctly.
   rep->internal_prefix_transform.reset(
-      new InternalKeySliceTransform(prefix_extractor));
+      new InternalKeySliceTransform(rep->ioptions.prefix_extractor));
   SetupCacheKeyPrefix(rep, file_size);
   unique_ptr<BlockBasedTable> new_table(new BlockBasedTable(rep));
 
@@ -888,14 +863,8 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
       // block_cache
 
       CachableEntry<IndexReader> index_entry;
-      bool prefix_extractor_changed = false;
-      // check prefix_extractor match only if hash based index is used
-      if (rep->index_type == BlockBasedTableOptions::kHashSearch) {
-        prefix_extractor_changed =
-            PrefixExtractorChanged(rep->table_properties, prefix_extractor);
-      }
-      unique_ptr<InternalIterator> iter(new_table->NewIndexIterator(
-          ReadOptions(), prefix_extractor_changed, nullptr, &index_entry));
+      unique_ptr<InternalIterator> iter(
+          new_table->NewIndexIterator(ReadOptions(), nullptr, &index_entry));
       s = iter->status();
       if (s.ok()) {
         // This is the first call to NewIndexIterator() since we're in Open().
@@ -910,9 +879,9 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
         }
 
         // Hack: Call GetFilter() to implicitly add filter to the block_cache
-        auto filter_entry = new_table->GetFilter(prefix_extractor);
+        auto filter_entry = new_table->GetFilter();
         if (filter_entry.value != nullptr) {
-          filter_entry.value->CacheDependencies(pin, prefix_extractor);
+          filter_entry.value->CacheDependencies(pin);
         }
         // if pin_l0_filter_and_index_blocks_in_cache is true, and this is
         // a level0 file, then save it in rep_->filter_entry; it will be
@@ -944,14 +913,13 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
       // Set filter block
       if (rep->filter_policy) {
         const bool is_a_filter_partition = true;
-        auto filter =
-            new_table->ReadFilter(prefetch_buffer.get(), rep->filter_handle,
-                                  !is_a_filter_partition, prefix_extractor);
+        auto filter = new_table->ReadFilter(
+            prefetch_buffer.get(), rep->filter_handle, !is_a_filter_partition);
         rep->filter.reset(filter);
         // Refer to the comment above about paritioned indexes always being
         // cached
         if (filter && (prefetch_index_and_filter_in_cache || level == 0)) {
-          filter->CacheDependencies(pin, prefix_extractor);
+          filter->CacheDependencies(pin);
         }
       }
     } else {
@@ -1247,8 +1215,7 @@ Status BlockBasedTable::PutDataBlockToCache(
 
 FilterBlockReader* BlockBasedTable::ReadFilter(
     FilePrefetchBuffer* prefetch_buffer, const BlockHandle& filter_handle,
-    const bool is_a_filter_partition,
-    const SliceTransform* prefix_extractor) const {
+    const bool is_a_filter_partition) const {
   auto& rep = rep_;
   // TODO: We might want to unify with ReadBlockFromFile() if we start
   // requiring checksum verification in Table::Open.
@@ -1281,14 +1248,14 @@ FilterBlockReader* BlockBasedTable::ReadFilter(
   switch (filter_type) {
     case Rep::FilterType::kPartitionedFilter: {
       return new PartitionedFilterBlockReader(
-          rep->prefix_filtering ? prefix_extractor : nullptr,
+          rep->prefix_filtering ? rep->ioptions.prefix_extractor : nullptr,
           rep->whole_key_filtering, std::move(block), nullptr,
           rep->ioptions.statistics, rep->internal_comparator, this);
     }
 
     case Rep::FilterType::kBlockFilter:
       return new BlockBasedFilterBlockReader(
-          rep->prefix_filtering ? prefix_extractor : nullptr,
+          rep->prefix_filtering ? rep->ioptions.prefix_extractor : nullptr,
           rep->table_options, rep->whole_key_filtering, std::move(block),
           rep->ioptions.statistics);
 
@@ -1297,7 +1264,7 @@ FilterBlockReader* BlockBasedTable::ReadFilter(
           rep->filter_policy->GetFilterBitsReader(block.data);
       assert(filter_bits_reader != nullptr);
       return new FullFilterBlockReader(
-          rep->prefix_filtering ? prefix_extractor : nullptr,
+          rep->prefix_filtering ? rep->ioptions.prefix_extractor : nullptr,
           rep->whole_key_filtering, std::move(block), filter_bits_reader,
           rep->ioptions.statistics);
     }
@@ -1311,18 +1278,18 @@ FilterBlockReader* BlockBasedTable::ReadFilter(
 }
 
 BlockBasedTable::CachableEntry<FilterBlockReader> BlockBasedTable::GetFilter(
-    const SliceTransform* prefix_extractor, FilePrefetchBuffer* prefetch_buffer,
-    bool no_io, GetContext* get_context) const {
+    FilePrefetchBuffer* prefetch_buffer, bool no_io,
+    GetContext* get_context) const {
   const BlockHandle& filter_blk_handle = rep_->filter_handle;
   const bool is_a_filter_partition = true;
   return GetFilter(prefetch_buffer, filter_blk_handle, !is_a_filter_partition,
-                   no_io, get_context, prefix_extractor);
+                   no_io, get_context);
 }
 
 BlockBasedTable::CachableEntry<FilterBlockReader> BlockBasedTable::GetFilter(
     FilePrefetchBuffer* prefetch_buffer, const BlockHandle& filter_blk_handle,
-    const bool is_a_filter_partition, bool no_io, GetContext* get_context,
-    const SliceTransform* prefix_extractor) const {
+    const bool is_a_filter_partition, bool no_io,
+    GetContext* get_context) const {
   // If cache_index_and_filter_blocks is false, filter should be pre-populated.
   // We will return rep_->filter anyway. rep_->filter can be nullptr if filter
   // read fails at Open() time. We don't want to reload again since it will
@@ -1362,8 +1329,8 @@ BlockBasedTable::CachableEntry<FilterBlockReader> BlockBasedTable::GetFilter(
     // Do not invoke any io.
     return CachableEntry<FilterBlockReader>();
   } else {
-    filter = ReadFilter(prefetch_buffer, filter_blk_handle,
-                        is_a_filter_partition, prefix_extractor);
+    filter =
+        ReadFilter(prefetch_buffer, filter_blk_handle, is_a_filter_partition);
     if (filter != nullptr) {
       Status s = block_cache->Insert(
           key, filter, filter->size(), &DeleteCachedFilterEntry, &cache_handle,
@@ -1395,23 +1362,18 @@ BlockBasedTable::CachableEntry<FilterBlockReader> BlockBasedTable::GetFilter(
   return { filter, cache_handle };
 }
 
-// disable_prefix_seek should be set to true when prefix_extractor found in SST
-// differs from the one in mutable_cf_options and index type is HashBasedIndex
 InternalIterator* BlockBasedTable::NewIndexIterator(
-    const ReadOptions& read_options, bool disable_prefix_seek,
-    BlockIter* input_iter, CachableEntry<IndexReader>* index_entry,
-    GetContext* get_context) {
+    const ReadOptions& read_options, BlockIter* input_iter,
+    CachableEntry<IndexReader>* index_entry, GetContext* get_context) {
   // index reader has already been pre-populated.
   if (rep_->index_reader) {
     return rep_->index_reader->NewIterator(
-        input_iter, read_options.total_order_seek || disable_prefix_seek,
-        read_options.fill_cache);
+        input_iter, read_options.total_order_seek, read_options.fill_cache);
   }
   // we have a pinned index block
   if (rep_->index_entry.IsSet()) {
     return rep_->index_entry.value->NewIterator(
-        input_iter, read_options.total_order_seek || disable_prefix_seek,
-        read_options.fill_cache);
+        input_iter, read_options.total_order_seek, read_options.fill_cache);
   }
 
   PERF_TIMER_GUARD(read_index_block_nanos);
@@ -1487,7 +1449,7 @@ InternalIterator* BlockBasedTable::NewIndexIterator(
 
   assert(cache_handle);
   auto* iter = index_reader->NewIterator(
-      input_iter, read_options.total_order_seek || disable_prefix_seek);
+      input_iter, read_options.total_order_seek);
 
   // the caller would like to take ownership of the index block
   // don't call RegisterCleanup() in this case, the caller will take care of it
@@ -1720,32 +1682,31 @@ BlockBasedTable::PartitionedIndexIteratorState::NewSecondaryIterator(
 // Otherwise, this method guarantees no I/O will be incurred.
 //
 // REQUIRES: this method shouldn't be called while the DB lock is held.
-bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key,
-                                     const SliceTransform* prefix_extractor) {
+bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key) {
   if (!rep_->filter_policy) {
     return true;
   }
 
-  assert(prefix_extractor != nullptr);
+  assert(rep_->ioptions.prefix_extractor != nullptr);
   auto user_key = ExtractUserKey(internal_key);
-  if (!prefix_extractor->InDomain(user_key)) {
+  if (!rep_->ioptions.prefix_extractor->InDomain(user_key) ||
+      rep_->table_properties->prefix_extractor_name.compare(
+          rep_->ioptions.prefix_extractor->Name()) != 0) {
     return true;
   }
-  assert(rep_->table_properties->prefix_extractor_name.compare(
-             prefix_extractor->Name()) == 0);
-  auto prefix = prefix_extractor->Transform(user_key);
+  auto prefix = rep_->ioptions.prefix_extractor->Transform(user_key);
 
   bool may_match = true;
   Status s;
 
   // First, try check with full filter
-  auto filter_entry = GetFilter(prefix_extractor);
+  auto filter_entry = GetFilter();
   FilterBlockReader* filter = filter_entry.value;
   if (filter != nullptr) {
     if (!filter->IsBlockBased()) {
       const Slice* const const_ikey_ptr = &internal_key;
-      may_match = filter->PrefixMayMatch(prefix, prefix_extractor, kNotValid,
-                                         false, const_ikey_ptr);
+      may_match =
+          filter->PrefixMayMatch(prefix, kNotValid, false, const_ikey_ptr);
     } else {
       InternalKey internal_key_prefix(prefix, kMaxSequenceNumber, kTypeValue);
       auto internal_prefix = internal_key_prefix.Encode();
@@ -1757,11 +1718,7 @@ bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key,
       no_io_read_options.read_tier = kBlockCacheTier;
 
       // Then, try find it within each block
-      // we already know prefix_extractor and prefix_extractor_name must match
-      // because `CheckPrefixMayMatch` first checks `check_filter_ == true`
-      bool prefix_extractor_changed = false;
-      unique_ptr<InternalIterator> iiter(
-          NewIndexIterator(no_io_read_options, prefix_extractor_changed));
+      unique_ptr<InternalIterator> iiter(NewIndexIterator(no_io_read_options));
       iiter->Seek(internal_prefix);
 
       if (!iiter->Valid()) {
@@ -1793,8 +1750,7 @@ bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key,
         BlockHandle handle;
         s = handle.DecodeFrom(&handle_value);
         assert(s.ok());
-        may_match =
-            filter->PrefixMayMatch(prefix, prefix_extractor, handle.offset());
+        may_match = filter->PrefixMayMatch(prefix, handle.offset());
       }
     }
   }
@@ -1816,7 +1772,7 @@ bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key,
 }
 
 void BlockBasedTableIterator::Seek(const Slice& target) {
-  if (!CheckPrefixMayMatch(target, prefix_extractor_)) {
+  if (!CheckPrefixMayMatch(target)) {
     ResetDataIter();
     return;
   }
@@ -1840,7 +1796,7 @@ void BlockBasedTableIterator::Seek(const Slice& target) {
 }
 
 void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
-  if (!CheckPrefixMayMatch(target, prefix_extractor_)) {
+  if (!CheckPrefixMayMatch(target)) {
     ResetDataIter();
     return;
   }
@@ -2019,29 +1975,22 @@ void BlockBasedTableIterator::FindKeyBackward() {
   // code simplicity.
 }
 
-InternalIterator* BlockBasedTable::NewIterator(
-    const ReadOptions& read_options, const SliceTransform* prefix_extractor,
-    Arena* arena, bool skip_filters) {
-  bool prefix_extractor_changed =
-      PrefixExtractorChanged(rep_->table_properties, prefix_extractor);
+InternalIterator* BlockBasedTable::NewIterator(const ReadOptions& read_options,
+                                               Arena* arena,
+                                               bool skip_filters) {
   if (arena == nullptr) {
     return new BlockBasedTableIterator(
         this, read_options, rep_->internal_comparator,
-        NewIndexIterator(
-            read_options,
-            prefix_extractor_changed &&
-            rep_->index_type == BlockBasedTableOptions::kHashSearch),
+        NewIndexIterator(read_options),
         !skip_filters && !read_options.total_order_seek &&
-        prefix_extractor != nullptr && !prefix_extractor_changed,
-        prefix_extractor);
+            rep_->ioptions.prefix_extractor != nullptr);
   } else {
     auto* mem = arena->AllocateAligned(sizeof(BlockBasedTableIterator));
     return new (mem) BlockBasedTableIterator(
         this, read_options, rep_->internal_comparator,
-        NewIndexIterator(read_options, prefix_extractor_changed),
+        NewIndexIterator(read_options),
         !skip_filters && !read_options.total_order_seek &&
-        prefix_extractor != nullptr && !prefix_extractor_changed,
-        prefix_extractor);
+            rep_->ioptions.prefix_extractor != nullptr);
   }
 }
 
@@ -2075,10 +2024,10 @@ InternalIterator* BlockBasedTable::NewRangeTombstoneIterator(
   return NewDataBlockIterator(rep_, read_options, Slice(str));
 }
 
-bool BlockBasedTable::FullFilterKeyMayMatch(
-    const ReadOptions& read_options, FilterBlockReader* filter,
-    const Slice& internal_key, const bool no_io,
-    const SliceTransform* prefix_extractor) const {
+bool BlockBasedTable::FullFilterKeyMayMatch(const ReadOptions& read_options,
+                                            FilterBlockReader* filter,
+                                            const Slice& internal_key,
+                                            const bool no_io) const {
   if (filter == nullptr || filter->IsBlockBased()) {
     return true;
   }
@@ -2086,15 +2035,15 @@ bool BlockBasedTable::FullFilterKeyMayMatch(
   const Slice* const const_ikey_ptr = &internal_key;
   bool may_match = true;
   if (filter->whole_key_filtering()) {
-    may_match = filter->KeyMayMatch(user_key, prefix_extractor, kNotValid,
-                                    no_io, const_ikey_ptr);
-  } else if (!read_options.total_order_seek && prefix_extractor &&
+    may_match = filter->KeyMayMatch(user_key, kNotValid, no_io, const_ikey_ptr);
+  } else if (!read_options.total_order_seek &&
+             rep_->ioptions.prefix_extractor &&
              rep_->table_properties->prefix_extractor_name.compare(
-                 prefix_extractor->Name()) == 0 &&
-             prefix_extractor->InDomain(user_key) &&
-             !filter->PrefixMayMatch(prefix_extractor->Transform(user_key),
-                                     prefix_extractor, kNotValid, false,
-                                     const_ikey_ptr)) {
+                 rep_->ioptions.prefix_extractor->Name()) == 0 &&
+             rep_->ioptions.prefix_extractor->InDomain(user_key) &&
+             !filter->PrefixMayMatch(
+                 rep_->ioptions.prefix_extractor->Transform(user_key),
+                 kNotValid, false, const_ikey_ptr)) {
     may_match = false;
   }
   if (may_match) {
@@ -2104,36 +2053,25 @@ bool BlockBasedTable::FullFilterKeyMayMatch(
 }
 
 Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
-                            GetContext* get_context,
-                            const SliceTransform* prefix_extractor,
-                            bool skip_filters) {
+                            GetContext* get_context, bool skip_filters) {
   Status s;
   const bool no_io = read_options.read_tier == kBlockCacheTier;
   CachableEntry<FilterBlockReader> filter_entry;
   if (!skip_filters) {
     filter_entry =
-        GetFilter(prefix_extractor, /*prefetch_buffer*/ nullptr,
+        GetFilter(/*prefetch_buffer*/ nullptr,
                   read_options.read_tier == kBlockCacheTier, get_context);
   }
   FilterBlockReader* filter = filter_entry.value;
 
   // First check the full filter
   // If full filter not useful, Then go into each block
-  if (!FullFilterKeyMayMatch(read_options, filter, key, no_io,
-                             prefix_extractor)) {
+  if (!FullFilterKeyMayMatch(read_options, filter, key, no_io)) {
     RecordTick(rep_->ioptions.statistics, BLOOM_FILTER_USEFUL);
   } else {
     BlockIter iiter_on_stack;
-    // if prefix_extractor found in block differs from options, disable
-    // BlockPrefixIndex. Only do this check when index_type is kHashSearch.
-    bool prefix_extractor_changed = false;
-    if (rep_->index_type == BlockBasedTableOptions::kHashSearch) {
-      prefix_extractor_changed =
-          PrefixExtractorChanged(rep_->table_properties, prefix_extractor);
-    }
-    auto iiter = NewIndexIterator(read_options, prefix_extractor_changed,
-                                  &iiter_on_stack, /* index_entry */ nullptr,
-                                  get_context);
+    auto iiter = NewIndexIterator(read_options, &iiter_on_stack,
+                                  /* index_entry */ nullptr, get_context);
     std::unique_ptr<InternalIterator> iiter_unique_ptr;
     if (iiter != &iiter_on_stack) {
       iiter_unique_ptr.reset(iiter);
@@ -2148,8 +2086,7 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
       bool not_exist_in_filter =
           filter != nullptr && filter->IsBlockBased() == true &&
           handle.DecodeFrom(&handle_value).ok() &&
-          !filter->KeyMayMatch(ExtractUserKey(key), prefix_extractor,
-                               handle.offset(), no_io);
+          !filter->KeyMayMatch(ExtractUserKey(key), handle.offset(), no_io);
 
       if (not_exist_in_filter) {
         // Not found
@@ -2221,7 +2158,7 @@ Status BlockBasedTable::Prefetch(const Slice* const begin,
   }
 
   BlockIter iiter_on_stack;
-  auto iiter = NewIndexIterator(ReadOptions(), false, &iiter_on_stack);
+  auto iiter = NewIndexIterator(ReadOptions(), &iiter_on_stack);
   std::unique_ptr<InternalIterator> iiter_unique_ptr;
   if (iiter != &iiter_on_stack) {
     iiter_unique_ptr = std::unique_ptr<InternalIterator>(iiter);
@@ -2278,8 +2215,7 @@ Status BlockBasedTable::VerifyChecksum() {
   }
   // Check Data blocks
   BlockIter iiter_on_stack;
-  InternalIterator* iiter =
-      NewIndexIterator(ReadOptions(), false, &iiter_on_stack);
+  InternalIterator* iiter = NewIndexIterator(ReadOptions(), &iiter_on_stack);
   std::unique_ptr<InternalIterator> iiter_unique_ptr;
   if (iiter != &iiter_on_stack) {
     iiter_unique_ptr = std::unique_ptr<InternalIterator>(iiter);
@@ -2372,20 +2308,20 @@ Status BlockBasedTable::CreateIndexReader(
     if (pos != props.end()) {
       index_type_on_file = static_cast<BlockBasedTableOptions::IndexType>(
           DecodeFixed32(pos->second.c_str()));
-      // update index_type with the true type
-      rep_->index_type = index_type_on_file;
     }
   }
 
   auto file = rep_->file.get();
   const InternalKeyComparator* icomparator = &rep_->internal_comparator;
   const Footer& footer = rep_->footer;
-
-  // kHashSearch requires non-empty prefix_extractor but bypass checking
-  // prefix_extractor here since we have no access to MutableCFOptions.
-  // Add prefix_extractor_changed flag in  BlockBasedTable::NewIndexIterator.
-  // If prefix_extractor does not match prefix_extractor_name from table
-  // properties, turn off Hash Index by setting total_order_seek to true
+  if (index_type_on_file == BlockBasedTableOptions::kHashSearch &&
+      rep_->ioptions.prefix_extractor == nullptr) {
+    ROCKS_LOG_WARN(rep_->ioptions.info_log,
+                   "BlockBasedTableOptions::kHashSearch requires "
+                   "options.prefix_extractor to be set."
+                   " Fall back to binary search index.");
+    index_type_on_file = BlockBasedTableOptions::kBinarySearch;
+  }
 
   switch (index_type_on_file) {
     case BlockBasedTableOptions::kTwoLevelIndexSearch: {
@@ -2525,8 +2461,7 @@ Status BlockBasedTable::GetKVPairsFromDataBlocks(
   return Status::OK();
 }
 
-Status BlockBasedTable::DumpTable(WritableFile* out_file,
-                                  const SliceTransform* prefix_extractor) {
+Status BlockBasedTable::DumpTable(WritableFile* out_file) {
   // Output Footer
   out_file->Append(
       "Footer Details:\n"
@@ -2607,7 +2542,7 @@ Status BlockBasedTable::DumpTable(WritableFile* out_file,
         s = block_fetcher.ReadBlockContents();
         if (!s.ok()) {
           rep_->filter.reset(new BlockBasedFilterBlockReader(
-              prefix_extractor, table_options,
+              rep_->ioptions.prefix_extractor, table_options,
               table_options.whole_key_filtering, std::move(block),
               rep_->ioptions.statistics));
         }
@@ -2692,6 +2627,7 @@ Status BlockBasedTable::DumpIndexBlock(WritableFile* out_file) {
   out_file->Append(
       "Index Details:\n"
       "--------------------------------------\n");
+
   std::unique_ptr<InternalIterator> blockhandles_iter(
       NewIndexIterator(ReadOptions()));
   Status s = blockhandles_iter->status();

table/block_based_table_reader.h

@@ -90,29 +90,25 @@ class BlockBasedTable : public TableReader {
                      const InternalKeyComparator& internal_key_comparator,
                      unique_ptr<RandomAccessFileReader>&& file,
                      uint64_t file_size, unique_ptr<TableReader>* table_reader,
-                     const SliceTransform* prefix_extractor = nullptr,
                      bool prefetch_index_and_filter_in_cache = true,
                      bool skip_filters = false, int level = -1);
 
-  bool PrefixMayMatch(const Slice& internal_key,
-                      const SliceTransform* prefix_extractor = nullptr);
+  bool PrefixMayMatch(const Slice& internal_key);
 
   // Returns a new iterator over the table contents.
   // The result of NewIterator() is initially invalid (caller must
   // call one of the Seek methods on the iterator before using it).
   // @param skip_filters Disables loading/accessing the filter block
-  InternalIterator* NewIterator(const ReadOptions&,
-                                const SliceTransform* prefix_extractor,
-                                Arena* arena = nullptr,
-                                bool skip_filters = false) override;
+  InternalIterator* NewIterator(
+      const ReadOptions&, Arena* arena = nullptr,
+      bool skip_filters = false) override;
 
   InternalIterator* NewRangeTombstoneIterator(
       const ReadOptions& read_options) override;
 
   // @param skip_filters Disables loading/accessing the filter block
   Status Get(const ReadOptions& readOptions, const Slice& key,
-             GetContext* get_context, const SliceTransform* prefix_extractor,
-             bool skip_filters = false) override;
+             GetContext* get_context, bool skip_filters = false) override;
 
   // Pre-fetch the disk blocks that correspond to the key range specified by
   // (kbegin, kend). The call will return error status in the event of
@@ -140,8 +136,7 @@ class BlockBasedTable : public TableReader {
   size_t ApproximateMemoryUsage() const override;
 
   // convert SST file to a human readable form
-  Status DumpTable(WritableFile* out_file,
-                   const SliceTransform* prefix_extractor = nullptr) override;
+  Status DumpTable(WritableFile* out_file) override;
 
   Status VerifyChecksum() override;
 
@@ -258,13 +253,12 @@ class BlockBasedTable : public TableReader {
   // if `no_io == true`, we will not try to read filter/index from sst file
   // were they not present in cache yet.
   CachableEntry<FilterBlockReader> GetFilter(
-      const SliceTransform* prefix_extractor = nullptr,
       FilePrefetchBuffer* prefetch_buffer = nullptr, bool no_io = false,
       GetContext* get_context = nullptr) const;
   virtual CachableEntry<FilterBlockReader> GetFilter(
       FilePrefetchBuffer* prefetch_buffer, const BlockHandle& filter_blk_handle,
-      const bool is_a_filter_partition, bool no_io, GetContext* get_context,
-      const SliceTransform* prefix_extractor = nullptr) const;
+      const bool is_a_filter_partition, bool no_io,
+      GetContext* get_context) const;
 
   // Get the iterator from the index reader.
   // If input_iter is not set, return new Iterator
@@ -277,8 +271,7 @@ class BlockBasedTable : public TableReader {
   //  3. We disallowed any io to be performed, that is, read_options ==
   //     kBlockCacheTier
   InternalIterator* NewIndexIterator(
-      const ReadOptions& read_options, bool prefix_extractor_changed = false,
-      BlockIter* input_iter = nullptr,
+      const ReadOptions& read_options, BlockIter* input_iter = nullptr,
       CachableEntry<IndexReader>* index_entry = nullptr,
       GetContext* get_context = nullptr);
 
@@ -332,10 +325,9 @@ class BlockBasedTable : public TableReader {
       InternalIterator* preloaded_meta_index_iter = nullptr,
       const int level = -1);
 
-  bool FullFilterKeyMayMatch(
-      const ReadOptions& read_options, FilterBlockReader* filter,
-      const Slice& user_key, const bool no_io,
-      const SliceTransform* prefix_extractor = nullptr) const;
+  bool FullFilterKeyMayMatch(const ReadOptions& read_options,
+                             FilterBlockReader* filter, const Slice& user_key,
+                             const bool no_io) const;
 
   // Read the meta block from sst.
   static Status ReadMetaBlock(Rep* rep, FilePrefetchBuffer* prefetch_buffer,
@@ -345,10 +337,9 @@ class BlockBasedTable : public TableReader {
   Status VerifyChecksumInBlocks(InternalIterator* index_iter);
 
   // Create the filter from the filter block.
-  FilterBlockReader* ReadFilter(
-      FilePrefetchBuffer* prefetch_buffer, const BlockHandle& filter_handle,
-      const bool is_a_filter_partition,
-      const SliceTransform* prefix_extractor = nullptr) const;
+  FilterBlockReader* ReadFilter(FilePrefetchBuffer* prefetch_buffer,
+                                const BlockHandle& filter_handle,
+                                const bool is_a_filter_partition) const;
 
   static void SetupCacheKeyPrefix(Rep* rep, uint64_t file_size);
 
@@ -429,7 +420,7 @@ struct BlockBasedTable::Rep {
 
   const ImmutableCFOptions& ioptions;
   const EnvOptions& env_options;
-  const BlockBasedTableOptions& table_options;
+  const BlockBasedTableOptions table_options;
   const FilterPolicy* const filter_policy;
   const InternalKeyComparator& internal_comparator;
   Status status;
@@ -508,16 +499,14 @@ class BlockBasedTableIterator : public InternalIterator {
   BlockBasedTableIterator(BlockBasedTable* table,
                           const ReadOptions& read_options,
                           const InternalKeyComparator& icomp,
-                          InternalIterator* index_iter, bool check_filter,
-                          const SliceTransform* prefix_extractor)
+                          InternalIterator* index_iter, bool check_filter)
       : table_(table),
         read_options_(read_options),
         icomp_(icomp),
         index_iter_(index_iter),
         pinned_iters_mgr_(nullptr),
         block_iter_points_to_real_block_(false),
-        check_filter_(check_filter),
-        prefix_extractor_(prefix_extractor) {}
+        check_filter_(check_filter) {}
 
   ~BlockBasedTableIterator() { delete index_iter_; }
 
@@ -563,9 +552,8 @@ class BlockBasedTableIterator : public InternalIterator {
            block_iter_points_to_real_block_;
   }
 
-  bool CheckPrefixMayMatch(const Slice& ikey,
-                           const SliceTransform* prefix_extractor = nullptr) {
-    if (check_filter_ && !table_->PrefixMayMatch(ikey, prefix_extractor)) {
+  bool CheckPrefixMayMatch(const Slice& ikey) {
+    if (check_filter_ && !table_->PrefixMayMatch(ikey)) {
       // TODO remember the iterator is invalidated because of prefix
       // match. This can avoid the upper level file iterator to falsely
       // believe the position is the end of the SST file and move to
@@ -611,7 +599,6 @@ class BlockBasedTableIterator : public InternalIterator {
   bool check_filter_;
   // TODO use block offset instead
   std::string prev_index_value_;
-  const SliceTransform* prefix_extractor_;
 
   static const size_t kInitReadaheadSize = 8 * 1024;
   // Found that 256 KB readahead size provides the best performance, based on

table/cuckoo_table_reader.cc

@@ -141,7 +141,6 @@ CuckooTableReader::CuckooTableReader(
 
 Status CuckooTableReader::Get(const ReadOptions& /*readOptions*/,
                               const Slice& key, GetContext* get_context,
-                              const SliceTransform* /* prefix_extractor */,
                               bool /*skip_filters*/) {
   assert(key.size() == key_length_ + (is_last_level_ ? 8 : 0));
   Slice user_key = ExtractUserKey(key);
@@ -378,9 +377,7 @@ extern InternalIterator* NewErrorInternalIterator(const Status& status,
                                                   Arena* arena);
 
 InternalIterator* CuckooTableReader::NewIterator(
-    const ReadOptions& /*read_options*/,
-    const SliceTransform* /* prefix_extractor */, Arena* arena,
-    bool /*skip_filters*/) {
+    const ReadOptions& /*read_options*/, Arena* arena, bool /*skip_filters*/) {
   if (!status().ok()) {
     return NewErrorInternalIterator(
         Status::Corruption("CuckooTableReader status is not okay."), arena);

table/cuckoo_table_reader.h

@@ -42,14 +42,12 @@ class CuckooTableReader: public TableReader {
 
   Status status() const { return status_; }
 
-  Status Get(const ReadOptions& readOptions, const Slice& key,
-             GetContext* get_context, const SliceTransform* prefix_extractor,
-             bool skip_filters = false) override;
+  Status Get(const ReadOptions& read_options, const Slice& key,
+             GetContext* get_context, bool skip_filters = false) override;
 
-  InternalIterator* NewIterator(const ReadOptions&,
-                                const SliceTransform* prefix_extractor,
-                                Arena* arena = nullptr,
-                                bool skip_filters = false) override;
+  InternalIterator* NewIterator(
+      const ReadOptions&, Arena* arena = nullptr,
+      bool skip_filters = false) override;
   void Prepare(const Slice& target) override;
 
   // Report an approximation of how much memory has been used.

table/cuckoo_table_reader_test.cc

@@ -127,8 +127,7 @@ class CuckooReaderTest : public testing::Test {
       GetContext get_context(ucomp, nullptr, nullptr, nullptr,
                              GetContext::kNotFound, Slice(user_keys[i]), &value,
                              nullptr, nullptr, nullptr, nullptr);
-      ASSERT_OK(
-          reader.Get(ReadOptions(), Slice(keys[i]), &get_context, nullptr));
+      ASSERT_OK(reader.Get(ReadOptions(), Slice(keys[i]), &get_context));
       ASSERT_STREQ(values[i].c_str(), value.data());
     }
   }
@@ -150,8 +149,7 @@ class CuckooReaderTest : public testing::Test {
     CuckooTableReader reader(ioptions, std::move(file_reader), file_size, ucomp,
                              GetSliceHash);
     ASSERT_OK(reader.status());
-    InternalIterator* it =
-        reader.NewIterator(ReadOptions(), nullptr, nullptr, false);
+    InternalIterator* it = reader.NewIterator(ReadOptions(), nullptr);
     ASSERT_OK(it->status());
     ASSERT_TRUE(!it->Valid());
     it->SeekToFirst();
@@ -190,7 +188,7 @@ class CuckooReaderTest : public testing::Test {
     delete it;
 
     Arena arena;
-    it = reader.NewIterator(ReadOptions(), nullptr, &arena);
+    it = reader.NewIterator(ReadOptions(), &arena);
     ASSERT_OK(it->status());
     ASSERT_TRUE(!it->Valid());
     it->Seek(keys[num_items/2]);
@@ -339,8 +337,7 @@ TEST_F(CuckooReaderTest, WhenKeyNotFound) {
   GetContext get_context(ucmp, nullptr, nullptr, nullptr, GetContext::kNotFound,
                          Slice(not_found_key), &value, nullptr, nullptr,
                          nullptr, nullptr);
-  ASSERT_OK(
-      reader.Get(ReadOptions(), Slice(not_found_key), &get_context, nullptr));
+  ASSERT_OK(reader.Get(ReadOptions(), Slice(not_found_key), &get_context));
   ASSERT_TRUE(value.empty());
   ASSERT_OK(reader.status());
   // Search for a key with an independent hash value.
@@ -353,8 +350,7 @@ TEST_F(CuckooReaderTest, WhenKeyNotFound) {
   GetContext get_context2(ucmp, nullptr, nullptr, nullptr,
                           GetContext::kNotFound, Slice(not_found_key2), &value,
                           nullptr, nullptr, nullptr, nullptr);
-  ASSERT_OK(
-      reader.Get(ReadOptions(), Slice(not_found_key2), &get_context2, nullptr));
+  ASSERT_OK(reader.Get(ReadOptions(), Slice(not_found_key2), &get_context2));
   ASSERT_TRUE(value.empty());
   ASSERT_OK(reader.status());
 
@@ -369,8 +365,7 @@ TEST_F(CuckooReaderTest, WhenKeyNotFound) {
   GetContext get_context3(ucmp, nullptr, nullptr, nullptr,
                           GetContext::kNotFound, Slice(unused_key), &value,
                           nullptr, nullptr, nullptr, nullptr);
-  ASSERT_OK(
-      reader.Get(ReadOptions(), Slice(unused_key), &get_context3, nullptr));
+  ASSERT_OK(reader.Get(ReadOptions(), Slice(unused_key), &get_context3));
   ASSERT_TRUE(value.empty());
   ASSERT_OK(reader.status());
 }
@@ -448,7 +443,7 @@ void WriteFile(const std::vector<std::string>& keys,
   for (uint64_t i = 0; i < num; ++i) {
     value.Reset();
     value.clear();
-    ASSERT_OK(reader.Get(r_options, Slice(keys[i]), &get_context, nullptr));
+    ASSERT_OK(reader.Get(r_options, Slice(keys[i]), &get_context));
     ASSERT_TRUE(Slice(keys[i]) == Slice(&keys[i][0], 4));
   }
 }
@@ -501,13 +496,13 @@ void ReadKeys(uint64_t num, uint32_t batch_size) {
       }
       for (uint64_t j = i; j < i+batch_size && j < num; ++j) {
         reader.Get(r_options, Slice(reinterpret_cast<char*>(&keys[j]), 16),
-                   &get_context, nullptr);
+                   &get_context);
       }
     }
   } else {
     for (uint64_t i = 0; i < num; i++) {
       reader.Get(r_options, Slice(reinterpret_cast<char*>(&keys[i]), 16),
-                 &get_context, nullptr);
+                 &get_context);
     }
   }
   float time_per_op = (env->NowMicros() - start_time) * 1.0f / num;

table/filter_block.h

@@ -93,21 +93,16 @@ class FilterBlockReader {
    * built upon InternalKey and must be provided via const_ikey_ptr when running
    * queries.
    */
-  virtual bool KeyMayMatch(const Slice& key,
-                           const SliceTransform* prefix_extractor,
-                           uint64_t block_offset = kNotValid,
+  virtual bool KeyMayMatch(const Slice& key, uint64_t block_offset = kNotValid,
                            const bool no_io = false,
                            const Slice* const const_ikey_ptr = nullptr) = 0;
-
   /**
    * no_io and const_ikey_ptr here means the same as in KeyMayMatch
    */
   virtual bool PrefixMayMatch(const Slice& prefix,
-                              const SliceTransform* prefix_extractor,
                               uint64_t block_offset = kNotValid,
                               const bool no_io = false,
                               const Slice* const const_ikey_ptr = nullptr) = 0;
-
   virtual size_t ApproximateMemoryUsage() const = 0;
   virtual size_t size() const { return size_; }
   virtual Statistics* statistics() const { return statistics_; }
@@ -120,8 +115,7 @@ class FilterBlockReader {
     return error_msg;
   }
 
-  virtual void CacheDependencies(bool /*pin*/,
-                                 const SliceTransform* /*prefix_extractor*/) {}
+  virtual void CacheDependencies(bool /*pin*/) {}
 
  protected:
   bool whole_key_filtering_;

table/full_filter_block.cc

@@ -72,8 +72,14 @@ inline void FullFilterBlockBuilder::AddPrefix(const Slice& key) {
   }
 }
 
+void FullFilterBlockBuilder::Reset() {
+  last_whole_key_recorded_ = false;
+  last_prefix_recorded_ = false;
+}
+
 Slice FullFilterBlockBuilder::Finish(const BlockHandle& /*tmp*/,
                                      Status* status) {
+  Reset();
   // In this impl we ignore BlockHandle
   *status = Status::OK();
   if (num_added_ != 0) {
@@ -103,10 +109,9 @@ FullFilterBlockReader::FullFilterBlockReader(
   block_contents_ = std::move(contents);
 }
 
-bool FullFilterBlockReader::KeyMayMatch(
-    const Slice& key, const SliceTransform* /*prefix_extractor*/,
-    uint64_t block_offset, const bool /*no_io*/,
-    const Slice* const /*const_ikey_ptr*/) {
+bool FullFilterBlockReader::KeyMayMatch(const Slice& key, uint64_t block_offset,
+                                        const bool /*no_io*/,
+                                        const Slice* const /*const_ikey_ptr*/) {
 #ifdef NDEBUG
   (void)block_offset;
 #endif
@@ -118,8 +123,7 @@ bool FullFilterBlockReader::KeyMayMatch(
 }
 
 bool FullFilterBlockReader::PrefixMayMatch(
-    const Slice& prefix, const SliceTransform* /* prefix_extractor */,
-    uint64_t block_offset, const bool /*no_io*/,
+    const Slice& prefix, uint64_t block_offset, const bool /*no_io*/,
     const Slice* const /*const_ikey_ptr*/) {
 #ifdef NDEBUG
   (void)block_offset;

table/full_filter_block.h

@@ -52,6 +52,7 @@ class FullFilterBlockBuilder : public FilterBlockBuilder {
  protected:
   virtual void AddKey(const Slice& key);
   std::unique_ptr<FilterBitsBuilder> filter_bits_builder_;
+  virtual void Reset();
 
  private:
   // important: all of these might point to invalid addresses
@@ -96,15 +97,13 @@ class FullFilterBlockReader : public FilterBlockReader {
   ~FullFilterBlockReader() {}
 
   virtual bool IsBlockBased() override { return false; }
-
   virtual bool KeyMayMatch(
-      const Slice& key, const SliceTransform* prefix_extractor,
-      uint64_t block_offset = kNotValid, const bool no_io = false,
+      const Slice& key, uint64_t block_offset = kNotValid,
+      const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
-
   virtual bool PrefixMayMatch(
-      const Slice& prefix, const SliceTransform* prefix_extractor,
-      uint64_t block_offset = kNotValid, const bool no_io = false,
+      const Slice& prefix, uint64_t block_offset = kNotValid,
+      const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual size_t ApproximateMemoryUsage() const override;
 

table/full_filter_block_test.cc

@@ -113,7 +113,7 @@ TEST_F(PluginFullFilterBlockTest, PluginEmptyBuilder) {
       nullptr, true, block,
       table_options_.filter_policy->GetFilterBitsReader(block), nullptr);
   // Remain same symantic with blockbased filter
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr));
+  ASSERT_TRUE(reader.KeyMayMatch("foo"));
 }
 
 TEST_F(PluginFullFilterBlockTest, PluginSingleChunk) {
@@ -128,13 +128,13 @@ TEST_F(PluginFullFilterBlockTest, PluginSingleChunk) {
   FullFilterBlockReader reader(
       nullptr, true, block,
       table_options_.filter_policy->GetFilterBitsReader(block), nullptr);
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr));
-  ASSERT_TRUE(reader.KeyMayMatch("bar", nullptr));
-  ASSERT_TRUE(reader.KeyMayMatch("box", nullptr));
-  ASSERT_TRUE(reader.KeyMayMatch("hello", nullptr));
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr));
-  ASSERT_TRUE(!reader.KeyMayMatch("missing", nullptr));
-  ASSERT_TRUE(!reader.KeyMayMatch("other", nullptr));
+  ASSERT_TRUE(reader.KeyMayMatch("foo"));
+  ASSERT_TRUE(reader.KeyMayMatch("bar"));
+  ASSERT_TRUE(reader.KeyMayMatch("box"));
+  ASSERT_TRUE(reader.KeyMayMatch("hello"));
+  ASSERT_TRUE(reader.KeyMayMatch("foo"));
+  ASSERT_TRUE(!reader.KeyMayMatch("missing"));
+  ASSERT_TRUE(!reader.KeyMayMatch("other"));
 }
 
 class FullFilterBlockTest : public testing::Test {
@@ -158,7 +158,7 @@ TEST_F(FullFilterBlockTest, EmptyBuilder) {
       nullptr, true, block,
       table_options_.filter_policy->GetFilterBitsReader(block), nullptr);
   // Remain same symantic with blockbased filter
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr));
+  ASSERT_TRUE(reader.KeyMayMatch("foo"));
 }
 
 TEST_F(FullFilterBlockTest, DuplicateEntries) {
@@ -208,13 +208,13 @@ TEST_F(FullFilterBlockTest, SingleChunk) {
   FullFilterBlockReader reader(
       nullptr, true, block,
       table_options_.filter_policy->GetFilterBitsReader(block), nullptr);
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr));
-  ASSERT_TRUE(reader.KeyMayMatch("bar", nullptr));
-  ASSERT_TRUE(reader.KeyMayMatch("box", nullptr));
-  ASSERT_TRUE(reader.KeyMayMatch("hello", nullptr));
-  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr));
-  ASSERT_TRUE(!reader.KeyMayMatch("missing", nullptr));
-  ASSERT_TRUE(!reader.KeyMayMatch("other", nullptr));
+  ASSERT_TRUE(reader.KeyMayMatch("foo"));
+  ASSERT_TRUE(reader.KeyMayMatch("bar"));
+  ASSERT_TRUE(reader.KeyMayMatch("box"));
+  ASSERT_TRUE(reader.KeyMayMatch("hello"));
+  ASSERT_TRUE(reader.KeyMayMatch("foo"));
+  ASSERT_TRUE(!reader.KeyMayMatch("missing"));
+  ASSERT_TRUE(!reader.KeyMayMatch("other"));
 }
 
 }  // namespace rocksdb

table/get_context.cc

@@ -200,7 +200,7 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
           merge_context_->PushOperand(value, false);
         }
         if (merge_operator_ != nullptr &&
-            merge_operator_->ShouldMerge(merge_context_->GetOperands())) {
+            merge_operator_->ShouldMerge(merge_context_->GetOperandsDirectionBackward())) {
           state_ = kFound;
           if (LIKELY(pinnable_val_ != nullptr)) {
             Status merge_status = MergeHelper::TimedFullMerge(

table/merging_iterator.cc

@@ -228,28 +228,12 @@ class MergingIterator : public InternalIterator {
       Slice target = key();
       for (auto& child : children_) {
         if (&child != current_) {
-          if (!prefix_seek_mode_) {
-            child.Seek(target);
-            if (child.Valid()) {
-              // Child is at first entry >= key().  Step back one to be < key()
-              TEST_SYNC_POINT_CALLBACK("MergeIterator::Prev:BeforePrev",
-                                       &child);
-              assert(child.status().ok());
-              child.Prev();
-            } else {
-              // Child has no entries >= key().  Position at last entry.
-              TEST_SYNC_POINT("MergeIterator::Prev:BeforeSeekToLast");
-              considerStatus(child.status());
-              child.SeekToLast();
-            }
+          child.SeekForPrev(target);
+          TEST_SYNC_POINT_CALLBACK("MergeIterator::Prev:BeforePrev", &child);
+          considerStatus(child.status());
+          if (child.Valid() && comparator_->Equal(target, child.key())) {
+            child.Prev();
             considerStatus(child.status());
-          } else {
-            child.SeekForPrev(target);
-            considerStatus(child.status());
-            if (child.Valid() && comparator_->Equal(target, child.key())) {
-              child.Prev();
-              considerStatus(child.status());
-            }
           }
         }
         if (child.Valid()) {

table/mock_table.cc

@@ -26,16 +26,14 @@ stl_wrappers::KVMap MakeMockFile(
   return stl_wrappers::KVMap(l, stl_wrappers::LessOfComparator(&icmp_));
 }
 
-InternalIterator* MockTableReader::NewIterator(
-    const ReadOptions&, const SliceTransform* /* prefix_extractor */,
-    Arena* /*arena*/, bool /*skip_filters*/) {
+InternalIterator* MockTableReader::NewIterator(const ReadOptions&,
+                                               Arena* /*arena*/,
+                                               bool /*skip_filters*/) {
   return new MockTableIterator(table_);
 }
 
 Status MockTableReader::Get(const ReadOptions&, const Slice& key,
-                            GetContext* get_context,
-                            const SliceTransform* /*prefix_extractor*/,
-                            bool /*skip_filters*/) {
+                            GetContext* get_context, bool /*skip_filters*/) {
   std::unique_ptr<MockTableIterator> iter(new MockTableIterator(table_));
   for (iter->Seek(key); iter->Valid(); iter->Next()) {
     ParsedInternalKey parsed_key;

table/mock_table.h

@@ -39,12 +39,10 @@ class MockTableReader : public TableReader {
   explicit MockTableReader(const stl_wrappers::KVMap& table) : table_(table) {}
 
   InternalIterator* NewIterator(const ReadOptions&,
-                                const SliceTransform* prefix_extractor,
-                                Arena* arena = nullptr,
+                                Arena* arena,
                                 bool skip_filters = false) override;
 
-  Status Get(const ReadOptions& readOptions, const Slice& key,
-             GetContext* get_context, const SliceTransform* prefix_extractor,
+  Status Get(const ReadOptions&, const Slice& key, GetContext* get_context,
              bool skip_filters = false) override;
 
   uint64_t ApproximateOffsetOf(const Slice& /*key*/) override { return 0; }

table/partitioned_filter_block.cc

@@ -48,6 +48,7 @@ void PartitionedFilterBlockBuilder::MaybeCutAFilterBlock() {
   std::string& index_key = p_index_builder_->GetPartitionKey();
   filters.push_back({index_key, filter});
   filters_in_partition_ = 0;
+  Reset();
 }
 
 void PartitionedFilterBlockBuilder::AddKey(const Slice& key) {
@@ -130,8 +131,7 @@ PartitionedFilterBlockReader::~PartitionedFilterBlockReader() {
 }
 
 bool PartitionedFilterBlockReader::KeyMayMatch(
-    const Slice& key, const SliceTransform* prefix_extractor,
-    uint64_t block_offset, const bool no_io,
+    const Slice& key, uint64_t block_offset, const bool no_io,
     const Slice* const const_ikey_ptr) {
   assert(const_ikey_ptr != nullptr);
   assert(block_offset == kNotValid);
@@ -146,14 +146,12 @@ bool PartitionedFilterBlockReader::KeyMayMatch(
     return false;
   }
   bool cached = false;
-  auto filter_partition =
-      GetFilterPartition(nullptr /* prefetch_buffer */, &filter_handle, no_io,
-                         &cached, prefix_extractor);
+  auto filter_partition = GetFilterPartition(nullptr /* prefetch_buffer */,
+                                             &filter_handle, no_io, &cached);
   if (UNLIKELY(!filter_partition.value)) {
     return true;
   }
-  auto res = filter_partition.value->KeyMayMatch(key, prefix_extractor,
-                                                 block_offset, no_io);
+  auto res = filter_partition.value->KeyMayMatch(key, block_offset, no_io);
   if (cached) {
     return res;
   }
@@ -166,8 +164,7 @@ bool PartitionedFilterBlockReader::KeyMayMatch(
 }
 
 bool PartitionedFilterBlockReader::PrefixMayMatch(
-    const Slice& prefix, const SliceTransform* prefix_extractor,
-    uint64_t block_offset, const bool no_io,
+    const Slice& prefix, uint64_t block_offset, const bool no_io,
     const Slice* const const_ikey_ptr) {
 #ifdef NDEBUG
   (void)block_offset;
@@ -185,14 +182,12 @@ bool PartitionedFilterBlockReader::PrefixMayMatch(
     return false;
   }
   bool cached = false;
-  auto filter_partition =
-      GetFilterPartition(nullptr /* prefetch_buffer */, &filter_handle, no_io,
-                         &cached, prefix_extractor);
+  auto filter_partition = GetFilterPartition(nullptr /* prefetch_buffer */,
+                                             &filter_handle, no_io, &cached);
   if (UNLIKELY(!filter_partition.value)) {
     return true;
   }
-  auto res = filter_partition.value->PrefixMayMatch(prefix, prefix_extractor,
-                                                    kNotValid, no_io);
+  auto res = filter_partition.value->PrefixMayMatch(prefix, kNotValid, no_io);
   if (cached) {
     return res;
   }
@@ -220,7 +215,7 @@ Slice PartitionedFilterBlockReader::GetFilterPartitionHandle(
 BlockBasedTable::CachableEntry<FilterBlockReader>
 PartitionedFilterBlockReader::GetFilterPartition(
     FilePrefetchBuffer* prefetch_buffer, Slice* handle_value, const bool no_io,
-    bool* cached, const SliceTransform* prefix_extractor) {
+    bool* cached) {
   BlockHandle fltr_blk_handle;
   auto s = fltr_blk_handle.DecodeFrom(handle_value);
   assert(s.ok());
@@ -243,10 +238,10 @@ PartitionedFilterBlockReader::GetFilterPartition(
     }
     return table_->GetFilter(/*prefetch_buffer*/ nullptr, fltr_blk_handle,
                              is_a_filter_partition, no_io,
-                             /* get_context */ nullptr, prefix_extractor);
+                             /* get_context */ nullptr);
   } else {
     auto filter = table_->ReadFilter(prefetch_buffer, fltr_blk_handle,
-                                     is_a_filter_partition, prefix_extractor);
+                                     is_a_filter_partition);
     return {filter, nullptr};
   }
 }
@@ -263,8 +258,7 @@ void ReleaseFilterCachedEntry(void* arg, void* h) {
 }
 
 // TODO(myabandeh): merge this with the same function in IndexReader
-void PartitionedFilterBlockReader::CacheDependencies(
-    bool pin, const SliceTransform* prefix_extractor) {
+void PartitionedFilterBlockReader::CacheDependencies(bool pin) {
   // Before read partitions, prefetch them to avoid lots of IOs
   auto rep = table_->rep_;
   BlockIter biter;
@@ -315,9 +309,9 @@ void PartitionedFilterBlockReader::CacheDependencies(
 
     const bool no_io = true;
     const bool is_a_filter_partition = true;
-    auto filter = table_->GetFilter(
-        prefetch_buffer.get(), handle, is_a_filter_partition, !no_io,
-        /* get_context */ nullptr, prefix_extractor);
+    auto filter = table_->GetFilter(prefetch_buffer.get(), handle,
+                                    is_a_filter_partition, !no_io,
+                                    /* get_context */ nullptr);
     if (LIKELY(filter.IsSet())) {
       if (pin) {
         filter_map_[handle.offset()] = std::move(filter);

table/partitioned_filter_block.h

@@ -79,12 +79,12 @@ class PartitionedFilterBlockReader : public FilterBlockReader,
 
   virtual bool IsBlockBased() override { return false; }
   virtual bool KeyMayMatch(
-      const Slice& key, const SliceTransform* prefix_extractor,
-      uint64_t block_offset = kNotValid, const bool no_io = false,
+      const Slice& key, uint64_t block_offset = kNotValid,
+      const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual bool PrefixMayMatch(
-      const Slice& prefix, const SliceTransform* prefix_extractor,
-      uint64_t block_offset = kNotValid, const bool no_io = false,
+      const Slice& prefix, uint64_t block_offset = kNotValid,
+      const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual size_t ApproximateMemoryUsage() const override;
 
@@ -92,9 +92,8 @@ class PartitionedFilterBlockReader : public FilterBlockReader,
   Slice GetFilterPartitionHandle(const Slice& entry);
   BlockBasedTable::CachableEntry<FilterBlockReader> GetFilterPartition(
       FilePrefetchBuffer* prefetch_buffer, Slice* handle, const bool no_io,
-      bool* cached, const SliceTransform* prefix_extractor = nullptr);
-  virtual void CacheDependencies(
-      bool bin, const SliceTransform* prefix_extractor) override;
+      bool* cached);
+  virtual void CacheDependencies(bool pin) override;
 
   const SliceTransform* prefix_extractor_;
   std::unique_ptr<Block> idx_on_fltr_blk_;

table/partitioned_filter_block_test.cc

@@ -29,8 +29,8 @@ class MockedBlockBasedTable : public BlockBasedTable {
 
   virtual CachableEntry<FilterBlockReader> GetFilter(
       FilePrefetchBuffer*, const BlockHandle& filter_blk_handle,
-      const bool /* unused */, bool /* unused */, GetContext* /* unused */,
-      const SliceTransform* /* unused */) const override {
+      const bool /* unused */, bool /* unused */,
+      GetContext* /* unused */) const override {
     Slice slice = slices[filter_blk_handle.offset()];
     auto obj = new FullFilterBlockReader(
         nullptr, true, BlockContents(slice, false, kNoCompression),
@@ -121,7 +121,6 @@ class PartitionedFilterBlockTest : public testing::Test {
     } while (status.IsIncomplete());
     const Options options;
     const ImmutableCFOptions ioptions(options);
-    const MutableCFOptions moptions(options);
     const EnvOptions env_options;
     table.reset(new MockedBlockBasedTable(new BlockBasedTable::Rep(
         ioptions, env_options, table_options_, icomp, false)));
@@ -139,27 +138,23 @@ class PartitionedFilterBlockTest : public testing::Test {
     for (auto key : keys) {
       auto ikey = InternalKey(key, 0, ValueType::kTypeValue);
       const Slice ikey_slice = Slice(*ikey.rep());
-      ASSERT_TRUE(
-          reader->KeyMayMatch(key, nullptr, kNotValid, !no_io, &ikey_slice));
+      ASSERT_TRUE(reader->KeyMayMatch(key, kNotValid, !no_io, &ikey_slice));
     }
     {
       // querying a key twice
       auto ikey = InternalKey(keys[0], 0, ValueType::kTypeValue);
       const Slice ikey_slice = Slice(*ikey.rep());
-      ASSERT_TRUE(reader->KeyMayMatch(keys[0], nullptr, kNotValid, !no_io,
-                                      &ikey_slice));
+      ASSERT_TRUE(reader->KeyMayMatch(keys[0], kNotValid, !no_io, &ikey_slice));
     }
     // querying missing keys
     for (auto key : missing_keys) {
       auto ikey = InternalKey(key, 0, ValueType::kTypeValue);
       const Slice ikey_slice = Slice(*ikey.rep());
       if (empty) {
-        ASSERT_TRUE(
-            reader->KeyMayMatch(key, nullptr, kNotValid, !no_io, &ikey_slice));
+        ASSERT_TRUE(reader->KeyMayMatch(key, kNotValid, !no_io, &ikey_slice));
       } else {
         // assuming a good hash function
-        ASSERT_FALSE(
-            reader->KeyMayMatch(key, nullptr, kNotValid, !no_io, &ikey_slice));
+        ASSERT_FALSE(reader->KeyMayMatch(key, kNotValid, !no_io, &ikey_slice));
       }
     }
   }

table/plain_table_builder.cc

@@ -57,7 +57,7 @@ extern const uint64_t kPlainTableMagicNumber = 0x8242229663bf9564ull;
 extern const uint64_t kLegacyPlainTableMagicNumber = 0x4f3418eb7a8f13b8ull;
 
 PlainTableBuilder::PlainTableBuilder(
-    const ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
+    const ImmutableCFOptions& ioptions,
     const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
         int_tbl_prop_collector_factories,
     uint32_t column_family_id, WritableFileWriter* file, uint32_t user_key_len,
@@ -66,21 +66,20 @@ PlainTableBuilder::PlainTableBuilder(
     uint32_t num_probes, size_t huge_page_tlb_size, double hash_table_ratio,
     bool store_index_in_file)
     : ioptions_(ioptions),
-      moptions_(moptions),
       bloom_block_(num_probes),
       file_(file),
       bloom_bits_per_key_(bloom_bits_per_key),
       huge_page_tlb_size_(huge_page_tlb_size),
-      encoder_(encoding_type, user_key_len, moptions.prefix_extractor.get(),
+      encoder_(encoding_type, user_key_len, ioptions.prefix_extractor,
                index_sparseness),
       store_index_in_file_(store_index_in_file),
-      prefix_extractor_(moptions.prefix_extractor.get()) {
+      prefix_extractor_(ioptions.prefix_extractor) {
   // Build index block and save it in the file if hash_table_ratio > 0
   if (store_index_in_file_) {
     assert(hash_table_ratio > 0 || IsTotalOrderMode());
-    index_builder_.reset(new PlainTableIndexBuilder(
-        &arena_, ioptions, moptions.prefix_extractor.get(), index_sparseness,
-        hash_table_ratio, huge_page_tlb_size_));
+    index_builder_.reset(
+        new PlainTableIndexBuilder(&arena_, ioptions, index_sparseness,
+                                   hash_table_ratio, huge_page_tlb_size_));
     properties_.user_collected_properties
         [PlainTablePropertyNames::kBloomVersion] = "1";  // For future use
   }
@@ -97,8 +96,8 @@ PlainTableBuilder::PlainTableBuilder(
   properties_.format_version = (encoding_type == kPlain) ? 0 : 1;
   properties_.column_family_id = column_family_id;
   properties_.column_family_name = column_family_name;
-  properties_.prefix_extractor_name = moptions_.prefix_extractor != nullptr
-                                          ? moptions_.prefix_extractor->Name()
+  properties_.prefix_extractor_name = ioptions_.prefix_extractor != nullptr
+                                          ? ioptions_.prefix_extractor->Name()
                                           : "nullptr";
 
   std::string val;
@@ -132,11 +131,11 @@ void PlainTableBuilder::Add(const Slice& key, const Slice& value) {
 
   // Store key hash
   if (store_index_in_file_) {
-    if (moptions_.prefix_extractor == nullptr) {
+    if (ioptions_.prefix_extractor == nullptr) {
       keys_or_prefixes_hashes_.push_back(GetSliceHash(internal_key.user_key));
     } else {
       Slice prefix =
-          moptions_.prefix_extractor->Transform(internal_key.user_key);
+          ioptions_.prefix_extractor->Transform(internal_key.user_key);
       keys_or_prefixes_hashes_.push_back(GetSliceHash(prefix));
     }
   }

table/plain_table_builder.h

@@ -32,7 +32,7 @@ class PlainTableBuilder: public TableBuilder {
   // will be part of level specified by 'level'.  A value of -1 means
   // that the caller does not know which level the output file will reside.
   PlainTableBuilder(
-      const ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
+      const ImmutableCFOptions& ioptions,
       const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
           int_tbl_prop_collector_factories,
       uint32_t column_family_id, WritableFileWriter* file,
@@ -79,7 +79,6 @@ class PlainTableBuilder: public TableBuilder {
  private:
   Arena arena_;
   const ImmutableCFOptions& ioptions_;
-  const MutableCFOptions& moptions_;
   std::vector<std::unique_ptr<IntTblPropCollector>>
       table_properties_collectors_;
 

table/plain_table_factory.cc

@@ -27,7 +27,7 @@ Status PlainTableFactory::NewTableReader(
       table_reader_options.internal_comparator, std::move(file), file_size,
       table, table_options_.bloom_bits_per_key, table_options_.hash_table_ratio,
       table_options_.index_sparseness, table_options_.huge_page_tlb_size,
-      table_options_.full_scan_mode, table_reader_options.prefix_extractor);
+      table_options_.full_scan_mode);
 }
 
 TableBuilder* PlainTableFactory::NewTableBuilder(
@@ -38,7 +38,7 @@ TableBuilder* PlainTableFactory::NewTableBuilder(
   // tables
   //
   return new PlainTableBuilder(
-      table_builder_options.ioptions, table_builder_options.moptions,
+      table_builder_options.ioptions,
       table_builder_options.int_tbl_prop_collector_factories, column_family_id,
       file, table_options_.user_key_len, table_options_.encoding_type,
       table_options_.index_sparseness, table_options_.bloom_bits_per_key,

table/plain_table_index.h

@@ -112,7 +112,6 @@ class PlainTableIndex {
 class PlainTableIndexBuilder {
  public:
   PlainTableIndexBuilder(Arena* arena, const ImmutableCFOptions& ioptions,
-                         const SliceTransform* prefix_extractor,
                          size_t index_sparseness, double hash_table_ratio,
                          size_t huge_page_tlb_size)
       : arena_(arena),
@@ -126,7 +125,7 @@ class PlainTableIndexBuilder {
         index_sparseness_(index_sparseness),
         index_size_(0),
         sub_index_size_(0),
-        prefix_extractor_(prefix_extractor),
+        prefix_extractor_(ioptions.prefix_extractor),
         hash_table_ratio_(hash_table_ratio),
         huge_page_tlb_size_(huge_page_tlb_size) {}
 

table/plain_table_reader.cc

@@ -97,13 +97,12 @@ PlainTableReader::PlainTableReader(const ImmutableCFOptions& ioptions,
                                    const InternalKeyComparator& icomparator,
                                    EncodingType encoding_type,
                                    uint64_t file_size,
-                                   const TableProperties* table_properties,
-                                   const SliceTransform* prefix_extractor)
+                                   const TableProperties* table_properties)
     : internal_comparator_(icomparator),
       encoding_type_(encoding_type),
       full_scan_mode_(false),
       user_key_len_(static_cast<uint32_t>(table_properties->fixed_key_len)),
-      prefix_extractor_(prefix_extractor),
+      prefix_extractor_(ioptions.prefix_extractor),
       enable_bloom_(false),
       bloom_(6, nullptr),
       file_info_(std::move(file), storage_options,
@@ -115,13 +114,15 @@ PlainTableReader::PlainTableReader(const ImmutableCFOptions& ioptions,
 PlainTableReader::~PlainTableReader() {
 }
 
-Status PlainTableReader::Open(
-    const ImmutableCFOptions& ioptions, const EnvOptions& env_options,
-    const InternalKeyComparator& internal_comparator,
-    unique_ptr<RandomAccessFileReader>&& file, uint64_t file_size,
-    unique_ptr<TableReader>* table_reader, const int bloom_bits_per_key,
-    double hash_table_ratio, size_t index_sparseness, size_t huge_page_tlb_size,
-    bool full_scan_mode, const SliceTransform* prefix_extractor) {
+Status PlainTableReader::Open(const ImmutableCFOptions& ioptions,
+                              const EnvOptions& env_options,
+                              const InternalKeyComparator& internal_comparator,
+                              unique_ptr<RandomAccessFileReader>&& file,
+                              uint64_t file_size,
+                              unique_ptr<TableReader>* table_reader,
+                              const int bloom_bits_per_key,
+                              double hash_table_ratio, size_t index_sparseness,
+                              size_t huge_page_tlb_size, bool full_scan_mode) {
   if (file_size > PlainTableIndex::kMaxFileSize) {
     return Status::NotSupported("File is too large for PlainTableReader!");
   }
@@ -140,12 +141,12 @@ Status PlainTableReader::Open(
   if (!full_scan_mode &&
       !prefix_extractor_in_file.empty() /* old version sst file*/
       && prefix_extractor_in_file != "nullptr") {
-    if (!prefix_extractor) {
+    if (!ioptions.prefix_extractor) {
       return Status::InvalidArgument(
           "Prefix extractor is missing when opening a PlainTable built "
           "using a prefix extractor");
-    } else if (prefix_extractor_in_file.compare(prefix_extractor->Name()) !=
-               0) {
+    } else if (prefix_extractor_in_file.compare(
+                   ioptions.prefix_extractor->Name()) != 0) {
       return Status::InvalidArgument(
           "Prefix extractor given doesn't match the one used to build "
           "PlainTable");
@@ -162,7 +163,7 @@ Status PlainTableReader::Open(
 
   std::unique_ptr<PlainTableReader> new_reader(new PlainTableReader(
       ioptions, std::move(file), env_options, internal_comparator,
-      encoding_type, file_size, props, prefix_extractor));
+      encoding_type, file_size, props));
 
   s = new_reader->MmapDataIfNeeded();
   if (!s.ok()) {
@@ -188,9 +189,9 @@ Status PlainTableReader::Open(
 void PlainTableReader::SetupForCompaction() {
 }
 
-InternalIterator* PlainTableReader::NewIterator(
-    const ReadOptions& options, const SliceTransform* /* prefix_extractor */,
-    Arena* arena, bool /*skip_filters*/) {
+InternalIterator* PlainTableReader::NewIterator(const ReadOptions& options,
+                                                Arena* arena,
+                                                bool /*skip_filters*/) {
   bool use_prefix_seek = !IsTotalOrderMode() && !options.total_order_seek;
   if (arena == nullptr) {
     return new PlainTableIterator(this, use_prefix_seek);
@@ -209,7 +210,7 @@ Status PlainTableReader::PopulateIndexRecordList(
   bool is_first_record = true;
   Slice key_prefix_slice;
   PlainTableKeyDecoder decoder(&file_info_, encoding_type_, user_key_len_,
-                               prefix_extractor_);
+                               ioptions_.prefix_extractor);
   while (pos < file_info_.data_end_offset) {
     uint32_t key_offset = pos;
     ParsedInternalKey key;
@@ -329,8 +330,9 @@ Status PlainTableReader::PopulateIndex(TableProperties* props,
     index_block = nullptr;
   }
 
-  if ((prefix_extractor_ == nullptr) && (hash_table_ratio != 0)) {
-    // moptions.prefix_extractor is requried for a hash-based look-up.
+  if ((ioptions_.prefix_extractor == nullptr) &&
+      (hash_table_ratio != 0)) {
+    // ioptions.prefix_extractor is requried for a hash-based look-up.
     return Status::NotSupported(
         "PlainTable requires a prefix extractor enable prefix hash mode.");
   }
@@ -375,9 +377,8 @@ Status PlainTableReader::PopulateIndex(TableProperties* props,
     bloom_bits_per_key = 0;
   }
 
-  PlainTableIndexBuilder index_builder(&arena_, ioptions_, prefix_extractor_,
-                                       index_sparseness, hash_table_ratio,
-                                       huge_page_tlb_size);
+  PlainTableIndexBuilder index_builder(&arena_, ioptions_, index_sparseness,
+                                       hash_table_ratio, huge_page_tlb_size);
 
   std::vector<uint32_t> prefix_hashes;
   if (!index_in_file) {
@@ -537,9 +538,7 @@ void PlainTableReader::Prepare(const Slice& target) {
 }
 
 Status PlainTableReader::Get(const ReadOptions& /*ro*/, const Slice& target,
-                             GetContext* get_context,
-                             const SliceTransform* /* prefix_extractor */,
-                             bool /*skip_filters*/) {
+                             GetContext* get_context, bool /*skip_filters*/) {
   // Check bloom filter first.
   Slice prefix_slice;
   uint32_t prefix_hash;
@@ -566,7 +565,7 @@ Status PlainTableReader::Get(const ReadOptions& /*ro*/, const Slice& target,
   uint32_t offset;
   bool prefix_match;
   PlainTableKeyDecoder decoder(&file_info_, encoding_type_, user_key_len_,
-                               prefix_extractor_);
+                               ioptions_.prefix_extractor);
   Status s = GetOffset(&decoder, target, prefix_slice, prefix_hash,
                        prefix_match, &offset);
 

table/plain_table_reader.h

@@ -76,18 +76,15 @@ class PlainTableReader: public TableReader {
                      uint64_t file_size, unique_ptr<TableReader>* table,
                      const int bloom_bits_per_key, double hash_table_ratio,
                      size_t index_sparseness, size_t huge_page_tlb_size,
-                     bool full_scan_mode,
-                     const SliceTransform* prefix_extractor = nullptr);
+                     bool full_scan_mode);
 
   InternalIterator* NewIterator(const ReadOptions&,
-                                const SliceTransform* prefix_extractor,
                                 Arena* arena = nullptr,
                                 bool skip_filters = false) override;
 
   void Prepare(const Slice& target) override;
 
-  Status Get(const ReadOptions& readOptions, const Slice& key,
-             GetContext* get_context, const SliceTransform* prefix_extractor,
+  Status Get(const ReadOptions&, const Slice& key, GetContext* get_context,
              bool skip_filters = false) override;
 
   uint64_t ApproximateOffsetOf(const Slice& key) override;
@@ -108,8 +105,7 @@ class PlainTableReader: public TableReader {
                    const EnvOptions& env_options,
                    const InternalKeyComparator& internal_comparator,
                    EncodingType encoding_type, uint64_t file_size,
-                   const TableProperties* table_properties,
-                   const SliceTransform* prefix_extractor);
+                   const TableProperties* table_properties);
   virtual ~PlainTableReader();
 
  protected:

table/sst_file_writer.cc

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

table/table_builder.h

@@ -27,19 +27,16 @@ class Status;
 struct TableReaderOptions {
   // @param skip_filters Disables loading/accessing the filter block
   TableReaderOptions(const ImmutableCFOptions& _ioptions,
-                     const SliceTransform* _prefix_extractor,
                      const EnvOptions& _env_options,
                      const InternalKeyComparator& _internal_comparator,
                      bool _skip_filters = false, int _level = -1)
       : ioptions(_ioptions),
-        prefix_extractor(_prefix_extractor),
         env_options(_env_options),
         internal_comparator(_internal_comparator),
         skip_filters(_skip_filters),
         level(_level) {}
 
   const ImmutableCFOptions& ioptions;
-  const SliceTransform* prefix_extractor;
   const EnvOptions& env_options;
   const InternalKeyComparator& internal_comparator;
   // This is only used for BlockBasedTable (reader)
@@ -50,7 +47,7 @@ struct TableReaderOptions {
 
 struct TableBuilderOptions {
   TableBuilderOptions(
-      const ImmutableCFOptions& _ioptions, const MutableCFOptions& _moptions,
+      const ImmutableCFOptions& _ioptions,
       const InternalKeyComparator& _internal_comparator,
       const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
           _int_tbl_prop_collector_factories,
@@ -60,7 +57,6 @@ struct TableBuilderOptions {
       const std::string& _column_family_name, int _level,
       const uint64_t _creation_time = 0, const int64_t _oldest_key_time = 0)
       : ioptions(_ioptions),
-        moptions(_moptions),
         internal_comparator(_internal_comparator),
         int_tbl_prop_collector_factories(_int_tbl_prop_collector_factories),
         compression_type(_compression_type),
@@ -72,7 +68,6 @@ struct TableBuilderOptions {
         creation_time(_creation_time),
         oldest_key_time(_oldest_key_time) {}
   const ImmutableCFOptions& ioptions;
-  const MutableCFOptions& moptions;
   const InternalKeyComparator& internal_comparator;
   const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
       int_tbl_prop_collector_factories;

table/table_properties.cc

@@ -107,11 +107,6 @@ std::string TableProperties::ToString(
       filter_policy_name.empty() ? std::string("N/A") : filter_policy_name,
       prop_delim, kv_delim);
 
-  AppendProperty(result, "prefix extractor name",
-                 prefix_extractor_name.empty() ? std::string("N/A")
-                                               : prefix_extractor_name,
-                 prop_delim, kv_delim);
-
   AppendProperty(result, "column family ID",
                  column_family_id == rocksdb::TablePropertiesCollectorFactory::
                                          Context::kUnknownColumnFamily

table/table_reader.h

@@ -9,7 +9,6 @@
 
 #pragma once
 #include <memory>
-#include "rocksdb/slice_transform.h"
 #include "table/internal_iterator.h"
 
 namespace rocksdb {
@@ -40,7 +39,6 @@ class TableReader {
   // skip_filters: disables checking the bloom filters even if they exist. This
   //               option is effective only for block-based table format.
   virtual InternalIterator* NewIterator(const ReadOptions&,
-                                        const SliceTransform* prefix_extractor,
                                         Arena* arena = nullptr,
                                         bool skip_filters = false) = 0;
 
@@ -81,9 +79,7 @@ class TableReader {
   // skip_filters: disables checking the bloom filters even if they exist. This
   //               option is effective only for block-based table format.
   virtual Status Get(const ReadOptions& readOptions, const Slice& key,
-                     GetContext* get_context,
-                     const SliceTransform* prefix_extractor,
-                     bool skip_filters = false) = 0;
+                     GetContext* get_context, bool skip_filters = false) = 0;
 
   // Prefetch data corresponding to a give range of keys
   // Typically this functionality is required for table implementations that
@@ -98,8 +94,7 @@ class TableReader {
   }
 
   // convert db file to a human readable form
-  virtual Status DumpTable(WritableFile* /*out_file*/,
-                           const SliceTransform* /*prefix_extractor*/) {
+  virtual Status DumpTable(WritableFile* /*out_file*/) {
     return Status::NotSupported("DumpTable() not supported");
   }