diff --git a/atree/cursor.go b/atree/cursor.go index 84f4960..083c96b 100644 --- a/atree/cursor.go +++ b/atree/cursor.go @@ -130,7 +130,7 @@ func (s *BackwardCursor) makeDecompressors() error { s.pageNo, payloadSize) } - s.timestampDecompressor = enc.NewReverseTimeDeltaOfDeltaDecompressor( + s.timestampDecompressor = enc.NewTimeDeltaDecompressor( s.pageData[:timestampsPayloadSize], ) @@ -138,11 +138,11 @@ func (s *BackwardCursor) makeDecompressors() error { switch s.metricType { case qb.Instant: - s.valueDecompressor = enc.NewReverseInstantDeltaDecompressor( + s.valueDecompressor = enc.NewInstantDeltaDecompressor( vbuf, s.fracDigits) case qb.Cumulative: - s.valueDecompressor = enc.NewReverseCumulativeDeltaDecompressor( + s.valueDecompressor = enc.NewCumulativeDeltaDecompressor( vbuf, s.fracDigits) default: @@ -164,7 +164,7 @@ func makeDecompressors(pageData []byte, metricType qb.MetricType, fracDigits byt payloadSize) } - timestampDecompressor := enc.NewReverseTimeDeltaOfDeltaDecompressor( + timestampDecompressor := enc.NewTimeDeltaDecompressor( pageData[:timestampsPayloadSize], ) @@ -173,11 +173,11 @@ func makeDecompressors(pageData []byte, metricType qb.MetricType, fracDigits byt var valueDecompressor qb.ValueDecompressor switch metricType { case qb.Instant: - valueDecompressor = enc.NewReverseInstantDeltaDecompressor( + valueDecompressor = enc.NewInstantDeltaDecompressor( vbuf, fracDigits) case qb.Cumulative: - valueDecompressor = enc.NewReverseCumulativeDeltaDecompressor( + valueDecompressor = enc.NewCumulativeDeltaDecompressor( vbuf, fracDigits) default: diff --git a/chunkenc/chunkenc.go b/chunkenc/chunkenc.go deleted file mode 100644 index d6192d8..0000000 --- a/chunkenc/chunkenc.go +++ /dev/null @@ -1,8 +0,0 @@ -package chunkenc - -const eps = 0.000001 - -const ( - flagLiteral = 128 - minBufferSize = 1024 -) diff --git a/chunkenc/chunkenc_test.go b/chunkenc/chunkenc_test.go deleted file mode 100644 index a5b1c0d..0000000 --- a/chunkenc/chunkenc_test.go +++ /dev/null @@ -1,634 +0,0 @@ -package chunkenc - -import ( - "bytes" - "fmt" - "slices" - "testing" -) - -func TestCumdelta(t *testing.T) { - var ( - fracDigits byte = 0 - buf = make([]byte, minBufferSize) - value float64 - done bool - ) - c := NewCumulativeDeltaCompressor(buf, 0, fracDigits) - // c.Append(1.55) - // c.Append(23) - // c.Append(23) - // c.Append(23) - // c.Append(23.5) - - c.Append(130) - c.Append(191) - - c.Append(248) - c.Append(305) - - //fmt.Printf("pos: %d\n", c.pos) - //fmt.Printf("%d\n", buf.Chunks()[0]) - - d := NewCumulativeDeltaDecompressor(buf, c.Size(), fracDigits) - - for range 8 { - value, done = d.NextValue() - fmt.Println(value, done) - } -} - -func TestInsdelta(t *testing.T) { - var ( - fracDigits byte = 2 - buf = make([]byte, minBufferSize) - value float64 - done bool - ) - c := NewInstantDeltaCompressor(buf, 0, fracDigits) - c.Append(-1.55) - c.Append(23) - - //fmt.Printf("pos: %d\n", c.pos) - //fmt.Printf("%d\n", buf.Chunks()[0]) - - c.Append(23) - c.Append(23) - - c.Append(-23.5) - - //fmt.Printf("pos: %d\n", c.pos) - //fmt.Printf("%d\n", buf.Chunks()[0]) - - d := NewInstantDeltaDecompressor(buf, c.Size(), fracDigits) - - for range 8 { - value, done = d.NextValue() - fmt.Println(value, done) - } -} - -func TestTimeDeltaCompressor(t *testing.T) { - var ( - testCases = []struct { - Nums []uint32 - RepeatLastDeltaNTimes int - Code int - RequiredSize int - Buf []byte - LastUnixtime uint32 - LastDelta uint32 - H byte - }{ - { - Nums: []uint32{ - 1780777000, - }, - Code: addUnixtime, - RequiredSize: 4, - Buf: []byte{ - 0, 0, 0, 0, 0, 0, 0, 0, - }, - LastUnixtime: 1780777000, - LastDelta: 0, - H: 0, - }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - }, - Code: add1stDelta, - RequiredSize: 6, - Buf: []byte{ - 0, 0, 0, 0, 0, 0, 0, 0, - }, - LastUnixtime: 1780777060, - LastDelta: 60, - H: 128, - }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - }, - Code: startRun, // literal changed by run - RequiredSize: 6, - Buf: []byte{ - 0, 0, 0, 0, 0, 0, 0, 0, - }, - LastUnixtime: 1780777120, - LastDelta: 60, - H: 0, // run, length = 2 - }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777130, // +70 - 1780777200, // +70 - }, - Code: startRun, // literal decreased by 1 - RequiredSize: 7, // unixtime, end of literal, new delta, new h - Buf: []byte{ - 0, 0, 0, 0, 0, 0, 128, 188, - }, - LastUnixtime: 1780777200, - LastDelta: 70, - H: 0, // run, length = 2 - }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - 1780777180, // +60 - }, - Code: incrementRun, - RequiredSize: 5, - Buf: []byte{ - 0, 0, 0, 0, 0, 0, 0, 0, - }, - LastUnixtime: 1780777180, - LastDelta: 60, - H: 1, // run, length = 3 - }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - 1780777180, // +60 - 1780777200, // +20 - }, - Code: endSeries, // run - RequiredSize: 8, - Buf: []byte{ - 0, 0, 0, 0, 0, 0, - 1, // h of run, length = 3 - 188, // varUint64(60) - }, - LastUnixtime: 1780777200, - LastDelta: 20, - H: 128, // literal, length = 1 - }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - }, - RepeatLastDeltaNTimes: 128, - Code: incrementRun, // h byte full filled - RequiredSize: 5, - Buf: []byte{ - 0, 0, 0, 0, 0, 0, 0, 0, - }, - LastUnixtime: 1780777060 + 128*60, - LastDelta: 60, - H: 127, // run, length = 129 - }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - }, - RepeatLastDeltaNTimes: 129, - Code: endSeries, // run, h byte overflowed - RequiredSize: 8, - Buf: []byte{ - 0, 0, 0, 0, 0, 0, - 127, // h - end of run (length = 129) - 188, // varUint64(60) - }, - LastUnixtime: 1780777060 + 129*60, - LastDelta: 60, - H: 128, // literal, length = 1 - }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777122, // +62 - }, - Code: incrementLiteral, - RequiredSize: 7, - Buf: []byte{ - 0, 0, 0, 0, 0, 0, 0, - 188, // varUint64(60) - }, - LastUnixtime: 1780777122, - LastDelta: 62, - H: 129, // literal, length = 2 - }, - } - ) - - for caseIdx, testCase := range testCases { - //fmt.Println("----------") - var ( - buf = make([]byte, 8) - code int - requiredSize int - ) - c := NewTimeDeltaCompressor(buf, 0) - for _, num := range testCase.Nums { - code, requiredSize = c.CalcRequiredSpace(num) - c.Append(code, num) - } - if testCase.RepeatLastDeltaNTimes > 0 { - for range testCase.RepeatLastDeltaNTimes { - num := c.lastUnixtime + c.lastDelta - code, requiredSize = c.CalcRequiredSpace(num) - c.Append(code, num) - } - } - if code != testCase.Code { - t.Fatalf("%d: got code %d are not equal expected %d", - caseIdx, code, testCase.Code) - } - if requiredSize != testCase.RequiredSize { - t.Fatalf("%d: got requiredSize %d are not equal expected %d", - caseIdx, requiredSize, testCase.RequiredSize) - } - if !bytes.Equal(buf, testCase.Buf) { - t.Fatalf("%d: got buf %v are not equal expected %v", - caseIdx, buf, testCase.Buf) - } - if c.lastUnixtime != testCase.LastUnixtime { - t.Fatalf("%d: got lastUnixtime %d are not equal expected %d", - caseIdx, c.lastUnixtime, testCase.LastUnixtime) - } - if c.lastDelta != testCase.LastDelta { - t.Fatalf("%d: got lastDelta %d are not equal expected %d", - caseIdx, c.lastDelta, testCase.LastDelta) - } - if c.h != testCase.H { - t.Fatalf("%d: got h %d are not equal expected %d", - caseIdx, c.h, testCase.H) - } - } -} - -func TestTimeDeltaDecompressorFromCapturedState(t *testing.T) { - var ( - testCases = []struct { - Nums []uint32 - //RepeatLastDeltaNTimes int - }{ - { - Nums: []uint32{ - 1780777000, - }, - }, - { - // add1stDelta - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - }, - }, - { - // startRun - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - }, - }, - { - // incrementRun - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - 1780777180, // +60 - }, - }, - { - // endRun - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - 1780777180, // +60 - 1780777200, // +20 - }, - }, - { - // incrementLiteral - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777122, // +62 - }, - }, - { - // ... - Nums: []uint32{ - 1780777000, - 1780777010, // +10 - 1780777120, // +10 - 1780777130, // +10 - 1780777135, // +5 - 1780777141, // +6 - 1780777148, // +7 - 1780777156, // +8 - 1780777165, // +9 - 1780777174, // +9 - 1780777183, // +9 - 1780777190, // +7 - }, - }, - } - ) - for caseIdx, testCase := range testCases { - var ( - buf = make([]byte, 16) - decodedNums []uint32 - ) - c := NewTimeDeltaCompressor(buf, 0) - for _, num := range testCase.Nums { - code, _ := c.CalcRequiredSpace(num) - c.Append(code, num) - } - - // - c.CaptureState() - - d := c.CreateDecompressor() - - //fmt.Println(testCase.Nums) - // fmt.Println("---------------") - // fmt.Println(buf) - - for { - num, done := d.NextValue() - //fmt.Println(num, done) - if done { - break - } - decodedNums = append(decodedNums, num) - } - - slices.Reverse(decodedNums) - - if !slices.Equal(testCase.Nums, decodedNums) { - t.Fatalf("%d: got nums %v not equal expected %v", caseIdx, decodedNums, testCase.Nums) - } - } -} - -// func TestTimeDelta(t *testing.T) { -// var ( -// buf = make([]byte, minBufferSize) -// value uint32 -// done bool -// ) -// c := NewReverseTimeDeltaCompressor(buf, 0) -// c.Append(10) - -// //c.Append(131) - -// //fmt.Printf("pos: %d\n", c.pos) -// //fmt.Printf("%d\n", buf.Chunks()[0]) - -// c.Append(70) - -// c.Append(130) -// c.Append(191) - -// c.Append(248) -// c.Append(305) -// c.Append(330) -// c.Append(390) -// c.Append(450) - -// // fmt.Printf("pos: %d\n", c.pos) -// // fmt.Printf("%d\n", buf.Chunks()[0]) - -// //c.Sync() - -// // bound := c.GetState() - -// // fmt.Println("AFTER SYNC") -// // fmt.Printf("pos: %d\n", c.pos) -// // chunks := buf.Chunks() -// // if len(chunks) > 0 { -// // fmt.Printf("%d\n", chunks[0]) -// // } - -// d := NewReverseTimeDeltaDecompressor(buf, c.Size()) -// // d.RestoreFromBound(bound) -// //d.RestoreFromEnd() - -// for range 12 { -// value, done = d.NextValue() -// fmt.Println(value, done) -// } -// } - -func TestCumdeltaBound(t *testing.T) { - var ( - fracDigits byte = 0 - buf = make([]byte, minBufferSize) - value float64 - done bool - ) - c := NewCumulativeDeltaCompressor(buf, 0, fracDigits) - // c.Append(1.55) - // c.Append(23) - // c.Append(23) - // c.Append(23) - // c.Append(23.5) - c.Append(10) - c.Append(70) - - c.Append(130) - - // fmt.Printf("pos: %d\n", c.pos) - // fmt.Printf("%d\n", buf.Chunks()[0]) - - c.Append(191) - - // fmt.Printf("pos: %d\n", c.pos) - // fmt.Printf("%d\n", buf.Chunks()[0]) - - c.Append(191) - - fmt.Printf("pos: %d\n", c.pos) - fmt.Printf("%d\n", c.Chunks()) - ////////////////////////// - - c.Lock() - - //fmt.Printf("bound: pos=%d, h=%d\n", bound.Pos, bound.H) - - c.Append(248) - c.Append(305) - - fmt.Printf("pos: %d\n", c.pos) - fmt.Printf("%d\n", c.Chunks()) - - d := NewCumulativeDeltaDecompressor(buf, c.Size(), fracDigits) - d.RestoreFromEnd() - - for range 8 { - value, done = d.NextValue() - fmt.Println(value, done) - } - - //boundDecompressor := NewReverseCumulativeDeltaDecompressor(buf, c.Size(), fracDigits) - //boundDecompressor.RestoreFromBound(bound) - boundDecompressor := c.CreateDecompressor(fracDigits) - - fmt.Println("from bound:") - for range 8 { - value, done = boundDecompressor.NextValue() - fmt.Println(value, done) - } -} - -func TestInsdeltaBound(t *testing.T) { - var ( - fracDigits byte = 0 - buf = make([]byte, minBufferSize) - value float64 - done bool - ) - c := NewInstantDeltaCompressor(buf, 0, fracDigits) - // c.Append(1.55) - // c.Append(23) - // c.Append(23) - // c.Append(23) - // c.Append(23.5) - c.Append(10) - c.Append(70) - - c.Append(130) - - // fmt.Printf("pos: %d\n", c.pos) - // fmt.Printf("%d\n", buf.Chunks()[0]) - - c.Append(191) - - // fmt.Printf("pos: %d\n", c.pos) - // fmt.Printf("%d\n", buf.Chunks()[0]) - - c.Append(191) - - fmt.Printf("pos: %d\n", c.pos) - fmt.Printf("%d\n", c.Chunks()) - ////////////////////////// - - //bound := c.GetState() - - //fmt.Printf("bound: pos=%d, h=%d\n", bound.Pos, bound.H) - - c.Append(248) - c.Append(305) - - fmt.Printf("pos: %d\n", c.pos) - fmt.Printf("%d\n", c.Chunks()) - - d := NewInstantDeltaDecompressor(buf, c.Size(), fracDigits) - d.RestoreFromEnd() - - for range 8 { - value, done = d.NextValue() - fmt.Println(value, done) - } - - boundDecompressor := NewInstantDeltaDecompressor(buf, c.Size(), fracDigits) - //boundDecompressor.RestoreFromBound(bound) - - fmt.Println("from bound:") - for range 8 { - value, done = boundDecompressor.NextValue() - fmt.Println(value, done) - } -} - -func TestInsdeltaBound2(t *testing.T) { - var ( - fracDigits byte = 0 - buf = make([]byte, minBufferSize) - value float64 - done bool - ) - c := NewInstantDeltaCompressor(buf, 0, fracDigits) - // c.Append(1.55) - // c.Append(23) - // c.Append(23) - // c.Append(23) - // c.Append(23.5) - c.Append(305) - c.Append(248) - - c.Append(191) - - // fmt.Printf("pos: %d\n", c.pos) - // fmt.Printf("%d\n", buf.Chunks()[0]) - - c.Append(191) - - // fmt.Printf("pos: %d\n", c.pos) - // fmt.Printf("%d\n", buf.Chunks()[0]) - - c.Append(130) - - fmt.Printf("pos: %d\n", c.pos) - fmt.Printf("%d\n", c.Chunks()) - ////////////////////////// - - //bound := c.GetState() - - //fmt.Printf("bound: pos=%d, h=%d\n", bound.Pos, bound.H) - - c.Append(70) - c.Append(10) - - fmt.Printf("pos: %d\n", c.pos) - fmt.Printf("%d\n", c.Chunks()) - - d := NewInstantDeltaDecompressor(buf, c.Size(), fracDigits) - d.RestoreFromEnd() - - for range 8 { - value, done = d.NextValue() - fmt.Println(value, done) - } - - boundDecompressor := NewInstantDeltaDecompressor(buf, c.Size(), fracDigits) - //sboundDecompressor.RestoreFromBound(bound) - - fmt.Println("from bound:") - for range 8 { - value, done = boundDecompressor.NextValue() - fmt.Println(value, done) - } -} - -func TestCap(t *testing.T) { - a := make([]byte, 0, 10) - fmt.Println("len:", len(a)) - fmt.Println("cap:", cap(a)) - - a = append(a, 1) - fmt.Println("after append:") - fmt.Println("len:", len(a)) - fmt.Println("cap:", cap(a)) - fmt.Println("a:", a) - - a[1] = 55 - fmt.Println("after []:") - fmt.Println("len:", len(a)) - fmt.Println("cap:", cap(a)) - fmt.Println("a:", a) - - a = append(a, 2) - fmt.Println("after 2nd append:") - fmt.Println("len:", len(a)) - fmt.Println("cap:", cap(a)) - fmt.Println("a:", a) -} diff --git a/chunkenc/cumdelta.go b/chunkenc/cumdelta.go deleted file mode 100644 index 7b90a62..0000000 --- a/chunkenc/cumdelta.go +++ /dev/null @@ -1,366 +0,0 @@ -package chunkenc - -import ( - "fmt" - "log" - "math" - - bin "gordenko.dev/dima/bin/little" - "gordenko.dev/dima/qb" -) - -// COMPRESSOR REVERSE - -/* -Формат: -base value (var u64) -delta (var u64) -q (qty of previous deltas; msb=0 - series, msb=1 - non series, low 7 bits - qty 1..128) -delta -delta -delta -q - -Дельта рахується від base value. -Декодування у зворотньому порядку. - -Після base value слідують run або literal блоки. -Run блок - це delta + header byte в кінці. -Literal блок - це від одної до N дельт + header byte в кінці. -Спочатку створюється literal блок. -Якщо для останної дельти додається дублікат, literal блок модифікується - -лічильник зменшується до 1. А остання дельта переміщюється в новий run блок. -Причому лічильник 0 - означає 2 елементи. Приклад: -До: -v1 v2 v3 h-byte(literal, 3) <- v3 -Після: -v1 v2 h-byte(literal, 2) v3 h-byte(run, 2) -*/ - -type CumulativeDeltaCompressor struct { - buf []byte - coef float64 - pos int - baseValue float64 - lastDelta uint64 - lastDeltaSize int - h byte - state *CumulativeDeltaBound -} - -func NewCumulativeDeltaCompressor(buf []byte, size int, fracDigits byte) *CumulativeDeltaCompressor { - var coef float64 = 1 - if fracDigits > 0 { - coef = math.Pow(10, float64(fracDigits)) - } - s := &CumulativeDeltaCompressor{ - buf: buf, - pos: size, // перший вільний байт - coef: coef, - } - if size > 0 { - u64, _, err := bin.GetVarUint64(s.buf) - if err != nil { - log.Fatalf("bug: get base value: %s", err) - } - s.baseValue = float64(u64) / s.coef - s.h = s.buf[s.pos-1] - s.lastDelta, s.lastDeltaSize, err = bin.ReverseGetVarUint64(s.buf[:s.pos-2]) - if err != nil { - log.Fatalf("bug: get last delta: %s", err) - } - } - return s -} - -func (s *CumulativeDeltaCompressor) Size() int { - return s.pos -} - -func (s *CumulativeDeltaCompressor) CalcRequiredSpace(value float64) int { - return 0 -} - -func (s *CumulativeDeltaCompressor) Append(value float64) { - if s.pos == 0 { - // base value - n, _ := bin.PutVarUint64(s.buf[s.pos:], uint64(value*s.coef)) - s.pos += n - s.baseValue = value - s.appendNewLiteral(0) - } else { - delta := uint64((value-s.baseValue)*s.coef + eps) - 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) - } - } - } - } -} - -func (s *CumulativeDeltaCompressor) 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:], s.lastDelta) - s.pos += s.lastDeltaSize - s.h = 0 // run блок, довжини 2 - s.buf[s.pos] = s.h - s.pos++ -} - -func (s *CumulativeDeltaCompressor) convertLiteralToRun() { - // Знімаю flagLiteral, а лічильник 0 дорівнює 2 елементам в серії. - s.h = 0 - s.buf[s.pos-1] = s.h -} - -func (s *CumulativeDeltaCompressor) appendDeltaToLiteral(delta uint64) { - s.h++ // збільшую к-сть дельт - s.lastDelta = delta - s.pos-- - s.lastDeltaSize, _ = bin.ReversePutVarUint64(s.buf[s.pos:], delta) - s.pos += s.lastDeltaSize - s.buf[s.pos] = s.h - s.pos++ -} - -func (s *CumulativeDeltaCompressor) appendNewLiteral(delta uint64) { - s.h = flagLiteral - s.lastDelta = delta - s.lastDeltaSize, _ = bin.ReversePutVarUint64(s.buf[s.pos:], delta) - s.pos += s.lastDeltaSize - s.buf[s.pos] = flagLiteral // literal, length = 1 - s.pos++ -} - -func (s *CumulativeDeltaCompressor) DeleteLast() { - -} - -type CumulativeDeltaBound struct { - Pos int - H byte - LastDelta uint64 - Chunks []byte -} - -// delta h -func (s *CumulativeDeltaCompressor) Lock() { - if s.state != nil { - qb.Abort(qb.RepeatableLock, nil) - } - // позиція посувається вліво, отже може перескочити на попередній chunk - pos := s.pos - 1 - s.lastDeltaSize - s.state = &CumulativeDeltaBound{ - Pos: pos, - H: s.h, - LastDelta: s.lastDelta, - Chunks: s.buf[:s.pos], // fix check ? - } -} - -// fix - повернути в Pool буфери -func (s *CumulativeDeltaCompressor) Unlock() { - s.state = nil -} - -func (s *CumulativeDeltaCompressor) Offset() int { - if s.state != nil { - return s.state.Pos - } - return 0 -} - -func (s *CumulativeDeltaCompressor) 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, s.state.LastDelta) - // buf.SetByte(pos, s.state.H) - // pos++ - // return chunks, pos - return nil, 0 -} - -func (s *CumulativeDeltaCompressor) CreateDecompressor(fracDigits byte) qb.ValueDecompressor { - if s.state == nil { - d := NewCumulativeDeltaDecompressor(s.buf, s.Size(), fracDigits) - d.RestoreFromEnd() - return d - } - d := NewCumulativeDeltaDecompressor(s.buf, s.Size(), fracDigits) - d.RestoreFromBound(*s.state) - return d -} - -func (s *CumulativeDeltaCompressor) Renew() { - // УВАГА! - // state не чіпаємо - s.buf = make([]byte, minBufferSize) - s.pos = 0 - // - s.baseValue = 0 - s.lastDelta = 0 - s.lastDeltaSize = 0 - s.h = 0 -} - -func (s *CumulativeDeltaCompressor) Chunks() []byte { - return s.buf -} - -// DECOMPRESSOR - -type CumulativeDeltaDecompressor struct { - buf []byte - coef float64 - pos int - bound int - baseValue float64 - lastValue float64 - isRun bool - pending int - done bool -} - -func NewCumulativeDeltaDecompressor(buf []byte, size int, fracDigits byte) *CumulativeDeltaDecompressor { - var coef float64 = 1 - if fracDigits > 0 { - coef = math.Pow(10, float64(fracDigits)) - } - u64, n, err := bin.GetVarUint64(buf) - if err != nil { - log.Fatalf("bug: get base value: %s", err) - } - //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) - return &CumulativeDeltaDecompressor{ - buf: buf, - coef: coef, - pos: size - 1, // last elem - baseValue: float64(u64) / coef, - bound: n, - } -} - -func (s *CumulativeDeltaDecompressor) RestoreFromEnd() { - if s.pos > s.bound { - // читаю заголовок наступної серії - s.readHeader() - s.readValue() - } else { - s.done = true - } -} - -func (s *CumulativeDeltaDecompressor) RestoreFromBound(bound CumulativeDeltaBound) { - s.pos = bound.Pos - 1 - s.lastValue = s.baseValue + float64(bound.LastDelta) - s.decodeHeaderByte(bound.H) - fmt.Printf("restore from bound: isRun=%t, pending=%d\n", s.isRun, s.pending) -} - -func (s *CumulativeDeltaDecompressor) NextValue() (value float64, done bool) { - //fmt.Printf("NextValue(): bound: %d, pos: %d, pending: %d\n", s.bound, s.pos, s.pending) - if s.done { - return 0, true - } - // повертаю значення, що було прочитано в методі RestoreFromBound/RestoreFromEnd - value = s.lastValue - s.pending-- - if s.pending > 0 { - // якщо в серії залишаються елементи - if !s.isRun { - s.readValue() - } - } else if s.pos > s.bound { - // в серії більше немає елементів, отже перевіряє чи є ще дані в буфері. - // дані є - читаю заголовок наступної серії - s.readHeader() - s.readValue() - } else { - s.done = true - } - // серія завершена - перевіряю чи є ще серії - return value, false -} - -func (s *CumulativeDeltaDecompressor) 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 *CumulativeDeltaDecompressor) readValue() { - 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.lastValue = s.baseValue + float64(u64)/s.coef -} - -func (s *CumulativeDeltaDecompressor) decodeHeaderByte(h byte) { - s.isRun = h < 128 - if s.isRun { - s.pending = int(h&127) + 2 - } else { - s.pending = int(h&127) + 1 - } -} - -// diff --git a/chunkenc/insdelta.go b/chunkenc/insdelta.go deleted file mode 100644 index 35c9336..0000000 --- a/chunkenc/insdelta.go +++ /dev/null @@ -1,420 +0,0 @@ -package chunkenc - -import ( - "fmt" - "log" - "math" - - "gordenko.dev/dima/qb" - "gordenko.dev/dima/qb/bin" -) - -type InstantDeltaCompressor struct { - buf []byte - coef float64 - pos int - baseValue float64 - lastDelta int64 - lastDeltaSize int - h byte - state *InstantDeltaBound -} - -func NewInstantDeltaCompressor(buf []byte, size int, fracDigits byte) *InstantDeltaCompressor { - var coef float64 = 1 - if fracDigits > 0 { - coef = math.Pow(10, float64(fracDigits)) - } - s := &InstantDeltaCompressor{ - buf: buf, - pos: size, - coef: coef, - } - if size > 0 { - i64, _, err := bin.GetVarInt64(s.buf) - if err != nil { - log.Fatalf("bug: get base value: %s", err) - } - s.baseValue = float64(i64) / s.coef - s.h = s.buf[s.pos-1] - s.lastDelta, s.lastDeltaSize, err = bin.ReverseGetVarInt64(s.buf[:s.pos-2]) - if err != nil { - log.Fatalf("bug: get last delta: %s", err) - } - } - return s -} - -func (s *InstantDeltaCompressor) Size() int { - return s.pos -} - -func (s *InstantDeltaCompressor) Append(value float64) { - if s.pos == 0 { - // base value - n, _ := bin.PutVarInt64(s.buf[s.pos:], int64(value*s.coef)) - s.pos += n - s.baseValue = value - s.appendNewLiteral(0) - } else { - tmp := (value - s.baseValue) * s.coef - if tmp > 0 { - tmp += eps - } else { - tmp -= eps - } - delta := int64(tmp) - 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 - // fix encode delta - s.appendNewLiteral(delta) - } - } else { - // literal блок. - // Якщо в ньому лише одне значення - перетворюю його на run блок. - // Інакше забираю останнє значення із literal блока і додаю новий run блок. - q := s.h & 127 - if q == 0 { // 1 кодується як 0 - s.convertLiteralToRun() - } else { - s.convertLastFromLiteralToRun() - } - } - } else { - if s.h < 127 { - // 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) - } - } - } - } -} - -func (s *InstantDeltaCompressor) convertLastFromLiteralToRun() { - // Зменшую кількість елементів в literal блоці - s.h-- - s.pos -= 1 + s.lastDeltaSize - s.buf[s.pos] = s.h // закриваю literal блок - s.pos++ - s.lastDeltaSize, _ = bin.ReversePutVarInt64(s.buf[:s.pos], s.lastDelta) - s.pos += s.lastDeltaSize - s.h = 0 // run блок, довжини 2 - s.buf[s.pos] = s.h - s.pos++ -} - -func (s *InstantDeltaCompressor) convertLiteralToRun() { - // Знімаю flagLiteral, а лічильник 0 дорівнює 2 елементам в серії. - s.h = 0 - s.buf[s.pos-1] = s.h -} - -func (s *InstantDeltaCompressor) appendDeltaToLiteral(delta int64) { - s.h++ // збільшую к-сть дельт - s.lastDelta = delta - s.pos-- - s.lastDeltaSize, _ = bin.ReversePutVarInt64(s.buf[s.pos:], delta) - s.pos += s.lastDeltaSize - s.buf[s.pos] = s.h - s.pos++ -} - -func (s *InstantDeltaCompressor) appendNewLiteral(delta int64) { - s.h = flagLiteral - s.lastDelta = delta - s.lastDeltaSize, _ = bin.ReversePutVarInt64(s.buf[s.pos:], delta) - s.pos += s.lastDeltaSize - s.buf[s.pos] = flagLiteral // literal, length = 1 - s.pos++ -} - -func (s *InstantDeltaCompressor) DeleteLast() {} - -type InstantDeltaBound struct { - Pos int - H byte - LastDelta int64 - Chunks []byte -} - -// delta h -func (s *InstantDeltaCompressor) Lock() { - if s.state != nil { - qb.Abort(qb.RepeatableLock, nil) - } - // позиція посувається вліво, отже може перескочити на попередній chunk - pos := s.pos - 1 - s.lastDeltaSize - s.state = &InstantDeltaBound{ - Pos: pos, - H: s.h, - LastDelta: s.lastDelta, - Chunks: s.buf[:s.pos], // fix check ? - } -} - -// fix - повернути в Pool буфери -func (s *InstantDeltaCompressor) Unlock() { - s.state = nil -} - -func (s *InstantDeltaCompressor) Offset() int { - if s.state != nil { - return s.state.Pos - } - return 0 -} - -func (s *InstantDeltaCompressor) 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.ReversePutVarInt64(pos, s.state.LastDelta) - // buf.SetByte(pos, s.state.H) - // pos++ - // return chunks, pos - return nil, 0 -} - -func (s *InstantDeltaCompressor) CreateDecompressor(fracDigits byte) qb.ValueDecompressor { - if s.state == nil { - d := NewInstantDeltaDecompressor(s.buf, s.Size(), fracDigits) - d.RestoreFromEnd() - return d - } - d := NewInstantDeltaDecompressor(s.buf, s.Size(), fracDigits) - d.RestoreFromBound(*s.state) - return d -} - -func (s *InstantDeltaCompressor) Renew() { - // УВАГА! - // state не чіпаємо - s.buf = make([]byte, minBufferSize) - s.pos = 0 - // - s.baseValue = 0 - s.lastDelta = 0 - s.lastDeltaSize = 0 - s.h = 0 -} - -func (s *InstantDeltaCompressor) CalcRequiredSpace(value float64) int { - return 0 -} - -func (s *InstantDeltaCompressor) Chunks() []byte { - return s.buf -} - -// DECOMPRESSOR - -type InstantDeltaDecompressor struct { - buf []byte - coef float64 - pos int - bound int - baseValue float64 - lastValue float64 - isRun bool - pending int - done bool -} - -func NewInstantDeltaDecompressor(buf []byte, size int, fracDigits byte) *InstantDeltaDecompressor { - var coef float64 = 1 - if fracDigits > 0 { - coef = math.Pow(10, float64(fracDigits)) - } - i64, n, err := bin.GetVarInt64(buf) - if err != nil { - log.Fatalf("bug: get base value: %s", err) - } - //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) - return &InstantDeltaDecompressor{ - buf: buf, - coef: coef, - pos: size - 1, // last elem - baseValue: float64(i64) / coef, - bound: n, - } -} - -func (s *InstantDeltaDecompressor) RestoreFromEnd() { - if s.pos > s.bound { - // читаю заголовок наступної серії - s.readHeader() - s.readValue() - } else { - s.done = true - } -} - -func (s *InstantDeltaDecompressor) RestoreFromBound(bound InstantDeltaBound) { - s.pos = bound.Pos - 1 - s.lastValue = s.baseValue + float64(bound.LastDelta) - s.decodeHeaderByte(bound.H) - fmt.Printf("restore from bound: isRun=%t, pending=%d\n", s.isRun, s.pending) -} - -func (s *InstantDeltaDecompressor) NextValue() (value float64, done bool) { - //fmt.Printf("NextValue(): bound: %d, pos: %d, pending: %d\n", s.bound, s.pos, s.pending) - if s.done { - return 0, true - } - // повертаю значення, що було прочитано в методі RestoreFromBound/RestoreFromEnd - value = s.lastValue - s.pending-- - if s.pending > 0 { - // якщо в серії залишаються елементи - if !s.isRun { - s.readValue() - } - } else if s.pos > s.bound { - // в серії більше немає елементів, отже перевіряє чи є ще дані в буфері. - // дані є - читаю заголовок наступної серії - s.readHeader() - s.readValue() - } else { - s.done = true - } - // серія завершена - перевіряю чи є ще серії - return value, false -} - -func (s *InstantDeltaDecompressor) 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 *InstantDeltaDecompressor) readValue() { - i64, n, err := bin.ReverseGetVarInt64(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.lastValue = s.baseValue + float64(i64)/s.coef -} - -func (s *InstantDeltaDecompressor) decodeHeaderByte(h byte) { - s.isRun = h < 128 - if s.isRun { - s.pending = int(h&127) + 2 - } else { - s.pending = int(h&127) + 1 - } -} - -// DECOMPRESSOR - -// type ReverseInstantDeltaDecompressor struct { -// buf *conbuf.ContinuousBuffer -// pos int -// bound int -// baseValue float64 -// lastValue float64 -// coef float64 -// isRun bool -// pending int -// } - -// func NewReverseInstantDeltaDecompressor(buf *conbuf.ContinuousBuffer, size int, fracDigits byte) *ReverseInstantDeltaDecompressor { -// var coef float64 = 1 -// if fracDigits > 0 { -// coef = math.Pow(10, float64(fracDigits)) -// } -// u64, n, err := buf.GetVarInt64(0) -// if err != nil { -// log.Fatalf("bug: get base value: %s", err) -// } -// //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) -// return &ReverseInstantDeltaDecompressor{ -// buf: buf, -// coef: coef, -// pos: size - 1, // last elem -// baseValue: float64(u64) / coef, -// bound: n, -// } -// } - -// func (s *ReverseInstantDeltaDecompressor) NextValue() (value float64, done bool) { -// //fmt.Printf("NextValue(): bound: %d, pos: %d, pending: %d\n", s.bound, s.pos, s.pending) -// if s.pending > 0 { -// // якщо в серії залишаються елементи -// if !s.isRun { -// s.readValue() -// } -// s.pending-- -// return s.lastValue, false -// } -// if s.pos > s.bound { -// // читаю заголовок наступної серії -// s.readHeader() -// s.readValue() -// s.pending-- -// return s.lastValue, false -// } -// // серія завершена - перевіряю чи є ще серії -// return 0, true -// } - -// func (s *ReverseInstantDeltaDecompressor) readHeader() { -// h := s.buf.GetByte(s.pos) -// s.pos-- -// s.isRun = h < 128 -// if s.isRun { -// s.pending = int(h&127) + 2 -// } else { -// s.pending = int(h&127) + 1 -// } -// // fmt.Println("h:", h) -// // fmt.Println("isRun:", s.isRun) -// // fmt.Println("pending:", s.pending) -// } -// func (s *ReverseInstantDeltaDecompressor) readValue() { -// i64, n, err := s.buf.ReverseGetVarInt64(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.lastValue = s.baseValue + float64(i64)/s.coef -// } diff --git a/database/metric.go b/database/metric.go index 6ba426f..ec87527 100644 --- a/database/metric.go +++ b/database/metric.go @@ -8,6 +8,8 @@ import ( // METRIC +const minBufferSize = 1024 + type IndexLevelTail struct { Payload []byte RecordsCount int @@ -21,38 +23,39 @@ type _metric struct { Since uint32 UntilValue float64 Until uint32 + Buffer []byte Timestamps qb.TimestampCompressor Values qb.ValueCompressor XLock bool RLocks int WaitQueue []any - IndexLevelTails []IndexLevelTail // root - last element + IndexLevelTails []txlog.IndexLevelTail // root - last element } //IndexLevels [][]IndexRec // root - last element -func (s *_metric) ReinitBy(timestamp uint32, value float64) { - s.Timestamps.Renew() - s.Values.Renew() +// func (s *_metric) ReinitBy(timestamp uint32, value float64) { +// s.Timestamps.Renew() +// s.Values.Renew() - s.Timestamps.Append(timestamp) - s.Values.Append(value) +// s.Timestamps.Append(timestamp) +// s.Values.Append(value) - s.Since = timestamp - s.SinceValue = value - s.Until = timestamp - s.UntilValue = value -} +// s.Since = timestamp +// s.SinceValue = value +// s.Until = timestamp +// s.UntilValue = value +// } func (s *_metric) DeleteMeasures() { - s.Timestamps.Renew() - s.Values.Renew() + // s.Timestamps.Renew() + // s.Values.Renew() - s.LastPageNo = 0 - s.Since = 0 - s.SinceValue = 0 - s.Until = 0 - s.UntilValue = 0 + // s.LastPageNo = 0 + // s.Since = 0 + // s.SinceValue = 0 + // s.Until = 0 + // s.UntilValue = 0 } // func (s *_metric) startAppendMeasures(req tryAppendMeasuresReq, sendToStorage func(txlog.AppendedMeasures)) { @@ -73,15 +76,15 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu timestamps = s.Timestamps values = s.Values - indexLevels []atree.IndexLevelTail - dataPages []atree.DataPayload + indexLevels []txlog.IndexLevelTail + dataPages []txlog.DataPayload //written int //resultCode byte ) - s.Values.Lock() - s.Timestamps.Lock() + s.Values.CaptureState() + s.Timestamps.CaptureState() for idx, measure := range req.Measures { if s.Since == 0 { @@ -91,8 +94,8 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu // якщо idx == 0 - одразу знімаю блокування і нічого не відправляю в txlog if measure.Timestamp <= s.Until { if idx == 0 { - s.Values.Unlock() - s.Timestamps.Unlock() + s.Values.ForgetCapturedState() + s.Timestamps.ForgetCapturedState() req.ResultCh <- tryAppendMeasuresResult{ ResultCode: ExpiredMeasure, @@ -107,8 +110,8 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu if s.MetricType == qb.Cumulative && measure.Value < s.UntilValue { if idx == 0 { - s.Values.Unlock() - s.Timestamps.Unlock() + s.Values.ForgetCapturedState() + s.Timestamps.ForgetCapturedState() req.ResultCh <- tryAppendMeasuresResult{ ResultCode: NonMonotonicValue, @@ -122,31 +125,38 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu } // fix - 1 + 8 bytes - extraSpace := timestamps.CalcRequiredSpace(measure.Timestamp) + - values.CalcRequiredSpace(measure.Value) - totalSpace := timestamps.Size() + values.Size() + extraSpace + timestampCompressionWay, timestampRequiredSpace := timestamps.Evaluate(measure.Timestamp) + valueCompressionWay, valueRequiredSpace := values.Evaluate(measure.Value) + + totalSpace := timestampRequiredSpace + valueRequiredSpace if totalSpace <= atree.DataPagePayloadSize { // накопичую - timestamps.Append(measure.Timestamp) - values.Append(measure.Value) + timestamps.Compress(timestampCompressionWay, measure.Timestamp) + values.Compress(valueCompressionWay, measure.Value) } else { // сторінка заповнена + buffer := make([]byte, minBufferSize) + + timestampsSize := timestamps.Rotate(buffer) + valuesSize := values.Rotate(buffer) // prevPageNo - виставляю в txlog, коли забираю номер сторінки із freeList або генерую новий - dataPages = append(dataPages, atree.DataPayload{ + dataPages = append(dataPages, txlog.DataPayload{ Since: s.Since, - Content: nil, - TimestampsSize: timestamps.Size(), - ValuesSize: values.Size(), + Content: s.Buffer, + TimestampsSize: timestampsSize, + ValuesSize: valuesSize, }) - // обнуляються буфери, але state незмінний - timestamps.Renew() - values.Renew() + // renew + s.Buffer = buffer - timestamps.Append(measure.Timestamp) - values.Append(measure.Value) + timestampCompressionWay, _ = timestamps.Evaluate(measure.Timestamp) + valueCompressionWay, _ = values.Evaluate(measure.Value) + + timestamps.Compress(timestampCompressionWay, measure.Timestamp) + values.Compress(valueCompressionWay, measure.Value) s.Since = measure.Timestamp } @@ -161,8 +171,8 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu if len(dataPages) > 0 { // пишу в txlog довгим шляхом через redo файл і запис в data файл for _, tail := range s.IndexLevelTails { - indexLevels = append(indexLevels, atree.IndexLevelTail{ - Payload: tail.Payload, + indexLevels = append(indexLevels, txlog.IndexLevelTail{ + Records: tail.Records, RecordsCount: tail.RecordsCount, }) } @@ -189,8 +199,8 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu func (s *_metric) FinAppendMeasures(rec txlog.AppendMeasuresSummary) { // Видаляю state. Оригінальні Timestamps і Values вже мають останню версію - s.Values.Unlock() - s.Timestamps.Unlock() + s.Values.ForgetCapturedState() + s.Timestamps.ForgetCapturedState() // fix write index levels // update prev pageNo // if len(rec.DataPages) > 0 { @@ -201,80 +211,80 @@ func (s *_metric) FinAppendMeasures(rec txlog.AppendMeasuresSummary) { } // В txlog я передав повний індекс. У нього додали елементи (можливо нові рівні). // Тому проста заміна - s.IndexLevels = rec.Index + s.IndexLevelTails = rec.Index } // READ func (s *_metric) StartRangeScan(req tryRangeScanReq) { - if s.Since == 0 { - req.ResultCh <- rangeScanResult{ - ResultCode: QueryDone, - } - return - } + // if s.Since == 0 { + // req.ResultCh <- rangeScanResult{ + // ResultCode: QueryDone, + // } + // return + // } - if req.Since > s.Until { - 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 - } - } + // 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) + // timestampDecompressor := s.Timestamps.CreateDecompressor() + // valueDecompressor := s.Values.CreateDecompressor(s.FracDigits) - for { - timestamp, done := timestampDecompressor.NextValue() - if done { - break - } + // for { + // timestamp, done := timestampDecompressor.NextValue() + // if done { + // break + // } - value, done := valueDecompressor.NextValue() - if done { - qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug) - } + // 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 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, - } - } + // 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) { diff --git a/enc/cumdelta.go b/enc/cumdelta.go index 45bab32..f140421 100644 --- a/enc/cumdelta.go +++ b/enc/cumdelta.go @@ -1,102 +1,440 @@ package enc import ( + "log" "math" - "gordenko.dev/dima/qb/bin" + bin "gordenko.dev/dima/bin/little" + "gordenko.dev/dima/qb" ) -type ReverseCumulativeDeltaDecompressor struct { - buf []byte - pos int - bound int - firstValue float64 - lastValue float64 - length uint16 - coef float64 - idxOf8 uint - s8 byte +// COMPRESSOR REVERSE + +/* +Формат: +base value (var u64) +delta (var u64) +q (qty of previous deltas; msb=0 - series, msb=1 - non series, low 7 bits - qty 1..128) +delta +delta +delta +q + +Дельта рахується від base value. +Декодування у зворотньому порядку. + +Після base value слідують run або literal блоки. +Run блок - це delta + header byte в кінці. +Literal блок - це від одної до N дельт + header byte в кінці. +Спочатку створюється literal блок. +Якщо для останної дельти додається дублікат, literal блок модифікується - +лічильник зменшується до 1. А остання дельта переміщюється в новий run блок. +Причому лічильник 0 - означає 2 елементи. Приклад: +До: +v1 v2 v3 h-byte(literal, 3) <- v3 +Після: +v1 v2 h-byte(literal, 2) v3 h-byte(run, 2) +*/ + +type CumulativeDeltaCompressor struct { + buf []byte + coef float64 + pos int + baseValue float64 + lastDelta uint64 + h byte + state *CumulativeDeltaCapturedState } -func NewReverseCumulativeDeltaDecompressor(buf []byte, fracDigits byte) *ReverseCumulativeDeltaDecompressor { +// Після відновлення із снапшота +func NewCumulativeDeltaCompressor(buf []byte, payloadSize int, fracDigits byte) *CumulativeDeltaCompressor { var coef float64 = 1 if fracDigits > 0 { coef = math.Pow(10, float64(fracDigits)) } - return &ReverseCumulativeDeltaDecompressor{ + s := &CumulativeDeltaCompressor{ buf: buf, + pos: payloadSize, // перший вільний байт coef: coef, - pos: len(buf), + } + if payloadSize > 0 { + u64, _, err := bin.GetVarUint64(buf) + if err != nil { + log.Fatalf("bug: get base value: %s", err) + } + s.baseValue = float64(u64) / s.coef + s.pos-- + s.h = s.buf[s.pos] + var n int + s.lastDelta, n, err = bin.ReverseGetVarUint64(s.buf[:s.pos]) + if err != nil { + log.Fatalf("bug: get last delta: %s", err) + } + s.pos -= n + } + return s +} + +// func (s *CumulativeDeltaCompressor) Size() int { +// return s.pos +// } + +func (s *CumulativeDeltaCompressor) Evaluate(value float64) (compressionWay int, requiredSpace int) { + var ( + delta = uint64((value-s.baseValue)*s.coef + eps) + ) + if s.pos > 0 { + if s.h < 128 { + // run + if delta == s.lastDelta && s.h < 127 { + compressionWay = incrementRun + requiredSpace += bin.CountVarUint64(uint64(s.lastDelta)) + // current delta + hSize + } else { + compressionWay = endSeries + requiredSpace += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta + hSize + // h - end of run + bin.CountVarUint64(uint64(delta)) + // new literal + hSize // new h + } + } else { + // literal + if delta != s.lastDelta { + if s.h < 255 { + compressionWay = incrementLiteral + requiredSpace += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta + bin.CountVarUint64(uint64(delta)) + // new delta + hSize + } else { + compressionWay = endSeries + requiredSpace += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta + hSize + // h - end of literal + bin.CountVarUint64(uint64(delta)) + // new literal + hSize // new h + } + } else { + compressionWay = startRun + if s.h > 128 { + requiredSpace += hSize // h - end of literal + } + requiredSpace += bin.CountVarUint64(uint64(delta)) + // new delta + hSize // new h + } + } + } else { + // encode base value + compressionWay = addBaseValue + requiredSpace = bin.CountVarUint64(uint64(value*s.coef)) + // base value + bin.CountVarUint64(uint64(delta)) + // new delta + +hSize // new h + } + return +} + +func (s *CumulativeDeltaCompressor) Compress(compressionWay int, value float64) { + delta := uint64((value-s.baseValue)*s.coef + eps) + switch compressionWay { + case incrementRun: + s.h++ + case incrementLiteral: + // write previous delta and increment counter + n, _ := bin.ReversePutVarUint64(s.buf[s.pos:], uint64(s.lastDelta)) + s.pos += n + s.lastDelta = delta + s.h++ + case endSeries: + n, _ := bin.ReversePutVarUint64(s.buf[s.pos:], uint64(s.lastDelta)) + s.pos += n + // write h - end of run + s.buf[s.pos] = s.h + s.pos++ + // start new literal (length=1) + s.lastDelta = delta + s.h = 128 + case startRun: + if s.h > 128 { + // write h - end of literal (because length > 1) + s.h-- + s.buf[s.pos] = s.h + s.pos++ + } + // start new run (length=2) + s.h = 0 + case addBaseValue: + // start new literal (length=1) + n, _ := bin.PutVarUint64(s.buf[s.pos:], uint64(value*s.coef)) + s.pos += n + s.baseValue = value + // start new literal (length=1) + s.lastDelta = 0 + s.h = 128 } } -func (s *ReverseCumulativeDeltaDecompressor) NextValue() (value float64, done bool) { - if s.length > 0 { - s.length-- - return s.lastValue, false +func (s *CumulativeDeltaCompressor) DeleteLast() { + +} + +type CumulativeDeltaCapturedState struct { + H byte + LastDelta uint64 + Payload []byte +} + +func (s *CumulativeDeltaCompressor) CaptureState() { + if s.state != nil { + qb.Abort(qb.RepeatableLock, nil) } - if s.pos < s.bound { + // позиція посувається вліво, отже може перескочити на попередній chunk + s.state = &CumulativeDeltaCapturedState{ + H: s.h, + LastDelta: s.lastDelta, + Payload: s.buf[:s.pos], + } +} + +// fix - повернути в Pool буфери +func (s *CumulativeDeltaCompressor) ForgetCapturedState() { + s.state = nil +} + +func (s *CumulativeDeltaCompressor) Offset() int { + // if s.state != nil { + // return s.state.Pos + // } + return 0 +} + +// Snapshot - для створення снапшота. +func (s *CumulativeDeltaCompressor) Snapshot() (left []byte, right []byte) { + if s.state == nil { + left = s.buf[:s.pos] + right = s.encodeTail(s.lastDelta, s.h) + } else { + left = s.state.Payload + right = s.encodeTail(s.state.LastDelta, s.state.H) + } + return +} + +func (s *CumulativeDeltaCompressor) encodeTail(lastDelta uint64, h byte) []byte { + tail := make([]byte, 10) // max var uint64 + h + n, _ := bin.PutVarUint64(tail, lastDelta) + tail[n] = h + return tail[:n+1] +} + +func (s *CumulativeDeltaCompressor) Rotate(newbuf []byte) (payloadSize int) { + n, _ := bin.PutVarUint64(s.buf[s.pos:], s.lastDelta) + s.pos += n + s.buf[s.pos] = s.h + s.pos++ + payloadSize = s.pos + // УВАГА! + // state не чіпаємо + s.buf = newbuf + s.pos = 0 + s.baseValue = 0 + s.lastDelta = 0 + s.h = 0 + return +} + +func (s *CumulativeDeltaCompressor) CreateDecompressor() qb.ValueDecompressor { + var ( + h = s.h + lastDelta = s.lastDelta + payload = s.buf[:s.pos] + ) + if s.state != nil { + h = s.state.H + lastDelta = s.state.LastDelta + payload = s.state.Payload + } + return NewCumulativeDeltaDecompressorFromState(CumulativeDeltaDecompressorFromStateOptions{ + Coef: s.coef, + H: h, + LastDelta: lastDelta, + Payload: payload, + }) +} + +func (s *CumulativeDeltaCompressor) Chunks() []byte { + return s.buf +} + +// DECOMPRESSOR + +type CumulativeDeltaDecompressor struct { + buf []byte + coef float64 + pos int + bound int + baseValue float64 + lastValue float64 + isRun bool + pending int + done bool +} + +func NewCumulativeDeltaDecompressor(buf []byte, fracDigits byte) *CumulativeDeltaDecompressor { + var coef float64 = 1 + if fracDigits > 0 { + coef = math.Pow(10, float64(fracDigits)) + } + u64, n, err := bin.GetVarUint64(buf) + if err != nil { + log.Fatalf("bug: get base value: %s", err) + } + //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) + s := &CumulativeDeltaDecompressor{ + buf: buf, + coef: coef, + pos: len(buf), // first free + baseValue: float64(u64) / coef, + bound: n, + } + // читаю заголовок наступної серії + s.readHeader() + s.readValue() + return s +} + +type CumulativeDeltaDecompressorFromStateOptions struct { + Coef float64 + H byte + LastDelta uint64 + Payload []byte +} + +func NewCumulativeDeltaDecompressorFromState(opt CumulativeDeltaDecompressorFromStateOptions) *CumulativeDeltaDecompressor { + u64, n, err := bin.GetVarUint64(opt.Payload) + if err != nil { + log.Fatalf("bug: get base value: %s", err) + } + //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/opt.Coef, n, size) + s := &CumulativeDeltaDecompressor{ + buf: opt.Payload, + coef: opt.Coef, + pos: len(opt.Payload), // first free + baseValue: float64(u64) / opt.Coef, + bound: n, + } + s.lastValue = s.baseValue + float64(opt.LastDelta)/s.coef + s.decodeHeaderByte(opt.H) + //fmt.Printf("payload: %d\n", opt.Payload) + //fmt.Printf("restore from state: bound=%d, isRun=%t, pending=%d, baseValue=%v, lastValue=%v\n", + // n, s.isRun, s.pending, s.baseValue, s.lastValue) + return s +} + +func (s *CumulativeDeltaDecompressor) NextValue() (value float64, done bool) { + //fmt.Printf("NextValue(): bound: %d, pos: %d, pending: %d\n", s.bound, s.pos, s.pending) + if s.done { return 0, true } - - if s.pos == len(s.buf) { - u64, n, err := bin.GetVarUint64(s.buf) - if err != nil { - panic(err) + // метод працює як do while - спочатку значення, а потім перевірка умови + value = s.lastValue + s.pending-- + if s.pending > 0 { + // якщо в серії залишаються елементи + if !s.isRun { + s.readValue() } - s.firstValue = float64(u64) / s.coef - s.bound = n - s.pos-- - s.idxOf8 = uint(8 - s.buf[s.pos]) - s.pos-- - s.s8 = s.buf[s.pos] - s.pos-- - s.readVar() - if s.length > 0 { - s.length-- - } - return s.lastValue, false - } - - if s.idxOf8 == 0 { - s.s8 = s.buf[s.pos] - s.pos-- - } - s.readVar() - if s.length > 0 { - s.length-- - } - return s.lastValue, false -} - -func (s *ReverseCumulativeDeltaDecompressor) readVar() { - u64, n, err := bin.ReverseGetVarUint64(s.buf[:s.pos+1]) - if err != nil { - panic(err) - } - s.pos -= n - s.lastValue = s.firstValue + float64(u64)/s.coef - - var flag byte = 1 << s.idxOf8 - if (s.s8 & flag) == flag { - s.decodeLength() - } - if s.idxOf8 == 7 { - s.idxOf8 = 0 + } else if s.pos > s.bound { + // в серії більше немає елементів, отже перевіряє чи є ще дані в буфері. + // дані є - читаю заголовок наступної серії + s.readHeader() + s.readValue() } else { - s.idxOf8++ + s.done = true } + // серія завершена - перевіряю чи є ще серії + return value, false } -func (s *ReverseCumulativeDeltaDecompressor) decodeLength() { - b1 := s.buf[s.pos] +func (s *CumulativeDeltaDecompressor) readHeader() { + //fmt.Println("read from pos:", s.pos) s.pos-- - if b1 < 128 { - s.length = uint16(b1) - } else { - b2 := s.buf[s.pos] - s.pos-- - s.length = uint16(b1&127) | (uint16(b2) << 7) - } - s.length += 2 + h := s.buf[s.pos] + s.decodeHeaderByte(h) + + // fmt.Println("h:", h) + // fmt.Println("isRun:", s.isRun) + // fmt.Println("pending:", s.pending) } +func (s *CumulativeDeltaDecompressor) readValue() { + 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.lastValue = s.baseValue + float64(u64)/s.coef + //fmt.Println(s.baseValue, float64(u64)/s.coef) +} + +func (s *CumulativeDeltaDecompressor) decodeHeaderByte(h byte) { + s.isRun = h < 128 + if s.isRun { + s.pending = int(h) + 2 + } else { + s.pending = int(h&127) + 1 + } +} + +// + +// func (s *CumulativeDeltaCompressor) Append(value float64) { +// if s.pos == 0 { +// // base value +// n, _ := bin.PutVarUint64(s.buf[s.pos:], uint64(value*s.coef)) +// s.pos += n +// s.baseValue = value +// s.appendNewLiteral(0) +// } else { +// delta := uint64((value-s.baseValue)*s.coef + eps) +// 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) +// } +// } +// } +// } +// } diff --git a/enc/enc.go b/enc/enc.go index 579d518..80bc136 100644 --- a/enc/enc.go +++ b/enc/enc.go @@ -1,3 +1,18 @@ package enc const eps = 0.000001 + +const ( + flagLiteral = 128 + + hSize = 1 + + // append scripts + addBaseValue = 0 // cumulative and instant only + addUnixtime = 0 // time only + incrementRun = 1 + incrementLiteral = 2 + endSeries = 3 + startRun = 4 + add1stDelta = 5 // time only +) diff --git a/enc/enc_test.go b/enc/enc_test.go new file mode 100644 index 0000000..c1ba3d3 --- /dev/null +++ b/enc/enc_test.go @@ -0,0 +1,1005 @@ +package enc + +import ( + "bytes" + "fmt" + "math" + "slices" + "testing" + + bin "gordenko.dev/dima/bin/little" +) + +func equalFloatSlices(a, b []float64, epsilon float64) bool { + return slices.EqualFunc(a, b, func(x, y float64) bool { + return math.Abs(x-y) <= epsilon + }) +} + +// CUMULATIVE + +func TestCumulativeDeltaCompressor(t *testing.T) { + var ( + testCases = []struct { + Nums []float64 + RepeatLastDeltaNTimes int + Code int + RequiredSpace int + Buf []byte + BaseValue float64 + LastDelta uint64 + H byte + }{ + { + Nums: []float64{ + 1.5, + }, + Code: addBaseValue, + RequiredSpace: 3, // base value, delta, h + Buf: []byte{ + 143, 0, 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 128, // literal, length = 1 + }, + { + Nums: []float64{ + 1.5, + 1.5, + }, + Code: startRun, // literal changed to run + RequiredSpace: 2, // delta, h + Buf: []byte{ + 143, 0, 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 0, // run, length = 2 + }, + { + Nums: []float64{ + 1.5, + 1.6, + 1.6, + }, + Code: startRun, // literal decreased by 1 + RequiredSpace: 3, // end of literal, new delta, new h + Buf: []byte{ + 143, // base value + 128, // literal 1st delta (0) + 128, // h - end of literal, length = 1 + 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 1, + H: 0, // run, length = 2 + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, + }, + Code: incrementRun, + RequiredSpace: 2, + Buf: []byte{ + 143, // base value + 0, 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 1, // run, length = 3 + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, + 1.6, + }, + Code: endSeries, // end of run + RequiredSpace: 4, + Buf: []byte{ + 143, // base value + 128, // run delta (0) + 1, // h of run, length = 3 + 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 1, + H: 128, // literal, length = 1 + }, + { + Nums: []float64{ + 1.5, + }, + RepeatLastDeltaNTimes: 128, + Code: incrementRun, // h byte full filled + RequiredSpace: 2, + Buf: []byte{ + 143, // base value + 0, 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 127, // run, length = 129 + }, + { + Nums: []float64{ + 1.5, + }, + RepeatLastDeltaNTimes: 129, + Code: endSeries, // run, h byte overflowed + RequiredSpace: 4, // run delta, h of run, 1st literal delta, h of literal + Buf: []byte{ + 143, // base value + 128, // run delta (0) + 127, // h of run (length = 129) + 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 128, // literal, length = 1 + }, + { + Nums: []float64{ + 1.5, + 1.6, + }, + Code: incrementLiteral, + RequiredSpace: 3, // previous delta, new delta, h + Buf: []byte{ + 143, // base value + 128, // literal 1st delta (0) + 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 1, + H: 129, // literal, length = 2 + }, + } + ) + + for caseIdx, testCase := range testCases { + var ( + fracDigits byte = 1 + buf = make([]byte, 8) + payloadSize = 0 + compressionWay int + requiredSpace int + ) + c := NewCumulativeDeltaCompressor(buf, payloadSize, fracDigits) + for _, num := range testCase.Nums { + compressionWay, requiredSpace = c.Evaluate(num) + c.Compress(compressionWay, num) + } + if testCase.RepeatLastDeltaNTimes > 0 { + num := testCase.Nums[len(testCase.Nums)-1] + for range testCase.RepeatLastDeltaNTimes { + compressionWay, requiredSpace = c.Evaluate(num) + c.Compress(compressionWay, num) + } + } + if compressionWay != testCase.Code { + t.Fatalf("%d: got code %d are not equal expected %d", + caseIdx, compressionWay, testCase.Code) + } + if requiredSpace != testCase.RequiredSpace { + t.Fatalf("%d: got requiredSpace %d are not equal expected %d", + caseIdx, requiredSpace, testCase.RequiredSpace) + } + if !bytes.Equal(buf, testCase.Buf) { + t.Fatalf("%d: got buf %v are not equal expected %v", + caseIdx, buf, testCase.Buf) + } + if c.baseValue != testCase.BaseValue { + t.Fatalf("%d: got lastUnixtime %v are not equal expected %v", + caseIdx, c.baseValue, testCase.BaseValue) + } + if c.lastDelta != testCase.LastDelta { + t.Fatalf("%d: got lastDelta %d are not equal expected %d", + caseIdx, c.lastDelta, testCase.LastDelta) + } + if c.h != testCase.H { + t.Fatalf("%d: got h %d are not equal expected %d", + caseIdx, c.h, testCase.H) + } + } +} + +func TestCumulativeDeltaDecompressorFromState(t *testing.T) { + var ( + testCases = []struct { + Nums []float64 + //RepeatLastDeltaNTimes int + }{ + { + // addBaseValue + Nums: []float64{ + 1.5, + }, + }, + { + // startRun + Nums: []float64{ + 1.5, + 1.5, + }, + }, + { + // startRun + Nums: []float64{ + 1.5, + 1.6, + 1.6, + }, + }, + { + // incrementRun + Nums: []float64{ + 1.5, + 1.5, + 1.5, + }, + }, + { + // endSeries run + Nums: []float64{ + 1.5, + 1.5, + 1.5, + 1.6, + }, + }, + { + // incrementLiteral + Nums: []float64{ + 1.5, + 1.6, + }, + }, + { + // ... + Nums: []float64{ + 1.5, + 1.5, + 1.5, // + 1.5, // + 1.6, // + 1.7, // + 1.8, // + 1.9, // + 2.0, // + 2.0, // + 2.0, // + 2.2, // + }, + }, + } + ) + for caseIdx, testCase := range testCases { + var ( + fracDigits byte = 1 + buf = make([]byte, 16) + payloadSize = 0 + decodedNums []float64 + ) + c := NewCumulativeDeltaCompressor(buf, payloadSize, fracDigits) + for _, num := range testCase.Nums { + compressionWay, _ := c.Evaluate(num) + c.Compress(compressionWay, num) + } + + // + c.CaptureState() + + fmt.Println("---------------") + + d := c.CreateDecompressor() + + //fmt.Println(buf) + + for { + num, done := d.NextValue() + //fmt.Println(num, done) + if done { + break + } + decodedNums = append(decodedNums, num) + } + + slices.Reverse(decodedNums) + + fmt.Println(testCase.Nums) + fmt.Println(decodedNums) + fmt.Println() + + if !equalFloatSlices(testCase.Nums, decodedNums, 0.0000001) { + t.Fatalf("%d: got nums %v not equal expected %v", caseIdx, decodedNums, testCase.Nums) + } + } +} + +// INSTANT + +func TestInstantDeltaCompressor(t *testing.T) { + var ( + testCases = []struct { + Nums []float64 + RepeatLastDeltaNTimes int + Code int + RequiredSpace int + Buf []byte + BaseValue float64 + LastDelta int64 + H byte + }{ + { + Nums: []float64{ + 1.5, + }, + Code: addBaseValue, + RequiredSpace: 3, // base value, delta, h + Buf: []byte{ + 158, 0, 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 128, // literal, length = 1 + }, + { + Nums: []float64{ + 1.5, + 1.5, + }, + Code: startRun, // literal changed to run + RequiredSpace: 2, // delta, h + Buf: []byte{ + 158, 0, 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 0, // run, length = 2 + }, + { + Nums: []float64{ + 1.5, + 1.6, + 1.6, + }, + Code: startRun, // literal decreased by 1 + RequiredSpace: 3, // end of literal, new delta, new h + Buf: []byte{ + 158, // base value + 128, // literal 1st delta (0) + 128, // h - end of literal, length = 1 + 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 1, + H: 0, // run, length = 2 + }, + { + // same as prev, but negative delta + Nums: []float64{ + 1.5, + 1.4, + 1.4, + }, + Code: startRun, // literal decreased by 1 + RequiredSpace: 3, // end of literal, new delta, new h + Buf: []byte{ + 158, // base value + 128, // literal 1st delta (0) + 128, // h - end of literal, length = 1 + 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: -1, + H: 0, // run, length = 2 + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, + }, + Code: incrementRun, + RequiredSpace: 2, + Buf: []byte{ + 158, // base value + 0, 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 1, // run, length = 3 + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, + 1.6, + }, + Code: endSeries, // end of run + RequiredSpace: 4, + Buf: []byte{ + 158, // base value + 128, // run delta (0) + 1, // h of run, length = 3 + 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 1, + H: 128, // literal, length = 1 + }, + { + // same as prev, but negative delta + Nums: []float64{ + 1.5, + 1.5, + 1.5, + 1.4, + }, + Code: endSeries, // end of run + RequiredSpace: 4, + Buf: []byte{ + 158, // base value + 128, // run delta (0) + 1, // h of run, length = 3 + 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: -1, + H: 128, // literal, length = 1 + }, + { + Nums: []float64{ + 1.5, + }, + RepeatLastDeltaNTimes: 128, + Code: incrementRun, // h byte full filled + RequiredSpace: 2, + Buf: []byte{ + 158, // base value + 0, 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 127, // run, length = 129 + }, + { + Nums: []float64{ + 1.5, + }, + RepeatLastDeltaNTimes: 129, + Code: endSeries, // run, h byte overflowed + RequiredSpace: 4, // run delta, h of run, 1st literal delta, h of literal + Buf: []byte{ + 158, // base value + 128, // run delta (0) + 127, // h of run (length = 129) + 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 0, + H: 128, // literal, length = 1 + }, + { + Nums: []float64{ + 1.5, + 1.6, + }, + Code: incrementLiteral, + RequiredSpace: 3, // previous delta, new delta, h + Buf: []byte{ + 158, // base value + 128, // literal 1st delta (0) + 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: 1, + H: 129, // literal, length = 2 + }, + { + // same as prev, but negative delta + Nums: []float64{ + 1.5, + 1.4, + }, + Code: incrementLiteral, + RequiredSpace: 3, // previous delta, new delta, h + Buf: []byte{ + 158, // base value + 128, // literal 1st delta (0) + 0, 0, 0, 0, 0, 0, + }, + BaseValue: 1.5, + LastDelta: -1, + H: 129, // literal, length = 2 + }, + } + ) + + for caseIdx, testCase := range testCases { + var ( + fracDigits byte = 1 + buf = make([]byte, 8) + payloadSize = 0 + compressionWay int + requiredSpace int + ) + c := NewInstantDeltaCompressor(buf, payloadSize, fracDigits) + for _, num := range testCase.Nums { + compressionWay, requiredSpace = c.Evaluate(num) + c.Compress(compressionWay, num) + } + if testCase.RepeatLastDeltaNTimes > 0 { + num := testCase.Nums[len(testCase.Nums)-1] + for range testCase.RepeatLastDeltaNTimes { + compressionWay, requiredSpace = c.Evaluate(num) + c.Compress(compressionWay, num) + } + } + if compressionWay != testCase.Code { + t.Fatalf("%d: got code %d are not equal expected %d", + caseIdx, compressionWay, testCase.Code) + } + if requiredSpace != testCase.RequiredSpace { + t.Fatalf("%d: got requiredSpace %d are not equal expected %d", + caseIdx, requiredSpace, testCase.RequiredSpace) + } + if !bytes.Equal(buf, testCase.Buf) { + t.Fatalf("%d: got buf %v are not equal expected %v", + caseIdx, buf, testCase.Buf) + } + if c.baseValue != testCase.BaseValue { + t.Fatalf("%d: got lastUnixtime %v are not equal expected %v", + caseIdx, c.baseValue, testCase.BaseValue) + } + if c.lastDelta != testCase.LastDelta { + t.Fatalf("%d: got lastDelta %d are not equal expected %d", + caseIdx, c.lastDelta, testCase.LastDelta) + } + if c.h != testCase.H { + t.Fatalf("%d: got h %d are not equal expected %d", + caseIdx, c.h, testCase.H) + } + } +} + +func TestInstantDeltaDecompressorFromState(t *testing.T) { + var ( + testCases = []struct { + Nums []float64 + //RepeatLastDeltaNTimes int + }{ + { + // addBaseValue + Nums: []float64{ + 1.5, + }, + }, + { + // startRun + Nums: []float64{ + 1.5, + 1.5, + }, + }, + { + // startRun + Nums: []float64{ + 1.5, + 1.6, + 1.6, + }, + }, + { + // incrementRun + Nums: []float64{ + 1.5, + 1.5, + 1.5, + }, + }, + { + // endSeries run + Nums: []float64{ + 1.5, + 1.5, + 1.5, + 1.6, + }, + }, + { + // incrementLiteral + Nums: []float64{ + 1.5, + 1.6, + }, + }, + { + // ... + Nums: []float64{ + 1.5, + 1.5, + 1.5, // + 1.5, // + 1.6, // + 1.7, // + 1.8, // + 1.9, // + 2.0, // + 2.0, // + 2.0, // + 2.2, // + }, + }, + } + ) + for caseIdx, testCase := range testCases { + var ( + fracDigits byte = 1 + buf = make([]byte, 16) + payloadSize = 0 + decodedNums []float64 + ) + c := NewInstantDeltaCompressor(buf, payloadSize, fracDigits) + for _, num := range testCase.Nums { + compressionWay, _ := c.Evaluate(num) + c.Compress(compressionWay, num) + } + + // + c.CaptureState() + + fmt.Println("---------------") + + d := c.CreateDecompressor() + + //fmt.Println(buf) + + for { + num, done := d.NextValue() + //fmt.Println(num, done) + if done { + break + } + decodedNums = append(decodedNums, num) + } + + slices.Reverse(decodedNums) + + fmt.Println(testCase.Nums) + fmt.Println(decodedNums) + fmt.Println() + + if !equalFloatSlices(testCase.Nums, decodedNums, 0.0000001) { + t.Fatalf("%d: got nums %v not equal expected %v", caseIdx, decodedNums, testCase.Nums) + } + } +} + +// TIME DELTA + +func TestTimeDeltaCompressor(t *testing.T) { + var ( + testCases = []struct { + Nums []uint32 + RepeatLastDeltaNTimes int + CompressionWay int + RequiredSpace int + Buf []byte + LastUnixtime uint32 + LastDelta uint32 + H byte + }{ + { + Nums: []uint32{ + 1780777000, + }, + CompressionWay: addUnixtime, + RequiredSpace: 4, + Buf: []byte{ + 0, 0, 0, 0, 0, 0, 0, 0, + }, + LastUnixtime: 1780777000, + LastDelta: 0, + H: 0, + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + }, + CompressionWay: add1stDelta, + RequiredSpace: 6, + Buf: []byte{ + 0, 0, 0, 0, 0, 0, 0, 0, + }, + LastUnixtime: 1780777060, + LastDelta: 60, + H: 128, + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 + }, + CompressionWay: startRun, // literal changed by run + RequiredSpace: 6, + Buf: []byte{ + 0, 0, 0, 0, 0, 0, 0, 0, + }, + LastUnixtime: 1780777120, + LastDelta: 60, + H: 0, // run, length = 2 + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777130, // +70 + 1780777200, // +70 + }, + CompressionWay: startRun, // literal decreased by 1 + RequiredSpace: 7, // unixtime, end of literal, new delta, new h + Buf: []byte{ + 0, 0, 0, 0, 0, 0, 128, 188, + }, + LastUnixtime: 1780777200, + LastDelta: 70, + H: 0, // run, length = 2 + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 + 1780777180, // +60 + }, + CompressionWay: incrementRun, + RequiredSpace: 6, + Buf: []byte{ + 0, 0, 0, 0, 0, 0, 0, 0, + }, + LastUnixtime: 1780777180, + LastDelta: 60, + H: 1, // run, length = 3 + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 + 1780777180, // +60 + 1780777200, // +20 + }, + CompressionWay: endSeries, // run + RequiredSpace: 8, + Buf: []byte{ + 0, 0, 0, 0, 0, 0, + 1, // h of run, length = 3 + 188, // varUint64(60) + }, + LastUnixtime: 1780777200, + LastDelta: 20, + H: 128, // literal, length = 1 + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + }, + RepeatLastDeltaNTimes: 128, + CompressionWay: incrementRun, // h byte full filled + RequiredSpace: 6, + Buf: []byte{ + 0, 0, 0, 0, 0, 0, 0, 0, + }, + LastUnixtime: 1780777060 + 128*60, + LastDelta: 60, + H: 127, // run, length = 129 + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + }, + RepeatLastDeltaNTimes: 129, + CompressionWay: endSeries, // run, h byte overflowed + RequiredSpace: 8, + Buf: []byte{ + 0, 0, 0, 0, 0, 0, + 127, // h - end of run (length = 129) + 188, // varUint64(60) + }, + LastUnixtime: 1780777060 + 129*60, + LastDelta: 60, + H: 128, // literal, length = 1 + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777122, // +62 + }, + CompressionWay: incrementLiteral, + RequiredSpace: 7, + Buf: []byte{ + 0, 0, 0, 0, 0, 0, 0, + 188, // varUint64(60) + }, + LastUnixtime: 1780777122, + LastDelta: 62, + H: 129, // literal, length = 2 + }, + } + ) + + for caseIdx, testCase := range testCases { + //fmt.Println("----------") + var ( + buf = make([]byte, 8) + compressionWay int + requiredSpace int + ) + c := NewTimeDeltaCompressor(buf, 0) + for _, num := range testCase.Nums { + compressionWay, requiredSpace = c.Evaluate(num) + c.Compress(compressionWay, num) + } + if testCase.RepeatLastDeltaNTimes > 0 { + for range testCase.RepeatLastDeltaNTimes { + num := c.lastUnixtime + c.lastDelta + compressionWay, requiredSpace = c.Evaluate(num) + c.Compress(compressionWay, num) + } + } + if compressionWay != testCase.CompressionWay { + t.Fatalf("%d: got code %d are not equal expected %d", + caseIdx, compressionWay, testCase.CompressionWay) + } + if requiredSpace != testCase.RequiredSpace { + t.Fatalf("%d: got requiredSpace %d are not equal expected %d", + caseIdx, requiredSpace, testCase.RequiredSpace) + } + if !bytes.Equal(buf, testCase.Buf) { + t.Fatalf("%d: got buf %v are not equal expected %v", + caseIdx, buf, testCase.Buf) + } + if c.lastUnixtime != testCase.LastUnixtime { + t.Fatalf("%d: got lastUnixtime %d are not equal expected %d", + caseIdx, c.lastUnixtime, testCase.LastUnixtime) + } + if c.lastDelta != testCase.LastDelta { + t.Fatalf("%d: got lastDelta %d are not equal expected %d", + caseIdx, c.lastDelta, testCase.LastDelta) + } + if c.h != testCase.H { + t.Fatalf("%d: got h %d are not equal expected %d", + caseIdx, c.h, testCase.H) + } + } +} + +func TestTimeDeltaDecompressorFromState(t *testing.T) { + var ( + testCases = []struct { + Nums []uint32 + //RepeatLastDeltaNTimes int + }{ + { + Nums: []uint32{ + 1780777000, + }, + }, + { + // add1stDelta + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + }, + }, + { + // startRun + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 + }, + }, + { + // incrementRun + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 + 1780777180, // +60 + }, + }, + { + // endRun + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 + 1780777180, // +60 + 1780777200, // +20 + }, + }, + { + // incrementLiteral + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777122, // +62 + }, + }, + { + // ... + Nums: []uint32{ + 1780777000, + 1780777010, // +10 + 1780777120, // +10 + 1780777130, // +10 + 1780777135, // +5 + 1780777141, // +6 + 1780777148, // +7 + 1780777156, // +8 + 1780777165, // +9 + 1780777174, // +9 + 1780777183, // +9 + 1780777190, // +7 + }, + }, + } + ) + for caseIdx, testCase := range testCases { + var ( + buf = make([]byte, 16) + decodedNums []uint32 + ) + c := NewTimeDeltaCompressor(buf, 0) + for _, num := range testCase.Nums { + compressionWay, _ := c.Evaluate(num) + c.Compress(compressionWay, num) + } + + // + c.CaptureState() + + d := c.CreateDecompressor() + + //fmt.Println(testCase.Nums) + // fmt.Println("---------------") + // fmt.Println(buf) + + for { + num, done := d.NextValue() + //fmt.Println(num, done) + if done { + break + } + decodedNums = append(decodedNums, num) + } + + slices.Reverse(decodedNums) + + if !slices.Equal(testCase.Nums, decodedNums) { + t.Fatalf("%d: got nums %v not equal expected %v", caseIdx, decodedNums, testCase.Nums) + } + } +} + +func TestVarUint64(t *testing.T) { + arr := make([]byte, 9) + n, err := bin.PutVarInt64(arr, 15) + if err != nil { + t.Fatal(err) + } + fmt.Println(arr[:n]) +} diff --git a/enc/insdelta.go b/enc/insdelta.go index a341e19..4782378 100644 --- a/enc/insdelta.go +++ b/enc/insdelta.go @@ -1,130 +1,496 @@ package enc import ( - "fmt" + "log" "math" - octopus "gordenko.dev/dima/qb" - "gordenko.dev/dima/qb/bin" + bin "gordenko.dev/dima/bin/little" + "gordenko.dev/dima/qb" ) -type ReverseInstantDeltaDecompressor struct { - buf []byte - pos int - bound int - firstValue float64 - lastValue float64 - length uint16 - coef float64 - idxOf8 uint - s8 byte +type InstantDeltaCompressor struct { + buf []byte + coef float64 + pos int + baseValue float64 + lastDelta int64 + h byte + state *InstantDeltaCapturedState } -func NewReverseInstantDeltaDecompressor(buf []byte, fracDigits byte) *ReverseInstantDeltaDecompressor { +func NewInstantDeltaCompressor(buf []byte, payloadSize int, fracDigits byte) *InstantDeltaCompressor { var coef float64 = 1 if fracDigits > 0 { coef = math.Pow(10, float64(fracDigits)) } - return &ReverseInstantDeltaDecompressor{ + s := &InstantDeltaCompressor{ buf: buf, + pos: payloadSize, coef: coef, - pos: len(buf), + } + if payloadSize > 0 { + u64, _, err := bin.GetVarInt64(buf) + if err != nil { + log.Fatalf("bug: get base value: %s", err) + } + s.baseValue = float64(u64) / s.coef + s.pos-- + s.h = s.buf[s.pos] + var n int + s.lastDelta, n, err = bin.ReverseGetVarInt64(s.buf[:s.pos]) + if err != nil { + log.Fatalf("bug: get last delta: %s", err) + } + s.pos -= n + } + return s +} + +// func (s *InstantDeltaCompressor) Size() int { +// return s.pos +// } + +func (s *InstantDeltaCompressor) Evaluate(value float64) (compressionWay int, requiredSpace int) { + tmp := (value - s.baseValue) * s.coef + if tmp > 0 { + tmp += eps + } else { + tmp -= eps + } + delta := int64(tmp) + if s.pos > 0 { + if s.h < 128 { + // run + if delta == s.lastDelta && s.h < 127 { + compressionWay = incrementRun + requiredSpace += bin.CountVarInt64(s.lastDelta) + // current delta + hSize + } else { + compressionWay = endSeries + requiredSpace += bin.CountVarInt64(s.lastDelta) + // previous delta + hSize + // h - end of run + bin.CountVarInt64(delta) + // new literal + hSize // new h + } + } else { + // literal + if delta != s.lastDelta { + if s.h < 255 { + compressionWay = incrementLiteral + requiredSpace += bin.CountVarInt64(s.lastDelta) + // previous delta + bin.CountVarInt64(delta) + // new delta + hSize + } else { + compressionWay = endSeries + requiredSpace += bin.CountVarInt64(s.lastDelta) + // previous delta + hSize + // h - end of literal + bin.CountVarInt64(delta) + // new literal + hSize // new h + } + } else { + compressionWay = startRun + if s.h > 128 { + requiredSpace += hSize // h - end of literal + } + requiredSpace += bin.CountVarInt64(delta) + // new delta + hSize // new h + } + } + } else { + // encode base value + compressionWay = addBaseValue + requiredSpace = bin.CountVarInt64(int64(value*s.coef)) + // base value + bin.CountVarInt64(delta) + // new delta + +hSize // new h + } + return +} + +func (s *InstantDeltaCompressor) Compress(compressionWay int, value float64) { + tmp := (value - s.baseValue) * s.coef + if tmp > 0 { + tmp += eps + } else { + tmp -= eps + } + delta := int64(tmp) + switch compressionWay { + case incrementRun: + s.h++ + case incrementLiteral: + // write previous delta and increment counter + n, _ := bin.ReversePutVarInt64(s.buf[s.pos:], s.lastDelta) + s.pos += n + s.lastDelta = delta + s.h++ + case endSeries: + n, _ := bin.ReversePutVarInt64(s.buf[s.pos:], s.lastDelta) + s.pos += n + // write h - end of run + s.buf[s.pos] = s.h + s.pos++ + // start new literal (length=1) + s.lastDelta = delta + s.h = 128 + case startRun: + if s.h > 128 { + // write h - end of literal (because length > 1) + s.h-- + s.buf[s.pos] = s.h + s.pos++ + } + // start new run (length=2) + s.h = 0 + case addBaseValue: + // start new literal (length=1) + n, _ := bin.PutVarInt64(s.buf[s.pos:], int64(value*s.coef)) + s.pos += n + s.baseValue = value + // start new literal (length=1) + s.lastDelta = 0 + s.h = 128 } } -func (s *ReverseInstantDeltaDecompressor) NextValue() (value float64, done bool) { - if s.length > 0 { - s.length-- - return s.lastValue, false +func (s *InstantDeltaCompressor) DeleteLast() {} + +type InstantDeltaCapturedState struct { + H byte + LastDelta int64 + Payload []byte +} + +func (s *InstantDeltaCompressor) CaptureState() { + if s.state != nil { + qb.Abort(qb.RepeatableLock, nil) } - if s.pos < s.bound { + // позиція посувається вліво, отже може перескочити на попередній chunk + s.state = &InstantDeltaCapturedState{ + H: s.h, + LastDelta: s.lastDelta, + Payload: s.buf[:s.pos], + } +} + +// fix - повернути в Pool буфери +func (s *InstantDeltaCompressor) ForgetCapturedState() { + s.state = nil +} + +func (s *InstantDeltaCompressor) Offset() int { + // if s.state != nil { + // return s.state.Pos + // } + return 0 +} + +// Snapshot - для створення снапшота. +func (s *InstantDeltaCompressor) Snapshot() (left []byte, right []byte) { + if s.state == nil { + left = s.buf[:s.pos] + right = s.encodeTail(s.lastDelta, s.h) + } else { + left = s.state.Payload + right = s.encodeTail(s.state.LastDelta, s.state.H) + } + return +} + +func (s *InstantDeltaCompressor) encodeTail(lastDelta int64, h byte) []byte { + tail := make([]byte, 10) // max var uint64 + h + n, _ := bin.PutVarInt64(tail, lastDelta) + tail[n] = h + return tail[:n+1] +} + +func (s *InstantDeltaCompressor) Rotate(newbuf []byte) (payloadSize int) { + n, _ := bin.PutVarInt64(s.buf[s.pos:], s.lastDelta) + s.pos += n + s.buf[s.pos] = s.h + s.pos++ + payloadSize = s.pos + // УВАГА! + // state не чіпаємо + s.buf = newbuf + s.pos = 0 + s.baseValue = 0 + s.lastDelta = 0 + s.h = 0 + return +} + +func (s *InstantDeltaCompressor) CreateDecompressor() qb.ValueDecompressor { + var ( + h = s.h + lastDelta = s.lastDelta + payload = s.buf[:s.pos] + ) + if s.state != nil { + h = s.state.H + lastDelta = s.state.LastDelta + payload = s.state.Payload + } + return NewInstantDeltaDecompressorFromState(InstantDeltaDecompressorFromStateOptions{ + Coef: s.coef, + H: h, + LastDelta: lastDelta, + Payload: payload, + }) +} + +func (s *InstantDeltaCompressor) Chunks() []byte { + return s.buf +} + +// DECOMPRESSOR + +type InstantDeltaDecompressor struct { + buf []byte + coef float64 + pos int + bound int + baseValue float64 + lastValue float64 + isRun bool + pending int + done bool +} + +func NewInstantDeltaDecompressor(buf []byte, fracDigits byte) *InstantDeltaDecompressor { + var coef float64 = 1 + if fracDigits > 0 { + coef = math.Pow(10, float64(fracDigits)) + } + i64, n, err := bin.GetVarInt64(buf) + if err != nil { + log.Fatalf("bug: get base value: %s", err) + } + //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) + s := &InstantDeltaDecompressor{ + buf: buf, + coef: coef, + pos: len(buf), // first free + baseValue: float64(i64) / coef, + bound: n, + } + s.readHeader() + s.readValue() + return s +} + +type InstantDeltaDecompressorFromStateOptions struct { + Coef float64 + H byte + LastDelta int64 + Payload []byte +} + +func NewInstantDeltaDecompressorFromState(opt InstantDeltaDecompressorFromStateOptions) *InstantDeltaDecompressor { + i64, n, err := bin.GetVarInt64(opt.Payload) + if err != nil { + log.Fatalf("bug: get base value: %s", err) + } + //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) + s := &InstantDeltaDecompressor{ + buf: opt.Payload, + coef: opt.Coef, + pos: len(opt.Payload), // first free + baseValue: float64(i64) / opt.Coef, + bound: n, + } + s.lastValue = s.baseValue + float64(opt.LastDelta)/s.coef + s.decodeHeaderByte(opt.H) + //fmt.Printf("restore from bound: isRun=%t, pending=%d\n", s.isRun, s.pending) + return s +} + +func (s *InstantDeltaDecompressor) NextValue() (value float64, done bool) { + //fmt.Printf("NextValue(): bound: %d, pos: %d, pending: %d\n", s.bound, s.pos, s.pending) + if s.done { return 0, true } - - if s.pos == len(s.buf) { - u64, n, err := bin.GetVarInt64(s.buf) - if err != nil { - panic(err) + // метод працює як do while - спочатку значення, а потім перевірка умови + value = s.lastValue + s.pending-- + if s.pending > 0 { + // якщо в серії залишаються елементи + if !s.isRun { + s.readValue() } - s.firstValue = float64(u64) / s.coef - s.bound = n - s.pos-- - s.idxOf8 = uint(8 - s.buf[s.pos]) - s.pos-- - s.s8 = s.buf[s.pos] - s.pos-- - s.readVar() - if s.length > 0 { - s.length-- - } - return s.lastValue, false - } - - if s.idxOf8 == 0 { - s.s8 = s.buf[s.pos] - s.pos-- - } - s.readVar() - if s.length > 0 { - s.length-- - } - return s.lastValue, false -} - -func (s *ReverseInstantDeltaDecompressor) readVar() { - i64, n, err := bin.ReverseGetVarInt64(s.buf[:s.pos+1]) - 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.decodeLength() - } - if s.idxOf8 == 7 { - s.idxOf8 = 0 + } else if s.pos > s.bound { + // в серії більше немає елементів, отже перевіряє чи є ще дані в буфері. + // дані є - читаю заголовок наступної серії + s.readHeader() + s.readValue() } else { - s.idxOf8++ + s.done = true } + return value, false } -func (s *ReverseInstantDeltaDecompressor) decodeLength() { - b1 := s.buf[s.pos] +func (s *InstantDeltaDecompressor) readHeader() { s.pos-- - if b1 < 128 { - s.length = uint16(b1) + h := s.buf[s.pos] + s.decodeHeaderByte(h) + // fmt.Println("h:", h) + // fmt.Println("isRun:", s.isRun) + // fmt.Println("pending:", s.pending) +} +func (s *InstantDeltaDecompressor) readValue() { + i64, n, err := bin.ReverseGetVarInt64(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.lastValue = s.baseValue + float64(i64)/s.coef +} + +func (s *InstantDeltaDecompressor) decodeHeaderByte(h byte) { + s.isRun = h < 128 + if s.isRun { + s.pending = int(h&127) + 2 } else { - b2 := s.buf[s.pos] - s.pos-- - s.length = uint16(b1&127) | (uint16(b2) << 7) - } - s.length += 2 -} - -func GetValueBounds(valuesBuf []byte, metricType octopus.MetricType, fracDigits byte) (sinceValue, untilValue float64) { - var decompressor octopus.ValueDecompressor - switch metricType { - case octopus.Instant: - decompressor = NewReverseInstantDeltaDecompressor(valuesBuf, fracDigits) - case octopus.Cumulative: - decompressor = NewReverseCumulativeDeltaDecompressor(valuesBuf, fracDigits) - default: - panic(fmt.Sprintf("unknown metricType %d", metricType)) - } - value, done := decompressor.NextValue() - if done { - return - } - - sinceValue = value - untilValue = value - for { - value, done = decompressor.NextValue() - if done { - return - } - sinceValue = value + s.pending = int(h&127) + 1 } } + +// DECOMPRESSOR + +// type ReverseInstantDeltaDecompressor struct { +// buf *conbuf.ContinuousBuffer +// pos int +// bound int +// baseValue float64 +// lastValue float64 +// coef float64 +// isRun bool +// pending int +// } + +// func NewReverseInstantDeltaDecompressor(buf *conbuf.ContinuousBuffer, size int, fracDigits byte) *ReverseInstantDeltaDecompressor { +// var coef float64 = 1 +// if fracDigits > 0 { +// coef = math.Pow(10, float64(fracDigits)) +// } +// u64, n, err := buf.GetVarInt64(0) +// if err != nil { +// log.Fatalf("bug: get base value: %s", err) +// } +// //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) +// return &ReverseInstantDeltaDecompressor{ +// buf: buf, +// coef: coef, +// pos: size - 1, // last elem +// baseValue: float64(u64) / coef, +// bound: n, +// } +// } + +// func (s *ReverseInstantDeltaDecompressor) NextValue() (value float64, done bool) { +// //fmt.Printf("NextValue(): bound: %d, pos: %d, pending: %d\n", s.bound, s.pos, s.pending) +// if s.pending > 0 { +// // якщо в серії залишаються елементи +// if !s.isRun { +// s.readValue() +// } +// s.pending-- +// return s.lastValue, false +// } +// if s.pos > s.bound { +// // читаю заголовок наступної серії +// s.readHeader() +// s.readValue() +// s.pending-- +// return s.lastValue, false +// } +// // серія завершена - перевіряю чи є ще серії +// return 0, true +// } + +// func (s *ReverseInstantDeltaDecompressor) readHeader() { +// h := s.buf.GetByte(s.pos) +// s.pos-- +// s.isRun = h < 128 +// if s.isRun { +// s.pending = int(h&127) + 2 +// } else { +// s.pending = int(h&127) + 1 +// } +// // fmt.Println("h:", h) +// // fmt.Println("isRun:", s.isRun) +// // fmt.Println("pending:", s.pending) +// } +// func (s *ReverseInstantDeltaDecompressor) readValue() { +// i64, n, err := s.buf.ReverseGetVarInt64(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.lastValue = s.baseValue + float64(i64)/s.coef +// } + +// func (s *InstantDeltaCompressor) Append(value float64) { +// if s.pos == 0 { +// // base value +// n, _ := bin.PutVarInt64(s.buf[s.pos:], int64(value*s.coef)) +// s.pos += n +// s.baseValue = value +// s.appendNewLiteral(0) +// } else { +// tmp := (value - s.baseValue) * s.coef +// if tmp > 0 { +// tmp += eps +// } else { +// tmp -= eps +// } +// delta := int64(tmp) +// 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 +// // fix encode delta +// s.appendNewLiteral(delta) +// } +// } else { +// // literal блок. +// // Якщо в ньому лише одне значення - перетворюю його на run блок. +// // Інакше забираю останнє значення із literal блока і додаю новий run блок. +// q := s.h & 127 +// if q == 0 { // 1 кодується як 0 +// s.convertLiteralToRun() +// } else { +// s.convertLastFromLiteralToRun() +// } +// } +// } else { +// if s.h < 127 { +// // 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) +// } +// } +// } +// } +// } diff --git a/chunkenc/time_delta.go b/enc/time_delta.go similarity index 68% rename from chunkenc/time_delta.go rename to enc/time_delta.go index 088f4cb..d221a1c 100644 --- a/chunkenc/time_delta.go +++ b/enc/time_delta.go @@ -1,4 +1,4 @@ -package chunkenc +package enc import ( "log" @@ -17,18 +17,6 @@ Timestamps читаються звичайними методами Get*, а п */ -const ( - hSize = 1 - - // append scripts - addUnixtime = 0 - incrementRun = 1 - incrementLiteral = 2 - endSeries = 3 - startRun = 5 - add1stDelta = 6 -) - type TimeDeltaCompressor struct { buf []byte // останній записаний байт, якщо рахувати справа наліво @@ -40,12 +28,12 @@ type TimeDeltaCompressor struct { } // Кодує значення справа наліво футнкціями TailPut*. Читає значення зліва направо функціями Get* -func NewTimeDeltaCompressor(buf []byte, size int) *TimeDeltaCompressor { +func NewTimeDeltaCompressor(buf []byte, payloadSize int) *TimeDeltaCompressor { s := &TimeDeltaCompressor{ buf: buf, - pos: len(buf) - size, + pos: len(buf) - payloadSize, } - if size > 0 { + if payloadSize > 0 { var err error s.lastUnixtime, err = bin.GetUint32(s.buf[s.pos:]) if err != nil { @@ -66,14 +54,13 @@ func NewTimeDeltaCompressor(buf []byte, size int) *TimeDeltaCompressor { return s } -func (s *TimeDeltaCompressor) Size() int { - return len(s.buf) - s.pos -} +// func (s *TimeDeltaCompressor) Size() int { +// return len(s.buf) - s.pos +// } // retrun -func (s *TimeDeltaCompressor) CalcRequiredSpace(unixtime uint32) (code int, size int) { - //fmt.Println(unixtime) - size = 4 // last unixtime +func (s *TimeDeltaCompressor) Evaluate(unixtime uint32) (compressionWay int, requiredSpace int) { + requiredSpace = 4 // last unixtime if s.lastUnixtime > 0 { delta := unixtime - s.lastUnixtime if s.lastDelta > 0 { @@ -81,11 +68,12 @@ func (s *TimeDeltaCompressor) CalcRequiredSpace(unixtime uint32) (code int, size if s.h < 128 { // run if delta == s.lastDelta && s.h < 127 { - code = incrementRun - size += hSize + compressionWay = incrementRun + requiredSpace += bin.CountVarUint64(uint64(s.lastDelta)) + // current delta + hSize } else { - code = endSeries - size += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta + compressionWay = endSeries + requiredSpace += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta hSize + // h - end of run bin.CountVarUint64(uint64(delta)) + // new literal hSize // new h @@ -94,40 +82,40 @@ func (s *TimeDeltaCompressor) CalcRequiredSpace(unixtime uint32) (code int, size // literal if delta != s.lastDelta { if s.h < 255 { - code = incrementLiteral - size += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta + compressionWay = incrementLiteral + requiredSpace += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta bin.CountVarUint64(uint64(delta)) + // new delta hSize } else { - code = endSeries - size += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta + compressionWay = endSeries + requiredSpace += bin.CountVarUint64(uint64(s.lastDelta)) + // previous delta hSize + // h - end of literal bin.CountVarUint64(uint64(delta)) + // new literal hSize // new h } } else { - code = startRun + compressionWay = startRun if s.h > 128 { - size += hSize // h - end of literal + requiredSpace += hSize // h - end of literal } - size += bin.CountVarUint64(uint64(delta)) + // new delta + requiredSpace += bin.CountVarUint64(uint64(delta)) + // new delta hSize // new h } } } else { - code = add1stDelta - size += bin.CountVarUint64(uint64(delta)) + // new literal + compressionWay = add1stDelta + requiredSpace += bin.CountVarUint64(uint64(delta)) + // new literal hSize // new h } } else { - code = addUnixtime + compressionWay = addUnixtime } return } -func (s *TimeDeltaCompressor) Append(code int, unixtime uint32) { +func (s *TimeDeltaCompressor) Compress(compressionWay int, unixtime uint32) { delta := unixtime - s.lastUnixtime - switch code { + switch compressionWay { case incrementRun: s.h++ case incrementLiteral: @@ -159,7 +147,8 @@ func (s *TimeDeltaCompressor) Append(code int, unixtime uint32) { s.lastDelta = delta s.h = 128 } - // 1st value + // 1st value or update after every update + // do not write immediatelly because of update after every update s.lastUnixtime = unixtime } @@ -167,18 +156,13 @@ func (s *TimeDeltaCompressor) DeleteLast() { } -func (s *TimeDeltaCompressor) Sync() { - //n, _ := bin.ReversePutVarUint64(s.buf[s.pos:], uint64(s.lastUnixtime)) - //s.pos += n -} - // FIX - check methods type TimeDeltaCapturedState struct { H byte LastUnixtime uint32 LastDelta uint32 - Buf []byte + Payload []byte } // delta h @@ -191,7 +175,7 @@ func (s *TimeDeltaCompressor) CaptureState() { H: s.h, LastUnixtime: s.lastUnixtime, LastDelta: s.lastDelta, - Buf: s.buf[s.pos:], + Payload: s.buf[s.pos:], } } @@ -209,14 +193,13 @@ func (s *TimeDeltaCompressor) Offset() int { } // Snapshot - для створення снапшота. -// return tail, head -func (s *TimeDeltaCompressor) Snapshot() (tail []byte, head []byte) { +func (s *TimeDeltaCompressor) Snapshot() (left []byte, right []byte) { if s.state == nil { - tail = s.encodeTail(s.lastUnixtime, s.lastDelta, s.h) - head = s.buf[s.pos:] + left = s.encodeTail(s.lastUnixtime, s.lastDelta, s.h) + right = s.buf[s.pos:] } else { - tail = s.encodeTail(s.state.LastUnixtime, s.state.LastDelta, s.state.H) - head = s.state.Buf + left = s.encodeTail(s.state.LastUnixtime, s.state.LastDelta, s.state.H) + right = s.state.Payload } return } @@ -224,26 +207,49 @@ func (s *TimeDeltaCompressor) Snapshot() (tail []byte, head []byte) { func (s *TimeDeltaCompressor) encodeTail(lastUnixtime, lastDelta uint32, h byte) []byte { tail := make([]byte, 9) bin.PutUint32(tail, lastUnixtime) - n, _ := bin.PutVarUint64(tail[4:], uint64(lastDelta)) - tail[n+4] = h + tail[4] = h + n, _ := bin.PutVarUint64(tail[5:], uint64(lastDelta)) return tail[:n+5] } -func (s *TimeDeltaCompressor) CreateDecompressor() qb.TimestampDecompressor { - if s.state == nil { - return NewTimeDeltaDecompressor(s.buf[s.pos:]) - } - return NewTimeDeltaDecompressorFromCapturedState(*s.state) -} - -func (s *TimeDeltaCompressor) Renew() { +func (s *TimeDeltaCompressor) Rotate(newbuf []byte) (payloadSize int) { + var n int + n, _ = bin.TailPutVarUint64(s.buf[:s.pos], uint64(s.lastDelta)) + s.pos -= n + s.pos-- + s.buf[s.pos] = s.h + n, _ = bin.TailPutVarUint64(s.buf[:s.pos], uint64(s.lastUnixtime)) + s.pos -= n + payloadSize = len(s.buf) - s.pos // УВАГА! // state не чіпаємо - s.buf = make([]byte, minBufferSize) + s.buf = newbuf s.pos = len(s.buf) s.lastUnixtime = 0 s.lastDelta = 0 s.h = 0 + return +} + +func (s *TimeDeltaCompressor) CreateDecompressor() qb.TimestampDecompressor { + var ( + h = s.h + 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) Chunks() []byte { @@ -280,14 +286,21 @@ func NewTimeDeltaDecompressor(buf []byte) *TimeDeltaDecompressor { return s } -func NewTimeDeltaDecompressorFromCapturedState(state TimeDeltaCapturedState) *TimeDeltaDecompressor { +type TimeDeltaDecompressorFromStateOptions struct { + H byte + LastUnixtime uint32 + LastDelta uint32 + Payload []byte +} + +func NewTimeDeltaDecompressorFromState(opt TimeDeltaDecompressorFromStateOptions) *TimeDeltaDecompressor { s := &TimeDeltaDecompressor{ - buf: state.Buf, - lastUnixtime: state.LastUnixtime, + buf: opt.Payload, + lastUnixtime: opt.LastUnixtime, } - if state.LastDelta > 0 { - s.lastDelta = state.LastDelta - s.decodeHeaderByte(state.H) + 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", @@ -362,47 +375,3 @@ func (s *TimeDeltaDecompressor) readDelta() { s.pos += n s.lastDelta = uint32(u64) } - -// func (s *ReverseTimeDeltaCompressor) Append(unixtime uint32) { -// if s.lastUnixtime > 0 { -// delta := unixtime - s.lastUnixtime -// if s.lastDelta > 0 { -// // 3rd value -// if s.h < 128 { -// // run -// if delta == s.lastDelta && s.h < 127 { -// s.h++ -// } else { -// //write h - end of run -// s.lastDelta = delta -// s.h = 128 // new literal (length=1) -// } -// } else { -// // literal -// if delta == s.lastDelta { -// if s.h > 128 { // -// // write h - end literal series if length > 1 -// } -// // last literal convert to run (length=2) -// // write delta -// s.h = 0 // run, length=2 -// } else { -// // white s.lastDelta -// if s.h < 255 { -// //s.lastDelta = delta -// s.h++ -// } else { -// // write h - end of literal -// s.lastDelta = delta -// s.h = 128 // new literal (length=1) -// } -// } -// } -// } else { -// // 2nd value -// s.lastDelta = delta -// } -// } -// // 1st value -// s.lastUnixtime = unixtime -// } diff --git a/enc/time_delta_delta.go b/enc/time_delta_delta.go deleted file mode 100644 index 90f3809..0000000 --- a/enc/time_delta_delta.go +++ /dev/null @@ -1,145 +0,0 @@ -package enc - -import ( - "gordenko.dev/dima/qb/bin" -) - -// REVERSE - -const ( - lastUnixtimeIdx = 0 - baseDeltaIdx = 4 -) - -type ReverseTimeDeltaOfDeltaDecompressor struct { - step byte - buf []byte - pos int - bound int - lastUnixtime uint32 - baseDelta uint32 - lastDeltaOfDelta int64 - length uint16 - idxOf8 uint - s8 byte -} - -func NewReverseTimeDeltaOfDeltaDecompressor(buf []byte) *ReverseTimeDeltaOfDeltaDecompressor { - return &ReverseTimeDeltaOfDeltaDecompressor{ - buf: buf, - pos: len(buf), - } -} - -func (s *ReverseTimeDeltaOfDeltaDecompressor) NextValue() (value uint32, done bool) { - if s.step == 0 { - if s.pos == 0 { - return 0, true - } - s.lastUnixtime = bin.GetUint32(s.buf[lastUnixtimeIdx:]) - s.step = 1 - return s.lastUnixtime, false - } - - if s.step == 1 { - if s.pos == baseDeltaIdx { - return 0, true - } - u64, n, err := bin.GetVarUint64(s.buf[baseDeltaIdx:]) - if err != nil { - panic("EOF") - } - s.bound = baseDeltaIdx + n - s.baseDelta = uint32(u64) - - s.pos-- - s.idxOf8 = uint(8 - s.buf[s.pos]) - s.pos-- - s.s8 = s.buf[s.pos] - s.pos-- - - s.readVar() - if s.length > 0 { - s.length-- - } - s.step = 2 - return s.lastUnixtime, false - } - - if s.length > 0 { - s.length-- - delta := int64(s.baseDelta) + s.lastDeltaOfDelta - s.lastUnixtime = uint32(int64(s.lastUnixtime) - delta) - return s.lastUnixtime, false - } - - if s.pos < s.bound { - return 0, true - } - if s.idxOf8 == 0 { - s.s8 = s.buf[s.pos] - s.pos-- - } - s.readVar() - if s.length > 0 { - s.length-- - } - return s.lastUnixtime, false -} - -func GetTimeRange(timestampsBuf []byte) (since, until uint32) { - decompressor := NewReverseTimeDeltaOfDeltaDecompressor(timestampsBuf) - value, done := decompressor.NextValue() - if done { - return - } - - since = value - until = value - for { - value, done = decompressor.NextValue() - if done { - return - } - since = value - } -} - -func (s *ReverseTimeDeltaOfDeltaDecompressor) readVar() { - var ( - n int - err error - ) - - s.lastDeltaOfDelta, n, err = bin.ReverseGetVarInt64(s.buf[:s.pos+1]) - if err != nil { - panic(err) - } - s.pos -= n - - delta := int64(s.baseDelta) + s.lastDeltaOfDelta - s.lastUnixtime = uint32(int64(s.lastUnixtime) - delta) - - var flag byte = 1 << s.idxOf8 - if (s.s8 & flag) == flag { - s.decodeLength() - } - if s.idxOf8 == 7 { - s.idxOf8 = 0 - } else { - s.idxOf8++ - } -} - -func (s *ReverseTimeDeltaOfDeltaDecompressor) decodeLength() { - b1 := s.buf[s.pos] - s.pos-- - if b1 < 128 { - s.length = uint16(b1) - } else { - b2 := s.buf[s.pos] - s.pos-- - s.length = uint16(b1&127) | (uint16(b2) << 7) - } - s.length += 2 -} diff --git a/qb.go b/qb.go index 1775649..6ddd515 100644 --- a/qb.go +++ b/qb.go @@ -23,34 +23,36 @@ const ( ) type TimestampCompressor interface { - CalcRequiredSpace(uint32) (int, int) - Append(int, uint32) - Size() int - Chunks() [][]byte - DeleteLast() - Renew() + // (timestamp) => compressionWay, requiredSpace + Evaluate(uint32) (int, int) + Compress(int, uint32) + //Size() int + //Chunks() [][]byte + //DeleteLast() CaptureState() ForgetCapturedState() CreateDecompressor() TimestampDecompressor - Snapshot() ([]byte, []byte) // payload, additional payload + Snapshot() ([]byte, []byte) // tail, head Offset() int - Sync() + Rotate([]byte) int // payloadSize //LastTimestamp() uint32 } type ValueCompressor interface { - CalcRequiredSpace(float64) int - Append(float64) - Size() int - Chunks() [][]byte - DeleteLast() + // (value) => compressionWay, requiredSpace + Evaluate(float64) (int, int) + Compress(int, float64) + //Size() int + //Chunks() [][]byte + //DeleteLast() Renew() // створює новий conbuf, але не чіпає state - Lock() - Unlock() + CaptureState() + ForgetCapturedState() // fracDigits CreateDecompressor(byte) ValueDecompressor - Snapshot() ([][]byte, int) // chunks, size + Snapshot() ([]byte, []byte) // tail, head Offset() int + Rotate([]byte) int // payloadSize //LastValue() float64 }