From 76298f4524825b72c960566dbc88391957777037 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 13:04:34 +0530 Subject: [PATCH 1/7] perf(engine): keep compaction reads out of the node cache The liveness walk and relocation decode every page they reach. On a cold cache all of those nodes were cached, survived the next young collection and were copied into the old generation, which showed up as one long GC pause while commits waited. Compaction now uses pages that are already cached but no longer adds the ones it decodes. --- .../hstore/engine/maintenance/Compactor.java | 2 +- .../hstore/engine/tree/PagedNodeSource.java | 22 ++++++++++++++++++- .../hstore/engine/txn/TransactionManager.java | 2 +- .../java/io/hstore/engine/CompactionTest.java | 15 +++++++++++++ 4 files changed, 38 insertions(+), 3 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java index fd519b1..8d2bb1f 100644 --- a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java +++ b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java @@ -36,7 +36,7 @@ public Compactor(TransactionManager transactions, PageStore pages, SlotRegistry public Map liveness() { VisitedPages visited = new VisitedPages(pages.segments().stream().mapToLong(SegmentInfo::pages).sum()); - TreeWalker walker = new TreeWalker(transactions.source()); + TreeWalker walker = new TreeWalker(transactions.source().scanning()); Stream.concat(transactions.history().stream(), Stream.of(transactions.current())) .flatMap(generation -> generation.branches().values().stream()) .forEach(branch -> branch.roots().roots().forEach((slot, ref) -> diff --git a/engine/src/main/java/io/hstore/engine/tree/PagedNodeSource.java b/engine/src/main/java/io/hstore/engine/tree/PagedNodeSource.java index 7f45720..5114978 100644 --- a/engine/src/main/java/io/hstore/engine/tree/PagedNodeSource.java +++ b/engine/src/main/java/io/hstore/engine/tree/PagedNodeSource.java @@ -18,6 +18,24 @@ public PagedNodeSource(PageStore store, NodeCache cache) { @Override public Node load(long pageId, TreeSchema schema) { + return load(pageId, schema, true); + } + + public NodeSource scanning() { + return new NodeSource() { + @Override + public Node load(long pageId, TreeSchema schema) { + return PagedNodeSource.this.load(pageId, schema, false); + } + + @Override + public Layout layout() { + return layout; + } + }; + } + + private Node load(long pageId, TreeSchema schema, boolean admit) { Node cached = cache.get(pageId); if (cached != null) { if (cached.schema.id() != schema.id()) { @@ -28,7 +46,9 @@ public Node load(long pageId, TreeSchema schema) { } IoTrace.recordRead(); Node node = NodeCodec.decode(store.read(pageId), pageId, schema); - cache.put(pageId, node); + if (admit) { + cache.put(pageId, node); + } return node; } diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java index f289e20..247e86a 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -322,7 +322,7 @@ public long relocate(LongPredicate moving) { } finally { commitLock.unlock(); } - TreeWalker walker = new TreeWalker(storage.source()); + TreeWalker walker = new TreeWalker(storage.source().scanning()); VisitedPages reached = new VisitedPages(); Map moved = new HashMap<>(); List records = new ArrayList<>(); diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java index 3fd337f..4efb9b6 100644 --- a/engine/src/test/java/io/hstore/engine/CompactionTest.java +++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java @@ -12,6 +12,7 @@ import io.hstore.engine.topology.EdgeKind; import io.hstore.engine.topology.Incidence; import io.hstore.engine.tree.Leaf; +import io.hstore.engine.tree.NodeCache; import io.hstore.engine.tree.Ref; import io.hstore.engine.tree.TreeWalker; import io.hstore.engine.txn.CrashPoint; @@ -132,6 +133,20 @@ void livenessCountsEveryPageOfEveryRetainedRootOnce() { } } + @Test + void livenessWalkLeavesTheNodeCacheAsItWas() { + try (StorageEngine engine = StorageEngine.open(directory, options())) { + Map> oracle = load(engine); + churn(engine, Branch.MAIN, oracle, RandomGeneratorFactory.of("L64X128MixRandom").create(7), 10); + NodeCache cache = engine.transactions().source().cache(); + cache.clear(); + Map first = engine.liveness(); + long hits = cache.hits(); + assertEquals(first, engine.liveness()); + assertEquals(hits, cache.hits()); + } + } + @Test void commitsGoThroughWhileCompactionCopiesPages() throws Exception { try (StorageEngine engine = StorageEngine.open(directory, options())) { From 4f2a61f774c4521c09d352c00a384c105af70eec Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 13:51:24 +0530 Subject: [PATCH 2/7] perf(engine): log relocated pages before taking the commit lock The swap appended one WAL record per copied page while holding the commit lock, so commits waited for tens of thousands of records to be encoded and written. Recovery matches records to their commit by transaction id, so the records are now logged right after the copies are written and the swap only appends the new roots. A relocation that swaps nothing logs an abort. --- .../io/hstore/engine/txn/TransactionManager.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java index 247e86a..fef698c 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -337,10 +337,11 @@ public long relocate(LongPredicate moving) { pairings.add(() -> walker.pairMoves(ref, copy, schema, moved)); })); writes.forEach(write -> storage.pages().write(write.pageId(), write.image())); + storage.wal().append(records); pairings.forEach(Runnable::run); pagesWritten.add(materializer.pagesWritten()); Set swapped = new HashSet<>(); - List unlogged = records; + boolean committed = false; long generation = snapshot.id(); while (true) { Appended appended = null; @@ -351,6 +352,9 @@ public long relocate(LongPredicate moving) { .filter(Branch::isActive).filter(candidate -> !swapped.contains(candidate.id())) .findFirst().orElse(null); if (branch == null) { + if (!committed && !records.isEmpty()) { + logAbort(txnId); + } return generation; } swapped.add(branch.id()); @@ -358,9 +362,9 @@ public long relocate(LongPredicate moving) { RootVector roots = branch.roots().map((slot, ref) -> walker.adopt(ref, storage.slots().slot(slot).schema(), moved, reached::contains, merge)); if (!roots.sameAs(branch.roots())) { - appended = append(latest, branch.withRoots(roots), unlogged.isEmpty() ? txns.getAndIncrement() : txnId, - List.of(), List.of(), unlogged, Long.MAX_VALUE); - unlogged = List.of(); + appended = append(latest, branch.withRoots(roots), committed ? txns.getAndIncrement() : txnId, + List.of(), List.of(), List.of(), Long.MAX_VALUE); + committed = true; relocationFence = appended.pending().generation().id(); } } finally { From 0e5dd85832865e42037ab9f462f683a4f4eef842 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 13:51:39 +0530 Subject: [PATCH 3/7] perf(engine): write relocated pages as soon as they are copied Relocation kept every copied page image in a list until the whole walk had finished, so a full pass held about as much heap as the live data it moved. The copies sit at fresh locations that no committed tree points to until the swap, so they are written straight away. Pages rebuilt for relocation are no longer added to the node cache either. --- .../hstore/engine/txn/TransactionManager.java | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java index fef698c..afcc0d7 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -47,6 +47,7 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.LongAdder; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; import java.util.function.Function; import java.util.function.LongPredicate; import java.util.function.Supplier; @@ -326,8 +327,8 @@ public long relocate(LongPredicate moving) { VisitedPages reached = new VisitedPages(); Map moved = new HashMap<>(); List records = new ArrayList<>(); - List writes = new ArrayList<>(); - Materializer materializer = materializer(snapshot.id() + 1, txnId, records, writes); + Consumer writes = write -> storage.pages().write(write.pageId(), write.image()); + Materializer materializer = materializer(snapshot.id() + 1, txnId, records, writes, false); WriteScope copies = new WriteScope(); List pairings = new ArrayList<>(); snapshot.branches().values().stream().filter(Branch::isActive).forEach(branch -> branch.roots().roots().forEach((slot, ref) -> { @@ -336,7 +337,6 @@ public long relocate(LongPredicate moving) { leaf -> copyPage(leaf, snapshot.id() + 1, txnId, records, writes)), schema); pairings.add(() -> walker.pairMoves(ref, copy, schema, moved)); })); - writes.forEach(write -> storage.pages().write(write.pageId(), write.image())); storage.wal().append(records); pairings.forEach(Runnable::run); pagesWritten.add(materializer.pagesWritten()); @@ -402,7 +402,7 @@ void logAbort(long txnId) { Ref spill(long txnId, Ref ref, TreeSchema schema) { List records = new ArrayList<>(); List writes = new ArrayList<>(); - Materializer materializer = materializer(head.get().id() + 1, txnId, records, writes); + Materializer materializer = materializer(head.get().id() + 1, txnId, records, writes::add, true); Ref stored = materializer.materialize(ref, schema); writes.forEach(write -> storage.pages().write(write.pageId(), write.image())); storage.wal().append(records); @@ -410,7 +410,7 @@ Ref spill(long txnId, Ref ref, TreeSchema schema) { return stored; } - private Materializer materializer(long epoch, long txnId, List records, List writes) { + private Materializer materializer(long epoch, long txnId, List records, Consumer writes, boolean admit) { return new Materializer(new Materializer.Sink() { @Override public long allocate(int length) { @@ -420,19 +420,21 @@ public long allocate(int length) { @Override public void accept(long pageId, MemorySegment image, Node frozen) { stage(txnId, pageId, image, records, writes); - storage.source().admit(pageId, frozen); + if (admit) { + storage.source().admit(pageId, frozen); + } } }, epoch, storage.pages().pageSize()); } - private void stage(long txnId, long pageId, MemorySegment image, List records, List writes) { + private void stage(long txnId, long pageId, MemorySegment image, List records, Consumer writes) { records.add(storage.walMode() == WalMode.PAGE_IMAGES ? new WalRecord.Page(txnId, pageId, image.toArray(ValueLayout.JAVA_BYTE)) : new WalRecord.PageRef(txnId, pageId, Math.toIntExact(image.byteSize()), Checksums.crc32c(image))); - writes.add(new PageWrite(pageId, image)); + writes.accept(new PageWrite(pageId, image)); } - private Ref.Stored copyPage(Ref.Stored stored, long epoch, long txnId, List records, List writes) { + private Ref.Stored copyPage(Ref.Stored stored, long epoch, long txnId, List records, Consumer writes) { MemorySegment page = storage.pages().read(stored.pageId()); if (PageHeader.verify(page, stored.pageId()).height() != 0) { return stored; @@ -456,7 +458,7 @@ private Appended append(Generation latest, Branch changed, long txnId, List records = new ArrayList<>(); List writes = new ArrayList<>(); records.add(new WalRecord.Begin(txnId, changed.id(), latest.id())); - Materializer materializer = materializer(generationId, txnId, records, writes); + Materializer materializer = materializer(generationId, txnId, records, writes::add, true); RootVector stored = changed.roots().map((slot, ref) -> materializer.materialize(ref, storage.slots().slot(slot).schema())); storage.faults().reach(CrashPoint.PAGE); Branch previous = latest.branches().get(changed.id()); From 3e79d9d64cc186a42270df2692cf6d6850f2f330 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 13:51:39 +0530 Subject: [PATCH 4/7] perf(engine): write rebuilt nodes while relocating instead of after Relocation built the whole rewritten tree for a slot in memory before writing any of it, which for the atoms tree means most of its leaves. Each rebuilt node is now written as soon as its children are, so only the current path stays on the heap. --- .../main/java/io/hstore/engine/tree/TreeWalker.java | 13 +++++++------ .../io/hstore/engine/txn/TransactionManager.java | 2 +- 2 files changed, 8 insertions(+), 7 deletions(-) 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 b559739..6ac30ce 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java @@ -3,6 +3,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.function.BiFunction; import java.util.function.LongConsumer; import java.util.function.LongPredicate; import java.util.function.UnaryOperator; @@ -72,16 +73,16 @@ public Ref relocate(Ref root, TreeSchema schema, LongPredicate moving, WriteS } public Ref relocate(Ref root, TreeSchema schema, LongPredicate moving, WriteScope scope, LongConsumer reached) { - return relocate(root, schema, moving, scope, reached, leaf -> leaf); + return relocate(root, schema, moving, scope, reached, leaf -> leaf, (ref, _) -> ref); } public Ref relocate(Ref root, TreeSchema schema, LongPredicate moving, WriteScope scope, LongConsumer reached, - UnaryOperator copyLeaf) { - return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope, reached, copyLeaf); + UnaryOperator copyLeaf, BiFunction, Ref> settle) { + return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope, reached, copyLeaf, settle); } private Ref relocate(Ref ref, TreeSchema schema, int height, LongPredicate inVictim, WriteScope scope, LongConsumer reached, - UnaryOperator copyLeaf) { + UnaryOperator copyLeaf, BiFunction, Ref> settle) { if (!(ref instanceof Ref.Stored stored)) { return ref; } @@ -99,8 +100,8 @@ private Ref relocate(Ref ref, TreeSchema schema, int height, LongPredicate in } } } - return rebuild(ref, pageId, schema, moving, scope, (child, childSchema, childHeight) -> - relocate(child, childSchema, childHeight, inVictim, scope, reached, copyLeaf)); + return settle.apply(rebuild(ref, pageId, schema, moving, scope, (child, childSchema, childHeight) -> + relocate(child, childSchema, childHeight, inVictim, scope, reached, copyLeaf, settle)), schema); } public void pairMoves(Ref before, Ref after, TreeSchema schema, Map moved) { diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java index afcc0d7..b602913 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -334,7 +334,7 @@ public long relocate(LongPredicate moving) { snapshot.branches().values().stream().filter(Branch::isActive).forEach(branch -> branch.roots().roots().forEach((slot, ref) -> { TreeSchema schema = storage.slots().slot(slot).schema(); Ref copy = materializer.materialize(walker.relocate(ref, schema, moving, copies, pageId -> reached.add(pageId, 0), - leaf -> copyPage(leaf, snapshot.id() + 1, txnId, records, writes)), schema); + leaf -> copyPage(leaf, snapshot.id() + 1, txnId, records, writes), materializer::materialize), schema); pairings.add(() -> walker.pairMoves(ref, copy, schema, moved)); })); storage.wal().append(records); From 60fc0d01a44423bf1e296f00fefb614c7979c750 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 13:57:07 +0530 Subject: [PATCH 5/7] perf(engine): sync checkpoints without holding the commit lock A checkpoint fsynced every segment file, the feed, the catalog and the WAL while holding the commit lock, so commits stalled for as long as the disk took to flush. Right after a compaction that is hundreds of MiB. The checkpoint now only captures its position under the lock: the current generation, history, segment list, WAL end and feed size. The syncs and the catalog write happen afterwards while commits continue. Everything up to the captured generation was written before the capture, so the syncs cover it, and later commits replay from the WAL as before. --- .../engine/maintenance/Checkpointer.java | 20 +++++------ .../io/hstore/engine/CrashRecoveryTest.java | 34 +++++++++++++++++++ 2 files changed, 44 insertions(+), 10 deletions(-) diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java index befdc25..93b3773 100644 --- a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java +++ b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java @@ -40,22 +40,22 @@ public long walSinceCheckpoint() { } public Result checkpoint() { - return transactions.exclusive(() -> { - pages.sync(); - feed.sync(); + CatalogImage image = transactions.exclusive(() -> { Generation current = transactions.current(); - long lsn = wal.end(); List history = transactions.history().stream() .filter(generation -> generation.id() != current.id()) .sorted((a, b) -> Long.compare(b.id(), a.id())) .limit(historyLimit) .toList(); - catalog.publish(new CatalogImage(current, history, pages.segments(), lsn, feed.size())); - wal.append(new WalRecord.Checkpoint(0, current.id(), lsn)); - wal.sync(); - wal.truncateBefore(lsn); - lastLsn = lsn; - return new Result(current.id(), lsn, wal.segmentCount()); + return new CatalogImage(current, history, pages.segments(), wal.end(), feed.size()); }); + pages.sync(); + feed.sync(); + catalog.publish(image); + wal.append(new WalRecord.Checkpoint(0, image.current().id(), image.checkpointLsn())); + wal.sync(); + wal.truncateBefore(image.checkpointLsn()); + lastLsn = image.checkpointLsn(); + return new Result(image.current().id(), image.checkpointLsn(), wal.segmentCount()); } } diff --git a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java index 2a93987..158f938 100644 --- a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java +++ b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java @@ -10,6 +10,7 @@ import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.MethodSource; import java.nio.channels.FileChannel; @@ -124,6 +125,39 @@ 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 < 2_000; 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++; + } + writer.join(); + engine.halt(); + assertTrue(checkpoints > 1, "only " + checkpoints + " checkpoints overlapped the writer"); + try (StorageEngine recovered = StorageEngine.open(directory, options)) { + recovered.read(snapshot -> { + assertEquals(2_000, snapshot.requireEdge(edge).size()); + assertEquals(2_000L, snapshot.countOfType(0, EngineTest.NODE)); + snapshot.requireEdge(edge).stream().forEach(member -> assertEquals(1, snapshot.degree(member.member()))); + return null; + }); + } + } + private Path lastSegment() throws Exception { try (Stream files = Files.list(directory.resolve("data").resolve("segments"))) { return files.max(Comparator.naturalOrder()).orElseThrow(); From 76686acac7d8b980a425e7b28a0af9797ace17e9 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 14:22:32 +0530 Subject: [PATCH 6/7] perf(engine): skip leaves without nested trees in the liveness walk The liveness walk decoded every leaf of trees whose values can point to nested trees, such as the posting lists of the indexes, only to find that most of them point nowhere. On a cold cache that decoded every inline posting list and allocated about 3 GiB per full pass. Leaves are now marked in the unused flags byte of the page header when none of their values references a nested tree. The walk reads just the header of an uncached leaf and stops there when the mark is set. Pages written before this change carry no mark and are decoded as before. Relocation still decodes, since it needs checksummed pages to be safe. --- .../io/hstore/engine/page/PageHeader.java | 23 ++++++ .../java/io/hstore/engine/page/PageStore.java | 8 ++ .../java/io/hstore/engine/tree/NodeCodec.java | 18 ++++- .../io/hstore/engine/tree/NodeSource.java | 4 + .../hstore/engine/tree/PagedNodeSource.java | 23 +++++- .../io/hstore/engine/tree/TreeWalker.java | 6 +- .../java/io/hstore/engine/CompactionTest.java | 21 ++++- .../io/hstore/engine/tree/NodeCodecTest.java | 77 +++++++++++++++++++ 8 files changed, 175 insertions(+), 5 deletions(-) create mode 100644 engine/src/test/java/io/hstore/engine/tree/NodeCodecTest.java diff --git a/engine/src/main/java/io/hstore/engine/page/PageHeader.java b/engine/src/main/java/io/hstore/engine/page/PageHeader.java index 4d014d3..08ccec6 100644 --- a/engine/src/main/java/io/hstore/engine/page/PageHeader.java +++ b/engine/src/main/java/io/hstore/engine/page/PageHeader.java @@ -21,6 +21,7 @@ public record PageHeader( public static final int MAGIC = 0x47505348; public static final int FORMAT = 1; public static final int SIZE = 80; + public static final int NO_NESTED_REFS = 1; private static final int PAGE_ID_OFFSET = 8; private static final int PAYLOAD_LENGTH_OFFSET = 24; private static final int CHECKSUM_OFFSET = 72; @@ -60,6 +61,28 @@ public static void seal(MemorySegment page, int payloadLength) { ByteCursor.over(page).position(CHECKSUM_OFFSET).putInt(checksum); } + public static boolean isPlainLeaf(MemorySegment head, long expectedPageId) { + if (head.byteSize() < SIZE) { + throw HStoreException.corrupt(expectedPageId, "page header is truncated"); + } + ByteCursor in = ByteCursor.over(head); + if (in.getInt() != MAGIC) { + throw HStoreException.corrupt(expectedPageId, "bad page magic"); + } + int format = in.getUnsignedByte(); + if (format != FORMAT) { + throw HStoreException.corrupt(expectedPageId, "unsupported page format " + format); + } + PageType type = PageType.of(in.getUnsignedByte()); + in.getUnsignedByte(); + int flags = in.getUnsignedByte(); + long pageId = in.getLong(); + if (pageId != expectedPageId) { + throw HStoreException.corrupt(expectedPageId, "page identity mismatch, found " + PageId.unpack(pageId)); + } + return type == PageType.LEAF && (flags & NO_NESTED_REFS) != 0; + } + public static PageHeader verify(MemorySegment page, long expectedPageId) { ByteCursor in = ByteCursor.over(page); if (in.getInt() != MAGIC) { diff --git a/engine/src/main/java/io/hstore/engine/page/PageStore.java b/engine/src/main/java/io/hstore/engine/page/PageStore.java index f58464f..c2d679c 100644 --- a/engine/src/main/java/io/hstore/engine/page/PageStore.java +++ b/engine/src/main/java/io/hstore/engine/page/PageStore.java @@ -129,6 +129,14 @@ public MemorySegment read(long pageId) { return file.read((long) PageId.offsetOf(pageId) * PageId.UNIT_BYTES, pageSize); } + public MemorySegment readHeader(long pageId) { + SegmentFile file = files.get(PageId.segmentOf(pageId)); + if (file == null) { + throw HStoreException.corrupt(pageId, "segment does not exist"); + } + return file.read((long) PageId.offsetOf(pageId) * PageId.UNIT_BYTES, PageHeader.SIZE); + } + public void sync() { files.values().forEach(SegmentFile::force); } diff --git a/engine/src/main/java/io/hstore/engine/tree/NodeCodec.java b/engine/src/main/java/io/hstore/engine/tree/NodeCodec.java index 39579bc..ee28a2a 100644 --- a/engine/src/main/java/io/hstore/engine/tree/NodeCodec.java +++ b/engine/src/main/java/io/hstore/engine/tree/NodeCodec.java @@ -33,13 +33,29 @@ public static MemorySegment encode(Node node, long pageId, long epoch, int pageS } int payloadLength = Math.toIntExact(out.position() - PageHeader.SIZE); PageType type = node instanceof Leaf ? PageType.LEAF : PageType.INTERNAL; - new PageHeader(type, node.schema.id(), 0, pageId, epoch, payloadLength, node.size(), summary.count(), + int flags = node instanceof Leaf leaf && !holdsNestedRefs(leaf) ? PageHeader.NO_NESTED_REFS : 0; + new PageHeader(type, node.schema.id(), flags, pageId, epoch, payloadLength, node.size(), summary.count(), summary.min(), summary.max(), node.height(), summary.fingerprint()).writeTo(out); MemorySegment image = out.segment().asSlice(0, PageHeader.SIZE + payloadLength); PageHeader.seal(image, payloadLength); return image; } + private static boolean holdsNestedRefs(Leaf leaf) { + if (!leaf.schema.codec().holdsRefs()) { + return false; + } + boolean[] found = {false}; + for (int i = 0; i < leaf.size() && !found[0]; i++) { + visitRefs(leaf.schema, leaf.value(i), (ref, _) -> found[0] |= !(ref instanceof Ref.Empty)); + } + return found[0]; + } + + private static void visitRefs(TreeSchema schema, Object value, ValueCodec.RefVisitor visitor) { + schema.codec().forEachRef(schema.cast(value), visitor); + } + private static void encodeLeaf(ByteCursor out, Leaf leaf) { long[] keys = leaf.keyArray(); out.putSignedVarLong(keys[0]); diff --git a/engine/src/main/java/io/hstore/engine/tree/NodeSource.java b/engine/src/main/java/io/hstore/engine/tree/NodeSource.java index 312ea1e..3a6917b 100644 --- a/engine/src/main/java/io/hstore/engine/tree/NodeSource.java +++ b/engine/src/main/java/io/hstore/engine/tree/NodeSource.java @@ -6,6 +6,10 @@ public interface NodeSource { Layout layout(); + default Node loadUnlessPlainLeaf(long pageId, TreeSchema schema) { + return load(pageId, schema); + } + static NodeSource ephemeral(int pageSize) { Layout layout = Layout.of(pageSize); return new NodeSource() { diff --git a/engine/src/main/java/io/hstore/engine/tree/PagedNodeSource.java b/engine/src/main/java/io/hstore/engine/tree/PagedNodeSource.java index 5114978..cb486a1 100644 --- a/engine/src/main/java/io/hstore/engine/tree/PagedNodeSource.java +++ b/engine/src/main/java/io/hstore/engine/tree/PagedNodeSource.java @@ -2,6 +2,7 @@ import io.hstore.engine.HStoreException; import io.hstore.engine.page.IoTrace; +import io.hstore.engine.page.PageHeader; import io.hstore.engine.page.PageStore; public final class PagedNodeSource implements NodeSource { @@ -28,6 +29,12 @@ public Node load(long pageId, TreeSchema schema) { return PagedNodeSource.this.load(pageId, schema, false); } + @Override + public Node loadUnlessPlainLeaf(long pageId, TreeSchema schema) { + Node cached = cached(pageId, schema); + return cached != null || PageHeader.isPlainLeaf(store.readHeader(pageId), pageId) ? cached : decode(pageId, schema, false); + } + @Override public Layout layout() { return layout; @@ -35,15 +42,29 @@ public Layout layout() { }; } + @Override + public Node loadUnlessPlainLeaf(long pageId, TreeSchema schema) { + Node cached = cached(pageId, schema); + return cached != null || PageHeader.isPlainLeaf(store.readHeader(pageId), pageId) ? cached : decode(pageId, schema, true); + } + private Node load(long pageId, TreeSchema schema, boolean admit) { + Node cached = cached(pageId, schema); + return cached != null ? cached : decode(pageId, schema, admit); + } + + private Node cached(long pageId, TreeSchema schema) { Node cached = cache.get(pageId); if (cached != null) { if (cached.schema.id() != schema.id()) { throw HStoreException.corrupt(pageId, "cached page belongs to " + cached.schema + ", expected " + schema); } IoTrace.recordHit(); - return cached; } + return cached; + } + + private Node decode(long pageId, TreeSchema schema, boolean admit) { IoTrace.recordRead(); Node node = NodeCodec.decode(store.read(pageId), pageId, schema); if (admit) { 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 6ac30ce..233c9af 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java @@ -50,7 +50,11 @@ void walk(Ref ref, TreeSchema schema, int height) { if (height == 0 && !schema.codec().holdsRefs()) { return; } - switch (source.load(pageId, schema)) { + Node node = height > 0 ? source.load(pageId, schema) : source.loadUnlessPlainLeaf(pageId, schema); + if (node == null) { + return; + } + switch (node) { case Leaf leaf -> { if (schema.codec().holdsRefs()) { for (int i = 0; i < leaf.size(); i++) { diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java index 4efb9b6..badc878 100644 --- a/engine/src/test/java/io/hstore/engine/CompactionTest.java +++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java @@ -11,9 +11,13 @@ import io.hstore.engine.page.SegmentState; import io.hstore.engine.topology.EdgeKind; import io.hstore.engine.topology.Incidence; +import io.hstore.engine.tree.Layout; import io.hstore.engine.tree.Leaf; +import io.hstore.engine.tree.Node; import io.hstore.engine.tree.NodeCache; +import io.hstore.engine.tree.NodeSource; import io.hstore.engine.tree.Ref; +import io.hstore.engine.tree.TreeSchema; import io.hstore.engine.tree.TreeWalker; import io.hstore.engine.txn.CrashPoint; import io.hstore.engine.txn.Snapshot; @@ -122,14 +126,27 @@ void livenessCountsEveryPageOfEveryRetainedRootOnce() { try (StorageEngine engine = StorageEngine.open(directory, options())) { Map> oracle = load(engine); churn(engine, Branch.MAIN, oracle, RandomGeneratorFactory.of("L64X128MixRandom").create(6), 30); + engine.transactions().source().cache().clear(); + Map live = engine.liveness(); Map pages = new HashMap<>(); - TreeWalker walker = new TreeWalker(engine.transactions().source()); + NodeSource decoding = new NodeSource() { + @Override + public Node load(long pageId, TreeSchema schema) { + return engine.transactions().source().load(pageId, schema); + } + + @Override + public Layout layout() { + return engine.transactions().source().layout(); + } + }; + TreeWalker walker = new TreeWalker(decoding); Stream.concat(engine.transactions().history().stream(), Stream.of(engine.transactions().current())) .flatMap(generation -> generation.branches().values().stream()) .forEach(branch -> branch.roots().roots().forEach((slot, ref) -> walker.visit(ref, engine.slots().slot(slot).schema(), pages))); Map expected = new TreeMap<>(); pages.forEach((pageId, units) -> expected.merge(PageId.segmentOf(pageId), (long) units * PageId.UNIT_BYTES, Long::sum)); - assertEquals(expected, engine.liveness()); + assertEquals(expected, live); } } diff --git a/engine/src/test/java/io/hstore/engine/tree/NodeCodecTest.java b/engine/src/test/java/io/hstore/engine/tree/NodeCodecTest.java new file mode 100644 index 0000000..c779ad4 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/tree/NodeCodecTest.java @@ -0,0 +1,77 @@ +package io.hstore.engine.tree; + +import io.hstore.engine.page.ByteCursor; +import io.hstore.engine.page.PageHeader; +import io.hstore.engine.page.PageId; +import org.junit.jupiter.api.Test; + +import java.lang.foreign.MemorySegment; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class NodeCodecTest { + + private static final long PAGE = PageId.pack(2, 40, 1); + + private static final TreeSchema NESTED = new TreeSchema<>(201, "nested", FingerprintMode.SET, new ValueCodec<>() { + @Override + public int maxSize(Ref value) { + return Ref.encodedSize(value); + } + + @Override + public void encode(ByteCursor out, long[] keys, List values) { + values.forEach(value -> Ref.write(out, value)); + } + + @Override + public void decode(ByteCursor in, long[] keys, Object[] into) { + for (int i = 0; i < into.length; i++) { + into[i] = Ref.read(in); + } + } + + @Override + public boolean holdsRefs() { + return true; + } + + @Override + public Ref mapRefs(Ref value, RefMapper mapper) { + return mapper.map(value, TreeTest.LONGS); + } + + @Override + public void forEachRef(Ref value, RefVisitor visitor) { + visitor.visit(value, TreeTest.LONGS); + } + }, (key, _, into) -> into.entry(key, Hashing.mix(key))); + + @Test + void leavesWithoutNestedTreesAreMarkedPlain() { + assertTrue(plain(Leaf.owned(TreeTest.LONGS, new Object(), new long[]{1, 2}, new Object[]{10L, 20L}, 2))); + assertTrue(plain(Leaf.owned(NESTED, new Object(), new long[]{1, 2}, new Object[]{Ref.EMPTY, Ref.EMPTY}, 2))); + } + + @Test + void leavesWithANestedTreeAndBranchesAreNotPlain() { + Ref nested = new Ref.Stored(PageId.pack(5, 0, 1), 2, Summary.EMPTY); + Leaf leaf = Leaf.owned(NESTED, new Object(), new long[]{1, 2}, new Object[]{Ref.EMPTY, nested}, 2); + assertFalse(plain(leaf)); + Leaf decoded = (Leaf) NodeCodec.decode(encode(leaf), PAGE, NESTED); + assertEquals(nested, decoded.value(1)); + Ref child = new Ref.Stored(PageId.pack(5, 4, 1), 2, Summary.EMPTY); + assertFalse(plain(Branch.owned(TreeTest.LONGS, new Object(), new long[]{1, 9}, new Ref[]{child, child}, 2, 1))); + } + + private static boolean plain(Node node) { + return PageHeader.isPlainLeaf(encode(node).asSlice(0, PageHeader.SIZE), PAGE); + } + + private static MemorySegment encode(Node node) { + return NodeCodec.encode(node, PAGE, 1, 4096); + } +} From 53cbb21d34142787e722cff8771da98688d49fdb Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 14:22:32 +0530 Subject: [PATCH 7/7] 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 9c7b4f7..74ad554 100644 --- a/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java +++ b/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java @@ -6,7 +6,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; @@ -78,10 +77,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;