qb/database/metric.go

package database

import (
	"io"

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

// METRIC

const minBufferSize = 1024

var (
	indexRecordSize = 8
)

type _metric struct {
	metricType qb.MetricType
	fracDigits byte
	lastPageNo uint32
	//SinceValue      float64
	//Since     uint32
	lastValue float64
	//Until           uint32
	buffer          []byte
	timestamps      qb.TimestampCompressor
	values          qb.ValueCompressor
	XLock           bool
	RLocks          int
	WaitQueue       []any
	indexLevelTails []storage.IndexLevelTail // root - last element
	capturedState   *CapturedState
}

//IndexLevels [][]IndexRec // root - last element

// func (s *_metric) ReinitBy(timestamp uint32, value float64) {
// 	s.Timestamps.Renew()
// 	s.Values.Renew()

// 	s.Timestamps.Append(timestamp)
// 	s.Values.Append(value)

// 	s.Since = timestamp
// 	s.SinceValue = value
// 	s.Until = timestamp
// 	s.UntilValue = value
// }

func (s *_metric) DeleteMeasures() {
	// s.Timestamps.Renew()
	// s.Values.Renew()

	// s.LastPageNo = 0
	// s.Since = 0
	// s.SinceValue = 0
	// s.Until = 0
	// s.UntilValue = 0
}

type CapturedState struct {
	Pages      []storage.DataPayload
	TimeState  qb.TimeDeltaCapturedState
	ValueState qb.ValueDeltaCapturedState
}

// func (s *_metric) startAppendMeasures(req tryAppendMeasuresReq, sendToStorage func(storage.AppendedMeasures)) {
func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, sendToStorage func(any)) {
	if s.capturedState != nil {
		s.WaitQueue = append(s.WaitQueue, req)
	}
	var (
		timestamps = s.timestamps
		values     = s.values

		// офсети на head сторінці
		timestampsOffset int
		valuesOffset     int

		written    int
		resultCode byte
	)

	s.capturedState = &CapturedState{
		//
		TimeState:  s.timestamps.CaptureState(),
		ValueState: s.values.CaptureState(),
	}

	for idx, measure := range req.Measures {
		if measure.Timestamp <= s.timestamps.LastTimestamp() {
			resultCode = ExpiredMeasure
			break
		}

		if s.metricType == qb.Cumulative && measure.Value < s.lastValue {
			resultCode = NonMonotonicValue
			break
		}

		// fix - 1 + 8 bytes

		tReport := timestamps.Evaluate(tmp, measure.Timestamp)
		vReport := values.Evaluate(tmp, measure.Value)

		totalRequiredSpace := tReport.TotalSpace + vReport.TotalSpace

		if totalRequiredSpace <= len(s.buffer) {
			// якщо на сторінці є місце
			if idx == 0 {
				timestampsOffset = tReport.Offset
				valuesOffset = vReport.Offset
			}
		} else {
			if len(s.buffer) < storage.DataPageSize {
				// allocate bigger buffer
				buf, databuf := growBuffers(growBuffersIn{
					databuf:       s.buffer,
					tSize:         timestamps.Size(),
					vSize:         values.Size(),
					requiredSpace: totalRequiredSpace,
				})

				s.buffer = buf
				// replace buffer in timestamps and values
				timestamps.Rotate(databuf) // fix pos
				values.Rotate(databuf)     // fix pos
			} else {
				// сторінка заповнена
				since := s.timestamps.ReplaceSinceWithUntil()

				if len(s.capturedState.Pages) == 0 {
					// head page FIX
					// timestampsPayload =
					// valuesPayload =
				}

				s.capturedState.Pages = append(s.capturedState.Pages, storage.DataPayload{
					Since:          since,
					Content:        s.buffer,
					TimestampsSize: timestamps.Size(),
					ValuesSize:     values.Size(),
				})

				buffer := make([]byte, minBufferSize)

				timestamps.Rotate(buffer)
				values.Rotate(buffer)

				// renew
				s.buffer = buffer
			}
		}

		timestamps.Append(tReport.Offset, tmp[:tReport.ChangeSize], measure.Timestamp)
		values.Append(vReport.Offset, tmp[7:7+vReport.ChangeSize], measure.Value, vReport.Delta)
		//
		s.lastValue = measure.Value
		written++
	}

	if written == 0 {
		s.capturedState = nil
		req.ResultCh <- tryAppendMeasuresResult{
			ResultCode: resultCode,
		}
		return
	}

	// виділити змінені байти.
	// скопіювати. Причому можна скопіювати зрізи chunks

	if len(s.capturedState.Pages) > 0 {
		head := s.capturedState.Pages[0]
		databuf := head.Content[:storage.DataPagePayloadSize]
		// пишу в storage довгим шляхом через redo файл і запис в data файл
		sendToStorage(storage.AppendedMeasures{
			MetricID:         req.MetricID,
			LastPageNo:       s.lastPageNo,
			TimestampsOffset: timestampsOffset,
			Timestamps:       databuf[len(databuf)-head.TimestampsSize : len(databuf)-timestampsOffset],
			ValuesOffset:     valuesOffset,
			Values:           databuf[valuesOffset:head.ValuesSize],
			IndexLevelTails:  s.indexLevelTails,
			DataPages:        s.capturedState.Pages,
			TailTimestamps:   nil, // payload
			TailValues:       s.buffer[:values.Size()],
			ResultCode:       resultCode,
			WrittenCount:     written,
			ResultCh:         nil,
		})
	} else {
		databuf := s.buffer
		if len(s.buffer) == storage.DataPageSize {
			databuf = s.buffer[:storage.DataPagePayloadSize]
		}
		// короткий шлях - запис лише в storage
		sendToStorage(storage.AppendedMeasures{
			MetricID:         req.MetricID,
			TimestampsOffset: timestampsOffset,
			ValuesOffset:     valuesOffset,
			Timestamps:       databuf[len(databuf)-timestamps.Size() : len(databuf)-timestampsOffset],
			Values:           databuf[valuesOffset:values.Size()],
			ResultCode:       resultCode,
			WrittenCount:     written,
			ResultCh:         nil,
		})
	}

}

func (s *_metric) FinAppendMeasures(rec storage.AppendMeasuresSummary) {
	// Видаляю state. Оригінальні Timestamps і Values вже мають останню версію
	s.capturedState = nil

	if rec.LastPageNo > 0 {
		s.lastPageNo = rec.LastPageNo
	}
	// В storage я передав повний індекс. У нього додали елементи (можливо нові рівні).
	// Тому проста заміна
	s.indexLevelTails = rec.Index
}

// READ

func (s *_metric) StartRangeScan(req tryRangeScanReq) {
	// if s.Since == 0 {
	// 	req.ResultCh <- rangeScanResult{
	// 		ResultCode: QueryDone,
	// 	}
	// 	return
	// }

	// if req.Since > s.Until {
	// 	req.ResultCh <- rangeScanResult{
	// 		ResultCode: QueryDone,
	// 	}
	// 	return
	// }

	// if req.Until < s.Since {
	// 	if s.RootPageNo > 0 {
	// 		req.ResultCh <- rangeScanResult{
	// 			ResultCode: UntilNotFound,
	// 			RootPageNo: s.RootPageNo,
	// 			FracDigits: s.FracDigits,
	// 		}
	// 		s.RLocks++
	// 		return
	// 	} else {
	// 		req.ResultCh <- rangeScanResult{
	// 			ResultCode: QueryDone,
	// 		}
	// 		return
	// 	}
	// }

	// timestampDecompressor := s.Timestamps.CreateDecompressor()
	// valueDecompressor := s.Values.CreateDecompressor(s.FracDigits)

	// for {
	// 	timestamp, done := timestampDecompressor.NextValue()
	// 	if done {
	// 		break
	// 	}

	// 	value, done := valueDecompressor.NextValue()
	// 	if done {
	// 		qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug)
	// 	}

	// 	if timestamp <= req.Until {
	// 		req.ResponseWriter.FeedNoSend(timestamp, value)
	// 		if timestamp < req.Since {
	// 			req.ResultCh <- rangeScanResult{
	// 				ResultCode: QueryDone,
	// 			}
	// 			return
	// 		}
	// 	}
	// }

	// if s.LastPageNo > 0 {
	// 	req.ResultCh <- rangeScanResult{
	// 		ResultCode: UntilFound,
	// 		LastPageNo: s.LastPageNo,
	// 		FracDigits: s.FracDigits,
	// 	}
	// 	s.RLocks++
	// } else {
	// 	req.ResultCh <- rangeScanResult{
	// 		ResultCode: QueryDone,
	// 	}
	// }
}

func (s *_metric) StartFullScan(req tryFullScanReq) {
	// if s.Since == 0 {
	// 	req.ResultCh <- fullScanResult{
	// 		ResultCode: QueryDone,
	// 	}
	// 	return
	// }

	// timestampDecompressor := s.Timestamps.CreateDecompressor()
	// valueDecompressor := s.Values.CreateDecompressor()

	// for {
	// 	timestamp, done := timestampDecompressor.NextValue()
	// 	if done {
	// 		break
	// 	}
	// 	value, done := valueDecompressor.NextValue()
	// 	if done {
	// 		qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug)
	// 	}
	// 	req.ResponseWriter.FeedNoSend(timestamp, value)
	// }

	// if s.LastPageNo > 0 {
	// 	req.ResultCh <- fullScanResult{
	// 		ResultCode: UntilFound,
	// 		LastPageNo: s.LastPageNo,
	// 		FracDigits: s.FracDigits,
	// 	}
	// 	s.RLocks++
	// } else {
	// 	req.ResultCh <- fullScanResult{
	// 		ResultCode: QueryDone,
	// 	}
	// }
}

// індекси
// Metric encode format:
// metricID - 4b
// metricType - 1b
// fracDigits - 1b
// lastPageNo - 4b
// timestamps size - 2b
// values size - 2b
// timestams payload - Nb
// values payload - Nb
// index levels count - varsize
// [
//    records qty - varsize
//    records - Nb
// ]

func (s *_metric) WriteTo(w io.Writer) (err error) {
	_, err = w.Write([]byte{
		byte(s.metricType),
		s.fracDigits,
	})
	if err != nil {
		return
	}
	err = bin.WriteUint32(w, s.lastPageNo)
	if err != nil {
		return
	}
	err = bin.WriteUint16(w, uint16(s.timestamps.Size()))
	if err != nil {
		return
	}
	err = bin.WriteUint16(w, uint16(s.values.Size()))
	if err != nil {
		return
	}
	var (
		timestampsCapturedState *qb.TimeDeltaCapturedState
		valuesCapturedState     *qb.ValueDeltaCapturedState
	)
	if s.capturedState != nil {
		timestampsCapturedState = &s.capturedState.TimeState
		valuesCapturedState = &s.capturedState.ValueState
	}
	// timestamps payload
	err = s.timestamps.WritePayloadTo(w, timestampsCapturedState)
	if err != nil {
		return
	}
	// values payload
	err = s.values.WritePayloadTo(w, valuesCapturedState)
	if err != nil {
		return
	}
	// indexes
	_, err = bin.WriteVarSize(w, len(s.indexLevelTails))
	if err != nil {
		return
	}
	for _, level := range s.indexLevelTails {
		_, err = bin.WriteVarSize(w, level.RecordsCount)
		if err != nil {
			return
		}
		_, err = w.Write(level.Buffer)
		if err != nil {
			return
		}
	}
	return
}

// since - 4b
// sinceValue - 8b -
// untilValue - 8b

// time

// if s.state != nil {
// 		h = s.state.H
// 		lastUnixtime = s.state.LastUnixtime
// 		lastDelta = s.state.LastDelta
// 		payload = s.state.Payload
// 	}
// return NewTimeDeltaDecompressorFromState(TimeDeltaDecompressorFromStateOptions{
// 	H:            h,
// 	LastUnixtime: lastUnixtime,
// 	LastDelta:    lastDelta,
// 	Payload:      payload,
// })

// value
// if s.state != nil {
// 	h = s.state.H
// 	lastDelta = s.state.LastDelta
// 	payload = s.state.Payload
// }
// return NewValueDeltaDecompressorFromState(ValueDeltaDecompressorFromStateOptions{
// 	Coef:      s.coef,
// 	ToFloat64: s.toFloat64,
// 	H:         s.buf[s.pos-1],
// 	LastDelta: s.lastDelta,
// 	Payload:   s.buf[:s.pos],
// })