Compact one file at a time for FIFO temperature change compactions (#13018)

Summary:
Per customer request, we should not merge multiple SST files together during temperature change compaction, since this can cause FIFO TTL compactions to be delayed. This PR changes the compaction picking logic to pick one file at a time.

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

Test Plan: * updated some existing unit tests to test this new behavior.

Reviewed By: jowlyzhang

Differential Revision: D62883292

Pulled By: cbi42

fbshipit-source-id: 6a9fc8c296b5d9b17168ef6645f25153241c8b93

db/compaction/compaction_picker_fifo.cc

@@ -294,7 +294,7 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
 Compaction* FIFOCompactionPicker::PickTemperatureChangeCompaction(
     const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
     const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-    LogBuffer* log_buffer) {
+    LogBuffer* log_buffer) const {
   const std::vector<FileTemperatureAge>& ages =
       mutable_cf_options.compaction_options_fifo
           .file_temperature_age_thresholds;
@@ -344,12 +344,10 @@ Compaction* FIFOCompactionPicker::PickTemperatureChangeCompaction(
   Temperature compaction_target_temp = Temperature::kLastTemperature;
   if (current_time > min_age) {
     uint64_t create_time_threshold = current_time - min_age;
-    uint64_t compaction_size = 0;
     // We will ideally identify a file qualifying for temperature change by
     // knowing the timestamp for the youngest entry in the file. However, right
     // now we don't have the information. We infer it by looking at timestamp of
     // the previous file's (which is just younger) oldest entry's timestamp.
-    Temperature cur_target_temp;
     // avoid index underflow
     assert(level_files.size() >= 1);
     for (size_t index = level_files.size() - 1; index >= 1; --index) {
@@ -374,7 +372,7 @@ Compaction* FIFOCompactionPicker::PickTemperatureChangeCompaction(
         // cur_file is too fresh
         break;
       }
-      cur_target_temp = ages[0].temperature;
+      Temperature cur_target_temp = ages[0].temperature;
       for (size_t i = 1; i < ages.size(); ++i) {
         if (current_time >= ages[i].age &&
             oldest_ancestor_time <= current_time - ages[i].age) {
@@ -382,35 +380,20 @@ Compaction* FIFOCompactionPicker::PickTemperatureChangeCompaction(
         }
       }
       if (cur_file->temperature == cur_target_temp) {
-        if (inputs[0].empty()) {
-          continue;
-        } else {
-          break;
-        }
+        continue;
       }
 
       // cur_file needs to change temperature
-      if (compaction_target_temp == Temperature::kLastTemperature) {
-        assert(inputs[0].empty());
-        compaction_target_temp = cur_target_temp;
-      } else if (cur_target_temp != compaction_target_temp) {
-        assert(!inputs[0].empty());
-        break;
-      }
-      if (inputs[0].empty() || compaction_size + cur_file->fd.GetFileSize() <=
-                                   mutable_cf_options.max_compaction_bytes) {
-        inputs[0].files.push_back(cur_file);
-        compaction_size += cur_file->fd.GetFileSize();
-        ROCKS_LOG_BUFFER(
-            log_buffer,
-            "[%s] FIFO compaction: picking file %" PRIu64
-            " with next file's oldest time %" PRIu64 " for temperature %s.",
-            cf_name.c_str(), cur_file->fd.GetNumber(), oldest_ancestor_time,
-            temperature_to_string[cur_target_temp].c_str());
-      }
-      if (compaction_size > mutable_cf_options.max_compaction_bytes) {
-        break;
-      }
+      assert(compaction_target_temp == Temperature::kLastTemperature);
+      compaction_target_temp = cur_target_temp;
+      inputs[0].files.push_back(cur_file);
+      ROCKS_LOG_BUFFER(
+          log_buffer,
+          "[%s] FIFO compaction: picking file %" PRIu64
+          " with next file's oldest time %" PRIu64 " for temperature %s.",
+          cf_name.c_str(), cur_file->fd.GetNumber(), oldest_ancestor_time,
+          temperature_to_string[cur_target_temp].c_str());
+      break;
     }
   }
 
@@ -418,7 +401,9 @@ Compaction* FIFOCompactionPicker::PickTemperatureChangeCompaction(
     return nullptr;
   }
   assert(compaction_target_temp != Temperature::kLastTemperature);
-
+  // Only compact one file at a time.
+  assert(inputs.size() == 1);
+  assert(inputs[0].size() == 1);
   Compaction* c = new Compaction(
       vstorage, ioptions_, mutable_cf_options, mutable_db_options,
       std::move(inputs), 0, 0 /* output file size limit */,

db/compaction/compaction_picker_fifo.h

@@ -53,9 +53,10 @@ class FIFOCompactionPicker : public CompactionPicker {
                                  VersionStorageInfo* version,
                                  LogBuffer* log_buffer);
 
+  // Will pick one file to compact at a time, starting from the oldest file.
   Compaction* PickTemperatureChangeCompaction(
       const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
       const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
-      LogBuffer* log_buffer);
+      LogBuffer* log_buffer) const;
 };
 }  // namespace ROCKSDB_NAMESPACE

db/compaction/compaction_picker_test.cc

@@ -1119,48 +1119,6 @@ TEST_F(CompactionPickerTest, FIFOToCold1) {
   ASSERT_EQ(3U, compaction->input(0, 0)->fd.GetNumber());
 }
 
-TEST_F(CompactionPickerTest, FIFOToCold2) {
-  NewVersionStorage(1, kCompactionStyleFIFO);
-  const uint64_t kFileSize = 100000;
-  const uint64_t kMaxSize = kFileSize * 100000;
-  uint64_t kColdThreshold = 2000;
-
-  fifo_options_.max_table_files_size = kMaxSize;
-  fifo_options_.file_temperature_age_thresholds = {
-      {Temperature::kCold, kColdThreshold}};
-  mutable_cf_options_.compaction_options_fifo = fifo_options_;
-  mutable_cf_options_.level0_file_num_compaction_trigger = 100;
-  mutable_cf_options_.max_compaction_bytes = kFileSize * 100;
-  FIFOCompactionPicker fifo_compaction_picker(ioptions_, &icmp_);
-
-  int64_t current_time = 0;
-  ASSERT_OK(Env::Default()->GetCurrentTime(&current_time));
-  uint64_t threshold_time =
-      static_cast<uint64_t>(current_time) - kColdThreshold;
-  Add(0, 6U, "240", "290", 2 * kFileSize, 0, 2900, 3000, 0, true,
-      Temperature::kUnknown, static_cast<uint64_t>(current_time) - 100);
-  Add(0, 4U, "260", "300", 1 * kFileSize, 0, 2500, 2600, 0, true,
-      Temperature::kUnknown, threshold_time);
-  // The following two files qualify for compaction to kCold.
-  Add(0, 3U, "200", "300", 4 * kFileSize, 0, 2300, 2400, 0, true,
-      Temperature::kUnknown, threshold_time - 3000);
-  Add(0, 2U, "200", "300", 4 * kFileSize, 0, 2100, 2200, 0, true,
-      Temperature::kUnknown, threshold_time - 4000);
-  UpdateVersionStorageInfo();
-
-  ASSERT_EQ(fifo_compaction_picker.NeedsCompaction(vstorage_.get()), true);
-  std::unique_ptr<Compaction> compaction(fifo_compaction_picker.PickCompaction(
-      cf_name_, mutable_cf_options_, mutable_db_options_, vstorage_.get(),
-      &log_buffer_));
-  ASSERT_TRUE(compaction.get() != nullptr);
-  ASSERT_EQ(compaction->compaction_reason(),
-            CompactionReason::kChangeTemperature);
-  ASSERT_EQ(compaction->output_temperature(), Temperature::kCold);
-  ASSERT_EQ(2U, compaction->num_input_files(0));
-  ASSERT_EQ(2U, compaction->input(0, 0)->fd.GetNumber());
-  ASSERT_EQ(3U, compaction->input(0, 1)->fd.GetNumber());
-}
-
 TEST_F(CompactionPickerTest, FIFOToColdMaxCompactionSize) {
   NewVersionStorage(1, kCompactionStyleFIFO);
   const uint64_t kFileSize = 100000;
@@ -1202,10 +1160,10 @@ TEST_F(CompactionPickerTest, FIFOToColdMaxCompactionSize) {
   ASSERT_TRUE(compaction.get() != nullptr);
   ASSERT_EQ(compaction->compaction_reason(),
             CompactionReason::kChangeTemperature);
+  // Compaction picker picks older files first and picks one file at a time.
   ASSERT_EQ(compaction->output_temperature(), Temperature::kCold);
-  ASSERT_EQ(2U, compaction->num_input_files(0));
+  ASSERT_EQ(1U, compaction->num_input_files(0));
   ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());
-  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());
 }
 
 TEST_F(CompactionPickerTest, FIFOToColdWithExistingCold) {
@@ -1248,10 +1206,10 @@ TEST_F(CompactionPickerTest, FIFOToColdWithExistingCold) {
   ASSERT_TRUE(compaction.get() != nullptr);
   ASSERT_EQ(compaction->compaction_reason(),
             CompactionReason::kChangeTemperature);
+  // Compaction picker picks older files first and picks one file at a time.
   ASSERT_EQ(compaction->output_temperature(), Temperature::kCold);
+  ASSERT_EQ(1U, compaction->num_input_files(0));
   ASSERT_EQ(2U, compaction->input(0, 0)->fd.GetNumber());
-  ASSERT_EQ(2U, compaction->num_input_files(0));
-  ASSERT_EQ(3U, compaction->input(0, 1)->fd.GetNumber());
 }
 
 TEST_F(CompactionPickerTest, FIFOToColdWithHotBetweenCold) {
@@ -1299,7 +1257,7 @@ TEST_F(CompactionPickerTest, FIFOToColdWithHotBetweenCold) {
   ASSERT_EQ(2U, compaction->input(0, 0)->fd.GetNumber());
 }
 
-TEST_F(CompactionPickerTest, FIFOToColdAndWarm) {
+TEST_F(CompactionPickerTest, FIFOToHotAndWarm) {
   NewVersionStorage(1, kCompactionStyleFIFO);
   const uint64_t kFileSize = 100000;
   const uint64_t kMaxSize = kFileSize * 100000;
@@ -1344,11 +1302,10 @@ TEST_F(CompactionPickerTest, FIFOToColdAndWarm) {
   ASSERT_TRUE(compaction.get() != nullptr);
   ASSERT_EQ(compaction->compaction_reason(),
             CompactionReason::kChangeTemperature);
-  // Assumes compaction picker picks older files first.
+  // Compaction picker picks older files first and picks one file at a time.
   ASSERT_EQ(compaction->output_temperature(), Temperature::kWarm);
-  ASSERT_EQ(2U, compaction->num_input_files(0));
+  ASSERT_EQ(1U, compaction->num_input_files(0));
   ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());
-  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());
 }
 
 TEST_F(CompactionPickerTest, CompactionPriMinOverlapping1) {

db/db_compaction_test.cc

@@ -9357,12 +9357,13 @@ TEST_F(DBCompactionTest, FIFOChangeTemperature) {
     ASSERT_OK(Flush());
 
     ASSERT_OK(Put(Key(0), "value1"));
-    env_->MockSleepForSeconds(800);
     ASSERT_OK(Put(Key(2), "value2"));
     ASSERT_OK(Flush());
 
+    // First two L0 files both become eligible for temperature change compaction
+    // They should be compacted one-by-one.
     ASSERT_OK(Put(Key(0), "value1"));
-    env_->MockSleepForSeconds(800);
+    env_->MockSleepForSeconds(1200);
     ASSERT_OK(Put(Key(2), "value2"));
     ASSERT_OK(Flush());
     ASSERT_OK(dbfull()->TEST_WaitForCompact());

include/rocksdb/advanced_options.h

@@ -89,7 +89,10 @@ struct CompactionOptionsFIFO {
   // Age (in seconds) threshold for different file temperatures.
   // When not empty, each element specifies an age threshold `age` and a
   // temperature such that if all the data in a file is older than `age`,
-  // RocksDB will compact the file to the specified `temperature`.
+  // RocksDB will compact the file to the specified `temperature`. Oldest file
+  // will be considered first. Only one file is compacted at a time,
+  // so multiple files qualifying to be compacted to be same temperature
+  // won't be merged together.
   //
   // Note:
   // - Flushed files will always have temperature kUnknown.

unreleased_history/behavior_changes/fifo-temp-compaction.md

@@ -0,0 +1 @@
+* 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).