refactor(telemetrystoretypes): back JSONValue with chcol.JSON for on-demand flat/nested reads

Holding the raw column lets a scalar-and-object key survive and keeps JSONValue the single
scan target, so WrapRows returns.

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

@@ -40,11 +40,19 @@ 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: the chcol.Variant the driver scans a Dynamic column (a JSON
+// path such as body_v2.level) into, and the JSONValue a whole JSON column scans into. The attributes
+// bag decodes flat so its dotted keys merge with the legacy attribute maps and a key stored at both a
+// scalar and an object path 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 telemetrystoretypes.JSONValue:
+		if name == "attributes" {
+			return v.Flatten()
+		}
+		return v.Nest()
 	}
 	return val
 }
@@ -54,11 +62,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)
 		}
@@ -478,7 +486,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(stripKeyAlias(colNames[i]), derefValue(cell))
 		}
 		data = append(data, row)
 	}
@@ -536,7 +544,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" {
@@ -575,12 +583,15 @@ 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)
+	// readAsRaw already decoded the JSON column into a flat bag of dotted keys via unwrapVariant.
+	attrJSON, _ := 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)
+		for k, v := range attrJSON {
+			attributes[k] = v
+		}
 		if m, ok := attrStr.(map[string]string); ok {
 			for k, v := range m {
 				attributes[k] = v

pkg/querier/consume_test.go

@@ -3,12 +3,8 @@ 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"
@@ -83,101 +79,40 @@ 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)},
+// jsonColumn builds a JSONValue from dotted leaf paths, as ClickHouse stores them.
+func jsonColumn(paths map[string]any) telemetrystoretypes.JSONValue {
+	j := chcol.NewJSON()
+	for path, value := range paths {
+		j.SetValueAtPath(path, value)
 	}
-
-	// 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"])
-	})
+	return telemetrystoretypes.NewJSONValue(*j)
 }
 
-// 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.
+// A JSON path (e.g. body_v2.level) comes back as a Dynamic column scanned into a chcol.Variant
+// envelope; a whole JSON column scans into a JSONValue, which decodes flat for the attributes bag and
+// nested for every other column.
 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)))
+	jsonCol := jsonColumn(map[string]any{"level": "error", "attrs.code": int64(500)})
+	nullCol := jsonColumn(map[string]any{"level": "error", "absent": chcol.Variant{}})
+
+	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: jsonCol, want: map[string]any{"level": "error", "attrs": map[string]any{"code": float64(500)}}},
+		{name: "AttributesColumnFlat", column: "attributes", input: jsonCol, want: map[string]any{"level": "error", "attrs.code": float64(500)}},
+		{name: "AttributesColumnDropsNullPaths", column: "attributes", input: nullCol, want: map[string]any{"level": "error"}},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, unwrapVariant(testCase.column, testCase.input))
+		})
+	}
 }
 
 func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
@@ -196,8 +131,8 @@ func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
 	}
 }
 
-// Arrays stay native leaves: the collector stringifies top-level arrays and explodes nested ones
-// into indexed keys in the legacy maps; the JSON home keeps them whole and we do not mimic either.
+// readAsRaw decodes the JSON column into a flat bag of dotted keys before this runs; mergeSpanAttributeColumns
+// copies that bag and overlays the legacy string/number/bool maps, which win on collision.
 func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
 	testCases := []struct {
 		name string
@@ -205,12 +140,9 @@ func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
 		want map[string]any
 	}{
 		{
-			name: "JSONOnly_FlattensNestedPaths_PreservesTypes",
+			name: "JSONOnly_FlatBagCopied_PreservesTypes",
 			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 +151,17 @@ 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: "MapOnly_EmptyJSONDoc_KeepsMapValues",
+			name: "MapOnly_EmptyJSONBag_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 +169,39 @@ 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",
+			name: "ScalarAndObjectKey_BothSurvive",
 			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{"db.function": "refresh", "db.function.arg": float64(2)},
 			},
-			want: map[string]any{"key": []any{map[string]any{"a": float64(1)}, map[string]any{"b": float64(2)}}},
+			want: map[string]any{"db.function": "refresh", "db.function.arg": float64(2)},
 		},
 		{
-			name: "NestedArrayOfMaps_StaysNativeLeaf_NoIndexPaths",
-			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{"http": map[string]any{"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",
+			name: "ScalarValueStaysScalar",
 			data: map[string]any{
-				"attributes": telemetrystoretypes.JSONValue{"k": nil},
-			},
-			want: map[string]any{"k": nil},
-		},
-		{
-			name: "KeyIsLeafValue_NotFlattened",
-			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,10 @@ 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
+	// Read JSON columns natively: the driver mis-decodes JSON(max_dynamic_paths=0) columns without
+	// the flattened serialization, and reading them as a string collapses a key stored at both a
+	// scalar and an object path.
+	options.Settings["output_format_native_use_flattened_dynamic_and_json_serialization"] = 1
 
 	chConn, err := clickhouse.Open(options)
 	if err != nil {

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/spantypes/thread.go

@@ -178,9 +178,8 @@ func newThreadSpan(traceID string, storable *StorableSpan) *ThreadSpan {
 	maps.Copy(resources, storable.ResourcesString)
 	timeUnixNano := uint64(storable.StartTime.UnixNano())
 	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)
+	// Spans from before the JSON rollout have an empty document and format to nothing.
+	formatted := aiobservabilitytypes.FormatMessages(storable.AttributesJSON.Nest())
 	return &ThreadSpan{
 		SpanID:            storable.SpanID,
 		TraceID:           traceID,
@@ -206,10 +205,9 @@ func newThreadSpan(traceID string, storable *StorableSpan) *ThreadSpan {
 // threadAttributes flattens the JSON column into dotted keys, 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.Flatten()
+	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,22 @@ package spantypes
 import (
 	"testing"
 
+	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
+	"github.com/SigNoz/signoz/pkg/types/aiobservabilitytypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrystoretypes"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+// jsonAttrs builds a JSONValue from dotted leaf paths, as ClickHouse stores them.
+func jsonAttrs(paths map[string]any) telemetrystoretypes.JSONValue {
+	j := chcol.NewJSON()
+	for path, value := range paths {
+		j.SetValueAtPath(path, value)
+	}
+	return telemetrystoretypes.NewJSONValue(*j)
+}
+
 func TestNewThreadQuery_Cursor(t *testing.T) {
 	testCases := []struct {
 		name    string
@@ -48,7 +59,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"},
@@ -66,6 +77,22 @@ func TestThreadAttributes(t *testing.T) {
 				"gen_ai.stream":             true,
 			},
 		},
+		{
+			// ClickHouse materializes a row's absent paths as null from the part's path schema.
+			name: "JSONColumn_DropsNullPaths",
+			span: StorableSpan{
+				AttributesJSON: jsonAttrs(map[string]any{"gen_ai.request.model": "gpt-4o", "gen_ai.output.messages": chcol.Variant{}}),
+			},
+			wantAttrs: map[string]any{"gen_ai.request.model": "gpt-4o"},
+		},
+		{
+			name: "JSONColumn_OnlyNullPaths_FallsBackToLegacyMaps",
+			span: StorableSpan{
+				AttributesJSON:   jsonAttrs(map[string]any{"gen_ai.output.messages": chcol.Variant{}}),
+				AttributesString: map[string]string{"gen_ai.input.messages": "legacy"},
+			},
+			wantAttrs: map[string]any{"gen_ai.input.messages": "legacy"},
+		},
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
@@ -73,3 +100,15 @@ func TestThreadAttributes(t *testing.T) {
 		})
 	}
 }
+
+func TestNewThreadSpan_FormatsJSONMessages(t *testing.T) {
+	storable := StorableSpan{
+		SpanID:         "f1fa1bc863e94dd0",
+		AttributesJSON: jsonAttrs(map[string]any{"gen_ai.input.messages": `[{"role":"user","content":"hi"}]`}),
+	}
+
+	span := newThreadSpan("trace", &storable)
+
+	require.Len(t, span.FormattedInput, 1)
+	assert.Equal(t, aiobservabilitytypes.RoleUser, span.FormattedInput[0].Role)
+}

pkg/types/spantypes/waterfall_span.go

@@ -324,7 +324,7 @@ func (item *StorableSpan) ToWaterfallSpan(traceID string) *WaterfallSpan {
 	resources := make(map[string]string)
 	maps.Copy(resources, item.ResourcesString)
 
-	formatted := aiobservabilitytypes.FormatMessages(item.AttributesJSON)
+	formatted := aiobservabilitytypes.FormatMessages(item.AttributesJSON.Nest())
 	return &WaterfallSpan{
 		Attributes:         item.Attributes(),
 		DBName:             item.DBName,

pkg/types/telemetrystoretypes/json.go

@@ -1,55 +1,70 @@
 package telemetrystoretypes
 
 import (
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/bytedance/sonic"
+	"encoding/json"
+
+	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
 )
 
-var ErrCodeUnmarshalJSONColumn = errors.MustNewCode("fail_unmarshal_json_column")
+var _ chcol.JSONDeserializer = (*JSONValue)(nil)
 
-// 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
+// JSONValue is the scan target for a ClickHouse native JSON column. It holds the column as the driver's
+// chcol.JSON, so a key stored at both a scalar and an object path (db.function and db.function.arg) stays
+// two paths that nesting would collapse into one. Read it with Flatten for the flat attribute bag or Nest
+// for a nested document; both drop the null paths ClickHouse materializes for a row from its data part's
+// shared path schema.
+type JSONValue struct {
+	column chcol.JSON
+}
 
-// 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)
-	}
+// NewJSONValue wraps a chcol.JSON a caller holds outside the native scan path.
+func NewJSONValue(column chcol.JSON) JSONValue {
+	return JSONValue{column: column}
+}
 
-	decoded := JSONValue{}
-	if err := sonic.Unmarshal(raw, &decoded); err != nil {
-		return errors.WrapInternalf(err, ErrCodeUnmarshalJSONColumn, "failed to unmarshal JSON column")
-	}
-	*v = decoded
+// DeserializeClickHouseJSON lets the native driver scan a JSON column straight into the wrapper.
+func (v *JSONValue) DeserializeClickHouseJSON(column *chcol.JSON) error {
+	v.column = *column
 	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
+// Flatten returns the column as a flat map of dotted path to value, the shape of an attribute bag.
+func (v JSONValue) Flatten() map[string]any {
+	flat := make(map[string]any)
+	for path, value := range v.column.ValuesByPath() {
+		if isNull(value) {
+			continue
 		}
+		flat[path] = value
 	}
+	return unwrap(flat)
+}
+
+// Nest returns the column as a nested document, the shape the logs body is returned to the client in.
+// A key stored at both a scalar and an object path collapses here, which Flatten avoids:
+// https://github.com/SigNoz/events-pod/issues/8
+func (v JSONValue) Nest() map[string]any {
+	return unwrap(v.column.NestedMap())
+}
+
+func isNull(value any) bool {
+	if value == nil {
+		return true
+	}
+	variant, ok := value.(chcol.Variant)
+	return ok && variant.Nil()
+}
+
+// unwrap resolves the driver's Dynamic and Variant wrappers to plain Go types through a JSON round-trip,
+// so callers can type-assert the values. A document that does not round-trip is returned unchanged.
+func unwrap(doc map[string]any) map[string]any {
+	raw, err := json.Marshal(doc)
+	if err != nil {
+		return doc
+	}
+	out := make(map[string]any, len(doc))
+	if err := json.Unmarshal(raw, &out); err != nil {
+		return doc
+	}
+	return out
 }

pkg/types/telemetrystoretypes/json_test.go

@@ -0,0 +1,61 @@
+package telemetrystoretypes
+
+import (
+	"testing"
+
+	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
+	"github.com/stretchr/testify/assert"
+)
+
+func jsonValue(paths map[string]any) JSONValue {
+	j := chcol.NewJSON()
+	for path, value := range paths {
+		j.SetValueAtPath(path, value)
+	}
+	return JSONValue{column: *j}
+}
+
+func TestFlatten(t *testing.T) {
+	testCases := []struct {
+		name  string
+		value JSONValue
+		want  map[string]any
+	}{
+		{
+			name:  "ScalarAndObjectKeyBothSurvive",
+			value: jsonValue(map[string]any{"db.function": "refresh", "db.function.arg": int64(2)}),
+			want:  map[string]any{"db.function": "refresh", "db.function.arg": float64(2)},
+		},
+		{
+			name:  "DropsNullPaths",
+			value: jsonValue(map[string]any{"gen_ai.request.model": "gpt-4o", "gen_ai.output.messages": chcol.Variant{}}),
+			want:  map[string]any{"gen_ai.request.model": "gpt-4o"},
+		},
+		{
+			name:  "Empty",
+			value: jsonValue(map[string]any{}),
+			want:  map[string]any{},
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, testCase.value.Flatten())
+		})
+	}
+}
+
+func TestNest(t *testing.T) {
+	value := jsonValue(map[string]any{
+		"gen_ai.request.model":    "gpt-4o",
+		"gen_ai.prompt.0.content": "hi",
+		"gen_ai.prompt.0.role":    "user",
+		"gen_ai.output.messages":  chcol.Variant{},
+	})
+	want := map[string]any{
+		"gen_ai": map[string]any{
+			"request": map[string]any{"model": "gpt-4o"},
+			"prompt":  map[string]any{"0": map[string]any{"content": "hi", "role": "user"}},
+		},
+	}
+	assert.Equal(t, want, value.Nest())
+}

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