Remove stale entries from L0 files when UDT is not persisted (#13035)

Summary:
When user-defined timestamps are not persisted, currently we replace the actual timestamp with min timestamp after an entry is output from compaction iterator. Compaction iterator won't be able to help with removing stale entries this way. This PR adds a wrapper iterator `TimestampStrippingIterator` for `MemTableIterator` that does the min timestamp replacement at the memtable iteration step. It is used by flush and can help remove stale entries from landing in L0 files.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/13035

Test Plan: Added unit test

Reviewed By: pdillinger, cbi42

Differential Revision: D63423682

Pulled By: jowlyzhang

fbshipit-source-id: 087dcc9cee97b9ea51b8d2b88dc91c2984d54e55

HISTORY.md

@@ -1,6 +1,36 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 9.7.2 (10/08/2024)
+### Bug Fixes
+* Fix a bug for surfacing write unix time: `Iterator::GetProperty("rocksdb.iterator.write-time")` for non-L0 files.
+
+## 9.7.1 (09/26/2024)
+### Bug Fixes
+* Several DB option settings could be lost through `GetOptionsFromString()`, possibly elsewhere as well. Affected options, now fixed:`background_close_inactive_wals`, `write_dbid_to_manifest`, `write_identity_file`, `prefix_seek_opt_in_only`
+* Fix under counting of allocated memory in the compressed secondary cache due to looking at the compressed block size rather than the actual memory allocated, which could be larger due to internal fragmentation.
+* Skip insertion of compressed blocks in the secondary cache if the lowest_used_cache_tier DB option is kVolatileTier.
+
+## 9.7.0 (09/20/2024)
+### New Features
+* Make Cache a customizable class that can be instantiated by the object registry.
+* Add new option `prefix_seek_opt_in_only` that makes iterators generally safer when you might set a `prefix_extractor`. When `prefix_seek_opt_in_only=true`, which is expected to be the future default, prefix seek is only used when `prefix_same_as_start` or `auto_prefix_mode` are set. Also, `prefix_same_as_start` and `auto_prefix_mode` now allow prefix filtering even with `total_order_seek=true`.
+* Add a new table property "rocksdb.key.largest.seqno" which records the largest sequence number of all keys in file. It is verified to be zero during SST file ingestion.
+
+### Behavior Changes
+* Changed the semantics of the BlobDB configuration option `blob_garbage_collection_force_threshold` to define a threshold for the overall garbage ratio of all blob files currently eligible for garbage collection (according to `blob_garbage_collection_age_cutoff`). This can provide better control over space amplification at the cost of slightly higher write amplification.
+* Set `write_dbid_to_manifest=true` by default. This means DB ID will now be preserved through backups, checkpoints, etc. by default. Also add `write_identity_file` option which can be set to false for anticipated future behavior.
+* In FIFO compaction, compactions for changing file temperature (configured by option `file_temperature_age_thresholds`) will compact one file at a time, instead of merging multiple eligible file together (#13018).
+* Support ingesting db generated files using hard link, i.e. IngestExternalFileOptions::move_files/link_files and IngestExternalFileOptions::allow_db_generated_files.
+* Add a new file ingestion option `IngestExternalFileOptions::link_files` to hard link input files and preserve original files links after ingestion.
+* DB::Close now untracks files in SstFileManager, making avaialble any space used
+by them. Prior to this change they would be orphaned until the DB is re-opened.
+
+### Bug Fixes
+* Fix a bug in CompactRange() where result files may not be compacted in any future compaction. This can only happen when users configure CompactRangeOptions::change_level to true and the change level step of manual compaction fails (#13009).
+* Fix handling of dynamic change of `prefix_extractor` with memtable prefix filter. Previously, prefix seek could mix different prefix interpretations between memtable and SST files. Now the latest `prefix_extractor` at the time of iterator creation or refresh is respected.
+* Fix a bug with manual_wal_flush and auto error recovery from WAL failure that may cause CFs to be inconsistent (#12995). The fix will set potential WAL write failure as fatal error when manual_wal_flush is true, and disables auto error recovery from these errors.
+
 ## 9.6.0 (08/19/2024)
 ### New Features
 * *Best efforts recovery supports recovering to incomplete Version with a clean seqno cut that presents a valid point in time view from the user's perspective, if versioning history doesn't include atomic flush.

cache/compressed_secondary_cache.cc

@@ -79,7 +79,11 @@ std::unique_ptr<SecondaryCacheResultHandle> CompressedSecondaryCache::Lookup(
     data_ptr = GetVarint32Ptr(data_ptr, data_ptr + 1,
                               static_cast<uint32_t*>(&source_32));
     source = static_cast<CacheTier>(source_32);
-    handle_value_charge -= (data_ptr - ptr->get());
+    uint64_t data_size = 0;
+    data_ptr = GetVarint64Ptr(data_ptr, ptr->get() + handle_value_charge,
+                              static_cast<uint64_t*>(&data_size));
+    assert(handle_value_charge > data_size);
+    handle_value_charge = data_size;
   }
   MemoryAllocator* allocator = cache_options_.memory_allocator.get();
 
@@ -169,13 +173,15 @@ Status CompressedSecondaryCache::InsertInternal(
   }
 
   auto internal_helper = GetHelper(cache_options_.enable_custom_split_merge);
-  char header[10];
+  char header[20];
   char* payload = header;
   payload = EncodeVarint32(payload, static_cast<uint32_t>(type));
   payload = EncodeVarint32(payload, static_cast<uint32_t>(source));
+  size_t data_size = (*helper->size_cb)(value);
+  char* data_size_ptr = payload;
+  payload = EncodeVarint64(payload, data_size);
 
   size_t header_size = payload - header;
-  size_t data_size = (*helper->size_cb)(value);
   size_t total_size = data_size + header_size;
   CacheAllocationPtr ptr =
       AllocateBlock(total_size, cache_options_.memory_allocator.get());
@@ -210,6 +216,8 @@ Status CompressedSecondaryCache::InsertInternal(
 
     val = Slice(compressed_val);
     data_size = compressed_val.size();
+    payload = EncodeVarint64(data_size_ptr, data_size);
+    header_size = payload - header;
     total_size = header_size + data_size;
     PERF_COUNTER_ADD(compressed_sec_cache_compressed_bytes, data_size);
 
@@ -222,14 +230,21 @@ Status CompressedSecondaryCache::InsertInternal(
 
   PERF_COUNTER_ADD(compressed_sec_cache_insert_real_count, 1);
   if (cache_options_.enable_custom_split_merge) {
-    size_t charge{0};
-    CacheValueChunk* value_chunks_head =
-        SplitValueIntoChunks(val, cache_options_.compression_type, charge);
-    return cache_->Insert(key, value_chunks_head, internal_helper, charge);
+    size_t split_charge{0};
+    CacheValueChunk* value_chunks_head = SplitValueIntoChunks(
+        val, cache_options_.compression_type, split_charge);
+    return cache_->Insert(key, value_chunks_head, internal_helper,
+                          split_charge);
   } else {
+#ifdef ROCKSDB_MALLOC_USABLE_SIZE
+    size_t charge = malloc_usable_size(ptr.get());
+#else
+    size_t charge = total_size;
+#endif
     std::memcpy(ptr.get(), header, header_size);
     CacheAllocationPtr* buf = new CacheAllocationPtr(std::move(ptr));
-    return cache_->Insert(key, buf, internal_helper, total_size);
+    charge += sizeof(CacheAllocationPtr);
+    return cache_->Insert(key, buf, internal_helper, charge);
   }
 }
 
@@ -398,6 +413,21 @@ const Cache::CacheItemHelper* CompressedSecondaryCache::GetHelper(
   }
 }
 
+size_t CompressedSecondaryCache::TEST_GetCharge(const Slice& key) {
+  Cache::Handle* lru_handle = cache_->Lookup(key);
+  if (lru_handle == nullptr) {
+    return 0;
+  }
+
+  size_t charge = cache_->GetCharge(lru_handle);
+  if (cache_->Value(lru_handle) != nullptr &&
+      !cache_options_.enable_custom_split_merge) {
+    charge -= 10;
+  }
+  cache_->Release(lru_handle, /*erase_if_last_ref=*/false);
+  return charge;
+}
+
 std::shared_ptr<SecondaryCache>
 CompressedSecondaryCacheOptions::MakeSharedSecondaryCache() const {
   return std::make_shared<CompressedSecondaryCache>(*this);

cache/compressed_secondary_cache.h

@@ -139,6 +139,8 @@ class CompressedSecondaryCache : public SecondaryCache {
                         const Cache::CacheItemHelper* helper,
                         CompressionType type, CacheTier source);
 
+  size_t TEST_GetCharge(const Slice& key);
+
   // TODO: clean up to use cleaner interfaces in typed_cache.h
   const Cache::CacheItemHelper* GetHelper(bool enable_custom_split_merge) const;
   std::shared_ptr<Cache> cache_;

cache/compressed_secondary_cache_test.cc

@@ -39,6 +39,8 @@ class CompressedSecondaryCacheTestBase : public testing::Test,
  protected:
   void BasicTestHelper(std::shared_ptr<SecondaryCache> sec_cache,
                        bool sec_cache_is_compressed) {
+    CompressedSecondaryCache* comp_sec_cache =
+        static_cast<CompressedSecondaryCache*>(sec_cache.get());
     get_perf_context()->Reset();
     bool kept_in_sec_cache{true};
     // Lookup an non-existent key.
@@ -66,6 +68,8 @@ class CompressedSecondaryCacheTestBase : public testing::Test,
     ASSERT_OK(sec_cache->Insert(key1, &item1, GetHelper(), false));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 1);
 
+    ASSERT_GT(comp_sec_cache->TEST_GetCharge(key1), 1000);
+
     std::unique_ptr<SecondaryCacheResultHandle> handle1_2 =
         sec_cache->Lookup(key1, GetHelper(), this, true, /*advise_erase=*/true,
                           /*stats=*/nullptr, kept_in_sec_cache);

cache/secondary_cache_adapter.cc

@@ -271,7 +271,8 @@ Status CacheWithSecondaryAdapter::Insert(const Slice& key, ObjectPtr value,
   // Warm up the secondary cache with the compressed block. The secondary
   // cache may choose to ignore it based on the admission policy.
   if (value != nullptr && !compressed_value.empty() &&
-      adm_policy_ == TieredAdmissionPolicy::kAdmPolicyThreeQueue) {
+      adm_policy_ == TieredAdmissionPolicy::kAdmPolicyThreeQueue &&
+      helper->IsSecondaryCacheCompatible()) {
     Status status = secondary_cache_->InsertSaved(key, compressed_value, type);
     assert(status.ok() || status.IsNotSupported());
   }

cache/tiered_secondary_cache_test.cc

@@ -765,6 +765,54 @@ TEST_F(DBTieredSecondaryCacheTest, IterateTest) {
   Destroy(options);
 }
 
+TEST_F(DBTieredSecondaryCacheTest, VolatileTierTest) {
+  if (!LZ4_Supported()) {
+    ROCKSDB_GTEST_SKIP("This test requires LZ4 support.");
+    return;
+  }
+
+  BlockBasedTableOptions table_options;
+  // We want a block cache of size 5KB, and a compressed secondary cache of
+  // size 5KB. However, we specify a block cache size of 256KB here in order
+  // to take into account the cache reservation in the block cache on
+  // behalf of the compressed cache. The unit of cache reservation is 256KB.
+  // The effective block cache capacity will be calculated as 256 + 5 = 261KB,
+  // and 256KB will be reserved for the compressed cache, leaving 5KB for
+  // the primary block cache. We only have to worry about this here because
+  // the cache size is so small.
+  table_options.block_cache = NewCache(256 * 1024, 5 * 1024, 256 * 1024);
+  table_options.block_size = 4 * 1024;
+  table_options.cache_index_and_filter_blocks = false;
+  Options options = GetDefaultOptions();
+  options.create_if_missing = true;
+  options.compression = kLZ4Compression;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+  // Disable paranoid_file_checks so that flush will not read back the newly
+  // written file
+  options.paranoid_file_checks = false;
+  options.lowest_used_cache_tier = CacheTier::kVolatileTier;
+  DestroyAndReopen(options);
+  Random rnd(301);
+  const int N = 256;
+  for (int i = 0; i < N; i++) {
+    std::string p_v;
+    test::CompressibleString(&rnd, 0.5, 1007, &p_v);
+    ASSERT_OK(Put(Key(i), p_v));
+  }
+
+  ASSERT_OK(Flush());
+
+  // Since lowest_used_cache_tier is the volatile tier, nothing should be
+  // inserted in the secondary cache.
+  std::string v = Get(Key(0));
+  ASSERT_EQ(1007, v.size());
+  ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 0u);
+  ASSERT_EQ(nvm_sec_cache()->num_misses(), 0u);
+
+  Destroy(options);
+}
+
 class DBTieredAdmPolicyTest
     : public DBTieredSecondaryCacheTest,
       public testing::WithParamInterface<TieredAdmissionPolicy> {};

db/builder.cc

@@ -206,10 +206,6 @@ Status BuildTable(
         /*compaction=*/nullptr, compaction_filter.get(),
         /*shutting_down=*/nullptr, db_options.info_log, full_history_ts_low);
 
-    const size_t ts_sz = ucmp->timestamp_size();
-    const bool logical_strip_timestamp =
-        ts_sz > 0 && !ioptions.persist_user_defined_timestamps;
-
     SequenceNumber smallest_preferred_seqno = kMaxSequenceNumber;
     std::string key_after_flush_buf;
     std::string value_buf;
@@ -222,16 +218,6 @@ Status BuildTable(
       Slice key_after_flush = key_after_flush_buf;
       Slice value_after_flush = value;
 
-      // If user defined timestamps will be stripped from user key after flush,
-      // the in memory version of the key act logically the same as one with a
-      // minimum timestamp. We update the timestamp here so file boundary and
-      // output validator, block builder all see the effect of the stripping.
-      if (logical_strip_timestamp) {
-        key_after_flush_buf.clear();
-        ReplaceInternalKeyWithMinTimestamp(&key_after_flush_buf, key, ts_sz);
-        key_after_flush = key_after_flush_buf;
-      }
-
       if (ikey.type == kTypeValuePreferredSeqno) {
         auto [unpacked_value, unix_write_time] =
             ParsePackedValueWithWriteTime(value);
@@ -291,11 +277,7 @@ Status BuildTable(
       Slice last_tombstone_start_user_key{};
       for (range_del_it->SeekToFirst(); range_del_it->Valid();
            range_del_it->Next()) {
-        // When user timestamp should not be persisted, we logically strip a
-        // range tombstone's start and end key's timestamp (replace it with min
-        // timestamp) before passing them along to table builder and to update
-        // file boundaries.
-        auto tombstone = range_del_it->Tombstone(logical_strip_timestamp);
+        auto tombstone = range_del_it->Tombstone();
         std::pair<InternalKey, Slice> kv = tombstone.Serialize();
         builder->Add(kv.first.Encode(), kv.second);
         InternalKey tombstone_end = tombstone.SerializeEndKey();

db/column_family_test.cc

@@ -3870,6 +3870,91 @@ TEST_F(ManualFlushSkipRetainUDTTest, ManualFlush) {
   Close();
 }
 
+TEST_F(ManualFlushSkipRetainUDTTest, FlushRemovesStaleEntries) {
+  column_family_options_.max_write_buffer_number = 4;
+  Open();
+  ASSERT_OK(db_->IncreaseFullHistoryTsLow(handles_[0], EncodeAsUint64(0)));
+
+  ColumnFamilyHandle* cfh = db_->DefaultColumnFamily();
+  ColumnFamilyData* cfd =
+      static_cast_with_check<ColumnFamilyHandleImpl>(cfh)->cfd();
+  for (int version = 0; version < 100; version++) {
+    if (version == 50) {
+      ASSERT_OK(static_cast_with_check<DBImpl>(db_)->TEST_SwitchMemtable(cfd));
+    }
+    ASSERT_OK(
+        Put(0, "foo", EncodeAsUint64(version), "v" + std::to_string(version)));
+  }
+
+  ASSERT_OK(Flush(0));
+  TablePropertiesCollection tables_properties;
+  ASSERT_OK(db_->GetPropertiesOfAllTables(&tables_properties));
+  ASSERT_EQ(1, tables_properties.size());
+  std::shared_ptr<const TableProperties> table_properties =
+      tables_properties.begin()->second;
+  ASSERT_EQ(1, table_properties->num_entries);
+  ASSERT_EQ(0, table_properties->num_deletions);
+  ASSERT_EQ(0, table_properties->num_range_deletions);
+  CheckEffectiveCutoffTime(100);
+  CheckAutomaticFlushRetainUDT(101);
+
+  Close();
+}
+
+TEST_F(ManualFlushSkipRetainUDTTest, RangeDeletionFlushRemovesStaleEntries) {
+  column_family_options_.max_write_buffer_number = 4;
+  Open();
+  // TODO(yuzhangyu): a non 0 full history ts low is needed for this garbage
+  // collection to kick in. This doesn't work well for the very first flush of
+  // the column family. Not a big issue, but would be nice to improve this.
+  ASSERT_OK(db_->IncreaseFullHistoryTsLow(handles_[0], EncodeAsUint64(9)));
+
+  for (int i = 10; i < 100; i++) {
+    ASSERT_OK(Put(0, "foo" + std::to_string(i), EncodeAsUint64(i),
+                  "val" + std::to_string(i)));
+    if (i % 2 == 1) {
+      ASSERT_OK(db_->DeleteRange(WriteOptions(), "foo" + std::to_string(i - 1),
+                                 "foo" + std::to_string(i), EncodeAsUint64(i)));
+    }
+  }
+
+  ASSERT_OK(Flush(0));
+  CheckEffectiveCutoffTime(100);
+  std::string read_ts = EncodeAsUint64(100);
+  std::string min_ts = EncodeAsUint64(0);
+  ReadOptions ropts;
+  Slice read_ts_slice = read_ts;
+  std::string value;
+  ropts.timestamp = &read_ts_slice;
+  {
+    Iterator* iter = db_->NewIterator(ropts);
+    iter->SeekToFirst();
+    int i = 11;
+    while (iter->Valid()) {
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ("foo" + std::to_string(i), iter->key());
+      ASSERT_EQ("val" + std::to_string(i), iter->value());
+      ASSERT_EQ(min_ts, iter->timestamp());
+      iter->Next();
+      i += 2;
+    }
+    ASSERT_OK(iter->status());
+    delete iter;
+  }
+  TablePropertiesCollection tables_properties;
+  ASSERT_OK(db_->GetPropertiesOfAllTables(&tables_properties));
+  ASSERT_EQ(1, tables_properties.size());
+  std::shared_ptr<const TableProperties> table_properties =
+      tables_properties.begin()->second;
+  // 45 point data + 45 range deletions. 45 obsolete point data are garbage
+  // collected.
+  ASSERT_EQ(90, table_properties->num_entries);
+  ASSERT_EQ(45, table_properties->num_deletions);
+  ASSERT_EQ(45, table_properties->num_range_deletions);
+
+  Close();
+}
+
 TEST_F(ManualFlushSkipRetainUDTTest, ManualCompaction) {
   Open();
   ASSERT_OK(db_->IncreaseFullHistoryTsLow(handles_[0], EncodeAsUint64(0)));

db/compaction/compaction_job_test.cc

@@ -50,7 +50,8 @@ void VerifyInitializationOfCompactionJobStats(
   ASSERT_EQ(compaction_job_stats.num_output_records, 0U);
   ASSERT_EQ(compaction_job_stats.num_output_files, 0U);
 
-  ASSERT_EQ(compaction_job_stats.is_manual_compaction, true);
+  ASSERT_TRUE(compaction_job_stats.is_manual_compaction);
+  ASSERT_FALSE(compaction_job_stats.is_remote_compaction);
 
   ASSERT_EQ(compaction_job_stats.total_input_bytes, 0U);
   ASSERT_EQ(compaction_job_stats.total_output_bytes, 0U);

db/compaction/compaction_service_job.cc

@@ -68,8 +68,11 @@ CompactionJob::ProcessKeyValueCompactionWithCompactionService(
       "[%s] [JOB %d] Starting remote compaction (output level: %d): %s",
       compaction->column_family_data()->GetName().c_str(), job_id_,
       compaction_input.output_level, input_files_oss.str().c_str());
-  CompactionServiceJobInfo info(dbname_, db_id_, db_session_id_,
-                                GetCompactionId(sub_compact), thread_pri_);
+  CompactionServiceJobInfo info(
+      dbname_, db_id_, db_session_id_, GetCompactionId(sub_compact),
+      thread_pri_, compaction->compaction_reason(),
+      compaction->is_full_compaction(), compaction->is_manual_compaction(),
+      compaction->bottommost_level());
   CompactionServiceScheduleResponse response =
       db_options_.compaction_service->Schedule(info, compaction_input_binary);
   switch (response.status) {
@@ -333,6 +336,7 @@ Status CompactionServiceCompactionJob::Run() {
   // Build compaction result
   compaction_result_->output_level = compact_->compaction->output_level();
   compaction_result_->output_path = output_path_;
+  compaction_result_->stats.is_remote_compaction = true;
   for (const auto& output_file : sub_compact->GetOutputs()) {
     auto& meta = output_file.meta;
     compaction_result_->output_files.emplace_back(
@@ -527,6 +531,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct CompactionJobStats, is_manual_compaction),
           OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kNone}},
+        {"is_remote_compaction",
+         {offsetof(struct CompactionJobStats, is_remote_compaction),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
         {"total_input_bytes",
          {offsetof(struct CompactionJobStats, total_input_bytes),
           OptionType::kUInt64T, OptionVerificationType::kNormal,

db/compaction/compaction_service_test.cc

@@ -21,8 +21,10 @@ class MyTestCompactionService : public CompactionService {
       : db_path_(std::move(db_path)),
         options_(options),
         statistics_(statistics),
-        start_info_("na", "na", "na", 0, Env::TOTAL),
-        wait_info_("na", "na", "na", 0, Env::TOTAL),
+        start_info_("na", "na", "na", 0, Env::TOTAL, CompactionReason::kUnknown,
+                    false, false, false),
+        wait_info_("na", "na", "na", 0, Env::TOTAL, CompactionReason::kUnknown,
+                   false, false, false),
         listeners_(listeners),
         table_properties_collector_factories_(
             std::move(table_properties_collector_factories)) {}
@@ -97,8 +99,12 @@ class MyTestCompactionService : public CompactionService {
     Status s =
         DB::OpenAndCompact(options, db_path_, db_path_ + "/" + scheduled_job_id,
                            compaction_input, result, options_override);
-    if (is_override_wait_result_) {
-      *result = override_wait_result_;
+    {
+      InstrumentedMutexLock l(&mutex_);
+      if (is_override_wait_result_) {
+        *result = override_wait_result_;
+      }
+      result_ = *result;
     }
     compaction_num_.fetch_add(1);
     if (s.ok()) {
@@ -141,6 +147,10 @@ class MyTestCompactionService : public CompactionService {
 
   void SetCanceled(bool canceled) { canceled_ = canceled; }
 
+  void GetResult(CompactionServiceResult* deserialized) {
+    CompactionServiceResult::Read(result_, deserialized).PermitUncheckedError();
+  }
+
   CompactionServiceJobStatus GetFinalCompactionServiceJobStatus() {
     return final_updated_status_.load();
   }
@@ -162,6 +172,7 @@ class MyTestCompactionService : public CompactionService {
   CompactionServiceJobStatus override_wait_status_ =
       CompactionServiceJobStatus::kFailure;
   bool is_override_wait_result_ = false;
+  std::string result_;
   std::string override_wait_result_;
   std::vector<std::shared_ptr<EventListener>> listeners_;
   std::vector<std::shared_ptr<TablePropertiesCollectorFactory>>
@@ -331,6 +342,14 @@ TEST_F(CompactionServiceTest, BasicCompactions) {
   ReopenWithColumnFamilies({kDefaultColumnFamilyName, "cf_1", "cf_2", "cf_3"},
                            options);
   ASSERT_GT(verify_passed, 0);
+  CompactionServiceResult result;
+  my_cs->GetResult(&result);
+  if (s.IsAborted()) {
+    ASSERT_NOK(result.status);
+  } else {
+    ASSERT_OK(result.status);
+  }
+  ASSERT_TRUE(result.stats.is_remote_compaction);
   Close();
 }
 
@@ -369,6 +388,12 @@ TEST_F(CompactionServiceTest, ManualCompaction) {
   ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
   ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
   VerifyTestData();
+
+  CompactionServiceResult result;
+  my_cs->GetResult(&result);
+  ASSERT_OK(result.status);
+  ASSERT_TRUE(result.stats.is_manual_compaction);
+  ASSERT_TRUE(result.stats.is_remote_compaction);
 }
 
 TEST_F(CompactionServiceTest, CancelCompactionOnRemoteSide) {
@@ -601,11 +626,20 @@ TEST_F(CompactionServiceTest, CompactionInfo) {
                               {file.db_path + "/" + file.name}, 2));
   info = my_cs->GetCompactionInfoForStart();
   ASSERT_EQ(Env::USER, info.priority);
+  ASSERT_EQ(CompactionReason::kManualCompaction, info.compaction_reason);
+  ASSERT_EQ(true, info.is_manual_compaction);
+  ASSERT_EQ(false, info.is_full_compaction);
+  ASSERT_EQ(true, info.bottommost_level);
   info = my_cs->GetCompactionInfoForWait();
   ASSERT_EQ(Env::USER, info.priority);
+  ASSERT_EQ(CompactionReason::kManualCompaction, info.compaction_reason);
+  ASSERT_EQ(true, info.is_manual_compaction);
+  ASSERT_EQ(false, info.is_full_compaction);
+  ASSERT_EQ(true, info.bottommost_level);
 
   // Test priority BOTTOM
   env_->SetBackgroundThreads(1, Env::BOTTOM);
+  // This will set bottommost_level = true but is_full_compaction = false
   options.num_levels = 2;
   ReopenWithCompactionService(&options);
   my_cs =
@@ -628,9 +662,71 @@ TEST_F(CompactionServiceTest, CompactionInfo) {
   }
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
   info = my_cs->GetCompactionInfoForStart();
+  ASSERT_EQ(CompactionReason::kLevelL0FilesNum, info.compaction_reason);
+  ASSERT_EQ(false, info.is_manual_compaction);
+  ASSERT_EQ(false, info.is_full_compaction);
+  ASSERT_EQ(true, info.bottommost_level);
   ASSERT_EQ(Env::BOTTOM, info.priority);
   info = my_cs->GetCompactionInfoForWait();
   ASSERT_EQ(Env::BOTTOM, info.priority);
+  ASSERT_EQ(CompactionReason::kLevelL0FilesNum, info.compaction_reason);
+  ASSERT_EQ(false, info.is_manual_compaction);
+  ASSERT_EQ(false, info.is_full_compaction);
+  ASSERT_EQ(true, info.bottommost_level);
+
+  // Test Non-Bottommost Level
+  options.num_levels = 4;
+  ReopenWithCompactionService(&options);
+  my_cs =
+      static_cast_with_check<MyTestCompactionService>(GetCompactionService());
+
+  for (int i = 0; i < options.level0_file_num_compaction_trigger; i++) {
+    for (int j = 0; j < 10; j++) {
+      int key_id = i * 10 + j;
+      ASSERT_OK(Put(Key(key_id), "value_new_new" + std::to_string(key_id)));
+    }
+    ASSERT_OK(Flush());
+  }
+
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  info = my_cs->GetCompactionInfoForStart();
+  ASSERT_EQ(false, info.is_manual_compaction);
+  ASSERT_EQ(false, info.is_full_compaction);
+  ASSERT_EQ(false, info.bottommost_level);
+  info = my_cs->GetCompactionInfoForWait();
+  ASSERT_EQ(false, info.is_manual_compaction);
+  ASSERT_EQ(false, info.is_full_compaction);
+  ASSERT_EQ(false, info.bottommost_level);
+
+  // Test Full Compaction + Bottommost Level
+  options.num_levels = 6;
+  ReopenWithCompactionService(&options);
+  my_cs =
+      static_cast_with_check<MyTestCompactionService>(GetCompactionService());
+
+  for (int i = 0; i < 20; i++) {
+    for (int j = 0; j < 10; j++) {
+      int key_id = i * 10 + j;
+      ASSERT_OK(Put(Key(key_id), "value_new_new" + std::to_string(key_id)));
+    }
+    ASSERT_OK(Flush());
+  }
+
+  CompactRangeOptions cro;
+  cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+  ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  info = my_cs->GetCompactionInfoForStart();
+  ASSERT_EQ(true, info.is_manual_compaction);
+  ASSERT_EQ(true, info.is_full_compaction);
+  ASSERT_EQ(true, info.bottommost_level);
+  ASSERT_EQ(CompactionReason::kManualCompaction, info.compaction_reason);
+  info = my_cs->GetCompactionInfoForWait();
+  ASSERT_EQ(true, info.is_manual_compaction);
+  ASSERT_EQ(true, info.is_full_compaction);
+  ASSERT_EQ(true, info.bottommost_level);
+  ASSERT_EQ(CompactionReason::kManualCompaction, info.compaction_reason);
 }
 
 TEST_F(CompactionServiceTest, FallbackLocalAuto) {

db/compaction/tiered_compaction_test.cc

@@ -2512,6 +2512,7 @@ TEST_P(IteratorWriteTimeTest, ReadFromMemtables) {
                               start_time + kSecondsPerRecording * (i + 1));
       }
     }
+    ASSERT_EQ(kNumKeys, i);
     ASSERT_OK(iter->status());
   }
 
@@ -2531,12 +2532,13 @@ TEST_P(IteratorWriteTimeTest, ReadFromMemtables) {
       }
     }
     ASSERT_OK(iter->status());
+    ASSERT_EQ(-1, i);
   }
 
   // Reopen the DB and disable the seqno to time recording, data with user
   // specified write time can still get a write time before it's flushed.
   options.preserve_internal_time_seconds = 0;
-  DestroyAndReopen(options);
+  Reopen(options);
   ASSERT_OK(TimedPut(Key(kKeyWithWriteTime), rnd.RandomString(100),
                      kUserSpecifiedWriteTime));
   {
@@ -2613,6 +2615,7 @@ TEST_P(IteratorWriteTimeTest, ReadFromSstFile) {
       }
     }
     ASSERT_OK(iter->status());
+    ASSERT_EQ(kNumKeys, i);
   }
 
   // Backward iteration
@@ -2632,12 +2635,13 @@ TEST_P(IteratorWriteTimeTest, ReadFromSstFile) {
       }
     }
     ASSERT_OK(iter->status());
+    ASSERT_EQ(-1, i);
   }
 
   // Reopen the DB and disable the seqno to time recording. Data retrieved from
   // SST files still have write time available.
   options.preserve_internal_time_seconds = 0;
-  DestroyAndReopen(options);
+  Reopen(options);
 
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(kSecondsPerRecording); });
@@ -2663,6 +2667,7 @@ TEST_P(IteratorWriteTimeTest, ReadFromSstFile) {
                               start_time + kSecondsPerRecording * (i + 1));
       }
     }
+    ASSERT_EQ(kNumKeys, i);
     ASSERT_OK(iter->status());
   }
 
@@ -2686,6 +2691,7 @@ TEST_P(IteratorWriteTimeTest, ReadFromSstFile) {
       VerifyKeyAndWriteTime(iter.get(), Key(i), 0);
     }
     ASSERT_OK(iter->status());
+    ASSERT_EQ(kNumKeys, i);
   }
   Close();
 }

db/db_impl/db_impl_open.cc

@@ -1667,10 +1667,19 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
   Arena arena;
   Status s;
   TableProperties table_properties;
+  const auto* ucmp = cfd->internal_comparator().user_comparator();
+  assert(ucmp);
+  const size_t ts_sz = ucmp->timestamp_size();
+  const bool logical_strip_timestamp =
+      ts_sz > 0 && !cfd->ioptions()->persist_user_defined_timestamps;
   {
     ScopedArenaPtr<InternalIterator> iter(
-        mem->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena,
-                         /*prefix_extractor=*/nullptr));
+        logical_strip_timestamp
+            ? mem->NewTimestampStrippingIterator(
+                  ro, /*seqno_to_time_mapping=*/nullptr, &arena,
+                  /*prefix_extractor=*/nullptr, ts_sz)
+            : mem->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena,
+                               /*prefix_extractor=*/nullptr));
     ROCKS_LOG_DEBUG(immutable_db_options_.info_log,
                     "[%s] [WriteLevel0TableForRecovery]"
                     " Level-0 table #%" PRIu64 ": started",
@@ -1705,11 +1714,14 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
       std::vector<std::unique_ptr<FragmentedRangeTombstoneIterator>>
           range_del_iters;
       auto range_del_iter =
-          // This is called during recovery, where a live memtable is flushed
-          // directly. In this case, no fragmented tombstone list is cached in
-          // this memtable yet.
-          mem->NewRangeTombstoneIterator(ro, kMaxSequenceNumber,
-                                         false /* immutable_memtable */);
+          logical_strip_timestamp
+              ? mem->NewTimestampStrippingRangeTombstoneIterator(
+                    ro, kMaxSequenceNumber, ts_sz)
+              // This is called during recovery, where a live memtable is
+              // flushed directly. In this case, no fragmented tombstone list is
+              // cached in this memtable yet.
+              : mem->NewRangeTombstoneIterator(ro, kMaxSequenceNumber,
+                                               false /* immutable_memtable */);
       if (range_del_iter != nullptr) {
         range_del_iters.emplace_back(range_del_iter);
       }
@@ -1795,9 +1807,7 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
 
     // For UDT in memtable only feature, move up the cutoff timestamp whenever
     // a flush happens.
-    const Comparator* ucmp = cfd->user_comparator();
-    size_t ts_sz = ucmp->timestamp_size();
-    if (ts_sz > 0 && !cfd->ioptions()->persist_user_defined_timestamps) {
+    if (logical_strip_timestamp) {
       Slice mem_newest_udt = mem->GetNewestUDT();
       std::string full_history_ts_low = cfd->GetFullHistoryTsLow();
       if (full_history_ts_low.empty() ||

db/dbformat.cc

@@ -272,11 +272,23 @@ LookupKey::LookupKey(const Slice& _user_key, SequenceNumber s,
 
 void IterKey::EnlargeBuffer(size_t key_size) {
   // If size is smaller than buffer size, continue using current buffer,
-  // or the static allocated one, as default
+  // or the inline one, as default
   assert(key_size > buf_size_);
   // Need to enlarge the buffer.
   ResetBuffer();
   buf_ = new char[key_size];
   buf_size_ = key_size;
 }
+
+void IterKey::EnlargeSecondaryBufferIfNeeded(size_t key_size) {
+  // If size is smaller than buffer size, continue using current buffer,
+  // or the inline one, as default
+  if (key_size <= secondary_buf_size_) {
+    return;
+  }
+  // Need to enlarge the secondary buffer.
+  ResetSecondaryBuffer();
+  secondary_buf_ = new char[key_size];
+  secondary_buf_size_ = key_size;
+}
 }  // namespace ROCKSDB_NAMESPACE

db/dbformat.h

@@ -10,6 +10,7 @@
 #pragma once
 #include <stdio.h>
 
+#include <array>
 #include <memory>
 #include <optional>
 #include <string>
@@ -562,18 +563,28 @@ inline uint64_t GetInternalKeySeqno(const Slice& internal_key) {
 //    allocation for smaller keys.
 // 3. It tracks user key or internal key, and allow conversion between them.
 class IterKey {
+  static constexpr size_t kInlineBufferSize = 39;
+  // This is only used by user-defined timestamps in MemTable only feature,
+  // which only supports uint64_t timestamps.
+  static constexpr char kTsMin[] = "\x00\x00\x00\x00\x00\x00\x00\x00";
+
  public:
   IterKey()
       : buf_(space_),
         key_(buf_),
         key_size_(0),
-        buf_size_(sizeof(space_)),
-        is_user_key_(true) {}
+        buf_size_(kInlineBufferSize),
+        is_user_key_(true),
+        secondary_buf_(space_for_secondary_buf_),
+        secondary_buf_size_(kInlineBufferSize) {}
   // No copying allowed
   IterKey(const IterKey&) = delete;
   void operator=(const IterKey&) = delete;
 
-  ~IterKey() { ResetBuffer(); }
+  ~IterKey() {
+    ResetBuffer();
+    ResetSecondaryBuffer();
+  }
 
   // The bool will be picked up by the next calls to SetKey
   void SetIsUserKey(bool is_user_key) { is_user_key_ = is_user_key; }
@@ -641,13 +652,15 @@ class IterKey {
                                const char* non_shared_data,
                                const size_t non_shared_len,
                                const size_t ts_sz) {
-    std::string kTsMin(ts_sz, static_cast<unsigned char>(0));
-    std::string key_with_ts;
-    std::vector<Slice> key_parts_with_ts;
+    // This function is only used by the UDT in memtable feature, which only
+    // support built in comparators with uint64 timestamps.
+    assert(ts_sz == sizeof(uint64_t));
+    size_t next_key_slice_index = 0;
     if (IsUserKey()) {
-      key_parts_with_ts = {Slice(key_, shared_len),
-                           Slice(non_shared_data, non_shared_len),
-                           Slice(kTsMin)};
+      key_slices_[next_key_slice_index++] = Slice(key_, shared_len);
+      key_slices_[next_key_slice_index++] =
+          Slice(non_shared_data, non_shared_len);
+      key_slices_[next_key_slice_index++] = Slice(kTsMin, ts_sz);
     } else {
       assert(shared_len + non_shared_len >= kNumInternalBytes);
       // Invaraint: shared_user_key_len + shared_internal_bytes_len = shared_len
@@ -664,30 +677,46 @@ class IterKey {
 
       // One Slice among the three Slices will get split into two Slices, plus
       // a timestamp slice.
-      key_parts_with_ts.reserve(5);
       bool ts_added = false;
       // Add slice parts and find the right location to add the min timestamp.
       MaybeAddKeyPartsWithTimestamp(
           key_, shared_user_key_len,
           shared_internal_bytes_len + non_shared_len < kNumInternalBytes,
-          shared_len + non_shared_len - kNumInternalBytes, kTsMin,
-          key_parts_with_ts, &ts_added);
+          shared_len + non_shared_len - kNumInternalBytes, ts_sz,
+          &next_key_slice_index, &ts_added);
       MaybeAddKeyPartsWithTimestamp(
           key_ + user_key_len, shared_internal_bytes_len,
           non_shared_len < kNumInternalBytes,
-          shared_internal_bytes_len + non_shared_len - kNumInternalBytes,
-          kTsMin, key_parts_with_ts, &ts_added);
+          shared_internal_bytes_len + non_shared_len - kNumInternalBytes, ts_sz,
+          &next_key_slice_index, &ts_added);
       MaybeAddKeyPartsWithTimestamp(non_shared_data, non_shared_len,
                                     non_shared_len >= kNumInternalBytes,
-                                    non_shared_len - kNumInternalBytes, kTsMin,
-                                    key_parts_with_ts, &ts_added);
+                                    non_shared_len - kNumInternalBytes, ts_sz,
+                                    &next_key_slice_index, &ts_added);
       assert(ts_added);
     }
+    SetKeyImpl(next_key_slice_index,
+               /* total_bytes= */ shared_len + non_shared_len + ts_sz);
+  }
 
-    Slice new_key(SliceParts(&key_parts_with_ts.front(),
-                             static_cast<int>(key_parts_with_ts.size())),
-                  &key_with_ts);
-    SetKey(new_key);
+  Slice SetKeyWithPaddedMinTimestamp(const Slice& key, size_t ts_sz) {
+    // This function is only used by the UDT in memtable feature, which only
+    // support built in comparators with uint64 timestamps.
+    assert(ts_sz == sizeof(uint64_t));
+    size_t num_key_slices = 0;
+    if (is_user_key_) {
+      key_slices_[0] = key;
+      key_slices_[1] = Slice(kTsMin, ts_sz);
+      num_key_slices = 2;
+    } else {
+      assert(key.size() >= kNumInternalBytes);
+      size_t user_key_size = key.size() - kNumInternalBytes;
+      key_slices_[0] = Slice(key.data(), user_key_size);
+      key_slices_[1] = Slice(kTsMin, ts_sz);
+      key_slices_[2] = Slice(key.data() + user_key_size, kNumInternalBytes);
+      num_key_slices = 3;
+    }
+    return SetKeyImpl(num_key_slices, key.size() + ts_sz);
   }
 
   Slice SetKey(const Slice& key, bool copy = true) {
@@ -718,15 +747,6 @@ class IterKey {
     return Slice(key_, key_n);
   }
 
-  // Copy the key into IterKey own buf_
-  void OwnKey() {
-    assert(IsKeyPinned() == true);
-
-    Reserve(key_size_);
-    memcpy(buf_, key_, key_size_);
-    key_ = buf_;
-  }
-
   // Update the sequence number in the internal key.  Guarantees not to
   // invalidate slices to the key (and the user key).
   void UpdateInternalKey(uint64_t seq, ValueType t, const Slice* ts = nullptr) {
@@ -738,10 +758,15 @@ class IterKey {
              ts->size());
     }
     uint64_t newval = (seq << 8) | t;
-    EncodeFixed64(&buf_[key_size_ - kNumInternalBytes], newval);
+    if (key_ == buf_) {
+      EncodeFixed64(&buf_[key_size_ - kNumInternalBytes], newval);
+    } else {
+      assert(key_ == secondary_buf_);
+      EncodeFixed64(&secondary_buf_[key_size_ - kNumInternalBytes], newval);
+    }
   }
 
-  bool IsKeyPinned() const { return (key_ != buf_); }
+  bool IsKeyPinned() const { return key_ != buf_ && key_ != secondary_buf_; }
 
   // If `ts` is provided, user_key should not contain timestamp,
   // and `ts` is appended after user_key.
@@ -806,8 +831,24 @@ class IterKey {
   const char* key_;
   size_t key_size_;
   size_t buf_size_;
-  char space_[39];  // Avoid allocation for short keys
+  char space_[kInlineBufferSize];  // Avoid allocation for short keys
   bool is_user_key_;
+  // Below variables are only used by user-defined timestamps in MemTable only
+  // feature for iterating keys in an index block or a data block.
+  //
+  // We will alternate between buf_ and secondary_buf_ to hold the key. key_
+  // will be modified in accordance to point to the right one. This is to avoid
+  // an extra copy when we need to copy some shared bytes from previous key
+  // (delta encoding), and we need to pad a min timestamp at the right location.
+  char space_for_secondary_buf_[kInlineBufferSize];  // Avoid allocation for
+                                                     // short keys
+  char* secondary_buf_;
+  size_t secondary_buf_size_;
+  // Use to track the pieces that together make the whole key. We then copy
+  // these pieces in order either into buf_ or secondary_buf_ depending on where
+  // the previous key is held.
+  std::array<Slice, 5> key_slices_;
+  // End of variables used by user-defined timestamps in MemTable only feature.
 
   Slice SetKeyImpl(const Slice& key, bool copy) {
     size_t size = key.size();
@@ -824,18 +865,64 @@ class IterKey {
     return Slice(key_, key_size_);
   }
 
+  Slice SetKeyImpl(size_t num_key_slices, size_t total_bytes) {
+    assert(num_key_slices <= 5);
+    char* buf_start = nullptr;
+    if (key_ == buf_) {
+      // If the previous key is in buf_, we copy key_slices_ in order into
+      // secondary_buf_.
+      EnlargeSecondaryBufferIfNeeded(total_bytes);
+      buf_start = secondary_buf_;
+      key_ = secondary_buf_;
+    } else {
+      // Copy key_slices_ in order into buf_.
+      EnlargeBufferIfNeeded(total_bytes);
+      buf_start = buf_;
+      key_ = buf_;
+    }
+#ifndef NDEBUG
+    size_t actual_total_bytes = 0;
+#endif  // NDEBUG
+    for (size_t i = 0; i < num_key_slices; i++) {
+      size_t key_slice_size = key_slices_[i].size();
+      memcpy(buf_start, key_slices_[i].data(), key_slice_size);
+      buf_start += key_slice_size;
+#ifndef NDEBUG
+      actual_total_bytes += key_slice_size;
+#endif  // NDEBUG
+    }
+#ifndef NDEBUG
+    assert(actual_total_bytes == total_bytes);
+#endif  // NDEBUG
+    key_size_ = total_bytes;
+    return Slice(key_, key_size_);
+  }
+
   void ResetBuffer() {
+    if (key_ == buf_) {
+      key_size_ = 0;
+    }
     if (buf_ != space_) {
       delete[] buf_;
       buf_ = space_;
     }
-    buf_size_ = sizeof(space_);
-    key_size_ = 0;
+    buf_size_ = kInlineBufferSize;
+  }
+
+  void ResetSecondaryBuffer() {
+    if (key_ == secondary_buf_) {
+      key_size_ = 0;
+    }
+    if (secondary_buf_ != space_for_secondary_buf_) {
+      delete[] secondary_buf_;
+      secondary_buf_ = space_for_secondary_buf_;
+    }
+    secondary_buf_size_ = kInlineBufferSize;
   }
 
   // Enlarge the buffer size if needed based on key_size.
-  // By default, static allocated buffer is used. Once there is a key
-  // larger than the static allocated buffer, another buffer is dynamically
+  // By default, inline buffer is used. Once there is a key
+  // larger than the inline buffer, another buffer is dynamically
   // allocated, until a larger key buffer is requested. In that case, we
   // reallocate buffer and delete the old one.
   void EnlargeBufferIfNeeded(size_t key_size) {
@@ -846,23 +933,27 @@ class IterKey {
     }
   }
 
+  void EnlargeSecondaryBufferIfNeeded(size_t key_size);
+
   void EnlargeBuffer(size_t key_size);
 
   void MaybeAddKeyPartsWithTimestamp(const char* slice_data,
                                      const size_t slice_sz, bool add_timestamp,
-                                     const size_t left_sz,
-                                     const std::string& min_timestamp,
-                                     std::vector<Slice>& key_parts,
+                                     const size_t left_sz, const size_t ts_sz,
+                                     size_t* next_key_slice_idx,
                                      bool* ts_added) {
+    assert(next_key_slice_idx);
     if (add_timestamp && !*ts_added) {
       assert(slice_sz >= left_sz);
-      key_parts.emplace_back(slice_data, left_sz);
-      key_parts.emplace_back(min_timestamp);
-      key_parts.emplace_back(slice_data + left_sz, slice_sz - left_sz);
+      key_slices_[(*next_key_slice_idx)++] = Slice(slice_data, left_sz);
+      key_slices_[(*next_key_slice_idx)++] = Slice(kTsMin, ts_sz);
+      key_slices_[(*next_key_slice_idx)++] =
+          Slice(slice_data + left_sz, slice_sz - left_sz);
       *ts_added = true;
     } else {
-      key_parts.emplace_back(slice_data, slice_sz);
+      key_slices_[(*next_key_slice_idx)++] = Slice(slice_data, slice_sz);
     }
+    assert(*next_key_slice_idx <= 5);
   }
 };
 
@@ -936,22 +1027,13 @@ struct RangeTombstone {
   // User-defined timestamp is enabled, `sk` and `ek` should be user key
   // with timestamp, `ts` will replace the timestamps in `sk` and
   // `ek`.
-  // When `logical_strip_timestamp` is true, the timestamps in `sk` and `ek`
-  // will be replaced with min timestamp.
-  RangeTombstone(Slice sk, Slice ek, SequenceNumber sn, Slice ts,
-                 bool logical_strip_timestamp)
-      : seq_(sn) {
+  RangeTombstone(Slice sk, Slice ek, SequenceNumber sn, Slice ts) : seq_(sn) {
     const size_t ts_sz = ts.size();
     assert(ts_sz > 0);
     pinned_start_key_.reserve(sk.size());
     pinned_end_key_.reserve(ek.size());
-    if (logical_strip_timestamp) {
-      AppendUserKeyWithMinTimestamp(&pinned_start_key_, sk, ts_sz);
-      AppendUserKeyWithMinTimestamp(&pinned_end_key_, ek, ts_sz);
-    } else {
-      AppendUserKeyWithDifferentTimestamp(&pinned_start_key_, sk, ts);
-      AppendUserKeyWithDifferentTimestamp(&pinned_end_key_, ek, ts);
-    }
+    AppendUserKeyWithDifferentTimestamp(&pinned_start_key_, sk, ts);
+    AppendUserKeyWithDifferentTimestamp(&pinned_end_key_, ek, ts);
     start_key_ = pinned_start_key_;
     end_key_ = pinned_end_key_;
     ts_ = Slice(pinned_start_key_.data() + sk.size() - ts_sz, ts_sz);

db/flush_job.cc

@@ -858,6 +858,12 @@ Status FlushJob::WriteLevel0Table() {
   meta_.temperature = mutable_cf_options_.default_write_temperature;
   file_options_.temperature = meta_.temperature;
 
+  const auto* ucmp = cfd_->internal_comparator().user_comparator();
+  assert(ucmp);
+  const size_t ts_sz = ucmp->timestamp_size();
+  const bool logical_strip_timestamp =
+      ts_sz > 0 && !cfd_->ioptions()->persist_user_defined_timestamps;
+
   std::vector<BlobFileAddition> blob_file_additions;
 
   {
@@ -893,10 +899,21 @@ Status FlushJob::WriteLevel0Table() {
           db_options_.info_log,
           "[%s] [JOB %d] Flushing memtable with next log file: %" PRIu64 "\n",
           cfd_->GetName().c_str(), job_context_->job_id, m->GetNextLogNumber());
-      memtables.push_back(m->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr,
-                                         &arena, /*prefix_extractor=*/nullptr));
-      auto* range_del_iter = m->NewRangeTombstoneIterator(
-          ro, kMaxSequenceNumber, true /* immutable_memtable */);
+      if (logical_strip_timestamp) {
+        memtables.push_back(m->NewTimestampStrippingIterator(
+            ro, /*seqno_to_time_mapping=*/nullptr, &arena,
+            /*prefix_extractor=*/nullptr, ts_sz));
+      } else {
+        memtables.push_back(
+            m->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena,
+                           /*prefix_extractor=*/nullptr));
+      }
+      auto* range_del_iter =
+          logical_strip_timestamp
+              ? m->NewTimestampStrippingRangeTombstoneIterator(
+                    ro, kMaxSequenceNumber, ts_sz)
+              : m->NewRangeTombstoneIterator(ro, kMaxSequenceNumber,
+                                             true /* immutable_memtable */);
       if (range_del_iter != nullptr) {
         range_del_iters.emplace_back(range_del_iter);
       }

db/flush_job_test.cc

@@ -874,9 +874,15 @@ TEST_P(FlushJobTimestampTest, NoKeyExpired) {
       expected_full_history_ts_low = full_history_ts_low;
     }
     InternalKey smallest(smallest_key, curr_seq_ - 1, ValueType::kTypeValue);
-    InternalKey largest(largest_key, kStartSeq, ValueType::kTypeValue);
-    CheckFileMetaData(cfd, smallest, largest, &fmeta);
-    CheckFullHistoryTsLow(cfd, expected_full_history_ts_low);
+    if (!persist_udt_) {
+      InternalKey largest(largest_key, curr_seq_ - 1, ValueType::kTypeValue);
+      CheckFileMetaData(cfd, smallest, largest, &fmeta);
+      CheckFullHistoryTsLow(cfd, expected_full_history_ts_low);
+    } else {
+      InternalKey largest(largest_key, kStartSeq, ValueType::kTypeValue);
+      CheckFileMetaData(cfd, smallest, largest, &fmeta);
+      CheckFullHistoryTsLow(cfd, expected_full_history_ts_low);
+    }
   }
   job_context.Clean();
   ASSERT_TRUE(to_delete.empty());

db/forward_iterator.cc

@@ -167,6 +167,10 @@ class ForwardLevelIterator : public InternalIterator {
     assert(valid_);
     return file_iter_->value();
   }
+  uint64_t write_unix_time() const override {
+    assert(valid_);
+    return file_iter_->write_unix_time();
+  }
   Status status() const override {
     if (!status_.ok()) {
       return status_;

db/memtable.cc

@@ -613,6 +613,135 @@ InternalIterator* MemTable::NewIterator(
                        seqno_to_time_mapping, arena, prefix_extractor);
 }
 
+// An iterator wrapper that wraps a MemTableIterator and logically strips each
+// key's user-defined timestamp.
+class TimestampStrippingIterator : public InternalIterator {
+ public:
+  TimestampStrippingIterator(
+      MemTableIterator::Kind kind, const MemTable& memtable,
+      const ReadOptions& read_options,
+      UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
+      const SliceTransform* cf_prefix_extractor, size_t ts_sz)
+      : arena_mode_(arena != nullptr), kind_(kind), ts_sz_(ts_sz) {
+    assert(ts_sz_ != 0);
+    void* mem = arena ? arena->AllocateAligned(sizeof(MemTableIterator)) :
+                      operator new(sizeof(MemTableIterator));
+    iter_ = new (mem)
+        MemTableIterator(kind, memtable, read_options, seqno_to_time_mapping,
+                         arena, cf_prefix_extractor);
+  }
+
+  // No copying allowed
+  TimestampStrippingIterator(const TimestampStrippingIterator&) = delete;
+  void operator=(const TimestampStrippingIterator&) = delete;
+
+  ~TimestampStrippingIterator() override {
+    if (arena_mode_) {
+      iter_->~MemTableIterator();
+    } else {
+      delete iter_;
+    }
+  }
+
+  void SetPinnedItersMgr(PinnedIteratorsManager* pinned_iters_mgr) override {
+    iter_->SetPinnedItersMgr(pinned_iters_mgr);
+  }
+
+  bool Valid() const override { return iter_->Valid(); }
+  void Seek(const Slice& k) override {
+    iter_->Seek(k);
+    UpdateKeyAndValueBuffer();
+  }
+  void SeekForPrev(const Slice& k) override {
+    iter_->SeekForPrev(k);
+    UpdateKeyAndValueBuffer();
+  }
+  void SeekToFirst() override {
+    iter_->SeekToFirst();
+    UpdateKeyAndValueBuffer();
+  }
+  void SeekToLast() override {
+    iter_->SeekToLast();
+    UpdateKeyAndValueBuffer();
+  }
+  void Next() override {
+    iter_->Next();
+    UpdateKeyAndValueBuffer();
+  }
+  bool NextAndGetResult(IterateResult* result) override {
+    iter_->Next();
+    UpdateKeyAndValueBuffer();
+    bool is_valid = Valid();
+    if (is_valid) {
+      result->key = key();
+      result->bound_check_result = IterBoundCheck::kUnknown;
+      result->value_prepared = true;
+    }
+    return is_valid;
+  }
+  void Prev() override {
+    iter_->Prev();
+    UpdateKeyAndValueBuffer();
+  }
+  Slice key() const override {
+    assert(Valid());
+    return key_buf_;
+  }
+
+  uint64_t write_unix_time() const override { return iter_->write_unix_time(); }
+  Slice value() const override {
+    if (kind_ == MemTableIterator::Kind::kRangeDelEntries) {
+      return value_buf_;
+    }
+    return iter_->value();
+  }
+  Status status() const override { return iter_->status(); }
+  bool IsKeyPinned() const override {
+    // Key is only in a buffer that is updated in each iteration.
+    return false;
+  }
+  bool IsValuePinned() const override {
+    if (kind_ == MemTableIterator::Kind::kRangeDelEntries) {
+      return false;
+    }
+    return iter_->IsValuePinned();
+  }
+
+ private:
+  void UpdateKeyAndValueBuffer() {
+    key_buf_.clear();
+    if (kind_ == MemTableIterator::Kind::kRangeDelEntries) {
+      value_buf_.clear();
+    }
+    if (!Valid()) {
+      return;
+    }
+    Slice original_key = iter_->key();
+    ReplaceInternalKeyWithMinTimestamp(&key_buf_, original_key, ts_sz_);
+    if (kind_ == MemTableIterator::Kind::kRangeDelEntries) {
+      Slice original_value = iter_->value();
+      AppendUserKeyWithMinTimestamp(&value_buf_, original_value, ts_sz_);
+    }
+  }
+  bool arena_mode_;
+  MemTableIterator::Kind kind_;
+  size_t ts_sz_;
+  MemTableIterator* iter_;
+  std::string key_buf_;
+  std::string value_buf_;
+};
+
+InternalIterator* MemTable::NewTimestampStrippingIterator(
+    const ReadOptions& read_options,
+    UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
+    const SliceTransform* prefix_extractor, size_t ts_sz) {
+  assert(arena != nullptr);
+  auto mem = arena->AllocateAligned(sizeof(TimestampStrippingIterator));
+  return new (mem) TimestampStrippingIterator(
+      MemTableIterator::kPointEntries, *this, read_options,
+      seqno_to_time_mapping, arena, prefix_extractor, ts_sz);
+}
+
 FragmentedRangeTombstoneIterator* MemTable::NewRangeTombstoneIterator(
     const ReadOptions& read_options, SequenceNumber read_seq,
     bool immutable_memtable) {
@@ -624,6 +753,30 @@ FragmentedRangeTombstoneIterator* MemTable::NewRangeTombstoneIterator(
                                            immutable_memtable);
 }
 
+FragmentedRangeTombstoneIterator*
+MemTable::NewTimestampStrippingRangeTombstoneIterator(
+    const ReadOptions& read_options, SequenceNumber read_seq, size_t ts_sz) {
+  if (read_options.ignore_range_deletions ||
+      is_range_del_table_empty_.load(std::memory_order_relaxed)) {
+    return nullptr;
+  }
+  if (!timestamp_stripping_fragmented_range_tombstone_list_) {
+    // TODO: plumb Env::IOActivity, Env::IOPriority
+    auto* unfragmented_iter = new TimestampStrippingIterator(
+        MemTableIterator::kRangeDelEntries, *this, ReadOptions(),
+        /*seqno_to_time_mapping*/ nullptr, /* arena */ nullptr,
+        /* prefix_extractor */ nullptr, ts_sz);
+
+    timestamp_stripping_fragmented_range_tombstone_list_ =
+        std::make_unique<FragmentedRangeTombstoneList>(
+            std::unique_ptr<InternalIterator>(unfragmented_iter),
+            comparator_.comparator);
+  }
+  return new FragmentedRangeTombstoneIterator(
+      timestamp_stripping_fragmented_range_tombstone_list_.get(),
+      comparator_.comparator, read_seq, read_options.timestamp);
+}
+
 FragmentedRangeTombstoneIterator* MemTable::NewRangeTombstoneIteratorInternal(
     const ReadOptions& read_options, SequenceNumber read_seq,
     bool immutable_memtable) {

db/memtable.h

@@ -213,6 +213,14 @@ class MemTable {
       UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
       const SliceTransform* prefix_extractor);
 
+  // Returns an iterator that wraps a MemTableIterator and logically strips the
+  // user-defined timestamp of each key. This API is only used by flush when
+  // user-defined timestamps in MemTable only feature is enabled.
+  InternalIterator* NewTimestampStrippingIterator(
+      const ReadOptions& read_options,
+      UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
+      const SliceTransform* prefix_extractor, size_t ts_sz);
+
   // Returns an iterator that yields the range tombstones of the memtable.
   // The caller must ensure that the underlying MemTable remains live
   // while the returned iterator is live.
@@ -227,6 +235,13 @@ class MemTable {
       const ReadOptions& read_options, SequenceNumber read_seq,
       bool immutable_memtable);
 
+  // Returns an iterator that yields the range tombstones of the memtable and
+  // logically strips the user-defined timestamp of each key (including start
+  // key, and end key). This API is only used by flush when user-defined
+  // timestamps in MemTable only feature is enabled.
+  FragmentedRangeTombstoneIterator* NewTimestampStrippingRangeTombstoneIterator(
+      const ReadOptions& read_options, SequenceNumber read_seq, size_t ts_sz);
+
   Status VerifyEncodedEntry(Slice encoded,
                             const ProtectionInfoKVOS64& kv_prot_info);
 
@@ -704,6 +719,12 @@ class MemTable {
   std::unique_ptr<FragmentedRangeTombstoneList>
       fragmented_range_tombstone_list_;
 
+  // The fragmented range tombstone of this memtable with all keys' user-defined
+  // timestamps logically stripped. This is constructed and used by flush when
+  // user-defined timestamps in memtable only feature is enabled.
+  std::unique_ptr<FragmentedRangeTombstoneList>
+      timestamp_stripping_fragmented_range_tombstone_list_;
+
   // makes sure there is a single range tombstone writer to invalidate cache
   std::mutex range_del_mutex_;
   CoreLocalArray<std::shared_ptr<FragmentedRangeTombstoneListCache>>

db/range_tombstone_fragmenter.h

@@ -197,11 +197,10 @@ class FragmentedRangeTombstoneIterator : public InternalIterator {
     pinned_seq_pos_ = tombstones_->seq_end();
   }
 
-  RangeTombstone Tombstone(bool logical_strip_timestamp = false) const {
+  RangeTombstone Tombstone() const {
     assert(Valid());
     if (icmp_->user_comparator()->timestamp_size()) {
-      return RangeTombstone(start_key(), end_key(), seq(), timestamp(),
-                            logical_strip_timestamp);
+      return RangeTombstone(start_key(), end_key(), seq(), timestamp());
     }
     return RangeTombstone(start_key(), end_key(), seq());
   }

db/version_set.cc

@@ -1050,6 +1050,11 @@ class LevelIterator final : public InternalIterator {
     return file_iter_.value();
   }
 
+  uint64_t write_unix_time() const override {
+    assert(Valid());
+    return file_iter_.write_unix_time();
+  }
+
   Status status() const override {
     return file_iter_.iter() ? file_iter_.status() : Status::OK();
   }

include/rocksdb/compaction_job_stats.h

@@ -19,80 +19,82 @@ struct CompactionJobStats {
   void Add(const CompactionJobStats& stats);
 
   // the elapsed time of this compaction in microseconds.
-  uint64_t elapsed_micros;
+  uint64_t elapsed_micros = 0;
 
   // the elapsed CPU time of this compaction in microseconds.
-  uint64_t cpu_micros;
+  uint64_t cpu_micros = 0;
 
   // Used internally indicating whether a subcompaction's
   // `num_input_records` is accurate.
-  bool has_num_input_records;
+  bool has_num_input_records = false;
   // the number of compaction input records.
-  uint64_t num_input_records;
+  uint64_t num_input_records = 0;
   // the number of blobs read from blob files
-  uint64_t num_blobs_read;
+  uint64_t num_blobs_read = 0;
   // the number of compaction input files (table files)
-  size_t num_input_files;
+  size_t num_input_files = 0;
   // the number of compaction input files at the output level (table files)
-  size_t num_input_files_at_output_level;
+  size_t num_input_files_at_output_level = 0;
 
   // the number of compaction output records.
-  uint64_t num_output_records;
+  uint64_t num_output_records = 0;
   // the number of compaction output files (table files)
-  size_t num_output_files;
+  size_t num_output_files = 0;
   // the number of compaction output files (blob files)
-  size_t num_output_files_blob;
+  size_t num_output_files_blob = 0;
 
   // true if the compaction is a full compaction (all live SST files input)
-  bool is_full_compaction;
+  bool is_full_compaction = false;
   // true if the compaction is a manual compaction
-  bool is_manual_compaction;
+  bool is_manual_compaction = false;
+  // true if the compaction ran in a remote worker
+  bool is_remote_compaction = false;
 
   // the total size of table files in the compaction input
-  uint64_t total_input_bytes;
+  uint64_t total_input_bytes = 0;
   // the total size of blobs read from blob files
-  uint64_t total_blob_bytes_read;
+  uint64_t total_blob_bytes_read = 0;
   // the total size of table files in the compaction output
-  uint64_t total_output_bytes;
+  uint64_t total_output_bytes = 0;
   // the total size of blob files in the compaction output
-  uint64_t total_output_bytes_blob;
+  uint64_t total_output_bytes_blob = 0;
 
   // number of records being replaced by newer record associated with same key.
   // this could be a new value or a deletion entry for that key so this field
   // sums up all updated and deleted keys
-  uint64_t num_records_replaced;
+  uint64_t num_records_replaced = 0;
 
   // the sum of the uncompressed input keys in bytes.
-  uint64_t total_input_raw_key_bytes;
+  uint64_t total_input_raw_key_bytes = 0;
   // the sum of the uncompressed input values in bytes.
-  uint64_t total_input_raw_value_bytes;
+  uint64_t total_input_raw_value_bytes = 0;
 
   // the number of deletion entries before compaction. Deletion entries
   // can disappear after compaction because they expired
-  uint64_t num_input_deletion_records;
+  uint64_t num_input_deletion_records = 0;
   // number of deletion records that were found obsolete and discarded
   // because it is not possible to delete any more keys with this entry
   // (i.e. all possible deletions resulting from it have been completed)
-  uint64_t num_expired_deletion_records;
+  uint64_t num_expired_deletion_records = 0;
 
   // number of corrupt keys (ParseInternalKey returned false when applied to
   // the key) encountered and written out.
-  uint64_t num_corrupt_keys;
+  uint64_t num_corrupt_keys = 0;
 
   // Following counters are only populated if
   // options.report_bg_io_stats = true;
 
   // Time spent on file's Append() call.
-  uint64_t file_write_nanos;
+  uint64_t file_write_nanos = 0;
 
   // Time spent on sync file range.
-  uint64_t file_range_sync_nanos;
+  uint64_t file_range_sync_nanos = 0;
 
   // Time spent on file fsync.
-  uint64_t file_fsync_nanos;
+  uint64_t file_fsync_nanos = 0;
 
   // Time spent on preparing file write (fallocate, etc)
-  uint64_t file_prepare_write_nanos;
+  uint64_t file_prepare_write_nanos = 0;
 
   // 0-terminated strings storing the first 8 bytes of the smallest and
   // largest key in the output.
@@ -102,10 +104,10 @@ struct CompactionJobStats {
   std::string largest_output_key_prefix;
 
   // number of single-deletes which do not meet a put
-  uint64_t num_single_del_fallthru;
+  uint64_t num_single_del_fallthru = 0;
 
   // number of single-deletes which meet something other than a put
-  uint64_t num_single_del_mismatch;
+  uint64_t num_single_del_mismatch = 0;
 
   // TODO: Add output_to_penultimate_level output information
 };

include/rocksdb/options.h

@@ -466,14 +466,27 @@ struct CompactionServiceJobInfo {
 
   Env::Priority priority;
 
+  // Additional Compaction Details that can be useful in the CompactionService
+  CompactionReason compaction_reason;
+  bool is_full_compaction;
+  bool is_manual_compaction;
+  bool bottommost_level;
+
   CompactionServiceJobInfo(std::string db_name_, std::string db_id_,
                            std::string db_session_id_, uint64_t job_id_,
-                           Env::Priority priority_)
+                           Env::Priority priority_,
+                           CompactionReason compaction_reason_,
+                           bool is_full_compaction_, bool is_manual_compaction_,
+                           bool bottommost_level_)
       : db_name(std::move(db_name_)),
         db_id(std::move(db_id_)),
         db_session_id(std::move(db_session_id_)),
         job_id(job_id_),
-        priority(priority_) {}
+        priority(priority_),
+        compaction_reason(compaction_reason_),
+        is_full_compaction(is_full_compaction_),
+        is_manual_compaction(is_manual_compaction_),
+        bottommost_level(bottommost_level_) {}
 };
 
 struct CompactionServiceScheduleResponse {

include/rocksdb/utilities/ldb_cmd.h

@@ -74,6 +74,7 @@ class LDBCommand {
   static const std::string ARG_DECODE_BLOB_INDEX;
   static const std::string ARG_DUMP_UNCOMPRESSED_BLOBS;
   static const std::string ARG_READ_TIMESTAMP;
+  static const std::string ARG_GET_WRITE_UNIX_TIME;
 
   struct ParsedParams {
     std::string cmd;

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 9
 #define ROCKSDB_MINOR 7
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 2
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

options/options_helper.cc

@@ -55,7 +55,13 @@ Status ValidateOptions(const DBOptions& db_opts,
 DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
                          const MutableDBOptions& mutable_db_options) {
   DBOptions options;
+  BuildDBOptions(immutable_db_options, mutable_db_options, options);
+  return options;
+}
 
+void BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
+                    const MutableDBOptions& mutable_db_options,
+                    DBOptions& options) {
   options.create_if_missing = immutable_db_options.create_if_missing;
   options.create_missing_column_families =
       immutable_db_options.create_missing_column_families;
@@ -88,9 +94,6 @@ DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.max_background_jobs = mutable_db_options.max_background_jobs;
   options.max_background_compactions =
       mutable_db_options.max_background_compactions;
-  options.bytes_per_sync = mutable_db_options.bytes_per_sync;
-  options.wal_bytes_per_sync = mutable_db_options.wal_bytes_per_sync;
-  options.strict_bytes_per_sync = mutable_db_options.strict_bytes_per_sync;
   options.max_subcompactions = mutable_db_options.max_subcompactions;
   options.max_background_flushes = mutable_db_options.max_background_flushes;
   options.max_log_file_size = immutable_db_options.max_log_file_size;
@@ -127,6 +130,9 @@ DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.writable_file_max_buffer_size =
       mutable_db_options.writable_file_max_buffer_size;
   options.use_adaptive_mutex = immutable_db_options.use_adaptive_mutex;
+  options.bytes_per_sync = mutable_db_options.bytes_per_sync;
+  options.wal_bytes_per_sync = mutable_db_options.wal_bytes_per_sync;
+  options.strict_bytes_per_sync = mutable_db_options.strict_bytes_per_sync;
   options.listeners = immutable_db_options.listeners;
   options.enable_thread_tracking = immutable_db_options.enable_thread_tracking;
   options.delayed_write_rate = mutable_db_options.delayed_write_rate;
@@ -161,9 +167,15 @@ DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.two_write_queues = immutable_db_options.two_write_queues;
   options.manual_wal_flush = immutable_db_options.manual_wal_flush;
   options.wal_compression = immutable_db_options.wal_compression;
+  options.background_close_inactive_wals =
+      immutable_db_options.background_close_inactive_wals;
   options.atomic_flush = immutable_db_options.atomic_flush;
   options.avoid_unnecessary_blocking_io =
       immutable_db_options.avoid_unnecessary_blocking_io;
+  options.write_dbid_to_manifest = immutable_db_options.write_dbid_to_manifest;
+  options.write_identity_file = immutable_db_options.write_identity_file;
+  options.prefix_seek_opt_in_only =
+      immutable_db_options.prefix_seek_opt_in_only;
   options.log_readahead_size = immutable_db_options.log_readahead_size;
   options.file_checksum_gen_factory =
       immutable_db_options.file_checksum_gen_factory;
@@ -189,7 +201,6 @@ DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.metadata_write_temperature =
       immutable_db_options.metadata_write_temperature;
   options.wal_write_temperature = immutable_db_options.wal_write_temperature;
-  return options;
 }
 
 ColumnFamilyOptions BuildColumnFamilyOptions(

options/options_helper.h

@@ -44,6 +44,10 @@ Status ValidateOptions(const DBOptions& db_opts,
 
 DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
                          const MutableDBOptions& mutable_db_options);
+// Overwrites `options`
+void BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
+                    const MutableDBOptions& mutable_db_options,
+                    DBOptions& options);
 
 ColumnFamilyOptions BuildColumnFamilyOptions(
     const ColumnFamilyOptions& ioptions,

options/options_settable_test.cc

@@ -271,6 +271,12 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
   ASSERT_GT(unset_bytes_base, 0);
   options->~DBOptions();
 
+  // Now also check that BuildDBOptions populates everything
+  FillWithSpecialChar(options_ptr, sizeof(DBOptions), kDBOptionsExcluded);
+  BuildDBOptions({}, {}, *options);
+  ASSERT_EQ(unset_bytes_base,
+            NumUnsetBytes(options_ptr, sizeof(DBOptions), kDBOptionsExcluded));
+
   options = new (options_ptr) DBOptions();
   FillWithSpecialChar(options_ptr, sizeof(DBOptions), kDBOptionsExcluded);
 
@@ -372,7 +378,11 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
                              "follower_catchup_retry_count=456;"
                              "follower_catchup_retry_wait_ms=789;"
                              "metadata_write_temperature=kCold;"
-                             "wal_write_temperature=kHot;",
+                             "wal_write_temperature=kHot;"
+                             "background_close_inactive_wals=true;"
+                             "write_dbid_to_manifest=true;"
+                             "write_identity_file=true;"
+                             "prefix_seek_opt_in_only=true;",
                              new_options));
 
   ASSERT_EQ(unset_bytes_base, NumUnsetBytes(new_options_ptr, sizeof(DBOptions),

table/block_based/block.h

@@ -575,13 +575,7 @@ class BlockIter : public InternalIteratorBase<TValue> {
 
   void UpdateRawKeyAndMaybePadMinTimestamp(const Slice& key) {
     if (pad_min_timestamp_) {
-      std::string buf;
-      if (raw_key_.IsUserKey()) {
-        AppendKeyWithMinTimestamp(&buf, key, ts_sz_);
-      } else {
-        PadInternalKeyWithMinTimestamp(&buf, key, ts_sz_);
-      }
-      raw_key_.SetKey(buf, true /* copy */);
+      raw_key_.SetKeyWithPaddedMinTimestamp(key, ts_sz_);
     } else {
       raw_key_.SetKey(key, false /* copy */);
     }

table/iterator.cc

@@ -74,6 +74,10 @@ class EmptyInternalIterator : public InternalIteratorBase<TValue> {
     assert(false);
     return TValue();
   }
+  uint64_t write_unix_time() const override {
+    assert(false);
+    return std::numeric_limits<uint64_t>::max();
+  }
   Status status() const override { return status_; }
 
  private:

tools/ldb_cmd.cc

@@ -27,6 +27,7 @@
 #include "db/write_batch_internal.h"
 #include "file/filename.h"
 #include "rocksdb/cache.h"
+#include "rocksdb/comparator.h"
 #include "rocksdb/experimental.h"
 #include "rocksdb/file_checksum.h"
 #include "rocksdb/filter_policy.h"
@@ -110,6 +111,7 @@ const std::string LDBCommand::ARG_DECODE_BLOB_INDEX = "decode_blob_index";
 const std::string LDBCommand::ARG_DUMP_UNCOMPRESSED_BLOBS =
     "dump_uncompressed_blobs";
 const std::string LDBCommand::ARG_READ_TIMESTAMP = "read_timestamp";
+const std::string LDBCommand::ARG_GET_WRITE_UNIX_TIME = "get_write_unix_time";
 
 const char* LDBCommand::DELIM = " ==> ";
 
@@ -3622,11 +3624,12 @@ void BatchPutCommand::OverrideBaseOptions() {
 ScanCommand::ScanCommand(const std::vector<std::string>& /*params*/,
                          const std::map<std::string, std::string>& options,
                          const std::vector<std::string>& flags)
-    : LDBCommand(options, flags, true,
-                 BuildCmdLineOptions(
-                     {ARG_TTL, ARG_NO_VALUE, ARG_HEX, ARG_KEY_HEX, ARG_TO,
-                      ARG_VALUE_HEX, ARG_FROM, ARG_TIMESTAMP, ARG_MAX_KEYS,
-                      ARG_TTL_START, ARG_TTL_END, ARG_READ_TIMESTAMP})),
+    : LDBCommand(
+          options, flags, true,
+          BuildCmdLineOptions({ARG_TTL, ARG_NO_VALUE, ARG_HEX, ARG_KEY_HEX,
+                               ARG_TO, ARG_VALUE_HEX, ARG_FROM, ARG_TIMESTAMP,
+                               ARG_MAX_KEYS, ARG_TTL_START, ARG_TTL_END,
+                               ARG_READ_TIMESTAMP, ARG_GET_WRITE_UNIX_TIME})),
       start_key_specified_(false),
       end_key_specified_(false),
       max_keys_scanned_(-1),
@@ -3670,6 +3673,7 @@ ScanCommand::ScanCommand(const std::vector<std::string>& /*params*/,
           ARG_MAX_KEYS + " has a value out-of-range");
     }
   }
+  get_write_unix_time_ = IsFlagPresent(flags_, ARG_GET_WRITE_UNIX_TIME);
 }
 
 void ScanCommand::Help(std::string& ret) {
@@ -3683,6 +3687,7 @@ void ScanCommand::Help(std::string& ret) {
   ret.append(" [--" + ARG_TTL_END + "=<N>:- is exclusive]");
   ret.append(" [--" + ARG_NO_VALUE + "]");
   ret.append(" [--" + ARG_READ_TIMESTAMP + "=<uint64_ts>] ");
+  ret.append(" [--" + ARG_GET_WRITE_UNIX_TIME + "]");
   ret.append("\n");
 }
 
@@ -3765,6 +3770,22 @@ void ScanCommand::DoCommand() {
       fprintf(stdout, "%s\n", str.c_str());
     }
 
+    if (get_write_unix_time_) {
+      std::string write_unix_time;
+      uint64_t write_time_int = std::numeric_limits<uint64_t>::max();
+      Status s =
+          it->GetProperty("rocksdb.iterator.write-time", &write_unix_time);
+      if (s.ok()) {
+        s = DecodeU64Ts(write_unix_time, &write_time_int);
+      }
+      if (!s.ok()) {
+        fprintf(stdout, "  Failed to get write unix time: %s\n",
+                s.ToString().c_str());
+      } else {
+        fprintf(stdout, "  write unix time: %s\n",
+                std::to_string(write_time_int).c_str());
+      }
+    }
     num_keys_scanned++;
     if (max_keys_scanned_ >= 0 && num_keys_scanned >= max_keys_scanned_) {
       break;

tools/ldb_cmd_impl.h

@@ -511,6 +511,7 @@ class ScanCommand : public LDBCommand {
   bool end_key_specified_;
   int max_keys_scanned_;
   bool no_value_;
+  bool get_write_unix_time_;
 };
 
 class DeleteCommand : public LDBCommand {

unreleased_history/behavior_changes/blob_garbage_collection_force_threshold.md

@@ -1 +0,0 @@
-Changed the semantics of the BlobDB configuration option `blob_garbage_collection_force_threshold` to define a threshold for the overall garbage ratio of all blob files currently eligible for garbage collection (according to `blob_garbage_collection_age_cutoff`). This can provide better control over space amplification at the cost of slightly higher write amplification.

unreleased_history/behavior_changes/db_id.md

@@ -1 +0,0 @@
-* Set `write_dbid_to_manifest=true` by default. This means DB ID will now be preserved through backups, checkpoints, etc. by default. Also add `write_identity_file` option which can be set to false for anticipated future behavior.

unreleased_history/behavior_changes/fifo-temp-compaction.md

@@ -1 +0,0 @@
-* In FIFO compaction, compactions for changing file temperature (configured by option `file_temperature_age_thresholds`) will compact one file at a time, instead of merging multiple eligible file together (#13018). 

unreleased_history/behavior_changes/ingest-live-file-with-move.md

@@ -1 +0,0 @@
-* Support ingesting db generated files using hard link, i.e. IngestExternalFileOptions::move_files/link_files and IngestExternalFileOptions::allow_db_generated_files.

unreleased_history/behavior_changes/link-file-ingest.md

@@ -1 +0,0 @@
-* Add a new file ingestion option `IngestExternalFileOptions::link_files` to hard link input files and preserve original files links after ingestion.

unreleased_history/behavior_changes/sst_file_manager_untrack_close.md

@@ -1,2 +0,0 @@
-DB::Close now untracks files in SstFileManager, making avaialble any space used
-by them. Prior to this change they would be orphaned until the DB is re-opened.

unreleased_history/bug_fixes/bug-refit-level.md

@@ -1 +0,0 @@
-* Fix a bug in CompactRange() where result files may not be compacted in any future compaction. This can only happen when users configure CompactRangeOptions::change_level to true and the change level step of manual compaction fails (#13009).

unreleased_history/bug_fixes/memtable_prefix.md

@@ -1 +0,0 @@
-* Fix handling of dynamic change of `prefix_extractor` with memtable prefix filter. Previously, prefix seek could mix different prefix interpretations between memtable and SST files. Now the latest `prefix_extractor` at the time of iterator creation or refresh is respected.

unreleased_history/bug_fixes/wal-error-recovery.md

@@ -1 +0,0 @@
-* Fix a bug with manual_wal_flush and auto error recovery from WAL failure that may cause CFs to be inconsistent (#12995). The fix will set potential WAL write failure as fatal error when manual_wal_flush is true, and disables auto error recovery from these errors.   

unreleased_history/new_features/customizable_cache.md

@@ -1 +0,0 @@
-Make Cache a customizable class that can be instantiated by the object registry.

unreleased_history/new_features/prefix_seek_opt_in_only.md

@@ -1 +0,0 @@
-* Add new option `prefix_seek_opt_in_only` that makes iterators generally safer when you might set a `prefix_extractor`. When `prefix_seek_opt_in_only=true`, which is expected to be the future default, prefix seek is only used when `prefix_same_as_start` or `auto_prefix_mode` are set. Also, `prefix_same_as_start` and `auto_prefix_mode` now allow prefix filtering even with `total_order_seek=true`.

unreleased_history/new_features/tp_largest_seqno.md

@@ -1 +0,0 @@
-* Add a new table property "rocksdb.key.largest.seqno" which records the largest sequence number of all keys in file. It is verified to be zero during SST file ingestion.

util/compaction_job_stats_impl.cc

@@ -24,6 +24,7 @@ void CompactionJobStats::Reset() {
 
   is_full_compaction = false;
   is_manual_compaction = false;
+  is_remote_compaction = false;
 
   total_input_bytes = 0;
   total_blob_bytes_read = 0;