chore(tests): name the query builders by what they build and make the request type explicit

build_scalar_query built a builder query envelope and did not decide the
request type; make_query_request did, with a silent time_series default.

Assisted-by: Claude Fable 5.1

pkg/querier/bucket_cache.go

@@ -915,13 +915,15 @@ func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, q qbtypes
 // as empty rows that the same request never has when it bypasses the cache.
 //
 // PromQL and ClickHouse queries name their own buckets, and a histogram's are
-// its `le` labels, so only a gauge, sum or summary axis is cut.
+// its `le` labels, so only a gauge, sum or summary axis and an event axis are cut.
 func (bc *bucketCache) trimHeatmapAxisToTheWindow(q qbtypes.Query, tsData *qbtypes.TimeSeriesData) {
-	metricQuery, isMetricQuery := q.(*builderQuery[qbtypes.MetricAggregation])
-	if !isMetricQuery {
-		return
-	}
-	if metricQuery.spec.Aggregations[0].Type == metrictypes.HistogramType {
+	switch query := q.(type) {
+	case *builderQuery[qbtypes.MetricAggregation]:
+		if query.spec.Aggregations[0].Type == metrictypes.HistogramType {
+			return
+		}
+	case *builderQuery[qbtypes.LogAggregation], *builderQuery[qbtypes.TraceAggregation]:
+	default:
 		return
 	}
 

pkg/querier/builder_query.go

@@ -77,7 +77,7 @@ func newBuilderQuery[T any](
 func (q *builderQuery[T]) Fingerprint() string {
 
 	if (q.spec.Signal == telemetrytypes.SignalTraces ||
-		q.spec.Signal == telemetrytypes.SignalLogs) && q.kind != qbtypes.RequestTypeTimeSeries {
+		q.spec.Signal == telemetrytypes.SignalLogs) && !q.kind.IsBuiltFromTimeSeries() {
 		// No caching for non-timeseries queries
 		return ""
 	}
@@ -150,6 +150,12 @@ func (q *builderQuery[T]) Fingerprint() string {
 		parts = append(parts, fmt.Sprintf("aggs=[%s]", strings.Join(aggParts, ",")))
 	}
 
+	// A metrics heatmap carries its axis on the aggregation, which the loop above
+	// already fingerprinted. Logs and traces carry it on the spec.
+	if q.kind.CountsIntoBuckets() && q.spec.Signal != telemetrytypes.SignalMetrics {
+		parts = append(parts, "buckets="+fingerprintHeatmapBucketing(q.spec.BucketOptions.ToHeatmapBucketing()))
+	}
+
 	// Add filter if present
 	if q.spec.Filter != nil && q.spec.Filter.Expression != "" {
 		parts = append(parts, fmt.Sprintf("filter=%s", q.spec.Filter.Expression))

pkg/querier/postprocess.go

@@ -191,6 +191,14 @@ func postProcessBuilderQuery[T any](
 	req *qbtypes.QueryRangeRequest,
 ) *qbtypes.Result {
 
+	// A disabled query feeds a formula as a time series, so only an enabled one
+	// came back bucketed.
+	if req.RequestType.CountsIntoBuckets() && !query.Disabled {
+		if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok {
+			resolveHeatmapBucketAxis(tsData, query.BucketOptions.ToHeatmapBucketing())
+		}
+	}
+
 	// the statement's top-N already skipped offset groups by their whole-window value; only reorder here
 	result = q.applySeriesLimit(result, query.Limit, 0, query.Order)
 

pkg/querier/querier.go

@@ -257,27 +257,30 @@ func (q *querier) buildQueries(
 			if !ok {
 				return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid AI builder query spec %T", query.Spec)
 			}
+			requestType := requestTypeForQuery(req.RequestType, spec.Disabled)
 			spec.ShiftBy = extractShiftFromBuilderQuery(spec)
-			timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
-			bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, q.aiTraceStmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
+			timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, requestType)
+			bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, q.aiTraceStmtBuilder, query.Type, spec, timeRange, requestType, tmplVars, builderConfig{})
 			queries[spec.Name] = bq
 			steps[spec.Name] = spec.StepInterval
 		case qbtypes.QueryTypeBuilder:
 			switch spec := query.Spec.(type) {
 			case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
+				requestType := requestTypeForQuery(req.RequestType, spec.Disabled)
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
-				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
-				bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, q.traceStmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
+				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, requestType)
+				bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, q.traceStmtBuilder, query.Type, spec, timeRange, requestType, tmplVars, builderConfig{})
 				queries[spec.Name] = bq
 				steps[spec.Name] = spec.StepInterval
 			case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
+				requestType := requestTypeForQuery(req.RequestType, spec.Disabled)
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
-				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
+				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, requestType)
 				stmtBuilder := q.logStmtBuilder
 				if spec.Source == telemetrytypes.SourceAudit {
 					stmtBuilder = q.auditStmtBuilder
 				}
-				bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, stmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, q.builderConfig)
+				bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, stmtBuilder, query.Type, spec, timeRange, requestType, tmplVars, q.builderConfig)
 				queries[spec.Name] = bq
 				steps[spec.Name] = spec.StepInterval
 			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
@@ -288,13 +291,7 @@ func (q *querier) buildQueries(
 				if missingMetricQuerySet[spec.Name] {
 					continue
 				}
-				requestType := req.RequestType
-				if requestType.CountsIntoBuckets() && spec.Disabled {
-					// A disabled query in a heatmap request feeds a formula, and the
-					// formula converts time series into heatmap data, so its inputs
-					// run as time series queries.
-					requestType = qbtypes.RequestTypeTimeSeries
-				}
+				requestType := requestTypeForQuery(req.RequestType, spec.Disabled)
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
 				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, requestType)
 				var bq *builderQuery[qbtypes.MetricAggregation]
@@ -1196,6 +1193,17 @@ func clampStep(qe *qbtypes.QueryEnvelope, recommended, min uint64, warnings *[]s
 	}
 }
 
+// requestTypeForQuery downgrades a disabled query in a heatmap request to a
+// time series one. A disabled query feeds a formula rather than the chart, and
+// the formula is what turns time series into cells, so its inputs state no
+// bucket axis of their own.
+func requestTypeForQuery(requestType qbtypes.RequestType, disabled bool) qbtypes.RequestType {
+	if requestType.CountsIntoBuckets() && disabled {
+		return qbtypes.RequestTypeTimeSeries
+	}
+	return requestType
+}
+
 // extractShiftFromBuilderQuery extracts the shift value from timeShift function if present.
 func extractShiftFromBuilderQuery[T any](spec qbtypes.QueryBuilderQuery[T]) int64 {
 	for _, fn := range spec.Functions {

pkg/querybuilder/heatmap.go

@@ -0,0 +1,136 @@
+package querybuilder
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"strconv"
+
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+// heatmapBucketValueAlias names the value a heatmap's bounds are cut from. The
+// bounds read it several times, so it is declared once as a WITH expression
+// alias rather than repeated.
+const heatmapBucketValueAlias = "__bucket_value"
+
+// RenderHeatmapBucketExprs renders the bucket (min, max] that valueExpr falls
+// in. Only the bucket under everything the axis covers is open below, and only
+// the one over it is open above.
+//
+// valueExpr is repeated several times in each expression, so it has to be free
+// of placeholders — HeatmapBucketColumns passes an alias for that reason.
+func RenderHeatmapBucketExprs(bucketing qbtypes.HeatmapBucketing, valueExpr string) (minExpr, maxExpr string) {
+	if bucketing.Kind == qbtypes.BucketsKindLinear {
+		return renderLinearBucketExprs(bucketing, valueExpr)
+	}
+	return renderLogBucketExprs(valueExpr)
+}
+
+func renderLinearBucketExprs(bucketing qbtypes.HeatmapBucketing, valueExpr string) (string, string) {
+	maxValue := FormatHeatmapFloat(bucketing.MaxValue)
+	numBuckets := strconv.Itoa(bucketing.NumBuckets)
+	index := fmt.Sprintf("least(greatest(ceil(%s * %s / %s), 1), %s)", valueExpr, numBuckets, maxValue, numBuckets)
+
+	minExpr := fmt.Sprintf(
+		"multiIf(%s <= 0, toFloat64('-Inf'), %s > %s, toFloat64(%s), (%s - 1) * %s / %s)",
+		valueExpr, valueExpr, maxValue, maxValue, index, maxValue, numBuckets,
+	)
+	maxExpr := fmt.Sprintf(
+		"multiIf(%s <= 0, toFloat64(0), %s > %s, toFloat64('+Inf'), %s * %s / %s)",
+		valueExpr, valueExpr, maxValue, index, maxValue, numBuckets,
+	)
+	return minExpr, maxExpr
+}
+
+// ClickHouse buckets at MaxLogScale whatever HeatmapBucketing.LogScale asks for;
+// postprocessing folds the axis down afterwards.
+func renderLogBucketExprs(valueExpr string) (string, string) {
+	bucketsPerDoubling := FormatHeatmapFloat(math.Exp2(qbtypes.MaxLogScale))
+	lowest := FormatHeatmapFloat(qbtypes.MinLogUpperBound)
+	highest := FormatHeatmapFloat(qbtypes.MaxLogUpperBound)
+
+	minExpr := fmt.Sprintf(
+		"multiIf(%s <= 0, toFloat64('-Inf'), %s <= %s, toFloat64(0), %s > %s, toFloat64(%s), pow(2, (ceil(log2(%s) * %s) - 1) / %s))",
+		valueExpr, valueExpr, lowest, valueExpr, highest, highest, valueExpr, bucketsPerDoubling, bucketsPerDoubling,
+	)
+	maxExpr := fmt.Sprintf(
+		"multiIf(%s <= 0, toFloat64(0), %s <= %s, %s, %s > %s, toFloat64('+Inf'), pow(2, ceil(log2(%s) * %s) / %s))",
+		valueExpr, valueExpr, lowest, lowest, valueExpr, highest, valueExpr, bucketsPerDoubling, bucketsPerDoubling,
+	)
+	return minExpr, maxExpr
+}
+
+// FormatHeatmapFloat renders a float64 as the shortest literal that reads back
+// as the same value, so an upper bound computed from it is identical on every
+// row.
+func FormatHeatmapFloat(v float64) string {
+	return strconv.FormatFloat(v, 'g', -1, 64)
+}
+
+// HeatmapFieldMayBeAbsent reports whether some row can lack key. A key that
+// resolves to a table column every row carries, such as a span's duration,
+// cannot, as the column is read first, and an EXISTS on it would drop its
+// zeros instead.
+func HeatmapFieldMayBeAbsent(
+	ctx context.Context,
+	info qbtypes.QueryInfo,
+	storage qbtypes.Storage,
+	key *telemetrytypes.TelemetryFieldKey,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+) (bool, error) {
+	resolved, err := Resolve(ctx, info, storage, key, qbtypes.FilterOperatorUnknown, nil, keys)
+	if err != nil {
+		return false, err
+	}
+	for _, field := range resolved.Fields {
+		read, err := LogicalRead(ctx, info, storage, field)
+		if err != nil {
+			return false, err
+		}
+		if read.WhenAbsent == qbtypes.AlwaysPresent {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+// HeatmapBucketColumns adds the bucket a row's value falls in to sb's select
+// list, as the (min, max] pair a heatmap statement reports. key is the field
+// the axis is cut from, already adjusted for context and data type.
+//
+// The bounds read the value through an expression alias rather than repeating
+// it, so the returned fragment has to be declared alongside the caller's CTEs.
+// It carries no arguments of its own.
+//
+// A row without a numeric value counts as 0.
+func HeatmapBucketColumns(
+	ctx context.Context,
+	info qbtypes.QueryInfo,
+	storage qbtypes.Storage,
+	sb *sqlbuilder.SelectBuilder,
+	key *telemetrytypes.TelemetryFieldKey,
+	bucketOptions *qbtypes.BucketOptions,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+) (valueFragment string, groupByAliases []string, warnings []string, err error) {
+	// Resolved directly rather than through ResolveColumn: a key nothing carries
+	// puts every row at 0, and the warning is the only thing that says why.
+	resolved, err := Resolve(ctx, info, storage, key, qbtypes.FilterOperatorUnknown, nil, keys)
+	if err != nil {
+		return "", nil, nil, err
+	}
+	valueExpr, err := Column(ctx, info, storage, resolved, telemetrytypes.FieldDataTypeFloat64)
+	if err != nil {
+		return "", nil, nil, err
+	}
+
+	bucketMin, bucketMax := RenderHeatmapBucketExprs(bucketOptions.ToHeatmapBucketing(), heatmapBucketValueAlias)
+
+	sb.SelectMore(sqlbuilder.Escape(fmt.Sprintf("%s AS %s", bucketMin, qbtypes.HeatmapBucketMinColumn)))
+	sb.SelectMore(sqlbuilder.Escape(fmt.Sprintf("%s AS %s", bucketMax, qbtypes.HeatmapBucketMaxColumn)))
+
+	valueFragment = fmt.Sprintf("ifNull(%s, 0) AS %s", valueExpr, heatmapBucketValueAlias)
+	return valueFragment, []string{qbtypes.HeatmapBucketMinColumn, qbtypes.HeatmapBucketMaxColumn}, resolved.Warnings, nil
+}

pkg/statementbuilder/auditstatementbuilder/statement_builder.go

@@ -110,6 +110,8 @@ func (b *auditQueryStatementBuilder) Build(
 		stmt, err = b.buildListQuery(ctx, orgID, q, query, start, end, keys, variables)
 	case qbtypes.RequestTypeTimeSeries:
 		stmt, err = b.buildTimeSeriesQuery(ctx, orgID, q, query, start, end, keys, variables)
+	case qbtypes.RequestTypeHeatmap, qbtypes.RequestTypeDistribution:
+		stmt, err = b.buildHeatmapTimeSeriesQuery(ctx, orgID, q, query, start, end, keys, variables)
 	case qbtypes.RequestTypeScalar:
 		stmt, err = b.buildScalarQuery(ctx, orgID, q, query, start, end, keys, false, variables)
 	default:
@@ -322,6 +324,48 @@ func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
 	start, end uint64,
 	keys map[string][]*telemetrytypes.TelemetryFieldKey,
 	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	return b.buildTimeSeriesOrHeatmapQuery(ctx, orgID, sb, query, start, end, keys, variables, nil)
+}
+
+// buildHeatmapTimeSeriesQuery counts rows per (time bucket, group, value
+// bucket) of the field its aggregation names.
+func (b *auditQueryStatementBuilder) buildHeatmapTimeSeriesQuery(
+	ctx context.Context,
+	orgID valuer.UUID,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	heatmapField, countQuery, err := qbtypes.HeatmapCountQuery(query)
+	if err != nil {
+		return nil, err
+	}
+	info := querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalLogs, nil, start, end)
+	mayBeAbsent, err := querybuilder.HeatmapFieldMayBeAbsent(ctx, info, b.storage, heatmapField, keys)
+	if err != nil {
+		return nil, err
+	}
+	if mayBeAbsent {
+		countQuery = qbtypes.RequireHeatmapField(countQuery, *heatmapField)
+	}
+	return b.buildTimeSeriesOrHeatmapQuery(ctx, orgID, sb, countQuery, start, end, keys, variables, heatmapField)
+}
+
+// buildTimeSeriesOrHeatmapQuery builds one row per (time bucket, group). A
+// non-nil heatmapField makes it a heatmap instead: the bucket a row's
+// heatmapField falls in joins the group keys, so a row is one cell.
+func (b *auditQueryStatementBuilder) buildTimeSeriesOrHeatmapQuery(
+	ctx context.Context,
+	orgID valuer.UUID,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+	heatmapField *telemetrytypes.TelemetryFieldKey,
 ) (*qbtypes.Statement, error) {
 	var (
 		cteFragments   []string
@@ -357,6 +401,17 @@ func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
 		fieldNames = append(fieldNames, sqlbuilder.Escape(clickhousesql.Identifier(gb.Name)))
 	}
 
+	var bucketAliases []string
+	if heatmapField != nil {
+		valueFragment, aliases, warnings, err := querybuilder.HeatmapBucketColumns(ctx, info, b.storage, sb, heatmapField, query.BucketOptions, keys)
+		if err != nil {
+			return nil, err
+		}
+		cteFragments = append(cteFragments, valueFragment)
+		bucketAliases = aliases
+		columnWarnings = append(columnWarnings, warnings...)
+	}
+
 	allAggChArgs := make([]any, 0)
 	for i, agg := range query.Aggregations {
 		rewritten, chArgs, aggWarnings, err := b.aggExprRewriter.Rewrite(ctx, orgID, start, end, agg.Expression, uint64(query.StepInterval.Seconds()), keys)
@@ -393,6 +448,7 @@ func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
 
 		sb.GroupBy("ts")
 		sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+		sb.GroupBy(bucketAliases...)
 		if query.Having != nil && query.Having.Expression != "" {
 			rewriter := querybuilder.NewHavingExpressionRewriter()
 			rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)
@@ -420,6 +476,7 @@ func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
 	} else {
 		sb.GroupBy("ts")
 		sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+		sb.GroupBy(bucketAliases...)
 		if query.Having != nil && query.Having.Expression != "" {
 			rewriter := querybuilder.NewHavingExpressionRewriter()
 			rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)

pkg/statementbuilder/auditstatementbuilder/statement_builder_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
 	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -224,3 +225,59 @@ func TestStatementBuilder(t *testing.T) {
 		})
 	}
 }
+
+func TestStatementBuilderHeatmap(t *testing.T) {
+	statementBuilder := newTestAuditStatementBuilder(t)
+
+	testCases := []struct {
+		name        string
+		requestType qbtypes.RequestType
+		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
+		expected    qbtypes.Statement
+	}{
+		{
+			name:        "AttributeField_GroupByMaterialized_LinearAxis_RequiresField",
+			requestType: qbtypes.RequestTypeHeatmap,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:        telemetrytypes.SignalLogs,
+				Source:        telemetrytypes.SourceAudit,
+				StepInterval:  qbtypes.Step{Duration: 60 * time.Second},
+				Aggregations:  []qbtypes.LogAggregation{{Expression: "http.response.status_code"}},
+				Filter:        &qbtypes.Filter{Expression: "signoz.audit.resource.kind = 'dashboard'"},
+				GroupBy:       []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "signoz.audit.action"}}},
+				BucketOptions: &qbtypes.BucketOptions{Kind: qbtypes.BucketsKindLinear, Spec: qbtypes.LinearBucketsSpec{MaxValue: 500, NumBuckets: 5}},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE ((simpleJSONExtractString(labels, 'signoz.audit.resource.kind') = ? AND labels LIKE ? AND labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), ifNull(multiIf(mapContains(attributes_number, 'http.response.status_code'), toFloat64(attributes_number['http.response.status_code']), NULL), 0) AS __bucket_value SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 60 SECOND) AS ts, toString(multiIf(`attribute_string_signoz$$audit$$action_exists` = true, `attribute_string_signoz$$audit$$action`, NULL)) AS `signoz.audit.action`, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value > 500, toFloat64(500), (least(greatest(ceil(__bucket_value * 5 / 500), 1), 5) - 1) * 500 / 5) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value > 500, toFloat64('+Inf'), least(greatest(ceil(__bucket_value * 5 / 500), 1), 5) * 500 / 5) AS __bucket_max, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND mapContains(attributes_number, 'http.response.status_code') AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY ts, `signoz.audit.action`, __bucket_min, __bucket_max",
+				Args:  []any{"dashboard", "%signoz.audit.resource.kind%", "%signoz.audit.resource.kind\":\"dashboard%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
+			},
+		},
+		{
+			name:        "Distribution_AttributeField_SameStatementAsHeatmap",
+			requestType: qbtypes.RequestTypeDistribution,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:        telemetrytypes.SignalLogs,
+				Source:        telemetrytypes.SourceAudit,
+				StepInterval:  qbtypes.Step{Duration: 60 * time.Second},
+				Aggregations:  []qbtypes.LogAggregation{{Expression: "http.response.status_code"}},
+				Filter:        &qbtypes.Filter{Expression: "signoz.audit.resource.kind = 'dashboard'"},
+				GroupBy:       []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "signoz.audit.action"}}},
+				BucketOptions: &qbtypes.BucketOptions{Kind: qbtypes.BucketsKindLinear, Spec: qbtypes.LinearBucketsSpec{MaxValue: 500, NumBuckets: 5}},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE ((simpleJSONExtractString(labels, 'signoz.audit.resource.kind') = ? AND labels LIKE ? AND labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), ifNull(multiIf(mapContains(attributes_number, 'http.response.status_code'), toFloat64(attributes_number['http.response.status_code']), NULL), 0) AS __bucket_value SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 60 SECOND) AS ts, toString(multiIf(`attribute_string_signoz$$audit$$action_exists` = true, `attribute_string_signoz$$audit$$action`, NULL)) AS `signoz.audit.action`, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value > 500, toFloat64(500), (least(greatest(ceil(__bucket_value * 5 / 500), 1), 5) - 1) * 500 / 5) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value > 500, toFloat64('+Inf'), least(greatest(ceil(__bucket_value * 5 / 500), 1), 5) * 500 / 5) AS __bucket_max, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND mapContains(attributes_number, 'http.response.status_code') AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY ts, `signoz.audit.action`, __bucket_min, __bucket_max",
+				Args:  []any{"dashboard", "%signoz.audit.resource.kind%", "%signoz.audit.resource.kind\":\"dashboard%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			statement, err := statementBuilder.Build(context.Background(), valuer.UUID{}, 1747947419000, 1747983448000, testCase.requestType, testCase.query, nil)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expected.Query, statement.Query)
+			assert.Equal(t, testCase.expected.Args, statement.Args)
+			assert.Equal(t, testCase.expected.Warnings, statement.Warnings)
+		})
+	}
+}

pkg/statementbuilder/logsstatementbuilder/statement_builder.go

@@ -142,6 +142,8 @@ func (b *logQueryStatementBuilder) Build(
 		stmt, err = b.buildListQuery(ctx, orgID, q, query, start, end, keys, variables)
 	case qbtypes.RequestTypeTimeSeries:
 		stmt, err = b.buildTimeSeriesQuery(ctx, orgID, q, query, start, end, keys, variables)
+	case qbtypes.RequestTypeHeatmap, qbtypes.RequestTypeDistribution:
+		stmt, err = b.buildHeatmapTimeSeriesQuery(ctx, orgID, q, query, start, end, keys, variables)
 	case qbtypes.RequestTypeScalar:
 		stmt, err = b.buildScalarQuery(ctx, orgID, q, query, start, end, keys, false, variables)
 	default:
@@ -442,6 +444,48 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
 	keys map[string][]*telemetrytypes.TelemetryFieldKey,
 	variables map[string]qbtypes.VariableItem,
 ) (*qbtypes.Statement, error) {
+	return b.buildTimeSeriesOrHeatmapQuery(ctx, orgID, sb, query, start, end, keys, variables, nil)
+}
+
+// buildHeatmapTimeSeriesQuery counts rows per (time bucket, group, value
+// bucket) of the field its aggregation names.
+func (b *logQueryStatementBuilder) buildHeatmapTimeSeriesQuery(
+	ctx context.Context,
+	orgID valuer.UUID,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	heatmapField, countQuery, err := qbtypes.HeatmapCountQuery(query)
+	if err != nil {
+		return nil, err
+	}
+	info := querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalLogs, nil, start, end)
+	mayBeAbsent, err := querybuilder.HeatmapFieldMayBeAbsent(ctx, info, b.storage, columnKey(heatmapField), keys)
+	if err != nil {
+		return nil, err
+	}
+	if mayBeAbsent {
+		countQuery = qbtypes.RequireHeatmapField(countQuery, *heatmapField)
+	}
+	return b.buildTimeSeriesOrHeatmapQuery(ctx, orgID, sb, countQuery, start, end, keys, variables, heatmapField)
+}
+
+// buildTimeSeriesOrHeatmapQuery builds one row per (time bucket, group). A
+// non-nil heatmapField makes it a heatmap instead: the bucket a row's
+// heatmapField falls in joins the group keys, so a row is one cell.
+func (b *logQueryStatementBuilder) buildTimeSeriesOrHeatmapQuery(
+	ctx context.Context,
+	orgID valuer.UUID,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+	heatmapField *telemetrytypes.TelemetryFieldKey,
+) (*qbtypes.Statement, error) {
 
 	var (
 		cteFragments   []string
@@ -481,6 +525,19 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
 		fieldNames = append(fieldNames, sqlbuilder.Escape(fieldAlias))
 	}
 
+	var bucketAliases []string
+	if heatmapField != nil {
+		valueFragment, aliases, warnings, err := querybuilder.HeatmapBucketColumns(
+			ctx, info, b.storage, sb, columnKey(heatmapField), query.BucketOptions, keys,
+		)
+		if err != nil {
+			return nil, err
+		}
+		cteFragments = append(cteFragments, valueFragment)
+		bucketAliases = aliases
+		columnWarnings = append(columnWarnings, warnings...)
+	}
+
 	// Aggregations
 	allAggChArgs := make([]any, 0)
 	for i, agg := range query.Aggregations {
@@ -527,6 +584,7 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
 		// Group by all dimensions
 		sb.GroupBy("ts")
 		sb.GroupBy(fieldNames...)
+		sb.GroupBy(bucketAliases...)
 		if query.Having != nil && query.Having.Expression != "" {
 			// Rewrite having expression to use SQL column names
 			rewriter := querybuilder.NewHavingExpressionRewriter()
@@ -561,6 +619,7 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
 	} else {
 		sb.GroupBy("ts")
 		sb.GroupBy(fieldNames...)
+		sb.GroupBy(bucketAliases...)
 		if query.Having != nil && query.Having.Expression != "" {
 			rewriter := querybuilder.NewHavingExpressionRewriter()
 			rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)

pkg/statementbuilder/logsstatementbuilder/stmt_builder_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
 	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -1271,3 +1272,119 @@ func newSkipResourceFingerprintLogsBuilder(
 		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: skipEnable, Threshold: threshold}},
 	)
 }
+
+func TestStatementBuilderHeatmap(t *testing.T) {
+	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
+	startTs := uint64(releaseTime.UnixNano()) + uint64(24*time.Hour.Nanoseconds())
+	endTs := uint64(releaseTime.UnixNano()) + uint64(48*time.Hour.Nanoseconds())
+
+	testCases := []struct {
+		name        string
+		requestType qbtypes.RequestType
+		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
+		expected    qbtypes.Statement
+	}{
+		{
+			name:        "AttributeField_LogAxis_RequiresField",
+			requestType: qbtypes.RequestTypeHeatmap,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.LogAggregation{{Expression: "duration"}},
+				Filter:       &qbtypes.Filter{Expression: "service.name = 'cartservice'"},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE ((simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), ifNull(multiIf(mapContains(attributes_number, 'duration'), toFloat64(attributes_number['duration']), NULL), 0) AS __bucket_value SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value <= 2.3283064365386963e-10, toFloat64(0), __bucket_value > 1.8446744073709552e+19, toFloat64(1.8446744073709552e+19), pow(2, (ceil(log2(__bucket_value) * 16) - 1) / 16)) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value <= 2.3283064365386963e-10, 2.3283064365386963e-10, __bucket_value > 1.8446744073709552e+19, toFloat64('+Inf'), pow(2, ceil(log2(__bucket_value) * 16) / 16)) AS __bucket_max, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND mapContains(attributes_number, 'duration') AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY ts, __bucket_min, __bucket_max",
+				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1705397400), uint64(1705485600), "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
+			},
+		},
+		{
+			name:        "AttributeField_GroupByWithLimit_LinearAxis_RequiresFieldInLimit",
+			requestType: qbtypes.RequestTypeHeatmap,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:        telemetrytypes.SignalLogs,
+				StepInterval:  qbtypes.Step{Duration: 60 * time.Second},
+				Aggregations:  []qbtypes.LogAggregation{{Expression: "http.status_code"}},
+				GroupBy:       []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "service.name"}}},
+				Limit:         5,
+				BucketOptions: &qbtypes.BucketOptions{Kind: qbtypes.BucketsKindLinear, Spec: qbtypes.LinearBucketsSpec{MaxValue: 600, NumBuckets: 6}},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH ifNull(multiIf(mapContains(attributes_number, 'http.status_code'), toFloat64(attributes_number['http.status_code']), NULL), 0) AS __bucket_value, __limit_cte AS (SELECT toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE mapContains(attributes_number, 'http.status_code') AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 60 SECOND) AS ts, toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value > 600, toFloat64(600), (least(greatest(ceil(__bucket_value * 6 / 600), 1), 6) - 1) * 600 / 6) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value > 600, toFloat64('+Inf'), least(greatest(ceil(__bucket_value * 6 / 600), 1), 6) * 600 / 6) AS __bucket_max, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE mapContains(attributes_number, 'http.status_code') AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name`, __bucket_min, __bucket_max",
+				Args:  []any{"1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600), 5, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
+			},
+		},
+		{
+			name:        "QualifiedField_OrderedByAggregation_OrdersByCount",
+			requestType: qbtypes.RequestTypeHeatmap,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.LogAggregation{{Expression: "attribute.amount"}},
+				GroupBy:      []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "service.name"}}},
+				Limit:        3,
+				Order: []qbtypes.OrderBy{{
+					Key:       qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "amount", FieldContext: telemetrytypes.FieldContextAttribute}},
+					Direction: qbtypes.OrderDirectionAsc,
+				}},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH ifNull(multiIf(mapContains(attributes_number, 'amount'), toFloat64(attributes_number['amount']), NULL), 0) AS __bucket_value, __limit_cte AS (SELECT toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE mapContains(attributes_number, 'amount') AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY __result_0 asc LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value <= 2.3283064365386963e-10, toFloat64(0), __bucket_value > 1.8446744073709552e+19, toFloat64(1.8446744073709552e+19), pow(2, (ceil(log2(__bucket_value) * 16) - 1) / 16)) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value <= 2.3283064365386963e-10, 2.3283064365386963e-10, __bucket_value > 1.8446744073709552e+19, toFloat64('+Inf'), pow(2, ceil(log2(__bucket_value) * 16) / 16)) AS __bucket_max, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE mapContains(attributes_number, 'amount') AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name`, __bucket_min, __bucket_max ORDER BY ts desc",
+				Args:  []any{"1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600), 3, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
+			},
+		},
+		{
+			name:        "IntrinsicColumn_NoExists",
+			requestType: qbtypes.RequestTypeHeatmap,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.LogAggregation{{Expression: "log.severity_number"}},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH ifNull(toFloat64(severity_number), 0) AS __bucket_value SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value <= 2.3283064365386963e-10, toFloat64(0), __bucket_value > 1.8446744073709552e+19, toFloat64(1.8446744073709552e+19), pow(2, (ceil(log2(__bucket_value) * 16) - 1) / 16)) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value <= 2.3283064365386963e-10, 2.3283064365386963e-10, __bucket_value > 1.8446744073709552e+19, toFloat64('+Inf'), pow(2, ceil(log2(__bucket_value) * 16) / 16)) AS __bucket_max, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY ts, __bucket_min, __bucket_max",
+				Args:  []any{"1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
+			},
+		},
+		{
+			name:        "Distribution_AttributeField_SameStatementAsHeatmap",
+			requestType: qbtypes.RequestTypeDistribution,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.LogAggregation{{Expression: "duration"}},
+				Filter:       &qbtypes.Filter{Expression: "service.name = 'cartservice'"},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE ((simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), ifNull(multiIf(mapContains(attributes_number, 'duration'), toFloat64(attributes_number['duration']), NULL), 0) AS __bucket_value SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value <= 2.3283064365386963e-10, toFloat64(0), __bucket_value > 1.8446744073709552e+19, toFloat64(1.8446744073709552e+19), pow(2, (ceil(log2(__bucket_value) * 16) - 1) / 16)) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value <= 2.3283064365386963e-10, 2.3283064365386963e-10, __bucket_value > 1.8446744073709552e+19, toFloat64('+Inf'), pow(2, ceil(log2(__bucket_value) * 16) / 16)) AS __bucket_max, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND mapContains(attributes_number, 'duration') AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY ts, __bucket_min, __bucket_max",
+				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1705397400), uint64(1705485600), "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
+			},
+		},
+	}
+
+	fl := flaggertest.New(t)
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
+	storage := logstelemetryschema.NewStorage()
+	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
+	statementBuilder := NewLogQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		mockMetadataStore,
+		storage,
+		aggExprRewriter,
+		logstelemetryschema.DefaultFullTextColumn,
+		fl,
+		nil,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
+	)
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			statement, err := statementBuilder.Build(context.Background(), valuer.UUID{}, startTs, endTs, testCase.requestType, testCase.query, nil)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expected.Query, statement.Query)
+			assert.Equal(t, testCase.expected.Args, statement.Args)
+			assert.Equal(t, testCase.expected.Warnings, statement.Warnings)
+		})
+	}
+}

pkg/statementbuilder/metricsstatementbuilder/statement_builder.go

@@ -4,14 +4,11 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
-	"math"
 	"slices"
-	"strconv"
 	"strings"
 	"time"
 
 	"github.com/SigNoz/signoz/pkg/clickhousesql"
-	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/flagger"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
@@ -917,6 +914,10 @@ const (
 
 	heatmapValueAlias = "__result_0"
 	heatmapWindow     = "__heatmap_window"
+
+	// The column __spatial_aggregation_cte carries the sample in, which is what
+	// the bucket axis is cut from.
+	heatmapValueColumn = "value"
 )
 
 func isHistogramBucket(k qbtypes.GroupByKey) bool { return k.Name == histogramBucketKey }
@@ -982,10 +983,7 @@ func buildValueHeatmapFinalSelect(
 	args []any,
 	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
 ) (*qbtypes.Statement, error) {
-	bucketMin, bucketMax, err := renderHeatmapBucketExprs(*query.Aggregations[0].HeatmapBucketing)
-	if err != nil {
-		return nil, err
-	}
+	bucketMin, bucketMax := querybuilder.RenderHeatmapBucketExprs(*query.Aggregations[0].HeatmapBucketing, heatmapValueColumn)
 
 	groupAliases := GroupByAliases(query.GroupBy)
 
@@ -1003,61 +1001,6 @@ func buildValueHeatmapFinalSelect(
 	return &qbtypes.Statement{Query: combined + q, Args: append(args, a...)}, nil
 }
 
-// renderHeatmapBucketExprs renders the bucket (min, max] that `value` falls in.
-// Only the bucket under everything the axis covers is open below, and only the
-// one over it is open above.
-func renderHeatmapBucketExprs(bucketing qbtypes.HeatmapBucketing) (minExpr, maxExpr string, err error) {
-	switch bucketing.Kind {
-	case qbtypes.BucketsKindLinear:
-		return renderLinearBucketExprs(bucketing)
-	case qbtypes.BucketsKindLog:
-		return renderLogBucketExprs()
-	default:
-		return "", "", errors.NewInvalidInputf(errors.CodeInvalidInput,
-			"unsupported bucketsScaling %q for heatmap requests", bucketing.Kind.StringValue())
-	}
-}
-
-func renderLinearBucketExprs(bucketing qbtypes.HeatmapBucketing) (string, string, error) {
-	maxValue := formatFloat(bucketing.MaxValue)
-	numBuckets := strconv.Itoa(bucketing.NumBuckets)
-	index := fmt.Sprintf("least(greatest(ceil(value * %s / %s), 1), %s)", numBuckets, maxValue, numBuckets)
-
-	minExpr := fmt.Sprintf(
-		"multiIf(value <= 0, toFloat64('-Inf'), value > %s, toFloat64(%s), (%s - 1) * %s / %s)",
-		maxValue, maxValue, index, maxValue, numBuckets,
-	)
-	maxExpr := fmt.Sprintf(
-		"multiIf(value <= 0, toFloat64(0), value > %s, toFloat64('+Inf'), %s * %s / %s)",
-		maxValue, index, maxValue, numBuckets,
-	)
-	return minExpr, maxExpr, nil
-}
-
-// ClickHouse buckets at MaxLogScale whatever HeatmapBucketing.LogScale asks for;
-// postprocessing folds the axis down afterwards.
-func renderLogBucketExprs() (string, string, error) {
-	bucketsPerDoubling := formatFloat(math.Exp2(qbtypes.MaxLogScale))
-	lowest := formatFloat(qbtypes.MinLogUpperBound)
-	highest := formatFloat(qbtypes.MaxLogUpperBound)
-
-	minExpr := fmt.Sprintf(
-		"multiIf(value <= 0, toFloat64('-Inf'), value <= %s, toFloat64(0), value > %s, toFloat64(%s), pow(2, (ceil(log2(value) * %s) - 1) / %s))",
-		lowest, highest, highest, bucketsPerDoubling, bucketsPerDoubling,
-	)
-	maxExpr := fmt.Sprintf(
-		"multiIf(value <= 0, toFloat64(0), value <= %s, %s, value > %s, toFloat64('+Inf'), pow(2, ceil(log2(value) * %s) / %s))",
-		lowest, lowest, highest, bucketsPerDoubling, bucketsPerDoubling,
-	)
-	return minExpr, maxExpr, nil
-}
-
-// formatFloat renders a float64 as the shortest literal that reads back as the
-// same value, so an upper bound computed from it is identical on every row.
-func formatFloat(v float64) string {
-	return strconv.FormatFloat(v, 'g', -1, 64)
-}
-
 func GroupByColumnAlias(i int, name string) string {
 	if name == histogramBucketKey {
 		return clickhousesql.Identifier(histogramBucketKey)

pkg/statementbuilder/tracesstatementbuilder/statement_builder.go

@@ -178,6 +178,8 @@ func (b *traceQueryStatementBuilder) Build(
 		return b.buildListQuery(ctx, orgID, q, query, start, end, keys, variables, isSelectFieldsEmpty)
 	case qbtypes.RequestTypeTimeSeries:
 		return b.buildTimeSeriesQuery(ctx, orgID, q, query, start, end, keys, variables)
+	case qbtypes.RequestTypeHeatmap, qbtypes.RequestTypeDistribution:
+		return b.buildHeatmapTimeSeriesQuery(ctx, orgID, q, query, start, end, keys, variables)
 	case qbtypes.RequestTypeScalar:
 		return b.buildScalarQuery(ctx, orgID, q, query, start, end, keys, variables, false, false)
 	case qbtypes.RequestTypeTrace:
@@ -570,6 +572,48 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
 	keys map[string][]*telemetrytypes.TelemetryFieldKey,
 	variables map[string]qbtypes.VariableItem,
 ) (*qbtypes.Statement, error) {
+	return b.buildTimeSeriesOrHeatmapQuery(ctx, orgID, sb, query, start, end, keys, variables, nil)
+}
+
+// buildHeatmapTimeSeriesQuery counts rows per (time bucket, group, value
+// bucket) of the field its aggregation names.
+func (b *traceQueryStatementBuilder) buildHeatmapTimeSeriesQuery(
+	ctx context.Context,
+	orgID valuer.UUID,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	heatmapField, countQuery, err := qbtypes.HeatmapCountQuery(query)
+	if err != nil {
+		return nil, err
+	}
+	info := querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end)
+	mayBeAbsent, err := querybuilder.HeatmapFieldMayBeAbsent(ctx, info, b.storage, heatmapField, keys)
+	if err != nil {
+		return nil, err
+	}
+	if mayBeAbsent {
+		countQuery = qbtypes.RequireHeatmapField(countQuery, *heatmapField)
+	}
+	return b.buildTimeSeriesOrHeatmapQuery(ctx, orgID, sb, countQuery, start, end, keys, variables, heatmapField)
+}
+
+// buildTimeSeriesOrHeatmapQuery builds one row per (time bucket, group). A
+// non-nil heatmapField makes it a heatmap instead: the bucket a row's
+// heatmapField falls in joins the group keys, so a row is one cell.
+func (b *traceQueryStatementBuilder) buildTimeSeriesOrHeatmapQuery(
+	ctx context.Context,
+	orgID valuer.UUID,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+	heatmapField *telemetrytypes.TelemetryFieldKey,
+) (*qbtypes.Statement, error) {
 
 	var (
 		cteFragments   []string
@@ -610,6 +654,19 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
 		fieldNames = append(fieldNames, sqlbuilder.Escape(fieldAlias))
 	}
 
+	var bucketAliases []string
+	if heatmapField != nil {
+		valueFragment, aliases, warnings, err := querybuilder.HeatmapBucketColumns(
+			ctx, info, b.storage, sb, heatmapField, query.BucketOptions, keys,
+		)
+		if err != nil {
+			return nil, err
+		}
+		cteFragments = append(cteFragments, valueFragment)
+		bucketAliases = aliases
+		columnWarnings = append(columnWarnings, warnings...)
+	}
+
 	// Aggregations
 	allAggChArgs := make([]any, 0)
 	for i, agg := range query.Aggregations {
@@ -653,6 +710,7 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
 		// Group by all dimensions
 		sb.GroupBy("ts")
 		sb.GroupBy(fieldNames...)
+		sb.GroupBy(bucketAliases...)
 		if query.Having != nil && query.Having.Expression != "" {
 			rewriter := querybuilder.NewHavingExpressionRewriter()
 			rewrittenExpr, err := rewriter.RewriteForTraces(query.Having.Expression, query.Aggregations)
@@ -686,6 +744,7 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
 	} else {
 		sb.GroupBy("ts")
 		sb.GroupBy(fieldNames...)
+		sb.GroupBy(bucketAliases...)
 		if query.Having != nil && query.Having.Expression != "" {
 			rewriter := querybuilder.NewHavingExpressionRewriter()
 			rewrittenExpr, err := rewriter.RewriteForTraces(query.Having.Expression, query.Aggregations)

pkg/statementbuilder/tracesstatementbuilder/stmt_builder_test.go

@@ -2113,3 +2113,103 @@ func TestStatementBuilderSemconvSingleSpelling(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "WITH __resource_filter AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'deployment.environment') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT toString(multiIf(resource.`deployment.environment` IS NOT NULL, resource.`deployment.environment`::String, mapContains(resources_string, 'deployment.environment'), resources_string['deployment.environment'], NULL)) AS `__GROUP_BY_KEY_0_deployment.environment.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_deployment.environment.name` ORDER BY __result_0 DESC", q.Query)
 }
+
+func TestStatementBuilderHeatmap(t *testing.T) {
+	testCases := []struct {
+		name        string
+		requestType qbtypes.RequestType
+		query       qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]
+		expected    qbtypes.Statement
+	}{
+		{
+			name:        "DurationNano_CollidingAttribute_NoExists",
+			requestType: qbtypes.RequestTypeHeatmap,
+			query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal:       telemetrytypes.SignalTraces,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.TraceAggregation{{Expression: "duration_nano"}},
+				Filter:       &qbtypes.Filter{Expression: "service.name = 'redis-manual'"},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), ifNull(toFloat64(duration_nano), 0) AS __bucket_value SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value <= 2.3283064365386963e-10, toFloat64(0), __bucket_value > 1.8446744073709552e+19, toFloat64(1.8446744073709552e+19), pow(2, (ceil(log2(__bucket_value) * 16) - 1) / 16)) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value <= 2.3283064365386963e-10, 2.3283064365386963e-10, __bucket_value > 1.8446744073709552e+19, toFloat64('+Inf'), pow(2, ceil(log2(__bucket_value) * 16) / 16)) AS __bucket_max, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY ts, __bucket_min, __bucket_max",
+				Args:  []any{"redis-manual", "%service.name%", "%service.name\":\"redis-manual%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
+			},
+		},
+		{
+			name:        "AttributeField_GroupByWithLimit_LinearAxis_RequiresFieldInLimit",
+			requestType: qbtypes.RequestTypeHeatmap,
+			query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal:        telemetrytypes.SignalTraces,
+				StepInterval:  qbtypes.Step{Duration: 60 * time.Second},
+				Aggregations:  []qbtypes.TraceAggregation{{Expression: "cart.items_count"}},
+				GroupBy:       []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "service.name"}}},
+				Limit:         5,
+				BucketOptions: &qbtypes.BucketOptions{Kind: qbtypes.BucketsKindLinear, Spec: qbtypes.LinearBucketsSpec{MaxValue: 50, NumBuckets: 5}},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH ifNull(multiIf(`attribute_number_cart$$items_count_exists` = true, toFloat64(`attribute_number_cart$$items_count`), NULL), 0) AS __bucket_value, __limit_cte AS (SELECT toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE `attribute_number_cart$$items_count_exists` = true AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts, toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value > 50, toFloat64(50), (least(greatest(ceil(__bucket_value * 5 / 50), 1), 5) - 1) * 50 / 5) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value > 50, toFloat64('+Inf'), least(greatest(ceil(__bucket_value * 5 / 50), 1), 5) * 50 / 5) AS __bucket_max, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE `attribute_number_cart$$items_count_exists` = true AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name`, __bucket_min, __bucket_max",
+				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 5, "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
+			},
+		},
+		{
+			name:        "QualifiedField_OrderedByAggregation_OrdersByCount",
+			requestType: qbtypes.RequestTypeHeatmap,
+			query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal:       telemetrytypes.SignalTraces,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.TraceAggregation{{Expression: "attribute.metric.max_count"}},
+				GroupBy:      []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "service.name"}}},
+				Limit:        3,
+				Order: []qbtypes.OrderBy{{
+					Key:       qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "metric.max_count", FieldContext: telemetrytypes.FieldContextAttribute}},
+					Direction: qbtypes.OrderDirectionDesc,
+				}},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH ifNull(multiIf(mapContains(attributes_number, 'metric.max_count'), toFloat64(attributes_number['metric.max_count']), NULL), 0) AS __bucket_value, __limit_cte AS (SELECT toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE mapContains(attributes_number, 'metric.max_count') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY __result_0 desc LIMIT ?) SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value <= 2.3283064365386963e-10, toFloat64(0), __bucket_value > 1.8446744073709552e+19, toFloat64(1.8446744073709552e+19), pow(2, (ceil(log2(__bucket_value) * 16) - 1) / 16)) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value <= 2.3283064365386963e-10, 2.3283064365386963e-10, __bucket_value > 1.8446744073709552e+19, toFloat64('+Inf'), pow(2, ceil(log2(__bucket_value) * 16) / 16)) AS __bucket_max, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE mapContains(attributes_number, 'metric.max_count') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name`, __bucket_min, __bucket_max ORDER BY ts desc",
+				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 3, "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
+			},
+		},
+		{
+			name:        "Distribution_DurationNano_SameStatementAsHeatmap",
+			requestType: qbtypes.RequestTypeDistribution,
+			query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal:       telemetrytypes.SignalTraces,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.TraceAggregation{{Expression: "duration_nano"}},
+				Filter:       &qbtypes.Filter{Expression: "service.name = 'redis-manual'"},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), ifNull(toFloat64(duration_nano), 0) AS __bucket_value SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, multiIf(__bucket_value <= 0, toFloat64('-Inf'), __bucket_value <= 2.3283064365386963e-10, toFloat64(0), __bucket_value > 1.8446744073709552e+19, toFloat64(1.8446744073709552e+19), pow(2, (ceil(log2(__bucket_value) * 16) - 1) / 16)) AS __bucket_min, multiIf(__bucket_value <= 0, toFloat64(0), __bucket_value <= 2.3283064365386963e-10, 2.3283064365386963e-10, __bucket_value > 1.8446744073709552e+19, toFloat64('+Inf'), pow(2, ceil(log2(__bucket_value) * 16) / 16)) AS __bucket_max, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY ts, __bucket_min, __bucket_max",
+				Args:  []any{"redis-manual", "%service.name%", "%service.name\":\"redis-manual%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
+			},
+		},
+	}
+
+	releaseTime := time.Date(2025, 5, 22, 0, 0, 0, 0, time.UTC)
+	fl := flaggertest.New(t)
+	storage := tracestelemetryschema.NewStorage(nil)
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
+	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
+	statementBuilder := NewTraceQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		mockMetadataStore,
+		storage,
+		aggExprRewriter,
+		nil,
+		fl,
+		false,
+		100000,
+	)
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			statement, err := statementBuilder.Build(context.Background(), valuer.UUID{}, 1747947419000, 1747983448000, testCase.requestType, testCase.query, nil)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expected.Query, statement.Query)
+			assert.Equal(t, testCase.expected.Args, statement.Args)
+			assert.Equal(t, testCase.expected.Warnings, statement.Warnings)
+		})
+	}
+}

pkg/types/querybuildertypes/querybuildertypesv5/heatmap.go

@@ -3,9 +3,11 @@ package querybuildertypesv5
 import (
 	"math"
 	"slices"
+	"strings"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
 
 const (
@@ -16,6 +18,9 @@ const (
 	HeatmapBucketMinColumn = "__bucket_min"
 	HeatmapBucketMaxColumn = "__bucket_max"
 
+	// A logs or traces heatmap counts events, so its field renders as count().
+	heatmapCountExpression = "count()"
+
 	DefaultNumBuckets = 60
 
 	// MaxLogScale is the resolution ClickHouse buckets every log heatmap at:
@@ -81,6 +86,68 @@ func (b *BucketOptions) ToHeatmapBucketing() HeatmapBucketing {
 	return resolved
 }
 
+// HeatmapAggregationField returns the field a logs or traces heatmap buckets,
+// written as the bare field. ok is false for any other aggregation.
+func HeatmapAggregationField(expression string) (telemetrytypes.TelemetryFieldKey, bool) {
+	keyText := strings.TrimSpace(expression)
+	if keyText == "" || strings.ContainsAny(keyText, "() \t\n") {
+		return telemetrytypes.TelemetryFieldKey{}, false
+	}
+	return telemetrytypes.GetFieldKeyFromKeyText(keyText), true
+}
+
+// HeatmapCountQuery splits a logs or traces heatmap query into the field its
+// aggregation buckets and a copy of query that counts events in its place,
+// which is what every cell holds.
+func HeatmapCountQuery[T LogAggregation | TraceAggregation](query QueryBuilderQuery[T]) (*telemetrytypes.TelemetryFieldKey, QueryBuilderQuery[T], error) {
+	if len(query.Aggregations) != 1 {
+		return nil, query, errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"a heatmap renders one bucket axis, but the query has %d aggregations", len(query.Aggregations))
+	}
+
+	query.Aggregations = slices.Clone(query.Aggregations)
+	var expression *string
+	switch aggregation := any(&query.Aggregations[0]).(type) {
+	case *LogAggregation:
+		expression = &aggregation.Expression
+	case *TraceAggregation:
+		expression = &aggregation.Expression
+	}
+
+	field, ok := HeatmapAggregationField(*expression)
+	if !ok {
+		return nil, query, errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"a logs or traces heatmap counts events by the field named as its aggregation, got %q", *expression)
+	}
+
+	// An order on the aggregation names it by its expression, which decoding
+	// splits into a context, a name and a data type, and which metadata may
+	// complete, as it does any field.
+	query.Order = slices.Clone(query.Order)
+	for i := range query.Order {
+		key := query.Order[i].Key.TelemetryFieldKey
+		sameField := key.Name == field.Name &&
+			(field.FieldContext == telemetrytypes.FieldContextUnspecified || key.FieldContext == field.FieldContext) &&
+			(field.FieldDataType == telemetrytypes.FieldDataTypeUnspecified || key.FieldDataType == field.FieldDataType)
+		if key.Name == *expression || sameField {
+			query.Order[i].Key = OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: heatmapCountExpression}}
+		}
+	}
+	*expression = heatmapCountExpression
+
+	return &field, query, nil
+}
+
+// RequireHeatmapField keeps only the rows of query that carry field.
+func RequireHeatmapField[T LogAggregation | TraceAggregation](query QueryBuilderQuery[T], field telemetrytypes.TelemetryFieldKey) QueryBuilderQuery[T] {
+	exists := telemetrytypes.TelemetryFieldKeyToText(&field) + " EXISTS"
+	if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
+		exists = "(" + query.Filter.Expression + ") AND " + exists
+	}
+	query.Filter = &Filter{Expression: exists}
+	return query
+}
+
 // This cannot be called in validateBucketCounting cuz metric type is resolved in querier.go.
 func (a *MetricAggregation) VerifyAndApplyBucketOptions(bucketOptions *BucketOptions) error {
 	switch a.Type {

pkg/types/querybuildertypes/querybuildertypesv5/validation.go

@@ -934,15 +934,26 @@ func (r *QueryRangeRequest) validateBucketCounting() error {
 				continue
 			}
 			enabled++
-		// An AI query decodes to the traces spec, so it lands here too. Admitting
-		// either signal means capping Aggregations at one: each carries its own
-		// Meta.Buckets, and a heatmap renders against a single bucket axis.
-		case QueryBuilderQuery[LogAggregation], QueryBuilderQuery[TraceAggregation]:
-			return errors.New(errors.TypeUnsupported, errors.CodeUnsupported,
-				"heatmaps are not supported for the logs and traces signals yet")
+		// An AI query decodes to the traces spec, so it lands here too.
+		case QueryBuilderQuery[LogAggregation]:
+			if err := validateEventHeatmapQuery(r.RequestType, spec.Functions, spec.Having, aggregationExpressions(spec.Aggregations), spec.Disabled); err != nil {
+				return err
+			}
+			if spec.Disabled {
+				continue
+			}
+			enabled++
+		case QueryBuilderQuery[TraceAggregation]:
+			if err := validateEventHeatmapQuery(r.RequestType, spec.Functions, spec.Having, aggregationExpressions(spec.Aggregations), spec.Disabled); err != nil {
+				return err
+			}
+			if spec.Disabled {
+				continue
+			}
+			enabled++
 		default:
 			return errors.NewInvalidInputf(errors.CodeInvalidInput,
-				"%s requests support one metrics builder query, one formula over them, one clickhouse query, or one promql query, got %q", r.RequestType.StringValue(), envelope.Type.StringValue())
+				"%s requests support one builder query, one formula over them, one clickhouse query, or one promql query, got %q", r.RequestType.StringValue(), envelope.Type.StringValue())
 		}
 	}
 
@@ -1046,6 +1057,48 @@ func validateBucketCountingQuery(requestType RequestType, functions []Function,
 	return nil
 }
 
+// validateEventHeatmapQuery holds a logs or traces query to what a per-event
+// heatmap can draw. A disabled query feeds a formula rather than the chart, so
+// it aggregates as a time series.
+func validateEventHeatmapQuery(requestType RequestType, functions []Function, having *Having, expressions []string, disabled bool) error {
+	if err := validateBucketCountingQuery(requestType, functions, having); err != nil {
+		return err
+	}
+
+	if disabled {
+		return nil
+	}
+
+	// Every aggregation carries its own Meta.Buckets, and a heatmap renders
+	// against a single bucket axis.
+	if len(expressions) != 1 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"a heatmap renders one bucket axis, but the query has %d aggregations", len(expressions)).
+			WithAdditional("Keep the one aggregation whose distribution you want to plot")
+	}
+
+	if _, ok := HeatmapAggregationField(expressions[0]); !ok {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"a logs or traces heatmap counts events by the field named as its aggregation, got %q", expressions[0]).
+			WithAdditional("Use an aggregation such as duration_nano")
+	}
+
+	return nil
+}
+
+func aggregationExpressions[T LogAggregation | TraceAggregation](aggregations []T) []string {
+	expressions := make([]string, len(aggregations))
+	for i, aggregation := range aggregations {
+		switch a := any(aggregation).(type) {
+		case LogAggregation:
+			expressions[i] = a.Expression
+		case TraceAggregation:
+			expressions[i] = a.Expression
+		}
+	}
+	return expressions
+}
+
 func GetValidationOptions(requestType RequestType) []ValidationOption {
 	switch requestType {
 	case RequestTypeTimeSeries, RequestTypeHeatmap, RequestTypeDistribution:

tests/fixtures/querier.py

@@ -173,7 +173,7 @@ def make_query_request(
     end_ms: int,
     queries: list[dict],
     *,
-    request_type: str = RequestType.TIME_SERIES,
+    request_type: str,
     format_options: dict | None = None,
     variables: dict | None = None,
     no_cache: bool = True,
@@ -209,7 +209,7 @@ def make_preview_query_request(
     end_ms: int,
     queries: list[dict],
     *,
-    request_type: str = RequestType.TIME_SERIES,
+    request_type: str,
     format_options: dict | None = None,
     variables: dict | None = None,
     verbose: bool = True,
@@ -282,6 +282,7 @@ def query_metric_values(  # pylint: disable=too-many-arguments,too-many-position
         start_ms=start_epoch * 1000,
         end_ms=end_epoch * 1000,
         queries=[build_builder_query("A", metric_name, time_agg, space_agg, step_interval=step_interval)],
+        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
     return sorted(get_series_values(response.json(), "A"), key=lambda v: v["timestamp"])
@@ -679,7 +680,7 @@ def assert_raw_row_subset(
     assert expected.items() <= data.items(), f"expected subset {expected}, got data {data}"
 
 
-def build_scalar_query(
+def build_aggregation_query(
     name: str,
     signal: str,
     aggregations: list[dict],
@@ -730,7 +731,45 @@ def build_scalar_query(
     return {"type": "builder_query", "spec": spec}
 
 
-def build_traces_scalar_query(
+def build_heatmap_query(
+    name: str,
+    signal: str,
+    expression: str,
+    *,
+    source: str | None = None,
+    group_by: list[dict] | None = None,
+    order: list[dict] | None = None,
+    filter_expression: str | None = None,
+    bucket_options: dict | None = None,
+    step_interval: int = DEFAULT_STEP_INTERVAL,
+) -> dict:
+    """Omitting bucket_options leaves the server's default log axis."""
+    spec: dict[str, Any] = {
+        "name": name,
+        "signal": signal,
+        "stepInterval": step_interval,
+        "aggregations": [{"expression": expression}],
+    }
+
+    if source:
+        spec["source"] = source
+
+    if group_by:
+        spec["groupBy"] = group_by
+
+    if order:
+        spec["order"] = order
+
+    if filter_expression:
+        spec["filter"] = {"expression": filter_expression}
+
+    if bucket_options is not None:
+        spec["bucketOptions"] = bucket_options
+
+    return {"type": "builder_query", "spec": spec}
+
+
+def build_traces_aggregation_query(
     aggregations: list[dict],
     *,
     name: str = "A",
@@ -743,8 +782,7 @@ def build_traces_scalar_query(
     disabled: bool = False,
     functions: list[dict] | None = None,
 ) -> dict:
-    """build_scalar_query pinned to the traces signal, with name defaulting to 'A'."""
-    return build_scalar_query(
+    return build_aggregation_query(
         name=name,
         signal="traces",
         aggregations=aggregations,
@@ -1283,7 +1321,7 @@ def run_query_case(signoz: types.SigNoz, token: str, now: datetime, case: dict[s
             aggregations = aggregation
         else:
             aggregations = []
-        query = build_scalar_query(
+        query = build_aggregation_query(
             name=case["name"],
             signal="logs",
             aggregations=aggregations,

tests/integration/tests/auditquerier/01_audit_logs.py

@@ -10,8 +10,8 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.querier import (
     BuilderQuery,
     build_aggregation,
+    build_aggregation_query,
     build_order_by,
-    build_scalar_query,
     make_query_request,
 )
 
@@ -361,7 +361,7 @@ def test_audit_scalar_count_failures(
         start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
         end_ms=int(now.timestamp() * 1000),
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="logs",
                 source="audit",

tests/integration/tests/auditquerier/02_heatmap.py

@@ -0,0 +1,83 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.audit import AuditLog
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import (
+    RequestType,
+    assert_distribution_sums_heatmap,
+    build_group_by_field,
+    build_heatmap_query,
+    build_linear_bucket_options,
+    get_all_series,
+    get_buckets,
+    index_series_by_label,
+    make_query_request,
+)
+
+
+def test_audit_heatmap(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_audit_logs: Callable[[list[AuditLog]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    changed_by_action = {"update": [3, 8], "delete": [25]}
+    insert_audit_logs(
+        [
+            AuditLog(
+                timestamp=now - timedelta(seconds=index + 1),
+                resources={"service.name": "signoz", "signoz.audit.resource.kind": "dashboard"},
+                attributes={"signoz.audit.action": action, "signoz.audit.fields_changed": fields_changed},
+                event_name=f"dashboard.{action}",
+            )
+            for action, values in changed_by_action.items()
+            for index, fields_changed in enumerate(values)
+        ]
+        # an event without the field is not counted
+        + [
+            AuditLog(
+                timestamp=now - timedelta(seconds=5),
+                resources={"service.name": "signoz", "signoz.audit.resource.kind": "dashboard"},
+                attributes={"signoz.audit.action": "update"},
+                event_name="dashboard.update",
+            )
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+    end_ms = int((now + timedelta(seconds=5)).timestamp() * 1000)
+    queries = [
+        build_heatmap_query(
+            "A",
+            "logs",
+            "signoz.audit.fields_changed",
+            source="audit",
+            group_by=[build_group_by_field("signoz.audit.action", "string", "attribute")],
+            filter_expression="signoz.audit.resource.kind = 'dashboard'",
+            bucket_options=build_linear_bucket_options(20, 4),
+        )
+    ]
+    response = make_query_request(signoz, token, start_ms, end_ms, queries, request_type=RequestType.HEATMAP)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_distribution_sums_heatmap(signoz, token, start_ms, end_ms, queries, response)
+
+    data = response.json()
+    # 20 split into 4 gives (0, 5], (5, 10], (10, 15], (15, 20]; the axis stops
+    # at the highest bucket reached, and 25 is past 20 so it counts in the
+    # trailing overflow
+    buckets = get_buckets(data, "A")
+    assert buckets == pytest.approx([0.0, 5.0, 10.0])
+
+    series_by_action = index_series_by_label(get_all_series(data, "A"), "signoz.audit.action")
+    totals_by_action = {action: [sum(column["values"][slot] for column in series["values"]) for slot in range(len(buckets) + 1)] for action, series in series_by_action.items()}
+    assert totals_by_action == {
+        "update": [0, 1, 1, 0],
+        "delete": [0, 0, 0, 1],
+    }

tests/integration/tests/metricreduction/03_reduced_recent_window.py

@@ -6,6 +6,7 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metricreduction import build_recent_gauge_data
 from fixtures.querier import (
+    RequestType,
     aligned_epoch,
     build_builder_query,
     get_all_series,
@@ -59,6 +60,7 @@ def test_recent_queries_group_by_full_labels(
         start_ms=base_epoch * 1000,
         end_ms=(base_epoch + MINUTES * 60) * 1000,
         queries=[build_builder_query("A", metric_name, "sum", "sum", step_interval=300, group_by=["service"])],
+        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
 

tests/integration/tests/promqlconformance/01_upstream_corpus.py

@@ -8,7 +8,7 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.promqltestcorpus import ingest_promqltest_corpus
-from fixtures.querier import get_all_series, make_query_request
+from fixtures.querier import RequestType, get_all_series, make_query_request
 
 TESTDATA_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "testdata")
 # The corpus (see fixtures/promqltestcorpus.py) is frozen from Prometheus' own
@@ -57,7 +57,7 @@ def test_upstream_promqltest_corpus(
         }
         case_id = f"{case['source']}[{case['variant']}]"
 
-        response = make_query_request(signoz, token, req_start_ms, end_ms, [query])
+        response = make_query_request(signoz, token, req_start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
         if response.status_code != HTTPStatus.OK:
             failures.append(f"{case_id}: HTTP {response.status_code} for {case['expr']!r}: {response.text[:200]}")
             continue

tests/integration/tests/promqlconformance/02_fingerprint_probe.py

@@ -5,7 +5,7 @@ from http import HTTPStatus
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import get_all_series, make_query_request
+from fixtures.querier import RequestType, get_all_series, make_query_request
 
 METRIC = "probe_requests"
 EVOLVED_METRIC = "probe_schema_evolution"
@@ -74,7 +74,7 @@ def test_identical_labelsets_merge_and_grouping(
 
     def run(promql: str, start_ms: int, end_ms: int) -> list[tuple[dict, list]]:
         q = {"type": "promql", "spec": {"name": "A", "query": promql, "step": 60, "skipStepAlignment": True}}
-        resp = make_query_request(signoz, token, start_ms, end_ms, [q])
+        resp = make_query_request(signoz, token, start_ms, end_ms, [q], request_type=RequestType.TIME_SERIES)
         assert resp.status_code == HTTPStatus.OK, f"{promql!r}: {resp.text[:300]}"
         out = []
         for series in get_all_series(resp.json(), "A") or []:

tests/integration/tests/promqlconformance/03_nonfinite_cache.py

@@ -6,7 +6,7 @@ from uuid import uuid4
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import get_all_series, make_query_request
+from fixtures.querier import RequestType, get_all_series, make_query_request
 
 HOUR_MS = 3_600_000
 SAMPLE_INTERVAL_MS = 60_000
@@ -40,7 +40,7 @@ def test_promql_ratio_with_zero_denominator_is_dropped_and_cached(
 
     def run() -> tuple[dict[str, dict[int, object]], int]:
         query = {"type": "promql", "spec": {"name": "A", "query": promql}}
-        response = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
+        response = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
         assert response.status_code == HTTPStatus.OK, response.text[:300]
         body = response.json()
         out: dict[str, dict[int, object]] = {}

tests/integration/tests/promqlconformance/04_no_step_subquery.py

@@ -6,7 +6,7 @@ from uuid import uuid4
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import get_all_series, make_query_request
+from fixtures.querier import RequestType, get_all_series, make_query_request
 
 MINUTE_MS = 60_000
 
@@ -41,7 +41,7 @@ def test_promql_subquery_without_step_evaluates(
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     query = {"type": "promql", "spec": {"name": "A", "query": f"max_over_time({metric}[5m:])", "skipStepAlignment": True}}
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK, response.text[:300]
     series = get_all_series(response.json(), "A")
     assert series, "the subquery must return the inserted series"
@@ -55,5 +55,6 @@ def test_promql_subquery_without_step_evaluates(
         start_ms,
         end_ms,
         [{"type": "promql", "spec": {"name": "A", "query": metric, "skipStepAlignment": True}}],
+        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text[:300]

tests/integration/tests/promqlconformance/05_or_subquery_duplicate_labelset.py

@@ -8,7 +8,7 @@ import requests
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import get_all_series, make_query_request
+from fixtures.querier import RequestType, get_all_series, make_query_request
 
 MINUTE_MS = 60_000
 QUERY_TIMEOUT = 30
@@ -73,7 +73,7 @@ def test_or_arms_merge_under_subquery_name_drop(
 
     # query_range over the same grid must agree point for point.
     spec = {"name": "A", "query": query, "step": 300, "skipStepAlignment": True}
-    response = make_query_request(signoz, token, start_ms, end_ms, [{"type": "promql", "spec": spec}])
+    response = make_query_request(signoz, token, start_ms, end_ms, [{"type": "promql", "spec": spec}], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK, response.text[:300]
     series = get_all_series(response.json(), "A")
     assert len(series) == 1, f"both or arms must merge into one series: {series}"

tests/integration/tests/querier_json_body/02_non_body_fields.py

@@ -7,10 +7,10 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
     build_aggregation,
+    build_aggregation_query,
     build_group_by_field,
     build_order_by,
     build_raw_query,
-    build_scalar_query,
     get_rows,
     get_scalar_table_data,
     make_query_request,
@@ -136,7 +136,7 @@ def test_non_body_filter_groupby_aggregation(
                 step_interval=60,
             )
         else:
-            query = build_scalar_query(
+            query = build_aggregation_query(
                 name=case["name"],
                 signal="logs",
                 aggregations=[build_aggregation(case["aggregation"])],
@@ -233,7 +233,7 @@ def test_non_body_orderby(
                 step_interval=60,
             )
         else:
-            query = build_scalar_query(
+            query = build_aggregation_query(
                 name=case["name"],
                 signal="logs",
                 aggregations=[build_aggregation(case["aggregation"])],

tests/integration/tests/querier_json_body/04_unknown_key_collisions.py

@@ -12,9 +12,9 @@ from fixtures.logs import Logs
 from fixtures.querier import (
     RequestType,
     build_aggregation,
+    build_aggregation_query,
     build_group_by_field,
     build_raw_query,
-    build_scalar_query,
     get_rows,
     get_scalar_table_data,
     make_query_request,
@@ -68,7 +68,7 @@ def test_json_body_unknown_key_collision_groupby(
         token,
         now,
         [
-            build_scalar_query(
+            build_aggregation_query(
                 "A",
                 "logs",
                 [build_aggregation("count()")],
@@ -175,7 +175,7 @@ def test_json_body_unknown_key_attribute_parity(
             token,
             now,
             [
-                build_scalar_query(
+                build_aggregation_query(
                     "A",
                     "logs",
                     [build_aggregation("count()")],
@@ -225,7 +225,7 @@ def test_json_body_unknown_key_aggregation_over_body(
         token,
         now,
         [
-            build_scalar_query(
+            build_aggregation_query(
                 "A",
                 "logs",
                 [build_aggregation("count(field)", "cnt"), build_aggregation("count_distinct(field)", "distinct")],
@@ -291,7 +291,7 @@ def test_json_body_unknown_key_body_path_parity(
             token,
             now,
             [
-                build_scalar_query(
+                build_aggregation_query(
                     "A",
                     "logs",
                     [build_aggregation("count()")],
@@ -308,7 +308,7 @@ def test_json_body_unknown_key_body_path_parity(
             token,
             now,
             [
-                build_scalar_query(
+                build_aggregation_query(
                     "A",
                     "logs",
                     [build_aggregation(f"count_distinct({field})")],

tests/integration/tests/querier_json_body/07_field_name_quoting.py

@@ -8,7 +8,7 @@ import pytest
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
-from fixtures.querier import build_aggregation, build_raw_query, build_scalar_query, get_rows, get_scalar_table_data, make_query_request
+from fixtures.querier import build_aggregation, build_aggregation_query, build_raw_query, get_rows, get_scalar_table_data, make_query_request
 
 SERVICE = "quoting-json-svc"
 
@@ -55,7 +55,7 @@ def test_body_json_group_by_and_select_hostile_key(
         token,
         start_ms,
         end_ms,
-        [build_scalar_query("A", "logs", [build_aggregation("count()")], group_by=[field], order=[{"key": {"name": key}, "direction": "asc"}], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_aggregation_query("A", "logs", [build_aggregation("count()")], group_by=[field], order=[{"key": {"name": key}, "direction": "asc"}], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text

tests/integration/tests/queriercommon/03_clickhouse_queries.py

@@ -118,7 +118,7 @@ INVALID_REGEXP = "write the pattern in RE2 syntax, which has no lookaround or ba
             id="regexp_in_sql",
         ),
         pytest.param(
-            querier.build_scalar_query("A", "logs", [querier.build_aggregation("count()")], filter_expression="body REGEXP '(?=a)'"),
+            querier.build_aggregation_query("A", "logs", [querier.build_aggregation("count()")], filter_expression="body REGEXP '(?=a)'"),
             "invalid_regexp",
             INVALID_REGEXP,
             id="regexp_in_filter",

tests/integration/tests/queriercommon/06_keyless_semantics.py

@@ -195,6 +195,7 @@ def test_metrics_negative_operators_include_keyless_series(
                 filter_expression=expression,
             )
         ],
+        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
 

tests/integration/tests/queriercommon/08_field_name_quoting.py

@@ -9,10 +9,11 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    RequestType,
     build_aggregation,
+    build_aggregation_query,
     build_builder_query,
     build_raw_query,
-    build_scalar_query,
     get_all_series,
     get_rows,
     get_scalar_table_data,
@@ -66,7 +67,7 @@ def test_logs_group_by_order_by_and_select_hostile_attribute(
         token,
         start_ms,
         end_ms,
-        [build_scalar_query("A", "logs", [build_aggregation("count()")], group_by=[field], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_aggregation_query("A", "logs", [build_aggregation("count()")], group_by=[field], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text
@@ -78,7 +79,7 @@ def test_logs_group_by_order_by_and_select_hostile_attribute(
         token,
         start_ms,
         end_ms,
-        [build_scalar_query("A", "logs", [build_aggregation("count()")], group_by=[field], order=[{"key": {"name": key}, "direction": "asc"}], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_aggregation_query("A", "logs", [build_aggregation("count()")], group_by=[field], order=[{"key": {"name": key}, "direction": "asc"}], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text
@@ -170,7 +171,7 @@ def test_logs_aggregation_with_dollar_in_key(
             token,
             start_ms,
             end_ms,
-            [build_scalar_query("A", "logs", [build_aggregation(expression)], filter_expression=f"service.name = '{SERVICE}'")],
+            [build_aggregation_query("A", "logs", [build_aggregation(expression)], filter_expression=f"service.name = '{SERVICE}'")],
             request_type="scalar",
         )
         assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
@@ -201,7 +202,7 @@ def test_logs_group_by_hostile_resource_attribute(
         token,
         int((now - timedelta(minutes=2)).timestamp() * 1000),
         int((now + timedelta(seconds=1)).timestamp() * 1000),
-        [build_scalar_query("A", "logs", [build_aggregation("count()")], group_by=[{"name": key, "fieldContext": "resource", "fieldDataType": "string"}], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_aggregation_query("A", "logs", [build_aggregation("count()")], group_by=[{"name": key, "fieldContext": "resource", "fieldDataType": "string"}], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text
@@ -236,7 +237,7 @@ def test_traces_group_by_and_select_hostile_attribute(
         token,
         start_ms,
         end_ms,
-        [build_scalar_query("A", "traces", [build_aggregation("count()")], group_by=[field], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_aggregation_query("A", "traces", [build_aggregation("count()")], group_by=[field], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text
@@ -274,6 +275,7 @@ def test_metrics_group_by_hostile_label(
         int((now - timedelta(minutes=5)).timestamp() * 1000),
         int((now + timedelta(minutes=1)).timestamp() * 1000),
         [build_builder_query("A", metric, "avg", "avg", group_by=[label])],
+        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
     labels = {tuple(sorted((entry["key"]["name"], entry["value"]) for entry in item.get("labels", []))) for item in get_all_series(response.json(), "A")}

tests/integration/tests/queriercommon/09_time_range.py

@@ -12,6 +12,7 @@ from fixtures.querier import (
     Aggregation,
     BuilderQuery,
     MetricAggregation,
+    RequestType,
     get_all_warnings,
     get_error_message,
     get_series_values,
@@ -47,6 +48,7 @@ def test_time_range_rejected(
         start_ms=start_ms,
         end_ms=end_ms,
         queries=[BuilderQuery(signal="logs", aggregations=[Aggregation(expression="count()")], step_interval=60).to_dict()],
+        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.BAD_REQUEST
@@ -79,6 +81,7 @@ def test_two_year_logs_timeseries_step_clamped(
             ).to_dict()
         ],
         format_options={"formatTableResultForUI": False, "fillGaps": True},
+        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK
@@ -117,6 +120,7 @@ def test_two_year_traces_timeseries_step_clamped(
             ).to_dict()
         ],
         format_options={"formatTableResultForUI": False, "fillGaps": True},
+        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK
@@ -155,6 +159,7 @@ def test_two_year_metrics_timeseries_step_clamped(
             ).to_dict()
         ],
         format_options={"formatTableResultForUI": False, "fillGaps": True},
+        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK

tests/integration/tests/queriercommon/10_cache_time_shift.py

@@ -7,10 +7,11 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
+    RequestType,
     build_aggregation,
+    build_aggregation_query,
     build_function,
     build_group_by_field,
-    build_scalar_query,
     get_all_series,
     index_series_by_label,
     make_query_request,
@@ -58,7 +59,7 @@ def test_time_shift_partial_hit_matches_no_cache(
         ]
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],
@@ -71,11 +72,11 @@ def test_time_shift_partial_hit_matches_no_cache(
     first_end_ms = int((now - 30 * MINUTE).timestamp() * 1000)
     second_end_ms = int((now - 10 * MINUTE).timestamp() * 1000)
 
-    warm = make_query_request(signoz, token, start_ms, first_end_ms, [query], no_cache=False)
+    warm = make_query_request(signoz, token, start_ms, first_end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
     assert warm.status_code == HTTPStatus.OK, warm.text
-    cached = make_query_request(signoz, token, start_ms, second_end_ms, [query], no_cache=False)
+    cached = make_query_request(signoz, token, start_ms, second_end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
     assert cached.status_code == HTTPStatus.OK, cached.text
-    fresh = make_query_request(signoz, token, start_ms, second_end_ms, [query], no_cache=True)
+    fresh = make_query_request(signoz, token, start_ms, second_end_ms, [query], no_cache=True, request_type=RequestType.TIME_SERIES)
     assert fresh.status_code == HTTPStatus.OK, fresh.text
 
     fresh_points = {service: {point["timestamp"]: point["value"] for point in series["values"]} for service, series in index_series_by_label(get_all_series(fresh.json(), "A"), "service.name").items()}

tests/integration/tests/queriercommon/12_cache_limited_group_by.py

@@ -7,10 +7,11 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
+    RequestType,
     build_aggregation,
+    build_aggregation_query,
     build_group_by_field,
     build_order_by,
-    build_scalar_query,
     get_all_series,
     index_series_by_label,
     make_query_request,
@@ -48,7 +49,7 @@ def test_limited_group_by_partial_hit_matches_no_cache(
         ]
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],
@@ -62,11 +63,11 @@ def test_limited_group_by_partial_hit_matches_no_cache(
     head_end_ms = int((head_start + 20 * MINUTE).timestamp() * 1000)
     tail_end_ms = int((head_start + 40 * MINUTE).timestamp() * 1000)
 
-    warm = make_query_request(signoz, token, head_start_ms, head_end_ms, [query], no_cache=False)
+    warm = make_query_request(signoz, token, head_start_ms, head_end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
     assert warm.status_code == HTTPStatus.OK, warm.text
-    cached = make_query_request(signoz, token, head_start_ms, tail_end_ms, [query], no_cache=False)
+    cached = make_query_request(signoz, token, head_start_ms, tail_end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
     assert cached.status_code == HTTPStatus.OK, cached.text
-    fresh = make_query_request(signoz, token, head_start_ms, tail_end_ms, [query], no_cache=True)
+    fresh = make_query_request(signoz, token, head_start_ms, tail_end_ms, [query], no_cache=True, request_type=RequestType.TIME_SERIES)
     assert fresh.status_code == HTTPStatus.OK, fresh.text
 
     fresh_points = {service: {point["timestamp"]: point["value"] for point in series["values"]} for service, series in index_series_by_label(get_all_series(fresh.json(), "A"), "service.name").items()}
@@ -106,7 +107,7 @@ def test_limited_group_by_shrunk_window_matches_no_cache(
         ]
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],
@@ -120,11 +121,11 @@ def test_limited_group_by_shrunk_window_matches_no_cache(
     middle_ms = int((start + 5 * MINUTE).timestamp() * 1000)
     end_ms = int((start + 10 * MINUTE).timestamp() * 1000)
 
-    warm = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
+    warm = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
     assert warm.status_code == HTTPStatus.OK, warm.text
-    cached = make_query_request(signoz, token, middle_ms, end_ms, [query], no_cache=False)
+    cached = make_query_request(signoz, token, middle_ms, end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
     assert cached.status_code == HTTPStatus.OK, cached.text
-    fresh = make_query_request(signoz, token, middle_ms, end_ms, [query], no_cache=True)
+    fresh = make_query_request(signoz, token, middle_ms, end_ms, [query], no_cache=True, request_type=RequestType.TIME_SERIES)
     assert fresh.status_code == HTTPStatus.OK, fresh.text
 
     fresh_points = {service: {point["timestamp"]: point["value"] for point in series["values"]} for service, series in index_series_by_label(get_all_series(fresh.json(), "A"), "service.name").items()}

tests/integration/tests/querierlogs/04_formula.py

@@ -9,10 +9,10 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
     build_aggregation,
+    build_aggregation_query,
     build_formula_query,
     build_group_by_field,
     build_order_by,
-    build_scalar_query,
     find_named_result,
     get_all_series,
     get_error_message,
@@ -101,7 +101,7 @@ def test_logs_formula_orderby_and_limit(
         end_ms=int(now.timestamp() * 1000),
         request_type="scalar",
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -109,7 +109,7 @@ def test_logs_formula_orderby_and_limit(
                 filter_expression="severity_text = 'ERROR'",
                 disabled=True,
             ),
-            build_scalar_query(
+            build_aggregation_query(
                 name="B",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -240,8 +240,8 @@ def test_logs_formula_having_time_series_matches_query_having(
         end_ms=int(now.timestamp() * 1000),
         request_type="time_series",
         queries=[
-            build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
-            build_scalar_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
+            build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
+            build_aggregation_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
             build_formula_query("F1", "A", having_expression=formula_having),
         ],
     )
@@ -290,8 +290,8 @@ def test_logs_formula_having_scalar_matches_query_having(
         end_ms=int(now.timestamp() * 1000),
         request_type="scalar",
         queries=[
-            build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
-            build_scalar_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
+            build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
+            build_aggregation_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
             build_formula_query("F1", "A", having_expression=formula_having),
         ],
     )
@@ -325,7 +325,7 @@ def test_logs_formula_having_invalid_reference(
         end_ms=int(now.timestamp() * 1000),
         request_type="time_series",
         queries=[
-            build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], disabled=True),
+            build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], disabled=True),
             build_formula_query("F1", "A", having_expression="count() > 0"),
         ],
     )

tests/integration/tests/querierlogs/11_aggregation_functions.py

@@ -10,9 +10,9 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
     build_aggregation,
+    build_aggregation_query,
     build_group_by_field,
     build_order_by,
-    build_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -72,7 +72,7 @@ def test_logs_aggregate_functions(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[
@@ -165,7 +165,7 @@ def test_logs_aggregate_if_predicates_use_filter_syntax(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[
@@ -240,7 +240,7 @@ def test_logs_aggregate_arguments_are_field_names(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[
@@ -288,7 +288,7 @@ def test_logs_aggregate_invalid_expression_is_bad_request(
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation(expression)])
+    query = build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation(expression)])
     response = make_scalar_query_request(signoz, token, now, [query])
 
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
@@ -335,7 +335,7 @@ def test_logs_aggregate_having(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation(expr) for expr in aggregations],

tests/integration/tests/querierlogs/12_aggregation_timeseries.py

@@ -10,9 +10,9 @@ from fixtures.logs import Logs
 from fixtures.querier import (
     RequestType,
     build_aggregation,
+    build_aggregation_query,
     build_group_by_field,
     build_order_by,
-    build_scalar_query,
     get_all_series,
     get_scalar_table_data,
     index_series_by_label,
@@ -64,7 +64,7 @@ def test_logs_aggregate_min_max(
         end_ms=int((now + timedelta(seconds=5)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[
@@ -135,7 +135,7 @@ def test_logs_aggregate_time_series_limit(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -187,7 +187,7 @@ def test_logs_aggregate_multi_group_by(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],

tests/integration/tests/querierlogs/13_unknown_key_collisions.py

@@ -10,9 +10,9 @@ from fixtures.logs import Logs
 from fixtures.querier import (
     RequestType,
     build_aggregation,
+    build_aggregation_query,
     build_group_by_field,
     build_raw_query,
-    build_scalar_query,
     get_rows,
     get_scalar_table_data,
     make_query_request,
@@ -92,7 +92,7 @@ def test_logs_unknown_key_collision_groupby_parity(
             token,
             now,
             [
-                build_scalar_query(
+                build_aggregation_query(
                     "A",
                     "logs",
                     [build_aggregation("count()")],
@@ -210,7 +210,7 @@ def test_logs_unknown_key_collision_aggregation_parity(
             token,
             now,
             [
-                build_scalar_query(
+                build_aggregation_query(
                     "A",
                     "logs",
                     [build_aggregation(f"count({key})", "cnt"), build_aggregation(f"count_distinct({key})", "distinct")],

tests/integration/tests/querierlogs/17_heatmap.py

@@ -0,0 +1,207 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logs import Logs
+from fixtures.querier import (
+    RequestType,
+    assert_distribution_sums_heatmap,
+    build_aggregation,
+    build_aggregation_query,
+    build_group_by_field,
+    build_heatmap_query,
+    build_linear_bucket_options,
+    build_log_bucket_options,
+    build_order_by,
+    get_all_series,
+    get_buckets,
+    get_error_message,
+    get_heatmap_columns,
+    index_series_by_label,
+    make_query_request,
+)
+
+
+def test_linear_heatmap_grouped(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    bytes_by_service = {"heatmap-svc-a": [50, 250], "heatmap-svc-b": [950, 5000]}
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=index + 1),
+                resources={"service.name": service},
+                attributes={"response_bytes": response_bytes},
+                body=f"{service} served {response_bytes} bytes",
+            )
+            for service, values in bytes_by_service.items()
+            for index, response_bytes in enumerate(values)
+        ]
+        # a log without the field is not counted
+        + [
+            Logs(
+                timestamp=now - timedelta(seconds=index + 1),
+                resources={"service.name": "heatmap-svc-a"},
+                body="no response size",
+            )
+            for index in range(2)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+    end_ms = int((now + timedelta(seconds=5)).timestamp() * 1000)
+    queries = [
+        build_heatmap_query(
+            "A",
+            "logs",
+            "response_bytes",
+            group_by=[build_group_by_field("service.name", "string", "resource")],
+            filter_expression="service.name IN ('heatmap-svc-a', 'heatmap-svc-b')",
+            bucket_options=build_linear_bucket_options(1000, 10),
+        )
+    ]
+    response = make_query_request(signoz, token, start_ms, end_ms, queries, request_type=RequestType.HEATMAP)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_distribution_sums_heatmap(signoz, token, start_ms, end_ms, queries, response)
+
+    data = response.json()
+    # 1000 split into 10 gives (0, 100], (100, 200], ... (900, 1000], and 5000
+    # is past 1000 so it counts in the trailing overflow
+    buckets = get_buckets(data, "A")
+    assert buckets == pytest.approx([0.0, 100.0, 200.0, 300.0, 400.0, 500.0, 600.0, 700.0, 800.0, 900.0, 1000.0])
+
+    series_by_service = index_series_by_label(get_all_series(data, "A"), "service.name")
+    assert set(series_by_service) == set(bytes_by_service)
+
+    totals_by_service = {service: [sum(column["values"][slot] for column in series["values"]) for slot in range(len(buckets) + 1)] for service, series in series_by_service.items()}
+    assert totals_by_service == {
+        "heatmap-svc-a": [0, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
+        "heatmap-svc-b": [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1],
+    }
+
+
+def test_log_heatmap(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=index + 1),
+                resources={"service.name": "heatmap-log-axis"},
+                attributes={"queue_depth": queue_depth},
+                body=f"queue depth {queue_depth}",
+            )
+            for index, queue_depth in enumerate([100, 300, 600])
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+    end_ms = int((now + timedelta(seconds=5)).timestamp() * 1000)
+    queries = [
+        build_heatmap_query(
+            "A",
+            "logs",
+            "attribute.queue_depth",
+            order=[build_order_by("attribute.queue_depth")],
+            filter_expression="service.name = 'heatmap-log-axis'",
+            bucket_options=build_log_bucket_options(0),
+        )
+    ]
+    response = make_query_request(signoz, token, start_ms, end_ms, queries, request_type=RequestType.HEATMAP)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_distribution_sums_heatmap(signoz, token, start_ms, end_ms, queries, response)
+
+    data = response.json()
+    # a doubling per bucket, with one rung under the lowest bucket reached so
+    # its lower bound is on the axis; (128, 256] holds nothing but stays
+    buckets = get_buckets(data, "A")
+    assert buckets == pytest.approx([64.0, 128.0, 256.0, 512.0, 1024.0])
+    assert [sum(column["values"][slot] for column in get_heatmap_columns(data, "A")) for slot in range(len(buckets) + 1)] == [0, 1, 0, 1, 1, 0]
+
+
+def test_default_axis(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=index + 1),
+                resources={"service.name": "heatmap-default-axis"},
+                attributes={"batch_size": 1000},
+                body="batch of 1000",
+            )
+            for index in range(2)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+    end_ms = int((now + timedelta(seconds=5)).timestamp() * 1000)
+    queries = [build_heatmap_query("A", "logs", "batch_size", filter_expression="service.name = 'heatmap-default-axis'")]
+    response = make_query_request(signoz, token, start_ms, end_ms, queries, request_type=RequestType.HEATMAP)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_distribution_sums_heatmap(signoz, token, start_ms, end_ms, queries, response)
+
+    data = response.json()
+    # no bucketOptions is the finest log axis, 16 buckets per doubling: 1000
+    # sits in (2^(159/16), 2^10]
+    buckets = get_buckets(data, "A")
+    totals = [sum(column["values"][slot] for column in get_heatmap_columns(data, "A")) for slot in range(len(buckets) + 1)]
+    holding = [slot for slot, total in enumerate(totals) if total]
+    assert holding == [len(buckets) - 1]
+    assert totals[holding[0]] == 2
+    assert buckets[-2:] == pytest.approx([2 ** (159 / 16), 1024.0])
+
+
+@pytest.mark.parametrize(
+    "aggregations, expected_message",
+    [
+        pytest.param(
+            [build_aggregation("count()")],
+            'a logs or traces heatmap counts events by the field named as its aggregation, got "count()"',
+            id="count_aggregation",
+        ),
+        pytest.param(
+            [build_aggregation("response_bytes"), build_aggregation("queue_depth")],
+            "a heatmap renders one bucket axis, but the query has 2 aggregations",
+            id="two_aggregations",
+        ),
+    ],
+)
+def test_heatmap_rejections(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    aggregations: list[dict],
+    expected_message: str,
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        int((now - timedelta(minutes=5)).timestamp() * 1000),
+        int(now.timestamp() * 1000),
+        [build_aggregation_query("A", "logs", aggregations)],
+        request_type=RequestType.HEATMAP,
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert expected_message in get_error_message(response.json())

tests/integration/tests/queriermetrics/02_warnings.py

@@ -7,6 +7,7 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    RequestType,
     build_builder_query,
     get_all_warnings,
     make_query_request,
@@ -50,12 +51,12 @@ def test_histogram_p90_returns_warning_outside_data_window(
     end_ms = int(now.timestamp() * 1000)
 
     start_4h = int((now - timedelta(hours=4)).timestamp() * 1000)
-    response = make_query_request(signoz, token, start_4h, end_ms, [query])
+    response = make_query_request(signoz, token, start_4h, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
     assert response.json()["status"] == "success"
 
     start_15m = int((now - timedelta(minutes=15)).timestamp() * 1000)
-    response = make_query_request(signoz, token, start_15m, end_ms, [query])
+    response = make_query_request(signoz, token, start_15m, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
     data = response.json()
     warnings = get_all_warnings(data)
@@ -83,7 +84,7 @@ def test_non_existent_metrics_returns_warning(
     end_ms = int(now.timestamp() * 1000)
 
     start_2h = int((now - timedelta(hours=2)).timestamp() * 1000)
-    response = make_query_request(signoz, token, start_2h, end_ms, [query])
+    response = make_query_request(signoz, token, start_2h, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -111,7 +112,7 @@ def test_non_existent_internal_metrics_returns_no_warning(
     end_ms = int(now.timestamp() * 1000)
 
     start_2h = int((now - timedelta(hours=2)).timestamp() * 1000)
-    response = make_query_request(signoz, token, start_2h, end_ms, [query])
+    response = make_query_request(signoz, token, start_2h, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
     data = response.json()
     assert get_all_warnings(data) == []
@@ -172,6 +173,7 @@ def test_variable_in_filter_returns_no_warning(
         end_ms,
         [query],
         variables={"tag": {"type": "query", "value": "service-a"}},
+        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK

tests/integration/tests/queriermetrics/04_histogram.py

@@ -9,6 +9,7 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    RequestType,
     build_builder_query,
     get_all_series,
     get_series_values,
@@ -75,7 +76,7 @@ def test_histogram_count_for_one_endpoint(
         filter_expression='endpoint = "/health"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -142,7 +143,7 @@ def test_histogram_count_for_one_service(
         filter_expression='service = "api"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -212,7 +213,7 @@ def test_histogram_count_for_delta_service(
         filter_expression='service = "web"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -271,7 +272,7 @@ def test_histogram_count_for_all_services(
         ## no services filter, this tests for multitemporality handling as well
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -308,7 +309,7 @@ def test_histogram_count_no_param(
         "count",
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -394,7 +395,7 @@ def test_histogram_percentile_for_all_services(
         space_agg,
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -445,7 +446,7 @@ def test_histogram_percentile_for_cumulative_service(
         filter_expression='service = "api"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -500,6 +501,7 @@ def test_histogram_percentile_with_a_gap_in_one_bucket(
         start_ms,
         end_ms,
         [build_builder_query("A", metric_name, "doesnotreallymatter", "p50")],
+        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
 
@@ -548,7 +550,7 @@ def test_histogram_percentile_for_delta_service(
         filter_expression='service = "web"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -590,6 +592,7 @@ def test_histogram_percentile_explicit_le_group_by(
             build_builder_query("A", metric_name, "doesnotreallymatter", "p95", group_by=["le", "service"]),
             build_builder_query("B", metric_name, "doesnotreallymatter", "p95", group_by=["service"]),
         ],
+        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
 

tests/integration/tests/queriermetrics/05_multi_temporality.py

@@ -10,6 +10,7 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    RequestType,
     build_builder_query,
     get_all_series,
     get_series_values,
@@ -59,7 +60,7 @@ def test_with_steady_values_and_reset(
         filter_expression='endpoint = "/orders"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -118,7 +119,7 @@ def test_group_by_endpoint(
         group_by=["endpoint"],
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -242,7 +243,7 @@ def test_for_service_with_switch(
         filter_expression='service = "api"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -294,7 +295,7 @@ def test_for_week_long_time_range(
         "sum",
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
     data = response.json()
     result_values = sorted(get_series_values(data, "A"), key=lambda x: x["timestamp"])
@@ -338,7 +339,7 @@ def test_for_month_long_time_range(
         "sum",
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
     data = response.json()
     result_values = sorted(get_series_values(data, "A"), key=lambda x: x["timestamp"])

tests/integration/tests/queriermetrics/06_gauge.py

@@ -9,6 +9,7 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    RequestType,
     build_builder_query,
     get_series_values,
     make_query_request,
@@ -72,7 +73,7 @@ def test_for_one_service(
         filter_expression=f'service = "{service}"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -128,7 +129,7 @@ def test_for_multiple_aggregations(
         space_agg,
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()

tests/integration/tests/queriermetrics/07_rate_cumulative.py

@@ -8,6 +8,7 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    RequestType,
     build_builder_query,
     get_all_series,
     get_series_values,
@@ -46,7 +47,7 @@ def test_rate_with_steady_values_and_reset(
         filter_expression='endpoint = "/orders"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -90,7 +91,7 @@ def test_rate_group_by_endpoint(
         group_by=["endpoint"],
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()

tests/integration/tests/queriermetrics/08_rate_delta.py

@@ -8,6 +8,7 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    RequestType,
     build_builder_query,
     get_all_series,
     get_series_values,
@@ -45,7 +46,7 @@ def test_rate_with_steady_values_and_reset(
         filter_expression='endpoint = "/orders"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -87,7 +88,7 @@ def test_rate_group_by_endpoint(
         group_by=["endpoint"],
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()

tests/integration/tests/queriermetrics/09_cost_meter.py

@@ -8,6 +8,7 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.meter import MeterSample, make_meter_samples
 from fixtures.querier import (
+    RequestType,
     build_builder_query,
     get_series_values,
     make_query_request,
@@ -49,7 +50,7 @@ def test_query_range_cost_meter(
         temporality="delta",
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()

tests/integration/tests/queriermetrics/10_key_resolution.py

@@ -45,7 +45,7 @@ def test_metrics_group_by_context_collapse(
             token,
             now,
             [
-                querier.build_scalar_query(
+                querier.build_aggregation_query(
                     name="A",
                     signal="metrics",
                     aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -93,7 +93,7 @@ def test_metrics_filter_label_context(
             token,
             now,
             [
-                querier.build_scalar_query(
+                querier.build_aggregation_query(
                     name="A",
                     signal="metrics",
                     aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -113,7 +113,7 @@ def test_metrics_filter_label_context(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -154,7 +154,7 @@ def test_metrics_group_by_unknown_label(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -199,7 +199,7 @@ def test_metrics_filter_unknown_label_matches_nothing(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -245,7 +245,7 @@ def test_metrics_full_text_filter_does_not_error(
             token,
             now,
             [
-                querier.build_scalar_query(
+                querier.build_aggregation_query(
                     name="A",
                     signal="metrics",
                     aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -290,7 +290,7 @@ def test_metrics_group_by_label_named_value(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],

tests/integration/tests/queriermetrics/12_delta_nonmonotonic_sum.py

@@ -7,7 +7,7 @@ import pytest
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import build_builder_query, get_series_values, make_query_request
+from fixtures.querier import RequestType, build_builder_query, get_series_values, make_query_request
 
 
 # A delta, non-monotonic Sum queried without an explicit type must be treated as
@@ -53,7 +53,7 @@ def test_delta_nonmonotonic_sum_is_treated_as_sum(
     # No type and no temporality: the server resolves both from the seeded series.
     query = build_builder_query("A", metric_name, time_aggregation, "sum")
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK, response.text
 
     values = get_series_values(response.json(), "A")

tests/integration/tests/queriermetrics/13_bool_label_filter.py

@@ -52,7 +52,7 @@ def test_metrics_filter_bool_label(
             token,
             now,
             [
-                querier.build_scalar_query(
+                querier.build_aggregation_query(
                     name="A",
                     signal="metrics",
                     aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],

tests/integration/tests/queriermetrics/14_exp_histogram.py

@@ -8,6 +8,7 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import ExpHistogramMetrics
 from fixtures.querier import (
+    RequestType,
     build_builder_query,
     get_all_series,
     get_all_warnings,
@@ -74,7 +75,7 @@ def test_exp_histogram_percentile_delta_grouped(
         group_by=["service.name"],
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK, response.text
 
     data = response.json()
@@ -123,7 +124,7 @@ def test_exp_histogram_percentile_delta_merges_across_series(
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
     query = build_builder_query("A", metric_name, "", "p95", temporality="delta")
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK, response.text
 
     data = response.json()
@@ -164,7 +165,7 @@ def test_exp_histogram_percentile_cumulative_is_rejected(
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
     query = build_builder_query("A", metric_name, "", "p95", temporality="cumulative")
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
 
     # the request is well formed; it is the stored metric that cannot be read, so
     # this is reported as unsupported rather than as bad input

tests/integration/tests/queriermetrics/17_cache.py

@@ -7,6 +7,7 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    RequestType,
     assert_results_equal,
     build_builder_query,
     get_series_values,
@@ -56,16 +57,16 @@ def test_builder_shortening_the_time_range_at_the_end(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert base_query.status_code == HTTPStatus.OK, base_query.text
     points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
     returned_points = [(point["value"], point.get("partial", False)) for point in points]
     assert returned_points == [(16, False), (4096, False)]
 
-    from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=False)
+    from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
-    uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=True)
+    uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=True, request_type=RequestType.TIME_SERIES)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened end")
@@ -116,16 +117,16 @@ def test_builder_shortening_the_time_range_at_the_start(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, no_cache=False)
+    base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert base_query.status_code == HTTPStatus.OK, base_query.text
     points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
     returned_points = [(point["value"], point.get("partial", False)) for point in points]
     assert returned_points == [(65536, False), (4096, False)]
 
-    from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=False)
+    from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
-    uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=True)
+    uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=True, request_type=RequestType.TIME_SERIES)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened start")
@@ -178,16 +179,16 @@ def test_builder_time_range_shorter_than_the_step(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert base_query.status_code == HTTPStatus.OK, base_query.text
     points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
     returned_points = [(point["value"], point.get("partial", False)) for point in points]
     assert returned_points == [(16, False), (4096, False)]
 
-    from_cache = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=False)
+    from_cache = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
-    uncached = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=True)
+    uncached = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=True, request_type=RequestType.TIME_SERIES)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "range shorter than the step")
@@ -238,10 +239,10 @@ def test_promql_running_the_same_query_twice(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
+    first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert first.status_code == HTTPStatus.OK, first.text
 
-    second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
+    second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert second.status_code == HTTPStatus.OK, second.text
 
     assert_results_equal(first.json(), second.json(), "A", "the same query twice")
@@ -296,7 +297,7 @@ def test_promql_shifting_the_time_range(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, no_cache=False)
+    aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert aligned_and_cached.status_code == HTTPStatus.OK, aligned_and_cached.text
 
     ## at each timestamp t, promql takes the highest sample in (t-2minutes, t],
@@ -312,7 +313,7 @@ def test_promql_shifting_the_time_range(
 
     # an unaligned range is rounded down to the step, so this asks for the same
     # instants as the aligned query and gets the same answer
-    unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=False)
+    unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert unaligned_and_cached.status_code == HTTPStatus.OK, unaligned_and_cached.text
     points = sorted(get_series_values(unaligned_and_cached.json(), "A"), key=lambda point: point["timestamp"])
     assert [(point["timestamp"], point["value"]) for point in points] == aligned_points, "unaligned query, rounded to the step"
@@ -329,7 +330,7 @@ def test_promql_shifting_the_time_range(
         (unaligned_end_time_ms, 1100),  # t = 3m30s
     ]
     for run in ("first", "second"):
-        unaligned_as_asked = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, unaligned_query, no_cache=False)
+        unaligned_as_asked = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, unaligned_query, no_cache=False, request_type=RequestType.TIME_SERIES)
         assert unaligned_as_asked.status_code == HTTPStatus.OK, unaligned_as_asked.text
         points = sorted(get_series_values(unaligned_as_asked.json(), "A"), key=lambda point: point["timestamp"])
         assert [(point["timestamp"], point["value"]) for point in points] == unaligned_points, f"unaligned query as asked, {run} run"
@@ -371,17 +372,17 @@ def test_promql_time_range_shorter_than_the_step(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert base_query.status_code == HTTPStatus.OK, base_query.text
     points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
     assert [(point["timestamp"], point["value"]) for point in points] == [(start_time_ms, 100), (end_time_ms_base_query, 1100)]
 
     # a range shorter than the step rounds down to the single instant at minute
     # 10, the instant the base query closed on
-    from_cache = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=False)
+    from_cache = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
     assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
-    uncached = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=True)
+    uncached = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=True, request_type=RequestType.TIME_SERIES)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "range shorter than the step")
@@ -427,7 +428,7 @@ def test_builder_refreshing_a_sliding_time_range(
     # a dashboard left open on a one hour range, re-running a minute later each time
     for refresh in range(20):
         refresh_start_ms = start_time_ms + refresh * MINUTE_MS
-        from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 60 * MINUTE_MS, query, no_cache=False)
+        from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 60 * MINUTE_MS, query, no_cache=False, request_type=RequestType.TIME_SERIES)
         assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
         # each refresh is stitched out of overlapping cached ranges, so this catches
@@ -438,7 +439,7 @@ def test_builder_refreshing_a_sliding_time_range(
         assert returned_points == expected_points, f"refresh {refresh} did not return the minutes it covers"
 
     last_refresh_start_ms = start_time_ms + 19 * MINUTE_MS
-    uncached = make_query_request(signoz, token, last_refresh_start_ms, last_refresh_start_ms + 60 * MINUTE_MS, query, no_cache=True)
+    uncached = make_query_request(signoz, token, last_refresh_start_ms, last_refresh_start_ms + 60 * MINUTE_MS, query, no_cache=True, request_type=RequestType.TIME_SERIES)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "the twentieth refresh")

tests/integration/tests/queriermetrics/19_offset.py

@@ -7,7 +7,7 @@ import pytest
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import build_builder_query, get_all_series, make_query_request
+from fixtures.querier import RequestType, build_builder_query, get_all_series, make_query_request
 
 
 @pytest.mark.parametrize(
@@ -62,6 +62,7 @@ def test_metrics_time_series_offset_skips_series_in_order(
                 offset=offset,
             )
         ],
+        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK, response.text

tests/integration/tests/querierscalar/01_logs.py

@@ -46,7 +46,7 @@ def build_logs_query(
     gb = [querier.build_group_by_field(f, "string", "resource") for f in group_by] if group_by else None
     order = [querier.build_order_by(name, direction) for name, direction in order_by] if order_by else None
 
-    return querier.build_scalar_query(
+    return querier.build_aggregation_query(
         name=name,
         signal="logs",
         aggregations=aggs,

tests/integration/tests/querierscalar/02_traces.py

@@ -52,7 +52,7 @@ def build_traces_query(
     gb = [querier.build_group_by_field(f, "string", "resource") for f in group_by] if group_by else None
     order = [querier.build_order_by(name, direction) for name, direction in order_by] if order_by else None
 
-    return querier.build_scalar_query(
+    return querier.build_aggregation_query(
         name=name,
         signal="traces",
         aggregations=aggs,

tests/integration/tests/querierscalar/03_metrics.py

@@ -48,7 +48,7 @@ def build_metrics_query(
     gb = [querier.build_group_by_field(f, "string", "attribute") for f in group_by] if group_by else None
     order = [querier.build_order_by(name, direction) for name, direction in order_by] if order_by else None
 
-    return querier.build_scalar_query(
+    return querier.build_aggregation_query(
         name=name,
         signal="metrics",
         aggregations=aggs,

tests/integration/tests/querierscalar/04_having.py

@@ -8,9 +8,9 @@ from fixtures.logs import Logs
 from fixtures.querier import (
     assert_scalar_result_order,
     build_aggregation,
+    build_aggregation_query,
     build_group_by_field,
     build_order_by,
-    build_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -41,7 +41,7 @@ def test_logs_scalar_group_by_having(
     insert_logs(logs)
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],

tests/integration/tests/querierscalar/05_reduce_to.py

@@ -9,10 +9,10 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
     assert_scalar_result_order,
+    build_aggregation_query,
     build_group_by_field,
     build_metrics_aggregation,
     build_order_by,
-    build_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -59,7 +59,7 @@ def test_metrics_scalar_reduce_to_modes(
 
     aggregation = build_metrics_aggregation("test.reduce.metric", "latest", "sum", "unspecified")
     aggregation["reduceTo"] = reduce_to
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="metrics",
         aggregations=[aggregation],
@@ -118,7 +118,7 @@ def test_metrics_scalar_limit_ranks_by_reduced_value(
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation(metric_name, "latest", "sum", "unspecified", reduce_to=reduce_to)],
@@ -175,7 +175,7 @@ def test_metrics_scalar_limit_ranks_after_functions(
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation(metric_name, "latest", "sum", "unspecified", reduce_to="max")],
@@ -200,7 +200,7 @@ def test_metrics_scalar_missing_reduce_to_rejected(
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation("test.reduce.metric", "latest", "sum", "unspecified")],
@@ -221,7 +221,7 @@ def test_metrics_scalar_invalid_reduce_to_rejected(
 
     aggregation = build_metrics_aggregation("test.reduce.metric", "latest", "sum", "unspecified")
     aggregation["reduceTo"] = "p99"
-    query = build_scalar_query(name="A", signal="metrics", aggregations=[aggregation])
+    query = build_aggregation_query(name="A", signal="metrics", aggregations=[aggregation])
     response = make_scalar_query_request(signoz, token, now, [query])
 
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text

tests/integration/tests/querierscalar/06_metrics_aggregations.py

@@ -8,9 +8,9 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
+    build_aggregation_query,
     build_group_by_field,
     build_metrics_aggregation,
-    build_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -52,7 +52,7 @@ def test_metrics_scalar_space_aggregations(
     insert_metrics([Metrics(metric_name="test.metric", labels={"service.name": service}, timestamp=now - timedelta(seconds=1), temporality="Unspecified", type_="Gauge", is_monotonic=False, value=value) for service, value in METRIC_VALUES.items()])
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation("test.metric", "latest", space_aggregation, "unspecified", reduce_to="last")],
@@ -93,7 +93,7 @@ def test_metrics_scalar_having(
     insert_metrics([Metrics(metric_name="test.metric", labels={"service.name": service}, timestamp=now - timedelta(seconds=1), temporality="Unspecified", type_="Gauge", is_monotonic=False, value=value) for service, value in METRIC_VALUES.items()])
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_scalar_query(
+    query = build_aggregation_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation("test.metric", "latest", "sum", "unspecified", reduce_to="last")],

tests/integration/tests/querierscalar/07_unknown_keys.py

@@ -32,7 +32,7 @@ def test_logs_scalar_filter_unknown_key_synthesizes(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[querier.build_aggregation("count()")],
@@ -64,7 +64,7 @@ def test_logs_scalar_group_by_over_empty_filter_returns_no_rows(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[querier.build_aggregation("count()")],
@@ -95,7 +95,7 @@ def test_logs_scalar_group_by_unknown_key_null_bucket(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[querier.build_aggregation("count()")],
@@ -142,7 +142,7 @@ def test_traces_scalar_filter_unknown_key_synthesizes(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="traces",
                 aggregations=[querier.build_aggregation("count()")],
@@ -187,7 +187,7 @@ def test_traces_scalar_group_by_unknown_key_null_bucket(
         token,
         now,
         [
-            querier.build_scalar_query(
+            querier.build_aggregation_query(
                 name="A",
                 signal="traces",
                 aggregations=[querier.build_aggregation("count()")],

tests/integration/tests/querierscalar/09_offset.py

@@ -11,11 +11,11 @@ from fixtures.metrics import Metrics
 from fixtures.querier import (
     RequestType,
     build_aggregation,
+    build_aggregation_query,
     build_formula_query,
     build_group_by_field,
     build_metrics_aggregation,
     build_order_by,
-    build_scalar_query,
     find_named_result,
     get_scalar_table_data,
     make_query_request,
@@ -62,7 +62,7 @@ def test_logs_scalar_offset_skips_rows_in_order(
         token,
         now,
         [
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -117,7 +117,7 @@ def test_traces_scalar_offset_skips_rows_in_order(
         token,
         now,
         [
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="traces",
                 aggregations=[build_aggregation("count()")],
@@ -162,7 +162,7 @@ def test_metrics_scalar_offset_skips_rows_in_order(
         token,
         now,
         [
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation("offset.gauge", "latest", "sum", "unspecified", reduce_to="last")],
@@ -205,7 +205,7 @@ def test_formula_scalar_offset_skips_rows_in_order(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -240,7 +240,7 @@ def test_negative_offset_is_rejected(
         signoz,
         token,
         datetime.now(tz=UTC).replace(second=0, microsecond=0),
-        [build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=-1)],
+        [build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=-1)],
     )
 
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
@@ -257,7 +257,7 @@ def test_offset_without_limit_is_rejected(
         signoz,
         token,
         datetime.now(tz=UTC).replace(second=0, microsecond=0),
-        [build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=1)],
+        [build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=1)],
     )
 
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text

tests/integration/tests/queriertraces/02_aggregation.py

@@ -18,7 +18,7 @@ from fixtures.querier import (
     build_aggregation,
     build_group_by_field,
     build_order_by,
-    build_traces_scalar_query,
+    build_traces_aggregation_query,
     get_all_series,
     get_scalar_table_data,
     index_series_by_label,
@@ -208,7 +208,7 @@ def test_traces_aggregate_functions(
         build_aggregation("rate_sum(duration_nano)", "rate_sum_d"),
         build_aggregation("avg(latency_ms)", "avg_lat"),
     ]
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=aggregations,
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", "desc")],
@@ -284,7 +284,7 @@ def test_traces_aggregate_calculated_range_countif(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=[build_aggregation("countIf(response_status_code >= 400 AND response_status_code < 500)")],
     )
     response = make_scalar_query_request(signoz, token, now, [query])
@@ -342,7 +342,7 @@ def test_traces_aggregate_having(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=[build_aggregation("count()")],
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", "desc")],
@@ -418,7 +418,7 @@ def test_traces_aggregate_time_series_limit(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=[build_aggregation("count()")],
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", direction)],
@@ -495,7 +495,7 @@ def test_traces_aggregate_unknown_key(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_scalar_query(aggregations=[build_aggregation(expression)])
+    query = build_traces_aggregation_query(aggregations=[build_aggregation(expression)])
     response = make_scalar_query_request(signoz, token, now, [query])
 
     assert response.status_code == HTTPStatus.OK, response.text

tests/integration/tests/queriertraces/03_fill.py

@@ -12,7 +12,7 @@ from fixtures.querier import (
     build_aggregation,
     build_formula_query,
     build_group_by_field,
-    build_traces_scalar_query,
+    build_traces_aggregation_query,
     find_named_result,
     get_all_series,
     index_series_by_label,
@@ -72,7 +72,7 @@ def test_traces_fill_plain(
         start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
-        queries=[build_traces_scalar_query(aggregations=[build_aggregation("count()")], functions=functions)],
+        queries=[build_traces_aggregation_query(aggregations=[build_aggregation("count()")], functions=functions)],
         format_options=format_options,
     )
 
@@ -138,7 +138,7 @@ def test_traces_fill_group_by(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("service.name", "string", "resource")],
                 functions=functions,
@@ -212,8 +212,8 @@ def test_traces_fill_formula(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(aggregations=[build_aggregation("count()")], filter_expression="resource.service.name = 'test'", disabled=True),
-            build_traces_scalar_query(name="B", aggregations=[build_aggregation("count()")], filter_expression="resource.service.name = 'another-test'", disabled=True),
+            build_traces_aggregation_query(aggregations=[build_aggregation("count()")], filter_expression="resource.service.name = 'test'", disabled=True),
+            build_traces_aggregation_query(name="B", aggregations=[build_aggregation("count()")], filter_expression="resource.service.name = 'another-test'", disabled=True),
             build_formula_query("F1", "A + B", functions=functions),
         ],
         format_options=format_options,
@@ -284,8 +284,8 @@ def test_traces_fill_formula_group_by(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
-            build_traces_scalar_query(name="B", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
+            build_traces_aggregation_query(aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
+            build_traces_aggregation_query(name="B", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
             build_formula_query("F1", "A + B", functions=functions),
         ],
         format_options=format_options,

tests/integration/tests/queriertraces/05_resource_evolution.py

@@ -9,7 +9,7 @@ from fixtures.querier import (
     assert_grouped_series,
     build_aggregation,
     build_group_by_field,
-    build_traces_scalar_query,
+    build_traces_aggregation_query,
     get_resource_evolution_time,
     index_series_by_label,
     make_query_request,
@@ -99,7 +99,7 @@ def test_traces_resource_evolution(
         end_ms=int((before_1 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("service.name")],
             )
@@ -131,7 +131,7 @@ def test_traces_resource_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("service.name")],
             )
@@ -163,7 +163,7 @@ def test_traces_resource_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("service.name")],
             )
@@ -202,7 +202,7 @@ def test_traces_resource_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count_distinct(service.name)")],
                 group_by=[build_group_by_field("deployment.environment")],
             )

tests/integration/tests/queriertraces/08_aggregation_min_max.py

@@ -10,7 +10,7 @@ from fixtures.querier import (
     RequestType,
     build_aggregation,
     build_group_by_field,
-    build_traces_scalar_query,
+    build_traces_aggregation_query,
     index_series_by_label,
     make_query_request,
 )
@@ -71,7 +71,7 @@ def test_traces_aggregate_min_max(
         end_ms=int((now + timedelta(seconds=5)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[
                     build_aggregation("max(duration_nano)", "maxDuration"),
                     build_aggregation("min(duration_nano)", "minDuration"),

tests/integration/tests/queriertraces/09_unknown_keys.py

@@ -11,7 +11,7 @@ from fixtures.querier import (
     build_aggregation,
     build_group_by_field,
     build_raw_query,
-    build_traces_scalar_query,
+    build_traces_aggregation_query,
     get_rows,
     get_scalar_table_data,
     make_query_request,
@@ -44,7 +44,7 @@ def test_traces_filter_unknown_key_synthesizes(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 [build_aggregation("count()")],
                 filter_expression='totally.unknown.key = "x"',
             )
@@ -152,7 +152,7 @@ def test_traces_group_by_unknown_key_null_bucket(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 [build_aggregation("count()")],
                 group_by=[build_group_by_field("totally.unknown.key", field_data_type="", field_context="")],
                 filter_expression='service.name = "groupby-unknown-svc"',

tests/integration/tests/queriertraces/11_aggregation_depth.py

@@ -11,7 +11,7 @@ from fixtures.querier import (
     build_aggregation,
     build_group_by_field,
     build_order_by,
-    build_traces_scalar_query,
+    build_traces_aggregation_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -73,7 +73,7 @@ def test_traces_aggregate_percentiles(
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     percentiles = [25, 50, 90, 95, 99]
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=[build_aggregation(f"p{p}(duration_nano)", f"p{p}") for p in percentiles],
     )
     response = make_scalar_query_request(signoz, token, now, [query])
@@ -138,7 +138,7 @@ def test_traces_aggregate_group_by_non_resource(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=[build_aggregation("count()")],
         group_by=[build_group_by_field(group_key, field_data_type="", field_context="")],
         filter_expression="resource.service.name = 'gb-svc'",
@@ -193,7 +193,7 @@ def test_traces_aggregate_multi_group_by(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=[build_aggregation("count()")],
         group_by=[
             build_group_by_field("service.name", "string", "resource"),
@@ -262,7 +262,7 @@ def test_traces_aggregate_having_breadth(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=[build_aggregation("count()"), build_aggregation("sum(duration_nano)")],
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", "desc")],
@@ -325,7 +325,7 @@ def test_traces_duration_nano_qol_filter_with_collision(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_scalar_query(
+    query = build_traces_aggregation_query(
         aggregations=[build_aggregation("count()", "cnt")],
         filter_expression=duration_filter,
     )

tests/integration/tests/queriertraces/13_attributes_json_evolution.py

@@ -12,7 +12,7 @@ from fixtures.querier import (
     assert_grouped_series,
     build_aggregation,
     build_group_by_field,
-    build_traces_scalar_query,
+    build_traces_aggregation_query,
     get_all_warnings,
     get_preview_sql,
     get_rows,
@@ -90,7 +90,7 @@ def test_traces_attributes_json_evolution(
         end_ms=int((before_1 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("http.route", field_context="attribute")],
             )
@@ -114,7 +114,7 @@ def test_traces_attributes_json_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("http.route", field_context="attribute")],
             )
@@ -138,7 +138,7 @@ def test_traces_attributes_json_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("http.route", field_context="attribute")],
             )
@@ -212,7 +212,7 @@ def test_traces_attributes_json_typed_filters(
             end_ms=end_ms,
             request_type=RequestType.TIME_SERIES,
             queries=[
-                build_traces_scalar_query(
+                build_traces_aggregation_query(
                     aggregations=[build_aggregation("count()")],
                     group_by=[build_group_by_field("http.route", field_context="attribute")],
                     filter_expression=filter_expression,
@@ -291,7 +291,7 @@ def test_traces_attributes_json_collision_and_map_parity(
         end_ms=end_ms,
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("app.status", field_context="attribute")],
                 filter_expression="app.status EXISTS",
@@ -325,7 +325,7 @@ def test_traces_attributes_json_collision_and_map_parity(
             end_ms=end_ms,
             request_type=RequestType.TIME_SERIES,
             queries=[
-                build_traces_scalar_query(
+                build_traces_aggregation_query(
                     aggregations=[build_aggregation("count()")],
                     group_by=[build_group_by_field("http.route", field_context="attribute")],
                     filter_expression=filter_expression,
@@ -382,7 +382,7 @@ def test_traces_attributes_json_number_reads(
             end_ms=end_ms,
             request_type=RequestType.TIME_SERIES,
             queries=[
-                build_traces_scalar_query(
+                build_traces_aggregation_query(
                     aggregations=[build_aggregation("count()")],
                     group_by=[build_group_by_field("http.route", field_context="attribute")],
                     filter_expression=filter_expression,
@@ -513,7 +513,7 @@ def test_traces_attributes_json_unreported_key(
     insert_traces(spans)
 
     queries = [
-        build_traces_scalar_query(
+        build_traces_aggregation_query(
             aggregations=[build_aggregation("count()")],
             filter_expression="unreported.key = 'value'",
         )
@@ -523,11 +523,11 @@ def test_traces_attributes_json_unreported_key(
     start_ms = int((before - timedelta(minutes=1)).timestamp() * 1000)
     end_ms = int((before + timedelta(minutes=1)).timestamp() * 1000)
 
-    sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries), "A")
+    sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries, request_type=RequestType.TIME_SERIES), "A")
     assert "attributes_string['unreported.key']" in sql
     assert "attributes.`unreported.key`" not in sql
 
-    response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries)
+    response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries, request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
     series = response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
     assert [value["value"] for value in series[0]["values"]] == [1]
@@ -537,11 +537,11 @@ def test_traces_attributes_json_unreported_key(
     start_ms = int((after - timedelta(minutes=1)).timestamp() * 1000)
     end_ms = int((after + timedelta(minutes=1)).timestamp() * 1000)
 
-    sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries), "A")
+    sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries, request_type=RequestType.TIME_SERIES), "A")
     assert "attributes.`unreported.key`" in sql
     assert "attributes_string" not in sql
 
-    response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries)
+    response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries, request_type=RequestType.TIME_SERIES)
     assert response.status_code == HTTPStatus.OK
     series = response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
     assert [value["value"] for value in series[0]["values"]] == [1]

tests/integration/tests/queriertraces/14_skip_indexes.py

@@ -7,7 +7,7 @@ import requests
 
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
-from fixtures.querier import RequestType, build_aggregation, build_traces_scalar_query, get_preview_statements, make_preview_query_request
+from fixtures.querier import RequestType, build_aggregation, build_traces_aggregation_query, get_preview_statements, make_preview_query_request
 from fixtures.traces import ATTRIBUTE_JSON_ROLLOUT_TIME, Traces
 
 PROMOTED_PATHS = "/api/v1/promoted_paths"
@@ -72,7 +72,7 @@ def test_traces_filters_use_skip_indexes(
                 end_ms=int(now.timestamp() * 1000),
                 request_type=RequestType.SCALAR,
                 # count() over primary-key filters is answered by the exact count projection, which reads no index
-                queries=[build_traces_scalar_query(aggregations=[build_aggregation("sum(duration_nano)")], filter_expression=filter_expression)],
+                queries=[build_traces_aggregation_query(aggregations=[build_aggregation("sum(duration_nano)")], filter_expression=filter_expression)],
             ),
             "A",
         )

tests/integration/tests/queriertraces/16_heatmap.py

@@ -0,0 +1,113 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import (
+    RequestType,
+    assert_distribution_sums_heatmap,
+    build_heatmap_query,
+    build_linear_bucket_options,
+    build_log_bucket_options,
+    get_buckets,
+    get_heatmap_columns,
+    make_query_request,
+)
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind
+
+
+def test_duration_heatmap(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    durations = [timedelta(0)] + [timedelta(milliseconds=5)] * 6 + [timedelta(seconds=1)] * 3 + [timedelta(seconds=30)]
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(minutes=2, seconds=index),
+                duration=duration,
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                name="GET /cart",
+                kind=TracesKind.SPAN_KIND_SERVER,
+                resources={"service.name": "heatmap-cart"},
+            )
+            for index, duration in enumerate(durations)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+    end_ms = int(now.timestamp() * 1000)
+    queries = [build_heatmap_query("A", "traces", "duration_nano", filter_expression="service.name = 'heatmap-cart'", bucket_options=build_log_bucket_options(0))]
+    response = make_query_request(signoz, token, start_ms, end_ms, queries, request_type=RequestType.HEATMAP)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_distribution_sums_heatmap(signoz, token, start_ms, end_ms, queries, response)
+
+    data = response.json()
+    # a doubling per bucket in nanoseconds: 5ms is in (2^22, 2^23], 1s in
+    # (2^29, 2^30] and 30s in (2^34, 2^35], and 2^22 is the rung under 5ms; a
+    # 0ns span is counted at 0
+    buckets = get_buckets(data, "A")
+    assert buckets == pytest.approx([0.0] + [float(2**power) for power in range(22, 36)])
+
+    totals = [sum(column["values"][slot] for column in get_heatmap_columns(data, "A")) for slot in range(len(buckets) + 1)]
+    assert {buckets[slot]: total for slot, total in enumerate(totals) if total} == {0.0: 1, 2.0**23: 6, 2.0**30: 3, 2.0**35: 1}
+    assert sum(totals) == len(durations)
+
+
+def test_span_attribute_heatmap(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    rows_returned = [5, 5, 5, 45, 45, 250]
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(minutes=2, seconds=index),
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                name="SELECT orders",
+                kind=TracesKind.SPAN_KIND_CLIENT,
+                resources={"service.name": "heatmap-orders-db"},
+                attributes={"db.rows_returned": rows},
+            )
+            for index, rows in enumerate(rows_returned)
+        ]
+        # a span without the attribute is not counted
+        + [
+            Traces(
+                timestamp=now - timedelta(minutes=2, seconds=index),
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                name="BEGIN",
+                kind=TracesKind.SPAN_KIND_CLIENT,
+                resources={"service.name": "heatmap-orders-db"},
+            )
+            for index in range(2)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+    end_ms = int(now.timestamp() * 1000)
+    queries = [build_heatmap_query("A", "traces", "db.rows_returned", filter_expression="service.name = 'heatmap-orders-db'", bucket_options=build_linear_bucket_options(100, 10))]
+    response = make_query_request(signoz, token, start_ms, end_ms, queries, request_type=RequestType.HEATMAP)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_distribution_sums_heatmap(signoz, token, start_ms, end_ms, queries, response)
+
+    data = response.json()
+    # 100 split into 10 gives (0, 10], (10, 20], ... (90, 100]: 5 is in the
+    # first, 45 in (40, 50], and 250 is past 100 so it counts in the overflow
+    buckets = get_buckets(data, "A")
+    assert buckets == pytest.approx([0.0, 10.0, 20.0, 30.0, 40.0, 50.0])
+    assert [sum(column["values"][slot] for column in get_heatmap_columns(data, "A")) for slot in range(len(buckets) + 1)] == [0, 3, 0, 0, 0, 2, 1]

tests/integration/tests/semconvfamilies/01_family_matrix.py

@@ -10,11 +10,11 @@ from fixtures.metadata import AttributesMetadata, get_field_keys, get_field_valu
 from fixtures.querier import (
     RequestType,
     build_aggregation,
+    build_aggregation_query,
     build_group_by_field,
     build_order_by,
     build_raw_query,
-    build_scalar_query,
-    build_traces_scalar_query,
+    build_traces_aggregation_query,
     get_all_warnings,
     get_column_data_from_response,
     get_scalar_columns,
@@ -178,7 +178,7 @@ def test_group_by_merges_and_echoes_requested_spelling(
         end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 [build_aggregation("count_distinct(name)")],
                 filter_expression=f"service.name LIKE '{PREFIX}%'",
                 group_by=[build_group_by_field(requested_key, "string", "resource")],
@@ -214,7 +214,7 @@ def test_log_group_by_merges_and_echoes_requested_spelling(
         end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count_distinct(body)")],
@@ -245,7 +245,7 @@ def test_time_series_group_by_merges_family(
         end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_scalar_query(
+            build_traces_aggregation_query(
                 [build_aggregation("count_distinct(name)")],
                 filter_expression=f"service.name LIKE '{PREFIX}%'",
                 group_by=[build_group_by_field(CURRENT_KEY, "string", "resource")],

tests/integration/tests/semconvfamilies/02_metric_families.py

@@ -9,9 +9,9 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
     RequestType,
+    build_aggregation_query,
     build_group_by_field,
     build_metrics_aggregation,
-    build_scalar_query,
     get_scalar_columns,
     get_scalar_table_data,
     make_query_request,
@@ -40,7 +40,7 @@ def test_metric_label_filter_merges_stored_spellings(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(LABEL_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -69,7 +69,7 @@ def test_metric_label_group_by_merges_stored_spellings(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(LABEL_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -113,7 +113,7 @@ def test_span_metric_label_filter_reads_resource_layout(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(SPAN_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -142,7 +142,7 @@ def test_metric_name_family_unions_storage_names(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(requested, "latest", "sum", "unspecified", reduce_to="last")],
@@ -177,7 +177,7 @@ def test_metric_name_union_double_counts_dual_emission(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(CURRENT_NAME_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -203,7 +203,7 @@ def test_metric_family_stays_literal_with_flag_off(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(LABEL_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -223,7 +223,7 @@ def test_metric_family_stays_literal_with_flag_off(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_scalar_query(
+            build_aggregation_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(OLD_NAME_METRIC, "latest", "sum", "unspecified", reduce_to="last")],