Fix O(k^2) behavior when begin is a prefix of end

ConflictSet.cpp

@@ -1187,25 +1187,34 @@ struct CheckRangeRightSide {
   bool step() {
     switch (phase) {
     case Search: {
-      if (n->maxVersion <= readVersion) {
-        ok = true;
-        return true;
-      }
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+      fprintf(
+          stderr,
+          "Search path: %s, searchPathLen: %d, prefixLen: %d, remaining: %s\n",
+          getSearchPathPrintable(n).c_str(), searchPathLen, prefixLen,
+          printable(remaining).c_str());
+#endif
+
+      assert(searchPathLen <= int(key.size()));
+
       if (remaining.size() == 0) {
-        assert(searchPathLen >= prefixLen);
-        ok = true;
-        return true;
+        return downLeftSpine();
       }
 
       if (searchPathLen >= prefixLen) {
+        if (n->entryPresent && n->entry.pointVersion > readVersion) {
+          ok = false;
+          return true;
+        }
+
         if (maxBetweenExclusive(n, -1, remaining[0]) > readVersion) {
           ok = false;
           return true;
         }
       }
 
-      if (searchPathLen >= prefixLen && searchPathLen < int(key.size()) &&
-          n->entryPresent && n->entry.pointVersion > readVersion) {
+      if (searchPathLen > prefixLen && n->entryPresent &&
+          n->entry.rangeVersion > readVersion) {
         ok = false;
         return true;
       }
@@ -1217,43 +1226,30 @@ struct CheckRangeRightSide {
         ++searchPathLen;
       } else {
         if (c >= 0) {
-          if (searchPathLen < prefixLen) {
-            n = getChildExists(n, c);
-            return downLeftSpine();
-          }
-          ok = true;
-          return true;
-        } else {
-          n = nextSibling(n);
+          n = getChildExists(n, c);
           return downLeftSpine();
+        } else {
+          return backtrack();
         }
       }
 
       if (n->partialKeyLen > 0) {
         int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
         for (int i = 0; i < commonLen; ++i) {
+          ++searchPathLen;
           auto c = n->partialKey[i] <=> remaining[i];
           if (c == 0) {
-            ++searchPathLen;
             continue;
           }
           if (c > 0) {
-            if (searchPathLen < prefixLen) {
-              return downLeftSpine();
-            }
-            if (n->entryPresent && n->entry.rangeVersion > readVersion) {
-              ok = false;
-              return true;
-            }
-            ok = true;
-            return true;
-          } else {
-            if (searchPathLen >= prefixLen && n->maxVersion > readVersion) {
-              ok = false;
-              return true;
-            }
-            n = nextSibling(n);
             return downLeftSpine();
+          } else {
+            if (searchPathLen > prefixLen && n->entryPresent &&
+                n->entry.rangeVersion > readVersion) {
+              ok = false;
+              return true;
+            }
+            return backtrack();
           }
         }
         if (commonLen == n->partialKeyLen) {
@@ -1261,15 +1257,7 @@ struct CheckRangeRightSide {
           remaining =
               remaining.subspan(commonLen, remaining.size() - commonLen);
         } else if (n->partialKeyLen > int(remaining.size())) {
-          if (searchPathLen < prefixLen) {
-            return downLeftSpine();
-          }
-          if (n->entryPresent && n->entry.rangeVersion > readVersion) {
-            ok = false;
-            return true;
-          }
-          ok = true;
-          return true;
+          return downLeftSpine();
         }
       }
     } break;
@@ -1285,6 +1273,30 @@ struct CheckRangeRightSide {
     }
     return false;
   }
+
+  bool backtrack() {
+    for (;;) {
+      if (searchPathLen > prefixLen && n->maxVersion > readVersion) {
+        ok = false;
+        return true;
+      }
+      if (n->parent == nullptr) {
+        ok = true;
+        return true;
+      }
+      auto next = getChildGeq(n->parent, n->parentsIndex + 1);
+      if (next < 0) {
+        searchPathLen -= 1 + n->partialKeyLen;
+        n = n->parent;
+      } else {
+        searchPathLen -= n->partialKeyLen;
+        n = getChildExists(n->parent, next);
+        searchPathLen += n->partialKeyLen;
+        return downLeftSpine();
+      }
+    }
+  }
+
   bool downLeftSpine() {
     phase = DownLeftSpine;
     if (n == nullptr) {
@@ -1330,16 +1342,10 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
   lcp -= consumed;
 
   if (lcp == int(begin.size())) {
-    for (int i = lcp; i < int(end.size()); ++i) {
-      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;
+    CheckRangeRightSide checkRangeRightSide{n, end, lcp, readVersion};
+    while (!checkRangeRightSide.step())
+      ;
+    return checkRangeRightSide.ok;
   }
 
   if (!checkRangeStartsWith(n, begin.subspan(0, lcp), begin[lcp], end[lcp],