Documentation
¶
Overview ¶
Package wal implements a Write-Ahead Log (WAL) for durable and atomic record storage.
A Write-Ahead Log is a reliability mechanism that ensures data persistence by writing records sequentially to disk before they are considered committed. The WAL is organized into segments, where each segment contains a sorted collection of records.
The WAL implementation provides the following guarantees:
- Durability: Records are persisted to disk before operations are considered complete
- Atomicity: Records within a segment are written atomically
- Order preservation: Records are maintained in sorted order within segments
Basic usage:
file, err := os.Create("mywal.db")
if err != nil {
log.Fatal(err)
}
// Create a new WAL writer with segments of up to 1000 records
writer, err := wal.NewWriter(file, 1000)
if err != nil {
log.Fatal(err)
}
// Write records
err = writer.Write(myRecord)
if err != nil {
log.Fatal(err)
}
// Close the writer
err = writer.Close()
if err != nil {
log.Fatal(err)
}
// Read records back
file, err = os.Open("mywal.db")
if err != nil {
log.Fatal(err)
}
reader := wal.NewReader(file)
for record := range reader.All() {
// Process record
}
File Format: Each segment in the WAL consists of:
- Segment header: Total size of the segment (8 bytes)
- Records: Series of variable-length records
The WAL automatically rotates segments when they reach the configured maximum number of records, ensuring efficient storage and retrieval.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
View Source
var ( ErrInvalidMaxRecords = errors.New("maxRecords must be greater than 0") ErrWALClosed = errors.New("WAL is closed") )
Functions ¶
This section is empty.
Types ¶
type Writer ¶
type Writer struct {
// contains filtered or unexported fields
}
Example ¶
ExampleWriter demonstrates creating a WAL, writing records, and reading them back.
package main
import (
"bytes"
"fmt"
"time"
"github.com/davidvella/xp/partition"
"github.com/davidvella/xp/wal"
)
type writeCloser struct {
*bytes.Buffer
}
func (w writeCloser) Close() error {
return nil
}
func main() {
buffer := bytes.NewBuffer([]byte{})
buf := writeCloser{buffer}
// Initialize the WAL writer with segments of up to 1000 records
writer, err := wal.NewWriter(buf, 1000)
if err != nil {
fmt.Printf("Error opening writer: %v\n", err)
return
}
// Create and write some records
records := []partition.Record{
&partition.RecordImpl{
ID: "key1",
PartitionKey: "partition1",
Timestamp: time.Now(),
Data: []byte("value1"),
},
&partition.RecordImpl{
ID: "key2",
PartitionKey: "partition1",
Timestamp: time.Now(),
Data: []byte("value2"),
},
}
for _, record := range records {
if err := writer.Write(record); err != nil {
fmt.Printf("Error writing record: %v\n", err)
return
}
}
// Close the writer to flush any remaining records
if err := writer.Close(); err != nil {
fmt.Printf("Error closing writer: %v\n", err)
return
}
// Initialize the WAL reader
readerAt := bytes.NewReader(buf.Bytes())
reader := wal.NewReader(readerAt)
// Read all records back
for record := range reader.All() {
fmt.Printf("Read record: %s = %s\n", record.GetID(), string(record.GetData()))
}
}
Output: Read record: key1 = value1 Read record: key2 = value2
Example (Segments) ¶
ExampleWriter_segments demonstrates how the WAL handles segment rotation.
package main
import (
"bytes"
"fmt"
"time"
"github.com/davidvella/xp/partition"
"github.com/davidvella/xp/wal"
)
type writeCloser struct {
*bytes.Buffer
}
func (w writeCloser) Close() error {
return nil
}
func main() {
buffer := bytes.NewBuffer([]byte{})
buf := writeCloser{buffer}
// Initialize the WAL writer with small segments (2 records per segment)
writer, err := wal.NewWriter(&buf, 2)
if err != nil {
fmt.Printf("Error opening writer: %v\n", err)
return
}
// Write three records to trigger segment rotation
records := []partition.Record{
&partition.RecordImpl{
ID: "1",
PartitionKey: "partition1",
Timestamp: time.Now(),
Data: []byte("first"),
},
&partition.RecordImpl{
ID: "2",
PartitionKey: "partition1",
Timestamp: time.Now(),
Data: []byte("second"),
},
&partition.RecordImpl{
ID: "3",
PartitionKey: "partition1",
Timestamp: time.Now(),
Data: []byte("third"),
},
}
for _, record := range records {
if err := writer.Write(record); err != nil {
fmt.Printf("Error writing record: %v\n", err)
return
}
}
if err := writer.Close(); err != nil {
fmt.Printf("Error closing writer: %v\n", err)
return
}
// Read all records back
readerAt := bytes.NewReader(buf.Bytes())
reader := wal.NewReader(readerAt)
for record := range reader.All() {
fmt.Printf("Read record: %s = %s\n", record.GetID(), string(record.GetData()))
}
}
Output: Read record: 1 = first Read record: 2 = second Read record: 3 = third
Source Files
Click to show internal directories.
Click to hide internal directories.