Consolidate SliceTransform object ownership

Summary:
(1) Fix SanitizeOptions() to also check HashLinkList. The current
dynamic case just happens to work because the 2 classes have the same
layout.
(2) Do not delete SliceTransform object in HashSkipListFactory and
HashLinkListFactory destructor. Reason: SanitizeOptions() enforces
prefix_extractor and SliceTransform to be the same object when
Hash**Factory is used. This makes the behavior strange: when
Hash**Factory is used, prefix_extractor will be released by RocksDB. If
other memtable factory is used, prefix_extractor should be released by
user.

Test Plan: db_bench && make asan_check

Reviewers: haobo, igor, sdong

Reviewed By: igor

CC: leveldb, dhruba

Differential Revision: https://reviews.facebook.net/D16587

HISTORY.md

@@ -10,6 +10,8 @@
 * Added "virtual void WaitForJoin() = 0" in class Env
 * Removed BackupEngine::DeleteBackupsNewerThan() function
 * Added new option -- verify_checksums_in_compaction
+* Chagned Options.prefix_extractor from raw pointer to shared_ptr (take ownership)
+  Changed HashSkipListRepFactory and HashLinkListRepFactory constructor to not take SliceTransform object (use Options.prefix_extractor implicitly)
 
 ### New Features
 * If we find one truncated record at the end of the MANIFEST or WAL files,

db/c.cc

@@ -294,10 +294,10 @@ struct rocksdb_universal_compaction_options_t {
 };
 
 static bool SaveError(char** errptr, const Status& s) {
-  assert(errptr != NULL);
+  assert(errptr != nullptr);
   if (s.ok()) {
     return false;
-  } else if (*errptr == NULL) {
+  } else if (*errptr == nullptr) {
     *errptr = strdup(s.ToString().c_str());
   } else {
     // TODO(sanjay): Merge with existing error?
@@ -319,7 +319,7 @@ rocksdb_t* rocksdb_open(
     char** errptr) {
   DB* db;
   if (SaveError(errptr, DB::Open(options->rep, std::string(name), &db))) {
-    return NULL;
+    return nullptr;
   }
   rocksdb_t* result = new rocksdb_t;
   result->rep = db;
@@ -373,7 +373,7 @@ char* rocksdb_get(
     const char* key, size_t keylen,
     size_t* vallen,
     char** errptr) {
-  char* result = NULL;
+  char* result = nullptr;
   std::string tmp;
   Status s = db->rep->Get(options->rep, Slice(key, keylen), &tmp);
   if (s.ok()) {
@@ -418,7 +418,7 @@ char* rocksdb_property_value(
     // We use strdup() since we expect human readable output.
     return strdup(tmp.c_str());
   } else {
-    return NULL;
+    return nullptr;
   }
 }
 
@@ -456,9 +456,9 @@ void rocksdb_compact_range(
     const char* limit_key, size_t limit_key_len) {
   Slice a, b;
   db->rep->CompactRange(
-      // Pass NULL Slice if corresponding "const char*" is NULL
-      (start_key ? (a = Slice(start_key, start_key_len), &a) : NULL),
-      (limit_key ? (b = Slice(limit_key, limit_key_len), &b) : NULL));
+      // Pass nullptr Slice if corresponding "const char*" is nullptr
+      (start_key ? (a = Slice(start_key, start_key_len), &a) : nullptr),
+      (limit_key ? (b = Slice(limit_key, limit_key_len), &b) : nullptr));
 }
 
 void rocksdb_flush(
@@ -647,7 +647,7 @@ void rocksdb_options_set_paranoid_checks(
 }
 
 void rocksdb_options_set_env(rocksdb_options_t* opt, rocksdb_env_t* env) {
-  opt->rep.env = (env ? env->rep : NULL);
+  opt->rep.env = (env ? env->rep : nullptr);
 }
 
 void rocksdb_options_set_info_log(rocksdb_options_t* opt, rocksdb_logger_t* l) {
@@ -765,7 +765,7 @@ void rocksdb_options_set_compression_options(
 
 void rocksdb_options_set_prefix_extractor(
     rocksdb_options_t* opt, rocksdb_slicetransform_t* prefix_extractor) {
-  opt->rep.prefix_extractor = prefix_extractor;
+  opt->rep.prefix_extractor.reset(prefix_extractor);
 }
 
 void rocksdb_options_set_whole_key_filtering(
@@ -1087,8 +1087,8 @@ rocksdb_filterpolicy_t* rocksdb_filterpolicy_create_bloom(int bits_per_key) {
   };
   Wrapper* wrapper = new Wrapper;
   wrapper->rep_ = NewBloomFilterPolicy(bits_per_key);
-  wrapper->state_ = NULL;
-  wrapper->delete_filter_ = NULL;
+  wrapper->state_ = nullptr;
+  wrapper->delete_filter_ = nullptr;
   wrapper->destructor_ = &Wrapper::DoNothing;
   return wrapper;
 }
@@ -1154,7 +1154,7 @@ void rocksdb_readoptions_set_prefix_seek(
 void rocksdb_readoptions_set_snapshot(
     rocksdb_readoptions_t* opt,
     const rocksdb_snapshot_t* snap) {
-  opt->rep.snapshot = (snap ? snap->rep : NULL);
+  opt->rep.snapshot = (snap ? snap->rep : nullptr);
 }
 
 void rocksdb_readoptions_set_prefix(
@@ -1280,7 +1280,7 @@ rocksdb_slicetransform_t* rocksdb_slicetransform_create_fixed_prefix(size_t pref
   };
   Wrapper* wrapper = new Wrapper;
   wrapper->rep_ = rocksdb::NewFixedPrefixTransform(prefixLen);
-  wrapper->state_ = NULL;
+  wrapper->state_ = nullptr;
   wrapper->destructor_ = &Wrapper::DoNothing;
   return wrapper;
 }

db/db_bench.cc

@@ -1538,9 +1538,10 @@ class Benchmark {
     options.compaction_style = FLAGS_compaction_style_e;
     options.block_size = FLAGS_block_size;
     options.filter_policy = filter_policy_;
-    options.prefix_extractor =
-      (FLAGS_use_plain_table || FLAGS_use_prefix_blooms) ? prefix_extractor_
-                                                         : nullptr;
+    if (FLAGS_use_plain_table || FLAGS_use_prefix_blooms) {
+      options.prefix_extractor.reset(
+          NewFixedPrefixTransform(FLAGS_prefix_size));
+    }
     options.memtable_prefix_bloom_bits = FLAGS_memtable_bloom_bits;
     options.max_open_files = FLAGS_open_files;
     options.statistics = dbstats;
@@ -1564,7 +1565,6 @@ class Benchmark {
     switch (FLAGS_rep_factory) {
       case kPrefixHash:
         options.memtable_factory.reset(NewHashSkipListRepFactory(
-            prefix_extractor_,
             FLAGS_hash_bucket_count));
         break;
       case kSkipList:
@@ -1572,7 +1572,6 @@ class Benchmark {
         break;
       case kHashLinkedList:
         options.memtable_factory.reset(NewHashLinkListRepFactory(
-            prefix_extractor_,
             FLAGS_hash_bucket_count));
         break;
       case kVectorRep:

db/db_impl.cc

@@ -56,6 +56,7 @@
 #include "util/build_version.h"
 #include "util/coding.h"
 #include "util/hash_skiplist_rep.h"
+#include "util/hash_linklist_rep.h"
 #include "util/logging.h"
 #include "util/log_buffer.h"
 #include "util/mutexlock.h"
@@ -176,19 +177,16 @@ Options SanitizeOptions(const std::string& dbname,
     Log(result.info_log, "Compaction filter specified, ignore factory");
   }
   if (result.prefix_extractor) {
-    // If a prefix extractor has been supplied and a HashSkipListRepFactory is
-    // being used, make sure that the latter uses the former as its transform
-    // function.
-    auto factory = dynamic_cast<HashSkipListRepFactory*>(
-      result.memtable_factory.get());
-    if (factory &&
-        factory->GetTransform() != result.prefix_extractor) {
-      Log(result.info_log, "A prefix hash representation factory was supplied "
-          "whose prefix extractor does not match options.prefix_extractor. "
-          "Falling back to skip list representation factory");
+    Log(result.info_log, "prefix extractor %s in use.",
+        result.prefix_extractor->Name());
+  } else {
+    assert(result.memtable_factory);
+    Slice name = result.memtable_factory->Name();
+    if (name.compare("HashSkipListRepFactory") == 0 ||
+        name.compare("HashLinkListRepFactory") == 0) {
+      Log(result.info_log, "prefix extractor is not provided while using %s. "
+          "fallback to skiplist", name.ToString().c_str());
       result.memtable_factory = std::make_shared<SkipListFactory>();
-    } else if (factory) {
-      Log(result.info_log, "Prefix hash memtable rep is in use.");
     }
   }
 
@@ -3207,7 +3205,7 @@ Iterator* DBImpl::NewIterator(const ReadOptions& options) {
     // use extra wrapper to exclude any keys from the results which
     // don't begin with the prefix
     iter = new PrefixFilterIterator(iter, *options.prefix,
-                                    options_.prefix_extractor);
+                                    options_.prefix_extractor.get());
   }
   return iter;
 }

db/db_test.cc

@@ -259,8 +259,6 @@ class SpecialEnv : public EnvWrapper {
 class DBTest {
  private:
   const FilterPolicy* filter_policy_;
-  static std::unique_ptr<const SliceTransform> prefix_1_transform;
-  static std::unique_ptr<const SliceTransform> noop_transform;
 
  protected:
   // Sequence of option configurations to try
@@ -374,18 +372,18 @@ class DBTest {
     Options options;
     switch (option_config_) {
       case kHashSkipList:
-        options.memtable_factory.reset(
-            NewHashSkipListRepFactory(NewFixedPrefixTransform(1)));
+        options.prefix_extractor.reset(NewFixedPrefixTransform(1));
+        options.memtable_factory.reset(NewHashSkipListRepFactory());
         break;
       case kPlainTableFirstBytePrefix:
         options.table_factory.reset(new PlainTableFactory());
-        options.prefix_extractor = prefix_1_transform.get();
+        options.prefix_extractor.reset(NewFixedPrefixTransform(1));
         options.allow_mmap_reads = true;
         options.max_sequential_skip_in_iterations = 999999;
         break;
       case kPlainTableAllBytesPrefix:
         options.table_factory.reset(new PlainTableFactory());
-        options.prefix_extractor = noop_transform.get();
+        options.prefix_extractor.reset(NewNoopTransform());
         options.allow_mmap_reads = true;
         options.max_sequential_skip_in_iterations = 999999;
         break;
@@ -425,8 +423,8 @@ class DBTest {
         options.memtable_factory.reset(new VectorRepFactory(100));
         break;
       case kHashLinkList:
-      options.memtable_factory.reset(
-          NewHashLinkListRepFactory(NewFixedPrefixTransform(1), 4));
+        options.prefix_extractor.reset(NewFixedPrefixTransform(1));
+        options.memtable_factory.reset(NewHashLinkListRepFactory(4));
         break;
       case kUniversalCompaction:
         options.compaction_style = kCompactionStyleUniversal;
@@ -819,10 +817,6 @@ class DBTest {
   }
 
 };
-std::unique_ptr<const SliceTransform> DBTest::prefix_1_transform(
-    NewFixedPrefixTransform(1));
-std::unique_ptr<const SliceTransform> DBTest::noop_transform(
-    NewNoopTransform());
 
 static std::string Key(int i) {
   char buf[100];
@@ -5404,7 +5398,7 @@ TEST(DBTest, PrefixScan) {
   options.env = env_;
   options.no_block_cache = true;
   options.filter_policy = NewBloomFilterPolicy(10);
-  options.prefix_extractor = NewFixedPrefixTransform(8);
+  options.prefix_extractor.reset(NewFixedPrefixTransform(8));
   options.whole_key_filtering = false;
   options.disable_auto_compactions = true;
   options.max_background_compactions = 2;
@@ -5412,8 +5406,7 @@ TEST(DBTest, PrefixScan) {
   options.disable_seek_compaction = true;
   // Tricky: options.prefix_extractor will be released by
   // NewHashSkipListRepFactory after use.
-  options.memtable_factory.reset(
-      NewHashSkipListRepFactory(options.prefix_extractor));
+  options.memtable_factory.reset(NewHashSkipListRepFactory());
 
   // prefix specified, with blooms: 2 RAND I/Os
   // SeekToFirst
@@ -5609,14 +5602,12 @@ TEST(DBTest, TailingIteratorPrefixSeek) {
   read_options.tailing = true;
   read_options.prefix_seek = true;
 
-  auto prefix_extractor = NewFixedPrefixTransform(2);
-
   Options options = CurrentOptions();
   options.env = env_;
   options.create_if_missing = true;
   options.disable_auto_compactions = true;
-  options.prefix_extractor = prefix_extractor;
-  options.memtable_factory.reset(NewHashSkipListRepFactory(prefix_extractor));
+  options.prefix_extractor.reset(NewFixedPrefixTransform(2));
+  options.memtable_factory.reset(NewHashSkipListRepFactory());
   DestroyAndReopen(&options);
 
   std::unique_ptr<Iterator> iter(db_->NewIterator(read_options));

db/memtable.cc

@@ -32,7 +32,8 @@ MemTable::MemTable(const InternalKeyComparator& cmp, const Options& options)
     : comparator_(cmp),
       refs_(0),
       arena_(options.arena_block_size),
-      table_(options.memtable_factory->CreateMemTableRep(comparator_, &arena_)),
+      table_(options.memtable_factory->CreateMemTableRep(
+             comparator_, &arena_, options.prefix_extractor.get())),
       flush_in_progress_(false),
       flush_completed_(false),
       file_number_(0),
@@ -41,7 +42,7 @@ MemTable::MemTable(const InternalKeyComparator& cmp, const Options& options)
       mem_logfile_number_(0),
       locks_(options.inplace_update_support ? options.inplace_update_num_locks
                                             : 0),
-      prefix_extractor_(options.prefix_extractor) {
+      prefix_extractor_(options.prefix_extractor.get()) {
   if (prefix_extractor_ && options.memtable_prefix_bloom_bits > 0) {
     prefix_bloom_.reset(new DynamicBloom(options.memtable_prefix_bloom_bits,
                                          options.memtable_prefix_bloom_probes));

db/plain_table_db_test.cc

@@ -44,7 +44,6 @@ class PlainTableDBTest {
   DB* db_;
 
   Options last_options_;
-  static std::unique_ptr<const SliceTransform> prefix_transform;
 
  public:
   PlainTableDBTest() : env_(Env::Default()) {
@@ -66,7 +65,7 @@ class PlainTableDBTest {
   Options CurrentOptions() {
     Options options;
     options.table_factory.reset(NewPlainTableFactory(16, 2, 0.8, 3));
-    options.prefix_extractor = prefix_transform.get();
+    options.prefix_extractor.reset(NewFixedPrefixTransform(8));
     options.allow_mmap_reads = true;
     return options;
   }
@@ -173,9 +172,6 @@ class PlainTableDBTest {
   }
 };
 
-std::unique_ptr<const SliceTransform> PlainTableDBTest::prefix_transform(
-    NewFixedPrefixTransform(8));
-
 TEST(PlainTableDBTest, Empty) {
   ASSERT_TRUE(dbfull() != nullptr);
   ASSERT_EQ("NOT_FOUND", Get("0000000000000foo"));

db/prefix_test.cc

@@ -161,16 +161,15 @@ class PrefixTest {
     // skip some options
     option_config_++;
     if (option_config_ < kEnd) {
-      auto prefix_extractor = NewFixedPrefixTransform(8);
-      options.prefix_extractor = prefix_extractor;
+      options.prefix_extractor.reset(NewFixedPrefixTransform(8));
       switch(option_config_) {
         case kHashSkipList:
-          options.memtable_factory.reset(NewHashSkipListRepFactory(
-              options.prefix_extractor, bucket_count, FLAGS_skiplist_height));
+          options.memtable_factory.reset(
+              NewHashSkipListRepFactory(bucket_count, FLAGS_skiplist_height));
           return true;
         case kHashLinkList:
-          options.memtable_factory.reset(NewHashLinkListRepFactory(
-              options.prefix_extractor, bucket_count));
+          options.memtable_factory.reset(
+              NewHashLinkListRepFactory(bucket_count));
           return true;
         default:
           return false;

db/skiplist.h

@@ -158,7 +158,7 @@ class SkipList {
 
 // Implementation details follow
 template<typename Key, class Comparator>
-struct SkipList<Key,Comparator>::Node {
+struct SkipList<Key, Comparator>::Node {
   explicit Node(const Key& k) : key(k) { }
 
   Key const key;
@@ -194,43 +194,43 @@ struct SkipList<Key,Comparator>::Node {
 };
 
 template<typename Key, class Comparator>
-typename SkipList<Key,Comparator>::Node*
-SkipList<Key,Comparator>::NewNode(const Key& key, int height) {
+typename SkipList<Key, Comparator>::Node*
+SkipList<Key, Comparator>::NewNode(const Key& key, int height) {
   char* mem = arena_->AllocateAligned(
       sizeof(Node) + sizeof(port::AtomicPointer) * (height - 1));
   return new (mem) Node(key);
 }
 
 template<typename Key, class Comparator>
-inline SkipList<Key,Comparator>::Iterator::Iterator(const SkipList* list) {
+inline SkipList<Key, Comparator>::Iterator::Iterator(const SkipList* list) {
   SetList(list);
 }
 
 template<typename Key, class Comparator>
-inline void SkipList<Key,Comparator>::Iterator::SetList(const SkipList* list) {
+inline void SkipList<Key, Comparator>::Iterator::SetList(const SkipList* list) {
   list_ = list;
   node_ = nullptr;
 }
 
 template<typename Key, class Comparator>
-inline bool SkipList<Key,Comparator>::Iterator::Valid() const {
+inline bool SkipList<Key, Comparator>::Iterator::Valid() const {
   return node_ != nullptr;
 }
 
 template<typename Key, class Comparator>
-inline const Key& SkipList<Key,Comparator>::Iterator::key() const {
+inline const Key& SkipList<Key, Comparator>::Iterator::key() const {
   assert(Valid());
   return node_->key;
 }
 
 template<typename Key, class Comparator>
-inline void SkipList<Key,Comparator>::Iterator::Next() {
+inline void SkipList<Key, Comparator>::Iterator::Next() {
   assert(Valid());
   node_ = node_->Next(0);
 }
 
 template<typename Key, class Comparator>
-inline void SkipList<Key,Comparator>::Iterator::Prev() {
+inline void SkipList<Key, Comparator>::Iterator::Prev() {
   // Instead of using explicit "prev" links, we just search for the
   // last node that falls before key.
   assert(Valid());
@@ -241,17 +241,17 @@ inline void SkipList<Key,Comparator>::Iterator::Prev() {
 }
 
 template<typename Key, class Comparator>
-inline void SkipList<Key,Comparator>::Iterator::Seek(const Key& target) {
+inline void SkipList<Key, Comparator>::Iterator::Seek(const Key& target) {
   node_ = list_->FindGreaterOrEqual(target, nullptr);
 }
 
 template<typename Key, class Comparator>
-inline void SkipList<Key,Comparator>::Iterator::SeekToFirst() {
+inline void SkipList<Key, Comparator>::Iterator::SeekToFirst() {
   node_ = list_->head_->Next(0);
 }
 
 template<typename Key, class Comparator>
-inline void SkipList<Key,Comparator>::Iterator::SeekToLast() {
+inline void SkipList<Key, Comparator>::Iterator::SeekToLast() {
   node_ = list_->FindLast();
   if (node_ == list_->head_) {
     node_ = nullptr;
@@ -259,7 +259,7 @@ inline void SkipList<Key,Comparator>::Iterator::SeekToLast() {
 }
 
 template<typename Key, class Comparator>
-int SkipList<Key,Comparator>::RandomHeight() {
+int SkipList<Key, Comparator>::RandomHeight() {
   // Increase height with probability 1 in kBranching
   int height = 1;
   while (height < kMaxHeight_ && ((rnd_.Next() % kBranching_) == 0)) {
@@ -271,14 +271,14 @@ int SkipList<Key,Comparator>::RandomHeight() {
 }
 
 template<typename Key, class Comparator>
-bool SkipList<Key,Comparator>::KeyIsAfterNode(const Key& key, Node* n) const {
+bool SkipList<Key, Comparator>::KeyIsAfterNode(const Key& key, Node* n) const {
   // nullptr n is considered infinite
   return (n != nullptr) && (compare_(n->key, key) < 0);
 }
 
 template<typename Key, class Comparator>
-typename SkipList<Key,Comparator>::Node* SkipList<Key,Comparator>::FindGreaterOrEqual(const Key& key, Node** prev)
-    const {
+typename SkipList<Key, Comparator>::Node* SkipList<Key, Comparator>::
+  FindGreaterOrEqual(const Key& key, Node** prev) const {
   // Use prev as an optimization hint and fallback to slow path
   if (prev && !KeyIsAfterNode(key, prev[0]->Next(0))) {
     Node* x = prev[0];
@@ -315,8 +315,8 @@ typename SkipList<Key,Comparator>::Node* SkipList<Key,Comparator>::FindGreaterOr
 }
 
 template<typename Key, class Comparator>
-typename SkipList<Key,Comparator>::Node*
-SkipList<Key,Comparator>::FindLessThan(const Key& key) const {
+typename SkipList<Key, Comparator>::Node*
+SkipList<Key, Comparator>::FindLessThan(const Key& key) const {
   Node* x = head_;
   int level = GetMaxHeight() - 1;
   while (true) {
@@ -336,7 +336,7 @@ SkipList<Key,Comparator>::FindLessThan(const Key& key) const {
 }
 
 template<typename Key, class Comparator>
-typename SkipList<Key,Comparator>::Node* SkipList<Key,Comparator>::FindLast()
+typename SkipList<Key, Comparator>::Node* SkipList<Key, Comparator>::FindLast()
     const {
   Node* x = head_;
   int level = GetMaxHeight() - 1;
@@ -356,7 +356,7 @@ typename SkipList<Key,Comparator>::Node* SkipList<Key,Comparator>::FindLast()
 }
 
 template<typename Key, class Comparator>
-SkipList<Key,Comparator>::SkipList(Comparator cmp, Arena* arena,
+SkipList<Key, Comparator>::SkipList(const Comparator cmp, Arena* arena,
                                    int32_t max_height,
                                    int32_t branching_factor)
     : kMaxHeight_(max_height),
@@ -380,7 +380,7 @@ SkipList<Key,Comparator>::SkipList(Comparator cmp, Arena* arena,
 }
 
 template<typename Key, class Comparator>
-void SkipList<Key,Comparator>::Insert(const Key& key) {
+void SkipList<Key, Comparator>::Insert(const Key& key) {
   // TODO(opt): We can use a barrier-free variant of FindGreaterOrEqual()
   // here since Insert() is externally synchronized.
   Node* x = FindGreaterOrEqual(key, prev_);
@@ -417,7 +417,7 @@ void SkipList<Key,Comparator>::Insert(const Key& key) {
 }
 
 template<typename Key, class Comparator>
-bool SkipList<Key,Comparator>::Contains(const Key& key) const {
+bool SkipList<Key, Comparator>::Contains(const Key& key) const {
   Node* x = FindGreaterOrEqual(key, nullptr);
   if (x != nullptr && Equal(key, x->key)) {
     return true;

db/tailing_iter.cc

@@ -155,7 +155,7 @@ bool TailingIterator::IsCurrentVersion() const {
 }
 
 bool TailingIterator::IsSamePrefix(const Slice& target) const {
-  const SliceTransform* extractor = db_->options_.prefix_extractor;
+  const SliceTransform* extractor = db_->options_.prefix_extractor.get();
 
   assert(extractor);
   assert(is_prev_set_);

include/rocksdb/memtablerep.h

@@ -160,8 +160,8 @@ class MemTableRep {
 class MemTableRepFactory {
  public:
   virtual ~MemTableRepFactory() {}
-  virtual MemTableRep* CreateMemTableRep(MemTableRep::KeyComparator&,
-                                         Arena*) = 0;
+  virtual MemTableRep* CreateMemTableRep(const MemTableRep::KeyComparator&,
+      Arena*, const SliceTransform*) = 0;
   virtual const char* Name() const = 0;
 };
 
@@ -178,8 +178,9 @@ class VectorRepFactory : public MemTableRepFactory {
 
  public:
   explicit VectorRepFactory(size_t count = 0) : count_(count) { }
-  virtual MemTableRep* CreateMemTableRep(MemTableRep::KeyComparator&,
-                                         Arena*) override;
+  virtual MemTableRep* CreateMemTableRep(
+      const MemTableRep::KeyComparator&, Arena*,
+      const SliceTransform*) override;
   virtual const char* Name() const override {
     return "VectorRepFactory";
   }
@@ -188,8 +189,9 @@ class VectorRepFactory : public MemTableRepFactory {
 // This uses a skip list to store keys. It is the default.
 class SkipListFactory : public MemTableRepFactory {
  public:
-  virtual MemTableRep* CreateMemTableRep(MemTableRep::KeyComparator&,
-                                         Arena*) override;
+  virtual MemTableRep* CreateMemTableRep(
+      const MemTableRep::KeyComparator&, Arena*,
+      const SliceTransform*) override;
   virtual const char* Name() const override {
     return "SkipListFactory";
   }
@@ -202,8 +204,8 @@ class SkipListFactory : public MemTableRepFactory {
 // skiplist_branching_factor: probabilistic size ratio between adjacent
 //                            link lists in the skiplist
 extern MemTableRepFactory* NewHashSkipListRepFactory(
-  const SliceTransform* transform, size_t bucket_count = 1000000,
-  int32_t skiplist_height = 4, int32_t skiplist_branching_factor = 4
+    size_t bucket_count = 1000000, int32_t skiplist_height = 4,
+    int32_t skiplist_branching_factor = 4
 );
 
 // The factory is to create memtables with a hashed linked list:
@@ -211,6 +213,6 @@ extern MemTableRepFactory* NewHashSkipListRepFactory(
 // linked list (null if the bucket is empty).
 // bucket_count: number of fixed array buckets
 extern MemTableRepFactory* NewHashLinkListRepFactory(
-    const SliceTransform* transform, size_t bucket_count = 50000);
+    size_t bucket_count = 50000);
 
 }  // namespace rocksdb

include/rocksdb/options.h

@@ -277,7 +277,7 @@ struct Options {
   // 4) prefix(prefix(key)) == prefix(key)
   //
   // Default: nullptr
-  const SliceTransform* prefix_extractor;
+  std::shared_ptr<const SliceTransform> prefix_extractor;
 
   // If true, place whole keys in the filter (not just prefixes).
   // This must generally be true for gets to be efficient.

table/filter_block.cc

@@ -24,7 +24,7 @@ static const size_t kFilterBase = 1 << kFilterBaseLg;
 FilterBlockBuilder::FilterBlockBuilder(const Options& opt,
                                        const Comparator* internal_comparator)
     : policy_(opt.filter_policy),
-      prefix_extractor_(opt.prefix_extractor),
+      prefix_extractor_(opt.prefix_extractor.get()),
       whole_key_filtering_(opt.whole_key_filtering),
       comparator_(internal_comparator) {}
 
@@ -133,7 +133,7 @@ void FilterBlockBuilder::GenerateFilter() {
 FilterBlockReader::FilterBlockReader(
     const Options& opt, const Slice& contents, bool delete_contents_after_use)
     : policy_(opt.filter_policy),
-      prefix_extractor_(opt.prefix_extractor),
+      prefix_extractor_(opt.prefix_extractor.get()),
       whole_key_filtering_(opt.whole_key_filtering),
       data_(nullptr),
       offset_(nullptr),

table/table_reader_bench.cc

@@ -240,8 +240,8 @@ int main(int argc, char** argv) {
   rocksdb::TableFactory* tf = new rocksdb::BlockBasedTableFactory();
   rocksdb::Options options;
   if (FLAGS_prefix_len < 16) {
-    options.prefix_extractor = rocksdb::NewFixedPrefixTransform(
-        FLAGS_prefix_len);
+    options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(
+        FLAGS_prefix_len));
   }
   rocksdb::ReadOptions ro;
   rocksdb::EnvOptions env_options;
@@ -254,8 +254,8 @@ int main(int argc, char** argv) {
     env_options.use_mmap_reads = true;
     tf = new rocksdb::PlainTableFactory(16, (FLAGS_prefix_len == 16) ? 0 : 8,
                                         0.75);
-    options.prefix_extractor = rocksdb::NewFixedPrefixTransform(
-        FLAGS_prefix_len);
+    options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(
+        FLAGS_prefix_len));
   } else {
     tf = new rocksdb::BlockBasedTableFactory();
   }

table/table_test.cc

@@ -696,7 +696,7 @@ class Harness {
       case PLAIN_TABLE_SEMI_FIXED_PREFIX:
         support_prev_ = false;
         only_support_prefix_seek_ = true;
-        options_.prefix_extractor = prefix_transform.get();
+        options_.prefix_extractor.reset(new FixedOrLessPrefixTransform(2));
         options_.allow_mmap_reads = true;
         options_.table_factory.reset(NewPlainTableFactory());
         constructor_ = new TableConstructor(options_.comparator, true, true);
@@ -706,7 +706,7 @@ class Harness {
       case PLAIN_TABLE_FULL_STR_PREFIX:
         support_prev_ = false;
         only_support_prefix_seek_ = true;
-        options_.prefix_extractor = noop_transform.get();
+        options_.prefix_extractor.reset(NewNoopTransform());
         options_.allow_mmap_reads = true;
         options_.table_factory.reset(NewPlainTableFactory());
         constructor_ = new TableConstructor(options_.comparator, true, true);
@@ -919,15 +919,8 @@ class Harness {
   bool support_prev_;
   bool only_support_prefix_seek_;
   shared_ptr<InternalKeyComparator> internal_comparator_;
-  static std::unique_ptr<const SliceTransform> noop_transform;
-  static std::unique_ptr<const SliceTransform> prefix_transform;
 };
 
-std::unique_ptr<const SliceTransform> Harness::noop_transform(
-    NewNoopTransform());
-std::unique_ptr<const SliceTransform> Harness::prefix_transform(
-    new FixedOrLessPrefixTransform(2));
-
 static bool Between(uint64_t val, uint64_t low, uint64_t high) {
   bool result = (val >= low) && (val <= high);
   if (!result) {

tools/db_stress.cc

@@ -688,9 +688,6 @@ class StressTest {
         filter_policy_(FLAGS_bloom_bits >= 0
                        ? NewBloomFilterPolicy(FLAGS_bloom_bits)
                        : nullptr),
-        prefix_extractor_(NewFixedPrefixTransform(
-                          FLAGS_test_batches_snapshots ?
-                          sizeof(long) : sizeof(long)-1)),
         db_(nullptr),
         num_times_reopened_(0) {
     if (FLAGS_destroy_db_initially) {
@@ -708,7 +705,6 @@ class StressTest {
   ~StressTest() {
     delete db_;
     delete filter_policy_;
-    delete prefix_extractor_;
   }
 
   void Run() {
@@ -1373,7 +1369,7 @@ class StressTest {
       static_cast<rocksdb::CompactionStyle>(FLAGS_compaction_style);
     options.block_size = FLAGS_block_size;
     options.filter_policy = filter_policy_;
-    options.prefix_extractor = prefix_extractor_;
+    options.prefix_extractor.reset(NewFixedPrefixTransform(FLAGS_prefix_size));
     options.max_open_files = FLAGS_open_files;
     options.statistics = dbstats;
     options.env = FLAGS_env;
@@ -1405,16 +1401,13 @@ class StressTest {
     }
     switch (FLAGS_rep_factory) {
       case kHashSkipList:
-        options.memtable_factory.reset(NewHashSkipListRepFactory(
-            NewFixedPrefixTransform(FLAGS_prefix_size)));
+        options.memtable_factory.reset(NewHashSkipListRepFactory());
         break;
       case kSkipList:
         // no need to do anything
         break;
       case kVectorRep:
-        options.memtable_factory.reset(
-          new VectorRepFactory()
-        );
+        options.memtable_factory.reset(new VectorRepFactory());
         break;
     }
     static Random purge_percent(1000); // no benefit from non-determinism here
@@ -1488,7 +1481,6 @@ class StressTest {
   shared_ptr<Cache> cache_;
   shared_ptr<Cache> compressed_cache_;
   const FilterPolicy* filter_policy_;
-  const SliceTransform* prefix_extractor_;
   DB* db_;
   StackableDB* sdb_;
   int num_times_reopened_;

tools/sst_dump.cc

@@ -130,7 +130,7 @@ Status SstFileReader::SetTableOptionsByMagicNumber(uint64_t table_magic_number,
     options_.allow_mmap_reads = true;
     options_.table_factory = std::make_shared<PlainTableFactory>(
         table_properties->fixed_key_len, 2, 0.8);
-    options_.prefix_extractor = NewNoopTransform();
+    options_.prefix_extractor.reset(NewNoopTransform());
     fprintf(stdout, "Sst file format: plain table\n");
   } else {
     char error_msg_buffer[80];

util/hash_linklist_rep.cc

@@ -55,7 +55,7 @@ private:
 
 class HashLinkListRep : public MemTableRep {
  public:
-  HashLinkListRep(MemTableRep::KeyComparator& compare, Arena* arena,
+  HashLinkListRep(const MemTableRep::KeyComparator& compare, Arena* arena,
                   const SliceTransform* transform, size_t bucket_size);
 
   virtual void Insert(const char* key) override;
@@ -81,7 +81,7 @@ class HashLinkListRep : public MemTableRep {
 
  private:
   friend class DynamicIterator;
-  typedef SkipList<const char*, MemTableRep::KeyComparator&> FullList;
+  typedef SkipList<const char*, const MemTableRep::KeyComparator&> FullList;
 
   size_t bucket_size_;
 
@@ -92,7 +92,7 @@ class HashLinkListRep : public MemTableRep {
   // The user-supplied transform whose domain is the user keys.
   const SliceTransform* transform_;
 
-  MemTableRep::KeyComparator& compare_;
+  const MemTableRep::KeyComparator& compare_;
   // immutable after construction
   Arena* const arena_;
 
@@ -314,7 +314,7 @@ class HashLinkListRep : public MemTableRep {
   };
 };
 
-HashLinkListRep::HashLinkListRep(MemTableRep::KeyComparator& compare,
+HashLinkListRep::HashLinkListRep(const MemTableRep::KeyComparator& compare,
                                  Arena* arena, const SliceTransform* transform,
                                  size_t bucket_size)
   : bucket_size_(bucket_size),
@@ -475,13 +475,13 @@ Node* HashLinkListRep::FindGreaterOrEqualInBucket(Node* head,
 } // anon namespace
 
 MemTableRep* HashLinkListRepFactory::CreateMemTableRep(
-    MemTableRep::KeyComparator& compare, Arena* arena) {
-  return new HashLinkListRep(compare, arena, transform_, bucket_count_);
+    const MemTableRep::KeyComparator& compare, Arena* arena,
+    const SliceTransform* transform) {
+  return new HashLinkListRep(compare, arena, transform, bucket_count_);
 }
 
-MemTableRepFactory* NewHashLinkListRepFactory(
-    const SliceTransform* transform, size_t bucket_count) {
-  return new HashLinkListRepFactory(transform, bucket_count);
+MemTableRepFactory* NewHashLinkListRepFactory(size_t bucket_count) {
+  return new HashLinkListRepFactory(bucket_count);
 }
 
 } // namespace rocksdb

util/hash_linklist_rep.h

@@ -14,25 +14,20 @@ namespace rocksdb {
 
 class HashLinkListRepFactory : public MemTableRepFactory {
  public:
-  explicit HashLinkListRepFactory(
-    const SliceTransform* transform,
-    size_t bucket_count)
-      : transform_(transform),
-        bucket_count_(bucket_count) { }
+  explicit HashLinkListRepFactory(size_t bucket_count)
+      : bucket_count_(bucket_count) { }
 
-  virtual ~HashLinkListRepFactory() { delete transform_; }
+  virtual ~HashLinkListRepFactory() {}
 
-  virtual MemTableRep* CreateMemTableRep(MemTableRep::KeyComparator& compare,
-                                         Arena* arena) override;
+  virtual MemTableRep* CreateMemTableRep(
+      const MemTableRep::KeyComparator& compare, Arena* arena,
+      const SliceTransform* transform) override;
 
   virtual const char* Name() const override {
     return "HashLinkListRepFactory";
   }
 
-  const SliceTransform* GetTransform() { return transform_; }
-
  private:
-  const SliceTransform* transform_;
   const size_t bucket_count_;
 };
 

util/hash_skiplist_rep.cc

@@ -21,7 +21,7 @@ namespace {
 
 class HashSkipListRep : public MemTableRep {
  public:
-  HashSkipListRep(MemTableRep::KeyComparator& compare, Arena* arena,
+  HashSkipListRep(const MemTableRep::KeyComparator& compare, Arena* arena,
                   const SliceTransform* transform, size_t bucket_size,
                   int32_t skiplist_height, int32_t skiplist_branching_factor);
 
@@ -48,7 +48,7 @@ class HashSkipListRep : public MemTableRep {
 
  private:
   friend class DynamicIterator;
-  typedef SkipList<const char*, MemTableRep::KeyComparator&> Bucket;
+  typedef SkipList<const char*, const MemTableRep::KeyComparator&> Bucket;
 
   size_t bucket_size_;
 
@@ -62,7 +62,7 @@ class HashSkipListRep : public MemTableRep {
   // The user-supplied transform whose domain is the user keys.
   const SliceTransform* transform_;
 
-  MemTableRep::KeyComparator& compare_;
+  const MemTableRep::KeyComparator& compare_;
   // immutable after construction
   Arena* const arena_;
 
@@ -221,7 +221,7 @@ class HashSkipListRep : public MemTableRep {
   };
 };
 
-HashSkipListRep::HashSkipListRep(MemTableRep::KeyComparator& compare,
+HashSkipListRep::HashSkipListRep(const MemTableRep::KeyComparator& compare,
                                  Arena* arena, const SliceTransform* transform,
                                  size_t bucket_size, int32_t skiplist_height,
                                  int32_t skiplist_branching_factor)
@@ -321,16 +321,17 @@ MemTableRep::Iterator* HashSkipListRep::GetDynamicPrefixIterator() {
 } // anon namespace
 
 MemTableRep* HashSkipListRepFactory::CreateMemTableRep(
-    MemTableRep::KeyComparator& compare, Arena* arena) {
-  return new HashSkipListRep(compare, arena, transform_, bucket_count_,
+    const MemTableRep::KeyComparator& compare, Arena* arena,
+    const SliceTransform* transform) {
+  return new HashSkipListRep(compare, arena, transform, bucket_count_,
                              skiplist_height_, skiplist_branching_factor_);
 }
 
 MemTableRepFactory* NewHashSkipListRepFactory(
-    const SliceTransform* transform, size_t bucket_count,
-    int32_t skiplist_height, int32_t skiplist_branching_factor) {
-  return new HashSkipListRepFactory(transform, bucket_count,
-                                    skiplist_height, skiplist_branching_factor);
+    size_t bucket_count, int32_t skiplist_height,
+    int32_t skiplist_branching_factor) {
+  return new HashSkipListRepFactory(bucket_count, skiplist_height,
+      skiplist_branching_factor);
 }
 
 } // namespace rocksdb

util/hash_skiplist_rep.h

@@ -15,28 +15,24 @@ namespace rocksdb {
 class HashSkipListRepFactory : public MemTableRepFactory {
  public:
   explicit HashSkipListRepFactory(
-    const SliceTransform* transform,
     size_t bucket_count,
     int32_t skiplist_height,
     int32_t skiplist_branching_factor)
-      : transform_(transform),
-        bucket_count_(bucket_count),
+      : bucket_count_(bucket_count),
         skiplist_height_(skiplist_height),
         skiplist_branching_factor_(skiplist_branching_factor) { }
 
-  virtual ~HashSkipListRepFactory() { delete transform_; }
+  virtual ~HashSkipListRepFactory() {}
 
-  virtual MemTableRep* CreateMemTableRep(MemTableRep::KeyComparator& compare,
-                                         Arena* arena) override;
+  virtual MemTableRep* CreateMemTableRep(
+      const MemTableRep::KeyComparator& compare, Arena* arena,
+      const SliceTransform* transform) override;
 
   virtual const char* Name() const override {
     return "HashSkipListRepFactory";
   }
 
-  const SliceTransform* GetTransform() { return transform_; }
-
  private:
-  const SliceTransform* transform_;
   const size_t bucket_count_;
   const int32_t skiplist_height_;
   const int32_t skiplist_branching_factor_;

util/skiplistrep.cc

@@ -10,11 +10,11 @@
 namespace rocksdb {
 namespace {
 class SkipListRep : public MemTableRep {
-  SkipList<const char*, MemTableRep::KeyComparator&> skip_list_;
+  SkipList<const char*, const MemTableRep::KeyComparator&> skip_list_;
 public:
-  explicit SkipListRep(MemTableRep::KeyComparator& compare, Arena* arena)
+  explicit SkipListRep(const MemTableRep::KeyComparator& compare, Arena* arena)
     : skip_list_(compare, arena) {
-}
+  }
 
   // Insert key into the list.
   // REQUIRES: nothing that compares equal to key is currently in the list.
@@ -47,12 +47,12 @@ public:
 
   // Iteration over the contents of a skip list
   class Iterator : public MemTableRep::Iterator {
-    SkipList<const char*, MemTableRep::KeyComparator&>::Iterator iter_;
+    SkipList<const char*, const MemTableRep::KeyComparator&>::Iterator iter_;
    public:
     // Initialize an iterator over the specified list.
     // The returned iterator is not valid.
     explicit Iterator(
-      const SkipList<const char*, MemTableRep::KeyComparator&>* list
+      const SkipList<const char*, const MemTableRep::KeyComparator&>* list
     ) : iter_(list) { }
 
     virtual ~Iterator() override { }
@@ -115,7 +115,8 @@ public:
 }
 
 MemTableRep* SkipListFactory::CreateMemTableRep(
-    MemTableRep::KeyComparator& compare, Arena* arena) {
+    const MemTableRep::KeyComparator& compare, Arena* arena,
+    const SliceTransform*) {
   return new SkipListRep(compare, arena);
 }
 

util/vectorrep.cc

@@ -271,7 +271,8 @@ MemTableRep::Iterator* VectorRep::GetIterator() {
 } // anon namespace
 
 MemTableRep* VectorRepFactory::CreateMemTableRep(
-    MemTableRep::KeyComparator& compare, Arena* arena) {
+    const MemTableRep::KeyComparator& compare, Arena* arena,
+    const SliceTransform*) {
   return new VectorRep(compare, arena, count_);
 }
 } // namespace rocksdb