From dee4e8e0edc378d900934ea2d82d9b0c84f9accd Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 09:13:49 +0530 Subject: [PATCH 1/3] perf(engine): copy relocated leaf pages as bytes Compaction decoded every moved leaf into objects, copied them and encoded them again. A leaf of a tree without nested references can be copied as it is: verify it against its old id, re-stamp the header with the new page id and checksum, and write it. Only branches and leaves holding nested trees are still rebuilt. Moves are paired after the copies are written, since copied pages never go through the node cache. --- .../io/hstore/engine/tree/TreeWalker.java | 28 ++++++++++++---- .../hstore/engine/txn/TransactionManager.java | 33 +++++++++++++++---- 2 files changed, 49 insertions(+), 12 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 8d3bef7..bd11609 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java @@ -5,6 +5,7 @@ import java.util.Map; import java.util.function.LongConsumer; import java.util.function.LongPredicate; +import java.util.function.UnaryOperator; public final class TreeWalker { @@ -50,20 +51,35 @@ 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, UNKNOWN_HEIGHT, moving, scope, reached); + return relocate(root, schema, moving, scope, reached, leaf -> leaf); } - private Ref relocate(Ref ref, TreeSchema schema, int height, LongPredicate inVictim, WriteScope scope, LongConsumer reached) { - if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _))) { + 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); + } + + private Ref relocate(Ref ref, TreeSchema schema, int height, LongPredicate inVictim, WriteScope scope, LongConsumer reached, + UnaryOperator copyLeaf) { + if (!(ref instanceof Ref.Stored stored)) { return ref; } + long pageId = stored.pageId(); reached.accept(pageId); boolean moving = inVictim.test(pageId); - if (height == 0 && !moving && !schema.codec().holdsRefs()) { - return ref; + if (!schema.codec().holdsRefs()) { + if (height == 0 && !moving) { + return ref; + } + if (moving && height <= 0) { + Ref.Stored copy = copyLeaf.apply(stored); + if (copy != stored) { + return copy; + } + } } return rebuild(ref, pageId, schema, moving, scope, (child, childSchema, childHeight) -> - relocate(child, childSchema, childHeight, inVictim, scope, reached)); + relocate(child, childSchema, childHeight, inVictim, scope, reached, copyLeaf)); } 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 19bc1ea..0e6c6ba 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -12,6 +12,8 @@ import io.hstore.engine.feed.CommitEvent; import io.hstore.engine.feed.FeedCodec; import io.hstore.engine.page.Checksums; +import io.hstore.engine.page.PageHeader; +import io.hstore.engine.page.PageId; import io.hstore.engine.page.PageStore; import io.hstore.engine.topology.MemberChange; import io.hstore.engine.tree.Hashing; @@ -326,12 +328,15 @@ public long relocate(LongPredicate moving) { List writes = new ArrayList<>(); Materializer materializer = materializer(snapshot.id() + 1, txnId, records, writes); WriteScope copies = new WriteScope(); + List pairings = new ArrayList<>(); 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, reached::add), schema); - walker.pairMoves(ref, copy, schema, moved); + Ref copy = materializer.materialize(walker.relocate(ref, schema, moving, copies, reached::add, + 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())); + pairings.forEach(Runnable::run); pagesWritten.add(materializer.pagesWritten()); Set swapped = new HashSet<>(); List unlogged = records; @@ -409,15 +414,31 @@ public long allocate(int length) { @Override public void accept(long pageId, MemorySegment image, Node frozen) { - 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)); + stage(txnId, pageId, image, records, writes); storage.source().admit(pageId, frozen); } }, epoch, storage.pages().pageSize()); } + private void stage(long txnId, long pageId, MemorySegment image, List records, List 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)); + } + + private Ref.Stored copyPage(Ref.Stored stored, long epoch, long txnId, List records, List writes) { + MemorySegment page = storage.pages().read(stored.pageId()); + if (PageHeader.verify(page, stored.pageId()).height() != 0) { + return stored; + } + MemorySegment image = page.asSlice(0, PageHeader.SIZE + PageHeader.payloadLength(page)); + long pageId = storage.pages().allocate(epoch, Math.toIntExact(image.byteSize())); + PageHeader.assign(image, pageId); + stage(txnId, pageId, image, records, writes); + return new Ref.Stored(pageId, PageId.unitsFor(image.byteSize()), stored.summary()); + } + private Appended append(Generation latest, Branch changed, long txnId, List members, List> slotChanges, List meta, long pageBudget) { long pendingPages = changed.roots().roots().entrySet().stream() From 7aedab899fbc678ffc18933cced69d6def7c8cc4 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Wed, 7 Oct 2026 09:35:31 +0530 Subject: [PATCH 2/3] test(engine): a corrupt page is not copied with a fresh checksum A corrupted membership leaf makes relocation fail instead of being re-sealed with a valid checksum; the test fails if verification is skipped. --- .../java/io/hstore/engine/CompactionTest.java | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java index 0f00557..c400cd2 100644 --- a/engine/src/test/java/io/hstore/engine/CompactionTest.java +++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java @@ -5,11 +5,14 @@ import io.hstore.engine.catalog.Branch; import io.hstore.engine.maintenance.Compactor; +import io.hstore.engine.page.PageHeader; import io.hstore.engine.page.PageId; import io.hstore.engine.page.SegmentInfo; import io.hstore.engine.page.SegmentState; import io.hstore.engine.topology.EdgeKind; import io.hstore.engine.topology.Incidence; +import io.hstore.engine.tree.Leaf; +import io.hstore.engine.tree.Ref; import io.hstore.engine.tree.TreeWalker; import io.hstore.engine.txn.CrashPoint; import io.hstore.engine.txn.Snapshot; @@ -19,7 +22,10 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; import java.nio.file.Path; +import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -238,6 +244,36 @@ void commitsOnTwoBranchesDuringCompactionAreKeptAndOldSegmentsGoAway() throws Ex } } + @Test + void aCorruptPageIsNotCopiedWithAFreshChecksum() throws Exception { + try (StorageEngine engine = StorageEngine.open(directory, options())) { + Map> oracle = load(engine); + Set sealed = engine.stats().segments().stream().filter(segment -> segment.state() == SegmentState.SEALED) + .map(SegmentInfo::id).collect(Collectors.toSet()); + long victim; + try (Snapshot snapshot = engine.snapshot(Branch.MAIN)) { + victim = oracle.keySet().stream() + .map(edge -> snapshot.requireEdge(edge).membership().root()) + .filter(root -> root instanceof Ref.Stored(long pageId, int _, var _) && sealed.contains(PageId.segmentOf(pageId))) + .mapToLong(root -> ((Ref.Stored) root).pageId()) + .filter(pageId -> engine.transactions().source().load(pageId, snapshot.requireEdge(oracle.keySet().iterator().next()).membership().schema()) instanceof Leaf) + .findFirst().orElse(0L); + } + assertTrue(victim != 0, "no single-leaf membership tree in a sealed segment"); + Path segment = directory.resolve("data").resolve("segments").resolve("%08x.seg".formatted(PageId.segmentOf(victim))); + try (FileChannel channel = FileChannel.open(segment, StandardOpenOption.READ, StandardOpenOption.WRITE)) { + long at = (long) PageId.offsetOf(victim) * PageId.UNIT_BYTES + PageHeader.SIZE + 1; + ByteBuffer one = ByteBuffer.allocate(1); + channel.read(one, at); + channel.write(ByteBuffer.wrap(new byte[]{(byte) (one.get(0) ^ 0x5A)}), at); + } + int corrupted = PageId.segmentOf(victim); + HStoreException failure = assertThrows(HStoreException.class, + () -> engine.transactions().relocate(pageId -> PageId.segmentOf(pageId) == corrupted)); + assertTrue(failure.getMessage().toLowerCase().contains("checksum"), failure.getMessage()); + } + } + @ParameterizedTest @EnumSource(value = CrashPoint.class, names = {"PAGE", "WAL_APPEND", "DATA_WRITE", "COMMIT_APPEND", "DATA_SYNC", "WAL_SYNC", "CATALOG_PUBLISH"}) void aCrashWhileSwappingInTheCopiesLosesNothing(CrashPoint point) { From 9cf85d2e7c7f65a82678f0638c7385a405ee544f Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Fri, 9 Oct 2026 02:20:47 +0530 Subject: [PATCH 3/3] fix(engine): copy a relocated page before giving it a new id copyPage wrote the new page id straight into the image that PageStore.read handed back. That only works while read returns a private buffer. Once pages are read through the segment map they come back read-only, and every compaction failed with "Attempt to write a read-only segment". Copy the page into its own array first. --- .../src/main/java/io/hstore/engine/txn/TransactionManager.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 0e6c6ba..a9b2886 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -432,7 +432,7 @@ private Ref.Stored copyPage(Ref.Stored stored, long epoch, long txnId, List