chore(promql): add clickhousev2 provider (push agg to CH) and shadow-compare promql behind a flag

.github/workflows/integrationci.yaml

@@ -58,6 +58,7 @@ jobs:
           - rootuser
           - serviceaccount
           - querier_json_body
+          - promqlparity
           - querier_skip_resource_fingerprint
           - ttl
         sqlstore-provider:

pkg/flagger/registry.go

@@ -15,6 +15,8 @@ var (
 	FeatureEnableAIObservability  = featuretypes.MustNewName("enable_ai_observability")
 	FeatureEnableMetricsReduction = featuretypes.MustNewName("enable_metrics_reduction")
 	FeatureUseInfraMonitoringV2   = featuretypes.MustNewName("use_infra_monitoring_v2")
+
+	FeatureUsePrometheusClickhouseV2 = featuretypes.MustNewName("use_prometheus_clickhouse_v2")
 )
 
 func MustNewRegistry() featuretypes.Registry {
@@ -115,6 +117,14 @@ func MustNewRegistry() featuretypes.Registry {
 			DefaultVariant: featuretypes.MustNewName("disabled"),
 			Variants:       featuretypes.NewBooleanVariants(),
 		},
+		&featuretypes.Feature{
+			Name:           FeatureUsePrometheusClickhouseV2,
+			Kind:           featuretypes.KindBoolean,
+			Stage:          featuretypes.StageExperimental,
+			Description:    "Runs PromQL queries on the clickhousev2 provider alongside the served engine result and logs any difference; serving is unaffected.",
+			DefaultVariant: featuretypes.MustNewName("disabled"),
+			Variants:       featuretypes.NewBooleanVariants(),
+		},
 	)
 	if err != nil {
 		panic(err)

pkg/prometheus/clickhouseprometheusv2/capture.go

@@ -0,0 +1,89 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"sync"
+
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/util/annotations"
+)
+
+// statementRecorder collects the statements a PromQL evaluation would run.
+// Safe for concurrent use: the engine may Select selectors concurrently.
+type statementRecorder struct {
+	mu         sync.Mutex
+	statements []prometheus.CapturedStatement
+}
+
+func (r *statementRecorder) record(query string, args []any) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.statements = append(r.statements, prometheus.CapturedStatement{Query: query, Args: args})
+}
+
+func (r *statementRecorder) Statements() []prometheus.CapturedStatement {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make([]prometheus.CapturedStatement, len(r.statements))
+	copy(out, r.statements)
+	return out
+}
+
+type captureQueryable struct {
+	client   *client
+	recorder *statementRecorder
+}
+
+func (c *captureQueryable) Querier(mint, maxt int64) (storage.Querier, error) {
+	return &captureQuerier{
+		querier:  querier{mint: mint, maxt: maxt, client: c.client},
+		recorder: c.recorder,
+	}, nil
+}
+
+// captureQuerier builds the same SQL as the live querier but records it and
+// returns no data. The fingerprint filter always takes the subquery form:
+// without executing the series lookup, the inline literal set is unknown.
+type captureQuerier struct {
+	querier
+	recorder *statementRecorder
+}
+
+func (c *captureQuerier) Select(ctx context.Context, _ bool, hints *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
+	if rawQuery, ok := rawSQLQuery(matchers); ok {
+		c.recorder.record(rawQuery, nil)
+		return storage.EmptySeriesSet()
+	}
+
+	start, end := c.window(hints)
+
+	samplesQuery, args, err := buildSamplesQuery(start, end, metricNamesFromMatchers(matchers), nil, matchers, c.lastSamplePerStepFor(ctx, hints))
+	if err != nil {
+		return storage.ErrSeriesSet(err)
+	}
+	c.recorder.record(samplesQuery, args)
+
+	return storage.EmptySeriesSet()
+}
+
+func (c *captureQuerier) LabelValues(context.Context, string, *storage.LabelHints, ...*labels.Matcher) ([]string, annotations.Annotations, error) {
+	return nil, nil, nil
+}
+
+func (c *captureQuerier) LabelNames(context.Context, *storage.LabelHints, ...*labels.Matcher) ([]string, annotations.Annotations, error) {
+	return nil, nil, nil
+}
+
+// metricNamesFromMatchers extracts the statically known metric name, if any.
+// The live path derives names from the matched series; the capture path has
+// no execution results, so only a __name__ equality contributes.
+func metricNamesFromMatchers(matchers []*labels.Matcher) []string {
+	for _, m := range matchers {
+		if m.Name == metricNameLabel && m.Type == labels.MatchEqual && m.Value != "" {
+			return []string{m.Value}
+		}
+	}
+	return nil
+}

pkg/prometheus/clickhouseprometheusv2/client.go

@@ -0,0 +1,282 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"math"
+	"slices"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
+	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/prometheus/prometheus/model/labels"
+	promValue "github.com/prometheus/prometheus/model/value"
+)
+
+// seriesLookup is a series-lookup result: matched fingerprints with their
+// labels, and the distinct metric names seen on them.
+type seriesLookup struct {
+	fingerprints map[uint64]labels.Labels
+	metricNames  []string
+}
+
+// client executes the series, samples and raw queries against ClickHouse.
+type client struct {
+	settings       factory.ScopedProviderSettings
+	telemetryStore telemetrystore.TelemetryStore
+	cfg            prometheus.ClickhouseV2Config
+	lookbackMs     int64
+}
+
+func newClient(settings factory.ScopedProviderSettings, telemetryStore telemetrystore.TelemetryStore, cfg prometheus.Config) *client {
+	lookback := cfg.LookbackDelta
+	if lookback <= 0 {
+		// Mirror the engine: promql defaults an unset lookback to 5m.
+		lookback = defaultLookbackDelta
+	}
+	return &client{
+		settings:       settings,
+		telemetryStore: telemetryStore,
+		cfg:            cfg.ClickhouseV2,
+		lookbackMs:     lookback.Milliseconds(),
+	}
+}
+
+func (c *client) withContext(ctx context.Context, functionName string) context.Context {
+	return ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "clickhouse-prometheus-v2",
+		instrumentationtypes.CodeFunctionName: functionName,
+	})
+}
+
+// selectSeries runs the series lookup for the given matchers and window.
+func (c *client) selectSeries(ctx context.Context, query string, args []any) (*seriesLookup, error) {
+	ctx = c.withContext(ctx, "selectSeries")
+	rows, err := c.telemetryStore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	lookup := &seriesLookup{fingerprints: make(map[uint64]labels.Labels)}
+	names := make(map[string]struct{})
+
+	var fingerprint uint64
+	var labelsJSON string
+	for rows.Next() {
+		if err := rows.Scan(&fingerprint, &labelsJSON); err != nil {
+			return nil, err
+		}
+		lset, err := unmarshalLabels(labelsJSON)
+		if err != nil {
+			return nil, err
+		}
+		lookup.fingerprints[fingerprint] = lset
+		if name := lset.Get(metricNameLabel); name != "" {
+			names[name] = struct{}{}
+		}
+		if c.cfg.MaxFetchedSeries > 0 && len(lookup.fingerprints) > c.cfg.MaxFetchedSeries {
+			return nil, errors.NewInvalidInputf(
+				errors.CodeInvalidInput,
+				"promql selector matched more than %d series; narrow the label matchers or raise prometheus::clickhousev2::max_fetched_series",
+				c.cfg.MaxFetchedSeries,
+			)
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	for name := range names {
+		lookup.metricNames = append(lookup.metricNames, name)
+	}
+	slices.Sort(lookup.metricNames)
+
+	return lookup, nil
+}
+
+// unmarshalLabels parses the labels JSON column. Unlike v1, the fingerprint
+// is not injected as a synthetic label (it would take part in `without (...)`
+// grouping and vector matching) and empty-valued labels are dropped: an empty
+// label value means "label absent" in Prometheus, and upstream never produces
+// such labels, but stored attribute JSON can carry them.
+func unmarshalLabels(s string) (labels.Labels, error) {
+	m := make(map[string]string)
+	if err := json.Unmarshal([]byte(s), &m); err != nil {
+		return labels.EmptyLabels(), err
+	}
+	builder := labels.NewScratchBuilder(len(m))
+	for k, v := range m {
+		if v == "" {
+			continue
+		}
+		builder.Add(k, v)
+	}
+	builder.Sort()
+	return builder.Labels(), nil
+}
+
+// selectSamples executes a samples query (raw or last-sample-per-step; both
+// produce the same column shape) and assembles the per-series sample slices.
+// Rows arrive ordered by (fingerprint, unix_milli). Rows whose fingerprint
+// is missing from the lookup are skipped (possible in the subquery filter
+// mode, where the fingerprint filter re-runs after the lookup and can see
+// series born in between). Stale flags map to the engine's StaleNaN.
+// Duplicate timestamps pass through as stored: upstream Prometheus cannot
+// produce them (its TSDB rejects them at ingest), our ingest can under
+// at-least-once retries, and v1 feeds them to the engine as-is —
+// deduplicating here would make this provider silently disagree with both
+// v1 and the transpiled statements over the same dirty data. Uniqueness
+// belongs to the ingest layer.
+func (c *client) selectSamples(ctx context.Context, query string, args []any, lookup *seriesLookup) ([]*series, error) {
+	ctx = c.withContext(ctx, "selectSamples")
+	rows, err := c.telemetryStore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var (
+		result       []*series
+		current      *series
+		fingerprint  uint64
+		prevFp       uint64
+		timestampMs  int64
+		val          float64
+		flags        uint32
+		first        = true
+		haveCurrent  bool
+		staleMarker  = math.Float64frombits(promValue.StaleNaN)
+		maxSamples   = c.cfg.MaxFetchedSamples
+		fetched      int64
+		unknownCount int
+	)
+
+	for rows.Next() {
+		if err := rows.Scan(&fingerprint, &timestampMs, &val, &flags); err != nil {
+			return nil, err
+		}
+
+		fetched++
+		if maxSamples > 0 && fetched > maxSamples {
+			return nil, errors.NewInvalidInputf(
+				errors.CodeInvalidInput,
+				"promql query would fetch more than %d samples; narrow the selector or time range, or raise prometheus::clickhousev2::max_fetched_samples",
+				maxSamples,
+			)
+		}
+
+		if first || fingerprint != prevFp {
+			first = false
+			prevFp = fingerprint
+			lset, ok := lookup.fingerprints[fingerprint]
+			if !ok {
+				unknownCount++
+				haveCurrent = false
+				continue
+			}
+			current = &series{lset: lset}
+			result = append(result, current)
+			haveCurrent = true
+		}
+		if !haveCurrent {
+			// Remaining rows of a fingerprint missing from the lookup.
+			continue
+		}
+
+		if flags&1 == 1 {
+			val = staleMarker
+		}
+		current.ts = append(current.ts, timestampMs)
+		current.vs = append(current.vs, val)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	if unknownCount > 0 {
+		c.settings.Logger().DebugContext(ctx, "skipped samples of fingerprints missing from series lookup",
+			slog.Int("unknown_fingerprints", unknownCount))
+	}
+
+	return result, nil
+}
+
+// queryRaw supports the {job="rawsql", query="..."} escape hatch: the value of
+// the query matcher runs as-is, each row becoming a single-sample series
+// stamped at the query end. Column "value" is the sample value; every other
+// column is a label.
+func (c *client) queryRaw(ctx context.Context, query string, ts int64) ([]*series, error) {
+	ctx = c.withContext(ctx, "queryRaw")
+	rows, err := c.telemetryStore.ClickhouseDB().Query(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	columns := rows.Columns()
+	targets := make([]any, len(columns))
+	for i := range targets {
+		targets[i] = new(scanner)
+	}
+
+	var result []*series
+	for rows.Next() {
+		if err := rows.Scan(targets...); err != nil {
+			return nil, err
+		}
+		builder := labels.NewScratchBuilder(len(columns))
+		var val float64
+		for i, col := range columns {
+			v := targets[i].(*scanner)
+			if col == "value" {
+				val = v.f
+				continue
+			}
+			builder.Add(col, v.s)
+		}
+		builder.Sort()
+		result = append(result, &series{
+			lset: builder.Labels(),
+			ts:   []int64{ts},
+			vs:   []float64{val},
+		})
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return result, nil
+}
+
+var _ sql.Scanner = (*scanner)(nil)
+
+type scanner struct {
+	f float64
+	s string
+}
+
+func (s *scanner) Scan(val any) error {
+	s.f = 0
+	s.s = ""
+
+	s.s = fmt.Sprintf("%v", val)
+	switch val := val.(type) {
+	case int64:
+		s.f = float64(val)
+	case uint64:
+		s.f = float64(val)
+	case float64:
+		s.f = val
+	case []byte:
+		s.s = string(val)
+	}
+	return nil
+}

pkg/prometheus/clickhouseprometheusv2/doc.go

@@ -0,0 +1,334 @@
+// Package clickhouseprometheusv2 is the second-generation ClickHouse-backed
+// Prometheus provider. It exists because the v1 provider fetches every raw
+// sample of a query's union window through the remote-read protobuf layer
+// and hands it to the engine — the cost is a function of ingested data, not
+// of the question asked, which is how a dashboard of PromQL panels takes an
+// instance down.
+//
+// Every query runs in one of two ways, decided per query:
+//
+//   - Transpiled: the query is evaluated entirely inside ClickHouse and only
+//     final (or near-final) per-group grid arrays come back, built on the
+//     timeSeries*ToGrid aggregate functions (the supported ClickHouse floor
+//     is >= 25.6, so they are assumed available).
+//   - Engine: the stock promql engine evaluates over this package's native
+//     storage.Querier. This is the path for everything not transpilable.
+//
+// Correctness is the constraint that shaped both paths: a PromQL result that
+// differs from upstream Prometheus is a lost user, so anything that cannot
+// reproduce engine semantics exactly falls back rather than approximate.
+// The rest of this comment is the PromQL -> SQL story, because that mapping
+// is where correctness is won or lost.
+//
+// # The evaluation model the SQL must reproduce
+//
+// A PromQL range query is an instant query evaluated at every grid point
+// t_i = start + i*step, i = 0..(end-start)/step. At each t_i:
+//
+//   - an instant selector resolves to the latest sample in the left-open
+//     lookback window (t_i - lookback, t_i], and to nothing when that latest
+//     sample is a stale marker — even if older real samples sit inside the
+//     window;
+//   - a range selector [r] collects every sample in (t_i - r, t_i], stale
+//     markers excluded;
+//   - offset d shifts both windows to (t_i - d - w, t_i - d].
+//
+// The transpilation invariant follows from this: every transpiled construct
+// produces, per output series, one array with exactly one slot per grid
+// point — slot i holds the value at t_i, NULL means absent. This is what
+// makes composition correct, not just convenient: the engine evaluates
+// these operators independently per t_i, so any representation that gets
+// every slot right gets the whole query right, and spatial aggregation over
+// arrays is sound because it combines values that belong to the same t_i by
+// construction. Slot index i maps back to t_i = start + i*step at scan time
+// (toMatrix). Everything below is about filling those slots with exactly
+// the numbers the engine would compute — and each equivalence was validated
+// against the vendored engine on live data before its shape entered the
+// allowlist; anything unproven stays on the engine path.
+//
+// # Classification: finding what a statement can answer
+//
+// classify walks the parsed AST looking for "core units" — maximal subtrees
+// of the shape
+//
+//	[agg by/without (...)] [fn(] selector[range] [offset d] [)] [op scalar]...
+//
+// classifyCore peels that chain from the outside in: an optional
+// sum/min/max/avg/count aggregation, then one of the allowlisted functions
+// or a bare instant selector, then the selector with its offset; on the way
+// out it accumulates number-literal arithmetic, comparisons (including
+// bool) and unary minus into a scalar-op pipeline. A node qualifies only if
+// its type, arguments and children are in the proven set — an allowlist, so
+// an overlooked construct becomes a fallback instead of a wrong number.
+//
+// Three unit kinds come out of this, each with its own SQL form:
+// unitRange (rate, irate, increase, delta, idelta over a range selector),
+// unitInstant (instant vector selection, bare or comparison-filtered) and
+// unitOverTime (avg/min/max/sum/count/last _over_time).
+//
+// If the entire tree is one unit, the plan is "full": the statement's rows
+// are the query result. Otherwise every maximal unit is cut out and replaced
+// in the expression with a synthetic selector __signoz_transpiled_N__, and
+// the rewritten expression runs in the engine over the units' materialized
+// results ("hybrid") — histogram_quantile, topk, or/and/unless and vector
+// matching keep exact engine semantics while their expensive inputs were
+// aggregated server-side.
+//
+// Classification refuses when exact semantics cannot be guaranteed
+// server-side: the @ modifier anywhere and default-resolution subqueries
+// (their resolution is a server runtime setting the transpiler cannot see);
+// steps or ranges that are not whole seconds (the grid functions take
+// whole-second parameters); grouping by or matching on __name__ in hybrid
+// plans (the synthetic name would leak into results); name-keeping units —
+// bare/comparison instant selectors and last_over_time keep their real
+// __name__ (keepsName), which substitution would replace, so they transpile
+// only as full plans; and every function outside the allowlist (changes,
+// resets, quantile_over_time, absent, native-histogram functions, ...).
+//
+// Units inside a fixed-resolution subquery evaluate on the subquery's own
+// grid instead of the query grid: epoch-aligned multiples of the resolution
+// strictly after outerStart - offset - range, ending at outer end - offset —
+// the exact derivation the engine uses, because a grid shifted by one step
+// changes which samples every window sees.
+//
+// # From one unit to one statement
+//
+// buildUnitSQL renders each unit as a single statement. For
+// sum by (pod) (rate(m{job="api"}[5m])) the skeleton is:
+//
+//	SELECT gkey, sumForEach(grid) AS grid FROM (
+//	    SELECT series.gkey AS gkey,
+//	           timeSeriesRateToGrid(<start>, <end>, <step>, <range>)(fromUnixTimestamp64Milli(unix_milli), value) AS grid
+//	    FROM signoz_metrics.distributed_samples_v4 AS points
+//	    INNER JOIN (
+//	        SELECT fingerprint, <group key expr> AS gkey
+//	        FROM signoz_metrics.time_series_v4
+//	        WHERE <series predicates>
+//	        GROUP BY fingerprint, gkey
+//	    ) AS series ON points.fingerprint = series.fingerprint
+//	    WHERE metric_name = ? AND temporality IN ['Cumulative', 'Unspecified']
+//	      AND points.fingerprint IN (<matched fingerprints>)
+//	      AND unix_milli > <start - range> AND unix_milli <= <end>
+//	      AND bitAnd(flags, 1) = 0
+//	    GROUP BY points.fingerprint, series.gkey
+//	) GROUP BY gkey
+//	SETTINGS allow_experimental_ts_to_grid_aggregate_function = 1
+//
+// Reading it inside out:
+//
+// The time window is the selector's semantics verbatim: strict > on the
+// lower bound and <= on the upper is the left-open (t - w, t] rule, with the
+// whole window shifted by the offset. bitAnd(flags, 1) = 0 drops stale
+// markers, which PromQL excludes from range vectors.
+//
+// The inner GROUP BY computes one grid array per series.
+// timeSeriesRateToGrid(start, end, step, range) is a parametric aggregate:
+// fed (timestamp, value) pairs it produces Array(Nullable(Float64)) with one
+// slot per grid point. Correct because it implements the engine's
+// extrapolatedRate decision for decision — counter resets, the zero-point
+// clamp, the extrapolation thresholds, the >= 2 samples rule, the left-open
+// window — verified by feeding identical samples to both and comparing
+// slot for slot: the only difference ever observed is the last bit
+// (ClickHouse's C++ and Go round the same formula differently), which is
+// the floating-point floor, not a semantic gap. irate/delta/idelta map to
+// their own timeSeries*ToGrid functions with the same verification;
+// increase has no function of its own and is emitted as
+// arrayMap(x -> x * <range seconds>, <rate expr>), exact by definition —
+// extrapolatedRate computes the same extrapolated delta for both and
+// divides by the range only when isRate, so multiplying it back is the
+// identity, not an approximation. The grid parameters are rendered as
+// literals, not bound args — they are aggregate-function parameters — and
+// the experimental gate rides as a SETTINGS clause on the statement itself
+// so telemetrystore hooks cannot clobber it.
+//
+// The join annotates each series with its group key: toJSONString of the
+// sorted [label, value] pairs the unit projects, extracted from the stored
+// labels JSON. by keeps the listed labels, without excludes them plus
+// __name__, no aggregation keeps everything minus __name__ unless the unit
+// keeps its name — the engine's name-dropping rules. Correct as a grouping
+// key because the pairs are sorted and empty values are filtered: key
+// equality is then exactly label-set equality on the projection —
+// Prometheus treats an empty label value as the label being absent, and
+// stored attribute JSON can carry empties that must not split groups — and
+// the same canonical string parses back into the output label set
+// (labelsFromGroupKey).
+//
+// The outer GROUP BY is the spatial aggregation: sum/min/max/avg/count
+// by/without become the -ForEach combinators. Element-wise aggregation over
+// grid arrays is the engine's per-t_i aggregation, because slot i of every
+// input array refers to the same t_i; the combinators skip NULLs, which is
+// the engine aggregating only the series present at t_i, and an index where
+// every series is absent stays NULL. Two edges need explicit handling:
+// countForEach wraps in a mapping of 0 back to NULL, because a count over
+// an all-absent index is an absent point, not 0; and a unit without
+// aggregation still passes through maxForEach — the identity for the common
+// one-fingerprint group, and a deterministic NULL-skipping merge when a
+// regex __name__ selector collapses distinct metrics onto one projected
+// label set. One caveat is inherent: summation order over series differs
+// from the engine's, so spatial aggregates can differ in the last ULP —
+// float addition is not associative; no ordering reproduces the engine's
+// bit-exactly from inside a GROUP BY.
+//
+// # Instant selectors: staleness needs two aggregates
+//
+// unitInstant uses window = lookback and must reproduce the shadowing rule:
+// the point is absent when the latest in-window sample is a stale marker.
+// timeSeriesLastToGrid alone cannot express that — skipping stale rows in
+// WHERE would resurrect the older real sample the marker was written to
+// bury. So stale rows stay in the scan for this kind only, and the grid
+// expression compares three aggregates per slot:
+//
+//	arrayMap((tall, tok, vok) -> if(tall IS NULL OR tok IS NULL OR tall != tok, NULL, vok),
+//	         timeSeriesLastToGrid(...)(ts, toFloat64(unix_milli)),                           -- last sample overall
+//	         timeSeriesLastToGridIf(...)(ts, toFloat64(unix_milli), bitAnd(flags, 1) = 0),   -- last non-stale, its timestamp
+//	         timeSeriesLastToGridIf(...)(ts, value, bitAnd(flags, 1) = 0))                   -- last non-stale, its value
+//
+// Correct by cases on a slot's window. No samples at all: both timestamp
+// aggregates are NULL, the slot is NULL — absent, as the engine says. Latest
+// sample non-stale: it is the latest overall and the latest non-stale, the
+// timestamps agree, the slot takes its value — the engine's pick. Latest
+// sample stale: the last-overall timestamp is the marker's, the
+// last-non-stale timestamp is older (or NULL when only markers are in
+// window), they disagree, the slot is NULL — the marker shadows, exactly
+// the engine's rule. Timestamps are unique per series (ingest dedups), so
+// timestamp equality identifies "the same sample" without ambiguity. The
+// -If combinator's applicability to these experimental aggregates was
+// probed before being trusted, not assumed.
+//
+// # Windowed *_over_time: fan-out instead of a grid function
+//
+// avg/min/max/sum/count _over_time aggregate every raw sample in the window,
+// and no timeSeries*ToGrid function computes them. (last_over_time is the
+// exception: the last sample of a range vector — stale markers excluded from
+// range vectors by PromQL, excluded here in WHERE — is exactly
+// timeSeriesLastToGrid.) Instead, each sample is fanned out to every grid
+// index whose window contains it:
+//
+//	ARRAY JOIN range(toUInt64(greatest(0, intDiv(unix_milli - <start> + <step> - 1, <step>))),
+//	                 toUInt64(least(<lastIdx>, intDiv(unix_milli + <range> - 1 - <start>, <step>)) + 1)) AS k
+//
+// Correct because the bounds solve the window condition for k. A sample at
+// ts contributes to slot k iff t_k - range < ts <= t_k. The right side
+// gives t_k >= ts, so the first index is ceil((ts - start)/step) — a sample
+// at exactly t_k belongs to k, the window is right-closed. The left side
+// gives t_k < ts + range, and with millisecond-integer timestamps that is
+// t_k <= ts + range - 1, so the last index is
+// floor((ts + range - 1 - start)/step) — a sample at exactly t_k - range is
+// excluded, the window is left-open. Clamped to the grid, the fan-out
+// therefore lands each sample in exactly the slots whose windows contain
+// it, and GROUP BY (fingerprint, k) with the plain aggregate (avg(value),
+// min(value), ...) computes per slot over precisely the engine's sample
+// multiset — the same numbers, since avg/min/max/sum/count are
+// order-insensitive on a given multiset (sum/avg up to summation order, the
+// float caveat above). A second level assembles the positional array with
+// groupArray + indexOf, mapping missing indices to NULL — groupArrayInsertAt
+// would coerce NULL defaults to 0, which is a value, not absence. The
+// group-key join happens at the initiator here, over rows already reduced
+// to per-(series, index); see the sharding section for why that costs
+// nothing.
+//
+// # Scalar ops, full plans, hybrid plans
+//
+// The scalar-op pipeline applies in Go to the returned arrays
+// (applyScalarOps), slot by slot: arithmetic operators compute, comparisons
+// filter (the slot keeps the vector-side value or becomes NULL) or return
+// 0/1 under bool. Correct trivially: it is the same float64 operation the
+// engine would apply to the same slot value, in the same operator order the
+// AST dictates — running it in Go instead of another SQL layer changes
+// where, not what.
+//
+// A full plan's arrays map straight to the result matrix. A hybrid plan
+// materializes each unit's arrays as synthetic series under its
+// __signoz_transpiled_N__ name and evaluates the rewritten expression over
+// a storage that serves synthetic names from memory and everything else
+// live. Substitution is sound because a unit's output is a plain instant
+// vector to the engine — same values at same timestamps under a different
+// name, and the name cannot matter: plans that group by or match on
+// __name__ were refused at classification, and name-keeping units are never
+// substituted. One subtlety makes it exact: stale markers are written at
+// absent grid points, because the engine's lookback would otherwise
+// resurrect a point from up to lookback earlier — the marker encodes
+// "absent here" the way the engine itself encodes it. Units evaluate
+// concurrently; each is one series lookup plus one grid statement. A step
+// of 0 is an instant query: a single evaluation at end.
+//
+// # Series lookup
+//
+// Both paths resolve matchers the same way, once per selector
+// (selectSeries): __name__ matchers translate to the metric_name column —
+// all four matcher types; the v1 client silently returned nothing for regex
+// metric names — and every other matcher to a JSONExtractString condition on
+// the labels column (applySeriesConditions). Regexes are anchored before
+// they reach match(): PromQL matchers match the whole value, ClickHouse
+// match() searches for a substring, and without anchoring =~"api" would
+// also select "x-api-y". An equality matcher against "" matches series
+// without the label, mirroring PromQL, because JSONExtractString returns ""
+// for missing keys. The series tables hold one row per (fingerprint, bucket)
+// at 1h/6h/1d/1w granularities; timeSeriesTableFor picks the table whose
+// bucket fits the window and rounds the window start down to the bucket
+// boundary. The resulting label sets drop what v1 leaked into results: the
+// synthetic fingerprint label (it would take part in without() grouping and
+// vector matching) and empty-valued labels. MaxFetchedSeries fails the
+// lookup with a typed invalid-input error past the ceiling — v1's behavior
+// for an oversized selector was to buffer everything and OOM, and a 4xx the
+// user can narrow beats a dead process serving nobody.
+//
+// # The engine path
+//
+// Queries that do not transpile run in the stock engine over this package's
+// storage.Querier, which is still not the v1 path. Samples are fetched per
+// selector using the engine's per-selector hints, not the query-wide union
+// window, so foo / foo offset 1d reads two narrow windows instead of the
+// widest one twice. Instant selectors of subquery-free queries fetch only
+// the last sample per step bucket (lastSamplePerStep): buckets anchor at the
+// selector's first evaluation timestamp — recovered from the hints as
+// hints.Start + lookback - 1ms, the inverse of how the engine derives
+// hints.Start — so bucket boundaries coincide with evaluation timestamps and
+// a non-final sample of a bucket can never be the latest sample in
+// (t - lookback, t] for any grid t. Real timestamps are preserved, so the
+// engine's own lookback and staleness handling stay exact. Range selectors
+// always fetch raw — every sample feeds the range function — and the
+// subquery-free proof travels in the context as prometheus.QueryTraits,
+// because subquery selectors evaluate at the subquery's step while the
+// hints carry the top-level step. Row assembly counts rows against
+// MaxFetchedSamples while scanning, keeps the first of consecutive equal
+// timestamps, maps stale flags to the engine's StaleNaN, and merges series
+// with identical label sets (sortAndMerge) — the engine assumes storages
+// never emit duplicates. A {job="rawsql", query="..."} selector bypasses all
+// of this and runs the query matcher's value verbatim.
+//
+// # Sharding
+//
+// samples_v4 and time_series_v4 (and all their rollups) shard on the same
+// key — cityHash64(env, temporality, metric_name, fingerprint) — so a
+// series' samples and catalog rows live on the same shard. The transpiled
+// statement above exploits that: the distributed samples table at the
+// top-level FROM makes ClickHouse rewrite the whole inner query per shard,
+// where the join against the shard-local series table and the per-series
+// grid aggregation run next to the data; the initiator only merges
+// aggregate states and applies the spatial -ForEach step. Same layout as
+// the telemetrymetrics statement builder. Fingerprint filters follow suit:
+// matched sets inline as sorted literals up to inlineFingerprintsLimit
+// (literals engage the samples primary key; sorting keeps statements
+// deterministic), beyond it the group-key join alone restricts — a
+// semi-join on the same predicates would only rescan the series table —
+// except the windowed *_over_time fan-out, which has no join and keeps a
+// shard-local IN subquery rather than expand every series of the metric.
+// The engine path's over-limit filter is the same shard-local subquery, not
+// a GLOBAL broadcast of the matched set. The temporality filter on every
+// samples statement is a semantic no-op — the matched fingerprints already
+// come from those temporalities — that engages the leading samples
+// primary-key column. Delta-temporality series stay invisible to PromQL
+// here exactly as they are in v1: the rollout gate is parity with v1, and
+// making Delta visible is its own change with its own semantics to design —
+// a Delta stream fed to rate() as-if-cumulative would be wrong, not just
+// new.
+//
+// # Observability
+//
+// Every statement carries a log_comment with
+// code.namespace=clickhouse-prometheus-v2 and code.function.name naming the
+// call site (selectSeries, selectSamples, transpiledUnit, LabelValues,
+// LabelNames), so this provider's work is attributable in system.query_log
+// without guessing from query text.
+package clickhouseprometheusv2

pkg/prometheus/clickhouseprometheusv2/last_sample_per_step_test.go

@@ -0,0 +1,191 @@
+package clickhouseprometheusv2
+
+import (
+	"fmt"
+	"math/rand"
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// The lastSamplePerStep correctness argument, executed: for instant selectors, keeping
+// only the last sample of every step bucket (bucket 0 = (start, firstEval],
+// bucket i = (firstEval+(i-1)·step, firstEval+i·step]) yields exactly the
+// same instant-vector selections as the raw samples, for every evaluation
+// timestamp on the grid. The engine picks the latest sample in
+// (t-lookback, t] per evaluation timestamp t and treats a stale marker as
+// absent; both behaviors are emulated here directly.
+
+type tsample struct {
+	ts    int64
+	value float64
+	stale bool
+}
+
+// engineSelect emulates the engine's instant-selector resolution at
+// evaluation timestamp t over samples ordered by timestamp: the latest sample
+// in (t-lookback, t], absent when none or when it is a stale marker.
+func engineSelect(samples []tsample, t, lookbackMs int64) (tsample, bool) {
+	var picked tsample
+	found := false
+	for _, s := range samples {
+		if s.ts > t-lookbackMs && s.ts <= t {
+			picked = s
+			found = true
+		}
+	}
+	if !found || picked.stale {
+		return tsample{}, false
+	}
+	return picked, true
+}
+
+// lastPerStep emulates the last-sample-per-step samples query: group samples into buckets and
+// keep only the last sample of each (ties keep either; ClickHouse argMax over
+// equal keys is unspecified, so generated timestamps are unique).
+func lastPerStep(samples []tsample, firstEvalMs, stepMs int64) []tsample {
+	last := make(map[int64]tsample)
+	for _, s := range samples {
+		var bucket int64
+		if stepMs > 0 && s.ts > firstEvalMs {
+			bucket = (s.ts-firstEvalMs-1)/stepMs + 1
+		}
+		if cur, ok := last[bucket]; !ok || s.ts > cur.ts {
+			last[bucket] = s
+		}
+	}
+	out := make([]tsample, 0, len(last))
+	for _, s := range last {
+		out = append(out, s)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].ts < out[j].ts })
+	return out
+}
+
+func TestLastSamplePerStepEquivalence(t *testing.T) {
+	rng := rand.New(rand.NewSource(42))
+
+	for caseIdx := 0; caseIdx < 2000; caseIdx++ {
+		// Random query shape. Units are milliseconds but kept small so bucket
+		// boundaries are hit often.
+		stepMs := []int64{1, 2, 5, 7, 30, 60}[rng.Intn(6)]
+		lookbackMs := []int64{1, 3, 5, 10, 45}[rng.Intn(5)]
+		queryStart := int64(1000)
+		numSteps := rng.Int63n(20)
+		queryEnd := queryStart + numSteps*stepMs + rng.Int63n(stepMs) // grid may not divide the range
+
+		// Engine-derived selector window for instant selectors:
+		// hints.Start = firstEval - (lookback - 1), hints.End = queryEnd.
+		hintsStart := queryStart - (lookbackMs - 1)
+		hintsEnd := queryEnd
+		firstEval := hintsStart + lookbackMs - 1
+		require.Equal(t, queryStart, firstEval)
+
+		// Random samples inside the fetch window [hints.Start, hints.End],
+		// with unique timestamps and occasional stale markers. The sample
+		// count is capped by the window size: timestamps are unique.
+		windowSize := hintsEnd - hintsStart + 1
+		numSamples := rng.Int63n(40)
+		if numSamples > windowSize {
+			numSamples = windowSize
+		}
+		seen := make(map[int64]bool)
+		var samples []tsample
+		for int64(len(samples)) < numSamples {
+			ts := hintsStart + rng.Int63n(windowSize)
+			if seen[ts] {
+				continue
+			}
+			seen[ts] = true
+			samples = append(samples, tsample{ts: ts, value: rng.Float64(), stale: rng.Intn(8) == 0})
+		}
+		sort.Slice(samples, func(i, j int) bool { return samples[i].ts < samples[j].ts })
+
+		reduced := lastPerStep(samples, firstEval, stepMs)
+
+		desc := fmt.Sprintf("case=%d step=%d lookback=%d start=%d end=%d samples=%d",
+			caseIdx, stepMs, lookbackMs, queryStart, queryEnd, len(samples))
+
+		for evalTs := queryStart; evalTs <= queryEnd; evalTs += stepMs {
+			rawPick, rawOK := engineSelect(samples, evalTs, lookbackMs)
+			reducedPick, reducedOK := engineSelect(reduced, evalTs, lookbackMs)
+
+			require.Equal(t, rawOK, reducedOK, "%s eval=%d presence mismatch", desc, evalTs)
+			if rawOK {
+				require.Equal(t, rawPick, reducedPick, "%s eval=%d sample mismatch", desc, evalTs)
+			}
+		}
+	}
+}
+
+// Instant queries (step 0) evaluate once at firstEval == hints.End; lastSamplePerStep
+// collapses to a single bucket over the whole window.
+func TestLastSamplePerStepEquivalenceInstantQuery(t *testing.T) {
+	rng := rand.New(rand.NewSource(7))
+
+	for caseIdx := 0; caseIdx < 500; caseIdx++ {
+		lookbackMs := []int64{1, 3, 5, 10, 45}[rng.Intn(5)]
+		evalTs := int64(1000)
+		hintsStart := evalTs - (lookbackMs - 1)
+		hintsEnd := evalTs
+		firstEval := hintsStart + lookbackMs - 1
+		require.Equal(t, evalTs, firstEval)
+
+		windowSize := hintsEnd - hintsStart + 1
+		numSamples := rng.Int63n(10)
+		if numSamples > windowSize {
+			numSamples = windowSize
+		}
+		seen := make(map[int64]bool)
+		var samples []tsample
+		for int64(len(samples)) < numSamples {
+			ts := hintsStart + rng.Int63n(windowSize)
+			if seen[ts] {
+				continue
+			}
+			seen[ts] = true
+			samples = append(samples, tsample{ts: ts, value: rng.Float64(), stale: rng.Intn(4) == 0})
+		}
+		sort.Slice(samples, func(i, j int) bool { return samples[i].ts < samples[j].ts })
+
+		reduced := lastPerStep(samples, firstEval, 0)
+		require.LessOrEqual(t, len(reduced), 1, "instant reduction must keep at most one sample")
+
+		rawPick, rawOK := engineSelect(samples, evalTs, lookbackMs)
+		reducedPick, reducedOK := engineSelect(reduced, evalTs, lookbackMs)
+		require.Equal(t, rawOK, reducedOK, "case=%d presence mismatch", caseIdx)
+		if rawOK {
+			require.Equal(t, rawPick, reducedPick, "case=%d sample mismatch", caseIdx)
+		}
+	}
+}
+
+// A stale marker that is the latest sample of its bucket must shadow older
+// samples: the engine sees the marker and reports the series absent, exactly
+// as with raw samples. Pre-filtering stale rows would instead resurrect the
+// older sample.
+func TestLastSamplePerStepKeepsStaleShadowing(t *testing.T) {
+	lookbackMs := int64(10)
+	stepMs := int64(5)
+	queryStart := int64(1000)
+
+	samples := []tsample{
+		{ts: 998, value: 1.0},                // bucket 0
+		{ts: 999, stale: true},               // bucket 0: marker shadows 998
+		{ts: 1003, value: 2.0},               // bucket 1
+		{ts: 1004, stale: true},              // bucket 1: marker shadows 1003
+		{ts: 1008, value: 3.0, stale: false}, // bucket 2
+	}
+	firstEval := queryStart
+	reduced := lastPerStep(samples, firstEval, stepMs)
+
+	for evalTs := queryStart; evalTs <= queryStart+2*stepMs; evalTs += stepMs {
+		rawPick, rawOK := engineSelect(samples, evalTs, lookbackMs)
+		reducedPick, reducedOK := engineSelect(reduced, evalTs, lookbackMs)
+		require.Equal(t, rawOK, reducedOK, "eval=%d", evalTs)
+		if rawOK {
+			require.Equal(t, rawPick, reducedPick, "eval=%d", evalTs)
+		}
+	}
+}

pkg/prometheus/clickhouseprometheusv2/provider.go

@@ -0,0 +1,84 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/storage"
+)
+
+// Provider ties the package together: its own engine and parser, the
+// ClickHouse client behind the native storage.Querier, and the transpiler
+// executor. See the package documentation for what runs where and why. It is
+// exported as a concrete type — pkg/querier holds it directly for shadow
+// comparison and pinned serving, and an interface with a single
+// implementation would only hide that dependency.
+type Provider struct {
+	settings factory.ScopedProviderSettings
+	engine   *prometheus.Engine
+	parser   prometheus.Parser
+	client   *client
+	executor *executor
+}
+
+var (
+	_ prometheus.Prometheus        = (*Provider)(nil)
+	_ prometheus.StatementCapturer = (*Provider)(nil)
+)
+
+func NewFactory(telemetryStore telemetrystore.TelemetryStore) factory.ProviderFactory[prometheus.Prometheus, prometheus.Config] {
+	return factory.NewProviderFactory(factory.MustNewName("clickhousev2"), func(ctx context.Context, providerSettings factory.ProviderSettings, config prometheus.Config) (prometheus.Prometheus, error) {
+		return New(ctx, providerSettings, config, telemetryStore)
+	})
+}
+
+func New(_ context.Context, providerSettings factory.ProviderSettings, config prometheus.Config, telemetryStore telemetrystore.TelemetryStore) (*Provider, error) {
+	settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2")
+
+	engine := prometheus.NewEngine(settings.Logger(), config)
+	parser := prometheus.NewParser()
+	client := newClient(settings, telemetryStore, config)
+
+	return &Provider{
+		settings: settings,
+		engine:   engine,
+		parser:   parser,
+		client:   client,
+		executor: &executor{client: client, engine: engine, parser: parser},
+	}, nil
+}
+
+// TryExecuteRange evaluates transpilable query shapes directly in ClickHouse
+// (see transpiler.go). ok=false means the shape is not transpilable and the
+// caller should evaluate through Engine over Storage instead.
+func (p *Provider) TryExecuteRange(ctx context.Context, query string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error) {
+	return p.executor.TryExecuteRange(ctx, query, start, end, step)
+}
+
+func (p *Provider) Engine() *prometheus.Engine {
+	return p.engine
+}
+
+func (p *Provider) Parser() prometheus.Parser {
+	return p.parser
+}
+
+func (p *Provider) Storage() storage.Queryable {
+	return p
+}
+
+func (p *Provider) Querier(mint, maxt int64) (storage.Querier, error) {
+	return &querier{mint: mint, maxt: maxt, client: p.client}, nil
+}
+
+// CapturingStorage implements prometheus.StatementCapturer: a storage that
+// records each selector's SQL without executing it, for the preview path.
+// A fresh recorder per call keeps concurrent dry-runs isolated.
+func (p *Provider) CapturingStorage() (storage.Queryable, prometheus.StatementRecorder) {
+	recorder := &statementRecorder{}
+	return &captureQueryable{client: p.client, recorder: recorder}, recorder
+}

pkg/prometheus/clickhouseprometheusv2/querier.go

@@ -0,0 +1,233 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"sort"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/huandu/go-sqlbuilder"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/util/annotations"
+)
+
+// defaultLookbackDelta mirrors promql's default when the config leaves the
+// lookback unset; the engine and the storage must agree on it for
+// last-sample-per-step bucket anchoring.
+const defaultLookbackDelta = 5 * time.Minute
+
+// querier is a native storage.Querier over ClickHouse. Unlike v1 it does not
+// round-trip through the remote-read protobuf machinery: Select builds SQL
+// directly from the matchers and hints, and the result set is assembled once
+// into compact series.
+type querier struct {
+	mint, maxt int64
+	client     *client
+}
+
+var _ storage.Querier = (*querier)(nil)
+
+func (q *querier) Select(ctx context.Context, sortSeries bool, hints *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
+	if rawQuery, ok := rawSQLQuery(matchers); ok {
+		_, end := q.window(hints)
+		list, err := q.client.queryRaw(ctx, rawQuery, end)
+		if err != nil {
+			return storage.ErrSeriesSet(err)
+		}
+		if sortSeries {
+			sort.Slice(list, func(i, j int) bool { return labels.Compare(list[i].lset, list[j].lset) < 0 })
+		}
+		return newSeriesSet(list)
+	}
+
+	start, end := q.window(hints)
+
+	seriesQuery, seriesArgs, err := buildSeriesQuery(start, end, matchers)
+	if err != nil {
+		return storage.ErrSeriesSet(err)
+	}
+	lookup, err := q.client.selectSeries(ctx, seriesQuery, seriesArgs)
+	if err != nil {
+		return storage.ErrSeriesSet(err)
+	}
+	if len(lookup.fingerprints) == 0 {
+		return storage.EmptySeriesSet()
+	}
+
+	list, err := q.fetchSamples(ctx, start, end, matchers, lookup, q.lastSamplePerStepFor(ctx, hints))
+	if err != nil {
+		return storage.ErrSeriesSet(err)
+	}
+
+	// Sorting doubles as duplicate-label-set detection, which the engine
+	// depends on storages never emitting; the cost is on series count, not
+	// samples.
+	list = sortAndMerge(list)
+	return newSeriesSet(list)
+}
+
+// LabelValues returns the values of a label across series matching the
+// matchers within the querier window.
+func (q *querier) LabelValues(ctx context.Context, name string, hints *storage.LabelHints, matchers ...*labels.Matcher) ([]string, annotations.Annotations, error) {
+	sb := sqlbuilder.NewSelectBuilder()
+	if name == metricNameLabel {
+		sb.Select("DISTINCT metric_name AS value")
+	} else {
+		sb.Select(fmt.Sprintf("DISTINCT JSONExtractString(labels, %s) AS value", sb.Var(name)))
+	}
+	adjustedStart, table := timeSeriesTableFor(q.mint, q.maxt)
+	sb.From(fmt.Sprintf("%s.%s", databaseName, table))
+	if err := applySeriesConditions(sb, adjustedStart, q.maxt, matchers); err != nil {
+		return nil, nil, err
+	}
+	sb.Where("value != ''")
+	if hints != nil && hints.Limit > 0 {
+		sb.Limit(hints.Limit)
+	}
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	values, err := q.selectStrings(ctx, "LabelValues", query, args)
+	if err != nil {
+		return nil, nil, err
+	}
+	slices.Sort(values)
+	return values, nil, nil
+}
+
+// LabelNames returns the label names present on series matching the matchers
+// within the querier window.
+func (q *querier) LabelNames(ctx context.Context, hints *storage.LabelHints, matchers ...*labels.Matcher) ([]string, annotations.Annotations, error) {
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select("DISTINCT arrayJoin(JSONExtractKeys(labels)) AS name")
+	adjustedStart, table := timeSeriesTableFor(q.mint, q.maxt)
+	sb.From(fmt.Sprintf("%s.%s", databaseName, table))
+	if err := applySeriesConditions(sb, adjustedStart, q.maxt, matchers); err != nil {
+		return nil, nil, err
+	}
+	if hints != nil && hints.Limit > 0 {
+		sb.Limit(hints.Limit)
+	}
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	names, err := q.selectStrings(ctx, "LabelNames", query, args)
+	if err != nil {
+		return nil, nil, err
+	}
+	slices.Sort(names)
+	return names, nil, nil
+}
+
+func (q *querier) Close() error {
+	return nil
+}
+
+// window returns the per-selector fetch window. The engine sends per-selector
+// bounds in the hints (already adjusted for offset, @, range and lookback);
+// they are always at least as tight as the querier-level mint/maxt, which
+// span the union of all selectors in the query.
+func (q *querier) window(hints *storage.SelectHints) (int64, int64) {
+	if hints != nil && hints.Start != 0 && hints.End != 0 && hints.Start <= hints.End {
+		return hints.Start, hints.End
+	}
+	return q.mint, q.maxt
+}
+
+// lastSamplePerStepFor decides whether the fetch can keep only the last
+// sample per step bucket, and computes the bucket parameters. Requirements:
+//   - the call site attached QueryTraits proving the query has no subquery
+//     (subquery selectors evaluate at the subquery's own step, but hints
+//     carry the top-level step);
+//   - the selector is an instant selector (hints.Range == 0); range selectors
+//     need every raw sample in the window;
+//   - per-selector hints are present.
+//
+// The engine derives hints.Start for instant selectors as
+// firstEval - (lookback - 1ms), so the first evaluation timestamp is
+// recovered as hints.Start + lookback - 1ms. Bucket boundaries then coincide
+// with evaluation timestamps, which is what makes keeping only the last
+// sample per bucket lossless.
+func (q *querier) lastSamplePerStepFor(ctx context.Context, hints *storage.SelectHints) *lastSamplePerStep {
+	if hints == nil || hints.Range != 0 || hints.Start <= 0 {
+		return nil
+	}
+	traits, ok := prometheus.QueryTraitsFromContext(ctx)
+	if !ok || !traits.SubqueryFree {
+		return nil
+	}
+	firstEval := hints.Start + q.client.lookbackMs - 1
+	if firstEval > hints.End {
+		// Defensive: never anchor a bucket past the window.
+		firstEval = hints.End
+	}
+	return &lastSamplePerStep{firstEvalMs: firstEval, stepMs: hints.Step}
+}
+
+// fetchSamples runs the samples query for the matched series. Small sets
+// inline the fingerprints as sorted uint64 literals — literals engage the
+// samples primary key, and sorting keeps the statement deterministic for
+// logging and tests. Larger sets re-run the series predicates as a
+// shard-local IN subquery instead: inlining hundreds of thousands of
+// literals makes the statement itself the bottleneck, while the subquery is
+// a cheap primary-key scan on each shard's own series table (see
+// localTimeSeriesTable for why that is complete).
+func (q *querier) fetchSamples(ctx context.Context, start, end int64, matchers []*labels.Matcher, lookup *seriesLookup, lastPerStep *lastSamplePerStep) ([]*series, error) {
+	var fingerprints []uint64
+	if len(lookup.fingerprints) <= inlineFingerprintsLimit {
+		fingerprints = make([]uint64, 0, len(lookup.fingerprints))
+		for fp := range lookup.fingerprints {
+			fingerprints = append(fingerprints, fp)
+		}
+		slices.Sort(fingerprints)
+	}
+	query, args, err := buildSamplesQuery(start, end, lookup.metricNames, fingerprints, matchers, lastPerStep)
+	if err != nil {
+		return nil, err
+	}
+	return q.client.selectSamples(ctx, query, args, lookup)
+}
+
+func (q *querier) selectStrings(ctx context.Context, fn, query string, args []any) ([]string, error) {
+	ctx = q.client.withContext(ctx, fn)
+	rows, err := q.client.telemetryStore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []string
+	var v string
+	for rows.Next() {
+		if err := rows.Scan(&v); err != nil {
+			return nil, err
+		}
+		out = append(out, v)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// rawSQLQuery detects the {job="rawsql", query="..."} escape hatch.
+func rawSQLQuery(matchers []*labels.Matcher) (string, bool) {
+	if len(matchers) != 2 {
+		return "", false
+	}
+	var hasJob bool
+	var query string
+	for _, m := range matchers {
+		if m.Type == labels.MatchEqual && m.Name == "job" && m.Value == "rawsql" {
+			hasJob = true
+		}
+		if m.Type == labels.MatchEqual && m.Name == "query" {
+			query = m.Value
+		}
+	}
+	if hasJob && query != "" {
+		return query, true
+	}
+	return "", false
+}

pkg/prometheus/clickhouseprometheusv2/querier_test.go

@@ -0,0 +1,221 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	cmock "github.com/SigNoz/clickhouse-go-mock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+)
+
+var (
+	seriesCols = []cmock.ColumnType{
+		{Name: "fingerprint", Type: "UInt64"},
+		{Name: "labels", Type: "String"},
+	}
+	samplesCols = []cmock.ColumnType{
+		{Name: "fingerprint", Type: "UInt64"},
+		{Name: "unix_milli", Type: "Int64"},
+		{Name: "value", Type: "Float64"},
+		{Name: "flags", Type: "UInt32"},
+	}
+)
+
+func newTestClient(t *testing.T, cfg prometheus.ClickhouseV2Config) (*client, *telemetrystoretest.Provider) {
+	t.Helper()
+	store := telemetrystoretest.New(telemetrystore.Config{Provider: "clickhouse"}, sqlmock.QueryMatcherRegexp)
+	settings := factory.NewScopedProviderSettings(instrumentationtest.New().ToProviderSettings(), "clickhouseprometheusv2_test")
+	promCfg := prometheus.Config{ClickhouseV2: cfg}
+	return newClient(settings, store, promCfg), store
+}
+
+func testMatchers(t *testing.T) []*labels.Matcher {
+	t.Helper()
+	return []*labels.Matcher{
+		mustMatcher(t, labels.MatchEqual, "__name__", "cpu_usage"),
+		mustMatcher(t, labels.MatchEqual, "job", "api"),
+	}
+}
+
+func TestQuerierSelectRawPath(t *testing.T) {
+	c, store := newTestClient(t, prometheus.ClickhouseV2Config{})
+	q := &querier{mint: 1000, maxt: 2000, client: c}
+
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("cpu_usage", int64(0), int64(2000), "job", "api").WillReturnRows(cmock.NewRows(seriesCols, [][]any{
+		{uint64(42), `{"__name__":"cpu_usage","job":"api","instance":"a"}`},
+		{uint64(7), `{"__name__":"cpu_usage","job":"api","instance":"b"}`},
+	}))
+	// Inline fingerprints (sorted), raw samples: no traits in ctx -> no
+	// last-sample-per-step reduction.
+	store.Mock().ExpectQuery("SELECT fingerprint, unix_milli, value, flags FROM signoz_metrics.distributed_samples_v4 WHERE metric_name = \\? AND temporality IN \\['Cumulative', 'Unspecified'\\] AND fingerprint IN \\(7, 42\\)").
+		WithArgs("cpu_usage", int64(1000), int64(2000)).
+		WillReturnRows(cmock.NewRows(samplesCols, [][]any{
+			{uint64(7), int64(1100), 1.5, uint32(0)},
+			{uint64(7), int64(1200), 2.5, uint32(0)},
+			{uint64(42), int64(1100), 3.5, uint32(1)}, // stale marker
+		}))
+
+	hints := &storage.SelectHints{Start: 1000, End: 2000, Step: 60_000}
+	set := q.Select(context.Background(), false, hints, testMatchers(t)...)
+
+	var got []*series
+	for set.Next() {
+		got = append(got, set.At().(*series))
+	}
+	require.NoError(t, set.Err())
+	require.Len(t, got, 2)
+
+	// Sorted by labels: instance=a (fp 42) before instance=b (fp 7).
+	assert.Equal(t, "a", got[0].lset.Get("instance"))
+	require.Len(t, got[0].ts, 1)
+	assert.True(t, got[0].vs[0] != got[0].vs[0], "stale marker must be NaN") //nolint:testifylint
+
+	assert.Equal(t, "b", got[1].lset.Get("instance"))
+	assert.Equal(t, []int64{1100, 1200}, got[1].ts)
+	assert.Equal(t, []float64{1.5, 2.5}, got[1].vs)
+
+	// No fingerprint label injected.
+	assert.Empty(t, got[0].lset.Get("fingerprint"))
+}
+
+// Wrong gating silently corrupts range functions (a rate over reduced
+// samples loses points), so the decision logic is pinned here even though
+// the helper is unexported: the integration suite would catch it too, but
+// with far worse failure locality.
+func TestLastSamplePerStepFor(t *testing.T) {
+	c, _ := newTestClient(t, prometheus.ClickhouseV2Config{})
+	q := &querier{mint: 0, maxt: 2000, client: c}
+	traitsCtx := prometheus.NewContextWithQueryTraits(context.Background(), prometheus.QueryTraits{SubqueryFree: true})
+
+	tests := []struct {
+		name  string
+		ctx   context.Context
+		hints *storage.SelectHints
+		want  *lastSamplePerStep
+	}{
+		{"no traits in context stays raw", context.Background(), &storage.SelectHints{Start: 1000, End: 2000, Step: 60_000}, nil},
+		{"subquery in the query stays raw", prometheus.NewContextWithQueryTraits(context.Background(), prometheus.QueryTraits{SubqueryFree: false}), &storage.SelectHints{Start: 1000, End: 2000, Step: 60_000}, nil},
+		{"range selector stays raw", traitsCtx, &storage.SelectHints{Start: 1000, End: 2000, Step: 60_000, Range: 300_000}, nil},
+		{"instant selector reduces, anchored at first eval", traitsCtx, &storage.SelectHints{Start: 1000, End: 2_000_000, Step: 60_000}, &lastSamplePerStep{firstEvalMs: 1000 + c.lookbackMs - 1, stepMs: 60_000}},
+		{"anchor never passes the window end", traitsCtx, &storage.SelectHints{Start: 1000, End: 2000, Step: 60_000}, &lastSamplePerStep{firstEvalMs: 2000, stepMs: 60_000}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, q.lastSamplePerStepFor(tt.ctx, tt.hints))
+		})
+	}
+}
+
+func TestQuerierSelectSeriesBudget(t *testing.T) {
+	c, store := newTestClient(t, prometheus.ClickhouseV2Config{MaxFetchedSeries: 1})
+	q := &querier{mint: 1000, maxt: 2000, client: c}
+
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("cpu_usage", int64(0), int64(2000), "job", "api").WillReturnRows(cmock.NewRows(seriesCols, [][]any{
+		{uint64(1), `{"__name__":"cpu_usage","instance":"a"}`},
+		{uint64(2), `{"__name__":"cpu_usage","instance":"b"}`},
+	}))
+
+	set := q.Select(context.Background(), false, &storage.SelectHints{Start: 1000, End: 2000}, testMatchers(t)...)
+	assert.False(t, set.Next())
+	require.Error(t, set.Err())
+	assert.True(t, errors.Ast(set.Err(), errors.TypeInvalidInput), "budget error must be typed invalid input, got %v", set.Err())
+}
+
+func TestQuerierSelectSamplesBudget(t *testing.T) {
+	c, store := newTestClient(t, prometheus.ClickhouseV2Config{MaxFetchedSamples: 2})
+	q := &querier{mint: 1000, maxt: 2000, client: c}
+
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("cpu_usage", int64(0), int64(2000)).WillReturnRows(cmock.NewRows(seriesCols, [][]any{
+		{uint64(7), `{"__name__":"cpu_usage"}`},
+	}))
+	store.Mock().ExpectQuery("SELECT fingerprint, unix_milli, value, flags").
+		WithArgs("cpu_usage", int64(1000), int64(2000)).
+		WillReturnRows(cmock.NewRows(samplesCols, [][]any{
+			{uint64(7), int64(1100), 1.0, uint32(0)},
+			{uint64(7), int64(1200), 2.0, uint32(0)},
+			{uint64(7), int64(1300), 3.0, uint32(0)},
+		}))
+
+	set := q.Select(context.Background(), false, &storage.SelectHints{Start: 1000, End: 2000},
+		mustMatcher(t, labels.MatchEqual, "__name__", "cpu_usage"))
+	assert.False(t, set.Next())
+	require.Error(t, set.Err())
+	assert.True(t, errors.Ast(set.Err(), errors.TypeInvalidInput))
+}
+
+func TestQuerierSelectSubqueryFilterOverInlineLimit(t *testing.T) {
+	c, store := newTestClient(t, prometheus.ClickhouseV2Config{})
+	q := &querier{mint: 1000, maxt: 2000, client: c}
+
+	seriesRows := make([][]any, inlineFingerprintsLimit+1)
+	for i := range seriesRows {
+		seriesRows[i] = []any{uint64(i + 1), fmt.Sprintf(`{"__name__":"cpu_usage","instance":"i%d"}`, i)}
+	}
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("cpu_usage", int64(0), int64(2000), "job", "api").WillReturnRows(cmock.NewRows(seriesCols, seriesRows))
+	// The over-limit samples query embeds the semi-join against the
+	// shard-local series table (fingerprint co-locality), not a GLOBAL
+	// broadcast; args follow placeholder order — samples metric name, then
+	// the semi-join's series predicates, then the samples window bounds.
+	store.Mock().ExpectQuery("fingerprint IN \\(SELECT fingerprint FROM signoz_metrics\\.time_series_v4").
+		WithArgs("cpu_usage", "cpu_usage", int64(0), int64(2000), "job", "api", int64(1000), int64(2000)).
+		WillReturnRows(cmock.NewRows(samplesCols, [][]any{}))
+
+	set := q.Select(context.Background(), false, &storage.SelectHints{Start: 1000, End: 2000}, testMatchers(t)...)
+	assert.False(t, set.Next())
+	require.NoError(t, set.Err())
+}
+
+func TestQuerierSelectRawSQLPassthrough(t *testing.T) {
+	c, store := newTestClient(t, prometheus.ClickhouseV2Config{})
+	q := &querier{mint: 1000, maxt: 2000, client: c}
+
+	rawCols := []cmock.ColumnType{
+		{Name: "le", Type: "String"},
+		{Name: "value", Type: "Float64"},
+	}
+	store.Mock().ExpectQuery("SELECT le, avg\\(v\\) AS value FROM t").WillReturnRows(cmock.NewRows(rawCols, [][]any{
+		{"0.5", 12.5},
+	}))
+
+	set := q.Select(context.Background(), false, &storage.SelectHints{Start: 1000, End: 2000},
+		mustMatcher(t, labels.MatchEqual, "job", "rawsql"),
+		mustMatcher(t, labels.MatchEqual, "query", "SELECT le, avg(v) AS value FROM t"),
+	)
+
+	require.True(t, set.Next())
+	s := set.At()
+	assert.Equal(t, "0.5", s.Labels().Get("le"))
+	it := s.Iterator(nil)
+	require.NotNil(t, it)
+	_, v := func() (int64, float64) { it.Next(); return it.At() }()
+	assert.Equal(t, 12.5, v)
+	assert.False(t, set.Next())
+}
+
+func TestCaptureQuerierRecordsWithoutExecuting(t *testing.T) {
+	c, _ := newTestClient(t, prometheus.ClickhouseV2Config{})
+	recorder := &statementRecorder{}
+	cq := &captureQuerier{querier: querier{mint: 1000, maxt: 2000, client: c}, recorder: recorder}
+
+	ctx := prometheus.NewContextWithQueryTraits(context.Background(), prometheus.QueryTraits{SubqueryFree: true})
+	set := cq.Select(ctx, false, &storage.SelectHints{Start: 1000, End: 2000, Step: 60_000}, testMatchers(t)...)
+	assert.False(t, set.Next())
+	require.NoError(t, set.Err())
+
+	statements := recorder.Statements()
+	require.Len(t, statements, 1)
+	assert.Contains(t, statements[0].Query, "IN (SELECT fingerprint FROM signoz_metrics.time_series_v4")
+	assert.Contains(t, statements[0].Query, "argMax(value, unix_milli)")
+}

pkg/prometheus/clickhouseprometheusv2/seriesset.go

@@ -0,0 +1,184 @@
+package clickhouseprometheusv2
+
+import (
+	"sort"
+
+	"github.com/prometheus/prometheus/model/histogram"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/tsdb/chunkenc"
+	"github.com/prometheus/prometheus/util/annotations"
+)
+
+// series is one time series with samples stored as parallel slices, ordered
+// by timestamp. The compact layout avoids per-sample allocations and keeps
+// iteration cache friendly.
+type series struct {
+	lset labels.Labels
+	ts   []int64
+	vs   []float64
+}
+
+var _ storage.Series = (*series)(nil)
+
+func (s *series) Labels() labels.Labels {
+	return s.lset
+}
+
+func (s *series) Iterator(it chunkenc.Iterator) chunkenc.Iterator {
+	if fit, ok := it.(*floatIterator); ok {
+		fit.reset(s)
+		return fit
+	}
+	fit := &floatIterator{}
+	fit.reset(s)
+	return fit
+}
+
+// floatIterator implements chunkenc.Iterator over a series' sample slices.
+type floatIterator struct {
+	s *series
+	i int
+}
+
+var _ chunkenc.Iterator = (*floatIterator)(nil)
+
+func (it *floatIterator) reset(s *series) {
+	it.s = s
+	it.i = -1
+}
+
+func (it *floatIterator) Next() chunkenc.ValueType {
+	it.i++
+	if it.i >= len(it.s.ts) {
+		return chunkenc.ValNone
+	}
+	return chunkenc.ValFloat
+}
+
+func (it *floatIterator) Seek(t int64) chunkenc.ValueType { //nolint:govet // stdmethods flags io.Seeker; this is chunkenc.Iterator's Seek
+	if it.i < 0 {
+		it.i = 0
+	}
+	if it.i >= len(it.s.ts) {
+		return chunkenc.ValNone
+	}
+	// The current position, once valid, must not move backwards.
+	if it.s.ts[it.i] >= t {
+		return chunkenc.ValFloat
+	}
+	it.i += sort.Search(len(it.s.ts)-it.i, func(j int) bool {
+		return it.s.ts[it.i+j] >= t
+	})
+	if it.i >= len(it.s.ts) {
+		return chunkenc.ValNone
+	}
+	return chunkenc.ValFloat
+}
+
+func (it *floatIterator) At() (int64, float64) {
+	return it.s.ts[it.i], it.s.vs[it.i]
+}
+
+func (it *floatIterator) AtHistogram(*histogram.Histogram) (int64, *histogram.Histogram) {
+	return 0, nil
+}
+
+func (it *floatIterator) AtFloatHistogram(*histogram.FloatHistogram) (int64, *histogram.FloatHistogram) {
+	return 0, nil
+}
+
+func (it *floatIterator) AtT() int64 {
+	return it.s.ts[it.i]
+}
+
+// AtST returns the current start timestamp; not tracked by this storage.
+func (it *floatIterator) AtST() int64 {
+	return 0
+}
+
+func (it *floatIterator) Err() error {
+	return nil
+}
+
+// seriesSet iterates a fully materialized, label-sorted list of series.
+type seriesSet struct {
+	series []*series
+	i      int
+}
+
+var _ storage.SeriesSet = (*seriesSet)(nil)
+
+func newSeriesSet(list []*series) *seriesSet {
+	return &seriesSet{series: list, i: -1}
+}
+
+func (s *seriesSet) Next() bool {
+	s.i++
+	return s.i < len(s.series)
+}
+
+func (s *seriesSet) At() storage.Series {
+	return s.series[s.i]
+}
+
+func (s *seriesSet) Err() error {
+	return nil
+}
+
+func (s *seriesSet) Warnings() annotations.Annotations {
+	return nil
+}
+
+// sortAndMerge orders series by label set and merges series whose label sets
+// are identical. Distinct fingerprints can carry identical label sets (e.g.
+// series differing only in a non-label dimension); Prometheus storages never
+// expose duplicate label sets to the engine, so merge their samples by
+// timestamp, keeping the first sample on ties.
+func sortAndMerge(list []*series) []*series {
+	if len(list) < 2 {
+		return list
+	}
+	sort.Slice(list, func(i, j int) bool {
+		return labels.Compare(list[i].lset, list[j].lset) < 0
+	})
+	out := list[:1]
+	for _, s := range list[1:] {
+		last := out[len(out)-1]
+		if labels.Compare(last.lset, s.lset) != 0 {
+			out = append(out, s)
+			continue
+		}
+		merged := mergeSamples(last, s)
+		out[len(out)-1] = merged
+	}
+	return out
+}
+
+func mergeSamples(a, b *series) *series {
+	ts := make([]int64, 0, len(a.ts)+len(b.ts))
+	vs := make([]float64, 0, len(a.ts)+len(b.ts))
+	i, j := 0, 0
+	for i < len(a.ts) && j < len(b.ts) {
+		switch {
+		case a.ts[i] < b.ts[j]:
+			ts = append(ts, a.ts[i])
+			vs = append(vs, a.vs[i])
+			i++
+		case a.ts[i] > b.ts[j]:
+			ts = append(ts, b.ts[j])
+			vs = append(vs, b.vs[j])
+			j++
+		default:
+			ts = append(ts, a.ts[i])
+			vs = append(vs, a.vs[i])
+			i++
+			j++
+		}
+	}
+	ts = append(ts, a.ts[i:]...)
+	vs = append(vs, a.vs[i:]...)
+	ts = append(ts, b.ts[j:]...)
+	vs = append(vs, b.vs[j:]...)
+	return &series{lset: a.lset, ts: ts, vs: vs}
+}

pkg/prometheus/clickhouseprometheusv2/sql.go

@@ -0,0 +1,199 @@
+package clickhouseprometheusv2
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/query-service/constants"
+	"github.com/huandu/go-sqlbuilder"
+	"github.com/prometheus/prometheus/model/labels"
+)
+
+// inlineFingerprintsLimit is the largest matched-series count inlined into
+// the samples query as literals. Literals engage the samples primary key and
+// avoid a second series-table scan; past a few thousand the statement itself
+// becomes the cost, and the shard-local subquery filter wins. Not
+// configurable: the crossover depends on statement parsing, not on any
+// property of a deployment an operator could know better.
+const inlineFingerprintsLimit = 5_000
+
+// buildSeriesQuery renders the series lookup: one row per matched fingerprint
+// with its labels.
+func buildSeriesQuery(start, end int64, matchers []*labels.Matcher) (string, []any, error) {
+	adjustedStart, table := timeSeriesTableFor(start, end)
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select("fingerprint", "any(labels)")
+	sb.From(fmt.Sprintf("%s.%s", databaseName, table))
+	if err := applySeriesConditions(sb, adjustedStart, end, matchers); err != nil {
+		return "", nil, err
+	}
+	sb.GroupBy("fingerprint")
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return query, args, nil
+}
+
+// buildSamplesQuery renders the samples fetch for the series selected by the
+// series lookup. Small matched sets pass inlineFingerprints — sorted uint64
+// literals that engage the samples primary key; nil means the set exceeded
+// the inline limit, and the filter becomes a semi-join re-running the series
+// predicates against the shard-local series table (complete by fingerprint
+// co-locality, see localTimeSeriesTable; a GLOBAL broadcast of the matched
+// set would ship it to every shard instead). metricNames narrows the
+// primary-key scan; when the selector had no __name__ equality, the names
+// observed on the matched series are used. A non-nil lastPerStep groups to
+// one (the last) sample per step bucket.
+func buildSamplesQuery(start, end int64, metricNames []string, inlineFingerprints []uint64, matchers []*labels.Matcher, lastPerStep *lastSamplePerStep) (string, []any, error) {
+	sb := sqlbuilder.NewSelectBuilder()
+	if lastPerStep != nil {
+		// Aliases must not shadow source columns: ClickHouse resolves aliases
+		// in WHERE too, and "max(unix_milli) AS unix_milli" would put an
+		// aggregate into the WHERE clause (error 184).
+		sb.Select("fingerprint", "max(unix_milli) AS ts", "argMax(value, unix_milli) AS val", "argMax(flags, unix_milli) AS fl")
+	} else {
+		sb.Select("fingerprint", "unix_milli", "value", "flags")
+	}
+	sb.From(fmt.Sprintf("%s.%s", databaseName, distributedSamplesV4))
+
+	switch len(metricNames) {
+	case 0:
+		// No name constraint derivable; correct but unable to use the
+		// metric_name primary-key prefix.
+	case 1:
+		sb.Where(sb.EQ("metric_name", metricNames[0]))
+	default:
+		sb.Where(sb.In("metric_name", sqlbuilder.List(metricNames)))
+	}
+	// temporality precedes metric_name in the samples primary key; the
+	// fingerprints already come from these temporalities, so this only helps
+	// granule pruning.
+	sb.Where("temporality IN ['Cumulative', 'Unspecified']")
+	if inlineFingerprints != nil {
+		sb.Where("fingerprint " + inlineFingerprintFilter(inlineFingerprints))
+	} else {
+		sub := sqlbuilder.NewSelectBuilder()
+		sub.Select("fingerprint")
+		adjustedStart, table := timeSeriesTableFor(start, end)
+		sub.From(fmt.Sprintf("%s.%s", databaseName, localTimeSeriesTable(table)))
+		if err := applySeriesConditions(sub, adjustedStart, end, matchers); err != nil {
+			return "", nil, err
+		}
+		sb.Where(sb.In("fingerprint", sub))
+	}
+	sb.Where(sb.GTE("unix_milli", start), sb.LTE("unix_milli", end))
+
+	if lastPerStep != nil {
+		sb.GroupBy("fingerprint")
+		if expr := lastPerStep.bucketExpr(); expr != "" {
+			sb.GroupBy(expr)
+		}
+		sb.OrderBy("fingerprint", "ts")
+	} else {
+		sb.OrderBy("fingerprint", "unix_milli")
+	}
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return query, args, nil
+}
+
+// applySeriesConditions adds the WHERE conditions of a series table scan for
+// the given matchers and window. __name__ matchers translate to the
+// metric_name column (all four matcher types — the v1 client silently
+// returned nothing for regex metric names); every other matcher translates
+// to a JSONExtractString condition on the labels column. An equality matcher
+// against "" matches series without the label, mirroring PromQL, because
+// JSONExtractString returns "" for missing keys. Regexes are anchored:
+// PromQL matchers match the whole value, while ClickHouse match() searches
+// for a partial match — without anchoring, =~"api" would also select
+// "x-api-y".
+func applySeriesConditions(sb *sqlbuilder.SelectBuilder, start, end int64, matchers []*labels.Matcher) error {
+	for _, m := range matchers {
+		if m.Name != metricNameLabel {
+			continue
+		}
+		switch m.Type {
+		case labels.MatchEqual:
+			sb.Where(sb.EQ("metric_name", m.Value))
+		case labels.MatchNotEqual:
+			sb.Where(sb.NE("metric_name", m.Value))
+		case labels.MatchRegexp:
+			sb.Where(fmt.Sprintf("match(metric_name, %s)", sb.Var(anchorRegex(m.Value))))
+		case labels.MatchNotRegexp:
+			sb.Where(fmt.Sprintf("NOT match(metric_name, %s)", sb.Var(anchorRegex(m.Value))))
+		default:
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported matcher type %q for __name__", m.Type)
+		}
+	}
+
+	sb.Where("temporality IN ['Cumulative', 'Unspecified']")
+	sb.Where(fmt.Sprintf("__normalized = %v", !constants.IsDotMetricsEnabled))
+	sb.Where(sb.GTE("unix_milli", start), sb.LT("unix_milli", end))
+
+	for _, m := range matchers {
+		if m.Name == metricNameLabel {
+			continue
+		}
+		switch m.Type {
+		case labels.MatchEqual:
+			sb.Where(fmt.Sprintf("JSONExtractString(labels, %s) = %s", sb.Var(m.Name), sb.Var(m.Value)))
+		case labels.MatchNotEqual:
+			sb.Where(fmt.Sprintf("JSONExtractString(labels, %s) != %s", sb.Var(m.Name), sb.Var(m.Value)))
+		case labels.MatchRegexp:
+			sb.Where(fmt.Sprintf("match(JSONExtractString(labels, %s), %s)", sb.Var(m.Name), sb.Var(anchorRegex(m.Value))))
+		case labels.MatchNotRegexp:
+			sb.Where(fmt.Sprintf("NOT match(JSONExtractString(labels, %s), %s)", sb.Var(m.Name), sb.Var(anchorRegex(m.Value))))
+		default:
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported matcher type %q", m.Type)
+		}
+	}
+
+	return nil
+}
+
+// anchorRegex turns a PromQL regex into its fully-anchored form (see
+// applySeriesConditions).
+func anchorRegex(v string) string {
+	return "^(?:" + v + ")$"
+}
+
+// inlineFingerprintFilter renders "IN (fp1, fp2, ...)" with literal uint64s.
+func inlineFingerprintFilter(fingerprints []uint64) string {
+	var b strings.Builder
+	b.Grow(len(fingerprints)*21 + 8)
+	b.WriteString("IN (")
+	for i, fp := range fingerprints {
+		if i > 0 {
+			b.WriteString(", ")
+		}
+		b.WriteString(strconv.FormatUint(fp, 10))
+	}
+	b.WriteString(")")
+	return b.String()
+}
+
+// lastSamplePerStep reduces an instant-selector fetch to the last sample of
+// each step bucket. Buckets are anchored at the selector's first evaluation
+// timestamp so that every bucket boundary coincides with an evaluation
+// timestamp: bucket 0 is (start, firstEval] (the initial lookback window)
+// and bucket i is (firstEval+(i-1)·step, firstEval+i·step]. Keeping only the
+// last sample per bucket is lossless: the engine resolves each evaluation
+// timestamp t to the latest sample in (t-lookback, t], and a non-final
+// sample of a bucket can never be that latest sample for any t on the
+// evaluation grid. Real timestamps are preserved, so the engine's own
+// lookback and staleness handling remain exact.
+type lastSamplePerStep struct {
+	firstEvalMs int64
+	stepMs      int64
+}
+
+func (t *lastSamplePerStep) bucketExpr() string {
+	if t.stepMs <= 0 {
+		// Instant query: a single evaluation at firstEval; one bucket.
+		return ""
+	}
+	return fmt.Sprintf(
+		"if(unix_milli <= %d, 0, intDiv(unix_milli - %d - 1, %d) + 1)",
+		t.firstEvalMs, t.firstEvalMs, t.stepMs,
+	)
+}

pkg/prometheus/clickhouseprometheusv2/sql_test.go

@@ -0,0 +1,148 @@
+package clickhouseprometheusv2
+
+import (
+	"testing"
+	"time"
+
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func mustMatcher(t *testing.T, mt labels.MatchType, name, value string) *labels.Matcher {
+	t.Helper()
+	m, err := labels.NewMatcher(mt, name, value)
+	require.NoError(t, err)
+	return m
+}
+
+func TestTimeSeriesTableFor(t *testing.T) {
+	base := time.Date(2026, 7, 10, 3, 27, 0, 0, time.UTC).UnixMilli()
+
+	tests := []struct {
+		name      string
+		span      time.Duration
+		wantTable string
+		roundTo   time.Duration
+	}{
+		{"under 6h uses hourly table", 2 * time.Hour, distributedTimeSeriesV4, time.Hour},
+		{"under 1d uses 6h table", 12 * time.Hour, distributedTimeSeriesV46hrs, 6 * time.Hour},
+		{"under 1w uses 1d table", 3 * 24 * time.Hour, distributedTimeSeriesV41day, 24 * time.Hour},
+		{"over 1w uses 1w table", 10 * 24 * time.Hour, distributedTimeSeriesV41week, 7 * 24 * time.Hour},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			start, table := timeSeriesTableFor(base, base+tt.span.Milliseconds())
+			assert.Equal(t, tt.wantTable, table)
+			assert.Zero(t, start%tt.roundTo.Milliseconds())
+			assert.LessOrEqual(t, start, base)
+		})
+	}
+}
+
+func TestBuildSeriesQuery(t *testing.T) {
+	start := int64(1_700_000_000_000)
+	end := start + time.Hour.Milliseconds()
+	// The series table window rounds down to the table's bucket boundary.
+	adjustedStart := start - (start % time.Hour.Milliseconds())
+
+	t.Run("equality name and label matchers", func(t *testing.T) {
+		query, args, err := buildSeriesQuery(start, end, []*labels.Matcher{
+			mustMatcher(t, labels.MatchEqual, "__name__", "http_requests_total"),
+			mustMatcher(t, labels.MatchEqual, "job", "api"),
+		})
+		require.NoError(t, err)
+		assert.Equal(t,
+			"SELECT fingerprint, any(labels) FROM signoz_metrics.distributed_time_series_v4 WHERE metric_name = ? AND temporality IN ['Cumulative', 'Unspecified'] AND __normalized = false AND unix_milli >= ? AND unix_milli < ? AND JSONExtractString(labels, ?) = ? GROUP BY fingerprint",
+			query,
+		)
+		assert.Equal(t, []any{"http_requests_total", adjustedStart, end, "job", "api"}, args)
+	})
+
+	t.Run("regex matchers are anchored", func(t *testing.T) {
+		_, args, err := buildSeriesQuery(start, end, []*labels.Matcher{
+			mustMatcher(t, labels.MatchEqual, "__name__", "up"),
+			mustMatcher(t, labels.MatchRegexp, "instance", "prod.*"),
+			mustMatcher(t, labels.MatchNotRegexp, "env", "dev|test"),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, []any{"up", adjustedStart, end, "instance", "^(?:prod.*)$", "env", "^(?:dev|test)$"}, args)
+	})
+
+	t.Run("regex name matcher uses metric_name column", func(t *testing.T) {
+		query, args, err := buildSeriesQuery(start, end, []*labels.Matcher{
+			mustMatcher(t, labels.MatchRegexp, "__name__", "node_cpu.*|node_memory.*"),
+		})
+		require.NoError(t, err)
+		assert.Contains(t, query, "match(metric_name, ?)")
+		assert.NotContains(t, query, "JSONExtractString")
+		assert.Equal(t, []any{"^(?:node_cpu.*|node_memory.*)$", adjustedStart, end}, args)
+	})
+
+	t.Run("no name matcher omits metric_name condition", func(t *testing.T) {
+		query, _, err := buildSeriesQuery(start, end, []*labels.Matcher{
+			mustMatcher(t, labels.MatchEqual, "job", "api"),
+		})
+		require.NoError(t, err)
+		assert.NotContains(t, query, "metric_name")
+	})
+}
+
+func TestBuildSamplesQuery(t *testing.T) {
+	start := int64(1_700_000_000_000)
+	end := start + time.Hour.Milliseconds()
+	adjustedStart := start - (start % time.Hour.Milliseconds())
+	matchers := []*labels.Matcher{
+		mustMatcher(t, labels.MatchEqual, "__name__", "up"),
+		mustMatcher(t, labels.MatchEqual, "job", "api"),
+	}
+
+	t.Run("raw with inline fingerprints", func(t *testing.T) {
+		query, args, err := buildSamplesQuery(start, end, []string{"up"}, []uint64{7, 42}, matchers, nil)
+		require.NoError(t, err)
+		assert.Equal(t,
+			"SELECT fingerprint, unix_milli, value, flags FROM signoz_metrics.distributed_samples_v4 WHERE metric_name = ? AND temporality IN ['Cumulative', 'Unspecified'] AND fingerprint IN (7, 42) AND unix_milli >= ? AND unix_milli <= ? ORDER BY fingerprint, unix_milli",
+			query,
+		)
+		assert.Equal(t, []any{"up", start, end}, args)
+	})
+
+	t.Run("last-sample-per-step groups by step bucket anchored at first eval", func(t *testing.T) {
+		lastPerStep := &lastSamplePerStep{firstEvalMs: start + 299_999, stepMs: 60_000}
+		query, _, err := buildSamplesQuery(start, end, []string{"up"}, []uint64{7}, matchers, lastPerStep)
+		require.NoError(t, err)
+		assert.Contains(t, query, "argMax(value, unix_milli) AS val")
+		assert.Contains(t, query, "argMax(flags, unix_milli) AS fl")
+		assert.Contains(t, query, "GROUP BY fingerprint, if(unix_milli <= 1700000299999, 0, intDiv(unix_milli - 1700000299999 - 1, 60000) + 1)")
+		assert.Contains(t, query, "ORDER BY fingerprint, ts")
+		// Aliases must not shadow the source columns referenced in WHERE.
+		assert.NotContains(t, query, "AS unix_milli")
+		assert.NotContains(t, query, "AS value")
+		assert.NotContains(t, query, "AS flags")
+	})
+
+	t.Run("instant query keeps one bucket", func(t *testing.T) {
+		lastPerStep := &lastSamplePerStep{firstEvalMs: end, stepMs: 0}
+		query, _, err := buildSamplesQuery(start, end, []string{"up"}, []uint64{7}, matchers, lastPerStep)
+		require.NoError(t, err)
+		assert.Contains(t, query, "GROUP BY fingerprint ORDER BY fingerprint, ts")
+		assert.NotContains(t, query, "intDiv")
+	})
+
+	t.Run("over-limit set becomes a shard-local semi-join", func(t *testing.T) {
+		query, args, err := buildSamplesQuery(start, end, []string{"up"}, nil, matchers, nil)
+		require.NoError(t, err)
+		assert.Contains(t, query, "fingerprint IN (SELECT fingerprint FROM signoz_metrics.time_series_v4 WHERE ")
+		assert.NotContains(t, query, "GLOBAL IN")
+		// Args follow placeholder order: samples metric name, the semi-join's
+		// series predicates, then the samples window bounds.
+		assert.Equal(t, []any{"up", "up", adjustedStart, end, "job", "api", start, end}, args)
+	})
+
+	t.Run("multiple metric names from regex selector", func(t *testing.T) {
+		query, args, err := buildSamplesQuery(start, end, []string{"node_cpu", "node_memory"}, []uint64{7}, matchers, nil)
+		require.NoError(t, err)
+		assert.Contains(t, query, "metric_name IN (?, ?)")
+		assert.Equal(t, []any{"node_cpu", "node_memory", start, end}, args)
+	})
+}

pkg/prometheus/clickhouseprometheusv2/tables.go

@@ -0,0 +1,65 @@
+package clickhouseprometheusv2
+
+import "time"
+
+const (
+	// metricNameLabel is the reserved PromQL label holding the metric name.
+	metricNameLabel string = "__name__"
+
+	databaseName                 string = "signoz_metrics"
+	distributedTimeSeriesV4      string = "distributed_time_series_v4"
+	distributedTimeSeriesV46hrs  string = "distributed_time_series_v4_6hrs"
+	distributedTimeSeriesV41day  string = "distributed_time_series_v4_1day"
+	distributedTimeSeriesV41week string = "distributed_time_series_v4_1week"
+	distributedSamplesV4         string = "distributed_samples_v4"
+
+	localTimeSeriesV4      string = "time_series_v4"
+	localTimeSeriesV46hrs  string = "time_series_v4_6hrs"
+	localTimeSeriesV41day  string = "time_series_v4_1day"
+	localTimeSeriesV41week string = "time_series_v4_1week"
+)
+
+// localTimeSeriesTable maps a distributed time series table to its shard-local
+// table. Samples and time series shard on the same key
+// (cityHash64(env, temporality, metric_name, fingerprint)), so a query whose
+// top-level FROM is the distributed samples table can join or semi-join the
+// local time series table inside each shard: the shard rewrite runs the
+// subquery against the shard's own series rows, which are exactly the series
+// of the shard's samples. No broadcast, no initiator-side join.
+func localTimeSeriesTable(distributed string) string {
+	switch distributed {
+	case distributedTimeSeriesV46hrs:
+		return localTimeSeriesV46hrs
+	case distributedTimeSeriesV41day:
+		return localTimeSeriesV41day
+	case distributedTimeSeriesV41week:
+		return localTimeSeriesV41week
+	default:
+		return localTimeSeriesV4
+	}
+}
+
+var (
+	oneHourInMilliseconds  = time.Hour.Milliseconds()
+	sixHoursInMilliseconds = time.Hour.Milliseconds() * 6
+	oneDayInMilliseconds   = time.Hour.Milliseconds() * 24
+	oneWeekInMilliseconds  = time.Hour.Milliseconds() * 24 * 7
+)
+
+// timeSeriesTableFor returns the adjusted start and the time series table for
+// the window. Time series tables hold one row per (fingerprint, bucket), with
+// bucket granularities of 1h, 6h, 1d and 1w; the start is rounded down to the
+// bucket boundary so a window beginning mid-bucket still matches the bucket's
+// row.
+func timeSeriesTableFor(start, end int64) (int64, string) {
+	switch {
+	case end-start < sixHoursInMilliseconds:
+		return start - (start % oneHourInMilliseconds), distributedTimeSeriesV4
+	case end-start < oneDayInMilliseconds:
+		return start - (start % sixHoursInMilliseconds), distributedTimeSeriesV46hrs
+	case end-start < oneWeekInMilliseconds:
+		return start - (start % oneDayInMilliseconds), distributedTimeSeriesV41day
+	default:
+		return start - (start % oneWeekInMilliseconds), distributedTimeSeriesV41week
+	}
+}

pkg/prometheus/clickhouseprometheusv2/transpiler.go

@@ -0,0 +1,485 @@
+package clickhouseprometheusv2
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql/parser"
+)
+
+// The compiler turns PromQL subtrees into single ClickHouse queries built on
+// the timeSeries*ToGrid aggregate functions (CH >= 25.6), whose semantics
+// were verified against this repo's vendored engine: exact extrapolatedRate
+// behavior including counter resets, the counter zero-point clamp, the
+// 1.1x-average extrapolation threshold, left-open windows, the >= 2 samples
+// rule, stale-marker shadowing, and millisecond grid starts. Sample rows
+// never leave ClickHouse: one row per output series comes back, holding the
+// whole grid as an array.
+//
+// Scope (the allowlist): an optional sum/min/max/avg/count by/without
+// aggregation over a core unit — a rate/increase/delta/irate/idelta range
+// selection, an instant vector selection, or an avg/min/max/sum/count/last
+// _over_time window — plus number-literal arithmetic/comparisons and unary
+// minus on top. Units inside fixed-resolution subqueries evaluate on the
+// subquery's own grid. Everything else either falls back to the engine over
+// this package's querier, or — when a transpilable subtree sits under a
+// non-transpilable node — runs hybrid: the subtree's grids are computed in
+// ClickHouse and substituted into the engine as synthetic series (see
+// compiler_exec.go). See doc.go for the fallback list and the reasons behind
+// each entry.
+
+// rangeFn is a transpilable range-vector function.
+type rangeFn string
+
+const (
+	fnRate     rangeFn = "rate"
+	fnIncrease rangeFn = "increase"
+	fnDelta    rangeFn = "delta"
+	fnIRate    rangeFn = "irate"
+	fnIDelta   rangeFn = "idelta"
+)
+
+var gridFunction = map[rangeFn]string{
+	fnRate:     "timeSeriesRateToGrid",
+	fnIncrease: "timeSeriesRateToGrid", // increase == rate * range seconds, exactly (same factor algebra)
+	fnDelta:    "timeSeriesDeltaToGrid",
+	fnIRate:    "timeSeriesInstantRateToGrid",
+	fnIDelta:   "timeSeriesInstantDeltaToGrid",
+}
+
+// scalarOp is one number-literal arithmetic or comparison applied to a
+// compiled vector, evaluated in Go during assembly with the same float64
+// operations the engine uses.
+type scalarOp struct {
+	op           parser.ItemType
+	scalar       float64
+	scalarOnLeft bool
+	returnBool   bool
+}
+
+// isComparison reports whether the op is a filtering/bool comparison, which
+// preserves the metric name (arithmetic drops it).
+func (o scalarOp) isComparison() bool {
+	return o.op.IsComparisonOperator()
+}
+
+// unitKind is the selector shape at the bottom of a core unit.
+type unitKind int
+
+const (
+	// unitRange: rate/increase/delta/irate/idelta over a matrix selector.
+	unitRange unitKind = iota
+	// unitInstant: a plain vector selector resolved per grid point with
+	// lookback and stale-marker shadowing.
+	unitInstant
+	// unitOverTime: avg/min/max/sum/count/last_over_time over a matrix
+	// selector (aggregation over the window's samples, stale rows excluded).
+	unitOverTime
+)
+
+// coreUnit is one transpilable subtree: selector [-> range function] ->
+// optional aggregation -> scalar op pipeline.
+type coreUnit struct {
+	kind     unitKind
+	matchers []*labels.Matcher
+	offsetMs int64
+	fn       rangeFn // unitRange
+	overFn   string  // unitOverTime: avg|min|max|sum|count|last
+	rangeMs  int64   // unitRange/unitOverTime window
+
+	hasAgg   bool
+	aggOp    parser.ItemType // SUM MIN MAX AVG COUNT
+	by       bool
+	grouping []string
+
+	ops []scalarOp
+}
+
+// keepsName reports whether the unit's output series keep their real
+// __name__: bare/comparison-filtered instant selectors and last_over_time do
+// (it returns the raw sample, name included); range functions, the other
+// *_over_time functions, aggregations, arithmetic and bool comparisons all
+// drop it — a bool comparison returns 0/1, not the sample, so the engine
+// drops the name there too. Units that keep the name cannot be substituted
+// as synthetic series in hybrid plans — the synthetic name would replace
+// the real one — but transpile fine as full plans, where assembly emits the
+// real names.
+func (u *coreUnit) keepsName() bool {
+	nameKeepingSelector := u.kind == unitInstant || (u.kind == unitOverTime && u.overFn == "last")
+	if !nameKeepingSelector || u.hasAgg {
+		return false
+	}
+	for _, op := range u.ops {
+		if !op.isComparison() || op.returnBool {
+			return false
+		}
+	}
+	return true
+}
+
+// gridContext is the evaluation grid a unit computes on. The query grid for
+// top-level units; for units inside subqueries, the subquery's own grid:
+// epoch-aligned multiples of its resolution covering the subquery window,
+// exactly as the engine derives it (engine.go, *parser.SubqueryExpr case).
+type gridContext struct {
+	startMs int64
+	endMs   int64
+	stepMs  int64
+}
+
+// subqueryGrid derives the inner grid for a subquery evaluated on outer:
+// interval S, end = outer end − offset, start = first multiple of S strictly
+// greater than outer start − offset − range.
+func subqueryGrid(outer gridContext, rangeMs, stepMs, offsetMs int64) gridContext {
+	lower := outer.startMs - offsetMs - rangeMs
+	start := stepMs * (lower / stepMs)
+	if start <= lower {
+		start += stepMs
+	}
+	return gridContext{startMs: start, endMs: outer.endMs - offsetMs, stepMs: stepMs}
+}
+
+// transpiledUnit is a coreUnit scheduled for execution, named for hybrid
+// substitution, carrying the grid it evaluates on.
+type transpiledUnit struct {
+	core coreUnit
+	name string // __signoz_transpiled_<n>__
+	grid gridContext
+}
+
+// transpilePlan is the outcome of classifying a query.
+type transpilePlan struct {
+	units []*transpiledUnit
+	grid  gridContext // the query's top-level grid
+	// full is set when the entire query is units[0]; otherwise rewritten
+	// holds the query with each unit replaced by a synthetic selector, to be
+	// evaluated by the engine over a hybrid storage.
+	full      bool
+	rewritten string
+}
+
+const syntheticNamePrefix = "__signoz_transpiled_"
+
+func syntheticName(i int) string {
+	return fmt.Sprintf("%s%d__", syntheticNamePrefix, i)
+}
+
+// classifyCore matches a subtree against the transpilable core shape.
+// stepMs gates second-granularity: the grid functions take whole-second step
+// and window parameters (grid *starts* are millisecond-precise).
+func classifyCore(node parser.Expr, stepMs int64) (*coreUnit, bool) {
+	unit := &coreUnit{}
+
+	expr := node
+	// Peel scalar ops and parens off the top, outermost first; ops apply in
+	// evaluation order, so prepend while peeling.
+	for {
+		switch n := expr.(type) {
+		case *parser.ParenExpr:
+			expr = n.Expr
+			continue
+		case *parser.UnaryExpr:
+			if n.Op != parser.SUB {
+				expr = n.Expr // unary '+' is a no-op
+				continue
+			}
+			// -x == -1 * x for every float64 (incl. NaN and signed zero).
+			unit.ops = append([]scalarOp{{op: parser.MUL, scalar: -1}}, unit.ops...)
+			expr = n.Expr
+			continue
+		case *parser.StepInvariantExpr:
+			// @-pinned expressions evaluate on a different grid.
+			return nil, false
+		case *parser.BinaryExpr:
+			lit, litOnLeft, ok := numberLiteralSide(n)
+			if !ok {
+				return nil, false
+			}
+			if !n.Op.IsOperator() && !n.Op.IsComparisonOperator() {
+				return nil, false
+			}
+			if n.Op == parser.ATAN2 {
+				// atan2 is arithmetic in PromQL but rarely used; keep the
+				// allowlist tight.
+				return nil, false
+			}
+			returnBool := n.ReturnBool
+			unit.ops = append([]scalarOp{{op: n.Op, scalar: lit, scalarOnLeft: litOnLeft, returnBool: returnBool}}, unit.ops...)
+			if litOnLeft {
+				expr = n.RHS
+			} else {
+				expr = n.LHS
+			}
+			continue
+		}
+		break
+	}
+
+	// Optional aggregation.
+	if agg, ok := expr.(*parser.AggregateExpr); ok {
+		switch agg.Op {
+		case parser.SUM, parser.MIN, parser.MAX, parser.AVG, parser.COUNT:
+		default:
+			return nil, false
+		}
+		for _, g := range agg.Grouping {
+			if g == metricNameLabel {
+				// by(__name__)/without(__name__) over synthetic or compiled
+				// output needs name bookkeeping the compiler doesn't do.
+				return nil, false
+			}
+		}
+		unit.hasAgg = true
+		unit.aggOp = agg.Op
+		unit.by = !agg.Without
+		unit.grouping = agg.Grouping
+		expr = agg.Expr
+		for {
+			if p, ok := expr.(*parser.ParenExpr); ok {
+				expr = p.Expr
+				continue
+			}
+			break
+		}
+	}
+
+	// The grid functions take whole-second steps; stepMs == 0 is an instant
+	// query (single-point grid).
+	if stepMs < 0 || stepMs%1000 != 0 {
+		return nil, false
+	}
+
+	// Bare instant selector: resolved per grid point with lookback and
+	// stale-marker shadowing (see compiler_sql.go).
+	if vs, ok := expr.(*parser.VectorSelector); ok {
+		if vs.Timestamp != nil || vs.StartOrEnd != 0 || vs.Anchored || vs.Smoothed {
+			return nil, false
+		}
+		offsetMs := vs.OriginalOffset.Milliseconds()
+		if offsetMs < 0 {
+			return nil, false
+		}
+		unit.kind = unitInstant
+		unit.offsetMs = offsetMs
+		unit.matchers = vs.LabelMatchers
+		return unit, true
+	}
+
+	// Range or *_over_time function over a plain matrix selector.
+	call, ok := expr.(*parser.Call)
+	if !ok {
+		return nil, false
+	}
+	var fn rangeFn
+	var overFn string
+	switch call.Func.Name {
+	case "rate":
+		fn = fnRate
+	case "increase":
+		fn = fnIncrease
+	case "delta":
+		fn = fnDelta
+	case "irate":
+		fn = fnIRate
+	case "idelta":
+		fn = fnIDelta
+	case "avg_over_time", "min_over_time", "max_over_time", "sum_over_time", "count_over_time", "last_over_time":
+		overFn = strings.TrimSuffix(call.Func.Name, "_over_time")
+	default:
+		return nil, false
+	}
+	if len(call.Args) != 1 {
+		return nil, false
+	}
+	ms, ok := call.Args[0].(*parser.MatrixSelector)
+	if !ok {
+		return nil, false
+	}
+	vs, ok := ms.VectorSelector.(*parser.VectorSelector)
+	if !ok {
+		return nil, false
+	}
+	if vs.Timestamp != nil || vs.StartOrEnd != 0 || vs.Anchored || vs.Smoothed {
+		return nil, false
+	}
+
+	rangeMs := ms.Range.Milliseconds()
+	offsetMs := vs.OriginalOffset.Milliseconds()
+	if rangeMs <= 0 || rangeMs%1000 != 0 || offsetMs < 0 {
+		return nil, false
+	}
+
+	if overFn != "" {
+		unit.kind = unitOverTime
+		unit.overFn = overFn
+	} else {
+		unit.kind = unitRange
+		unit.fn = fn
+	}
+	unit.rangeMs = rangeMs
+	unit.offsetMs = offsetMs
+	unit.matchers = vs.LabelMatchers
+	return unit, true
+}
+
+// numberLiteralSide returns the number literal on one side of a binary
+// expression (peeling parens and unary minus), and which side it is on.
+func numberLiteralSide(b *parser.BinaryExpr) (float64, bool, bool) {
+	if v, ok := literalValue(b.LHS); ok {
+		return v, true, true
+	}
+	if v, ok := literalValue(b.RHS); ok {
+		return v, false, true
+	}
+	return 0, false, false
+}
+
+func literalValue(e parser.Expr) (float64, bool) {
+	neg := false
+	for {
+		switch n := e.(type) {
+		case *parser.ParenExpr:
+			e = n.Expr
+			continue
+		case *parser.StepInvariantExpr:
+			e = n.Expr
+			continue
+		case *parser.UnaryExpr:
+			if n.Op == parser.SUB {
+				neg = !neg
+			}
+			e = n.Expr
+			continue
+		case *parser.NumberLiteral:
+			if neg {
+				return -n.Val, true
+			}
+			return n.Val, true
+		default:
+			return 0, false
+		}
+	}
+}
+
+// classify builds the compile plan for a query: full when the root is a core
+// unit, hybrid when core units sit strictly below the root (including inside
+// fixed-resolution subqueries, computed on the subquery grid), none
+// otherwise.
+func classify(root parser.Expr, grid gridContext) (*transpilePlan, bool) {
+	if unit, ok := classifyCore(root, grid.stepMs); ok {
+		return &transpilePlan{
+			units: []*transpiledUnit{{core: *unit, name: syntheticName(0), grid: grid}},
+			grid:  grid,
+			full:  true,
+		}, true
+	}
+
+	plan := &transpilePlan{grid: grid}
+	rewritten := rewrite(root, grid, plan, false)
+	if len(plan.units) == 0 {
+		return nil, false
+	}
+	plan.rewritten = rewritten.String()
+	return plan, true
+}
+
+// rewrite walks top-down replacing maximal transpilable subtrees with synthetic
+// vector selectors. nameSensitive marks scopes where an ancestor's semantics
+// depend on __name__ (grouping or vector matching on it): synthetic series
+// carry a synthetic __name__, so substitution there would change results.
+// Fixed-resolution subqueries recurse with the subquery's own grid; scopes
+// whose evaluation grid is unknowable (@-pinned, default-resolution
+// subqueries) are not entered.
+func rewrite(node parser.Expr, grid gridContext, plan *transpilePlan, nameSensitive bool) parser.Expr {
+	if node == nil {
+		return nil
+	}
+
+	if !nameSensitive {
+		// Units whose output keeps the real __name__ (bare instant selectors)
+		// cannot be substituted: the synthetic name would replace it in the
+		// engine's output. They still compile as full plans.
+		if unit, ok := classifyCore(node, grid.stepMs); ok && !unit.keepsName() {
+			cu := &transpiledUnit{core: *unit, name: syntheticName(len(plan.units)), grid: grid}
+			plan.units = append(plan.units, cu)
+			return &parser.VectorSelector{
+				Name: cu.name,
+				LabelMatchers: []*labels.Matcher{
+					labels.MustNewMatcher(labels.MatchEqual, metricNameLabel, cu.name),
+				},
+				PosRange: node.PositionRange(),
+			}
+		}
+	}
+
+	switch n := node.(type) {
+	case *parser.ParenExpr:
+		n.Expr = rewrite(n.Expr, grid, plan, nameSensitive)
+	case *parser.UnaryExpr:
+		n.Expr = rewrite(n.Expr, grid, plan, nameSensitive)
+	case *parser.AggregateExpr:
+		sensitive := nameSensitive || groupingUsesName(n.Grouping)
+		n.Expr = rewrite(n.Expr, grid, plan, sensitive)
+		// n.Param is a scalar/string; nothing transpilable inside for our core.
+	case *parser.Call:
+		for i, arg := range n.Args {
+			n.Args[i] = rewrite(arg, grid, plan, nameSensitive)
+		}
+	case *parser.BinaryExpr:
+		sensitive := nameSensitive || vectorMatchingUsesName(n.VectorMatching)
+		n.LHS = rewrite(n.LHS, grid, plan, sensitive)
+		n.RHS = rewrite(n.RHS, grid, plan, sensitive)
+	case *parser.SubqueryExpr:
+		// The alert-smoothing idiom fn_over_time((expr)[R:S]) dominates real
+		// rule fleets; inner units evaluate on the subquery grid, and the
+		// engine does the smoothing over the synthetic series. Requires an
+		// explicit whole-second resolution (S == 0 needs the engine's
+		// default-interval function) and no @ pinning.
+		stepMs := n.Step.Milliseconds()
+		rangeMs := n.Range.Milliseconds()
+		offsetMs := n.OriginalOffset.Milliseconds()
+		if n.Timestamp == nil && n.StartOrEnd == 0 &&
+			stepMs > 0 && stepMs%1000 == 0 && rangeMs%1000 == 0 && offsetMs >= 0 {
+			inner := subqueryGrid(grid, rangeMs, stepMs, offsetMs)
+			n.Expr = rewrite(n.Expr, inner, plan, nameSensitive)
+		}
+	case *parser.StepInvariantExpr, *parser.MatrixSelector,
+		*parser.VectorSelector, *parser.NumberLiteral, *parser.StringLiteral:
+		// Leaves, or scopes substitution must not enter.
+	}
+	return node
+}
+
+func groupingUsesName(grouping []string) bool {
+	for _, g := range grouping {
+		if g == metricNameLabel {
+			return true
+		}
+	}
+	return false
+}
+
+func vectorMatchingUsesName(vm *parser.VectorMatching) bool {
+	if vm == nil {
+		return false
+	}
+	for _, l := range append(append([]string{}, vm.MatchingLabels...), vm.Include...) {
+		if l == metricNameLabel {
+			return true
+		}
+	}
+	// Default (all-labels) matching ignores __name__, and by()/ignoring()
+	// lists were checked above.
+	return false
+}
+
+// isSyntheticSelector reports whether matchers target a compiled unit.
+func isSyntheticSelector(matchers []*labels.Matcher) (string, bool) {
+	for _, m := range matchers {
+		if m.Name == metricNameLabel && m.Type == labels.MatchEqual && strings.HasPrefix(m.Value, syntheticNamePrefix) {
+			return m.Value, true
+		}
+	}
+	return "", false
+}

pkg/prometheus/clickhouseprometheusv2/transpiler_corpus_test.go

@@ -0,0 +1,150 @@
+package clickhouseprometheusv2
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"os"
+	"regexp"
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/prometheus/promql/parser"
+	"github.com/stretchr/testify/require"
+)
+
+// TestClassifyCorpus measures real-workload compiler coverage: it classifies
+// every query of a JSON-lines corpus (one JSON-encoded PromQL string per
+// line) with the live classifier and reports full / hybrid / fallback
+// shares. Skipped unless PROMQL_CORPUS points to one or more files
+// (comma-separated). Dashboard template variables are substituted with
+// placeholder values before parsing, mirroring the production render step.
+//
+//	PROMQL_CORPUS=corpus-a.jsonl,corpus-b.jsonl go test -run TestClassifyCorpus -v
+func TestClassifyCorpus(t *testing.T) {
+	corpus := os.Getenv("PROMQL_CORPUS")
+	if corpus == "" {
+		t.Skip("PROMQL_CORPUS not set")
+	}
+
+	varRe := regexp.MustCompile(`\{\{\s*\.?[\w.]+\s*\}\}|\[\[\s*[\w.]+\s*\]\]|\$[\w.]+`)
+	promParser := parser.NewParser(parser.Options{})
+
+	for _, path := range strings.Split(corpus, ",") {
+		f, err := os.Open(path)
+		require.NoError(t, err)
+
+		var full, hybrid, fallbackInstant, fallbackOther, parseErrs int
+		fallbackReasons := map[string]int{}
+
+		scanner := bufio.NewScanner(f)
+		scanner.Buffer(make([]byte, 1024*1024), 1024*1024)
+		for scanner.Scan() {
+			var query string
+			require.NoError(t, json.Unmarshal(scanner.Bytes(), &query))
+			query = varRe.ReplaceAllString(query, "placeholder")
+
+			expr, err := promParser.ParseExpr(query)
+			if err != nil {
+				parseErrs++
+				continue
+			}
+
+			plan, ok := classify(expr, gridContext{startMs: 1_700_000_000_000, endMs: 1_700_007_200_000, stepMs: 60_000})
+			switch {
+			case ok && plan.full:
+				full++
+			case ok:
+				hybrid++
+			default:
+				reason := fallbackShape(expr)
+				fallbackReasons[reason]++
+				if reason == "instant-selector shape (last-sample-per-step engine path)" {
+					fallbackInstant++
+				} else {
+					fallbackOther++
+				}
+			}
+		}
+		require.NoError(t, scanner.Err())
+		_ = f.Close()
+
+		total := full + hybrid + fallbackInstant + fallbackOther
+		if total == 0 {
+			t.Logf("%s: no parseable queries (%d parse errors)", path, parseErrs)
+			continue
+		}
+		t.Logf("%s: %d queries — full=%d (%.0f%%) hybrid=%d (%.0f%%) fallback=%d (%.0f%%; instant-shape=%d) parse_errors=%d",
+			path, total,
+			full, 100*float64(full)/float64(total),
+			hybrid, 100*float64(hybrid)/float64(total),
+			fallbackInstant+fallbackOther, 100*float64(fallbackInstant+fallbackOther)/float64(total),
+			fallbackInstant, parseErrs)
+
+		reasons := make([]string, 0, len(fallbackReasons))
+		for r := range fallbackReasons {
+			reasons = append(reasons, r)
+		}
+		sort.Slice(reasons, func(i, j int) bool { return fallbackReasons[reasons[i]] > fallbackReasons[reasons[j]] })
+		for _, r := range reasons {
+			t.Logf("  fallback %4d  %s", fallbackReasons[r], r)
+		}
+	}
+}
+
+// fallbackShape buckets a non-transpilable query by why it stays on the engine
+// path, to separate "already served well" (instant selectors on the last-sample-per-step
+// path) from genuine compiler gaps.
+func fallbackShape(expr parser.Expr) string {
+	var hasMatrix, hasSubquery, hasAt, overTime bool
+	rangeFns := map[string]bool{"rate": true, "increase": true, "delta": true, "irate": true, "idelta": true}
+	var unsupportedFns []string
+	parser.Inspect(expr, func(node parser.Node, _ []parser.Node) error {
+		switch n := node.(type) {
+		case *parser.MatrixSelector:
+			hasMatrix = true
+		case *parser.SubqueryExpr:
+			hasSubquery = true
+		case *parser.VectorSelector:
+			if n.Timestamp != nil || n.StartOrEnd != 0 {
+				hasAt = true
+			}
+		case *parser.Call:
+			if strings.HasSuffix(n.Func.Name, "_over_time") {
+				overTime = true
+			} else if !rangeFns[n.Func.Name] {
+				unsupportedFns = append(unsupportedFns, n.Func.Name)
+			}
+		}
+		return nil
+	})
+
+	switch {
+	case hasSubquery:
+		return "subquery"
+	case hasAt:
+		return "@ modifier"
+	case overTime:
+		return "*_over_time range function"
+	case !hasMatrix:
+		return "instant-selector shape (last-sample-per-step engine path)"
+	case len(unsupportedFns) > 0:
+		return fmt.Sprintf("range shape with unsupported function(s): %s", strings.Join(dedupe(unsupportedFns), ",")) //nolint:makezero
+	default:
+		return "other range shape"
+	}
+}
+
+func dedupe(in []string) []string {
+	seen := map[string]bool{}
+	var out []string
+	for _, s := range in {
+		if !seen[s] {
+			seen[s] = true
+			out = append(out, s)
+		}
+	}
+	sort.Strings(out)
+	return out
+}

pkg/prometheus/clickhouseprometheusv2/transpiler_exec.go

@@ -0,0 +1,463 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"sort"
+	"sync/atomic"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
+	promValue "github.com/prometheus/prometheus/model/value"
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/promql/parser"
+	"github.com/prometheus/prometheus/storage"
+	"golang.org/x/sync/errgroup"
+)
+
+// executor evaluates transpilable PromQL directly in ClickHouse, falling
+// back (ok=false) whenever the query shape or the step doesn't qualify. The
+// timeSeries*ToGrid functions it builds on are assumed available: the
+// supported ClickHouse floor is >= 25.6.
+type executor struct {
+	client *client
+	engine *prometheus.Engine
+	parser prometheus.Parser
+}
+
+// TryExecuteRange transpiles and runs the query in ClickHouse when its shape
+// is in the allowlist. ok=false means "not transpilable" and carries no
+// error; the caller runs the engine path.
+func (e *executor) TryExecuteRange(ctx context.Context, qs string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error) {
+	expr, err := e.parser.ParseExpr(qs)
+	if err != nil {
+		// Let the engine path produce the (enhanced) parse error.
+		return nil, false, nil
+	}
+
+	plan, ok := classify(expr, queryGrid(start, end, step))
+	if !ok {
+		return nil, false, nil
+	}
+
+	// timeSeriesLastToGrid widens its window to max(window, step) — probed: a
+	// sample aged (window, step] still fills the slot — while the rate/delta
+	// family enforces the window strictly. The Last-style kinds therefore
+	// transpile only when their window covers the step; otherwise the engine
+	// path serves them exactly.
+	for _, unit := range plan.units {
+		lastStyle := unit.core.kind == unitInstant || (unit.core.kind == unitOverTime && unit.core.overFn == "last")
+		if !lastStyle {
+			continue
+		}
+		windowMs := unit.core.rangeMs
+		if unit.core.kind == unitInstant {
+			windowMs = e.client.lookbackMs
+		}
+		if windowMs < unit.grid.stepMs {
+			return nil, false, nil
+		}
+	}
+
+	// Evaluate every unit concurrently on its own grid (the query grid, or a
+	// subquery grid); each is one series lookup plus one grid query. The
+	// units share one grid-cell budget: transpiled results never pass
+	// through the engine's sample limiter, so without this a large
+	// series-count x grid-width query would buffer unbounded arrays — the
+	// OOM this provider exists to prevent.
+	results := make([][]transpiledSeries, len(plan.units))
+	var gridCells atomic.Int64
+	eg, egCtx := errgroup.WithContext(ctx)
+	for i, unit := range plan.units {
+		eg.Go(func() error {
+			res, err := e.executeUnit(egCtx, &unit.core, unit.grid, &gridCells)
+			if err != nil {
+				return err
+			}
+			results[i] = res
+			return nil
+		})
+	}
+	if err := eg.Wait(); err != nil {
+		return nil, true, err
+	}
+
+	if plan.full {
+		g := plan.units[0].grid
+		return toMatrix(results[0], g.startMs, g.stepMs), true, nil
+	}
+
+	matrix, err := e.executeHybrid(ctx, plan, results)
+	if err != nil {
+		return nil, true, err
+	}
+	return matrix, true, nil
+}
+
+// queryGrid derives the top-level evaluation grid; step 0 is an instant
+// query: a single evaluation at end, whatever start was.
+func queryGrid(start, end time.Time, step time.Duration) gridContext {
+	startMs, endMs, stepMs := start.UnixMilli(), end.UnixMilli(), step.Milliseconds()
+	if stepMs == 0 {
+		startMs = endMs
+	}
+	return gridContext{startMs: startMs, endMs: endMs, stepMs: stepMs}
+}
+
+// transpiledSeries is one output series of a unit: projected labels and one
+// value pointer per grid point (nil = absent).
+type transpiledSeries struct {
+	lset   labels.Labels
+	values []*float64
+}
+
+// executeUnit runs one core unit on its grid: series lookup (budgets,
+// fingerprints, metric names), then the single grid query, then the
+// scalar-op pipeline.
+func (e *executor) executeUnit(ctx context.Context, unit *coreUnit, grid gridContext, gridCells *atomic.Int64) ([]transpiledSeries, error) {
+	startMs, endMs, stepMs := grid.startMs, grid.endMs, grid.stepMs
+	windowMs := unit.rangeMs
+	if unit.kind == unitInstant {
+		windowMs = e.client.lookbackMs
+	}
+	dataStart := startMs - unit.offsetMs - windowMs
+	dataEnd := endMs - unit.offsetMs
+
+	seriesQuery, seriesArgs, err := buildSeriesQuery(dataStart, dataEnd, unit.matchers)
+	if err != nil {
+		return nil, err
+	}
+	lookup, err := e.client.selectSeries(ctx, seriesQuery, seriesArgs)
+	if err != nil {
+		return nil, err
+	}
+	if len(lookup.fingerprints) == 0 {
+		return nil, nil
+	}
+
+	// The result buffers one grid array per series; series count times grid
+	// width is the transpiled equivalent of fetched samples, counted before
+	// the arrays exist rather than after the memory is spent.
+	gridLen := int64(1)
+	if stepMs > 0 {
+		gridLen = (endMs-startMs)/stepMs + 1
+	}
+	if maxSamples := e.client.cfg.MaxFetchedSamples; maxSamples > 0 && gridCells.Add(int64(len(lookup.fingerprints))*gridLen) > maxSamples {
+		return nil, errors.NewInvalidInputf(
+			errors.CodeInvalidInput,
+			"promql query would buffer more than %d output points; narrow the selector or time range, or raise prometheus::clickhousev2::max_fetched_samples",
+			maxSamples,
+		)
+	}
+
+	query, args, err := buildUnitSQL(unit, lookup.metricNames, transpiledFingerprintFilter(lookup), dataStart, dataEnd, startMs, endMs, stepMs, e.client.lookbackMs)
+	if err != nil {
+		return nil, err
+	}
+
+	rows, err := e.client.telemetryStore.ClickhouseDB().Query(e.client.withContext(ctx, "transpiledUnit"), query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	// Name-dropping units keep __name__ in the SQL group key so distinct
+	// metrics never merge server-side; the name comes off here, and a
+	// post-strip collision is the engine's duplicate-labelset error — v1
+	// would have errored, so silently inventing a merged series would be a
+	// divergence.
+	stripName := !unit.hasAgg && !unit.keepsName()
+	seen := make(map[uint64]string)
+
+	var out []transpiledSeries
+	var gkey string
+	var gridValues []*float64
+	for rows.Next() {
+		if err := rows.Scan(&gkey, &gridValues); err != nil {
+			return nil, err
+		}
+		lset, err := labelsFromGroupKey(gkey)
+		if err != nil {
+			return nil, err
+		}
+		if stripName {
+			name := lset.Get(metricNameLabel)
+			lset = labels.NewBuilder(lset).Del(metricNameLabel).Labels()
+			if prev, ok := seen[lset.Hash()]; ok && prev != name {
+				return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "vector cannot contain metrics with the same labelset")
+			}
+			seen[lset.Hash()] = name
+		}
+		values := make([]*float64, len(gridValues))
+		copy(values, gridValues)
+		applyScalarOps(unit.ops, values)
+		out = append(out, transpiledSeries{lset: lset, values: values})
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	sort.Slice(out, func(i, j int) bool { return labels.Compare(out[i].lset, out[j].lset) < 0 })
+	return out, nil
+}
+
+// transpiledFingerprintFilter returns the matched fingerprints as a sorted
+// slice when they fit the inline limit — literals engage the samples primary
+// key, and sorting keeps the statement deterministic for logging and tests.
+// Over the limit it returns nil: the unit query's INNER JOIN against the
+// local series subquery restricts to exactly the matched fingerprints
+// already, and a semi-join on the same predicates would only rescan the
+// series table.
+func transpiledFingerprintFilter(lookup *seriesLookup) []uint64 {
+	if len(lookup.fingerprints) > inlineFingerprintsLimit {
+		return nil
+	}
+	fingerprints := make([]uint64, 0, len(lookup.fingerprints))
+	for fp := range lookup.fingerprints {
+		fingerprints = append(fingerprints, fp)
+	}
+	sort.Slice(fingerprints, func(i, j int) bool { return fingerprints[i] < fingerprints[j] })
+	return fingerprints
+}
+
+// labelsFromGroupKey parses the toJSONString'd sorted [key, value] pairs.
+func labelsFromGroupKey(gkey string) (labels.Labels, error) {
+	var pairs [][]string
+	if err := json.Unmarshal([]byte(gkey), &pairs); err != nil {
+		return labels.EmptyLabels(), errors.WrapInternalf(err, errors.CodeInternal, "malformed compiled group key %q", gkey)
+	}
+	builder := labels.NewScratchBuilder(len(pairs))
+	for _, p := range pairs {
+		if len(p) != 2 {
+			return labels.EmptyLabels(), errors.NewInternalf(errors.CodeInternal, "malformed compiled group key pair %q", gkey)
+		}
+		builder.Add(p[0], p[1])
+	}
+	builder.Sort()
+	return builder.Labels(), nil
+}
+
+// applyScalarOps applies the number-literal op pipeline in place, with the
+// same float64 arithmetic and comparison-filter semantics as the engine.
+func applyScalarOps(ops []scalarOp, values []*float64) {
+	for _, op := range ops {
+		for i, v := range values {
+			if v == nil {
+				continue
+			}
+			lhs, rhs := *v, op.scalar
+			if op.scalarOnLeft {
+				lhs, rhs = op.scalar, *v
+			}
+			switch op.op {
+			case parser.ADD:
+				res := lhs + rhs
+				values[i] = &res
+			case parser.SUB:
+				res := lhs - rhs
+				values[i] = &res
+			case parser.MUL:
+				res := lhs * rhs
+				values[i] = &res
+			case parser.DIV:
+				res := lhs / rhs
+				values[i] = &res
+			case parser.MOD:
+				res := math.Mod(lhs, rhs)
+				values[i] = &res
+			case parser.POW:
+				res := math.Pow(lhs, rhs)
+				values[i] = &res
+			default:
+				keep := compare(op.op, lhs, rhs)
+				switch {
+				case op.returnBool:
+					res := 0.0
+					if keep {
+						res = 1.0
+					}
+					values[i] = &res
+				case keep:
+					// Filter comparisons keep the vector-side value.
+					vec := *v
+					values[i] = &vec
+				default:
+					values[i] = nil
+				}
+			}
+		}
+	}
+}
+
+func compare(op parser.ItemType, lhs, rhs float64) bool {
+	switch op {
+	case parser.EQLC:
+		return lhs == rhs
+	case parser.NEQ:
+		return lhs != rhs
+	case parser.GTR:
+		return lhs > rhs
+	case parser.LSS:
+		return lhs < rhs
+	case parser.GTE:
+		return lhs >= rhs
+	case parser.LTE:
+		return lhs <= rhs
+	}
+	return false
+}
+
+// toMatrix converts a unit result to a promql matrix on the query grid.
+func toMatrix(series []transpiledSeries, startMs, stepMs int64) promql.Matrix {
+	matrix := make(promql.Matrix, 0, len(series))
+	for _, s := range series {
+		var floats []promql.FPoint
+		for i, v := range s.values {
+			if v == nil {
+				continue
+			}
+			floats = append(floats, promql.FPoint{T: startMs + int64(i)*stepMs, F: *v})
+		}
+		if len(floats) == 0 {
+			continue
+		}
+		matrix = append(matrix, promql.Series{Metric: s.lset, Floats: floats})
+	}
+	return matrix
+}
+
+// executeHybrid substitutes each unit's grids into the engine as synthetic
+// series and evaluates the rewritten query over a storage that serves
+// synthetic selectors from memory and everything else from the live querier.
+// Absent grid points become stale markers so the engine's lookback cannot
+// resurrect the previous grid point. Each unit's synthetic samples sit on its
+// own grid (query grid, or subquery grid for units inside subqueries).
+func (e *executor) executeHybrid(ctx context.Context, plan *transpilePlan, results [][]transpiledSeries) (promql.Matrix, error) {
+	synthetic := make(map[string][]*series, len(plan.units))
+	staleMarker := math.Float64frombits(promValue.StaleNaN)
+
+	queryGrid := plan.grid
+
+	for i, unit := range plan.units {
+		g := unit.grid
+		gridLen := 1
+		if g.stepMs > 0 {
+			gridLen = int((g.endMs-g.startMs)/g.stepMs) + 1
+		}
+		list := make([]*series, 0, len(results[i]))
+		for _, cs := range results[i] {
+			builder := labels.NewBuilder(cs.lset)
+			builder.Set(metricNameLabel, unit.name)
+			s := &series{lset: builder.Labels()}
+			s.ts = make([]int64, 0, gridLen)
+			s.vs = make([]float64, 0, gridLen)
+			for idx := 0; idx < gridLen; idx++ {
+				t := g.startMs + int64(idx)*g.stepMs
+				var v float64
+				if idx < len(cs.values) && cs.values[idx] != nil {
+					v = *cs.values[idx]
+				} else {
+					v = staleMarker
+				}
+				s.ts = append(s.ts, t)
+				s.vs = append(s.vs, v)
+			}
+			list = append(list, s)
+		}
+		synthetic[unit.name] = list
+	}
+
+	hybrid := &hybridQueryable{client: e.client, synthetic: synthetic}
+
+	var qry promql.Query
+	var err error
+	if queryGrid.stepMs == 0 {
+		qry, err = e.engine.NewInstantQuery(ctx, hybrid, nil, plan.rewritten, time.UnixMilli(queryGrid.endMs))
+	} else {
+		qry, err = e.engine.NewRangeQuery(ctx, hybrid, nil, plan.rewritten, time.UnixMilli(queryGrid.startMs), time.UnixMilli(queryGrid.endMs), time.Duration(queryGrid.stepMs)*time.Millisecond)
+	}
+	if err != nil {
+		return nil, err
+	}
+	defer qry.Close()
+
+	res := qry.Exec(ctx)
+	if res.Err != nil {
+		return nil, res.Err
+	}
+
+	matrix, err := resultToMatrix(res)
+	if err != nil {
+		return nil, err
+	}
+
+	// Deep-copy before Close returns the result's slices to the engine pool,
+	// and drop the synthetic __name__ that filter comparisons preserve.
+	out := make(promql.Matrix, 0, len(matrix))
+	for _, s := range matrix {
+		lset := s.Metric
+		if name := lset.Get(metricNameLabel); len(name) >= len(syntheticNamePrefix) && name[:len(syntheticNamePrefix)] == syntheticNamePrefix {
+			builder := labels.NewBuilder(lset)
+			builder.Del(metricNameLabel)
+			lset = builder.Labels()
+		}
+		floats := make([]promql.FPoint, len(s.Floats))
+		copy(floats, s.Floats)
+		out = append(out, promql.Series{Metric: lset.Copy(), Floats: floats})
+	}
+	sort.Slice(out, func(i, j int) bool { return labels.Compare(out[i].Metric, out[j].Metric) < 0 })
+	return out, nil
+}
+
+func resultToMatrix(res *promql.Result) (promql.Matrix, error) {
+	switch v := res.Value.(type) {
+	case promql.Matrix:
+		return v, nil
+	case promql.Vector:
+		matrix := make(promql.Matrix, 0, len(v))
+		for _, s := range v {
+			matrix = append(matrix, promql.Series{Metric: s.Metric, Floats: []promql.FPoint{{T: s.T, F: s.F}}})
+		}
+		return matrix, nil
+	case promql.Scalar:
+		return promql.Matrix{{Metric: labels.EmptyLabels(), Floats: []promql.FPoint{{T: v.T, F: v.V}}}}, nil
+	default:
+		return nil, errors.NewInternalf(errors.CodeInternal, "unexpected hybrid result type %T", res.Value)
+	}
+}
+
+// hybridQueryable serves synthetic (compiled) selectors from memory and
+// everything else from the live storage.
+type hybridQueryable struct {
+	client    *client
+	synthetic map[string][]*series
+}
+
+func (h *hybridQueryable) Querier(mint, maxt int64) (storage.Querier, error) {
+	return &hybridQuerier{
+		querier:   querier{mint: mint, maxt: maxt, client: h.client},
+		synthetic: h.synthetic,
+	}, nil
+}
+
+type hybridQuerier struct {
+	querier
+	synthetic map[string][]*series
+}
+
+func (h *hybridQuerier) Select(ctx context.Context, sortSeries bool, hints *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
+	if name, ok := isSyntheticSelector(matchers); ok {
+		list := h.synthetic[name]
+		if sortSeries {
+			sorted := make([]*series, len(list))
+			copy(sorted, list)
+			sort.Slice(sorted, func(i, j int) bool { return labels.Compare(sorted[i].lset, sorted[j].lset) < 0 })
+			list = sorted
+		}
+		return newSeriesSet(list)
+	}
+	return h.querier.Select(ctx, sortSeries, hints, matchers...)
+}

pkg/prometheus/clickhouseprometheusv2/transpiler_sql.go

@@ -0,0 +1,303 @@
+package clickhouseprometheusv2
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/huandu/go-sqlbuilder"
+)
+
+// experimental gate for the timeSeries*ToGrid aggregate functions; attached
+// as a SETTINGS clause so telemetrystore hooks cannot clobber it.
+const gridFunctionsSetting = "SETTINGS allow_experimental_ts_to_grid_aggregate_function = 1"
+
+var aggForEach = map[string]string{
+	"sum":   "sumForEach",
+	"min":   "minForEach",
+	"max":   "maxForEach",
+	"avg":   "avgForEach",
+	"count": "countForEach",
+}
+
+// buildUnitSQL renders the single ClickHouse query evaluating a core unit
+// over the [startMs, endMs] / stepMs evaluation grid: per-series grids via a
+// timeSeries*ToGrid aggregate (or a windowed aggregation for *_over_time),
+// then spatial aggregation with -ForEach combinators grouped by a canonical
+// JSON key of the projected label pairs.
+//
+// The heavy level is shaped to run on the shards: the top-level FROM is the
+// distributed samples table and the group-key join partner is a subquery on
+// the shard-local time series table, so the shard rewrite executes the join
+// and the per-(fingerprint, gkey) grid aggregation next to the data —
+// complete by fingerprint co-locality (see localTimeSeriesTable) — and the
+// initiator only merges per-series grid states and applies the spatial
+// -ForEach step. Same layout as the telemetrymetrics statement builder. The
+// windowed *_over_time form is the exception: its ARRAY JOIN level
+// aggregates on the shards the same way, but the group-key join happens at
+// the initiator over the already-reduced per-(series, index) rows — pushing
+// it down would not move any data off the initiator (the reduced rows arrive
+// there either way), so the combined ARRAY JOIN + JOIN form buys nothing.
+//
+// inlineFingerprints carries the matched set when it fits the inline limit;
+// nil means over the limit, where the group-key join restricts on its own
+// (the windowed form, whose fan-out query has no join, falls back to a
+// shard-local semi-join so it does not expand every series of the metric).
+//
+// The selector's data window is offset-shifted; the resulting grid indices
+// map 1:1 onto the query grid (output ts = startMs + i*stepMs). Grid
+// parameters are rendered as literals — they are aggregate-function
+// parameters, not bindable values.
+//
+// Statements nest builder-rendered SQL as text, so the returned args must be
+// ordered by where each fragment lands in the final statement: ClickHouse
+// binds ? placeholders by position. A JOIN renders before WHERE, so a joined
+// subquery's args precede the outer query's own condition args.
+//
+// Row shape: (gkey String, grid Array(Nullable(Float64))). gkey is
+// toJSONString of the sorted projected [key, value] pairs; NULL grid points
+// are absent points (the engine's "no value here"), which the -ForEach
+// combinators preserve: an index where every series is NULL aggregates to
+// NULL, and countForEach's 0 is mapped back to NULL.
+func buildUnitSQL(unit *coreUnit, metricNames []string, inlineFingerprints []uint64, dataStart, dataEnd int64, startMs, endMs, stepMs, lookbackMs int64) (string, []any, error) {
+	selStart := startMs - unit.offsetMs
+	selEnd := endMs - unit.offsetMs
+	stepSec := stepMs / 1000
+	if stepSec == 0 {
+		// Instant query: start == end, so the grid has one point for any
+		// positive step.
+		stepSec = 1
+	}
+	windowMs := unit.rangeMs
+	if unit.kind == unitInstant {
+		windowMs = lookbackMs
+	}
+	windowSec := windowMs / 1000
+
+	adjustedTsStart, tsTable := timeSeriesTableFor(dataStart, dataEnd)
+
+	// seriesSub computes fingerprint -> group key. It reads the local series
+	// table when it rides inside the shard-rewritten samples query, and the
+	// distributed one when it joins at the initiator (windowed form).
+	seriesSub := func(table string) (string, []any, error) {
+		sub := sqlbuilder.NewSelectBuilder()
+		sub.Select("fingerprint", groupKeyExpr(unit)+" AS gkey")
+		sub.From(fmt.Sprintf("%s.%s", databaseName, table))
+		if err := applySeriesConditions(sub, adjustedTsStart, dataEnd, unit.matchers); err != nil {
+			return "", nil, err
+		}
+		sub.GroupBy("fingerprint", "gkey")
+		q, args := sub.BuildWithFlavor(sqlbuilder.ClickHouse)
+		return q, args, nil
+	}
+
+	// samplesConditions adds the samples-side WHERE. The samples table is
+	// aliased "points" in every kind: under the group-key join both sides
+	// carry a fingerprint column, so the filter must qualify it. A nil
+	// inline set adds no fingerprint condition — the join restricts.
+	samplesConditions := func(sb *sqlbuilder.SelectBuilder, excludeStale bool) {
+		switch len(metricNames) {
+		case 0:
+			// No name constraint derivable; correct but unable to use the
+			// metric_name primary-key prefix.
+		case 1:
+			sb.Where(sb.EQ("metric_name", metricNames[0]))
+		default:
+			sb.Where(sb.In("metric_name", sqlbuilder.List(metricNames)))
+		}
+		// temporality precedes metric_name in the samples primary key; the
+		// fingerprints already come from these temporalities, so this only
+		// helps granule pruning.
+		sb.Where("temporality IN ['Cumulative', 'Unspecified']")
+		if inlineFingerprints != nil {
+			sb.Where("points.fingerprint " + inlineFingerprintFilter(inlineFingerprints))
+		}
+		// Left-open window: a sample exactly at the window's lower boundary
+		// is never used (range selectors and lookback are both left-open).
+		sb.Where(sb.GT("unix_milli", selStart-windowMs), sb.LTE("unix_milli", selEnd))
+		if excludeStale {
+			// PromQL excludes stale markers from range vectors. Instant
+			// selectors need the stale rows for shadowing instead.
+			sb.Where("bitAnd(flags, 1) = 0")
+		}
+	}
+
+	// joinedInner builds the shard-side SELECT for the single-pass kinds:
+	// grid expression per (fingerprint, gkey), group-key join against the
+	// local series table.
+	joinedInner := func(gridExpr string, excludeStale bool) (string, []any, error) {
+		seriesSQL, seriesArgs, err := seriesSub(localTimeSeriesTable(tsTable))
+		if err != nil {
+			return "", nil, err
+		}
+		sb := sqlbuilder.NewSelectBuilder()
+		sb.Select("series.gkey AS gkey", gridExpr+" AS grid")
+		sb.From(fmt.Sprintf("%s.%s AS points", databaseName, distributedSamplesV4))
+		sb.JoinWithOption(sqlbuilder.InnerJoin, fmt.Sprintf("(%s) AS series", seriesSQL), "points.fingerprint = series.fingerprint")
+		samplesConditions(sb, excludeStale)
+		sb.GroupBy("points.fingerprint", "series.gkey")
+		q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+		// The join text renders before WHERE: its args come first.
+		return q, append(seriesArgs, args...), nil
+	}
+
+	var inner string
+	var innerArgs []any
+	var err error
+	switch unit.kind {
+	case unitInstant:
+		// Instant selection with stale shadowing: the grid value is the last
+		// non-stale sample in (t-lookback, t], absent when the overall last
+		// sample in that window is a stale marker (verified semantics: the
+		// -If combinator applies to the grid aggregates, and NULL comparisons
+		// make a stale-latest point absent).
+		gridParams := fmt.Sprintf("(fromUnixTimestamp64Milli(%d), fromUnixTimestamp64Milli(%d), %d, %d)", selStart, selEnd, stepSec, windowSec)
+		gridExpr := fmt.Sprintf(
+			"arrayMap((tall, tok, vok) -> if(tall IS NULL OR tok IS NULL OR tall != tok, NULL, vok), timeSeriesLastToGrid%s(fromUnixTimestamp64Milli(unix_milli), toFloat64(unix_milli)), timeSeriesLastToGridIf%s(fromUnixTimestamp64Milli(unix_milli), toFloat64(unix_milli), bitAnd(flags, 1) = 0), timeSeriesLastToGridIf%s(fromUnixTimestamp64Milli(unix_milli), value, bitAnd(flags, 1) = 0))",
+			gridParams, gridParams, gridParams,
+		)
+		inner, innerArgs, err = joinedInner(gridExpr, false)
+	case unitOverTime:
+		if unit.overFn == "last" {
+			// last_over_time == last non-stale sample in the window: the
+			// stale rows are already excluded in WHERE.
+			gridExpr := fmt.Sprintf(
+				"timeSeriesLastToGrid(fromUnixTimestamp64Milli(%d), fromUnixTimestamp64Milli(%d), %d, %d)(fromUnixTimestamp64Milli(unix_milli), value)",
+				selStart, selEnd, stepSec, windowSec,
+			)
+			inner, innerArgs, err = joinedInner(gridExpr, true)
+			break
+		}
+		inner, innerArgs, err = windowedInner(unit, samplesConditions, seriesSub, inlineFingerprints == nil, adjustedTsStart, dataEnd, tsTable, selStart, selEnd, stepMs, windowMs)
+	default: // unitRange
+		gridExpr := fmt.Sprintf(
+			"%s(fromUnixTimestamp64Milli(%d), fromUnixTimestamp64Milli(%d), %d, %d)(fromUnixTimestamp64Milli(unix_milli), value)",
+			gridFunction[unit.fn], selStart, selEnd, stepSec, windowSec,
+		)
+		if unit.fn == fnIncrease {
+			// increase == rate * range-seconds, exactly: extrapolatedRate
+			// divides by the range only when isRate.
+			gridExpr = fmt.Sprintf("arrayMap(x -> x * %d, %s)", windowSec, gridExpr)
+		}
+		inner, innerArgs, err = joinedInner(gridExpr, true)
+	}
+	if err != nil {
+		return "", nil, err
+	}
+
+	spatial := "maxForEach(grid)"
+	switch {
+	case !unit.hasAgg:
+		// Per-series output: one row per (labels-minus-__name__) group.
+		// Distinct fingerprints can collapse onto the same projected label
+		// set only via a regex __name__ selector over metrics with identical
+		// other labels; maxForEach is a deterministic NULL-skipping merge and
+		// the identity for the overwhelmingly common one-fingerprint group.
+	case unit.aggOp.String() == "count":
+		// count over an all-absent index is an absent point, not 0.
+		spatial = "arrayMap(c -> if(c = 0, NULL, toFloat64(c)), countForEach(grid))"
+	default:
+		spatial = fmt.Sprintf("%s(grid)", aggForEach[unit.aggOp.String()])
+	}
+
+	query := fmt.Sprintf("SELECT gkey, %s AS grid FROM (%s) GROUP BY gkey %s", spatial, inner, gridFunctionsSetting)
+	return query, innerArgs, nil
+}
+
+// windowedInner builds the avg/min/max/sum/count _over_time form: each
+// sample fans out to every grid index k whose window (t_k - range, t_k]
+// contains it (ARRAY JOIN), aggregates per (fingerprint, k) — shard-side
+// partials over the distributed table — then assembles the positional grid
+// and joins the group key at the initiator over the reduced rows. The
+// group-key subquery reads the distributed series table here because it does
+// not ride inside a shard-rewritten query.
+//
+// This is the one form whose samples query has no series join, so an
+// over-the-limit fingerprint set (semiJoin) must fall back to the
+// shard-local semi-join: without it the fan-out would expand every series of
+// the metric and discard the unmatched ones only at the group-key join.
+func windowedInner(unit *coreUnit, samplesConditions func(*sqlbuilder.SelectBuilder, bool), seriesSub func(string) (string, []any, error), semiJoin bool, adjustedTsStart, dataEnd int64, tsTable string, selStart, selEnd, stepMs, windowMs int64) (string, []any, error) {
+	aggExpr := map[string]string{
+		"avg":   "avg(value)",
+		"min":   "min(value)",
+		"max":   "max(value)",
+		"sum":   "sum(value)",
+		"count": "toFloat64(count(value))",
+	}[unit.overFn]
+	effStepMs := stepMs
+	if effStepMs == 0 {
+		effStepMs = 1000
+	}
+	lastIdx := (selEnd - selStart) / effStepMs
+
+	perWindow := sqlbuilder.NewSelectBuilder()
+	perWindow.Select("fingerprint", "k", aggExpr+" AS v")
+	perWindow.From(fmt.Sprintf(
+		"%s.%s AS points ARRAY JOIN range(toUInt64(greatest(0, intDiv(unix_milli - %d + %d - 1, %d))), toUInt64(least(%d, intDiv(unix_milli + %d - 1 - %d, %d)) + 1)) AS k",
+		databaseName, distributedSamplesV4,
+		selStart, effStepMs, effStepMs,
+		lastIdx, windowMs, selStart, effStepMs,
+	))
+	samplesConditions(perWindow, true)
+	if semiJoin {
+		sub := sqlbuilder.NewSelectBuilder()
+		sub.Select("fingerprint")
+		sub.From(fmt.Sprintf("%s.%s", databaseName, localTimeSeriesTable(tsTable)))
+		if err := applySeriesConditions(sub, adjustedTsStart, dataEnd, unit.matchers); err != nil {
+			return "", nil, err
+		}
+		perWindow.Where(perWindow.In("points.fingerprint", sub))
+	}
+	perWindow.GroupBy("fingerprint", "k")
+	perWindowSQL, perWindowArgs := perWindow.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	grids := fmt.Sprintf(
+		"SELECT fingerprint, arrayMap(i -> if(indexOf(ks, i) = 0, NULL, vs[indexOf(ks, i)]), range(toUInt64(%d))) AS grid FROM (SELECT fingerprint, groupArray(k) AS ks, groupArray(v) AS vs FROM (%s) GROUP BY fingerprint)",
+		lastIdx+1, perWindowSQL,
+	)
+
+	seriesSQL, seriesArgs, err := seriesSub(tsTable)
+	if err != nil {
+		return "", nil, err
+	}
+	inner := fmt.Sprintf(
+		"SELECT series.gkey AS gkey, points.grid AS grid FROM (%s) AS points INNER JOIN (%s) AS series ON points.fingerprint = series.fingerprint",
+		grids, seriesSQL,
+	)
+	return inner, append(perWindowArgs, seriesArgs...), nil
+}
+
+// groupKeyExpr renders the canonical group key for a unit: the sorted
+// [key, value] pairs of the projected labels, JSON-encoded.
+//   - by (a, b): keep only the listed labels (absent labels stay absent,
+//     matching PromQL's by() over missing labels);
+//   - without (a, b): keep everything except the listed labels and __name__;
+//   - no aggregation: keep everything including __name__ — even when the
+//     unit drops the name from its OUTPUT, the key must keep it so distinct
+//     metrics never merge in SQL; executeUnit strips the name afterwards and
+//     turns a post-strip collision into the engine's duplicate-labelset
+//     error instead of a silently invented merge.
+func groupKeyExpr(unit *coreUnit) string {
+	// An empty label value means "label absent" in Prometheus; the stored
+	// labels JSON can carry empty attribute values, which must not become
+	// output labels or group keys.
+	pairs := "arraySort(JSONExtractKeysAndValues(labels, 'String'))"
+	if !unit.hasAgg {
+		return fmt.Sprintf("toJSONString(arrayFilter(p -> p.2 != '', %s))", pairs)
+	}
+	if unit.by {
+		if len(unit.grouping) == 0 {
+			return "'[]'"
+		}
+		return fmt.Sprintf("toJSONString(arrayFilter(p -> p.2 != '' AND p.1 IN (%s), %s))", quotedList(unit.grouping), pairs)
+	}
+	excluded := append([]string{metricNameLabel}, unit.grouping...)
+	return fmt.Sprintf("toJSONString(arrayFilter(p -> p.2 != '' AND p.1 NOT IN (%s), %s))", quotedList(excluded), pairs)
+}
+
+func quotedList(items []string) string {
+	quoted := make([]string, len(items))
+	for i, s := range items {
+		quoted[i] = "'" + strings.ReplaceAll(s, "'", "\\'") + "'"
+	}
+	return strings.Join(quoted, ", ")
+}

pkg/prometheus/clickhouseprometheusv2/transpiler_test.go

@@ -0,0 +1,517 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	cmock "github.com/SigNoz/clickhouse-go-mock"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql/parser"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func parse(t *testing.T, q string) parser.Expr {
+	t.Helper()
+	expr, err := parser.NewParser(parser.Options{}).ParseExpr(q)
+	require.NoError(t, err)
+	return expr
+}
+
+func TestClassifyFullShapes(t *testing.T) {
+	tests := []struct {
+		name  string
+		query string
+		check func(t *testing.T, u *coreUnit)
+	}{
+		{
+			name:  "sum by rate",
+			query: `sum by (pod) (rate(http_requests_total{job="api"}[5m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, fnRate, u.fn)
+				assert.Equal(t, int64(300_000), u.rangeMs)
+				assert.True(t, u.hasAgg)
+				assert.True(t, u.by)
+				assert.Equal(t, []string{"pod"}, u.grouping)
+			},
+		},
+		{
+			name:  "bare increase with offset",
+			query: `increase(errors_total[10m] offset 30m)`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, fnIncrease, u.fn)
+				assert.Equal(t, int64(1_800_000), u.offsetMs)
+				assert.False(t, u.hasAgg)
+			},
+		},
+		{
+			name:  "avg without over delta",
+			query: `avg without (instance) (delta(gauge_metric[15m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, fnDelta, u.fn)
+				assert.True(t, u.hasAgg)
+				assert.False(t, u.by)
+			},
+		},
+		{
+			name:  "scalar pipeline with comparison",
+			query: `sum(rate(x[5m])) * 100 > 5`,
+			check: func(t *testing.T, u *coreUnit) {
+				require.Len(t, u.ops, 2)
+				assert.Equal(t, parser.ItemType(parser.MUL), u.ops[0].op)
+				assert.Equal(t, 100.0, u.ops[0].scalar)
+				assert.Equal(t, parser.ItemType(parser.GTR), u.ops[1].op)
+			},
+		},
+		{
+			name:  "scalar on left with unary minus",
+			query: `-1 * sum(rate(x[5m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				require.Len(t, u.ops, 1)
+				assert.True(t, u.ops[0].scalarOnLeft)
+				assert.Equal(t, -1.0, u.ops[0].scalar)
+			},
+		},
+		{
+			name:  "bool comparison",
+			query: `sum(rate(x[5m])) >= bool 0.5`,
+			check: func(t *testing.T, u *coreUnit) {
+				require.Len(t, u.ops, 1)
+				assert.True(t, u.ops[0].returnBool)
+			},
+		},
+		{
+			name:  "irate utf8 name",
+			query: `sum by ("k8s.pod.name") (irate({"k8s.container.cpu.time"}[2m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, fnIRate, u.fn)
+				assert.Equal(t, []string{"k8s.pod.name"}, u.grouping)
+			},
+		},
+		{
+			name:  "bare instant selector keeps name",
+			query: `up{job="api"}`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitInstant, u.kind)
+				assert.True(t, u.keepsName())
+			},
+		},
+		{
+			name:  "gauge aggregation",
+			query: `sum by (pod) (container_memory offset 5m)`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitInstant, u.kind)
+				assert.Equal(t, int64(300_000), u.offsetMs)
+				assert.True(t, u.hasAgg)
+				assert.False(t, u.keepsName())
+			},
+		},
+		{
+			name:  "gauge comparison keeps name",
+			query: `container_memory > 100`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitInstant, u.kind)
+				assert.True(t, u.keepsName())
+			},
+		},
+		{
+			name:  "gauge arithmetic drops name",
+			query: `container_memory / 1024`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitInstant, u.kind)
+				assert.False(t, u.keepsName())
+			},
+		},
+		{
+			name:  "avg_over_time",
+			query: `max by (node) (avg_over_time(load1[10m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitOverTime, u.kind)
+				assert.Equal(t, "avg", u.overFn)
+				assert.Equal(t, int64(600_000), u.rangeMs)
+			},
+		},
+		{
+			name:  "last_over_time keeps name",
+			query: `last_over_time(load1[10m])`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitOverTime, u.kind)
+				assert.Equal(t, "last", u.overFn)
+				assert.True(t, u.keepsName())
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			plan, ok := classify(parse(t, tt.query), testGrid(60_000))
+			require.True(t, ok, "expected transpilable")
+			require.True(t, plan.full, "expected full compilation")
+			require.Len(t, plan.units, 1)
+			tt.check(t, &plan.units[0].core)
+		})
+	}
+}
+
+func TestClassifyFallbackShapes(t *testing.T) {
+	queries := []struct {
+		name  string
+		query string
+		step  int64
+	}{
+		{"default-resolution subquery", `max_over_time(rate(x[5m])[30m:])`, 60_000},
+		{"at modifier", `sum(rate(x[5m] @ 1609746000))`, 60_000},
+		{"at modifier on gauge", `sum(container_memory @ 1609746000)`, 60_000},
+		{"sub-second step", `sum(rate(x[5m]))`, 500},
+		{"sub-second range", `sum(rate(x[1500ms]))`, 60_000},
+		{"by __name__ full", `sum by (__name__) (rate({__name__=~"a|b"}[5m]))`, 60_000},
+		{"quantile_over_time unsupported", `quantile_over_time(0.9, load1[10m])`, 60_000},
+	}
+	for _, tt := range queries {
+		t.Run(tt.name, func(t *testing.T) {
+			_, ok := classify(parse(t, tt.query), testGrid(tt.step))
+			assert.False(t, ok, "expected fallback for %s", tt.query)
+		})
+	}
+}
+
+func TestClassifyHybridShapes(t *testing.T) {
+	tests := []struct {
+		name          string
+		query         string
+		wantUnits     int
+		wantRewritten string
+	}{
+		{
+			name:          "histogram quantile",
+			query:         `histogram_quantile(0.95, sum by (le) (rate(http_bucket[5m])))`,
+			wantUnits:     1,
+			wantRewritten: `histogram_quantile(0.95, __signoz_transpiled_0__)`,
+		},
+		{
+			name:          "topk over compiled",
+			query:         `topk(5, sum by (pod) (rate(x[5m])))`,
+			wantUnits:     1,
+			wantRewritten: `topk(5, __signoz_transpiled_0__)`,
+		},
+		{
+			name:          "ratio of compiled units",
+			query:         `sum(rate(a[5m])) / sum(rate(b[5m]))`,
+			wantUnits:     2,
+			wantRewritten: `__signoz_transpiled_0__ / __signoz_transpiled_1__`,
+		},
+		{
+			name:          "or vector zero",
+			query:         `sum(rate(a[5m])) or vector(0)`,
+			wantUnits:     1,
+			wantRewritten: `__signoz_transpiled_0__ or vector(0)`,
+		},
+		{
+			name:          "quantile agg over compiled rate",
+			query:         `quantile(0.9, rate(x[5m]))`,
+			wantUnits:     1,
+			wantRewritten: `quantile(0.9, __signoz_transpiled_0__)`,
+		},
+		{
+			name:          "non-literal scalar side stays engine-side",
+			query:         `sum(rate(x[5m])) * scalar(y)`,
+			wantUnits:     1,
+			wantRewritten: `__signoz_transpiled_0__ * scalar(y)`,
+		},
+		{
+			name:          "compiled mixed with raw selector",
+			query:         `sum by (pod) (rate(a[5m])) / on (pod) group_left () b`,
+			wantUnits:     1,
+			wantRewritten: `__signoz_transpiled_0__ / on (pod) group_left () b`,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			plan, ok := classify(parse(t, tt.query), testGrid(60_000))
+			require.True(t, ok)
+			assert.False(t, plan.full)
+			assert.Len(t, plan.units, tt.wantUnits)
+			assert.Equal(t, tt.wantRewritten, plan.rewritten)
+		})
+	}
+}
+
+func TestClassifyHybridGuards(t *testing.T) {
+	t.Run("no substitution under on(__name__)", func(t *testing.T) {
+		plan, ok := classify(parse(t, `sum(rate(a[5m])) * on (__name__) b`), testGrid(60_000))
+		_ = plan
+		assert.False(t, ok, "matching on __name__ must not see synthetic names")
+	})
+	t.Run("no substitution inside @-pinned subquery", func(t *testing.T) {
+		_, ok := classify(parse(t, `max_over_time(rate(x[5m])[30m:1m] @ 1609746000)`), testGrid(60_000))
+		assert.False(t, ok)
+	})
+}
+
+// The alert-smoothing idiom: units inside a fixed-resolution subquery
+// evaluate on the subquery grid — epoch-aligned multiples of the resolution,
+// starting strictly after (outer start - range), exactly as the engine
+// derives it.
+func TestClassifySubqueryUnits(t *testing.T) {
+	grid := gridContext{startMs: 1_700_000_030_000, endMs: 1_700_007_200_000, stepMs: 60_000}
+
+	plan, ok := classify(parse(t, `min_over_time((sum by (ns) (increase(x[5m])))[10m:5m]) > 0`), grid)
+	require.True(t, ok)
+	require.False(t, plan.full)
+	require.Len(t, plan.units, 1)
+	assert.Equal(t, `min_over_time(__signoz_transpiled_0__[10m:5m]) > 0`, plan.rewritten)
+
+	unit := plan.units[0]
+	// lower bound = outer start - range = 1_699_999_430_000; first multiple
+	// of 300_000 strictly greater is 1_699_999_500_000.
+	assert.Equal(t, int64(1_699_999_500_000), unit.grid.startMs)
+	assert.Equal(t, grid.endMs, unit.grid.endMs)
+	assert.Equal(t, int64(300_000), unit.grid.stepMs)
+	assert.Equal(t, fnIncrease, unit.core.fn)
+
+	t.Run("subquery offset shifts the grid", func(t *testing.T) {
+		plan, ok := classify(parse(t, `max_over_time((sum(rate(x[5m])))[10m:5m] offset 30m)`), grid)
+		require.True(t, ok)
+		require.Len(t, plan.units, 1)
+		// lower = start - offset - range = 1_699_997_630_000 -> first
+		// multiple of 300_000 above = 1_699_997_700_000; end shifts too.
+		assert.Equal(t, int64(1_699_997_700_000), plan.units[0].grid.startMs)
+		assert.Equal(t, grid.endMs-1_800_000, plan.units[0].grid.endMs)
+	})
+
+	t.Run("mollusk ratio-inside-subquery idiom", func(t *testing.T) {
+		q := `min_over_time(((sum by (a) (rate(m1[5m]))) / (avg by (a) (m2)))[5m:1m])`
+		plan, ok := classify(parse(t, q), grid)
+		require.True(t, ok)
+		// Both sides compile on the subquery grid: the rate side and the
+		// gauge aggregation side; the engine joins them and smooths.
+		require.Len(t, plan.units, 2)
+		assert.Equal(t, int64(60_000), plan.units[0].grid.stepMs)
+		assert.Equal(t, unitInstant, plan.units[1].core.kind)
+		assert.Contains(t, plan.rewritten, `__signoz_transpiled_0__ / __signoz_transpiled_1__`)
+	})
+}
+
+func TestBuildUnitSQL(t *testing.T) {
+	unit := &coreUnit{
+		fn:       fnRate,
+		rangeMs:  300_000,
+		hasAgg:   true,
+		aggOp:    parser.SUM,
+		by:       true,
+		grouping: []string{"pod"},
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "http_requests_total")},
+	}
+	sql, args, err := buildUnitSQL(unit, []string{"http_requests_total"}, []uint64{7, 42}, 1_699_999_700_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 300_000)
+	require.NoError(t, err)
+
+	assert.Contains(t, sql, "timeSeriesRateToGrid(fromUnixTimestamp64Milli(1700000000000), fromUnixTimestamp64Milli(1700003600000), 60, 300)(fromUnixTimestamp64Milli(unix_milli), value)")
+	assert.Contains(t, sql, "unix_milli > ? AND unix_milli <= ?")
+	assert.Contains(t, sql, "bitAnd(flags, 1) = 0")
+	assert.Contains(t, sql, "sumForEach(grid)")
+	// The group-key join rides inside the shard query: distributed samples
+	// at the top level, the local series table in the join subquery, the
+	// grid aggregation grouped per (fingerprint, gkey) shard-side.
+	assert.Contains(t, sql, "FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint,")
+	assert.Contains(t, sql, "FROM signoz_metrics.time_series_v4 WHERE")
+	assert.Contains(t, sql, "GROUP BY points.fingerprint, series.gkey")
+	assert.Contains(t, sql, "points.fingerprint IN (7, 42)")
+	assert.Contains(t, sql, `toJSONString(arrayFilter(p -> p.2 != '' AND p.1 IN ('pod'),`)
+	assert.Contains(t, sql, "SETTINGS allow_experimental_ts_to_grid_aggregate_function = 1")
+	// Args follow placeholder order: the joined series subquery renders
+	// before the samples WHERE.
+	assert.Equal(t, []any{"http_requests_total", int64(1_699_999_200_000), int64(1_700_003_600_000), "http_requests_total", int64(1_699_999_700_000), int64(1_700_003_600_000)}, args)
+}
+
+func TestBuildUnitSQLIncreaseAndOffset(t *testing.T) {
+	unit := &coreUnit{
+		fn:       fnIncrease,
+		rangeMs:  600_000,
+		offsetMs: 1_800_000,
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "errors_total")},
+	}
+	sql, _, err := buildUnitSQL(unit, nil, []uint64{7}, 1_699_997_600_000, 1_700_001_800_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 300_000)
+	require.NoError(t, err)
+
+	// Grid and window shift by the offset; increase multiplies rate by the
+	// range in seconds.
+	assert.Contains(t, sql, "fromUnixTimestamp64Milli(1699998200000), fromUnixTimestamp64Milli(1700001800000)")
+	assert.Contains(t, sql, "arrayMap(x -> x * 600, timeSeriesRateToGrid")
+	assert.Contains(t, sql, "maxForEach(grid)")
+}
+
+func TestBuildUnitSQLOverLimitJoinOnly(t *testing.T) {
+	// Past the inline limit no fingerprint filter is rendered: the series
+	// join restricts to the matched fingerprints on its own.
+	unit := &coreUnit{
+		fn:       fnRate,
+		rangeMs:  300_000,
+		hasAgg:   true,
+		aggOp:    parser.SUM,
+		by:       true,
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "http_requests_total")},
+	}
+	sql, _, err := buildUnitSQL(unit, []string{"http_requests_total"}, nil, 1_699_999_700_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 300_000)
+	require.NoError(t, err)
+
+	assert.NotContains(t, sql, "points.fingerprint IN")
+	assert.Contains(t, sql, "INNER JOIN (SELECT fingerprint,")
+	assert.Contains(t, sql, "FROM signoz_metrics.time_series_v4 WHERE")
+}
+
+func TestBuildUnitSQLOverLimitWindowedSemiJoin(t *testing.T) {
+	// The windowed *_over_time fan-out has no series join, so the over-limit
+	// regime falls back to the shard-local semi-join instead of expanding
+	// every series of the metric.
+	unit := &coreUnit{
+		kind:     unitOverTime,
+		overFn:   "avg",
+		rangeMs:  600_000,
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "node_load1")},
+	}
+	sql, _, err := buildUnitSQL(unit, []string{"node_load1"}, nil, 1_699_999_400_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 300_000)
+	require.NoError(t, err)
+
+	assert.Contains(t, sql, "points.fingerprint IN (SELECT fingerprint FROM signoz_metrics.time_series_v4 WHERE ")
+	assert.Contains(t, sql, "ARRAY JOIN range(")
+}
+
+func TestApplyScalarOps(t *testing.T) {
+	f := func(v float64) *float64 { return &v }
+
+	t.Run("arithmetic chain", func(t *testing.T) {
+		values := []*float64{f(2), nil, f(4)}
+		applyScalarOps([]scalarOp{{op: parser.MUL, scalar: 100}, {op: parser.ADD, scalar: 1}}, values)
+		require.NotNil(t, values[0])
+		assert.Equal(t, 201.0, *values[0])
+		assert.Nil(t, values[1])
+		assert.Equal(t, 401.0, *values[2])
+	})
+
+	t.Run("comparison filters points", func(t *testing.T) {
+		values := []*float64{f(1), f(10)}
+		applyScalarOps([]scalarOp{{op: parser.GTR, scalar: 5}}, values)
+		assert.Nil(t, values[0])
+		require.NotNil(t, values[1])
+		assert.Equal(t, 10.0, *values[1], "filter comparisons keep the original value")
+	})
+
+	t.Run("bool comparison emits 0/1", func(t *testing.T) {
+		values := []*float64{f(1), f(10)}
+		applyScalarOps([]scalarOp{{op: parser.GTR, scalar: 5, returnBool: true}}, values)
+		assert.Equal(t, 0.0, *values[0])
+		assert.Equal(t, 1.0, *values[1])
+	})
+
+	t.Run("scalar on left division", func(t *testing.T) {
+		values := []*float64{f(4)}
+		applyScalarOps([]scalarOp{{op: parser.DIV, scalar: 100, scalarOnLeft: true}}, values)
+		assert.Equal(t, 25.0, *values[0])
+	})
+}
+
+func TestLabelsFromGroupKey(t *testing.T) {
+	lset, err := labelsFromGroupKey(`[["pod","api-0"],["ns","prod"]]`)
+	require.NoError(t, err)
+	assert.Equal(t, "api-0", lset.Get("pod"))
+	assert.Equal(t, "prod", lset.Get("ns"))
+
+	empty, err := labelsFromGroupKey(`[]`)
+	require.NoError(t, err)
+	assert.True(t, empty.IsEmpty())
+}
+
+// testGrid is a 2h query grid ending on a round timestamp.
+func testGrid(stepMs int64) gridContext {
+	return gridContext{startMs: 1_700_000_000_000, endMs: 1_700_007_200_000, stepMs: stepMs}
+}
+
+// A bool comparison returns 0/1, not the sample, so the engine drops
+// __name__; keeping it would change downstream vector matching.
+func TestKeepsName_BoolComparisonDropsName(t *testing.T) {
+	plan, ok := classify(parse(t, `up > bool 0`), testGrid(60_000))
+	require.True(t, ok)
+	assert.False(t, plan.units[0].core.keepsName())
+
+	plan, ok = classify(parse(t, `up > 0`), testGrid(60_000))
+	require.True(t, ok)
+	assert.True(t, plan.units[0].core.keepsName())
+}
+
+// timeSeriesLastToGrid widens its window to max(window, step) — probed on
+// 25.12 — so Last-style units at window < step must fall back or they would
+// resurrect samples the engine's lookback already dropped.
+func TestTryExecuteRange_LastStyleWindowBelowStepFallsBack(t *testing.T) {
+	c, _ := newTestClient(t, prometheus.ClickhouseV2Config{})
+	e := &executor{client: c, parser: prometheus.NewParser()}
+
+	start := time.UnixMilli(1_700_000_000_000)
+	end := time.UnixMilli(1_700_003_600_000)
+
+	_, ok, err := e.TryExecuteRange(context.Background(), `sum by (pod) (up)`, start, end, time.Hour)
+	require.NoError(t, err)
+	assert.False(t, ok, "instant selection at step > lookback must not transpile")
+
+	_, ok, err = e.TryExecuteRange(context.Background(), `last_over_time(up[10m])`, start, end, time.Hour)
+	require.NoError(t, err)
+	assert.False(t, ok, "last_over_time at range < step must not transpile")
+}
+
+// Transpiled results never pass the engine's sample limiter, so the grid
+// cells (series x grid width) must be budgeted before the arrays exist —
+// otherwise a wide query rebuilds the OOM this provider exists to prevent.
+func TestExecuteUnit_GridCellBudget(t *testing.T) {
+	c, store := newTestClient(t, prometheus.ClickhouseV2Config{MaxFetchedSamples: 100})
+	e := &executor{client: c, parser: prometheus.NewParser()}
+
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("up", int64(1_699_999_200_000), int64(1_700_003_600_000)).WillReturnRows(cmock.NewRows(seriesCols, [][]any{
+		{uint64(1), `{"__name__":"up","instance":"a"}`},
+		{uint64(2), `{"__name__":"up","instance":"b"}`},
+	}))
+
+	plan, ok := classify(parse(t, `sum(rate(up[5m]))`), gridContext{startMs: 1_700_000_000_000, endMs: 1_700_003_600_000, stepMs: 60_000})
+	require.True(t, ok)
+
+	var cells atomic.Int64
+	// 2 series x 61 grid points = 122 cells > 100.
+	_, err := e.executeUnit(context.Background(), &plan.units[0].core, plan.units[0].grid, &cells)
+	require.Error(t, err)
+	assert.True(t, errors.Ast(err, errors.TypeInvalidInput), "budget refusal must be typed invalid input, got %v", err)
+}
+
+// Two metrics collapsing onto one labelset after the name drop is the
+// engine's duplicate-labelset error; silently merging them would invent a
+// series no engine would produce.
+func TestExecuteUnit_NameCollisionErrors(t *testing.T) {
+	c, store := newTestClient(t, prometheus.ClickhouseV2Config{})
+	e := &executor{client: c, parser: prometheus.NewParser()}
+
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("^(?:a|b)$", int64(1_699_999_200_000), int64(1_700_003_600_000)).WillReturnRows(cmock.NewRows(seriesCols, [][]any{
+		{uint64(1), `{"__name__":"a","job":"x"}`},
+		{uint64(2), `{"__name__":"b","job":"x"}`},
+	}))
+	store.Mock().ExpectQuery("SELECT gkey").
+		WithArgs("^(?:a|b)$", int64(1_699_999_200_000), int64(1_700_003_600_000), "a", "b", int64(1_699_999_700_000), int64(1_700_003_600_000)).
+		WillReturnRows(cmock.NewRows(gkeyCols, [][]any{
+			{`[["__name__","a"],["job","x"]]`, []*float64{f64(1)}},
+			{`[["__name__","b"],["job","x"]]`, []*float64{f64(2)}},
+		}))
+
+	plan, ok := classify(parse(t, `rate({__name__=~"a|b"}[5m])`), gridContext{startMs: 1_700_000_000_000, endMs: 1_700_003_600_000, stepMs: 60_000})
+	require.True(t, ok)
+
+	var cells atomic.Int64
+	_, err := e.executeUnit(context.Background(), &plan.units[0].core, plan.units[0].grid, &cells)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "vector cannot contain metrics with the same labelset")
+}
+
+var gkeyCols = []cmock.ColumnType{
+	{Name: "gkey", Type: "String"},
+	{Name: "grid", Type: "Array(Nullable(Float64))"},
+}
+
+func f64(v float64) *float64 { return &v }

pkg/prometheus/config.go

@@ -13,6 +13,11 @@ type ActiveQueryTrackerConfig struct {
 	MaxConcurrent int    `mapstructure:"max_concurrent"`
 }
 
+type ClickhouseV2Config struct {
+	MaxFetchedSeries  int   `mapstructure:"max_fetched_series"`
+	MaxFetchedSamples int64 `mapstructure:"max_fetched_samples"`
+}
+
 type Config struct {
 	ActiveQueryTrackerConfig ActiveQueryTrackerConfig `mapstructure:"active_query_tracker"`
 
@@ -24,6 +29,13 @@ type Config struct {
 
 	// Timeout is the maximum time a query is allowed to run before being aborted.
 	Timeout time.Duration `mapstructure:"timeout"`
+
+	// ProviderName selects the storage provider: "clickhouse" (default) or
+	// "clickhousev2".
+	ProviderName string `mapstructure:"provider"`
+
+	// ClickhouseV2 configures the clickhousev2 provider.
+	ClickhouseV2 ClickhouseV2Config `mapstructure:"clickhousev2"`
 }
 
 func NewConfigFactory() factory.ConfigFactory {
@@ -37,7 +49,12 @@ func newConfig() factory.Config {
 			Path:          "",
 			MaxConcurrent: 20,
 		},
-		Timeout: 2 * time.Minute,
+		Timeout:      2 * time.Minute,
+		ProviderName: "clickhouse",
+		ClickhouseV2: ClickhouseV2Config{
+			MaxFetchedSeries:  500_000,
+			MaxFetchedSamples: 50_000_000,
+		},
 	}
 }
 
@@ -45,9 +62,18 @@ func (c Config) Validate() error {
 	if c.Timeout <= 0 {
 		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "prometheus::timeout must be greater than 0")
 	}
+	if c.ProviderName != "" && c.ProviderName != "clickhouse" && c.ProviderName != "clickhousev2" {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "prometheus::provider must be one of [clickhouse, clickhousev2], got %q", c.ProviderName)
+	}
+	if c.ClickhouseV2.MaxFetchedSeries < 0 || c.ClickhouseV2.MaxFetchedSamples < 0 {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "prometheus::clickhousev2 limits must not be negative")
+	}
 	return nil
 }
 
 func (c Config) Provider() string {
-	return "clickhouse"
+	if c.ProviderName == "" {
+		return "clickhouse"
+	}
+	return c.ProviderName
 }

pkg/prometheus/prometheus.go

@@ -35,3 +35,9 @@ type StatementRecorder interface {
 type StatementCapturer interface {
 	CapturingStorage() (storage.Queryable, StatementRecorder)
 }
+
+// ProviderClickhouseV2 is the clickhousev2 provider name: the factory
+// registration, the prometheus::provider config value and the
+// X-SigNoz-PromQL-Provider request header all use it, so they cannot drift
+// apart.
+const ProviderClickhouseV2 = "clickhousev2"

pkg/prometheus/traits.go

@@ -0,0 +1,49 @@
+package prometheus
+
+import (
+	"context"
+
+	"github.com/prometheus/prometheus/promql/parser"
+)
+
+type queryTraitsKey struct{}
+
+// QueryTraits carries per-query facts a storage implementation cannot derive
+// from SelectHints alone. Call sites that parse the PromQL expression attach
+// traits to the context before handing it to the engine; storages treat a
+// missing traits value as "unknown" and stay conservative.
+type QueryTraits struct {
+	// SubqueryFree is true when the query contains no subquery expression.
+	// Subquery selectors are evaluated at the subquery's own step, but
+	// SelectHints.Step always carries the top-level step, so step-aligned
+	// storage optimizations (e.g. keeping only the last sample per step
+	// bucket) are safe only when this is true.
+	SubqueryFree bool
+}
+
+// DetectQueryTraits derives QueryTraits from a parsed PromQL expression.
+func DetectQueryTraits(expr parser.Expr) QueryTraits {
+	subqueryFree := true
+	parser.Inspect(expr, func(node parser.Node, _ []parser.Node) error {
+		if _, ok := node.(*parser.SubqueryExpr); ok {
+			subqueryFree = false
+		}
+		return nil
+	})
+	return QueryTraits{SubqueryFree: subqueryFree}
+}
+
+// NewContextWithQueryTraits returns a context carrying the given traits.
+func NewContextWithQueryTraits(ctx context.Context, traits QueryTraits) context.Context {
+	return context.WithValue(ctx, queryTraitsKey{}, traits)
+}
+
+// QueryTraitsFromContext returns the traits attached to ctx, if any.
+//
+// Context is used here, unlike for backend selection, because traits must
+// cross the promql engine to reach storage.Querier.Select, and the engine's
+// interfaces offer no other channel; the alternative is a Prometheus fork.
+func QueryTraitsFromContext(ctx context.Context) (QueryTraits, bool) {
+	traits, ok := ctx.Value(queryTraitsKey{}).(QueryTraits)
+	return traits, ok
+}

pkg/querier/api.go

@@ -50,6 +50,7 @@ func (handler *handler) QueryRange(rw http.ResponseWriter, req *http.Request) {
 		render.Error(rw, err)
 		return
 	}
+	queryRangeRequest.PromQLProvider = req.Header.Get("X-SigNoz-PromQL-Provider")
 
 	// Validate the query request
 	if err := queryRangeRequest.Validate(); err != nil {

pkg/querier/preview.go

@@ -230,7 +230,7 @@ func (q *querier) buildPreviewProviders(
 			sub.CompositeQuery = qbtypes.CompositeQuery{Queries: []qbtypes.QueryEnvelope{query}}
 		}
 
-		built, _, bErr := q.buildQueries(&sub, deps, missingMetricQuerySet, event)
+		built, _, bErr := q.buildQueries(&sub, deps, missingMetricQuerySet, event, promqlOptions{})
 		if bErr != nil {
 			errs[name] = bErr
 			continue

pkg/querier/promql_query.go

@@ -8,15 +8,19 @@ import (
 	"regexp"
 	"sort"
 	"strings"
+	"sync"
 	"text/template"
 	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2"
+
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/promql/parser"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
 	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
@@ -39,6 +43,13 @@ var quotedMetricOutsideBracesPattern = regexp.MustCompile(`"([^"]+)"\s*\{`)
 // tryEnhancePromQLExecError attempts to convert a PromQL execution error into
 // a properly typed error. Returns nil if the error is not a recognized execution error.
 func tryEnhancePromQLExecError(execErr error) error {
+	// A storage may fail a query with an already-typed error (e.g. the
+	// clickhousev2 series/sample budgets); surface it as-is instead of
+	// flattening it into an internal error.
+	if typed := typedStorageError(execErr); typed != nil {
+		return typed
+	}
+
 	var eqc promql.ErrQueryCanceled
 	var eqt promql.ErrQueryTimeout
 	var es promql.ErrStorage
@@ -58,6 +69,30 @@ func tryEnhancePromQLExecError(execErr error) error {
 	}
 }
 
+// typedStorageError walks an engine execution error chain looking for a
+// SigNoz-typed invalid-input error raised by the storage layer (the budget
+// refusals). Every wrapper level is stepped through by hand: Ast is a bare
+// type cast, not an unwrap — it misses a typed error behind the engine's
+// "expanding series: %w" — and promql.ErrStorage has no Unwrap method at
+// all, so a plain unwrap loop would stop at it.
+func typedStorageError(execErr error) error {
+	for e := execErr; e != nil; {
+		if errors.Ast(e, errors.TypeInvalidInput) {
+			return e
+		}
+		if es, ok := e.(promql.ErrStorage); ok {
+			e = es.Err
+			continue
+		}
+		u, ok := e.(interface{ Unwrap() error })
+		if !ok {
+			return nil
+		}
+		e = u.Unwrap()
+	}
+	return nil
+}
+
 // enhancePromQLError adds helpful context to PromQL parse errors,
 // particularly for UTF-8 syntax migration issues where metric and label
 // names containing dots need to be quoted.
@@ -98,6 +133,24 @@ type promqlQuery struct {
 	tr          qbv5.TimeRange
 	requestType qbv5.RequestType
 	vars        map[string]qbv5.VariableItem
+	opts        promqlOptions
+}
+
+// promqlOptions is how a PromQL query relates to the clickhousev2 provider
+// (see querier.promqlOptions for where the fields come from and why they are
+// flag-gated). Both providers are nil for a plain request, so a plain
+// request costs nothing extra.
+type promqlOptions struct {
+	// shadow, when set, runs the query on this provider after serving and
+	// logs any result difference; the response is never affected.
+	shadow *clickhouseprometheusv2.Provider
+	// shadowSlots is the querier-wide admission for shadow runs, shared by
+	// every query so the bound holds per process.
+	shadowSlots chan struct{}
+	// serve, when set, serves the response from this provider instead of the
+	// default path. Comparison callers fetch the default and the pinned
+	// result as two API calls and diff them.
+	serve *clickhouseprometheusv2.Provider
 }
 
 var _ qbv5.Query = (*promqlQuery)(nil)
@@ -110,6 +163,7 @@ func newPromqlQuery(
 	tr qbv5.TimeRange,
 	requestType qbv5.RequestType,
 	variables map[string]qbv5.VariableItem,
+	opts promqlOptions,
 ) *promqlQuery {
 	return &promqlQuery{
 		logger:      logger,
@@ -119,10 +173,20 @@ func newPromqlQuery(
 		tr:          tr,
 		requestType: requestType,
 		vars:        variables,
+		opts:        opts,
 	}
 }
 
 func (q *promqlQuery) Fingerprint() string {
+	// A pinned request must not share cache entries with default serving: a
+	// cached default result would satisfy the pin without running the pinned
+	// provider, and a pinned result would poison normal serving. No
+	// fingerprint means no caching at all — the pin exists to observe a
+	// provider, so a cache in front of it defeats the point.
+	if q.opts.serve != nil {
+		return ""
+	}
+
 	query, err := q.renderVars(q.query.Query, q.vars, q.tr.From, q.tr.To)
 	if err != nil {
 		q.logger.ErrorContext(context.TODO(), "failed render template variables", slog.String("query", q.query.Query))
@@ -248,7 +312,16 @@ func (q *promqlQuery) PreviewStatements(ctx context.Context) ([]prometheus.Captu
 	start := int64(querybuilder.ToNanoSecs(q.tr.From))
 	end := int64(querybuilder.ToNanoSecs(q.tr.To))
 
+	// Attach the same query traits as Execute so the captured statements
+	// match what the live path would run.
+	if expr, parseErr := q.parser.ParseExpr(rendered); parseErr == nil {
+		ctx = prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
+	}
+
 	capStorage, recorder := storer.CapturingStorage()
+	if capStorage == nil {
+		return nil, nil
+	}
 	qry, err := q.promEngine.Engine().NewRangeQuery(
 		ctx,
 		capStorage,
@@ -292,6 +365,58 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
 		return nil, err
 	}
 
+	// Attach query traits so the storage can prove step-aligned optimizations
+	// safe (see prometheus.QueryTraits). A parse failure surfaces below via
+	// the engine with the enhanced error message.
+	if expr, parseErr := q.parser.ParseExpr(query); parseErr == nil {
+		ctx = prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
+	}
+
+	// Accumulate ClickHouse-side scan stats across every storage query this
+	// evaluation issues (engine selectors or the compiled executor): progress
+	// options propagate to each ClickHouse query through the context.
+	var statsMu sync.Mutex
+	var rowsScanned, bytesScanned uint64
+	ctx = clickhouse.Context(ctx, clickhouse.WithProgress(func(p *clickhouse.Progress) {
+		statsMu.Lock()
+		rowsScanned += p.Rows
+		bytesScanned += p.Bytes
+		statsMu.Unlock()
+	}))
+
+	began := time.Now()
+
+	// A pinned provider serves directly from it: comparison callers fetch
+	// the default result and the pinned result as two API calls and diff
+	// them.
+	if q.opts.serve != nil {
+		matrix, err := q.serveFromProvider(ctx, query, start, end)
+		if err != nil {
+			if enhanced := tryEnhancePromQLExecError(err); enhanced != nil {
+				return nil, enhanced
+			}
+			return nil, err
+		}
+		return q.toResult(matrix, nil, began, &statsMu, &rowsScanned, &bytesScanned), nil
+	}
+
+	// When the serving provider itself is clickhousev2
+	// (prometheus::provider: clickhousev2), serve the way the provider is
+	// designed to serve: transpiled when the shape allows. Without this the
+	// override would silently run the engine path only.
+	if prov, ok := q.promEngine.(*clickhouseprometheusv2.Provider); ok {
+		matrix, served, err := prov.TryExecuteRange(ctx, query, time.Unix(0, start), time.Unix(0, end), q.query.Step.Duration)
+		if err != nil {
+			if enhanced := tryEnhancePromQLExecError(err); enhanced != nil {
+				return nil, enhanced
+			}
+			return nil, err
+		}
+		if served {
+			return q.toResult(matrix, nil, began, &statsMu, &rowsScanned, &bytesScanned), nil
+		}
+	}
+
 	qry, err := q.promEngine.Engine().NewRangeQuery(
 		ctx,
 		q.promEngine.Storage(),
@@ -327,6 +452,34 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
 		return nil, errors.WrapInternalf(promErr, errors.CodeInternal, "error getting matrix from promql query %q", query)
 	}
 
+	if q.opts.shadow != nil {
+		// Shadows detach from the request, so without admission a dashboard
+		// burst would stack unbounded ClickHouse work for up to the shadow
+		// timeout — the concurrency pattern behind the original outages.
+		// Non-blocking: at the cap the comparison is skipped, not queued;
+		// a sampled shadow stream is exactly as useful for rollout evidence.
+		select {
+		case q.opts.shadowSlots <- struct{}{}:
+			// The engine pools the result's sample slices on Close; the
+			// shadow comparison needs a stable copy of what was served.
+			served := copyMatrix(matrix)
+			servedIn := time.Since(began)
+			go func() {
+				defer func() { <-q.opts.shadowSlots }()
+				q.runShadowCompare(context.WithoutCancel(ctx), query, start, end, served, servedIn)
+			}()
+		default:
+			q.logger.DebugContext(ctx, "promql shadow skipped: at concurrency cap", slog.String("query", query))
+		}
+	}
+
+	warnings, _ := res.Warnings.AsStrings(query, 10, 0)
+	return q.toResult(matrix, warnings, began, &statsMu, &rowsScanned, &bytesScanned), nil
+}
+
+// toResult converts an evaluated matrix into the v5 result shape, attaching
+// the ClickHouse scan stats accumulated during evaluation.
+func (q *promqlQuery) toResult(matrix promql.Matrix, warnings []string, began time.Time, statsMu *sync.Mutex, rowsScanned, bytesScanned *uint64) *qbv5.Result {
 	excludeLabel := func(labelName string) bool {
 		if labelName == "__name__" {
 			return false
@@ -359,7 +512,13 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
 		series = append(series, &s)
 	}
 
-	warnings, _ := res.Warnings.AsStrings(query, 10, 0)
+	statsMu.Lock()
+	stats := qbv5.ExecStats{
+		RowsScanned:  *rowsScanned,
+		BytesScanned: *bytesScanned,
+		DurationMS:   uint64(time.Since(began).Milliseconds()),
+	}
+	statsMu.Unlock()
 
 	return &qbv5.Result{
 		Type: q.requestType,
@@ -372,6 +531,6 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
 			},
 		},
 		Warnings: warnings,
-		// TODO: map promql stats?
-	}, nil
+		Stats:    stats,
+	}
 }

pkg/querier/promql_query_test.go

@@ -7,7 +7,9 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
 	qbv5 "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -440,3 +442,35 @@ func TestQuotedMetricOutsideBracesPattern(t *testing.T) {
 		})
 	}
 }
+
+// wrappedErr stands in for the engine's fmt-based "expanding series: %w"
+// wrapper: an ordinary error with an Unwrap chain that is not a SigNoz base
+// error itself.
+type wrappedErr struct{ inner error }
+
+func (w wrappedErr) Error() string { return "expanding series: " + w.inner.Error() }
+func (w wrappedErr) Unwrap() error { return w.inner }
+
+// A typed budget refusal must survive the engine's wrapping and reach the
+// API as invalid input; flattened to internal it becomes a 500 the user
+// cannot act on — the exact failure this error type exists to prevent.
+func TestTypedStorageError_SeesThroughEngineWrappers(t *testing.T) {
+	budget := errors.NewInvalidInputf(errors.CodeInvalidInput, "promql selector matched more than 500000 series")
+
+	assert.NotNil(t, typedStorageError(wrappedErr{inner: budget}), "typed error behind an Unwrap wrapper")
+	assert.NotNil(t, typedStorageError(promql.ErrStorage{Err: wrappedErr{inner: budget}}), "typed error behind ErrStorage then a wrapper")
+	assert.NotNil(t, typedStorageError(wrappedErr{inner: promql.ErrStorage{Err: budget}}), "typed error behind a wrapper then ErrStorage")
+	assert.Nil(t, typedStorageError(wrappedErr{inner: errors.NewInternalf(errors.CodeInternal, "boom")}), "internal errors stay internal")
+}
+
+// A pinned request must not share cache entries with default serving: a
+// cached default result would satisfy the pin without running the pinned
+// provider.
+func TestFingerprint_PinnedProviderBypassesCache(t *testing.T) {
+	q := &promqlQuery{
+		logger: slog.Default(),
+		query:  qbv5.PromQuery{Query: "up"},
+		opts:   promqlOptions{serve: &clickhouseprometheusv2.Provider{}},
+	}
+	assert.Empty(t, q.Fingerprint())
+}

pkg/querier/promql_shadow.go

@@ -0,0 +1,195 @@
+package querier
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"math"
+	"sort"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
+)
+
+// shadowTimeout bounds a shadow evaluation; a shadow run must never outlive
+// the request by much or pile up.
+const shadowTimeout = 2 * time.Minute
+
+// runShadowCompare executes the query on the clickhousev2 provider exactly
+// as it would serve (transpiled when the shape allows, engine over the v2
+// querier otherwise), compares against the served result and logs the
+// outcome. Serving is never affected: this runs after the response, off the
+// request context, and only logs. The mismatch and failure logs are the
+// rollout evidence — serving cuts over to v2 only after they stay clean.
+func (q *promqlQuery) runShadowCompare(ctx context.Context, query string, startNs, endNs int64, served promql.Matrix, servedIn time.Duration) {
+	defer func() {
+		if r := recover(); r != nil {
+			q.logger.ErrorContext(ctx, "promql shadow comparison panicked", slog.Any("panic", r), slog.String("query", query))
+		}
+	}()
+
+	ctx, cancel := context.WithTimeout(ctx, shadowTimeout)
+	defer cancel()
+
+	// The request context carries the served response's scan-stats progress
+	// callback; without replacing it the shadow's ClickHouse progress would
+	// race into the served stats. The response itself was already sent.
+	ctx = clickhouse.Context(ctx, clickhouse.WithProgress(func(*clickhouse.Progress) {}))
+
+	if expr, parseErr := q.parser.ParseExpr(query); parseErr == nil {
+		ctx = prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
+	}
+
+	start, end := time.Unix(0, startNs), time.Unix(0, endNs)
+	began := time.Now()
+	shadow, transpiled, err := executeOnProvider(ctx, q.opts.shadow, query, start, end, q.query.Step.Duration)
+	shadowIn := time.Since(began)
+
+	logAttrs := []any{
+		slog.String("query", query),
+		slog.Int64("start_ms", startNs/int64(time.Millisecond)),
+		slog.Int64("end_ms", endNs/int64(time.Millisecond)),
+		slog.Duration("step", q.query.Step.Duration),
+		slog.Bool("transpiled", transpiled),
+		slog.Duration("served_in", servedIn),
+		slog.Duration("shadow_in", shadowIn),
+	}
+
+	if err != nil {
+		// A shadow failure would be a serving failure after rollout; surface
+		// it at the same level as a result mismatch.
+		q.logger.WarnContext(ctx, "promql shadow execution failed", append(logAttrs, slog.Any("error", err))...)
+		return
+	}
+
+	servedNorm := normalizeShadowMatrix(served)
+	shadowNorm := normalizeShadowMatrix(shadow)
+	if diff := diffShadowMatrices(servedNorm, shadowNorm); diff != "" {
+		q.logger.WarnContext(ctx, "promql shadow comparison mismatch", append(logAttrs,
+			slog.String("diff", diff),
+			slog.Int("served_series", len(servedNorm)),
+			slog.Int("shadow_series", len(shadowNorm)),
+		)...)
+		return
+	}
+	// Matches log the timings: served_in vs shadow_in across the fleet is
+	// the perf evidence for the cutover, gathered for free.
+	q.logger.DebugContext(ctx, "promql shadow comparison matched", logAttrs...)
+}
+
+// serveFromProvider evaluates the query the way the pinned provider would
+// serve it.
+func (q *promqlQuery) serveFromProvider(ctx context.Context, query string, startNs, endNs int64) (promql.Matrix, error) {
+	matrix, _, err := executeOnProvider(ctx, q.opts.serve, query, time.Unix(0, startNs), time.Unix(0, endNs), q.query.Step.Duration)
+	return matrix, err
+}
+
+// executeOnProvider evaluates the query the way the provider would serve it:
+// transpiled in ClickHouse when the shape allows, the engine over the
+// provider's storage otherwise. The returned matrix is an owned copy.
+func executeOnProvider(ctx context.Context, prov *clickhouseprometheusv2.Provider, query string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error) {
+	matrix, ok, err := prov.TryExecuteRange(ctx, query, start, end, step)
+	if err != nil {
+		return nil, true, err
+	}
+	if ok {
+		return matrix, true, nil
+	}
+
+	qry, err := prov.Engine().NewRangeQuery(ctx, prov.Storage(), nil, query, start, end, step)
+	if err != nil {
+		return nil, false, err
+	}
+	defer qry.Close()
+
+	res := qry.Exec(ctx)
+	if res.Err != nil {
+		return nil, false, res.Err
+	}
+	matrix, err = res.Matrix()
+	if err != nil {
+		return nil, false, err
+	}
+	// Close returns the result's sample slices to the engine pool.
+	return copyMatrix(matrix), false, nil
+}
+
+func copyMatrix(matrix promql.Matrix) promql.Matrix {
+	out := make(promql.Matrix, 0, len(matrix))
+	for _, s := range matrix {
+		floats := make([]promql.FPoint, len(s.Floats))
+		copy(floats, s.Floats)
+		out = append(out, promql.Series{Metric: s.Metric.Copy(), Floats: floats})
+	}
+	return out
+}
+
+// normalizeShadowMatrix strips the labels the two providers legitimately
+// disagree on — v1 injects a synthetic fingerprint label and leaks
+// empty-valued labels from the stored attribute JSON, both removed from API
+// responses anyway — and sorts by label set.
+func normalizeShadowMatrix(matrix promql.Matrix) promql.Matrix {
+	out := make(promql.Matrix, 0, len(matrix))
+	for _, s := range matrix {
+		builder := labels.NewBuilder(s.Metric)
+		builder.Del(prometheus.FingerprintAsPromLabelName)
+		s.Metric.Range(func(l labels.Label) {
+			if l.Value == "" {
+				builder.Del(l.Name)
+			}
+		})
+		out = append(out, promql.Series{Metric: builder.Labels(), Floats: s.Floats})
+	}
+	sort.Slice(out, func(i, j int) bool { return labels.Compare(out[i].Metric, out[j].Metric) < 0 })
+	return out
+}
+
+// diffShadowMatrices returns a description of the first difference, or "".
+// Values compare with relative tolerance: spatial aggregations accumulate
+// floats in storage order, which differs between the providers in the last
+// ULP.
+func diffShadowMatrices(served, shadow promql.Matrix) string {
+	const relTol = 1e-9
+	if len(served) != len(shadow) {
+		return fmt.Sprintf("series count: served=%d shadow=%d", len(served), len(shadow))
+	}
+	for i := range served {
+		if labels.Compare(served[i].Metric, shadow[i].Metric) != 0 {
+			return fmt.Sprintf("series %d labels: served=%s shadow=%s", i, served[i].Metric, shadow[i].Metric)
+		}
+		if len(served[i].Floats) != len(shadow[i].Floats) {
+			return fmt.Sprintf("series %s points: served=%d shadow=%d", served[i].Metric, len(served[i].Floats), len(shadow[i].Floats))
+		}
+		for j := range served[i].Floats {
+			a, b := served[i].Floats[j], shadow[i].Floats[j]
+			if a.T != b.T {
+				return fmt.Sprintf("series %s point %d ts: served=%d shadow=%d", served[i].Metric, j, a.T, b.T)
+			}
+			// NaN and infinities first: NaN != NaN and Inf-Inf arithmetic
+			// would otherwise make one-sided NaN and Inf-vs-finite compare
+			// as equal (NaN > x and Inf > Inf are both false).
+			if math.IsNaN(a.F) || math.IsNaN(b.F) {
+				if math.IsNaN(a.F) != math.IsNaN(b.F) {
+					return fmt.Sprintf("series %s @%d value: served=%v shadow=%v", served[i].Metric, a.T, a.F, b.F)
+				}
+				continue
+			}
+			if math.IsInf(a.F, 0) || math.IsInf(b.F, 0) {
+				if a.F != b.F {
+					return fmt.Sprintf("series %s @%d value: served=%v shadow=%v", served[i].Metric, a.T, a.F, b.F)
+				}
+				continue
+			}
+			diff := math.Abs(a.F - b.F)
+			scale := math.Max(math.Abs(a.F), math.Abs(b.F))
+			if diff > relTol*math.Max(scale, 1e-300) && diff > 1e-12 {
+				return fmt.Sprintf("series %s @%d value: served=%v shadow=%v", served[i].Metric, a.T, a.F, b.F)
+			}
+		}
+	}
+	return ""
+}

pkg/querier/promql_shadow_test.go

@@ -0,0 +1,66 @@
+package querier
+
+import (
+	"math"
+	"testing"
+
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNormalizeShadowMatrix(t *testing.T) {
+	matrix := promql.Matrix{
+		{
+			Metric: labels.FromStrings("__name__", "up", "fingerprint", "42", "empty", "", "job", "api"),
+			Floats: []promql.FPoint{{T: 1000, F: 1}},
+		},
+		{
+			Metric: labels.FromStrings("a", "1"),
+			Floats: []promql.FPoint{{T: 1000, F: 2}},
+		},
+	}
+	norm := normalizeShadowMatrix(matrix)
+	// sorted by labels; fingerprint and empty-valued labels stripped
+	assert.Equal(t, labels.FromStrings("__name__", "up", "job", "api"), norm[0].Metric)
+	assert.Equal(t, labels.FromStrings("a", "1"), norm[1].Metric)
+}
+
+func TestDiffShadowMatrices(t *testing.T) {
+	series := func(v float64) promql.Matrix {
+		return promql.Matrix{{Metric: labels.FromStrings("a", "1"), Floats: []promql.FPoint{{T: 1000, F: v}}}}
+	}
+
+	assert.Empty(t, diffShadowMatrices(series(1.5), series(1.5)))
+	// last-ULP differences from storage-order float accumulation are expected
+	assert.Empty(t, diffShadowMatrices(series(0.08888888888888889), series(0.08888888888888888)))
+	assert.Empty(t, diffShadowMatrices(series(math.NaN()), series(math.NaN())))
+
+	assert.Contains(t, diffShadowMatrices(series(1.5), series(1.6)), "value")
+	assert.Contains(t, diffShadowMatrices(series(1.5), promql.Matrix{}), "series count")
+	assert.Contains(t, diffShadowMatrices(
+		series(1.5),
+		promql.Matrix{{Metric: labels.FromStrings("a", "2"), Floats: []promql.FPoint{{T: 1000, F: 1.5}}}},
+	), "labels")
+	assert.Contains(t, diffShadowMatrices(
+		series(1.5),
+		promql.Matrix{{Metric: labels.FromStrings("a", "1"), Floats: []promql.FPoint{{T: 2000, F: 1.5}}}},
+	), "ts")
+}
+
+// One-sided NaN makes every float comparison false, and Inf-Inf arithmetic
+// yields Inf > Inf == false; without explicit handling both divergences log
+// as matched — a shadow comparator that cannot see them would green-light a
+// broken rollout.
+func TestDiffShadowMatrices_SpecialFloats(t *testing.T) {
+	point := func(v float64) promql.Matrix {
+		return promql.Matrix{{Metric: labels.FromStrings("a", "1"), Floats: []promql.FPoint{{T: 1000, F: v}}}}
+	}
+
+	assert.NotEmpty(t, diffShadowMatrices(point(math.NaN()), point(1.5)), "one-sided NaN must diff")
+	assert.NotEmpty(t, diffShadowMatrices(point(1.5), point(math.NaN())), "one-sided NaN must diff either way")
+	assert.NotEmpty(t, diffShadowMatrices(point(math.Inf(1)), point(1.5)), "Inf vs finite must diff")
+	assert.NotEmpty(t, diffShadowMatrices(point(math.Inf(1)), point(math.Inf(-1))), "opposite infinities must diff")
+	assert.Empty(t, diffShadowMatrices(point(math.Inf(1)), point(math.Inf(1))), "equal infinities match")
+	assert.Empty(t, diffShadowMatrices(point(math.NaN()), point(math.NaN())), "both NaN match")
+}

pkg/querier/querier.go

@@ -19,10 +19,12 @@ import (
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/flagger"
 	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
 	"github.com/SigNoz/signoz/pkg/query-service/utils"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
 	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
+	"github.com/SigNoz/signoz/pkg/types/featuretypes"
 	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -36,11 +38,19 @@ var (
 )
 
 type querier struct {
-	logger                   *slog.Logger
-	fl                       flagger.Flagger
-	telemetryStore           telemetrystore.TelemetryStore
-	metadataStore            telemetrytypes.MetadataStore
-	promEngine               prometheus.Prometheus
+	logger         *slog.Logger
+	fl             flagger.Flagger
+	telemetryStore telemetrystore.TelemetryStore
+	metadataStore  telemetrytypes.MetadataStore
+	promEngine     prometheus.Prometheus
+	// promV2 is the clickhousev2 prometheus provider, wired only when the
+	// serving provider is the default one (nil otherwise). It reads the same
+	// ClickHouse data through a different implementation; PromQL queries
+	// shadow-compare against it behind the use_prometheus_clickhouse_v2 flag
+	// and can be pinned to it for a response (see promqlOptions). It never
+	// serves by default — that cutover happens only after the shadow logs
+	// stay clean.
+	promV2                   *clickhouseprometheusv2.Provider
 	traceStmtBuilder         qbtypes.StatementBuilder[qbtypes.TraceAggregation]
 	logStmtBuilder           qbtypes.StatementBuilder[qbtypes.LogAggregation]
 	auditStmtBuilder         qbtypes.StatementBuilder[qbtypes.LogAggregation]
@@ -51,8 +61,16 @@ type querier struct {
 	liveDataRefresh          time.Duration
 	builderConfig            builderConfig
 	maxConcurrentQueries     int
+	// shadowSlots bounds concurrent shadow comparisons per process; shadows
+	// detach from their requests, so nothing else limits how many pile up.
+	shadowSlots chan struct{}
 }
 
+// maxConcurrentShadows is deliberately small: a shadow is a full extra
+// ClickHouse evaluation, and a sampled stream of comparisons is exactly as
+// useful for rollout evidence as an exhaustive one under load.
+const maxConcurrentShadows = 8
+
 var _ Querier = (*querier)(nil)
 
 func New(
@@ -60,6 +78,7 @@ func New(
 	telemetryStore telemetrystore.TelemetryStore,
 	metadataStore telemetrytypes.MetadataStore,
 	promEngine prometheus.Prometheus,
+	promV2 *clickhouseprometheusv2.Provider,
 	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
@@ -81,6 +100,7 @@ func New(
 		telemetryStore:           telemetryStore,
 		metadataStore:            metadataStore,
 		promEngine:               promEngine,
+		promV2:                   promV2,
 		traceStmtBuilder:         traceStmtBuilder,
 		logStmtBuilder:           logStmtBuilder,
 		auditStmtBuilder:         auditStmtBuilder,
@@ -93,6 +113,7 @@ func New(
 			logTraceIDWindowPaddingMS: uint64(logTraceIDWindowPadding.Milliseconds()),
 		},
 		maxConcurrentQueries: maxConcurrentQueries,
+		shadowSlots:          make(chan struct{}, maxConcurrentShadows),
 	}
 }
 
@@ -132,7 +153,11 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 		missingMetricQuerySet[name] = true
 	}
 
-	queries, steps, err := q.buildQueries(req, dependencyQueries, missingMetricQuerySet, event)
+	promqlOpts, err := q.promqlOptions(ctx, orgID, req)
+	if err != nil {
+		return nil, err
+	}
+	queries, steps, err := q.buildQueries(req, dependencyQueries, missingMetricQuerySet, event, promqlOpts)
 	if err != nil {
 		return nil, err
 	}
@@ -175,11 +200,40 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 	return qbResp, qbErr
 }
 
+// promqlOptions derives the PromQL execution options for a request. With the
+// org's use_prometheus_clickhouse_v2 flag on, queries are shadow-compared
+// against the clickhousev2 provider (serving unaffected, diffs logged; see
+// promql_shadow.go). The X-SigNoz-PromQL-Provider header may instead pin the
+// response to that provider — integration tests and support fetch both
+// results for comparison — so it is deliberately flag-gated too: without the
+// gate the header would be an unaudited switch onto a provider still under
+// validation.
+func (q *querier) promqlOptions(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (promqlOptions, error) {
+	enabled := q.fl.BooleanOrEmpty(ctx, flagger.FeatureUsePrometheusClickhouseV2, featuretypes.NewFlaggerEvaluationContext(orgID))
+	if req.PromQLProvider == "" {
+		if enabled && q.promV2 != nil {
+			return promqlOptions{shadow: q.promV2, shadowSlots: q.shadowSlots}, nil
+		}
+		return promqlOptions{}, nil
+	}
+	if req.PromQLProvider != prometheus.ProviderClickhouseV2 {
+		return promqlOptions{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown promql provider %q", req.PromQLProvider)
+	}
+	if !enabled {
+		return promqlOptions{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "promql provider %q requires the use_prometheus_clickhouse_v2 flag", req.PromQLProvider)
+	}
+	if q.promV2 == nil {
+		return promqlOptions{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "promql provider %q is not available", req.PromQLProvider)
+	}
+	return promqlOptions{serve: q.promV2}, nil
+}
+
 func (q *querier) buildQueries(
 	req *qbtypes.QueryRangeRequest,
 	dependencyQueries map[string]bool,
 	missingMetricQuerySet map[string]bool,
 	event *qbtypes.QBEvent,
+	promqlOpts promqlOptions,
 ) (map[string]qbtypes.Query, map[string]qbtypes.Step, error) {
 
 	tmplVars := req.Variables
@@ -204,7 +258,7 @@ func (q *querier) buildQueries(
 			if !ok {
 				return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid promql query spec %T", query.Spec)
 			}
-			promqlQuery := newPromqlQuery(q.logger, q.promEngine, promQuery, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType, tmplVars)
+			promqlQuery := newPromqlQuery(q.logger, q.promEngine, promQuery, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType, tmplVars, promqlOpts)
 			queries[promQuery.Name] = promqlQuery
 			steps[promQuery.Name] = promQuery.Step
 		case qbtypes.QueryTypeClickHouseSQL:
@@ -846,7 +900,7 @@ func (q *querier) createRangedQuery(originalQuery qbtypes.Query, timeRange qbtyp
 	switch qt := originalQuery.(type) {
 	case *promqlQuery:
 		queryCopy := qt.query.Copy()
-		return newPromqlQuery(q.logger, q.promEngine, queryCopy, timeRange, qt.requestType, qt.vars)
+		return newPromqlQuery(q.logger, qt.promEngine, queryCopy, timeRange, qt.requestType, qt.vars, qt.opts)
 
 	case *chSQLQuery:
 		queryCopy := qt.query.Copy()

pkg/querier/querier_test.go

@@ -48,6 +48,7 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
 		nil, // telemetryStore
 		metadataStore,
 		nil,                // prometheus
+		nil,                // promV2
 		nil,                // traceStmtBuilder
 		nil,                // logStmtBuilder
 		nil,                // auditStmtBuilder
@@ -120,6 +121,7 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 		telemetryStore,
 		metadataStore,
 		nil,                      // prometheus
+		nil,                      // promV2
 		nil,                      // traceStmtBuilder
 		nil,                      // logStmtBuilder
 		nil,                      // auditStmtBuilder

pkg/querier/signozquerier/provider.go

@@ -7,6 +7,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/flagger"
 	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/telemetryaudit"
@@ -22,6 +23,7 @@ import (
 func NewFactory(
 	telemetryStore telemetrystore.TelemetryStore,
 	prometheus prometheus.Prometheus,
+	promV2 *clickhouseprometheusv2.Provider,
 	cache cache.Cache,
 	flagger flagger.Flagger,
 ) factory.ProviderFactory[querier.Querier, querier.Config] {
@@ -32,7 +34,7 @@ func NewFactory(
 			settings factory.ProviderSettings,
 			cfg querier.Config,
 		) (querier.Querier, error) {
-			return newProvider(ctx, settings, cfg, telemetryStore, prometheus, cache, flagger)
+			return newProvider(ctx, settings, cfg, telemetryStore, prometheus, promV2, cache, flagger)
 		},
 	)
 }
@@ -43,6 +45,7 @@ func newProvider(
 	cfg querier.Config,
 	telemetryStore telemetrystore.TelemetryStore,
 	prometheus prometheus.Prometheus,
+	promV2 *clickhouseprometheusv2.Provider,
 	cache cache.Cache,
 	flagger flagger.Flagger,
 ) (querier.Querier, error) {
@@ -184,6 +187,7 @@ func newProvider(
 		telemetryStore,
 		telemetryMetadataStore,
 		prometheus,
+		promV2,
 		traceStmtBuilder,
 		logStmtBuilder,
 		auditStmtBuilder,

pkg/query-service/rules/managertestfactory.go

@@ -105,7 +105,7 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
 	}
 
 	// Create querier with test values
-	providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, cache, flagger)
+	providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, nil, cache, flagger)
 	mockQuerier, err := providerFactory.New(context.Background(), providerSettings, querier.Config{})
 	require.NoError(t, err)
 

pkg/query-service/rules/setups_test.go

@@ -47,6 +47,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		telemetryStore,
 		metadataStore,
 		nil, // prometheus
+		nil, // promV2
 		nil, // traceStmtBuilder
 		nil, // logStmtBuilder
 		nil, // auditStmtBuilder
@@ -102,6 +103,7 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
 		telemetryStore,
 		metadataStore,
 		nil,            // prometheus
+		nil,            // promV2
 		nil,            // traceStmtBuilder
 		logStmtBuilder, // logStmtBuilder
 		nil,            // auditStmtBuilder
@@ -151,6 +153,7 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
 		telemetryStore,
 		metadataStore,
 		nil,              // prometheus
+		nil,              // promV2
 		traceStmtBuilder, // traceStmtBuilder
 		nil,              // logStmtBuilder
 		nil,              // auditStmtBuilder

pkg/signoz/provider.go

@@ -44,6 +44,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/pprof/nooppprof"
 	"github.com/SigNoz/signoz/pkg/prometheus"
 	"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheus"
+	"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/querier/signozquerier"
 	"github.com/SigNoz/signoz/pkg/sharder"
@@ -235,6 +236,7 @@ func NewTelemetryStoreProviderFactories() factory.NamedMap[factory.ProviderFacto
 func NewPrometheusProviderFactories(telemetryStore telemetrystore.TelemetryStore) factory.NamedMap[factory.ProviderFactory[prometheus.Prometheus, prometheus.Config]] {
 	return factory.MustNewNamedMap(
 		clickhouseprometheus.NewFactory(telemetryStore),
+		clickhouseprometheusv2.NewFactory(telemetryStore),
 	)
 }
 
@@ -276,9 +278,9 @@ func NewStatsReporterProviderFactories(aggregator statsreporter.Aggregator, orgG
 	)
 }
 
-func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, cache cache.Cache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
+func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, promV2 *clickhouseprometheusv2.Provider, cache cache.Cache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
 	return factory.MustNewNamedMap(
-		signozquerier.NewFactory(telemetryStore, prometheus, cache, flagger),
+		signozquerier.NewFactory(telemetryStore, prometheus, promV2, cache, flagger),
 	)
 }
 

pkg/signoz/signoz.go

@@ -39,6 +39,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/tag/impltag"
 	"github.com/SigNoz/signoz/pkg/modules/user/impluser"
 	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/queryparser"
 	"github.com/SigNoz/signoz/pkg/ruler"
@@ -241,6 +242,11 @@ func New(
 
 	retentionGetter := implretention.NewGetter(implretention.NewStore(sqlstore))
 
+	// promV2 is the clickhousev2 provider handed to the querier for shadow
+	// comparison and pinned serving (declared before the serving provider,
+	// whose variable shadows the package name below).
+	var promV2 *clickhouseprometheusv2.Provider
+
 	// Initialize prometheus from the available prometheus provider factories
 	prometheus, err := factory.NewProviderFromNamedMap(
 		ctx,
@@ -253,12 +259,29 @@ func New(
 		return nil, err
 	}
 
+	// With the default provider, also stand up the clickhousev2 provider for
+	// the querier: PromQL queries shadow-compare against it behind the
+	// use_prometheus_clickhouse_v2 flag (see pkg/querier/promql_shadow.go).
+	// It never serves by default. An explicit
+	// prometheus::provider: clickhousev2 makes v2 the serving provider
+	// outright, so there is nothing to compare against.
+	if config.Prometheus.Provider() == "clickhouse" {
+		v2Config := config.Prometheus
+		// The v2 engine only evaluates shadow and pinned queries; disable its
+		// active query tracker so two trackers never share a file.
+		v2Config.ActiveQueryTrackerConfig.Enabled = false
+		promV2, err = clickhouseprometheusv2.New(ctx, providerSettings, v2Config, telemetrystore)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	// Initialize querier from the available querier provider factories
 	querier, err := factory.NewProviderFromNamedMap(
 		ctx,
 		providerSettings,
 		config.Querier,
-		NewQuerierProviderFactories(telemetrystore, prometheus, cache, flagger),
+		NewQuerierProviderFactories(telemetrystore, prometheus, promV2, cache, flagger),
 		config.Querier.Provider(),
 	)
 	if err != nil {

pkg/types/querybuildertypes/querybuildertypesv5/req.go

@@ -370,6 +370,14 @@ type QueryRangeRequest struct {
 	// NoCache is a flag to disable caching for the request.
 	NoCache bool `json:"noCache,omitempty"`
 
+	// PromQLProvider serves this request's PromQL queries via the named
+	// prometheus provider ("clickhousev2") instead of the default — the same
+	// data read through a different implementation. It is set from the
+	// X-SigNoz-PromQL-Provider header by the API handler, never from the
+	// body: a rollout-scoped comparison hook for integration tests and
+	// support should not become part of the public request schema.
+	PromQLProvider string `json:"-"`
+
 	FormatOptions *FormatOptions `json:"formatOptions,omitempty"`
 }
 

tests/integration/tests/promqlparity/01_provider_parity.py

@@ -0,0 +1,257 @@
+"""PromQL provider parity: the CI guard for the clickhousev2 rollout.
+
+Every battery query is fetched through /api/v5/query_range twice — once
+served from the default provider, once pinned to the clickhousev2 provider
+via the flag-gated X-SigNoz-PromQL-Provider header — over several evaluation
+grids, and the two responses are compared series-by-series, point-by-point,
+in the test. The providers read the same data through different
+implementations, so timestamps and label sets must match exactly; values
+within 1e-9 relative (they accumulate floats in different orders in the last
+bit).
+
+The fixtures are deterministic and target the semantics that historically
+diverge between implementations:
+
+- parity.counter: job=a resets mid-window; job=b is clean; job=c resets
+  twice AND has a gap longer than the 5m lookback, so its series vanishes
+  from instant selections and re-enters extrapolation windows.
+- parity.gauge: pod=p1 lives forever; pod=p2 emits one stale marker and
+  resumes (a scrape blip); pod=p3 dies with a stale marker and stays dead;
+  pod=p4 is born mid-window; pod=p5 dies and resurrects 10 minutes later.
+  Per-pod magnitudes are distinct so topk never sees ties (tied topk picks
+  winners by evaluation order — legitimately different between two correct
+  implementations).
+- parity.hist.bucket: a classic cumulative histogram.
+
+All values are powers of two so float aggregation is order-independent.
+
+The evaluation grids vary the query window and step: aligned and unaligned
+starts (grid anchoring), a step that is no multiple of the scrape cadence,
+a coarse step, and a window that begins before any data exists.
+"""
+
+import math
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import requests
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metrics import Metrics
+
+REL_TOL = 1e-9
+
+QUERIES = [
+    # instant selectors: lookback, stale-marker shadowing, name keeping
+    '{"parity.gauge"}',
+    '{"parity.gauge"} > 4',
+    'sum by (pod) ({"parity.gauge"})',
+    'sum({"parity.gauge"} offset 10m)',
+    # instant selection over a counter with a lookback-sized gap
+    '{"parity.counter", job="c"}',
+    # range functions: counter resets, double resets, gaps, extrapolation
+    'sum by (job) (rate({"parity.counter"}[5m]))',
+    'rate({"parity.counter", job="c"}[5m])',
+    'increase({"parity.counter", job="c"}[15m])',
+    'sum(increase({"parity.counter"}[10m] offset 15m))',
+    'sum by (pod) (delta({"parity.gauge"}[15m]))',
+    'irate({"parity.counter"}[5m])',
+    'idelta({"parity.gauge"}[5m])',
+    # *_over_time
+    'max by (pod) (avg_over_time({"parity.gauge"}[10m]))',
+    'count_over_time({"parity.gauge"}[10m])',
+    'last_over_time({"parity.gauge"}[10m])',
+    'min_over_time({"parity.gauge"}[7m])',
+    'sum_over_time({"parity.counter", job="c"}[10m])',
+    # hybrid shapes: quantiles, ratios, topk, or-fill
+    'histogram_quantile(0.9, sum by (le) (rate({"parity.hist.bucket"}[5m])))',
+    'sum(rate({"parity.counter"}[5m])) / sum(rate({"parity.counter"}[10m]))',
+    'topk(2, sum by (pod) ({"parity.gauge"}))',
+    'sum(rate({"parity.counter", job="missing"}[5m])) or vector(0)',
+    # subquery smoothing evaluates inner units on the subquery grid
+    'min_over_time((sum by (job) (increase({"parity.counter"}[5m])))[10m:5m])',
+    'avg_over_time((sum by (pod) ({"parity.gauge"}))[10m:2m])',
+]
+
+
+def grids(now_ms: int) -> list[tuple[str, int, int, int]]:
+    """(description, start_ms, end_ms, step_seconds) variations. Different
+    starts anchor the evaluation grid differently; PromQL evaluates at
+    start + k*step, so an unaligned start shifts which samples each lookback
+    window sees. The long windows select the coarser series tables
+    (time_series_v4_6hrs at > 6h, _1week at > 1w) so the table choice and
+    its bucket rounding are exercised end to end, not just in unit tests."""
+    minute = 60_000
+    return [
+        ("90m window, 60s step, minute-aligned", now_ms - 90 * minute, now_ms, 60),
+        ("90m window, 60s step, start unaligned by 17s", now_ms - 90 * minute + 17_000, now_ms, 60),
+        ("90m window, 90s step (no cadence multiple)", now_ms - 90 * minute, now_ms, 90),
+        ("3h window starting before any data, 300s step", now_ms - 180 * minute, now_ms, 300),
+        ("7h window (6h series table), 300s step", now_ms - 420 * minute, now_ms, 300),
+        ("8d window (1w series table), 3600s step", now_ms - 8 * 24 * 60 * minute, now_ms, 3600),
+    ]
+
+
+def seed(insert_metrics, now: datetime) -> None:
+    """95 minutes of 30s-cadence series ending at now (see module docstring
+    for the shapes). Values are powers of two so float aggregation is
+    order-independent."""
+    metrics: list[Metrics] = []
+    start = now - timedelta(minutes=95)
+
+    counters = {"a": 0.0, "b": 0.0, "c": 0.0}
+    step = 0
+    ts = start
+    while ts <= now:
+        for i, job in enumerate(("a", "b", "c")):
+            counters[job] += 2 << i
+            if job == "a" and ts == start + timedelta(minutes=45):
+                counters[job] = 8  # counter reset
+            if job == "c":
+                # Two resets and a >5m gap: the gap exceeds the lookback, so
+                # the series vanishes from instant selections mid-window.
+                if ts == start + timedelta(minutes=30) or ts == start + timedelta(minutes=70):
+                    counters[job] = 4
+                if start + timedelta(minutes=50) < ts <= start + timedelta(minutes=56):
+                    continue
+            metrics.append(
+                Metrics(
+                    metric_name="parity.counter",
+                    labels={"job": job},
+                    timestamp=ts,
+                    value=counters[job],
+                    temporality="Cumulative",
+                    type_="Sum",
+                )
+            )
+
+        # Distinct per-pod magnitudes keep topk free of ties: tied series
+        # would make the winner storage-order-dependent and the comparison
+        # nondeterministic.
+        for pod, scale in (("p1", 1), ("p2", 8), ("p3", 64), ("p4", 128), ("p5", 256)):
+            flags = 0
+            value = float((1 << (step % 6)) * scale)
+            if pod == "p2" and ts == start + timedelta(minutes=40):
+                flags, value = 1, 0.0  # stale marker, series resumes after
+            if pod == "p3":
+                if ts > start + timedelta(minutes=50, seconds=30):
+                    continue
+                if ts == start + timedelta(minutes=50, seconds=30):
+                    flags, value = 1, 0.0  # series dies and stays dead
+            if pod == "p4" and ts < start + timedelta(minutes=60):
+                continue  # born mid-window
+            if pod == "p5":
+                # Dies with a stale marker, resurrects 10 minutes later.
+                if ts == start + timedelta(minutes=20):
+                    flags, value = 1, 0.0
+                elif start + timedelta(minutes=20) < ts < start + timedelta(minutes=30):
+                    continue
+            metrics.append(
+                Metrics(
+                    metric_name="parity.gauge",
+                    labels={"pod": pod},
+                    timestamp=ts,
+                    value=value,
+                    temporality="Unspecified",
+                    type_="Gauge",
+                    is_monotonic=False,
+                    flags=flags,
+                )
+            )
+
+        for i, le in enumerate(("0.5", "2", "+Inf")):
+            metrics.append(
+                Metrics(
+                    metric_name="parity.hist.bucket",
+                    labels={"le": le},
+                    timestamp=ts,
+                    value=float((step + 1) * (2 << i)),
+                    temporality="Cumulative",
+                    type_="Histogram",
+                )
+            )
+
+        step += 1
+        ts += timedelta(seconds=30)
+
+    insert_metrics(metrics)
+
+
+def normalize(response_json: dict) -> dict[tuple, list[tuple]]:
+    """Response results as {sorted-label-pairs: [(ts, value), ...]}."""
+    out: dict[tuple, list[tuple]] = {}
+    for result in response_json["data"]["data"]["results"]:
+        for aggregation in result.get("aggregations") or []:
+            for series in aggregation.get("series") or []:
+                key = tuple(
+                    sorted((label["key"]["name"], label["value"]) for label in (series.get("labels") or []))
+                )
+                out[key] = [(v["timestamp"], v["value"]) for v in (series.get("values") or [])]
+    return out
+
+
+def values_equal(a: float, b: float) -> bool:
+    if isinstance(a, str) or isinstance(b, str):  # "NaN" and friends
+        return str(a) == str(b)
+    if math.isnan(a) and math.isnan(b):
+        return True
+    return math.isclose(a, b, rel_tol=REL_TOL, abs_tol=1e-12)
+
+
+def fetch(signoz: types.SigNoz, token: str, query: str, start_ms: int, end_ms: int, step: int, provider: str | None):
+    payload = {
+        "schemaVersion": "v1",
+        "start": start_ms,
+        "end": end_ms,
+        "requestType": "time_series",
+        "compositeQuery": {"queries": [{"type": "promql", "spec": {"name": "A", "query": query, "step": step}}]},
+        "formatOptions": {"formatTableResultForUI": False, "fillGaps": False},
+        "noCache": True,
+    }
+    headers = {"authorization": f"Bearer {token}"}
+    if provider:
+        headers["X-SigNoz-PromQL-Provider"] = provider
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v5/query_range"),
+        timeout=60,
+        headers=headers,
+        json=payload,
+    )
+    assert response.status_code == HTTPStatus.OK, f"{query} (provider={provider}): {response.text}"
+    return normalize(response.json())
+
+
+def test_provider_parity(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token,
+    insert_metrics,
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    seed(insert_metrics, now)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    now_ms = int(now.timestamp() * 1000)
+
+    served_any_data = False
+    for grid_desc, start_ms, end_ms, step in grids(now_ms):
+        for query in QUERIES:
+            where = f"{query} [{grid_desc}]"
+            served = fetch(signoz, token, query, start_ms, end_ms, step, provider=None)
+            pinned = fetch(signoz, token, query, start_ms, end_ms, step, provider="clickhousev2")
+
+            assert served.keys() == pinned.keys(), (
+                f"{where}: series sets differ\nonly default: {sorted(set(served) - set(pinned))}"
+                f"\nonly clickhousev2: {sorted(set(pinned) - set(served))}"
+            )
+            for key, served_points in served.items():
+                pinned_points = pinned[key]
+                assert len(served_points) == len(pinned_points), f"{where} {key}: point counts differ ({len(served_points)} vs {len(pinned_points)})"
+                for (ts_a, val_a), (ts_b, val_b) in zip(served_points, pinned_points):
+                    assert ts_a == ts_b, f"{where} {key}: timestamps differ ({ts_a} vs {ts_b})"
+                    assert values_equal(val_a, val_b), f"{where} {key} @{ts_a}: values differ ({val_a} vs {val_b})"
+            if served:
+                served_any_data = True
+
+    assert served_any_data, "fixtures produced no data; the comparison would be over empties"

tests/integration/tests/promqlparity/conftest.py

@@ -0,0 +1,44 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.signoz import create_signoz
+
+
+@pytest.fixture(name="signoz", scope="package")
+def signoz_promql_shadow(
+    network: Network,
+    migrator: types.Operation,  # pylint: disable=unused-argument
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    """
+    Package-scoped SigNoz with the PromQL shadow-comparison flag on: every
+    PromQL query is served from the default engine path and re-run on the
+    clickhousev2 provider, with differences logged.
+    """
+    # The clickhousev2 provider assumes the timeSeries*ToGrid aggregate
+    # functions exist (ClickHouse >= 25.6); older versions cannot serve the
+    # pinned side of the comparison at all.
+    version = pytestconfig.getoption("--clickhouse-version")
+    major, minor = (int(part) for part in str(version).split(".")[:2])
+    if (major, minor) < (25, 6):
+        pytest.skip(f"clickhousev2 requires ClickHouse >= 25.6, matrix leg runs {version}")
+
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-promql-shadow",
+        env_overrides={
+            "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_USE__PROMETHEUS__CLICKHOUSE__V2": True,
+        },
+    )