diff --git a/benchmarks/src/main/java/io/hstore/bench/Comparison.java b/benchmarks/src/main/java/io/hstore/bench/Comparison.java index d6b8db8b..b787e406 100644 --- a/benchmarks/src/main/java/io/hstore/bench/Comparison.java +++ b/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 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 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; diff --git a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java index f0dafb9d..8611c144 100644 --- a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java +++ b/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 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 diff --git a/database/src/main/java/io/hstore/db/Reader.java b/database/src/main/java/io/hstore/db/Reader.java index 9aafec35..c5060453 100644 --- a/database/src/main/java/io/hstore/db/Reader.java +++ b/database/src/main/java/io/hstore/db/Reader.java @@ -237,6 +237,12 @@ public Stream 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) diff --git a/database/src/test/java/io/hstore/db/DatabaseTest.java b/database/src/test/java/io/hstore/db/DatabaseTest.java index f08b43ad..1656ab4c 100644 --- a/database/src/test/java/io/hstore/db/DatabaseTest.java +++ b/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 -> { diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 80869f53..386a1e42 100644 --- a/docs/benchmarks.md +++ b/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 diff --git a/docs/database/data-model.md b/docs/database/data-model.md index ff0cec33..36a94834 100644 --- a/docs/database/data-model.md +++ b/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) | diff --git a/engine/src/main/java/io/hstore/engine/index/PostingIndex.java b/engine/src/main/java/io/hstore/engine/index/PostingIndex.java index 56d7b07a..5c2f76bb 100644 --- a/engine/src/main/java/io/hstore/engine/index/PostingIndex.java +++ b/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> stream(Tree> index, long key) { }).orElseGet(Stream::empty); } + public void forEachPosting(Tree> index, long key, LongConsumer action) { + index.get(key).ifPresent(value -> { + switch (value) { + case Postings.Inline

inline -> { + for (long posting : inline.keys()) { + action.accept(posting); + } + } + case Postings.Promoted

(Ref root) -> subtree(index, root).forEachKey(action); + } + }); + } + public Tree

tree(Tree> index, long key) { return index.get(key).map(value -> switch (value) { case Postings.Inline

(long[] keys, List

values, long _, int _) -> { diff --git a/engine/src/main/java/io/hstore/engine/tree/Leaf.java b/engine/src/main/java/io/hstore/engine/tree/Leaf.java index 74169b0f..1da4499c 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Leaf.java +++ b/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; } diff --git a/engine/src/main/java/io/hstore/engine/tree/Tree.java b/engine/src/main/java/io/hstore/engine/tree/Tree.java index e85d173e..fb3492da 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Tree.java +++ b/engine/src/main/java/io/hstore/engine/tree/Tree.java @@ -219,10 +219,20 @@ public Comparator getComparator() { } public void forEachKey(LongConsumer action) { + if (root instanceof Ref.Inline(Leaf leaf)) { + leaf.forEachKey(action); + return; + } cursor().forEachRemainingKey(action); } public void forEachValue(Consumer 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 cursor = cursor(); cursor.hasNext(); cursor.advance()) { action.accept(cursor.value()); } diff --git a/engine/src/main/java/io/hstore/engine/txn/View.java b/engine/src/main/java/io/hstore/engine/txn/View.java index d9b2d19e..6be337f7 100644 --- a/engine/src/main/java/io/hstore/engine/txn/View.java +++ b/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 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); } diff --git a/engine/src/test/java/io/hstore/engine/InlineMembersTest.java b/engine/src/test/java/io/hstore/engine/InlineMembersTest.java index f0cc0f59..b56eda49 100644 --- a/engine/src/test/java/io/hstore/engine/InlineMembersTest.java +++ b/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 ordered(StorageEngine engine, long edge) { return engine.read(snapshot -> snapshot.requireEdge(edge).stream().map(Incidence::member).toList()); } + private static List visited(StorageEngine engine, long edge) { + List members = new ArrayList<>(); + engine.read(snapshot -> { + snapshot.requireEdge(edge).forEachMemberId(members::add); + return null; + }); + return members; + } + private static void assertShapes(StorageEngine engine, Map> expected, Map> shapes) { expected.forEach((edge, members) -> { assertEquals(members, members(engine, edge)); diff --git a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java b/engine/src/test/java/io/hstore/engine/tree/TreeTest.java index 097426a5..dc9b57e5 100644 --- a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java +++ b/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 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 visited = new ArrayList<>(); + tree.forEachValue(incidence -> visited.add(incidence.member())); + assertEquals(List.of(10L, 20L, 30L), visited); } }