From 57c18420abbd34877bf57cd5bd1572d36e155e37 Mon Sep 17 00:00:00 2001 From: beorn7 Date: Wed, 18 Jan 2023 13:47:22 +0100 Subject: [PATCH] histograms: General readability tweaks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Adjust doc comments to go1.19 style. - Break down some overly long lines. - Minor doc comment tweaks and fixes. - Some renaming. Some rationales for the last point: I have renamed “interjections” into “inserts”, mostly because it is shorter, and the word shows up a lot by now (and the concept is cryptic enough to not obfuscate it even more with abbreviations). I have also tried to find more descriptive naming for the “compare spans” functions. Signed-off-by: beorn7 --- tsdb/chunkenc/float_histogram.go | 55 ++++---- tsdb/chunkenc/float_histogram_test.go | 2 +- tsdb/chunkenc/histogram.go | 68 ++++----- tsdb/chunkenc/histogram_meta.go | 196 +++++++++++++------------- tsdb/chunkenc/histogram_meta_test.go | 58 ++++---- tsdb/chunkenc/histogram_test.go | 2 +- tsdb/head_append.go | 49 ++++--- 7 files changed, 218 insertions(+), 212 deletions(-) diff --git a/tsdb/chunkenc/float_histogram.go b/tsdb/chunkenc/float_histogram.go index 538af364ae..437a5712e2 100644 --- a/tsdb/chunkenc/float_histogram.go +++ b/tsdb/chunkenc/float_histogram.go @@ -217,10 +217,10 @@ func (a *FloatHistogramAppender) AppendHistogram(int64, *histogram.Histogram) { panic("appended an integer histogram to a float histogram chunk") } -// Appendable returns whether the chunk can be appended to, and if so -// whether any recoding needs to happen using the provided interjections -// (in case of any new buckets, positive or negative range, respectively). -// If the sample is a gauge histogram, AppendableGauge must be used instead. +// Appendable returns whether the chunk can be appended to, and if so whether +// any recoding needs to happen using the provided inserts (in case of any new +// buckets, positive or negative range, respectively). If the sample is a gauge +// histogram, AppendableGauge must be used instead. // // The chunk is not appendable in the following cases: // - The schema has changed. @@ -233,7 +233,7 @@ func (a *FloatHistogramAppender) AppendHistogram(int64, *histogram.Histogram) { // because of a counter reset. If the given sample is stale, it is always ok to // append. If counterReset is true, okToAppend is always false. func (a *FloatHistogramAppender) Appendable(h *histogram.FloatHistogram) ( - positiveInterjections, negativeInterjections []Interjection, + positiveInserts, negativeInserts []Insert, okToAppend, counterReset bool, ) { if a.NumSamples() > 0 && a.GetCounterResetHeader() == GaugeType { @@ -267,12 +267,12 @@ func (a *FloatHistogramAppender) Appendable(h *histogram.FloatHistogram) ( } var ok bool - positiveInterjections, ok = forwardCompareSpans(a.pSpans, h.PositiveSpans) + positiveInserts, ok = expandSpansForward(a.pSpans, h.PositiveSpans) if !ok { counterReset = true return } - negativeInterjections, ok = forwardCompareSpans(a.nSpans, h.NegativeSpans) + negativeInserts, ok = expandSpansForward(a.nSpans, h.NegativeSpans) if !ok { counterReset = true return @@ -280,7 +280,7 @@ func (a *FloatHistogramAppender) Appendable(h *histogram.FloatHistogram) ( if counterResetInAnyFloatBucket(a.pBuckets, h.PositiveBuckets, a.pSpans, h.PositiveSpans) || counterResetInAnyFloatBucket(a.nBuckets, h.NegativeBuckets, a.nSpans, h.NegativeSpans) { - counterReset, positiveInterjections, negativeInterjections = true, nil, nil + counterReset, positiveInserts, negativeInserts = true, nil, nil return } @@ -290,10 +290,11 @@ func (a *FloatHistogramAppender) Appendable(h *histogram.FloatHistogram) ( // AppendableGauge returns whether the chunk can be appended to, and if so // whether: -// 1. Any recoding needs to happen to the chunk using the provided interjections +// 1. Any recoding needs to happen to the chunk using the provided inserts // (in case of any new buckets, positive or negative range, respectively). -// 2. Any recoding needs to happen for the histogram being appended, using the backward interjections -// (in case of any missing buckets, positive or negative range, respectively). +// 2. Any recoding needs to happen for the histogram being appended, using the +// backward inserts (in case of any missing buckets, positive or negative +// range, respectively). // // This method must be only used for gauge histograms. // @@ -302,8 +303,8 @@ func (a *FloatHistogramAppender) Appendable(h *histogram.FloatHistogram) ( // - The threshold for the zero bucket has changed. // - The last sample in the chunk was stale while the current sample is not stale. func (a *FloatHistogramAppender) AppendableGauge(h *histogram.FloatHistogram) ( - positiveInterjections, negativeInterjections []Interjection, - backwardPositiveInterjections, backwardNegativeInterjections []Interjection, + positiveInserts, negativeInserts []Insert, + backwardPositiveInserts, backwardNegativeInserts []Insert, positiveSpans, negativeSpans []histogram.Span, okToAppend bool, ) { @@ -325,8 +326,8 @@ func (a *FloatHistogramAppender) AppendableGauge(h *histogram.FloatHistogram) ( return } - positiveInterjections, backwardPositiveInterjections, positiveSpans = bidirectionalCompareSpans(a.pSpans, h.PositiveSpans) - negativeInterjections, backwardNegativeInterjections, negativeSpans = bidirectionalCompareSpans(a.nSpans, h.NegativeSpans) + positiveInserts, backwardPositiveInserts, positiveSpans = expandSpansBothWays(a.pSpans, h.PositiveSpans) + negativeInserts, backwardNegativeInserts, negativeSpans = expandSpansBothWays(a.nSpans, h.NegativeSpans) okToAppend = true return } @@ -501,12 +502,12 @@ func (a *FloatHistogramAppender) writeXorValue(old *xorValue, v float64) { } // Recode converts the current chunk to accommodate an expansion of the set of -// (positive and/or negative) buckets used, according to the provided -// interjections, resulting in the honoring of the provided new positive and -// negative spans. To continue appending, use the returned Appender rather than -// the receiver of this method. +// (positive and/or negative) buckets used, according to the provided inserts, +// resulting in the honoring of the provided new positive and negative spans. To +// continue appending, use the returned Appender rather than the receiver of +// this method. func (a *FloatHistogramAppender) Recode( - positiveInterjections, negativeInterjections []Interjection, + positiveInserts, negativeInserts []Insert, positiveSpans, negativeSpans []histogram.Span, ) (Chunk, Appender) { // TODO(beorn7): This currently just decodes everything and then encodes @@ -539,11 +540,11 @@ func (a *FloatHistogramAppender) Recode( // Save the modified histogram to the new chunk. hOld.PositiveSpans, hOld.NegativeSpans = positiveSpans, negativeSpans - if len(positiveInterjections) > 0 { - hOld.PositiveBuckets = interject(hOld.PositiveBuckets, positiveBuckets, positiveInterjections, false) + if len(positiveInserts) > 0 { + hOld.PositiveBuckets = insert(hOld.PositiveBuckets, positiveBuckets, positiveInserts, false) } - if len(negativeInterjections) > 0 { - hOld.NegativeBuckets = interject(hOld.NegativeBuckets, negativeBuckets, negativeInterjections, false) + if len(negativeInserts) > 0 { + hOld.NegativeBuckets = insert(hOld.NegativeBuckets, negativeBuckets, negativeInserts, false) } app.AppendFloatHistogram(tOld, hOld) } @@ -556,15 +557,15 @@ func (a *FloatHistogramAppender) Recode( // (positive and/or negative) buckets used. func (a *FloatHistogramAppender) RecodeHistogramm( fh *histogram.FloatHistogram, - pBackwardInter, nBackwardInter []Interjection, + pBackwardInter, nBackwardInter []Insert, ) { if len(pBackwardInter) > 0 { numPositiveBuckets := countSpans(fh.PositiveSpans) - fh.PositiveBuckets = interject(fh.PositiveBuckets, make([]float64, numPositiveBuckets), pBackwardInter, false) + fh.PositiveBuckets = insert(fh.PositiveBuckets, make([]float64, numPositiveBuckets), pBackwardInter, false) } if len(nBackwardInter) > 0 { numNegativeBuckets := countSpans(fh.NegativeSpans) - fh.NegativeBuckets = interject(fh.NegativeBuckets, make([]float64, numNegativeBuckets), nBackwardInter, false) + fh.NegativeBuckets = insert(fh.NegativeBuckets, make([]float64, numNegativeBuckets), nBackwardInter, false) } } diff --git a/tsdb/chunkenc/float_histogram_test.go b/tsdb/chunkenc/float_histogram_test.go index b080fe676c..e945ce1e67 100644 --- a/tsdb/chunkenc/float_histogram_test.go +++ b/tsdb/chunkenc/float_histogram_test.go @@ -138,7 +138,7 @@ func TestFloatHistogramChunkSameBuckets(t *testing.T) { require.Equal(t, ValNone, it4.Seek(exp[len(exp)-1].t+1)) } -// Mimics the scenario described for compareSpans(). +// Mimics the scenario described for expandSpansForward. func TestFloatHistogramChunkBucketChanges(t *testing.T) { c := Chunk(NewFloatHistogramChunk()) diff --git a/tsdb/chunkenc/histogram.go b/tsdb/chunkenc/histogram.go index 6ceb866a52..dd50a01e8a 100644 --- a/tsdb/chunkenc/histogram.go +++ b/tsdb/chunkenc/histogram.go @@ -243,10 +243,10 @@ func (a *HistogramAppender) AppendFloatHistogram(int64, *histogram.FloatHistogra panic("appended a float histogram to a histogram chunk") } -// Appendable returns whether the chunk can be appended to, and if so -// whether any recoding needs to happen using the provided interjections -// (in case of any new buckets, positive or negative range, respectively). -// If the sample is a gauge histogram, AppendableGauge must be used instead. +// Appendable returns whether the chunk can be appended to, and if so whether +// any recoding needs to happen using the provided inserts (in case of any new +// buckets, positive or negative range, respectively). If the sample is a gauge +// histogram, AppendableGauge must be used instead. // // The chunk is not appendable in the following cases: // @@ -261,7 +261,7 @@ func (a *HistogramAppender) AppendFloatHistogram(int64, *histogram.FloatHistogra // because of a counter reset. If the given sample is stale, it is always ok to // append. If counterReset is true, okToAppend is always false. func (a *HistogramAppender) Appendable(h *histogram.Histogram) ( - positiveInterjections, negativeInterjections []Interjection, + positiveInserts, negativeInserts []Insert, okToAppend, counterReset bool, ) { if a.NumSamples() > 0 && a.GetCounterResetHeader() == GaugeType { @@ -295,12 +295,12 @@ func (a *HistogramAppender) Appendable(h *histogram.Histogram) ( } var ok bool - positiveInterjections, ok = forwardCompareSpans(a.pSpans, h.PositiveSpans) + positiveInserts, ok = expandSpansForward(a.pSpans, h.PositiveSpans) if !ok { counterReset = true return } - negativeInterjections, ok = forwardCompareSpans(a.nSpans, h.NegativeSpans) + negativeInserts, ok = expandSpansForward(a.nSpans, h.NegativeSpans) if !ok { counterReset = true return @@ -308,7 +308,7 @@ func (a *HistogramAppender) Appendable(h *histogram.Histogram) ( if counterResetInAnyBucket(a.pBuckets, h.PositiveBuckets, a.pSpans, h.PositiveSpans) || counterResetInAnyBucket(a.nBuckets, h.NegativeBuckets, a.nSpans, h.NegativeSpans) { - counterReset, positiveInterjections, negativeInterjections = true, nil, nil + counterReset, positiveInserts, negativeInserts = true, nil, nil return } @@ -318,10 +318,11 @@ func (a *HistogramAppender) Appendable(h *histogram.Histogram) ( // AppendableGauge returns whether the chunk can be appended to, and if so // whether: -// 1. Any recoding needs to happen to the chunk using the provided interjections +// 1. Any recoding needs to happen to the chunk using the provided inserts // (in case of any new buckets, positive or negative range, respectively). -// 2. Any recoding needs to happen for the histogram being appended, using the backward interjections -// (in case of any missing buckets, positive or negative range, respectively). +// 2. Any recoding needs to happen for the histogram being appended, using the +// backward inserts (in case of any missing buckets, positive or negative +// range, respectively). // // This method must be only used for gauge histograms. // @@ -330,8 +331,8 @@ func (a *HistogramAppender) Appendable(h *histogram.Histogram) ( // - The threshold for the zero bucket has changed. // - The last sample in the chunk was stale while the current sample is not stale. func (a *HistogramAppender) AppendableGauge(h *histogram.Histogram) ( - positiveInterjections, negativeInterjections []Interjection, - backwardPositiveInterjections, backwardNegativeInterjections []Interjection, + positiveInserts, negativeInserts []Insert, + backwardPositiveInserts, backwardNegativeInserts []Insert, positiveSpans, negativeSpans []histogram.Span, okToAppend bool, ) { @@ -353,8 +354,8 @@ func (a *HistogramAppender) AppendableGauge(h *histogram.Histogram) ( return } - positiveInterjections, backwardPositiveInterjections, positiveSpans = bidirectionalCompareSpans(a.pSpans, h.PositiveSpans) - negativeInterjections, backwardNegativeInterjections, negativeSpans = bidirectionalCompareSpans(a.nSpans, h.NegativeSpans) + positiveInserts, backwardPositiveInserts, positiveSpans = expandSpansBothWays(a.pSpans, h.PositiveSpans) + negativeInserts, backwardNegativeInserts, negativeSpans = expandSpansBothWays(a.nSpans, h.NegativeSpans) okToAppend = true return } @@ -488,8 +489,9 @@ func (a *HistogramAppender) AppendHistogram(t int64, h *histogram.Histogram) { putVarbitInt(a.b, b) } } else { - // The case for the 2nd sample with single deltas is implicitly handled correctly with the double delta code, - // so we don't need a separate single delta logic for the 2nd sample. + // The case for the 2nd sample with single deltas is implicitly + // handled correctly with the double delta code, so we don't + // need a separate single delta logic for the 2nd sample. tDelta = t - a.t cntDelta = int64(h.Count) - int64(a.cnt) @@ -539,12 +541,12 @@ func (a *HistogramAppender) AppendHistogram(t int64, h *histogram.Histogram) { } // Recode converts the current chunk to accommodate an expansion of the set of -// (positive and/or negative) buckets used, according to the provided -// interjections, resulting in the honoring of the provided new positive and -// negative spans. To continue appending, use the returned Appender rather than -// the receiver of this method. +// (positive and/or negative) buckets used, according to the provided inserts, +// resulting in the honoring of the provided new positive and negative spans. To +// continue appending, use the returned Appender rather than the receiver of +// this method. func (a *HistogramAppender) Recode( - positiveInterjections, negativeInterjections []Interjection, + positiveInserts, negativeInserts []Insert, positiveSpans, negativeSpans []histogram.Span, ) (Chunk, Appender) { // TODO(beorn7): This currently just decodes everything and then encodes @@ -577,11 +579,11 @@ func (a *HistogramAppender) Recode( // Save the modified histogram to the new chunk. hOld.PositiveSpans, hOld.NegativeSpans = positiveSpans, negativeSpans - if len(positiveInterjections) > 0 { - hOld.PositiveBuckets = interject(hOld.PositiveBuckets, positiveBuckets, positiveInterjections, true) + if len(positiveInserts) > 0 { + hOld.PositiveBuckets = insert(hOld.PositiveBuckets, positiveBuckets, positiveInserts, true) } - if len(negativeInterjections) > 0 { - hOld.NegativeBuckets = interject(hOld.NegativeBuckets, negativeBuckets, negativeInterjections, true) + if len(negativeInserts) > 0 { + hOld.NegativeBuckets = insert(hOld.NegativeBuckets, negativeBuckets, negativeInserts, true) } app.AppendHistogram(tOld, hOld) } @@ -590,19 +592,19 @@ func (a *HistogramAppender) Recode( return hc, app } -// RecodeHistogram converts the current histogram (in-place) to accommodate an expansion of the set of -// (positive and/or negative) buckets used. +// RecodeHistogram converts the current histogram (in-place) to accommodate an +// expansion of the set of (positive and/or negative) buckets used. func (a *HistogramAppender) RecodeHistogram( h *histogram.Histogram, - pBackwardInter, nBackwardInter []Interjection, + pBackwardInserts, nBackwardInserts []Insert, ) { - if len(pBackwardInter) > 0 { + if len(pBackwardInserts) > 0 { numPositiveBuckets := countSpans(h.PositiveSpans) - h.PositiveBuckets = interject(h.PositiveBuckets, make([]int64, numPositiveBuckets), pBackwardInter, true) + h.PositiveBuckets = insert(h.PositiveBuckets, make([]int64, numPositiveBuckets), pBackwardInserts, true) } - if len(nBackwardInter) > 0 { + if len(nBackwardInserts) > 0 { numNegativeBuckets := countSpans(h.NegativeSpans) - h.NegativeBuckets = interject(h.NegativeBuckets, make([]int64, numNegativeBuckets), nBackwardInter, true) + h.NegativeBuckets = insert(h.NegativeBuckets, make([]int64, numNegativeBuckets), nBackwardInserts, true) } } diff --git a/tsdb/chunkenc/histogram_meta.go b/tsdb/chunkenc/histogram_meta.go index 345b8cc519..ca60f350bc 100644 --- a/tsdb/chunkenc/histogram_meta.go +++ b/tsdb/chunkenc/histogram_meta.go @@ -19,7 +19,10 @@ import ( "github.com/prometheus/prometheus/model/histogram" ) -func writeHistogramChunkLayout(b *bstream, schema int32, zeroThreshold float64, positiveSpans, negativeSpans []histogram.Span) { +func writeHistogramChunkLayout( + b *bstream, schema int32, zeroThreshold float64, + positiveSpans, negativeSpans []histogram.Span, +) { putZeroThreshold(b, zeroThreshold) putVarbitInt(b, int64(schema)) putHistogramChunkLayoutSpans(b, positiveSpans) @@ -91,9 +94,7 @@ func readHistogramChunkLayoutSpans(b *bstreamReader) ([]histogram.Span, error) { // putZeroThreshold writes the zero threshold to the bstream. It stores typical // values in just one byte, but needs 9 bytes for other values. In detail: -// -// * If the threshold is 0, store a single zero byte. -// +// - If the threshold is 0, store a single zero byte. // - If the threshold is a power of 2 between (and including) 2^-243 and 2^10, // take the exponent from the IEEE 754 representation of the threshold, which // covers a range between (and including) -242 and 11. (2^-243 is 0.5*2^-242 @@ -103,7 +104,6 @@ func readHistogramChunkLayoutSpans(b *bstreamReader) ([]histogram.Span, error) { // threshold. The default value for the zero threshold is 2^-128 (or // 0.5*2^-127 in IEEE 754 representation) and will therefore be encoded as a // single byte (with value 116). -// // - In all other cases, store 255 as a single byte, followed by the 8 bytes of // the threshold as a float64, i.e. taking 9 bytes in total. func putZeroThreshold(b *bstream, threshold float64) { @@ -186,16 +186,16 @@ func (b *bucketIterator) Next() (int, bool) { return 0, false } -// An Interjection describes how many new buckets have to be introduced before -// processing the pos'th delta from the original slice. -type Interjection struct { +// An Insert describes how many new buckets have to be inserted before +// processing the pos'th bucket from the original slice. +type Insert struct { pos int num int } -// forwardCompareSpans returns the interjections to convert a slice of deltas to a new -// slice representing an expanded set of buckets, or false if incompatible -// (e.g. if buckets were removed). +// expandSpansForward returns the inserts to expand the bucket spans 'a' so that +// they match the spans in 'b'. 'b' must cover the same or more buckets than +// 'a', otherwise the function will return false. // // Example: // @@ -222,25 +222,25 @@ type Interjection struct { // deltas 6 -3 -3 3 -3 0 2 2 1 -5 1 // delta mods: / \ / \ / \ // -// Note that whenever any new buckets are introduced, the subsequent "old" -// bucket needs to readjust its delta to the new base of 0. Thus, for the caller -// who wants to transform the set of original deltas to a new set of deltas to -// match a new span layout that adds buckets, we simply need to generate a list -// of interjections. +// Note for histograms with delta-encoded buckets: Whenever any new buckets are +// introduced, the subsequent "old" bucket needs to readjust its delta to the +// new base of 0. Thus, for the caller who wants to transform the set of +// original deltas to a new set of deltas to match a new span layout that adds +// buckets, we simply need to generate a list of inserts. // -// Note: Within forwardCompareSpans we don't have to worry about the changes to the +// Note: Within expandSpansForward we don't have to worry about the changes to the // spans themselves, thanks to the iterators we get to work with the more useful // bucket indices (which of course directly correspond to the buckets we have to // adjust). -func forwardCompareSpans(a, b []histogram.Span) (forward []Interjection, ok bool) { +func expandSpansForward(a, b []histogram.Span) (forward []Insert, ok bool) { ai := newBucketIterator(a) bi := newBucketIterator(b) - var interjections []Interjection + var inserts []Insert - // When inter.num becomes > 0, this becomes a valid interjection that - // should be yielded when we finish a streak of new buckets. - var inter Interjection + // When inter.num becomes > 0, this becomes a valid insert that should + // be yielded when we finish a streak of new buckets. + var inter Insert av, aOK := ai.Next() bv, bOK := bi.Next() @@ -250,43 +250,46 @@ loop: case aOK && bOK: switch { case av == bv: // Both have an identical value. move on! - // Finish WIP interjection and reset. + // Finish WIP insert and reset. if inter.num > 0 { - interjections = append(interjections, inter) + inserts = append(inserts, inter) } inter.num = 0 av, aOK = ai.Next() bv, bOK = bi.Next() inter.pos++ case av < bv: // b misses a value that is in a. - return interjections, false + return inserts, false case av > bv: // a misses a value that is in b. Forward b and recompare. inter.num++ bv, bOK = bi.Next() } case aOK && !bOK: // b misses a value that is in a. - return interjections, false + return inserts, false case !aOK && bOK: // a misses a value that is in b. Forward b and recompare. inter.num++ bv, bOK = bi.Next() default: // Both iterators ran out. We're done. if inter.num > 0 { - interjections = append(interjections, inter) + inserts = append(inserts, inter) } break loop } } - return interjections, true + return inserts, true } -// bidirectionalCompareSpans does everything that forwardCompareSpans does and -// also returns interjections in the other direction (i.e. buckets missing in b that are missing in a). -func bidirectionalCompareSpans(a, b []histogram.Span) (forward, backward []Interjection, mergedSpans []histogram.Span) { +// expandSpansBothWays is similar to expandSpansForward, but now b may also +// cover an entirely different set of buckets. The function returns the +// “forward” inserts to expand 'a' to also cover all the buckets exclusively +// covered by 'b', and it returns the “backward” inserts to expand 'b' to also +// cover all the buckets exclusively covered by 'a' +func expandSpansBothWays(a, b []histogram.Span) (forward, backward []Insert, mergedSpans []histogram.Span) { ai := newBucketIterator(a) bi := newBucketIterator(b) - var interjections, bInterjections []Interjection + var fInserts, bInserts []Insert var lastBucket int addBucket := func(b int) { offset := b - lastBucket - 1 @@ -305,9 +308,10 @@ func bidirectionalCompareSpans(a, b []histogram.Span) (forward, backward []Inter lastBucket = b } - // When inter.num becomes > 0, this becomes a valid interjection that - // should be yielded when we finish a streak of new buckets. - var inter, bInter Interjection + // When fInter.num (or bInter.num, respectively) becomes > 0, this + // becomes a valid insert that should be yielded when we finish a streak + // of new buckets. + var fInter, bInter Insert av, aOK := ai.Next() bv, bOK := bi.Next() @@ -317,37 +321,37 @@ loop: case aOK && bOK: switch { case av == bv: // Both have an identical value. move on! - // Finish WIP interjection and reset. - if inter.num > 0 { - interjections = append(interjections, inter) - inter.num = 0 + // Finish WIP insert and reset. + if fInter.num > 0 { + fInserts = append(fInserts, fInter) + fInter.num = 0 } if bInter.num > 0 { - bInterjections = append(bInterjections, bInter) + bInserts = append(bInserts, bInter) bInter.num = 0 } addBucket(av) av, aOK = ai.Next() bv, bOK = bi.Next() - inter.pos++ + fInter.pos++ bInter.pos++ case av < bv: // b misses a value that is in a. bInter.num++ - // Collect the forward interjection before advancing the - // position of 'a'. - if inter.num > 0 { - interjections = append(interjections, inter) - inter.num = 0 + // Collect the forward inserts before advancing + // the position of 'a'. + if fInter.num > 0 { + fInserts = append(fInserts, fInter) + fInter.num = 0 } addBucket(av) - inter.pos++ + fInter.pos++ av, aOK = ai.Next() case av > bv: // a misses a value that is in b. Forward b and recompare. - inter.num++ - // Collect the backward interjection before advancing the + fInter.num++ + // Collect the backward inserts before advancing the // position of 'b'. if bInter.num > 0 { - bInterjections = append(bInterjections, bInter) + bInserts = append(bInserts, bInter) bInter.num = 0 } addBucket(bv) @@ -359,92 +363,92 @@ loop: addBucket(av) av, aOK = ai.Next() case !aOK && bOK: // a misses a value that is in b. Forward b and recompare. - inter.num++ + fInter.num++ addBucket(bv) bv, bOK = bi.Next() default: // Both iterators ran out. We're done. - if inter.num > 0 { - interjections = append(interjections, inter) + if fInter.num > 0 { + fInserts = append(fInserts, fInter) } if bInter.num > 0 { - bInterjections = append(bInterjections, bInter) + bInserts = append(bInserts, bInter) } break loop } } - return interjections, bInterjections, mergedSpans + return fInserts, bInserts, mergedSpans } type bucketValue interface { int64 | float64 } -// interject merges 'in' with the provided interjections and writes them into -// 'out', which must already have the appropriate length. -func interject[BV bucketValue](in, out []BV, interjections []Interjection, deltas bool) []BV { +// insert merges 'in' with the provided inserts and writes them into 'out', +// which must already have the appropriate length. 'out' is also returned for +// convenience. +func insert[BV bucketValue](in, out []BV, inserts []Insert, deltas bool) []BV { var ( - j int // Position in out. - v BV // The last value seen. - interj int // The next interjection to process. + oi int // Position in out. + v BV // The last value seen. + ii int // The next insert to process. ) for i, d := range in { - if interj < len(interjections) && i == interjections[interj].pos { - - // We have an interjection! - // Add interjection.num new delta values such that their bucket values equate 0. - // When deltas==false, it means that it is an absolute value. So we set it to 0 directly. + if ii < len(inserts) && i == inserts[ii].pos { + // We have an insert! + // Add insert.num new delta values such that their + // bucket values equate 0. When deltas==false, it means + // that it is an absolute value. So we set it to 0 + // directly. if deltas { - out[j] = -v + out[oi] = -v } else { - out[j] = 0 + out[oi] = 0 } - j++ - for x := 1; x < interjections[interj].num; x++ { - out[j] = 0 - j++ + oi++ + for x := 1; x < inserts[ii].num; x++ { + out[oi] = 0 + oi++ } - interj++ + ii++ // Now save the value from the input. The delta value we // should save is the original delta value + the last - // value of the point before the interjection (to undo - // the delta that was introduced by the interjection). - // When deltas==false, it means that it is an absolute value, + // value of the point before the insert (to undo the + // delta that was introduced by the insert). When + // deltas==false, it means that it is an absolute value, // so we set it directly to the value in the 'in' slice. if deltas { - out[j] = d + v + out[oi] = d + v } else { - out[j] = d + out[oi] = d } - j++ + oi++ v = d + v continue } - - // If there was no interjection, the original delta is still - // valid. - out[j] = d - j++ + // If there was no insert, the original delta is still valid. + out[oi] = d + oi++ v += d } - switch interj { - case len(interjections): - // All interjections processed. Nothing more to do. - case len(interjections) - 1: - // One more interjection to process at the end. + switch ii { + case len(inserts): + // All inserts processed. Nothing more to do. + case len(inserts) - 1: + // One more insert to process at the end. if deltas { - out[j] = -v + out[oi] = -v } else { - out[j] = 0 + out[oi] = 0 } - j++ - for x := 1; x < interjections[interj].num; x++ { - out[j] = 0 - j++ + oi++ + for x := 1; x < inserts[ii].num; x++ { + out[oi] = 0 + oi++ } default: - panic("unprocessed interjections left") + panic("unprocessed inserts left") } return out } diff --git a/tsdb/chunkenc/histogram_meta_test.go b/tsdb/chunkenc/histogram_meta_test.go index a4ce62f3b7..0b2b187475 100644 --- a/tsdb/chunkenc/histogram_meta_test.go +++ b/tsdb/chunkenc/histogram_meta_test.go @@ -78,7 +78,7 @@ func TestBucketIterator(t *testing.T) { }, idxs: []int{100, 101, 102, 103, 112, 113, 114, 115, 116, 117, 118, 119}, }, - // The below 2 sets ore the ones described in compareSpans's comments. + // The below 2 sets ore the ones described in expandSpansForward's comments. { spans: []histogram.Span{ {Offset: 0, Length: 2}, @@ -111,12 +111,12 @@ func TestBucketIterator(t *testing.T) { } } -func TestCompareSpansAndInterject(t *testing.T) { +func TestCompareSpansAndInsert(t *testing.T) { scenarios := []struct { - description string - spansA, spansB []histogram.Span - interjections, backwardInterjections []Interjection - bucketsIn, bucketsOut []int64 + description string + spansA, spansB []histogram.Span + fInserts, bInserts []Insert + bucketsIn, bucketsOut []int64 }{ { description: "single prepend at the beginning", @@ -126,7 +126,7 @@ func TestCompareSpansAndInterject(t *testing.T) { spansB: []histogram.Span{ {Offset: -11, Length: 4}, }, - interjections: []Interjection{ + fInserts: []Insert{ { pos: 0, num: 1, @@ -143,7 +143,7 @@ func TestCompareSpansAndInterject(t *testing.T) { spansB: []histogram.Span{ {Offset: -10, Length: 4}, }, - interjections: []Interjection{ + fInserts: []Insert{ { pos: 3, num: 1, @@ -160,7 +160,7 @@ func TestCompareSpansAndInterject(t *testing.T) { spansB: []histogram.Span{ {Offset: -12, Length: 5}, }, - interjections: []Interjection{ + fInserts: []Insert{ { pos: 0, num: 2, @@ -177,7 +177,7 @@ func TestCompareSpansAndInterject(t *testing.T) { spansB: []histogram.Span{ {Offset: -10, Length: 5}, }, - interjections: []Interjection{ + fInserts: []Insert{ { pos: 3, num: 2, @@ -194,7 +194,7 @@ func TestCompareSpansAndInterject(t *testing.T) { spansB: []histogram.Span{ {Offset: -12, Length: 7}, }, - interjections: []Interjection{ + fInserts: []Insert{ { pos: 0, num: 2, @@ -215,7 +215,7 @@ func TestCompareSpansAndInterject(t *testing.T) { spansB: []histogram.Span{ {Offset: -9, Length: 3}, }, - backwardInterjections: []Interjection{ + bInserts: []Insert{ {pos: 0, num: 1}, }, }, @@ -228,7 +228,7 @@ func TestCompareSpansAndInterject(t *testing.T) { {Offset: -10, Length: 2}, {Offset: 1, Length: 1}, }, - backwardInterjections: []Interjection{ + bInserts: []Insert{ {pos: 2, num: 1}, }, }, @@ -240,7 +240,7 @@ func TestCompareSpansAndInterject(t *testing.T) { spansB: []histogram.Span{ {Offset: -10, Length: 3}, }, - backwardInterjections: []Interjection{ + bInserts: []Insert{ {pos: 3, num: 1}, }, }, @@ -259,7 +259,7 @@ func TestCompareSpansAndInterject(t *testing.T) { {Offset: 1, Length: 4}, {Offset: 3, Length: 3}, }, - interjections: []Interjection{ + fInserts: []Insert{ { pos: 2, num: 1, @@ -277,7 +277,7 @@ func TestCompareSpansAndInterject(t *testing.T) { bucketsOut: []int64{6, -3, -3, 3, -3, 0, 2, 2, 1, -5, 1}, }, { - description: "both forward and backward interjections, complex case", + description: "both forward and backward inserts, complex case", spansA: []histogram.Span{ {Offset: 0, Length: 2}, {Offset: 2, Length: 1}, @@ -292,7 +292,7 @@ func TestCompareSpansAndInterject(t *testing.T) { {Offset: 1, Length: 1}, {Offset: 4, Length: 1}, }, - interjections: []Interjection{ + fInserts: []Insert{ { pos: 2, num: 1, @@ -306,7 +306,7 @@ func TestCompareSpansAndInterject(t *testing.T) { num: 1, }, }, - backwardInterjections: []Interjection{ + bInserts: []Insert{ { pos: 0, num: 1, @@ -329,22 +329,22 @@ func TestCompareSpansAndInterject(t *testing.T) { for _, s := range scenarios { t.Run(s.description, func(t *testing.T) { - if len(s.backwardInterjections) > 0 { - interjections, bInterjections, _ := bidirectionalCompareSpans(s.spansA, s.spansB) - require.Equal(t, s.interjections, interjections) - require.Equal(t, s.backwardInterjections, bInterjections) + if len(s.bInserts) > 0 { + fInserts, bInserts, _ := expandSpansBothWays(s.spansA, s.spansB) + require.Equal(t, s.fInserts, fInserts) + require.Equal(t, s.bInserts, bInserts) } - interjections, valid := forwardCompareSpans(s.spansA, s.spansB) - if len(s.backwardInterjections) > 0 { + inserts, valid := expandSpansForward(s.spansA, s.spansB) + if len(s.bInserts) > 0 { require.False(t, valid, "compareScan unexpectedly returned true") return } require.True(t, valid, "compareScan unexpectedly returned false") - require.Equal(t, s.interjections, interjections) + require.Equal(t, s.fInserts, inserts) gotBuckets := make([]int64, len(s.bucketsOut)) - interject(s.bucketsIn, gotBuckets, interjections, true) + insert(s.bucketsIn, gotBuckets, inserts, true) require.Equal(t, s.bucketsOut, gotBuckets) floatBucketsIn := make([]float64, len(s.bucketsIn)) @@ -362,7 +362,7 @@ func TestCompareSpansAndInterject(t *testing.T) { floatBucketsOut[i] = float64(last) } gotFloatBuckets := make([]float64, len(floatBucketsOut)) - interject(floatBucketsIn, gotFloatBuckets, interjections, false) + insert(floatBucketsIn, gotFloatBuckets, inserts, false) require.Equal(t, floatBucketsOut, gotFloatBuckets) }) } @@ -564,12 +564,12 @@ func TestSpansFromBidirectionalCompareSpans(t *testing.T) { copy(s1c, c.s1) copy(s2c, c.s2) - _, _, act := bidirectionalCompareSpans(c.s1, c.s2) + _, _, act := expandSpansBothWays(c.s1, c.s2) require.Equal(t, c.exp, act) // Check that s1 and s2 are not modified. require.Equal(t, s1c, c.s1) require.Equal(t, s2c, c.s2) - _, _, act = bidirectionalCompareSpans(c.s2, c.s1) + _, _, act = expandSpansBothWays(c.s2, c.s1) require.Equal(t, c.exp, act) } } diff --git a/tsdb/chunkenc/histogram_test.go b/tsdb/chunkenc/histogram_test.go index 73851c9dfb..c5b6faa2a0 100644 --- a/tsdb/chunkenc/histogram_test.go +++ b/tsdb/chunkenc/histogram_test.go @@ -149,7 +149,7 @@ func TestHistogramChunkSameBuckets(t *testing.T) { require.Equal(t, ValNone, it4.Seek(exp[len(exp)-1].t+1)) } -// Mimics the scenario described for compareSpans(). +// Mimics the scenario described for expandSpansForward. func TestHistogramChunkBucketChanges(t *testing.T) { c := Chunk(NewHistogramChunk()) diff --git a/tsdb/head_append.go b/tsdb/head_append.go index d4534c6695..f3585a87be 100644 --- a/tsdb/head_append.go +++ b/tsdb/head_append.go @@ -1141,7 +1141,6 @@ func (s *memSeries) append(t int64, v float64, appendID uint64, chunkDiskMapper // appendHistogram adds the histogram. // It is unsafe to call this concurrently with s.iterator(...) without holding the series lock. -// TODO(codesome): Support gauge histograms here. func (s *memSeries) appendHistogram(t int64, h *histogram.Histogram, appendID uint64, chunkDiskMapper *chunks.ChunkDiskMapper, chunkRange int64) (sampleInOrder, chunkCreated bool) { // Head controls the execution of recoding, so that we own the proper // chunk reference afterwards. We check for Appendable from appender before @@ -1150,10 +1149,10 @@ func (s *memSeries) appendHistogram(t int64, h *histogram.Histogram, appendID ui // meta properly. app, _ := s.app.(*chunkenc.HistogramAppender) var ( - positiveInterjections, negativeInterjections []chunkenc.Interjection - pBackwardInter, nBackwardInter []chunkenc.Interjection - pMergedSpans, nMergedSpans []histogram.Span - okToAppend, counterReset bool + pForwardInserts, nForwardInserts []chunkenc.Insert + pBackwardInserts, nBackwardInserts []chunkenc.Insert + pMergedSpans, nMergedSpans []histogram.Span + okToAppend, counterReset bool ) c, sampleInOrder, chunkCreated := s.appendPreprocessor(t, chunkenc.EncHistogram, chunkDiskMapper, chunkRange) if !sampleInOrder { @@ -1162,32 +1161,32 @@ func (s *memSeries) appendHistogram(t int64, h *histogram.Histogram, appendID ui gauge := h.CounterResetHint == histogram.GaugeType if app != nil { if gauge { - positiveInterjections, negativeInterjections, pBackwardInter, nBackwardInter, pMergedSpans, nMergedSpans, okToAppend = app.AppendableGauge(h) + pForwardInserts, nForwardInserts, pBackwardInserts, nBackwardInserts, pMergedSpans, nMergedSpans, okToAppend = app.AppendableGauge(h) } else { - positiveInterjections, negativeInterjections, okToAppend, counterReset = app.Appendable(h) + pForwardInserts, nForwardInserts, okToAppend, counterReset = app.Appendable(h) } } if !chunkCreated { - if len(pBackwardInter)+len(nBackwardInter) > 0 { + if len(pBackwardInserts)+len(nBackwardInserts) > 0 { h.PositiveSpans = pMergedSpans h.NegativeSpans = nMergedSpans - app.RecodeHistogram(h, pBackwardInter, nBackwardInter) + app.RecodeHistogram(h, pBackwardInserts, nBackwardInserts) } // We have 3 cases here // - !okToAppend -> We need to cut a new chunk. - // - okToAppend but we have interjections → Existing chunk needs + // - okToAppend but we have inserts → Existing chunk needs // recoding before we can append our histogram. - // - okToAppend and no interjections → Chunk is ready to support our histogram. + // - okToAppend and no inserts → Chunk is ready to support our histogram. if !okToAppend || counterReset { c = s.cutNewHeadChunk(t, chunkenc.EncHistogram, chunkDiskMapper, chunkRange) chunkCreated = true - } else if len(positiveInterjections) > 0 || len(negativeInterjections) > 0 { + } else if len(pForwardInserts) > 0 || len(nForwardInserts) > 0 { // New buckets have appeared. We need to recode all // prior histogram samples within the chunk before we // can process this one. chunk, app := app.Recode( - positiveInterjections, negativeInterjections, + pForwardInserts, nForwardInserts, h.PositiveSpans, h.NegativeSpans, ) c.chunk = chunk @@ -1233,10 +1232,10 @@ func (s *memSeries) appendFloatHistogram(t int64, fh *histogram.FloatHistogram, // meta properly. app, _ := s.app.(*chunkenc.FloatHistogramAppender) var ( - positiveInterjections, negativeInterjections []chunkenc.Interjection - pBackwardInter, nBackwardInter []chunkenc.Interjection - pMergedSpans, nMergedSpans []histogram.Span - okToAppend, counterReset bool + pForwardInserts, nForwardInserts []chunkenc.Insert + pBackwardInserts, nBackwardInserts []chunkenc.Insert + pMergedSpans, nMergedSpans []histogram.Span + okToAppend, counterReset bool ) c, sampleInOrder, chunkCreated := s.appendPreprocessor(t, chunkenc.EncFloatHistogram, chunkDiskMapper, chunkRange) if !sampleInOrder { @@ -1245,33 +1244,33 @@ func (s *memSeries) appendFloatHistogram(t int64, fh *histogram.FloatHistogram, gauge := fh.CounterResetHint == histogram.GaugeType if app != nil { if gauge { - positiveInterjections, negativeInterjections, pBackwardInter, nBackwardInter, + pForwardInserts, nForwardInserts, pBackwardInserts, nBackwardInserts, pMergedSpans, nMergedSpans, okToAppend = app.AppendableGauge(fh) } else { - positiveInterjections, negativeInterjections, okToAppend, counterReset = app.Appendable(fh) + pForwardInserts, nForwardInserts, okToAppend, counterReset = app.Appendable(fh) } } if !chunkCreated { - if len(pBackwardInter)+len(nBackwardInter) > 0 { + if len(pBackwardInserts)+len(nBackwardInserts) > 0 { fh.PositiveSpans = pMergedSpans fh.NegativeSpans = nMergedSpans - app.RecodeHistogramm(fh, pBackwardInter, nBackwardInter) + app.RecodeHistogramm(fh, pBackwardInserts, nBackwardInserts) } // We have 3 cases here // - !okToAppend -> We need to cut a new chunk. - // - okToAppend but we have interjections → Existing chunk needs + // - okToAppend but we have inserts → Existing chunk needs // recoding before we can append our histogram. - // - okToAppend and no interjections → Chunk is ready to support our histogram. + // - okToAppend and no inserts → Chunk is ready to support our histogram. if !okToAppend || counterReset { c = s.cutNewHeadChunk(t, chunkenc.EncFloatHistogram, chunkDiskMapper, chunkRange) chunkCreated = true - } else if len(positiveInterjections) > 0 || len(negativeInterjections) > 0 { + } else if len(pForwardInserts) > 0 || len(nForwardInserts) > 0 { // New buckets have appeared. We need to recode all // prior histogram samples within the chunk before we // can process this one. chunk, app := app.Recode( - positiveInterjections, negativeInterjections, + pForwardInserts, nForwardInserts, fh.PositiveSpans, fh.NegativeSpans, ) c.chunk = chunk