From ed203195faa795ab7d79dc73acc9341bdc003c88 Mon Sep 17 00:00:00 2001 From: dima <1.e4.kc6@gmail.com> Date: Thu, 11 Jun 2026 14:27:38 +0000 Subject: [PATCH] wp --- database/metric.go | 112 ++++++------ enc/cumdelta.go | 484 ------------------------------------------------ enc/enc_test.go | 447 ++++++++++++++++++++++++--------------------- enc/insdelta.go | 526 ----------------------------------------------------- enc/time_delta.go | 7 +- enc/value_delta.go | 138 +++++++++++--- qb.go | 12 +- 7 files changed, 431 insertions(+), 1295 deletions(-) delete mode 100644 enc/cumdelta.go delete mode 100644 enc/insdelta.go diff --git a/database/metric.go b/database/metric.go index bbfb8f1..4f3b263 100644 --- a/database/metric.go +++ b/database/metric.go @@ -60,25 +60,17 @@ func (s *_metric) DeleteMeasures() { } // func (s *_metric) startAppendMeasures(req tryAppendMeasuresReq, sendToStorage func(storage.AppendedMeasures)) { -func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage func(any)) { +func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, sendToStorage func(any)) { var ( - // AppendedMeasures struct { - // MetricID uint32 - // TimestampsOffset int // (заповнені одразу) - // ValuesOffset int // (заповнені одразу) - // Timestamps [][]byte - // TimestampsSize int - // Values [][]byte - // ValuesSize int - // IndexLevels []*atree.IndexLevel - // DataPages []*atree.DataPage // fix - prevPageNo for the 1st data page - // } - timestamps = s.timestamps values = s.values dataPages []storage.DataPayload + // офсети на head сторінці + timestampsOffset int + valuesOffset int + //written int //resultCode byte ) @@ -87,9 +79,6 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu s.timestamps.CaptureState() for idx, measure := range req.Measures { - // if s.Since == 0 { - // s.Since = measure.Timestamp - // } else { // FIX - у випадку помилки треба в транзакції зберегти що помилка, але також зафіксувати скільки елементів збережено. // якщо idx == 0 - одразу знімаю блокування і нічого не відправляю в storage if measure.Timestamp <= s.timestamps.LastTimestamp() { @@ -126,51 +115,60 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu // fix - 1 + 8 bytes - timestampCompressionWay, timestampRequiredSpace := timestamps.Evaluate(measure.Timestamp) - valueCompressionWay, valueRequiredSpace := values.Evaluate(measure.Value) + tReport := timestamps.Evaluate(tmp, measure.Timestamp) + vReport := values.Evaluate(tmp, measure.Value) - totalRequiredSpace := timestampRequiredSpace + valueRequiredSpace + totalRequiredSpace := tReport.TotalSpace + vReport.TotalSpace if totalRequiredSpace <= len(s.buffer) { - // накопичую - timestamps.Compress(timestampCompressionWay, measure.Timestamp) - values.Compress(valueCompressionWay, measure.Value) - } else if len(s.buffer) < storage.DataPagePayloadSize { - // allocate bigger buffer - buffer := make([]byte, len(s.buffer)*2) - // copy timestamps - // copy values - // replace buffer in timestamps and values - - s.buffer = buffer - - timestamps.Compress(timestampCompressionWay, measure.Timestamp) - values.Compress(valueCompressionWay, measure.Value) + // якщо на сторінці є місце + if idx == 0 { + timestampsOffset = tReport.Offset + valuesOffset = vReport.Offset + } } else { - // сторінка заповнена - since := s.timestamps.ReplaceSinceWithUntil() + if len(s.buffer) < storage.DataPageSize { + // allocate bigger buffer + buf, databuf := growBuffers(growBuffersIn{ + databuf: s.buffer, + tSize: timestamps.Size(), + vSize: values.Size(), + requiredSpace: totalRequiredSpace, + }) - dataPages = append(dataPages, storage.DataPayload{ - Since: since, - Content: s.buffer, - TimestampsSize: timestamps.Size(), - ValuesSize: values.Size(), - }) + s.buffer = buf + // replace buffer in timestamps and values + timestamps.Rotate(databuf) // fix pos + values.Rotate(databuf) // fix pos + } else { + // сторінка заповнена + since := s.timestamps.ReplaceSinceWithUntil() - buffer := make([]byte, minBufferSize) + if len(dataPages) == 0 { + // head page FIX + // timestampsPayload = + // valuesPayload = + } - timestamps.Rotate(buffer) - values.Rotate(buffer) + dataPages = append(dataPages, storage.DataPayload{ + Since: since, + Content: s.buffer, + TimestampsSize: timestamps.Size(), + ValuesSize: values.Size(), + }) - // renew - s.buffer = buffer + buffer := make([]byte, minBufferSize) - timestampCompressionWay, _ = timestamps.Evaluate(measure.Timestamp) - valueCompressionWay, _ = values.Evaluate(measure.Value) + timestamps.Rotate(buffer) + values.Rotate(buffer) - timestamps.Compress(timestampCompressionWay, measure.Timestamp) - values.Compress(valueCompressionWay, measure.Value) + // renew + s.buffer = buffer + } } + + timestamps.Append(tReport.Offset, tmp[:tReport.ChangeSize], measure.Timestamp) + values.Append(vReport.Offset, tmp[7:7+vReport.ChangeSize], measure.Value, vReport.Delta) // s.lastValue = measure.Value } @@ -183,8 +181,8 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu sendToStorage(storage.AppendedMeasures{ MetricID: req.MetricID, LastPageNo: s.lastPageNo, - TimestampsOffset: timestamps.Offset(), // state.Pos() з якої позиції дописувати дані на сторінку 0 (при відновленні) - ValuesOffset: values.Offset(), // state.Pos() + TimestampsOffset: timestampsOffset, + ValuesOffset: valuesOffset, //Payload: s.payload, // fix - timestamps + values ? or timestamps and values (for WAL) IndexLevelTails: s.indexLevelTails, DataPages: dataPages, @@ -196,6 +194,16 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, sendToStorage fu }) } else { // короткий шлях - запис лише в storage + sendToStorage(storage.AppendedMeasures{ + MetricID: req.MetricID, + TimestampsOffset: timestampsOffset, + ValuesOffset: valuesOffset, + Timestamps: nil, // pos - offset + Values: nil, + //ResultCode: resultCode, + //WrittenCount: wri, + ResultCh: nil, + }) } } diff --git a/enc/cumdelta.go b/enc/cumdelta.go deleted file mode 100644 index 07d9e9d..0000000 --- a/enc/cumdelta.go +++ /dev/null @@ -1,484 +0,0 @@ -package enc - -import ( - "io" - "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 - h byte - state *CumulativeDeltaCapturedState -} - -// Після відновлення із снапшота -func NewCumulativeDeltaCompressor(fracDigits byte, buf []byte, payloadSize int) *CumulativeDeltaCompressor { - var coef float64 = 1 - if fracDigits > 0 { - coef = math.Pow(10, float64(fracDigits)) - } - s := &CumulativeDeltaCompressor{ - buf: buf, - coef: coef, - } - if payloadSize > 0 { - s.pos = payloadSize - 1 // завжди показує на h - // base value на початку - 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] - s.lastDelta, _, err = bin.ReverseGetVarUint64(s.buf[:s.pos-1]) - if err != nil { - log.Fatalf("bug: get last delta: %s", err) - } - } - return s -} - -// func (s *CumulativeDeltaCompressor) RestoreState() { - -// } - -// func (s *CumulativeDeltaCompressor) Size() int { -// return s.pos -// } - -type EvaluationReport struct { - CompressionWay int - TotalSpace int - AdditionalSpace int - Offset int - BytesCount int - Delta uint64 -} - -// можна виділити буфер максимального розміру, що може бути змінено під час кодування -// закодувати на етапі evaluate значення і повернути buf + offset. Причому буфер може бути -// спільний на всі metrics -func (s *CumulativeDeltaCompressor) Evaluate(value float64) (compressionWay int, requiredSpace int) { - delta := uint64((value-s.baseValue)*s.coef + eps) // fix - if delta 0, no eps - requiredSpace = s.pos - if s.pos > 0 { - if s.h < 128 { - // run - if delta == s.lastDelta && s.h < 127 { - compressionWay = incrementRun - // offset = s.pos - // bytesCount = 1 - } else { - compressionWay = endSeries - requiredSpace += bin.CountVarUint64(uint64(delta)) + hSize - // offset = s.pos + 1 - // bytesCount = bin.CountVarUint64(uint64(delta)) + hSize - } - } else { - // literal - if delta != s.lastDelta { - if s.h < 255 { - compressionWay = incrementLiteral - requiredSpace += bin.CountVarUint64(uint64(delta)) - // offset = s.pos - // bytesCount = bin.CountVarUint64(uint64(delta)) // hSize просто переміщається - } else { - compressionWay = endSeries - requiredSpace += bin.CountVarUint64(uint64(delta)) + hSize - } - } else { - compressionWay = startRun - if s.h > 128 { - // offset = s.pos - 1 - bin.CountVarUint64(uint64(delta)) - requiredSpace += hSize // h - end of run - } - } - } - } else { - // encode base value - compressionWay = addBaseValue - requiredSpace = bin.CountVarUint64(uint64(value*s.coef)) + // base value - bin.CountVarUint64(0) + // new delta - hSize // new h - } - return -} - -// pos завжди вказує на h -func (s *CumulativeDeltaCompressor) Compress(compressionWay int, value float64) { - delta := uint64((value-s.baseValue)*s.coef + eps) - switch compressionWay { - case incrementRun: - s.h++ - case incrementLiteral: - s.lastDelta = delta - s.h++ - // write previous delta and increment counter - n, _ := bin.ReversePutVarUint64(s.buf[s.pos:], delta) - s.pos += n - case endSeries: - s.lastDelta = delta - n, _ := bin.ReversePutVarUint64(s.buf[s.pos:], delta) - s.pos += n - s.h = 128 // start new literal (length=1) - case startRun: - if s.h > 128 { - // write h - end of literal (because length > 1) - s.pos -= 1 + bin.CountVarUint64(s.lastDelta) - s.h-- - s.buf[s.pos] = s.h - s.pos++ - // run delta - n, _ := bin.ReversePutVarUint64(s.buf[s.pos:], delta) - s.pos += n - } else { - } - // 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 - - s.lastDelta = 0 - n, _ = bin.ReversePutVarUint64(s.buf[s.pos:], s.lastDelta) - s.pos += n - // start new literal (length=1) - s.h = 128 - } - s.buf[s.pos] = s.h -} - -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) - } - // позиція посувається вліво, отже може перескочити на попередній chunk - pos := s.pos - hSize - bin.CountVarUint64(s.lastDelta) - s.state = &CumulativeDeltaCapturedState{ - H: s.h, - LastDelta: s.lastDelta, - Payload: s.buf[: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 -} - -func (s *CumulativeDeltaCompressor) Size() int { - if s.state == nil { - return s.pos - } else { - return bin.CountVarUint64(s.state.LastDelta) + hSize + len(s.state.Payload) - } -} - -// Snapshot - для створення снапшота. -func (s *CumulativeDeltaCompressor) Payload() []byte { - if s.state == nil { - return s.buf[:s.pos] - } else { - return s.state.Payload - } -} - -func (s *CumulativeDeltaCompressor) WritePayloadTo(w io.Writer) (err error) { - if s.state == nil { - _, err = w.Write(s.buf[:s.pos]) - return - } else { - _, err = w.Write(s.state.Payload) - if err != nil { - return - } - _, err = bin.WriteVarUint64(w, s.state.LastDelta) - if err != nil { - return - } - _, err = w.Write([]byte{ - 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) { - // 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 -} - -func (s *CumulativeDeltaCompressor) LastValue() float64 { - if s.state == nil { - return s.baseValue + float64(s.lastDelta)*s.coef - } else { - return s.baseValue + float64(s.state.LastDelta)*s.coef - } -} - -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, - }) -} - -// 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 - } - // метод працює як do while - спочатку значення, а потім перевірка умови - 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() { - //fmt.Println("read from pos:", s.pos) - s.pos-- - 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_test.go b/enc/enc_test.go index 8d52cab..9fb8c37 100644 --- a/enc/enc_test.go +++ b/enc/enc_test.go @@ -19,7 +19,7 @@ func equalFloatSlices(a, b []float64, epsilon float64) bool { // CUMULATIVE -func TestValueDeltaCompressor(t *testing.T) { +func TestCumulativeDeltaCompressor(t *testing.T) { var ( testCases = []struct { Nums []float64 @@ -76,7 +76,7 @@ func TestValueDeltaCompressor(t *testing.T) { 0x8f, // base value 0x80, // delta 0 0x80, // literal (len = 1) - 0x81, // delta 10 + 0x81, // delta 1 0x00, // run (len = 2) 0x00, 0x00, 0x00, }, @@ -115,45 +115,43 @@ func TestValueDeltaCompressor(t *testing.T) { 0x8f, // base value 0x80, // delta 0 0x01, // run (len = 3) - 0x81, // delta 10 + 0x81, // delta 1 0x80, // literal (len = 1) 0x00, 0x00, 0x00, }, BaseValue: 1.5, LastDelta: 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: repeatFloat64(1.5, 129), + Name: "increment run to full fill h-byte", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x8f, // base value + 0x80, // delta 0 + 0x7f, // run (len = 129) + 0x00, 0x00, 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: 0, + }, + { + Nums: repeatFloat64(1.5, 130), + Name: "run switch to literal after h-byte overflow", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x8f, // base value + 0x80, // delta 0 + 0x7f, // run (len = 129) + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: 0, + }, { Nums: []float64{ 1.5, @@ -166,8 +164,8 @@ func TestValueDeltaCompressor(t *testing.T) { Buf: []byte{ 0x8f, // base value 0x80, // delta 0 - 0x81, // delta 10 - 0x82, // delta 20 + 0x81, // delta 1 + 0x82, // delta 2 0x82, // literal (len = 3) 0x00, 0x00, 0x00, }, @@ -180,12 +178,13 @@ func TestValueDeltaCompressor(t *testing.T) { for _, testCase := range testCases { var ( tmp = make([]byte, tmpValueSize) + metricType = qb.Cumulative fracDigits byte = 1 buf = make([]byte, 8) payloadSize = 0 report qb.ValueEvaluationReport ) - c := NewValueDeltaCompressor(fracDigits, buf, payloadSize) + c := NewValueDeltaCompressor(metricType, fracDigits, buf, payloadSize) for _, num := range testCase.Nums { report = c.Evaluate(tmp, num) c.Append(report.Offset, tmp[:report.ChangeSize], num, report.Delta) @@ -285,15 +284,17 @@ func TestCumulativeDeltaDecompressorFromState(t *testing.T) { ) for caseIdx, testCase := range testCases { var ( + metricType = qb.Cumulative fracDigits byte = 1 + tmp = make([]byte, tmpValueSize) buf = make([]byte, 16) payloadSize = 0 decodedNums []float64 ) - c := NewCumulativeDeltaCompressor(fracDigits, buf, payloadSize) + c := NewValueDeltaCompressor(metricType, fracDigits, buf, payloadSize) for _, num := range testCase.Nums { - compressionWay, _ := c.Evaluate(num) - c.Compress(compressionWay, num) + report := c.Evaluate(tmp, num) + c.Append(report.Offset, tmp[:report.ChangeSize], num, report.Delta) } // @@ -331,41 +332,62 @@ func TestCumulativeDeltaDecompressorFromState(t *testing.T) { 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 + Name string + Offset int + ChangeSize int + Buf []byte + BaseValue float64 + LastDelta uint64 }{ { Nums: []float64{ 1.5, }, - Code: addBaseValue, - RequiredSpace: 3, // base value, delta, h + Name: "add 1st value", + Offset: 0, + ChangeSize: 3, Buf: []byte{ - 158, 0, 0, 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, 0x00, 0x00, }, BaseValue: 1.5, LastDelta: 0, - H: 128, // literal, length = 1 + }, + { + Nums: []float64{ + -1.5, + }, + Name: "add 1st value (negative)", + Offset: 0, + ChangeSize: 3, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, 0x00, 0x00, + }, + BaseValue: -1.5, + LastDelta: 0, }, { Nums: []float64{ 1.5, 1.5, }, - Code: startRun, // literal changed to run - RequiredSpace: 2, // delta, h + Name: "literal switch to run", + Offset: 1, + ChangeSize: 1, Buf: []byte{ - 158, 0, 0, 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, 0x00, 0x00, }, BaseValue: 1.5, LastDelta: 0, - H: 0, // run, length = 2 }, { Nums: []float64{ @@ -373,36 +395,39 @@ func TestInstantDeltaCompressor(t *testing.T) { 1.6, 1.6, }, - Code: startRun, // literal decreased by 1 - RequiredSpace: 3, // end of literal, new delta, new h + Name: "literal decrease by 1 and switch to run", + Offset: 2, + ChangeSize: 3, Buf: []byte{ - 158, // base value - 128, // literal 1st delta (0) - 128, // h - end of literal, length = 1 - 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x82, // delta 1 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, }, BaseValue: 1.5, - LastDelta: 1, - H: 0, // run, length = 2 + LastDelta: bin.EncodeZigZag(1), }, { - // 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 + Name: "literal decrease by 1 and switch to run (negative delta)", + Offset: 2, + ChangeSize: 3, Buf: []byte{ - 158, // base value - 128, // literal 1st delta (0) - 128, // h - end of literal, length = 1 - 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x81, // delta -1 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, }, BaseValue: 1.5, - LastDelta: -1, - H: 0, // run, length = 2 + LastDelta: bin.EncodeZigZag(-1), }, { Nums: []float64{ @@ -410,15 +435,17 @@ func TestInstantDeltaCompressor(t *testing.T) { 1.5, 1.5, }, - Code: incrementRun, - RequiredSpace: 2, + Name: "increment run", + Offset: 1, + ChangeSize: 1, Buf: []byte{ - 158, // base value - 0, 0, 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x00, 0x00, 0x00, 0x00, 0x00, }, BaseValue: 1.5, LastDelta: 0, - H: 1, // run, length = 3 }, { Nums: []float64{ @@ -427,149 +454,149 @@ func TestInstantDeltaCompressor(t *testing.T) { 1.5, 1.6, }, - Code: endSeries, // end of run - RequiredSpace: 4, + Name: "run switch to literal", + Offset: 0, + ChangeSize: 2, Buf: []byte{ - 158, // base value - 128, // run delta (0) - 1, // h of run, length = 3 - 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x82, // delta 1 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, }, BaseValue: 1.5, - LastDelta: 1, - H: 128, // literal, length = 1 + LastDelta: bin.EncodeZigZag(1), }, { - // same as prev, but negative delta Nums: []float64{ 1.5, 1.5, 1.5, 1.4, }, - Code: endSeries, // end of run - RequiredSpace: 4, + Name: "run switch to literal (negative delta)", + Offset: 0, + ChangeSize: 2, Buf: []byte{ - 158, // base value - 128, // run delta (0) - 1, // h of run, length = 3 - 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x81, // delta -1 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, }, BaseValue: 1.5, - LastDelta: -1, - H: 128, // literal, length = 1 + LastDelta: bin.EncodeZigZag(-1), }, { - Nums: []float64{ - 1.5, - }, - RepeatLastDeltaNTimes: 128, - Code: incrementRun, // h byte full filled - RequiredSpace: 2, + Nums: repeatFloat64(1.5, 129), + Name: "increment run to full fill h-byte", + Offset: 1, + ChangeSize: 1, Buf: []byte{ - 158, // base value - 0, 0, 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x7f, // run (len = 129) + 0x00, 0x00, 0x00, 0x00, 0x00, }, 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 + Nums: repeatFloat64(1.5, 130), + Name: "run switch to literal after h-byte overflow", + Offset: 0, + ChangeSize: 2, Buf: []byte{ - 158, // base value - 128, // run delta (0) - 127, // h of run (length = 129) - 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x7f, // run (len = 129) + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, }, BaseValue: 1.5, LastDelta: 0, - H: 128, // literal, length = 1 }, { Nums: []float64{ 1.5, 1.6, + 1.7, }, - Code: incrementLiteral, - RequiredSpace: 3, // previous delta, new delta, h + Name: "increment literal", + Offset: 1, + ChangeSize: 2, Buf: []byte{ - 158, // base value - 128, // literal 1st delta (0) - 0, 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x82, // delta 1 + 0x84, // delta 2 + 0x82, // literal (len = 3) + 0x00, 0x00, 0x00, }, BaseValue: 1.5, - LastDelta: 1, - H: 129, // literal, length = 2 + LastDelta: bin.EncodeZigZag(2), }, { - // same as prev, but negative delta Nums: []float64{ 1.5, 1.4, + 1.3, }, - Code: incrementLiteral, - RequiredSpace: 3, // previous delta, new delta, h + Name: "increment literal (negative delta)", + Offset: 1, + ChangeSize: 2, Buf: []byte{ - 158, // base value - 128, // literal 1st delta (0) - 0, 0, 0, 0, 0, 0, + 0x9e, // base value + 0x80, // delta 0 + 0x81, // delta -1 + 0x83, // delta -2 + 0x82, // literal (len = 3) + 0x00, 0x00, 0x00, }, BaseValue: 1.5, - LastDelta: -1, - H: 129, // literal, length = 2 + LastDelta: bin.EncodeZigZag(-2), }, } ) - for caseIdx, testCase := range testCases { + for _, testCase := range testCases { + // fmt.Println("-----") + // fmt.Println("nums", testCase.Nums) var ( - fracDigits byte = 1 - buf = make([]byte, 8) - payloadSize = 0 - compressionWay int - requiredSpace int + tmp = make([]byte, tmpValueSize) + metricType = qb.Instant + fracDigits byte = 1 + buf = make([]byte, 8) + payloadSize = 0 + report qb.ValueEvaluationReport ) - c := NewInstantDeltaCompressor(fracDigits, buf, payloadSize) + c := NewValueDeltaCompressor(metricType, fracDigits, buf, payloadSize) for _, num := range testCase.Nums { - compressionWay, requiredSpace = c.Evaluate(num) - c.Compress(compressionWay, num) + report = c.Evaluate(tmp, num) + c.Append(report.Offset, tmp[:report.ChangeSize], num, report.Delta) } - 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 report.Offset != testCase.Offset { + t.Fatalf("%s: got offset %d are not equal expected %d", + testCase.Name, report.Offset, testCase.Offset) } - 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 report.ChangeSize != testCase.ChangeSize { + t.Fatalf("%s: got changeSize %d are not equal expected %d", + testCase.Name, report.ChangeSize, testCase.ChangeSize) } if !bytes.Equal(buf, testCase.Buf) { - t.Fatalf("%d: got buf %v are not equal expected %v", - caseIdx, buf, testCase.Buf) + t.Fatalf("%s: got buf % x are not equal expected % x", + testCase.Name, buf, testCase.Buf) } if c.baseValue != testCase.BaseValue { - t.Fatalf("%d: got lastUnixtime %v are not equal expected %v", - caseIdx, c.baseValue, testCase.BaseValue) + t.Fatalf("%s: got baseValue %v are not equal expected %v", + testCase.Name, 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) + t.Fatalf("%s: got lastDelta %d are not equal expected %d", + testCase.Name, c.lastDelta, testCase.LastDelta) } } } @@ -646,15 +673,17 @@ func TestInstantDeltaDecompressorFromState(t *testing.T) { ) for caseIdx, testCase := range testCases { var ( + metricType = qb.Instant fracDigits byte = 1 + tmp = make([]byte, tmpValueSize) buf = make([]byte, 16) payloadSize = 0 decodedNums []float64 ) - c := NewInstantDeltaCompressor(fracDigits, buf, payloadSize) + c := NewValueDeltaCompressor(metricType, fracDigits, buf, payloadSize) for _, num := range testCase.Nums { - compressionWay, _ := c.Evaluate(num) - c.Compress(compressionWay, num) + report := c.Evaluate(tmp, num) + c.Append(report.Offset, tmp[:report.ChangeSize], num, report.Delta) } // @@ -809,42 +838,35 @@ func TestTimeDeltaCompressor(t *testing.T) { LastUnixtime: 1780777200, LastDelta: 20, }, - // { - // Nums: []uint32{ - // 1780777000, - // 1780777060, // +60 - // }, - // RepeatLastDeltaNTimes: 128, - // CompressionWay: incrementRun, // h byte full filled - // RequiredSpace: 6, - // Buf: []byte{ - // 0x80, // h-byte (literal, len=1) - // 0x94, // delta 20 - // 0x01, // h-byte (run, len=3) - // 0xbc, // delta 60 - // 0x28, 0x80, 0x24, 0x6a, // since - // }, - // 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: generateProgression(1780777000, 60, 130), + Name: "increment run up to full filled h-byte", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x00, 0x00, + 0x7f, // h-byte (run, len=129) + 0xbc, // delta 60 + 0x28, 0x80, 0x24, 0x6a, // since + }, + LastUnixtime: 1780777000 + 129*60, + LastDelta: 60, + }, + { + Nums: generateProgression(1780777000, 60, 131), + Name: "run switch to literal after h-byte overflow", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x80, // h-byte (literal, len=1) + 0xbc, // delta 60 + 0x7f, // h-byte (run, len=129) + 0xbc, // delta 60 + 0x28, 0x80, 0x24, 0x6a, // since + }, + LastUnixtime: 1780777000 + 130*60, + LastDelta: 60, + }, { Nums: []uint32{ 1780777000, @@ -1019,11 +1041,28 @@ func TestTimeDeltaDecompressorFromState(t *testing.T) { } } -func TestVarUint64(t *testing.T) { +func TestVarInt64(t *testing.T) { arr := make([]byte, 9) - n, err := bin.PutVarInt64(arr, 15) + n, err := bin.PutVarInt64(arr, -15) if err != nil { t.Fatal(err) } fmt.Println(arr[:n]) } + +func repeatFloat64(num float64, n int) []float64 { + nums := make([]float64, n) + for i := range nums { + nums[i] = num + } + return nums +} + +func generateProgression(start uint32, delta uint32, n int) []uint32 { + nums := make([]uint32, n) + nums[0] = start + for i := 1; i < n; i++ { + nums[i] = nums[i-1] + delta + } + return nums +} diff --git a/enc/insdelta.go b/enc/insdelta.go deleted file mode 100644 index a2fc972..0000000 --- a/enc/insdelta.go +++ /dev/null @@ -1,526 +0,0 @@ -package enc - -import ( - "io" - "log" - "math" - - bin "gordenko.dev/dima/bin/little" - "gordenko.dev/dima/qb" -) - -type InstantDeltaCompressor struct { - buf []byte - coef float64 - pos int - baseValue float64 - lastDelta int64 - h byte - state *InstantDeltaCapturedState -} - -func NewInstantDeltaCompressor(fracDigits byte, buf []byte, payloadSize int) *InstantDeltaCompressor { - var coef float64 = 1 - if fracDigits > 0 { - coef = math.Pow(10, float64(fracDigits)) - } - s := &InstantDeltaCompressor{ - buf: buf, - coef: coef, - } - if payloadSize > 0 { - s.pos = payloadSize - 1 // завжди показує на h - // base value на початку - 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] - s.lastDelta, _, err = bin.ReverseGetVarInt64(s.buf[:s.pos-1]) - if err != nil { - log.Fatalf("bug: get last delta: %s", err) - } - } - return s -} - -// func (s *InstantDeltaCompressor) RestoreState() { - -// } - -// 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 *InstantDeltaCompressor) DeleteLast() {} - -type InstantDeltaCapturedState struct { - H byte - LastDelta int64 - Payload []byte -} - -func (s *InstantDeltaCompressor) CaptureState() { - if s.state != nil { - qb.Abort(qb.RepeatableLock, nil) - } - // позиція посувається вліво, отже може перескочити на попередній 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 -} - -func (s *InstantDeltaCompressor) Size() int { - if s.state == nil { - return s.pos - } else { - return bin.CountVarInt64(s.state.LastDelta) + hSize + len(s.state.Payload) - } -} - -// Snapshot - для створення снапшота. -func (s *InstantDeltaCompressor) Payload() []byte { - if s.state == nil { - return s.buf[:s.pos] - } else { - return s.state.Payload - } -} - -func (s *InstantDeltaCompressor) WritePayloadTo(w io.Writer) (err error) { - if s.state == nil { - _, err = w.Write(s.buf[:s.pos]) - return - } else { - _, err = w.Write(s.state.Payload) - if err != nil { - return - } - _, err = bin.WriteVarInt64(w, s.state.LastDelta) - if err != nil { - return - } - _, err = w.Write([]byte{ - 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) { - // УВАГА! - // state не чіпаємо - s.buf = newbuf - s.pos = 0 - s.baseValue = 0 - s.lastDelta = 0 - s.h = 0 -} - -func (s *InstantDeltaCompressor) LastValue() float64 { - if s.state == nil { - return s.baseValue + float64(s.lastDelta)*s.coef - } else { - return s.baseValue + float64(s.state.LastDelta)*s.coef - } -} - -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 - } - // метод працює як do while - спочатку значення, а потім перевірка умови - 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() { - s.pos-- - 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 { - 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/enc/time_delta.go b/enc/time_delta.go index c510084..1138d8c 100644 --- a/enc/time_delta.go +++ b/enc/time_delta.go @@ -194,8 +194,11 @@ func (s *TimeDeltaCompressor) ReplaceSinceWithUntil() uint32 { return since } -// fix -// func (s *TimeDeltaCompressor) Size() int { +// для зростання буфера під час вставки даних. State не цікавить +func (s *TimeDeltaCompressor) Size() int { + return len(s.buf) - s.pos +} + // if s.state == nil { // return len(s.buf) - s.pos // } else { diff --git a/enc/value_delta.go b/enc/value_delta.go index c08b134..4e64714 100644 --- a/enc/value_delta.go +++ b/enc/value_delta.go @@ -20,10 +20,16 @@ type ValueDeltaCompressor struct { baseValue float64 lastDelta uint64 state *ValueDeltaCapturedState + // (baseValue, coef, value) => delta + calcDelta func(float64, float64, float64) uint64 + // (value, coef) => uint64 + toUint64 func(float64, float64) uint64 + // (value, coef) => uint64 + toFloat64 func(uint64, float64) float64 } // Після відновлення із снапшота -func NewValueDeltaCompressor(fracDigits byte, buf []byte, payloadSize int) *ValueDeltaCompressor { +func NewValueDeltaCompressor(metricType qb.MetricType, fracDigits byte, buf []byte, payloadSize int) *ValueDeltaCompressor { var coef float64 = 1 if fracDigits > 0 { coef = math.Pow(10, float64(fracDigits)) @@ -33,6 +39,15 @@ func NewValueDeltaCompressor(fracDigits byte, buf []byte, payloadSize int) *Valu coef: coef, pos: payloadSize, } + if metricType == qb.Cumulative { + s.calcDelta = calcCumulativeDelta + s.toUint64 = toCumulativeUint64 + s.toFloat64 = toCumulativeFloat64 + } else { + s.calcDelta = calcInstantDelta + s.toUint64 = toInstantUint64 + s.toFloat64 = toInstantFloat64 + } if payloadSize > 0 { // base value на початку u64, _, err := bin.GetVarUint64(s.buf) @@ -59,7 +74,19 @@ func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvalu i int ) if s.pos > 0 { - delta = uint64((value-s.baseValue)*s.coef + eps) + delta = s.calcDelta(s.baseValue, s.coef, value) + // if false { + // delta = uint64((value-s.baseValue)*s.coef + eps) + + // } else { + // f64 := (value - s.baseValue) * s.coef + // if f64 > 0 { + // f64 += eps + // } else { + // f64 -= eps + // } + // delta = bin.EncodeZigZag(int64(f64)) + // } h := s.buf[s.pos-1] if h < 128 { // run @@ -110,7 +137,13 @@ func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvalu } } } else { - n, _ := bin.PutVarUint64(tmp, uint64(value*s.coef)) // base value + // var baseValue uint64 + // if false { + // baseValue = uint64(value * s.coef) + // } else { + // baseValue = bin.EncodeZigZag(int64(value * s.coef)) + // } + n, _ := bin.PutVarUint64(tmp, s.toUint64(value, s.coef)) i += n n, _ = bin.ReversePutVarUint64(tmp[i:], 0) // delta i += n @@ -164,12 +197,13 @@ func (s *ValueDeltaCompressor) ForgetCapturedState() { s.state = nil } +// для зростання буфера під час вставки даних. State не цікавить func (s *ValueDeltaCompressor) Size() int { - if s.state == nil { - return s.pos - } else { - return bin.CountVarUint64(s.state.LastDelta) + hSize + len(s.state.Payload) - } + //if s.state == nil { + return s.pos + // } else { + // return bin.CountVarUint64(s.state.LastDelta) + hSize + len(s.state.Payload) + // } } // // Snapshot - для створення снапшота. @@ -217,7 +251,7 @@ func (s *ValueDeltaCompressor) LastValue() float64 { } else { delta = s.state.LastDelta } - return s.baseValue + float64(delta)*s.coef + return s.baseValue + s.toFloat64(delta, s.coef) } func (s *ValueDeltaCompressor) CreateDecompressor() qb.ValueDecompressor { @@ -236,6 +270,7 @@ func (s *ValueDeltaCompressor) CreateDecompressor() qb.ValueDecompressor { } return NewValueDeltaDecompressorFromState(ValueDeltaDecompressorFromStateOptions{ Coef: s.coef, + ToFloat64: s.toFloat64, H: h, LastDelta: lastDelta, Payload: payload, @@ -254,9 +289,11 @@ type ValueDeltaDecompressor struct { isRun bool pending int done bool + // (value, coef) => uint64 + toFloat64 func(uint64, float64) float64 } -func NewValueDeltaDecompressor(buf []byte, fracDigits byte) *ValueDeltaDecompressor { +func NewValueDeltaDecompressor(fracDigits byte, buf []byte, toFloat64 func(uint64, float64) float64) *ValueDeltaDecompressor { var coef float64 = 1 if fracDigits > 0 { coef = math.Pow(10, float64(fracDigits)) @@ -265,14 +302,21 @@ func NewValueDeltaDecompressor(buf []byte, fracDigits byte) *ValueDeltaDecompres if err != nil { log.Fatalf("bug: get base value: %s", err) } + // if false { + // baseValue = float64(u64) / coef + // } else { + // baseValue = float64(bin.DecodeZigZag(u64)) / coef + // } //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) s := &ValueDeltaDecompressor{ - buf: buf, - coef: coef, - pos: len(buf), // first free - baseValue: float64(u64) / coef, + buf: buf, + coef: coef, + pos: len(buf), // first free + // baseValue: baseValue, bound: n, + toFloat64: toFloat64, } + s.baseValue = s.toFloat64(u64, coef) // читаю заголовок наступної серії s.readHeader() s.readValue() @@ -280,7 +324,9 @@ func NewValueDeltaDecompressor(buf []byte, fracDigits byte) *ValueDeltaDecompres } type ValueDeltaDecompressorFromStateOptions struct { - Coef float64 + Coef float64 + // (value, coef) => uint64 + ToFloat64 func(uint64, float64) float64 H byte LastDelta uint64 Payload []byte @@ -291,15 +337,28 @@ func NewValueDeltaDecompressorFromState(opt ValueDeltaDecompressorFromStateOptio if err != nil { log.Fatalf("bug: get base value: %s", err) } + // var baseValue float64 + // if false { + // baseValue = float64(u64) / opt.Coef + // } else { + // baseValue = float64(bin.DecodeZigZag(u64)) / opt.Coef + // } //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/opt.Coef, n, size) s := &ValueDeltaDecompressor{ - buf: opt.Payload, - coef: opt.Coef, - pos: len(opt.Payload), // first free - baseValue: float64(u64) / opt.Coef, + buf: opt.Payload, + coef: opt.Coef, + pos: len(opt.Payload), // first free + //baseValue: baseValue, bound: n, + toFloat64: opt.ToFloat64, } - s.lastValue = s.baseValue + float64(opt.LastDelta)/s.coef + s.baseValue = s.toFloat64(u64, s.coef) + // if false { + // s.lastValue = s.baseValue + float64(opt.LastDelta)/s.coef + // } else { + // s.lastValue = s.baseValue + float64(bin.DecodeZigZag(opt.LastDelta))/s.coef + // } + s.lastValue = s.baseValue + s.toFloat64(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", @@ -353,7 +412,12 @@ func (s *ValueDeltaDecompressor) readValue() { // fmt.Println("read delta n:", n) // fmt.Println() s.pos -= n - s.lastValue = s.baseValue + float64(u64)/s.coef + // if false { + // s.lastValue = s.baseValue + float64(u64)/s.coef + // } else { + // s.lastValue = s.baseValue + float64(bin.DecodeZigZag(u64))/s.coef + // } + s.lastValue = s.baseValue + s.toFloat64(u64, s.coef) //fmt.Println(s.baseValue, float64(u64)/s.coef) } @@ -366,6 +430,38 @@ func (s *ValueDeltaDecompressor) decodeHeaderByte(h byte) { } } +// HELPERS + +func calcCumulativeDelta(baseValue float64, coef float64, value float64) uint64 { + return uint64((value-baseValue)*coef + eps) +} + +func calcInstantDelta(baseValue float64, coef float64, value float64) uint64 { + f64 := (value - baseValue) * coef + if f64 > 0 { + f64 += eps + } else { + f64 -= eps + } + return bin.EncodeZigZag(int64(f64)) +} + +func toCumulativeUint64(value float64, coef float64) uint64 { + return uint64(value * coef) +} + +func toInstantUint64(value float64, coef float64) uint64 { + return bin.EncodeZigZag(int64(value * coef)) +} + +func toCumulativeFloat64(value uint64, coef float64) float64 { + return float64(value) / coef +} + +func toInstantFloat64(value uint64, coef float64) float64 { + return float64(bin.DecodeZigZag(value)) / coef +} + /* Формат: base value (var u64) diff --git a/qb.go b/qb.go index 5e29b6a..083b4ba 100644 --- a/qb.go +++ b/qb.go @@ -26,9 +26,9 @@ const ( type TimestampCompressor interface { // (tmp, timestamp) Evaluate([]byte, uint32) TimeEvaluationReport - // (offset, tmp) - Append(int, []byte) - //Size() int + // (offset, change, timestamp) + Append(int, []byte, uint32) + Size() int //Chunks() [][]byte //DeleteLast() CaptureState() @@ -51,9 +51,9 @@ type TimeEvaluationReport struct { type ValueCompressor interface { // (tmp, value) Evaluate([]byte, float64) ValueEvaluationReport - // (offset, tmp, delta) - Append(int, []byte, uint64) - //Size() int + // (offset, change, value, delta) + Append(int, []byte, float64, uint64) + Size() int //Chunks() [][]byte //DeleteLast() CaptureState()