mirror of
https://github.com/SigNoz/signoz.git
synced 2026-09-04 18:40:40 +01:00
Compare commits
18 Commits
nv/heatmap
...
worktree-t
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c94fab8f3f | ||
|
|
4ac1251b90 | ||
|
|
736f3a7eab | ||
|
|
143b880204 | ||
|
|
8b42912b90 | ||
|
|
1fca671118 | ||
|
|
b67d5894ea | ||
|
|
97b8ef2e3c | ||
|
|
de9054ef7f | ||
|
|
bad7906dbb | ||
|
|
e8b74856ac | ||
|
|
9c6d3e8436 | ||
|
|
0f10fbb632 | ||
|
|
6a93afa794 | ||
|
|
95d8ea7602 | ||
|
|
47c3f3d96a | ||
|
|
0503672992 | ||
|
|
fae0f88450 |
@@ -256,7 +256,7 @@ Tests can be configured using pytest options:
|
||||
- `--sqlite-mode` — SQLite journal mode: `delete` or `wal` (default: `delete`). Only relevant when `--sqlstore-provider=sqlite`.
|
||||
- `--postgres-version` — PostgreSQL version (default: `15`)
|
||||
- `--clickhouse-version` — ClickHouse version, also used for ClickHouse Keeper (default: `25.12.5`)
|
||||
- `--schema-migrator-version` — SigNoz schema migrator version (default: `v0.144.6`)
|
||||
- `--schema-migrator-version` — SigNoz schema migrator version (default: `v0.144.9`)
|
||||
- `--with-web` — Build the frontend into the SigNoz image (required for e2e)
|
||||
|
||||
Example:
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -2500,13 +2500,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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -73,6 +73,8 @@ func (c *conditionBuilder) conditionFor(
|
||||
// the first member stands in for the field.
|
||||
fieldExpression, value = querybuilder.DataTypeCollisionHandledFieldName(logical.Single(), value, fieldExpression, operator)
|
||||
|
||||
fieldExpression = foldAbsentJSONReadToTypeDefault(logical.Single(), operator, fieldExpression)
|
||||
|
||||
// regular operators
|
||||
switch operator {
|
||||
// regular operators
|
||||
@@ -177,6 +179,31 @@ func (c *conditionBuilder) conditionFor(
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// foldAbsentJSONReadToTypeDefault gives negative operators on a numeric/bool JSON attribute the
|
||||
// legacy Map's absent-key semantics. Negative operators carry no guard, so NULL <> x would drop rows
|
||||
// lacking the key, whereas the Map defaulted them to the type zero and kept them (0 <> x).
|
||||
// String needs no fold — its ::String value already reads absent as ”.
|
||||
func foldAbsentJSONReadToTypeDefault(key *telemetrytypes.TelemetryFieldKey, operator qbtypes.FilterOperator, expr string) string {
|
||||
if !operator.IsNegativeOperator() || operator == qbtypes.FilterOperatorNotExists {
|
||||
return expr
|
||||
}
|
||||
if key.FieldContext != telemetrytypes.FieldContextAttribute {
|
||||
return expr
|
||||
}
|
||||
if !attributeColumnEvolutionRegistered(key, SpanAttributesColumn) {
|
||||
return expr
|
||||
}
|
||||
switch key.FieldDataType {
|
||||
case telemetrytypes.FieldDataTypeInt64,
|
||||
telemetrytypes.FieldDataTypeFloat64,
|
||||
telemetrytypes.FieldDataTypeNumber:
|
||||
return fmt.Sprintf("ifNull(%s, 0)", expr)
|
||||
case telemetrytypes.FieldDataTypeBool:
|
||||
return fmt.Sprintf("ifNull(%s, false)", expr)
|
||||
}
|
||||
return expr
|
||||
}
|
||||
|
||||
// isFoldContext reports whether the context is one CandidateKeys would fold the prefix into
|
||||
// the key name for (span/trace). These behave like a default context that also addresses
|
||||
// columns and attributes, unlike strict resource/attribute/scope contexts.
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -3,6 +3,7 @@ package tracestelemetryschema
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
|
||||
@@ -52,8 +53,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 +187,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 {
|
||||
@@ -260,7 +275,7 @@ func (m *fieldMapper) FieldFor(
|
||||
for i, expr := range exprs {
|
||||
finalExprs = append(finalExprs, fmt.Sprintf("%s, %s", existExpr[i], expr))
|
||||
}
|
||||
return "multiIf(" + strings.Join(finalExprs, ", ") + ", NULL)", nil
|
||||
return fmt.Sprintf("multiIf(%s, NULL)", strings.Join(finalExprs, ", ")), nil
|
||||
}
|
||||
|
||||
// should not reach here
|
||||
@@ -309,8 +324,12 @@ 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, attributeJSONValueExpr(path, 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)
|
||||
return nil, nil, nil, errors.NewInternalf(errors.CodeInternal, "only resource, scope and attribute context fields are supported for json columns, got %s", key.FieldContext.String)
|
||||
}
|
||||
case schema.ColumnTypeEnumString,
|
||||
schema.ColumnTypeEnumUInt64,
|
||||
@@ -353,6 +372,33 @@ func (m *fieldMapper) resolveColumnExprs(
|
||||
return exprs, existExprs, columns, 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
|
||||
}
|
||||
|
||||
// attributeJSONValueExpr renders the value expression for a span attribute read from the JSON
|
||||
// column. Absent path reads ” the way the Map default does.
|
||||
// Numeric and bool use accurateCastOrNull, which reads an absent path or a same-named key
|
||||
// stored as another type as NULL rather than erroring — unlike a bare ::Int64/::Bool cast, and
|
||||
// unlike ::Nullable(Bool), which throws on a non-bool string.
|
||||
func attributeJSONValueExpr(path string, dataType telemetrytypes.FieldDataType) string {
|
||||
switch dataType {
|
||||
case telemetrytypes.FieldDataTypeInt64,
|
||||
telemetrytypes.FieldDataTypeFloat64,
|
||||
telemetrytypes.FieldDataTypeNumber,
|
||||
telemetrytypes.FieldDataTypeBool:
|
||||
return fmt.Sprintf("accurateCastOrNull(%s, '%s')", path, telemetrytypes.MappingFieldDataTypeToJSONDataType[dataType].StringValue())
|
||||
default:
|
||||
return path + "::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
|
||||
@@ -439,6 +485,16 @@ func (m *fieldMapper) ColumnExpressionFor(
|
||||
// Group-by/order (String) and aggregation (String/Float64): every candidate is
|
||||
// exists-guarded and coerced to requiredDataType, in a single multiIf. Raw select
|
||||
// (Unspecified) keeps the lighter native shape below.
|
||||
// A JSON type-collision shows up as several candidates sharing one physical path (and so one
|
||||
// raw-path guard). Guarding each branch by that shared path can't tell the types apart, so
|
||||
// discriminate by castability instead. Map candidates keep distinct per-column guards and are
|
||||
// left to the normal folds below.
|
||||
if fold, ok, err := m.foldCastDiscriminated(ctx, startNs, endNs, candidates, requiredDataType); err != nil {
|
||||
return "", err
|
||||
} else if ok {
|
||||
return fold, nil
|
||||
}
|
||||
|
||||
if requiredDataType != telemetrytypes.FieldDataTypeUnspecified {
|
||||
var dummyValue any = ""
|
||||
if requiredDataType == telemetrytypes.FieldDataTypeFloat64 {
|
||||
@@ -500,6 +556,99 @@ func (m *fieldMapper) ColumnExpressionFor(
|
||||
return fmt.Sprintf("multiIf(%s, NULL)", strings.Join(args, ", ")), nil
|
||||
}
|
||||
|
||||
// foldCastDiscriminated renders a JSON type-collision: candidates that each resolve to a single
|
||||
// column in the window and share an identical raw-path existence guard (every type of one
|
||||
// attribute lives at one JSON path). It guards each numeric/bool branch by whether the path casts
|
||||
// to that type (`<cast> IS NOT NULL`) and keeps the ::String branch as the last-resort fallback,
|
||||
// so each row is read as its actual stored type instead of the first branch always winning.
|
||||
//
|
||||
// It returns ok=false — leaving the caller's normal fold in place — unless the candidates are
|
||||
// exactly such a collision: fewer than two candidates, a family, a straddle/multi-column
|
||||
// candidate, or candidates with distinct guards (map columns) all decline.
|
||||
func (m *fieldMapper) foldCastDiscriminated(
|
||||
ctx context.Context,
|
||||
startNs, endNs uint64,
|
||||
candidates []*telemetrytypes.LogicalField,
|
||||
requiredDataType telemetrytypes.FieldDataType,
|
||||
) (string, bool, error) {
|
||||
if len(candidates) < 2 {
|
||||
return "", false, nil
|
||||
}
|
||||
|
||||
type branch struct {
|
||||
guard string
|
||||
value string
|
||||
member *telemetrytypes.TelemetryFieldKey
|
||||
catchAll bool
|
||||
}
|
||||
branches := make([]branch, 0, len(candidates))
|
||||
rawGuardCount := make(map[string]int, len(candidates))
|
||||
for _, logical := range candidates {
|
||||
if logical.IsFamily() {
|
||||
return "", false, nil
|
||||
}
|
||||
member := logical.Single()
|
||||
exprs, existExprs, _, err := m.resolveColumnExprs(ctx, startNs, endNs, member)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
if len(exprs) != 1 || len(existExprs) != 1 {
|
||||
return "", false, nil
|
||||
}
|
||||
rawGuardCount[existExprs[0]]++
|
||||
catchAll := member.FieldDataType == telemetrytypes.FieldDataTypeString ||
|
||||
member.FieldDataType == telemetrytypes.FieldDataTypeUnspecified
|
||||
guard := existExprs[0]
|
||||
if !catchAll {
|
||||
guard = exprs[0] + " IS NOT NULL"
|
||||
}
|
||||
branches = append(branches, branch{guard: guard, value: exprs[0], member: member, catchAll: catchAll})
|
||||
}
|
||||
|
||||
collision := false
|
||||
for _, n := range rawGuardCount {
|
||||
if n > 1 {
|
||||
collision = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !collision {
|
||||
return "", false, nil
|
||||
}
|
||||
|
||||
slices.SortStableFunc(branches, func(a, b branch) int {
|
||||
switch {
|
||||
case a.catchAll == b.catchAll:
|
||||
return 0
|
||||
case a.catchAll:
|
||||
return 1
|
||||
default:
|
||||
return -1
|
||||
}
|
||||
})
|
||||
|
||||
var dummyValue any = ""
|
||||
if requiredDataType == telemetrytypes.FieldDataTypeFloat64 {
|
||||
dummyValue = 0.0
|
||||
}
|
||||
stmts := make([]string, 0, len(branches)*2)
|
||||
seen := make(map[string]struct{}, len(branches))
|
||||
for _, br := range branches {
|
||||
if _, dup := seen[br.guard]; dup {
|
||||
continue
|
||||
}
|
||||
seen[br.guard] = struct{}{}
|
||||
value := br.value
|
||||
if requiredDataType == telemetrytypes.FieldDataTypeUnspecified {
|
||||
value = fmt.Sprintf("toString(%s)", value)
|
||||
} else {
|
||||
value, _ = querybuilder.DataTypeCollisionHandledFieldName(br.member, dummyValue, value, qbtypes.FilterOperatorUnknown)
|
||||
}
|
||||
stmts = append(stmts, br.guard, value)
|
||||
}
|
||||
return fmt.Sprintf("multiIf(%s, NULL)", strings.Join(stmts, ", ")), true, nil
|
||||
}
|
||||
|
||||
// logicalIsTemporal reports whether the logical field resolves to a single time
|
||||
// column. A family is attribute-backed and never temporal.
|
||||
func (m *fieldMapper) logicalIsTemporal(ctx context.Context, startNs, endNs uint64, logical *telemetrytypes.LogicalField) (bool, error) {
|
||||
|
||||
@@ -0,0 +1,474 @@
|
||||
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, "accurateCastOrNull(attributes.`user.id`, 'Float64')"},
|
||||
{"number straddle -> dual", telemetrytypes.FieldDataTypeNumber, attrWindowStraddle, "multiIf(attributes.`user.id` IS NOT NULL, accurateCastOrNull(attributes.`user.id`, 'Float64'), mapContains(attributes_number, 'user.id'), attributes_number['user.id'], NULL)"},
|
||||
{"int64 after -> json", telemetrytypes.FieldDataTypeInt64, attrWindowAfter, "accurateCastOrNull(attributes.`user.id`, 'Int64')"},
|
||||
{"bool before -> map", telemetrytypes.FieldDataTypeBool, attrWindowBefore, "attributes_bool['user.id']"},
|
||||
{"bool after -> json", telemetrytypes.FieldDataTypeBool, attrWindowAfter, "accurateCastOrNull(attributes.`user.id`, 'Bool')"},
|
||||
{"bool straddle -> dual", telemetrytypes.FieldDataTypeBool, attrWindowStraddle, "multiIf(attributes.`user.id` IS NOT NULL, accurateCastOrNull(attributes.`user.id`, '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(accurateCastOrNull(attributes.`http.status_code`, '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(accurateCastOrNull(attributes.`latency`, '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(accurateCastOrNull(attributes.`latency`, '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(accurateCastOrNull(attributes.`http.status_code`, '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. On the JSON column both interpretations read the same path, so the raw-path guard
|
||||
// can't tell them apart; each branch is instead guarded by whether the path casts to its type,
|
||||
// with the ::String branch as the last-resort fallback. A row is read as its actual stored type
|
||||
// (int via accurateCastOrNull to Int64, everything else via ::String) rather than the first branch winning.
|
||||
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(accurateCastOrNull(attributes.`http.status_code`, 'Int64') IS NOT NULL, toString(accurateCastOrNull(attributes.`http.status_code`, 'Int64')), attributes.`http.status_code` IS NOT NULL, attributes.`http.status_code`::String, NULL)",
|
||||
got)
|
||||
}
|
||||
|
||||
// TestColumnExpressionForAttributeJSONTypeCollisionNumericAgg covers a numeric aggregation over a
|
||||
// name colliding as Number and String: the numeric branch is read natively when the path casts to
|
||||
// a number, and only rows that are not numeric fall through to the string parse — so a genuinely
|
||||
// string-stored value is never silently nulled by a numeric-first cast.
|
||||
func TestColumnExpressionForAttributeJSONTypeCollisionNumericAgg(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fm := NewFieldMapper(flaggertest.New(t))
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
numKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeNumber, evo)
|
||||
strKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeString, evo)
|
||||
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
"http.status_code": {&numKey, &strKey},
|
||||
}
|
||||
|
||||
ref := attrKey("http.status_code", telemetrytypes.FieldDataTypeUnspecified, nil)
|
||||
got, err := fm.ColumnExpressionFor(ctx, valuer.UUID{}, attrWindowAfter[0], attrWindowAfter[1], &ref, telemetrytypes.FieldDataTypeFloat64, fieldKeys)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t,
|
||||
"multiIf(accurateCastOrNull(attributes.`http.status_code`, 'Float64') IS NOT NULL, toFloat64(accurateCastOrNull(attributes.`http.status_code`, 'Float64')), attributes.`http.status_code` IS NOT NULL, toFloat64OrNull(attributes.`http.status_code`::String), 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)
|
||||
}
|
||||
|
||||
// TestConditionForAttributeJSONNegativeOperatorParity pins Map parity for numeric/bool attributes.
|
||||
// The value reads an absent key as NULL (accurateCastOrNull, or the straddle multiIf else); a
|
||||
// positive operator excludes such a row via the exists guard, but a negative operator has no guard,
|
||||
// so the condition builder folds the NULL to the Map's type zero (ifNull) for negatives only.
|
||||
// String needs no fold — ::String already reads absent as ”. The fold rides the attributes
|
||||
// evolution: a key without it (the pre-rollout system) is byte-identical to today.
|
||||
func TestConditionForAttributeJSONNegativeOperatorParity(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fm := NewFieldMapper(flaggertest.New(t))
|
||||
cb := NewConditionBuilder(fm, flaggertest.New(t))
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
build := func(t *testing.T, key telemetrytypes.TelemetryFieldKey, window [2]uint64, op qbtypes.FilterOperator, value any) string {
|
||||
t.Helper()
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, window[0], window[1], &key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, qbtypes.ConditionBuilderOptions{}, op, value, sb)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conds...)
|
||||
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
return sql
|
||||
}
|
||||
|
||||
t.Run("not equal number after -> NULL folded to 0", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
|
||||
sql := build(t, key, attrWindowAfter, qbtypes.FilterOperatorNotEqual, float64(200))
|
||||
assert.Contains(t, sql, "ifNull(toFloat64(accurateCastOrNull(attributes.`http.status_code`, 'Int64')), 0) <> ?")
|
||||
})
|
||||
|
||||
t.Run("not equal bool after -> NULL folded to false", func(t *testing.T) {
|
||||
key := attrKey("http.cache.hit", telemetrytypes.FieldDataTypeBool, evo)
|
||||
sql := build(t, key, attrWindowAfter, qbtypes.FilterOperatorNotEqual, true)
|
||||
assert.Contains(t, sql, "ifNull(accurateCastOrNull(attributes.`http.cache.hit`, 'Bool'), false) <> ?")
|
||||
})
|
||||
|
||||
t.Run("equal number after -> not folded, exists guard excludes absent", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
|
||||
sql := build(t, key, attrWindowAfter, qbtypes.FilterOperatorEqual, float64(0))
|
||||
assert.Contains(t, sql, "(toFloat64(accurateCastOrNull(attributes.`http.status_code`, 'Int64')) = ? AND attributes.`http.status_code` IS NOT NULL)")
|
||||
assert.NotContains(t, sql, "ifNull")
|
||||
})
|
||||
|
||||
t.Run("not in number after -> each operand folded to 0", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
|
||||
sql := build(t, key, attrWindowAfter, qbtypes.FilterOperatorNotIn, []any{float64(200), float64(404)})
|
||||
assert.Contains(t, sql, "(ifNull(toFloat64(accurateCastOrNull(attributes.`http.status_code`, 'Int64')), 0) <> ? AND ifNull(toFloat64(accurateCastOrNull(attributes.`http.status_code`, 'Int64')), 0) <> ?)")
|
||||
})
|
||||
|
||||
t.Run("not equal number straddle -> whole multiIf folded", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
|
||||
sql := build(t, key, attrWindowStraddle, qbtypes.FilterOperatorNotEqual, float64(200))
|
||||
assert.Contains(t, sql, "ifNull(toFloat64(multiIf(attributes.`http.status_code` IS NOT NULL, accurateCastOrNull(attributes.`http.status_code`, 'Int64'), mapContains(attributes_number, 'http.status_code'), attributes_number['http.status_code'], NULL)), 0) <> ?")
|
||||
})
|
||||
|
||||
t.Run("not equal number before -> harmless fold over the map read", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
|
||||
sql := build(t, key, attrWindowBefore, qbtypes.FilterOperatorNotEqual, float64(200))
|
||||
assert.Contains(t, sql, "ifNull(toFloat64(attributes_number['http.status_code']), 0) <> ?")
|
||||
})
|
||||
|
||||
t.Run("not equal number without rollout -> byte-identical to today", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, nil)
|
||||
sql := build(t, key, attrWindowBefore, qbtypes.FilterOperatorNotEqual, float64(200))
|
||||
assert.Contains(t, sql, "toFloat64(attributes_number['http.status_code']) <> ?")
|
||||
assert.NotContains(t, sql, "ifNull")
|
||||
})
|
||||
|
||||
t.Run("not equal string after -> '' default, never folded", func(t *testing.T) {
|
||||
key := attrKey("user.id", telemetrytypes.FieldDataTypeString, evo)
|
||||
sql := build(t, key, attrWindowAfter, qbtypes.FilterOperatorNotEqual, "admin")
|
||||
assert.Contains(t, sql, "attributes.`user.id`::String <> ?")
|
||||
assert.NotContains(t, sql, "ifNull")
|
||||
})
|
||||
}
|
||||
|
||||
// TestConditionForAttributeJSONStraddleAbsentKeyExclusion guards the straddle exists path: because
|
||||
// the value reads absent-in-both-homes as NULL (multiIf else), a positive zero-value comparison and
|
||||
// EXISTS/NOT EXISTS must still exclude a key absent from every home, rather than matching it.
|
||||
func TestConditionForAttributeJSONStraddleAbsentKeyExclusion(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fm := NewFieldMapper(flaggertest.New(t))
|
||||
cb := NewConditionBuilder(fm, flaggertest.New(t))
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
build := func(t *testing.T, key telemetrytypes.TelemetryFieldKey, op qbtypes.FilterOperator, value any) string {
|
||||
t.Helper()
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, attrWindowStraddle[0], attrWindowStraddle[1], &key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, qbtypes.ConditionBuilderOptions{}, op, value, sb)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conds...)
|
||||
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
return sql
|
||||
}
|
||||
|
||||
guard := "multiIf(attributes.`http.status_code` IS NOT NULL, accurateCastOrNull(attributes.`http.status_code`, 'Int64'), mapContains(attributes_number, 'http.status_code'), attributes_number['http.status_code'], NULL) IS NOT NULL"
|
||||
|
||||
t.Run("equal zero keeps the exists guard", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
|
||||
assert.Contains(t, build(t, key, qbtypes.FilterOperatorEqual, float64(0)), guard)
|
||||
})
|
||||
t.Run("exists is the raw multiIf, not always-true", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
|
||||
assert.Contains(t, build(t, key, qbtypes.FilterOperatorExists, nil), guard)
|
||||
})
|
||||
t.Run("not exists negates the raw multiIf", func(t *testing.T) {
|
||||
key := attrKey("http.status_code", telemetrytypes.FieldDataTypeInt64, evo)
|
||||
sql := build(t, key, qbtypes.FilterOperatorNotExists, nil)
|
||||
assert.Contains(t, sql, ", NULL) IS NULL")
|
||||
})
|
||||
}
|
||||
@@ -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")
|
||||
})
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -23,8 +23,21 @@ func SelectEvolutionsForColumns(columns []*schema.Column, evolutions []*telemetr
|
||||
return columns, nil, nil
|
||||
}
|
||||
|
||||
sortedEvolutions := make([]*telemetrytypes.EvolutionEntry, len(evolutions))
|
||||
copy(sortedEvolutions, evolutions)
|
||||
// Derive the base column from the candidate columns.
|
||||
seen := make(map[string]struct{}, len(evolutions))
|
||||
for _, e := range evolutions {
|
||||
seen[e.ColumnName] = struct{}{}
|
||||
}
|
||||
|
||||
// never modify evolutions in place, it may be cached and shared across queries.
|
||||
sortedEvolutions := make([]*telemetrytypes.EvolutionEntry, 0, len(evolutions)+len(columns))
|
||||
sortedEvolutions = append(sortedEvolutions, evolutions...)
|
||||
for _, c := range columns {
|
||||
if _, ok := seen[c.Name]; ok {
|
||||
continue
|
||||
}
|
||||
sortedEvolutions = append(sortedEvolutions, &telemetrytypes.EvolutionEntry{ColumnName: c.Name, ReleaseTime: time.Unix(0, 0)})
|
||||
}
|
||||
|
||||
// sort the evolutions by ReleaseTime ascending
|
||||
sort.Slice(sortedEvolutions, func(i, j int) bool {
|
||||
|
||||
@@ -396,14 +396,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
|
||||
|
||||
@@ -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))
|
||||
})
|
||||
}
|
||||
@@ -114,6 +114,6 @@ def pytest_addoption(parser: pytest.Parser):
|
||||
parser.addoption(
|
||||
"--schema-migrator-version",
|
||||
action="store",
|
||||
default="v0.144.6",
|
||||
default="v0.144.9", # todo(nikhil): change to 0.144.10
|
||||
help="schema migrator version",
|
||||
)
|
||||
|
||||
40
tests/fixtures/traces.py
vendored
40
tests/fixtures/traces.py
vendored
@@ -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,
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user