[Draft] Persist fast SST open metadata

db/builder.cc

@@ -467,7 +467,9 @@ Status BuildTable(
           MaxFileSizeForL0MetaPin(mutable_cf_options),
           /*smallest_compaction_key=*/nullptr,
           /*largest_compaction_key*/ nullptr,
-          /*allow_unprepared_value*/ false));
+          /*allow_unprepared_value*/ false,
+          /*range_del_read_seqno=*/nullptr, /*range_del_iter=*/nullptr,
+          /*maybe_pin_table_handle=*/false, &meta->file_open_metadata));
       s = it->status();
       if (s.ok() && paranoid_file_checks) {
         OutputValidator file_validator(tboptions.internal_comparator,

db/compaction/compaction_job.cc

@@ -2133,6 +2133,18 @@ Status CompactionJob::FinishCompactionOutputFile(
     meta = nullptr;
   }
 
+  if (s.ok() && meta != nullptr && db_options_.fast_sst_open) {
+    FileOptions metadata_file_options = file_options_;
+    metadata_file_options.temperature = meta->temperature;
+    metadata_file_options.file_checksum = meta->file_checksum;
+    metadata_file_options.file_checksum_func_name =
+        meta->file_checksum_func_name;
+    IOStatus metadata_io_s = MaybeGetFileOpenMetadata(
+        cfd->ioptions().fs.get(), GetTableFileName(meta->fd.GetNumber()),
+        metadata_file_options, &meta->file_open_metadata);
+    metadata_io_s.PermitUncheckedError();
+  }
+
   if (s.ok() && (current_entries > 0 || tp.num_range_deletions > 0)) {
     // Output to event logger and fire events.
     outputs.UpdateTableProperties();
@@ -2862,6 +2874,11 @@ Status CompactionJob::ReadTablePropertiesDirectly(
   FileOptions fopts = file_options_;
   fopts.file_checksum = file_meta->file_checksum;
   fopts.file_checksum_func_name = file_meta->file_checksum_func_name;
+  std::string file_open_metadata;
+  if (ioptions.fast_sst_open && !file_meta->file_open_metadata.empty()) {
+    file_open_metadata = file_meta->file_open_metadata;
+    fopts.file_metadata = &file_open_metadata;
+  }
   Status s = ioptions.fs->NewRandomAccessFile(file_name, fopts, &file,
                                               nullptr /* dbg */);
   if (!s.ok()) {

db/compaction/compaction_job.h

@@ -617,6 +617,7 @@ struct CompactionServiceOutputFile {
   bool marked_for_compaction;
   UniqueId64x2 unique_id{};
   TableProperties table_properties;
+  std::string file_open_metadata;
   bool is_proximal_level_output;
   Temperature file_temperature = Temperature::kUnknown;
 
@@ -629,7 +630,8 @@ struct CompactionServiceOutputFile {
       const std::string& _file_checksum,
       const std::string& _file_checksum_func_name, uint64_t _paranoid_hash,
       bool _marked_for_compaction, UniqueId64x2 _unique_id,
-      const TableProperties& _table_properties, bool _is_proximal_level_output,
+      const TableProperties& _table_properties,
+      std::string _file_open_metadata, bool _is_proximal_level_output,
       Temperature _file_temperature)
       : file_name(name),
         file_size(size),
@@ -646,6 +648,7 @@ struct CompactionServiceOutputFile {
         marked_for_compaction(_marked_for_compaction),
         unique_id(std::move(_unique_id)),
         table_properties(_table_properties),
+        file_open_metadata(std::move(_file_open_metadata)),
         is_proximal_level_output(_is_proximal_level_output),
         file_temperature(_file_temperature) {}
 };

db/compaction/compaction_job_test.cc

@@ -1686,6 +1686,7 @@ TEST_F(CompactionJobTest, ResultSerialization) {
         file_checksum_func_name /* file_checksum_func_name */,
         rnd64.Uniform(UINT64_MAX) /* paranoid_hash */,
         rnd.OneIn(2) /* marked_for_compaction */, id /* unique_id */, tp,
+        rnd.RandomBinaryString(rnd.Uniform(kStrMaxLen)) /* file_open_metadata */,
         false /* is_proximal_level_output */, Temperature::kHot);
   }
   result.output_level = rnd.Uniform(10);

db/compaction/compaction_service_job.cc

@@ -258,6 +258,7 @@ CompactionJob::ProcessKeyValueCompactionWithCompactionService(
     meta.file_checksum_func_name = file.file_checksum_func_name;
     meta.marked_for_compaction = file.marked_for_compaction;
     meta.unique_id = file.unique_id;
+    meta.file_open_metadata = file.file_open_metadata;
     meta.temperature = file.file_temperature;
     meta.tail_size =
         FileMetaData::CalculateTailSize(file_size, file.table_properties);
@@ -445,6 +446,7 @@ Status CompactionServiceCompactionJob::Run() {
           meta.file_checksum, meta.file_checksum_func_name,
           output_file.validator.GetHash(), meta.marked_for_compaction,
           meta.unique_id, *output_file.table_properties,
+          meta.file_open_metadata,
           output_file.is_proximal_level, meta.temperature);
     }
   }
@@ -622,6 +624,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
             const auto that_one = static_cast<const TableProperties*>(addr2);
             return this_one->AreEqual(opts, that_one, mismatch);
           }}},
+        {"file_open_metadata",
+         {offsetof(struct CompactionServiceOutputFile, file_open_metadata),
+          OptionType::kEncodedString, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
         {"is_proximal_level_output",
          {offsetof(struct CompactionServiceOutputFile,
                    is_proximal_level_output),