Update HISTORY for 9.7.2

Summary:
The write unix time from non L0 files are not surfaced properly because the level's wrapper iterator doesn't have a `write_unix_time` implementation that delegates to the corresponding file. The unit test didn't catch this because it incorrectly destroy the old db and reopen to check write time, instead of just reopen and check. This fix also include a change to support ldb's scan command to get write time for easier debugging.

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

Test Plan: Updated unit tests

Reviewed By: pdillinger

Differential Revision: D64015107

Pulled By: jowlyzhang

fbshipit-source-id: 244474f78a034f80c9235eea2aa8a0f4e54dff59

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/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/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/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/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;