mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-31 16:40:44 +01:00
Compare commits
3 Commits
qf-values-
...
worktree-t
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
088d742bf3 | ||
|
|
814a65c1d6 | ||
|
|
04adfc7f30 |
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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,34 @@ 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 evolution entries are the rollout control. The JSON `attributes` column
|
||||
// is added once its column-wide entry is registered; `attributes_promoted` is
|
||||
// added per path once that path's promotion entry is registered. Promotion is
|
||||
// just a third evolution column: SelectEvolutionsForColumns picks a single home
|
||||
// per time window (promoted after its release, attributes before it, Map before
|
||||
// the JSON rollout), reading more than one only across an evolution boundary.
|
||||
cols := make([]*schema.Column, 0, 3)
|
||||
if attributeColumnEvolutionRegistered(key, SpanAttributesPromotedColumn) {
|
||||
cols = append(cols, indexV3Columns["attributes_promoted"])
|
||||
}
|
||||
if attributeColumnEvolutionRegistered(key, SpanAttributesColumn) {
|
||||
cols = append(cols, indexV3Columns["attributes"])
|
||||
}
|
||||
cols = append(cols, mapCol)
|
||||
return cols, nil
|
||||
case telemetrytypes.FieldContextSpan:
|
||||
// Check if this is a span scope field
|
||||
if strings.ToLower(key.Name) == SpanSearchScopeRoot || strings.ToLower(key.Name) == SpanSearchScopeEntryPoint {
|
||||
@@ -279,6 +299,7 @@ func (m *fieldMapper) resolveColumnExprs(
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
|
||||
key = narrowEvolutionsToColumns(key, columns)
|
||||
newColumns, evolutionsEntries, err := qbtypes.SelectEvolutionsForColumns(columns, key.Evolutions, startNs, endNs)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
@@ -309,6 +330,17 @@ 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:
|
||||
// Span attributes are flat shared-data paths keyed by the attribute name
|
||||
// verbatim (no nested object), so the name addresses the path directly.
|
||||
// String casts to ::String so an absent path folds to '' — matching the
|
||||
// Map column's default and preserving negative-operator parity; typed
|
||||
// numeric/bool cast to Nullable so a missing or wrong-typed path reads
|
||||
// NULL rather than 0/false. Existence tests the raw path (the cast folds
|
||||
// NULL) and is index-eligible via attributes_paths_tokenbf.
|
||||
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 +385,65 @@ func (m *fieldMapper) resolveColumnExprs(
|
||||
return exprs, existExprs, columns, nil
|
||||
}
|
||||
|
||||
// attributeColumnEvolutionRegistered reports whether key carries an evolution entry for the
|
||||
// given column, i.e. that column is a rollout-registered home for this attribute key.
|
||||
func attributeColumnEvolutionRegistered(key *telemetrytypes.TelemetryFieldKey, columnName string) bool {
|
||||
for _, e := range key.Evolutions {
|
||||
if e != nil && e.ColumnName == columnName {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// narrowEvolutionsToColumns returns key.Evolutions filtered to entries whose column is in cols.
|
||||
// A metadata attribute key carries the `__all__` evolutions for every attribute-context column
|
||||
// (all three legacy maps plus the JSON column), but getColumn resolves a typed key to only its
|
||||
// own map + the JSON column; without this filter SelectEvolutionsForColumns would reject the
|
||||
// in-range sibling-map entries as columns not present in the slice. A no-op for every other
|
||||
// context, where getColumn already returns exactly the columns the evolutions name.
|
||||
func narrowEvolutionsToColumns(key *telemetrytypes.TelemetryFieldKey, cols []*schema.Column) *telemetrytypes.TelemetryFieldKey {
|
||||
if len(key.Evolutions) == 0 {
|
||||
return key
|
||||
}
|
||||
allowed := make(map[string]struct{}, len(cols))
|
||||
for _, c := range cols {
|
||||
allowed[c.Name] = struct{}{}
|
||||
}
|
||||
filtered := make([]*telemetrytypes.EvolutionEntry, 0, len(key.Evolutions))
|
||||
for _, e := range key.Evolutions {
|
||||
if e == nil {
|
||||
continue
|
||||
}
|
||||
if _, ok := allowed[e.ColumnName]; ok {
|
||||
filtered = append(filtered, e)
|
||||
}
|
||||
}
|
||||
if len(filtered) == len(key.Evolutions) {
|
||||
return key
|
||||
}
|
||||
narrowed := *key
|
||||
narrowed.Evolutions = filtered
|
||||
return &narrowed
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -516,6 +607,7 @@ func (m *fieldMapper) columnIsTemporal(ctx context.Context, startNs, endNs uint6
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
key = narrowEvolutionsToColumns(key, columns)
|
||||
newColumns, _, err := qbtypes.SelectEvolutionsForColumns(columns, key.Evolutions, startNs, endNs)
|
||||
if err != nil {
|
||||
return false, err
|
||||
@@ -638,6 +730,7 @@ func (m *fieldMapper) ExistsFor(
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
key = narrowEvolutionsToColumns(key, columns)
|
||||
fieldExpression, err := m.FieldFor(ctx, orgID, tsStart, tsEnd, key)
|
||||
if err != nil {
|
||||
return "", err
|
||||
|
||||
@@ -0,0 +1,254 @@
|
||||
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")
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
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(
|
||||
"attributes_string", "Map(LowCardinality(String), String)", "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(
|
||||
"attributes_string", "Map(LowCardinality(String), String)", "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")
|
||||
})
|
||||
}
|
||||
@@ -154,6 +154,42 @@ func BuildCompleteFieldKeyMap(releaseTime time.Time) map[string][]*telemetrytype
|
||||
return keysMap
|
||||
}
|
||||
|
||||
// MockAttributeEvolutionData returns the attribute-context evolution timeline: the three legacy
|
||||
// map columns at epoch 0 and the JSON `attributes` column released at releaseTime. Every entry
|
||||
// is field_name "__all__", so a typed attribute key carries all four; getColumn keeps only its
|
||||
// own map plus the JSON column and narrowEvolutionsToColumns drops the rest before selection.
|
||||
func MockAttributeEvolutionData(releaseTime time.Time) []*telemetrytypes.EvolutionEntry {
|
||||
entry := func(col, typ string, rt time.Time) *telemetrytypes.EvolutionEntry {
|
||||
return &telemetrytypes.EvolutionEntry{
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
ColumnName: col,
|
||||
ColumnType: typ,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldName: "__all__",
|
||||
ReleaseTime: rt,
|
||||
}
|
||||
}
|
||||
return []*telemetrytypes.EvolutionEntry{
|
||||
entry("attributes_string", "Map(LowCardinality(String), String)", time.Unix(0, 0)),
|
||||
entry("attributes_number", "Map(LowCardinality(String), Float64)", time.Unix(0, 0)),
|
||||
entry("attributes_bool", "Map(LowCardinality(String), Bool)", time.Unix(0, 0)),
|
||||
entry("attributes", "JSON()", releaseTime),
|
||||
}
|
||||
}
|
||||
|
||||
// MockPromotedAttributeEvolutionData returns an attribute timeline where a single path is
|
||||
// promoted: the datatype's legacy map at epoch 0, the JSON `attributes` column at jsonRelease
|
||||
// (field_name "__all__"), and the per-path `attributes_promoted` column at promoteRelease
|
||||
// (field_name = path). This is what a promoted key's Evolutions look like once the metadata
|
||||
// layer composes the column-wide and per-path entries.
|
||||
func MockPromotedAttributeEvolutionData(mapColumn, mapType, path string, jsonRelease, promoteRelease time.Time) []*telemetrytypes.EvolutionEntry {
|
||||
return []*telemetrytypes.EvolutionEntry{
|
||||
{Signal: telemetrytypes.SignalTraces, ColumnName: mapColumn, ColumnType: mapType, FieldContext: telemetrytypes.FieldContextAttribute, FieldName: "__all__", ReleaseTime: time.Unix(0, 0)},
|
||||
{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 {
|
||||
|
||||
Reference in New Issue
Block a user