perf(querier): classify time-series columns once per query

The time-series reader decided what each cell was by switching on its pointer
type, per cell per row — and the switch listed 32- and 64-bit widths only, so
a raw ClickHouse query aggregating or grouping by an Int8, UInt8, Int16,
UInt16 or Bool column (max(severity_number), GROUP BY has_error) either
emptied the chart or merged every group into one unlabelled series.

Classify each column once, up front, into what it contributes to every row:
timestamp, numeric, bool, string, or document. Numeric coverage now comes
from the same helper that counts numeric columns, so the two cannot disagree,
and any width works. Rows reuse the scratch that used to be allocated per row
— the aggregation map, the label slices, and the label objects for rows that
land in existing series — and values dodge the reflection boxing on the way
out of the driver. On a 200k-series group-by this is a quarter fewer bytes
allocated and a fifth less CPU per request.

The __result_<n> index is now also bounded before it sizes the aggregation
slots: the alias is user-written in raw SQL, and an overflowing or huge index
panicked the reader — on main too, where the bucket slice is sized from the
same number after the rows are read.

pkg/querier/consume.go

@@ -1,22 +1,22 @@
 package querier
 
 import (
+	"encoding/json"
 	"fmt"
 	"math"
 	"reflect"
 	"regexp"
 	"slices"
-	"sort"
 	"strconv"
 	"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/telemetrystoretypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-	"github.com/bytedance/sonic"
 )
 
 var (
@@ -30,8 +30,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 +38,32 @@ 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
+}
+
+// labelValue renders a group-by value the payload cannot carry as a scalar — a JSON column, or a
+// Dynamic one — as a stable string, so that rows differing only in that value land in different
+// 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)
+	if val == nil {
+		return ""
+	}
+	if v, ok := val.(telemetrystoretypes.JSONValue); ok {
+		if raw, err := json.Marshal(v); err == nil {
+			return string(raw)
+		}
+	}
+	return fmt.Sprint(val)
+}
+
 // consume reads every row and shapes it into the payload expected for the
 // given request type.
 //
@@ -66,10 +90,148 @@ func consume(rows driver.Rows, kind qbtypes.RequestType, queryWindow *qbtypes.Ti
 	return payload, err
 }
 
+// labelPair is a label held in per-row scratch, so that rows landing in an existing series do not
+// allocate label objects only to drop them. It keeps the value unboxed — an any field would put
+// every label of every row on the heap — and boxes once, for the row that creates the series.
+type labelPair struct {
+	name    string
+	display string // the form the series key is built from
+	num     float64
+	class   tsColumnClass
+}
+
+func materialiseLabels(pairs []labelPair) []*qbtypes.Label {
+	labels := make([]*qbtypes.Label, len(pairs))
+	for i, pair := range pairs {
+		var value any = pair.display
+		switch pair.class {
+		case tsColumnNumeric:
+			value = pair.num
+		case tsColumnBool:
+			value = pair.num != 0
+		}
+		labels[i] = &qbtypes.Label{
+			Key:   telemetrytypes.TelemetryFieldKey{Name: pair.name},
+			Value: value,
+		}
+	}
+	return labels
+}
+
+// The *FromSlot helpers skip the reflection in derefValue for the types that actually turn up;
+// anything else still goes through it, so a missing width is slower, never dropped.
+func numericFromSlot(ptr any) (float64, bool) {
+	switch v := ptr.(type) {
+	case *float64:
+		return *v, true
+	case *uint64:
+		return float64(*v), true
+	case *int64:
+		return float64(*v), true
+	case *uint32:
+		return float64(*v), true
+	case *int32:
+		return float64(*v), true
+	case *float32:
+		return float64(*v), true
+	case *uint8:
+		return float64(*v), true
+	case *int8:
+		return float64(*v), true
+	}
+	val := derefValue(ptr)
+	if val == nil {
+		return 0, false
+	}
+	return numericAsFloat(val), true
+}
+
+func boolFromSlot(ptr any) bool {
+	if flag, ok := ptr.(*bool); ok {
+		return *flag
+	}
+	flag, _ := derefValue(ptr).(bool)
+	return flag
+}
+
+func stringFromSlot(ptr any) string {
+	switch v := ptr.(type) {
+	case *string:
+		return *v
+	case **string:
+		if *v == nil {
+			return ""
+		}
+		return **v
+	}
+	str, _ := derefValue(ptr).(string)
+	return str
+}
+
+// maxAggregationIndex bounds the __result_<n> indices honored as aggregations; anything past it
+// is read as a plain numeric column.
+const maxAggregationIndex = 1000
+
+type tsColumnClass uint8
+
+const (
+	tsColumnSkip tsColumnClass = iota
+	tsColumnTimestamp
+	tsColumnNumeric
+	tsColumnBool
+	tsColumnString
+	tsColumnDocument // a JSON or Dynamic column, rendered as a label
+)
+
+// tsColumn is what a result column contributes to every row of a time series. The class and the
+// role are the same for all of them, so they are worked out once instead of per cell.
+type tsColumn struct {
+	name          string
+	class         tsColumnClass
+	aggIdx        int // -1 unless the column is aliased as an aggregation
+	isTargetAlias bool
+}
+
+func planColumns(colNames []string, colTypes []driver.ColumnType) []tsColumn {
+	plan := make([]tsColumn, len(colTypes))
+	for i, colType := range colTypes {
+		name := stripKeyAlias(colNames[i])
+		col := tsColumn{
+			name:          name,
+			aggIdx:        -1,
+			isTargetAlias: slices.Contains(legacyReservedColumnTargetAliases, name),
+		}
+		// A raw ClickHouse query writes its own aliases, and aggValues and the result buckets are
+		// sized from this index — an unchecked __result_<n> is an allocation of the user's choosing.
+		if m := aggRe.FindStringSubmatch(name); m != nil {
+			if idx, err := strconv.Atoi(m[1]); err == nil && idx < maxAggregationIndex {
+				col.aggIdx = idx
+			}
+		}
+
+		typ := baseType(colType.ScanType())
+		switch {
+		case typ == timeType:
+			col.class = tsColumnTimestamp
+		case typ == jsonValueType, typ == variantType:
+			col.class = tsColumnDocument
+		case isNumericKind(typ):
+			col.class = tsColumnNumeric
+		case typ.Kind() == reflect.Bool:
+			col.class = tsColumnBool
+		case typ.Kind() == reflect.String:
+			col.class = tsColumnString
+		}
+		plan[i] = col
+	}
+	return plan
+}
+
 func readAsTimeSeries(rows driver.Rows, queryWindow *qbtypes.TimeRange, step qbtypes.Step, queryName string) (*qbtypes.TimeSeriesData, error) {
 	colTypes := rows.ColumnTypes()
 	colNames := rows.Columns()
 
+	plan := planColumns(colNames, colTypes)
 	slots := make([]any, len(colTypes))
 	numericColsCount := 0
 	for i, ct := range colTypes {
@@ -122,109 +284,122 @@ func readAsTimeSeries(rows driver.Rows, queryWindow *qbtypes.TimeRange, step qbt
 		lblValsCapacity = 0
 	}
 
+	// Every row writes into the same scratch: the label objects are materialised only for the row
+	// that creates a series, and aggregation slots are indexed by the alias instead of hashed.
+	aggCount := 1
+	for _, col := range plan {
+		if col.aggIdx >= aggCount {
+			aggCount = col.aggIdx + 1
+		}
+	}
+	var (
+		aggValues = make([]float64, aggCount)
+		aggSeen   = make([]bool, aggCount)
+		lblVals   = make([]string, 0, lblValsCapacity)
+		lblPairs  = make([]labelPair, 0, lblValsCapacity)
+	)
+
 	for rows.Next() {
 		if err := rows.Scan(slots...); err != nil {
 			return nil, err
 		}
 
+		clear(aggSeen)
+		lblVals = lblVals[:0]
+		lblPairs = lblPairs[:0]
+
 		var (
 			ts            int64
-			lblVals       = make([]string, 0, lblValsCapacity)
-			lblObjs       = make([]*qbtypes.Label, 0, lblValsCapacity)
-			aggValues     = map[int]float64{} // all __result_N in this row
-			fallbackValue float64             // value when NO __result_N columns exist
+			anyAgg        bool
+			fallbackValue float64 // value when NO __result_N columns exist
 			fallbackSeen  bool
 		)
 
 		for idx, ptr := range slots {
-			name := stripKeyAlias(colNames[idx])
+			col := plan[idx]
 
-			switch v := ptr.(type) {
-			case *time.Time:
-				ts = v.UnixMilli()
+			switch col.class {
+			case tsColumnTimestamp:
+				if t, ok := ptr.(*time.Time); ok {
+					ts = t.UnixMilli()
+				} else if t, ok := derefValue(ptr).(time.Time); ok {
+					ts = t.UnixMilli()
+				}
 
-			case *float64, *float32, *int64, *int32, *uint64, *uint32:
-				val := numericAsFloat(reflect.ValueOf(ptr).Elem().Interface())
-				if m := aggRe.FindStringSubmatch(name); m != nil {
-					id, _ := strconv.Atoi(m[1])
-					aggValues[id] = val
-				} else if numericColsCount == 1 { // classic single-value query
-					fallbackValue = val
+			case tsColumnNumeric:
+				num, ok := numericFromSlot(ptr)
+				if !ok { // a NULL number is neither a value nor a label
+					continue
+				}
+				switch {
+				case col.aggIdx >= 0:
+					aggValues[col.aggIdx] = num
+					aggSeen[col.aggIdx] = true
+					anyAgg = true
+				case numericColsCount == 1, col.isTargetAlias: // classic single-value query
+					fallbackValue = num
 					fallbackSeen = true
-				} else if slices.Contains(legacyReservedColumnTargetAliases, name) {
-					fallbackValue = val
-					fallbackSeen = true
-				} else {
-					// numeric label
-					lblVals = append(lblVals, fmt.Sprint(val))
-					lblObjs = append(lblObjs, &qbtypes.Label{
-						Key:   telemetrytypes.TelemetryFieldKey{Name: name},
-						Value: val,
+				default:
+					display := strconv.FormatFloat(num, 'g', -1, 64)
+					lblVals = append(lblVals, display)
+					lblPairs = append(lblPairs, labelPair{
+						name:    col.name,
+						display: display,
+						num:     num,
+						class:   tsColumnNumeric,
 					})
 				}
 
-			case **float64, **float32, **int64, **int32, **uint64, **uint32:
-				tempVal := reflect.ValueOf(ptr)
-				if tempVal.IsValid() && !tempVal.IsNil() && !tempVal.Elem().IsNil() {
-					val := numericAsFloat(tempVal.Elem().Elem().Interface())
-					if m := aggRe.FindStringSubmatch(name); m != nil {
-						id, _ := strconv.Atoi(m[1])
-						aggValues[id] = val
-					} else if numericColsCount == 1 { // classic single-value query
-						fallbackValue = val
-						fallbackSeen = true
-					} else if slices.Contains(legacyReservedColumnTargetAliases, name) {
-						fallbackValue = val
-						fallbackSeen = true
-					} else {
-						// numeric label
-						lblVals = append(lblVals, fmt.Sprint(val))
-						lblObjs = append(lblObjs, &qbtypes.Label{
-							Key:   telemetrytypes.TelemetryFieldKey{Name: name},
-							Value: val,
-						})
-					}
+			case tsColumnBool:
+				flag := boolFromSlot(ptr)
+				switch {
+				case col.aggIdx >= 0:
+					aggValues[col.aggIdx] = boolAsFloat(flag)
+					aggSeen[col.aggIdx] = true
+					anyAgg = true
+				case col.isTargetAlias:
+					fallbackValue = boolAsFloat(flag)
+					fallbackSeen = true
+				default:
+					display := strconv.FormatBool(flag)
+					lblVals = append(lblVals, display)
+					lblPairs = append(lblPairs, labelPair{
+						name:    col.name,
+						display: display,
+						num:     boolAsFloat(flag),
+						class:   tsColumnBool,
+					})
 				}
 
-			case *string:
-				lblVals = append(lblVals, *v)
-				lblObjs = append(lblObjs, &qbtypes.Label{
-					Key:   telemetrytypes.TelemetryFieldKey{Name: name},
-					Value: *v,
-				})
+			case tsColumnString:
+				label := stringFromSlot(ptr) // a NULL string labels the series with the empty string
+				lblVals = append(lblVals, label)
+				lblPairs = append(lblPairs, labelPair{name: col.name, display: label, class: tsColumnString})
 
-			case **string:
-				val := *v
-				if val == nil {
-					var empty string
-					val = &empty
-				}
-				lblVals = append(lblVals, *val)
-				lblObjs = append(lblObjs, &qbtypes.Label{
-					Key:   telemetrytypes.TelemetryFieldKey{Name: name},
-					Value: *val,
-				})
-
-			default:
-				continue
+			case tsColumnDocument:
+				label := labelValue(derefValue(ptr))
+				lblVals = append(lblVals, label)
+				lblPairs = append(lblPairs, labelPair{name: col.name, display: label, class: tsColumnDocument})
 			}
 		}
 
 		// Edge-case: no __result_N columns, but a single numeric column present
-		if len(aggValues) == 0 && fallbackSeen {
+		if !anyAgg && fallbackSeen {
 			aggValues[0] = fallbackValue
+			aggSeen[0] = true
+			anyAgg = true
 		}
 
-		if ts == 0 || len(aggValues) == 0 {
+		if ts == 0 || !anyAgg {
 			continue // nothing useful
 		}
 
-		sort.Strings(lblVals)
+		slices.Sort(lblVals)
 		labelsKey := strings.Join(lblVals, ",")
 
 		// one point per aggregation in this row
 		for aggIdx, val := range aggValues {
-			if math.IsNaN(val) || math.IsInf(val, 0) {
+			if !aggSeen[aggIdx] || math.IsNaN(val) || math.IsInf(val, 0) {
 				continue
 			}
 
@@ -232,7 +407,7 @@ func readAsTimeSeries(rows driver.Rows, queryWindow *qbtypes.TimeRange, step qbt
 
 			series, ok := seriesMap[key]
 			if !ok {
-				series = &qbtypes.TimeSeries{Labels: lblObjs}
+				series = &qbtypes.TimeSeries{Labels: materialiseLabels(lblPairs)}
 				seriesMap[key] = series
 			}
 			series.Values = append(series.Values, &qbtypes.TimeSeriesValue{
@@ -282,6 +457,20 @@ func readAsTimeSeries(rows driver.Rows, queryWindow *qbtypes.TimeRange, step qbt
 	}, nil
 }
 
+var (
+	timeType      = reflect.TypeFor[time.Time]()
+	jsonValueType = reflect.TypeFor[telemetrystoretypes.JSONValue]()
+	variantType   = reflect.TypeFor[chcol.Variant]()
+)
+
+// baseType unwraps pointer levels, so that a Nullable column is classified like the column it wraps.
+func baseType(t reflect.Type) reflect.Type {
+	for t.Kind() == reflect.Pointer {
+		t = t.Elem()
+	}
+	return t
+}
+
 func isNumericKind(t reflect.Type) bool {
 	if t == nil {
 		return false
@@ -345,7 +534,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 +571,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 +592,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" {
@@ -512,6 +669,13 @@ func mergeSpanAttributeColumns(data map[string]any) {
 	}
 }
 
+func boolAsFloat(v bool) float64 {
+	if v {
+		return 1
+	}
+	return 0
+}
+
 // numericAsFloat converts numeric types to float64 efficiently.
 func numericAsFloat(v any) float64 {
 	switch x := v.(type) {

pkg/querier/consume_test.go

@@ -3,8 +3,17 @@ package querier
 import (
 	"reflect"
 	"testing"
+	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
+	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 +84,103 @@ 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)
+	})
+
+	// 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"])
+	})
+}
+
+// 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{},
@@ -90,3 +196,114 @@ func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
 		t.Fatalf("expected empty []spantypes.Link, got %#v", data["links"])
 	}
 }
+
+// boolRows reports a bool scan type for one column, which cmock cannot do on its own.
+type boolRows struct {
+	driver.Rows
+	col string
+}
+
+func (r boolRows) ColumnTypes() []driver.ColumnType {
+	out := r.Rows.ColumnTypes()
+	for i, colType := range out {
+		if colType.Name() == r.col {
+			out[i] = cmock.NewColumnType(colType.Name(), "Bool", false, reflect.TypeOf(true))
+		}
+	}
+	return out
+}
+
+// A time-series result can carry numeric widths narrower than 32 bits — max(severity_number) is
+// UInt8, kind is Int8, has_error is Bool. Dropping them empties the chart when the column holds the
+// aggregation, and merges every group into one series when it is a group-by key.
+func TestConsume_NarrowNumericWidths(t *testing.T) {
+	ts := time.Date(2026, 8, 19, 10, 0, 0, 0, time.UTC)
+
+	seriesOf := func(t *testing.T, rows driver.Rows) []*qbtypes.TimeSeries {
+		t.Helper()
+		payload, err := consume(rows, qbtypes.RequestTypeTimeSeries, nil, qbtypes.Step{}, "A")
+		require.NoError(t, err)
+		data := payload.(*qbtypes.TimeSeriesData)
+		require.Len(t, data.Aggregations, 1)
+		return data.Aggregations[0].Series
+	}
+
+	t.Run("UInt8 aggregation", func(t *testing.T) {
+		series := seriesOf(t, cmock.NewRows([]cmock.ColumnType{
+			{Name: "ts", Type: "DateTime"},
+			{Name: "__result_0", Type: "UInt8"},
+		}, [][]any{{ts, uint8(17)}}))
+
+		require.Len(t, series, 1)
+		require.Len(t, series[0].Values, 1)
+		assert.Equal(t, float64(17), series[0].Values[0].Value)
+	})
+
+	t.Run("Int8 group-by", func(t *testing.T) {
+		series := seriesOf(t, cmock.NewRows([]cmock.ColumnType{
+			{Name: "ts", Type: "DateTime"},
+			{Name: "kind", Type: "Int8"},
+			{Name: "__result_0", Type: "UInt64"},
+		}, [][]any{{ts, int8(2), uint64(7)}, {ts, int8(3), uint64(2)}}))
+
+		got := map[float64]float64{}
+		for _, s := range series {
+			require.Len(t, s.Labels, 1)
+			require.Len(t, s.Values, 1)
+			got[s.Labels[0].Value.(float64)] = s.Values[0].Value
+		}
+		assert.Equal(t, map[float64]float64{2: 7, 3: 2}, got)
+	})
+
+	t.Run("Bool aggregation", func(t *testing.T) {
+		series := seriesOf(t, boolRows{col: "__result_0", Rows: cmock.NewRows([]cmock.ColumnType{
+			{Name: "ts", Type: "DateTime"},
+			{Name: "__result_0", Type: "UInt8"},
+		}, [][]any{{ts, uint8(1)}})})
+
+		require.Len(t, series, 1)
+		require.Len(t, series[0].Values, 1)
+		assert.Equal(t, float64(1), series[0].Values[0].Value)
+	})
+
+	t.Run("Bool group-by", func(t *testing.T) {
+		series := seriesOf(t, boolRows{col: "has_error", Rows: cmock.NewRows([]cmock.ColumnType{
+			{Name: "ts", Type: "DateTime"},
+			{Name: "has_error", Type: "UInt8"},
+			{Name: "__result_0", Type: "UInt64"},
+		}, [][]any{{ts, uint8(1), uint64(7)}, {ts, uint8(0), uint64(2)}})})
+
+		got := map[bool]float64{}
+		for _, s := range series {
+			require.Len(t, s.Labels, 1)
+			require.Len(t, s.Values, 1)
+			got[s.Labels[0].Value.(bool)] = s.Values[0].Value
+		}
+		assert.Equal(t, map[bool]float64{true: 7, false: 2}, got)
+	})
+}
+
+// A raw ClickHouse query picks its own aliases, so __result_<n> can carry any index — including
+// one that overflows int or would size a slice in the terabytes.
+func TestConsume_HugeAggregationAlias(t *testing.T) {
+	ts := time.Date(2026, 8, 19, 10, 0, 0, 0, time.UTC)
+
+	for _, alias := range []string{"__result_99999999999999999999", "__result_4000000000"} {
+		t.Run(alias, func(t *testing.T) {
+			rows := cmock.NewRows([]cmock.ColumnType{
+				{Name: "ts", Type: "DateTime"},
+				{Name: alias, Type: "UInt64"},
+			}, [][]any{{ts, uint64(7)}})
+
+			payload, err := consume(rows, qbtypes.RequestTypeTimeSeries, nil, qbtypes.Step{}, "A")
+			require.NoError(t, err)
+
+			// the alias is not honored as an aggregation; the single numeric column carries the value
+			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(7), data.Aggregations[0].Series[0].Values[0].Value)
+		})
+	}
+}

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