Compare commits

...

4 Commits

Author SHA1 Message Date
nityanandagohain
d93f034e41 fix: more tests 2026-07-08 19:58:42 +05:30
nityanandagohain
ccc1bca3ff fix: take perf into consideration 2026-07-08 19:30:26 +05:30
nityanandagohain
cee28bf9b3 feat: support llm trace list and span list 2026-07-08 12:43:46 +05:30
Tushar Vats
7e08f757c8 chore: refactor qb to lossen tight coupling of querybuilder with other modules 2026-07-02 17:40:21 +05:30
48 changed files with 3302 additions and 471 deletions

View File

@@ -21,6 +21,42 @@ func newConditionBuilder(fm qbtypes.FieldMapper) qbtypes.ConditionBuilder {
}
func (c *conditionBuilder) ConditionFor(
ctx context.Context,
startNs uint64,
endNs uint64,
key *telemetrytypes.TelemetryFieldKey,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
// has/hasAny/hasAll/hasToken are logs-body-only; reject for rule state history.
if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
return nil, nil, err
}
keys, warning := querybuilder.ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
if len(keys) == 0 {
return nil, warnings, querybuilder.NewKeyNotFoundError(key.Name)
}
conds := make([]string, 0, len(keys))
for _, k := range keys {
cond, err := c.conditionForKey(ctx, startNs, endNs, k, operator, value, sb)
if err != nil {
return nil, nil, err
}
conds = append(conds, cond)
}
return conds, warnings, nil
}
func (c *conditionBuilder) conditionForKey(
ctx context.Context,
startNs uint64,
endNs uint64,

View File

@@ -35,20 +35,21 @@ var (
)
type querier struct {
logger *slog.Logger
fl flagger.Flagger
telemetryStore telemetrystore.TelemetryStore
metadataStore telemetrytypes.MetadataStore
promEngine prometheus.Prometheus
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation]
meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation]
traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder
bucketCache BucketCache
liveDataRefresh time.Duration
builderConfig builderConfig
logger *slog.Logger
fl flagger.Flagger
telemetryStore telemetrystore.TelemetryStore
metadataStore telemetrytypes.MetadataStore
promEngine prometheus.Prometheus
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation]
meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation]
traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder
bucketCache BucketCache
liveDataRefresh time.Duration
builderConfig builderConfig
}
var _ Querier = (*querier)(nil)
@@ -59,6 +60,7 @@ func New(
metadataStore telemetrytypes.MetadataStore,
promEngine prometheus.Prometheus,
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
@@ -70,19 +72,20 @@ func New(
) *querier {
querierSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/querier")
return &querier{
logger: querierSettings.Logger(),
fl: flagger,
telemetryStore: telemetryStore,
metadataStore: metadataStore,
promEngine: promEngine,
traceStmtBuilder: traceStmtBuilder,
logStmtBuilder: logStmtBuilder,
auditStmtBuilder: auditStmtBuilder,
metricStmtBuilder: metricStmtBuilder,
meterStmtBuilder: meterStmtBuilder,
traceOperatorStmtBuilder: traceOperatorStmtBuilder,
bucketCache: bucketCache,
liveDataRefresh: 5 * time.Second,
logger: querierSettings.Logger(),
fl: flagger,
telemetryStore: telemetryStore,
metadataStore: metadataStore,
promEngine: promEngine,
traceStmtBuilder: traceStmtBuilder,
aiTraceStmtBuilder: aiTraceStmtBuilder,
logStmtBuilder: logStmtBuilder,
auditStmtBuilder: auditStmtBuilder,
metricStmtBuilder: metricStmtBuilder,
meterStmtBuilder: meterStmtBuilder,
traceOperatorStmtBuilder: traceOperatorStmtBuilder,
bucketCache: bucketCache,
liveDataRefresh: 5 * time.Second,
builderConfig: builderConfig{
logTraceIDWindowPaddingMS: uint64(logTraceIDWindowPadding.Milliseconds()),
},
@@ -228,10 +231,18 @@ func (q *querier) buildQueries(
case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
spec.ShiftBy = extractShiftFromBuilderQuery(spec)
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
bq := newBuilderQuery(q.logger, q.telemetryStore, q.traceStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
stmtBuilder := q.traceStmtBuilder
if spec.Source == telemetrytypes.SourceAI {
event.Source = telemetrytypes.SourceAI.StringValue()
stmtBuilder = q.aiTraceStmtBuilder
}
bq := newBuilderQuery(q.logger, q.telemetryStore, stmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
queries[spec.Name] = bq
steps[spec.Name] = spec.StepInterval
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
if spec.Source == telemetrytypes.SourceAI {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "source \"ai\" is only supported for the traces signal, not logs")
}
spec.ShiftBy = extractShiftFromBuilderQuery(spec)
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
stmtBuilder := q.logStmtBuilder
@@ -242,6 +253,9 @@ func (q *querier) buildQueries(
queries[spec.Name] = bq
steps[spec.Name] = spec.StepInterval
case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
if spec.Source == telemetrytypes.SourceAI {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "source \"ai\" is only supported for the traces signal, not metrics")
}
// Spec was already patched by resolveMetricMetadata. Queries
// whose every aggregation was missing live in
// missingMetricQuerySet and produce empty preseeded results
@@ -828,7 +842,11 @@ func (q *querier) createRangedQuery(originalQuery qbtypes.Query, timeRange qbtyp
specCopy := qt.spec.Copy()
specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
return newBuilderQuery(q.logger, q.telemetryStore, q.traceStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
shiftStmtBuilder := q.traceStmtBuilder
if qt.spec.Source == telemetrytypes.SourceAI {
shiftStmtBuilder = q.aiTraceStmtBuilder
}
return newBuilderQuery(q.logger, q.telemetryStore, shiftStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
case *builderQuery[qbtypes.LogAggregation]:
specCopy := qt.spec.Copy()

View File

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

View File

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

View File

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

View File

@@ -220,9 +220,7 @@ func (v *exprVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
FieldKeys: v.fieldKeys,
FieldMapper: v.fieldMapper,
ConditionBuilder: v.conditionBuilder,
BodyJSONEnabled: bodyJSONEnabled,
FullTextColumn: v.fullTextColumn,
JsonKeyToKey: v.jsonKeyToKey,
StartNs: v.startNs,
EndNs: v.endNs,
},

View File

@@ -47,11 +47,11 @@ func CollisionHandledFinalExpr(
addCondition := func(key *telemetrytypes.TelemetryFieldKey) error {
sb := sqlbuilder.NewSelectBuilder()
condition, err := cb.ConditionFor(ctx, startNs, endNs, key, qbtypes.FilterOperatorExists, nil, sb)
conds, _, err := cb.ConditionFor(ctx, startNs, endNs, key, []*telemetrytypes.TelemetryFieldKey{key}, qbtypes.FilterOperatorExists, nil, sb)
if err != nil {
return err
}
sb.Where(condition)
sb.Where(conds[0])
expr, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
expr = strings.TrimPrefix(expr, "WHERE ")

View File

@@ -0,0 +1,147 @@
package querybuilder
import (
"strings"
"github.com/SigNoz/signoz/pkg/errors"
grammar "github.com/SigNoz/signoz/pkg/parser/filterquery/grammar"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/antlr4-go/antlr/v4"
)
// SplitFilterForAggregates partitions a single filter expression into a span-level
// part (a WHERE over spans) and a trace-level part (a HAVING over per-trace
// aggregates), splitting on the top-level AND.
//
// A key is trace-level when it carries the `trace`/`tracefield` field context (e.g.
// `trace.completion_tokens`) or, with no explicit context, its name is in
// aggregateNames. Trace-level and span-level keys may be AND-combined (they run at
// different query stages) but not OR-combined; an OR that mixes the two is an error.
//
// Syntax errors are ignored here — each part is re-parsed downstream (PrepareWhereClause
// for the span part, the HAVING rewriter for the trace part), which surface them.
func SplitFilterForAggregates(query string, aggregateNames map[string]struct{}) (spanExpr string, havingExpr string, err error) {
if strings.TrimSpace(query) == "" {
return "", "", nil
}
s := filterSplitter{query: query, aggregateNames: aggregateNames}
s.visit(parseFilterQuery(query))
if s.mixedOR {
return "", "", errors.NewInvalidInputf(errors.CodeInvalidInput,
"trace-level and span-level filters cannot be combined with OR; use AND")
}
return strings.Join(s.span, " AND "), strings.Join(s.having, " AND "), nil
}
func parseFilterQuery(query string) antlr.Tree {
lexer := grammar.NewFilterQueryLexer(antlr.NewInputStream(query))
lexer.RemoveErrorListeners()
parser := grammar.NewFilterQueryParser(antlr.NewCommonTokenStream(lexer, 0))
parser.RemoveErrorListeners()
return parser.Query()
}
// filterSplitter walks the parse tree once, flattening the top-level AND chain and
// routing each atom (a comparison, a NOT expression, or a whole multi-branch OR group)
// to the span or having bucket by the class of the keys it references.
type filterSplitter struct {
query string
aggregateNames map[string]struct{}
span []string
having []string
mixedOR bool
}
func (s *filterSplitter) visit(node antlr.Tree) {
switch n := node.(type) {
case *grammar.QueryContext:
if n.Expression() != nil {
s.visit(n.Expression())
}
case *grammar.ExpressionContext:
if n.OrExpression() != nil {
s.visit(n.OrExpression())
}
case *grammar.OrExpressionContext:
// a single branch is just an AND chain; multiple branches are a real OR, kept
// whole so a class-mixing OR can be rejected.
if ands := n.AllAndExpression(); len(ands) == 1 {
s.visit(ands[0])
} else {
s.route(n)
}
case *grammar.AndExpressionContext:
for _, u := range n.AllUnaryExpression() {
s.visit(u)
}
case *grammar.UnaryExpressionContext:
if n.NOT() != nil {
s.route(n)
} else if n.Primary() != nil {
s.visit(n.Primary())
}
case *grammar.PrimaryContext:
if n.OrExpression() != nil { // parenthesized sub-expression
s.visit(n.OrExpression())
} else {
s.route(n)
}
}
}
// route classifies an atom and appends its original source text to the right bucket.
func (s *filterSplitter) route(atom antlr.ParserRuleContext) {
isTrace, isSpan := classifyKeys(atom, s.aggregateNames)
if isTrace && isSpan {
s.mixedOR = true
return
}
text := atomSourceText(s.query, atom)
if isTrace {
s.having = append(s.having, text)
} else {
s.span = append(s.span, text)
}
}
// classifyKeys reports whether a subtree references trace-level and/or span-level keys.
// Explicit contexts win; an unspecified-context name is trace-level if it is an
// aggregate name (bare or with the `trace.` prefix).
func classifyKeys(node antlr.Tree, aggregateNames map[string]struct{}) (isTrace, isSpan bool) {
kc, ok := node.(*grammar.KeyContext)
if ok {
key := telemetrytypes.GetFieldKeyFromKeyText(kc.GetText())
switch {
case key.FieldContext == telemetrytypes.FieldContextTrace:
isTrace = true
case key.FieldContext == telemetrytypes.FieldContextUnspecified:
if _, agg := aggregateNames[strings.TrimPrefix(key.Name, "trace.")]; agg {
isTrace = true
} else {
isSpan = true
}
default:
isSpan = true
}
return
}
for i := 0; i < node.GetChildCount(); i++ {
t, s := classifyKeys(node.GetChild(i), aggregateNames)
isTrace = isTrace || t
isSpan = isSpan || s
}
return
}
// atomSourceText returns the original source substring for an atom, preserving
// whitespace. The token stream drops skipped whitespace, which would glue word
// operators (OR/AND/NOT) to their operands, so slice the input by char offsets.
func atomSourceText(query string, atom antlr.ParserRuleContext) string {
start, stop := atom.GetStart(), atom.GetStop()
if start == nil || stop == nil || start.GetStart() < 0 || stop.GetStop() >= len(query) || stop.GetStop() < start.GetStart() {
return atom.GetText()
}
return query[start.GetStart() : stop.GetStop()+1]
}

View File

@@ -0,0 +1,51 @@
package querybuilder
import (
"strings"
"testing"
)
func TestSplitFilterForAggregates(t *testing.T) {
agg := map[string]struct{}{"completion_tokens": {}, "span_count": {}, "prompt_tokens": {}}
cases := []struct {
name string
query string
wantSpan string // substring expected in span part ("" => span empty)
wantHaving string // substring expected in having part ("" => having empty)
wantErr bool
}{
{"span only", "service.name = 'x'", "service.name = 'x'", "", false},
{"agg only bare", "completion_tokens > 1000", "", "completion_tokens > 1000", false},
{"agg only trace ctx", "trace.completion_tokens > 1000", "", "trace.completion_tokens > 1000", false},
{"span AND agg", "service.name = 'x' AND completion_tokens > 1000", "service.name = 'x'", "completion_tokens > 1000", false},
// the whitespace-preservation case: OR between two aggregates must keep spaces
{"agg OR agg", "completion_tokens > 1000 OR span_count > 3", "", "completion_tokens > 1000 OR span_count > 3", false},
{"span OR span", "service.name = 'x' OR kind_string = 'Internal'", "service.name = 'x' OR kind_string = 'Internal'", "", false},
{"agg OR span rejected", "completion_tokens > 1000 OR service.name = 'x'", "", "", true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
span, having, err := SplitFilterForAggregates(tc.query, agg)
if (err != nil) != tc.wantErr {
t.Fatalf("query %q: err=%v wantErr=%v", tc.query, err, tc.wantErr)
}
if tc.wantErr {
return
}
if tc.wantSpan == "" && span != "" {
t.Errorf("query %q: expected empty span, got %q", tc.query, span)
}
if tc.wantSpan != "" && !strings.Contains(span, tc.wantSpan) {
t.Errorf("query %q: span %q missing %q", tc.query, span, tc.wantSpan)
}
if tc.wantHaving == "" && having != "" {
t.Errorf("query %q: expected empty having, got %q", tc.query, having)
}
if tc.wantHaving != "" && !strings.Contains(having, tc.wantHaving) {
t.Errorf("query %q: having %q missing %q", tc.query, having, tc.wantHaving)
}
})
}
}

View File

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

View File

@@ -0,0 +1,82 @@
package querybuilder
import (
"fmt"
"github.com/SigNoz/signoz/pkg/errors"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
const (
// FieldContextDataTypesDocURL documents how to disambiguate a key that resolves
// to multiple field context / data type combinations.
FieldContextDataTypesDocURL = "https://signoz.io/docs/userguide/field-context-data-types/"
// KeyNotFoundDocURL documents the "key not found" error.
KeyNotFoundDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#key-fieldname-not-found"
// Doc URLs for the has/hasAny/hasAll and hasToken "unsupported" errors.
functionBodyJSONSearchDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#function-supports-only-body-json-search"
hasTokenFunctionDocURL = "https://signoz.io/docs/userguide/functions-reference/#hastoken-function"
)
// ResolveKeys picks which matching field keys a filter term builds conditions for.
// With 0 or 1 match it returns the input unchanged and no warning. When a name is
// ambiguous it returns a warning; a resource+attribute mix defaults to the resource
// keys (the common intent), noted in the warning.
func ResolveKeys(field *telemetrytypes.TelemetryFieldKey, fieldKeysForName []*telemetrytypes.TelemetryFieldKey) ([]*telemetrytypes.TelemetryFieldKey, string) {
if len(fieldKeysForName) <= 1 {
return fieldKeysForName, ""
}
warning := fmt.Sprintf(
"Key `%s` is ambiguous, found %d different combinations of field context / data type: %v.",
field.Name,
len(fieldKeysForName),
fieldKeysForName,
)
hasResource, hasAttribute := false, false
for _, item := range fieldKeysForName {
switch item.FieldContext {
case telemetrytypes.FieldContextResource:
hasResource = true
case telemetrytypes.FieldContextAttribute:
hasAttribute = true
}
}
// when there is both resource and attribute context, default to resource only
if hasResource && hasAttribute {
filteredKeys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(fieldKeysForName))
for _, item := range fieldKeysForName {
if item.FieldContext == telemetrytypes.FieldContextResource {
filteredKeys = append(filteredKeys, item)
}
}
fieldKeysForName = filteredKeys
warning += " " + "Using `resource` context by default. To query attributes explicitly, " +
fmt.Sprintf("use the fully qualified name (e.g., 'attribute.%s')", field.Name)
}
return fieldKeysForName, warning
}
// NewKeyNotFoundError builds the error a condition builder returns when a filter term
// references a key it has no matching field key for.
func NewKeyNotFoundError(name string) error {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "key `%s` not found", name).WithUrl(KeyNotFoundDocURL)
}
// NewFunctionUnsupportedError returns the error for a has/hasAny/hasAll/hasToken operator
// on a builder that doesn't support it (logs body only), or nil for other operators.
func NewFunctionUnsupportedError(operator qbtypes.FilterOperator) error {
switch operator {
case qbtypes.FilterOperatorHasToken:
return errors.NewInvalidInputf(errors.CodeInvalidInput, "function `hasToken` only supports body field as first parameter").WithUrl(hasTokenFunctionDocURL)
case qbtypes.FilterOperatorHas, qbtypes.FilterOperatorHasAny, qbtypes.FilterOperatorHasAll:
return errors.NewInvalidInputf(errors.CodeInvalidInput, "function `%s` supports only body JSON search", operator.FunctionName()).WithUrl(functionBodyJSONSearchDocURL)
default:
return nil
}
}

View File

@@ -25,7 +25,6 @@ const stringMatchingOperatorDocURL = "https://signoz.io/docs/userguide/operators
// to convert the parsed filter expressions into ClickHouse WHERE clause.
type filterExpressionVisitor struct {
context context.Context
logger *slog.Logger
fieldMapper qbtypes.FieldMapper
conditionBuilder qbtypes.ConditionBuilder
warnings []string
@@ -35,12 +34,8 @@ type filterExpressionVisitor struct {
mainErrorURL string
builder *sqlbuilder.SelectBuilder
fullTextColumn *telemetrytypes.TelemetryFieldKey
jsonKeyToKey qbtypes.JsonKeyToFieldFunc
bodyJSONEnabled bool
skipResourceFilter bool
skipFullTextFilter bool
skipFunctionCalls bool
ignoreNotFoundKeys bool
variables map[string]qbtypes.VariableItem
keysWithWarnings map[string]bool
@@ -56,12 +51,8 @@ type FilterExprVisitorOpts struct {
FieldKeys map[string][]*telemetrytypes.TelemetryFieldKey
Builder *sqlbuilder.SelectBuilder
FullTextColumn *telemetrytypes.TelemetryFieldKey
JsonKeyToKey qbtypes.JsonKeyToFieldFunc
BodyJSONEnabled bool
SkipResourceFilter bool
SkipFullTextFilter bool
SkipFunctionCalls bool
IgnoreNotFoundKeys bool
Variables map[string]qbtypes.VariableItem
StartNs uint64
EndNs uint64
@@ -71,18 +62,13 @@ type FilterExprVisitorOpts struct {
func newFilterExpressionVisitor(opts FilterExprVisitorOpts) *filterExpressionVisitor {
return &filterExpressionVisitor{
context: opts.Context,
logger: opts.Logger,
fieldMapper: opts.FieldMapper,
conditionBuilder: opts.ConditionBuilder,
fieldKeys: opts.FieldKeys,
builder: opts.Builder,
fullTextColumn: opts.FullTextColumn,
jsonKeyToKey: opts.JsonKeyToKey,
bodyJSONEnabled: opts.BodyJSONEnabled,
skipResourceFilter: opts.SkipResourceFilter,
skipFullTextFilter: opts.SkipFullTextFilter,
skipFunctionCalls: opts.SkipFunctionCalls,
ignoreNotFoundKeys: opts.IgnoreNotFoundKeys,
variables: opts.Variables,
keysWithWarnings: make(map[string]bool),
startNs: opts.StartNs,
@@ -360,16 +346,17 @@ func (v *filterExpressionVisitor) VisitPrimary(ctx *grammar.PrimaryContext) any
return ErrorConditionLiteral
}
}
cond, err := v.conditionBuilder.ConditionFor(context.Background(), v.startNs, v.endNs, v.fullTextColumn, qbtypes.FilterOperatorRegexp, FormatFullTextSearch(searchText), v.builder)
if err != nil {
v.errors = append(v.errors, fmt.Sprintf("failed to build full text search condition: %s", err.Error()))
conds, ok := v.buildConditions(v.fullTextColumn, []*telemetrytypes.TelemetryFieldKey{v.fullTextColumn}, qbtypes.FilterOperatorRegexp, FormatFullTextSearch(searchText))
if !ok {
return ErrorConditionLiteral
}
if v.bodyJSONEnabled && v.fullTextColumn.Name == "body" {
v.warnings = append(v.warnings, BodyFullTextSearchDefaultWarning)
if len(conds) == 0 {
return SkipConditionLiteral
}
return cond
if len(conds) == 1 {
return conds[0]
}
return v.builder.Or(conds...)
}
return ErrorConditionLiteral // Should not happen with valid input
@@ -377,26 +364,26 @@ func (v *filterExpressionVisitor) VisitPrimary(ctx *grammar.PrimaryContext) any
// VisitComparison handles all comparison operators.
func (v *filterExpressionVisitor) VisitComparison(ctx *grammar.ComparisonContext) any {
keys := v.Visit(ctx.Key()).([]*telemetrytypes.TelemetryFieldKey)
key := v.Visit(ctx.Key()).(*telemetrytypes.TelemetryFieldKey)
matching := matchingFieldKeys(key, v.fieldKeys)
// no keys resolved; VisitKey already recorded the error, skip this condition
if len(keys) == 0 {
return ErrorConditionLiteral
}
// this is used to skip the resource filtering on main table if
// the query may use the resources table sub-query filter
if v.skipResourceFilter {
filteredKeys := []*telemetrytypes.TelemetryFieldKey{}
for _, key := range keys {
if key.FieldContext != telemetrytypes.FieldContextResource {
filteredKeys = append(filteredKeys, key)
// Skip resource filtering on the main table when a sub-query covers it. Resolve
// ambiguity first (resource+attribute defaults to resource), then drop resource
// matches; skip the term if nothing remains. Empty matches flow to the builder.
if v.skipResourceFilter && len(matching) > 0 {
resolved, warning := ResolveKeys(key, matching)
// emit the ambiguity warning even when the term is skipped below
v.addWarnings([]string{warning}, len(matching) > 1)
filtered := []*telemetrytypes.TelemetryFieldKey{}
for _, k := range resolved {
if k.FieldContext != telemetrytypes.FieldContextResource {
filtered = append(filtered, k)
}
}
keys = filteredKeys
if len(keys) == 0 {
if len(filtered) == 0 {
return SkipConditionLiteral
}
matching = filtered
}
// Handle EXISTS specially
@@ -405,18 +392,12 @@ func (v *filterExpressionVisitor) VisitComparison(ctx *grammar.ComparisonContext
if ctx.NOT() != nil {
op = qbtypes.FilterOperatorNotExists
}
var conds []string
for _, key := range keys {
condition, err := v.conditionBuilder.ConditionFor(v.context, v.startNs, v.endNs, key, op, nil, v.builder)
if err != nil {
v.errors = append(v.errors, fmt.Sprintf("failed to build condition: %s", err.Error()))
return ErrorConditionLiteral
}
if slices.Contains(SkippableConditionLiterals, condition) {
continue
}
conds = append(conds, condition)
conds, ok := v.buildConditions(key, matching, op, nil)
if !ok {
return ErrorConditionLiteral
}
if len(conds) == 0 {
return SkipConditionLiteral
}
@@ -484,17 +465,12 @@ func (v *filterExpressionVisitor) VisitComparison(ctx *grammar.ComparisonContext
if ctx.NotInClause() != nil {
op = qbtypes.FilterOperatorNotIn
}
var conds []string
for _, key := range keys {
condition, err := v.conditionBuilder.ConditionFor(v.context, v.startNs, v.endNs, key, op, values, v.builder)
if err != nil {
return ErrorConditionLiteral
}
if slices.Contains(SkippableConditionLiterals, condition) {
continue
}
conds = append(conds, condition)
conds, ok := v.buildConditions(key, matching, op, values)
if !ok {
return ErrorConditionLiteral
}
if len(conds) == 0 {
return SkipConditionLiteral
}
@@ -525,29 +501,22 @@ func (v *filterExpressionVisitor) VisitComparison(ctx *grammar.ComparisonContext
switch value1.(type) {
case float64:
if _, ok := value2.(float64); !ok {
v.errors = append(v.errors, fmt.Sprintf("value type mismatch for key %s: expected number for both operands", keys[0].Name))
v.errors = append(v.errors, fmt.Sprintf("value type mismatch for key %s: expected number for both operands", key.Name))
return ErrorConditionLiteral
}
case string:
if _, ok := value2.(string); !ok {
v.errors = append(v.errors, fmt.Sprintf("value type mismatch for key %s: expected string for both operands", keys[0].Name))
v.errors = append(v.errors, fmt.Sprintf("value type mismatch for key %s: expected string for both operands", key.Name))
return ErrorConditionLiteral
}
default:
v.errors = append(v.errors, fmt.Sprintf("value type mismatch for key %s: operands must be number or string", keys[0].Name))
v.errors = append(v.errors, fmt.Sprintf("value type mismatch for key %s: operands must be number or string", key.Name))
return ErrorConditionLiteral
}
var conds []string
for _, key := range keys {
condition, err := v.conditionBuilder.ConditionFor(v.context, v.startNs, v.endNs, key, op, []any{value1, value2}, v.builder)
if err != nil {
return ErrorConditionLiteral
}
if slices.Contains(SkippableConditionLiterals, condition) {
continue
}
conds = append(conds, condition)
conds, ok := v.buildConditions(key, matching, op, []any{value1, value2})
if !ok {
return ErrorConditionLiteral
}
if len(conds) == 0 {
return SkipConditionLiteral
@@ -629,18 +598,11 @@ func (v *filterExpressionVisitor) VisitComparison(ctx *grammar.ComparisonContext
}
}
var conds []string
for _, key := range keys {
condition, err := v.conditionBuilder.ConditionFor(v.context, v.startNs, v.endNs, key, op, value, v.builder)
if err != nil {
v.errors = append(v.errors, fmt.Sprintf("failed to build condition: %s", err.Error()))
return ErrorConditionLiteral
}
if slices.Contains(SkippableConditionLiterals, condition) {
continue
}
conds = append(conds, condition)
conds, ok := v.buildConditions(key, matching, op, value)
if !ok {
return ErrorConditionLiteral
}
if len(conds) == 0 {
return SkipConditionLiteral
}
@@ -718,35 +680,37 @@ func (v *filterExpressionVisitor) VisitFullText(ctx *grammar.FullTextContext) an
v.errors = append(v.errors, "full text search is not supported")
return ErrorConditionLiteral
}
cond, err := v.conditionBuilder.ConditionFor(v.context, v.startNs, v.endNs, v.fullTextColumn, qbtypes.FilterOperatorRegexp, FormatFullTextSearch(text), v.builder)
if err != nil {
v.errors = append(v.errors, fmt.Sprintf("failed to build full text search condition: %s", err.Error()))
conds, ok := v.buildConditions(v.fullTextColumn, []*telemetrytypes.TelemetryFieldKey{v.fullTextColumn}, qbtypes.FilterOperatorRegexp, FormatFullTextSearch(text))
if !ok {
return ErrorConditionLiteral
}
if v.bodyJSONEnabled && v.fullTextColumn.Name == "body" {
v.warnings = append(v.warnings, BodyFullTextSearchDefaultWarning)
if len(conds) == 0 {
return SkipConditionLiteral
}
return cond
if len(conds) == 1 {
return conds[0]
}
return v.builder.Or(conds...)
}
// VisitFunctionCall handles function calls like has(), hasAny(), etc.
func (v *filterExpressionVisitor) VisitFunctionCall(ctx *grammar.FunctionCallContext) any {
if v.skipFunctionCalls {
return SkipConditionLiteral
}
// Get function name based on which token is present
var functionName string
var operator qbtypes.FilterOperator
if ctx.HAS() != nil {
functionName = "has"
operator = qbtypes.FilterOperatorHas
} else if ctx.HASANY() != nil {
functionName = "hasAny"
operator = qbtypes.FilterOperatorHasAny
} else if ctx.HASALL() != nil {
functionName = "hasAll"
operator = qbtypes.FilterOperatorHasAll
} else if ctx.HASTOKEN() != nil {
functionName = "hasToken"
operator = qbtypes.FilterOperatorHasToken
} else {
// Default fallback
v.errors = append(v.errors, fmt.Sprintf("unknown function `%s`", ctx.GetText()))
@@ -759,86 +723,22 @@ func (v *filterExpressionVisitor) VisitFunctionCall(ctx *grammar.FunctionCallCon
return ErrorConditionLiteral
}
keys, ok := params[0].([]*telemetrytypes.TelemetryFieldKey)
key, ok := params[0].(*telemetrytypes.TelemetryFieldKey)
if !ok {
v.errors = append(v.errors, fmt.Sprintf("function `%s` expects key parameter to be a field key", functionName))
return ErrorConditionLiteral
}
// TODO(Tushar): thread orgID here to evaluate correctly
if v.bodyJSONEnabled && functionName != "hasToken" {
filteredKeys := []*telemetrytypes.TelemetryFieldKey{}
for _, key := range keys {
if key.FieldDataType.IsArray() {
filteredKeys = append(filteredKeys, key)
}
}
if len(filteredKeys) == 0 {
v.errors = append(v.errors, fmt.Sprintf("function `%s` expects key parameter to be an array field; no array fields found", functionName))
return ErrorConditionLiteral
}
keys = filteredKeys
}
value := params[1:]
var conds []string
for _, key := range keys {
var fieldName string
if functionName == "hasToken" {
if key.Name != "body" {
if v.mainErrorURL == "" {
v.mainErrorURL = "https://signoz.io/docs/userguide/functions-reference/#hastoken-function"
}
v.errors = append(v.errors, fmt.Sprintf("function `%s` only supports body field as first parameter", functionName))
}
// this will only work with string.
if _, ok := value[0].(string); !ok {
if v.mainErrorURL == "" {
v.mainErrorURL = "https://signoz.io/docs/userguide/functions-reference/#hastoken-function"
}
v.errors = append(v.errors, fmt.Sprintf("function `%s` expects value parameter to be a string", functionName))
return ErrorConditionLiteral
}
conds = append(conds, fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", key.Name, v.builder.Var(value[0])))
} else {
// this is that all other functions only support array fields
if key.FieldContext == telemetrytypes.FieldContextBody {
var err error
if v.bodyJSONEnabled {
fieldName, err = v.fieldMapper.FieldFor(v.context, v.startNs, v.endNs, key)
if err != nil {
v.errors = append(v.errors, fmt.Sprintf("failed to get field name for key %s: %s", key.Name, err.Error()))
return ErrorConditionLiteral
}
} else {
fieldName, _ = v.jsonKeyToKey(v.context, key, qbtypes.FilterOperatorUnknown, value)
}
} else {
// TODO(add docs for json body search)
if v.mainErrorURL == "" {
v.mainErrorURL = "https://signoz.io/docs/userguide/search-troubleshooting/#function-supports-only-body-json-search"
}
v.errors = append(v.errors, fmt.Sprintf("function `%s` supports only body JSON search", functionName))
return ErrorConditionLiteral
}
var cond string
// Map our functions to ClickHouse equivalents
switch functionName {
case "has":
cond = fmt.Sprintf("has(%s, %s)", fieldName, v.builder.Var(value[0]))
case "hasAny":
cond = fmt.Sprintf("hasAny(%s, %s)", fieldName, v.builder.Var(value[0]))
case "hasAll":
cond = fmt.Sprintf("hasAll(%s, %s)", fieldName, v.builder.Var(value[0]))
}
conds = append(conds, cond)
}
conds, ok := v.buildConditions(key, matchingFieldKeys(key, v.fieldKeys), operator, value)
if !ok {
return ErrorConditionLiteral
}
if len(conds) == 0 {
return SkipConditionLiteral
}
if len(conds) == 1 {
return conds[0]
}
@@ -903,106 +803,41 @@ func (v *filterExpressionVisitor) VisitValue(ctx *grammar.ValueContext) any {
return ErrorConditionLiteral // Should not happen with valid input
}
// VisitKey handles field/column references.
// VisitKey resolves a field/column reference to its parsed key. It makes no
// decisions — the condition builder owns those.
func (v *filterExpressionVisitor) VisitKey(ctx *grammar.KeyContext) any {
fieldKey := telemetrytypes.GetFieldKeyFromKeyText(ctx.GetText())
keyName := fieldKey.Name
return &fieldKey
}
// GetFieldKeyFromKeyText function extracts the context prefix (attribute., resource., body.) if present
// extracts data type if present (e.g., :string, :int)
// so we need to filter the available field keys based on the provided context and data type
fieldKeysForName := []*telemetrytypes.TelemetryFieldKey{}
// buildConditions invokes the condition builder for a filter term, folding its
// warnings/errors into visitor state. It returns the conditions and false if an error
// was recorded.
func (v *filterExpressionVisitor) buildConditions(key *telemetrytypes.TelemetryFieldKey, matching []*telemetrytypes.TelemetryFieldKey, op qbtypes.FilterOperator, value any) ([]string, bool) {
conds, warns, err := v.conditionBuilder.ConditionFor(v.context, v.startNs, v.endNs, key, matching, op, value, v.builder)
if err != nil {
_, _, _, _, errURL, _ := errors.Unwrapb(err)
assignIfEmpty(&v.mainErrorURL, errURL)
v.errors = append(v.errors, err.Error())
return nil, false
}
v.addWarnings(warns, len(matching) > 1)
return conds, true
}
// If user said key = 'xyz', we look up in the map for all possible field keys with name 'xyz'
for _, item := range v.fieldKeys[keyName] {
// Match items where:
// 1) user didn't specify context OR context matches, AND
// 2) user didn't specify data type OR data type matches.
if (fieldKey.FieldContext == telemetrytypes.FieldContextUnspecified || fieldKey.FieldContext == item.FieldContext) &&
(fieldKey.FieldDataType == telemetrytypes.FieldDataTypeUnspecified || fieldKey.FieldDataType == item.FieldDataType) {
fieldKeysForName = append(fieldKeysForName, item)
// addWarnings appends de-duplicated warnings to the visitor. ambiguous marks warnings
// from a multi-match key so the field-context doc URL is attached.
func (v *filterExpressionVisitor) addWarnings(warns []string, ambiguous bool) {
for _, w := range warns {
if w == "" || v.keysWithWarnings[w] {
continue
}
v.keysWithWarnings[w] = true
v.warnings = append(v.warnings, w)
if ambiguous {
assignIfEmpty(&v.mainWarnURL, FieldContextDataTypesDocURL)
}
}
// Now consider that GetFieldKeyFromKeyText may have extracted the context which was actually part of the name
// If user said attribute.key = 'xyz',
// 1. either user meant key ( this is already handled above in fieldKeysForName )
// 2. or user meant `attribute.key` we look up in the map for all possible field keys with name 'attribute.key'
// Note:
// If user only wants to search `attribute.key`, then they have to use `attribute.attribute.key`
// If user only wants to search `key`, then they have to use `key`
// If user wants to search both, they can use `attribute.key` and we will resolve the ambiguity
if fieldKey.FieldContext != telemetrytypes.FieldContextUnspecified {
contextPrefixedFieldName := fmt.Sprintf("%s.%s", fieldKey.FieldContext.StringValue(), fieldKey.Name)
for _, item := range v.fieldKeys[contextPrefixedFieldName] {
// Match items where: user didn't specify data type OR data type matches. Context filtering not needed here since context was used to construct the lookup key.
if fieldKey.FieldDataType == telemetrytypes.FieldDataTypeUnspecified || item.FieldDataType == fieldKey.FieldDataType {
fieldKeysForName = append(fieldKeysForName, item)
}
}
}
// for the body json search, we need to add search on the body field even
// if there is a field with the same name as attribute/resource attribute
// Since it will ORed with the fieldKeysForName, it will not result empty
// when either of them have values
// Note: Skip this logic if body json query is enabled so we can look up the key inside fields
//
// TODO(Piyush): After entire migration this is supposed to be removed.
// TODO(Tushar): thread orgID here to evaluate correctly
if fieldKey.FieldContext == telemetrytypes.FieldContextBody && !v.bodyJSONEnabled {
fieldKeysForName = append(fieldKeysForName, &fieldKey)
}
if len(fieldKeysForName) == 0 {
if fieldKey.FieldContext == telemetrytypes.FieldContextBody && keyName == "" {
v.errors = append(v.errors, "missing key for body json search - expected key of the form `body.key` (ex: `body.status`)")
} else if !v.ignoreNotFoundKeys {
// TODO(srikanthccv): do we want to return an error here?
// should we infer the type and auto-magically build a key for expression?
v.errors = append(v.errors, fmt.Sprintf("key `%s` not found", fieldKey.Name))
v.mainErrorURL = "https://signoz.io/docs/userguide/search-troubleshooting/#key-fieldname-not-found"
}
}
if len(fieldKeysForName) > 1 {
warnMsg := fmt.Sprintf(
"Key `%s` is ambiguous, found %d different combinations of field context / data type: %v.",
fieldKey.Name,
len(fieldKeysForName),
fieldKeysForName,
)
mixedFieldContext := map[string]bool{}
for _, item := range fieldKeysForName {
mixedFieldContext[item.FieldContext.StringValue()] = true
}
// when there is both resource and attribute context, default to resource only
if mixedFieldContext[telemetrytypes.FieldContextResource.StringValue()] &&
mixedFieldContext[telemetrytypes.FieldContextAttribute.StringValue()] {
filteredKeys := []*telemetrytypes.TelemetryFieldKey{}
for _, item := range fieldKeysForName {
if item.FieldContext != telemetrytypes.FieldContextResource {
continue
}
filteredKeys = append(filteredKeys, item)
}
fieldKeysForName = filteredKeys
warnMsg += " " + "Using `resource` context by default. To query attributes explicitly, " +
fmt.Sprintf("use the fully qualified name (e.g., 'attribute.%s')", fieldKey.Name)
}
if !v.keysWithWarnings[keyName] {
v.mainWarnURL = "https://signoz.io/docs/userguide/field-context-data-types/"
// this is warning state, we must have a unambiguous key
v.warnings = append(v.warnings, warnMsg)
}
v.keysWithWarnings[keyName] = true
v.logger.Warn("ambiguous key", slog.String("field_key_name", fieldKey.Name)) //nolint:sloglint
}
return fieldKeysForName
}
// hasLikeWildcards checks if a value contains LIKE wildcards (% or _).
@@ -1028,3 +863,38 @@ func trimQuotes(txt string) string {
txt = strings.ReplaceAll(txt, `\'`, `'`)
return txt
}
func assignIfEmpty(s *string, value string) {
if *s == "" {
*s = value
}
}
// matchingFieldKeys returns the field keys from the provided map that match the given
// key, honoring any context/data type the user specified. It also resolves the case
// where GetFieldKeyFromKeyText split a context off a name that legitimately contained it.
func matchingFieldKeys(field *telemetrytypes.TelemetryFieldKey, fieldKeys map[string][]*telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
fieldKeysForName := []*telemetrytypes.TelemetryFieldKey{}
// match by name; keep items whose context and data type match (unspecified matches any)
for _, item := range fieldKeys[field.Name] {
if (field.FieldContext == telemetrytypes.FieldContextUnspecified || field.FieldContext == item.FieldContext) &&
(field.FieldDataType == telemetrytypes.FieldDataTypeUnspecified || field.FieldDataType == item.FieldDataType) {
fieldKeysForName = append(fieldKeysForName, item)
}
}
// A context may have been split off a name that legitimately contained it (e.g.
// `attribute.key`); also look up the context-prefixed name so both readings resolve.
if field.FieldContext != telemetrytypes.FieldContextUnspecified {
contextPrefixedFieldName := fmt.Sprintf("%s.%s", field.FieldContext.StringValue(), field.Name)
for _, item := range fieldKeys[contextPrefixedFieldName] {
// Context already matched via the lookup key; only data type needs checking.
if field.FieldDataType == telemetrytypes.FieldDataTypeUnspecified || item.FieldDataType == field.FieldDataType {
fieldKeysForName = append(fieldKeysForName, item)
}
}
}
return fieldKeysForName
}

View File

@@ -3,10 +3,10 @@ package querybuilder
import (
"context"
"fmt"
"log/slog"
"strings"
"testing"
"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"
@@ -79,15 +79,13 @@ func TestPrepareWhereClause_EmptyVariableList(t *testing.T) {
}
// createTestVisitor creates a filterExpressionVisitor for testing VisitKey.
func createTestVisitor(t *testing.T, fieldKeys map[string][]*telemetrytypes.TelemetryFieldKey, ignoreNotFoundKeys bool) *filterExpressionVisitor {
func createTestVisitor(t *testing.T, fieldKeys map[string][]*telemetrytypes.TelemetryFieldKey) *filterExpressionVisitor {
t.Helper()
return &filterExpressionVisitor{
context: t.Context(),
logger: slog.Default(),
fieldKeys: fieldKeys,
ignoreNotFoundKeys: ignoreNotFoundKeys,
keysWithWarnings: make(map[string]bool),
builder: sqlbuilder.NewSelectBuilder(),
context: t.Context(),
fieldKeys: fieldKeys,
keysWithWarnings: make(map[string]bool),
builder: sqlbuilder.NewSelectBuilder(),
}
}
@@ -160,7 +158,7 @@ func TestVisitKey(t *testing.T) {
name: "Key not found",
keyText: "unknown_key",
fieldKeys: map[string][]*telemetrytypes.TelemetryFieldKey{
"service": []*telemetrytypes.TelemetryFieldKey{
"service": {
{
Name: "service",
Signal: telemetrytypes.SignalLogs,
@@ -361,7 +359,7 @@ func TestVisitKey(t *testing.T) {
name: "Unknown key with ignoreNotFoundKeys=true",
keyText: "unknown_key",
fieldKeys: map[string][]*telemetrytypes.TelemetryFieldKey{
"service": []*telemetrytypes.TelemetryFieldKey{
"service": {
{
Name: "service",
Signal: telemetrytypes.SignalLogs,
@@ -574,7 +572,7 @@ func TestVisitKey(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
visitor := createTestVisitor(t, tt.fieldKeys, tt.ignoreNotFoundKeys)
visitor := createTestVisitor(t, tt.fieldKeys)
keyCtx := parseKeyContext(tt.keyText)
if keyCtx == nil {
@@ -582,10 +580,30 @@ func TestVisitKey(t *testing.T) {
}
result := visitor.VisitKey(keyCtx)
keys, ok := result.([]*telemetrytypes.TelemetryFieldKey)
key, ok := result.(*telemetrytypes.TelemetryFieldKey)
if !ok {
t.Fatalf("expected []*TelemetryFieldKey, got %T", result)
t.Fatalf("expected *TelemetryFieldKey, got %T", result)
}
// VisitKey only parses; the condition builder matches, resolves ambiguity
// and decides not-found handling. Replay that here against the generic
// builder behavior (error unless the key is ignored).
matching := matchingFieldKeys(key, tt.fieldKeys)
keys, warning := ResolveKeys(key, matching)
var gotErrors []string
var gotMainErrURL, gotMainWrnURL string
var gotWarnings []string
if len(keys) == 0 && !tt.ignoreNotFoundKeys {
err := NewKeyNotFoundError(key.Name)
gotErrors = append(gotErrors, err.Error())
_, _, _, _, gotMainErrURL, _ = errors.Unwrapb(err)
}
if warning != "" {
gotWarnings = append(gotWarnings, warning)
if len(matching) > 1 {
gotMainWrnURL = FieldContextDataTypesDocURL
}
}
// Check expected keys count
@@ -612,55 +630,55 @@ func TestVisitKey(t *testing.T) {
// Check errors
if tt.expectedErrors != nil {
if len(visitor.errors) != len(tt.expectedErrors) {
t.Errorf("expected %d errors, got %d: %v", len(tt.expectedErrors), len(visitor.errors), visitor.errors)
if len(gotErrors) != len(tt.expectedErrors) {
t.Errorf("expected %d errors, got %d: %v", len(tt.expectedErrors), len(gotErrors), gotErrors)
}
for _, expectedError := range tt.expectedErrors {
found := false
for _, err := range visitor.errors {
for _, err := range gotErrors {
if strings.Contains(err, expectedError) {
found = true
break
}
}
if !found {
t.Errorf("expected error containing %q, got errors: %v", expectedError, visitor.errors)
t.Errorf("expected error containing %q, got errors: %v", expectedError, gotErrors)
}
}
} else {
if len(visitor.errors) != 0 {
t.Errorf("expected no errors, got %d: %v", len(visitor.errors), visitor.errors)
if len(gotErrors) != 0 {
t.Errorf("expected no errors, got %d: %v", len(gotErrors), gotErrors)
}
}
// Check mainErrorURL
if visitor.mainErrorURL != tt.expectedMainErrURL {
t.Errorf("expected mainErrorURL %q, got %q", tt.expectedMainErrURL, visitor.mainErrorURL)
if gotMainErrURL != tt.expectedMainErrURL {
t.Errorf("expected mainErrorURL %q, got %q", tt.expectedMainErrURL, gotMainErrURL)
}
// Check warnings
if tt.expectedWarnings != nil {
for _, expectedWarn := range tt.expectedWarnings {
found := false
for _, warn := range visitor.warnings {
for _, warn := range gotWarnings {
if strings.Contains(strings.ToLower(warn), strings.ToLower(expectedWarn)) {
found = true
break
}
}
if !found {
t.Errorf("expected warning containing %q, got warnings: %v", expectedWarn, visitor.warnings)
t.Errorf("expected warning containing %q, got warnings: %v", expectedWarn, gotWarnings)
}
}
} else {
if len(visitor.warnings) != 0 {
t.Errorf("expected no warnings, got %d: %v", len(visitor.warnings), visitor.warnings)
if len(gotWarnings) != 0 {
t.Errorf("expected no warnings, got %d: %v", len(gotWarnings), gotWarnings)
}
}
// Check mainWarnURL
if visitor.mainWarnURL != tt.expectedMainWrnURL {
t.Errorf("expected mainWarnURL %q, got %q", tt.expectedMainWrnURL, visitor.mainWarnURL)
if gotMainWrnURL != tt.expectedMainWrnURL {
t.Errorf("expected mainWarnURL %q, got %q", tt.expectedMainWrnURL, gotMainWrnURL)
}
})
}
@@ -673,13 +691,12 @@ func TestVisitKey(t *testing.T) {
// This suite exercises the visitor with two different configurations
// side-by-side for each expression, asserting both expected outputs:
//
// resourceConditionBuilder (wantRSB) — returns TrueConditionLiteral for
// non-resource keys and "{name}_cond" for resource keys (x, y, z).
// Opts: SkipFullTextFilter:true, SkipFunctionCalls:true, IgnoreNotFoundKeys:true.
// resourceConditionBuilder (wantRSB) — produces "{name}_cond" only for resource
// keys (x, y, z); non-resource keys, unknown keys, and function calls yield no
// condition. Opts: SkipFullTextFilter:true.
//
// conditionBuilder (wantSB) — returns "{name}_cond" for every key regardless
// of FieldContext. Opts: SkipFullTextFilter:false, SkipFunctionCalls:false,
// IgnoreNotFoundKeys:false, FullTextColumn:bodyCol.
// of FieldContext. Opts: SkipFullTextFilter:false, FullTextColumn:bodyCol.
//
// Key behavioral rules:
//
@@ -732,16 +749,32 @@ func (b *resourceConditionBuilder) ConditionFor(
_ uint64,
_ uint64,
key *telemetrytypes.TelemetryFieldKey,
_ qbtypes.FilterOperator,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
_ any,
_ *sqlbuilder.SelectBuilder,
) (string, error) {
) ([]string, []string, error) {
if key.FieldContext != telemetrytypes.FieldContextResource {
return SkipConditionLiteral, nil
// mirror the real resource builder: function operators never apply to resources
if operator.IsFunctionOperator() {
return nil, nil, nil
}
return fmt.Sprintf("%s_cond", key.Name), nil
keys, warning := ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
var conds []string
for _, k := range keys {
// only resource keys contribute; others (and unknown keys) are ignored
if k.FieldContext != telemetrytypes.FieldContextResource {
continue
}
conds = append(conds, fmt.Sprintf("%s_cond", k.Name))
}
return conds, warnings, nil
}
type conditionBuilder struct{}
@@ -751,19 +784,44 @@ func (b *conditionBuilder) ConditionFor(
_ uint64,
_ uint64,
key *telemetrytypes.TelemetryFieldKey,
_ qbtypes.FilterOperator,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
_ any,
_ *sqlbuilder.SelectBuilder,
) (string, error) {
) ([]string, []string, error) {
return fmt.Sprintf("%s_cond", key.Name), nil
// has/hasAny/hasAll/hasToken only support body fields; mirror the real
// condition builder which now owns this validation and errors for non-body keys.
switch operator {
case qbtypes.FilterOperatorHas, qbtypes.FilterOperatorHasAny, qbtypes.FilterOperatorHasAll, qbtypes.FilterOperatorHasToken:
if key.FieldContext != telemetrytypes.FieldContextBody {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "function supports only body JSON search")
}
return []string{fmt.Sprintf("%s_cond", key.Name)}, nil, nil
}
keys, warning := ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
if len(keys) == 0 {
// errors on unknown keys (no IgnoreNotFoundKeys equivalent for this builder)
return nil, warnings, NewKeyNotFoundError(key.Name)
}
conds := make([]string, 0, len(keys))
for _, k := range keys {
conds = append(conds, fmt.Sprintf("%s_cond", k.Name))
}
return conds, warnings, nil
}
// visitComparisonCase is a single test case for the TestVisitComparison_* family.
// Each case is run under two independent configurations:
//
// - rsbOpts (resourceConditionBuilder): skips full-text and function calls,
// ignores unknown keys, produces conditions only for resource-context keys.
// - rsbOpts (resourceConditionBuilder): skips full-text; its builder skips function
// calls and unknown keys, producing conditions only for resource-context keys.
//
// - sbOpts (conditionBuilder): skips resource-context keys (unless OR is present),
// evaluates full-text, errors on unknown keys.
@@ -799,8 +857,6 @@ func visitComparisonOpts(t *testing.T) (rsbOpts, sbOpts FilterExprVisitorOpts) {
Variables: allVariable,
SkipResourceFilter: false,
SkipFullTextFilter: true,
SkipFunctionCalls: true,
IgnoreNotFoundKeys: true,
}
sbOpts = FilterExprVisitorOpts{
Context: t.Context(),
@@ -809,8 +865,6 @@ func visitComparisonOpts(t *testing.T) (rsbOpts, sbOpts FilterExprVisitorOpts) {
Variables: allVariable,
SkipResourceFilter: true,
SkipFullTextFilter: false,
SkipFunctionCalls: false,
IgnoreNotFoundKeys: false,
FullTextColumn: bodyCol,
}
return
@@ -1537,9 +1591,9 @@ func TestVisitComparison_AllVariable(t *testing.T) {
}
// TestVisitComparison_FunctionCalls covers function call expressions (has, hasAny, hasAll).
// rsbOpts has SkipFunctionCalls=true → TrueConditionLiteral (function never evaluated).
// sbOpts has SkipFunctionCalls=false; has/hasAny/hasAll only support FieldContextBody,
// so calls on attribute/resource keys return an error.
// The resource builder skips function operators, so they yield no condition in RSB.
// In SB, has/hasAny/hasAll only support FieldContextBody, so calls on attribute/resource
// keys return an error.
func TestVisitComparison_FunctionCalls(t *testing.T) {
rsbOpts, sbOpts := visitComparisonOpts(t)
tests := []visitComparisonCase{

View File

@@ -0,0 +1,55 @@
package telemetryai
import (
"strings"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// BaseConditionProvider defines which spans (and traces) are in scope, decoupled
// from the query. Swap it to redefine "an AI trace" without touching the builder.
//
// It only *declares* the gate (a grammar expression + its field keys); the builder
// resolves those keys through the field mapper, so all attribute access is
// materialization/evolution aware — no hardcoded map lookups.
type BaseConditionProvider interface {
// FilterExpression is the grammar-level (EXISTS) gate, resolved via the visitor.
FilterExpression() string
// FieldKeys are the gate's keys: registered in metadata and used to build the
// per-span mask (OR of resolved EXISTS conditions).
FieldKeys() []*telemetrytypes.TelemetryFieldKey
}
// genAIBaseConditionProvider: an AI trace has >=1 gen_ai LLM, tool, or agent span.
type genAIBaseConditionProvider struct {
keys []string
}
var _ BaseConditionProvider = (*genAIBaseConditionProvider)(nil)
func NewGenAIBaseConditionProvider() BaseConditionProvider {
return &genAIBaseConditionProvider{
keys: []string{telemetrytypes.GenAIRequestModel, telemetrytypes.GenAIToolName, telemetrytypes.GenAIAgentName},
}
}
func (p *genAIBaseConditionProvider) FilterExpression() string {
parts := make([]string, 0, len(p.keys))
for _, k := range p.keys {
parts = append(parts, k+" EXISTS")
}
return strings.Join(parts, " OR ")
}
func (p *genAIBaseConditionProvider) FieldKeys() []*telemetrytypes.TelemetryFieldKey {
keys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(p.keys))
for _, k := range p.keys {
keys = append(keys, &telemetrytypes.TelemetryFieldKey{
Name: k,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
})
}
return keys
}

View File

@@ -0,0 +1,99 @@
package telemetryai
import (
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
type TraceColumn struct {
Alias string
// Orderable marks a column computable in the mask-pruned `matched` pass (a
// gen_ai-scoped aggregate): the only columns usable for ORDER BY and the
// aggregate filter. All-span columns (span_count, duration_nano, …) are false —
// they are output-only, since the matched pass sees only gen_ai spans.
Orderable bool
Intrinsic string // fixed SQL over intrinsic columns; used verbatim (nil Attr)
Attr *AttrAgg // field-mapper-resolved aggregate (nil Intrinsic)
}
// AttrAgg describes an aggregate the builder assembles after resolving fields:
// - Func "count" + ExistsKey -> countIf(<ExistsKey EXISTS>) (e.g. llm_call_count)
// - Func + ValueKey -> Func(<resolved value>) (e.g. sum tokens)
// - Func + ValueExpr + Scoped -> FuncIf(<value>, <gate mask>) (e.g. last_activity_time)
type AttrAgg struct {
Func string // sum | max | min | count
ValueKey *telemetrytypes.TelemetryFieldKey // attribute value to aggregate (resolved as Float64)
ValueExpr string // fixed value expr when ValueKey is nil (e.g. "timestamp")
Scoped bool // wrap as <Func>If(value, <gate mask>)
ExistsKey *telemetrytypes.TelemetryFieldKey // count spans where this key exists (countIf)
}
// ProjectionProvider decides which per-trace columns the list computes and which
// are sortable, decoupled from selection and topology.
type ProjectionProvider interface {
Columns() []TraceColumn
// DefaultOrderAlias is sorted by (desc) when the query gives no order.
DefaultOrderAlias() string
// AggregateAliases are the computed per-trace (trace-level) column names, used to
// classify which filter keys are trace-level vs span-level. Only the Orderable
// subset is actually usable in ORDER BY / the aggregate filter; the rest are
// output-only. Excludes aliases that are also real span/resource keys (service.name).
AggregateAliases() []string
}
// CommonTraceColumns are domain-neutral intrinsic columns any trace list can reuse.
// All are over-all-spans intrinsics, so none is Orderable (not computable in the
// mask-pruned matched pass) — they are output-only, computed in the enrichment scan.
func CommonTraceColumns() []TraceColumn {
return []TraceColumn{
{Alias: "start_time", Intrinsic: "min(timestamp)", Orderable: false},
{Alias: "end_time", Intrinsic: "max(timestamp)", Orderable: false},
{Alias: "duration_nano", Intrinsic: "(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp)))", Orderable: false},
{Alias: "span_count", Intrinsic: "count()", Orderable: false},
{Alias: "root_span_name", Intrinsic: "anyIf(name, parent_span_id = '')", Orderable: false},
{Alias: "service.name", Intrinsic: "any(resource_string_service$$name)", Orderable: false},
}
}
// genAIProjectionProvider adds AI/LLM per-trace metrics to the common columns.
type genAIProjectionProvider struct{}
var _ ProjectionProvider = (*genAIProjectionProvider)(nil)
func NewGenAIProjectionProvider() ProjectionProvider {
return &genAIProjectionProvider{}
}
func (genAIProjectionProvider) Columns() []TraceColumn {
// Key definitions (name/context/type) live once in GenAIFieldDefinitions, shared
// with the metadata enrichment. Copy to locals so we can take their address.
reqModel := telemetrytypes.GenAIFieldDefinitions[telemetrytypes.GenAIRequestModel]
inTok := telemetrytypes.GenAIFieldDefinitions[telemetrytypes.GenAIUsageInputTokens]
outTok := telemetrytypes.GenAIFieldDefinitions[telemetrytypes.GenAIUsageOutputTokens]
cols := CommonTraceColumns()
return append(cols,
// LLM calls only (request model present), not the full gate (tool/agent too).
TraceColumn{Alias: "llm_call_count", Orderable: true, Attr: &AttrAgg{Func: "count", ExistsKey: &reqModel}},
// tokens live only on LLM spans, so a plain sum over resolved values (NULL
// elsewhere) is correct without scoping to the mask. Aliases match the OTel
// source attributes (gen_ai.usage.input_tokens / output_tokens).
TraceColumn{Alias: "input_tokens", Orderable: true, Attr: &AttrAgg{Func: "sum", ValueKey: &inTok}},
TraceColumn{Alias: "output_tokens", Orderable: true, Attr: &AttrAgg{Func: "sum", ValueKey: &outTok}},
// timestamp of the last in-scope (gen_ai: LLM/tool/agent) span.
TraceColumn{Alias: "last_activity_time", Orderable: true, Attr: &AttrAgg{Func: "max", ValueExpr: "timestamp", Scoped: true}},
)
}
func (genAIProjectionProvider) DefaultOrderAlias() string { return "last_activity_time" }
func (genAIProjectionProvider) AggregateAliases() []string {
// every computed column except service.name (a real resource key, filterable
// at the span level). Of these, only the gen_ai-scoped ones (llm_call_count,
// input_tokens, output_tokens, last_activity_time) are Orderable and thus usable
// in ORDER BY / the aggregate filter; the rest are output-only.
return []string{
"start_time", "end_time", "duration_nano", "span_count", "root_span_name",
"llm_call_count", "input_tokens", "output_tokens", "last_activity_time",
}
}

View File

@@ -0,0 +1,722 @@
package telemetryai
import (
"context"
"fmt"
"log/slog"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/huandu/go-sqlbuilder"
)
var (
ErrUnsupportedRequestType = errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported request type for source=ai")
)
// scopedTraceStatementBuilder builds a trace list scoped to a span-selection
// category. Topology is fixed; selection (BaseConditionProvider) and columns
// (ProjectionProvider) are pluggable, so a new category is a new pair of
// providers, not new topology.
type scopedTraceStatementBuilder struct {
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
baseCond BaseConditionProvider
projection ProjectionProvider
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
}
var _ qbtypes.StatementBuilder[qbtypes.TraceAggregation] = (*scopedTraceStatementBuilder)(nil)
// NewScopedTraceStatementBuilder wires the generic trace-list builder. The trace
// builder is reused for the span-list (raw) path.
func NewScopedTraceStatementBuilder(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
fieldMapper qbtypes.FieldMapper,
conditionBuilder qbtypes.ConditionBuilder,
baseCond BaseConditionProvider,
projection ProjectionProvider,
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
) *scopedTraceStatementBuilder {
aiSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryai")
return &scopedTraceStatementBuilder{
logger: aiSettings.Logger(),
metadataStore: metadataStore,
fm: fieldMapper,
cb: conditionBuilder,
baseCond: baseCond,
projection: projection,
traceStmtBuilder: traceStmtBuilder,
}
}
// NewAITraceStatementBuilder is the scoped builder with the gen_ai gate + AI projection.
func NewAITraceStatementBuilder(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
fieldMapper qbtypes.FieldMapper,
conditionBuilder qbtypes.ConditionBuilder,
baseCond BaseConditionProvider,
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
) *scopedTraceStatementBuilder {
return NewScopedTraceStatementBuilder(settings, metadataStore, fieldMapper, conditionBuilder, baseCond, NewGenAIProjectionProvider(), traceStmtBuilder)
}
func (b *scopedTraceStatementBuilder) Build(
ctx context.Context,
start uint64,
end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
switch requestType {
case qbtypes.RequestTypeTrace:
return b.buildTraceListQuery(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), query, variables)
case qbtypes.RequestTypeRaw:
return b.buildDelegated(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).
func (b *scopedTraceStatementBuilder) buildDelegated(
ctx context.Context,
start, end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
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)
}
// 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)
}
// buildTraceListQuery is the map for the whole file. It resolves the columns, then
// wires the CTE pipeline that was benchmarked (see ai-qb-handoff.md): a single
// windowed pass picks the top-N traces, then a bucket-pruned pass enriches only those.
// The helpers appear in this file in the order they run here.
//
// RESOLVE (turn keys/columns into SQL, field-mapper aware)
// fetchKeys → metadata for the keys we reference
// resolveMask → the "is a gen_ai span" predicate (OR of EXISTS) [existsExpr]
// resolveColumns → per-trace column SQL: intrinsics + resolved aggregates
// resolveListOrders→ which resolved columns to ORDER BY
// splitFilter → span-level predicate + trace-level HAVING expression
//
// BUILD (compose the CTE pipeline)
// matched [buildMatchedCTE] ONE windowed, mask-pruned GROUP BY trace_id pass over
// │ the span index that applies the gate (+ span filter as
// │ countIf existence), the trace-level HAVING, ORDER BY and
// │ LIMIT/OFFSET in a single scan → top-N trace_ids + their
// │ gen_ai-scoped ranking metrics. No giant gate id-set.
// ▼
// ranked [buildRankedCTE] per-trace [start,end] bounds for those N traces, read
// │ from the small distributed_trace_summary table.
// ▼
// buckets [buildBucketsCTE] the exact ts_bucket_start values those N traces touch,
// │ so the enrichment scan is primary-key pruned.
// ▼
// enrichment[buildEnrichmentSelect] all per-trace columns for the N traces, scanning
// only their buckets (full trace, not window-clipped).
//
// Only gen_ai-scoped aggregates (tokens, llm activity, llm_call_count) are computable in
// the mask-pruned `matched` pass, so only those are orderable / usable in the aggregate
// filter. All-span columns (span_count, duration_nano, …) are output-only.
//
// start/end are nanoseconds.
func (b *scopedTraceStatementBuilder) buildTraceListQuery(
ctx context.Context,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
endBucket := end / querybuilder.NsToSeconds
limit := query.Limit
if limit <= 0 {
limit = 100
}
// Resolve the gate keys + columns once; every attribute access below goes through
// the field mapper (materialization/evolution aware), never a hardcoded map lookup.
keys, err := b.fetchKeys(ctx)
if err != nil {
return nil, err
}
maskExpr, maskArgs, err := b.resolveMask(ctx, start, end, keys)
if err != nil {
return nil, err
}
resolved, err := b.resolveColumns(ctx, start, end, keys, maskExpr, maskArgs)
if err != nil {
return nil, err
}
orders, err := b.resolveListOrders(query.Order, resolved)
if err != nil {
return nil, err
}
orderableSet := orderableAliasSet(resolved)
// Split the user filter: span-level predicate + trace-level HAVING expression.
fp, err := b.splitFilter(ctx, query, b.aggregateAliasSet(), orderableSet, start, end, 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, limit, query.Offset)
if err != nil {
return nil, err
}
rankedFrag, rankedArgs := b.buildRankedCTE(start, end)
bucketsFrag := buildBucketsCTE()
mainSQL, mainArgs := b.buildEnrichmentSelect(resolved, orders)
cteFragments := []string{matchedFrag, rankedFrag, bucketsFrag}
cteArgs := [][]any{matchedArgs, rankedArgs, nil}
finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL + " SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000"
finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
return &qbtypes.Statement{
Query: finalSQL,
Args: finalArgs,
Warnings: fp.warnings,
WarningsDocURL: fp.warningsURL,
}, nil
}
// ---------------------------------------------------------------------------
// RESOLVE — turn keys/columns into field-mapper-aware SQL
// ---------------------------------------------------------------------------
func (b *scopedTraceStatementBuilder) fetchKeys(ctx context.Context) (map[string][]*telemetrytypes.TelemetryFieldKey, error) {
fields := b.resolverFieldKeys()
selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(fields))
for _, k := range fields {
selectors = append(selectors, &telemetrytypes.FieldKeySelector{
Name: k.Name,
Signal: k.Signal,
FieldContext: k.FieldContext,
})
}
keys, _, err := b.metadataStore.GetKeysMulti(ctx, selectors)
return keys, err
}
func (b *scopedTraceStatementBuilder) resolverFieldKeys() []*telemetrytypes.TelemetryFieldKey {
seen := make(map[string]struct{})
var out []*telemetrytypes.TelemetryFieldKey
add := func(k *telemetrytypes.TelemetryFieldKey) {
if k == nil {
return
}
if _, dup := seen[k.Name]; dup {
return
}
seen[k.Name] = struct{}{}
out = append(out, k)
}
for _, k := range b.baseCond.FieldKeys() {
add(k)
}
for _, c := range b.projection.Columns() {
if c.Attr != nil {
add(c.Attr.ValueKey)
add(c.Attr.ExistsKey)
}
}
return out
}
// resolveMask builds the per-span in-scope mask: OR of resolved EXISTS predicates
// over the base condition's field keys.
func (b *scopedTraceStatementBuilder) resolveMask(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey) (string, []any, error) {
fieldKeys := b.baseCond.FieldKeys()
parts := make([]string, 0, len(fieldKeys))
var args []any
for _, key := range fieldKeys {
e, a, err := b.existsExpr(ctx, start, end, keys, key)
if err != nil {
return "", nil, err
}
parts = append(parts, e)
args = append(args, a...)
}
return "(" + strings.Join(parts, " OR ") + ")", args, nil
}
// existsExpr resolves a field-mapper-aware EXISTS predicate for key (materialized
// column when present, else the map). Escaped once so it round-trips when embedded
// in an outer builder.
func (b *scopedTraceStatementBuilder) existsExpr(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey, key *telemetrytypes.TelemetryFieldKey) (string, []any, error) {
resolvedKey := key
cands := keys[key.Name]
if len(cands) == 0 {
cands = []*telemetrytypes.TelemetryFieldKey{key}
} else {
resolvedKey = cands[0]
}
sb := sqlbuilder.NewSelectBuilder()
conds, _, err := b.cb.ConditionFor(ctx, start, end, resolvedKey, cands, qbtypes.FilterOperatorExists, nil, sb)
if err != nil {
return "", nil, err
}
sb.Where(conds[0])
expr, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
expr = strings.TrimPrefix(expr, "WHERE ")
return sqlbuilder.Escape(expr), args, nil
}
// resolvedColumn is a projection column whose attribute access has been resolved to
// SQL via the field mapper. expr is escaped once, ready to embed in an outer SELECT.
type resolvedColumn struct {
alias string
expr string
args []any
orderable bool
}
// resolveColumns turns the projection's declarative columns into SQL, resolving all
// attribute access through the field mapper.
func (b *scopedTraceStatementBuilder) resolveColumns(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey, maskExpr string, maskArgs []any) ([]resolvedColumn, error) {
cols := b.projection.Columns()
out := make([]resolvedColumn, 0, len(cols))
for _, c := range cols {
rc := resolvedColumn{alias: c.Alias, orderable: c.Orderable}
if c.Attr == nil {
// intrinsic: escape once (no-op unless it references a $$ column).
rc.expr = sqlbuilder.Escape(c.Intrinsic)
out = append(out, rc)
continue
}
a := c.Attr
switch {
case a.Func == "count" && a.ExistsKey != nil:
cond, cargs, err := b.existsExpr(ctx, start, end, keys, a.ExistsKey)
if err != nil {
return nil, err
}
rc.expr = fmt.Sprintf("countIf(%s)", cond)
rc.args = cargs
default:
var vexpr string
var vargs []any
if a.ValueKey != nil {
e, ar, err := querybuilder.CollisionHandledFinalExpr(ctx, start, end, a.ValueKey, b.fm, b.cb, keys, telemetrytypes.FieldDataTypeFloat64, nil, false)
if err != nil {
return nil, err
}
vexpr, vargs = e, ar
} else {
vexpr = a.ValueExpr
}
if a.Scoped {
rc.expr = fmt.Sprintf("%sIf(%s, %s)", a.Func, vexpr, maskExpr)
rc.args = append(append([]any{}, vargs...), maskArgs...)
} else {
rc.expr = fmt.Sprintf("%s(%s)", a.Func, vexpr)
rc.args = vargs
}
}
out = append(out, rc)
}
return out, nil
}
// listOrder resolves a sort key to an aggregate-column alias + direction. Both the
// matched CTE and the enrichment select that alias, so both ORDER BY it.
type listOrder struct {
alias string
direction string
}
// resolveListOrders maps order keys to the resolved orderable columns; non-orderable
// columns are rejected. Defaults to the projection's default order.
func (b *scopedTraceStatementBuilder) resolveListOrders(order []qbtypes.OrderBy, resolved []resolvedColumn) ([]listOrder, error) {
byAlias := make(map[string]resolvedColumn, len(resolved))
orderable := make([]string, 0, len(resolved))
for _, rc := range resolved {
byAlias[rc.alias] = rc
if rc.orderable {
orderable = append(orderable, rc.alias)
}
}
if len(order) == 0 {
return []listOrder{{alias: b.projection.DefaultOrderAlias(), direction: "DESC"}}, nil
}
orders := make([]listOrder, 0, len(order))
for _, o := range order {
direction := "DESC"
if o.Direction == qbtypes.OrderDirectionAsc {
direction = "ASC"
}
rc, ok := byAlias[o.Key.Name]
if !ok || !rc.orderable {
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"unsupported order key %q for the trace list; orderable keys: %s", o.Key.Name, strings.Join(orderable, ", "))
}
orders = append(orders, listOrder{alias: rc.alias, direction: direction})
}
return orders, nil
}
// filterParts is the split of the user filter into a resolved span-level predicate
// (used both to widen the matched WHERE prune and as a countIf existence in HAVING)
// and a trace-level HAVING expression.
type filterParts struct {
spanPred string
spanArgs []any
hasSpanFilter bool
havingExpr string
warnings []string
warningsURL string
}
// splitFilter partitions query.Filter into a span-level predicate and a trace-level
// HAVING expression; an explicit query.Having is ANDed onto the latter. The trace-level
// expression is validated against the aggregates computable in the matched pass.
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet, orderableSet map[string]struct{}, start, end uint64, variables map[string]qbtypes.VariableItem) (filterParts, error) {
var fp filterParts
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
if strings.TrimSpace(spanExpr) != "" {
pred, args, warnings, url, err := b.resolveSpanPredicate(ctx, start, end, spanExpr, variables)
if err != nil {
return fp, err
}
fp.spanPred, fp.spanArgs, fp.hasSpanFilter = pred, args, true
fp.warnings, fp.warningsURL = warnings, url
}
}
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
if fp.havingExpr != "" {
fp.havingExpr = fmt.Sprintf("(%s) AND (%s)", fp.havingExpr, query.Having.Expression)
} else {
fp.havingExpr = query.Having.Expression
}
}
if err := validateAggregateFilter(fp.havingExpr, orderableSet); err != nil {
return fp, err
}
return fp, nil
}
// resolveSpanPredicate resolves a span-level filter expression to a bare boolean SQL
// predicate (escaped) + args via the field mapper.
func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context, start, end uint64, expr string, variables map[string]qbtypes.VariableItem) (string, []any, []string, string, error) {
selectors := querybuilder.QueryStringToKeysSelectors(expr)
for i := range selectors {
selectors[i].Signal = telemetrytypes.SignalTraces
}
keys, _, err := b.metadataStore.GetKeysMulti(ctx, selectors)
if err != nil {
return "", nil, nil, "", err
}
prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
Context: ctx,
Logger: b.logger,
FieldMapper: b.fm,
ConditionBuilder: b.cb,
FieldKeys: keys,
Variables: variables,
StartNs: start,
EndNs: end,
})
if err != nil {
return "", nil, nil, "", err
}
if prepared.IsEmpty() {
return "", nil, nil, "", nil
}
sb := sqlbuilder.NewSelectBuilder()
sb.Select("1")
sb.AddWhereClause(prepared.WhereClause)
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
pred := sql[strings.Index(sql, "WHERE ")+len("WHERE "):]
return sqlbuilder.Escape(pred), args, prepared.Warnings, prepared.WarningsDocURL, nil
}
// ---------------------------------------------------------------------------
// BUILD — compose the CTE pipeline (matched → ranked → buckets → enrichment)
// ---------------------------------------------------------------------------
// buildMatchedCTE builds `matched`: a single windowed GROUP BY trace_id pass that
// picks the top-N traces. The WHERE prunes to gen_ai spans (widened with the span-level
// predicate when present); HAVING enforces the gate (countIf(mask) > 0), the span-level
// filter as an existence check, and the trace-level aggregate filter; ORDER BY + LIMIT
// select the winners. Only gen_ai-scoped aggregates are computable here, and only those
// actually referenced by ORDER BY / the aggregate filter are selected (the rest are
// computed once, later, in the enrichment scan) — this keeps the hot ranking pass lean.
func (b *scopedTraceStatementBuilder) buildMatchedCTE(start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, orderableSet map[string]struct{}, maskExpr string, maskArgs []any, fp filterParts, 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)
selects := []string{"trace_id"}
for _, rc := range resolved {
if _, ok := needed[rc.alias]; !ok {
continue
}
selects = append(selects, embedExpr(sb, rc.expr, rc.args)+" AS "+quoteAlias(rc.alias))
}
sb.Select(selects...)
sb.From(spanTable())
// WHERE: window + coarse prune to gen_ai spans (widened so span-filter spans are
// visible for the countIf existence check below).
win := windowWhere(sb, start, end, startBucket, endBucket)
prune := "(" + embedExpr(sb, maskExpr, maskArgs)
if fp.hasSpanFilter {
prune += " OR " + embedExpr(sb, fp.spanPred, fp.spanArgs)
}
prune += ")"
sb.Where(append(win, prune)...)
sb.GroupBy("trace_id")
// HAVING: the gate + span-existence checks are only needed once the WHERE has been
// widened by a span filter; otherwise WHERE = mask already enforces the gate.
var having []string
if fp.hasSpanFilter {
having = append(having, "countIf("+embedExpr(sb, maskExpr, maskArgs)+") > 0")
having = append(having, "countIf("+embedExpr(sb, fp.spanPred, fp.spanArgs)+") > 0")
}
if strings.TrimSpace(fp.havingExpr) != "" {
hv, err := b.buildHaving(fp.havingExpr, orderableSet)
if err != nil {
return "", nil, err
}
if hv != "" {
having = append(having, hv)
}
}
if len(having) > 0 {
sb.Having(strings.Join(having, " AND "))
}
sb.OrderBy(orderClause(orders)...)
sb.Limit(limit)
if offset > 0 {
sb.Offset(offset)
}
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return fmt.Sprintf("matched AS (%s)", sql), args, nil
}
// buildRankedCTE builds `ranked`: per-trace [start,end] bounds for the matched traces,
// read from the small trace-summary table (used to derive the bucket prune).
func (b *scopedTraceStatementBuilder) buildRankedCTE(start, end uint64) (string, []any) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select("trace_id", "min(start) AS t_start", "max(end) AS t_end")
sb.From(summaryTable())
sb.Where(
"trace_id GLOBAL IN (SELECT trace_id FROM matched)",
"end >= fromUnixTimestamp64Nano("+sb.Var(start)+")",
"start < fromUnixTimestamp64Nano("+sb.Var(end)+")",
)
sb.GroupBy("trace_id")
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return fmt.Sprintf("ranked AS (%s)", sql), args
}
// buildBucketsCTE builds `buckets`: the exact ts_bucket_start values the matched traces
// span, so the enrichment scan is pruned to those primary-key buckets. No args.
func buildBucketsCTE() string {
adj := querybuilder.BucketAdjustment // 30-min bucket width in seconds
return fmt.Sprintf("buckets AS (SELECT DISTINCT b AS ts_bucket FROM ranked "+
"ARRAY JOIN range("+
"toUInt64(intDiv(toUnixTimestamp(t_start), %d) * %d - %d), "+
"toUInt64(intDiv(toUnixTimestamp(t_end), %d) * %d + %d), "+
"%d) AS b)", adj, adj, adj, adj, adj, adj, adj)
}
// buildEnrichmentSelect builds the final SELECT: all per-trace columns for the matched
// traces, scanning only their buckets (full trace, not window-clipped). SELECT-expr
// args lead; the WHERE / ORDER BY carry none.
func (b *scopedTraceStatementBuilder) buildEnrichmentSelect(resolved []resolvedColumn, orders []listOrder) (string, []any) {
sb := sqlbuilder.NewSelectBuilder()
selects, selectArgs := selectAllColumns(resolved)
sb.Select(selects...)
sb.From(spanTable())
sb.Where(
"ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)",
"trace_id GLOBAL IN (SELECT trace_id FROM ranked)",
)
sb.GroupBy("trace_id")
sb.OrderBy(orderClause(orders)...)
sql, builtArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return sql, append(append([]any{}, selectArgs...), builtArgs...)
}
// buildHaving rewrites a trace-level HAVING expression against the aggregate column
// aliases computable in the matched pass; bare, trace., and tracefield. forms all map
// to the selected 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
// ---------------------------------------------------------------------------
// spanTable is the fully-qualified span index table.
func spanTable() string {
return fmt.Sprintf("%s.%s", telemetrytraces.DBName, telemetrytraces.SpanIndexV3TableName)
}
// summaryTable is the fully-qualified trace-summary table.
func summaryTable() string {
return fmt.Sprintf("%s.%s", telemetrytraces.DBName, telemetrytraces.TraceSummaryTableName)
}
// aggregateAliasSet is the set of all trace-level (computed) column aliases, used to
// classify which filter keys are trace-level vs span-level.
func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
set := make(map[string]struct{}, len(b.projection.AggregateAliases()))
for _, a := range b.projection.AggregateAliases() {
set[a] = struct{}{}
}
return set
}
// orderableAliasSet is the subset of aggregate aliases computable in the matched pass
// (gen_ai-scoped): the only ones usable for 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 set of aggregate aliases the matched pass must
// select: those referenced by ORDER BY plus those referenced in the aggregate HAVING
// (bare / trace. / tracefield. forms). Anything else is left to the enrichment scan.
func neededMatchedAliases(orders []listOrder, havingExpr string, orderableSet map[string]struct{}) 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{}{}
}
}
return needed
}
// validateAggregateFilter rejects a trace-level filter that references an aggregate not
// computable in the matched pass (e.g. span_count, duration_nano), with a clear message.
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)
}
for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
name := strings.TrimPrefix(strings.TrimPrefix(sel.Name, "trace."), "tracefield.")
if _, ok := orderableSet[name]; !ok {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregate %q cannot be used in an AI trace-list filter; filterable aggregates: %s", name, strings.Join(allowed, ", "))
}
}
return nil
}
// embedExpr inlines a resolved (escaped) expr carrying `?` placeholders into sb by
// replacing each `?` with a builder Var, so go-sqlbuilder tracks the args in appearance
// order and un-escapes the expr at Build time.
func embedExpr(sb *sqlbuilder.SelectBuilder, expr string, args []any) string {
var out strings.Builder
ai := 0
for i := 0; i < len(expr); i++ {
if expr[i] == '?' && ai < len(args) {
out.WriteString(sb.Var(args[ai]))
ai++
continue
}
out.WriteByte(expr[i])
}
return out.String()
}
// windowWhere binds the shared time-window predicates to sb and returns them, so a
// caller can add its own predicate in the same Where call.
func windowWhere(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64) []string {
return []string{
sb.GE("timestamp", fmt.Sprintf("%d", start)),
sb.L("timestamp", fmt.Sprintf("%d", end)),
sb.GE("ts_bucket_start", startBucket),
sb.LE("ts_bucket_start", endBucket),
}
}
// orderClause renders the ORDER BY terms (by column alias) + the trace_id tiebreak.
func orderClause(orders []listOrder) []string {
out := make([]string, 0, len(orders)+1)
for _, o := range orders {
out = append(out, fmt.Sprintf("%s %s", quoteAlias(o.alias), o.direction))
}
return append(out, "trace_id DESC")
}
// selectAllColumns renders `expr AS alias` for every resolved column and returns their
// field-mapper args in select order.
func selectAllColumns(resolved []resolvedColumn) ([]string, []any) {
selects := []string{"trace_id"}
var args []any
for _, rc := range resolved {
selects = append(selects, rc.expr+" AS "+quoteAlias(rc.alias))
args = append(args, rc.args...)
}
return selects, args
}
// quoteAlias backticks an alias that carries characters special to the SQL builder.
func quoteAlias(alias string) string {
if strings.ContainsAny(alias, ".$`") {
return "`" + alias + "`"
}
return alias
}

View File

@@ -0,0 +1,700 @@
package telemetryai
import (
"context"
"fmt"
"strings"
"testing"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/stretchr/testify/require"
)
// otelKeysMap seeds the OpenTelemetry gen_ai semantic-convention keys the AI
// queries reference, so the metadata-backed field resolution succeeds in tests.
func otelKeysMap() map[string][]*telemetrytypes.TelemetryFieldKey {
strKey := func(name string) *telemetrytypes.TelemetryFieldKey {
return &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
}
}
numKey := func(name string) *telemetrytypes.TelemetryFieldKey {
return &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeFloat64,
}
}
return map[string][]*telemetrytypes.TelemetryFieldKey{
"gen_ai.request.model": {strKey("gen_ai.request.model")},
"gen_ai.tool.name": {strKey("gen_ai.tool.name")},
"gen_ai.agent.name": {strKey("gen_ai.agent.name")},
"gen_ai.user.id": {strKey("gen_ai.user.id")},
"gen_ai.usage.input_tokens": {numKey("gen_ai.usage.input_tokens")},
"gen_ai.usage.output_tokens": {numKey("gen_ai.usage.output_tokens")},
"gen_ai.usage.cost": {numKey("gen_ai.usage.cost")},
"gen_ai.usage.cached_input_tokens": {numKey("gen_ai.usage.cached_input_tokens")},
"has_error": {{
Name: "has_error",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextSpan,
FieldDataType: telemetrytypes.FieldDataTypeBool,
}},
}
}
// standard test window (ms), matching the traces builder tests.
const (
testStartMs = uint64(1747947419000)
testEndMs = uint64(1747983448000)
)
func newTestBuilder(t *testing.T) *scopedTraceStatementBuilder {
return newTestBuilderWithKeys(t, otelKeysMap())
}
// newTestBuilderWithKeys mirrors the production wiring in signozquerier's provider.
// The gen_ai keys are seeded via keysMap here; in production the metadata store
// surfaces them itself (enrichWithGenAIKeys).
func newTestBuilderWithKeys(t *testing.T, keysMap map[string][]*telemetrytypes.TelemetryFieldKey) *scopedTraceStatementBuilder {
t.Helper()
settings := instrumentationtest.New().ToProviderSettings()
fm := telemetrytraces.NewFieldMapper()
cb := telemetrytraces.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = keysMap
fl := flaggertest.New(t)
baseCond := NewGenAIBaseConditionProvider()
// In production the metadata store enriches gen_ai keys (enrichWithGenAIKeys);
// here the mock is seeded directly via keysMap.
metadataStore := telemetrytypes.MetadataStore(mockMetadataStore)
rewriter := querybuilder.NewAggExprRewriter(settings, nil, fm, cb, nil, fl)
traceStmtBuilder := telemetrytraces.NewTraceQueryStatementBuilder(
settings,
metadataStore,
fm,
cb,
rewriter,
nil,
fl,
false,
100000,
)
return NewAITraceStatementBuilder(
settings,
metadataStore,
fm,
cb,
baseCond,
traceStmtBuilder,
)
}
// ---------------------------------------------------------------------------
// Full-query golden tests
//
// Each pins the WHOLE generated statement, with bound args inlined into the `?`
// placeholders, as ONE self-contained literal — so a failure diff shows the entire
// query and the expected SQL can be copied straight into a ClickHouse client. The
// `want` strings are formatted for readability; the comparison is whitespace- and
// backtick-insensitive (see normalizeSQL), so only the SQL tokens themselves matter.
//
// The four trace-list goldens cover the corners of how `matched` is assembled —
// {no span filter, span filter} × {no aggregate filter, aggregate filter} — plus a
// mixed filter + multi-key order, plus the delegated span list. Note `matched` selects
// only the aggregates ORDER BY / HAVING reference; the rest appear only in enrichment.
//
// Run `go test ./pkg/telemetryai/ -run TestBuild_FullSQL -v` to also print each query.
// ---------------------------------------------------------------------------
// renderSQL substitutes bound args into the `?` placeholders so the whole statement
// reads as one literal SQL string.
func renderSQL(t *testing.T, stmt *qbtypes.Statement) string {
t.Helper()
var b strings.Builder
argi := 0
for i := 0; i < len(stmt.Query); i++ {
if stmt.Query[i] == '?' {
require.Less(t, argi, len(stmt.Args), "more ? than args in query")
b.WriteString(formatArg(stmt.Args[argi]))
argi++
continue
}
b.WriteByte(stmt.Query[i])
}
require.Equal(t, len(stmt.Args), argi, "arg count does not match number of placeholders")
return b.String()
}
func formatArg(a any) string {
if s, ok := a.(string); ok {
return "'" + s + "'"
}
return fmt.Sprintf("%v", a)
}
// normalizeSQL makes the comparison insensitive to formatting: it drops identifier
// backticks, collapses whitespace runs to a single space, and removes spaces directly
// inside parentheses. This lets the golden strings be freely indented/wrapped (and
// written as Go raw literals, which cannot contain backticks) — only the SQL tokens
// and their order matter.
func normalizeSQL(s string) string {
s = strings.Join(strings.Fields(strings.ReplaceAll(s, "`", "")), " ")
s = strings.ReplaceAll(s, "( ", "(")
s = strings.ReplaceAll(s, " )", ")")
return s
}
func requireSQLEqual(t *testing.T, want string, stmt *qbtypes.Statement) {
t.Helper()
got := renderSQL(t, stmt)
t.Logf("\n%s", got)
require.Equal(t, normalizeSQL(want), normalizeSQL(got))
}
// No filter: matched selects only the default order key (last_activity_time), WHERE is
// just window + gate mask, no HAVING.
func TestBuild_FullSQL_TraceList_NoFilter(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI, Limit: 20,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
LIMIT 20
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Span-level AND trace-level filter, order by the aggregate, pagination. matched selects
// only output_tokens (the sole aggregate referenced by both ORDER BY and HAVING) — not
// input_tokens/llm_call_count/last_activity_time. The span predicate widens the WHERE
// prune and becomes a countIf(...) > 0 existence check alongside the gate countIf.
func TestBuild_FullSQL_TraceList_SpanAndTraceFilter(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND output_tokens > 1000"},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "output_tokens"}}, Direction: qbtypes.OrderDirectionDesc}},
Limit: 10, Offset: 30,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
OR (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model') = true))
GROUP BY trace_id
HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
AND countIf((attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
AND output_tokens > 1000
ORDER BY output_tokens DESC, trace_id DESC
LIMIT 10 OFFSET 30
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY output_tokens DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Aggregate-only filter (no span filter). WHERE prune is NOT widened, there is no
// gate/span countIf, just the aggregate HAVING. `trace.output_tokens` rewrites to the
// output_tokens alias. matched selects output_tokens (HAVING) + last_activity_time (default order).
func TestBuild_FullSQL_TraceList_AggregateFilterOnly(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
Limit: 20,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
GROUP BY trace_id
HAVING output_tokens > 1000
ORDER BY last_activity_time DESC, trace_id DESC
LIMIT 20
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Span-only filter (no aggregate filter). WHERE is widened; HAVING has the gate + span
// countIf pair but no trailing aggregate. `has_error = true` resolves to a
// materialized-column predicate (not a map access). matched selects only the default order key.
func TestBuild_FullSQL_TraceList_SpanFilterOnly(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "has_error = true"},
Limit: 20,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
OR has_error = true)
GROUP BY trace_id
HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
AND countIf(has_error = true) > 0
ORDER BY last_activity_time DESC, trace_id DESC
LIMIT 20
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY last_activity_time DESC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Mixed filter (two span predicates AND'd into one existence check + an aggregate) with
// a two-key order on different aggregates than the filter. matched selects input_tokens
// + last_activity_time (ORDER BY) and output_tokens (HAVING) — three of four; llm_call_count is not.
func TestBuild_FullSQL_TraceList_MixedFiltersMultiOrder(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o' AND has_error = true AND output_tokens > 500"},
Order: []qbtypes.OrderBy{
{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "input_tokens"}}, Direction: qbtypes.OrderDirectionDesc},
{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "last_activity_time"}}, Direction: qbtypes.OrderDirectionAsc},
},
Limit: 15,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
WITH matched AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
OR ((attributes_string['gen_ai.request.model'] = 'gpt-4o' AND mapContains(attributes_string, 'gen_ai.request.model') = true) AND has_error = true))
GROUP BY trace_id
HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
AND countIf(((attributes_string['gen_ai.request.model'] = 'gpt-4o' AND mapContains(attributes_string, 'gen_ai.request.model') = true) AND has_error = true)) > 0
AND output_tokens > 500
ORDER BY input_tokens DESC, last_activity_time ASC, trace_id DESC
LIMIT 15
),
ranked AS (
SELECT trace_id, min(start) AS t_start, max(end) AS t_end
FROM signoz_traces.distributed_trace_summary
WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
AND end >= fromUnixTimestamp64Nano(1747947419000000000)
AND start < fromUnixTimestamp64Nano(1747983448000000000)
GROUP BY trace_id
),
buckets AS (
SELECT DISTINCT b AS ts_bucket
FROM ranked
ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
)
SELECT trace_id,
min(timestamp) AS start_time,
max(timestamp) AS end_time,
(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
count() AS span_count,
anyIf(name, parent_span_id = '') AS root_span_name,
any(resource_string_service$$name) AS service.name,
countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
FROM signoz_traces.distributed_signoz_index_v3
WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
GROUP BY trace_id
ORDER BY input_tokens DESC, last_activity_time ASC, trace_id DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Span list (requestType raw): delegated to the traces builder with the gate ANDed
// into the user filter, so only gen_ai spans matching the filter come back. Standard
// span columns, single SELECT (no CTE pipeline).
func TestBuild_FullSQL_SpanList_Raw(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini'"},
Limit: 10,
}, nil)
require.NoError(t, err)
requireSQLEqual(t, `
SELECT timestamp AS timestamp, trace_id AS trace_id, span_id AS span_id,
trace_state AS trace_state, parent_span_id AS parent_span_id, flags AS flags,
name AS name, kind AS kind, kind_string AS kind_string, duration_nano AS duration_nano,
status_code AS status_code, status_message AS status_message,
status_code_string AS status_code_string, events AS events, links AS links,
response_status_code AS response_status_code, external_http_url AS external_http_url,
http_url AS http_url, external_http_method AS external_http_method,
http_method AS http_method, http_host AS http_host, db_name AS db_name,
db_operation AS db_operation, has_error AS has_error, is_remote AS is_remote,
attributes_string, attributes_number, attributes_bool, resources_string
FROM signoz_traces.distributed_signoz_index_v3
WHERE (((mapContains(attributes_string, 'gen_ai.request.model') = true
OR mapContains(attributes_string, 'gen_ai.tool.name') = true
OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
AND ((attributes_string['gen_ai.request.model'] = 'gpt-4o-mini'
AND mapContains(attributes_string, 'gen_ai.request.model') = true)))
AND timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
LIMIT 10
`, stmt)
}
// ---------------------------------------------------------------------------
// Behavior / branch tests not covered by the goldens above
// ---------------------------------------------------------------------------
// Trace-level and span-level predicates may not be OR-combined.
func TestBuild_TraceList_TraceOrSpanMixRejected(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000 OR gen_ai.request.model = 'x'"},
Limit: 10,
}
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "cannot be combined")
}
// An output-only aggregate (span_count / duration_nano) can be displayed but not used
// in the aggregate filter or ORDER BY — it is not computable in the matched pass.
func TestBuild_TraceList_OutputOnlyAggregateRejected(t *testing.T) {
b := newTestBuilder(t)
// filter by span_count -> rejected
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: "span_count > 3"},
}, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "span_count")
// order by duration_nano -> rejected
_, err = b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "duration_nano"}}, Direction: qbtypes.OrderDirectionDesc}},
}, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "unsupported order key")
}
// A HAVING referencing a non-aggregate column is rejected.
func TestBuild_TraceList_Having_UnknownColumn(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
Having: &qbtypes.Having{Expression: "service.name > 1"}, // not an aggregate column
Limit: 10,
}
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
require.Error(t, err)
}
// Ordering by an unknown key is rejected.
func TestBuild_TraceList_UnsupportedOrderKey(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
Order: []qbtypes.OrderBy{
{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "http.request.method"}}, Direction: qbtypes.OrderDirectionDesc},
},
}
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
require.Error(t, err)
require.Contains(t, err.Error(), "unsupported order key")
}
// With no limit set, the builder applies the default of 100.
func TestBuild_TraceList_DefaultLimit(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
}
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
require.NoError(t, err)
require.Contains(t, stmt.Query, "LIMIT ?")
require.Contains(t, stmt.Args, 100)
}
// Only trace list and span list (raw) are supported; distribution is not.
func TestBuild_UnsupportedRequestType(t *testing.T) {
b := newTestBuilder(t)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Source: telemetrytypes.SourceAI,
Aggregations: []qbtypes.TraceAggregation{
{Expression: "count()"},
},
}
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeDistribution, query, nil)
require.ErrorIs(t, err, ErrUnsupportedRequestType)
}
// ---------------------------------------------------------------------------
// Filter operator resolution
//
// The goldens pin the CTE structure; these pin how each filter OPERATOR resolves into
// SQL (the part that varies with the operator, not the pipeline). Span-level operators
// resolve to a predicate that appears both in the widened WHERE prune and as a
// countIf(...) > 0 existence check; aggregate operators become a HAVING (values inlined).
// ---------------------------------------------------------------------------
func TestBuild_TraceList_SpanFilterOperatorResolution(t *testing.T) {
b := newTestBuilder(t)
build := func(t *testing.T, expr string) string {
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: expr}, Limit: 5,
}, nil)
require.NoError(t, err)
return stmt.Query
}
cases := []struct{ name, expr, frag string }{
{"not_equal", "gen_ai.request.model != 'gpt-4o'",
"attributes_string['gen_ai.request.model'] <> ?"},
{"in", "gen_ai.request.model IN ('gpt-4o', 'gpt-4')",
"(attributes_string['gen_ai.request.model'] = ? OR attributes_string['gen_ai.request.model'] = ?) AND mapContains(attributes_string, 'gen_ai.request.model') = ?"},
{"exists", "gen_ai.user.id EXISTS", // non-gate key, so the fragment is unambiguous
"mapContains(attributes_string, 'gen_ai.user.id') = ?"},
{"not_exists", "gen_ai.user.id NOT EXISTS",
"mapContains(attributes_string, 'gen_ai.user.id') <> ?"},
{"contains", "gen_ai.request.model CONTAINS 'gpt'", // case-insensitive
"LOWER(attributes_string['gen_ai.request.model']) LIKE LOWER(?)"},
{"like", "gen_ai.request.model LIKE 'gpt%'",
"attributes_string['gen_ai.request.model'] LIKE ?"},
{"numeric_gte", "gen_ai.usage.output_tokens >= 100", // span attr (Float64), distinct from the output_tokens aggregate
"toFloat64(attributes_number['gen_ai.usage.output_tokens']) >= ? AND mapContains(attributes_number, 'gen_ai.usage.output_tokens') = ?"},
{"not_group", "NOT (gen_ai.request.model = 'gpt-4o')",
"NOT (((attributes_string['gen_ai.request.model'] = ? AND mapContains(attributes_string, 'gen_ai.request.model') = ?)))"},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
// the resolved predicate appears in the widened WHERE prune and (wrapped) in
// the countIf existence check; the goldens pin the countIf structure, here we
// pin only the operator's resolution.
require.Contains(t, build(t, c.expr), c.frag)
})
}
}
func TestBuild_TraceList_AggregateFilterOperatorResolution(t *testing.T) {
b := newTestBuilder(t)
build := func(t *testing.T, expr string) (string, error) {
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
Filter: &qbtypes.Filter{Expression: expr}, Limit: 5,
}, nil)
if err != nil {
return "", err
}
return stmt.Query, nil
}
// values are inlined by the HAVING rewriter (not parameterized).
cases := []struct{ name, expr, having string }{
{"less_than", "output_tokens < 500", "HAVING output_tokens < 500"},
{"not_equal", "output_tokens != 0", "HAVING output_tokens != 0"},
{"range_and", "output_tokens >= 500 AND output_tokens <= 1000", "HAVING (output_tokens >= 500 AND output_tokens <= 1000)"},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
q, err := build(t, c.expr)
require.NoError(t, err)
require.Contains(t, q, c.having)
})
}
// BETWEEN is not supported by the HAVING rewriter — surfaced as an error, not silently wrong.
t.Run("between_unsupported", func(t *testing.T) {
_, err := build(t, "output_tokens BETWEEN 500 AND 1000")
require.Error(t, err)
})
}

View File

@@ -171,6 +171,42 @@ func (c *conditionBuilder) conditionFor(
}
func (c *conditionBuilder) ConditionFor(
ctx context.Context,
startNs uint64,
endNs uint64,
key *telemetrytypes.TelemetryFieldKey,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
// has/hasAny/hasAll/hasToken are logs-body-only; reject for audit.
if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
return nil, nil, err
}
keys, warning := querybuilder.ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
if len(keys) == 0 {
return nil, warnings, querybuilder.NewKeyNotFoundError(key.Name)
}
conds := make([]string, 0, len(keys))
for _, k := range keys {
cond, err := c.conditionForKey(ctx, startNs, endNs, k, operator, value, sb)
if err != nil {
return nil, nil, err
}
conds = append(conds, cond)
}
return conds, warnings, nil
}
func (c *conditionBuilder) conditionForKey(
ctx context.Context,
startNs uint64,
endNs uint64,

View File

@@ -49,7 +49,6 @@ func NewAuditQueryStatementBuilder(
telemetrytypes.SourceAudit,
metadataStore,
fullTextColumn,
jsonKeyToKey,
flagger,
)
@@ -551,7 +550,6 @@ func (b *auditQueryStatementBuilder) addFilterCondition(
FieldKeys: keys,
SkipResourceFilter: true,
FullTextColumn: b.fullTextColumn,
JsonKeyToKey: b.jsonKeyToKey,
Variables: variables,
StartNs: start,
EndNs: end,

View File

@@ -25,6 +25,97 @@ func NewConditionBuilder(fm qbtypes.FieldMapper, fl flagger.Flagger) *conditionB
return &conditionBuilder{fm: fm, fl: fl}
}
// isBodyJSONSearch reports whether a key addresses a path within the body JSON. Only
// an explicit Body context qualifies; a bare, context-less `body` (e.g. full-text
// `count_distinct(body)` or `body EXISTS`) is a full-text match, not a `$.body` path.
func isBodyJSONSearch(key *telemetrytypes.TelemetryFieldKey, columns []*schema.Column) bool {
if key.FieldContext != telemetrytypes.FieldContextBody {
return false
}
for _, column := range columns {
if column.Name == LogsV2BodyColumn || column.Name == LogsV2BodyV2Column {
return true
}
}
return false
}
// conditionForArrayFunction builds `has/hasAny/hasAll(<arrayFieldExpr>, value)` over a
// body JSON array field. The field expression uses the JSON accessor (flag on) or
// legacy string extraction (flag off); value[0] is the needle.
func (c *conditionBuilder) conditionForArrayFunction(
ctx context.Context,
startNs, endNs uint64,
key *telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
value any,
columns []*schema.Column,
sb *sqlbuilder.SelectBuilder,
) (string, error) {
if !isBodyJSONSearch(key, columns) {
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
"function `%s` supports only body JSON search", operator.FunctionName()).WithUrl(functionBodyJSONSearchDocURL)
}
needle := value
if args, ok := value.([]any); ok && len(args) > 0 {
needle = args[0]
}
var fieldExpr string
if c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
fe, err := c.fm.FieldFor(ctx, startNs, endNs, key)
if err != nil {
return "", err
}
fieldExpr = fe
} else {
// legacy string-body path; value drives array-type inference (e.g. `[*]` paths)
fieldExpr, _ = GetBodyJSONKey(ctx, key, qbtypes.FilterOperatorUnknown, value)
}
return fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), fieldExpr, sb.Var(needle)), nil
}
// conditionForHasToken builds `hasToken(LOWER(<bodyColumn>), LOWER(<needle>))`, a
// full-text token search over the body column. It resolves the column from the key
// name + use_json_body flag, validates the field/value, and tags errors with the doc URL.
func (c *conditionBuilder) conditionForHasToken(
ctx context.Context,
key *telemetrytypes.TelemetryFieldKey,
value any,
sb *sqlbuilder.SelectBuilder,
) (string, error) {
// hasToken takes a single needle; unwrap it from the function-argument slice.
needle := value
if args, ok := value.([]any); ok && len(args) > 0 {
needle = args[0]
}
// TODO(Tushar): thread orgID here to evaluate correctly
bodyJSONEnabled := c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
columnName := LogsV2BodyColumn
if bodyJSONEnabled {
if key.Name != LogsV2BodyColumn && key.Name != bodyMessageField {
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
"function `hasToken` only supports body/body.message field as first parameter").WithUrl(hasTokenFunctionDocURL)
}
columnName = bodyMessageField
} else if key.Name != LogsV2BodyColumn {
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
"function `hasToken` only supports body field as first parameter").WithUrl(hasTokenFunctionDocURL)
}
// hasToken matches string tokens only.
if _, ok := needle.(string); !ok {
return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
"function `hasToken` expects value parameter to be a string").WithUrl(hasTokenFunctionDocURL)
}
return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", columnName, sb.Var(needle)), nil
}
func (c *conditionBuilder) conditionFor(
ctx context.Context,
startNs, endNs uint64,
@@ -33,15 +124,27 @@ func (c *conditionBuilder) conditionFor(
value any,
sb *sqlbuilder.SelectBuilder,
) (string, error) {
// hasToken is a token search over the body column resolved purely from the key
// name + flag, independent of column resolution, so handle it before anything else.
if operator == qbtypes.FilterOperatorHasToken {
return c.conditionForHasToken(ctx, key, value, sb)
}
columns, err := c.fm.ColumnFor(ctx, startNs, endNs, key)
if err != nil {
return "", err
}
// has/hasAny/hasAll build `has(<arrayFieldExpr>, value)` over body JSON arrays
// rather than going through the normal operator paths, so handle them up front.
if operator.IsArrayFunctionOperator() {
return c.conditionForArrayFunction(ctx, startNs, endNs, key, operator, value, columns, sb)
}
// TODO(Piyush): Update this to support multiple JSON columns based on evolutions
for _, column := range columns {
// TODO(Tushar): thread orgID here to evaluate correctly
if column.Type.GetType() == schema.ColumnTypeEnumJSON && key.FieldContext == telemetrytypes.FieldContextBody && c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) && key.Name != messageSubField {
if column.Type.GetType() == schema.ColumnTypeEnumJSON && isBodyJSONSearch(key, columns) && c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) && key.Name != messageSubField {
valueType, value := InferDataType(value, operator, key)
cond, err := NewJSONConditionBuilder(key, valueType).buildJSONCondition(operator, value, sb)
if err != nil {
@@ -60,9 +163,9 @@ func (c *conditionBuilder) conditionFor(
return "", err
}
// Check if this is a body JSON search - either by FieldContext
// Check if this is a body JSON search (legacy string-body path, JSON flag off).
// TODO(Tushar): thread orgID here to evaluate correctly
if key.FieldContext == telemetrytypes.FieldContextBody && !c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
if isBodyJSONSearch(key, columns) && !c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
fieldExpression, value = GetBodyJSONKey(ctx, key, operator, value)
}
@@ -174,7 +277,7 @@ func (c *conditionBuilder) conditionFor(
// key membership checks, so depending on the column type, the condition changes
case qbtypes.FilterOperatorExists, qbtypes.FilterOperatorNotExists:
// TODO(Tushar): thread orgID here to evaluate correctly
if key.FieldContext == telemetrytypes.FieldContextBody && !c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
if isBodyJSONSearch(key, columns) && !c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
if operator == qbtypes.FilterOperatorExists {
return GetBodyJSONKeyForExists(ctx, key, operator, value), nil
} else {
@@ -268,6 +371,82 @@ func (c *conditionBuilder) conditionFor(
}
func (c *conditionBuilder) ConditionFor(
ctx context.Context,
startNs uint64,
endNs uint64,
key *telemetrytypes.TelemetryFieldKey,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
keys, warning := querybuilder.ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
if len(keys) == 0 {
// No known field key matched. Legacy string-body mode still searches unknown
// Body-context keys as body JSON paths; JSON-body mode requires a metadata match.
// TODO(Tushar): thread orgID here to evaluate correctly
switch {
case key.FieldContext == telemetrytypes.FieldContextBody && key.Name == "":
return nil, warnings, errors.NewInvalidInputf(errors.CodeInvalidInput, "missing key for body json search - expected key of the form `body.key` (ex: `body.status`)")
case key.FieldContext == telemetrytypes.FieldContextBody &&
!c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})):
keys = []*telemetrytypes.TelemetryFieldKey{key}
default:
return nil, warnings, querybuilder.NewKeyNotFoundError(key.Name)
}
}
// has/hasAny/hasAll need an array field: in JSON-body mode drop non-array matches so a
// scalar errors clearly instead of failing at ClickHouse runtime (legacy mode skips this).
if operator.IsArrayFunctionOperator() &&
c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
arrayKeys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(keys))
for _, k := range keys {
if k.FieldDataType.IsArray() {
arrayKeys = append(arrayKeys, k)
}
}
if len(arrayKeys) == 0 {
return nil, warnings, errors.NewInvalidInputf(errors.CodeInvalidInput,
"function `%s` expects key parameter to be an array field; no array fields found", operator.FunctionName())
}
keys = arrayKeys
}
conds := make([]string, 0, len(keys))
for _, k := range keys {
cond, err := c.conditionForKey(ctx, startNs, endNs, k, operator, value, sb)
if err != nil {
return nil, nil, err
}
conds = append(conds, cond)
if w := c.bodyFullTextDefaultWarning(ctx, startNs, endNs, k, operator); w != "" {
warnings = append(warnings, w)
}
}
return conds, warnings, nil
}
// bodyFullTextDefaultWarning returns the advisory shown when a regexp full-text
// search on `body` resolves to the body.message sub-field (JSON mode), else "". This
// keeps the JSON-vs-legacy decision in the builder rather than the filter visitor.
func (c *conditionBuilder) bodyFullTextDefaultWarning(ctx context.Context, startNs, endNs uint64, key *telemetrytypes.TelemetryFieldKey, operator qbtypes.FilterOperator) string {
if operator != qbtypes.FilterOperatorRegexp || key.Name != LogsV2BodyColumn {
return ""
}
if field, err := c.fm.FieldFor(ctx, startNs, endNs, key); err == nil && field == messageSubColumn {
return querybuilder.BodyFullTextSearchDefaultWarning
}
return ""
}
func (c *conditionBuilder) conditionForKey(
ctx context.Context,
startNs uint64,
endNs uint64,

View File

@@ -131,8 +131,8 @@ func TestExistsConditionForWithEvolutions(t *testing.T) {
for _, tc := range testCases {
sb := sqlbuilder.NewSelectBuilder()
t.Run(tc.name, func(t *testing.T) {
cond, err := conditionBuilder.ConditionFor(ctx, tc.startTs, tc.endTs, &tc.key, tc.operator, tc.value, sb)
sb.Where(cond)
cond, _, err := conditionBuilder.ConditionFor(ctx, tc.startTs, tc.endTs, &tc.key, []*telemetrytypes.TelemetryFieldKey{&tc.key}, tc.operator, tc.value, sb)
sb.Where(cond...)
if tc.expectedError != nil {
assert.Equal(t, tc.expectedError, err)
@@ -522,8 +522,8 @@ func TestConditionFor(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
t.Run(tc.name, func(t *testing.T) {
tc.key.Evolutions = tc.evolutions
cond, err := conditionBuilder.ConditionFor(ctx, 0, 0, &tc.key, tc.operator, tc.value, sb)
sb.Where(cond)
cond, _, err := conditionBuilder.ConditionFor(ctx, 0, 0, &tc.key, []*telemetrytypes.TelemetryFieldKey{&tc.key}, tc.operator, tc.value, sb)
sb.Where(cond...)
if tc.expectedError != nil {
assert.Equal(t, tc.expectedError, err)

View File

@@ -44,6 +44,16 @@ const (
messageSubField = "message"
messageSubColumn = "body_v2.message"
bodySearchDefaultWarning = "body searches default to `body.message:string`. Use `body.<key>` to search a different field inside body"
// bodyMessageField is the field name addressing the message sub-field of the
// body when use_json_body is enabled (i.e. `body` + `.` + `message`). hasToken
// targets this column in that mode.
bodyMessageField = "body.message"
// Documentation URLs attached to function-call errors so the visitor can
// surface them to the user without knowing function-specific details.
hasTokenFunctionDocURL = "https://signoz.io/docs/userguide/functions-reference/#hastoken-function"
functionBodyJSONSearchDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#function-supports-only-body-json-search"
)
var (

View File

@@ -28,7 +28,6 @@ func TestLikeAndILikeWithoutWildcards_Warns(t *testing.T) {
ConditionBuilder: cb,
FieldKeys: keys,
FullTextColumn: DefaultFullTextColumn,
JsonKeyToKey: GetBodyJSONKey,
}
tests := []string{
@@ -66,9 +65,7 @@ func TestLikeAndILikeWithWildcards_NoWarn(t *testing.T) {
FieldMapper: fm,
ConditionBuilder: cb,
FieldKeys: keys,
FullTextColumn: DefaultFullTextColumn,
JsonKeyToKey: GetBodyJSONKey,
}
FullTextColumn: DefaultFullTextColumn}
tests := []string{
"service.name LIKE 'demo-%'",

View File

@@ -30,7 +30,6 @@ func TestFilterExprLogsBodyJSON(t *testing.T) {
ConditionBuilder: cb,
FieldKeys: keys,
FullTextColumn: &telemetrytypes.TelemetryFieldKey{Name: "body"},
JsonKeyToKey: GetBodyJSONKey,
}
testCases := []struct {

View File

@@ -50,7 +50,6 @@ func TestFilterExprLogs(t *testing.T) {
ConditionBuilder: cb,
FieldKeys: keys,
FullTextColumn: DefaultFullTextColumn,
JsonKeyToKey: GetBodyJSONKey,
StartNs: uint64(releaseTime.Add(-5 * time.Minute).UnixNano()),
EndNs: uint64(releaseTime.Add(5 * time.Minute).UnixNano()),
}
@@ -2467,7 +2466,6 @@ func TestFilterExprLogsConflictNegation(t *testing.T) {
ConditionBuilder: cb,
FieldKeys: keys,
FullTextColumn: DefaultFullTextColumn,
JsonKeyToKey: GetBodyJSONKey,
}
testCases := []struct {

View File

@@ -58,7 +58,6 @@ func NewLogQueryStatementBuilder(
telemetrytypes.SourceUnspecified,
metadataStore,
fullTextColumn,
jsonKeyToKey,
fl,
telemetryStore,
skipResourceFingerprintThreshold,
@@ -659,8 +658,6 @@ func (b *logQueryStatementBuilder) addFilterCondition(
var preparedWhereClause querybuilder.PreparedWhereClause
var err error
// TODO(Tushar): thread orgID here to evaluate correctly
bodyJSONEnabled := b.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
if query.Filter != nil && query.Filter.Expression != "" {
// add filter expression
@@ -670,10 +667,8 @@ func (b *logQueryStatementBuilder) addFilterCondition(
FieldMapper: b.fm,
ConditionBuilder: b.cb,
FieldKeys: keys,
BodyJSONEnabled: bodyJSONEnabled,
SkipResourceFilter: skipResourceFilter,
FullTextColumn: b.fullTextColumn,
JsonKeyToKey: b.jsonKeyToKey,
Variables: variables,
StartNs: start,
EndNs: end,

View File

@@ -20,6 +20,40 @@ func NewConditionBuilder(fm qbtypes.FieldMapper) *conditionBuilder {
}
func (c *conditionBuilder) ConditionFor(
ctx context.Context,
tsStart, tsEnd uint64,
key *telemetrytypes.TelemetryFieldKey,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
// has/hasAny/hasAll/hasToken are logs-body-only; reject to avoid malformed related-values SQL.
if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
return nil, nil, err
}
// metadata builds best-effort filters for related-values lookups; an unknown key
// simply yields no condition rather than an error.
keys, warning := querybuilder.ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
conds := make([]string, 0, len(keys))
for _, k := range keys {
cond, err := c.conditionForKey(ctx, tsStart, tsEnd, k, operator, value, sb)
if err != nil {
return nil, nil, err
}
conds = append(conds, cond)
}
return conds, warnings, nil
}
func (c *conditionBuilder) conditionForKey(
ctx context.Context,
tsStart, tsEnd uint64,
key *telemetrytypes.TelemetryFieldKey,

View File

@@ -53,8 +53,8 @@ func TestConditionFor(t *testing.T) {
for _, tc := range testCases {
sb := sqlbuilder.NewSelectBuilder()
t.Run(tc.name, func(t *testing.T) {
cond, err := conditionBuilder.ConditionFor(ctx, 0, 0, &tc.key, tc.operator, tc.value, sb)
sb.Where(cond)
cond, _, err := conditionBuilder.ConditionFor(ctx, 0, 0, &tc.key, []*telemetrytypes.TelemetryFieldKey{&tc.key}, tc.operator, tc.value, sb)
sb.Where(cond...)
if tc.expectedError != nil {
assert.Equal(t, tc.expectedError, err)

View File

@@ -1190,6 +1190,34 @@ func enrichWithIntrinsicMetricKeys(keys map[string][]*telemetrytypes.TelemetryFi
return keys
}
// enrichWithGenAIKeys surfaces the gen_ai semantic-convention span attributes for
// trace queries even when they have not been ingested yet, so the AI query builder's
// gate and aggregate columns resolve on a fresh install. Only fills a name the store
// did not already return (ingested data wins).
func enrichWithGenAIKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) map[string][]*telemetrytypes.TelemetryFieldKey {
if len(selectors) == 0 {
return keys
}
for _, selector := range selectors {
if selector.Signal != telemetrytypes.SignalTraces && selector.Signal != telemetrytypes.SignalUnspecified {
continue
}
for name, def := range telemetrytypes.GenAIFieldDefinitions {
if len(keys[name]) > 0 {
continue // already resolved from ingested data
}
if !selectorMatchesIntrinsicField(selector, def) {
continue
}
keyCopy := def
keys[name] = []*telemetrytypes.TelemetryFieldKey{&keyCopy}
}
}
return keys
}
func selectorMatchesIntrinsicField(selector *telemetrytypes.FieldKeySelector, definition telemetrytypes.TelemetryFieldKey) bool {
if selector.FieldContext != telemetrytypes.FieldContextUnspecified && selector.FieldContext != definition.FieldContext {
return false
@@ -1275,6 +1303,7 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, fieldKeySelector *tele
applyBackwardCompatibleKeys(mapOfKeys)
mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, selectors)
mapOfKeys = enrichWithGenAIKeys(mapOfKeys, selectors)
return mapOfKeys, complete, nil
}
@@ -1353,6 +1382,7 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
applyBackwardCompatibleKeys(mapOfKeys)
mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, fieldKeySelectors)
mapOfKeys = enrichWithGenAIKeys(mapOfKeys, fieldKeySelectors)
return mapOfKeys, complete, nil
}
@@ -1372,9 +1402,6 @@ func (t *telemetryMetaStore) getRelatedValues(ctx context.Context, fieldValueSel
instrumentationtypes.CodeFunctionName: "getRelatedValues",
})
// TODO(Tushar): thread orgID here to evaluate correctly
bodyJSONEnabled := t.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
// nothing to return as "related" value if there is nothing to filter on
if fieldValueSelector.ExistingQuery == "" {
return nil, true, nil
@@ -1424,7 +1451,6 @@ func (t *telemetryMetaStore) getRelatedValues(ctx context.Context, fieldValueSel
FieldMapper: t.fm,
ConditionBuilder: t.conditionBuilder,
FieldKeys: keys,
BodyJSONEnabled: bodyJSONEnabled,
})
if err != nil {
t.logger.WarnContext(ctx, "error parsing existing query for related values", errors.Attr(err))
@@ -1450,22 +1476,22 @@ func (t *telemetryMetaStore) getRelatedValues(ctx context.Context, fieldValueSel
// search on attributes
key.FieldContext = telemetrytypes.FieldContextAttribute
cond, err := t.conditionBuilder.ConditionFor(ctx, 0, 0, key, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
attrConds, _, err := t.conditionBuilder.ConditionFor(ctx, 0, 0, key, []*telemetrytypes.TelemetryFieldKey{key}, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
if err == nil {
conds = append(conds, cond)
conds = append(conds, attrConds...)
}
// search on resource
key.FieldContext = telemetrytypes.FieldContextResource
cond, err = t.conditionBuilder.ConditionFor(ctx, 0, 0, key, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
resourceConds, _, err := t.conditionBuilder.ConditionFor(ctx, 0, 0, key, []*telemetrytypes.TelemetryFieldKey{key}, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
if err == nil {
conds = append(conds, cond)
conds = append(conds, resourceConds...)
}
key.FieldContext = origContext
} else {
cond, err := t.conditionBuilder.ConditionFor(ctx, 0, 0, key, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
keyConds, _, err := t.conditionBuilder.ConditionFor(ctx, 0, 0, key, []*telemetrytypes.TelemetryFieldKey{key}, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
if err == nil {
conds = append(conds, cond)
conds = append(conds, keyConds...)
}
}

View File

@@ -142,6 +142,42 @@ func (c *conditionBuilder) conditionFor(
}
func (c *conditionBuilder) ConditionFor(
ctx context.Context,
startNs uint64,
endNs uint64,
key *telemetrytypes.TelemetryFieldKey,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
// has/hasAny/hasAll/hasToken are logs-body-only; reject for metrics.
if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
return nil, nil, err
}
keys, warning := querybuilder.ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
if len(keys) == 0 {
return nil, warnings, querybuilder.NewKeyNotFoundError(key.Name)
}
conds := make([]string, 0, len(keys))
for _, k := range keys {
cond, err := c.conditionForKey(ctx, startNs, endNs, k, operator, value, sb)
if err != nil {
return nil, nil, err
}
conds = append(conds, cond)
}
return conds, warnings, nil
}
func (c *conditionBuilder) conditionForKey(
ctx context.Context,
startNs uint64,
endNs uint64,

View File

@@ -234,8 +234,8 @@ func TestConditionFor(t *testing.T) {
for _, tc := range testCases {
sb := sqlbuilder.NewSelectBuilder()
t.Run(tc.name, func(t *testing.T) {
cond, err := conditionBuilder.ConditionFor(ctx, 0, 0, &tc.key, tc.operator, tc.value, sb)
sb.Where(cond)
cond, _, err := conditionBuilder.ConditionFor(ctx, 0, 0, &tc.key, []*telemetrytypes.TelemetryFieldKey{&tc.key}, tc.operator, tc.value, sb)
sb.Where(cond...)
if tc.expectedError != nil {
assert.Equal(t, tc.expectedError, err)
@@ -289,8 +289,8 @@ func TestConditionForMultipleKeys(t *testing.T) {
t.Run(tc.name, func(t *testing.T) {
var err error
for _, key := range tc.keys {
cond, err := conditionBuilder.ConditionFor(ctx, 0, 0, &key, tc.operator, tc.value, sb)
sb.Where(cond)
cond, _, err := conditionBuilder.ConditionFor(ctx, 0, 0, &key, []*telemetrytypes.TelemetryFieldKey{&key}, tc.operator, tc.value, sb)
sb.Where(cond...)
if err != nil {
t.Fatalf("Error getting condition for key %s: %v", key.Name, err)
}

View File

@@ -44,6 +44,46 @@ func keyIndexFilter(key *telemetrytypes.TelemetryFieldKey) any {
}
func (b *defaultConditionBuilder) ConditionFor(
ctx context.Context,
startNs uint64,
endNs uint64,
key *telemetrytypes.TelemetryFieldKey,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
op qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
// has/hasAny/hasAll/hasToken are logs-body-only functions; they never apply to the
// resource fingerprint table, so skip them (the main query still evaluates them).
if op.IsFunctionOperator() {
return nil, nil, nil
}
keys, warning := querybuilder.ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
conds := make([]string, 0, len(keys))
for _, k := range keys {
// the resource fingerprint table only stores resource attributes; keys from
// any other context contribute no condition and are omitted. An empty result
// (including an unknown key) lets the caller skip this filter entirely.
if k.FieldContext != telemetrytypes.FieldContextResource {
continue
}
cond, err := b.conditionForKey(ctx, startNs, endNs, k, op, value, sb)
if err != nil {
return nil, nil, err
}
conds = append(conds, cond)
}
return conds, warnings, nil
}
func (b *defaultConditionBuilder) conditionForKey(
ctx context.Context,
startNs uint64,
endNs uint64,
@@ -53,10 +93,6 @@ func (b *defaultConditionBuilder) ConditionFor(
sb *sqlbuilder.SelectBuilder,
) (string, error) {
if key.FieldContext != telemetrytypes.FieldContextResource {
return querybuilder.SkipConditionLiteral, nil
}
// except for in, not in, between, not between all other operators should have formatted value
// as we store resource values as string
formattedValue := querybuilder.FormatValueForContains(value)

View File

@@ -205,8 +205,8 @@ func TestConditionBuilder(t *testing.T) {
for _, tc := range testCases {
sb := sqlbuilder.NewSelectBuilder()
t.Run(tc.name, func(t *testing.T) {
cond, err := conditionBuilder.ConditionFor(context.Background(), 0, 0, tc.key, tc.op, tc.value, sb)
sb.Where(cond)
cond, _, err := conditionBuilder.ConditionFor(context.Background(), 0, 0, tc.key, []*telemetrytypes.TelemetryFieldKey{tc.key}, tc.op, tc.value, sb)
sb.Where(cond...)
if tc.expectedErr != nil {
assert.Error(t, err)

View File

@@ -24,7 +24,6 @@ func NewResolver[T any](
source telemetrytypes.Source,
metadataStore telemetrytypes.MetadataStore,
fullTextColumn *telemetrytypes.TelemetryFieldKey,
jsonKeyToKey qbtypes.JsonKeyToFieldFunc,
fl flagger.Flagger,
telemetryStore telemetrystore.TelemetryStore,
threshold uint64,
@@ -38,7 +37,6 @@ func NewResolver[T any](
source,
metadataStore,
fullTextColumn,
jsonKeyToKey,
fl,
),
telemetryStore: telemetryStore,

View File

@@ -8,10 +8,8 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
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"
)
@@ -28,7 +26,6 @@ type resourceFilterStatementBuilder[T any] struct {
flagger flagger.Flagger
fullTextColumn *telemetrytypes.TelemetryFieldKey
jsonKeyToKey qbtypes.JsonKeyToFieldFunc
}
// Ensure interface compliance at compile time.
@@ -45,7 +42,6 @@ func New[T any](
source telemetrytypes.Source,
metadataStore telemetrytypes.MetadataStore,
fullTextColumn *telemetrytypes.TelemetryFieldKey,
jsonKeyToKey qbtypes.JsonKeyToFieldFunc,
fl flagger.Flagger,
) *resourceFilterStatementBuilder[T] {
set := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryresourcefilter")
@@ -62,7 +58,6 @@ func New[T any](
source: source,
flagger: fl,
fullTextColumn: fullTextColumn,
jsonKeyToKey: jsonKeyToKey,
}
}
@@ -158,9 +153,6 @@ func (b *resourceFilterStatementBuilder[T]) addConditions(
variables map[string]qbtypes.VariableItem,
) (bool, error) {
// TODO(Tushar): thread orgID here to evaluate correctly
bodyJSONEnabled := b.flagger.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
// Add filter condition if present
if query.Filter != nil && query.Filter.Expression != "" {
@@ -171,16 +163,13 @@ func (b *resourceFilterStatementBuilder[T]) addConditions(
FieldMapper: b.fieldMapper,
ConditionBuilder: b.conditionBuilder,
FieldKeys: keys,
BodyJSONEnabled: bodyJSONEnabled,
FullTextColumn: b.fullTextColumn,
JsonKeyToKey: b.jsonKeyToKey,
SkipFullTextFilter: true,
SkipFunctionCalls: true,
// there is no need for "key" not found error for resource filtering
IgnoreNotFoundKeys: true,
Variables: variables,
StartNs: start,
EndNs: end,
// the resource-filter condition builder ignores keys it can't resolve (and
// skips function calls), so no "key not found" error arises here.
Variables: variables,
StartNs: start,
EndNs: end,
})
if err != nil {

View File

@@ -361,7 +361,6 @@ func TestResourceFilterStatementBuilder_Traces(t *testing.T) {
telemetrytypes.SourceUnspecified,
mockMetadataStore,
nil,
nil,
flaggertest.New(t),
)
@@ -555,7 +554,6 @@ func TestResourceFilterStatementBuilder_Logs(t *testing.T) {
telemetrytypes.SourceUnspecified,
mockMetadataStore,
nil,
nil,
flaggertest.New(t),
)
@@ -623,7 +621,6 @@ func TestResourceFilterStatementBuilder_Variables(t *testing.T) {
telemetrytypes.SourceUnspecified,
mockMetadataStore,
nil,
nil,
flaggertest.New(t),
)

View File

@@ -255,6 +255,42 @@ func (c *conditionBuilder) conditionFor(
}
func (c *conditionBuilder) ConditionFor(
ctx context.Context,
startNs uint64,
endNs uint64,
key *telemetrytypes.TelemetryFieldKey,
fieldKeysForName []*telemetrytypes.TelemetryFieldKey,
operator qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
// has/hasAny/hasAll/hasToken are logs-body-only; reject for traces.
if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
return nil, nil, err
}
keys, warning := querybuilder.ResolveKeys(key, fieldKeysForName)
var warnings []string
if warning != "" {
warnings = append(warnings, warning)
}
if len(keys) == 0 {
return nil, warnings, querybuilder.NewKeyNotFoundError(key.Name)
}
conds := make([]string, 0, len(keys))
for _, k := range keys {
cond, err := c.conditionForKey(ctx, startNs, endNs, k, operator, value, sb)
if err != nil {
return nil, nil, err
}
conds = append(conds, cond)
}
return conds, warnings, nil
}
func (c *conditionBuilder) conditionForKey(
ctx context.Context,
startNs uint64,
endNs uint64,

View File

@@ -293,8 +293,8 @@ func TestConditionFor(t *testing.T) {
for _, tc := range testCases {
sb := sqlbuilder.NewSelectBuilder()
t.Run(tc.name, func(t *testing.T) {
cond, err := conditionBuilder.ConditionFor(ctx, 1761437108000000000, 1761458708000000000, &tc.key, tc.operator, tc.value, sb)
sb.Where(cond)
cond, _, err := conditionBuilder.ConditionFor(ctx, 1761437108000000000, 1761458708000000000, &tc.key, []*telemetrytypes.TelemetryFieldKey{&tc.key}, tc.operator, tc.value, sb)
sb.Where(cond...)
if tc.expectedError != nil {
assert.Equal(t, tc.expectedError, err)
@@ -380,9 +380,9 @@ func TestConditionForResourceWithEvolution(t *testing.T) {
for _, tc := range testCases {
sb := sqlbuilder.NewSelectBuilder()
t.Run(tc.name, func(t *testing.T) {
cond, err := conditionBuilder.ConditionFor(ctx, tc.tsStart, tc.tsEnd, &tc.key, tc.operator, nil, sb)
cond, _, err := conditionBuilder.ConditionFor(ctx, tc.tsStart, tc.tsEnd, &tc.key, []*telemetrytypes.TelemetryFieldKey{&tc.key}, tc.operator, nil, sb)
require.NoError(t, err)
sb.Where(cond)
sb.Where(cond...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
assert.Contains(t, sql, tc.expectedSQL)
})

View File

@@ -54,7 +54,6 @@ func NewTraceQueryStatementBuilder(
telemetrytypes.SourceUnspecified,
metadataStore,
nil,
nil,
flagger,
telemetryStore,
skipResourceFingerprintThreshold,

View File

@@ -44,7 +44,6 @@ func NewTraceOperatorStatementBuilder(
telemetrytypes.SourceUnspecified,
metadataStore,
nil,
nil,
flagger,
)

View File

@@ -8,6 +8,7 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/query-service/agentConf"
"github.com/SigNoz/signoz/pkg/types"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/uptrace/bun"
)
@@ -15,11 +16,13 @@ import (
const (
LLMCostFeatureType agentConf.AgentFeatureType = "llm_pricing"
GenAIRequestModel = "gen_ai.request.model"
GenAIUsageInputTokens = "gen_ai.usage.input_tokens"
GenAIUsageOutputTokens = "gen_ai.usage.output_tokens"
GenAIUsageCacheReadInputTokens = "gen_ai.usage.cache_read.input_tokens"
GenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation.input_tokens"
// gen_ai semconv keys live in telemetrytypes (single source); aliased here
// for existing pricing consumers.
GenAIRequestModel = telemetrytypes.GenAIRequestModel
GenAIUsageInputTokens = telemetrytypes.GenAIUsageInputTokens
GenAIUsageOutputTokens = telemetrytypes.GenAIUsageOutputTokens
GenAIUsageCacheReadInputTokens = telemetrytypes.GenAIUsageCacheReadInputTokens
GenAIUsageCacheCreationInputTokens = telemetrytypes.GenAIUsageCacheCreationInputTokens
SignozGenAICostInput = "_signoz.gen_ai.cost_input"
SignozGenAICostOutput = "_signoz.gen_ai.cost_output"

View File

@@ -111,6 +111,13 @@ const (
FilterOperatorContains
FilterOperatorNotContains
// has/hasAny/hasAll are array membership functions; hasToken is a full-text token
// search (not array membership), so it is excluded from IsArrayFunctionOperator.
FilterOperatorHas
FilterOperatorHasToken
FilterOperatorHasAny
FilterOperatorHasAll
)
var operatorInverseMapping = map[FilterOperator]FilterOperator{
@@ -224,6 +231,45 @@ func (f FilterOperator) IsArrayOperator() bool {
}
}
// IsArrayFunctionOperator reports whether the operator is one of the array
// membership functions (has/hasAny/hasAll) that operate over array fields.
func (f FilterOperator) IsArrayFunctionOperator() bool {
switch f {
case FilterOperatorHas, FilterOperatorHasAny, FilterOperatorHasAll:
return true
default:
return false
}
}
// IsFunctionOperator reports whether the operator is a query function
// (has/hasAny/hasAll/hasToken); these apply to the logs body column only.
func (f FilterOperator) IsFunctionOperator() bool {
switch f {
case FilterOperatorHas, FilterOperatorHasAny, FilterOperatorHasAll, FilterOperatorHasToken:
return true
default:
return false
}
}
// FunctionName returns the query-text name of a function operator
// (has/hasAny/hasAll/hasToken), or "" for any non-function operator.
func (f FilterOperator) FunctionName() string {
switch f {
case FilterOperatorHas:
return "has"
case FilterOperatorHasAny:
return "hasAny"
case FilterOperatorHasAll:
return "hasAll"
case FilterOperatorHasToken:
return "hasToken"
default:
return ""
}
}
type OrderDirection struct {
valuer.String
}

View File

@@ -10,7 +10,7 @@ import (
)
var (
ErrColumnNotFound = errors.Newf(errors.TypeNotFound, errors.CodeNotFound, "field not found")
ErrColumnNotFound = errors.NewNotFoundf(errors.CodeNotFound, "field not found")
ErrBetweenValues = errors.NewInvalidInputf(errors.CodeInvalidInput, "(not) between operator requires two values")
ErrBetweenValuesType = errors.NewInvalidInputf(errors.CodeInvalidInput, "(not) between operator requires two values of the number type")
ErrInValues = errors.NewInvalidInputf(errors.CodeInvalidInput, "(not) in operator requires a list of values")
@@ -29,10 +29,14 @@ type FieldMapper interface {
ColumnExpressionFor(ctx context.Context, tsStart, tsEnd uint64, key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) (string, error)
}
// ConditionBuilder builds the condition for the filter.
// ConditionBuilder builds the conditions for the filter.
type ConditionBuilder interface {
// ConditionFor returns the condition for the given key, operator and value.
ConditionFor(ctx context.Context, startNs uint64, endNs uint64, key *telemetrytypes.TelemetryFieldKey, operator FilterOperator, value any, sb *sqlbuilder.SelectBuilder) (string, error)
// ConditionFor returns the conditions and any advisory warnings for a filter
// term. key is the field key as parsed from the query text; fieldKeysForName is
// the set of known field keys matching it (may be empty). The builder owns the
// decision of what to do — resolve ambiguity, fall back to a body JSON search,
// emit a "not found" error, or skip — and which errors/warnings are apt.
ConditionFor(ctx context.Context, startNs uint64, endNs uint64, key *telemetrytypes.TelemetryFieldKey, fieldKeysForName []*telemetrytypes.TelemetryFieldKey, operator FilterOperator, value any, sb *sqlbuilder.SelectBuilder) ([]string, []string, error)
}
type AggExprRewriter interface {

View File

@@ -172,58 +172,13 @@ func (f *TelemetryFieldKey) Normalize() {
}
// GetFieldKeyFromKeyText returns a TelemetryFieldKey from a key text.
// The key text is expected to be in the format of `fieldContext.fieldName:fieldDataType` in the search query.
// Both fieldContext and :fieldDataType are optional.
// fieldName can contain dots and can start with a dot (e.g., ".http_code").
// Special cases:
// - When key exactly matches a field context name (e.g., "body", "attribute"), use unspecified context.
// - When key starts with "body." prefix, use "body" as context with remainder as field name.
// GetFieldKeyFromKeyText returns a TelemetryFieldKey parsed from a key text of the
// form `fieldContext.fieldName:fieldDataType` (context and :dataType optional). It
// delegates to Normalize; see Normalize for the parsing rules and special cases.
func GetFieldKeyFromKeyText(key string) TelemetryFieldKey {
var explicitFieldDataType = FieldDataTypeUnspecified
var fieldName string
// Step 1: Parse data type from the right (after the last ":")
var keyWithoutDataType string
if colonIdx := strings.LastIndex(key, ":"); colonIdx != -1 {
potentialDataType := key[colonIdx+1:]
if dt, ok := fieldDataTypes[potentialDataType]; ok && dt != FieldDataTypeUnspecified {
explicitFieldDataType = dt
keyWithoutDataType = key[:colonIdx]
} else {
// No valid data type found, treat the entire key as the field name
keyWithoutDataType = key
}
} else {
keyWithoutDataType = key
}
// Step 2: Parse field context from the left
if dotIdx := strings.Index(keyWithoutDataType, "."); dotIdx != -1 {
potentialContext := keyWithoutDataType[:dotIdx]
if fc, ok := fieldContexts[potentialContext]; ok && fc != FieldContextUnspecified {
fieldName = keyWithoutDataType[dotIdx+1:]
// Step 2a: Handle special case for log.body.* fields
if fc == FieldContextLog && strings.HasPrefix(fieldName, BodyJSONStringSearchPrefix) {
fc = FieldContextBody
fieldName = strings.TrimPrefix(fieldName, BodyJSONStringSearchPrefix)
}
return TelemetryFieldKey{
Name: fieldName,
FieldContext: fc,
FieldDataType: explicitFieldDataType,
}
}
}
// Step 3: No context found, entire key is the field name
return TelemetryFieldKey{
Name: keyWithoutDataType,
FieldContext: FieldContextUnspecified,
FieldDataType: explicitFieldDataType,
}
f := TelemetryFieldKey{Name: key}
f.Normalize()
return f
}
func TelemetryFieldKeyToText(key *TelemetryFieldKey) string {

View File

@@ -0,0 +1,30 @@
package telemetrytypes
// OpenTelemetry gen_ai semantic-convention attribute keys. Single source of truth
// shared by the AI query builder and the LLM pricing pipeline.
const (
GenAIRequestModel = "gen_ai.request.model"
GenAIToolName = "gen_ai.tool.name"
GenAIAgentName = "gen_ai.agent.name"
GenAIUsageInputTokens = "gen_ai.usage.input_tokens"
GenAIUsageOutputTokens = "gen_ai.usage.output_tokens"
GenAIUsageCacheReadInputTokens = "gen_ai.usage.cache_read.input_tokens"
GenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation.input_tokens"
)
// GenAIFieldDefinitions are the gen_ai semantic-convention span attributes the AI
// query builder relies on. They are surfaced by the metadata store for trace
// queries regardless of whether they have been ingested yet, so the AI gate/columns
// resolve on a fresh install (mirrors intrinsic metric keys). String keys are the
// gate; the usage keys are numeric.
var GenAIFieldDefinitions = map[string]TelemetryFieldKey{
GenAIRequestModel: {Name: GenAIRequestModel, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
GenAIToolName: {Name: GenAIToolName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
GenAIAgentName: {Name: GenAIAgentName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
GenAIUsageInputTokens: {Name: GenAIUsageInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
GenAIUsageOutputTokens: {Name: GenAIUsageOutputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
GenAIUsageCacheReadInputTokens: {Name: GenAIUsageCacheReadInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
GenAIUsageCacheCreationInputTokens: {Name: GenAIUsageCacheCreationInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
}

View File

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

View File

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

View File

@@ -0,0 +1,544 @@
"""
Integration tests for source="ai" over the traces signal.
These ingest OpenTelemetry gen_ai spans into real ClickHouse via the insert_traces
fixture and exercise the actual /api/v5/query_range API, so they validate the whole
path: payload -> AI statement builder -> ClickHouse -> response.
Data shape (generic OTel gen_ai semantic conventions):
- a root span (no gen_ai attributes)
- an LLM span carrying gen_ai.request.model (str) and numeric usage attributes
(gen_ai.usage.input_tokens / output_tokens / cost) plus gen_ai.user.id
Each test tags its spans with a unique service.name and filters on it, so tests do
not interfere with each other's data.
"""
import json
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.querier import (
BuilderQuery,
MetricAggregation,
OrderBy,
RequestType,
TelemetryFieldKey,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
def _ai_trace(
*,
now: datetime,
service: str,
user: str,
in_tokens: int,
out_tokens: int,
cost: float,
llm_duration_s: float = 1.0,
error: bool = False,
) -> 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()
llm_id = TraceIdGenerator.span_id()
resources = {"service.name": service, "deployment.environment": "production"}
root = Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=llm_duration_s + 0.1),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
llm = Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=llm_duration_s),
trace_id=trace_id,
span_id=llm_id,
parent_span_id=root_id,
name="chat gpt-4o-mini",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=(TracesStatusCode.STATUS_CODE_ERROR if error else TracesStatusCode.STATUS_CODE_OK),
resources=resources,
attributes={
"gen_ai.request.model": "gpt-4o-mini",
"gen_ai.system": "openai",
"gen_ai.user.id": user,
# numeric values land in attributes_number
"gen_ai.usage.input_tokens": in_tokens,
"gen_ai.usage.output_tokens": out_tokens,
"gen_ai.usage.cost": cost,
},
)
return [root, llm]
def _non_ai_trace(*, now: datetime, service: str) -> list[Traces]:
"""A plain HTTP trace with no gen_ai attributes; must be excluded by the AI gate."""
trace_id = TraceIdGenerator.trace_id()
span_id = TraceIdGenerator.span_id()
return [
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=span_id,
parent_span_id="",
name="GET /health",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources={"service.name": service},
attributes={"http.request.method": "GET"},
)
]
def _window_ms(now: datetime) -> tuple[int, int]:
start_ms = int((now - timedelta(minutes=10)).timestamp() * 1000)
end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
return start_ms, end_ms
def test_ai_list_excludes_non_ai(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Trace-list panel (requestType="trace"): returns AI traces and excludes the
non-AI trace. Asserts on the raw response payload to stay agnostic to the exact
row schema.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-list"
ai = _ai_trace(now=now, service=service, user="alice", in_tokens=100, out_tokens=20, cost=0.5)
non_ai = _non_ai_trace(now=now, service=service)
ai_trace_id = ai[0].trace_id
non_ai_trace_id = non_ai[0].trace_id
insert_traces(ai + non_ai)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
limit=10,
)
response = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
)
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert ai_trace_id in body, f"expected AI trace {ai_trace_id} in list response"
assert non_ai_trace_id not in body, f"non-AI trace {non_ai_trace_id} should be excluded by the gate"
def _ai_trace_mixed_spans(*, now: datetime, service: str, user: str) -> list[Traces]:
"""
Root + one LLM span + one tool span + one agent span. The gate matches all three
child spans, but only the LLM span carries gen_ai.request.model.
"""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service, "deployment.environment": "production"}
def _span(name, kind, attributes, offset_s):
return Traces(
timestamp=now - timedelta(seconds=offset_s),
duration=timedelta(seconds=0.5),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name=name,
kind=kind,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes=attributes,
)
root = Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=4),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
llm = _span("chat gpt-4o-mini", TracesKind.SPAN_KIND_CLIENT, {
"gen_ai.request.model": "gpt-4o-mini",
"gen_ai.system": "openai",
"gen_ai.user.id": user,
"gen_ai.usage.input_tokens": 100,
"gen_ai.usage.output_tokens": 20,
}, 4)
tool = _span("execute_tool", TracesKind.SPAN_KIND_INTERNAL, {
"gen_ai.tool.name": "get_weather",
"gen_ai.tool.type": "function",
}, 3)
agent = _span("agent.step", TracesKind.SPAN_KIND_INTERNAL, {
"gen_ai.agent.name": "chat-agent",
}, 2)
return [root, llm, tool, agent]
def test_ai_list_llm_call_count_counts_llm_only(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
llm_call_count counts LLM spans only (gen_ai.request.model), not the full gate:
a trace with 1 LLM + 1 tool + 1 agent span (4 spans incl. root) reports
llm_call_count == 1 and span_count == 4.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-llmcount"
insert_traces(_ai_trace_mixed_spans(now=now, service=service, user="alice"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
limit=10,
)
response = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
)
assert response.status_code == HTTPStatus.OK, response.text
rows = response.json()["data"]["data"]["results"][0]["rows"]
assert len(rows) == 1, f"expected exactly one AI trace, got: {rows}"
data = rows[0]["data"]
assert data["llm_call_count"] == 1, f"llm_call_count should count LLM spans only: {data}"
assert data["span_count"] == 4, f"span_count should include all spans: {data}"
def test_ai_list_having_aggregate_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Aggregate filter written in the SAME filter box: the span-level predicate narrows
to the service, the trace-level `output_tokens > 100` keeps the large-token
trace and drops the small one (split internally into WHERE + HAVING).
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-having"
small = _ai_trace(now=now, service=service, user="alice", in_tokens=10, out_tokens=20, cost=0.1)
large = _ai_trace(now=now, service=service, user="bob", in_tokens=10, out_tokens=500, cost=0.2)
small_id = small[0].trace_id
large_id = large[0].trace_id
insert_traces(small + large)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}' AND output_tokens > 100",
limit=10,
)
response = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
)
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert large_id in body, f"trace with 500 out-tokens should pass output_tokens > 100"
assert small_id not in body, f"trace with 20 out-tokens should be filtered out by HAVING"
def test_ai_list_order_limit_offset(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Trace list honors order by (aggregate column) + limit + offset (pagination)."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-order"
traces: list[Traces] = []
for out in (100, 200, 300, 400, 500):
traces += _ai_trace(now=now, service=service, user="u", in_tokens=10, out_tokens=out, cost=0.1)
insert_traces(traces)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
def page(offset: int) -> list[int]:
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
order=[OrderBy(key=TelemetryFieldKey(name="output_tokens"), direction="desc")],
limit=2,
offset=offset,
)
resp = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
)
assert resp.status_code == HTTPStatus.OK, resp.text
rows = resp.json()["data"]["data"]["results"][0]["rows"]
return [int(r["data"]["output_tokens"]) for r in rows]
assert page(0) == [500, 400], "first page: highest output_tokens, desc"
assert page(2) == [300, 200], "second page (offset 2): next two, desc"
def test_ai_span_list_limit(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Span list honors limit (delegated raw path): 6 gen_ai spans available, capped to 4."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-spanlimit"
insert_traces(
_ai_trace_mixed_spans(now=now, service=service, user="a")
+ _ai_trace_mixed_spans(now=now, service=service, user="b")
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
limit=4,
)
resp = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW
)
assert resp.status_code == HTTPStatus.OK, resp.text
rows = resp.json()["data"]["data"]["results"][0]["rows"]
assert len(rows) == 4, f"limit should cap at 4 (6 gen_ai spans available), got {len(rows)}"
def test_ai_span_list_excludes_non_gen_ai_spans(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Span list (requestType=raw): returns only the gen_ai spans (LLM/tool/agent); the
root span of the same trace (no gen_ai attributes) is excluded by the span-level gate.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-spanlist"
insert_traces(_ai_trace_mixed_spans(now=now, service=service, user="alice"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}'",
select_fields=[TelemetryFieldKey(name="name", field_context="span")],
limit=50,
)
response = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW
)
assert response.status_code == HTTPStatus.OK, response.text
rows = response.json()["data"]["data"]["results"][0]["rows"]
names = sorted(r["data"]["name"] for r in rows)
assert names == ["agent.step", "chat gpt-4o-mini", "execute_tool"], names
assert "POST /api/chat" not in names # root span excluded
def test_ai_list_having_or_aggregates(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""
Two aggregate conditions OR-ed within the filter box (regression guard for OR-group
whitespace handling): output_tokens > 100 OR span_count > 100 keeps only the
large-token trace (span_count is 2 for both, so that branch never matches).
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-having-or"
small = _ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=20, cost=0.1)
large = _ai_trace(now=now, service=service, user="b", in_tokens=10, out_tokens=500, cost=0.2)
small_id, large_id = small[0].trace_id, large[0].trace_id
insert_traces(small + large)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}' AND (output_tokens > 100 OR span_count > 100)",
limit=10,
)
response = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
)
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert large_id in body
assert small_id not in body
def test_ai_list_having_trace_context_prefix(
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 `trace.` context prefix on an aggregate column works like the bare name."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-trace-ctx"
small = _ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=20, cost=0.1)
large = _ai_trace(now=now, service=service, user="b", in_tokens=10, out_tokens=500, cost=0.2)
small_id, large_id = small[0].trace_id, large[0].trace_id
insert_traces(small + large)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
source="ai",
name="A",
filter_expression=f"service.name = '{service}' AND trace.output_tokens > 100",
limit=10,
)
response = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
)
assert response.status_code == HTTPStatus.OK, response.text
body = json.dumps(response.json())
assert large_id in body
assert small_id not in body
def test_ai_source_rejected_on_logs(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
) -> None:
"""source=ai is only valid for traces; on logs it must be a validation error."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(signal="logs", source="ai", name="A", limit=5)
response = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert "traces signal" in response.text
def test_ai_source_rejected_on_metrics(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
) -> None:
"""source=ai on metrics must be a validation error, not a silent normal query."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="metrics",
source="ai",
name="A",
aggregations=[MetricAggregation(
metric_name="system_memory_usage",
time_aggregation="avg",
space_aggregation="avg",
temporality="unspecified",
)],
step_interval=60,
)
response = make_query_request(
signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.TIME_SERIES
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert "traces signal" in response.text
def test_ai_list_rejects_aggregate_or_span_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
) -> None:
"""
Aggregate (HAVING) columns may not be OR-ed with span-level keys in the trace
list; a span-OR-span filter is fine.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
# aggregate OR span -> rejected
bad = BuilderQuery(
signal="traces", source="ai", name="A", limit=10,
filter_expression="output_tokens > 1000 OR service.name = 'ai-it-orfilter'",
)
response = make_query_request(
signoz, token, start_ms, end_ms, [bad.to_dict()], request_type="trace"
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert "cannot be combined" in response.text
# span OR span -> accepted (empty result is fine; just not an error)
ok = BuilderQuery(
signal="traces", source="ai", name="A", limit=10,
filter_expression="service.name = 'ai-it-orfilter' OR has_error = true",
)
response = make_query_request(
signoz, token, start_ms, end_ms, [ok.to_dict()], request_type="trace"
)
assert response.status_code == HTTPStatus.OK, response.text