diff --git a/client/client.go b/client/client.go index 37c4f23..98b477d 100644 --- a/client/client.go +++ b/client/client.go @@ -67,7 +67,7 @@ func (s *Connection) AddMetric(req proto.AddMetricReq) (err error) { proto.TypeAddMetric, 0, 0, 0, 0, // byte(req.MetricType), - byte(req.FracDigits), + req.FracDigits, } bin.PutUint32(arr[1:], req.MetricID) // req @@ -82,7 +82,7 @@ func (s *Connection) AddMetric(req proto.AddMetricReq) (err error) { // arr := []byte{ // proto.TypeUpdateMetric, // 0, 0, 0, 0, // -// byte(req.FracDigits), +// req.FracDigits, // } // bin.PutUint32(arr[1:], req.MetricID) diff --git a/database/database.go b/database/database.go index e4199c7..db8d10b 100644 --- a/database/database.go +++ b/database/database.go @@ -288,67 +288,6 @@ func (s *Database) relayMetricsToMetrics(replayMetrics map[uint32]*ReplayMetric) } } -// FIX -// УВАГА! -// Якщо буфер metric.buffer < розміру сторінки - для timestamps доступний весь буфер. -// Якщо буфер досяг розміру сторінки - в timestamps я передаю buffer[:DataPagePayloadSize] - -var dataBufferSizes = []int{ - 1024, - 2048, - 4096, - 8192, -} - -func calculateDataBufferSize(payloadSize int) int { - for _, bufferSize := range dataBufferSizes { - if payloadSize <= bufferSize { - return bufferSize - } - } - return storage.DataPageSize -} - -func allocateBuffers(requiredSpace int) (buf, databuf []byte) { - bufferSize := calculateDataBufferSize(requiredSpace) - buf = make([]byte, bufferSize) - if bufferSize == storage.DataPageSize { - databuf = buf[:storage.DataPagePayloadSize] - } else { - databuf = buf - } - return -} - -// src - databuf -type growBuffersIn struct { - databuf []byte - tSize int - vSize int - requiredSpace int -} - -func growBuffers(in growBuffersIn) (buf, databuf []byte) { - buf, databuf = allocateBuffers(in.requiredSpace) - copy(databuf[len(databuf)-in.tSize:], in.databuf[len(in.databuf)-in.tSize:]) - copy(databuf, in.databuf[:in.vSize]) - return -} - -type shrinkBufferIn struct { - Buffer []byte - TimestampsSize int - ValuesSize int - NewBufferSize int -} - -func shrinkBuffer(in shrinkBufferIn) []byte { - buf := make([]byte, in.NewBufferSize) - copy(buf, in.Buffer[:in.ValuesSize]) - copy(buf[:len(buf)-in.TimestampsSize], in.Buffer[len(in.Buffer)-in.TimestampsSize:]) - return buf -} - //func (s *Database) verifySnapshot(fileName string) (_ bool, err error) { // file, err := os.Open(fileName) // if err != nil { diff --git a/database/metric.go b/database/metric.go index c718fc9..8ddda8b 100644 --- a/database/metric.go +++ b/database/metric.go @@ -10,7 +10,7 @@ import ( // METRIC -const minBufferSize = 1024 +//const minBufferSize = 1024 var ( indexRecordSize = 8 @@ -79,29 +79,31 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, send timestampsOffset int valuesOffset int + headTimestamps []byte + headValues []byte + written int resultCode byte ) s.capturedState = &CapturedState{ // - TimeState: s.timestamps.CaptureState(), - ValueState: s.values.CaptureState(), + } + s.timestamps.CaptureState() + s.values.CaptureState() + for idx, measure := range req.Measures { if measure.Timestamp <= s.timestamps.LastTimestamp() { resultCode = ExpiredMeasure break } - if s.metricType == qb.Cumulative && measure.Value < s.lastValue { resultCode = NonMonotonicValue break } - // fix - 1 + 8 bytes - tReport := timestamps.Evaluate(tmp, measure.Timestamp) vReport := values.Evaluate(tmp, measure.Value) @@ -114,48 +116,33 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, send valuesOffset = vReport.Offset } } else { - if len(s.buffer) < storage.DataPageSize { - // allocate bigger buffer - buf, databuf := growBuffers(growBuffersIn{ - databuf: s.buffer, - tSize: timestamps.Size(), - vSize: values.Size(), - requiredSpace: totalRequiredSpace, - }) + // сторінка заповнена + since := s.timestamps.ReplaceSinceWithUntil() - s.buffer = buf - // replace buffer in timestamps and values - timestamps.Rotate(databuf) // fix pos - values.Rotate(databuf) // fix pos - } else { - // сторінка заповнена - since := s.timestamps.ReplaceSinceWithUntil() - - if len(s.capturedState.Pages) == 0 { - // head page FIX - // timestampsPayload = - // valuesPayload = - } - - s.capturedState.Pages = append(s.capturedState.Pages, storage.DataPayload{ - Since: since, - Content: s.buffer, - TimestampsSize: timestamps.Size(), - ValuesSize: values.Size(), - }) - - buffer := make([]byte, minBufferSize) - - timestamps.Rotate(buffer) - values.Rotate(buffer) - - // renew - s.buffer = buffer + if len(s.capturedState.Pages) == 0 { + headTimestamps = timestamps.Tail(timestampsOffset) + headValues = values.Tail(valuesOffset) } + + s.capturedState.Pages = append(s.capturedState.Pages, storage.DataPayload{ + Since: since, + Content: s.buffer, + TimestampsSize: timestamps.Size(), + ValuesSize: values.Size(), + }) + + buf := make([]byte, storage.DataPageSize) + databuf := buf[:storage.DataPagePayloadSize] + + timestamps.ReplaceBuffer(databuf) + values.ReplaceBuffer(databuf) + + // renew + s.buffer = buf } - timestamps.Append(tReport.Offset, tmp[:tReport.ChangeSize], measure.Timestamp) - values.Append(vReport.Offset, tmp[7:7+vReport.ChangeSize], measure.Value, vReport.Delta) + timestamps.Append(tReport.RewindOffset, tmp[:tReport.ChangeSize], measure.Timestamp) + values.Append(vReport.RewindOffset, tmp[7:7+vReport.ChangeSize], measure.Value, vReport.Delta) // s.lastValue = measure.Value written++ @@ -173,39 +160,33 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, send // скопіювати. Причому можна скопіювати зрізи chunks if len(s.capturedState.Pages) > 0 { - head := s.capturedState.Pages[0] - databuf := head.Content[:storage.DataPagePayloadSize] // пишу в storage довгим шляхом через redo файл і запис в data файл sendToStorage(storage.AppendedMeasures{ MetricID: req.MetricID, LastPageNo: s.lastPageNo, TimestampsOffset: timestampsOffset, - Timestamps: databuf[len(databuf)-head.TimestampsSize : len(databuf)-timestampsOffset], + Timestamps: headTimestamps, ValuesOffset: valuesOffset, - Values: databuf[valuesOffset:head.ValuesSize], + Values: headValues, IndexLevelTails: s.indexLevelTails, DataPages: s.capturedState.Pages, - TailTimestamps: nil, // payload - TailValues: s.buffer[:values.Size()], + TailTimestamps: timestamps.Tail(0), // payload + TailValues: values.Tail(0), ResultCode: resultCode, WrittenCount: written, - ResultCh: nil, + //ResultCh: req.ResultCh, }) } else { - databuf := s.buffer - if len(s.buffer) == storage.DataPageSize { - databuf = s.buffer[:storage.DataPagePayloadSize] - } // короткий шлях - запис лише в storage sendToStorage(storage.AppendedMeasures{ MetricID: req.MetricID, TimestampsOffset: timestampsOffset, ValuesOffset: valuesOffset, - Timestamps: databuf[len(databuf)-timestamps.Size() : len(databuf)-timestampsOffset], - Values: databuf[valuesOffset:values.Size()], + Timestamps: timestamps.Tail(timestampsOffset), + Values: values.Tail(valuesOffset), ResultCode: resultCode, WrittenCount: written, - ResultCh: nil, + //ResultCh: req.ResultCh, }) } diff --git a/database/proc.go b/database/proc.go index b099f2a..3ee0768 100644 --- a/database/proc.go +++ b/database/proc.go @@ -472,14 +472,15 @@ func (s *Database) applyAddMetric(rec storage.MetricAddRecord) { func (s *Database) addMetric(rec storage.MetricAddRecord) { var ( - buffer = make([]byte, minBufferSize) + buf = make([]byte, storage.DataPageSize) + databuf = buf[:storage.DataPagePayloadSize] ) s.metrics[rec.MetricID] = &_metric{ metricType: rec.MetricType, - fracDigits: byte(rec.FracDigits), - buffer: buffer, - timestamps: enc.NewTimeDeltaCompressor(buffer, 0), - values: enc.NewValueDeltaCompressor(rec.MetricType, byte(rec.FracDigits), buffer, 0), + fracDigits: rec.FracDigits, + buffer: buf, + timestamps: enc.NewTimeDeltaCompressor(databuf, 0), + values: enc.NewValueDeltaCompressor(rec.MetricType, rec.FracDigits, databuf, 0), } } diff --git a/database/replay_metric.go b/database/replay_metric.go index 2628b05..1c34d18 100644 --- a/database/replay_metric.go +++ b/database/replay_metric.go @@ -271,57 +271,17 @@ func composeHeadDataPage(databuf []byte, head storage.HeadDataPage) storage.Page } func (s *ReplayMetric) ToMetric() *_metric { - var ( - requiredSpace = s.tSize + s.vSize - ) - bufferSize := calculateDataBufferSize(requiredSpace) - if len(s.buf) > bufferSize { - s.buf = shrinkBuffer(shrinkBufferIn{ - Buffer: s.buf, - TimestampsSize: s.tSize, - ValuesSize: s.vSize, - NewBufferSize: bufferSize, - }) - } - values := enc.NewValueDeltaCompressor(s.metricType, s.fracDigits, s.buf, s.vSize) + + databuf := s.buf[:storage.DataPagePayloadSize] + values := enc.NewValueDeltaCompressor(s.metricType, s.fracDigits, databuf, s.vSize) return &_metric{ metricType: s.metricType, fracDigits: s.fracDigits, lastPageNo: s.lastPageNo, lastValue: values.LastValue(), buffer: s.buf, - timestamps: enc.NewTimeDeltaCompressor(s.buf, s.tSize), + timestamps: enc.NewTimeDeltaCompressor(databuf, s.tSize), values: values, indexLevelTails: s.indexLevelTails, } } - -// func (s *ReplayMetric) ApplyHeadDataPage(p storage.HeadDataPage) { -// // буфер має бути розміру сторінки -// if len(s.buf) != atree.DataPageSize { -// s.buf, s.databuf = growBuffers(growBuffersIn{ -// databuf: s.databuf, -// tSize: s.tSize, -// vSize: s.vSize, -// requiredSpace: atree.DataPageSize, -// }) -// } -// // дописую дані по зміщенню -// pos := len(s.databuf) - p.TimestampsOffset - len(p.Timestamps) -// copy(s.databuf[pos:], p.Timestamps) -// copy(s.databuf[p.ValuesOffset:], p.Values) -// // -// s.tSize = p.TimestampsOffset + len(p.Timestamps) -// s.tSize = p.ValuesOffset + len(p.Values) - -// // fix - заполнить страницу -// // fix - check checksum -// } - -// func (s *ReplayMetric) ApplyHeadIndexPages(changedIndexLevels storage.ChangedIndexLevel) { -// for levelIdx, changedLevel := range changedIndexLevels { -// - -// -// } -// } diff --git a/database/tmp b/database/tmp new file mode 100644 index 0000000..8013628 --- /dev/null +++ b/database/tmp @@ -0,0 +1,93 @@ +// IN APPEND MEASURES +// if len(s.buffer) < storage.DataPageSize { +// // allocate bigger buffer +// buf, databuf := growBuffers(growBuffersIn{ +// databuf: s.buffer, +// tSize: timestamps.Size(), +// vSize: values.Size(), +// requiredSpace: totalRequiredSpace, +// }) + +// s.buffer = buf +// // replace buffer in timestamps and values +// timestamps.Rotate(databuf) // fix pos +// values.Rotate(databuf) // fix pos +// } else { + +// IN DATABASE + +// FIX +// УВАГА! +// Якщо буфер metric.buffer < розміру сторінки - для timestamps доступний весь буфер. +// Якщо буфер досяг розміру сторінки - в timestamps я передаю buffer[:DataPagePayloadSize] + +// var dataBufferSizes = []int{ +// 1024, +// 2048, +// 4096, +// 8192, +// } + +// func calculateDataBufferSize(payloadSize int) int { +// for _, bufferSize := range dataBufferSizes { +// if payloadSize <= bufferSize { +// return bufferSize +// } +// } +// return storage.DataPageSize +// } + +// func allocateBuffers(requiredSpace int) (buf, databuf []byte) { +// bufferSize := calculateDataBufferSize(requiredSpace) +// buf = make([]byte, bufferSize) +// if bufferSize == storage.DataPageSize { +// databuf = buf[:storage.DataPagePayloadSize] +// } else { +// databuf = buf +// } +// return +// } + +// // src - databuf +// type growBuffersIn struct { +// databuf []byte +// tSize int +// vSize int +// requiredSpace int +// } + +// func growBuffers(in growBuffersIn) (buf, databuf []byte) { +// buf, databuf = allocateBuffers(in.requiredSpace) +// copy(databuf[len(databuf)-in.tSize:], in.databuf[len(in.databuf)-in.tSize:]) +// copy(databuf, in.databuf[:in.vSize]) +// return +// } + +// type shrinkBufferIn struct { +// Buffer []byte +// TimestampsSize int +// ValuesSize int +// NewBufferSize int +// } + +// func shrinkBuffer(in shrinkBufferIn) []byte { +// buf := make([]byte, in.NewBufferSize) +// copy(buf, in.Buffer[:in.ValuesSize]) +// copy(buf[:len(buf)-in.TimestampsSize], in.Buffer[len(in.Buffer)-in.TimestampsSize:]) +// return buf +// } + +// IN REPLAY_METRIC.TO_METRIC + +// var ( +// requiredSpace = s.tSize + s.vSize +// ) +// bufferSize := calculateDataBufferSize(requiredSpace) +// if len(s.buf) > bufferSize { +// s.buf = shrinkBuffer(shrinkBufferIn{ +// Buffer: s.buf, +// TimestampsSize: s.tSize, +// ValuesSize: s.vSize, +// NewBufferSize: bufferSize, +// }) +// } \ No newline at end of file diff --git a/enc/enc_test.go b/enc/enc_test.go index 9fb8c37..ec03ae6 100644 --- a/enc/enc_test.go +++ b/enc/enc_test.go @@ -19,163 +19,163 @@ func equalFloatSlices(a, b []float64, epsilon float64) bool { // CUMULATIVE -func TestCumulativeDeltaCompressor(t *testing.T) { - var ( - testCases = []struct { - Nums []float64 - Name string - Offset int - ChangeSize int - Buf []byte - BaseValue float64 - LastDelta uint64 - }{ - { - Nums: []float64{ - 1.5, - }, - Name: "add 1st value", - Offset: 0, - ChangeSize: 3, - Buf: []byte{ - 0x8f, // base value - 0x80, // delta 0 - 0x80, // literal (len = 1) - 0x00, 0x00, 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 0, +var ( + cumulativeTestCases = []struct { + Nums []float64 + Name string + Offset int + ChangeSize int + Buf []byte + BaseValue float64 + LastDelta uint64 + }{ + { + Nums: []float64{ + 1.5, }, - { - Nums: []float64{ - 1.5, - 1.5, - }, - Name: "literal switch to run", - Offset: 1, - ChangeSize: 1, - Buf: []byte{ - 0x8f, // base value - 0x80, // delta 0 - 0x00, // run (len = 2) - 0x00, 0x00, 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 0, + Name: "add 1st value", + Offset: 0, + ChangeSize: 3, + Buf: []byte{ + 0x8f, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, 0x00, 0x00, }, - { - Nums: []float64{ - 1.5, - 1.6, - 1.6, - }, - Name: "literal decrease by 1 and switch to run", - Offset: 2, - ChangeSize: 3, - Buf: []byte{ - 0x8f, // base value - 0x80, // delta 0 - 0x80, // literal (len = 1) - 0x81, // delta 1 - 0x00, // run (len = 2) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 1, + BaseValue: 1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + 1.5, + 1.5, }, - { - Nums: []float64{ - 1.5, - 1.5, - 1.5, - }, - Name: "increment run", - Offset: 1, - ChangeSize: 1, - Buf: []byte{ - 0x8f, // base value - 0x80, // delta 0 - 0x01, // run (len = 3) - 0x00, 0x00, 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 0, + Name: "literal switch to run", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x8f, // base value + 0x80, // delta 0 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, 0x00, 0x00, }, - { - Nums: []float64{ - 1.5, - 1.5, - 1.5, - 1.6, - }, - Name: "run switch to literal", - Offset: 0, - ChangeSize: 2, - Buf: []byte{ - 0x8f, // base value - 0x80, // delta 0 - 0x01, // run (len = 3) - 0x81, // delta 1 - 0x80, // literal (len = 1) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 1, + BaseValue: 1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + 1.5, + 1.6, + 1.6, }, - { - 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, + Name: "literal decrease by 1 and switch to run", + Offset: 2, + ChangeSize: 3, + Buf: []byte{ + 0x8f, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x81, // delta 1 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, }, - { - 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, + BaseValue: 1.5, + LastDelta: 1, + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, }, - { - Nums: []float64{ - 1.5, - 1.6, - 1.7, - }, - Name: "increment literal", - Offset: 1, - ChangeSize: 2, - Buf: []byte{ - 0x8f, // base value - 0x80, // delta 0 - 0x81, // delta 1 - 0x82, // delta 2 - 0x82, // literal (len = 3) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 2, + Name: "increment run", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x8f, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x00, 0x00, 0x00, 0x00, 0x00, }, - } - ) + BaseValue: 1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, + 1.6, + }, + Name: "run switch to literal", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x8f, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x81, // delta 1 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: 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, + 1.6, + 1.7, + }, + Name: "increment literal", + Offset: 1, + ChangeSize: 2, + Buf: []byte{ + 0x8f, // base value + 0x80, // delta 0 + 0x81, // delta 1 + 0x82, // delta 2 + 0x82, // literal (len = 3) + 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: 2, + }, + } +) - for _, testCase := range testCases { +func TestCumulativeDeltaCompressor(t *testing.T) { + for _, testCase := range cumulativeTestCases { var ( tmp = make([]byte, tmpValueSize) metricType = qb.Cumulative @@ -212,77 +212,37 @@ func TestCumulativeDeltaCompressor(t *testing.T) { } } -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, // - }, - }, +func TestRestoreCumulativeDeltaCompressor(t *testing.T) { + for _, testCase := range cumulativeTestCases { + var ( + tmp = make([]byte, tmpValueSize) + metricType = qb.Cumulative + fracDigits byte = 1 + buf = make([]byte, 8) + payloadSize = 0 + report qb.ValueEvaluationReport + ) + 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) } - ) - for caseIdx, testCase := range testCases { + + restored := NewValueDeltaCompressor(metricType, fracDigits, buf, c.Size()) + + if restored.baseValue != testCase.BaseValue { + t.Fatalf("%s: got baseValue %v are not equal expected %v", + testCase.Name, restored.baseValue, testCase.BaseValue) + } + if restored.lastDelta != testCase.LastDelta { + t.Fatalf("%s: got lastDelta %d are not equal expected %d", + testCase.Name, restored.lastDelta, testCase.LastDelta) + } + } +} + +func TestCumulativeDeltaDecompressorFromState(t *testing.T) { + for _, testCase := range cumulativeTestCases { var ( metricType = qb.Cumulative fracDigits byte = 1 @@ -297,18 +257,9 @@ func TestCumulativeDeltaDecompressorFromState(t *testing.T) { c.Append(report.Offset, tmp[:report.ChangeSize], num, report.Delta) } - // - c.CaptureState() - - fmt.Println("---------------") - - d := c.CreateDecompressor() - - //fmt.Println(buf) - + d := c.CreateDecompressor(metricType, fracDigits) for { num, done := d.NextValue() - //fmt.Println(num, done) if done { break } @@ -317,254 +268,471 @@ func TestCumulativeDeltaDecompressorFromState(t *testing.T) { slices.Reverse(decodedNums) - fmt.Println(testCase.Nums) - fmt.Println(decodedNums) - fmt.Println() + if !equalFloatSlices(testCase.Nums, decodedNums, 0.0000001) { + t.Fatalf("%s: got nums %v not equal expected %v", + testCase.Name, decodedNums, testCase.Nums) + } + } +} + +func TestCumulativeDeltaDecompressorFromEnd(t *testing.T) { + for _, testCase := range cumulativeTestCases { + var ( + metricType = qb.Cumulative + fracDigits byte = 1 + tmp = make([]byte, tmpValueSize) + buf = make([]byte, 16) + payloadSize = 0 + decodedNums []float64 + ) + 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) + } + + d := NewValueDeltaDecompressor(metricType, fracDigits) + d.RestoreFromEnd(buf[:c.Size()]) + for { + num, done := d.NextValue() + if done { + break + } + decodedNums = append(decodedNums, num) + } + + slices.Reverse(decodedNums) if !equalFloatSlices(testCase.Nums, decodedNums, 0.0000001) { - t.Fatalf("%d: got nums %v not equal expected %v", caseIdx, decodedNums, testCase.Nums) + t.Fatalf("%s: got nums %v not equal expected %v", + testCase.Name, decodedNums, testCase.Nums) } } } // INSTANT -func TestInstantDeltaCompressor(t *testing.T) { - var ( - testCases = []struct { - Nums []float64 - Name string - Offset int - ChangeSize int - Buf []byte - BaseValue float64 - LastDelta uint64 - }{ - { - Nums: []float64{ - 1.5, - }, - Name: "add 1st value", - Offset: 0, - ChangeSize: 3, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x80, // literal (len = 1) - 0x00, 0x00, 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 0, +var ( + instantTestCases = []struct { + Nums []float64 + Name string + Offset int + ChangeSize int + Buf []byte + BaseValue float64 + LastDelta uint64 + }{ + { + Nums: []float64{ + 1.5, }, - { - 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, + Name: "add 1st value", + Offset: 0, + ChangeSize: 3, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, 0x00, 0x00, }, - { - Nums: []float64{ - 1.5, - 1.5, - }, - Name: "literal switch to run", - Offset: 1, - ChangeSize: 1, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x00, // run (len = 2) - 0x00, 0x00, 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 0, + BaseValue: 1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + -1.5, }, - { - Nums: []float64{ - 1.5, - 1.6, - 1.6, - }, - Name: "literal decrease by 1 and switch to run", - Offset: 2, - ChangeSize: 3, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x80, // literal (len = 1) - 0x82, // delta 1 - 0x00, // run (len = 2) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: bin.EncodeZigZag(1), + 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, }, - { - Nums: []float64{ - 1.5, - 1.4, - 1.4, - }, - Name: "literal decrease by 1 and switch to run (negative delta)", - Offset: 2, - ChangeSize: 3, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x80, // literal (len = 1) - 0x81, // delta -1 - 0x00, // run (len = 2) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: bin.EncodeZigZag(-1), + BaseValue: -1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + 1.5, + 1.5, }, - { - Nums: []float64{ - 1.5, - 1.5, - 1.5, - }, - Name: "increment run", - Offset: 1, - ChangeSize: 1, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x01, // run (len = 3) - 0x00, 0x00, 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 0, + Name: "literal switch to run", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, 0x00, 0x00, }, - { - Nums: []float64{ - 1.5, - 1.5, - 1.5, - 1.6, - }, - Name: "run switch to literal", - Offset: 0, - ChangeSize: 2, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x01, // run (len = 3) - 0x82, // delta 1 - 0x80, // literal (len = 1) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: bin.EncodeZigZag(1), + BaseValue: 1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + -1.5, + -1.5, }, - { - Nums: []float64{ - 1.5, - 1.5, - 1.5, - 1.4, - }, - Name: "run switch to literal (negative delta)", - Offset: 0, - ChangeSize: 2, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x01, // run (len = 3) - 0x81, // delta -1 - 0x80, // literal (len = 1) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: bin.EncodeZigZag(-1), + Name: "literal switch to run (negative)", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, 0x00, 0x00, }, - { - Nums: repeatFloat64(1.5, 129), - Name: "increment run to full fill h-byte", - Offset: 1, - ChangeSize: 1, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x7f, // run (len = 129) - 0x00, 0x00, 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: 0, + BaseValue: -1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + 1.5, + 1.6, + 1.6, }, - { - Nums: repeatFloat64(1.5, 130), - Name: "run switch to literal after h-byte overflow", - Offset: 0, - ChangeSize: 2, - Buf: []byte{ - 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, + Name: "literal decrease by 1 and switch to run", + Offset: 2, + ChangeSize: 3, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x82, // delta 1 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, }, - { - Nums: []float64{ - 1.5, - 1.6, - 1.7, - }, - Name: "increment literal", - Offset: 1, - ChangeSize: 2, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x82, // delta 1 - 0x84, // delta 2 - 0x82, // literal (len = 3) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: bin.EncodeZigZag(2), + BaseValue: 1.5, + LastDelta: bin.EncodeZigZag(1), + }, + { + Nums: []float64{ + -1.5, + -1.6, + -1.6, }, - { - Nums: []float64{ - 1.5, - 1.4, - 1.3, - }, - Name: "increment literal (negative delta)", - Offset: 1, - ChangeSize: 2, - Buf: []byte{ - 0x9e, // base value - 0x80, // delta 0 - 0x81, // delta -1 - 0x83, // delta -2 - 0x82, // literal (len = 3) - 0x00, 0x00, 0x00, - }, - BaseValue: 1.5, - LastDelta: bin.EncodeZigZag(-2), + Name: "literal decrease by 1 and switch to run (negative)", + Offset: 2, + ChangeSize: 3, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x81, // delta -1 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, }, - } - ) + BaseValue: -1.5, + LastDelta: bin.EncodeZigZag(-1), + }, + { + Nums: []float64{ + 1.5, + 1.4, + 1.4, + }, + Name: "literal decrease by 1 and switch to run (negative delta)", + Offset: 2, + ChangeSize: 3, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x81, // delta -1 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: bin.EncodeZigZag(-1), + }, + { + Nums: []float64{ + -1.5, + -1.4, + -1.4, + }, + Name: "literal decrease by 1 and switch to run (positive delta, negative)", + Offset: 2, + ChangeSize: 3, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x80, // literal (len = 1) + 0x82, // delta 1 + 0x00, // run (len = 2) + 0x00, 0x00, 0x00, + }, + BaseValue: -1.5, + LastDelta: bin.EncodeZigZag(1), + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, + }, + Name: "increment run", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x00, 0x00, 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + -1.5, + -1.5, + -1.5, + }, + Name: "increment run (negative)", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x00, 0x00, 0x00, 0x00, 0x00, + }, + BaseValue: -1.5, + LastDelta: 0, + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, + 1.6, + }, + Name: "run switch to literal", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x82, // delta 1 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: bin.EncodeZigZag(1), + }, + { + Nums: []float64{ + -1.5, + -1.5, + -1.5, + -1.6, + }, + Name: "run switch to literal (negative)", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x81, // delta -1 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, + }, + BaseValue: -1.5, + LastDelta: bin.EncodeZigZag(-1), + }, + { + Nums: []float64{ + 1.5, + 1.5, + 1.5, + 1.4, + }, + Name: "run switch to literal (negative delta)", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x81, // delta -1 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: bin.EncodeZigZag(-1), + }, + { + Nums: []float64{ + -1.5, + -1.5, + -1.5, + -1.4, + }, + Name: "run switch to literal (positive delta, negative version)", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x01, // run (len = 3) + 0x82, // delta 1 + 0x80, // literal (len = 1) + 0x00, 0x00, 0x00, + }, + BaseValue: -1.5, + LastDelta: bin.EncodeZigZag(1), + }, + { + Nums: repeatFloat64(1.5, 129), + Name: "increment run to full fill h-byte", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x7f, // run (len = 129) + 0x00, 0x00, 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: 0, + }, + { + Nums: repeatFloat64(-1.5, 129), + Name: "increment run to full fill h-byte (negative version)", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x9d, // 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{ + 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, + }, + { + Nums: repeatFloat64(-1.5, 130), + Name: "run switch to literal after h-byte overflow (negative version)", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x9d, // 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, + 1.6, + 1.7, + }, + Name: "increment literal", + Offset: 1, + ChangeSize: 2, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x82, // delta 1 + 0x84, // delta 2 + 0x82, // literal (len = 3) + 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: bin.EncodeZigZag(2), + }, + { + Nums: []float64{ + -1.5, + -1.6, + -1.7, + }, + Name: "increment literal (negative version)", + Offset: 1, + ChangeSize: 2, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x81, // delta -1 + 0x83, // delta -2 + 0x82, // literal (len = 3) + 0x00, 0x00, 0x00, + }, + BaseValue: -1.5, + LastDelta: bin.EncodeZigZag(-2), + }, + { + Nums: []float64{ + 1.5, + 1.4, + 1.3, + }, + Name: "increment literal (negative delta)", + Offset: 1, + ChangeSize: 2, + Buf: []byte{ + 0x9e, // base value + 0x80, // delta 0 + 0x81, // delta -1 + 0x83, // delta -2 + 0x82, // literal (len = 3) + 0x00, 0x00, 0x00, + }, + BaseValue: 1.5, + LastDelta: bin.EncodeZigZag(-2), + }, + { + Nums: []float64{ + -1.5, + -1.4, + -1.3, + }, + Name: "increment literal (positive delta, negative version)", + Offset: 1, + ChangeSize: 2, + Buf: []byte{ + 0x9d, // base value + 0x80, // delta 0 + 0x82, // delta 1 + 0x84, // delta 2 + 0x82, // literal (len = 3) + 0x00, 0x00, 0x00, + }, + BaseValue: -1.5, + LastDelta: bin.EncodeZigZag(2), + }, + } +) - for _, testCase := range testCases { - // fmt.Println("-----") - // fmt.Println("nums", testCase.Nums) +func TestInstantDeltaCompressor(t *testing.T) { + for _, testCase := range instantTestCases { var ( tmp = make([]byte, tmpValueSize) metricType = qb.Instant @@ -601,77 +769,35 @@ func TestInstantDeltaCompressor(t *testing.T) { } } -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, // - }, - }, +func TestRestoreInstantDeltaCompressor(t *testing.T) { + for _, testCase := range instantTestCases { + var ( + tmp = make([]byte, tmpValueSize) + metricType = qb.Instant + fracDigits byte = 1 + buf = make([]byte, 8) + payloadSize = 0 + report qb.ValueEvaluationReport + ) + 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) } - ) - for caseIdx, testCase := range testCases { + restored := NewValueDeltaCompressor(metricType, fracDigits, buf, c.Size()) + if restored.baseValue != testCase.BaseValue { + t.Fatalf("%s: got baseValue %v are not equal expected %v", + testCase.Name, restored.baseValue, testCase.BaseValue) + } + if restored.lastDelta != testCase.LastDelta { + t.Fatalf("%s: got lastDelta %d are not equal expected %d", + testCase.Name, restored.lastDelta, testCase.LastDelta) + } + } +} + +func TestInstantDeltaDecompressorFromState(t *testing.T) { + for _, testCase := range instantTestCases { var ( metricType = qb.Instant fracDigits byte = 1 @@ -686,18 +812,9 @@ func TestInstantDeltaDecompressorFromState(t *testing.T) { c.Append(report.Offset, tmp[:report.ChangeSize], num, report.Delta) } - // - c.CaptureState() - - fmt.Println("---------------") - - d := c.CreateDecompressor() - - //fmt.Println(buf) - + d := c.CreateDecompressor(metricType, fracDigits) for { num, done := d.NextValue() - //fmt.Println(num, done) if done { break } @@ -706,191 +823,222 @@ func TestInstantDeltaDecompressorFromState(t *testing.T) { slices.Reverse(decodedNums) - fmt.Println(testCase.Nums) - fmt.Println(decodedNums) - fmt.Println() + if !equalFloatSlices(testCase.Nums, decodedNums, 0.0000001) { + t.Fatalf("%s: got nums %v not equal expected %v", + testCase.Name, decodedNums, testCase.Nums) + } + } +} + +func TestInstantDeltaDecompressorFromEnd(t *testing.T) { + for _, testCase := range instantTestCases { + var ( + metricType = qb.Instant + fracDigits byte = 1 + tmp = make([]byte, tmpValueSize) + buf = make([]byte, 8) + payloadSize = 0 + decodedNums []float64 + ) + 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) + } + + d := NewValueDeltaDecompressor(metricType, fracDigits) + d.RestoreFromEnd(buf[:c.Size()]) + for { + num, done := d.NextValue() + if done { + break + } + decodedNums = append(decodedNums, num) + } + + slices.Reverse(decodedNums) if !equalFloatSlices(testCase.Nums, decodedNums, 0.0000001) { - t.Fatalf("%d: got nums %v not equal expected %v", caseIdx, decodedNums, testCase.Nums) + t.Fatalf("%s: got nums %v not equal expected %v", + testCase.Name, decodedNums, testCase.Nums) } } } // TIME DELTA -func TestTimeDeltaCompressor(t *testing.T) { - var ( - testCases = []struct { - Name string - Nums []uint32 - Buf []byte - LastUnixtime uint32 - LastDelta uint32 - Offset int - ChangeSize int - }{ - { - Nums: []uint32{ - 1780777000, - }, - Name: "add 1st value", - Offset: 0, - ChangeSize: 4, - Buf: []byte{ - 0x00, 0x00, 0x00, 0x00, - 0x28, 0x80, 0x24, 0x6a, // since - }, - LastUnixtime: 1780777000, - LastDelta: 0, +var ( + timeTestCases = []struct { + Name string + Nums []uint32 + Buf []byte + LastUnixtime uint32 + LastDelta uint32 + Offset int + ChangeSize int + }{ + { + Nums: []uint32{ + 1780777000, }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - }, - Name: "add 1st delta", - Offset: 0, - ChangeSize: 2, - Buf: []byte{ - 0x00, 0x00, - 0x80, // h-byte (literal, len=1) - 0xbc, // delta (60) - 0x28, 0x80, 0x24, 0x6a, // since - }, - LastUnixtime: 1780777060, - LastDelta: 60, + Name: "add 1st value", + Offset: 0, + ChangeSize: 4, + Buf: []byte{ + 0x00, 0x00, 0x00, 0x00, + 0x28, 0x80, 0x24, 0x6a, // since }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - }, - Name: "literal changed to run", - Offset: 1, - ChangeSize: 1, - Buf: []byte{ - 0x00, 0x00, - 0x00, // h-byte (run, len=2) - 0xbc, // delta - 0x28, 0x80, 0x24, 0x6a, // since - }, - LastUnixtime: 1780777120, - LastDelta: 60, + LastUnixtime: 1780777000, + LastDelta: 0, + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777130, // +70 - 1780777200, // +70 - }, - Name: "literal decrease by 1 and switch to run", - Offset: 2, - ChangeSize: 3, - Buf: []byte{ - 0x00, // h-byte (run, len=2) - 0xc6, // delta 70 - 0x80, // h-byte (literal, len=1) - 0xbc, // delta 60 - 0x28, 0x80, 0x24, 0x6a, // since - }, - LastUnixtime: 1780777200, - LastDelta: 70, + Name: "add 1st delta", + Offset: 0, + ChangeSize: 2, + Buf: []byte{ + 0x00, 0x00, + 0x80, // h-byte (literal, len=1) + 0xbc, // delta (60) + 0x28, 0x80, 0x24, 0x6a, // since }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - 1780777180, // +60 - }, - Name: "increment run", - Offset: 1, - ChangeSize: 1, - Buf: []byte{ - 0x00, 0x00, - 0x01, // h-byte (run, len=2) - 0xbc, // delta 60 - 0x28, 0x80, 0x24, 0x6a, // since - }, - LastUnixtime: 1780777180, - LastDelta: 60, + LastUnixtime: 1780777060, + LastDelta: 60, + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777120, // +60 - 1780777180, // +60 - 1780777200, // +20 - }, - Name: "switch run to literal", - Offset: 0, - ChangeSize: 2, - 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: 1780777200, - LastDelta: 20, + Name: "literal changed to run", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x00, 0x00, + 0x00, // h-byte (run, len=2) + 0xbc, // delta + 0x28, 0x80, 0x24, 0x6a, // since }, - { - 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, + LastUnixtime: 1780777120, + LastDelta: 60, + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777130, // +70 + 1780777200, // +70 }, - { - 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, + Name: "literal decrease by 1 and switch to run", + Offset: 2, + ChangeSize: 3, + Buf: []byte{ + 0x00, // h-byte (run, len=2) + 0xc6, // delta 70 + 0x80, // h-byte (literal, len=1) + 0xbc, // delta 60 + 0x28, 0x80, 0x24, 0x6a, // since }, - { - Nums: []uint32{ - 1780777000, - 1780777060, // +60 - 1780777130, // +70 - }, - Name: "increment literal", - Offset: 1, - ChangeSize: 2, - Buf: []byte{ - 0x00, - 0x81, // h-byte (literal, len=2) - 0xc6, // delta 70 - 0xbc, // delta 60 - 0x28, 0x80, 0x24, 0x6a, // since - }, - LastUnixtime: 1780777130, - LastDelta: 70, + LastUnixtime: 1780777200, + LastDelta: 70, + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 + 1780777180, // +60 }, - } - ) + Name: "increment run", + Offset: 1, + ChangeSize: 1, + Buf: []byte{ + 0x00, 0x00, + 0x01, // h-byte (run, len=2) + 0xbc, // delta 60 + 0x28, 0x80, 0x24, 0x6a, // since + }, + LastUnixtime: 1780777180, + LastDelta: 60, + }, + { + Nums: []uint32{ + 1780777000, + 1780777060, // +60 + 1780777120, // +60 + 1780777180, // +60 + 1780777200, // +20 + }, + Name: "switch run to literal", + Offset: 0, + ChangeSize: 2, + 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: 1780777200, + LastDelta: 20, + }, + { + 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, + 1780777060, // +60 + 1780777130, // +70 + }, + Name: "increment literal", + Offset: 1, + ChangeSize: 2, + Buf: []byte{ + 0x00, + 0x81, // h-byte (literal, len=2) + 0xc6, // delta 70 + 0xbc, // delta 60 + 0x28, 0x80, 0x24, 0x6a, // since + }, + LastUnixtime: 1780777130, + LastDelta: 70, + }, + } +) - for _, testCase := range testCases { - //fmt.Println("----------") +func TestTimeDeltaCompressor(t *testing.T) { + for _, testCase := range timeTestCases { var ( tmp = make([]byte, tmpTimeSize) buf = make([]byte, 8) @@ -900,11 +1048,6 @@ func TestTimeDeltaCompressor(t *testing.T) { c := NewTimeDeltaCompressor(buf, payloadSize) for _, num := range testCase.Nums { report = c.Evaluate(tmp, num) - - // fmt.Println("----------\npos", c.pos) - // fmt.Printf("buf: % x\n", c.buf[c.pos:]) - // pretty.Println(report) - // fmt.Printf("change: % x\n", tmp[:report.ChangeSize]) c.Append(report.Offset, tmp[:report.ChangeSize], num) } if report.Offset != testCase.Offset { @@ -930,82 +1073,36 @@ func TestTimeDeltaCompressor(t *testing.T) { } } -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 { +func TestRestoreTimeDeltaCompressor(t *testing.T) { + for _, testCase := range timeTestCases { var ( tmp = make([]byte, tmpTimeSize) - buf = make([]byte, 16) + buf = make([]byte, 8) + payloadSize = 0 + report qb.TimeEvaluationReport + ) + c := NewTimeDeltaCompressor(buf, payloadSize) + for _, num := range testCase.Nums { + report = c.Evaluate(tmp, num) + c.Append(report.Offset, tmp[:report.ChangeSize], num) + } + restored := NewTimeDeltaCompressor(buf, c.Size()) + if restored.lastUnixtime != testCase.LastUnixtime { + t.Fatalf("%s: got lastUnixtime %d are not equal expected %d", + testCase.Name, restored.lastUnixtime, testCase.LastUnixtime) + } + if restored.lastDelta != testCase.LastDelta { + t.Fatalf("%s: got lastDelta %d are not equal expected %d", + testCase.Name, restored.lastDelta, testCase.LastDelta) + } + } +} + +func TestTimeDeltaDecompressorFromState(t *testing.T) { + for _, testCase := range timeTestCases { + var ( + tmp = make([]byte, tmpTimeSize) + buf = make([]byte, 8) payloadSize = 0 decodedNums []uint32 ) @@ -1015,18 +1112,9 @@ func TestTimeDeltaDecompressorFromState(t *testing.T) { c.Append(report.Offset, tmp[:report.ChangeSize], 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 } @@ -1036,7 +1124,43 @@ func TestTimeDeltaDecompressorFromState(t *testing.T) { slices.Reverse(decodedNums) if !slices.Equal(testCase.Nums, decodedNums) { - t.Fatalf("%d: got nums %v not equal expected %v", caseIdx, decodedNums, testCase.Nums) + t.Fatalf("%s: got nums %v not equal expected %v", + testCase.Name, decodedNums, testCase.Nums) + } + } +} + +func TestTimeDeltaDecompressorFromEnd(t *testing.T) { + for _, testCase := range timeTestCases { + var ( + tmp = make([]byte, tmpTimeSize) + buf = make([]byte, 8) + payloadSize = 0 + decodedNums []uint32 + ) + c := NewTimeDeltaCompressor(buf, payloadSize) + for _, num := range testCase.Nums { + report := c.Evaluate(tmp, num) + c.Append(report.Offset, tmp[:report.ChangeSize], num) + } + + c.ReplaceSinceWithUntil() + + d := NewTimeDeltaDecompressor() + d.RestoreFromEnd(buf[len(buf)-c.Size():]) + for { + num, done := d.NextValue() + if done { + break + } + decodedNums = append(decodedNums, num) + } + + slices.Reverse(decodedNums) + + if !slices.Equal(testCase.Nums, decodedNums) { + t.Fatalf("%s: got nums %v not equal expected %v", + testCase.Name, decodedNums, testCase.Nums) } } } diff --git a/enc/time_delta.go b/enc/time_delta.go index 529ca18..9bea0d0 100644 --- a/enc/time_delta.go +++ b/enc/time_delta.go @@ -28,6 +28,7 @@ type TimeDeltaCompressor struct { pos int lastUnixtime uint32 lastDelta uint32 + state *qb.TimeDeltaCapturedState } // Початкове створення, коли даних немає @@ -42,29 +43,57 @@ func NewTimeDeltaCompressor(buf []byte, payloadSize int) *TimeDeltaCompressor { pos: len(buf) - payloadSize, } if payloadSize > 0 { - var err error - s.lastUnixtime, err = bin.GetUint32(s.buf[len(s.buf)-4:]) - if err != nil { - log.Fatalf("bug: get since: %s", err) - } - //s.lastUnixtime = lastTimestamp // fix - порахувати - - if payloadSize > 4 { - // u64, _, err := bin.GetVarUint64(s.buf[s.pos+1:]) - // if err != nil { - // log.Fatalf("bug: get last delta: %s", err) - // } - // s.lastDelta = uint32(u64) - } + s.restore(payloadSize) } return s } +func (s *TimeDeltaCompressor) restore(payloadSize int) { + bound := len(s.buf) - 4 + since, err := bin.GetUint32(s.buf[bound:]) + if err != nil { + log.Fatalf("bug: get since: %s", err) + } + if payloadSize == 4 { + s.lastUnixtime = since + return + } + var ( + i = s.pos + totalDelta uint64 + ) + for i < bound { + h := s.buf[i] + i++ + if h < 128 { + // run + count := h + 2 + delta, n, _ := bin.GetVarUint64(s.buf[i:]) + i += n + totalDelta += delta * uint64(count) + } else { + // literal + count := (h & 127) + 1 + for range count { + delta, n, _ := bin.GetVarUint64(s.buf[i:]) + i += n + totalDelta += delta + } + } + } + u64, _, err := bin.GetVarUint64(s.buf[s.pos+1:]) + if err != nil { + log.Fatalf("bug: get last delta: %s", err) + } + s.lastDelta = uint32(u64) + s.lastUnixtime = since + uint32(totalDelta) +} + func (s *TimeDeltaCompressor) Evaluate(tmp []byte, timestamp uint32) qb.TimeEvaluationReport { var ( - delta = timestamp - s.lastUnixtime - offset int - i int + delta = timestamp - s.lastUnixtime + rewindOffset int + i int ) if s.pos < len(s.buf) { if s.lastDelta > 0 { @@ -75,7 +104,7 @@ func (s *TimeDeltaCompressor) Evaluate(tmp []byte, timestamp uint32) qb.TimeEval // incrementRun tmp[i] = h + 1 i++ - offset = 1 // перезапис h + rewindOffset = 1 // перезапис h } else { // endSeries n, _ := bin.PutVarUint64(tmp, uint64(delta)) @@ -92,7 +121,7 @@ func (s *TimeDeltaCompressor) Evaluate(tmp []byte, timestamp uint32) qb.TimeEval i += n tmp[i] = h + 1 i++ - offset = 1 // перезапис h + rewindOffset = 1 // перезапис h } else { // endSeries n, _ := bin.PutVarUint64(tmp, uint64(delta)) @@ -109,11 +138,11 @@ func (s *TimeDeltaCompressor) Evaluate(tmp []byte, timestamp uint32) qb.TimeEval i += n tmp[i] = 0 // start new run (length=2) i++ - offset = 1 + n // перезапис пари delta/h + rewindOffset = 1 + n // перезапис пари delta/h } else { tmp[i] = 0 // change literal (length=1) to run (length=2) i++ - offset = 1 + rewindOffset = 1 } } } @@ -129,15 +158,16 @@ func (s *TimeDeltaCompressor) Evaluate(tmp []byte, timestamp uint32) qb.TimeEval i += 4 } return qb.TimeEvaluationReport{ - Offset: offset, - ChangeSize: i, - TotalSpace: s.pos - offset + i, + RewindOffset: rewindOffset, + Offset: s.pos + rewindOffset, + ChangeSize: i, + TotalSpace: s.pos - rewindOffset + i, } } -func (s *TimeDeltaCompressor) Append(offset int, change []byte, timestamp uint32) { +func (s *TimeDeltaCompressor) Append(rewindOffset int, change []byte, timestamp uint32) { if s.pos < len(s.buf) { - i := s.pos + offset - 1 // -1, because s.pos always points to h byte + i := s.pos + rewindOffset - 1 // -1, because s.pos always points to h byte for _, b := range change { s.buf[i] = b i-- @@ -146,7 +176,7 @@ func (s *TimeDeltaCompressor) Append(offset int, change []byte, timestamp uint32 } else { copy(s.buf[len(s.buf)-4:], change) // 4b since } - s.pos -= len(change) - offset + s.pos -= len(change) - rewindOffset s.lastUnixtime = timestamp } @@ -154,17 +184,25 @@ func (s *TimeDeltaCompressor) Append(offset int, change []byte, timestamp uint32 // } -// delta h -func (s *TimeDeltaCompressor) CaptureState() qb.TimeDeltaCapturedState { - // позиція посувається вліво, отже може перескочити на попередній chunk +func (s *TimeDeltaCompressor) getState() qb.TimeDeltaCapturedState { + bound := s.pos + 1 + bin.CountVarUint64(uint64(s.lastDelta)) return qb.TimeDeltaCapturedState{ H: s.buf[s.pos], LastUnixtime: s.lastUnixtime, LastDelta: s.lastDelta, - Payload: s.buf[s.pos:], + Payload: s.buf[bound:], } } +func (s *TimeDeltaCompressor) CaptureState() { + state := s.getState() + s.state = &state +} + +func (s *TimeDeltaCompressor) Tail(offset int) []byte { + return s.buf[s.pos : len(s.buf)-offset] +} + // коли сторінка заповнена і відправляється в txlog, since замінюю на until func (s *TimeDeltaCompressor) ReplaceSinceWithUntil() uint32 { pos := len(s.buf) - 4 @@ -178,12 +216,13 @@ func (s *TimeDeltaCompressor) Size() int { return len(s.buf) - s.pos } -// if s.state == nil { -// return len(s.buf) - s.pos -// } else { -// return bin.CountVarUint64(uint64(s.state.LastDelta)) + hSize + len(s.state.Payload) -// } -// } +func (s *TimeDeltaCompressor) StoredSize() int { + if s.state == nil { + return len(s.buf) - s.pos + } else { + return bin.CountVarUint64(uint64(s.state.LastDelta)) + 1 + len(s.state.Payload) + } +} // Snapshot - для створення снапшота. // FIX - encode firstUnixtime in 1st 4 bytes @@ -217,9 +256,7 @@ func (s *TimeDeltaCompressor) WritePayloadTo(w io.Writer, state *qb.TimeDeltaCap } } -func (s *TimeDeltaCompressor) Rotate(newbuf []byte) { - // УВАГА! - // state не чіпаємо +func (s *TimeDeltaCompressor) ReplaceBuffer(newbuf []byte) { s.buf = newbuf s.pos = len(s.buf) s.lastUnixtime = 0 @@ -230,12 +267,18 @@ func (s *TimeDeltaCompressor) CreateDecompressor() qb.TimestampDecompressor { if s.pos == len(s.buf) { return nil } - return NewTimeDeltaDecompressorFromState(TimeDeltaDecompressorFromStateOptions{ - H: s.buf[s.pos], - LastUnixtime: s.lastUnixtime, - LastDelta: s.lastDelta, - Payload: s.buf[s.pos:], - }) + d := NewTimeDeltaDecompressor() + if s.state != nil { + d.RestoreFromState(*s.state) + } else { + d.RestoreFromState(s.getState()) + } + // fmt.Printf("h: % x\n", s.buf[s.pos]) + // fmt.Printf("lastUnixtime: % x\n", s.lastUnixtime) + // fmt.Printf("lastDelta: % x\n", s.lastDelta) + // fmt.Printf("buf: % x\n", s.buf) + // fmt.Printf("pay: % x\n", s.buf[bound:]) + return d } func (s *TimeDeltaCompressor) LastTimestamp() uint32 { @@ -254,66 +297,57 @@ type TimeDeltaDecompressor struct { done bool } -func NewTimeDeltaDecompressor(buf []byte) *TimeDeltaDecompressor { - s := &TimeDeltaDecompressor{ - buf: buf, - } - if s.pos < len(s.buf) { - var err error - s.lastUnixtime, err = bin.GetUint32(s.buf) - if err != nil { - log.Fatalf("bug: get last unixtime: %s", err) - } - s.pos += 4 - } else { - s.done = true - } - return s +func NewTimeDeltaDecompressor() *TimeDeltaDecompressor { + return new(TimeDeltaDecompressor) } -type TimeDeltaDecompressorFromStateOptions struct { - H byte - LastUnixtime uint32 - LastDelta uint32 - Payload []byte +// викликається для повних сторінок +func (s *TimeDeltaDecompressor) RestoreFromEnd(buf []byte) { + s.buf = buf + var err error + s.lastUnixtime, err = bin.GetUint32(buf[len(buf)-4:]) + if err != nil { + log.Fatalf("bug: get last unixtime: %s", err) + } + if len(buf) > 4 { + s.readHeader() + s.readDelta() + } + // fmt.Printf("h: % x\n", s.buf[s.pos]) + // fmt.Printf("lastUnixtime: %d\n", s.lastUnixtime) + // fmt.Printf("lastDelta: %d\n", s.lastDelta) + // fmt.Printf("buf: % x\n", s.buf) } -func NewTimeDeltaDecompressorFromState(opt TimeDeltaDecompressorFromStateOptions) *TimeDeltaDecompressor { - s := &TimeDeltaDecompressor{ - buf: opt.Payload, - lastUnixtime: opt.LastUnixtime, +// викликається для data level tail +func (s *TimeDeltaDecompressor) RestoreFromState(state qb.TimeDeltaCapturedState) { + s.buf = state.Payload + s.lastUnixtime = state.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) - } - return s } func (s *TimeDeltaDecompressor) NextValue() (value uint32, done bool) { if s.done { return 0, true } - // повертаю значення, що було прочитано в методі RestoreFromBound/RestoreFromEnd value = s.lastUnixtime if s.lastDelta > 0 { s.lastUnixtime -= s.lastDelta s.pending-- if s.pending > 0 { - // якщо в серії залишаються елементи if !s.isRun { s.readDelta() } - } else if s.pos < len(s.buf) { - // в серії більше немає елементів, отже перевіряє чи є ще дані в буфері. - // дані є - читаю заголовок наступної серії + } else if s.pos < len(s.buf)-4 { s.readHeader() s.readDelta() } else { s.lastDelta = 0 } } else { - // був закодований лише last unixtime s.done = true } return value, false diff --git a/enc/value_delta.go b/enc/value_delta.go index ddb596f..f42b266 100644 --- a/enc/value_delta.go +++ b/enc/value_delta.go @@ -25,19 +25,19 @@ type ValueDeltaCompressor struct { toUint64 func(float64, float64) uint64 // (value, coef) => uint64 toFloat64 func(uint64, float64) float64 + state *qb.ValueDeltaCapturedState } // Після відновлення із снапшота 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)) - } s := &ValueDeltaCompressor{ buf: buf, - coef: coef, + coef: 1, pos: payloadSize, } + if fracDigits > 0 { + s.coef = math.Pow(10, float64(fracDigits)) + } if metricType == qb.Cumulative { s.calcDelta = calcCumulativeDelta s.toUint64 = toCumulativeUint64 @@ -48,16 +48,19 @@ func NewValueDeltaCompressor(metricType qb.MetricType, fracDigits byte, buf []by s.toFloat64 = toInstantFloat64 } if payloadSize > 0 { - // base value на початку + // fmt.Printf("% x\n", buf) + // fmt.Printf("% x\n", buf[:s.pos]) + // fmt.Printf("pos: %d\n", s.pos) u64, _, err := bin.GetVarUint64(s.buf) if err != nil { log.Fatalf("bug: get base value: %s", err) } - s.baseValue = float64(u64) / s.coef - s.lastDelta, _, err = bin.ReverseGetVarUint64(s.buf[:s.pos-2]) // skip h byte + s.baseValue = s.toFloat64(u64, s.coef) + s.lastDelta, _, err = bin.ReverseGetVarUint64(s.buf[:s.pos-1]) // skip h byte if err != nil { log.Fatalf("bug: get last delta: %s", err) } + //fmt.Printf("lastDelta: %d\n", s.lastDelta) } return s } @@ -68,9 +71,9 @@ func NewValueDeltaCompressor(metricType qb.MetricType, fracDigits byte, buf []by // arr - func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvaluationReport { var ( - delta uint64 - offset int - i int + delta uint64 + rewindOffset int + i int ) if s.pos > 0 { delta = s.calcDelta(s.baseValue, s.coef, value) @@ -81,7 +84,7 @@ func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvalu // incrementRun tmp[i] = h + 1 i++ - offset = 1 // перезапис h + rewindOffset = 1 // перезапис h } else { // endSeries n, _ := bin.ReversePutVarUint64(tmp, delta) @@ -98,7 +101,7 @@ func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvalu i += n tmp[i] = h + 1 i++ - offset = 1 // перезапис h + rewindOffset = 1 // перезапис h } else { // endSeries n, _ := bin.ReversePutVarUint64(tmp, delta) @@ -115,11 +118,11 @@ func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvalu i += n tmp[i] = 0 // start new run (length=2) i++ - offset = 1 + n // перезапис пари delta/h + rewindOffset = 1 + n // перезапис пари delta/h } else { tmp[i] = 0 // change literal (length=1) to run (length=2) i++ - offset = 1 + rewindOffset = 1 } } } @@ -132,51 +135,61 @@ func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvalu i++ } return qb.ValueEvaluationReport{ - Offset: offset, - ChangeSize: i, - TotalSpace: s.pos + i - offset, - Delta: delta, + RewindOffset: rewindOffset, + Offset: s.pos - rewindOffset, + ChangeSize: i, + TotalSpace: s.pos + i - rewindOffset, + Delta: delta, } } // pos завжди вказує на h -func (s *ValueDeltaCompressor) Append(offset int, change []byte, value float64, delta uint64) { +func (s *ValueDeltaCompressor) Append(rewindOffset int, change []byte, value float64, delta uint64) { if s.pos > 0 { s.lastDelta = delta } else { s.baseValue = value } - copy(s.buf[s.pos-offset:], change) - s.pos += len(change) - offset + copy(s.buf[s.pos-rewindOffset:], change) + s.pos += len(change) - rewindOffset } func (s *ValueDeltaCompressor) DeleteLast() { } -func (s *ValueDeltaCompressor) CaptureState() qb.ValueDeltaCapturedState { - // if s.state != nil { - // qb.Abort(qb.RepeatableLock, nil) - // } - // позиція посувається вліво, отже може перескочити на попередній chunk - pos := s.pos - 1 - bin.CountVarUint64(s.lastDelta) +func (s *ValueDeltaCompressor) getState() qb.ValueDeltaCapturedState { + bound := s.pos - 1 - bin.CountVarUint64(s.lastDelta) return qb.ValueDeltaCapturedState{ H: s.buf[s.pos-1], LastDelta: s.lastDelta, - Payload: s.buf[:pos], + Payload: s.buf[:bound], } } +func (s *ValueDeltaCompressor) CaptureState() { + state := s.getState() + s.state = &state +} + +func (s *ValueDeltaCompressor) Tail(offset int) []byte { + return s.buf[offset:s.pos] +} + // для зростання буфера під час вставки даних. 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) - // } } -// // Snapshot - для створення снапшота. +func (s *ValueDeltaCompressor) StoredSize() int { + if s.state == nil { + return len(s.buf) - s.pos + } else { + return bin.CountVarUint64(uint64(s.state.LastDelta)) + 1 + len(s.state.Payload) + } +} + +// Snapshot - для створення снапшота. // func (s *ValueDeltaCompressor) Payload() []byte { // if s.state == nil { // return s.buf[:s.pos] @@ -205,9 +218,7 @@ func (s *ValueDeltaCompressor) WritePayloadTo(w io.Writer, state *qb.ValueDeltaC } } -func (s *ValueDeltaCompressor) Rotate(newbuf []byte) { - // УВАГА! - // state не чіпаємо +func (s *ValueDeltaCompressor) ReplaceBuffer(newbuf []byte) { s.buf = newbuf s.pos = 0 s.baseValue = 0 @@ -226,17 +237,18 @@ func (s *ValueDeltaCompressor) LastValue() float64 { return 0 } -func (s *ValueDeltaCompressor) CreateDecompressor() qb.ValueDecompressor { +// Decompressor читає виключно збережені на диск дані +func (s *ValueDeltaCompressor) CreateDecompressor(metricType qb.MetricType, fracDigits byte) qb.ValueDecompressor { if s.pos == 0 { return nil } - return NewValueDeltaDecompressorFromState(ValueDeltaDecompressorFromStateOptions{ - Coef: s.coef, - ToFloat64: s.toFloat64, - H: s.buf[s.pos-1], - LastDelta: s.lastDelta, - Payload: s.buf[:s.pos], - }) + d := NewValueDeltaDecompressor(metricType, fracDigits) + if s.state != nil { + d.RestoreFromState(*s.state) + } else { + d.RestoreFromState(s.getState()) + } + return d } // DECOMPRESSOR @@ -255,78 +267,62 @@ type ValueDeltaDecompressor struct { toFloat64 func(uint64, float64) float64 } -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)) - } - u64, n, err := bin.GetVarUint64(buf) - if err != nil { - log.Fatalf("bug: get base value: %s", err) - } +func NewValueDeltaDecompressor(metricType qb.MetricType, fracDigits byte) *ValueDeltaDecompressor { s := &ValueDeltaDecompressor{ - buf: buf, - coef: coef, - pos: len(buf), // first free - bound: n, - toFloat64: toFloat64, + coef: 1, } - s.baseValue = s.toFloat64(u64, coef) - // читаю заголовок наступної серії + if fracDigits > 0 { + s.coef = math.Pow(10, float64(fracDigits)) + } + if metricType == qb.Cumulative { + s.toFloat64 = toCumulativeFloat64 + } else { + s.toFloat64 = toInstantFloat64 + } + return s +} + +func (s *ValueDeltaDecompressor) RestoreFromEnd(buf []byte) { + s.buf = buf + s.pos = len(buf) // first free + s.readBaseValue() s.readHeader() s.readValue() - return s } -type ValueDeltaDecompressorFromStateOptions struct { - Coef float64 - // (value, coef) => uint64 - ToFloat64 func(uint64, float64) float64 - H byte - LastDelta uint64 - Payload []byte +func (s *ValueDeltaDecompressor) RestoreFromState(state qb.ValueDeltaCapturedState) { + s.buf = state.Payload + s.pos = len(s.buf) // first free + s.readBaseValue() + s.lastValue = s.baseValue + s.toFloat64(state.LastDelta, s.coef) + s.decodeHeaderByte(state.H) } -func NewValueDeltaDecompressorFromState(opt ValueDeltaDecompressorFromStateOptions) *ValueDeltaDecompressor { - u64, n, err := bin.GetVarUint64(opt.Payload) +func (s *ValueDeltaDecompressor) readBaseValue() { + u64, n, err := bin.GetVarUint64(s.buf) if err != nil { log.Fatalf("bug: get base value: %s", err) } - s := &ValueDeltaDecompressor{ - buf: opt.Payload, - coef: opt.Coef, - pos: len(opt.Payload), // first free - bound: n, - toFloat64: opt.ToFloat64, - } s.baseValue = s.toFloat64(u64, s.coef) - s.lastValue = s.baseValue + s.toFloat64(opt.LastDelta, s.coef) - s.decodeHeaderByte(opt.H) - return s + s.bound = n } func (s *ValueDeltaDecompressor) 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 } @@ -335,6 +331,7 @@ func (s *ValueDeltaDecompressor) readHeader() { h := s.buf[s.pos] s.decodeHeaderByte(h) } + func (s *ValueDeltaDecompressor) readValue() { u64, n, err := bin.ReverseGetVarUint64(s.buf[:s.pos]) if err != nil { diff --git a/examples/requests/requests.go b/examples/requests/requests.go index 191587b..12c2467 100644 --- a/examples/requests/requests.go +++ b/examples/requests/requests.go @@ -14,7 +14,7 @@ func sendRequests(conn *client.Connection) { var ( instantMetricID uint32 = 10000 cumulativeMetricID uint32 = 10001 - fracDigits int = 2 + fracDigits byte = 2 err error ) diff --git a/proto/proto.go b/proto/proto.go index ad6c25d..6f362a1 100644 --- a/proto/proto.go +++ b/proto/proto.go @@ -162,7 +162,7 @@ func ReadListCurrentValuesReq(r *bufreader.BufferedReader) (m ListCurrentValuesR type AddMetricReq struct { MetricID uint32 MetricType qb.MetricType - FracDigits int + FracDigits byte } func ReadAddMetricReq(r *bufreader.BufferedReader) (m AddMetricReq, err error) { @@ -177,7 +177,7 @@ func ReadAddMetricReq(r *bufreader.BufferedReader) (m AddMetricReq, err error) { func UnpackAddMetricReq(arr []byte) (m AddMetricReq) { m.MetricID, _ = bin.GetUint32(arr) m.MetricType = qb.MetricType(arr[4]) - m.FracDigits = int(arr[5]) + m.FracDigits = arr[5] return } diff --git a/qb.go b/qb.go index 670dcc0..2ddb421 100644 --- a/qb.go +++ b/qb.go @@ -26,16 +26,18 @@ const ( type TimestampCompressor interface { // (tmp, timestamp) Evaluate([]byte, uint32) TimeEvaluationReport - // (offset, change, timestamp) + // (rewindOffset, change, timestamp) Append(int, []byte, uint32) Size() int //Chunks() [][]byte //DeleteLast() - CaptureState() TimeDeltaCapturedState + CaptureState() + // (offset) => payload + Tail(int) []byte CreateDecompressor() TimestampDecompressor //Payload() []byte // для снапшота // Offset() int - Rotate([]byte) + ReplaceBuffer([]byte) WritePayloadTo(io.Writer, *TimeDeltaCapturedState) error LastTimestamp() uint32 ReplaceSinceWithUntil() uint32 @@ -49,25 +51,28 @@ type TimeDeltaCapturedState struct { } type TimeEvaluationReport struct { - TotalSpace int - Offset int - ChangeSize int + TotalSpace int + Offset int // from payload start + RewindOffset int // steps to back from pos + ChangeSize int } type ValueCompressor interface { // (tmp, value) Evaluate([]byte, float64) ValueEvaluationReport - // (offset, change, value, delta) + // (rewindOffset, change, value, delta) Append(int, []byte, float64, uint64) Size() int //Chunks() [][]byte //DeleteLast() - CaptureState() ValueDeltaCapturedState + CaptureState() + // (offset) => payload + Tail(int) []byte // fracDigits - CreateDecompressor() ValueDecompressor + CreateDecompressor(MetricType, byte) ValueDecompressor //Payload() []byte // для снапшота //Offset() int - Rotate([]byte) + ReplaceBuffer([]byte) WritePayloadTo(io.Writer, *ValueDeltaCapturedState) error LastValue() float64 } @@ -79,10 +84,11 @@ type ValueDeltaCapturedState struct { } type ValueEvaluationReport struct { - TotalSpace int - Offset int - ChangeSize int - Delta uint64 + TotalSpace int + Offset int // from payload start + RewindOffset int // steps to back from pos + ChangeSize int + Delta uint64 } type TimestampDecompressor interface { diff --git a/storage/page_preparer.go b/storage/page_preparer.go index afa86be..d3f45da 100644 --- a/storage/page_preparer.go +++ b/storage/page_preparer.go @@ -15,23 +15,11 @@ import ( // fix - reduce leves type PagePreparer struct { - dataChecksumIdx int - maxRecordsOnIndexPage int - indexPageIncSize int // кратно indexPageSize - timestampsSizeIdx int - valuesSizeIdx int - prevPageIdx int - indexRecordsCountIdx int - isZeroLevelIdx int - indexChecksumIdx int - getDataPageNumber func() (uint32, bool, error) - getIndexPageNumber func() (uint32, bool, error) + getDataPageNumber func() (uint32, bool, error) + getIndexPageNumber func() (uint32, bool, error) } type PagePreparerOptions struct { - IndexPageSize int - IndexPageIncSize int - DataPageSize int GetDataPageNumber func() (uint32, bool, error) GetIndexPageNumber func() (uint32, bool, error) } @@ -43,22 +31,10 @@ func NewPagePreparer(opt PagePreparerOptions) (*PagePreparer, error) { if opt.GetDataPageNumber == nil { return nil, errors.New("missing required option: GetDataPageNumber") } - if (opt.IndexPageSize % opt.IndexPageIncSize) != 0 { - return nil, errors.New("IndexPageIncSize must multiple of IndexPageSize") - } s := &PagePreparer{ - dataChecksumIdx: opt.DataPageSize - 4, - indexPageIncSize: opt.IndexPageIncSize, - timestampsSizeIdx: opt.DataPageSize - 6, - valuesSizeIdx: opt.DataPageSize - 8, - prevPageIdx: opt.DataPageSize - 12, - indexRecordsCountIdx: opt.IndexPageSize - 6, - isZeroLevelIdx: opt.IndexPageSize - 7, - indexChecksumIdx: opt.IndexPageSize - 4, - getIndexPageNumber: opt.GetIndexPageNumber, - getDataPageNumber: opt.GetDataPageNumber, + getIndexPageNumber: opt.GetIndexPageNumber, + getDataPageNumber: opt.GetDataPageNumber, } - s.maxRecordsOnIndexPage = (opt.IndexPageSize - 7) / indexRecordSize return s, nil } @@ -70,7 +46,7 @@ type appendIndexRecordIn struct { } func (s *PagePreparer) appendIndexRecord(in appendIndexRecordIn) []byte { - if in.RecordsCount < s.maxRecordsOnIndexPage { + if in.RecordsCount < maxRecordsOnIndexPage { var ( pos int buf = in.Records @@ -78,7 +54,7 @@ func (s *PagePreparer) appendIndexRecord(in appendIndexRecordIn) []byte { if in.RecordsCount > 0 { pos = in.RecordsCount * indexRecordSize } else { - buf = make([]byte, s.indexPageIncSize) + buf = make([]byte, IndexPageSize) // IndexPageIncSize } bin.PutUint32(buf[pos:], in.Timestamp) bin.PutUint32(buf[pos+4:], in.PageNo) @@ -88,39 +64,6 @@ func (s *PagePreparer) appendIndexRecord(in appendIndexRecordIn) []byte { return nil } -type sealDataPageIn struct { - Content []byte - PrevPageNo uint32 - TimestampsSize int - ValuesSize int -} - -func (s *PagePreparer) sealDataPage(in sealDataPageIn) (checksum uint32) { - bin.PutUint16(in.Content[s.timestampsSizeIdx:], uint16(in.TimestampsSize)) - bin.PutUint16(in.Content[s.valuesSizeIdx:], uint16(in.ValuesSize)) - bin.PutUint32(in.Content[s.prevPageIdx:], in.PrevPageNo) - - checksum = util.CalcChecksum(in.Content[:s.dataChecksumIdx]) - bin.PutUint32(in.Content[s.dataChecksumIdx:], checksum) - return -} - -type sealIndexPageIn struct { - Content []byte - RecordsCount int - LastLevel bool -} - -func (s *PagePreparer) sealIndexPage(in sealIndexPageIn) (checksum uint32) { - bin.PutUint16(in.Content[s.indexRecordsCountIdx:], uint16(in.RecordsCount)) - if in.LastLevel { - in.Content[s.isZeroLevelIdx] = 1 - } - checksum = util.CalcChecksum(in.Content[:s.indexChecksumIdx]) - bin.PutUint32(in.Content[s.indexChecksumIdx:], checksum) - return -} - type IndexLevelTail struct { Buffer []byte // розмір більший за кількість RecordsCount int @@ -210,7 +153,7 @@ func (s *PagePreparer) SealPages(in SealPagesIn) (result SealResult) { PrevPageNo: prevPageNo, PageNo: pageNo, Reused: reused, - Checksum: s.sealDataPage(sealDataPageIn{ + Checksum: SealDataPage(SealDataPageIn{ Content: d.Content, PrevPageNo: prevPageNo, TimestampsSize: d.TimestampsSize, @@ -232,7 +175,7 @@ func (s *PagePreparer) SealPages(in SealPagesIn) (result SealResult) { break } level := sealedIndexLevels[levelIdx] - if level.TailRecordsCount < s.maxRecordsOnIndexPage { + if level.TailRecordsCount < maxRecordsOnIndexPage { level.TailRecords = s.appendIndexRecord(appendIndexRecordIn{ Records: level.TailRecords, Timestamp: upTimestamp, @@ -252,10 +195,10 @@ func (s *PagePreparer) SealPages(in SealPagesIn) (result SealResult) { Content: level.TailRecords, PageNo: pageNo, Reused: reused, - Checksum: s.sealIndexPage(sealIndexPageIn{ + Checksum: SealIndexPage(SealIndexPageIn{ Content: level.TailRecords, RecordsCount: level.TailRecordsCount, - LastLevel: levelIdx == 0, + ZeroLevel: levelIdx == 0, }), } level.IndexPages = append(level.IndexPages, filled) @@ -288,12 +231,12 @@ type SealDataPageIn struct { } func SealDataPage(in SealDataPageIn) (checksum uint32) { - // bin.PutUint16(in.Content[s.timestampsSizeIdx:], uint16(in.TimestampsSize)) - // bin.PutUint16(in.Content[s.valuesSizeIdx:], uint16(in.ValuesSize)) - // bin.PutUint32(in.Content[s.prevPageIdx:], in.PrevPageNo) + bin.PutUint16(in.Content[timestampsSizeIdx:], uint16(in.TimestampsSize)) + bin.PutUint16(in.Content[valuesSizeIdx:], uint16(in.ValuesSize)) + bin.PutUint32(in.Content[prevPageIdx:], in.PrevPageNo) - // checksum = util.CalcChecksum(in.Content[:s.dataChecksumIdx]) - // bin.PutUint32(in.Content[s.dataChecksumIdx:], checksum) + checksum = util.CalcChecksum(in.Content[:dataCRC32Idx]) + bin.PutUint32(in.Content[dataCRC32Idx:], checksum) return } @@ -304,11 +247,11 @@ type SealIndexPageIn struct { } func SealIndexPage(in SealIndexPageIn) (checksum uint32) { - // bin.PutUint16(in.Content[s.indexRecordsCountIdx:], uint16(in.RecordsCount)) - // if in.LastLevel { - // in.Content[s.isZeroLevelIdx] = 1 - // } - // checksum = util.CalcChecksum(in.Content[:s.indexChecksumIdx]) - // bin.PutUint32(in.Content[s.indexChecksumIdx:], checksum) + bin.PutUint16(in.Content[indexRecordsCountIdx:], uint16(in.RecordsCount)) + if in.ZeroLevel { + in.Content[isZeroLevelIdx] = 1 + } + checksum = util.CalcChecksum(in.Content[:indexCRC32Idx]) + bin.PutUint32(in.Content[indexCRC32Idx:], checksum) return } diff --git a/storage/wal_records.go b/storage/wal_records.go index ef6fe1f..5e479cd 100644 --- a/storage/wal_records.go +++ b/storage/wal_records.go @@ -360,7 +360,7 @@ func (s *MeasuresAppendRecord) Parse(r *bytes.Buffer) (err error) { type MetricAddRecord struct { MetricID uint32 MetricType qb.MetricType - FracDigits int + FracDigits byte } func (s MetricAddRecord) Pack(w io.Writer) { @@ -368,7 +368,7 @@ func (s MetricAddRecord) Pack(w io.Writer) { CodeMetricAdd, 0, 0, 0, 0, // byte(s.MetricType), - byte(s.FracDigits), + s.FracDigits, } bin.PutUint32(arr[1:], s.MetricID) w.Write(arr) @@ -381,7 +381,7 @@ func (s *MetricAddRecord) Parse(src io.Reader) (err error) { } s.MetricID, _ = bin.GetUint32(arr) s.MetricType = qb.MetricType(arr[4]) - s.FracDigits = int(arr[5]) + s.FracDigits = arr[5] return nil }