qb/chunkenc/time_delta.go

package chunkenc

import (
	"fmt"
	"log"

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

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

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

*/

type ReverseTimeDeltaCompressor struct {
	buf           []byte
	pos           int
	lastUnixtime  uint32
	lastDelta     uint32
	lastDeltaSize int
	h             byte
	state         *TimeDeltaBound
}

func NewReverseTimeDeltaCompressor(buf []byte, size int) *ReverseTimeDeltaCompressor {
	s := &ReverseTimeDeltaCompressor{
		buf: buf,
		pos: size, // перший вільний байт
	}
	if size > 0 {
		u64, n, err := bin.ReverseGetVarUint64(s.buf[:s.pos-1])
		if err != nil {
			log.Fatalf("bug: get last unixtime: %s", err)
		}
		s.lastUnixtime = uint32(u64)
		s.pos -= n
		if s.pos > 0 {
			s.h = s.buf[s.pos-1]
			u64, s.lastDeltaSize, err = bin.ReverseGetVarUint64(s.buf[:s.pos-2])
			if err != nil {
				log.Fatalf("bug: get last delta: %s", err)
			}
			s.lastDelta = uint32(u64)
		}
	}
	return s
}

func (s *ReverseTimeDeltaCompressor) Size() int {
	return s.pos
}

func (s *ReverseTimeDeltaCompressor) CalcRequiredSpace(value uint32) int {
	return 0
}

func (s *ReverseTimeDeltaCompressor) Append(unixtime uint32) {
	if s.lastUnixtime > 0 {
		// base value
		//s.pos += s.buf.PutVarUint64(s.pos, uint64(value*s.coef))
		//s.baseValue = value
		//s.appendNewLiteral(0)

		//} else {
		delta := unixtime - s.lastUnixtime
		if delta == s.lastDelta {
			if s.h < 128 {
				// run блок - отже треба збільшити лічильник
				if s.h < 127 {
					// increase counter
					s.h++
					s.buf[s.pos-1] = s.h
				} else {
					// не можу збільшити - буде переповнення. Додаю новий literal блок
					// counter overflow
					s.appendNewLiteral(delta)
				}
			} else {
				// literal блок.
				// Якщо в ньому лише одне значення - перетворюю його на run блок.
				// Інакше забираю останнє значення із literal блока і додаю новий run блок.
				q := s.h & 127
				if q == 0 { // 1 кодується як 0
					s.convertLiteralToRun()
				} else {
					// забираю останнє значення із literal блоку щоб зробити run блок
					s.convertLastFromLiteralToRun()
				}
			}
		} else {
			if s.h < 127 { // fix 128?
				// end of run
				s.appendNewLiteral(delta)
			} else {
				if s.h < 255 {
					// encode value from pos - 1, then append h byte
					s.appendDeltaToLiteral(delta)
				} else {
					// overflowed - encode new
					s.appendNewLiteral(delta)
				}
			}
		}
	}
	s.lastUnixtime = unixtime
}

func (s *ReverseTimeDeltaCompressor) convertLastFromLiteralToRun() {
	// Зменшую кількість елементів в literal блоці
	s.h--
	s.pos -= 1 + s.lastDeltaSize
	s.buf[s.pos] = s.h // закриваю literal блок
	s.pos++
	s.lastDeltaSize, _ = bin.ReversePutVarUint64(s.buf[s.pos:], uint64(s.lastDelta))
	s.pos += s.lastDeltaSize
	s.h = 0 // run блок, довжини 2
	s.buf[s.pos] = s.h
	s.pos++
}

func (s *ReverseTimeDeltaCompressor) convertLiteralToRun() {
	// Знімаю flagLiteral, а лічильник 0 дорівнює 2 елементам в серії.
	s.h = 0
	s.buf[s.pos-1] = s.h
}

func (s *ReverseTimeDeltaCompressor) appendDeltaToLiteral(delta uint32) {
	s.h++ // збільшую к-сть дельт
	s.lastDelta = delta
	s.pos--
	s.lastDeltaSize, _ = bin.ReversePutVarUint64(s.buf[s.pos:], uint64(delta))
	s.pos += s.lastDeltaSize
	s.buf[s.pos] = s.h
	s.pos++
}

func (s *ReverseTimeDeltaCompressor) appendNewLiteral(delta uint32) {
	//fmt.Println("appendNewLiteral", delta)
	s.h = flagLiteral
	s.lastDelta = delta
	s.lastDeltaSize, _ = bin.ReversePutVarUint64(s.buf[s.pos:], uint64(delta))
	s.pos += s.lastDeltaSize
	s.buf[s.pos] = flagLiteral // literal, length = 1
	s.pos++
}

func (s *ReverseTimeDeltaCompressor) DeleteLast() {

}

func (s *ReverseTimeDeltaCompressor) Sync() {
	n, _ := bin.ReversePutVarUint64(s.buf[s.pos:], uint64(s.lastUnixtime))
	s.pos += n
}

// FIX - check methods

type TimeDeltaBound struct {
	Pos          int
	H            byte
	LastUnixtime uint32
	LastDelta    uint32
	Chunks       []byte
}

// delta h
// func (s *ReverseTimeDeltaCompressor) GetState() TimeDeltaBound { // fix replace by Lock
// 	bound := TimeDeltaBound{
// 		LastUnixtime: s.lastUnixtime,
// 	}
// 	if s.pos > 0 {
// 		bound.Pos = s.pos - 1 - s.lastDeltaSize
// 		bound.H = s.h
// 		bound.LastDelta = s.lastDelta
// 	}
// 	return bound
// }

// delta h
func (s *ReverseTimeDeltaCompressor) Lock() {
	if s.state != nil {
		qb.Abort(qb.RepeatableLock, nil)
	}
	// позиція посувається вліво, отже може перескочити на попередній chunk
	pos := s.pos - 1 - s.lastDeltaSize
	s.state = &TimeDeltaBound{
		Pos:          pos,
		H:            s.h,
		LastUnixtime: s.lastUnixtime, // fix
		LastDelta:    s.lastDelta,
		Chunks:       s.buf[:s.pos], // fix check pos?
	}
}

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

func (s *ReverseTimeDeltaCompressor) Offset() int {
	if s.state != nil {
		return s.state.Pos
	}
	return 0
}

// fix -
func (s *ReverseTimeDeltaCompressor) Snapshot() ([]byte, int) {
	if s.state == nil {
		return s.buf, s.Size()
	}
	// ВАЖЛИВО!
	// Треба відтворити стан останнього чанка
	var (
		pos          = s.state.Pos
		chunk        = make([]byte, conbuf.ChunkSize)
		lastChunkIdx = len(s.state.Chunks) - 1
		qtyToCopy    = pos % conbuf.ChunkSize
	)
	copy(chunk, s.state.Chunks[lastChunkIdx][:qtyToCopy])
	chunks := append(s.state.Chunks[:lastChunkIdx], chunk)

	buf := conbuf.New(chunks)
	pos += buf.ReversePutVarUint64(pos, uint64(s.state.LastDelta))
	buf.SetByte(pos, s.state.H)
	pos++
	return chunks, pos
}

func (s *ReverseTimeDeltaCompressor) CreateDecompressor() qb.TimestampDecompressor {
	if s.state == nil {
		d := NewReverseTimeDeltaDecompressor(s.buf, s.Size())
		d.RestoreFromEnd()
		return d
	}
	d := NewReverseTimeDeltaDecompressor(s.buf, s.Size())
	d.RestoreFromBound(*s.state)
	return d
}

func (s *ReverseTimeDeltaCompressor) Renew() {
	// УВАГА!
	// state не чіпаємо
	s.buf = make([]byte, minBufferSize)
	s.pos = 0
	//
	//s.baseValue = 0
	s.lastDelta = 0
	s.lastDeltaSize = 0
	s.h = 0
}

func (s *ReverseTimeDeltaCompressor) Chunks() []byte {
	return s.buf
}

// DECOMPRESSOR

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

func NewReverseTimeDeltaDecompressor(buf []byte, size int) *ReverseTimeDeltaDecompressor {
	return &ReverseTimeDeltaDecompressor{
		buf: buf,
		pos: size,
	}
}

func (s *ReverseTimeDeltaDecompressor) RestoreFromEnd() {
	fmt.Println("RestoreFromEnd", s.pos)
	if s.pos > 0 {
		s.pos-- // перший байт даних
		u64, n, err := bin.ReverseGetVarUint64(s.buf[:s.pos])
		if err != nil {
			log.Fatalf("bug: get last unixtime: %s", err)
		}
		s.lastUnixtime = uint32(u64)
		s.pos -= n
		fmt.Println("restored", s.lastUnixtime, s.pos)
		if s.pos > 0 {
			s.readHeader()
			s.readDelta()
		}
	} else {
		s.done = true
	}
}

func (s *ReverseTimeDeltaDecompressor) RestoreFromBound(bound TimeDeltaBound) {
	pretty.Println(bound)
	s.lastUnixtime = bound.LastUnixtime
	if bound.LastDelta > 0 {
		s.pos = bound.Pos - 1
		s.lastDelta = bound.LastDelta
		s.decodeHeaderByte(bound.H)
	} else {
		s.pos = -1
	}
	fmt.Printf("restore from bound: pos=%d, isRun=%t, pending=%d\n", s.pos, s.isRun, s.pending)
}

func (s *ReverseTimeDeltaDecompressor) 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.isRegular {
		s.pending--
		if s.pending > 0 {
			// якщо в серії залишаються елементи
			if !s.isRun {
				s.readDelta()
			}
			s.lastUnixtime -= s.lastDelta
		} else if s.pos > 0 {
			// в серії більше немає елементів, отже перевіряє чи є ще дані в буфері.
			// дані є - читаю заголовок наступної серії
			s.readHeader()
			s.readDelta()
			s.lastUnixtime -= s.lastDelta
		} else {
			s.done = true
		}
	} else {
		if s.lastDelta > 0 {
			s.isRegular = true
			s.lastUnixtime -= s.lastDelta
		} else {
			// був закодований лише last unixtime
			s.done = true
		}
	}
	return value, false
}

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

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

func (s *ReverseTimeDeltaDecompressor) readDelta() {
	u64, n, err := bin.ReverseGetVarUint64(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)
}