Keep commits fast while compaction runs

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<P> typed(Object[] decoded) {
+        for (Object value : decoded) {
+            postings.cast(value);
+        }
+        return (List<P>) List.of(decoded);
+    }
+
     @Override
     public boolean holdsRefs() {
         return true;

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

engine/src/main/java/io/hstore/engine/maintenance/Compactor.java

@@ -36,7 +36,7 @@ public Compactor(TransactionManager transactions, PageStore pages, SlotRegistry
 
     public Map<Integer, Long> 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) ->

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) {

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

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 <V> void visitRefs(TreeSchema<V> 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]);

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

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

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<Ref.Stored> copyLeaf) {
-        return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope, reached, copyLeaf);
+                        UnaryOperator<Ref.Stored> copyLeaf, BiFunction<Ref, TreeSchema<?>, 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<Ref.Stored> copyLeaf) {
+                         UnaryOperator<Ref.Stored> copyLeaf, BiFunction<Ref, TreeSchema<?>, 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<Long, Ref> moved) {