Move prefix_extractor to MutableCFOptions

Summary:
Currently it is not possible to change bloom filter config without restart the db, which is causing a lot of operational complexity for users.
This PR aims to make it possible to dynamically change bloom filter config.
Closes https://github.com/facebook/rocksdb/pull/3601

Differential Revision: D7253114

Pulled By: miasantreble

fbshipit-source-id: f22595437d3e0b86c95918c484502de2ceca120c

69 files changed, 1306 insertions, 752 deletions

diff --git a/HISTORY.md b/HISTORY.md
index c299a0505..c340d3fee 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -21,6 +21,7 @@
 * 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.
diff --git a/db/builder.cc b/db/builder.cc
index e0c39be84..8e944983d 100644
--- a/db/builder.cc
+++ b/db/builder.cc
@@ -39,7 +39,7 @@ namespace rocksdb {
 class TableFactory;
 
 TableBuilder* NewTableBuilder(
-    const ImmutableCFOptions& ioptions,
+    const ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
     const InternalKeyComparator& internal_comparator,
     const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
         int_tbl_prop_collector_factories,
@@ -52,19 +52,20 @@ TableBuilder* NewTableBuilder(
           TablePropertiesCollectorFactory::Context::kUnknownColumnFamily) ==
          column_family_name.empty());
   return ioptions.table_factory->NewTableBuilder(
-      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),
+      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),
       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,
@@ -122,10 +123,11 @@ Status BuildTable(
       file_writer.reset(new WritableFileWriter(std::move(file), env_options,
                                                ioptions.statistics));
       builder = NewTableBuilder(
-          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);
+          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);
     }
 
     MergeHelper merge(env, internal_comparator.user_comparator(),
@@ -195,7 +197,8 @@ 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 */, nullptr,
+          nullptr /* range_del_agg */,
+          mutable_cf_options.prefix_extractor.get(), nullptr,
           (internal_stats == nullptr) ? nullptr
                                       : internal_stats->GetFileReadHist(0),
           false /* for_compaction */, nullptr /* arena */,
diff --git a/db/builder.h b/db/builder.h
index d83644499..0e8218e74 100644
--- a/db/builder.h
+++ b/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 ImmutableCFOptions& options, const MutableCFOptions& moptions,
     const InternalKeyComparator& internal_comparator,
     const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
         int_tbl_prop_collector_factories,
diff --git a/db/compacted_db_impl.cc b/db/compacted_db_impl.cc
index 8f44b08b8..823276c4d 100644
--- a/db/compacted_db_impl.cc
+++ b/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);
+  files_.files[FindFile(key)].fd.table_reader->Get(options, lkey.internal_key(),
+                                                   &get_context, nullptr);
   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);
+      r->Get(options, lkey.internal_key(), &get_context, nullptr);
       value.assign(pinnable_val.data(), pinnable_val.size());
       if (get_context.State() == GetContext::kFound) {
         statuses[idx] = Status::OK();
diff --git a/db/compaction_job.cc b/db/compaction_job.cc
index b302f2dd6..8bafabd5d 100644
--- a/db/compaction_job.cc
+++ b/db/compaction_job.cc
@@ -1183,7 +1183,9 @@ 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 */, nullptr,
+        nullptr /* range_del_agg */,
+        sub_compact->compaction->mutable_cf_options()->prefix_extractor.get(),
+        nullptr,
         cfd->internal_stats()->GetFileReadHist(
             compact_->compaction->output_level()),
         false, nullptr /* arena */, false /* skip_filters */,
@@ -1386,9 +1388,10 @@ Status CompactionJob::OpenCompactionOutputFile(
   }
 
   sub_compact->builder.reset(NewTableBuilder(
-      *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(), *(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()->compression_opts,
       sub_compact->compaction->output_level(), &sub_compact->compression_dict,
       skip_filters, output_file_creation_time));
diff --git a/db/convenience.cc b/db/convenience.cc
index bfa11f202..22ffd821c 100644
--- a/db/convenience.cc
+++ b/db/convenience.cc
@@ -50,8 +50,9 @@ 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, env_options, internal_comparator,
-                         false /* skip_filters */, -1 /* level */),
+      TableReaderOptions(ioptions, options.prefix_extractor.get(), 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()) {
diff --git a/db/db_impl.cc b/db/db_impl.cc
index cc1140160..1be9c29c7 100644
--- a/db/db_impl.cc
+++ b/db/db_impl.cc
@@ -969,7 +969,7 @@ InternalIterator* DBImpl::NewInternalIterator(
   MergeIteratorBuilder merge_iter_builder(
       &cfd->internal_comparator(), arena,
       !read_options.total_order_seek &&
-          cfd->ioptions()->prefix_extractor != nullptr);
+          super_version->mutable_cf_options.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(),
-                      cfd->user_comparator(), iter, kMaxSequenceNumber,
-                      sv->mutable_cf_options.max_sequential_skip_in_iterations,
-                      read_callback);
+    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);
 #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(), snapshot,
+      env_, read_options, *cfd->ioptions(), sv->mutable_cf_options, 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(), cfd->user_comparator(), iter,
-          kMaxSequenceNumber,
+          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));
     }
@@ -2863,7 +2863,9 @@ Status DBImpl::IngestExternalFile(
     pending_output_elem = CaptureCurrentFileNumberInPendingOutputs();
   }
 
-  status = ingestion_job.Prepare(external_files);
+  SuperVersion* super_version = cfd->GetReferencedSuperVersion(&mutex_);
+  status = ingestion_job.Prepare(external_files, super_version);
+  CleanupSuperVersion(super_version);
   if (!status.ok()) {
     return status;
   }
diff --git a/db/db_impl_readonly.cc b/db/db_impl_readonly.cc
index b4f140f5a..e302344d9 100644
--- a/db/db_impl_readonly.cc
+++ b/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(),
+      env_, read_options, *cfd->ioptions(), super_version->mutable_cf_options,
       (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(),
+        env_, read_options, *cfd->ioptions(), sv->mutable_cf_options,
         (read_options.snapshot != nullptr
              ? reinterpret_cast<const SnapshotImpl*>(read_options.snapshot)
                    ->number_
diff --git a/db/db_iter.cc b/db/db_iter.cc
index d2d1d0451..e24db0fb4 100644
--- a/db/db_iter.cc
+++ b/db/db_iter.cc
@@ -110,7 +110,8 @@ class DBIter final: public Iterator {
   };
 
   DBIter(Env* _env, const ReadOptions& read_options,
-         const ImmutableCFOptions& cf_options, const Comparator* cmp,
+         const ImmutableCFOptions& cf_options,
+         const MutableCFOptions& mutable_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)
@@ -138,7 +139,7 @@ class DBIter final: public Iterator {
         is_blob_(false),
         start_seqnum_(read_options.iter_start_seqnum) {
     RecordTick(statistics_, NO_ITERATORS);
-    prefix_extractor_ = cf_options.prefix_extractor;
+    prefix_extractor_ = mutable_cf_options.prefix_extractor.get();
     max_skip_ = max_sequential_skip_in_iterations;
     max_skippable_internal_keys_ = read_options.max_skippable_internal_keys;
     if (pin_thru_lifetime_) {
@@ -1426,14 +1427,15 @@ 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, user_key_comparator,
-                 internal_iter, sequence, false,
+      new DBIter(env, read_options, cf_options, mutable_cf_options,
+                 user_key_comparator, internal_iter, sequence, false,
                  max_sequential_skip_in_iterations, read_callback, allow_blob);
   return db_iter;
 }
@@ -1477,6 +1479,7 @@ 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,
@@ -1484,9 +1487,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, cf_options.user_comparator, nullptr,
-             sequence, true, max_sequential_skip_in_iteration, read_callback,
-             allow_blob);
+      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);
   sv_number_ = version_number;
   allow_refresh_ = allow_refresh;
 }
@@ -1508,8 +1511,8 @@ Status ArenaWrappedDBIter::Refresh() {
     new (&arena_) Arena();
 
     SuperVersion* sv = cfd_->GetReferencedSuperVersion(db_impl_->mutex());
-    Init(env, read_options_, *(cfd_->ioptions()), latest_seq,
-         sv->mutable_cf_options.max_sequential_skip_in_iterations,
+    Init(env, read_options_, *(cfd_->ioptions()), sv->mutable_cf_options,
+         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(
@@ -1524,12 +1527,13 @@ Status ArenaWrappedDBIter::Refresh() {
 
 ArenaWrappedDBIter* NewArenaWrappedDbIterator(
     Env* env, const ReadOptions& read_options,
-    const ImmutableCFOptions& cf_options, const SequenceNumber& sequence,
+    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, DBImpl* db_impl, ColumnFamilyData* cfd,
     bool allow_blob, bool allow_refresh) {
   ArenaWrappedDBIter* iter = new ArenaWrappedDBIter();
-  iter->Init(env, read_options, cf_options, sequence,
+  iter->Init(env, read_options, cf_options, mutable_cf_options, sequence,
              max_sequential_skip_in_iterations, version_number, read_callback,
              allow_blob, allow_refresh);
   if (db_impl != nullptr && cfd != nullptr && allow_refresh) {
diff --git a/db/db_iter.h b/db/db_iter.h
index df7c1e1da..989a0ba1f 100644
--- a/db/db_iter.h
+++ b/db/db_iter.h
@@ -30,6 +30,7 @@ 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,
@@ -71,6 +72,7 @@ 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);
@@ -102,7 +104,8 @@ class ArenaWrappedDBIter : public Iterator {
 // be supported.
 extern ArenaWrappedDBIter* NewArenaWrappedDbIterator(
     Env* env, const ReadOptions& read_options,
-    const ImmutableCFOptions& cf_options, const SequenceNumber& sequence,
+    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, DBImpl* db_impl = nullptr,
     ColumnFamilyData* cfd = nullptr, bool allow_blob = false,
diff --git a/db/db_iter_stress_test.cc b/db/db_iter_stress_test.cc
index cebe1ff90..a0f1dfeab 100644
--- a/db/db_iter_stress_test.cc
+++ b/db/db_iter_stress_test.cc
@@ -509,7 +509,8 @@ TEST_F(DBIteratorStressTest, StressTest) {
                   internal_iter->trace = trace;
                   db_iter.reset(NewDBIterator(
                       env_, ropt, ImmutableCFOptions(options),
-                      BytewiseComparator(), internal_iter, sequence,
+                      MutableCFOptions(options), BytewiseComparator(),
+                      internal_iter, sequence,
                       options.max_sequential_skip_in_iterations,
                       nullptr /*read_callback*/));
                 }
diff --git a/db/db_iter_test.cc b/db/db_iter_test.cc
index 4af678b4a..aa4b39c38 100644
--- a/db/db_iter_test.cc
+++ b/db/db_iter_test.cc
@@ -234,7 +234,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,8 +248,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -280,8 +281,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -306,8 +308,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -338,8 +341,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -373,8 +377,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(!db_iter->Valid());
@@ -402,8 +407,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 7,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 7, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     SetPerfLevel(kEnableCount);
     ASSERT_TRUE(GetPerfLevel() == kEnableCount);
@@ -439,8 +445,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 4,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 4, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -464,8 +471,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(!db_iter->Valid());
@@ -486,8 +494,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -521,8 +530,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
     ro.iterate_upper_bound = &prefix;
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 7,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 7, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     SetPerfLevel(kEnableCount);
     ASSERT_TRUE(GetPerfLevel() == kEnableCount);
@@ -550,8 +560,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -592,8 +603,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 2,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -623,8 +635,9 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 
     ReadOptions ro;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -645,6 +658,7 @@ TEST_F(DBIteratorTest, DBIteratorPrevNext) {
 TEST_F(DBIteratorTest, DBIteratorEmpty) {
   Options options;
   ImmutableCFOptions cf_options = ImmutableCFOptions(options);
+  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
   ReadOptions ro;
 
   {
@@ -652,8 +666,9 @@ TEST_F(DBIteratorTest, DBIteratorEmpty) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 0, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(!db_iter->Valid());
   }
@@ -663,8 +678,9 @@ TEST_F(DBIteratorTest, DBIteratorEmpty) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 0, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToFirst();
     ASSERT_TRUE(!db_iter->Valid());
   }
@@ -684,10 +700,10 @@ TEST_F(DBIteratorTest, DBIteratorUseSkipCountSkips) {
   }
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(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), MutableCFOptions(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");
@@ -716,6 +732,7 @@ 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) {
@@ -729,8 +746,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
 
       options.statistics = rocksdb::CreateDBStatistics();
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, BytewiseComparator(), internal_iter, i + 2,
-          options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+          internal_iter, i + 2, options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -765,8 +782,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->Finish();
 
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, BytewiseComparator(), internal_iter, i + 2,
-          options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+          internal_iter, i + 2, options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -794,8 +811,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->Finish();
 
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, BytewiseComparator(), internal_iter, 202,
-          options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+          internal_iter, 202, options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -827,8 +844,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->AddPut("c", "200");
       internal_iter->Finish();
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, BytewiseComparator(), internal_iter, i,
-          options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+          internal_iter, i, options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(!db_iter->Valid());
@@ -844,8 +861,9 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
     internal_iter->AddPut("c", "200");
     internal_iter->Finish();
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 200,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -878,8 +896,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->Finish();
 
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, BytewiseComparator(), internal_iter, i + 2,
-          options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+          internal_iter, i + 2, options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -913,8 +931,8 @@ TEST_F(DBIteratorTest, DBIteratorUseSkip) {
       internal_iter->Finish();
 
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, BytewiseComparator(), internal_iter, i + 2,
-          options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+          internal_iter, i + 2, options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
       db_iter->SeekToLast();
       ASSERT_TRUE(db_iter->Valid());
@@ -945,6 +963,7 @@ 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.
@@ -962,8 +981,9 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 0;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1008,8 +1028,9 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1052,8 +1073,9 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1090,8 +1112,9 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1125,8 +1148,9 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -1155,8 +1179,9 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1192,8 +1217,9 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
     ro.max_skippable_internal_keys = 2;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 10, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
 
     db_iter->SeekToFirst();
     ASSERT_TRUE(db_iter->Valid());
@@ -1229,8 +1255,8 @@ TEST_F(DBIteratorTest, DBIteratorSkipInternalKeys) {
 
       ro.max_skippable_internal_keys = i;
       std::unique_ptr<Iterator> db_iter(NewDBIterator(
-          env_, ro, cf_options, BytewiseComparator(), internal_iter, 2 * i + 1,
-          options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+          internal_iter, 2 * i + 1, options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
 
       db_iter->SeekToFirst();
@@ -1283,8 +1309,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, BytewiseComparator(), internal_iter, 2 * i + 1,
-          options.max_sequential_skip_in_iterations,
+          env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+          internal_iter, 2 * i + 1, options.max_sequential_skip_in_iterations,
           nullptr /*read_callback*/));
 
       db_iter->SeekToFirst();
@@ -1321,10 +1347,10 @@ TEST_F(DBIteratorTest, DBIterator1) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(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), MutableCFOptions(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");
@@ -1349,10 +1375,10 @@ TEST_F(DBIteratorTest, DBIterator2) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(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), MutableCFOptions(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");
@@ -1374,10 +1400,10 @@ TEST_F(DBIteratorTest, DBIterator3) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(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), MutableCFOptions(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");
@@ -1399,10 +1425,10 @@ TEST_F(DBIteratorTest, DBIterator4) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(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), MutableCFOptions(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");
@@ -1420,6 +1446,7 @@ 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());
@@ -1433,8 +1460,9 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1455,8 +1483,9 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 1,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1477,8 +1506,9 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 2,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1499,8 +1529,9 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 3,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1521,8 +1552,9 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 4,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1543,8 +1575,9 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 5,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1565,8 +1598,9 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 6,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1585,8 +1619,9 @@ TEST_F(DBIteratorTest, DBIterator5) {
     internal_iter->AddPut("b", "val_b");
     internal_iter->Finish();
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 10,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1601,6 +1636,7 @@ 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());
@@ -1614,8 +1650,9 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1636,8 +1673,9 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 1,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1658,8 +1696,9 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 2,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1680,8 +1719,9 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 3,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 3, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(!db_iter->Valid());
   }
@@ -1698,8 +1738,9 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 4,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1720,8 +1761,9 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 5,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1742,8 +1784,9 @@ TEST_F(DBIteratorTest, DBIterator6) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 6,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1758,6 +1801,7 @@ 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());
@@ -1783,8 +1827,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 0,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_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");
@@ -1817,8 +1862,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 2,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 2, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -1857,8 +1903,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 4,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 4, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -1897,8 +1944,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 5,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 5, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -1942,8 +1990,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 6,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 6, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -1988,8 +2037,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 7,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 7, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2028,8 +2078,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 9,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 9, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2074,8 +2125,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 13,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 13, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2121,8 +2173,9 @@ TEST_F(DBIteratorTest, DBIterator7) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, cf_options, BytewiseComparator(), internal_iter, 14,
-        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
+        env_, ro, cf_options, mutable_cf_options, BytewiseComparator(),
+        internal_iter, 14, options.max_sequential_skip_in_iterations,
+        nullptr /*read_callback*/));
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
 
@@ -2151,9 +2204,9 @@ TEST_F(DBIteratorTest, DBIterator8) {
   internal_iter->Finish();
 
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-      internal_iter, 10, options.max_sequential_skip_in_iterations,
-      nullptr /*read_callback*/));
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(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");
@@ -2182,9 +2235,9 @@ TEST_F(DBIteratorTest, DBIterator9) {
     internal_iter->Finish();
 
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-        internal_iter, 10, options.max_sequential_skip_in_iterations,
-        nullptr /*read_callback*/));
+        env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
+        BytewiseComparator(), internal_iter, 10,
+        options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
     db_iter->SeekToLast();
     ASSERT_TRUE(db_iter->Valid());
@@ -2249,9 +2302,9 @@ TEST_F(DBIteratorTest, DBIterator10) {
   internal_iter->Finish();
 
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-      internal_iter, 10, options.max_sequential_skip_in_iterations,
-      nullptr /*read_callback*/));
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
+      BytewiseComparator(), internal_iter, 10,
+      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
   db_iter->Seek("c");
   ASSERT_TRUE(db_iter->Valid());
@@ -2289,8 +2342,9 @@ TEST_F(DBIteratorTest, SeekToLastOccurrenceSeq0) {
   internal_iter->Finish();
 
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-      internal_iter, 10, 0 /* force seek */, nullptr /*read_callback*/));
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(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");
@@ -2316,10 +2370,10 @@ TEST_F(DBIteratorTest, DBIterator11) {
   internal_iter->AddMerge("b", "2");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(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), MutableCFOptions(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");
@@ -2343,9 +2397,9 @@ TEST_F(DBIteratorTest, DBIterator12) {
   internal_iter->AddSingleDeletion("b");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-                    internal_iter, 10, 0, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(NewDBIterator(
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
+      BytewiseComparator(), internal_iter, 10, 0, nullptr /*read_callback*/));
   db_iter->SeekToLast();
   ASSERT_TRUE(db_iter->Valid());
   ASSERT_EQ(db_iter->key().ToString(), "c");
@@ -2380,9 +2434,9 @@ TEST_F(DBIteratorTest, DBIterator13) {
   internal_iter->AddPut(key, "8");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-                    internal_iter, 2, 3, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(NewDBIterator(
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(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);
@@ -2408,9 +2462,9 @@ TEST_F(DBIteratorTest, DBIterator14) {
   internal_iter->AddPut("c", "9");
   internal_iter->Finish();
 
-  std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-                    internal_iter, 4, 1, nullptr /*read_callback*/));
+  std::unique_ptr<Iterator> db_iter(NewDBIterator(
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(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");
@@ -2435,10 +2489,10 @@ TEST_F(DBIteratorTest, DBIteratorTestDifferentialSnapshots) {
     }
     internal_iter->Finish();
 
-    std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-                    internal_iter, 13,
-                    options.max_sequential_skip_in_iterations, nullptr));
+    std::unique_ptr<Iterator> db_iter(NewDBIterator(
+        env_, ro, ImmutableCFOptions(options), MutableCFOptions(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"};
@@ -2470,10 +2524,10 @@ TEST_F(DBIteratorTest, DBIteratorTestDifferentialSnapshots) {
     }
     internal_iter->Finish();
 
-    std::unique_ptr<Iterator> db_iter(
-      NewDBIterator(env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-                    internal_iter, 13,
-                    options.max_sequential_skip_in_iterations, nullptr));
+    std::unique_ptr<Iterator> db_iter(NewDBIterator(
+        env_, ro, ImmutableCFOptions(options), MutableCFOptions(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,
@@ -2521,8 +2575,9 @@ class DBIterWithMergeIterTest : public testing::Test {
         NewMergingIterator(&icomp_, &child_iters[0], 2u);
 
     db_iter_.reset(NewDBIterator(
-        env_, ro_, ImmutableCFOptions(options_), BytewiseComparator(),
-        merge_iter, 8 /* read data earlier than seqId 8 */,
+        env_, ro_, ImmutableCFOptions(options_), MutableCFOptions(options_),
+        BytewiseComparator(), merge_iter,
+        8 /* read data earlier than seqId 8 */,
         3 /* max iterators before reseek */, nullptr /*read_callback*/));
   }
 
@@ -2960,9 +3015,9 @@ TEST_F(DBIteratorTest, SeekPrefixTombstones) {
 
   ro.prefix_same_as_start = true;
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-      internal_iter, 10, options.max_sequential_skip_in_iterations,
-      nullptr /*read_callback*/));
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
+      BytewiseComparator(), internal_iter, 10,
+      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
   int skipped_keys = 0;
 
@@ -2996,9 +3051,10 @@ TEST_F(DBIteratorTest, SeekToFirstLowerBound) {
     ro.iterate_lower_bound = &lower_bound;
     Options options;
     std::unique_ptr<Iterator> db_iter(NewDBIterator(
-        env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-        internal_iter, 10 /* sequence */,
-        options.max_sequential_skip_in_iterations, nullptr /* read_callback */));
+        env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
+        BytewiseComparator(), internal_iter, 10 /* sequence */,
+        options.max_sequential_skip_in_iterations,
+        nullptr /* read_callback */));
 
     db_iter->SeekToFirst();
     if (i == kNumKeys + 1) {
@@ -3034,8 +3090,8 @@ TEST_F(DBIteratorTest, PrevLowerBound) {
   ro.iterate_lower_bound = &lower_bound;
   Options options;
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-      internal_iter, 10 /* sequence */,
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(options),
+      BytewiseComparator(), internal_iter, 10 /* sequence */,
       options.max_sequential_skip_in_iterations, nullptr /* read_callback */));
 
   db_iter->SeekToLast();
@@ -3062,8 +3118,8 @@ TEST_F(DBIteratorTest, SeekLessLowerBound) {
   ro.iterate_lower_bound = &lower_bound;
   Options options;
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ro, ImmutableCFOptions(options), BytewiseComparator(),
-      internal_iter, 10 /* sequence */,
+      env_, ro, ImmutableCFOptions(options), MutableCFOptions(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);
@@ -3087,9 +3143,9 @@ TEST_F(DBIteratorTest, ReverseToForwardWithDisappearingKeys) {
   internal_iter->Finish();
 
   std::unique_ptr<Iterator> db_iter(NewDBIterator(
-      env_, ReadOptions(), ImmutableCFOptions(options), BytewiseComparator(),
-      internal_iter, 10, options.max_sequential_skip_in_iterations,
-      nullptr /*read_callback*/));
+      env_, ReadOptions(), ImmutableCFOptions(options),
+      MutableCFOptions(options), BytewiseComparator(), internal_iter, 10,
+      options.max_sequential_skip_in_iterations, nullptr /*read_callback*/));
 
   db_iter->SeekForPrev("a");
   ASSERT_TRUE(db_iter->Valid());
diff --git a/db/db_iterator_test.cc b/db/db_iterator_test.cc
index dd2d1ce49..e6504dc50 100644
--- a/db/db_iterator_test.cc
+++ b/db/db_iterator_test.cc
@@ -862,6 +862,8 @@ 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_;
@@ -946,9 +948,7 @@ TEST_P(DBIteratorTest, DBIteratorBoundTest) {
   }
 
   // prefix is the first letter of the key
-  options.prefix_extractor.reset(NewFixedPrefixTransform(1));
-
-  DestroyAndReopen(options);
+  ASSERT_OK(dbfull()->SetOptions({{"prefix_extractor", "fixed:1"}}));
   ASSERT_OK(Put("a", "0"));
   ASSERT_OK(Put("foo", "bar"));
   ASSERT_OK(Put("foo1", "bar1"));
@@ -1035,6 +1035,7 @@ 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;
diff --git a/db/db_test.cc b/db/db_test.cc
index 351c9be22..c2dce2b3a 100644
--- a/db/db_test.cc
+++ b/db/db_test.cc
@@ -4607,6 +4607,181 @@ 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;
@@ -5445,18 +5620,17 @@ 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) {
diff --git a/db/external_sst_file_ingestion_job.cc b/db/external_sst_file_ingestion_job.cc
index b0dce8aec..4042961c9 100644
--- a/db/external_sst_file_ingestion_job.cc
+++ b/db/external_sst_file_ingestion_job.cc
@@ -29,13 +29,13 @@
 namespace rocksdb {
 
 Status ExternalSstFileIngestionJob::Prepare(
-    const std::vector<std::string>& external_files_paths) {
+    const std::vector<std::string>& external_files_paths, SuperVersion* sv) {
   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);
+    status = GetIngestedFileInfo(file_path, &file_to_ingest, sv);
     if (!status.ok()) {
       return status;
     }
@@ -284,7 +284,8 @@ void ExternalSstFileIngestionJob::Cleanup(const Status& status) {
 }
 
 Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
-    const std::string& external_file, IngestedFileInfo* file_to_ingest) {
+    const std::string& external_file, IngestedFileInfo* file_to_ingest,
+    SuperVersion* sv) {
   file_to_ingest->external_file_path = external_file;
 
   // Get external file size
@@ -306,8 +307,9 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
                                                    external_file));
 
   status = cfd_->ioptions()->table_factory->NewTableReader(
-      TableReaderOptions(*cfd_->ioptions(), env_options_,
-                         cfd_->internal_comparator()),
+      TableReaderOptions(*cfd_->ioptions(),
+                         sv->mutable_cf_options.prefix_extractor.get(),
+                         env_options_, cfd_->internal_comparator()),
       std::move(sst_file_reader), file_to_ingest->file_size, &table_reader);
   if (!status.ok()) {
     return status;
@@ -363,7 +365,8 @@ 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));
+  std::unique_ptr<InternalIterator> iter(table_reader->NewIterator(
+      ro, sv->mutable_cf_options.prefix_extractor.get()));
 
   // Get first (smallest) key from file
   iter->SeekToFirst();
diff --git a/db/external_sst_file_ingestion_job.h b/db/external_sst_file_ingestion_job.h
index 44d1314ee..7833e0b37 100644
--- a/db/external_sst_file_ingestion_job.h
+++ b/db/external_sst_file_ingestion_job.h
@@ -86,7 +86,8 @@ 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);
+  Status Prepare(const std::vector<std::string>& external_files_paths,
+                 SuperVersion* sv);
 
   // 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
@@ -119,7 +120,8 @@ 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);
+                             IngestedFileInfo* file_to_ingest,
+                             SuperVersion* sv);
 
   // Assign `file_to_ingest` the appropriate sequence number and  the lowest
   // possible level that it can be ingested to according to compaction_style.
diff --git a/db/forward_iterator.cc b/db/forward_iterator.cc
index 1401bb0ab..00cfffc49 100644
--- a/db/forward_iterator.cc
+++ b/db/forward_iterator.cc
@@ -33,14 +33,16 @@ class ForwardLevelIterator : public InternalIterator {
  public:
   ForwardLevelIterator(const ColumnFamilyData* const cfd,
                        const ReadOptions& read_options,
-                       const std::vector<FileMetaData*>& files)
+                       const std::vector<FileMetaData*>& files,
+                       const SliceTransform* prefix_extractor)
       : 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) {}
+        pinned_iters_mgr_(nullptr),
+        prefix_extractor_(prefix_extractor) {}
 
   ~ForwardLevelIterator() {
     // Reset current pointer
@@ -75,7 +77,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,
-        nullptr /* table_reader_ptr */, nullptr, false);
+        prefix_extractor_, nullptr /* table_reader_ptr */, nullptr, false);
     file_iter_->SetPinnedItersMgr(pinned_iters_mgr_);
     valid_ = false;
     if (!range_del_agg.IsEmpty()) {
@@ -188,6 +190,7 @@ 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,
@@ -196,7 +199,7 @@ ForwardIterator::ForwardIterator(DBImpl* db, const ReadOptions& read_options,
     : db_(db),
       read_options_(read_options),
       cfd_(cfd),
-      prefix_extractor_(cfd->ioptions()->prefix_extractor),
+      prefix_extractor_(current_sv->mutable_cf_options.prefix_extractor.get()),
       user_comparator_(cfd->user_comparator()),
       immutable_min_heap_(MinIterComparator(&cfd_->internal_comparator())),
       sv_(current_sv),
@@ -633,7 +636,8 @@ 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));
+        read_options_.ignore_range_deletions ? nullptr : &range_del_agg,
+        sv_->mutable_cf_options.prefix_extractor.get()));
   }
   BuildLevelIterators(vstorage);
   current_ = nullptr;
@@ -703,7 +707,8 @@ 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));
+        read_options_.ignore_range_deletions ? nullptr : &range_del_agg,
+        svnew->mutable_cf_options.prefix_extractor.get()));
   }
 
   for (auto* f : l0_iters_) {
@@ -744,8 +749,9 @@ 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));
+      level_iters_.push_back(new ForwardLevelIterator(
+          cfd_, read_options_, level_files,
+          sv_->mutable_cf_options.prefix_extractor.get()));
     }
   }
 }
@@ -760,7 +766,8 @@ 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 */);
+        l0_files[i]->fd, nullptr /* range_del_agg */,
+        sv_->mutable_cf_options.prefix_extractor.get());
     l0_iters_[i]->SetPinnedItersMgr(pinned_iters_mgr_);
   }
 
diff --git a/db/memtable.cc b/db/memtable.cc
index f2d2881d9..91559a596 100644
--- a/db/memtable.cc
+++ b/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_, ioptions.prefix_extractor, ioptions.info_log,
-          column_family_id)),
+          comparator_, &arena_, mutable_cf_options.prefix_extractor.get(),
+          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_(ioptions.prefix_extractor),
+      prefix_extractor_(mutable_cf_options.prefix_extractor.get()),
       flush_state_(FLUSH_NOT_REQUESTED),
       env_(ioptions.env),
       insert_with_hint_prefix_extractor_(
diff --git a/db/plain_table_db_test.cc b/db/plain_table_db_test.cc
index de1f67fe0..b2ccad5f6 100644
--- a/db/plain_table_db_test.cc
+++ b/db/plain_table_db_test.cc
@@ -262,11 +262,13 @@ 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),
+                         encoding_type, file_size, table_properties,
+                         prefix_extractor),
         expect_bloom_not_match_(expect_bloom_not_match) {
     Status s = MmapDataIfNeeded();
     EXPECT_TRUE(s.ok());
@@ -360,7 +362,8 @@ 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, expect_bloom_not_match_,
+        std::move(file), table_reader_options.ioptions,
+        table_reader_options.prefix_extractor, expect_bloom_not_match_,
         store_index_in_file_, column_family_id_, column_family_name_));
 
     *table = std::move(new_reader);
diff --git a/db/repair.cc b/db/repair.cc
index 62609481f..4e41960ff 100644
--- a/db/repair.cc
+++ b/db/repair.cc
@@ -501,7 +501,8 @@ class Repairer {
     if (status.ok()) {
       InternalIterator* iter = table_cache_->NewIterator(
           ReadOptions(), env_options_, cfd->internal_comparator(), t->meta.fd,
-          nullptr /* range_del_agg */);
+          nullptr /* range_del_agg */,
+          cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
       bool empty = true;
       ParsedInternalKey parsed;
       t->min_sequence = 0;
diff --git a/db/table_cache.cc b/db/table_cache.cc
index 2e7c58957..c38514cb0 100644
--- a/db/table_cache.cc
+++ b/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,
-    bool skip_filters, int level, bool prefetch_index_and_filter_in_cache,
-    bool for_compaction) {
+    const SliceTransform* prefix_extractor, 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_, env_options, internal_comparator,
-                           skip_filters, level),
+        TableReaderOptions(ioptions_, prefix_extractor, 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,6 +134,7 @@ 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,
@@ -154,7 +155,8 @@ 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,
-                       skip_filters, level, prefetch_index_and_filter_in_cache);
+                       prefix_extractor, skip_filters, level,
+                       prefetch_index_and_filter_in_cache);
     if (!s.ok()) {
       assert(table_reader == nullptr);
       RecordTick(ioptions_.statistics, NO_FILE_ERRORS);
@@ -175,9 +177,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, TableReader** table_reader_ptr,
-    HistogramImpl* file_read_hist, bool for_compaction, Arena* arena,
-    bool skip_filters, int level) {
+    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) {
   PERF_TIMER_GUARD(new_table_iterator_nanos);
 
   Status s;
@@ -210,7 +212,7 @@ InternalIterator* TableCache::NewIterator(
     s = GetTableReader(
         env_options, icomparator, fd, true /* sequential_mode */, readahead,
         !for_compaction /* record stats */, nullptr, &table_reader_unique_ptr,
-        false /* skip_filters */, level,
+        prefix_extractor, false /* skip_filters */, level,
         true /* prefetch_index_and_filter_in_cache */, for_compaction);
     if (s.ok()) {
       table_reader = table_reader_unique_ptr.release();
@@ -218,7 +220,7 @@ InternalIterator* TableCache::NewIterator(
   } else {
     table_reader = fd.table_reader;
     if (table_reader == nullptr) {
-      s = FindTable(env_options, icomparator, fd, &handle,
+      s = FindTable(env_options, icomparator, fd, &handle, prefix_extractor,
                     options.read_tier == kBlockCacheTier /* no_io */,
                     !for_compaction /* record read_stats */, file_read_hist,
                     skip_filters, level);
@@ -233,7 +235,8 @@ InternalIterator* TableCache::NewIterator(
         !options.table_filter(*table_reader->GetTableProperties())) {
       result = NewEmptyInternalIterator(arena);
     } else {
-      result = table_reader->NewIterator(options, arena, skip_filters);
+      result = table_reader->NewIterator(options, prefix_extractor, arena,
+                                         skip_filters);
     }
     if (create_new_table_reader) {
       assert(handle == nullptr);
@@ -276,12 +279,13 @@ 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) {
+    HistogramImpl* file_read_hist, bool skip_filters, int level,
+    const SliceTransform* prefix_extractor) {
   Status s;
   Cache::Handle* handle = nullptr;
   TableReader* table_reader = fd.table_reader;
   if (table_reader == nullptr) {
-    s = FindTable(env_options, icomparator, fd, &handle,
+    s = FindTable(env_options, icomparator, fd, &handle, prefix_extractor,
                   options.read_tier == kBlockCacheTier /* no_io */,
                   true /* record read_stats */, file_read_hist, skip_filters,
                   level);
@@ -313,8 +317,10 @@ InternalIterator* TableCache::NewRangeTombstoneIterator(
 Status TableCache::Get(const ReadOptions& options,
                        const InternalKeyComparator& internal_comparator,
                        const FileDescriptor& fd, const Slice& k,
-                       GetContext* get_context, HistogramImpl* file_read_hist,
-                       bool skip_filters, int level) {
+                       GetContext* get_context,
+                       const SliceTransform* prefix_extractor,
+                       HistogramImpl* file_read_hist, bool skip_filters,
+                       int level) {
   std::string* row_cache_entry = nullptr;
   bool done = false;
 #ifndef ROCKSDB_LITE
@@ -378,10 +384,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,
-                    options.read_tier == kBlockCacheTier /* no_io */,
-                    true /* record_read_stats */, file_read_hist, skip_filters,
-                    level);
+      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);
       if (s.ok()) {
         t = GetTableReaderFromHandle(handle);
       }
@@ -400,7 +406,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, skip_filters);
+      s = t->Get(options, k, get_context, prefix_extractor, 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
@@ -430,7 +436,8 @@ 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, bool no_io) {
+    std::shared_ptr<const TableProperties>* properties,
+    const SliceTransform* prefix_extractor, bool no_io) {
   Status s;
   auto table_reader = fd.table_reader;
   // table already been pre-loaded?
@@ -441,7 +448,8 @@ Status TableCache::GetTableProperties(
   }
 
   Cache::Handle* table_handle = nullptr;
-  s = FindTable(env_options, internal_comparator, fd, &table_handle, no_io);
+  s = FindTable(env_options, internal_comparator, fd, &table_handle,
+                prefix_extractor, no_io);
   if (!s.ok()) {
     return s;
   }
@@ -454,8 +462,8 @@ Status TableCache::GetTableProperties(
 
 size_t TableCache::GetMemoryUsageByTableReader(
     const EnvOptions& env_options,
-    const InternalKeyComparator& internal_comparator,
-    const FileDescriptor& fd) {
+    const InternalKeyComparator& internal_comparator, const FileDescriptor& fd,
+    const SliceTransform* prefix_extractor) {
   Status s;
   auto table_reader = fd.table_reader;
   // table already been pre-loaded?
@@ -464,7 +472,8 @@ size_t TableCache::GetMemoryUsageByTableReader(
   }
 
   Cache::Handle* table_handle = nullptr;
-  s = FindTable(env_options, internal_comparator, fd, &table_handle, true);
+  s = FindTable(env_options, internal_comparator, fd, &table_handle,
+                prefix_extractor, true);
   if (!s.ok()) {
     return 0;
   }
diff --git a/db/table_cache.h b/db/table_cache.h
index 8b65bafa3..2b550a869 100644
--- a/db/table_cache.h
+++ b/db/table_cache.h
@@ -54,6 +54,7 @@ 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);
@@ -62,7 +63,8 @@ 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);
+      bool skip_filters, int level,
+      const SliceTransform* prefix_extractor = nullptr);
 
   // If a seek to internal key "k" in specified file finds an entry,
   // call (*handle_result)(arg, found_key, found_value) repeatedly until
@@ -75,8 +77,10 @@ class TableCache {
   Status Get(const ReadOptions& options,
              const InternalKeyComparator& internal_comparator,
              const FileDescriptor& file_fd, const Slice& k,
-             GetContext* get_context, HistogramImpl* file_read_hist = nullptr,
-             bool skip_filters = false, int level = -1);
+             GetContext* get_context,
+             const SliceTransform* prefix_extractor = nullptr,
+             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);
@@ -91,6 +95,7 @@ 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,
@@ -109,6 +114,7 @@ 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.
@@ -116,7 +122,8 @@ class TableCache {
   size_t GetMemoryUsageByTableReader(
       const EnvOptions& toptions,
       const InternalKeyComparator& internal_comparator,
-      const FileDescriptor& fd);
+      const FileDescriptor& fd,
+      const SliceTransform* prefix_extractor = nullptr);
 
   // Release the handle from a cache
   void ReleaseHandle(Cache::Handle* handle);
@@ -133,6 +140,7 @@ 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);
diff --git a/db/table_properties_collector_test.cc b/db/table_properties_collector_test.cc
index bf382b4fd..6db6d371f 100644
--- a/db/table_properties_collector_test.cc
+++ b/db/table_properties_collector_test.cc
@@ -39,6 +39,7 @@ 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,
@@ -48,10 +49,9 @@ void MakeBuilder(const Options& options, const ImmutableCFOptions& ioptions,
   writable->reset(new WritableFileWriter(std::move(wf), EnvOptions()));
   int unknown_level = -1;
   builder->reset(NewTableBuilder(
-      ioptions, internal_comparator, int_tbl_prop_collector_factories,
-      kTestColumnFamilyId, kTestColumnFamilyName,
-      writable->get(), options.compression, options.compression_opts,
-      unknown_level));
+      ioptions, moptions, internal_comparator, int_tbl_prop_collector_factories,
+      kTestColumnFamilyId, kTestColumnFamilyName, writable->get(),
+      options.compression, options.compression_opts, unknown_level));
 }
 }  // namespace
 
@@ -251,6 +251,7 @@ 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) {
@@ -259,7 +260,7 @@ void TestCustomizedTablePropertiesCollector(
   } else {
     GetIntTblPropCollectorFactory(ioptions, &int_tbl_prop_collector_factories);
   }
-  MakeBuilder(options, ioptions, internal_comparator,
+  MakeBuilder(options, ioptions, moptions, internal_comparator,
               &int_tbl_prop_collector_factories, &writer, &builder);
 
   SequenceNumber seqNum = 0U;
@@ -401,10 +402,11 @@ void TestInternalKeyPropertiesCollector(
         new InternalKeyPropertiesCollectorFactory);
   }
   const ImmutableCFOptions ioptions(options);
+  MutableCFOptions moptions(options);
 
   for (int iter = 0; iter < 2; ++iter) {
-    MakeBuilder(options, ioptions, pikc, &int_tbl_prop_collector_factories,
-                &writable, &builder);
+    MakeBuilder(options, ioptions, moptions, pikc,
+                &int_tbl_prop_collector_factories, &writable, &builder);
     for (const auto& k : keys) {
       builder->Add(k.Encode(), "val");
     }
diff --git a/db/version_builder.cc b/db/version_builder.cc
index 6507b8e2f..2d381ec61 100644
--- a/db/version_builder.cc
+++ b/db/version_builder.cc
@@ -368,7 +368,8 @@ class VersionBuilder::Rep {
   }
 
   void LoadTableHandlers(InternalStats* internal_stats, int max_threads,
-                         bool prefetch_index_and_filter_in_cache) {
+                         bool prefetch_index_and_filter_in_cache,
+                         const SliceTransform* prefix_extractor) {
     assert(table_cache_ != nullptr);
     // <file metadata, level>
     std::vector<std::pair<FileMetaData*, int>> files_meta;
@@ -390,12 +391,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,
-                                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, prefix_extractor,
+            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(
@@ -455,11 +456,12 @@ void VersionBuilder::SaveTo(VersionStorageInfo* vstorage) {
   rep_->SaveTo(vstorage);
 }
 
-void VersionBuilder::LoadTableHandlers(
-    InternalStats* internal_stats, int max_threads,
-    bool prefetch_index_and_filter_in_cache) {
+void VersionBuilder::LoadTableHandlers(InternalStats* internal_stats,
+                                       int max_threads,
+                                       bool prefetch_index_and_filter_in_cache,
+                                       const SliceTransform* prefix_extractor) {
   rep_->LoadTableHandlers(internal_stats, max_threads,
-                          prefetch_index_and_filter_in_cache);
+                          prefetch_index_and_filter_in_cache, prefix_extractor);
 }
 
 void VersionBuilder::MaybeAddFile(VersionStorageInfo* vstorage, int level,
diff --git a/db/version_builder.h b/db/version_builder.h
index 440d4eaf6..4d32aec2b 100644
--- a/db/version_builder.h
+++ b/db/version_builder.h
@@ -9,6 +9,7 @@
 //
 #pragma once
 #include "rocksdb/env.h"
+#include "rocksdb/slice_transform.h"
 
 namespace rocksdb {
 
@@ -33,7 +34,8 @@ 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);
+                         bool prefetch_index_and_filter_in_cache,
+                         const SliceTransform* prefix_extractor);
   void MaybeAddFile(VersionStorageInfo* vstorage, int level, FileMetaData* f);
 
  private:
diff --git a/db/version_set.cc b/db/version_set.cc
index d1a63bce6..b14bea897 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -463,7 +463,8 @@ class LevelIterator final : public InternalIterator {
   LevelIterator(TableCache* table_cache, const ReadOptions& read_options,
                 const EnvOptions& env_options,
                 const InternalKeyComparator& icomparator,
-                const LevelFilesBrief* flevel, bool should_sample,
+                const LevelFilesBrief* flevel,
+                const SliceTransform* prefix_extractor, bool should_sample,
                 HistogramImpl* file_read_hist, bool for_compaction,
                 bool skip_filters, int level, RangeDelAggregator* range_del_agg)
       : table_cache_(table_cache),
@@ -471,6 +472,7 @@ 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),
@@ -547,8 +549,9 @@ class LevelIterator final : public InternalIterator {
 
     return table_cache_->NewIterator(
         read_options_, env_options_, icomparator_, file_meta.fd, range_del_agg_,
-        nullptr /* don't need reference to table */, file_read_hist_,
-        for_compaction_, nullptr /* arena */, skip_filters_, level_);
+        prefix_extractor_, nullptr /* don't need reference to table */,
+        file_read_hist_, for_compaction_, nullptr /* arena */, skip_filters_,
+        level_);
   }
 
   TableCache* table_cache_;
@@ -557,6 +560,7 @@ 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_;
@@ -722,8 +726,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, true /* no io */);
+      env_options_, cfd_->internal_comparator(), file_meta->fd, tp,
+      mutable_cf_options_.prefix_extractor.get(), true /* no io */);
   if (s.ok()) {
     return s;
   }
@@ -857,8 +861,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);
+          env_options_, cfd_->internal_comparator(), file_level.files[i].fd,
+          mutable_cf_options_.prefix_extractor.get());
     }
   }
   return total_usage;
@@ -996,8 +1000,9 @@ 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, nullptr, cfd_->internal_stats()->GetFileReadHist(0),
-          false, arena, false /* skip_filters */, 0 /* level */));
+          range_del_agg, mutable_cf_options_.prefix_extractor.get(), 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
@@ -1016,7 +1021,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),
-        should_sample_file_read(),
+        mutable_cf_options_.prefix_extractor.get(), should_sample_file_read(),
         cfd_->internal_stats()->GetFileReadHist(level),
         false /* for_compaction */, IsFilterSkipped(level), level,
         range_del_agg));
@@ -1048,8 +1053,9 @@ 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, nullptr, cfd_->internal_stats()->GetFileReadHist(0),
-          false, &arena, false /* skip_filters */, 0 /* level */));
+          &range_del_agg, mutable_cf_options_.prefix_extractor.get(), 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) {
@@ -1061,7 +1067,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),
-        should_sample_file_read(),
+        mutable_cf_options_.prefix_extractor.get(), should_sample_file_read(),
         cfd_->internal_stats()->GetFileReadHist(level),
         false /* for_compaction */, IsFilterSkipped(level), level,
         &range_del_agg));
@@ -1122,7 +1128,9 @@ VersionStorageInfo::VersionStorageInfo(
 }
 
 Version::Version(ColumnFamilyData* column_family_data, VersionSet* vset,
-                 const EnvOptions& env_opt, uint64_t version_number)
+                 const EnvOptions& env_opt,
+                 const MutableCFOptions mutable_cf_options,
+                 uint64_t version_number)
     : env_(vset->env_),
       cfd_(column_family_data),
       info_log_((cfd_ == nullptr) ? nullptr : cfd_->ioptions()->info_log),
@@ -1146,6 +1154,7 @@ 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,
@@ -1189,6 +1198,7 @@ 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()),
@@ -2775,7 +2785,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_,
+    v = new Version(column_family_data, this, env_options_, mutable_cf_options,
                     current_version_number_++);
     builder_guard.reset(new BaseReferencedVersionBuilder(column_family_data));
     auto* builder = builder_guard->version_builder();
@@ -2836,7 +2846,8 @@ 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 */);
+          true /* prefetch_index_and_filter_in_cache */,
+          mutable_cf_options.prefix_extractor.get());
     }
 
     // This is fine because everything inside of this block is serialized --
@@ -3327,11 +3338,13 @@ 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 */);
+            false /* prefetch_index_and_filter_in_cache */,
+            cfd->GetLatestMutableCFOptions()->prefix_extractor.get());
       }
 
-      Version* v =
-          new Version(cfd, this, env_options_, current_version_number_++);
+      Version* v = new Version(cfd, this, env_options_,
+                               *cfd->GetLatestMutableCFOptions(),
+                               current_version_number_++);
       builder->SaveTo(v->storage_info());
 
       // Install recovered version
@@ -3696,8 +3709,9 @@ 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_, current_version_number_++);
+      Version* v = new Version(cfd, this, env_options_,
+                               *cfd->GetLatestMutableCFOptions(),
+                               current_version_number_++);
       builder->SaveTo(v->storage_info());
       v->PrepareApply(*cfd->GetLatestMutableCFOptions(), false);
 
@@ -3920,7 +3934,8 @@ 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 */, &table_reader_ptr);
+        nullptr /* range_del_agg */,
+        v->GetMutableCFOptions().prefix_extractor.get(), &table_reader_ptr);
     if (table_reader_ptr != nullptr) {
       result = table_reader_ptr->ApproximateOffsetOf(key);
     }
@@ -4000,6 +4015,7 @@ 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 */,
@@ -4010,6 +4026,7 @@ 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 */,
@@ -4149,7 +4166,9 @@ ColumnFamilyData* VersionSet::CreateColumnFamily(
     const ColumnFamilyOptions& cf_options, VersionEdit* edit) {
   assert(edit->is_column_family_add_);
 
-  Version* dummy_versions = new Version(nullptr, this, env_options_);
+  MutableCFOptions dummy_cf_options;
+  Version* dummy_versions =
+      new Version(nullptr, this, env_options_, dummy_cf_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();
@@ -4157,8 +4176,9 @@ ColumnFamilyData* VersionSet::CreateColumnFamily(
       edit->column_family_name_, edit->column_family_, dummy_versions,
       cf_options);
 
-  Version* v =
-      new Version(new_cfd, this, env_options_, current_version_number_++);
+  Version* v = new Version(new_cfd, this, env_options_,
+                           *new_cfd->GetLatestMutableCFOptions(),
+                           current_version_number_++);
 
   // Fill level target base information.
   v->storage_info()->CalculateBaseBytes(*new_cfd->ioptions(),
diff --git a/db/version_set.h b/db/version_set.h
index 563054d17..0a5fa504d 100644
--- a/db/version_set.h
+++ b/db/version_set.h
@@ -633,6 +633,8 @@ class Version {
 
   uint64_t GetSstFilesSize();
 
+  MutableCFOptions GetMutableCFOptions() { return mutable_cf_options_; }
+
  private:
   Env* env_;
   friend class VersionSet;
@@ -680,13 +682,14 @@ 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,
-          uint64_t version_number = 0);
+          MutableCFOptions mutable_cf_options, uint64_t version_number = 0);
 
   ~Version();
 
diff --git a/options/cf_options.cc b/options/cf_options.cc
index bc0195e41..e5d62c24f 100644
--- a/options/cf_options.cc
+++ b/options/cf_options.cc
@@ -27,7 +27,6 @@ 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()),
@@ -143,6 +142,9 @@ 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,
@@ -189,4 +191,7 @@ void MutableCFOptions::Dump(Logger* log) const {
                  static_cast<int>(compression));
 }
 
+MutableCFOptions::MutableCFOptions(const Options& options)
+    : MutableCFOptions(ColumnFamilyOptions(options)) {}
+
 }  // namespace rocksdb
diff --git a/options/cf_options.h b/options/cf_options.h
index f54224230..1974ba6d7 100644
--- a/options/cf_options.h
+++ b/options/cf_options.h
@@ -30,8 +30,6 @@ struct ImmutableCFOptions {
 
   CompactionPri compaction_pri;
 
-  const SliceTransform* prefix_extractor;
-
   const Comparator* user_comparator;
   InternalKeyComparator internal_comparator;
 
@@ -134,6 +132,7 @@ 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),
@@ -168,6 +167,7 @@ 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,6 +185,8 @@ 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);
 
@@ -210,6 +212,7 @@ 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;
diff --git a/options/options_helper.cc b/options/options_helper.cc
index 6d412b5ee..eba08cb85 100644
--- a/options/options_helper.cc
+++ b/options/options_helper.cc
@@ -145,6 +145,7 @@ 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 =
@@ -383,7 +384,8 @@ 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) {
@@ -395,6 +397,10 @@ 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 {
@@ -1791,7 +1797,7 @@ std::unordered_map<std::string, OptionTypeInfo>
         {"prefix_extractor",
          {offset_of(&ColumnFamilyOptions::prefix_extractor),
           OptionType::kSliceTransform, OptionVerificationType::kByNameAllowNull,
-          false, 0}},
+          true, offsetof(struct MutableCFOptions, prefix_extractor)}},
         {"memtable_insert_with_hint_prefix_extractor",
          {offset_of(
               &ColumnFamilyOptions::memtable_insert_with_hint_prefix_extractor),
diff --git a/table/block_based_filter_block.cc b/table/block_based_filter_block.cc
index fc24f41a6..ffd37f01d 100644
--- a/table/block_based_filter_block.cc
+++ b/table/block_based_filter_block.cc
@@ -186,7 +186,8 @@ BlockBasedFilterBlockReader::BlockBasedFilterBlockReader(
 }
 
 bool BlockBasedFilterBlockReader::KeyMayMatch(
-    const Slice& key, uint64_t block_offset, const bool /*no_io*/,
+    const Slice& key, const SliceTransform* /* prefix_extractor */,
+    uint64_t block_offset, const bool /*no_io*/,
     const Slice* const /*const_ikey_ptr*/) {
   assert(block_offset != kNotValid);
   if (!whole_key_filtering_) {
@@ -196,7 +197,8 @@ bool BlockBasedFilterBlockReader::KeyMayMatch(
 }
 
 bool BlockBasedFilterBlockReader::PrefixMayMatch(
-    const Slice& prefix, uint64_t block_offset, const bool /*no_io*/,
+    const Slice& prefix, const SliceTransform* /* prefix_extractor */,
+    uint64_t block_offset, const bool /*no_io*/,
     const Slice* const /*const_ikey_ptr*/) {
   assert(block_offset != kNotValid);
   if (!prefix_extractor_) {
diff --git a/table/block_based_filter_block.h b/table/block_based_filter_block.h
index 3bfb3b24a..96a75e361 100644
--- a/table/block_based_filter_block.h
+++ b/table/block_based_filter_block.h
@@ -83,13 +83,14 @@ class BlockBasedFilterBlockReader : public FilterBlockReader {
                               bool whole_key_filtering,
                               BlockContents&& contents, Statistics* statistics);
   virtual bool IsBlockBased() override { return true; }
+
   virtual bool KeyMayMatch(
-      const Slice& key, uint64_t block_offset = kNotValid,
-      const bool no_io = false,
+      const Slice& key, const SliceTransform* prefix_extractor,
+      uint64_t block_offset = kNotValid, const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual bool PrefixMayMatch(
-      const Slice& prefix, uint64_t block_offset = kNotValid,
-      const bool no_io = false,
+      const Slice& prefix, const SliceTransform* prefix_extractor,
+      uint64_t block_offset = kNotValid, const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual size_t ApproximateMemoryUsage() const override;
 
diff --git a/table/block_based_filter_block_test.cc b/table/block_based_filter_block_test.cc
index dece461e3..8de857f4e 100644
--- a/table/block_based_filter_block_test.cc
+++ b/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", 0));
-  ASSERT_TRUE(reader.KeyMayMatch("foo", 100000));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr, uint64_t{0}));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr, 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", 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));
+  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));
 }
 
 TEST_F(FilterBlockTest, MultiChunk) {
@@ -112,28 +112,28 @@ TEST_F(FilterBlockTest, MultiChunk) {
                                      std::move(block), nullptr);
 
   // Check first filter
-  ASSERT_TRUE(reader.KeyMayMatch("foo", 0));
-  ASSERT_TRUE(reader.KeyMayMatch("bar", 2000));
-  ASSERT_TRUE(!reader.KeyMayMatch("box", 0));
-  ASSERT_TRUE(!reader.KeyMayMatch("hello", 0));
+  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}));
 
   // Check second filter
-  ASSERT_TRUE(reader.KeyMayMatch("box", 3100));
-  ASSERT_TRUE(!reader.KeyMayMatch("foo", 3100));
-  ASSERT_TRUE(!reader.KeyMayMatch("bar", 3100));
-  ASSERT_TRUE(!reader.KeyMayMatch("hello", 3100));
+  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));
 
   // Check third filter (empty)
-  ASSERT_TRUE(!reader.KeyMayMatch("foo", 4100));
-  ASSERT_TRUE(!reader.KeyMayMatch("bar", 4100));
-  ASSERT_TRUE(!reader.KeyMayMatch("box", 4100));
-  ASSERT_TRUE(!reader.KeyMayMatch("hello", 4100));
+  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));
 
   // Check last filter
-  ASSERT_TRUE(reader.KeyMayMatch("box", 9000));
-  ASSERT_TRUE(reader.KeyMayMatch("hello", 9000));
-  ASSERT_TRUE(!reader.KeyMayMatch("foo", 9000));
-  ASSERT_TRUE(!reader.KeyMayMatch("bar", 9000));
+  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));
 }
 
 // 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", 0));
-  ASSERT_TRUE(reader->KeyMayMatch("foo", 100000));
+  ASSERT_TRUE(reader->KeyMayMatch("foo", nullptr, uint64_t{0}));
+  ASSERT_TRUE(reader->KeyMayMatch("foo", nullptr, 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", 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));
+  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));
 
   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", 0));
-  ASSERT_TRUE(reader->KeyMayMatch("bar", 2000));
-  ASSERT_TRUE(!reader->KeyMayMatch("box", 0));
-  ASSERT_TRUE(!reader->KeyMayMatch("hello", 0));
+  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}));
 
   // Check second filter
-  ASSERT_TRUE(reader->KeyMayMatch("box", 3100));
-  ASSERT_TRUE(!reader->KeyMayMatch("foo", 3100));
-  ASSERT_TRUE(!reader->KeyMayMatch("bar", 3100));
-  ASSERT_TRUE(!reader->KeyMayMatch("hello", 3100));
+  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));
 
   // Check third filter (empty)
-  ASSERT_TRUE(!reader->KeyMayMatch("foo", 4100));
-  ASSERT_TRUE(!reader->KeyMayMatch("bar", 4100));
-  ASSERT_TRUE(!reader->KeyMayMatch("box", 4100));
-  ASSERT_TRUE(!reader->KeyMayMatch("hello", 4100));
+  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));
 
   // Check last filter
-  ASSERT_TRUE(reader->KeyMayMatch("box", 9000));
-  ASSERT_TRUE(reader->KeyMayMatch("hello", 9000));
-  ASSERT_TRUE(!reader->KeyMayMatch("foo", 9000));
-  ASSERT_TRUE(!reader->KeyMayMatch("bar", 9000));
+  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));
 
   delete builder;
   delete reader;
diff --git a/table/block_based_table_builder.cc b/table/block_based_table_builder.cc
index 85c41fe51..8c4ced1dd 100644
--- a/table/block_based_table_builder.cc
+++ b/table/block_based_table_builder.cc
@@ -62,14 +62,16 @@ namespace {
 
 // Create a filter block builder based on its type.
 FilterBlockBuilder* CreateFilterBlockBuilder(
-    const ImmutableCFOptions& opt, const BlockBasedTableOptions& table_opt,
+    const ImmutableCFOptions& /*opt*/, const MutableCFOptions& mopt,
+    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(opt.prefix_extractor, table_opt);
+    return new BlockBasedFilterBlockBuilder(mopt.prefix_extractor.get(),
+                                            table_opt);
   } else {
     if (table_opt.partition_filters) {
       assert(p_index_builder != nullptr);
@@ -82,11 +84,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(
-          opt.prefix_extractor, table_opt.whole_key_filtering,
+          mopt.prefix_extractor.get(), table_opt.whole_key_filtering,
           filter_bits_builder, table_opt.index_block_restart_interval,
           p_index_builder, partition_size);
     } else {
-      return new FullFilterBlockBuilder(opt.prefix_extractor,
+      return new FullFilterBlockBuilder(mopt.prefix_extractor.get(),
                                         table_opt.whole_key_filtering,
                                         filter_bits_builder);
     }
@@ -244,6 +246,7 @@ class BlockBasedTableBuilder::BlockBasedTablePropertiesCollector
 
 struct BlockBasedTableBuilder::Rep {
   const ImmutableCFOptions ioptions;
+  const MutableCFOptions moptions;
   const BlockBasedTableOptions table_options;
   const InternalKeyComparator& internal_comparator;
   WritableFileWriter* file;
@@ -280,7 +283,7 @@ struct BlockBasedTableBuilder::Rep {
 
   std::vector<std::unique_ptr<IntTblPropCollector>> table_properties_collectors;
 
-  Rep(const ImmutableCFOptions& _ioptions,
+  Rep(const ImmutableCFOptions& _ioptions, const MutableCFOptions& _moptions,
       const BlockBasedTableOptions& table_opt,
       const InternalKeyComparator& icomparator,
       const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
@@ -292,6 +295,7 @@ 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),
@@ -301,7 +305,7 @@ struct BlockBasedTableBuilder::Rep {
         data_block(table_options.block_restart_interval,
                    table_options.use_delta_encoding),
         range_del_block(1 /* block_restart_interval */),
-        internal_prefix_transform(_ioptions.prefix_extractor),
+        internal_prefix_transform(_moptions.prefix_extractor.get()),
         compression_type(_compression_type),
         compression_opts(_compression_opts),
         compression_dict(_compression_dict),
@@ -326,8 +330,8 @@ struct BlockBasedTableBuilder::Rep {
     if (skip_filters) {
       filter_builder = nullptr;
     } else {
-      filter_builder.reset(
-          CreateFilterBlockBuilder(_ioptions, table_options, p_index_builder_));
+      filter_builder.reset(CreateFilterBlockBuilder(
+          _ioptions, _moptions, table_options, p_index_builder_));
     }
 
     for (auto& collector_factories : *int_tbl_prop_collector_factories) {
@@ -337,12 +341,12 @@ struct BlockBasedTableBuilder::Rep {
     table_properties_collectors.emplace_back(
         new BlockBasedTablePropertiesCollector(
             table_options.index_type, table_options.whole_key_filtering,
-            _ioptions.prefix_extractor != nullptr));
+            _moptions.prefix_extractor != nullptr));
   }
 };
 
 BlockBasedTableBuilder::BlockBasedTableBuilder(
-    const ImmutableCFOptions& ioptions,
+    const ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
     const BlockBasedTableOptions& table_options,
     const InternalKeyComparator& internal_comparator,
     const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
@@ -365,11 +369,11 @@ BlockBasedTableBuilder::BlockBasedTableBuilder(
     sanitized_table_options.format_version = 1;
   }
 
-  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);
+  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);
 
   if (rep_->filter_builder != nullptr) {
     rep_->filter_builder->StartBlock(0);
@@ -737,8 +741,8 @@ Status BlockBasedTableBuilder::Finish() {
                                          : "nullptr";
       r->props.compression_name = CompressionTypeToString(r->compression_type);
       r->props.prefix_extractor_name =
-          r->ioptions.prefix_extractor != nullptr
-              ? r->ioptions.prefix_extractor->Name()
+          r->moptions.prefix_extractor != nullptr
+              ? r->moptions.prefix_extractor->Name()
               : "nullptr";
 
       std::string property_collectors_names = "[";
diff --git a/table/block_based_table_builder.h b/table/block_based_table_builder.h
index 392dedc1f..a0ba87f14 100644
--- a/table/block_based_table_builder.h
+++ b/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 ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
       const BlockBasedTableOptions& table_options,
       const InternalKeyComparator& internal_comparator,
       const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
diff --git a/table/block_based_table_factory.cc b/table/block_based_table_factory.cc
index 1896041f4..d53b9833c 100644
--- a/table/block_based_table_factory.cc
+++ b/table/block_based_table_factory.cc
@@ -69,16 +69,17 @@ 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, prefetch_index_and_filter_in_cache,
-      table_reader_options.skip_filters, table_reader_options.level);
+      file_size, table_reader, table_reader_options.prefix_extractor,
+      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_options_,
-      table_builder_options.internal_comparator,
+      table_builder_options.ioptions, table_builder_options.moptions,
+      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,
diff --git a/table/block_based_table_reader.cc b/table/block_based_table_reader.cc
index e3c218474..e719ee31d 100644
--- a/table/block_based_table_reader.cc
+++ b/table/block_based_table_reader.cc
@@ -173,6 +173,25 @@ 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::string prefix_extractor_block,
+                            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 || prefix_extractor_block.empty()) {
+    return true;
+  }
+  // prefix_extractor and prefix_extractor_block are both non-empty
+  if (prefix_extractor_block.compare(prefix_extractor->Name()) != 0) {
+    return true;
+  } else {
+    return false;
+  }
+}
+
 }  // namespace
 
 // Index that allows binary search lookup in a two-level index structure.
@@ -221,7 +240,8 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
       ro.fill_cache = fill_cache;
       return new BlockBasedTableIterator(
           table_, ro, *icomparator_,
-          index_block_->NewIterator(icomparator_, nullptr, true), false);
+          index_block_->NewIterator(icomparator_, nullptr, true), false,
+          /* prefix_extractor */ nullptr);
     }
     // TODO(myabandeh): Update TwoLevelIterator to be able to make use of
     // on-stack BlockIter while the state is on heap. Currentlly it assumes
@@ -646,6 +666,7 @@ 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();
@@ -697,7 +718,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(rep->ioptions.prefix_extractor));
+      new InternalKeySliceTransform(prefix_extractor));
   SetupCacheKeyPrefix(rep, file_size);
   unique_ptr<BlockBasedTable> new_table(new BlockBasedTable(rep));
 
@@ -863,8 +884,14 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
       // block_cache
 
       CachableEntry<IndexReader> index_entry;
-      unique_ptr<InternalIterator> iter(
-          new_table->NewIndexIterator(ReadOptions(), nullptr, &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_name, prefix_extractor);
+      }
+      unique_ptr<InternalIterator> iter(new_table->NewIndexIterator(
+          ReadOptions(), prefix_extractor_changed, nullptr, &index_entry));
       s = iter->status();
       if (s.ok()) {
         // This is the first call to NewIndexIterator() since we're in Open().
@@ -879,9 +906,9 @@ Status BlockBasedTable::Open(const ImmutableCFOptions& ioptions,
         }
 
         // Hack: Call GetFilter() to implicitly add filter to the block_cache
-        auto filter_entry = new_table->GetFilter();
+        auto filter_entry = new_table->GetFilter(prefix_extractor);
         if (filter_entry.value != nullptr) {
-          filter_entry.value->CacheDependencies(pin);
+          filter_entry.value->CacheDependencies(pin, prefix_extractor);
         }
         // 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
@@ -913,13 +940,14 @@ 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);
+        auto filter =
+            new_table->ReadFilter(prefetch_buffer.get(), rep->filter_handle,
+                                  !is_a_filter_partition, prefix_extractor);
         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);
+          filter->CacheDependencies(pin, prefix_extractor);
         }
       }
     } else {
@@ -1215,7 +1243,8 @@ Status BlockBasedTable::PutDataBlockToCache(
 
 FilterBlockReader* BlockBasedTable::ReadFilter(
     FilePrefetchBuffer* prefetch_buffer, const BlockHandle& filter_handle,
-    const bool is_a_filter_partition) const {
+    const bool is_a_filter_partition,
+    const SliceTransform* prefix_extractor) const {
   auto& rep = rep_;
   // TODO: We might want to unify with ReadBlockFromFile() if we start
   // requiring checksum verification in Table::Open.
@@ -1248,14 +1277,14 @@ FilterBlockReader* BlockBasedTable::ReadFilter(
   switch (filter_type) {
     case Rep::FilterType::kPartitionedFilter: {
       return new PartitionedFilterBlockReader(
-          rep->prefix_filtering ? rep->ioptions.prefix_extractor : nullptr,
+          rep->prefix_filtering ? 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 ? rep->ioptions.prefix_extractor : nullptr,
+          rep->prefix_filtering ? prefix_extractor : nullptr,
           rep->table_options, rep->whole_key_filtering, std::move(block),
           rep->ioptions.statistics);
 
@@ -1264,7 +1293,7 @@ FilterBlockReader* BlockBasedTable::ReadFilter(
           rep->filter_policy->GetFilterBitsReader(block.data);
       assert(filter_bits_reader != nullptr);
       return new FullFilterBlockReader(
-          rep->prefix_filtering ? rep->ioptions.prefix_extractor : nullptr,
+          rep->prefix_filtering ? prefix_extractor : nullptr,
           rep->whole_key_filtering, std::move(block), filter_bits_reader,
           rep->ioptions.statistics);
     }
@@ -1278,18 +1307,18 @@ FilterBlockReader* BlockBasedTable::ReadFilter(
 }
 
 BlockBasedTable::CachableEntry<FilterBlockReader> BlockBasedTable::GetFilter(
-    FilePrefetchBuffer* prefetch_buffer, bool no_io,
-    GetContext* get_context) const {
+    const SliceTransform* prefix_extractor, 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);
+                   no_io, get_context, prefix_extractor);
 }
 
 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 {
+    const bool is_a_filter_partition, bool no_io, GetContext* get_context,
+    const SliceTransform* prefix_extractor) 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
@@ -1329,8 +1358,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);
+    filter = ReadFilter(prefetch_buffer, filter_blk_handle,
+                        is_a_filter_partition, prefix_extractor);
     if (filter != nullptr) {
       Status s = block_cache->Insert(
           key, filter, filter->size(), &DeleteCachedFilterEntry, &cache_handle,
@@ -1362,18 +1391,23 @@ 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, BlockIter* input_iter,
-    CachableEntry<IndexReader>* index_entry, GetContext* get_context) {
+    const ReadOptions& read_options, bool disable_prefix_seek,
+    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, read_options.fill_cache);
+        input_iter, read_options.total_order_seek || disable_prefix_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, read_options.fill_cache);
+        input_iter, read_options.total_order_seek || disable_prefix_seek,
+        read_options.fill_cache);
   }
 
   PERF_TIMER_GUARD(read_index_block_nanos);
@@ -1449,7 +1483,7 @@ InternalIterator* BlockBasedTable::NewIndexIterator(
 
   assert(cache_handle);
   auto* iter = index_reader->NewIterator(
-      input_iter, read_options.total_order_seek);
+      input_iter, read_options.total_order_seek || disable_prefix_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
@@ -1682,31 +1716,32 @@ 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) {
+bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key,
+                                     const SliceTransform* prefix_extractor) {
   if (!rep_->filter_policy) {
     return true;
   }
 
-  assert(rep_->ioptions.prefix_extractor != nullptr);
+  assert(prefix_extractor != nullptr);
   auto user_key = ExtractUserKey(internal_key);
-  if (!rep_->ioptions.prefix_extractor->InDomain(user_key) ||
-      rep_->table_properties->prefix_extractor_name.compare(
-          rep_->ioptions.prefix_extractor->Name()) != 0) {
+  if (!prefix_extractor->InDomain(user_key)) {
     return true;
   }
-  auto prefix = rep_->ioptions.prefix_extractor->Transform(user_key);
+  assert(rep_->table_properties->prefix_extractor_name.compare(
+             prefix_extractor->Name()) == 0);
+  auto prefix = prefix_extractor->Transform(user_key);
 
   bool may_match = true;
   Status s;
 
   // First, try check with full filter
-  auto filter_entry = GetFilter();
+  auto filter_entry = GetFilter(prefix_extractor);
   FilterBlockReader* filter = filter_entry.value;
   if (filter != nullptr) {
     if (!filter->IsBlockBased()) {
       const Slice* const const_ikey_ptr = &internal_key;
-      may_match =
-          filter->PrefixMayMatch(prefix, kNotValid, false, const_ikey_ptr);
+      may_match = filter->PrefixMayMatch(prefix, prefix_extractor, kNotValid,
+                                         false, const_ikey_ptr);
     } else {
       InternalKey internal_key_prefix(prefix, kMaxSequenceNumber, kTypeValue);
       auto internal_prefix = internal_key_prefix.Encode();
@@ -1718,7 +1753,11 @@ bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key) {
       no_io_read_options.read_tier = kBlockCacheTier;
 
       // Then, try find it within each block
-      unique_ptr<InternalIterator> iiter(NewIndexIterator(no_io_read_options));
+      // 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));
       iiter->Seek(internal_prefix);
 
       if (!iiter->Valid()) {
@@ -1750,7 +1789,8 @@ bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key) {
         BlockHandle handle;
         s = handle.DecodeFrom(&handle_value);
         assert(s.ok());
-        may_match = filter->PrefixMayMatch(prefix, handle.offset());
+        may_match =
+            filter->PrefixMayMatch(prefix, prefix_extractor, handle.offset());
       }
     }
   }
@@ -1772,7 +1812,7 @@ bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key) {
 }
 
 void BlockBasedTableIterator::Seek(const Slice& target) {
-  if (!CheckPrefixMayMatch(target)) {
+  if (!CheckPrefixMayMatch(target, prefix_extractor_)) {
     ResetDataIter();
     return;
   }
@@ -1796,7 +1836,7 @@ void BlockBasedTableIterator::Seek(const Slice& target) {
 }
 
 void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
-  if (!CheckPrefixMayMatch(target)) {
+  if (!CheckPrefixMayMatch(target, prefix_extractor_)) {
     ResetDataIter();
     return;
   }
@@ -1975,22 +2015,29 @@ void BlockBasedTableIterator::FindKeyBackward() {
   // code simplicity.
 }
 
-InternalIterator* BlockBasedTable::NewIterator(const ReadOptions& read_options,
-                                               Arena* arena,
-                                               bool skip_filters) {
+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_name, prefix_extractor);
   if (arena == nullptr) {
     return new BlockBasedTableIterator(
         this, read_options, rep_->internal_comparator,
-        NewIndexIterator(read_options),
+        NewIndexIterator(
+            read_options,
+            prefix_extractor_changed &&
+            rep_->index_type == BlockBasedTableOptions::kHashSearch),
         !skip_filters && !read_options.total_order_seek &&
-            rep_->ioptions.prefix_extractor != nullptr);
+        prefix_extractor != nullptr && !prefix_extractor_changed,
+        prefix_extractor);
   } else {
     auto* mem = arena->AllocateAligned(sizeof(BlockBasedTableIterator));
     return new (mem) BlockBasedTableIterator(
         this, read_options, rep_->internal_comparator,
-        NewIndexIterator(read_options),
+        NewIndexIterator(read_options, prefix_extractor_changed),
         !skip_filters && !read_options.total_order_seek &&
-            rep_->ioptions.prefix_extractor != nullptr);
+        prefix_extractor != nullptr && !prefix_extractor_changed,
+        prefix_extractor);
   }
 }
 
@@ -2024,10 +2071,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 {
+bool BlockBasedTable::FullFilterKeyMayMatch(
+    const ReadOptions& read_options, FilterBlockReader* filter,
+    const Slice& internal_key, const bool no_io,
+    const SliceTransform* prefix_extractor) const {
   if (filter == nullptr || filter->IsBlockBased()) {
     return true;
   }
@@ -2035,15 +2082,15 @@ bool BlockBasedTable::FullFilterKeyMayMatch(const ReadOptions& read_options,
   const Slice* const const_ikey_ptr = &internal_key;
   bool may_match = true;
   if (filter->whole_key_filtering()) {
-    may_match = filter->KeyMayMatch(user_key, kNotValid, no_io, const_ikey_ptr);
-  } else if (!read_options.total_order_seek &&
-             rep_->ioptions.prefix_extractor &&
+    may_match = filter->KeyMayMatch(user_key, prefix_extractor, kNotValid,
+                                    no_io, const_ikey_ptr);
+  } else if (!read_options.total_order_seek && prefix_extractor &&
              rep_->table_properties->prefix_extractor_name.compare(
-                 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)) {
+                 prefix_extractor->Name()) == 0 &&
+             prefix_extractor->InDomain(user_key) &&
+             !filter->PrefixMayMatch(prefix_extractor->Transform(user_key),
+                                     prefix_extractor, kNotValid, false,
+                                     const_ikey_ptr)) {
     may_match = false;
   }
   if (may_match) {
@@ -2053,25 +2100,36 @@ bool BlockBasedTable::FullFilterKeyMayMatch(const ReadOptions& read_options,
 }
 
 Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
-                            GetContext* get_context, bool skip_filters) {
+                            GetContext* get_context,
+                            const SliceTransform* prefix_extractor,
+                            bool skip_filters) {
   Status s;
   const bool no_io = read_options.read_tier == kBlockCacheTier;
   CachableEntry<FilterBlockReader> filter_entry;
   if (!skip_filters) {
     filter_entry =
-        GetFilter(/*prefetch_buffer*/ nullptr,
+        GetFilter(prefix_extractor, /*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)) {
+  if (!FullFilterKeyMayMatch(read_options, filter, key, no_io,
+                             prefix_extractor)) {
     RecordTick(rep_->ioptions.statistics, BLOOM_FILTER_USEFUL);
   } else {
     BlockIter iiter_on_stack;
-    auto iiter = NewIndexIterator(read_options, &iiter_on_stack,
-                                  /* index_entry */ nullptr, get_context);
+    // 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_name, prefix_extractor);
+    }
+    auto iiter = NewIndexIterator(read_options, prefix_extractor_changed,
+                                  &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);
@@ -2086,7 +2144,8 @@ 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), handle.offset(), no_io);
+          !filter->KeyMayMatch(ExtractUserKey(key), prefix_extractor,
+                               handle.offset(), no_io);
 
       if (not_exist_in_filter) {
         // Not found
@@ -2158,7 +2217,7 @@ Status BlockBasedTable::Prefetch(const Slice* const begin,
   }
 
   BlockIter iiter_on_stack;
-  auto iiter = NewIndexIterator(ReadOptions(), &iiter_on_stack);
+  auto iiter = NewIndexIterator(ReadOptions(), false, &iiter_on_stack);
   std::unique_ptr<InternalIterator> iiter_unique_ptr;
   if (iiter != &iiter_on_stack) {
     iiter_unique_ptr = std::unique_ptr<InternalIterator>(iiter);
@@ -2215,7 +2274,8 @@ Status BlockBasedTable::VerifyChecksum() {
   }
   // Check Data blocks
   BlockIter iiter_on_stack;
-  InternalIterator* iiter = NewIndexIterator(ReadOptions(), &iiter_on_stack);
+  InternalIterator* iiter =
+      NewIndexIterator(ReadOptions(), false, &iiter_on_stack);
   std::unique_ptr<InternalIterator> iiter_unique_ptr;
   if (iiter != &iiter_on_stack) {
     iiter_unique_ptr = std::unique_ptr<InternalIterator>(iiter);
@@ -2308,20 +2368,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;
-  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;
-  }
+
+  // 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
 
   switch (index_type_on_file) {
     case BlockBasedTableOptions::kTwoLevelIndexSearch: {
@@ -2461,7 +2521,8 @@ Status BlockBasedTable::GetKVPairsFromDataBlocks(
   return Status::OK();
 }
 
-Status BlockBasedTable::DumpTable(WritableFile* out_file) {
+Status BlockBasedTable::DumpTable(WritableFile* out_file,
+                                  const SliceTransform* prefix_extractor) {
   // Output Footer
   out_file->Append(
       "Footer Details:\n"
@@ -2542,7 +2603,7 @@ Status BlockBasedTable::DumpTable(WritableFile* out_file) {
         s = block_fetcher.ReadBlockContents();
         if (!s.ok()) {
           rep_->filter.reset(new BlockBasedFilterBlockReader(
-              rep_->ioptions.prefix_extractor, table_options,
+              prefix_extractor, table_options,
               table_options.whole_key_filtering, std::move(block),
               rep_->ioptions.statistics));
         }
@@ -2627,7 +2688,6 @@ 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();
diff --git a/table/block_based_table_reader.h b/table/block_based_table_reader.h
index 3b4e49d76..0b58c84e1 100644
--- a/table/block_based_table_reader.h
+++ b/table/block_based_table_reader.h
@@ -90,25 +90,29 @@ 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);
+  bool PrefixMayMatch(const Slice& internal_key,
+                      const SliceTransform* prefix_extractor = nullptr);
 
   // 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&, Arena* arena = nullptr,
-      bool skip_filters = false) override;
+  InternalIterator* NewIterator(const ReadOptions&,
+                                const SliceTransform* prefix_extractor,
+                                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, bool skip_filters = false) override;
+             GetContext* get_context, const SliceTransform* prefix_extractor,
+             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
@@ -136,7 +140,8 @@ class BlockBasedTable : public TableReader {
   size_t ApproximateMemoryUsage() const override;
 
   // convert SST file to a human readable form
-  Status DumpTable(WritableFile* out_file) override;
+  Status DumpTable(WritableFile* out_file,
+                   const SliceTransform* prefix_extractor = nullptr) override;
 
   Status VerifyChecksum() override;
 
@@ -253,12 +258,13 @@ 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;
+      const bool is_a_filter_partition, bool no_io, GetContext* get_context,
+      const SliceTransform* prefix_extractor = nullptr) const;
 
   // Get the iterator from the index reader.
   // If input_iter is not set, return new Iterator
@@ -271,7 +277,8 @@ class BlockBasedTable : public TableReader {
   //  3. We disallowed any io to be performed, that is, read_options ==
   //     kBlockCacheTier
   InternalIterator* NewIndexIterator(
-      const ReadOptions& read_options, BlockIter* input_iter = nullptr,
+      const ReadOptions& read_options, bool prefix_extractor_changed = false,
+      BlockIter* input_iter = nullptr,
       CachableEntry<IndexReader>* index_entry = nullptr,
       GetContext* get_context = nullptr);
 
@@ -325,9 +332,10 @@ 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;
+  bool FullFilterKeyMayMatch(
+      const ReadOptions& read_options, FilterBlockReader* filter,
+      const Slice& user_key, const bool no_io,
+      const SliceTransform* prefix_extractor = nullptr) const;
 
   // Read the meta block from sst.
   static Status ReadMetaBlock(Rep* rep, FilePrefetchBuffer* prefetch_buffer,
@@ -337,9 +345,10 @@ 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;
+  FilterBlockReader* ReadFilter(
+      FilePrefetchBuffer* prefetch_buffer, const BlockHandle& filter_handle,
+      const bool is_a_filter_partition,
+      const SliceTransform* prefix_extractor = nullptr) const;
 
   static void SetupCacheKeyPrefix(Rep* rep, uint64_t file_size);
 
@@ -499,14 +508,16 @@ class BlockBasedTableIterator : public InternalIterator {
   BlockBasedTableIterator(BlockBasedTable* table,
                           const ReadOptions& read_options,
                           const InternalKeyComparator& icomp,
-                          InternalIterator* index_iter, bool check_filter)
+                          InternalIterator* index_iter, bool check_filter,
+                          const SliceTransform* prefix_extractor)
       : 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) {}
+        check_filter_(check_filter),
+        prefix_extractor_(prefix_extractor) {}
 
   ~BlockBasedTableIterator() { delete index_iter_; }
 
@@ -552,8 +563,9 @@ class BlockBasedTableIterator : public InternalIterator {
            block_iter_points_to_real_block_;
   }
 
-  bool CheckPrefixMayMatch(const Slice& ikey) {
-    if (check_filter_ && !table_->PrefixMayMatch(ikey)) {
+  bool CheckPrefixMayMatch(const Slice& ikey,
+                           const SliceTransform* prefix_extractor = nullptr) {
+    if (check_filter_ && !table_->PrefixMayMatch(ikey, prefix_extractor)) {
       // 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
@@ -599,6 +611,7 @@ 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
diff --git a/table/cuckoo_table_reader.cc b/table/cuckoo_table_reader.cc
index d611352b2..a287566e8 100644
--- a/table/cuckoo_table_reader.cc
+++ b/table/cuckoo_table_reader.cc
@@ -141,6 +141,7 @@ 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);
@@ -377,7 +378,9 @@ extern InternalIterator* NewErrorInternalIterator(const Status& status,
                                                   Arena* arena);
 
 InternalIterator* CuckooTableReader::NewIterator(
-    const ReadOptions& /*read_options*/, Arena* arena, bool /*skip_filters*/) {
+    const ReadOptions& /*read_options*/,
+    const SliceTransform* /* prefix_extractor */, Arena* arena,
+    bool /*skip_filters*/) {
   if (!status().ok()) {
     return NewErrorInternalIterator(
         Status::Corruption("CuckooTableReader status is not okay."), arena);
diff --git a/table/cuckoo_table_reader.h b/table/cuckoo_table_reader.h
index 2988c88e0..c0d70174b 100644
--- a/table/cuckoo_table_reader.h
+++ b/table/cuckoo_table_reader.h
@@ -42,12 +42,14 @@ class CuckooTableReader: public TableReader {
 
   Status status() const { return status_; }
 
-  Status Get(const ReadOptions& read_options, const Slice& key,
-             GetContext* get_context, bool skip_filters = false) override;
+  Status Get(const ReadOptions& readOptions, const Slice& key,
+             GetContext* get_context, const SliceTransform* prefix_extractor,
+             bool skip_filters = false) override;
 
-  InternalIterator* NewIterator(
-      const ReadOptions&, Arena* arena = nullptr,
-      bool skip_filters = false) override;
+  InternalIterator* NewIterator(const ReadOptions&,
+                                const SliceTransform* prefix_extractor,
+                                Arena* arena = nullptr,
+                                bool skip_filters = false) override;
   void Prepare(const Slice& target) override;
 
   // Report an approximation of how much memory has been used.
diff --git a/table/cuckoo_table_reader_test.cc b/table/cuckoo_table_reader_test.cc
index d831e02e4..65c93d196 100644
--- a/table/cuckoo_table_reader_test.cc
+++ b/table/cuckoo_table_reader_test.cc
@@ -127,7 +127,8 @@ 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));
+      ASSERT_OK(
+          reader.Get(ReadOptions(), Slice(keys[i]), &get_context, nullptr));
       ASSERT_STREQ(values[i].c_str(), value.data());
     }
   }
@@ -149,7 +150,8 @@ 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);
+    InternalIterator* it =
+        reader.NewIterator(ReadOptions(), nullptr, nullptr, false);
     ASSERT_OK(it->status());
     ASSERT_TRUE(!it->Valid());
     it->SeekToFirst();
@@ -188,7 +190,7 @@ class CuckooReaderTest : public testing::Test {
     delete it;
 
     Arena arena;
-    it = reader.NewIterator(ReadOptions(), &arena);
+    it = reader.NewIterator(ReadOptions(), nullptr, &arena);
     ASSERT_OK(it->status());
     ASSERT_TRUE(!it->Valid());
     it->Seek(keys[num_items/2]);
@@ -337,7 +339,8 @@ 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));
+  ASSERT_OK(
+      reader.Get(ReadOptions(), Slice(not_found_key), &get_context, nullptr));
   ASSERT_TRUE(value.empty());
   ASSERT_OK(reader.status());
   // Search for a key with an independent hash value.
@@ -350,7 +353,8 @@ 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));
+  ASSERT_OK(
+      reader.Get(ReadOptions(), Slice(not_found_key2), &get_context2, nullptr));
   ASSERT_TRUE(value.empty());
   ASSERT_OK(reader.status());
 
@@ -365,7 +369,8 @@ 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));
+  ASSERT_OK(
+      reader.Get(ReadOptions(), Slice(unused_key), &get_context3, nullptr));
   ASSERT_TRUE(value.empty());
   ASSERT_OK(reader.status());
 }
@@ -443,7 +448,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));
+    ASSERT_OK(reader.Get(r_options, Slice(keys[i]), &get_context, nullptr));
     ASSERT_TRUE(Slice(keys[i]) == Slice(&keys[i][0], 4));
   }
 }
@@ -496,13 +501,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);
+                   &get_context, nullptr);
       }
     }
   } else {
     for (uint64_t i = 0; i < num; i++) {
       reader.Get(r_options, Slice(reinterpret_cast<char*>(&keys[i]), 16),
-                 &get_context);
+                 &get_context, nullptr);
     }
   }
   float time_per_op = (env->NowMicros() - start_time) * 1.0f / num;
diff --git a/table/filter_block.h b/table/filter_block.h
index 97f493fb7..8f3666e1b 100644
--- a/table/filter_block.h
+++ b/table/filter_block.h
@@ -93,16 +93,21 @@ class FilterBlockReader {
    * built upon InternalKey and must be provided via const_ikey_ptr when running
    * queries.
    */
-  virtual bool KeyMayMatch(const Slice& key, uint64_t block_offset = kNotValid,
+  virtual bool KeyMayMatch(const Slice& key,
+                           const SliceTransform* prefix_extractor,
+                           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_; }
@@ -115,7 +120,8 @@ class FilterBlockReader {
     return error_msg;
   }
 
-  virtual void CacheDependencies(bool /*pin*/) {}
+  virtual void CacheDependencies(bool /*pin*/,
+                                 const SliceTransform* /*prefix_extractor*/) {}
 
  protected:
   bool whole_key_filtering_;
diff --git a/table/full_filter_block.cc b/table/full_filter_block.cc
index 0e621042d..359b19c62 100644
--- a/table/full_filter_block.cc
+++ b/table/full_filter_block.cc
@@ -103,9 +103,10 @@ FullFilterBlockReader::FullFilterBlockReader(
   block_contents_ = std::move(contents);
 }
 
-bool FullFilterBlockReader::KeyMayMatch(const Slice& key, uint64_t block_offset,
-                                        const bool /*no_io*/,
-                                        const Slice* const /*const_ikey_ptr*/) {
+bool FullFilterBlockReader::KeyMayMatch(
+    const Slice& key, const SliceTransform* /*prefix_extractor*/,
+    uint64_t block_offset, const bool /*no_io*/,
+    const Slice* const /*const_ikey_ptr*/) {
 #ifdef NDEBUG
   (void)block_offset;
 #endif
@@ -117,7 +118,8 @@ bool FullFilterBlockReader::KeyMayMatch(const Slice& key, uint64_t block_offset,
 }
 
 bool FullFilterBlockReader::PrefixMayMatch(
-    const Slice& prefix, uint64_t block_offset, const bool /*no_io*/,
+    const Slice& prefix, const SliceTransform* /* prefix_extractor */,
+    uint64_t block_offset, const bool /*no_io*/,
     const Slice* const /*const_ikey_ptr*/) {
 #ifdef NDEBUG
   (void)block_offset;
diff --git a/table/full_filter_block.h b/table/full_filter_block.h
index 352051052..628cefdd8 100644
--- a/table/full_filter_block.h
+++ b/table/full_filter_block.h
@@ -96,13 +96,15 @@ class FullFilterBlockReader : public FilterBlockReader {
   ~FullFilterBlockReader() {}
 
   virtual bool IsBlockBased() override { return false; }
+
   virtual bool KeyMayMatch(
-      const Slice& key, uint64_t block_offset = kNotValid,
-      const bool no_io = false,
+      const Slice& key, const SliceTransform* prefix_extractor,
+      uint64_t block_offset = kNotValid, const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
+
   virtual bool PrefixMayMatch(
-      const Slice& prefix, uint64_t block_offset = kNotValid,
-      const bool no_io = false,
+      const Slice& prefix, const SliceTransform* prefix_extractor,
+      uint64_t block_offset = kNotValid, const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual size_t ApproximateMemoryUsage() const override;
 
diff --git a/table/full_filter_block_test.cc b/table/full_filter_block_test.cc
index eb77ec056..b2d81eee3 100644
--- a/table/full_filter_block_test.cc
+++ b/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"));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr));
 }
 
 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"));
-  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"));
+  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));
 }
 
 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"));
+  ASSERT_TRUE(reader.KeyMayMatch("foo", nullptr));
 }
 
 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"));
-  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"));
+  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));
 }
 
 }  // namespace rocksdb
diff --git a/table/mock_table.cc b/table/mock_table.cc
index e5210b2c1..32704dbca 100644
--- a/table/mock_table.cc
+++ b/table/mock_table.cc
@@ -26,14 +26,16 @@ stl_wrappers::KVMap MakeMockFile(
   return stl_wrappers::KVMap(l, stl_wrappers::LessOfComparator(&icmp_));
 }
 
-InternalIterator* MockTableReader::NewIterator(const ReadOptions&,
-                                               Arena* /*arena*/,
-                                               bool /*skip_filters*/) {
+InternalIterator* MockTableReader::NewIterator(
+    const ReadOptions&, const SliceTransform* /* prefix_extractor */,
+    Arena* /*arena*/, bool /*skip_filters*/) {
   return new MockTableIterator(table_);
 }
 
 Status MockTableReader::Get(const ReadOptions&, const Slice& key,
-                            GetContext* get_context, bool /*skip_filters*/) {
+                            GetContext* get_context,
+                            const SliceTransform* /*prefix_extractor*/,
+                            bool /*skip_filters*/) {
   std::unique_ptr<MockTableIterator> iter(new MockTableIterator(table_));
   for (iter->Seek(key); iter->Valid(); iter->Next()) {
     ParsedInternalKey parsed_key;
diff --git a/table/mock_table.h b/table/mock_table.h
index f5fb4009e..a95725b6a 100644
--- a/table/mock_table.h
+++ b/table/mock_table.h
@@ -39,10 +39,12 @@ class MockTableReader : public TableReader {
   explicit MockTableReader(const stl_wrappers::KVMap& table) : table_(table) {}
 
   InternalIterator* NewIterator(const ReadOptions&,
-                                Arena* arena,
+                                const SliceTransform* prefix_extractor,
+                                Arena* arena = nullptr,
                                 bool skip_filters = false) override;
 
-  Status Get(const ReadOptions&, const Slice& key, GetContext* get_context,
+  Status Get(const ReadOptions& readOptions, const Slice& key,
+             GetContext* get_context, const SliceTransform* prefix_extractor,
              bool skip_filters = false) override;
 
   uint64_t ApproximateOffsetOf(const Slice& /*key*/) override { return 0; }
diff --git a/table/partitioned_filter_block.cc b/table/partitioned_filter_block.cc
index ab8483f12..2f03f2fec 100644
--- a/table/partitioned_filter_block.cc
+++ b/table/partitioned_filter_block.cc
@@ -130,7 +130,8 @@ PartitionedFilterBlockReader::~PartitionedFilterBlockReader() {
 }
 
 bool PartitionedFilterBlockReader::KeyMayMatch(
-    const Slice& key, uint64_t block_offset, const bool no_io,
+    const Slice& key, const SliceTransform* prefix_extractor,
+    uint64_t block_offset, const bool no_io,
     const Slice* const const_ikey_ptr) {
   assert(const_ikey_ptr != nullptr);
   assert(block_offset == kNotValid);
@@ -145,12 +146,14 @@ bool PartitionedFilterBlockReader::KeyMayMatch(
     return false;
   }
   bool cached = false;
-  auto filter_partition = GetFilterPartition(nullptr /* prefetch_buffer */,
-                                             &filter_handle, no_io, &cached);
+  auto filter_partition =
+      GetFilterPartition(nullptr /* prefetch_buffer */, &filter_handle, no_io,
+                         &cached, prefix_extractor);
   if (UNLIKELY(!filter_partition.value)) {
     return true;
   }
-  auto res = filter_partition.value->KeyMayMatch(key, block_offset, no_io);
+  auto res = filter_partition.value->KeyMayMatch(key, prefix_extractor,
+                                                 block_offset, no_io);
   if (cached) {
     return res;
   }
@@ -163,7 +166,8 @@ bool PartitionedFilterBlockReader::KeyMayMatch(
 }
 
 bool PartitionedFilterBlockReader::PrefixMayMatch(
-    const Slice& prefix, uint64_t block_offset, const bool no_io,
+    const Slice& prefix, const SliceTransform* prefix_extractor,
+    uint64_t block_offset, const bool no_io,
     const Slice* const const_ikey_ptr) {
 #ifdef NDEBUG
   (void)block_offset;
@@ -181,12 +185,14 @@ bool PartitionedFilterBlockReader::PrefixMayMatch(
     return false;
   }
   bool cached = false;
-  auto filter_partition = GetFilterPartition(nullptr /* prefetch_buffer */,
-                                             &filter_handle, no_io, &cached);
+  auto filter_partition =
+      GetFilterPartition(nullptr /* prefetch_buffer */, &filter_handle, no_io,
+                         &cached, prefix_extractor);
   if (UNLIKELY(!filter_partition.value)) {
     return true;
   }
-  auto res = filter_partition.value->PrefixMayMatch(prefix, kNotValid, no_io);
+  auto res = filter_partition.value->PrefixMayMatch(prefix, prefix_extractor,
+                                                    kNotValid, no_io);
   if (cached) {
     return res;
   }
@@ -214,7 +220,7 @@ Slice PartitionedFilterBlockReader::GetFilterPartitionHandle(
 BlockBasedTable::CachableEntry<FilterBlockReader>
 PartitionedFilterBlockReader::GetFilterPartition(
     FilePrefetchBuffer* prefetch_buffer, Slice* handle_value, const bool no_io,
-    bool* cached) {
+    bool* cached, const SliceTransform* prefix_extractor) {
   BlockHandle fltr_blk_handle;
   auto s = fltr_blk_handle.DecodeFrom(handle_value);
   assert(s.ok());
@@ -237,10 +243,10 @@ PartitionedFilterBlockReader::GetFilterPartition(
     }
     return table_->GetFilter(/*prefetch_buffer*/ nullptr, fltr_blk_handle,
                              is_a_filter_partition, no_io,
-                             /* get_context */ nullptr);
+                             /* get_context */ nullptr, prefix_extractor);
   } else {
     auto filter = table_->ReadFilter(prefetch_buffer, fltr_blk_handle,
-                                     is_a_filter_partition);
+                                     is_a_filter_partition, prefix_extractor);
     return {filter, nullptr};
   }
 }
@@ -257,7 +263,8 @@ void ReleaseFilterCachedEntry(void* arg, void* h) {
 }
 
 // TODO(myabandeh): merge this with the same function in IndexReader
-void PartitionedFilterBlockReader::CacheDependencies(bool pin) {
+void PartitionedFilterBlockReader::CacheDependencies(
+    bool pin, const SliceTransform* prefix_extractor) {
   // Before read partitions, prefetch them to avoid lots of IOs
   auto rep = table_->rep_;
   BlockIter biter;
@@ -308,9 +315,9 @@ void PartitionedFilterBlockReader::CacheDependencies(bool pin) {
 
     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);
+    auto filter = table_->GetFilter(
+        prefetch_buffer.get(), handle, is_a_filter_partition, !no_io,
+        /* get_context */ nullptr, prefix_extractor);
     if (LIKELY(filter.IsSet())) {
       if (pin) {
         filter_map_[handle.offset()] = std::move(filter);
diff --git a/table/partitioned_filter_block.h b/table/partitioned_filter_block.h
index 4c3611cd8..c9b358f9d 100644
--- a/table/partitioned_filter_block.h
+++ b/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, uint64_t block_offset = kNotValid,
-      const bool no_io = false,
+      const Slice& key, const SliceTransform* prefix_extractor,
+      uint64_t block_offset = kNotValid, const bool no_io = false,
       const Slice* const const_ikey_ptr = nullptr) override;
   virtual bool PrefixMayMatch(
-      const Slice& prefix, uint64_t block_offset = kNotValid,
-      const bool no_io = false,
+      const Slice& prefix, const SliceTransform* prefix_extractor,
+      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,8 +92,9 @@ class PartitionedFilterBlockReader : public FilterBlockReader,
   Slice GetFilterPartitionHandle(const Slice& entry);
   BlockBasedTable::CachableEntry<FilterBlockReader> GetFilterPartition(
       FilePrefetchBuffer* prefetch_buffer, Slice* handle, const bool no_io,
-      bool* cached);
-  virtual void CacheDependencies(bool pin) override;
+      bool* cached, const SliceTransform* prefix_extractor = nullptr);
+  virtual void CacheDependencies(
+      bool bin, const SliceTransform* prefix_extractor) override;
 
   const SliceTransform* prefix_extractor_;
   std::unique_ptr<Block> idx_on_fltr_blk_;
diff --git a/table/partitioned_filter_block_test.cc b/table/partitioned_filter_block_test.cc
index aef6dd39b..ac47f2504 100644
--- a/table/partitioned_filter_block_test.cc
+++ b/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 override {
+      const bool /* unused */, bool /* unused */, GetContext* /* unused */,
+      const SliceTransform* /* unused */) const override {
     Slice slice = slices[filter_blk_handle.offset()];
     auto obj = new FullFilterBlockReader(
         nullptr, true, BlockContents(slice, false, kNoCompression),
@@ -121,6 +121,7 @@ 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)));
@@ -138,23 +139,27 @@ 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, kNotValid, !no_io, &ikey_slice));
+      ASSERT_TRUE(
+          reader->KeyMayMatch(key, nullptr, 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], kNotValid, !no_io, &ikey_slice));
+      ASSERT_TRUE(reader->KeyMayMatch(keys[0], nullptr, 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, kNotValid, !no_io, &ikey_slice));
+        ASSERT_TRUE(
+            reader->KeyMayMatch(key, nullptr, kNotValid, !no_io, &ikey_slice));
       } else {
         // assuming a good hash function
-        ASSERT_FALSE(reader->KeyMayMatch(key, kNotValid, !no_io, &ikey_slice));
+        ASSERT_FALSE(
+            reader->KeyMayMatch(key, nullptr, kNotValid, !no_io, &ikey_slice));
       }
     }
   }
diff --git a/table/plain_table_builder.cc b/table/plain_table_builder.cc
index 964804358..717635cc1 100644
--- a/table/plain_table_builder.cc
+++ b/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 ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
     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,20 +66,21 @@ 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, ioptions.prefix_extractor,
+      encoder_(encoding_type, user_key_len, moptions.prefix_extractor.get(),
                index_sparseness),
       store_index_in_file_(store_index_in_file),
-      prefix_extractor_(ioptions.prefix_extractor) {
+      prefix_extractor_(moptions.prefix_extractor.get()) {
   // 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, index_sparseness,
-                                   hash_table_ratio, huge_page_tlb_size_));
+    index_builder_.reset(new PlainTableIndexBuilder(
+        &arena_, ioptions, moptions.prefix_extractor.get(), index_sparseness,
+        hash_table_ratio, huge_page_tlb_size_));
     properties_.user_collected_properties
         [PlainTablePropertyNames::kBloomVersion] = "1";  // For future use
   }
@@ -96,8 +97,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 = ioptions_.prefix_extractor != nullptr
-                                          ? ioptions_.prefix_extractor->Name()
+  properties_.prefix_extractor_name = moptions_.prefix_extractor != nullptr
+                                          ? moptions_.prefix_extractor->Name()
                                           : "nullptr";
 
   std::string val;
@@ -131,11 +132,11 @@ void PlainTableBuilder::Add(const Slice& key, const Slice& value) {
 
   // Store key hash
   if (store_index_in_file_) {
-    if (ioptions_.prefix_extractor == nullptr) {
+    if (moptions_.prefix_extractor == nullptr) {
       keys_or_prefixes_hashes_.push_back(GetSliceHash(internal_key.user_key));
     } else {
       Slice prefix =
-          ioptions_.prefix_extractor->Transform(internal_key.user_key);
+          moptions_.prefix_extractor->Transform(internal_key.user_key);
       keys_or_prefixes_hashes_.push_back(GetSliceHash(prefix));
     }
   }
diff --git a/table/plain_table_builder.h b/table/plain_table_builder.h
index 1d1f6c758..ca0879a4e 100644
--- a/table/plain_table_builder.h
+++ b/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 ImmutableCFOptions& ioptions, const MutableCFOptions& moptions,
       const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
           int_tbl_prop_collector_factories,
       uint32_t column_family_id, WritableFileWriter* file,
@@ -79,6 +79,7 @@ class PlainTableBuilder: public TableBuilder {
  private:
   Arena arena_;
   const ImmutableCFOptions& ioptions_;
+  const MutableCFOptions& moptions_;
   std::vector<std::unique_ptr<IntTblPropCollector>>
       table_properties_collectors_;
 
diff --git a/table/plain_table_factory.cc b/table/plain_table_factory.cc
index ac0c4c80f..b88a689d4 100644
--- a/table/plain_table_factory.cc
+++ b/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_options_.full_scan_mode, table_reader_options.prefix_extractor);
 }
 
 TableBuilder* PlainTableFactory::NewTableBuilder(
@@ -38,7 +38,7 @@ TableBuilder* PlainTableFactory::NewTableBuilder(
   // tables
   //
   return new PlainTableBuilder(
-      table_builder_options.ioptions,
+      table_builder_options.ioptions, table_builder_options.moptions,
       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,
diff --git a/table/plain_table_index.h b/table/plain_table_index.h
index f4adf0258..360d99827 100644
--- a/table/plain_table_index.h
+++ b/table/plain_table_index.h
@@ -112,6 +112,7 @@ 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),
@@ -125,7 +126,7 @@ class PlainTableIndexBuilder {
         index_sparseness_(index_sparseness),
         index_size_(0),
         sub_index_size_(0),
-        prefix_extractor_(ioptions.prefix_extractor),
+        prefix_extractor_(prefix_extractor),
         hash_table_ratio_(hash_table_ratio),
         huge_page_tlb_size_(huge_page_tlb_size) {}
 
diff --git a/table/plain_table_reader.cc b/table/plain_table_reader.cc
index 8408abfad..5df7486b8 100644
--- a/table/plain_table_reader.cc
+++ b/table/plain_table_reader.cc
@@ -97,12 +97,13 @@ PlainTableReader::PlainTableReader(const ImmutableCFOptions& ioptions,
                                    const InternalKeyComparator& icomparator,
                                    EncodingType encoding_type,
                                    uint64_t file_size,
-                                   const TableProperties* table_properties)
+                                   const TableProperties* table_properties,
+                                   const SliceTransform* prefix_extractor)
     : 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_(ioptions.prefix_extractor),
+      prefix_extractor_(prefix_extractor),
       enable_bloom_(false),
       bloom_(6, nullptr),
       file_info_(std::move(file), storage_options,
@@ -114,15 +115,13 @@ 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) {
+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) {
   if (file_size > PlainTableIndex::kMaxFileSize) {
     return Status::NotSupported("File is too large for PlainTableReader!");
   }
@@ -141,12 +140,12 @@ Status PlainTableReader::Open(const ImmutableCFOptions& ioptions,
   if (!full_scan_mode &&
       !prefix_extractor_in_file.empty() /* old version sst file*/
       && prefix_extractor_in_file != "nullptr") {
-    if (!ioptions.prefix_extractor) {
+    if (!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(
-                   ioptions.prefix_extractor->Name()) != 0) {
+    } else if (prefix_extractor_in_file.compare(prefix_extractor->Name()) !=
+               0) {
       return Status::InvalidArgument(
           "Prefix extractor given doesn't match the one used to build "
           "PlainTable");
@@ -163,7 +162,7 @@ Status PlainTableReader::Open(const ImmutableCFOptions& ioptions,
 
   std::unique_ptr<PlainTableReader> new_reader(new PlainTableReader(
       ioptions, std::move(file), env_options, internal_comparator,
-      encoding_type, file_size, props));
+      encoding_type, file_size, props, prefix_extractor));
 
   s = new_reader->MmapDataIfNeeded();
   if (!s.ok()) {
@@ -189,9 +188,9 @@ Status PlainTableReader::Open(const ImmutableCFOptions& ioptions,
 void PlainTableReader::SetupForCompaction() {
 }
 
-InternalIterator* PlainTableReader::NewIterator(const ReadOptions& options,
-                                                Arena* arena,
-                                                bool /*skip_filters*/) {
+InternalIterator* PlainTableReader::NewIterator(
+    const ReadOptions& options, const SliceTransform* /* prefix_extractor */,
+    Arena* arena, bool /*skip_filters*/) {
   bool use_prefix_seek = !IsTotalOrderMode() && !options.total_order_seek;
   if (arena == nullptr) {
     return new PlainTableIterator(this, use_prefix_seek);
@@ -210,7 +209,7 @@ Status PlainTableReader::PopulateIndexRecordList(
   bool is_first_record = true;
   Slice key_prefix_slice;
   PlainTableKeyDecoder decoder(&file_info_, encoding_type_, user_key_len_,
-                               ioptions_.prefix_extractor);
+                               prefix_extractor_);
   while (pos < file_info_.data_end_offset) {
     uint32_t key_offset = pos;
     ParsedInternalKey key;
@@ -330,9 +329,8 @@ Status PlainTableReader::PopulateIndex(TableProperties* props,
     index_block = nullptr;
   }
 
-  if ((ioptions_.prefix_extractor == nullptr) &&
-      (hash_table_ratio != 0)) {
-    // ioptions.prefix_extractor is requried for a hash-based look-up.
+  if ((prefix_extractor_ == nullptr) && (hash_table_ratio != 0)) {
+    // moptions.prefix_extractor is requried for a hash-based look-up.
     return Status::NotSupported(
         "PlainTable requires a prefix extractor enable prefix hash mode.");
   }
@@ -377,8 +375,9 @@ Status PlainTableReader::PopulateIndex(TableProperties* props,
     bloom_bits_per_key = 0;
   }
 
-  PlainTableIndexBuilder index_builder(&arena_, ioptions_, index_sparseness,
-                                       hash_table_ratio, huge_page_tlb_size);
+  PlainTableIndexBuilder index_builder(&arena_, ioptions_, prefix_extractor_,
+                                       index_sparseness, hash_table_ratio,
+                                       huge_page_tlb_size);
 
   std::vector<uint32_t> prefix_hashes;
   if (!index_in_file) {
@@ -538,7 +537,9 @@ void PlainTableReader::Prepare(const Slice& target) {
 }
 
 Status PlainTableReader::Get(const ReadOptions& /*ro*/, const Slice& target,
-                             GetContext* get_context, bool /*skip_filters*/) {
+                             GetContext* get_context,
+                             const SliceTransform* /* prefix_extractor */,
+                             bool /*skip_filters*/) {
   // Check bloom filter first.
   Slice prefix_slice;
   uint32_t prefix_hash;
@@ -565,7 +566,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_,
-                               ioptions_.prefix_extractor);
+                               prefix_extractor_);
   Status s = GetOffset(&decoder, target, prefix_slice, prefix_hash,
                        prefix_match, &offset);
 
diff --git a/table/plain_table_reader.h b/table/plain_table_reader.h
index 6bf8da2f9..5fdfa4786 100644
--- a/table/plain_table_reader.h
+++ b/table/plain_table_reader.h
@@ -76,15 +76,18 @@ 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);
+                     bool full_scan_mode,
+                     const SliceTransform* prefix_extractor = nullptr);
 
   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&, const Slice& key, GetContext* get_context,
+  Status Get(const ReadOptions& readOptions, const Slice& key,
+             GetContext* get_context, const SliceTransform* prefix_extractor,
              bool skip_filters = false) override;
 
   uint64_t ApproximateOffsetOf(const Slice& key) override;
@@ -105,7 +108,8 @@ class PlainTableReader: public TableReader {
                    const EnvOptions& env_options,
                    const InternalKeyComparator& internal_comparator,
                    EncodingType encoding_type, uint64_t file_size,
-                   const TableProperties* table_properties);
+                   const TableProperties* table_properties,
+                   const SliceTransform* prefix_extractor);
   virtual ~PlainTableReader();
 
  protected:
diff --git a/table/sst_file_writer.cc b/table/sst_file_writer.cc
index 2ed387446..d1134d36e 100644
--- a/table/sst_file_writer.cc
+++ b/table/sst_file_writer.cc
@@ -189,10 +189,10 @@ Status SstFileWriter::Open(const std::string& file_path) {
   }
 
   TableBuilderOptions table_builder_options(
-      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->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->file_writer.reset(
       new WritableFileWriter(std::move(sst_file), r->env_options));
 
diff --git a/table/table_builder.h b/table/table_builder.h
index e5e7d6e22..c8af93406 100644
--- a/table/table_builder.h
+++ b/table/table_builder.h
@@ -27,16 +27,19 @@ 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)
@@ -47,7 +50,7 @@ struct TableReaderOptions {
 
 struct TableBuilderOptions {
   TableBuilderOptions(
-      const ImmutableCFOptions& _ioptions,
+      const ImmutableCFOptions& _ioptions, const MutableCFOptions& _moptions,
       const InternalKeyComparator& _internal_comparator,
       const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
           _int_tbl_prop_collector_factories,
@@ -57,6 +60,7 @@ 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),
@@ -68,6 +72,7 @@ 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;
diff --git a/table/table_properties.cc b/table/table_properties.cc
index 4adced5ec..306f5c17e 100644
--- a/table/table_properties.cc
+++ b/table/table_properties.cc
@@ -107,6 +107,11 @@ 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
diff --git a/table/table_reader.h b/table/table_reader.h
index 37a282b10..43a146b2b 100644
--- a/table/table_reader.h
+++ b/table/table_reader.h
@@ -9,6 +9,7 @@
 
 #pragma once
 #include <memory>
+#include "rocksdb/slice_transform.h"
 #include "table/internal_iterator.h"
 
 namespace rocksdb {
@@ -39,6 +40,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 InternalIterator* NewIterator(const ReadOptions&,
+                                        const SliceTransform* prefix_extractor,
                                         Arena* arena = nullptr,
                                         bool skip_filters = false) = 0;
 
@@ -79,7 +81,9 @@ 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, bool skip_filters = false) = 0;
+                     GetContext* get_context,
+                     const SliceTransform* prefix_extractor,
+                     bool skip_filters = false) = 0;
 
   // Prefetch data corresponding to a give range of keys
   // Typically this functionality is required for table implementations that
@@ -94,7 +98,8 @@ class TableReader {
   }
 
   // convert db file to a human readable form
-  virtual Status DumpTable(WritableFile* /*out_file*/) {
+  virtual Status DumpTable(WritableFile* /*out_file*/,
+                           const SliceTransform* /*prefix_extractor*/) {
     return Status::NotSupported("DumpTable() not supported");
   }
 
diff --git a/table/table_reader_bench.cc b/table/table_reader_bench.cc
index f9732772e..9bce610fd 100644
--- a/table/table_reader_bench.cc
+++ b/table/table_reader_bench.cc
@@ -84,6 +84,8 @@ void TableReaderBenchmark(Options& opts, EnvOptions& env_options,
   DB* db = nullptr;
   Status s;
   const ImmutableCFOptions ioptions(opts);
+  const ColumnFamilyOptions cfo(opts);
+  const MutableCFOptions moptions(cfo);
   unique_ptr<WritableFileWriter> file_writer;
   if (!through_db) {
     unique_ptr<WritableFile> file;
@@ -95,12 +97,11 @@ void TableReaderBenchmark(Options& opts, EnvOptions& env_options,
     file_writer.reset(new WritableFileWriter(std::move(file), env_options));
     int unknown_level = -1;
     tb = opts.table_factory->NewTableBuilder(
-        TableBuilderOptions(ioptions, ikc, &int_tbl_prop_collector_factories,
-                            CompressionType::kNoCompression,
-                            CompressionOptions(),
-                            nullptr /* compression_dict */,
-                            false /* skip_filters */, kDefaultColumnFamilyName,
-                            unknown_level),
+        TableBuilderOptions(
+            ioptions, moptions, ikc, &int_tbl_prop_collector_factories,
+            CompressionType::kNoCompression, CompressionOptions(),
+            nullptr /* compression_dict */, false /* skip_filters */,
+            kDefaultColumnFamilyName, unknown_level),
         0 /* column_family_id */, file_writer.get());
   } else {
     s = DB::Open(opts, dbname, &db);
@@ -138,8 +139,9 @@ void TableReaderBenchmark(Options& opts, EnvOptions& env_options,
     unique_ptr<RandomAccessFileReader> file_reader(
         new RandomAccessFileReader(std::move(raf), file_name));
     s = opts.table_factory->NewTableReader(
-        TableReaderOptions(ioptions, env_options, ikc), std::move(file_reader),
-        file_size, &table_reader);
+        TableReaderOptions(ioptions, moptions.prefix_extractor.get(),
+                           env_options, ikc),
+        std::move(file_reader), file_size, &table_reader);
     if (!s.ok()) {
       fprintf(stderr, "Open Table Error: %s\n", s.ToString().c_str());
       exit(1);
@@ -173,7 +175,7 @@ void TableReaderBenchmark(Options& opts, EnvOptions& env_options,
                                    ioptions.statistics, GetContext::kNotFound,
                                    Slice(key), &value, nullptr, &merge_context,
                                    &range_del_agg, env);
-            s = table_reader->Get(read_options, key, &get_context);
+            s = table_reader->Get(read_options, key, &get_context, nullptr);
           } else {
             s = db->Get(read_options, key, &result);
           }
@@ -195,7 +197,7 @@ void TableReaderBenchmark(Options& opts, EnvOptions& env_options,
           Iterator* iter = nullptr;
           InternalIterator* iiter = nullptr;
           if (!through_db) {
-            iiter = table_reader->NewIterator(read_options);
+            iiter = table_reader->NewIterator(read_options, nullptr);
           } else {
             iter = db->NewIterator(read_options);
           }
diff --git a/table/table_test.cc b/table/table_test.cc
index 19bd8043d..79a43a2c9 100644
--- a/table/table_test.cc
+++ b/table/table_test.cc
@@ -163,6 +163,7 @@ class Constructor {
   // been added so far.  Returns the keys in sorted order in "*keys"
   // and stores the key/value pairs in "*kvmap"
   void Finish(const Options& options, const ImmutableCFOptions& ioptions,
+              const MutableCFOptions& moptions,
               const BlockBasedTableOptions& table_options,
               const InternalKeyComparator& internal_comparator,
               std::vector<std::string>* keys, stl_wrappers::KVMap* kvmap) {
@@ -173,7 +174,7 @@ class Constructor {
       keys->push_back(kv.first);
     }
     data_.clear();
-    Status s = FinishImpl(options, ioptions, table_options,
+    Status s = FinishImpl(options, ioptions, moptions, table_options,
                           internal_comparator, *kvmap);
     ASSERT_TRUE(s.ok()) << s.ToString();
   }
@@ -181,11 +182,13 @@ class Constructor {
   // Construct the data structure from the data in "data"
   virtual Status FinishImpl(const Options& options,
                             const ImmutableCFOptions& ioptions,
+                            const MutableCFOptions& moptions,
                             const BlockBasedTableOptions& table_options,
                             const InternalKeyComparator& internal_comparator,
                             const stl_wrappers::KVMap& data) = 0;
 
-  virtual InternalIterator* NewIterator() const = 0;
+  virtual InternalIterator* NewIterator(
+      const SliceTransform* prefix_extractor = nullptr) const = 0;
 
   virtual const stl_wrappers::KVMap& data() { return data_; }
 
@@ -213,6 +216,7 @@ class BlockConstructor: public Constructor {
   }
   virtual Status FinishImpl(
       const Options& /*options*/, const ImmutableCFOptions& /*ioptions*/,
+      const MutableCFOptions& /*moptions*/,
       const BlockBasedTableOptions& table_options,
       const InternalKeyComparator& /*internal_comparator*/,
       const stl_wrappers::KVMap& kv_map) override {
@@ -231,7 +235,8 @@ class BlockConstructor: public Constructor {
     block_ = new Block(std::move(contents), kDisableGlobalSequenceNumber);
     return Status::OK();
   }
-  virtual InternalIterator* NewIterator() const override {
+  virtual InternalIterator* NewIterator(
+      const SliceTransform* /*prefix_extractor*/) const override {
     return block_->NewIterator(comparator_);
   }
 
@@ -311,6 +316,7 @@ class TableConstructor: public Constructor {
 
   virtual Status FinishImpl(const Options& options,
                             const ImmutableCFOptions& ioptions,
+                            const MutableCFOptions& moptions,
                             const BlockBasedTableOptions& /*table_options*/,
                             const InternalKeyComparator& internal_comparator,
                             const stl_wrappers::KVMap& kv_map) override {
@@ -323,10 +329,10 @@ class TableConstructor: public Constructor {
     std::string column_family_name;
     builder.reset(ioptions.table_factory->NewTableBuilder(
         TableBuilderOptions(
-            ioptions, internal_comparator, &int_tbl_prop_collector_factories,
-            options.compression, CompressionOptions(),
-            nullptr /* compression_dict */, false /* skip_filters */,
-            column_family_name, level_),
+            ioptions, moptions, internal_comparator,
+            &int_tbl_prop_collector_factories, options.compression,
+            CompressionOptions(), nullptr /* compression_dict */,
+            false /* skip_filters */, column_family_name, level_),
         TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
         file_writer_.get()));
 
@@ -353,14 +359,15 @@ class TableConstructor: public Constructor {
         GetSink()->contents(), uniq_id_, ioptions.allow_mmap_reads)));
     const bool skip_filters = false;
     return ioptions.table_factory->NewTableReader(
-        TableReaderOptions(ioptions, soptions, internal_comparator,
-                           skip_filters, level_),
+        TableReaderOptions(ioptions, moptions.prefix_extractor.get(), soptions,
+                           internal_comparator, skip_filters, level_),
         std::move(file_reader_), GetSink()->contents().size(), &table_reader_);
   }
 
-  virtual InternalIterator* NewIterator() const override {
+  virtual InternalIterator* NewIterator(
+      const SliceTransform* prefix_extractor) const override {
     ReadOptions ro;
-    InternalIterator* iter = table_reader_->NewIterator(ro);
+    InternalIterator* iter = table_reader_->NewIterator(ro, prefix_extractor);
     if (convert_to_internal_key_) {
       return new KeyConvertingIterator(iter);
     } else {
@@ -377,11 +384,13 @@ class TableConstructor: public Constructor {
     return table_reader_->ApproximateOffsetOf(key);
   }
 
-  virtual Status Reopen(const ImmutableCFOptions& ioptions) {
+  virtual Status Reopen(const ImmutableCFOptions& ioptions,
+                        const MutableCFOptions& moptions) {
     file_reader_.reset(test::GetRandomAccessFileReader(new test::StringSource(
         GetSink()->contents(), uniq_id_, ioptions.allow_mmap_reads)));
     return ioptions.table_factory->NewTableReader(
-        TableReaderOptions(ioptions, soptions, *last_internal_key_),
+        TableReaderOptions(ioptions, moptions.prefix_extractor.get(), soptions,
+                           *last_internal_key_),
         std::move(file_reader_), GetSink()->contents().size(), &table_reader_);
   }
 
@@ -442,6 +451,7 @@ class MemTableConstructor: public Constructor {
   }
   virtual Status FinishImpl(
       const Options&, const ImmutableCFOptions& ioptions,
+      const MutableCFOptions& /*moptions*/,
       const BlockBasedTableOptions& /*table_options*/,
       const InternalKeyComparator& /*internal_comparator*/,
       const stl_wrappers::KVMap& kv_map) override {
@@ -458,7 +468,8 @@ class MemTableConstructor: public Constructor {
     }
     return Status::OK();
   }
-  virtual InternalIterator* NewIterator() const override {
+  virtual InternalIterator* NewIterator(
+      const SliceTransform* /*prefix_extractor*/) const override {
     return new KeyConvertingIterator(
         memtable_->NewIterator(ReadOptions(), &arena_), true);
   }
@@ -509,6 +520,7 @@ class DBConstructor: public Constructor {
   }
   virtual Status FinishImpl(
       const Options& /*options*/, const ImmutableCFOptions& /*ioptions*/,
+      const MutableCFOptions& /*moptions*/,
       const BlockBasedTableOptions& /*table_options*/,
       const InternalKeyComparator& /*internal_comparator*/,
       const stl_wrappers::KVMap& kv_map) override {
@@ -523,7 +535,8 @@ class DBConstructor: public Constructor {
     return Status::OK();
   }
 
-  virtual InternalIterator* NewIterator() const override {
+  virtual InternalIterator* NewIterator(
+      const SliceTransform* /*prefix_extractor*/) const override {
     return new InternalIteratorFromIterator(db_->NewIterator(ReadOptions()));
   }
 
@@ -684,6 +697,7 @@ class HarnessTest : public testing::Test {
  public:
   HarnessTest()
       : ioptions_(options_),
+        moptions_(options_),
         constructor_(nullptr),
         write_buffer_(options_.db_write_buffer_size) {}
 
@@ -782,6 +796,7 @@ class HarnessTest : public testing::Test {
         break;
     }
     ioptions_ = ImmutableCFOptions(options_);
+    moptions_ = MutableCFOptions(options_);
   }
 
   ~HarnessTest() { delete constructor_; }
@@ -793,7 +808,7 @@ class HarnessTest : public testing::Test {
   void Test(Random* rnd) {
     std::vector<std::string> keys;
     stl_wrappers::KVMap data;
-    constructor_->Finish(options_, ioptions_, table_options_,
+    constructor_->Finish(options_, ioptions_, moptions_, table_options_,
                          *internal_comparator_, &keys, &data);
 
     TestForwardScan(keys, data);
@@ -996,6 +1011,7 @@ class HarnessTest : public testing::Test {
  private:
   Options options_ = Options();
   ImmutableCFOptions ioptions_;
+  MutableCFOptions moptions_;
   BlockBasedTableOptions table_options_ = BlockBasedTableOptions();
   Constructor* constructor_;
   WriteBufferManager write_buffer_;
@@ -1104,8 +1120,9 @@ TEST_F(BlockBasedTableTest, BasicBlockBasedTableProperties) {
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
   ImmutableCFOptions ioptions(options);
+  MutableCFOptions moptions(options);
   ioptions.statistics = options.statistics.get();
-  c.Finish(options, ioptions, table_options,
+  c.Finish(options, ioptions, moptions, table_options,
            GetPlainInternalComparator(options.comparator), &keys, &kvmap);
   ASSERT_EQ(options.statistics->getTickerCount(NUMBER_BLOCK_NOT_COMPRESSED), 0);
 
@@ -1152,8 +1169,9 @@ uint64_t BlockBasedTableTest::IndexUncompressedHelper(bool compressed) {
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
   ImmutableCFOptions ioptions(options);
+  MutableCFOptions moptions(options);
   ioptions.statistics = options.statistics.get();
-  c.Finish(options, ioptions, table_options,
+  c.Finish(options, ioptions, moptions, table_options,
            GetPlainInternalComparator(options.comparator), &keys, &kvmap);
   c.ResetTableReader();
   return options.statistics->getTickerCount(NUMBER_BLOCK_COMPRESSED);
@@ -1178,7 +1196,8 @@ TEST_F(BlockBasedTableTest, BlockBasedTableProperties2) {
     options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
     const ImmutableCFOptions ioptions(options);
-    c.Finish(options, ioptions, table_options,
+    const MutableCFOptions moptions(options);
+    c.Finish(options, ioptions, moptions, table_options,
              GetPlainInternalComparator(options.comparator), &keys, &kvmap);
 
     auto& props = *c.GetTableReader()->GetTableProperties();
@@ -1211,7 +1230,8 @@ TEST_F(BlockBasedTableTest, BlockBasedTableProperties2) {
         new DummyPropertiesCollectorFactory2());
 
     const ImmutableCFOptions ioptions(options);
-    c.Finish(options, ioptions, table_options,
+    const MutableCFOptions moptions(options);
+    c.Finish(options, ioptions, moptions, table_options,
              GetPlainInternalComparator(options.comparator), &keys, &kvmap);
 
     auto& props = *c.GetTableReader()->GetTableProperties();
@@ -1246,10 +1266,11 @@ TEST_F(BlockBasedTableTest, RangeDelBlock) {
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
   const ImmutableCFOptions ioptions(options);
+  const MutableCFOptions moptions(options);
   std::unique_ptr<InternalKeyComparator> internal_cmp(
       new InternalKeyComparator(options.comparator));
-  c.Finish(options, ioptions, table_options, *internal_cmp, &sorted_keys,
-           &kvmap);
+  c.Finish(options, ioptions, moptions, table_options, *internal_cmp,
+           &sorted_keys, &kvmap);
 
   for (int j = 0; j < 2; ++j) {
     std::unique_ptr<InternalIterator> iter(
@@ -1287,7 +1308,8 @@ TEST_F(BlockBasedTableTest, FilterPolicyNameProperties) {
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options,
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options,
            GetPlainInternalComparator(options.comparator), &keys, &kvmap);
   auto& props = *c.GetTableReader()->GetTableProperties();
   ASSERT_EQ("rocksdb.BuiltinBloomFilter", props.filter_policy_name);
@@ -1330,7 +1352,8 @@ void PrefetchRange(TableConstructor* c, Options* opt,
   table_options->block_cache = NewLRUCache(16 * 1024 * 1024, 4);
   opt->table_factory.reset(NewBlockBasedTableFactory(*table_options));
   const ImmutableCFOptions ioptions2(*opt);
-  ASSERT_OK(c->Reopen(ioptions2));
+  const MutableCFOptions moptions(*opt);
+  ASSERT_OK(c->Reopen(ioptions2, moptions));
 
   // prefetch
   auto* table_reader = dynamic_cast<BlockBasedTable*>(c->GetTableReader());
@@ -1387,7 +1410,8 @@ TEST_F(BlockBasedTableTest, PrefetchTest) {
   std::vector<std::string> keys;
   stl_wrappers::KVMap kvmap;
   const ImmutableCFOptions ioptions(opt);
-  c.Finish(opt, ioptions, table_options, *ikc, &keys, &kvmap);
+  const MutableCFOptions moptions(opt);
+  c.Finish(opt, ioptions, moptions, table_options, *ikc, &keys, &kvmap);
   c.ResetTableReader();
 
   // We get the following data spread :
@@ -1482,14 +1506,16 @@ TEST_F(BlockBasedTableTest, TotalOrderSeekOnHashIndex) {
     std::vector<std::string> keys;
     stl_wrappers::KVMap kvmap;
     const ImmutableCFOptions ioptions(options);
-    c.Finish(options, ioptions, table_options,
+    const MutableCFOptions moptions(options);
+    c.Finish(options, ioptions, moptions, table_options,
              GetPlainInternalComparator(options.comparator), &keys, &kvmap);
     auto props = c.GetTableReader()->GetTableProperties();
     ASSERT_EQ(7u, props->num_data_blocks);
     auto* reader = c.GetTableReader();
     ReadOptions ro;
     ro.total_order_seek = true;
-    std::unique_ptr<InternalIterator> iter(reader->NewIterator(ro));
+    std::unique_ptr<InternalIterator> iter(
+        reader->NewIterator(ro, moptions.prefix_extractor.get()));
 
     iter->Seek(InternalKey("b", 0, kTypeValue).Encode());
     ASSERT_OK(iter->status());
@@ -1538,15 +1564,17 @@ TEST_F(BlockBasedTableTest, NoopTransformSeek) {
   std::vector<std::string> keys;
   stl_wrappers::KVMap kvmap;
   const ImmutableCFOptions ioptions(options);
+  const MutableCFOptions moptions(options);
   const InternalKeyComparator internal_comparator(options.comparator);
-  c.Finish(options, ioptions, table_options, internal_comparator, &keys,
-           &kvmap);
+  c.Finish(options, ioptions, moptions, table_options, internal_comparator,
+           &keys, &kvmap);
 
   auto* reader = c.GetTableReader();
   for (int i = 0; i < 2; ++i) {
     ReadOptions ro;
     ro.total_order_seek = (i == 0);
-    std::unique_ptr<InternalIterator> iter(reader->NewIterator(ro));
+    std::unique_ptr<InternalIterator> iter(
+        reader->NewIterator(ro, moptions.prefix_extractor.get()));
 
     iter->Seek(key.Encode());
     ASSERT_OK(iter->status());
@@ -1573,14 +1601,18 @@ TEST_F(BlockBasedTableTest, SkipPrefixBloomFilter) {
   std::vector<std::string> keys;
   stl_wrappers::KVMap kvmap;
   const ImmutableCFOptions ioptions(options);
+  const MutableCFOptions moptions(options);
   const InternalKeyComparator internal_comparator(options.comparator);
-  c.Finish(options, ioptions, table_options, internal_comparator, &keys,
-           &kvmap);
+  c.Finish(options, ioptions, moptions, table_options, internal_comparator,
+           &keys, &kvmap);
+  // TODO(Zhongyi): update test to use MutableCFOptions
   options.prefix_extractor.reset(NewFixedPrefixTransform(9));
   const ImmutableCFOptions new_ioptions(options);
-  c.Reopen(new_ioptions);
+  const MutableCFOptions new_moptions(options);
+  c.Reopen(new_ioptions, new_moptions);
   auto reader = c.GetTableReader();
-  std::unique_ptr<InternalIterator> db_iter(reader->NewIterator(ReadOptions()));
+  std::unique_ptr<InternalIterator> db_iter(
+      reader->NewIterator(ReadOptions(), new_moptions.prefix_extractor.get()));
 
   // Test point lookup
   // only one kv
@@ -1637,14 +1669,17 @@ void TableTest::IndexTest(BlockBasedTableOptions table_options) {
   std::unique_ptr<InternalKeyComparator> comparator(
       new InternalKeyComparator(BytewiseComparator()));
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options, *comparator, &keys, &kvmap);
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options, *comparator, &keys,
+           &kvmap);
   auto reader = c.GetTableReader();
 
   auto props = reader->GetTableProperties();
   ASSERT_EQ(5u, props->num_data_blocks);
 
+  // TODO(Zhongyi): update test to use MutableCFOptions
   std::unique_ptr<InternalIterator> index_iter(
-      reader->NewIterator(ReadOptions()));
+      reader->NewIterator(ReadOptions(), moptions.prefix_extractor.get()));
 
   // -- Find keys do not exist, but have common prefix.
   std::vector<std::string> prefixes = {"001", "003", "005", "007", "009"};
@@ -1773,7 +1808,8 @@ TEST_F(BlockBasedTableTest, IndexSizeStat) {
     options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
     const ImmutableCFOptions ioptions(options);
-    c.Finish(options, ioptions, table_options,
+    const MutableCFOptions moptions(options);
+    c.Finish(options, ioptions, moptions, table_options,
              GetPlainInternalComparator(options.comparator), &ks, &kvmap);
     auto index_size = c.GetTableReader()->GetTableProperties()->index_size;
     ASSERT_GT(index_size, last_index_size);
@@ -1801,7 +1837,8 @@ TEST_F(BlockBasedTableTest, NumBlockStat) {
   std::vector<std::string> ks;
   stl_wrappers::KVMap kvmap;
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options,
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options,
            GetPlainInternalComparator(options.comparator), &ks, &kvmap);
   ASSERT_EQ(kvmap.size(),
             c.GetTableReader()->GetTableProperties()->num_data_blocks);
@@ -1887,7 +1924,8 @@ TEST_F(BlockBasedTableTest, BlockCacheDisabledTest) {
   TableConstructor c(BytewiseComparator(), true /* convert_to_internal_key_ */);
   c.Add("key", "value");
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options,
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options,
            GetPlainInternalComparator(options.comparator), &keys, &kvmap);
 
   // preloading filter/index blocks is enabled.
@@ -1907,7 +1945,8 @@ TEST_F(BlockBasedTableTest, BlockCacheDisabledTest) {
                            GetContext::kNotFound, Slice(), nullptr, nullptr,
                            nullptr, nullptr, nullptr);
     // a hack that just to trigger BlockBasedTable::GetFilter.
-    reader->Get(ReadOptions(), "non-exist-key", &get_context);
+    reader->Get(ReadOptions(), "non-exist-key", &get_context,
+                moptions.prefix_extractor.get());
     BlockCachePropertiesSnapshot props(options.statistics.get());
     props.AssertIndexBlockStat(0, 0);
     props.AssertFilterBlockStat(0, 0);
@@ -1933,7 +1972,8 @@ TEST_F(BlockBasedTableTest, FilterBlockInBlockCache) {
   TableConstructor c(BytewiseComparator(), true /* convert_to_internal_key_ */);
   c.Add("key", "value");
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options,
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options,
            GetPlainInternalComparator(options.comparator), &keys, &kvmap);
   // preloading filter/index blocks is prohibited.
   auto* reader = dynamic_cast<BlockBasedTable*>(c.GetTableReader());
@@ -1959,7 +1999,7 @@ TEST_F(BlockBasedTableTest, FilterBlockInBlockCache) {
 
   // Only index block will be accessed
   {
-    iter.reset(c.NewIterator());
+    iter.reset(c.NewIterator(moptions.prefix_extractor.get()));
     BlockCachePropertiesSnapshot props(options.statistics.get());
     // NOTE: to help better highlight the "detla" of each ticker, I use
     // <last_value> + <added_value> to indicate the increment of changed
@@ -1988,7 +2028,7 @@ TEST_F(BlockBasedTableTest, FilterBlockInBlockCache) {
 
   // Data block will be in cache
   {
-    iter.reset(c.NewIterator());
+    iter.reset(c.NewIterator(moptions.prefix_extractor.get()));
     iter->SeekToFirst();
     BlockCachePropertiesSnapshot props(options.statistics.get());
     props.AssertEqual(1, 1 + 1, /* index block hit */
@@ -2010,7 +2050,8 @@ TEST_F(BlockBasedTableTest, FilterBlockInBlockCache) {
   options.statistics = CreateDBStatistics();
   options.table_factory.reset(new BlockBasedTableFactory(table_options));
   const ImmutableCFOptions ioptions2(options);
-  c.Reopen(ioptions2);
+  const MutableCFOptions moptions2(options);
+  c.Reopen(ioptions2, moptions2);
   {
     BlockCachePropertiesSnapshot props(options.statistics.get());
     props.AssertEqual(1,  // index block miss
@@ -2023,7 +2064,7 @@ TEST_F(BlockBasedTableTest, FilterBlockInBlockCache) {
     // Both index and data block get accessed.
     // It first cache index block then data block. But since the cache size
     // is only 1, index block will be purged after data block is inserted.
-    iter.reset(c.NewIterator());
+    iter.reset(c.NewIterator(moptions2.prefix_extractor.get()));
     BlockCachePropertiesSnapshot props(options.statistics.get());
     props.AssertEqual(1 + 1,  // index block miss
                       0, 0,   // data block miss
@@ -2055,8 +2096,9 @@ TEST_F(BlockBasedTableTest, FilterBlockInBlockCache) {
   InternalKey internal_key(user_key, 0, kTypeValue);
   c3.Add(internal_key.Encode().ToString(), "hello");
   ImmutableCFOptions ioptions3(options);
+  MutableCFOptions moptions3(options);
   // Generate table without filter policy
-  c3.Finish(options, ioptions3, table_options,
+  c3.Finish(options, ioptions3, moptions3, table_options,
             GetPlainInternalComparator(options.comparator), &keys, &kvmap);
   c3.ResetTableReader();
 
@@ -2065,14 +2107,16 @@ TEST_F(BlockBasedTableTest, FilterBlockInBlockCache) {
   options.table_factory.reset(new BlockBasedTableFactory(table_options));
   options.statistics = CreateDBStatistics();
   ImmutableCFOptions ioptions4(options);
-  ASSERT_OK(c3.Reopen(ioptions4));
+  MutableCFOptions moptions4(options);
+  ASSERT_OK(c3.Reopen(ioptions4, moptions4));
   reader = dynamic_cast<BlockBasedTable*>(c3.GetTableReader());
   ASSERT_TRUE(!reader->TEST_filter_block_preloaded());
   PinnableSlice value;
   GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
                          GetContext::kNotFound, user_key, &value, nullptr,
                          nullptr, nullptr, nullptr);
-  ASSERT_OK(reader->Get(ReadOptions(), user_key, &get_context));
+  ASSERT_OK(reader->Get(ReadOptions(), user_key, &get_context,
+                        moptions4.prefix_extractor.get()));
   ASSERT_STREQ(value.data(), "hello");
   BlockCachePropertiesSnapshot props(options.statistics.get());
   props.AssertFilterBlockStat(0, 0);
@@ -2147,8 +2191,9 @@ TEST_F(BlockBasedTableTest, BlockReadCountTest) {
       std::string encoded_key = internal_key.Encode().ToString();
       c.Add(encoded_key, "hello");
       ImmutableCFOptions ioptions(options);
+      MutableCFOptions moptions(options);
       // Generate table with filter policy
-      c.Finish(options, ioptions, table_options,
+      c.Finish(options, ioptions, moptions, table_options,
                GetPlainInternalComparator(options.comparator), &keys, &kvmap);
       auto reader = c.GetTableReader();
       PinnableSlice value;
@@ -2156,7 +2201,8 @@ TEST_F(BlockBasedTableTest, BlockReadCountTest) {
                              GetContext::kNotFound, user_key, &value, nullptr,
                              nullptr, nullptr, nullptr);
       get_perf_context()->Reset();
-      ASSERT_OK(reader->Get(ReadOptions(), encoded_key, &get_context));
+      ASSERT_OK(reader->Get(ReadOptions(), encoded_key, &get_context,
+                            moptions.prefix_extractor.get()));
       if (index_and_filter_in_cache) {
         // data, index and filter block
         ASSERT_EQ(get_perf_context()->block_read_count, 3);
@@ -2177,7 +2223,8 @@ TEST_F(BlockBasedTableTest, BlockReadCountTest) {
                                GetContext::kNotFound, user_key, &value, nullptr,
                                nullptr, nullptr, nullptr);
       get_perf_context()->Reset();
-      ASSERT_OK(reader->Get(ReadOptions(), encoded_key, &get_context));
+      ASSERT_OK(reader->Get(ReadOptions(), encoded_key, &get_context,
+                            moptions.prefix_extractor.get()));
       ASSERT_EQ(get_context.State(), GetContext::kNotFound);
 
       if (index_and_filter_in_cache) {
@@ -2296,7 +2343,9 @@ TEST_F(BlockBasedTableTest, NoObjectInCacheAfterTableClose) {
             std::vector<std::string> keys;
             stl_wrappers::KVMap kvmap;
             const ImmutableCFOptions ioptions(opt);
-            c.Finish(opt, ioptions, table_options, *ikc, &keys, &kvmap);
+            const MutableCFOptions moptions(opt);
+            c.Finish(opt, ioptions, moptions, table_options, *ikc, &keys,
+                     &kvmap);
 
             // Doing a read to make index/filter loaded into the cache
             auto table_reader =
@@ -2306,7 +2355,8 @@ TEST_F(BlockBasedTableTest, NoObjectInCacheAfterTableClose) {
                                    GetContext::kNotFound, user_key, &value,
                                    nullptr, nullptr, nullptr, nullptr);
             InternalKey ikey(user_key, 0, kTypeValue);
-            auto s = table_reader->Get(ReadOptions(), key, &get_context);
+            auto s = table_reader->Get(ReadOptions(), key, &get_context,
+                                       moptions.prefix_extractor.get());
             ASSERT_EQ(get_context.State(), GetContext::kFound);
             ASSERT_STREQ(value.data(), "hello");
 
@@ -2358,9 +2408,11 @@ TEST_F(BlockBasedTableTest, BlockCacheLeak) {
   std::vector<std::string> keys;
   stl_wrappers::KVMap kvmap;
   const ImmutableCFOptions ioptions(opt);
-  c.Finish(opt, ioptions, table_options, *ikc, &keys, &kvmap);
+  const MutableCFOptions moptions(opt);
+  c.Finish(opt, ioptions, moptions, table_options, *ikc, &keys, &kvmap);
 
-  unique_ptr<InternalIterator> iter(c.NewIterator());
+  unique_ptr<InternalIterator> iter(
+      c.NewIterator(moptions.prefix_extractor.get()));
   iter->SeekToFirst();
   while (iter->Valid()) {
     iter->key();
@@ -2371,7 +2423,8 @@ TEST_F(BlockBasedTableTest, BlockCacheLeak) {
   iter.reset();
 
   const ImmutableCFOptions ioptions1(opt);
-  ASSERT_OK(c.Reopen(ioptions1));
+  const MutableCFOptions moptions1(opt);
+  ASSERT_OK(c.Reopen(ioptions1, moptions1));
   auto table_reader = dynamic_cast<BlockBasedTable*>(c.GetTableReader());
   for (const std::string& key : keys) {
     ASSERT_TRUE(table_reader->TEST_KeyInCache(ReadOptions(), key));
@@ -2382,7 +2435,8 @@ TEST_F(BlockBasedTableTest, BlockCacheLeak) {
   table_options.block_cache = NewLRUCache(16 * 1024 * 1024, 4);
   opt.table_factory.reset(NewBlockBasedTableFactory(table_options));
   const ImmutableCFOptions ioptions2(opt);
-  ASSERT_OK(c.Reopen(ioptions2));
+  const MutableCFOptions moptions2(opt);
+  ASSERT_OK(c.Reopen(ioptions2, moptions2));
   table_reader = dynamic_cast<BlockBasedTable*>(c.GetTableReader());
   for (const std::string& key : keys) {
     ASSERT_TRUE(!table_reader->TEST_KeyInCache(ReadOptions(), key));
@@ -2405,7 +2459,8 @@ TEST_F(BlockBasedTableTest, NewIndexIteratorLeak) {
   table_options.block_cache = NewLRUCache(0);
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options,
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options,
            GetPlainInternalComparator(options.comparator), &keys, &kvmap);
 
   rocksdb::SyncPoint::GetInstance()->LoadDependencyAndMarkers(
@@ -2432,13 +2487,16 @@ TEST_F(BlockBasedTableTest, NewIndexIteratorLeak) {
 
   std::function<void()> func1 = [&]() {
     TEST_SYNC_POINT("BlockBasedTableTest::NewIndexIteratorLeak:Thread1Marker");
-    std::unique_ptr<InternalIterator> iter(reader->NewIterator(ro));
+    // TODO(Zhongyi): update test to use MutableCFOptions
+    std::unique_ptr<InternalIterator> iter(
+        reader->NewIterator(ro, moptions.prefix_extractor.get()));
     iter->Seek(InternalKey("a1", 0, kTypeValue).Encode());
   };
 
   std::function<void()> func2 = [&]() {
     TEST_SYNC_POINT("BlockBasedTableTest::NewIndexIteratorLeak:Thread2Marker");
-    std::unique_ptr<InternalIterator> iter(reader->NewIterator(ro));
+    std::unique_ptr<InternalIterator> iter(
+        reader->NewIterator(ro, moptions.prefix_extractor.get()));
   };
 
   auto thread1 = port::Thread(func1);
@@ -2463,17 +2521,17 @@ TEST_F(PlainTableTest, BasicPlainTableProperties) {
       test::GetWritableFileWriter(new test::StringSink()));
   Options options;
   const ImmutableCFOptions ioptions(options);
+  const MutableCFOptions moptions(options);
   InternalKeyComparator ikc(options.comparator);
   std::vector<std::unique_ptr<IntTblPropCollectorFactory>>
       int_tbl_prop_collector_factories;
   std::string column_family_name;
   int unknown_level = -1;
   std::unique_ptr<TableBuilder> builder(factory.NewTableBuilder(
-      TableBuilderOptions(ioptions, ikc, &int_tbl_prop_collector_factories,
-                          kNoCompression, CompressionOptions(),
-                          nullptr /* compression_dict */,
-                          false /* skip_filters */, column_family_name,
-                          unknown_level),
+      TableBuilderOptions(
+          ioptions, moptions, ikc, &int_tbl_prop_collector_factories,
+          kNoCompression, CompressionOptions(), nullptr /* compression_dict */,
+          false /* skip_filters */, column_family_name, unknown_level),
       TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
       file_writer.get()));
 
@@ -2525,7 +2583,8 @@ TEST_F(GeneralTableTest, ApproximateOffsetOfPlain) {
   BlockBasedTableOptions table_options;
   table_options.block_size = 1024;
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options, internal_comparator,
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options, internal_comparator,
            &keys, &kvmap);
 
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("abc"),       0,      0));
@@ -2560,7 +2619,8 @@ static void DoCompressionTest(CompressionType comp) {
   BlockBasedTableOptions table_options;
   table_options.block_size = 1024;
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options, ikc, &keys, &kvmap);
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options, ikc, &keys, &kvmap);
 
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("abc"),       0,      0));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k01"),       0,      0));
@@ -2930,10 +2990,13 @@ TEST_P(IndexBlockRestartIntervalTest, IndexBlockRestartInterval) {
   std::unique_ptr<InternalKeyComparator> comparator(
       new InternalKeyComparator(BytewiseComparator()));
   const ImmutableCFOptions ioptions(options);
-  c.Finish(options, ioptions, table_options, *comparator, &keys, &kvmap);
+  const MutableCFOptions moptions(options);
+  c.Finish(options, ioptions, moptions, table_options, *comparator, &keys,
+           &kvmap);
   auto reader = c.GetTableReader();
 
-  std::unique_ptr<InternalIterator> db_iter(reader->NewIterator(ReadOptions()));
+  std::unique_ptr<InternalIterator> db_iter(
+      reader->NewIterator(ReadOptions(), moptions.prefix_extractor.get()));
 
   // Test point lookup
   for (auto& kv : kvmap) {
@@ -3044,6 +3107,7 @@ TEST_F(BlockBasedTableTest, TableWithGlobalSeqno) {
   Options options;
   options.table_factory.reset(NewBlockBasedTableFactory(bbto));
   const ImmutableCFOptions ioptions(options);
+  const MutableCFOptions moptions(options);
   InternalKeyComparator ikc(options.comparator);
   std::vector<std::unique_ptr<IntTblPropCollectorFactory>>
       int_tbl_prop_collector_factories;
@@ -3052,9 +3116,9 @@ TEST_F(BlockBasedTableTest, TableWithGlobalSeqno) {
                                                   0 /* global_seqno*/));
   std::string column_family_name;
   std::unique_ptr<TableBuilder> builder(options.table_factory->NewTableBuilder(
-      TableBuilderOptions(ioptions, ikc, &int_tbl_prop_collector_factories,
-                          kNoCompression, CompressionOptions(),
-                          nullptr /* compression_dict */,
+      TableBuilderOptions(ioptions, moptions, ikc,
+                          &int_tbl_prop_collector_factories, kNoCompression,
+                          CompressionOptions(), nullptr /* compression_dict */,
                           false /* skip_filters */, column_family_name, -1),
       TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
       file_writer.get()));
@@ -3112,10 +3176,12 @@ TEST_F(BlockBasedTableTest, TableWithGlobalSeqno) {
             new test::StringSource(ss_rw.contents(), 73342, true)));
 
     options.table_factory->NewTableReader(
-        TableReaderOptions(ioptions, EnvOptions(), ikc), std::move(file_reader),
-        ss_rw.contents().size(), &table_reader);
+        TableReaderOptions(ioptions, moptions.prefix_extractor.get(),
+                           EnvOptions(), ikc),
+        std::move(file_reader), ss_rw.contents().size(), &table_reader);
 
-    return table_reader->NewIterator(ReadOptions());
+    return table_reader->NewIterator(ReadOptions(),
+                                     moptions.prefix_extractor.get());
   };
 
   GetVersionAndGlobalSeqno();
@@ -3223,14 +3289,15 @@ TEST_F(BlockBasedTableTest, BlockAlignTest) {
   options.compression = kNoCompression;
   options.table_factory.reset(NewBlockBasedTableFactory(bbto));
   const ImmutableCFOptions ioptions(options);
+  const MutableCFOptions moptions(options);
   InternalKeyComparator ikc(options.comparator);
   std::vector<std::unique_ptr<IntTblPropCollectorFactory>>
       int_tbl_prop_collector_factories;
   std::string column_family_name;
   std::unique_ptr<TableBuilder> builder(options.table_factory->NewTableBuilder(
-      TableBuilderOptions(ioptions, ikc, &int_tbl_prop_collector_factories,
-                          kNoCompression, CompressionOptions(),
-                          nullptr /* compression_dict */,
+      TableBuilderOptions(ioptions, moptions, ikc,
+                          &int_tbl_prop_collector_factories, kNoCompression,
+                          CompressionOptions(), nullptr /* compression_dict */,
                           false /* skip_filters */, column_family_name, -1),
       TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
       file_writer.get()));
@@ -3275,13 +3342,16 @@ TEST_F(BlockBasedTableTest, BlockAlignTest) {
   Options options2;
   options2.table_factory.reset(NewBlockBasedTableFactory(bbto));
   ImmutableCFOptions ioptions2(options2);
+  const MutableCFOptions moptions2(options2);
+
   ASSERT_OK(ioptions.table_factory->NewTableReader(
-      TableReaderOptions(ioptions2, EnvOptions(),
+      TableReaderOptions(ioptions2, moptions2.prefix_extractor.get(),
+                         EnvOptions(),
                          GetPlainInternalComparator(options2.comparator)),
       std::move(file_reader), ss_rw.contents().size(), &table_reader));
 
-  std::unique_ptr<InternalIterator> db_iter(
-      table_reader->NewIterator(ReadOptions()));
+  std::unique_ptr<InternalIterator> db_iter(table_reader->NewIterator(
+      ReadOptions(), moptions2.prefix_extractor.get()));
 
   int expected_key = 1;
   for (db_iter->SeekToFirst(); db_iter->Valid(); db_iter->Next()) {
diff --git a/tools/sst_dump_test.cc b/tools/sst_dump_test.cc
index d7f4b5887..eaf35916e 100644
--- a/tools/sst_dump_test.cc
+++ b/tools/sst_dump_test.cc
@@ -49,6 +49,7 @@ void createSST(const std::string& file_name,
   ReadOptions read_options;
   Options opts;
   const ImmutableCFOptions imoptions(opts);
+  const MutableCFOptions moptions(opts);
   rocksdb::InternalKeyComparator ikc(opts.comparator);
   unique_ptr<TableBuilder> tb;
 
@@ -61,11 +62,11 @@ void createSST(const std::string& file_name,
   std::string column_family_name;
   int unknown_level = -1;
   tb.reset(opts.table_factory->NewTableBuilder(
-      TableBuilderOptions(imoptions, ikc, &int_tbl_prop_collector_factories,
-                          CompressionType::kNoCompression, CompressionOptions(),
-                          nullptr /* compression_dict */,
-                          false /* skip_filters */, column_family_name,
-                          unknown_level),
+      TableBuilderOptions(
+          imoptions, moptions, ikc, &int_tbl_prop_collector_factories,
+          CompressionType::kNoCompression, CompressionOptions(),
+          nullptr /* compression_dict */, false /* skip_filters */,
+          column_family_name, unknown_level),
       TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
       file_writer.get()));
 
diff --git a/tools/sst_dump_tool.cc b/tools/sst_dump_tool.cc
index 4480bb09e..79bbf0d53 100644
--- a/tools/sst_dump_tool.cc
+++ b/tools/sst_dump_tool.cc
@@ -43,12 +43,15 @@
 
 namespace rocksdb {
 
-SstFileReader::SstFileReader(const std::string& file_path,
-                             bool verify_checksum,
+SstFileReader::SstFileReader(const std::string& file_path, bool verify_checksum,
                              bool output_hex)
-    :file_name_(file_path), read_num_(0), verify_checksum_(verify_checksum),
-    output_hex_(output_hex), ioptions_(options_),
-    internal_comparator_(BytewiseComparator()) {
+    : file_name_(file_path),
+      read_num_(0),
+      verify_checksum_(verify_checksum),
+      output_hex_(output_hex),
+      ioptions_(options_),
+      moptions_(ColumnFamilyOptions(options_)),
+      internal_comparator_(BytewiseComparator()) {
   fprintf(stdout, "Process %s\n", file_path.c_str());
   init_result_ = GetTableReader(file_name_);
 }
@@ -128,14 +131,16 @@ Status SstFileReader::NewTableReader(
   // BlockBasedTable
   if (BlockBasedTableFactory::kName == options_.table_factory->Name()) {
     return options_.table_factory->NewTableReader(
-        TableReaderOptions(ioptions_, soptions_, internal_comparator_,
+        TableReaderOptions(ioptions_, moptions_.prefix_extractor.get(),
+                           soptions_, internal_comparator_,
                            /*skip_filters=*/false),
         std::move(file_), file_size, &table_reader_, /*enable_prefetch=*/false);
   }
 
   // For all other factory implementation
   return options_.table_factory->NewTableReader(
-      TableReaderOptions(ioptions_, soptions_, internal_comparator_),
+      TableReaderOptions(ioptions_, moptions_.prefix_extractor.get(), soptions_,
+                         internal_comparator_),
       std::move(file_), file_size, &table_reader_);
 }
 
@@ -147,7 +152,8 @@ Status SstFileReader::DumpTable(const std::string& out_filename) {
   unique_ptr<WritableFile> out_file;
   Env* env = Env::Default();
   env->NewWritableFile(out_filename, &out_file, soptions_);
-  Status s = table_reader_->DumpTable(out_file.get());
+  Status s = table_reader_->DumpTable(out_file.get(),
+                                      moptions_.prefix_extractor.get());
   out_file->Close();
   return s;
 }
@@ -167,7 +173,8 @@ uint64_t SstFileReader::CalculateCompressedTableSize(
       tb_options,
       TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
       dest_writer.get()));
-  unique_ptr<InternalIterator> iter(table_reader_->NewIterator(ReadOptions()));
+  unique_ptr<InternalIterator> iter(table_reader_->NewIterator(
+      ReadOptions(), moptions_.prefix_extractor.get()));
   for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
     if (!iter->status().ok()) {
       fputs(iter->status().ToString().c_str(), stderr);
@@ -192,6 +199,8 @@ int SstFileReader::ShowAllCompressionSizes(
   ReadOptions read_options;
   Options opts;
   const ImmutableCFOptions imoptions(opts);
+  const ColumnFamilyOptions cfo(opts);
+  const MutableCFOptions moptions(cfo);
   rocksdb::InternalKeyComparator ikc(opts.comparator);
   std::vector<std::unique_ptr<IntTblPropCollectorFactory> >
       block_based_table_factories;
@@ -203,11 +212,10 @@ int SstFileReader::ShowAllCompressionSizes(
       CompressionOptions compress_opt;
       std::string column_family_name;
       int unknown_level = -1;
-      TableBuilderOptions tb_opts(imoptions, ikc, &block_based_table_factories,
-                                  i.first, compress_opt,
-                                  nullptr /* compression_dict */,
-                                  false /* skip_filters */, column_family_name,
-                                  unknown_level);
+      TableBuilderOptions tb_opts(
+          imoptions, moptions, ikc, &block_based_table_factories, i.first,
+          compress_opt, nullptr /* compression_dict */,
+          false /* skip_filters */, column_family_name, unknown_level);
       uint64_t file_size = CalculateCompressedTableSize(tb_opts, block_size);
       fprintf(stdout, "Compression: %s", i.second);
       fprintf(stdout, " Size: %" PRIu64 "\n", file_size);
@@ -291,8 +299,8 @@ Status SstFileReader::ReadSequential(bool print_kv, uint64_t read_num,
     return init_result_;
   }
 
-  InternalIterator* iter =
-      table_reader_->NewIterator(ReadOptions(verify_checksum_, false));
+  InternalIterator* iter = table_reader_->NewIterator(
+      ReadOptions(verify_checksum_, false), moptions_.prefix_extractor.get());
   uint64_t i = 0;
   if (has_from) {
     InternalKey ikey;
diff --git a/tools/sst_dump_tool_imp.h b/tools/sst_dump_tool_imp.h
index 9531b5415..ca60dd93c 100644
--- a/tools/sst_dump_tool_imp.h
+++ b/tools/sst_dump_tool_imp.h
@@ -74,6 +74,7 @@ class SstFileReader {
   unique_ptr<RandomAccessFileReader> file_;
 
   const ImmutableCFOptions ioptions_;
+  const MutableCFOptions moptions_;
   InternalKeyComparator internal_comparator_;
   unique_ptr<TableProperties> table_properties_;
 };
diff --git a/utilities/column_aware_encoding_util.cc b/utilities/column_aware_encoding_util.cc
index 18b24d5b8..6d15e77ad 100644
--- a/utilities/column_aware_encoding_util.cc
+++ b/utilities/column_aware_encoding_util.cc
@@ -38,6 +38,7 @@ ColumnAwareEncodingReader::ColumnAwareEncodingReader(
     const std::string& file_path)
     : file_name_(file_path),
       ioptions_(options_),
+      moptions_(options_),
       internal_comparator_(BytewiseComparator()) {
   InitTableReader(file_name_);
 }
@@ -55,7 +56,8 @@ void ColumnAwareEncodingReader::InitTableReader(const std::string& file_path) {
 
   std::unique_ptr<TableReader> table_reader;
   options_.table_factory->NewTableReader(
-      TableReaderOptions(ioptions_, soptions_, internal_comparator_,
+      TableReaderOptions(ioptions_, moptions_.prefix_extractor.get(), soptions_,
+                         internal_comparator_,
                          /*skip_filters=*/false),
       std::move(file_), file_size, &table_reader, /*enable_prefetch=*/false);
 
diff --git a/utilities/column_aware_encoding_util.h b/utilities/column_aware_encoding_util.h
index 385d410d1..c2c4fa2d6 100644
--- a/utilities/column_aware_encoding_util.h
+++ b/utilities/column_aware_encoding_util.h
@@ -71,6 +71,7 @@ class ColumnAwareEncodingReader {
   std::unique_ptr<RandomAccessFileReader> file_;
 
   const ImmutableCFOptions ioptions_;
+  const MutableCFOptions moptions_;
   InternalKeyComparator internal_comparator_;
   std::unique_ptr<TableProperties> table_properties_;
 };
diff --git a/utilities/date_tiered/date_tiered_db_impl.cc b/utilities/date_tiered/date_tiered_db_impl.cc
index 596e7f027..978bfb2e4 100644
--- a/utilities/date_tiered/date_tiered_db_impl.cc
+++ b/utilities/date_tiered/date_tiered_db_impl.cc
@@ -32,6 +32,7 @@ DateTieredDBImpl::DateTieredDBImpl(
     : db_(db),
       cf_options_(ColumnFamilyOptions(options)),
       ioptions_(ImmutableCFOptions(options)),
+      moptions_(MutableCFOptions(options)),
       icomp_(cf_options_.comparator),
       ttl_(ttl),
       column_family_interval_(column_family_interval),
@@ -379,7 +380,7 @@ Iterator* DateTieredDBImpl::NewIterator(const ReadOptions& opts) {
   DBImpl* db_impl = reinterpret_cast<DBImpl*>(db_);
 
   auto db_iter = NewArenaWrappedDbIterator(
-      db_impl->GetEnv(), opts, ioptions_, kMaxSequenceNumber,
+      db_impl->GetEnv(), opts, ioptions_, moptions_, kMaxSequenceNumber,
       cf_options_.max_sequential_skip_in_iterations, 0,
       nullptr /*read_callback*/);
 
diff --git a/utilities/date_tiered/date_tiered_db_impl.h b/utilities/date_tiered/date_tiered_db_impl.h
index bbefabe33..7a6a6b75a 100644
--- a/utilities/date_tiered/date_tiered_db_impl.h
+++ b/utilities/date_tiered/date_tiered_db_impl.h
@@ -56,6 +56,8 @@ class DateTieredDBImpl : public DateTieredDB {
 
   const ImmutableCFOptions ioptions_;
 
+  const MutableCFOptions moptions_;
+
   const InternalKeyComparator icomp_;
 
   // Storing all column family handles for time series data.