Remove DB::VerifyFileChecksums

db/corruption_test.cc

@@ -840,13 +840,15 @@ TEST_F(CorruptionTest, VerifyWholeTableChecksum) {
 
   Build(10, 5);
 
-  ASSERT_OK(db_->VerifyFileChecksums(ReadOptions()));
+  auto* dbi = static_cast_with_check<DBImpl>(db_);
+  ASSERT_OK(dbi->VerifyFileChecksums(ReadOptions()));
   CloseDb();
 
   // Corrupt the first byte of each table file, this must be data block.
   Corrupt(kTableFile, 0, 1);
 
   ASSERT_OK(TryReopen(&options));
+  dbi = static_cast_with_check<DBImpl>(db_);
 
   SyncPoint::GetInstance()->DisableProcessing();
   SyncPoint::GetInstance()->ClearAllCallBacks();
@@ -859,8 +861,23 @@ TEST_F(CorruptionTest, VerifyWholeTableChecksum) {
         ASSERT_NOK(*s);
       });
   SyncPoint::GetInstance()->EnableProcessing();
-  ASSERT_TRUE(db_->VerifyFileChecksums(ReadOptions()).IsCorruption());
+  ASSERT_TRUE(dbi->VerifyFileChecksums(ReadOptions()).IsCorruption());
   ASSERT_EQ(1, count);
+
+  CloseDb();
+  ASSERT_OK(DestroyDB(dbname_, options));
+  Reopen(&options);
+  Build(10, 5);
+  dbi = static_cast_with_check<DBImpl>(db_);
+  ASSERT_OK(dbi->VerifyFileChecksums(ReadOptions()));
+  CloseDb();
+  Corrupt(kTableFile, 0, 1);
+
+  // Set best_efforts_recovery to true
+  options.best_efforts_recovery = true;
+#ifdef OS_LINUX
+  ASSERT_TRUE(TryReopen(&options).IsCorruption());
+#endif  // OS_LINUX
 }
 
 }  // namespace ROCKSDB_NAMESPACE

db/db_basic_test.cc

@@ -2201,6 +2201,8 @@ TEST_F(DBBasicTest, MultiGetIOBufferOverrun) {
 
 TEST_F(DBBasicTest, IncrementalRecoveryNoCorrupt) {
   Options options = CurrentOptions();
+  options.file_checksum_gen_factory =
+      ROCKSDB_NAMESPACE::GetFileChecksumGenCrc32cFactory();
   DestroyAndReopen(options);
   CreateAndReopenWithCF({"pikachu", "eevee"}, options);
   size_t num_cfs = handles_.size();
@@ -2239,6 +2241,8 @@ TEST_F(DBBasicTest, IncrementalRecoveryNoCorrupt) {
 
 TEST_F(DBBasicTest, BestEffortsRecoveryWithVersionBuildingFailure) {
   Options options = CurrentOptions();
+  options.file_checksum_gen_factory =
+      ROCKSDB_NAMESPACE::GetFileChecksumGenCrc32cFactory();
   DestroyAndReopen(options);
   ASSERT_OK(Put("foo", "value"));
   ASSERT_OK(Flush());
@@ -2285,6 +2289,8 @@ TEST_F(DBBasicTest, RecoverWithMissingFiles) {
   // Disable auto compaction to simplify SST file name tracking.
   options.disable_auto_compactions = true;
   options.listeners.emplace_back(listener);
+  options.file_checksum_gen_factory =
+      ROCKSDB_NAMESPACE::GetFileChecksumGenCrc32cFactory();
   CreateAndReopenWithCF({"pikachu", "eevee"}, options);
   std::vector<std::string> all_cf_names = {kDefaultColumnFamilyName, "pikachu",
                                            "eevee"};
@@ -2345,6 +2351,8 @@ TEST_F(DBBasicTest, RecoverWithMissingFiles) {
 
 TEST_F(DBBasicTest, BestEffortsRecoveryTryMultipleManifests) {
   Options options = CurrentOptions();
+  options.file_checksum_gen_factory =
+      ROCKSDB_NAMESPACE::GetFileChecksumGenCrc32cFactory();
   options.env = env_;
   DestroyAndReopen(options);
   ASSERT_OK(Put("foo", "value0"));
@@ -2371,6 +2379,8 @@ TEST_F(DBBasicTest, BestEffortsRecoveryTryMultipleManifests) {
 
 TEST_F(DBBasicTest, RecoverWithNoCurrentFile) {
   Options options = CurrentOptions();
+  options.file_checksum_gen_factory =
+      ROCKSDB_NAMESPACE::GetFileChecksumGenCrc32cFactory();
   options.env = env_;
   DestroyAndReopen(options);
   CreateAndReopenWithCF({"pikachu"}, options);
@@ -2394,6 +2404,8 @@ TEST_F(DBBasicTest, RecoverWithNoCurrentFile) {
 
 TEST_F(DBBasicTest, RecoverWithNoManifest) {
   Options options = CurrentOptions();
+  options.file_checksum_gen_factory =
+      ROCKSDB_NAMESPACE::GetFileChecksumGenCrc32cFactory();
   options.env = env_;
   DestroyAndReopen(options);
   ASSERT_OK(Put("foo", "value"));
@@ -2423,6 +2435,8 @@ TEST_F(DBBasicTest, RecoverWithNoManifest) {
 
 TEST_F(DBBasicTest, SkipWALIfMissingTableFiles) {
   Options options = CurrentOptions();
+  options.file_checksum_gen_factory =
+      ROCKSDB_NAMESPACE::GetFileChecksumGenCrc32cFactory();
   DestroyAndReopen(options);
   TableFileListener* listener = new TableFileListener();
   options.listeners.emplace_back(listener);
@@ -3311,17 +3325,17 @@ TEST_F(DBBasicTest, VerifyFileChecksums) {
   DestroyAndReopen(options);
   ASSERT_OK(Put("a", "value"));
   ASSERT_OK(Flush());
-  ASSERT_TRUE(db_->VerifyFileChecksums(ReadOptions()).IsInvalidArgument());
+  ASSERT_TRUE(dbfull()->VerifyFileChecksums(ReadOptions()).IsInvalidArgument());
 
   options.file_checksum_gen_factory = GetFileChecksumGenCrc32cFactory();
   Reopen(options);
-  ASSERT_OK(db_->VerifyFileChecksums(ReadOptions()));
+  ASSERT_OK(dbfull()->VerifyFileChecksums(ReadOptions()));
 
   // Write an L0 with checksum computed.
   ASSERT_OK(Put("b", "value"));
   ASSERT_OK(Flush());
 
-  ASSERT_OK(db_->VerifyFileChecksums(ReadOptions()));
+  ASSERT_OK(dbfull()->VerifyFileChecksums(ReadOptions()));
 }
 #endif  // !ROCKSDB_LITE
 

db/db_impl/db_impl.cc

@@ -4778,11 +4778,11 @@ Status DBImpl::VerifyChecksumInternal(const ReadOptions& read_options,
            j++) {
         const auto& fd_with_krange = vstorage->LevelFilesBrief(i).files[j];
         const auto& fd = fd_with_krange.fd;
-        const FileMetaData* fmeta = fd_with_krange.file_metadata;
-        assert(fmeta);
         std::string fname = TableFileName(cfd->ioptions()->cf_paths,
                                           fd.GetNumber(), fd.GetPathId());
         if (use_file_checksum) {
+          const FileMetaData* fmeta = fd_with_krange.file_metadata;
+          assert(fmeta);
           s = VerifySstFileChecksum(*fmeta, fname, read_options);
         } else {
           s = ROCKSDB_NAMESPACE::VerifySstFileChecksum(opts, file_options_,

db/db_impl/db_impl.h

@@ -373,6 +373,12 @@ class DBImpl : public DB {
       uint64_t start_time, uint64_t end_time,
       std::unique_ptr<StatsHistoryIterator>* stats_iterator) override;
 
+  // If immutable_db_options_.best_efforts_recovery is true, and
+  // RocksDbFileChecksumsVerificationEnabledOnRecovery is defined and returns
+  // true, and immutable_db_options_.file_checksum_gen_factory is not nullptr,
+  // then call VerifyFileChecksums().
+  Status MaybeVerifyFileChecksums();
+
 #ifndef ROCKSDB_LITE
   using DB::ResetStats;
   virtual Status ResetStats() override;
@@ -431,8 +437,7 @@ class DBImpl : public DB {
       const ExportImportFilesMetaData& metadata,
       ColumnFamilyHandle** handle) override;
 
-  using DB::VerifyFileChecksums;
-  Status VerifyFileChecksums(const ReadOptions& read_options) override;
+  Status VerifyFileChecksums(const ReadOptions& read_options);
 
   using DB::VerifyChecksum;
   virtual Status VerifyChecksum(const ReadOptions& /*read_options*/) override;

db/db_impl/db_impl_open.cc

@@ -23,6 +23,15 @@
 #include "test_util/sync_point.h"
 #include "util/rate_limiter.h"
 
+#if !defined(ROCKSDB_LITE) && defined(OS_LINUX)
+// VerifyFileChecksums is a weak symbol.
+// If it is defined and returns true, and options.best_efforts_recovery = true,
+// and file checksum is enabled, then the checksums of table files will be
+// computed and verified with MANIFEST.
+extern "C" bool RocksDbFileChecksumsVerificationEnabledOnRecovery()
+    __attribute__((__weak__));
+#endif  // !ROCKSDB_LITE && OS_LINUX
+
 namespace ROCKSDB_NAMESPACE {
 Options SanitizeOptions(const std::string& dbname, const Options& src) {
   auto db_options = SanitizeOptions(dbname, DBOptions(src));
@@ -1404,6 +1413,22 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
   return s;
 }
 
+Status DBImpl::MaybeVerifyFileChecksums() {
+  Status s;
+#if !defined(ROCKSDB_LITE) && defined(OS_LINUX)
+  // TODO: remove the VerifyFileChecksums() call because it's very expensive.
+  if (immutable_db_options_.best_efforts_recovery &&
+      RocksDbFileChecksumsVerificationEnabledOnRecovery &&
+      RocksDbFileChecksumsVerificationEnabledOnRecovery() &&
+      immutable_db_options_.file_checksum_gen_factory) {
+    s = VerifyFileChecksums(ReadOptions());
+    ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                   "Verified file checksums: %s\n", s.ToString().c_str());
+  }
+#endif  // !ROCKSDB_LITE && OS_LINUX
+  return s;
+}
+
 Status DB::Open(const Options& options, const std::string& dbname, DB** dbptr) {
   DBOptions db_options(options);
   ColumnFamilyOptions cf_options(options);
@@ -1779,6 +1804,9 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
                    "Persisting Option File error: %s",
                    persist_options_status.ToString().c_str());
   }
+  if (s.ok()) {
+    s = impl->MaybeVerifyFileChecksums();
+  }
   if (s.ok()) {
     impl->StartPeriodicWorkScheduler();
   } else {

db/db_impl/db_impl_readonly.cc

@@ -233,6 +233,9 @@ Status DBImplReadOnly::OpenForReadOnlyWithoutCheck(
   }
   impl->mutex_.Unlock();
   sv_context.Clean();
+  if (s.ok()) {
+    s = impl->MaybeVerifyFileChecksums();
+  }
   if (s.ok()) {
     *dbptr = impl;
     for (auto* h : *handles) {

db/db_test_util.cc

@@ -15,6 +15,10 @@
 #include "rocksdb/utilities/object_registry.h"
 #include "util/random.h"
 
+extern "C" bool RocksDbFileChecksumsVerificationEnabledOnRecovery() {
+  return true;
+}
+
 namespace ROCKSDB_NAMESPACE {
 
 namespace {

db/db_test_util.h

@@ -49,6 +49,8 @@
 #include "util/string_util.h"
 #include "utilities/merge_operators.h"
 
+extern "C" bool RocksDbFileChecksumsVerificationEnabledOnRecovery();
+
 namespace ROCKSDB_NAMESPACE {
 
 namespace anon {

include/rocksdb/db.h

@@ -1443,12 +1443,6 @@ class DB {
       const ExportImportFilesMetaData& metadata,
       ColumnFamilyHandle** handle) = 0;
 
-  // Verify the checksums of files in db. Currently the whole-file checksum of
-  // table files are checked.
-  virtual Status VerifyFileChecksums(const ReadOptions& /*read_options*/) {
-    return Status::NotSupported("File verification not supported");
-  }
-
   // Verify the block checksums of files in db. The block checksums of table
   // files are checked.
   virtual Status VerifyChecksum(const ReadOptions& read_options) = 0;

include/rocksdb/utilities/stackable_db.h

@@ -141,11 +141,6 @@ class StackableDB : public DB {
                                              import_options, metadata, handle);
   }
 
-  using DB::VerifyFileChecksums;
-  Status VerifyFileChecksums(const ReadOptions& read_opts) override {
-    return db_->VerifyFileChecksums(read_opts);
-  }
-
   virtual Status VerifyChecksum() override { return db_->VerifyChecksum(); }
 
   virtual Status VerifyChecksum(const ReadOptions& options) override {