qb/enc/time_delta.go

package enc

import (
	"io"
	"log"

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

/*
Payload data файла має таку структуру:
vvvvvvvv->   <-ttttttt
де v - це values, які додаються зліва направо, а читаються справа наліво;
t - це timestamps, які додаються справа наліво, а читаються зліва направо.

Timestamps читаються звичайними методами Get*, а писатись мають типу ByRightBound

*/

const (
	tmpTimeSize = 7 // max when start run
)

type TimeDeltaCompressor struct {
	buf []byte
	// останній записаний байт, якщо рахувати справа наліво
	pos          int
	lastUnixtime uint32
	lastDelta    uint32
	state        *TimeDeltaCapturedState
}

// Початкове створення, коли даних немає
// Створення після page filled, коли одразу є початкове значення
// Відновлення стану із снапшота. LastTimestamp передаю окермо від payload,
// а lastDelta і h треба прочитати із payload (якщо є)

// Кодує значення справа наліво функціями TailPut*. Читає значення зліва направо функціями Get*
func NewTimeDeltaCompressor(buf []byte, payloadSize int) *TimeDeltaCompressor {
	s := &TimeDeltaCompressor{
		buf: buf,
		pos: len(buf) - payloadSize,
	}
	if payloadSize > 0 {
		var err error
		s.lastUnixtime, err = bin.GetUint32(s.buf[len(s.buf)-4:])
		if err != nil {
			log.Fatalf("bug: get since: %s", err)
		}
		//s.lastUnixtime = lastTimestamp // fix - порахувати

		if payloadSize > 4 {
			// u64, _, err := bin.GetVarUint64(s.buf[s.pos+1:])
			// if err != nil {
			// 	log.Fatalf("bug: get last delta: %s", err)
			// }
			// s.lastDelta = uint32(u64)
		}
	}
	return s
}

// FIX - restire lastUnixtime
// func (s *TimeDeltaCompressor) RestoreState() {

// }

func (s *TimeDeltaCompressor) Evaluate(tmp []byte, timestamp uint32) qb.TimeEvaluationReport {
	var (
		delta  = timestamp - s.lastUnixtime
		offset int
		i      int
	)
	if s.pos < len(s.buf) {
		if s.lastDelta > 0 {
			h := s.buf[s.pos]
			if h < 128 {
				// run
				if delta == s.lastDelta && h < 127 {
					// incrementRun
					tmp[i] = h + 1
					i++
					offset = 1 // перезапис h
				} else {
					// endSeries
					n, _ := bin.PutVarUint64(tmp, uint64(delta))
					i += n
					tmp[i] = 128 // start new literal (length=1)
					i++
				}
			} else {
				// literal
				if delta != s.lastDelta {
					if h < 255 {
						// incrementLiteral
						n, _ := bin.PutVarUint64(tmp, uint64(delta))
						i += n
						tmp[i] = h + 1
						i++
						offset = 1 // перезапис h
					} else {
						// endSeries
						n, _ := bin.PutVarUint64(tmp, uint64(delta))
						i += n
						tmp[i] = 128 // start new literal (length=1)
						i++
					}
				} else {
					// startRun
					if h > 128 {
						tmp[i] = h - 1 // зменшую довжину попередньої серії на 1
						i++
						n, _ := bin.PutVarUint64(tmp[i:], uint64(delta))
						i += n
						tmp[i] = 0 // start new run (length=2)
						i++
						offset = 1 + n // перезапис пари delta/h
					} else {
						tmp[i] = 0 // change literal (length=1) to run (length=2)
						i++
						offset = 1
					}
				}
			}
		} else {
			// add 1st delta
			n, _ := bin.PutVarUint64(tmp, uint64(delta))
			i += n
			tmp[i] = 128 // start new literal (length=1)
			i++
		}
	} else {
		bin.PutUint32(tmp, timestamp) // 1st timestamp (since)
		i += 4
	}
	return qb.TimeEvaluationReport{
		Offset:     offset,
		ChangeSize: i,
		TotalSpace: s.pos - offset + i,
	}
}

func (s *TimeDeltaCompressor) Append(offset int, change []byte, timestamp uint32) {
	if s.pos < len(s.buf) {
		i := s.pos + offset - 1 // -1, because s.pos always points to h byte
		for _, b := range change {
			s.buf[i] = b
			i--
		}
		s.lastDelta = timestamp - s.lastUnixtime
	} else {
		copy(s.buf[len(s.buf)-4:], change) // 4b since
	}
	s.pos -= len(change) - offset
	s.lastUnixtime = timestamp
}

// func (s *TimeDeltaCompressor) DeleteLast() {

// }

type TimeDeltaCapturedState struct {
	H            byte
	LastUnixtime uint32
	LastDelta    uint32
	Payload      []byte
}

// delta h
func (s *TimeDeltaCompressor) CaptureState() {
	if s.state != nil {
		qb.Abort(qb.RepeatableLock, nil)
	}
	// позиція посувається вліво, отже може перескочити на попередній chunk
	s.state = &TimeDeltaCapturedState{
		H:            s.buf[s.pos],
		LastUnixtime: s.lastUnixtime,
		LastDelta:    s.lastDelta,
		Payload:      s.buf[s.pos:],
	}
}

// fix - повернути в Pool буфери
func (s *TimeDeltaCompressor) ForgetCapturedState() {
	s.state = nil
}

// коли сторінка заповнена і відправляється в txlog, since замінюю на until
func (s *TimeDeltaCompressor) ReplaceSinceWithUntil() uint32 {
	pos := len(s.buf) - 4
	since, _ := bin.GetUint32(s.buf[pos:])
	bin.PutUint32(s.buf[pos:], s.lastUnixtime)
	return since
}

// для зростання буфера під час вставки даних. State не цікавить
func (s *TimeDeltaCompressor) Size() int {
	return len(s.buf) - s.pos
}

// 	if s.state == nil {
// 		return len(s.buf) - s.pos
// 	} else {
// 		return bin.CountVarUint64(uint64(s.state.LastDelta)) + hSize + len(s.state.Payload)
// 	}
// }

// Snapshot - для створення снапшота.
// FIX - encode firstUnixtime in 1st 4 bytes
//
//	while page not completed - has first unixtime, after - last unixtime
// func (s *TimeDeltaCompressor) Payload() []byte {
// 	if s.state == nil {
// 		return s.buf[s.pos:]
// 	} else {
// 		return s.state.Payload
// 	}
// }

func (s *TimeDeltaCompressor) WritePayloadTo(w io.Writer) (err error) {
	if s.state == nil {
		_, err = w.Write(s.buf[:s.pos])
		return
	} else {
		_, err = w.Write(s.state.Payload)
		if err != nil {
			return
		}
		_, err = bin.WriteVarUint64(w, uint64(s.state.LastDelta))
		if err != nil {
			return
		}
		_, err = w.Write([]byte{
			s.state.H,
		})
		return
	}
}

// func (s *TimeDeltaCompressor) encodeTail(lastUnixtime, lastDelta uint32, h byte) []byte {
// 	tail := make([]byte, 9)
// 	bin.PutUint32(tail, lastUnixtime)
// 	tail[4] = h
// 	n, _ := bin.PutVarUint64(tail[5:], uint64(lastDelta))
// 	return tail[:n+5]
// }

func (s *TimeDeltaCompressor) Rotate(newbuf []byte) {
	// УВАГА!
	// state не чіпаємо
	s.buf = newbuf
	s.pos = len(s.buf)
	s.lastUnixtime = 0
	s.lastDelta = 0
}

func (s *TimeDeltaCompressor) CreateDecompressor() qb.TimestampDecompressor {
	if s.pos == len(s.buf) {
		return nil
	}
	var (
		h            = s.buf[s.pos]
		lastUnixtime = s.lastUnixtime
		lastDelta    = s.lastDelta
		payload      = s.buf[s.pos:]
	)
	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,
	})
}

func (s *TimeDeltaCompressor) LastTimestamp() uint32 {
	return s.lastUnixtime
}

// DECOMPRESSOR

type TimeDeltaDecompressor struct {
	buf          []byte
	pos          int
	lastDelta    uint32
	lastUnixtime uint32
	isRun        bool
	pending      int
	done         bool
}

func NewTimeDeltaDecompressor(buf []byte) *TimeDeltaDecompressor {
	s := &TimeDeltaDecompressor{
		buf: buf,
	}
	if s.pos < len(s.buf) {
		var err error
		s.lastUnixtime, err = bin.GetUint32(s.buf)
		if err != nil {
			log.Fatalf("bug: get last unixtime: %s", err)
		}
		s.pos += 4
		//fmt.Println("restored last unixtime", s.lastUnixtime)
	} else {
		s.done = true
	}
	return s
}

type TimeDeltaDecompressorFromStateOptions struct {
	H            byte
	LastUnixtime uint32
	LastDelta    uint32
	Payload      []byte
}

func NewTimeDeltaDecompressorFromState(opt TimeDeltaDecompressorFromStateOptions) *TimeDeltaDecompressor {
	s := &TimeDeltaDecompressor{
		buf:          opt.Payload,
		lastUnixtime: opt.LastUnixtime,
	}
	if opt.LastDelta > 0 {
		s.lastDelta = opt.LastDelta
		s.decodeHeaderByte(opt.H)
	}
	//fmt.Println("-------------------")
	//fmt.Printf("restore from bound: isRun=%t, pending=%d, lastUnix=%d, lastDelta=%d\n",
	//s.isRun, s.pending, s.lastUnixtime, s.lastDelta)
	//fmt.Printf("buf: %d\n", s.buf)
	return s
}

func (s *TimeDeltaDecompressor) NextValue() (value uint32, done bool) {
	//fmt.Printf("NextValue(): pos: %d, isRegular: %t, pending: %d\n", s.pos, s.isRegular, s.pending)
	if s.done {
		return 0, true
	}
	// повертаю значення, що було прочитано в методі RestoreFromBound/RestoreFromEnd
	value = s.lastUnixtime
	if s.lastDelta > 0 {
		s.lastUnixtime -= s.lastDelta
		s.pending--
		//fmt.Printf("lastUnix: %d, s.pending: %d\n", s.lastUnixtime, s.pending)
		if s.pending > 0 {
			// якщо в серії залишаються елементи
			if !s.isRun {
				s.readDelta()
			}
		} else if s.pos < len(s.buf) {
			// в серії більше немає елементів, отже перевіряє чи є ще дані в буфері.
			// дані є - читаю заголовок наступної серії
			s.readHeader()
			s.readDelta()
		} else {
			//s.done = true
			s.lastDelta = 0
		}
	} else {
		// був закодований лише last unixtime
		s.done = true
	}
	return value, false
}

func (s *TimeDeltaDecompressor) readHeader() {
	h := s.buf[s.pos]
	s.pos++
	s.decodeHeaderByte(h)
	// fmt.Println()
	// fmt.Println("read from pos:", s.pos)
	// fmt.Println("h:", h)
	// fmt.Println("isRun:", s.isRun)
	// fmt.Println("pending:", s.pending)
	// fmt.Println()
}

func (s *TimeDeltaDecompressor) decodeHeaderByte(h byte) {
	s.isRun = h < 128
	if s.isRun {
		s.pending = int(h) + 2
	} else {
		s.pending = int(h&127) + 1
	}
}

func (s *TimeDeltaDecompressor) readDelta() {
	u64, n, err := bin.GetVarUint64(s.buf[s.pos:])
	if err != nil {
		log.Fatalln(err)
	}
	// fmt.Println()
	// fmt.Println("read from pos:", s.pos)
	// fmt.Println("read delta:", u64)
	// fmt.Println("read delta n:", n)
	// fmt.Println()
	s.pos += n
	s.lastDelta = uint32(u64)
}