Merge branch 'issue-5535' of github.com:SigNoz/signoz into issue-5535

.github/workflows/integrationci.yaml

@@ -60,6 +60,8 @@ jobs:
           - querier_json_body
           - querier_skip_resource_fingerprint
           - ttl
+          - clickhousecluster
+          - metricreduction
         sqlstore-provider:
           - postgres
           - sqlite

frontend/__mocks__/motionMock.tsx

@@ -1,88 +0,0 @@
-// oxlint-disable-next-line no-restricted-imports
-import * as React from 'react';
-
-// In jsdom, AnimatePresence from motion/react keeps children in DOM during exit
-// animations (awaiting rAF-driven completion that never fully runs in jsdom).
-// This mock makes AnimatePresence render children immediately and makes motion.*
-// elements render as their plain HTML equivalents without animation side-effects.
-//
-// IMPORTANT: motion component references are cached so React sees a stable
-// component identity across re-renders and does not enter an infinite remount loop.
-
-const MOTION_PROPS_TO_STRIP = new Set([
-	'initial',
-	'animate',
-	'exit',
-	'variants',
-	'transition',
-	'whileHover',
-	'whileTap',
-	'whileFocus',
-	'whileInView',
-	'layout',
-	'layoutId',
-	'onAnimationStart',
-	'onAnimationComplete',
-]);
-
-const cache = new Map<string, React.ComponentType>();
-
-function getMotionComponent(tag: string): React.ComponentType {
-	if (!cache.has(tag)) {
-		const Component = React.forwardRef<HTMLElement, Record<string, unknown>>(
-			(props, ref) => {
-				const domProps: Record<string, unknown> = {};
-				for (const [k, v] of Object.entries(props)) {
-					if (!MOTION_PROPS_TO_STRIP.has(k)) {
-						domProps[k] = v;
-					}
-				}
-				return React.createElement(tag, { ...domProps, ref });
-			},
-		);
-		Component.displayName = `motion.${tag}`;
-		cache.set(tag, Component as unknown as React.ComponentType);
-	}
-	return cache.get(tag) as React.ComponentType;
-}
-
-const motionHandler: ProxyHandler<Record<string, React.ComponentType>> = {
-	get(_target, prop: string) {
-		return getMotionComponent(prop);
-	},
-};
-
-export const AnimatePresence: React.FC<{
-	children?: React.ReactNode;
-	mode?: string;
-}> = ({ children }) => React.createElement(React.Fragment, null, children);
-
-export const motion = new Proxy(
-	{} as Record<string, React.ComponentType>,
-	motionHandler,
-);
-
-export const useAnimation = (): Record<string, unknown> => ({
-	start: (): unknown => Promise.resolve(),
-	stop: (): unknown => undefined,
-	set: (): unknown => undefined,
-});
-
-export const useMotionValue = (
-	initial: unknown,
-): { get: () => unknown; set: () => void } => ({
-	get: (): unknown => initial,
-	set: (): unknown => undefined,
-});
-
-export const useTransform = (): { get: () => number } => ({
-	get: (): number => 0,
-});
-
-export const useSpring = (v: unknown): unknown => v;
-
-export const useScroll = (): { scrollY: { get: () => number } } => ({
-	scrollY: { get: (): number => 0 },
-});
-
-export default { motion, AnimatePresence };

frontend/jest.config.ts

@@ -20,7 +20,6 @@ const config: Config.InitialOptions = {
 		'\\.module\\.mjs$': '<rootDir>/__mocks__/cssMock.ts',
 		'\\.md$': '<rootDir>/__mocks__/cssMock.ts',
 		'^uplot$': '<rootDir>/__mocks__/uplotMock.ts',
-		'^motion/react$': '<rootDir>/__mocks__/motionMock.tsx',
 		'^@signozhq/resizable$': '<rootDir>/__mocks__/resizableMock.tsx',
 		'^hooks/useSafeNavigate$': USE_SAFE_NAVIGATE_MOCK_PATH,
 		'^src/hooks/useSafeNavigate$': USE_SAFE_NAVIGATE_MOCK_PATH,

frontend/src/components/CreateServiceAccountModal/CreateServiceAccountModal.tsx

@@ -92,7 +92,6 @@ function CreateServiceAccountModal(): JSX.Element {
 			width="narrow"
 			className="create-sa-modal"
 			disableOutsideClick={isErrorModalVisible}
-			testId="create-service-account-modal"
 		>
 			<div className="create-sa-modal__content">
 				<form

frontend/src/components/CreateServiceAccountModal/__tests__/CreateServiceAccountModal.test.tsx

@@ -1,7 +1,13 @@
 import { toast } from '@signozhq/ui/sonner';
 import { rest, server } from 'mocks-server/server';
 import { NuqsTestingAdapter } from 'nuqs/adapters/testing';
-import { render, screen, userEvent, waitFor } from 'tests/test-utils';
+import {
+	render,
+	screen,
+	userEvent,
+	waitFor,
+	waitForElementToBeRemoved,
+} from 'tests/test-utils';
 
 import CreateServiceAccountModal from '../CreateServiceAccountModal';
 
@@ -83,7 +89,7 @@ describe('CreateServiceAccountModal', () => {
 
 		await waitFor(() => {
 			expect(
-				screen.queryByTestId('create-service-account-modal'),
+				screen.queryByRole('dialog', { name: /New Service Account/i }),
 			).not.toBeInTheDocument();
 		});
 	});
@@ -123,7 +129,7 @@ describe('CreateServiceAccountModal', () => {
 		});
 
 		expect(
-			screen.getByTestId('create-service-account-modal'),
+			screen.getByRole('dialog', { name: /New Service Account/i }),
 		).toBeInTheDocument();
 	});
 
@@ -131,14 +137,15 @@ describe('CreateServiceAccountModal', () => {
 		const user = userEvent.setup({ pointerEventsCheck: 0 });
 		renderModal();
 
-		await screen.findByTestId('create-service-account-modal');
+		const dialog = await screen.findByRole('dialog', {
+			name: /New Service Account/i,
+		});
 		await user.click(screen.getByRole('button', { name: /Cancel/i }));
 
-		await waitFor(() => {
-			expect(
-				screen.queryByTestId('create-service-account-modal'),
-			).not.toBeInTheDocument();
-		});
+		await waitForElementToBeRemoved(dialog);
+		expect(
+			screen.queryByRole('dialog', { name: /New Service Account/i }),
+		).not.toBeInTheDocument();
 	});
 
 	it('shows "Name is required" after clearing the name field', async () => {

frontend/src/components/ServiceAccountDrawer/AddKeyModal/index.tsx

@@ -151,7 +151,6 @@ function AddKeyModal(): JSX.Element {
 			className="add-key-modal"
 			showCloseButton
 			disableOutsideClick={isErrorModalVisible}
-			testId="add-key-modal"
 		>
 			{phase === Phase.FORM && (
 				<KeyFormPhase

frontend/src/components/ServiceAccountDrawer/DeleteAccountModal.tsx

@@ -91,7 +91,6 @@ function DeleteAccountModal(): JSX.Element {
 				color="destructive"
 				loading={isDeleting}
 				onClick={handleConfirm}
-				data-testid="confirm-delete-btn"
 			>
 				<Trash2 size={12} />
 				Delete
@@ -112,7 +111,6 @@ function DeleteAccountModal(): JSX.Element {
 			className="alert-dialog sa-delete-dialog"
 			showCloseButton={false}
 			disableOutsideClick={isErrorModalVisible}
-			testId="delete-service-account-modal"
 			footer={footer}
 		>
 			{content}

frontend/src/components/ServiceAccountDrawer/EditKeyModal/index.tsx

@@ -175,7 +175,6 @@ function EditKeyModal({ keyItem }: EditKeyModalProps): JSX.Element {
 			}
 			showCloseButton={!isRevokeConfirmOpen}
 			disableOutsideClick={isErrorModalVisible}
-			testId="edit-key-modal"
 			footer={
 				isRevokeConfirmOpen ? (
 					<RevokeKeyFooter

frontend/src/components/ServiceAccountDrawer/__tests__/AddKeyModal.test.tsx

@@ -2,7 +2,13 @@ import { toast } from '@signozhq/ui/sonner';
 import { setupAuthzAdmin } from 'lib/authz/utils/authz-test-utils';
 import { rest, server } from 'mocks-server/server';
 import { NuqsTestingAdapter } from 'nuqs/adapters/testing';
-import { render, screen, userEvent, waitFor } from 'tests/test-utils';
+import {
+	render,
+	screen,
+	userEvent,
+	waitFor,
+	waitForElementToBeRemoved,
+} from 'tests/test-utils';
 
 import AddKeyModal from '../AddKeyModal';
 
@@ -91,7 +97,7 @@ describe('AddKeyModal', () => {
 
 		await screen.findByText('snz_abc123xyz456secret');
 		expect(screen.getByText(/Store the key securely/i)).toBeInTheDocument();
-		expect(screen.getByTestId('add-key-modal')).toBeInTheDocument();
+		await screen.findByRole('dialog', { name: /Key Created Successfully/i });
 	});
 
 	it('copy button writes key to clipboard and shows toast.success', async () => {
@@ -127,11 +133,9 @@ describe('AddKeyModal', () => {
 		const user = userEvent.setup({ pointerEventsCheck: 0 });
 		renderModal();
 
-		await screen.findByTestId('add-key-modal');
+		const dialog = await screen.findByRole('dialog', { name: /Add a New Key/i });
 		await user.click(screen.getByRole('button', { name: /Cancel/i }));
 
-		await waitFor(() => {
-			expect(screen.queryByTestId('add-key-modal')).not.toBeInTheDocument();
-		});
+		await waitForElementToBeRemoved(dialog);
 	});
 });

frontend/src/components/ServiceAccountDrawer/__tests__/EditKeyModal.test.tsx

@@ -73,7 +73,9 @@ describe('EditKeyModal (URL-controlled)', () => {
 	it('renders nothing when edit-key param is absent', () => {
 		renderModal(null, { account: 'sa-1' });
 
-		expect(screen.queryByTestId('edit-key-modal')).not.toBeInTheDocument();
+		expect(
+			screen.queryByRole('dialog', { name: /Edit Key Details/i }),
+		).not.toBeInTheDocument();
 	});
 
 	it('renders key data from prop when edit-key param is set', async () => {
@@ -100,7 +102,9 @@ describe('EditKeyModal (URL-controlled)', () => {
 		});
 
 		await waitFor(() => {
-			expect(screen.queryByTestId('edit-key-modal')).not.toBeInTheDocument();
+			expect(
+				screen.queryByRole('dialog', { name: /Edit Key Details/i }),
+			).not.toBeInTheDocument();
 		});
 	});
 
@@ -127,7 +131,9 @@ describe('EditKeyModal (URL-controlled)', () => {
 		expect(latestUrlUpdate.queryString).not.toContain('edit-key=');
 
 		await waitFor(() => {
-			expect(screen.queryByTestId('edit-key-modal')).not.toBeInTheDocument();
+			expect(
+				screen.queryByRole('dialog', { name: /Edit Key Details/i }),
+			).not.toBeInTheDocument();
 		});
 	});
 
@@ -139,7 +145,9 @@ describe('EditKeyModal (URL-controlled)', () => {
 		await user.click(screen.getByRole('button', { name: /Revoke Key/i }));
 
 		// Same dialog, now showing revoke confirmation
-		expect(screen.getByTestId('edit-key-modal')).toBeInTheDocument();
+		await expect(
+			screen.findByRole('dialog', { name: /Revoke Original Key Name/i }),
+		).resolves.toBeInTheDocument();
 		expect(
 			screen.getByText(/Revoking this key will permanently invalidate it/i),
 		).toBeInTheDocument();
@@ -162,7 +170,9 @@ describe('EditKeyModal (URL-controlled)', () => {
 		});
 
 		await waitFor(() => {
-			expect(screen.queryByTestId('edit-key-modal')).not.toBeInTheDocument();
+			expect(
+				screen.queryByRole('dialog', { name: /Edit Key Details/i }),
+			).not.toBeInTheDocument();
 		});
 	});
 });

frontend/src/components/ServiceAccountDrawer/__tests__/ServiceAccountDrawer.test.tsx

@@ -222,20 +222,21 @@ describe('ServiceAccountDrawer', () => {
 			screen.getByRole('button', { name: /Delete Service Account/i }),
 		);
 
-		await screen.findByTestId('delete-service-account-modal');
-		expect(
-			screen.getByTestId('delete-service-account-modal'),
-		).toBeInTheDocument();
+		const dialog = await screen.findByRole('dialog', {
+			name: /Delete service account CI Bot/i,
+		});
+		expect(dialog).toBeInTheDocument();
 
-		await user.click(screen.getByTestId('confirm-delete-btn'));
+		const confirmBtns = screen.getAllByRole('button', { name: /^Delete$/i });
+		await user.click(confirmBtns[confirmBtns.length - 1]);
 
-		await waitFor(
-			() => {
-				expect(deleteSpy).toHaveBeenCalled();
-				expect(screen.queryByDisplayValue('CI Bot')).not.toBeInTheDocument();
-			},
-			{ timeout: 3000 },
-		);
+		await waitFor(() => {
+			expect(deleteSpy).toHaveBeenCalled();
+		});
+
+		await waitFor(() => {
+			expect(screen.queryByDisplayValue('CI Bot')).not.toBeInTheDocument();
+		});
 	});
 
 	it('deleted account shows read-only name, no Save button, no Delete button', async () => {

frontend/src/container/ServiceAccountsSettings/__tests__/ServiceAccountsSettings.authz.integration.test.tsx

@@ -208,7 +208,7 @@ describe('ServiceAccountsSettings (integration)', () => {
 
 		fireEvent.click(screen.getByRole('button', { name: /New Service Account/i }));
 
-		await screen.findByTestId('create-service-account-modal');
+		await screen.findByRole('dialog', { name: /New Service Account/i });
 		expect(screen.getByPlaceholderText('Enter a name')).toBeInTheDocument();
 	});
 

frontend/src/pages/TraceDetailsV3/TraceDetailsHeader/MissingSpansBanner.tsx

@@ -5,7 +5,7 @@ import { ArrowUpRight } from '@signozhq/icons';
 import styles from './MissingSpansBanner.module.scss';
 
 const MISSING_SPANS_DOCS_URL =
-	'https://signoz.io/docs/traces-management/troubleshooting/faqs/#q-why-are-some-spans-missing-from-a-trace';
+	'https://signoz.io/docs/userguide/traces/#missing-spans';
 
 function MissingSpansBanner(): JSX.Element | null {
 	// Session-only dismissal — not persisted, so the banner returns on reload.

frontend/src/utils/docLinks.ts

@@ -14,8 +14,7 @@ const DOCLINKS = {
 		'https://signoz.io/docs/userguide/logs_clickhouse_queries/',
 	QUERY_CLICKHOUSE_METRICS:
 		'https://signoz.io/docs/userguide/write-a-metrics-clickhouse-query/',
-	AGENT_SKILL_INSTALL:
-		'https://signoz.io/docs/ai/agent-skills/#install-the-plugin',
+	AGENT_SKILL_INSTALL: 'https://signoz.io/docs/ai/agent-skills/#installation',
 };
 
 export default DOCLINKS;

pkg/modules/inframonitoring/implinframonitoring/checks_constants.go

@@ -16,11 +16,11 @@ const (
 
 // Documentation links — one per component. User-facing; emitted on missing-entries.
 const (
-	docLinkHostMetricsReceiver        = "https://signoz.io/docs/infrastructure-monitoring/hostmetrics/#configure-the-hostmetrics-receiver"
-	docLinkKubeletStatsReceiver       = "https://signoz.io/docs/infrastructure-monitoring/k8s-metrics/#2-configure-the-kubelet-stats-receiver"
-	docLinkK8sClusterReceiver         = "https://signoz.io/docs/infrastructure-monitoring/k8s-metrics/#1-configure-the-k8s-cluster-receiver"
-	docLinkResourceDetectionProcessor = "https://signoz.io/docs/infrastructure-monitoring/hostmetrics/#configure-the-processors"
-	docLinkK8sAttributesProcessor     = "https://signoz.io/docs/infrastructure-monitoring/k8s-metrics/#3-enable-kubernetes-metadata"
+	docLinkHostMetricsReceiver        = "https://signoz.io/docs/infrastructure-monitoring/user-guides/hostmetrics/#configure-the-hostmetrics-receiver"
+	docLinkKubeletStatsReceiver       = "https://signoz.io/docs/infrastructure-monitoring/user-guides/k8s-metrics/#setup-kubelet-stats-receiver"
+	docLinkK8sClusterReceiver         = "https://signoz.io/docs/infrastructure-monitoring/user-guides/k8s-metrics/#setup-k8s-cluster-receiver"
+	docLinkResourceDetectionProcessor = "https://signoz.io/docs/infrastructure-monitoring/user-guides/hostmetrics/#configure-the-resourcedetection-processor"
+	docLinkK8sAttributesProcessor     = "https://signoz.io/docs/infrastructure-monitoring/user-guides/k8s-metrics/#3-setup-k8sattributesprocessor-to-enable-kubernetes-metadata"
 )
 
 var (

pkg/prometheus/clickhouseprometheus/client.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
-	"github.com/SigNoz/signoz/pkg/query-service/constants"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
 	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
 	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
@@ -143,10 +142,6 @@ func (client *client) queryToClickhouseQuery(_ context.Context, query *prompb.Qu
 	conditions = append(conditions, "temporality IN ['Cumulative', 'Unspecified']")
 	conditions = append(conditions, fmt.Sprintf("unix_milli >= %d AND unix_milli < %d", start, end))
 
-	normalized := !constants.IsDotMetricsEnabled
-
-	conditions = append(conditions, fmt.Sprintf("__normalized = %v", normalized))
-
 	args = append(args, metricName)
 	for _, m := range query.Matchers {
 		switch m.Type {

pkg/query-service/constants/constants.go

@@ -124,7 +124,7 @@ const (
 // alert related constants
 const (
 	// AlertHelpPage is used in case default alert repo url is not set
-	AlertHelpPage   = "https://signoz.io/docs/alerts/"
+	AlertHelpPage   = "https://signoz.io/docs/userguide/alerts-management/#generator-url"
 	AlertTimeFormat = "2006-01-02 15:04:05"
 )
 

pkg/query-service/rules/threshold_rule_test.go

@@ -628,7 +628,7 @@ func TestThresholdRuleUnitCombinations(t *testing.T) {
 		queryString := "SELECT any"
 		telemetryStore.Mock().
 			ExpectQuery(queryString).
-			WithArgs(nil, nil, nil, nil, nil, nil, nil, nil, nil).
+			WithArgs(nil, nil, nil, nil, nil, nil, nil, nil).
 			WillReturnRows(rows)
 		postableRule.RuleCondition.CompareOperator = c.compareOperator
 		postableRule.RuleCondition.MatchType = c.matchType
@@ -737,7 +737,7 @@ func TestThresholdRuleNoData(t *testing.T) {
 		queryString := "SELECT any"
 		telemetryStore.Mock().
 			ExpectQuery(queryString).
-			WithArgs(nil, nil, nil, nil, nil, nil, nil, nil).
+			WithArgs(nil, nil, nil, nil, nil, nil, nil).
 			WillReturnRows(rows)
 
 		querier, mockMetadataStore := prepareQuerierForMetrics(t, telemetryStore)
@@ -1129,7 +1129,7 @@ func TestMultipleThresholdRule(t *testing.T) {
 		queryString := "SELECT any"
 		telemetryStore.Mock().
 			ExpectQuery(queryString).
-			WithArgs(nil, nil, nil, nil, nil, nil, nil, nil).
+			WithArgs(nil, nil, nil, nil, nil, nil, nil).
 			WillReturnRows(rows)
 
 		querier, mockMetadataStore := prepareQuerierForMetrics(t, telemetryStore)
@@ -1922,7 +1922,7 @@ func TestThresholdEval_RequireMinPoints(t *testing.T) {
 		queryString := "SELECT any"
 		telemetryStore.Mock().
 			ExpectQuery(queryString).
-			WithArgs(nil, nil, nil, nil, nil, nil, nil, nil).
+			WithArgs(nil, nil, nil, nil, nil, nil, nil).
 			WillReturnRows(rows)
 
 		querier, mockMetadataStore := prepareQuerierForMetrics(t, telemetryStore)

pkg/query-service/rules/utils.go

@@ -57,9 +57,6 @@ func GenerateMetricQueryCHArgs(
 		queryArgs = append(queryArgs, temporality.StringValue())
 	}
 
-	// Add normalized flag
-	queryArgs = append(queryArgs, false)
-
 	// Step2: Add temporal aggregation args
 	// build args for filtering signoz_metrics.distributed_samples_v4 table
 	temporalAggArgs := []interface{}{

pkg/querybuilder/key_resolution.go

@@ -13,10 +13,10 @@ const (
 	// to multiple field context / data type combinations.
 	FieldContextDataTypesDocURL = "https://signoz.io/docs/userguide/field-context-data-types/"
 	// KeyNotFoundDocURL documents the "key not found" error.
-	KeyNotFoundDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#q-im-getting-key-fieldname-not-found--why-cant-it-find-my-field"
+	KeyNotFoundDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#key-fieldname-not-found"
 
 	// Doc URLs for the has/hasAny/hasAll and hasToken "unsupported" errors.
-	functionBodyJSONSearchDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#q-im-getting-function-supports-only-body-json-search--can-i-use-functions-on-other-fields"
+	functionBodyJSONSearchDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#function-supports-only-body-json-search"
 	hasTokenFunctionDocURL       = "https://signoz.io/docs/userguide/functions-reference/#hastoken-function"
 )
 

pkg/querybuilder/where_clause_visitor_test.go

@@ -169,7 +169,7 @@ func TestVisitKey(t *testing.T) {
 			},
 			expectedKeys:       []telemetrytypes.TelemetryFieldKey{},
 			expectedErrors:     []string{"key `unknown_key` not found"},
-			expectedMainErrURL: "https://signoz.io/docs/userguide/search-troubleshooting/#q-im-getting-key-fieldname-not-found--why-cant-it-find-my-field",
+			expectedMainErrURL: "https://signoz.io/docs/userguide/search-troubleshooting/#key-fieldname-not-found",
 			expectedWarnings:   nil,
 			expectedMainWrnURL: "",
 		},
@@ -351,7 +351,7 @@ func TestVisitKey(t *testing.T) {
 			ignoreNotFoundKeys: false,
 			expectedKeys:       []telemetrytypes.TelemetryFieldKey{},
 			expectedErrors:     []string{"key `unknown_key` not found"},
-			expectedMainErrURL: "https://signoz.io/docs/userguide/search-troubleshooting/#q-im-getting-key-fieldname-not-found--why-cant-it-find-my-field",
+			expectedMainErrURL: "https://signoz.io/docs/userguide/search-troubleshooting/#key-fieldname-not-found",
 			expectedWarnings:   nil,
 			expectedMainWrnURL: "",
 		},

pkg/telemetrylogs/const.go

@@ -53,7 +53,7 @@ const (
 	// Documentation URLs attached to function-call errors so the visitor can
 	// surface them to the user without knowing function-specific details.
 	hasTokenFunctionDocURL       = "https://signoz.io/docs/userguide/functions-reference/#hastoken-function"
-	functionBodyJSONSearchDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#q-im-getting-function-supports-only-body-json-search--can-i-use-functions-on-other-fields"
+	functionBodyJSONSearchDocURL = "https://signoz.io/docs/userguide/search-troubleshooting/#function-supports-only-body-json-search"
 )
 
 var (

pkg/telemetrymetrics/const.go

@@ -3,7 +3,6 @@ package telemetrymetrics
 import "github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 
 var IntrinsicFields = []string{
-	"__normalized",
 	"temporality",
 	"metric_name",
 	"type",

pkg/telemetrymetrics/reduced_test.go

@@ -39,80 +39,80 @@ func TestReducedStatementBuilder(t *testing.T) {
 			name:  "gauge_sum_latest",
 			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationLatest, metrictypes.SpaceAggregationSum),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, anyLast(last) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, argMax(value, unix_milli) AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum_last`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
-				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, anyLast(last) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT points.reduced_fingerprint AS fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, argMax(`sum_last`, unix_milli) AS per_series_value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points FINAL INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746997200000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			name:  "gauge_avg_avg",
 			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationAvg, metrictypes.SpaceAggregationAvg),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(sum) / sum(count) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, avg(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum_last`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
-				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(sum) / sum(count) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT points.reduced_fingerprint AS fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, avg(`sum_last`) AS per_series_value, avg(`count_series`) AS per_series_weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points FINAL INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746997200000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			name:  "gauge_min_min",
 			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationMin, metrictypes.SpaceAggregationMin),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, min(min) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, min(value) AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`min`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
-				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, min(min) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, min(`min`) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points FINAL INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746997200000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			name:  "gauge_max_max",
 			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationMax, metrictypes.SpaceAggregationMax),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(value) AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`max`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
-				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(`max`) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points FINAL INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746997200000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			name:  "counter_sum_rate",
 			query: reducedQuery("test.metric.sum", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationRate, metrictypes.SpaceAggregationSum),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(value) / 300 AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
-				Args:  []any{"test.metric.sum", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric.sum", uint64(1746999600000), uint64(1747172760000), 0, "test.metric.sum", uint64(1746999600000), uint64(1747172760000), "test.metric.sum", uint64(1746999600000), uint64(1747172760000), false},
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(`sum`) / 300 AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s AS points FINAL INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+				Args:  []any{"test.metric.sum", uint64(1746921600000), uint64(1747172760000), "cumulative", "test.metric.sum", uint64(1746999600000), uint64(1747172760000), 0, "test.metric.sum", uint64(1746997200000), uint64(1747172760000), "test.metric.sum", uint64(1746999600000), uint64(1747172760000)},
 			},
 		},
 		{
 			name:  "counter_avg_increase",
 			query: reducedQuery("test.metric", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationIncrease, metrictypes.SpaceAggregationAvg),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value, per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
-				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric", uint64(1746999600000), uint64(1747172760000), 0, "test.metric", uint64(1746999600000), uint64(1747172760000), "test.metric", uint64(1746999600000), uint64(1747172760000), false},
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value, per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT points.reduced_fingerprint AS fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(`sum`) AS per_series_value, avg(`count_series`) AS per_series_weight FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s AS points FINAL INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", "test.metric", uint64(1746999600000), uint64(1747172760000), 0, "test.metric", uint64(1746997200000), uint64(1747172760000), "test.metric", uint64(1746999600000), uint64(1747172760000)},
 			},
 		},
 		{
 			name:  "counter_min_omitted",
 			query: reducedQuery("test.metric", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationRate, metrictypes.SpaceAggregationMin),
 			expected: qbtypes.Statement{
-				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts",
-				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric", uint64(1746999600000), uint64(1747172760000), 0},
+				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", "test.metric", uint64(1746999600000), uint64(1747172760000), 0},
 			},
 		},
 		{
 			name:  "counter_max_omitted",
 			query: reducedQuery("test.metric", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationRate, metrictypes.SpaceAggregationMax),
 			expected: qbtypes.Statement{
-				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts",
-				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric", uint64(1746999600000), uint64(1747172760000), 0},
+				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", "test.metric", uint64(1746999600000), uint64(1747172760000), 0},
 			},
 		},
 		{
 			name:  "histogram_p99",
 			query: reducedQuery("test.metric.bucket", metrictypes.HistogramType, metrictypes.Cumulative, metrictypes.TimeAggregationUnspecified, metrictypes.SpaceAggregationPercentile99),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, `le`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `le`, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `le` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `le`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `le`) SELECT ts, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.990) AS value FROM __spatial_aggregation_cte GROUP BY ts ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `le`, sum(value) / 300 AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts, `le`), __spatial_aggregation_cte AS (SELECT ts, `le`, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts, `le`) SELECT ts, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.990) AS value FROM __spatial_aggregation_cte GROUP BY ts ORDER BY ts) ORDER BY ts",
-				Args:  []any{"test.metric.bucket", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric.bucket", uint64(1746999900000), uint64(1747172760000), 0, "test.metric.bucket", uint64(1746999900000), uint64(1747172760000), "test.metric.bucket", uint64(1746999900000), uint64(1747172760000), false},
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, `le`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `le`, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `le` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `le`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `le`) SELECT ts, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.990) AS value FROM __spatial_aggregation_cte GROUP BY ts ORDER BY ts) UNION ALL SELECT * FROM (WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `le`, sum(`sum`) / 300 AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s AS points FINAL INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint, `le`) AS filtered_time_series ON points.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `le`) SELECT ts, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.990) AS value FROM __spatial_aggregation_cte GROUP BY ts ORDER BY ts) ORDER BY ts SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+				Args:  []any{"test.metric.bucket", uint64(1746921600000), uint64(1747172760000), "cumulative", "test.metric.bucket", uint64(1746999900000), uint64(1747172760000), 0, "test.metric.bucket", uint64(1746997200000), uint64(1747172760000), "test.metric.bucket", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			name:  "summary_avg",
 			query: reducedQuery("test.metric", metrictypes.SummaryType, metrictypes.Unspecified, metrictypes.TimeAggregationAvg, metrictypes.SpaceAggregationAvg),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(sum) / sum(count) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, avg(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum_last`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
-				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(sum) / sum(count) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT points.reduced_fingerprint AS fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, avg(`sum_last`) AS per_series_value, avg(`count_series`) AS per_series_weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points FINAL INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? GROUP BY fingerprint) AS filtered_time_series ON points.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746997200000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 	}

pkg/telemetrymetrics/statement_builder.go

@@ -228,10 +228,18 @@ func (b *MetricQueryStatementBuilder) buildPipelineStatement(
 	if agg.Reduced && !useBuffer {
 		var tsCTE string
 		var tsArgs []any
-		if tsCTE, tsArgs, err = b.buildReducedTimeSeriesCTE(ctx, start, end, query, keys, variables); err != nil {
+		// time series rows are written on hour boundaries
+		tsStart := start - (start % oneHourInMilliseconds)
+		if tsCTE, tsArgs, err = b.buildReducedTimeSeriesCTE(ctx, tsStart, end, query, keys, variables); err != nil {
 			return nil, err
 		}
-		if temporalFrag, temporalArgs, ok := b.buildReducedTemporalAggregationCTE(start, end, query, tsCTE, tsArgs); ok {
+		if qbtypes.CanShortCircuitReduced(agg) {
+			// spatial_aggregation_cte directly, no per-series level
+			if spatialFrag, spatialArgs, ok := b.buildReducedSpatialAggFastPath(start, end, query, tsCTE, tsArgs); ok {
+				reducedFragments = []string{spatialFrag}
+				reducedArgs = [][]any{spatialArgs}
+			}
+		} else if temporalFrag, temporalArgs, ok := b.buildReducedTemporalAggregationCTE(start, end, query, tsCTE, tsArgs); ok {
 			spatialFrag, spatialArgs := b.buildReducedSpatialAggregationCTE(query)
 			reducedFragments = []string{temporalFrag, spatialFrag}
 			reducedArgs = [][]any{temporalArgs, spatialArgs}
@@ -262,7 +270,10 @@ func unionStatements(main, reduced *qbtypes.Statement, query qbtypes.QueryBuilde
 	for _, g := range query.GroupBy {
 		orderBy = fmt.Sprintf("`%s`, ", g.Name) + orderBy
 	}
-	q := fmt.Sprintf("SELECT * FROM (%s) UNION ALL SELECT * FROM (%s) ORDER BY %s", main.Query, reduced.Query, orderBy)
+	q := fmt.Sprintf(
+		"SELECT * FROM (%s) UNION ALL SELECT * FROM (%s) ORDER BY %s SETTINGS do_not_merge_across_partitions_select_final = 1, optimize_move_to_prewhere_if_final = 1",
+		main.Query, reduced.Query, orderBy,
+	)
 	args := append(append([]any{}, main.Args...), reduced.Args...)
 	warnings := append(append([]string{}, main.Warnings...), reduced.Warnings...)
 	return &qbtypes.Statement{Query: q, Args: args, Warnings: warnings}, nil
@@ -309,7 +320,6 @@ func (b *MetricQueryStatementBuilder) buildReducedTimeSeriesCTE(
 		sb.In("metric_name", query.Aggregations[0].MetricName),
 		sb.GTE("unix_milli", start),
 		sb.LTE("unix_milli", end),
-		sb.EQ("__normalized", false),
 	)
 
 	if !preparedWhereClause.IsEmpty() {
@@ -322,6 +332,46 @@ func (b *MetricQueryStatementBuilder) buildReducedTimeSeriesCTE(
 	return fmt.Sprintf("(%s) AS filtered_time_series", q), args, nil
 }
 
+// buildReducedSpatialAggFastPath is the reduced analog of
+// buildTemporalAggDeltaFastPath: for combinations where the temporal and
+// spatial aggregations collapse (CanShortCircuitReduced), it emits the
+// spatial_aggregation_cte in one level with no per-series grouping, so shards
+// send one state per (step, group) instead of per (series, step, group).
+// FINAL still dedups recomputed 60s buckets at scan time.
+func (b *MetricQueryStatementBuilder) buildReducedSpatialAggFastPath(
+	start, end uint64,
+	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	timeSeriesCTE string,
+	timeSeriesCTEArgs []any,
+) (string, []any, bool) {
+	agg := query.Aggregations[0]
+	stepSec := int64(query.StepInterval.Seconds())
+
+	value, _, ok := ReducedValueColumn(agg.Type, agg.SpaceAggregation)
+	if !ok {
+		return "", nil, false
+	}
+
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select(fmt.Sprintf("toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts", stepSec))
+	for _, g := range query.GroupBy {
+		sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
+	}
+	sb.SelectMore(fmt.Sprintf("%s AS value", ReducedTimeAggregationColumn(agg.TimeAggregation, stepSec, value)))
+	sb.From(fmt.Sprintf("%s.%s AS points FINAL", DBName, WhichReducedSamplesTableToUse(agg.Type)))
+	sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.reduced_fingerprint = filtered_time_series.fingerprint")
+	sb.Where(
+		sb.In("metric_name", agg.MetricName),
+		sb.GTE("unix_milli", start),
+		sb.LT("unix_milli", end),
+	)
+	sb.GroupBy("ts")
+	sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+
+	q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, timeSeriesCTEArgs...)
+	return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args, true
+}
+
 func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
 	start, end uint64,
 	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
@@ -336,41 +386,31 @@ func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
 		return "", nil, false
 	}
 
-	// dedup recomputed buckets: latest computed_at wins per (series, 60s bucket)
-	dedup := sqlbuilder.NewSelectBuilder()
-	dedup.Select("reduced_fingerprint AS fingerprint", "unix_milli")
-	dedup.SelectMore(fmt.Sprintf("argMax(%s, computed_at) AS value", value))
-	if weight != "" {
-		dedup.SelectMore(fmt.Sprintf("argMax(%s, computed_at) AS weight", weight))
-	}
-	dedup.From(fmt.Sprintf("%s.%s", DBName, WhichReducedSamplesTableToUse(agg.Type)))
-	dedup.Where(
-		dedup.In("metric_name", agg.MetricName),
-		dedup.GTE("unix_milli", start),
-		dedup.LT("unix_milli", end),
-	)
-	dedup.GroupBy("reduced_fingerprint", "unix_milli")
-	dedupQuery, dedupArgs := dedup.BuildWithFlavor(sqlbuilder.ClickHouse)
-
+	// TODO(srikanthccv): add _5m/_30m tables similar to samples_v4
+	// and wire them up in querier before GA
 	sb := sqlbuilder.NewSelectBuilder()
-	sb.Select("fingerprint")
+	sb.Select("points.reduced_fingerprint AS fingerprint")
 	sb.SelectMore(fmt.Sprintf("toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts", stepSec))
 	for _, g := range query.GroupBy {
 		sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
 	}
-	sb.SelectMore(fmt.Sprintf("%s AS per_series_value", ReducedTimeAggregationColumn(agg.TimeAggregation, stepSec)))
+	sb.SelectMore(fmt.Sprintf("%s AS per_series_value", ReducedTimeAggregationColumn(agg.TimeAggregation, stepSec, value)))
 	if weight != "" {
 		// count_series is a series count, not additive over time, so the avg
 		// denominator is reduced with avg
-		sb.SelectMore("avg(weight) AS per_series_weight")
+		sb.SelectMore(fmt.Sprintf("avg(%s) AS per_series_weight", weight))
 	}
-	sb.From(fmt.Sprintf("(%s) AS points", dedupQuery))
-	sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
+	sb.From(fmt.Sprintf("%s.%s AS points FINAL", DBName, WhichReducedSamplesTableToUse(agg.Type)))
+	sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.reduced_fingerprint = filtered_time_series.fingerprint")
+	sb.Where(
+		sb.In("metric_name", agg.MetricName),
+		sb.GTE("unix_milli", start),
+		sb.LT("unix_milli", end),
+	)
 	sb.GroupBy("fingerprint", "ts")
 	sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
 
-	initArgs := append(append([]any{}, dedupArgs...), timeSeriesCTEArgs...)
-	q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, initArgs...)
+	q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, timeSeriesCTEArgs...)
 	return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, true
 }
 
@@ -503,11 +543,6 @@ func (b *MetricQueryStatementBuilder) buildTimeSeriesCTE(
 		sb.Where(sb.ILike("temporality", query.Aggregations[0].Temporality.StringValue()))
 	}
 
-	// TODO configurable if we don't rollout the new un-normalized metrics
-	sb.Where(
-		sb.EQ("__normalized", false),
-	)
-
 	// the buffer holds both raw rows and the reduced catalog rows; the raw read
 	// only wants the original series
 	if tsTable == TimeseriesV4BufferLocalTableName {

pkg/telemetrymetrics/stmt_builder_test.go

@@ -50,8 +50,8 @@ func TestStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `service.name`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `service.name` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `service.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `service.name`, ts",
-				Args:  []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "cumulative", false, "cartservice", "signoz_calls_total", uint64(1747947360000), uint64(1747983420000), 0},
+				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `service.name`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `service.name` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `service.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `service.name`, ts",
+				Args:  []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "cumulative", "cartservice", "signoz_calls_total", uint64(1747947360000), uint64(1747983420000), 0},
 			},
 			expectedErr: nil,
 		},
@@ -83,8 +83,8 @@ func TestStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `service.name`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND (match(JSONExtractString(labels, 'materialized.key.name'), ?) OR JSONExtractString(labels, 'service.name') = ?) GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `service.name` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `service.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `service.name`, ts",
-				Args:  []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "cumulative", false, "cartservice", "cartservice", "signoz_calls_total", uint64(1747947360000), uint64(1747983420000), 0},
+				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `service.name`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND (match(JSONExtractString(labels, 'materialized.key.name'), ?) OR JSONExtractString(labels, 'service.name') = ?) GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `service.name` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `service.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `service.name`, ts",
+				Args:  []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "cumulative", "cartservice", "cartservice", "signoz_calls_total", uint64(1747947360000), uint64(1747983420000), 0},
 			},
 			expectedErr: nil,
 		},
@@ -116,8 +116,8 @@ func TestStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, sum(value)/30 AS value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `service.name`, ts",
-				Args:  []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "delta", false, "cartservice", "signoz_calls_total", uint64(1747947390000), uint64(1747983420000)},
+				Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, sum(value)/30 AS value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `service.name`, ts",
+				Args:  []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "delta", "cartservice", "signoz_calls_total", uint64(1747947390000), uint64(1747983420000)},
 			},
 			expectedErr: nil,
 		},
@@ -149,8 +149,8 @@ func TestStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, `le`, sum(value)/30 AS value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name`, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `service.name`, `le`) SELECT ts, `service.name`, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.950) AS value FROM __spatial_aggregation_cte GROUP BY `service.name`, ts ORDER BY `service.name`, ts",
-				Args:  []any{"signoz_latency", uint64(1747936800000), uint64(1747983420000), "delta", false, "cartservice", "signoz_latency", uint64(1747947390000), uint64(1747983420000)},
+				Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, `le`, sum(value)/30 AS value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name`, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `service.name`, `le`) SELECT ts, `service.name`, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.950) AS value FROM __spatial_aggregation_cte GROUP BY `service.name`, ts ORDER BY `service.name`, ts",
+				Args:  []any{"signoz_latency", uint64(1747936800000), uint64(1747983420000), "delta", "cartservice", "signoz_latency", uint64(1747947390000), uint64(1747983420000)},
 			},
 			expectedErr: nil,
 		},
@@ -182,8 +182,8 @@ func TestStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `host.name`, avg(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'host.name') AS `host.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'host.name') = ? GROUP BY fingerprint, `host.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `host.name` ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, `host.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `host.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `host.name`, ts",
-				Args:  []any{"system.memory.usage", uint64(1747936800000), uint64(1747983420000), "unspecified", false, "big-data-node-1", "system.memory.usage", uint64(1747947390000), uint64(1747983420000), 0},
+				Query: "WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `host.name`, avg(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'host.name') AS `host.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND JSONExtractString(labels, 'host.name') = ? GROUP BY fingerprint, `host.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `host.name` ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, `host.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `host.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `host.name`, ts",
+				Args:  []any{"system.memory.usage", uint64(1747936800000), uint64(1747983420000), "unspecified", "big-data-node-1", "system.memory.usage", uint64(1747947390000), uint64(1747983420000), 0},
 			},
 			expectedErr: nil,
 		},
@@ -212,8 +212,8 @@ func TestStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `service.name`, `le`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, `le`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name`, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint, `service.name`, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `service.name`, `le` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `service.name`, `le`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`, `le`) SELECT ts, `service.name`, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.950) AS value FROM __spatial_aggregation_cte GROUP BY `service.name`, ts ORDER BY `service.name`, ts",
-				Args:  []any{"http_server_duration_bucket", uint64(1747936800000), uint64(1747983420000), "cumulative", false, "http_server_duration_bucket", uint64(1747947360000), uint64(1747983420000), 0},
+				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `service.name`, `le`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, `le`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name`, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, `service.name`, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `service.name`, `le` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `service.name`, `le`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`, `le`) SELECT ts, `service.name`, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.950) AS value FROM __spatial_aggregation_cte GROUP BY `service.name`, ts ORDER BY `service.name`, ts",
+				Args:  []any{"http_server_duration_bucket", uint64(1747936800000), uint64(1747983420000), "cumulative", "http_server_duration_bucket", uint64(1747947360000), uint64(1747983420000), 0},
 			},
 			expectedErr: nil,
 		},
@@ -244,8 +244,8 @@ func TestStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `k8s.statefulset.name`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `k8s.statefulset.name`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'k8s.statefulset.name') AS `k8s.statefulset.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'k8s.statefulset.name') = ? GROUP BY fingerprint, `k8s.statefulset.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `k8s.statefulset.name` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `k8s.statefulset.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `k8s.statefulset.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `k8s.statefulset.name`, ts",
-				Args:     []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "cumulative", false, "my-statefulset", "signoz_calls_total", uint64(1747947360000), uint64(1747983420000), 0},
+				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `k8s.statefulset.name`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `k8s.statefulset.name`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'k8s.statefulset.name') AS `k8s.statefulset.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND JSONExtractString(labels, 'k8s.statefulset.name') = ? GROUP BY fingerprint, `k8s.statefulset.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `k8s.statefulset.name` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `k8s.statefulset.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `k8s.statefulset.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `k8s.statefulset.name`, ts",
+				Args:  []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "cumulative", "my-statefulset", "signoz_calls_total", uint64(1747947360000), uint64(1747983420000), 0},
 			},
 			expectedErr: nil,
 		},

pkg/telemetrymetrics/tables.go

@@ -393,25 +393,24 @@ func ReducedValueColumn(metricType metrictypes.Type, space metrictypes.SpaceAggr
 	return "", "", false
 }
 
-// ReducedTimeAggregationColumn applies the time aggregation to the reduced `value`
-// column over the step's 60s buckets. latest uses argMax over the bucket timestamp
-// (the buckets have no read order); rate divides the per-step sum by the step.
-func ReducedTimeAggregationColumn(timeAggregation metrictypes.TimeAggregation, stepSec int64) string {
+// ReducedTimeAggregationColumn applies the time aggregation to the reduced value
+// column over the step's 60s buckets.
+func ReducedTimeAggregationColumn(timeAggregation metrictypes.TimeAggregation, stepSec int64, value string) string {
 	switch timeAggregation {
 	case metrictypes.TimeAggregationLatest:
-		return "argMax(value, unix_milli)"
+		return fmt.Sprintf("argMax(%s, unix_milli)", value)
 	case metrictypes.TimeAggregationAvg:
-		return "avg(value)"
+		return fmt.Sprintf("avg(%s)", value)
 	case metrictypes.TimeAggregationMin:
-		return "min(value)"
+		return fmt.Sprintf("min(%s)", value)
 	case metrictypes.TimeAggregationMax:
-		return "max(value)"
+		return fmt.Sprintf("max(%s)", value)
 	case metrictypes.TimeAggregationCount:
-		return "count(value)"
+		return fmt.Sprintf("count(%s)", value)
 	case metrictypes.TimeAggregationRate:
-		return fmt.Sprintf("sum(value) / %d", stepSec)
+		return fmt.Sprintf("sum(%s) / %d", value, stepSec)
 	default: // sum, increase
-		return "sum(value)"
+		return fmt.Sprintf("sum(%s)", value)
 	}
 }
 

pkg/types/dashboardtypes/perses_dashboard_convertors_test.go

@@ -38,7 +38,7 @@ func newTestDashboardV2(t *testing.T, orgID valuer.UUID, source Source) *Dashboa
 								LineInterpolation: LineInterpolationSpline,
 								LineStyle:         LineStyleSolid,
 								FillMode:          FillModeSolid,
-								SpanGaps:          SpanGaps{FillLessThan: "60s"},
+								SpanGaps:          SpanGaps{FillLessThan: valuer.MustParseTextDuration("60s")},
 							},
 							Legend: Legend{Position: LegendPositionBottom, Mode: LegendModeList},
 						},

pkg/types/dashboardtypes/perses_dashboard_test.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/perses/spec/go/dashboard"
@@ -752,7 +753,7 @@ func TestInvalidateBadPanelSpecValues(t *testing.T) {
 						"spec": {
 							"plugin": {
 								"kind": "signoz/TimeSeriesPanel",
-								"spec": {"chartAppearance": {"spanGaps": {"fillOnlyBelow": true, "fillLessThan": "notaduration"}}}
+								"spec": {"chartAppearance": {"spanGaps": {"fillLessThan": "notaduration"}}}
 							}
 						}
 					}
@@ -1370,49 +1371,23 @@ func TestSpanGaps(t *testing.T) {
 	t.Run("defaults", func(t *testing.T) {
 		var sg SpanGaps
 		assert.False(t, sg.FillOnlyBelow, "expected FillOnlyBelow default false")
-		assert.Empty(t, sg.FillLessThan, "expected FillLessThan default empty")
+		assert.True(t, sg.FillLessThan.IsZero(), "expected FillLessThan default zero")
 	})
 
 	t.Run("fillOnlyBelow true", func(t *testing.T) {
-		sg := unmarshal(t, `{"fillOnlyBelow": true, "fillLessThan": "5m"}`)
+		sg := unmarshal(t, `{"fillOnlyBelow": true}`)
 		assert.True(t, sg.FillOnlyBelow)
 	})
 
-	t.Run("fillLessThan ignored when fillOnlyBelow is false", func(t *testing.T) {
-		sg := unmarshal(t, `{"fillOnlyBelow": false, "fillLessThan": ""}`)
-		assert.False(t, sg.FillOnlyBelow)
-		assert.Empty(t, sg.FillLessThan)
-	})
-
 	t.Run("fillLessThan duration", func(t *testing.T) {
-		sg := unmarshal(t, `{"fillOnlyBelow": true, "fillLessThan": "5m"}`)
-		assert.True(t, sg.FillOnlyBelow)
-		assert.Equal(t, "5m", sg.FillLessThan)
+		sg := unmarshal(t, `{"fillOnlyBelow": false, "fillLessThan": "5m"}`)
+		assert.False(t, sg.FillOnlyBelow)
+		assert.Equal(t, 5*time.Minute, sg.FillLessThan.Duration())
 	})
 
 	t.Run("fillLessThan compound duration", func(t *testing.T) {
-		sg := unmarshal(t, `{"fillOnlyBelow": true, "fillLessThan": "1h30m"}`)
-		assert.Equal(t, "1h30m", sg.FillLessThan)
-	})
-
-	t.Run("fillLessThan day duration", func(t *testing.T) {
-		sg := unmarshal(t, `{"fillOnlyBelow": true, "fillLessThan": "1d"}`)
-		assert.Equal(t, "1d", sg.FillLessThan)
-	})
-
-	t.Run("fillLessThan required when fillOnlyBelow is true", func(t *testing.T) {
-		var sg SpanGaps
-		require.Error(t, json.Unmarshal([]byte(`{"fillOnlyBelow": true}`), &sg))
-	})
-
-	t.Run("invalid fillLessThan rejected on unmarshal", func(t *testing.T) {
-		var sg SpanGaps
-		require.Error(t, json.Unmarshal([]byte(`{"fillOnlyBelow": true, "fillLessThan": "not-a-duration"}`), &sg))
-	})
-
-	t.Run("non-positive fillLessThan rejected on unmarshal", func(t *testing.T) {
-		var sg SpanGaps
-		require.Error(t, json.Unmarshal([]byte(`{"fillOnlyBelow": true, "fillLessThan": "0s"}`), &sg))
+		sg := unmarshal(t, `{"fillLessThan": "1h30m"}`)
+		assert.Equal(t, 90*time.Minute, sg.FillLessThan.Duration())
 	})
 }
 

pkg/types/dashboardtypes/perses_signoz_plugins.go

@@ -8,7 +8,6 @@ import (
 	qb "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
-	"github.com/prometheus/common/model"
 	"github.com/swaggest/jsonschema-go"
 )
 
@@ -622,39 +621,8 @@ func (fm *FillMode) UnmarshalJSON(data []byte) error {
 }
 
 type SpanGaps struct {
-	FillOnlyBelow bool   `json:"fillOnlyBelow" description:"Controls whether lines connect across null values. When false (default), all gaps are connected. When true, only gaps smaller than fillLessThan are connected."`
-	FillLessThan  string `json:"fillLessThan" description:"The maximum gap size to connect when fillOnlyBelow is true. Gaps larger than this duration are left disconnected."`
-}
-
-func (sg *SpanGaps) UnmarshalJSON(data []byte) error {
-	type alias SpanGaps
-	var tmp alias
-	if err := json.Unmarshal(data, &tmp); err != nil {
-		return errors.WrapInvalidInputf(err, ErrCodeDashboardInvalidInput, "invalid spanGaps")
-	}
-	*sg = SpanGaps(tmp)
-	return sg.validate()
-}
-
-// validate enforces FillLessThan only when FillOnlyBelow is set, since that is
-// the only mode in which it applies. It must then be a valid positive duration.
-// prometheus's parser accepts day/week/year units (e.g. "1d"); time.ParseDuration
-// caps at hours.
-func (sg SpanGaps) validate() error {
-	if !sg.FillOnlyBelow {
-		return nil
-	}
-	if sg.FillLessThan == "" {
-		return errors.NewInvalidInputf(ErrCodeDashboardInvalidInput, "spanGaps.fillLessThan is required when fillOnlyBelow is true")
-	}
-	d, err := model.ParseDuration(sg.FillLessThan)
-	if err != nil {
-		return errors.WrapInvalidInputf(err, ErrCodeDashboardInvalidInput, "invalid spanGaps.fillLessThan duration %q", sg.FillLessThan)
-	}
-	if d <= 0 {
-		return errors.NewInvalidInputf(ErrCodeDashboardInvalidInput, "spanGaps.fillLessThan duration must be positive, got %q", sg.FillLessThan)
-	}
-	return nil
+	FillOnlyBelow bool                `json:"fillOnlyBelow" description:"Controls whether lines connect across null values. When false (default), all gaps are connected. When true, only gaps smaller than fillLessThan are connected."`
+	FillLessThan  valuer.TextDuration `json:"fillLessThan" description:"The maximum gap size to connect when fillOnlyBelow is true. Gaps larger than this duration are left disconnected."`
 }
 
 type PrecisionOption struct{ valuer.String }

pkg/types/dashboardtypes/testdata/perses_with_sections.json

@@ -76,7 +76,7 @@
                             "showPoints": false,
                             "lineStyle": "solid",
                             "fillMode": "none",
-                            "spanGaps": {"fillOnlyBelow": true, "fillLessThan": "5m"}
+                            "spanGaps": {"fillOnlyBelow": true}
                         },
                         "legend": {
                             "position": "bottom"

pkg/types/querybuildertypes/querybuildertypesv5/builder_query.go

@@ -256,3 +256,30 @@ func CanShortCircuitDelta(metricAgg MetricAggregation) bool {
 
 	return false
 }
+
+// CanShortCircuitReduced is like CanShortCircuitDelta but for reduced.
+func CanShortCircuitReduced(metricAgg MetricAggregation) bool {
+	if metricAgg.ValueFilter != nil {
+		return false
+	}
+
+	ta := metricAgg.TimeAggregation
+	sa := metricAgg.SpaceAggregation
+
+	if metricAgg.Type == metrictypes.SumType || metricAgg.Type == metrictypes.HistogramType {
+		return (ta == metrictypes.TimeAggregationRate || ta == metrictypes.TimeAggregationIncrease || ta == metrictypes.TimeAggregationSum) &&
+			sa == metrictypes.SpaceAggregationSum
+	}
+
+	if ta == metrictypes.TimeAggregationSum && sa == metrictypes.SpaceAggregationSum {
+		return true
+	}
+	if ta == metrictypes.TimeAggregationMin && sa == metrictypes.SpaceAggregationMin {
+		return true
+	}
+	if ta == metrictypes.TimeAggregationMax && sa == metrictypes.SpaceAggregationMax {
+		return true
+	}
+
+	return false
+}

tests/conftest.py

@@ -10,7 +10,7 @@ pytest_plugins = [
     "fixtures.postgres",
     "fixtures.sql",
     "fixtures.sqlite",
-    "fixtures.zookeeper",
+    "fixtures.keeper",
     "fixtures.signoz",
     "fixtures.audit",
     "fixtures.logs",
@@ -74,12 +74,6 @@ def pytest_addoption(parser: pytest.Parser):
         default="25.5.6",
         help="clickhouse version",
     )
-    parser.addoption(
-        "--zookeeper-version",
-        action="store",
-        default="3.7.1",
-        help="zookeeper version",
-    )
     parser.addoption(
         "--schema-migrator-version",
         action="store",

tests/fixtures/clickhouse.py

@@ -2,6 +2,7 @@ import os
 from collections.abc import Callable, Generator
 from datetime import datetime
 from typing import Any
+from uuid import uuid4
 
 import clickhouse_connect
 import clickhouse_connect.driver
@@ -17,30 +18,88 @@ from fixtures.logger import setup_logger
 
 logger = setup_logger(__name__)
 
+CLICKHOUSE_USERNAME = "signoz"
+CLICKHOUSE_PASSWORD = "password"
 
-@pytest.fixture(name="clickhouse", scope="package")
-def clickhouse(
-    tmpfs: Generator[types.LegacyPath, Any],
-    network: Network,
-    zookeeper: types.TestContainerDocker,
-    request: pytest.FixtureRequest,
-    pytestconfig: pytest.Config,
-) -> types.TestContainerClickhouse:
-    """
-    Package-scoped fixture for Clickhouse TestContainer.
+CUSTOM_FUNCTION_CONFIG = """
+<functions>
+    <function>
+        <type>executable</type>
+        <name>histogramQuantile</name>
+        <return_type>Float64</return_type>
+        <argument>
+            <type>Array(Float64)</type>
+            <name>buckets</name>
+        </argument>
+        <argument>
+            <type>Array(Float64)</type>
+            <name>counts</name>
+        </argument>
+        <argument>
+            <type>Float64</type>
+            <name>quantile</name>
+        </argument>
+        <format>CSV</format>
+        <command>./histogramQuantile</command>
+    </function>
+</functions>
+"""
+
+# Distributed inserts to a remote shard are async by default. We force
+# sycn at the profile level for deterministic tests.
+CLUSTER_USERS_CONFIG = """
+<clickhouse>
+    <profiles>
+        <default>
+            <insert_distributed_sync>1</insert_distributed_sync>
+        </default>
+    </profiles>
+</clickhouse>
+"""
+
+
+def render_remote_servers(shard_hosts: list[tuple[str, int]], secret: str | None = None) -> str:
+    """Render the <remote_servers> block for a cluster named `cluster` with one
+    single-replica shard per (host, port).
     """
+    shards = "".join(
+        f"""
+                <shard>
+                    <replica>
+                        <host>{host}</host>
+                        <port>{port}</port>
+                    </replica>
+                </shard>"""
+        for host, port in shard_hosts
+    )
 
-    def create() -> types.TestContainerClickhouse:
-        version = request.config.getoption("--clickhouse-version")
+    # Multi-node clusters need `secret` because distributed queries otherwise
+    # authenticate as the `default` user, which the docker entrypoint restricts
+    # to localhost when a custom user is configured.
+    secret_block = (
+        f"""
+                    <secret>{secret}</secret>"""
+        if secret
+        else ""
+    )
 
-        container = ClickHouseContainer(
-            image=f"clickhouse/clickhouse-server:{version}",
-            port=9000,
-            username="signoz",
-            password="password",
-        )
+    return f"""
+            <remote_servers>
+                <cluster>{secret_block}{shards}
+                </cluster>
+            </remote_servers>"""
 
-        cluster_config = f"""
+
+def render_node_config(
+    keeper_address: str,
+    keeper_port: int,
+    shard: str,
+    remote_servers: str,
+    distributed_ddl_path: str = "/clickhouse/task_queue/ddl",
+) -> str:
+    # <zookeeper> is ClickHouse's config section name for any coordination
+    # service, including ClickHouse Keeper.
+    return f"""
         <clickhouse>
             <logger>
                 <level>information</level>
@@ -55,33 +114,23 @@ def clickhouse(
             </logger>
 
             <macros>
-                <shard>01</shard>
+                <shard>{shard}</shard>
                 <replica>01</replica>
             </macros>
 
             <zookeeper>
                 <node>
-                    <host>{zookeeper.container_configs["2181"].address}</host>
-                    <port>{zookeeper.container_configs["2181"].port}</port>
+                    <host>{keeper_address}</host>
+                    <port>{keeper_port}</port>
                 </node>
             </zookeeper>
-
-            <remote_servers>
-                <cluster>
-                    <shard>
-                        <replica>
-                            <host>127.0.0.1</host>
-                            <port>9000</port>
-                        </replica>
-                    </shard>
-                </cluster>
-            </remote_servers>
+{remote_servers}
 
             <user_defined_executable_functions_config>*function.xml</user_defined_executable_functions_config>
             <user_scripts_path>/var/lib/clickhouse/user_scripts/</user_scripts_path>
 
             <distributed_ddl>
-                <path>/clickhouse/task_queue/ddl</path>
+                <path>{distributed_ddl_path}</path>
                 <profile>default</profile>
             </distributed_ddl>
 
@@ -122,38 +171,66 @@ def clickhouse(
         </clickhouse>
         """
 
-        custom_function_config = """
-        <functions>
-            <function>
-                <type>executable</type>
-                <name>histogramQuantile</name>
-                <return_type>Float64</return_type>
-                <argument>
-                    <type>Array(Float64)</type>
-                    <name>buckets</name>
-                </argument>
-                <argument>
-                    <type>Array(Float64)</type>
-                    <name>counts</name>
-                </argument>
-                <argument>
-                    <type>Float64</type>
-                    <name>quantile</name>
-                </argument>
-                <format>CSV</format>
-                <command>./histogramQuantile</command>
-            </function>
-        </functions>
-        """
 
-        tmp_dir = tmpfs("clickhouse")
+def install_histogram_quantile(container: ClickHouseContainer) -> None:
+    wrapped = container.get_wrapped_container()
+    exit_code, output = wrapped.exec_run(
+        [
+            "bash",
+            "-c",
+            (
+                'version="v0.0.1" && '
+                'node_os=$(uname -s | tr "[:upper:]" "[:lower:]") && '
+                "node_arch=$(uname -m | sed s/aarch64/arm64/ | sed s/x86_64/amd64/) && "
+                "cd /tmp && "
+                'wget -O histogram-quantile.tar.gz "https://github.com/SigNoz/signoz/releases/download/histogram-quantile%2F${version}/histogram-quantile_${node_os}_${node_arch}.tar.gz" && '
+                "tar -xzf histogram-quantile.tar.gz && "
+                "mkdir -p /var/lib/clickhouse/user_scripts && "
+                "mv histogram-quantile /var/lib/clickhouse/user_scripts/histogramQuantile && "
+                "chmod +x /var/lib/clickhouse/user_scripts/histogramQuantile"
+            ),
+        ],
+    )
+    if exit_code != 0:
+        raise RuntimeError(f"Failed to install histogramQuantile binary: {output.decode()}")
+
+
+def create_clickhouse(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    cache_key: str = "clickhouse",
+    version: str | None = None,
+) -> types.TestContainerClickhouse:
+    coordinator = next(iter(keeper.container_configs.values()))
+
+    def create() -> types.TestContainerClickhouse:
+        clickhouse_version = version or request.config.getoption("--clickhouse-version")
+
+        container = ClickHouseContainer(
+            image=f"clickhouse/clickhouse-server:{clickhouse_version}",
+            port=9000,
+            username=CLICKHOUSE_USERNAME,
+            password=CLICKHOUSE_PASSWORD,
+        )
+
+        cluster_config = render_node_config(
+            keeper_address=coordinator.address,
+            keeper_port=coordinator.port,
+            shard="01",
+            remote_servers=render_remote_servers([("127.0.0.1", 9000)]),
+        )
+
+        tmp_dir = tmpfs(cache_key)
         cluster_config_file_path = os.path.join(tmp_dir, "cluster.xml")
         with open(cluster_config_file_path, "w", encoding="utf-8") as f:
             f.write(cluster_config)
 
         custom_function_file_path = os.path.join(tmp_dir, "custom-function.xml")
         with open(custom_function_file_path, "w", encoding="utf-8") as f:
-            f.write(custom_function_config)
+            f.write(CUSTOM_FUNCTION_CONFIG)
 
         container.with_volume_mapping(cluster_config_file_path, "/etc/clickhouse-server/config.d/cluster.xml")
         container.with_volume_mapping(
@@ -163,27 +240,7 @@ def clickhouse(
         container.with_network(network)
         container.start()
 
-        # Download and install the histogramQuantile binary
-        wrapped = container.get_wrapped_container()
-        exit_code, output = wrapped.exec_run(
-            [
-                "bash",
-                "-c",
-                (
-                    'version="v0.0.1" && '
-                    'node_os=$(uname -s | tr "[:upper:]" "[:lower:]") && '
-                    "node_arch=$(uname -m | sed s/aarch64/arm64/ | sed s/x86_64/amd64/) && "
-                    "cd /tmp && "
-                    'wget -O histogram-quantile.tar.gz "https://github.com/SigNoz/signoz/releases/download/histogram-quantile%2F${version}/histogram-quantile_${node_os}_${node_arch}.tar.gz" && '
-                    "tar -xzf histogram-quantile.tar.gz && "
-                    "mkdir -p /var/lib/clickhouse/user_scripts && "
-                    "mv histogram-quantile /var/lib/clickhouse/user_scripts/histogramQuantile && "
-                    "chmod +x /var/lib/clickhouse/user_scripts/histogramQuantile"
-                ),
-            ],
-        )
-        if exit_code != 0:
-            raise RuntimeError(f"Failed to install histogramQuantile binary: {output.decode()}")
+        install_histogram_quantile(container)
 
         connection = clickhouse_connect.get_client(
             user=container.username,
@@ -253,7 +310,7 @@ def clickhouse(
     return reuse.wrap(
         request,
         pytestconfig,
-        "clickhouse",
+        cache_key,
         empty=lambda: types.TestContainerSQL(
             container=types.TestContainerDocker(id="", host_configs={}, container_configs={}),
             conn=None,
@@ -265,6 +322,212 @@ def clickhouse(
     )
 
 
+@pytest.fixture(name="clickhouse", scope="package")
+def clickhouse(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerClickhouse:
+    """
+    Package-scoped fixture for Clickhouse TestContainer.
+    """
+    return create_clickhouse(
+        tmpfs=tmpfs,
+        network=network,
+        keeper=keeper,
+        request=request,
+        pytestconfig=pytestconfig,
+    )
+
+
+@pytest.fixture(name="clickhouse_node_conns", scope="function")
+def clickhouse_node_conns(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[list[clickhouse_connect.driver.client.Client], Any]:
+    """Per-node clients (index 0 = the initiator) for asserting shard-local
+    state via the local, non-distributed tables. Empty for single-node
+    fixtures, which don't populate `nodes`."""
+    conns = [
+        clickhouse_connect.get_client(
+            user=clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_USERNAME"],
+            password=clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_PASSWORD"],
+            host=node.host_configs["8123"].address,
+            port=node.host_configs["8123"].port,
+        )
+        for node in clickhouse.nodes
+    ]
+    yield conns
+    for conn in conns:
+        conn.close()
+
+
+def create_clickhouse_cluster(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    cache_key: str = "clickhouse_cluster",
+    shards: int = 2,
+    version: str | None = None,
+) -> types.TestContainerClickhouse:
+    """
+    To some extent, taken inspiration from how ClickHouse's own integration
+    harness composes real clusters: deterministic hostnames
+    (network aliases), per-node shard macros, and a shared cluster definition
+    named `cluster`.
+
+    `conn`/`env` point at node 1 i.e the initiator every query-service query and
+    migration goes through. Per-node containers are exposed via `nodes` so
+    tests can assert shard-local state.
+    """
+    coordinator = next(iter(keeper.container_configs.values()))
+
+    def create() -> types.TestContainerClickhouse:
+        clickhouse_version = version or request.config.getoption("--clickhouse-version")
+
+        # Unique aliases per creation: docker allows duplicate network aliases
+        # (DNS round-robin), so a stale cluster must never share names with a
+        # fresh one.
+        suffix = uuid4().hex[:6]
+        aliases = [f"signoz-ch-{suffix}-{i:02d}" for i in range(1, shards + 1)]
+        remote_servers = render_remote_servers([(alias, 9000) for alias in aliases], secret=cache_key)
+        # Own DDL queue path: the keeper instance may be shared with other
+        # environments under --reuse; its DDL queue stays separate.
+        distributed_ddl_path = f"/clickhouse/{cache_key}-{suffix}/task_queue/ddl"
+
+        nodes: list[types.TestContainerDocker] = []
+        started: list[ClickHouseContainer] = []
+        try:
+            for i, alias in enumerate(aliases, start=1):
+                node_config = render_node_config(
+                    keeper_address=coordinator.address,
+                    keeper_port=coordinator.port,
+                    shard=f"{i:02d}",
+                    remote_servers=remote_servers,
+                    distributed_ddl_path=distributed_ddl_path,
+                )
+
+                tmp_dir = tmpfs(f"clickhouse-{suffix}-{i:02d}")
+                cluster_config_file_path = os.path.join(tmp_dir, "cluster.xml")
+                with open(cluster_config_file_path, "w", encoding="utf-8") as f:
+                    f.write(node_config)
+                custom_function_file_path = os.path.join(tmp_dir, "custom-function.xml")
+                with open(custom_function_file_path, "w", encoding="utf-8") as f:
+                    f.write(CUSTOM_FUNCTION_CONFIG)
+                users_config_file_path = os.path.join(tmp_dir, "users.xml")
+                with open(users_config_file_path, "w", encoding="utf-8") as f:
+                    f.write(CLUSTER_USERS_CONFIG)
+
+                container = ClickHouseContainer(
+                    image=f"clickhouse/clickhouse-server:{clickhouse_version}",
+                    port=9000,
+                    username=CLICKHOUSE_USERNAME,
+                    password=CLICKHOUSE_PASSWORD,
+                )
+                container.with_volume_mapping(cluster_config_file_path, "/etc/clickhouse-server/config.d/cluster.xml")
+                container.with_volume_mapping(custom_function_file_path, "/etc/clickhouse-server/custom-function.xml")
+                container.with_volume_mapping(users_config_file_path, "/etc/clickhouse-server/users.d/integration-cluster.xml")
+                container.with_network(network)
+                container.with_network_aliases(alias)
+                container.start()
+                started.append(container)
+
+                install_histogram_quantile(container)
+
+                nodes.append(
+                    types.TestContainerDocker(
+                        id=container.get_wrapped_container().id,
+                        host_configs={
+                            "9000": types.TestContainerUrlConfig(
+                                "tcp",
+                                container.get_container_host_ip(),
+                                container.get_exposed_port(9000),
+                            ),
+                            "8123": types.TestContainerUrlConfig(
+                                "tcp",
+                                container.get_container_host_ip(),
+                                container.get_exposed_port(8123),
+                            ),
+                        },
+                        container_configs={
+                            "9000": types.TestContainerUrlConfig("tcp", alias, 9000),
+                            "8123": types.TestContainerUrlConfig("tcp", alias, 8123),
+                        },
+                    )
+                )
+        except Exception:
+            for container in started:
+                container.stop()
+            raise
+
+        connection = clickhouse_connect.get_client(
+            user=CLICKHOUSE_USERNAME,
+            password=CLICKHOUSE_PASSWORD,
+            host=nodes[0].host_configs["8123"].address,
+            port=nodes[0].host_configs["8123"].port,
+        )
+
+        return types.TestContainerClickhouse(
+            container=nodes[0],
+            conn=connection,
+            env={
+                "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN": f"tcp://{CLICKHOUSE_USERNAME}:{CLICKHOUSE_PASSWORD}@{aliases[0]}:{9000}",
+                "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_USERNAME": CLICKHOUSE_USERNAME,
+                "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_PASSWORD": CLICKHOUSE_PASSWORD,
+                "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER": "cluster",
+            },
+            nodes=nodes,
+        )
+
+    def delete(resource: types.TestContainerClickhouse) -> None:
+        client = docker.from_env()
+        for node in resource.nodes or [resource.container]:
+            try:
+                client.containers.get(container_id=node.id).stop()
+                client.containers.get(container_id=node.id).remove(v=True)
+            except docker.errors.NotFound:
+                logger.info(
+                    "Skipping removal of Clickhouse cluster node, node(%s) not found. Maybe it was manually removed?",
+                    {"id": node.id},
+                )
+
+    def restore(cache: dict) -> types.TestContainerClickhouse:
+        nodes = [types.TestContainerDocker.from_cache(node) for node in cache["nodes"]]
+        env = cache["env"]
+        host_config = nodes[0].host_configs["8123"]
+
+        conn = clickhouse_connect.get_client(
+            user=env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_USERNAME"],
+            password=env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_PASSWORD"],
+            host=host_config.address,
+            port=host_config.port,
+        )
+
+        return types.TestContainerClickhouse(
+            container=nodes[0],
+            conn=conn,
+            env=env,
+            nodes=nodes,
+        )
+
+    return reuse.wrap(
+        request,
+        pytestconfig,
+        cache_key,
+        empty=lambda: types.TestContainerClickhouse(
+            container=types.TestContainerDocker(id="", host_configs={}, container_configs={}),
+            conn=None,
+            env={},
+        ),
+        create=create,
+        delete=delete,
+        restore=restore,
+    )
+
+
 @pytest.fixture(name="check_query_log")
 def check_query_log(
     signoz: types.SigNoz,

tests/fixtures/keeper.py

@@ -0,0 +1,121 @@
+import os
+from collections.abc import Generator
+from typing import Any
+
+import docker
+import docker.errors
+import pytest
+from testcontainers.core.container import DockerContainer, Network
+
+from fixtures import reuse, types
+from fixtures.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+KEEPER_CONFIG = """
+<clickhouse>
+    <listen_host>0.0.0.0</listen_host>
+    <keeper_server>
+        <tcp_port>9181</tcp_port>
+        <server_id>1</server_id>
+        <log_storage_path>/var/lib/clickhouse-keeper/coordination/log</log_storage_path>
+        <snapshot_storage_path>/var/lib/clickhouse-keeper/coordination/snapshots</snapshot_storage_path>
+        <coordination_settings>
+            <operation_timeout_ms>10000</operation_timeout_ms>
+            <session_timeout_ms>30000</session_timeout_ms>
+            <raft_logs_level>warning</raft_logs_level>
+        </coordination_settings>
+        <raft_configuration>
+            <server>
+                <id>1</id>
+                <hostname>localhost</hostname>
+                <port>9234</port>
+            </server>
+        </raft_configuration>
+    </keeper_server>
+</clickhouse>
+"""
+
+
+def create_clickhouse_keeper(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    cache_key: str = "clickhousekeeper",
+    version: str | None = None,
+) -> types.TestContainerDocker:
+
+    def create() -> types.TestContainerDocker:
+        keeper_version = version or request.config.getoption("--clickhouse-version")
+
+        tmp_dir = tmpfs(cache_key)
+        keeper_config_file_path = os.path.join(tmp_dir, "keeper_config.xml")
+        with open(keeper_config_file_path, "w", encoding="utf-8") as f:
+            f.write(KEEPER_CONFIG)
+
+        container = DockerContainer(image=f"clickhouse/clickhouse-keeper:{keeper_version}")
+        container.with_volume_mapping(keeper_config_file_path, "/etc/clickhouse-keeper/keeper_config.xml")
+        container.with_exposed_ports(9181)
+        container.with_network(network=network)
+
+        container.start()
+        return types.TestContainerDocker(
+            id=container.get_wrapped_container().id,
+            host_configs={
+                "9181": types.TestContainerUrlConfig(
+                    scheme="tcp",
+                    address=container.get_container_host_ip(),
+                    port=container.get_exposed_port(9181),
+                )
+            },
+            container_configs={
+                "9181": types.TestContainerUrlConfig(
+                    scheme="tcp",
+                    address=container.get_wrapped_container().name,
+                    port=9181,
+                )
+            },
+        )
+
+    def delete(container: types.TestContainerDocker):
+        client = docker.from_env()
+        try:
+            client.containers.get(container_id=container.id).stop()
+            client.containers.get(container_id=container.id).remove(v=True)
+        except docker.errors.NotFound:
+            logger.info(
+                "Skipping removal of ClickHouse Keeper, Keeper(%s) not found. Maybe it was manually removed?",
+                {"id": container.id},
+            )
+
+    def restore(cache: dict) -> types.TestContainerDocker:
+        return types.TestContainerDocker.from_cache(cache)
+
+    return reuse.wrap(
+        request,
+        pytestconfig,
+        cache_key,
+        lambda: types.TestContainerDocker(id="", host_configs={}, container_configs={}),
+        create,
+        delete,
+        restore,
+    )
+
+
+@pytest.fixture(name="keeper", scope="package")
+def keeper(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    """
+    Package-scoped fixture for ClickHouse Keeper TestContainer.
+    """
+    return create_clickhouse_keeper(
+        tmpfs=tmpfs,
+        network=network,
+        request=request,
+        pytestconfig=pytestconfig,
+    )

tests/fixtures/metricreduction.py

@@ -0,0 +1,83 @@
+import datetime
+from collections.abc import Sequence
+
+import clickhouse_connect.driver.client
+
+from fixtures.metrics import MetricsBufferSample, MetricsBufferTimeSeries
+
+
+def local_series_counts(
+    node_conns: list[clickhouse_connect.driver.client.Client],
+    table: str,
+    metric_name: str,
+) -> list[int]:
+    """Distinct series per node via the LOCAL (non-distributed) table."""
+    return [
+        int(
+            conn.query(
+                f"SELECT count(DISTINCT fingerprint) FROM signoz_metrics.{table} WHERE metric_name = %(metric_name)s",
+                parameters={"metric_name": metric_name},
+            ).result_rows[0][0]
+        )
+        for conn in node_conns
+    ]
+
+
+def assert_spans_shards(
+    node_conns: list[clickhouse_connect.driver.client.Client],
+    table: str,
+    metric_name: str,
+    total: int,
+) -> None:
+    """Guard for distributed tests: a green run on a cluster proves nothing
+    unless the seeded series actually landed on more than one shard."""
+    counts = local_series_counts(node_conns, table, metric_name)
+    assert sum(counts) == total, f"expected {total} series in {table} across shards, got {counts}"
+    assert min(counts) > 0, f"seeded series in {table} all landed on one shard: {counts}"
+
+
+def build_recent_gauge_data(
+    metric_name: str,
+    base_epoch: int,
+    services: Sequence[str],
+    pods_per_service: int,
+    minutes: int,
+    value: float = 1.0,
+) -> tuple[list[MetricsBufferTimeSeries], list[MetricsBufferSample]]:
+    """Collector-shaped buffer rows for a gauge under a reduction rule that
+    keeps `service`: per raw series a raw series row (is_reduced=false, full
+    labels, reduced_fingerprint -> group) plus the group's reduced series row
+    (is_reduced=true, kept labels), and one raw sample per series per minute
+    carrying both fingerprints. Returns (time_series, samples) for
+    insert_buffer_metrics."""
+    reduced_series = {
+        service: MetricsBufferTimeSeries(
+            metric_name=metric_name,
+            labels={"service": service},
+            timestamp=datetime.datetime.fromtimestamp(base_epoch, tz=datetime.UTC),
+            is_reduced=True,
+        )
+        for service in services
+    }
+    raw_series = [
+        MetricsBufferTimeSeries(
+            metric_name=metric_name,
+            labels={"service": service, "pod": f"pod-{service}-{i}"},
+            timestamp=datetime.datetime.fromtimestamp(base_epoch, tz=datetime.UTC),
+            reduced_fingerprint=reduced_series[service].fingerprint,
+        )
+        for service in services
+        for i in range(pods_per_service)
+    ]
+    samples = [
+        MetricsBufferSample(
+            metric_name=metric_name,
+            fingerprint=ts.fingerprint,
+            timestamp=datetime.datetime.fromtimestamp(base_epoch + minute * 60, tz=datetime.UTC),
+            value=value,
+            reduced_fingerprint=ts.reduced_fingerprint,
+        )
+        for ts in raw_series
+        for minute in range(minutes)
+    ]
+    return raw_series + list(reduced_series.values()), samples

tests/fixtures/metrics.py

@@ -11,6 +11,14 @@ import pytest
 from fixtures import types
 from fixtures.time import parse_timestamp
 
+_REDUCED_METRICS_TABLES_TO_TRUNCATE = [
+    "time_series_v4_reduced",
+    "samples_v4_reduced_last_60s",
+    "samples_v4_reduced_sum_60s",
+    "time_series_v4_buffer",
+    "samples_v4_buffer",
+]
+
 
 class MetricsTimeSeries(ABC):
     """Represents a row in the time_series_v4 table."""
@@ -28,7 +36,6 @@ class MetricsTimeSeries(ABC):
     attrs: dict[str, str]
     scope_attrs: dict[str, str]
     resource_attrs: dict[str, str]
-    __normalized: bool
 
     def __init__(
         self,
@@ -60,7 +67,6 @@ class MetricsTimeSeries(ABC):
         self.scope_attrs = scope_attrs
         self.resource_attrs = resource_attrs
         self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
-        self.__normalized = False
 
         # Calculate fingerprint from metric_name + labels
         fingerprint_str = metric_name + self.labels
@@ -81,7 +87,6 @@ class MetricsTimeSeries(ABC):
             self.attrs,
             self.scope_attrs,
             self.resource_attrs,
-            self.__normalized,
         ]
 
 
@@ -414,6 +419,263 @@ class Metrics(ABC):
         return metrics
 
 
+class MetricsReducedTimeSeries(ABC):
+    """Represents a row in the time_series_v4_reduced table i.e what
+    the time_series_v4_reduced_mv materializes for a metric under a
+    reduction rule. One row per kept-label group. `fingerprint` holds the
+    reduced fingerprint and `labels` contains only the kept labels.
+
+    The fingerprint recipe (md5, like MetricsTimeSeries) does not match the
+    collector's real hash; it only needs to be consistent with the
+    reduced_fingerprint used in the reduced samples rows.
+    """
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        kept_labels: dict[str, str],
+        timestamp: datetime.datetime,
+        temporality: str = "Unspecified",
+        description: str = "",
+        unit: str = "",
+        type_: str = "Gauge",
+        is_monotonic: bool = False,
+        env: str = "default",
+    ) -> None:
+        kept_labels = dict(kept_labels)
+        kept_labels["__name__"] = metric_name
+        self.env = env
+        # mirror time_series_v4_reduced_mv: monotonic cumulative counters are
+        # reduced as deltas
+        if temporality == "Cumulative" and is_monotonic:
+            temporality = "Delta"
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.description = description
+        self.unit = unit
+        self.type = type_
+        self.is_monotonic = is_monotonic
+        self.labels = json.dumps(kept_labels, separators=(",", ":"))
+        self.attrs = kept_labels
+        self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3) // 3600000 * 3600000)
+
+        fingerprint_str = metric_name + self.labels
+        self.fingerprint = np.uint64(int(hashlib.md5(fingerprint_str.encode()).hexdigest()[:16], 16))
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.description,
+            self.unit,
+            self.type,
+            self.is_monotonic,
+            self.fingerprint,
+            self.unix_milli,
+            self.labels,
+            self.attrs,
+            {},
+            {},
+        ]
+
+
+class MetricsReducedSampleLast60s(ABC):
+    """Represents a row in the samples_v4_reduced_last_60s table. One 60s
+    bucket per reduced group, as the samples_v4_reduced_last_60s_mv refresh
+    would emit it (gauges and non-monotonic cumulative sums)."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        reduced_fingerprint: np.uint64,
+        timestamp: datetime.datetime,
+        sum_last: float,
+        min_value: float,
+        max_value: float,
+        sum_values: float,
+        count_series: int,
+        count_samples: int,
+        temporality: str = "Unspecified",
+        env: str = "default",
+        computed_at: datetime.datetime | None = None,
+    ) -> None:
+        self.env = env
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.reduced_fingerprint = reduced_fingerprint
+        # buckets are 60s-aligned: intDiv(unix_milli, 60000) * 60000
+        self.unix_milli = np.int64((int(timestamp.timestamp() * 1e3) // 60000) * 60000)
+        self.sum_last = np.float64(sum_last)
+        self.min = np.float64(min_value)
+        self.max = np.float64(max_value)
+        self.sum_values = np.float64(sum_values)
+        self.count_series = np.uint64(count_series)
+        self.count_samples = np.uint64(count_samples)
+        # the refresh stamps now(); default to shortly after the bucket closes
+        if computed_at is None:
+            computed_at = datetime.datetime.fromtimestamp(int(self.unix_milli) / 1e3, tz=datetime.UTC) + datetime.timedelta(seconds=180)
+        self.computed_at = computed_at
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.reduced_fingerprint,
+            self.unix_milli,
+            self.sum_last,
+            self.min,
+            self.max,
+            self.sum_values,
+            self.count_series,
+            self.count_samples,
+            self.computed_at,
+        ]
+
+
+class MetricsReducedSampleSum60s(ABC):
+    """Represents a row in the samples_v4_reduced_sum_60s table. One 60s
+    bucket per reduced group for delta counters and histograms."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        reduced_fingerprint: np.uint64,
+        timestamp: datetime.datetime,
+        sum_value: float,
+        count_series: int,
+        count_samples: int,
+        temporality: str = "Delta",
+        env: str = "default",
+        computed_at: datetime.datetime | None = None,
+    ) -> None:
+        self.env = env
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.reduced_fingerprint = reduced_fingerprint
+        self.unix_milli = np.int64((int(timestamp.timestamp() * 1e3) // 60000) * 60000)
+        self.sum = np.float64(sum_value)
+        self.count_series = np.uint64(count_series)
+        self.count_samples = np.uint64(count_samples)
+        if computed_at is None:
+            computed_at = datetime.datetime.fromtimestamp(int(self.unix_milli) / 1e3, tz=datetime.UTC) + datetime.timedelta(seconds=180)
+        self.computed_at = computed_at
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.reduced_fingerprint,
+            self.unix_milli,
+            self.sum,
+            self.count_series,
+            self.count_samples,
+            self.computed_at,
+        ]
+
+
+class MetricsBufferTimeSeries(ABC):
+    """Represents a row in the time_series_v4_buffer table. This is the collector's
+    universal landing target under cardinality control. For a ruled metric the
+    collector writes two rows per series: the raw one (is_reduced=false, full
+    labels, reduced_fingerprint pointing at its group) and the group's reduced
+    one (is_reduced=true, kept labels, fingerprint = reduced fingerprint)."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        labels: dict[str, str],
+        timestamp: datetime.datetime,
+        reduced_fingerprint: np.uint64 | int = 0,
+        is_reduced: bool = False,
+        temporality: str = "Unspecified",
+        description: str = "",
+        unit: str = "",
+        type_: str = "Gauge",
+        is_monotonic: bool = False,
+        env: str = "default",
+    ) -> None:
+        labels = dict(labels)
+        labels["__name__"] = metric_name
+        self.env = env
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.description = description
+        self.unit = unit
+        self.type = type_
+        self.is_monotonic = is_monotonic
+        self.reduced_fingerprint = np.uint64(reduced_fingerprint)
+        self.is_reduced = is_reduced
+        self.labels = json.dumps(labels, separators=(",", ":"))
+        self.attrs = labels
+        self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3) // 3600000 * 3600000)
+
+        fingerprint_str = metric_name + self.labels
+        self.fingerprint = np.uint64(int(hashlib.md5(fingerprint_str.encode()).hexdigest()[:16], 16))
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.description,
+            self.unit,
+            self.type,
+            self.is_monotonic,
+            self.fingerprint,
+            self.reduced_fingerprint,
+            self.is_reduced,
+            self.unix_milli,
+            self.labels,
+            self.attrs,
+            {},
+            {},
+        ]
+
+
+class MetricsBufferSample(ABC):
+    """Represents a row in the samples_v4_buffer table. Ruled samples carry
+    the raw fingerprint plus the group's reduced_fingerprint; unruled samples
+    have reduced_fingerprint = 0."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        fingerprint: np.uint64,
+        timestamp: datetime.datetime,
+        value: float,
+        reduced_fingerprint: np.uint64 | int = 0,
+        is_monotonic: bool = False,
+        temporality: str = "Unspecified",
+        env: str = "default",
+        flags: int = 0,
+    ) -> None:
+        self.env = env
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.fingerprint = fingerprint
+        self.reduced_fingerprint = np.uint64(reduced_fingerprint)
+        self.is_monotonic = is_monotonic
+        self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
+        self.value = np.float64(value)
+        self.flags = np.uint32(flags)
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.fingerprint,
+            self.reduced_fingerprint,
+            self.is_monotonic,
+            self.unix_milli,
+            self.value,
+            self.flags,
+        ]
+
+
 def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
     """
     Insert metrics into ClickHouse tables.
@@ -449,7 +711,6 @@ def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
                 "attrs",
                 "scope_attrs",
                 "resource_attrs",
-                "__normalized",
             ],
             data=[ts.to_row() for ts in time_series_map.values()],
         )
@@ -576,6 +837,161 @@ def insert_metrics(
     )
 
 
+def insert_reduced_metrics_to_clickhouse(
+    conn,
+    time_series: list[MetricsReducedTimeSeries],
+    last_samples: list[MetricsReducedSampleLast60s] | None = None,
+    sum_samples: list[MetricsReducedSampleSum60s] | None = None,
+) -> None:
+    """Insert reduced series into distributed_time_series_v4_reduced and 60s
+    buckets into the reduced samples tables. These tables exist only when
+    the schema migrator version includes the metrics cardinality-control
+    migration."""
+    if time_series:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_time_series_v4_reduced",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "description",
+                "unit",
+                "type",
+                "is_monotonic",
+                "fingerprint",
+                "unix_milli",
+                "labels",
+                "attrs",
+                "scope_attrs",
+                "resource_attrs",
+            ],
+            data=[ts.to_row() for ts in time_series],
+        )
+
+    if last_samples:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_samples_v4_reduced_last_60s",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "reduced_fingerprint",
+                "unix_milli",
+                "sum_last",
+                "min",
+                "max",
+                "sum_values",
+                "count_series",
+                "count_samples",
+                "computed_at",
+            ],
+            data=[sample.to_row() for sample in last_samples],
+        )
+
+    if sum_samples:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_samples_v4_reduced_sum_60s",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "reduced_fingerprint",
+                "unix_milli",
+                "sum",
+                "count_series",
+                "count_samples",
+                "computed_at",
+            ],
+            data=[sample.to_row() for sample in sum_samples],
+        )
+
+
+def insert_buffer_metrics_to_clickhouse(
+    conn,
+    time_series: list[MetricsBufferTimeSeries],
+    samples: list[MetricsBufferSample],
+) -> None:
+    if time_series:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_time_series_v4_buffer",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "description",
+                "unit",
+                "type",
+                "is_monotonic",
+                "fingerprint",
+                "reduced_fingerprint",
+                "is_reduced",
+                "unix_milli",
+                "labels",
+                "attrs",
+                "scope_attrs",
+                "resource_attrs",
+            ],
+            data=[ts.to_row() for ts in time_series],
+        )
+
+    if samples:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_samples_v4_buffer",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "fingerprint",
+                "reduced_fingerprint",
+                "is_monotonic",
+                "unix_milli",
+                "value",
+                "flags",
+            ],
+            data=[sample.to_row() for sample in samples],
+        )
+
+
+@pytest.fixture(name="insert_reduced_metrics", scope="function")
+def insert_reduced_metrics(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[Callable[..., None], Any]:
+    def _insert_reduced_metrics(
+        time_series: list[MetricsReducedTimeSeries],
+        last_samples: list[MetricsReducedSampleLast60s] | None = None,
+        sum_samples: list[MetricsReducedSampleSum60s] | None = None,
+    ) -> None:
+        insert_reduced_metrics_to_clickhouse(clickhouse.conn, time_series, last_samples, sum_samples)
+
+    yield _insert_reduced_metrics
+
+    cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
+    for table in _REDUCED_METRICS_TABLES_TO_TRUNCATE:
+        clickhouse.conn.query(f"TRUNCATE TABLE signoz_metrics.{table} ON CLUSTER '{cluster}' SYNC")
+
+
+@pytest.fixture(name="insert_buffer_metrics", scope="function")
+def insert_buffer_metrics(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[Callable[..., None], Any]:
+    def _insert_buffer_metrics(
+        time_series: list[MetricsBufferTimeSeries],
+        samples: list[MetricsBufferSample],
+    ) -> None:
+        insert_buffer_metrics_to_clickhouse(clickhouse.conn, time_series, samples)
+
+    yield _insert_buffer_metrics
+
+    cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
+    for table in _REDUCED_METRICS_TABLES_TO_TRUNCATE:
+        clickhouse.conn.query(f"TRUNCATE TABLE signoz_metrics.{table} ON CLUSTER '{cluster}' SYNC")
+
+
 @pytest.fixture(name="remove_metrics_ttl_and_storage_settings", scope="function")
 def remove_metrics_ttl_and_storage_settings(signoz: types.SigNoz):
     """

tests/fixtures/migrator.py

@@ -8,27 +8,30 @@ from fixtures.logger import setup_logger
 logger = setup_logger(__name__)
 
 
-def create_migrator(
+def create_migrator(  # pylint: disable=too-many-arguments,too-many-positional-arguments
     network: Network,
     clickhouse: types.TestContainerClickhouse,
     request: pytest.FixtureRequest,
     pytestconfig: pytest.Config,
     cache_key: str = "migrator",
     env_overrides: dict | None = None,
+    version: str | None = None,
 ) -> types.Operation:
     """
     Factory function for running schema migrations.
-    Accepts optional env_overrides to customize the migrator environment.
+    Accepts optional env_overrides to customize the migrator environment, and
+    an optional version to pin a schema-migrator release different from the
+    --schema-migrator-version option.
     """
 
     def create() -> None:
-        version = request.config.getoption("--schema-migrator-version")
+        migrator_version = version or request.config.getoption("--schema-migrator-version")
         client = docker.from_env()
 
         environment = dict(env_overrides) if env_overrides else {}
 
         container = client.containers.run(
-            image=f"signoz/signoz-schema-migrator:{version}",
+            image=f"signoz/signoz-schema-migrator:{migrator_version}",
             command=f"sync --replication=true --cluster-name=cluster --up= --dsn={clickhouse.env['SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN']}",
             detach=True,
             auto_remove=False,
@@ -47,7 +50,7 @@ def create_migrator(
         container.remove()
 
         container = client.containers.run(
-            image=f"signoz/signoz-schema-migrator:{version}",
+            image=f"signoz/signoz-schema-migrator:{migrator_version}",
             command=f"async --replication=true --cluster-name=cluster --up= --dsn={clickhouse.env['SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN']}",
             detach=True,
             auto_remove=False,

tests/fixtures/querier.py

@@ -189,6 +189,38 @@ def make_query_request(
     )
 
 
+def aligned_epoch(ago: timedelta, step_seconds: int = DEFAULT_STEP_INTERVAL) -> int:
+    """Epoch seconds for `now - ago`, floored to a step boundary so seeded
+    points land exactly on the query's toStartOfInterval buckets."""
+    epoch = (int((datetime.now(tz=UTC) - ago).timestamp()) // step_seconds) * step_seconds
+    if epoch % 3600 == 0:
+        epoch += step_seconds
+    return epoch
+
+
+def query_metric_values(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    token: str,
+    metric_name: str,
+    start_epoch: int,
+    end_epoch: int,
+    time_agg: str,
+    space_agg: str,
+    step_interval: int = DEFAULT_STEP_INTERVAL,
+) -> list[dict]:
+    """Run a single metrics builder query over [start_epoch, end_epoch) in
+    epoch seconds and return its series values sorted by timestamp."""
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=start_epoch * 1000,
+        end_ms=end_epoch * 1000,
+        queries=[build_builder_query("A", metric_name, time_agg, space_agg, step_interval=step_interval)],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    return sorted(get_series_values(response.json(), "A"), key=lambda v: v["timestamp"])
+
+
 def build_builder_query(
     name: str,
     metric_name: str,

tests/fixtures/types.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Literal
 from urllib.parse import urljoin
 
@@ -84,11 +84,16 @@ class TestContainerClickhouse:
     container: TestContainerDocker
     conn: clickhouse_connect.driver.client.Client
     env: dict[str, str]
+    # Per-node containers when running a multi-node cluster. Empty for the
+    # default single-node setup; nodes[0] is the node `conn`/`env` point at
+    # (the initiator every query goes through).
+    nodes: list[TestContainerDocker] = field(default_factory=list)
 
     def __cache__(self) -> dict:
         return {
             "container": self.container.__cache__(),
             "env": self.env,
+            "nodes": [node.__cache__() for node in self.nodes],
         }
 
     def __log__(self) -> str:

tests/fixtures/zookeeper.py

@@ -1,67 +0,0 @@
-import docker
-import docker.errors
-import pytest
-from testcontainers.core.container import DockerContainer, Network
-
-from fixtures import reuse, types
-from fixtures.logger import setup_logger
-
-logger = setup_logger(__name__)
-
-
-@pytest.fixture(name="zookeeper", scope="package")
-def zookeeper(network: Network, request: pytest.FixtureRequest, pytestconfig: pytest.Config) -> types.TestContainerDocker:
-    """
-    Package-scoped fixture for Zookeeper TestContainer.
-    """
-
-    def create() -> types.TestContainerDocker:
-        version = request.config.getoption("--zookeeper-version")
-
-        container = DockerContainer(image=f"signoz/zookeeper:{version}")
-        container.with_env("ALLOW_ANONYMOUS_LOGIN", "yes")
-        container.with_exposed_ports(2181)
-        container.with_network(network=network)
-
-        container.start()
-        return types.TestContainerDocker(
-            id=container.get_wrapped_container().id,
-            host_configs={
-                "2181": types.TestContainerUrlConfig(
-                    scheme="tcp",
-                    address=container.get_container_host_ip(),
-                    port=container.get_exposed_port(2181),
-                )
-            },
-            container_configs={
-                "2181": types.TestContainerUrlConfig(
-                    scheme="tcp",
-                    address=container.get_wrapped_container().name,
-                    port=2181,
-                )
-            },
-        )
-
-    def delete(container: types.TestContainerDocker):
-        client = docker.from_env()
-        try:
-            client.containers.get(container_id=container.id).stop()
-            client.containers.get(container_id=container.id).remove(v=True)
-        except docker.errors.NotFound:
-            logger.info(
-                "Skipping removal of Zookeeper, Zookeeper(%s) not found. Maybe it was manually removed?",
-                {"id": container.id},
-            )
-
-    def restore(cache: dict) -> types.TestContainerDocker:
-        return types.TestContainerDocker.from_cache(cache)
-
-    return reuse.wrap(
-        request,
-        pytestconfig,
-        "zookeeper",
-        lambda: types.TestContainerDocker(id="", host_configs={}, container_configs={}),
-        create,
-        delete,
-        restore,
-    )

tests/integration/testdata/alerts/test_scenarios/threshold_above_at_least_once/rule.json

@@ -25,7 +25,7 @@
           "type": "clickhouse_sql",
           "spec": {
             "name": "A",
-            "query": "WITH __temporal_aggregation_cte AS (\n SELECT \n fingerprint, \n toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(60)) AS ts, \n avg(value) AS per_series_value \n FROM signoz_metrics.distributed_samples_v4 AS points \n INNER JOIN (\n SELECT fingerprint \n FROM signoz_metrics.time_series_v4 \n WHERE metric_name IN ('request_total_threshold_above_at_least_once') \n AND LOWER(temporality) LIKE LOWER('cumulative') \n AND __normalized = false \n GROUP BY fingerprint\n ) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint \n WHERE metric_name IN ('request_total_threshold_above_at_least_once') \n AND unix_milli >= $start_timestamp_ms \n AND unix_milli < $end_timestamp_ms \n GROUP BY fingerprint, ts \n ORDER BY fingerprint, ts\n), \n__spatial_aggregation_cte AS (\n SELECT \n ts, \n sum(per_series_value) AS value \n FROM __temporal_aggregation_cte \n WHERE isNaN(per_series_value) = 0 \n GROUP BY ts\n) \nSELECT * FROM __spatial_aggregation_cte \nORDER BY ts"
+            "query": "WITH __temporal_aggregation_cte AS (\n SELECT \n fingerprint, \n toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(60)) AS ts, \n avg(value) AS per_series_value \n FROM signoz_metrics.distributed_samples_v4 AS points \n INNER JOIN (\n SELECT fingerprint \n FROM signoz_metrics.time_series_v4 \n WHERE metric_name IN ('request_total_threshold_above_at_least_once') \n AND LOWER(temporality) LIKE LOWER('cumulative') \n GROUP BY fingerprint\n ) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint \n WHERE metric_name IN ('request_total_threshold_above_at_least_once') \n AND unix_milli >= $start_timestamp_ms \n AND unix_milli < $end_timestamp_ms \n GROUP BY fingerprint, ts \n ORDER BY fingerprint, ts\n), \n__spatial_aggregation_cte AS (\n SELECT \n ts, \n sum(per_series_value) AS value \n FROM __temporal_aggregation_cte \n WHERE isNaN(per_series_value) = 0 \n GROUP BY ts\n) \nSELECT * FROM __spatial_aggregation_cte \nORDER BY ts"
           }
         }
       ]

tests/integration/tests/clickhousecluster/01_basic_cluster.py

@@ -0,0 +1,52 @@
+import clickhouse_connect.driver.client
+
+from fixtures import types
+
+TOTAL_ROWS = 64
+
+
+def test_topology(
+    clickhouse: types.TestContainerClickhouse,
+    clickhouse_node_conns: list[clickhouse_connect.driver.client.Client],
+) -> None:
+    aliases = {node.container_configs["9000"].address for node in clickhouse.nodes}
+
+    # Every node sees the same 2-shard cluster definition and identifies
+    # exactly itself as the local replica
+
+    for i, conn in enumerate(clickhouse_node_conns, start=1):
+        rows = conn.query("SELECT shard_num, host_name, is_local FROM system.clusters WHERE cluster = 'cluster' ORDER BY shard_num").result_rows
+        assert [row[0] for row in rows] == [1, 2], f"node {i}: expected 2 shards, got {rows}"
+        assert {row[1] for row in rows} == aliases, f"node {i}: cluster hosts {rows} != node aliases {aliases}"
+        local = [row[0] for row in rows if row[2]]
+        assert local == [i], f"node {i}: expected to be local for shard {i} only, got {local}"
+
+
+def test_replicated_distributed_round_trip(
+    clickhouse: types.TestContainerClickhouse,
+    clickhouse_node_conns: list[clickhouse_connect.driver.client.Client],
+) -> None:
+    # ON CLUSTER DDL reaches both nodes, Replicated engines register with the
+    # keeper via per-node macros, and a sharded Distributed insert scatters rows
+    # across shards while the distributed read returns the union.
+    conn = clickhouse.conn
+    try:
+        conn.query("CREATE DATABASE IF NOT EXISTS it_cluster ON CLUSTER 'cluster'")
+        conn.query("CREATE TABLE it_cluster.events ON CLUSTER 'cluster' (id UInt64, payload String) ENGINE = ReplicatedMergeTree ORDER BY id")
+        conn.query("CREATE TABLE it_cluster.distributed_events ON CLUSTER 'cluster' AS it_cluster.events ENGINE = Distributed('cluster', 'it_cluster', 'events', cityHash64(id))")
+
+        conn.insert(
+            database="it_cluster",
+            table="distributed_events",
+            column_names=["id", "payload"],
+            data=[[i, f"payload-{i:03d}"] for i in range(TOTAL_ROWS)],
+        )
+
+        distributed_count = int(conn.query("SELECT count() FROM it_cluster.distributed_events").result_rows[0][0])
+        assert distributed_count == TOTAL_ROWS
+
+        local_counts = [int(node_conn.query("SELECT count() FROM it_cluster.events").result_rows[0][0]) for node_conn in clickhouse_node_conns]
+        assert sum(local_counts) == TOTAL_ROWS, f"local counts {local_counts} do not add up to {TOTAL_ROWS}"
+        assert min(local_counts) > 0, f"all rows landed on one shard: {local_counts}"
+    finally:
+        conn.query("DROP DATABASE IF EXISTS it_cluster ON CLUSTER 'cluster' SYNC")

tests/integration/tests/clickhousecluster/conftest.py

@@ -0,0 +1,48 @@
+from collections.abc import Generator
+from typing import Any
+
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.clickhouse import create_clickhouse_cluster
+from fixtures.keeper import create_clickhouse_keeper
+
+CLICKHOUSE_VERSION = "25.12.5"
+
+
+@pytest.fixture(name="keeper", scope="package")
+def keeper_cluster(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    return create_clickhouse_keeper(
+        tmpfs=tmpfs,
+        network=network,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="keeper_cluster",
+        version=CLICKHOUSE_VERSION,
+    )
+
+
+@pytest.fixture(name="clickhouse", scope="package")
+def clickhouse_cluster(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerClickhouse:
+    return create_clickhouse_cluster(
+        tmpfs=tmpfs,
+        network=network,
+        keeper=keeper,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="clickhouse_cluster",
+        shards=2,
+        version=CLICKHOUSE_VERSION,
+    )

tests/integration/tests/metricreduction/01_reduced_gauge.py

@@ -0,0 +1,203 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import clickhouse_connect.driver.client
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metricreduction import assert_spans_shards
+from fixtures.metrics import (
+    Metrics,
+    MetricsReducedSampleLast60s,
+    MetricsReducedTimeSeries,
+)
+from fixtures.querier import aligned_epoch, query_metric_values
+
+
+def test_query_spanning_rule_activation_combines_raw_and_reduced_data(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+    insert_reduced_metrics: Callable[..., None],
+    clickhouse_node_conns: list[clickhouse_connect.driver.client.Client],
+) -> None:
+    """Before a reduction rule activates, data lives in the raw tables; after,
+    only the reduced tables have data. A single query spanning the activation
+    time must return one continuous series with no gap and no double counting:
+    32 raw series at 2.0 collapse into 16 groups whose per-minute total is
+    4.0, so the summed value stays 320 per step on both sides. Enough series
+    are seeded that both shards hold data (checked below), so correct totals
+    also prove the queries read every shard."""
+    metric_name = "test_reduction_activation_boundary"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+    services = [f"svc-{i:02d}" for i in range(16)]
+
+    # first 30 minutes: raw data (2 pods per service, one sample per minute)
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": service, "pod": f"{service}-pod-{pod}"},
+                timestamp=datetime.fromtimestamp(base_epoch + minute * 60, tz=UTC),
+                value=2.0,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for service in services
+            for pod in range(2)
+            for minute in range(30)
+        ]
+    )
+
+    # next 30 minutes: reduced data only (one row per service per minute)
+    time_series = [
+        MetricsReducedTimeSeries(
+            metric_name=metric_name,
+            kept_labels={"service": service},
+            timestamp=datetime.fromtimestamp(base_epoch + 30 * 60, tz=UTC),
+        )
+        for service in services
+    ]
+    insert_reduced_metrics(
+        time_series,
+        [
+            MetricsReducedSampleLast60s(
+                metric_name=metric_name,
+                reduced_fingerprint=ts.fingerprint,
+                timestamp=datetime.fromtimestamp(base_epoch + (30 + minute) * 60, tz=UTC),
+                sum_last=4.0,
+                min_value=2.0,
+                max_value=2.0,
+                sum_values=4.0,
+                count_series=2,
+                count_samples=2,
+            )
+            for ts in time_series
+            for minute in range(30)
+        ],
+    )
+
+    assert_spans_shards(clickhouse_node_conns, "time_series_v4", metric_name, total=len(services) * 2)
+    assert_spans_shards(clickhouse_node_conns, "time_series_v4_reduced", metric_name, total=len(services))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + 3600, "sum", "sum", step_interval=300)
+
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(12)]
+    assert [v["value"] for v in values] == [320.0] * 12
+
+
+@pytest.mark.parametrize(
+    "space_agg, expected",
+    [
+        ("sum", 12.0),  # sum_last: 4 + 8
+        ("avg", 3.0),  # sum(sum_last) / sum(count_series): 12 / 4
+        ("min", 1.0),  # min(min)
+        ("max", 6.0),  # max(max)
+    ],
+)
+def test_aggregations_across_series(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_reduced_metrics: Callable[..., None],
+    space_agg: str,
+    expected: float,
+) -> None:
+    """Aggregating across series reads the pre-aggregated reduced columns:
+    sum/avg from sum_last with the count_series weight, min/max from the
+    min/max columns."""
+    metric_name = f"test_reduction_across_series_{space_agg}"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+
+    groups = [
+        # (service, sum_last, min, max, count_series)
+        ("a", 4.0, 1.0, 3.0, 2),
+        ("b", 8.0, 2.0, 6.0, 2),
+    ]
+    time_series = {
+        service: MetricsReducedTimeSeries(
+            metric_name=metric_name,
+            kept_labels={"service": service},
+            timestamp=datetime.fromtimestamp(base_epoch, tz=UTC),
+        )
+        for service, _, _, _, _ in groups
+    }
+    insert_reduced_metrics(
+        list(time_series.values()),
+        [
+            MetricsReducedSampleLast60s(
+                metric_name=metric_name,
+                reduced_fingerprint=time_series[service].fingerprint,
+                timestamp=datetime.fromtimestamp(base_epoch + minute * 60, tz=UTC),
+                sum_last=sum_last,
+                min_value=min_value,
+                max_value=max_value,
+                sum_values=sum_last,
+                count_series=count_series,
+                count_samples=count_series,
+            )
+            for service, sum_last, min_value, max_value, count_series in groups
+            for minute in range(20)
+        ],
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + 20 * 60, "avg", space_agg, step_interval=300)
+
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(4)]
+    assert [v["value"] for v in values] == [expected] * 4
+
+
+def test_recomputed_minutes_use_only_the_newest_values(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_reduced_metrics: Callable[..., None],
+) -> None:
+    """The collector rewrites recent minutes on every refresh, so the same
+    minute exists multiple times with increasing computed_at. Queries must
+    count each minute once, using its newest version: write the same minutes
+    twice with different values and only the second write may show up."""
+    metric_name = "test_reduction_recompute"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+
+    time_series = [
+        MetricsReducedTimeSeries(
+            metric_name=metric_name,
+            kept_labels={"service": service},
+            timestamp=datetime.fromtimestamp(base_epoch, tz=UTC),
+        )
+        for service in ("a", "b")
+    ]
+
+    def minute_rows(sum_last: float, computed_at_offset_seconds: int) -> list[MetricsReducedSampleLast60s]:
+        return [
+            MetricsReducedSampleLast60s(
+                metric_name=metric_name,
+                reduced_fingerprint=ts.fingerprint,
+                timestamp=datetime.fromtimestamp(base_epoch + minute * 60, tz=UTC),
+                sum_last=sum_last,
+                min_value=sum_last,
+                max_value=sum_last,
+                sum_values=sum_last,
+                count_series=1,
+                count_samples=1,
+                computed_at=datetime.fromtimestamp(base_epoch + minute * 60 + computed_at_offset_seconds, tz=UTC),
+            )
+            for ts in time_series
+            for minute in range(10)
+        ]
+
+    # first write says 1.0; a later rewrite of the same minutes says 5.0
+    insert_reduced_metrics(time_series, minute_rows(sum_last=1.0, computed_at_offset_seconds=120))
+    insert_reduced_metrics(time_series, minute_rows(sum_last=5.0, computed_at_offset_seconds=180))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + 10 * 60, "sum", "sum", step_interval=300)
+
+    # 2 groups x 5 minutes x 5.0 per step; the 1.0 rows must not contribute
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(2)]
+    assert [v["value"] for v in values] == [50.0] * 2

tests/integration/tests/metricreduction/02_reduced_counter.py

@@ -0,0 +1,70 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metrics import (
+    MetricsReducedSampleSum60s,
+    MetricsReducedTimeSeries,
+)
+from fixtures.querier import aligned_epoch, query_metric_values
+
+
+@pytest.mark.parametrize(
+    "time_agg, expected",
+    [
+        # 2 groups x 5 minutes x 30.0 per 300s step
+        ("rate", 1.0),  # 300 / 300s
+        ("increase", 300.0),
+    ],
+)
+def test_counter_rate_and_increase(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_reduced_metrics: Callable[..., None],
+    time_agg: str,
+    expected: float,
+) -> None:
+    metric_name = f"test_reduction_counter_{time_agg}"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+
+    # monotonic cumulative counter: MetricsReducedTimeSeries mirrors the
+    # collector's temporality rewrite to Delta
+    time_series = [
+        MetricsReducedTimeSeries(
+            metric_name=metric_name,
+            kept_labels={"service": service},
+            timestamp=datetime.fromtimestamp(base_epoch, tz=UTC),
+            temporality="Cumulative",
+            type_="Sum",
+            is_monotonic=True,
+        )
+        for service in ("a", "b")
+    ]
+    assert all(ts.temporality == "Delta" for ts in time_series)
+
+    insert_reduced_metrics(
+        time_series,
+        sum_samples=[
+            MetricsReducedSampleSum60s(
+                metric_name=metric_name,
+                reduced_fingerprint=ts.fingerprint,
+                timestamp=datetime.fromtimestamp(base_epoch + minute * 60, tz=UTC),
+                sum_value=30.0,
+                count_series=2,
+                count_samples=2,
+                temporality="Delta",
+            )
+            for ts in time_series
+            for minute in range(20)
+        ],
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + 20 * 60, time_agg, "sum", step_interval=300)
+
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(4)]
+    assert [v["value"] for v in values] == [expected] * 4

tests/integration/tests/metricreduction/03_reduced_recent_window.py

@@ -0,0 +1,70 @@
+from collections.abc import Callable
+from datetime import timedelta
+from http import HTTPStatus
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metricreduction import build_recent_gauge_data
+from fixtures.querier import (
+    aligned_epoch,
+    build_builder_query,
+    get_all_series,
+    index_series_by_label,
+    make_query_request,
+    query_metric_values,
+)
+
+SERVICES = ("a", "b")
+PODS_PER_SERVICE = 2
+MINUTES = 20
+
+
+def test_recent_queries_return_full_resolution_totals(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_buffer_metrics: Callable[..., None],
+) -> None:
+    metric_name = "test_reduction_recent_totals"
+    # samples span [now-25m, now-5m); the query window sits inside the last 24h
+    base_epoch = aligned_epoch(timedelta(minutes=25), step_seconds=300)
+    insert_buffer_metrics(*build_recent_gauge_data(metric_name, base_epoch, SERVICES, PODS_PER_SERVICE, MINUTES))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + MINUTES * 60, "sum", "sum", step_interval=300)
+
+    # 4 raw series x 5 samples x 1.0 per step: full raw resolution, and the
+    # reduced series rows must not be counted (their fingerprints match no
+    # samples, and the time-series lookup filters them out)
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(4)]
+    assert [v["value"] for v in values] == [float(len(SERVICES) * PODS_PER_SERVICE * 5)] * 4
+
+
+def test_recent_queries_group_by_full_labels(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_buffer_metrics: Callable[..., None],
+) -> None:
+    """Group-by resolves against the raw buffer series rows (full labels), so
+    grouping by the kept label still sees every raw series underneath."""
+    metric_name = "test_reduction_recent_groupby"
+    base_epoch = aligned_epoch(timedelta(minutes=25), step_seconds=300)
+    insert_buffer_metrics(*build_recent_gauge_data(metric_name, base_epoch, SERVICES, PODS_PER_SERVICE, MINUTES))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=base_epoch * 1000,
+        end_ms=(base_epoch + MINUTES * 60) * 1000,
+        queries=[build_builder_query("A", metric_name, "sum", "sum", step_interval=300, group_by=["service"])],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    series_by_service = index_series_by_label(get_all_series(response.json(), "A"), "service")
+    assert set(series_by_service.keys()) == set(SERVICES)
+    for service in SERVICES:
+        values = sorted(series_by_service[service]["values"], key=lambda v: v["timestamp"])
+        # 2 pods x 5 samples x 1.0 per step
+        assert [v["value"] for v in values] == [float(PODS_PER_SERVICE * 5)] * 4

tests/integration/tests/metricreduction/conftest.py

@@ -0,0 +1,99 @@
+from collections.abc import Generator
+from typing import Any
+
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.auth import register_admin
+from fixtures.clickhouse import create_clickhouse_cluster
+from fixtures.keeper import create_clickhouse_keeper
+from fixtures.migrator import create_migrator
+from fixtures.signoz import create_signoz
+
+SCHEMA_MIGRATOR_VERSION = "v0.144.6-rc.2"
+CLICKHOUSE_VERSION = "25.12.5"
+
+
+@pytest.fixture(name="keeper", scope="package")
+def keeper_metricreduction(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    return create_clickhouse_keeper(
+        tmpfs=tmpfs,
+        network=network,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="keeper_metricreduction",
+        version=CLICKHOUSE_VERSION,
+    )
+
+
+@pytest.fixture(name="clickhouse", scope="package")
+def clickhouse_metricreduction(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerClickhouse:
+    return create_clickhouse_cluster(
+        tmpfs=tmpfs,
+        network=network,
+        keeper=keeper,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="clickhouse_metricreduction",
+        shards=2,
+        version=CLICKHOUSE_VERSION,
+    )
+
+
+@pytest.fixture(name="migrator", scope="package")
+def migrator_metricreduction(
+    network: Network,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.Operation:
+    return create_migrator(
+        network=network,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="migrator_metricreduction",
+        version=SCHEMA_MIGRATOR_VERSION,
+    )
+
+
+@pytest.fixture(name="signoz", scope="package")
+def signoz_metricreduction(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz_metricreduction",
+        env_overrides={
+            "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_ENABLE__METRICS__REDUCTION": True,
+        },
+    )
+
+
+@pytest.fixture(name="create_user_admin", scope="package")
+def create_user_admin_metricreduction(signoz: types.SigNoz, request: pytest.FixtureRequest, pytestconfig: pytest.Config) -> types.Operation:
+    return register_admin(signoz, request, pytestconfig, cache_key="create_user_admin_metricreduction")