qb/database/wal_replayer.go

package database

import (
	"errors"
	"fmt"

	"gordenko.dev/dima/qb/storage"
)

type WALReplayer struct {
	walReader      *storage.WALReader
	metrics        map[uint32]*ReplayMetric
	freeIndexPages []uint32
	freeDataPages  []uint32
	indexPages     []storage.PageToWrite
	dataPages      []storage.PageToWrite
}

type WALReplayerOptions struct {
	WALReader      *storage.WALReader
	Metrics        map[uint32]*ReplayMetric // from snapshot
	FreeIndexPages []uint32                 // from snapshot
	FreeDataPages  []uint32                 // from snapshot
}

func NewWALReplayer(opt WALReplayerOptions) (*WALReplayer, error) {
	if opt.WALReader == nil {
		return nil, errors.New("required option: WALReader")
	}
	if opt.Metrics == nil {
		return nil, errors.New("required option: Metrics")
	}
	return &WALReplayer{
		walReader:      opt.WALReader,
		metrics:        opt.Metrics,
		freeIndexPages: opt.FreeIndexPages,
		freeDataPages:  opt.FreeDataPages,
	}, nil
}

func (s *WALReplayer) IndexPages() []storage.PageToWrite {
	return s.indexPages
}

func (s *WALReplayer) DataPages() []storage.PageToWrite {
	return s.dataPages
}

func (s *WALReplayer) Replay() error {
	for {
		records, isLastPacket, err := s.walReader.NextPacket()
		if err != nil {
			return err
		}
		for _, rec := range records {
			if err = s.replayRecord(rec, isLastPacket); err != nil {
				return err
			}
		}
	}
}

func (s *WALReplayer) replayRecord(untyped any, isLastPacket bool) (err error) {
	switch rec := untyped.(type) {
	case storage.MeasuresAppendRecord:
		if err = s.onMeasuresAppend(rec); err != nil {
			return
		}
	case storage.MeasuresAppendWithGrowRecord:
		if err = s.onMeasuresAppendWithGrow(rec, isLastPacket); err != nil {
			return
		}
	case storage.MetricAddRecord:
		if err = s.onMetricAdd(rec); err != nil {
			return
		}
	case storage.MetricDeleteRecord:
		if err = s.onMetricDelete(rec); err != nil {
			return
		}
		// case storage.DeletedMeasures:
		// 	metric, ok := s.metrics[rec.MetricID]
		// 	if ok {
		// 		metric.DeleteMeasures()
		// 		if len(rec.FreePageNumbers) > 0 {
		// 			s.freeList.AddPages(rec.FreePageNumbers)
		// 		}
		// 	}

		// default:
		// 	qb.Abort(qb.UnknownstorageRecordTypeBug,
		// 		fmt.Errorf("bug: unknown record type %T in TransactionLog", rec))
	}
	return nil
}

func (s *WALReplayer) onMetricAdd(rec storage.MetricAddRecord) (err error) {
	_, ok := s.metrics[rec.MetricID]
	if ok {
		return fmt.Errorf("metric %d add failed: already added", rec.MetricID)
	}
	var (
		buf = make([]byte, storage.DataPageSize)
	)
	s.metrics[rec.MetricID] = &ReplayMetric{
		metricType: rec.MetricType,
		fracDigits: byte(rec.FracDigits),
		buf:        buf,
		databuf:    buf[:storage.DataPagePayloadSize],
	}
	return
}

func (s *WALReplayer) onMetricDelete(rec storage.MetricDeleteRecord) error {
	_, ok := s.metrics[rec.MetricID]
	if !ok {
		return fmt.Errorf("metric %d deletion failed: not found", rec.MetricID)
	}
	delete(s.metrics, rec.MetricID)
	s.freeIndexPages = append(s.freeIndexPages, rec.FreeIndexPages...)
	s.freeDataPages = append(s.freeDataPages, rec.FreeDataPages...)
	return nil
}

func (s *WALReplayer) onMeasuresAppend(rec storage.MeasuresAppendRecord) (err error) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		return fmt.Errorf("append measures failed: metric %d not found", rec.MetricID)
	}
	metric.MeasuresAppend(rec)
	return
}

func (s *WALReplayer) onMeasuresAppendWithGrow(rec storage.MeasuresAppendWithGrowRecord, isLastPacket bool) (err error) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		return fmt.Errorf("append measures failed: metric %d not found", rec.MetricID)
	}
	result := metric.MeasuresAppendWithGrow(rec, isLastPacket)
	//
	if isLastPacket {
		s.indexPages = append(s.indexPages, result.IndexPages...)
		s.dataPages = append(s.dataPages, result.DataPages...)
	}
	//
	s.freeIndexPages = s.freeIndexPages[:len(s.freeIndexPages)-result.ReusedIndexPagesCount]
	s.freeDataPages = s.freeDataPages[:len(s.freeDataPages)-result.ReusedDataPagesCount]
	return
}

//func (s *WALReplayer) collectPagesToWrite(metric *ReplayMetric, rec storage.WALRecordAppendMeasures) {
// metric.ApplyHeadDataPage(rec.CompletedDataPage)

// s.dataPages = append(s.dataPages, storage.PageToWrite{
// 	PageNo:  p.PageNo,
// 	Content: metric.buf,
// })

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

//}

// func (s *WALReplayer) replayPacket(records []any, isLastPacket bool) {
// 	for _, untyped := range records {
// 		switch rec := untyped.(type) {
// 		case storage.WALRecordAppendMeasures:
// 			s.onWALRecordAppendMeasures(rec, isLastPacket)
// 		}
// 	}
// 	return
// }