execution: push limits into operator telemetry

MichaHoffmann · MichaHoffmann · commit d6daab4b7f5b · 2026-04-08T06:57:59.000Z
Signed-off-by: Michael Hoffmann &lt;mhoffmann@cloudflare.com&gt;
diff --git a/engine/engine.go b/engine/engine.go
@@ -446,7 +446,7 @@ func (e *Engine) makeQueryOpts(start time.Time, end time.Time, step time.Duratio
 		EnableAnalysis:           e.enableAnalysis,
 		NoStepSubqueryIntervalFn: e.noStepSubqueryIntervalFn,
 		DecodingConcurrency:      e.decodingConcurrency,
-		SampleTracker:            query.NewSampleTracker(e.maxSamplesPerQuery),
+		SampleLimiter:            telemetry.NewSampleLimiter(e.maxSamplesPerQuery, start.UnixMilli(), end.UnixMilli(), telemetry.StepTrackingInterval(step)),
 	}
 
 	if opts == nil {
@@ -459,7 +459,6 @@ func (e *Engine) makeQueryOpts(start time.Time, end time.Time, step time.Duratio
 	if opts.EnablePerStepStats() {
 		res.EnablePerStepStats = opts.EnablePerStepStats()
 	}
-
 	if opts.DecodingConcurrency != 0 {
 		res.DecodingConcurrency = opts.DecodingConcurrency
 	}
diff --git a/engine/explain.go b/engine/explain.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/thanos-io/promql-engine/execution/model"
 	"github.com/thanos-io/promql-engine/execution/telemetry"
-	"github.com/thanos-io/promql-engine/logicalplan"
 
 	"github.com/prometheus/prometheus/promql"
 )
@@ -64,10 +63,10 @@ func (a *AnalyzeOutputNode) aggregateSamples() {
 			childPeak := child.PeakSamples()
 			a.peakSamples = max(a.peakSamples, childPeak)
 
-			switch a.OperatorTelemetry.LogicalNode().(type) {
-			case *logicalplan.Subquery:
+			switch {
+			case a.OperatorTelemetry.IsSubquery():
 				// Skip aggregating samples for subquery
-			case *logicalplan.StepInvariantExpr:
+			case a.OperatorTelemetry.IsStepInvariant():
 				childSamples := child.TotalSamples()
 				for i := range a.totalSamplesPerStep {
 					a.totalSamples += childSamples
diff --git a/execution/aggregate/count_values.go b/execution/aggregate/count_values.go
@@ -49,7 +49,7 @@ func NewCountValues(next model.VectorOperator, param string, by bool, grouping [
 		by:         by,
 		grouping:   grouping,
 	}
-	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op)
+	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op)
 }
 
 func (c *countValuesOperator) Explain() []model.VectorOperator {
diff --git a/execution/aggregate/hashaggregate.go b/execution/aggregate/hashaggregate.go
@@ -69,7 +69,7 @@ func NewHashAggregate(
 		params:      make([]float64, opts.StepsBatch),
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(a, opts), a), nil
+	return telemetry.NewOperator(telemetry.NewTelemetry(a, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), a), nil
 }
 
 func (a *aggregate) String() string {
diff --git a/execution/aggregate/khashaggregate.go b/execution/aggregate/khashaggregate.go
@@ -80,7 +80,7 @@ func NewKHashAggregate(
 		stepsBatch:  opts.StepsBatch,
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op), nil
+	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op), nil
 }
 
 func (a *kAggregate) Next(ctx context.Context, buf []model.StepVector) (int, error) {
diff --git a/execution/binary/scalar.go b/execution/binary/scalar.go
@@ -56,7 +56,7 @@ func NewScalar(
 		stepsBatch: opts.StepsBatch,
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op), nil
+	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op), nil
 }
 
 func (o *scalarOperator) Explain() (next []model.VectorOperator) {
diff --git a/execution/binary/vector.go b/execution/binary/vector.go
@@ -70,7 +70,7 @@ func NewVectorOperator(
 		stepsBatch: opts.StepsBatch,
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op), nil
+	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op), nil
 }
 
 func (o *vectorOperator) String() string {
diff --git a/execution/exchange/coalesce.go b/execution/exchange/coalesce.go
@@ -63,7 +63,7 @@ func NewCoalesce(opts *query.Options, batchSize int64, operators ...model.Vector
 		batchSize:     batchSize,
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (c *coalesce) Explain() (next []model.VectorOperator) {
diff --git a/execution/exchange/concurrent.go b/execution/exchange/concurrent.go
@@ -47,7 +47,7 @@ func NewConcurrent(next model.VectorOperator, bufferSize int, opts *query.Option
 		returnChan: make(chan []model.StepVector, bufferSize+2),
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (c *concurrencyOperator) Explain() (next []model.VectorOperator) {
diff --git a/execution/exchange/dedup.go b/execution/exchange/dedup.go
@@ -44,7 +44,7 @@ func NewDedupOperator(next model.VectorOperator, opts *query.Options) model.Vect
 	oper := &dedupOperator{
 		next: next,
 	}
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (d *dedupOperator) Next(ctx context.Context, buf []model.StepVector) (int, error) {
diff --git a/execution/exchange/duplicate_label.go b/execution/exchange/duplicate_label.go
@@ -29,7 +29,7 @@ func NewDuplicateLabelCheck(next model.VectorOperator, opts *query.Options) mode
 	oper := &duplicateLabelCheckOperator{
 		next: next,
 	}
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (d *duplicateLabelCheckOperator) Next(ctx context.Context, buf []model.StepVector) (int, error) {
diff --git a/execution/execution.go b/execution/execution.go
@@ -50,6 +50,7 @@ func New(ctx context.Context, expr logicalplan.Node, storage storage.Scanners, o
 		End:   opts.End.UnixMilli(),
 		Step:  opts.Step.Milliseconds(),
 	}
+
 	return newOperator(ctx, expr, storage, opts, hints)
 }
 
@@ -211,7 +212,6 @@ func newSubqueryFunction(ctx context.Context, e *logicalplan.FunctionCall, t *lo
 		outerOpts.Start = time.UnixMilli(*t.Timestamp)
 		outerOpts.End = time.UnixMilli(*t.Timestamp)
 	}
-
 	var scalarArg model.VectorOperator
 	var scalarArg2 model.VectorOperator
 	switch e.Func.Name {
@@ -239,6 +239,8 @@ func newSubqueryFunction(ctx context.Context, e *logicalplan.FunctionCall, t *lo
 		}
 	}
 
+	// The outer subquery operator reports samples at parent step timestamps,
+	// so it uses the parent limiter.
 	return scan.NewSubqueryOperator(inner, scalarArg, scalarArg2, &outerOpts, e, t)
 }
 
diff --git a/execution/function/absent.go b/execution/function/absent.go
@@ -31,7 +31,7 @@ func newAbsentOperator(
 		funcExpr: funcExpr,
 		next:     next,
 	}
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (o *absentOperator) String() string {
diff --git a/execution/function/histogram.go b/execution/function/histogram.go
@@ -92,7 +92,7 @@ func newHistogramOperator(
 	default:
 		panic("unsupported function passed")
 	}
-	return telemetry.NewOperator(telemetry.NewTelemetry(o, opts), o)
+	return telemetry.NewOperator(telemetry.NewTelemetry(o, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), o)
 }
 
 func (o *histogramOperator) String() string {
diff --git a/execution/function/operator.go b/execution/function/operator.go
@@ -75,7 +75,7 @@ func newNoArgsFunctionOperator(funcExpr *logicalplan.FunctionCall, stepsBatch in
 		op.sampleIDs = []uint64{0}
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op), nil
+	return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op), nil
 }
 
 // functionOperator returns []model.StepVector after processing input with desired function.
@@ -122,7 +122,7 @@ func newInstantVectorFunctionOperator(funcExpr *logicalplan.FunctionCall, nextOp
 	// Check selector type.
 	switch funcExpr.Args[f.vectorIndex].ReturnType() {
 	case parser.ValueTypeVector, parser.ValueTypeScalar:
-		return telemetry.NewOperator(telemetry.NewTelemetry(f, opts), f), nil
+		return telemetry.NewOperator(telemetry.NewTelemetry(f, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), f), nil
 	default:
 		return nil, errors.Wrapf(parse.ErrNotImplemented, "got %s:", funcExpr.String())
 	}
diff --git a/execution/function/relabel.go b/execution/function/relabel.go
@@ -35,7 +35,7 @@ func newRelabelOperator(
 		next:     next,
 		funcExpr: funcExpr,
 	}
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (o *relabelOperator) String() string {
diff --git a/execution/function/scalar.go b/execution/function/scalar.go
@@ -23,7 +23,7 @@ func newScalarOperator(next model.VectorOperator, opts *query.Options) model.Vec
 		next: next,
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (o *scalarOperator) String() string {
diff --git a/execution/function/timestamp.go b/execution/function/timestamp.go
@@ -26,7 +26,7 @@ func newTimestampOperator(next model.VectorOperator, opts *query.Options) model.
 	oper := &timestampOperator{
 		next: next,
 	}
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (o *timestampOperator) Explain() (next []model.VectorOperator) {
diff --git a/execution/remote/operator.go b/execution/remote/operator.go
@@ -33,17 +33,17 @@ type Execution struct {
 }
 
 func NewExecution(query promql.Query, queryRangeStart, queryRangeEnd time.Time, engineLabels []labels.Labels, opts *query.Options, _ storage.SelectHints) model.VectorOperator {
-	storage := newStorageFromQuery(query, opts, engineLabels)
+	stor := newStorageFromQuery(query, opts, engineLabels)
 	oper := &Execution{
-		storage:         storage,
+		storage:         stor,
 		query:           query,
 		opts:            opts,
 		queryRangeStart: queryRangeStart,
 		queryRangeEnd:   queryRangeEnd,
-		vectorSelector:  promstorage.NewVectorSelector(storage, opts, 0, 0, false, 0, 1),
+		vectorSelector:  promstorage.NewVectorSelector(stor, opts, 0, 0, false, 0, 1),
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (e *Execution) Series(ctx context.Context) ([]labels.Labels, error) {
diff --git a/execution/scan/literal_selector.go b/execution/scan/literal_selector.go
@@ -38,7 +38,7 @@ func NewNumberLiteralSelector(opts *query.Options, val float64) model.VectorOper
 		val:         val,
 	}
 
-	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
+	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)
 }
 
 func (o *numberLiteralSelector) Explain() (next []model.VectorOperator) {
diff --git a/execution/scan/subquery.go b/execution/scan/subquery.go
@@ -20,8 +20,6 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 )
 
-const sampleLimitCheckPercentage = 0.05
-
 type subqueryOperator struct {
 	next      model.VectorOperator
 	paramOp   model.VectorOperator
@@ -54,9 +52,6 @@ type subqueryOperator struct {
 	paramBuf  []model.StepVector
 	param2Buf []model.StepVector
 	tempBuf   []model.StepVector
-
-	currentTrackedSamples int
-	lastTrackedSamples    int
 }
 
 func NewSubqueryOperator(next, paramOp, paramOp2 model.VectorOperator, opts *query.Options, funcExpr *logicalplan.FunctionCall, subQuery *logicalplan.Subquery) (model.VectorOperator, error) {
@@ -86,7 +81,7 @@ func NewSubqueryOperator(next, paramOp, paramOp2 model.VectorOperator, opts *que
 		params:        make([]float64, opts.StepsBatch),
 		params2:       make([]float64, opts.StepsBatch),
 	}
-	o.telemetry = telemetry.NewSubqueryTelemetry(o, opts)
+	o.telemetry = telemetry.NewSubqueryTelemetry(o, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter)
 	return telemetry.NewOperator(o.telemetry, o), nil
 }
 
@@ -155,9 +150,6 @@ func (o *subqueryOperator) Next(ctx context.Context, buf []model.StepVector) (in
 		for _, b := range o.buffers {
 			b.Reset(mint, maxt+o.subQuery.Offset.Milliseconds())
 		}
-		o.currentTrackedSamples = 0
-		o.lastTrackedSamples = 0
-		checkSampleLimitCounter := 0
 		if len(o.lastVectors) > 0 {
 			for _, v := range o.lastVectors[o.lastCollected+1:] {
 				if v.T > maxt {
@@ -167,7 +159,6 @@ func (o *subqueryOperator) Next(ctx context.Context, buf []model.StepVector) (in
 				o.lastCollected++
 			}
 			if o.lastCollected == len(o.lastVectors)-1 {
-
 				o.lastVectors = nil
 				o.lastCollected = -1
 			}
@@ -192,18 +183,6 @@ func (o *subqueryOperator) Next(ctx context.Context, buf []model.StepVector) (in
 				o.collect(vector, mint)
 			}
 
-			checkSampleLimitCounter++
-			if o.shouldCheckSampleLimit(checkSampleLimitCounter) {
-				if err := o.checkSampleLimit(); err != nil {
-					return 0, err
-				}
-				checkSampleLimitCounter = 0
-			}
-		}
-		if checkSampleLimitCounter > 0 {
-			if err := o.checkSampleLimit(); err != nil {
-				return 0, err
-			}
 		}
 
 		buf[n].Reset(o.currentStep)
@@ -221,7 +200,9 @@ func (o *subqueryOperator) Next(ctx context.Context, buf []model.StepVector) (in
 					buf[n].AppendSampleWithSizeHint(uint64(sampleId), f, hint)
 				}
 			}
-			o.telemetry.IncrementSamplesAtTimestamp(rangeSamples.SampleCount(), buf[n].T)
+			if err := o.telemetry.IncrementSamplesAtTimestamp(rangeSamples.SampleCount(), buf[n].T); err != nil {
+				return 0, err
+			}
 		}
 		n++
 		o.currentStep += o.step
@@ -230,15 +211,6 @@ func (o *subqueryOperator) Next(ctx context.Context, buf []model.StepVector) (in
 	return n, nil
 }
 
-func (o *subqueryOperator) checkSampleLimit() error {
-	delta := o.currentTrackedSamples - o.lastTrackedSamples
-	if delta > 0 {
-		o.opts.SampleTracker.Add(delta)
-	}
-	o.lastTrackedSamples = o.currentTrackedSamples
-	return o.opts.SampleTracker.CheckLimit()
-}
-
 func (o *subqueryOperator) collect(v model.StepVector, mint int64) {
 	if v.T < mint {
 		return
@@ -249,7 +221,6 @@ func (o *subqueryOperator) collect(v model.StepVector, mint int64) {
 			continue
 		}
 		buffer.Push(v.T, ringbuffer.Value{F: s})
-		o.currentTrackedSamples++
 	}
 	for i, s := range v.Histograms {
 		buffer := o.buffers[v.HistogramIDs[i]]
@@ -275,9 +246,7 @@ func (o *subqueryOperator) collect(v model.StepVector, mint int64) {
 			s.CounterResetHint = histogram.UnknownCounterReset
 		}
 		buffer.Push(v.T, ringbuffer.Value{H: s})
-		o.currentTrackedSamples += telemetry.CalculateHistogramSampleCount(s)
 	}
-
 }
 
 func (o *subqueryOperator) Series(ctx context.Context) ([]labels.Labels, error) {
@@ -322,21 +291,3 @@ func (o *subqueryOperator) initSeries(ctx context.Context) error {
 	})
 	return err
 }
-
-func (o *subqueryOperator) shouldCheckSampleLimit(checkSampleLimitCounter int) bool {
-	if len(o.series) == 0 {
-		return checkSampleLimitCounter >= 1
-	}
-
-	limit := o.opts.SampleTracker.Limit()
-	targetSamplesPerCheck := int(float64(limit) * sampleLimitCheckPercentage)
-
-	maxSamplesPerCall := len(o.series) * o.stepsBatch
-	if maxSamplesPerCall == 0 {
-		return checkSampleLimitCounter >= 1
-	}
-
-	interval := max(targetSamplesPerCheck/maxSamplesPerCall, 1)
-
-	return checkSampleLimitCounter >= interval
-}
diff --git a/execution/step_invariant/step_invariant.go b/execution/step_invariant/step_invariant.go
@@ -65,7 +65,7 @@ func NewStepInvariantOperator(
 		u.cacheResult = false
 	}
 
-	return telemetry.NewOperator(telemetry.NewStepInvariantTelemetry(u, opts), u), nil
+	return telemetry.NewOperator(telemetry.NewStepInvariantTelemetry(u, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), u), nil
 }
 
 func (u *stepInvariantOperator) Series(ctx context.Context) ([]labels.Labels, error) {
diff --git a/execution/telemetry/limiter.go b/execution/telemetry/limiter.go
diff --git a/execution/telemetry/telemetry.go b/execution/telemetry/telemetry.go
diff --git a/execution/unary/unary.go b/execution/unary/unary.go
diff --git a/query/options.go b/query/options.go
diff --git a/query/sample_tracker.go b/query/sample_tracker.go
diff --git a/query/sample_tracker_test.go b/query/sample_tracker_test.go
diff --git a/storage/prometheus/matrix_selector.go b/storage/prometheus/matrix_selector.go
diff --git a/storage/prometheus/vector_selector.go b/storage/prometheus/vector_selector.go

Original file line number	Diff line number	Diff line change
`@@ -446,7 +446,7 @@ func (e *Engine) makeQueryOpts(start time.Time, end time.Time, step time.Duratio`
`446`	`446`	`EnableAnalysis: e.enableAnalysis,`
`447`	`447`	`NoStepSubqueryIntervalFn: e.noStepSubqueryIntervalFn,`
`448`	`448`	`DecodingConcurrency: e.decodingConcurrency,`
`449`		`- SampleTracker: query.NewSampleTracker(e.maxSamplesPerQuery),`
	`449`	`+ SampleLimiter: telemetry.NewSampleLimiter(e.maxSamplesPerQuery, start.UnixMilli(), end.UnixMilli(), telemetry.StepTrackingInterval(step)),`
`450`	`450`	`}`
`451`	`451`
`452`	`452`	`if opts == nil {`
`@@ -459,7 +459,6 @@ func (e *Engine) makeQueryOpts(start time.Time, end time.Time, step time.Duratio`
`459`	`459`	`if opts.EnablePerStepStats() {`
`460`	`460`	`res.EnablePerStepStats = opts.EnablePerStepStats()`
`461`	`461`	`}`
`462`		`-`
`463`	`462`	`if opts.DecodingConcurrency != 0 {`
`464`	`463`	`res.DecodingConcurrency = opts.DecodingConcurrency`
`465`	`464`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ func NewCountValues(next model.VectorOperator, param string, by bool, grouping [`
`49`	`49`	`by: by,`
`50`	`50`	`grouping: grouping,`
`51`	`51`	`}`
`52`		`- return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op)`
	`52`	`+ return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op)`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`func (c *countValuesOperator) Explain() []model.VectorOperator {`
Original file line number	Diff line number	Diff line change
`@@ -69,7 +69,7 @@ func NewHashAggregate(`
`69`	`69`	`params: make([]float64, opts.StepsBatch),`
`70`	`70`	`}`
`71`	`71`
`72`		`- return telemetry.NewOperator(telemetry.NewTelemetry(a, opts), a), nil`
	`72`	`+ return telemetry.NewOperator(telemetry.NewTelemetry(a, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), a), nil`
`73`	`73`	`}`
`74`	`74`
`75`	`75`	`func (a *aggregate) String() string {`
Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,7 @@ func NewKHashAggregate(`
`80`	`80`	`stepsBatch: opts.StepsBatch,`
`81`	`81`	`}`
`82`	`82`
`83`		`- return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op), nil`
	`83`	`+ return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op), nil`
`84`	`84`	`}`
`85`	`85`
`86`	`86`	`func (a *kAggregate) Next(ctx context.Context, buf []model.StepVector) (int, error) {`
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,7 @@ func NewScalar(`
`56`	`56`	`stepsBatch: opts.StepsBatch,`
`57`	`57`	`}`
`58`	`58`
`59`		`- return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op), nil`
	`59`	`+ return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op), nil`
`60`	`60`	`}`
`61`	`61`
`62`	`62`	`func (o *scalarOperator) Explain() (next []model.VectorOperator) {`
Original file line number	Diff line number	Diff line change
`@@ -70,7 +70,7 @@ func NewVectorOperator(`
`70`	`70`	`stepsBatch: opts.StepsBatch,`
`71`	`71`	`}`
`72`	`72`
`73`		`- return telemetry.NewOperator(telemetry.NewTelemetry(op, opts), op), nil`
	`73`	`+ return telemetry.NewOperator(telemetry.NewTelemetry(op, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), op), nil`
`74`	`74`	`}`
`75`	`75`
`76`	`76`	`func (o *vectorOperator) String() string {`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ func NewCoalesce(opts *query.Options, batchSize int64, operators ...model.Vector`
`63`	`63`	`batchSize: batchSize,`
`64`	`64`	`}`
`65`	`65`
`66`		`- return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)`
	`66`	`+ return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)`
`67`	`67`	`}`
`68`	`68`
`69`	`69`	`func (c *coalesce) Explain() (next []model.VectorOperator) {`
Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ func NewConcurrent(next model.VectorOperator, bufferSize int, opts *query.Option`
`47`	`47`	`returnChan: make(chan []model.StepVector, bufferSize+2),`
`48`	`48`	`}`
`49`	`49`
`50`		`- return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)`
	`50`	`+ return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`func (c *concurrencyOperator) Explain() (next []model.VectorOperator) {`
Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ func NewDedupOperator(next model.VectorOperator, opts *query.Options) model.Vect`
`44`	`44`	`oper := &dedupOperator{`
`45`	`45`	`next: next,`
`46`	`46`	`}`
`47`		`- return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)`
	`47`	`+ return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts.EnableAnalysis, opts.EnablePerStepStats, opts.Start.UnixMilli(), opts.End.UnixMilli(), opts.Step, opts.SampleLimiter), oper)`
`48`	`48`	`}`
`49`	`49`
`50`	`50`	`func (d *dedupOperator) Next(ctx context.Context, buf []model.StepVector) (int, error) {`