Updated GenerateOneFileChecksum to use requested_checksum_func_name (#7586)

Summary:
CreateFileChecksumGenerator may uses requested_checksum_func_name in generator context to decide which generator will be used. GenerateOneFileChecksum has not being updated to use it, which will always get the generator when the name is empty. Fix it.

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

Test Plan: make check

Reviewed By: riversand963

Differential Revision: D24491989

Pulled By: zhichao-cao

fbshipit-source-id: d9fdfdd431240f0a9a2e781ddbd48a7d6c609aad

db/external_sst_file_ingestion_job.cc

@@ -192,10 +192,13 @@ Status ExternalSstFileIngestionJob::Prepare(
     // Step 1: generate the checksum for ingested sst file.
     if (need_generate_file_checksum_) {
       for (size_t i = 0; i < files_to_ingest_.size(); i++) {
-        std::string generated_checksum, generated_checksum_func_name;
+        std::string generated_checksum;
+        std::string generated_checksum_func_name;
+        std::string requested_checksum_func_name;
         IOStatus io_s = GenerateOneFileChecksum(
             fs_.get(), files_to_ingest_[i].internal_file_path,
-            db_options_.file_checksum_gen_factory.get(), &generated_checksum,
+            db_options_.file_checksum_gen_factory.get(),
+            requested_checksum_func_name, &generated_checksum,
             &generated_checksum_func_name,
             ingestion_options_.verify_checksums_readahead_size,
             db_options_.allow_mmap_reads, io_tracer_);
@@ -839,11 +842,13 @@ IOStatus ExternalSstFileIngestionJob::GenerateChecksumForIngestedFile(
     // file checksum generated during Prepare(). This step will be skipped.
     return IOStatus::OK();
   }
-  std::string file_checksum, file_checksum_func_name;
+  std::string file_checksum;
+  std::string file_checksum_func_name;
+  std::string requested_checksum_func_name;
   IOStatus io_s = GenerateOneFileChecksum(
       fs_.get(), file_to_ingest->internal_file_path,
-      db_options_.file_checksum_gen_factory.get(), &file_checksum,
-      &file_checksum_func_name,
+      db_options_.file_checksum_gen_factory.get(), requested_checksum_func_name,
+      &file_checksum, &file_checksum_func_name,
       ingestion_options_.verify_checksums_readahead_size,
       db_options_.allow_mmap_reads, io_tracer_);
   if (!io_s.ok()) {

file/file_util.cc

@@ -123,13 +123,16 @@ bool IsWalDirSameAsDBPath(const ImmutableDBOptions* db_options) {
   return same;
 }
 
-IOStatus GenerateOneFileChecksum(FileSystem* fs, const std::string& file_path,
-                                 FileChecksumGenFactory* checksum_factory,
-                                 std::string* file_checksum,
-                                 std::string* file_checksum_func_name,
-                                 size_t verify_checksums_readahead_size,
-                                 bool allow_mmap_reads,
-                                 std::shared_ptr<IOTracer>& io_tracer) {
+// requested_checksum_func_name brings the function name of the checksum
+// generator in checksum_factory. Checksum factories may use or ignore
+// requested_checksum_func_name.
+IOStatus GenerateOneFileChecksum(
+    FileSystem* fs, const std::string& file_path,
+    FileChecksumGenFactory* checksum_factory,
+    const std::string& requested_checksum_func_name, std::string* file_checksum,
+    std::string* file_checksum_func_name,
+    size_t verify_checksums_readahead_size, bool allow_mmap_reads,
+    std::shared_ptr<IOTracer>& io_tracer) {
   if (checksum_factory == nullptr) {
     return IOStatus::InvalidArgument("Checksum factory is invalid");
   }
@@ -137,8 +140,25 @@ IOStatus GenerateOneFileChecksum(FileSystem* fs, const std::string& file_path,
   assert(file_checksum_func_name != nullptr);
 
   FileChecksumGenContext gen_context;
+  gen_context.requested_checksum_func_name = requested_checksum_func_name;
+  gen_context.file_name = file_path;
   std::unique_ptr<FileChecksumGenerator> checksum_generator =
       checksum_factory->CreateFileChecksumGenerator(gen_context);
+  if (checksum_generator == nullptr) {
+    std::string msg =
+        "Cannot get the file checksum generator based on the requested "
+        "checksum function name: " +
+        requested_checksum_func_name +
+        " from checksum factory: " + checksum_factory->Name();
+    return IOStatus::InvalidArgument(msg);
+  }
+
+  // For backward compatable, requested_checksum_func_name can be empty.
+  // If we give the requested checksum function name, we expect it is the
+  // same name of the checksum generator.
+  assert(!checksum_generator || requested_checksum_func_name.empty() ||
+         requested_checksum_func_name == checksum_generator->Name());
+
   uint64_t size;
   IOStatus io_s;
   std::unique_ptr<RandomAccessFileReader> reader;

file/file_util.h

@@ -35,7 +35,8 @@ extern bool IsWalDirSameAsDBPath(const ImmutableDBOptions* db_options);
 
 extern IOStatus GenerateOneFileChecksum(
     FileSystem* fs, const std::string& file_path,
-    FileChecksumGenFactory* checksum_factory, std::string* file_checksum,
+    FileChecksumGenFactory* checksum_factory,
+    const std::string& requested_checksum_func_name, std::string* file_checksum,
     std::string* file_checksum_func_name,
     size_t verify_checksums_readahead_size, bool allow_mmap_reads,
     std::shared_ptr<IOTracer>& io_tracer);