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; 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/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/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 7f45720..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 { @@ -18,17 +19,57 @@ 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 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; + } + }; + } + + @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); - cache.put(pageId, node); + if (admit) { + cache.put(pageId, node); + } return node; } 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..233c9af 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; @@ -49,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++) { @@ -72,16 +77,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 +104,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 f289e20..b602913 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; @@ -322,25 +323,25 @@ 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<>(); - 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) -> { 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)); })); - 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 { @@ -398,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); @@ -406,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) { @@ -416,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; @@ -452,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()); diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java index 3fd337f..badc878 100644 --- a/engine/src/test/java/io/hstore/engine/CompactionTest.java +++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java @@ -11,8 +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; @@ -121,14 +126,41 @@ 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); + } + } + + @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()); } } 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(); 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); + } +}