Compare commits

...

1 Commits

Author SHA1 Message Date
Nikhil Soni
69fbf1b75d feat(traces-qb): read unreported span attributes through the JSON rollout
A key metadata does not report used to fall back to the legacy maps in
every window. It now carries the column-wide attribute evolutions, loaded
once per request, so it follows the same rollout as a reported key.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 21:13:44 +05:30
28 changed files with 387 additions and 103 deletions

View File

@@ -188,7 +188,10 @@ func (s *traceStore) genAISpanColumns(ctx context.Context, orgID valuer.UUID, bo
return nil, err
}
q := querybuilder.NewQueryInfo(ctx, orgID, s.flagger, telemetrytypes.SignalTraces, nil, uint64(bounds.Start.UnixNano()), uint64(bounds.End.UnixNano()))
q, err := querybuilder.LoadColumnEvolutions(ctx, querybuilder.NewQueryInfo(ctx, orgID, s.flagger, telemetrytypes.SignalTraces, nil, uint64(bounds.Start.UnixNano()), uint64(bounds.End.UnixNano())), s.metadataStore)
if err != nil {
return nil, err
}
isGenAI, err := s.anyExistsCondition(ctx, q, aiobservabilitytypes.GenAISpanGateKeys, keys, sb)
if err != nil {
return nil, err
@@ -288,7 +291,10 @@ func (s *traceStore) selectStorableSpans(ctx context.Context, orgID valuer.UUID)
}
func (s *traceStore) GetThreadSpans(ctx context.Context, orgID valuer.UUID, traceID string, bounds *spantypes.TraceBounds, page spantypes.ThreadPage) ([]spantypes.StorableSpan, error) {
q := querybuilder.NewQueryInfo(ctx, orgID, s.flagger, telemetrytypes.SignalTraces, nil, uint64(bounds.Start.UnixNano()), uint64(bounds.End.UnixNano()))
q, err := querybuilder.LoadColumnEvolutions(ctx, querybuilder.NewQueryInfo(ctx, orgID, s.flagger, telemetrytypes.SignalTraces, nil, uint64(bounds.Start.UnixNano()), uint64(bounds.End.UnixNano())), s.metadataStore)
if err != nil {
return nil, err
}
sb := sqlbuilder.NewSelectBuilder()
sb.Select(
"DISTINCT ON (span_id) timestamp", "duration_nano", "span_id", "parent_span_id", "has_error", "name", "kind_string",

View File

@@ -9,7 +9,6 @@ import (
chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
@@ -20,8 +19,6 @@ type aggExprRewriter struct {
logger *slog.Logger
fullTextColumn *telemetrytypes.TelemetryFieldKey
storage qbtypes.Storage
flagger flagger.Flagger
signal telemetrytypes.Signal
}
var _ qbtypes.AggExprRewriter = (*aggExprRewriter)(nil)
@@ -30,8 +27,6 @@ func NewAggExprRewriter(
settings factory.ProviderSettings,
fullTextColumn *telemetrytypes.TelemetryFieldKey,
storage qbtypes.Storage,
fl flagger.Flagger,
signal telemetrytypes.Signal,
) *aggExprRewriter {
set := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/querybuilder/agg_rewrite")
@@ -39,8 +34,6 @@ func NewAggExprRewriter(
logger: set.Logger(),
fullTextColumn: fullTextColumn,
storage: storage,
flagger: fl,
signal: signal,
}
}
@@ -49,9 +42,7 @@ func NewAggExprRewriter(
// and the args if the parametric aggregation function is used.
func (r *aggExprRewriter) Rewrite(
ctx context.Context,
orgID valuer.UUID,
startNs uint64,
endNs uint64,
q qbtypes.QueryInfo,
expr string,
rateInterval uint64,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
@@ -80,7 +71,7 @@ func (r *aggExprRewriter) Rewrite(
visitor := &exprVisitor{
ctx: ctx,
query: NewQueryInfo(ctx, orgID, r.flagger, r.signal, nil, startNs, endNs),
query: q,
logger: r.logger,
fieldKeys: keys,
fullTextColumn: r.fullTextColumn,
@@ -100,9 +91,7 @@ func (r *aggExprRewriter) Rewrite(
// RewriteMulti rewrites a slice of expressions.
func (r *aggExprRewriter) RewriteMulti(
ctx context.Context,
orgID valuer.UUID,
startNs uint64,
endNs uint64,
q qbtypes.QueryInfo,
exprs []string,
rateInterval uint64,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
@@ -111,7 +100,7 @@ func (r *aggExprRewriter) RewriteMulti(
var errs []error
var chArgsList [][]any
for i, e := range exprs {
w, chArgs, err := r.Rewrite(ctx, orgID, startNs, endNs, e, rateInterval, keys)
w, chArgs, err := r.Rewrite(ctx, q, e, rateInterval, keys)
if err != nil {
errs = append(errs, err)
out[i] = e

View File

@@ -31,6 +31,21 @@ func NewQueryInfo(ctx context.Context, orgID valuer.UUID, fl flagger.Flagger, si
return q
}
// LoadColumnEvolutions loads the column-wide evolution entries a key metadata
// does not report still reads through: the JSON span attributes column, when
// the flag puts it on the query path. Any other query keeps q as it is.
func LoadColumnEvolutions(ctx context.Context, q qbtypes.QueryInfo, store telemetrytypes.MetadataStore) (qbtypes.QueryInfo, error) {
if store == nil || q.Signal != telemetrytypes.SignalTraces || !q.TraceAttrsJSONOn {
return q, nil
}
evolutions, err := store.GetColumnEvolutions(ctx, q.Signal, telemetrytypes.FieldContextAttribute)
if err != nil {
return q, err
}
q.Evolutions = evolutions
return q, nil
}
// Resolve turns one requested key into its meanings, one time for each
// use. The order is the same for every storage, in the filter and in every
// select field, group by, order by, and aggregation:

View File

@@ -42,7 +42,7 @@ func NewFactory(
factory.MustNewName("audit"),
func(_ context.Context, settings factory.ProviderSettings, _ statementbuilder.Config) (qbtypes.StatementBuilder[qbtypes.LogAggregation], error) {
storage := audittelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, audittelemetryschema.DefaultFullTextColumn, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, audittelemetryschema.DefaultFullTextColumn, storage)
return NewAuditQueryStatementBuilder(
settings, metadataStore, storage, aggExprRewriter, audittelemetryschema.DefaultFullTextColumn, fl,
), nil
@@ -354,7 +354,7 @@ func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
allAggChArgs := make([]any, 0)
for i, agg := range query.Aggregations {
rewritten, chArgs, err := b.aggExprRewriter.Rewrite(ctx, orgID, start, end, agg.Expression, uint64(query.StepInterval.Seconds()), keys)
rewritten, chArgs, err := b.aggExprRewriter.Rewrite(ctx, info, agg.Expression, uint64(query.StepInterval.Seconds()), keys)
if err != nil {
return nil, err
}
@@ -492,7 +492,7 @@ func (b *auditQueryStatementBuilder) buildScalarQuery(
if len(query.Aggregations) > 0 {
for idx := range query.Aggregations {
aggExpr := query.Aggregations[idx]
rewritten, chArgs, err := b.aggExprRewriter.Rewrite(ctx, orgID, start, end, aggExpr.Expression, rateInterval, keys)
rewritten, chArgs, err := b.aggExprRewriter.Rewrite(ctx, info, aggExpr.Expression, rateInterval, keys)
if err != nil {
return nil, err
}

View File

@@ -56,7 +56,7 @@ func newTestAuditStatementBuilder(t *testing.T) *auditQueryStatementBuilder {
mockMetadataStore.KeysMap = auditFieldKeyMap()
storage := audittelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
return NewAuditQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),

View File

@@ -58,7 +58,7 @@ func TestStatementBuilderResolvesLogFamilies(t *testing.T) {
}
mockMetadataStore.KeysMap = keys
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, storage, aggExprRewriter,
@@ -114,7 +114,7 @@ func TestStatementBuilderResolvesLogFamilyFilteredAggregation(t *testing.T) {
}
mockMetadataStore.KeysMap = keys
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, storage, aggExprRewriter,
@@ -169,7 +169,7 @@ func TestStatementBuilderResolvesSingleSpellingAcrossNames(t *testing.T) {
}}
mockMetadataStore.KeysMap = keys
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, storage, aggExprRewriter,
@@ -229,7 +229,7 @@ func TestStatementBuilderResolvesLogFamilyGroupBy(t *testing.T) {
}
mockMetadataStore.KeysMap = keys
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, storage, aggExprRewriter,

View File

@@ -47,7 +47,7 @@ func TestStatementBuilderGroupByUnknownKey(t *testing.T) {
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, storage, aggExprRewriter,

View File

@@ -1335,7 +1335,7 @@ func buildJSONTestStatementBuilder(t *testing.T, addIndexes bool) (*logQueryStat
fl := flaggertest.WithUseJSONBody(t, true)
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),

View File

@@ -42,7 +42,7 @@ func TestStatementBuilderResourceBodyConflict(t *testing.T) {
fl := flaggertest.WithUseJSONBody(t, true)
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
store,

View File

@@ -29,7 +29,7 @@ func TestSearchCostGuard(t *testing.T) {
storage := logstelemetryschema.NewStorage()
store := telemetrytypestest.NewMockMetadataStore()
store.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
sb := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
store, storage, rewriter, logstelemetryschema.DefaultFullTextColumn, fl, nil,
@@ -59,7 +59,7 @@ func TestSearchCostGuardJSONBody(t *testing.T) {
storage := logstelemetryschema.NewStorage()
store := telemetrytypestest.NewMockMetadataStore()
store.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
sb := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
store, storage, rewriter, logstelemetryschema.DefaultFullTextColumn, fl, nil,

View File

@@ -60,7 +60,7 @@ func NewFactory(
factory.MustNewName("logs"),
func(_ context.Context, settings factory.ProviderSettings, cfg statementbuilder.Config) (qbtypes.StatementBuilder[qbtypes.LogAggregation], error) {
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, logstelemetryschema.DefaultFullTextColumn, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, logstelemetryschema.DefaultFullTextColumn, storage)
return NewLogQueryStatementBuilder(
settings, metadataStore, storage, aggExprRewriter, logstelemetryschema.DefaultFullTextColumn,
fl, telemetryStore, cfg,
@@ -480,7 +480,7 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
allAggChArgs := make([]any, 0)
for i, agg := range query.Aggregations {
rewritten, chArgs, err := b.aggExprRewriter.Rewrite(
ctx, orgID, start, end, agg.Expression,
ctx, info, agg.Expression,
uint64(query.StepInterval.Seconds()),
keys,
)
@@ -649,7 +649,7 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
for idx := range query.Aggregations {
aggExpr := query.Aggregations[idx]
rewritten, chArgs, err := b.aggExprRewriter.Rewrite(
ctx, orgID, start, end, aggExpr.Expression,
ctx, info, aggExpr.Expression,
rateInterval,
keys,
)

View File

@@ -220,7 +220,7 @@ func TestStatementBuilderTimeSeries(t *testing.T) {
storage := logstelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -360,7 +360,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -510,7 +510,7 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -585,7 +585,7 @@ func TestStatementBuilderTimeSeriesBodyGroupBy(t *testing.T) {
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -680,7 +680,7 @@ func TestStatementBuilderListQueryServiceCollision(t *testing.T) {
storage := logstelemetryschema.NewStorage()
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMapCollision()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -903,7 +903,7 @@ func TestAdjustKey(t *testing.T) {
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMapCollision()
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -1048,7 +1048,7 @@ func TestStmtBuilderBodyField(t *testing.T) {
f := field
mockMetadataStore.KeysMap[field.Name] = append(mockMetadataStore.KeysMap[field.Name], &f)
}
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
@@ -1146,7 +1146,7 @@ func TestStmtBuilderBodyFullTextSearch(t *testing.T) {
f := field
mockMetadataStore.KeysMap[field.Name] = append(mockMetadataStore.KeysMap[field.Name], &f)
}
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
@@ -1258,7 +1258,7 @@ func newSkipResourceFingerprintLogsBuilder(
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC))
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), logstelemetryschema.DefaultFullTextColumn, storage, fl, telemetrytypes.SignalLogs)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), logstelemetryschema.DefaultFullTextColumn, storage)
return NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),

View File

@@ -226,7 +226,10 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
return nil, err
}
matchedSB := sqlbuilder.NewSelectBuilder()
q := querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end)
q, err := querybuilder.LoadColumnEvolutions(ctx, querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end), b.metadataStore)
if err != nil {
return nil, err
}
maskExpr, resolved, err := b.resolveFor(ctx, q, keys, matchedSB)
if err != nil {
return nil, err
@@ -245,7 +248,7 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
return nil, err
}
fp, err := b.splitFilter(ctx, orgID, query, b.aggregateAliasSet(), keys, start, end, variables, matchedSB)
fp, err := b.splitFilter(ctx, q, query, b.aggregateAliasSet(), keys, variables, matchedSB)
if err != nil {
return nil, err
}
@@ -445,7 +448,7 @@ type filterParts struct {
// splitFilter splits query.Filter into a span-level predicate and a trace-level
// HAVING (explicit query.Having ANDed on before resolution); args bind into sb.
// keys must cover the filter's span-level selectors.
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID valuer.UUID, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet map[string]struct{}, keys map[string][]*telemetrytypes.TelemetryFieldKey, start, end uint64, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (filterParts, error) {
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, q qbtypes.QueryInfo, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet map[string]struct{}, keys map[string][]*telemetrytypes.TelemetryFieldKey, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (filterParts, error) {
var fp filterParts
havingExpr := ""
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
@@ -455,7 +458,7 @@ func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID val
}
havingExpr = traceExpr
if strings.TrimSpace(spanExpr) != "" {
pred, warnings, url, err := b.resolveSpanPredicate(ctx, querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end), spanExpr, keys, variables, sb)
pred, warnings, url, err := b.resolveSpanPredicate(ctx, q, spanExpr, keys, variables, sb)
if err != nil {
return fp, err
}

View File

@@ -340,8 +340,12 @@ func (b *scopedTraceStatementBuilder) buildQualifiedStatement(
if err != nil {
return nil, nil, err
}
q, err := querybuilder.LoadColumnEvolutions(ctx, querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end), b.metadataStore)
if err != nil {
return nil, nil, err
}
sb := sqlbuilder.NewSelectBuilder()
maskExpr, resolved, err := b.resolveFor(ctx, querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end), keys, sb)
maskExpr, resolved, err := b.resolveFor(ctx, q, keys, sb)
if err != nil {
return nil, nil, err
}
@@ -511,15 +515,13 @@ type scanContext struct {
func (b *scopedTraceStatementBuilder) newScanContext(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
q qbtypes.QueryInfo,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
spanExpr, traceExpr string,
variables map[string]qbtypes.VariableItem,
) (*scanContext, error) {
sc := &scanContext{sb: sqlbuilder.NewSelectBuilder()}
var err error
q := querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end)
sc.maskExpr, sc.resolved, err = b.resolveFor(ctx, q, keys, sc.sb)
if err != nil {
return nil, err
@@ -566,6 +568,10 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
if err != nil {
return nil, err
}
q, err := querybuilder.LoadColumnEvolutions(ctx, querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end), b.metadataStore)
if err != nil {
return nil, err
}
resourceFrag, resourceArgs, resourcePred, err := b.maybeAttachResourceFilter(ctx, orgID, query, start, end, variables)
if err != nil {
@@ -582,7 +588,7 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
// __qualified: its own scan resolution, HAVING = the trace-level filter part
qualified := false
if strings.TrimSpace(traceExpr) != "" {
qsc, err := b.newScanContext(ctx, orgID, start, end, keys, "", traceExpr, variables)
qsc, err := b.newScanContext(ctx, q, keys, "", traceExpr, variables)
if err != nil {
return nil, err
}
@@ -598,7 +604,7 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
}
}
groupCols, err := b.resolveGroupColumns(ctx, querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end), query.GroupBy, keys)
groupCols, err := b.resolveGroupColumns(ctx, q, query.GroupBy, keys)
if err != nil {
return nil, err
}
@@ -640,7 +646,7 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
// (exact for non-composable aggregates) and prune the main scan to the top-N.
limitPred := ""
if requestType == qbtypes.RequestTypeTimeSeries && query.Limit > 0 && len(groupCols) > 0 {
tsc, err := b.newScanContext(ctx, orgID, start, end, keys, spanExpr, "", variables)
tsc, err := b.newScanContext(ctx, q, keys, spanExpr, "", variables)
if err != nil {
return nil, err
}
@@ -666,7 +672,7 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
strings.Join(exprs, ", "), strings.Join(groupNames, ", "))
}
msc, err := b.newScanContext(ctx, orgID, start, end, keys, spanExpr, "", variables)
msc, err := b.newScanContext(ctx, q, keys, spanExpr, "", variables)
if err != nil {
return nil, err
}

View File

@@ -37,7 +37,7 @@ func newBulkTestBuilder(t *testing.T, releaseTime time.Time) *traceQueryStatemen
}}
store.ColumnEvolutionMetadataMap["traces:attribute:__all__"] = tracestelemetryschema.MockAttributeEvolutionData(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
return NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
store, storage, aggExprRewriter, nil, fl, false, 100000,
@@ -148,3 +148,71 @@ func TestGroupByAttributeHomeAcrossRollout(t *testing.T) {
})
}
}
// TestUnknownAttributeFollowsRollout: a filter or group-by key metadata does not report reads the
// home the window selects, the same as a reported key.
func TestUnknownAttributeFollowsRollout(t *testing.T) {
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
rel := releaseTime.UnixMilli()
day := int64(24 * time.Hour / time.Millisecond)
b := newBulkTestBuilder(t, releaseTime)
testCases := []struct {
name string
startMs uint64
endMs uint64
wantContains []string
wantNotContains []string
}{
{
name: "BeforeRollout_ReadsLegacyMap",
startMs: uint64(rel - 2*day),
endMs: uint64(rel - day),
wantContains: []string{"attributes_string['unknown.attr'] = ?", "mapContains(attributes_string, 'unknown.attr')"},
wantNotContains: []string{"attributes.`unknown.attr`"},
},
{
name: "StraddlingRollout_JSONThenLegacyFallback",
startMs: uint64(rel - day),
endMs: uint64(rel + day),
wantContains: []string{"attributes.`unknown.attr` IS NOT NULL", "attributes.`unknown.attr`::String", "attributes_string['unknown.attr']"},
wantNotContains: nil,
},
{
name: "AfterRollout_ReadsJSONOnly",
startMs: uint64(rel + day),
endMs: uint64(rel + 2*day),
wantContains: []string{"attributes.`unknown.attr`::String = ?"},
wantNotContains: []string{"attributes_string"},
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
stmt, err := b.Build(
context.Background(), valuer.UUID{}, testCase.startMs, testCase.endMs,
qbtypes.RequestTypeTimeSeries,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
Filter: &qbtypes.Filter{Expression: "unknown.attr = 'x'"},
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "unknown.attr",
}}},
Limit: 10,
},
nil,
)
require.NoError(t, err)
for _, want := range testCase.wantContains {
assert.Contains(t, stmt.Query, want)
}
for _, unwanted := range testCase.wantNotContains {
assert.NotContains(t, stmt.Query, unwanted)
}
assert.Contains(t, strings.Join(stmt.Warnings, "\n"), "not found in metadata")
})
}
}

View File

@@ -55,7 +55,7 @@ func NewFactory(
factory.MustNewName("traces"),
func(_ context.Context, settings factory.ProviderSettings, cfg statementbuilder.Config) (qbtypes.StatementBuilder[qbtypes.TraceAggregation], error) {
storage := tracestelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, nil, storage)
return NewTraceQueryStatementBuilder(
settings, metadataStore, storage, aggExprRewriter, telemetryStore, fl,
cfg.SkipResourceFingerprint.Enabled, cfg.SkipResourceFingerprint.Threshold,
@@ -169,18 +169,22 @@ func (b *traceQueryStatementBuilder) Build(
for _, action := range adjustTraceKeys(keys, &query, requestType) {
b.logger.DebugContext(ctx, "key adjustment action", slog.String("action", action))
}
info, err := querybuilder.LoadColumnEvolutions(ctx, querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end), b.metadataStore)
if err != nil {
return nil, err
}
// Create SQL builder
q := sqlbuilder.NewSelectBuilder()
switch requestType {
case qbtypes.RequestTypeRaw:
return b.buildListQuery(ctx, orgID, q, query, start, end, keys, variables, isSelectFieldsEmpty)
return b.buildListQuery(ctx, orgID, q, query, start, end, info, keys, variables, isSelectFieldsEmpty)
case qbtypes.RequestTypeTimeSeries:
return b.buildTimeSeriesQuery(ctx, orgID, q, query, start, end, keys, variables)
return b.buildTimeSeriesQuery(ctx, orgID, q, query, start, end, info, keys, variables)
case qbtypes.RequestTypeScalar:
return b.buildScalarQuery(ctx, orgID, q, query, start, end, keys, variables, false, false)
return b.buildScalarQuery(ctx, orgID, q, query, start, end, info, keys, variables, false, false)
case qbtypes.RequestTypeTrace:
return b.buildTraceQuery(ctx, orgID, q, query, start, end, keys, variables)
return b.buildTraceQuery(ctx, orgID, q, query, start, end, info, keys, variables)
}
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported request type: %s", requestType)
@@ -355,6 +359,7 @@ func (b *traceQueryStatementBuilder) buildListQuery(
sb *sqlbuilder.SelectBuilder,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
info qbtypes.QueryInfo,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
isSelectFieldsEmpty bool,
@@ -379,7 +384,6 @@ func (b *traceQueryStatementBuilder) buildListQuery(
cteArgs = append(cteArgs, scopeArgs...)
}
info := querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end)
for i, field := range query.SelectFields {
expr, err := querybuilder.ResolveColumn(ctx, info, b.storage, &field, telemetrytypes.FieldDataTypeUnspecified, keys)
if err != nil {
@@ -398,7 +402,7 @@ func (b *traceQueryStatementBuilder) buildListQuery(
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, info, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
}
@@ -444,6 +448,7 @@ func (b *traceQueryStatementBuilder) buildTraceQuery(
_ *sqlbuilder.SelectBuilder,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
info qbtypes.QueryInfo,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
@@ -470,7 +475,7 @@ func (b *traceQueryStatementBuilder) buildTraceQuery(
}
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, distSB, start, end, query, keys, variables, skipResourceFilter)
preparedWhereClause, err := b.addFilterCondition(ctx, distSB, start, end, info, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
}
@@ -563,6 +568,7 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
sb *sqlbuilder.SelectBuilder,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
info qbtypes.QueryInfo,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
@@ -593,7 +599,6 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
// Keep original column expressions so we can build the tuple
fieldNames := make([]string, 0, len(query.GroupBy))
info := querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end)
for i, gb := range query.GroupBy {
expr, err := querybuilder.ResolveColumn(ctx, info, b.storage, &gb.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
@@ -608,7 +613,7 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
allAggChArgs := make([]any, 0)
for i, agg := range query.Aggregations {
rewritten, chArgs, err := b.aggExprRewriter.Rewrite(
ctx, orgID, start, end, agg.Expression,
ctx, info, agg.Expression,
uint64(query.StepInterval.Seconds()),
keys,
)
@@ -620,7 +625,7 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
}
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, info, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
}
@@ -631,7 +636,7 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
if query.Limit > 0 && len(query.GroupBy) > 0 {
// build the scalar “top/bottom-N” query in its own builder.
cteSB := sqlbuilder.NewSelectBuilder()
cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, query, start, end, keys, variables, true, true)
cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, query, start, end, info, keys, variables, true, true)
if err != nil {
return nil, err
}
@@ -727,6 +732,7 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
sb *sqlbuilder.SelectBuilder,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
info qbtypes.QueryInfo,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
skipResourceCTE bool,
@@ -757,7 +763,6 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
allAggChArgs := []any{}
fieldNames := make([]string, 0, len(query.GroupBy))
info := querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end)
for i, gb := range query.GroupBy {
expr, err := querybuilder.ResolveColumn(ctx, info, b.storage, &gb.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
@@ -776,7 +781,7 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
for idx := range query.Aggregations {
aggExpr := query.Aggregations[idx]
rewritten, chArgs, err := b.aggExprRewriter.Rewrite(
ctx, orgID, start, end, aggExpr.Expression,
ctx, info, aggExpr.Expression,
rateInterval,
keys,
)
@@ -792,7 +797,7 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, info, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
}
@@ -854,9 +859,9 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
// buildFilterCondition builds SQL condition from filter expression.
func (b *traceQueryStatementBuilder) addFilterCondition(
ctx context.Context,
orgID valuer.UUID,
sb *sqlbuilder.SelectBuilder,
start, end uint64,
info qbtypes.QueryInfo,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
@@ -870,7 +875,7 @@ func (b *traceQueryStatementBuilder) addFilterCondition(
// add filter expression
preparedWhereClause, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
Context: ctx,
Query: querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end),
Query: info,
Storage: b.storage,
Logger: b.logger,
FieldKeys: keys,

View File

@@ -468,7 +468,7 @@ func TestStatementBuilder(t *testing.T) {
storage := tracestelemetryschema.NewStorage()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -767,7 +767,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
storage := tracestelemetryschema.NewStorage()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -1032,7 +1032,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
if mockMetadataStore.KeysMap == nil {
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
}
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -1102,7 +1102,7 @@ func TestStatementBuilderGroupByResourceEvolution(t *testing.T) {
storage := tracestelemetryschema.NewStorage()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -1267,7 +1267,7 @@ func TestStatementBuilderTraceQuery(t *testing.T) {
storage := tracestelemetryschema.NewStorage()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -1905,7 +1905,7 @@ func newSkipResourceFingerprintBuilder(
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
return NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -1929,7 +1929,7 @@ func TestStatementBuilderGroupByUnseenKey(t *testing.T) {
storage := tracestelemetryschema.NewStorage()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -1969,7 +1969,7 @@ func TestStatementBuilderAggregationUnseenKey(t *testing.T) {
storage := tracestelemetryschema.NewStorage()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -2042,7 +2042,7 @@ func TestStatementBuilderSemconvFamilies(t *testing.T) {
storage := tracestelemetryschema.NewStorage()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
@@ -2081,7 +2081,7 @@ func TestStatementBuilderSemconvSingleSpelling(t *testing.T) {
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
}
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),

View File

@@ -28,6 +28,7 @@ type traceOperatorCTEBuilder struct {
start uint64
end uint64
orgID valuer.UUID
queryInfo qbtypes.QueryInfo
operator *qbtypes.QueryBuilderTraceOperator
stmtBuilder *traceOperatorStatementBuilder
queries map[string]*qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]
@@ -265,7 +266,7 @@ func (b *traceOperatorCTEBuilder) buildQueryCTE(ctx context.Context, queryName s
query.Filter.Expression,
querybuilder.FilterExprVisitorOpts{
Context: ctx,
Query: querybuilder.NewQueryInfo(ctx, b.orgID, b.stmtBuilder.fl, telemetrytypes.SignalTraces, nil, b.start, b.end),
Query: b.queryInfo,
Storage: b.stmtBuilder.storage,
Logger: b.stmtBuilder.logger,
FieldKeys: keys,
@@ -485,7 +486,7 @@ func (b *traceOperatorCTEBuilder) buildListQuery(ctx context.Context, selectFrom
}
// Add selectFields since we now have all base table columns
info := querybuilder.NewQueryInfo(ctx, b.orgID, b.stmtBuilder.fl, telemetrytypes.SignalTraces, nil, b.start, b.end)
info := b.queryInfo
for i, field := range b.operator.SelectFields {
if selectedFields[field.Name] {
continue
@@ -629,7 +630,7 @@ func (b *traceOperatorCTEBuilder) buildTimeSeriesQuery(ctx context.Context, sele
int64(b.operator.StepInterval.Seconds()),
))
info := querybuilder.NewQueryInfo(ctx, b.orgID, b.stmtBuilder.fl, telemetrytypes.SignalTraces, nil, b.start, b.end)
info := b.queryInfo
for _, gb := range b.operator.GroupBy {
expr, err := querybuilder.ResolveColumn(ctx, info, b.stmtBuilder.storage, &gb.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
@@ -647,9 +648,7 @@ func (b *traceOperatorCTEBuilder) buildTimeSeriesQuery(ctx context.Context, sele
for i, agg := range b.operator.Aggregations {
rewritten, chArgs, err := b.stmtBuilder.aggExprRewriter.Rewrite(
ctx,
b.orgID,
b.start,
b.end,
b.queryInfo,
agg.Expression,
uint64(b.operator.StepInterval.Seconds()),
keys,
@@ -726,7 +725,7 @@ func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFro
sb := sqlbuilder.NewSelectBuilder()
info := querybuilder.NewQueryInfo(ctx, b.orgID, b.stmtBuilder.fl, telemetrytypes.SignalTraces, nil, b.start, b.end)
info := b.queryInfo
for _, gb := range b.operator.GroupBy {
expr, err := querybuilder.ResolveColumn(ctx, info, b.stmtBuilder.storage, &gb.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
@@ -746,9 +745,7 @@ func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFro
for i, agg := range b.operator.Aggregations {
rewritten, chArgs, err := b.stmtBuilder.aggExprRewriter.Rewrite(
ctx,
b.orgID,
b.start,
b.end,
b.queryInfo,
agg.Expression,
rateInterval,
keys,
@@ -853,7 +850,7 @@ func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFro
func (b *traceOperatorCTEBuilder) buildScalarQuery(ctx context.Context, selectFromCTE string, keys map[string][]*telemetrytypes.TelemetryFieldKey) (*qbtypes.Statement, error) {
sb := sqlbuilder.NewSelectBuilder()
info := querybuilder.NewQueryInfo(ctx, b.orgID, b.stmtBuilder.fl, telemetrytypes.SignalTraces, nil, b.start, b.end)
info := b.queryInfo
for _, gb := range b.operator.GroupBy {
expr, err := querybuilder.ResolveColumn(ctx, info, b.stmtBuilder.storage, &gb.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
@@ -871,9 +868,7 @@ func (b *traceOperatorCTEBuilder) buildScalarQuery(ctx context.Context, selectFr
for i, agg := range b.operator.Aggregations {
rewritten, chArgs, err := b.stmtBuilder.aggExprRewriter.Rewrite(
ctx,
b.orgID,
b.start,
b.end,
b.queryInfo,
agg.Expression,
uint64((b.end-b.start)/querybuilder.NsToSeconds),
keys,

View File

@@ -23,7 +23,7 @@ func newTestTraceOperatorStatementBuilder(t *testing.T) *traceOperatorStatementB
storage := tracestelemetryschema.NewStorage()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
traceStmtBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, storage, aggExprRewriter, nil, fl, false, 100000,

View File

@@ -42,7 +42,7 @@ func NewOperatorFactory(
factory.MustNewName("traceoperator"),
func(_ context.Context, settings factory.ProviderSettings, cfg statementbuilder.Config) (qbtypes.TraceOperatorStatementBuilder, error) {
storage := tracestelemetryschema.NewStorage()
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, nil, storage)
traceStmtBuilder := NewTraceQueryStatementBuilder(
settings, metadataStore, storage, aggExprRewriter, telemetryStore, fl,
cfg.SkipResourceFingerprint.Enabled, cfg.SkipResourceFingerprint.Threshold,
@@ -116,11 +116,17 @@ func (b *traceOperatorStatementBuilder) Build(
slog.String("expression", query.Expression),
slog.Any("request_type", requestType))
info, err := querybuilder.LoadColumnEvolutions(ctx, querybuilder.NewQueryInfo(ctx, orgID, b.fl, telemetrytypes.SignalTraces, nil, start, end), b.metadataStore)
if err != nil {
return nil, err
}
// Build the CTE-based query
builder := &traceOperatorCTEBuilder{
start: start,
end: end,
orgID: orgID,
queryInfo: info,
operator: &query,
stmtBuilder: b,
queries: make(map[string]*qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]),

View File

@@ -39,7 +39,7 @@ func TestTraceTimeRangeOptimization(t *testing.T) {
Signal: telemetrytypes.SignalTraces,
}}
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),

View File

@@ -2430,6 +2430,16 @@ func (k *telemetryMetaStore) fetchEvolutionEntryFromClickHouse(ctx context.Conte
return entries, nil
}
func (k *telemetryMetaStore) GetColumnEvolutions(ctx context.Context, signal telemetrytypes.Signal, fieldContext telemetrytypes.FieldContext) ([]*telemetrytypes.EvolutionEntry, error) {
ctx = withTelemetryContext(ctx, signal, "GetColumnEvolutions")
selector := &telemetrytypes.EvolutionSelector{Signal: signal, FieldContext: fieldContext, FieldName: "__all__"}
evolutions, err := k.fetchEvolutionEntryFromClickHouse(ctx, []*telemetrytypes.EvolutionSelector{selector})
if err != nil {
return nil, errors.WrapInternalf(err, errors.CodeInternal, "failed to fetch column evolutions")
}
return evolutions, nil
}
// updateColumnEvolutionMetadataForKeys updates the evolution field for keys.
func (k *telemetryMetaStore) updateColumnEvolutionMetadataForKeys(ctx context.Context, keysToUpdate []*telemetrytypes.TelemetryFieldKey) error {
// an empty selector list would run the evolution query without a filter

View File

@@ -531,7 +531,8 @@ func foldAbsentJSONReadToTypeDefault(key *telemetrytypes.TelemetryFieldKey, oper
// real column is that column. A span or trace context is kept as written,
// and corrects to the attribute maps when it names no column. A strict
// context synthesizes its type variants under the stripped and the literal
// spelling.
// spelling. A synthesized attribute key reads through the column-wide
// evolutions of the query, so it follows the JSON rollout like a reported one.
func (m *storage) Fallback(ctx context.Context, q qbtypes.QueryInfo, key *telemetrytypes.TelemetryFieldKey, _ qbtypes.FilterOperator, value any) ([]*telemetrytypes.LogicalField, error) {
var keys []*telemetrytypes.TelemetryFieldKey
switch key.FieldContext {
@@ -560,9 +561,29 @@ func (m *storage) Fallback(ctx context.Context, q qbtypes.QueryInfo, key *teleme
literal := telemetrytypes.NewTelemetryFieldKey(key.FieldContext.StringValue()+"."+key.Name, key.FieldContext, key.FieldDataType)
keys = append(querybuilder.SynthesizeKeys(key, value), querybuilder.SynthesizeKeys(literal, value)...)
}
stampAttributeEvolutions(q, keys)
return querybuilder.WrapAsLogicalFields(key.Name, keys), nil
}
// stampAttributeEvolutions gives every synthesized attribute key without
// evolutions the query's column-wide attribute entries.
func stampAttributeEvolutions(q qbtypes.QueryInfo, keys []*telemetrytypes.TelemetryFieldKey) {
var evolutions []*telemetrytypes.EvolutionEntry
for _, e := range q.Evolutions {
if e != nil && e.FieldContext == telemetrytypes.FieldContextAttribute {
evolutions = append(evolutions, e)
}
}
if len(evolutions) == 0 {
return
}
for _, key := range keys {
if key.FieldContext == telemetrytypes.FieldContextAttribute && len(key.Evolutions) == 0 {
key.Evolutions = evolutions
}
}
}
// stampColumnType gives an untyped column key the data type its column reads
// as, so the intrinsic-column step can compare it with same-named metadata.
func stampColumnType(key *telemetrytypes.TelemetryFieldKey, columns []*schema.Column) *telemetrytypes.TelemetryFieldKey {

View File

@@ -527,3 +527,81 @@ func TestConditionForAttributeJSONStraddleAbsentKeyExclusion(t *testing.T) {
assert.Contains(t, sql, ", NULL) IS NULL")
})
}
// TestConditionForAttributeJSONUnknownKey: a key metadata does not report follows the rollout like
// a reported one, through the query's column-wide evolutions; without them it stays on the Map.
func TestConditionForAttributeJSONUnknownKey(t *testing.T) {
ctx := context.Background()
storage := NewStorage()
evo := MockAttributeEvolutionData(attrJSONRelease)
build := func(t *testing.T, q qbtypes.QueryInfo, key telemetrytypes.TelemetryFieldKey, op qbtypes.FilterOperator, value any) string {
t.Helper()
sb := sqlbuilder.NewSelectBuilder()
conds, _, err := querybuilder.Conditions(ctx, q, storage, &key, op, value, map[string][]*telemetrytypes.TelemetryFieldKey{}, false, sb)
require.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return sql
}
query := func(window [2]uint64, evolutions []*telemetrytypes.EvolutionEntry) qbtypes.QueryInfo {
return qbtypes.QueryInfo{StartNs: window[0], EndNs: window[1], TraceAttrsJSONOn: true, Evolutions: evolutions}
}
bare := telemetrytypes.TelemetryFieldKey{Name: "unknown.attr"}
t.Run("before -> map", func(t *testing.T) {
sql := build(t, query(attrWindowBefore, evo), bare, qbtypes.FilterOperatorEqual, "x")
assert.Contains(t, sql, "attributes_string['unknown.attr'] = ?")
assert.NotContains(t, sql, "attributes.`unknown.attr`")
})
t.Run("after -> json", func(t *testing.T) {
sql := build(t, query(attrWindowAfter, evo), bare, qbtypes.FilterOperatorEqual, "x")
assert.Contains(t, sql, "attributes.`unknown.attr`::String = ?")
assert.NotContains(t, sql, "attributes_string")
})
t.Run("straddle -> dual", func(t *testing.T) {
sql := build(t, query(attrWindowStraddle, evo), bare, qbtypes.FilterOperatorEqual, "x")
assert.Contains(t, sql, "multiIf(attributes.`unknown.attr` IS NOT NULL, attributes.`unknown.attr`::String, mapContains(attributes_string, 'unknown.attr'), attributes_string['unknown.attr'], NULL) = ?")
})
t.Run("after number -> type-gated json", func(t *testing.T) {
sql := build(t, query(attrWindowAfter, evo), bare, qbtypes.FilterOperatorEqual, float64(1))
assert.Contains(t, sql, "accurateCastOrNull(attributes.`unknown.attr`, 'Float64')")
assert.NotContains(t, sql, "attributes_number")
})
t.Run("after exists fans out over json only", func(t *testing.T) {
sql := build(t, query(attrWindowAfter, evo), bare, qbtypes.FilterOperatorExists, nil)
assert.Contains(t, sql, "attributes.`unknown.attr` IS NOT NULL")
assert.NotContains(t, sql, "mapContains")
})
t.Run("after qualified attribute spelling -> json for both spellings", func(t *testing.T) {
qualified := telemetrytypes.TelemetryFieldKey{Name: "unknown.attr", FieldContext: telemetrytypes.FieldContextAttribute}
sql := build(t, query(attrWindowAfter, evo), qualified, qbtypes.FilterOperatorEqual, "x")
assert.Contains(t, sql, "attributes.`unknown.attr`::String = ?")
assert.Contains(t, sql, "attributes.`attribute.unknown.attr`::String = ?")
assert.NotContains(t, sql, "attributes_string")
})
t.Run("after span spelling corrects to the json attribute", func(t *testing.T) {
span := telemetrytypes.TelemetryFieldKey{Name: "unknown.attr", FieldContext: telemetrytypes.FieldContextSpan}
sql := build(t, query(attrWindowAfter, evo), span, qbtypes.FilterOperatorEqual, "x")
assert.Contains(t, sql, "attributes.`unknown.attr`::String = ?")
assert.NotContains(t, sql, "attributes_string")
})
t.Run("without column-wide evolutions -> map in every window", func(t *testing.T) {
sql := build(t, query(attrWindowAfter, nil), bare, qbtypes.FilterOperatorEqual, "x")
assert.Contains(t, sql, "attributes_string['unknown.attr'] = ?")
assert.NotContains(t, sql, "attributes.`unknown.attr`")
})
t.Run("flag off -> map in every window", func(t *testing.T) {
q := query(attrWindowAfter, evo)
q.TraceAttrsJSONOn = false
sql := build(t, q, bare, qbtypes.FilterOperatorEqual, "x")
assert.Contains(t, sql, "attributes_string['unknown.attr'] = ?")
assert.NotContains(t, sql, "attributes.`unknown.attr`")
})
t.Run("resource spelling is untouched by attribute evolutions", func(t *testing.T) {
resource := telemetrytypes.TelemetryFieldKey{Name: "unknown.attr", FieldContext: telemetrytypes.FieldContextResource}
sql := build(t, query(attrWindowAfter, evo), resource, qbtypes.FilterOperatorEqual, "x")
assert.Contains(t, sql, "resource.`unknown.attr`::String")
assert.Contains(t, sql, "resources_string['unknown.attr']")
})
}

View File

@@ -40,7 +40,9 @@ var (
// every read needs, the signal and queried metric that family admission
// needs, and the query-path flags evaluated one time per request. The
// generic flows read FamiliesOn. Only the logs storage reads BodyJSONOn, and
// only the traces storage reads TraceAttrsJSONOn.
// only the traces storage reads TraceAttrsJSONOn. Evolutions holds the
// column-wide entries (field_name `__all__`) a key metadata does not report
// still reads through, loaded one time per request.
type QueryInfo struct {
StartNs uint64
EndNs uint64
@@ -49,6 +51,7 @@ type QueryInfo struct {
FamiliesOn bool
BodyJSONOn bool
TraceAttrsJSONOn bool
Evolutions []*telemetrytypes.EvolutionEntry
}
// Absent is how a field key reads for a row that does not carry it, with
@@ -214,8 +217,8 @@ type Resolved struct {
type AggExprRewriter interface {
// Rewrite rewrites the aggregation expression to be used in the query.
Rewrite(ctx context.Context, orgID valuer.UUID, startNs, endNs uint64, expr string, rateInterval uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey) (string, []any, error)
RewriteMulti(ctx context.Context, orgID valuer.UUID, startNs, endNs uint64, exprs []string, rateInterval uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey) ([]string, [][]any, error)
Rewrite(ctx context.Context, q QueryInfo, expr string, rateInterval uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey) (string, []any, error)
RewriteMulti(ctx context.Context, q QueryInfo, exprs []string, rateInterval uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey) ([]string, [][]any, error)
}
type Statement struct {

View File

@@ -45,6 +45,11 @@ type MetadataStore interface {
// rows templated by entry; FieldName and ReleaseTime are set per path.
PromotePaths(ctx context.Context, entry EvolutionEntry, paths ...string) error
// GetColumnEvolutions lists the column-wide evolution entries (field_name
// `__all__`) of the signal's field context: the homes a key metadata does
// not report still reads through.
GetColumnEvolutions(ctx context.Context, signal Signal, fieldContext FieldContext) ([]*EvolutionEntry, error)
// GetFirstSeenFromMetricMetadata gets the first seen timestamp for a metric metadata lookup key.
GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []MetricMetadataLookupKey) (map[MetricMetadataLookupKey]int64, error)

View File

@@ -419,6 +419,11 @@ func (m *MockMetadataStore) updateColumnEvolutionMetadataForKeys(_ context.Conte
return result
}
func (m *MockMetadataStore) GetColumnEvolutions(_ context.Context, signal telemetrytypes.Signal, fieldContext telemetrytypes.FieldContext) ([]*telemetrytypes.EvolutionEntry, error) {
selector := &telemetrytypes.EvolutionSelector{Signal: signal, FieldContext: fieldContext, FieldName: "__all__"}
return m.ColumnEvolutionMetadataMap[selector.QualifiedName()], nil
}
func (m *MockMetadataStore) GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []telemetrytypes.MetricMetadataLookupKey) (map[telemetrytypes.MetricMetadataLookupKey]int64, error) {
return m.LookupKeysMap, nil
}

View File

@@ -13,6 +13,7 @@ from fixtures.querier import (
build_aggregation,
build_group_by_field,
build_traces_scalar_query,
get_all_warnings,
get_rows,
index_series_by_label,
make_query_request,
@@ -410,3 +411,71 @@ def test_traces_attributes_json_list_view(
assert row["data"]["name"] == name
# int/float compare equal in Python, so map-sourced float64 and JSON-sourced numbers both match.
assert row["data"]["attributes"] == attributes, name
def test_traces_attributes_json_unknown_key_follows_rollout(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
seed_attribute_evolution: Callable[[str, datetime], None],
) -> None:
"""A key with no metadata rows resolves through the column-wide evolution row like a reported
one: the Map before the rollout, the JSON column after it, both while straddling. The before
span writes only the Map and the after span only the JSON column, so each window's rows prove
which home was read. The not-found warning still fires."""
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
evolution_time = datetime.now(tz=UTC).replace(second=0, microsecond=0) - timedelta(minutes=30)
seed_attribute_evolution("traces", evolution_time)
before = evolution_time - timedelta(minutes=5)
after = evolution_time + timedelta(minutes=5)
spans = [
Traces(
timestamp=before,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name="before",
attributes={"ad.hoc": "/old"},
attribute_write_mode="legacy_only",
),
Traces(
timestamp=after,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name="after",
attributes={"ad.hoc": "/new"},
attribute_write_mode="json_only",
),
]
for span in spans:
span.attribute_keys = []
span.tag_attributes = []
insert_traces(spans)
windows = {
"before": (before - timedelta(minutes=1), before + timedelta(minutes=1), {"/old": {int(before.timestamp() * 1000): 1}}),
"after": (after - timedelta(minutes=1), after + timedelta(minutes=1), {"/new": {int(after.timestamp() * 1000): 1}}),
"straddle": (before - timedelta(minutes=1), after + timedelta(minutes=1), {"/old": {int(before.timestamp() * 1000): 1}, "/new": {int(after.timestamp() * 1000): 1}}),
}
for window, (start, end, expected) in windows.items():
response = make_query_request(
signoz,
token,
start_ms=int(start.timestamp() * 1000),
end_ms=int(end.timestamp() * 1000),
request_type=RequestType.TIME_SERIES,
queries=[
build_traces_scalar_query(
aggregations=[build_aggregation("count()")],
filter_expression="ad.hoc EXISTS",
group_by=[build_group_by_field("ad.hoc", field_context="attribute")],
)
],
)
assert response.status_code == HTTPStatus.OK, window
series = index_series_by_label(response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"], "ad.hoc")
assert_grouped_series(series, expected_values_by_group=expected)
assert any("not found in metadata" in warning.get("message", "") for warning in get_all_warnings(response.json())), window