test(promql): replay the conformance corpus on both providers

Every corpus case now runs twice — default provider, and pinned to
clickhousev2 via the flag-gated X-SigNoz-PromQL-Provider header — each leg
asserted against the same frozen expectations, each leg with its own
known-divergences ledger enforced in both directions. The new
known_divergences_v2.json is the rollout scorecard: the provider swap is
measured by burning it to empty.

The legs are never asserted against each other: both can sit within one
rounding quantum of the expected value yet differ by up to two quanta at a
rounding boundary, so a leg-vs-leg check would reintroduce the boundary
noise the tolerance absorbs. Anchoring both to the same oracle over the
same ingested bytes already localizes any disagreement to a provider.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

pkg/flagger/registry.go

@@ -14,6 +14,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 {
@@ -106,6 +108,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,85 @@
+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 {
+
+	start, end := c.window(hints)
+
+	samplesQuery, args, err := buildSamplesQuery(start, end, metricNamesFromMatchers(matchers), 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,180 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"math"
+	"slices"
+
+	"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 holds a series lookup's 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
+	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,
+		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,
+	})
+}
+
+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 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, dropping empty-valued
+// labels: empty means "absent" in Prometheus, 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 assembles per-series sample slices from a samples query (raw
+// or last-sample-per-step; same column shape), whose rows arrive ordered by
+// (fingerprint, unix_milli). Fingerprints missing from the lookup are skipped:
+// the samples query's semi-join re-runs the series predicates and can match
+// series registered after the lookup ran — the lookup is the read snapshot.
+// Stale flags map to the engine's StaleNaN. Duplicate
+// timestamps pass through: uniqueness is ingest's job, and v1 feeds them to
+// the engine as-is over the same dirty data.
+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)
+		unknownCount int
+	)
+
+	for rows.Next() {
+		if err := rows.Scan(&fingerprint, &timestampMs, &val, &flags); err != nil {
+			return nil, err
+		}
+
+		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 {
+			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
+}

pkg/prometheus/clickhouseprometheusv2/doc.go

@@ -0,0 +1,49 @@
+// 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. Here the stock promql engine evaluates over a
+// native storage.Querier: no translation layer, per-selector fetch windows,
+// and fetch reductions that are provably invisible to the engine. Every
+// reduction either preserves engine semantics exactly or is not performed.
+//
+// # Series lookup
+//
+// Matchers resolve to series once per selector (selectSeries) against the
+// time-series tables, which 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.
+// How matchers become SQL, and why regexes are anchored, is documented at
+// applySeriesConditions. Empty-valued labels come off at this boundary: an
+// empty value means "label absent" in Prometheus, but stored attribute JSON
+// can carry them.
+//
+// # Sample fetch
+//
+// 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 — see
+// lastSamplePerStep for the correctness argument — while range selectors
+// always fetch raw: every sample feeds the range function. Row assembly maps
+// stale flags to the engine's StaleNaN and merges series with identical
+// label sets (sortAndMerge), because the engine assumes storages never emit
+// duplicates.
+//
+// # 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 samples
+// fetch exploits that: it restricts by a shard-local series subquery, not a
+// GLOBAL broadcast of the matched set. Delta-temporality
+// series stay invisible to PromQL exactly as they are in v1: the rollout
+// gate is parity with v1, and 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, so this provider's work is attributable in system.query_log.
+package clickhouseprometheusv2

pkg/prometheus/clickhouseprometheusv2/provider.go

@@ -0,0 +1,72 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/prometheus/prometheus/storage"
+)
+
+// Provider ties the package together: its own engine and parser, and the
+// ClickHouse client behind the native storage.Querier. See the package
+// documentation for how the read path differs from v1. It is exported as a
+// concrete type — callers hold it directly, 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
+}
+
+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,
+	}, nil
+}
+
+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,170 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"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: Select builds SQL
+// directly from the matchers and hints, with no remote-read protobuf layer.
+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 {
+	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)
+	}
+
+	// The engine assumes storages never emit duplicate label sets.
+	list = sortAndMerge(list)
+	return newSeriesSet(list)
+}
+
+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
+}
+
+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 from the hints (already
+// adjusted for offset, @, range and lookback); mint/maxt span the union of
+// all selectors and are the fallback.
+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 (see lastSamplePerStep). Only instant selectors
+// (hints.Range == 0) of subquery-free queries qualify: range selectors need
+// every raw sample, and subquery selectors evaluate at the subquery's own
+// step while hints carry the top-level step — the subquery-free proof
+// arrives as QueryTraits in the context. The first evaluation timestamp is
+// recovered as hints.Start + lookback - 1ms, inverting how the engine
+// derives hints.Start.
+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 (see
+// buildSamplesQuery).
+func (q *querier) fetchSamples(ctx context.Context, start, end int64, matchers []*labels.Matcher, lookup *seriesLookup, lastPerStep *lastSamplePerStep) ([]*series, error) {
+	query, args, err := buildSamplesQuery(start, end, lookup.metricNames, 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
+}

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 as parallel slices ordered by
+// timestamp. Deliberately not storage.NewListSeries: that boxes every sample
+// as an interface value, a per-sample allocation this fetch path exists to
+// avoid.
+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 identical label sets
+// by timestamp (first sample wins ties): distinct fingerprints can carry
+// identical label sets, and the engine assumes storages never emit
+// duplicates.
+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,164 @@
+package clickhouseprometheusv2
+
+import (
+	"fmt"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/huandu/go-sqlbuilder"
+	"github.com/prometheus/prometheus/model/labels"
+)
+
+// 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 matched series: a
+// semi-join re-runs the series predicates against the shard-local series
+// table (complete by fingerprint co-locality, see localTimeSeriesTable — a
+// GLOBAL broadcast would ship the matched set to every shard, and ClickHouse
+// materializes the subquery's set per shard before the scan, so it still
+// engages the fingerprint primary-key column). metricNames (observed on the
+// matched series when the selector had no __name__ equality) narrows the
+// primary-key scan. A non-nil lastPerStep groups to the last sample per
+// step bucket.
+func buildSamplesQuery(start, end int64, metricNames []string, 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; the primary-key prefix goes unused.
+	case 1:
+		sb.Where(sb.EQ("metric_name", metricNames[0]))
+	default:
+		sb.Where(sb.In("metric_name", sqlbuilder.List(metricNames)))
+	}
+	// Semantically redundant (the fingerprints already come from these
+	// temporalities) but engages the leading primary-key column.
+	sb.Where("temporality IN ['Cumulative', 'Unspecified']")
+	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:
+// __name__ matchers (all four types) translate to the metric_name column,
+// every other matcher 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, ClickHouse match()
+// searches for a substring.
+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']")
+	// Inclusive upper bound: registration rows are hour-floored (and 6h/1d/1w
+	// for the rollup tables) by the exporter, so a series first registered in
+	// the bucket starting exactly at `end` would otherwise be invisible while
+	// its samples (<= end) are in range.
+	sb.Where(sb.GTE("unix_milli", start), sb.LTE("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 + ")$"
+}
+
+// lastSamplePerStep reduces an instant-selector fetch to the last sample of
+// each step bucket: bucket 0 is (start, firstEval], bucket i is
+// (firstEval+(i-1)·step, firstEval+i·step]. Anchoring buckets at the first
+// evaluation timestamp makes every bucket boundary an evaluation timestamp,
+// so a non-final sample of a bucket can never be the latest sample in any
+// (t-lookback, t] the engine resolves — the reduction is lossless. 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,139 @@
+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 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 fetch filters by a shard-local semi-join", func(t *testing.T) {
+		query, args, err := buildSamplesQuery(start, end, []string{"up"}, 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")
+		assert.Contains(t, query, "ORDER BY fingerprint, unix_milli")
+		// 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("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"}, 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"}, matchers, lastPerStep)
+		require.NoError(t, err)
+		assert.Contains(t, query, "GROUP BY fingerprint ORDER BY fingerprint, ts")
+		assert.NotContains(t, query, "intDiv")
+	})
+
+	t.Run("multiple metric names from regex selector", func(t *testing.T) {
+		query, args, err := buildSamplesQuery(start, end, []string{"node_cpu", "node_memory"}, matchers, nil)
+		require.NoError(t, err)
+		assert.Contains(t, query, "metric_name IN (?, ?)")
+		assert.Equal(t, []any{"node_cpu", "node_memory", "up", adjustedStart, end, "job", "api", start, end}, args)
+	})
+}

pkg/prometheus/clickhouseprometheusv2/tables.go

@@ -0,0 +1,70 @@
+package clickhouseprometheusv2
+
+import "time"
+
+// TODO(srikanthccv): consolidate the metrics table names and the
+// window-to-table selection across the query builders (telemetrymetrics)
+// and the prometheus providers instead of each package carrying its own
+// copy.
+
+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/config.go

@@ -24,6 +24,10 @@ 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"`
 }
 
 func NewConfigFactory() factory.ConfigFactory {
@@ -37,7 +41,8 @@ func newConfig() factory.Config {
 			Path:          "",
 			MaxConcurrent: 20,
 		},
-		Timeout: 2 * time.Minute,
+		Timeout:      2 * time.Minute,
+		ProviderName: "clickhouse",
 	}
 }
 
@@ -45,9 +50,15 @@ 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)
+	}
 	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

@@ -231,7 +231,7 @@ func (q *querier) buildPreviewProviders(
 			sub.CompositeQuery = qbtypes.CompositeQuery{Queries: []qbtypes.QueryEnvelope{query}}
 		}
 
-		built, _, bErr := q.buildQueries(orgID, &sub, deps, missingMetricQuerySet, event)
+		built, _, bErr := q.buildQueries(orgID, &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"
@@ -98,6 +102,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 +132,7 @@ func newPromqlQuery(
 	tr qbv5.TimeRange,
 	requestType qbv5.RequestType,
 	variables map[string]qbv5.VariableItem,
+	opts promqlOptions,
 ) *promqlQuery {
 	return &promqlQuery{
 		logger:      logger,
@@ -119,10 +142,19 @@ 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 ""
+	}
 	if q.requestType != qbv5.RequestTypeTimeSeries {
 		return ""
 	}
@@ -252,7 +284,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,
@@ -296,6 +337,41 @@ 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: 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
+	}
+
 	qry, err := q.promEngine.Engine().NewRangeQuery(
 		ctx,
 		q.promEngine.Storage(),
@@ -331,6 +407,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 {
 	// Hide only known SigNoz storage keys: label names are user data and may
 	// legitimately start with "__" (e.g. __address__), so a blanket dunder
 	// strip mangles user labelsets. The __scope./__resource. prefixes cover
@@ -366,7 +470,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()
 
 	tsData := &qbv5.TimeSeriesData{
 		QueryName: q.query.Name,
@@ -400,6 +510,6 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
 		Type:     q.requestType,
 		Value:    payload,
 		Warnings: warnings,
-		// TODO: map promql stats?
-	}, nil
+		Stats:    stats,
+	}
 }

pkg/querier/promql_query_test.go

@@ -7,6 +7,7 @@ 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/stretchr/testify/assert"
 )
@@ -440,3 +441,15 @@ func TestQuotedMetricOutsideBracesPattern(t *testing.T) {
 		})
 	}
 }
+
+// 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,176 @@
+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 (the engine over the v2 querier), 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, 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.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) {
+	return executeOnProvider(ctx, q.opts.serve, query, time.Unix(0, startNs), time.Unix(0, endNs), q.query.Step.Duration)
+}
+
+// executeOnProvider evaluates the query the way the provider would serve it:
+// the engine over the provider's storage. 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, error) {
+	qry, err := prov.Engine().NewRangeQuery(ctx, prov.Storage(), nil, query, start, end, step)
+	if err != nil {
+		return nil, err
+	}
+	defer qry.Close()
+
+	res := qry.Exec(ctx)
+	if res.Err != nil {
+		return nil, res.Err
+	}
+	matrix, err := res.Matrix()
+	if err != nil {
+		return nil, err
+	}
+	// Close returns the result's sample slices to the engine pool.
+	return copyMatrix(matrix), 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 sorts by label set for order-independent
+// comparison. Both providers now resolve series identity the same way
+// (empty-valued labels dropped at read, no synthetic fingerprint label
+// since the v1 series-identity fix), so labels need no normalization.
+func normalizeShadowMatrix(matrix promql.Matrix) promql.Matrix {
+	out := make(promql.Matrix, 0, len(matrix))
+	out = append(out, matrix...)
+	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,67 @@
+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", "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 label set; labels pass through untouched — both providers
+	// resolve series identity identically since the v1 series-identity fix
+	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(orgID, req, dependencyQueries, missingMetricQuerySet, event)
+	promqlOpts, err := q.promqlOptions(ctx, orgID, req)
+	if err != nil {
+		return nil, err
+	}
+	queries, steps, err := q.buildQueries(orgID, req, dependencyQueries, missingMetricQuerySet, event, promqlOpts)
 	if err != nil {
 		return nil, err
 	}
@@ -175,12 +200,41 @@ 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(
 	orgID valuer.UUID,
 	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
@@ -205,7 +259,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:
@@ -848,7 +902,7 @@ func (q *querier) createRangedQuery(_ valuer.UUID, originalQuery qbtypes.Query,
 	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) {
@@ -180,6 +183,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
@@ -100,6 +101,7 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
 		telemetryStore,
 		metadataStore,
 		nil,            // prometheus
+		nil,            // promV2
 		nil,            // traceStmtBuilder
 		logStmtBuilder, // logStmtBuilder
 		nil,            // auditStmtBuilder
@@ -149,6 +151,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

@@ -45,6 +45,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"
@@ -244,6 +245,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),
 	)
 }
 
@@ -285,9 +287,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

@@ -40,6 +40,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"
@@ -242,6 +243,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,
@@ -254,12 +260,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/fixtures/querier.py

@@ -168,6 +168,7 @@ def make_query_request(
     variables: dict | None = None,
     no_cache: bool = True,
     timeout: int = QUERY_TIMEOUT,
+    headers: dict | None = None,
 ) -> requests.Response:
     if format_options is None:
         format_options = {"formatTableResultForUI": False, "fillGaps": False}
@@ -187,7 +188,7 @@ def make_query_request(
     return requests.post(
         signoz.self.host_configs["8080"].get("/api/v5/query_range"),
         timeout=timeout,
-        headers={"authorization": f"Bearer {token}"},
+        headers={"authorization": f"Bearer {token}", **(headers or {})},
         json=payload,
     )
 

tests/integration/testdata/promqltestcorpus/known_divergences_v2.json

@@ -0,0 +1,4 @@
+{
+  "note": "Divergences of the clickhousev2 provider (pinned via X-SigNoz-PromQL-Provider) from the upstream reference engine, enforced exactly by 01_upstream_corpus.py in both directions. This ledger is the rollout scorecard for the provider swap: the default provider cannot be replaced by clickhousev2 while anything is listed here. Entries must carry the defect's cause and be REMOVED as the provider is fixed.",
+  "divergences": {}
+}

tests/integration/tests/promqlconformance/01_upstream_corpus.py

@@ -3,13 +3,26 @@ Upstream promqltest conformance: replay the frozen corpus extracted from
 Prometheus' own promql/promqltest testdata and assert our API returns the
 reference engine's answers.
 
-Unlike the parity suites, the oracle here is a committed file
+Unlike live-vs-live parity suites, the oracle here is a committed file
 (tests/integration/testdata/promqltestcorpus/corpus.json), generated by
 scripts/promqltestcorpus from upstream's load scripts and the vendored
 reference engine. It therefore keeps working when the serving path itself is
 the thing being changed — the one situation where comparing two live paths
 against each other is blind.
 
+Every case replays on both serving paths — the default provider, and the
+clickhousev2 provider pinned via the flag-gated X-SigNoz-PromQL-Provider
+header (see conftest.py) — and each leg is asserted against the same frozen
+expectations, each leg against its own known-divergences ledger. The legs
+are deliberately never asserted against each other: both can sit within one
+rounding quantum of the expected value yet differ from each other by up to
+two quanta when a true value straddles a rounding boundary, so a leg-vs-leg
+equality check would reintroduce exactly the boundary noise the quantum
+tolerance exists to absorb. Because both legs anchor to the same oracle over
+the same ingested bytes, a case failing on one leg while passing on the
+other already localizes the defect to that provider — and the printed
+DIVERGED lines for both legs are the side-by-side view for triage.
+
 Datasets are placed on disjoint time windows (2h isolation gaps, far beyond
 the 5m lookback) so one bulk ingest serves every case without cross-talk.
 Expected values carry the API's 3-significant-decimal rounding, mirrored by
@@ -31,12 +44,30 @@ from fixtures.querier import get_all_series, make_query_request
 
 TESTDATA_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "testdata")
 CORPUS_FILE = os.path.join(TESTDATA_DIR, "promqltestcorpus", "corpus.json")
-KNOWN_DIVERGENCES_FILE = os.path.join(TESTDATA_DIR, "promqltestcorpus", "known_divergences.json")
+
+# One ledger per leg, enforced exactly in both directions. The default leg's
+# ledger is empty and pinned there; the clickhousev2 ledger is the rollout
+# scorecard — the provider swap is measured by burning it down to empty.
+LEDGER_FILES = {
+    "default": os.path.join(TESTDATA_DIR, "promqltestcorpus", "known_divergences.json"),
+    "clickhousev2": os.path.join(TESTDATA_DIR, "promqltestcorpus", "known_divergences_v2.json"),
+}
+
+LEGS: list[tuple[str, dict | None]] = [
+    ("default", None),
+    ("clickhousev2", {"X-SigNoz-PromQL-Provider": "clickhousev2"}),
+]
 
 ISOLATION_GAP_MS = 2 * 3600 * 1000
 SPECIALS = {"NaN": math.nan, "Inf": math.inf, "-Inf": -math.inf}
 
 
+def _decode(v: float | str) -> float:
+    if isinstance(v, str):
+        return SPECIALS[v]
+    return float(v)
+
+
 def _values_close(a: float, b: float) -> bool:
     if math.isnan(a) or math.isnan(b):
         return math.isnan(a) and math.isnan(b)
@@ -59,6 +90,10 @@ def _values_close(a: float, b: float) -> bool:
     return abs(a - b) <= quantum + 1e-12
 
 
+def _labelset(labels: dict[str, str]) -> tuple:
+    return tuple(sorted(labels.items()))
+
+
 def _response_series(data: dict) -> tuple[dict[tuple, dict[int, float]], list[tuple]]:
     """Returns (series map, duplicate labelsets). A response carrying several
     series with identical visible labels is itself a defect signal (e.g. a
@@ -69,14 +104,70 @@ def _response_series(data: dict) -> tuple[dict[tuple, dict[int, float]], list[tu
     # Empty results serialize with null aggregations/series/values fields.
     for series in get_all_series(data, "A") or []:
         lbls = {l["key"]["name"]: str(l["value"]) for l in series.get("labels") or []}
-        points = {int(v["timestamp"]): SPECIALS[v["value"]] if isinstance(v["value"], str) else float(v["value"]) for v in series.get("values") or []}
-        key = tuple(sorted(lbls.items()))
+        points = {int(v["timestamp"]): _decode(v["value"]) for v in series.get("values") or []}
+        key = _labelset(lbls)
         if key in out:
             duplicates.append(key)
         out[key] = points
     return out, duplicates
 
 
+def _case_failure(
+    signoz: types.SigNoz,
+    token: str,
+    case: dict,
+    base: int,
+    headers: dict | None,
+) -> str | None:
+    """Replays one corpus case on one leg; returns a failure line or None."""
+    start_ms = base + case["start_ms"]
+    end_ms = base + case["end_ms"]
+    step_s = max(1, case["step_ms"] // 1000)
+    req_start_ms = start_ms
+    if case["instant"]:
+        # The API rejects start == end; ask for one extra step backward
+        # and compare only at the instant timestamp. Nudging the start
+        # earlier instead of the end later keeps every window that the
+        # expected values were computed from untouched.
+        req_start_ms = start_ms - step_s * 1000
+    query = {
+        "type": "promql",
+        "spec": {"name": "A", "query": case["expr"], "step": step_s},
+    }
+
+    case_id = f"{case['source']}[{case['variant']}]"
+    response = make_query_request(signoz, token, req_start_ms, end_ms, [query], headers=headers)
+    if response.status_code != HTTPStatus.OK:
+        return f"{case_id}: HTTP {response.status_code} for {case['expr']!r}: {response.text[:200]}"
+
+    actual, duplicates = _response_series(response.json())
+    if duplicates:
+        return f"{case_id}: response carries multiple series with identical labels for {case['expr']!r}: {[dict(d) for d in duplicates[:3]]}"
+    if case["instant"]:
+        # Keep only the instant point; the extra grid step is a request
+        # encoding byproduct, not part of the assertion.
+        actual = {lset: {ts: v for ts, v in pts.items() if ts == end_ms} for lset, pts in actual.items()}
+        actual = {lset: pts for lset, pts in actual.items() if pts}
+    expected: dict[tuple, dict[int, float]] = {}
+    for res in case["expected"]:
+        points = {base + off_ms: _decode(v) for off_ms, v in res["points"]}
+        expected[_labelset(res["labels"])] = points
+
+    if set(actual) != set(expected):
+        missing = set(expected) - set(actual)
+        extra = set(actual) - set(expected)
+        return f"{case_id}: series mismatch for {case['expr']!r} (missing={sorted(missing)[:3]} extra={sorted(extra)[:3]}) actual={[(dict(k), {t - base: v for t, v in pts.items()}) for k, pts in actual.items()]}"
+
+    for lset, exp_points in expected.items():
+        act_points = actual[lset]
+        if set(act_points) != set(exp_points):
+            return f"{case_id}: timestamp mismatch for {case['expr']!r} series {dict(lset)} (expected {len(exp_points)} points, got {len(act_points)})"
+        for ts, exp_v in exp_points.items():
+            if not _values_close(act_points[ts], exp_v):
+                return f"{case_id}: value mismatch for {case['expr']!r} series {dict(lset)} at {ts}: expected {exp_v}, got {act_points[ts]}"
+    return None
+
+
 def test_upstream_promqltest_corpus(
     signoz: types.SigNoz,
     create_user_admin: None,  # pylint: disable=unused-argument
@@ -121,7 +212,7 @@ def test_upstream_promqltest_corpus(
                         metric_name=metric_name,
                         labels=labels,
                         timestamp=datetime.fromtimestamp((cursor + off_ms) / 1000, tz=UTC),
-                        value=0.0 if stale else (SPECIALS[raw] if isinstance(raw, str) else float(raw)),
+                        value=0.0 if stale else _decode(raw),
                         flags=1 if stale else 0,
                     )
                 )
@@ -130,79 +221,36 @@ def test_upstream_promqltest_corpus(
     insert_metrics(metrics)
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    failures: list[str] = []
+    failures: dict[str, list[str]] = {leg: [] for leg, _ in LEGS}
     for case in corpus["cases"]:
-        base = bases[case["dataset"]]
-        start_ms = base + case["start_ms"]
-        end_ms = base + case["end_ms"]
-        step_s = max(1, case["step_ms"] // 1000)
-        req_start_ms = start_ms
-        if case["instant"]:
-            # The API rejects start == end; ask for one extra step backward
-            # and compare only at the instant timestamp. Nudging the start
-            # earlier instead of the end later keeps every window that the
-            # expected values were computed from untouched.
-            req_start_ms = start_ms - step_s * 1000
-        query = {
-            "type": "promql",
-            "spec": {"name": "A", "query": case["expr"], "step": step_s},
-        }
+        for leg, headers in LEGS:
+            f_line = _case_failure(signoz, token, case, bases[case["dataset"]], headers)
+            if f_line:
+                failures[leg].append(f_line)
 
-        case_id = f"{case['source']}[{case['variant']}]"
-        response = make_query_request(signoz, token, req_start_ms, end_ms, [query])
-        if response.status_code != HTTPStatus.OK:
-            failures.append(f"{case_id}: HTTP {response.status_code} for {case['expr']!r}: {response.text[:200]}")
-            continue
+    for leg, _ in LEGS:
+        for f_line in failures[leg]:
+            print("DIVERGED", f"[{leg}]", f_line)
 
-        actual, duplicates = _response_series(response.json())
-        if duplicates:
-            failures.append(f"{case_id}: response carries multiple series with identical labels for {case['expr']!r}: {[dict(d) for d in duplicates[:3]]}")
-            continue
-        if case["instant"]:
-            # Keep only the instant point; the extra grid step is a request
-            # encoding byproduct, not part of the assertion.
-            actual = {lset: {ts: v for ts, v in pts.items() if ts == end_ms} for lset, pts in actual.items()}
-            actual = {lset: pts for lset, pts in actual.items() if pts}
-        expected: dict[tuple, dict[int, float]] = {}
-        for res in case["expected"]:
-            points = {base + off_ms: SPECIALS[v] if isinstance(v, str) else float(v) for off_ms, v in res["points"]}
-            expected[tuple(sorted(res["labels"].items()))] = points
-
-        if set(actual) != set(expected):
-            missing = set(expected) - set(actual)
-            extra = set(actual) - set(expected)
-            failures.append(f"{case_id}: series mismatch for {case['expr']!r} (missing={sorted(missing)[:3]} extra={sorted(extra)[:3]}) actual={[(dict(k), {t - base: v for t, v in pts.items()}) for k, pts in actual.items()]}")
-            continue
-
-        for lset, exp_points in expected.items():
-            act_points = actual[lset]
-            if set(act_points) != set(exp_points):
-                failures.append(f"{case_id}: timestamp mismatch for {case['expr']!r} series {dict(lset)} (expected {len(exp_points)} points, got {len(act_points)})")
-                break
-            for ts, exp_v in exp_points.items():
-                if not _values_close(act_points[ts], exp_v):
-                    failures.append(f"{case_id}: value mismatch for {case['expr']!r} series {dict(lset)} at {ts}: expected {exp_v}, got {act_points[ts]}")
-                    break
-            else:
-                continue
-            break
-
-    for f_line in failures:
-        print("DIVERGED", f_line)
-
-    # Known divergences are defects of the current serving path, frozen with
-    # reasons. The set is enforced exactly in both directions: a NEW
+    # Known divergences are defects of that leg's serving path, frozen with
+    # reasons. Each set is enforced exactly in both directions: a NEW
     # divergence is a regression, and a known divergence that starts passing
-    # must be removed from the file — that is the ledger the serving-path
-    # swap is measured against.
-    known: dict[str, str] = {}
-    if os.path.exists(KNOWN_DIVERGENCES_FILE):
-        with open(KNOWN_DIVERGENCES_FILE, encoding="utf-8") as f:
-            known = json.load(f)["divergences"]
+    # must be removed from the file. Problems across both legs are collected
+    # before asserting so one leg's failure never hides the other's.
+    problems: list[str] = []
+    for leg, _ in LEGS:
+        known: dict[str, str] = {}
+        if os.path.exists(LEDGER_FILES[leg]):
+            with open(LEDGER_FILES[leg], encoding="utf-8") as f:
+                known = json.load(f)["divergences"]
 
-    failed_ids = {f_line.split(": ", 1)[0] for f_line in failures}
-    unexpected = [f_line for f_line in failures if f_line.split(": ", 1)[0] not in known]
-    now_passing = sorted(set(known) - failed_ids)
+        failed_ids = {f_line.split(": ", 1)[0] for f_line in failures[leg]}
+        unexpected = [f_line for f_line in failures[leg] if f_line.split(": ", 1)[0] not in known]
+        now_passing = sorted(set(known) - failed_ids)
 
-    assert not unexpected, f"{len(unexpected)} corpus cases diverged beyond the known set:\n" + "\n".join(unexpected[:25])
-    assert not now_passing, f"{len(now_passing)} known divergences now pass — remove them from known_divergences.json: {now_passing[:25]}"
+        if unexpected:
+            problems.append(f"[{leg}] {len(unexpected)} corpus cases diverged beyond the known set:\n" + "\n".join(unexpected[:25]))
+        if now_passing:
+            problems.append(f"[{leg}] {len(now_passing)} known divergences now pass — remove them from {os.path.basename(LEDGER_FILES[leg])}: {now_passing[:25]}")
+
+    assert not problems, "\n\n".join(problems)

tests/integration/tests/promqlconformance/conftest.py

@@ -0,0 +1,39 @@
+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_conformance(
+    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 use_prometheus_clickhouse_v2 on, so the corpus
+    can replay every case twice: once against the default provider and once
+    pinned to the clickhousev2 provider via the X-SigNoz-PromQL-Provider
+    header (which the flag gates). Each leg is asserted against the same
+    frozen expectations — see 01_upstream_corpus.py for why the legs are
+    never asserted against each other.
+    """
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-promql-conformance",
+        env_overrides={
+            "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_USE__PROMETHEUS__CLICKHOUSE__V2": True,
+        },
+    )