qb/txlog/page_preparer.go

package txlog

import (
	"errors"

	"gordenko.dev/dima/qb"
	"gordenko.dev/dima/qb/bin"
	"gordenko.dev/dima/qb/util"
)

// Задача метода:
// - для data сторінок отримати pageNo та reused, задати sizes та prevPageNo, порахувати CRC32
// - додати пари до index, згенерувати нові index сторінки
// levels []*IndexLevel, dataPages []*DataPage
// fix - reduce leves

type PagePreparer struct {
	dataChecksumIdx       int
	maxRecordsOnIndexPage int
	indexPageIncSize      int // кратно indexPageSize
	timestampsSizeIdx     int
	valuesSizeIdx         int
	prevPageIdx           int
	indexRecordsCountIdx  int
	isZeroLevelIdx        int
	indexChecksumIdx      int
	getDataPageNumber     func() (uint32, bool, error)
	getIndexPageNumber    func() (uint32, bool, error)
}

type PagePreparerOptions struct {
	IndexPageSize      int
	IndexPageIncSize   int
	DataPageSize       int
	GetDataPageNumber  func() (uint32, bool, error)
	GetIndexPageNumber func() (uint32, bool, error)
}

func NewPagePreparer(opt PagePreparerOptions) (*PagePreparer, error) {
	if opt.GetIndexPageNumber == nil {
		return nil, errors.New("missing required option: GetIndexPageNumber")
	}
	if opt.GetDataPageNumber == nil {
		return nil, errors.New("missing required option: GetDataPageNumber")
	}
	if (opt.IndexPageSize % opt.IndexPageIncSize) != 0 {
		return nil, errors.New("IndexPageIncSize must multiple of IndexPageSize")
	}
	s := &PagePreparer{
		dataChecksumIdx:      opt.DataPageSize - 4,
		indexPageIncSize:     opt.IndexPageIncSize,
		timestampsSizeIdx:    opt.DataPageSize - 6,
		valuesSizeIdx:        opt.DataPageSize - 8,
		prevPageIdx:          opt.DataPageSize - 12,
		indexRecordsCountIdx: opt.IndexPageSize - 6,
		isZeroLevelIdx:       opt.IndexPageSize - 7,
		indexChecksumIdx:     opt.IndexPageSize - 4,
		getIndexPageNumber:   opt.GetIndexPageNumber,
		getDataPageNumber:    opt.GetDataPageNumber,
	}
	s.maxRecordsOnIndexPage = (opt.IndexPageSize - 7) / indexRecordSize
	return s, nil
}

type appendIndexRecordIn struct {
	Records      []byte
	RecordsCount int
	Timestamp    uint32
	PageNo       uint32
}

func (s *PagePreparer) appendIndexRecord(in appendIndexRecordIn) []byte {
	if in.RecordsCount < s.maxRecordsOnIndexPage {
		var (
			pos int
			buf = in.Records
		)
		if in.RecordsCount > 0 {
			pos = in.RecordsCount * indexRecordSize
		} else {
			buf = make([]byte, s.indexPageIncSize)
		}
		bin.PutUint32(buf[pos:], in.Timestamp)
		bin.PutUint32(buf[pos+4:], in.PageNo)
		return buf
	}
	qb.Abort(qb.NoSpaceOnIndexPage, nil)
	return nil
}

type sealDataPageIn struct {
	Content        []byte
	PrevPageNo     uint32
	TimestampsSize int
	ValuesSize     int
}

func (s *PagePreparer) sealDataPage(in sealDataPageIn) (checksum uint32) {
	bin.PutUint16(in.Content[s.timestampsSizeIdx:], uint16(in.TimestampsSize))
	bin.PutUint16(in.Content[s.valuesSizeIdx:], uint16(in.ValuesSize))
	bin.PutUint32(in.Content[s.prevPageIdx:], in.PrevPageNo)

	checksum = util.CalcChecksum(in.Content[:s.dataChecksumIdx])
	bin.PutUint32(in.Content[s.dataChecksumIdx:], checksum)
	return
}

type sealIndexPageIn struct {
	Content      []byte
	RecordsCount int
	LastLevel    bool
}

func (s *PagePreparer) sealIndexPage(in sealIndexPageIn) (checksum uint32) {
	bin.PutUint16(in.Content[s.indexRecordsCountIdx:], uint16(in.RecordsCount))
	if in.LastLevel {
		in.Content[s.isZeroLevelIdx] = 1
	}
	checksum = util.CalcChecksum(in.Content[:s.indexChecksumIdx])
	bin.PutUint32(in.Content[s.indexChecksumIdx:], checksum)
	return
}

type IndexLevelTail struct {
	Records      []byte // розмір більший за кількість
	RecordsCount int
}

type DataPayload struct {
	Since          uint32
	Content        []byte
	TimestampsSize int
	ValuesSize     int
}

type SealPagesIn struct {
	LastPageNo      uint32
	IndexLevelTails []IndexLevelTail
	DataPages       []DataPayload
}

type SealedDataPage struct {
	PageNo     uint32
	Reused     bool
	PrevPageNo uint32
	Checksum   uint32
	Content    []byte
}

type SealedIndexPage struct {
	Content  []byte
	PageNo   uint32
	Reused   bool
	Checksum uint32
}

// SealedIndexLevel - як зрозуміти чи потрібно щось писати в WAL чи index файл,
// чи додавання нових data сторінок не зачепило індексний рівень?
// Якщо є IndexPages - пишемо весь рівень в WAL і готові сторінки в index файл.
// Якщо IndexPages пустий - порівнюємо Offset і len(Payload) - якщо довжина
// більша за змішення - дані додані на незаповнену сторінку рівня,
// отже різницю пишемо в WAL.
type SealedIndexLevel struct {
	// Якщо є IndexPages, offset - це зміщення із якого починаються дані першої сторінки,
	// які ще не зписані в WAL.
	// Якщо IndexPages пустий, offset - це зміщення із якого починаються дані payload,
	// які ще не зписані в WAL.
	// В payload в будь-якому разі дані останньої індексної сторінки рівня (незаповненої).
	SkipRecords      int
	IndexPages       []SealedIndexPage
	TailRecords      []byte
	TailRecordsCount int
}

type SealResult struct {
	DataPages   []SealedDataPage
	IndexLevels []*SealedIndexLevel
}

func (s *PagePreparer) SealPages(in SealPagesIn) (result SealResult) {
	var (
		sealedDataPages   []SealedDataPage
		sealedIndexLevels []*SealedIndexLevel
	)
	for _, tail := range in.IndexLevelTails {
		sealedIndexLevels = append(sealedIndexLevels, &SealedIndexLevel{
			SkipRecords:      tail.RecordsCount,
			TailRecords:      tail.Records,
			TailRecordsCount: tail.RecordsCount,
		})
	}
	for dataPageIdx, d := range in.DataPages {
		var (
			upTimestamp = d.Since
			upPageNo    uint32
			levelIdx    = 0
			prevPageNo  uint32
		)
		if dataPageIdx == 0 {
			prevPageNo = in.LastPageNo
		} else {
			prevPageNo = result.DataPages[dataPageIdx-1].PageNo
		}
		pageNo, reused, err := s.getDataPageNumber()
		if err != nil {
			qb.Abort(qb.FailedGetPageNumber, err)
		}
		sealedDataPages = append(sealedDataPages, SealedDataPage{
			Content:    d.Content,
			PrevPageNo: prevPageNo,
			PageNo:     pageNo,
			Reused:     reused,
			Checksum: s.sealDataPage(sealDataPageIn{
				Content:        d.Content,
				PrevPageNo:     prevPageNo,
				TimestampsSize: d.TimestampsSize,
				ValuesSize:     d.ValuesSize,
			}),
		})
		upPageNo = pageNo
		for {
			if levelIdx == len(sealedIndexLevels) {
				// новий root
				sealedIndexLevels = append(sealedIndexLevels, &SealedIndexLevel{
					TailRecords: s.appendIndexRecord(appendIndexRecordIn{
						Records:   nil,
						Timestamp: upTimestamp,
						PageNo:    upPageNo,
					}),
					TailRecordsCount: 1,
				})
				break
			}
			level := sealedIndexLevels[levelIdx]
			if level.TailRecordsCount < s.maxRecordsOnIndexPage {
				level.TailRecords = s.appendIndexRecord(appendIndexRecordIn{
					Records:   level.TailRecords,
					Timestamp: upTimestamp,
					PageNo:    upPageNo,
				})
				level.TailRecordsCount++
				break
			}
			// в буфері немає місця для додавання нової пари, отже це
			// заповнена сторінка
			pageNo, reused, err := s.getIndexPageNumber()
			if err != nil {
				qb.Abort(qb.FailedGetPageNumber, err)
			}

			filled := SealedIndexPage{
				Content: level.TailRecords,
				PageNo:  pageNo,
				Reused:  reused,
				Checksum: s.sealIndexPage(sealIndexPageIn{
					Content:      level.TailRecords,
					RecordsCount: level.TailRecordsCount,
					LastLevel:    levelIdx == 0,
				}),
			}
			level.IndexPages = append(level.IndexPages, filled)
			// новий tail на індексному рівні
			level.TailRecords = s.appendIndexRecord(appendIndexRecordIn{
				Records:   nil,
				Timestamp: upTimestamp,
				PageNo:    upPageNo,
			})
			level.TailRecordsCount = 1
			//
			upPageNo = pageNo
			upTimestamp = bin.GetUint32(filled.Content)
			levelIdx++
		}
	}
	return SealResult{
		DataPages:   sealedDataPages,
		IndexLevels: sealedIndexLevels,
	}
}