qb/chunkenc/insdelta.go

package chunkenc

import (
	"fmt"
	"math"

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

// REVERSE

type ReverseInstantDeltaCompressor struct {
	buf        *conbuf.ContinuousBuffer
	coef       float64
	pos        int
	firstValue float64
	lastDelta  int64
	length     uint16
	numIdx     int
}

func NewReverseInstantDeltaCompressor(buf *conbuf.ContinuousBuffer, size int, fracDigits byte) *ReverseInstantDeltaCompressor {
	var coef float64 = 1
	if fracDigits > 0 {
		coef = math.Pow(10, float64(fracDigits))
	}
	s := &ReverseInstantDeltaCompressor{
		buf:  buf,
		pos:  size,
		coef: coef,
	}
	if size > 0 {
		s.restoreState()
	}
	return s
}

func (s *ReverseInstantDeltaCompressor) restoreState() {
	i64, n, err := s.buf.GetVarInt64(0)
	if err != nil {
		panic(fmt.Sprintf("bug: get first value: %s", err))
	}
	s.firstValue = float64(i64) / s.coef

	if s.pos > n {
		pos := s.pos - 1
		idxOf8 := uint(8 - s.buf.GetByte(pos))
		pos--
		s8 := s.buf.GetByte(pos)
		pos--

		var n int
		s.lastDelta, n, err = s.buf.ReverseGetVarInt64(pos)
		if err != nil {
			panic(fmt.Sprintf("bug: get last delta: %s", err))
		}
		pos -= n
		s.numIdx = pos + 1
		var flag byte = 1 << idxOf8
		if (s8 & flag) == flag {
			s.length, _ = s.buf.DecodeRunLength(pos)
		}
	}
}

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

func (s *ReverseInstantDeltaCompressor) CalcRequiredSpace(value float64) int {
	if s.pos == 0 {
		n := bin.CalcVarInt64Length(int64(value * s.coef))
		return n + 3
	}

	tmp := (value - s.firstValue) * s.coef
	if tmp > 0 {
		tmp += eps
	} else {
		tmp -= eps
	}
	delta := int64(tmp)

	if delta == s.lastDelta {
		if s.length == 0 {
			return 1
		} else {
			newLength := s.length + 1
			if newLength < 130 {
				return 0
			} else if newLength == 130 {
				return 1
			} else {
				if newLength < 32769 {
					return 0
				} else {
					n := bin.CalcVarInt64Length(delta)
					n += 2
					s8q := s.buf.GetByte(s.pos - 1)
					if s8q == 8 {
						n -= 1
					} else {
						n -= 2
					}
					return n
				}
			}
		}
	} else {
		n := bin.CalcVarInt64Length(delta)
		n += 2
		s8q := s.buf.GetByte(s.pos - 1)
		if s8q == 8 {
			n -= 1
		} else {
			n -= 2
		}
		return n
	}
}

// В начале буфера кодирую базовое значение.
func (s *ReverseInstantDeltaCompressor) Append(value float64) {
	if s.pos == 0 {
		n := s.buf.PutVarInt64(s.pos, int64(value*s.coef))
		s.pos += n
		s.firstValue = value
		s.encodeNewDelta(0, 0, 1)
	} else {
		tmp := (value - s.firstValue) * s.coef
		if tmp > 0 {
			tmp += eps
		} else {
			tmp -= eps
		}
		delta := int64(tmp)

		if delta == s.lastDelta {
			if s.length == 0 {
				s.length = 2
				s.shiftOnePosToRight()
				s.buf.SetByte(s.numIdx-1, 0)
				s8q := s.buf.GetByte(s.pos - 1)
				s.buf.SetFlag(s.pos-2, 1<<(8-s8q))
			} else {
				s.length++
				if s.length < 130 {
					s.buf.SetByte(s.numIdx-1, byte(s.length-2))
				} else if s.length == 130 {
					s.shiftOnePosToRight()
					s.encode2bLength()
				} else {
					if s.length < 32769 {
						s.encode2bLength()
					} else {
						s.appendNewDelta(delta)
					}
				}
			}
		} else {
			s.appendNewDelta(delta)
		}
	}
}

func (s *ReverseInstantDeltaCompressor) appendNewDelta(delta int64) {
	s.length = 0

	s8 := s.buf.GetByte(s.pos - 2)
	s8q := s.buf.GetByte(s.pos - 1)

	if s8q == 8 {
		s.pos -= 1
		s8 = 0
		s8q = 1
	} else {
		s.pos -= 2
		s8q++
	}
	s.encodeNewDelta(delta, s8, s8q)
}

func (s *ReverseInstantDeltaCompressor) encodeNewDelta(delta int64, s8 byte, s8q byte) {
	s.lastDelta = delta
	s.numIdx = s.pos
	n := s.buf.ReversePutVarInt64(s.pos, s.lastDelta)
	s.pos += n
	s.buf.SetByte(s.pos, s8)
	s.pos++
	s.buf.SetByte(s.pos, s8q)
	s.pos++
}

func (s *ReverseInstantDeltaCompressor) shiftOnePosToRight() {
	s.buf.ShiftOnePosToRight(s.numIdx, s.pos)
	s.pos++
	s.numIdx++
}

func (s *ReverseInstantDeltaCompressor) encode2bLength() {
	num := s.length - 2
	s.buf.SetByte(s.numIdx-1, byte(num&127)|128)
	s.buf.SetByte(s.numIdx-2, byte(num>>7))
}

func (s *ReverseInstantDeltaCompressor) DeleteLast() {
	var (
		s8q       = s.buf.GetByte(s.pos - 1)
		s8        = s.buf.GetByte(s.pos - 2)
		flag byte = 1 << uint(8-s8q)
	)

	if s.length > 0 {
		if s.length == 2 {
			s.length = 0
			s.buf.UnsetFlag(s.pos-2, flag)
			s.buf.ShiftOnePosToLeft(s.numIdx, s.pos)
			s.numIdx--
			s.pos--
		} else if s.length < 130 {
			s.length--
			s.buf.SetByte(s.numIdx-1, byte(s.length)-2)
		} else if s.length == 130 {
			s.length--
			s.buf.ShiftOnePosToLeft(s.numIdx, s.pos)
			s.numIdx--
			s.pos--
			s.buf.SetByte(s.numIdx-1, byte(s.length)-2)
		} else {
			s.length--
			s.encode2bLength()
		}
	} else {
		if s8q > 1 {
			s8q--
			flag = 1 << uint(8-s8q)
			s.pos = s.numIdx + 2
			s.buf.SetByte(s.pos-2, s8)
			s.buf.SetByte(s.pos-1, s8q)
		} else {
			s.pos = s.numIdx + 1
			s.buf.SetByte(s.pos-1, 8)
			s8 = s.buf.GetByte(s.pos - 2)
			flag = 1
		}
		var (
			pos = s.pos - 3
			n   int
			err error
		)
		s.lastDelta, n, err = s.buf.ReverseGetVarInt64(pos)
		if err != nil {
			panic(err)
		}
		s.numIdx = pos - n
		if (s8 & flag) == flag {
			s.length, _ = s.buf.DecodeRunLength(s.numIdx - 1)
		}
	}
}

type ReverseInstantDeltaDecompressor struct {
	step       byte
	buf        *conbuf.ContinuousBuffer
	pos        int
	bound      int
	firstValue float64
	lastValue  float64
	length     uint16
	coef       float64
	idxOf8     uint
	s8         byte
}

func NewReverseInstantDeltaDecompressor(buf *conbuf.ContinuousBuffer, size int, fracDigits byte) *ReverseInstantDeltaDecompressor {
	var coef float64 = 1
	if fracDigits > 0 {
		coef = math.Pow(10, float64(fracDigits))
	}
	return &ReverseInstantDeltaDecompressor{
		buf:  buf,
		coef: coef,
		pos:  size,
	}
}

func (s *ReverseInstantDeltaDecompressor) NextValue() (value float64, done bool) {
	if s.step > 0 {
		if s.length > 0 {
			s.length--
			return s.lastValue, false
		}
		if s.pos < s.bound {
			return 0, true
		}

		if s.idxOf8 == 0 {
			s.s8 = s.buf.GetByte(s.pos)
			s.pos--
		}
		s.readVar()
		if s.length > 0 {
			s.length--
		}
		return s.lastValue, false
	}
	i64, n, err := s.buf.GetVarInt64(0)
	if err != nil {
		panic(err)
	}
	s.firstValue = float64(i64) / s.coef
	s.bound = n
	s.pos--
	s.idxOf8 = uint(8 - s.buf.GetByte(s.pos))
	s.pos--
	s.s8 = s.buf.GetByte(s.pos)
	s.pos--
	s.readVar()
	if s.length > 0 {
		s.length--
	}
	s.step = 1
	return s.lastValue, false
}

func (s *ReverseInstantDeltaDecompressor) readVar() {
	i64, n, err := s.buf.ReverseGetVarInt64(s.pos)
	if err != nil {
		panic(err)
	}
	s.pos -= n
	s.lastValue = s.firstValue + float64(i64)/s.coef

	var flag byte = 1 << s.idxOf8
	if (s.s8 & flag) == flag {
		s.length, n = s.buf.DecodeRunLength(s.pos)
		s.pos -= n
	}
	if s.idxOf8 == 7 {
		s.idxOf8 = 0
	} else {
		s.idxOf8++
	}
}