Add precondition to longestCommonPrefixPartialKey

The benchmark has this as faster

CMakeLists.txt

@@ -28,7 +28,8 @@ add_compile_options(-fdata-sections -ffunction-sections)
 include_directories(SYSTEM ${CMAKE_SOURCE_DIR}/third_party/valgrind)
 
 if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
-  add_compile_options(-Wno-maybe-uninitialized)
+  add_compile_options(-Wno-maybe-uninitialized
+                      $<$<COMPILE_LANGUAGE:CXX>:-Wno-invalid-offsetof>)
 endif()
 
 if(APPLE)

ConflictSet.cpp

@@ -21,6 +21,7 @@ limitations under the License.
 #include <bit>
 #include <cassert>
 #include <compare>
+#include <cstddef>
 #include <cstdint>
 #include <cstring>
 #include <inttypes.h>
@@ -89,45 +90,6 @@ private:
   void *freeList = nullptr;
 };
 
-enum class Type : int8_t {
-  Node4,
-  Node16,
-  Node48,
-  Node256,
-  Invalid,
-};
-struct Node {
-  /* begin section that's copied to the next node */
-  Node *parent = nullptr;
-  // The max write version over all keys that start with the search path up to
-  // this point
-  int64_t maxVersion;
-  Entry entry;
-  int16_t numChildren = 0;
-  bool entryPresent = false;
-  uint8_t parentsIndex = 0;
-  constexpr static auto kPartialKeyMaxLen = 26;
-  uint8_t partialKey[kPartialKeyMaxLen];
-  int8_t partialKeyLen = 0;
-  /* end section that's copied to the next node */
-
-  Type type = Type::Invalid;
-};
-
-struct Node4 : Node {
-  // Sorted
-  uint8_t index[4] = {};
-  Node *children[4] = {};
-  Node4() { this->type = Type::Node4; }
-};
-
-struct Node16 : Node {
-  // Sorted
-  uint8_t index[16] = {};
-  Node *children[16] = {};
-  Node16() { this->type = Type::Node16; }
-};
-
 struct BitSet {
   bool test(int i) const {
     assert(0 <= i);
@@ -185,9 +147,49 @@ private:
   __uint128_t hi = 0;
 };
 
+enum class Type : int8_t {
+  Node4,
+  Node16,
+  Node48,
+  Node256,
+  Invalid,
+};
+struct Node {
+  /* begin section that's copied to the next node */
+  Node *parent = nullptr;
+  // The max write version over all keys that start with the search path up to
+  // this point
+  int64_t maxVersion;
+  Entry entry;
+  int16_t numChildren = 0;
+  bool entryPresent = false;
+  uint8_t parentsIndex = 0;
+  constexpr static auto kPartialKeyMaxLen = 26;
+  uint8_t partialKey[kPartialKeyMaxLen];
+  int8_t partialKeyLen = 0;
+  /* end section that's copied to the next node */
+
+  Type type = Type::Invalid;
+};
+
+struct Node4 : Node {
+  // Sorted
+  uint8_t index[16]; // 16 so that we can use the same simd index search
+                     // implementation for Node4 as Node16
+  Node *children[4];
+  Node4() { this->type = Type::Node4; }
+};
+
+struct Node16 : Node {
+  // Sorted
+  uint8_t index[16];
+  Node *children[16];
+  Node16() { this->type = Type::Node16; }
+};
+
 struct Node48 : Node {
   BitSet bitSet;
-  Node *children[48] = {};
+  Node *children[48];
   int8_t nextFree = 0;
   int8_t index[256];
   Node48() {
@@ -209,15 +211,6 @@ struct NodeAllocators {
   BoundedFreeListAllocator<Node256> node256;
 };
 
-int getNodeIndex(Node4 *self, uint8_t index) {
-  for (int i = 0; i < self->numChildren; ++i) {
-    if (self->index[i] == index) {
-      return i;
-    }
-  }
-  return -1;
-}
-
 int getNodeIndex(Node16 *self, uint8_t index) {
 #ifdef HAS_AVX
   // Based on https://www.the-paper-trail.org/post/art-paper-notes/
@@ -278,10 +271,7 @@ int getNodeIndex(Node16 *self, uint8_t index) {
 
 // Precondition - an entry for index must exist in the node
 Node *&getChildExists(Node *self, uint8_t index) {
-  if (self->type == Type::Node4) {
-    auto *self4 = static_cast<Node4 *>(self);
-    return self4->children[getNodeIndex(self4, index)];
-  } else if (self->type == Type::Node16) {
+  if (self->type <= Type::Node16) {
     auto *self16 = static_cast<Node16 *>(self);
     return self16->children[getNodeIndex(self16, index)];
   } else if (self->type == Type::Node48) {
@@ -295,21 +285,32 @@ Node *&getChildExists(Node *self, uint8_t index) {
   __builtin_unreachable(); // GCOVR_EXCL_LINE
 }
 
+Node *getChild(Node *self, uint8_t index) {
+  if (self->type <= Type::Node16) {
+    auto *self16 = static_cast<Node16 *>(self);
+    int i = getNodeIndex(self16, index);
+    if (i >= 0) {
+      return self16->children[i];
+    }
+    return nullptr;
+  } else if (self->type == Type::Node48) {
+    auto *self48 = static_cast<Node48 *>(self);
+    int secondIndex = self48->index[index];
+    if (secondIndex >= 0) {
+      return self48->children[secondIndex];
+    }
+    return nullptr;
+  } else {
+    auto *self256 = static_cast<Node256 *>(self);
+    return self256->children[index];
+  }
+}
+
 int getChildGeq(Node *self, int child) {
   if (child > 255) {
     return -1;
   }
-  if (self->type == Type::Node4) {
-    auto *self4 = static_cast<Node4 *>(self);
-    for (int i = 0; i < self->numChildren; ++i) {
-      if (i > 0) {
-        assert(self4->index[i - 1] < self4->index[i]);
-      }
-      if (self4->index[i] >= child) {
-        return self4->index[i];
-      }
-    }
-  } else if (self->type == Type::Node16) {
+  if (self->type <= Type::Node16) {
     auto *self16 = static_cast<Node16 *>(self);
 #ifdef HAS_AVX
     __m128i key_vec = _mm_set1_epi8(child);
@@ -363,12 +364,10 @@ int getChildGeq(Node *self, int child) {
       }
     }
 #endif
-  } else if (self->type == Type::Node48) {
+  } else {
+    static_assert(offsetof(Node48, bitSet) == offsetof(Node256, bitSet));
     auto *self48 = static_cast<Node48 *>(self);
     return self48->bitSet.firstSetGeq(child);
-  } else {
-    auto *self256 = static_cast<Node256 *>(self);
-    return self256->bitSet.firstSetGeq(child);
   }
   return -1;
 }
@@ -386,7 +385,7 @@ Node *&getOrCreateChild(Node *&self, uint8_t index,
   if (self->type == Type::Node4) {
     auto *self4 = static_cast<Node4 *>(self);
     {
-      int i = getNodeIndex(self4, index);
+      int i = getNodeIndex((Node16 *)self4, index);
       if (i >= 0) {
         return self4->children[i];
       }
@@ -517,15 +516,7 @@ void eraseChild(Node *self, uint8_t index, NodeAllocators *allocators) {
     __builtin_unreachable(); // GCOVR_EXCL_LINE
   }
 
-  if (self->type == Type::Node4) {
-    auto *self4 = static_cast<Node4 *>(self);
-    int nodeIndex = getNodeIndex(self4, index);
-    memmove(self4->index + nodeIndex, self4->index + nodeIndex + 1,
-            sizeof(self4->index[0]) * (self->numChildren - (nodeIndex + 1)));
-    memmove(self4->children + nodeIndex, self4->children + nodeIndex + 1,
-            sizeof(self4->children[0]) * // NOLINT
-                (self->numChildren - (nodeIndex + 1)));
-  } else if (self->type == Type::Node16) {
+  if (self->type <= Type::Node16) {
     auto *self16 = static_cast<Node16 *>(self);
     int nodeIndex = getNodeIndex(self16, index);
     memmove(self16->index + nodeIndex, self16->index + nodeIndex + 1,
@@ -599,221 +590,6 @@ Node *nextSibling(Node *node) {
   }
 }
 
-// Performs a physical search for remaining
-struct SearchStepWise {
-  Node *n;
-  std::span<const uint8_t> remaining;
-
-  SearchStepWise() {}
-  SearchStepWise(Node *n, std::span<const uint8_t> remaining)
-      : n(n), remaining(remaining) {
-    assert(n->partialKeyLen == 0);
-  }
-
-  bool step() {
-    if (remaining.size() == 0) {
-      return true;
-    } else {
-      int c = getChildGeq(n, remaining[0]);
-      if (c == remaining[0]) {
-        auto *child = getChildExists(n, c);
-        int i = 0;
-        for (; i < child->partialKeyLen; ++i) {
-          if (!(i + 1 < int(remaining.size()) &&
-                remaining[i + 1] == child->partialKey[i])) {
-            break;
-          }
-        }
-        if (i != child->partialKeyLen) {
-          return true;
-        }
-        n = child;
-        remaining =
-            remaining.subspan(1 + child->partialKeyLen,
-                              remaining.size() - (1 + child->partialKeyLen));
-      } else {
-        return true;
-      }
-    }
-    return false;
-  }
-};
-
-namespace {
-std::string getSearchPathPrintable(Node *n);
-}
-
-// Logically this is the same as performing firstGeq and then checking against
-// point or range version according to cmp, but this version short circuits as
-// soon as it can prove that there's no conflict.
-bool checkPointRead(Node *n, const std::span<const uint8_t> key,
-                    int64_t readVersion) {
-#if DEBUG_VERBOSE && !defined(NDEBUG)
-  fprintf(stderr, "Check point read: %s\n", printable(key).c_str());
-#endif
-  auto remaining = key;
-  for (;;) {
-    if (n->maxVersion <= readVersion) {
-      return true;
-    }
-    if (remaining.size() == 0) {
-      if (n->entryPresent) {
-        return n->entry.pointVersion <= readVersion;
-      }
-      int c = getChildGeq(n, 0);
-      assert(c >= 0);
-      n = getChildExists(n, c);
-      goto downLeftSpine;
-    } else {
-      int c = getChildGeq(n, remaining[0]);
-      if (c == remaining[0]) {
-        n = getChildExists(n, c);
-        remaining = remaining.subspan(1, remaining.size() - 1);
-      } else {
-        if (c >= 0) {
-          n = getChildExists(n, c);
-          goto downLeftSpine;
-        } 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) {
-          continue;
-        }
-        if (c > 0) {
-          goto downLeftSpine;
-        } 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())) {
-        // n is the first physical node greater than remaining, and there's no
-        // eq node
-        goto downLeftSpine;
-      }
-    }
-  }
-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);
-  }
-}
-
-int64_t maxBetweenExclusive(Node *n, int begin, int end) {
-  assert(-1 <= begin);
-  assert(begin <= 256);
-  assert(-1 <= end);
-  assert(end <= 256);
-  assert(begin < end);
-  int64_t result = std::numeric_limits<int64_t>::lowest();
-  constexpr int kSparseThreshold = 32;
-  {
-    int c = getChildGeq(n, begin + 1);
-    if (c >= 0 && c < end) {
-      auto *child = getChildExists(n, c);
-      if (child->entryPresent) {
-        result = std::max(result, child->entry.rangeVersion);
-      }
-    }
-  }
-  switch (n->type) {
-  case Type::Node4: {
-    auto *self = static_cast<Node4 *>(n);
-    for (int i = 0; i < self->numChildren && self->index[i] < end; ++i) {
-      if (begin < self->index[i]) {
-        result = std::max(result, self->children[i]->maxVersion);
-      }
-    }
-    break;
-  }
-  case Type::Node16: {
-    auto *self = static_cast<Node16 *>(n);
-    for (int i = 0; i < self->numChildren && self->index[i] < end; ++i) {
-      if (begin < self->index[i]) {
-        result = std::max(result, self->children[i]->maxVersion);
-      }
-    }
-    break;
-  }
-  case Type::Node48: {
-    auto *self = static_cast<Node48 *>(n);
-    if (self->numChildren < kSparseThreshold) {
-      for (int i = self->bitSet.firstSetGeq(begin + 1); i < end && i >= 0;
-           i = self->bitSet.firstSetGeq(i + 1)) {
-        if (self->index[i] != -1) {
-          result = std::max(result, self->children[self->index[i]]->maxVersion);
-        }
-      }
-    } else {
-      for (int i = begin + 1; i < end; ++i) {
-        if (self->index[i] != -1) {
-          result = std::max(result, self->children[self->index[i]]->maxVersion);
-        }
-      }
-    }
-    break;
-  }
-  case Type::Node256: {
-    auto *self = static_cast<Node256 *>(n);
-    if (self->numChildren < kSparseThreshold) {
-      for (int i = self->bitSet.firstSetGeq(begin + 1); i < end && i >= 0;
-           i = self->bitSet.firstSetGeq(i + 1)) {
-        result = std::max(result, self->children[i]->maxVersion);
-      }
-    } else {
-      for (int i = begin + 1; i < end; ++i) {
-        if (self->children[i] != nullptr) {
-          result = std::max(result, self->children[i]->maxVersion);
-        }
-      }
-    }
-    break;
-  }
-  case Type::Invalid:
-    __builtin_unreachable(); // GCOVR_EXCL_LINE
-  }
-#if DEBUG_VERBOSE && !defined(NDEBUG)
-  fprintf(stderr, "At `%s', max version in (%02x, %02x) is %" PRId64 "\n",
-          getSearchPathPrintable(n).c_str(), begin, end, result);
-#endif
-  return result;
-}
-
-Vector<uint8_t> getSearchPath(Arena &arena, Node *n) {
-  assert(n != nullptr);
-  auto result = vector<uint8_t>(arena);
-  for (;;) {
-    for (int i = n->partialKeyLen - 1; i >= 0; --i) {
-      result.push_back(n->partialKey[i]);
-    }
-    if (n->parent == nullptr) {
-      break;
-    }
-    result.push_back(n->parentsIndex);
-    n = n->parent;
-  }
-  std::reverse(result.begin(), result.end());
-  return result;
-}
-
 #if defined(HAS_AVX) || defined(HAS_ARM_NEON)
 constexpr int kStride = 64;
 #else
@@ -850,7 +626,6 @@ bool compareStride(const uint8_t *ap, const uint8_t *bp) {
   return eq;
 }
 
-// Precondition: ap[0:kStride] != bp[0:kStride]
 int firstNeqStride(const uint8_t *ap, const uint8_t *bp) {
 #if defined(HAS_AVX)
   static_assert(kStride == 64);
@@ -888,6 +663,9 @@ int firstNeqStride(const uint8_t *ap, const uint8_t *bp) {
 }
 
 int longestCommonPrefix(const uint8_t *ap, const uint8_t *bp, int cl) {
+  if (cl < 0) {
+    __builtin_unreachable(); // GCOVR_EXCL_LINE
+  }
   int i = 0;
   int end;
 
@@ -961,6 +739,218 @@ bytes:
   return i;
 }
 
+int longestCommonPrefixPartialKey(const uint8_t *ap, const uint8_t *bp,
+                                  int cl) {
+  assert(cl <= Node::kPartialKeyMaxLen);
+  int i = 0;
+  for (; i < cl; ++i) {
+    if (*ap++ != *bp++) {
+      break;
+    }
+  }
+  return i;
+}
+
+// Performs a physical search for remaining
+struct SearchStepWise {
+  Node *n;
+  std::span<const uint8_t> remaining;
+
+  SearchStepWise() {}
+  SearchStepWise(Node *n, std::span<const uint8_t> remaining)
+      : n(n), remaining(remaining) {
+    assert(n->partialKeyLen == 0);
+  }
+
+  bool step() {
+    if (remaining.size() == 0) {
+      return true;
+    }
+    auto *child = getChild(n, remaining[0]);
+    if (child == nullptr) {
+      return true;
+    }
+    int cl = std::min<int>(child->partialKeyLen, remaining.size() - 1);
+    int i = longestCommonPrefixPartialKey(child->partialKey,
+                                          remaining.data() + 1, cl);
+    if (i != child->partialKeyLen) {
+      return true;
+    }
+    n = child;
+    remaining =
+        remaining.subspan(1 + child->partialKeyLen,
+                          remaining.size() - (1 + child->partialKeyLen));
+    return false;
+  }
+};
+
+namespace {
+std::string getSearchPathPrintable(Node *n);
+}
+
+// Logically this is the same as performing firstGeq and then checking against
+// point or range version according to cmp, but this version short circuits as
+// soon as it can prove that there's no conflict.
+bool checkPointRead(Node *n, const std::span<const uint8_t> key,
+                    int64_t readVersion) {
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+  fprintf(stderr, "Check point read: %s\n", printable(key).c_str());
+#endif
+  auto remaining = key;
+  for (;;) {
+    if (n->maxVersion <= readVersion) {
+      return true;
+    }
+    if (remaining.size() == 0) {
+      if (n->entryPresent) {
+        return n->entry.pointVersion <= readVersion;
+      }
+      int c = getChildGeq(n, 0);
+      assert(c >= 0);
+      n = getChildExists(n, c);
+      goto downLeftSpine;
+    }
+
+    int c = getChildGeq(n, remaining[0]);
+    if (c != remaining[0]) {
+      if (c >= 0) {
+        n = getChildExists(n, c);
+        goto downLeftSpine;
+      } else {
+        n = nextSibling(n);
+        goto downLeftSpine;
+      }
+    }
+
+    n = getChildExists(n, c);
+    remaining = remaining.subspan(1, remaining.size() - 1);
+
+    if (n->partialKeyLen > 0) {
+      int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
+      int i = longestCommonPrefix(n->partialKey, remaining.data(), commonLen);
+      if (i < commonLen) {
+        auto c = n->partialKey[i] <=> remaining[i];
+        if (c > 0) {
+          goto downLeftSpine;
+        } 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())) {
+        // n is the first physical node greater than remaining, and there's no
+        // eq node
+        goto downLeftSpine;
+      }
+    }
+  }
+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);
+  }
+}
+
+int64_t maxBetweenExclusive(Node *n, int begin, int end) {
+  assert(-1 <= begin);
+  assert(begin <= 256);
+  assert(-1 <= end);
+  assert(end <= 256);
+  assert(begin < end);
+  int64_t result = std::numeric_limits<int64_t>::lowest();
+  constexpr int kSparseThreshold = 32;
+  {
+    int c = getChildGeq(n, begin + 1);
+    if (c >= 0 && c < end) {
+      auto *child = getChildExists(n, c);
+      if (child->entryPresent) {
+        result = std::max(result, child->entry.rangeVersion);
+      }
+    }
+  }
+  switch (n->type) {
+  case Type::Node4:
+    [[fallthrough]];
+  case Type::Node16: {
+    auto *self = static_cast<Node16 *>(n);
+    for (int i = 0; i < self->numChildren && self->index[i] < end; ++i) {
+      if (begin < self->index[i]) {
+        result = std::max(result, self->children[i]->maxVersion);
+      }
+    }
+    break;
+  }
+  case Type::Node48: {
+    auto *self = static_cast<Node48 *>(n);
+    if (self->numChildren < kSparseThreshold) {
+      for (int i = self->bitSet.firstSetGeq(begin + 1); i < end && i >= 0;
+           i = self->bitSet.firstSetGeq(i + 1)) {
+        if (self->index[i] != -1) {
+          result = std::max(result, self->children[self->index[i]]->maxVersion);
+        }
+      }
+    } else {
+      for (int i = begin + 1; i < end; ++i) {
+        if (self->index[i] != -1) {
+          result = std::max(result, self->children[self->index[i]]->maxVersion);
+        }
+      }
+    }
+    break;
+  }
+  case Type::Node256: {
+    auto *self = static_cast<Node256 *>(n);
+    if (self->numChildren < kSparseThreshold) {
+      for (int i = self->bitSet.firstSetGeq(begin + 1); i < end && i >= 0;
+           i = self->bitSet.firstSetGeq(i + 1)) {
+        result = std::max(result, self->children[i]->maxVersion);
+      }
+    } else {
+      for (int i = begin + 1; i < end; ++i) {
+        if (self->children[i] != nullptr) {
+          result = std::max(result, self->children[i]->maxVersion);
+        }
+      }
+    }
+    break;
+  }
+  case Type::Invalid:
+    __builtin_unreachable(); // GCOVR_EXCL_LINE
+  }
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+  fprintf(stderr, "At `%s', max version in (%02x, %02x) is %" PRId64 "\n",
+          getSearchPathPrintable(n).c_str(), begin, end, result);
+#endif
+  return result;
+}
+
+Vector<uint8_t> getSearchPath(Arena &arena, Node *n) {
+  assert(n != nullptr);
+  auto result = vector<uint8_t>(arena);
+  for (;;) {
+    for (int i = n->partialKeyLen - 1; i >= 0; --i) {
+      result.push_back(n->partialKey[i]);
+    }
+    if (n->parent == nullptr) {
+      break;
+    }
+    result.push_back(n->parentsIndex);
+    n = n->parent;
+  }
+  std::reverse(result.begin(), result.end());
+  return result;
+}
+
 // Return true if the max version among all keys that start with key + [child],
 // where begin < child < end, is <= readVersion
 bool checkRangeStartsWith(Node *n, std::span<const uint8_t> key, int begin,
@@ -978,10 +968,7 @@ bool checkRangeStartsWith(Node *n, std::span<const uint8_t> key, int begin,
     }
 
     int c = getChildGeq(n, remaining[0]);
-    if (c == remaining[0]) {
-      n = getChildExists(n, c);
-      remaining = remaining.subspan(1, remaining.size() - 1);
-    } else {
+    if (c != remaining[0]) {
       if (c >= 0) {
         n = getChildExists(n, c);
         goto downLeftSpine;
@@ -991,13 +978,15 @@ bool checkRangeStartsWith(Node *n, std::span<const uint8_t> key, int begin,
       }
     }
 
+    n = getChildExists(n, c);
+    remaining = remaining.subspan(1, remaining.size() - 1);
+
     if (n->partialKeyLen > 0) {
       int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
-      for (int i = 0; i < commonLen; ++i) {
+      int i = longestCommonPrefixPartialKey(n->partialKey, remaining.data(),
+                                            commonLen);
+      if (i < commonLen) {
         auto c = n->partialKey[i] <=> remaining[i];
-        if (c == 0) {
-          continue;
-        }
         if (c > 0) {
           goto downLeftSpine;
         } else {
@@ -1077,11 +1066,7 @@ struct CheckRangeLeftSide {
       }
 
       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 != remaining[0]) {
         if (c >= 0) {
           if (searchPathLen < prefixLen) {
             n = getChildExists(n, c);
@@ -1096,14 +1081,16 @@ struct CheckRangeLeftSide {
         }
       }
 
+      n = getChildExists(n, c);
+      remaining = remaining.subspan(1, remaining.size() - 1);
+      ++searchPathLen;
+
       if (n->partialKeyLen > 0) {
         int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
-        for (int i = 0; i < commonLen; ++i) {
+        int i = longestCommonPrefix(n->partialKey, remaining.data(), commonLen);
+        searchPathLen += i;
+        if (i < commonLen) {
           auto c = n->partialKey[i] <=> remaining[i];
-          if (c == 0) {
-            ++searchPathLen;
-            continue;
-          }
           if (c > 0) {
             if (searchPathLen < prefixLen) {
               return downLeftSpine();
@@ -1220,11 +1207,7 @@ struct CheckRangeRightSide {
       }
 
       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 != remaining[0]) {
         if (c >= 0) {
           n = getChildExists(n, c);
           return downLeftSpine();
@@ -1233,14 +1216,18 @@ struct CheckRangeRightSide {
         }
       }
 
+      n = getChildExists(n, c);
+      remaining = remaining.subspan(1, remaining.size() - 1);
+      ++searchPathLen;
+
       if (n->partialKeyLen > 0) {
         int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
-        for (int i = 0; i < commonLen; ++i) {
+        int i = longestCommonPrefixPartialKey(n->partialKey, remaining.data(),
+                                              commonLen);
+        searchPathLen += i;
+        if (i < commonLen) {
           ++searchPathLen;
           auto c = n->partialKey[i] <=> remaining[i];
-          if (c == 0) {
-            continue;
-          }
           if (c > 0) {
             return downLeftSpine();
           } else {
@@ -1376,14 +1363,7 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
     break;
   }
 
-  if (!checkRangeLeftSide.ok) {
-    return false;
-  }
-  if (!checkRangeRightSide.ok) {
-    return false;
-  }
-
-  return true;
+  return checkRangeLeftSide.ok & checkRangeRightSide.ok;
 }
 
 // Returns a pointer to the newly inserted node.  caller is reponsible for
@@ -1396,10 +1376,10 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
   for (;;) {
     auto &self = *self_;
     // Handle an existing partial key
-    int partialKeyIndex = 0;
-    for (; partialKeyIndex < self->partialKeyLen; ++partialKeyIndex) {
-      if (partialKeyIndex == int(key.size()) ||
-          self->partialKey[partialKeyIndex] != key[partialKeyIndex]) {
+    int commonLen = std::min<int>(self->partialKeyLen, key.size());
+    int partialKeyIndex =
+        longestCommonPrefixPartialKey(self->partialKey, key.data(), commonLen);
+    if (partialKeyIndex < self->partialKeyLen) {
       auto *old = self;
       self = allocators->node4.allocate();
       self->maxVersion = old->maxVersion;
@@ -1416,8 +1396,6 @@ bool checkRangeRead(Node *n, std::span<const uint8_t> begin,
       memmove(old->partialKey, old->partialKey + partialKeyIndex + 1,
               old->partialKeyLen - (partialKeyIndex + 1));
       old->partialKeyLen -= partialKeyIndex + 1;
-        break;
-      }
     }
     key = key.subspan(partialKeyIndex, key.size() - partialKeyIndex);
 
@@ -1490,47 +1468,39 @@ void addPointWrite(Node *&root, int64_t oldestVersion,
 }
 
 void addWriteRange(Node *&root, int64_t oldestVersion,
-                   const ConflictSet::WriteRange &w,
-                   NodeAllocators *allocators) {
-
-  auto begin = std::span<const uint8_t>(w.begin.p, w.begin.len);
-  auto end = std::span<const uint8_t>(w.end.p, w.end.len);
+                   std::span<const uint8_t> begin, std::span<const uint8_t> end,
+                   int64_t writeVersion, NodeAllocators *allocators) {
 
   int lcp = longestCommonPrefix(begin.data(), end.data(),
                                 std::min(begin.size(), end.size()));
   if (lcp == int(begin.size()) && end.size() == begin.size() + 1 &&
       end.back() == 0) {
-    return addPointWrite(root, oldestVersion, begin, w.writeVersion,
-                         allocators);
+    return addPointWrite(root, oldestVersion, begin, writeVersion, allocators);
   }
   auto remaining = begin.subspan(0, lcp);
 
   auto *n = root;
 
   for (;;) {
-    if (int(remaining.size()) < n->partialKeyLen + 1) {
+    if (int(remaining.size()) <= n->partialKeyLen) {
       break;
     }
-    int i = 0;
-    for (; i < n->partialKeyLen; ++i) {
-      if (remaining[i] != n->partialKey[i]) {
-        break;
-      }
-    }
+    int i = longestCommonPrefixPartialKey(n->partialKey, remaining.data(),
+                                          n->partialKeyLen);
     if (i != n->partialKeyLen) {
       break;
     }
 
-    int c = getChildGeq(n, remaining[n->partialKeyLen]);
-    if (c != remaining[n->partialKeyLen]) {
+    auto *child = getChild(n, remaining[n->partialKeyLen]);
+    if (child == nullptr) {
       break;
     }
 
-    n->maxVersion = std::max(n->maxVersion, w.writeVersion);
+    n->maxVersion = std::max(n->maxVersion, writeVersion);
 
     remaining = remaining.subspan(n->partialKeyLen + 1,
                                   remaining.size() - (n->partialKeyLen + 1));
-    n = getChildExists(n, c);
+    n = child;
   }
 
   Node **useAsRoot = n->parent == nullptr
@@ -1542,7 +1512,7 @@ void addWriteRange(Node *&root, int64_t oldestVersion,
   begin = begin.subspan(consumed, begin.size() - consumed);
   end = end.subspan(consumed, end.size() - consumed);
 
-  auto *beginNode = insert(useAsRoot, begin, w.writeVersion, true, allocators);
+  auto *beginNode = insert(useAsRoot, begin, writeVersion, true, allocators);
 
   const bool insertedBegin = !std::exchange(beginNode->entryPresent, true);
 
@@ -1550,14 +1520,14 @@ void addWriteRange(Node *&root, int64_t oldestVersion,
     auto *p = nextLogical(beginNode);
     beginNode->entry.rangeVersion =
         p != nullptr ? p->entry.rangeVersion : oldestVersion;
-    beginNode->entry.pointVersion = w.writeVersion;
-    beginNode->maxVersion = w.writeVersion;
+    beginNode->entry.pointVersion = writeVersion;
+    beginNode->maxVersion = writeVersion;
   }
-  beginNode->maxVersion = std::max(beginNode->maxVersion, w.writeVersion);
+  beginNode->maxVersion = std::max(beginNode->maxVersion, writeVersion);
   beginNode->entry.pointVersion =
-      std::max(beginNode->entry.pointVersion, w.writeVersion);
+      std::max(beginNode->entry.pointVersion, writeVersion);
 
-  auto *endNode = insert(useAsRoot, end, w.writeVersion, false, allocators);
+  auto *endNode = insert(useAsRoot, end, writeVersion, false, allocators);
 
   const bool insertedEnd = !std::exchange(endNode->entryPresent, true);
 
@@ -1568,11 +1538,11 @@ void addWriteRange(Node *&root, int64_t oldestVersion,
     endNode->maxVersion =
         std::max(endNode->maxVersion, endNode->entry.pointVersion);
   }
-  endNode->entry.rangeVersion = w.writeVersion;
+  endNode->entry.rangeVersion = writeVersion;
 
   if (insertedEnd) {
     // beginNode may have been invalidated
-    beginNode = insert(useAsRoot, begin, w.writeVersion, true, allocators);
+    beginNode = insert(useAsRoot, begin, writeVersion, true, allocators);
   }
 
   for (beginNode = nextLogical(beginNode); beginNode != endNode;) {
@@ -1605,18 +1575,16 @@ struct FirstGeqStepwise {
   // Not being done implies that n is not the firstGeq
   bool step() {
     switch (phase) {
-    case Search:
+    case Search: {
       if (remaining.size() == 0) {
         int c = getChildGeq(n, 0);
         assert(c >= 0);
         n = getChildExists(n, c);
         return downLeftSpine();
-      } else {
+      }
+
       int c = getChildGeq(n, remaining[0]);
-        if (c == remaining[0]) {
-          n = getChildExists(n, c);
-          remaining = remaining.subspan(1, remaining.size() - 1);
-        } else {
+      if (c != remaining[0]) {
         if (c >= 0) {
           n = getChildExists(n, c);
           return downLeftSpine();
@@ -1625,14 +1593,16 @@ struct FirstGeqStepwise {
           return downLeftSpine();
         }
       }
-      }
+
+      n = getChildExists(n, c);
+      remaining = remaining.subspan(1, remaining.size() - 1);
+
       if (n->partialKeyLen > 0) {
         int commonLen = std::min<int>(n->partialKeyLen, remaining.size());
-        for (int i = 0; i < commonLen; ++i) {
+        int i = longestCommonPrefixPartialKey(n->partialKey, remaining.data(),
+                                              commonLen);
+        if (i < commonLen) {
           auto c = n->partialKey[i] <=> remaining[i];
-          if (c == 0) {
-            continue;
-          }
           if (c > 0) {
             return downLeftSpine();
           } else {
@@ -1650,6 +1620,7 @@ struct FirstGeqStepwise {
           return downLeftSpine();
         }
       }
+    }
       [[fallthrough]];
     case Init:
       phase = Search;
@@ -1697,19 +1668,14 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
 
   void check(const ReadRange *reads, Result *result, int count) const {
     for (int i = 0; i < count; ++i) {
+      const auto &r = reads[i];
+      auto begin = std::span<const uint8_t>(r.begin.p, r.begin.len);
+      auto end = std::span<const uint8_t>(r.end.p, r.end.len);
       result[i] =
           reads[i].readVersion < oldestVersion ? TooOld
-          : (reads[i].end.len > 0
-                 ? checkRangeRead(root,
-                                  std::span<const uint8_t>(reads[i].begin.p,
-                                                           reads[i].begin.len),
-                                  std::span<const uint8_t>(reads[i].end.p,
-                                                           reads[i].end.len),
-                                  reads[i].readVersion)
-                 : checkPointRead(root,
-                                  std::span<const uint8_t>(reads[i].begin.p,
-                                                           reads[i].begin.len),
-                                  reads[i].readVersion))
+          : (end.size() > 0
+                 ? checkRangeRead(root, begin, end, reads[i].readVersion)
+                 : checkPointRead(root, begin, reads[i].readVersion))
               ? Commit
               : Conflict;
     }
@@ -1718,14 +1684,15 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
   void addWrites(const WriteRange *writes, int count) {
     for (int i = 0; i < count; ++i) {
       const auto &w = writes[i];
+      auto begin = std::span<const uint8_t>(w.begin.p, w.begin.len);
+      auto end = std::span<const uint8_t>(w.end.p, w.end.len);
       if (w.end.len > 0) {
         keyUpdates += 2;
-        addWriteRange(root, oldestVersion, w, &allocators);
+        addWriteRange(root, oldestVersion, begin, end, w.writeVersion,
+                      &allocators);
       } else {
         keyUpdates += 1;
-        addPointWrite(root, oldestVersion,
-                      std::span<const uint8_t>(w.begin.p, w.begin.len),
-                      w.writeVersion, &allocators);
+        addPointWrite(root, oldestVersion, begin, w.writeVersion, &allocators);
       }
     }
   }