hanzoai/o11y
2
0
Fork 0
mirror of https://github.com/hanzoai/o11y.git synced 2026-10-15 12:58:33 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity
Files
main
o11y/pkg/telemetrytraces/trace_operator_cte_builder_test.go
T
zooqueenandHanzo Dev e7a15ed8c9 Logs and traces reads bind the tenant's org 
The v5 logs and traces statement builders read event.log and event.span,
where every row carries its org. Each read now binds `org = ?` to the
tenant: the validated X-Org-Id for a request (set by the Hanzo IAM
identity), or the rule's org name for a scheduled rule that reads logs or
traces. A read with no tenant is refused as forbidden instead of reading
every org. This covers list, time series, scalar and trace requests and
both span reads of the trace operator.

Co-authored-by: Hanzo Dev <[email protected]>
2026-09-24 16:40:08 -07:00

736 lines
52 KiB
Go
Raw Permalink Blame History

package telemetrytraces
import (
"testing"
"time"
"github.com/hanzoai/o11y/pkg/flagger/flaggertest"
"github.com/hanzoai/o11y/pkg/instrumentation/instrumentationtest"
"github.com/hanzoai/o11y/pkg/querybuilder"
qbtypes "github.com/hanzoai/o11y/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/hanzoai/o11y/pkg/types/telemetrytypes"
"github.com/hanzoai/o11y/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/stretchr/testify/require"
)
func newTestTraceOperatorStatementBuilder(t *testing.T) *traceOperatorStatementBuilder {
t.Helper()
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
fl := flaggertest.New(t)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)
traceStmtBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, fm, cb, aggExprRewriter, nil, fl, false, 100000,
)
return NewTraceOperatorStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, fm, cb, traceStmtBuilder, aggExprRewriter, fl,
)
}
func TestTraceOperatorStatementBuilder(t *testing.T) {
cases := []struct {
name string
requestType qbtypes.RequestType
operator qbtypes.QueryBuilderTraceOperator
compositeQuery *qbtypes.CompositeQuery
expected qbtypes.Statement
expectedErr error
}{
{
name: "simple direct descendant operator",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A => B",
SelectFields: []telemetrytypes.TelemetryFieldKey{
{
Name: "service.name",
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
{
Name: "name",
FieldContext: telemetrytypes.FieldContextSpan,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
Limit: 10,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'frontend'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'backend'",
},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_DIR_DESC_B AS (SELECT p.* FROM A AS p INNER JOIN B AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent) SELECT time AS `timestamp`, trace_id, span_id, name, duration AS `duration_nano`, parent AS `parent_span_id`, service AS `service.name` FROM A_DIR_DESC_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, 10},
},
expectedErr: nil,
},
{
name: "indirect descendant operator",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A -> B",
Limit: 5,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'gateway'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'database'",
},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id) SELECT time AS `timestamp`, trace_id, span_id, name, duration AS `duration_nano`, parent AS `parent_span_id`, attributes['trace_state'] AS `trace_state`, toUInt32OrZero(attributes['flags']) AS `flags`, toInt8(multiIf(kind = 'internal', 1, kind = 'server', 2, kind = 'client', 3, kind = 'producer', 4, kind = 'consumer', 5, 0)) AS `kind`, kind AS `kind_string`, toInt16(multiIf(status = 'error', 2, status = 'ok', 1, 0)) AS `status_code`, attributes['status.message'] AS `status_message`, status AS `status_code_string`, JSONExtractArrayRaw(attributes['span.events']) AS `events`, attributes['links'] AS `links`, if(attributes['http.response.status_code'] != '', attributes['http.response.status_code'], attributes['http.status_code']) AS `response_status_code`, url AS `external_http_url`, url AS `http_url`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `external_http_method`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `http_method`, host AS `http_host`, attributes['db.name'] AS `db_name`, attributes['db.operation'] AS `db_operation`, toBool(status = 'error') AS `has_error`, attributes['is_remote'] AS `is_remote`, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string FROM A_INDIR_DESC_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, 5},
},
expectedErr: nil,
},
{
name: "AND operator",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A && B",
Limit: 15,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'frontend'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'backend'",
},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_AND_B AS (SELECT l.* FROM A AS l INNER JOIN B AS r ON l.trace_id = r.trace_id) SELECT time AS `timestamp`, trace_id, span_id, name, duration AS `duration_nano`, parent AS `parent_span_id`, attributes['trace_state'] AS `trace_state`, toUInt32OrZero(attributes['flags']) AS `flags`, toInt8(multiIf(kind = 'internal', 1, kind = 'server', 2, kind = 'client', 3, kind = 'producer', 4, kind = 'consumer', 5, 0)) AS `kind`, kind AS `kind_string`, toInt16(multiIf(status = 'error', 2, status = 'ok', 1, 0)) AS `status_code`, attributes['status.message'] AS `status_message`, status AS `status_code_string`, JSONExtractArrayRaw(attributes['span.events']) AS `events`, attributes['links'] AS `links`, if(attributes['http.response.status_code'] != '', attributes['http.response.status_code'], attributes['http.status_code']) AS `response_status_code`, url AS `external_http_url`, url AS `http_url`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `external_http_method`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `http_method`, host AS `http_host`, attributes['db.name'] AS `db_name`, attributes['db.operation'] AS `db_operation`, toBool(status = 'error') AS `has_error`, attributes['is_remote'] AS `is_remote`, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string FROM A_AND_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, 15},
},
expectedErr: nil,
},
{
name: "OR operator",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A || B",
Limit: 20,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'frontend'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'backend'",
},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_OR_B AS (SELECT * FROM A UNION DISTINCT SELECT * FROM B) SELECT time AS `timestamp`, trace_id, span_id, name, duration AS `duration_nano`, parent AS `parent_span_id`, attributes['trace_state'] AS `trace_state`, toUInt32OrZero(attributes['flags']) AS `flags`, toInt8(multiIf(kind = 'internal', 1, kind = 'server', 2, kind = 'client', 3, kind = 'producer', 4, kind = 'consumer', 5, 0)) AS `kind`, kind AS `kind_string`, toInt16(multiIf(status = 'error', 2, status = 'ok', 1, 0)) AS `status_code`, attributes['status.message'] AS `status_message`, status AS `status_code_string`, JSONExtractArrayRaw(attributes['span.events']) AS `events`, attributes['links'] AS `links`, if(attributes['http.response.status_code'] != '', attributes['http.response.status_code'], attributes['http.status_code']) AS `response_status_code`, url AS `external_http_url`, url AS `http_url`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `external_http_method`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `http_method`, host AS `http_host`, attributes['db.name'] AS `db_name`, attributes['db.operation'] AS `db_operation`, toBool(status = 'error') AS `has_error`, attributes['is_remote'] AS `is_remote`, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string FROM A_OR_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, 20},
},
expectedErr: nil,
},
{
name: "NOT operator",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A NOT B",
Limit: 10,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'frontend'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'backend'",
},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_not_B AS (SELECT l.* FROM A AS l WHERE l.trace_id GLOBAL NOT IN (SELECT DISTINCT trace_id FROM B)) SELECT time AS `timestamp`, trace_id, span_id, name, duration AS `duration_nano`, parent AS `parent_span_id`, attributes['trace_state'] AS `trace_state`, toUInt32OrZero(attributes['flags']) AS `flags`, toInt8(multiIf(kind = 'internal', 1, kind = 'server', 2, kind = 'client', 3, kind = 'producer', 4, kind = 'consumer', 5, 0)) AS `kind`, kind AS `kind_string`, toInt16(multiIf(status = 'error', 2, status = 'ok', 1, 0)) AS `status_code`, attributes['status.message'] AS `status_message`, status AS `status_code_string`, JSONExtractArrayRaw(attributes['span.events']) AS `events`, attributes['links'] AS `links`, if(attributes['http.response.status_code'] != '', attributes['http.response.status_code'], attributes['http.status_code']) AS `response_status_code`, url AS `external_http_url`, url AS `http_url`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `external_http_method`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `http_method`, host AS `http_host`, attributes['db.name'] AS `db_name`, attributes['db.operation'] AS `db_operation`, toBool(status = 'error') AS `has_error`, attributes['is_remote'] AS `is_remote`, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string FROM A_not_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, 10},
},
expectedErr: nil,
},
{
name: "time series query with aggregations",
requestType: qbtypes.RequestTypeTimeSeries,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A => B",
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
Aggregations: []qbtypes.TraceAggregation{
{
Expression: "count()",
},
},
GroupBy: []qbtypes.GroupByKey{
{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "service.name",
},
},
},
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'frontend'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'backend'",
},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_DIR_DESC_B AS (SELECT p.* FROM A AS p INNER JOIN B AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent) SELECT toStartOfInterval(time, INTERVAL 60 SECOND) AS ts, toString(multiIf(notEmpty(service), service, NULL)) AS `service.name`, count() AS __result_0 FROM A_DIR_DESC_B GROUP BY ts, `service.name` ORDER BY ts desc SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant},
},
expectedErr: nil,
},
{
name: "scalar query with aggregation and group by",
requestType: qbtypes.RequestTypeScalar,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A && B",
Aggregations: []qbtypes.TraceAggregation{
{
Expression: "avg(duration_nano)",
},
},
GroupBy: []qbtypes.GroupByKey{
{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "service.name",
},
},
},
Order: []qbtypes.OrderBy{
{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "__result_0",
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
Limit: 10,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'frontend'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "response_status_code < 400",
},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), B AS (SELECT * FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ? AND toFloat64(if(attributes['http.response.status_code'] != '', attributes['http.response.status_code'], attributes['http.status_code'])) < ?), A_AND_B AS (SELECT l.* FROM A AS l INNER JOIN B AS r ON l.trace_id = r.trace_id) SELECT toString(multiIf(notEmpty(service), service, NULL)) AS `service.name`, avg(multiIf(duration <> ?, duration, NULL)) AS __result_0 FROM A_AND_B GROUP BY `service.name` ORDER BY __result_0 desc SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, float64(400), 0},
},
expectedErr: nil,
},
{
name: "complex nested expression",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "(A => B) && (C => D)",
Limit: 5,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'frontend'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'backend'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "C",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'auth'",
},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "D",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'database'",
},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_DIR_DESC_B AS (SELECT p.* FROM A AS p INNER JOIN B AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent), __resource_filter_C AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_D AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), D AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_D) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), C_DIR_DESC_D AS (SELECT p.* FROM C AS p INNER JOIN D AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent), A_DIR_DESC_B_AND_C_DIR_DESC_D AS (SELECT l.* FROM A_DIR_DESC_B AS l INNER JOIN C_DIR_DESC_D AS r ON l.trace_id = r.trace_id) SELECT time AS `timestamp`, trace_id, span_id, name, duration AS `duration_nano`, parent AS `parent_span_id`, attributes['trace_state'] AS `trace_state`, toUInt32OrZero(attributes['flags']) AS `flags`, toInt8(multiIf(kind = 'internal', 1, kind = 'server', 2, kind = 'client', 3, kind = 'producer', 4, kind = 'consumer', 5, 0)) AS `kind`, kind AS `kind_string`, toInt16(multiIf(status = 'error', 2, status = 'ok', 1, 0)) AS `status_code`, attributes['status.message'] AS `status_message`, status AS `status_code_string`, JSONExtractArrayRaw(attributes['span.events']) AS `events`, attributes['links'] AS `links`, if(attributes['http.response.status_code'] != '', attributes['http.response.status_code'], attributes['http.status_code']) AS `response_status_code`, url AS `external_http_url`, url AS `http_url`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `external_http_method`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `http_method`, host AS `http_host`, attributes['db.name'] AS `db_name`, attributes['db.operation'] AS `db_operation`, toBool(status = 'error') AS `has_error`, attributes['is_remote'] AS `is_remote`, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string FROM A_DIR_DESC_B_AND_C_DIR_DESC_D ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "auth", "%service.name%", "%service.name\":\"auth%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, 5},
},
expectedErr: nil,
},
{
name: "returnSpansFrom B: A -> B return B spans filtered by operator",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A -> B",
ReturnSpansFrom: "B",
Limit: 10,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "service.name = 'gateway'"},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "service.name = 'database'"},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __return_from_B AS (SELECT * FROM B WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B)) SELECT time AS `timestamp`, trace_id, span_id, name, duration AS `duration_nano`, parent AS `parent_span_id`, attributes['trace_state'] AS `trace_state`, toUInt32OrZero(attributes['flags']) AS `flags`, toInt8(multiIf(kind = 'internal', 1, kind = 'server', 2, kind = 'client', 3, kind = 'producer', 4, kind = 'consumer', 5, 0)) AS `kind`, kind AS `kind_string`, toInt16(multiIf(status = 'error', 2, status = 'ok', 1, 0)) AS `status_code`, attributes['status.message'] AS `status_message`, status AS `status_code_string`, JSONExtractArrayRaw(attributes['span.events']) AS `events`, attributes['links'] AS `links`, if(attributes['http.response.status_code'] != '', attributes['http.response.status_code'], attributes['http.status_code']) AS `response_status_code`, url AS `external_http_url`, url AS `http_url`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `external_http_method`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `http_method`, host AS `http_host`, attributes['db.name'] AS `db_name`, attributes['db.operation'] AS `db_operation`, toBool(status = 'error') AS `has_error`, attributes['is_remote'] AS `is_remote`, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string FROM __return_from_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, 10},
},
expectedErr: nil,
},
{
name: "returnSpansFrom C: (A -> B) && C return C spans filtered by operator",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "(A -> B) && C",
ReturnSpansFrom: "C",
Limit: 10,
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "service.name = 'gateway'"},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "service.name = 'database'"},
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "C",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "service.name = 'auth'"},
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, service AS `service.name` FROM event.span WHERE time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_A AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), __resource_filter_B AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __resource_filter_C AS (SELECT fingerprint FROM event.span_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM event.span WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND time >= ? AND time < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND org = ?), A_INDIR_DESC_B_AND_C AS (SELECT l.* FROM A_INDIR_DESC_B AS l INNER JOIN C AS r ON l.trace_id = r.trace_id), __return_from_C AS (SELECT * FROM C WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B_AND_C)) SELECT time AS `timestamp`, trace_id, span_id, name, duration AS `duration_nano`, parent AS `parent_span_id`, attributes['trace_state'] AS `trace_state`, toUInt32OrZero(attributes['flags']) AS `flags`, toInt8(multiIf(kind = 'internal', 1, kind = 'server', 2, kind = 'client', 3, kind = 'producer', 4, kind = 'consumer', 5, 0)) AS `kind`, kind AS `kind_string`, toInt16(multiIf(status = 'error', 2, status = 'ok', 1, 0)) AS `status_code`, attributes['status.message'] AS `status_message`, status AS `status_code_string`, JSONExtractArrayRaw(attributes['span.events']) AS `events`, attributes['links'] AS `links`, if(attributes['http.response.status_code'] != '', attributes['http.response.status_code'], attributes['http.status_code']) AS `response_status_code`, url AS `external_http_url`, url AS `http_url`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `external_http_method`, if(attributes['http.request.method'] != '', attributes['http.request.method'], attributes['http.method']) AS `http_method`, host AS `http_host`, attributes['db.name'] AS `db_name`, attributes['db.operation'] AS `db_operation`, toBool(status = 'error') AS `has_error`, attributes['is_remote'] AS `is_remote`, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string FROM __return_from_C ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, "auth", "%service.name%", "%service.name\":\"auth%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), testTenant, 10},
},
expectedErr: nil,
},
}
statementBuilder := newTestTraceOperatorStatementBuilder(t)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
// Parse the operator expression
err := c.operator.ParseExpression()
require.NoError(t, err)
q, err := statementBuilder.Build(
tenantCtx(),
1747947419000,
1747983448000,
c.requestType,
c.operator,
c.compositeQuery,
)
if c.expectedErr != nil {
require.Error(t, err)
require.Contains(t, err.Error(), c.expectedErr.Error())
} else {
require.NoError(t, err)
require.Equal(t, c.expected.Query, q.Query)
require.Equal(t, c.expected.Args, q.Args)
require.Equal(t, c.expected.Warnings, q.Warnings)
}
})
}
}
func TestTraceOperatorStatementBuilderErrors(t *testing.T) {
cases := []struct {
name string
operator qbtypes.QueryBuilderTraceOperator
compositeQuery *qbtypes.CompositeQuery
expectedErr string
}{
{
name: "missing referenced query",
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A => B",
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
},
},
// Missing query B
},
},
expectedErr: "referenced query 'B' not found",
},
{
name: "nil composite query",
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A => B",
},
compositeQuery: nil,
expectedErr: "compositeQuery cannot be nil",
},
{
name: "unsupported operator",
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A XOR B", // Assuming XOR is not supported
},
compositeQuery: &qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
},
},
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "B",
Signal: telemetrytypes.SignalTraces,
},
},
},
},
expectedErr: "invalid query reference 'A XOR B'",
},
}
statementBuilder := newTestTraceOperatorStatementBuilder(t)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
// Parse the operator expression
err := c.operator.ParseExpression()
if err == nil { // Only proceed if parsing succeeded
_, err = statementBuilder.Build(
tenantCtx(),
1747947419000,
1747983448000,
qbtypes.RequestTypeRaw,
c.operator,
c.compositeQuery,
)
}
require.Error(t, err)
require.Contains(t, err.Error(), c.expectedErr)
})
}
}
func TestTraceOperatorStatementBuilderAdjustsKeys(t *testing.T) {
cases := []struct {
name string
requestType qbtypes.RequestType
operator qbtypes.QueryBuilderTraceOperator
builderFilter string
wantSQL string
wantArgs []any
}{
{
name: "deprecated duration filter in referenced builder query",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A",
Limit: 10,
},
builderFilter: "durationNano = '3s'",
wantSQL: "duration = ?",
wantArgs: []any{int64(3000000000)},
},
{
name: "context-prefixed aggregation alias in order by",
requestType: qbtypes.RequestTypeScalar,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A",
Aggregations: []qbtypes.TraceAggregation{
{
Expression: "count()",
Alias: "span.count_",
},
},
Order: []qbtypes.OrderBy{
{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "count_",
FieldContext: telemetrytypes.FieldContextSpan,
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
},
wantSQL: "ORDER BY __result_0 desc",
},
}
statementBuilder := newTestTraceOperatorStatementBuilder(t)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
err := c.operator.ParseExpression()
require.NoError(t, err)
filter := c.builderFilter
if filter == "" {
filter = "service.name = 'frontend'"
}
q, err := statementBuilder.Build(
tenantCtx(),
1747947419000,
1747983448000,
c.requestType,
c.operator,
&qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: filter},
},
},
},
},
)
require.NoError(t, err)
require.Contains(t, q.Query, c.wantSQL)
for _, arg := range c.wantArgs {
require.Contains(t, q.Args, arg)
}
})
}
}
// TestTraceOperatorStatementBuilderDeduplicatesKeys checks that a trace
// operator with the same field name listed twice in GroupBy (once with a
// context, once without) ends up with a single column in the outer SELECT
// and a single entry in GROUP BY.
func TestTraceOperatorStatementBuilderDeduplicatesKeys(t *testing.T) {
statementBuilder := newTestTraceOperatorStatementBuilder(t)
operator := qbtypes.QueryBuilderTraceOperator{
Expression: "A",
Aggregations: []qbtypes.TraceAggregation{
{Expression: "count()"},
},
GroupBy: []qbtypes.GroupByKey{
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "http.method",
FieldContext: telemetrytypes.FieldContextAttribute,
}},
// Same name, no context — should be merged with the entry above.
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "http.method",
}},
},
}
require.NoError(t, operator.ParseExpression())
q, err := statementBuilder.Build(
tenantCtx(),
1747947419000,
1747983448000,
qbtypes.RequestTypeScalar,
operator,
&qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "service.name = 'frontend'"},
},
},
},
},
)
require.NoError(t, err)
require.Contains(t, q.Query,
"SELECT toString(multiIf(mapContains(attributes, 'http.method') = ?, attributes['http.method'], NULL)) AS `http.method`, count() AS __result_0 FROM A GROUP BY `http.method` ORDER BY __result_0 DESC")
}
Reference in New Issue View Git Blame Copy Permalink