Method Combine

pkg/traceql/engine_metrics.go:1864–1928

(in []*tempopb.TimeSeries)

}

func (h *HistogramAggregator) Combine(in []*tempopb.TimeSeries) {
	for _, ts := range in {
		// Convert proto labels to traceql labels while stripping the bucket label
		var bucket Static
		h.labelBuffer, bucket = convertProtoLabelsToTraceQL(ts.Labels, true)

		if bucket.Type == TypeNil {
			// Bad __bucket label?
			continue
		}

		withoutBucketStr := h.labelBuffer.String()

		existing, ok := h.ss[withoutBucketStr]
		if !ok {
			// Create a copy of the labels since the buffer will be reused
			withoutBucket := make(Labels, len(h.labelBuffer))
			copy(withoutBucket, h.labelBuffer)

			// Create a fresh histogram slice for each series to avoid sharing memory
			// Pool reuse was causing data contamination between series
			histSlice := make([]Histogram, h.intervalMapper.IntervalCount())

			existing = histSeries{
				labels: withoutBucket,
				hist:   histSlice,
				// Pre-allocate exemplars slice with capacity hint to reduce allocations
				exemplars:       make([]Exemplar, 0, len(ts.Exemplars)),
				exemplarBuckets: h.exemplarBucketsCreator(),
			}
		}

		b := bucket.Float()

		for _, sample := range ts.Samples {
			if sample.Value == 0 {
				continue
			}
			j := h.intervalMapper.IntervalMs(sample.TimestampMs)
			if j >= 0 && j < len(existing.hist) {
				existing.hist[j].Record(b, int(sample.Value))
			}
		}

		// Collect exemplars per series, not globally
		for _, exemplar := range ts.Exemplars {
			if existing.exemplarBuckets.testTotal() {
				break
			}
			if existing.exemplarBuckets.addAndTest(uint64(exemplar.TimestampMs)) {
				continue // Skip this exemplar and continue, next exemplar might fit in a different bucket
			}

			// Convert exemplar labels directly (no need for buffer reuse here)
			labels, _ := convertProtoLabelsToTraceQL(exemplar.Labels, false)

			existing.exemplars = append(existing.exemplars, Exemplar{
				Labels:      labels,