db: VersionEdit can understand the format introduced in PR 3488.

RocksDB's PR 3488 introduced a new format of `VersionEdit`
encoding which is not understandable for older versions.
Thus, the change broke forward compatibility and has been
reverted in PR 3762. Later, PR 3765 reiterated the concept
but in a way that does not provide compatibility with the very
short-living format of PR 3488.

This change tries to address the issue of accessing DBs
in format of 3488 by ignoring the special entries.

Fixes: http://tracker.ceph.com/issues/25146
Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>

diff --git a/db/version_edit.cc b/db/version_edit.cc
index ecadf6e39807a9336f3cf19f8042777563a12c62..0190e60a41fb0df966c8a907df244395489bc3d0 100644 (file)
--- a/db/version_edit.cc
+++ b/db/version_edit.cc
@@ -238,6 +238,24 @@ bool VersionEdit::GetLevel(Slice* input, int* level, const char** /*msg*/) {
   }
 }
 
+static bool is_pseudo_new_file_record_pr3488(
+  const int level,
+  const uint64_t number,
+  const uint64_t file_size,
+  InternalKey& smallest,
+  InternalKey& largest,
+  const bool has_min_log_number_to_keep_) {
+
+  if (level == 0 && number == 0 && file_size == 0 &&
+      has_min_log_number_to_keep_) {
+    InternalKey dummy_key(Slice("dummy_key"), 0ull, ValueType::kTypeValue);
+    return (*smallest.rep() == *dummy_key.rep() &&
+            *largest.rep() == *dummy_key.rep());
+  } else {
+    return false;
+  }
+}
+
 const char* VersionEdit::DecodeNewFile4From(Slice* input) {
   const char* msg = nullptr;
   int level;
@@ -303,6 +321,12 @@ const char* VersionEdit::DecodeNewFile4From(Slice* input) {
   } else {
     return "new-file4 entry";
   }
+  if (is_pseudo_new_file_record_pr3488(level, number, file_size,
+                                       f.smallest, f.largest,
+                                       has_min_log_number_to_keep_)) {
+    // Since this has nothing to do with NewFile, return immediately.
+    return nullptr;
+  }
   f.fd =
       FileDescriptor(number, path_id, file_size, smallest_seqno, largest_seqno);
   new_files_.push_back(std::make_pair(level, f));