feat(traces-qb): read the list-view attributes bag from the JSON column

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

Assisted-by: Claude Opus 4.8

pkg/querier/consume.go

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

pkg/querier/consume_test.go

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

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, key.Name)
+		rawPath := fmt.Sprintf("%s.%s", columnName, ClickHouseIdentifier(key.Name))
 		if exists {
 			return rawPath + " IS NOT NULL", nil
 		}

pkg/statementbuilder/tracesstatementbuilder/bulk_attributes_test.go

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

pkg/statementbuilder/tracesstatementbuilder/statement_builder.go

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

pkg/telemetrymetadata/metadata.go

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

pkg/telemetryschema/tracestelemetryschema/const.go

@@ -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 (

pkg/telemetryschema/tracestelemetryschema/field_mapper.go

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

pkg/telemetryschema/tracestelemetryschema/field_mapper_attributes_json_test.go

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

pkg/telemetryschema/tracestelemetryschema/field_mapper_attributes_promoted_test.go

@@ -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")
+	})
+}

pkg/telemetryschema/tracestelemetryschema/test_data.go

@@ -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 {

pkg/types/querybuildertypes/querybuildertypesv5/evolution.go

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

pkg/types/telemetrytypes/evolution_metadata.go

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

pkg/types/telemetrytypes/store.go

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

pkg/types/telemetrytypes/telemetrytypestest/metadata_store.go

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

pkg/types/telemetrytypes/telemetrytypestest/metadata_store_test.go

@@ -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))
+	})
+}

tests/fixtures/traces.py

@@ -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,

tests/integration/tests/queriertraces/13_attributes_json_evolution.py

@@ -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