test: pin the semconv family matrix on a live stack

A dedicated package runs SigNoz with resolve_semconv_families on and a
second instance with the flag at its default. The fleet has one
identity per state: OLD (old spelling only), NEW (current only), BOTH
(current staging and old production), NEITHER (keyless).

- 36 filter cells: nine operators, both spellings, both contexts. The
  result sets do not depend on the requested spelling; the current
  spelling wins on the conflict row; negative operators keep keyless
  rows, exactly like a single key.
- Group-by merges the fleet into one production group, a staging group,
  and a NULL group, and the group column carries the requested spelling.
- A bare name with the family under two contexts warns and keeps the
  resource side; the family survives the collision as one unit.
- Logs stay literal with the flag on, and everything stays literal with
  the flag off.

Verified: 42 passed against the live stack.

Assisted-by: Claude Fable 5

pkg/modules/rulestatehistory/implrulestatehistory/field_mapper.go

@@ -88,5 +88,5 @@ func (m *fieldMapper) ColumnExpressionFor(ctx context.Context, orgID valuer.UUID
 	if err != nil {
 		return "", err
 	}
-	return fmt.Sprintf("%s AS %s", sqlbuilder.Escape(colName), querybuilder.ClickHouseIdentifier(field.Name)), nil
+	return fmt.Sprintf("%s AS `%s`", sqlbuilder.Escape(colName), field.Name), nil
 }

pkg/querybuilder/exists_expr.go

@@ -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, ClickHouseIdentifier(key.Name))
+		rawPath := fmt.Sprintf("%s.`%s`", columnName, key.Name)
 		if exists {
 			return rawPath + " IS NOT NULL", nil
 		}
@@ -88,7 +88,7 @@ func ExistsExpression(columns []*schema.Column, key *telemetrytypes.TelemetryFie
 
 		switch valueType := column.Type.(schema.MapColumnType).ValueType; valueType.GetType() {
 		case schema.ColumnTypeEnumString, schema.ColumnTypeEnumBool, schema.ColumnTypeEnumFloat64:
-			leftOperand := fmt.Sprintf("mapContains(%s, %s)", column.Name, ClickHouseStringLiteral(key.Name))
+			leftOperand := fmt.Sprintf("mapContains(%s, '%s')", column.Name, key.Name)
 			if key.Materialized {
 				leftOperand = telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key)
 			}

pkg/statementbuilder/resourcefilter/condition_builder.go

@@ -96,7 +96,7 @@ func valueIndexCondition(
 func memberPresenceCondition(sb *sqlbuilder.SelectBuilder, column string, members []*telemetrytypes.TelemetryFieldKey, exists bool) string {
 	conditions := make([]string, 0, len(members))
 	for _, member := range members {
-		field := fmt.Sprintf("simpleJSONHas(%s, %s)", column, querybuilder.ClickHouseStringLiteral(member.Name))
+		field := fmt.Sprintf("simpleJSONHas(%s, '%s')", column, member.Name)
 		if exists {
 			conditions = append(conditions, sb.E(field, true))
 		} else {

pkg/statementbuilder/resourcefilter/field_mapper.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 
 	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
-	"github.com/SigNoz/signoz/pkg/querybuilder"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -67,7 +66,7 @@ func (m *defaultFieldMapper) FieldFor(
 		return "", err
 	}
 	if key.FieldContext == telemetrytypes.FieldContextResource {
-		return fmt.Sprintf("simpleJSONExtractString(%s, %s)", columns[0].Name, querybuilder.ClickHouseStringLiteral(key.Name)), nil
+		return fmt.Sprintf("simpleJSONExtractString(%s, '%s')", columns[0].Name, key.Name), nil
 	}
 	return columns[0].Name, nil
 }
@@ -92,7 +91,7 @@ func (m *defaultFieldMapper) ExistsFor(
 		}
 		return "false", nil
 	}
-	pred := fmt.Sprintf("simpleJSONHas(%s, %s)", columns[0].Name, querybuilder.ClickHouseStringLiteral(key.Name))
+	pred := fmt.Sprintf("simpleJSONHas(%s, '%s')", columns[0].Name, key.Name)
 	if exists {
 		return pred, nil
 	}
@@ -111,5 +110,5 @@ func (m *defaultFieldMapper) ColumnExpressionFor(
 	if err != nil {
 		return "", err
 	}
-	return fmt.Sprintf("%s AS %s", fieldExpression, querybuilder.ClickHouseIdentifier(key.Name)), nil
+	return fmt.Sprintf("%s AS `%s`", fieldExpression, key.Name), nil
 }

pkg/telemetrymetadata/condition_builder.go

@@ -168,7 +168,7 @@ func (c *conditionBuilder) conditionForKey(
 			KeyType:   schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString},
 			ValueType: schema.ColumnTypeString,
 		}:
-			leftOperand := fmt.Sprintf("mapContains(%s, %s)", columns[0].Name, querybuilder.ClickHouseStringLiteral(key.Name))
+			leftOperand := fmt.Sprintf("mapContains(%s, '%s')", columns[0].Name, key.Name)
 			if operator == qbtypes.FilterOperatorExists {
 				cond = sb.E(leftOperand, true)
 			} else {

pkg/telemetrymetadata/field_mapper.go

@@ -7,7 +7,6 @@ import (
 
 	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/querybuilder"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -65,7 +64,7 @@ func (m *fieldMapper) ExistsFor(ctx context.Context, _ valuer.UUID, tsStart, tsE
 	if err != nil {
 		return "", err
 	}
-	pred := fmt.Sprintf("mapContains(%s, %s)", columns[0].Name, querybuilder.ClickHouseStringLiteral(key.Name))
+	pred := fmt.Sprintf("mapContains(%s, '%s')", columns[0].Name, key.Name)
 	if exists {
 		return pred, nil
 	}
@@ -83,7 +82,7 @@ func (m *fieldMapper) FieldFor(ctx context.Context, _ valuer.UUID, startNs, endN
 		KeyType:   schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString},
 		ValueType: schema.ColumnTypeString,
 	}:
-		return fmt.Sprintf("%s[%s]", columns[0].Name, querybuilder.ClickHouseStringLiteral(key.Name)), nil
+		return fmt.Sprintf("%s['%s']", columns[0].Name, key.Name), nil
 	}
 	return columns[0].Name, nil
 }
@@ -131,5 +130,5 @@ func (m *fieldMapper) ColumnExpressionFor(
 		}
 	}
 
-	return fmt.Sprintf("%s AS %s", sqlbuilder.Escape(fieldExpression), querybuilder.ClickHouseIdentifier(field.Name)), nil
+	return fmt.Sprintf("%s AS `%s`", sqlbuilder.Escape(fieldExpression), field.Name), nil
 }

pkg/telemetryschema/audittelemetryschema/field_mapper.go

@@ -68,7 +68,7 @@ func (m *fieldMapper) FieldFor(ctx context.Context, _ valuer.UUID, _, _ uint64,
 		if key.FieldContext != telemetrytypes.FieldContextResource {
 			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "only resource context fields are supported for json columns in audit, got %s", key.FieldContext.String)
 		}
-		return fmt.Sprintf("%s.%s::String", column.Name, querybuilder.ClickHouseIdentifier(key.Name)), nil
+		return fmt.Sprintf("%s.`%s`::String", column.Name, key.Name), nil
 	case schema.ColumnTypeEnumLowCardinality:
 		return column.Name, nil
 	case schema.ColumnTypeEnumString, schema.ColumnTypeEnumUInt64, schema.ColumnTypeEnumUInt32, schema.ColumnTypeEnumUInt8:
@@ -84,7 +84,7 @@ func (m *fieldMapper) FieldFor(ctx context.Context, _ valuer.UUID, _, _ uint64,
 			if key.Materialized {
 				return telemetrytypes.FieldKeyToMaterializedColumnName(key), nil
 			}
-			return fmt.Sprintf("%s[%s]", column.Name, querybuilder.ClickHouseStringLiteral(key.Name)), nil
+			return fmt.Sprintf("%s['%s']", column.Name, key.Name), nil
 		default:
 			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported map value type %s", valueType)
 		}
@@ -156,7 +156,7 @@ func (m *fieldMapper) ColumnExpressionFor(
 		return fmt.Sprintf("multiIf(%s, %s, NULL)", guard, coerced), nil
 	}
 
-	return fmt.Sprintf("%s AS %s", sqlbuilder.Escape(fieldExpression), querybuilder.ClickHouseIdentifier(field.Name)), nil
+	return fmt.Sprintf("%s AS `%s`", sqlbuilder.Escape(fieldExpression), field.Name), nil
 }
 
 // CandidateKeys returns nil: audit has no synthesize-on-unknown-key fallback, so an

pkg/telemetryschema/logstelemetryschema/field_mapper.go

@@ -141,8 +141,8 @@ func (m *fieldMapper) FieldFor(ctx context.Context, orgID valuer.UUID, tsStart,
 		case schema.ColumnTypeEnumJSON:
 			switch key.FieldContext {
 			case telemetrytypes.FieldContextResource:
-				exprs = append(exprs, fmt.Sprintf("%s.%s::String", columnName, querybuilder.ClickHouseIdentifier(key.Name)))
-				existExpr = append(existExpr, fmt.Sprintf("%s.%s IS NOT NULL", columnName, querybuilder.ClickHouseIdentifier(key.Name)))
+				exprs = append(exprs, fmt.Sprintf("%s.`%s`::String", columnName, key.Name))
+				existExpr = append(existExpr, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, key.Name))
 			case telemetrytypes.FieldContextBody:
 				if key.Name == messageSubField {
 					exprs = append(exprs, messageSubColumn)
@@ -186,8 +186,8 @@ func (m *fieldMapper) FieldFor(ctx context.Context, orgID valuer.UUID, tsStart,
 					exprs = append(exprs, telemetrytypes.FieldKeyToMaterializedColumnName(key))
 					existExpr = append(existExpr, telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key))
 				} else {
-					exprs = append(exprs, fmt.Sprintf("%s[%s]", columnName, querybuilder.ClickHouseStringLiteral(key.Name)))
-					existExpr = append(existExpr, fmt.Sprintf("mapContains(%s, %s)", columnName, querybuilder.ClickHouseStringLiteral(key.Name)))
+					exprs = append(exprs, fmt.Sprintf("%s['%s']", columnName, key.Name))
+					existExpr = append(existExpr, fmt.Sprintf("mapContains(%s, '%s')", columnName, key.Name))
 				}
 			default:
 				return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "exists operator is not supported for map column type %s", valueType)
@@ -415,7 +415,7 @@ func (m *fieldMapper) buildFieldForJSON(key *telemetrytypes.TelemetryFieldKey) (
 			elemType = telemetrytypes.String
 		}
 
-		fieldPath := fmt.Sprintf("%s.%s", LogsV2BodyV2Column, querybuilder.ClickHouseIdentifier(key.Name))
+		fieldPath := fmt.Sprintf("%s.`%s`", LogsV2BodyV2Column, key.Name)
 		return fmt.Sprintf("dynamicElement(%s, '%s')", fieldPath, elemType.StringValue()), nil
 	}
 

pkg/telemetryschema/logstelemetryschema/json_condition_builder.go

@@ -41,7 +41,7 @@ func (c *jsonConditionBuilder) buildJSONCondition(operator qbtypes.FilterOperato
 
 	// path index
 	if operator.AddDefaultExistsFilter() {
-		pathIndex := fmt.Sprintf(`has(%s, %s)`, schemamigrator.JSONPathsIndexExpr(LogsV2BodyV2Column), querybuilder.ClickHouseStringLiteral(c.key.ArrayParentPaths()[0]))
+		pathIndex := fmt.Sprintf(`has(%s, '%s')`, schemamigrator.JSONPathsIndexExpr(LogsV2BodyV2Column), c.key.ArrayParentPaths()[0])
 		return sb.And(baseCond, pathIndex), nil
 	}
 

pkg/telemetryschema/metricstelemetryschema/condition_builder.go

@@ -136,9 +136,9 @@ func (c *conditionBuilder) conditionFor(
 		}
 
 		if operator == qbtypes.FilterOperatorExists {
-			return fmt.Sprintf("has(JSONExtractKeys(labels), %s)", querybuilder.ClickHouseStringLiteral(key.Name)), nil
+			return fmt.Sprintf("has(JSONExtractKeys(labels), '%s')", key.Name), nil
 		}
-		return fmt.Sprintf("not has(JSONExtractKeys(labels), %s)", querybuilder.ClickHouseStringLiteral(key.Name)), nil
+		return fmt.Sprintf("not has(JSONExtractKeys(labels), '%s')", key.Name), nil
 	}
 	return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported operator: %v", operator)
 }

pkg/telemetryschema/metricstelemetryschema/field_mapper.go

@@ -6,7 +6,6 @@ import (
 	"slices"
 
 	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
-	"github.com/SigNoz/signoz/pkg/querybuilder"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -80,14 +79,14 @@ func (m *fieldMapper) FieldFor(ctx context.Context, _ valuer.UUID, startNs, endN
 
 	switch key.FieldContext {
 	case telemetrytypes.FieldContextResource, telemetrytypes.FieldContextScope, telemetrytypes.FieldContextAttribute:
-		return fmt.Sprintf("JSONExtractString(%s, %s)", columns[0].Name, querybuilder.ClickHouseStringLiteral(key.Name)), nil
+		return fmt.Sprintf("JSONExtractString(%s, '%s')", columns[0].Name, key.Name), nil
 	case telemetrytypes.FieldContextMetric:
 		return columns[0].Name, nil
 	case telemetrytypes.FieldContextUnspecified:
 		if slices.Contains(IntrinsicFields, key.Name) {
 			return columns[0].Name, nil
 		}
-		return fmt.Sprintf("JSONExtractString(%s, %s)", columns[0].Name, querybuilder.ClickHouseStringLiteral(key.Name)), nil
+		return fmt.Sprintf("JSONExtractString(%s, '%s')", columns[0].Name, key.Name), nil
 	}
 
 	return columns[0].Name, nil
@@ -104,9 +103,9 @@ func (m *fieldMapper) ExistsFor(_ context.Context, _ valuer.UUID, _, _ uint64, k
 		return "true", nil
 	}
 	if exists {
-		return fmt.Sprintf("has(JSONExtractKeys(labels), %s)", querybuilder.ClickHouseStringLiteral(key.Name)), nil
+		return fmt.Sprintf("has(JSONExtractKeys(labels), '%s')", key.Name), nil
 	}
-	return fmt.Sprintf("not has(JSONExtractKeys(labels), %s)", querybuilder.ClickHouseStringLiteral(key.Name)), nil
+	return fmt.Sprintf("not has(JSONExtractKeys(labels), '%s')", key.Name), nil
 }
 
 func (m *fieldMapper) ColumnExpressionFor(

pkg/telemetryschema/tracestelemetryschema/field_mapper.go

@@ -298,8 +298,8 @@ func (m *fieldMapper) resolveColumnExprs(
 			}
 			// have to add ::string as clickHouse throws an error :- data types Variant/Dynamic are not allowed in GROUP BY
 			// once clickHouse dependency is updated, we need to check if we can remove it.
-			exprs = append(exprs, fmt.Sprintf("%s.%s::String", columnName, querybuilder.ClickHouseIdentifier(key.Name)))
-			existExprs = append(existExprs, fmt.Sprintf("%s.%s IS NOT NULL", columnName, querybuilder.ClickHouseIdentifier(key.Name)))
+			exprs = append(exprs, fmt.Sprintf("%s.`%s`::String", columnName, key.Name))
+			existExprs = append(existExprs, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, key.Name))
 		case schema.ColumnTypeEnumString,
 			schema.ColumnTypeEnumUInt64,
 			schema.ColumnTypeEnumUInt32,
@@ -329,8 +329,8 @@ func (m *fieldMapper) resolveColumnExprs(
 					exprs = append(exprs, telemetrytypes.FieldKeyToMaterializedColumnName(key))
 					existExprs = append(existExprs, telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key))
 				} else {
-					exprs = append(exprs, fmt.Sprintf("%s[%s]", columnName, querybuilder.ClickHouseStringLiteral(key.Name)))
-					existExprs = append(existExprs, fmt.Sprintf("mapContains(%s, %s)", columnName, querybuilder.ClickHouseStringLiteral(key.Name)))
+					exprs = append(exprs, fmt.Sprintf("%s['%s']", columnName, key.Name))
+					existExprs = append(existExprs, fmt.Sprintf("mapContains(%s, '%s')", columnName, key.Name))
 				}
 			default:
 				return nil, nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "value type %s is not supported for map column type %s", valueType, column.Type)

tests/conftest.py

@@ -19,6 +19,7 @@ pytest_plugins = [
     "fixtures.traces",
     "fixtures.metrics",
     "fixtures.queriercommon",
+    "fixtures.semconvfamilies",
     "fixtures.metadata",
     "fixtures.meter",
     "fixtures.browser",

tests/fixtures/semconvfamilies.py

@@ -0,0 +1,75 @@
+"""Seed data for the semconv family matrix tests.
+
+Four identities cover every fleet state of the deployment.environment(.name)
+family. The tests assert which identities a filter returns, so BOTH (a row
+that carries the two spellings with different values) and NEITHER (a keyless
+row) are the point of most cases.
+
+Each row carries its family pairs in the resource attributes and in the span
+attributes, so one fleet serves the resource-context and attribute-context
+matrices. The same rows exist as logs for the logs literalness guard.
+"""
+
+from collections.abc import Callable, Generator
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from fixtures.logs import Logs
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
+
+PREFIX = "semconv-fam"
+CURRENT_KEY = "deployment.environment.name"
+OLD_KEY = "deployment.environment"
+
+# Row identities. The span name, the log body, and service.name are the identity.
+OLD = f"{PREFIX}-old"  # only the old spelling, value "production"
+NEW = f"{PREFIX}-new"  # only the current spelling, value "production"
+BOTH = f"{PREFIX}-both"  # current "staging" and old "production" - the conflict row
+NEITHER = f"{PREFIX}-neither"  # no member at all
+
+_ROWS = [
+    (OLD, {OLD_KEY: "production"}, timedelta(seconds=4)),
+    (NEW, {CURRENT_KEY: "production"}, timedelta(seconds=3)),
+    (BOTH, {CURRENT_KEY: "staging", OLD_KEY: "production"}, timedelta(seconds=2)),
+    (NEITHER, {}, timedelta(seconds=1)),
+]
+
+
+@pytest.fixture(name="family_fleet", scope="function")
+def family_fleet(
+    insert_logs: Callable[[list[Logs]], None],
+    insert_traces: Callable[[list[Traces]], None],
+) -> Generator[datetime]:
+    """Inserts one span and one log per identity and yields the base
+    timestamp."""
+    now = datetime.now(tz=UTC).replace(microsecond=0) - timedelta(minutes=1)
+
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - offset,
+                duration=timedelta(milliseconds=10),
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                name=identity,
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                resources={"service.name": identity, **family},
+                attributes=dict(family),
+            )
+            for identity, family, offset in _ROWS
+        ]
+    )
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - offset,
+                body=identity,
+                resources={"service.name": identity, **family},
+                attributes=dict(family),
+            )
+            for identity, family, offset in _ROWS
+        ]
+    )
+    yield now

tests/integration/tests/semconvfamilies/01_family_matrix.py

@@ -0,0 +1,220 @@
+"""The phase-1 matrix for semantic-convention family resolution.
+
+The package runs SigNoz with resolve_semconv_families on. The fleet in
+fixtures/semconvfamilies.py has one identity per state: OLD (old spelling
+only), NEW (current only), BOTH (current "staging" and old "production"),
+NEITHER (keyless). Each case asserts which identities a filter returns, with
+either spelling as the requested name and for both contexts.
+
+The pinned facts:
+  - Both spellings resolve to the same merged field; the result sets do not
+    depend on the requested spelling.
+  - The current spelling wins on a row that carries both (BOTH reads
+    "staging", never "production").
+  - Negative operators keep keyless rows (NEITHER), exactly like a single
+    key; presence stays an explicit EXISTS opt-in.
+  - Logs stay literal: only traces have family support today.
+  - With the flag off, everything stays literal.
+"""
+
+from collections.abc import Callable
+from datetime import datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import (
+    RequestType,
+    build_aggregation,
+    build_group_by_field,
+    build_order_by,
+    build_raw_query,
+    build_traces_scalar_query,
+    get_column_data_from_response,
+    make_query_request,
+)
+from fixtures.semconvfamilies import (
+    BOTH,
+    CURRENT_KEY,
+    NEITHER,
+    NEW,
+    OLD,
+    OLD_KEY,
+    PREFIX,
+)
+
+FILTER_MATRIX = [
+    pytest.param("{key} = 'production'", {OLD, NEW}, id="eq_matches_either_spelling"),
+    pytest.param("{key} = 'staging'", {BOTH}, id="eq_current_wins_on_conflict"),
+    pytest.param("{key} != 'production'", {BOTH, NEITHER}, id="neq_keeps_keyless_and_conflict"),
+    pytest.param("{key} IN ['production', 'staging']", {OLD, NEW, BOTH}, id="in_matches_merged_value"),
+    pytest.param("{key} NOT IN ['production']", {BOTH, NEITHER}, id="not_in_keeps_keyless"),
+    pytest.param("{key} LIKE '%prod%'", {OLD, NEW}, id="like_matches_merged_value"),
+    pytest.param("{key} EXISTS", {OLD, NEW, BOTH}, id="exists_is_any_member"),
+    pytest.param("{key} NOT EXISTS", {NEITHER}, id="not_exists_is_no_member"),
+    pytest.param("{key} != 'production' AND {key} EXISTS", {BOTH}, id="neq_composed_with_exists"),
+]
+
+LITERAL_MATRIX = [
+    pytest.param("{key} = 'production'", {NEW}, id="literal_eq_reads_one_spelling"),
+    pytest.param("{key} != 'production'", {OLD, BOTH, NEITHER}, id="literal_neq_reads_one_spelling"),
+]
+
+
+def _trace_identities(
+    signoz: types.SigNoz,
+    token: str,
+    base: datetime,
+    expression: str,
+    signal: str = "traces",
+) -> set[str]:
+    identity_field = "span.name" if signal == "traces" else "body"
+    identity_column = "name" if signal == "traces" else "body"
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((base - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((base + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                signal,
+                limit=100,
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": identity_field}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    # Sets keep the assertion stable when the shared stack is reused and older
+    # rows with the same identities remain.
+    return {name for name in get_column_data_from_response(response.json(), identity_column) if name.startswith(PREFIX)}
+
+
+@pytest.mark.parametrize("expression_template,expected", FILTER_MATRIX)
+@pytest.mark.parametrize("requested_key", [CURRENT_KEY, OLD_KEY], ids=["current", "old"])
+@pytest.mark.parametrize("context", ["resource", "attribute"])
+def test_family_filters(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+    context: str,
+    requested_key: str,
+    expression_template: str,
+    expected: set[str],
+) -> None:
+    """One matrix cell: a filter on one spelling, in one context. The result
+    set is a property of the family, not of the requested spelling."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(key=f"{context}.{requested_key}")
+    assert _trace_identities(signoz, token, family_fleet, expression) == expected, expression
+
+
+@pytest.mark.parametrize("expression_template,expected", LITERAL_MATRIX)
+def test_flag_off_stays_literal(
+    signoz_families_off: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+    expression_template: str,
+    expected: set[str],
+) -> None:
+    """The same fleet through an instance with the flag at its default: the
+    current spelling reads only rows that carry the current spelling."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(key=f"resource.{CURRENT_KEY}")
+    assert _trace_identities(signoz_families_off, token, family_fleet, expression) == expected, expression
+
+
+@pytest.mark.parametrize("expression_template,expected", LITERAL_MATRIX)
+def test_logs_stay_literal_with_flag_on(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+    expression_template: str,
+    expected: set[str],
+) -> None:
+    """Only traces have family support. The same filters on the logs copy of
+    the fleet behave literally even with the flag on."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(key=f"resource.{CURRENT_KEY}")
+    assert _trace_identities(signoz, token, family_fleet, expression, signal="logs") == expected, expression
+
+
+def test_group_by_merges_and_echoes_requested_spelling(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+) -> None:
+    """Group by the current spelling over the fleet: OLD and NEW land in one
+    "production" group, BOTH lands in "staging", and the group column carries
+    the requested spelling."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.SCALAR,
+        queries=[
+            build_traces_scalar_query(
+                [build_aggregation("count()")],
+                filter_expression=f"service.name LIKE '{PREFIX}%'",
+                group_by=[build_group_by_field(CURRENT_KEY, "string", "resource")],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    result = response.json()["data"]["data"]["results"][0]
+    group_column = result["columns"][0]
+    assert group_column["name"] == CURRENT_KEY, group_column
+    assert group_column["columnType"] == "group", group_column
+
+    groups = {row[0] for row in result["data"]}
+    assert {"production", "staging"}.issubset(groups), groups
+    assert None in groups, groups
+
+
+def test_bare_name_prefers_resource_and_warns(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+) -> None:
+    """The fleet stores the family under the resource and the attribute
+    contexts, so a bare name is ambiguous. Resolution warns and keeps the
+    resource side; the family survives the collision as one unit."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                "traces",
+                limit=100,
+                filter_expression=f"{CURRENT_KEY} = 'production'",
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": "span.name"}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    matched = {name for name in get_column_data_from_response(response.json(), "name") if name.startswith(PREFIX)}
+    assert matched == {OLD, NEW}
+
+    warning = response.json()["data"].get("warning") or {}
+    messages = " ".join(entry.get("message", "") for entry in warning.get("warnings", []))
+    assert "ambiguous" in messages.lower(), messages

tests/integration/tests/semconvfamilies/conftest.py

@@ -0,0 +1,56 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.signoz import create_signoz
+
+
+@pytest.fixture(name="signoz", scope="package")
+def signoz_semconv_families(
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    """Package-scoped SigNoz with resolve_semconv_families on."""
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-semconv-families",
+        env_overrides={
+            "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_RESOLVE__SEMCONV__FAMILIES": True,
+        },
+    )
+
+
+@pytest.fixture(name="signoz_families_off", scope="package")
+def signoz_families_off(
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    """A second instance with the flag at its default (off). It shares the
+    sqlstore and clickhouse, so the same admin token and seeded rows work."""
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-semconv-families-off",
+        env_overrides={},
+    )