Add filename to several Corruption messages (#9239)

Summary:
This change adds the filename of the offending filen to several place that produce Status objects with code `kCorruption`.
This is not an attempt to have every Corruption message in the codebase extended with the filename, but it is a start.
The motivation for the change was to quickly diagnose which file is corrupted when a large database is openend and there is not option to copy it offsite for analysis, run strace or install the ldb tool.
In the particular case in question, the error message improved from a mere
```
Corruption: checksum mismatch
```
to
```
Corruption: checksum mismatch in file /path/to/db/engine-rocksdb/MANIFEST-000171
```

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

Reviewed By: jay-zhuang

Differential Revision: D33237742

Pulled By: riversand963

fbshipit-source-id: bd42559cfbf786a0a674d091671d1a2bf07bdd31

db/version_edit_handler.cc

@@ -10,6 +10,7 @@
 #include "db/version_edit_handler.h"
 
 #include <cinttypes>
+#include <sstream>
 
 #include "db/blob/blob_file_cache.h"
 #include "db/blob/blob_file_reader.h"
@@ -68,6 +69,26 @@ void VersionEditHandlerBase::Iterate(log::Reader& reader,
   CheckIterationResult(reader, &s);
 
   if (!s.ok()) {
+    if (s.IsCorruption()) {
+      // when we find a Corruption error, something is
+      // wrong with the underlying file. in this case we
+      // want to report the filename, so in here we append
+      // the filename to the Corruption message
+      assert(reader.file());
+
+      // build a new error message
+      std::stringstream message;
+      // append previous dynamic state message
+      const char* state = s.getState();
+      if (state != nullptr) {
+        message << state;
+        message << ' ';
+      }
+      // append the filename to the corruption message
+      message << "in file " << reader.file()->file_name();
+      // overwrite the status with the extended status
+      s = Status(s.code(), s.subcode(), s.severity(), message.str());
+    }
     status_ = s;
   }
   TEST_SYNC_POINT_CALLBACK("VersionEditHandlerBase::Iterate:Finish",

db/version_set_test.cc

@@ -2629,6 +2629,7 @@ TEST_P(VersionSetTestEmptyDb, OpenFromIncompleteManifest0) {
   if (iter == cf_names.end()) {
     ASSERT_TRUE(s.IsInvalidArgument());
   } else {
+    ASSERT_NE(s.ToString().find(manifest_path), std::string::npos);
     ASSERT_TRUE(s.IsCorruption());
   }
 }
@@ -2670,6 +2671,7 @@ TEST_P(VersionSetTestEmptyDb, OpenFromIncompleteManifest1) {
   if (iter == cf_names.end()) {
     ASSERT_TRUE(s.IsInvalidArgument());
   } else {
+    ASSERT_NE(s.ToString().find(manifest_path), std::string::npos);
     ASSERT_TRUE(s.IsCorruption());
   }
 }
@@ -2716,6 +2718,7 @@ TEST_P(VersionSetTestEmptyDb, OpenFromInCompleteManifest2) {
   if (iter == cf_names.end()) {
     ASSERT_TRUE(s.IsInvalidArgument());
   } else {
+    ASSERT_NE(s.ToString().find(manifest_path), std::string::npos);
     ASSERT_TRUE(s.IsCorruption());
   }
 }
@@ -2773,6 +2776,7 @@ TEST_P(VersionSetTestEmptyDb, OpenManifestWithUnknownCF) {
   if (iter == cf_names.end()) {
     ASSERT_TRUE(s.IsInvalidArgument());
   } else {
+    ASSERT_NE(s.ToString().find(manifest_path), std::string::npos);
     ASSERT_TRUE(s.IsCorruption());
   }
 }