feat(promql): transpile allowlisted query shapes to ClickHouse grid statements

An allowlist compiler (classify/rewrite) evaluates proven PromQL shapes
entirely inside ClickHouse on the timeSeries*ToGrid aggregate functions
(CH >= 25.6): one row per output series comes back instead of every raw
sample. Everything not provably equivalent falls back to the engine over
the native querier; transpilable subtrees under non-transpilable nodes run
hybrid (materialized as synthetic series, engine on top). TryExecuteRange
slots into the serve/shadow paths, which until now ran engine-only.

Window-sliver filtering folded in: when the window is narrower than the
step, a lattice predicate admits only the samples any grid point can see -
measured 74s/28GiB -> 16s/4.3GiB on a 36k-series 1w rate, and a
2.67B-sample 1w case that died at 150GiB completes in 19s/17GiB. Over
slivered rows the last-style gates lift and disjoint over_time forms drop
the divisibility gate.

The classification golden freezes the routing decision (full/hybrid/
fallback + reason) for every conformance-corpus expression: silently
falling back costs the pushdown, silently transpiling an unproven shape
risks wrong numbers - both now surface in review as a golden diff.

The dual-leg conformance suite already earned its keep on its first
transpiled run:

- It caught the classifier reading a duration expression's offset
  (x offset step()) as zero - offset expressions parse WITHOUT the
  experimental-parser flag, so they reach production. Such selectors are
  now refused and the engine evaluates them exactly.
- It caught name-drop assembly treating temporally-disjoint same-labelset
  twins as separate series (-{job="api"} spanning http_requests and
  http_errors 400'd; hybrid -metric_a or -metric_b returned duplicate {}
  series). Both paths now merge by labelset slot-wise, raising the
  engine's duplicate error only on a same-timestamp conflict - the
  engine's actual rule.
- The 12 remaining divergences are one class, recorded with causes in
  known_divergences_v2.json (the swap scorecard): the engine sums with
  Kahan compensation and an overflow-free incremental mean, ClickHouse's
  sum/avg/arraySum are naive - visible only at 1e100-class cancellation
  and near-max-float overflow.

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

.github/workflows/cacheci.yml

@@ -1,92 +0,0 @@
-name: cacheci
-
-on:
-  push:
-    branches:
-      - main
-  workflow_dispatch:
-
-permissions:
-  contents: read
-  actions: write
-
-# Cancelling mid-rotation is safe: the sequential delete-then-save order
-# leaves at most one key missing at any moment.
-concurrency:
-  group: cacheci
-  cancel-in-progress: true
-
-jobs:
-  tests:
-    runs-on: ubuntu-latest
-    steps:
-      - name: checkout
-        uses: actions/checkout@v4
-      - name: restore
-        id: restore
-        uses: actions/cache/restore@v4
-        with:
-          path: ${{ runner.temp }}/cacheci
-          key: tests-primary
-          restore-keys: |
-            tests-secondary
-      - name: inject
-        if: steps.restore.outputs.cache-matched-key != ''
-        run: |
-          cat > "$RUNNER_TEMP/inject.Dockerfile" <<'EOF'
-          FROM busybox:1.37
-          RUN --mount=type=cache,target=/root/.cache/go-build \
-              --mount=type=cache,target=/go/pkg/mod \
-              --mount=type=cache,target=/pnpm/store \
-              --mount=type=bind,target=/restored \
-              tar -xf /restored/go-build.tar -C /root/.cache/go-build && \
-              tar -xf /restored/go-mod.tar -C /go/pkg/mod && \
-              tar -xf /restored/pnpm-store.tar -C /pnpm/store
-          EOF
-          docker build -f "$RUNNER_TEMP/inject.Dockerfile" "$RUNNER_TEMP/cacheci"
-      - name: build
-        run: |
-          docker build -f cmd/enterprise/Dockerfile.integration --build-arg TARGETARCH=amd64 --build-arg ZEUSURL=http://zeus:8080 .
-          docker build -f cmd/enterprise/Dockerfile.with-web.integration --build-arg TARGETARCH=amd64 --build-arg ZEUSURL=http://zeus:8080 .
-      # docker cp instead of --output type=local (the local exporter stalls on
-      # multi-GB outputs); tarballs instead of raw trees so the host never hits
-      # the permission and symlink semantics that broke docker cp.
-      - name: extract
-        run: |
-          rm -rf "$RUNNER_TEMP/cacheci"
-          mkdir -p "$RUNNER_TEMP/cacheci" "$RUNNER_TEMP/extract-context"
-          cat > "$RUNNER_TEMP/extract.Dockerfile" <<'EOF'
-          FROM busybox:1.37
-          RUN --mount=type=cache,target=/root/.cache/go-build \
-              --mount=type=cache,target=/go/pkg/mod \
-              --mount=type=cache,target=/pnpm/store \
-              mkdir -p /out && \
-              tar -cf /out/go-build.tar -C /root/.cache/go-build . && \
-              tar -cf /out/go-mod.tar -C /go/pkg/mod . && \
-              tar -cf /out/pnpm-store.tar -C /pnpm/store .
-          EOF
-          docker build -f "$RUNNER_TEMP/extract.Dockerfile" -t cacheci-extract "$RUNNER_TEMP/extract-context"
-          id=$(docker create cacheci-extract)
-          docker cp "$id":/out/. "$RUNNER_TEMP/cacheci/"
-          docker rm "$id"
-      # Fixed cache keys are immutable, so each key must be deleted before it
-      # can be saved again. Rotating primary and secondary one after the other
-      # keeps at least one key restorable for concurrent test runs.
-      - name: delete-primary
-        env:
-          GH_TOKEN: ${{ github.token }}
-        run: gh cache delete tests-primary --repo "$GITHUB_REPOSITORY" || true
-      - name: save-primary
-        uses: actions/cache/save@v4
-        with:
-          path: ${{ runner.temp }}/cacheci
-          key: tests-primary
-      - name: delete-secondary
-        env:
-          GH_TOKEN: ${{ github.token }}
-        run: gh cache delete tests-secondary --repo "$GITHUB_REPOSITORY" || true
-      - name: save-secondary
-        uses: actions/cache/save@v4
-        with:
-          path: ${{ runner.temp }}/cacheci
-          key: tests-secondary

.github/workflows/e2eci.yaml

@@ -75,30 +75,6 @@ jobs:
           docker rm pw
           echo "PLAYWRIGHT_BROWSERS_PATH=$RUNNER_TEMP/ms-playwright" >> "$GITHUB_ENV"
           cd tests/e2e && pnpm playwright install-deps ${{ matrix.project }}
-      # Restore-only: the cacheci workflow owns cache saves. Seeds the
-      # BuildKit cache mounts so the in-test image build is incremental.
-      - name: restore
-        id: restore
-        uses: actions/cache/restore@v4
-        with:
-          path: ${{ runner.temp }}/cacheci
-          key: tests-primary
-          restore-keys: |
-            tests-secondary
-      - name: inject
-        if: steps.restore.outputs.cache-matched-key != ''
-        run: |
-          cat > "$RUNNER_TEMP/inject.Dockerfile" <<'EOF'
-          FROM busybox:1.37
-          RUN --mount=type=cache,target=/root/.cache/go-build \
-              --mount=type=cache,target=/go/pkg/mod \
-              --mount=type=cache,target=/pnpm/store \
-              --mount=type=bind,target=/restored \
-              tar -xf /restored/go-build.tar -C /root/.cache/go-build && \
-              tar -xf /restored/go-mod.tar -C /go/pkg/mod && \
-              tar -xf /restored/pnpm-store.tar -C /pnpm/store
-          EOF
-          docker build -f "$RUNNER_TEMP/inject.Dockerfile" "$RUNNER_TEMP/cacheci"
       - name: bring-up-stack
         run: |
           cd tests && \

.github/workflows/integrationci.yaml

@@ -110,30 +110,6 @@ jobs:
           sudo mv chromedriver-linux64/chromedriver /usr/local/bin/chromedriver
           chromedriver -version
           google-chrome-stable --version
-      # Restore-only: the cacheci workflow owns cache saves. Seeds the
-      # BuildKit cache mounts so the in-test image build is incremental.
-      - name: restore
-        id: restore
-        uses: actions/cache/restore@v4
-        with:
-          path: ${{ runner.temp }}/cacheci
-          key: tests-primary
-          restore-keys: |
-            tests-secondary
-      - name: inject
-        if: steps.restore.outputs.cache-matched-key != ''
-        run: |
-          cat > "$RUNNER_TEMP/inject.Dockerfile" <<'EOF'
-          FROM busybox:1.37
-          RUN --mount=type=cache,target=/root/.cache/go-build \
-              --mount=type=cache,target=/go/pkg/mod \
-              --mount=type=cache,target=/pnpm/store \
-              --mount=type=bind,target=/restored \
-              tar -xf /restored/go-build.tar -C /root/.cache/go-build && \
-              tar -xf /restored/go-mod.tar -C /go/pkg/mod && \
-              tar -xf /restored/pnpm-store.tar -C /pnpm/store
-          EOF
-          docker build -f "$RUNNER_TEMP/inject.Dockerfile" "$RUNNER_TEMP/cacheci"
       - name: run
         run: |
           cd tests && \

Makefile

@@ -209,8 +209,8 @@ py-lint: ## Run ruff check across the shared tests project
 	@cd tests && uv run ruff check --fix .
 
 .PHONY: py-test-setup
-py-test-setup: ## Bring up the shared SigNoz backend used by integration and e2e tests, rebuilding signoz from the current sources
-	@cd tests && uv run pytest --basetemp=./tmp/ -vv --reuse --rebuild --capture=no integration/bootstrap/setup.py::test_setup
+py-test-setup: ## Bring up the shared SigNoz backend used by integration and e2e tests
+	@cd tests && uv run pytest --basetemp=./tmp/ -vv --reuse --capture=no integration/bootstrap/setup.py::test_setup
 
 .PHONY: py-test-teardown
 py-test-teardown: ## Tear down the shared SigNoz backend

cmd/enterprise/Dockerfile.integration

@@ -4,13 +4,9 @@ ARG OS="linux"
 ARG TARGETARCH
 ARG ZEUSURL
 
-# HOME comes from the build user, not the image config; declare it so the
-# /root paths below trace to it.
-ENV HOME=/root
-
 # This path is important for stacktraces
 WORKDIR $GOPATH/src/github.com/signoz/signoz
-WORKDIR $HOME
+WORKDIR /root
 
 RUN set -eux; \
 	apt-get update; \
@@ -18,36 +14,23 @@ RUN set -eux; \
 		g++ \
 		gcc \
 		libc6-dev \
+		make \
 		pkg-config \
 	; \
 	rm -rf /var/lib/apt/lists/*
 
-# Keep the literal cache-mount targets below in sync with these. The caches
-# are shared with Dockerfile.with-web.integration (same target paths).
-ENV GOCACHE=$HOME/.cache/go-build
-ENV GOMODCACHE=$GOPATH/pkg/mod
-
 COPY go.mod go.sum ./
 
-RUN --mount=type=cache,target=/go/pkg/mod \
-	go mod download
+RUN go mod download
 
 COPY ./cmd/ ./cmd/
 COPY ./ee/ ./ee/
 COPY ./pkg/ ./pkg/
 COPY ./templates /root/templates
 
-# Invoked directly instead of via make so Makefile changes don't invalidate
-# this layer; the Makefile's git-derived ldflags resolve to empty in here
-# anyway (.git is dockerignored).
-RUN --mount=type=cache,target=/go/pkg/mod \
-	--mount=type=cache,target=/root/.cache/go-build \
-	GOARCH=${TARGETARCH} GOOS=${OS} go build -C ./cmd/enterprise -race -tags timetzdata -o /root/signoz \
-	-ldflags "-s -w \
-		-X github.com/SigNoz/signoz/pkg/version.version=integration \
-		-X github.com/SigNoz/signoz/pkg/version.variant=enterprise \
-		-X github.com/SigNoz/signoz/ee/zeus.url=${ZEUSURL} \
-		-X github.com/SigNoz/signoz/ee/zeus.deprecatedURL=${ZEUSURL}/api/v1"
+COPY Makefile Makefile
+RUN TARGET_DIR=/root ARCHS=${TARGETARCH} ZEUS_URL=${ZEUSURL} LICENSE_URL=${ZEUSURL}/api/v1 make go-build-enterprise-race
+RUN mv /root/linux-${TARGETARCH}/signoz /root/signoz
 
 RUN chmod 755 /root /root/signoz
 

cmd/enterprise/Dockerfile.with-web.integration

@@ -1,23 +1,10 @@
 FROM node:22-bookworm AS build
 
 WORKDIR /opt/
-
-# HOME comes from the build user, not the image config.
-ENV HOME=/root
-# pnpm's store lives at $PNPM_HOME/store — a dedicated directory pnpm
-# manages. Keep the literal cache-mount targets below in sync.
-ENV PNPM_HOME=/pnpm
-ENV NODE_OPTIONS=--max-old-space-size=8192
-
-RUN CI=1 npm i -g pnpm@10
-
-# pnpm fetch resolves from the lockfile alone and runs no lifecycle scripts;
-# the repo's postinstall needs source files that are not copied yet.
-COPY ./frontend/package.json ./frontend/pnpm-lock.yaml ./frontend/pnpm-workspace.yaml ./
-RUN --mount=type=cache,target=/pnpm/store CI=1 pnpm fetch
-
 COPY ./frontend/ ./
-RUN --mount=type=cache,target=/pnpm/store CI=1 pnpm install --offline
+ENV NODE_OPTIONS=--max-old-space-size=8192
+RUN CI=1 npm i -g pnpm@10
+RUN CI=1 pnpm install
 RUN CI=1 pnpm build
 
 FROM golang:1.25-bookworm
@@ -26,13 +13,9 @@ ARG OS="linux"
 ARG TARGETARCH
 ARG ZEUSURL
 
-# HOME comes from the build user, not the image config; declare it so the
-# /root paths below trace to it.
-ENV HOME=/root
-
 # This path is important for stacktraces
 WORKDIR $GOPATH/src/github.com/signoz/signoz
-WORKDIR $HOME
+WORKDIR /root
 
 RUN set -eux; \
 	apt-get update; \
@@ -40,36 +23,23 @@ RUN set -eux; \
 		g++ \
 		gcc \
 		libc6-dev \
+		make \
 		pkg-config \
 	; \
 	rm -rf /var/lib/apt/lists/*
 
-# Keep the literal cache-mount targets below in sync with these. The caches
-# are shared with Dockerfile.integration (same target paths).
-ENV GOCACHE=$HOME/.cache/go-build
-ENV GOMODCACHE=$GOPATH/pkg/mod
-
 COPY go.mod go.sum ./
 
-RUN --mount=type=cache,target=/go/pkg/mod \
-	go mod download
+RUN go mod download
 
 COPY ./cmd/ ./cmd/
 COPY ./ee/ ./ee/
 COPY ./pkg/ ./pkg/
 COPY ./templates /root/templates
 
-# Invoked directly instead of via make so Makefile changes don't invalidate
-# this layer; the Makefile's git-derived ldflags resolve to empty in here
-# anyway (.git is dockerignored).
-RUN --mount=type=cache,target=/go/pkg/mod \
-	--mount=type=cache,target=/root/.cache/go-build \
-	GOARCH=${TARGETARCH} GOOS=${OS} go build -C ./cmd/enterprise -race -tags timetzdata -o /root/signoz \
-	-ldflags "-s -w \
-		-X github.com/SigNoz/signoz/pkg/version.version=integration \
-		-X github.com/SigNoz/signoz/pkg/version.variant=enterprise \
-		-X github.com/SigNoz/signoz/ee/zeus.url=${ZEUSURL} \
-		-X github.com/SigNoz/signoz/ee/zeus.deprecatedURL=${ZEUSURL}/api/v1"
+COPY Makefile Makefile
+RUN TARGET_DIR=/root ARCHS=${TARGETARCH} ZEUS_URL=${ZEUSURL} LICENSE_URL=${ZEUSURL}/api/v1 make go-build-enterprise-race
+RUN mv /root/linux-${TARGETARCH}/signoz /root/signoz
 
 COPY --from=build /opt/build ./web/
 

docs/contributing/prometheus.md

@@ -2,8 +2,8 @@
 
 This document is the subsystem context for `pkg/prometheus/clickhouseprometheusv2`,
 the second-generation ClickHouse-backed Prometheus provider. It explains why the
-package exists, the correctness constraints that shaped it, and how each fetch
-reduction is proven not to change results. Any change to the provider must keep
+package exists, the correctness constraints that shaped it, and how each
+construct is proven not to change results. Any change to the provider must keep
 these invariants; if a change would violate one, it must be flagged and
 discussed.
 
@@ -17,107 +17,337 @@ window is fetched, serialized, and handed to the engine. The cost is a function
 of ingested data, not of the question asked — which is how a dashboard of PromQL
 panels can take an instance down.
 
-In v2 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.
+In v2, every query runs in one of two ways, decided per query:
 
-**The core constraint: every reduction either preserves engine semantics exactly
-or is not performed.** A PromQL result that differs from upstream Prometheus is
-a lost user. The conformance suite
+- **Transpiled**: the query is evaluated entirely inside ClickHouse and only
+  final (or near-final) per-group grid arrays come back, built on the
+  `timeSeries*ToGrid` aggregate functions (the supported ClickHouse floor is
+  >= 25.6, so they are assumed available).
+- **Engine**: the stock promql engine evaluates over this package's native
+  `storage.Querier`. This is the path for everything not transpilable.
+
+**The core constraint: a PromQL result that differs from upstream Prometheus is
+a lost user, so anything that cannot reproduce engine semantics exactly falls
+back rather than approximate.** The conformance suite
 (`tests/integration/tests/promqlconformance/`) replays Prometheus' own test
-corpus against both providers and is the arbiter.
+corpus against both providers and is the arbiter; the classification golden
+(`testdata/classification_golden.json`) freezes which of the two ways each
+corpus expression takes. The rest of this document is the PromQL -> SQL story,
+because that mapping is where correctness is won or lost.
 
 ---
 
+## The evaluation model the SQL must reproduce
+
+A PromQL range query is an instant query evaluated at every grid point
+t_i = start + i*step, i = 0..(end-start)/step. At each t_i:
+
+  - an instant selector resolves to the latest sample in the left-open
+    lookback window (t_i - lookback, t_i], and to nothing when that latest
+    sample is a stale marker — even if older real samples sit inside the
+    window;
+  - a range selector [r] collects every sample in (t_i - r, t_i], stale
+    markers excluded;
+  - offset d shifts both windows to (t_i - d - w, t_i - d].
+
+The transpilation invariant follows from this: every transpiled construct
+produces, per output series, one array with exactly one slot per grid
+point — slot i holds the value at t_i, NULL means absent. This is what
+makes composition correct, not just convenient: the engine evaluates
+these operators independently per t_i, so any representation that gets
+every slot right gets the whole query right, and spatial aggregation over
+arrays is sound because it combines values that belong to the same t_i by
+construction. Slot index i maps back to t_i = start + i*step at scan time
+(toMatrix). Everything below is about filling those slots with exactly
+the numbers the engine would compute — and each equivalence was validated
+against the vendored engine on live data before its shape entered the
+allowlist; anything unproven stays on the engine path.
+
+## Classification: finding what a statement can answer
+
+classify walks the parsed AST looking for "core units" — maximal subtrees
+of the shape
+
+	[agg by/without (...)] [fn(] selector[range] [offset d] [)] [op scalar]...
+
+classifyCore peels that chain from the outside in: an optional
+sum/min/max/avg/count aggregation, then one of the allowlisted functions
+or a bare instant selector, then the selector with its offset; on the way
+out it accumulates number-literal arithmetic, comparisons (including
+bool) and unary minus into a scalar-op pipeline. A node qualifies only if
+its type, arguments and children are in the proven set — an allowlist, so
+an overlooked construct becomes a fallback instead of a wrong number.
+
+Three unit kinds come out of this, each with its own SQL form:
+unitRange (rate, irate, increase, delta, idelta over a range selector),
+unitInstant (instant vector selection, bare or comparison-filtered) and
+unitOverTime (avg/min/max/sum/count/last _over_time).
+
+If the entire tree is one unit, the plan is "full": the statement's rows
+are the query result. Otherwise every maximal unit is cut out and replaced
+in the expression with a synthetic selector __signoz_transpiled_N__, and
+the rewritten expression runs in the engine over the units' materialized
+results ("hybrid") — histogram_quantile, topk, or/and/unless and vector
+matching keep exact engine semantics while their expensive inputs were
+aggregated server-side.
+
+Classification refuses when exact semantics cannot be guaranteed
+server-side: the @ modifier anywhere and default-resolution subqueries
+(their resolution is a server runtime setting the transpiler cannot see);
+duration expressions (offset step(), [range()], ...) anywhere — they are
+resolved into the selector's static fields only at evaluation time, so at
+classification time those fields still hold their zero values and
+transpiling would silently use the wrong offset or range;
+steps or ranges that are not whole seconds (the grid functions take
+whole-second parameters); grouping by or matching on __name__ in hybrid
+plans (the synthetic name would leak into results); name-keeping units —
+bare/comparison instant selectors and last_over_time keep their real
+__name__ (keepsName), which substitution would replace, so they transpile
+only as full plans; and every function outside the allowlist (changes,
+resets, quantile_over_time, absent, native-histogram functions, ...).
+
+Units inside a fixed-resolution subquery evaluate on the subquery's own
+grid instead of the query grid: epoch-aligned multiples of the resolution
+strictly after outerStart - offset - range, ending at outer end - offset —
+the exact derivation the engine uses, because a grid shifted by one step
+changes which samples every window sees.
+
+## From one unit to one statement
+
+buildUnitSQL renders each unit as a single statement. For
+sum by (pod) (rate(m{job="api"}[5m])) the skeleton is:
+
+	SELECT gkey, sumForEach(grid) AS grid FROM (
+	    SELECT series.gkey AS gkey,
+	           timeSeriesRateToGrid(<start>, <end>, <step>, <range>)(fromUnixTimestamp64Milli(unix_milli), value) AS grid
+	    FROM signoz_metrics.distributed_samples_v4 AS points
+	    INNER JOIN (
+	        SELECT fingerprint, <group key expr> AS gkey
+	        FROM signoz_metrics.time_series_v4
+	        WHERE <series predicates>
+	        GROUP BY fingerprint, gkey
+	    ) AS series ON points.fingerprint = series.fingerprint
+	    WHERE metric_name = ? AND temporality IN ['Cumulative', 'Unspecified']
+	      AND points.fingerprint IN (<matched fingerprints>)
+	      AND unix_milli > <start - range> AND unix_milli <= <end>
+	      AND bitAnd(flags, 1) = 0
+	    GROUP BY points.fingerprint, series.gkey
+	) GROUP BY gkey
+	SETTINGS allow_experimental_ts_to_grid_aggregate_function = 1
+
+Reading it inside out:
+
+The time window is the selector's semantics verbatim: strict > on the
+lower bound and <= on the upper is the left-open (t - w, t] rule, with the
+whole window shifted by the offset. bitAnd(flags, 1) = 0 drops stale
+markers, which PromQL excludes from range vectors.
+
+The inner GROUP BY computes one grid array per series.
+timeSeriesRateToGrid(start, end, step, range) is a parametric aggregate:
+fed (timestamp, value) pairs it produces Array(Nullable(Float64)) with one
+slot per grid point. Correct because it implements the engine's
+extrapolatedRate decision for decision — counter resets, the zero-point
+clamp, the extrapolation thresholds, the >= 2 samples rule, the left-open
+window — verified by feeding identical samples to both and comparing
+slot for slot: the only difference ever observed is the last bit
+(ClickHouse's C++ and Go round the same formula differently), which is
+the floating-point floor, not a semantic gap. irate/delta/idelta map to
+their own timeSeries*ToGrid functions with the same verification;
+increase has no function of its own and is emitted as
+arrayMap(x -> x * <range seconds>, <rate expr>), exact by definition —
+extrapolatedRate computes the same extrapolated delta for both and
+divides by the range only when isRate, so multiplying it back is the
+identity, not an approximation. The grid parameters are rendered as
+literals, not bound args — they are aggregate-function parameters — and
+the experimental gate rides as a SETTINGS clause on the statement itself
+so telemetrystore hooks cannot clobber it.
+
+The join annotates each series with its group key, in one of two forms.
+by (...) extracts each listed label as a plain column
+(JSONExtractString(labels, 'pod') AS g0) and groups on the columns
+directly: the projection is a known short list and the label names live
+in Go, so building, sorting and stringifying every label pair per row
+would be waste. Correct because column-tuple equality is label-set
+equality on the projection, and an extracted '' is the label being
+absent — Prometheus semantics for by() over missing labels, and empties
+are skipped when the columns turn back into labels. without and
+no-aggregation project a label SET that varies per series, so they get
+the canonical key: toJSONString of the sorted [label, value] pairs the
+unit projects (without excludes the listed labels plus __name__; no
+aggregation keeps everything, the name coming off in Go per the engine's
+name-dropping rules). There the sort is load-bearing — stored JSON key
+order is not canonical across fingerprints, and two orderings of the same
+labels must land in one group — empty values are filtered for the same
+absent-label reason, and the same string parses back into the output
+label set (labelsFromGroupKey).
+
+The outer GROUP BY is the spatial aggregation: sum/min/max/avg/count
+by/without become the -ForEach combinators. Element-wise aggregation over
+grid arrays is the engine's per-t_i aggregation, because slot i of every
+input array refers to the same t_i; the combinators skip NULLs, which is
+the engine aggregating only the series present at t_i, and an index where
+every series is absent stays NULL. Two edges need explicit handling:
+countForEach wraps in a mapping of 0 back to NULL, because a count over
+an all-absent index is an absent point, not 0; and a unit without
+aggregation still passes through maxForEach — the identity for the common
+one-fingerprint group, and a deterministic NULL-skipping merge when a
+regex __name__ selector collapses distinct metrics onto one projected
+label set. One caveat is inherent: summation order over series differs
+from the engine's, so spatial aggregates can differ in the last ULP —
+float addition is not associative; no ordering reproduces the engine's
+bit-exactly from inside a GROUP BY.
+
+## Instant selectors: staleness needs two aggregates
+
+unitInstant uses window = lookback and must reproduce the shadowing rule:
+the point is absent when the latest in-window sample is a stale marker.
+timeSeriesLastToGrid alone cannot express that — skipping stale rows in
+WHERE would resurrect the older real sample the marker was written to
+bury. So stale rows stay in the scan for this kind only, and the grid
+expression compares three aggregates per slot:
+
+	arrayMap((tall, tok, vok) -> if(tall IS NULL OR tok IS NULL OR tall != tok, NULL, vok),
+	         timeSeriesLastToGrid(...)(ts, toFloat64(unix_milli)),                           -- last sample overall
+	         timeSeriesLastToGridIf(...)(ts, toFloat64(unix_milli), bitAnd(flags, 1) = 0),   -- last non-stale, its timestamp
+	         timeSeriesLastToGridIf(...)(ts, value, bitAnd(flags, 1) = 0))                   -- last non-stale, its value
+
+Correct by cases on a slot's window. No samples at all: both timestamp
+aggregates are NULL, the slot is NULL — absent, as the engine says. Latest
+sample non-stale: it is the latest overall and the latest non-stale, the
+timestamps agree, the slot takes its value — the engine's pick. Latest
+sample stale: the last-overall timestamp is the marker's, the
+last-non-stale timestamp is older (or NULL when only markers are in
+window), they disagree, the slot is NULL — the marker shadows, exactly
+the engine's rule. Timestamps are unique per series (ingest dedups), so
+timestamp equality identifies "the same sample" without ambiguity. The
+-If combinator's applicability to these experimental aggregates was
+probed before being trusted, not assumed.
+
+## Windowed *_over_time: whole buckets instead of a grid function
+
+avg/min/max/sum/count _over_time aggregate every raw sample in the window,
+and no timeSeries*ToGrid function computes them. (last_over_time is the
+exception: the last sample of a range vector — stale markers excluded from
+range vectors by PromQL, excluded here in WHERE — is exactly
+timeSeriesLastToGrid.) These transpile only when the range is a whole
+multiple of the step, and then the window needs no per-sample fan-out at
+all: with W = range/step, the window (t_k - range, t_k] is exactly the
+union of W step buckets — both are left-open on the same boundaries — so
+bucket membership fully determines window membership. Each sample lands
+in exactly one bucket by a plain GROUP BY:
+
+	intDiv(unix_milli - <start> + <range> - 1, <step>) AS jj
+
+(ceil((ts - start)/step) shifted by W-1 so the earliest in-window sample
+sits at 0; slot k's window is buckets jj in [k, k+W-1]). The alternative —
+fanning each sample into all W windows that cover it — multiplies rows by
+W, which for a long range over a short step is a row explosion measured
+in billions; the bucketed form's row count is series x buckets, the size
+of the output, regardless of W.
+
+The shard level aggregates per (series, group key, bucket): a bucket
+count plus the function's value aggregate (sum for sum/avg, min, max).
+The assembly level places the partials into dense arrays
+(groupArrayInsertAt — positions are unique, one row per bucket; counts
+and sums default to 0, which contributes nothing) and slides: slot k
+combines its at-most-W bucket partials by direct aggregation, so window
+sums are added the way the engine adds them — no prefix-sum differencing,
+whose large-minus-large cancellation would drift past the shadow
+tolerance on counter-sized values. Correct per slot because the bucket
+union is the exact window multiset and avg/min/max/sum/count are
+order-insensitive on a multiset (sum/avg up to summation order, the float
+caveat above). A slot with zero window count is absent; min/max filter
+their slices on the bucket counts, so an empty bucket's default can never
+be mistaken for a value — a real sample can legitimately be +Inf.
+
+Ranges that don't divide the step, and windows wider than
+maxWindowBuckets buckets (the slide costs W combines per slot), fall back
+to the engine path, which is exact.
+
+## Scalar ops, full plans, hybrid plans
+
+The scalar-op pipeline applies in Go to the returned arrays
+(applyScalarOps), slot by slot: arithmetic operators compute, comparisons
+filter (the slot keeps the vector-side value or becomes NULL) or return
+0/1 under bool. Correct trivially: it is the same float64 operation the
+engine would apply to the same slot value, in the same operator order the
+AST dictates — running it in Go instead of another SQL layer changes
+where, not what.
+
+A full plan's arrays map straight to the result matrix. A hybrid plan
+materializes each unit's arrays as synthetic series under its
+__signoz_transpiled_N__ name and evaluates the rewritten expression over
+a storage that serves synthetic names from memory and everything else
+live. Substitution is sound because a unit's output is a plain instant
+vector to the engine — same values at same timestamps under a different
+name, and the name cannot matter: plans that group by or match on
+__name__ were refused at classification, and name-keeping units are never
+substituted. One subtlety makes it exact: stale markers are written at
+absent grid points, because the engine's lookback would otherwise
+resurrect a point from up to lookback earlier — the marker encodes
+"absent here" the way the engine itself encodes it. Units evaluate
+concurrently; each is one series lookup plus one grid statement. A step
+of 0 is an instant query: a single evaluation at end.
+
 ## Series lookup
 
-Matchers resolve to series once per selector (`selectSeries`) against the series
-tables, which hold one row per (fingerprint, bucket) at 1h/6h/1d/1w
-granularities. Table selection and window rounding delegate to the shared
-metrics schema package (`pkg/telemetryschema/metricstelemetryschema`); the
-window start rounds down to the bucket boundary so a window beginning mid-bucket
-still matches the bucket's row.
+Both paths resolve matchers the same way, once per selector
+(selectSeries), against the series tables holding 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.
 
-How matchers become SQL is documented at `applySeriesConditions`. The rules that
-carry semantics:
+## The engine path
 
-- `__name__` matchers (all four types) translate to the `metric_name` column.
-- Every other matcher becomes 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 (`^(?:...)$`) before they reach `match()`: PromQL
-  matchers match the whole value, ClickHouse `match()` searches for a
-  substring.
-- The series-lookup upper bound is inclusive (`unix_milli <= end`) because the
-  exporter floors registration rows to the bucket start: a series first
-  registered in the bucket beginning exactly at `end` would otherwise be
-  invisible while its samples are in range.
-
-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 — `foo / foo offset 1d` reads two narrow windows
-instead of the widest one twice.
-
-**Last-sample-per-step reduction.** Instant selectors of subquery-free queries
-fetch only the last sample per step bucket. The engine resolves an instant
-selector at each grid timestamp `t` to the latest sample in the left-open
-lookback window `(t − lookback, t]`. Buckets anchor at the selector's first
-evaluation timestamp — recovered from the hints as
-`hints.Start + lookback − 1ms`, the inverse of how the engine derives
-`hints.Start` — so bucket boundaries coincide with evaluation timestamps, and a
-non-final sample of a bucket can never be the latest sample in
-`(t − lookback, t]` for any grid `t`. Real timestamps are preserved, so the
-engine's own lookback and staleness handling stay exact.
-
-Range selectors always fetch raw — every sample feeds the range function. The
-subquery-free proof travels in the context as `prometheus.QueryTraits`, because
-subquery selectors evaluate at the subquery's step while the hints carry the
-top-level step; call sites that do not attach traits get the conservative raw
-fetch.
-
-**Row assembly** maps stale flags to the engine's `StaleNaN` and merges series
-with identical label sets (`sortAndMerge`) — the engine assumes storages never
-emit duplicates. Duplicate timestamps pass through as stored: uniqueness is
-ingest's job, and v1 feeds them to the engine as-is over the same data.
-
-**The fingerprint filter is a shard-local semi-join.** The samples query
-restricts to the matched series by re-running the series predicates as an
-`IN (SELECT fingerprint FROM <local series table> ...)` subquery, not a GLOBAL
-broadcast of the matched set. ClickHouse materializes the subquery's set per
-shard before the scan, so it still engages the fingerprint primary-key column.
-Because the subquery re-executes the predicates after the lookup ran, it can
-match series registered in between; sample rows whose fingerprint the lookup
-never saw are skipped — the lookup is the read snapshot.
-
----
+Queries that do not transpile run in the stock engine over this package's
+storage.Querier, which is still not the v1 path. Samples are fetched per
+selector using the engine's per-selector hints, not the query-wide union
+window, so foo / foo offset 1d reads two narrow windows instead of the
+widest one twice. Instant selectors of subquery-free queries fetch only
+the last sample per step bucket (lastSamplePerStep): buckets anchor at the
+selector's first evaluation timestamp — recovered from the hints as
+hints.Start + lookback - 1ms, the inverse of how the engine derives
+hints.Start — so bucket boundaries coincide with evaluation timestamps and
+a non-final sample of a bucket can never be the latest sample in
+(t - lookback, t] for any grid t. Real timestamps are preserved, so the
+engine's own lookback and staleness handling stay exact. Range selectors
+always fetch raw — every sample feeds the range function — and the
+subquery-free proof travels in the context as prometheus.QueryTraits,
+because subquery selectors evaluate at the subquery's step while the
+hints carry the top-level step. Row assembly maps stale flags to the
+engine's StaleNaN and merges series with identical label sets
+(sortAndMerge) — 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 semi-join above exploits
-that: each shard filters by its own series rows, which are exactly the series
-of that shard's samples.
-
-The temporality filter on every samples statement
-(`temporality IN ['Cumulative', 'Unspecified']`) is a semantic no-op — the
-matched fingerprints already come from those temporalities — that engages the
-leading samples primary-key column.
-
-Delta-temporality series stay invisible to PromQL 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.
-
----
+samples_v4 and time_series_v4 (and all their rollups) shard on the same
+key — cityHash64(env, temporality, metric_name, fingerprint) — so a
+series' samples and catalog rows live on the same shard. The transpiled
+statement above exploits that: the distributed samples table at the
+top-level FROM makes ClickHouse rewrite the whole inner query per shard,
+where the join against the shard-local series table and the per-series
+grid aggregation run next to the data; the initiator only merges
+aggregate states and applies the spatial -ForEach step. Same layout as
+the telemetrymetrics statement builder. The group-key join alone
+restricts the transpiled scan to the matched series; the engine path's
+samples fetch restricts by the same predicates as a shard-local
+semi-join, not a GLOBAL broadcast of the matched set. The temporality
+filter on every
+samples statement is a semantic no-op — the matched fingerprints already
+come from those temporalities — that engages the leading samples
+primary-key column. Delta-temporality series stay invisible to PromQL
+here exactly as they are in v1: the rollout gate is parity with v1, and
+making Delta visible is its own change with its own semantics to design —
+a Delta stream fed to rate() as-if-cumulative would be wrong, not just
+new.
 
 ## Observability
 
-Every statement carries a `log_comment` with
-`code.namespace=clickhouse-prometheus-v2` and `code.function.name` naming the
-call site, so this provider's work is attributable in `system.query_log`.
+Every statement carries a log_comment with
+code.namespace=clickhouse-prometheus-v2 and code.function.name naming the
+call site (selectSeries, selectSamples, transpiledUnit, LabelValues,
+LabelNames), so this provider's work is attributable in system.query_log
+without guessing from query text.

docs/contributing/tests/e2e.md

@@ -30,11 +30,11 @@ yarn install:browsers   # one-time Playwright browser install
 
 ### Starting the Test Environment
 
-To spin up the backend stack (SigNoz, ClickHouse, Postgres, ClickHouse Keeper, Zeus mock, gateway mock, seeder, migrator-with-web) and keep it running:
+To spin up the backend stack (SigNoz, ClickHouse, Postgres, Zookeeper, Zeus mock, gateway mock, seeder, migrator-with-web) and keep it running:
 
 ```bash
 cd tests
-uv run pytest --basetemp=./tmp/ -vv --reuse --rebuild --with-web \
+uv run pytest --basetemp=./tmp/ -vv --reuse --with-web \
   e2e/bootstrap/setup.py::test_setup
 ```
 
@@ -45,13 +45,8 @@ This command will:
 - Start the HTTP seeder container (`tests/seeder/` — exposing `/telemetry/{traces,logs,metrics}` POST + DELETE)
 - Write backend coordinates to `tests/e2e/.env.local` (loaded by `playwright.config.ts` via dotenv)
 - Keep containers running via the `--reuse` flag
-- Rebuild the SigNoz container from the current sources via the `--rebuild` flag
 
-The `--with-web` flag builds the frontend into the SigNoz container — required for E2E. The build takes ~4 mins on a cold start; later builds are incremental.
-
-### Rebuilding After Source Changes
-
-The `--with-web` image bakes the built frontend in, so neither backend nor frontend changes are picked up while `--reuse` keeps the container running. `--rebuild` fixes that for both: it kills the SigNoz container, rebuilds the image incrementally (go build cache + pnpm store — a frontend-only change rebuilds in about a minute), and starts a fresh one while databases, mocks, migrations, and the seeder stay reused. The setup command above passes it, so the iteration loop is: change code → re-run the setup command → re-run your specs. `--rebuild` requires `--reuse` and cannot be combined with `--teardown` or `--clean`.
+The `--with-web` flag builds the frontend into the SigNoz container — required for E2E. The build takes ~4 mins on a cold start.
 
 ### Stopping the Test Environment
 
@@ -286,16 +281,13 @@ The full `playwright.config.ts` is the source of truth. Common things to tweak:
 The same pytest flags integration tests expose work here, since E2E reuses the shared fixture graph:
 
 - `--reuse` — keep containers warm between runs (required for all iteration).
-- `--rebuild` — recreate the SigNoz container from the current sources (backend and, with `--with-web`, frontend) while the rest of the stack stays up. Requires `--reuse`.
 - `--teardown` — tear everything down.
-- `--clean` — prune the docker build caches, forcing the next image build to start cold.
 - `--with-web` — build the frontend into the SigNoz container. **Required for E2E**; integration tests don't need it.
-- `--sqlstore-provider`, `--postgres-version`, `--clickhouse-version`, etc. — see `docs/contributing/tests/integration.md`.
+- `--sqlstore-provider`, `--postgres-version`, `--clickhouse-version`, etc. — see `docs/contributing/integration.md`.
 
 ## What should I remember?
 
-- **Always use the `--reuse` flag** when setting up the E2E stack. `--with-web` adds a ~4 min frontend build on a cold start; later builds are incremental.
-- **Changed backend or frontend code? Re-run the setup command** — it passes `--rebuild`, swapping the SigNoz container for one built from your current sources while the rest of the stack stays up.
+- **Always use the `--reuse` flag** when setting up the E2E stack. `--with-web` adds a ~4 min frontend build; you only want to pay that once.
 - **Don't teardown before setup.** `--reuse` correctly handles partially-set-up state, so chaining teardown → setup wastes time.
 - **Prefer UI-driven flows.** Playwright captures BE requests in the trace; a parallel `fetch` probe is almost always redundant. Drop to `page.request.*` only when the UI can't reach what you need.
 - **Use `page.waitForResponse` on UI clicks** to assert BE contracts — it still exercises the UI trigger path.

docs/contributing/tests/integration.md

@@ -37,34 +37,13 @@ make py-test-setup
 Under the hood this runs, from `tests/`:
 
 ```bash
-uv run pytest --basetemp=./tmp/ -vv --reuse --rebuild --capture=no integration/bootstrap/setup.py::test_setup
+uv run pytest --basetemp=./tmp/ -vv --reuse integration/bootstrap/setup.py::test_setup
 ```
 
 This command will:
-- Start all required services (ClickHouse, PostgreSQL, ClickHouse Keeper, SigNoz, Zeus mock, gateway mock)
+- Start all required services (ClickHouse, PostgreSQL, Zookeeper, SigNoz, Zeus mock, gateway mock)
 - Register an admin user
 - Keep containers running via the `--reuse` flag
-- Rebuild the SigNoz container from the current sources via the `--rebuild` flag
-
-### Rebuilding After Source Changes
-
-`--reuse` keeps the running SigNoz container, which means backend source changes are not picked up. `--rebuild` fixes exactly that: it kills the existing SigNoz container, rebuilds the image (incremental — only changed packages recompile thanks to the build cache), and starts a fresh one, while everything else (databases, mocks, migrations) stays reused. `make py-test-setup` passes it by default, so the iteration loop is simply:
-
-```bash
-make py-test-setup      # (re)build signoz from your current sources
-uv run pytest --basetemp=./tmp/ -vv --reuse integration/tests/<suite>/
-# ... edit backend code or tests ...
-make py-test-setup      # pick up the backend changes
-uv run pytest --basetemp=./tmp/ -vv --reuse integration/tests/<suite>/
-```
-
-The same applies to the e2e stack. `--rebuild` requires `--reuse` and cannot be combined with `--teardown` or `--clean`.
-
-Some suites define their own SigNoz variant in a suite-local `conftest.py` (`create_signoz(..., cache_key=...)` — e.g. `basepath`, `metricreduction`, `querier_json_body`). Those containers are not touched by `make py-test-setup`, which only rebuilds the default instance. For such suites, pass `--rebuild` on the suite run itself — it rebuilds every SigNoz variant the run instantiates:
-
-```bash
-uv run pytest --basetemp=./tmp/ -vv --reuse --rebuild integration/tests/<suite>/
-```
 
 ### Stopping the Test Environment
 
@@ -77,21 +56,11 @@ make py-test-teardown
 Which runs:
 
 ```bash
-uv run pytest --basetemp=./tmp/ -vv --teardown --capture=no integration/bootstrap/setup.py::test_teardown
+uv run pytest --basetemp=./tmp/ -vv --teardown integration/bootstrap/setup.py::test_teardown
 ```
 
 This destroys the running integration test setup and cleans up resources.
 
-### Cleaning the Image Build Cache
-
-The `signoz:integration` image build keeps its Go build and module caches in BuildKit cache mounts, so rebuilds only recompile what changed. These caches survive `--teardown` (they belong to the Docker builder, not to any container). If a cache ever needs to be nuked — suspected corruption, disk pressure, or to force a genuinely cold build — pass the `--clean` flag:
-
-```bash
-uv run pytest --basetemp=./tmp/ -vv --teardown --clean integration/bootstrap/setup.py::test_teardown
-```
-
-`--clean` prunes the docker build artifacts backing the incremental image build at session start, so the next build starts from a clean slate. Images and regular layer cache stay intact, but note the pruning is host-wide — it clears build caches for other projects too, not just SigNoz's. The flag composes with any invocation — passing it on a normal `--reuse` run simply makes the next image build start cold (~3–4 minutes instead of seconds).
-
 ## Understanding the Integration Test Framework
 
 Python and pytest form the foundation of the integration testing framework. Testcontainers are used to spin up disposable integration environments. WireMock is used to spin up **test doubles** of external services (Zeus cloud API, gateway, etc.).
@@ -130,7 +99,7 @@ tests/
 │       ├── passwordauthn/
 │       ├── querier/
 │       └── ...
-└── e2e/                     # Playwright suite (see docs/contributing/tests/e2e.md)
+└── e2e/                     # Playwright suite (see docs/contributing/e2e.md)
 ```
 
 Each test suite follows these principles:
@@ -255,9 +224,9 @@ Tests can be configured using pytest options:
 - `--sqlstore-provider` — Choose the SQL store provider (default: `postgres`)
 - `--sqlite-mode` — SQLite journal mode: `delete` or `wal` (default: `delete`). Only relevant when `--sqlstore-provider=sqlite`.
 - `--postgres-version` — PostgreSQL version (default: `15`)
-- `--clickhouse-version` — ClickHouse version, also used for ClickHouse Keeper (default: `25.12.5`)
-- `--schema-migrator-version` — SigNoz schema migrator version (default: `v0.144.6`)
-- `--with-web` — Build the frontend into the SigNoz image (required for e2e)
+- `--clickhouse-version` — ClickHouse version (default: `25.5.6`)
+- `--zookeeper-version` — Zookeeper version (default: `3.7.1`)
+- `--schema-migrator-version` — SigNoz schema migrator version (default: `v0.144.2`)
 
 Example:
 
@@ -270,7 +239,6 @@ uv run pytest --basetemp=./tmp/ -vv --reuse \
 ## What should I remember?
 
 - **Always use the `--reuse` flag** when setting up the environment or running tests to keep containers warm. Without it every run rebuilds the stack (~4 mins).
-- **Changed backend code? Re-run `make py-test-setup`** — it passes `--rebuild`, swapping the SigNoz container for one built from your current sources while the rest of the stack stays up.
 - **Use the `--teardown` flag** only when cleaning up — mixing `--teardown` with `--reuse` is a contradiction.
 - **Do not pre-emptively teardown before setup.** If the stack is partially up, `--reuse` picks up from wherever it is. `make py-test-teardown` then `make py-test-setup` wastes minutes.
 - **Follow the naming convention** with two-digit numeric prefixes (`01_`, `02_`) for ordered test execution within a suite.
@@ -279,5 +247,5 @@ uv run pytest --basetemp=./tmp/ -vv --reuse \
 - **Use descriptive test names** that clearly indicate what is being tested.
 - **Leverage fixtures** for common setup. The shared fixture package is at `tests/fixtures/` — reuse before adding new ones.
 - **Test both success and failure scenarios** (4xx / 5xx paths) to ensure robust functionality.
-- **Run `make py-fmt` and `make py-lint` before committing** Python changes — ruff format + ruff check.
+- **Run `make py-fmt` and `make py-lint` before committing** Python changes — black + isort + autoflake + pylint.
 - **`--sqlite-mode=wal` does not work on macOS.** The integration test environment runs SigNoz inside a Linux container with the SQLite database file mounted from the macOS host. WAL mode requires shared memory between connections, and connections crossing the VM boundary (macOS host ↔ Linux container) cannot share the WAL index, resulting in `SQLITE_IOERR_SHORT_READ`. WAL mode is tested in CI on Linux only.

go.mod

@@ -4,7 +4,7 @@ go 1.25.7
 
 require (
 	dario.cat/mergo v1.0.2
-	github.com/AfterShip/clickhouse-sql-parser v0.5.4
+	github.com/AfterShip/clickhouse-sql-parser v0.5.3
 	github.com/ClickHouse/clickhouse-go/v2 v2.44.0
 	github.com/DATA-DOG/go-sqlmock v1.5.2
 	github.com/SigNoz/clickhouse-go-mock v0.14.0

go.sum

@@ -66,8 +66,8 @@ dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
 filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
-github.com/AfterShip/clickhouse-sql-parser v0.5.4 h1:yiCQaMq8EO+dpKdnpP9YYd/ne6MSuOXgsMsNL33NiTI=
-github.com/AfterShip/clickhouse-sql-parser v0.5.4/go.mod h1:Qi3qvPTfZb/aFwI5V4WFOahgjsLJa4MzVijIAfwOhDw=
+github.com/AfterShip/clickhouse-sql-parser v0.5.3 h1:6iap8XGjuSjD3w7r1UNrg66ljBugcv2P39s4eo/ZLRw=
+github.com/AfterShip/clickhouse-sql-parser v0.5.3/go.mod h1:Qi3qvPTfZb/aFwI5V4WFOahgjsLJa4MzVijIAfwOhDw=
 github.com/Azure/azure-sdk-for-go v68.0.0+incompatible h1:fcYLmCpyNYRnvJbPerq7U0hS+6+I79yEDJBqVNcqUzU=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0 h1:fou+2+WFTib47nS+nz/ozhEBnvU96bKHy6LjRsY4E28=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0/go.mod h1:t76Ruy8AHvUAC8GfMWJMa0ElSbuIcO03NLpynfbgsPA=

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/classification_golden_test.go

@@ -0,0 +1,119 @@
+package clickhouseprometheusv2
+
+import (
+	"encoding/json"
+	"flag"
+	"os"
+	"path/filepath"
+	"sort"
+	"testing"
+
+	"github.com/prometheus/prometheus/promql/parser"
+	"github.com/stretchr/testify/require"
+)
+
+var updateGolden = flag.Bool("update", false, "rewrite the classification golden file")
+
+const goldenFile = "testdata/classification_golden.json"
+
+// corpusFile is the conformance corpus the integration suite replays; the
+// golden freezes how the classifier routes every one of its expressions.
+const corpusFile = "../../../tests/integration/testdata/promqltestcorpus/corpus.json"
+
+type goldenEntry struct {
+	Expr    string `json:"expr"`
+	StartMs int64  `json:"start_ms"`
+	EndMs   int64  `json:"end_ms"`
+	StepMs  int64  `json:"step_ms"`
+	// Plan is the routing decision: "full" (whole query in ClickHouse),
+	// "hybrid" (units substituted, engine on top), "fallback" (engine over
+	// the native querier).
+	Plan string `json:"plan"`
+	// Units is the substituted-unit count for hybrid plans.
+	Units int `json:"units,omitempty"`
+	// Reason is the coarse fallback bucket (fallbackShape).
+	Reason string `json:"reason,omitempty"`
+}
+
+// TestClassificationGolden freezes the classifier's routing decision for
+// every (expression, grid) of the conformance corpus. Routing is a
+// correctness surface of its own: a change that silently sends rate() to the
+// engine path costs the pushdown, and one that silently starts transpiling a
+// shape never proven equivalent risks wrong numbers — both must show up in
+// review as a diff of this file, with the corpus suite's clickhousev2 leg
+// judging whether the new routing still returns the reference answers.
+//
+// Regenerate after intentional classifier changes:
+//
+//	go test ./pkg/prometheus/clickhouseprometheusv2 -run TestClassificationGolden -update
+func TestClassificationGolden(t *testing.T) {
+	raw, err := os.ReadFile(corpusFile)
+	require.NoError(t, err)
+
+	var corpus struct {
+		Cases []struct {
+			Expr    string `json:"expr"`
+			StartMs int64  `json:"start_ms"`
+			EndMs   int64  `json:"end_ms"`
+			StepMs  int64  `json:"step_ms"`
+		} `json:"cases"`
+	}
+	require.NoError(t, json.Unmarshal(raw, &corpus))
+	require.NotEmpty(t, corpus.Cases)
+
+	promParser := parser.NewParser(parser.Options{})
+	seen := map[goldenEntry]bool{}
+	var entries []goldenEntry
+	for _, c := range corpus.Cases {
+		key := goldenEntry{Expr: c.Expr, StartMs: c.StartMs, EndMs: c.EndMs, StepMs: c.StepMs}
+		if seen[key] {
+			continue
+		}
+		seen[key] = true
+
+		expr, err := promParser.ParseExpr(c.Expr)
+		require.NoError(t, err, "corpus expression must parse: %q", c.Expr)
+
+		entry := key
+		plan, ok := classify(expr, gridContext{startMs: c.StartMs, endMs: c.EndMs, stepMs: c.StepMs})
+		switch {
+		case ok && plan.full:
+			entry.Plan = "full"
+		case ok:
+			entry.Plan = "hybrid"
+			entry.Units = len(plan.units)
+		default:
+			entry.Plan = "fallback"
+			entry.Reason = fallbackShape(expr)
+		}
+		entries = append(entries, entry)
+	}
+	sort.Slice(entries, func(i, j int) bool {
+		a, b := entries[i], entries[j]
+		if a.Expr != b.Expr {
+			return a.Expr < b.Expr
+		}
+		if a.StartMs != b.StartMs {
+			return a.StartMs < b.StartMs
+		}
+		if a.EndMs != b.EndMs {
+			return a.EndMs < b.EndMs
+		}
+		return a.StepMs < b.StepMs
+	})
+
+	got, err := json.MarshalIndent(entries, "", "  ")
+	require.NoError(t, err)
+	got = append(got, '\n')
+
+	if *updateGolden {
+		require.NoError(t, os.MkdirAll(filepath.Dir(goldenFile), 0o755))
+		require.NoError(t, os.WriteFile(goldenFile, got, 0o644))
+		return
+	}
+
+	want, err := os.ReadFile(goldenFile)
+	require.NoError(t, err, "golden missing — generate it with -update")
+	require.Equal(t, string(want), string(got),
+		"classification routing changed; if intentional, regenerate with -update and justify the diff in review")
+}

pkg/prometheus/clickhouseprometheusv2/provider.go

@@ -2,27 +2,33 @@ package clickhouseprometheusv2
 
 import (
 	"context"
+	"time"
 
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/prometheus"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/storage"
 )
 
-// provider ties the package together: its own engine and parser, and the
-// ClickHouse client behind the native storage.Querier. It stays unexported:
-// callers hold the prometheus.Prometheus interface, which is the boundary
-// between the two provider implementations.
+// provider ties the package together: its own engine and parser, the
+// ClickHouse client behind the native storage.Querier, and the transpiler
+// executor. It stays unexported: callers hold the prometheus.Prometheus
+// interface, which is the boundary between the two provider implementations,
+// and reach the transpiler only through the prometheus.RangeExecutor
+// capability.
 type provider struct {
 	settings factory.ScopedProviderSettings
 	engine   *prometheus.Engine
 	parser   prometheus.Parser
 	client   *client
+	executor *executor
 }
 
 var (
 	_ prometheus.Prometheus        = (*provider)(nil)
 	_ prometheus.StatementCapturer = (*provider)(nil)
+	_ prometheus.RangeExecutor     = (*provider)(nil)
 )
 
 func NewFactory(telemetryStore telemetrystore.TelemetryStore) factory.ProviderFactory[prometheus.Prometheus, prometheus.Config] {
@@ -43,9 +49,17 @@ func New(_ context.Context, providerSettings factory.ProviderSettings, config pr
 		engine:   engine,
 		parser:   parser,
 		client:   client,
+		executor: &executor{client: client, engine: engine, parser: parser},
 	}, nil
 }
 
+// TryExecuteRange evaluates transpilable query shapes directly in ClickHouse
+// (see transpiler.go). ok=false means the shape is not transpilable and the
+// caller should evaluate through Engine over Storage instead.
+func (p *provider) TryExecuteRange(ctx context.Context, query string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error) {
+	return p.executor.TryExecuteRange(ctx, query, start, end, step)
+}
+
 func (p *provider) Engine() *prometheus.Engine {
 	return p.engine
 }

pkg/prometheus/clickhouseprometheusv2/transpiler.go

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

pkg/prometheus/clickhouseprometheusv2/transpiler_corpus_test.go

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

pkg/prometheus/clickhouseprometheusv2/transpiler_exec.go

@@ -0,0 +1,550 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"sort"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
+	promValue "github.com/prometheus/prometheus/model/value"
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/promql/parser"
+	"github.com/prometheus/prometheus/storage"
+	"golang.org/x/sync/errgroup"
+)
+
+// executor evaluates transpilable PromQL directly in ClickHouse, falling
+// back (ok=false) whenever the query shape or the step doesn't qualify. The
+// timeSeries*ToGrid functions it builds on are assumed available: the
+// supported ClickHouse floor is >= 25.6.
+type executor struct {
+	client *client
+	engine *prometheus.Engine
+	parser prometheus.Parser
+}
+
+// maxWindowBuckets caps range/step for the windowed *_over_time form: every
+// grid slot combines that many bucket partials, and the fleet's windows sit
+// well under it ([1m]..[17m] at 30-60s steps) — anything larger is a
+// long-range query whose step a dashboard scales up anyway, and the engine
+// path serves the rest.
+const maxWindowBuckets = 64
+
+// TryExecuteRange transpiles and runs the query in ClickHouse when its shape
+// is in the allowlist. ok=false means "not transpilable" and carries no
+// error; the caller runs the engine path.
+func (e *executor) TryExecuteRange(ctx context.Context, qs string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error) {
+	expr, err := e.parser.ParseExpr(qs)
+	if err != nil {
+		// Let the engine path produce the (enhanced) parse error.
+		return nil, false, nil
+	}
+
+	plan, ok := classify(expr, queryGrid(start, end, step))
+	if !ok {
+		return nil, false, nil
+	}
+
+	// timeSeriesLastToGrid widens its window to max(window, step) — probed: a
+	// sample aged (window, step] still fills the slot — while the rate/delta
+	// family enforces the window strictly. The Last-style kinds used to fall
+	// back when window < step because of that widening; the window-sliver
+	// filter (see samplesConditions) makes the widening harmless there:
+	// samples exist only inside (t_k - window, t_k] slivers, so the widened
+	// window intersected with the data IS the lookback window — and if a
+	// future ClickHouse stops widening, the unwidened window is the sliver
+	// too. Correct either way. A non-positive window still falls back: the
+	// sliver argument needs a real window to filter to.
+	//
+	// The windowed *_over_time form gates only the range >= step regime: it
+	// decomposes the window into whole step buckets (see windowedInner),
+	// which is exact only when the range is a multiple of the step, and its
+	// per-slot slide costs range/step bucket combines — bounded by
+	// maxWindowBuckets so a long-range short-step query cannot turn the
+	// slide into the bottleneck. range < step needs neither gate: the
+	// windows are disjoint slivers, aggregated one slot each with no slide.
+	// Every miss falls back to the engine path, which is exact.
+	for _, unit := range plan.units {
+		stepMs := unit.grid.stepMs
+		if stepMs == 0 {
+			stepMs = 1000
+		}
+		switch {
+		case unit.core.kind == unitInstant || (unit.core.kind == unitOverTime && unit.core.overFn == "last"):
+			windowMs := unit.core.rangeMs
+			if unit.core.kind == unitInstant {
+				windowMs = e.client.lookbackMs
+			}
+			if windowMs <= 0 {
+				return nil, false, nil
+			}
+		case unit.core.kind == unitOverTime:
+			if unit.core.rangeMs < unit.grid.stepMs {
+				// Disjoint slivers: no divisibility or width requirement.
+				continue
+			}
+			if unit.core.rangeMs%stepMs != 0 || unit.core.rangeMs/stepMs > maxWindowBuckets {
+				return nil, false, nil
+			}
+		}
+	}
+
+	// Evaluate every unit concurrently on its own grid (the query grid, or a
+	// subquery grid); each is one series lookup plus one grid query.
+	results := make([][]transpiledSeries, len(plan.units))
+	eg, egCtx := errgroup.WithContext(ctx)
+	for i, unit := range plan.units {
+		eg.Go(func() error {
+			res, err := e.executeUnit(egCtx, &unit.core, unit.grid)
+			if err != nil {
+				return err
+			}
+			results[i] = res
+			return nil
+		})
+	}
+	if err := eg.Wait(); err != nil {
+		return nil, true, err
+	}
+
+	if plan.full {
+		g := plan.units[0].grid
+		return toMatrix(results[0], g.startMs, g.stepMs), true, nil
+	}
+
+	matrix, err := e.executeHybrid(ctx, plan, results)
+	if err != nil {
+		return nil, true, err
+	}
+	return matrix, true, nil
+}
+
+// queryGrid derives the top-level evaluation grid; step 0 is an instant
+// query: a single evaluation at end, whatever start was.
+func queryGrid(start, end time.Time, step time.Duration) gridContext {
+	startMs, endMs, stepMs := start.UnixMilli(), end.UnixMilli(), step.Milliseconds()
+	if stepMs == 0 {
+		startMs = endMs
+	}
+	return gridContext{startMs: startMs, endMs: endMs, stepMs: stepMs}
+}
+
+// transpiledSeries is one output series of a unit: projected labels and one
+// value pointer per grid point (nil = absent).
+type transpiledSeries struct {
+	lset   labels.Labels
+	values []*float64
+}
+
+// executeUnit runs one core unit on its grid: series lookup (budgets,
+// fingerprints, metric names), then the single grid query, then the
+// scalar-op pipeline.
+func (e *executor) executeUnit(ctx context.Context, unit *coreUnit, grid gridContext) ([]transpiledSeries, error) {
+	startMs, endMs, stepMs := grid.startMs, grid.endMs, grid.stepMs
+	windowMs := unit.rangeMs
+	if unit.kind == unitInstant {
+		windowMs = e.client.lookbackMs
+	}
+	dataStart := startMs - unit.offsetMs - windowMs
+	dataEnd := endMs - unit.offsetMs
+
+	seriesQuery, seriesArgs, err := buildSeriesQuery(dataStart, dataEnd, unit.matchers)
+	if err != nil {
+		return nil, err
+	}
+	lookup, err := e.client.selectSeries(ctx, seriesQuery, seriesArgs)
+	if err != nil {
+		return nil, err
+	}
+	if len(lookup.fingerprints) == 0 {
+		return nil, nil
+	}
+
+	query, args, err := buildUnitSQL(unit, lookup.metricNames, dataStart, dataEnd, startMs, endMs, stepMs, e.client.lookbackMs)
+	if err != nil {
+		return nil, err
+	}
+
+	rows, err := e.client.telemetryStore.ClickhouseDB().Query(e.client.withContext(ctx, "transpiledUnit"), query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	// Name-dropping units keep __name__ in the SQL group key so distinct
+	// metrics never merge server-side; the name comes off here. Two metrics
+	// can then share a labelset — the engine merges their samples into one
+	// series when they never overlap in time (a selector spanning metrics
+	// whose series alternate across lookback windows) and raises the
+	// duplicate-labelset error only when two samples land on the same
+	// evaluation timestamp. mergeSameLabelsetSeries reproduces exactly that.
+	stripName := !unit.hasAgg && !unit.keepsName()
+
+	// by (...) units return one plain column per grouped label; everything
+	// else returns the single canonical JSON key (see groupKeyColumns).
+	keyNames := groupKeyColumns(unit)
+	keyVals := make([]string, max(len(keyNames), 1))
+	targets := make([]any, 0, len(keyVals)+1)
+	for i := range keyVals {
+		targets = append(targets, &keyVals[i])
+	}
+	var gridValues []*float64
+	targets = append(targets, &gridValues)
+
+	var out []transpiledSeries
+	for rows.Next() {
+		if err := rows.Scan(targets...); err != nil {
+			return nil, err
+		}
+		var lset labels.Labels
+		if keyNames != nil {
+			builder := labels.NewScratchBuilder(len(keyNames))
+			for i, name := range keyNames {
+				// An empty extracted value is the label being absent.
+				if keyVals[i] != "" {
+					builder.Add(name, keyVals[i])
+				}
+			}
+			builder.Sort()
+			lset = builder.Labels()
+		} else {
+			lset, err = labelsFromGroupKey(keyVals[0])
+			if err != nil {
+				return nil, err
+			}
+		}
+		if stripName {
+			lset = labels.NewBuilder(lset).Del(metricNameLabel).Labels()
+		}
+		values := make([]*float64, len(gridValues))
+		copy(values, gridValues)
+		applyScalarOps(unit.ops, values)
+		out = append(out, transpiledSeries{lset: lset, values: values})
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	if stripName {
+		if out, err = mergeSameLabelsetSeries(out); err != nil {
+			return nil, err
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return labels.Compare(out[i].lset, out[j].lset) < 0 })
+	return out, nil
+}
+
+// mergeSameLabelsetSeries combines series left with identical labelsets by a
+// name strip, slot by slot: the engine assembles its result matrix by
+// labelset, so post-strip twins whose points interleave in time are one
+// series to it, and two values on the same evaluation timestamp are its
+// duplicate-labelset error — v1 would have errored there too, so silently
+// picking one value would be a divergence.
+func mergeSameLabelsetSeries(in []transpiledSeries) ([]transpiledSeries, error) {
+	index := make(map[uint64]int, len(in))
+	out := in[:0]
+	for _, s := range in {
+		hash := s.lset.Hash()
+		idx, ok := index[hash]
+		if ok && labels.Equal(out[idx].lset, s.lset) {
+			dst := out[idx].values
+			for k, v := range s.values {
+				if v == nil {
+					continue
+				}
+				if dst[k] != nil {
+					return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "vector cannot contain metrics with the same labelset")
+				}
+				dst[k] = v
+			}
+			continue
+		}
+		index[hash] = len(out)
+		out = append(out, s)
+	}
+	return out, nil
+}
+
+// labelsFromGroupKey parses the toJSONString'd sorted [key, value] pairs.
+func labelsFromGroupKey(gkey string) (labels.Labels, error) {
+	var pairs [][]string
+	if err := json.Unmarshal([]byte(gkey), &pairs); err != nil {
+		return labels.EmptyLabels(), errors.WrapInternalf(err, errors.CodeInternal, "malformed compiled group key %q", gkey)
+	}
+	builder := labels.NewScratchBuilder(len(pairs))
+	for _, p := range pairs {
+		if len(p) != 2 {
+			return labels.EmptyLabels(), errors.NewInternalf(errors.CodeInternal, "malformed compiled group key pair %q", gkey)
+		}
+		builder.Add(p[0], p[1])
+	}
+	builder.Sort()
+	return builder.Labels(), nil
+}
+
+// applyScalarOps applies the number-literal op pipeline in place, with the
+// same float64 arithmetic and comparison-filter semantics as the engine.
+func applyScalarOps(ops []scalarOp, values []*float64) {
+	for _, op := range ops {
+		for i, v := range values {
+			if v == nil {
+				continue
+			}
+			lhs, rhs := *v, op.scalar
+			if op.scalarOnLeft {
+				lhs, rhs = op.scalar, *v
+			}
+			switch op.op {
+			case parser.ADD:
+				res := lhs + rhs
+				values[i] = &res
+			case parser.SUB:
+				res := lhs - rhs
+				values[i] = &res
+			case parser.MUL:
+				res := lhs * rhs
+				values[i] = &res
+			case parser.DIV:
+				res := lhs / rhs
+				values[i] = &res
+			case parser.MOD:
+				res := math.Mod(lhs, rhs)
+				values[i] = &res
+			case parser.POW:
+				res := math.Pow(lhs, rhs)
+				values[i] = &res
+			default:
+				keep := compare(op.op, lhs, rhs)
+				switch {
+				case op.returnBool:
+					res := 0.0
+					if keep {
+						res = 1.0
+					}
+					values[i] = &res
+				case keep:
+					// Filter comparisons keep the vector-side value.
+					vec := *v
+					values[i] = &vec
+				default:
+					values[i] = nil
+				}
+			}
+		}
+	}
+}
+
+func compare(op parser.ItemType, lhs, rhs float64) bool {
+	switch op {
+	case parser.EQLC:
+		return lhs == rhs
+	case parser.NEQ:
+		return lhs != rhs
+	case parser.GTR:
+		return lhs > rhs
+	case parser.LSS:
+		return lhs < rhs
+	case parser.GTE:
+		return lhs >= rhs
+	case parser.LTE:
+		return lhs <= rhs
+	}
+	return false
+}
+
+// toMatrix converts a unit result to a promql matrix on the query grid.
+func toMatrix(series []transpiledSeries, startMs, stepMs int64) promql.Matrix {
+	matrix := make(promql.Matrix, 0, len(series))
+	for _, s := range series {
+		var floats []promql.FPoint
+		for i, v := range s.values {
+			if v == nil {
+				continue
+			}
+			floats = append(floats, promql.FPoint{T: startMs + int64(i)*stepMs, F: *v})
+		}
+		if len(floats) == 0 {
+			continue
+		}
+		matrix = append(matrix, promql.Series{Metric: s.lset, Floats: floats})
+	}
+	return matrix
+}
+
+// executeHybrid substitutes each unit's grids into the engine as synthetic
+// series and evaluates the rewritten query over a storage that serves
+// synthetic selectors from memory and everything else from the live querier.
+// Absent grid points become stale markers so the engine's lookback cannot
+// resurrect the previous grid point. Each unit's synthetic samples sit on its
+// own grid (query grid, or subquery grid for units inside subqueries).
+func (e *executor) executeHybrid(ctx context.Context, plan *transpilePlan, results [][]transpiledSeries) (promql.Matrix, error) {
+	synthetic := make(map[string][]*series, len(plan.units))
+	staleMarker := math.Float64frombits(promValue.StaleNaN)
+
+	queryGrid := plan.grid
+
+	for i, unit := range plan.units {
+		g := unit.grid
+		gridLen := 1
+		if g.stepMs > 0 {
+			gridLen = int((g.endMs-g.startMs)/g.stepMs) + 1
+		}
+		list := make([]*series, 0, len(results[i]))
+		for _, cs := range results[i] {
+			builder := labels.NewBuilder(cs.lset)
+			builder.Set(metricNameLabel, unit.name)
+			s := &series{lset: builder.Labels()}
+			s.ts = make([]int64, 0, gridLen)
+			s.vs = make([]float64, 0, gridLen)
+			for idx := 0; idx < gridLen; idx++ {
+				t := g.startMs + int64(idx)*g.stepMs
+				var v float64
+				if idx < len(cs.values) && cs.values[idx] != nil {
+					v = *cs.values[idx]
+				} else {
+					v = staleMarker
+				}
+				s.ts = append(s.ts, t)
+				s.vs = append(s.vs, v)
+			}
+			list = append(list, s)
+		}
+		synthetic[unit.name] = list
+	}
+
+	hybrid := &hybridQueryable{client: e.client, synthetic: synthetic}
+
+	var qry promql.Query
+	var err error
+	if queryGrid.stepMs == 0 {
+		qry, err = e.engine.NewInstantQuery(ctx, hybrid, nil, plan.rewritten, time.UnixMilli(queryGrid.endMs))
+	} else {
+		qry, err = e.engine.NewRangeQuery(ctx, hybrid, nil, plan.rewritten, time.UnixMilli(queryGrid.startMs), time.UnixMilli(queryGrid.endMs), time.Duration(queryGrid.stepMs)*time.Millisecond)
+	}
+	if err != nil {
+		return nil, err
+	}
+	defer qry.Close()
+
+	res := qry.Exec(ctx)
+	if res.Err != nil {
+		return nil, res.Err
+	}
+
+	matrix, err := resultToMatrix(res)
+	if err != nil {
+		return nil, err
+	}
+
+	// Deep-copy before Close returns the result's slices to the engine pool,
+	// and drop the synthetic __name__ that filter comparisons preserve.
+	out := make(promql.Matrix, 0, len(matrix))
+	for _, s := range matrix {
+		lset := s.Metric
+		if name := lset.Get(metricNameLabel); len(name) >= len(syntheticNamePrefix) && name[:len(syntheticNamePrefix)] == syntheticNamePrefix {
+			builder := labels.NewBuilder(lset)
+			builder.Del(metricNameLabel)
+			lset = builder.Labels()
+		}
+		floats := make([]promql.FPoint, len(s.Floats))
+		copy(floats, s.Floats)
+		out = append(out, promql.Series{Metric: lset.Copy(), Floats: floats})
+	}
+	// The strip can leave twins: two units' outputs distinguishable only by
+	// their synthetic names (e.g. -metric_a or -metric_b, both {} to the
+	// engine's real evaluation once names dropped). The engine assembles its
+	// matrix by labelset, merging such temporally-disjoint elements into one
+	// series; reproduce that, with its duplicate error on same-timestamp
+	// overlap.
+	out, err = mergeMatrixByLabelset(out)
+	if err != nil {
+		return nil, err
+	}
+	sort.Slice(out, func(i, j int) bool { return labels.Compare(out[i].Metric, out[j].Metric) < 0 })
+	return out, nil
+}
+
+// mergeMatrixByLabelset merges series sharing a labelset by interleaving
+// their points in timestamp order; a timestamp present in both is the
+// engine's duplicate-labelset error.
+func mergeMatrixByLabelset(matrix promql.Matrix) (promql.Matrix, error) {
+	index := make(map[uint64]int, len(matrix))
+	out := matrix[:0]
+	for _, s := range matrix {
+		hash := s.Metric.Hash()
+		idx, ok := index[hash]
+		if ok && labels.Equal(out[idx].Metric, s.Metric) {
+			merged := make([]promql.FPoint, 0, len(out[idx].Floats)+len(s.Floats))
+			a, b := out[idx].Floats, s.Floats
+			for len(a) > 0 && len(b) > 0 {
+				switch {
+				case a[0].T < b[0].T:
+					merged, a = append(merged, a[0]), a[1:]
+				case b[0].T < a[0].T:
+					merged, b = append(merged, b[0]), b[1:]
+				default:
+					return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "vector cannot contain metrics with the same labelset")
+				}
+			}
+			out[idx].Floats = append(append(merged, a...), b...)
+			continue
+		}
+		index[hash] = len(out)
+		out = append(out, s)
+	}
+	return out, nil
+}
+
+func resultToMatrix(res *promql.Result) (promql.Matrix, error) {
+	switch v := res.Value.(type) {
+	case promql.Matrix:
+		return v, nil
+	case promql.Vector:
+		matrix := make(promql.Matrix, 0, len(v))
+		for _, s := range v {
+			matrix = append(matrix, promql.Series{Metric: s.Metric, Floats: []promql.FPoint{{T: s.T, F: s.F}}})
+		}
+		return matrix, nil
+	case promql.Scalar:
+		return promql.Matrix{{Metric: labels.EmptyLabels(), Floats: []promql.FPoint{{T: v.T, F: v.V}}}}, nil
+	default:
+		return nil, errors.NewInternalf(errors.CodeInternal, "unexpected hybrid result type %T", res.Value)
+	}
+}
+
+// hybridQueryable serves synthetic (compiled) selectors from memory and
+// everything else from the live storage.
+type hybridQueryable struct {
+	client    *client
+	synthetic map[string][]*series
+}
+
+func (h *hybridQueryable) Querier(mint, maxt int64) (storage.Querier, error) {
+	return &hybridQuerier{
+		querier:   querier{mint: mint, maxt: maxt, client: h.client},
+		synthetic: h.synthetic,
+	}, nil
+}
+
+type hybridQuerier struct {
+	querier
+	synthetic map[string][]*series
+}
+
+func (h *hybridQuerier) Select(ctx context.Context, sortSeries bool, hints *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
+	if name, ok := isSyntheticSelector(matchers); ok {
+		list := h.synthetic[name]
+		if sortSeries {
+			sorted := make([]*series, len(list))
+			copy(sorted, list)
+			sort.Slice(sorted, func(i, j int) bool { return labels.Compare(sorted[i].lset, sorted[j].lset) < 0 })
+			list = sorted
+		}
+		return newSeriesSet(list)
+	}
+	return h.querier.Select(ctx, sortSeries, hints, matchers...)
+}

pkg/prometheus/clickhouseprometheusv2/transpiler_sql.go

@@ -0,0 +1,414 @@
+package clickhouseprometheusv2
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+// experimental gate for the timeSeries*ToGrid aggregate functions; attached
+// as a SETTINGS clause so telemetrystore hooks cannot clobber it.
+const gridFunctionsSetting = "SETTINGS allow_experimental_ts_to_grid_aggregate_function = 1"
+
+var aggForEach = map[string]string{
+	"sum":   "sumForEach",
+	"min":   "minForEach",
+	"max":   "maxForEach",
+	"avg":   "avgForEach",
+	"count": "countForEach",
+}
+
+// buildUnitSQL renders the single ClickHouse query evaluating a core unit
+// over the [startMs, endMs] / stepMs evaluation grid: per-series grids via a
+// timeSeries*ToGrid aggregate (or a windowed aggregation for *_over_time),
+// then spatial aggregation with -ForEach combinators grouped by a canonical
+// JSON key of the projected label pairs.
+//
+// The heavy level is shaped to run on the shards: the top-level FROM is the
+// distributed samples table and the group-key join partner is a subquery on
+// the shard-local time series table, so the shard rewrite executes the join
+// and the per-(fingerprint, group key) aggregation next to the data —
+// complete by fingerprint co-locality (see localTimeSeriesTable) — and the
+// initiator only merges the per-series states and applies the spatial
+// -ForEach step. Same layout as the telemetrymetrics statement builder.
+// The windowed *_over_time form shares the frame but aggregates per
+// (series, group key, step bucket) instead of straight to grids
+// (see windowedInner).
+//
+// The selector's data window is offset-shifted; the resulting grid indices
+// map 1:1 onto the query grid (output ts = startMs + i*stepMs). Grid
+// parameters are rendered as literals — they are aggregate-function
+// parameters, not bindable values.
+//
+// Statements nest builder-rendered SQL as text, so the returned args must be
+// ordered by where each fragment lands in the final statement: ClickHouse
+// binds ? placeholders by position. A JOIN renders before WHERE, so a joined
+// subquery's args precede the outer query's own condition args.
+//
+// Row shape: the group-key columns (see groupKeyColumns) followed by
+// grid Array(Nullable(Float64)); NULL grid points are absent points (the
+// engine's "no value here"), which the -ForEach combinators preserve: an
+// index where every series is NULL aggregates to NULL, and countForEach's 0
+// is mapped back to NULL.
+func buildUnitSQL(unit *coreUnit, metricNames []string, dataStart, dataEnd int64, startMs, endMs, stepMs, lookbackMs int64) (string, []any, error) {
+	selStart := startMs - unit.offsetMs
+	selEnd := endMs - unit.offsetMs
+	stepSec := stepMs / 1000
+	if stepSec == 0 {
+		// Instant query: start == end, so the grid has one point for any
+		// positive step.
+		stepSec = 1
+	}
+	windowMs := unit.rangeMs
+	if unit.kind == unitInstant {
+		windowMs = lookbackMs
+	}
+	windowSec := windowMs / 1000
+
+	adjustedTsStartU, _, _, localTsTable := metricstelemetryschema.WhichTSTableToUse(uint64(dataStart), uint64(dataEnd), false, nil)
+	adjustedTsStart := int64(adjustedTsStartU)
+	keyNames := groupKeyColumns(unit)
+
+	// seriesSub computes fingerprint -> group key columns. It reads the
+	// local series table when it rides inside the shard-rewritten samples
+	// query, and the distributed one when it joins at the initiator
+	// (windowed form).
+	seriesSub := func(table string) (string, []any, error) {
+		sub := sqlbuilder.NewSelectBuilder()
+		selects := []string{"fingerprint"}
+		if keyNames == nil {
+			selects = append(selects, groupKeyExpr(unit)+" AS gkey")
+		} else {
+			// by (...) grouping extracts exactly the listed labels as plain
+			// columns: no reason to build, sort and stringify every label
+			// pair per row when the projection is a known short list and
+			// the label names live in Go anyway.
+			for i, name := range keyNames {
+				selects = append(selects, fmt.Sprintf("JSONExtractString(labels, %s) AS g%d", sub.Var(name), i))
+			}
+		}
+		sub.Select(selects...)
+		sub.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, table))
+		if err := applySeriesConditions(sub, adjustedTsStart, dataEnd, unit.matchers); err != nil {
+			return "", nil, err
+		}
+		sub.GroupBy(append([]string{"fingerprint"}, keyColumnAliases(keyNames)...)...)
+		q, args := sub.BuildWithFlavor(sqlbuilder.ClickHouse)
+		return q, args, nil
+	}
+
+	// samplesConditions adds the samples-side WHERE. The group-key join
+	// restricts to the matched series; no fingerprint condition is added
+	// here.
+	samplesConditions := func(sb *sqlbuilder.SelectBuilder, excludeStale bool) {
+		switch len(metricNames) {
+		case 0:
+			// No name constraint derivable; correct but unable to use the
+			// metric_name primary-key prefix.
+		case 1:
+			sb.Where(sb.EQ("metric_name", metricNames[0]))
+		default:
+			sb.Where(sb.In("metric_name", sqlbuilder.List(metricNames)))
+		}
+		// temporality precedes metric_name in the samples primary key; the
+		// fingerprints already come from these temporalities, so this only
+		// helps granule pruning.
+		sb.Where("temporality IN ['Cumulative', 'Unspecified']")
+		// When the window is narrower than the step, the grid windows
+		// (t_k − window, t_k] cover only window/step of the selector's
+		// timeline; a sample in a gap belongs to no window and cannot move
+		// any grid point, but the grid aggregate buffers every row it is
+		// fed. Keeping only in-window rows cut a 36k-series 1w rate from
+		// 74s/28GiB to 16s/4.3GiB on fleet data — the read stays the same,
+		// the aggregate input shrinks by the coverage ratio. The lattice
+		// anchors at selStart (end may sit off-lattice on unaligned grids),
+		// positiveModulo because samples above selStart make the dividend
+		// negative, and the upper bound tightens to the last grid point —
+		// rows past it are equally windowless. window >= step tiles the
+		// timeline and keeps today's plain bounds.
+		sliver := stepMs > 0 && windowMs > 0 && windowMs < stepMs
+		upper := selEnd
+		if sliver {
+			upper = selStart + (selEnd-selStart)/stepMs*stepMs
+		}
+		// Left-open window: a sample exactly at the window's lower boundary
+		// is never used (range selectors and lookback are both left-open).
+		sb.Where(sb.GT("unix_milli", selStart-windowMs), sb.LTE("unix_milli", upper))
+		if sliver {
+			sb.Where(fmt.Sprintf("positiveModulo(%s - unix_milli, %s) < %s",
+				sb.Var(selStart), sb.Var(stepMs), sb.Var(windowMs)))
+		}
+		if excludeStale {
+			// PromQL excludes stale markers from range vectors. Instant
+			// selectors need the stale rows for shadowing instead.
+			sb.Where("bitAnd(flags, 1) = 0")
+		}
+	}
+
+	keyCols := keyColumnAliases(keyNames)
+
+	// joinedInner builds the shard-side SELECT for the single-pass kinds:
+	// grid expression per (fingerprint, group key), group-key join against
+	// the local series table.
+	joinedInner := func(gridExpr string, excludeStale bool) (string, []any, error) {
+		seriesSQL, seriesArgs, err := seriesSub(localTsTable)
+		if err != nil {
+			return "", nil, err
+		}
+		sb := sqlbuilder.NewSelectBuilder()
+		selects := make([]string, 0, len(keyCols)+1)
+		// A fingerprint is the hash of one labelset, so every group-key
+		// column is functionally dependent on it: any() is exact, and
+		// grouping by the fingerprint alone spares hashing the joined
+		// string per sample row — measured -10-13% on a 1.9B-row rate.
+		for _, col := range keyCols {
+			selects = append(selects, fmt.Sprintf("any(series.%s) AS %s", col, col))
+		}
+		sb.Select(append(selects, gridExpr+" AS grid")...)
+		sb.From(fmt.Sprintf("%s.%s AS points", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4TableName))
+		sb.JoinWithOption(sqlbuilder.InnerJoin, fmt.Sprintf("(%s) AS series", seriesSQL), "points.fingerprint = series.fingerprint")
+		samplesConditions(sb, excludeStale)
+		sb.GroupBy("points.fingerprint")
+		q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+		// The join text renders before WHERE: its args come first.
+		return q, append(seriesArgs, args...), nil
+	}
+
+	var inner string
+	var innerArgs []any
+	var err error
+	switch unit.kind {
+	case unitInstant:
+		// Instant selection with stale shadowing: the grid value is the last
+		// non-stale sample in (t-lookback, t], absent when the overall last
+		// sample in that window is a stale marker (verified semantics: the
+		// -If combinator applies to the grid aggregates, and NULL comparisons
+		// make a stale-latest point absent).
+		gridParams := fmt.Sprintf("(fromUnixTimestamp64Milli(%d), fromUnixTimestamp64Milli(%d), %d, %d)", selStart, selEnd, stepSec, windowSec)
+		gridExpr := fmt.Sprintf(
+			"arrayMap((tall, tok, vok) -> if(tall IS NULL OR tok IS NULL OR tall != tok, NULL, vok), timeSeriesLastToGrid%s(fromUnixTimestamp64Milli(unix_milli), toFloat64(unix_milli)), timeSeriesLastToGridIf%s(fromUnixTimestamp64Milli(unix_milli), toFloat64(unix_milli), bitAnd(flags, 1) = 0), timeSeriesLastToGridIf%s(fromUnixTimestamp64Milli(unix_milli), value, bitAnd(flags, 1) = 0))",
+			gridParams, gridParams, gridParams,
+		)
+		inner, innerArgs, err = joinedInner(gridExpr, false)
+	case unitOverTime:
+		if unit.overFn == "last" {
+			// last_over_time == last non-stale sample in the window: the
+			// stale rows are already excluded in WHERE.
+			gridExpr := fmt.Sprintf(
+				"timeSeriesLastToGrid(fromUnixTimestamp64Milli(%d), fromUnixTimestamp64Milli(%d), %d, %d)(fromUnixTimestamp64Milli(unix_milli), value)",
+				selStart, selEnd, stepSec, windowSec,
+			)
+			inner, innerArgs, err = joinedInner(gridExpr, true)
+			break
+		}
+		inner, innerArgs, err = windowedInner(unit, samplesConditions, seriesSub, keyCols, localTsTable, selStart, selEnd, stepMs, windowMs)
+	default: // unitRange
+		gridExpr := fmt.Sprintf(
+			"%s(fromUnixTimestamp64Milli(%d), fromUnixTimestamp64Milli(%d), %d, %d)(fromUnixTimestamp64Milli(unix_milli), value)",
+			gridFunction[unit.fn], selStart, selEnd, stepSec, windowSec,
+		)
+		if unit.fn == fnIncrease {
+			// increase == rate * range-seconds, exactly: extrapolatedRate
+			// divides by the range only when isRate.
+			gridExpr = fmt.Sprintf("arrayMap(x -> x * %d, %s)", windowSec, gridExpr)
+		}
+		inner, innerArgs, err = joinedInner(gridExpr, true)
+	}
+	if err != nil {
+		return "", nil, err
+	}
+
+	spatial := "maxForEach(grid)"
+	switch {
+	case !unit.hasAgg:
+		// Per-series output: one row per (labels-minus-__name__) group.
+		// Distinct fingerprints can collapse onto the same projected label
+		// set only via a regex __name__ selector over metrics with identical
+		// other labels; maxForEach is a deterministic NULL-skipping merge and
+		// the identity for the overwhelmingly common one-fingerprint group.
+	case unit.aggOp.String() == "count":
+		// count over an all-absent index is an absent point, not 0.
+		spatial = "arrayMap(c -> if(c = 0, NULL, toFloat64(c)), countForEach(grid))"
+	default:
+		spatial = fmt.Sprintf("%s(grid)", aggForEach[unit.aggOp.String()])
+	}
+
+	keyList := strings.Join(keyCols, ", ")
+	query := fmt.Sprintf("SELECT %s, %s AS grid FROM (%s) GROUP BY %s %s", keyList, spatial, inner, keyList, gridFunctionsSetting)
+	return query, innerArgs, nil
+}
+
+// groupKeyColumns returns the label names to extract as plain group-key
+// columns, or nil when the unit needs the canonical JSON key instead. Only
+// by (...) grouping qualifies: its projection is a known short list, so
+// extracting each label directly beats building, sorting and stringifying
+// every label pair per row. without and no-aggregation project a label SET
+// that varies per series — there the sorted-JSON key is load-bearing: the
+// sort is what makes two fingerprints with different stored JSON key order
+// land in one group, and the string carries the labels back out.
+func groupKeyColumns(unit *coreUnit) []string {
+	if unit.hasAgg && unit.by && len(unit.grouping) > 0 {
+		return unit.grouping
+	}
+	return nil
+}
+
+// keyColumnAliases names the group-key columns in every SELECT level: g0..gN
+// for direct extraction, the single canonical gkey otherwise.
+func keyColumnAliases(keyNames []string) []string {
+	if keyNames == nil {
+		return []string{"gkey"}
+	}
+	cols := make([]string, len(keyNames))
+	for i := range keyNames {
+		cols[i] = fmt.Sprintf("g%d", i)
+	}
+	return cols
+}
+
+// windowedInner builds the avg/min/max/sum/count _over_time form without
+// fanning samples out. It runs only when the range is a whole multiple of
+// the step (see the transpile gate), because then the window
+// (t_k - range, t_k] is exactly the union of W = range/step step buckets —
+// both are left-open on the same boundaries — so bucket membership fully
+// determines window membership. Fanning each sample into all W windows it
+// covers (ARRAY JOIN) multiplies rows by W, which at long ranges over short
+// steps is a row explosion measured in billions.
+//
+// The bucketing itself is the -Resample combinator: one group per (series,
+// group key) whose state is a fixed array of per-bucket aggregates, updated
+// in place per sample. Grouping by (series, bucket) instead — measured on a
+// 100k-series x 371-bucket workload — creates a 37M-entry hash aggregation
+// whose per-thread partial tables scale memory WITH max_threads (12 -> 48
+// GiB from 2 to 8 threads, dead at 16) and ships one row per group to the
+// initiator; the Resample form carries the same numbers in 100k compact
+// array states, like every other unit kind.
+//
+// The wrapper level slides the window: slot k combines buckets k..k+W-1 by
+// direct aggregation over at most W partials — no prefix-sum tricks, so no
+// large-minus-large cancellation against the engine's directly-summed
+// windows. A slot with zero window count is absent, which also keeps
+// min/max honest: their slices filter on the bucket counts, so an empty
+// bucket's zero-fill can never be mistaken for a value (a real sample can
+// legitimately be 0 or +Inf).
+func windowedInner(unit *coreUnit, samplesConditions func(*sqlbuilder.SelectBuilder, bool), seriesSub func(string) (string, []any, error), keyCols []string, localSeriesTable string, selStart, selEnd, stepMs, windowMs int64) (string, []any, error) {
+	effStepMs := stepMs
+	if effStepMs == 0 {
+		effStepMs = 1000
+	}
+	lastIdx := (selEnd - selStart) / effStepMs
+	gridLen := lastIdx + 1
+	w := windowMs / effStepMs
+	bucketLen := gridLen + w
+
+	// A window narrower than the step makes the windows (t_k - range, t_k]
+	// pairwise disjoint: there is nothing to slide, each slot reads exactly
+	// its own window's aggregate. This is exact ONLY over sliver-filtered
+	// rows (samplesConditions adds the window<step predicate): the index
+	// below assigns every gap sample to the window above it, and the filter
+	// is what removes them. Requires a real step — instant queries carry no
+	// sliver filter, so they keep the tiled form and its gates.
+	disjoint := stepMs > 0 && windowMs < stepMs
+	if disjoint {
+		w = 1
+		bucketLen = gridLen
+	}
+
+	seriesSQL, seriesArgs, err := seriesSub(localSeriesTable)
+	if err != nil {
+		return "", nil, err
+	}
+
+	// Bucket index, shifted so the earliest in-window sample lands at 0:
+	// jj = ceil((ts - selStart)/step) + W - 1, folded into one intDiv. Slot
+	// k's window is then buckets jj in [k, k+W-1]. In the disjoint form the
+	// same ceil lands each in-window sample directly on its slot (W = 1),
+	// and the numerator stays positive: the fetch floor is
+	// selStart - range > selStart - step.
+	jjShift := windowMs
+	if disjoint {
+		jjShift = effStepMs
+	}
+	jj := fmt.Sprintf("intDiv(unix_milli - %d + %d - 1, %d)", selStart, jjShift, effStepMs)
+	buckets := sqlbuilder.NewSelectBuilder()
+	selects := make([]string, 0, len(keyCols)+2)
+	// any() over the group key: exact because the key is functionally
+	// dependent on the fingerprint (see joinedInner).
+	for _, col := range keyCols {
+		selects = append(selects, fmt.Sprintf("any(series.%s) AS %s", col, col))
+	}
+	selects = append(selects, fmt.Sprintf("countResample(0, %d, 1)(value, %s) AS cnts", bucketLen, jj))
+	if unit.overFn != "count" {
+		selects = append(selects, fmt.Sprintf("%sResample(0, %d, 1)(value, %s) AS vals", map[string]string{
+			"avg": "sum",
+			"sum": "sum",
+			"min": "min",
+			"max": "max",
+		}[unit.overFn], bucketLen, jj))
+	}
+	buckets.Select(selects...)
+	buckets.From(fmt.Sprintf("%s.%s AS points", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4TableName))
+	buckets.JoinWithOption(sqlbuilder.InnerJoin, fmt.Sprintf("(%s) AS series", seriesSQL), "points.fingerprint = series.fingerprint")
+	samplesConditions(buckets, true)
+	buckets.GroupBy("points.fingerprint")
+	bucketsSQL, bucketsArgs := buckets.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	windowCnt := fmt.Sprintf("arraySum(arraySlice(cnts, k + 1, %d))", w)
+	var slot string
+	switch unit.overFn {
+	case "count":
+		slot = fmt.Sprintf("if(%s = 0, NULL, toFloat64(%s))", windowCnt, windowCnt)
+	case "sum":
+		slot = fmt.Sprintf("if(%s = 0, NULL, arraySum(arraySlice(vals, k + 1, %d)))", windowCnt, w)
+	case "avg":
+		slot = fmt.Sprintf("if(%s = 0, NULL, arraySum(arraySlice(vals, k + 1, %d)) / %s)", windowCnt, w, windowCnt)
+	case "min":
+		slot = fmt.Sprintf("if(%s = 0, NULL, arrayMin(arrayFilter((v, c) -> c > 0, arraySlice(vals, k + 1, %d), arraySlice(cnts, k + 1, %d))))", windowCnt, w, w)
+	case "max":
+		slot = fmt.Sprintf("if(%s = 0, NULL, arrayMax(arrayFilter((v, c) -> c > 0, arraySlice(vals, k + 1, %d), arraySlice(cnts, k + 1, %d))))", windowCnt, w, w)
+	}
+
+	keyList := strings.Join(keyCols, ", ")
+	inner := fmt.Sprintf(
+		"SELECT %s, arrayMap(k -> %s, range(toUInt64(%d))) AS grid FROM (%s)",
+		keyList, slot, gridLen, bucketsSQL,
+	)
+	return inner, append(seriesArgs, bucketsArgs...), nil
+}
+
+// groupKeyExpr renders the canonical JSON group key for the units whose
+// projected label SET varies per series (see groupKeyColumns): the sorted
+// [key, value] pairs of the projected labels, JSON-encoded.
+//   - by () with no labels: one constant group;
+//   - without (a, b): keep everything except the listed labels and __name__;
+//   - no aggregation: keep everything including __name__ — even when the
+//     unit drops the name from its OUTPUT, the key must keep it so distinct
+//     metrics never merge in SQL; executeUnit strips the name afterwards and
+//     turns a post-strip collision into the engine's duplicate-labelset
+//     error instead of a silently invented merge.
+func groupKeyExpr(unit *coreUnit) string {
+	// An empty label value means "label absent" in Prometheus; the stored
+	// labels JSON can carry empty attribute values, which must not become
+	// output labels or group keys.
+	pairs := "arraySort(JSONExtractKeysAndValues(labels, 'String'))"
+	if !unit.hasAgg {
+		return fmt.Sprintf("toJSONString(arrayFilter(p -> p.2 != '', %s))", pairs)
+	}
+	if unit.by {
+		// Non-empty by (...) never reaches here; groupKeyColumns extracts
+		// those labels as plain columns instead.
+		return "'[]'"
+	}
+	excluded := append([]string{metricNameLabel}, unit.grouping...)
+	return fmt.Sprintf("toJSONString(arrayFilter(p -> p.2 != '' AND p.1 NOT IN (%s), %s))", quotedList(excluded), pairs)
+}
+
+func quotedList(items []string) string {
+	quoted := make([]string, len(items))
+	for i, s := range items {
+		quoted[i] = "'" + strings.ReplaceAll(s, "'", "\\'") + "'"
+	}
+	return strings.Join(quoted, ", ")
+}

pkg/prometheus/clickhouseprometheusv2/transpiler_test.go

@@ -0,0 +1,751 @@
+package clickhouseprometheusv2
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	cmock "github.com/SigNoz/clickhouse-go-mock"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/promql/parser"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newTestClient(t *testing.T) (*client, *telemetrystoretest.Provider) {
+	t.Helper()
+	store := telemetrystoretest.New(telemetrystore.Config{Provider: "clickhouse"}, sqlmock.QueryMatcherRegexp)
+	settings := factory.NewScopedProviderSettings(instrumentationtest.New().ToProviderSettings(), "clickhouseprometheusv2_test")
+	return newClient(settings, store, prometheus.Config{}), store
+}
+
+var seriesCols = []cmock.ColumnType{
+	{Name: "fingerprint", Type: "UInt64"},
+	{Name: "labels", Type: "String"},
+}
+
+func parse(t *testing.T, q string) parser.Expr {
+	t.Helper()
+	expr, err := parser.NewParser(parser.Options{}).ParseExpr(q)
+	require.NoError(t, err)
+	return expr
+}
+
+func TestClassifyFullShapes(t *testing.T) {
+	tests := []struct {
+		name  string
+		query string
+		check func(t *testing.T, u *coreUnit)
+	}{
+		{
+			name:  "sum by rate",
+			query: `sum by (pod) (rate(http_requests_total{job="api"}[5m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, fnRate, u.fn)
+				assert.Equal(t, int64(300_000), u.rangeMs)
+				assert.True(t, u.hasAgg)
+				assert.True(t, u.by)
+				assert.Equal(t, []string{"pod"}, u.grouping)
+			},
+		},
+		{
+			name:  "bare increase with offset",
+			query: `increase(errors_total[10m] offset 30m)`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, fnIncrease, u.fn)
+				assert.Equal(t, int64(1_800_000), u.offsetMs)
+				assert.False(t, u.hasAgg)
+			},
+		},
+		{
+			name:  "avg without over delta",
+			query: `avg without (instance) (delta(gauge_metric[15m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, fnDelta, u.fn)
+				assert.True(t, u.hasAgg)
+				assert.False(t, u.by)
+			},
+		},
+		{
+			name:  "scalar pipeline with comparison",
+			query: `sum(rate(x[5m])) * 100 > 5`,
+			check: func(t *testing.T, u *coreUnit) {
+				require.Len(t, u.ops, 2)
+				assert.Equal(t, parser.ItemType(parser.MUL), u.ops[0].op)
+				assert.Equal(t, 100.0, u.ops[0].scalar)
+				assert.Equal(t, parser.ItemType(parser.GTR), u.ops[1].op)
+			},
+		},
+		{
+			name:  "scalar on left with unary minus",
+			query: `-1 * sum(rate(x[5m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				require.Len(t, u.ops, 1)
+				assert.True(t, u.ops[0].scalarOnLeft)
+				assert.Equal(t, -1.0, u.ops[0].scalar)
+			},
+		},
+		{
+			name:  "bool comparison",
+			query: `sum(rate(x[5m])) >= bool 0.5`,
+			check: func(t *testing.T, u *coreUnit) {
+				require.Len(t, u.ops, 1)
+				assert.True(t, u.ops[0].returnBool)
+			},
+		},
+		{
+			name:  "irate utf8 name",
+			query: `sum by ("k8s.pod.name") (irate({"k8s.container.cpu.time"}[2m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, fnIRate, u.fn)
+				assert.Equal(t, []string{"k8s.pod.name"}, u.grouping)
+			},
+		},
+		{
+			name:  "bare instant selector keeps name",
+			query: `up{job="api"}`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitInstant, u.kind)
+				assert.True(t, u.keepsName())
+			},
+		},
+		{
+			name:  "gauge aggregation",
+			query: `sum by (pod) (container_memory offset 5m)`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitInstant, u.kind)
+				assert.Equal(t, int64(300_000), u.offsetMs)
+				assert.True(t, u.hasAgg)
+				assert.False(t, u.keepsName())
+			},
+		},
+		{
+			name:  "gauge comparison keeps name",
+			query: `container_memory > 100`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitInstant, u.kind)
+				assert.True(t, u.keepsName())
+			},
+		},
+		{
+			name:  "gauge arithmetic drops name",
+			query: `container_memory / 1024`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitInstant, u.kind)
+				assert.False(t, u.keepsName())
+			},
+		},
+		{
+			name:  "avg_over_time",
+			query: `max by (node) (avg_over_time(load1[10m]))`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitOverTime, u.kind)
+				assert.Equal(t, "avg", u.overFn)
+				assert.Equal(t, int64(600_000), u.rangeMs)
+			},
+		},
+		{
+			name:  "last_over_time keeps name",
+			query: `last_over_time(load1[10m])`,
+			check: func(t *testing.T, u *coreUnit) {
+				assert.Equal(t, unitOverTime, u.kind)
+				assert.Equal(t, "last", u.overFn)
+				assert.True(t, u.keepsName())
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			plan, ok := classify(parse(t, tt.query), testGrid(60_000))
+			require.True(t, ok, "expected transpilable")
+			require.True(t, plan.full, "expected full compilation")
+			require.Len(t, plan.units, 1)
+			tt.check(t, &plan.units[0].core)
+		})
+	}
+}
+
+func TestClassifyFallbackShapes(t *testing.T) {
+	queries := []struct {
+		name  string
+		query string
+		step  int64
+	}{
+		{"default-resolution subquery", `max_over_time(rate(x[5m])[30m:])`, 60_000},
+		{"at modifier", `sum(rate(x[5m] @ 1609746000))`, 60_000},
+		{"at modifier on gauge", `sum(container_memory @ 1609746000)`, 60_000},
+		{"sub-second step", `sum(rate(x[5m]))`, 500},
+		{"sub-second range", `sum(rate(x[1500ms]))`, 60_000},
+		{"by __name__ full", `sum by (__name__) (rate({__name__=~"a|b"}[5m]))`, 60_000},
+		{"quantile_over_time unsupported", `quantile_over_time(0.9, load1[10m])`, 60_000},
+		// Duration expressions resolve into the selectors' static fields only
+		// at evaluation time; classification reads those fields as zero, so
+		// transpiling would silently use the wrong offset (caught by the
+		// conformance corpus' duration_expression.test cases). Offset
+		// expressions parse without the experimental-parser flag, so they do
+		// reach the transpiler; range-position expressions are rejected at
+		// parse (the RangeExpr/StepExpr guards are defense-in-depth).
+		{"duration expression offset on instant", `x offset step()`, 60_000},
+		{"duration expression offset arithmetic", `x offset -step()*2`, 60_000},
+		{"duration expression offset on range", `sum(rate(x[5m] offset max(3s, step())))`, 60_000},
+		{"duration expression subquery step", `max_over_time(rate(x[5m])[30m:step()])`, 60_000},
+	}
+	for _, tt := range queries {
+		t.Run(tt.name, func(t *testing.T) {
+			_, ok := classify(parse(t, tt.query), testGrid(tt.step))
+			assert.False(t, ok, "expected fallback for %s", tt.query)
+		})
+	}
+}
+
+func TestClassifyHybridShapes(t *testing.T) {
+	tests := []struct {
+		name          string
+		query         string
+		wantUnits     int
+		wantRewritten string
+	}{
+		{
+			name:          "histogram quantile",
+			query:         `histogram_quantile(0.95, sum by (le) (rate(http_bucket[5m])))`,
+			wantUnits:     1,
+			wantRewritten: `histogram_quantile(0.95, __signoz_transpiled_0__)`,
+		},
+		{
+			name:          "topk over compiled",
+			query:         `topk(5, sum by (pod) (rate(x[5m])))`,
+			wantUnits:     1,
+			wantRewritten: `topk(5, __signoz_transpiled_0__)`,
+		},
+		{
+			name:          "ratio of compiled units",
+			query:         `sum(rate(a[5m])) / sum(rate(b[5m]))`,
+			wantUnits:     2,
+			wantRewritten: `__signoz_transpiled_0__ / __signoz_transpiled_1__`,
+		},
+		{
+			name:          "or vector zero",
+			query:         `sum(rate(a[5m])) or vector(0)`,
+			wantUnits:     1,
+			wantRewritten: `__signoz_transpiled_0__ or vector(0)`,
+		},
+		{
+			name:          "quantile agg over compiled rate",
+			query:         `quantile(0.9, rate(x[5m]))`,
+			wantUnits:     1,
+			wantRewritten: `quantile(0.9, __signoz_transpiled_0__)`,
+		},
+		{
+			name:          "non-literal scalar side stays engine-side",
+			query:         `sum(rate(x[5m])) * scalar(y)`,
+			wantUnits:     1,
+			wantRewritten: `__signoz_transpiled_0__ * scalar(y)`,
+		},
+		{
+			name:          "compiled mixed with raw selector",
+			query:         `sum by (pod) (rate(a[5m])) / on (pod) group_left () b`,
+			wantUnits:     1,
+			wantRewritten: `__signoz_transpiled_0__ / on (pod) group_left () b`,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			plan, ok := classify(parse(t, tt.query), testGrid(60_000))
+			require.True(t, ok)
+			assert.False(t, plan.full)
+			assert.Len(t, plan.units, tt.wantUnits)
+			assert.Equal(t, tt.wantRewritten, plan.rewritten)
+		})
+	}
+}
+
+func TestClassifyHybridGuards(t *testing.T) {
+	t.Run("no substitution under on(__name__)", func(t *testing.T) {
+		plan, ok := classify(parse(t, `sum(rate(a[5m])) * on (__name__) b`), testGrid(60_000))
+		_ = plan
+		assert.False(t, ok, "matching on __name__ must not see synthetic names")
+	})
+	t.Run("no substitution inside @-pinned subquery", func(t *testing.T) {
+		_, ok := classify(parse(t, `max_over_time(rate(x[5m])[30m:1m] @ 1609746000)`), testGrid(60_000))
+		assert.False(t, ok)
+	})
+}
+
+// The alert-smoothing idiom: units inside a fixed-resolution subquery
+// evaluate on the subquery grid — epoch-aligned multiples of the resolution,
+// starting strictly after (outer start - range), exactly as the engine
+// derives it.
+func TestClassifySubqueryUnits(t *testing.T) {
+	grid := gridContext{startMs: 1_700_000_030_000, endMs: 1_700_007_200_000, stepMs: 60_000}
+
+	plan, ok := classify(parse(t, `min_over_time((sum by (ns) (increase(x[5m])))[10m:5m]) > 0`), grid)
+	require.True(t, ok)
+	require.False(t, plan.full)
+	require.Len(t, plan.units, 1)
+	assert.Equal(t, `min_over_time(__signoz_transpiled_0__[10m:5m]) > 0`, plan.rewritten)
+
+	unit := plan.units[0]
+	// lower bound = outer start - range = 1_699_999_430_000; first multiple
+	// of 300_000 strictly greater is 1_699_999_500_000.
+	assert.Equal(t, int64(1_699_999_500_000), unit.grid.startMs)
+	assert.Equal(t, grid.endMs, unit.grid.endMs)
+	assert.Equal(t, int64(300_000), unit.grid.stepMs)
+	assert.Equal(t, fnIncrease, unit.core.fn)
+
+	t.Run("subquery offset shifts the grid", func(t *testing.T) {
+		plan, ok := classify(parse(t, `max_over_time((sum(rate(x[5m])))[10m:5m] offset 30m)`), grid)
+		require.True(t, ok)
+		require.Len(t, plan.units, 1)
+		// lower = start - offset - range = 1_699_997_630_000 -> first
+		// multiple of 300_000 above = 1_699_997_700_000; end shifts too.
+		assert.Equal(t, int64(1_699_997_700_000), plan.units[0].grid.startMs)
+		assert.Equal(t, grid.endMs-1_800_000, plan.units[0].grid.endMs)
+	})
+
+	t.Run("mollusk ratio-inside-subquery idiom", func(t *testing.T) {
+		q := `min_over_time(((sum by (a) (rate(m1[5m]))) / (avg by (a) (m2)))[5m:1m])`
+		plan, ok := classify(parse(t, q), grid)
+		require.True(t, ok)
+		// Both sides compile on the subquery grid: the rate side and the
+		// gauge aggregation side; the engine joins them and smooths.
+		require.Len(t, plan.units, 2)
+		assert.Equal(t, int64(60_000), plan.units[0].grid.stepMs)
+		assert.Equal(t, unitInstant, plan.units[1].core.kind)
+		assert.Contains(t, plan.rewritten, `__signoz_transpiled_0__ / __signoz_transpiled_1__`)
+	})
+}
+
+func TestBuildUnitSQL(t *testing.T) {
+	unit := &coreUnit{
+		fn:       fnRate,
+		rangeMs:  300_000,
+		hasAgg:   true,
+		aggOp:    parser.SUM,
+		by:       true,
+		grouping: []string{"pod"},
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "http_requests_total")},
+	}
+	sql, args, err := buildUnitSQL(unit, []string{"http_requests_total"}, 1_699_999_700_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 300_000)
+	require.NoError(t, err)
+
+	assert.Contains(t, sql, "timeSeriesRateToGrid(fromUnixTimestamp64Milli(1700000000000), fromUnixTimestamp64Milli(1700003600000), 60, 300)(fromUnixTimestamp64Milli(unix_milli), value)")
+	assert.Contains(t, sql, "unix_milli > ? AND unix_milli <= ?")
+	assert.Contains(t, sql, "bitAnd(flags, 1) = 0")
+	assert.Contains(t, sql, "sumForEach(grid)")
+	// The group-key join rides inside the shard query: distributed samples
+	// at the top level, the local series table in the join subquery, the
+	// grid aggregation grouped per (fingerprint, group key) shard-side.
+	assert.Contains(t, sql, "FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint,")
+	assert.Contains(t, sql, "FROM signoz_metrics.time_series_v4 WHERE")
+	// The group key is functionally dependent on the fingerprint (one
+	// labelset per fingerprint): any() is exact and the per-row hash key
+	// shrinks to the fingerprint alone.
+	assert.Contains(t, sql, "any(series.g0) AS g0")
+	assert.Contains(t, sql, "GROUP BY points.fingerprint)")
+	// No samples-side fingerprint condition: the group-key join restricts.
+	assert.NotContains(t, sql, "points.fingerprint IN (")
+	// by (pod) extracts the grouped label directly — no per-row JSON
+	// build/sort/stringify for a known projection.
+	assert.Contains(t, sql, "JSONExtractString(labels, ?) AS g0")
+	assert.NotContains(t, sql, "toJSONString")
+	assert.Contains(t, sql, "SETTINGS allow_experimental_ts_to_grid_aggregate_function = 1")
+	// Args follow placeholder order: the joined series subquery renders
+	// before the samples WHERE, and its select list ('pod') renders before
+	// its own conditions.
+	assert.Equal(t, []any{"pod", "http_requests_total", int64(1_699_999_200_000), int64(1_700_003_600_000), "http_requests_total", int64(1_699_999_700_000), int64(1_700_003_600_000)}, args)
+}
+
+func TestBuildUnitSQLIncreaseAndOffset(t *testing.T) {
+	unit := &coreUnit{
+		fn:       fnIncrease,
+		rangeMs:  600_000,
+		offsetMs: 1_800_000,
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "errors_total")},
+	}
+	sql, _, err := buildUnitSQL(unit, nil, 1_699_997_600_000, 1_700_001_800_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 300_000)
+	require.NoError(t, err)
+
+	// Grid and window shift by the offset; increase multiplies rate by the
+	// range in seconds.
+	assert.Contains(t, sql, "fromUnixTimestamp64Milli(1699998200000), fromUnixTimestamp64Milli(1700001800000)")
+	assert.Contains(t, sql, "arrayMap(x -> x * 600, timeSeriesRateToGrid")
+	assert.Contains(t, sql, "maxForEach(grid)")
+}
+
+func TestBuildUnitSQLOverLimitJoinOnly(t *testing.T) {
+	// Past the inline limit no fingerprint filter is rendered: the series
+	// join restricts to the matched fingerprints on its own.
+	unit := &coreUnit{
+		fn:       fnRate,
+		rangeMs:  300_000,
+		hasAgg:   true,
+		aggOp:    parser.SUM,
+		by:       true,
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "http_requests_total")},
+	}
+	sql, _, err := buildUnitSQL(unit, []string{"http_requests_total"}, 1_699_999_700_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 300_000)
+	require.NoError(t, err)
+
+	assert.NotContains(t, sql, "points.fingerprint IN")
+	assert.Contains(t, sql, "INNER JOIN (SELECT fingerprint,")
+	assert.Contains(t, sql, "FROM signoz_metrics.time_series_v4 WHERE")
+}
+
+func TestBuildUnitSQLWindowSliver(t *testing.T) {
+	// rate[5m] on a 30m grid evaluates only a 5m sliver before each grid
+	// point — samples in the gaps belong to no window and would only be
+	// buffered by the grid aggregate. The WHERE must keep exactly the
+	// in-window rows: positiveModulo anchored at the selector start (end
+	// can sit off-lattice on unaligned grids, and samples above the start
+	// make the plain modulo dividend negative), and the scan capped at the
+	// last grid point — rows past it are equally windowless.
+	unit := &coreUnit{
+		fn:       fnRate,
+		rangeMs:  300_000,
+		hasAgg:   true,
+		aggOp:    parser.SUM,
+		by:       true,
+		grouping: []string{"pod"},
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "http_requests_total")},
+	}
+	sql, args, err := buildUnitSQL(unit, []string{"http_requests_total"}, 1_699_999_700_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 1_800_000, 300_000)
+	require.NoError(t, err)
+
+	assert.Contains(t, sql, "positiveModulo(? - unix_milli, ?) < ?")
+	assert.Equal(t, []any{"pod", "http_requests_total", int64(1_699_999_200_000), int64(1_700_003_600_000), "http_requests_total", int64(1_699_999_700_000), int64(1_700_003_600_000), int64(1_700_000_000_000), int64(1_800_000), int64(300_000)}, args)
+
+	t.Run("off-lattice end caps the scan at the last grid point", func(t *testing.T) {
+		// end - start = 50m at a 30m step: the only grid points are start
+		// and start+30m; samples in the trailing 20m serve no window.
+		_, args, err := buildUnitSQL(unit, []string{"http_requests_total"}, 1_699_999_700_000, 1_700_003_000_000, 1_700_000_000_000, 1_700_003_000_000, 1_800_000, 300_000)
+		require.NoError(t, err)
+		assert.Contains(t, args, int64(1_700_001_800_000))
+	})
+
+	t.Run("window covering the step keeps plain bounds", func(t *testing.T) {
+		sql, _, err := buildUnitSQL(unit, []string{"http_requests_total"}, 1_699_999_700_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 300_000)
+		require.NoError(t, err)
+		assert.NotContains(t, sql, "positiveModulo")
+	})
+}
+
+func TestBuildUnitSQLWindowedBucketsWithoutFanOut(t *testing.T) {
+	// The window is W = range/step whole buckets, so each sample lands in
+	// exactly one bucket via GROUP BY and the window slides over bucket
+	// partials — fanning samples into every covered window (ARRAY JOIN)
+	// multiplies rows by W, a row explosion at long ranges.
+	unit := &coreUnit{
+		kind:     unitOverTime,
+		overFn:   "avg",
+		rangeMs:  600_000,
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "node_load1")},
+	}
+	sql, _, err := buildUnitSQL(unit, []string{"node_load1"}, 1_699_999_400_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 60_000, 600_000)
+	require.NoError(t, err)
+
+	assert.NotContains(t, sql, "ARRAY JOIN")
+	// One group per series with fixed per-bucket arrays (-Resample); the
+	// bucket index jj = ceil((ts - start)/step) + W - 1 folded into a single
+	// intDiv. Grouping by (series, bucket) instead measured 37M hash groups
+	// whose per-thread partials scale memory with max_threads.
+	assert.Contains(t, sql, "countResample(0, 71, 1)(value, intDiv(unix_milli - 1700000000000 + 600000 - 1, 60000)) AS cnts")
+	assert.Contains(t, sql, "sumResample(0, 71, 1)(value, intDiv(unix_milli - 1700000000000 + 600000 - 1, 60000)) AS vals")
+	assert.Contains(t, sql, "any(series.gkey) AS gkey")
+	assert.Contains(t, sql, "GROUP BY points.fingerprint)")
+	assert.NotContains(t, sql, "jj) AS jj")
+	assert.Contains(t, sql, "INNER JOIN (SELECT fingerprint,")
+	assert.Contains(t, sql, "FROM signoz_metrics.time_series_v4 WHERE")
+	// Slide: W = 10 buckets per slot, absent when the window count is 0.
+	assert.Contains(t, sql, "arraySum(arraySlice(cnts, k + 1, 10))")
+	assert.Contains(t, sql, "arraySum(arraySlice(vals, k + 1, 10))")
+}
+
+func TestBuildUnitSQLDisjointOverTime(t *testing.T) {
+	// avg_over_time[5m] on a 30m grid: the windows are pairwise disjoint,
+	// so there is no slide — one Resample bucket per grid slot, read
+	// directly. Exact only together with the window-sliver predicate, which
+	// removes the gap samples the ceil index would otherwise assign to the
+	// window above them.
+	unit := &coreUnit{
+		kind:     unitOverTime,
+		overFn:   "avg",
+		rangeMs:  300_000,
+		matchers: []*labels.Matcher{mustMatcher(t, labels.MatchEqual, "__name__", "node_load1")},
+	}
+	sql, _, err := buildUnitSQL(unit, []string{"node_load1"}, 1_699_999_700_000, 1_700_003_600_000, 1_700_000_000_000, 1_700_003_600_000, 1_800_000, 300_000)
+	require.NoError(t, err)
+
+	assert.NotContains(t, sql, "ARRAY JOIN")
+	// gridLen = 3 slots, bucket array the same length — no W tail.
+	assert.Contains(t, sql, "countResample(0, 3, 1)(value, intDiv(unix_milli - 1700000000000 + 1800000 - 1, 1800000)) AS cnts")
+	assert.Contains(t, sql, "sumResample(0, 3, 1)(value, intDiv(unix_milli - 1700000000000 + 1800000 - 1, 1800000)) AS vals")
+	// Single-bucket window: the slide degenerates to reading one slot.
+	assert.Contains(t, sql, "arraySum(arraySlice(cnts, k + 1, 1))")
+	// The sliver predicate is the correctness precondition of this form.
+	assert.Contains(t, sql, "positiveModulo(? - unix_milli, ?) < ?")
+}
+
+// TestDisjointWindowLattice brute-forces the disjoint-form arithmetic: a
+// sample survives the sliver predicate exactly when some grid window
+// contains it, and the ceil bucket index then lands it on that window's
+// slot. This is the pure-Go mirror of the SQL expressions — the predicate
+// in samplesConditions and jj in windowedInner — over random lattices,
+// including off-lattice ends and samples beyond the last grid point.
+func TestDisjointWindowLattice(t *testing.T) {
+	rng := func(seed *uint64) int64 {
+		*seed = *seed*6364136223846793005 + 1442695040888963407
+		return int64(*seed >> 33)
+	}
+	seed := uint64(42)
+	for trial := 0; trial < 2000; trial++ {
+		stepMs := 1_000 * (1 + rng(&seed)%3600)
+		windowMs := 1 + rng(&seed)%(stepMs-1) // strictly below the step
+		selStart := 1_700_000_000_000 + rng(&seed)%1_000_000
+		selEnd := selStart + rng(&seed)%(50*stepMs) // end may sit off-lattice
+		lastIdx := (selEnd - selStart) / stepMs
+		upper := selStart + lastIdx*stepMs
+
+		for i := 0; i < 50; i++ {
+			u := selStart - windowMs - stepMs + rng(&seed)%(selEnd-selStart+3*stepMs)
+
+			// Oracle: is u inside any window (t_k - window, t_k]?
+			inWindow := false
+			var slot int64 = -1
+			for k := int64(0); k <= lastIdx; k++ {
+				tk := selStart + k*stepMs
+				if u > tk-windowMs && u <= tk {
+					inWindow = true
+					slot = k
+					break
+				}
+			}
+
+			// The SQL: fetch bounds, then the sliver predicate
+			// positiveModulo(selStart - u, step) < window.
+			kept := u > selStart-windowMs && u <= upper
+			if kept {
+				pmod := (selStart - u) % stepMs
+				if pmod < 0 {
+					pmod += stepMs
+				}
+				kept = pmod < windowMs
+			}
+
+			require.Equal(t, inWindow, kept,
+				"sliver keep mismatch: u=%d selStart=%d step=%d window=%d", u, selStart, stepMs, windowMs)
+			if !kept {
+				continue
+			}
+			// jj = ceil((u - selStart)/step) via one intDiv; numerator is
+			// positive because u > selStart - window > selStart - step.
+			jj := (u - selStart + stepMs - 1) / stepMs
+			require.Equal(t, slot, jj,
+				"slot mismatch: u=%d selStart=%d step=%d window=%d", u, selStart, stepMs, windowMs)
+		}
+	}
+}
+
+func TestTryExecuteRange_WindowedGateFallsBack(t *testing.T) {
+	c, store := newTestClient(t)
+	e := &executor{client: c, parser: prometheus.NewParser()}
+
+	start := time.UnixMilli(1_700_000_000_000)
+	end := time.UnixMilli(1_700_003_600_000)
+
+	// 10m range at 90s step: the window is not a whole number of buckets.
+	_, ok, err := e.TryExecuteRange(context.Background(), `avg_over_time(up[10m])`, start, end, 90*time.Second)
+	require.NoError(t, err)
+	assert.False(t, ok, "range not divisible by step must not transpile")
+
+	// 1d range at 60s step: 1440 bucket combines per slot, over the cap.
+	_, ok, err = e.TryExecuteRange(context.Background(), `avg_over_time(up[1d])`, start, end, time.Minute)
+	require.NoError(t, err)
+	assert.False(t, ok, "range/step above maxWindowBuckets must not transpile")
+
+	// 1m range at 5m step: the windows are disjoint slivers — no
+	// divisibility or width requirement, so this transpiles.
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("up", int64(1_699_999_200_000), int64(1_700_003_600_000)).WillReturnRows(cmock.NewRows(seriesCols, [][]any{}))
+	_, ok, err = e.TryExecuteRange(context.Background(), `avg_over_time(up[1m])`, start, end, 5*time.Minute)
+	require.NoError(t, err)
+	assert.True(t, ok, "range below step is the disjoint form and must transpile")
+}
+
+func TestApplyScalarOps(t *testing.T) {
+	f := func(v float64) *float64 { return &v }
+
+	t.Run("arithmetic chain", func(t *testing.T) {
+		values := []*float64{f(2), nil, f(4)}
+		applyScalarOps([]scalarOp{{op: parser.MUL, scalar: 100}, {op: parser.ADD, scalar: 1}}, values)
+		require.NotNil(t, values[0])
+		assert.Equal(t, 201.0, *values[0])
+		assert.Nil(t, values[1])
+		assert.Equal(t, 401.0, *values[2])
+	})
+
+	t.Run("comparison filters points", func(t *testing.T) {
+		values := []*float64{f(1), f(10)}
+		applyScalarOps([]scalarOp{{op: parser.GTR, scalar: 5}}, values)
+		assert.Nil(t, values[0])
+		require.NotNil(t, values[1])
+		assert.Equal(t, 10.0, *values[1], "filter comparisons keep the original value")
+	})
+
+	t.Run("bool comparison emits 0/1", func(t *testing.T) {
+		values := []*float64{f(1), f(10)}
+		applyScalarOps([]scalarOp{{op: parser.GTR, scalar: 5, returnBool: true}}, values)
+		assert.Equal(t, 0.0, *values[0])
+		assert.Equal(t, 1.0, *values[1])
+	})
+
+	t.Run("scalar on left division", func(t *testing.T) {
+		values := []*float64{f(4)}
+		applyScalarOps([]scalarOp{{op: parser.DIV, scalar: 100, scalarOnLeft: true}}, values)
+		assert.Equal(t, 25.0, *values[0])
+	})
+}
+
+func TestLabelsFromGroupKey(t *testing.T) {
+	lset, err := labelsFromGroupKey(`[["pod","api-0"],["ns","prod"]]`)
+	require.NoError(t, err)
+	assert.Equal(t, "api-0", lset.Get("pod"))
+	assert.Equal(t, "prod", lset.Get("ns"))
+
+	empty, err := labelsFromGroupKey(`[]`)
+	require.NoError(t, err)
+	assert.True(t, empty.IsEmpty())
+}
+
+// testGrid is a 2h query grid ending on a round timestamp.
+func testGrid(stepMs int64) gridContext {
+	return gridContext{startMs: 1_700_000_000_000, endMs: 1_700_007_200_000, stepMs: stepMs}
+}
+
+// A bool comparison returns 0/1, not the sample, so the engine drops
+// __name__; keeping it would change downstream vector matching.
+func TestKeepsName_BoolComparisonDropsName(t *testing.T) {
+	plan, ok := classify(parse(t, `up > bool 0`), testGrid(60_000))
+	require.True(t, ok)
+	assert.False(t, plan.units[0].core.keepsName())
+
+	plan, ok = classify(parse(t, `up > 0`), testGrid(60_000))
+	require.True(t, ok)
+	assert.True(t, plan.units[0].core.keepsName())
+}
+
+// timeSeriesLastToGrid widens its window to max(window, step) — probed on
+// 25.12 — so Last-style units at window < step must fall back or they would
+// resurrect samples the engine's lookback already dropped.
+func TestTryExecuteRange_LastStyleWindowBelowStepTranspiles(t *testing.T) {
+	// These used to fall back because timeSeriesLastToGrid widens its window
+	// to max(window, step). Over sliver-filtered rows the widening is
+	// harmless — the widened window intersected with the data IS the
+	// lookback window — so the gate is gone and both shapes transpile. The
+	// mock returns no series: the point here is the routing, the value
+	// semantics are the parity suite's job.
+	c, store := newTestClient(t)
+	e := &executor{client: c, parser: prometheus.NewParser()}
+
+	start := time.UnixMilli(1_700_000_000_000)
+	end := time.UnixMilli(1_700_003_600_000)
+
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("up", int64(1_699_999_200_000), int64(1_700_003_600_000)).WillReturnRows(cmock.NewRows(seriesCols, [][]any{}))
+	_, ok, err := e.TryExecuteRange(context.Background(), `sum by (pod) (up)`, start, end, time.Hour)
+	require.NoError(t, err)
+	assert.True(t, ok, "instant selection at step > lookback must transpile")
+
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("up", int64(1_699_999_200_000), int64(1_700_003_600_000)).WillReturnRows(cmock.NewRows(seriesCols, [][]any{}))
+	_, ok, err = e.TryExecuteRange(context.Background(), `last_over_time(up[10m])`, start, end, time.Hour)
+	require.NoError(t, err)
+	assert.True(t, ok, "last_over_time at range < step must transpile")
+}
+
+// Two metrics collapsing onto one labelset after the name drop, with values
+// on the same grid slot, is the engine's duplicate-labelset error; merging
+// them would invent a series no engine would produce. (Temporally disjoint
+// twins merge instead — see TestMergeSameLabelsetSeries.)
+func TestExecuteUnit_NameCollisionErrors(t *testing.T) {
+	c, store := newTestClient(t)
+	e := &executor{client: c, parser: prometheus.NewParser()}
+
+	store.Mock().ExpectQuery("SELECT fingerprint, any\\(labels\\)").WithArgs("^(?:a|b)$", int64(1_699_999_200_000), int64(1_700_003_600_000)).WillReturnRows(cmock.NewRows(seriesCols, [][]any{
+		{uint64(1), `{"__name__":"a","job":"x"}`},
+		{uint64(2), `{"__name__":"b","job":"x"}`},
+	}))
+	store.Mock().ExpectQuery("SELECT gkey").
+		WithArgs("^(?:a|b)$", int64(1_699_999_200_000), int64(1_700_003_600_000), "a", "b", int64(1_699_999_700_000), int64(1_700_003_600_000)).
+		WillReturnRows(cmock.NewRows(gkeyCols, [][]any{
+			{`[["__name__","a"],["job","x"]]`, []*float64{f64(1)}},
+			{`[["__name__","b"],["job","x"]]`, []*float64{f64(2)}},
+		}))
+
+	plan, ok := classify(parse(t, `rate({__name__=~"a|b"}[5m])`), gridContext{startMs: 1_700_000_000_000, endMs: 1_700_003_600_000, stepMs: 60_000})
+	require.True(t, ok)
+
+	_, err := e.executeUnit(context.Background(), &plan.units[0].core, plan.units[0].grid)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "vector cannot contain metrics with the same labelset")
+}
+
+var gkeyCols = []cmock.ColumnType{
+	{Name: "gkey", Type: "String"},
+	{Name: "grid", Type: "Array(Nullable(Float64))"},
+}
+
+func f64(v float64) *float64 { return &v }
+
+// A nameless selector can span metrics whose series alternate in time (one
+// dies inside the lookback before the other appears); after the name drop
+// the engine merges them into ONE series and errors only when two samples
+// share an evaluation timestamp. Pinned by conformance cases
+// operators.test:994/997 (-{job="api"} over http_requests/http_errors).
+func TestMergeSameLabelsetSeries(t *testing.T) {
+	f := func(v float64) *float64 { return &v }
+	api := labels.FromStrings("job", "api")
+
+	out, err := mergeSameLabelsetSeries([]transpiledSeries{
+		{lset: api, values: []*float64{f(-2), nil}},
+		{lset: api, values: []*float64{nil, f(-4)}},
+		{lset: labels.FromStrings("job", "web"), values: []*float64{f(7), nil}},
+	})
+	require.NoError(t, err)
+	require.Len(t, out, 2)
+	assert.Equal(t, []*float64{f(-2), f(-4)}, out[0].values, "temporally disjoint twins must merge into one series")
+
+	_, err = mergeSameLabelsetSeries([]transpiledSeries{
+		{lset: api, values: []*float64{f(1), nil}},
+		{lset: api, values: []*float64{f(2), nil}},
+	})
+	require.Error(t, err, "two values on one evaluation timestamp is the engine's duplicate error")
+	assert.True(t, errors.Ast(err, errors.TypeInvalidInput))
+}
+
+// Hybrid twin case: stripping the synthetic __name__ can leave two engine
+// output series distinguishable only by those names (-metric_a or -metric_b:
+// both {} once real names are dropped). Pinned by conformance cases
+// name_label_dropping.test:137 and operators.test:1016.
+func TestMergeMatrixByLabelset(t *testing.T) {
+	empty := labels.EmptyLabels()
+
+	out, err := mergeMatrixByLabelset(promql.Matrix{
+		{Metric: empty, Floats: []promql.FPoint{{T: 0, F: -1}}},
+		{Metric: empty, Floats: []promql.FPoint{{T: 600_000, F: -4}}},
+	})
+	require.NoError(t, err)
+	require.Len(t, out, 1)
+	assert.Equal(t, []promql.FPoint{{T: 0, F: -1}, {T: 600_000, F: -4}}, out[0].Floats)
+
+	_, err = mergeMatrixByLabelset(promql.Matrix{
+		{Metric: empty, Floats: []promql.FPoint{{T: 0, F: -1}}},
+		{Metric: empty, Floats: []promql.FPoint{{T: 0, F: -3}}},
+	})
+	require.Error(t, err)
+	assert.True(t, errors.Ast(err, errors.TypeInvalidInput))
+}

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

@@ -1,6 +1,9 @@
 package prometheus
 
 import (
+	"context"
+	"time"
+
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/promql/parser"
 	"github.com/prometheus/prometheus/storage"
@@ -35,3 +38,20 @@ 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"
+
+// RangeExecutor is the optional capability of a provider that can evaluate
+// some range queries entirely inside the datastore. ok=false means the query
+// is not evaluable that way and the caller should run the engine over the
+// provider's Storage instead — which is always exact. Only the clickhousev2
+// provider implements it; once that provider is the only one, the capability
+// folds into Prometheus itself and the engine-vs-datastore decision becomes
+// internal.
+type RangeExecutor interface {
+	TryExecuteRange(ctx context.Context, query string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error)
+}

pkg/querier/api.go

@@ -57,6 +57,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,9 +8,12 @@ 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"
@@ -98,6 +101,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 prometheus.Prometheus
+	// 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 prometheus.Prometheus
 }
 
 var _ qbv5.Query = (*promqlQuery)(nil)
@@ -110,6 +131,7 @@ func newPromqlQuery(
 	tr qbv5.TimeRange,
 	requestType qbv5.RequestType,
 	variables map[string]qbv5.VariableItem,
+	opts promqlOptions,
 ) *promqlQuery {
 	return &promqlQuery{
 		logger:      logger,
@@ -119,10 +141,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 +283,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 +336,58 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
 		return nil, err
 	}
 
+	// Attach query traits so the storage can prove step-aligned optimizations
+	// safe (see prometheus.QueryTraits). A parse failure surfaces below via
+	// the engine with the enhanced error message.
+	if expr, parseErr := q.parser.ParseExpr(query); parseErr == nil {
+		ctx = prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
+	}
+
+	// Accumulate ClickHouse-side scan stats across every storage query this
+	// evaluation issues (engine selectors or the compiled executor): progress
+	// options propagate to each ClickHouse query through the context.
+	var statsMu sync.Mutex
+	var rowsScanned, bytesScanned uint64
+	ctx = clickhouse.Context(ctx, clickhouse.WithProgress(func(p *clickhouse.Progress) {
+		statsMu.Lock()
+		rowsScanned += p.Rows
+		bytesScanned += p.Bytes
+		statsMu.Unlock()
+	}))
+
+	began := time.Now()
+
+	// A pinned provider serves directly from it: comparison callers fetch
+	// the default result and the pinned result as two API calls and diff
+	// them.
+	if q.opts.serve != nil {
+		matrix, err := q.serveFromProvider(ctx, query, start, end)
+		if err != nil {
+			if enhanced := tryEnhancePromQLExecError(err); enhanced != nil {
+				return nil, enhanced
+			}
+			return nil, err
+		}
+		return q.toResult(matrix, nil, began, &statsMu, &rowsScanned, &bytesScanned), nil
+	}
+
+	// When the serving provider has the RangeExecutor capability
+	// (prometheus::provider: clickhousev2), serve the way the provider is
+	// designed to serve: transpiled when the shape allows. Without this the
+	// override would silently run the engine path only.
+	if re, ok := q.promEngine.(prometheus.RangeExecutor); ok {
+		matrix, served, err := re.TryExecuteRange(ctx, query, time.Unix(0, start), time.Unix(0, end), q.query.Step.Duration)
+		if err != nil {
+			if enhanced := tryEnhancePromQLExecError(err); enhanced != nil {
+				return nil, enhanced
+			}
+			return nil, err
+		}
+		if served {
+			return q.toResult(matrix, nil, began, &statsMu, &rowsScanned, &bytesScanned), nil
+		}
+	}
+
 	qry, err := q.promEngine.Engine().NewRangeQuery(
 		ctx,
 		q.promEngine.Storage(),
@@ -331,6 +423,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 +486,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 +526,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/prometheustest"
 	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: &prometheustest.Provider{}},
+	}
+	assert.Empty(t, q.Fingerprint())
+}

pkg/querier/promql_shadow.go

@@ -0,0 +1,188 @@
+package querier
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"math"
+	"sort"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/SigNoz/signoz/pkg/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
+)
+
+// shadowTimeout bounds a shadow evaluation; a shadow run must never outlive
+// the request by much or pile up.
+const shadowTimeout = 2 * time.Minute
+
+// runShadowCompare executes the query on the clickhousev2 provider exactly
+// as it would serve (transpiled when the shape allows, engine over the v2
+// querier otherwise), compares against the served result and logs the
+// outcome. Serving is never affected: this runs after the response, off the
+// request context, and only logs. The mismatch and failure logs are the
+// rollout evidence — serving cuts over to v2 only after they stay clean.
+func (q *promqlQuery) runShadowCompare(ctx context.Context, query string, startNs, endNs int64, served promql.Matrix, servedIn time.Duration) {
+	defer func() {
+		if r := recover(); r != nil {
+			q.logger.ErrorContext(ctx, "promql shadow comparison panicked", slog.Any("panic", r), slog.String("query", query))
+		}
+	}()
+
+	ctx, cancel := context.WithTimeout(ctx, shadowTimeout)
+	defer cancel()
+
+	// The request context carries the served response's scan-stats progress
+	// callback; without replacing it the shadow's ClickHouse progress would
+	// race into the served stats. The response itself was already sent.
+	ctx = clickhouse.Context(ctx, clickhouse.WithProgress(func(*clickhouse.Progress) {}))
+
+	if expr, parseErr := q.parser.ParseExpr(query); parseErr == nil {
+		ctx = prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
+	}
+
+	start, end := time.Unix(0, startNs), time.Unix(0, endNs)
+	began := time.Now()
+	shadow, transpiled, err := executeOnProvider(ctx, q.opts.shadow, query, start, end, q.query.Step.Duration)
+	shadowIn := time.Since(began)
+
+	logAttrs := []any{
+		slog.String("query", query),
+		slog.Int64("start_ms", startNs/int64(time.Millisecond)),
+		slog.Int64("end_ms", endNs/int64(time.Millisecond)),
+		slog.Duration("step", q.query.Step.Duration),
+		slog.Bool("transpiled", transpiled),
+		slog.Duration("served_in", servedIn),
+		slog.Duration("shadow_in", shadowIn),
+	}
+
+	if err != nil {
+		// A shadow failure would be a serving failure after rollout; surface
+		// it at the same level as a result mismatch.
+		q.logger.WarnContext(ctx, "promql shadow execution failed", append(logAttrs, slog.Any("error", err))...)
+		return
+	}
+
+	servedNorm := normalizeShadowMatrix(served)
+	shadowNorm := normalizeShadowMatrix(shadow)
+	if diff := diffShadowMatrices(servedNorm, shadowNorm); diff != "" {
+		q.logger.WarnContext(ctx, "promql shadow comparison mismatch", append(logAttrs,
+			slog.String("diff", diff),
+			slog.Int("served_series", len(servedNorm)),
+			slog.Int("shadow_series", len(shadowNorm)),
+		)...)
+		return
+	}
+	// Matches log the timings: served_in vs shadow_in across the fleet is
+	// the perf evidence for the cutover, gathered for free.
+	q.logger.DebugContext(ctx, "promql shadow comparison matched", logAttrs...)
+}
+
+// serveFromProvider evaluates the query the way the pinned provider would
+// serve it.
+func (q *promqlQuery) serveFromProvider(ctx context.Context, query string, startNs, endNs int64) (promql.Matrix, error) {
+	matrix, _, err := executeOnProvider(ctx, q.opts.serve, query, time.Unix(0, startNs), time.Unix(0, endNs), q.query.Step.Duration)
+	return matrix, err
+}
+
+// executeOnProvider evaluates the query the way the provider would serve it:
+// transpiled in the datastore when the provider has the RangeExecutor
+// capability and the shape allows, the engine over the provider's storage
+// otherwise. The returned matrix is an owned copy.
+func executeOnProvider(ctx context.Context, prov prometheus.Prometheus, query string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error) {
+	if re, ok := prov.(prometheus.RangeExecutor); ok {
+		matrix, served, err := re.TryExecuteRange(ctx, query, start, end, step)
+		if err != nil {
+			return nil, true, err
+		}
+		if served {
+			return matrix, true, nil
+		}
+	}
+
+	qry, err := prov.Engine().NewRangeQuery(ctx, prov.Storage(), nil, query, start, end, step)
+	if err != nil {
+		return nil, false, err
+	}
+	defer qry.Close()
+
+	res := qry.Exec(ctx)
+	if res.Err != nil {
+		return nil, false, res.Err
+	}
+	matrix, err := res.Matrix()
+	if err != nil {
+		return nil, false, err
+	}
+	// Close returns the result's sample slices to the engine pool.
+	return copyMatrix(matrix), false, nil
+}
+
+func copyMatrix(matrix promql.Matrix) promql.Matrix {
+	out := make(promql.Matrix, 0, len(matrix))
+	for _, s := range matrix {
+		floats := make([]promql.FPoint, len(s.Floats))
+		copy(floats, s.Floats)
+		out = append(out, promql.Series{Metric: s.Metric.Copy(), Floats: floats})
+	}
+	return out
+}
+
+// normalizeShadowMatrix 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

@@ -24,6 +24,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/statsreporter"
 	"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"
@@ -46,11 +47,19 @@ type Querier interface {
 }
 
 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                   prometheus.Prometheus
 	traceStmtBuilder         qbtypes.StatementBuilder[qbtypes.TraceAggregation]
 	logStmtBuilder           qbtypes.StatementBuilder[qbtypes.LogAggregation]
 	auditStmtBuilder         qbtypes.StatementBuilder[qbtypes.LogAggregation]
@@ -61,8 +70,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(
@@ -70,6 +87,7 @@ func New(
 	telemetryStore telemetrystore.TelemetryStore,
 	metadataStore telemetrytypes.MetadataStore,
 	promEngine prometheus.Prometheus,
+	promV2 prometheus.Prometheus,
 	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
@@ -91,6 +109,7 @@ func New(
 		telemetryStore:           telemetryStore,
 		metadataStore:            metadataStore,
 		promEngine:               promEngine,
+		promV2:                   promV2,
 		traceStmtBuilder:         traceStmtBuilder,
 		logStmtBuilder:           logStmtBuilder,
 		auditStmtBuilder:         auditStmtBuilder,
@@ -103,6 +122,7 @@ func New(
 			logTraceIDWindowPaddingMS: uint64(logTraceIDWindowPadding.Milliseconds()),
 		},
 		maxConcurrentQueries: maxConcurrentQueries,
+		shadowSlots:          make(chan struct{}, maxConcurrentShadows),
 	}
 }
 
@@ -142,7 +162,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
 	}
@@ -185,12 +209,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
@@ -215,7 +268,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:
@@ -858,7 +911,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

@@ -18,6 +18,7 @@ import (
 func NewFactory(
 	telemetryStore telemetrystore.TelemetryStore,
 	prometheus prometheus.Prometheus,
+	promV2 prometheus.Prometheus,
 	metadataStore telemetrytypes.MetadataStore,
 	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
@@ -40,6 +41,7 @@ func NewFactory(
 				telemetryStore,
 				metadataStore,
 				prometheus,
+				promV2,
 				traceStmtBuilder,
 				logStmtBuilder,
 				auditStmtBuilder,

pkg/query-service/rules/managertestfactory.go

@@ -128,7 +128,7 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
 	meterStmtBuilder, err := meterstatementbuilder.NewFactory(metadataStore, flagger).New(ctx, providerSettings, cfg)
 	require.NoError(t, err)
 	bucketCache := querier.NewBucketCache(providerSettings, cache, 0, 0)
-	providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger)
+	providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, nil, metadataStore, traceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger)
 	mockQuerier, err := providerFactory.New(context.Background(), providerSettings, querier.Config{})
 	require.NoError(t, err)
 

pkg/query-service/rules/setups_test.go

@@ -40,6 +40,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		telemetryStore,
 		metadataStore,
 		nil, // prometheus
+		nil, // promV2
 		nil, // traceStmtBuilder
 		nil, // logStmtBuilder
 		nil, // auditStmtBuilder
@@ -74,6 +75,7 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
 		telemetryStore,
 		metadataStore,
 		nil, // prometheus
+		nil, // promV2
 		nil, // traceStmtBuilder
 		logStmtBuilder,
 		nil, // auditStmtBuilder
@@ -109,6 +111,7 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
 		telemetryStore,
 		metadataStore,
 		nil, // prometheus
+		nil, // promV2
 		traceStmtBuilder,
 		nil, // logStmtBuilder
 		nil, // auditStmtBuilder

pkg/querybuilder/clickhouse_sql.go

@@ -3,8 +3,6 @@ package querybuilder
 import (
 	"context"
 	"log/slog"
-	"maps"
-	"slices"
 	"strings"
 
 	chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
@@ -27,23 +25,7 @@ var internalDatabases = map[string]struct{}{
 	"information_schema": {},
 }
 
-// generatorTableFunctions compute their rows from their arguments alone. They open no file or socket, reach no other host, and name no table, database or dictionary, so none of them can read through anything the rules here exist to protect. Can be used to build a dense axis to join a sparse series against. Every other table function is refused.
-//
-// Keyed by the lowercased name so that matching is case-insensitive, valued by the spelling to name it back to the caller.
-//
-// TODO(@therealpandey): take a deployment level allow list on top of this, so an operator can permit more without a release.
-var generatorTableFunctions = map[string]string{
-	"numbers":         "numbers",
-	"numbers_mt":      "numbers_mt",
-	"zeros":           "zeros",
-	"zeros_mt":        "zeros_mt",
-	"generateseries":  "generateSeries",
-	"generate_series": "generate_series",
-}
-
-var generatorTableFunctionsMessage = "allowed table functions are " + strings.Join(slices.Sorted(maps.Values(generatorTableFunctions)), ", ")
-
-// The parser's grammar has gaps against SQL that ClickHouse itself accepts.
+// The parser's grammar has gaps against SQL that ClickHouse itself accepts. See TestErrIfStatementIsNotValid_ShouldPassButFails.
 func ErrIfStatementIsNotValid(query string) (err error) {
 	defer func() {
 		// The parser has a history of panicking on malformed input rather than returning an error.
@@ -70,17 +52,8 @@ func ErrIfStatementIsNotValid(query string) (err error) {
 	visitor := &chparser.DefaultASTVisitor{Visit: func(node chparser.Expr) error {
 		switch expr := node.(type) {
 		case *chparser.TableFunctionExpr:
-			// Source table functions remain usable in ClickHouse read-only mode. Arguments are
-			// visited before this, so a read smuggled into one is already refused by the time
-			// an allowed generator gets here.
-			name := chparser.Format(expr.Name)
-			if _, ok := generatorTableFunctions[strings.ToLower(name)]; ok {
-				return nil
-			}
-
-			return errors.
-				NewInvalidInputf(CodeClickHouseSQLTableFunction, "ClickHouse table functions are not allowed in SQL queries: %s", name).
-				WithAdditional(generatorTableFunctionsMessage)
+			// Source table functions remain usable in ClickHouse read-only mode.
+			return errors.NewInvalidInputf(CodeClickHouseSQLTableFunction, "ClickHouse table functions are not allowed in SQL queries: %s", chparser.Format(expr.Name))
 
 		case *chparser.TableIdentifier:
 			// Reading these is unaffected by ClickHouse read-only mode.

pkg/querybuilder/clickhouse_sql_test.go

@@ -2,11 +2,12 @@ package querybuilder
 
 import (
 	"testing"
-	"time"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 
+	chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestErrIfStatementIsNotValid_Pass(t *testing.T) {
@@ -20,9 +21,6 @@ func TestErrIfStatementIsNotValid_Pass(t *testing.T) {
 		{"CommonTableExpression", "WITH t AS (SELECT fingerprint FROM signoz_metrics.time_series_v4) SELECT * FROM t"},
 		{"Join", "SELECT * FROM t1 LEFT JOIN t2 ON t1.a = t2.b"},
 		{"GlobalIn", "SELECT a FROM t WHERE a GLOBAL IN (SELECT b FROM t2)"},
-		// GLOBAL parsed only when the join type was omitted, and only before IN. https://github.com/AfterShip/clickhouse-sql-parser/pull/293
-		{"GlobalLeftJoin", "SELECT * FROM t1 GLOBAL LEFT JOIN t2 ON t1.a = t2.a"},
-		{"GlobalNotIn", "SELECT a FROM t WHERE a GLOBAL NOT IN (SELECT b FROM t2)"},
 		{"Union", "SELECT * FROM t UNION ALL SELECT * FROM t2"},
 		{"Intersect", "SELECT * FROM t INTERSECT SELECT * FROM t2"},
 		{"WindowFunction", "SELECT sum(v) OVER (PARTITION BY a ORDER BY t) FROM t"},
@@ -38,50 +36,18 @@ func TestErrIfStatementIsNotValid_Pass(t *testing.T) {
 		// order by interval
 		{"OrderByInterval", "SELECT toStartOfInterval(timestamp, INTERVAL 1 MINUTE) AS interval ORDER BY interval"},
 		{"OrderByIntervalAndDirection", "SELECT toStartOfInterval(timestamp, INTERVAL 1 MINUTE) AS `interval` ORDER BY `interval` ASC"},
-		// `interval` is a unit keyword, so unquoting it was rejected everywhere the parser
-		// expected a plain identifier. https://github.com/AfterShip/clickhouse-sql-parser/pull/296
-		{"OrderByUnquotedIntervalAsc", "SELECT toStartOfInterval(timestamp, INTERVAL 1 MINUTE) AS interval FROM t GROUP BY interval ORDER BY interval ASC"},
-		{"OrderByUnquotedIntervalDesc", "SELECT toStartOfInterval(timestamp, INTERVAL 1 MINUTE) AS interval FROM t GROUP BY interval ORDER BY interval DESC"},
-		{"UnquotedIntervalInGroupByTuple", "SELECT a FROM t GROUP BY (`service.name`, `service.version`, interval)"},
-		{"UnquotedIntervalProductionQuery", "SELECT toStartOfInterval(timestamp, INTERVAL 1 MINUTE) AS interval, resource_string_service$$name AS `service.name`, attributes_string['http.route'] AS `http.route`, quantile(0.95)(duration_nano) / 1000000000 AS value FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_string_service$$name = 'svc-a' AND resources_string['deployment.environment'] = 'dev' AND attributes_string['http.route'] = '/v1' AND http_method = 'POST' AND timestamp BETWEEN toDateTime(1784601720) AND toDateTime(1784602620) AND ts_bucket_start BETWEEN 1784601720 - 1800 AND 1784602620 GROUP BY `service.name`, `http.route`, interval ORDER BY interval ASC"},
-		// Separating the two readings of INTERVAL needs backtracking as per the current implementation which could have performance regressions.
-		// https://github.com/AfterShip/clickhouse-sql-parser/pull/296#issuecomment-5150316367
-		{"UnquotedIntervalRepeatedThirtyTimes", "SELECT interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval + interval AS total FROM t WHERE interval > 0 ORDER BY interval ASC"},
+		// Unspaced, so rejected until the parser stopped lexing a signed literal after a
+		// closing bracket. The spaced form above no longer needs to be spaced.
+		// https://github.com/AfterShip/clickhouse-sql-parser/issues/286
 		{"SignedLiteralAfterClosingParenUnspaced", "SELECT now() AS ts, toFloat64(count()) AS value FROM ( SELECT attributes_string['TableName'] AS T, attributes_string['MissingId'] AS M, max(fromUnixTimestamp64Nano(timestamp)) AS last_seen, dateDiff('minute', min(fromUnixTimestamp64Nano(timestamp)), max(fromUnixTimestamp64Nano(timestamp))) AS age_min FROM signoz_logs.distributed_logs_v2 WHERE body='missing_map_record' AND timestamp >= (toUnixTimestamp(now())-3600)*1000000000 GROUP BY T, M ) WHERE age_min >= 20 AND last_seen >= now() - toIntervalMinute(8)"},
 		{"SignedLiteralAfterClosingParenMinimal", "SELECT (1)-1"},
 		{"TrimFunction", "SELECT trimBoth('/api/endpoint/', '/');"},
-		// The SQL-standard keyword-separated argument forms, which took commas only. https://github.com/AfterShip/clickhouse-sql-parser/pull/290
-		{"StandardTrimSyntax", "SELECT trim(BOTH ' ' FROM body) FROM t"},
-		{"StandardSubstringSyntax", "SELECT substring(body FROM 2 FOR 3) FROM t"},
-		{"StandardOverlaySyntax", "SELECT overlay(body PLACING 'x' FROM 2) FROM t"},
-		// Row generators compute their rows from their arguments, so they read through nothing. This is the shape they get used for: a dense interval axis to CROSS JOIN a sparse series against.
-		{"NumbersTableFunction", "SELECT intervals.interval AS interval, active.cluster AS cluster, toFloat64(if(ts_data.has_data = 0, 0, 1)) AS value FROM ( SELECT DISTINCT JSONExtractString(labels, 'k8s.cluster.name') AS cluster FROM signoz_metrics.distributed_time_series_v4 WHERE metric_name = 'my_metric' AND unix_milli >= toUnixTimestamp(now() - INTERVAL 30 DAY) * 1000 HAVING cluster != '' ) AS active CROSS JOIN ( SELECT toStartOfInterval( toDateTime(toUnixTimestamp(now() - INTERVAL 30 MINUTE) + number * 60), INTERVAL 1 MINUTE ) AS interval FROM numbers(31) ) AS intervals LEFT JOIN ( SELECT toStartOfInterval( toDateTime(intDiv(s.unix_milli, 1000)), INTERVAL 1 MINUTE ) AS interval, JSONExtractString(ts.labels, 'k8s.cluster.name') AS cluster, 1 AS has_data FROM signoz_metrics.distributed_samples_v4 s INNER JOIN ( SELECT DISTINCT fingerprint, labels FROM signoz_metrics.distributed_time_series_v4 WHERE metric_name = 'my_metric' ) AS ts ON s.fingerprint = ts.fingerprint WHERE s.metric_name = 'my_metric' AND s.unix_milli >= toUnixTimestamp(now() - INTERVAL 30 MINUTE) * 1000 GROUP BY interval, cluster ) AS ts_data ON active.cluster = ts_data.cluster AND intervals.interval = ts_data.interval ORDER BY interval ASC"},
-		{"NumbersMtTableFunction", "SELECT * FROM numbers_mt(31)"},
-		{"ZerosTableFunction", "SELECT * FROM zeros(31)"},
-		{"ZerosMtTableFunction", "SELECT * FROM zeros_mt(31)"},
-		{"GenerateSeriesTableFunction", "SELECT * FROM generateSeries(1, 10)"},
-		{"GenerateSeriesSnakeCaseTableFunction", "SELECT * FROM generate_series(1, 10)"},
-		{"GeneratorTableFunctionUppercase", "SELECT * FROM NUMBERS(31)"},
-		{"GeneratorTableFunctionParenthesisedArgument", "SELECT * FROM NUMBERS((31))"},
-		{"GeneratorTableFunctionInJoin", "SELECT * FROM signoz_logs.distributed_logs_v2 AS l CROSS JOIN numbers(31) AS n"},
-		{"GeneratorTableFunctionInCommonTableExpression", "WITH axis AS (SELECT number FROM numbers(31)) SELECT * FROM axis"},
-		{"GeneratorTableFunctionInWhereSubquery", "SELECT * FROM t WHERE a IN (SELECT number FROM numbers(31))"},
-		{"GeneratorTableFunctionInUnion", "SELECT number FROM numbers(31) UNION ALL SELECT number FROM zeros(31)"},
 	}
 
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			// Bounded rather than called directly: a parser that backtracks without memoising
-			// hangs instead of returning. Every case here parses in well under a millisecond.
-			errC := make(chan error, 1)
-			go func() { errC <- ErrIfStatementIsNotValid(testCase.query) }()
-
-			select {
-			case err := <-errC:
-				assert.NoError(t, err)
-			case <-time.After(10 * time.Second):
-				assert.Fail(t, "timed out, which means the parser is no longer bounding its backtracking")
-			}
+			err := ErrIfStatementIsNotValid(testCase.query)
+			assert.NoError(t, err)
 		})
 	}
 }
@@ -104,8 +70,6 @@ func TestErrIfStatementIsNotValid_Fail(t *testing.T) {
 		{"CreateTable", "CREATE TABLE evil (a Int) ENGINE = Memory", CodeClickHouseSQLNotSelect},
 		{"Grant", "GRANT ALL ON *.* TO admin", CodeClickHouseSQLNotSelect},
 		{"Set", "SET readonly = 0", CodeClickHouseSQLNotSelect},
-		// The parser still dereferences nil on a DEFAULT expression it cannot read, so the recover is what turns this into a rejection rather than a crash.
-		{"UnparseableDefaultExpression", "CREATE TABLE t (a String DEFAULT foo(b FROM 2)) ENGINE = Memory", CodeClickHouseSQLParserPanic},
 		// These the parser rejects outright rather than classifying.
 		{"ShowGrants", "SHOW GRANTS", CodeClickHouseSQLUnparseable},
 		{"IntoOutfile", "SELECT * FROM t INTO OUTFILE '/tmp/x.csv'", CodeClickHouseSQLUnparseable},
@@ -117,20 +81,6 @@ func TestErrIfStatementIsNotValid_Fail(t *testing.T) {
 		{"TableFunctionInCommonTableExpression", "WITH c AS (SELECT * FROM url('http://x', CSV, 'a String')) SELECT * FROM c", CodeClickHouseSQLTableFunction},
 		{"TableFunctionInWhereSubquery", "SELECT * FROM t WHERE a IN (SELECT * FROM file('/etc/passwd', CSV, 'a String'))", CodeClickHouseSQLTableFunction},
 		{"TableFunctionInUnion", "SELECT * FROM t UNION ALL SELECT * FROM url('http://x', CSV, 'a String')", CodeClickHouseSQLTableFunction},
-		// These reach the internal databases without ever naming one, so the table-function rule is the only thing that sees them.
-		{"MergeTableFunction", "SELECT * FROM merge('system', '.*')", CodeClickHouseSQLTableFunction},
-		{"RemoteTableFunction", "SELECT * FROM remote('other-host', 'system.users')", CodeClickHouseSQLTableFunction},
-		{"ClusterTableFunction", "SELECT * FROM cluster('c', 'system.users')", CodeClickHouseSQLTableFunction},
-		// Pure, but excluded: generateRandom streams rows the arguments do not bound, and values has no use here that an array literal does not already cover.
-		{"GenerateRandomTableFunction", "SELECT * FROM generateRandom('a UInt64')", CodeClickHouseSQLTableFunction},
-		{"ValuesTableFunction", "SELECT * FROM values('a UInt64', 1, 2)", CodeClickHouseSQLTableFunction},
-		// Arguments are visited before the table function itself, so allowing a generator does not give anyone a wrapper to smuggle a read through.
-		{"InternalDatabaseInsideAllowedTableFunction", "SELECT * FROM numbers((SELECT count() FROM system.users))", CodeClickHouseSQLInternalDatabase},
-		{"InternalDatabaseJoinedOntoAllowedTableFunction", "SELECT * FROM numbers(31) AS n JOIN system.users AS u ON 1 = 1", CodeClickHouseSQLInternalDatabase},
-		{"InternalDatabaseUnionedWithAllowedTableFunction", "SELECT number FROM numbers(31) UNION ALL SELECT name FROM system.users", CodeClickHouseSQLInternalDatabase},
-		{"RefusedTableFunctionJoinedOntoAllowedTableFunction", "SELECT * FROM numbers(31) AS n JOIN url('http://x', CSV, 'a String') AS u ON 1 = 1", CodeClickHouseSQLTableFunction},
-		{"RefusedTableFunctionInsideAllowedTableFunction", "SELECT * FROM numbers((SELECT count() FROM file('/etc/passwd', CSV, 'a String')))", CodeClickHouseSQLTableFunction},
-		{"InternalDatabaseInsideAllowedTableFunctionCommonTableExpression", "WITH axis AS (SELECT * FROM numbers((SELECT count() FROM system.users))) SELECT * FROM axis", CodeClickHouseSQLInternalDatabase},
 		// Internal databases, which hold grants and server metadata rather than telemetry.
 		{"SystemUsers", "SELECT * FROM system.users", CodeClickHouseSQLInternalDatabase},
 		{"SystemUppercase", "SELECT * FROM SYSTEM.USERS", CodeClickHouseSQLInternalDatabase},
@@ -153,3 +103,50 @@ func TestErrIfStatementIsNotValid_Fail(t *testing.T) {
 		})
 	}
 }
+
+// Queries the parser cannot read. ClickHouse runs all of them.
+func TestErrIfStatementIsNotValid_ShouldPassButFails(t *testing.T) {
+	testCases := []struct {
+		name  string
+		query string
+		// The construct the parser stops after, which is the one it cannot read.
+		expectedStopsAfter string
+		// The same construct written so the parser accepts it.
+		fix string
+	}{
+		{
+			name:               "IntervalAliasInOrderBy",
+			query:              "SELECT toStartOfInterval(timestamp, INTERVAL 1 MINUTE) AS interval, resource_string_service$$name AS `service.name`, attributes_string['http.route'] AS `http.route`, quantile(0.95)(duration_nano) / 1000000000 AS value FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_string_service$$name = 'svc-a' AND resources_string['deployment.environment'] = 'dev' AND attributes_string['http.route'] = '/v1' AND http_method = 'POST' AND timestamp BETWEEN toDateTime(1784601720) AND toDateTime(1784602620) AND ts_bucket_start BETWEEN 1784601720 - 1800 AND 1784602620 GROUP BY `service.name`, `http.route`, interval ORDER BY interval ASC",
+			expectedStopsAfter: "ORDER BY interval ASC",
+			fix:                "SELECT count() AS interval FROM t ORDER BY `interval` ASC",
+		},
+		{
+			name:               "IntervalAliasInOrderByDesc",
+			query:              "SELECT count() AS value, toStartOfInterval(timestamp, INTERVAL 1 MINUTE) AS interval, serviceName, resourceTagsMap['deployment.environment'] AS environment, exceptionStacktrace FROM signoz_traces.distributed_signoz_error_index_v2 WHERE exceptionType != 'OSError' AND resourceTagsMap['deployment.environment'] = 'staging' AND timestamp BETWEEN toDateTime(1785186300) AND toDateTime(1785186600) GROUP BY serviceName, interval, environment, exceptionStacktrace ORDER BY interval DESC",
+			expectedStopsAfter: "ORDER BY interval DESC",
+			fix:                "SELECT count() AS interval FROM t ORDER BY `interval` DESC",
+		},
+		{
+			name:               "StandardTrimSyntax",
+			query:              "SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 5 MINUTE) AS interval, resources_string['host.name'] as host_name, toFloat64(countIf( lower(trim(BOTH ' ' FROM replaceOne( JSONExtractString(body, 'Action'), 'health_status: ', '' ))) IN ('unhealthy','starting','failing') )) as value FROM signoz_logs.distributed_logs_v2 WHERE timestamp BETWEEN 1784602320000000000 AND 1784602620000000000 AND ts_bucket_start BETWEEN 1784602320 - 300 AND 1784602620 AND JSONExtractString(body, 'Type') = 'container' AND JSONExtractString(body, 'Actor', 'Attributes', 'name') IS NOT NULL AND resources_string['host.name'] IS NOT NULL AND resources_string['host.name'] = 'aihub-nightly' GROUP BY interval, host_name ORDER BY interval, host_name",
+			expectedStopsAfter: "trim(BOTH '",
+			fix:                "SELECT trimBoth(replaceOne( JSONExtractString(body, 'Action'), 'health_status: ', '' ), ' ')",
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			err := ErrIfStatementIsNotValid(testCase.query)
+
+			var parseErr *chparser.ParseError
+			require.ErrorAs(t, err, &parseErr, "expected a parser failure rather than a rule violation")
+
+			// The parser reports the offset it stopped at, which sits just past the construct
+			// it choked on, so the text leading up to it is what needs looking at.
+			consumed := testCase.query[:parseErr.Pos]
+			assert.Equal(t, testCase.expectedStopsAfter, consumed[max(0, len(consumed)-len(testCase.expectedStopsAfter)):])
+
+			assert.NoError(t, ErrIfStatementIsNotValid(testCase.fix))
+		})
+	}
+}

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"
@@ -248,6 +249,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),
 	)
 }
 
@@ -289,9 +291,9 @@ func NewStatsReporterProviderFactories(aggregator statsreporter.Aggregator, orgG
 	)
 }
 
-func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
+func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, promV2 prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
 	return factory.MustNewNamedMap(
-		signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
+		signozquerier.NewFactory(telemetryStore, prometheus, promV2, metadataStore, traceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, 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"
@@ -299,6 +300,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 prometheus.Prometheus
+
 	// Initialize prometheus from the available prometheus provider factories
 	prometheus, err := factory.NewProviderFromNamedMap(
 		ctx,
@@ -311,6 +317,23 @@ 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
+		}
+	}
+
 	// Assemble the query stack (metadata store, statement builders, bucket cache) once,
 	// and reuse the single metadata store everywhere downstream.
 	telemetryMetadataStore, traceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, err := newQueryStack(ctx, providerSettings, config, telemetrystore, cache, flagger)
@@ -323,7 +346,7 @@ func New(
 		ctx,
 		providerSettings,
 		config.Querier,
-		NewQuerierProviderFactories(telemetrystore, prometheus, telemetryMetadataStore, traceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
+		NewQuerierProviderFactories(telemetrystore, prometheus, promV2, telemetryMetadataStore, traceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, 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"`
 }
 

scripts/promqltestcorpus/go.mod

@@ -76,14 +76,14 @@ require (
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/goleak v1.3.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
-	golang.org/x/crypto v0.52.0 // indirect
+	golang.org/x/crypto v0.50.0 // indirect
 	golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa // indirect
-	golang.org/x/net v0.55.0 // indirect
+	golang.org/x/net v0.53.0 // indirect
 	golang.org/x/oauth2 v0.36.0 // indirect
 	golang.org/x/sync v0.20.0 // indirect
-	golang.org/x/sys v0.45.0 // indirect
-	golang.org/x/term v0.43.0 // indirect
-	golang.org/x/text v0.37.0 // indirect
+	golang.org/x/sys v0.43.0 // indirect
+	golang.org/x/term v0.42.0 // indirect
+	golang.org/x/text v0.36.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
 	google.golang.org/api v0.272.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect

scripts/promqltestcorpus/go.sum

@@ -377,29 +377,29 @@ go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
 go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
-golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=
-golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc=
+golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
+golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
 golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa h1:Zt3DZoOFFYkKhDT3v7Lm9FDMEV06GpzjG2jrqW+QTE0=
 golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa/go.mod h1:K79w1Vqn7PoiZn+TkNpx3BUWUQksGO3JcVX6qIjytmA=
-golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM=
-golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU=
-golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
-golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
+golang.org/x/mod v0.34.0 h1:xIHgNUUnW6sYkcM5Jleh05DvLOtwc6RitGHbDk4akRI=
+golang.org/x/mod v0.34.0/go.mod h1:ykgH52iCZe79kzLLMhyCUzhMci+nQj+0XkbXpNYtVjY=
+golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
+golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
 golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
 golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
 golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
 golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
-golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4=
-golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk=
-golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
-golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
+golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
+golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/term v0.42.0 h1:UiKe+zDFmJobeJ5ggPwOshJIVt6/Ft0rcfrXZDLWAWY=
+golang.org/x/term v0.42.0/go.mod h1:Dq/D+snpsbazcBG5+F9Q1n2rXV8Ma+71xEjTRufARgY=
+golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
+golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
 golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
-golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c=
-golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI=
+golang.org/x/tools v0.43.0 h1:12BdW9CeB3Z+J/I/wj34VMl8X+fEXBxVR90JeMX5E7s=
+golang.org/x/tools v0.43.0/go.mod h1:uHkMso649BX2cZK6+RpuIPXS3ho2hZo4FVwfoy1vIk0=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
 google.golang.org/api v0.272.0 h1:eLUQZGnAS3OHn31URRf9sAmRk3w2JjMx37d2k8AjJmA=

tests/conftest.py

@@ -1,5 +1,3 @@
-import subprocess
-
 import pytest
 
 pytest_plugins = [
@@ -34,22 +32,6 @@ pytest_plugins = [
 ]
 
 
-def pytest_configure(config: pytest.Config):
-    if config.getoption("--rebuild"):
-        if not config.getoption("--reuse"):
-            raise pytest.UsageError("--rebuild requires --reuse: it replaces the signoz container within an environment that is being reused.")
-        if config.getoption("--teardown"):
-            raise pytest.UsageError("--rebuild cannot be combined with --teardown.")
-        if config.getoption("--clean"):
-            raise pytest.UsageError("--rebuild cannot be combined with --clean: --clean forces a cold build, which defeats the purpose of --rebuild.")
-
-
-def pytest_sessionstart(session: pytest.Session):
-    if session.config.getoption("--clean"):
-        # The type filter removes only cache mounts, leaving images and layer cache intact.
-        subprocess.run(["docker", "builder", "prune", "--force", "--filter", "type=exec.cachemount"], check=True)
-
-
 def pytest_addoption(parser: pytest.Parser):
     parser.addoption(
         "--reuse",
@@ -63,18 +45,6 @@ def pytest_addoption(parser: pytest.Parser):
         default=False,
         help="Teardown environment. Run pytest --basetemp=./tmp/ -vv --teardown src/bootstrap/setup::test_teardown to teardown your local dev environment.",
     )
-    parser.addoption(
-        "--rebuild",
-        action="store_true",
-        default=False,
-        help="Rebuild the signoz container from the current sources while reusing the rest of the stack (databases, mocks, migrations). Only meaningful together with --reuse: pytest --basetemp=./tmp/ -vv --reuse --rebuild integration/bootstrap/setup.py::test_setup.",
-    )
-    parser.addoption(
-        "--clean",
-        action="store_true",
-        default=False,
-        help="Prune the BuildKit cache mounts (go build and module caches) used by the signoz image build, forcing the next build to start cold. Combine with --teardown to reset everything: pytest --basetemp=./tmp/ -vv --teardown --clean integration/bootstrap/setup.py::test_teardown.",
-    )
     parser.addoption(
         "--with-web",
         action="store_true",

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/fixtures/reuse.py

@@ -41,7 +41,6 @@ def wrap(  # pylint: disable=too-many-arguments,too-many-positional-arguments
     create: Callable[[], T],
     delete: Callable[[T], None],
     restore: Callable[[dict], T],
-    rebuild: bool = False,
 ) -> T:
     """
     Wraps a resource creation and cleanup process with reuse and teardown options.
@@ -52,7 +51,6 @@ def wrap(  # pylint: disable=too-many-arguments,too-many-positional-arguments
     - create: function to create the resource
     - delete: function to delete the resource
     - restore: function to restore resource from cache
-    - rebuild: under --reuse, delete the cached resource and recreate it instead of restoring it
     """
     resource = empty()
 
@@ -60,13 +58,8 @@ def wrap(  # pylint: disable=too-many-arguments,too-many-positional-arguments
         existing_resource = pytestconfig.cache.get(key, None)
         if existing_resource:
             assert isinstance(existing_resource, dict)
-            if rebuild:
-                logger.info("Rebuilding %s(%s), removing the existing one", key, existing_resource)
-                delete(restore(existing_resource))
-                pytestconfig.cache.set(key, None)
-            else:
-                logger.info("Reusing existing %s(%s)", key, existing_resource)
-                return restore(existing_resource)
+            logger.info("Reusing existing %s(%s)", key, existing_resource)
+            return restore(existing_resource)
 
     if not teardown(request):
         resource = create()

tests/fixtures/signoz.py

@@ -1,6 +1,4 @@
-import os
 import platform
-import subprocess
 import time
 from http import HTTPStatus
 from os import path
@@ -10,6 +8,7 @@ import docker.errors
 import pytest
 import requests
 from testcontainers.core.container import DockerContainer, Network
+from testcontainers.core.image import DockerImage
 
 from fixtures import reuse, types
 from fixtures.logger import setup_logger
@@ -51,28 +50,18 @@ def create_signoz(
 
         # Docker build context is the repo root — one up from pytest's
         # rootdir (tests/).
-        context = pytestconfig.rootpath.parent
-
-        # The docker CLI is required: the Dockerfiles use BuildKit cache
-        # mounts, which docker-py does not support.
-        subprocess.run(
-            [
-                "docker",
-                "build",
-                "--file",
-                str(context / dockerfile_path),
-                "--tag",
-                "signoz:integration",
-                "--build-arg",
-                f"TARGETARCH={arch}",
-                "--build-arg",
-                f"ZEUSURL={zeus.container_configs['8080'].base()}",
-                str(context),
-            ],
-            check=True,
-            env=os.environ | {"DOCKER_BUILDKIT": "1"},
+        self = DockerImage(
+            path=str(pytestconfig.rootpath.parent),
+            dockerfile_path=dockerfile_path,
+            tag="signoz:integration",
+            buildargs={
+                "TARGETARCH": arch,
+                "ZEUSURL": zeus.container_configs["8080"].base(),
+            },
         )
 
+        self.build()
+
         env = (
             {
                 "SIGNOZ_WEB_ENABLED": False,
@@ -211,7 +200,6 @@ def create_signoz(
         create=create,
         delete=delete,
         restore=restore,
-        rebuild=pytestconfig.getoption("--rebuild"),
     )
 
 

tests/integration/testdata/promqltestcorpus/known_divergences_v2.json

@@ -0,0 +1,17 @@
+{
+  "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. Current class: the engine aggregates floats with Kahan compensated summation (sum, sum_over_time) and an overflow-free incremental mean (avg); ClickHouse's sumForEach/avgForEach/arraySum are naive, so extreme-magnitude corpus data (±1e100 cancellation, ±1.8e308 overflow) diverges on transpiled plans. Burn-down candidates: sumKahanForEach for the cancellation class; the overflow class needs an incremental-mean aggregate ClickHouse does not have.",
+  "divergences": {
+    "aggregators.test:651[base]": "avg over near-max-float64 values: engine's incremental mean never forms the overflowing sum; avgForEach sums then divides, overflowing to +Inf",
+    "aggregators.test:651[instant-coarse]": "same as aggregators.test:651[base] on the coarse-step grid variant",
+    "aggregators.test:654[base]": "avg over near-min-float64 values: engine's incremental mean never forms the overflowing sum; avgForEach overflows to -Inf",
+    "aggregators.test:654[instant-coarse]": "same as aggregators.test:654[base] on the coarse-step grid variant",
+    "aggregators.test:687[base]": "sum over {1e100, -1e100, small}: engine uses Kahan compensated summation; sumForEach's naive summation loses the small terms to cancellation and returns 0",
+    "aggregators.test:687[instant-coarse]": "same as aggregators.test:687[base] on the coarse-step grid variant",
+    "aggregators.test:695[base]": "avg over {1e100, -1e100, small}: same Kahan-vs-naive cancellation as aggregators.test:687, divided by count",
+    "aggregators.test:695[instant-coarse]": "same as aggregators.test:695[base] on the coarse-step grid variant",
+    "functions.test:1084[instant-coarse]": "sum_over_time over a window containing ±1e100: the disjoint coarse-step form's arraySum slide is naive summation, cancelling to 0 (the base variant's W>64 shape falls back to the engine and is exact)",
+    "functions.test:1087[instant-coarse]": "avg_over_time, same window and cancellation as functions.test:1084[instant-coarse]",
+    "functions.test:1149[base]": "avg_over_time over ±2.258e220-magnitude samples: engine's Kahan-compensated incremental mean cancels exactly to 0; the bucketed form's naive slide summation leaves a ~1e202 residue",
+    "functions.test:1149[instant-coarse]": "same as functions.test:1149[base] through the disjoint coarse-step form"
+  }
+}

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