SigNoz/signoz
1
0
Fork 0
mirror of https://github.com/SigNoz/signoz.git synced 2026-08-14 00:40:35 +01:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity
Files
2ea58235600ea72fcfae54c9a00a55bcbdcab3b3
signoz/pkg/statementbuilder/meterstatementbuilder/statement_builder.go
Tushar Vats a7fd14eac9 refactor(metrics): alias group-by columns positionally like logs and traces (#12464) 
A metric label may be named after a column the generated query builds
for itself, and the metrics and meter builders selected group-by columns
under the label's own name — so `group by ts` or `group by value`
produced SQL with two columns of that name, which ClickHouse rejects.

### What

- Metrics and meter now alias group-by columns
`__GROUP_BY_KEY_<i>_<name>`, the scheme the logs and traces statement
builders already use.
- `pkg/querier/consume.go` already strips that prefix on all three read
paths (time-series, scalar, raw), so API responses are unchanged.
- Metrics' `ColumnExpressionFor` now returns the bare expression like
the logs and traces mappers. It was the only one returning an aliased
expression (`expr AS <name>`), which `agg_rewrite.go` splices inside a
function argument — giving `sum(expr AS <name>)` if metrics ever grows
expression aggregations. Callers alias and escape, as logs does.
- The histogram pipeline derives its CTE-side query once — `le` appended
last, plus the existing rate/sum rewrite — instead of mutating the query
and restoring it around the whole pipeline. The final select takes the
original minus `le`, so the remaining keys hold the positions their CTE
aliases were built from.

With a label named `ts`, before:

```sql
SELECT ts, `ts`, multiIf(…) … GROUP BY fingerprint, ts, `ts`
```

and after:

```sql
SELECT ts, `__GROUP_BY_KEY_0_ts`, multiIf(…) …
```

A label named `value` was the quieter case — the spatial CTE selected it
next to the aggregate of the same name:

```sql
SELECT ts, `value`, sum(per_series_value) AS value …
```

### Notes

- Meter comes along because it holds a
`*metricsstatementbuilder.StatementBuilder` and calls the shared
`BuildFinalSelect`; aliasing metrics alone would leave meter ordering by
an alias its own select never produced. `GroupByColumnAlias` /
`GroupByAliases` are exported for it, alongside the `GetKeySelectors` /
`RateTmpl` already shared across that boundary.
- Meter had the identical collision, so this fixes it there too.

### Testing

- `TestGroupByAliasAvoidsColumnCollision` covers `ts`, `value`,
`fingerprint` and an ordinary label, in both the metrics and meter
builders; all three collision cases fail without the change.
- `reduced_test.go` gains `histogram_p99_group_by` and
`gauge_avg_avg_group_by` — the reduced path had no group-by coverage at
all, so neither the aliases in its four CTE builders nor the union's
`ORDER BY` were exercised. The histogram case pins that both `UNION ALL`
branches emit the same columns.
- `test_histogram_count_no_param` pins the `SELECT *` branch, where `le`
stays unaliased so `ORDER BY toFloat64(le)` resolves.
- Both new behaviours were mutation-checked: appending `le` first
instead of last, and returning the bare name from `GroupByColumnAlias`,
each turn the relevant tests red.
- Twelve expected-SQL blobs regenerated across the metrics and meter
statement builder tests — alias-only diffs.
- `go test ./...` green, `make go-lint` clean.

Fixes https://github.com/SigNoz/engineering-pod/issues/5868
2026-08-13 16:35:36 +00:00

431 lines
16 KiB
Go
Raw Blame History

package meterstatementbuilder
import (
"context"
"fmt"
"log/slog"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
)
type meterQueryStatementBuilder struct {
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
metricsStatementBuilder *metricsstatementbuilder.StatementBuilder
}
var _ qbtypes.StatementBuilder[qbtypes.MetricAggregation] = (*meterQueryStatementBuilder)(nil)
// NewFactory returns a provider factory for the meter statement builder. Its New
// reuses the metrics FieldMapper/ConditionBuilder and delegates the final SELECT
// to a metrics statement builder built via the metrics factory.
func NewFactory(
metadataStore telemetrytypes.MetadataStore,
fl flagger.Flagger,
) factory.ProviderFactory[qbtypes.StatementBuilder[qbtypes.MetricAggregation], statementbuilder.Config] {
return factory.NewProviderFactory(
factory.MustNewName("meter"),
func(ctx context.Context, settings factory.ProviderSettings, cfg statementbuilder.Config) (qbtypes.StatementBuilder[qbtypes.MetricAggregation], error) {
metricsStatementBuilder, err := metricsstatementbuilder.NewFactory(metadataStore, fl).New(ctx, settings, cfg)
if err != nil {
return nil, err
}
fm := metricstelemetryschema.NewFieldMapper()
cb := metricstelemetryschema.NewConditionBuilder(fm)
return NewMeterQueryStatementBuilder(settings, metadataStore, fm, cb, metricsStatementBuilder), nil
},
)
}
func NewMeterQueryStatementBuilder(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
fieldMapper qbtypes.FieldMapper,
conditionBuilder qbtypes.ConditionBuilder,
metricsStatementBuilder *metricsstatementbuilder.StatementBuilder,
) *meterQueryStatementBuilder {
metricsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema")
return &meterQueryStatementBuilder{
logger: metricsSettings.Logger(),
metadataStore: metadataStore,
fm: fieldMapper,
cb: conditionBuilder,
metricsStatementBuilder: metricsStatementBuilder,
}
}
func (b *meterQueryStatementBuilder) Build(
ctx context.Context,
orgID valuer.UUID,
start uint64,
end uint64,
_ qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
keySelectors := metricsstatementbuilder.GetKeySelectors(query)
keys, _, err := b.metadataStore.GetKeysMulti(ctx, orgID, keySelectors)
if err != nil {
return nil, err
}
start, end = querybuilder.AdjustedMetricTimeRange(start, end, uint64(query.StepInterval.Seconds()), query)
return b.buildPipelineStatement(ctx, orgID, start, end, query, keys, variables)
}
func (b *meterQueryStatementBuilder) buildPipelineStatement(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
var (
cteFragments []string
cteArgs [][]any
)
if qbtypes.CanShortCircuitDelta(query.Aggregations[0]) {
// spatial_aggregation_cte directly for certain delta queries
if frag, args, err := b.buildTemporalAggDeltaFastPath(ctx, orgID, start, end, query, keys, variables); err != nil {
return nil, err
} else if frag != "" {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
} else {
// temporal_aggregation_cte
if frag, args, err := b.buildTemporalAggregationCTE(ctx, orgID, start, end, query, keys, variables); err != nil {
return nil, err
} else if frag != "" {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
// spatial_aggregation_cte
if frag, args, err := b.buildSpatialAggregationCTE(ctx, start, end, query, keys); err != nil {
return nil, err
} else if frag != "" {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
}
// final SELECT
return b.metricsStatementBuilder.BuildFinalSelect(cteFragments, cteArgs, query)
}
func (b *meterQueryStatementBuilder) buildTemporalAggDeltaFastPath(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
) (string, []any, error) {
var filterWhere querybuilder.PreparedWhereClause
var err error
stepSec := int64(query.StepInterval.Seconds())
sb := sqlbuilder.NewSelectBuilder()
sb.SelectMore(fmt.Sprintf(
"toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts",
stepSec,
))
for i, g := range query.GroupBy {
col, err := b.fm.ColumnExpressionFor(ctx, orgID, start, end, &g.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
return "", nil, err
}
sb.SelectMore(fmt.Sprintf("%s AS `%s`", sqlbuilder.Escape(col), metricsstatementbuilder.GroupByColumnAlias(i, g.Name)))
}
tbl := metertelemetryschema.WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := metertelemetryschema.AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
if err != nil {
return "", nil, err
}
if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
}
sb.SelectMore(fmt.Sprintf("%s AS value", aggCol))
sb.From(fmt.Sprintf("%s.%s AS points", metertelemetryschema.DBName, tbl))
sb.Where(
sb.In("metric_name", query.Aggregations[0].MetricName),
sb.GTE("unix_milli", start),
sb.LT("unix_milli", end),
)
if query.Filter != nil && query.Filter.Expression != "" {
filterWhere, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
Context: ctx,
OrgID: orgID,
Logger: b.logger,
FieldMapper: b.fm,
ConditionBuilder: b.cb,
FieldKeys: keys,
FullTextColumn: &telemetrytypes.TelemetryFieldKey{Name: "labels"},
Variables: variables,
StartNs: start,
EndNs: end,
})
if err != nil {
return "", nil, err
}
}
if !filterWhere.IsEmpty() {
sb.AddWhereClause(filterWhere.WhereClause)
}
if query.Aggregations[0].Temporality != metrictypes.Unknown {
sb.Where(sb.ILike("temporality", query.Aggregations[0].Temporality.StringValue()))
}
sb.GroupBy("ts")
sb.GroupBy(metricsstatementbuilder.GroupByAliases(query.GroupBy)...)
q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args, nil
}
func (b *meterQueryStatementBuilder) buildTemporalAggregationCTE(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
) (string, []any, error) {
if query.Aggregations[0].Temporality == metrictypes.Delta {
return b.buildTemporalAggDelta(ctx, orgID, start, end, query, keys, variables)
}
return b.buildTemporalAggCumulativeOrUnspecified(ctx, orgID, start, end, query, keys, variables)
}
func (b *meterQueryStatementBuilder) buildTemporalAggDelta(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
) (string, []any, error) {
var filterWhere querybuilder.PreparedWhereClause
var err error
stepSec := int64(query.StepInterval.Seconds())
sb := sqlbuilder.NewSelectBuilder()
sb.Select("fingerprint")
sb.SelectMore(fmt.Sprintf(
"toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts",
stepSec,
))
for i, g := range query.GroupBy {
col, err := b.fm.ColumnExpressionFor(ctx, orgID, start, end, &g.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
return "", nil, err
}
sb.SelectMore(fmt.Sprintf("%s AS `%s`", sqlbuilder.Escape(col), metricsstatementbuilder.GroupByColumnAlias(i, g.Name)))
}
tbl := metertelemetryschema.WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := metertelemetryschema.AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality,
query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
if err != nil {
return "", nil, err
}
if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
}
sb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
sb.From(fmt.Sprintf("%s.%s AS points", metertelemetryschema.DBName, tbl))
sb.Where(
sb.In("metric_name", query.Aggregations[0].MetricName),
sb.GTE("unix_milli", start),
sb.LT("unix_milli", end),
)
if query.Filter != nil && query.Filter.Expression != "" {
filterWhere, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
Context: ctx,
OrgID: orgID,
Logger: b.logger,
FieldMapper: b.fm,
ConditionBuilder: b.cb,
FieldKeys: keys,
FullTextColumn: &telemetrytypes.TelemetryFieldKey{Name: "labels"},
Variables: variables,
StartNs: start,
EndNs: end,
})
if err != nil {
return "", nil, err
}
}
if !filterWhere.IsEmpty() {
sb.AddWhereClause(filterWhere.WhereClause)
}
if query.Aggregations[0].Temporality != metrictypes.Unknown {
sb.Where(sb.ILike("temporality", query.Aggregations[0].Temporality.StringValue()))
}
sb.GroupBy("fingerprint", "ts")
sb.GroupBy(metricsstatementbuilder.GroupByAliases(query.GroupBy)...)
sb.OrderBy("fingerprint", "ts")
q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil
}
func (b *meterQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
) (string, []any, error) {
var filterWhere querybuilder.PreparedWhereClause
var err error
stepSec := int64(query.StepInterval.Seconds())
baseSb := sqlbuilder.NewSelectBuilder()
baseSb.Select("fingerprint")
baseSb.SelectMore(fmt.Sprintf(
"toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts",
stepSec,
))
for i, g := range query.GroupBy {
col, err := b.fm.ColumnExpressionFor(ctx, orgID, start, end, &g.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
return "", nil, err
}
baseSb.SelectMore(fmt.Sprintf("%s AS `%s`", sqlbuilder.Escape(col), metricsstatementbuilder.GroupByColumnAlias(i, g.Name)))
}
tbl := metertelemetryschema.WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := metertelemetryschema.AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
if err != nil {
return "", nil, err
}
baseSb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
baseSb.From(fmt.Sprintf("%s.%s AS points", metertelemetryschema.DBName, tbl))
baseSb.Where(
baseSb.In("metric_name", query.Aggregations[0].MetricName),
baseSb.GTE("unix_milli", start),
baseSb.LT("unix_milli", end),
)
if query.Filter != nil && query.Filter.Expression != "" {
filterWhere, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
Context: ctx,
OrgID: orgID,
Logger: b.logger,
FieldMapper: b.fm,
ConditionBuilder: b.cb,
FieldKeys: keys,
FullTextColumn: &telemetrytypes.TelemetryFieldKey{Name: "labels"},
Variables: variables,
StartNs: start,
EndNs: end,
})
if err != nil {
return "", nil, err
}
}
if !filterWhere.IsEmpty() {
baseSb.AddWhereClause(filterWhere.WhereClause)
}
if query.Aggregations[0].Temporality != metrictypes.Unknown {
baseSb.Where(baseSb.ILike("temporality", query.Aggregations[0].Temporality.StringValue()))
}
baseSb.GroupBy("fingerprint", "ts")
baseSb.GroupBy(metricsstatementbuilder.GroupByAliases(query.GroupBy)...)
baseSb.OrderBy("fingerprint", "ts")
innerQuery, innerArgs := baseSb.BuildWithFlavor(sqlbuilder.ClickHouse)
switch query.Aggregations[0].TimeAggregation {
case metrictypes.TimeAggregationRate:
wrapped := sqlbuilder.NewSelectBuilder()
wrapped.Select("ts")
for i, g := range query.GroupBy {
wrapped.SelectMore(fmt.Sprintf("`%s`", metricsstatementbuilder.GroupByColumnAlias(i, g.Name)))
}
wrapped.SelectMore(fmt.Sprintf("%s AS per_series_value", metricsstatementbuilder.RateTmpl))
wrapped.From(fmt.Sprintf("(%s) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)", innerQuery))
q, args := wrapped.BuildWithFlavor(sqlbuilder.ClickHouse, innerArgs...)
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil
case metrictypes.TimeAggregationIncrease:
wrapped := sqlbuilder.NewSelectBuilder()
wrapped.Select("ts")
for i, g := range query.GroupBy {
wrapped.SelectMore(fmt.Sprintf("`%s`", metricsstatementbuilder.GroupByColumnAlias(i, g.Name)))
}
wrapped.SelectMore(fmt.Sprintf("%s AS per_series_value", metricsstatementbuilder.IncreaseTmpl))
wrapped.From(fmt.Sprintf("(%s) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)", innerQuery))
q, args := wrapped.BuildWithFlavor(sqlbuilder.ClickHouse, innerArgs...)
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil
default:
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", innerQuery), innerArgs, nil
}
}
func (b *meterQueryStatementBuilder) buildSpatialAggregationCTE(
_ context.Context,
_ uint64,
_ uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
_ map[string][]*telemetrytypes.TelemetryFieldKey,
) (string, []any, error) {
if query.Aggregations[0].SpaceAggregation.IsZero() {
return "", nil, errors.Newf(
errors.TypeInvalidInput,
errors.CodeInvalidInput,
"invalid space aggregation, should be one of the following: [`sum`, `avg`, `min`, `max`, `count`]",
)
}
sb := sqlbuilder.NewSelectBuilder()
sb.Select("ts")
for i, g := range query.GroupBy {
sb.SelectMore(fmt.Sprintf("`%s`", metricsstatementbuilder.GroupByColumnAlias(i, g.Name)))
}
sb.SelectMore(fmt.Sprintf("%s(per_series_value) AS value", query.Aggregations[0].SpaceAggregation.StringValue()))
sb.From("__temporal_aggregation_cte")
sb.Where(sb.EQ("isNaN(per_series_value)", 0))
if query.Aggregations[0].ValueFilter != nil {
sb.Where(sb.EQ("per_series_value", query.Aggregations[0].ValueFilter.Value))
}
sb.GroupBy("ts")
sb.GroupBy(metricsstatementbuilder.GroupByAliases(query.GroupBy)...)
q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args, nil
}
Reference in New Issue View Git Blame Copy Permalink