Compare commits

...

12 Commits

Author SHA1 Message Date
nityanandagohain
d250f190a7 Squashed commit of the following:
commit 6d0b607f41b80583b2f4dc015c728d57fd0a3f6a
Author: nityanandagohain <nityanandagohain@gmail.com>
Date:   Wed Jul 15 10:30:43 2026 +0530

    fix: update tests

commit 20f42474b0901e49cc745155a2c70bb829988a94
Merge: 36dca5dab0 31efe177a4
Author: nityanandagohain <nityanandagohain@gmail.com>
Date:   Tue Jul 14 23:09:00 2026 +0530

    Merge branch 'issue_5601': scoped-trace builder package + port aggregations

    Merges the telemetryscopedtraces refactor (generic scoped-trace topology moved
    out of telemetryai, feature-flagged gen_ai key enrichment, tracefield. context
    classification) and ports the scalar/time-series work onto it:

    - trace_aggregation.go / trace_having.go now live in telemetryscopedtraces
    - trace-level HAVING resolution composes the canonical pkg/variables
      replacement (dynamic __all__ drops a condition for any operator) with the
      standard filter pipeline, so values still bind as args
    - buildPerTraceScan adopts the merged embedExpr placeholder/arg checking
    - ColumnProvider gains ActivityGateAlias; gen_ai provider gates trace-level
      aggregations on llm_call_count

commit 36dca5dab0c0abb08aeb2645fee07f28cc764eb8
Author: nityanandagohain <nityanandagohain@gmail.com>
Date:   Tue Jul 14 23:09:00 2026 +0530

    feat: trace-level aggregations for scalar/time-series and trace-scoped span list

    - trace.-prefixed aggregations (avg(trace.output_tokens), count(trace.trace_id),
      arithmetic) run over window-clipped per-trace values via a native CTE pipeline
    - a trace-level filter condition qualifies traces in scalar/time-series/raw via
      __qualified / __trace_scope, resource-fingerprint pruned like the trace list
    - trace-level filter conditions resolve through the standard filter pipeline
      (bound args, operators, query variables) against the per-trace column aliases
    - per-trace rows with no LLM span in the window are dropped from trace-level
      aggregations (LLM-activity gate)
    - targeted rejections for group-by/order-by on trace-level columns
2026-07-15 12:38:57 +05:30
nityanandagohain
97c49c870b feat: support ai trace aggregate filtering in ai span list 2026-07-15 12:27:55 +05:30
nityanandagohain
31efe177a4 fix: address comments 2026-07-14 18:53:30 +05:30
nityanandagohain
d502d12ac3 fix: update openapi 2026-07-10 14:27:07 +05:30
nityanandagohain
bd9f15a716 fix: update integration test 2026-07-10 14:21:16 +05:30
nityanandagohain
813ef988c9 fix: edge cases and correct cost key 2026-07-10 12:06:34 +05:30
nityanandagohain
40e6799285 fix: add resource fingerprint cte 2026-07-10 00:36:25 +05:30
nityanandagohain
1caa60a3cd fix: cleanup and more tests 2026-07-09 23:54:05 +05:30
nityanandagohain
3f781f0083 fix: more cleanup 2026-07-09 12:39:01 +05:30
nityanandagohain
6aec05cf7a fix: more tests 2026-07-09 08:45:22 +05:30
nityanandagohain
683a52f35a fix: take perf into consideration 2026-07-09 08:45:22 +05:30
nityanandagohain
e924fa1e62 feat: support llm trace list and span list 2026-07-09 08:45:20 +05:30
36 changed files with 6004 additions and 21 deletions

View File

@@ -53,6 +53,7 @@ jobs:
- queriermetrics
- querierscalar
- queriercommon
- querierai
- rawexportdata
- role
- rootuser

View File

@@ -8312,6 +8312,7 @@ components:
TelemetrytypesSource:
enum:
- meter
- ai
type: string
TelemetrytypesTelemetryFieldKey:
properties:

View File

@@ -3631,6 +3631,7 @@ export enum Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTypesQue
}
export enum TelemetrytypesSourceDTO {
meter = 'meter',
ai = 'ai',
}
export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTypesQuerybuildertypesQuerybuildertypesv5LogAggregationDTO {
/**

View File

@@ -213,18 +213,18 @@ func (module *module) discoverModels(ctx context.Context, orgID valuer.UUID) ([]
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: fmt.Sprintf("%s EXISTS", llmpricingruletypes.GenAIRequestModel)},
Filter: &qbtypes.Filter{Expression: fmt.Sprintf("%s EXISTS", telemetrytypes.GenAIRequestModel)},
Aggregations: []qbtypes.TraceAggregation{
{Expression: "count()", Alias: "spanCount"},
},
GroupBy: []qbtypes.GroupByKey{
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: llmpricingruletypes.GenAIRequestModel,
Name: telemetrytypes.GenAIRequestModel,
FieldContext: telemetrytypes.FieldContextSpan,
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: llmpricingruletypes.GenAIProviderName,
Name: telemetrytypes.GenAIProviderName,
FieldContext: telemetrytypes.FieldContextSpan,
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
@@ -254,9 +254,9 @@ func (module *module) discoverModels(ctx context.Context, orgID valuer.UUID) ([]
switch c.Type {
case qbtypes.ColumnTypeGroup:
switch c.Name {
case llmpricingruletypes.GenAIRequestModel:
case telemetrytypes.GenAIRequestModel:
modelIdx = i
case llmpricingruletypes.GenAIProviderName:
case telemetrytypes.GenAIProviderName:
providerIdx = i
}
case qbtypes.ColumnTypeAggregation:

View File

@@ -42,6 +42,7 @@ type querier struct {
metadataStore telemetrytypes.MetadataStore
promEngine prometheus.Prometheus
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation]
@@ -61,6 +62,7 @@ func New(
metadataStore telemetrytypes.MetadataStore,
promEngine prometheus.Prometheus,
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
@@ -82,6 +84,7 @@ func New(
metadataStore: metadataStore,
promEngine: promEngine,
traceStmtBuilder: traceStmtBuilder,
aiTraceStmtBuilder: aiTraceStmtBuilder,
logStmtBuilder: logStmtBuilder,
auditStmtBuilder: auditStmtBuilder,
metricStmtBuilder: metricStmtBuilder,
@@ -235,10 +238,18 @@ func (q *querier) buildQueries(
case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
spec.ShiftBy = extractShiftFromBuilderQuery(spec)
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
bq := newBuilderQuery(q.logger, q.telemetryStore, q.traceStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
stmtBuilder := q.traceStmtBuilder
if spec.Source == telemetrytypes.SourceAI {
event.Source = telemetrytypes.SourceAI.StringValue()
stmtBuilder = q.aiTraceStmtBuilder
}
bq := newBuilderQuery(q.logger, q.telemetryStore, stmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
queries[spec.Name] = bq
steps[spec.Name] = spec.StepInterval
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
if spec.Source == telemetrytypes.SourceAI {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "source \"ai\" is only supported for the traces signal, not logs")
}
spec.ShiftBy = extractShiftFromBuilderQuery(spec)
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
stmtBuilder := q.logStmtBuilder
@@ -249,6 +260,9 @@ func (q *querier) buildQueries(
queries[spec.Name] = bq
steps[spec.Name] = spec.StepInterval
case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
if spec.Source == telemetrytypes.SourceAI {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "source \"ai\" is only supported for the traces signal, not metrics")
}
// Spec was already patched by resolveMetricMetadata. Queries
// whose every aggregation was missing live in
// missingMetricQuerySet and produce empty preseeded results
@@ -479,6 +493,10 @@ func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qb
if query.Type == qbtypes.QueryTypeBuilder {
switch spec := query.Spec.(type) {
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
if spec.Source == telemetrytypes.SourceAI {
client.Error <- errors.NewInvalidInputf(errors.CodeInvalidInput, "source \"ai\" is only supported for the traces signal, not logs")
return
}
event.FilterApplied = spec.Filter != nil && spec.Filter.Expression != ""
default:
// return if it's not log aggregation
@@ -858,7 +876,11 @@ func (q *querier) createRangedQuery(originalQuery qbtypes.Query, timeRange qbtyp
specCopy := qt.spec.Copy()
specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
return newBuilderQuery(q.logger, q.telemetryStore, q.traceStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
shiftStmtBuilder := q.traceStmtBuilder
if qt.spec.Source == telemetrytypes.SourceAI {
shiftStmtBuilder = q.aiTraceStmtBuilder
}
return newBuilderQuery(q.logger, q.telemetryStore, shiftStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
case *builderQuery[qbtypes.LogAggregation]:
specCopy := qt.spec.Copy()

View File

@@ -49,6 +49,7 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
metadataStore,
nil, // prometheus
nil, // traceStmtBuilder
nil, // aiTraceStmtBuilder
nil, // logStmtBuilder
nil, // auditStmtBuilder
nil, // metricStmtBuilder
@@ -121,6 +122,7 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
metadataStore,
nil, // prometheus
nil, // traceStmtBuilder
nil, // aiTraceStmtBuilder
nil, // logStmtBuilder
nil, // auditStmtBuilder
&mockMetricStmtBuilder{}, // metricStmtBuilder

View File

@@ -9,6 +9,7 @@ import (
"github.com/SigNoz/signoz/pkg/prometheus"
"github.com/SigNoz/signoz/pkg/querier"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryai"
"github.com/SigNoz/signoz/pkg/telemetryaudit"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/telemetrymetadata"
@@ -92,6 +93,22 @@ func newProvider(
cfg.SkipResourceFingerprint.Threshold,
)
// AI trace statement builder (source=ai). The gen_ai gate/column keys are
// surfaced by the metadata store itself (enrichWithGenAIKeys), so queries work
// before any gen_ai metadata is ingested — no per-builder decoration needed.
// The standard trace builder doubles as the delegate for the span-list path.
aiBaseCondition := telemetryai.NewGenAIBaseConditionProvider()
aiTraceStmtBuilder := telemetryai.NewAITraceStatementBuilder(
settings,
telemetryMetadataStore,
aiBaseCondition,
traceStmtBuilder,
telemetryStore,
flagger,
cfg.SkipResourceFingerprint.Enabled,
cfg.SkipResourceFingerprint.Threshold,
)
// Create trace operator statement builder
traceOperatorStmtBuilder := telemetrytraces.NewTraceOperatorStatementBuilder(
settings,
@@ -185,6 +202,7 @@ func newProvider(
telemetryMetadataStore,
prometheus,
traceStmtBuilder,
aiTraceStmtBuilder,
logStmtBuilder,
auditStmtBuilder,
metricStmtBuilder,

View File

@@ -48,6 +48,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
metadataStore,
nil, // prometheus
nil, // traceStmtBuilder
nil, // aiTraceStmtBuilder
nil, // logStmtBuilder
nil, // auditStmtBuilder
metricStmtBuilder,
@@ -103,6 +104,7 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
metadataStore,
nil, // prometheus
nil, // traceStmtBuilder
nil, // aiTraceStmtBuilder
logStmtBuilder, // logStmtBuilder
nil, // auditStmtBuilder
nil, // metricStmtBuilder
@@ -152,6 +154,7 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
metadataStore,
nil, // prometheus
traceStmtBuilder, // traceStmtBuilder
nil, // aiTraceStmtBuilder
nil, // logStmtBuilder
nil, // auditStmtBuilder
nil, // metricStmtBuilder

View File

@@ -0,0 +1,76 @@
package querybuilder
import (
"strings"
"github.com/SigNoz/signoz/pkg/errors"
grammar "github.com/SigNoz/signoz/pkg/parser/filterquery/grammar"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/antlr4-go/antlr/v4"
)
// ExprKeys returns the field keys referenced in *key positions* of a filter
// expression. Unlike QueryStringToKeysSelectors (which scans raw KEY tokens and so
// also picks up unquoted values — in `x > $threshold` it reports `$threshold`), this
// walks the parse tree and collects only KeyContext nodes.
func ExprKeys(query string) []*telemetrytypes.TelemetryFieldKey {
var keys []*telemetrytypes.TelemetryFieldKey
var walk func(node antlr.Tree)
walk = func(node antlr.Tree) {
if kc, ok := node.(*grammar.KeyContext); ok {
key := telemetrytypes.GetFieldKeyFromKeyText(kc.GetText())
keys = append(keys, &key)
return
}
for i := 0; i < node.GetChildCount(); i++ {
walk(node.GetChild(i))
}
}
walk(parseFilterQuery(query))
return keys
}
// ValidateVariablesInExpr checks the variable references in an expression's value
// positions upfront, so a broken reference fails with a targeted error instead of
// the where-clause visitor's combined "Found N errors" (whose details ride in the
// error's additionals). Lookup mirrors the visitor: verbatim, then with a leading
// `$` stripped. A `$`-prefixed token that resolves to nothing is an error — it can
// never be a valid literal; a bare token that resolves to nothing is left to mean
// itself.
func ValidateVariablesInExpr(query string, variables map[string]qbtypes.VariableItem) error {
var err error
var walk func(node antlr.Tree)
walk = func(node antlr.Tree) {
if err != nil {
return
}
if vc, ok := node.(*grammar.ValueContext); ok {
// only unquoted textual values can be variable references
if vc.KEY() == nil {
return
}
text := vc.GetText()
item, ok := variables[text]
if !ok {
item, ok = variables[strings.TrimPrefix(text, "$")]
}
if !ok {
if strings.HasPrefix(text, "$") {
err = errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown variable %q", text)
}
return
}
if values, isList := item.Value.([]any); isList && len(values) == 0 {
err = errors.NewInvalidInputf(errors.CodeInvalidInput,
"variable %q used in expression has an empty list value", strings.TrimPrefix(text, "$"))
}
return
}
for i := 0; i < node.GetChildCount(); i++ {
walk(node.GetChild(i))
}
}
walk(parseFilterQuery(query))
return err
}

View File

@@ -0,0 +1,39 @@
package querybuilder
import (
"testing"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/stretchr/testify/require"
)
func TestExprKeys(t *testing.T) {
names := func(expr string) []string {
var out []string
for _, k := range ExprKeys(expr) {
out = append(out, k.Name)
}
return out
}
// value-position tokens are not keys, unlike QueryStringToKeysSelectors
require.Equal(t, []string{"output_tokens"}, names("output_tokens > $threshold"))
require.Equal(t, []string{"trace.output_tokens"}, names("trace.output_tokens > 1000"))
require.Equal(t, []string{"a", "b"}, names("a > 1 AND b IN ('x', 'y')"))
}
func TestValidateVariablesInExpr(t *testing.T) {
vars := map[string]qbtypes.VariableItem{
"threshold": {Type: qbtypes.TextBoxVariableType, Value: float64(1000)},
"empty": {Type: qbtypes.QueryVariableType, Value: []any{}},
"all": {Type: qbtypes.DynamicVariableType, Value: "__all__"},
}
require.NoError(t, ValidateVariablesInExpr("x > $threshold", vars))
require.NoError(t, ValidateVariablesInExpr("x > threshold", vars))
require.NoError(t, ValidateVariablesInExpr("x IN $all", vars))
require.NoError(t, ValidateVariablesInExpr("m = 'cost$usd'", vars)) // quoted literals are not references
require.NoError(t, ValidateVariablesInExpr("x > bare_word", vars)) // bare non-variable means itself
require.ErrorContains(t, ValidateVariablesInExpr("x > $bogus", vars), `unknown variable "$bogus"`)
require.ErrorContains(t, ValidateVariablesInExpr("x IN $empty", vars), "empty list")
}

View File

@@ -0,0 +1,160 @@
package querybuilder
import (
"strings"
"github.com/SigNoz/signoz/pkg/errors"
grammar "github.com/SigNoz/signoz/pkg/parser/filterquery/grammar"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/antlr4-go/antlr/v4"
)
// SplitFilterForAggregates partitions a single filter expression into a span-level
// part (a WHERE over spans) and a trace-level part (a HAVING over per-trace
// aggregates), splitting on the top-level AND.
//
// A key is trace-level when, with no explicit field context, its name is in
// aggregateNames — written bare (`completion_tokens`) or with the user-facing `trace.`
// prefix (`trace.completion_tokens`). Any explicit context (`span.`, `resource.`, …) is
// span-level. Trace-level and span-level keys may be AND-combined (they run at different
// query stages) but not OR-combined; an OR that mixes the two is an error.
//
// Syntax errors are ignored here — each part is re-parsed downstream (PrepareWhereClause
// for the span part, the HAVING rewriter for the trace part), which surface them.
func SplitFilterForAggregates(query string, aggregateNames map[string]struct{}) (spanExpr string, havingExpr string, err error) {
if strings.TrimSpace(query) == "" {
return "", "", nil
}
s := filterSplitter{query: []rune(query), aggregateNames: aggregateNames}
s.visit(parseFilterQuery(query))
if s.mixed {
return "", "", errors.NewInvalidInputf(errors.CodeInvalidInput,
"trace-level and span-level filters cannot be combined within an OR/NOT group; separate them with a top-level AND")
}
return strings.Join(s.span, " AND "), strings.Join(s.having, " AND "), nil
}
func parseFilterQuery(query string) antlr.Tree {
lexer := grammar.NewFilterQueryLexer(antlr.NewInputStream(query))
lexer.RemoveErrorListeners()
parser := grammar.NewFilterQueryParser(antlr.NewCommonTokenStream(lexer, 0))
parser.RemoveErrorListeners()
return parser.Query()
}
// filterSplitter walks the parse tree once, flattening the top-level AND chain and
// routing each atom (a comparison, a NOT expression, or a whole multi-branch OR group)
// to the span or having bucket by the class of the keys it references.
type filterSplitter struct {
query []rune
aggregateNames map[string]struct{}
span []string
having []string
mixed bool
}
func (s *filterSplitter) visit(node antlr.Tree) {
switch n := node.(type) {
case *grammar.QueryContext:
if n.Expression() != nil {
s.visit(n.Expression())
}
case *grammar.ExpressionContext:
if n.OrExpression() != nil {
s.visit(n.OrExpression())
}
case *grammar.OrExpressionContext:
// a single branch is just an AND chain; multiple branches are a real OR, kept
// whole so a class-mixing OR can be rejected.
if ands := n.AllAndExpression(); len(ands) == 1 {
s.visit(ands[0])
} else {
s.route(n)
}
case *grammar.AndExpressionContext:
for _, u := range n.AllUnaryExpression() {
s.visit(u)
}
case *grammar.UnaryExpressionContext:
if n.NOT() != nil {
s.route(n)
} else if n.Primary() != nil {
s.visit(n.Primary())
}
case *grammar.PrimaryContext:
if n.OrExpression() != nil { // parenthesized sub-expression
s.visit(n.OrExpression())
} else {
s.route(n)
}
}
}
// route classifies an atom and appends its original source text to the right bucket.
func (s *filterSplitter) route(atom antlr.ParserRuleContext) {
isTrace, isSpan := classifyKeys(atom, s.aggregateNames)
if isTrace && isSpan {
s.mixed = true
return
}
text := atomSourceText(s.query, atom)
// A multi-branch OR group's source slice excludes its enclosing parens (they belong
// to the parent Primary). Re-wrap it so rejoining a bucket with " AND " cannot invert
// OR/AND precedence, e.g. `a AND (b OR c)` must not flatten to `a AND b OR c`.
if or, ok := atom.(*grammar.OrExpressionContext); ok && len(or.AllAndExpression()) > 1 {
text = "(" + text + ")"
}
if isTrace {
s.having = append(s.having, text)
} else {
s.span = append(s.span, text)
}
}
// classifyKeys reports whether a subtree references trace-level and/or span-level keys.
// A key is trace-level when it has no explicit field context and its name — after the
// optional user-facing `trace.` prefix is stripped — is a known aggregate, or when it
// carries the trace field context explicitly (`tracefield.`, which Normalize parses
// into FieldContextTrace; no span field mapper resolves that context, so it can only
// mean a trace-level aggregate — an unknown name is then rejected by the aggregate
// validation with a targeted error instead of failing as an unknown span field). Any
// other explicit context (`span.`, `resource.`, …) is span-level.
func classifyKeys(node antlr.Tree, aggregateNames map[string]struct{}) (isTrace, isSpan bool) {
kc, ok := node.(*grammar.KeyContext)
if ok {
key := telemetrytypes.GetFieldKeyFromKeyText(kc.GetText())
switch key.FieldContext {
case telemetrytypes.FieldContextUnspecified:
// `trace.` is the user-facing prefix for trace-level aggregates. It is not a
// registered field context, so it stays on the name; strip it before matching.
name := strings.TrimPrefix(key.Name, telemetrytypes.FieldContextTrace.StringValue()+".")
_, isTrace = aggregateNames[name]
isSpan = !isTrace
case telemetrytypes.FieldContextTrace:
isTrace = true
default:
isSpan = true
}
return
}
for i := 0; i < node.GetChildCount(); i++ {
t, s := classifyKeys(node.GetChild(i), aggregateNames)
isTrace = isTrace || t
isSpan = isSpan || s
}
return
}
// atomSourceText returns the original source substring for an atom, preserving
// whitespace. The token stream drops skipped whitespace, which would glue word
// operators (OR/AND/NOT) to their operands, so slice the input by token offsets.
// ANTLR offsets are rune indices (InputStream holds []rune), hence the rune slice.
func atomSourceText(query []rune, atom antlr.ParserRuleContext) string {
start, stop := atom.GetStart(), atom.GetStop()
if start == nil || stop == nil || start.GetStart() < 0 || stop.GetStop() >= len(query) || stop.GetStop() < start.GetStart() {
return atom.GetText()
}
return string(query[start.GetStart() : stop.GetStop()+1])
}

View File

@@ -0,0 +1,174 @@
package querybuilder
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestSplitFilterForAggregates(t *testing.T) {
agg := map[string]struct{}{"completion_tokens": {}, "span_count": {}, "prompt_tokens": {}}
type tc struct {
name string
query string
span string // expected span-level (WHERE) part; "" => empty
having string // expected trace-level (HAVING) part; "" => empty
wantErr bool
}
cases := []tc{
// --- empty input ---------------------------------------------------------
{
name: "empty",
},
{
name: "whitespace only",
query: " ",
},
// --- single class --------------------------------------------------------
{
name: "span only",
query: "service.name = 'x'",
span: "service.name = 'x'",
},
{
name: "agg only bare",
query: "completion_tokens > 1000",
having: "completion_tokens > 1000",
},
{
// the user-facing `trace.` prefix marks a trace-level aggregate.
name: "agg only trace prefix",
query: "trace.completion_tokens > 1000",
having: "trace.completion_tokens > 1000",
},
{
// `tracefield.` is the explicit trace field context (Normalize parses it into
// FieldContextTrace), so it marks a trace-level aggregate like `trace.`.
name: "agg only tracefield prefix",
query: "tracefield.completion_tokens > 1000",
having: "tracefield.completion_tokens > 1000",
},
{
// an unknown name under the explicit trace context still routes trace-level,
// so the aggregate validation rejects it with a targeted error instead of the
// span path failing on an unknown field.
name: "unknown agg under tracefield prefix stays trace-level",
query: "tracefield.not_an_aggregate > 1000",
having: "tracefield.not_an_aggregate > 1000",
},
{
// ANTLR token offsets are rune indices; slicing must not shift after a
// multi-byte char (this used to truncate 1000 → 100).
name: "unicode value before the split",
query: "service.name = 'héllo' AND completion_tokens > 1000",
span: "service.name = 'héllo'",
having: "completion_tokens > 1000",
},
// --- top-level AND splits across the two buckets -------------------------
{
name: "span AND agg",
query: "service.name = 'x' AND completion_tokens > 1000",
span: "service.name = 'x'",
having: "completion_tokens > 1000",
},
{
// order within a bucket is preserved; the two span atoms join with AND.
name: "span AND span AND agg",
query: "service.name = 'x' AND kind_string = 'Internal' AND completion_tokens > 1000",
span: "service.name = 'x' AND kind_string = 'Internal'",
having: "completion_tokens > 1000",
},
{
// a parenthesized top-level AND still splits across the two buckets.
name: "parenthesized span AND agg",
query: "(service.name = 'x' AND completion_tokens > 1000)",
span: "service.name = 'x'",
having: "completion_tokens > 1000",
},
// --- OR groups are re-wrapped in parens so a later AND-join can't invert
// precedence (`a AND (b OR c)` must not flatten to `a AND b OR c`) ------
{
name: "agg OR agg",
query: "completion_tokens > 1000 OR span_count > 3",
having: "(completion_tokens > 1000 OR span_count > 3)",
},
{
name: "span OR span",
query: "service.name = 'x' OR kind_string = 'Internal'",
span: "(service.name = 'x' OR kind_string = 'Internal')",
},
{
name: "span AND (span OR span)",
query: "service.name = 'x' AND (kind_string = 'Internal' OR kind_string = 'Client')",
span: "service.name = 'x' AND (kind_string = 'Internal' OR kind_string = 'Client')",
},
{
name: "agg AND (agg OR agg)",
query: "prompt_tokens > 5 AND (completion_tokens > 1000 OR span_count > 3)",
having: "prompt_tokens > 5 AND (completion_tokens > 1000 OR span_count > 3)",
},
{
// the OR group routes to span, the trailing aggregate to having.
name: "span AND (span OR span) AND agg",
query: "a.b = 'x' AND (c.d = 'y' OR e.f = 'z') AND completion_tokens > 1000",
span: "a.b = 'x' AND (c.d = 'y' OR e.f = 'z')",
having: "completion_tokens > 1000",
},
// --- a nested AND group flattens across the buckets (no spurious parens) --
{
name: "(span AND agg) AND agg",
query: "(service.name = 'x' AND completion_tokens > 1000) AND prompt_tokens > 5",
span: "service.name = 'x'",
having: "completion_tokens > 1000 AND prompt_tokens > 5",
},
// --- NOT wrapping a single-class group is routed whole to that class ------
{
name: "not agg",
query: "NOT (completion_tokens > 1000)",
having: "NOT (completion_tokens > 1000)",
},
{
name: "not span",
query: "NOT (service.name = 'x')",
span: "NOT (service.name = 'x')",
},
// --- class-mixing is rejected in an OR group, a NOT group, or a nested OR -
{
name: "agg OR span rejected",
query: "completion_tokens > 1000 OR service.name = 'x'",
wantErr: true,
},
{
name: "not mixed rejected",
query: "NOT (completion_tokens > 1000 AND service.name = 'x')",
wantErr: true,
},
{
name: "span AND (agg OR span) rejected",
query: "service.name = 'x' AND (completion_tokens > 1000 OR kind_string = 'Client')",
wantErr: true,
},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
span, having, err := SplitFilterForAggregates(c.query, agg)
if c.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
require.Equal(t, c.span, span, "span part")
require.Equal(t, c.having, having, "having part")
})
}
}

View File

@@ -18,6 +18,19 @@ func NewHavingExpressionRewriter() *HavingExpressionRewriter {
}
}
// Rewrite rewrites and validates a HAVING expression against a caller-supplied
// column map (user-facing name -> SQL identifier/expression). Values are inlined, so
// the result is a bare SQL boolean expression with no bound args. Used by callers
// that project their own aggregate columns (e.g. the AI trace list) rather than the
// query's Aggregations.
func (r *HavingExpressionRewriter) Rewrite(expression string, columnMap map[string]string) (string, error) {
if len(strings.TrimSpace(expression)) == 0 {
return "", nil
}
r.columnMap = columnMap
return r.rewriteAndValidate(expression)
}
// RewriteForTraces rewrites and validates the HAVING expression for a traces query.
func (r *HavingExpressionRewriter) RewriteForTraces(expression string, aggregations []qbtypes.TraceAggregation) (string, error) {
if len(strings.TrimSpace(expression)) == 0 {

View File

@@ -0,0 +1,40 @@
package telemetryai
import (
"strings"
scopedtraces "github.com/SigNoz/signoz/pkg/telemetryscopedtraces"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// genAIBaseConditionProvider: an AI trace has >=1 gen_ai LLM, tool, or agent span.
type genAIBaseConditionProvider struct {
keys []string
}
var _ scopedtraces.BaseConditionProvider = (*genAIBaseConditionProvider)(nil)
func NewGenAIBaseConditionProvider() scopedtraces.BaseConditionProvider {
return &genAIBaseConditionProvider{
keys: []string{telemetrytypes.GenAIRequestModel, telemetrytypes.GenAIToolName, telemetrytypes.GenAIAgentName},
}
}
func (p *genAIBaseConditionProvider) FilterExpression() string {
parts := make([]string, 0, len(p.keys))
for _, k := range p.keys {
parts = append(parts, k+" EXISTS")
}
return strings.Join(parts, " OR ")
}
func (p *genAIBaseConditionProvider) FieldKeys() []*telemetrytypes.TelemetryFieldKey {
// Definitions come from GenAIFieldDefinitions so they can't drift from the
// canonical semconv keys; copy to take the address.
keys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(p.keys))
for _, k := range p.keys {
def := telemetrytypes.GenAIFieldDefinitions[k]
keys = append(keys, &def)
}
return keys
}

View File

@@ -0,0 +1,69 @@
package telemetryai
import (
scopedtraces "github.com/SigNoz/signoz/pkg/telemetryscopedtraces"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// genAIColumnProvider adds AI/LLM per-trace metrics on top of the common columns.
type genAIColumnProvider struct{}
var _ scopedtraces.ColumnProvider = (*genAIColumnProvider)(nil)
func NewGenAIColumnProvider() scopedtraces.ColumnProvider {
return &genAIColumnProvider{}
}
func (genAIColumnProvider) Columns() []scopedtraces.TraceColumn {
defs := telemetrytypes.GenAIFieldDefinitions
reqModel := defs[telemetrytypes.GenAIRequestModel]
toolName := defs[telemetrytypes.GenAIToolName]
inTok := defs[telemetrytypes.GenAIUsageInputTokens]
outTok := defs[telemetrytypes.GenAIUsageOutputTokens]
cost := defs[telemetrytypes.SignozGenAITotalCost]
inMsg := defs[telemetrytypes.GenAIInputMessages]
outMsg := defs[telemetrytypes.GenAIOutputMessages]
str := telemetrytypes.FieldDataTypeString
return append(scopedtraces.CommonTraceColumns(),
// LLM calls only (request model present), not the full gate.
scopedtraces.TraceColumn{Alias: "llm_call_count", Orderable: true, Expr: scopedtraces.CountExists(&reqModel)},
scopedtraces.TraceColumn{Alias: "tool_call_count", Orderable: true, Expr: scopedtraces.CountExists(&toolName)},
scopedtraces.TraceColumn{Alias: "distinct_tool_count", Orderable: true, Expr: scopedtraces.UniqCount(&toolName, str)},
// tokens live only on LLM spans, so a plain sum needs no gate scoping.
scopedtraces.TraceColumn{Alias: "input_tokens", Orderable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &inTok)},
scopedtraces.TraceColumn{Alias: "output_tokens", Orderable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &outTok)},
scopedtraces.TraceColumn{Alias: "total_tokens", Orderable: true, Expr: scopedtraces.SumOfKeys(telemetrytypes.FieldDataTypeFloat64, &inTok, &outTok)},
// per-span cost attached by the SigNoz LLM pricing processor.
scopedtraces.TraceColumn{Alias: "estimated_cost_usd", Orderable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &cost)},
// slowest single LLM call in the trace.
scopedtraces.TraceColumn{Alias: "max_llm_latency_ns", Orderable: true, Expr: scopedtraces.ScopedToKeyColumn(scopedtraces.AggMax, "duration_nano", &reqModel)},
// errors across the whole trace (any span), so display-only.
scopedtraces.TraceColumn{Alias: "error_count", Expr: scopedtraces.PredicateCount("has_error = true")},
// timestamp of the last gen_ai span (LLM/tool/agent), hence gate-scoped.
scopedtraces.TraceColumn{Alias: "last_activity_time", Orderable: true, Expr: scopedtraces.ScopedReduce(scopedtraces.AggMax, "timestamp")},
// previews: first call's input (the prompt), last call's output (the answer).
scopedtraces.TraceColumn{Alias: "input", SpanLevel: true, Expr: scopedtraces.PickBy(&inMsg, str, "timestamp", scopedtraces.PickEarliest)},
scopedtraces.TraceColumn{Alias: "output", SpanLevel: true, Expr: scopedtraces.PickBy(&outMsg, str, "timestamp", scopedtraces.PickLatest)},
)
}
func (genAIColumnProvider) DefaultOrderAlias() string { return "last_activity_time" }
// Trace-level aggregations only see traces with LLM activity in the window/bucket —
// a tool/agent-only slice would contribute NULL tokens but still count as a trace,
// making count(trace.trace_id) and avg(trace.output_tokens) disagree.
func (genAIColumnProvider) ActivityGateAlias() string { return "llm_call_count" }
func (p genAIColumnProvider) AggregateAliases() []string {
// Derived from Columns() so a new column can't be forgotten; SpanLevel columns
// are filtered span-level, so skip them.
cols := p.Columns()
aliases := make([]string, 0, len(cols))
for _, c := range cols {
if !c.SpanLevel {
aliases = append(aliases, c.Alias)
}
}
return aliases
}

View File

@@ -0,0 +1,26 @@
package telemetryai
import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
scopedtraces "github.com/SigNoz/signoz/pkg/telemetryscopedtraces"
"github.com/SigNoz/signoz/pkg/telemetrystore"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// NewAITraceStatementBuilder wires the generic scoped-trace builder with the gen_ai
// gate and AI columns. This package holds only gen_ai domain knowledge; the query
// topology lives in telemetryscopedtraces.
func NewAITraceStatementBuilder(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
baseCond scopedtraces.BaseConditionProvider,
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
telemetryStore telemetrystore.TelemetryStore,
fl flagger.Flagger,
skipResourceFingerprintEnable bool,
skipResourceFingerprintThreshold uint64,
) qbtypes.StatementBuilder[qbtypes.TraceAggregation] {
return scopedtraces.NewScopedTraceStatementBuilder(settings, metadataStore, baseCond, NewGenAIColumnProvider(), traceStmtBuilder, telemetryStore, fl, skipResourceFingerprintEnable, skipResourceFingerprintThreshold)
}

View File

@@ -0,0 +1,955 @@
package telemetryai
import (
"context"
"fmt"
"strings"
"testing"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
scopedtraces "github.com/SigNoz/signoz/pkg/telemetryscopedtraces"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/stretchr/testify/require"
)
// otelKeysMap seeds the OpenTelemetry gen_ai semantic-convention keys the AI
// queries reference, so the metadata-backed field resolution succeeds in tests.
func otelKeysMap() map[string][]*telemetrytypes.TelemetryFieldKey {
strKey := func(name string) *telemetrytypes.TelemetryFieldKey {
return &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
}
}
numKey := func(name string) *telemetrytypes.TelemetryFieldKey {
return &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeFloat64,
}
}
m := make(map[string][]*telemetrytypes.TelemetryFieldKey)
// gen_ai semconv keys sourced from the single source of truth, mirroring what the
// production metadata store surfaces via enrichWithGenAIKeys.
for name, def := range telemetrytypes.GenAIFieldDefinitions {
keyCopy := def
m[name] = []*telemetrytypes.TelemetryFieldKey{&keyCopy}
}
// Extra keys these tests reference that aren't gen_ai semconv definitions.
m["gen_ai.user.id"] = []*telemetrytypes.TelemetryFieldKey{strKey("gen_ai.user.id")}
m["_signoz.gen_ai.total_cost"] = []*telemetrytypes.TelemetryFieldKey{numKey("_signoz.gen_ai.total_cost")}
m["gen_ai.usage.cached_input_tokens"] = []*telemetrytypes.TelemetryFieldKey{numKey("gen_ai.usage.cached_input_tokens")}
m["has_error"] = []*telemetrytypes.TelemetryFieldKey{{
Name: "has_error",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextSpan,
FieldDataType: telemetrytypes.FieldDataTypeBool,
}}
return m
}
// standard test window (ms), matching the traces builder tests.
const (
testStartMs = uint64(1747947419000)
testEndMs = uint64(1747983448000)
)
func newTestBuilder(t *testing.T) qbtypes.StatementBuilder[qbtypes.TraceAggregation] {
return newTestBuilderWithKeys(t, otelKeysMap())
}
// newTestBuilderWithKeys mirrors the production wiring in signozquerier's provider.
// The gen_ai keys are seeded via keysMap here; in production the metadata store
// surfaces them itself (enrichWithGenAIKeys).
func newTestBuilderWithKeys(t *testing.T, keysMap map[string][]*telemetrytypes.TelemetryFieldKey) qbtypes.StatementBuilder[qbtypes.TraceAggregation] {
t.Helper()
settings := instrumentationtest.New().ToProviderSettings()
fm := telemetrytraces.NewFieldMapper()
cb := telemetrytraces.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = keysMap
fl := flaggertest.New(t)
baseCond := NewGenAIBaseConditionProvider()
// In production the metadata store enriches gen_ai keys (enrichWithGenAIKeys);
// here the mock is seeded directly via keysMap.
metadataStore := telemetrytypes.MetadataStore(mockMetadataStore)
rewriter := querybuilder.NewAggExprRewriter(settings, nil, fm, cb, nil, fl)
traceStmtBuilder := telemetrytraces.NewTraceQueryStatementBuilder(
settings,
metadataStore,
fm,
cb,
rewriter,
nil,
fl,
false,
100000,
)
return NewAITraceStatementBuilder(
settings,
metadataStore,
baseCond,
traceStmtBuilder,
nil, // telemetryStore: only used by the skip-fingerprint count query, which is disabled here
fl,
false,
100000,
)
}
// ---------------------------------------------------------------------------
// Full-query golden tests
//
// Each pins the WHOLE generated statement, with bound args inlined into the `?`
// placeholders, as ONE self-contained literal — so a failure diff shows the entire
// query and the expected SQL can be copied straight into a ClickHouse client. The
// `want` strings are formatted for readability; the comparison is whitespace- and
// backtick-insensitive (see normalizeSQL), so only the SQL tokens themselves matter.
//
// The four trace-list goldens cover the corners of how `matched` is assembled —
// {no span filter, span filter} × {no aggregate filter, aggregate filter} — plus a
// mixed filter + multi-key order, plus the delegated span list. Note `matched` selects
// only the aggregates ORDER BY / HAVING reference; the rest appear only in enrichment.
//
// Run `go test ./pkg/telemetryai/ -run TestBuild_FullSQL -v` to also print each query.
// ---------------------------------------------------------------------------
// renderSQL substitutes bound args into the `?` placeholders so the whole statement
// reads as one literal SQL string.
func renderSQL(t *testing.T, stmt *qbtypes.Statement) string {
t.Helper()
var b strings.Builder
argi := 0
for i := 0; i < len(stmt.Query); i++ {
if stmt.Query[i] == '?' {
require.Less(t, argi, len(stmt.Args), "more ? than args in query")
b.WriteString(formatArg(stmt.Args[argi]))
argi++
continue
}
b.WriteByte(stmt.Query[i])
}
require.Equal(t, len(stmt.Args), argi, "arg count does not match number of placeholders")
return b.String()
}
func formatArg(a any) string {
if s, ok := a.(string); ok {
return "'" + s + "'"
}
return fmt.Sprintf("%v", a)
}
// normalizeSQL makes the comparison insensitive to formatting: it drops identifier
// backticks, collapses whitespace runs to a single space, and removes spaces directly
// inside parentheses. This lets the golden strings be freely indented/wrapped (and
// written as Go raw literals, which cannot contain backticks) — only the SQL tokens
// and their order matter.
func normalizeSQL(s string) string {
s = strings.Join(strings.Fields(strings.ReplaceAll(s, "`", "")), " ")
s = strings.ReplaceAll(s, "( ", "(")
s = strings.ReplaceAll(s, " )", ")")
return s
}
func requireSQLEqual(t *testing.T, want string, stmt *qbtypes.Statement) {
t.Helper()
got := renderSQL(t, stmt)
t.Logf("\n%s", got)
require.Equal(t, normalizeSQL(want), normalizeSQL(got))
}
// No filter: matched selects only the default order key (last_activity_time), WHERE is
// just window + gate mask, no HAVING.
func TestBuild_FullSQL_TraceList_NoFilter(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI, Limit: 20,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
LIMIT 20
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
countIf(mapContains(attributes_string, 'gen_ai.tool.name') = true) AS tool_call_count,
uniqIf(multiIf(mapContains(attributes_string, 'gen_ai.tool.name') = true, attributes_string['gen_ai.tool.name'], NULL), mapContains(attributes_string, 'gen_ai.tool.name') = true) AS distinct_tool_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens,
sum(multiIf(mapContains(attributes_number, '_signoz.gen_ai.total_cost') = true, toFloat64(attributes_number['_signoz.gen_ai.total_cost']), NULL)) AS estimated_cost_usd,
maxIf(signoz_traces.distributed_signoz_index_v3.duration_nano, mapContains(attributes_string, 'gen_ai.request.model') = true) AS max_llm_latency_ns,
countIf(has_error = true) AS error_count,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time,
argMinIf(multiIf(mapContains(attributes_string, 'gen_ai.input.messages') = true, attributes_string['gen_ai.input.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.input.messages') = true) AS input,
argMaxIf(multiIf(mapContains(attributes_string, 'gen_ai.output.messages') = true, attributes_string['gen_ai.output.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.output.messages') = true) AS output
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Promotion: a materialized gen_ai attribute must resolve to its materialized column
// everywhere it appears — gate mask, countIf/scoped existence, and value columns —
// while un-promoted attributes stay in the attributes map, so one query mixes both
// forms. Here gen_ai.request.model and gen_ai.usage.input_tokens are materialized:
// the gate/llm_call_count/max_llm_latency use `..._exists`, input_tokens/total_tokens
// use the materialized value column, and tool/output_tokens/cost/messages stay in the map.
func TestBuild_FullSQL_TraceList_MaterializedColumns(t *testing.T) {
keys := otelKeysMap()
for _, name := range []string{"gen_ai.request.model", "gen_ai.usage.input_tokens"} {
for _, k := range keys[name] {
k.Materialized = true
}
}
b := newTestBuilderWithKeys(t, keys)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI, Limit: 20,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
maxIf(timestamp, (attribute_string_gen_ai$$request$$model_exists = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((attribute_string_gen_ai$$request$$model_exists = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
LIMIT 20
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(attribute_string_gen_ai$$request$$model_exists = true) AS llm_call_count,
countIf(mapContains(attributes_string, 'gen_ai.tool.name') = true) AS tool_call_count,
uniqIf(multiIf(mapContains(attributes_string, 'gen_ai.tool.name') = true, attributes_string['gen_ai.tool.name'], NULL), mapContains(attributes_string, 'gen_ai.tool.name') = true) AS distinct_tool_count,
sum(multiIf(attribute_number_gen_ai$$usage$$input_tokens_exists = true, toFloat64(attribute_number_gen_ai$$usage$$input_tokens), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
coalesce(sum(multiIf(attribute_number_gen_ai$$usage$$input_tokens_exists = true, toFloat64(attribute_number_gen_ai$$usage$$input_tokens), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens,
sum(multiIf(mapContains(attributes_number, '_signoz.gen_ai.total_cost') = true, toFloat64(attributes_number['_signoz.gen_ai.total_cost']), NULL)) AS estimated_cost_usd,
maxIf(signoz_traces.distributed_signoz_index_v3.duration_nano, attribute_string_gen_ai$$request$$model_exists = true) AS max_llm_latency_ns,
countIf(has_error = true) AS error_count,
maxIf(timestamp, (attribute_string_gen_ai$$request$$model_exists = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time,
argMinIf(multiIf(mapContains(attributes_string, 'gen_ai.input.messages') = true, attributes_string['gen_ai.input.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.input.messages') = true) AS input,
argMaxIf(multiIf(mapContains(attributes_string, 'gen_ai.output.messages') = true, attributes_string['gen_ai.output.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.output.messages') = true) AS output
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Span-level AND trace-level filter, order by the aggregate, pagination. matched selects
// only output_tokens (the sole aggregate referenced by both ORDER BY and HAVING) — not
// input_tokens/llm_call_count/last_activity_time. The span predicate widens the WHERE
// prune and becomes a countIf(...) > 0 existence check alongside the gate countIf.
func TestBuild_FullSQL_TraceList_SpanAndTraceFilter(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND output_tokens > 1000"},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "output_tokens"}}, Direction: qbtypes.OrderDirectionDesc}},
Limit: 10, Offset: 30,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
OR (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model') = true))
GROUP BY trace_id
HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
AND countIf((attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
AND output_tokens > 1000
ORDER BY output_tokens DESC, trace_id DESC
LIMIT 10 OFFSET 30
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
countIf(mapContains(attributes_string, 'gen_ai.tool.name') = true) AS tool_call_count,
uniqIf(multiIf(mapContains(attributes_string, 'gen_ai.tool.name') = true, attributes_string['gen_ai.tool.name'], NULL), mapContains(attributes_string, 'gen_ai.tool.name') = true) AS distinct_tool_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens,
sum(multiIf(mapContains(attributes_number, '_signoz.gen_ai.total_cost') = true, toFloat64(attributes_number['_signoz.gen_ai.total_cost']), NULL)) AS estimated_cost_usd,
maxIf(signoz_traces.distributed_signoz_index_v3.duration_nano, mapContains(attributes_string, 'gen_ai.request.model') = true) AS max_llm_latency_ns,
countIf(has_error = true) AS error_count,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time,
argMinIf(multiIf(mapContains(attributes_string, 'gen_ai.input.messages') = true, attributes_string['gen_ai.input.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.input.messages') = true) AS input,
argMaxIf(multiIf(mapContains(attributes_string, 'gen_ai.output.messages') = true, attributes_string['gen_ai.output.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.output.messages') = true) AS output
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY output_tokens DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Aggregate-only filter (no span filter). WHERE prune is NOT widened, there is no
// gate/span countIf, just the aggregate HAVING. `trace.output_tokens` rewrites to the
// output_tokens alias. matched selects output_tokens (HAVING) + last_activity_time (default order).
func TestBuild_FullSQL_TraceList_AggregateFilterOnly(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
Limit: 20,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
GROUP BY trace_id
HAVING output_tokens > 1000
ORDER BY last_activity_time DESC, trace_id DESC
LIMIT 20
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
countIf(mapContains(attributes_string, 'gen_ai.tool.name') = true) AS tool_call_count,
uniqIf(multiIf(mapContains(attributes_string, 'gen_ai.tool.name') = true, attributes_string['gen_ai.tool.name'], NULL), mapContains(attributes_string, 'gen_ai.tool.name') = true) AS distinct_tool_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens,
sum(multiIf(mapContains(attributes_number, '_signoz.gen_ai.total_cost') = true, toFloat64(attributes_number['_signoz.gen_ai.total_cost']), NULL)) AS estimated_cost_usd,
maxIf(signoz_traces.distributed_signoz_index_v3.duration_nano, mapContains(attributes_string, 'gen_ai.request.model') = true) AS max_llm_latency_ns,
countIf(has_error = true) AS error_count,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time,
argMinIf(multiIf(mapContains(attributes_string, 'gen_ai.input.messages') = true, attributes_string['gen_ai.input.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.input.messages') = true) AS input,
argMaxIf(multiIf(mapContains(attributes_string, 'gen_ai.output.messages') = true, attributes_string['gen_ai.output.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.output.messages') = true) AS output
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Span-only filter (no aggregate filter). WHERE is widened; HAVING has the gate + span
// countIf pair but no trailing aggregate. `has_error = true` resolves to a
// materialized-column predicate (not a map access). matched selects only the default order key.
func TestBuild_FullSQL_TraceList_SpanFilterOnly(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "has_error = true"},
Limit: 20,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
OR has_error = true)
GROUP BY trace_id
HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
AND countIf(has_error = true) > 0
ORDER BY last_activity_time DESC, trace_id DESC
LIMIT 20
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
countIf(mapContains(attributes_string, 'gen_ai.tool.name') = true) AS tool_call_count,
uniqIf(multiIf(mapContains(attributes_string, 'gen_ai.tool.name') = true, attributes_string['gen_ai.tool.name'], NULL), mapContains(attributes_string, 'gen_ai.tool.name') = true) AS distinct_tool_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens,
sum(multiIf(mapContains(attributes_number, '_signoz.gen_ai.total_cost') = true, toFloat64(attributes_number['_signoz.gen_ai.total_cost']), NULL)) AS estimated_cost_usd,
maxIf(signoz_traces.distributed_signoz_index_v3.duration_nano, mapContains(attributes_string, 'gen_ai.request.model') = true) AS max_llm_latency_ns,
countIf(has_error = true) AS error_count,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time,
argMinIf(multiIf(mapContains(attributes_string, 'gen_ai.input.messages') = true, attributes_string['gen_ai.input.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.input.messages') = true) AS input,
argMaxIf(multiIf(mapContains(attributes_string, 'gen_ai.output.messages') = true, attributes_string['gen_ai.output.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.output.messages') = true) AS output
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Resource filter: a resource attribute in the filter is pulled into a __resource_filter
// CTE (fingerprints matching the resource condition), and the `matched` scan is narrowed
// by `resource_fingerprint GLOBAL IN (…)`. The resource key is dropped from the span
// predicate (skipResourceFilter), so here there is no span-level existence check — the
// prune stays the gate mask and the whole match is scoped to the resource fingerprints.
func TestBuild_FullSQL_TraceList_ResourceFilter(t *testing.T) {
keys := otelKeysMap()
keys["service.name"] = []*telemetrytypes.TelemetryFieldKey{{
Name: "service.name",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
}}
b := newTestBuilderWithKeys(t, keys)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "resource.service.name = 'checkout'"},
Limit: 20,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH __resource_filter AS (
SELECT fingerprint
FROM signoz_traces.distributed_traces_v3_resource
WHERE (simpleJSONExtractString(labels, 'service.name') = 'checkout' AND labels LIKE '%service.name%' AND labels LIKE '%service.name":"checkout%')
AND seen_at_ts_bucket_start >= 1747945619
AND seen_at_ts_bucket_start <= 1747983448
GROUP BY fingerprint
),
matched AS (
SELECT trace_id,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
AND resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
LIMIT 20
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
countIf(mapContains(attributes_string, 'gen_ai.tool.name') = true) AS tool_call_count,
uniqIf(multiIf(mapContains(attributes_string, 'gen_ai.tool.name') = true, attributes_string['gen_ai.tool.name'], NULL), mapContains(attributes_string, 'gen_ai.tool.name') = true) AS distinct_tool_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens,
sum(multiIf(mapContains(attributes_number, '_signoz.gen_ai.total_cost') = true, toFloat64(attributes_number['_signoz.gen_ai.total_cost']), NULL)) AS estimated_cost_usd,
maxIf(signoz_traces.distributed_signoz_index_v3.duration_nano, mapContains(attributes_string, 'gen_ai.request.model') = true) AS max_llm_latency_ns,
countIf(has_error = true) AS error_count,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time,
argMinIf(multiIf(mapContains(attributes_string, 'gen_ai.input.messages') = true, attributes_string['gen_ai.input.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.input.messages') = true) AS input,
argMaxIf(multiIf(mapContains(attributes_string, 'gen_ai.output.messages') = true, attributes_string['gen_ai.output.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.output.messages') = true) AS output
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Mixed filter (two span predicates AND'd into one existence check + an aggregate) with
// a two-key order on different aggregates than the filter. matched selects input_tokens
// + last_activity_time (ORDER BY) and output_tokens (HAVING) — three of four; llm_call_count is not.
func TestBuild_FullSQL_TraceList_MixedFiltersMultiOrder(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o' AND has_error = true AND output_tokens > 500"},
Order: []qbtypes.OrderBy{
{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "input_tokens"}}, Direction: qbtypes.OrderDirectionDesc},
{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "last_activity_time"}}, Direction: qbtypes.OrderDirectionAsc},
},
Limit: 15,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
OR ((attributes_string['gen_ai.request.model'] = 'gpt-4o' AND mapContains(attributes_string, 'gen_ai.request.model') = true) AND has_error = true))
GROUP BY trace_id
HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
AND countIf(((attributes_string['gen_ai.request.model'] = 'gpt-4o' AND mapContains(attributes_string, 'gen_ai.request.model') = true) AND has_error = true)) > 0
AND output_tokens > 500
ORDER BY input_tokens DESC, last_activity_time ASC, trace_id DESC
LIMIT 15
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
countIf(mapContains(attributes_string, 'gen_ai.tool.name') = true) AS tool_call_count,
uniqIf(multiIf(mapContains(attributes_string, 'gen_ai.tool.name') = true, attributes_string['gen_ai.tool.name'], NULL), mapContains(attributes_string, 'gen_ai.tool.name') = true) AS distinct_tool_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens,
sum(multiIf(mapContains(attributes_number, '_signoz.gen_ai.total_cost') = true, toFloat64(attributes_number['_signoz.gen_ai.total_cost']), NULL)) AS estimated_cost_usd,
maxIf(signoz_traces.distributed_signoz_index_v3.duration_nano, mapContains(attributes_string, 'gen_ai.request.model') = true) AS max_llm_latency_ns,
countIf(has_error = true) AS error_count,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time,
argMinIf(multiIf(mapContains(attributes_string, 'gen_ai.input.messages') = true, attributes_string['gen_ai.input.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.input.messages') = true) AS input,
argMaxIf(multiIf(mapContains(attributes_string, 'gen_ai.output.messages') = true, attributes_string['gen_ai.output.messages'], NULL), timestamp, mapContains(attributes_string, 'gen_ai.output.messages') = true) AS output
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY input_tokens DESC, last_activity_time ASC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Span list (requestType raw): delegated to the traces builder with the gate ANDed
// into the user filter, so only gen_ai spans matching the filter come back. Standard
// span columns, single SELECT (no CTE pipeline).
func TestBuild_FullSQL_SpanList_Raw(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini'"},
Limit: 10,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
SELECT timestamp AS timestamp, trace_id AS trace_id, span_id AS span_id,
trace_state AS trace_state, parent_span_id AS parent_span_id, flags AS flags,
name AS name, kind AS kind, kind_string AS kind_string, duration_nano AS duration_nano,
status_code AS status_code, status_message AS status_message,
status_code_string AS status_code_string, events AS events, links AS links,
response_status_code AS response_status_code, external_http_url AS external_http_url,
http_url AS http_url, external_http_method AS external_http_method,
http_method AS http_method, http_host AS http_host, db_name AS db_name,
db_operation AS db_operation, has_error AS has_error, is_remote AS is_remote,
attributes_string, attributes_number, attributes_bool, resources_string
FROM signoz_traces.distributed_signoz_index_v3
WHERE (((mapContains(attributes_string, 'gen_ai.request.model') = true
OR mapContains(attributes_string, 'gen_ai.tool.name') = true
OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
AND ((attributes_string['gen_ai.request.model'] = 'gpt-4o-mini'
AND mapContains(attributes_string, 'gen_ai.request.model') = true)))
AND timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
LIMIT 10
`, stmt)
}
// ---------------------------------------------------------------------------
// Behavior / branch tests not covered by the goldens above
// ---------------------------------------------------------------------------
// resourceKeysMap returns the gen_ai keys plus a resource-context service.name key so
// resource.* filters route into the fingerprint CTE.
func resourceKeysMap() map[string][]*telemetrytypes.TelemetryFieldKey {
keys := otelKeysMap()
keys["service.name"] = []*telemetrytypes.TelemetryFieldKey{{
Name: "service.name",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
}}
return keys
}
// A filter mixing a resource attribute with a span-level and an aggregate condition:
// the resource key routes into __resource_filter (fingerprint prune), the span key stays
// as a countIf existence check, and the aggregate becomes a HAVING — all AND-combined.
func TestBuild_TraceList_ResourcePlusSpanPlusAggregateFilter(t *testing.T) {
b := newTestBuilderWithKeys(t, resourceKeysMap())
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "resource.service.name = 'checkout' AND has_error = true AND output_tokens > 1000"},
Limit: 10,
}, nil)
require.NoError(t, err)
got := renderSQL(t, stmt)
// resource condition -> fingerprint CTE + prune, not applied on the span index.
require.Contains(t, got, "__resource_filter AS (")
require.Contains(t, got, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
require.NotContains(t, got, "resources_string['service.name']")
// span condition -> existence check in matched HAVING.
require.Contains(t, got, "countIf(has_error = true) > 0")
// aggregate condition -> HAVING on the matched aggregate alias.
require.Contains(t, got, "output_tokens")
}
// The resolver-unset (nil) fallback is covered in pkg/telemetryscopedtraces, which
// can construct that builder state directly.
// Trace-level and span-level predicates may not be OR-combined.
func TestBuild_TraceList_TraceOrSpanMixRejected(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000 OR gen_ai.request.model = 'x'"},
Limit: 10,
}
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "cannot be combined")
}
// An output-only aggregate (span_count / duration_nano) can be displayed but not used
// in the aggregate filter or ORDER BY — it is not computable in the matched pass.
func TestBuild_TraceList_OutputOnlyAggregateRejected(t *testing.T) {
b := newTestBuilder(t)
// filter by span_count -> rejected
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "span_count > 3"},
}, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "span_count")
// order by duration_nano -> rejected
_, err = b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "duration_nano"}}, Direction: qbtypes.OrderDirectionDesc}},
}, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "unsupported order key")
}
// A HAVING referencing a non-aggregate column is rejected.
func TestBuild_TraceList_Having_UnknownColumn(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
Having: &qbtypes.Having{Expression: "service.name > 1"}, // not an aggregate column
Limit: 10,
}
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
require.Error(t, err)
}
// Ordering by an unknown key is rejected.
func TestBuild_TraceList_UnsupportedOrderKey(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
Order: []qbtypes.OrderBy{
{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "http.request.method"}}, Direction: qbtypes.OrderDirectionDesc},
},
}
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "unsupported order key")
}
// With no limit set, the builder applies the default of 100.
func TestBuild_TraceList_DefaultLimit(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
}
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
require.NoError(t, err)
require.Contains(t, stmt.Query, "LIMIT ?")
require.Contains(t, stmt.Args, 100)
}
// Only trace list and span list (raw) are supported; distribution is not.
func TestBuild_UnsupportedRequestType(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{
{Expression: "count()"},
},
}
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeDistribution, query, nil)
require.ErrorIs(t, err, scopedtraces.ErrUnsupportedRequestType)
}
// A gate key ingested under several data types (e.g. string + number from a
// misbehaving SDK) contributes ALL variants to the mask, OR-combined — not just
// the first — matching the standard visitor's EXISTS handling.
func TestBuild_TraceList_MultiVariantGateKey(t *testing.T) {
keys := otelKeysMap()
keys[telemetrytypes.GenAIToolName] = append(keys[telemetrytypes.GenAIToolName], &telemetrytypes.TelemetryFieldKey{
Name: telemetrytypes.GenAIToolName,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeFloat64,
})
b := newTestBuilderWithKeys(t, keys)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI, Limit: 10,
}, nil)
require.NoError(t, err)
got := renderSQL(t, stmt)
require.Contains(t, got, "mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_number, 'gen_ai.tool.name') = true")
}
// `tracefield.` is the explicit trace field context, so in a filter it marks a
// trace-level aggregate exactly like the user-facing `trace.` prefix — same statement,
// and the same targeted rejection for a non-filterable aggregate. (Filter and Having
// accept the same forms; the splitter used to misroute tracefield. as span-level.)
func TestBuild_TraceList_TracefieldPrefixMatchesTracePrefix(t *testing.T) {
b := newTestBuilder(t)
build := func(expr string) (*qbtypes.Statement, error) {
return b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: expr},
Limit: 20,
}, nil)
}
viaTrace, err := build("trace.output_tokens > 1000")
require.NoError(t, err)
viaTracefield, err := build("tracefield.output_tokens > 1000")
require.NoError(t, err)
require.Equal(t, viaTrace.Query, viaTracefield.Query)
require.Equal(t, viaTrace.Args, viaTracefield.Args)
// output-only aggregate under tracefield. gets the aggregate rejection, not an
// unknown-span-field failure.
_, err = build("tracefield.span_count > 3")
require.Error(t, err)
require.Contains(t, err.Error(), "cannot be used")
}
// Query variables in a trace-level condition resolve through the standard filter
// pipeline, exactly like span-level filters: bound args, list/IN handling, dynamic
// __all__ dropping the condition.
func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
b := newTestBuilder(t)
build := func(expr string, vars map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
return b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: expr},
Limit: 20,
}, vars)
}
// scalar variable -> replaced to a literal (canonical pkg/variables semantics),
// then parsed and bound as an arg by the filter pipeline
stmt, err := build("trace.output_tokens > $threshold",
map[string]qbtypes.VariableItem{"threshold": {Value: 700}})
require.NoError(t, err)
require.Contains(t, stmt.Query, "HAVING output_tokens > ?")
require.Contains(t, stmt.Args, float64(700))
// list variable with IN
stmt, err = build("trace.llm_call_count IN $counts",
map[string]qbtypes.VariableItem{"counts": {Value: []any{1, 2}}})
require.NoError(t, err)
require.Contains(t, stmt.Query, "HAVING llm_call_count IN (?, ?)")
// dynamic __all__ -> condition dropped, no HAVING at all
stmt, err = build("trace.output_tokens > $threshold",
map[string]qbtypes.VariableItem{"threshold": {Type: qbtypes.DynamicVariableType, Value: "__all__"}})
require.NoError(t, err)
require.NotContains(t, stmt.Query, "HAVING")
// unresolved variable -> rejected, not compared as a literal
_, err = build("trace.output_tokens > $missing", map[string]qbtypes.VariableItem{"other": {Value: 1}})
require.Error(t, err)
}

View File

@@ -0,0 +1,398 @@
package telemetryai
import (
"context"
"strings"
"testing"
"time"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/stretchr/testify/require"
)
// Scalar / time-series (trace-level aggregation) Build tests. The
// rewriteTraceAggregation unit tests live in pkg/telemetryscopedtraces; these
// exercise the full builder through the gen_ai provider pair.
// Mixing domains across separate aggregations of one query is rejected.
func TestBuild_Aggregation_MixedDomainsRejected(t *testing.T) {
b := newTestBuilder(t)
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{
{Expression: "avg(trace.output_tokens)"},
{Expression: "sum(gen_ai.usage.output_tokens)"},
},
}, nil)
require.ErrorContains(t, err, "cannot be mixed")
}
// A trace-level filter over an output-only aggregate is rejected on the
// aggregation paths too (it is not computable in the mask-pruned scan).
func TestBuild_Aggregation_OutputOnlyFilterRejected(t *testing.T) {
b := newTestBuilder(t)
for _, rt := range []qbtypes.RequestType{qbtypes.RequestTypeScalar, qbtypes.RequestTypeRaw} {
_, err := b.Build(context.Background(), testStartMs, testEndMs, rt,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
Filter: &qbtypes.Filter{Expression: "trace.span_count > 3"},
}, nil)
require.ErrorContains(t, err, `aggregate "span_count" cannot be used`)
}
}
// Trace-level per-trace columns are rejected as group-by / order keys with a
// targeted error (not the field mapper's generic "field not found").
func TestBuild_Aggregation_GroupByOrderValidation(t *testing.T) {
b := newTestBuilder(t)
ctx := context.Background()
_, err := b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.llm_call_count"}}},
}, nil)
require.ErrorContains(t, err, `grouping by trace-level aggregate "trace.llm_call_count" is not supported`)
_, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.output_tokens"}}}},
}, nil)
require.ErrorContains(t, err, `ordering the span list by trace-level aggregate "trace.output_tokens" is not supported`)
_, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.total_tokens"}}}},
}, nil)
require.ErrorContains(t, err, `ordering by trace-level aggregate "trace.total_tokens" is not supported`)
// ordering by the aggregation itself (expression or alias) stays valid
_, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)", Alias: "avg_out"}},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "avg_out"}}, Direction: qbtypes.OrderDirectionAsc}},
}, nil)
require.NoError(t, err)
}
// Query variables resolve inside trace-level filter conditions on every request type,
// as bound args via the standard filter pipeline; unknown $vars fail with a variable
// error, not an "unknown aggregate" one.
func TestBuild_Aggregation_VariablesInTraceFilter(t *testing.T) {
b := newTestBuilder(t)
ctx := context.Background()
vars := map[string]qbtypes.VariableItem{
"threshold": {Type: qbtypes.TextBoxVariableType, Value: float64(1000)},
}
for _, rt := range []qbtypes.RequestType{qbtypes.RequestTypeScalar, qbtypes.RequestTypeRaw, qbtypes.RequestTypeTrace} {
q := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > $threshold"},
}
if rt == qbtypes.RequestTypeScalar {
q.Aggregations = []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}}
}
stmt, err := b.Build(ctx, testStartMs, testEndMs, rt, q, vars)
require.NoError(t, err, rt.StringValue())
require.Contains(t, stmt.Query, "HAVING output_tokens > ?", rt.StringValue())
require.Contains(t, stmt.Args, float64(1000), rt.StringValue())
_, err = b.Build(ctx, testStartMs, testEndMs, rt, q, nil)
require.ErrorContains(t, err, `unknown variable "$threshold"`, rt.StringValue())
}
// a dynamic variable resolved to __all__ skips the trace-level condition, exactly
// like span filters — no qualification CTE is built
stmt, err := b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Filter: &qbtypes.Filter{Expression: "trace.output_tokens IN $models"},
}, map[string]qbtypes.VariableItem{
"models": {Type: qbtypes.DynamicVariableType, Value: "__all__"},
})
require.NoError(t, err)
require.NotContains(t, stmt.Query, "__qualified")
// list variables render as IN with bound args
stmt, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "count(trace.trace_id)"}},
Filter: &qbtypes.Filter{Expression: "trace.llm_call_count IN $counts"},
}, map[string]qbtypes.VariableItem{
"counts": {Type: qbtypes.QueryVariableType, Value: []any{float64(1), float64(2)}},
})
require.NoError(t, err)
require.Contains(t, stmt.Query, "HAVING llm_call_count IN (?, ?)")
}
// A resource-attribute condition prunes the qualification scan the same way it prunes
// the trace list's matched pass: __qualified references the __resource_filter CTE, the
// delegated __trace_scope inlines the fingerprint subquery.
func TestBuild_Aggregation_QualificationResourcePruned(t *testing.T) {
keys := otelKeysMap()
keys["service.name"] = []*telemetrytypes.TelemetryFieldKey{{
Name: "service.name",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
}}
b := newTestBuilderWithKeys(t, keys)
ctx := context.Background()
filter := &qbtypes.Filter{Expression: "service.name = 'api' AND trace.output_tokens > 1000"}
stmt, err := b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Filter: filter,
}, nil)
require.NoError(t, err)
qualified := stmt.Query[strings.Index(stmt.Query, "__qualified"):strings.Index(stmt.Query, "__ai_traces")]
require.Contains(t, qualified, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
stmt, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: filter,
Limit: 10,
}, nil)
require.NoError(t, err)
scope := stmt.Query[strings.Index(stmt.Query, "__trace_scope"):strings.Index(stmt.Query, "SELECT timestamp")]
require.Contains(t, scope, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM (SELECT")
}
// ---------------------------------------------------------------------------
// Full-query goldens — native trace-domain pipeline
// ---------------------------------------------------------------------------
// Scalar over per-trace values, no filter: one window-clipped per-trace scan, outer
// avg across traces.
func TestBuild_FullSQL_Scalar_TraceAgg(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH __ai_traces AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
GROUP BY trace_id
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
)
SELECT avg(output_tokens) AS __result_0
FROM __ai_traces
ORDER BY __result_0 DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Scalar with a span-level + trace-level filter and a groupBy: __qualified holds the
// whole-window qualification, the per-trace scan is per (trace, group) with the span
// filter ANDed, the outer aggregation is per group.
func TestBuild_FullSQL_Scalar_TraceAgg_FilterAndGroupBy(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(trace.total_tokens)"}},
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND trace.output_tokens > 1000"},
GroupBy: []qbtypes.GroupByKey{
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}},
},
Limit: 5,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH __qualified AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
GROUP BY trace_id
HAVING output_tokens > 1000
),
__ai_traces AS (
SELECT trace_id,
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model') = true, attributes_string['gen_ai.request.model'], NULL)) AS gen_ai.request.model,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model') = true)
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
GROUP BY trace_id, gen_ai.request.model
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
)
SELECT gen_ai.request.model, sum(total_tokens) AS __result_0
FROM __ai_traces
GROUP BY gen_ai.request.model
ORDER BY __result_0 DESC
LIMIT 5
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Time series over per-trace values: the per-trace scan buckets by span time
// (per-bucket clipping), the outer aggregation is per bucket.
func TestBuild_FullSQL_TimeSeries_TraceAgg(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH __ai_traces AS (
SELECT trace_id,
toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
GROUP BY trace_id, ts
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
)
SELECT ts, avg(output_tokens) AS __result_0
FROM __ai_traces
GROUP BY ts
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Grouped, limited time series: groups are ranked on whole-window per-trace values
// (__ai_traces_total → __limit_cte) and the main per-bucket query is constrained to
// the top-N groups.
func TestBuild_FullSQL_TimeSeries_TraceAgg_TopN(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
GroupBy: []qbtypes.GroupByKey{
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}},
},
Limit: 3,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH __ai_traces_total AS (
SELECT trace_id,
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model') = true, attributes_string['gen_ai.request.model'], NULL)) AS gen_ai.request.model,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
GROUP BY trace_id, gen_ai.request.model
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
),
__limit_cte AS (
SELECT gen_ai.request.model, avg(output_tokens) AS __result_0
FROM __ai_traces_total
GROUP BY gen_ai.request.model
ORDER BY __result_0 DESC
LIMIT 3
),
__ai_traces AS (
SELECT trace_id,
toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model') = true, attributes_string['gen_ai.request.model'], NULL)) AS gen_ai.request.model,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
GROUP BY trace_id, ts, gen_ai.request.model
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
)
SELECT ts, gen_ai.request.model, avg(output_tokens) AS __result_0
FROM __ai_traces
WHERE (gen_ai.request.model) IN (SELECT gen_ai.request.model FROM __limit_cte)
GROUP BY ts, gen_ai.request.model
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// ---------------------------------------------------------------------------
// Full-query goldens — delegated span-domain with a trace-level filter
// ---------------------------------------------------------------------------
// Span-level scalar with a trace-level filter: delegated to the trace builder with
// the gate ANDed, constrained by the __trace_scope qualification.
func TestBuild_FullSQL_Scalar_SpanAgg_TraceScoped(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(gen_ai.usage.output_tokens)"}},
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH __trace_scope AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
GROUP BY trace_id
HAVING output_tokens > 1000
)
SELECT sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS __result_0
FROM signoz_traces.distributed_signoz_index_v3
WHERE trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)
AND (mapContains(attributes_string, 'gen_ai.request.model') = true
OR mapContains(attributes_string, 'gen_ai.tool.name') = true
OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
AND timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
ORDER BY __result_0 DESC
`, stmt)
}

View File

@@ -0,0 +1,142 @@
package telemetryai
import (
"context"
"testing"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/stretchr/testify/require"
)
// Span list with a mixed filter: gen_ai spans matching the span-level part, in
// traces whose window-clipped aggregates satisfy the trace-level part (the
// __trace_scope qualification on the delegated path).
func TestBuild_FullSQL_SpanList_TraceScoped(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND trace.output_tokens > 1000"},
Limit: 10,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH __trace_scope AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
GROUP BY trace_id
HAVING output_tokens > 1000
)
SELECT timestamp AS timestamp, trace_id AS trace_id, span_id AS span_id,
trace_state AS trace_state, parent_span_id AS parent_span_id, flags AS flags,
name AS name, kind AS kind, kind_string AS kind_string, duration_nano AS duration_nano,
status_code AS status_code, status_message AS status_message,
status_code_string AS status_code_string, events AS events, links AS links,
response_status_code AS response_status_code, external_http_url AS external_http_url,
http_url AS http_url, external_http_method AS external_http_method,
http_method AS http_method, http_host AS http_host, db_name AS db_name,
db_operation AS db_operation, has_error AS has_error, is_remote AS is_remote,
attributes_string, attributes_number, attributes_bool, resources_string
FROM signoz_traces.distributed_signoz_index_v3
WHERE trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)
AND (((mapContains(attributes_string, 'gen_ai.request.model') = true
OR mapContains(attributes_string, 'gen_ai.tool.name') = true
OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
AND ((attributes_string['gen_ai.request.model'] = 'gpt-4o-mini'
AND mapContains(attributes_string, 'gen_ai.request.model') = true)))
AND timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
LIMIT 10
`, stmt)
}
// Without a trace-level condition nothing changes: the span list stays a single
// gated span scan (no __trace_scope CTE).
func TestBuild_SpanList_NoTraceFilter_NoScope(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini'"},
Limit: 10,
}, nil)
require.NoError(t, err)
require.NotContains(t, stmt.Query, "__trace_scope")
}
// The span-list trace-level filter shares the trace list's rules: output-only
// aggregates are rejected, OR-mixing the two classes is rejected, and explicitly
// trace-level order keys get a targeted error — while bare span columns that happen
// to share a name with an aggregate alias (duration_nano) stay orderable.
func TestBuild_SpanList_TraceFilter_Validation(t *testing.T) {
b := newTestBuilder(t)
build := func(q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) error {
q.Signal = telemetrytypes.SignalTraces
q.Source = telemetrytypes.SourceAI
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeRaw, q, nil)
return err
}
err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "trace.span_count > 3"},
})
require.ErrorContains(t, err, `aggregate "span_count" cannot be used`)
err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000 OR kind_string = 'Client'"},
})
require.ErrorContains(t, err, "cannot be combined")
err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.output_tokens"}}}},
})
require.ErrorContains(t, err, `ordering the span list by trace-level aggregate "trace.output_tokens" is not supported`)
err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "duration_nano"}}, Direction: qbtypes.OrderDirectionDesc}},
Limit: 10,
})
require.NoError(t, err, "bare duration_nano is a span column, not a trace-level key")
}
// Variables in a trace-level condition on the span list get the trace list's
// treatment: resolved and bound as args, __all__ drops the condition (no scope CTE),
// tracefield. spelling behaves like trace..
func TestBuild_SpanList_TraceFilter_Variables(t *testing.T) {
b := newTestBuilder(t)
build := func(expr string, vars map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
return b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: expr},
Limit: 10,
}, vars)
}
stmt, err := build("trace.output_tokens > $threshold",
map[string]qbtypes.VariableItem{"threshold": {Value: 700}})
require.NoError(t, err)
require.Contains(t, stmt.Query, "HAVING output_tokens > ?")
require.Contains(t, stmt.Args, float64(700))
stmt, err = build("trace.output_tokens > $threshold",
map[string]qbtypes.VariableItem{"threshold": {Type: qbtypes.DynamicVariableType, Value: "__all__"}})
require.NoError(t, err)
require.NotContains(t, stmt.Query, "__trace_scope")
viaTrace, err := build("trace.output_tokens > 1000", nil)
require.NoError(t, err)
viaTracefield, err := build("tracefield.output_tokens > 1000", nil)
require.NoError(t, err)
require.Equal(t, viaTrace.Query, viaTracefield.Query)
}

View File

@@ -19,6 +19,7 @@ import (
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/types/authtypes"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
@@ -1190,6 +1191,45 @@ func enrichWithIntrinsicMetricKeys(keys map[string][]*telemetrytypes.TelemetryFi
return keys
}
// genAIEnrichmentEnabled reports whether the org in ctx has AI observability enabled.
// The static gen_ai key definitions are surfaced (autocomplete + query-time resolution
// before any gen_ai data is ingested) only for those orgs, so other tenants don't see
// gen_ai keys in trace autocomplete. Contexts without claims (internal paths) resolve
// to false — an org relying on enrichment has no gen_ai data ingested, so those paths
// had nothing to resolve anyway.
func (t *telemetryMetaStore) genAIEnrichmentEnabled(ctx context.Context) bool {
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
return false
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
return false
}
return t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableAIObservability, featuretypes.NewFlaggerEvaluationContext(orgID))
}
// enrichWithGenAIKeys adds keys that can be queried for GenAI signals, even though they have not been ingested yet.
func enrichWithGenAIKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) map[string][]*telemetrytypes.TelemetryFieldKey {
for _, selector := range selectors {
if selector.Signal != telemetrytypes.SignalTraces && selector.Signal != telemetrytypes.SignalUnspecified {
continue
}
for name, def := range telemetrytypes.GenAIFieldDefinitions {
if len(keys[name]) > 0 {
continue // already resolved from ingested data
}
if !selectorMatchesIntrinsicField(selector, def) {
continue
}
keyCopy := def
keys[name] = []*telemetrytypes.TelemetryFieldKey{&keyCopy}
}
}
return keys
}
func selectorMatchesIntrinsicField(selector *telemetrytypes.FieldKeySelector, definition telemetrytypes.TelemetryFieldKey) bool {
if selector.FieldContext != telemetrytypes.FieldContextUnspecified && selector.FieldContext != definition.FieldContext {
return false
@@ -1275,6 +1315,9 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, fieldKeySelector *tele
applyBackwardCompatibleKeys(mapOfKeys)
mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, selectors)
if t.genAIEnrichmentEnabled(ctx) {
mapOfKeys = enrichWithGenAIKeys(mapOfKeys, selectors)
}
return mapOfKeys, complete, nil
}
@@ -1353,6 +1396,9 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
applyBackwardCompatibleKeys(mapOfKeys)
mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, fieldKeySelectors)
if t.genAIEnrichmentEnabled(ctx) {
mapOfKeys = enrichWithGenAIKeys(mapOfKeys, fieldKeySelectors)
}
return mapOfKeys, complete, nil
}

View File

@@ -0,0 +1,17 @@
package telemetryscopedtraces
import (
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// BaseConditionProvider defines which spans are in scope. It only declares the gate
// (a filter expression + its field keys); the builder resolves the keys through the
// field mapper, so attribute access stays materialization-aware.
type BaseConditionProvider interface {
// FilterExpression is the grammar-level (EXISTS) gate, used on the delegated
// span-list path.
FilterExpression() string
// FieldKeys are the gate's keys, used to build the per-span mask
// (OR of resolved EXISTS conditions).
FieldKeys() []*telemetrytypes.TelemetryFieldKey
}

View File

@@ -0,0 +1,184 @@
package telemetryscopedtraces
import (
"fmt"
"strings"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/huandu/go-sqlbuilder"
)
// TraceColumn is one per-trace output column.
type TraceColumn struct {
Alias string
// Orderable columns can be used in ORDER BY and the aggregate filter. All-span
// aggregates (span_count, duration_nano, …) are display-only and set false.
Orderable bool
// SpanLevel columns surface a real span/resource attribute (service.name,
// input/output messages); a filter on them is applied span-level, so they are
// excluded from AggregateAliases.
SpanLevel bool
Expr Aggregate
}
// Aggregate renders one column's SQL and lists the attribute keys it references so
// the builder can pre-fetch their metadata. Build one with the constructors below;
// the zero value is not usable.
type Aggregate struct {
keys []*telemetrytypes.TelemetryFieldKey
render func(r aggResolver) (expr string, args []any, err error)
}
// aggResolver hands each aggregate the field-mapper primitives it may need — an
// EXISTS predicate, a resolved value expression, and the gate mask. Populated per
// query by resolveColumns.
type aggResolver struct {
exists func(key *telemetrytypes.TelemetryFieldKey) (string, []any, error)
value func(key *telemetrytypes.TelemetryFieldKey, dt telemetrytypes.FieldDataType) (string, []any, error)
maskExpr string
maskArgs []any
}
// AggFunc is a ClickHouse aggregate function name.
type AggFunc string
const (
AggSum AggFunc = "sum"
AggMax AggFunc = "max"
AggMin AggFunc = "min"
)
// PickDirection selects the earliest (argMin) or latest (argMax) span by ordering.
type PickDirection int
const (
PickLatest PickDirection = iota
PickEarliest
)
// Intrinsic emits fixed intrinsic-column SQL verbatim (escaped once).
func Intrinsic(text string) Aggregate {
return Aggregate{render: func(aggResolver) (string, []any, error) {
return sqlbuilder.Escape(text), nil, nil
}}
}
// CountExists renders countIf(<key> EXISTS) — counts spans carrying key.
func CountExists(key *telemetrytypes.TelemetryFieldKey) Aggregate {
return Aggregate{keys: keysOf(key), render: func(r aggResolver) (string, []any, error) {
cond, args, err := r.exists(key)
return fmt.Sprintf("countIf(%s)", cond), args, err
}}
}
// Reduce renders <fn>(<value>) over a resolved numeric attribute value.
func Reduce(fn AggFunc, valueKey *telemetrytypes.TelemetryFieldKey) Aggregate {
return Aggregate{keys: keysOf(valueKey), render: func(r aggResolver) (string, []any, error) {
v, args, err := r.value(valueKey, telemetrytypes.FieldDataTypeFloat64)
return fmt.Sprintf("%s(%s)", fn, v), args, err
}}
}
// ScopedReduce renders <fn>If(<valueExpr>, <gate mask>) over a fixed value expression.
func ScopedReduce(fn AggFunc, valueExpr string) Aggregate {
return Aggregate{render: func(r aggResolver) (string, []any, error) {
return fmt.Sprintf("%sIf(%s, %s)", fn, valueExpr, r.maskExpr), append([]any{}, r.maskArgs...), nil
}}
}
// ScopedToKeyColumn renders <fn>If(<column>, <scopeKey> EXISTS) — a physical
// span-index column aggregated over spans carrying scopeKey (e.g. max LLM latency).
// Providers pass the bare column name; it is table-qualified here so it binds to the
// physical column and not a same-named output alias, which ClickHouse would reject
// as an aggregate inside an aggregate.
func ScopedToKeyColumn(fn AggFunc, column string, scopeKey *telemetrytypes.TelemetryFieldKey) Aggregate {
return Aggregate{keys: keysOf(scopeKey), render: func(r aggResolver) (string, []any, error) {
cond, args, err := r.exists(scopeKey)
return fmt.Sprintf("%sIf(%s.%s, %s)", fn, spanTable(), column, cond), args, err
}}
}
// PickBy renders argMinIf/argMaxIf(<value>, <orderExpr>, <value> EXISTS) — the value
// from the earliest/latest span that carries it.
func PickBy(valueKey *telemetrytypes.TelemetryFieldKey, dt telemetrytypes.FieldDataType, orderExpr string, dir PickDirection) Aggregate {
fn := "argMaxIf"
if dir == PickEarliest {
fn = "argMinIf"
}
return Aggregate{keys: keysOf(valueKey), render: func(r aggResolver) (string, []any, error) {
v, vargs, err := r.value(valueKey, dt)
if err != nil {
return "", nil, err
}
cond, cargs, err := r.exists(valueKey)
return fmt.Sprintf("%s(%s, %s, %s)", fn, v, orderExpr, cond), append(vargs, cargs...), err
}}
}
// UniqCount renders uniqIf(<value>, <value> EXISTS) — distinct count of an attribute.
func UniqCount(valueKey *telemetrytypes.TelemetryFieldKey, dt telemetrytypes.FieldDataType) Aggregate {
return Aggregate{keys: keysOf(valueKey), render: func(r aggResolver) (string, []any, error) {
v, vargs, err := r.value(valueKey, dt)
if err != nil {
return "", nil, err
}
cond, cargs, err := r.exists(valueKey)
return fmt.Sprintf("uniqIf(%s, %s)", v, cond), append(vargs, cargs...), err
}}
}
// PredicateCount renders countIf(<predicate>) over a fixed boolean predicate.
func PredicateCount(predicate string) Aggregate {
return Aggregate{render: func(aggResolver) (string, []any, error) {
return fmt.Sprintf("countIf(%s)", sqlbuilder.Escape(predicate)), nil, nil
}}
}
// SumOfKeys renders coalesce(sum(<v1>), 0) + coalesce(sum(<v2>), 0) + … over several
// numeric attributes. Coalesced because a key absent from every span sums to NULL and
// NULL + n = NULL — a trace with only output tokens would otherwise total NULL.
func SumOfKeys(dt telemetrytypes.FieldDataType, valueKeys ...*telemetrytypes.TelemetryFieldKey) Aggregate {
return Aggregate{keys: valueKeys, render: func(r aggResolver) (string, []any, error) {
parts := make([]string, 0, len(valueKeys))
var args []any
for _, k := range valueKeys {
v, vargs, err := r.value(k, dt)
if err != nil {
return "", nil, err
}
parts = append(parts, fmt.Sprintf("coalesce(sum(%s), 0)", v))
args = append(args, vargs...)
}
return strings.Join(parts, " + "), args, nil
}}
}
func keysOf(k *telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
return []*telemetrytypes.TelemetryFieldKey{k}
}
// ColumnProvider supplies the columns a trace list computes.
type ColumnProvider interface {
Columns() []TraceColumn
// DefaultOrderAlias is sorted by (desc) when the query gives no order.
DefaultOrderAlias() string
// AggregateAliases are the computed per-trace column names, used to classify a
// filter key as trace-level vs span-level. Excludes SpanLevel columns.
AggregateAliases() []string
// ActivityGateAlias names the column that must be > 0 for a per-trace row to feed
// trace-level (trace.) aggregations; empty disables the gate.
ActivityGateAlias() string
}
// CommonTraceColumns are domain-neutral columns any trace list can reuse. All
// aggregate over every span, so none is Orderable.
func CommonTraceColumns() []TraceColumn {
return []TraceColumn{
{Alias: "start_time", Expr: Intrinsic("min(timestamp)")},
{Alias: "end_time", Expr: Intrinsic("max(timestamp)")},
{Alias: "duration_nano", Expr: Intrinsic("(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp)))")},
{Alias: "span_count", Expr: Intrinsic("count()")},
{Alias: "root_span_name", Expr: Intrinsic("anyIf(name, parent_span_id = '')")},
{Alias: "service.name", SpanLevel: true, Expr: Intrinsic("any(resource_string_service$$name)")},
}
}

View File

@@ -0,0 +1,840 @@
package telemetryscopedtraces
import (
"context"
"fmt"
"log/slog"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryresourcefilter"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/huandu/go-sqlbuilder"
)
var (
ErrUnsupportedRequestType = errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported request type for the scoped trace builder")
)
// scopedTraceStatementBuilder builds a trace list scoped to one span category
// (e.g. gen_ai spans). The query shape is fixed; BaseConditionProvider decides which
// spans are in scope and ColumnProvider decides the per-trace columns, so a new
// category only needs a new pair of providers.
type scopedTraceStatementBuilder struct {
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
baseCond BaseConditionProvider
columnProvider ColumnProvider
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
resourceFilterResolver *telemetryresourcefilter.ResourceFingerprintResolver[qbtypes.TraceAggregation]
skipResourceFingerprintEnabled bool
}
var _ qbtypes.StatementBuilder[qbtypes.TraceAggregation] = (*scopedTraceStatementBuilder)(nil)
// NewScopedTraceStatementBuilder wires the generic trace-list builder. The field
// mapper / condition builder are built here, not injected — the list always scans the
// telemetrytraces span index. traceStmtBuilder (the delegate for the span-list path)
// is injected because the provider already has the canonical instance.
func NewScopedTraceStatementBuilder(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
baseCond BaseConditionProvider,
columnProvider ColumnProvider,
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
telemetryStore telemetrystore.TelemetryStore,
fl flagger.Flagger,
skipResourceFingerprintEnable bool,
skipResourceFingerprintThreshold uint64,
) qbtypes.StatementBuilder[qbtypes.TraceAggregation] {
scopedSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryscopedtraces")
fieldMapper := telemetrytraces.NewFieldMapper()
conditionBuilder := telemetrytraces.NewConditionBuilder(fieldMapper)
// Same resource-fingerprint prune as the standard trace builder — the list scans
// the same span index.
resourceFilterResolver := telemetryresourcefilter.NewResolver[qbtypes.TraceAggregation](
settings,
telemetrytraces.DBName,
telemetrytraces.TracesResourceV3TableName,
telemetrytypes.SignalTraces,
telemetrytypes.SourceUnspecified,
metadataStore,
nil,
fl,
telemetryStore,
skipResourceFingerprintThreshold,
)
return &scopedTraceStatementBuilder{
logger: scopedSettings.Logger(),
metadataStore: metadataStore,
fm: fieldMapper,
cb: conditionBuilder,
baseCond: baseCond,
columnProvider: columnProvider,
traceStmtBuilder: traceStmtBuilder,
resourceFilterResolver: resourceFilterResolver,
skipResourceFingerprintEnabled: skipResourceFingerprintEnable,
}
}
func (b *scopedTraceStatementBuilder) Build(
ctx context.Context,
start uint64,
end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
switch requestType {
case qbtypes.RequestTypeTrace:
return b.buildTraceListQuery(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), query, variables)
case qbtypes.RequestTypeRaw:
if err := b.validateGroupByAndOrder(requestType, query); err != nil {
return nil, err
}
return b.buildDelegated(ctx, start, end, requestType, query, variables)
case qbtypes.RequestTypeScalar, qbtypes.RequestTypeTimeSeries:
return b.buildAggregation(ctx, start, end, requestType, query, variables)
default:
return nil, ErrUnsupportedRequestType
}
}
// traceScopedStatementBuilder is the delegate's optional capability of constraining a
// query to a set of trace ids (implemented by the telemetrytraces builder).
type traceScopedStatementBuilder interface {
BuildTraceScoped(ctx context.Context, start, end uint64, requestType qbtypes.RequestType, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], variables map[string]qbtypes.VariableItem, traceScope *qbtypes.Statement) (*qbtypes.Statement, error)
}
// splitUserFilter partitions query.Filter into a span-level expression (re-parsed by
// the delegate / span predicate resolution) and the resolved trace-level part (nil
// when there is none, or when every trace-level condition was skipped).
func (b *scopedTraceStatementBuilder) splitUserFilter(ctx context.Context, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], variables map[string]qbtypes.VariableItem) (string, *traceHaving, error) {
if query.Filter == nil || strings.TrimSpace(query.Filter.Expression) == "" {
return "", nil, nil
}
spanExpr, traceExpr, err := querybuilder.SplitFilterForAggregates(query.Filter.Expression, b.aggregateAliasSet())
if err != nil {
return "", nil, err
}
having, err := b.resolveTraceHaving(ctx, traceExpr, variables)
if err != nil {
return "", nil, err
}
return spanExpr, having, nil
}
// buildDelegated splits the user filter, ANDs the base gate into its span-level part,
// and delegates to the standard trace builder. A trace-level part (trace.output_tokens
// > 1000) becomes a window-clipped qualification the delegate constrains trace_id by.
// Serves the span list (raw) and span-level scalar/time-series.
func (b *scopedTraceStatementBuilder) buildDelegated(
ctx context.Context,
start, end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
spanExpr, having, err := b.splitUserFilter(ctx, query, variables)
if err != nil {
return nil, err
}
gate := b.baseCond.FilterExpression()
expr := gate
if strings.TrimSpace(spanExpr) != "" {
expr = fmt.Sprintf("(%s) AND (%s)", gate, spanExpr)
}
// shallow copy; only Filter is replaced, caller's query untouched
gated := query
gated.Filter = &qbtypes.Filter{Expression: expr}
if having == nil {
return b.traceStmtBuilder.Build(ctx, start, end, requestType, gated, variables)
}
scoped, ok := b.traceStmtBuilder.(traceScopedStatementBuilder)
if !ok {
return nil, errors.NewInternalf(errors.CodeInternal, "trace statement builder does not support trace-scoped queries")
}
scope, err := b.buildQualifiedStatement(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), having, query, variables)
if err != nil {
return nil, err
}
return scoped.BuildTraceScoped(ctx, start, end, requestType, gated, variables, scope)
}
// buildTraceListQuery wires the CTE pipeline (benchmarked, see ai-qb-handoff.md): one
// windowed pass picks the top-N traces, then a bucket-pruned pass enriches only those.
// Helpers appear in this file in the order they run. start/end are nanoseconds.
//
// RESOLVE (keys/columns → SQL via the field mapper)
// fetchKeys metadata for every key we reference
// resolveMask the "span is in scope" predicate (OR of EXISTS)
// resolveColumns per-trace column SQL
// resolveListOrders which columns to ORDER BY
// splitFilter span-level predicate + trace-level HAVING
//
// BUILD
// matched one windowed, mask-pruned GROUP BY trace_id scan fusing gate + span
// │ filter + HAVING + ORDER BY + LIMIT/OFFSET → the top-N trace_ids
// ▼
// ranked [start,end] bounds of those traces, from the small summary table
// ▼
// buckets the ts_bucket_start values they touch, to prune the next scan
// ▼
// enrichment every per-trace column for those traces over their full extent
// (not window-clipped), scanning only their buckets
//
// Only Orderable columns are computable in the mask-pruned matched pass, so only they
// can be ordered or filtered on; all-span columns (span_count, …) are output-only.
func (b *scopedTraceStatementBuilder) buildTraceListQuery(
ctx context.Context,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
endBucket := end / querybuilder.NsToSeconds
limit := query.Limit
if limit <= 0 {
limit = 100
}
// Resolve keys and columns once; all attribute access goes through the field mapper.
keys, err := b.fetchKeys(ctx)
if err != nil {
return nil, err
}
maskExpr, maskArgs, err := b.resolveMask(ctx, start, end, keys)
if err != nil {
return nil, err
}
resolved, err := b.resolveColumns(ctx, start, end, keys, maskExpr, maskArgs)
if err != nil {
return nil, err
}
orders, err := b.resolveListOrders(query.Order, resolved)
if err != nil {
return nil, err
}
// If the filter references resource attributes, add a __resource_filter CTE and
// narrow the matched scan by resource_fingerprint; skipResourceFilter then drops
// those keys from the span predicate so they aren't applied twice.
resourceFrag, resourceArgs, resourcePred, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, query, start, end, variables)
if err != nil {
return nil, err
}
// Split the user filter: span-level predicate + trace-level HAVING expression.
fp, err := b.splitFilter(ctx, query, b.aggregateAliasSet(), start, end, skipResourceFilter, variables)
if err != nil {
return nil, err
}
// matched → ranked → buckets → enrichment
matchedFrag, matchedArgs, err := b.buildMatchedCTE(start, end, startBucket, endBucket, resolved, orders, maskExpr, maskArgs, fp, resourcePred, limit, query.Offset)
if err != nil {
return nil, err
}
rankedFrag, rankedArgs := b.buildRankedCTE(start, end)
bucketsFrag := buildBucketsCTE()
mainSQL, mainArgs := b.buildEnrichmentSelect(resolved, orders)
cteFragments := []string{matchedFrag, rankedFrag, bucketsFrag}
cteArgs := [][]any{matchedArgs, rankedArgs, nil}
// __resource_filter must precede `matched`, which references it.
if resourceFrag != "" {
cteFragments = append([]string{resourceFrag}, cteFragments...)
cteArgs = append([][]any{resourceArgs}, cteArgs...)
}
finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL + " SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000"
finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
return &qbtypes.Statement{
Query: finalSQL,
Args: finalArgs,
Warnings: fp.warnings,
WarningsDocURL: fp.warningsURL,
}, nil
}
// maybeAttachResourceFilter builds the __resource_filter CTE (fingerprints matching
// the filter's resource conditions) and the predicate narrowing the span scan by
// resource_fingerprint, mirroring the standard trace builder. With no resolver or no
// resource conditions it returns empty fragments and the resource keys stay in the
// span predicate (skipResourceFilter=false).
func (b *scopedTraceStatementBuilder) maybeAttachResourceFilter(
ctx context.Context,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
variables map[string]qbtypes.VariableItem,
) (cteFrag string, cteArgs []any, fingerprintPred string, skipResourceFilter bool, err error) {
stmt, skipResourceFilter, err := b.resolveResourceFilterStmt(ctx, query, start, end, variables)
if err != nil || stmt == nil {
return "", nil, "", skipResourceFilter, err
}
return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args,
"resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)", skipResourceFilter, nil
}
// resolveResourceFilterStmt resolves the fingerprint statement for the resource
// attributes in the query's filter; nil when there are none (or the skip decision
// applies). skipResourceFilter follows the same contract as maybeAttachResourceFilter.
func (b *scopedTraceStatementBuilder) resolveResourceFilterStmt(
ctx context.Context,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, bool, error) {
if b.resourceFilterResolver == nil {
return nil, false, nil
}
if b.skipResourceFingerprintEnabled {
decision, err := b.resourceFilterResolver.Resolve(ctx, query, start, end, variables)
if err != nil {
return nil, true, err
}
switch decision {
case qbtypes.ResourceFilterResolveKindNoOp:
return nil, true, nil
case qbtypes.ResourceFilterResolveKindFallback:
return nil, false, nil
}
}
stmt, err := b.resourceFilterResolver.StatementBuilder().Build(
ctx, start, end, qbtypes.RequestTypeRaw, query, variables,
)
if err != nil {
return nil, true, err
}
if stmt == nil {
return nil, true, nil
}
return stmt, true, nil
}
// ---------------------------------------------------------------------------
// RESOLVE — turn keys/columns into field-mapper-aware SQL
// ---------------------------------------------------------------------------
func (b *scopedTraceStatementBuilder) fetchKeys(ctx context.Context) (map[string][]*telemetrytypes.TelemetryFieldKey, error) {
fields := b.resolverFieldKeys()
selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(fields))
for _, k := range fields {
selectors = append(selectors, &telemetrytypes.FieldKeySelector{
Name: k.Name,
Signal: k.Signal,
FieldContext: k.FieldContext,
})
}
keys, _, err := b.metadataStore.GetKeysMulti(ctx, selectors)
return keys, err
}
func (b *scopedTraceStatementBuilder) resolverFieldKeys() []*telemetrytypes.TelemetryFieldKey {
seen := make(map[string]struct{})
var out []*telemetrytypes.TelemetryFieldKey
add := func(k *telemetrytypes.TelemetryFieldKey) {
if k == nil {
return
}
if _, dup := seen[k.Name]; dup {
return
}
seen[k.Name] = struct{}{}
out = append(out, k)
}
for _, k := range b.baseCond.FieldKeys() {
add(k)
}
for _, c := range b.columnProvider.Columns() {
for _, k := range c.Expr.keys {
add(k)
}
}
return out
}
// resolveMask builds the per-span in-scope mask: OR of resolved EXISTS predicates
// over the base condition's field keys.
func (b *scopedTraceStatementBuilder) resolveMask(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey) (string, []any, error) {
fieldKeys := b.baseCond.FieldKeys()
parts := make([]string, 0, len(fieldKeys))
var args []any
for _, key := range fieldKeys {
e, a, err := b.existsExpr(ctx, start, end, keys, key)
if err != nil {
return "", nil, err
}
parts = append(parts, e)
args = append(args, a...)
}
return "(" + strings.Join(parts, " OR ") + ")", args, nil
}
// existsExpr resolves an EXISTS predicate for key via the field mapper (materialized
// column when present, else map access). Escaped once so it can be embedded in an
// outer builder.
func (b *scopedTraceStatementBuilder) existsExpr(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey, key *telemetrytypes.TelemetryFieldKey) (string, []any, error) {
resolvedKey := key
cands := keys[key.Name]
if len(cands) == 0 {
cands = []*telemetrytypes.TelemetryFieldKey{key}
} else {
resolvedKey = cands[0]
}
sb := sqlbuilder.NewSelectBuilder()
conds, _, err := b.cb.ConditionFor(ctx, start, end, resolvedKey, cands, qbtypes.FilterOperatorExists, nil, sb)
if err != nil {
return "", nil, err
}
// One condition per candidate variant (a key can be ingested under several data
// types); OR them all, like the visitor does for EXISTS.
if len(conds) == 1 {
sb.Where(conds[0])
} else {
sb.Where(sb.Or(conds...))
}
expr, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
expr = strings.TrimPrefix(expr, "WHERE ")
return sqlbuilder.Escape(expr), args, nil
}
// resolvedColumn is a column resolved to SQL via the field mapper; expr is escaped
// once, ready to embed in an outer SELECT.
type resolvedColumn struct {
alias string
expr string
args []any
orderable bool
}
// resolveColumns turns the declarative columns into SQL, resolving all
// attribute access through the field mapper.
func (b *scopedTraceStatementBuilder) resolveColumns(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey, maskExpr string, maskArgs []any) ([]resolvedColumn, error) {
r := aggResolver{
exists: func(key *telemetrytypes.TelemetryFieldKey) (string, []any, error) {
return b.existsExpr(ctx, start, end, keys, key)
},
value: func(key *telemetrytypes.TelemetryFieldKey, dt telemetrytypes.FieldDataType) (string, []any, error) {
// Use the metadata variant, which carries Materialized — a provider's static
// definition never does, so a promoted attribute would otherwise fall back
// to map access. Mirrors existsExpr.
if cands := keys[key.Name]; len(cands) > 0 {
key = cands[0]
}
return querybuilder.CollisionHandledFinalExpr(ctx, start, end, key, b.fm, b.cb, keys, dt, nil, false)
},
maskExpr: maskExpr,
maskArgs: maskArgs,
}
cols := b.columnProvider.Columns()
out := make([]resolvedColumn, 0, len(cols))
for _, c := range cols {
expr, args, err := c.Expr.render(r)
if err != nil {
return nil, err
}
out = append(out, resolvedColumn{alias: c.Alias, expr: expr, args: args, orderable: c.Orderable})
}
return out, nil
}
// listOrder is a sort key resolved to a column alias + direction; both the matched
// CTE and the enrichment ORDER BY it.
type listOrder struct {
alias string
direction string
}
// resolveListOrders maps order keys to the resolved orderable columns; non-orderable
// columns are rejected. Defaults to the column provider's default order.
func (b *scopedTraceStatementBuilder) resolveListOrders(order []qbtypes.OrderBy, resolved []resolvedColumn) ([]listOrder, error) {
byAlias := make(map[string]resolvedColumn, len(resolved))
orderable := make([]string, 0, len(resolved))
for _, rc := range resolved {
byAlias[rc.alias] = rc
if rc.orderable {
orderable = append(orderable, rc.alias)
}
}
if len(order) == 0 {
return []listOrder{{alias: b.columnProvider.DefaultOrderAlias(), direction: "DESC"}}, nil
}
orders := make([]listOrder, 0, len(order))
for _, o := range order {
direction := "DESC"
if o.Direction == qbtypes.OrderDirectionAsc {
direction = "ASC"
}
rc, ok := byAlias[o.Key.Name]
if !ok || !rc.orderable {
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"unsupported order key %q for the trace list; orderable keys: %s", o.Key.Name, strings.Join(orderable, ", "))
}
orders = append(orders, listOrder{alias: rc.alias, direction: direction})
}
return orders, nil
}
// filterParts is the user filter split into a span-level predicate (widens the
// matched WHERE prune and becomes a countIf existence check in HAVING) and the
// resolved trace-level HAVING (nil when there is none).
type filterParts struct {
spanPred string
spanArgs []any
hasSpanFilter bool
having *traceHaving
warnings []string
warningsURL string
}
// splitFilter splits query.Filter into a span-level predicate and a trace-level
// HAVING; an explicit query.Having is ANDed onto the latter before resolution.
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet map[string]struct{}, start, end uint64, skipResourceFilter bool, variables map[string]qbtypes.VariableItem) (filterParts, error) {
var fp filterParts
havingExpr := ""
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
spanExpr, traceExpr, err := querybuilder.SplitFilterForAggregates(query.Filter.Expression, classifySet)
if err != nil {
return fp, err
}
havingExpr = traceExpr
if strings.TrimSpace(spanExpr) != "" {
pred, args, warnings, url, err := b.resolveSpanPredicate(ctx, start, end, spanExpr, skipResourceFilter, variables)
if err != nil {
return fp, err
}
// pred is empty when the span-level keys were all resource attributes
// already handled by the __resource_filter CTE.
if strings.TrimSpace(pred) != "" {
fp.spanPred, fp.spanArgs, fp.hasSpanFilter = pred, args, true
}
fp.warnings, fp.warningsURL = warnings, url
}
}
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
if havingExpr != "" {
havingExpr = fmt.Sprintf("(%s) AND (%s)", havingExpr, query.Having.Expression)
} else {
havingExpr = query.Having.Expression
}
}
having, err := b.resolveTraceHaving(ctx, havingExpr, variables)
if err != nil {
return fp, err
}
fp.having = having
return fp, nil
}
// resolveSpanPredicate resolves a span-level filter expression to a bare boolean
// SQL predicate + args via the field mapper.
func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context, start, end uint64, expr string, skipResourceFilter bool, variables map[string]qbtypes.VariableItem) (string, []any, []string, string, error) {
selectors := querybuilder.QueryStringToKeysSelectors(expr)
for i := range selectors {
selectors[i].Signal = telemetrytypes.SignalTraces
}
keys, _, err := b.metadataStore.GetKeysMulti(ctx, selectors)
if err != nil {
return "", nil, nil, "", err
}
prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
Context: ctx,
Logger: b.logger,
FieldMapper: b.fm,
ConditionBuilder: b.cb,
FieldKeys: keys,
SkipResourceFilter: skipResourceFilter,
Variables: variables,
StartNs: start,
EndNs: end,
})
if err != nil {
return "", nil, nil, "", err
}
if prepared.IsEmpty() {
return "", nil, nil, "", nil
}
sb := sqlbuilder.NewSelectBuilder()
sb.Select("1")
sb.AddWhereClause(prepared.WhereClause)
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
pred := sql[strings.Index(sql, "WHERE ")+len("WHERE "):]
return sqlbuilder.Escape(pred), args, prepared.Warnings, prepared.WarningsDocURL, nil
}
// buildMatchedCTE builds `matched`: the single windowed GROUP BY trace_id scan that
// fuses gate + span filter + HAVING + ORDER BY + LIMIT/OFFSET, selecting only the
// aliases the ORDER BY / HAVING reference.
func (b *scopedTraceStatementBuilder) buildMatchedCTE(start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, maskExpr string, maskArgs []any, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
sb := sqlbuilder.NewSelectBuilder()
// SELECT trace_id + only the aggregates ORDER BY / HAVING reference (as aliases).
needed := neededMatchedAliases(orders, fp.having)
selects := []string{"trace_id"}
for _, rc := range resolved {
if _, ok := needed[rc.alias]; !ok {
continue
}
colExpr, err := embedExpr(sb, rc.expr, rc.args)
if err != nil {
return "", nil, err
}
selects = append(selects, colExpr+" AS "+quoteAlias(rc.alias))
}
sb.Select(selects...)
sb.From(spanTable())
// WHERE: window + prune to in-scope spans, widened by the span filter so its
// spans survive for the countIf existence check below.
win := windowWhere(sb, start, end, startBucket, endBucket)
mask, err := embedExpr(sb, maskExpr, maskArgs)
if err != nil {
return "", nil, err
}
prune := "(" + mask
if fp.hasSpanFilter {
spanPred, err := embedExpr(sb, fp.spanPred, fp.spanArgs)
if err != nil {
return "", nil, err
}
prune += " OR " + spanPred
}
prune += ")"
where := append(win, prune)
if resourcePred != "" {
where = append(where, resourcePred)
}
sb.Where(where...)
sb.GroupBy("trace_id")
// HAVING: the gate/span existence checks are only needed when the WHERE was
// widened by a span filter; otherwise the mask alone already enforces the gate.
var having []string
if fp.hasSpanFilter {
havingMask, err := embedExpr(sb, maskExpr, maskArgs)
if err != nil {
return "", nil, err
}
havingPred, err := embedExpr(sb, fp.spanPred, fp.spanArgs)
if err != nil {
return "", nil, err
}
having = append(having, "countIf("+havingMask+") > 0")
having = append(having, "countIf("+havingPred+") > 0")
}
if fp.having != nil {
hv, err := embedExpr(sb, fp.having.pred, fp.having.args)
if err != nil {
return "", nil, err
}
having = append(having, hv)
}
if len(having) > 0 {
sb.Having(strings.Join(having, " AND "))
}
sb.OrderBy(orderClause(orders)...)
sb.Limit(limit)
if offset > 0 {
sb.Offset(offset)
}
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return fmt.Sprintf("matched AS (%s)", sql), args, nil
}
// buildRankedCTE builds `ranked`: [start,end] bounds per matched trace, read from the
// small trace-summary table.
func (b *scopedTraceStatementBuilder) buildRankedCTE(start, end uint64) (string, []any) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select("trace_id", "min(start) AS t_start", "max(end) AS t_end")
sb.From(summaryTable())
sb.Where(
"trace_id GLOBAL IN (SELECT trace_id FROM matched)",
"end >= fromUnixTimestamp64Nano("+sb.Var(start)+")",
"start < fromUnixTimestamp64Nano("+sb.Var(end)+")",
)
sb.GroupBy("trace_id")
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return fmt.Sprintf("ranked AS (%s)", sql), args
}
// buildBucketsCTE builds `buckets`: the ts_bucket_start values the matched traces
// span, so the enrichment scan is primary-key pruned. No args.
func buildBucketsCTE() string {
adj := querybuilder.BucketAdjustment // 30-min bucket width in seconds
return fmt.Sprintf("buckets AS (SELECT DISTINCT b AS ts_bucket FROM ranked "+
"ARRAY JOIN range("+
"toUInt64(intDiv(toUnixTimestamp(t_start), %d) * %d - %d), "+
"toUInt64(intDiv(toUnixTimestamp(t_end), %d) * %d + %d), "+
"%d) AS b)", adj, adj, adj, adj, adj, adj, adj)
}
// buildEnrichmentSelect builds the final SELECT: every per-trace column for the
// matched traces over their full extent, scanning only their buckets.
//
// Accepted discrepancy: matched ranks/paginates on window-clipped values, this pass
// recomputes and ORDER BYs full-trace values, so a trace with activity outside the
// window can sort differently than it ranked. Page membership is unaffected
// (LIMIT/OFFSET runs only in matched); rows still sort by the values the user sees.
// Ordering by matched's values instead would re-run the matched scan (ClickHouse
// re-executes a CTE per reference) without fixing the visible cross-page artifact.
func (b *scopedTraceStatementBuilder) buildEnrichmentSelect(resolved []resolvedColumn, orders []listOrder) (string, []any) {
sb := sqlbuilder.NewSelectBuilder()
selects, selectArgs := selectAllColumns(resolved)
sb.Select(selects...)
sb.From(spanTable())
sb.Where(
"ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)",
"trace_id GLOBAL IN (SELECT trace_id FROM ranked)",
)
sb.GroupBy("trace_id")
sb.OrderBy(orderClause(orders)...)
sql, builtArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return sql, append(append([]any{}, selectArgs...), builtArgs...)
}
// ---------------------------------------------------------------------------
// Small shared SQL-builder utilities
// ---------------------------------------------------------------------------
// spanTable is the fully-qualified span index table.
func spanTable() string {
return fmt.Sprintf("%s.%s", telemetrytraces.DBName, telemetrytraces.SpanIndexV3TableName)
}
// summaryTable is the fully-qualified trace-summary table.
func summaryTable() string {
return fmt.Sprintf("%s.%s", telemetrytraces.DBName, telemetrytraces.TraceSummaryTableName)
}
// aggregateAliasSet is every trace-level column alias, used to classify filter keys
// as trace-level vs span-level.
func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
set := make(map[string]struct{}, len(b.columnProvider.AggregateAliases()))
for _, a := range b.columnProvider.AggregateAliases() {
set[a] = struct{}{}
}
return set
}
// neededMatchedAliases is the minimal alias set the matched pass must select: those
// in ORDER BY plus those the resolved trace-level HAVING touches. Everything else is
// left to the enrichment scan.
func neededMatchedAliases(orders []listOrder, having *traceHaving) map[string]struct{} {
needed := make(map[string]struct{})
for _, o := range orders {
needed[o.alias] = struct{}{}
}
if having != nil {
for a := range having.used {
needed[a] = struct{}{}
}
}
return needed
}
// validateAggregateFilter rejects a trace-level filter referencing an aggregate not
// computable in the matched pass (e.g. span_count, duration_nano) with a targeted
// top-level error — the same check inside the where-clause visitor would surface only
// as a detail of a combined error. Key positions only: `x > $threshold` references x.
func validateAggregateFilter(havingExpr string, orderableSet map[string]struct{}) error {
if strings.TrimSpace(havingExpr) == "" {
return nil
}
for _, key := range querybuilder.ExprKeys(havingExpr) {
name := strings.TrimPrefix(strings.TrimPrefix(key.Name, "trace."), "tracefield.")
if _, ok := orderableSet[name]; !ok {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregate %q cannot be used in the trace-list filter; filterable aggregates: %s", name, strings.Join(sortedAliases(orderableSet), ", "))
}
}
return nil
}
// embedExpr inlines a resolved expr into sb, replacing each `?` placeholder with a
// builder Var so the args are tracked in appearance order. Resolved exprs carry
// values only as bound args, so every `?` is a placeholder; a count mismatch would
// silently shift args into the wrong slots — error out instead.
func embedExpr(sb *sqlbuilder.SelectBuilder, expr string, args []any) (string, error) {
if n := strings.Count(expr, "?"); n != len(args) {
return "", errors.NewInternalf(errors.CodeInternal,
"scoped trace builder: %d placeholders != %d args embedding %q", n, len(args), expr)
}
var out strings.Builder
ai := 0
for i := 0; i < len(expr); i++ {
if expr[i] == '?' {
out.WriteString(sb.Var(args[ai]))
ai++
continue
}
out.WriteByte(expr[i])
}
return out.String(), nil
}
// windowWhere binds the time-window predicates to sb and returns them so the caller
// can add its own predicates in the same Where call.
func windowWhere(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64) []string {
return []string{
sb.GE("timestamp", fmt.Sprintf("%d", start)),
sb.L("timestamp", fmt.Sprintf("%d", end)),
sb.GE("ts_bucket_start", startBucket),
sb.LE("ts_bucket_start", endBucket),
}
}
// orderClause renders the ORDER BY terms plus the trace_id tiebreak.
func orderClause(orders []listOrder) []string {
out := make([]string, 0, len(orders)+1)
for _, o := range orders {
out = append(out, fmt.Sprintf("%s %s", quoteAlias(o.alias), o.direction))
}
return append(out, "trace_id DESC")
}
// selectAllColumns renders `expr AS alias` for every resolved column, args in select
// order.
func selectAllColumns(resolved []resolvedColumn) ([]string, []any) {
selects := []string{"trace_id"}
var args []any
for _, rc := range resolved {
selects = append(selects, rc.expr+" AS "+quoteAlias(rc.alias))
args = append(args, rc.args...)
}
return selects, args
}
// quoteAlias backticks an alias containing characters special to the SQL builder.
func quoteAlias(alias string) string {
if strings.ContainsAny(alias, ".$`") {
return "`" + alias + "`"
}
return alias
}

View File

@@ -0,0 +1,140 @@
package telemetryscopedtraces
import (
"context"
"fmt"
"strings"
"testing"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/huandu/go-sqlbuilder"
"github.com/stretchr/testify/require"
)
// The full-pipeline golden tests live in pkg/telemetryai, which exercises this
// builder through its production provider pair. The tests here cover only what
// needs the package internals: builder states not reachable through the
// constructor.
// stubGate scopes to spans carrying a single attribute key.
type stubGate struct {
key *telemetrytypes.TelemetryFieldKey
}
func (s stubGate) FilterExpression() string { return s.key.Name + " EXISTS" }
func (s stubGate) FieldKeys() []*telemetrytypes.TelemetryFieldKey {
return []*telemetrytypes.TelemetryFieldKey{s.key}
}
// stubColumns is the common columns plus one orderable scoped aggregate, the
// minimum a column provider must supply (a default order key).
type stubColumns struct {
key *telemetrytypes.TelemetryFieldKey
}
func (s stubColumns) Columns() []TraceColumn {
return append(CommonTraceColumns(),
TraceColumn{Alias: "scoped_span_count", Orderable: true, Expr: CountExists(s.key)})
}
func (s stubColumns) DefaultOrderAlias() string { return "scoped_span_count" }
func (s stubColumns) ActivityGateAlias() string { return "" }
func (s stubColumns) AggregateAliases() []string {
aliases := make([]string, 0)
for _, c := range s.Columns() {
if !c.SpanLevel {
aliases = append(aliases, c.Alias)
}
}
return aliases
}
// renderSQL substitutes bound args into the `?` placeholders so assertions can
// match the statement as one literal SQL string.
func renderSQL(t *testing.T, stmt *qbtypes.Statement) string {
t.Helper()
var b strings.Builder
argi := 0
for i := 0; i < len(stmt.Query); i++ {
if stmt.Query[i] == '?' {
require.Less(t, argi, len(stmt.Args), "more ? than args in query")
if s, ok := stmt.Args[argi].(string); ok {
b.WriteString("'" + s + "'")
} else {
fmt.Fprintf(&b, "%v", stmt.Args[argi])
}
argi++
continue
}
b.WriteByte(stmt.Query[i])
}
require.Equal(t, len(stmt.Args), argi, "arg count does not match number of placeholders")
return b.String()
}
// With the resolver unset (nil), the resource filter falls back to being applied inline
// on the span index — no fingerprint CTE — so existing behavior is preserved. The nil
// state is not reachable through the constructor, hence the direct struct literal.
func TestBuild_TraceList_ResourceFilter_NoResolver(t *testing.T) {
gateKey := &telemetrytypes.TelemetryFieldKey{
Name: "scope.marker",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
}
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = map[string][]*telemetrytypes.TelemetryFieldKey{
gateKey.Name: {gateKey},
"service.name": {{
Name: "service.name",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
}
settings := factory.NewScopedProviderSettings(instrumentationtest.New().ToProviderSettings(), "github.com/SigNoz/signoz/pkg/telemetryscopedtraces")
fm := telemetrytraces.NewFieldMapper()
b := &scopedTraceStatementBuilder{
logger: settings.Logger(),
metadataStore: mockMetadataStore,
fm: fm,
cb: telemetrytraces.NewConditionBuilder(fm),
baseCond: stubGate{key: gateKey},
columnProvider: stubColumns{key: gateKey},
resourceFilterResolver: nil,
}
stmt, err := b.Build(context.Background(), 1747947419000, 1747983448000, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "resource.service.name = 'checkout'"},
Limit: 10,
}, nil)
require.NoError(t, err)
got := renderSQL(t, stmt)
require.NotContains(t, got, "__resource_filter")
require.Contains(t, got, "resources_string['service.name']")
}
// embedExpr treats every `?` byte as a placeholder; a count/args mismatch (an expr
// carrying a literal `?`, or a dropped arg) must fail loudly instead of silently
// shifting every subsequent arg into the wrong placeholder.
func TestEmbedExpr_PlaceholderArgMismatch(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
out, err := embedExpr(sb, "x = ? AND y = ?", []any{1, 2})
require.NoError(t, err)
require.Equal(t, 2, strings.Count(out, "$"), "both placeholders bound as builder vars")
_, err = embedExpr(sb, "x = ? AND y LIKE 'a?b'", []any{1})
require.Error(t, err, "literal ? in the expr must not pass as a placeholder")
_, err = embedExpr(sb, "x = ?", []any{1, 2})
require.Error(t, err, "extra args must not be silently dropped")
}

View File

@@ -0,0 +1,785 @@
package telemetryscopedtraces
import (
"context"
"fmt"
"sort"
"strings"
chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/querybuilder"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
)
// This file implements scalar / time-series for source=ai.
//
// Aggregations come in two domains, chosen per expression by the `trace.` prefix:
// - span-level (bare keys): aggregate over individual gen_ai spans. Delegated to the
// standard trace builder with the gate ANDed in; a trace-level filter part becomes
// a __trace_scope qualification (see buildDelegated).
// - trace-level (`trace.` prefix): aggregate over window-clipped per-trace values
// (avg(trace.output_tokens) = average per trace). Runs the native pipeline below.
//
// Native pipeline (buildTraceAggregationQuery):
//
// __qualified traces whose window-clipped aggregates satisfy the trace-level
// │ filter — whole-window values, so a trace qualifies once. Only
// ▼ present when the filter has a trace-level part.
// __ai_traces per-trace values: windowed, mask-pruned GROUP BY trace_id
// │ (+ time bucket for time series → per-bucket clipping, + group-by
// ▼ columns), spans filtered by gate AND span-level filter; rows with
// main no LLM activity are dropped (activity gate). Outer aggregation over
// the per-trace rows → __result_i.
// traceAggregation is one aggregation rewritten to run over the per-trace scan.
type traceAggregation struct {
expr string // rewritten SQL over the per-trace column aliases
used map[string]struct{} // per-trace aliases referenced
isRate bool
}
// buildAggregation routes scalar/time-series requests by aggregation domain.
func (b *scopedTraceStatementBuilder) buildAggregation(
ctx context.Context,
start, end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
traceAggs, err := b.classifyAggregations(query.Aggregations)
if err != nil {
return nil, err
}
if err := b.validateGroupByAndOrder(requestType, query); err != nil {
return nil, err
}
if len(traceAggs) == 0 {
return b.buildDelegated(ctx, start, end, requestType, query, variables)
}
return b.buildTraceAggregationQuery(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), requestType, query, variables, traceAggs)
}
// classifyAggregations splits the aggregations into span-domain (delegated) vs
// trace-domain (over per-trace values). Returns the rewritten trace-domain
// aggregations, nil when all are span-domain; mixing the two domains is rejected.
func (b *scopedTraceStatementBuilder) classifyAggregations(aggs []qbtypes.TraceAggregation) ([]traceAggregation, error) {
traceCols := b.orderableColumnSet()
var out []traceAggregation
spanCount := 0
for _, agg := range aggs {
ta, isTrace, err := rewriteTraceAggregation(agg.Expression, traceCols)
if err != nil {
return nil, err
}
if isTrace {
out = append(out, *ta)
} else {
spanCount++
}
}
if len(out) > 0 && spanCount > 0 {
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"span-level and trace-level (trace.) aggregations cannot be mixed in one query")
}
return out, nil
}
// orderableColumnSet is the gen_ai-scoped per-trace column set (static, from the
// provider) usable in trace-level aggregations and filters.
func (b *scopedTraceStatementBuilder) orderableColumnSet() map[string]struct{} {
set := make(map[string]struct{})
for _, c := range b.columnProvider.Columns() {
if c.Orderable {
set[c.Alias] = struct{}{}
}
}
return set
}
// validateGroupByAndOrder rejects trace-level (trace.) per-trace columns used as a
// group-by key or an order key with a targeted error, instead of the generic "field
// not found" the field mapper would raise. An order key that names an aggregation
// (alias / expression / index) is exempt — that is the way to order by a trace-level
// aggregation's result.
func (b *scopedTraceStatementBuilder) validateGroupByAndOrder(requestType qbtypes.RequestType, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) error {
aliases := b.aggregateAliasSet()
for _, gb := range query.GroupBy {
if isTraceLevelKey(gb.Name, gb.FieldContext, aliases) {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"grouping by trace-level aggregate %q is not supported; group by span attributes instead (e.g. gen_ai.request.model, service.name)", gb.Name)
}
}
for _, o := range query.Order {
if _, isAgg := traceAggOrderIndex(o, query); isAgg {
continue
}
if !isTraceLevelKey(o.Key.Name, o.Key.FieldContext, aliases) {
continue
}
if requestType == qbtypes.RequestTypeRaw {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"ordering the span list by trace-level aggregate %q is not supported; order by span columns instead (e.g. timestamp, duration_nano)", o.Key.Name)
}
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"ordering by trace-level aggregate %q is not supported; order by the aggregation itself (its alias or expression) or a group-by key", o.Key.Name)
}
return nil
}
// isTraceLevelKey reports whether a group-by / order key explicitly names a
// trace-level per-trace aggregate (trace./tracefield. prefix or trace field context).
// Bare names pass through: they may legitimately be span columns that share a name
// with an aggregate alias (duration_nano, timestamp).
func isTraceLevelKey(name string, fieldContext telemetrytypes.FieldContext, aliases map[string]struct{}) bool {
stripped := strings.TrimPrefix(strings.TrimPrefix(name, "tracefield."), "trace.")
if _, ok := aliases[stripped]; !ok {
return false
}
return stripped != name || fieldContext == telemetrytypes.FieldContextTrace
}
// rewriteTraceAggregation parses one aggregation expression. When it references
// trace.-prefixed per-trace columns it returns the expression rewritten to run over
// the per-trace scan (trace.output_tokens → output_tokens, arithmetic between
// trace. columns allowed, function names mapped via AggreFuncMap) with isTrace=true;
// a pure span-level expression returns isTrace=false and is left for the delegate.
func rewriteTraceAggregation(expr string, traceCols map[string]struct{}) (*traceAggregation, bool, error) {
p := chparser.NewParser("SELECT " + expr)
stmts, err := p.ParseStmts()
if err != nil {
return nil, false, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "failed to parse aggregation expression %q", expr)
}
if len(stmts) == 0 {
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid aggregation expression %q", expr)
}
sel, ok := stmts[0].(*chparser.SelectQuery)
if !ok || len(sel.SelectItems) == 0 {
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid aggregation expression %q", expr)
}
v := &traceAggVisitor{traceCols: traceCols, used: make(map[string]struct{})}
if err := sel.SelectItems[0].Accept(v); err != nil {
return nil, false, err
}
if !v.hasTrace {
return nil, false, nil
}
if v.hasSpan {
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregation %q mixes trace-level (trace.) and span-level columns; use one domain per aggregation", expr)
}
return &traceAggregation{expr: sel.SelectItems[0].String(), used: v.used, isRate: v.isRate}, true, nil
}
// traceAggVisitor walks the aggregation AST, classifying column references and
// rewriting trace.-prefixed ones (bare paths, backquoted identifiers, and either
// nested in arithmetic) to the per-trace column aliases in place. It keeps an
// ancestor stack (Enter/Leave) to tell a column identifier from a path segment,
// function name, or alias, and to reject trace. columns inside *If combinators.
type traceAggVisitor struct {
chparser.DefaultASTVisitor
traceCols map[string]struct{}
used map[string]struct{}
stack []chparser.Expr
hasTrace bool
hasSpan bool
isRate bool
}
func (v *traceAggVisitor) Enter(expr chparser.Expr) { v.stack = append(v.stack, expr) }
func (v *traceAggVisitor) Leave(expr chparser.Expr) { v.stack = v.stack[:len(v.stack)-1] }
// parent is the node enclosing the one currently being visited (the visited node
// itself is the stack top).
func (v *traceAggVisitor) parent() chparser.Expr {
if len(v.stack) < 2 {
return nil
}
return v.stack[len(v.stack)-2]
}
// enclosingCombinator returns the name of a surrounding *If-combinator function, if any.
func (v *traceAggVisitor) enclosingCombinator() (string, bool) {
for _, e := range v.stack {
fn, ok := e.(*chparser.FunctionExpr)
if !ok {
continue
}
if agg, known := querybuilder.AggreFuncMap[valuer.NewString(strings.ToLower(fn.Name.Name))]; known && agg.FuncCombinator {
return fn.Name.Name, true
}
}
return "", false
}
// VisitPath classifies a dotted reference (trace.output_tokens); trace-level ones are
// rewritten in place to the bare per-trace alias.
func (v *traceAggVisitor) VisitPath(p *chparser.Path) error {
col, isTrace := traceColumnRef(p.String())
if !isTrace {
v.hasSpan = true
return nil
}
if err := v.acceptTraceColumn(p.String(), col); err != nil {
return err
}
p.Fields = p.Fields[len(p.Fields)-1:]
p.Fields[0].Name = col
return nil
}
// VisitIdent classifies a plain identifier: a backquoted `trace.output_tokens` is a
// trace-level reference (rewritten in place); any other column identifier is
// span-level. Path segments, function names, and aliases are structural, not columns.
func (v *traceAggVisitor) VisitIdent(i *chparser.Ident) error {
switch parent := v.parent().(type) {
case *chparser.Path:
return nil // segments are classified whole by VisitPath
case *chparser.FunctionExpr:
if parent.Name == i {
return nil
}
case *chparser.ColumnExpr:
if parent.Alias == i {
return nil
}
}
col, isTrace := traceColumnRef(i.Name)
if !isTrace {
v.hasSpan = true
return nil
}
if err := v.acceptTraceColumn(i.Name, col); err != nil {
return err
}
i.Name = col
return nil
}
// acceptTraceColumn validates one trace-level column reference and records it.
func (v *traceAggVisitor) acceptTraceColumn(ref, col string) error {
if name, in := v.enclosingCombinator(); in {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"%q over trace-level (trace.) columns is not supported; put the trace-level condition in the filter expression instead", name)
}
// trace_id is always selected by the per-trace scan (count(trace.trace_id)
// counts traces); everything else must be a gen_ai-scoped column.
if col != "trace_id" {
if _, known := v.traceCols[col]; !known {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"unknown trace-level aggregation column %q; usable columns: %s", ref, strings.Join(sortedAliases(v.traceCols), ", "))
}
v.used[col] = struct{}{}
}
v.hasTrace = true
return nil
}
// VisitFunctionExpr validates and maps the function name. Children were already
// visited (post-order), so classification is complete for this subtree.
func (v *traceAggVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
name := strings.ToLower(fn.Name.Name)
aggFunc, ok := querybuilder.AggreFuncMap[valuer.NewString(name)]
if !ok {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "unrecognized function: %s", name)
}
if fn.Params != nil && fn.Params.Items != nil && len(fn.Params.Items.Items) > 0 && aggFunc.FuncCombinator {
// combinator predicates over span columns stay span-level (countIf(has_error=true))
v.hasSpan = true
return nil
}
fn.Name.Name = aggFunc.FuncName
if aggFunc.Rate {
v.isRate = true
}
return nil
}
// traceColumnRef reports whether text is a pure trace.-prefixed column reference
// (trace.output_tokens / tracefield.output_tokens) and returns the bare column name.
func traceColumnRef(text string) (string, bool) {
text = strings.TrimSpace(text)
var rest string
if r, ok := strings.CutPrefix(text, "trace."); ok {
rest = r
} else if r, ok := strings.CutPrefix(text, "tracefield."); ok {
rest = r
} else {
return "", false
}
if rest == "" || strings.ContainsAny(rest, " ()'\"`,+-*/<>=!") {
return "", false
}
return rest, true
}
func sortedAliases(set map[string]struct{}) []string {
out := make([]string, 0, len(set))
for a := range set {
out = append(out, a)
}
sort.Strings(out)
return out
}
// ---------------------------------------------------------------------------
// Qualification + per-trace scan
// ---------------------------------------------------------------------------
// buildQualifiedStatement builds the qualification statement — trace ids whose
// window-clipped per-trace aggregates satisfy the resolved trace-level filter — used
// as the delegate's __trace_scope and the native pipeline's __qualified. When the
// query's filter references resource attributes, the scan is pruned to matching
// resource fingerprints (inlined, since the caller embeds this statement standalone),
// matching the trace list's matched pass. start/end are ns.
func (b *scopedTraceStatementBuilder) buildQualifiedStatement(
ctx context.Context,
start, end uint64,
having *traceHaving,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
keys, err := b.fetchKeys(ctx)
if err != nil {
return nil, err
}
maskExpr, maskArgs, err := b.resolveMask(ctx, start, end, keys)
if err != nil {
return nil, err
}
resolved, err := b.resolveColumns(ctx, start, end, keys, maskExpr, maskArgs)
if err != nil {
return nil, err
}
var resourcePred string
var resourceArgs []any
if stmt, _, err := b.resolveResourceFilterStmt(ctx, query, start, end, variables); err != nil {
return nil, err
} else if stmt != nil {
resourcePred = fmt.Sprintf("resource_fingerprint GLOBAL IN (SELECT fingerprint FROM (%s))", sqlbuilder.Escape(stmt.Query))
resourceArgs = stmt.Args
}
sql, args, err := b.qualifiedScanSQL(start, end, having, resolved, maskExpr, maskArgs, resourcePred, resourceArgs)
if err != nil {
return nil, err
}
return &qbtypes.Statement{Query: sql, Args: args}, nil
}
// qualifiedScanSQL renders the qualification scan given already-resolved columns.
func (b *scopedTraceStatementBuilder) qualifiedScanSQL(start, end uint64, having *traceHaving, resolved []resolvedColumn, maskExpr string, maskArgs []any, resourcePred string, resourceArgs []any) (string, []any, error) {
return b.buildPerTraceScan(start, end, resolved, maskExpr, maskArgs, perTraceScanOpts{
needed: having.used,
havingExpr: having.pred,
havingArgs: having.args,
resourcePred: resourcePred,
resourcePredArgs: resourceArgs,
})
}
// groupColumn is a resolved span-attribute group-by column.
type groupColumn struct {
name string
expr string
args []any
}
// perTraceScanOpts parametrize one windowed, mask-pruned GROUP BY trace_id scan.
type perTraceScanOpts struct {
stepSeconds int64 // >0 → bucket per-trace values by time (ts column)
groupCols []groupColumn
needed map[string]struct{} // per-trace aliases to select
spanPred string // resolved span-level filter, ANDed per span
spanPredArgs []any
resourcePred string // resource-fingerprint prune (CTE reference or inline subquery)
resourcePredArgs []any
qualified bool // constrain to __qualified
havingExpr string // resolved HAVING predicate over the selected aliases
havingArgs []any
activityExpr string // aggregate expr that must be > 0 for a row to survive (LLM-activity gate)
activityArgs []any
}
// buildPerTraceScan renders the scan: window + gate mask (+ span filter, resource
// prune, qualification), grouped by trace_id (+ ts bucket, group-by columns).
func (b *scopedTraceStatementBuilder) buildPerTraceScan(start, end uint64, resolved []resolvedColumn, maskExpr string, maskArgs []any, o perTraceScanOpts) (string, []any, error) {
startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
endBucket := end / querybuilder.NsToSeconds
sb := sqlbuilder.NewSelectBuilder()
selects := []string{"trace_id"}
if o.stepSeconds > 0 {
selects = append(selects, fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts", o.stepSeconds))
}
for _, gc := range o.groupCols {
gcExpr, err := embedExpr(sb, gc.expr, gc.args)
if err != nil {
return "", nil, err
}
selects = append(selects, fmt.Sprintf("toString(%s) AS `%s`", gcExpr, gc.name))
}
for _, rc := range resolved {
if _, ok := o.needed[rc.alias]; !ok {
continue
}
colExpr, err := embedExpr(sb, rc.expr, rc.args)
if err != nil {
return "", nil, err
}
selects = append(selects, colExpr+" AS "+quoteAlias(rc.alias))
}
sb.Select(selects...)
sb.From(spanTable())
where := windowWhere(sb, start, end, startBucket, endBucket)
mask, err := embedExpr(sb, maskExpr, maskArgs)
if err != nil {
return "", nil, err
}
where = append(where, mask)
if strings.TrimSpace(o.spanPred) != "" {
pred, err := embedExpr(sb, o.spanPred, o.spanPredArgs)
if err != nil {
return "", nil, err
}
where = append(where, pred)
}
if o.resourcePred != "" {
pred, err := embedExpr(sb, o.resourcePred, o.resourcePredArgs)
if err != nil {
return "", nil, err
}
where = append(where, pred)
}
if o.qualified {
where = append(where, "trace_id GLOBAL IN (SELECT trace_id FROM __qualified)")
}
sb.Where(where...)
groupBy := []string{"trace_id"}
if o.stepSeconds > 0 {
groupBy = append(groupBy, "ts")
}
for _, gc := range o.groupCols {
groupBy = append(groupBy, "`"+gc.name+"`")
}
sb.GroupBy(groupBy...)
var having []string
if strings.TrimSpace(o.activityExpr) != "" {
activity, err := embedExpr(sb, o.activityExpr, o.activityArgs)
if err != nil {
return "", nil, err
}
having = append(having, "("+activity+") > 0")
}
if strings.TrimSpace(o.havingExpr) != "" {
hv, err := embedExpr(sb, o.havingExpr, o.havingArgs)
if err != nil {
return "", nil, err
}
having = append(having, hv)
}
if len(having) > 0 {
sb.Having(strings.Join(having, " AND "))
}
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return sql, args, nil
}
// resolveGroupColumns resolves span-attribute group-by keys through the field mapper
// (metadata-aware), for selection inside the per-trace scan.
func (b *scopedTraceStatementBuilder) resolveGroupColumns(ctx context.Context, start, end uint64, groupBy []qbtypes.GroupByKey) ([]groupColumn, error) {
if len(groupBy) == 0 {
return nil, nil
}
selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(groupBy))
for i := range groupBy {
selectors = append(selectors, &telemetrytypes.FieldKeySelector{
Name: groupBy[i].Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: groupBy[i].FieldContext,
FieldDataType: groupBy[i].FieldDataType,
SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
})
}
keys, _, err := b.metadataStore.GetKeysMulti(ctx, selectors)
if err != nil {
return nil, err
}
out := make([]groupColumn, 0, len(groupBy))
for i := range groupBy {
expr, args, err := querybuilder.CollisionHandledFinalExpr(ctx, start, end, &groupBy[i].TelemetryFieldKey, b.fm, b.cb, keys, telemetrytypes.FieldDataTypeString, nil, false)
if err != nil {
return nil, err
}
out = append(out, groupColumn{name: groupBy[i].Name, expr: expr, args: args})
}
return out, nil
}
// ---------------------------------------------------------------------------
// Native trace-domain aggregation query
// ---------------------------------------------------------------------------
// buildTraceAggregationQuery builds the native pipeline (see the file comment).
// start/end are ns.
func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
ctx context.Context,
start, end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
traceAggs []traceAggregation,
) (*qbtypes.Statement, error) {
keys, err := b.fetchKeys(ctx)
if err != nil {
return nil, err
}
maskExpr, maskArgs, err := b.resolveMask(ctx, start, end, keys)
if err != nil {
return nil, err
}
resolved, err := b.resolveColumns(ctx, start, end, keys, maskExpr, maskArgs)
if err != nil {
return nil, err
}
spanExpr, traceHavingPart, err := b.splitUserFilter(ctx, query, variables)
if err != nil {
return nil, err
}
resourceFrag, resourceArgs, resourcePred, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, query, start, end, variables)
if err != nil {
return nil, err
}
var warnings []string
var warningsURL string
var spanPred string
var spanPredArgs []any
if strings.TrimSpace(spanExpr) != "" {
pred, args, warns, url, err := b.resolveSpanPredicate(ctx, start, end, spanExpr, skipResourceFilter, variables)
if err != nil {
return nil, err
}
spanPred, spanPredArgs, warnings, warningsURL = pred, args, warns, url
}
var cteFragments []string
var cteArgs [][]any
if resourceFrag != "" {
cteFragments = append(cteFragments, resourceFrag)
cteArgs = append(cteArgs, resourceArgs)
}
qualified := traceHavingPart != nil
if qualified {
qsql, qargs, err := b.qualifiedScanSQL(start, end, traceHavingPart, resolved, maskExpr, maskArgs, resourcePred, nil)
if err != nil {
return nil, err
}
cteFragments = append(cteFragments, fmt.Sprintf("__qualified AS (%s)", qsql))
cteArgs = append(cteArgs, qargs)
}
groupCols, err := b.resolveGroupColumns(ctx, start, end, query.GroupBy)
if err != nil {
return nil, err
}
groupNames := make([]string, 0, len(groupCols))
for _, gc := range groupCols {
groupNames = append(groupNames, "`"+gc.name+"`")
}
needed := make(map[string]struct{})
for _, ta := range traceAggs {
for a := range ta.used {
needed[a] = struct{}{}
}
}
stepSeconds := int64(0)
rateInterval := (end - start) / querybuilder.NsToSeconds
if requestType == qbtypes.RequestTypeTimeSeries {
stepSeconds = int64(query.StepInterval.Seconds())
rateInterval = uint64(stepSeconds)
}
// LLM-activity gate: per-trace rows with no LLM span in their window/bucket slice
// are dropped, so e.g. count(trace.trace_id) and avg(trace.output_tokens) agree on
// the set of traces they see.
activityExpr, activityArgs := activityGate(b.columnProvider, resolved)
scanOpts := perTraceScanOpts{
stepSeconds: stepSeconds,
groupCols: groupCols,
needed: needed,
spanPred: spanPred,
spanPredArgs: spanPredArgs,
resourcePred: resourcePred,
qualified: qualified,
activityExpr: activityExpr,
activityArgs: activityArgs,
}
// outer aggregation over the per-trace rows
sb := sqlbuilder.NewSelectBuilder()
selects := []string{}
if stepSeconds > 0 {
selects = append(selects, "ts")
}
selects = append(selects, groupNames...)
for i, ta := range traceAggs {
selects = append(selects, fmt.Sprintf("%s AS __result_%d", ta.rendered(rateInterval), i))
}
sb.Select(selects...)
sb.From("__ai_traces")
// grouped, limited time series → rank groups on whole-window per-trace values
// (exact for non-composable aggregates) and constrain the main query to the top-N.
if requestType == qbtypes.RequestTypeTimeSeries && query.Limit > 0 && len(groupCols) > 0 {
totalOpts := scanOpts
totalOpts.stepSeconds = 0
totalSQL, totalArgs, err := b.buildPerTraceScan(start, end, resolved, maskExpr, maskArgs, totalOpts)
if err != nil {
return nil, err
}
cteFragments = append(cteFragments, fmt.Sprintf("__ai_traces_total AS (%s)", totalSQL))
cteArgs = append(cteArgs, totalArgs)
limitSQL, limitArgs := outerLimitSQL(query, traceAggs, groupNames, (end-start)/querybuilder.NsToSeconds)
cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", limitSQL))
cteArgs = append(cteArgs, limitArgs)
tuple := "(" + strings.Join(groupNames, ", ") + ")"
sb.Where(fmt.Sprintf("%s IN (SELECT %s FROM __limit_cte)", tuple, strings.Join(groupNames, ", ")))
}
perTraceSQL, perTraceArgs, err := b.buildPerTraceScan(start, end, resolved, maskExpr, maskArgs, scanOpts)
if err != nil {
return nil, err
}
cteFragments = append(cteFragments, fmt.Sprintf("__ai_traces AS (%s)", perTraceSQL))
cteArgs = append(cteArgs, perTraceArgs)
groupBys := []string{}
if stepSeconds > 0 {
groupBys = append(groupBys, "ts")
}
groupBys = append(groupBys, groupNames...)
if len(groupBys) > 0 {
sb.GroupBy(groupBys...)
}
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
rewritten, err := querybuilder.NewHavingExpressionRewriter().RewriteForTraces(query.Having.Expression, query.Aggregations)
if err != nil {
return nil, err
}
sb.Having(rewritten)
}
if requestType == qbtypes.RequestTypeTimeSeries {
if len(query.Order) != 0 {
for _, orderBy := range query.Order {
if _, ok := traceAggOrderIndex(orderBy, query); !ok {
sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
}
}
sb.OrderBy("ts desc")
}
} else {
for _, orderBy := range query.Order {
if idx, ok := traceAggOrderIndex(orderBy, query); ok {
sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
} else {
sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
}
}
if len(query.Order) == 0 {
sb.OrderBy("__result_0 DESC")
}
if query.Limit > 0 {
sb.Limit(query.Limit)
}
}
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL + " SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000"
finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
return &qbtypes.Statement{
Query: finalSQL,
Args: finalArgs,
Warnings: warnings,
WarningsDocURL: warningsURL,
}, nil
}
// activityGate resolves the provider's activity-gate column (llm_call_count for
// gen_ai) to its aggregate expression; empty when the provider declares none.
func activityGate(provider ColumnProvider, resolved []resolvedColumn) (string, []any) {
alias := provider.ActivityGateAlias()
if alias == "" {
return "", nil
}
for _, rc := range resolved {
if rc.alias == alias {
return rc.expr, rc.args
}
}
return "", nil
}
// rendered returns the outer aggregation SQL, dividing rate aggregations by the
// interval (step for time series, window length for scalar).
func (ta traceAggregation) rendered(rateInterval uint64) string {
if ta.isRate {
return fmt.Sprintf("%s/%d", ta.expr, rateInterval)
}
return ta.expr
}
// outerLimitSQL renders the top-N group selection for a grouped, limited time
// series: the outer aggregations over whole-window per-trace values, ranked and
// limited.
func outerLimitSQL(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], traceAggs []traceAggregation, groupNames []string, windowSeconds uint64) (string, []any) {
sb := sqlbuilder.NewSelectBuilder()
selects := append([]string{}, groupNames...)
for i, ta := range traceAggs {
selects = append(selects, fmt.Sprintf("%s AS __result_%d", ta.rendered(windowSeconds), i))
}
sb.Select(selects...)
sb.From("__ai_traces_total")
sb.GroupBy(groupNames...)
for _, orderBy := range query.Order {
if idx, ok := traceAggOrderIndex(orderBy, query); ok {
sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
} else {
sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
}
}
if len(query.Order) == 0 {
sb.OrderBy("__result_0 DESC")
}
sb.Limit(query.Limit)
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
}
// traceAggOrderIndex reports whether an order key refers to the i-th aggregation
// (by alias, expression, or index), mirroring the trace builder.
func traceAggOrderIndex(k qbtypes.OrderBy, q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) (int, bool) {
for i, agg := range q.Aggregations {
if k.Key.Name == agg.Alias ||
k.Key.Name == agg.Expression ||
k.Key.Name == fmt.Sprintf("%d", i) {
return i, true
}
}
return 0, false
}

View File

@@ -0,0 +1,63 @@
package telemetryscopedtraces
import (
"testing"
"github.com/stretchr/testify/require"
)
// ---------------------------------------------------------------------------
// rewriteTraceAggregation unit tests
// ---------------------------------------------------------------------------
func TestRewriteTraceAggregation(t *testing.T) {
cols := map[string]struct{}{
"input_tokens": {}, "output_tokens": {}, "total_tokens": {}, "llm_call_count": {}, "max_llm_latency_ns": {},
}
cases := []struct {
name string
expr string
isTrace bool
want string // rewritten expr, only checked when isTrace
used []string
wantErr string
}{
{name: "avg trace col", expr: "avg(trace.output_tokens)", isTrace: true, want: "avg(output_tokens)", used: []string{"output_tokens"}},
{name: "tracefield prefix", expr: "sum(tracefield.total_tokens)", isTrace: true, want: "sum(total_tokens)", used: []string{"total_tokens"}},
{name: "count traces", expr: "count(trace.trace_id)", isTrace: true, want: "count(trace_id)"},
{name: "p90 trace col", expr: "p90(trace.max_llm_latency_ns)", isTrace: true, want: "quantile(0.90)(max_llm_latency_ns)", used: []string{"max_llm_latency_ns"}},
{name: "arithmetic between trace cols", expr: "avg(trace.output_tokens + trace.input_tokens)", isTrace: true, want: "avg(output_tokens + input_tokens)", used: []string{"output_tokens", "input_tokens"}},
{name: "arithmetic with constant", expr: "sum(trace.output_tokens * 1.5)", isTrace: true, want: "sum(output_tokens * 1.5)", used: []string{"output_tokens"}},
{name: "ratio of two aggregations", expr: "sum(trace.output_tokens)/count(trace.trace_id)", isTrace: true, want: "sum(output_tokens) / count(trace_id)", used: []string{"output_tokens"}},
{name: "backquoted trace col", expr: "avg(`trace.output_tokens`)", isTrace: true, want: "avg(`output_tokens`)", used: []string{"output_tokens"}},
{name: "bare count is span-level", expr: "count()", isTrace: false},
{name: "span attribute is span-level", expr: "sum(gen_ai.usage.output_tokens)", isTrace: false},
{name: "countIf span predicate is span-level", expr: "countIf(has_error = true)", isTrace: false},
{name: "mixed domains in one expression", expr: "sum(trace.output_tokens) + sum(gen_ai.usage.input_tokens)", wantErr: "mixes trace-level"},
{name: "mixed domains in one function", expr: "sum(trace.output_tokens + gen_ai.usage.input_tokens)", wantErr: "mixes trace-level"},
{name: "output-only column rejected", expr: "avg(trace.span_count)", wantErr: "unknown trace-level aggregation column"},
{name: "unknown column rejected", expr: "avg(trace.bogus)", wantErr: "unknown trace-level aggregation column"},
{name: "countIf over trace col rejected", expr: "countIf(trace.output_tokens > 1000)", wantErr: "not supported"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
ta, isTrace, err := rewriteTraceAggregation(tc.expr, cols)
if tc.wantErr != "" {
require.ErrorContains(t, err, tc.wantErr)
return
}
require.NoError(t, err)
require.Equal(t, tc.isTrace, isTrace)
if !tc.isTrace {
return
}
require.Equal(t, tc.want, ta.expr)
for _, u := range tc.used {
require.Contains(t, ta.used, u)
}
require.Len(t, ta.used, len(tc.used))
})
}
}

View File

@@ -0,0 +1,156 @@
package telemetryscopedtraces
import (
"context"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/querybuilder"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
qbvariables "github.com/SigNoz/signoz/pkg/variables"
"github.com/huandu/go-sqlbuilder"
)
// traceHaving is the resolved trace-level filter part: a HAVING predicate over the
// per-trace column aliases (escaped, `?` placeholders) plus the aliases it references
// (so scans select only what the predicate needs).
type traceHaving struct {
pred string
args []any
used map[string]struct{}
}
// resolveTraceHaving resolves a trace-level filter expression through the standard
// filter pipeline (PrepareWhereClause) against the per-trace column aliases, so
// operators and bound args behave exactly as in span-level filters. Query variables
// are resolved by the canonical replacement (pkg/variables) first — a dynamic
// variable set to __all__ drops its condition for any operator — and the resulting
// literals are then parsed and bound as args. Returns nil when the expression is
// empty or every condition was dropped.
func (b *scopedTraceStatementBuilder) resolveTraceHaving(ctx context.Context, expr string, variables map[string]qbtypes.VariableItem) (*traceHaving, error) {
if strings.TrimSpace(expr) == "" {
return nil, nil
}
allowed := b.orderableColumnSet()
// upfront targeted errors: the visitor folds condition errors into a combined
// "Found N errors" whose details are not part of the error message
if err := validateAggregateFilter(expr, allowed); err != nil {
return nil, err
}
if err := querybuilder.ValidateVariablesInExpr(expr, variables); err != nil {
return nil, err
}
if len(variables) > 0 {
replaced, err := qbvariables.ReplaceVariablesInExpression(expr, variables)
if err != nil {
return nil, err
}
expr = replaced
if strings.TrimSpace(expr) == "" {
return nil, nil
}
}
// every user-facing spelling of an alias resolves to the same synthetic key:
// bare + trace.-prefixed here; tracefield. parses to FieldContextTrace, which
// matches the bare entry's context
fieldKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey, len(allowed)*2)
for alias := range allowed {
key := &telemetrytypes.TelemetryFieldKey{Name: alias, FieldContext: telemetrytypes.FieldContextTrace}
fieldKeys[alias] = []*telemetrytypes.TelemetryFieldKey{key}
fieldKeys["trace."+alias] = []*telemetrytypes.TelemetryFieldKey{key}
}
cb := &aliasConditionBuilder{allowed: allowed, used: make(map[string]struct{})}
prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
Context: ctx,
Logger: b.logger,
ConditionBuilder: cb,
FieldKeys: fieldKeys,
Variables: variables,
})
if err != nil {
return nil, err
}
if prepared.IsEmpty() {
return nil, nil
}
sb := sqlbuilder.NewSelectBuilder()
sb.Select("1")
sb.AddWhereClause(prepared.WhereClause)
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
pred := sql[strings.Index(sql, "WHERE ")+len("WHERE "):]
return &traceHaving{pred: sqlbuilder.Escape(pred), args: args, used: cb.used}, nil
}
// aliasConditionBuilder renders filter conditions directly against the per-trace
// column aliases. It records the aliases it touches; a key that resolves to no alias
// is an unknown/unfilterable aggregate.
type aliasConditionBuilder struct {
allowed map[string]struct{}
used map[string]struct{}
}
var _ qbtypes.ConditionBuilder = (*aliasConditionBuilder)(nil)
func (c *aliasConditionBuilder) ConditionFor(
_ context.Context,
_, _ uint64,
key *telemetrytypes.TelemetryFieldKey,
matching []*telemetrytypes.TelemetryFieldKey,
op qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
if len(matching) == 0 {
name := strings.TrimPrefix(strings.TrimPrefix(key.Name, "tracefield."), "trace.")
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregate %q cannot be used in an AI trace-list filter; filterable aggregates: %s",
name, strings.Join(sortedAliases(c.allowed), ", "))
}
alias := matching[0].Name
c.used[alias] = struct{}{}
col := quoteAlias(alias)
var cond string
switch op {
case qbtypes.FilterOperatorEqual:
cond = sb.E(col, value)
case qbtypes.FilterOperatorNotEqual:
cond = sb.NE(col, value)
case qbtypes.FilterOperatorGreaterThan:
cond = sb.G(col, value)
case qbtypes.FilterOperatorGreaterThanOrEq:
cond = sb.GE(col, value)
case qbtypes.FilterOperatorLessThan:
cond = sb.L(col, value)
case qbtypes.FilterOperatorLessThanOrEq:
cond = sb.LE(col, value)
case qbtypes.FilterOperatorIn, qbtypes.FilterOperatorNotIn:
values, ok := value.([]any)
if !ok {
values = []any{value}
}
if op == qbtypes.FilterOperatorIn {
cond = sb.In(col, values...)
} else {
cond = sb.NotIn(col, values...)
}
case qbtypes.FilterOperatorBetween, qbtypes.FilterOperatorNotBetween:
values, ok := value.([]any)
if !ok || len(values) != 2 {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"between on trace-level aggregate %q requires exactly two values", alias)
}
if op == qbtypes.FilterOperatorBetween {
cond = sb.Between(col, values[0], values[1])
} else {
cond = sb.NotBetween(col, values[0], values[1])
}
default:
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"trace-level aggregate %q supports only comparison operators (=, !=, <, <=, >, >=, in, between)", alias)
}
return []string{cond}, nil, nil
}

View File

@@ -29,6 +29,10 @@ type traceQueryStatementBuilder struct {
resourceFilterResolver *telemetryresourcefilter.ResourceFingerprintResolver[qbtypes.TraceAggregation]
aggExprRewriter qbtypes.AggExprRewriter
skipResourceFingerprintEnabled bool
// traceScope, when set (only on the per-call copy made by BuildTraceScoped),
// constrains raw/scalar/time-series queries to spans whose trace_id is in the
// scope statement, attached as a __trace_scope CTE.
traceScope *qbtypes.Statement
}
var _ qbtypes.StatementBuilder[qbtypes.TraceAggregation] = (*traceQueryStatementBuilder)(nil)
@@ -70,6 +74,33 @@ func NewTraceQueryStatementBuilder(
}
}
// BuildTraceScoped is Build with the query additionally constrained to spans whose
// trace_id is selected by traceScope. The receiver is copied so the shared builder
// stays stateless.
func (b *traceQueryStatementBuilder) BuildTraceScoped(
ctx context.Context,
start uint64,
end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
traceScope *qbtypes.Statement,
) (*qbtypes.Statement, error) {
scoped := *b
scoped.traceScope = traceScope
return scoped.Build(ctx, start, end, requestType, query, variables)
}
// attachTraceScope adds the trace-scope condition to sb and returns the CTE fragment
// + args to prepend; both empty when no scope is set.
func (b *traceQueryStatementBuilder) attachTraceScope(sb *sqlbuilder.SelectBuilder) (string, []any) {
if b.traceScope == nil {
return "", nil
}
sb.Where("trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)")
return fmt.Sprintf("__trace_scope AS (%s)", b.traceScope.Query), b.traceScope.Args
}
// Build builds a SQL query for traces based on the given parameters.
func (b *traceQueryStatementBuilder) Build(
ctx context.Context,
@@ -292,6 +323,11 @@ func (b *traceQueryStatementBuilder) buildListQuery(
cteArgs = append(cteArgs, args)
}
if scopeFrag, scopeArgs := b.attachTraceScope(sb); scopeFrag != "" {
cteFragments = append(cteFragments, scopeFrag)
cteArgs = append(cteArgs, scopeArgs)
}
for _, field := range query.SelectFields {
colExpr, err := b.fm.ColumnExpressionFor(ctx, start, end, &field, keys)
if err != nil {
@@ -491,6 +527,11 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
cteArgs = append(cteArgs, args)
}
if scopeFrag, scopeArgs := b.attachTraceScope(sb); scopeFrag != "" {
cteFragments = append(cteFragments, scopeFrag)
cteArgs = append(cteArgs, scopeArgs)
}
sb.SelectMore(fmt.Sprintf(
"toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts",
int64(query.StepInterval.Seconds()),
@@ -645,6 +686,13 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
cteArgs = append(cteArgs, args)
}
// skipResourceCTE means this scalar is embedded as a CTE of a time-series query,
// which has already emitted the __trace_scope fragment — add only the condition.
if scopeFrag, scopeArgs := b.attachTraceScope(sb); scopeFrag != "" && !skipResourceCTE {
cteFragments = append(cteFragments, scopeFrag)
cteArgs = append(cteArgs, scopeArgs)
}
allAggChArgs := []any{}
var allGroupByArgs []any

View File

@@ -16,18 +16,10 @@ import (
const (
LLMCostFeatureType agentConf.AgentFeatureType = "llm_pricing"
GenAIRequestModel = "gen_ai.request.model"
GenAIProviderName = "gen_ai.provider.name"
GenAIUsageInputTokens = "gen_ai.usage.input_tokens"
GenAIUsageOutputTokens = "gen_ai.usage.output_tokens"
GenAIUsageCacheReadInputTokens = "gen_ai.usage.cache_read.input_tokens"
GenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation.input_tokens"
SignozGenAICostInput = "_signoz.gen_ai.cost_input"
SignozGenAICostOutput = "_signoz.gen_ai.cost_output"
SignozGenAICostCacheRead = "_signoz.gen_ai.cost_cache_read"
SignozGenAICostCacheWrite = "_signoz.gen_ai.cost_cache_write"
SignozGenAITotalCost = "_signoz.gen_ai.total_cost"
)
var (

View File

@@ -4,6 +4,7 @@ import (
"bytes"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"gopkg.in/yaml.v3"
)
@@ -83,11 +84,11 @@ func buildProcessorConfig(rules []*LLMPricingRule) *LLMPricingRuleProcessorConfi
return &LLMPricingRuleProcessorConfig{
Attrs: LLMPricingRuleProcessorAttrs{
Model: GenAIRequestModel,
In: GenAIUsageInputTokens,
Out: GenAIUsageOutputTokens,
CacheRead: GenAIUsageCacheReadInputTokens,
CacheWrite: GenAIUsageCacheCreationInputTokens,
Model: telemetrytypes.GenAIRequestModel,
In: telemetrytypes.GenAIUsageInputTokens,
Out: telemetrytypes.GenAIUsageOutputTokens,
CacheRead: telemetrytypes.GenAIUsageCacheReadInputTokens,
CacheWrite: telemetrytypes.GenAIUsageCacheCreationInputTokens,
},
DefaultPricing: LLMPricingRuleProcessorDefaultPricing{
Rules: pricingRules,
@@ -97,7 +98,7 @@ func buildProcessorConfig(rules []*LLMPricingRule) *LLMPricingRuleProcessorConfi
Out: SignozGenAICostOutput,
CacheRead: SignozGenAICostCacheRead,
CacheWrite: SignozGenAICostCacheWrite,
Total: SignozGenAITotalCost,
Total: telemetrytypes.SignozGenAITotalCost,
},
}
}

View File

@@ -0,0 +1,43 @@
package telemetrytypes
// OpenTelemetry gen_ai semantic-convention attribute keys. Single source of truth
// shared by the AI query builder and the LLM pricing pipeline.
const (
GenAIRequestModel = "gen_ai.request.model"
GenAIToolName = "gen_ai.tool.name"
GenAIAgentName = "gen_ai.agent.name"
GenAIProviderName = "gen_ai.provider.name"
GenAIUsageInputTokens = "gen_ai.usage.input_tokens"
GenAIUsageOutputTokens = "gen_ai.usage.output_tokens"
GenAIUsageCacheReadInputTokens = "gen_ai.usage.cache_read.input_tokens"
GenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation.input_tokens"
GenAIInputMessages = "gen_ai.input.messages"
GenAIOutputMessages = "gen_ai.output.messages"
// SignozGenAITotalCost is not OTel semconv: it is the per-span total cost the
// SigNoz LLM pricing processor computes and attaches (see llmpricingruletypes).
SignozGenAITotalCost = "_signoz.gen_ai.total_cost"
)
// GenAIFieldDefinitions are the gen_ai semantic-convention span attributes the AI
// query builder relies on. They are surfaced by the metadata store for trace
// queries regardless of whether they have been ingested yet, so the AI gate/columns
// resolve on a fresh install (mirrors intrinsic metric keys). String keys are the
// gate; the usage keys are numeric.
var GenAIFieldDefinitions = map[string]TelemetryFieldKey{
GenAIRequestModel: {Name: GenAIRequestModel, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
GenAIToolName: {Name: GenAIToolName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
GenAIAgentName: {Name: GenAIAgentName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
GenAIProviderName: {Name: GenAIProviderName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
GenAIUsageInputTokens: {Name: GenAIUsageInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
GenAIUsageOutputTokens: {Name: GenAIUsageOutputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
GenAIUsageCacheReadInputTokens: {Name: GenAIUsageCacheReadInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
GenAIUsageCacheCreationInputTokens: {Name: GenAIUsageCacheCreationInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
SignozGenAITotalCost: {Name: SignozGenAITotalCost, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
GenAIInputMessages: {Name: GenAIInputMessages, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
GenAIOutputMessages: {Name: GenAIOutputMessages, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
}

View File

@@ -9,6 +9,7 @@ type Source struct {
var (
SourceAudit = Source{valuer.NewString("audit")}
SourceMeter = Source{valuer.NewString("meter")}
SourceAI = Source{valuer.NewString("ai")}
SourceUnspecified = Source{valuer.NewString("")}
)
@@ -17,5 +18,6 @@ var (
func (Source) Enum() []any {
return []any{
SourceMeter,
SourceAI,
}
}

View File

@@ -79,10 +79,13 @@ class BuilderQuery:
name: str = "A"
source: str | None = None
limit: int | None = None
offset: int | None = None
filter_expression: str | None = None
having_expression: str | None = None
select_fields: list[TelemetryFieldKey] | None = None
order: list[OrderBy] | None = None
aggregations: list[Aggregation | MetricAggregation] | None = None
group_by: list[TelemetryFieldKey] | None = None
step_interval: int | None = None
def to_dict(self) -> dict:
@@ -94,14 +97,20 @@ class BuilderQuery:
spec["source"] = self.source
if self.limit is not None:
spec["limit"] = self.limit
if self.offset is not None:
spec["offset"] = self.offset
if self.filter_expression:
spec["filter"] = {"expression": self.filter_expression}
if self.having_expression:
spec["having"] = {"expression": self.having_expression}
if self.select_fields:
spec["selectFields"] = [f.to_dict() for f in self.select_fields]
if self.order:
spec["order"] = [o.to_dict() if hasattr(o, "to_dict") else o for o in self.order]
if self.aggregations:
spec["aggregations"] = [agg.to_dict() if hasattr(agg, "to_dict") else agg for agg in self.aggregations]
if self.group_by:
spec["groupBy"] = [k.to_dict() for k in self.group_by]
if self.step_interval is not None:
spec["stepInterval"] = self.step_interval

View File

@@ -0,0 +1,881 @@
"""
Integration tests for source="ai" over the traces signal.
Data shape (generic OTel gen_ai semantic conventions):
- a root span (no gen_ai attributes)
- an LLM span carrying gen_ai.request.model (str) and numeric usage attributes
(gen_ai.usage.input_tokens / output_tokens / cost) plus gen_ai.user.id
Each test tags its spans with a unique service.name and filters on it, so tests do
not interfere with each other's data.
"""
import json
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 (
BuilderQuery,
OrderBy,
RequestType,
TelemetryFieldKey,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
def _ai_trace(
*,
now: datetime,
service: str,
user: str,
in_tokens: int | None,
out_tokens: int,
cost: float,
model: str = "gpt-4o-mini",
llm_duration_s: float = 1.0,
error: bool = False,
environment: str = "production",
) -> list[Traces]:
"""A minimal AI trace: root span + one LLM span with gen_ai attributes.
in_tokens=None omits the input-tokens attribute entirely (not zero)."""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
llm_id = TraceIdGenerator.span_id()
resources = {"service.name": service, "deployment.environment": environment}
root = Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=llm_duration_s + 0.1),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
attributes = {
"gen_ai.request.model": model,
"gen_ai.system": "openai",
"gen_ai.user.id": user,
# numeric values land in attributes_number
"gen_ai.usage.output_tokens": out_tokens,
"_signoz.gen_ai.total_cost": cost,
}
if in_tokens is not None:
attributes["gen_ai.usage.input_tokens"] = in_tokens
llm = Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=llm_duration_s),
trace_id=trace_id,
span_id=llm_id,
parent_span_id=root_id,
name="chat gpt-4o-mini",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=(TracesStatusCode.STATUS_CODE_ERROR if error else TracesStatusCode.STATUS_CODE_OK),
resources=resources,
attributes=attributes,
)
return [root, llm]
def _non_ai_trace(*, now: datetime, service: str) -> list[Traces]:
"""A plain HTTP trace with no gen_ai attributes; must be excluded by the AI gate."""
trace_id = TraceIdGenerator.trace_id()
span_id = TraceIdGenerator.span_id()
return [
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=span_id,
parent_span_id="",
name="GET /health",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources={"service.name": service},
attributes={"http.request.method": "GET"},
)
]
def _window_ms(now: datetime) -> tuple[int, int]:
start_ms = int((now - timedelta(minutes=10)).timestamp() * 1000)
end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
return start_ms, end_ms
def test_ai_list_excludes_non_ai(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Trace-list panel (requestType="trace"): returns AI traces and excludes the
non-AI trace. Asserts on the raw response payload to stay agnostic to the exact
row schema.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-list"
ai = _ai_trace(now=now, service=service, user="alice", in_tokens=100, out_tokens=20, cost=0.5)
non_ai = _non_ai_trace(now=now, service=service)
ai_trace_id = ai[0].trace_id
non_ai_trace_id = non_ai[0].trace_id
insert_traces(ai + non_ai)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
limit=10,
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert ai_trace_id in body, f"expected AI trace {ai_trace_id} in list response"
assert non_ai_trace_id not in body, f"non-AI trace {non_ai_trace_id} should be excluded by the gate"
def _ai_trace_mixed_spans(*, now: datetime, service: str, user: str) -> list[Traces]:
"""
Root + one LLM span + one tool span + one agent span. The gate matches all three
child spans, but only the LLM span carries gen_ai.request.model.
"""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service, "deployment.environment": "production"}
def _span(name, kind, attributes, offset_s):
return Traces(
timestamp=now - timedelta(seconds=offset_s),
duration=timedelta(seconds=0.5),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name=name,
kind=kind,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes=attributes,
)
root = Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=4),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
llm = _span(
"chat gpt-4o-mini",
TracesKind.SPAN_KIND_CLIENT,
{
"gen_ai.request.model": "gpt-4o-mini",
"gen_ai.system": "openai",
"gen_ai.user.id": user,
"gen_ai.usage.input_tokens": 100,
"gen_ai.usage.output_tokens": 20,
},
4,
)
tool = _span(
"execute_tool",
TracesKind.SPAN_KIND_INTERNAL,
{
"gen_ai.tool.name": "get_weather",
"gen_ai.tool.type": "function",
},
3,
)
agent = _span(
"agent.step",
TracesKind.SPAN_KIND_INTERNAL,
{
"gen_ai.agent.name": "chat-agent",
},
2,
)
return [root, llm, tool, agent]
def test_ai_list_having_aggregate_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Aggregate filter written in the SAME filter box: the span-level predicate narrows
to the service, the trace-level `output_tokens > 100` keeps the large-token
trace and drops the small one (split internally into WHERE + HAVING). All three
spellings of a trace-level aggregate — bare, `trace.`, `tracefield.` — behave
identically (unit tests pin them to byte-identical SQL; this covers the wiring
once end-to-end). An output-only aggregate is rejected under any spelling.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-having"
small = _ai_trace(now=now, service=service, user="alice", in_tokens=10, out_tokens=20, cost=0.1)
large = _ai_trace(now=now, service=service, user="bob", in_tokens=10, out_tokens=500, cost=0.2)
small_id = small[0].trace_id
large_id = large[0].trace_id
insert_traces(small + large)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
for spelling in ("output_tokens", "trace.output_tokens", "tracefield.output_tokens"):
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}' AND {spelling} > 100",
limit=10,
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, f"{spelling}: {response.text}"
body = json.dumps(response.json())
assert large_id in body, f"{spelling}: trace with 500 out-tokens should pass > 100"
assert small_id not in body, f"{spelling}: trace with 20 out-tokens should be filtered out by HAVING"
# output-only aggregate gets the targeted rejection, also under the explicit context.
bad = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression="tracefield.span_count > 3",
limit=10,
)
response = make_query_request(signoz, token, start_ms, end_ms, [bad.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert "cannot be used" in response.text
def test_ai_list_order_limit_offset(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Trace list honors order by (aggregate column) + limit + offset (pagination)."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-order"
traces: list[Traces] = []
for out in (100, 200, 300, 400, 500):
traces += _ai_trace(now=now, service=service, user="u", in_tokens=10, out_tokens=out, cost=0.1)
insert_traces(traces)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
def page(offset: int) -> list[int]:
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
order=[OrderBy(key=TelemetryFieldKey(name="output_tokens"), direction="desc")],
limit=2,
offset=offset,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert resp.status_code == HTTPStatus.OK, resp.text
rows = resp.json()["data"]["data"]["results"][0]["rows"]
return [int(r["data"]["output_tokens"]) for r in rows]
assert page(0) == [500, 400], "first page: highest output_tokens, desc"
assert page(2) == [300, 200], "second page (offset 2): next two, desc"
def test_ai_span_list_limit(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Span list honors limit (delegated raw path): 6 gen_ai spans available, capped to 4."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-spanlimit"
insert_traces(_ai_trace_mixed_spans(now=now, service=service, user="a") + _ai_trace_mixed_spans(now=now, service=service, user="b"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
limit=4,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW)
assert resp.status_code == HTTPStatus.OK, resp.text
rows = resp.json()["data"]["data"]["results"][0]["rows"]
assert len(rows) == 4, f"limit should cap at 4 (6 gen_ai spans available), got {len(rows)}"
def test_ai_span_list_excludes_non_gen_ai_spans(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Span list (requestType=raw): returns only the gen_ai spans (LLM/tool/agent); the
root span of the same trace (no gen_ai attributes) is excluded by the span-level gate.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-spanlist"
insert_traces(_ai_trace_mixed_spans(now=now, service=service, user="alice"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
select_fields=[TelemetryFieldKey(name="name", field_context="span")],
limit=50,
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW)
assert response.status_code == HTTPStatus.OK, response.text
rows = response.json()["data"]["data"]["results"][0]["rows"]
names = sorted(r["data"]["name"] for r in rows)
assert names == ["agent.step", "chat gpt-4o-mini", "execute_tool"], names
assert "POST /api/chat" not in names # root span excluded
def test_ai_span_list_trace_level_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Span list (raw) with a trace-level condition: only gen_ai spans of traces whose
window-clipped aggregates qualify come back (the __trace_scope qualification on
the delegated path). Two traces with out-tokens 100 / 300: `trace.output_tokens
> 100` keeps only the large trace's LLM span.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-spanlist-tracefilter"
small = _ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=100, cost=0.1)
large = _ai_trace(now=now, service=service, user="b", in_tokens=30, out_tokens=300, cost=0.2)
insert_traces(small + large)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}' AND trace.output_tokens > 100",
limit=10,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW)
assert resp.status_code == HTTPStatus.OK, resp.text
rows = resp.json()["data"]["data"]["results"][0]["rows"]
assert len(rows) == 1, f"expected only the large trace's LLM span, got {len(rows)} rows"
body = json.dumps(rows)
assert large[0].trace_id in body
assert small[0].trace_id not in body
def test_ai_list_having_or_aggregates(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Two trace-level aggregates OR-ed within the filter box (regression guard for OR-group
whitespace handling): output_tokens > 100 OR input_tokens > 1000 keeps only the
large-output trace (input_tokens is 10 for both, so that branch never matches).
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-having-or"
small = _ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=20, cost=0.1)
large = _ai_trace(now=now, service=service, user="b", in_tokens=10, out_tokens=500, cost=0.2)
small_id, large_id = small[0].trace_id, large[0].trace_id
insert_traces(small + large)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}' AND (output_tokens > 100 OR input_tokens > 1000)",
limit=10,
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert large_id in body
assert small_id not in body
def test_ai_list_resource_filter_isolates_by_fingerprint(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
A resource attribute in the filter is pulled into the __resource_filter fingerprint
CTE (see maybeAttachResourceFilter). Two traces on the same service but different
deployment.environment: `resource.deployment.environment = 'production'` must keep
the production trace and drop the staging one — the fingerprint prune isolates by
the resource, not by any span attribute.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-resfilter"
prod = _ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=20, cost=0.1, environment="production")
stag = _ai_trace(now=now, service=service, user="b", in_tokens=10, out_tokens=20, cost=0.1, environment="staging")
prod_id, stag_id = prod[0].trace_id, stag[0].trace_id
insert_traces(prod + stag)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=(f"resource.service.name = '{service}' AND resource.deployment.environment = 'production'"),
limit=10,
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert prod_id in body, "production trace should match the resource filter"
assert stag_id not in body, "staging trace should be excluded by the resource fingerprint prune"
def test_ai_list_rejects_aggregate_or_span_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Aggregate (HAVING) columns may not be OR-ed with span-level keys in the trace
list; a span-OR-span filter is fine.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-orfilter"
# seed a trace so service.name resolves as a known key in this window (resource
# keys are discovered from ingested data).
insert_traces(_ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=20, cost=0.1))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
# aggregate OR span -> rejected
bad = BuilderQuery(
signal="traces",
source="ai",
name="A",
limit=10,
filter_expression=f"output_tokens > 1000 OR service.name = '{service}'",
)
response = make_query_request(signoz, token, start_ms, end_ms, [bad.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert "cannot be combined" in response.text
# span OR span -> accepted (result content doesn't matter; just not an error)
ok = BuilderQuery(
signal="traces",
source="ai",
name="A",
limit=10,
filter_expression=f"service.name = '{service}' OR has_error = true",
)
response = make_query_request(signoz, token, start_ms, end_ms, [ok.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, response.text
def test_ai_list_nested_group_span_or_and_aggregate(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
A complex filter that mixes all three routing paths in one expression:
service.name = X AND (has_error = true OR gen_ai.request.model = 'gpt-4o') AND total_tokens > 100
The nested (span OR span) group must not flatten (precedence), the span predicates
go to WHERE as a trace-existence check, and the new `total_tokens` aggregate goes to
HAVING. Three traces isolate each discriminator:
- t_ok: gpt-4o, out=500 -> OR matches (model) AND total_tokens>100 -> IN
- t_or_miss: gpt-4o-mini, out=500 -> OR fails (no error, wrong model) -> OUT
- t_agg_miss: gpt-4o, out=20 -> OR matches but total_tokens<=100 -> OUT
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-nested"
t_ok = _ai_trace(now=now, service=service, user="a", model="gpt-4o", in_tokens=10, out_tokens=500, cost=0.1)
t_or_miss = _ai_trace(now=now, service=service, user="b", model="gpt-4o-mini", in_tokens=10, out_tokens=500, cost=0.1)
t_agg_miss = _ai_trace(now=now, service=service, user="c", model="gpt-4o", in_tokens=10, out_tokens=20, cost=0.1)
insert_traces(t_ok + t_or_miss + t_agg_miss)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=(f"service.name = '{service}' AND (has_error = true OR gen_ai.request.model = 'gpt-4o') AND total_tokens > 100"),
limit=10,
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert t_ok[0].trace_id in body
assert t_or_miss[0].trace_id not in body, "nested (span OR span) group must exclude the wrong-model, no-error trace"
assert t_agg_miss[0].trace_id not in body, "HAVING total_tokens > 100 must exclude the low-token trace"
def test_ai_list_rejects_unknown_aggregate_key(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
) -> None:
"""A trace-level filter on an unknown aggregate name is rejected, not silently run."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
limit=10,
filter_expression="trace.bogus_tokens > 1",
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
def test_ai_list_rejects_order_by_span_attribute(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
) -> None:
"""Only gen_ai-scoped aggregates are orderable; ordering by a span/resource key errors."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
limit=5,
order=[OrderBy(key=TelemetryFieldKey(name="service.name"), direction="asc")],
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert "order key" in response.text
def test_ai_list_total_tokens_output_only(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
A trace whose LLM span carries only output tokens (no input-tokens attribute at
all) must still total: total_tokens is coalesce(sum(in),0)+coalesce(sum(out),0),
since sum over an absent attribute is NULL and NULL + n = NULL in ClickHouse.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-total-coalesce"
insert_traces(_ai_trace(now=now, service=service, user="a", in_tokens=None, out_tokens=300, cost=0.1))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
limit=10,
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, response.text
rows = response.json()["data"]["data"]["results"][0]["rows"]
assert len(rows) == 1, f"expected one trace, got: {rows}"
data = rows[0]["data"]
assert data["input_tokens"] is None, data # attribute absent -> NULL, not 0
assert data["output_tokens"] == 300, data
assert data["total_tokens"] == 300, f"total must coalesce the missing input side: {data}"
def test_ai_list_variable_in_aggregate_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""A query variable in a trace-level condition is substituted into the HAVING."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-having-var"
small = _ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=20, cost=0.1)
large = _ai_trace(now=now, service=service, user="b", in_tokens=10, out_tokens=500, cost=0.2)
small_id, large_id = small[0].trace_id, large[0].trace_id
insert_traces(small + large)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}' AND trace.output_tokens > $threshold",
limit=10,
)
response = make_query_request(
signoz,
token,
start_ms,
end_ms,
[query.to_dict()],
request_type="trace",
variables={"threshold": {"type": "custom", "value": 100}},
)
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert large_id in body
assert small_id not in body
def _ai_trace_two_llm(*, now: datetime, service: str) -> list[Traces]:
"""Root + two LLM spans at different times, each with distinct input/output messages."""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service}
def _llm(offset_s: float, prompt: str, answer: str) -> Traces:
return Traces(
timestamp=now - timedelta(seconds=offset_s),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={
"gen_ai.request.model": "gpt-4o-mini",
"gen_ai.input.messages": prompt,
"gen_ai.output.messages": answer,
},
)
root = Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=4),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
# earlier call is the "first" (its input is the prompt), later call is the "last"
# (its output is the final answer).
first = _llm(4, "first prompt", "first answer")
last = _llm(2, "second prompt", "second answer")
return [root, first, last]
def test_ai_list_messages_first_input_last_output(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
`input` is the FIRST LLM span's prompt (argMin over timestamp) and `output` is the
LAST LLM span's answer (argMax) — the question -> final-answer preview.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-messages"
insert_traces(_ai_trace_two_llm(now=now, service=service))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
limit=10,
filter_expression=f"service.name = '{service}'",
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, response.text
rows = response.json()["data"]["data"]["results"][0]["rows"]
assert len(rows) == 1, f"expected one trace, got: {rows}"
data = rows[0]["data"]
assert data["input"] == "first prompt", f"input should be the earliest call's prompt: {data}"
assert data["output"] == "second answer", f"output should be the latest call's answer: {data}"
def _ai_trace_for_metrics(*, now: datetime, service: str) -> list[Traces]:
"""
Root + one errored LLM span (tokens/cost) + three tool spans (two 'get_weather',
one 'get_time') + one agent span, so the derived per-trace metrics have distinct
expected values. The agent span is in the gen_ai gate but carries no request.model,
so it must NOT count toward llm_call_count (only span_count / last_activity_time).
"""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service}
def _tool(name: str, offset_s: float) -> Traces:
return Traces(
timestamp=now - timedelta(seconds=offset_s),
duration=timedelta(seconds=0.2),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="execute_tool",
kind=TracesKind.SPAN_KIND_INTERNAL,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.tool.name": name, "gen_ai.tool.type": "function"},
)
root = Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=4),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
llm = Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=2),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat gpt-4o-mini",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_ERROR, # -> has_error, drives error_count
resources=resources,
attributes={
"gen_ai.request.model": "gpt-4o-mini",
"gen_ai.usage.input_tokens": 100,
"gen_ai.usage.output_tokens": 20,
"_signoz.gen_ai.total_cost": 0.5,
},
)
agent = Traces(
timestamp=now - timedelta(seconds=1),
duration=timedelta(seconds=0.5),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="agent.step",
kind=TracesKind.SPAN_KIND_INTERNAL,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.agent.name": "chat-agent"},
)
return [root, llm, _tool("get_weather", 3), _tool("get_weather", 2.5), _tool("get_time", 2), agent]
def test_ai_list_enrichment_values(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
End-to-end values of the derived per-trace columns (only integration can check that
ClickHouse computes uniqIf / sum+sum / countIf(predicate) correctly, not just that
the SQL is shaped right). One trace: root + 1 errored LLM + 3 tool spans
(get_weather x2, get_time x1) + 1 agent span. The tool and agent spans are in the
gen_ai gate but carry no request.model, so llm_call_count stays 1 while span_count
counts them all.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-metrics"
insert_traces(_ai_trace_for_metrics(now=now, service=service))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
limit=10,
filter_expression=f"service.name = '{service}'",
)
response = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace")
assert response.status_code == HTTPStatus.OK, response.text
rows = response.json()["data"]["data"]["results"][0]["rows"]
assert len(rows) == 1, f"expected one trace, got: {rows}"
data = rows[0]["data"]
assert data["span_count"] == 6, data # root + llm + 3 tools + agent
assert data["llm_call_count"] == 1, data # only the request.model span, not tool/agent
assert data["tool_call_count"] == 3, data # all three tool spans
assert data["distinct_tool_count"] == 2, data # get_weather, get_time
assert data["input_tokens"] == 100, data
assert data["output_tokens"] == 20, data
assert data["total_tokens"] == 120, data # input + output
assert data["estimated_cost_usd"] == pytest.approx(0.5), data
assert data["error_count"] == 1, data # the errored LLM span
assert data["max_llm_latency_ns"] > 0, data # scoped max over LLM spans

View File

@@ -0,0 +1,598 @@
"""
Integration tests for source="ai" scalar / time-series aggregations.
Aggregations come in two domains, chosen per expression by the `trace.` prefix:
- span-level (bare keys): over individual gen_ai spans (count(), sum(gen_ai.*))
- trace-level (trace.*): over window-clipped per-trace values (avg(trace.output_tokens))
A trace-level condition in the filter (trace.output_tokens > N) qualifies traces by
their window-clipped per-trace values in every request type — the span-list variant
of this is covered in 01_ai_traces.py (test_ai_span_list_trace_level_filter).
Each test tags its spans with a unique service.name and filters on it, so tests do
not interfere with each other's data.
"""
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 (
Aggregation,
BuilderQuery,
OrderBy,
RequestType,
TelemetryFieldKey,
get_scalar_table_data,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
def _ai_trace(
*,
now: datetime,
service: str,
in_tokens: int,
out_tokens: int,
model: str = "gpt-4o-mini",
) -> list[Traces]:
"""A minimal AI trace: root span + one LLM span with gen_ai attributes."""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service}
root = Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=2),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
llm = Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={
"gen_ai.request.model": model,
"gen_ai.usage.input_tokens": in_tokens,
"gen_ai.usage.output_tokens": out_tokens,
},
)
return [root, llm]
def _tool_only_trace(*, now: datetime, service: str) -> list[Traces]:
"""Root + one tool span: passes the gen_ai gate but has NO LLM span."""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service}
return [
Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=2),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/tool",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
),
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=0.5),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="execute_tool",
kind=TracesKind.SPAN_KIND_INTERNAL,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.tool.name": "get_weather", "gen_ai.tool.type": "function"},
),
]
def _window_ms(now: datetime) -> tuple[int, int]:
start_ms = int((now - timedelta(minutes=10)).timestamp() * 1000)
end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
return start_ms, end_ms
def _scalar_query(
service: str,
expression: str,
*,
filter_extra: str = "",
group_by: list[TelemetryFieldKey] | None = None,
alias: str | None = None,
having: str | None = None,
limit: int | None = None,
) -> dict:
filter_expression = f"service.name = '{service}'"
if filter_extra:
filter_expression += f" AND {filter_extra}"
return BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=filter_expression,
aggregations=[Aggregation(expression=expression, alias=alias)],
group_by=group_by,
having_expression=having,
limit=limit,
).to_dict()
def _scalar_value(signoz: types.SigNoz, token: str, start_ms: int, end_ms: int, service: str, expression: str) -> float:
"""Run one single-aggregation scalar query and return its value."""
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[_scalar_query(service, expression)],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, f"{expression}: {resp.text}"
data = get_scalar_table_data(resp.json())
assert len(data) == 1, f"{expression}: expected one row, got {data}"
return float(data[0][-1])
def _series_values(response_json: dict) -> list[list[float]]:
"""Per-series lists of bucket values (bucket order as returned)."""
series = response_json["data"]["data"]["results"][0]["aggregations"][0]["series"]
return [[v["value"] for v in ser["values"]] for ser in series]
def test_ai_scalar_trace_level_aggregations(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Trace-level (trace.) scalar aggregations over per-trace values: two traces with
out-tokens 100 / 300 give avg(trace.output_tokens)=200 and count(trace.trace_id)=2,
while the span-level count() sees the two LLM spans (root spans are gated out).
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-scalar"
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
def scalar_value(expression: str) -> float:
return _scalar_value(signoz, token, start_ms, end_ms, service, expression)
assert scalar_value("avg(trace.output_tokens)") == pytest.approx(200)
assert scalar_value("count(trace.trace_id)") == 2
assert scalar_value("max(trace.total_tokens)") == pytest.approx(330)
assert scalar_value("p50(trace.output_tokens)") == pytest.approx(200) # AggreFuncMap -> quantile(0.50)
# arithmetic inside one function and between functions
assert scalar_value("avg(trace.output_tokens + trace.input_tokens)") == pytest.approx(220)
assert scalar_value("sum(trace.output_tokens)/count(trace.trace_id)") == pytest.approx(200)
# span-level domain still works through the same request type
assert scalar_value("count()") == 2 # the two LLM spans; roots are not gen_ai
assert scalar_value("sum(gen_ai.usage.output_tokens)") == pytest.approx(400)
# multiple trace-level aggregations in one query -> one column per aggregation
multi = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)"), Aggregation(expression="count(trace.trace_id)")],
)
resp = make_query_request(signoz, token, start_ms, end_ms, [multi.to_dict()], request_type=RequestType.SCALAR)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and [float(v) for v in data[0]] == [pytest.approx(200), 2], data
def test_ai_scalar_trace_level_filter_qualifies_traces(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
A trace-level condition in the filter qualifies whole traces before aggregation:
with out-tokens 100 / 300, `trace.output_tokens > 100` keeps only the 300 trace
for both trace-level and span-level aggregations.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-qualify"
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
for expression, expected in (
("sum(trace.output_tokens)", 300), # native trace-domain path
("sum(gen_ai.usage.output_tokens)", 300), # delegated span-domain path (__trace_scope)
):
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[_scalar_query(service, expression, filter_extra="trace.output_tokens > 100")],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and float(data[0][-1]) == pytest.approx(expected), f"{expression}: {data}"
# the qualification also constrains delegated (span-domain) time series
ts = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}' AND trace.output_tokens > 100",
aggregations=[Aggregation(expression="sum(gen_ai.usage.output_tokens)")],
step_interval=60,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [ts.to_dict()], request_type=RequestType.TIME_SERIES)
assert resp.status_code == HTTPStatus.OK, resp.text
values = _series_values(resp.json())
assert values == [[pytest.approx(300)]], values
def test_ai_scalar_group_by_model(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Trace-level aggregation grouped by a span attribute: per-model avg of per-trace tokens."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-groupby"
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o") + _ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + _ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[_scalar_query(service, "avg(trace.output_tokens)", group_by=[TelemetryFieldKey(name="gen_ai.request.model")])],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
by_model = {row[0]: float(row[-1]) for row in data}
assert by_model == {"gpt-4o": pytest.approx(200), "gpt-4o-mini": pytest.approx(50)}, data
def test_ai_timeseries_trace_level_aggregation(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Time-series over per-trace values: all spans fall in one step bucket, avg=200."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-ts"
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
step_interval=60,
)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[query.to_dict()],
request_type=RequestType.TIME_SERIES,
)
assert resp.status_code == HTTPStatus.OK, resp.text
values = _series_values(resp.json())
assert values == [[pytest.approx(200)]], values
def test_ai_timeseries_top_n_groups(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Grouped, limited time series ranks groups on whole-window per-trace values
(__ai_traces_total -> __limit_cte) and returns only the top-N: gpt-4o sums to
400 across two traces vs gpt-4o-mini's 50, so limit=1 keeps only gpt-4o.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-topn"
insert_traces(
_ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o")
+ _ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o")
+ _ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini")
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="sum(trace.output_tokens)")],
group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
step_interval=60,
limit=1,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.TIME_SERIES)
assert resp.status_code == HTTPStatus.OK, resp.text
series = resp.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
assert len(series) == 1, f"limit=1 must keep only the top group, got {len(series)} series"
assert series[0]["labels"][0]["value"] == "gpt-4o", series[0]["labels"]
assert [v["value"] for v in series[0]["values"]] == [pytest.approx(400)]
def test_ai_timeseries_span_time_bucketing(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Per-trace values are clipped per (bucket, trace): one trace with two LLM calls
two minutes apart contributes each call's tokens to its own bucket, not the
whole-trace total to both.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-buckets"
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service}
def _llm(offset_s: float, out_tokens: int) -> Traces:
return Traces(
timestamp=now - timedelta(seconds=offset_s),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.request.model": "gpt-4o-mini", "gen_ai.usage.output_tokens": out_tokens},
)
root = Traces(
timestamp=now - timedelta(seconds=130),
duration=timedelta(seconds=130),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
insert_traces([root, _llm(124, 100), _llm(4, 300)])
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
step_interval=60,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.TIME_SERIES)
assert resp.status_code == HTTPStatus.OK, resp.text
values = _series_values(resp.json())
assert len(values) == 1, values
assert sorted(values[0]) == [pytest.approx(100), pytest.approx(300)], f"each call's tokens in its own bucket: {values}"
def test_ai_scalar_variables_in_trace_level_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Query variables resolve inside trace-level conditions with span-filter semantics;
an unresolvable $var is a 400, not a silent literal comparison.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-vars"
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = _scalar_query(service, "sum(trace.output_tokens)", filter_extra="trace.output_tokens > $threshold")
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[query],
request_type=RequestType.SCALAR,
variables={"threshold": {"type": "text", "value": 100}},
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and float(data[0][-1]) == pytest.approx(300), data
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[query],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
assert "unknown variable" in resp.text
# a dynamic variable resolved to __all__ drops the condition (both traces count)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[query],
request_type=RequestType.SCALAR,
variables={"threshold": {"type": "dynamic", "value": "__all__"}},
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and float(data[0][-1]) == pytest.approx(400), data
def test_ai_scalar_activity_gate_excludes_tool_only_traces(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
A tool-only trace (in the gen_ai gate, no LLM span) must not feed trace-level
aggregations: count(trace.trace_id) sees only the LLM trace, while the span-level
count() still sees both gen_ai spans (LLM + tool).
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-gate"
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _tool_only_trace(now=now, service=service))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
def scalar_value(expression: str) -> float:
return _scalar_value(signoz, token, start_ms, end_ms, service, expression)
# count and avg agree on the trace set — the gate's purpose
assert scalar_value("count(trace.trace_id)") == 1, "tool-only trace must be dropped by the LLM-activity gate"
assert scalar_value("avg(trace.output_tokens)") == pytest.approx(100), "avg over the same gated trace set"
assert scalar_value("count()") == 2, "span-level count still sees the tool span"
def test_ai_scalar_having_on_aggregation(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""The outer having filters aggregation results per group (by alias)."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-having"
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + _ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[
_scalar_query(
service,
"avg(trace.output_tokens)",
group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
alias="avg_out",
having="avg_out > 100",
)
],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and data[0][0] == "gpt-4o", data
def test_ai_aggregation_rejections(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Targeted 400s: mixed domains, group-by on a trace column, raw order by a trace column."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-reject"
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
# span-level and trace-level aggregations cannot be mixed in one query
mixed = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)"), Aggregation(expression="count()")],
)
resp = make_query_request(signoz, token, start_ms, end_ms, [mixed.to_dict()], request_type=RequestType.SCALAR)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
assert "cannot be mixed" in resp.text
# grouping by a trace-level per-trace column is rejected with a targeted error
bad_group = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
group_by=[TelemetryFieldKey(name="trace.llm_call_count")],
)
resp = make_query_request(signoz, token, start_ms, end_ms, [bad_group.to_dict()], request_type=RequestType.SCALAR)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
assert "grouping by trace-level aggregate" in resp.text
# ordering the span list by a trace-level column is rejected with a targeted error
bad_order = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
order=[OrderBy(key=TelemetryFieldKey(name="trace.output_tokens"), direction="desc")],
limit=10,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [bad_order.to_dict()], request_type=RequestType.RAW)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
assert "ordering the span list by trace-level aggregate" in resp.text

View File

@@ -0,0 +1,38 @@
import pytest
from testcontainers.core.container import Network
from fixtures import types
from fixtures.signoz import create_signoz
@pytest.fixture(name="signoz", scope="package")
def signoz_ai_observability(
network: Network,
migrator: types.Operation, # pylint: disable=unused-argument
zeus: types.TestContainerDocker,
gateway: types.TestContainerDocker,
sqlstore: types.TestContainerSQL,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.SigNoz:
"""
Package-scoped SigNoz instance with AI observability enabled. source=ai
queries rely on the metadata store surfacing the static gen_ai key
definitions (enrichWithGenAIKeys), which is gated on this flag — without it
the gate keys (gen_ai.tool.name, gen_ai.agent.name, ...) only resolve once
a span carrying them has been ingested.
"""
return create_signoz(
network=network,
zeus=zeus,
gateway=gateway,
sqlstore=sqlstore,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="signoz-ai-observability",
env_overrides={
"SIGNOZ_FLAGGER_CONFIG_BOOLEAN_ENABLE__AI__OBSERVABILITY": True,
},
)