From 8a54241c8962acebb8419e9180f7e12c9c03d77d Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 14:22:32 +0530 Subject: [PATCH 1/3] perf(engine): decode inline posting lists without streams Each inline posting list went through a stream, a list that allows nulls and then a second copy in Postings.Inline. It is now built once with List.of, which Inline keeps as it is. --- .../java/io/hstore/engine/index/PostingsCodec.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java b/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java index 6174a11..e66c1c0 100644 --- a/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java +++ b/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java @@ -9,7 +9,6 @@ import io.hstore.engine.tree.TreeSchema; import io.hstore.engine.tree.ValueCodec; -import java.util.Arrays; import java.util.List; import java.util.function.ToLongFunction; @@ -81,10 +80,18 @@ public void decode(ByteCursor in, long[] keys, Object[] into) { } Object[] decoded = new Object[count]; postings.codec().decode(in, postingKeys, decoded); - into[i] = inline(postingKeys, Arrays.stream(decoded).map(postings::cast).toList()); + into[i] = inline(postingKeys, typed(decoded)); } } + @SuppressWarnings("unchecked") + private List

typed(Object[] decoded) { + for (Object value : decoded) { + postings.cast(value); + } + return (List

) List.of(decoded); + } + @Override public boolean holdsRefs() { return true; From 1ad571e76730300bee26fc680b8a8b038a9b9f9e Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 09:59:41 +0530 Subject: [PATCH 2/3] perf(engine): visit nested trees without rebuilding records Codecs that hold tree references only had mapRefs, which builds a new record even when every reference maps to itself. The reference counter walks every leaf value of every commit's changed nodes this way, and so does the check that sets a leaf's flag for page references. mapRefs now returns the same record when nothing changed, and a new forEachRef visits the references without building anything; the counter and the flag check use it. --- .../io/hstore/db/property/PropertySlots.java | 8 ++- .../io/hstore/engine/catalog/AtomCodec.java | 18 +++++-- .../io/hstore/engine/index/PostingsCodec.java | 13 ++++- .../io/hstore/engine/tree/TreeWalker.java | 9 ++-- .../io/hstore/engine/tree/ValueCodec.java | 14 ++++-- .../hstore/engine/catalog/AtomCodecTest.java | 49 +++++++++++++++++++ .../engine/index/PostingsCodecTest.java | 46 +++++++++++++++++ 7 files changed, 143 insertions(+), 14 deletions(-) create mode 100644 engine/src/test/java/io/hstore/engine/catalog/AtomCodecTest.java create mode 100644 engine/src/test/java/io/hstore/engine/index/PostingsCodecTest.java diff --git a/database/src/main/java/io/hstore/db/property/PropertySlots.java b/database/src/main/java/io/hstore/db/property/PropertySlots.java index 2851d53..af64c60 100644 --- a/database/src/main/java/io/hstore/db/property/PropertySlots.java +++ b/database/src/main/java/io/hstore/db/property/PropertySlots.java @@ -62,7 +62,13 @@ public boolean refersToPages(IndexRoot value) { @Override public IndexRoot mapRefs(IndexRoot value, RefMapper mapper) { - return new IndexRoot(mapper.map(value.root(), VALUE_POSTINGS.schema())); + Ref mapped = mapper.map(value.root(), VALUE_POSTINGS.schema()); + return mapped == value.root() ? value : new IndexRoot(mapped); + } + + @Override + public void forEachRef(IndexRoot value, RefVisitor visitor) { + visitor.visit(value.root(), VALUE_POSTINGS.schema()); } }, EntryMeasure.keyed(root -> root.root().summary().fingerprint())); diff --git a/engine/src/main/java/io/hstore/engine/catalog/AtomCodec.java b/engine/src/main/java/io/hstore/engine/catalog/AtomCodec.java index 3b4727b..21e46f1 100644 --- a/engine/src/main/java/io/hstore/engine/catalog/AtomCodec.java +++ b/engine/src/main/java/io/hstore/engine/catalog/AtomCodec.java @@ -88,10 +88,20 @@ public boolean refersToPages(AtomRecord record) { public AtomRecord mapRefs(AtomRecord record, RefMapper mapper) { return switch (record) { case NodeRecord node -> node; - case EdgeRecord edge -> new EdgeRecord(edge.type(), edge.kind(), - mapper.map(edge.members(), TopologySchemas.members(edge.kind())), - mapper.map(edge.order(), TopologySchemas.ORDER_INDEX), - edge.version(), edge.dataRef(), edge.tenant(), edge.isolation()); + case EdgeRecord edge -> { + Ref members = mapper.map(edge.members(), TopologySchemas.members(edge.kind())); + Ref order = mapper.map(edge.order(), TopologySchemas.ORDER_INDEX); + yield members == edge.members() && order == edge.order() ? edge + : new EdgeRecord(edge.type(), edge.kind(), members, order, edge.version(), edge.dataRef(), edge.tenant(), edge.isolation()); + } }; } + + @Override + public void forEachRef(AtomRecord record, RefVisitor visitor) { + if (record instanceof EdgeRecord edge) { + visitor.visit(edge.members(), TopologySchemas.members(edge.kind())); + visitor.visit(edge.order(), TopologySchemas.ORDER_INDEX); + } + } } diff --git a/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java b/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java index e66c1c0..f440431 100644 --- a/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java +++ b/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java @@ -104,6 +104,17 @@ public boolean refersToPages(Postings

value) { @Override public Postings

mapRefs(Postings

value, RefMapper mapper) { - return value instanceof Postings.Promoted

(Ref root) ? new Postings.Promoted<>(mapper.map(root, postings)) : value; + if (!(value instanceof Postings.Promoted

(Ref root))) { + return value; + } + Ref mapped = mapper.map(root, postings); + return mapped == root ? value : new Postings.Promoted<>(mapped); + } + + @Override + public void forEachRef(Postings

value, RefVisitor visitor) { + if (value instanceof Postings.Promoted

(Ref root)) { + visitor.visit(root, postings); + } } } diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java index b0bc36c..b8bc6a9 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java @@ -29,14 +29,13 @@ public void references(long pageId, TreeSchema schema, int height, Reference switch (node) { case Leaf leaf -> { if (schema.codec().holdsRefs()) { - ValueCodec.RefMapper mapper = (nested, nestedSchema) -> { + ValueCodec.RefVisitor visitor = (nested, nestedSchema) -> { if (nested instanceof Ref.Stored(long child, int _, Summary _)) { into.accept(child, nestedSchema, UNKNOWN_HEIGHT); } - return nested; }; for (int i = 0; i < leaf.size(); i++) { - nested(schema, leaf.value(i), mapper); + nested(schema, leaf.value(i), visitor); } } } @@ -50,7 +49,7 @@ public void references(long pageId, TreeSchema schema, int height, Reference } } - private static void nested(TreeSchema schema, Object value, ValueCodec.RefMapper mapper) { - schema.codec().mapRefs(schema.cast(value), mapper); + private static void nested(TreeSchema schema, Object value, ValueCodec.RefVisitor visitor) { + schema.codec().forEachRef(schema.cast(value), visitor); } } diff --git a/engine/src/main/java/io/hstore/engine/tree/ValueCodec.java b/engine/src/main/java/io/hstore/engine/tree/ValueCodec.java index ef60d3c..49cf9a9 100644 --- a/engine/src/main/java/io/hstore/engine/tree/ValueCodec.java +++ b/engine/src/main/java/io/hstore/engine/tree/ValueCodec.java @@ -32,11 +32,15 @@ default V mapRefs(V value, RefMapper mapper) { default boolean refersToPages(V value) { boolean[] found = new boolean[1]; - mapRefs(value, (ref, _) -> { - found[0] |= ref instanceof Ref.Stored; + forEachRef(value, (ref, _) -> found[0] |= ref instanceof Ref.Stored); + return found[0]; + } + + default void forEachRef(V value, RefVisitor visitor) { + mapRefs(value, (ref, schema) -> { + visitor.visit(ref, schema); return ref; }); - return found[0]; } static ValueCodec rows(ToIntFunction size, BiConsumer writer, Function reader) { @@ -64,4 +68,8 @@ public void decode(ByteCursor in, long[] keys, Object[] into) { interface RefMapper { Ref map(Ref ref, TreeSchema schema); } + + interface RefVisitor { + void visit(Ref ref, TreeSchema schema); + } } diff --git a/engine/src/test/java/io/hstore/engine/catalog/AtomCodecTest.java b/engine/src/test/java/io/hstore/engine/catalog/AtomCodecTest.java new file mode 100644 index 0000000..95ef218 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/catalog/AtomCodecTest.java @@ -0,0 +1,49 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.catalog; + +import io.hstore.engine.page.PageId; +import io.hstore.engine.topology.EdgeKind; +import io.hstore.engine.tree.Ref; +import io.hstore.engine.tree.Summary; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; + +class AtomCodecTest { + + private static final Ref MEMBERS = new Ref.Stored(PageId.pack(310, 0), 4, Summary.EMPTY); + private static final Ref ORDER = new Ref.Stored(PageId.pack(320, 0), 2, Summary.EMPTY); + private static final AtomRecord.EdgeRecord EDGE = new AtomRecord.EdgeRecord(7, EdgeKind.ORDERED, MEMBERS, ORDER, 1, 0, 0, 0); + + @Test + void mappingRefsToThemselvesKeepsTheSameRecord() { + assertSame(EDGE, AtomCodec.INSTANCE.mapRefs(EDGE, (ref, _) -> ref)); + Ref moved = new Ref.Stored(PageId.pack(900, 0), 4, Summary.EMPTY); + AtomRecord.EdgeRecord changed = (AtomRecord.EdgeRecord) AtomCodec.INSTANCE.mapRefs(EDGE, (ref, _) -> ref == MEMBERS ? moved : ref); + assertNotSame(EDGE, changed); + assertSame(moved, changed.members()); + assertSame(ORDER, changed.order()); + } + + @Test + void forEachRefVisitsWhatMapRefsSees() { + List mapped = new ArrayList<>(); + AtomCodec.INSTANCE.mapRefs(EDGE, (ref, _) -> { + mapped.add(ref); + return ref; + }); + List visited = new ArrayList<>(); + AtomCodec.INSTANCE.forEachRef(EDGE, (ref, _) -> visited.add(ref)); + assertEquals(mapped, visited); + List none = new ArrayList<>(); + AtomCodec.INSTANCE.forEachRef(new AtomRecord.NodeRecord(1, "k", 0, 0, 0, 0, 0), (ref, _) -> none.add(ref)); + assertEquals(List.of(), none); + } +} diff --git a/engine/src/test/java/io/hstore/engine/index/PostingsCodecTest.java b/engine/src/test/java/io/hstore/engine/index/PostingsCodecTest.java new file mode 100644 index 0000000..775dccf --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/index/PostingsCodecTest.java @@ -0,0 +1,46 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.index; + +import io.hstore.engine.page.ByteCursor; +import io.hstore.engine.page.PageId; +import io.hstore.engine.tree.FingerprintMode; +import io.hstore.engine.tree.Hashing; +import io.hstore.engine.tree.Ref; +import io.hstore.engine.tree.Summary; +import io.hstore.engine.tree.TreeSchema; +import io.hstore.engine.tree.ValueCodec; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; + +class PostingsCodecTest { + + private static final TreeSchema LONGS = new TreeSchema<>(210, "postings-test", FingerprintMode.SET, + ValueCodec.rows(ByteCursor::signedVarLongSize, ByteCursor::putSignedVarLong, ByteCursor::getSignedVarLong), + (key, value, into) -> into.entry(key, Hashing.of(key, value))); + private static final PostingsCodec CODEC = new PostingsCodec<>(LONGS, value -> value); + private static final Ref ROOT = new Ref.Stored(PageId.pack(408, 0), 3, Summary.EMPTY); + + @Test + void promotedPostingsKeepTheirIdentityWhenNothingMoves() { + Postings promoted = new Postings.Promoted<>(ROOT); + assertSame(promoted, CODEC.mapRefs(promoted, (ref, _) -> ref)); + Ref moved = new Ref.Stored(PageId.pack(500, 0), 3, Summary.EMPTY); + Postings changed = CODEC.mapRefs(promoted, (ref, _) -> moved); + assertNotSame(promoted, changed); + assertEquals(new Postings.Promoted<>(moved), changed); + List visited = new ArrayList<>(); + CODEC.forEachRef(promoted, (ref, schema) -> { + visited.add(ref); + assertSame(LONGS, schema); + }); + assertEquals(List.of(ROOT), visited); + } +} From cd30b4b6a4ef2ae59e5388ed0265fd9cce2bdb5a Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Sun, 11 Oct 2026 15:07:16 +0530 Subject: [PATCH 3/3] test(engine): check that checkpoints taken while commits run lose nothing A writer commits while the test thread checkpoints every 20 ms, then the engine stops without a clean shutdown. Every commit has to come back, with the reference counts matching the trees. The test comes from #76, whose compaction changes #81 replaced. --- .../io/hstore/engine/CrashRecoveryTest.java | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java index cbed659..f205845 100644 --- a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java +++ b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java @@ -24,6 +24,7 @@ import java.nio.file.Path; import java.nio.file.StandardOpenOption; import java.util.Comparator; +import java.util.concurrent.locks.LockSupport; import java.util.EnumSet; import java.util.List; import java.util.concurrent.ExecutorService; @@ -190,6 +191,41 @@ void concurrentCommitsShareDurabilityBarriers() throws Exception { } } + @ParameterizedTest + @EnumSource(WalMode.class) + void checkpointsTakenWhileCommitsRunLoseNothing(WalMode mode) throws Exception { + EngineOptions options = EngineTest.small().withWalMode(mode); + StorageEngine engine = StorageEngine.open(directory, options); + long edge = engine.write(txn -> txn.createEdge(EngineTest.EDGE, EdgeKind.ORDERED)); + Thread writer = Thread.ofPlatform().start(() -> { + for (int i = 0; i < 400; i++) { + int n = i; + engine.write(txn -> { + txn.append(edge, Incidence.of(txn.createNode(EngineTest.NODE, "n" + n))); + return null; + }); + } + }); + int checkpoints = 0; + while (writer.isAlive()) { + engine.checkpoint(); + checkpoints++; + LockSupport.parkNanos(20_000_000); + } + writer.join(); + engine.halt(); + assertTrue(checkpoints > 1, "only " + checkpoints + " checkpoints overlapped the writer"); + try (StorageEngine recovered = StorageEngine.open(directory, options)) { + recovered.read(snapshot -> { + assertEquals(400, snapshot.requireEdge(edge).size()); + assertEquals(400L, snapshot.countOfType(0, EngineTest.NODE)); + snapshot.requireEdge(edge).stream().forEach(member -> assertEquals(1, snapshot.degree(member.member()))); + return null; + }); + recovered.verifyReferences(); + } + } + private Path lastSegment() throws Exception { try (Stream files = Files.list(directory.resolve("data").resolve("segments"))) { return files.filter(file -> file.getFileName().toString().endsWith(".seg")).max(Comparator.naturalOrder()).orElseThrow();