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>
(cherry picked from commit ffed5e68515620e1de0b66d96d2ec15cef1c82a7)

diff --git a/db/version_edit.cc b/db/version_edit.cc
index 5d171442109e1b09b606664fdffe7c3b10e5baf0..e45e8265685c19d4fb4b4362335fbe52eb09faf7 100644 (file)
--- a/db/version_edit.cc
+++ b/db/version_edit.cc
@@ -310,6 +310,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 = 0;
@@ -396,6 +414,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));