Add filename to corruption errors

When a corruption happens, report the filename of the file where
corruptions happens.  This will aid in diagnosing database corruption
issues.

Adds a GetName() to all the environment file classes to make it possible
to report a filename downstream.

Author: laanwj

db/db_impl.cc

@@ -428,7 +428,7 @@ Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log,
   while (reader.ReadRecord(&record, &scratch) && status.ok()) {
     if (record.size() < 12) {
       reporter.Corruption(record.size(),
-                          Status::Corruption("log record too small"));
+                          Status::Corruption("log record too small", fname));
       continue;
     }
     WriteBatchInternal::SetContents(&batch, record);

db/db_test.cc

@@ -158,6 +158,7 @@ class SpecialEnv : public EnvWrapper {
         }
         return base_->Sync();
       }
+      std::string GetName() const override { return ""; }
     };
     class ManifestFile : public WritableFile {
      private:
@@ -183,6 +184,7 @@ class SpecialEnv : public EnvWrapper {
           return base_->Sync();
         }
       }
+      std::string GetName() const override { return ""; }
     };
 
     if (non_writable_.load(std::memory_order_acquire)) {
@@ -216,6 +218,7 @@ class SpecialEnv : public EnvWrapper {
         counter_->Increment();
         return target_->Read(offset, n, result, scratch);
       }
+      std::string GetName() const override { return ""; }
     };
 
     Status s = target()->NewRandomAccessFile(f, r);

db/fault_injection_test.cc

@@ -114,6 +114,7 @@ class TestWritableFile : public WritableFile {
   Status Close() override;
   Status Flush() override;
   Status Sync() override;
+  std::string GetName() const override { return ""; }
 
  private:
   FileState state_;

db/leveldbutil.cc

@@ -20,6 +20,7 @@ class StdoutPrinter : public WritableFile {
   Status Close() override { return Status::OK(); }
   Status Flush() override { return Status::OK(); }
   Status Sync() override { return Status::OK(); }
+  std::string GetName() const override { return "[stdout]"; }
 };
 
 bool HandleDumpCommand(Env* env, char** files, int num) {

db/log_reader.cc

@@ -176,7 +176,7 @@ bool Reader::ReadRecord(Slice* record, std::string* scratch) {
 uint64_t Reader::LastRecordOffset() { return last_record_offset_; }
 
 void Reader::ReportCorruption(uint64_t bytes, const char* reason) {
-  ReportDrop(bytes, Status::Corruption(reason));
+  ReportDrop(bytes, Status::Corruption(reason, file_->GetName()));
 }
 
 void Reader::ReportDrop(uint64_t bytes, const Status& reason) {

db/log_test.cc

@@ -168,6 +168,7 @@ class LogTest {
       contents_.append(slice.data(), slice.size());
       return Status::OK();
     }
+    std::string GetName() const override { return ""; }
 
     std::string contents_;
   };
@@ -204,6 +205,7 @@ class LogTest {
 
       return Status::OK();
     }
+    std::string GetName() const { return ""; }
 
     Slice contents_;
     bool force_error_;

db/repair.cc

@@ -183,7 +183,7 @@ class Repairer {
     while (reader.ReadRecord(&record, &scratch)) {
       if (record.size() < 12) {
         reporter.Corruption(record.size(),
-                            Status::Corruption("log record too small"));
+                            Status::Corruption("log record too small", logname));
         continue;
       }
       WriteBatchInternal::SetContents(&batch, record);

helpers/memenv/memenv.cc

@@ -178,6 +178,7 @@ class SequentialFileImpl : public SequentialFile {
     return Status::OK();
   }
 
+  virtual std::string GetName() const override { return "[memenv]"; }
  private:
   FileState* file_;
   uint64_t pos_;
@@ -194,6 +195,7 @@ class RandomAccessFileImpl : public RandomAccessFile {
     return file_->Read(offset, n, result, scratch);
   }
 
+  virtual std::string GetName() const override { return "[memenv]"; }
  private:
   FileState* file_;
 };
@@ -210,6 +212,7 @@ class WritableFileImpl : public WritableFile {
   Status Flush() override { return Status::OK(); }
   Status Sync() override { return Status::OK(); }
 
+  virtual std::string GetName() const override { return "[memenv]"; }
  private:
   FileState* file_;
 };

include/leveldb/env.h

@@ -217,6 +217,9 @@ class LEVELDB_EXPORT SequentialFile {
   //
   // REQUIRES: External synchronization
   virtual Status Skip(uint64_t n) = 0;
+
+  // Get a name for the file, only for error reporting
+  virtual std::string GetName() const = 0;
 };
 
 // A file abstraction for randomly reading the contents of a file.
@@ -240,6 +243,9 @@ class LEVELDB_EXPORT RandomAccessFile {
   // Safe for concurrent use by multiple threads.
   virtual Status Read(uint64_t offset, size_t n, Slice* result,
                       char* scratch) const = 0;
+
+  // Get a name for the file, only for error reporting
+  virtual std::string GetName() const = 0;
 };
 
 // A file abstraction for sequential writing.  The implementation
@@ -258,6 +264,9 @@ class LEVELDB_EXPORT WritableFile {
   virtual Status Close() = 0;
   virtual Status Flush() = 0;
   virtual Status Sync() = 0;
+
+  // Get a name for the file, only for error reporting
+  virtual std::string GetName() const = 0;
 };
 
 // An interface for writing log messages.

table/format.cc

@@ -79,7 +79,7 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
   }
   if (contents.size() != n + kBlockTrailerSize) {
     delete[] buf;
-    return Status::Corruption("truncated block read");
+    return Status::Corruption("truncated block read", file->GetName());
   }
 
   // Check the crc of the type and the block contents
@@ -89,7 +89,7 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
     const uint32_t actual = crc32c::Value(data, n + 1);
     if (actual != crc) {
       delete[] buf;
-      s = Status::Corruption("block checksum mismatch");
+      s = Status::Corruption("block checksum mismatch", file->GetName());
       return s;
     }
   }
@@ -116,13 +116,13 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
       size_t ulength = 0;
       if (!port::Snappy_GetUncompressedLength(data, n, &ulength)) {
         delete[] buf;
-        return Status::Corruption("corrupted compressed block contents");
+        return Status::Corruption("corrupted compressed block contents", file->GetName());
       }
       char* ubuf = new char[ulength];
       if (!port::Snappy_Uncompress(data, n, ubuf)) {
         delete[] buf;
         delete[] ubuf;
-        return Status::Corruption("corrupted compressed block contents");
+        return Status::Corruption("corrupted compressed block contents", file->GetName());
       }
       delete[] buf;
       result->data = Slice(ubuf, ulength);
@@ -132,7 +132,7 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
     }
     default:
       delete[] buf;
-      return Status::Corruption("bad block type");
+      return Status::Corruption("bad block type", file->GetName());
   }
 
   return Status::OK();