Files
Abstraction/vendor/github.com/akrylysov/pogreb/segment.go
2022-12-11 06:18:00 +00:00

157 lines
3.7 KiB
Go

package pogreb
import (
"bufio"
"encoding/binary"
"fmt"
"hash/crc32"
"io"
)
type recordType int
const (
recordTypePut recordType = iota
recordTypeDelete
segmentExt = ".psg"
)
// segment is a write-ahead log segment.
// It consists of a sequence of binary-encoded variable length records.
type segment struct {
*file
id uint16 // Physical segment identifier.
sequenceID uint64 // Logical monotonically increasing segment identifier.
name string
meta *segmentMeta
}
func segmentName(id uint16, sequenceID uint64) string {
return fmt.Sprintf("%05d-%d%s", id, sequenceID, segmentExt)
}
type segmentMeta struct {
Full bool
PutRecords uint32
DeleteRecords uint32
DeletedKeys uint32
DeletedBytes uint32
}
func segmentMetaName(id uint16, sequenceID uint64) string {
return segmentName(id, sequenceID) + metaExt
}
// Binary representation of a segment record:
// +---------------+------------------+------------------+-...-+--...--+----------+
// | Key Size (2B) | Record Type (1b) | Value Size (31b) | Key | Value | CRC (4B) |
// +---------------+------------------+------------------+-...-+--...--+----------+
type record struct {
rtype recordType
segmentID uint16
offset uint32
data []byte
key []byte
value []byte
}
func encodedRecordSize(kvSize uint32) uint32 {
// key size, value size, key, value, crc32
return 2 + 4 + kvSize + 4
}
func encodeRecord(key []byte, value []byte, rt recordType) []byte {
size := encodedRecordSize(uint32(len(key) + len(value)))
data := make([]byte, size)
binary.LittleEndian.PutUint16(data[:2], uint16(len(key)))
valLen := uint32(len(value))
if rt == recordTypeDelete { // Set delete bit.
valLen |= 1 << 31
}
binary.LittleEndian.PutUint32(data[2:], valLen)
copy(data[6:], key)
copy(data[6+len(key):], value)
checksum := crc32.ChecksumIEEE(data[:6+len(key)+len(value)])
binary.LittleEndian.PutUint32(data[size-4:size], checksum)
return data
}
func encodePutRecord(key []byte, value []byte) []byte {
return encodeRecord(key, value, recordTypePut)
}
func encodeDeleteRecord(key []byte) []byte {
return encodeRecord(key, nil, recordTypeDelete)
}
// segmentIterator iterates over segment records.
type segmentIterator struct {
f *segment
offset uint32
r *bufio.Reader
buf []byte // kv size and crc32 reusable buffer.
}
func newSegmentIterator(f *segment) (*segmentIterator, error) {
if _, err := f.Seek(int64(headerSize), io.SeekStart); err != nil {
return nil, err
}
return &segmentIterator{
f: f,
offset: headerSize,
r: bufio.NewReader(f),
buf: make([]byte, 6),
}, nil
}
func (it *segmentIterator) next() (record, error) {
// Read key and value size.
kvSizeBuf := it.buf
if _, err := io.ReadFull(it.r, kvSizeBuf); err != nil {
if err == io.EOF {
return record{}, ErrIterationDone
}
return record{}, err
}
// Decode key size.
keySize := uint32(binary.LittleEndian.Uint16(kvSizeBuf[:2]))
// Decode value size and record type.
rt := recordTypePut
valueSize := binary.LittleEndian.Uint32(kvSizeBuf[2:])
if valueSize&(1<<31) != 0 {
rt = recordTypeDelete
valueSize &^= 1 << 31
}
// Read key, value and checksum.
recordSize := encodedRecordSize(keySize + valueSize)
data := make([]byte, recordSize)
copy(data, kvSizeBuf)
if _, err := io.ReadFull(it.r, data[6:]); err != nil {
return record{}, err
}
// Verify checksum.
checksum := binary.LittleEndian.Uint32(data[len(data)-4:])
if checksum != crc32.ChecksumIEEE(data[:len(data)-4]) {
return record{}, errCorrupted
}
offset := it.offset
it.offset += recordSize
rec := record{
rtype: rt,
segmentID: it.f.id,
offset: offset,
data: data,
key: data[6 : 6+keySize],
value: data[6+keySize : 6+keySize+valueSize],
}
return rec, nil
}