Stop allocating per frame when the WAL is opened

docs/transactions/wal-and-recovery.md

@@ -70,7 +70,7 @@ The checksum covers the LSN, type, transaction id and payload. Embedding the LSN
 
 **Sync.** `sync()` calls `FileChannel.force(false)` on the active segment if anything was written since the last sync (`dirty`). Rolled segments were forced when they were closed, so one call makes the entire log durable.
 
-**Open.** `restore()` lists `*.wal`, scans only the **last** segment frame by frame, truncates it to the end of the last valid frame and forces it. Recovery therefore always appends after a clean frame boundary.
+**Open.** `restore()` lists `*.wal`, scans only the **last** segment frame by frame, truncates it to the end of the last valid frame and forces it. Recovery therefore always appends after a clean frame boundary. The scan checks each frame's header and CRC-32C but doesn't decode the records, and a reader reuses one header and one body buffer, growing the body only for a larger frame, so opening doesn't allocate per frame.
 
 **Read.** `read(from)` iterates frames starting at LSN `from` across segments. When fewer than 25 header bytes remain in a segment, the reader treats it as the clean end of that segment and continues with the next one. A *damaged* frame is one whose length is out of range, whose embedded LSN differs from the expected one, whose body is short, or whose CRC does not match. What happens next depends on where it is:
 

engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java

@@ -69,7 +69,7 @@ private void restore() {
             }
             Map.Entry<Long, Path> last = segments.lastEntry();
             long validEnd = last.getKey();
-            for (Positioned positioned : (Iterable<Positioned>) () -> new Reader(last.getKey())) {
+            for (Positioned positioned : (Iterable<Positioned>) () -> new Reader(last.getKey(), false)) {
                 validEnd = positioned.next();
             }
             activeStart = last.getKey();
@@ -184,7 +184,7 @@ public long bytesAppended() {
     }
 
     public Stream<Positioned> read(long from) {
-        return StreamSupport.stream(Spliterators.spliteratorUnknownSize(new Reader(from),
+        return StreamSupport.stream(Spliterators.spliteratorUnknownSize(new Reader(from, true),
                 Spliterator.ORDERED | Spliterator.NONNULL), false);
     }
 
@@ -240,8 +240,12 @@ private final class Reader implements Iterator<Positioned> {
         private long lsn;
         private Positioned next;
         private boolean finished;
+        private final ByteBuffer header = ByteBuffer.allocate(HEADER);
+        private final boolean decode;
+        private ByteBuffer body = ByteBuffer.allocate(4096);
 
-        Reader(long from) {
+        Reader(long from, boolean decode) {
+            this.decode = decode;
             Long floor = segments.floorKey(from);
             this.files = segments.tailMap(floor == null ? Long.MIN_VALUE : floor, true).entrySet().iterator();
             this.lsn = from;
@@ -293,7 +297,7 @@ private boolean openNext() {
 
         private Positioned readFrame() {
             try {
-                ByteBuffer header = ByteBuffer.allocate(HEADER);
+                header.clear();
                 if (!readFully(header, lsn - segmentStart)) {
                     return header.position() == 0 ? null : damaged("frame header is truncated");
                 }
@@ -306,7 +310,11 @@ private Positioned readFrame() {
                 if (payload < 0 || payload > MAX_RECORD || frameLsn != lsn) {
                     return damaged("frame header is invalid");
                 }
-                ByteBuffer body = ByteBuffer.allocate(HEADER - 8 + payload);
+                int size = HEADER - 8 + payload;
+                if (body.capacity() < size) {
+                    body = ByteBuffer.allocate(Math.max(size, body.capacity() * 2));
+                }
+                body.clear().limit(size);
                 body.put(header.position(8));
                 if (!readFully(body, lsn - segmentStart + HEADER)) {
                     return damaged("frame is truncated");
@@ -315,7 +323,7 @@ private Positioned readFrame() {
                 if (Checksums.crc32c(frame) != crc) {
                     return damaged("frame checksum does not match");
                 }
-                WalRecord record = WalRecord.readPayload(type, txnId, ByteCursor.over(frame).position(HEADER - 8));
+                WalRecord record = decode ? WalRecord.readPayload(type, txnId, ByteCursor.over(frame).position(HEADER - 8)) : null;
                 long start = lsn;
                 lsn += HEADER + payload;
                 return new Positioned(start, lsn, record);

engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java

@@ -0,0 +1,106 @@
+package io.hstore.engine.wal;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.random.RandomGenerator;
+import java.util.random.RandomGeneratorFactory;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class WriteAheadLogTest {
+
+    private static final long SEGMENT_BYTES = 64 * 1024;
+
+    @TempDir
+    Path directory;
+
+    private static List<WalRecord> records(int count, long seed) {
+        RandomGenerator random = RandomGeneratorFactory.of("L64X128MixRandom").create(seed);
+        List<WalRecord> records = new ArrayList<>();
+        for (int i = 0; i < count; i++) {
+            byte[] bytes = new byte[random.nextInt(random.nextInt(10) == 0 ? 20_000 : 200)];
+            random.nextBytes(bytes);
+            records.add(switch (i % 4) {
+                case 0 -> new WalRecord.Begin(i, random.nextInt(4), random.nextLong(1_000_000));
+                case 1 -> new WalRecord.Page(i, random.nextLong(), bytes);
+                case 2 -> new WalRecord.Feed(i, bytes);
+                default -> new WalRecord.Commit(i, i, random.nextLong(), random.nextLong(1 << 30), random.nextLong(1 << 30));
+            });
+        }
+        return records;
+    }
+
+    private static String describe(WalRecord record) {
+        return switch (record) {
+            case WalRecord.Page(long txn, long page, byte[] image) -> "page " + txn + " " + page + " " + HexFormat.of().formatHex(image);
+            case WalRecord.Feed(long txn, byte[] payload) -> "feed " + txn + " " + HexFormat.of().formatHex(payload);
+            default -> record.toString();
+        };
+    }
+
+    private static List<String> replay(WriteAheadLog wal) {
+        try (Stream<WriteAheadLog.Positioned> stream = wal.read(0)) {
+            return stream.map(positioned -> describe(positioned.record())).toList();
+        }
+    }
+
+    @Test
+    void reopenedLogReplaysRecordsOfMixedSizesAcrossSegments() {
+        List<WalRecord> written = records(3_000, 7);
+        long end;
+        try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) {
+            written.forEach(wal::append);
+            end = wal.end();
+            assertTrue(wal.segmentCount() > 3, "segments " + wal.segmentCount());
+        }
+        try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) {
+            assertEquals(end, wal.end());
+            assertEquals(written.stream().map(WriteAheadLogTest::describe).toList(), replay(wal));
+        }
+    }
+
+    @Test
+    void aTornFinalFrameIsDroppedAndEverythingBeforeItKept() throws IOException {
+        List<WalRecord> written = records(400, 11);
+        long lastStart;
+        try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) {
+            written.subList(0, written.size() - 1).forEach(wal::append);
+            lastStart = wal.end();
+            wal.append(written.getLast());
+        }
+        Path last;
+        try (Stream<Path> files = Files.list(directory)) {
+            last = files.filter(path -> path.toString().endsWith(".wal")).max(Path::compareTo).orElseThrow();
+        }
+        try (FileChannel channel = FileChannel.open(last, StandardOpenOption.WRITE)) {
+            channel.truncate(channel.size() - 3);
+        }
+        try (WriteAheadLog wal = WriteAheadLog.open(directory, SEGMENT_BYTES)) {
+            assertEquals(lastStart, wal.end());
+            assertEquals(written.subList(0, written.size() - 1).stream().map(WriteAheadLogTest::describe).toList(), replay(wal));
+        }
+    }
+
+    @Test
+    void aSmallFrameAfterALargeOneReadsBackIntact() {
+        List<WalRecord> written = List.of(new WalRecord.Feed(1, new byte[]{1, 2, 3}), new WalRecord.Page(2, 9, new byte[30_000]),
+                new WalRecord.Feed(3, new byte[]{4, 5}), new WalRecord.Commit(4, 4, 5, 6, 7), new WalRecord.Feed(5, new byte[0]));
+        try (WriteAheadLog wal = WriteAheadLog.open(directory, 1 << 20)) {
+            written.forEach(wal::append);
+        }
+        try (WriteAheadLog wal = WriteAheadLog.open(directory, 1 << 20)) {
+            assertEquals(written.stream().map(WriteAheadLogTest::describe).toList(), replay(wal));
+        }
+    }
+}