diff --git a/src/sstable.zig b/src/sstable.zig index 45f6c0d..f523bfc 100644 --- a/src/sstable.zig +++ b/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")); +}