Add support for topk and bottomk functions for TraceQL metrics (#4646)

Add support for topk and bottomk functions on top of TraceQL Metrics. you can now write TraceQL Queries like {} | rate() by (span.client_ip) | topk(10) or {} | rate() by (span.client_ip) | bottomk(10), and only get top or bottomk series from the underlaying TraceQL Metrics Queries. topk and bottomk behaves like Prometheus topk and bottomk functions.
2025-04-10 23:08:24 +05:30
parent 33829c9cef
commit ef2cee7420
25 changed files with 2292 additions and 1241 deletions
@@ -1,4 +1,6 @@
 ## main / unreleased
+
+* [FEATURE] Add support for topk and bottomk functions for TraceQL metrics [#4646](https://github.com/grafana/tempo/pull/4646/) @electron0zero
 * [CHANGE] Update query range error message [#4929](https://github.com/grafana/tempo/pull/4929) (@joey-grafana)
 * [CHANGE] **BREAKING CHANGE** Upgrade OTEL Collector to v0.122.1 [#4893](https://github.com/grafana/tempo/pull/4893) (@javiermolinar)
  The `name` dimension from `tempo_receiver_accepted_span` and `tempo_receiver_refused_spans` changes from `tempo/jaeger_receiver` to `jaeger/jaeger_receiver`
@@ -26,7 +26,8 @@ refs:
 TraceQL metrics is an experimental feature in Grafana Tempo that creates metrics from traces.

 Metric queries extend trace queries by applying a function to trace query results.
-This powerful feature allows for ad hoc aggregation of any existing TraceQL query by any dimension available in your traces, much in the same way that LogQL metric queries create metrics from logs.
+This powerful feature allows for ad hoc aggregation of any existing TraceQL query by any dimension available in your
+traces, much in the same way that LogQL metric queries create metrics from logs.

 Traces are a unique observability signal that contain causal relationships between the components in your system.

@@ -38,7 +39,8 @@ TraceQL metrics can help answer questions like this:

 TraceQL metrics can help you answer these questions by parsing your traces in aggregate.

-TraceQL metrics are powered by the [TraceQL metrics API](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics).
+TraceQL metrics are powered by
+the [TraceQL metrics API](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics).

 ![Metrics visualization in Grafana](/media/docs/tempo/metrics-explore-sample-2.4.png)

@@ -50,16 +52,19 @@ RED is an acronym for three types of metrics:
 - Errors, the number of those requests that are failing
 - Duration, the amount of time those requests take

-For more information about the RED method, refer to [The RED Method: how to instrument your services](/blog/2018/08/02/the-red-method-how-to-instrument-your-services/).
+For more information about the RED method, refer
+to [The RED Method: how to instrument your services](/blog/2018/08/02/the-red-method-how-to-instrument-your-services/).

 You can write TraceQL metrics queries to compute rate, errors, and durations over different groups of spans.

-For more information on how to use TraceQL metrics to investigate issues, refer to [Solve problems with metrics queries](ref:solve-problems-mq).
+For more information on how to use TraceQL metrics to investigate issues, refer
+to [Solve problems with metrics queries](ref:solve-problems-mq).

 ## Enable and use TraceQL metrics

 To use TraceQL metrics, you need to enable them on your Tempo database.
-Refer to [Configure TraceQL metrics](https://grafana.com/docs/tempo/<TEMPO_VERSION>/operations/traceql-metrics/) for more information.
+Refer to [Configure TraceQL metrics](https://grafana.com/docs/tempo/<TEMPO_VERSION>/operations/traceql-metrics/) for
+more information.

 From there, you can either query the TraceQL metrics API directly (for example, with `curl`) or using Grafana
 (recommended).
@@ -71,9 +76,13 @@ Refer to [Solve problems using metrics queries](ref:solve-problems-mq) for some

 ### Functions

-TraceQL metrics queries currently include the following functions for aggregating over groups of spans: `rate`, `count_over_time`, `sum_over_time`, `max_over_time`, `min_over_time`, `avg_over_time`, `quantile_over_time`, `histogram_over_time`, and `compare`.
+TraceQL metrics queries currently include the following functions for aggregating over groups of spans: `rate`,
+`count_over_time`, `sum_over_time`, `max_over_time`, `min_over_time`, `avg_over_time`, `quantile_over_time`,
+`histogram_over_time`, and `compare`.
 These functions can be added as an operator at the end of any TraceQL query.

+The `topk` and `bottomk` functions are supported on TraceQL metrics functions.
+
 For detailed information and example queries for each function, refer to [TraceQL metrics functions](ref:mq-functions).

 ### Exemplars
@@ -83,7 +92,8 @@ They allow you to see an exact trace that contributed to a given metric value.
 This is particularly useful when you want to understand why a given metric is high or low.

 Exemplars are available in TraceQL metrics for all range queries.
-To get exemplars, you need to configure it in the query-frontend with the parameter `query_frontend.metrics.max_exemplars`,
+To get exemplars, you need to configure it in the query-frontend with the parameter
+`query_frontend.metrics.max_exemplars`,
 or pass a query hint in your query.

 Example:
@@ -12,7 +12,8 @@ keywords:

 <!-- If you add a new function to this page, make sure you also add it to the _index.md#functions section.-->

-TraceQL supports `rate`, `count_over_time`, `sum_over_time`, `min_over_time`, `avg_over_time`, `quantile_over_time`, `histogram_over_time`, and `compare` functions.
+TraceQL supports `rate`, `count_over_time`, `sum_over_time`, `min_over_time`, `avg_over_time`, `quantile_over_time`,
+`histogram_over_time`, and `compare` functions.

 ## Available functions

@@ -22,19 +23,24 @@ These functions can be added as an operator at the end of any TraceQL query.
 : Calculates the number of matching spans per second.

 `count_over_time`
-: Counts the number of matching spans per time interval (refer to the [`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs)).
+: Counts the number of matching spans per time interval (refer to the [
+`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs)).

 `sum_over_time`
-: Sums the value for the specified attribute across all matching spans per time interval (refer to the [`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs)).
+: Sums the value for the specified attribute across all matching spans per time interval (refer to the [
+`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs)).

 `min_over_time`
-: Returns the minimum value for the specified attribute across all matching spans per time interval (refer to the [`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics)).
+: Returns the minimum value for the specified attribute across all matching spans per time interval (refer to the [
+`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics)).

 `max_over_time`
-: Returns the maximum value for the specified attribute across all matching spans per time interval (refer to the [`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics)).
+: Returns the maximum value for the specified attribute across all matching spans per time interval (refer to the [
+`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics)).

 `avg_over_time`
-: Returns the average value for the specified attribute across all matching spans per time interval (refer to the [`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics)).
+: Returns the average value for the specified attribute across all matching spans per time interval (refer to the [
+`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics)).

 `quantile_over_time`
 : The quantile of the values in the specified interval.
@@ -43,7 +49,8 @@ These functions can be added as an operator at the end of any TraceQL query.
 : Evaluate frequency distribution over time. Example: `histogram_over_time(duration) by (span.foo)`.

 `compare`
-: Used to split the stream of spans into two groups: a selection and a baseline. The function returns time-series for all attributes found on the spans to highlight the differences between the two groups.
+: Used to split the stream of spans into two groups: a selection and a baseline. The function returns time-series for
+all attributes found on the spans to highlight the differences between the two groups.

 ## The `rate` function

@@ -56,7 +63,8 @@ None.
 ## Examples

 The following query shows the rate of errors by service and span name.
-This is a TraceQL specific way of gathering rate metrics that would otherwise be generated by the span metrics processor.
+This is a TraceQL specific way of gathering rate metrics that would otherwise be generated by the span metrics
+processor.

 For example, this query:

@@ -88,34 +96,39 @@ spans than `/api/happy`, for example.

 The `count_over_time()` function counts the number of matching spans per time interval.
 The time interval that the count will be computed over is set by the `step` parameter.
-For more information, refer to the [`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics).
+For more information, refer to the [
+`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics).

 ### Example

-This example counts the number of spans with name `"GET /:endpoint"` broken down by status code. You might see that there are 10 `"GET /:endpoint"` spans with status code 200 and 15 `"GET /:endpoint"` spans with status code 400.
+This example counts the number of spans with name `"GET /:endpoint"` broken down by status code. You might see that
+there are 10 `"GET /:endpoint"` spans with status code 200 and 15 `"GET /:endpoint"` spans with status code 400.

 ```
 { name = "GET /:endpoint" } | count_over_time() by (span.http.status_code)

 ```

-
 ## The `sum_over_time`, `min_over_time`, `max_over_time`, and `avg_over_time`  functions

 The `sum_over_time()` lets you aggregate numerical values by computing the sum value of them.
 The time interval that the sum is computed over is set by the `step` parameter.

 The `min_over_time()` function lets you aggregate numerical attributes by calculating their minimum value.
-For example, you could choose to calculate the minimum duration of a group of spans, or you could choose to calculate the minimum value of a custom attribute you've attached to your spans, like `span.shopping.cart.entries`.
+For example, you could choose to calculate the minimum duration of a group of spans, or you could choose to calculate
+the minimum value of a custom attribute you've attached to your spans, like `span.shopping.cart.entries`.
 The time interval that the minimum is computed over is set by the `step` parameter.

-The `max_over_time()` lets you aggregate numerical values by computing the maximum value of them, such as the all important span duration.
+The `max_over_time()` lets you aggregate numerical values by computing the maximum value of them, such as the all
+important span duration.
 The time interval that the maximum is computed over is set by the `step` parameter.

-The `avg_over_time()` function lets you aggregate numerical values by computing the maximum value of them, such as the all important span duration.
+The `avg_over_time()` function lets you aggregate numerical values by computing the maximum value of them, such as the
+all important span duration.
 The time interval that the maximum is computer over is set by the `step` parameter.

-For more information, refer to the [`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics).
+For more information, refer to the [
+`step` API parameter](https://grafana.com/docs/tempo/<TEMPO_VERSION>/api_docs/#traceql-metrics).

 ### Parameters

@@ -158,10 +171,12 @@ This example computes the average duration for each `http.status_code` of all sp

 ## The `quantile_over_time` and `histogram_over_time` functions

-The `quantile_over_time()` and `histogram_over_time()` functions let you aggregate numerical values, such as the all important span duration.
+The `quantile_over_time()` and `histogram_over_time()` functions let you aggregate numerical values, such as the all
+important span duration.
 You can specify multiple quantiles in the same query.

-The example below computes the 99th, 90th, and 50th percentile of the duration attribute on all spans with name  `GET /:endpoint`.
+The example below computes the 99th, 90th, and 50th percentile of the duration attribute on all spans with name
+`GET /:endpoint`.

 ```
 { name = "GET /:endpoint" } | quantile_over_time(duration, .99, .9, .5)
@@ -181,21 +196,50 @@ To demonstrate this flexibility, consider this nonsensical quantile on `span.htt
 { name = "GET /:endpoint" } | quantile_over_time(span.http.status_code, .99, .9, .5)
 ```

-This computes the 99th, 90th, and 50th percentile of the values of the `status_code` attribute for all spans named `GET /:endpoint`.
+This computes the 99th, 90th, and 50th percentile of the values of the `status_code` attribute for all spans named
+`GET /:endpoint`.
 This is unlikely to tell you anything useful (what does a median status code of `347` mean?), but it works.

 As a further example, imagine a custom attribute like `span.temperature`.
-You could use a similar query to know what the 50th percentile and 95th percentile temperatures were across all your spans.
+You could use a similar query to know what the 50th percentile and 95th percentile temperatures were across all your
+spans.
+
+## `topk` and `bottomk` functions
+
+TraceQL supports the `topk` and `bottomk` functions that let you aggregate and process TraceQL metrics.
+These functions are similar to their equivalent PromQL functions. For example:
+
+- `{ } | rate() by(resource.service.name) | bottomk(5)`
+- `{ } | rate() by(resource.service.name) | topk(5)`
+
+When a query response is larger than the maximum, you can use these functions to return only the specified number
+from 1 through `k` of the number of the top or bottom results.
+
+For example: `{ resource.service.name = "foo" } | rate() by (span.http.url)  | topk(10)`
+The first part, `{ resource.service.name = "foo" }`, takes all spans in the service `foo` The spans
+are rated by the URL, for example, the most active endpoints on a service.
+
+Adding `topk(10)` returns the top 10 most common instead of the entire list.
+Conversely, you can use `bottomk(10)` to see the least most used ones.
+
+In TraceQL, `topk` and `bottomk` work similar to how they function in PromQL. Both `topk` and `bottomk`
+are evaluated at each data point.
+
+If you do a `topk` of 10, you might get a 20 series. For example, on this data point, the top 10 are `A` through `J`.
+
+On the next data point, `A` through `I` might still be the top 9, but `J` might have fallen off for `K`.
+Because it's evaluated at each data point, you'll get the top series for each data point.

 ## The `compare` function

 The `compare` function is used to split a set of spans into two groups: a selection and a baseline.
 It returns time-series for all attributes found on the spans to highlight the differences between the two groups.

-This is a powerful function that's best understood by using the [**Comparison** tab in Traces Drilldown](https://grafana.com/docs/grafana/<GRAFANA_VERSION>/explore/simplified-exploration/traces/investigate/#comparison).
+This powerful function is best understood by using the [**Comparison** tab in Traces Drilldown](https://grafana.com/docs/grafana/<GRAFANA_VERSION>/explore/simplified-exploration/traces/investigate/#comparison).
 You can also under this function by looking at example outputs below.

-The function is used like other metrics functions: when it's placed after any trace query, it converts the query into a metrics query:
+The function is used like other metrics functions: when it's placed after any trace query, it converts the query into a
+metrics query:
 `...any spanset pipeline... | compare({subset filters}, <topN>, <start timestamp>, <end timestamp>)`

 Example:
@@ -206,18 +250,26 @@ Example:

 This function is generally run as an instant query.
 An instant query gives a single value at the end of the selected time range.
-[Instant queries](https://prometheus.io/docs/prometheus/latest/querying/api/#instant-queries) are quicker to execute and it often easier to understand their results
+[Instant queries](https://prometheus.io/docs/prometheus/latest/querying/api/#instant-queries) are quicker to execute and
+it often easier to understand their results
 The returns may exceed gRPC payloads when run as a range query.

 ### Parameters

 The `compare` function has four parameters:

-1. Required. The first parameter is a spanset filter for choosing the subset of spans. This filter is executed against the incoming spans. If it matches, then the span is considered to be part of the selection. Otherwise, it is part of the baseline.  Common filters are expected to be things like `{status=error}` (what is different about errors?) or `{duration>1s}` (what is different about slow spans?)
+1. Required. The first parameter is a spanset filter for choosing the subset of spans. This filter is executed against
+   the incoming spans. If it matches, then the span is considered to be part of the selection. Otherwise, it is part of
+   the baseline. Common filters are expected to be things like `{status=error}` (what is different about errors?) or
+   `{duration>1s}` (what is different about slow spans?)

-2. Optional. The second parameter is the top `N` values to return per attribute. If an attribute exceeds this limit in either the selection group or baseline group, then only the top `N` values (based on frequency) are returned, and an error indicator for the attribute is included output (see below).  Defaults to `10`.
+2. Optional. The second parameter is the top `N` values to return per attribute. If an attribute exceeds this limit in
+   either the selection group or baseline group, then only the top `N` values (based on frequency) are returned, and an
+   error indicator for the attribute is included output (see below). Defaults to `10`.

-3. Optional. Start and End timestamps in Unix nanoseconds, which can be used to constrain the selection window by time, in addition to the filter. For example, the overall query could cover the past hour, and the selection window only a 5 minute time period in which there was an anomaly. These timestamps must both be given, or neither.
+3. Optional. Start and End timestamps in Unix nanoseconds, which can be used to constrain the selection window by time,
+   in addition to the filter. For example, the overall query could cover the past hour, and the selection window only a
+   5 minute time period in which there was an anomaly. These timestamps must both be given, or neither.

 ### Output

@@ -226,6 +278,7 @@ The outputs are flat time-series for each attribute/value found in the spans.
 Each series has a label `__meta_type` which denotes which group it is in, either `selection` or `baseline`.

 Example output series:
+
 ```
 { __meta_type="baseline", resource.cluster="prod" } 123
 { __meta_type="baseline", resource.cluster="qa" } 124
@@ -236,6 +289,7 @@ Example output series:

 When an attribute reaches the topN limit, there will also be present an error indicator.
 This example means the attribute `resource.cluster` had too many values.
+
 ```
 { __meta_error="__too_many_values__", resource.cluster=<nil> }
 ```
@@ -23,7 +23,7 @@ query_frontend:
  trace_by_id:
    duration_slo: 100ms
  metrics:
-    max_duration: 120h                # maximum duration of a metrics query, increase for local setups
+    max_duration: 200h                # maximum duration of a metrics query, increase for local setups
    query_backend_after: 5m
    duration_slo: 5s
    throughput_bytes_slo: 1.073741824e+09
@@ -56,7 +56,7 @@ ingester:

 compactor:
  compaction:
-    block_retention: 24h                # overall Tempo trace retention. set for demo purposes
+    block_retention: 720h                # overall Tempo trace retention. set for demo purposes

 metrics_generator:
  registry:
@@ -74,7 +74,7 @@ func (s queryRangeSharder) RoundTrip(pipelineRequest pipeline.Request) (pipeline
 		return pipeline.NewBadRequest(err), nil
 	}

-	expr, _, _, _, err := traceql.Compile(req.Query)
+	expr, _, _, _, _, err := traceql.Compile(req.Query)
 	if err != nil {
 		return pipeline.NewBadRequest(err), nil
 	}
@@ -102,7 +102,7 @@ func (c weightRequestWare) setTraceQLWeight(req Request) {
 		return
 	}

-	_, _, _, spanRequest, err := traceql.Compile(traceQLQuery)
+	_, _, _, _, spanRequest, err := traceql.Compile(traceQLQuery)
 	if err != nil || spanRequest == nil {
 		return
 	}
@@ -3,6 +3,7 @@ package querier
 import (
 	"context"
 	"fmt"
+	"math"
 	"time"

 	"github.com/go-kit/log/level"
@@ -158,6 +159,12 @@ func queryRangeTraceQLToProto(set traceql.SeriesSet, req *tempopb.QueryRangeRequ

 		exemplars := make([]tempopb.Exemplar, 0, len(s.Exemplars))
 		for _, e := range s.Exemplars {
+			// skip exemplars that has NaN value
+			i := traceql.IntervalOfMs(int64(e.TimestampMs), req.Start, req.End, req.Step)
+			if i < 0 || i >= len(s.Values) || math.IsNaN(s.Values[i]) { // strict bounds check
+				continue
+			}
+
 			lbls := make([]v1.KeyValue, 0, len(e.Labels))
 			for _, label := range e.Labels {
 				lbls = append(lbls,
@@ -11,7 +11,6 @@ import (
 	"unsafe"

 	"github.com/grafana/tempo/pkg/regexp"
-	"github.com/grafana/tempo/pkg/tempopb"
 )

 type Element interface {
@@ -19,16 +18,6 @@ type Element interface {
 	validate() error
 }

-type metricsFirstStageElement interface {
-	Element
-	extractConditions(request *FetchSpansRequest)
-	init(req *tempopb.QueryRangeRequest, mode AggregateMode)
-	observe(Span) // TODO - batching?
-	observeExemplar(Span)
-	observeSeries([]*tempopb.TimeSeries) // Re-entrant metrics on the query-frontend.  Using proto version for efficiency
-	result() SeriesSet
-}
-
 type pipelineElement interface {
 	Element
 	extractConditions(request *FetchSpansRequest)
@@ -40,9 +29,10 @@ type typedExpression interface {
 }

 type RootExpr struct {
-	Pipeline        Pipeline
-	MetricsPipeline metricsFirstStageElement
-	Hints           *Hints
+	Pipeline           Pipeline
+	MetricsPipeline    firstStageElement
+	MetricsSecondStage secondStageElement
+	Hints              *Hints
 }

 func newRootExpr(e pipelineElement) *RootExpr {
@@ -56,7 +46,7 @@ func newRootExpr(e pipelineElement) *RootExpr {
 	}
 }

-func newRootExprWithMetrics(e pipelineElement, m metricsFirstStageElement) *RootExpr {
+func newRootExprWithMetrics(e pipelineElement, m firstStageElement) *RootExpr {
 	p, ok := e.(Pipeline)
 	if !ok {
 		p = newPipeline(e)
@@ -68,12 +58,25 @@ func newRootExprWithMetrics(e pipelineElement, m metricsFirstStageElement) *Root
 	}
 }

+func newRootExprWithMetricsTwoStage(e pipelineElement, m1 firstStageElement, m2 secondStageElement) *RootExpr {
+	p, ok := e.(Pipeline)
+	if !ok {
+		p = newPipeline(e)
+	}
+
+	return &RootExpr{
+		Pipeline:           p,
+		MetricsPipeline:    m1,
+		MetricsSecondStage: m2,
+	}
+}
+
 func (r *RootExpr) withHints(h *Hints) *RootExpr {
 	r.Hints = h
 	return r
 }

-// IsNoop detects trival noop queries like {false} which never return
+// IsNoop detects trivial noop queries like {false} which never return
 // results and can be used to exit early.
 func (r *RootExpr) IsNoop() bool {
 	isNoopFilter := func(x any) bool {
@@ -1112,281 +1115,3 @@ var (
 	_ pipelineElement = (*ScalarFilter)(nil)
 	_ pipelineElement = (*GroupOperation)(nil)
 )
-
-type getExemplar func(Span) (float64, uint64)
-
-// MetricsAggregate is a placeholder in the AST for a metrics aggregation
-// pipeline element. It has a superset of the properties of them all, and
-// builds them later via init() so that appropriate buffers can be allocated
-// for the query time range and step, and different implementations for
-// shardable and unshardable pipelines.
-type MetricsAggregate struct {
-	op         MetricsAggregateOp
-	by         []Attribute
-	attr       Attribute
-	floats     []float64
-	agg        SpanAggregator
-	seriesAgg  SeriesAggregator
-	exemplarFn getExemplar
-	// Type of operation for simple aggregatation in layers 2 and 3
-	simpleAggregationOp SimpleAggregationOp
-}
-
-func newMetricsAggregate(agg MetricsAggregateOp, by []Attribute) *MetricsAggregate {
-	return &MetricsAggregate{
-		op: agg,
-		by: by,
-	}
-}
-
-func newMetricsAggregateWithAttr(agg MetricsAggregateOp, attr Attribute, by []Attribute) *MetricsAggregate {
-	return &MetricsAggregate{
-		op:   agg,
-		attr: attr,
-		by:   by,
-	}
-}
-
-func newMetricsAggregateQuantileOverTime(attr Attribute, qs []float64, by []Attribute) *MetricsAggregate {
-	return &MetricsAggregate{
-		op:     metricsAggregateQuantileOverTime,
-		floats: qs,
-		attr:   attr,
-		by:     by,
-	}
-}
-
-func (a *MetricsAggregate) extractConditions(request *FetchSpansRequest) {
-	// For metrics aggregators based on a span attribute we have to include it
-	includeAttribute := a.attr != (Attribute{}) && !request.HasAttribute(a.attr)
-	if includeAttribute {
-		request.SecondPassConditions = append(request.SecondPassConditions, Condition{
-			Attribute: a.attr,
-		})
-	}
-
-	for _, b := range a.by {
-		if !request.HasAttribute(b) {
-			request.SecondPassConditions = append(request.SecondPassConditions, Condition{
-				Attribute: b,
-			})
-		}
-	}
-}
-
-func (a *MetricsAggregate) init(q *tempopb.QueryRangeRequest, mode AggregateMode) {
-	// Raw mode:
-
-	var innerAgg func() VectorAggregator
-	var byFunc func(Span) (Static, bool)
-	var byFuncLabel string
-
-	switch a.op {
-	case metricsAggregateCountOverTime:
-		innerAgg = func() VectorAggregator { return NewCountOverTimeAggregator() }
-		a.simpleAggregationOp = sumAggregation
-		a.exemplarFn = exemplarNaN
-
-	case metricsAggregateMinOverTime:
-		innerAgg = func() VectorAggregator { return NewOverTimeAggregator(a.attr, minOverTimeAggregation) }
-		a.simpleAggregationOp = minOverTimeAggregation
-		a.exemplarFn = exemplarFnFor(a.attr)
-
-	case metricsAggregateMaxOverTime:
-		innerAgg = func() VectorAggregator { return NewOverTimeAggregator(a.attr, maxOverTimeAggregation) }
-		a.simpleAggregationOp = maxOverTimeAggregation
-		a.exemplarFn = exemplarFnFor(a.attr)
-
-	case metricsAggregateSumOverTime:
-		innerAgg = func() VectorAggregator { return NewOverTimeAggregator(a.attr, sumOverTimeAggregation) }
-		a.simpleAggregationOp = sumOverTimeAggregation
-		a.exemplarFn = exemplarFnFor(a.attr)
-
-	case metricsAggregateRate:
-		innerAgg = func() VectorAggregator { return NewRateAggregator(1.0 / time.Duration(q.Step).Seconds()) }
-		a.simpleAggregationOp = sumAggregation
-		a.exemplarFn = exemplarNaN
-
-	case metricsAggregateHistogramOverTime:
-		innerAgg = func() VectorAggregator { return NewCountOverTimeAggregator() }
-		byFunc = bucketizeFnFor(a.attr)
-		byFuncLabel = internalLabelBucket
-		a.simpleAggregationOp = sumAggregation
-		a.exemplarFn = exemplarNaN // Histogram final series are counts so exemplars are placeholders
-
-	case metricsAggregateQuantileOverTime:
-		innerAgg = func() VectorAggregator { return NewCountOverTimeAggregator() }
-		byFunc = bucketizeFnFor(a.attr)
-		byFuncLabel = internalLabelBucket
-		a.simpleAggregationOp = sumAggregation
-		a.exemplarFn = exemplarFnFor(a.attr)
-	}
-
-	switch mode {
-	case AggregateModeSum:
-		a.initSum(q)
-		return
-
-	case AggregateModeFinal:
-		a.initFinal(q)
-		return
-	}
-
-	a.agg = NewGroupingAggregator(a.op.String(), func() RangeAggregator {
-		return NewStepAggregator(q.Start, q.End, q.Step, innerAgg)
-	}, a.by, byFunc, byFuncLabel)
-}
-
-func bucketizeFnFor(attr Attribute) func(Span) (Static, bool) {
-	switch attr {
-	case IntrinsicDurationAttribute:
-		// Optimal implementation for duration attribute
-		return bucketizeDuration
-	default:
-		// Basic implementation for all other attributes
-		return bucketizeAttribute(attr)
-	}
-}
-
-func bucketizeDuration(s Span) (Static, bool) {
-	d := s.DurationNanos()
-	if d < 2 {
-		return NewStaticNil(), false
-	}
-	// Bucket is in seconds
-	return NewStaticFloat(Log2Bucketize(d) / float64(time.Second)), true
-}
-
-// exemplarAttribute captures a closure around the attribute so it doesn't have to be passed along with every span.
-// should be more efficient.
-func bucketizeAttribute(a Attribute) func(Span) (Static, bool) {
-	return func(s Span) (Static, bool) {
-		f, t := FloatizeAttribute(s, a)
-
-		switch t {
-		case TypeInt:
-			if f < 2 {
-				return NewStaticNil(), false
-			}
-			// Bucket is the value rounded up to the nearest power of 2
-			return NewStaticFloat(Log2Bucketize(uint64(f))), true
-		case TypeDuration:
-			if f < 2 {
-				return NewStaticNil(), false
-			}
-			// Bucket is log2(nanos) converted to float seconds
-			return NewStaticFloat(Log2Bucketize(uint64(f)) / float64(time.Second)), true
-		default:
-			// TODO(mdisibio) - Add support for floats, we need to map them into buckets.
-			// Because of the range of floats, we need a native histogram approach.
-			return NewStaticNil(), false
-		}
-	}
-}
-
-func exemplarFnFor(a Attribute) func(Span) (float64, uint64) {
-	switch a {
-	case IntrinsicDurationAttribute:
-		return exemplarDuration
-	case Attribute{}:
-		// This records exemplars without a value, and they
-		// are attached to the series at the end.
-		return exemplarNaN
-	default:
-		return exemplarAttribute(a)
-	}
-}
-
-func exemplarNaN(s Span) (float64, uint64) {
-	return math.NaN(), s.StartTimeUnixNanos() / uint64(time.Millisecond)
-}
-
-func exemplarDuration(s Span) (float64, uint64) {
-	v := float64(s.DurationNanos()) / float64(time.Second)
-	t := s.StartTimeUnixNanos() / uint64(time.Millisecond)
-	return v, t
-}
-
-// exemplarAttribute captures a closure around the attribute so it doesn't have to be passed along with every span.
-// should be more efficient.
-func exemplarAttribute(a Attribute) func(Span) (float64, uint64) {
-	return func(s Span) (float64, uint64) {
-		v, _ := FloatizeAttribute(s, a)
-		t := s.StartTimeUnixNanos() / uint64(time.Millisecond)
-		return v, t
-	}
-}
-
-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)
-}
-
-func (a *MetricsAggregate) initFinal(q *tempopb.QueryRangeRequest) {
-	switch a.op {
-	case metricsAggregateQuantileOverTime:
-		a.seriesAgg = NewHistogramAggregator(q, a.floats)
-	default:
-		// These are simple additions by series
-		a.seriesAgg = NewSimpleCombiner(q, a.simpleAggregationOp)
-	}
-}
-
-func (a *MetricsAggregate) observe(span Span) {
-	a.agg.Observe(span)
-}
-
-func (a *MetricsAggregate) observeExemplar(span Span) {
-	v, ts := a.exemplarFn(span)
-	a.agg.ObserveExemplar(span, v, ts)
-}
-
-func (a *MetricsAggregate) observeSeries(ss []*tempopb.TimeSeries) {
-	a.seriesAgg.Combine(ss)
-}
-
-func (a *MetricsAggregate) result() SeriesSet {
-	if a.agg != nil {
-		return a.agg.Series()
-	}
-
-	// In the frontend-version the results come from
-	// the job-level aggregator
-	return a.seriesAgg.Results()
-}
-
-func (a *MetricsAggregate) validate() error {
-	switch a.op {
-	case metricsAggregateCountOverTime:
-	case metricsAggregateMinOverTime:
-	case metricsAggregateMaxOverTime:
-	case metricsAggregateSumOverTime:
-	case metricsAggregateRate:
-	case metricsAggregateHistogramOverTime:
-		if len(a.by) >= maxGroupBys {
-			// We reserve a spot for the bucket so quantile has 1 less group by
-			return newUnsupportedError(fmt.Sprintf("metrics group by %v values", len(a.by)))
-		}
-	case metricsAggregateQuantileOverTime:
-		if len(a.by) >= maxGroupBys {
-			// We reserve a spot for the bucket so quantile has 1 less group by
-			return newUnsupportedError(fmt.Sprintf("metrics group by %v values", len(a.by)))
-		}
-		for _, q := range a.floats {
-			if q < 0 || q > 1 {
-				return fmt.Errorf("quantile must be between 0 and 1: %v", q)
-			}
-		}
-	default:
-		return newUnsupportedError(fmt.Sprintf("metrics aggregate operation (%v)", a.op))
-	}
-
-	if len(a.by) > maxGroupBys {
-		return newUnsupportedError(fmt.Sprintf("metrics group by %v values", len(a.by)))
-	}
-
-	return nil
-}
-
-var _ metricsFirstStageElement = (*MetricsAggregate)(nil)
@@ -0,0 +1,376 @@
+package traceql
+
+import (
+	"fmt"
+	"math"
+	"time"
+
+	"github.com/grafana/tempo/pkg/tempopb"
+)
+
+type firstStageElement interface {
+	Element
+	extractConditions(request *FetchSpansRequest)
+	init(req *tempopb.QueryRangeRequest, mode AggregateMode)
+	observe(Span) // TODO - batching?
+	observeExemplar(Span)
+	observeSeries([]*tempopb.TimeSeries) // Re-entrant metrics on the query-frontend.  Using proto version for efficiency
+	result() SeriesSet
+}
+
+type getExemplar func(Span) (float64, uint64)
+
+// MetricsAggregate is a placeholder in the AST for a metrics aggregation
+// pipeline element. It has a superset of the properties of them all, and
+// builds them later via init() so that appropriate buffers can be allocated
+// for the query time range and step, and different implementations for
+// shardable and unshardable pipelines.
+type MetricsAggregate struct {
+	op         MetricsAggregateOp
+	by         []Attribute
+	attr       Attribute
+	floats     []float64
+	agg        SpanAggregator
+	seriesAgg  SeriesAggregator
+	exemplarFn getExemplar
+	// Type of operation for simple aggregatation in layers 2 and 3
+	simpleAggregationOp SimpleAggregationOp
+}
+
+func newMetricsAggregate(agg MetricsAggregateOp, by []Attribute) *MetricsAggregate {
+	return &MetricsAggregate{
+		op: agg,
+		by: by,
+	}
+}
+
+func newMetricsAggregateWithAttr(agg MetricsAggregateOp, attr Attribute, by []Attribute) *MetricsAggregate {
+	return &MetricsAggregate{
+		op:   agg,
+		attr: attr,
+		by:   by,
+	}
+}
+
+func newMetricsAggregateQuantileOverTime(attr Attribute, qs []float64, by []Attribute) *MetricsAggregate {
+	return &MetricsAggregate{
+		op:     metricsAggregateQuantileOverTime,
+		floats: qs,
+		attr:   attr,
+		by:     by,
+	}
+}
+
+func (a *MetricsAggregate) extractConditions(request *FetchSpansRequest) {
+	// For metrics aggregators based on a span attribute we have to include it
+	includeAttribute := a.attr != (Attribute{}) && !request.HasAttribute(a.attr)
+	if includeAttribute {
+		request.SecondPassConditions = append(request.SecondPassConditions, Condition{
+			Attribute: a.attr,
+		})
+	}
+
+	for _, b := range a.by {
+		if !request.HasAttribute(b) {
+			request.SecondPassConditions = append(request.SecondPassConditions, Condition{
+				Attribute: b,
+			})
+		}
+	}
+}
+
+func (a *MetricsAggregate) init(q *tempopb.QueryRangeRequest, mode AggregateMode) {
+	// Raw mode:
+
+	var innerAgg func() VectorAggregator
+	var byFunc func(Span) (Static, bool)
+	var byFuncLabel string
+
+	switch a.op {
+	case metricsAggregateCountOverTime:
+		innerAgg = func() VectorAggregator { return NewCountOverTimeAggregator() }
+		a.simpleAggregationOp = sumAggregation
+		a.exemplarFn = exemplarNaN
+
+	case metricsAggregateMinOverTime:
+		innerAgg = func() VectorAggregator { return NewOverTimeAggregator(a.attr, minOverTimeAggregation) }
+		a.simpleAggregationOp = minOverTimeAggregation
+		a.exemplarFn = exemplarFnFor(a.attr)
+
+	case metricsAggregateMaxOverTime:
+		innerAgg = func() VectorAggregator { return NewOverTimeAggregator(a.attr, maxOverTimeAggregation) }
+		a.simpleAggregationOp = maxOverTimeAggregation
+		a.exemplarFn = exemplarFnFor(a.attr)
+
+	case metricsAggregateSumOverTime:
+		innerAgg = func() VectorAggregator { return NewOverTimeAggregator(a.attr, sumOverTimeAggregation) }
+		a.simpleAggregationOp = sumOverTimeAggregation
+		a.exemplarFn = exemplarFnFor(a.attr)
+
+	case metricsAggregateRate:
+		innerAgg = func() VectorAggregator { return NewRateAggregator(1.0 / time.Duration(q.Step).Seconds()) }
+		a.simpleAggregationOp = sumAggregation
+		a.exemplarFn = exemplarNaN
+
+	case metricsAggregateHistogramOverTime:
+		innerAgg = func() VectorAggregator { return NewCountOverTimeAggregator() }
+		byFunc = bucketizeFnFor(a.attr)
+		byFuncLabel = internalLabelBucket
+		a.simpleAggregationOp = sumAggregation
+		a.exemplarFn = exemplarNaN // Histogram final series are counts so exemplars are placeholders
+
+	case metricsAggregateQuantileOverTime:
+		innerAgg = func() VectorAggregator { return NewCountOverTimeAggregator() }
+		byFunc = bucketizeFnFor(a.attr)
+		byFuncLabel = internalLabelBucket
+		a.simpleAggregationOp = sumAggregation
+		a.exemplarFn = exemplarFnFor(a.attr)
+	}
+
+	switch mode {
+	case AggregateModeSum:
+		a.initSum(q)
+		return
+
+	case AggregateModeFinal:
+		a.initFinal(q)
+		return
+	}
+
+	a.agg = NewGroupingAggregator(a.op.String(), func() RangeAggregator {
+		return NewStepAggregator(q.Start, q.End, q.Step, innerAgg)
+	}, a.by, byFunc, byFuncLabel)
+}
+
+func bucketizeFnFor(attr Attribute) func(Span) (Static, bool) {
+	switch attr {
+	case IntrinsicDurationAttribute:
+		// Optimal implementation for duration attribute
+		return bucketizeDuration
+	default:
+		// Basic implementation for all other attributes
+		return bucketizeAttribute(attr)
+	}
+}
+
+func bucketizeDuration(s Span) (Static, bool) {
+	d := s.DurationNanos()
+	if d < 2 {
+		return NewStaticNil(), false
+	}
+	// Bucket is in seconds
+	return NewStaticFloat(Log2Bucketize(d) / float64(time.Second)), true
+}
+
+// exemplarAttribute captures a closure around the attribute so it doesn't have to be passed along with every span.
+// should be more efficient.
+func bucketizeAttribute(a Attribute) func(Span) (Static, bool) {
+	return func(s Span) (Static, bool) {
+		f, t := FloatizeAttribute(s, a)
+
+		switch t {
+		case TypeInt:
+			if f < 2 {
+				return NewStaticNil(), false
+			}
+			// Bucket is the value rounded up to the nearest power of 2
+			return NewStaticFloat(Log2Bucketize(uint64(f))), true
+		case TypeDuration:
+			if f < 2 {
+				return NewStaticNil(), false
+			}
+			// Bucket is log2(nanos) converted to float seconds
+			return NewStaticFloat(Log2Bucketize(uint64(f)) / float64(time.Second)), true
+		default:
+			// TODO(mdisibio) - Add support for floats, we need to map them into buckets.
+			// Because of the range of floats, we need a native histogram approach.
+			return NewStaticNil(), false
+		}
+	}
+}
+
+func exemplarFnFor(a Attribute) func(Span) (float64, uint64) {
+	switch a {
+	case IntrinsicDurationAttribute:
+		return exemplarDuration
+	case Attribute{}:
+		// This records exemplars without a value, and they
+		// are attached to the series at the end.
+		return exemplarNaN
+	default:
+		return exemplarAttribute(a)
+	}
+}
+
+func exemplarNaN(s Span) (float64, uint64) {
+	return math.NaN(), s.StartTimeUnixNanos() / uint64(time.Millisecond)
+}
+
+func exemplarDuration(s Span) (float64, uint64) {
+	v := float64(s.DurationNanos()) / float64(time.Second)
+	t := s.StartTimeUnixNanos() / uint64(time.Millisecond)
+	return v, t
+}
+
+// exemplarAttribute captures a closure around the attribute so it doesn't have to be passed along with every span.
+// should be more efficient.
+func exemplarAttribute(a Attribute) func(Span) (float64, uint64) {
+	return func(s Span) (float64, uint64) {
+		v, _ := FloatizeAttribute(s, a)
+		t := s.StartTimeUnixNanos() / uint64(time.Millisecond)
+		return v, t
+	}
+}
+
+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)
+}
+
+func (a *MetricsAggregate) initFinal(q *tempopb.QueryRangeRequest) {
+	switch a.op {
+	case metricsAggregateQuantileOverTime:
+		a.seriesAgg = NewHistogramAggregator(q, a.floats)
+	default:
+		// These are simple additions by series
+		a.seriesAgg = NewSimpleCombiner(q, a.simpleAggregationOp)
+	}
+}
+
+func (a *MetricsAggregate) observe(span Span) {
+	a.agg.Observe(span)
+}
+
+func (a *MetricsAggregate) observeExemplar(span Span) {
+	v, ts := a.exemplarFn(span)
+	a.agg.ObserveExemplar(span, v, ts)
+}
+
+func (a *MetricsAggregate) observeSeries(ss []*tempopb.TimeSeries) {
+	a.seriesAgg.Combine(ss)
+}
+
+func (a *MetricsAggregate) result() SeriesSet {
+	if a.agg != nil {
+		return a.agg.Series()
+	}
+
+	// In the frontend-version the results come from
+	// the job-level aggregator
+	return a.seriesAgg.Results()
+}
+
+func (a *MetricsAggregate) validate() error {
+	switch a.op {
+	case metricsAggregateCountOverTime:
+	case metricsAggregateMinOverTime:
+	case metricsAggregateMaxOverTime:
+	case metricsAggregateSumOverTime:
+	case metricsAggregateRate:
+	case metricsAggregateHistogramOverTime:
+		if len(a.by) >= maxGroupBys {
+			// We reserve a spot for the bucket so quantile has 1 less group by
+			return newUnsupportedError(fmt.Sprintf("metrics group by %v values", len(a.by)))
+		}
+	case metricsAggregateQuantileOverTime:
+		if len(a.by) >= maxGroupBys {
+			// We reserve a spot for the bucket so quantile has 1 less group by
+			return newUnsupportedError(fmt.Sprintf("metrics group by %v values", len(a.by)))
+		}
+		for _, q := range a.floats {
+			if q < 0 || q > 1 {
+				return fmt.Errorf("quantile must be between 0 and 1: %v", q)
+			}
+		}
+	default:
+		return newUnsupportedError(fmt.Sprintf("metrics aggregate operation (%v)", a.op))
+	}
+
+	if len(a.by) > maxGroupBys {
+		return newUnsupportedError(fmt.Sprintf("metrics group by %v values", len(a.by)))
+	}
+
+	return nil
+}
+
+var _ firstStageElement = (*MetricsAggregate)(nil)
+
+// secondStageElement represents operations that are performed
+// in the second stage metrics pipeline, such as topK/bottomK, etc.
+// these operations are performed on the metrics generated by the first stage.
+//
+// NOTE: find a batter name for this, maybe something like AggregateStage.
+// This stage operates on metrics generated by the first stage and performs aggregation on traceql metrics.
+// for now, calling it second stage is fine because it is the second stage in the pipeline.
+// and we already have MetricsAggregate which is the in the first stage that operates on spans
+// and produces metrics so we need to rename that to make things clear and avoid confusion.
+type secondStageElement interface {
+	Element
+	init(req *tempopb.QueryRangeRequest)
+	process(input SeriesSet) SeriesSet
+}
+
+type SecondStageOp int
+
+const (
+	OpTopK SecondStageOp = iota
+	OpBottomK
+)
+
+var errInvalidLimit = fmt.Errorf("limit must be greater than 0")
+
+func (op SecondStageOp) String() string {
+	switch op {
+	case OpTopK:
+		return "topk"
+	case OpBottomK:
+		return "bottomk"
+	}
+	return "unknown"
+}
+
+// TopKBottomK handles second stage topK/bottomK operations
+type TopKBottomK struct {
+	op     SecondStageOp
+	limit  int
+	length int
+}
+
+func newTopKBottomK(op SecondStageOp, limit int) *TopKBottomK {
+	return &TopKBottomK{op: op, limit: limit}
+}
+
+func (m *TopKBottomK) String() string {
+	return fmt.Sprintf("%s(%d)", m.op.String(), m.limit)
+}
+
+func (m *TopKBottomK) validate() error {
+	if m.limit <= 0 {
+		return errInvalidLimit
+	}
+	return nil
+}
+
+func (m *TopKBottomK) init(req *tempopb.QueryRangeRequest) {
+	m.length = IntervalCount(req.Start, req.End, req.Step)
+}
+
+func (m *TopKBottomK) process(input SeriesSet) SeriesSet {
+	// if input size is less or equal to limit, return input as is
+	if len(input) <= m.limit {
+		return input
+	}
+
+	switch m.op {
+	case OpTopK:
+		return processTopK(input, m.length, m.limit)
+	case OpBottomK:
+		return processBottomK(input, m.length, m.limit)
+	default:
+		// unknown operation, return input SeriesSet, we shouldn't reach here
+		return input
+	}
+}
+
+var _ secondStageElement = (*TopKBottomK)(nil)
@@ -14,6 +14,10 @@ func (r RootExpr) String() string {
 		s.WriteString(" | ")
 		s.WriteString(r.MetricsPipeline.String())
 	}
+	if r.MetricsSecondStage != nil {
+		s.WriteString(" | ")
+		s.WriteString(r.MetricsSecondStage.String())
+	}
 	if r.Hints != nil {
 		s.WriteString(" ")
 		s.WriteString(r.Hints.String())
@@ -25,7 +25,26 @@ func (r RootExpr) validate() error {
 	}

 	if r.MetricsPipeline != nil {
-		return r.MetricsPipeline.validate()
+		err := r.MetricsPipeline.validate()
+		if err != nil {
+			return err
+		}
+	}
+
+	if r.MetricsSecondStage != nil {
+		err := r.MetricsSecondStage.validate()
+		if err != nil {
+			return err
+		}
+	}
+
+	// extra validation to disallow compare() with second stage functions
+	// for example: `{} | compare({status=error}) | topk(10)` doesn't make sense
+	if r.MetricsPipeline != nil && r.MetricsSecondStage != nil {
+		// cast and check if the first stage is a compare operation
+		if _, ok := r.MetricsPipeline.(*MetricsCompare); ok {
+			return fmt.Errorf("`compare()` cannot be used with second stage functions")
+		}
 	}

 	return nil
@@ -27,10 +27,10 @@ func NewEngine() *Engine {
 	return &Engine{}
 }

-func Compile(query string) (*RootExpr, SpansetFilterFunc, metricsFirstStageElement, *FetchSpansRequest, error) {
+func Compile(query string) (*RootExpr, SpansetFilterFunc, firstStageElement, secondStageElement, *FetchSpansRequest, error) {
 	expr, err := Parse(query)
 	if err != nil {
-		return nil, nil, nil, nil, err
+		return nil, nil, nil, nil, nil, err
 	}

 	req := &FetchSpansRequest{
@@ -40,17 +40,17 @@ func Compile(query string) (*RootExpr, SpansetFilterFunc, metricsFirstStageEleme

 	err = expr.validate()
 	if err != nil {
-		return nil, nil, nil, nil, err
+		return nil, nil, nil, nil, nil, err
 	}

-	return expr, expr.Pipeline.evaluate, expr.MetricsPipeline, req, nil
+	return expr, expr.Pipeline.evaluate, expr.MetricsPipeline, expr.MetricsSecondStage, req, nil
 }

 func (e *Engine) ExecuteSearch(ctx context.Context, searchReq *tempopb.SearchRequest, spanSetFetcher SpansetFetcher) (*tempopb.SearchResponse, error) {
 	ctx, span := tracer.Start(ctx, "traceql.Engine.ExecuteSearch")
 	defer span.End()

-	rootExpr, _, _, fetchSpansRequest, err := Compile(searchReq.Query)
+	rootExpr, _, _, _, fetchSpansRequest, err := Compile(searchReq.Query)
 	if err != nil {
 		return nil, err
 	}
@@ -1,6 +1,7 @@
 package traceql

 import (
+	"container/heap"
 	"context"
 	"errors"
 	"fmt"
@@ -282,6 +283,12 @@ func (set SeriesSet) ToProto(req *tempopb.QueryRangeRequest) []*tempopb.TimeSeri
 			exemplars = make([]tempopb.Exemplar, 0, len(s.Exemplars))
 		}
 		for _, e := range s.Exemplars {
+			// skip exemplars that has NaN value
+			i := IntervalOfMs(int64(e.TimestampMs), start, end, req.Step)
+			if i < 0 || i >= len(s.Values) || math.IsNaN(s.Values[i]) { // strict bounds check
+				continue
+			}
+
 			labels := make([]commonv1proto.KeyValue, 0, len(e.Labels))
 			for _, label := range e.Labels {
 				labels = append(labels,
@@ -787,24 +794,33 @@ func (e *Engine) CompileMetricsQueryRangeNonRaw(req *tempopb.QueryRangeRequest,
 		return nil, fmt.Errorf("step required")
 	}

-	_, _, metricsPipeline, _, err := Compile(req.Query)
+	_, _, metricsPipeline, metricsSecondStage, _, err := Compile(req.Query)
 	if err != nil {
 		return nil, fmt.Errorf("compiling query: %w", err)
 	}

+	// for metrics queries, we need a metrics pipeline
 	if metricsPipeline == nil {
 		return nil, fmt.Errorf("not a metrics query")
 	}

 	metricsPipeline.init(req, mode)
-
-	return &MetricsFrontendEvaluator{
+	mfe := &MetricsFrontendEvaluator{
 		metricsPipeline: metricsPipeline,
-	}, nil
+	}
+
+	// only run metrics second stage if we have second stage and query mode = final,
+	// as we are not sharding them now in lower layers.
+	if metricsSecondStage != nil && mode == AggregateModeFinal {
+		metricsSecondStage.init(req)
+		mfe.metricsSecondStage = metricsSecondStage
+	}
+
+	return mfe, nil
 }

 // CompileMetricsQueryRange returns an evaluator that can be reused across multiple data sources.
-// Dedupe spans parameter is an indicator of whether to expect duplicates in the datasource. For
+// Dedupe spans parameter is an indicator of whether to expected duplicates in the datasource. For
 // example if the datasource is replication factor=1 or only a single block then we know there
 // aren't duplicates, and we can make some optimizations.
 func (e *Engine) CompileMetricsQueryRange(req *tempopb.QueryRangeRequest, exemplars int, timeOverlapCutoff float64, allowUnsafeQueryHints bool) (*MetricsEvaluator, error) {
@@ -821,7 +837,7 @@ func (e *Engine) CompileMetricsQueryRange(req *tempopb.QueryRangeRequest, exempl
 		return nil, fmt.Errorf("step required")
 	}

-	expr, eval, metricsPipeline, storageReq, err := Compile(req.Query)
+	expr, eval, metricsPipeline, _, storageReq, err := Compile(req.Query)
 	if err != nil {
 		return nil, fmt.Errorf("compiling query: %w", err)
 	}
@@ -991,7 +1007,7 @@ type MetricsEvaluator struct {
 	exemplarMap                     map[string]struct{}
 	timeOverlapCutoff               float64
 	storageReq                      *FetchSpansRequest
-	metricsPipeline                 metricsFirstStageElement
+	metricsPipeline                 firstStageElement
 	spansTotal, spansDeduped, bytes uint64
 	mtx                             sync.Mutex
 }
@@ -1108,6 +1124,11 @@ func (e *MetricsEvaluator) Metrics() (uint64, uint64, uint64) {
 }

 func (e *MetricsEvaluator) Results() SeriesSet {
+	// NOTE: skip processing of second stage because not all first stage functions can't be pushed down.
+	// for example: if query has avg_over_time(), then we can't push it down to second stage, and second stage
+	// can only be processed on the frontend.
+	// we could do this but it would require knowing if the first stage functions
+	// can be pushed down to second stage or not so we are skipping it for now, and will handle it later.
 	return e.metricsPipeline.result()
 }

@@ -1133,8 +1154,9 @@ func (e *MetricsEvaluator) sampleExemplar(id []byte) bool {
 // MetricsFrontendEvaluator pipes the sharded job results back into the engine for the rest
 // of the pipeline.  i.e. This evaluator is for the query-frontend.
 type MetricsFrontendEvaluator struct {
-	mtx             sync.Mutex
-	metricsPipeline metricsFirstStageElement
+	mtx                sync.Mutex
+	metricsPipeline    firstStageElement
+	metricsSecondStage secondStageElement
 }

 func (m *MetricsFrontendEvaluator) ObserveSeries(in []*tempopb.TimeSeries) {
@@ -1148,7 +1170,16 @@ func (m *MetricsFrontendEvaluator) Results() SeriesSet {
 	m.mtx.Lock()
 	defer m.mtx.Unlock()

-	return m.metricsPipeline.result()
+	results := m.metricsPipeline.result()
+
+	if m.metricsSecondStage != nil {
+		// metrics second stage is only set when query has second stage function and mode = final
+		// if we have metrics second stage, pass first stage results through
+		// second stage for further processing.
+		results = m.metricsSecondStage.process(results)
+	}
+
+	return results
 }

 type SeriesAggregator interface {
@@ -1528,3 +1559,131 @@ func FloatizeAttribute(s Span, a Attribute) (float64, StaticType) {
 	}
 	return f, v.Type
 }
+
+// processTopK implements TopKBottomK topk method
+func processTopK(input SeriesSet, valueLength, limit int) SeriesSet {
+	result := make(SeriesSet)
+	// Min heap for top-k (smallest values at top for easy replacement)
+	h := &seriesHeap{}
+	heap.Init(h)
+
+	// process each timestamp
+	for i := 0; i < valueLength; i++ {
+		// process each series for this timestamp
+		for key, series := range input {
+			if i >= len(series.Values) {
+				continue
+			}
+
+			value := series.Values[i]
+			if math.IsNaN(value) {
+				continue // Skip NaN values
+			}
+
+			// If heap not full yet, add the value
+			if h.Len() < limit {
+				heap.Push(h, seriesValue{
+					key:   key,
+					value: value,
+				})
+				continue
+			}
+
+			// If new value is greater than smallest in heap, replace it
+			smallest := (*h)[0]
+			if value > smallest.value {
+				heap.Pop(h)
+				heap.Push(h, seriesValue{
+					key:   key,
+					value: value,
+				})
+			}
+		}
+
+		// we have iterated over all series for this timestamp
+		// empty the heap and record these series in result set
+		for h.Len() > 0 {
+			sv := heap.Pop(h).(seriesValue)
+			initSeriesInResult(result, sv.key, input, valueLength)
+			// Set only this timestamp's value
+			result[sv.key].Values[i] = input[sv.key].Values[i]
+		}
+	}
+
+	return result
+}
+
+// processBottomK implements TopKBottomK bottomk method
+func processBottomK(input SeriesSet, valueLength, limit int) SeriesSet {
+	result := make(SeriesSet)
+
+	// Max heap for bottom-k (largest values at top for easy replacement)
+	h := &reverseSeriesHeap{}
+	heap.Init(h)
+
+	// Process each timestamp
+	for i := 0; i < valueLength; i++ {
+		// Process each series for this timestamp
+		for key, series := range input {
+			if i >= len(series.Values) {
+				continue
+			}
+
+			value := series.Values[i]
+			if math.IsNaN(value) {
+				continue // Skip NaN values
+			}
+
+			// If heap not full yet, add the value
+			if h.Len() < limit {
+				heap.Push(h, seriesValue{
+					key:   key,
+					value: value,
+				})
+				continue
+			}
+
+			// If new value is less than largest in heap, replace it
+			largest := (*h)[0]
+			if value < largest.value {
+				heap.Pop(h)
+				heap.Push(h, seriesValue{
+					key:   key,
+					value: value,
+				})
+			}
+		}
+
+		// we have iterated over all series for this timestamp
+		// empty the heap and record these series in result set
+		for h.Len() > 0 {
+			sv := heap.Pop(h).(seriesValue)
+			initSeriesInResult(result, sv.key, input, valueLength)
+			// Set only this timestamp's value
+			result[sv.key].Values[i] = input[sv.key].Values[i]
+		}
+	}
+
+	return result
+}
+
+// initSeriesInResult ensures that a series exists in the result map, and
+// initializes it with NaN values if it doesn't exist in the result set.
+func initSeriesInResult(result SeriesSet, key string, input SeriesSet, valueLength int) {
+	if _, exists := result[key]; exists {
+		// series already exists, no need to initialize
+		return
+	}
+	// series doesn't exist, initialize it
+	// Copy the series labels and exemplars from the input
+	result[key] = TimeSeries{
+		Labels:    input[key].Labels,
+		Values:    make([]float64, valueLength),
+		Exemplars: input[key].Exemplars,
+	}
+
+	// Initialize all values to NaN because we only want to set values for this timestamp
+	for j := range result[key].Values {
+		result[key].Values[j] = math.NaN()
+	}
+}
@@ -23,7 +23,7 @@ type averageOverTimeAggregator struct {
 	mode       AggregateMode
 }

-var _ metricsFirstStageElement = (*averageOverTimeAggregator)(nil)
+var _ firstStageElement = (*averageOverTimeAggregator)(nil)

 func newAverageOverTimeMetricsAggregator(attr Attribute, by []Attribute) *averageOverTimeAggregator {
 	return &averageOverTimeAggregator{
@@ -312,7 +312,7 @@ func (m *MetricsCompare) String() string {
 	return "compare(" + m.f.String() + "}"
 }

-var _ metricsFirstStageElement = (*MetricsCompare)(nil)
+var _ firstStageElement = (*MetricsCompare)(nil)

 // BaselineAggregator is a special series combiner for the compare() function.
 // It resplits job-level results into baseline and selection buffers, and if
@@ -760,6 +760,46 @@ func TestAvgOverTimeForDuration(t *testing.T) {
 	assert.True(t, math.IsNaN(fooBar.Values[2]))
 }

+func TestAvgOverTimeForDurationWithSecondStage(t *testing.T) {
+	req := &tempopb.QueryRangeRequest{
+		Start: uint64(1 * time.Second),
+		End:   uint64(3 * time.Second),
+		Step:  uint64(1 * time.Second),
+		Query: "{ } | avg_over_time(duration) by (span.foo) | topk(1)",
+	}
+
+	// A variety of spans across times, durations, and series. All durations are powers of 2 for simplicity
+	in := []Span{
+		newMockSpan(nil).WithStartTime(uint64(1*time.Second)).WithSpanString("foo", "bar").WithDuration(100),
+		newMockSpan(nil).WithStartTime(uint64(1*time.Second)).WithSpanString("foo", "bar").WithDuration(100),
+		newMockSpan(nil).WithStartTime(uint64(1*time.Second)).WithSpanString("foo", "bar").WithDuration(100),
+
+		newMockSpan(nil).WithStartTime(uint64(2*time.Second)).WithSpanString("foo", "bar").WithDuration(100),
+		newMockSpan(nil).WithStartTime(uint64(2*time.Second)).WithSpanString("foo", "bar").WithDuration(100),
+		newMockSpan(nil).WithStartTime(uint64(2*time.Second)).WithSpanString("foo", "bar").WithDuration(100),
+		newMockSpan(nil).WithStartTime(uint64(2*time.Second)).WithSpanString("foo", "bar").WithDuration(500),
+
+		newMockSpan(nil).WithStartTime(uint64(3*time.Second)).WithSpanString("foo", "baz").WithDuration(100),
+		newMockSpan(nil).WithStartTime(uint64(3*time.Second)).WithSpanString("foo", "baz").WithDuration(200),
+		newMockSpan(nil).WithStartTime(uint64(3*time.Second)).WithSpanString("foo", "baz").WithDuration(300),
+	}
+
+	result, err := runTraceQLMetric(req, in)
+	require.NoError(t, err)
+
+	fooBaz := result[`{"span.foo"="baz"}`]
+	fooBar := result[`{"span.foo"="bar"}`]
+
+	// We cannot compare with require.Equal because NaN != NaN
+	assert.True(t, math.IsNaN(fooBaz.Values[0]))
+	assert.True(t, math.IsNaN(fooBaz.Values[1]))
+	assert.Equal(t, 200., fooBaz.Values[2]*float64(time.Second))
+
+	assert.Equal(t, 100., fooBar.Values[0]*float64(time.Second))
+	assert.Equal(t, 200., fooBar.Values[1]*float64(time.Second))
+	assert.True(t, math.IsNaN(fooBar.Values[2]))
+}
+
 func TestAvgOverTimeForDurationWithoutAggregation(t *testing.T) {
 	req := &tempopb.QueryRangeRequest{
 		Start: uint64(1 * time.Second),
@@ -1336,6 +1376,365 @@ func TestHistogramOverTime(t *testing.T) {
 	require.Equal(t, out, result)
 }

+func TestSecondStageTopK(t *testing.T) {
+	req := &tempopb.QueryRangeRequest{
+		Start: uint64(1 * time.Second),
+		End:   uint64(8 * time.Second),
+		Step:  uint64(1 * time.Second),
+		Query: "{ } | rate() by (span.foo) | topk(2)",
+	}
+
+	in := make([]Span, 0)
+	// 15 spans, at different start times across 3 series
+	in = append(in, generateSpans(7, []int{1, 2, 3, 4, 5, 6, 7, 8}, "bar")...)
+	in = append(in, generateSpans(5, []int{1, 2, 3, 4, 5, 6, 7, 8}, "baz")...)
+	in = append(in, generateSpans(3, []int{1, 2, 3, 4, 5, 6, 7, 8}, "quax")...)
+
+	result, err := runTraceQLMetric(req, in)
+	require.NoError(t, err)
+
+	// bar and baz have more spans so they should be the top 2
+	resultBar := result[`{"span.foo"="bar"}`]
+	require.Equal(t, []float64{7, 7, 7, 7, 7, 7, 7, 7}, resultBar.Values)
+	resultBaz := result[`{"span.foo"="baz"}`]
+	require.Equal(t, []float64{5, 5, 5, 5, 5, 5, 5, 5}, resultBaz.Values)
+}
+
+func TestSecondStageTopKAverage(t *testing.T) {
+	req := &tempopb.QueryRangeRequest{
+		Start: uint64(1 * time.Second),
+		End:   uint64(8 * time.Second),
+		Step:  uint64(1 * time.Second),
+		Query: "{ } | avg_over_time(duration) by (span.foo) | topk(2)",
+	}
+
+	in := make([]Span, 0)
+	// 15 spans, at different start times across 3 series
+	in = append(in, generateSpans(7, []int{1, 2, 3, 4, 5, 6, 7, 8}, "bar")...)
+	in = append(in, generateSpans(5, []int{1, 2, 3, 4, 5, 6, 7, 8}, "baz")...)
+	in = append(in, generateSpans(3, []int{1, 2, 3, 4, 5, 6, 7, 8}, "quax")...)
+
+	result, err := runTraceQLMetric(req, in)
+	require.NoError(t, err)
+
+	resultBar := result[`{"span.foo"="bar"}`]
+	val1 := 0.000000512
+	require.Equal(t, []float64{val1, val1, val1, val1, val1, val1, val1, val1}, resultBar.Values)
+	resultBaz := result[`{"span.foo"="baz"}`]
+	val2 := 0.00000038400000000000005
+	require.Equal(t, []float64{val2, val2, val2, val2, val2, val2, val2, val2}, resultBaz.Values)
+}
+
+func TestSecondStageBottomK(t *testing.T) {
+	req := &tempopb.QueryRangeRequest{
+		Start: uint64(1 * time.Second),
+		End:   uint64(8 * time.Second),
+		Step:  uint64(1 * time.Second),
+		Query: "{ } | rate() by (span.foo) | bottomk(2)",
+	}
+
+	in := make([]Span, 0)
+	// 15 spans, at different start times across 3 series
+	in = append(in, generateSpans(7, []int{1, 2, 3, 4, 5, 6, 7, 8}, "bar")...)
+	in = append(in, generateSpans(5, []int{1, 2, 3, 4, 5, 6, 7, 8}, "baz")...)
+	in = append(in, generateSpans(3, []int{1, 2, 3, 4, 5, 6, 7, 8}, "quax")...)
+
+	result, err := runTraceQLMetric(req, in)
+	require.NoError(t, err)
+
+	// quax and baz have the lowest spans so they should be the bottom 2
+	resultBar := result[`{"span.foo"="quax"}`]
+	require.Equal(t, []float64{3, 3, 3, 3, 3, 3, 3, 3}, resultBar.Values)
+	resultBaz := result[`{"span.foo"="baz"}`]
+	require.Equal(t, []float64{5, 5, 5, 5, 5, 5, 5, 5}, resultBaz.Values)
+}
+
+func TestProcessTopK(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    SeriesSet
+		limit    int
+		expected SeriesSet
+	}{
+		{
+			name: "topk selection",
+			input: createSeriesSet(map[string][]float64{
+				"a": {1, 5, 3},
+				"b": {2, 6, 2},
+				"c": {3, 1, 1},
+				"d": {4, 2, 4},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {math.NaN(), 5, 3},          // Top-2 at timestamp 1, 2
+				"b": {math.NaN(), 6, math.NaN()}, // Top-2 at timestamp 1
+				"c": {3, math.NaN(), math.NaN()}, // Top-2 at timestamp 0
+				"d": {4, math.NaN(), 4},          // Top-2 at timestamps 0, 2
+			}),
+		},
+		{
+			name: "topk selection at each timestamp",
+			input: createSeriesSet(map[string][]float64{
+				"a": {1, 2, 3},
+				"b": {2, 3, 4},
+				"c": {3, 4, 5},
+				"d": {1, 1, 1},
+				"e": {0.5, 2, 1},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"b": {2, 3, 4}, // Top-2 at all timestamps
+				"c": {3, 4, 5}, // Top-2 at all timestamps
+			}),
+		},
+		{
+			name: "select single highest value at specific timestamp",
+			input: createSeriesSet(map[string][]float64{
+				"a": {1, 6, 3},
+				"b": {4, 5, 1},
+				"c": {2, 3, 7},
+			}),
+			limit: 1,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {math.NaN(), 6, math.NaN()}, // top at timestamp 1
+				"b": {4, math.NaN(), math.NaN()}, // top at timestamp 0
+				"c": {math.NaN(), math.NaN(), 7}, // top at timestamp 2
+			}),
+		},
+		{
+			name: "with NaN values",
+			input: createSeriesSet(map[string][]float64{
+				"a": {1, math.NaN(), 3},
+				"b": {2, 6, math.NaN()},
+				"c": {3, 1, 1},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {math.NaN(), math.NaN(), 3}, // Top-2 at timestamp 2
+				"b": {2, 6, math.NaN()},          // Top-2 at timestamp 0, 1
+				"c": {3, 1, 1},                   // Top-2 at timestamp 0, 2
+			}),
+		},
+		{
+			name: "series with all NaN values is skipped",
+			input: createSeriesSet(map[string][]float64{
+				"a": {math.NaN(), math.NaN(), math.NaN()},
+				"b": {2, 6, math.NaN()},
+				"c": {3, 1, 1},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"b": {2, 6, math.NaN()},
+				"c": {3, 1, 1},
+			}),
+		},
+		{
+			name: "all series with all NaN values",
+			input: createSeriesSet(map[string][]float64{
+				"a": {math.NaN(), math.NaN(), math.NaN()},
+				"b": {math.NaN(), math.NaN(), math.NaN()},
+			}),
+			limit:    2,
+			expected: SeriesSet{},
+		},
+		{
+			name:     "empty input",
+			input:    SeriesSet{},
+			limit:    2,
+			expected: SeriesSet{},
+		},
+		{
+			name: "limit larger than series count",
+			input: createSeriesSet(map[string][]float64{
+				"a": {1, 5, 3},
+				"b": {2, 6, 2},
+			}),
+			limit: 5,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {1, 5, 3},
+				"b": {2, 6, 2},
+			}),
+		},
+		{
+			name: "negative and infinity values",
+			input: createSeriesSet(map[string][]float64{
+				"a": {-1, 5, math.Inf(-1)},
+				"b": {-2, 6, math.Inf(1)},
+				"c": {-3, 7, 1},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {-1, math.NaN(), math.NaN()}, // Top-2 at timestamp 0
+				"b": {-2, 6, math.Inf(1)},         // Top-2 at timestamps 1, 2
+				"c": {math.NaN(), 7, 1},           // Top-2 at timestamp 1
+			}),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := processTopK(tt.input, 3, tt.limit)
+			expectSeriesSet(t, tt.expected, result)
+		})
+	}
+}
+
+func TestProcessBottomK(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    SeriesSet
+		limit    int
+		expected SeriesSet
+	}{
+		{
+			name: "bottomk selection",
+			input: createSeriesSet(map[string][]float64{
+				"a": {1, 5, 3},
+				"b": {2, 6, 2},
+				"c": {3, 1, 1},
+				"d": {4, 2, 4},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {1, math.NaN(), math.NaN()}, // Bottom-2 at timestamp 0
+				"b": {2, math.NaN(), 2},          // Bottom-2 at timestamps 0, 2
+				"c": {math.NaN(), 1, 1},          // Bottom-2 at timestamps 1, 2
+				"d": {math.NaN(), 2, math.NaN()}, // Bottom-2 at timestamp 1
+			}),
+		},
+		{
+			name: "bottomk selection at each timestamp",
+			input: createSeriesSet(map[string][]float64{
+				"a": {5, 4, 3},
+				"b": {6, 5, 4},
+				"c": {7, 6, 1},
+				"d": {3, 3, 3},
+				"e": {4, 2, 2},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"c": {math.NaN(), math.NaN(), 1}, // bottom 2 at timestamp 2
+				"d": {3, 3, math.NaN()},          // bottom 2 at timestamp 0, 1
+				"e": {4, 2, 2},                   // bottom 2 at timestamp 0, 1, 2
+			}),
+		},
+		{
+			name: "select single lowest value at specific timestamp",
+			input: createSeriesSet(map[string][]float64{
+				"a": {3, 1, 5},
+				"b": {1, 2, 3},
+				"c": {4, 5, 1},
+			}),
+			limit: 1,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {math.NaN(), 1, math.NaN()}, // Lowest at timestamp 1
+				"b": {1, math.NaN(), math.NaN()}, // Lowest at timestamp 0
+				"c": {math.NaN(), math.NaN(), 1}, // Lowest at timestamp 2
+			}),
+		},
+		{
+			name: "with NaN values",
+			input: createSeriesSet(map[string][]float64{
+				"a": {1, math.NaN(), 3},
+				"b": {4, 6, math.NaN()},
+				"c": {5, 1, 1},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {1, math.NaN(), 3}, // Bottom-2 at timestamp 0
+				"b": {4, 6, math.NaN()}, // NaN values are skipped in comparison
+				"c": {math.NaN(), 1, 1}, // Bottom-2 at timestamps 1, 2
+			}),
+		},
+		{
+			name: "all series with NaN values",
+			input: createSeriesSet(map[string][]float64{
+				"a": {math.NaN(), math.NaN(), math.NaN()},
+				"b": {math.NaN(), math.NaN(), math.NaN()},
+				"c": {math.NaN(), math.NaN(), math.NaN()},
+			}),
+			limit:    2,
+			expected: SeriesSet{},
+		},
+		{
+			name:     "empty input",
+			input:    SeriesSet{},
+			limit:    2,
+			expected: SeriesSet{},
+		},
+		{
+			name: "limit larger than series count",
+			input: createSeriesSet(map[string][]float64{
+				"a": {1, 5, 3},
+				"b": {2, 6, 2},
+			}),
+			limit: 5,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {1, 5, 3},
+				"b": {2, 6, 2},
+			}),
+		},
+		{
+			name: "negative and infinity values",
+			input: createSeriesSet(map[string][]float64{
+				"a": {-1, 5, math.Inf(-1)},
+				"b": {-2, 6, math.Inf(1)},
+				"c": {-3, 7, 1},
+			}),
+			limit: 2,
+			expected: createSeriesSet(map[string][]float64{
+				"a": {math.NaN(), 5, math.Inf(-1)}, // Bottom-2 at timestamp 2
+				"b": {-2, 6, math.NaN()},           // Bottom-2 at timestamp 0
+				"c": {-3, math.NaN(), 1},           // Bottom-2 at timestamps 0, 2
+			}),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fmt.Printf("input: %v\n", tt.input)
+			result := processBottomK(tt.input, 3, tt.limit)
+			fmt.Printf("result: %v\n", result)
+			expectSeriesSet(t, tt.expected, result)
+		})
+	}
+}
+
+func TestTiesInTopK(t *testing.T) {
+	input := createSeriesSet(map[string][]float64{
+		"a": {10, 5, 1},
+		"b": {10, 4, 2},
+		"c": {10, 3, 3},
+	})
+	result := processTopK(input, 3, 2)
+	fmt.Printf("result: %v\n", result)
+
+	// because of ties, we can have different result at index 0
+	// "a" can be [10, 5, NaN] OR [NaN, 5, NaN]
+	// "b" can be [10, 4, 2] OR [NaN, 4, 2]
+	// "c" can be [10, NaN, 3] OR [NaN, NaN, 3]
+	checkEqualForTies(t, result[`{label="a"}`].Values, []float64{10, 5, math.NaN()})
+	checkEqualForTies(t, result[`{label="b"}`].Values, []float64{10, 4, 2})
+	checkEqualForTies(t, result[`{label="c"}`].Values, []float64{10, math.NaN(), 3})
+}
+
+func TestTiesInBottomK(t *testing.T) {
+	input := createSeriesSet(map[string][]float64{
+		"a": {10, 5, 1},
+		"b": {10, 4, 2},
+		"c": {10, 3, 3},
+	})
+	result := processBottomK(input, 3, 2)
+
+	// because of ties, we can have different result at index 0
+	// "a" can be [10, NaN, 1] OR [NaN, NaN, 1]
+	// "b" can be [10, 4, 2] OR [NaN, 4, 2]
+	// "c" can be [10, 3, NaN] OR [NaN, 3, NaN]
+	checkEqualForTies(t, result[`{label="a"}`].Values, []float64{10, math.NaN(), 1})
+	checkEqualForTies(t, result[`{label="b"}`].Values, []float64{10, 4, 2})
+	checkEqualForTies(t, result[`{label="c"}`].Values, []float64{10, 3, math.NaN()})
+}
+
 func runTraceQLMetric(req *tempopb.QueryRangeRequest, inSpans ...[]Span) (SeriesSet, error) {
 	e := NewEngine()

@@ -1380,6 +1779,66 @@ func randFloat(minimum, maximum float64) float64 {
 	return rand.Float64()*(maximum-minimum) + minimum
 }

+func generateSpans(count int, startTimes []int, value string) []Span {
+	spans := make([]Span, 0)
+	for i := 0; i < count; i++ {
+		for _, t := range startTimes {
+			sTime := uint64(time.Duration(t) * time.Second)
+			spans = append(spans, newMockSpan(nil).WithStartTime(sTime).WithSpanString("foo", value).WithDuration(128*uint64(i+1)))
+		}
+	}
+	return spans
+}
+
+// createSeriesSet to create a SeriesSet from a map of values
+func createSeriesSet(data map[string][]float64) SeriesSet {
+	seriesSet := SeriesSet{}
+	labelName := "label"
+	for key, values := range data {
+		seriesSet[fmt.Sprintf(`{%s="%s"}`, labelName, key)] = TimeSeries{
+			Values: values,
+			Labels: Labels{Label{Name: labelName, Value: NewStaticString(key)}},
+		}
+	}
+	return seriesSet
+}
+
+// expectSeriesSet validates SeriesSet equality, and also considers NaN values
+func expectSeriesSet(t *testing.T, expected, result SeriesSet) {
+	for expectedKey, expectedSeries := range expected {
+		resultSeries, ok := result[expectedKey]
+		require.True(t, ok, "expected series %s to be in result", expectedKey)
+		require.Equal(t, expectedSeries.Labels, resultSeries.Labels)
+
+		// check values, including NaN values
+		require.Equal(t, len(expectedSeries.Values), len(resultSeries.Values))
+		for i, expectedValue := range expectedSeries.Values {
+			if math.IsNaN(expectedValue) {
+				require.True(t, math.IsNaN(resultSeries.Values[i]), "expected NaN at index %d", i)
+			} else {
+				require.Equal(t, expectedValue, resultSeries.Values[i])
+			}
+		}
+	}
+}
+
+func checkEqualForTies(t *testing.T, result, expected []float64) {
+	for i := range result {
+		switch i {
+		// at index 0, we have a tie so it can be sometimes NaN
+		case 0:
+			require.True(t, math.IsNaN(result[0]) || result[0] == expected[0],
+				"index 0: expected NaN or %v, got %v", expected[0], result[0])
+		default:
+			if math.IsNaN(expected[i]) {
+				require.True(t, math.IsNaN(result[i]), "index %d: expected NaN, got %v", i, result[i])
+			} else {
+				require.Equal(t, expected[i], result[i], "index %d: expected %v", i, expected[i])
+			}
+		}
+	}
+}
+
 func BenchmarkSumOverTime(b *testing.B) {
 	totalSpans := 1_000_000
 	in := make([]Span, 0, totalSpans)
@@ -531,21 +531,21 @@ func TestExamplesInEngine(t *testing.T) {

 	for _, q := range queries.Valid {
 		t.Run("valid - "+q, func(t *testing.T) {
-			_, _, _, _, err := Compile(q)
+			_, _, _, _, _, err := Compile(q)
 			require.NoError(t, err)
 		})
 	}

 	for _, q := range queries.ParseFails {
 		t.Run("parse fails - "+q, func(t *testing.T) {
-			_, _, _, _, err := Compile(q)
+			_, _, _, _, _, err := Compile(q)
 			require.Error(t, err)
 		})
 	}

 	for _, q := range queries.ValidateFails {
 		t.Run("validate fails - "+q, func(t *testing.T) {
-			_, _, _, _, err := Compile(q)
+			_, _, _, _, _, err := Compile(q)
 			require.Error(t, err)
 			var unErr *unsupportedError
 			require.False(t, errors.As(err, &unErr))
@@ -554,7 +554,7 @@ func TestExamplesInEngine(t *testing.T) {

 	for _, q := range queries.Unsupported {
 		t.Run("unsupported - "+q, func(t *testing.T) {
-			_, _, _, _, err := Compile(q)
+			_, _, _, _, _, err := Compile(q)
 			require.Error(t, err)
 			var unErr *unsupportedError
 			require.True(t, errors.As(err, &unErr))
@@ -29,7 +29,8 @@ import (
    wrappedScalarPipeline Pipeline
    scalarPipeline Pipeline
    aggregate Aggregate
-    metricsAggregation metricsFirstStageElement
+    metricsAggregation firstStageElement
+    metricsSecondStage secondStageElement

    fieldExpression FieldExpression
    static Static
@@ -64,6 +65,7 @@ import (
 %type <scalarFilter> scalarFilter
 %type <scalarFilterOperation> scalarFilterOperation
 %type <metricsAggregation> metricsAggregation
+%type <metricsSecondStage> metricsSecondStage

 %type <scalarPipelineExpressionFilter> scalarPipelineExpressionFilter
 %type <scalarPipelineExpression> scalarPipelineExpression
@@ -101,6 +103,7 @@ import (
                        BY COALESCE SELECT
                        END_ATTRIBUTE
                        RATE COUNT_OVER_TIME MIN_OVER_TIME MAX_OVER_TIME AVG_OVER_TIME SUM_OVER_TIME QUANTILE_OVER_TIME HISTOGRAM_OVER_TIME COMPARE
+                        TOPK BOTTOMK
                        WITH

 // Operators are listed with increasing precedence.
@@ -121,6 +124,8 @@ root:
  | spansetPipelineExpression                   { yylex.(*lexer).expr = newRootExpr($1) }
  | scalarPipelineExpressionFilter              { yylex.(*lexer).expr = newRootExpr($1) } 
  | spansetPipeline PIPE metricsAggregation     { yylex.(*lexer).expr = newRootExprWithMetrics($1, $3) }
+  // note: would only work for single metrics pipeline and not for multiple metrics pipelines before the fucntions
+  | spansetPipeline PIPE metricsAggregation PIPE metricsSecondStage  { yylex.(*lexer).expr = newRootExprWithMetricsTwoStage($1, $3, $5) }
  | root hints                                  { yylex.(*lexer).expr.withHints($2) }
  ;

@@ -292,6 +297,7 @@ aggregate:

 // **********************
 // Metrics
+// TODO: rename metricsAggregation -> metricsFirstStage
 // **********************
 metricsAggregation:
      RATE            OPEN_PARENS CLOSE_PARENS                                                                          { $$ = newMetricsAggregate(metricsAggregateRate, nil) }
@@ -315,6 +321,14 @@ metricsAggregation:
    | COMPARE OPEN_PARENS spansetFilter COMMA INTEGER COMMA INTEGER COMMA INTEGER CLOSE_PARENS                          { $$ = newMetricsCompare($3, $5, $7, $9)}
  ;

+// **********************
+// Metrics Second Stage Functions
+// **********************
+metricsSecondStage:
+    TOPK OPEN_PARENS INTEGER CLOSE_PARENS                        { $$ = newTopKBottomK(OpTopK, $3) }
+    | BOTTOMK OPEN_PARENS INTEGER CLOSE_PARENS                   { $$ = newTopKBottomK(OpBottomK, $3) }
+  ;
+
 // **********************
 // Hints
 // **********************
@@ -110,6 +110,8 @@ var tokens = map[string]int{
 	"quantile_over_time":  QUANTILE_OVER_TIME,
 	"histogram_over_time": HISTOGRAM_OVER_TIME,
 	"compare":             COMPARE,
+	"topk":                TOPK,
+	"bottomk":             BOTTOMK,
 	"with":                WITH,
 }

@@ -1442,3 +1442,118 @@ func TestMetrics(t *testing.T) {
 		})
 	}
 }
+
+func TestMetricsSecondStage(t *testing.T) {
+	tests := []struct {
+		in       string
+		expected *RootExpr
+	}{
+		{
+			in: `{ } | rate() | topk(10)`,
+			expected: newRootExprWithMetricsTwoStage(
+				newPipeline(newSpansetFilter(NewStaticBool(true))),
+				newMetricsAggregate(metricsAggregateRate, nil),
+				newTopKBottomK(OpTopK, 10),
+			),
+		},
+		{
+			in: `{ } | count_over_time() by(name, span.http.status_code) | topk(10)`,
+			expected: newRootExprWithMetricsTwoStage(
+				newPipeline(newSpansetFilter(NewStaticBool(true))),
+				newMetricsAggregate(metricsAggregateCountOverTime, []Attribute{
+					NewIntrinsic(IntrinsicName),
+					NewScopedAttribute(AttributeScopeSpan, false, "http.status_code"),
+				}),
+				newTopKBottomK(OpTopK, 10),
+			),
+		},
+		{
+			in: `{ } | rate() | topk(10) with(foo="bar")`,
+			expected: newRootExprWithMetricsTwoStage(
+				newPipeline(newSpansetFilter(NewStaticBool(true))),
+				newMetricsAggregate(metricsAggregateRate, nil),
+				newTopKBottomK(OpTopK, 10),
+			).withHints(newHints([]*Hint{
+				newHint("foo", NewStaticString("bar")),
+			})),
+		},
+		{
+			in: `{ } | rate() | bottomk(10)`,
+			expected: newRootExprWithMetricsTwoStage(
+				newPipeline(newSpansetFilter(NewStaticBool(true))),
+				newMetricsAggregate(metricsAggregateRate, nil),
+				newTopKBottomK(OpBottomK, 10),
+			),
+		},
+		{
+			in: `{ } | count_over_time() by(name, span.http.status_code) | bottomk(10)`,
+			expected: newRootExprWithMetricsTwoStage(
+				newPipeline(newSpansetFilter(NewStaticBool(true))),
+				newMetricsAggregate(metricsAggregateCountOverTime, []Attribute{
+					NewIntrinsic(IntrinsicName),
+					NewScopedAttribute(AttributeScopeSpan, false, "http.status_code"),
+				}),
+				newTopKBottomK(OpBottomK, 10),
+			),
+		},
+		{
+			in: `{ } | rate() | bottomk(10) with(foo="bar")`,
+			expected: newRootExprWithMetricsTwoStage(
+				newPipeline(newSpansetFilter(NewStaticBool(true))),
+				newMetricsAggregate(metricsAggregateRate, nil),
+				newTopKBottomK(OpBottomK, 10),
+			).withHints(newHints([]*Hint{
+				newHint("foo", NewStaticString("bar")),
+			})),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.in, func(t *testing.T) {
+			actual, err := Parse(tc.in)
+
+			require.NoError(t, err)
+			require.Equal(t, tc.expected, actual)
+		})
+	}
+}
+
+func TestMetricsSecondStageErrors(t *testing.T) {
+	tests := []struct {
+		in  string
+		err error
+	}{
+		{
+			in:  "{} | topk(10)",
+			err: newParseError("syntax error: unexpected topk", 1, 6),
+		},
+		{
+			in:  "{} | topk(10) with(sample=0.1)",
+			err: newParseError("syntax error: unexpected topk", 1, 6),
+		},
+		{
+			in:  "{} | rate() | topk(-1)",
+			err: newParseError("syntax error: unexpected -, expecting INTEGER", 1, 20),
+		},
+		{
+			in:  "{} | bottomk(10)",
+			err: newParseError("syntax error: unexpected bottomk", 1, 6),
+		},
+		{
+			in:  "{} | bottomk(10) with(sample=0.1)",
+			err: newParseError("syntax error: unexpected bottomk", 1, 6),
+		},
+		{
+			in:  "{} | rate() | bottomk(-1)",
+			err: newParseError("syntax error: unexpected -, expecting INTEGER", 1, 23),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.in, func(t *testing.T) {
+			_, err := Parse(tc.in)
+
+			require.Equal(t, tc.err, err)
+		})
+	}
+}
@@ -0,0 +1,49 @@
+package traceql
+
+// seriesValue keeps a value from a time series with its key
+type seriesValue struct {
+	key   string
+	value float64
+}
+
+// seriesHeap implements a min-heap of seriesValue
+type seriesHeap []seriesValue
+
+func (h seriesHeap) Len() int { return len(h) }
+
+func (h seriesHeap) Less(i, j int) bool { return h[i].value < h[j].value }
+
+func (h seriesHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
+
+func (h *seriesHeap) Push(x interface{}) {
+	*h = append(*h, x.(seriesValue))
+}
+
+func (h *seriesHeap) Pop() interface{} {
+	old := *h
+	n := len(old)
+	x := old[n-1]
+	*h = old[0 : n-1]
+	return x
+}
+
+// reverseSeriesHeap implements a max-heap of seriesValue
+type reverseSeriesHeap []seriesValue
+
+func (h reverseSeriesHeap) Len() int { return len(h) }
+
+func (h reverseSeriesHeap) Less(i, j int) bool { return h[i].value > h[j].value }
+
+func (h reverseSeriesHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
+
+func (h *reverseSeriesHeap) Push(x interface{}) {
+	*h = append(*h, x.(seriesValue))
+}
+
+func (h *reverseSeriesHeap) Pop() interface{} {
+	old := *h
+	n := len(old)
+	x := old[n-1]
+	*h = old[0 : n-1]
+	return x
+}
@@ -151,6 +151,17 @@ valid:
  - '{} | avg_over_time(duration) by (span.http.path)'
  - '{} | sum_over_time(duration) by (span.http.path)'
  - '{} | quantile_over_time(duration, 0, 0.9, 1) by (span.http.path)'
+  # second stage on metrics
+  - '{} | rate() | topk(10)'
+  - '{} | rate() by (span.client_ip) | topk(10)'
+  - '{} | count_over_time() by (name) | topk(10) with(sample=0.1)'
+  - '{} | quantile_over_time(duration, 0, 0.9, 1) by (span.http.path) | topk(10)'
+  - '{} | avg_over_time(duration) by (span:name) | topk(10)'
+  - '{} | rate() by (span.client_ip) | bottomk(10)'
+  - '{} | rate() | bottomk(10)'
+  - '{} | count_over_time() by (name) | bottomk(10) with(sample=0.1)'
+  - '{} | quantile_over_time(duration, 0, 0.9, 1) by (span.http.path) | bottomk(10)'
+  - '{} | avg_over_time(duration) by (span:name) | bottomk(10)'
  # undocumented - nested set
  - '{ nestedSetLeft > 3 }'
  - '{ } >> { kind = server } | select(nestedSetLeft, nestedSetRight, nestedSetParent)'
@@ -219,6 +230,14 @@ parse_fails:
  - '{ trace:name = "a" }'
  - '{ trace:rootServiceName = "a" }' # should be trace:rootService only 
  - '{ span:rootName = "bar" }'
+  # invalid metrics second stage functions
+  - '{} | topk(10)'
+  - '{} | topk(10) with(sample=0.1)'
+  - '{} | bottomk(10)'
+  - '{} | bottomk(10) with(sample=0.1)'
+  - '{} | rate() | topk(-1)'
+  - '{} | rate() | bottomk(-1)'
+

 # validate_fails parse correctly and return an error **besides unsupported** when calling .validate()
 validate_fails:
@@ -281,6 +300,14 @@ validate_fails:
  - '{ nestedSetLeft = "foo" }'
  - '{ nestedSetRight = false }'
  - '{ nestedSetParent > "foo" }'
+  # invalid metrics second stage functions
+  - '{} | rate() | topk(0)'
+  - '{} | rate() | topk(0) with(sample=0.1)'
+  - '{} | rate() | bottomk(0)'
+  - '{} | rate() | bottomk(0) with(sample=0.1)'
+  # compare function with second stage functions is not valid
+  - '{} | compare({status=error}) | topk(10)'
+  - '{} | compare({status=error}) | bottomk(10)'

 # unsupported parse correctly and return an unsupported error when calling .validate()
 unsupported:
@@ -226,7 +226,7 @@ func GetMetrics(ctx context.Context, query, groupBy string, spanLimit int, start
 		groupByKeys[i] = groupBys[i][0].String()
 	}

-	_, eval, _, req, err := traceql.Compile(query)
+	_, eval, _, _, req, err := traceql.Compile(query)
 	if err != nil {
 		return nil, fmt.Errorf("compiling query: %w", err)
 	}
@@ -730,7 +730,7 @@ func TestBackendBlockSelectAll(t *testing.T) {

 	b := makeBackendBlockWithTraces(t, traces)

-	_, _, _, req, err := traceql.Compile("{}")
+	_, _, _, _, req, err := traceql.Compile("{}")
 	require.NoError(t, err)
 	req.SecondPass = func(inSS *traceql.Spanset) ([]*traceql.Spanset, error) { return []*traceql.Spanset{inSS}, nil }
 	req.SecondPassSelectAll = true