Extend existing unit tests to run with WriteUnPrepared as well

Summary:
DBImpl::FindObsoleteFiles() may call GetChildren() multiple times if different CFs are on the same path. Fix it.
Closes https://github.com/facebook/rocksdb/pull/3885

Differential Revision: D8084634

Pulled By: siying

fbshipit-source-id: b471fbc251f6a05e9243304dc14c0831060cc0b0

CMakeLists.txt

@@ -657,6 +657,8 @@ set(SOURCES
         utilities/transactions/transaction_util.cc
         utilities/transactions/write_prepared_txn.cc
         utilities/transactions/write_prepared_txn_db.cc
+        utilities/transactions/write_unprepared_txn.cc
+        utilities/transactions/write_unprepared_txn_db.cc
         utilities/ttl/db_ttl_impl.cc
         utilities/write_batch_with_index/write_batch_with_index.cc
         utilities/write_batch_with_index/write_batch_with_index_internal.cc

HISTORY.md

@@ -2,6 +2,13 @@
 ## Unreleased
 ### Public API Change
 * For users of `Statistics` objects created via `CreateDBStatistics()`, the format of the string returned by its `ToString()` method has changed.
+* With LRUCache, when high_pri_pool_ratio > 0, midpoint insertion strategy will be enabled to put low-pri items to the tail of low-pri list (the midpoint) when they first inserted into the cache. This is to make cache entries never get hit age out faster, improving cache efficiency when large background scan presents.
+
+### New Features
+* Changes the format of index blocks by storing the key in their raw form rather than converting them to InternalKey. This saves 8 bytes per index key. The feature is backward compatbile but not forward compatible. It is disabled by default unless format_version 3 or above is used.
+
+### Bug Fixes
+* fix deadlock with enable_pipelined_write=true and max_successive_merges > 0
 
 ## 5.14.0 (5/16/2018)
 ### Public API Change

TARGETS

@@ -280,6 +280,8 @@ cpp_library(
         "utilities/transactions/transaction_util.cc",
         "utilities/transactions/write_prepared_txn.cc",
         "utilities/transactions/write_prepared_txn_db.cc",
+        "utilities/transactions/write_unprepared_txn.cc",
+        "utilities/transactions/write_unprepared_txn_db.cc",
         "utilities/ttl/db_ttl_impl.cc",
         "utilities/write_batch_with_index/write_batch_with_index.cc",
         "utilities/write_batch_with_index/write_batch_with_index_internal.cc",

cache/lru_cache.cc

@@ -199,7 +199,7 @@ void LRUCacheShard::LRU_Remove(LRUHandle* e) {
 void LRUCacheShard::LRU_Insert(LRUHandle* e) {
   assert(e->next == nullptr);
   assert(e->prev == nullptr);
-  if (high_pri_pool_ratio_ > 0 && e->IsHighPri()) {
+  if (high_pri_pool_ratio_ > 0 && (e->IsHighPri() || e->HasHit())) {
     // Inset "e" to head of LRU list.
     e->next = &lru_;
     e->prev = lru_.prev;
@@ -246,18 +246,6 @@ void LRUCacheShard::EvictFromLRU(size_t charge,
   }
 }
 
-void* LRUCacheShard::operator new(size_t size) {
-  return port::cacheline_aligned_alloc(size);
-}
-
-void* LRUCacheShard::operator new(size_t /*size*/, void* ptr) { return ptr; }
-
-void LRUCacheShard::operator delete(void *memblock) {
-  port::cacheline_aligned_free(memblock);
-}
-
-void LRUCacheShard::operator delete(void* /*memblock*/, void* /*ptr*/) {}
-
 void LRUCacheShard::SetCapacity(size_t capacity) {
   autovector<LRUHandle*> last_reference_list;
   {
@@ -287,6 +275,7 @@ Cache::Handle* LRUCacheShard::Lookup(const Slice& key, uint32_t hash) {
       LRU_Remove(e);
     }
     e->refs++;
+    e->SetHit();
   }
   return reinterpret_cast<Cache::Handle*>(e);
 }
@@ -485,10 +474,13 @@ LRUCache::LRUCache(size_t capacity, int num_shard_bits,
 }
 
 LRUCache::~LRUCache() {
-  for (int i = 0; i < num_shards_; i++) {
-    shards_[i].~LRUCacheShard();
+  if (shards_ != nullptr) {
+    assert(num_shards_ > 0);
+    for (int i = 0; i < num_shards_; i++) {
+      shards_[i].~LRUCacheShard();
+    }
+    port::cacheline_aligned_free(shards_);
   }
-  port::cacheline_aligned_free(shards_);
 }
 
 CacheShard* LRUCache::GetShard(int shard) {
@@ -515,6 +507,7 @@ void LRUCache::DisownData() {
 // Do not drop data if compile with ASAN to suppress leak warning.
 #ifndef __SANITIZE_ADDRESS__
   shards_ = nullptr;
+  num_shards_ = 0;
 #endif  // !__SANITIZE_ADDRESS__
 }
 

cache/lru_cache.h

@@ -77,6 +77,7 @@ struct LRUHandle {
   bool InCache() { return flags & 1; }
   bool IsHighPri() { return flags & 2; }
   bool InHighPriPool() { return flags & 4; }
+  bool HasHit() { return flags & 8; }
 
   void SetInCache(bool in_cache) {
     if (in_cache) {
@@ -102,6 +103,8 @@ struct LRUHandle {
     }
   }
 
+  void SetHit() { flags |= 8; }
+
   void Free() {
     assert((refs == 1 && InCache()) || (refs == 0 && !InCache()));
     if (deleter) {
@@ -206,18 +209,6 @@ class ALIGN_AS(CACHE_LINE_SIZE) LRUCacheShard : public CacheShard {
   //  Retrives high pri pool ratio
   double GetHighPriPoolRatio();
 
-  // Overloading to aligned it to cache line size
-  // They are used by tests.
-  void* operator new(size_t);
-
-  // placement new
-  void* operator new(size_t, void*);
-
-  void operator delete(void *);
-
-  // placement delete, does nothing.
-  void operator delete(void*, void*);
-
  private:
   void LRU_Remove(LRUHandle* e);
   void LRU_Insert(LRUHandle* e);
@@ -304,7 +295,7 @@ class LRUCache : public ShardedCache {
   double GetHighPriPoolRatio();
 
  private:
-  LRUCacheShard* shards_;
+  LRUCacheShard* shards_ = nullptr;
   int num_shards_ = 0;
 };
 

cache/lru_cache_test.cc

@@ -15,11 +15,22 @@ namespace rocksdb {
 class LRUCacheTest : public testing::Test {
  public:
   LRUCacheTest() {}
-  ~LRUCacheTest() {}
+  ~LRUCacheTest() { DeleteCache(); }
+
+  void DeleteCache() {
+    if (cache_ != nullptr) {
+      cache_->~LRUCacheShard();
+      port::cacheline_aligned_free(cache_);
+      cache_ = nullptr;
+    }
+  }
 
   void NewCache(size_t capacity, double high_pri_pool_ratio = 0.0) {
-    cache_.reset(new LRUCacheShard(capacity, false /*strict_capcity_limit*/,
-                                   high_pri_pool_ratio));
+    DeleteCache();
+    cache_ = reinterpret_cast<LRUCacheShard*>(
+        port::cacheline_aligned_alloc(sizeof(LRUCacheShard)));
+    new (cache_) LRUCacheShard(capacity, false /*strict_capcity_limit*/,
+                               high_pri_pool_ratio);
   }
 
   void Insert(const std::string& key,
@@ -75,7 +86,7 @@ class LRUCacheTest : public testing::Test {
   }
 
  private:
-  std::unique_ptr<LRUCacheShard> cache_;
+  LRUCacheShard* cache_ = nullptr;
 };
 
 TEST_F(LRUCacheTest, BasicLRU) {
@@ -104,6 +115,29 @@ TEST_F(LRUCacheTest, BasicLRU) {
   ValidateLRUList({"e", "z", "d", "u", "v"});
 }
 
+TEST_F(LRUCacheTest, MidpointInsertion) {
+  // Allocate 2 cache entries to high-pri pool.
+  NewCache(5, 0.45);
+
+  Insert("a", Cache::Priority::LOW);
+  Insert("b", Cache::Priority::LOW);
+  Insert("c", Cache::Priority::LOW);
+  Insert("x", Cache::Priority::HIGH);
+  Insert("y", Cache::Priority::HIGH);
+  ValidateLRUList({"a", "b", "c", "x", "y"}, 2);
+
+  // Low-pri entries inserted to the tail of low-pri list (the midpoint).
+  // After lookup, it will move to the tail of the full list.
+  Insert("d", Cache::Priority::LOW);
+  ValidateLRUList({"b", "c", "d", "x", "y"}, 2);
+  ASSERT_TRUE(Lookup("d"));
+  ValidateLRUList({"b", "c", "x", "y", "d"}, 2);
+
+  // High-pri entries will be inserted to the tail of full list.
+  Insert("z", Cache::Priority::HIGH);
+  ValidateLRUList({"c", "x", "y", "d", "z"}, 2);
+}
+
 TEST_F(LRUCacheTest, EntriesWithPriority) {
   // Allocate 2 cache entries to high-pri pool.
   NewCache(5, 0.45);
@@ -130,15 +164,15 @@ TEST_F(LRUCacheTest, EntriesWithPriority) {
   Insert("a", Cache::Priority::LOW);
   ValidateLRUList({"v", "X", "a", "Y", "Z"}, 2);
 
-  // Low-pri entries will be inserted to head of low-pri pool after lookup.
+  // Low-pri entries will be inserted to head of high-pri pool after lookup.
   ASSERT_TRUE(Lookup("v"));
-  ValidateLRUList({"X", "a", "v", "Y", "Z"}, 2);
+  ValidateLRUList({"X", "a", "Y", "Z", "v"}, 2);
 
   // High-pri entries will be inserted to the head of the list after lookup.
   ASSERT_TRUE(Lookup("X"));
-  ValidateLRUList({"a", "v", "Y", "Z", "X"}, 2);
+  ValidateLRUList({"a", "Y", "Z", "v", "X"}, 2);
   ASSERT_TRUE(Lookup("Z"));
-  ValidateLRUList({"a", "v", "Y", "X", "Z"}, 2);
+  ValidateLRUList({"a", "Y", "v", "X", "Z"}, 2);
 
   Erase("Y");
   ValidateLRUList({"a", "v", "X", "Z"}, 2);
@@ -151,7 +185,7 @@ TEST_F(LRUCacheTest, EntriesWithPriority) {
   Insert("g", Cache::Priority::LOW);
   ValidateLRUList({"d", "e", "f", "g", "Z"}, 1);
   ASSERT_TRUE(Lookup("d"));
-  ValidateLRUList({"e", "f", "g", "d", "Z"}, 1);
+  ValidateLRUList({"e", "f", "g", "Z", "d"}, 2);
 }
 
 }  // namespace rocksdb

db/c.cc

@@ -90,6 +90,7 @@ using rocksdb::LiveFileMetaData;
 using rocksdb::BackupEngine;
 using rocksdb::BackupableDBOptions;
 using rocksdb::BackupInfo;
+using rocksdb::BackupID;
 using rocksdb::RestoreOptions;
 using rocksdb::CompactRangeOptions;
 using rocksdb::RateLimiter;
@@ -531,10 +532,18 @@ rocksdb_backup_engine_t* rocksdb_backup_engine_open(
 }
 
 void rocksdb_backup_engine_create_new_backup(rocksdb_backup_engine_t* be,
-                                             rocksdb_t* db, char** errptr) {
+                                             rocksdb_t* db,
+                                             char** errptr) {
   SaveError(errptr, be->rep->CreateNewBackup(db->rep));
 }
 
+void rocksdb_backup_engine_create_new_backup_flush(rocksdb_backup_engine_t* be,
+                                                   rocksdb_t* db,
+                                                   unsigned char flush_before_backup,
+                                                   char** errptr) {
+  SaveError(errptr, be->rep->CreateNewBackup(db->rep, flush_before_backup));
+}
+
 void rocksdb_backup_engine_purge_old_backups(rocksdb_backup_engine_t* be,
                                              uint32_t num_backups_to_keep,
                                              char** errptr) {
@@ -554,6 +563,12 @@ void rocksdb_restore_options_set_keep_log_files(rocksdb_restore_options_t* opt,
   opt->rep.keep_log_files = v;
 }
 
+
+void rocksdb_backup_engine_verify_backup(rocksdb_backup_engine_t* be,
+    uint32_t backup_id, char** errptr) {
+  SaveError(errptr, be->rep->VerifyBackup(static_cast<BackupID>(backup_id)));
+}
+
 void rocksdb_backup_engine_restore_db_from_latest_backup(
     rocksdb_backup_engine_t* be, const char* db_dir, const char* wal_dir,
     const rocksdb_restore_options_t* restore_options, char** errptr) {

db/column_family_test.cc

@@ -2827,7 +2827,28 @@ TEST_F(ColumnFamilyTest, CreateAndDestoryOptions) {
   ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
 }
 
+TEST_F(ColumnFamilyTest, CreateDropAndDestroy) {
+  ColumnFamilyHandle* cfh;
+  Open();
+  ASSERT_OK(db_->CreateColumnFamily(ColumnFamilyOptions(), "yoyo", &cfh));
+  ASSERT_OK(db_->Put(WriteOptions(), cfh, "foo", "bar"));
+  ASSERT_OK(db_->Flush(FlushOptions(), cfh));
+  ASSERT_OK(db_->DropColumnFamily(cfh));
+  ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
+}
+
 #ifndef ROCKSDB_LITE
+TEST_F(ColumnFamilyTest, CreateDropAndDestroyWithoutFileDeletion) {
+  ColumnFamilyHandle* cfh;
+  Open();
+  ASSERT_OK(db_->CreateColumnFamily(ColumnFamilyOptions(), "yoyo", &cfh));
+  ASSERT_OK(db_->Put(WriteOptions(), cfh, "foo", "bar"));
+  ASSERT_OK(db_->Flush(FlushOptions(), cfh));
+  ASSERT_OK(db_->DisableFileDeletions());
+  ASSERT_OK(db_->DropColumnFamily(cfh));
+  ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
+}
+
 TEST_F(ColumnFamilyTest, FlushCloseWALFiles) {
   SpecialEnv env(Env::Default());
   db_options_.env = &env;

db/compaction_picker.cc

@@ -366,7 +366,7 @@ bool CompactionPicker::IsRangeInCompaction(VersionStorageInfo* vstorage,
   assert(level < NumberLevels());
 
   vstorage->GetOverlappingInputs(level, smallest, largest, &inputs,
-                                 *level_index, level_index);
+                                 level_index ? *level_index : 0, level_index);
   return AreFilesInCompaction(inputs);
 }
 
@@ -947,6 +947,78 @@ void CompactionPicker::UnregisterCompaction(Compaction* c) {
   compactions_in_progress_.erase(c);
 }
 
+void CompactionPicker::PickFilesMarkedForCompaction(
+    const std::string& cf_name, VersionStorageInfo* vstorage, int* start_level,
+    int* output_level, CompactionInputFiles* start_level_inputs) {
+  if (vstorage->FilesMarkedForCompaction().empty()) {
+    return;
+  }
+
+  auto continuation = [&, cf_name](std::pair<int, FileMetaData*> level_file) {
+    // If it's being compacted it has nothing to do here.
+    // If this assert() fails that means that some function marked some
+    // files as being_compacted, but didn't call ComputeCompactionScore()
+    assert(!level_file.second->being_compacted);
+    *start_level = level_file.first;
+    *output_level =
+        (*start_level == 0) ? vstorage->base_level() : *start_level + 1;
+
+    if (*start_level == 0 && !level0_compactions_in_progress()->empty()) {
+      return false;
+    }
+
+    start_level_inputs->files = {level_file.second};
+    start_level_inputs->level = *start_level;
+    return ExpandInputsToCleanCut(cf_name, vstorage, start_level_inputs);
+  };
+
+  // take a chance on a random file first
+  Random64 rnd(/* seed */ reinterpret_cast<uint64_t>(vstorage));
+  size_t random_file_index = static_cast<size_t>(rnd.Uniform(
+      static_cast<uint64_t>(vstorage->FilesMarkedForCompaction().size())));
+
+  if (continuation(vstorage->FilesMarkedForCompaction()[random_file_index])) {
+    // found the compaction!
+    return;
+  }
+
+  for (auto& level_file : vstorage->FilesMarkedForCompaction()) {
+    if (continuation(level_file)) {
+      // found the compaction!
+      return;
+    }
+  }
+  start_level_inputs->files.clear();
+}
+
+bool CompactionPicker::GetOverlappingL0Files(
+    VersionStorageInfo* vstorage, CompactionInputFiles* start_level_inputs,
+    int output_level, int* parent_index) {
+  // Two level 0 compaction won't run at the same time, so don't need to worry
+  // about files on level 0 being compacted.
+  assert(level0_compactions_in_progress()->empty());
+  InternalKey smallest, largest;
+  GetRange(*start_level_inputs, &smallest, &largest);
+  // Note that the next call will discard the file we placed in
+  // c->inputs_[0] earlier and replace it with an overlapping set
+  // which will include the picked file.
+  start_level_inputs->files.clear();
+  vstorage->GetOverlappingInputs(0, &smallest, &largest,
+                                 &(start_level_inputs->files));
+
+  // If we include more L0 files in the same compaction run it can
+  // cause the 'smallest' and 'largest' key to get extended to a
+  // larger range. So, re-invoke GetRange to get the new key range
+  GetRange(*start_level_inputs, &smallest, &largest);
+  if (IsRangeInCompaction(vstorage, &smallest, &largest, output_level,
+                          parent_index)) {
+    return false;
+  }
+  assert(!start_level_inputs->files.empty());
+
+  return true;
+}
+
 bool LevelCompactionPicker::NeedsCompaction(
     const VersionStorageInfo* vstorage) const {
   if (!vstorage->ExpiredTtlFiles().empty()) {
@@ -1018,9 +1090,6 @@ class LevelCompactionBuilder {
   // otherwise, returns false.
   bool PickIntraL0Compaction();
 
-  // If there is any file marked for compaction, put put it into inputs.
-  void PickFilesMarkedForCompaction();
-
   void PickExpiredTtlFiles();
 
   const std::string& cf_name_;
@@ -1049,50 +1118,6 @@ class LevelCompactionBuilder {
   static const int kMinFilesForIntraL0Compaction = 4;
 };
 
-void LevelCompactionBuilder::PickFilesMarkedForCompaction() {
-  if (vstorage_->FilesMarkedForCompaction().empty()) {
-    return;
-  }
-
-  auto continuation = [&](std::pair<int, FileMetaData*> level_file) {
-    // If it's being compacted it has nothing to do here.
-    // If this assert() fails that means that some function marked some
-    // files as being_compacted, but didn't call ComputeCompactionScore()
-    assert(!level_file.second->being_compacted);
-    start_level_ = level_file.first;
-    output_level_ =
-        (start_level_ == 0) ? vstorage_->base_level() : start_level_ + 1;
-
-    if (start_level_ == 0 &&
-        !compaction_picker_->level0_compactions_in_progress()->empty()) {
-      return false;
-    }
-
-    start_level_inputs_.files = {level_file.second};
-    start_level_inputs_.level = start_level_;
-    return compaction_picker_->ExpandInputsToCleanCut(cf_name_, vstorage_,
-                                                      &start_level_inputs_);
-  };
-
-  // take a chance on a random file first
-  Random64 rnd(/* seed */ reinterpret_cast<uint64_t>(vstorage_));
-  size_t random_file_index = static_cast<size_t>(rnd.Uniform(
-      static_cast<uint64_t>(vstorage_->FilesMarkedForCompaction().size())));
-
-  if (continuation(vstorage_->FilesMarkedForCompaction()[random_file_index])) {
-    // found the compaction!
-    return;
-  }
-
-  for (auto& level_file : vstorage_->FilesMarkedForCompaction()) {
-    if (continuation(level_file)) {
-      // found the compaction!
-      return;
-    }
-  }
-  start_level_inputs_.files.clear();
-}
-
 void LevelCompactionBuilder::PickExpiredTtlFiles() {
   if (vstorage_->ExpiredTtlFiles().empty()) {
     return;
@@ -1182,7 +1207,9 @@ void LevelCompactionBuilder::SetupInitialFiles() {
   if (start_level_inputs_.empty()) {
     parent_index_ = base_index_ = -1;
 
-    PickFilesMarkedForCompaction();
+    // PickFilesMarkedForCompaction();
+    compaction_picker_->PickFilesMarkedForCompaction(
+        cf_name_, vstorage_, &start_level_, &output_level_, &start_level_inputs_);
     if (!start_level_inputs_.empty()) {
       is_manual_ = true;
       compaction_reason_ = CompactionReason::kFilesMarkedForCompaction;
@@ -1220,29 +1247,9 @@ void LevelCompactionBuilder::SetupInitialFiles() {
 
 bool LevelCompactionBuilder::SetupOtherL0FilesIfNeeded() {
   if (start_level_ == 0 && output_level_ != 0) {
-    // Two level 0 compaction won't run at the same time, so don't need to worry
-    // about files on level 0 being compacted.
-    assert(compaction_picker_->level0_compactions_in_progress()->empty());
-    InternalKey smallest, largest;
-    compaction_picker_->GetRange(start_level_inputs_, &smallest, &largest);
-    // Note that the next call will discard the file we placed in
-    // c->inputs_[0] earlier and replace it with an overlapping set
-    // which will include the picked file.
-    start_level_inputs_.files.clear();
-    vstorage_->GetOverlappingInputs(0, &smallest, &largest,
-                                    &start_level_inputs_.files);
-
-    // If we include more L0 files in the same compaction run it can
-    // cause the 'smallest' and 'largest' key to get extended to a
-    // larger range. So, re-invoke GetRange to get the new key range
-    compaction_picker_->GetRange(start_level_inputs_, &smallest, &largest);
-    if (compaction_picker_->IsRangeInCompaction(
-            vstorage_, &smallest, &largest, output_level_, &parent_index_)) {
-      return false;
-    }
+    return compaction_picker_->GetOverlappingL0Files(
+        vstorage_, &start_level_inputs_, output_level_, &parent_index_);
   }
-  assert(!start_level_inputs_.files.empty());
-
   return true;
 }
 

db/compaction_picker.h

@@ -175,6 +175,15 @@ class CompactionPicker {
                        const CompactionInputFiles& output_level_inputs,
                        std::vector<FileMetaData*>* grandparents);
 
+  void PickFilesMarkedForCompaction(const std::string& cf_name,
+                                    VersionStorageInfo* vstorage,
+                                    int* start_level, int* output_level,
+                                    CompactionInputFiles* start_level_inputs);
+
+  bool GetOverlappingL0Files(VersionStorageInfo* vstorage,
+                             CompactionInputFiles* start_level_inputs,
+                             int output_level, int* parent_index);
+
   // Register this compaction in the set of running compactions
   void RegisterCompaction(Compaction* c);
 

db/compaction_picker_test.cc

@@ -391,10 +391,10 @@ TEST_F(CompactionPickerTest, NeedsCompactionUniversal) {
   NewVersionStorage(1, kCompactionStyleUniversal);
   UniversalCompactionPicker universal_compaction_picker(
       ioptions_, &icmp_);
+  UpdateVersionStorageInfo();
   // must return false when there's no files.
   ASSERT_EQ(universal_compaction_picker.NeedsCompaction(vstorage_.get()),
             false);
-  UpdateVersionStorageInfo();
 
   // verify the trigger given different number of L0 files.
   for (int i = 1;
@@ -415,6 +415,7 @@ TEST_F(CompactionPickerTest, CompactionUniversalIngestBehindReservedLevel) {
   ioptions_.allow_ingest_behind = true;
   ioptions_.num_levels = 3;
   UniversalCompactionPicker universal_compaction_picker(ioptions_, &icmp_);
+  UpdateVersionStorageInfo();
   // must return false when there's no files.
   ASSERT_EQ(universal_compaction_picker.NeedsCompaction(vstorage_.get()),
             false);
@@ -448,6 +449,7 @@ TEST_F(CompactionPickerTest, CannotTrivialMoveUniversal) {
   mutable_cf_options_.compaction_options_universal.allow_trivial_move = true;
   NewVersionStorage(1, kCompactionStyleUniversal);
   UniversalCompactionPicker universal_compaction_picker(ioptions_, &icmp_);
+  UpdateVersionStorageInfo();
   // must return false when there's no files.
   ASSERT_EQ(universal_compaction_picker.NeedsCompaction(vstorage_.get()),
             false);

db/compaction_picker_universal.cc

@@ -162,7 +162,13 @@ bool UniversalCompactionPicker::IsInputFilesNonOverlapping(Compaction* c) {
 bool UniversalCompactionPicker::NeedsCompaction(
     const VersionStorageInfo* vstorage) const {
   const int kLevel0 = 0;
-  return vstorage->CompactionScore(kLevel0) >= 1;
+  if (vstorage->CompactionScore(kLevel0) >= 1) {
+    return true;
+  }
+  if (!vstorage->FilesMarkedForCompaction().empty()) {
+    return true;
+  }
+  return false;
 }
 
 void UniversalCompactionPicker::SortedRun::Dump(char* out_buf,
@@ -257,8 +263,9 @@ Compaction* UniversalCompactionPicker::PickCompaction(
       CalculateSortedRuns(*vstorage, ioptions_, mutable_cf_options);
 
   if (sorted_runs.size() == 0 ||
-      sorted_runs.size() <
-          (unsigned int)mutable_cf_options.level0_file_num_compaction_trigger) {
+      (vstorage->FilesMarkedForCompaction().empty() &&
+       sorted_runs.size() < (unsigned int)mutable_cf_options
+                                .level0_file_num_compaction_trigger)) {
     ROCKS_LOG_BUFFER(log_buffer, "[%s] Universal: nothing to do\n",
                      cf_name.c_str());
     TEST_SYNC_POINT_CALLBACK("UniversalCompactionPicker::PickCompaction:Return",
@@ -272,58 +279,73 @@ Compaction* UniversalCompactionPicker::PickCompaction(
       cf_name.c_str(), sorted_runs.size(), vstorage->LevelSummary(&tmp));
 
   // Check for size amplification first.
-  Compaction* c;
-  if ((c = PickCompactionToReduceSizeAmp(cf_name, mutable_cf_options, vstorage,
-                                         score, sorted_runs, log_buffer)) !=
-      nullptr) {
-    ROCKS_LOG_BUFFER(log_buffer, "[%s] Universal: compacting for size amp\n",
-                     cf_name.c_str());
-  } else {
-    // Size amplification is within limits. Try reducing read
-    // amplification while maintaining file size ratios.
-    unsigned int ratio =
-        mutable_cf_options.compaction_options_universal.size_ratio;
-
-    if ((c = PickCompactionToReduceSortedRuns(
-             cf_name, mutable_cf_options, vstorage, score, ratio, UINT_MAX,
-             sorted_runs, log_buffer)) != nullptr) {
-      ROCKS_LOG_BUFFER(log_buffer,
-                       "[%s] Universal: compacting for size ratio\n",
+  Compaction* c = nullptr;
+  if (sorted_runs.size() >=
+      static_cast<size_t>(
+          mutable_cf_options.level0_file_num_compaction_trigger)) {
+    if ((c = PickCompactionToReduceSizeAmp(cf_name, mutable_cf_options,
+                                           vstorage, score, sorted_runs,
+                                           log_buffer)) != nullptr) {
+      ROCKS_LOG_BUFFER(log_buffer, "[%s] Universal: compacting for size amp\n",
                        cf_name.c_str());
     } else {
-      // Size amplification and file size ratios are within configured limits.
-      // If max read amplification is exceeding configured limits, then force
-      // compaction without looking at filesize ratios and try to reduce
-      // the number of files to fewer than level0_file_num_compaction_trigger.
-      // This is guaranteed by NeedsCompaction()
-      assert(sorted_runs.size() >=
-             static_cast<size_t>(
-                 mutable_cf_options.level0_file_num_compaction_trigger));
-      // Get the total number of sorted runs that are not being compacted
-      int num_sr_not_compacted = 0;
-      for (size_t i = 0; i < sorted_runs.size(); i++) {
-        if (sorted_runs[i].being_compacted == false) {
-          num_sr_not_compacted++;
-        }
-      }
+      // Size amplification is within limits. Try reducing read
+      // amplification while maintaining file size ratios.
+      unsigned int ratio =
+          mutable_cf_options.compaction_options_universal.size_ratio;
 
-      // The number of sorted runs that are not being compacted is greater than
-      // the maximum allowed number of sorted runs
-      if (num_sr_not_compacted >
-          mutable_cf_options.level0_file_num_compaction_trigger) {
-        unsigned int num_files =
-            num_sr_not_compacted -
-            mutable_cf_options.level0_file_num_compaction_trigger + 1;
-        if ((c = PickCompactionToReduceSortedRuns(
-                 cf_name, mutable_cf_options, vstorage, score, UINT_MAX,
-                 num_files, sorted_runs, log_buffer)) != nullptr) {
-          ROCKS_LOG_BUFFER(log_buffer,
-                           "[%s] Universal: compacting for file num -- %u\n",
-                           cf_name.c_str(), num_files);
+      if ((c = PickCompactionToReduceSortedRuns(
+               cf_name, mutable_cf_options, vstorage, score, ratio, UINT_MAX,
+               sorted_runs, log_buffer)) != nullptr) {
+        ROCKS_LOG_BUFFER(log_buffer,
+                         "[%s] Universal: compacting for size ratio\n",
+                         cf_name.c_str());
+      } else {
+        // Size amplification and file size ratios are within configured limits.
+        // If max read amplification is exceeding configured limits, then force
+        // compaction without looking at filesize ratios and try to reduce
+        // the number of files to fewer than level0_file_num_compaction_trigger.
+        // This is guaranteed by NeedsCompaction()
+        assert(sorted_runs.size() >=
+               static_cast<size_t>(
+                   mutable_cf_options.level0_file_num_compaction_trigger));
+        // Get the total number of sorted runs that are not being compacted
+        int num_sr_not_compacted = 0;
+        for (size_t i = 0; i < sorted_runs.size(); i++) {
+          if (sorted_runs[i].being_compacted == false) {
+            num_sr_not_compacted++;
+          }
+        }
+
+        // The number of sorted runs that are not being compacted is greater
+        // than the maximum allowed number of sorted runs
+        if (num_sr_not_compacted >
+            mutable_cf_options.level0_file_num_compaction_trigger) {
+          unsigned int num_files =
+              num_sr_not_compacted -
+              mutable_cf_options.level0_file_num_compaction_trigger + 1;
+          if ((c = PickCompactionToReduceSortedRuns(
+                   cf_name, mutable_cf_options, vstorage, score, UINT_MAX,
+                   num_files, sorted_runs, log_buffer)) != nullptr) {
+            ROCKS_LOG_BUFFER(log_buffer,
+                             "[%s] Universal: compacting for file num -- %u\n",
+                             cf_name.c_str(), num_files);
+          }
         }
       }
     }
   }
+
+  if (c == nullptr) {
+    if ((c = PickDeleteTriggeredCompaction(cf_name, mutable_cf_options,
+                                           vstorage, score, sorted_runs,
+                                           log_buffer)) != nullptr) {
+      ROCKS_LOG_BUFFER(log_buffer,
+                       "[%s] Universal: delete triggered compaction\n",
+                       cf_name.c_str());
+    }
+  }
+
   if (c == nullptr) {
     TEST_SYNC_POINT_CALLBACK("UniversalCompactionPicker::PickCompaction:Return",
                              nullptr);
@@ -753,6 +775,125 @@ Compaction* UniversalCompactionPicker::PickCompactionToReduceSizeAmp(
       score, false /* deletion_compaction */,
       CompactionReason::kUniversalSizeAmplification);
 }
+
+// Pick files marked for compaction. Typically, files are marked by
+// CompactOnDeleteCollector due to the presence of tombstones.
+Compaction* UniversalCompactionPicker::PickDeleteTriggeredCompaction(
+    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+    VersionStorageInfo* vstorage, double score,
+    const std::vector<SortedRun>& /*sorted_runs*/, LogBuffer* /*log_buffer*/) {
+  CompactionInputFiles start_level_inputs;
+  int output_level;
+  std::vector<CompactionInputFiles> inputs;
+
+  if (vstorage->num_levels() == 1) {
+    // This is single level universal. Since we're basically trying to reclaim
+    // space by processing files marked for compaction due to high tombstone
+    // density, let's do the same thing as compaction to reduce size amp which
+    // has the same goals.
+    bool compact = false;
+
+    start_level_inputs.level = 0;
+    start_level_inputs.files.clear();
+    output_level = 0;
+    for (FileMetaData* f : vstorage->LevelFiles(0)) {
+      if (f->marked_for_compaction) {
+        compact = true;
+      }
+      if (compact) {
+        start_level_inputs.files.push_back(f);
+      }
+    }
+    if (start_level_inputs.size() <= 1) {
+      // If only the last file in L0 is marked for compaction, ignore it
+      return nullptr;
+    }
+    inputs.push_back(start_level_inputs);
+  } else {
+    int start_level;
+
+    // For multi-level universal, the strategy is to make this look more like
+    // leveled. We pick one of the files marked for compaction and compact with
+    // overlapping files in the adjacent level.
+    PickFilesMarkedForCompaction(cf_name, vstorage, &start_level, &output_level,
+                                 &start_level_inputs);
+    if (start_level_inputs.empty()) {
+      return nullptr;
+    }
+
+    // Pick the first non-empty level after the start_level
+    for (output_level = start_level + 1; output_level < vstorage->num_levels();
+         output_level++) {
+      if (vstorage->NumLevelFiles(output_level) != 0) {
+        break;
+      }
+    }
+
+    // If all higher levels are empty, pick the highest level as output level
+    if (output_level == vstorage->num_levels()) {
+      if (start_level == 0) {
+        output_level = vstorage->num_levels() - 1;
+      } else {
+        // If start level is non-zero and all higher levels are empty, this
+        // compaction will translate into a trivial move. Since the idea is
+        // to reclaim space and trivial move doesn't help with that, we
+        // skip compaction in this case and return nullptr
+        return nullptr;
+      }
+    }
+    if (ioptions_.allow_ingest_behind &&
+        output_level == vstorage->num_levels() - 1) {
+      assert(output_level > 1);
+      output_level--;
+    }
+
+    if (output_level != 0) {
+      if (start_level == 0) {
+        if (!GetOverlappingL0Files(vstorage, &start_level_inputs, output_level,
+                                   nullptr)) {
+          return nullptr;
+        }
+      }
+
+      CompactionInputFiles output_level_inputs;
+      int parent_index = -1;
+
+      output_level_inputs.level = output_level;
+      if (!SetupOtherInputs(cf_name, mutable_cf_options, vstorage,
+                            &start_level_inputs, &output_level_inputs,
+                            &parent_index, -1)) {
+        return nullptr;
+      }
+      inputs.push_back(start_level_inputs);
+      if (!output_level_inputs.empty()) {
+        inputs.push_back(output_level_inputs);
+      }
+      if (FilesRangeOverlapWithCompaction(inputs, output_level)) {
+        return nullptr;
+      }
+    } else {
+      inputs.push_back(start_level_inputs);
+    }
+  }
+
+  uint64_t estimated_total_size = 0;
+  // Use size of the output level as estimated file size
+  for (FileMetaData* f : vstorage->LevelFiles(output_level)) {
+    estimated_total_size += f->fd.GetFileSize();
+  }
+  uint32_t path_id =
+      GetPathId(ioptions_, mutable_cf_options, estimated_total_size);
+  return new Compaction(
+      vstorage, ioptions_, mutable_cf_options, std::move(inputs), output_level,
+      MaxFileSizeForLevel(mutable_cf_options, output_level,
+                          kCompactionStyleUniversal),
+      /* max_grandparent_overlap_bytes */ LLONG_MAX, path_id,
+      GetCompressionType(ioptions_, vstorage, mutable_cf_options, output_level,
+                         1),
+      /* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ true,
+      score, false /* deletion_compaction */,
+      CompactionReason::kFilesMarkedForCompaction);
+}
 }  // namespace rocksdb
 
 #endif  // !ROCKSDB_LITE

db/compaction_picker_universal.h

@@ -73,6 +73,11 @@ class UniversalCompactionPicker : public CompactionPicker {
       VersionStorageInfo* vstorage, double score,
       const std::vector<SortedRun>& sorted_runs, LogBuffer* log_buffer);
 
+  Compaction* PickDeleteTriggeredCompaction(
+      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+      VersionStorageInfo* vstorage, double score,
+      const std::vector<SortedRun>& sorted_runs, LogBuffer* log_buffer);
+
   // Used in universal compaction when the enabled_trivial_move
   // option is set. Checks whether there are any overlapping files
   // in the input. Returns true if the input files are non

db/db_block_cache_test.cc

@@ -390,7 +390,10 @@ class MockCache : public LRUCache {
   static uint32_t high_pri_insert_count;
   static uint32_t low_pri_insert_count;
 
-  MockCache() : LRUCache(1 << 25, 0, false, 0.0) {}
+  MockCache()
+      : LRUCache((size_t)1 << 25 /*capacity*/, 0 /*num_shard_bits*/,
+                 false /*strict_capacity_limit*/, 0.0 /*high_pri_pool_ratio*/) {
+  }
 
   virtual Status Insert(const Slice& key, void* value, size_t charge,
                         void (*deleter)(const Slice& key, void* value),

db/db_compaction_filter_test.cc

@@ -45,6 +45,7 @@ class DBTestCompactionFilterWithCompactParam
   }
 };
 
+#ifndef ROCKSDB_VALGRIND_RUN
 INSTANTIATE_TEST_CASE_P(
     DBTestCompactionFilterWithCompactOption,
     DBTestCompactionFilterWithCompactParam,
@@ -53,6 +54,12 @@ INSTANTIATE_TEST_CASE_P(
                       DBTestBase::OptionConfig::kUniversalCompactionMultiLevel,
                       DBTestBase::OptionConfig::kLevelSubcompactions,
                       DBTestBase::OptionConfig::kUniversalSubcompactions));
+#else
+// Run fewer cases in valgrind
+INSTANTIATE_TEST_CASE_P(DBTestCompactionFilterWithCompactOption,
+                        DBTestCompactionFilterWithCompactParam,
+                        ::testing::Values(DBTestBase::OptionConfig::kDefault));
+#endif  // ROCKSDB_VALGRIND_RUN
 
 class KeepFilter : public CompactionFilter {
  public:

db/db_compaction_test.cc

@@ -251,6 +251,7 @@ const SstFileMetaData* PickFileRandomly(
 }
 }  // anonymous namespace
 
+#ifndef ROCKSDB_VALGRIND_RUN
 // All the TEST_P tests run once with sub_compactions disabled (i.e.
 // options.max_subcompactions = 1) and once with it enabled
 TEST_P(DBCompactionTestWithParam, CompactionDeletionTrigger) {
@@ -293,6 +294,7 @@ TEST_P(DBCompactionTestWithParam, CompactionDeletionTrigger) {
     ASSERT_GT(db_size[0] / 3, db_size[1]);
   }
 }
+#endif  // ROCKSDB_VALGRIND_RUN
 
 TEST_P(DBCompactionTestWithParam, CompactionsPreserveDeletes) {
   //  For each options type we test following

db/db_impl_files.cc

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

db/db_impl_write.cc

@@ -1281,7 +1281,11 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
   // In case of pipelined write is enabled, wait for all pending memtable
   // writers.
   if (immutable_db_options_.enable_pipelined_write) {
+    // Memtable writers may call DB::Get in case max_successive_merges > 0,
+    // which may lock mutex. Unlocking mutex here to avoid deadlock.
+    mutex_.Unlock();
     write_thread_.WaitForMemTableWriters();
+    mutex_.Lock();
   }
 
   // Attempt to switch to a new memtable and trigger flush of old.

db/db_properties_test.cc

@@ -177,17 +177,16 @@ void ParseTablePropertiesString(std::string tp_string, TableProperties* tp) {
   std::replace(tp_string.begin(), tp_string.end(), ';', ' ');
   std::replace(tp_string.begin(), tp_string.end(), '=', ' ');
   ResetTableProperties(tp);
-
   sscanf(tp_string.c_str(),
          "# data blocks %" SCNu64 " # entries %" SCNu64 " raw key size %" SCNu64
          " raw average key size %lf "
          " raw value size %" SCNu64
          " raw average value size %lf "
-         " data block size %" SCNu64 " index block size %" SCNu64
-         " filter block size %" SCNu64,
+         " data block size %" SCNu64 " index block size (user-key? %" SCNu64
+         ") %" SCNu64 " filter block size %" SCNu64,
          &tp->num_data_blocks, &tp->num_entries, &tp->raw_key_size,
          &dummy_double, &tp->raw_value_size, &dummy_double, &tp->data_size,
-         &tp->index_size, &tp->filter_size);
+         &tp->index_key_is_user_key, &tp->index_size, &tp->filter_size);
 }
 
 void VerifySimilar(uint64_t a, uint64_t b, double bias) {
@@ -224,7 +223,8 @@ void GetExpectedTableProperties(TableProperties* expected_tp,
                                 const int kKeySize, const int kValueSize,
                                 const int kKeysPerTable, const int kTableCount,
                                 const int kBloomBitsPerKey,
-                                const size_t kBlockSize) {
+                                const size_t kBlockSize,
+                                const bool index_key_is_user_key) {
   const int kKeyCount = kTableCount * kKeysPerTable;
   const int kAvgSuccessorSize = kKeySize / 5;
   const int kEncodingSavePerKey = kKeySize / 4;
@@ -238,7 +238,8 @@ void GetExpectedTableProperties(TableProperties* expected_tp,
   expected_tp->data_size =
       kTableCount * (kKeysPerTable * (kKeySize + 8 + kValueSize));
   expected_tp->index_size =
-      expected_tp->num_data_blocks * (kAvgSuccessorSize + 8);
+      expected_tp->num_data_blocks *
+      (kAvgSuccessorSize + (index_key_is_user_key ? 0 : 8));
   expected_tp->filter_size =
       kTableCount * (kKeysPerTable * kBloomBitsPerKey / 8);
 }
@@ -315,14 +316,14 @@ TEST_F(DBPropertiesTest, AggregatedTableProperties) {
     }
     std::string property;
     db_->GetProperty(DB::Properties::kAggregatedTableProperties, &property);
+    TableProperties output_tp;
+    ParseTablePropertiesString(property, &output_tp);
+    bool index_key_is_user_key = output_tp.index_key_is_user_key > 0;
 
     TableProperties expected_tp;
     GetExpectedTableProperties(&expected_tp, kKeySize, kValueSize,
                                kKeysPerTable, kTableCount, kBloomBitsPerKey,
-                               table_options.block_size);
-
-    TableProperties output_tp;
-    ParseTablePropertiesString(property, &output_tp);
+                               table_options.block_size, index_key_is_user_key);
 
     VerifyTableProperties(expected_tp, output_tp);
   }
@@ -489,6 +490,7 @@ TEST_F(DBPropertiesTest, AggregatedTablePropertiesAtLevel) {
     }
     db_->GetProperty(DB::Properties::kAggregatedTableProperties, &tp_string);
     ParseTablePropertiesString(tp_string, &tp);
+    bool index_key_is_user_key = tp.index_key_is_user_key > 0;
     ASSERT_EQ(sum_tp.data_size, tp.data_size);
     ASSERT_EQ(sum_tp.index_size, tp.index_size);
     ASSERT_EQ(sum_tp.filter_size, tp.filter_size);
@@ -497,9 +499,9 @@ TEST_F(DBPropertiesTest, AggregatedTablePropertiesAtLevel) {
     ASSERT_EQ(sum_tp.num_data_blocks, tp.num_data_blocks);
     ASSERT_EQ(sum_tp.num_entries, tp.num_entries);
     if (table > 3) {
-      GetExpectedTableProperties(&expected_tp, kKeySize, kValueSize,
-                                 kKeysPerTable, table, kBloomBitsPerKey,
-                                 table_options.block_size);
+      GetExpectedTableProperties(
+          &expected_tp, kKeySize, kValueSize, kKeysPerTable, table,
+          kBloomBitsPerKey, table_options.block_size, index_key_is_user_key);
       // Gives larger bias here as index block size, filter block size,
       // and data block size become much harder to estimate in this test.
       VerifyTableProperties(tp, expected_tp, 0.5, 0.4, 0.4, 0.25);

db/db_test.cc

@@ -2561,6 +2561,7 @@ class ModelDB : public DB {
   std::string name_ = "";
 };
 
+#ifndef ROCKSDB_VALGRIND_RUN
 static std::string RandomKey(Random* rnd, int minimum = 0) {
   int len;
   do {
@@ -2717,6 +2718,7 @@ TEST_P(DBTestRandomized, Randomized) {
   if (model_snap != nullptr) model.ReleaseSnapshot(model_snap);
   if (db_snap != nullptr) db_->ReleaseSnapshot(db_snap);
 }
+#endif  // ROCKSDB_VALGRIND_RUN
 
 TEST_F(DBTest, BlockBasedTablePrefixIndexTest) {
   // create a DB with block prefix index

db/db_universal_compaction_test.cc

@@ -10,6 +10,7 @@
 #include "db/db_test_util.h"
 #include "port/stack_trace.h"
 #if !defined(ROCKSDB_LITE)
+#include "rocksdb/utilities/table_properties_collectors.h"
 #include "util/sync_point.h"
 
 namespace rocksdb {
@@ -40,6 +41,12 @@ class DBTestUniversalCompaction : public DBTestUniversalCompactionBase {
       DBTestUniversalCompactionBase("/db_universal_compaction_test") {}
 };
 
+class DBTestUniversalDeleteTrigCompaction : public DBTestBase {
+ public:
+  DBTestUniversalDeleteTrigCompaction()
+      : DBTestBase("/db_universal_compaction_test") {}
+};
+
 namespace {
 void VerifyCompactionResult(
     const ColumnFamilyMetaData& cf_meta,
@@ -655,7 +662,7 @@ TEST_P(DBTestUniversalCompaction, UniversalCompactionTargetLevel) {
   ASSERT_EQ("0,0,0,0,1", FilesPerLevel(0));
 }
 
-
+#ifndef ROCKSDB_VALGRIND_RUN
 class DBTestUniversalCompactionMultiLevels
     : public DBTestUniversalCompactionBase {
  public:
@@ -691,6 +698,7 @@ TEST_P(DBTestUniversalCompactionMultiLevels, UniversalCompactionMultiLevels) {
     ASSERT_EQ(Get(1, Key(i % num_keys)), Key(i));
   }
 }
+
 // Tests universal compaction with trivial move enabled
 TEST_P(DBTestUniversalCompactionMultiLevels, UniversalCompactionTrivialMove) {
   int32_t trivial_move = 0;
@@ -933,6 +941,7 @@ INSTANTIATE_TEST_CASE_P(DBTestUniversalCompactionParallel,
                         DBTestUniversalCompactionParallel,
                         ::testing::Combine(::testing::Values(1, 10),
                                            ::testing::Values(false)));
+#endif  // ROCKSDB_VALGRIND_RUN
 
 TEST_P(DBTestUniversalCompaction, UniversalCompactionOptions) {
   Options options = CurrentOptions();
@@ -1148,6 +1157,7 @@ TEST_P(DBTestUniversalCompaction, UniversalCompactionCompressRatio2) {
   ASSERT_LT(TotalSize(), 120000U * 12 * 0.8 + 120000 * 2);
 }
 
+#ifndef ROCKSDB_VALGRIND_RUN
 // Test that checks trivial move in universal compaction
 TEST_P(DBTestUniversalCompaction, UniversalCompactionTrivialMoveTest1) {
   int32_t trivial_move = 0;
@@ -1240,6 +1250,7 @@ TEST_P(DBTestUniversalCompaction, UniversalCompactionTrivialMoveTest2) {
 
   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
 }
+#endif  // ROCKSDB_VALGRIND_RUN
 
 TEST_P(DBTestUniversalCompaction, UniversalCompactionFourPaths) {
   Options options = CurrentOptions();
@@ -1845,6 +1856,241 @@ INSTANTIATE_TEST_CASE_P(DBTestUniversalManualCompactionOutputPathId,
                         ::testing::Combine(::testing::Values(1, 8),
                                            ::testing::Bool()));
 
+TEST_F(DBTestUniversalDeleteTrigCompaction, BasicL0toL1) {
+  const int kNumKeys = 3000;
+  const int kWindowSize = 100;
+  const int kNumDelsTrigger = 90;
+
+  Options opts = CurrentOptions();
+  opts.table_properties_collector_factories.emplace_back(
+      NewCompactOnDeletionCollectorFactory(kWindowSize, kNumDelsTrigger));
+  opts.compaction_style = kCompactionStyleUniversal;
+  opts.level0_file_num_compaction_trigger = 2;
+  opts.compression = kNoCompression;
+  opts.compaction_options_universal.size_ratio = 10;
+  opts.compaction_options_universal.min_merge_width = 2;
+  opts.compaction_options_universal.max_size_amplification_percent = 200;
+  Reopen(opts);
+
+  // add an L1 file to prevent tombstones from dropping due to obsolescence
+  // during flush
+  int i;
+  for (i = 0; i < 2000; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  //  MoveFilesToLevel(6);
+  dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+
+  for (i = 1999; i < kNumKeys; ++i) {
+    if (i >= kNumKeys - kWindowSize &&
+        i < kNumKeys - kWindowSize + kNumDelsTrigger) {
+      Delete(Key(i));
+    } else {
+      Put(Key(i), "val");
+    }
+  }
+  Flush();
+
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+  ASSERT_GT(NumTableFilesAtLevel(6), 0);
+}
+
+TEST_F(DBTestUniversalDeleteTrigCompaction, SingleLevel) {
+  const int kNumKeys = 3000;
+  const int kWindowSize = 100;
+  const int kNumDelsTrigger = 90;
+
+  Options opts = CurrentOptions();
+  opts.table_properties_collector_factories.emplace_back(
+      NewCompactOnDeletionCollectorFactory(kWindowSize, kNumDelsTrigger));
+  opts.compaction_style = kCompactionStyleUniversal;
+  opts.level0_file_num_compaction_trigger = 2;
+  opts.compression = kNoCompression;
+  opts.num_levels = 1;
+  opts.compaction_options_universal.size_ratio = 10;
+  opts.compaction_options_universal.min_merge_width = 2;
+  opts.compaction_options_universal.max_size_amplification_percent = 200;
+  Reopen(opts);
+
+  // add an L1 file to prevent tombstones from dropping due to obsolescence
+  // during flush
+  int i;
+  for (i = 0; i < 2000; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+
+  for (i = 1999; i < kNumKeys; ++i) {
+    if (i >= kNumKeys - kWindowSize &&
+        i < kNumKeys - kWindowSize + kNumDelsTrigger) {
+      Delete(Key(i));
+    } else {
+      Put(Key(i), "val");
+    }
+  }
+  Flush();
+
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ(1, NumTableFilesAtLevel(0));
+}
+
+TEST_F(DBTestUniversalDeleteTrigCompaction, MultipleLevels) {
+  const int kWindowSize = 100;
+  const int kNumDelsTrigger = 90;
+
+  Options opts = CurrentOptions();
+  opts.table_properties_collector_factories.emplace_back(
+      NewCompactOnDeletionCollectorFactory(kWindowSize, kNumDelsTrigger));
+  opts.compaction_style = kCompactionStyleUniversal;
+  opts.level0_file_num_compaction_trigger = 4;
+  opts.compression = kNoCompression;
+  opts.compaction_options_universal.size_ratio = 10;
+  opts.compaction_options_universal.min_merge_width = 2;
+  opts.compaction_options_universal.max_size_amplification_percent = 200;
+  Reopen(opts);
+
+  // add an L1 file to prevent tombstones from dropping due to obsolescence
+  // during flush
+  int i;
+  for (i = 0; i < 500; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  for (i = 500; i < 1000; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  for (i = 1000; i < 1500; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  for (i = 1500; i < 2000; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+  ASSERT_GT(NumTableFilesAtLevel(6), 0);
+
+  for (i = 1999; i < 2333; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  for (i = 2333; i < 2666; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  for (i = 2666; i < 2999; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+  ASSERT_GT(NumTableFilesAtLevel(6), 0);
+  ASSERT_GT(NumTableFilesAtLevel(5), 0);
+
+  for (i = 1900; i < 2100; ++i) {
+    Delete(Key(i));
+  }
+  Flush();
+
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+  ASSERT_EQ(0, NumTableFilesAtLevel(1));
+  ASSERT_EQ(0, NumTableFilesAtLevel(2));
+  ASSERT_EQ(0, NumTableFilesAtLevel(3));
+  ASSERT_EQ(0, NumTableFilesAtLevel(4));
+  ASSERT_EQ(0, NumTableFilesAtLevel(5));
+  ASSERT_GT(NumTableFilesAtLevel(6), 0);
+}
+
+TEST_F(DBTestUniversalDeleteTrigCompaction, OverlappingL0) {
+  const int kWindowSize = 100;
+  const int kNumDelsTrigger = 90;
+
+  Options opts = CurrentOptions();
+  opts.table_properties_collector_factories.emplace_back(
+      NewCompactOnDeletionCollectorFactory(kWindowSize, kNumDelsTrigger));
+  opts.compaction_style = kCompactionStyleUniversal;
+  opts.level0_file_num_compaction_trigger = 5;
+  opts.compression = kNoCompression;
+  opts.compaction_options_universal.size_ratio = 10;
+  opts.compaction_options_universal.min_merge_width = 2;
+  opts.compaction_options_universal.max_size_amplification_percent = 200;
+  Reopen(opts);
+
+  // add an L1 file to prevent tombstones from dropping due to obsolescence
+  // during flush
+  int i;
+  for (i = 0; i < 2000; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  for (i = 2000; i < 3000; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  for (i = 3500; i < 4000; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  for (i = 2900; i < 3100; ++i) {
+    Delete(Key(i));
+  }
+  Flush();
+
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ(2, NumTableFilesAtLevel(0));
+  ASSERT_GT(NumTableFilesAtLevel(6), 0);
+}
+
+TEST_F(DBTestUniversalDeleteTrigCompaction, IngestBehind) {
+  const int kNumKeys = 3000;
+  const int kWindowSize = 100;
+  const int kNumDelsTrigger = 90;
+
+  Options opts = CurrentOptions();
+  opts.table_properties_collector_factories.emplace_back(
+      NewCompactOnDeletionCollectorFactory(kWindowSize, kNumDelsTrigger));
+  opts.compaction_style = kCompactionStyleUniversal;
+  opts.level0_file_num_compaction_trigger = 2;
+  opts.compression = kNoCompression;
+  opts.allow_ingest_behind = true;
+  opts.compaction_options_universal.size_ratio = 10;
+  opts.compaction_options_universal.min_merge_width = 2;
+  opts.compaction_options_universal.max_size_amplification_percent = 200;
+  Reopen(opts);
+
+  // add an L1 file to prevent tombstones from dropping due to obsolescence
+  // during flush
+  int i;
+  for (i = 0; i < 2000; ++i) {
+    Put(Key(i), "val");
+  }
+  Flush();
+  //  MoveFilesToLevel(6);
+  dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+
+  for (i = 1999; i < kNumKeys; ++i) {
+    if (i >= kNumKeys - kWindowSize &&
+        i < kNumKeys - kWindowSize + kNumDelsTrigger) {
+      Delete(Key(i));
+    } else {
+      Put(Key(i), "val");
+    }
+  }
+  Flush();
+
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+  ASSERT_EQ(0, NumTableFilesAtLevel(6));
+  ASSERT_GT(NumTableFilesAtLevel(5), 0);
+}
+
 }  // namespace rocksdb
 
 #endif  // !defined(ROCKSDB_LITE)

db/internal_stats.cc

@@ -788,8 +788,8 @@ bool InternalStats::HandleEstimateTableReadersMem(uint64_t* value,
 }
 
 bool InternalStats::HandleEstimateLiveDataSize(uint64_t* value, DBImpl* /*db*/,
-                                               Version* /*version*/) {
-  const auto* vstorage = cfd_->current()->storage_info();
+                                               Version* version) {
+  const auto* vstorage = version->storage_info();
   *value = vstorage->EstimateLiveDataSize();
   return true;
 }

db/write_batch.cc

@@ -39,6 +39,7 @@
 #include <stdexcept>
 #include <type_traits>
 #include <vector>
+#include <csignal>
 
 #include "db/column_family.h"
 #include "db/db_impl.h"
@@ -513,6 +514,7 @@ Status WriteBatch::Iterate(Handler* handler) const {
         handler->MarkBeginPrepare();
         empty_batch = false;
         if (handler->WriteAfterCommit()) {
+          std::raise(SIGINT);
           s = Status::NotSupported(
               "WritePrepared txn tag when write_after_commit_ is enabled (in "
               "default WriteCommitted mode). If it is not due to corruption, "

env/env_posix.cc

@@ -241,9 +241,9 @@ class PosixEnv : public Env {
                                                   size, options));
         } else {
           s = IOError("while mmap file for read", fname, errno);
+          close(fd);
         }
       }
-      close(fd);
     } else {
       if (options.use_direct_reads && !options.use_mmap_reads) {
 #ifdef OS_MACOSX

env/env_test.cc

@@ -232,10 +232,10 @@ TEST_F(EnvPosixTest, MemoryMappedFileBuffer) {
 
   ASSERT_OK(status);
   ASSERT_NE(nullptr, mmap_buffer.get());
-  ASSERT_NE(nullptr, mmap_buffer->base);
-  ASSERT_EQ(kFileBytes, mmap_buffer->length);
-  std::string actual_data(static_cast<char*>(mmap_buffer->base),
-                          mmap_buffer->length);
+  ASSERT_NE(nullptr, mmap_buffer->GetBase());
+  ASSERT_EQ(kFileBytes, mmap_buffer->GetLen());
+  std::string actual_data(reinterpret_cast<const char*>(mmap_buffer->GetBase()),
+                          mmap_buffer->GetLen());
   ASSERT_EQ(expected_data, actual_data);
 }
 
@@ -1437,10 +1437,8 @@ TEST_P(EnvPosixTestWithParam, PosixRandomRWFile) {
 
   std::unique_ptr<RandomRWFile> file;
 
-#ifdef OS_LINUX
   // Cannot open non-existing file.
   ASSERT_NOK(env_->NewRandomRWFile(path, &file, EnvOptions()));
-#endif
 
   // Create the file using WriteableFile
   {

env/io_posix.cc

@@ -456,6 +456,7 @@ PosixMmapReadableFile::~PosixMmapReadableFile() {
     fprintf(stdout, "failed to munmap %p length %" ROCKSDB_PRIszt " \n",
             mmapped_region_, length_);
   }
+  close(fd_);
 }
 
 Status PosixMmapReadableFile::Read(uint64_t offset, size_t n, Slice* result,
@@ -1054,7 +1055,7 @@ Status PosixRandomRWFile::Close() {
 
 PosixMemoryMappedFileBuffer::~PosixMemoryMappedFileBuffer() {
   // TODO should have error handling though not much we can do...
-  munmap(this->base, length);
+  munmap(this->base_, length_);
 }
 
 /*

include/rocksdb/c.h

@@ -145,6 +145,10 @@ extern ROCKSDB_LIBRARY_API rocksdb_backup_engine_t* rocksdb_backup_engine_open(
 extern ROCKSDB_LIBRARY_API void rocksdb_backup_engine_create_new_backup(
     rocksdb_backup_engine_t* be, rocksdb_t* db, char** errptr);
 
+extern ROCKSDB_LIBRARY_API void rocksdb_backup_engine_create_new_backup_flush(
+    rocksdb_backup_engine_t* be, rocksdb_t* db, unsigned char flush_before_backup,
+    char** errptr);
+
 extern ROCKSDB_LIBRARY_API void rocksdb_backup_engine_purge_old_backups(
     rocksdb_backup_engine_t* be, uint32_t num_backups_to_keep, char** errptr);
 
@@ -155,6 +159,10 @@ extern ROCKSDB_LIBRARY_API void rocksdb_restore_options_destroy(
 extern ROCKSDB_LIBRARY_API void rocksdb_restore_options_set_keep_log_files(
     rocksdb_restore_options_t* opt, int v);
 
+extern ROCKSDB_LIBRARY_API void
+rocksdb_backup_engine_verify_backup(rocksdb_backup_engine_t* be,
+    uint32_t backup_id, char** errptr);
+
 extern ROCKSDB_LIBRARY_API void
 rocksdb_backup_engine_restore_db_from_latest_backup(
     rocksdb_backup_engine_t* be, const char* db_dir, const char* wal_dir,

include/rocksdb/cache.h

@@ -47,6 +47,15 @@ struct LRUCacheOptions {
   bool strict_capacity_limit = false;
 
   // Percentage of cache reserved for high priority entries.
+  // If greater than zero, the LRU list will be split into a high-pri
+  // list and a low-pri list. High-pri entries will be insert to the
+  // tail of high-pri list, while low-pri entries will be first inserted to
+  // the low-pri list (the midpoint). This is refered to as
+  // midpoint insertion strategy to make entries never get hit in cache
+  // age out faster.
+  //
+  // See also
+  // BlockBasedTableOptions::cache_index_and_filter_blocks_with_high_priority.
   double high_pri_pool_ratio = 0.0;
 
   LRUCacheOptions() {}

include/rocksdb/env.h

@@ -42,7 +42,7 @@ class SequentialFile;
 class Slice;
 class WritableFile;
 class RandomRWFile;
-struct MemoryMappedFileBuffer;
+class MemoryMappedFileBuffer;
 class Directory;
 struct DBOptions;
 struct ImmutableDBOptions;
@@ -822,13 +822,24 @@ class RandomRWFile {
 
 // MemoryMappedFileBuffer object represents a memory-mapped file's raw buffer.
 // Subclasses should release the mapping upon destruction.
-struct MemoryMappedFileBuffer {
+class MemoryMappedFileBuffer {
+public:
   MemoryMappedFileBuffer(void* _base, size_t _length)
-      : base(_base), length(_length) {}
+      : base_(_base), length_(_length) {}
+
   virtual ~MemoryMappedFileBuffer() = 0;
 
-  void* const base;
-  const size_t length;
+  // We do not want to unmap this twice. We can make this class
+  // movable if desired, however, since
+  MemoryMappedFileBuffer(const MemoryMappedFileBuffer&) = delete;
+  MemoryMappedFileBuffer& operator=(const MemoryMappedFileBuffer&) = delete;
+
+  void*       GetBase() const { return base_;   }
+  size_t      GetLen() const  { return length_; }
+
+protected:
+  void*        base_;
+  const size_t length_;
 };
 
 // Directory object represents collection of files and implements

include/rocksdb/table.h

@@ -214,8 +214,11 @@ struct BlockBasedTableOptions {
   // encode compressed blocks with LZ4, BZip2 and Zlib compression. If you
   // don't plan to run RocksDB before version 3.10, you should probably use
   // this.
-  // This option only affects newly written tables. When reading existing tables,
-  // the information about version is read from the footer.
+  // 3 -- Can be read by RocksDB's versions since 5.15. Changes the way we
+  // encode the keys in index blocks. If you don't plan to run RocksDB before
+  // version 5.15, you should probably use this.
+  // This option only affects newly written tables. When reading existing
+  // tables, the information about version is read from the footer.
   uint32_t format_version = 2;
 
   // Store index blocks on disk in compressed format. Changing this option to

include/rocksdb/table_properties.h

@@ -33,6 +33,7 @@ struct TablePropertiesNames {
   static const std::string kIndexSize;
   static const std::string kIndexPartitions;
   static const std::string kTopLevelIndexSize;
+  static const std::string kIndexKeyIsUserKey;
   static const std::string kFilterSize;
   static const std::string kRawKeySize;
   static const std::string kRawValueSize;
@@ -134,6 +135,9 @@ struct TableProperties {
   uint64_t index_partitions = 0;
   // Size of the top-level index if kTwoLevelIndexSearch is used
   uint64_t top_level_index_size = 0;
+  // Whether the index key is user key. Otherwise it includes 8 byte of sequence
+  // number added by internal key format.
+  uint64_t index_key_is_user_key = 0;
   // the size of filter block.
   uint64_t filter_size = 0;
   // total raw key size

java/rocksjni/transaction_db.cc

@@ -309,7 +309,7 @@ jobject Java_org_rocksdb_TransactionDB_getLockStatusData(JNIEnv* env,
 
   const rocksdb::HashMapJni::FnMapKV<const int32_t, const rocksdb::KeyLockInfo>
       fn_map_kv =
-          [env, txn_db, &lock_status_data](
+          [env](
               const std::pair<const int32_t, const rocksdb::KeyLockInfo>&
                   pair) {
             const jobject jlong_column_family_id =

memtable/inlineskiplist_test.cc

@@ -309,6 +309,7 @@ TEST_F(InlineSkipTest, InsertWithHint_CompatibleWithInsertWithoutHint) {
   Validate(&list);
 }
 
+#ifndef ROCKSDB_VALGRIND_RUN
 // We want to make sure that with a single writer and multiple
 // concurrent readers (with no synchronization other than when a
 // reader's iterator is created), the reader always observes all the
@@ -635,6 +636,7 @@ TEST_F(InlineSkipTest, ConcurrentInsert1) { RunConcurrentInsert(1); }
 TEST_F(InlineSkipTest, ConcurrentInsert2) { RunConcurrentInsert(2); }
 TEST_F(InlineSkipTest, ConcurrentInsert3) { RunConcurrentInsert(3); }
 
+#endif  // ROCKSDB_VALGRIND_RUN
 }  // namespace rocksdb
 
 int main(int argc, char** argv) {

port/port_posix.cc

@@ -188,8 +188,6 @@ int GetMaxOpenFiles() {
 void *cacheline_aligned_alloc(size_t size) {
 #if __GNUC__ < 5 && defined(__SANITIZE_ADDRESS__)
   return malloc(size);
-#elif defined(_ISOC11_SOURCE)
-  return aligned_alloc(CACHE_LINE_SIZE, size);
 #elif ( _POSIX_C_SOURCE >= 200112L || _XOPEN_SOURCE >= 600 || defined(__APPLE__))
   void *m;
   errno = posix_memalign(&m, CACHE_LINE_SIZE, size);

port/win/env_win.cc

@@ -112,6 +112,15 @@ Status WinEnvIO::DeleteFile(const std::string& fname) {
   return result;
 }
 
+Status WinEnvIO::Truncate(const std::string& fname, size_t size) {
+  Status s;
+  int result = truncate(fname.c_str(), size);
+  if (result != 0) {
+    s = IOError("Failed to truncate: " + fname, errno);
+  }
+  return s;
+}
+
 Status WinEnvIO::GetCurrentTime(int64_t* unix_time) {
   time_t time = std::time(nullptr);
   if (time == (time_t)(-1)) {
@@ -344,7 +353,7 @@ Status WinEnvIO::NewRandomRWFile(const std::string & fname,
   // Random access is to disable read-ahead as the system reads too much data
   DWORD desired_access = GENERIC_READ | GENERIC_WRITE;
   DWORD shared_mode = FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE;
-  DWORD creation_disposition = OPEN_ALWAYS; // Create if necessary or open existing
+  DWORD creation_disposition = OPEN_EXISTING; // Fail if file does not exist
   DWORD file_flags = FILE_FLAG_RANDOM_ACCESS;
 
   if (options.use_direct_reads && options.use_direct_writes) {
@@ -380,6 +389,84 @@ Status WinEnvIO::NewRandomRWFile(const std::string & fname,
   return s;
 }
 
+Status WinEnvIO::NewMemoryMappedFileBuffer(const std::string & fname,
+  std::unique_ptr<MemoryMappedFileBuffer>* result) {
+  Status s;
+  result->reset();
+
+  DWORD fileFlags = FILE_ATTRIBUTE_READONLY;
+
+  HANDLE hFile = INVALID_HANDLE_VALUE;
+  {
+    IOSTATS_TIMER_GUARD(open_nanos);
+    hFile = CreateFileA(
+      fname.c_str(), GENERIC_READ | GENERIC_WRITE,
+      FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
+      NULL,
+      OPEN_EXISTING,  // Open only if it exists
+      fileFlags,
+      NULL);
+  }
+
+  if (INVALID_HANDLE_VALUE == hFile) {
+    auto lastError = GetLastError();
+    s = IOErrorFromWindowsError("Failed to open NewMemoryMappedFileBuffer: " + fname,
+      lastError);
+    return s;
+  }
+  UniqueCloseHandlePtr fileGuard(hFile, CloseHandleFunc);
+
+  uint64_t fileSize = 0;
+  s = GetFileSize(fname, &fileSize);
+  if (!s.ok()) {
+    return s;
+  }
+  // Will not map empty files
+  if (fileSize == 0) {
+    return Status::NotSupported("NewMemoryMappedFileBuffer can not map zero length files: " + fname);
+  }
+
+  // size_t is 32-bit with 32-bit builds
+  if (fileSize > std::numeric_limits<size_t>::max()) {
+    return Status::NotSupported(
+      "The specified file size does not fit into 32-bit memory addressing: " + fname);
+  }
+
+  HANDLE hMap = CreateFileMappingA(hFile, NULL, PAGE_READWRITE,
+      0,  // Whole file at its present length
+      0,
+      NULL);  // Mapping name
+
+  if (!hMap) {
+    auto lastError = GetLastError();
+    return IOErrorFromWindowsError(
+      "Failed to create file mapping for NewMemoryMappedFileBuffer: " + fname,
+      lastError);
+  }
+  UniqueCloseHandlePtr mapGuard(hMap, CloseHandleFunc);
+
+  void* base = MapViewOfFileEx(hMap, FILE_MAP_WRITE,
+    0,  // High DWORD of access start
+    0,  // Low DWORD
+    fileSize,
+    NULL);  // Let the OS choose the mapping
+
+  if (!base) {
+    auto lastError = GetLastError();
+    return IOErrorFromWindowsError(
+      "Failed to MapViewOfFile for NewMemoryMappedFileBuffer: " + fname,
+      lastError);
+  }
+
+  result->reset(new WinMemoryMappedBuffer(hFile, hMap, 
+    base, static_cast<size_t>(fileSize)));
+
+  mapGuard.release();
+  fileGuard.release();
+
+  return s;
+}
+
 Status WinEnvIO::NewDirectory(const std::string& name,
   std::unique_ptr<Directory>* result) {
   Status s;
@@ -928,27 +1015,33 @@ std::string WinEnvIO::TimeToString(uint64_t secondsSince1970) {
 
 EnvOptions WinEnvIO::OptimizeForLogWrite(const EnvOptions& env_options,
   const DBOptions& db_options) const {
-  EnvOptions optimized = env_options;
+  EnvOptions optimized(env_options);
+  // These two the same as default optimizations
   optimized.bytes_per_sync = db_options.wal_bytes_per_sync;
-  optimized.use_mmap_writes = false;
-  // This is because we flush only whole pages on unbuffered io and
-  // the last records are not guaranteed to be flushed.
-  optimized.use_direct_writes = false;
-  // TODO(icanadi) it's faster if fallocate_with_keep_size is false, but it
-  // breaks TransactionLogIteratorStallAtLastRecord unit test. Fix the unit
-  // test and make this false
-  optimized.fallocate_with_keep_size = true;
   optimized.writable_file_max_buffer_size =
       db_options.writable_file_max_buffer_size;
+
+  // This adversely affects %999 on windows
+  optimized.use_mmap_writes = false;
+  // Direct writes will produce a huge perf impact on
+  // Windows. Pre-allocate space for WAL.
+  optimized.use_direct_writes = false;
   return optimized;
 }
 
 EnvOptions WinEnvIO::OptimizeForManifestWrite(
   const EnvOptions& env_options) const {
-  EnvOptions optimized = env_options;
+  EnvOptions optimized(env_options);
   optimized.use_mmap_writes = false;
-  optimized.use_direct_writes = false;
-  optimized.fallocate_with_keep_size = true;
+  optimized.use_direct_reads = false;
+  return optimized;
+}
+
+EnvOptions WinEnvIO::OptimizeForManifestRead(
+  const EnvOptions& env_options) const {
+  EnvOptions optimized(env_options);
+  optimized.use_mmap_writes = false;
+  optimized.use_direct_reads = false;
   return optimized;
 }
 
@@ -1145,6 +1238,10 @@ Status WinEnv::DeleteFile(const std::string& fname) {
   return winenv_io_.DeleteFile(fname);
 }
 
+Status WinEnv::Truncate(const std::string& fname, size_t size) {
+  return winenv_io_.Truncate(fname, size);
+}
+
 Status WinEnv::GetCurrentTime(int64_t* unix_time) {
   return winenv_io_.GetCurrentTime(unix_time);
 }
@@ -1173,10 +1270,15 @@ Status WinEnv::ReopenWritableFile(const std::string& fname,
 }
 
 Status WinEnv::NewRandomRWFile(const std::string & fname,
-  unique_ptr<RandomRWFile>* result, const EnvOptions & options) {
+  std::unique_ptr<RandomRWFile>* result, const EnvOptions & options) {
   return winenv_io_.NewRandomRWFile(fname, result, options);
 }
 
+Status WinEnv::NewMemoryMappedFileBuffer(const std::string& fname,
+  std::unique_ptr<MemoryMappedFileBuffer>* result) {
+  return winenv_io_.NewMemoryMappedFileBuffer(fname, result);
+}
+
 Status WinEnv::NewDirectory(const std::string& name,
   std::unique_ptr<Directory>* result) {
   return winenv_io_.NewDirectory(name, result);
@@ -1310,6 +1412,11 @@ void  WinEnv::IncBackgroundThreadsIfNeeded(int num, Env::Priority pri) {
   return winenv_threads_.IncBackgroundThreadsIfNeeded(num, pri);
 }
 
+EnvOptions WinEnv::OptimizeForManifestRead(
+  const EnvOptions& env_options) const {
+  return winenv_io_.OptimizeForManifestRead(env_options);
+}
+
 EnvOptions WinEnv::OptimizeForLogWrite(const EnvOptions& env_options,
   const DBOptions& db_options) const {
   return winenv_io_.OptimizeForLogWrite(env_options, db_options);

port/win/env_win.h

@@ -89,6 +89,8 @@ public:
 
   virtual Status DeleteFile(const std::string& fname);
 
+  Status Truncate(const std::string& fname, size_t size);
+
   virtual Status GetCurrentTime(int64_t* unix_time);
 
   virtual Status NewSequentialFile(const std::string& fname,
@@ -110,6 +112,10 @@ public:
     unique_ptr<RandomRWFile>* result,
     const EnvOptions& options);
 
+  virtual Status NewMemoryMappedFileBuffer(
+    const std::string& fname,
+    std::unique_ptr<MemoryMappedFileBuffer>* result);
+
   virtual Status NewDirectory(const std::string& name,
     std::unique_ptr<Directory>* result);
 
@@ -168,6 +174,9 @@ public:
   virtual EnvOptions OptimizeForManifestWrite(
     const EnvOptions& env_options) const;
 
+  virtual EnvOptions OptimizeForManifestRead(
+    const EnvOptions& env_options) const;
+
   size_t GetPageSize() const { return page_size_; }
 
   size_t GetAllocationGranularity() const { return allocation_granularity_; }
@@ -197,6 +206,8 @@ public:
 
   Status DeleteFile(const std::string& fname) override;
 
+  Status Truncate(const std::string& fname, size_t size) override;
+
   Status GetCurrentTime(int64_t* unix_time) override;
 
   Status NewSequentialFile(const std::string& fname,
@@ -224,9 +235,13 @@ public:
 
   // The returned file will only be accessed by one thread at a time.
   Status NewRandomRWFile(const std::string& fname,
-    unique_ptr<RandomRWFile>* result,
+    std::unique_ptr<RandomRWFile>* result,
     const EnvOptions& options) override;
 
+  Status NewMemoryMappedFileBuffer(
+    const std::string& fname,
+    std::unique_ptr<MemoryMappedFileBuffer>* result) override;
+
   Status NewDirectory(const std::string& name,
     std::unique_ptr<Directory>* result) override;
 
@@ -302,12 +317,16 @@ public:
 
   void IncBackgroundThreadsIfNeeded(int num, Env::Priority pri) override;
 
+  EnvOptions OptimizeForManifestRead(
+    const EnvOptions& env_options) const override;
+
   EnvOptions OptimizeForLogWrite(const EnvOptions& env_options,
     const DBOptions& db_options) const override;
 
   EnvOptions OptimizeForManifestWrite(
     const EnvOptions& env_options) const override;
 
+
 private:
 
   WinEnvIO      winenv_io_;

port/win/io_win.cc

@@ -1061,6 +1061,27 @@ Status WinRandomRWFile::Close() {
   return CloseImpl();
 }
 
+//////////////////////////////////////////////////////////////////////////
+/// WinMemoryMappedBufer
+WinMemoryMappedBuffer::~WinMemoryMappedBuffer() {
+  BOOL ret = FALSE;
+  if (base_ != nullptr) {
+    ret = ::UnmapViewOfFile(base_);
+    assert(ret);
+    base_ = nullptr;
+  }
+  if (map_handle_ != NULL && map_handle_ != INVALID_HANDLE_VALUE) {
+    ret = ::CloseHandle(map_handle_);
+    assert(ret);
+    map_handle_ = NULL;
+  }
+  if (file_handle_ != NULL && file_handle_ != INVALID_HANDLE_VALUE) {
+    ret = ::CloseHandle(file_handle_);
+    assert(ret);
+    file_handle_ = NULL;
+  }
+}
+
 //////////////////////////////////////////////////////////////////////////
 /// WinDirectory
 

port/win/io_win.h

@@ -411,6 +411,18 @@ class WinRandomRWFile : private WinFileData,
   virtual Status Close() override;
 };
 
+class WinMemoryMappedBuffer : public MemoryMappedFileBuffer {
+private:
+  HANDLE  file_handle_;
+  HANDLE  map_handle_;
+public:
+  WinMemoryMappedBuffer(HANDLE file_handle, HANDLE map_handle, void* base, size_t size) :
+    MemoryMappedFileBuffer(base, size),
+    file_handle_(file_handle),
+    map_handle_(map_handle) {}
+  ~WinMemoryMappedBuffer() override;
+};
+
 class WinDirectory : public Directory {
   HANDLE handle_;
  public:

src.mk

@@ -208,6 +208,8 @@ LIB_SOURCES =                                                   \
   utilities/transactions/transaction_util.cc                    \
   utilities/transactions/write_prepared_txn.cc                  \
   utilities/transactions/write_prepared_txn_db.cc               \
+  utilities/transactions/write_unprepared_txn.cc                \
+  utilities/transactions/write_unprepared_txn_db.cc             \
   utilities/ttl/db_ttl_impl.cc                                  \
   utilities/write_batch_with_index/write_batch_with_index.cc    \
   utilities/write_batch_with_index/write_batch_with_index_internal.cc    \

table/block.cc

@@ -87,7 +87,11 @@ void BlockIter::Prev() {
     const Slice current_key(key_ptr, current_prev_entry.key_size);
 
     current_ = current_prev_entry.offset;
-    key_.SetInternalKey(current_key, false /* copy */);
+    if (key_includes_seq_) {
+      key_.SetInternalKey(current_key, false /* copy */);
+    } else {
+      key_.SetUserKey(current_key, false /* copy */);
+    }
     value_ = current_prev_entry.value;
 
     return;
@@ -136,6 +140,10 @@ void BlockIter::Prev() {
 }
 
 void BlockIter::Seek(const Slice& target) {
+  Slice seek_key = target;
+  if (!key_includes_seq_) {
+    seek_key = ExtractUserKey(target);
+  }
   PERF_TIMER_GUARD(block_seek_nanos);
   if (data_ == nullptr) {  // Not init yet
     return;
@@ -145,7 +153,7 @@ void BlockIter::Seek(const Slice& target) {
   if (prefix_index_) {
     ok = PrefixSeek(target, &index);
   } else {
-    ok = BinarySeek(target, 0, num_restarts_ - 1, &index);
+    ok = BinarySeek(seek_key, 0, num_restarts_ - 1, &index);
   }
 
   if (!ok) {
@@ -155,7 +163,7 @@ void BlockIter::Seek(const Slice& target) {
   // Linear search (within restart block) for first key >= target
 
   while (true) {
-    if (!ParseNextKey() || Compare(key_.GetInternalKey(), target) >= 0) {
+    if (!ParseNextKey() || Compare(key_, seek_key) >= 0) {
       return;
     }
   }
@@ -163,24 +171,28 @@ void BlockIter::Seek(const Slice& target) {
 
 void BlockIter::SeekForPrev(const Slice& target) {
   PERF_TIMER_GUARD(block_seek_nanos);
+  Slice seek_key = target;
+  if (!key_includes_seq_) {
+    seek_key = ExtractUserKey(target);
+  }
   if (data_ == nullptr) {  // Not init yet
     return;
   }
   uint32_t index = 0;
-  bool ok = BinarySeek(target, 0, num_restarts_ - 1, &index);
+  bool ok = BinarySeek(seek_key, 0, num_restarts_ - 1, &index);
 
   if (!ok) {
     return;
   }
   SeekToRestartPoint(index);
-  // Linear search (within restart block) for first key >= target
+  // Linear search (within restart block) for first key >= seek_key
 
-  while (ParseNextKey() && Compare(key_.GetInternalKey(), target) < 0) {
+  while (ParseNextKey() && Compare(key_, seek_key) < 0) {
   }
   if (!Valid()) {
     SeekToLast();
   } else {
-    while (Valid() && Compare(key_.GetInternalKey(), target) > 0) {
+    while (Valid() && Compare(key_, seek_key) > 0) {
       Prev();
     }
   }
@@ -233,7 +245,11 @@ bool BlockIter::ParseNextKey() {
     if (shared == 0) {
       // If this key dont share any bytes with prev key then we dont need
       // to decode it and can use it's address in the block directly.
-      key_.SetInternalKey(Slice(p, non_shared), false /* copy */);
+      if (key_includes_seq_) {
+        key_.SetInternalKey(Slice(p, non_shared), false /* copy */);
+      } else {
+        key_.SetUserKey(Slice(p, non_shared), false /* copy */);
+      }
       key_pinned_ = true;
     } else {
       // This key share `shared` bytes with prev key, we need to decode it
@@ -380,6 +396,10 @@ bool BlockIter::BinaryBlockIndexSeek(const Slice& target, uint32_t* block_ids,
 
 bool BlockIter::PrefixSeek(const Slice& target, uint32_t* index) {
   assert(prefix_index_);
+  Slice seek_key = target;
+  if (!key_includes_seq_) {
+    seek_key = ExtractUserKey(target);
+  }
   uint32_t* block_ids = nullptr;
   uint32_t num_blocks = prefix_index_->GetBlocks(target, &block_ids);
 
@@ -387,7 +407,7 @@ bool BlockIter::PrefixSeek(const Slice& target, uint32_t* index) {
     current_ = restarts_;
     return false;
   } else  {
-    return BinaryBlockIndexSeek(target, block_ids, 0, num_blocks - 1, index);
+    return BinaryBlockIndexSeek(seek_key, block_ids, 0, num_blocks - 1, index);
   }
 }
 
@@ -422,8 +442,9 @@ Block::Block(BlockContents&& contents, SequenceNumber _global_seqno,
   }
 }
 
-BlockIter* Block::NewIterator(const Comparator* cmp, BlockIter* iter,
-                              bool total_order_seek, Statistics* stats) {
+BlockIter* Block::NewIterator(const Comparator* cmp, const Comparator* ucmp,
+                              BlockIter* iter, bool total_order_seek,
+                              Statistics* stats, bool key_includes_seq) {
   BlockIter* ret_iter;
   if (iter != nullptr) {
     ret_iter = iter;
@@ -441,9 +462,9 @@ BlockIter* Block::NewIterator(const Comparator* cmp, BlockIter* iter,
   } else {
     BlockPrefixIndex* prefix_index_ptr =
         total_order_seek ? nullptr : prefix_index_.get();
-    ret_iter->Initialize(cmp, data_, restart_offset_, num_restarts_,
+    ret_iter->Initialize(cmp, ucmp, data_, restart_offset_, num_restarts_,
                          prefix_index_ptr, global_seqno_,
-                         read_amp_bitmap_.get());
+                         read_amp_bitmap_.get(), key_includes_seq);
 
     if (read_amp_bitmap_) {
       if (read_amp_bitmap_->GetStatistics() != stats) {

table/block.h

@@ -162,6 +162,9 @@ class Block {
   // the iterator will simply be set as "invalid", rather than returning
   // the key that is just pass the target key.
   //
+  // If comparator is InternalKeyComparator, user_comparator is its user
+  // comparator; they are equal otherwise.
+  //
   // If iter is null, return new Iterator
   // If iter is not null, update this one and return it as Iterator*
   //
@@ -169,9 +172,11 @@ class Block {
   // This option only applies for index block. For data block, hash_index_
   // and prefix_index_ are null, so this option does not matter.
   BlockIter* NewIterator(const Comparator* comparator,
+                         const Comparator* user_comparator,
                          BlockIter* iter = nullptr,
                          bool total_order_seek = true,
-                         Statistics* stats = nullptr);
+                         Statistics* stats = nullptr,
+                         bool key_includes_seq = true);
   void SetBlockPrefixIndex(BlockPrefixIndex* prefix_index);
 
   // Report an approximation of how much memory has been used.
@@ -203,6 +208,7 @@ class BlockIter final : public InternalIterator {
   // and status() is OK.
   BlockIter()
       : comparator_(nullptr),
+        user_comparator_(nullptr),
         data_(nullptr),
         restarts_(0),
         num_restarts_(0),
@@ -211,26 +217,30 @@ class BlockIter final : public InternalIterator {
         status_(Status::OK()),
         prefix_index_(nullptr),
         key_pinned_(false),
+        key_includes_seq_(true),
         global_seqno_(kDisableGlobalSequenceNumber),
         read_amp_bitmap_(nullptr),
         last_bitmap_offset_(0) {}
 
-  BlockIter(const Comparator* comparator, const char* data, uint32_t restarts,
-            uint32_t num_restarts, BlockPrefixIndex* prefix_index,
-            SequenceNumber global_seqno, BlockReadAmpBitmap* read_amp_bitmap)
+  BlockIter(const Comparator* comparator, const Comparator* user_comparator,
+            const char* data, uint32_t restarts, uint32_t num_restarts,
+            BlockPrefixIndex* prefix_index, SequenceNumber global_seqno,
+            BlockReadAmpBitmap* read_amp_bitmap, bool key_includes_seq)
       : BlockIter() {
-    Initialize(comparator, data, restarts, num_restarts, prefix_index,
-               global_seqno, read_amp_bitmap);
+    Initialize(comparator, user_comparator, data, restarts, num_restarts,
+               prefix_index, global_seqno, read_amp_bitmap, key_includes_seq);
   }
 
-  void Initialize(const Comparator* comparator, const char* data,
+  void Initialize(const Comparator* comparator,
+                  const Comparator* user_comparator, const char* data,
                   uint32_t restarts, uint32_t num_restarts,
                   BlockPrefixIndex* prefix_index, SequenceNumber global_seqno,
-                  BlockReadAmpBitmap* read_amp_bitmap) {
+                  BlockReadAmpBitmap* read_amp_bitmap, bool key_includes_seq) {
     assert(data_ == nullptr);           // Ensure it is called only once
     assert(num_restarts > 0);           // Ensure the param is valid
 
     comparator_ = comparator;
+    user_comparator_ = user_comparator;
     data_ = data;
     restarts_ = restarts;
     num_restarts_ = num_restarts;
@@ -240,6 +250,7 @@ class BlockIter final : public InternalIterator {
     global_seqno_ = global_seqno;
     read_amp_bitmap_ = read_amp_bitmap;
     last_bitmap_offset_ = current_ + 1;
+    key_includes_seq_ = key_includes_seq;
   }
 
   // Makes Valid() return false, status() return `s`, and Seek()/Prev()/etc do
@@ -263,7 +274,7 @@ class BlockIter final : public InternalIterator {
   virtual Status status() const override { return status_; }
   virtual Slice key() const override {
     assert(Valid());
-    return key_.GetInternalKey();
+    return key_includes_seq_ ? key_.GetInternalKey() : key_.GetUserKey();
   }
   virtual Slice value() const override {
     assert(Valid());
@@ -312,7 +323,11 @@ class BlockIter final : public InternalIterator {
   }
 
  private:
+  // Note: The type could be changed to InternalKeyComparator but we see a weird
+  // performance drop by that.
   const Comparator* comparator_;
+  // Same as comparator_ if comparator_ is not InernalKeyComparator
+  const Comparator* user_comparator_;
   const char* data_;       // underlying block contents
   uint32_t restarts_;      // Offset of restart array (list of fixed32)
   uint32_t num_restarts_;  // Number of uint32_t entries in restart array
@@ -325,8 +340,11 @@ class BlockIter final : public InternalIterator {
   Status status_;
   BlockPrefixIndex* prefix_index_;
   bool key_pinned_;
+  // Key is in InternalKey format
+  bool key_includes_seq_;
   SequenceNumber global_seqno_;
 
+ public:
   // read-amp bitmap
   BlockReadAmpBitmap* read_amp_bitmap_;
   // last `current_` value we report to read-amp bitmp
@@ -357,7 +375,19 @@ class BlockIter final : public InternalIterator {
   int32_t prev_entries_idx_ = -1;
 
   inline int Compare(const Slice& a, const Slice& b) const {
-    return comparator_->Compare(a, b);
+    if (key_includes_seq_) {
+      return comparator_->Compare(a, b);
+    } else {
+      return user_comparator_->Compare(a, b);
+    }
+  }
+
+  inline int Compare(const IterKey& ikey, const Slice& b) const {
+    if (key_includes_seq_) {
+      return comparator_->Compare(ikey.GetInternalKey(), b);
+    } else {
+      return user_comparator_->Compare(ikey.GetUserKey(), b);
+    }
   }
 
   // Return the offset in data_ just past the end of the current entry.

table/block_based_table_builder.cc

@@ -763,6 +763,8 @@ Status BlockBasedTableBuilder::Finish() {
         r->props.top_level_index_size =
             r->p_index_builder_->EstimateTopLevelIndexSize(r->offset);
       }
+      r->props.index_key_is_user_key =
+          !r->index_builder->seperator_is_key_plus_seq();
       r->props.creation_time = r->creation_time;
       r->props.oldest_key_time = r->oldest_key_time;
 

table/block_based_table_reader.cc

@@ -212,7 +212,7 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
                        const InternalKeyComparator* icomparator,
                        IndexReader** index_reader,
                        const PersistentCacheOptions& cache_options,
-                       const int level) {
+                       const int level, const bool index_key_includes_seq) {
     std::unique_ptr<Block> index_block;
     auto s = ReadBlockFromFile(
         file, prefetch_buffer, footer, ReadOptions(), index_handle,
@@ -221,9 +221,9 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
         kDisableGlobalSequenceNumber, 0 /* read_amp_bytes_per_bit */);
 
     if (s.ok()) {
-      *index_reader =
-          new PartitionIndexReader(table, icomparator, std::move(index_block),
-                                   ioptions.statistics, level);
+      *index_reader = new PartitionIndexReader(
+          table, icomparator, std::move(index_block), ioptions.statistics,
+          level, index_key_includes_seq);
     }
 
     return s;
@@ -237,15 +237,19 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
     if (!partition_map_.empty()) {
       return NewTwoLevelIterator(
           new BlockBasedTable::PartitionedIndexIteratorState(
-              table_, partition_map_.size() ? &partition_map_ : nullptr),
-          index_block_->NewIterator(icomparator_, nullptr, true));
+              table_, &partition_map_, index_key_includes_seq_),
+          index_block_->NewIterator(
+              icomparator_, icomparator_->user_comparator(), nullptr, true));
     } else {
       auto ro = ReadOptions();
       ro.fill_cache = fill_cache;
+      bool kIsIndex = true;
       return new BlockBasedTableIterator(
           table_, ro, *icomparator_,
-          index_block_->NewIterator(icomparator_, nullptr, true), false,
-          /* prefix_extractor */ nullptr);
+          index_block_->NewIterator(
+              icomparator_, icomparator_->user_comparator(), nullptr, true),
+          false,
+          /* prefix_extractor */ nullptr, kIsIndex, index_key_includes_seq_);
     }
     // TODO(myabandeh): Update TwoLevelIterator to be able to make use of
     // on-stack BlockIter while the state is on heap. Currentlly it assumes
@@ -258,7 +262,8 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
     auto rep = table_->rep_;
     BlockIter biter;
     BlockHandle handle;
-    index_block_->NewIterator(icomparator_, &biter, true);
+    index_block_->NewIterator(icomparator_, icomparator_->user_comparator(),
+                              &biter, true);
     // Index partitions are assumed to be consecuitive. Prefetch them all.
     // Read the first block offset
     biter.SeekToFirst();
@@ -347,16 +352,18 @@ class PartitionIndexReader : public IndexReader, public Cleanable {
   PartitionIndexReader(BlockBasedTable* table,
                        const InternalKeyComparator* icomparator,
                        std::unique_ptr<Block>&& index_block, Statistics* stats,
-                       const int /*level*/)
+                       const int /*level*/, const bool index_key_includes_seq)
       : IndexReader(icomparator, stats),
         table_(table),
-        index_block_(std::move(index_block)) {
+        index_block_(std::move(index_block)),
+        index_key_includes_seq_(index_key_includes_seq) {
     assert(index_block_ != nullptr);
   }
   BlockBasedTable* table_;
   std::unique_ptr<Block> index_block_;
   std::unordered_map<uint64_t, BlockBasedTable::CachableEntry<Block>>
       partition_map_;
+  const bool index_key_includes_seq_;
 };
 
 // Index that allows binary search lookup for the first key of each block.
@@ -374,7 +381,8 @@ class BinarySearchIndexReader : public IndexReader {
                        const ImmutableCFOptions& ioptions,
                        const InternalKeyComparator* icomparator,
                        IndexReader** index_reader,
-                       const PersistentCacheOptions& cache_options) {
+                       const PersistentCacheOptions& cache_options,
+                       const bool index_key_includes_seq) {
     std::unique_ptr<Block> index_block;
     auto s = ReadBlockFromFile(
         file, prefetch_buffer, footer, ReadOptions(), index_handle,
@@ -384,7 +392,8 @@ class BinarySearchIndexReader : public IndexReader {
 
     if (s.ok()) {
       *index_reader = new BinarySearchIndexReader(
-          icomparator, std::move(index_block), ioptions.statistics);
+          icomparator, std::move(index_block), ioptions.statistics,
+          index_key_includes_seq);
     }
 
     return s;
@@ -393,7 +402,9 @@ class BinarySearchIndexReader : public IndexReader {
   virtual InternalIterator* NewIterator(BlockIter* iter = nullptr,
                                         bool /*dont_care*/ = true,
                                         bool /*dont_care*/ = true) override {
-    return index_block_->NewIterator(icomparator_, iter, true);
+    return index_block_->NewIterator(icomparator_,
+                                     icomparator_->user_comparator(), iter,
+                                     true, nullptr, index_key_includes_seq_);
   }
 
   virtual size_t size() const override { return index_block_->size(); }
@@ -409,11 +420,14 @@ class BinarySearchIndexReader : public IndexReader {
  private:
   BinarySearchIndexReader(const InternalKeyComparator* icomparator,
                           std::unique_ptr<Block>&& index_block,
-                          Statistics* stats)
-      : IndexReader(icomparator, stats), index_block_(std::move(index_block)) {
+                          Statistics* stats, const bool index_key_includes_seq)
+      : IndexReader(icomparator, stats),
+        index_block_(std::move(index_block)),
+        index_key_includes_seq_(index_key_includes_seq) {
     assert(index_block_ != nullptr);
   }
   std::unique_ptr<Block> index_block_;
+  const bool index_key_includes_seq_;
 };
 
 // Index that leverages an internal hash table to quicken the lookup for a given
@@ -429,7 +443,8 @@ class HashIndexReader : public IndexReader {
                        InternalIterator* meta_index_iter,
                        IndexReader** index_reader,
                        bool /*hash_index_allow_collision*/,
-                       const PersistentCacheOptions& cache_options) {
+                       const PersistentCacheOptions& cache_options,
+                       const bool index_key_includes_seq) {
     std::unique_ptr<Block> index_block;
     auto s = ReadBlockFromFile(
         file, prefetch_buffer, footer, ReadOptions(), index_handle,
@@ -447,7 +462,7 @@ class HashIndexReader : public IndexReader {
 
     auto new_index_reader =
         new HashIndexReader(icomparator, std::move(index_block),
-          ioptions.statistics);
+                            ioptions.statistics, index_key_includes_seq);
     *index_reader = new_index_reader;
 
     // Get prefixes block
@@ -484,7 +499,7 @@ class HashIndexReader : public IndexReader {
         file, prefetch_buffer, footer, ReadOptions(), prefixes_meta_handle,
         &prefixes_meta_contents, ioptions, true /* decompress */,
         dummy_comp_dict /*compression dict*/, cache_options);
-    prefixes_meta_block_fetcher.ReadBlockContents();
+    s = prefixes_meta_block_fetcher.ReadBlockContents();
     if (!s.ok()) {
       // TODO: log error
       return Status::OK();
@@ -504,7 +519,9 @@ class HashIndexReader : public IndexReader {
   virtual InternalIterator* NewIterator(BlockIter* iter = nullptr,
                                         bool total_order_seek = true,
                                         bool /*dont_care*/ = true) override {
-    return index_block_->NewIterator(icomparator_, iter, total_order_seek);
+    return index_block_->NewIterator(
+        icomparator_, icomparator_->user_comparator(), iter, total_order_seek,
+        nullptr, index_key_includes_seq_);
   }
 
   virtual size_t size() const override { return index_block_->size(); }
@@ -520,8 +537,11 @@ class HashIndexReader : public IndexReader {
 
  private:
   HashIndexReader(const InternalKeyComparator* icomparator,
-                  std::unique_ptr<Block>&& index_block, Statistics* stats)
-      : IndexReader(icomparator, stats), index_block_(std::move(index_block)) {
+                  std::unique_ptr<Block>&& index_block, Statistics* stats,
+                  const bool index_key_includes_seq)
+      : IndexReader(icomparator, stats),
+        index_block_(std::move(index_block)),
+        index_key_includes_seq_(index_key_includes_seq) {
     assert(index_block_ != nullptr);
   }
 
@@ -530,6 +550,7 @@ class HashIndexReader : public IndexReader {
 
   std::unique_ptr<Block> index_block_;
   BlockContents prefixes_contents_;
+  const bool index_key_includes_seq_;
 };
 
 // Helper function to setup the cache key's prefix for the Table.
@@ -1026,7 +1047,8 @@ Status BlockBasedTable::ReadMetaBlock(Rep* rep,
 
   *meta_block = std::move(meta);
   // meta block uses bytewise comparator.
-  iter->reset(meta_block->get()->NewIterator(BytewiseComparator()));
+  iter->reset(meta_block->get()->NewIterator(BytewiseComparator(),
+                                             BytewiseComparator()));
   return Status::OK();
 }
 
@@ -1502,14 +1524,15 @@ InternalIterator* BlockBasedTable::NewIndexIterator(
 
 BlockIter* BlockBasedTable::NewDataBlockIterator(
     Rep* rep, const ReadOptions& ro, const Slice& index_value,
-    BlockIter* input_iter, bool is_index, GetContext* get_context) {
+    BlockIter* input_iter, bool is_index, bool key_includes_seq,
+    GetContext* get_context) {
   BlockHandle handle;
   Slice input = index_value;
   // We intentionally allow extra stuff in index_value so that we
   // can add more features in the future.
   Status s = handle.DecodeFrom(&input);
   return NewDataBlockIterator(rep, ro, handle, input_iter, is_index,
-                              get_context, s);
+                              key_includes_seq, get_context, s);
 }
 
 // Convert an index iterator value (i.e., an encoded BlockHandle)
@@ -1518,7 +1541,8 @@ BlockIter* BlockBasedTable::NewDataBlockIterator(
 // If input_iter is not null, update this iter and return it
 BlockIter* BlockBasedTable::NewDataBlockIterator(
     Rep* rep, const ReadOptions& ro, const BlockHandle& handle,
-    BlockIter* input_iter, bool is_index, GetContext* get_context, Status s) {
+    BlockIter* input_iter, bool is_index, bool key_includes_seq,
+    GetContext* get_context, Status s) {
   PERF_TIMER_GUARD(new_table_block_iter_nanos);
 
   const bool no_io = (ro.read_tier == kBlockCacheTier);
@@ -1564,8 +1588,9 @@ BlockIter* BlockBasedTable::NewDataBlockIterator(
 
   if (s.ok()) {
     assert(block.value != nullptr);
-    iter = block.value->NewIterator(&rep->internal_comparator, iter, true,
-                                    rep->ioptions.statistics);
+    iter = block.value->NewIterator(
+        &rep->internal_comparator, rep->internal_comparator.user_comparator(),
+        iter, true, rep->ioptions.statistics, key_includes_seq);
     if (block.cache_handle != nullptr) {
       iter->RegisterCleanup(&ReleaseCachedEntry, block_cache,
                             block.cache_handle);
@@ -1677,8 +1702,11 @@ Status BlockBasedTable::MaybeLoadDataBlockToCache(
 
 BlockBasedTable::PartitionedIndexIteratorState::PartitionedIndexIteratorState(
     BlockBasedTable* table,
-    std::unordered_map<uint64_t, CachableEntry<Block>>* block_map)
-    : table_(table), block_map_(block_map) {}
+    std::unordered_map<uint64_t, CachableEntry<Block>>* block_map,
+    bool index_key_includes_seq)
+    : table_(table),
+      block_map_(block_map),
+      index_key_includes_seq_(index_key_includes_seq) {}
 
 const size_t BlockBasedTableIterator::kMaxReadaheadSize = 256 * 1024;
 
@@ -1701,8 +1729,9 @@ BlockBasedTable::PartitionedIndexIteratorState::NewSecondaryIterator(
     assert(block_cache);
     RecordTick(rep->ioptions.statistics, BLOCK_CACHE_BYTES_READ,
                block_cache->GetUsage(block->second.cache_handle));
-    return block->second.value->NewIterator(&rep->internal_comparator, nullptr,
-                                            true, rep->ioptions.statistics);
+    return block->second.value->NewIterator(
+        &rep->internal_comparator, rep->internal_comparator.user_comparator(),
+        nullptr, true, rep->ioptions.statistics, index_key_includes_seq_);
   }
   // Create an empty iterator
   return new BlockIter();
@@ -1770,7 +1799,10 @@ bool BlockBasedTable::PrefixMayMatch(const Slice& internal_key,
         // and we're not really sure that we're past the end
         // of the file
         may_match = iiter->status().IsIncomplete();
-      } else if (ExtractUserKey(iiter->key())
+      } else if ((rep_->table_properties &&
+                          rep_->table_properties->index_key_is_user_key
+                      ? iiter->key()
+                      : ExtractUserKey(iiter->key()))
                      .starts_with(ExtractUserKey(internal_prefix))) {
         // we need to check for this subtle case because our only
         // guarantee is that "the key is a string >= last key in that data
@@ -1836,7 +1868,11 @@ void BlockBasedTableIterator::Seek(const Slice& target) {
 
   FindKeyForward();
   assert(!data_block_iter_.Valid() ||
-         icomp_.Compare(target, data_block_iter_.key()) <= 0);
+         (key_includes_seq_ &&
+          icomp_.Compare(target, data_block_iter_.key()) <= 0) ||
+         (!key_includes_seq_ &&
+          icomp_.user_comparator()->Compare(ExtractUserKey(target),
+                                            data_block_iter_.key()) <= 0));
 }
 
 void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
@@ -1952,7 +1988,8 @@ void BlockBasedTableIterator::InitDataBlock() {
     }
 
     BlockBasedTable::NewDataBlockIterator(rep, read_options_, data_block_handle,
-                                          &data_block_iter_, false,
+                                          &data_block_iter_, is_index_,
+                                          key_includes_seq_,
                                           /* get_context */ nullptr, s);
     block_iter_points_to_real_block_ = true;
   }
@@ -2024,24 +2061,25 @@ InternalIterator* BlockBasedTable::NewIterator(
     Arena* arena, bool skip_filters) {
   bool prefix_extractor_changed =
       PrefixExtractorChanged(rep_->table_properties, prefix_extractor);
+  const bool kIsNotIndex = false;
   if (arena == nullptr) {
     return new BlockBasedTableIterator(
         this, read_options, rep_->internal_comparator,
         NewIndexIterator(
             read_options,
             prefix_extractor_changed &&
-            rep_->index_type == BlockBasedTableOptions::kHashSearch),
+                rep_->index_type == BlockBasedTableOptions::kHashSearch),
         !skip_filters && !read_options.total_order_seek &&
-        prefix_extractor != nullptr && !prefix_extractor_changed,
-        prefix_extractor);
+            prefix_extractor != nullptr && !prefix_extractor_changed,
+        prefix_extractor, kIsNotIndex);
   } else {
     auto* mem = arena->AllocateAligned(sizeof(BlockBasedTableIterator));
     return new (mem) BlockBasedTableIterator(
         this, read_options, rep_->internal_comparator,
         NewIndexIterator(read_options, prefix_extractor_changed),
         !skip_filters && !read_options.total_order_seek &&
-        prefix_extractor != nullptr && !prefix_extractor_changed,
-        prefix_extractor);
+            prefix_extractor != nullptr && !prefix_extractor_changed,
+        prefix_extractor, kIsNotIndex);
   }
 }
 
@@ -2061,7 +2099,8 @@ InternalIterator* BlockBasedTable::NewRangeTombstoneIterator(
     assert(block_cache != nullptr);
     if (block_cache->Ref(rep_->range_del_entry.cache_handle)) {
       auto iter = rep_->range_del_entry.value->NewIterator(
-          &rep_->internal_comparator, nullptr /* iter */,
+          &rep_->internal_comparator,
+          rep_->internal_comparator.user_comparator(), nullptr /* iter */,
           true /* total_order_seek */, rep_->ioptions.statistics);
       iter->RegisterCleanup(&ReleaseCachedEntry, block_cache,
                             rep_->range_del_entry.cache_handle);
@@ -2107,6 +2146,7 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
                             GetContext* get_context,
                             const SliceTransform* prefix_extractor,
                             bool skip_filters) {
+  assert(key.size() >= 8);  // key must be internal key
   Status s;
   const bool no_io = read_options.read_tier == kBlockCacheTier;
   CachableEntry<FilterBlockReader> filter_entry;
@@ -2215,6 +2255,7 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
 Status BlockBasedTable::Prefetch(const Slice* const begin,
                                  const Slice* const end) {
   auto& comparator = rep_->internal_comparator;
+  auto user_comparator = comparator.user_comparator();
   // pre-condition
   if (begin && end && comparator.Compare(*begin, *end) > 0) {
     return Status::InvalidArgument(*begin, *end);
@@ -2238,8 +2279,12 @@ Status BlockBasedTable::Prefetch(const Slice* const begin,
   for (begin ? iiter->Seek(*begin) : iiter->SeekToFirst(); iiter->Valid();
        iiter->Next()) {
     Slice block_handle = iiter->value();
-
-    if (end && comparator.Compare(iiter->key(), *end) >= 0) {
+    const bool is_user_key = rep_->table_properties &&
+                             rep_->table_properties->index_key_is_user_key > 0;
+    if (end &&
+        ((!is_user_key && comparator.Compare(iiter->key(), *end) >= 0) ||
+         (is_user_key &&
+          user_comparator->Compare(iiter->key(), ExtractUserKey(*end)) >= 0))) {
       if (prefetching_boundary_page) {
         break;
       }
@@ -2392,12 +2437,16 @@ Status BlockBasedTable::CreateIndexReader(
       return PartitionIndexReader::Create(
           this, file, prefetch_buffer, footer, footer.index_handle(),
           rep_->ioptions, icomparator, index_reader,
-          rep_->persistent_cache_options, level);
+          rep_->persistent_cache_options, level,
+          rep_->table_properties == nullptr ||
+              rep_->table_properties->index_key_is_user_key == 0);
     }
     case BlockBasedTableOptions::kBinarySearch: {
       return BinarySearchIndexReader::Create(
           file, prefetch_buffer, footer, footer.index_handle(), rep_->ioptions,
-          icomparator, index_reader, rep_->persistent_cache_options);
+          icomparator, index_reader, rep_->persistent_cache_options,
+          rep_->table_properties == nullptr ||
+              rep_->table_properties->index_key_is_user_key == 0);
     }
     case BlockBasedTableOptions::kHashSearch: {
       std::unique_ptr<Block> meta_guard;
@@ -2415,7 +2464,9 @@ Status BlockBasedTable::CreateIndexReader(
           return BinarySearchIndexReader::Create(
               file, prefetch_buffer, footer, footer.index_handle(),
               rep_->ioptions, icomparator, index_reader,
-              rep_->persistent_cache_options);
+              rep_->persistent_cache_options,
+              rep_->table_properties == nullptr ||
+                  rep_->table_properties->index_key_is_user_key == 0);
         }
         meta_index_iter = meta_iter_guard.get();
       }
@@ -2424,7 +2475,9 @@ Status BlockBasedTable::CreateIndexReader(
           rep_->internal_prefix_transform.get(), footer, file, prefetch_buffer,
           rep_->ioptions, icomparator, footer.index_handle(), meta_index_iter,
           index_reader, rep_->hash_index_allow_collision,
-          rep_->persistent_cache_options);
+          rep_->persistent_cache_options,
+          rep_->table_properties == nullptr ||
+              rep_->table_properties->index_key_is_user_key == 0);
     }
     default: {
       std::string error_message =
@@ -2709,16 +2762,23 @@ Status BlockBasedTable::DumpIndexBlock(WritableFile* out_file) {
       break;
     }
     Slice key = blockhandles_iter->key();
+    Slice user_key;
     InternalKey ikey;
-    ikey.DecodeFrom(key);
+    if (rep_->table_properties &&
+        rep_->table_properties->index_key_is_user_key != 0) {
+      user_key = key;
+    } else {
+      ikey.DecodeFrom(key);
+      user_key = ikey.user_key();
+    }
 
     out_file->Append("  HEX    ");
-    out_file->Append(ikey.user_key().ToString(true).c_str());
+    out_file->Append(user_key.ToString(true).c_str());
     out_file->Append(": ");
     out_file->Append(blockhandles_iter->value().ToString(true).c_str());
     out_file->Append("\n");
 
-    std::string str_key = ikey.user_key().ToString();
+    std::string str_key = user_key.ToString();
     std::string res_key("");
     char cspace = ' ';
     for (size_t i = 0; i < str_key.size(); i++) {

table/block_based_table_reader.h

@@ -217,11 +217,13 @@ class BlockBasedTable : public TableReader {
                                          const Slice& index_value,
                                          BlockIter* input_iter = nullptr,
                                          bool is_index = false,
+                                         bool key_includes_seq = true,
                                          GetContext* get_context = nullptr);
   static BlockIter* NewDataBlockIterator(Rep* rep, const ReadOptions& ro,
                                          const BlockHandle& block_hanlde,
                                          BlockIter* input_iter = nullptr,
                                          bool is_index = false,
+                                         bool key_includes_seq = true,
                                          GetContext* get_context = nullptr,
                                          Status s = Status());
 
@@ -378,13 +380,15 @@ class BlockBasedTable::PartitionedIndexIteratorState
  public:
   PartitionedIndexIteratorState(
       BlockBasedTable* table,
-      std::unordered_map<uint64_t, CachableEntry<Block>>* block_map = nullptr);
+      std::unordered_map<uint64_t, CachableEntry<Block>>* block_map,
+      const bool index_key_includes_seq);
   InternalIterator* NewSecondaryIterator(const Slice& index_value) override;
 
  private:
   // Don't own table_
   BlockBasedTable* table_;
   std::unordered_map<uint64_t, CachableEntry<Block>>* block_map_;
+  bool index_key_includes_seq_;
 };
 
 // CachableEntry represents the entries that *may* be fetched from block cache.
@@ -429,7 +433,7 @@ struct BlockBasedTable::Rep {
 
   const ImmutableCFOptions& ioptions;
   const EnvOptions& env_options;
-  const BlockBasedTableOptions& table_options;
+  const BlockBasedTableOptions table_options;
   const FilterPolicy* const filter_policy;
   const InternalKeyComparator& internal_comparator;
   Status status;
@@ -509,7 +513,8 @@ class BlockBasedTableIterator : public InternalIterator {
                           const ReadOptions& read_options,
                           const InternalKeyComparator& icomp,
                           InternalIterator* index_iter, bool check_filter,
-                          const SliceTransform* prefix_extractor)
+                          const SliceTransform* prefix_extractor, bool is_index,
+                          bool key_includes_seq = true)
       : table_(table),
         read_options_(read_options),
         icomp_(icomp),
@@ -517,6 +522,8 @@ class BlockBasedTableIterator : public InternalIterator {
         pinned_iters_mgr_(nullptr),
         block_iter_points_to_real_block_(false),
         check_filter_(check_filter),
+        is_index_(is_index),
+        key_includes_seq_(key_includes_seq),
         prefix_extractor_(prefix_extractor) {}
 
   ~BlockBasedTableIterator() { delete index_iter_; }
@@ -609,6 +616,10 @@ class BlockBasedTableIterator : public InternalIterator {
   bool block_iter_points_to_real_block_;
   bool is_out_of_bound_ = false;
   bool check_filter_;
+  // If the blocks over which we iterate are index blocks
+  bool is_index_;
+  // If the keys in the blocks over which we iterate include 8 byte sequence
+  bool key_includes_seq_;
   // TODO use block offset instead
   std::string prev_index_value_;
   const SliceTransform* prefix_extractor_;

table/block_test.cc

@@ -99,7 +99,8 @@ TEST_F(BlockTest, SimpleTest) {
 
   // read contents of block sequentially
   int count = 0;
-  InternalIterator *iter = reader.NewIterator(options.comparator);
+  InternalIterator *iter =
+      reader.NewIterator(options.comparator, options.comparator);
   for (iter->SeekToFirst();iter->Valid(); count++, iter->Next()) {
 
     // read kv from block
@@ -113,7 +114,7 @@ TEST_F(BlockTest, SimpleTest) {
   delete iter;
 
   // read block contents randomly
-  iter = reader.NewIterator(options.comparator);
+  iter = reader.NewIterator(options.comparator, options.comparator);
   for (int i = 0; i < num_records; i++) {
 
     // find a random key in the lookaside array
@@ -163,7 +164,7 @@ void CheckBlockContents(BlockContents contents, const int max_key,
       NewFixedPrefixTransform(prefix_size));
 
   std::unique_ptr<InternalIterator> regular_iter(
-      reader2.NewIterator(BytewiseComparator()));
+      reader2.NewIterator(BytewiseComparator(), BytewiseComparator()));
 
   // Seek existent keys
   for (size_t i = 0; i < keys.size(); i++) {
@@ -388,8 +389,8 @@ TEST_F(BlockTest, BlockWithReadAmpBitmap) {
 
     // read contents of block sequentially
     size_t read_bytes = 0;
-    BlockIter *iter = static_cast<BlockIter *>(
-        reader.NewIterator(options.comparator, nullptr, true, stats.get()));
+    BlockIter *iter = static_cast<BlockIter *>(reader.NewIterator(
+        options.comparator, options.comparator, nullptr, true, stats.get()));
     for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
       iter->value();
       read_bytes += iter->TEST_CurrentEntrySize();
@@ -421,8 +422,8 @@ TEST_F(BlockTest, BlockWithReadAmpBitmap) {
                  kBytesPerBit, stats.get());
 
     size_t read_bytes = 0;
-    BlockIter *iter = static_cast<BlockIter *>(
-        reader.NewIterator(options.comparator, nullptr, true, stats.get()));
+    BlockIter *iter = static_cast<BlockIter *>(reader.NewIterator(
+        options.comparator, options.comparator, nullptr, true, stats.get()));
     for (int i = 0; i < num_records; i++) {
       Slice k(keys[i]);
 
@@ -457,8 +458,8 @@ TEST_F(BlockTest, BlockWithReadAmpBitmap) {
                  kBytesPerBit, stats.get());
 
     size_t read_bytes = 0;
-    BlockIter *iter = static_cast<BlockIter *>(
-        reader.NewIterator(options.comparator, nullptr, true, stats.get()));
+    BlockIter *iter = static_cast<BlockIter *>(reader.NewIterator(
+        options.comparator, options.comparator, nullptr, true, stats.get()));
     std::unordered_set<int> read_keys;
     for (int i = 0; i < num_records; i++) {
       int index = rnd.Uniform(num_records);

table/index_builder.cc

@@ -31,13 +31,15 @@ IndexBuilder* IndexBuilder::CreateIndexBuilder(
   IndexBuilder* result = nullptr;
   switch (index_type) {
     case BlockBasedTableOptions::kBinarySearch: {
-      result =  new ShortenedIndexBuilder(comparator,
-                                       table_opt.index_block_restart_interval);
+      result = new ShortenedIndexBuilder(comparator,
+                                         table_opt.index_block_restart_interval,
+                                         table_opt.format_version);
     }
   break;
     case BlockBasedTableOptions::kHashSearch: {
       result = new HashIndexBuilder(comparator, int_key_slice_transform,
-                                  table_opt.index_block_restart_interval);
+                                    table_opt.index_block_restart_interval,
+                                    table_opt.format_version);
     }
   break;
     case BlockBasedTableOptions::kTwoLevelIndexSearch: {
@@ -62,9 +64,11 @@ PartitionedIndexBuilder::PartitionedIndexBuilder(
     const InternalKeyComparator* comparator,
     const BlockBasedTableOptions& table_opt)
     : IndexBuilder(comparator),
-      index_block_builder_(table_opt.index_block_restart_interval),
+      index_block_builder_(table_opt.index_block_restart_interval,
+                           table_opt.format_version),
       sub_index_builder_(nullptr),
-      table_opt_(table_opt) {}
+      table_opt_(table_opt),
+      seperator_is_key_plus_seq_(false) {}
 
 PartitionedIndexBuilder::~PartitionedIndexBuilder() {
   delete sub_index_builder_;
@@ -73,7 +77,8 @@ PartitionedIndexBuilder::~PartitionedIndexBuilder() {
 void PartitionedIndexBuilder::MakeNewSubIndexBuilder() {
   assert(sub_index_builder_ == nullptr);
   sub_index_builder_ = new ShortenedIndexBuilder(
-      comparator_, table_opt_.index_block_restart_interval);
+      comparator_, table_opt_.index_block_restart_interval,
+      table_opt_.format_version);
   flush_policy_.reset(FlushBlockBySizePolicyFactory::NewFlushBlockPolicy(
       table_opt_.metadata_block_size, table_opt_.block_size_deviation,
       sub_index_builder_->index_block_builder_));
@@ -95,6 +100,10 @@ void PartitionedIndexBuilder::AddIndexEntry(
     }
     sub_index_builder_->AddIndexEntry(last_key_in_current_block,
                                       first_key_in_next_block, block_handle);
+    if (sub_index_builder_->seperator_is_key_plus_seq_) {
+      // then we need to apply it to all sub-index builders
+      seperator_is_key_plus_seq_ = true;
+    }
     sub_index_last_key_ = std::string(*last_key_in_current_block);
     entries_.push_back(
         {sub_index_last_key_,
@@ -123,6 +132,10 @@ void PartitionedIndexBuilder::AddIndexEntry(
     sub_index_builder_->AddIndexEntry(last_key_in_current_block,
                                       first_key_in_next_block, block_handle);
     sub_index_last_key_ = std::string(*last_key_in_current_block);
+    if (sub_index_builder_->seperator_is_key_plus_seq_) {
+      // then we need to apply it to all sub-index builders
+      seperator_is_key_plus_seq_ = true;
+    }
   }
 }
 
@@ -146,6 +159,8 @@ Status PartitionedIndexBuilder::Finish(
     // Finish the next partition index in line and Incomplete() to indicate we
     // expect more calls to Finish
     Entry& entry = entries_.front();
+    // Apply the policy to all sub-indexes
+    entry.value->seperator_is_key_plus_seq_ = seperator_is_key_plus_seq_;
     auto s = entry.value->Finish(index_blocks);
     finishing_indexes = true;
     return s.ok() ? Status::Incomplete() : s;

table/index_builder.h

@@ -99,6 +99,8 @@ class IndexBuilder {
   // Get the estimated size for index block.
   virtual size_t EstimatedSize() const = 0;
 
+  virtual bool seperator_is_key_plus_seq() { return true; }
+
  protected:
   const InternalKeyComparator* comparator_;
 };
@@ -115,9 +117,14 @@ class IndexBuilder {
 class ShortenedIndexBuilder : public IndexBuilder {
  public:
   explicit ShortenedIndexBuilder(const InternalKeyComparator* comparator,
-                                 int index_block_restart_interval)
+                                 int index_block_restart_interval,
+                                 uint32_t format_version)
       : IndexBuilder(comparator),
-        index_block_builder_(index_block_restart_interval) {}
+        index_block_builder_(index_block_restart_interval),
+        index_block_builder_without_seq_(index_block_restart_interval) {
+    // Making the default true will disable the feature for old versions
+    seperator_is_key_plus_seq_ = (format_version <= 2);
+  }
 
   virtual void AddIndexEntry(std::string* last_key_in_current_block,
                              const Slice* first_key_in_next_block,
@@ -125,31 +132,57 @@ class ShortenedIndexBuilder : public IndexBuilder {
     if (first_key_in_next_block != nullptr) {
       comparator_->FindShortestSeparator(last_key_in_current_block,
                                          *first_key_in_next_block);
+      if (!seperator_is_key_plus_seq_ &&
+          comparator_->user_comparator()->Compare(
+              ExtractUserKey(*last_key_in_current_block),
+              ExtractUserKey(*first_key_in_next_block)) == 0) {
+        seperator_is_key_plus_seq_ = true;
+      }
     } else {
       comparator_->FindShortSuccessor(last_key_in_current_block);
     }
+    auto sep = Slice(*last_key_in_current_block);
 
     std::string handle_encoding;
     block_handle.EncodeTo(&handle_encoding);
-    index_block_builder_.Add(*last_key_in_current_block, handle_encoding);
+    index_block_builder_.Add(sep, handle_encoding);
+    if (!seperator_is_key_plus_seq_) {
+      index_block_builder_without_seq_.Add(ExtractUserKey(sep),
+                                           handle_encoding);
+    }
   }
 
   using IndexBuilder::Finish;
   virtual Status Finish(
       IndexBlocks* index_blocks,
       const BlockHandle& /*last_partition_block_handle*/) override {
-    index_blocks->index_block_contents = index_block_builder_.Finish();
+    if (seperator_is_key_plus_seq_) {
+      index_blocks->index_block_contents = index_block_builder_.Finish();
+    } else {
+      index_blocks->index_block_contents =
+          index_block_builder_without_seq_.Finish();
+    }
     return Status::OK();
   }
 
   virtual size_t EstimatedSize() const override {
-    return index_block_builder_.CurrentSizeEstimate();
+    if (seperator_is_key_plus_seq_) {
+      return index_block_builder_.CurrentSizeEstimate();
+    } else {
+      return index_block_builder_without_seq_.CurrentSizeEstimate();
+    }
+  }
+
+  virtual bool seperator_is_key_plus_seq() override {
+    return seperator_is_key_plus_seq_;
   }
 
   friend class PartitionedIndexBuilder;
 
  private:
   BlockBuilder index_block_builder_;
+  BlockBuilder index_block_builder_without_seq_;
+  bool seperator_is_key_plus_seq_;
 };
 
 // HashIndexBuilder contains a binary-searchable primary index and the
@@ -183,9 +216,11 @@ class HashIndexBuilder : public IndexBuilder {
  public:
   explicit HashIndexBuilder(const InternalKeyComparator* comparator,
                             const SliceTransform* hash_key_extractor,
-                            int index_block_restart_interval)
+                            int index_block_restart_interval,
+                            int format_version)
       : IndexBuilder(comparator),
-        primary_index_builder_(comparator, index_block_restart_interval),
+        primary_index_builder_(comparator, index_block_restart_interval,
+                               format_version),
         hash_key_extractor_(hash_key_extractor) {}
 
   virtual void AddIndexEntry(std::string* last_key_in_current_block,
@@ -240,6 +275,10 @@ class HashIndexBuilder : public IndexBuilder {
            prefix_meta_block_.size();
   }
 
+  virtual bool seperator_is_key_plus_seq() override {
+    return primary_index_builder_.seperator_is_key_plus_seq();
+  }
+
  private:
   void FlushPendingPrefix() {
     prefix_block_.append(pending_entry_prefix_.data(),
@@ -316,6 +355,10 @@ class PartitionedIndexBuilder : public IndexBuilder {
   // cutting the next partition
   void RequestPartitionCut();
 
+  virtual bool seperator_is_key_plus_seq() override {
+    return seperator_is_key_plus_seq_;
+  }
+
  private:
   void MakeNewSubIndexBuilder();
 
@@ -333,6 +376,7 @@ class PartitionedIndexBuilder : public IndexBuilder {
   // true if Finish is called once but not complete yet.
   bool finishing_indexes = false;
   const BlockBasedTableOptions& table_opt_;
+  bool seperator_is_key_plus_seq_;
   // true if an external entity (such as filter partition builder) request
   // cutting the next partition
   bool partition_cut_requested_ = true;

table/meta_blocks.cc

@@ -71,6 +71,7 @@ void PropertyBlockBuilder::AddTableProperty(const TableProperties& props) {
     Add(TablePropertiesNames::kIndexPartitions, props.index_partitions);
     Add(TablePropertiesNames::kTopLevelIndexSize, props.top_level_index_size);
   }
+  Add(TablePropertiesNames::kIndexKeyIsUserKey, props.index_key_is_user_key);
   Add(TablePropertiesNames::kNumEntries, props.num_entries);
   Add(TablePropertiesNames::kNumDataBlocks, props.num_data_blocks);
   Add(TablePropertiesNames::kFilterSize, props.filter_size);
@@ -192,7 +193,8 @@ Status ReadProperties(const Slice& handle_value, RandomAccessFileReader* file,
   Block properties_block(std::move(block_contents),
                          kDisableGlobalSequenceNumber);
   BlockIter iter;
-  properties_block.NewIterator(BytewiseComparator(), &iter);
+  properties_block.NewIterator(BytewiseComparator(), BytewiseComparator(),
+                               &iter);
 
   auto new_table_properties = new TableProperties();
   // All pre-defined properties of type uint64_t
@@ -203,6 +205,8 @@ Status ReadProperties(const Slice& handle_value, RandomAccessFileReader* file,
        &new_table_properties->index_partitions},
       {TablePropertiesNames::kTopLevelIndexSize,
        &new_table_properties->top_level_index_size},
+      {TablePropertiesNames::kIndexKeyIsUserKey,
+       &new_table_properties->index_key_is_user_key},
       {TablePropertiesNames::kFilterSize, &new_table_properties->filter_size},
       {TablePropertiesNames::kRawKeySize, &new_table_properties->raw_key_size},
       {TablePropertiesNames::kRawValueSize,
@@ -312,7 +316,7 @@ Status ReadTableProperties(RandomAccessFileReader* file, uint64_t file_size,
   Block metaindex_block(std::move(metaindex_contents),
                         kDisableGlobalSequenceNumber);
   std::unique_ptr<InternalIterator> meta_iter(
-      metaindex_block.NewIterator(BytewiseComparator()));
+      metaindex_block.NewIterator(BytewiseComparator(), BytewiseComparator()));
 
   // -- Read property block
   bool found_properties_block = true;
@@ -375,7 +379,8 @@ Status FindMetaBlock(RandomAccessFileReader* file, uint64_t file_size,
                         kDisableGlobalSequenceNumber);
 
   std::unique_ptr<InternalIterator> meta_iter;
-  meta_iter.reset(metaindex_block.NewIterator(BytewiseComparator()));
+  meta_iter.reset(
+      metaindex_block.NewIterator(BytewiseComparator(), BytewiseComparator()));
 
   return FindMetaBlock(meta_iter.get(), meta_block_name, block_handle);
 }
@@ -416,7 +421,8 @@ Status ReadMetaBlock(RandomAccessFileReader* file,
                         kDisableGlobalSequenceNumber);
 
   std::unique_ptr<InternalIterator> meta_iter;
-  meta_iter.reset(metaindex_block.NewIterator(BytewiseComparator()));
+  meta_iter.reset(
+      metaindex_block.NewIterator(BytewiseComparator(), BytewiseComparator()));
 
   BlockHandle block_handle;
   status = FindMetaBlock(meta_iter.get(), meta_block_name, &block_handle);

table/partitioned_filter_block.cc

@@ -113,7 +113,7 @@ PartitionedFilterBlockReader::~PartitionedFilterBlockReader() {
   char cache_key[BlockBasedTable::kMaxCacheKeyPrefixSize + kMaxVarint64Length];
   BlockIter biter;
   BlockHandle handle;
-  idx_on_fltr_blk_->NewIterator(&comparator_, &biter, true);
+  idx_on_fltr_blk_->NewIterator(&comparator_, &comparator_, &biter, true);
   biter.SeekToFirst();
   for (; biter.Valid(); biter.Next()) {
     auto input = biter.value();
@@ -207,7 +207,7 @@ bool PartitionedFilterBlockReader::PrefixMayMatch(
 Slice PartitionedFilterBlockReader::GetFilterPartitionHandle(
     const Slice& entry) {
   BlockIter iter;
-  idx_on_fltr_blk_->NewIterator(&comparator_, &iter, true);
+  idx_on_fltr_blk_->NewIterator(&comparator_, &comparator_, &iter, true);
   iter.Seek(entry);
   if (UNLIKELY(!iter.Valid())) {
     return Slice();
@@ -269,7 +269,7 @@ void PartitionedFilterBlockReader::CacheDependencies(
   auto rep = table_->rep_;
   BlockIter biter;
   BlockHandle handle;
-  idx_on_fltr_blk_->NewIterator(&comparator_, &biter, true);
+  idx_on_fltr_blk_->NewIterator(&comparator_, &comparator_, &biter, true);
   // Index partitions are assumed to be consecuitive. Prefetch them all.
   // Read the first block offset
   biter.SeekToFirst();

table/table_properties.cc

@@ -90,7 +90,12 @@ std::string TableProperties::ToString(
                  prop_delim, kv_delim);
 
   AppendProperty(result, "data block size", data_size, prop_delim, kv_delim);
-  AppendProperty(result, "index block size", index_size, prop_delim, kv_delim);
+  char index_block_size_str[80];
+  snprintf(index_block_size_str, sizeof(index_block_size_str),
+           "index block size (user-key? %d)",
+           static_cast<int>(index_key_is_user_key));
+  AppendProperty(result, index_block_size_str, index_size, prop_delim,
+                 kv_delim);
   if (index_partitions != 0) {
     AppendProperty(result, "# index partitions", index_partitions, prop_delim,
                    kv_delim);
@@ -155,6 +160,7 @@ void TableProperties::Add(const TableProperties& tp) {
   index_size += tp.index_size;
   index_partitions += tp.index_partitions;
   top_level_index_size += tp.top_level_index_size;
+  index_key_is_user_key += tp.index_key_is_user_key;
   filter_size += tp.filter_size;
   raw_key_size += tp.raw_key_size;
   raw_value_size += tp.raw_value_size;
@@ -170,6 +176,8 @@ const std::string TablePropertiesNames::kIndexPartitions =
     "rocksdb.index.partitions";
 const std::string TablePropertiesNames::kTopLevelIndexSize =
     "rocksdb.top-level.index.size";
+const std::string TablePropertiesNames::kIndexKeyIsUserKey =
+    "rocksdb.index.key.is.user.key";
 const std::string TablePropertiesNames::kFilterSize =
     "rocksdb.filter.size";
 const std::string TablePropertiesNames::kRawKeySize =

table/table_test.cc

@@ -237,7 +237,7 @@ class BlockConstructor: public Constructor {
   }
   virtual InternalIterator* NewIterator(
       const SliceTransform* /*prefix_extractor*/) const override {
-    return block_->NewIterator(comparator_);
+    return block_->NewIterator(comparator_, comparator_);
   }
 
  private:
@@ -2115,7 +2115,7 @@ TEST_F(BlockBasedTableTest, FilterBlockInBlockCache) {
   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(), internal_key.Encode(), &get_context,
                         moptions4.prefix_extractor.get()));
   ASSERT_STREQ(value.data(), "hello");
   BlockCachePropertiesSnapshot props(options.statistics.get());
@@ -2427,7 +2427,8 @@ TEST_F(BlockBasedTableTest, BlockCacheLeak) {
   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));
+    InternalKey ikey(key, kMaxSequenceNumber, kTypeValue);
+    ASSERT_TRUE(table_reader->TEST_KeyInCache(ReadOptions(), ikey.Encode()));
   }
   c.ResetTableReader();
 
@@ -2439,7 +2440,8 @@ TEST_F(BlockBasedTableTest, BlockCacheLeak) {
   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));
+    InternalKey ikey(key, kMaxSequenceNumber, kTypeValue);
+    ASSERT_TRUE(!table_reader->TEST_KeyInCache(ReadOptions(), ikey.Encode()));
   }
   c.ResetTableReader();
 }

tools/db_bench_tool.cc

@@ -100,13 +100,13 @@ DEFINE_string(
     "readreverse,"
     "compact,"
     "compactall,"
-    "readrandom,"
     "multireadrandom,"
     "readseq,"
     "readtocache,"
     "readreverse,"
     "readwhilewriting,"
     "readwhilemerging,"
+    "readwhilescanning,"
     "readrandomwriterandom,"
     "updaterandom,"
     "xorupdaterandom,"
@@ -149,6 +149,8 @@ DEFINE_string(
     "reads\n"
     "\treadwhilemerging      -- 1 merger, N threads doing random "
     "reads\n"
+    "\treadwhilescanning     -- 1 thread doing full table scan, "
+    "N threads doing random reads\n"
     "\treadrandomwriterandom -- N threads doing random-read, "
     "random-write\n"
     "\tupdaterandom  -- N threads doing read-modify-write for random "
@@ -2524,6 +2526,9 @@ void VerifyDBFromDB(std::string& truth_db_name) {
       } else if (name == "readwhilemerging") {
         num_threads++;  // Add extra thread for writing
         method = &Benchmark::ReadWhileMerging;
+      } else if (name == "readwhilescanning") {
+        num_threads++;  // Add extra thread for scaning
+        method = &Benchmark::ReadWhileScanning;
       } else if (name == "readrandomwriterandom") {
         method = &Benchmark::ReadRandomWriteRandom;
       } else if (name == "readrandommergerandom") {
@@ -4507,6 +4512,45 @@ void VerifyDBFromDB(std::string& truth_db_name) {
     thread->stats.AddBytes(bytes);
   }
 
+  void ReadWhileScanning(ThreadState* thread) {
+    if (thread->tid > 0) {
+      ReadRandom(thread);
+    } else {
+      BGScan(thread);
+    }
+  }
+
+  void BGScan(ThreadState* thread) {
+    if (FLAGS_num_multi_db > 0) {
+      fprintf(stderr, "Not supporting multiple DBs.\n");
+      abort();
+    }
+    assert(db_.db != nullptr);
+    ReadOptions read_options;
+    Iterator* iter = db_.db->NewIterator(read_options);
+
+    fprintf(stderr, "num reads to do %" PRIu64 "\n", reads_);
+    Duration duration(FLAGS_duration, reads_);
+    uint64_t num_seek_to_first = 0;
+    uint64_t num_next = 0;
+    while (!duration.Done(1)) {
+      if (!iter->Valid()) {
+        iter->SeekToFirst();
+        num_seek_to_first++;
+      } else if (!iter->status().ok()) {
+        fprintf(stderr, "Iterator error: %s\n",
+                iter->status().ToString().c_str());
+        abort();
+      } else {
+        iter->Next();
+        num_next++;
+      }
+
+      thread->stats.FinishedOps(&db_, db_.db, 1, kSeek);
+    }
+    delete iter;
+  }
+
   // Given a key K and value V, this puts (K+"0", V), (K+"1", V), (K+"2", V)
   // in DB atomically i.e in a single batch. Also refer GetMany.
   Status PutMany(DB* db, const WriteOptions& writeoptions, const Slice& key,

tools/db_stress.cc

@@ -879,9 +879,9 @@ class SharedState {
             FLAGS_expected_values_path, &expected_mmap_buffer_);
       }
       if (status.ok()) {
-        assert(expected_mmap_buffer_->length == expected_values_size);
+        assert(expected_mmap_buffer_->GetLen() == expected_values_size);
         values_ =
-            static_cast<std::atomic<uint32_t>*>(expected_mmap_buffer_->base);
+            static_cast<std::atomic<uint32_t>*>(expected_mmap_buffer_->GetBase());
         assert(values_ != nullptr);
       } else {
         fprintf(stderr, "Failed opening shared file '%s' with error: %s\n",

util/testutil.h

@@ -87,13 +87,6 @@ class PlainInternalKeyComparator : public InternalKeyComparator {
   virtual int Compare(const Slice& a, const Slice& b) const override {
     return user_comparator()->Compare(a, b);
   }
-  virtual void FindShortestSeparator(std::string* start,
-                                     const Slice& limit) const override {
-    user_comparator()->FindShortestSeparator(start, limit);
-  }
-  virtual void FindShortSuccessor(std::string* key) const override {
-    user_comparator()->FindShortSuccessor(key);
-  }
 };
 #endif
 

utilities/persistent_cache/persistent_cache_test.cc

@@ -157,7 +157,7 @@ TEST_F(PersistentCacheTierTest, DISABLED_BlockCacheInsertWithFileCreateError) {
   rocksdb::SyncPoint::GetInstance()->ClearAllCallBacks();
 }
 
-#ifdef TRAVIS
+#if defined(TRAVIS) || defined(ROCKSDB_VALGRIND_RUN)
 // Travis is unable to handle the normal version of the tests running out of
 // fds, out of space and timeouts. This is an easier version of the test
 // specifically written for Travis
@@ -435,7 +435,7 @@ void PersistentCacheDBTest::RunTest(
   }
 }
 
-#ifdef TRAVIS
+#if defined(TRAVIS) || defined(ROCKSDB_VALGRIND_RUN)
 // Travis is unable to handle the normal version of the tests running out of
 // fds, out of space and timeouts. This is an easier version of the test
 // specifically written for Travis

utilities/transactions/pessimistic_transaction_db.cc

@@ -26,6 +26,7 @@
 #include "utilities/transactions/pessimistic_transaction.h"
 #include "utilities/transactions/transaction_db_mutex_impl.h"
 #include "utilities/transactions/write_prepared_txn_db.h"
+#include "utilities/transactions/write_unprepared_txn_db.h"
 
 namespace rocksdb {
 
@@ -216,7 +217,7 @@ Status TransactionDB::Open(
   DBOptions db_options_2pc = db_options;
   PrepareWrap(&db_options_2pc, &column_families_copy,
               &compaction_enabled_cf_indices);
-  const bool use_seq_per_batch = txn_db_options.write_policy == WRITE_PREPARED;
+  const bool use_seq_per_batch = txn_db_options.write_policy >= WRITE_PREPARED;
   s = DBImpl::Open(db_options_2pc, dbname, column_families_copy, handles, &db,
                    use_seq_per_batch);
   if (s.ok()) {
@@ -264,7 +265,9 @@ Status TransactionDB::WrapDB(
   std::unique_ptr<PessimisticTransactionDB> txn_db;
   switch (txn_db_options.write_policy) {
     case WRITE_UNPREPARED:
-      return Status::NotSupported("WRITE_UNPREPARED is not implemented yet");
+      txn_db.reset(new WriteUnpreparedTxnDB(
+          db, PessimisticTransactionDB::ValidateTxnDBOptions(txn_db_options)));
+      break;
     case WRITE_PREPARED:
       txn_db.reset(new WritePreparedTxnDB(
           db, PessimisticTransactionDB::ValidateTxnDBOptions(txn_db_options)));
@@ -297,7 +300,9 @@ Status TransactionDB::WrapStackableDB(
 
   switch (txn_db_options.write_policy) {
     case WRITE_UNPREPARED:
-      return Status::NotSupported("WRITE_UNPREPARED is not implemented yet");
+      txn_db.reset(new WriteUnpreparedTxnDB(
+          db, PessimisticTransactionDB::ValidateTxnDBOptions(txn_db_options)));
+      break;
     case WRITE_PREPARED:
       txn_db.reset(new WritePreparedTxnDB(
           db, PessimisticTransactionDB::ValidateTxnDBOptions(txn_db_options)));

utilities/transactions/transaction_test.cc

@@ -45,11 +45,17 @@ INSTANTIATE_TEST_CASE_P(
     ::testing::Values(std::make_tuple(false, false, WRITE_COMMITTED),
                       std::make_tuple(false, true, WRITE_COMMITTED),
                       std::make_tuple(false, false, WRITE_PREPARED),
-                      std::make_tuple(false, true, WRITE_PREPARED)));
+                      std::make_tuple(false, true, WRITE_PREPARED),
+                      std::make_tuple(false, false, WRITE_UNPREPARED),
+                      std::make_tuple(false, true, WRITE_UNPREPARED)));
 INSTANTIATE_TEST_CASE_P(
     StackableDBAsBaseDB, TransactionTest,
     ::testing::Values(std::make_tuple(true, true, WRITE_COMMITTED),
-                      std::make_tuple(true, true, WRITE_PREPARED)));
+                      std::make_tuple(true, true, WRITE_PREPARED),
+                      std::make_tuple(true, true, WRITE_UNPREPARED)));
+
+// MySQLStyleTransactionTest takes far too long for valgrind to run.
+#ifndef ROCKSDB_VALGRIND_RUN
 INSTANTIATE_TEST_CASE_P(
     MySQLStyleTransactionTest, MySQLStyleTransactionTest,
     ::testing::Values(std::make_tuple(false, false, WRITE_COMMITTED),
@@ -59,7 +65,12 @@ INSTANTIATE_TEST_CASE_P(
                       std::make_tuple(false, false, WRITE_PREPARED),
                       std::make_tuple(false, true, WRITE_PREPARED),
                       std::make_tuple(true, false, WRITE_PREPARED),
-                      std::make_tuple(true, true, WRITE_PREPARED)));
+                      std::make_tuple(true, true, WRITE_PREPARED),
+                      std::make_tuple(false, false, WRITE_UNPREPARED),
+                      std::make_tuple(false, true, WRITE_UNPREPARED),
+                      std::make_tuple(true, false, WRITE_UNPREPARED),
+                      std::make_tuple(true, true, WRITE_UNPREPARED)));
+#endif  // ROCKSDB_VALGRIND_RUN
 
 TEST_P(TransactionTest, DoubleEmptyWrite) {
   WriteOptions write_options;
@@ -1775,10 +1786,10 @@ TEST_P(TransactionTest, TwoPhaseLogRollingTest2) {
       ASSERT_EQ(cfh_a->cfd()->GetLogNumber(), db_impl->TEST_LogfileNumber());
       break;
     case WRITE_PREPARED:
+    case WRITE_UNPREPARED:
       // This cf is not flushed yet and should ref the log that has its data
       ASSERT_EQ(cfh_a->cfd()->GetLogNumber(), prepare_log_no);
       break;
-    case WRITE_UNPREPARED:
     default:
       assert(false);
   }
@@ -4803,6 +4814,7 @@ TEST_P(TransactionTest, ExpiredTransactionDataRace1) {
   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
 }
 
+#ifndef ROCKSDB_VALGRIND_RUN
 namespace {
 Status TransactionStressTestInserter(TransactionDB* db,
                                      const size_t num_transactions,
@@ -4890,6 +4902,7 @@ TEST_P(MySQLStyleTransactionTest, TransactionStressTest) {
                                                !TAKE_SNAPSHOT);
   ASSERT_OK(s);
 }
+#endif  // ROCKSDB_VALGRIND_RUN
 
 TEST_P(TransactionTest, MemoryLimitTest) {
   TransactionOptions txn_options;

utilities/transactions/write_prepared_transaction_test.cc

@@ -521,6 +521,7 @@ class WritePreparedTransactionTest
                                          std::get<2>(GetParam())){};
 };
 
+#ifndef ROCKSDB_VALGRIND_RUN
 class SnapshotConcurrentAccessTest
     : public WritePreparedTransactionTestBase,
       virtual public ::testing::WithParamInterface<
@@ -539,6 +540,7 @@ class SnapshotConcurrentAccessTest
   size_t split_id_;
   size_t split_cnt_;
 };
+#endif  // ROCKSDB_VALGRIND_RUN
 
 class SeqAdvanceConcurrentTest
     : public WritePreparedTransactionTestBase,
@@ -562,8 +564,10 @@ class SeqAdvanceConcurrentTest
 INSTANTIATE_TEST_CASE_P(
     WritePreparedTransactionTest, WritePreparedTransactionTest,
     ::testing::Values(std::make_tuple(false, false, WRITE_PREPARED),
-                      std::make_tuple(false, true, WRITE_PREPARED)));
+                      std::make_tuple(false, true, WRITE_PREPARED),
+                      std::make_tuple(false, true, WRITE_UNPREPARED)));
 
+#ifndef ROCKSDB_VALGRIND_RUN
 INSTANTIATE_TEST_CASE_P(
     TwoWriteQueues, SnapshotConcurrentAccessTest,
     ::testing::Values(std::make_tuple(false, true, WRITE_PREPARED, 0, 20),
@@ -585,7 +589,27 @@ INSTANTIATE_TEST_CASE_P(
                       std::make_tuple(false, true, WRITE_PREPARED, 16, 20),
                       std::make_tuple(false, true, WRITE_PREPARED, 17, 20),
                       std::make_tuple(false, true, WRITE_PREPARED, 18, 20),
-                      std::make_tuple(false, true, WRITE_PREPARED, 19, 20)));
+                      std::make_tuple(false, true, WRITE_PREPARED, 19, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 0, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 1, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 2, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 3, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 4, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 5, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 6, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 7, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 8, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 9, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 10, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 11, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 12, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 13, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 14, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 15, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 16, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 17, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 18, 20),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 19, 20)));
 
 INSTANTIATE_TEST_CASE_P(
     OneWriteQueue, SnapshotConcurrentAccessTest,
@@ -608,7 +632,27 @@ INSTANTIATE_TEST_CASE_P(
                       std::make_tuple(false, false, WRITE_PREPARED, 16, 20),
                       std::make_tuple(false, false, WRITE_PREPARED, 17, 20),
                       std::make_tuple(false, false, WRITE_PREPARED, 18, 20),
-                      std::make_tuple(false, false, WRITE_PREPARED, 19, 20)));
+                      std::make_tuple(false, false, WRITE_PREPARED, 19, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 0, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 1, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 2, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 3, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 4, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 5, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 6, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 7, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 8, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 9, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 10, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 11, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 12, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 13, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 14, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 15, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 16, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 17, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 18, 20),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 19, 20)));
 
 INSTANTIATE_TEST_CASE_P(
     TwoWriteQueues, SeqAdvanceConcurrentTest,
@@ -621,7 +665,17 @@ INSTANTIATE_TEST_CASE_P(
                       std::make_tuple(false, true, WRITE_PREPARED, 6, 10),
                       std::make_tuple(false, true, WRITE_PREPARED, 7, 10),
                       std::make_tuple(false, true, WRITE_PREPARED, 8, 10),
-                      std::make_tuple(false, true, WRITE_PREPARED, 9, 10)));
+                      std::make_tuple(false, true, WRITE_PREPARED, 9, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 0, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 1, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 2, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 3, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 4, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 5, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 6, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 7, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 8, 10),
+                      std::make_tuple(false, true, WRITE_UNPREPARED, 9, 10)));
 
 INSTANTIATE_TEST_CASE_P(
     OneWriteQueue, SeqAdvanceConcurrentTest,
@@ -634,7 +688,18 @@ INSTANTIATE_TEST_CASE_P(
                       std::make_tuple(false, false, WRITE_PREPARED, 6, 10),
                       std::make_tuple(false, false, WRITE_PREPARED, 7, 10),
                       std::make_tuple(false, false, WRITE_PREPARED, 8, 10),
-                      std::make_tuple(false, false, WRITE_PREPARED, 9, 10)));
+                      std::make_tuple(false, false, WRITE_PREPARED, 9, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 0, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 1, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 2, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 3, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 4, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 5, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 6, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 7, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 8, 10),
+                      std::make_tuple(false, false, WRITE_UNPREPARED, 9, 10)));
+#endif  // ROCKSDB_VALGRIND_RUN
 
 TEST_P(WritePreparedTransactionTest, CommitMapTest) {
   WritePreparedTxnDB* wp_db = dynamic_cast<WritePreparedTxnDB*>(db);
@@ -841,6 +906,7 @@ TEST_P(WritePreparedTransactionTest, CheckAgainstSnapshotsTest) {
 
 // This test is too slow for travis
 #ifndef TRAVIS
+#ifndef ROCKSDB_VALGRIND_RUN
 // Test that CheckAgainstSnapshots will not miss a live snapshot if it is run in
 // parallel with UpdateSnapshots.
 TEST_P(SnapshotConcurrentAccessTest, SnapshotConcurrentAccessTest) {
@@ -919,6 +985,7 @@ TEST_P(SnapshotConcurrentAccessTest, SnapshotConcurrentAccessTest) {
   }
   printf("\n");
 }
+#endif  // ROCKSDB_VALGRIND_RUN
 #endif  // TRAVIS
 
 // This test clarifies the contract of AdvanceMaxEvictedSeq method

utilities/transactions/write_unprepared_txn.cc

@@ -0,0 +1,19 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#ifndef ROCKSDB_LITE
+
+#include "utilities/transactions/write_unprepared_txn.h"
+
+#ifndef __STDC_FORMAT_MACROS
+#define __STDC_FORMAT_MACROS
+#endif
+
+namespace rocksdb {
+
+
+}  // namespace rocksdb
+
+#endif  // ROCKSDB_LITE

utilities/transactions/write_unprepared_txn.h

@@ -0,0 +1,21 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#ifndef ROCKSDB_LITE
+
+#include "utilities/transactions/write_prepared_txn.h"
+
+namespace rocksdb {
+
+class WriteUnpreparedTxn : public WritePreparedTxn {
+  using WritePreparedTxn::WritePreparedTxn;
+
+};
+
+}  // namespace rocksdb
+
+#endif  // ROCKSDB_LITE

utilities/transactions/write_unprepared_txn_db.cc

@@ -0,0 +1,29 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#ifndef ROCKSDB_LITE
+
+#ifndef __STDC_FORMAT_MACROS
+#define __STDC_FORMAT_MACROS
+#endif
+
+#include "utilities/transactions/write_unprepared_txn_db.h"
+#include "rocksdb/utilities/transaction_db.h"
+
+namespace rocksdb {
+
+Transaction* WriteUnpreparedTxnDB::BeginTransaction(
+const WriteOptions& write_options, const TransactionOptions& txn_options,
+Transaction* old_txn) {
+  if (old_txn != nullptr) {
+    ReinitializeTransaction(old_txn, write_options, txn_options);
+    return old_txn;
+  } else {
+    return new WriteUnpreparedTxn(this, write_options, txn_options);
+  }
+}
+
+}  //  namespace rocksdb
+#endif  // ROCKSDB_LITE

utilities/transactions/write_unprepared_txn_db.h

@@ -0,0 +1,27 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+#ifndef ROCKSDB_LITE
+
+#ifndef __STDC_FORMAT_MACROS
+#define __STDC_FORMAT_MACROS
+#endif
+
+#include "utilities/transactions/write_prepared_txn_db.h"
+
+#include "utilities/transactions/write_unprepared_txn.h"
+
+namespace rocksdb {
+
+class WriteUnpreparedTxnDB : public WritePreparedTxnDB {
+  using WritePreparedTxnDB::WritePreparedTxnDB;
+
+  Transaction* BeginTransaction(const WriteOptions& write_options, const TransactionOptions& txn_options,
+                                Transaction* old_txn) override;
+};
+
+}  //  namespace rocksdb
+#endif  // ROCKSDB_LITE