chore: undo the changes to tests/fixtures/http.py

.github/workflows/integrationci.yaml

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

frontend/src/pages/DashboardPageV2/DashboardContainer/DashboardSettings/Variables/VariableForm/useVariableForm.ts

@@ -46,12 +46,8 @@ export interface UseVariableForm {
 	handleSave: () => void;
 }
 
-// `defaultValue` is a string | string[] on the wire; the editor uses a single
-// string, so take the first when it's an array.
-const readDefaultValue = (model: VariableFormModel): string => {
-	const dv = model.defaultValue;
-	return Array.isArray(dv) ? (dv[0] ?? '') : (dv ?? '');
-};
+const readDefaultValue = (model: VariableFormModel): string =>
+	((model.defaultValue as { value?: string })?.value ?? '') as string;
 
 /** Form state, derivations and handlers for the variable editor. */
 export function useVariableForm({

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/VariableSelector.tsx

@@ -18,22 +18,22 @@ interface VariableSelectorProps {
 	variable: VariableFormModel;
 	/** All variables (Dynamic uses them to scope options by sibling selections). */
 	variables: VariableFormModel[];
+	/** Names this variable depends on (for Query gating). */
+	parents: string[];
 	/** All current selections (Query passes them as the request payload). */
 	selections: VariableSelectionMap;
 	selection: VariableSelection;
 	onChange: (selection: VariableSelection) => void;
-	/** Batched fill applied when options resolve (Query/Dynamic auto-selection). */
-	onAutoSelect: (selection: VariableSelection) => void;
 }
 
 /** One labelled variable control; dispatches on the variable type. */
 function VariableSelector({
 	variable,
 	variables,
+	parents,
 	selections,
 	selection,
 	onChange,
-	onAutoSelect,
 }: VariableSelectorProps): JSX.Element {
 	const customOptions = useMemo(
 		() =>
@@ -61,10 +61,10 @@ function VariableSelector({
 				return (
 					<QuerySelector
 						variable={variable}
+						parents={parents}
 						selections={selections}
 						selection={selection}
 						onChange={onChange}
-						onAutoSelect={onAutoSelect}
 					/>
 				);
 			case 'DYNAMIC':
@@ -75,7 +75,6 @@ function VariableSelector({
 						selections={selections}
 						selection={selection}
 						onChange={onChange}
-						onAutoSelect={onAutoSelect}
 					/>
 				);
 			case 'CUSTOM':

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/VariablesBar.module.scss

@@ -112,13 +112,3 @@
 		box-shadow: none !important;
 	}
 }
-
-.overflowTooltip {
-	display: flex;
-	flex-direction: column;
-	gap: 2px;
-}
-
-.overflowName {
-	font-weight: var(--font-weight-medium);
-}

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/VariablesBar.tsx

@@ -1,33 +1,14 @@
 import { useState } from 'react';
 import { ChevronLeft } from '@signozhq/icons';
 import { Button } from '@signozhq/ui/button';
-import { TooltipSimple } from '@signozhq/ui/tooltip';
 import cx from 'classnames';
 import type { DashboardtypesGettableDashboardV2DTO } from 'api/generated/services/sigNoz.schemas';
 import { useInlineOverflowCount } from 'hooks/useInlineOverflowCount';
 
-import type { VariableSelection } from './selectionTypes';
 import { useVariableSelection } from './useVariableSelection';
 import VariableSelector from './VariableSelector';
 import styles from './VariablesBar.module.scss';
 
-// Short display of a variable's current selection, for the collapsed +N tooltip.
-function formatSelection(selection: VariableSelection | undefined): string {
-	if (!selection) {
-		return '—';
-	}
-	if (selection.allSelected) {
-		return 'ALL';
-	}
-	const { value } = selection;
-	if (Array.isArray(value)) {
-		return value.length > 0 ? value.join(', ') : '—';
-	}
-	return value === '' || value === null || value === undefined
-		? '—'
-		: String(value);
-}
-
 interface VariablesBarProps {
 	dashboard: DashboardtypesGettableDashboardV2DTO;
 }
@@ -42,7 +23,7 @@ interface VariablesBarProps {
  * either way so auto-selection and option fetching keep driving the panels.
  */
 function VariablesBar({ dashboard }: VariablesBarProps): JSX.Element | null {
-	const { variables, selection, setSelection, autoSelect } =
+	const { variables, dependencyData, selection, setSelection } =
 		useVariableSelection(dashboard);
 	const [expanded, setExpanded] = useState(false);
 	const { containerRef, visibleCount, overflowCount } = useInlineOverflowCount({
@@ -57,22 +38,6 @@ function VariablesBar({ dashboard }: VariablesBarProps): JSX.Element | null {
 	}
 
 	const hasOverflow = overflowCount > 0;
-	const hiddenVariables =
-		!expanded && hasOverflow ? variables.slice(visibleCount) : [];
-
-	const moreButton = (
-		<Button
-			variant="outlined"
-			color="secondary"
-			size="md"
-			prefix={expanded ? <ChevronLeft size={14} /> : undefined}
-			aria-expanded={expanded}
-			testId="dashboard-variables-more"
-			onClick={(): void => setExpanded((prev) => !prev)}
-		>
-			{expanded ? 'Less' : `+${overflowCount}`}
-		</Button>
-	);
 
 	return (
 		<div className={styles.bar} data-testid="dashboard-variables-bar">
@@ -92,6 +57,7 @@ function VariablesBar({ dashboard }: VariablesBarProps): JSX.Element | null {
 						<VariableSelector
 							variable={variable}
 							variables={variables}
+							parents={dependencyData.parentGraph[variable.name] ?? []}
 							selections={selection}
 							selection={
 								selection[variable.name] ?? {
@@ -100,32 +66,23 @@ function VariablesBar({ dashboard }: VariablesBarProps): JSX.Element | null {
 								}
 							}
 							onChange={(next): void => setSelection(variable.name, next)}
-							onAutoSelect={(next): void => autoSelect(variable.name, next)}
 						/>
 					</div>
 				))}
 
 				{hasOverflow && (
 					<span className={styles.moreButton}>
-						{expanded ? (
-							moreButton
-						) : (
-							<TooltipSimple
-								side="top"
-								title={
-									<div className={styles.overflowTooltip}>
-										{hiddenVariables.map((variable) => (
-											<div key={variable.name}>
-												<span className={styles.overflowName}>{variable.name}</span>:{' '}
-												{formatSelection(selection[variable.name])}
-											</div>
-										))}
-									</div>
-								}
-							>
-								{moreButton}
-							</TooltipSimple>
-						)}
+						<Button
+							variant="outlined"
+							color="secondary"
+							size="md"
+							prefix={expanded ? <ChevronLeft size={14} /> : undefined}
+							aria-expanded={expanded}
+							testId="dashboard-variables-more"
+							onClick={(): void => setExpanded((prev) => !prev)}
+						>
+							{expanded ? 'Less' : `+${overflowCount}`}
+						</Button>
 					</span>
 				)}
 			</div>

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/selectors/DynamicSelector.tsx

@@ -1,8 +1,7 @@
 import { useMemo } from 'react';
-import { useQuery } from 'react-query';
 // eslint-disable-next-line no-restricted-imports
 import { useSelector } from 'react-redux';
-import { getFieldValues } from 'api/dynamicVariables/getFieldValues';
+import { useGetFieldValues } from 'hooks/dynamicVariables/useGetFieldValues';
 import type { AppState } from 'store/reducers';
 import type { GlobalReducer } from 'types/reducer/globalTime';
 
@@ -11,14 +10,12 @@ import {
 	sortValuesByOrder,
 } from '../../DashboardSettings/Variables/variableFormModel';
 import type { VariableFormModel } from '../../DashboardSettings/Variables/variableFormModel';
-import { useDashboardStore } from '../../store/useDashboardStore';
 import { buildExistingDynamicVariableQuery } from '../dynamicFilter';
 import type {
 	VariableSelection,
 	VariableSelectionMap,
 } from '../selectionTypes';
 import { useAutoSelect } from '../useAutoSelect';
-import { useVariableFetchState } from '../useVariableFetchState';
 import ValueSelector from './ValueSelector';
 
 interface DynamicSelectorProps {
@@ -28,16 +25,12 @@ interface DynamicSelectorProps {
 	selections: VariableSelectionMap;
 	selection: VariableSelection;
 	onChange: (selection: VariableSelection) => void;
-	/** Batched auto-selection fill applied when options resolve. */
-	onAutoSelect: (selection: VariableSelection) => void;
 }
 
 /**
  * Dynamic-variable options sourced from live telemetry field values for the
  * chosen signal + attribute, scoped by the other dynamic variables' selections
- * (so e.g. `pod` narrows to the chosen `namespace`). WHEN to fetch is owned by
- * the runtime fetch engine: dynamics fetch together once the query variables have
- * values, and refetch (via a `cycleId` bump) whenever any variable value changes.
+ * (so e.g. `pod` narrows to the chosen `namespace`).
  */
 function DynamicSelector({
 	variable,
@@ -45,7 +38,6 @@ function DynamicSelector({
 	selections,
 	selection,
 	onChange,
-	onAutoSelect,
 }: DynamicSelectorProps): JSX.Element {
 	const { minTime, maxTime } = useSelector<AppState, GlobalReducer>(
 		(state) => state.globalTime,
@@ -56,51 +48,14 @@ function DynamicSelector({
 		[variables, selections, variable.name],
 	);
 
-	const {
-		variableFetchCycleId,
-		isVariableFetching,
-		isVariableSettled,
-		isVariableWaiting,
-		hasVariableFetchedOnce,
-	} = useVariableFetchState(variable.name);
-	const onVariableFetchComplete = useDashboardStore(
-		(s) => s.onVariableFetchComplete,
-	);
-	const onVariableFetchFailure = useDashboardStore(
-		(s) => s.onVariableFetchFailure,
-	);
-
-	const { data, isFetching } = useQuery(
-		[
-			'dashboard-variable-dynamic',
-			variable.name,
-			variable.dynamicSignal,
-			variable.dynamicAttribute,
-			existingQuery,
-			minTime,
-			maxTime,
-			variableFetchCycleId,
-		],
-		() =>
-			getFieldValues(
-				signalForApi(variable.dynamicSignal),
-				variable.dynamicAttribute,
-				undefined,
-				minTime,
-				maxTime,
-				existingQuery || undefined,
-			),
-		{
-			enabled:
-				!!variable.dynamicAttribute &&
-				(isVariableFetching || (isVariableSettled && hasVariableFetchedOnce)),
-			refetchOnWindowFocus: false,
-			onSettled: (_, error) =>
-				error
-					? onVariableFetchFailure(variable.name)
-					: onVariableFetchComplete(variable.name),
-		},
-	);
+	const { data, isFetching } = useGetFieldValues({
+		signal: signalForApi(variable.dynamicSignal),
+		name: variable.dynamicAttribute,
+		startUnixMilli: minTime,
+		endUnixMilli: maxTime,
+		existingQuery: existingQuery || undefined,
+		enabled: !!variable.dynamicAttribute,
+	});
 
 	const options = useMemo(() => {
 		const payload = data?.data;
@@ -109,14 +64,14 @@ function DynamicSelector({
 		return sortValuesByOrder(values, variable.sort).map(String);
 	}, [data, variable.sort]);
 
-	useAutoSelect(variable, options, selection, onAutoSelect);
+	useAutoSelect(variable, options, selection, onChange);
 
 	return (
 		<ValueSelector
 			options={options}
 			multiSelect={variable.multiSelect}
 			showAllOption={variable.showAllOption}
-			loading={isFetching || isVariableWaiting}
+			loading={isFetching}
 			selection={selection}
 			onChange={onChange}
 			testId={`variable-select-${variable.name}`}

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/selectors/QuerySelector.tsx

@@ -8,82 +8,58 @@ import type { GlobalReducer } from 'types/reducer/globalTime';
 
 import { sortValuesByOrder } from '../../DashboardSettings/Variables/variableFormModel';
 import type { VariableFormModel } from '../../DashboardSettings/Variables/variableFormModel';
-import { useDashboardStore } from '../../store/useDashboardStore';
 import type {
 	VariableSelection,
 	VariableSelectionMap,
 } from '../selectionTypes';
-import { selectionToPayload } from '../selectionUtils';
+import { isResolved, selectionToPayload } from '../selectionUtils';
 import { useAutoSelect } from '../useAutoSelect';
-import { useVariableFetchState } from '../useVariableFetchState';
 import ValueSelector from './ValueSelector';
 
 interface QuerySelectorProps {
 	variable: VariableFormModel;
+	/** Names this variable's query references; it waits until they're resolved. */
+	parents: string[];
 	/** All current selections, fed to the query as `{ name: value }`. */
 	selections: VariableSelectionMap;
 	selection: VariableSelection;
 	onChange: (selection: VariableSelection) => void;
-	/** Batched auto-selection fill applied when options resolve. */
-	onAutoSelect: (selection: VariableSelection) => void;
 }
 
 /**
- * Query-driven options. WHEN to fetch is owned by the runtime fetch engine
- * (`variableFetchSlice`): the query is `enabled` while this variable is fetching
- * (or settled-after-a-first-fetch, so a cycle bump re-runs it), and the engine's
- * per-variable `cycleId` keys the request — so a parent's value change refetches
- * only the dependent variables, in dependency order. The current selections feed
- * the request payload but are deliberately NOT in the key (V1 parity).
+ * Query-driven options. Dependency orchestration is declarative: the query is
+ * `enabled` only once every parent is resolved, and the parent values are in the
+ * query key — so it refetches automatically when a parent changes (and a cyclic
+ * dependency is simply never enabled).
  */
 function QuerySelector({
 	variable,
+	parents,
 	selections,
 	selection,
 	onChange,
-	onAutoSelect,
 }: QuerySelectorProps): JSX.Element {
 	const { minTime, maxTime } = useSelector<AppState, GlobalReducer>(
 		(state) => state.globalTime,
 	);
 	const payload = useMemo(() => selectionToPayload(selections), [selections]);
-
-	const {
-		variableFetchCycleId,
-		isVariableFetching,
-		isVariableSettled,
-		isVariableWaiting,
-		hasVariableFetchedOnce,
-	} = useVariableFetchState(variable.name);
-	const onVariableFetchComplete = useDashboardStore(
-		(s) => s.onVariableFetchComplete,
-	);
-	const onVariableFetchFailure = useDashboardStore(
-		(s) => s.onVariableFetchFailure,
-	);
+	const enabled = parents.every((parent) => isResolved(selections[parent]));
 
 	const { data, isFetching } = useQuery(
 		[
 			'dashboard-variable',
 			variable.name,
 			variable.queryValue,
+			payload,
 			minTime,
 			maxTime,
-			variableFetchCycleId,
 		],
 		() =>
 			dashboardVariablesQuery({
 				query: variable.queryValue,
 				variables: payload,
 			}),
-		{
-			enabled: isVariableFetching || (isVariableSettled && hasVariableFetchedOnce),
-			refetchOnWindowFocus: false,
-			onSettled: (_, error) =>
-				error
-					? onVariableFetchFailure(variable.name)
-					: onVariableFetchComplete(variable.name),
-		},
+		{ enabled, refetchOnWindowFocus: false },
 	);
 
 	const options = useMemo(() => {
@@ -96,14 +72,14 @@ function QuerySelector({
 		).map(String);
 	}, [data, variable.sort]);
 
-	useAutoSelect(variable, options, selection, onAutoSelect);
+	useAutoSelect(variable, options, selection, onChange);
 
 	return (
 		<ValueSelector
 			options={options}
 			multiSelect={variable.multiSelect}
 			showAllOption={variable.showAllOption}
-			loading={isFetching || isVariableWaiting}
+			loading={isFetching}
 			selection={selection}
 			onChange={onChange}
 			testId={`variable-select-${variable.name}`}

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/useAutoSelect.ts

@@ -12,7 +12,7 @@ export function useAutoSelect(
 	variable: VariableFormModel,
 	options: string[],
 	selection: VariableSelection,
-	onAutoSelect: (selection: VariableSelection) => void,
+	onChange: (selection: VariableSelection) => void,
 ): void {
 	useEffect(() => {
 		if (options.length === 0 || selection.allSelected) {
@@ -28,11 +28,11 @@ export function useAutoSelect(
 		if (isValid) {
 			return;
 		}
-		const dv = variable.defaultValue;
-		const fallback = Array.isArray(dv) ? dv[0] : dv;
+		const fallback = (variable.defaultValue as { value?: string } | undefined)
+			?.value;
 		const initial =
 			fallback && options.includes(fallback) ? fallback : options[0];
-		onAutoSelect({
+		onChange({
 			value: variable.multiSelect ? [initial] : initial,
 			allSelected: false,
 		});

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/useVariableFetchState.ts

@@ -1,44 +0,0 @@
-import {
-	selectVariableCycleId,
-	selectVariableFetchedOnce,
-	selectVariableFetchState,
-	type VariableFetchState,
-} from '../store/slices/variableFetchSlice';
-import { useDashboardStore } from '../store/useDashboardStore';
-
-export interface VariableFetchStateResult {
-	variableFetchState: VariableFetchState;
-	/** Include in the selector's react-query key to auto-cancel stale requests. */
-	variableFetchCycleId: number;
-	/** Actively fetching (first load or revalidating). */
-	isVariableFetching: boolean;
-	/** Stable — the fetch completed (or errored). */
-	isVariableSettled: boolean;
-	/** Blocked on parent dependencies (query order) or query variables (dynamics). */
-	isVariableWaiting: boolean;
-	/** Completed at least one fetch — keeps the query subscribed so a cycle bump refetches. */
-	hasVariableFetchedOnce: boolean;
-}
-
-/**
- * Per-variable view of the runtime fetch engine (`variableFetchSlice`), consumed
- * by the Query/Dynamic selectors to gate their fetch and key it by cycle id.
- * V2-native equivalent of V1's `useVariableFetchState`.
- */
-export function useVariableFetchState(name: string): VariableFetchStateResult {
-	const variableFetchState = useDashboardStore(selectVariableFetchState(name));
-	const variableFetchCycleId = useDashboardStore(selectVariableCycleId(name));
-	const hasVariableFetchedOnce = useDashboardStore(
-		selectVariableFetchedOnce(name),
-	);
-
-	return {
-		variableFetchState,
-		variableFetchCycleId,
-		isVariableFetching:
-			variableFetchState === 'loading' || variableFetchState === 'revalidating',
-		isVariableSettled: variableFetchState === 'idle',
-		isVariableWaiting: variableFetchState === 'waiting',
-		hasVariableFetchedOnce,
-	};
-}

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/useVariableSelection.ts

@@ -1,10 +1,6 @@
-import { useCallback, useEffect, useMemo, useRef } from 'react';
+import { useCallback, useEffect, useMemo } from 'react';
 import { parseAsJson, useQueryState } from 'nuqs';
-// eslint-disable-next-line no-restricted-imports -- global time selector still on redux
-import { useSelector } from 'react-redux';
 import type { DashboardtypesGettableDashboardV2DTO } from 'api/generated/services/sigNoz.schemas';
-import type { AppState } from 'store/reducers';
-import type { GlobalReducer } from 'types/reducer/globalTime';
 
 import { dtoToFormModel } from '../DashboardSettings/Variables/variableAdapters';
 import type { VariableFormModel } from '../DashboardSettings/Variables/variableFormModel';
@@ -16,8 +12,8 @@ import type {
 	VariableSelectionMap,
 } from './selectionTypes';
 import {
-	deriveFetchContext,
-	doAllQueryVariablesHaveValues,
+	computeVariableDependencies,
+	type VariableDependencyData,
 } from './variableDependencies';
 
 /** URL sentinel for an "ALL values selected" state (matches V1). */
@@ -33,14 +29,12 @@ export const variablesUrlParser = parseAsJson<
 );
 
 function defaultSelection(model: VariableFormModel): VariableSelection {
-	// `defaultValue` is a string | string[] on the wire.
-	const def = model.defaultValue;
-	if (Array.isArray(def) && def.length > 0) {
+	const def = (
+		model.defaultValue as { value?: SelectedVariableValue } | undefined
+	)?.value;
+	if (def !== undefined && def !== null && def !== '') {
 		return { value: def, allSelected: false };
 	}
-	if (typeof def === 'string' && def !== '') {
-		return { value: model.multiSelect ? [def] : def, allSelected: false };
-	}
 	return { value: model.multiSelect ? [] : '', allSelected: false };
 }
 
@@ -52,14 +46,9 @@ function fromUrlValue(raw: SelectedVariableValue): VariableSelection {
 
 interface UseVariableSelection {
 	variables: VariableFormModel[];
+	dependencyData: VariableDependencyData;
 	selection: VariableSelectionMap;
 	setSelection: (name: string, selection: VariableSelection) => void;
-	/**
-	 * Auto-selection fill (default/first-option) applied when options arrive. Unlike
-	 * {@link UseVariableSelection.setSelection}, fills from the initial load burst are
-	 * coalesced into one store write + one downstream refresh.
-	 */
-	autoSelect: (name: string, selection: VariableSelection) => void;
 }
 
 /**
@@ -76,37 +65,27 @@ export function useVariableSelection(
 		() => (dashboard.spec?.variables ?? []).map(dtoToFormModel),
 		[dashboard.spec?.variables],
 	);
-	const fetchContext = useMemo(() => deriveFetchContext(variables), [variables]);
+	const dependencyData = useMemo(
+		() => computeVariableDependencies(variables),
+		[variables],
+	);
 
 	const selection = useDashboardStore(selectVariableValues(dashboardId));
 	const setVariableValue = useDashboardStore((s) => s.setVariableValue);
 	const setVariableValues = useDashboardStore((s) => s.setVariableValues);
-	const initVariableFetch = useDashboardStore((s) => s.initVariableFetch);
-	const enqueueFetchAll = useDashboardStore((s) => s.enqueueFetchAll);
-	const enqueueDescendants = useDashboardStore((s) => s.enqueueDescendants);
-	const enqueueDescendantsBatch = useDashboardStore(
-		(s) => s.enqueueDescendantsBatch,
-	);
-
-	const { minTime, maxTime } = useSelector<AppState, GlobalReducer>(
-		(state) => state.globalTime,
-	);
-
-	// Latest selection, read by the fetch-cycle effect without subscribing to it
-	// (so a value change doesn't re-trigger a full fetch cycle).
-	const selectionRef = useRef(selection);
-	selectionRef.current = selection;
 
 	const [urlValues, setUrlValues] = useQueryState(
 		'variables',
 		variablesUrlParser.withOptions({ history: 'replace' }),
 	);
 
-	// Seed selections: URL wins, then persisted store, then default.
+	// Seed selections for this dashboard: URL wins, then persisted store, then default.
 	useEffect(() => {
 		if (!dashboardId || variables.length === 0) {
 			return;
 		}
+		// `selection` here is the persisted (localStorage) map on mount — the
+		// effect deliberately doesn't depend on it, so seeding runs once per set.
 		const stored = selection;
 		const seeded: VariableSelectionMap = {};
 		variables.forEach((variable) => {
@@ -123,83 +102,16 @@ export function useVariableSelection(
 		// eslint-disable-next-line react-hooks/exhaustive-deps
 	}, [dashboardId, variables]);
 
-	// Start a full fetch cycle on load / dependency-order / time change. Runs after
-	// the seeding effect above, so it reads the seeded selection from the store; a
-	// value change instead goes through `enqueueDescendants`, not this effect.
-	const orderKey = `${fetchContext.queryVariableOrder.join(
-		',',
-	)}|${fetchContext.dynamicVariableOrder.join(',')}`;
-	useEffect(() => {
-		if (!dashboardId || variables.length === 0) {
-			return;
-		}
-		const names = variables
-			.map((v) => v.name)
-			.filter((name): name is string => !!name);
-		initVariableFetch(names, fetchContext);
-		enqueueFetchAll(
-			doAllQueryVariablesHaveValues(variables, selectionRef.current),
-		);
-		// eslint-disable-next-line react-hooks/exhaustive-deps
-	}, [dashboardId, orderKey, minTime, maxTime]);
-
 	const setSelection = useCallback(
 		(name: string, next: VariableSelection): void => {
 			setVariableValue(dashboardId, name, next);
-			enqueueDescendants(name);
 			void setUrlValues((prev) => ({
 				...(prev ?? {}),
 				[name]: next.allSelected ? ALL_SELECTED : next.value,
 			}));
 		},
-		[dashboardId, setVariableValue, enqueueDescendants, setUrlValues],
+		[dashboardId, setVariableValue, setUrlValues],
 	);
 
-	// Coalesce the initial load burst of auto-selections: each selector fills its
-	// value as its options resolve (at different times). Collecting them into one
-	// store write + one `enqueueDescendantsBatch` means dependents re-fetch once
-	// with the settled parent values, instead of once per fill.
-	const pendingAutoFillRef = useRef<VariableSelectionMap>({});
-	const autoFillFrameRef = useRef<number | null>(null);
-
-	const flushAutoFills = useCallback((): void => {
-		autoFillFrameRef.current = null;
-		const fills = pendingAutoFillRef.current;
-		pendingAutoFillRef.current = {};
-		const names = Object.keys(fills);
-		if (names.length === 0 || !dashboardId) {
-			return;
-		}
-		setVariableValues(dashboardId, { ...selectionRef.current, ...fills });
-		void setUrlValues((prev) => {
-			const next = { ...(prev ?? {}) };
-			names.forEach((name) => {
-				const sel = fills[name];
-				next[name] = sel.allSelected ? ALL_SELECTED : sel.value;
-			});
-			return next;
-		});
-		enqueueDescendantsBatch(names);
-	}, [dashboardId, setVariableValues, setUrlValues, enqueueDescendantsBatch]);
-
-	const autoSelect = useCallback(
-		(name: string, next: VariableSelection): void => {
-			pendingAutoFillRef.current[name] = next;
-			if (autoFillFrameRef.current == null) {
-				autoFillFrameRef.current = requestAnimationFrame(flushAutoFills);
-			}
-		},
-		[flushAutoFills],
-	);
-
-	useEffect(
-		() => (): void => {
-			if (autoFillFrameRef.current != null) {
-				cancelAnimationFrame(autoFillFrameRef.current);
-			}
-		},
-		[],
-	);
-
-	return { variables, selection, setSelection, autoSelect };
+	return { variables, dependencyData, selection, setSelection };
 }

frontend/src/pages/DashboardPageV2/DashboardContainer/VariablesBar/variableDependencies.ts

@@ -1,11 +1,6 @@
 import { textContainsVariableReference } from 'lib/dashboardVariables/variableReference';
 
-import type {
-	VariableFormModel,
-	VariableType,
-} from '../DashboardSettings/Variables/variableFormModel';
-import type { VariableSelectionMap } from './selectionTypes';
-import { isResolved } from './selectionUtils';
+import type { VariableFormModel } from '../DashboardSettings/Variables/variableFormModel';
 
 /**
  * Inter-variable dependency graph for runtime selection. A QUERY variable
@@ -202,57 +197,3 @@ export function computeVariableDependencies(
 ): VariableDependencyData {
 	return buildDependencyData(buildDependencies(variables));
 }
-
-/**
- * Static context the runtime fetch engine (`variableFetchSlice`) needs to order
- * fetches: the dependency graph plus the per-name type index and the QUERY /
- * DYNAMIC fetch orders. Derived from the variable definitions; stable until the
- * spec's variables change. Mirrors V1's `getVariableDependencyContext`.
- */
-export interface VariableFetchContext {
-	dependencyData: VariableDependencyData;
-	/** variable name → its type. */
-	variableTypes: Record<string, VariableType>;
-	/** QUERY variables in topological (parent-before-child) order. */
-	queryVariableOrder: string[];
-	/** DYNAMIC variable names (they implicitly depend on all QUERY values). */
-	dynamicVariableOrder: string[];
-}
-
-export function deriveFetchContext(
-	variables: VariableFormModel[],
-): VariableFetchContext {
-	const dependencyData = computeVariableDependencies(variables);
-	const variableTypes: Record<string, VariableType> = {};
-	variables.forEach((v) => {
-		if (v.name) {
-			variableTypes[v.name] = v.type;
-		}
-	});
-	const queryVariableOrder = dependencyData.order.filter(
-		(name) => variableTypes[name] === 'QUERY',
-	);
-	const dynamicVariableOrder = variables
-		.filter((v) => v.type === 'DYNAMIC' && !!v.name)
-		.map((v) => v.name);
-	return {
-		dependencyData,
-		variableTypes,
-		queryVariableOrder,
-		dynamicVariableOrder,
-	};
-}
-
-/**
- * Whether every QUERY variable already has a usable selection — decides at load
- * time whether dynamic variables may fetch immediately or must wait for the
- * query variables to settle first (V1 parity).
- */
-export function doAllQueryVariablesHaveValues(
-	variables: VariableFormModel[],
-	selection: VariableSelectionMap,
-): boolean {
-	return variables
-		.filter((v) => v.type === 'QUERY')
-		.every((v) => isResolved(selection[v.name]));
-}

frontend/src/pages/DashboardPageV2/DashboardContainer/store/slices/__tests__/variableFetchSlice.test.ts

@@ -1,97 +0,0 @@
-import {
-	emptyVariableFormModel,
-	type VariableFormModel,
-} from '../../../DashboardSettings/Variables/variableFormModel';
-import { deriveFetchContext } from '../../../VariablesBar/variableDependencies';
-import { useDashboardStore } from '../../useDashboardStore';
-
-function model(overrides: Partial<VariableFormModel>): VariableFormModel {
-	return { ...emptyVariableFormModel(), ...overrides };
-}
-
-// q1 (root query) → q2 (query referencing $q1) ; d1 (dynamic).
-const q1 = model({ name: 'q1', type: 'QUERY', queryValue: 'SELECT 1' });
-const q2 = model({ name: 'q2', type: 'QUERY', queryValue: 'SELECT $q1' });
-const d1 = model({ name: 'd1', type: 'DYNAMIC', dynamicAttribute: 'pod' });
-const context = deriveFetchContext([q1, q2, d1]);
-
-function store(): ReturnType<typeof useDashboardStore.getState> {
-	return useDashboardStore.getState();
-}
-function states(): Record<string, string> {
-	return store().variableFetchStates;
-}
-
-beforeEach(() => {
-	useDashboardStore.setState({
-		variableFetchStates: {},
-		variableLastUpdated: {},
-		variableCycleIds: {},
-		variableFetchContext: null,
-	});
-	store().initVariableFetch(['q1', 'q2', 'd1'], context);
-});
-
-describe('variableFetchSlice', () => {
-	it('initializes every variable to idle', () => {
-		expect(states()).toStrictEqual({ q1: 'idle', q2: 'idle', d1: 'idle' });
-	});
-
-	it('enqueueFetchAll loads roots, waits dependents and (ungated) dynamics', () => {
-		store().enqueueFetchAll(false);
-		expect(states()).toStrictEqual({
-			q1: 'loading',
-			q2: 'waiting',
-			d1: 'waiting',
-		});
-		expect(store().variableCycleIds).toStrictEqual({ q1: 1, q2: 1, d1: 1 });
-	});
-
-	it('enqueueFetchAll loads dynamics immediately when query values exist', () => {
-		store().enqueueFetchAll(true);
-		expect(states().d1).toBe('loading');
-	});
-
-	it('completing a parent unblocks its query child, then unlocks dynamics', () => {
-		store().enqueueFetchAll(false);
-		store().onVariableFetchComplete('q1');
-		expect(states()).toMatchObject({ q1: 'idle', q2: 'loading', d1: 'waiting' });
-
-		store().onVariableFetchComplete('q2');
-		expect(states()).toMatchObject({ q1: 'idle', q2: 'idle', d1: 'loading' });
-	});
-
-	it('enqueueDescendants revalidates only descendants + dynamics', () => {
-		store().enqueueFetchAll(false);
-		store().onVariableFetchComplete('q1');
-		store().onVariableFetchComplete('q2');
-		store().onVariableFetchComplete('d1');
-
-		store().enqueueDescendants('q1');
-		// q2 depends on q1 (settled) → revalidates; d1 waits (q2 no longer settled).
-		expect(states().q2).toBe('revalidating');
-		expect(states().d1).toBe('waiting');
-	});
-
-	it('enqueueDescendantsBatch bumps each descendant + dynamic exactly once', () => {
-		store().enqueueFetchAll(false);
-		store().onVariableFetchComplete('q1');
-		store().onVariableFetchComplete('q2');
-		store().onVariableFetchComplete('d1');
-		const before = { ...store().variableCycleIds };
-
-		// q1 and q2 auto-select together: q2 is a descendant of q1 but is also in
-		// the batch — it should still bump only once, as should the dynamic.
-		store().enqueueDescendantsBatch(['q1', 'q2']);
-		const after = store().variableCycleIds;
-		expect(after.q2).toBe(before.q2 + 1);
-		expect(after.d1).toBe(before.d1 + 1);
-	});
-
-	it('a failed parent idles its query descendants', () => {
-		store().enqueueFetchAll(false);
-		store().onVariableFetchFailure('q1');
-		expect(states().q1).toBe('error');
-		expect(states().q2).toBe('idle');
-	});
-});

frontend/src/pages/DashboardPageV2/DashboardContainer/store/slices/variableFetchSlice.ts

@@ -1,261 +0,0 @@
-import type { StateCreator } from 'zustand';
-
-import type { VariableFetchContext } from '../../VariablesBar/variableDependencies';
-import type { DashboardStore } from '../useDashboardStore';
-import {
-	areAllQueryVariablesSettled,
-	type FetchMaps,
-	isSettled,
-	resolveFetchState,
-	unlockWaitingDynamicVariables,
-	type VariableFetchState,
-} from './variableFetchSlice.utils';
-
-export type { VariableFetchState } from './variableFetchSlice.utils';
-
-/**
- * Runtime fetch orchestration for dashboard variables — native port of V1's
- * `variableFetchStore`. Decides WHEN each variable's options fetch: query
- * variables in dependency order, dynamics together once query values exist,
- * text/custom never. `cycleIds` is a per-variable request nonce keyed into each
- * selector's react-query key (bump = fresh fetch, auto-cancel stale). Transient.
- * `enqueueFetchAll` = load/time change; `enqueueDescendants` = one value changed.
- */
-export interface VariableFetchSlice {
-	variableFetchStates: Record<string, VariableFetchState>;
-	variableLastUpdated: Record<string, number>;
-	variableCycleIds: Record<string, number>;
-	/** Static dependency context, set by `initVariableFetch` (null before init). */
-	variableFetchContext: VariableFetchContext | null;
-
-	/** Seed state entries for the current variable set and store the context. */
-	initVariableFetch: (names: string[], context: VariableFetchContext) => void;
-	/** Start a full fetch cycle for every fetchable variable (load / time change). */
-	enqueueFetchAll: (doAllQueryVariablesHaveValuesSelected: boolean) => void;
-	/** Mark a variable's fetch as done; unblock its waiting children / dynamics. */
-	onVariableFetchComplete: (name: string) => void;
-	/** Mark a variable's fetch as failed; idle its query descendants. */
-	onVariableFetchFailure: (name: string) => void;
-	/** Cascade a value change to a variable's query descendants + the dynamics. */
-	enqueueDescendants: (name: string) => void;
-	/**
-	 * Batched value-change cascade: refresh the union of the given variables'
-	 * query descendants plus the dynamics, each exactly once. Used to collapse the
-	 * initial burst of auto-selections into a single downstream fetch.
-	 */
-	enqueueDescendantsBatch: (names: string[]) => void;
-}
-
-/** Snapshot the three fetch maps into mutable clones for a single action. */
-function cloneMaps(state: DashboardStore): FetchMaps {
-	return {
-		states: { ...state.variableFetchStates },
-		lastUpdated: { ...state.variableLastUpdated },
-		cycleIds: { ...state.variableCycleIds },
-	};
-}
-
-export const createVariableFetchSlice: StateCreator<
-	DashboardStore,
-	[['zustand/persist', unknown]],
-	[],
-	VariableFetchSlice
-> = (set, get) => ({
-	variableFetchStates: {},
-	variableLastUpdated: {},
-	variableCycleIds: {},
-	variableFetchContext: null,
-
-	initVariableFetch: (names, context): void => {
-		const maps = cloneMaps(get());
-		// Initialize new variables to idle, preserving existing states.
-		names.forEach((name) => {
-			if (!maps.states[name]) {
-				maps.states[name] = 'idle';
-			}
-		});
-		// Drop entries for variables that no longer exist.
-		const nameSet = new Set(names);
-		Object.keys(maps.states).forEach((name) => {
-			if (!nameSet.has(name)) {
-				delete maps.states[name];
-				delete maps.lastUpdated[name];
-				delete maps.cycleIds[name];
-			}
-		});
-		set({
-			variableFetchStates: maps.states,
-			variableLastUpdated: maps.lastUpdated,
-			variableCycleIds: maps.cycleIds,
-			variableFetchContext: context,
-		});
-	},
-
-	enqueueFetchAll: (doAllQueryVariablesHaveValuesSelected): void => {
-		const { variableFetchContext } = get();
-		if (!variableFetchContext) {
-			return;
-		}
-		const {
-			dependencyData,
-			variableTypes,
-			queryVariableOrder,
-			dynamicVariableOrder,
-		} = variableFetchContext;
-		const maps = cloneMaps(get());
-
-		// Query variables: roots start immediately, dependents wait for parents.
-		queryVariableOrder.forEach((name) => {
-			maps.cycleIds[name] = (maps.cycleIds[name] || 0) + 1;
-			const parents = dependencyData.parentGraph[name] || [];
-			const hasQueryParents = parents.some((p) => variableTypes[p] === 'QUERY');
-			maps.states[name] = hasQueryParents
-				? 'waiting'
-				: resolveFetchState(maps, name);
-		});
-
-		// Dynamic variables: start now if query variables already have values,
-		// otherwise wait until the query variables settle.
-		dynamicVariableOrder.forEach((name) => {
-			maps.cycleIds[name] = (maps.cycleIds[name] || 0) + 1;
-			maps.states[name] = doAllQueryVariablesHaveValuesSelected
-				? resolveFetchState(maps, name)
-				: 'waiting';
-		});
-
-		set({
-			variableFetchStates: maps.states,
-			variableLastUpdated: maps.lastUpdated,
-			variableCycleIds: maps.cycleIds,
-		});
-	},
-
-	onVariableFetchComplete: (name): void => {
-		const { variableFetchContext } = get();
-		const maps = cloneMaps(get());
-		maps.states[name] = 'idle';
-		maps.lastUpdated[name] = Date.now();
-
-		if (variableFetchContext) {
-			const { dependencyData, variableTypes, dynamicVariableOrder } =
-				variableFetchContext;
-			// Unblock waiting query-type children.
-			(dependencyData.graph[name] || []).forEach((child) => {
-				if (variableTypes[child] === 'QUERY' && maps.states[child] === 'waiting') {
-					maps.states[child] = resolveFetchState(maps, child);
-				}
-			});
-			// Once all query variables settle, unlock any waiting dynamics.
-			if (
-				variableTypes[name] === 'QUERY' &&
-				areAllQueryVariablesSettled(maps.states, variableTypes)
-			) {
-				unlockWaitingDynamicVariables(maps, dynamicVariableOrder);
-			}
-		}
-
-		set({
-			variableFetchStates: maps.states,
-			variableLastUpdated: maps.lastUpdated,
-			variableCycleIds: maps.cycleIds,
-		});
-	},
-
-	onVariableFetchFailure: (name): void => {
-		const { variableFetchContext } = get();
-		const maps = cloneMaps(get());
-		maps.states[name] = 'error';
-
-		if (variableFetchContext) {
-			const { dependencyData, variableTypes, dynamicVariableOrder } =
-				variableFetchContext;
-			// Query descendants can't proceed without this parent — idle them.
-			(dependencyData.transitiveDescendants[name] || []).forEach((desc) => {
-				if (variableTypes[desc] === 'QUERY') {
-					maps.states[desc] = 'idle';
-				}
-			});
-			if (
-				variableTypes[name] === 'QUERY' &&
-				areAllQueryVariablesSettled(maps.states, variableTypes)
-			) {
-				unlockWaitingDynamicVariables(maps, dynamicVariableOrder);
-			}
-		}
-
-		set({
-			variableFetchStates: maps.states,
-			variableLastUpdated: maps.lastUpdated,
-			variableCycleIds: maps.cycleIds,
-		});
-	},
-
-	enqueueDescendants: (name): void => {
-		get().enqueueDescendantsBatch([name]);
-	},
-
-	enqueueDescendantsBatch: (names): void => {
-		const { variableFetchContext } = get();
-		if (!variableFetchContext || names.length === 0) {
-			return;
-		}
-		const { dependencyData, variableTypes, dynamicVariableOrder } =
-			variableFetchContext;
-		const maps = cloneMaps(get());
-
-		// Union of the changed variables' query descendants, refreshed once each:
-		// refetch when all their parents are settled, else wait.
-		const queryDescendants = new Set<string>();
-		names.forEach((name) => {
-			(dependencyData.transitiveDescendants[name] || []).forEach((desc) => {
-				if (variableTypes[desc] === 'QUERY') {
-					queryDescendants.add(desc);
-				}
-			});
-		});
-		queryDescendants.forEach((desc) => {
-			maps.cycleIds[desc] = (maps.cycleIds[desc] || 0) + 1;
-			const parents = dependencyData.parentGraph[desc] || [];
-			const allParentsSettled = parents.every((p) => isSettled(maps.states[p]));
-			maps.states[desc] = allParentsSettled
-				? resolveFetchState(maps, desc)
-				: 'waiting';
-		});
-
-		// Dynamics implicitly depend on all query values: refetch now if the query
-		// variables are settled, otherwise wait for them.
-		dynamicVariableOrder.forEach((dynName) => {
-			maps.cycleIds[dynName] = (maps.cycleIds[dynName] || 0) + 1;
-			maps.states[dynName] = areAllQueryVariablesSettled(
-				maps.states,
-				variableTypes,
-			)
-				? resolveFetchState(maps, dynName)
-				: 'waiting';
-		});
-
-		set({
-			variableFetchStates: maps.states,
-			variableLastUpdated: maps.lastUpdated,
-			variableCycleIds: maps.cycleIds,
-		});
-	},
-});
-
-/** Selector: the fetch state for a single variable (defaults to idle). */
-export const selectVariableFetchState =
-	(name: string) =>
-	(state: DashboardStore): VariableFetchState =>
-		state.variableFetchStates[name] ?? 'idle';
-
-/** Selector: the current fetch cycle id for a single variable (defaults to 0). */
-export const selectVariableCycleId =
-	(name: string) =>
-	(state: DashboardStore): number =>
-		state.variableCycleIds[name] ?? 0;
-
-/** Selector: whether a variable has completed at least one fetch. */
-export const selectVariableFetchedOnce =
-	(name: string) =>
-	(state: DashboardStore): boolean =>
-		(state.variableLastUpdated[name] ?? 0) > 0;

frontend/src/pages/DashboardPageV2/DashboardContainer/store/slices/variableFetchSlice.utils.ts

@@ -1,51 +0,0 @@
-import type { VariableType } from '../../DashboardSettings/Variables/variableFormModel';
-
-/** Per-variable fetch lifecycle (ported from V1's `variableFetchStore`). */
-export type VariableFetchState =
-	| 'idle'
-	| 'loading'
-	| 'revalidating'
-	| 'waiting'
-	| 'error';
-
-/** Mutable clones a fetch action works over before committing back in one `set`. */
-export interface FetchMaps {
-	states: Record<string, VariableFetchState>;
-	lastUpdated: Record<string, number>;
-	cycleIds: Record<string, number>;
-}
-
-/** Settled = can make no further progress (idle or error). */
-export function isSettled(state: VariableFetchState | undefined): boolean {
-	return state === 'idle' || state === 'error';
-}
-
-/** Fetch-start state: `revalidating` if fetched before, else `loading`. */
-export function resolveFetchState(
-	maps: FetchMaps,
-	name: string,
-): VariableFetchState {
-	return (maps.lastUpdated[name] || 0) > 0 ? 'revalidating' : 'loading';
-}
-
-/** True once every QUERY variable is settled. */
-export function areAllQueryVariablesSettled(
-	states: Record<string, VariableFetchState>,
-	variableTypes: Record<string, VariableType>,
-): boolean {
-	return Object.entries(variableTypes)
-		.filter(([, type]) => type === 'QUERY')
-		.every(([name]) => isSettled(states[name]));
-}
-
-/** Move any `waiting` dynamic variables into loading/revalidating. */
-export function unlockWaitingDynamicVariables(
-	maps: FetchMaps,
-	dynamicVariableOrder: string[],
-): void {
-	dynamicVariableOrder.forEach((dynName) => {
-		if (maps.states[dynName] === 'waiting') {
-			maps.states[dynName] = resolveFetchState(maps, dynName);
-		}
-	});
-}

frontend/src/pages/DashboardPageV2/DashboardContainer/store/useDashboardStore.ts

@@ -13,15 +13,10 @@ import {
 	createVariableSelectionSlice,
 	type VariableSelectionSlice,
 } from './slices/variableSelectionSlice';
-import {
-	createVariableFetchSlice,
-	type VariableFetchSlice,
-} from './slices/variableFetchSlice';
 
 export type DashboardStore = EditContextSlice &
 	CollapseSlice &
-	VariableSelectionSlice &
-	VariableFetchSlice;
+	VariableSelectionSlice;
 
 /**
  * V2 dashboard session store. Holds cross-cutting client state only — never the
@@ -36,7 +31,6 @@ export const useDashboardStore = create<DashboardStore>()(
 			...createEditContextSlice(...a),
 			...createCollapseSlice(...a),
 			...createVariableSelectionSlice(...a),
-			...createVariableFetchSlice(...a),
 		}),
 		{
 			name: '@signoz/dashboard-v2',

pkg/telemetrymetrics/reduced_test.go

@@ -39,48 +39,48 @@ 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(?) 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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum_last`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s 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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, 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",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), false, "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(?) 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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum_last`, points.computed_at) AS value, argMax(`count_series`, points.computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s 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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, 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",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), false, "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(?) 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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`min`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s 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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "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(?) 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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`max`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s 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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "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(?) 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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s 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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, 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",
+				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), false, "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(?) 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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum`, points.computed_at) AS value, argMax(`count_series`, points.computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s 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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, 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",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric", uint64(1746999600000), uint64(1747172760000), 0, "test.metric", uint64(1746999600000), uint64(1747172760000), false, "test.metric", uint64(1746999600000), uint64(1747172760000)},
 			},
 		},
 		{
@@ -103,16 +103,16 @@ func TestReducedStatementBuilder(t *testing.T) {
 			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(?) 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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, `le`, argMax(`sum`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s 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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli, `le`) 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), false, "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(?) 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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum_last`, points.computed_at) AS value, argMax(`count_series`, points.computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s 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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, 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",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), false, "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 	}

pkg/telemetrymetrics/statement_builder.go

@@ -337,20 +337,28 @@ func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
 	}
 
 	// dedup recomputed buckets: latest computed_at wins per (series, 60s bucket)
+	// TODO(srikanthccv): add _5m/_30m tables similar to samples_v4
+	// and wrie them up in querier before GA
+	// TODO(srikanthccv): FINAL clause for the reduced table.
 	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.Select("points.reduced_fingerprint AS fingerprint", "points.unix_milli AS unix_milli")
+	for _, g := range query.GroupBy {
+		dedup.SelectMore(fmt.Sprintf("`%s`", g.Name))
 	}
-	dedup.From(fmt.Sprintf("%s.%s", DBName, WhichReducedSamplesTableToUse(agg.Type)))
+	dedup.SelectMore(fmt.Sprintf("argMax(%s, points.computed_at) AS value", value))
+	if weight != "" {
+		dedup.SelectMore(fmt.Sprintf("argMax(%s, points.computed_at) AS weight", weight))
+	}
+	dedup.From(fmt.Sprintf("%s.%s AS points", DBName, WhichReducedSamplesTableToUse(agg.Type)))
+	dedup.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.reduced_fingerprint = filtered_time_series.fingerprint")
 	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)
+	dedup.GroupBy("fingerprint", "unix_milli")
+	dedup.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+	dedupQuery, dedupArgs := dedup.BuildWithFlavor(sqlbuilder.ClickHouse, timeSeriesCTEArgs...)
 
 	sb := sqlbuilder.NewSelectBuilder()
 	sb.Select("fingerprint")
@@ -364,13 +372,11 @@ func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
 		// denominator is reduced with avg
 		sb.SelectMore("avg(weight) AS per_series_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)", dedupQuery))
 	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, dedupArgs...)
 	return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, true
 }
 

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",
@@ -80,12 +80,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."""
@@ -414,6 +422,267 @@ 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))
+        self.normalized = False
+
+        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,
+            {},
+            {},
+            self.normalized,
+        ]
+
+
+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))
+        self.normalized = False
+
+        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,
+            {},
+            {},
+            self.normalized,
+        ]
+
+
+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.
@@ -576,6 +845,163 @@ 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",
+                "__normalized",
+            ],
+            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",
+                "__normalized",
+            ],
+            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,35 @@ 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."""
+    return (int((datetime.now(tz=UTC) - ago).timestamp()) // step_seconds) * step_seconds
+
+
+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/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")