Merge pull request #6686 from influxdata/jw-timestamp

Optimize timestamp run-length decoding

CHANGELOG.md

@@ -18,6 +18,7 @@
 - [#6621](https://github.com/influxdata/influxdb/pull/6621): Add Holt-Winter forecasting function.
 - [#6655](https://github.com/influxdata/influxdb/issues/6655): Add HTTP(s) based subscriptions.
 - [#5906](https://github.com/influxdata/influxdb/issues/5906): Dynamically update the documentation link in the admin UI.
+- [#6686](https://github.com/influxdata/influxdb/pull/6686): Optimize timestamp run-length decoding
 
 ### Bugfixes
 

tsdb/engine/tsm1/timestamp.go

@@ -184,22 +184,40 @@ func (e *encoder) encodeRLE(first, delta, div uint64, n int) ([]byte, error) {
 }
 
 type TimeDecoder struct {
-	v   int64
-	i   int
-	ts  []uint64
-	dec simple8b.Decoder
-	err error
+	v    int64
+	i, n int
+	ts   []uint64
+	dec  simple8b.Decoder
+	err  error
+
+	// The delta value for a run-length encoded byte slice
+	rleDelta int64
+
+	encoding byte
 }
 
 func (d *TimeDecoder) Init(b []byte) {
 	d.v = 0
 	d.i = 0
 	d.ts = d.ts[:0]
 	d.err = nil
+	if len(b) > 0 {
+		// Encoding type is stored in the 4 high bits of the first byte
+		d.encoding = b[0] >> 4
+	}
 	d.decode(b)
 }
 
 func (d *TimeDecoder) Next() bool {
+	if d.encoding == timeCompressedRLE {
+		if d.i >= d.n {
+			return false
+		}
+		d.i++
+		d.v += d.rleDelta
+		return d.i < d.n
+	}
+
 	if d.i >= len(d.ts) {
 		return false
 	}
@@ -221,17 +239,15 @@ func (d *TimeDecoder) decode(b []byte) {
 		return
 	}
 
-	// Encoding type is stored in the 4 high bits of the first byte
-	encoding := b[0] >> 4
-	switch encoding {
+	switch d.encoding {
 	case timeUncompressed:
 		d.decodeRaw(b[1:])
 	case timeCompressedRLE:
 		d.decodeRLE(b)
 	case timeCompressedPackedSimple:
 		d.decodePacked(b)
 	default:
-		d.err = fmt.Errorf("unknown encoding: %v", encoding)
+		d.err = fmt.Errorf("unknown encoding: %v", d.encoding)
 	}
 }
 
@@ -279,20 +295,11 @@ func (d *TimeDecoder) decodeRLE(b []byte) {
 	// Last 1-10 bytes is how many times the value repeats
 	count, _ := binary.Uvarint(b[i:])
 
-	// Rebuild construct the original values now
-	deltas := d.ts[:0]
-	for i := 0; i < int(count); i++ {
-		deltas = append(deltas, value)
-	}
-
-	// Reverse the delta-encoding
-	deltas[0] = first
-	for i := 1; i < len(deltas); i++ {
-		deltas[i] = deltas[i-1] + deltas[i]
-	}
+	d.v = int64(first - value)
+	d.rleDelta = int64(value)
 
-	d.i = 0
-	d.ts = deltas
+	d.i = -1
+	d.n = int(count)
 }
 
 func (d *TimeDecoder) decodeRaw(b []byte) {