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/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/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,
+        },
+    )