perf: warm reads up fairly and visit incident edges without records

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

@@ -71,7 +71,7 @@ private interface Slice {
 
     private static final int BATCH = 1_000;
     private static final int SCAN_PASSES = 400;
-    private static final int SCAN_WARMUP = 40;
+    private static final long WARM_UP_NANOS = 2_000_000_000L;
     private static final int COMMITS = 5_000;
     private static final int ID_BYTES = 8;
     private static final int UPDATE_BYTES = ID_BYTES + 24;
@@ -169,11 +169,9 @@ private static void run(Map<String, String> options) throws Exception {
                 store.largeEdge(largeSize);
                 return largeSize;
             })));
-            for (int i = 0; i < SCAN_WARMUP; i++) {
-                store.scanLargeEdge();
-            }
-            results.add(reading(store, () -> measureReading("large.scan", "enumerate all members of the large hyperedge (x%d, after %d warm-up passes)"
-                    .formatted(SCAN_PASSES, SCAN_WARMUP), (long) SCAN_PASSES * largeSize, () -> {
+            warmUp(store::scanLargeEdge);
+            results.add(reading(store, () -> measureReading("large.scan", "enumerate all members of the large hyperedge (x%d, after a warm-up)"
+                    .formatted(SCAN_PASSES), (long) SCAN_PASSES * largeSize, () -> {
                 long total = 0;
                 for (int i = 0; i < SCAN_PASSES; i++) {
                     total += store.scanLargeEdge();
@@ -315,9 +313,16 @@ private static long batched(int length, Slice slice) {
         return total;
     }
 
+    private static void warmUp(LongSupplier pass) {
+        long deadline = System.nanoTime() + WARM_UP_NANOS;
+        do {
+            pass.getAsLong();
+        } while (System.nanoTime() < deadline);
+    }
+
     private static Result read(Store store, String workload, String description, int operations, Slice slice, boolean warm) {
         if (warm) {
-            batched(Math.max(BATCH, operations / 10), slice);
+            warmUp(() -> batched(operations, slice));
         }
         return reading(store, () -> measureReading(workload, description, operations, () -> batched(operations, slice)));
     }
@@ -337,9 +342,13 @@ private static Result reading(Store store, Supplier<Result> workload) {
 
     private static Result latency(String workload, String description, int operations, IntToLongFunction operation, boolean warm) {
         if (warm) {
-            for (int i = 0; i < operations / 10; i++) {
-                operation.applyAsLong(i);
-            }
+            warmUp(() -> {
+                long total = 0;
+                for (int i = 0; i < operations; i++) {
+                    total += operation.applyAsLong(i);
+                }
+                return total;
+            });
         }
         long[] nanos = new long[operations];
         long checksum = 0;

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

@@ -17,6 +17,7 @@
 
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.function.IntConsumer;
@@ -130,18 +131,18 @@ private List<Long> members(int[] indexes) {
     @Override
     public long incidence(int[] probes, int from, int to) {
         return database.read(reader -> {
-            long total = 0;
+            IdCount count = new IdCount();
             for (int i = from; i < to; i++) {
-                total += reader.incident(nodes[probes[i]]).filter(incident -> incident.edge() != 0).count();
+                reader.forEachIncidentEdge(nodes[probes[i]], count);
             }
-            return total;
+            return count.total;
         });
     }
 
     @Override
     public long members(int[] probes, int from, int to) {
         return database.read(reader -> {
-            MemberCount count = new MemberCount();
+            IdCount count = new IdCount();
             for (int i = from; i < to; i++) {
                 reader.forEachMemberId(edges[probes[i]], count);
             }
@@ -164,11 +165,13 @@ public long coMembership(int[][] pairs, int from, int to) {
     @Override
     public long twoHop(int[] probes, int from, int to) {
         return database.read(reader -> {
+            Reached reached = new Reached();
             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();
+                reached.size = 0;
+                reader.forEachIncidentEdge(start, edge -> reader.forEachMemberId(edge, reached));
+                total += reached.distinctExcept(start);
             }
             return total;
         });
@@ -200,7 +203,7 @@ public void largeEdge(int count) {
     @Override
     public long scanLargeEdge() {
         return database.read(reader -> {
-            MemberCount count = new MemberCount();
+            IdCount count = new IdCount();
             reader.forEachMemberId(large, count);
             return count.total;
         });
@@ -272,7 +275,31 @@ public void close() {
         database.close();
     }
 
-    private static final class MemberCount implements LongConsumer {
+    private static final class Reached implements LongConsumer {
+        long[] atoms = new long[64];
+        int size;
+
+        @Override
+        public void accept(long atom) {
+            if (size == atoms.length) {
+                atoms = Arrays.copyOf(atoms, size * 2);
+            }
+            atoms[size++] = atom;
+        }
+
+        long distinctExcept(long start) {
+            Arrays.sort(atoms, 0, size);
+            long distinct = 0;
+            for (int i = 0; i < size; i++) {
+                if (atoms[i] != start && (i == 0 || atoms[i] != atoms[i - 1])) {
+                    distinct++;
+                }
+            }
+            return distinct;
+        }
+    }
+
+    private static final class IdCount implements LongConsumer {
         long total;
 
         @Override

database/src/main/java/io/hstore/db/Reader.java

@@ -237,6 +237,12 @@ public Stream<Incident> incident(long atom) {
         return view.incident(atom).map(incident -> new Incident(incident.edge(), roles(incident.roleSet()), incident.locator()));
     }
 
+    public void forEachIncidentEdge(long atom, LongConsumer action) {
+        if (onlyDefaultTenant() || visible(atom)) {
+            view.forEachIncidentEdge(atom, action);
+        }
+    }
+
     public long position(Incident incident) {
         return view.atom(incident.edge())
                 .filter(record -> record instanceof AtomRecord.EdgeRecord edge && edge.kind() == EdgeKind.ORDERED)

database/src/test/java/io/hstore/db/DatabaseTest.java

@@ -136,6 +136,34 @@ void membersCarryRolesWeightsAndValidity() {
         });
     }
 
+    @Test
+    void incidentEdgeVisitorsSeeTheSameEdgesAsTheIncidenceStream() {
+        long[] people = database.write(writer -> {
+            long hub = writer.node("Person", "hub", Map.of("name", "Hub"));
+            long alice = writer.node("Person", "alice", Map.of("name", "Alice"));
+            long loner = writer.node("Person", "loner", Map.of("name", "Loner"));
+            for (int i = 0; i < 120; i++) {
+                long edge = writer.edge("Coauthorship", Map.of("year", 2000 + i));
+                writer.add(edge, hub, MemberSpec.roles("author"));
+                if (i % 40 == 0) {
+                    writer.add(edge, alice, MemberSpec.roles("lead"));
+                }
+            }
+            return new long[]{hub, alice, loner};
+        });
+        database.read(reader -> {
+            for (long person : people) {
+                LongStream.Builder visited = LongStream.builder();
+                reader.forEachIncidentEdge(person, visited::add);
+                assertEquals(reader.incident(person).map(Incident::edge).toList(), visited.build().boxed().toList());
+            }
+            LongStream.Builder hub = LongStream.builder();
+            reader.forEachIncidentEdge(people[0], hub::add);
+            assertEquals(120, hub.build().count());
+            return null;
+        });
+    }
+
     @Test
     void documentsAndJsonPathIndexes() {
         database.write(writer -> {

docs/benchmarks.md

@@ -42,18 +42,18 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000
 |---|---|---|---|
 | `ingest.nodes` | insert nodes, 1,000 per transaction | `Writer.node` | `graph.add(String)` in `transact` |
 | `ingest.edges` | insert hyperedges, 1,000 per transaction | `Writer.edge` + `Writer.load` | `graph.add(new HGPlainLink(…))` |
-| `read.incidence` | enumerate a node's incidence set, 1,000 probes per read transaction | `Reader.incident` | `getIncidenceSet(h).getSearchResult()` |
+| `read.incidence` | enumerate a node's incidence set, 1,000 probes per read transaction | `Reader.forEachIncidentEdge` | `getIncidenceSet(h).getSearchResult()` |
 | `read.members` | enumerate a hyperedge's members | `Reader.forEachMemberId` | `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.twohop` | count the distinct nodes that share a hyperedge with a node, 10,000 probes | `Reader.forEachIncidentEdge` + `Reader.forEachMemberId` | `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()` |
+| `latency.read` | one incidence lookup per read transaction, each timed on its own | `Reader.forEachIncidentEdge` | `getIncidenceSet(h).getSearchResult()` |
 | `write.update` | replace a node's value, 1,000 per transaction | `Writer.set` | `graph.replace` |
 | `latency.commit` | 5,000 write transactions of one update each, each timed on its own | `Writer.set` | `graph.replace` |
-| `mixed.read` | single incidence lookups from `threads − 1` threads for 10 s while `mixed.write` runs | `Reader.incident` | `getIncidenceSet(h).getSearchResult()` |
+| `mixed.read` | single incidence lookups from `threads − 1` threads for 10 s while `mixed.write` runs | `Reader.forEachIncidentEdge` | `getIncidenceSet(h).getSearchResult()` |
 | `mixed.write` | one thread committing 10 updates at a time on a fixed schedule of 1,000 updates/s | `Writer.set` | `graph.replace` |
 | `large.ingest` | one hyperedge containing every node | `Writer.load` | one `HGPlainLink` |
-| `large.scan` | enumerate the large hyperedge 400 times, after 40 warm-up passes | `Reader.forEachMemberId` | `getTargetAt` loop |
+| `large.scan` | enumerate the large hyperedge 400 times, after the warm-up described below | `Reader.forEachMemberId` | `getTargetAt` loop |
 | `large.probe` | test whether a node belongs to the large hyperedge | `View.incidence(edge, node)` (membership-tree lookup) | `getIncidenceSet(node).contains(edge)` |
 | `reopen` | close and reopen, including recovery | `HypergraphDatabase.open` | `HGEnvironment.get` |
 | `read.incidence.cold` | `read.incidence` immediately after reopening | | |
@@ -67,7 +67,12 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000
 | `disk.compacted` | bytes on disk after the compaction and a clean shutdown | | |
 | `recover` | open the store after the process ingesting into it was killed with SIGKILL | `HypergraphDatabase.open` | `HGEnvironment.get` |
 
-Read workloads run a warm-up pass over 10% of the probes before timing, except `read.incidence.cold`.
+Read workloads, `large.scan` and `latency.read` first repeat the whole workload for at least 2 seconds without
+timing it. The exception is `read.incidence.cold`. Both engines get the same warm-up. The numbers describe a server that has been running for
+a while, not one that has just started. With a warm-up over 10% of the probes, as earlier versions of this
+benchmark did, most of the timed pass still ran before the JIT had compiled the read path. The timings mostly
+showed how quickly each engine's code got compiled, and `read.members` came out about three times lower for
+HStore than its steady rate.
 
 The two `latency.*` workloads time every operation separately with `System.nanoTime()`. Each operation is a whole
 transaction, as it would be for a request to a server. The report adds a second table with p50, p99, p99.9 and the

docs/database/data-model.md

@@ -469,5 +469,6 @@ Optional.empty
 | `memberIds(edge)` | member ids only, as a `LongStream` in membership order, without building `Member` records |
 | `forEachMemberId(edge, action)` | the same ids handed to a `LongConsumer`, with nothing allocated per member |
 | `cardinality`, `degree`, `incident`, `position` | topology statistics and reverse lookup |
+| `forEachIncidentEdge(atom, action)` | the ids of the hyperedges containing an atom, handed to a `LongConsumer` without building `Incident` records |
 | `lookup`, `range` | property and JSON index access with exact re-check |
 | `similar(text, k, consistency)`, `similar(model, vector, k, consistency)`, `embedding` | [semantic](semantic.md) |

engine/src/main/java/io/hstore/engine/index/PostingIndex.java

@@ -18,6 +18,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.LongConsumer;
 import java.util.function.ToLongFunction;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -110,6 +111,19 @@ public Stream<Entry<P>> stream(Tree<Postings<P>> index, long key) {
         }).orElseGet(Stream::empty);
     }
 
+    public void forEachPosting(Tree<Postings<P>> index, long key, LongConsumer action) {
+        index.get(key).ifPresent(value -> {
+            switch (value) {
+                case Postings.Inline<P> inline -> {
+                    for (long posting : inline.keys()) {
+                        action.accept(posting);
+                    }
+                }
+                case Postings.Promoted<P>(Ref root) -> subtree(index, root).forEachKey(action);
+            }
+        });
+    }
+
     public Tree<P> tree(Tree<Postings<P>> index, long key) {
         return index.get(key).map(value -> switch (value) {
             case Postings.Inline<P>(long[] keys, List<P> values, long _, int _) -> {

engine/src/main/java/io/hstore/engine/tree/Leaf.java

@@ -9,6 +9,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.LongConsumer;
 
 public final class Leaf extends Node {
 
@@ -135,6 +136,12 @@ long key(int index) {
         return keys[index];
     }
 
+    void forEachKey(LongConsumer action) {
+        for (int i = 0; i < size; i++) {
+            action.accept(keys[i]);
+        }
+    }
+
     long[] impliedWords() {
         return values == null ? words : null;
     }

engine/src/main/java/io/hstore/engine/tree/Tree.java

@@ -219,10 +219,20 @@ public Comparator<? super Long> getComparator() {
     }
 
     public void forEachKey(LongConsumer action) {
+        if (root instanceof Ref.Inline(Leaf leaf)) {
+            leaf.forEachKey(action);
+            return;
+        }
         cursor().forEachRemainingKey(action);
     }
 
     public void forEachValue(Consumer<? super V> action) {
+        if (root instanceof Ref.Inline(Leaf leaf)) {
+            for (int i = 0, size = leaf.size(); i < size; i++) {
+                action.accept(schema.cast(leaf.value(i)));
+            }
+            return;
+        }
         for (TreeCursor<V> cursor = cursor(); cursor.hasNext(); cursor.advance()) {
             action.accept(cursor.value());
         }

engine/src/main/java/io/hstore/engine/txn/View.java

@@ -18,6 +18,7 @@
 
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.function.LongConsumer;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
@@ -90,6 +91,10 @@ public Stream<IncidentEdge> incident(long atom) {
         return Views.incident(lazyTree(EngineSlots.REVERSE), atom);
     }
 
+    public void forEachIncidentEdge(long atom, LongConsumer action) {
+        EngineSlots.REVERSE_INDEX.forEachPosting(lazyTree(EngineSlots.REVERSE), atom, action);
+    }
+
     public Tree<?> incidentTree(long atom) {
         return EngineSlots.REVERSE_INDEX.tree(lazyTree(EngineSlots.REVERSE), atom);
     }

engine/src/test/java/io/hstore/engine/InlineMembersTest.java

@@ -100,6 +100,7 @@ void orderedEdgesKeepBothTreesInline() {
                 return null;
             });
             assertEquals(List.of(BASE + 9, BASE + 2, BASE + 5), ordered(engine, edge));
+            assertEquals(List.of(BASE + 9, BASE + 2, BASE + 5), visited(engine, edge));
             engine.write(txn -> {
                 for (int i = 0; i < 40; i++) {
                     txn.insertAt(edge, 0, Incidence.of(BASE + 100 + i));
@@ -459,6 +460,15 @@ private static List<Long> ordered(StorageEngine engine, long edge) {
         return engine.read(snapshot -> snapshot.requireEdge(edge).stream().map(Incidence::member).toList());
     }
 
+    private static List<Long> visited(StorageEngine engine, long edge) {
+        List<Long> members = new ArrayList<>();
+        engine.read(snapshot -> {
+            snapshot.requireEdge(edge).forEachMemberId(members::add);
+            return null;
+        });
+        return members;
+    }
+
     private static void assertShapes(StorageEngine engine, Map<Long, List<Long>> expected, Map<Long, Class<? extends Ref>> shapes) {
         expected.forEach((edge, members) -> {
             assertEquals(members, members(engine, edge));

engine/src/test/java/io/hstore/engine/tree/TreeTest.java

@@ -317,7 +317,14 @@ void inlineTreesRoundTripInsideTheRecordThatHoldsThem() {
             Ref.Inline inline = (Ref.Inline) read.members();
             assertTrue(Ref.same(new Ref.Inline(members), inline));
             assertEquals(members.summary(), inline.summary());
-            assertEquals(List.of(10L, 20L, 30L), new Tree<>(TopologySchemas.SET_MEMBERS, NodeSource.ephemeral(4096), inline).keys().boxed().toList());
+            Tree<Incidence> tree = new Tree<>(TopologySchemas.SET_MEMBERS, NodeSource.ephemeral(4096), inline);
+            assertEquals(List.of(10L, 20L, 30L), tree.keys().boxed().toList());
+            LongStream.Builder keys = LongStream.builder();
+            tree.forEachKey(keys::add);
+            assertArrayEquals(new long[]{10, 20, 30}, keys.build().toArray());
+            List<Long> visited = new ArrayList<>();
+            tree.forEachValue(incidence -> visited.add(incidence.member()));
+            assertEquals(List.of(10L, 20L, 30L), visited);
         }
     }