prometheus/model/histogram/float_histogram.go
beorn7 9aadd54786 histogram: Fix bounds of buckets returned by floatBucketIterator
The bounds weren't really used so far, so no actual bug in the code so
far. But it's obviously confusing if the bounds returned by a
floatBucketIterator with a target schema different from the original
schema are wrong.

Signed-off-by: beorn7 <beorn@grafana.com>
2023-07-19 18:19:18 +02:00

985 lines
34 KiB
Go

// Copyright 2021 The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package histogram
import (
"fmt"
"strings"
)
// FloatHistogram is similar to Histogram but uses float64 for all
// counts. Additionally, bucket counts are absolute and not deltas.
//
// A FloatHistogram is needed by PromQL to handle operations that might result
// in fractional counts. Since the counts in a histogram are unlikely to be too
// large to be represented precisely by a float64, a FloatHistogram can also be
// used to represent a histogram with integer counts and thus serves as a more
// generalized representation.
type FloatHistogram struct {
// Counter reset information.
CounterResetHint CounterResetHint
// Currently valid schema numbers are -4 <= n <= 8. They are all for
// base-2 bucket schemas, where 1 is a bucket boundary in each case, and
// then each power of two is divided into 2^n logarithmic buckets. Or
// in other words, each bucket boundary is the previous boundary times
// 2^(2^-n).
Schema int32
// Width of the zero bucket.
ZeroThreshold float64
// Observations falling into the zero bucket. Must be zero or positive.
ZeroCount float64
// Total number of observations. Must be zero or positive.
Count float64
// Sum of observations. This is also used as the stale marker.
Sum float64
// Spans for positive and negative buckets (see Span below).
PositiveSpans, NegativeSpans []Span
// Observation counts in buckets. Each represents an absolute count and
// must be zero or positive.
PositiveBuckets, NegativeBuckets []float64
}
// Copy returns a deep copy of the Histogram.
func (h *FloatHistogram) Copy() *FloatHistogram {
c := *h
if h.PositiveSpans != nil {
c.PositiveSpans = make([]Span, len(h.PositiveSpans))
copy(c.PositiveSpans, h.PositiveSpans)
}
if h.NegativeSpans != nil {
c.NegativeSpans = make([]Span, len(h.NegativeSpans))
copy(c.NegativeSpans, h.NegativeSpans)
}
if h.PositiveBuckets != nil {
c.PositiveBuckets = make([]float64, len(h.PositiveBuckets))
copy(c.PositiveBuckets, h.PositiveBuckets)
}
if h.NegativeBuckets != nil {
c.NegativeBuckets = make([]float64, len(h.NegativeBuckets))
copy(c.NegativeBuckets, h.NegativeBuckets)
}
return &c
}
// CopyToSchema works like Copy, but the returned deep copy has the provided
// target schema, which must be ≤ the original schema (i.e. it must have a lower
// resolution).
func (h *FloatHistogram) CopyToSchema(targetSchema int32) *FloatHistogram {
if targetSchema == h.Schema {
// Fast path.
return h.Copy()
}
if targetSchema > h.Schema {
panic(fmt.Errorf("cannot copy from schema %d to %d", h.Schema, targetSchema))
}
c := FloatHistogram{
Schema: targetSchema,
ZeroThreshold: h.ZeroThreshold,
ZeroCount: h.ZeroCount,
Count: h.Count,
Sum: h.Sum,
}
// TODO(beorn7): This is a straight-forward implementation using merging
// iterators for the original buckets and then adding one merged bucket
// after another to the newly created FloatHistogram. It's well possible
// that a more involved implementation performs much better, which we
// could do if this code path turns out to be performance-critical.
var iInSpan, index int32
for iSpan, iBucket, it := -1, -1, h.floatBucketIterator(true, 0, targetSchema); it.Next(); {
b := it.At()
c.PositiveSpans, c.PositiveBuckets, iSpan, iBucket, iInSpan = addBucket(
b, c.PositiveSpans, c.PositiveBuckets, iSpan, iBucket, iInSpan, index,
)
index = b.Index
}
for iSpan, iBucket, it := -1, -1, h.floatBucketIterator(false, 0, targetSchema); it.Next(); {
b := it.At()
c.NegativeSpans, c.NegativeBuckets, iSpan, iBucket, iInSpan = addBucket(
b, c.NegativeSpans, c.NegativeBuckets, iSpan, iBucket, iInSpan, index,
)
index = b.Index
}
return &c
}
// String returns a string representation of the Histogram.
func (h *FloatHistogram) String() string {
var sb strings.Builder
fmt.Fprintf(&sb, "{count:%g, sum:%g", h.Count, h.Sum)
var nBuckets []Bucket[float64]
for it := h.NegativeBucketIterator(); it.Next(); {
bucket := it.At()
if bucket.Count != 0 {
nBuckets = append(nBuckets, it.At())
}
}
for i := len(nBuckets) - 1; i >= 0; i-- {
fmt.Fprintf(&sb, ", %s", nBuckets[i].String())
}
if h.ZeroCount != 0 {
fmt.Fprintf(&sb, ", %s", h.ZeroBucket().String())
}
for it := h.PositiveBucketIterator(); it.Next(); {
bucket := it.At()
if bucket.Count != 0 {
fmt.Fprintf(&sb, ", %s", bucket.String())
}
}
sb.WriteRune('}')
return sb.String()
}
// ZeroBucket returns the zero bucket.
func (h *FloatHistogram) ZeroBucket() Bucket[float64] {
return Bucket[float64]{
Lower: -h.ZeroThreshold,
Upper: h.ZeroThreshold,
LowerInclusive: true,
UpperInclusive: true,
Count: h.ZeroCount,
}
}
// Mul multiplies the FloatHistogram by the provided factor, i.e. it scales all
// bucket counts including the zero bucket and the count and the sum of
// observations. The bucket layout stays the same. This method changes the
// receiving histogram directly (rather than acting on a copy). It returns a
// pointer to the receiving histogram for convenience.
func (h *FloatHistogram) Mul(factor float64) *FloatHistogram {
h.ZeroCount *= factor
h.Count *= factor
h.Sum *= factor
for i := range h.PositiveBuckets {
h.PositiveBuckets[i] *= factor
}
for i := range h.NegativeBuckets {
h.NegativeBuckets[i] *= factor
}
return h
}
// Div works like Scale but divides instead of multiplies.
// When dividing by 0, everything will be set to Inf.
func (h *FloatHistogram) Div(scalar float64) *FloatHistogram {
h.ZeroCount /= scalar
h.Count /= scalar
h.Sum /= scalar
for i := range h.PositiveBuckets {
h.PositiveBuckets[i] /= scalar
}
for i := range h.NegativeBuckets {
h.NegativeBuckets[i] /= scalar
}
return h
}
// Add adds the provided other histogram to the receiving histogram. Count, Sum,
// and buckets from the other histogram are added to the corresponding
// components of the receiving histogram. Buckets in the other histogram that do
// not exist in the receiving histogram are inserted into the latter. The
// resulting histogram might have buckets with a population of zero or directly
// adjacent spans (offset=0). To normalize those, call the Compact method.
//
// The method reconciles differences in the zero threshold and in the schema,
// but the schema of the other histogram must be ≥ the schema of the receiving
// histogram (i.e. must have an equal or higher resolution). This means that the
// schema of the receiving histogram won't change. Its zero threshold, however,
// will change if needed. The other histogram will not be modified in any case.
//
// This method returns a pointer to the receiving histogram for convenience.
func (h *FloatHistogram) Add(other *FloatHistogram) *FloatHistogram {
switch {
case other.CounterResetHint == h.CounterResetHint:
// Adding apples to apples, all good. No need to change anything.
case h.CounterResetHint == GaugeType:
// Adding something else to a gauge. That's probably OK. Outcome is a gauge.
// Nothing to do since the receiver is already marked as gauge.
case other.CounterResetHint == GaugeType:
// Similar to before, but this time the receiver is "something else" and we have to change it to gauge.
h.CounterResetHint = GaugeType
case h.CounterResetHint == UnknownCounterReset:
// With the receiver's CounterResetHint being "unknown", this could still be legitimate
// if the caller knows what they are doing. Outcome is then again "unknown".
// No need to do anything since the receiver's CounterResetHint is already "unknown".
case other.CounterResetHint == UnknownCounterReset:
// Similar to before, but now we have to set the receiver's CounterResetHint to "unknown".
h.CounterResetHint = UnknownCounterReset
default:
// All other cases shouldn't actually happen.
// They are a direct collision of CounterReset and NotCounterReset.
// Conservatively set the CounterResetHint to "unknown" and isse a warning.
h.CounterResetHint = UnknownCounterReset
// TODO(trevorwhitney): Actually issue the warning as soon as the plumbing for it is in place
}
otherZeroCount := h.reconcileZeroBuckets(other)
h.ZeroCount += otherZeroCount
h.Count += other.Count
h.Sum += other.Sum
// TODO(beorn7): If needed, this can be optimized by inspecting the
// spans in other and create missing buckets in h in batches.
var iInSpan, index int32
for iSpan, iBucket, it := -1, -1, other.floatBucketIterator(true, h.ZeroThreshold, h.Schema); it.Next(); {
b := it.At()
h.PositiveSpans, h.PositiveBuckets, iSpan, iBucket, iInSpan = addBucket(
b, h.PositiveSpans, h.PositiveBuckets, iSpan, iBucket, iInSpan, index,
)
index = b.Index
}
for iSpan, iBucket, it := -1, -1, other.floatBucketIterator(false, h.ZeroThreshold, h.Schema); it.Next(); {
b := it.At()
h.NegativeSpans, h.NegativeBuckets, iSpan, iBucket, iInSpan = addBucket(
b, h.NegativeSpans, h.NegativeBuckets, iSpan, iBucket, iInSpan, index,
)
index = b.Index
}
return h
}
// Sub works like Add but subtracts the other histogram.
func (h *FloatHistogram) Sub(other *FloatHistogram) *FloatHistogram {
otherZeroCount := h.reconcileZeroBuckets(other)
h.ZeroCount -= otherZeroCount
h.Count -= other.Count
h.Sum -= other.Sum
// TODO(beorn7): If needed, this can be optimized by inspecting the
// spans in other and create missing buckets in h in batches.
var iInSpan, index int32
for iSpan, iBucket, it := -1, -1, other.floatBucketIterator(true, h.ZeroThreshold, h.Schema); it.Next(); {
b := it.At()
b.Count *= -1
h.PositiveSpans, h.PositiveBuckets, iSpan, iBucket, iInSpan = addBucket(
b, h.PositiveSpans, h.PositiveBuckets, iSpan, iBucket, iInSpan, index,
)
index = b.Index
}
for iSpan, iBucket, it := -1, -1, other.floatBucketIterator(false, h.ZeroThreshold, h.Schema); it.Next(); {
b := it.At()
b.Count *= -1
h.NegativeSpans, h.NegativeBuckets, iSpan, iBucket, iInSpan = addBucket(
b, h.NegativeSpans, h.NegativeBuckets, iSpan, iBucket, iInSpan, index,
)
index = b.Index
}
return h
}
// Equals returns true if the given float histogram matches exactly.
// Exact match is when there are no new buckets (even empty) and no missing buckets,
// and all the bucket values match. Spans can have different empty length spans in between,
// but they must represent the same bucket layout to match.
func (h *FloatHistogram) Equals(h2 *FloatHistogram) bool {
if h2 == nil {
return false
}
if h.Schema != h2.Schema || h.ZeroThreshold != h2.ZeroThreshold ||
h.ZeroCount != h2.ZeroCount || h.Count != h2.Count || h.Sum != h2.Sum {
return false
}
if !spansMatch(h.PositiveSpans, h2.PositiveSpans) {
return false
}
if !spansMatch(h.NegativeSpans, h2.NegativeSpans) {
return false
}
if !bucketsMatch(h.PositiveBuckets, h2.PositiveBuckets) {
return false
}
if !bucketsMatch(h.NegativeBuckets, h2.NegativeBuckets) {
return false
}
return true
}
// addBucket takes the "coordinates" of the last bucket that was handled and
// adds the provided bucket after it. If a corresponding bucket exists, the
// count is added. If not, the bucket is inserted. The updated slices and the
// coordinates of the inserted or added-to bucket are returned.
func addBucket(
b Bucket[float64],
spans []Span, buckets []float64,
iSpan, iBucket int,
iInSpan, index int32,
) (
newSpans []Span, newBuckets []float64,
newISpan, newIBucket int, newIInSpan int32,
) {
if iSpan == -1 {
// First add, check if it is before all spans.
if len(spans) == 0 || spans[0].Offset > b.Index {
// Add bucket before all others.
buckets = append(buckets, 0)
copy(buckets[1:], buckets)
buckets[0] = b.Count
if len(spans) > 0 && spans[0].Offset == b.Index+1 {
spans[0].Length++
spans[0].Offset--
return spans, buckets, 0, 0, 0
}
spans = append(spans, Span{})
copy(spans[1:], spans)
spans[0] = Span{Offset: b.Index, Length: 1}
if len(spans) > 1 {
// Convert the absolute offset in the formerly
// first span to a relative offset.
spans[1].Offset -= b.Index + 1
}
return spans, buckets, 0, 0, 0
}
if spans[0].Offset == b.Index {
// Just add to first bucket.
buckets[0] += b.Count
return spans, buckets, 0, 0, 0
}
// We are behind the first bucket, so set everything to the
// first bucket and continue normally.
iSpan, iBucket, iInSpan = 0, 0, 0
index = spans[0].Offset
}
deltaIndex := b.Index - index
for {
remainingInSpan := int32(spans[iSpan].Length) - iInSpan
if deltaIndex < remainingInSpan {
// Bucket is in current span.
iBucket += int(deltaIndex)
iInSpan += deltaIndex
buckets[iBucket] += b.Count
return spans, buckets, iSpan, iBucket, iInSpan
}
deltaIndex -= remainingInSpan
iBucket += int(remainingInSpan)
iSpan++
if iSpan == len(spans) || deltaIndex < spans[iSpan].Offset {
// Bucket is in gap behind previous span (or there are no further spans).
buckets = append(buckets, 0)
copy(buckets[iBucket+1:], buckets[iBucket:])
buckets[iBucket] = b.Count
if deltaIndex == 0 {
// Directly after previous span, extend previous span.
if iSpan < len(spans) {
spans[iSpan].Offset--
}
iSpan--
iInSpan = int32(spans[iSpan].Length)
spans[iSpan].Length++
return spans, buckets, iSpan, iBucket, iInSpan
}
if iSpan < len(spans) && deltaIndex == spans[iSpan].Offset-1 {
// Directly before next span, extend next span.
iInSpan = 0
spans[iSpan].Offset--
spans[iSpan].Length++
return spans, buckets, iSpan, iBucket, iInSpan
}
// No next span, or next span is not directly adjacent to new bucket.
// Add new span.
iInSpan = 0
if iSpan < len(spans) {
spans[iSpan].Offset -= deltaIndex + 1
}
spans = append(spans, Span{})
copy(spans[iSpan+1:], spans[iSpan:])
spans[iSpan] = Span{Length: 1, Offset: deltaIndex}
return spans, buckets, iSpan, iBucket, iInSpan
}
// Try start of next span.
deltaIndex -= spans[iSpan].Offset
iInSpan = 0
}
}
// Compact eliminates empty buckets at the beginning and end of each span, then
// merges spans that are consecutive or at most maxEmptyBuckets apart, and
// finally splits spans that contain more consecutive empty buckets than
// maxEmptyBuckets. (The actual implementation might do something more efficient
// but with the same result.) The compaction happens "in place" in the
// receiving histogram, but a pointer to it is returned for convenience.
//
// The ideal value for maxEmptyBuckets depends on circumstances. The motivation
// to set maxEmptyBuckets > 0 is the assumption that is less overhead to
// represent very few empty buckets explicitly within one span than cutting the
// one span into two to treat the empty buckets as a gap between the two spans,
// both in terms of storage requirement as well as in terms of encoding and
// decoding effort. However, the tradeoffs are subtle. For one, they are
// different in the exposition format vs. in a TSDB chunk vs. for the in-memory
// representation as Go types. In the TSDB, as an additional aspects, the span
// layout is only stored once per chunk, while many histograms with that same
// chunk layout are then only stored with their buckets (so that even a single
// empty bucket will be stored many times).
//
// For the Go types, an additional Span takes 8 bytes. Similarly, an additional
// bucket takes 8 bytes. Therefore, with a single separating empty bucket, both
// options have the same storage requirement, but the single-span solution is
// easier to iterate through. Still, the safest bet is to use maxEmptyBuckets==0
// and only use a larger number if you know what you are doing.
func (h *FloatHistogram) Compact(maxEmptyBuckets int) *FloatHistogram {
h.PositiveBuckets, h.PositiveSpans = compactBuckets(
h.PositiveBuckets, h.PositiveSpans, maxEmptyBuckets, false,
)
h.NegativeBuckets, h.NegativeSpans = compactBuckets(
h.NegativeBuckets, h.NegativeSpans, maxEmptyBuckets, false,
)
return h
}
// DetectReset returns true if the receiving histogram is missing any buckets
// that have a non-zero population in the provided previous histogram. It also
// returns true if any count (in any bucket, in the zero count, or in the count
// of observations, but NOT the sum of observations) is smaller in the receiving
// histogram compared to the previous histogram. Otherwise, it returns false.
//
// This method will shortcut to true if a CounterReset is detected, and shortcut
// to false if NotCounterReset is detected. Otherwise it will do the work to detect
// a reset.
//
// Special behavior in case the Schema or the ZeroThreshold are not the same in
// both histograms:
//
// - A decrease of the ZeroThreshold or an increase of the Schema (i.e. an
// increase of resolution) can only happen together with a reset. Thus, the
// method returns true in either case.
//
// - Upon an increase of the ZeroThreshold, the buckets in the previous
// histogram that fall within the new ZeroThreshold are added to the ZeroCount
// of the previous histogram (without mutating the provided previous
// histogram). The scenario that a populated bucket of the previous histogram
// is partially within, partially outside of the new ZeroThreshold, can only
// happen together with a counter reset and therefore shortcuts to returning
// true.
//
// - Upon a decrease of the Schema, the buckets of the previous histogram are
// merged so that they match the new, lower-resolution schema (again without
// mutating the provided previous histogram).
func (h *FloatHistogram) DetectReset(previous *FloatHistogram) bool {
if h.CounterResetHint == CounterReset {
return true
}
if h.CounterResetHint == NotCounterReset {
return false
}
// In all other cases of CounterResetHint (UnknownCounterReset and GaugeType),
// we go on as we would otherwise, for reasons explained below.
//
// If the CounterResetHint is UnknownCounterReset, we do not know yet if this histogram comes
// with a counter reset. Therefore, we have to do all the detailed work to find out if there
// is a counter reset or not.
// We do the same if the CounterResetHint is GaugeType, which should not happen, but PromQL still
// allows the user to apply functions to gauge histograms that are only meant for counter histograms.
// In this case, we treat the gauge histograms as a counter histograms
// (and we plan to return a warning about it to the user).
if h.Count < previous.Count {
return true
}
if h.Schema > previous.Schema {
return true
}
if h.ZeroThreshold < previous.ZeroThreshold {
// ZeroThreshold decreased.
return true
}
previousZeroCount, newThreshold := previous.zeroCountForLargerThreshold(h.ZeroThreshold)
if newThreshold != h.ZeroThreshold {
// ZeroThreshold is within a populated bucket in previous
// histogram.
return true
}
if h.ZeroCount < previousZeroCount {
return true
}
currIt := h.floatBucketIterator(true, h.ZeroThreshold, h.Schema)
prevIt := previous.floatBucketIterator(true, h.ZeroThreshold, h.Schema)
if detectReset(currIt, prevIt) {
return true
}
currIt = h.floatBucketIterator(false, h.ZeroThreshold, h.Schema)
prevIt = previous.floatBucketIterator(false, h.ZeroThreshold, h.Schema)
return detectReset(currIt, prevIt)
}
func detectReset(currIt, prevIt BucketIterator[float64]) bool {
if !prevIt.Next() {
return false // If no buckets in previous histogram, nothing can be reset.
}
prevBucket := prevIt.At()
if !currIt.Next() {
// No bucket in current, but at least one in previous
// histogram. Check if any of those are non-zero, in which case
// this is a reset.
for {
if prevBucket.Count != 0 {
return true
}
if !prevIt.Next() {
return false
}
}
}
currBucket := currIt.At()
for {
// Forward currIt until we find the bucket corresponding to prevBucket.
for currBucket.Index < prevBucket.Index {
if !currIt.Next() {
// Reached end of currIt early, therefore
// previous histogram has a bucket that the
// current one does not have. Unlass all
// remaining buckets in the previous histogram
// are unpopulated, this is a reset.
for {
if prevBucket.Count != 0 {
return true
}
if !prevIt.Next() {
return false
}
}
}
currBucket = currIt.At()
}
if currBucket.Index > prevBucket.Index {
// Previous histogram has a bucket the current one does
// not have. If it's populated, it's a reset.
if prevBucket.Count != 0 {
return true
}
} else {
// We have reached corresponding buckets in both iterators.
// We can finally compare the counts.
if currBucket.Count < prevBucket.Count {
return true
}
}
if !prevIt.Next() {
// Reached end of prevIt without finding offending buckets.
return false
}
prevBucket = prevIt.At()
}
}
// PositiveBucketIterator returns a BucketIterator to iterate over all positive
// buckets in ascending order (starting next to the zero bucket and going up).
func (h *FloatHistogram) PositiveBucketIterator() BucketIterator[float64] {
return h.floatBucketIterator(true, 0, h.Schema)
}
// NegativeBucketIterator returns a BucketIterator to iterate over all negative
// buckets in descending order (starting next to the zero bucket and going
// down).
func (h *FloatHistogram) NegativeBucketIterator() BucketIterator[float64] {
return h.floatBucketIterator(false, 0, h.Schema)
}
// PositiveReverseBucketIterator returns a BucketIterator to iterate over all
// positive buckets in descending order (starting at the highest bucket and
// going down towards the zero bucket).
func (h *FloatHistogram) PositiveReverseBucketIterator() BucketIterator[float64] {
return newReverseFloatBucketIterator(h.PositiveSpans, h.PositiveBuckets, h.Schema, true)
}
// NegativeReverseBucketIterator returns a BucketIterator to iterate over all
// negative buckets in ascending order (starting at the lowest bucket and going
// up towards the zero bucket).
func (h *FloatHistogram) NegativeReverseBucketIterator() BucketIterator[float64] {
return newReverseFloatBucketIterator(h.NegativeSpans, h.NegativeBuckets, h.Schema, false)
}
// AllBucketIterator returns a BucketIterator to iterate over all negative,
// zero, and positive buckets in ascending order (starting at the lowest bucket
// and going up). If the highest negative bucket or the lowest positive bucket
// overlap with the zero bucket, their upper or lower boundary, respectively, is
// set to the zero threshold.
func (h *FloatHistogram) AllBucketIterator() BucketIterator[float64] {
return &allFloatBucketIterator{
h: h,
leftIter: h.NegativeReverseBucketIterator(),
rightIter: h.PositiveBucketIterator(),
state: -1,
}
}
// AllReverseBucketIterator returns a BucketIterator to iterate over all negative,
// zero, and positive buckets in descending order (starting at the lowest bucket
// and going up). If the highest negative bucket or the lowest positive bucket
// overlap with the zero bucket, their upper or lower boundary, respectively, is
// set to the zero threshold.
func (h *FloatHistogram) AllReverseBucketIterator() BucketIterator[float64] {
return &allFloatBucketIterator{
h: h,
leftIter: h.PositiveReverseBucketIterator(),
rightIter: h.NegativeBucketIterator(),
state: -1,
}
}
// zeroCountForLargerThreshold returns what the histogram's zero count would be
// if the ZeroThreshold had the provided larger (or equal) value. If the
// provided value is less than the histogram's ZeroThreshold, the method panics.
// If the largerThreshold ends up within a populated bucket of the histogram, it
// is adjusted upwards to the lower limit of that bucket (all in terms of
// absolute values) and that bucket's count is included in the returned
// count. The adjusted threshold is returned, too.
func (h *FloatHistogram) zeroCountForLargerThreshold(largerThreshold float64) (count, threshold float64) {
// Fast path.
if largerThreshold == h.ZeroThreshold {
return h.ZeroCount, largerThreshold
}
if largerThreshold < h.ZeroThreshold {
panic(fmt.Errorf("new threshold %f is less than old threshold %f", largerThreshold, h.ZeroThreshold))
}
outer:
for {
count = h.ZeroCount
i := h.PositiveBucketIterator()
for i.Next() {
b := i.At()
if b.Lower >= largerThreshold {
break
}
count += b.Count // Bucket to be merged into zero bucket.
if b.Upper > largerThreshold {
// New threshold ended up within a bucket. if it's
// populated, we need to adjust largerThreshold before
// we are done here.
if b.Count != 0 {
largerThreshold = b.Upper
}
break
}
}
i = h.NegativeBucketIterator()
for i.Next() {
b := i.At()
if b.Upper <= -largerThreshold {
break
}
count += b.Count // Bucket to be merged into zero bucket.
if b.Lower < -largerThreshold {
// New threshold ended up within a bucket. If
// it's populated, we need to adjust
// largerThreshold and have to redo the whole
// thing because the treatment of the positive
// buckets is invalid now.
if b.Count != 0 {
largerThreshold = -b.Lower
continue outer
}
break
}
}
return count, largerThreshold
}
}
// trimBucketsInZeroBucket removes all buckets that are within the zero
// bucket. It assumes that the zero threshold is at a bucket boundary and that
// the counts in the buckets to remove are already part of the zero count.
func (h *FloatHistogram) trimBucketsInZeroBucket() {
i := h.PositiveBucketIterator()
bucketsIdx := 0
for i.Next() {
b := i.At()
if b.Lower >= h.ZeroThreshold {
break
}
h.PositiveBuckets[bucketsIdx] = 0
bucketsIdx++
}
i = h.NegativeBucketIterator()
bucketsIdx = 0
for i.Next() {
b := i.At()
if b.Upper <= -h.ZeroThreshold {
break
}
h.NegativeBuckets[bucketsIdx] = 0
bucketsIdx++
}
// We are abusing Compact to trim the buckets set to zero
// above. Premature compacting could cause additional cost, but this
// code path is probably rarely used anyway.
h.Compact(0)
}
// reconcileZeroBuckets finds a zero bucket large enough to include the zero
// buckets of both histograms (the receiving histogram and the other histogram)
// with a zero threshold that is not within a populated bucket in either
// histogram. This method modifies the receiving histogram accourdingly, but
// leaves the other histogram as is. Instead, it returns the zero count the
// other histogram would have if it were modified.
func (h *FloatHistogram) reconcileZeroBuckets(other *FloatHistogram) float64 {
otherZeroCount := other.ZeroCount
otherZeroThreshold := other.ZeroThreshold
for otherZeroThreshold != h.ZeroThreshold {
if h.ZeroThreshold > otherZeroThreshold {
otherZeroCount, otherZeroThreshold = other.zeroCountForLargerThreshold(h.ZeroThreshold)
}
if otherZeroThreshold > h.ZeroThreshold {
h.ZeroCount, h.ZeroThreshold = h.zeroCountForLargerThreshold(otherZeroThreshold)
h.trimBucketsInZeroBucket()
}
}
return otherZeroCount
}
// floatBucketIterator is a low-level constructor for bucket iterators.
//
// If positive is true, the returned iterator iterates through the positive
// buckets, otherwise through the negative buckets.
//
// If absoluteStartValue is < the lowest absolute value of any upper bucket
// boundary, the iterator starts with the first bucket. Otherwise, it will skip
// all buckets with an absolute value of their upper boundary ≤
// absoluteStartValue.
//
// targetSchema must be ≤ the schema of FloatHistogram (and of course within the
// legal values for schemas in general). The buckets are merged to match the
// targetSchema prior to iterating (without mutating FloatHistogram).
func (h *FloatHistogram) floatBucketIterator(
positive bool, absoluteStartValue float64, targetSchema int32,
) *floatBucketIterator {
if targetSchema > h.Schema {
panic(fmt.Errorf("cannot merge from schema %d to %d", h.Schema, targetSchema))
}
i := &floatBucketIterator{
baseBucketIterator: baseBucketIterator[float64, float64]{
schema: h.Schema,
positive: positive,
},
targetSchema: targetSchema,
absoluteStartValue: absoluteStartValue,
}
if positive {
i.spans = h.PositiveSpans
i.buckets = h.PositiveBuckets
} else {
i.spans = h.NegativeSpans
i.buckets = h.NegativeBuckets
}
return i
}
// reverseFloatbucketiterator is a low-level constructor for reverse bucket iterators.
func newReverseFloatBucketIterator(
spans []Span, buckets []float64, schema int32, positive bool,
) *reverseFloatBucketIterator {
r := &reverseFloatBucketIterator{
baseBucketIterator: baseBucketIterator[float64, float64]{
schema: schema,
spans: spans,
buckets: buckets,
positive: positive,
},
}
r.spansIdx = len(r.spans) - 1
r.bucketsIdx = len(r.buckets) - 1
if r.spansIdx >= 0 {
r.idxInSpan = int32(r.spans[r.spansIdx].Length) - 1
}
r.currIdx = 0
for _, s := range r.spans {
r.currIdx += s.Offset + int32(s.Length)
}
return r
}
type floatBucketIterator struct {
baseBucketIterator[float64, float64]
targetSchema int32 // targetSchema is the schema to merge to and must be ≤ schema.
origIdx int32 // The bucket index within the original schema.
absoluteStartValue float64 // Never return buckets with an upper bound ≤ this value.
}
func (i *floatBucketIterator) At() Bucket[float64] {
// Need to use i.targetSchema rather than i.baseBucketIterator.schema.
return i.baseBucketIterator.at(i.targetSchema)
}
func (i *floatBucketIterator) Next() bool {
if i.spansIdx >= len(i.spans) {
return false
}
// Copy all of these into local variables so that we can forward to the
// next bucket and then roll back if needed.
origIdx, spansIdx, idxInSpan := i.origIdx, i.spansIdx, i.idxInSpan
span := i.spans[spansIdx]
firstPass := true
i.currCount = 0
mergeLoop: // Merge together all buckets from the original schema that fall into one bucket in the targetSchema.
for {
if i.bucketsIdx == 0 {
// Seed origIdx for the first bucket.
origIdx = span.Offset
} else {
origIdx++
}
for idxInSpan >= span.Length {
// We have exhausted the current span and have to find a new
// one. We even handle pathologic spans of length 0 here.
idxInSpan = 0
spansIdx++
if spansIdx >= len(i.spans) {
if firstPass {
return false
}
break mergeLoop
}
span = i.spans[spansIdx]
origIdx += span.Offset
}
currIdx := i.targetIdx(origIdx)
switch {
case firstPass:
i.currIdx = currIdx
firstPass = false
case currIdx != i.currIdx:
// Reached next bucket in targetSchema.
// Do not actually forward to the next bucket, but break out.
break mergeLoop
}
i.currCount += i.buckets[i.bucketsIdx]
idxInSpan++
i.bucketsIdx++
i.origIdx, i.spansIdx, i.idxInSpan = origIdx, spansIdx, idxInSpan
if i.schema == i.targetSchema {
// Don't need to test the next bucket for mergeability
// if we have no schema change anyway.
break mergeLoop
}
}
// Skip buckets before absoluteStartValue.
// TODO(beorn7): Maybe do something more efficient than this recursive call.
if getBound(i.currIdx, i.targetSchema) <= i.absoluteStartValue {
return i.Next()
}
return true
}
// targetIdx returns the bucket index within i.targetSchema for the given bucket
// index within i.schema.
func (i *floatBucketIterator) targetIdx(idx int32) int32 {
if i.schema == i.targetSchema {
// Fast path for the common case. The below would yield the same
// result, just with more effort.
return idx
}
return ((idx - 1) >> (i.schema - i.targetSchema)) + 1
}
type reverseFloatBucketIterator struct {
baseBucketIterator[float64, float64]
idxInSpan int32 // Changed from uint32 to allow negative values for exhaustion detection.
}
func (i *reverseFloatBucketIterator) Next() bool {
i.currIdx--
if i.bucketsIdx < 0 {
return false
}
for i.idxInSpan < 0 {
// We have exhausted the current span and have to find a new
// one. We'll even handle pathologic spans of length 0.
i.spansIdx--
i.idxInSpan = int32(i.spans[i.spansIdx].Length) - 1
i.currIdx -= i.spans[i.spansIdx+1].Offset
}
i.currCount = i.buckets[i.bucketsIdx]
i.bucketsIdx--
i.idxInSpan--
return true
}
type allFloatBucketIterator struct {
h *FloatHistogram
leftIter, rightIter BucketIterator[float64]
// -1 means we are iterating negative buckets.
// 0 means it is time for the zero bucket.
// 1 means we are iterating positive buckets.
// Anything else means iteration is over.
state int8
currBucket Bucket[float64]
}
func (i *allFloatBucketIterator) Next() bool {
switch i.state {
case -1:
if i.leftIter.Next() {
i.currBucket = i.leftIter.At()
switch {
case i.currBucket.Upper < 0 && i.currBucket.Upper > -i.h.ZeroThreshold:
i.currBucket.Upper = -i.h.ZeroThreshold
case i.currBucket.Lower > 0 && i.currBucket.Lower < i.h.ZeroThreshold:
i.currBucket.Lower = i.h.ZeroThreshold
}
return true
}
i.state = 0
return i.Next()
case 0:
i.state = 1
if i.h.ZeroCount > 0 {
i.currBucket = Bucket[float64]{
Lower: -i.h.ZeroThreshold,
Upper: i.h.ZeroThreshold,
LowerInclusive: true,
UpperInclusive: true,
Count: i.h.ZeroCount,
// Index is irrelevant for the zero bucket.
}
return true
}
return i.Next()
case 1:
if i.rightIter.Next() {
i.currBucket = i.rightIter.At()
switch {
case i.currBucket.Lower > 0 && i.currBucket.Lower < i.h.ZeroThreshold:
i.currBucket.Lower = i.h.ZeroThreshold
case i.currBucket.Upper < 0 && i.currBucket.Upper > -i.h.ZeroThreshold:
i.currBucket.Upper = -i.h.ZeroThreshold
}
return true
}
i.state = 42
return false
}
return false
}
func (i *allFloatBucketIterator) At() Bucket[float64] {
return i.currBucket
}