add visitor that find all values by a prefix of the key

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

@@ -137,6 +137,34 @@ class NodeEntry {
 		return key.length - otherKey.length;
 	}
 
+	public boolean isPrefix(final byte[] keyPrefix) {
+
+		return compareKeyPrefix(keyPrefix) == 0;
+	}
+
+	/**
+	 * Same as {@link #compare(byte[])}, but return 0 if prefix is a prefix of the
+	 * key. {@link #compare(byte[])} return values >0 in that case, because key
+	 * is longer than the prefix.
+	 *
+	 * @param prefix the prefix
+	 * @return 0 if {@code prefix} is a prefix of the key otherwise the value is
+	 *         defined by {@link #compare(byte[])}
+	 */
+	public int compareKeyPrefix(final byte[] prefix) {
+
+		int i = 0;
+		while (i < key.length && i < prefix.length) {
+			if (key[i] != prefix[i]) {
+				return key[i] - prefix[i];
+			}
+			i++;
+		}
+
+		return key.length > prefix.length ? 0 : key.length - prefix.length;
+
+	}
+
 	public boolean equal(final byte[] otherKey) {
 		return compare(otherKey) == 0;
 	}

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

@@ -4,6 +4,9 @@ import java.io.IOException;
 import java.io.PrintStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Stack;
 
 import org.lucares.pdb.blockstorage.intsequence.VariableByteEncoder;
@@ -28,6 +31,10 @@ public class PersistentMap<K, V> {
 		void visit(PersistentMapDiskNode node, PersistentMapDiskNode parentNode, NodeEntry nodeEntry, int depth);
 	}
 
+	interface Visitor<K, V> {
+		void visit(K key, V value);
+	}
+
 	public interface EncoderDecoder<O> {
 		public byte[] encode(O object);
 
@@ -109,6 +116,12 @@ public class PersistentMap<K, V> {
 		}
 	}
 
+	public void putAllValues(final Map<K, V> map) throws IOException {
+		for (final Entry<K, V> e : map.entrySet()) {
+			putValue(e.getKey(), e.getValue());
+		}
+	}
+
 	public V putValue(final K key, final V value) throws IOException {
 		final byte[] encodedKey = keyEncoder.encode(key);
 		final byte[] encodedValue = valueEncoder.encode(value);
@@ -305,6 +318,44 @@ public class PersistentMap<K, V> {
 		}
 	}
 
+	enum VisitByPrefixMode {
+		FIND, ITERATE
+	}
+
+	public void visitValues(final K keyPrefix, final Visitor<K, V> visitor) throws IOException {
+		final byte[] encodedKeyPrefix = keyEncoder.encode(keyPrefix);
+
+		final long rootNodeOffset = readNodeOffsetOfRootNode();
+		iterateNodeEntryByPrefix(rootNodeOffset, encodedKeyPrefix, visitor);
+	}
+
+	private void iterateNodeEntryByPrefix(final long nodeOffest, final byte[] keyPrefix, final Visitor<K, V> visitor)
+			throws IOException {
+		final PersistentMapDiskNode node = getNode(nodeOffest);
+
+		// list of children that might contain a key with the keyPrefix
+		final List<NodeEntry> nodesForPrefix = node.getNodesByPrefix(keyPrefix);
+
+		for (final NodeEntry entry : nodesForPrefix) {
+
+			if (entry.isDataNode()) {
+				final int prefixCompareResult = entry.compareKeyPrefix(keyPrefix);
+				if (prefixCompareResult == 0) {
+
+					final K key = keyEncoder.decode(entry.getKey());
+					final V value = valueEncoder.decode(entry.getValue());
+					visitor.visit(key, value);
+					// System.out.println("--> " + key + "=" + value);
+				} else if (prefixCompareResult > 0) {
+					break;
+				}
+			} else {
+				final long childNodeOffset = toNodeOffset(entry);
+				iterateNodeEntryByPrefix(childNodeOffset, keyPrefix, visitor);
+			}
+		}
+	}
+
 	private long readNodeOffsetOfRootNode() throws IOException {
 		final DiskBlock diskBlock = diskStore.getDiskBlock(NODE_OFFSET_TO_ROOT_NODE, diskStore.minAllocationSize());
 
@@ -316,4 +367,5 @@ public class PersistentMap<K, V> {
 		diskBlock.getByteBuffer().putLong(0, newNodeOffsetToRootNode);
 		diskBlock.force();
 	}
+
 }

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

@@ -102,13 +102,32 @@ public class PersistentMapDiskNode {
 
 			if (entry.compare(key) >= 0) {
 				return entry;
-			} else {
-				// break;
 			}
 		}
 		return result;
 	}
 
+	public List<NodeEntry> getNodesByPrefix(final byte[] keyPrefix) {
+		final List<NodeEntry> result = new ArrayList<>();
+
+		for (final NodeEntry nodeEntry : entries) {
+			final int prefixCompareResult = nodeEntry.compareKeyPrefix(keyPrefix);
+			if (prefixCompareResult == 0) {
+				// add all entries where keyPrefix is a prefix of the key
+				result.add(nodeEntry);
+			} else if (prefixCompareResult > 0) {
+				// Only add the first entry where the keyPrefix is smaller (as defined by
+				// compareKeyPrefix) than the key.
+				// These are entries that might contain key with the keyPrefix. But only the
+				// first of those can really have such keys.
+				result.add(nodeEntry);
+				break;
+			}
+		}
+
+		return result;
+	}
+
 	public void addKeyValue(final byte[] key, final byte[] value) {
 		addNode(ValueType.VALUE_INLINE, key, value);
 	}