Keep the parts of the old compaction work that still help

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()));

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);
+        }
+    }
 }

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<P> typed(Object[] decoded) {
+        for (Object value : decoded) {
+            postings.cast(value);
+        }
+        return (List<P>) List.of(decoded);
+    }
+
     @Override
     public boolean holdsRefs() {
         return true;
@@ -97,6 +104,17 @@ public boolean refersToPages(Postings<P> value) {
 
     @Override
     public Postings<P> mapRefs(Postings<P> value, RefMapper mapper) {
-        return value instanceof Postings.Promoted<P>(Ref root) ? new Postings.Promoted<>(mapper.map(root, postings)) : value;
+        if (!(value instanceof Postings.Promoted<P>(Ref root))) {
+            return value;
+        }
+        Ref mapped = mapper.map(root, postings);
+        return mapped == root ? value : new Postings.Promoted<>(mapped);
+    }
+
+    @Override
+    public void forEachRef(Postings<P> value, RefVisitor visitor) {
+        if (value instanceof Postings.Promoted<P>(Ref root)) {
+            visitor.visit(root, postings);
+        }
     }
 }

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 <V> void nested(TreeSchema<V> schema, Object value, ValueCodec.RefMapper mapper) {
-        schema.codec().mapRefs(schema.cast(value), mapper);
+    private static <V> void nested(TreeSchema<V> schema, Object value, ValueCodec.RefVisitor visitor) {
+        schema.codec().forEachRef(schema.cast(value), visitor);
     }
 }

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 <V> ValueCodec<V> rows(ToIntFunction<V> size, BiConsumer<ByteCursor, V> writer, Function<ByteCursor, V> 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);
+    }
 }

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<Path> files = Files.list(directory.resolve("data").resolve("segments"))) {
             return files.filter(file -> file.getFileName().toString().endsWith(".seg")).max(Comparator.naturalOrder()).orElseThrow();

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<Ref> mapped = new ArrayList<>();
+        AtomCodec.INSTANCE.mapRefs(EDGE, (ref, _) -> {
+            mapped.add(ref);
+            return ref;
+        });
+        List<Ref> visited = new ArrayList<>();
+        AtomCodec.INSTANCE.forEachRef(EDGE, (ref, _) -> visited.add(ref));
+        assertEquals(mapped, visited);
+        List<Ref> none = new ArrayList<>();
+        AtomCodec.INSTANCE.forEachRef(new AtomRecord.NodeRecord(1, "k", 0, 0, 0, 0, 0), (ref, _) -> none.add(ref));
+        assertEquals(List.of(), none);
+    }
+}

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<Long> 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<Long> 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<Long> 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<Long> changed = CODEC.mapRefs(promoted, (ref, _) -> moved);
+        assertNotSame(promoted, changed);
+        assertEquals(new Postings.Promoted<>(moved), changed);
+        List<Ref> visited = new ArrayList<>();
+        CODEC.forEachRef(promoted, (ref, schema) -> {
+            visited.add(ref);
+            assertSame(LONGS, schema);
+        });
+        assertEquals(List.of(ROOT), visited);
+    }
+}