refactor(aiobservabilitytypes): normalize message inputs in the formatter

The formatter now json-normalizes structured inputs, so callers hand it the
driver's native JSON Variants directly instead of pre-flattening per consumer.

Assisted-by: Claude Opus 4.8

go.mod

@@ -12,7 +12,6 @@ require (
 	github.com/SigNoz/signoz-otel-collector v0.144.6
 	github.com/antlr4-go/antlr/v4 v4.13.1
 	github.com/antonmedv/expr v1.15.3
-	github.com/bytedance/sonic v1.14.1
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/coreos/go-oidc/v3 v3.17.0
 	github.com/dgraph-io/ristretto/v2 v2.3.0
@@ -112,6 +111,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.9 // indirect
 	github.com/aws/smithy-go v1.24.2 // indirect
 	github.com/bytedance/gopkg v0.1.3 // indirect
+	github.com/bytedance/sonic v1.14.1 // indirect
 	github.com/bytedance/sonic/loader v0.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // indirect

pkg/querier/consume.go

@@ -17,7 +17,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/errors"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/spantypes"
-	"github.com/SigNoz/signoz/pkg/types/telemetrystoretypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
 
@@ -40,11 +39,18 @@ func stripKeyAlias(name string) string {
 	return keyAliasRe.ReplaceAllString(name, "")
 }
 
-// unwrapVariant returns the concrete value inside the chcol.Variant envelope the driver scans a
-// Dynamic column — a JSON path such as body_v2.level — into.
-func unwrapVariant(val any) any {
-	if v, ok := val.(chcol.Variant); ok {
+// unwrapVariant decodes a scan envelope: a chcol.Variant to its value, a chcol.JSON column to a map.
+// The attributes bag decodes flat so its dotted keys merge with the legacy attribute maps and a
+// scalar-and-object key stays two keys; every other JSON column decodes nested.
+func unwrapVariant(name string, val any) any {
+	switch v := val.(type) {
+	case chcol.Variant:
 		return v.Any()
+	case chcol.JSON:
+		if name == "attributes" {
+			return v.ValuesByPath()
+		}
+		return v.NestedMap()
 	}
 	return val
 }
@@ -54,11 +60,11 @@ func unwrapVariant(val any) any {
 // series. JSON goes through encoding/json for its sorted map keys: ClickHouse groups documents by
 // structure, so two rows it considers equal have to produce the same label.
 func labelValue(val any) string {
-	val = unwrapVariant(val)
+	val = unwrapVariant("", val)
 	if val == nil {
 		return ""
 	}
-	if v, ok := val.(telemetrystoretypes.JSONValue); ok {
+	if v, ok := val.(map[string]any); ok {
 		if raw, err := json.Marshal(v); err == nil {
 			return string(raw)
 		}
@@ -204,7 +210,7 @@ func readAsTimeSeries(rows driver.Rows, queryWindow *qbtypes.TimeRange, step qbt
 					Value: *val,
 				})
 
-			case *telemetrystoretypes.JSONValue, *chcol.Variant:
+			case *chcol.JSON, *chcol.Variant:
 				val := labelValue(derefValue(ptr))
 				lblVals = append(lblVals, val)
 				lblObjs = append(lblObjs, &qbtypes.Label{
@@ -478,7 +484,7 @@ func readAsScalar(rows driver.Rows, queryName string) (*qbtypes.ScalarData, erro
 		// 2. deref each slot into the output row
 		row := make([]any, len(scan))
 		for i, cell := range scan {
-			row[i] = unwrapVariant(derefValue(cell))
+			row[i] = unwrapVariant(cd[i].Name, derefValue(cell))
 		}
 		data = append(data, row)
 	}
@@ -536,7 +542,7 @@ func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
 			name := stripKeyAlias(colNames[i])
 
 			// de-reference the typed pointer to any
-			val := unwrapVariant(reflect.ValueOf(cellPtr).Elem().Interface())
+			val := unwrapVariant(name, reflect.ValueOf(cellPtr).Elem().Interface())
 
 			// special-case: timestamp column
 			if name == "timestamp" || name == "timestamp_datetime" {
@@ -570,17 +576,18 @@ func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
 // unified "attributes" and "resource" keys, and parses the stringified `events`
 // and `links` columns into structured slices. Raw DB columns are removed.
 //
-// The `attributes` JSON column is flattened in first and the legacy maps merged over it, so maps win on collision.
+// The legacy attribute maps are merged over the JSON attributes bag, so maps win on collision.
 func mergeSpanAttributeColumns(data map[string]any) {
 	attrStr, hasStr := data["attributes_string"]
 	attrNum, hasNum := data["attributes_number"]
 	attrBool, hasBool := data["attributes_bool"]
-	attrJSON, _ := data["attributes"].(telemetrystoretypes.JSONValue)
+	attributes, _ := data["attributes"].(map[string]any)
 	// todo(nitya): move to resource json
 	resStr, hasRes := data["resources_string"]
-	if hasStr || hasNum || hasBool || attrJSON != nil || hasRes {
-		attributes := make(map[string]any)
-		attrJSON.FlattenInto("", attributes)
+	if hasStr || hasNum || hasBool || attributes != nil || hasRes {
+		if attributes == nil {
+			attributes = map[string]any{}
+		}
 		if m, ok := attrStr.(map[string]string); ok {
 			for k, v := range m {
 				attributes[k] = v

pkg/querier/consume_test.go

@@ -3,14 +3,9 @@ package querier
 import (
 	"reflect"
 	"testing"
-	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
-	cmock "github.com/SigNoz/clickhouse-go-mock"
-	"github.com/SigNoz/signoz/pkg/telemetrystore"
-	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/spantypes"
-	"github.com/SigNoz/signoz/pkg/types/telemetrystoretypes"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -83,101 +78,47 @@ func TestMergeSpanAttributeColumns_ParsesEventsAndLinks(t *testing.T) {
 	}
 }
 
-// A ClickHouse query can put a JSON column in the result of any request type — e.g.
-// `select * from signoz_logs.logs_v2` on a body_v2 stack, where `*` covers body_v2.
-func TestConsume_JSONColumn(t *testing.T) {
-	ts := time.Date(2026, 8, 14, 10, 0, 0, 0, time.UTC)
-	body := `{"level":"error","attrs":{"code":500}}`
-	wantBody := telemetrystoretypes.JSONValue{
-		"level": "error",
-		"attrs": map[string]any{"code": float64(500)},
+func TestUnwrapVariant(t *testing.T) {
+	j := chcol.NewJSON()
+	j.SetValueAtPath("level", "error")
+	j.SetValueAtPath("attrs.code", int64(500))
+
+	testCases := []struct {
+		name   string
+		column string
+		input  any
+		want   any
+	}{
+		{name: "VariantScalar", column: "", input: chcol.NewDynamicWithType("error", "String"), want: "error"},
+		{name: "EmptyDynamic", column: "", input: chcol.Dynamic{}, want: nil},
+		{name: "PlainValuePassthrough", column: "", input: uint64(3), want: uint64(3)},
+		{name: "JSONColumnNested", column: "body_v2", input: *j, want: map[string]any{"level": "error", "attrs": map[string]any{"code": int64(500)}}},
+		{name: "AttributesColumnFlat", column: "attributes", input: *j, want: map[string]any{"level": "error", "attrs.code": int64(500)}},
 	}
 
-	// the scalar reader reuses its scan slots across rows, so each row must still carry its own body
-	t.Run("scalar", func(t *testing.T) {
-		rows := telemetrystore.WrapRows(cmock.NewRows([]cmock.ColumnType{
-			{Name: "body_v2", Type: "JSON"},
-			{Name: "__result_0", Type: "UInt64"},
-		}, [][]any{{body, uint64(3)}, {`{"level":"warn"}`, uint64(1)}}))
-
-		payload, err := consume(rows, qbtypes.RequestTypeScalar, nil, qbtypes.Step{}, "A")
-		require.NoError(t, err)
-
-		data := payload.(*qbtypes.ScalarData)
-		require.Len(t, data.Data, 2)
-		assert.Equal(t, wantBody, data.Data[0][0])
-		assert.Equal(t, uint64(3), data.Data[0][1])
-		assert.Equal(t, telemetrystoretypes.JSONValue{"level": "warn"}, data.Data[1][0])
-		assert.Equal(t, uint64(1), data.Data[1][1])
-	})
-
-	t.Run("time series", func(t *testing.T) {
-		rows := telemetrystore.WrapRows(cmock.NewRows([]cmock.ColumnType{
-			{Name: "ts", Type: "DateTime"},
-			{Name: "body_v2", Type: "JSON"},
-			{Name: "__result_0", Type: "UInt64"},
-		}, [][]any{{ts, body, uint64(3)}}))
-
-		payload, err := consume(rows, qbtypes.RequestTypeTimeSeries, nil, qbtypes.Step{}, "A")
-		require.NoError(t, err)
-
-		data := payload.(*qbtypes.TimeSeriesData)
-		require.Len(t, data.Aggregations, 1)
-		require.Len(t, data.Aggregations[0].Series, 1)
-		require.Len(t, data.Aggregations[0].Series[0].Values, 1)
-		assert.Equal(t, float64(3), data.Aggregations[0].Series[0].Values[0].Value)
-	})
-
-	// grouping by a JSON column is legal in ClickHouse, so each document has to label its own
-	// series rather than being dropped, which would merge every group into one
-	t.Run("time series grouped by the JSON column", func(t *testing.T) {
-		rows := telemetrystore.WrapRows(cmock.NewRows([]cmock.ColumnType{
-			{Name: "ts", Type: "DateTime"},
-			{Name: "body_v2", Type: "JSON"},
-			{Name: "__result_0", Type: "UInt64"},
-		}, [][]any{
-			{ts, `{"level":"error"}`, uint64(7)},
-			{ts, `{"level":"warn"}`, uint64(2)},
-		}))
-
-		payload, err := consume(rows, qbtypes.RequestTypeTimeSeries, nil, qbtypes.Step{}, "A")
-		require.NoError(t, err)
-
-		data := payload.(*qbtypes.TimeSeriesData)
-		require.Len(t, data.Aggregations, 1)
-		require.Len(t, data.Aggregations[0].Series, 2)
-
-		got := map[string]float64{}
-		for _, series := range data.Aggregations[0].Series {
-			require.Len(t, series.Labels, 1)
-			require.Len(t, series.Values, 1)
-			got[series.Labels[0].Value.(string)] = series.Values[0].Value
-		}
-		assert.Equal(t, map[string]float64{`{"level":"error"}`: 7, `{"level":"warn"}`: 2}, got)
-	})
-
-	t.Run("raw", func(t *testing.T) {
-		rows := telemetrystore.WrapRows(cmock.NewRows([]cmock.ColumnType{
-			{Name: "timestamp", Type: "DateTime"},
-			{Name: "body_v2", Type: "JSON"},
-		}, [][]any{{ts, body}}))
-
-		payload, err := consume(rows, qbtypes.RequestTypeRaw, nil, qbtypes.Step{}, "A")
-		require.NoError(t, err)
-
-		data := payload.(*qbtypes.RawData)
-		require.Len(t, data.Rows, 1)
-		assert.Equal(t, ts, data.Rows[0].Timestamp.UTC())
-		assert.Equal(t, wantBody, data.Rows[0].Data["body_v2"])
-	})
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, unwrapVariant(testCase.column, testCase.input))
+		})
+	}
 }
 
-// A JSON path (e.g. `body_v2.level`) comes back as a Dynamic column, which the driver scans
-// into a chcol.Variant envelope rather than the value itself.
-func TestUnwrapVariant(t *testing.T) {
-	assert.Equal(t, "error", unwrapVariant(chcol.NewDynamicWithType("error", "String")))
-	assert.Nil(t, unwrapVariant(chcol.Dynamic{}))
-	assert.Equal(t, uint64(3), unwrapVariant(uint64(3)))
+func TestLabelValue(t *testing.T) {
+	testCases := []struct {
+		name  string
+		input any
+		want  string
+	}{
+		{name: "Nil", input: nil, want: ""},
+		{name: "Variant", input: chcol.NewDynamicWithType("error", "String"), want: "error"},
+		{name: "JSONMap", input: map[string]any{"level": "error", "attrs": map[string]any{"code": 500}}, want: `{"attrs":{"code":500},"level":"error"}`},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, labelValue(testCase.input))
+		})
+	}
 }
 
 func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
@@ -205,12 +146,9 @@ func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
 		want map[string]any
 	}{
 		{
-			name: "JSONOnly_FlattensNestedPaths_PreservesTypes",
+			name: "JSONOnly_FlatPathsPreserveTypes",
 			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{
-					"http":      map[string]any{"route": "/api/pay", "retry": map[string]any{"count": float64(3)}},
-					"cache.hit": true,
-				},
+				"attributes": map[string]any{"http.route": "/api/pay", "http.retry.count": float64(3), "cache.hit": true},
 			},
 			want: map[string]any{"http.route": "/api/pay", "http.retry.count": float64(3), "cache.hit": true},
 		},
@@ -219,17 +157,24 @@ func TestMergeSpanAttributeColumns_JSONColumn(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"}, "only.json": "j"},
+				"attributes":        map[string]any{"http.route": "/new", "only.json": "j"},
 			},
 			want: map[string]any{"http.route": "/old", "only.map": "m", "http.status": float64(500), "only.json": "j"},
 		},
+		{
+			name: "ScalarAndObjectKey_BothSurvive",
+			data: map[string]any{
+				"attributes": map[string]any{"db.function": "refresh", "db.function.arg": float64(2)},
+			},
+			want: map[string]any{"db.function": "refresh", "db.function.arg": float64(2)},
+		},
 		{
 			name: "MapOnly_EmptyJSONDoc_KeepsMapValues",
 			data: map[string]any{
 				"attributes_string": map[string]string{"http.route": "/map"},
 				"attributes_number": map[string]float64{"http.status": 200},
 				"attributes_bool":   map[string]bool{"cache.hit": true},
-				"attributes":        telemetrystoretypes.JSONValue{},
+				"attributes":        map[string]any{},
 			},
 			want: map[string]any{"http.route": "/map", "http.status": float64(200), "cache.hit": true},
 		},
@@ -237,60 +182,53 @@ func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
 			name: "MapOnly_NilJSON_BehavesAsAbsent",
 			data: map[string]any{
 				"attributes_string": map[string]string{"http.route": "/map"},
-				"attributes":        telemetrystoretypes.JSONValue(nil),
+				"attributes":        map[string]any(nil),
 			},
 			want: map[string]any{"http.route": "/map"},
 		},
 		{
 			name: "Arrays_StayLeafValues",
 			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{"http": map[string]any{"tags": []any{"a", "b"}, "codes": []any{float64(1), float64(2)}}},
+				"attributes": map[string]any{"http.tags": []any{"a", "b"}, "http.codes": []any{float64(1), float64(2)}},
 			},
 			want: map[string]any{"http.tags": []any{"a", "b"}, "http.codes": []any{float64(1), float64(2)}},
 		},
 		{
 			name: "TopLevelArrayOfMaps_StaysNativeLeaf",
 			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{"key": []any{map[string]any{"a": float64(1)}, map[string]any{"b": float64(2)}}},
+				"attributes": map[string]any{"key": []any{map[string]any{"a": float64(1)}, map[string]any{"b": float64(2)}}},
 			},
 			want: map[string]any{"key": []any{map[string]any{"a": float64(1)}, map[string]any{"b": float64(2)}}},
 		},
 		{
-			name: "NestedArrayOfMaps_StaysNativeLeaf_NoIndexPaths",
+			name: "ArrayOfMapsAtDottedPath_StaysNativeLeaf",
 			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{"http": map[string]any{"items": []any{map[string]any{"a": float64(1)}}}},
+				"attributes": map[string]any{"http.items": []any{map[string]any{"a": float64(1)}}},
 			},
 			want: map[string]any{"http.items": []any{map[string]any{"a": float64(1)}}},
 		},
 		{
-			name: "DualWritten_NestedArray_IndexKeysAndJSONArrayCoexist",
+			name: "DualWritten_IndexKeysAndJSONArrayCoexist",
 			data: map[string]any{
 				"attributes_number": map[string]float64{"http.items.0.a": 1},
-				"attributes":        telemetrystoretypes.JSONValue{"http": map[string]any{"items": []any{map[string]any{"a": float64(1)}}}},
+				"attributes":        map[string]any{"http.items": []any{map[string]any{"a": float64(1)}}},
 			},
 			want: map[string]any{"http.items.0.a": float64(1), "http.items": []any{map[string]any{"a": float64(1)}}},
 		},
 		{
 			name: "JSONNull_KeptAsNil",
 			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{"k": nil},
+				"attributes": map[string]any{"k": nil},
 			},
 			want: map[string]any{"k": nil},
 		},
 		{
-			name: "KeyIsLeafValue_NotFlattened",
+			name: "ScalarValueStaysScalar",
 			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{"http": "plaintext"},
+				"attributes": map[string]any{"http": "plaintext"},
 			},
 			want: map[string]any{"http": "plaintext"},
 		},
-		{
-			name: "KeyIsParent_FlattensToDottedPath",
-			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{"http": map[string]any{"route": "/a"}},
-			},
-			want: map[string]any{"http.route": "/a"},
-		},
 	}
 
 	for _, testCase := range testCases {

pkg/querier/postprocess.go

@@ -16,7 +16,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/types/featuretypes"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
-	"github.com/SigNoz/signoz/pkg/types/telemetrystoretypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
@@ -1202,7 +1201,7 @@ func (q *querier) postProcessLogBody(ctx context.Context, orgID valuer.UUID, res
 // carried one. Anything that is not a decoded document — the legacy string body, a NULL cell —
 // is legal under these names and left alone.
 func stripEmptyBodyMessage(val any) {
-	bodyMap, ok := val.(telemetrystoretypes.JSONValue)
+	bodyMap, ok := val.(map[string]any)
 	if !ok {
 		return
 	}

pkg/telemetrystore/clickhousetelemetrystore/provider.go

@@ -97,8 +97,8 @@ func New(ctx context.Context, providerSettings factory.ProviderSettings, config
 	options.MaxIdleConns = config.Connection.MaxIdleConns
 	options.MaxOpenConns = config.Connection.MaxOpenConns
 	options.DialTimeout = config.Connection.DialTimeout
-	// This is to avoid the driver decoding issues with JSON columns
-	options.Settings["output_format_native_write_json_as_string"] = 1
+	// Native flattened JSON serialization (CH 25.6+): clickhouse-go mis-decodes JSON(max_dynamic_paths=0) columns without it.
+	options.Settings["output_format_native_use_flattened_dynamic_and_json_serialization"] = 1
 
 	chConn, err := clickhouse.Open(options)
 	if err != nil {
@@ -184,7 +184,7 @@ func (p *provider) Query(ctx context.Context, query string, args ...interface{})
 	}
 
 	return &rowsWithHooks{
-		Rows:    telemetrystore.WrapRows(rows),
+		Rows:    rows,
 		ctx:     ctx,
 		event:   event,
 		onClose: func() { telemetrystore.WrapAfterQuery(p.hooks, ctx, event) },

pkg/telemetrystore/rows.go

@@ -1,39 +0,0 @@
-package telemetrystore
-
-import (
-	"reflect"
-	"strings"
-
-	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
-	"github.com/SigNoz/signoz/pkg/types/telemetrystoretypes"
-)
-
-// WrapRows reports JSONValue as the scan type of every JSON column. Nested JSON — Array(JSON),
-// Map(String, JSON) — is not covered.
-func WrapRows(rows driver.Rows) driver.Rows {
-	return &rowsWithJSONScanType{Rows: rows}
-}
-
-type rowsWithJSONScanType struct {
-	driver.Rows
-}
-
-func (r *rowsWithJSONScanType) ColumnTypes() []driver.ColumnType {
-	colTypes := r.Rows.ColumnTypes()
-	wrapped := make([]driver.ColumnType, len(colTypes))
-	for i, colType := range colTypes {
-		wrapped[i] = colType
-		if strings.HasPrefix(strings.ToUpper(colType.DatabaseTypeName()), "JSON") {
-			wrapped[i] = jsonColumnType{ColumnType: colType}
-		}
-	}
-	return wrapped
-}
-
-type jsonColumnType struct {
-	driver.ColumnType
-}
-
-func (jsonColumnType) ScanType() reflect.Type {
-	return reflect.TypeFor[telemetrystoretypes.JSONValue]()
-}

pkg/telemetrystore/telemetrystoretest/conn.go

@@ -1,23 +0,0 @@
-package telemetrystoretest
-
-import (
-	"context"
-
-	"github.com/ClickHouse/clickhouse-go/v2"
-	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
-	"github.com/SigNoz/signoz/pkg/telemetrystore"
-)
-
-// conn wraps rows the way the clickhouse provider does, so mocked JSON columns report the scan
-// type they do in production.
-type conn struct {
-	clickhouse.Conn
-}
-
-func (c conn) Query(ctx context.Context, query string, args ...any) (driver.Rows, error) {
-	rows, err := c.Conn.Query(ctx, query, args...)
-	if err != nil {
-		return nil, err
-	}
-	return telemetrystore.WrapRows(rows), nil
-}

pkg/telemetrystore/telemetrystoretest/provider.go

@@ -32,7 +32,7 @@ func New(_ telemetrystore.Config, matcher sqlmock.QueryMatcher) *Provider {
 
 // ClickhouseDB returns the mock Clickhouse connection.
 func (p *Provider) ClickhouseDB() clickhouse.Conn {
-	return conn{Conn: p.clickhouseDB.(clickhouse.Conn)}
+	return p.clickhouseDB.(clickhouse.Conn)
 }
 
 // Cluster returns the cluster name.

pkg/types/aiobservabilitytypes/message_format.go

@@ -139,12 +139,13 @@ func genericMessage(value any) OutputMessage {
 	return OutputMessage{Parts: Parts{part(GenericPart{"type": FormatterGeneric, "content": stringOf(value)})}}
 }
 
-// decode parses a JSON-encoded attribute value, twice when an SDK encoded the string again.
-// Anything that is not JSON is returned as is.
+// decode parses a JSON-encoded attribute value, twice when an SDK encoded the string again. A
+// structured value is normalized to plain []any and map[string]any so side can assert it; the
+// ClickHouse driver decodes JSON columns into wrapper types that side would otherwise miss.
 func decode(value any) any {
 	s, ok := value.(string)
 	if !ok {
-		return value
+		return plainJSON(value)
 	}
 	trimmed := strings.TrimSpace(s)
 	if trimmed == "" || !strings.ContainsAny(trimmed[:1], `[{"`) {
@@ -160,6 +161,20 @@ func decode(value any) any {
 	return decoded
 }
 
+// plainJSON round-trips a structured value through encoding/json, turning wrapper types such as the
+// ClickHouse driver's JSON Variants into plain Go types. Values that cannot be marshaled pass through.
+func plainJSON(value any) any {
+	raw, err := json.Marshal(value)
+	if err != nil {
+		return value
+	}
+	var decoded any
+	if err := json.Unmarshal(raw, &decoded); err != nil {
+		return value
+	}
+	return decoded
+}
+
 func role(value string) Role {
 	lower := strings.ToLower(strings.TrimSpace(value))
 	switch lower {

pkg/types/spantypes/thread.go

@@ -181,7 +181,7 @@ func newThreadSpan(traceID string, storable *StorableSpan) *ThreadSpan {
 	attributes := threadAttributes(storable)
 	// Spans from before the JSON rollout have no document and are not decoded: the legacy maps
 	// hold an object split into one key per field.
-	formatted := aiobservabilitytypes.FormatMessages(storable.AttributesJSON)
+	formatted := aiobservabilitytypes.FormatMessages(storable.AttributesJSON.NestedMap())
 	return &ThreadSpan{
 		SpanID:            storable.SpanID,
 		TraceID:           traceID,
@@ -205,13 +205,12 @@ func newThreadSpan(traceID string, storable *StorableSpan) *ThreadSpan {
 	}
 }
 
-// threadAttributes flattens the JSON column into dotted keys, as the querier does for list
+// threadAttributes returns the span's flat dotted attributes, as the querier does for list
 // responses, and falls back to the legacy maps for spans written before the JSON rollout.
 func threadAttributes(storable *StorableSpan) map[string]any {
-	if len(storable.AttributesJSON) == 0 {
+	attributes := storable.AttributesJSON.ValuesByPath()
+	if len(attributes) == 0 {
 		return storable.Attributes()
 	}
-	attributes := make(map[string]any, len(storable.AttributesJSON))
-	storable.AttributesJSON.FlattenInto("", attributes)
 	return attributes
 }

pkg/types/spantypes/thread_test.go

@@ -3,11 +3,21 @@ package spantypes
 import (
 	"testing"
 
-	"github.com/SigNoz/signoz/pkg/types/telemetrystoretypes"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
+	"github.com/SigNoz/signoz/pkg/types/aiobservabilitytypes"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+// jsonAttrs builds a native JSON column from dotted leaf paths, as ClickHouse stores them.
+func jsonAttrs(paths map[string]any) chcol.JSON {
+	j := chcol.NewJSON()
+	for path, value := range paths {
+		j.SetValueAtPath(path, value)
+	}
+	return *j
+}
+
 func TestNewThreadQuery_Cursor(t *testing.T) {
 	testCases := []struct {
 		name    string
@@ -48,7 +58,7 @@ func TestThreadAttributes(t *testing.T) {
 		{
 			name: "JSONColumnPresent_LegacyMapsIgnored",
 			span: StorableSpan{
-				AttributesJSON:   telemetrystoretypes.JSONValue{"gen_ai": map[string]any{"request": map[string]any{"model": "json"}}},
+				AttributesJSON:   jsonAttrs(map[string]any{"gen_ai.request.model": "json"}),
 				AttributesString: map[string]string{"gen_ai.request.model": "map", "http.method": "GET"},
 			},
 			wantAttrs: map[string]any{"gen_ai.request.model": "json"},
@@ -73,3 +83,21 @@ func TestThreadAttributes(t *testing.T) {
 		})
 	}
 }
+
+func TestNewThreadSpan_FormatsJSONMessages(t *testing.T) {
+	storable := StorableSpan{
+		SpanID: "f1fa1bc863e94dd0",
+		AttributesJSON: jsonAttrs(map[string]any{
+			"gen_ai.input.messages":  []any{map[string]any{"role": "user", "content": "hi"}},
+			"gen_ai.output.messages": []any{map[string]any{"role": "assistant", "content": "hello"}},
+		}),
+	}
+
+	span := newThreadSpan("trace", &storable)
+
+	assert.Equal(t, aiobservabilitytypes.FormatterOpenAIChat, span.Formatter)
+	require.Len(t, span.FormattedInput, 1)
+	assert.Equal(t, aiobservabilitytypes.RoleUser, span.FormattedInput[0].Role)
+	require.Len(t, span.FormattedOutput, 1)
+	assert.Equal(t, aiobservabilitytypes.RoleAssistant, span.FormattedOutput[0].Role)
+}

pkg/types/spantypes/waterfall_span.go

@@ -7,8 +7,9 @@ import (
 	"sort"
 	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
+
 	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/types/telemetrystoretypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
 
@@ -94,36 +95,36 @@ type WaterfallSpan struct {
 
 // StorableSpan is the ClickHouse scan struct for the v3 waterfall query.
 type StorableSpan struct {
-	StartTime          time.Time                     `ch:"timestamp"`
-	DurationNano       uint64                        `ch:"duration_nano"`
-	SpanID             string                        `ch:"span_id"`
-	HasError           bool                          `ch:"has_error"`
-	Kind               int8                          `ch:"kind"`
-	ServiceName        string                        `ch:"resource_string_service$$name"`
-	Name               string                        `ch:"name"`
-	AttributesString   map[string]string             `ch:"attributes_string"`
-	AttributesNumber   map[string]float64            `ch:"attributes_number"`
-	AttributesBool     map[string]bool               `ch:"attributes_bool"`
-	AttributesJSON     telemetrystoretypes.JSONValue `ch:"attributes"`
-	ResourcesString    map[string]string             `ch:"resources_string"`
-	Events             []string                      `ch:"events"`
-	StatusMessage      string                        `ch:"status_message"`
-	StatusCodeString   string                        `ch:"status_code_string"`
-	SpanKind           string                        `ch:"kind_string"`
-	ParentSpanID       string                        `ch:"parent_span_id"`
-	Flags              uint32                        `ch:"flags"`
-	IsRemote           string                        `ch:"is_remote"`
-	TraceState         string                        `ch:"trace_state"`
-	StatusCode         int16                         `ch:"status_code"`
-	DBName             string                        `ch:"db_name"`
-	DBOperation        string                        `ch:"db_operation"`
-	HTTPMethod         string                        `ch:"http_method"`
-	HTTPURL            string                        `ch:"http_url"`
-	HTTPHost           string                        `ch:"http_host"`
-	ExternalHTTPMethod string                        `ch:"external_http_method"`
-	ExternalHTTPURL    string                        `ch:"external_http_url"`
-	ResponseStatusCode string                        `ch:"response_status_code"`
-	References         string                        `ch:"references"`
+	StartTime          time.Time          `ch:"timestamp"`
+	DurationNano       uint64             `ch:"duration_nano"`
+	SpanID             string             `ch:"span_id"`
+	HasError           bool               `ch:"has_error"`
+	Kind               int8               `ch:"kind"`
+	ServiceName        string             `ch:"resource_string_service$$name"`
+	Name               string             `ch:"name"`
+	AttributesString   map[string]string  `ch:"attributes_string"`
+	AttributesNumber   map[string]float64 `ch:"attributes_number"`
+	AttributesBool     map[string]bool    `ch:"attributes_bool"`
+	AttributesJSON     chcol.JSON         `ch:"attributes"`
+	ResourcesString    map[string]string  `ch:"resources_string"`
+	Events             []string           `ch:"events"`
+	StatusMessage      string             `ch:"status_message"`
+	StatusCodeString   string             `ch:"status_code_string"`
+	SpanKind           string             `ch:"kind_string"`
+	ParentSpanID       string             `ch:"parent_span_id"`
+	Flags              uint32             `ch:"flags"`
+	IsRemote           string             `ch:"is_remote"`
+	TraceState         string             `ch:"trace_state"`
+	StatusCode         int16              `ch:"status_code"`
+	DBName             string             `ch:"db_name"`
+	DBOperation        string             `ch:"db_operation"`
+	HTTPMethod         string             `ch:"http_method"`
+	HTTPURL            string             `ch:"http_url"`
+	HTTPHost           string             `ch:"http_host"`
+	ExternalHTTPMethod string             `ch:"external_http_method"`
+	ExternalHTTPURL    string             `ch:"external_http_url"`
+	ResponseStatusCode string             `ch:"response_status_code"`
+	References         string             `ch:"references"`
 }
 
 // MinimalSpan with only the fields needed to build the parent-child tree.

pkg/types/telemetrystoretypes/json.go

@@ -1,55 +0,0 @@
-package telemetrystoretypes
-
-import (
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/bytedance/sonic"
-)
-
-var ErrCodeUnmarshalJSONColumn = errors.MustNewCode("fail_unmarshal_json_column")
-
-// JSONValue is the scan target for a ClickHouse JSON column: the connection sets
-// output_format_native_write_json_as_string, so the column arrives as a raw document rather than
-// the chcol.JSON the driver reports as its scan type.
-type JSONValue map[string]any
-
-// Scan decodes into a fresh map every time: a scan target is reused across rows, and unmarshalling
-// into the map already there would both keep its keys and hand every row the same map.
-func (v *JSONValue) Scan(src any) error {
-	var raw []byte
-	switch value := src.(type) {
-	case nil:
-		*v = nil
-		return nil
-	case string:
-		raw = []byte(value)
-	case []byte:
-		raw = value
-	default:
-		return errors.NewInternalf(ErrCodeUnmarshalJSONColumn, "cannot decode %T as a JSON column", src)
-	}
-
-	decoded := JSONValue{}
-	if err := sonic.Unmarshal(raw, &decoded); err != nil {
-		return errors.WrapInternalf(err, ErrCodeUnmarshalJSONColumn, "failed to unmarshal JSON column")
-	}
-	*v = decoded
-	return nil
-}
-
-// FlattenInto writes v into out under dotted keys, overwriting existing keys.
-func (v JSONValue) FlattenInto(prefix string, out map[string]any) {
-	for k, value := range v {
-		key := k
-		if prefix != "" {
-			key = prefix + "." + k
-		}
-		switch child := value.(type) {
-		case map[string]any:
-			JSONValue(child).FlattenInto(key, out)
-		case JSONValue:
-			child.FlattenInto(key, out)
-		default:
-			out[key] = value
-		}
-	}
-}

tests/integration/tests/querier_json_body/09_scalar_object_body.py

@@ -0,0 +1,52 @@
+import json
+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.logs import Logs
+from fixtures.querier import RequestType, build_raw_query, get_rows, make_query_request
+
+SERVICE = "scalar-object-body"
+
+
+def test_scalar_and_object_body_key_collapses(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    export_json_types: Callable[[list[Logs]], None],
+) -> None:
+    """A log body key stored as both a scalar and an object (db.function and db.function.arg) keeps
+    only one: the body is returned nested, not flattened to dotted keys, so it cannot hold both."""
+    now = datetime.now(tz=UTC)
+    logs = [
+        Logs(
+            timestamp=now - timedelta(seconds=1),
+            resources={"service.name": SERVICE},
+            body_v2=json.dumps({"db.function": "refresh", "db.function.arg": 2}),
+            body_promoted="",
+        ),
+    ]
+    export_json_types(logs)
+    insert_logs(logs)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[build_raw_query("A", "logs", limit=10, filter_expression=f"service.name = '{SERVICE}'")],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    rows = get_rows(response)
+    assert len(rows) == 1
+    body = rows[0]["data"]["body"]
+    # The scalar survives as a string; the sibling object path db.function.arg is not preserved
+    # under it, so db.function is a plain value rather than an object.
+    assert body["db"]["function"] == "refresh"
+    assert isinstance(body["db"]["function"], str)

tests/integration/tests/queriertraces/14_scalar_object_attribute.py

@@ -0,0 +1,74 @@
+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 (
+    BuilderQuery,
+    OrderBy,
+    RequestType,
+    TelemetryFieldKey,
+    get_rows,
+    make_query_request,
+)
+from fixtures.traces import TraceIdGenerator, Traces
+
+
+def test_traces_scalar_and_object_key_both_survive(
+    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:
+    """A span attribute stored as both a scalar and an object prefix (db.function and
+    db.function.arg) keeps both dotted keys through the native JSON read. json_only writes only the
+    JSON column, so no legacy attribute map backfills the keys."""
+    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)
+
+    service = "scalar-object-service"
+    span_time = evolution_time + timedelta(minutes=5)
+    insert_traces(
+        [
+            Traces(
+                timestamp=span_time,
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                name="scalar and object",
+                resources={"service.name": service},
+                attributes={"db.function": "refresh", "db.function.arg": 2},
+                attribute_write_mode="json_only",
+            ),
+        ]
+    )
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((span_time - timedelta(minutes=1)).timestamp() * 1000),
+        end_ms=int((span_time + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            BuilderQuery(
+                signal="traces",
+                name="A",
+                limit=10,
+                filter_expression=f"resource.service.name = '{service}'",
+                order=[OrderBy(TelemetryFieldKey("timestamp"), "asc")],
+            ).to_dict()
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    rows = get_rows(response)
+    assert len(rows) == 1
+    attributes = rows[0]["data"]["attributes"]
+    assert attributes["db.function"] == "refresh"
+    assert isinstance(attributes["db.function"], str)
+    assert attributes["db.function.arg"] == 2
+    # the native read returns flat dotted keys, so no Go-side flattening is needed: db is not nested
+    assert "db" not in attributes