Implement checkRangeLeftSide

ConflictSet.cpp

@@ -1056,6 +1056,102 @@ downLeftSpine:
   }
 }
 
+// Return true if the max version among all keys that start with key[:prefixLen]
+// that are >= key is <= readVersion
+bool checkRangeLeftSide(Node *n, std::span<const uint8_t> key, int prefixLen,
+                        int64_t readVersion) {
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+  fprintf(stderr, "Check range left side from %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 n->maxVersion <= readVersion;
+    }
+
+    if (searchPathLen >= prefixLen) {
+      if (maxBetweenExclusive(n, remaining[0], 256) > 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, c, 256) > readVersion) {
+          return false;
+        }
+        n = getChildExists(n, c);
+        return n->maxVersion <= readVersion;
+      } 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 n->maxVersion <= readVersion;
+        } else {
+          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 n->maxVersion <= readVersion;
+      }
+    }
+  }
+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(),
@@ -1099,7 +1195,7 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
     return true;
   }
 
-  if (!checkRangeStartsWith(n, begin, readVersion)) {
+  if (!checkRangeLeftSide(n, begin, lcp + 1, readVersion)) {
     return false;
   }
 
@@ -1107,12 +1203,7 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
                             readVersion)) {
     return false;
   }
-  for (int i = lcp + 1; i < int(begin.size()); ++i) {
+
-    if (!checkRangeStartsWith(n, begin.subspan(0, i), int(begin[i]), 256,
-                              readVersion)) {
-      return false;
-    }
-  }
   for (int i = lcp + 1; i < int(end.size()); ++i) {
     if (!checkPointRead(n, end.subspan(0, i), readVersion)) {
       return false;