qb/storage/wal_replayer.go

package storage

import (
	"errors"
	"fmt"
)

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

type WALReplayerOptions struct {
	WALReader      *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) IndexPagesToRewrite() []PageToWrite {
	return s.indexPages
}

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

func (s *WALReplayer) FreeIndexPages() []uint32 {
	return s.freeIndexPages
}

func (s *WALReplayer) FreeDataPages() []uint32 {
	return s.freeDataPages
}

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 MeasuresAppendRecord:
		if err = s.onMeasuresAppend(rec); err != nil {
			return
		}
	case MeasuresAppendWithGrowRecord:
		if err = s.onMeasuresAppendWithGrow(rec, isLastPacket); err != nil {
			return
		}
	case MetricAddRecord:
		if err = s.onMetricAdd(rec); err != nil {
			return
		}
	case MetricDeleteRecord:
		if err = s.onMetricDelete(rec); err != nil {
			return
		}
		// case 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 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, DataPageSize)
	)
	s.metrics[rec.MetricID] = &ReplayMetric{
		MetricType: rec.MetricType,
		FracDigits: rec.FracDigits,
		Buf:        buf,
	}
	return
}

func (s *WALReplayer) onMetricDelete(rec 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 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 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.IndexPagesToRewrite...)
		s.dataPages = append(s.dataPages, result.DataPagesToRewrite...)
	}
	//
	s.freeIndexPages = s.freeIndexPages[:len(s.freeIndexPages)-result.ReusedIndexPagesCount]
	s.freeDataPages = s.freeDataPages[:len(s.freeDataPages)-result.ReusedDataPagesCount]
	return
}