fix(querier): decode JSON columns for scalar and time-series results

Share the raw reader's JSON scan target across all three readers, and unwrap
Dynamic columns from their chcol.Variant envelope.

pkg/querier/consume.go

@@ -11,6 +11,7 @@ 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"
@@ -40,6 +41,45 @@ func stripKeyAlias(name string) string {
 	return keyAliasRe.ReplaceAllString(name, "")
 }
 
+func isJSONColumn(colType driver.ColumnType) bool {
+	return strings.HasPrefix(strings.ToUpper(colType.DatabaseTypeName()), "JSON")
+}
+
+// scanTarget returns the scan destination for a column.
+//
+// A JSON column is read as raw bytes: its ScanType is chcol.JSON, whose Scan accepts only a
+// JSON or a map[string]any and rejects the string the server sends us (we ask for JSON as
+// strings via output_format_native_write_json_as_string). decodeColumnValue turns the bytes
+// back into a map. Dynamic/Variant columns scan natively into a chcol.Variant envelope.
+func scanTarget(colType driver.ColumnType) any {
+	if isJSONColumn(colType) {
+		var v []byte
+		return &v
+	}
+	return reflect.New(colType.ScanType()).Interface()
+}
+
+// decodeColumnValue turns a scanned value into the value carried in the payload: JSON bytes
+// become a map, and a Dynamic/Variant envelope is unwrapped to the concrete value it holds.
+func decodeColumnValue(colType driver.ColumnType, name string, val any) (any, error) {
+	if isJSONColumn(colType) {
+		b, ok := val.([]byte)
+		if !ok {
+			// already a structured type (map[string]any, []any, etc.)
+			return val, nil
+		}
+		var m map[string]any
+		if err := sonic.Unmarshal(b, &m); err != nil {
+			return nil, errors.WrapInternalf(err, CodeFailUnmarshalJSONColumn, "failed to unmarshal JSON column %s", name)
+		}
+		return m, nil
+	}
+	if v, ok := val.(chcol.Variant); ok {
+		return v.Any(), nil
+	}
+	return val, nil
+}
+
 // consume reads every row and shapes it into the payload expected for the
 // given request type.
 //
@@ -73,7 +113,7 @@ func readAsTimeSeries(rows driver.Rows, queryWindow *qbtypes.TimeRange, step qbt
 	slots := make([]any, len(colTypes))
 	numericColsCount := 0
 	for i, ct := range colTypes {
-		slots[i] = reflect.New(ct.ScanType()).Interface()
+		slots[i] = scanTarget(ct)
 		if isNumericKind(ct.ScanType()) {
 			numericColsCount++
 		}
@@ -332,7 +372,7 @@ func readAsScalar(rows driver.Rows, queryName string) (*qbtypes.ScalarData, erro
 	// Pre-allocate scan slots once
 	scan := make([]any, len(colTypes))
 	for i := range scan {
-		scan[i] = reflect.New(colTypes[i].ScanType()).Interface()
+		scan[i] = scanTarget(colTypes[i])
 	}
 
 	var data [][]any
@@ -345,7 +385,11 @@ 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)
+			val, err := decodeColumnValue(colTypes[i], cd[i].Name, derefValue(cell))
+			if err != nil {
+				return nil, err
+			}
+			row[i] = val
 		}
 		data = append(data, row)
 	}
@@ -382,31 +426,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] = scanTarget(colTypes[i])
 		}
 
 		if err := rows.Scan(scan...); err != nil {
@@ -421,20 +447,9 @@ 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, err := decodeColumnValue(colTypes[i], name, reflect.ValueOf(cellPtr).Elem().Interface())
+			if err != nil {
+				return nil, err
 			}
 
 			// special-case: timestamp column

pkg/querier/consume_test.go

@@ -3,8 +3,14 @@ package querier
 import (
 	"reflect"
 	"testing"
+	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
+	cmock "github.com/SigNoz/clickhouse-go-mock"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/spantypes"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestMergeSpanAttributeColumns_ParsesEventsAndLinks(t *testing.T) {
@@ -75,6 +81,78 @@ 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 := map[string]any{
+		"level": "error",
+		"attrs": map[string]any{"code": float64(500)},
+	}
+
+	t.Run("scalar", func(t *testing.T) {
+		rows := cmock.NewRows([]cmock.ColumnType{
+			{Name: "body_v2", Type: "JSON"},
+			{Name: "__result_0", Type: "UInt64"},
+		}, [][]any{{body, uint64(3)}})
+
+		payload, err := consume(rows, qbtypes.RequestTypeScalar, nil, qbtypes.Step{}, "A")
+		require.NoError(t, err)
+
+		data := payload.(*qbtypes.ScalarData)
+		require.Len(t, data.Data, 1)
+		assert.Equal(t, wantBody, data.Data[0][0])
+		assert.Equal(t, uint64(3), data.Data[0][1])
+	})
+
+	t.Run("time series", func(t *testing.T) {
+		rows := 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 := 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 TestDecodeColumnValue_Dynamic(t *testing.T) {
+	colType := cmock.NewColumnType("level", "Dynamic", false, reflect.TypeOf(chcol.Variant{}))
+
+	val, err := decodeColumnValue(colType, "level", chcol.NewDynamicWithType("error", "String"))
+	require.NoError(t, err)
+	assert.Equal(t, "error", val)
+
+	val, err = decodeColumnValue(colType, "level", chcol.Dynamic{})
+	require.NoError(t, err)
+	assert.Nil(t, val)
+}
+
 func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
 	data := map[string]any{
 		"events": []string{},