add a pointer to the root node

Before the offset of the root node was hard-coded.
Now the offset of the pointer to the root node is hard-coded.
That allows us to replace the root node.

block-storage/src/main/java/org/lucares/pdb/diskstorage/DiskStorage.java

@@ -250,4 +250,8 @@ public class DiskStorage implements AutoCloseable {
 	public long size() throws IOException {
 		return fileChannel.size();
 	}
+
+	public int minAllocationSize() {
+		return FREE_LIST_NODE_SIZE;
+	}
 }

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

@@ -18,7 +18,7 @@ public class PersistentMap {
 
 	private static final Charset UTF8 = StandardCharsets.UTF_8;
 	static final int BLOCK_SIZE = 4096;
-	private static final int ROOT_NODE_OFFEST = 4096;
+	static final long NODE_OFFSET_TO_ROOT_NODE = 8;
 
 	private final DiskStorage diskStore;
 
@@ -29,14 +29,20 @@ public class PersistentMap {
 
 	private void initIfNew() throws IOException {
 		if (diskStore.size() < BLOCK_SIZE) {
-			// this map is new:
-			// 1. make sure that new blocks are aligned to the block size (for faster disk
+			final long nodeOffsetToRootNode = diskStore.allocateBlock(diskStore.minAllocationSize());
+			Preconditions.checkEqual(nodeOffsetToRootNode, NODE_OFFSET_TO_ROOT_NODE,
+					"The offset of the pointer to the root node must be at a well known location. "
+							+ "Otherwise we would not be able to find it in an already existing file.");
+
+			// 2. make sure that new blocks are aligned to the block size (for faster disk
 			// IO)
 			diskStore.ensureAlignmentForNewBlocks(BLOCK_SIZE);
 
-			// 2. initialize an empty root node
+			// 3. initialize an empty root node
 			final long blockOffset = diskStore.allocateBlock(BLOCK_SIZE);
-			assert blockOffset == ROOT_NODE_OFFEST : "offset was: " + blockOffset;
+
+			// 4. upate pointer to root node
+			writeNodeOffsetOfRootNode(blockOffset);
 		}
 	}
 
@@ -62,11 +68,13 @@ public class PersistentMap {
 	}
 
 	public byte[] put(final byte[] key, final byte[] value) throws IOException {
-		return insert(ROOT_NODE_OFFEST, key, value);
+		final long rootNodeOffset = readNodeOffsetOfRootNode();
+		return insert(rootNodeOffset, key, value);
 	}
 
 	public byte[] get(final byte[] key) throws IOException {
-		final NodeEntry entry = findNodeEntry(ROOT_NODE_OFFEST, key);
+		final long rootNodeOffset = readNodeOffsetOfRootNode();
+		final NodeEntry entry = findNodeEntry(rootNodeOffset, key);
 
 		return entry == null ? null : entry.getValue();
 	}
@@ -180,7 +188,8 @@ public class PersistentMap {
 	}
 
 	public void visitPreOrder(final VisitorCallback visitor) throws IOException {
-		visitPreOrderRecursively(ROOT_NODE_OFFEST, visitor, 0);
+		final long rootNodeOffset = readNodeOffsetOfRootNode();
+		visitPreOrderRecursively(rootNodeOffset, visitor, 0);
 	}
 
 	private void visitPreOrderRecursively(final long nodeOffset, final VisitorCallback visitor, final int depth)
@@ -196,4 +205,16 @@ public class PersistentMap {
 			}
 		}
 	}
+
+	private long readNodeOffsetOfRootNode() throws IOException {
+		final DiskBlock diskBlock = diskStore.getDiskBlock(NODE_OFFSET_TO_ROOT_NODE, diskStore.minAllocationSize());
+
+		return diskBlock.getByteBuffer().getLong(0);
+	}
+
+	private void writeNodeOffsetOfRootNode(final long newNodeOffsetToRootNode) throws IOException {
+		final DiskBlock diskBlock = diskStore.getDiskBlock(NODE_OFFSET_TO_ROOT_NODE, diskStore.minAllocationSize());
+		diskBlock.getByteBuffer().putLong(0, newNodeOffsetToRootNode);
+		diskBlock.force();
+	}
 }