qb/txlog/x.go

package txlog

import (
	"bytes"
	"fmt"
	"log"
	"os"

	bin "gordenko.dev/dima/bin/little"
	"gordenko.dev/dima/qb/atree"
)

const recordSize = 8

type LogWriter struct {
	w         *bytes.Buffer
	wal       *os.File
	dataFile  *os.File
	indexFile *os.File
	dataPage  []byte
	indexPage []byte
}

func (s *LogWriter) writePagesToAtree(levels []*atree.IndexLevel, dataPages []*atree.DataPage) (err error) {
	for _, p := range dataPages {
		atree.ChunksToDataPage(s.dataPage, atree.ChunksToDataPageReq{
			PrevPageNo:     p.PrevPageNo,
			Timestamps:     p.Timestamps,
			TimestampsSize: p.TimestampsSize,
			Values:         p.Values,
			ValuesSize:     p.ValuesSize,
		})
		off := (p.PageNo - 1) * atree.DataPageSize
		n, err := s.dataFile.WriteAt(s.dataPage, int64(off))
		if err != nil {
			return err
		}
		if n != atree.DataPageSize {
			return fmt.Errorf("write %d instead of %d", n, atree.DataPageSize)
		}
	}
	for levelIdx, level := range levels {
		for _, p := range level.Filled {
			// if len(p.Data) != atree.PageSize {
			// 	return fmt.Errorf("wrong page %d size: %d",
			// 		p.PageNo, len(p.Data))
			// }
			atree.DataToIndexPage(s.indexPage, p.Data, levelIdx == 0)
			off := (p.PageNo - 1) * atree.IndexPageSize
			n, err := s.indexFile.WriteAt(s.indexPage, int64(off))
			if err != nil {
				return err
			}
			if n != atree.IndexPageSize {
				return fmt.Errorf("write %d instead of %d", n, atree.IndexPageSize)
			}
		}
	}
	return nil
}

type ToWrite struct {
	MetricID         uint32
	TimestampsOffset int // (заповнені одразу)
	ValuesOffset     int // (заповнені одразу)
	Timestamps       [][]byte
	TimestampsSize   int
	Values           [][]byte
	ValuesSize       int
	IndexLevels      []*atree.IndexLevel
	DataPages        []*atree.DataPage // fix - prevPageNo for the 1st data page
}

type DataPage struct {
	PageNo     uint32
	Reused     bool
	PrevPageNo uint32
	Checksum   uint32
	Timestamps []byte
	Values     []byte
}

type IndexPage struct {
	PageNo   uint32
	Reused   bool
	Checksum uint32
	Records  []byte
}

type IndexLevel struct {
	Pages []IndexPage
	// offset потрібен тому що дані не додаються в кінець, а перезаписують кілька
	// останніх байтів unfilled даних. Потрібно для коректного recovery.
	// Якщо є Pages, то застосовуються для 1-ї Index сторінки, інакше - до records (unfilled)
	RecordsOffset int
	Records       []byte // unfilled
}

type TxAppendedMeasures struct {
	MetricID  uint32
	DataPages []DataPage
	// offsets потрібні тому що дані не додаються в кінець, а перезаписують кілька
	// останніх байтів unfilled даних. Потрібно для коректного recovery.
	// Якщо є DataPages, то застосовуються для 1-ї Data сторінки, інакше - до timestamps і values (unfilled)
	TimestampsOffset int
	ValuesOffset     int
	Timestamps       []byte // unfilled
	Values           []byte // unfilled
	IndexLevels      []DataPage
}

/*
Format appended measures:
1b - tx type
4b - metricID
Nb - qty of filled data pages (varsize)
[
	4b - pageNo
	1b - reused
	4b - prevPageNo
	4b - page crc32
	Nb - timestamps size (varsize)
	Nb - timestamps payload
	Nb - values size (varsize)
	Nb - values payload
]
2b - timestamps offset (for the 1st data page only)
2b - values offset (for the 1st data page only)
Nb - timestamps size unfilled (varsize)
Nb - timestamps payload unfilled
Nb - values size unfilled (varsize)
Nb - values payload unfilled
Nb - qty of index levels (varsize)
[
	Nb - qty of level filled pages (varsize)
	[
		4b - pageNo
		1b - reused
		4b - page crc32
		Nb - records qty (varsize)
		Nb - records payload
	]
	Nb - records offset (for the 1st index page only)
	Nb - records qty unfilled (varsize)
	Nb - records payload unfilled
]
*/

func (s *LogWriter) writeToWAL(req ToWrite) (err error) {
	// fix write code
	bin.WriteUint32(s.w, req.MetricID)
	bin.WriteVarSize(s.w, len(req.DataPages))
	for idx, p := range req.DataPages {
		bin.WriteUint32(s.w, p.PageNo)
		var reused byte
		if p.IsReused {
			reused = 1
		}
		s.w.WriteByte(reused)
		bin.WriteUint32(s.w, p.PrevPageNo)
		//bin.WriteUint32(s.w, p.CRC32) fix
		var (
			timestampsOffset int
			valuesOffset     int
		)
		if idx == 0 {
			timestampsOffset = req.TimestampsOffset
			valuesOffset = req.ValuesOffset
		}
		copyPayloadFromChunks(s.w, p.Timestamps, p.TimestampsSize, timestampsOffset)
		copyPayloadFromChunks(s.w, p.Values, p.ValuesSize, valuesOffset)
	}
	bin.WriteUint16(s.w, uint16(req.TimestampsOffset))
	bin.WriteUint16(s.w, uint16(req.ValuesOffset))
	// fix offset
	copyPayloadFromChunks(s.w, req.Timestamps, req.TimestampsSize, req.TimestampsOffset)
	copyPayloadFromChunks(s.w, req.Values, req.ValuesSize, req.ValuesOffset)
	if len(req.DataPages) == 0 {
		return
	}
	bin.WriteVarSize(s.w, len(req.IndexLevels))
	for _, level := range req.IndexLevels {
		bin.WriteVarSize(s.w, len(level.Filled))
		for idx, p := range level.Filled {
			bin.WriteUint32(s.w, p.PageNo)
			var reused byte
			if p.IsReused {
				reused = 1
			}
			s.w.WriteByte(reused)
			//bin.WriteUint32(s.w, p.CRC32) fix
			records := p.Data
			if idx == 0 {
				records = p.Data[level.Offset:]
			}
			if (len(records) % recordSize) != 0 {
				log.Fatalf("wrong records length: %d", len(records))
			}
			bin.WriteVarSize(s.w, len(records)/recordSize)
			s.w.Write(records)
		}
		if len(level.Filled) > 0 {
			s.w.Write(level.Data)
		} else {
			s.w.Write(level.Data[level.Offset:])
		}
	}
	return
}

const chunkSize = 24

func copyPayloadFromChunks(w *bytes.Buffer, chunks [][]byte, size int, offset int) {
	bin.WriteVarSize(w, size)
	var (
		chunkIdx = offset / chunkSize
		byteIdx  = offset % chunkSize
	)
	for _, chunk := range chunks[chunkIdx:] {
		available := len(chunk) - byteIdx
		if available <= size {
			w.Write(chunk[byteIdx:])
			size -= available
			if size == 0 {
				return
			}
			byteIdx = 0
		} else {
			w.Write(chunk[byteIdx : byteIdx+size])
			return
		}
	}
}