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

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

db/leveldbutil.cc

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

db/log_reader.cc

@@ -186,7 +186,7 @@ uint64_t Reader::LastRecordOffset() {
 }
 
 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/repair.cc

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

helpers/memenv/memenv.cc

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

include/leveldb/env.h

@@ -191,6 +191,9 @@ class 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;
+
  private:
   // No copying allowed
   SequentialFile(const SequentialFile&);
@@ -215,6 +218,9 @@ class RandomAccessFile {
   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;
+
  private:
   // No copying allowed
   RandomAccessFile(const RandomAccessFile&);
@@ -234,6 +240,9 @@ class WritableFile {
   virtual Status Flush() = 0;
   virtual Status Sync() = 0;
 
+  // Get a name for the file, only for error reporting
+  virtual std::string GetName() const = 0;
+
  private:
   // No copying allowed
   WritableFile(const WritableFile&);

table/format.cc

@@ -82,7 +82,7 @@ Status ReadBlock(RandomAccessFile* file,
   }
   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
@@ -92,7 +92,7 @@ Status ReadBlock(RandomAccessFile* file,
     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;
     }
   }
@@ -119,13 +119,13 @@ Status ReadBlock(RandomAccessFile* file,
       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);
@@ -135,7 +135,7 @@ Status ReadBlock(RandomAccessFile* file,
     }
     default:
       delete[] buf;
-      return Status::Corruption("bad block type");
+      return Status::Corruption("bad block type", file->GetName());
   }
 
   return Status::OK();

util/env_posix.cc

@@ -121,6 +121,8 @@ class PosixSequentialFile: public SequentialFile {
     }
     return Status::OK();
   }
+
+  virtual std::string GetName() const { return filename_; }
 };
 
 // pread() based random-access
@@ -172,6 +174,8 @@ class PosixRandomAccessFile: public RandomAccessFile {
     }
     return s;
   }
+
+  virtual std::string GetName() const { return filename_; }
 };
 
 // mmap() based random-access
@@ -206,6 +210,8 @@ class PosixMmapReadableFile: public RandomAccessFile {
     }
     return s;
   }
+
+  virtual std::string GetName() const { return filename_; }
 };
 
 class PosixWritableFile : public WritableFile {
@@ -287,6 +293,8 @@ class PosixWritableFile : public WritableFile {
     }
     return s;
   }
+
+  virtual std::string GetName() const { return filename_; }
 };
 
 static int LockOrUnlock(int fd, bool lock) {

util/env_win.cc

@@ -78,6 +78,7 @@ class Win32SequentialFile : public SequentialFile
     virtual Status Read(size_t n, Slice* result, char* scratch);
     virtual Status Skip(uint64_t n);
     BOOL isEnable();
+    virtual std::string GetName() const { return _filename; }
 private:
     BOOL _Init();
     void _CleanUp();
@@ -94,6 +95,7 @@ class Win32RandomAccessFile : public RandomAccessFile
     virtual ~Win32RandomAccessFile();
     virtual Status Read(uint64_t offset, size_t n, Slice* result,char* scratch) const;
     BOOL isEnable();
+    virtual std::string GetName() const { return _filename; }
 private:
     BOOL _Init(LPCWSTR path);
     void _CleanUp();
@@ -114,6 +116,7 @@ class Win32WritableFile : public WritableFile
     virtual Status Flush();
     virtual Status Sync();
     BOOL isEnable();
+    virtual std::string GetName() const { return filename_; }
 private:
     std::string filename_;
     ::HANDLE _hFile;