Add an API to check if an SST file is generated by SstFileWriter (#13072)

Summary:
Some users want to check if a file in their DB was created by SstFileWriter and ingested into hte DB. Files created by SstFileWriter records additional table properties https://github.com/facebook/rocksdb/blob/cbebbad7d9353173bb0e2580da2eef71c5c18199/table/sst_file_writer_collectors.h#L17-L21. These are not exposed so I'm adding an API to SstFileWriter to check for them.

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

Test Plan: added some assertions.

Reviewed By: jowlyzhang

Differential Revision: D64411518

Pulled By: cbi42

fbshipit-source-id: 279bfef48615aa6f78287b643d8445a1471e7f07

db/external_sst_file_ingestion_job.cc

@@ -756,6 +756,7 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
   // Get table version
   auto version_iter = uprops.find(ExternalSstFilePropertyNames::kVersion);
   if (version_iter == uprops.end()) {
+    assert(!SstFileWriter::CreatedBySstFileWriter(*props));
     if (!ingestion_options_.allow_db_generated_files) {
       return Status::Corruption("External file version not found");
     } else {
@@ -764,6 +765,7 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
       file_to_ingest->version = 0;
     }
   } else {
+    assert(SstFileWriter::CreatedBySstFileWriter(*props));
     file_to_ingest->version = DecodeFixed32(version_iter->second.c_str());
   }
 

include/rocksdb/sst_file_writer.h

@@ -195,6 +195,9 @@ class SstFileWriter {
   // Return the current file size.
   uint64_t FileSize();
 
+  // Check if a file with input table property is created by SstFileWriter.
+  static bool CreatedBySstFileWriter(const TableProperties&);
+
  private:
   void InvalidatePageCache(bool closing);
   struct Rep;

table/sst_file_writer.cc

@@ -533,4 +533,9 @@ Status SstFileWriter::Finish(ExternalSstFileInfo* file_info) {
 
 uint64_t SstFileWriter::FileSize() { return rep_->file_info.file_size; }
 
+bool SstFileWriter::CreatedBySstFileWriter(const TableProperties& tp) {
+  const auto& uprops = tp.user_collected_properties;
+  return uprops.find(ExternalSstFilePropertyNames::kVersion) != uprops.end();
+}
+
 }  // namespace ROCKSDB_NAMESPACE