Implement checkRangeRightSide

ConflictSet.cpp

@@ -1152,6 +1152,109 @@ downLeftSpine:
   }
 }
 
+// Return true if the max version among all keys that start with key[:prefixLen]
+// that are < key is <= readVersion
+bool checkRangeRightSide(Node *n, std::span<const uint8_t> key, int prefixLen,
+                         int64_t readVersion) {
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+  fprintf(stderr, "Check range right side to %s for keys starting with %s\n",
+          printable(key).c_str(), printable(key.subspan(0, prefixLen)).c_str());
+#endif
+  auto remaining = key;
+  int searchPathLen = 0;
+  for (;;) {
+    if (n->maxVersion <= readVersion) {
+      return true;
+    }
+    if (remaining.size() == 0) {
+      assert(searchPathLen >= prefixLen);
+      return true;
+    }
+
+    if (searchPathLen >= prefixLen) {
+      if (maxBetweenExclusive(n, -1, remaining[0]) > readVersion) {
+        return false;
+      }
+    }
+
+    if (searchPathLen >= prefixLen && searchPathLen < int(key.size()) &&
+        n->entryPresent && n->entry.pointVersion > readVersion) {
+      return false;
+    }
+
+    int c = getChildGeq(n, remaining[0]);
+    if (c == remaining[0]) {
+      n = getChildExists(n, c);
+      remaining = remaining.subspan(1, remaining.size() - 1);
+      ++searchPathLen;
+    } else {
+      if (c >= 0) {
+        if (searchPathLen < prefixLen) {
+          n = getChildExists(n, c);
+          goto downLeftSpine;
+        }
+        if (maxBetweenExclusive(n, -1, c) > readVersion) {
+          return false;
+        }
+        return true;
+      } else {
+        n = nextSibling(n);
+        goto downLeftSpine;
+      }
+    }
+
+    if (n->partialKeyLen > 0) {
+      int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
+      for (int i = 0; i < commonLen; ++i) {
+        auto c = n->partialKey[i] <=> remaining[i];
+        if (c == 0) {
+          ++searchPathLen;
+          continue;
+        }
+        if (c > 0) {
+          if (searchPathLen < prefixLen) {
+            goto downLeftSpine;
+          }
+          if (n->entryPresent && n->entry.rangeVersion > readVersion) {
+            return false;
+          }
+          return true;
+        } else {
+          if (searchPathLen >= prefixLen && n->maxVersion > readVersion) {
+            return false;
+          }
+          n = nextSibling(n);
+          goto downLeftSpine;
+        }
+      }
+      if (commonLen == n->partialKeyLen) {
+        // partial key matches
+        remaining = remaining.subspan(commonLen, remaining.size() - commonLen);
+      } else if (n->partialKeyLen > int(remaining.size())) {
+        if (searchPathLen < prefixLen) {
+          goto downLeftSpine;
+        }
+        if (n->entryPresent && n->entry.rangeVersion > readVersion) {
+          return false;
+        }
+        return true;
+      }
+    }
+  }
+downLeftSpine:
+  if (n == nullptr) {
+    return true;
+  }
+  for (;;) {
+    if (n->entryPresent) {
+      return n->entry.rangeVersion <= readVersion;
+    }
+    int c = getChildGeq(n, 0);
+    assert(c >= 0);
+    n = getChildExists(n, c);
+  }
+}
+
 bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
                     std::span<const uint8_t> end, int64_t readVersion) {
   int lcp = longestCommonPrefix(begin.data(), end.data(),
@@ -1204,14 +1307,10 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
     return false;
   }
 
-  for (int i = lcp + 1; i < int(end.size()); ++i) {
+  if (!checkRangeRightSide(n, end, lcp + 1, readVersion)) {
-    if (!checkPointRead(n, end.subspan(0, i), readVersion)) {
     return false;
   }
-    if (!checkRangeStartsWith(n, end.subspan(0, i), -1, end[i], readVersion)) {
+
-      return false;
-    }
-  }
   return true;
 }