diff --git a/tsdb/chunkenc/chunk.go b/tsdb/chunkenc/chunk.go index 5f9349f05..5f228df48 100644 --- a/tsdb/chunkenc/chunk.go +++ b/tsdb/chunkenc/chunk.go @@ -41,14 +41,29 @@ const ( // Chunk holds a sequence of sample pairs that can be iterated over and appended to. type Chunk interface { + // Bytes returns the underlying byte slice of the chunk. Bytes() []byte + + // Encoding returns the encoding type of the chunk. Encoding() Encoding + + // Appender returns an appender to append samples to the chunk. Appender() (Appender, error) + // The iterator passed as argument is for re-use. // Depending on implementation, the iterator can // be re-used or a new iterator can be allocated. Iterator(Iterator) Iterator + + // NumSamples returns the number of samples in the chunk. NumSamples() int + + // Compact is called whenever a chunk is expected to be complete (no more + // samples appended) and the underlying implementation can eventually + // optimize the chunk. + // There's no strong guarantee that no samples will be appended once + // Compact() is called. Implementing this function is optional. + Compact() } // Appender adds sample pairs to a chunk. diff --git a/tsdb/chunkenc/xor.go b/tsdb/chunkenc/xor.go index ce6e0a951..5401261d9 100644 --- a/tsdb/chunkenc/xor.go +++ b/tsdb/chunkenc/xor.go @@ -49,6 +49,10 @@ import ( "math/bits" ) +const ( + chunkCompactCapacityThreshold = 32 +) + // XORChunk holds XOR encoded sample data. type XORChunk struct { b bstream @@ -75,6 +79,14 @@ func (c *XORChunk) NumSamples() int { return int(binary.BigEndian.Uint16(c.Bytes())) } +func (c *XORChunk) Compact() { + if l := len(c.b.stream); cap(c.b.stream) > l+chunkCompactCapacityThreshold { + buf := make([]byte, l) + copy(buf, c.b.stream) + c.b.stream = buf + } +} + // Appender implements the Chunk interface. func (c *XORChunk) Appender() (Appender, error) { it := c.iterator(nil) diff --git a/tsdb/head.go b/tsdb/head.go index ce4a36719..5769c770a 100644 --- a/tsdb/head.go +++ b/tsdb/head.go @@ -1691,6 +1691,11 @@ func (s *memSeries) cut(mint int64) *memChunk { s.chunks = append(s.chunks, c) s.headChunk = c + // Remove exceeding capacity from the previous chunk byte slice to save memory. + if l := len(s.chunks); l > 1 { + s.chunks[l-2].chunk.Compact() + } + // Set upper bound on when the next chunk must be started. An earlier timestamp // may be chosen dynamically at a later point. s.nextAt = rangeForTimestamp(mint, s.chunkRange)