Compare commits

...

2 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
16 changed files with 2507 additions and 91 deletions

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

@@ -50,6 +50,11 @@ func (genAIColumnProvider) Columns() []scopedtraces.TraceColumn {
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.

View File

@@ -915,8 +915,9 @@ func TestBuild_TraceList_TracefieldPrefixMatchesTracePrefix(t *testing.T) {
require.Contains(t, err.Error(), "cannot be used")
}
// Query variables in a trace-level condition are substituted into the HAVING (the
// span path binds them via PrepareWhereClause; the HAVING is a text rewrite).
// 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) {
@@ -928,17 +929,19 @@ func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
}, vars)
}
// scalar variable -> literal in HAVING
// 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 > 700")
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")
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",

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

@@ -165,6 +165,9 @@ type ColumnProvider interface {
// 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

View File

@@ -4,7 +4,6 @@ import (
"context"
"fmt"
"log/slog"
"sort"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
@@ -16,7 +15,6 @@ import (
"github.com/SigNoz/signoz/pkg/telemetrytraces"
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"
)
@@ -102,14 +100,45 @@ func (b *scopedTraceStatementBuilder) Build(
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
}
}
// buildDelegated ANDs the base gate into the user filter and delegates to the
// standard trace builder (the span-list / raw path).
// 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,
@@ -117,17 +146,34 @@ func (b *scopedTraceStatementBuilder) buildDelegated(
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 query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
expr = fmt.Sprintf("(%s) AND (%s)", gate, query.Filter.Expression)
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}
return b.traceStmtBuilder.Build(ctx, start, end, requestType, gated, variables)
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
@@ -186,7 +232,6 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
if err != nil {
return nil, err
}
orderableSet := orderableAliasSet(resolved)
// If the filter references resource attributes, add a __resource_filter CTE and
// narrow the matched scan by resource_fingerprint; skipResourceFilter then drops
@@ -197,13 +242,13 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
}
// Split the user filter: span-level predicate + trace-level HAVING expression.
fp, err := b.splitFilter(ctx, query, b.aggregateAliasSet(), orderableSet, start, end, skipResourceFilter, variables)
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, orderableSet, maskExpr, maskArgs, fp, resourcePred, limit, query.Offset)
matchedFrag, matchedArgs, err := b.buildMatchedCTE(start, end, startBucket, endBucket, resolved, orders, maskExpr, maskArgs, fp, resourcePred, limit, query.Offset)
if err != nil {
return nil, err
}
@@ -242,20 +287,37 @@ func (b *scopedTraceStatementBuilder) maybeAttachResourceFilter(
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
return nil, false, nil
}
if b.skipResourceFingerprintEnabled {
decision, err := b.resourceFilterResolver.Resolve(ctx, query, start, end, variables)
if err != nil {
return "", nil, "", true, err
return nil, true, err
}
switch decision {
case qbtypes.ResourceFilterResolveKindNoOp:
return "", nil, "", true, nil
return nil, true, nil
case qbtypes.ResourceFilterResolveKindFallback:
return "", nil, "", false, nil
return nil, false, nil
}
}
@@ -263,13 +325,12 @@ func (b *scopedTraceStatementBuilder) maybeAttachResourceFilter(
ctx, start, end, qbtypes.RequestTypeRaw, query, variables,
)
if err != nil {
return "", nil, "", true, err
return nil, true, err
}
if stmt == nil {
return "", nil, "", true, nil
return nil, true, nil
}
return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args,
"resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)", true, nil
return stmt, true, nil
}
// ---------------------------------------------------------------------------
@@ -440,28 +501,28 @@ func (b *scopedTraceStatementBuilder) resolveListOrders(order []qbtypes.OrderBy,
}
// 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 a
// trace-level HAVING expression.
// 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
havingExpr string
having *traceHaving
warnings []string
warningsURL string
}
// splitFilter splits query.Filter into a span-level predicate and a trace-level
// HAVING expression (an explicit query.Having is ANDed onto the latter), then
// validates the trace-level part against the matched-pass aggregates.
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet, orderableSet map[string]struct{}, start, end uint64, skipResourceFilter bool, variables map[string]qbtypes.VariableItem) (filterParts, error) {
// 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
}
fp.havingExpr = traceExpr
havingExpr = traceExpr
if strings.TrimSpace(spanExpr) != "" {
pred, args, warnings, url, err := b.resolveSpanPredicate(ctx, start, end, spanExpr, skipResourceFilter, variables)
if err != nil {
@@ -476,25 +537,17 @@ func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, query qbt
}
}
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
if fp.havingExpr != "" {
fp.havingExpr = fmt.Sprintf("(%s) AND (%s)", fp.havingExpr, query.Having.Expression)
if havingExpr != "" {
havingExpr = fmt.Sprintf("(%s) AND (%s)", havingExpr, query.Having.Expression)
} else {
fp.havingExpr = query.Having.Expression
havingExpr = query.Having.Expression
}
}
// The span predicate binds variables via PrepareWhereClause; the HAVING is a plain
// text rewrite, so substitute variables here (list/IN quoting, __all__ drops the
// condition) before validating.
if strings.TrimSpace(fp.havingExpr) != "" && len(variables) > 0 {
replaced, err := qbvariables.ReplaceVariablesInExpression(fp.havingExpr, variables)
if err != nil {
return fp, err
}
fp.havingExpr = replaced
}
if err := validateAggregateFilter(fp.havingExpr, orderableSet); err != nil {
having, err := b.resolveTraceHaving(ctx, havingExpr, variables)
if err != nil {
return fp, err
}
fp.having = having
return fp, nil
}
@@ -537,11 +590,11 @@ func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context,
// 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, orderableSet map[string]struct{}, maskExpr string, maskArgs []any, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
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.havingExpr, orderableSet)
needed := neededMatchedAliases(orders, fp.having)
selects := []string{"trace_id"}
for _, rc := range resolved {
if _, ok := needed[rc.alias]; !ok {
@@ -594,14 +647,12 @@ func (b *scopedTraceStatementBuilder) buildMatchedCTE(start, end, startBucket, e
having = append(having, "countIf("+havingMask+") > 0")
having = append(having, "countIf("+havingPred+") > 0")
}
if strings.TrimSpace(fp.havingExpr) != "" {
hv, err := b.buildHaving(fp.havingExpr, orderableSet)
if fp.having != nil {
hv, err := embedExpr(sb, fp.having.pred, fp.having.args)
if err != nil {
return "", nil, err
}
if hv != "" {
having = append(having, hv)
}
having = append(having, hv)
}
if len(having) > 0 {
sb.Having(strings.Join(having, " AND "))
@@ -668,18 +719,6 @@ func (b *scopedTraceStatementBuilder) buildEnrichmentSelect(resolved []resolvedC
return sql, append(append([]any{}, selectArgs...), builtArgs...)
}
// buildHaving rewrites a trace-level HAVING expression to the matched-pass column
// aliases; bare, trace., and tracefield. forms all map to the same alias.
func (b *scopedTraceStatementBuilder) buildHaving(havingExpr string, orderableSet map[string]struct{}) (string, error) {
columnMap := make(map[string]string, len(orderableSet)*3)
for a := range orderableSet {
columnMap[a] = quoteAlias(a)
columnMap["trace."+a] = quoteAlias(a)
columnMap["tracefield."+a] = quoteAlias(a)
}
return querybuilder.NewHavingExpressionRewriter().Rewrite(havingExpr, columnMap)
}
// ---------------------------------------------------------------------------
// Small shared SQL-builder utilities
// ---------------------------------------------------------------------------
@@ -704,51 +743,35 @@ func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
return set
}
// orderableAliasSet is the subset of aliases computable in the matched pass — the
// only ones usable in ORDER BY and the aggregate filter.
func orderableAliasSet(resolved []resolvedColumn) map[string]struct{} {
set := make(map[string]struct{})
for _, rc := range resolved {
if rc.orderable {
set[rc.alias] = struct{}{}
}
}
return set
}
// neededMatchedAliases is the minimal alias set the matched pass must select: those
// in ORDER BY plus those in the aggregate HAVING. Everything else is left to the
// enrichment scan.
func neededMatchedAliases(orders []listOrder, havingExpr string, orderableSet map[string]struct{}) map[string]struct{} {
// 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{}{}
}
for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
name := strings.TrimPrefix(strings.TrimPrefix(sel.Name, "trace."), "tracefield.")
if _, ok := orderableSet[name]; ok {
needed[name] = 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).
// 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
}
allowed := make([]string, 0, len(orderableSet))
for a := range orderableSet {
allowed = append(allowed, a)
}
sort.Strings(allowed)
for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
name := strings.TrimPrefix(strings.TrimPrefix(sel.Name, "trace."), "tracefield.")
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(allowed, ", "))
"aggregate %q cannot be used in the trace-list filter; filterable aggregates: %s", name, strings.Join(sortedAliases(orderableSet), ", "))
}
}
return nil

View File

@@ -42,6 +42,7 @@ func (s stubColumns) Columns() []TraceColumn {
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() {

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

@@ -369,6 +369,44 @@ def test_ai_span_list_excludes_non_gen_ai_spans(
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

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,
},
)