refactor(telemetrystore): move JSONValue to telemetrystoretypes

The scan target is a type consumers name, so it belongs with the store's
other types; WrapRows stays with the provider it wraps rows for.

pkg/querier/consume.go

@@ -11,12 +11,11 @@ import (
 	"strings"
 	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
-	"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/telemetrytypes"
-	"github.com/bytedance/sonic"
 )
 
 var (
@@ -30,8 +29,6 @@ var (
 	// written clickhouse query. The column alias indcate which value is
 	// to be considered as final result (or target).
 	legacyReservedColumnTargetAliases = []string{"__result", "__value", "result", "res", "value"}
-
-	CodeFailUnmarshalJSONColumn = errors.MustNewCode("fail_unmarshal_json_column")
 )
 
 // stripKeyAlias removes the __SELECT_KEY_<n>_ / __GROUP_BY_KEY_<n>_ prefix from a result
@@ -40,6 +37,15 @@ 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 {
+		return v.Any()
+	}
+	return val
+}
+
 // consume reads every row and shapes it into the payload expected for the
 // given request type.
 //
@@ -345,7 +351,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] = derefValue(cell)
+			row[i] = unwrapVariant(derefValue(cell))
 		}
 		data = append(data, row)
 	}
@@ -382,31 +388,13 @@ func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
 	colTypes := rows.ColumnTypes()
 	colCnt := len(colNames)
 
-	// Helper that decides scan target per column based on DB type
-	makeScanTarget := func(i int) any {
-		dbt := strings.ToUpper(colTypes[i].DatabaseTypeName())
-		if strings.HasPrefix(dbt, "JSON") {
-			// Since the driver fails to decode JSON/Dynamic into native Go values, we read it as raw bytes
-			// TODO: check in future if fixed in the driver
-			var v []byte
-			return &v
-		}
-		return reflect.New(colTypes[i].ScanType()).Interface()
-	}
-
-	// Build a template slice of correctly-typed pointers once
-	scanTpl := make([]any, colCnt)
-	for i := range colTypes {
-		scanTpl[i] = makeScanTarget(i)
-	}
-
 	var outRows []*qbtypes.RawRow
 
 	for rows.Next() {
 		// fresh copy of the scan slice (otherwise the driver reuses pointers)
 		scan := make([]any, colCnt)
-		for i := range scanTpl {
-			scan[i] = makeScanTarget(i)
+		for i := range colTypes {
+			scan[i] = reflect.New(colTypes[i].ScanType()).Interface()
 		}
 
 		if err := rows.Scan(scan...); err != nil {
@@ -421,21 +409,7 @@ func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
 			name := stripKeyAlias(colNames[i])
 
 			// de-reference the typed pointer to any
-			val := reflect.ValueOf(cellPtr).Elem().Interface()
-			// Post-process JSON columns: unmarshal bytes into map[string]any
-			if strings.HasPrefix(strings.ToUpper(colTypes[i].DatabaseTypeName()), "JSON") {
-				switch x := val.(type) {
-				case []byte:
-					var m map[string]any
-					err := sonic.Unmarshal(x, &m)
-					if err != nil {
-						return nil, errors.WrapInternalf(err, CodeFailUnmarshalJSONColumn, "failed to unmarshal JSON column %s", name)
-					}
-					val = m
-				default:
-					// already a structured type (map[string]any, []any, etc.)
-				}
-			}
+			val := unwrapVariant(reflect.ValueOf(cellPtr).Elem().Interface())
 
 			// special-case: timestamp column
 			if name == "timestamp" || name == "timestamp_datetime" {

pkg/querier/consume_test.go

@@ -3,8 +3,16 @@ 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"
 )
 
 func TestMergeSpanAttributeColumns_ParsesEventsAndLinks(t *testing.T) {
@@ -75,6 +83,75 @@ 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)},
+	}
+
+	// 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)
+	})
+
+	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"])
+	})
+}
+
+// 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 TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
 	data := map[string]any{
 		"events": []string{},

pkg/querier/postprocess.go

@@ -16,6 +16,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/types/featuretypes"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	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"
 )
@@ -1065,7 +1066,7 @@ func (q *querier) postProcessLogBody(ctx context.Context, orgID valuer.UUID, res
 		return result
 	}
 	for _, row := range rawData.Rows {
-		bodyMap, ok := row.Data["body"].(map[string]any)
+		bodyMap, ok := row.Data["body"].(telemetrystoretypes.JSONValue)
 		if !ok {
 			continue
 		}

pkg/telemetrystore/clickhousetelemetrystore/provider.go

@@ -184,7 +184,7 @@ func (p *provider) Query(ctx context.Context, query string, args ...interface{})
 	}
 
 	return &rowsWithHooks{
-		Rows:    rows,
+		Rows:    telemetrystore.WrapRows(rows),
 		ctx:     ctx,
 		event:   event,
 		onClose: func() { telemetrystore.WrapAfterQuery(p.hooks, ctx, event) },

pkg/telemetrystore/rows.go

@@ -0,0 +1,39 @@
+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/provider.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"github.com/DATA-DOG/go-sqlmock"
 	cmock "github.com/SigNoz/clickhouse-go-mock"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
@@ -32,7 +33,21 @@ func New(_ telemetrystore.Config, matcher sqlmock.QueryMatcher) *Provider {
 
 // ClickhouseDB returns the mock Clickhouse connection.
 func (p *Provider) ClickhouseDB() clickhouse.Conn {
-	return p.clickhouseDB.(clickhouse.Conn)
+	return conn{Conn: p.clickhouseDB.(clickhouse.Conn)}
+}
+
+// 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
 }
 
 // Cluster returns the cluster name.

pkg/types/telemetrystoretypes/json.go

@@ -0,0 +1,37 @@
+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
+}