Log more information for the add file with overlapping range failure

Summary: crash_test sometimes fails, hitting the add file overlapping assert. Add information in info logs help us to find the bug.

Test Plan: Run all test suites. Do some manual tests to make sure printing is correct.

Reviewers: kradhakrishnan, yhchiang, anthony, IslamAbdelRahman, rven, igor

Reviewed By: igor

Subscribers: leveldb, dhruba

Differential Revision: https://reviews.facebook.net/D49017

db/version_builder.cc

@@ -82,6 +82,7 @@ class VersionBuilder::Rep {
   };
 
   const EnvOptions& env_options_;
+  Logger* info_log_;
   TableCache* table_cache_;
   VersionStorageInfo* base_vstorage_;
   LevelState* levels_;
@@ -89,9 +90,10 @@ class VersionBuilder::Rep {
   FileComparator level_nonzero_cmp_;
 
  public:
-  Rep(const EnvOptions& env_options, TableCache* table_cache,
+  Rep(const EnvOptions& env_options, Logger* info_log, TableCache* table_cache,
       VersionStorageInfo* base_vstorage)
       : env_options_(env_options),
+        info_log_(info_log),
         table_cache_(table_cache),
         base_vstorage_(base_vstorage) {
     levels_ = new LevelState[base_vstorage_->num_levels()];
@@ -335,15 +337,16 @@ class VersionBuilder::Rep {
     if (levels_[level].deleted_files.count(f->fd.GetNumber()) > 0) {
       // File is deleted: do nothing
     } else {
-      vstorage->AddFile(level, f);
+      vstorage->AddFile(level, f, info_log_);
     }
   }
 };
 
 VersionBuilder::VersionBuilder(const EnvOptions& env_options,
                                TableCache* table_cache,
-                               VersionStorageInfo* base_vstorage)
-    : rep_(new Rep(env_options, table_cache, base_vstorage)) {}
+                               VersionStorageInfo* base_vstorage,
+                               Logger* info_log)
+    : rep_(new Rep(env_options, info_log, table_cache, base_vstorage)) {}
 VersionBuilder::~VersionBuilder() { delete rep_; }
 void VersionBuilder::CheckConsistency(VersionStorageInfo* vstorage) {
   rep_->CheckConsistency(vstorage);

db/version_builder.h

@@ -24,7 +24,7 @@ class InternalStats;
 class VersionBuilder {
  public:
   VersionBuilder(const EnvOptions& env_options, TableCache* table_cache,
-                 VersionStorageInfo* base_vstorage);
+                 VersionStorageInfo* base_vstorage, Logger* info_log = nullptr);
   ~VersionBuilder();
   void CheckConsistency(VersionStorageInfo* vstorage);
   void CheckConsistencyForDeletes(VersionEdit* edit, uint64_t number,

db/version_set.cc

@@ -524,7 +524,7 @@ class BaseReferencedVersionBuilder {
   explicit BaseReferencedVersionBuilder(ColumnFamilyData* cfd)
       : version_builder_(new VersionBuilder(
             cfd->current()->version_set()->env_options(), cfd->table_cache(),
-            cfd->current()->storage_info())),
+            cfd->current()->storage_info(), cfd->ioptions()->info_log)),
         version_(cfd->current()) {
     version_->Ref();
   }
@@ -1262,13 +1262,27 @@ bool CompareCompensatedSizeDescending(const Fsize& first, const Fsize& second) {
 }
 } // anonymous namespace
 
-void VersionStorageInfo::AddFile(int level, FileMetaData* f) {
+void VersionStorageInfo::AddFile(int level, FileMetaData* f, Logger* info_log) {
   auto* level_files = &files_[level];
   // Must not overlap
-  assert(level <= 0 || level_files->empty() ||
+#ifndef NDEBUG
+  if (level > 0 && !level_files->empty() &&
       internal_comparator_->Compare(
-             (*level_files)[level_files->size() - 1]->largest, f->smallest) <
-             0);
+          (*level_files)[level_files->size() - 1]->largest, f->smallest) >= 0) {
+    auto* f2 = (*level_files)[level_files->size() - 1];
+    if (info_log != nullptr) {
+      Error(info_log, "Adding new file %" PRIu64
+                      " range (%s, %s) to level %d but overlapping "
+                      "with existing file %" PRIu64 " %s %s",
+            f->fd.GetNumber(), f->smallest.DebugString(true).c_str(),
+            f->largest.DebugString(true).c_str(), level, f2->fd.GetNumber(),
+            f2->smallest.DebugString(true).c_str(),
+            f2->largest.DebugString(true).c_str());
+      LogFlush(info_log);
+    }
+    assert(false);
+  }
+#endif
   f->refs++;
   level_files->push_back(f);
 }

db/version_set.h

@@ -97,7 +97,7 @@ class VersionStorageInfo {
 
   void Reserve(int level, size_t size) { files_[level].reserve(size); }
 
-  void AddFile(int level, FileMetaData* f);
+  void AddFile(int level, FileMetaData* f, Logger* info_log = nullptr);
 
   void SetFinalized();