From 40b5b44a6e7175572748e0f623875b193bd1222c Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:32:57 +0530 Subject: [PATCH 1/6] perf(feed): read only the header of a frame that replay skips Each subscriber calls replay(acknowledged) after every commit. replay starts at the nearest index entry, which can be up to 63 frames back, and read and checksummed the payload of every one of those frames only to drop them. With three subscribers (statistics, materialized views, semantic index) and 1,000 hyperedges per commit, ingest.edges at scale 1 allocated 39.2 KB per hyperedge across all threads before this change and 32.0 KB after it (median of three runs of the same probe). FrameReader now takes the generation to skip through and steps over the payload of an older frame after reading its 16-byte header. It also reuses one header buffer instead of allocating a buffer and a memory segment per frame. restore() and truncateAfter() still read and verify every frame. --- docs/storage/change-feed.md | 2 +- .../io/hstore/engine/feed/ChangeFeed.java | 64 +++++++++++-------- .../io/hstore/engine/feed/ChangeFeedTest.java | 57 +++++++++++++++++ 3 files changed, 96 insertions(+), 27 deletions(-) create mode 100644 engine/src/test/java/io/hstore/engine/feed/ChangeFeedTest.java diff --git a/docs/storage/change-feed.md b/docs/storage/change-feed.md index 7436b2fc..415cdf46 100644 --- a/docs/storage/change-feed.md +++ b/docs/storage/change-feed.md @@ -77,7 +77,7 @@ offset size field **Open.** `restore()` opens every `*.feed`, scans all frames from the first segment's start to the end of the last segment, and stops at the first frame that is short, has a negative or overlong length, fails its CRC, or is not newer than the frame before it. The CRC covers only the payload, so a frame also counts as damaged unless its payload is non-empty and starts with the same generation as its header (`FeedCodec` always writes the generation first); this keeps a damaged header, or a run of zeros left where an unsynced frame never reached the disk, from passing as a frame. `cutAt(position)` then deletes every segment that starts after the failure point and truncates the segment containing it, so the feed always ends at a clean frame. -**Index.** Each `Segment` remembers `firstGeneration`, the generation of its first frame. An in-memory sparse index `generation -> global position` holds the first frame of **every segment** and every 64th frame thereafter (`INDEX_STRIDE = 64`). `startFor(g)` seeks to the floor entry (never before the first retained segment) and scans forward, so locating a generation costs at most 63 frame reads plus segment boundaries. +**Index.** Each `Segment` remembers `firstGeneration`, the generation of its first frame. An in-memory sparse index `generation -> global position` holds the first frame of **every segment** and every 64th frame thereafter (`INDEX_STRIDE = 64`). `startFor(g)` seeks to the floor entry (never before the first retained segment) and scans forward, so locating a generation costs at most 63 frame reads plus segment boundaries. `replay(after)` reads only the 16-byte header of a frame whose generation is at most `after` and steps over its payload, so a subscriber that wakes after every commit doesn't read the payloads of the frames it has already consumed again. ```text feed/ diff --git a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java index 45610b3b..6f802c9c 100644 --- a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java +++ b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java @@ -10,8 +10,8 @@ import java.io.IOException; import java.io.UncheckedIOException; -import java.lang.foreign.MemorySegment; import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; @@ -307,7 +307,7 @@ public void truncateAfter(long generation) { } public Stream replay(long afterGeneration) { - Iterator frames = new FrameReader(startFor(afterGeneration), end); + Iterator frames = new FrameReader(startFor(afterGeneration), end, afterGeneration); return StreamSupport.stream(Spliterators.spliteratorUnknownSize(frames, Spliterator.ORDERED | Spliterator.NONNULL), false) .filter(frame -> frame.generation() > afterGeneration) .map(frame -> codec.decode(frame.payload())); @@ -351,13 +351,20 @@ private record Frame(long start, long next, long generation, byte[] payload) { } private final class FrameReader implements Iterator { + private final ByteBuffer header = ByteBuffer.allocate(FRAME_HEADER).order(ByteOrder.LITTLE_ENDIAN); private final long limit; + private final long skipThrough; private long position; private Frame next; FrameReader(long from, long limit) { + this(from, limit, Long.MIN_VALUE); + } + + FrameReader(long from, long limit, long skipThrough) { this.position = from; this.limit = limit; + this.skipThrough = skipThrough; } @Override @@ -389,32 +396,37 @@ private Frame read() { } private Frame readLocked() { - Map.Entry entry = segments.floorEntry(position); - if (entry == null || !entry.getValue().file.isOpen()) { - return null; - } - FileHandle file = entry.getValue().file; - long local = position - entry.getKey(); try { - ByteBuffer header = ByteBuffer.allocate(FRAME_HEADER); - if (!readFully(file, header, local)) { - return null; - } - ByteCursor in = ByteCursor.over(MemorySegment.ofBuffer(header.flip())); - int length = in.getInt(); - int crc = in.getInt(); - long generation = in.getLong(); - if (length < 0 || position + FRAME_HEADER + length > limit) { - return null; - } - ByteBuffer body = ByteBuffer.allocate(length); - if (!readFully(file, body, local + FRAME_HEADER)) { - return null; + while (position + FRAME_HEADER <= limit) { + Map.Entry entry = segments.floorEntry(position); + if (entry == null || !entry.getValue().file.isOpen()) { + return null; + } + FileHandle file = entry.getValue().file; + long local = position - entry.getKey(); + if (!readFully(file, header.clear(), local)) { + return null; + } + int length = header.getInt(0); + int crc = header.getInt(Integer.BYTES); + long generation = header.getLong(2 * Integer.BYTES); + if (length < 0 || position + FRAME_HEADER + length > limit) { + return null; + } + if (generation <= skipThrough) { + position += FRAME_HEADER + length; + continue; + } + ByteBuffer body = ByteBuffer.allocate(length); + if (!readFully(file, body, local + FRAME_HEADER)) { + return null; + } + byte[] payload = body.array(); + return Checksums.crc32c(payload) == crc && carries(payload, generation) + ? new Frame(position, position + FRAME_HEADER + length, generation, payload) + : null; } - byte[] payload = body.array(); - return Checksums.crc32c(payload) == crc && carries(payload, generation) - ? new Frame(position, position + FRAME_HEADER + length, generation, payload) - : null; + return null; } catch (IOException e) { throw new UncheckedIOException(e); } diff --git a/engine/src/test/java/io/hstore/engine/feed/ChangeFeedTest.java b/engine/src/test/java/io/hstore/engine/feed/ChangeFeedTest.java new file mode 100644 index 00000000..3f714d74 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/feed/ChangeFeedTest.java @@ -0,0 +1,57 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.feed; + +import io.hstore.engine.catalog.SlotRegistry; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class ChangeFeedTest { + + @TempDir + Path directory; + + @Test + void replayReadsOnlyTheHeaderOfAFrameItSkips() throws IOException { + FeedCodec codec = new FeedCodec(new SlotRegistry(List.of())); + try (ChangeFeed feed = ChangeFeed.open(directory, codec)) { + for (long generation = 1; generation <= 5; generation++) { + feed.append(new CommitEvent(generation, generation, generation * 1000, 0, List.of(), List.of())); + } + Path file; + try (Stream files = Files.list(directory)) { + file = files.filter(path -> path.toString().endsWith(".feed")).findFirst().orElseThrow(); + } + try (RandomAccessFile raw = new RandomAccessFile(file.toFile(), "rw")) { + long second = 16 + frameLength(raw, 0); + long payload = second + 16; + raw.seek(payload); + int value = raw.read(); + raw.seek(payload); + raw.write(value ^ 0xFF); + } + assertEquals(List.of(3L, 4L, 5L), feed.replay(2).map(CommitEvent::generation).toList()); + assertEquals(List.of(1L), feed.replay(0).map(CommitEvent::generation).toList()); + } + } + + private static int frameLength(RandomAccessFile raw, long at) throws IOException { + byte[] header = new byte[4]; + raw.seek(at); + raw.readFully(header); + return ByteBuffer.wrap(header).order(ByteOrder.LITTLE_ENDIAN).getInt(); + } +} From 9efcdf884b40eb867194cd993ffe381c5e2a10bb Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:32:57 +0530 Subject: [PATCH 2/6] perf(tree): walk a leaf by position when diffing two trees TreeDiff expanded a leaf by pushing one Element object per entry onto the walker's stack. The change log diffs every changed catalog and incidence leaf at commit, and Hyperedge.diff diffs every loaded member tree, so most of those objects were thrown away a moment later. A leaf now goes on the stack as one Run that holds the leaf and the position of the next entry. The deltas and their order are unchanged. A randomized test checks diff against an oracle in both directions and against an empty tree. --- .../java/io/hstore/engine/tree/TreeDiff.java | 66 ++++++++++++------- .../java/io/hstore/engine/tree/TreeTest.java | 52 +++++++++++++++ 2 files changed, 95 insertions(+), 23 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java b/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java index 44b37e18..c85d0644 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java @@ -88,7 +88,21 @@ private sealed interface Item { record Subtree(Ref ref) implements Item { } - record Element(long key, Object value) implements Item { + final class Run implements Item { + private final Leaf leaf; + private int index; + + Run(Leaf leaf) { + this.leaf = leaf; + } + + long key() { + return leaf.key(index); + } + + Object value() { + return leaf.value(index); + } } } @@ -139,23 +153,21 @@ private Delta step() { } return switch (a) { case Item.Subtree(Ref ra) when b instanceof Item.Subtree(Ref rb) -> compareSubtrees(ra, rb); - case Item.Subtree(Ref ra) when b instanceof Item.Element(long key, Object value) -> { - if (ra.summary().min() > key) { - right.pop(); - yield new Delta.Added<>(key, after.schema().cast(value)); + case Item.Subtree(Ref ra) when b instanceof Item.Run run -> { + if (ra.summary().min() > run.key()) { + yield new Delta.Added<>(run.key(), after.schema().cast(take(right, run))); } expand(left, before); yield null; } - case Item.Element(long key, Object value) when b instanceof Item.Subtree(Ref rb) -> { - if (rb.summary().min() > key) { - left.pop(); - yield new Delta.Removed<>(key, before.schema().cast(value)); + case Item.Run run when b instanceof Item.Subtree(Ref rb) -> { + if (rb.summary().min() > run.key()) { + yield new Delta.Removed<>(run.key(), before.schema().cast(take(left, run))); } expand(right, after); yield null; } - case Item.Element(long ka, Object va) when b instanceof Item.Element(long kb, Object vb) -> compareElements(ka, va, kb, vb); + case Item.Run ra when b instanceof Item.Run rb -> compareElements(ra, rb); default -> throw new IllegalStateException(); }; } @@ -183,17 +195,17 @@ private Delta compareSubtrees(Ref ra, Ref rb) { return null; } - private Delta compareElements(long ka, Object va, long kb, Object vb) { + private Delta compareElements(Item.Run ra, Item.Run rb) { + long ka = ra.key(); + long kb = rb.key(); if (ka < kb) { - left.pop(); - return new Delta.Removed<>(ka, before.schema().cast(va)); + return new Delta.Removed<>(ka, before.schema().cast(take(left, ra))); } if (kb < ka) { - right.pop(); - return new Delta.Added<>(kb, after.schema().cast(vb)); + return new Delta.Added<>(kb, after.schema().cast(take(right, rb))); } - left.pop(); - right.pop(); + Object va = take(left, ra); + Object vb = take(right, rb); return Objects.equals(va, vb) ? null : new Delta.Changed<>(ka, before.schema().cast(va), after.schema().cast(vb)); } @@ -201,20 +213,28 @@ private Delta drain(Deque side, Tree tree, boolean removed) { while (side.peek() instanceof Item.Subtree) { expand(side, tree); } - if (side.isEmpty()) { + if (!(side.peek() instanceof Item.Run run)) { return null; } - Item.Element element = (Item.Element) side.pop(); - V value = tree.schema().cast(element.value()); - return removed ? new Delta.Removed<>(element.key(), value) : new Delta.Added<>(element.key(), value); + long key = run.key(); + V value = tree.schema().cast(take(side, run)); + return removed ? new Delta.Removed<>(key, value) : new Delta.Added<>(key, value); + } + + private static Object take(Deque side, Item.Run run) { + Object value = run.value(); + if (++run.index == run.leaf.size()) { + side.pop(); + } + return value; } private void expand(Deque side, Tree tree) { Item.Subtree subtree = (Item.Subtree) side.pop(); switch (tree.node(subtree.ref())) { case Leaf leaf -> { - for (int i = leaf.size() - 1; i >= 0; i--) { - side.push(new Item.Element(leaf.key(i), leaf.value(i))); + if (leaf.size() > 0) { + side.push(new Item.Run(leaf)); } } case Branch branch -> { 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 52dbcfcf..69957df8 100644 --- a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java +++ b/engine/src/test/java/io/hstore/engine/tree/TreeTest.java @@ -230,6 +230,58 @@ void diffReportsExactDeltasAndSkipsSharedPages() { assertTrue(pages.loads < 40, "diff visited " + pages.loads + " pages"); } + @ParameterizedTest + @ValueSource(longs = {1, 2, 3, 4, 5}) + void diffMatchesAnOracleAcrossRandomEdits(long seed) { + RandomGenerator random = random(seed); + MemoryPages pages = new MemoryPages(1024); + TreeMap first = new TreeMap<>(); + for (int i = random.nextInt(3000); i > 0; i--) { + long key = random.nextLong(5000); + first.put(key, key); + } + Tree before = Tree.empty(LONGS, pages); + WriteScope scope = new WriteScope(); + for (Map.Entry entry : first.entrySet()) { + before = before.put(scope, entry.getKey(), entry.getValue()); + } + before = pages.persist(before); + TreeMap second = new TreeMap<>(first); + Tree after = before; + WriteScope edits = new WriteScope(); + for (int i = random.nextInt(400); i > 0; i--) { + long key = random.nextLong(6000); + if (random.nextBoolean() && !second.isEmpty()) { + long existing = second.ceilingKey(key) != null ? second.ceilingKey(key) : second.firstKey(); + second.remove(existing); + after = after.remove(edits, existing); + } else { + long value = random.nextLong(3); + second.put(key, value); + after = after.put(edits, key, value); + } + } + List> expected = new ArrayList<>(); + TreeSet keys = new TreeSet<>(first.keySet()); + keys.addAll(second.keySet()); + for (long key : keys) { + Long was = first.get(key); + Long now = second.get(key); + if (was == null) { + expected.add(new TreeDiff.Delta.Added<>(key, now)); + } else if (now == null) { + expected.add(new TreeDiff.Delta.Removed<>(key, was)); + } else if (!was.equals(now)) { + expected.add(new TreeDiff.Delta.Changed<>(key, was, now)); + } + } + assertEquals(expected, TreeDiff.diff(before, after).toList()); + assertEquals(expected.size(), TreeDiff.diff(after, before).count()); + assertEquals(List.of(), TreeDiff.diff(after, after).toList()); + assertEquals(second.size(), TreeDiff.diff(Tree.empty(LONGS, pages), after).count()); + assertEquals(second.size(), TreeDiff.diff(after, Tree.empty(LONGS, pages)).count()); + } + @Test void keysetPaginationIsStable() { MemoryPages pages = new MemoryPages(1024); From f437e4a2239c12f21b9d9f92d8ddc58cc3d4fe6a Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:32:57 +0530 Subject: [PATCH 3/6] perf(tree): grow a bulk builder's arrays from eight entries BulkBuilder allocated a 64-entry key array and value array up front. Hyperedge.build creates two builders per hyperedge, members and the order index, and the order index stays empty for a SET edge, so a 4-member edge allocated about 1.6 KB of arrays to hold four entries. The arrays now start empty and grow to 8 entries on the first add, then double as before. --- .../main/java/io/hstore/engine/tree/BulkBuilder.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java b/engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java index 01793bb0..5c664b86 100644 --- a/engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java +++ b/engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java @@ -14,6 +14,7 @@ public final class BulkBuilder { private static final int SPILL_BATCH = 32; + private static final int INITIAL_CAPACITY = 8; private final TreeSchema schema; private final NodeSource source; @@ -21,8 +22,8 @@ public final class BulkBuilder { private final Layout layout; private final List leaves = new ArrayList<>(); private BinaryOperator merge = (_, latest) -> latest; - private long[] keys = new long[64]; - private Object[] values = new Object[64]; + private long[] keys = new long[0]; + private Object[] values = new Object[0]; private int size; private int bytes; private int unspilled; @@ -58,8 +59,9 @@ public BulkBuilder add(long key, V value) { } entryBytes += size == 0 ? ByteCursor.signedVarLongSize(key) : ByteCursor.varLongSize(key - keys[size - 1]); if (size == keys.length) { - keys = Arrays.copyOf(keys, size * 2); - values = Arrays.copyOf(values, size * 2); + int capacity = Math.max(INITIAL_CAPACITY, size * 2); + keys = Arrays.copyOf(keys, capacity); + values = Arrays.copyOf(values, capacity); } keys[size] = key; values[size] = value; From 7363578a964f40ac78d654aee981514d549fc19b Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:32:58 +0530 Subject: [PATCH 4/6] perf(topology): find present incidence columns without a stream IncidenceCodec.encode built a stream per column to find which columns any value uses. The change feed encodes each member change on its own, so a hyperedge with four members created 24 streams at commit. A plain loop gives the same flags. --- .../java/io/hstore/engine/topology/IncidenceCodec.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java b/engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java index 075a91dd..575a1513 100644 --- a/engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java +++ b/engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java @@ -79,8 +79,11 @@ public int maxSize(Incidence incidence) { public void encode(ByteCursor out, long[] keys, List values) { int flags = 0; for (Column column : COLUMNS) { - if (values.stream().anyMatch(column.present)) { - flags |= 1 << column.ordinal(); + for (int i = 0; i < values.size(); i++) { + if (column.present.test(values.get(i))) { + flags |= 1 << column.ordinal(); + break; + } } } out.putByte(flags); From 641411d1a1e63fe95d93b29a18e702ce14b65644 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:32:58 +0530 Subject: [PATCH 5/6] perf(db): load members from a long[] and check the edge once per load Writer.load took a List and, for every member, checked write permission again, decoded the edge and its type, streamed over the roles, interned the role set and streamed over the weight. Only the member's own existence check depends on the member. load(edge, long[], spec) checks the edge and roles and builds the first incidence the way add() does, then checks each remaining member and copies the first incidence's columns. The List overload unboxes into the array and calls it. Errors are the same as before and come in the same order. Writer.assign checks required properties with a loop instead of a stream. --- .../src/main/java/io/hstore/db/Writer.java | 50 +++++++++++++------ .../test/java/io/hstore/db/DatabaseTest.java | 43 ++++++++++++++++ docs/database/data-model.md | 2 +- 3 files changed, 80 insertions(+), 15 deletions(-) diff --git a/database/src/main/java/io/hstore/db/Writer.java b/database/src/main/java/io/hstore/db/Writer.java index f6325a37..cc1f3b24 100644 --- a/database/src/main/java/io/hstore/db/Writer.java +++ b/database/src/main/java/io/hstore/db/Writer.java @@ -21,6 +21,7 @@ import io.hstore.db.value.TypeTag; import io.hstore.db.value.Value; import io.hstore.db.value.Values; +import io.hstore.engine.Dictionary; import io.hstore.engine.HStoreException; import io.hstore.engine.catalog.AtomRecord; import io.hstore.engine.catalog.EngineSlots; @@ -31,6 +32,7 @@ import io.hstore.engine.txn.Transaction; import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Optional; @@ -138,13 +140,11 @@ private TypeDef kindOf(String type, AtomKind kind) { } private void assign(long atom, TypeDef definition, Map properties) { - definition.properties().stream() - .filter(PropertyDef::required) - .filter(property -> !properties.containsKey(property.name())) - .findFirst() - .ifPresent(missing -> { - throw HStoreException.invalid(definition.name() + " requires property '" + missing.name() + "'"); - }); + for (PropertyDef property : definition.properties()) { + if (property.required() && !properties.containsKey(property.name())) { + throw HStoreException.invalid(definition.name() + " requires property '" + property.name() + "'"); + } + } properties.forEach((name, value) -> set(atom, name, Value.of(value))); } @@ -178,7 +178,19 @@ public void removeAt(long edge, long index) { } public void load(long edge, List members, MemberSpec spec) { - txn.load(edge, members.stream().map(member -> incidence(edge, member, spec)).toList()); + load(edge, members.stream().mapToLong(Long::longValue).toArray(), spec); + } + + public void load(long edge, long[] members, MemberSpec spec) { + Incidence[] incidences = new Incidence[members.length]; + if (members.length > 0) { + Incidence first = incidences[0] = incidence(edge, members[0], spec); + for (int i = 1; i < members.length; i++) { + requireVisible(members[i]); + incidences[i] = new Incidence(members[i], first.roleSet(), first.weight(), first.validFrom(), first.validTo(), first.dataRef(), first.qualifier()); + } + } + txn.load(edge, Arrays.asList(incidences)); } public void qualify(long edge, long member, long qualifier) { @@ -188,20 +200,30 @@ public void qualify(long edge, long member, long qualifier) { } private Incidence incidence(long edge, long member, MemberSpec spec) { + requireMembership(edge, spec); + requireVisible(member); + int roleSet = spec.roles().isEmpty() ? Dictionary.NO_ROLES : database.engine().dictionary().roleSet(spec.roles()); + long weight = spec.weight().isPresent() ? Weight.of(spec.weight().getAsDouble()) : Weight.ONE; + return new Incidence(member, roleSet, weight, spec.validity().from(), spec.validity().to(), spec.dataRef(), spec.qualifier()); + } + + private void requireMembership(long edge, MemberSpec spec) { principal().requireWrite(); Atom container = require(edge); if (!container.isEdge()) { throw HStoreException.invalid("atom " + edge + " is not a hyperedge"); } - spec.roles().stream().filter(role -> !container.type().allowsRole(role)).findFirst().ifPresent(role -> { - throw HStoreException.invalid(container.type().name() + " does not declare role '" + role + "'"); - }); + for (String role : spec.roles()) { + if (!container.type().allowsRole(role)) { + throw HStoreException.invalid(container.type().name() + " does not declare role '" + role + "'"); + } + } + } + + private void requireVisible(long member) { if (!visible(member)) { throw HStoreException.invalid("member atom " + member + " does not exist"); } - int roleSet = database.engine().dictionary().roleSet(spec.roles()); - return new Incidence(member, roleSet, spec.weight().stream().mapToLong(Weight::of).findFirst().orElse(Weight.ONE), - spec.validity().from(), spec.validity().to(), spec.dataRef(), spec.qualifier()); } public void set(long atom, String name, Value value) { diff --git a/database/src/test/java/io/hstore/db/DatabaseTest.java b/database/src/test/java/io/hstore/db/DatabaseTest.java index 1656ab4c..7754709a 100644 --- a/database/src/test/java/io/hstore/db/DatabaseTest.java +++ b/database/src/test/java/io/hstore/db/DatabaseTest.java @@ -136,6 +136,49 @@ void membersCarryRolesWeightsAndValidity() { }); } + @Test + void bulkLoadedMembersMatchMembersAddedOneAtATime() { + MemberSpec spec = MemberSpec.roles("author").withWeight(0.5).withValidity(new Validity(10, 20)); + long[] edges = database.write(writer -> { + long[] people = new long[5]; + for (int i = 0; i < people.length; i++) { + people[i] = writer.node("Person", "p" + i, Map.of("name", "Person " + i)); + } + long added = writer.edge("Coauthorship", Map.of("year", 2024)); + for (long person : people) { + writer.add(added, person, spec); + } + long fromArray = writer.edge("Coauthorship", Map.of("year", 2024)); + writer.load(fromArray, new long[]{people[3], people[0], people[4], people[1], people[2]}, spec); + long fromList = writer.edge("Coauthorship", Map.of("year", 2024)); + writer.load(fromList, List.of(people[4], people[2], people[0], people[3], people[1]), spec); + long empty = writer.edge("Coauthorship", Map.of("year", 2024)); + writer.load(empty, new long[0], spec); + return new long[]{added, fromArray, fromList, empty}; + }); + database.read(reader -> { + List expected = reader.members(edges[0]).toList(); + assertEquals(5, expected.size()); + assertEquals(expected, reader.members(edges[1]).toList()); + assertEquals(expected, reader.members(edges[2]).toList()); + assertEquals(0, reader.cardinality(edges[3])); + return null; + }); + long node = database.read(reader -> reader.resolve("Person", "p0")); + assertThrows(HStoreException.InvalidSchema.class, () -> database.write(writer -> { + writer.load(edges[3], new long[]{node}, MemberSpec.roles("reviewer")); + return null; + })); + assertThrows(HStoreException.class, () -> database.write(writer -> { + writer.load(edges[3], new long[]{node, 999_999_999L}, spec); + return null; + })); + assertThrows(HStoreException.class, () -> database.write(writer -> { + writer.load(node, new long[]{node}, spec); + return null; + })); + } + @Test void incidentEdgeVisitorsSeeTheSameEdgesAsTheIncidenceStream() { long[] people = database.write(writer -> { diff --git a/docs/database/data-model.md b/docs/database/data-model.md index 36a94834..15b8a85d 100644 --- a/docs/database/data-model.md +++ b/docs/database/data-model.md @@ -450,7 +450,7 @@ Optional.empty | `add(edge, member, spec)` | WRITER | upsert an incidence (roles validated against the type) | | `insertAt(edge, index, member, spec)` / `append(...)` | WRITER | ordered edges | | `remove(edge, member)` / `removeAt(edge, index)` | WRITER | membership removal | -| `load(edge, members, spec)` | WRITER | replaces the whole membership with a tree built bottom-up from `members` (`Hyperedge.build`) | +| `load(edge, members, spec)` | WRITER | replaces the whole membership with a tree built bottom-up from `members` (`Hyperedge.build`); `members` is a `List` or a `long[]`, and the edge and the roles are checked once per call | | `qualify(edge, member, qualifier)` | WRITER | attach an assertion id to an incidence | | `set(atom, name, value[, validity])` / `unset(atom, name)` | WRITER | per-key OCC on the property bag | | `document(atom, json)` | WRITER | stores `$document` | From 419b19c02715d3e0097dffd1d7174b3f18973f2e Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 14 Oct 2026 12:32:58 +0530 Subject: [PATCH 6/6] perf(bench): pass member ids to Writer.load as a long[] The adapter boxed every member id into a List before loading a hyperedge. It now uses the long[] overload of Writer.load, which is part of the public API. --- .../main/java/io/hstore/bench/HStoreStore.java | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java index 8611c144..e4519e77 100644 --- a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java +++ b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java @@ -16,7 +16,6 @@ import io.hstore.engine.txn.Durability; import java.nio.file.Path; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -120,10 +119,10 @@ public void ingestEdges(Dataset dataset, int batch) { } } - private List members(int[] indexes) { - List members = new ArrayList<>(indexes.length); - for (int index : indexes) { - members.add(nodes[index]); + private long[] members(int[] indexes) { + long[] members = new long[indexes.length]; + for (int i = 0; i < indexes.length; i++) { + members[i] = nodes[indexes[i]]; } return members; } @@ -191,11 +190,7 @@ public void update(int[] probes, int from, int to, int round) { public void largeEdge(int count) { large = database.write(writer -> { long edge = writer.edge(EDGE); - List members = new ArrayList<>(count); - for (int i = 0; i < count; i++) { - members.add(nodes[i]); - } - writer.load(edge, members, MemberSpec.PLAIN); + writer.load(edge, Arrays.copyOf(nodes, count), MemberSpec.PLAIN); return edge; }); }