qb/database/replay_metric.go

package database

import (
	"fmt"
	"io"

	bin "gordenko.dev/dima/bin/little"
	"gordenko.dev/dima/qb"
	"gordenko.dev/dima/qb/enc"
	"gordenko.dev/dima/qb/storage"
)

type ReplayMetric struct {
	metricType      qb.MetricType
	fracDigits      byte
	lastPageNo      uint32
	buf             []byte
	databuf         []byte // якщо розмір buf == DataPageSize, то databuf буде менше (мінус футер)
	tSize           int
	vSize           int
	indexLevelTails []storage.IndexLevelTail // root - last element
}

func (s *ReplayMetric) ReadFrom(r io.Reader) (err error) {
	metricType, err := bin.ReadByte(r)
	if err != nil {
		return
	}
	s.metricType = qb.MetricType(metricType)
	s.fracDigits, err = bin.ReadByte(r)
	if err != nil {
		return
	}
	s.lastPageNo, err = bin.ReadUint32(r)
	if err != nil {
		return
	}
	s.tSize, err = bin.ReadUint16AsInt(r)
	if err != nil {
		return
	}
	s.vSize, err = bin.ReadUint16AsInt(r)
	if err != nil {
		return
	}
	err = bin.ReadNInto(r, s.databuf[len(s.databuf)-s.tSize:])
	if err != nil {
		return
	}
	err = bin.ReadNInto(r, s.databuf[:s.vSize])
	if err != nil {
		return
	}
	//
	levelsCount, err := bin.ReadVarSize(r)
	if err != nil {
		return
	}
	for range levelsCount {
		var (
			buf   = make([]byte, storage.IndexPageSize)
			count int
		)
		count, err = bin.ReadVarSize(r)
		if err != nil {
			return
		}
		err = bin.ReadNInto(r, buf[:count*indexRecordSize])
		if err != nil {
			return
		}
		s.indexLevelTails = append(s.indexLevelTails, storage.IndexLevelTail{
			Buffer:       buf,
			RecordsCount: count,
		})
	}
	return
}

// func appendNewRecordsToIndexTail(level storage.IndexLevelTail, records []byte) {

// }

// func countReusedIndexPages(changedIndexLevels []storage.ChangedIndexLevel) (count int) {

// 	return
// }

func (s *ReplayMetric) MeasuresAppend(rec storage.MeasuresAppendRecord) {
	timestampsSize := rec.TimestampsRewindOffset - len(rec.Timestamps)
	// копіюю нові дані із WAL з урахуванням offset-ів
	copy(s.databuf[rec.ValuesRewindOffset:], rec.Values)
	copy(s.databuf[len(s.databuf)-timestampsSize:], rec.Timestamps)
	//
	s.tSize = timestampsSize
	s.vSize = rec.ValuesRewindOffset - len(rec.Values)
}

type AppendMeasuresResult struct {
	IndexPages            []storage.PageToWrite
	DataPages             []storage.PageToWrite
	ReusedIndexPagesCount int
	ReusedDataPagesCount  int
}

func (s *ReplayMetric) MeasuresAppendWithGrow(rec storage.MeasuresAppendWithGrowRecord, isLastPacket bool) (_ AppendMeasuresResult) {
	var (
		reusedIndexPages int
		reusedDataPages  int
		indexPages       []storage.PageToWrite
		dataPages        []storage.PageToWrite
	)
	// Додати в index level tails недостаючі дані, або замінити
	for levelIdx, change := range rec.ChangedIndexLevels {
		var (
			head            = change.IndexPageTail
			newRecordsCount = len(change.TailRecords) / indexRecordSize
		)
		// 1. набиваю сторінки для перезапису в index файлі
		if isLastPacket {
			if head != nil {
				indexPages = append(indexPages,
					composeHeadIndexPage(levelIdx, s.indexLevelTails[levelIdx], head))
			}
			for _, x := range change.IndexPages {
				indexPages = append(indexPages, storage.PageToWrite{
					PageNo:  x.PageNo,
					Content: x.Content,
				})
			}
		}

		// 2. Вношу зміни в поточні індексні хвости
		if levelIdx < len(s.indexLevelTails) {
			level := s.indexLevelTails[levelIdx]
			if head != nil {
				// попередній хвіст індексного рівня перетворився на head сторінку,
				// отже TailRecords - це новий хвіст
				copy(level.Buffer, change.TailRecords)
				level.RecordsCount = newRecordsCount
			} else {
				// TailRecords - це нові дані, які треба додати
				pos := level.RecordsCount * indexRecordSize
				copy(level.Buffer[pos:], change.TailRecords)
				level.RecordsCount += newRecordsCount
			}
			s.indexLevelTails[levelIdx] = level
		} else {
			// додаю новий індексний рівень
			buf := make([]byte, storage.IndexPageSize)
			copy(buf, change.TailRecords)
			//
			s.indexLevelTails = append(s.indexLevelTails, storage.IndexLevelTail{
				Buffer:       buf,
				RecordsCount: newRecordsCount,
			})
		}

		// 3. рахую кількість reused індексних сторінок
		if head != nil {
			if head.Reused {
				reusedIndexPages++
			}
		}
		for _, x := range change.IndexPages {
			if x.Reused {
				reusedIndexPages++
			}
		}
	}

	// 4. набиваю сторінки для перезапису в data файлі
	if isLastPacket {
		dataPages = append(dataPages, composeHeadDataPage(s.databuf, rec.DataPageTail))

		for _, x := range rec.DataPages {
			dataPages = append(dataPages, storage.PageToWrite{
				PageNo:  x.PageNo,
				Content: x.Content,
			})
		}
	}

	// рахую кількість reused дата сторінок
	if rec.DataPageTail.Reused {
		reusedDataPages++
	}
	for _, x := range rec.DataPages {
		if x.Reused {
			reusedDataPages++
		}
	}
	// HeadDataPage є полюбому, оскільки це транзакція із мінімум однією заповненою
	// data сторінкою. Отже TailTimestamps і TailValues - це нові хвости data рівня.
	copy(s.databuf, rec.TailValues)
	pos := len(s.databuf) - len(rec.TailTimestamps)
	copy(s.databuf[pos:], rec.TailTimestamps)

	s.tSize = len(rec.TailTimestamps)
	s.vSize = len(rec.TailValues)

	if len(rec.DataPages) == 0 {
		s.lastPageNo = rec.DataPageTail.PageNo
	} else {
		s.lastPageNo = rec.DataPages[len(rec.DataPages)-1].PageNo
	}

	return AppendMeasuresResult{
		IndexPages:            indexPages,
		DataPages:             dataPages,
		ReusedIndexPagesCount: reusedIndexPages,
		ReusedDataPagesCount:  reusedDataPages,
	}
}

// HELPERS

func composeHeadIndexPage(levelIdx int, level storage.IndexLevelTail, head *storage.IndexPageTail) storage.PageToWrite {
	// розраховую pos, з якого буду дописувати хвіст
	pos := level.RecordsCount * indexRecordSize
	// створюю копію сторінки
	page := make([]byte, storage.IndexPageSize)
	copy(page, level.Buffer[:pos]) // поточні дані
	copy(page[pos:], head.Records)
	// запечатати сторінку
	calculatedCRC := storage.SealIndexPage(storage.SealIndexPageIn{
		Content:      page,
		RecordsCount: level.RecordsCount + len(head.Records)/indexRecordSize,
		ZeroLevel:    levelIdx == 0,
	})
	// перевірка CRC
	if calculatedCRC != head.CRC32 {
		qb.Abort(qb.WALReplayFailed,
			fmt.Errorf("calculated CRC %d not equal expected %d of head page %d on index level %d",
				calculatedCRC, head.CRC32, head.PageNo, levelIdx))
	}
	return storage.PageToWrite{
		PageNo:  head.PageNo,
		Content: page,
	}
}

func composeHeadDataPage(databuf []byte, head storage.DataPageTail) storage.PageToWrite {
	// створюю копію сторінки
	var (
		page           = make([]byte, storage.DataPageSize)
		timestampsSize = head.TimestampsRewindOffset - len(head.Timestamps)
	)
	// копіюю поточні дані
	copy(page, databuf)
	// копіюю нові дані із WAL з урахуванням offset-ів
	copy(page[head.ValuesRewindOffset:], head.Values)
	copy(page[len(page)-timestampsSize:], head.Timestamps)
	// запечатати сторінку
	calculatedCRC := storage.SealDataPage(storage.SealDataPageIn{
		Content:        page,
		PrevPageNo:     head.PrevPageNo,
		TimestampsSize: timestampsSize,
		ValuesSize:     head.ValuesRewindOffset + len(head.Values),
	})
	// перевірка CRC
	if calculatedCRC != head.CRC32 {
		qb.Abort(qb.WALReplayFailed,
			fmt.Errorf("calculated CRC %d not equal expected %d of head data page %d",
				calculatedCRC, head.CRC32, head.PageNo))
	}
	return storage.PageToWrite{
		PageNo:  head.PageNo,
		Content: page,
	}
}

func (s *ReplayMetric) ToMetric() *_metric {

	databuf := s.buf[:storage.DataPagePayloadSize]
	values := enc.NewValueDeltaCompressor(s.metricType, s.fracDigits, databuf, s.vSize)
	return &_metric{
		metricType:      s.metricType,
		fracDigits:      s.fracDigits,
		lastPageNo:      s.lastPageNo,
		lastValue:       values.LastValue(),
		buffer:          s.buf,
		timestamps:      enc.NewTimeDeltaCompressor(databuf, s.tSize),
		values:          values,
		indexLevelTails: s.indexLevelTails,
	}
}