dima/qb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
qb/storage/storage_test.go

1155 lines
27 KiB
Go
Raw Permalink Normal View History

wp
2026-06-13 08:01:42 +03:00
package storage
import (
"bytes"
"fmt"
wp
2026-06-14 23:12:03 +03:00
"io"
wp
2026-06-13 08:01:42 +03:00
"math"
"reflect"
"slices"
"testing"
bin "gordenko.dev/dima/bin/little"
"gordenko.dev/dima/qb"
"gordenko.dev/dima/qb/util"
)
func TestAppendIndexRecord(t *testing.T) {
IndexPageSize = 24
var (
after = []byte{
0x01, 0x00, 0x00, 0x00, 0x08, 0x00, 0x00, 0x00,
0x02, 0x00, 0x00, 0x00, 0x09, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
}
)
// func creates index page buffer
buf := appendIndexRecord(appendIndexRecordIn{
Timestamp: 1,
PageNo: 8,
})
buf = appendIndexRecord(appendIndexRecordIn{
Records: buf,
RecordsCount: 1,
Timestamp: 2,
PageNo: 9,
})
if !bytes.Equal(buf, after) {
t.Fatalf("before not equal after:\nbefore: % x\n after: % x",
buf, after)
}
}
func TestSealIndexPage(t *testing.T) {
IndexPageSize = 16
indexCRC32Idx = IndexPageSize - 4
indexRecordsCountIdx = IndexPageSize - 6
isZeroLevelIdx = IndexPageSize - 7
var (
before = []byte{
0x07, 0x07, 0x07, 0x07, 0x08, 0x08, 0x08, 0x08, 0x00, // payload
0x00, // zero level
0x00, 0x00, // records count
0x00, 0x00, 0x00, 0x00, // crc32
}
after = []byte{
0x07, 0x07, 0x07, 0x07, 0x08, 0x08, 0x08, 0x08, 0x00, // payload
0x01, // zero level
0x02, 0x00, // records count
0x8d, 0xcc, 0x3c, 0xc0, // crc32
}
)
calculatedCRC := util.CalculateCRC32(after[:indexCRC32Idx])
checksum := SealIndexPage(SealIndexPageIn{
Content: before,
RecordsCount: 2,
ZeroLevel: true,
})
if calculatedCRC != checksum {
t.Fatalf("calculated CRC %d not equal returned %d",
calculatedCRC, checksum)
}
if !bytes.Equal(before, after) {
t.Fatalf("before not equal after:\nbefore: % x\n after: % x",
before, after)
}
}
func TestSealDataPage(t *testing.T) {
DataPageSize = 16
dataCRC32Idx = DataPageSize - 4
timestampsSizeIdx = DataPageSize - 6
valuesSizeIdx = DataPageSize - 8
prevPageIdx = DataPageSize - 12
var (
before = []byte{
0x07, 0x07, 0x07, 0x07, // payload
0x00, 0x00, 0x00, 0x00, // prevPageNo
0x00, 0x00, // values size
0x00, 0x00, // timestamps size
0x00, 0x00, 0x00, 0x00, // crc32
}
after = []byte{
0x07, 0x07, 0x07, 0x07, // payload
0x09, 0x00, 0x00, 0x00, // prevPageNo
0x03, 0x00, // values size
0x01, 0x00, // timestamps size
0xa9, 0xec, 0xc8, 0x91, // crc32
}
)
calculatedCRC := util.CalculateCRC32(after[:dataCRC32Idx])
checksum := SealDataPage(SealDataPageIn{
Content: before,
PrevPageNo: 9,
TimestampsSize: 1,
ValuesSize: 3,
})
if calculatedCRC != checksum {
t.Fatalf("calculated CRC %d not equal returned %d",
calculatedCRC, checksum)
}
if !bytes.Equal(before, after) {
t.Fatalf("before not equal after:\nbefore: % x\n after: % x",
before, after)
}
}
func TestPageIngester(t *testing.T) {
// index
IndexPageSize = 24
indexCRC32Idx = IndexPageSize - 4
indexRecordsCountIdx = IndexPageSize - 6
isZeroLevelIdx = IndexPageSize - 7
maxRecordsOnIndexPage = 2
// data
DataPageSize = 16
DataPagePayloadSize = DataPageSize - DataPageFooterSize
dataCRC32Idx = DataPageSize - 4
timestampsSizeIdx = DataPageSize - 6
valuesSizeIdx = DataPageSize - 8
prevPageIdx = DataPageSize - 12
var testCases = []struct {
Name string
DataPages []DataPayload
ExpectedIndexLevels []ExpectedIndexLevel
ExpectedDataPages []ExpectedDataPage
}{
{
Name: "create root index tail",
DataPages: []DataPayload{
{
Since: 100,
Content: makeDataPage([]byte{0x01, 0x00, 0x00, 0x00}),
},
},
ExpectedDataPages: []ExpectedDataPage{
{
PageNo: 1,
PrevPageNo: 0,
Payload: []byte{0x01, 0x00, 0x00, 0x00},
},
},
ExpectedIndexLevels: []ExpectedIndexLevel{
{
TailRecords: []IndexRecord{
{Timestamp: 100, PageNo: 1},
},
TailRecordsCount: 1,
},
},
},
{
Name: "just append record to index tail",
DataPages: []DataPayload{
{
Since: 100,
Content: makeDataPage([]byte{0x01, 0x00, 0x00, 0x00}),
},
{
Since: 200,
Content: makeDataPage([]byte{0x02, 0x00, 0x00, 0x00}),
},
},
ExpectedDataPages: []ExpectedDataPage{
{
PageNo: 1,
PrevPageNo: 0,
Payload: []byte{0x01, 0x00, 0x00, 0x00},
},
{
PageNo: 2,
PrevPageNo: 1,
Payload: []byte{0x02, 0x00, 0x00, 0x00},
},
},
ExpectedIndexLevels: []ExpectedIndexLevel{
{
TailRecords: []IndexRecord{
{Timestamp: 100, PageNo: 1},
{Timestamp: 200, PageNo: 2},
},
TailRecordsCount: 2,
},
},
},
{
Name: "create index page and upper level (root)",
DataPages: []DataPayload{
{
Since: 100,
Content: makeDataPage([]byte{0x01, 0x00, 0x00, 0x00}),
},
{
Since: 200,
Content: makeDataPage([]byte{0x02, 0x00, 0x00, 0x00}),
},
{
Since: 300,
Content: makeDataPage([]byte{0x03, 0x00, 0x00, 0x00}),
},
},
ExpectedDataPages: []ExpectedDataPage{
{
PageNo: 1,
PrevPageNo: 0,
Payload: []byte{0x01, 0x00, 0x00, 0x00},
},
{
PageNo: 2,
PrevPageNo: 1,
Payload: []byte{0x02, 0x00, 0x00, 0x00},
},
{
PageNo: 3,
PrevPageNo: 2,
Payload: []byte{0x03, 0x00, 0x00, 0x00},
},
},
ExpectedIndexLevels: []ExpectedIndexLevel{
{
IndexPages: []ExpectedIndexPage{
{
PageNo: 1,
ZeroLevel: true,
Records: []IndexRecord{
{Timestamp: 100, PageNo: 1},
{Timestamp: 200, PageNo: 2},
},
},
},
TailRecords: []IndexRecord{
{Timestamp: 300, PageNo: 3},
},
TailRecordsCount: 1,
},
{
TailRecords: []IndexRecord{
{Timestamp: 100, PageNo: 1},
},
TailRecordsCount: 1,
},
},
},
{
Name: "just append record to tail on zero level",
DataPages: []DataPayload{
{
Since: 100,
Content: makeDataPage([]byte{0x01, 0x00, 0x00, 0x00}),
},
{
Since: 200,
Content: makeDataPage([]byte{0x02, 0x00, 0x00, 0x00}),
},
{
Since: 300,
Content: makeDataPage([]byte{0x03, 0x00, 0x00, 0x00}),
},
{
Since: 400,
Content: makeDataPage([]byte{0x04, 0x00, 0x00, 0x00}),
},
},
ExpectedDataPages: []ExpectedDataPage{
{
PageNo: 1,
PrevPageNo: 0,
Payload: []byte{0x01, 0x00, 0x00, 0x00},
},
{
PageNo: 2,
PrevPageNo: 1,
Payload: []byte{0x02, 0x00, 0x00, 0x00},
},
{
PageNo: 3,
PrevPageNo: 2,
Payload: []byte{0x03, 0x00, 0x00, 0x00},
},
{
PageNo: 4,
PrevPageNo: 3,
Payload: []byte{0x04, 0x00, 0x00, 0x00},
},
},
ExpectedIndexLevels: []ExpectedIndexLevel{
{
IndexPages: []ExpectedIndexPage{
{
PageNo: 1,
ZeroLevel: true,
Records: []IndexRecord{
{Timestamp: 100, PageNo: 1},
{Timestamp: 200, PageNo: 2},
},
},
},
TailRecords: []IndexRecord{
{Timestamp: 300, PageNo: 3},
{Timestamp: 400, PageNo: 4},
},
TailRecordsCount: 2,
},
{
TailRecords: []IndexRecord{
{Timestamp: 100, PageNo: 1},
},
TailRecordsCount: 1,
},
},
},
}
for _, testCase := range testCases {
pageManager := new(PageNumbersMock)
ingester, err := NewPageIngester(PageIngesterOptions{
GetIndexPageNumber: pageManager.GetIndexPageNumber,
GetDataPageNumber: pageManager.GetDataPageNumber,
})
if err != nil {
t.Fatal(err)
}
sealedDataPages, sealedIndexLevels := ingester.Ingest(IngestIn{
LastPageNo: 0,
IndexLevelTails: nil,
DataPages: testCase.DataPages,
})
if len(testCase.ExpectedDataPages) != len(sealedDataPages) {
t.Fatalf("%s: got %d data pages, but not %d",
testCase.Name, len(sealedDataPages), len(testCase.ExpectedDataPages))
}
for idx, page := range testCase.ExpectedDataPages {
err := cmpDataPage(page, sealedDataPages[idx])
if err != nil {
t.Fatalf("%s: data page #%d: %s", testCase.Name, idx, err)
}
}
if len(testCase.ExpectedIndexLevels) != len(sealedIndexLevels) {
t.Fatalf("%s: got %d index levels, but not %d",
testCase.Name, len(sealedIndexLevels), len(testCase.ExpectedIndexLevels))
}
for idx, level := range testCase.ExpectedIndexLevels {
err := cmpIndexLevel(level, *sealedIndexLevels[idx])
if err != nil {
t.Fatalf("%s: index level #%d: %s", testCase.Name, idx, err)
}
}
}
}
type PageNumbersMock struct {
indexPageNo uint32
dataPageNo uint32
}
func (s *PageNumbersMock) GetIndexPageNumber() (uint32, bool, error) {
s.indexPageNo++
return s.indexPageNo, false, nil
}
func (s *PageNumbersMock) GetDataPageNumber() (uint32, bool, error) {
s.dataPageNo++
return s.dataPageNo, false, nil
}
// func getIndexRecords(buf []byte, count int) (list []IndexRecord) {
// i := 0
// for range count {
// var rec IndexRecord
// rec.Timestamp, _ = bin.GetUint32(buf[i:])
// rec.PageNo, _ = bin.GetUint32(buf[i+4:])
// list = append(list, rec)
// i += indexRecordSize
// }
// return
// }
type ExpectedIndexPage struct {
PageNo uint32
Records []IndexRecord
ZeroLevel bool
}
func cmpIndexPage(expected ExpectedIndexPage, page SealedIndexPage) error {
if expected.PageNo != page.PageNo {
return fmt.Errorf("got pageNo %d not equal expected %d",
page.PageNo, expected.PageNo)
}
buf := page.Content
calculatedCRC := util.CalculateCRC32(buf[:indexCRC32Idx])
checksum, _ := bin.GetUint32(buf[indexCRC32Idx:])
if calculatedCRC != checksum {
return fmt.Errorf("calculated CRC %d not equal written %d",
calculatedCRC, checksum)
}
zeroLevel, _ := bin.GetBool(buf[isZeroLevelIdx:])
if expected.ZeroLevel != zeroLevel {
return fmt.Errorf("expected zero level %t not equal written %t",
expected.ZeroLevel, zeroLevel)
}
count, _ := bin.GetUint16(buf[indexRecordsCountIdx:])
records := getIndexRecords(buf, int(count))
if !slices.Equal(expected.Records, records) {
return fmt.Errorf("expected records %v not equal written %v",
expected.Records, records)
}
return nil
}
type ExpectedIndexLevel struct {
SkipRecords int
IndexPages []ExpectedIndexPage
TailRecords []IndexRecord
TailRecordsCount int
}
func cmpIndexLevel(expected ExpectedIndexLevel, level SealedIndexLevel) error {
if expected.SkipRecords != level.SkipRecords {
return fmt.Errorf("got skipRecords %d not equal expected %d",
level.SkipRecords, expected.SkipRecords)
}
if expected.TailRecordsCount != level.TailRecordsCount {
return fmt.Errorf("got tailRecordsCount %d not equal expected %d",
level.TailRecordsCount, expected.TailRecordsCount)
}
records := getIndexRecords(level.TailRecords, level.TailRecordsCount)
if !slices.Equal(expected.TailRecords, records) {
return fmt.Errorf("expected tailRecords %v not equal written %v",
expected.TailRecords, records)
}
if len(expected.IndexPages) != len(level.IndexPages) {
return fmt.Errorf("got %d index pages, but not %d",
len(level.IndexPages), len(expected.IndexPages))
}
for idx, page := range expected.IndexPages {
err := cmpIndexPage(page, level.IndexPages[idx])
if err != nil {
return fmt.Errorf("index page #%d: %s", idx, err)
}
}
return nil
}
type ExpectedDataPage struct {
PageNo uint32
Payload []byte
PrevPageNo uint32
}
func cmpDataPage(expected ExpectedDataPage, page SealedDataPage) error {
buf := page.Content
if expected.PageNo != page.PageNo {
return fmt.Errorf("got pageNo %d not equal expected %d",
page.PageNo, expected.PageNo)
}
prevPageNo, _ := bin.GetUint32(buf[prevPageIdx:])
if expected.PrevPageNo != prevPageNo {
return fmt.Errorf("written prevPageNo %d not equal expected %d",
prevPageNo, expected.PrevPageNo)
}
calculatedCRC := util.CalculateCRC32(buf[:dataCRC32Idx])
checksum, _ := bin.GetUint32(buf[dataCRC32Idx:])
if calculatedCRC != checksum {
return fmt.Errorf("calculated CRC %d not equal written %d",
calculatedCRC, checksum)
}
writtenPayload := page.Content[:DataPagePayloadSize]
if !slices.Equal(expected.Payload, writtenPayload) {
return fmt.Errorf("expected payload % x not equal written % x",
expected.Payload, writtenPayload)
}
return nil
}
func makeDataPage(payload []byte) []byte {
buf := make([]byte, DataPageSize)
copy(buf, payload)
return buf
}
func TestMetricAddRecord(t *testing.T) {
rec := MetricAddRecord{
MetricID: 12345,
MetricType: qb.Cumulative,
FracDigits: 5,
}
buf := bytes.NewBuffer(nil)
rec.Pack(buf)
recordType, _ := buf.ReadByte()
if recordType != CodeMetricAdd {
t.Fatalf("wrong record type: %d", recordType)
}
decoded := new(MetricAddRecord)
err := decoded.Parse(buf)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(rec, *decoded) {
t.Fatalf("decoded are not equal origin: %v", decoded)
}
}
func TestMetricDeleteRecord(t *testing.T) {
rec := MetricDeleteRecord{
MetricID: 12345,
FreeIndexPages: []uint32{1, math.MaxUint32},
FreeDataPages: []uint32{1, math.MaxUint32},
}
buf := bytes.NewBuffer(nil)
rec.Pack(buf)
recordType, _ := buf.ReadByte()
if recordType != CodeMetricDelete {
t.Fatalf("wrong record type: %d", recordType)
}
decoded := new(MetricDeleteRecord)
err := decoded.Parse(buf)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(rec, *decoded) {
t.Fatalf("decoded are not equal origin: %v", decoded)
}
}
func TestMeasuresDeleteRecord(t *testing.T) {
rec := MeasuresDeleteRecord{
MetricID: 12345,
FreeIndexPages: []uint32{1, math.MaxUint32},
FreeDataPages: []uint32{1, math.MaxUint32},
}
buf := bytes.NewBuffer(nil)
rec.Pack(buf)
recordType, _ := buf.ReadByte()
if recordType != CodeMeasuresDelete {
t.Fatalf("wrong record type: %d", recordType)
}
decoded := new(MeasuresDeleteRecord)
err := decoded.Parse(buf)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(rec, *decoded) {
t.Fatalf("decoded are not equal origin: %v", decoded)
}
}
func TestMeasuresAppendRecord(t *testing.T) {
rec := MeasuresAppendRecord{
MetricID: 12345,
TimestampsRewindOffset: 7,
Timestamps: []byte{
0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07,
},
ValuesRewindOffset: 11,
Values: []byte{
0x08, 0x09, 0x0a,
},
}
buf := bytes.NewBuffer(nil)
rec.Pack(buf)
recordType, _ := buf.ReadByte()
if recordType != CodeMeasuresAppend {
t.Fatalf("wrong record type: %d", recordType)
}
decoded := new(MeasuresAppendRecord)
err := decoded.Parse(buf)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(rec, *decoded) {
t.Fatalf("decoded are not equal origin: %v", decoded)
}
}
func TestMeasuresAppendWithGrowRecord(t *testing.T) {
DataPageSize = 8
IndexPageSize = 4
//
rec := MeasuresAppendWithGrowRecord{
MetricID: 12345,
DataPageTail: DataPageTail{
PageNo: math.MaxUint32,
Reused: true,
PrevPageNo: 4127831,
CRC32: 316278321,
TimestampsRewindOffset: 7,
Timestamps: []byte{
0x07, 0x07, 0x07, 0x07, 0x07,
},
Values: []byte{
0x08, 0x08, 0x08, 0x08,
},
ValuesRewindOffset: 11,
},
DataPages: []SealedDataPage{
{
PageNo: 1000,
Reused: true,
//PrevPageNo: math.MaxUint32,
//CRC32: 312,
Content: []byte{
0x05, 0x05, 0x05, 0x05, 0x05, 0x05, 0x05, 0x05,
},
},
{
PageNo: 1001,
Reused: false,
//PrevPageNo: 1000,
//CRC32: 432432,
Content: []byte{
0x06, 0x06, 0x06, 0x06, 0x06, 0x06, 0x06, 0x06,
},
},
},
TailTimestamps: []byte{
0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07,
},
TailValues: []byte{
0x08, 0x09, 0x0a,
},
ChangedIndexLevels: []ChangedIndexLevel{
{
IndexPageTail: &IndexPageTail{
PageNo: 20000,
Reused: true,
CRC32: 32897,
Records: []byte{
0x01, 0x00, 0x00, 0x00,
},
},
IndexPages: []SealedIndexPage{
{
PageNo: 20001,
Reused: true,
//CRC32: 32897,
Content: []byte{
0x01, 0x00, 0x00, 0x01,
},
},
{
PageNo: 20002,
Reused: false,
//CRC32: 3284397,
Content: []byte{
0x01, 0x02, 0x03, 0x04,
},
},
},
TailRecords: []byte{
0xaa, 0xbb,
},
},
{
IndexPageTail: nil,
TailRecords: []byte{
0xdd, 0xee, 0xff, 0x77,
},
},
},
}
buf := bytes.NewBuffer(nil)
rec.Pack(buf)
recordType, _ := buf.ReadByte()
if recordType != CodeMeasuresAppendWithGrow {
t.Fatalf("wrong record type: %d", recordType)
}
decoded := new(MeasuresAppendWithGrowRecord)
err := decoded.Parse(buf)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(rec, *decoded) {
t.Fatalf("decoded are not equal origin: %v", decoded)
}
}
func TestWritePreparer(t *testing.T) {
// data page
DataPageSize = 24
dataCRC32Idx = DataPageSize - 4
timestampsSizeIdx = DataPageSize - 6
valuesSizeIdx = DataPageSize - 8
prevPageIdx = DataPageSize - 12
// index page
wp
2026-06-13 07:13:03 +00:00
IndexPageSize = 24
wp
2026-06-13 08:01:42 +03:00
maxRecordsOnIndexPage = 2
indexCRC32Idx = IndexPageSize - 4
indexRecordsCountIdx = IndexPageSize - 6
isZeroLevelIdx = IndexPageSize - 7
pageManager := new(PageNumbersMock)
pageIngester, err := NewPageIngester(PageIngesterOptions{
GetIndexPageNumber: pageManager.GetIndexPageNumber,
GetDataPageNumber: pageManager.GetDataPageNumber,
})
if err != nil {
t.Fatal(err)
}
writePreparer := NewWritePreparer(WritePreparerOptions{
MinWALBufferSize: 128,
PageIngester: pageIngester,
})
rec := MeasuresAppendWithGrow{
MetricID: 12345,
LastPageNo: 1,
TimestampsRewindOffset: 2,
Timestamps: []byte{
0x04, 0x05, 0x06,
},
ValuesRewindOffset: 1,
Values: []byte{
0x04, 0x03,
},
DataPages: []DataPayload{
// 1. fills existent index tail
{
Since: 513, // 0x01 0x02
// full page
Content: []byte{
0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01,
0x00, 0x00, 0x00, 0x00, // prevPageNo
0x00, 0x00, // values size
0x00, 0x00, // timestamps size
0x00, 0x00, 0x00, 0x00, // crc32
},
TimestampsSize: 6,
ValuesSize: 4,
},
// 2. creates index page tail (zero level) and level 1
{
Since: 769, // 0x01 0x03
// full page
Content: []byte{
0xa1, 0xa2, 0xa3, 0xa4, 0xa5, 0x00, 0x00, 0xa5, 0xa4, 0xa3, 0xa2, 0xa1,
0x00, 0x00, 0x00, 0x00, // prevPageNo
0x00, 0x00, // values size
0x00, 0x00, // timestamps size
0x00, 0x00, 0x00, 0x00, // crc32
},
TimestampsSize: 5,
ValuesSize: 5,
},
// 2. fills index tail records (zero level)
{
Since: 1025, // 0x01 0x04
// full page
Content: []byte{
0xb1, 0xb2, 0xb3, 0xb4, 0xb5, 0xb6, 0xb7, 0xb5, 0xb4, 0xb3, 0xb2, 0xb1,
0x00, 0x00, 0x00, 0x00, // prevPageNo
0x00, 0x00, // values size
0x00, 0x00, // timestamps size
0x00, 0x00, 0x00, 0x00, // crc32
},
TimestampsSize: 7,
ValuesSize: 5,
},
// 3. creates index page
{
Since: 1025, // 0x01 0x04
// full page
Content: []byte{
0xc1, 0xc2, 0xc3, 0xc4, 0xc5, 0xc6, 0xc6, 0xc5, 0xc4, 0xc3, 0xc2, 0xc1,
0x00, 0x00, 0x00, 0x00, // prevPageNo
0x00, 0x00, // values size
0x00, 0x00, // timestamps size
0x00, 0x00, 0x00, 0x00, // crc32
},
TimestampsSize: 6,
ValuesSize: 6,
},
},
TailTimestamps: []byte{
0xc1, 0xc2,
},
TailValues: []byte{
0xd1, 0xd2,
},
IndexLevelTails: []IndexLevelTail{
{
Buffer: []byte{
wp
2026-06-13 07:13:03 +00:00
0x01, 0x01, 0x00, 0x00, // since
0x0a, 0x00, 0x00, 0x00, // pageNo 10
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // empty space
wp
2026-06-13 08:01:42 +03:00
0x00, // zero level
0x00, 0x00, // records count
0x00, 0x00, 0x00, 0x00, // crc32
},
RecordsCount: 1,
},
},
WrittenCount: 10,
}
prepared := writePreparer.Prepare([]any{rec})
// fmt.Printf("%v\n", prepared)
// pretty.PPrintln("prepared", prepared)
fmt.Printf("wal: % x\n", prepared.Packet)
fmt.Println("index to write")
for _, p := range prepared.WriteToIndex {
fmt.Printf("%d: % x\n", p.PageNo, p.Content)
}
fmt.Println("data to write")
for _, p := range prepared.WriteToData {
fmt.Printf("%d: % x\n", p.PageNo, p.Content)
}
fmt.Println("commits")
for _, x := range prepared.Commits {
fmt.Printf("%#v\n", x)
}
}
wp
2026-06-13 22:43:17 +00:00
wp
2026-06-14 07:57:01 +03:00
func TestWriteTimestampsWithRewind(t *testing.T) {
DataPagePayloadSize = 8
var (
payload = []byte{
0xaa, 0xbb, 0xcc,
wp
2026-06-13 22:43:17 +00:00
}
wp
2026-06-14 07:57:01 +03:00
before = []byte{
0x00, 0x00, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01,
}
occupied = 4
after = []byte{
0x00, 0x00, 0x00, 0xaa, 0xbb, 0xcc, 0x02, 0x01,
wp
2026-06-13 22:43:17 +00:00
}
wp
2026-06-14 07:57:01 +03:00
occupiedAfter = 5
)
occupied = writeTimestampsWithRewind(before, occupied, payload, 2)
if occupied != occupiedAfter {
t.Fatalf("got occupied %d are not equal expected %d", occupied, occupiedAfter)
}
if !bytes.Equal(before, after) {
t.Fatalf("got buf % x are not equal expected % x", before, after)
wp
2026-06-13 22:43:17 +00:00
}
}
wp
2026-06-14 07:57:01 +03:00
func TestWriteValuesWithRewind(t *testing.T) {
var (
payload = []byte{
0xaa, 0xbb, 0xcc,
}
before = []byte{
0x01, 0x02, 0x03, 0x04, 0x00, 0x00, 0x00, 0x00,
}
occupied = 4
after = []byte{
0x01, 0x02, 0xaa, 0xbb, 0xcc, 0x00, 0x00, 0x00,
}
occupiedAfter = 5
)
occupied = writeValuesWithRewind(before, occupied, payload, 2)
if occupied != occupiedAfter {
t.Fatalf("got occupied %d are not equal expected %d", occupied, occupiedAfter)
wp
2026-06-13 22:43:17 +00:00
}
wp
2026-06-14 07:57:01 +03:00
if !bytes.Equal(before, after) {
t.Fatalf("got buf % x are not equal expected % x", before, after)
wp
2026-06-13 22:43:17 +00:00
}
}
wp
2026-06-14 23:12:03 +03:00
func TestWALReader(t *testing.T) {
rec1 := MetricAddRecord{
MetricID: 100,
MetricType: qb.Cumulative,
FracDigits: 4,
}
rec2 := MetricAddRecord{
MetricID: 200,
MetricType: qb.Instant,
FracDigits: 1,
}
rec3 := MetricDeleteRecord{
MetricID: 100,
FreeIndexPages: []uint32{
1, 2, 3,
},
FreeDataPages: []uint32{
10, 20,
},
}
rec4 := MetricDeleteRecord{
MetricID: 200,
}
packet1 := createPacket([]Packable{rec1, rec2})
packet2 := createPacket([]Packable{rec3, rec4})
src := bytes.NewBuffer(nil)
src.Write(packet1)
src.Write(packet2)
walReader, err := NewWALReader(src)
if err != nil {
t.Fatal(err)
}
i := 0
for {
records, isLastPacket, err := walReader.NextPacket()
if err != nil {
t.Fatal(err)
}
if records == nil {
break
}
switch i {
case 0:
if isLastPacket {
t.Fatalf("1st packet can't be the last packet in WAL")
}
if !reflect.DeepEqual(records, []any{rec1, rec2}) {
t.Fatalf("1st packet decoded incorrectly")
}
case 1:
if !isLastPacket {
t.Fatalf("2nd packet must be the last packet in WAL")
}
if !reflect.DeepEqual(records, []any{rec3, rec4}) {
t.Fatalf("2nd packet decoded incorrectly")
}
}
i++
}
}
func createPacket(records []Packable) []byte {
w := bytes.NewBuffer(nil)
w.Write([]byte{
0, 0, 0, 0, 0, 0, 0, 0, 0, // size (max 9 byte)
})
hasher := util.NewHasher()
multi := io.MultiWriter(w, hasher)
for _, rec := range records {
rec.Pack(multi)
}
return finalizePacket(w, hasher.Sum32())
}
func TestReplayMetric(t *testing.T) {
// data
DataPageSize = 24
DataPagePayloadSize = 12
dataCRC32Idx = DataPageSize - 4
timestampsSizeIdx = DataPageSize - 6
valuesSizeIdx = DataPageSize - 8
prevPageIdx = DataPageSize - 12
// index
IndexPageSize = 24
indexCRC32Idx = IndexPageSize - 4
indexRecordsCountIdx = IndexPageSize - 6
isZeroLevelIdx = IndexPageSize - 7
maxRecordsOnIndexPage = 2
//
metric := &ReplayMetric{
Buf: []byte{
0xa1, 0xa2, 0xa3, 0xa4,
0x00, 0x00, 0x00, 0x00,
0xb4, 0xb3, 0xb2, 0xb1,
0x01, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
},
TimestampsSize: 4,
ValuesSize: 4,
IndexLevelTails: []IndexLevelTail{
{
Buffer: []byte{
0x01, 0x01, 0x01, 0x01,
0x02, 0x02, 0x02, 0x02,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
},
RecordsCount: 1,
},
},
}
result := metric.MeasuresAppendWithGrow(MeasuresAppendWithGrowRecord{
DataPageTail: DataPageTail{
PrevPageNo: 1,
CRC32: 3235260875,
TimestampsRewindOffset: 1,
Timestamps: []byte{0x55, 0x44},
ValuesRewindOffset: 2,
Values: []byte{0x33, 0x44},
},
TailTimestamps: []byte{
0xf3, 0xf2, 0xf1,
},
TailValues: []byte{
0xd1, 0xd2,
},
ChangedIndexLevels: []ChangedIndexLevel{
{
IndexPageTail: &IndexPageTail{
CRC32: 2138044007,
Records: []byte{
0x03, 0x03, 0x03, 0x03,
0x04, 0x04, 0x04, 0x04,
},
},
TailRecords: []byte{
0x08, 0x08, 0x08, 0x08,
0x09, 0x09, 0x09, 0x09,
},
},
{
TailRecords: []byte{
0x01, 0x01, 0x01, 0x01,
0x02, 0x02, 0x02, 0x02,
},
},
},
}, true) // last packet
fmt.Printf("% x\n", metric.Buf)
fmt.Printf("% x\n", result.DataPagesToRewrite[0].Content)
fmt.Printf("% x\n", result.IndexPagesToRewrite[0].Content)
fmt.Printf("%d: % x\n", metric.IndexLevelTails[0].RecordsCount, metric.IndexLevelTails[0].Buffer)
fmt.Printf("%d: % x\n", metric.IndexLevelTails[1].RecordsCount, metric.IndexLevelTails[1].Buffer)
}
wp
2026-06-15 10:47:24 +00:00
// func TestComposeHeadIndexPage(t *testing.T) {
// var (
// levelIdx = 0
// level = storage.IndexLevelTail{
// Buffer: []byte{
// 1, 2, 3,
// },
// RecordsCount: 2,
// }
// head = &storage.IndexPageTail{
// PageNo: 100,
// CRC32: 12345,
// Records: []byte{},
// }
// )
// page := composeHeadIndexPage(levelIdx, level, head)
// if page.PageNo != head.PageNo {
// t.Fatalf("PageNo: got %d are not equal expected %v",
// page.PageNo, head.PageNo)
// }
// // fix compare pages
// }
func TestFindPageOnIndexTails(t *testing.T) {
levels := []IndexLevelTail{
{
Buffer: []byte{
0x64, 0x00, 0x00, 0x00, 0x0a, 0x00, 0x00, 0x00, // 100 => 10
0x6e, 0x00, 0x00, 0x00, 0x0b, 0x00, 0x00, 0x00, // 110 => 11
},
RecordsCount: 2,
},
{
Buffer: []byte{
0x0a, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, // 10 => 1
0x28, 0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, // 40 => 4
0x46, 0x00, 0x00, 0x00, 0x07, 0x00, 0x00, 0x00, // 70 => 7
},
RecordsCount: 3,
},
}
testCases := []struct {
Timestamp uint32
PageNo uint32
IsDataPage bool
}{
{Timestamp: 9, PageNo: 0, IsDataPage: false},
{Timestamp: 10, PageNo: 1, IsDataPage: false},
{Timestamp: 20, PageNo: 1, IsDataPage: false},
{Timestamp: 40, PageNo: 4, IsDataPage: false},
{Timestamp: 60, PageNo: 4, IsDataPage: false},
{Timestamp: 70, PageNo: 7, IsDataPage: false},
{Timestamp: 90, PageNo: 7, IsDataPage: false},
{Timestamp: 100, PageNo: 10, IsDataPage: true},
{Timestamp: 105, PageNo: 10, IsDataPage: true},
{Timestamp: 110, PageNo: 11, IsDataPage: true},
{Timestamp: 120, PageNo: 11, IsDataPage: true},
}
for _, testCase := range testCases {
pageNo, isDataPage := FindPageOnIndexLevelTails(levels, testCase.Timestamp)
if pageNo != testCase.PageNo {
t.Fatalf("timestamp %d: got pageNo %d are not equal expected %d",
testCase.Timestamp, pageNo, testCase.PageNo)
}
if isDataPage != testCase.IsDataPage {
t.Fatalf("timestamp %d: got isDataPage %t are not equal expected %t",
testCase.Timestamp, isDataPage, testCase.IsDataPage)
}
}
}
Reference in New Issue Copy Permalink