Read members and scan large edges without per-member objects

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

@@ -70,6 +70,8 @@ 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 int COMMITS = 5_000;
     private static final int ID_BYTES = 8;
     private static final int UPDATE_BYTES = ID_BYTES + 24;
@@ -167,10 +169,13 @@ private static void run(Map<String, String> options) throws Exception {
                 store.largeEdge(largeSize);
                 return largeSize;
             })));
-            store.scanLargeEdge();
-            results.add(reading(store, () -> measure("large.scan", "enumerate all members of the large hyperedge (x20)", 20L * 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, () -> {
                 long total = 0;
-                for (int i = 0; i < 20; i++) {
+                for (int i = 0; i < SCAN_PASSES; i++) {
                     total += store.scanLargeEdge();
                 }
                 return total;
@@ -295,6 +300,13 @@ private static Result measure(String workload, String description, long operatio
         return new Result(workload, description, operations, (System.nanoTime() - started) / 1e6, checksum).using(JvmUsage.now().since(usage));
     }
 
+    private static Result measureReading(String workload, String description, long operations, LongSupplier work) {
+        JvmUsage usage = JvmUsage.onThisThread();
+        long started = System.nanoTime();
+        long checksum = work.getAsLong();
+        return new Result(workload, description, operations, (System.nanoTime() - started) / 1e6, checksum).using(JvmUsage.onThisThread().since(usage));
+    }
+
     private static long batched(int length, Slice slice) {
         long total = 0;
         for (int from = 0; from < length; from += BATCH) {
@@ -307,7 +319,7 @@ private static Result read(Store store, String workload, String description, int
         if (warm) {
             batched(Math.max(BATCH, operations / 10), slice);
         }
-        return reading(store, () -> measure(workload, description, operations, () -> batched(operations, slice)));
+        return reading(store, () -> measureReading(workload, description, operations, () -> batched(operations, slice)));
     }
 
     private static Result writing(Store store, long logicalBytes, Supplier<Result> workload) {

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

@@ -20,6 +20,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.IntConsumer;
+import java.util.function.LongConsumer;
 
 final class HStoreStore implements Store {
 
@@ -140,11 +141,11 @@ public long incidence(int[] probes, int from, int to) {
     @Override
     public long members(int[] probes, int from, int to) {
         return database.read(reader -> {
-            long total = 0;
+            MemberCount count = new MemberCount();
             for (int i = from; i < to; i++) {
-                total += reader.memberIds(edges[probes[i]]).filter(atom -> atom != 0).count();
+                reader.forEachMemberId(edges[probes[i]], count);
             }
-            return total;
+            return count.total;
         });
     }
 
@@ -198,7 +199,11 @@ public void largeEdge(int count) {
 
     @Override
     public long scanLargeEdge() {
-        return database.read(reader -> reader.memberIds(large).filter(atom -> atom != 0).count());
+        return database.read(reader -> {
+            MemberCount count = new MemberCount();
+            reader.forEachMemberId(large, count);
+            return count.total;
+        });
     }
 
     @Override
@@ -266,4 +271,15 @@ public void reopen(Runnable whileClosed) {
     public void close() {
         database.close();
     }
+
+    private static final class MemberCount implements LongConsumer {
+        long total;
+
+        @Override
+        public void accept(long atom) {
+            if (atom != 0) {
+                total++;
+            }
+        }
+    }
 }

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

@@ -9,13 +9,22 @@
 record JvmUsage(long collections, long pauseMillis, long allocated) {
 
     static JvmUsage now() {
+        return now(true);
+    }
+
+    static JvmUsage onThisThread() {
+        return now(false);
+    }
+
+    private static JvmUsage now(boolean allThreads) {
         long collections = 0;
         long pauseMillis = 0;
         for (GarbageCollectorMXBean collector : ManagementFactory.getGarbageCollectorMXBeans()) {
             collections += Math.max(0, collector.getCollectionCount());
             pauseMillis += Math.max(0, collector.getCollectionTime());
         }
-        long allocated = ((com.sun.management.ThreadMXBean) ManagementFactory.getThreadMXBean()).getTotalThreadAllocatedBytes();
+        com.sun.management.ThreadMXBean threads = (com.sun.management.ThreadMXBean) ManagementFactory.getThreadMXBean();
+        long allocated = allThreads ? threads.getTotalThreadAllocatedBytes() : threads.getCurrentThreadAllocatedBytes();
         return new JvmUsage(collections, pauseMillis, allocated);
     }
 

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

@@ -32,6 +32,7 @@
 import java.util.OptionalInt;
 import java.util.OptionalLong;
 import java.util.Set;
+import java.util.function.LongConsumer;
 import java.util.function.Predicate;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
@@ -203,6 +204,10 @@ public LongStream memberIds(long edge) {
         return edge(edge).memberIds();
     }
 
+    public void forEachMemberId(long edge, LongConsumer action) {
+        edge(edge).forEachMemberId(action);
+    }
+
     public Stream<Member> members(long edge, long validAt) {
         return edge(edge).validAt(validAt).map(incidence -> member(incidence, -1));
     }

docs/architecture/overview.md

@@ -154,7 +154,7 @@ sequenceDiagram
 
 Notes:
 
-* Nodes are decoded once and then shared by every reader through `NodeCache`. It has 16 shards, each a `ConcurrentHashMap` plus a CLOCK (second-chance) ring, with a total byte budget of `node_cache_mb`, each node counted at an estimate of its heap size (`Node.heapBytes`). Lookups are lock-free; only admission takes a per-shard lock. Decoded nodes are immutable (`owner == null`), so sharing them needs no copying. See [pages.md](../storage/pages.md#caching-and-io-accounting).
+* Nodes are decoded once and then shared by every reader through `NodeCache`. It has 16 shards, each an open-addressing table keyed by page number plus a CLOCK (second-chance) ring, with a total byte budget of `node_cache_mb`, each node counted at an estimate of its heap size (`Node.heapBytes`). Lookups are lock-free; only admission takes a per-shard lock. Decoded nodes are immutable (`owner == null`), so sharing them needs no copying. See [pages.md](../storage/pages.md#caching-and-io-accounting).
 * `IoTrace` is bound with `ScopedValue.where(...)` around a query. `recordRead` throws `HStoreException.limit` once page reads plus cache hits exceed the query's page budget (`query_page_budget`).
 * Historical reads use `TransactionManager.snapshotAt(generation, branch)`. It looks the generation up in the retained `history` map; see [catalog-and-generations.md](../storage/catalog-and-generations.md#history-retention-and-time-travel).
 

docs/benchmarks.md

@@ -43,7 +43,7 @@ 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.members` | enumerate a hyperedge's members | `Reader.members` | `HGLink.getTargetAt` over `graph.get(h)` |
+| `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.incidence.parallel` | `read.incidence` from one thread per core | same | same |
@@ -53,7 +53,7 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000
 | `mixed.read` | single incidence lookups from `threads − 1` threads for 10 s while `mixed.write` runs | `Reader.incident` | `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 twenty times | `Reader.members` | `getTargetAt` loop |
+| `large.scan` | enumerate the large hyperedge 400 times, after 40 warm-up passes | `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 | | |
@@ -144,8 +144,9 @@ then the rest of the page image if it's bigger. JE reports sequential plus rando
 cold reads and with a small `CACHE_MB`. The logical sizes are rough estimates; compare the two stores against each
 other rather than reading the ratios as absolute numbers.
 
-**Memory.** Every workload also records, from the JVM's own MXBeans, the bytes the whole process allocated
-(`getTotalThreadAllocatedBytes`) and the GC pause time (`GarbageCollectorMXBean.getCollectionTime`, which is
+**Memory.** Every workload also records, from the JVM's own MXBeans, the bytes allocated
+(`getTotalThreadAllocatedBytes` for the whole process; for the single-threaded read workloads, only the reading
+thread's `getCurrentThreadAllocatedBytes`, so background maintenance doesn't count against a read) and the GC pause time (`GarbageCollectorMXBean.getCollectionTime`, which is
 stop-the-world time with ParallelGC). Rows where both stores paused for 0 ms are left out. After ingest, the run
 forces a full GC and records how much heap the loaded store keeps. A baseline taken before the store opens is
 subtracted, so the shared dataset arrays don't count. This number mostly reflects how each engine's cache is
@@ -244,8 +245,9 @@ not what drives HStore's footprint:
   atom cache, which holds a Java array of target handles, so a scan is an array loop. HStore walks a B+tree
   cursor, decodes incidence columns, resolves role sets and builds a `Member` record for each member. That
   record carries roles, weight, validity, position and qualifier. The cost buys per-member metadata and
-  snapshot-consistent reads, which a plain link cannot express. This is the clearest optimisation target
-  (a raw `LongStream` member path that skips record materialisation).
+  snapshot-consistent reads, which a plain link cannot express. The benchmark now reads ids through
+  `Reader.forEachMemberId`, which builds neither `Member` records nor tree entries; the numbers below predate that.
+  What remains is that every hyperedge keeps its members in a tree of its own, one more page to read.
 * **Edge ingest at small batch sizes (about 0.6–0.8× in async mode).** Each commit rewrites the incident-tree
   paths of every touched member, which is copy-on-write write amplification. JE appends log records and
   defers the work to its cleaner. In `sync` mode group commit amortises the fsync and HStore leads (1.35×).

docs/database/data-model.md

@@ -461,6 +461,7 @@ Optional.empty
 | `property(atom, name[, validAt])`, `properties`, `text(value)`, `document` | property access; `text` dereferences payloads |
 | `edge`, `members`, `members(edge, validAt)`, `membersWithRole`, `membersWeighted` | membership access with summary pruning |
 | `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 |
 | `lookup`, `range` | property and JSON index access with exact re-check |
 | `similar(text, k, consistency)`, `similar(model, vector, k, consistency)`, `embedding` | [semantic](semantic.md) |

docs/storage/pages.md

@@ -313,11 +313,11 @@ read-only memory, so nothing can write into a stored image through a read.
 
 The cache stores **decoded, frozen** `Node` objects, not bytes. A hit therefore costs no decoding and no allocation. The cache has 16 shards; a node goes to shard `mix(pageId) >>> 60`. Each shard has:
 
-* a `ConcurrentHashMap<Long, Slot>` from page id to `Slot(pageId, node, volatile referenced)`;
+* an open-addressing table of `Slot(pageId, node, volatile referenced)`, probed linearly from `mix(pageId)`, so a lookup needs no boxed key. It is kept at most half full, counting removed slots, and is rebuilt into a new array when it would pass that;
 * a byte budget of `max(64 KiB, node_cache_mb / 16)` and a circular, doubly linked ring of slots with a CLOCK hand. Each slot weighs `Node.heapBytes()`, an estimate of the decoded node's heap size: 144 bytes for the node and its summary, its key and value or child arrays, then 96 bytes plus the encoded value per leaf entry, or 128 bytes per child reference in a branch. Those constants come from measuring retained heap against cache contents at two dataset sizes;
 * a `ReentrantLock` taken only for admission.
 
-**Reads are lock-free.** `get` is a `ConcurrentHashMap.get`, plus setting `referenced = true` if it was clear. The check avoids a redundant volatile write on hot entries.
+**Reads are lock-free.** `get` probes the table, reading each slot with acquire ordering, plus setting `referenced = true` if it was clear. Writers change the table only under the shard lock, publishing each slot with release ordering, and a rebuild publishes the new array through a volatile field. A reader still probing the old array can find a node that was just evicted; that is harmless, because a page number's contents never change. The check avoids a redundant volatile write on hot entries.
 
 **Admission (`put`) uses CLOCK second-chance eviction under the shard lock:**
 

docs/storage/persistent-tree.md

@@ -35,7 +35,7 @@ Leaf                                   Branch
   Summary  summary   (volatile, recomputed lazily after mutation)
 ```
 
-**Routing.** `Branch.route(key)` binary-searches `separators[1..size-1]` for the last separator `<= key` and returns index 0 when there is none. `separators[0]` therefore never takes part in routing. A freshly grown root stores `Long.MIN_VALUE` there. When a branch is frozen (`Branch.frozenWith`), `separators[0]` is rewritten to the first child's `summary().min()`, so the encoded page carries its real lower bound and delta-encodes compactly.
+**Routing.** `Branch.route(key)` searches `separators[1..size-1]` for the last separator `<= key` and returns index 0 when there is none. In a branch with 16 or more children it first guesses the position from where `key` falls between the first and last separator, checks a window of 8 around the guess, and only then falls back to a binary search over what is left. Page numbers and atom ids are handed out in order, so the guess is usually right. `Leaf.search` does the same over a leaf's keys and returns exactly what `Arrays.binarySearch` would. `separators[0]` therefore never takes part in routing. A freshly grown root stores `Long.MIN_VALUE` there. When a branch is frozen (`Branch.frozenWith`), `separators[0]` is rewritten to the first child's `summary().min()`, so the encoded page carries its real lower bound and delta-encodes compactly.
 
 **Page budget.** `Layout.of(pageSize)` computes:
 
@@ -180,7 +180,9 @@ A `WriteScope` built with a `PageSink` (every `Transaction` has one) can spill c
 
 ### Cursors with summary pruning
 
-A `TreeCursor` (`TreeCursor.java`) keeps an explicit stack of up to 48 `(branch, slot)` pairs and is built with an `admit: Predicate<Summary>`. A child subtree is entered only if `admit.test(child.summary())`. Because the summary sits in the parent, a rejected subtree is skipped without I/O. Queries built on this:
+A `TreeCursor` (`TreeCursor.java`) keeps an explicit stack of `(branch, slot)` pairs, with room for four levels to start with and growing up to 48, and is built with an `admit: Predicate<Summary>`. A child subtree is entered only if `admit.test(child.summary())`. Because the summary sits in the parent, a rejected subtree is skipped without I/O. Queries built on this:
+
+`Tree.keys()`, `Tree.forEachKey` and `TreeCursor.forEachRemainingKey` walk the keys straight out of each leaf's `long[]` without building an `Entry` per key, and `Tree.forEachValue` does the same for values. Use them when only keys or only values are needed.
 
 | Query | Predicate |
 |---|---|

engine/src/main/java/io/hstore/engine/topology/Hyperedge.java

@@ -23,6 +23,7 @@
 import java.util.Set;
 import java.util.function.BinaryOperator;
 import java.util.function.Consumer;
+import java.util.function.LongConsumer;
 import java.util.function.UnaryOperator;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
@@ -127,6 +128,13 @@ public LongStream memberIds() {
         };
     }
 
+    public void forEachMemberId(LongConsumer action) {
+        switch (kind) {
+            case SET -> members.forEachKey(action);
+            case ORDERED -> members.forEachValue(incidence -> action.accept(incidence.member()));
+        }
+    }
+
     public Stream<Incidence> validAt(long instant) {
         return members.scan(summary -> summary.mayBeValidAt(instant))
                 .map(Entry::value)

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

@@ -47,6 +47,23 @@ int route(long key) {
         int low = 1;
         int high = size - 1;
         int result = 0;
+        if (size >= Trees.INTERPOLATE_FROM && key > separators[1] && key < separators[size - 1]) {
+            long first = separators[1];
+            int guess = Math.clamp(1 + (long) (((double) key - first) / ((double) separators[size - 1] - first) * (size - 2)), 1, size - 1);
+            if (separators[guess] <= key) {
+                int end = Math.min(size - 1, guess + Trees.INTERPOLATE_WINDOW);
+                result = separators[end] <= key ? end : guess;
+                low = result + 1;
+                high = separators[end] <= key ? high : end - 1;
+            } else {
+                int start = Math.max(1, guess - Trees.INTERPOLATE_WINDOW);
+                if (separators[start] <= key) {
+                    result = start;
+                    low = start + 1;
+                }
+                high = separators[start] <= key ? guess - 1 : start - 1;
+            }
+        }
         while (low <= high) {
             int mid = (low + high) >>> 1;
             if (separators[mid] <= key) {

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

@@ -65,6 +65,23 @@ int heapBytes() {
     }
 
     int search(long key) {
+        if (size >= Trees.INTERPOLATE_FROM) {
+            long low = keys[0];
+            long high = keys[size - 1];
+            if (key > low && key < high) {
+                int guess = Math.clamp((long) (((double) key - low) / ((double) high - low) * (size - 1)), 0, size - 1);
+                long found = keys[guess];
+                if (found == key) {
+                    return guess;
+                }
+                if (found < key) {
+                    int end = Math.min(size, guess + Trees.INTERPOLATE_WINDOW + 1);
+                    return keys[end - 1] >= key ? Arrays.binarySearch(keys, guess + 1, end, key) : Arrays.binarySearch(keys, end, size, key);
+                }
+                int start = Math.max(0, guess - Trees.INTERPOLATE_WINDOW);
+                return keys[start] <= key ? Arrays.binarySearch(keys, start, guess, key) : Arrays.binarySearch(keys, 0, start, key);
+            }
+        }
         return Arrays.binarySearch(keys, 0, size, key);
     }