Benchmark two-hop reads

benchmarks/src/main/java/io/hstore/bench/Comparison.java

@@ -133,6 +133,9 @@ private static void run(Map<String, String> options) throws Exception {
                     (from, to) -> store.members(dataset.probeEdges(), from, to), true));
             results.add(read("read.comembership", "count hyperedges containing both atoms of a pair", dataset.probePairs().length,
                     (from, to) -> store.coMembership(dataset.probePairs(), from, to), true));
+            int[] twoHop = dataset.probeTwoHop();
+            results.add(read("read.twohop", "count the distinct nodes that share a hyperedge with a node", twoHop.length,
+                    (from, to) -> store.twoHop(twoHop, from, to), true));
             results.add(concurrent(store, dataset, threads));
             results.add(latency("latency.read", "one incidence lookup per read transaction", nodes.length,
                     i -> store.incidence(nodes, i, i + 1), true));

benchmarks/src/main/java/io/hstore/bench/Dataset.java

@@ -4,7 +4,7 @@
 import java.util.SplittableRandom;
 
 record Dataset(int nodes, int[][] edges, int[] probeNodes, int[] probeEdges, int[][] probePairs, int[] updates, int[] deletions,
-               int[][] removals) {
+               int[][] removals, int[] probeTwoHop) {
 
     private static final int NODES_PER_SCALE = 50_000;
     private static final int EDGES_PER_SCALE = 100_000;
@@ -14,6 +14,7 @@ record Dataset(int nodes, int[][] edges, int[] probeNodes, int[] probeEdges, int
     private static final int MAX_CARDINALITY = 32;
     private static final double DELETED_FRACTION = 0.2;
     private static final int REMOVALS = 20_000;
+    private static final int TWO_HOP_PROBES = 10_000;
 
     static Dataset generate(int scale, long seed) {
         SplittableRandom random = new SplittableRandom(seed);
@@ -43,7 +44,8 @@ static Dataset generate(int scale, long seed) {
                 .limit(REMOVALS)
                 .mapToObj(edge -> new int[]{edge, edges[edge][random.nextInt(edges[edge].length)]})
                 .toArray(int[][]::new);
-        return new Dataset(nodes, edges, probeNodes, probeEdges, pairs, updates, deletions, removals);
+        int[] probeTwoHop = random.ints(TWO_HOP_PROBES, 0, nodes).toArray();
+        return new Dataset(nodes, edges, probeNodes, probeEdges, pairs, updates, deletions, removals, probeTwoHop);
     }
 
     private static int[] shuffled(SplittableRandom random, int length) {

benchmarks/src/main/java/io/hstore/bench/HStoreStore.java

@@ -149,6 +149,19 @@ public long coMembership(int[][] pairs, int from, int to) {
         });
     }
 
+    @Override
+    public long twoHop(int[] probes, int from, int to) {
+        return database.read(reader -> {
+            long total = 0;
+            for (int i = from; i < to; i++) {
+                long start = nodes[probes[i]];
+                total += reader.incident(start).flatMapToLong(incident -> reader.memberIds(incident.edge()))
+                        .filter(atom -> atom != start).sorted().distinct().count();
+            }
+            return total;
+        });
+    }
+
     @Override
     public void update(int[] probes, int from, int to, int round) {
         database.write(writer -> {

benchmarks/src/main/java/io/hstore/bench/HyperGraphDbStore.java

@@ -18,6 +18,8 @@
 import org.hypergraphdb.storage.bje.BJEStorageImplementation;
 
 import java.nio.file.Path;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.Callable;
 
 final class HyperGraphDbStore implements Store {
@@ -157,6 +159,31 @@ public long coMembership(int[][] pairs, int from, int to) {
         });
     }
 
+    @Override
+    public long twoHop(int[] probes, int from, int to) {
+        return read(() -> {
+            long total = 0;
+            for (int i = from; i < to; i++) {
+                HGPersistentHandle start = nodes[probes[i]];
+                Set<HGPersistentHandle> reached = new HashSet<>();
+                HGSearchResult<HGHandle> incident = graph.getIncidenceSet(start).getSearchResult();
+                try {
+                    while (incident.hasNext()) {
+                        HGLink link = graph.get(incident.next());
+                        for (int t = 0; t < link.getArity(); t++) {
+                            reached.add(graph.getPersistentHandle(link.getTargetAt(t)));
+                        }
+                    }
+                } finally {
+                    incident.close();
+                }
+                reached.remove(start);
+                total += reached.size();
+            }
+            return total;
+        });
+    }
+
     @Override
     public void update(int[] probes, int from, int to, int round) {
         transact(() -> {

benchmarks/src/main/java/io/hstore/bench/Store.java

@@ -24,6 +24,8 @@ interface Store extends AutoCloseable {
 
     long coMembership(int[][] pairs, int from, int to);
 
+    long twoHop(int[] probes, int from, int to);
+
     void update(int[] nodes, int from, int to, int round);
 
     void largeEdge(int members);

docs/benchmarks.md

@@ -45,6 +45,7 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000
 | `read.incidence` | enumerate a node's incidence set, 1,000 probes per read transaction | `Reader.incident` | `getIncidenceSet(h).getSearchResult()` |
 | `read.members` | enumerate a hyperedge's members | `Reader.members` | `HGLink.getTargetAt` over `graph.get(h)` |
 | `read.comembership` | count hyperedges containing both nodes of a pair | `TreeAlgebra.intersectKeys` (leapfrog) | `hg.count(hg.and(hg.incident(a), hg.incident(b)))` |
+| `read.twohop` | count the distinct nodes that share a hyperedge with a node, 10,000 probes | `Reader.incident` + `Reader.memberIds` | `getIncidenceSet` + the link's targets |
 | `read.incidence.parallel` | `read.incidence` from one thread per core | same | same |
 | `latency.read` | one incidence lookup per read transaction, each timed on its own | `Reader.incident` | `getIncidenceSet(h).getSearchResult()` |
 | `write.update` | replace a node's value, 1,000 per transaction | `Writer.set` | `graph.replace` |
@@ -77,6 +78,13 @@ the writer quietly slowing down. The writer stops at the deadline even when it i
 keep up shows an achieved rate below 1,000 updates/s. Readers keep a uniform random sample of 200,000 latencies
 each, plus the exact maximum. Both results depend on timing, so *Results agree* shows "n/a".
 
+**Two hops.** `read.twohop` follows each probe node to its hyperedges and from there to their members. It runs
+before `large.ingest`, because once the hyperedge containing every node exists, every node is two hops from all
+the others. Each adapter uses plain incidence and member calls and its own id type: `long` ids, sorted and
+de-duplicated, for HStore, and a `HashSet` of persistent handles for HyperGraphDB, which has no numeric ids.
+HyperGraphDB's `HGBreadthFirstTraversal` isn't used, since it would measure that framework rather than the
+storage underneath.
+
 **Deletes.** The churn set is drawn from the same seed after everything else, so the rest of the dataset is
 unchanged. It has 20% of the hyperedges to delete and 20,000 others, each with at least three members, that lose
 one member. `disk` is recorded while the store is closed for a reopen, before the churn starts, so it still means