feat: extend sstable with writeRawRecord

src/sstable.zig

@@ -915,3 +915,198 @@ test "sstable reverse cursor honors upper bound" {
     defer below.deinit();
     try std.testing.expect(!below.valid());
 }
+
+fn writeRawRecord(w: encode.ByteWriter, kind: Kind, seq: u64, shared: u32, unshared: []const u8, value: []const u8) !void {
+    try w.byte(@intFromEnum(kind));
+    try w.putU64(seq);
+    try w.putU32(shared);
+    try w.putU32(@intCast(unshared.len));
+    try w.putU32(@intCast(value.len));
+    try w.slice(unshared);
+    try w.slice(value);
+}
+
+test "expandBlock reconstructs prefix-compressed records" {
+    const allocator = std.testing.allocator;
+    var buf = std.ArrayList(u8).init(allocator);
+    defer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+
+    try writeRawRecord(w, .value, 1, 0, "apple", "1");
+    try writeRawRecord(w, .value, 2, 3, "ly", "2"); // "app" + "ly" = "apply"
+    try writeRawRecord(w, .tombstone, 3, 2, "ricot", ""); // "ap" + "ricot" = "apricot"
+    try w.putU32(0); // one restart, at offset 0
+    try w.putU32(1);
+
+    const expanded = try expandBlock(allocator, buf.items);
+    defer allocator.free(expanded.bytes);
+    defer allocator.free(expanded.restarts);
+
+    const d0 = decodeRecord(expanded.bytes, 0);
+    try std.testing.expectEqualStrings("apple", d0.record.key);
+    try std.testing.expectEqualStrings("1", d0.record.value);
+    try std.testing.expectEqual(Kind.value, d0.record.kind);
+
+    const d1 = decodeRecord(expanded.bytes, d0.next);
+    try std.testing.expectEqualStrings("apply", d1.record.key);
+    try std.testing.expectEqualStrings("2", d1.record.value);
+
+    const d2 = decodeRecord(expanded.bytes, d1.next);
+    try std.testing.expectEqualStrings("apricot", d2.record.key);
+    try std.testing.expectEqual(Kind.tombstone, d2.record.kind);
+    try std.testing.expectEqual(d2.next, expanded.bytes.len);
+
+    // Only the first record is a restart in this fixture (interval=16,
+    // 3 records) - `flatRestarts` must record exactly its flat offset (0).
+    try std.testing.expectEqualSlices(u32, &.{0}, expanded.restarts);
+}
+
+test "expandBlock rejects a restart declared at the wrong record" {
+    const allocator = std.testing.allocator;
+    var buf = std.ArrayList(u8).init(allocator);
+    defer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+
+    const record0End = 1 + 8 + 4 + 4 + 4 + 5 + 1; // header + "apple" + "1"
+    try writeRawRecord(w, .value, 1, 0, "apple", "1");
+    try writeRawRecord(w, .value, 2, 3, "ly", "2");
+    try w.putU32(@intCast(record0End)); // restart claims record 1's offset, not record 0's
+    try w.putU32(1);
+
+    try std.testing.expectError(error.CorruptSstable, expandBlock(allocator, buf.items));
+}
+
+test "expandBlock rejects a missing restart declaration" {
+    const allocator = std.testing.allocator;
+    var buf = std.ArrayList(u8).init(allocator);
+    defer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+
+    try writeRawRecord(w, .value, 1, 0, "apple", "1");
+    try w.putU32(0); // trailer present but claims zero restarts
+
+    try std.testing.expectError(error.CorruptSstable, expandBlock(allocator, buf.items));
+}
+
+test "expandBlock rejects shared length exceeding the previous key" {
+    const allocator = std.testing.allocator;
+    var buf = std.ArrayList(u8).init(allocator);
+    defer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+
+    try writeRawRecord(w, .value, 1, 0, "ab", "1"); // prevKey becomes "ab", len 2
+    try writeRawRecord(w, .value, 2, 5, "xyz", "2"); // shared=5 > len("ab")
+    try w.putU32(0);
+    try w.putU32(1);
+
+    try std.testing.expectError(error.CorruptSstable, expandBlock(allocator, buf.items));
+}
+
+test "expandBlock rejects a record stream that overruns its declared lengths" {
+    const allocator = std.testing.allocator;
+    var buf = std.ArrayList(u8).init(allocator);
+    defer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+
+    try w.byte(@intFromEnum(Kind.value));
+    try w.putU64(1);
+    try w.putU32(0);
+    try w.putU32(100); // claims 100 unshared key bytes that are never written
+    try w.putU32(0);
+    try w.putU32(0);
+    try w.putU32(1);
+
+    try std.testing.expectError(error.CorruptSstable, expandBlock(allocator, buf.items));
+}
+
+test "expandBlock rejects trailing bytes that do not form a full record" {
+    const allocator = std.testing.allocator;
+    var buf = std.ArrayList(u8).init(allocator);
+    defer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+
+    try writeRawRecord(w, .value, 1, 0, "apple", "1");
+    try w.byte(0xff); // stray byte before the restart trailer
+    try w.putU32(0);
+    try w.putU32(1);
+
+    try std.testing.expectError(error.CorruptSstable, expandBlock(allocator, buf.items));
+}
+
+test "Reader.open rejects an sstable written by an unsupported format version" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    try buildFixture(allocator, tmp.dir, "test.sst", 10);
+
+    {
+        var file = try tmp.dir.openFile("test.sst", .{ .mode = .read_write });
+        defer file.close();
+        const size = try file.getEndPos();
+        // version u32 sits right after the magic u32 at the footer's start.
+        try file.seekTo(size - footerLen + 4);
+        var bogus: [4]u8 = undefined;
+        std.mem.writeInt(u32, &bogus, version + 1, encode.endian);
+        try file.writeAll(&bogus);
+    }
+
+    try std.testing.expectError(error.UnsupportedSstableVersion, Reader.open(allocator, tmp.dir, "test.sst", 1));
+}
+
+test "prefix compression measurably shrinks sstables with a shared key prefix" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    const count = 3000;
+
+    {
+        var builder = try Builder.init(allocator, tmp.dir, "shared.sst", count);
+        defer builder.deinit();
+        var buf: [64]u8 = undefined;
+        var i: usize = 0;
+        while (i < count) : (i += 1) {
+            const key = try std.fmt.bufPrint(&buf, "users\x00city\x00warsaw\x00pk{d:0>8}", .{i});
+            try builder.add(.{ .key = key, .value = "v", .seq = i + 1, .kind = .value });
+        }
+        try builder.finish();
+    }
+    {
+        var builder = try Builder.init(allocator, tmp.dir, "distinct.sst", count);
+        defer builder.deinit();
+        var buf: [64]u8 = undefined;
+        var i: usize = 0;
+        while (i < count) : (i += 1) {
+            // Same value payload and roughly the same key length as above,
+            // but scrambled so consecutive keys share no meaningful prefix -
+            // isolates the saving to prefix compression rather than the
+            // block-level RLE codec, which would compress a repeated value
+            // just as well in both files.
+            const scrambled = i *% 2654435761;
+            const key = try std.fmt.bufPrint(&buf, "{d:0>20}", .{scrambled});
+            try builder.add(.{ .key = key, .value = "v", .seq = i + 1, .kind = .value });
+        }
+        try builder.finish();
+    }
+
+    var sharedReader = try Reader.open(allocator, tmp.dir, "shared.sst", 1);
+    defer sharedReader.deinit();
+    var distinctReader = try Reader.open(allocator, tmp.dir, "distinct.sst", 2);
+    defer distinctReader.deinit();
+
+    try std.testing.expectEqual(@as(u64, count), sharedReader.numEntries);
+    try std.testing.expectEqual(@as(u64, count), distinctReader.numEntries);
+    // Prefix compression must shrink the file meaningfully below the
+    // no-shared-prefix baseline (measured ~25% smaller at this fixture
+    // size), not just avoid growing it. Bloom filter and index overhead
+    // is identical in both files, so the margin here comes entirely from
+    // smaller data blocks.
+    try std.testing.expect(sharedReader.physicalBytes * 5 < distinctReader.physicalBytes * 4);
+
+    // Correctness survives the smaller on-disk encoding.
+    try std.testing.expectEqualStrings("v", sharedReader.find("users\x00city\x00warsaw\x00pk00000000").?.value);
+    try std.testing.expectEqualStrings("v", sharedReader.find("users\x00city\x00warsaw\x00pk00000150").?.value);
+    try std.testing.expectEqualStrings("v", sharedReader.find("users\x00city\x00warsaw\x00pk00000299").?.value);
+    try std.testing.expectEqual(@as(?Record, null), sharedReader.find("users\x00city\x00warsaw\x00pk99999999"));
+}