Merge branch 'main' into tvats-breakdown-query-range-into-smaller-units

pkg/querier/querier.go

@@ -80,244 +80,53 @@ func New(
 	}
 }
 
-// extractShiftFromBuilderQuery extracts the shift value from timeShift function if present.
-func extractShiftFromBuilderQuery[T any](spec qbtypes.QueryBuilderQuery[T]) int64 {
-	for _, fn := range spec.Functions {
-		if fn.Name == qbtypes.FunctionNameTimeShift && len(fn.Args) > 0 {
-			switch v := fn.Args[0].Value.(type) {
-			case float64:
-				return int64(v)
-			case int64:
-				return v
-			case int:
-				return int64(v)
-			case string:
-				if shiftFloat, err := strconv.ParseFloat(v, 64); err == nil {
-					return int64(shiftFloat)
-				}
-			}
-		}
-	}
-	return 0
-}
-
-// adjustTimeRangeForShift adjusts the time range based on the shift value from timeShift function.
-func adjustTimeRangeForShift[T any](spec qbtypes.QueryBuilderQuery[T], tr qbtypes.TimeRange, kind qbtypes.RequestType) qbtypes.TimeRange {
-	// Only apply time shift for time series and scalar queries
-	// Raw/list queries don't support timeshift
-	if kind != qbtypes.RequestTypeTimeSeries && kind != qbtypes.RequestTypeScalar {
-		return tr
-	}
-
-	// Use the ShiftBy field if it's already populated, otherwise extract it
-	shiftBy := spec.ShiftBy
-	if shiftBy == 0 {
-		shiftBy = extractShiftFromBuilderQuery(spec)
-	}
-
-	if shiftBy == 0 {
-		return tr
-	}
-
-	// ShiftBy is in seconds, convert to milliseconds and shift backward in time
-	shiftMS := shiftBy * 1000
-	return qbtypes.TimeRange{
-		From: tr.From - uint64(shiftMS),
-		To:   tr.To - uint64(shiftMS),
-	}
-}
-
 func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error) {
 
 	tmplVars := req.Variables
 	if tmplVars == nil {
 		tmplVars = make(map[string]qbtypes.VariableItem)
 	}
+
 	event := &qbtypes.QBEvent{
 		Version:         "v5",
 		NumberOfQueries: len(req.CompositeQuery.Queries),
 		PanelType:       req.RequestType.StringValue(),
 	}
-	intervalWarnings := []string{}
+	q.populateQBEvent(event, req.CompositeQuery.Queries)
 
-	dependencyQueries := make(map[string]bool)
-	traceOperatorQueries := make(map[string]qbtypes.QueryBuilderTraceOperator)
-
-	for _, query := range req.CompositeQuery.Queries {
-		if query.Type == qbtypes.QueryTypeTraceOperator {
-			if spec, ok := query.Spec.(qbtypes.QueryBuilderTraceOperator); ok {
-				// Parse expression to find dependencies
-				if err := spec.ParseExpression(); err != nil {
-					return nil, err
-				}
-
-				deps := spec.CollectReferencedQueries(spec.ParsedExpression)
-				for _, dep := range deps {
-					dependencyQueries[dep] = true
-				}
-				traceOperatorQueries[spec.Name] = spec
-			}
-		}
+	// TraceOperatorQuery leverages other queries defined in the rangeRequest
+	// Eg: C := A => B
+	// Need to create dependency map { "A": true, "B": true }
+	dependencyQueries, err := q.constructTraceOperatorDependencyMap(req.CompositeQuery.Queries)
+	if err != nil {
+		return nil, err
 	}
 
-	// First pass: collect all metric names that need temporality
-	metricNames := make([]string, 0)
-	for idx, query := range req.CompositeQuery.Queries {
-		event.QueryType = query.Type.StringValue()
-		switch query.Type {
-		case qbtypes.QueryTypeBuilder:
-			if spec, ok := query.Spec.(qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]); ok {
-				for _, agg := range spec.Aggregations {
-					if agg.MetricName != "" {
-						metricNames = append(metricNames, agg.MetricName)
-					}
-				}
-			}
-			// if step interval is not set, we set it ourselves with recommended value
-			// if step interval is set to value which could result in points more than
-			// allowed, we override it.
-			switch spec := query.Spec.(type) {
-			case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
-				event.TracesUsed = true
-				event.FilterApplied = spec.Filter != nil && spec.Filter.Expression != ""
-				event.GroupByApplied = len(spec.GroupBy) > 0
-				if spec.StepInterval.Seconds() == 0 {
-					spec.StepInterval = qbtypes.Step{
-						Duration: time.Second * time.Duration(querybuilder.RecommendedStepInterval(req.Start, req.End)),
-					}
-				}
-				if spec.StepInterval.Seconds() < float64(querybuilder.MinAllowedStepInterval(req.Start, req.End)) {
-					newStep := qbtypes.Step{
-						Duration: time.Second * time.Duration(querybuilder.MinAllowedStepInterval(req.Start, req.End)),
-					}
-					intervalWarnings = append(intervalWarnings, fmt.Sprintf(intervalWarn, spec.Name, spec.StepInterval.Seconds(), newStep.Seconds()))
-					spec.StepInterval = newStep
-				}
-				req.CompositeQuery.Queries[idx].Spec = spec
-			case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
-				event.LogsUsed = true
-				event.FilterApplied = spec.Filter != nil && spec.Filter.Expression != ""
-				event.GroupByApplied = len(spec.GroupBy) > 0
-				if spec.StepInterval.Seconds() == 0 {
-					spec.StepInterval = qbtypes.Step{
-						Duration: time.Second * time.Duration(querybuilder.RecommendedStepInterval(req.Start, req.End)),
-					}
-				}
-				if spec.StepInterval.Seconds() < float64(querybuilder.MinAllowedStepInterval(req.Start, req.End)) {
-					newStep := qbtypes.Step{
-						Duration: time.Second * time.Duration(querybuilder.MinAllowedStepInterval(req.Start, req.End)),
-					}
-					intervalWarnings = append(intervalWarnings, fmt.Sprintf(intervalWarn, spec.Name, spec.StepInterval.Seconds(), newStep.Seconds()))
-					spec.StepInterval = newStep
-				}
-				req.CompositeQuery.Queries[idx].Spec = spec
-			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
-				event.MetricsUsed = true
-				event.FilterApplied = spec.Filter != nil && spec.Filter.Expression != ""
-				event.GroupByApplied = len(spec.GroupBy) > 0
-
-				if spec.Source == telemetrytypes.SourceMeter {
-					if spec.StepInterval.Seconds() == 0 {
-						spec.StepInterval = qbtypes.Step{Duration: time.Second * time.Duration(querybuilder.RecommendedStepIntervalForMeter(req.Start, req.End))}
-					}
-
-					if spec.StepInterval.Seconds() < float64(querybuilder.MinAllowedStepIntervalForMeter(req.Start, req.End)) {
-						newStep := qbtypes.Step{
-							Duration: time.Second * time.Duration(querybuilder.MinAllowedStepIntervalForMeter(req.Start, req.End)),
-						}
-						spec.StepInterval = newStep
-					}
-				} else {
-					if spec.StepInterval.Seconds() == 0 {
-						spec.StepInterval = qbtypes.Step{
-							Duration: time.Second * time.Duration(querybuilder.RecommendedStepIntervalForMetric(req.Start, req.End)),
-						}
-					}
-					if spec.StepInterval.Seconds() < float64(querybuilder.MinAllowedStepIntervalForMetric(req.Start, req.End)) {
-						newStep := qbtypes.Step{
-							Duration: time.Second * time.Duration(querybuilder.MinAllowedStepIntervalForMetric(req.Start, req.End)),
-						}
-						intervalWarnings = append(intervalWarnings, fmt.Sprintf(intervalWarn, spec.Name, spec.StepInterval.Seconds(), newStep.Seconds()))
-						spec.StepInterval = newStep
-					}
-				}
-				req.CompositeQuery.Queries[idx].Spec = spec
-			}
-		case qbtypes.QueryTypePromQL:
-			event.MetricsUsed = true
-			switch spec := query.Spec.(type) {
-			case qbtypes.PromQuery:
-				if spec.Step.Seconds() == 0 {
-					spec.Step = qbtypes.Step{
-						Duration: time.Second * time.Duration(querybuilder.RecommendedStepIntervalForMetric(req.Start, req.End)),
-					}
-				}
-				req.CompositeQuery.Queries[idx].Spec = spec
-			}
-		case qbtypes.QueryTypeClickHouseSQL:
-			switch spec := query.Spec.(type) {
-			case qbtypes.ClickHouseQuery:
-				if strings.TrimSpace(spec.Query) != "" {
-					event.MetricsUsed = strings.Contains(spec.Query, "signoz_metrics")
-					event.LogsUsed = strings.Contains(spec.Query, "signoz_logs")
-					event.TracesUsed = strings.Contains(spec.Query, "signoz_traces")
-				}
-			}
-		case qbtypes.QueryTypeTraceOperator:
-			if spec, ok := query.Spec.(qbtypes.QueryBuilderTraceOperator); ok {
-				if spec.StepInterval.Seconds() == 0 {
-					spec.StepInterval = qbtypes.Step{
-						Duration: time.Second * time.Duration(querybuilder.RecommendedStepInterval(req.Start, req.End)),
-					}
-				}
-
-				if spec.StepInterval.Seconds() < float64(querybuilder.MinAllowedStepInterval(req.Start, req.End)) {
-					newStep := qbtypes.Step{
-						Duration: time.Second * time.Duration(querybuilder.MinAllowedStepInterval(req.Start, req.End)),
-					}
-					intervalWarnings = append(intervalWarnings, fmt.Sprintf(intervalWarn, spec.Name, spec.StepInterval.Seconds(), newStep.Seconds()))
-					spec.StepInterval = newStep
-				}
-				req.CompositeQuery.Queries[idx].Spec = spec
-			}
-		}
-	}
+	// Step interval is the aggregation parameter for timeseries requests.
+	// We need to set if it is unspecified or adjust it if value is not within recommended range
+	intervalWarnings := q.adjustStepInterval(req.CompositeQuery.Queries, req.Start, req.End)
 
 	queries := make(map[string]qbtypes.Query)
 	steps := make(map[string]qbtypes.Step)
-	missingMetrics := []string{}
-	missingMetricQueries := []string{}
+
+	// Resolve metric metadata once per request: patches each metric-aggregation
+	// query's spec in place, returns the queries whose every aggregation was
+	// missing (used for preseeded empty results), and any dormant-metric
+	// warning string. NotFound errors for never-seen metrics are propagated.
+	missingMetricQueries, dormantMetricsWarningMsg, err := q.resolveMetricMetadata(ctx, req.CompositeQuery.Queries, req.Start, req.End)
+	if err != nil {
+		return nil, err
+	}
+	missingMetricQuerySet := make(map[string]bool, len(missingMetricQueries))
+	for _, name := range missingMetricQueries {
+		missingMetricQuerySet[name] = true
+	}
 
 	for _, query := range req.CompositeQuery.Queries {
-		var queryName string
-		var isTraceOperator bool
+		queryName := query.GetQueryName()
 
-		switch query.Type {
-		case qbtypes.QueryTypeTraceOperator:
-			if spec, ok := query.Spec.(qbtypes.QueryBuilderTraceOperator); ok {
-				queryName = spec.Name
-				isTraceOperator = true
-			}
-		case qbtypes.QueryTypePromQL:
-			if spec, ok := query.Spec.(qbtypes.PromQuery); ok {
-				queryName = spec.Name
-			}
-		case qbtypes.QueryTypeClickHouseSQL:
-			if spec, ok := query.Spec.(qbtypes.ClickHouseQuery); ok {
-				queryName = spec.Name
-			}
-		case qbtypes.QueryTypeBuilder:
-			switch spec := query.Spec.(type) {
-			case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
-				queryName = spec.Name
-			case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
-				queryName = spec.Name
-			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
-				queryName = spec.Name
-			}
-		}
-
-		if !isTraceOperator && dependencyQueries[queryName] {
+		// skip if it is dependecy of traceOperatorQuery
+		if query.GetType() != qbtypes.QueryTypeTraceOperator && dependencyQueries[queryName] {
 			continue
 		}
 
@@ -372,40 +181,13 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 				queries[spec.Name] = bq
 				steps[spec.Name] = spec.StepInterval
 			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
-				var metricTemporality map[string]metrictypes.Temporality
-				var metricTypes map[string]metrictypes.Type
-				if len(metricNames) > 0 {
-					var err error
-					metricTemporality, metricTypes, err = q.metadataStore.FetchTemporalityAndTypeMulti(ctx, req.Start, req.End, metricNames...)
-					if err != nil {
-						q.logger.WarnContext(ctx, "failed to fetch metric temporality", errors.Attr(err), slog.Any("metrics", metricNames))
-						return nil, errors.NewInternalf(errors.CodeInternal, "failed to fetch metrics temporality")
-					}
-					q.logger.DebugContext(ctx, "fetched metric temporalities and types", slog.Any("metric_temporality", metricTemporality), slog.Any("metric_types", metricTypes))
-				}
-				presentAggregations := []qbtypes.MetricAggregation{}
-				for i := range spec.Aggregations {
-					if spec.Aggregations[i].MetricName != "" && spec.Aggregations[i].Temporality == metrictypes.Unknown {
-						if temp, ok := metricTemporality[spec.Aggregations[i].MetricName]; ok && temp != metrictypes.Unknown {
-							spec.Aggregations[i].Temporality = temp
-						}
-					}
-					if spec.Aggregations[i].MetricName != "" && spec.Aggregations[i].Type == metrictypes.UnspecifiedType {
-						if foundMetricType, ok := metricTypes[spec.Aggregations[i].MetricName]; ok && foundMetricType != metrictypes.UnspecifiedType {
-							spec.Aggregations[i].Type = foundMetricType
-						}
-					}
-					if spec.Aggregations[i].Type == metrictypes.UnspecifiedType {
-						missingMetrics = append(missingMetrics, spec.Aggregations[i].MetricName)
-						continue
-					}
-					presentAggregations = append(presentAggregations, spec.Aggregations[i])
-				}
-				if len(presentAggregations) == 0 {
-					missingMetricQueries = append(missingMetricQueries, spec.Name)
+				// Spec was already patched by resolveMetricMetadata. Queries
+				// whose every aggregation was missing live in
+				// missingMetricQuerySet and produce empty preseeded results
+				// rather than running here.
+				if missingMetricQuerySet[spec.Name] {
 					continue
 				}
-				spec.Aggregations = presentAggregations
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
 				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
 				var bq *builderQuery[qbtypes.MetricAggregation]
@@ -424,38 +206,6 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 			}
 		}
 	}
-	nonExistentMetrics := []string{}
-	var dormantMetricsWarningMsg string
-	if len(missingMetrics) > 0 {
-		lastSeenInfo, _ := q.metadataStore.FetchLastSeenInfoMulti(ctx, missingMetrics...)
-		for _, missingMetricName := range missingMetrics {
-			if ts, ok := lastSeenInfo[missingMetricName]; ok && ts > 0 {
-				continue
-			}
-			nonExistentMetrics = append(nonExistentMetrics, missingMetricName)
-		}
-		if len(nonExistentMetrics) == 1 {
-			return nil, errors.NewNotFoundf(errors.CodeNotFound, "could not find the metric %s", nonExistentMetrics[0])
-		} else if len(nonExistentMetrics) > 1 {
-			return nil, errors.NewNotFoundf(errors.CodeNotFound, "the following metrics were not found: %s", strings.Join(nonExistentMetrics, ", "))
-		}
-		lastSeenStr := func(name string) string {
-			if ts, ok := lastSeenInfo[name]; ok && ts > 0 {
-				ago := humanize.RelTime(time.UnixMilli(ts), time.Now(), "ago", "from now")
-				return fmt.Sprintf("%s (last seen %s)", name, ago)
-			}
-			return name // this case won't come cuz lastSeenStr is never called for metrics in nonExistentMetrics
-		}
-		if len(missingMetrics) == 1 {
-			dormantMetricsWarningMsg = fmt.Sprintf("no data found for the metric %s in the query time range", lastSeenStr(missingMetrics[0]))
-		} else {
-			parts := make([]string, len(missingMetrics))
-			for i, m := range missingMetrics {
-				parts[i] = lastSeenStr(m)
-			}
-			dormantMetricsWarningMsg = fmt.Sprintf("no data found for the following metrics in the query time range: %s", strings.Join(parts, ", "))
-		}
-	}
 	preseededResults := make(map[string]any)
 	for _, name := range missingMetricQueries { // at this point missing metrics will not have any non existent metrics, only normal ones
 		switch req.RequestType {
@@ -492,6 +242,166 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 	return qbResp, qbErr
 }
 
+func (q *querier) populateQBEvent(event *qbtypes.QBEvent, queries []qbtypes.QueryEnvelope) {
+	for _, query := range queries {
+		// BUG: QueryType doesn't make sense as range_request can have multiple query types.
+		event.QueryType = query.Type.StringValue()
+
+		switch query.Type {
+		case qbtypes.QueryTypeBuilder:
+			filter := query.GetFilter()
+			event.FilterApplied = event.FilterApplied || (filter != nil && filter.Expression != "")
+			event.GroupByApplied = event.GroupByApplied || len(query.GetGroupBy()) > 0
+			switch query.Spec.(type) {
+			case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
+				event.TracesUsed = true
+			case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
+				event.LogsUsed = true
+			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
+				event.MetricsUsed = true
+			}
+		case qbtypes.QueryTypePromQL:
+			event.MetricsUsed = true
+		case qbtypes.QueryTypeTraceOperator:
+			event.TracesUsed = true
+		case qbtypes.QueryTypeClickHouseSQL:
+			sql := query.GetQuery()
+			if strings.TrimSpace(sql) != "" {
+				event.MetricsUsed = strings.Contains(sql, "signoz_metrics")
+				event.LogsUsed = strings.Contains(sql, "signoz_logs")
+				event.TracesUsed = strings.Contains(sql, "signoz_traces")
+			}
+		}
+	}
+}
+
+// resolveMetricMetadata fetches metadata for every metric referenced by builder
+// metric-aggregation queries, patches each query's aggregations in place with
+// the resolved values, and classifies any metric that could not be resolved.
+//
+// Side effects on queries:
+//   - Aggregations with Unknown Temporality / UnspecifiedType are filled in from
+//     the metadata store.
+//   - Aggregations whose Type is still UnspecifiedType after the patch are
+//     dropped from the spec.
+//   - Queries whose entire aggregation list was dropped are NOT patched and are
+//     surfaced via the returned missingMetricQueries; the caller should skip
+//     them.
+//
+// Returns:
+//   - missingMetricQueries: names of queries whose every aggregation was
+//     missing. Used downstream to preseed empty result placeholders so the
+//     response still has an entry per requested query name.
+//   - dormantWarning: a human-readable warning describing metrics that exist in
+//     the store but produced no data within the query window. Empty when no
+//     such metrics are present.
+//   - err: NotFound when one or more referenced metrics have never been seen,
+//     or Internal when a metadata fetch fails.
+func (q *querier) resolveMetricMetadata(ctx context.Context, queries []qbtypes.QueryEnvelope, start, end uint64) (missingMetricQueries []string, dormantWarning string, err error) {
+	metricNames := make([]string, 0)
+	for idx := range queries {
+		if queries[idx].Type != qbtypes.QueryTypeBuilder {
+			continue
+		}
+		spec, ok := queries[idx].Spec.(qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation])
+		if !ok {
+			continue
+		}
+		for _, agg := range spec.Aggregations {
+			if agg.MetricName != "" {
+				metricNames = append(metricNames, agg.MetricName)
+			}
+		}
+	}
+
+	if len(metricNames) == 0 {
+		return nil, "", nil
+	}
+
+	metricTemporality, metricTypes, err := q.metadataStore.FetchTemporalityAndTypeMulti(ctx, start, end, metricNames...)
+	if err != nil {
+		q.logger.WarnContext(ctx, "failed to fetch metric temporality", errors.Attr(err), slog.Any("metrics", metricNames))
+		return nil, "", errors.NewInternalf(errors.CodeInternal, "failed to fetch metrics temporality")
+	}
+	q.logger.DebugContext(ctx, "fetched metric temporalities and types", slog.Any("metric_temporality", metricTemporality), slog.Any("metric_types", metricTypes))
+
+	missingMetrics := []string{}
+	for idx := range queries {
+		if queries[idx].Type != qbtypes.QueryTypeBuilder {
+			continue
+		}
+		spec, ok := queries[idx].Spec.(qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation])
+		if !ok {
+			continue
+		}
+
+		presentAggregations := make([]qbtypes.MetricAggregation, 0, len(spec.Aggregations))
+		for i := range spec.Aggregations {
+			if spec.Aggregations[i].MetricName != "" && spec.Aggregations[i].Temporality == metrictypes.Unknown {
+				if temp, ok := metricTemporality[spec.Aggregations[i].MetricName]; ok && temp != metrictypes.Unknown {
+					spec.Aggregations[i].Temporality = temp
+				}
+			}
+			if spec.Aggregations[i].MetricName != "" && spec.Aggregations[i].Type == metrictypes.UnspecifiedType {
+				if foundMetricType, ok := metricTypes[spec.Aggregations[i].MetricName]; ok && foundMetricType != metrictypes.UnspecifiedType {
+					spec.Aggregations[i].Type = foundMetricType
+				}
+			}
+			if spec.Aggregations[i].Type == metrictypes.UnspecifiedType {
+				missingMetrics = append(missingMetrics, spec.Aggregations[i].MetricName)
+				continue
+			}
+			presentAggregations = append(presentAggregations, spec.Aggregations[i])
+		}
+		if len(presentAggregations) == 0 {
+			missingMetricQueries = append(missingMetricQueries, spec.Name)
+			continue
+		}
+		spec.Aggregations = presentAggregations
+		queries[idx].Spec = spec
+	}
+
+	if len(missingMetrics) == 0 {
+		return missingMetricQueries, "", nil
+	}
+
+	// Classify each missing metric: never-seen → NotFound error; seen-but-no-
+	// data-in-window → dormant warning.
+	lastSeenInfo, _ := q.metadataStore.FetchLastSeenInfoMulti(ctx, missingMetrics...)
+	nonExistentMetrics := []string{}
+	for _, name := range missingMetrics {
+		if ts, ok := lastSeenInfo[name]; ok && ts > 0 {
+			continue
+		}
+		nonExistentMetrics = append(nonExistentMetrics, name)
+	}
+	if len(nonExistentMetrics) == 1 {
+		return nil, "", errors.NewNotFoundf(errors.CodeNotFound, "could not find the metric %s", nonExistentMetrics[0])
+	}
+	if len(nonExistentMetrics) > 1 {
+		return nil, "", errors.NewNotFoundf(errors.CodeNotFound, "the following metrics were not found: %s", strings.Join(nonExistentMetrics, ", "))
+	}
+
+	// All missing metrics are dormant — assemble the warning string.
+	lastSeenStr := func(name string) string {
+		if ts, ok := lastSeenInfo[name]; ok && ts > 0 {
+			ago := humanize.RelTime(time.UnixMilli(ts), time.Now(), "ago", "from now")
+			return fmt.Sprintf("%s (last seen %s)", name, ago)
+		}
+		return name
+	}
+	if len(missingMetrics) == 1 {
+		dormantWarning = fmt.Sprintf("no data found for the metric %s in the query time range", lastSeenStr(missingMetrics[0]))
+	} else {
+		parts := make([]string, len(missingMetrics))
+		for i, m := range missingMetrics {
+			parts[i] = lastSeenStr(m)
+		}
+		dormantWarning = fmt.Sprintf("no data found for the following metrics in the query time range: %s", strings.Join(parts, ", "))
+	}
+	return missingMetricQueries, dormantWarning, nil
+}
+
 func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, client *qbtypes.RawStream) {
 
 	event := &qbtypes.QBEvent{
@@ -1089,3 +999,129 @@ func (q *querier) mergeTimeSeriesResults(cachedValue *qbtypes.TimeSeriesData, fr
 
 	return result
 }
+
+func secondsStep(s uint64) qbtypes.Step {
+	return qbtypes.Step{Duration: time.Second * time.Duration(s)}
+}
+
+// clampStep sets the step to recommended when zero and clamps to min when below it.
+// When clamped and warn is true, a warning is appended for the user.
+func clampStep(qe *qbtypes.QueryEnvelope, recommended, min uint64, warnings *[]string) {
+	step := qe.GetStepInterval()
+	if step.Seconds() == 0 {
+		step = secondsStep(recommended)
+		qe.SetStepInterval(step)
+	}
+	if step.Seconds() < float64(min) {
+		newStep := secondsStep(min)
+		*warnings = append(*warnings, fmt.Sprintf(intervalWarn, qe.GetQueryName(), step.Seconds(), newStep.Seconds()))
+		qe.SetStepInterval(newStep)
+	}
+}
+
+// extractShiftFromBuilderQuery extracts the shift value from timeShift function if present.
+func extractShiftFromBuilderQuery[T any](spec qbtypes.QueryBuilderQuery[T]) int64 {
+	for _, fn := range spec.Functions {
+		if fn.Name == qbtypes.FunctionNameTimeShift && len(fn.Args) > 0 {
+			switch v := fn.Args[0].Value.(type) {
+			case float64:
+				return int64(v)
+			case int64:
+				return v
+			case int:
+				return int64(v)
+			case string:
+				if shiftFloat, err := strconv.ParseFloat(v, 64); err == nil {
+					return int64(shiftFloat)
+				}
+			}
+		}
+	}
+	return 0
+}
+
+// adjustTimeRangeForShift adjusts the time range based on the shift value from timeShift function.
+func adjustTimeRangeForShift[T any](spec qbtypes.QueryBuilderQuery[T], tr qbtypes.TimeRange, kind qbtypes.RequestType) qbtypes.TimeRange {
+	// Only apply time shift for time series and scalar queries
+	// Raw/list queries don't support timeshift
+	if kind != qbtypes.RequestTypeTimeSeries && kind != qbtypes.RequestTypeScalar {
+		return tr
+	}
+
+	// Use the ShiftBy field if it's already populated, otherwise extract it
+	shiftBy := spec.ShiftBy
+	if shiftBy == 0 {
+		shiftBy = extractShiftFromBuilderQuery(spec)
+	}
+
+	if shiftBy == 0 {
+		return tr
+	}
+
+	// ShiftBy is in seconds, convert to milliseconds and shift backward in time
+	shiftMS := shiftBy * 1000
+	return qbtypes.TimeRange{
+		From: tr.From - uint64(shiftMS),
+		To:   tr.To - uint64(shiftMS),
+	}
+}
+
+func (q *querier) constructTraceOperatorDependencyMap(queries []qbtypes.QueryEnvelope) (map[string]bool, error) {
+	dependencyQueries := make(map[string]bool)
+
+	for _, query := range queries {
+		if query.Type == qbtypes.QueryTypeTraceOperator {
+			if spec, ok := query.Spec.(qbtypes.QueryBuilderTraceOperator); ok {
+				// Parse expression to find dependencies
+				if err := spec.ParseExpression(); err != nil {
+					return nil, err
+				}
+
+				deps := spec.CollectReferencedQueries(spec.ParsedExpression)
+				for _, dep := range deps {
+					dependencyQueries[dep] = true
+				}
+			}
+		}
+	}
+
+	return dependencyQueries, nil
+}
+
+// adjustStepInterval normalizes each query's step interval in place and returns
+// any clamp warnings emitted along the way.
+func (q *querier) adjustStepInterval(queries []qbtypes.QueryEnvelope, start, end uint64) []string {
+	// Compute the per-signal bounds once per call — they only depend on start/end.
+	traceLogRecommended := querybuilder.RecommendedStepInterval(start, end)
+	traceLogMin := querybuilder.MinAllowedStepInterval(start, end)
+	meterRecommended := querybuilder.RecommendedStepIntervalForMeter(start, end)
+	meterMin := querybuilder.MinAllowedStepIntervalForMeter(start, end)
+	metricRecommended := querybuilder.RecommendedStepIntervalForMetric(start, end)
+	metricMin := querybuilder.MinAllowedStepIntervalForMetric(start, end)
+
+	warnings := make([]string, 0)
+	for idx := range queries {
+		qe := &queries[idx]
+		switch qe.Type {
+		case qbtypes.QueryTypeBuilder:
+			switch qe.Spec.(type) {
+			case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
+				clampStep(qe, traceLogRecommended, traceLogMin, &warnings)
+			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
+				if qe.GetSource() == telemetrytypes.SourceMeter {
+					clampStep(qe, meterRecommended, meterMin, &warnings)
+				} else {
+					clampStep(qe, metricRecommended, metricMin, &warnings)
+				}
+			}
+		case qbtypes.QueryTypePromQL:
+			// PromQL only fills an unset step — no min clamp.
+			if qe.GetStepInterval().Seconds() == 0 {
+				qe.SetStepInterval(secondsStep(metricRecommended))
+			}
+		case qbtypes.QueryTypeTraceOperator:
+			clampStep(qe, traceLogRecommended, traceLogMin, &warnings)
+		}
+	}
+	return warnings
+}