Changelog rc.1 (#5215)

* update changelog

Signed-off-by: Joe Elliott <number101010@gmail.com>

* ci?

---------

Signed-off-by: Joe Elliott <number101010@gmail.com>

.github/workflows/docker.yml

@@ -25,7 +25,7 @@ jobs:
           persist-credentials: false
 
       - name: fetch tags
-        run: git fetch --tags
+        run: git fetch --tags --force
 
       - id: get-tag
         run: |
@@ -50,7 +50,7 @@ jobs:
           persist-credentials: false
 
       - name: fetch tags
-        run: git fetch --tags
+        run: git fetch --tags --force
 
       - name: build-tempo-binaries
         run: |
@@ -118,7 +118,7 @@ jobs:
           persist-credentials: false
 
       - name: fetch tags
-        run: git fetch --tags
+        run: git fetch --tags --force
 
       - name: get-tag
         run: |

.github/workflows/release.yml

@@ -12,7 +12,7 @@ jobs:
     if: github.repository == 'grafana/tempo'  # skip in forks
     runs-on: ubuntu-24.04
     permissions:
-      contents: read
+      contents: write
       id-token: write
     env:
       NFPM_SIGNING_KEY_FILE: /tmp/nfpm-private-key.key
@@ -23,7 +23,7 @@ jobs:
           persist-credentials: false
 
       - name: fetch tags
-        run: git fetch --tags
+        run: git fetch --tags --force
 
       - id: "get-secrets"
         name: "get nfpm signing keys"

.goreleaser.yml

@@ -11,7 +11,7 @@ builds:
     env:
       - CGO_ENABLED=0
     goos:
-      - darwin
+#      - darwin  re-enable when https://github.com/golang/go/issues/73617 is fixed
       - linux
       - windows
     goarch:
@@ -41,7 +41,7 @@ builds:
     env:
       - CGO_ENABLED=0
     goos:
-      - darwin
+#      - darwin  re-enable when https://github.com/golang/go/issues/73617 is fixed
       - linux
       - windows
     goarch:
@@ -71,7 +71,7 @@ builds:
     env:
       - CGO_ENABLED=0
     goos:
-      - darwin
+#      - darwin  re-enable when https://github.com/golang/go/issues/73617 is fixed
       - linux
       - windows
     goarch:

CHANGELOG.md

@@ -1,5 +1,14 @@
 ## main / unreleased
 
+# v2.8.0-rc.1
+
+* [ENHANCEMENT] TraceQL Metrics: distribute exemplars over time [#5158](https://github.com/grafana/tempo/pull/5158) (@ruslan-mikhailov)
+* [ENHANCEMENT] TraceQL Metrics: hard limit number of exemplars [#5158](https://github.com/grafana/tempo/pull/5158) (@ruslan-mikhailov)
+* [BUGFIX] Excluded nestedSetParent and other values from compare() function [#5196](https://github.com/grafana/tempo/pull/5196) (@mdisibio)
+* [BUGFIX] Fix distributor issue where a hash collision could lead to spans stored incorrectly [#5186](https://github.com/grafana/tempo/pull/5186) (@mdisibio)
+* [BUGFIX] Fix structural metrics rate by aggregation [#5204](https://github.com/grafana/tempo/pull/5204) (@zalegrala)
+* [BUGFIX] TraceQL Metrics: right exemplars for histogram and quantiles [#5145](https://github.com/grafana/tempo/pull/5145) (@ruslan-mikhailov)
+
 # v2.8.0-rc.0
 
 * [CHANGE] **BREAKING CHANGE** Change default http-listen-port from 80 to 3200 [#4960](https://github.com/grafana/tempo/pull/4960) (@martialblog)

integration/e2e/api/query_range_test.go

@@ -2,7 +2,6 @@ package api
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -54,7 +53,20 @@ sendLoop:
 	for {
 		select {
 		case <-ticker.C:
-			require.NoError(t, jaegerClient.EmitBatch(context.Background(), util.MakeThriftBatch()))
+			require.NoError(t, jaegerClient.EmitBatch(context.Background(),
+				util.MakeThriftBatchWithSpanCountAttributeAndName(
+					1, "my operation",
+					"res_val", "span_val",
+					"res_attr", "span_attr",
+				),
+			))
+			require.NoError(t, jaegerClient.EmitBatch(context.Background(),
+				util.MakeThriftBatchWithSpanCountAttributeAndName(
+					1, "my operation",
+					"res_val2", "span_val2",
+					"res_attr", "span_attr",
+				),
+			))
 		case <-timer.C:
 			break sendLoop
 		}
@@ -64,34 +76,164 @@ sendLoop:
 	require.NoError(t, tempo.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"tempo_metrics_generator_processor_local_blocks_spans_total"}, e2e.WaitMissingMetrics))
 	require.NoError(t, tempo.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"tempo_metrics_generator_processor_local_blocks_cut_blocks"}, e2e.WaitMissingMetrics))
 
-	for _, query := range []string{
-		"{} | rate()",
-		"{} | compare({status=error})",
-		"{} | count_over_time()",
-		"{} | min_over_time(duration)",
-		"{} | max_over_time(duration)",
-		"{} | avg_over_time(duration)",
-		"{} | sum_over_time(duration)",
-		"{} | quantile_over_time(duration, .5)",
-		"{} | quantile_over_time(duration, .5, 0.9, 0.99)",
-		"{} | histogram_over_time(duration)",
-		"{} | count_over_time() by (status)",
-		"{status != error} | count_over_time() by (status)",
+	for _, exeplarsCase := range []struct {
+		name              string
+		exemplars         int
+		expectedExemplars int
+	}{
+		{
+			name:              "default",
+			exemplars:         0,
+			expectedExemplars: 100, // if set to 0, then limits to 100
+		},
+		{
+			name:              "5 exemplar",
+			exemplars:         5,
+			expectedExemplars: 5,
+		},
+		{
+			name:              "25 exemplars",
+			exemplars:         25,
+			expectedExemplars: 25,
+		},
+		{
+			name:              "capped exemplars",
+			exemplars:         1000,
+			expectedExemplars: 100, // capped to 100
+		},
 	} {
-		t.Run(query, func(t *testing.T) {
-			queryRangeRes := callQueryRange(t, tempo.Endpoint(tempoPort), query, debugMode)
+		for _, query := range []string{
+			"{} | rate()",
+			"{} | compare({status=error})",
+			"{} | count_over_time()",
+			"{} | min_over_time(duration)",
+			"{} | max_over_time(duration)",
+			"{} | avg_over_time(duration)",
+			"{} | sum_over_time(duration)",
+			"{} | quantile_over_time(duration, .5)",
+			"{} | quantile_over_time(duration, .5, 0.9, 0.99)",
+
+			"{} | count_over_time() by (span.span_attr)",
+			"{} | count_over_time() by (resource.res_attr)",
+			"{} | count_over_time() by (.span_attr)",
+			"{} | count_over_time() by (.res_attr)",
+
+			"{} | histogram_over_time(duration)",
+			"{} | count_over_time() by (status)",
+			"{status != error} | count_over_time() by (status)",
+		} {
+			t.Run(fmt.Sprintf("%s: %s", exeplarsCase.name, query), func(t *testing.T) {
+				queryRangeRes := callQueryRange(t, tempo.Endpoint(tempoPort), query, exeplarsCase.exemplars, debugMode)
+				require.NotNil(t, queryRangeRes)
+				require.GreaterOrEqual(t, len(queryRangeRes.GetSeries()), 1)
+				if query == "{} | quantile_over_time(duration, .5, 0.9, 0.99)" {
+					// Bug: https://github.com/grafana/tempo/issues/5167
+					t.Skip("Bug in quantile_over_time in calculating exemplars")
+				}
+
+				exemplarCount := 0
+
+				for _, series := range queryRangeRes.GetSeries() {
+					exemplarCount += len(series.GetExemplars())
+				}
+				assert.LessOrEqual(t, exemplarCount, exeplarsCase.expectedExemplars)
+				assert.GreaterOrEqual(t, exemplarCount, 1)
+			})
+		}
+	}
+
+	// check exemplars in more detail
+	for _, testCase := range []struct {
+		query                   string
+		targetAttribute         string
+		targetExemplarAttribute string
+	}{
+		{
+			query:                   "{} | quantile_over_time(duration, .9) by (span.span_attr)",
+			targetAttribute:         "span.span_attr",
+			targetExemplarAttribute: "span.span_attr",
+		},
+		{
+			query:                   "{} | quantile_over_time(duration, .9) by (resource.res_attr)",
+			targetAttribute:         "resource.res_attr",
+			targetExemplarAttribute: "resource.res_attr",
+		},
+		{
+			query:                   "{} | quantile_over_time(duration, .9) by (.span_attr)",
+			targetAttribute:         ".span_attr",
+			targetExemplarAttribute: "span.span_attr",
+		},
+		{
+			query:                   "{} | quantile_over_time(duration, .9) by (.res_attr)",
+			targetAttribute:         ".res_attr",
+			targetExemplarAttribute: "resource.res_attr",
+		},
+		{
+			query:                   "{} | rate() by (span.span_attr)",
+			targetAttribute:         "span.span_attr",
+			targetExemplarAttribute: "span.span_attr",
+		},
+		{
+			query:                   "{} | count_over_time() by (span.span_attr)",
+			targetAttribute:         "span.span_attr",
+			targetExemplarAttribute: "span.span_attr",
+		},
+		{
+			query:                   "{} | min_over_time(duration) by (span.span_attr)",
+			targetAttribute:         "span.span_attr",
+			targetExemplarAttribute: "span.span_attr",
+		},
+		{
+			query:                   "{} | max_over_time(duration) by (span.span_attr)",
+			targetAttribute:         "span.span_attr",
+			targetExemplarAttribute: "span.span_attr",
+		},
+		{
+			query:                   "{} | avg_over_time(duration) by (span.span_attr)",
+			targetAttribute:         "span.span_attr",
+			targetExemplarAttribute: "span.span_attr",
+		},
+		{
+			query:                   "{} | sum_over_time(duration) by (span.span_attr)",
+			targetAttribute:         "span.span_attr",
+			targetExemplarAttribute: "span.span_attr",
+		},
+	} {
+		t.Run(testCase.query, func(t *testing.T) {
+			queryRangeRes := callQueryRange(t, tempo.Endpoint(tempoPort), testCase.query, 100, debugMode)
 			require.NotNil(t, queryRangeRes)
-			require.GreaterOrEqual(t, len(queryRangeRes.GetSeries()), 1)
-			exemplarCount := 0
-			for _, series := range queryRangeRes.GetSeries() {
-				exemplarCount += len(series.GetExemplars())
+			require.Equal(t, len(queryRangeRes.GetSeries()), 2)
+
+			// Verify that all exemplars in this series belongs to the right series
+			// by matching attribute values
+			for _, series := range queryRangeRes.Series {
+				// search attribute value for the series
+				var expectedAttrValue string
+				for _, label := range series.Labels {
+					if label.Key == testCase.targetAttribute {
+						expectedAttrValue = label.Value.GetStringValue()
+						break
+					}
+				}
+				require.NotEmpty(t, expectedAttrValue)
+
+				// check attribute value in exemplars
+				for _, exemplar := range series.Exemplars {
+					var actualAttrValue string
+					for _, label := range exemplar.Labels {
+						if label.Key == testCase.targetExemplarAttribute {
+							actualAttrValue = label.Value.GetStringValue()
+							break
+						}
+					}
+					require.Equal(t, expectedAttrValue, actualAttrValue)
+				}
 			}
-			require.GreaterOrEqual(t, exemplarCount, 1)
 		})
 	}
 
 	// invalid query
-	res := doRequest(t, tempo.Endpoint(tempoPort), "{. a}")
+	res := doRequest(t, tempo.Endpoint(tempoPort), "{. a}", 100)
 	require.Equal(t, 400, res.StatusCode)
 
 	// query with empty results
@@ -102,7 +244,7 @@ sendLoop:
 		`{span.randomattr = "doesnotexist"} | count_over_time()`,
 	} {
 		t.Run(query, func(t *testing.T) {
-			queryRangeRes := callQueryRange(t, tempo.Endpoint(tempoPort), query, debugMode)
+			queryRangeRes := callQueryRange(t, tempo.Endpoint(tempoPort), query, 100, debugMode)
 			require.NotNil(t, queryRangeRes)
 			// it has time series but they are empty and has no exemplars
 			require.GreaterOrEqual(t, len(queryRangeRes.GetSeries()), 1)
@@ -142,7 +284,7 @@ func TestQueryRangeSingleTrace(t *testing.T) {
 
 	// Query the trace by count. As we have only one trace, we should get one dot with value 1
 	query := "{} | count_over_time()"
-	queryRangeRes := callQueryRange(t, tempo.Endpoint(tempoPort), query, debugMode)
+	queryRangeRes := callQueryRange(t, tempo.Endpoint(tempoPort), query, 100, debugMode)
 	require.NotNil(t, queryRangeRes)
 	require.Equal(t, len(queryRangeRes.GetSeries()), 1)
 
@@ -359,8 +501,8 @@ sendLoop:
 	require.Equal(t, spanCount, len(queryRangeRes.GetSeries()))
 }
 
-func callQueryRange(t *testing.T, endpoint, query string, printBody bool) tempopb.QueryRangeResponse {
-	res := doRequest(t, endpoint, query)
+func callQueryRange(t *testing.T, endpoint, query string, exemplars int, printBody bool) tempopb.QueryRangeResponse {
+	res := doRequest(t, endpoint, query, exemplars)
 	require.Equal(t, http.StatusOK, res.StatusCode)
 
 	// Read body and print it
@@ -371,12 +513,14 @@ func callQueryRange(t *testing.T, endpoint, query string, printBody bool) tempop
 	}
 
 	queryRangeRes := tempopb.QueryRangeResponse{}
-	require.NoError(t, json.Unmarshal(body, &queryRangeRes))
+	readBody := strings.NewReader(string(body))
+	err = new(jsonpb.Unmarshaler).Unmarshal(readBody, &queryRangeRes)
+	require.NoError(t, err)
 	return queryRangeRes
 }
 
-func doRequest(t *testing.T, endpoint, query string) *http.Response {
-	url := buildURL(endpoint, fmt.Sprintf("%s with(exemplars=true)", query))
+func doRequest(t *testing.T, endpoint, query string, exemplars int) *http.Response {
+	url := buildURL(endpoint, fmt.Sprintf("%s with(exemplars=true)", query), exemplars)
 	req, err := http.NewRequest(http.MethodGet, url, nil)
 	require.NoError(t, err)
 
@@ -385,13 +529,14 @@ func doRequest(t *testing.T, endpoint, query string) *http.Response {
 	return res
 }
 
-func buildURL(endpoint, query string) string {
+func buildURL(endpoint, query string, exemplars int) string {
 	return fmt.Sprintf(
-		"http://%s/api/metrics/query_range?query=%s&start=%d&end=%d&step=%s",
+		"http://%s/api/metrics/query_range?query=%s&start=%d&end=%d&step=%s&exemplars=%d",
 		endpoint,
 		url.QueryEscape(query),
 		time.Now().Add(-5*time.Minute).UnixNano(),
 		time.Now().Add(time.Minute).UnixNano(),
 		"5s",
+		exemplars,
 	)
 }

integration/e2e/limits_test.go

@@ -239,6 +239,9 @@ func TestQueryLimits(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		_, err = client.QueryTrace(tempoUtil.TraceIDToHexString(traceID[:]))
+		if err == nil {
+			return false
+		}
 		return strings.Contains(err.Error(), trace.ErrTraceTooLarge.Error())
 	}, time.Minute, time.Second)
 

modules/distributor/distributor.go

@@ -42,8 +42,7 @@ import (
 	v1_common "github.com/grafana/tempo/pkg/tempopb/common/v1"
 	v1 "github.com/grafana/tempo/pkg/tempopb/trace/v1"
 	"github.com/grafana/tempo/pkg/usagestats"
-	tempo_util "github.com/grafana/tempo/pkg/util"
-
+	"github.com/grafana/tempo/pkg/util"
 	"github.com/grafana/tempo/pkg/validation"
 )
 
@@ -472,7 +471,7 @@ func (d *Distributor) PushTraces(ctx context.Context, traces ptrace.Traces) (*te
 
 	maxAttributeBytes := d.getMaxAttributeBytes(userID)
 
-	keys, rebatchedTraces, truncatedAttributeCount, err := requestsByTraceID(batches, userID, spanCount, maxAttributeBytes)
+	ringTokens, rebatchedTraces, truncatedAttributeCount, err := requestsByTraceID(batches, userID, spanCount, maxAttributeBytes)
 	if err != nil {
 		logDiscardedResourceSpans(batches, userID, &d.cfg.LogDiscardedSpans, d.logger)
 		return nil, err
@@ -482,7 +481,7 @@ func (d *Distributor) PushTraces(ctx context.Context, traces ptrace.Traces) (*te
 		metricAttributesTruncated.WithLabelValues(userID).Add(float64(truncatedAttributeCount))
 	}
 
-	err = d.sendToIngestersViaBytes(ctx, userID, rebatchedTraces, keys)
+	err = d.sendToIngestersViaBytes(ctx, userID, rebatchedTraces, ringTokens)
 	if err != nil {
 		return nil, err
 	}
@@ -492,7 +491,7 @@ func (d *Distributor) PushTraces(ctx context.Context, traces ptrace.Traces) (*te
 	}
 
 	if d.kafkaProducer != nil {
-		err := d.sendToKafka(ctx, userID, keys, rebatchedTraces)
+		err := d.sendToKafka(ctx, userID, ringTokens, rebatchedTraces)
 		if err != nil {
 			level.Error(d.logger).Log("msg", "failed to write to kafka", "err", err)
 			return nil, err
@@ -500,7 +499,7 @@ func (d *Distributor) PushTraces(ctx context.Context, traces ptrace.Traces) (*te
 	} else {
 		// See if we need to send to the generators
 		if len(d.overrides.MetricsGeneratorProcessors(userID)) > 0 {
-			d.generatorForwarder.SendTraces(ctx, userID, keys, rebatchedTraces)
+			d.generatorForwarder.SendTraces(ctx, userID, ringTokens, rebatchedTraces)
 		}
 	}
 
@@ -704,11 +703,11 @@ func (d *Distributor) sendToKafka(ctx context.Context, userID string, keys []uin
 // and traces to pass onto the ingesters.
 func requestsByTraceID(batches []*v1.ResourceSpans, userID string, spanCount, maxSpanAttrSize int) ([]uint32, []*rebatchedTrace, int, error) {
 	const tracesPerBatch = 20 // p50 of internal env
-	tracesByID := make(map[uint32]*rebatchedTrace, tracesPerBatch)
+	tracesByID := make(map[uint64]*rebatchedTrace, tracesPerBatch)
 	truncatedAttributeCount := 0
 
 	for _, b := range batches {
-		spansByILS := make(map[uint32]*v1.ScopeSpans)
+		spansByILS := make(map[uint64]*v1.ScopeSpans)
 		// check resource for large attributes
 		if maxSpanAttrSize > 0 && b.Resource != nil {
 			resourceAttrTruncatedCount := processAttributes(b.Resource.Attributes, maxSpanAttrSize)
@@ -745,11 +744,11 @@ func requestsByTraceID(batches []*v1.ResourceSpans, userID string, spanCount, ma
 					return nil, nil, 0, status.Errorf(codes.InvalidArgument, "trace ids must be 128 bit, received %d bits", len(traceID)*8)
 				}
 
-				traceKey := tempo_util.TokenFor(userID, traceID)
+				traceKey := util.HashForTraceID(traceID)
 				ilsKey := traceKey
 				if ils.Scope != nil {
-					ilsKey = fnv1a.AddString32(ilsKey, ils.Scope.Name)
-					ilsKey = fnv1a.AddString32(ilsKey, ils.Scope.Version)
+					ilsKey = fnv1a.AddString64(ilsKey, ils.Scope.Name)
+					ilsKey = fnv1a.AddString64(ilsKey, ils.Scope.Version)
 				}
 
 				existingILS, ilsAdded := spansByILS[ilsKey]
@@ -800,15 +799,15 @@ func requestsByTraceID(batches []*v1.ResourceSpans, userID string, spanCount, ma
 
 	metricTracesPerBatch.Observe(float64(len(tracesByID)))
 
-	keys := make([]uint32, 0, len(tracesByID))
+	ringTokens := make([]uint32, 0, len(tracesByID))
 	traces := make([]*rebatchedTrace, 0, len(tracesByID))
 
-	for k, r := range tracesByID {
-		keys = append(keys, k)
-		traces = append(traces, r)
+	for _, tr := range tracesByID {
+		ringTokens = append(ringTokens, util.TokenFor(userID, tr.id))
+		traces = append(traces, tr)
 	}
 
-	return keys, traces, truncatedAttributeCount, nil
+	return ringTokens, traces, truncatedAttributeCount, nil
 }
 
 // find and truncate the span attributes that are too large
@@ -993,7 +992,7 @@ func logSpans(batches []*v1.ResourceSpans, cfg *LogSpansConfig, logger log.Logge
 				loggerWithAtts = log.With(
 					loggerWithAtts,
 					"span_"+strutil.SanitizeLabelName(a.GetKey()),
-					tempo_util.StringifyAnyValue(a.GetValue()))
+					util.StringifyAnyValue(a.GetValue()))
 			}
 		}
 
@@ -1015,7 +1014,7 @@ func logSpan(s *v1.Span, allAttributes bool, logger log.Logger) {
 			logger = log.With(
 				logger,
 				"span_"+strutil.SanitizeLabelName(a.GetKey()),
-				tempo_util.StringifyAnyValue(a.GetValue()))
+				util.StringifyAnyValue(a.GetValue()))
 		}
 
 		latencySeconds := float64(s.GetEndTimeUnixNano()-s.GetStartTimeUnixNano()) / float64(time.Second.Nanoseconds())

modules/distributor/distributor_test.go

@@ -76,8 +76,13 @@ func TestRequestsByTraceID(t *testing.T) {
 	traceIDA := []byte{0x0A, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F}
 	traceIDB := []byte{0x0B, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F}
 
+	// These 2 trace IDs are known to collide under fnv32
+	collision1, _ := util.HexStringToTraceID("fd5980503add11f09f80f77608c1b2da")
+	collision2, _ := util.HexStringToTraceID("091ea7803ade11f0998a055186ee1243")
+
 	tests := []struct {
 		name           string
+		emptyTenant    bool
 		batches        []*v1.ResourceSpans
 		expectedKeys   []uint32
 		expectedTraces []*tempopb.Trace
@@ -715,32 +720,206 @@ func TestRequestsByTraceID(t *testing.T) {
 			expectedStarts: []uint32{10, 60},
 			expectedEnds:   []uint32{50, 80},
 		},
+		{
+			// These 2 trace IDs are known to collide under fnv32
+			name:        "known collisions",
+			emptyTenant: true,
+			batches: []*v1.ResourceSpans{
+				{
+					Resource: &v1_resource.Resource{
+						DroppedAttributesCount: 3,
+					},
+					ScopeSpans: []*v1.ScopeSpans{
+						{
+							Scope: &v1_common.InstrumentationScope{
+								Name: "test",
+							},
+							Spans: []*v1.Span{
+								{
+									TraceId:           collision2,
+									Name:              "spanA",
+									StartTimeUnixNano: uint64(30 * time.Second),
+									EndTimeUnixNano:   uint64(40 * time.Second),
+								},
+								{
+									TraceId:           collision2,
+									Name:              "spanC",
+									StartTimeUnixNano: uint64(20 * time.Second),
+									EndTimeUnixNano:   uint64(50 * time.Second),
+								},
+								{
+									TraceId:           collision1,
+									Name:              "spanE",
+									StartTimeUnixNano: uint64(70 * time.Second),
+									EndTimeUnixNano:   uint64(80 * time.Second),
+								},
+							},
+						},
+					},
+				},
+				{
+					Resource: &v1_resource.Resource{
+						DroppedAttributesCount: 4,
+					},
+					ScopeSpans: []*v1.ScopeSpans{
+						{
+							Scope: &v1_common.InstrumentationScope{
+								Name: "test2",
+							},
+							Spans: []*v1.Span{
+								{
+									TraceId:           collision2,
+									Name:              "spanB",
+									StartTimeUnixNano: uint64(10 * time.Second),
+									EndTimeUnixNano:   uint64(30 * time.Second),
+								},
+								{
+									TraceId:           collision1,
+									Name:              "spanD",
+									StartTimeUnixNano: uint64(60 * time.Second),
+									EndTimeUnixNano:   uint64(80 * time.Second),
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedKeys: []uint32{
+				util.TokenFor("", collision1),
+				util.TokenFor("", collision2),
+			},
+			expectedTraces: []*tempopb.Trace{
+				{
+					ResourceSpans: []*v1.ResourceSpans{
+						{
+							Resource: &v1_resource.Resource{
+								DroppedAttributesCount: 3,
+							},
+							ScopeSpans: []*v1.ScopeSpans{
+								{
+									Scope: &v1_common.InstrumentationScope{
+										Name: "test",
+									},
+									Spans: []*v1.Span{
+										{
+											TraceId:           collision1,
+											Name:              "spanE",
+											StartTimeUnixNano: uint64(70 * time.Second),
+											EndTimeUnixNano:   uint64(80 * time.Second),
+										},
+									},
+								},
+							},
+						},
+						{
+							Resource: &v1_resource.Resource{
+								DroppedAttributesCount: 4,
+							},
+							ScopeSpans: []*v1.ScopeSpans{
+								{
+									Scope: &v1_common.InstrumentationScope{
+										Name: "test2",
+									},
+									Spans: []*v1.Span{
+										{
+											TraceId:           collision1,
+											Name:              "spanD",
+											StartTimeUnixNano: uint64(60 * time.Second),
+											EndTimeUnixNano:   uint64(80 * time.Second),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					ResourceSpans: []*v1.ResourceSpans{
+						{
+							Resource: &v1_resource.Resource{
+								DroppedAttributesCount: 3,
+							},
+							ScopeSpans: []*v1.ScopeSpans{
+								{
+									Scope: &v1_common.InstrumentationScope{
+										Name: "test",
+									},
+									Spans: []*v1.Span{
+										{
+											TraceId:           collision2,
+											Name:              "spanA",
+											StartTimeUnixNano: uint64(30 * time.Second),
+											EndTimeUnixNano:   uint64(40 * time.Second),
+										},
+										{
+											TraceId:           collision2,
+											Name:              "spanC",
+											StartTimeUnixNano: uint64(20 * time.Second),
+											EndTimeUnixNano:   uint64(50 * time.Second),
+										},
+									},
+								},
+							},
+						},
+						{
+							Resource: &v1_resource.Resource{
+								DroppedAttributesCount: 4,
+							},
+							ScopeSpans: []*v1.ScopeSpans{
+								{
+									Scope: &v1_common.InstrumentationScope{
+										Name: "test2",
+									},
+									Spans: []*v1.Span{
+										{
+											TraceId:           collision2,
+											Name:              "spanB",
+											StartTimeUnixNano: uint64(10 * time.Second),
+											EndTimeUnixNano:   uint64(30 * time.Second),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedIDs: [][]byte{
+				collision1,
+				collision2,
+			},
+			expectedStarts: []uint32{60, 10},
+			expectedEnds:   []uint32{80, 50},
+		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			keys, rebatchedTraces, _, err := requestsByTraceID(tt.batches, util.FakeTenantID, 1, 1000)
-			require.Equal(t, len(keys), len(rebatchedTraces))
+			tenant := util.FakeTenantID
+			if tt.emptyTenant {
+				tenant = ""
+			}
+			ringTokens, rebatchedTraces, _, err := requestsByTraceID(tt.batches, tenant, 1, 1000)
+			require.Equal(t, len(ringTokens), len(rebatchedTraces))
 
-			for i, expectedKey := range tt.expectedKeys {
+			for i, expectedID := range tt.expectedIDs {
 				foundIndex := -1
-				for j, key := range keys {
-					if expectedKey == key {
+				for j, tr := range rebatchedTraces {
+					if bytes.Equal(expectedID, tr.id) {
 						foundIndex = j
+						break
 					}
 				}
 				require.NotEqual(t, -1, foundIndex, "expected key %d not found", foundIndex)
 
 				// now confirm that the request at this position is the expected one
-				expectedReq := tt.expectedTraces[i]
-				actualReq := rebatchedTraces[foundIndex].trace
-				assert.Equal(t, expectedReq, actualReq)
-				assert.Equal(t, tt.expectedIDs[i], rebatchedTraces[foundIndex].id)
-				assert.Equal(t, tt.expectedStarts[i], rebatchedTraces[foundIndex].start)
-				assert.Equal(t, tt.expectedEnds[i], rebatchedTraces[foundIndex].end)
+				require.Equal(t, tt.expectedIDs[i], rebatchedTraces[foundIndex].id)
+				require.Equal(t, tt.expectedTraces[i], rebatchedTraces[foundIndex].trace)
+				require.Equal(t, tt.expectedStarts[i], rebatchedTraces[foundIndex].start)
+				require.Equal(t, tt.expectedEnds[i], rebatchedTraces[foundIndex].end)
 			}
 
-			assert.Equal(t, tt.expectedErr, err)
+			require.Equal(t, tt.expectedErr, err)
 		})
 	}
 }

modules/frontend/metrics_query_range_sharder.go

@@ -129,9 +129,12 @@ func (s queryRangeSharder) RoundTrip(pipelineRequest pipeline.Request) (pipeline
 		cutoff                = time.Now().Add(-s.cfg.QueryBackendAfter)
 	)
 
+	backendExemplars, generatorExemplars := s.exemplarsCutoff(*req, cutoff)
+	req.Exemplars = generatorExemplars
 	generatorReq := s.generatorRequest(tenantID, pipelineRequest, *req, cutoff)
-	reqCh := make(chan pipeline.Request, 2) // buffer of 2 allows us to insert generatorReq and metrics
+	req.Exemplars = backendExemplars
 
+	reqCh := make(chan pipeline.Request, 2) // buffer of 2 allows us to insert generatorReq and metrics
 	if generatorReq != nil {
 		reqCh <- generatorReq
 	}
@@ -165,11 +168,70 @@ func (s queryRangeSharder) RoundTrip(pipelineRequest pipeline.Request) (pipeline
 	return pipeline.NewAsyncSharderChan(ctx, s.cfg.ConcurrentRequests, reqCh, jobMetricsResponse, s.next), nil
 }
 
-func (s *queryRangeSharder) exemplarsPerShard(total uint32, exemplars uint32) uint32 {
-	if exemplars == 0 || total == 0 {
-		return 0
+// exemplarsPerShard fairly distributes exemplars over time per each shard.
+// Example: with metas containing 2 blocks, where first block duration is 90 seconds and
+// second block duration is 10 seconds, with limit = 100, the first block will
+// get 90*1.2 exemplars and the second block will get 10*1.2 exemplars
+func (s *queryRangeSharder) exemplarsPerShard(metas []*backend.BlockMeta, limit uint32) []uint32 {
+	const overhead = 1.2 // 20% overhead for shard size
+
+	exemplars := make([]uint32, len(metas))
+	if limit == 0 || len(metas) == 0 {
+		return exemplars
 	}
-	return max(uint32(math.Ceil(float64(exemplars)*1.2))/total, 1) // require at least 1 exemplar per shard
+
+	// Calculate total duration across all blocks
+	var totalDurationNanos int64
+	for _, meta := range metas {
+		if meta.EndTime.Before(meta.StartTime) { // Skip blocks with invalid time ranges
+			continue
+		}
+
+		totalDurationNanos += meta.EndTime.UnixNano() - meta.StartTime.UnixNano()
+	}
+
+	if totalDurationNanos <= 0 {
+		return exemplars
+	}
+
+	// Distribute exemplars proportionally based on block duration
+	var blockDuration int64
+	var share float64
+	for i, meta := range metas {
+		if meta.EndTime.Before(meta.StartTime) { // Skip blocks with invalid time ranges
+			continue
+		}
+
+		blockDuration = meta.EndTime.UnixNano() - meta.StartTime.UnixNano()
+		share = (float64(blockDuration) / float64(totalDurationNanos)) * float64(limit) * overhead
+		exemplars[i] = max(uint32(math.Ceil(share)), 1)
+	}
+
+	return exemplars
+}
+
+// exemplarsCutoff calculates how to distribute exemplars between the generator (for recent data) and
+// backend blocks. It returns two values: the number of exemplars for blocks before the cutoff time,
+// and the number of exemplars for data after the cutoff time. The distribution is proportional to
+// the time range of each segment relative to the total query time range.
+func (s *queryRangeSharder) exemplarsCutoff(req tempopb.QueryRangeRequest, cutoff time.Time) (uint32, uint32) {
+	timeRange := req.End - req.Start
+	limit := req.Exemplars
+	traceql.TrimToAfter(&req, cutoff)
+
+	if req.Start >= req.End { // no need to query generator
+		return limit, 0 // after - no exemplars needed
+	}
+	if req.End-req.Start >= timeRange { // no need to query backend
+		return 0, limit
+	}
+
+	shareAfterCutoff := float64(limit) * float64(req.End-req.Start) / float64(timeRange)
+	shareAfterCutoffCeil := uint32(math.Ceil(shareAfterCutoff))
+	if limit <= shareAfterCutoffCeil {
+		return 0, limit // after - receives all exemplars
+	}
+	return limit - shareAfterCutoffCeil, shareAfterCutoffCeil
 }
 
 func (s *queryRangeSharder) backendRequests(ctx context.Context, tenantID string, parent pipeline.Request, searchReq tempopb.QueryRangeRequest, cutoff time.Time, targetBytesPerRequest int, reqCh chan pipeline.Request) (totalJobs, totalBlocks uint32, totalBlockBytes uint64) {
@@ -228,8 +290,8 @@ func (s *queryRangeSharder) buildBackendRequests(ctx context.Context, tenantID s
 	queryHash := hashForQueryRangeRequest(&searchReq)
 	colsToJSON := api.NewDedicatedColumnsToJSON()
 
-	exemplarsPerBlock := s.exemplarsPerShard(uint32(len(metas)), searchReq.Exemplars)
-	for _, m := range metas {
+	exemplarsPerBlock := s.exemplarsPerShard(metas, searchReq.Exemplars)
+	for i, m := range metas {
 		if m.EndTime.Before(m.StartTime) {
 			// Ignore blocks with bad timings from debugging
 			continue
@@ -240,7 +302,7 @@ func (s *queryRangeSharder) buildBackendRequests(ctx context.Context, tenantID s
 			continue
 		}
 
-		exemplars := exemplarsPerBlock
+		exemplars := exemplarsPerBlock[i]
 		if exemplars > 0 {
 			// Scale the number of exemplars per block to match the size
 			// of each sub request on this block. For very small blocks or other edge cases, return at least 1.

modules/frontend/metrics_query_range_sharder_test.go

@@ -17,24 +17,6 @@ import (
 	"github.com/grafana/tempo/tempodb/backend"
 )
 
-func FuzzExemplarsPerShard(f *testing.F) {
-	f.Add(uint32(1), uint32(10))  // total = 1, exemplars = 10
-	f.Add(uint32(100), uint32(1)) // total = 100, exemplars = 1
-	f.Add(uint32(10), uint32(0))  // total = 10, exemplars = 0
-
-	s := &queryRangeSharder{}
-
-	f.Fuzz(func(t *testing.T, total uint32, exemplars uint32) {
-		result := s.exemplarsPerShard(total, exemplars)
-
-		if exemplars == 0 || total == 0 {
-			assert.Equal(t, uint32(0), result, "if exemplars is 0 or total is 0, result should be 0")
-		} else {
-			assert.Greater(t, result, uint32(0), "result should be greater than 0")
-		}
-	})
-}
-
 func TestBuildBackendRequestsExemplarsOneBlock(t *testing.T) {
 	// Create the test sharder with exemplars enabled
 	sharder := &queryRangeSharder{
@@ -192,3 +174,244 @@ func extractExemplarsValue(t *testing.T, uri string) int {
 
 	return exemplarsValue
 }
+
+func TestExemplarsPerShard(t *testing.T) {
+	s := &queryRangeSharder{}
+
+	createBlockMeta := func(durationSeconds int) *backend.BlockMeta {
+		now := time.Now()
+		return &backend.BlockMeta{
+			BlockID:   backend.MustParse(uuid.NewString()),
+			StartTime: now.Add(-time.Duration(durationSeconds) * time.Second),
+			EndTime:   now,
+		}
+	}
+
+	testCases := []struct {
+		name           string
+		metas          []*backend.BlockMeta
+		limit          uint32
+		expectedResult []uint32
+	}{
+		{
+			name:           "limit is zero",
+			metas:          []*backend.BlockMeta{createBlockMeta(60)},
+			limit:          0,
+			expectedResult: []uint32{0},
+		},
+		{
+			name:           "metas is empty",
+			metas:          []*backend.BlockMeta{},
+			limit:          100,
+			expectedResult: []uint32{},
+		},
+		{
+			name: "proportional distribution based on duration",
+			metas: []*backend.BlockMeta{
+				createBlockMeta(90),
+				createBlockMeta(10),
+			},
+			limit:          100,
+			expectedResult: []uint32{108, 12}, // 90*1.2 = 108, 10*1.2 = 12
+		},
+		{
+			name: "at least one exemplar per valid block",
+			metas: []*backend.BlockMeta{
+				createBlockMeta(1000), // large block
+				createBlockMeta(1),    // very small block
+			},
+			limit:          10,
+			expectedResult: []uint32{12, 1}, // First gets 9*1.2, second gets at least 1
+		},
+		{
+			name: "mixed valid and invalid blocks",
+			metas: []*backend.BlockMeta{
+				createBlockMeta(60),
+				createBlockMeta(-60), // invalid block
+				createBlockMeta(60),
+			},
+			limit:          100,
+			expectedResult: []uint32{60, 0, 60},
+		},
+		{
+			name: "only invalid blocks",
+			metas: []*backend.BlockMeta{
+				createBlockMeta(-60),
+				createBlockMeta(-60),
+			},
+			limit:          100,
+			expectedResult: []uint32{0, 0},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := s.exemplarsPerShard(tc.metas, tc.limit)
+			assert.Equal(t, tc.expectedResult, result)
+		})
+	}
+}
+
+func FuzzExemplarsPerShard(f *testing.F) {
+	f.Add(uint32(100), uint32(60)) // limit = 100, duration = 60s
+	f.Add(uint32(0), uint32(30))   // limit = 0, duration = 30s
+	f.Add(uint32(1000), uint32(0)) // limit = 1000, duration = 0s
+
+	s := &queryRangeSharder{}
+
+	f.Fuzz(func(t *testing.T, limit uint32, value uint32) {
+		now := time.Now()
+		metas := []*backend.BlockMeta{
+			{
+				BlockID:   backend.MustParse(uuid.NewString()),
+				StartTime: now.Add(-time.Duration(value) * time.Second),
+				EndTime:   now,
+			},
+		}
+
+		result := s.exemplarsPerShard(metas, limit)
+		require.Len(t, result, 1, "result should have one element")
+
+		if limit == 0 || value == 0 {
+			assert.Equal(t, uint32(0), result[0], "result should be 0")
+		} else {
+			assert.Greater(t, result[0], uint32(0), "result should be greater than 0")
+		}
+	})
+}
+
+// nolint: gosec // G115
+func TestExemplarsCutoff(t *testing.T) {
+	s := &queryRangeSharder{}
+	now := time.Now()
+	cutoff := now.Add(-1 * time.Hour)
+
+	testCases := []struct {
+		name              string
+		req               tempopb.QueryRangeRequest
+		expectedBeforeCut uint32
+		expectedAfterCut  uint32
+	}{
+		{
+			// When all data is after the cutoff, all exemplars should go to the 'after' portion
+			name: "all data after cutoff",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(50 * time.Minute).UnixNano()),
+				End:       uint64(now.UnixNano()),
+				Exemplars: 100,
+			},
+			expectedBeforeCut: 0,
+			expectedAfterCut:  100,
+		},
+		{
+			// When all data is before the cutoff, all exemplars should go to the 'before' portion
+			name: "all data before cutoff",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(-2 * time.Hour).UnixNano()),
+				End:       uint64(cutoff.Add(-10 * time.Minute).UnixNano()),
+				Exemplars: 100,
+			},
+			expectedBeforeCut: 100,
+			expectedAfterCut:  0,
+		},
+		{
+			name: "data spans the cutoff - 75% after",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(-20 * time.Minute).UnixNano()),
+				End:       uint64(now.UnixNano()),
+				Exemplars: 100,
+			},
+			expectedBeforeCut: 25,
+			expectedAfterCut:  75,
+		},
+		{
+			name: "data spans the cutoff - 25% after",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(-3 * time.Hour).UnixNano()),
+				End:       uint64(cutoff.Add(1 * time.Hour).UnixNano()),
+				Exemplars: 100,
+			},
+			expectedBeforeCut: 75,
+			expectedAfterCut:  25,
+		},
+		// in case of small limits, it gives favor to after (request to generator)
+		{
+			name: "small limit: 25% after",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(-3 * time.Hour).UnixNano()),
+				End:       uint64(cutoff.Add(1 * time.Hour).UnixNano()),
+				Exemplars: 2,
+			},
+			expectedBeforeCut: 1,
+			expectedAfterCut:  1,
+		},
+		{
+			name: "small limit: 25% after",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(-3 * time.Hour).UnixNano()),
+				End:       uint64(cutoff.Add(1 * time.Hour).UnixNano()),
+				Exemplars: 1,
+			},
+			expectedBeforeCut: 0,
+			expectedAfterCut:  1,
+		},
+		{
+			name: "small limit: 75% after",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(-20 * time.Minute).UnixNano()),
+				End:       uint64(now.UnixNano()),
+				Exemplars: 2,
+			},
+			expectedBeforeCut: 0,
+			expectedAfterCut:  2,
+		},
+		{
+			name: "small limit: 75% after",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(-20 * time.Minute).UnixNano()),
+				End:       uint64(now.UnixNano()),
+				Exemplars: 1,
+			},
+			expectedBeforeCut: 0,
+			expectedAfterCut:  1,
+		},
+		{
+			name: "exactly at cutoff",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.UnixNano()),
+				End:       uint64(now.UnixNano()),
+				Exemplars: 100,
+			},
+			expectedBeforeCut: 0,
+			expectedAfterCut:  100,
+		},
+		{
+			name: "exactly at cutoff",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(cutoff.Add(-1 * time.Hour).UnixNano()),
+				End:       uint64(cutoff.UnixNano()),
+				Exemplars: 100,
+			},
+			expectedBeforeCut: 100,
+			expectedAfterCut:  0,
+		},
+		{
+			name: "start equals end",
+			req: tempopb.QueryRangeRequest{
+				Start:     uint64(now.UnixNano()),
+				End:       uint64(now.UnixNano()),
+				Exemplars: 100,
+			},
+			expectedBeforeCut: 100,
+			expectedAfterCut:  0,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			beforeCut, afterCut := s.exemplarsCutoff(tc.req, cutoff)
+			assert.Equal(t, tc.expectedBeforeCut, beforeCut, "Exemplars before cutoff should match expected value")
+			assert.Equal(t, tc.expectedAfterCut, afterCut, "Exemplars after cutoff should match expected value")
+		})
+	}
+}

pkg/traceql/ast_metrics.go

@@ -72,10 +72,24 @@ func (a *MetricsAggregate) extractConditions(request *FetchSpansRequest) {
 	}
 
 	for _, b := range a.by {
-		if !request.HasAttribute(b) {
-			request.SecondPassConditions = append(request.SecondPassConditions, Condition{
-				Attribute: b,
-			})
+		// In the case of the AllConditions, it is enough to check that the
+		// attribute is present in any of the passes.
+		if request.AllConditions {
+			if !request.HasAttribute(b) {
+				request.SecondPassConditions = append(request.SecondPassConditions, Condition{
+					Attribute: b,
+				})
+			}
+		} else {
+			// In the case of AllConditions set to false, as is the case with a
+			// structural query, we need to ensure that the `by` attribute is present
+			// in the second pass conditions as well, so that we load the column and
+			// can return the appropriate values.
+			if !request.SecondPassHasAttribute(b) {
+				request.SecondPassConditions = append(request.SecondPassConditions, Condition{
+					Attribute: b,
+				})
+			}
 		}
 	}
 }
@@ -226,16 +240,16 @@ func exemplarAttribute(a Attribute) func(Span) (float64, uint64) {
 func (a *MetricsAggregate) initSum(q *tempopb.QueryRangeRequest) {
 	// Currently all metrics are summed by job to produce
 	// intermediate results. This will change when adding min/max/topk/etc
-	a.seriesAgg = NewSimpleCombiner(q, a.simpleAggregationOp)
+	a.seriesAgg = NewSimpleCombiner(q, a.simpleAggregationOp, maxExemplars)
 }
 
 func (a *MetricsAggregate) initFinal(q *tempopb.QueryRangeRequest) {
 	switch a.op {
 	case metricsAggregateQuantileOverTime:
-		a.seriesAgg = NewHistogramAggregator(q, a.floats)
+		a.seriesAgg = NewHistogramAggregator(q, a.floats, q.Exemplars)
 	default:
 		// These are simple additions by series
-		a.seriesAgg = NewSimpleCombiner(q, a.simpleAggregationOp)
+		a.seriesAgg = NewSimpleCombiner(q, a.simpleAggregationOp, q.Exemplars)
 	}
 }
 

pkg/traceql/engine_metrics.go

@@ -444,16 +444,18 @@ func NewStepAggregator(start, end, step uint64, innerAgg func() VectorAggregator
 		vectors[i] = innerAgg()
 	}
 
-	exemplars := make([]Exemplar, 0, maxExemplars)
-
 	return &StepAggregator{
-		start:           start,
-		end:             end,
-		step:            step,
-		intervals:       intervals,
-		vectors:         vectors,
-		exemplars:       exemplars,
-		exemplarBuckets: newBucketSet(intervals),
+		start:     start,
+		end:       end,
+		step:      step,
+		intervals: intervals,
+		vectors:   vectors,
+		exemplars: make([]Exemplar, 0, maxExemplars),
+		exemplarBuckets: newBucketSet(
+			maxExemplars,
+			alignStart(start, step),
+			alignEnd(end, step),
+		),
 	}
 }
 
@@ -469,8 +471,7 @@ func (s *StepAggregator) ObserveExemplar(value float64, ts uint64, lbls Labels)
 	if s.exemplarBuckets.testTotal() {
 		return
 	}
-	interval := IntervalOfMs(int64(ts), s.start, s.end, s.step)
-	if s.exemplarBuckets.addAndTest(interval) {
+	if s.exemplarBuckets.addAndTest(ts) {
 		return
 	}
 
@@ -1240,7 +1241,7 @@ type SimpleAggregator struct {
 	initWithNaN      bool
 }
 
-func NewSimpleCombiner(req *tempopb.QueryRangeRequest, op SimpleAggregationOp) *SimpleAggregator {
+func NewSimpleCombiner(req *tempopb.QueryRangeRequest, op SimpleAggregationOp, exemplars uint32) *SimpleAggregator {
 	l := IntervalCount(req.Start, req.End, req.Step)
 	var initWithNaN bool
 	var f func(existingValue float64, newValue float64) float64
@@ -1263,8 +1264,12 @@ func NewSimpleCombiner(req *tempopb.QueryRangeRequest, op SimpleAggregationOp) *
 
 	}
 	return &SimpleAggregator{
-		ss:              make(SeriesSet),
-		exemplarBuckets: newBucketSet(l),
+		ss: make(SeriesSet),
+		exemplarBuckets: newBucketSet(
+			exemplars,
+			alignStart(req.Start, req.Step),
+			alignEnd(req.End, req.Step),
+		),
 		len:             l,
 		start:           req.Start,
 		end:             req.End,
@@ -1321,8 +1326,7 @@ func (b *SimpleAggregator) aggregateExemplars(ts *tempopb.TimeSeries, existing *
 		if b.exemplarBuckets.testTotal() {
 			break
 		}
-		interval := IntervalOfMs(exemplar.TimestampMs, b.start, b.end, b.step)
-		if b.exemplarBuckets.addAndTest(interval) {
+		if b.exemplarBuckets.addAndTest(uint64(exemplar.TimestampMs)) { //nolint: gosec // G115
 			continue // Skip this exemplar and continue, next exemplar might fit in a different bucket	}
 		}
 		labels := make(Labels, 0, len(exemplar.Labels))
@@ -1372,8 +1376,9 @@ func (h *Histogram) Record(bucket float64, count int) {
 }
 
 type histSeries struct {
-	labels Labels
-	hist   []Histogram
+	labels    Labels
+	hist      []Histogram
+	exemplars []Exemplar
 }
 
 type HistogramAggregator struct {
@@ -1381,20 +1386,23 @@ type HistogramAggregator struct {
 	qs               []float64
 	len              int
 	start, end, step uint64
-	exemplars        []Exemplar
 	exemplarBuckets  *bucketSet
 }
 
-func NewHistogramAggregator(req *tempopb.QueryRangeRequest, qs []float64) *HistogramAggregator {
+func NewHistogramAggregator(req *tempopb.QueryRangeRequest, qs []float64, exemplars uint32) *HistogramAggregator {
 	l := IntervalCount(req.Start, req.End, req.Step)
 	return &HistogramAggregator{
-		qs:              qs,
-		ss:              make(map[string]histSeries),
-		len:             l,
-		start:           req.Start,
-		end:             req.End,
-		step:            req.Step,
-		exemplarBuckets: newBucketSet(l),
+		qs:    qs,
+		ss:    make(map[string]histSeries),
+		len:   l,
+		start: req.Start,
+		end:   req.End,
+		step:  req.Step,
+		exemplarBuckets: newBucketSet(
+			exemplars,
+			alignStart(req.Start, req.Step),
+			alignEnd(req.End, req.Step),
+		),
 	}
 }
 
@@ -1408,7 +1416,6 @@ func (h *HistogramAggregator) Combine(in []*tempopb.TimeSeries) {
 		var bucket Static
 		for _, l := range ts.Labels {
 			if l.Key == internalLabelBucket {
-				// bucket = int(l.Value.GetIntValue())
 				bucket = StaticFromAnyValue(l.Value)
 				continue
 			}
@@ -1431,7 +1438,6 @@ func (h *HistogramAggregator) Combine(in []*tempopb.TimeSeries) {
 				labels: withoutBucket,
 				hist:   make([]Histogram, h.len),
 			}
-			h.ss[withoutBucketStr] = existing
 		}
 
 		b := bucket.Float()
@@ -1450,8 +1456,7 @@ func (h *HistogramAggregator) Combine(in []*tempopb.TimeSeries) {
 			if h.exemplarBuckets.testTotal() {
 				break
 			}
-			interval := IntervalOfMs(exemplar.TimestampMs, h.start, h.end, h.step)
-			if h.exemplarBuckets.addAndTest(interval) {
+			if h.exemplarBuckets.addAndTest(uint64(exemplar.TimestampMs)) { //nolint: gosec // G115
 				continue // Skip this exemplar and continue, next exemplar might fit in a different bucket
 			}
 
@@ -1462,12 +1467,13 @@ func (h *HistogramAggregator) Combine(in []*tempopb.TimeSeries) {
 					Value: StaticFromAnyValue(l.Value),
 				})
 			}
-			h.exemplars = append(h.exemplars, Exemplar{
+			existing.exemplars = append(existing.exemplars, Exemplar{
 				Labels:      labels,
 				Value:       exemplar.Value,
 				TimestampMs: uint64(exemplar.TimestampMs),
 			})
 		}
+		h.ss[withoutBucketStr] = existing
 	}
 }
 
@@ -1485,7 +1491,7 @@ func (h *HistogramAggregator) Results() SeriesSet {
 			ts := TimeSeries{
 				Labels:    labels,
 				Values:    make([]float64, len(in.hist)),
-				Exemplars: h.exemplars,
+				Exemplars: in.exemplars,
 			}
 			for i := range in.hist {
 

pkg/traceql/engine_metrics_average.go

@@ -39,7 +39,11 @@ func (a *averageOverTimeAggregator) init(q *tempopb.QueryRangeRequest, mode Aggr
 		start:                 q.Start,
 		end:                   q.End,
 		step:                  q.Step,
-		exemplarBuckets:       newBucketSet(IntervalCount(q.Start, q.End, q.Step)),
+		exemplarBuckets: newBucketSet(
+			maxExemplars,
+			alignStart(q.Start, q.Step),
+			alignEnd(q.End, q.Step),
+		),
 	}
 
 	if mode == AggregateModeRaw {
@@ -292,8 +296,7 @@ func (b *averageOverTimeSeriesAggregator) aggregateExemplars(ts *tempopb.TimeSer
 		if b.exemplarBuckets.testTotal() {
 			break
 		}
-		interval := IntervalOfMs(exemplar.TimestampMs, b.start, b.end, b.step)
-		if b.exemplarBuckets.addAndTest(interval) {
+		if b.exemplarBuckets.addAndTest(uint64(exemplar.TimestampMs)) { //nolint: gosec // G115
 			continue // Skip this exemplar and continue, next exemplar might fit in a different bucket	}
 		}
 		labels := make(Labels, 0, len(exemplar.Labels))
@@ -451,8 +454,7 @@ func (g *avgOverTimeSpanAggregator[F, S]) ObserveExemplar(span Span, value float
 	if s.exemplarBuckets.testTotal() {
 		return
 	}
-	interval := IntervalOfMs(int64(ts), g.start, g.end, g.step)
-	if s.exemplarBuckets.addAndTest(interval) {
+	if s.exemplarBuckets.addAndTest(ts) {
 		return
 	}
 
@@ -533,10 +535,14 @@ func (g *avgOverTimeSpanAggregator[F, S]) getSeries(span Span) avgOverTimeSeries
 	if !ok {
 		intervals := IntervalCount(g.start, g.end, g.step)
 		s = avgOverTimeSeries[S]{
-			vals:            g.buf.vals,
-			average:         newAverageSeries(intervals, maxExemplars, nil),
-			exemplarBuckets: newBucketSet(intervals),
-			initialized:     true,
+			vals:    g.buf.vals,
+			average: newAverageSeries(intervals, maxExemplars, nil),
+			exemplarBuckets: newBucketSet(
+				maxExemplars,
+				alignStart(g.start, g.step),
+				alignEnd(g.end, g.step),
+			),
+			initialized: true,
 		}
 		g.series[g.buf.fast] = s
 	}

pkg/traceql/engine_metrics_compare.go

@@ -79,11 +79,11 @@ func (m *MetricsCompare) init(q *tempopb.QueryRangeRequest, mode AggregateMode)
 		m.selectionTotals = make(map[Attribute][]float64)
 
 	case AggregateModeSum:
-		m.seriesAgg = NewSimpleCombiner(q, sumAggregation)
+		m.seriesAgg = NewSimpleCombiner(q, sumAggregation, maxExemplars)
 		return
 
 	case AggregateModeFinal:
-		m.seriesAgg = NewBaselineAggregator(q, m.topN)
+		m.seriesAgg = NewBaselineAggregator(q, m.topN, q.Exemplars)
 		return
 	}
 }
@@ -136,11 +136,15 @@ func (m *MetricsCompare) observe(span Span) {
 			return
 		}
 
-		// These attributes get pulled back by select all but we never
-		// group by them because the cardinality isn't useful.
+		// These attributes get pulled back by select all, or might be used in a comparison query,
+		// but we never group by them because the cardinality isn't useful.
 		switch a {
 		case IntrinsicSpanStartTimeAttribute,
-			IntrinsicTraceIDAttribute:
+			IntrinsicTraceIDAttribute,
+			IntrinsicParentIDAttribute,
+			IntrinsicNestedSetLeftAttribute,
+			IntrinsicNestedSetRightAttribute,
+			IntrinsicNestedSetParentAttribute:
 			return
 		}
 
@@ -339,7 +343,7 @@ type staticWithTimeSeries struct {
 	series TimeSeries
 }
 
-func NewBaselineAggregator(req *tempopb.QueryRangeRequest, topN int) *BaselineAggregator {
+func NewBaselineAggregator(req *tempopb.QueryRangeRequest, topN int, exemplars uint32) *BaselineAggregator {
 	l := IntervalCount(req.Start, req.End, req.Step)
 	return &BaselineAggregator{
 		baseline:        make(map[string]map[StaticMapKey]staticWithTimeSeries),
@@ -352,7 +356,11 @@ func NewBaselineAggregator(req *tempopb.QueryRangeRequest, topN int) *BaselineAg
 		end:             req.End,
 		step:            req.Step,
 		topN:            topN,
-		exemplarBuckets: newBucketSet(l),
+		exemplarBuckets: newBucketSet(
+			exemplars,
+			alignStart(req.Start, req.Step),
+			alignEnd(req.End, req.Step),
+		),
 	}
 }
 
@@ -434,8 +442,7 @@ func (b *BaselineAggregator) Combine(ss []*tempopb.TimeSeries) {
 			if b.exemplarBuckets.testTotal() {
 				break
 			}
-			interval := IntervalOfMs(exemplar.TimestampMs, b.start, b.end, b.step)
-			if b.exemplarBuckets.addAndTest(interval) {
+			if b.exemplarBuckets.addAndTest(uint64(exemplar.TimestampMs)) { //nolint: gosec // G115
 				continue
 			}
 

pkg/traceql/engine_metrics_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/grafana/tempo/pkg/tempopb"
+	commonv1proto "github.com/grafana/tempo/pkg/tempopb/common/v1"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -302,6 +303,31 @@ func TestCompileMetricsQueryRangeFetchSpansRequest(t *testing.T) {
 				},
 			},
 		},
+		"structural_rate_by": {
+			q: "{name=`foo`} > {} | rate() by (name)",
+			expectedReq: FetchSpansRequest{
+				AllConditions: false,
+				Conditions: []Condition{
+					{
+						Attribute: NewIntrinsic(IntrinsicStructuralChild),
+					},
+					{
+						Attribute: IntrinsicNameAttribute,
+						Op:        OpEqual,
+						Operands:  Operands{NewStaticString("foo")},
+					},
+				},
+				SecondPassConditions: []Condition{
+					{
+						Attribute: IntrinsicNameAttribute,
+					},
+					{
+						// Since there is already a second pass then span start time isn't optimized to the first pass.
+						Attribute: IntrinsicSpanStartTimeAttribute,
+					},
+				},
+			},
+		},
 	}
 
 	for n, tc := range tc {
@@ -1883,3 +1909,117 @@ func BenchmarkSumOverTime(b *testing.B) {
 		_, _, _ = runTraceQLMetric(req, in, in2)
 	}
 }
+
+func BenchmarkHistogramAggregator_Combine(b *testing.B) {
+	// nolint:gosec // G115
+	req := &tempopb.QueryRangeRequest{
+		Start:     uint64(time.Now().Add(-1 * time.Hour).UnixNano()),
+		End:       uint64(time.Now().UnixNano()),
+		Step:      uint64(15 * time.Second.Nanoseconds()),
+		Exemplars: maxExemplars,
+	}
+	const seriesCount = 6
+
+	benchmarks := []struct {
+		name          string
+		samplesCount  int
+		exemplarCount int
+	}{
+		{"Small", 10, 5},
+		{"Medium", 100, 20},
+		{"Large", 1000, 100},
+	}
+
+	for _, bm := range benchmarks {
+		b.Run(bm.name, func(b *testing.B) {
+			series := generateTestTimeSeries(seriesCount, bm.samplesCount, bm.exemplarCount, req.Start, req.End)
+
+			for b.Loop() {
+				agg := NewHistogramAggregator(req, []float64{0.5, 0.9, 0.99}, uint32(bm.exemplarCount)) // nolint: gosec // G115
+				agg.Combine(series)
+			}
+		})
+	}
+}
+
+// generateTestTimeSeries creates test time series data for benchmarking
+// nolint:gosec // G115
+func generateTestTimeSeries(seriesCount, samplesCount, exemplarCount int, start, end uint64) []*tempopb.TimeSeries {
+	result := make([]*tempopb.TimeSeries, seriesCount)
+
+	timeRange := end - start
+
+	for i := 0; i < seriesCount; i++ {
+		// Create unique labels for each series
+		labels := []commonv1proto.KeyValue{
+			{
+				Key: "service",
+				Value: &commonv1proto.AnyValue{
+					Value: &commonv1proto.AnyValue_StringValue{
+						StringValue: "service-" + fmt.Sprintf("%d", i),
+					},
+				},
+			},
+			{
+				Key: internalLabelBucket,
+				Value: &commonv1proto.AnyValue{
+					Value: &commonv1proto.AnyValue_DoubleValue{
+						DoubleValue: math.Pow(2, float64(i%20)), // Power of 2 as bucket
+					},
+				},
+			},
+		}
+
+		samples := make([]tempopb.Sample, samplesCount)
+		for j := 0; j < samplesCount; j++ {
+			// Distribute samples evenly across the time range
+			offset := (uint64(j) * timeRange) / uint64(samplesCount)
+			ts := time.Unix(0, int64(start+offset)).UnixMilli()
+			samples[j] = tempopb.Sample{
+				TimestampMs: ts,
+				Value:       float64(j % 100), // Simple pattern for test data
+			}
+		}
+
+		// Create exemplars
+		exemplars := make([]tempopb.Exemplar, exemplarCount)
+		for j := 0; j < exemplarCount; j++ {
+			// Distribute exemplars evenly across the time range
+			offset := (uint64(j) * timeRange) / uint64(exemplarCount)
+			ts := time.Unix(0, int64(start+offset)).UnixMilli()
+			exemplarLabels := []commonv1proto.KeyValue{
+				{
+					Key: "trace_id",
+					Value: &commonv1proto.AnyValue{
+						Value: &commonv1proto.AnyValue_StringValue{
+							StringValue: fmt.Sprintf("trace-%d", i*1000+j),
+						},
+					},
+				},
+				{
+					Key: "span_id",
+					Value: &commonv1proto.AnyValue{
+						Value: &commonv1proto.AnyValue_StringValue{
+							StringValue: fmt.Sprintf("span-%d", j),
+						},
+					},
+				},
+			}
+			exemplars[j] = tempopb.Exemplar{
+				Labels: exemplarLabels,
+
+				Value:       float64(j % 100), // Simple pattern for test data
+				TimestampMs: ts,
+			}
+		}
+
+		result[i] = &tempopb.TimeSeries{
+			PromLabels: fmt.Sprintf("{service=\"service-%d\",bucket=\"%d\"}", i, i%20),
+			Labels:     labels,
+			Samples:    samples,
+			Exemplars:  exemplars,
+		}
+	}
+
+	return result
+}

pkg/traceql/storage.go

@@ -124,6 +124,16 @@ func (f *FetchSpansRequest) HasAttribute(a Attribute) bool {
 	return false
 }
 
+func (f *FetchSpansRequest) SecondPassHasAttribute(a Attribute) bool {
+	for _, cc := range f.SecondPassConditions {
+		if cc.Attribute == a {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (f *FetchSpansRequest) HasAttributeWithOp(a Attribute, o Operator) bool {
 	for _, cc := range f.Conditions {
 		if cc.Attribute == a && cc.Op == o {

pkg/traceql/util.go

@@ -48,14 +48,35 @@ func MakeCollectTagValueFunc(collect func(tempopb.TagValue) bool) func(v Static)
 type bucketSet struct {
 	sz, maxTotal, maxBucket int
 	buckets                 []int
+	start, end, bucketWidth uint64
 }
 
-func newBucketSet(size int) *bucketSet {
+// newBucketSet creates a new bucket set for the given time range
+// start and end are in nanoseconds
+func newBucketSet(exemplars uint32, start, end uint64) *bucketSet {
+	if exemplars > maxExemplars || exemplars == 0 {
+		exemplars = maxExemplars
+	}
+	buckets := exemplars / maxExemplarsPerBucket
+	if buckets == 0 { // edge case for few exemplars
+		buckets = 1
+	}
+
+	// convert nanoseconds to milliseconds
+	start /= uint64(time.Millisecond.Nanoseconds()) //nolint: gosec // G115
+	end /= uint64(time.Millisecond.Nanoseconds())   //nolint: gosec // G115
+
+	interval := end - start
+	bucketWidth := interval / uint64(buckets)
+
 	return &bucketSet{
-		sz:        size,
-		maxTotal:  maxExemplars,
-		maxBucket: maxExemplarsPerBucket,
-		buckets:   make([]int, size+1), // +1 for total count
+		sz:          int(buckets),
+		maxTotal:    int(exemplars),
+		maxBucket:   maxExemplarsPerBucket,
+		buckets:     make([]int, buckets+1), // +1 for total count
+		start:       start,
+		end:         end,
+		bucketWidth: bucketWidth,
 	}
 }
 
@@ -67,15 +88,33 @@ func (b *bucketSet) testTotal() bool {
 	return b.len() >= b.maxTotal
 }
 
-func (b *bucketSet) inRange(i int) bool {
-	return i >= 0 && i < b.sz
+func (b *bucketSet) inRange(ts uint64) bool {
+	return b.start <= ts && ts <= b.end
 }
 
-func (b *bucketSet) addAndTest(i int) bool {
-	if !b.inRange(i) || b.testTotal() {
+func (b *bucketSet) bucket(ts uint64) int {
+	if b.start == b.end {
+		return 0
+	}
+
+	bucket := int((ts - b.start) / b.bucketWidth) //nolint: gosec // G115
+
+	// Clamp to last bucket to handle edge rounding
+	if bucket >= b.sz {
+		bucket = b.sz - 1
+	}
+
+	return bucket
+}
+
+// addAndTest adds a timestamp to the bucket set and returns true if the total exceeds the max total
+// the timestamp is in milliseconds
+func (b *bucketSet) addAndTest(ts uint64) bool {
+	if !b.inRange(ts) || b.testTotal() || b.sz == 0 {
 		return true
 	}
 
+	i := b.bucket(ts)
 	if b.buckets[i] >= b.maxBucket {
 		return true
 	}

pkg/traceql/util_test.go

@@ -2,22 +2,104 @@ package traceql
 
 import (
 	"testing"
+	"time"
 
-	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/assert"
 )
 
+func TestBucketSet_Bucket(t *testing.T) {
+	testCases := []struct {
+		name     string
+		start    uint64
+		end      uint64
+		ts       uint64
+		expected int
+	}{
+		{
+			name:     "timestamp at start",
+			start:    100,
+			end:      200,
+			ts:       100,
+			expected: 0,
+		},
+		{
+			name:     "timestamp at end",
+			start:    100,
+			end:      200,
+			ts:       200,
+			expected: 49, // Should be last bucket (50 buckets, 0-49)
+		},
+		{
+			name:     "timestamp in middle",
+			start:    100,
+			end:      200,
+			ts:       150,
+			expected: 25, // Middle bucket
+		},
+		{
+			name:     "timestamp just past middle",
+			start:    100,
+			end:      200,
+			ts:       151,
+			expected: 25, // Should be in the same bucket as 150
+		},
+		{
+			name:     "start equals end",
+			start:    100,
+			end:      100,
+			ts:       100,
+			expected: 0, // Should be the first and only bucket
+		},
+		{
+			name:     "timestamp almost at end",
+			start:    100,
+			end:      200,
+			ts:       199,
+			expected: 49, // Should be in the last bucket
+		},
+		{
+			name:     "large range",
+			start:    0,
+			end:      1000000,
+			ts:       500000,
+			expected: 25, // Middle bucket
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			bs := newBucketSet(maxExemplars, tc.start*uint64(time.Second.Nanoseconds()), tc.end*uint64(time.Second.Nanoseconds())) //nolint: gosec // G115
+			actual := bs.bucket(tc.ts * uint64(time.Second.Milliseconds()))                                                        //nolint: gosec // G115
+			assert.Equal(t, tc.expected, actual)
+		})
+	}
+}
+
 func TestBucketSet(t *testing.T) {
-	buckets := 50
-	s := newBucketSet(buckets)
+	s := newBucketSet(maxExemplars, uint64(100*time.Second.Nanoseconds()), uint64(199*time.Second.Nanoseconds())) //nolint: gosec // G115
 
 	// Add two to each bucket
-	for i := 0; i < buckets; i++ {
-		require.False(t, s.addAndTest(i))
-		require.False(t, s.addAndTest(i))
+	for ts := uint64(100); ts <= 199; ts += 2 { // 100 in total
+		tsMilli := uint64(int64(ts) * time.Second.Milliseconds()) //nolint: gosec // G115
+
+		assert.False(t, s.addAndTest(tsMilli), "ts=%d should be added to bucket", ts)
+		assert.False(t, s.addAndTest(tsMilli), "ts=%d should be added to bucket", ts)
 	}
-	require.Equal(t, buckets*2, s.len())
+	assert.Equal(t, 50*2, s.len())
 
 	// Should be full and reject new adds
-	require.True(t, s.testTotal())
-	require.True(t, s.addAndTest(0))
+	assert.True(t, s.testTotal())
+	for ts := uint64(100); ts <= 199; ts += 2 {
+		tsMilli := uint64(int64(ts) * time.Second.Milliseconds()) //nolint: gosec // G115
+		assert.True(t, s.addAndTest(tsMilli), "ts=%d should be added to bucket", ts)
+		assert.True(t, s.addAndTest(tsMilli), "ts=%d should be added to bucket", ts)
+	}
+	assert.True(t, s.addAndTest(0))
+}
+
+func TestBucketSetSingleExemplar(t *testing.T) {
+	s := newBucketSet(1, uint64(100*time.Second.Nanoseconds()), uint64(199*time.Second.Nanoseconds())) //nolint: gosec // G115
+	tsMilli := uint64(100 * time.Second.Milliseconds())                                                //nolint: gosec // G115
+	assert.False(t, s.addAndTest(tsMilli), "ts=%d should be added to bucket", 100)
+	assert.True(t, s.addAndTest(tsMilli), "ts=%d should not be added to bucket", 100)
 }

pkg/util/hash.go

@@ -4,7 +4,9 @@ import (
 	"hash/fnv"
 )
 
-// TokenFor generates a token used for finding ingesters from ring
+// TokenFor generates a token used for finding ingesters from ring.
+// Not suitable for in-memory hashing or deduping because it is only 32-bit.
+// The collision rate is about 1 in 8000.
 func TokenFor(userID string, b []byte) uint32 {
 	h := fnv.New32()
 	_, _ = h.Write([]byte(userID))
@@ -12,9 +14,17 @@ func TokenFor(userID string, b []byte) uint32 {
 	return h.Sum32()
 }
 
-// TokenForTraceID generates a hashed value for a trace id
+// TokenForTraceID generates a hashed value for a trace id.  Used for bloom lookups.
+// Do not change because it will break lookups on existing bloom filters.
 func TokenForTraceID(b []byte) uint32 {
 	h := fnv.New32()
 	_, _ = h.Write(b)
 	return h.Sum32()
 }
+
+// HashForTraceID generates a generic hash for the trace ID, suitable for mapping and deduping.
+func HashForTraceID(tid []byte) uint64 {
+	h := fnv.New64()
+	_, _ = h.Write(tid)
+	return h.Sum64()
+}

pkg/util/hash_test.go

@@ -0,0 +1,62 @@
+package util
+
+import (
+	"bytes"
+	"crypto/rand"
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestHashForNoCollisions(t *testing.T) {
+	// Verify cases of known collisions under TokenFor don't collide in HashForTraceID.
+	pairs := [][2]string{
+		{"fd5980503add11f09f80f77608c1b2da", "091ea7803ade11f0998a055186ee1243"},
+		{"9e0d446036dc11f09ac04988d2097052", "a61ed97036dc11f0883771db3b51b1ec"},
+		{"6b27f5501eda11f09e99db1b2c23c542", "6b4149b01eda11f0b0e2a966cf7ebbc8"},
+		{"3e9582202f9a11f0afb01b7c06024bd6", "370db6802f9a11f0a9a212dff3125239"},
+		{"978d70802a7311f0991f350653ef0ab4", "9b66da202a7311f09d292db17ccfd31a"},
+		{"de567f703bb711f0b8c377682d1667e6", "dc2d0fc03bb711f091de732fcf93048c"},
+	}
+	for _, pair := range pairs {
+		b1, _ := HexStringToTraceID(pair[0])
+		b2, _ := HexStringToTraceID(pair[1])
+
+		t1 := HashForTraceID(b1)
+		t2 := HashForTraceID(b2)
+
+		require.NotEqual(t, t1, t2)
+	}
+}
+
+// Verify HashForTraceID doesn't collide within reasonable numbers, and estimate the hash collision rate if it does.
+func TestHashForCollisionRate(t *testing.T) {
+	var (
+		n      = 1_000_000
+		tokens = map[uint64]struct{}{}
+		IDs    = make([][]byte, 0, n)
+	)
+
+	for i := 0; i < n; i++ {
+		traceID := make([]byte, 16)
+		_, err := rand.Read(traceID)
+		require.NoError(t, err)
+
+		IDs = append(IDs, traceID)
+		tokens[HashForTraceID(traceID)] = struct{}{}
+	}
+
+	// Ensure no duplicate span IDs accidentally generated
+	sort.Slice(IDs, func(i, j int) bool {
+		return bytes.Compare(IDs[i], IDs[j]) == -1
+	})
+	for i := 1; i < len(IDs); i++ {
+		if bytes.Equal(IDs[i-1], IDs[i]) {
+			panic("same trace ID was generated, oops")
+		}
+	}
+
+	missing := n - len(tokens)
+	require.Zerof(t, missing, "missing 1 out of every %.2f trace ids", float32(n)/float32(missing))
+}

tempodb/encoding/vparquet3/block_traceql.go

@@ -1660,9 +1660,11 @@ func createSpanIterator(makeIter makeIterFn, primaryIter parquetquery.Iterator,
 			if entry.scope != intrinsicScopeSpan {
 				continue
 			}
-			// These intrinsics aren't included in select all because I say so.
+			// These intrinsics aren't included in select all because they
+			// aren't useful for compare().
 			switch intrins {
 			case traceql.IntrinsicSpanID,
+				traceql.IntrinsicParentID,
 				traceql.IntrinsicSpanStartTime,
 				traceql.IntrinsicStructuralDescendant,
 				traceql.IntrinsicStructuralChild,
@@ -2132,13 +2134,22 @@ func createAttributeIterator(makeIter makeIterFn, conditions []traceql.Condition
 ) (parquetquery.Iterator, error) {
 	if selectAll {
 		// Select all with no filtering
+		// Levels such as resource/instrumentation/span may have no attributes. When that
+		// occurs the columns are encoded as single null values, and the current attribute
+		// collector reads them as Nils.  We could skip them in the attribute collector,
+		// but this is more performant because it's at the lowest level.
+		// Alternatively, JoinIterators don't pay attention to -1 (undefined) when checking
+		// the definition level matches.  Fixing that would also work but would need wider testing first.
+		skipNils := &parquetquery.SkipNilsPredicate{}
 		return parquetquery.NewLeftJoinIterator(definitionLevel,
-			[]parquetquery.Iterator{makeIter(keyPath, nil, "key")},
 			[]parquetquery.Iterator{
-				makeIter(strPath, nil, "string"),
-				makeIter(intPath, nil, "int"),
-				makeIter(floatPath, nil, "float"),
-				makeIter(boolPath, nil, "bool"),
+				makeIter(keyPath, skipNils, "key"),
+			},
+			[]parquetquery.Iterator{
+				makeIter(strPath, skipNils, "string"),
+				makeIter(intPath, skipNils, "int"),
+				makeIter(floatPath, skipNils, "float"),
+				makeIter(boolPath, skipNils, "bool"),
 			},
 			&attributeCollector{},
 			parquetquery.WithPool(pqAttrPool))

tempodb/encoding/vparquet4/block_traceql.go

@@ -2036,9 +2036,11 @@ func createSpanIterator(makeIter makeIterFn, innerIterators []parquetquery.Itera
 			if entry.scope != intrinsicScopeSpan {
 				continue
 			}
-			// These intrinsics aren't included in select all because I say so.
+			// These intrinsics aren't included in select all because they
+			// aren't useful for compare().
 			switch intrins {
 			case traceql.IntrinsicSpanID,
+				traceql.IntrinsicParentID,
 				traceql.IntrinsicSpanStartTime,
 				traceql.IntrinsicStructuralDescendant,
 				traceql.IntrinsicStructuralChild,
@@ -2605,13 +2607,23 @@ func createAttributeIterator(makeIter makeIterFn, conditions []traceql.Condition
 	allConditions bool, selectAll bool,
 ) (parquetquery.Iterator, error) {
 	if selectAll {
+		// Select all with no filtering
+		// Levels such as resource/instrumentation/span may have no attributes. When that
+		// occurs the columns are encoded as single null values, and the current attribute
+		// collector reads them as Nils.  We could skip them in the attribute collector,
+		// but this is more performant because it's at the lowest level.
+		// Alternatively, JoinIterators don't pay attention to -1 (undefined) when checking
+		// the definition level matches.  Fixing that would also work but would need wider testing first.
+		skipNils := &parquetquery.SkipNilsPredicate{}
 		return parquetquery.NewLeftJoinIterator(definitionLevel,
-			[]parquetquery.Iterator{makeIter(keyPath, nil, "key")},
 			[]parquetquery.Iterator{
-				makeIter(strPath, nil, "string"),
-				makeIter(intPath, nil, "int"),
-				makeIter(floatPath, nil, "float"),
-				makeIter(boolPath, nil, "bool"),
+				makeIter(keyPath, skipNils, "key"),
+			},
+			[]parquetquery.Iterator{
+				makeIter(strPath, skipNils, "string"),
+				makeIter(intPath, skipNils, "int"),
+				makeIter(floatPath, skipNils, "float"),
+				makeIter(boolPath, skipNils, "bool"),
 			},
 			&attributeCollector{},
 			parquetquery.WithPool(pqAttrPool))
@@ -3156,6 +3168,7 @@ func (c *attributeCollector) KeepGroup(res *parquetquery.IteratorResult) bool {
 		switch e.Key {
 		case "key":
 			key = unsafeToString(e.Value.Bytes())
+
 		case "string":
 			c.strBuffer = append(c.strBuffer, unsafeToString(e.Value.Bytes()))
 		case "int":

tempodb/encoding/vparquet4/block_traceql_test.go

@@ -730,9 +730,9 @@ func TestBackendBlockSelectAll(t *testing.T) {
 
 	b := makeBackendBlockWithTraces(t, traces)
 
-	_, _, _, _, req, err := traceql.Compile("{}")
+	_, eval, _, _, req, err := traceql.Compile("{}")
 	require.NoError(t, err)
-	req.SecondPass = func(inSS *traceql.Spanset) ([]*traceql.Spanset, error) { return []*traceql.Spanset{inSS}, nil }
+	req.SecondPass = func(inSS *traceql.Spanset) ([]*traceql.Spanset, error) { return eval([]*traceql.Spanset{inSS}) }
 	req.SecondPassSelectAll = true
 
 	resp, err := b.Fetch(ctx, *req, common.DefaultSearchOptions())
@@ -887,7 +887,6 @@ func flattenForSelectAll(tr *Trace, dcm dedicatedColumnMapping) *traceql.Spanset
 				newS.addSpanAttr(traceql.IntrinsicNameAttribute, traceql.NewStaticString(s.Name))
 				newS.addSpanAttr(traceql.IntrinsicStatusAttribute, traceql.NewStaticStatus(otlpStatusToTraceqlStatus(uint64(s.StatusCode))))
 				newS.addSpanAttr(traceql.IntrinsicStatusMessageAttribute, traceql.NewStaticString(s.StatusMessage))
-				newS.addSpanAttr(traceql.IntrinsicParentIDAttribute, traceql.NewStaticString(util.SpanIDToHexString(s.ParentSpanID)))
 
 				if s.HttpStatusCode != nil {
 					newS.addSpanAttr(traceql.NewScopedAttribute(traceql.AttributeScopeSpan, false, LabelHTTPStatusCode), traceql.NewStaticInt(int(*s.HttpStatusCode)))
@@ -1115,6 +1114,7 @@ func BenchmarkBackendBlockQueryRange(b *testing.B) {
 		"{} | max_over_time(duration) by (span.http.status_code)",
 		"{} | min_over_time(duration) by (span.http.status_code)",
 		"{ name != nil } | compare({status=error})",
+		"{} > {} | rate() by (name)", // structural
 	}
 
 	e := traceql.NewEngine()