Merge Node4 with one child into child if favorable

CC #9

ConflictSet.cpp

@@ -263,8 +263,8 @@ struct Node256 : Node {
 
 // Bound memory usage following the analysis in the ART paper
 
-constexpr int kBytesPerKey = 86;
+constexpr int kBytesPerKey = 121;
-constexpr int kMinChildrenNode4 = 2;
+constexpr int kMinChildrenNode4 = 1;
 constexpr int kMinChildrenNode16 = 5;
 constexpr int kMinChildrenNode48 = 17;
 constexpr int kMinChildrenNode256 = 49;
@@ -713,13 +713,55 @@ Node *&getOrCreateChild(Node *&self, uint8_t index,
 }
 
 // TODO fuse into erase child so we don't need to repeat branches on type
-void maybeDownsize(Node *self, Node *&root, NodeAllocators *allocators) {
+void maybeDownsize(Node *self, Node *&root, NodeAllocators *allocators,
+                   ConflictSet::Impl *impl) {
   switch (self->type) {
   case Type::Node0:
     __builtin_unreachable(); // GCOVR_EXCL_LINE
   case Type::Node4:
-    if (self->numChildren < kMinChildrenNode4) {
+    if (self->numChildren < 2) {
+      if (!self->entryPresent) {
+        auto *self4 = (Node4 *)self;
+        auto *child = self4->children[0].child;
+        int minCapacity = self4->partialKeyLen + 1 + child->partialKeyLen;
+
+        if (minCapacity > child->partialKeyCapacity) {
+          // TODO resize child? It seems to be quite challenging to implement,
+          // since callers would now have to account for erase invalidating
+          // other nodes. We could lower kBytesPerKey by doing this though.
+          return;
+        }
+
         // Merge partial key with child
+#if DEBUG_VERBOSE && !defined(NDEBUG)
+        fprintf(stderr, "Merge %s into %s\n",
+                getSearchPathPrintable(self).c_str(),
+                getSearchPathPrintable(child).c_str());
+#endif
+
+        int64_t childMaxVersion = maxVersion(child, impl);
+
+        // Construct new partial key for child
+        memmove(child->partialKey() + self4->partialKeyLen + 1,
+                child->partialKey(), child->partialKeyLen);
+        memcpy(child->partialKey(), self4->partialKey(), self->partialKeyLen);
+        child->partialKey()[self4->partialKeyLen] = self4->index[0];
+        child->partialKeyLen += 1 + self4->partialKeyLen;
+
+        child->parent = self->parent;
+        child->parentsIndex = self->parentsIndex;
+        if (self->parent == nullptr) {
+          root = child;
+        } else {
+          getChildExists(self->parent, self->parentsIndex) = child;
+        }
+
+        // Max versions are stored in the parent, so we need to update it now
+        // that we have a new parent.
+        maxVersion(child, impl) = childMaxVersion;
+
+        allocators->node4.release(self4);
+      }
     }
     break;
   case Type::Node16:
@@ -800,7 +842,7 @@ void maybeDownsize(Node *self, Node *&root, NodeAllocators *allocators) {
 
 // Precondition - an entry for index must exist in the node
 void eraseChild(Node *self, uint8_t index, NodeAllocators *allocators,
-                Node *&root) {
+                Node *&root, ConflictSet::Impl *impl) {
   auto *child = getChildExists(self, index);
   switch (child->type) {
   case Type::Node0:
@@ -850,9 +892,9 @@ void eraseChild(Node *self, uint8_t index, NodeAllocators *allocators,
   --self->numChildren;
   if (self->numChildren == 0 && !self->entryPresent &&
       self->parent != nullptr) {
-    eraseChild(self->parent, self->parentsIndex, allocators, root);
+    eraseChild(self->parent, self->parentsIndex, allocators, root, impl);
   } else {
-    maybeDownsize(self, root, allocators);
+    maybeDownsize(self, root, allocators, impl);
   }
 }
 
@@ -1868,7 +1910,7 @@ void addWriteRange(Node *&root, int64_t oldestVersion,
     beginNode = nextLogical(beginNode);
     old->entryPresent = false;
     if (old->numChildren == 0 && old->parent != nullptr) {
-      eraseChild(old->parent, old->parentsIndex, allocators, root);
+      eraseChild(old->parent, old->parentsIndex, allocators, root, impl);
     }
   }
 }
@@ -1999,7 +2041,7 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
         assert(n->entry.rangeVersion <= oldestVersion);
         prev->entryPresent = false;
         if (prev->numChildren == 0 && prev->parent != nullptr) {
-          eraseChild(prev->parent, prev->parentsIndex, &allocators, root);
+          eraseChild(prev->parent, prev->parentsIndex, &allocators, root, this);
         }
       }