[rocksdb] Support output temperature in CompactFiles

Author: Josh Kang

db/compaction/compaction_picker.cc

@@ -375,13 +375,18 @@ Compaction* CompactionPicker::PickCompactionForCompactFiles(
     // without configurable `CompressionOptions`, which is inconsistent.
     compression_type = compact_options.compression;
   }
+
+  auto writeTemperature =
+      compact_options.output_temperature == Temperature::kUnknown
+          ? mutable_cf_options.default_write_temperature
+          : compact_options.output_temperature;
+
   auto c = new Compaction(
       vstorage, ioptions_, mutable_cf_options, mutable_db_options, input_files,
       output_level, compact_options.output_file_size_limit,
       mutable_cf_options.max_compaction_bytes, output_path_id, compression_type,
       GetCompressionOptions(mutable_cf_options, vstorage, output_level),
-      mutable_cf_options.default_write_temperature,
-      compact_options.max_subcompactions,
+      writeTemperature, compact_options.max_subcompactions,
       /* grandparents */ {}, earliest_snapshot, snapshot_checker,
       CompactionReason::kManualCompaction);
   RegisterCompaction(c);

db/compaction/compaction_picker_test.cc

@@ -1470,6 +1470,29 @@ TEST_F(CompactionPickerTest, FIFOToHotAndWarm) {
   }
 }
 
+TEST_F(CompactionPickerTest, CompactFilesOutputTemperature) {
+  NewVersionStorage(6, kCompactionStyleLevel);
+  auto file_number = 66U;
+  Add(0, file_number, "150", "200", 1000000000U);
+  UpdateVersionStorageInfo();
+
+  std::unordered_set<uint64_t> input{file_number};
+  std::vector<CompactionInputFiles> input_files;
+  ASSERT_OK(level_compaction_picker.GetCompactionInputsFromFileNumbers(
+      &input_files, &input, vstorage_.get(), CompactionOptions()));
+
+  auto compactionOptions = CompactionOptions();
+  compactionOptions.output_temperature = Temperature::kCold;
+
+  std::unique_ptr<Compaction> compaction(
+      level_compaction_picker.PickCompactionForCompactFiles(
+          compactionOptions, input_files, 1, vstorage_.get(),
+          mutable_cf_options_, mutable_db_options_, /*output_path_id=*/0));
+
+  ASSERT_TRUE(compaction.get() != nullptr);
+  ASSERT_EQ(compaction->output_temperature(), Temperature::kCold);
+}
+
 TEST_F(CompactionPickerTest, CompactionPriMinOverlapping1) {
   NewVersionStorage(6, kCompactionStyleLevel);
   ioptions_.compaction_pri = kMinOverlappingRatio;

include/rocksdb/options.h

@@ -2351,6 +2351,11 @@ struct CompactionOptions {
   // canceled variable in CompactionOptions, as it does for CompactRangeOptions
   // - this is because ManualCompactionState is not used
 
+  // Create output compaction file using this file temperature. If unset, will
+  // default to "last_level_temperature" if output level is last level otherwise
+  // "default_write_temperature"
+  Temperature output_temperature = Temperature::kUnknown;
+
   CompactionOptions()
       : compression(kDisableCompressionOption),
         output_file_size_limit(std::numeric_limits<uint64_t>::max()),