the root node can have more than two children it it is an inner node

It is not yet possible to split inner nodes or the root node.

block-storage/src/main/java/org/lucares/pdb/map/NodeEntry.java

@@ -5,6 +5,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
+import java.util.function.Predicate;
 
 import org.lucares.collections.LongList;
 import org.lucares.pdb.blockstorage.intsequence.VariableByteEncoder;
@@ -30,6 +31,20 @@ class NodeEntry {
 		}
 	}
 
+	static final class KeyMatches implements Predicate<NodeEntry> {
+
+		private final byte[] key;
+
+		public KeyMatches(final byte[] key) {
+			this.key = key;
+		}
+
+		@Override
+		public boolean test(final NodeEntry t) {
+			return Arrays.equals(key, t.getKey());
+		}
+	}
+
 	public static final Comparator<NodeEntry> SORT_BY_KEY = (a, b) -> a.compare(b.getKey());
 
 	private final ValueType type;

block-storage/src/main/java/org/lucares/pdb/map/PersistentMap.java

@@ -3,7 +3,8 @@ package org.lucares.pdb.map;
 import java.io.IOException;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
-import java.util.List;
+import java.util.Collections;
+import java.util.Stack;
 
 import org.lucares.pdb.blockstorage.intsequence.VariableByteEncoder;
 import org.lucares.pdb.diskstorage.DiskBlock;
@@ -16,6 +17,10 @@ public class PersistentMap {
 		void visit(NodeEntry nodeEntry, int depth);
 	}
 
+	interface NodeVisitorCallback {
+		void visit(PersistentMapDiskNode node, int depth);
+	}
+
 	private static final Charset UTF8 = StandardCharsets.UTF_8;
 	static final int BLOCK_SIZE = 4096;
 	static final long NODE_OFFSET_TO_ROOT_NODE = 8;
@@ -69,7 +74,8 @@ public class PersistentMap {
 
 	public byte[] put(final byte[] key, final byte[] value) throws IOException {
 		final long rootNodeOffset = readNodeOffsetOfRootNode();
-		return insert(rootNodeOffset, key, value);
+		final Stack<PersistentMapDiskNode> parents = new Stack<>();
+		return insert(parents, rootNodeOffset, key, value);
 	}
 
 	public byte[] get(final byte[] key) throws IOException {
@@ -79,7 +85,8 @@ public class PersistentMap {
 		return entry == null ? null : entry.getValue();
 	}
 
-	private byte[] insert(final long nodeOffest, final byte[] key, final byte[] value) throws IOException {
+	private byte[] insert(final Stack<PersistentMapDiskNode> parents, final long nodeOffest, final byte[] key,
+			final byte[] value) throws IOException {
 		final PersistentMapDiskNode node = getNode(nodeOffest);
 
 		final var entry = node.getNodeEntryTo(key);
@@ -101,45 +108,55 @@ public class PersistentMap {
 			if (node.canAdd(key, value)) {
 				// insert in existing node
 				node.addKeyValue(key, value);
-				writeNode(nodeOffest, node);
+				writeNode(node);
 				return oldValue;
 			} else {
 				// add new node
 				// 1. split current node into A and B
-				splitNode(nodeOffest, node);
+				splitNode(parents, node);
 
 				// 2. insert the value
-				return insert(nodeOffest, key, value);
+				// start from the root, because we might have added a new root node
+				return put(key, value);
 			}
 		} else {
 			final long childNodeOffset = toNodeOffset(entry);
-			return insert(childNodeOffset, key, value);
+			parents.add(node);
+			return insert(parents, childNodeOffset, key, value);
 		}
 	}
 
-	private void splitNode(final long nodeOffest, final PersistentMapDiskNode node) throws IOException {
+	private void splitNode(final Stack<PersistentMapDiskNode> parents, final PersistentMapDiskNode node)
+			throws IOException {
 
-		final long newLeftBlockOffset = diskStore.allocateBlock(BLOCK_SIZE);
+		final long newBlockOffset = diskStore.allocateBlock(BLOCK_SIZE);
-		final long newRightBlockOffset = diskStore.allocateBlock(BLOCK_SIZE);
 
-		final List<NodeEntry> entries = node.getEntries();
+		final PersistentMapDiskNode newNode = node.split(newBlockOffset);
-		final var leftEntries = entries.subList(0, entries.size() / 2);
-		final var rightEntries = entries.subList(entries.size() / 2, entries.size());
 
-		final PersistentMapDiskNode leftNode = new PersistentMapDiskNode(leftEntries);
+		final PersistentMapDiskNode parent = parents.isEmpty() ? null : parents.pop();
-		final PersistentMapDiskNode rightNode = new PersistentMapDiskNode(rightEntries);
+		if (parent != null) {
+			final byte[] newNodeKey = newNode.getTopNodeEntry().getKey();
+			parent.addKeyNodePointer(newNodeKey, newBlockOffset);
 
-		node.clear();
+			final byte[] oldNodeKey = node.getTopNodeEntry().getKey();
+			parent.addKeyNodePointer(oldNodeKey, node.getNodeOffset());
+			writeNode(parent);
+		} else {
+			// has no parent -> create a new parent (the new parent will also be the new
+			// root)
+			final long newRootOffset = diskStore.allocateBlock(BLOCK_SIZE);
+			final PersistentMapDiskNode rootNode = new PersistentMapDiskNode(newRootOffset, Collections.emptyList());
+			final byte[] newNodeKey = newNode.getTopNodeEntry().getKey();
+			rootNode.addKeyNodePointer(newNodeKey, newBlockOffset);
 
-		final NodeEntry lastLeftEntry = leftEntries.get(leftEntries.size() - 1);
+			final byte[] oldNodeKey = node.getTopNodeEntry().getKey();
-		final NodeEntry lastRightEntry = rightEntries.get(rightEntries.size() - 1);
+			rootNode.addKeyNodePointer(oldNodeKey, node.getNodeOffset());
+			writeNode(rootNode);
+			writeNodeOffsetOfRootNode(newRootOffset);
+		}
 
-		node.addKeyNodePointer(lastLeftEntry.getKey(), newLeftBlockOffset);
+		writeNode(newNode);
-		node.addKeyNodePointer(lastRightEntry.getKey(), newRightBlockOffset);
+		writeNode(node);
-
-		writeNode(newLeftBlockOffset, leftNode);
-		writeNode(newRightBlockOffset, rightNode);
-		writeNode(nodeOffest, node);
 	}
 
 	private NodeEntry findNodeEntry(final long nodeOffest, final byte[] key) throws IOException {
@@ -168,11 +185,12 @@ public class PersistentMap {
 	private PersistentMapDiskNode getNode(final long nodeOffset) throws IOException {
 		final DiskBlock diskBlock = diskStore.getDiskBlock(nodeOffset, BLOCK_SIZE);
 		final byte[] buffer = diskBlock.getBuffer();
-		final PersistentMapDiskNode node = PersistentMapDiskNode.parse(buffer);
+		final PersistentMapDiskNode node = PersistentMapDiskNode.parse(nodeOffset, buffer);
 		return node;
 	}
 
-	private void writeNode(final long nodeOffest, final PersistentMapDiskNode node) throws IOException {
+	private void writeNode(final PersistentMapDiskNode node) throws IOException {
+		final long nodeOffest = node.getNodeOffset();
 		final DiskBlock diskBlock = diskStore.getDiskBlock(nodeOffest, BLOCK_SIZE);
 		final byte[] buffer = diskBlock.getBuffer();
 		final byte[] newBuffer = node.serialize();
@@ -183,17 +201,17 @@ public class PersistentMap {
 
 	public void print() throws IOException {
 
-		visitPreOrder((nodeEntry, depth) -> System.out.println("    ".repeat(depth) + nodeEntry));
+		visitNodeEntriesPreOrder((nodeEntry, depth) -> System.out.println("    ".repeat(depth) + nodeEntry));
 
 	}
 
-	public void visitPreOrder(final VisitorCallback visitor) throws IOException {
+	public void visitNodeEntriesPreOrder(final VisitorCallback visitor) throws IOException {
 		final long rootNodeOffset = readNodeOffsetOfRootNode();
-		visitPreOrderRecursively(rootNodeOffset, visitor, 0);
+		visitNodeEntriesPreOrderRecursively(rootNodeOffset, visitor, 0);
 	}
 
-	private void visitPreOrderRecursively(final long nodeOffset, final VisitorCallback visitor, final int depth)
+	private void visitNodeEntriesPreOrderRecursively(final long nodeOffset, final VisitorCallback visitor,
-			throws IOException {
+			final int depth) throws IOException {
 		final PersistentMapDiskNode node = getNode(nodeOffset);
 
 		for (final NodeEntry child : node.getEntries()) {
@@ -201,7 +219,25 @@ public class PersistentMap {
 			visitor.visit(child, depth);
 			if (child.isInnerNode()) {
 				final long childNodeOffset = VariableByteEncoder.decodeFirstValue(child.getValue());
-				visitPreOrderRecursively(childNodeOffset, visitor, depth + 1);
+				visitNodeEntriesPreOrderRecursively(childNodeOffset, visitor, depth + 1);
+			}
+		}
+	}
+
+	public void visitNodesPreOrder(final NodeVisitorCallback visitor) throws IOException {
+		final long rootNodeOffset = readNodeOffsetOfRootNode();
+		visitNodesPreOrderRecursively(rootNodeOffset, visitor, 0);
+	}
+
+	private void visitNodesPreOrderRecursively(final long nodeOffset, final NodeVisitorCallback visitor,
+			final int depth) throws IOException {
+		final PersistentMapDiskNode node = getNode(nodeOffset);
+
+		visitor.visit(node, depth);
+		for (final NodeEntry child : node.getEntries()) {
+			if (child.isInnerNode()) {
+				final long childNodeOffset = VariableByteEncoder.decodeFirstValue(child.getValue());
+				visitNodesPreOrderRecursively(childNodeOffset, visitor, depth + 1);
 			}
 		}
 	}

block-storage/src/main/java/org/lucares/pdb/map/PersistentMapDiskNode.java

@@ -14,24 +14,33 @@ import org.lucares.pdb.map.NodeEntry.ValueType;
  * ┏━━━┳━━━━━┳━━━━━┳━━━━━┳╸╺╸╺╸╺╸╺┳━━━━━━━━━━━┳━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━┓
  * ┃ 6 ┃ 5,6 ┃ 3,6 ┃ 3,2 ┃        ┃"ba"->"147"┃"foobar"->"467"┃"foobaz"->"value"┃
  * ┗━━━┻━━━━━┻━━━━━┻━━━━━┻╸╺╸╺╸╺╸╺┻━━━━━━━━━━━┻━━━━━━━━━━━━━━━┻━━━━━━━━━━━━━━━━━┛
+ *   │   │ │   │ │   │ └▶ size of the third last key ("ba" in this example)
+ *   │   │ │   │ │   └▶ size of the third last value ("147" in this example)
+ *   │   │ │   │ └▶ size of the second last key ("foobar" in this example)
+ *   │   │ │   └▶ size of the second last value ("467" in this example)
+ *   │   │ └▶ size of the last key ("foobaz" in this example)
+ *   │   └▶ size of the last value (the string "value" in this example)
+ *   └▶ number of entries * 2
  * </pre>
  */
 public class PersistentMapDiskNode {
 
 	private final List<NodeEntry> entries;
+	private final long nodeOffset;
 
-	public PersistentMapDiskNode(final List<NodeEntry> entries) {
+	public PersistentMapDiskNode(final long nodeOffset, final List<NodeEntry> entries) {
-		this.entries = entries;
+		this.nodeOffset = nodeOffset;
+		this.entries = new ArrayList<>(entries);
 	}
 
-	public static PersistentMapDiskNode parse(final byte[] data) {
+	public static PersistentMapDiskNode parse(final long nodeOffset, final byte[] data) {
 		if (data.length != PersistentMap.BLOCK_SIZE) {
 			throw new IllegalStateException(
 					"block size must be " + PersistentMap.BLOCK_SIZE + " but was " + data.length);
 		}
 
 		final List<NodeEntry> entries = NodeEntry.deserialize(data);
-		return new PersistentMapDiskNode(entries);
+		return new PersistentMapDiskNode(nodeOffset, entries);
 	}
 
 	public byte[] serialize() {
@@ -39,6 +48,10 @@ public class PersistentMapDiskNode {
 		return NodeEntry.serialize(entries);
 	}
 
+	public long getNodeOffset() {
+		return nodeOffset;
+	}
+
 	public NodeEntry getNodeEntryTo(final byte[] key) {
 
 		final NodeEntry result = null;
@@ -63,6 +76,9 @@ public class PersistentMapDiskNode {
 	}
 
 	public void addNode(final ValueType valueType, final byte[] key, final byte[] value) {
+
+		entries.removeIf(new NodeEntry.KeyMatches(key));
+
 		final NodeEntry entry = new NodeEntry(valueType, key, value);
 		entries.add(entry);
 		Collections.sort(entries, NodeEntry.SORT_BY_KEY);
@@ -93,4 +109,19 @@ public class PersistentMapDiskNode {
 		return String.join("\n", entries.stream().map(NodeEntry::toString).collect(Collectors.toList()));
 	}
 
+	public NodeEntry getTopNodeEntry() {
+		return entries.get(entries.size() - 1);
+	}
+
+	public PersistentMapDiskNode split(final long newBlockOffset) {
+
+		final var leftEntries = new ArrayList<>(entries.subList(0, entries.size() / 2));
+		final var rightEntries = new ArrayList<>(entries.subList(entries.size() / 2, entries.size()));
+
+		entries.clear();
+		entries.addAll(leftEntries);
+
+		return new PersistentMapDiskNode(newBlockOffset, rightEntries);
+	}
+
 }

block-storage/src/test/java/org/lucares/pdb/map/PersistentMapTest.java

@@ -1,6 +1,7 @@
 package org.lucares.pdb.map;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.HashMap;
@@ -78,13 +79,22 @@ public class PersistentMapTest {
 		try (final DiskStorage ds = new DiskStorage(file)) {
 			final PersistentMap map = new PersistentMap(ds);
 
-			map.visitPreOrder((nodeEntry, depth) -> {
+			map.visitNodeEntriesPreOrder((nodeEntry, depth) -> {
 				if (nodeEntry.isInnerNode()) {
 					System.out.println("    ".repeat(depth) + nodeEntry);
+				} else {
+					System.out.println("    ".repeat(depth) + nodeEntry);
 				}
 			});
 			final AtomicInteger counter = new AtomicInteger();
-			map.visitPreOrder((nodeEntry, depth) -> counter.addAndGet(nodeEntry.isInnerNode() ? 1 : 0));
+			map.visitNodeEntriesPreOrder((nodeEntry, depth) -> counter.addAndGet(nodeEntry.isInnerNode() ? 1 : 0));
+
+			System.out.println(" -------------");
+			map.visitNodesPreOrder((node, depth) -> {
+				final String key = new String(node.getTopNodeEntry().getKey(), StandardCharsets.UTF_8);
+				System.out.println("    ".repeat(depth) + node.getNodeOffset() + " " + key + " (children: "
+						+ node.getEntries().size() + ")");
+			});
 
 //			Assert.assertEquals(counter.get(), 3,
 //					"number of nodes should be small. Any number larger than 3 indicates, "