mirror of
https://github.com/SigNoz/signoz.git
synced 2026-09-29 14:50:41 +01:00
Compare commits
8 Commits
feat/ai-tr
...
ns/read-at
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2177206b35 | ||
|
|
ee1aacc32b | ||
|
|
7cd0135fa4 | ||
|
|
33db5c467c | ||
|
|
dcdf63f823 | ||
|
|
aecbb593bd | ||
|
|
6a4e412a47 | ||
|
|
90be4a498d |
@@ -40,11 +40,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.
|
||||
// unwrapVariant returns the concrete value inside the driver's scan envelopes: chcol.Variant for a
|
||||
// Dynamic column (a JSON path such as body_v2.level), and chcol.JSON for a whole JSON column, decoded
|
||||
// into a nested document.
|
||||
func unwrapVariant(val any) any {
|
||||
if v, ok := val.(chcol.Variant); ok {
|
||||
switch v := val.(type) {
|
||||
case chcol.Variant:
|
||||
return v.Any()
|
||||
case chcol.JSON:
|
||||
return telemetrystoretypes.NestedJSON(v)
|
||||
}
|
||||
return val
|
||||
}
|
||||
@@ -58,7 +62,7 @@ func labelValue(val any) string {
|
||||
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 +208,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{
|
||||
@@ -536,7 +540,14 @@ 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())
|
||||
raw := reflect.ValueOf(cellPtr).Elem().Interface()
|
||||
// the attributes bag is flattened to dotted keys downstream; decode it flat so a key stored as both a scalar and an object is not collapsed into a mislabeled key.
|
||||
var val any
|
||||
if j, ok := raw.(chcol.JSON); ok && name == "attributes" {
|
||||
val = telemetrystoretypes.FlattenJSON(j)
|
||||
} else {
|
||||
val = unwrapVariant(raw)
|
||||
}
|
||||
|
||||
// special-case: timestamp column
|
||||
if name == "timestamp" || name == "timestamp_datetime" {
|
||||
@@ -576,8 +587,6 @@ func flattenJSONPaths(prefix string, m map[string]any, out map[string]any) {
|
||||
switch child := v.(type) {
|
||||
case map[string]any:
|
||||
flattenJSONPaths(key, child, out)
|
||||
case telemetrystoretypes.JSONValue:
|
||||
flattenJSONPaths(key, child, out)
|
||||
default:
|
||||
out[key] = v
|
||||
}
|
||||
@@ -593,7 +602,7 @@ 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)
|
||||
attrJSON, _ := data["attributes"].(map[string]any)
|
||||
// todo(nitya): move to resource json
|
||||
resStr, hasRes := data["resources_string"]
|
||||
if hasStr || hasNum || hasBool || attrJSON != nil || hasRes {
|
||||
|
||||
@@ -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,31 @@ 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.
|
||||
// A JSON path (e.g. `body_v2.level`) comes back as a Dynamic column, which the driver scans into a
|
||||
// chcol.Variant envelope; a whole JSON column comes back as chcol.JSON, decoded into a nested document.
|
||||
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)))
|
||||
|
||||
j := chcol.NewJSON()
|
||||
j.SetValueAtPath("level", "error")
|
||||
j.SetValueAtPath("attrs.code", int64(500))
|
||||
assert.Equal(t, map[string]any{
|
||||
"level": "error",
|
||||
"attrs": map[string]any{"code": float64(500)},
|
||||
}, unwrapVariant(*j))
|
||||
}
|
||||
|
||||
// labelValue renders a JSON group-by value as a stable, sorted-key string so structurally equal
|
||||
// documents share a series.
|
||||
func TestLabelValue(t *testing.T) {
|
||||
assert.Equal(t, "", labelValue(nil))
|
||||
assert.Equal(t, "error", labelValue(chcol.NewDynamicWithType("error", "String")))
|
||||
assert.Equal(t, `{"attrs":{"code":500},"level":"error"}`, labelValue(map[string]any{
|
||||
"level": "error",
|
||||
"attrs": map[string]any{"code": 500},
|
||||
}))
|
||||
}
|
||||
|
||||
func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
|
||||
@@ -207,7 +132,7 @@ func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
|
||||
{
|
||||
name: "JSONOnly_FlattensNestedPaths_PreservesTypes",
|
||||
data: map[string]any{
|
||||
"attributes": telemetrystoretypes.JSONValue{
|
||||
"attributes": map[string]any{
|
||||
"http": map[string]any{"route": "/api/pay", "retry": map[string]any{"count": float64(3)}},
|
||||
"cache.hit": true,
|
||||
},
|
||||
@@ -219,7 +144,7 @@ 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": map[string]any{"route": "/new"}, "only.json": "j"},
|
||||
},
|
||||
want: map[string]any{"http.route": "/old", "only.map": "m", "http.status": float64(500), "only.json": "j"},
|
||||
},
|
||||
@@ -229,7 +154,7 @@ func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
|
||||
"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,28 +162,28 @@ 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": map[string]any{"tags": []any{"a", "b"}, "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",
|
||||
data: map[string]any{
|
||||
"attributes": telemetrystoretypes.JSONValue{"http": map[string]any{"items": []any{map[string]any{"a": float64(1)}}}},
|
||||
"attributes": map[string]any{"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)}}},
|
||||
},
|
||||
@@ -266,28 +191,28 @@ func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
|
||||
name: "DualWritten_NestedArray_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": map[string]any{"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",
|
||||
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"}},
|
||||
"attributes": map[string]any{"http": map[string]any{"route": "/a"}},
|
||||
},
|
||||
want: map[string]any{"http.route": "/a"},
|
||||
},
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
// Decode JSON columns via the flattened native serialization (CH 25.6+); without it clickhouse-go mis-decodes the SharedData layout of JSON(max_dynamic_paths=0) columns and desyncs the native protocol.
|
||||
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) },
|
||||
|
||||
@@ -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]()
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -1,37 +1,51 @@
|
||||
package telemetrystoretypes
|
||||
|
||||
import (
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
|
||||
"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
|
||||
// NestedJSON decodes a native JSON column into a nested document via the driver's own marshaler, so
|
||||
// arrays of objects and typed sub-paths survive and Dynamic values arrive unwrapped. A key stored as
|
||||
// both a scalar and an object collapses, as the nested form cannot hold both.
|
||||
func NestedJSON(j chcol.JSON) map[string]any {
|
||||
raw, err := j.MarshalJSON()
|
||||
if err != 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")
|
||||
var out map[string]any
|
||||
if err := sonic.Unmarshal(raw, &out); err != nil {
|
||||
return nil
|
||||
}
|
||||
*v = decoded
|
||||
return nil
|
||||
return out
|
||||
}
|
||||
|
||||
// FlattenJSON decodes a native JSON column into its leaf paths as dotted keys, so a key stored as
|
||||
// both a scalar and an object survives as two distinct keys — unlike the nested form, which cannot
|
||||
// hold both. Dynamic values arrive unwrapped, arrays of objects intact.
|
||||
func FlattenJSON(j chcol.JSON) map[string]any {
|
||||
paths := j.ValuesByPath()
|
||||
out := make(map[string]any, len(paths))
|
||||
for path, value := range paths {
|
||||
out[path] = decodePathValue(value)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func decodePathValue(value any) any {
|
||||
variant, ok := value.(chcol.Variant)
|
||||
if !ok {
|
||||
return value
|
||||
}
|
||||
raw, err := json.Marshal(variant)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
var out any
|
||||
if err := sonic.Unmarshal(raw, &out); err != nil {
|
||||
return nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
82
pkg/types/telemetrystoretypes/json_test.go
Normal file
82
pkg/types/telemetrystoretypes/json_test.go
Normal file
@@ -0,0 +1,82 @@
|
||||
package telemetrystoretypes
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestNestedJSON(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
paths map[string]any
|
||||
want map[string]any
|
||||
}{
|
||||
{
|
||||
name: "Empty",
|
||||
paths: nil,
|
||||
want: map[string]any{},
|
||||
},
|
||||
{
|
||||
name: "FlatScalars",
|
||||
paths: map[string]any{"level": "error", "status": int64(500)},
|
||||
want: map[string]any{"level": "error", "status": float64(500)},
|
||||
},
|
||||
{
|
||||
name: "DottedPathsBecomeNested",
|
||||
paths: map[string]any{"attrs.code": int64(500), "attrs.path": "/checkout"},
|
||||
want: map[string]any{"attrs": map[string]any{"code": float64(500), "path": "/checkout"}},
|
||||
},
|
||||
{
|
||||
name: "ArrayOfObjectsPreserved",
|
||||
paths: map[string]any{"education": []any{map[string]any{"name": "IIT"}}},
|
||||
want: map[string]any{"education": []any{map[string]any{"name": "IIT"}}},
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
j := chcol.NewJSON()
|
||||
for path, value := range testCase.paths {
|
||||
j.SetValueAtPath(path, value)
|
||||
}
|
||||
assert.Equal(t, testCase.want, NestedJSON(*j))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFlattenJSON(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
paths map[string]any
|
||||
want map[string]any
|
||||
}{
|
||||
{
|
||||
name: "Empty",
|
||||
paths: nil,
|
||||
want: map[string]any{},
|
||||
},
|
||||
{
|
||||
name: "DottedPathsStayFlat",
|
||||
paths: map[string]any{"http.method": "GET", "level": "error"},
|
||||
want: map[string]any{"http.method": "GET", "level": "error"},
|
||||
},
|
||||
{
|
||||
// A scalar and an object under the same prefix survive as two distinct dotted keys.
|
||||
name: "ScalarAndObjectKey_BothSurvive",
|
||||
paths: map[string]any{"scope": "x", "scope.attributes.name": "y"},
|
||||
want: map[string]any{"scope": "x", "scope.attributes.name": "y"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
j := chcol.NewJSON()
|
||||
for path, value := range testCase.paths {
|
||||
j.SetValueAtPath(path, value)
|
||||
}
|
||||
assert.Equal(t, testCase.want, FlattenJSON(*j))
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user