feat(wal,memtable): implement segment writer, block writer, and MemTable
- wal/block_writer.go: 32KB block buffer with padding and flush - wal/segment_writer.go: WAL segment file with durable-ready protocol - memtable/memtable.go: Arena+SkipList wrapper with publish/abort semantics - Comprehensive tests for all modules, all pass with -race
This commit is contained in:
dailz
@@ -0,0 +1,339 @@
|
||||
package wal
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/dailz/go-kv/config"
|
||||
)
|
||||
|
||||
func testWalConfig() *config.WalConfig {
|
||||
cfg := config.Defaults()
|
||||
return &cfg
|
||||
}
|
||||
|
||||
func TestSegmentWriterCreation(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfg := testWalConfig()
|
||||
|
||||
sw, err := NewSegmentWriter(dir, 1, 100, cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSegmentWriter: %v", err)
|
||||
}
|
||||
|
||||
expectedPath := filepath.Join(dir, "segment-1.wal")
|
||||
if sw.SegmentPath() != expectedPath {
|
||||
t.Errorf("SegmentPath = %q, want %q", sw.SegmentPath(), expectedPath)
|
||||
}
|
||||
if sw.SegmentID() != 1 {
|
||||
t.Errorf("SegmentID = %d, want 1", sw.SegmentID())
|
||||
}
|
||||
if sw.CurrentOffset() != WalFileHeaderSize {
|
||||
t.Errorf("CurrentOffset = %d, want %d", sw.CurrentOffset(), WalFileHeaderSize)
|
||||
}
|
||||
|
||||
// Verify the file exists and has correct header.
|
||||
data, err := os.ReadFile(expectedPath)
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile: %v", err)
|
||||
}
|
||||
if len(data) != WalFileHeaderSize {
|
||||
t.Errorf("file size = %d, want %d (header only)", len(data), WalFileHeaderSize)
|
||||
}
|
||||
|
||||
hdr, err := DecodeWalHeader(data)
|
||||
if err != nil {
|
||||
t.Fatalf("DecodeWalHeader: %v", err)
|
||||
}
|
||||
if hdr.SegmentID != 1 {
|
||||
t.Errorf("header SegmentID = %d, want 1", hdr.SegmentID)
|
||||
}
|
||||
if hdr.StartSequence != 100 {
|
||||
t.Errorf("header StartSequence = %d, want 100", hdr.StartSequence)
|
||||
}
|
||||
if hdr.BlockSize != cfg.BlockSize {
|
||||
t.Errorf("header BlockSize = %d, want %d", hdr.BlockSize, cfg.BlockSize)
|
||||
}
|
||||
|
||||
// No .tmp file should remain.
|
||||
tmpPath := filepath.Join(dir, "segment-1.wal.tmp")
|
||||
if _, err := os.Stat(tmpPath); !os.IsNotExist(err) {
|
||||
t.Errorf("temp file %q should not exist", tmpPath)
|
||||
}
|
||||
|
||||
if err := sw.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSegmentWriterAppendBatch(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfg := testWalConfig()
|
||||
|
||||
sw, err := NewSegmentWriter(dir, 42, 0, cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSegmentWriter: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { sw.Close() })
|
||||
|
||||
// Encode a small batch.
|
||||
entries := []*WalEntry{
|
||||
{OpType: OpPut, ValueKind: VKInline, Key: []byte("key1"), Value: []byte("val1")},
|
||||
{OpType: OpPut, ValueKind: VKInline, Key: []byte("key2"), Value: []byte("val2")},
|
||||
}
|
||||
encoded, err := EncodeWalBatch(0, entries)
|
||||
if err != nil {
|
||||
t.Fatalf("EncodeWalBatch: %v", err)
|
||||
}
|
||||
|
||||
if err := sw.AppendBatch(encoded); err != nil {
|
||||
t.Fatalf("AppendBatch: %v", err)
|
||||
}
|
||||
|
||||
if err := sw.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
|
||||
// Read back the file and verify records.
|
||||
data, err := os.ReadFile(sw.SegmentPath())
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile: %v", err)
|
||||
}
|
||||
|
||||
// Skip file header.
|
||||
body := data[WalFileHeaderSize:]
|
||||
|
||||
// Use FragmentCollector to reassemble.
|
||||
fc := NewFragmentCollector()
|
||||
offset := 0
|
||||
for offset < len(body) {
|
||||
// Check for trailing zeros (block padding).
|
||||
if body[offset] == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
rec, consumed, err := DecodePhysicalRecord(body[offset:])
|
||||
if err != nil {
|
||||
t.Fatalf("DecodePhysicalRecord at offset %d: %v", offset, err)
|
||||
}
|
||||
|
||||
if err := fc.Append(rec.Type, rec.Payload); err != nil {
|
||||
t.Fatalf("FragmentCollector.Append: %v", err)
|
||||
}
|
||||
offset += consumed
|
||||
}
|
||||
|
||||
if !fc.IsComplete() {
|
||||
t.Fatal("fragment collector should be complete")
|
||||
}
|
||||
|
||||
decoded, err := DecodeWalBatch(fc.BatchData())
|
||||
if err != nil {
|
||||
t.Fatalf("DecodeWalBatch: %v", err)
|
||||
}
|
||||
|
||||
if decoded.EntryCount != 2 {
|
||||
t.Errorf("EntryCount = %d, want 2", decoded.EntryCount)
|
||||
}
|
||||
if decoded.BaseSequence != 0 {
|
||||
t.Errorf("BaseSequence = %d, want 0", decoded.BaseSequence)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSegmentWriterMultipleBatches(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfg := testWalConfig()
|
||||
|
||||
sw, err := NewSegmentWriter(dir, 1, 0, cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSegmentWriter: %v", err)
|
||||
}
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
entries := []*WalEntry{
|
||||
{
|
||||
OpType: OpPut,
|
||||
ValueKind: VKInline,
|
||||
Key: []byte("key"),
|
||||
Value: []byte("val"),
|
||||
},
|
||||
}
|
||||
encoded, err := EncodeWalBatch(uint64(i), entries)
|
||||
if err != nil {
|
||||
t.Fatalf("EncodeWalBatch %d: %v", i, err)
|
||||
}
|
||||
if err := sw.AppendBatch(encoded); err != nil {
|
||||
t.Fatalf("AppendBatch %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := sw.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
|
||||
data, err := os.ReadFile(sw.SegmentPath())
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile: %v", err)
|
||||
}
|
||||
|
||||
body := data[WalFileHeaderSize:]
|
||||
fc := NewFragmentCollector()
|
||||
batchCount := 0
|
||||
offset := 0
|
||||
|
||||
for offset < len(body) {
|
||||
if body[offset] == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
rec, consumed, err := DecodePhysicalRecord(body[offset:])
|
||||
if err != nil {
|
||||
t.Fatalf("DecodePhysicalRecord at offset %d: %v", offset, err)
|
||||
}
|
||||
offset += consumed
|
||||
|
||||
if err := fc.Append(rec.Type, rec.Payload); err != nil {
|
||||
t.Fatalf("FragmentCollector.Append: %v", err)
|
||||
}
|
||||
|
||||
if fc.IsComplete() {
|
||||
batchCount++
|
||||
decoded, err := DecodeWalBatch(fc.BatchData())
|
||||
if err != nil {
|
||||
t.Fatalf("DecodeWalBatch %d: %v", batchCount, err)
|
||||
}
|
||||
if decoded.BaseSequence != uint64(batchCount-1) {
|
||||
t.Errorf("batch %d BaseSequence = %d, want %d",
|
||||
batchCount, decoded.BaseSequence, batchCount-1)
|
||||
}
|
||||
fc.Reset()
|
||||
}
|
||||
}
|
||||
|
||||
if batchCount != 5 {
|
||||
t.Errorf("decoded %d batches, want 5", batchCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSegmentWriterRemainingPayload(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfg := testWalConfig()
|
||||
|
||||
sw, err := NewSegmentWriter(dir, 1, 0, cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSegmentWriter: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { sw.Close() })
|
||||
|
||||
initialRemaining := sw.RemainingPayload()
|
||||
|
||||
entries := []*WalEntry{
|
||||
{OpType: OpPut, ValueKind: VKInline, Key: []byte("k"), Value: []byte("v")},
|
||||
}
|
||||
encoded, err := EncodeWalBatch(0, entries)
|
||||
if err != nil {
|
||||
t.Fatalf("EncodeWalBatch: %v", err)
|
||||
}
|
||||
if err := sw.AppendBatch(encoded); err != nil {
|
||||
t.Fatalf("AppendBatch: %v", err)
|
||||
}
|
||||
|
||||
if sw.RemainingPayload() >= initialRemaining {
|
||||
t.Errorf("RemainingPayload should decrease after write, got %d >= %d",
|
||||
sw.RemainingPayload(), initialRemaining)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSegmentWriterLargeBatch(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfg := testWalConfig()
|
||||
|
||||
sw, err := NewSegmentWriter(dir, 1, 0, cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSegmentWriter: %v", err)
|
||||
}
|
||||
|
||||
// Create a batch larger than one block payload (~32KB - 7 bytes).
|
||||
largeValue := make([]byte, 40*1024)
|
||||
for i := range largeValue {
|
||||
largeValue[i] = byte(i % 256)
|
||||
}
|
||||
|
||||
entries := []*WalEntry{
|
||||
{OpType: OpPut, ValueKind: VKValueLogPointer, Key: []byte("large-key"), Value: largeValue},
|
||||
}
|
||||
encoded, err := EncodeWalBatch(0, entries)
|
||||
if err != nil {
|
||||
t.Fatalf("EncodeWalBatch: %v", err)
|
||||
}
|
||||
|
||||
if err := sw.AppendBatch(encoded); err != nil {
|
||||
t.Fatalf("AppendBatch: %v", err)
|
||||
}
|
||||
|
||||
if err := sw.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
|
||||
// Read back and verify.
|
||||
data, err := os.ReadFile(sw.SegmentPath())
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFile: %v", err)
|
||||
}
|
||||
|
||||
body := data[WalFileHeaderSize:]
|
||||
fc := NewFragmentCollector()
|
||||
offset := 0
|
||||
fragCount := 0
|
||||
|
||||
for offset < len(body) {
|
||||
if body[offset] == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
rec, consumed, err := DecodePhysicalRecord(body[offset:])
|
||||
if err != nil {
|
||||
t.Fatalf("DecodePhysicalRecord at offset %d: %v", offset, err)
|
||||
}
|
||||
offset += consumed
|
||||
fragCount++
|
||||
|
||||
if err := fc.Append(rec.Type, rec.Payload); err != nil {
|
||||
t.Fatalf("FragmentCollector.Append type=%d: %v", rec.Type, err)
|
||||
}
|
||||
}
|
||||
|
||||
if !fc.IsComplete() {
|
||||
t.Fatal("fragment collector should be complete after large batch")
|
||||
}
|
||||
if fragCount < 2 {
|
||||
t.Errorf("expected multiple fragments for large batch, got %d", fragCount)
|
||||
}
|
||||
|
||||
decoded, err := DecodeWalBatch(fc.BatchData())
|
||||
if err != nil {
|
||||
t.Fatalf("DecodeWalBatch: %v", err)
|
||||
}
|
||||
if decoded.EntryCount != 1 {
|
||||
t.Errorf("EntryCount = %d, want 1", decoded.EntryCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSegmentWriterSync(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfg := testWalConfig()
|
||||
|
||||
sw, err := NewSegmentWriter(dir, 1, 0, cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSegmentWriter: %v", err)
|
||||
}
|
||||
|
||||
if err := sw.Sync(); err != nil {
|
||||
t.Fatalf("Sync: %v", err)
|
||||
}
|
||||
|
||||
if err := sw.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user