Compare commits

...

15 Commits

Author SHA1 Message Date
Nikhil Soni
14131aaf56 feat(traces-qb): read the list-view attributes bag from the JSON column
- empty-selectFields ("all fields") list queries now pick their attribute
  homes by column evolution: the legacy maps before the JSON rollout, the
  `attributes` JSON column after it, and both across it — via a new
  MetadataStore.GetColumnEvolutions and BulkAttributeColumns, mirroring the
  per-key home selection.
- consume merges the natively-decoded JSON column (relies on #12555's
  JSONValue scan) into the flat attributes bag: nested paths are flattened
  back to dotted keys and win over the legacy maps on collision, so the bag
  has one stable shape across the rollout with a single decode and no
  re-marshal.
- per-key group-by/aggregation is unchanged (already JSON-only after the
  rollout), so the bag stays complete once map dual-write stops.

Assisted-by: Claude Opus 4.8
2026-09-02 12:59:23 +05:30
Nikhil Soni
8b42912b90 test(traces-qb): cover attribute data-type collisions on the JSON column
A name stored under two data types (e.g. http.status_code as String and
Int64) resolves to two logical fields. These tests pin that behavior on the
JSON attributes column: an untyped filter fans out to one exists-guarded
condition per type (both reading the same physical path, casts differing) and
surfaces the ambiguity warning; group-by folds both interpretations into one
multiIf output column. A before-rollout case anchors the legacy map parity the
JSON path preserves (two separate physical columns, each mapContains-guarded).

Assisted-by: Claude Opus 4.8
2026-09-02 01:57:49 +05:30
Nikhil Soni
1fca671118 docs(telemetrymetadata): note ignored evolve-before-__all__ edge case 2026-09-02 01:55:54 +05:30
Nikhil Soni
b67d5894ea chore: trim comments 2026-09-02 01:55:54 +05:30
Nikhil Soni
97b8ef2e3c refactor(traces-qb): gate attributes_promoted, drop isMap base heuristic
getColumn now offers attributes_promoted as a candidate only when the key
carries its own promotion entry, so every JSON candidate it returns has an
evolution entry. SelectEvolutionsForColumns can then synthesize the epoch-0
base for any unnamed candidate (always the legacy map) without switching on
column type, removing the attributes-specific MapColumnType check.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
de9054ef7f test(traces-qb): exercise nested JSON attribute keys
Use dotted keys (http.route, http.retry.count, http.cache.hit) so the
attributes JSON column nests them under a path while the legacy map keys
them verbatim, covering nested-path resolution end to end across the
evolution boundary and per type.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
bad7906dbb test(traces-qb): add attributes-JSON evolution integration tests
Cover map/JSON/straddle resolution across the attribute column-evolution
boundary plus the per-type casts (string/int/bool/exists) end to end.
Adds the attributes JSON column and an attribute_write_mode to the Traces
fixture, and a seed_attribute_evolution fixture that seeds the evolution
row and deletes it on teardown.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
e8b74856ac chore(traces-qb): trim verbose comments in attribute evolution/mapper 2026-09-02 01:55:54 +05:30
Nikhil Soni
9c6d3e8436 refactor(traces-qb): synthesize the Map base instead of registering it
Only JSON columns are recorded as evolution rows now; the legacy Map column is
the implicit epoch-0 base. SelectEvolutionsForColumns synthesizes a base entry
for any Map candidate the metadata does not name, and a JSON candidate with no
entry (attributes_promoted for an unpromoted key) is left unselected. This drops
the sibling-map __all__ rows a typed attribute key used to inherit, so
narrowEvolutionsToColumns is no longer needed and is removed.

The metadata append (composing the attributes __all__ entry with a per-path
attributes_promoted entry) is still required: a promoted key must keep its
attributes home, or a query between the JSON rollout and promotion would fall
back to the now-empty synthesized Map. The mock store is made faithful to the
real store (appends and assigns key.Evolutions instead of discarding), with a
test covering the promoted vs non-promoted composition.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
0f10fbb632 fix(metadata): append per-field evolution entries instead of replacing __all__
Root cause of the promoted-attribute breakage: updateColumnEvolutionMetadataForKeys
overwrote a key's column-wide (__all__) evolution homes with any per-field
(field_name = key.Name) entries. A promoted path's per-field attributes_promoted
entry is additive, not a re-specification, so replacing dropped its Map/base-JSON
homes and broke queries over ranges before promotion.

Fix is to append the per-field entries to the column-wide ones (both are already
fetched in the same query). This reverts the earlier MergeEvolutions helper, its
override-by-column semantics, the mock-store rewrite, and the extra tests — none
were needed, since a promoted path's per-field entry names a different column and
has nothing to dedup or override.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
6a93afa794 fix(metadata): merge column-wide and per-field evolution entries
updateColumnEvolutionMetadataForKeys composed a key's evolution homes by first
reading the column-wide (field_name "__all__") entries and then OVERWRITING them
with any per-field (field_name = key.Name) entries. A promoted attribute — whose
per-path attributes_promoted entry lives under its own field name — therefore
lost its Map and base-JSON homes and could not be queried for time ranges before
it was promoted.

Compose the two instead: column-wide entries provide the base homes shared by
every field, and per-field entries add homes specific to the field (a per-field
entry overrides a column-wide one only for the same column). Extracted as
telemetrytypes.MergeEvolutions and used by both the real store and the mock
(which previously discarded its result and never set key.Evolutions at all).

Provably a no-op for every existing case: only a key with BOTH kinds of entries
changes, and the only such key today is a promoted attribute.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
95d8ea7602 refactor(traces-qb): let attributes_promoted ride along on the attributes gate
getColumn no longer checks for a separate attributes_promoted evolution entry.
Once the attributes column is registered, attributes_promoted is always returned
as a candidate home; SelectEvolutionsForColumns selects it only when this key has
a promotion entry in range and drops it otherwise (a column with no evolution
entry is never selected). Same resolved SQL, less duplicated gating.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
47c3f3d96a feat(traces-qb): read promoted span attributes from attributes_promoted
Model attribute promotion as a third evolution column rather than a separate
mechanism. A promoted path carries a per-path evolution entry
(column_name=attributes_promoted, field_name=<path>) at its promotion release
time; getColumn adds attributes_promoted to the key's column set when that entry
is present, and SelectEvolutionsForColumns then picks a single physical home per
query window:

- before the JSON rollout        -> the legacy Map column
- between rollout and promotion   -> attributes
- after promotion                 -> attributes_promoted only (its own index)
- across an evolution boundary     -> a multiIf over just the two adjacent homes

So a query fully after promotion reads only attributes_promoted (fast, pruned by
attributes_promoted_paths_tokenbf); both JSON columns are read only transiently
in the straddle window. No coalesce and no new promotion flag are needed - the
existing generic JSON rendering handles any column name from the evolution entry.

Verified against real rows on ClickHouse: attributes_promoted holds a promoted
subset with its own tokenbf index; typed access and IS NOT NULL pruning work.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
0503672992 test(traces-qb): expand attribute-JSON coverage (agg, group-by, resolution)
Extend the attribute-JSON unit tests to the remaining functional requirements
on the JSON-on path (window fully after release):

- IN / NOT IN / BETWEEN operators.
- ColumnExpressionFor group-by (coerced to String) and aggregation (coerced to
  Float64), both exists-guarded.
- No-ambiguity-warning: a plain attribute filter must not warn, since the JSON
  column is a second physical home for one logical field, not a second field.
- Branch-flip guard: a data-type-unspecified attribute key still resolves to no
  column, keeping bare keys on the legacy CandidateKeys/synthesis path.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
Nikhil Soni
fae0f88450 feat(traces-qb): read span attributes from the JSON column (evolution-gated)
Resolve span attribute filters, group-bys and aggregations from the native
`attributes` JSON(max_dynamic_paths=0) column in addition to the legacy
attributes_string/number/bool maps, mirroring the resource/scope JSON handling.

Rollout is controlled entirely by the column-evolution entry: a key resolves to
the JSON column only once its evolution set names `attributes`. With no such
entry the key resolves to the Map column exactly as before, so default behaviour
is byte-for-byte unchanged (no existing golden test moves).

- getColumn returns [attributes, <map>] for a typed attribute key only when the
  attributes evolution is registered; data-type-unspecified keys keep the legacy
  CandidateKeys/synthesis path (no branch-flip of the common query shape).
- resolveColumnExprs renders a type-aware cast: String -> ::String (folds an
  absent path to '' for Map parity on negative operators), numeric/bool ->
  ::Nullable(T) (NULL on absent/type-mismatch, and GROUP BY-safe unlike Dynamic).
- Existence tests the raw path `attributes.`k` IS NOT NULL`, index-eligible via
  attributes_paths_tokenbf; ExistsExpression now quotes JSON paths with
  ClickHouseIdentifier so value and existence agree for keys with special chars.
- narrowEvolutionsToColumns drops the sibling-map evolution entries a typed key
  inherits from the `__all__` fetch, so its [attributes, <its map>] pair passes
  SelectEvolutionsForColumns.

Type preservation, flat dotted paths and index eligibility were verified against
real ingested rows on ClickHouse.

Assisted-by: Claude Opus 4.8
2026-09-02 01:55:54 +05:30
18 changed files with 1185 additions and 25 deletions

View File

@@ -466,16 +466,60 @@ func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
}, nil
}
// jsonAttributeMap returns the decoded document of a scanned JSON attributes column (the driver
// scans it into telemetrystoretypes.JSONValue) and whether one was present.
func jsonAttributeMap(v any) (map[string]any, bool) {
switch m := v.(type) {
case telemetrystoretypes.JSONValue:
if m == nil {
return nil, false
}
return m, true
case map[string]any:
if m == nil {
return nil, false
}
return m, true
default:
return nil, false
}
}
// flattenJSONPaths flattens a decoded JSON document into dotted keys (a nested {"http":{"route":x}}
// becomes "http.route": x), matching the legacy map representation so the attributes bag has the
// same flat shape whether it was read from the maps or the JSON column. Existing keys are
// overwritten, so a JSON path wins over a same-named map entry.
func flattenJSONPaths(prefix string, m map[string]any, out map[string]any) {
for k, v := range m {
key := k
if prefix != "" {
key = prefix + "." + k
}
switch child := v.(type) {
case map[string]any:
flattenJSONPaths(key, child, out)
case telemetrystoretypes.JSONValue:
flattenJSONPaths(key, child, out)
default:
out[key] = v
}
}
}
// mergeSpanAttributeColumns merges (attributes_string, attributes_number, attributes_bool, resources_string) into
// unified "attributes" and "resource" keys, and parses the stringified `events`
// and `links` columns into structured slices. Raw DB columns are removed.
//
// After the JSON rollout the attributes bag is read from the `attributes` JSON column instead of,
// or alongside, the legacy maps; those paths are flattened in and win over the maps on collision.
func mergeSpanAttributeColumns(data map[string]any) {
attrStr, hasStr := data["attributes_string"]
attrNum, hasNum := data["attributes_number"]
attrBool, hasBool := data["attributes_bool"]
attrJSON, hasJSON := jsonAttributeMap(data["attributes"])
// todo(nitya): move to resource json
resStr, hasRes := data["resources_string"]
if hasStr || hasNum || hasBool || hasRes {
if hasStr || hasNum || hasBool || hasJSON || hasRes {
attributes := make(map[string]any)
if m, ok := attrStr.(map[string]string); ok {
for k, v := range m {
@@ -492,6 +536,9 @@ func mergeSpanAttributeColumns(data map[string]any) {
attributes[k] = v
}
}
if hasJSON {
flattenJSONPaths("", attrJSON, attributes)
}
delete(data, "attributes_string")
delete(data, "attributes_number")
delete(data, "attributes_bool")

View File

@@ -195,3 +195,48 @@ func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
t.Fatalf("expected empty []spantypes.Link, got %#v", data["links"])
}
}
func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
t.Run("json only flattens nested paths and preserves types", func(t *testing.T) {
data := map[string]any{
"attributes": telemetrystoretypes.JSONValue{
"http": map[string]any{
"route": "/api/pay",
"retry": map[string]any{"count": float64(3)},
},
"cache.hit": true,
},
"resources_string": map[string]string{"service.name": "api"},
}
mergeSpanAttributeColumns(data)
attrs, ok := data["attributes"].(map[string]any)
require.True(t, ok, "attributes should be flattened to map[string]any, got %T", data["attributes"])
assert.Equal(t, "/api/pay", attrs["http.route"])
assert.Equal(t, float64(3), attrs["http.retry.count"])
assert.Equal(t, true, attrs["cache.hit"])
_, nested := attrs["http"]
assert.False(t, nested, "nested objects must be flattened away, not kept")
})
t.Run("straddle: json paths win over legacy map on collision, union otherwise", func(t *testing.T) {
data := map[string]any{
"attributes_string": map[string]string{"http.route": "/old", "only.map": "m"},
"attributes_number": map[string]float64{"http.status": 500},
"attributes": telemetrystoretypes.JSONValue{"http": map[string]any{"route": "/new"}},
"resources_string": map[string]string{"service.name": "api"},
}
mergeSpanAttributeColumns(data)
attrs := data["attributes"].(map[string]any)
assert.Equal(t, "/new", attrs["http.route"], "json home wins on collision")
assert.Equal(t, "m", attrs["only.map"], "map-only key survives")
assert.Equal(t, float64(500), attrs["http.status"], "number map key survives")
for _, removed := range []string{"attributes_string", "attributes_number", "attributes_bool"} {
_, present := data[removed]
assert.False(t, present, "%s should be removed", removed)
}
})
}

View File

@@ -43,7 +43,7 @@ func ExistsExpression(columns []*schema.Column, key *telemetrytypes.TelemetryFie
if len(evolutionsEntries) > 0 && evolutionsEntries[0] != nil {
columnName = evolutionsEntries[0].ColumnName
}
rawPath := fmt.Sprintf("%s.`%s`", columnName, key.Name)
rawPath := fmt.Sprintf("%s.%s", columnName, ClickHouseIdentifier(key.Name))
if exists {
return rawPath + " IS NOT NULL", nil
}

View File

@@ -0,0 +1,121 @@
package tracesstatementbuilder
import (
"context"
"regexp"
"strings"
"testing"
"time"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// jsonAttrColRe matches the raw `attributes` JSON column in the SELECT list (a bare column, not
// the `attributes_string`/`_number`/`_bool` maps).
var jsonAttrColRe = regexp.MustCompile(`,\s*attributes\s*,`)
func newBulkTestBuilder(t *testing.T, releaseTime time.Time) (*traceQueryStatementBuilder, *telemetrytypestest.MockMetadataStore) {
t.Helper()
fl := flaggertest.New(t)
fm := tracestelemetryschema.NewFieldMapper(fl)
cb := tracestelemetryschema.NewConditionBuilder(fm, fl)
store := telemetrytypestest.NewMockMetadataStore()
store.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
store.KeysMap["http.route"] = []*telemetrytypes.TelemetryFieldKey{{
Name: "http.route",
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
Signal: telemetrytypes.SignalTraces,
}}
store.ColumnEvolutionMetadataMap["traces:attribute:__all__"] = tracestelemetryschema.MockAttributeEvolutionData(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
b := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
store, fm, cb, aggExprRewriter, nil, fl, false, 100000,
)
return b, store
}
// TestBulkAttributeColumnsAcrossWindows asserts the empty-selectFields ("all fields") list query
// scans the attribute homes the column evolution resolves to for the window: the legacy maps before
// the JSON rollout, the JSON column after it, and both across it.
func TestBulkAttributeColumnsAcrossWindows(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)
cases := []struct {
name string
startMs uint64
endMs uint64
wantJSON bool
wantLegacyMap bool
}{
{"before rollout -> maps only", uint64(rel - 2*day), uint64(rel - day), false, true},
{"after rollout -> json only", uint64(rel + day), uint64(rel + 2*day), true, false},
{"straddle rollout -> both", uint64(rel - day), uint64(rel + day), true, true},
}
for _, tt := range cases {
t.Run(tt.name, func(t *testing.T) {
stmt, err := b.Build(
context.Background(), valuer.UUID{}, tt.startMs, tt.endMs,
qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{Signal: telemetrytypes.SignalTraces},
nil,
)
require.NoError(t, err)
selectList := stmt.Query[:strings.Index(stmt.Query, " FROM ")]
assert.Equal(t, tt.wantJSON, jsonAttrColRe.MatchString(stmt.Query),
"json `attributes` column presence; select=%s", selectList)
assert.Equal(t, tt.wantLegacyMap, strings.Contains(stmt.Query, "attributes_string"),
"legacy map presence; select=%s", selectList)
// resources_string stays a legacy map in every window (out of scope for this change).
assert.Contains(t, stmt.Query, "resources_string")
})
}
}
// TestGroupByAttributeAfterRolloutReadsJSON pins the post-dual-write guarantee at the statement
// level: a group-by on an attribute key in a window fully after the rollout reads the JSON column,
// never the legacy map — so aggregations keep working once map dual-write stops.
func TestGroupByAttributeAfterRolloutReadsJSON(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)
stmt, err := b.Build(
context.Background(), valuer.UUID{}, uint64(rel+day), uint64(rel+2*day),
qbtypes.RequestTypeTimeSeries,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "http.route",
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
}}},
Limit: 10,
},
nil,
)
require.NoError(t, err)
assert.Contains(t, stmt.Query, "attributes.`http.route`::String")
assert.NotContains(t, stmt.Query, "attributes_string", "post-rollout group-by must not read the legacy map")
}

View File

@@ -352,6 +352,30 @@ func lookupIntrinsicOrCalculatedField(name string) (telemetrytypes.TelemetryFiel
}
// buildListQuery builds a query for list panel type.
// bulkAttributeColumnNames returns the attribute columns the empty-selectFields list path scans:
// the physical homes the attributes column-evolution resolves to for [start, end] (legacy maps,
// the JSON column, or both across the rollout). resources_string is appended separately by the
// caller and stays a legacy map until the resource column moves to JSON.
func (b *traceQueryStatementBuilder) bulkAttributeColumnNames(ctx context.Context, start, end uint64) ([]string, error) {
evolutions, err := b.metadataStore.GetColumnEvolutions(ctx, []*telemetrytypes.EvolutionSelector{{
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldName: telemetrytypes.EvolutionFieldNameAll,
}})
if err != nil {
return nil, err
}
cols, err := tracestelemetryschema.BulkAttributeColumns(evolutions, start, end)
if err != nil {
return nil, err
}
names := make([]string, len(cols))
for i, c := range cols {
names[i] = c.Name
}
return names, nil
}
func (b *traceQueryStatementBuilder) buildListQuery(
ctx context.Context,
orgID valuer.UUID,
@@ -386,9 +410,14 @@ func (b *traceQueryStatementBuilder) buildListQuery(
}
if isSelectFieldsEmpty {
for _, col := range tracestelemetryschema.ContextualSpanColumns {
attrCols, err := b.bulkAttributeColumnNames(ctx, start, end)
if err != nil {
return nil, err
}
for _, col := range attrCols {
sb.SelectMore(col)
}
sb.SelectMore(tracestelemetryschema.SpanResourcesStringColumn)
}
// From table

View File

@@ -2407,6 +2407,13 @@ func (t *telemetryMetaStore) fetchMeterSourceMetricsTemporalityAndType(ctx conte
return temporalities, types, nil
}
func (k *telemetryMetaStore) GetColumnEvolutions(ctx context.Context, selectors []*telemetrytypes.EvolutionSelector) ([]*telemetrytypes.EvolutionEntry, error) {
if len(selectors) == 0 {
return nil, nil
}
return k.fetchEvolutionEntryFromClickHouse(ctx, selectors)
}
func (k *telemetryMetaStore) fetchEvolutionEntryFromClickHouse(ctx context.Context, selectors []*telemetrytypes.EvolutionSelector) ([]*telemetrytypes.EvolutionEntry, error) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select("signal", "column_name", "column_type", "field_context", "field_name", "version", "release_time")
@@ -2500,13 +2507,15 @@ func (k *telemetryMetaStore) updateColumnEvolutionMetadataForKeys(ctx context.Co
FieldContext: key.FieldContext,
FieldName: "__all__",
}
// first check if there is evolutions that with field name as __all__
if keyEvolutions, ok := evolutionsByUniqueKey[selector.QualifiedName()]; ok {
keysToUpdate[i].Evolutions = keyEvolutions
}
// then check for specific field name
// the per-field entries add to the column-wide ones, they don't replace them.
// NOTE: if a field evolved to its own column before an __all__ migration for the
// same signal+context, that later __all__ entry does not really apply to this field
// (the field had already moved). We ignore that case as it does not occur currently.
var keyEvolutions []*telemetrytypes.EvolutionEntry
keyEvolutions = append(keyEvolutions, evolutionsByUniqueKey[selector.QualifiedName()]...)
selector.FieldName = key.Name
if keyEvolutions, ok := evolutionsByUniqueKey[selector.QualifiedName()]; ok {
keyEvolutions = append(keyEvolutions, evolutionsByUniqueKey[selector.QualifiedName()]...)
if len(keyEvolutions) > 0 {
keysToUpdate[i].Evolutions = keyEvolutions
}
}

View File

@@ -40,10 +40,12 @@ const (
SpanIsRemoteColumn = "is_remote"
// Contextual Columns.
SpanAttributesStringColumn = "attributes_string"
SpanAttributesNumberColumn = "attributes_number"
SpanAttributesBoolColumn = "attributes_bool"
SpanResourcesStringColumn = "resources_string"
SpanAttributesStringColumn = "attributes_string"
SpanAttributesNumberColumn = "attributes_number"
SpanAttributesBoolColumn = "attributes_bool"
SpanAttributesColumn = "attributes"
SpanAttributesPromotedColumn = "attributes_promoted"
SpanResourcesStringColumn = "resources_string"
)
var (

View File

@@ -52,8 +52,10 @@ var (
KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString},
ValueType: schema.ColumnTypeString,
}},
"resource": {Name: "resource", Type: schema.JSONColumnType{}},
"scope": {Name: "scope", Type: schema.JSONColumnType{}},
"resource": {Name: "resource", Type: schema.JSONColumnType{}},
"scope": {Name: "scope", Type: schema.JSONColumnType{}},
"attributes": {Name: "attributes", Type: schema.JSONColumnType{}},
"attributes_promoted": {Name: "attributes_promoted", Type: schema.JSONColumnType{}},
"events": {Name: "events", Type: schema.ArrayColumnType{
ElementType: schema.ColumnTypeString,
@@ -184,16 +186,28 @@ func (m *fieldMapper) getColumn(
case telemetrytypes.FieldContextScope:
return []*schema.Column{indexV3Columns["scope"]}, nil
case telemetrytypes.FieldContextAttribute:
var mapCol *schema.Column
switch key.FieldDataType {
case telemetrytypes.FieldDataTypeString:
return []*schema.Column{indexV3Columns["attributes_string"]}, nil
mapCol = indexV3Columns["attributes_string"]
case telemetrytypes.FieldDataTypeInt64,
telemetrytypes.FieldDataTypeFloat64,
telemetrytypes.FieldDataTypeNumber:
return []*schema.Column{indexV3Columns["attributes_number"]}, nil
mapCol = indexV3Columns["attributes_number"]
case telemetrytypes.FieldDataTypeBool:
return []*schema.Column{indexV3Columns["attributes_bool"]}, nil
mapCol = indexV3Columns["attributes_bool"]
default:
return nil, qbtypes.ErrColumnNotFound
}
// The `attributes` evolution entry is the rollout control.
if attributeColumnEvolutionRegistered(key, SpanAttributesColumn) {
cols := make([]*schema.Column, 0, 3)
if attributeColumnEvolutionRegistered(key, SpanAttributesPromotedColumn) {
cols = append(cols, indexV3Columns["attributes_promoted"])
}
return append(cols, indexV3Columns["attributes"], mapCol), nil
}
return []*schema.Column{mapCol}, nil
case telemetrytypes.FieldContextSpan:
// Check if this is a span scope field
if strings.ToLower(key.Name) == SpanSearchScopeRoot || strings.ToLower(key.Name) == SpanSearchScopeEntryPoint {
@@ -309,6 +323,10 @@ func (m *fieldMapper) resolveColumnExprs(
exprs = append(exprs, fmt.Sprintf("%s.attributes.%s::String", columnName, querybuilder.ClickHouseIdentifier(attributeName)))
existExprs = append(existExprs, fmt.Sprintf("%s.attributes.%s IS NOT NULL", columnName, querybuilder.ClickHouseIdentifier(attributeName)))
}
case telemetrytypes.FieldContextAttribute:
path := fmt.Sprintf("%s.%s", columnName, querybuilder.ClickHouseIdentifier(key.Name))
exprs = append(exprs, fmt.Sprintf("%s::%s", path, attributeJSONCast(key.FieldDataType)))
existExprs = append(existExprs, fmt.Sprintf("%s IS NOT NULL", path))
default:
return nil, nil, nil, errors.NewInternalf(errors.CodeInternal, "only resource and scope context fields are supported for json columns, got %s", key.FieldContext.String)
}
@@ -353,6 +371,55 @@ func (m *fieldMapper) resolveColumnExprs(
return exprs, existExprs, columns, nil
}
// BulkAttributeColumns returns the physical attribute columns a whole-bag (list-view) read must
// scan over [startNs, endNs] given the attributes column-evolution entries: the three legacy maps
// before the JSON rollout, the JSON column after it, and both across the rollout — the same
// per-window home selection the per-key path uses, applied to the whole column. With no rollout
// entry it stays on the legacy maps.
func BulkAttributeColumns(evolutions []*telemetrytypes.EvolutionEntry, startNs, endNs uint64) ([]*schema.Column, error) {
maps := []*schema.Column{
indexV3Columns["attributes_string"],
indexV3Columns["attributes_number"],
indexV3Columns["attributes_bool"],
}
if len(evolutions) == 0 {
return maps, nil
}
family := append([]*schema.Column{indexV3Columns["attributes"]}, maps...)
cols, _, err := qbtypes.SelectEvolutionsForColumns(family, evolutions, startNs, endNs)
if err != nil {
return nil, err
}
return cols, nil
}
// attributeColumnEvolutionRegistered reports whether key carries an evolution entry for the given column.
func attributeColumnEvolutionRegistered(key *telemetrytypes.TelemetryFieldKey, columnName string) bool {
for _, e := range key.Evolutions {
if e != nil && e.ColumnName == columnName {
return true
}
}
return false
}
// attributeJSONCast returns the ClickHouse cast target for a span attribute read from the
// JSON column. String (and data-type-unspecified) casts to non-nullable String so an absent
// path folds to ” the way the Map column's default does, keeping negative-operator parity;
// numeric and bool cast to Nullable so an absent or wrong-typed path reads NULL instead of a
// spurious 0/false. GROUP BY accepts these Nullable scalar casts (unlike a raw Dynamic).
func attributeJSONCast(dataType telemetrytypes.FieldDataType) string {
switch dataType {
case telemetrytypes.FieldDataTypeInt64,
telemetrytypes.FieldDataTypeFloat64,
telemetrytypes.FieldDataTypeNumber,
telemetrytypes.FieldDataTypeBool:
return fmt.Sprintf("Nullable(%s)", telemetrytypes.MappingFieldDataTypeToJSONDataType[dataType].StringValue())
default:
return "String"
}
}
// upgradeToFamilies swaps single-member candidates for their family when the
// metadata map proves membership. Candidate order and every non-family
// candidate stay exactly as the legacy flow produced them; sibling candidates

View File

@@ -0,0 +1,337 @@
package tracestelemetryschema
import (
"context"
"testing"
"time"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
var (
attrJSONRelease = time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
attrWindowBefore = [2]uint64{tsNano(2024, 1), tsNano(2024, 6)}
attrWindowAfter = [2]uint64{tsNano(2025, 6), tsNano(2025, 7)}
attrWindowStraddle = [2]uint64{tsNano(2024, 6), tsNano(2025, 6)}
)
func tsNano(y int, m time.Month) uint64 {
return uint64(time.Date(y, m, 1, 0, 0, 0, 0, time.UTC).UnixNano())
}
func attrKey(name string, dt telemetrytypes.FieldDataType, evo []*telemetrytypes.EvolutionEntry) telemetrytypes.TelemetryFieldKey {
return telemetrytypes.TelemetryFieldKey{
Name: name,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: dt,
Evolutions: evo,
}
}
// TestFieldForAttributeJSONEvolution asserts the value expression across the rollout window:
// before release the legacy Map lookup (byte-for-byte today), after release the type-aware JSON
// cast, straddling a dual-read multiIf with the JSON column first.
func TestFieldForAttributeJSONEvolution(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
testCases := []struct {
name string
dataType telemetrytypes.FieldDataType
window [2]uint64
expected string
}{
{"string before -> map", telemetrytypes.FieldDataTypeString, attrWindowBefore, "attributes_string['user.id']"},
{"string after -> json", telemetrytypes.FieldDataTypeString, attrWindowAfter, "attributes.`user.id`::String"},
{"string straddle -> dual", telemetrytypes.FieldDataTypeString, attrWindowStraddle, "multiIf(attributes.`user.id` IS NOT NULL, attributes.`user.id`::String, mapContains(attributes_string, 'user.id'), attributes_string['user.id'], NULL)"},
{"number before -> map", telemetrytypes.FieldDataTypeNumber, attrWindowBefore, "attributes_number['user.id']"},
{"number after -> json", telemetrytypes.FieldDataTypeNumber, attrWindowAfter, "attributes.`user.id`::Nullable(Float64)"},
{"number straddle -> dual", telemetrytypes.FieldDataTypeNumber, attrWindowStraddle, "multiIf(attributes.`user.id` IS NOT NULL, attributes.`user.id`::Nullable(Float64), mapContains(attributes_number, 'user.id'), attributes_number['user.id'], NULL)"},
{"int64 after -> json", telemetrytypes.FieldDataTypeInt64, attrWindowAfter, "attributes.`user.id`::Nullable(Int64)"},
{"bool before -> map", telemetrytypes.FieldDataTypeBool, attrWindowBefore, "attributes_bool['user.id']"},
{"bool after -> json", telemetrytypes.FieldDataTypeBool, attrWindowAfter, "attributes.`user.id`::Nullable(Bool)"},
{"bool straddle -> dual", telemetrytypes.FieldDataTypeBool, attrWindowStraddle, "multiIf(attributes.`user.id` IS NOT NULL, attributes.`user.id`::Nullable(Bool), mapContains(attributes_bool, 'user.id'), attributes_bool['user.id'], NULL)"},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
key := attrKey("user.id", tc.dataType, evo)
got, err := fm.FieldFor(ctx, valuer.UUID{}, tc.window[0], tc.window[1], &key)
require.NoError(t, err)
assert.Equal(t, tc.expected, got)
})
}
}
// TestFieldForAttributeNoEvolutionParity proves the JSON column is untouched until the evolution
// entry is registered: a key with no evolutions resolves to the Map column for every window.
func TestFieldForAttributeNoEvolutionParity(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
for _, dt := range []struct {
dataType telemetrytypes.FieldDataType
expected string
}{
{telemetrytypes.FieldDataTypeString, "attributes_string['user.id']"},
{telemetrytypes.FieldDataTypeNumber, "attributes_number['user.id']"},
{telemetrytypes.FieldDataTypeBool, "attributes_bool['user.id']"},
} {
key := attrKey("user.id", dt.dataType, nil)
got, err := fm.FieldFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &key)
require.NoError(t, err)
assert.Equal(t, dt.expected, got, "no evolution entry must keep the Map path")
}
}
// TestConditionForAttributeJSON asserts the emitted WHERE fragment per operator against the JSON
// column (window fully after release). Positive operators carry the raw-path existence guard;
// numeric comparisons keep numeric semantics; existence never tests the ::String cast.
func TestConditionForAttributeJSON(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
cb := NewConditionBuilder(fm, flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
testCases := []struct {
name string
key telemetrytypes.TelemetryFieldKey
operator qbtypes.FilterOperator
value any
expected string
}{
{
name: "equal string",
key: attrKey("user.id", telemetrytypes.FieldDataTypeString, evo),
operator: qbtypes.FilterOperatorEqual, value: "admin",
expected: "(attributes.`user.id`::String = ? AND attributes.`user.id` IS NOT NULL)",
},
{
name: "not equal string has no exists guard",
key: attrKey("user.id", telemetrytypes.FieldDataTypeString, evo),
operator: qbtypes.FilterOperatorNotEqual, value: "admin",
expected: "attributes.`user.id`::String <> ?",
},
{
name: "greater than number",
key: attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo),
operator: qbtypes.FilterOperatorGreaterThan, value: float64(200),
expected: "toFloat64(attributes.`http.status_code`::Nullable(Int64)) > ?",
},
{
name: "ilike string",
key: attrKey("user.id", telemetrytypes.FieldDataTypeString, evo),
operator: qbtypes.FilterOperatorILike, value: "%adm%",
expected: "LOWER(attributes.`user.id`::String) LIKE LOWER(?)",
},
{
name: "exists uses raw path",
key: attrKey("user.id", telemetrytypes.FieldDataTypeString, evo),
operator: qbtypes.FilterOperatorExists, value: nil,
expected: "attributes.`user.id` IS NOT NULL",
},
{
name: "not exists uses raw path",
key: attrKey("user.id", telemetrytypes.FieldDataTypeString, evo),
operator: qbtypes.FilterOperatorNotExists, value: nil,
expected: "attributes.`user.id` IS NULL",
},
{
name: "in string",
key: attrKey("user.id", telemetrytypes.FieldDataTypeString, evo),
operator: qbtypes.FilterOperatorIn, value: []any{"a", "b"},
expected: "((attributes.`user.id`::String = ? OR attributes.`user.id`::String = ?) AND attributes.`user.id` IS NOT NULL)",
},
{
name: "not in string has no exists guard",
key: attrKey("user.id", telemetrytypes.FieldDataTypeString, evo),
operator: qbtypes.FilterOperatorNotIn, value: []any{"a", "b"},
expected: "(attributes.`user.id`::String <> ? AND attributes.`user.id`::String <> ?)",
},
{
name: "between number",
key: attrKey("latency", telemetrytypes.FieldDataTypeNumber, evo),
operator: qbtypes.FilterOperatorBetween, value: []any{float64(1), float64(9)},
expected: "toFloat64(attributes.`latency`::Nullable(Float64)) BETWEEN ? AND ?",
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &tc.key,
map[string][]*telemetrytypes.TelemetryFieldKey{tc.key.Name: {&tc.key}}, qbtypes.ConditionBuilderOptions{}, tc.operator, tc.value, sb)
require.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
assert.Contains(t, sql, tc.expected)
})
}
}
// TestConditionForAttributeJSONNotExistsDualRead covers NOT EXISTS across both homes during the
// dual-read window: it must AND the JSON IS NULL with NOT mapContains so a row present in either
// home is excluded (De Morgan), including rows that predate the JSON column.
func TestConditionForAttributeJSONNotExistsDualRead(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
cb := NewConditionBuilder(fm, flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
key := attrKey("user.id", telemetrytypes.FieldDataTypeString, evo)
sb := sqlbuilder.NewSelectBuilder()
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, attrWindowStraddle[0], attrWindowStraddle[1], &key,
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotExists, nil, sb)
require.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
// the value multiIf resolves the row's home; NOT EXISTS negates the whole thing to IS NULL
assert.Contains(t, sql, "IS NULL")
assert.Contains(t, sql, "attributes.`user.id` IS NOT NULL")
assert.Contains(t, sql, "mapContains(attributes_string, 'user.id')")
}
// TestColumnExpressionForAttributeJSON covers group-by (coerced to String) and aggregation
// (coerced to Float64) over a JSON attribute after release: both are exists-guarded so an absent
// path is NULL rather than a spurious ”/0, and the numeric branch keeps its toFloat64 coercion.
func TestColumnExpressionForAttributeJSON(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
t.Run("group by string", func(t *testing.T) {
key := attrKey("user.id", telemetrytypes.FieldDataTypeString, evo)
got, err := fm.ColumnExpressionFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &key, telemetrytypes.FieldDataTypeString, nil)
require.NoError(t, err)
assert.Equal(t, "multiIf(attributes.`user.id` IS NOT NULL, attributes.`user.id`::String, NULL)", got)
})
t.Run("aggregation numeric", func(t *testing.T) {
key := attrKey("latency", telemetrytypes.FieldDataTypeNumber, evo)
got, err := fm.ColumnExpressionFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &key, telemetrytypes.FieldDataTypeFloat64, nil)
require.NoError(t, err)
assert.Equal(t, "multiIf(attributes.`latency` IS NOT NULL, toFloat64(attributes.`latency`::Nullable(Float64)), NULL)", got)
})
}
// TestAttributeJSONNoAmbiguityWarning guards against a visible regression: the JSON column is a
// second physical home for the same logical field, not a second logical field, so a plain
// attribute filter must not emit the "ambiguous key" warning.
func TestAttributeJSONNoAmbiguityWarning(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
cb := NewConditionBuilder(fm, flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
key := attrKey("user.id", telemetrytypes.FieldDataTypeString, evo)
sb := sqlbuilder.NewSelectBuilder()
_, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &key,
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "x", sb)
require.NoError(t, err)
assert.Empty(t, warnings, "a plain attribute filter must not emit an ambiguity warning")
}
// TestConditionForAttributeJSONTypeCollision covers a name stored under two data types (String
// and Int64) in the JSON column: an untyped filter fans out to one exists-guarded condition per
// type, both reading the same physical path with their own cast, and surfaces the ambiguity
// warning. In the JSON column the two branches share the raw path; only the cast differs.
func TestConditionForAttributeJSONTypeCollision(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
cb := NewConditionBuilder(fm, flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
strKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeString, evo)
intKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
"http.status_code": {&strKey, &intKey},
}
ref := attrKey("http.status_code", telemetrytypes.FieldDataTypeUnspecified, nil)
sb := sqlbuilder.NewSelectBuilder()
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &ref,
fieldKeys, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, float64(200), sb)
require.NoError(t, err)
require.Len(t, conds, 2, "a colliding name must build one condition per data type")
sb.Where(sb.Or(conds...))
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
assert.Contains(t, sql, "toFloat64OrNull(attributes.`http.status_code`::String) = ?")
assert.Contains(t, sql, "toFloat64(attributes.`http.status_code`::Nullable(Int64)) = ?")
assert.Contains(t, sql, "attributes.`http.status_code` IS NOT NULL")
assert.NotEmpty(t, warnings, "a colliding name must surface the ambiguity warning")
}
// TestColumnExpressionForAttributeJSONTypeCollision covers group-by on a name stored under two
// data types: both interpretations fold into a single multiIf output column, each guarded on the
// shared raw path and coerced to the group-by type (String rendered directly, Int64 via toString).
func TestColumnExpressionForAttributeJSONTypeCollision(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
strKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeString, evo)
intKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
"http.status_code": {&strKey, &intKey},
}
ref := attrKey("http.status_code", telemetrytypes.FieldDataTypeUnspecified, nil)
got, err := fm.ColumnExpressionFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &ref, telemetrytypes.FieldDataTypeString, fieldKeys)
require.NoError(t, err)
assert.Equal(t,
"multiIf(attributes.`http.status_code` IS NOT NULL, attributes.`http.status_code`::String, attributes.`http.status_code` IS NOT NULL, toString(attributes.`http.status_code`::Nullable(Int64)), NULL)",
got)
}
// TestConditionForAttributeMapTypeCollisionParity anchors the legacy behavior the JSON path must
// preserve: before the rollout the same colliding name fans out to two separate physical map
// columns (attributes_string / attributes_number), each with its own mapContains guard.
func TestConditionForAttributeMapTypeCollisionParity(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
cb := NewConditionBuilder(fm, flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
strKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeString, evo)
intKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
"http.status_code": {&strKey, &intKey},
}
ref := attrKey("http.status_code", telemetrytypes.FieldDataTypeUnspecified, nil)
sb := sqlbuilder.NewSelectBuilder()
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, attrWindowBefore[0], attrWindowBefore[1], &ref,
fieldKeys, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, float64(200), sb)
require.NoError(t, err)
require.Len(t, conds, 2, "a colliding name must build one condition per data type")
sb.Where(sb.Or(conds...))
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
assert.Contains(t, sql, "toFloat64OrNull(attributes_string['http.status_code']) = ?")
assert.Contains(t, sql, "mapContains(attributes_string, 'http.status_code')")
assert.Contains(t, sql, "toFloat64(attributes_number['http.status_code']) = ?")
assert.Contains(t, sql, "mapContains(attributes_number, 'http.status_code')")
}
// TestColumnForUnspecifiedAttributeNoBranchFlip pins the branch-flip decision: a
// data-type-unspecified attribute key resolves to no column (even with the evolution present), so
// bare attribute keys keep taking the legacy CandidateKeys/synthesis path rather than becoming
// metadata-first resolvable.
func TestColumnForUnspecifiedAttributeNoBranchFlip(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
evo := MockAttributeEvolutionData(attrJSONRelease)
key := attrKey("user.id", telemetrytypes.FieldDataTypeUnspecified, evo)
_, err := fm.ColumnFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &key)
assert.ErrorIs(t, err, qbtypes.ErrColumnNotFound)
}

View File

@@ -0,0 +1,104 @@
package tracestelemetryschema
import (
"context"
"testing"
"time"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
var (
promoJSONRelease = time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
promoPromoRelease = time.Date(2025, 6, 1, 0, 0, 0, 0, time.UTC)
)
// TestFieldForAttributePromotedEvolution proves promotion is just a third evolution column:
// evolution selection reads a single physical home per window — the legacy Map before the JSON
// rollout, `attributes` between the JSON rollout and the path's promotion, and
// `attributes_promoted` alone after promotion — fanning out only across an evolution boundary.
func TestFieldForAttributePromotedEvolution(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
evo := MockPromotedAttributeEvolutionData("span.operation", promoJSONRelease, promoPromoRelease)
win := func(from, to string) [2]uint64 {
a, _ := time.Parse("2006-01-02", from)
b, _ := time.Parse("2006-01-02", to)
return [2]uint64{uint64(a.UnixNano()), uint64(b.UnixNano())}
}
testCases := []struct {
name string
window [2]uint64
expected string
}{
{"before json rollout -> map", win("2024-01-01", "2024-06-01"), "attributes_string['span.operation']"},
{"between json and promotion -> attributes", win("2025-02-01", "2025-04-01"), "attributes.`span.operation`::String"},
{"after promotion -> promoted only", win("2025-07-01", "2025-08-01"), "attributes_promoted.`span.operation`::String"},
{"straddle promotion -> attributes_promoted + attributes", win("2025-04-01", "2025-08-01"), "multiIf(attributes_promoted.`span.operation` IS NOT NULL, attributes_promoted.`span.operation`::String, attributes.`span.operation` IS NOT NULL, attributes.`span.operation`::String, NULL)"},
{"straddle json rollout -> attributes + map", win("2024-06-01", "2025-03-01"), "multiIf(attributes.`span.operation` IS NOT NULL, attributes.`span.operation`::String, mapContains(attributes_string, 'span.operation'), attributes_string['span.operation'], NULL)"},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
key := telemetrytypes.TelemetryFieldKey{
Name: "span.operation",
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
Evolutions: evo,
}
got, err := fm.FieldFor(ctx, valuer.UUID{}, tc.window[0], tc.window[1], &key)
require.NoError(t, err)
assert.Equal(t, tc.expected, got)
})
}
}
// TestConditionForAttributePromoted asserts a filter over a window fully after promotion reads
// only the promoted column, with existence testing the promoted raw path (index-eligible via
// attributes_promoted_paths_tokenbf) — not the attributes column.
func TestConditionForAttributePromoted(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
cb := NewConditionBuilder(fm, flaggertest.New(t))
evo := MockPromotedAttributeEvolutionData("span.operation", promoJSONRelease, promoPromoRelease)
afterPromo := [2]uint64{
uint64(time.Date(2025, 7, 1, 0, 0, 0, 0, time.UTC).UnixNano()),
uint64(time.Date(2025, 8, 1, 0, 0, 0, 0, time.UTC).UnixNano()),
}
key := telemetrytypes.TelemetryFieldKey{
Name: "span.operation",
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
Evolutions: evo,
}
t.Run("equal reads promoted column only", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, afterPromo[0], afterPromo[1], &key,
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "GET", sb)
require.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
assert.Contains(t, sql, "(attributes_promoted.`span.operation`::String = ? AND attributes_promoted.`span.operation` IS NOT NULL)")
assert.NotContains(t, sql, "attributes.`span.operation`")
})
t.Run("exists uses promoted raw path", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, afterPromo[0], afterPromo[1], &key,
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorExists, nil, sb)
require.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
assert.Contains(t, sql, "attributes_promoted.`span.operation` IS NOT NULL")
})
}

View File

@@ -154,6 +154,34 @@ func BuildCompleteFieldKeyMap(releaseTime time.Time) map[string][]*telemetrytype
return keysMap
}
// MockAttributeEvolutionData returns the attribute-context evolution timeline: only the JSON
// `attributes` migration released at releaseTime, field_name "__all__". The legacy map columns
// are the epoch-0 base and are not stored as evolution rows; SelectEvolutionsForColumns
// synthesizes the base entry for whichever typed map getColumn resolves the key to.
func MockAttributeEvolutionData(releaseTime time.Time) []*telemetrytypes.EvolutionEntry {
return []*telemetrytypes.EvolutionEntry{
{
Signal: telemetrytypes.SignalTraces,
ColumnName: "attributes",
ColumnType: "JSON()",
FieldContext: telemetrytypes.FieldContextAttribute,
FieldName: "__all__",
ReleaseTime: releaseTime,
},
}
}
// MockPromotedAttributeEvolutionData returns a promoted attribute's evolution timeline: the JSON
// `attributes` column at jsonRelease (field_name "__all__") and the per-path `attributes_promoted`
// column at promoteRelease (field_name = path). The legacy map is the synthesized epoch-0 base and
// is not stored as an evolution row.
func MockPromotedAttributeEvolutionData(path string, jsonRelease, promoteRelease time.Time) []*telemetrytypes.EvolutionEntry {
return []*telemetrytypes.EvolutionEntry{
{Signal: telemetrytypes.SignalTraces, ColumnName: "attributes", ColumnType: "JSON()", FieldContext: telemetrytypes.FieldContextAttribute, FieldName: "__all__", ReleaseTime: jsonRelease},
{Signal: telemetrytypes.SignalTraces, ColumnName: "attributes_promoted", ColumnType: "JSON()", FieldContext: telemetrytypes.FieldContextAttribute, FieldName: path, ReleaseTime: promoteRelease},
}
}
// MockEvolutionData returns the canonical resource-column evolution timeline used in tests:
// the legacy resources_string map at epoch 0 and the JSON resource column released at releaseTime.
func MockEvolutionData(releaseTime time.Time) []*telemetrytypes.EvolutionEntry {

View File

@@ -23,6 +23,18 @@ func SelectEvolutionsForColumns(columns []*schema.Column, evolutions []*telemetr
return columns, nil, nil
}
// Derive the base column from the candidate columns.
seen := make(map[string]struct{}, len(evolutions))
for _, e := range evolutions {
seen[e.ColumnName] = struct{}{}
}
for _, c := range columns {
if _, ok := seen[c.Name]; ok {
continue
}
evolutions = append(evolutions, &telemetrytypes.EvolutionEntry{ColumnName: c.Name, ReleaseTime: time.Unix(0, 0)})
}
sortedEvolutions := make([]*telemetrytypes.EvolutionEntry, len(evolutions))
copy(sortedEvolutions, evolutions)

View File

@@ -4,6 +4,10 @@ import (
"time"
)
// EvolutionFieldNameAll is the field_name of a column-wide (whole-signal-context) evolution
// entry, as opposed to a per-field promotion entry.
const EvolutionFieldNameAll = "__all__"
type EvolutionEntry struct {
Signal Signal `json:"signal"`
ColumnName string `json:"column_name"`

View File

@@ -19,6 +19,11 @@ type MetadataStore interface {
// GetKey returns a list of keys with the given name.
GetKey(ctx context.Context, orgID valuer.UUID, fieldKeySelector *FieldKeySelector) ([]*TelemetryFieldKey, error)
// GetColumnEvolutions returns the column-evolution entries matching the selectors (including
// __all__ family rows) without resolving concrete keys, so a whole-column read such as the
// list-view attributes bag can pick its physical homes for the query window.
GetColumnEvolutions(ctx context.Context, selectors []*EvolutionSelector) ([]*EvolutionEntry, error)
// GetRelatedValues returns a list of related values for the given key name
// and the existing selection of keys.
GetRelatedValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *FieldValueSelector) ([]string, bool, error)

View File

@@ -163,6 +163,14 @@ func (m *MockMetadataStore) GetKey(ctx context.Context, _ valuer.UUID, fieldKeyS
return result, nil
}
func (m *MockMetadataStore) GetColumnEvolutions(_ context.Context, selectors []*telemetrytypes.EvolutionSelector) ([]*telemetrytypes.EvolutionEntry, error) {
var evolutions []*telemetrytypes.EvolutionEntry
for _, selector := range selectors {
evolutions = append(evolutions, m.ColumnEvolutionMetadataMap[selector.QualifiedName()]...)
}
return evolutions, nil
}
// GetRelatedValues returns a list of related values for the given key name and selection.
func (m *MockMetadataStore) GetRelatedValues(ctx context.Context, _ valuer.UUID, fieldValueSelector *telemetrytypes.FieldValueSelector) ([]string, bool, error) {
if fieldValueSelector == nil {
@@ -396,14 +404,15 @@ func (m *MockMetadataStore) updateColumnEvolutionMetadataForKeys(_ context.Conte
FieldContext: selector.FieldContext,
FieldName: "__all__",
}
key := sel.QualifiedName()
if entries, exists := m.ColumnEvolutionMetadataMap[key]; exists {
result[key] = entries
}
// column-wide (__all__) homes plus this field's own homes, appended not replaced,
// mirroring the real store
var evolutions []*telemetrytypes.EvolutionEntry
evolutions = append(evolutions, m.ColumnEvolutionMetadataMap[sel.QualifiedName()]...)
sel.FieldName = metadataKeySelectors[i].FieldName
key = sel.QualifiedName()
if entries, exists := m.ColumnEvolutionMetadataMap[key]; exists {
result[key] = entries
evolutions = append(evolutions, m.ColumnEvolutionMetadataMap[sel.QualifiedName()]...)
if len(evolutions) > 0 {
keysToUpdate[i].Evolutions = evolutions
result[sel.QualifiedName()] = evolutions
}
}
return result

View File

@@ -0,0 +1,76 @@
package telemetrytypestest
import (
"context"
"testing"
"time"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestEvolutionAppendsPerFieldToColumnWide covers the metadata enrichment: a key's column-wide
// (__all__) evolution homes and its per-field homes are appended, not replaced. A promoted
// attribute (whose attributes_promoted entry lives under its own field name) must therefore keep
// its Map and base-JSON homes for time ranges before it was promoted.
func TestEvolutionAppendsPerFieldToColumnWide(t *testing.T) {
jsonRel := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
promoRel := time.Date(2025, 6, 1, 0, 0, 0, 0, time.UTC)
mk := func(col, field string, rt time.Time) *telemetrytypes.EvolutionEntry {
return &telemetrytypes.EvolutionEntry{
Signal: telemetrytypes.SignalTraces, ColumnName: col,
FieldContext: telemetrytypes.FieldContextAttribute, FieldName: field, ReleaseTime: rt,
}
}
columnNames := func(entries []*telemetrytypes.EvolutionEntry) []string {
out := make([]string, 0, len(entries))
for _, e := range entries {
out = append(out, e.ColumnName)
}
return out
}
// Only JSON columns are recorded as evolution rows; the legacy Map column is the
// synthesized epoch-0 base and is not stored here.
newStore := func() *MockMetadataStore {
store := NewMockMetadataStore()
store.ColumnEvolutionMetadataMap["traces:attribute:__all__"] = []*telemetrytypes.EvolutionEntry{
mk("attributes", "__all__", jsonRel),
}
return store
}
resolve := func(t *testing.T, store *MockMetadataStore, name string) *telemetrytypes.TelemetryFieldKey {
t.Helper()
key := &telemetrytypes.TelemetryFieldKey{
Name: name, Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute, FieldDataType: telemetrytypes.FieldDataTypeString,
}
store.KeysMap[name] = []*telemetrytypes.TelemetryFieldKey{key}
selector := &telemetrytypes.FieldKeySelector{
Name: name, Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute, SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
}
_, _, err := store.GetKeysMulti(context.Background(), valuer.UUID{}, []*telemetrytypes.FieldKeySelector{selector})
require.NoError(t, err)
return key
}
t.Run("promoted key keeps the column-wide attributes home and gains the promoted column", func(t *testing.T) {
store := newStore()
store.ColumnEvolutionMetadataMap["traces:attribute:span.operation"] = []*telemetrytypes.EvolutionEntry{
mk("attributes_promoted", "span.operation", promoRel),
}
key := resolve(t, store, "span.operation")
assert.ElementsMatch(t, []string{"attributes", "attributes_promoted"}, columnNames(key.Evolutions))
})
t.Run("non-promoted key gets only the column-wide home", func(t *testing.T) {
store := newStore()
key := resolve(t, store, "user.id")
assert.ElementsMatch(t, []string{"attributes"}, columnNames(key.Evolutions))
})
}

View File

@@ -292,6 +292,7 @@ class Traces(ABC):
events: list[dict[str, Any]]
links: list[dict[str, Any]]
resource_json: dict[str, str]
attributes_json: dict[str, Any]
response_status_code: str
external_http_url: str
http_url: str
@@ -330,6 +331,7 @@ class Traces(ABC):
flags: np.uint32 = 0,
scope: dict[str, Any] = {},
resource_write_mode: Literal["legacy_only", "dual_write"] = "dual_write",
attribute_write_mode: Literal["legacy_only", "dual_write"] = "dual_write",
) -> None:
if timestamp is None:
timestamp = datetime.datetime.now()
@@ -510,6 +512,11 @@ class Traces(ABC):
)
)
# Spans before the attribute JSON-evolution time populate only the legacy
# attributes_{string,number,bool} maps; spans at or after it dual-write the
# native-typed `attributes` JSON column too.
self.attributes_json = {} if attribute_write_mode == "legacy_only" else dict(attributes)
# Process events and derive error events. self.events holds the parsed
# response shape; np_arr() encodes back to the DB format on insert.
self.events = []
@@ -689,6 +696,7 @@ class Traces(ABC):
self.is_remote,
self.resource_json,
self.scope_json,
self.attributes_json,
],
dtype=object,
)
@@ -860,6 +868,7 @@ def insert_traces_to_clickhouse(conn, traces: list[Traces]) -> None:
"is_remote",
"resource",
"scope",
"attributes",
],
data=[trace.np_arr() for trace in traces],
)
@@ -923,6 +932,37 @@ def insert_traces(
)
def insert_attribute_evolution_to_clickhouse(conn, signal: str, release_time: datetime.datetime) -> None:
"""Seed the `attributes` JSON column-evolution row for a signal at release_time. Unlike the
resource row (seeded by the migrator at install), the attribute JSON rollout is install-specific
and not migrator-seeded, so tests insert it to gate map-vs-JSON resolution across a window."""
conn.command(
"""
INSERT INTO signoz_metadata.distributed_column_evolution_metadata
(signal, column_name, column_type, field_context, field_name, version, release_time)
VALUES (%(signal)s, 'attributes', 'JSON()', 'attribute', '__all__', 1, %(release_time_ns)s)
""",
parameters={"signal": signal, "release_time_ns": int(release_time.timestamp() * 1e9)},
)
@pytest.fixture(name="seed_attribute_evolution", scope="function")
def seed_attribute_evolution(
clickhouse: types.TestContainerClickhouse,
) -> Generator[Callable[[str, datetime.datetime], None], Any]:
def _seed(signal: str, release_time: datetime.datetime) -> None:
insert_attribute_evolution_to_clickhouse(clickhouse.conn, signal, release_time)
yield _seed
cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
clickhouse.conn.query(
f"ALTER TABLE signoz_metadata.column_evolution_metadata ON CLUSTER '{cluster}' "
"DELETE WHERE column_name = 'attributes' AND field_context = 'attribute' AND field_name = '__all__' "
"SETTINGS mutations_sync = 1"
)
@pytest.fixture(name="insert_top_level_operations", scope="function")
def insert_top_level_operations(
clickhouse: types.TestContainerClickhouse,

View File

@@ -0,0 +1,225 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.querier import (
RequestType,
assert_grouped_series,
build_aggregation,
build_group_by_field,
build_traces_scalar_query,
index_series_by_label,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces
def test_traces_attributes_json_evolution(
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:
"""`http.route` is a dotted key, so the `attributes` JSON column nests it under the path
http.route while the legacy attributes_string map keys it verbatim. Spans before the attribute
JSON-evolution time write only the map; spans at or after it dual-write the JSON column too. A
query window resolves the attribute to the map (before), the JSON nested path (after), or a
map+JSON multiIf (straddling), and must return identical rows across the boundary."""
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_2 = evolution_time - timedelta(minutes=10)
before_1 = evolution_time - timedelta(minutes=5)
after_1 = evolution_time + timedelta(minutes=5)
after_2 = evolution_time + timedelta(minutes=10)
insert_traces(
[
Traces(
timestamp=before_2,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name="before 2",
attributes={"http.route": "/d"},
attribute_write_mode="legacy_only",
),
Traces(
timestamp=before_1,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name="before 1",
attributes={"http.route": "/c"},
attribute_write_mode="legacy_only",
),
Traces(
timestamp=after_1,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name="after 1",
attributes={"http.route": "/a", "http.retry.count": 5, "http.cache.hit": True},
attribute_write_mode="dual_write",
),
Traces(
timestamp=after_2,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name="after 2",
attributes={"http.route": "/b", "http.retry.count": 1, "http.cache.hit": False},
attribute_write_mode="dual_write",
),
]
)
# before window -> map-only resolution
response = make_query_request(
signoz,
token,
start_ms=int((before_2 - timedelta(minutes=1)).timestamp() * 1000),
end_ms=int((before_1 + timedelta(minutes=1)).timestamp() * 1000),
request_type=RequestType.TIME_SERIES,
queries=[
build_traces_scalar_query(
aggregations=[build_aggregation("count()")],
group_by=[build_group_by_field("http.route", field_context="attribute")],
)
],
)
assert response.status_code == HTTPStatus.OK
before_series = index_series_by_label(
response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"], "http.route"
)
assert_grouped_series(
before_series,
expected_values_by_group={
"/d": {int(before_2.timestamp() * 1000): 1},
"/c": {int(before_1.timestamp() * 1000): 1},
},
)
# after window -> JSON-only resolution (nested path)
response = make_query_request(
signoz,
token,
start_ms=int((after_1 - timedelta(minutes=1)).timestamp() * 1000),
end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
request_type=RequestType.TIME_SERIES,
queries=[
build_traces_scalar_query(
aggregations=[build_aggregation("count()")],
group_by=[build_group_by_field("http.route", field_context="attribute")],
)
],
)
assert response.status_code == HTTPStatus.OK
after_series = index_series_by_label(
response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"], "http.route"
)
assert_grouped_series(
after_series,
expected_values_by_group={
"/a": {int(after_1.timestamp() * 1000): 1},
"/b": {int(after_2.timestamp() * 1000): 1},
},
)
# straddling window -> map + JSON multiIf resolution
response = make_query_request(
signoz,
token,
start_ms=int(before_2.timestamp() * 1000),
end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
request_type=RequestType.TIME_SERIES,
queries=[
build_traces_scalar_query(
aggregations=[build_aggregation("count()")],
group_by=[build_group_by_field("http.route", field_context="attribute")],
)
],
)
assert response.status_code == HTTPStatus.OK
spanning_series = index_series_by_label(
response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"], "http.route"
)
assert_grouped_series(
spanning_series,
expected_values_by_group={
"/d": {int(before_2.timestamp() * 1000): 1},
"/c": {int(before_1.timestamp() * 1000): 1},
"/a": {int(after_1.timestamp() * 1000): 1},
"/b": {int(after_2.timestamp() * 1000): 1},
},
)
def test_traces_attributes_json_typed_filters(
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:
"""In the JSON-only window each dotted attribute reads through the nested path with its native
cast: string (::String), Int64 (toFloat64(...::Nullable(Float64))), Bool (::Nullable(Bool)),
and existence via the raw path. Filters must select the same rows the Map path would."""
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)
hit = evolution_time + timedelta(minutes=5)
miss = evolution_time + timedelta(minutes=6)
insert_traces(
[
Traces(
timestamp=hit,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name="hit",
attributes={"http.route": "/a", "http.retry.count": 5, "http.cache.hit": True},
attribute_write_mode="dual_write",
),
Traces(
timestamp=miss,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name="miss",
attributes={"http.route": "/b", "http.retry.count": 1, "http.cache.hit": False},
attribute_write_mode="dual_write",
),
]
)
start_ms = int((hit - timedelta(minutes=1)).timestamp() * 1000)
end_ms = int((miss + timedelta(minutes=1)).timestamp() * 1000)
for label, filter_expression, expected in [
("string_eq", "http.route = '/a'", {"/a"}),
("int_gt", "http.retry.count > 1", {"/a"}),
("bool_eq", "http.cache.hit = true", {"/a"}),
("exists", "http.cache.hit EXISTS", {"/a", "/b"}),
]:
response = make_query_request(
signoz,
token,
start_ms=start_ms,
end_ms=end_ms,
request_type=RequestType.TIME_SERIES,
queries=[
build_traces_scalar_query(
aggregations=[build_aggregation("count()")],
group_by=[build_group_by_field("http.route", field_context="attribute")],
filter_expression=filter_expression,
)
],
)
assert response.status_code == HTTPStatus.OK, label
series = index_series_by_label(
response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"], "http.route"
)
assert set(series.keys()) == expected, label