Offline file checksum manifest retriever (#13178)

Summary:
This change introduces a new, lightweight _experimental_ API that reconstructs the [file # -> file checksum -> file checksum function] 1-1-1 mapping directly from the `MANIFEST` file considered `CURRENT` in scope of specific DB instance at the time. The goal is to provide a cheap alternative to `DB::GetLiveFilesMetaData` that doesn't require opening the database, reconstructing version sets and/or accessing files that are _potentially_ in disaggregated storage.

### Housekeeping:

1. Moved the `GetCurrentManifestPath` out of `version_set` to a new `manifest_ops` file(s) dedicated to manifest related operations.
2. Introduced new `Env::IOActivity::kReadManifest` to better reflect the IO intent in offline file checksum retrieving function.

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

Test Plan:
Added a unit test comparing the outcome of newly introduced API against the established `GetLiveFilesMetaData`:

```hcl
./db_test2 --gtest_filter="*GetFileChecksumsFromCurrentManifest_CRC32*"
```

Reviewed By: pdillinger

Differential Revision: D66711910

Pulled By: mszeszko-meta

fbshipit-source-id: 57091c550a14ac2e832bf7eea136dab5450e71bc

BUCK

@@ -83,6 +83,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
         "db/log_writer.cc",
         "db/logs_with_prep_tracker.cc",
         "db/malloc_stats.cc",
+        "db/manifest_ops.cc",
         "db/memtable.cc",
         "db/memtable_list.cc",
         "db/merge_helper.cc",

CMakeLists.txt

@@ -716,6 +716,7 @@ set(SOURCES
         db/log_reader.cc
         db/log_writer.cc
         db/malloc_stats.cc
+        db/manifest_ops.cc
         db/memtable.cc
         db/memtable_list.cc
         db/merge_helper.cc

db/db_impl/db_impl_readonly.cc

@@ -8,7 +8,7 @@
 #include "db/arena_wrapped_db_iter.h"
 #include "db/db_impl/compacted_db_impl.h"
 #include "db/db_impl/db_impl.h"
-#include "db/db_iter.h"
+#include "db/manifest_ops.h"
 #include "db/merge_context.h"
 #include "logging/logging.h"
 #include "monitoring/perf_context_imp.h"
@@ -265,9 +265,8 @@ Status OpenForReadOnlyCheckExistence(const DBOptions& db_options,
     const std::shared_ptr<FileSystem>& fs = db_options.env->GetFileSystem();
     std::string manifest_path;
     uint64_t manifest_file_number;
-    s = VersionSet::GetCurrentManifestPath(dbname, fs.get(), /*is_retry=*/false,
-                                           &manifest_path,
-                                           &manifest_file_number);
+    s = GetCurrentManifestPath(dbname, fs.get(), /*is_retry=*/false,
+                               &manifest_path, &manifest_file_number);
   } else {
     // Historic behavior that doesn't necessarily make sense
     s = db_options.env->CreateDirIfMissing(dbname);

db/db_test2.cc

@@ -16,6 +16,7 @@
 #include "db/db_test_util.h"
 #include "db/read_callback.h"
 #include "db/version_edit.h"
+#include "env/fs_readonly.h"
 #include "options/options_helper.h"
 #include "port/port.h"
 #include "port/stack_trace.h"
@@ -8061,6 +8062,63 @@ TEST_F(DBTest2, TableCacheMissDuringReadFromBlockCacheTier) {
   ASSERT_EQ(orig_num_file_opens, TestGetTickerCount(options, NO_FILE_OPENS));
 }
 
+TEST_F(DBTest2, GetFileChecksumsFromCurrentManifest_CRC32) {
+  Options opts = CurrentOptions();
+  opts.create_if_missing = true;
+  opts.file_checksum_gen_factory = GetFileChecksumGenCrc32cFactory();
+  opts.level0_file_num_compaction_trigger = 10;
+
+  // Bootstrap the test database.
+  DB* db = nullptr;
+  std::string dbname = test::PerThreadDBPath("file_chksum");
+  ASSERT_OK(DB::Open(opts, dbname, &db));
+
+  WriteOptions wopts;
+  FlushOptions fopts;
+  fopts.wait = true;
+  Random rnd(test::RandomSeed());
+  for (int i = 0; i < 4; i++) {
+    ASSERT_OK(db->Put(wopts, Key(i), rnd.RandomString(100)));
+    ASSERT_OK(db->Flush(fopts));
+  }
+
+  // Obtain rich files metadata for source of truth.
+  std::vector<LiveFileMetaData> live_files;
+  db->GetLiveFilesMetaData(&live_files);
+
+  ASSERT_OK(db->Close());
+  delete db;
+  db = nullptr;
+
+  // Process current MANIFEST file and build internal file checksum mappings.
+  std::unique_ptr<FileChecksumList> checksum_list(NewFileChecksumList());
+  auto read_only_fs =
+      std::make_shared<ReadOnlyFileSystem>(env_->GetFileSystem());
+  ASSERT_OK(experimental::GetFileChecksumsFromCurrentManifest(
+      read_only_fs.get(), dbname, checksum_list.get()));
+
+  ASSERT_TRUE(checksum_list != nullptr);
+
+  // Retrieve files, related checksums and checksum functions.
+  std::vector<uint64_t> file_numbers;
+  std::vector<std::string> checksums;
+  std::vector<std::string> checksum_func_names;
+  ASSERT_OK(checksum_list->GetAllFileChecksums(&file_numbers, &checksums,
+                                               &checksum_func_names));
+
+  // Compare results.
+  ASSERT_EQ(live_files.size(), checksum_list->size());
+  for (size_t i = 0; i < live_files.size(); i++) {
+    std::string stored_checksum;
+    std::string stored_func_name;
+    ASSERT_OK(checksum_list->SearchOneFileChecksum(
+        live_files[i].file_number, &stored_checksum, &stored_func_name));
+
+    ASSERT_EQ(live_files[i].file_checksum, stored_checksum);
+    ASSERT_EQ(live_files[i].file_checksum_func_name, stored_func_name);
+  }
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

db/experimental.cc

@@ -13,6 +13,8 @@
 #include <vector>
 
 #include "db/db_impl/db_impl.h"
+#include "db/manifest_ops.h"
+#include "db/version_edit_handler.h"
 #include "db/version_util.h"
 #include "logging/logging.h"
 #include "util/atomic.h"
@@ -40,6 +42,58 @@ Status SuggestCompactRange(DB* db, const Slice* begin, const Slice* end) {
   return SuggestCompactRange(db, db->DefaultColumnFamily(), begin, end);
 }
 
+Status GetFileChecksumsFromCurrentManifest(FileSystem* fs,
+                                           const std::string& dbname,
+                                           FileChecksumList* checksum_list) {
+  std::string manifest_path;
+  uint64_t manifest_file_number;
+  Status s = GetCurrentManifestPath(dbname, fs, true /* is_retry */,
+                                    &manifest_path, &manifest_file_number);
+  if (!s.ok()) {
+    return s;
+  }
+
+  if (checksum_list == nullptr) {
+    return Status::InvalidArgument("checksum_list is nullptr");
+  }
+  assert(checksum_list);
+
+  const ReadOptions read_options(Env::IOActivity::kReadManifest);
+  checksum_list->reset();
+
+  std::unique_ptr<SequentialFileReader> file_reader;
+  {
+    std::unique_ptr<FSSequentialFile> file;
+    s = fs->NewSequentialFile(manifest_path,
+                              fs->OptimizeForManifestRead(FileOptions()), &file,
+                              nullptr /* dbg */);
+    if (!s.ok()) {
+      return s;
+    }
+    file_reader.reset(new SequentialFileReader(std::move(file), manifest_path));
+  }
+
+  struct LogReporter : public log::Reader::Reporter {
+    Status* status_ptr;
+    void Corruption(size_t /*bytes*/, const Status& st) override {
+      if (status_ptr->ok()) {
+        *status_ptr = st;
+      }
+    }
+  } reporter;
+  reporter.status_ptr = &s;
+  log::Reader reader(nullptr, std::move(file_reader), &reporter,
+                     true /* checksum */, 0 /* log_number */);
+
+  // Read all records from the manifest file...
+  uint64_t manifest_file_size = std::numeric_limits<uint64_t>::max();
+  FileChecksumRetriever retriever(read_options, manifest_file_size,
+                                  *checksum_list);
+  retriever.Iterate(reader, &s);
+
+  return retriever.status();
+}
+
 Status UpdateManifestForFilesState(
     const DBOptions& db_opts, const std::string& db_name,
     const std::vector<ColumnFamilyDescriptor>& column_families,

db/manifest_ops.cc

@@ -0,0 +1,47 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "db/manifest_ops.h"
+
+#include "file/filename.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+Status GetCurrentManifestPath(const std::string& dbname, FileSystem* fs,
+                              bool is_retry, std::string* manifest_path,
+                              uint64_t* manifest_file_number) {
+  assert(fs != nullptr);
+  assert(manifest_path != nullptr);
+  assert(manifest_file_number != nullptr);
+
+  IOOptions opts;
+  std::string fname;
+  if (is_retry) {
+    opts.verify_and_reconstruct_read = true;
+  }
+  Status s = ReadFileToString(fs, CurrentFileName(dbname), opts, &fname);
+  if (!s.ok()) {
+    return s;
+  }
+  if (fname.empty() || fname.back() != '\n') {
+    return Status::Corruption("CURRENT file does not end with newline");
+  }
+  // remove the trailing '\n'
+  fname.resize(fname.size() - 1);
+  FileType type;
+  bool parse_ok = ParseFileName(fname, manifest_file_number, &type);
+  if (!parse_ok || type != kDescriptorFile) {
+    return Status::Corruption("CURRENT file corrupted");
+  }
+  *manifest_path = dbname;
+  if (dbname.back() != '/') {
+    manifest_path->push_back('/');
+  }
+  manifest_path->append(fname);
+  return Status::OK();
+}
+
+}  // namespace ROCKSDB_NAMESPACE

db/manifest_ops.h

@@ -0,0 +1,20 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+#include <cassert>
+
+#include "rocksdb/env.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+// Parameter is_retry=true sets the verify_and_reconstruct_read flag.
+// It comes handy when caller intends to re-read the data with much stronger
+// data integrity checking - e.g. in case of a perceived file corruption.
+Status GetCurrentManifestPath(const std::string& dbname, FileSystem* fs,
+                              bool is_retry, std::string* manifest_path,
+                              uint64_t* manifest_file_number);
+}  // namespace ROCKSDB_NAMESPACE

db/version_set.cc

@@ -31,6 +31,7 @@
 #include "db/internal_stats.h"
 #include "db/log_reader.h"
 #include "db/log_writer.h"
+#include "db/manifest_ops.h"
 #include "db/memtable.h"
 #include "db/merge_context.h"
 #include "db/merge_helper.h"
@@ -6011,41 +6012,6 @@ Status VersionSet::LogAndApplyHelper(ColumnFamilyData* cfd,
   return builder ? builder->Apply(edit) : Status::OK();
 }
 
-Status VersionSet::GetCurrentManifestPath(const std::string& dbname,
-                                          FileSystem* fs, bool is_retry,
-                                          std::string* manifest_path,
-                                          uint64_t* manifest_file_number) {
-  assert(fs != nullptr);
-  assert(manifest_path != nullptr);
-  assert(manifest_file_number != nullptr);
-
-  IOOptions opts;
-  std::string fname;
-  if (is_retry) {
-    opts.verify_and_reconstruct_read = true;
-  }
-  Status s = ReadFileToString(fs, CurrentFileName(dbname), opts, &fname);
-  if (!s.ok()) {
-    return s;
-  }
-  if (fname.empty() || fname.back() != '\n') {
-    return Status::Corruption("CURRENT file does not end with newline");
-  }
-  // remove the trailing '\n'
-  fname.resize(fname.size() - 1);
-  FileType type;
-  bool parse_ok = ParseFileName(fname, manifest_file_number, &type);
-  if (!parse_ok || type != kDescriptorFile) {
-    return Status::Corruption("CURRENT file corrupted");
-  }
-  *manifest_path = dbname;
-  if (dbname.back() != '/') {
-    manifest_path->push_back('/');
-  }
-  manifest_path->append(fname);
-  return Status::OK();
-}
-
 Status VersionSet::Recover(
     const std::vector<ColumnFamilyDescriptor>& column_families, bool read_only,
     std::string* db_id, bool no_error_if_files_missing, bool is_retry,

db/version_set.h

@@ -1277,10 +1277,6 @@ class VersionSet {
       const std::vector<std::function<void(const Status&)>>& manifest_wcbs =
           {});
 
-  static Status GetCurrentManifestPath(const std::string& dbname,
-                                       FileSystem* fs, bool is_retry,
-                                       std::string* manifest_filename,
-                                       uint64_t* manifest_file_number);
   void WakeUpWaitingManifestWriters();
 
   // Recover the last saved descriptor (MANIFEST) from persistent storage.

db/version_set_test.cc

@@ -15,6 +15,7 @@
 #include "db/db_impl/db_impl.h"
 #include "db/db_test_util.h"
 #include "db/log_writer.h"
+#include "db/manifest_ops.h"
 #include "db/version_edit.h"
 #include "rocksdb/advanced_options.h"
 #include "rocksdb/convenience.h"
@@ -1390,18 +1391,16 @@ class VersionSetTestBase {
   void GetManifestPath(std::string* manifest_path) const {
     assert(manifest_path != nullptr);
     uint64_t manifest_file_number = 0;
-    Status s = versions_->GetCurrentManifestPath(
-        dbname_, fs_.get(), /*is_retry=*/false, manifest_path,
-        &manifest_file_number);
+    Status s = GetCurrentManifestPath(dbname_, fs_.get(), /*is_retry=*/false,
+                                      manifest_path, &manifest_file_number);
     ASSERT_OK(s);
   }
 
   void VerifyManifest(std::string* manifest_path) const {
     assert(manifest_path != nullptr);
     uint64_t manifest_file_number = 0;
-    Status s = versions_->GetCurrentManifestPath(
-        dbname_, fs_.get(), /*is_retry=*/false, manifest_path,
-        &manifest_file_number);
+    Status s = GetCurrentManifestPath(dbname_, fs_.get(), /*is_retry=*/false,
+                                      manifest_path, &manifest_file_number);
     ASSERT_OK(s);
     ASSERT_EQ(1, manifest_file_number);
   }

include/rocksdb/env.h

@@ -458,6 +458,7 @@ class Env : public Customizable {
     kVerifyFileChecksums = 7,
     kGetEntity = 8,
     kMultiGetEntity = 9,
+    kReadManifest = 10,
     kUnknown,  // Keep last for easy array of non-unknowns
   };
 

include/rocksdb/experimental.h

@@ -30,6 +30,13 @@ Status SuggestCompactRange(DB* db, const Slice* begin, const Slice* end);
 Status PromoteL0(DB* db, ColumnFamilyHandle* column_family,
                  int target_level = 1);
 
+// This function is intended to be called with the DB closed and does not write
+// to the DB path. It will usually work on an open DB but may fail spuriously
+// in that case, and results may be out of date on return.
+Status GetFileChecksumsFromCurrentManifest(FileSystem* fs,
+                                           const std::string& dbname,
+                                           FileChecksumList* checksum_list);
+
 struct UpdateManifestForFilesStateOptions {
   // When true, read current file temperatures from FileSystem and update in
   // DB manifest when a temperature other than Unknown is reported and

include/rocksdb/thread_status.h

@@ -64,6 +64,7 @@ struct ThreadStatus {
     OP_VERIFY_FILE_CHECKSUMS,
     OP_GETENTITY,
     OP_MULTIGETENTITY,
+    OP_READ_MANIFEST,
     NUM_OP_TYPES
   };
 

monitoring/thread_status_util_debug.cc

@@ -50,6 +50,8 @@ Env::IOActivity ThreadStatusUtil::TEST_GetExpectedIOActivity(
       return Env::IOActivity::kGetEntity;
     case ThreadStatus::OperationType::OP_MULTIGETENTITY:
       return Env::IOActivity::kMultiGetEntity;
+    case ThreadStatus::OperationType::OP_READ_MANIFEST:
+      return Env::IOActivity::kReadManifest;
     default:
       return Env::IOActivity::kUnknown;
   }

src.mk

@@ -75,6 +75,7 @@ LIB_SOURCES =                                                   \
   db/log_reader.cc                                              \
   db/log_writer.cc                                              \
   db/malloc_stats.cc                                            \
+  db/manifest_ops.cc                                            \
   db/memtable.cc                                                \
   db/memtable_list.cc                                           \
   db/merge_helper.cc                                            \

util/file_checksum_helper.cc

@@ -9,12 +9,6 @@
 
 #include "util/file_checksum_helper.h"
 
-#include <unordered_set>
-
-#include "db/log_reader.h"
-#include "db/version_edit.h"
-#include "db/version_edit_handler.h"
-#include "file/sequence_file_reader.h"
 #include "rocksdb/utilities/customizable_util.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -91,52 +85,6 @@ std::shared_ptr<FileChecksumGenFactory> GetFileChecksumGenCrc32cFactory() {
   return default_crc32c_gen_factory;
 }
 
-Status GetFileChecksumsFromManifest(Env* src_env, const std::string& abs_path,
-                                    uint64_t manifest_file_size,
-                                    FileChecksumList* checksum_list) {
-  if (checksum_list == nullptr) {
-    return Status::InvalidArgument("checksum_list is nullptr");
-  }
-  assert(checksum_list);
-  // TODO: plumb Env::IOActivity, Env::IOPriority
-  const ReadOptions read_options;
-  checksum_list->reset();
-  Status s;
-
-  std::unique_ptr<SequentialFileReader> file_reader;
-  {
-    std::unique_ptr<FSSequentialFile> file;
-    const std::shared_ptr<FileSystem>& fs = src_env->GetFileSystem();
-    s = fs->NewSequentialFile(abs_path,
-                              fs->OptimizeForManifestRead(FileOptions()), &file,
-                              nullptr /* dbg */);
-    if (!s.ok()) {
-      return s;
-    }
-    file_reader.reset(new SequentialFileReader(std::move(file), abs_path));
-  }
-
-  struct LogReporter : public log::Reader::Reporter {
-    Status* status_ptr;
-    void Corruption(size_t /*bytes*/, const Status& st) override {
-      if (status_ptr->ok()) {
-        *status_ptr = st;
-      }
-    }
-  } reporter;
-  reporter.status_ptr = &s;
-  log::Reader reader(nullptr, std::move(file_reader), &reporter,
-                     true /* checksum */, 0 /* log_number */);
-  FileChecksumRetriever retriever(read_options, manifest_file_size,
-                                  *checksum_list);
-  retriever.Iterate(reader, &s);
-  assert(!retriever.status().ok() ||
-         manifest_file_size == std::numeric_limits<uint64_t>::max() ||
-         reader.LastRecordEnd() == manifest_file_size);
-
-  return retriever.status();
-}
-
 namespace {
 static int RegisterFileChecksumGenFactories(ObjectLibrary& library,
                                             const std::string& /*arg*/) {

util/file_checksum_helper.h

@@ -4,13 +4,10 @@
 //  (found in the LICENSE.Apache file in the root directory).
 
 #pragma once
-#include <cassert>
+
 #include <unordered_map>
 
-#include "port/port.h"
-#include "rocksdb/env.h"
 #include "rocksdb/file_checksum.h"
-#include "rocksdb/status.h"
 #include "util/coding.h"
 #include "util/crc32c.h"
 #include "util/math.h"
@@ -92,10 +89,4 @@ class FileChecksumListImpl : public FileChecksumList {
       checksum_map_;
 };
 
-// If manifest_file_size < std::numeric_limits<uint64_t>::max(), only use
-// that length prefix of the manifest file.
-Status GetFileChecksumsFromManifest(Env* src_env, const std::string& abs_path,
-                                    uint64_t manifest_file_size,
-                                    FileChecksumList* checksum_list);
-
 }  // namespace ROCKSDB_NAMESPACE

util/thread_operation.h

@@ -47,6 +47,7 @@ static OperationInfo global_operation_table[] = {
     {ThreadStatus::OP_VERIFY_FILE_CHECKSUMS, "VerifyFileChecksums"},
     {ThreadStatus::OP_GETENTITY, "GetEntity"},
     {ThreadStatus::OP_MULTIGETENTITY, "MultiGetEntity"},
+    {ThreadStatus::OP_READ_MANIFEST, "ReadManifest"},
 
 };