mirror of
https://github.com/SigNoz/signoz.git
synced 2026-10-16 07:01:11 +01:00
Compare commits
1 Commits
refactor/r
...
ns/trace-a
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
69fbf1b75d |
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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]),
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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']")
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user