chore: add todos

.github/workflows/integrationci.yaml

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

frontend/__mocks__/lib/env.ts

@@ -1,3 +0,0 @@
-export const IS_DEV = false;
-export const IS_PROD = true;
-export const MODE = 'test';

frontend/jest.config.ts

@@ -29,7 +29,6 @@ const config: Config.InitialOptions = {
 		'^constants/env$': '<rootDir>/__mocks__/env.ts',
 		'^src/constants/env$': '<rootDir>/__mocks__/env.ts',
 		'^@signozhq/icons$': '<rootDir>/__mocks__/signozhqIconsMock.tsx',
-		'^lib/env$': '<rootDir>/__mocks__/lib/env.ts',
 		'^test-mocks/(.*)$': '<rootDir>/__mocks__/$1',
 		'^react-syntax-highlighter/dist/esm/(.*)$':
 			'<rootDir>/node_modules/react-syntax-highlighter/dist/cjs/$1',

frontend/src/components/cmdKPalette/cmdKPalette.tsx

@@ -22,7 +22,6 @@ import {
 } from 'container/AIAssistant/store/useAIAssistantStore';
 import { useThemeMode } from 'hooks/useDarkMode';
 import { useIsAIAssistantEnabled } from 'hooks/useIsAIAssistantEnabled';
-import { IS_DEV } from 'lib/env';
 import history from 'lib/history';
 import { ROLES as UserRole } from 'types/roles';
 
@@ -31,33 +30,6 @@ import { useCmdK } from '../../providers/cmdKProvider';
 
 import './cmdKPalette.scss';
 
-const AuthZDevModal = IS_DEV
-	? React.lazy(() =>
-			import('lib/authz/devtools/AuthZDevModal/AuthZDevModal').then((m) => ({
-				default: m.AuthZDevModal,
-			})),
-		)
-	: null;
-
-const AuthZDevFloatingIndicator = IS_DEV
-	? React.lazy(() =>
-			import('lib/authz/devtools/AuthZDevFloatingIndicator/AuthZDevFloatingIndicator').then(
-				(m) => ({
-					default: m.AuthZDevFloatingIndicator,
-				}),
-			),
-		)
-	: null;
-
-const openAuthZDevModal = IS_DEV
-	? (): void => {
-			void import('lib/authz/devtools/useAuthZDevStore').then((m) => {
-				m.openAuthZDevModal();
-				return m;
-			});
-		}
-	: undefined;
-
 type CmdAction = {
 	id: string;
 	name: string;
@@ -138,7 +110,6 @@ export function CmdKPalette({
 		aiAssistant: isAIAssistantEnabled
 			? { open: handleOpenAIAssistant }
 			: undefined,
-		authzDevTools: openAuthZDevModal ? { open: openAuthZDevModal } : undefined,
 	});
 
 	// RBAC filter: show action if no roles set OR current user role is included
@@ -175,57 +146,37 @@ export function CmdKPalette({
 	};
 
 	return (
-		<>
-			<CommandDialog
-				open={open}
-				onOpenChange={setOpen}
-				position="top"
-				offset={110}
-			>
-				<CommandInput placeholder="Search…" className="cmdk-input-wrapper" />
-				<CommandList className="cmdk-list-scroll">
-					<CommandEmpty>No results</CommandEmpty>
-					{grouped.map(([section, items]) => (
-						<CommandGroup
-							key={section}
-							heading={section}
-							className="cmdk-section-heading"
-						>
-							{items.map((it) => (
-								<CommandItem
-									key={it.id}
-									onSelect={(): void => handleInvoke(it)}
-									value={it.name}
-									className={theme === 'light' ? 'cmdk-item-light' : 'cmdk-item'}
+		<CommandDialog open={open} onOpenChange={setOpen} position="top" offset={110}>
+			<CommandInput placeholder="Search…" className="cmdk-input-wrapper" />
+			<CommandList className="cmdk-list-scroll">
+				<CommandEmpty>No results</CommandEmpty>
+				{grouped.map(([section, items]) => (
+					<CommandGroup
+						key={section}
+						heading={section}
+						className="cmdk-section-heading"
+					>
+						{items.map((it) => (
+							<CommandItem
+								key={it.id}
+								onSelect={(): void => handleInvoke(it)}
+								value={it.name}
+								className={theme === 'light' ? 'cmdk-item-light' : 'cmdk-item'}
+							>
+								<span
+									className={cx('cmd-item-icon', it.id === 'ai-assistant' && 'noz-icon')}
 								>
-									<span
-										className={cx(
-											'cmd-item-icon',
-											it.id === 'ai-assistant' && 'noz-icon',
-										)}
-									>
-										{it.icon}
-									</span>
-									{it.name}
-									{it.shortcut && it.shortcut.length > 0 && (
-										<CommandShortcut>{it.shortcut.join(' • ')}</CommandShortcut>
-									)}
-								</CommandItem>
-							))}
-						</CommandGroup>
-					))}
-				</CommandList>
-			</CommandDialog>
-			{IS_DEV && AuthZDevModal && (
-				<React.Suspense fallback={null}>
-					<AuthZDevModal />
-				</React.Suspense>
-			)}
-			{IS_DEV && AuthZDevFloatingIndicator && (
-				<React.Suspense fallback={null}>
-					<AuthZDevFloatingIndicator />
-				</React.Suspense>
-			)}
-		</>
+									{it.icon}
+								</span>
+								{it.name}
+								{it.shortcut && it.shortcut.length > 0 && (
+									<CommandShortcut>{it.shortcut.join(' • ')}</CommandShortcut>
+								)}
+							</CommandItem>
+						))}
+					</CommandGroup>
+				))}
+			</CommandList>
+		</CommandDialog>
 	);
 }

frontend/src/constants/shortcutActions.tsx

@@ -43,17 +43,10 @@ type ActionDeps = {
 	aiAssistant?: {
 		open: () => void;
 	};
-	/**
-	 * Provided only in development mode. Opens the AuthZ DevTools modal
-	 * for testing permission overrides.
-	 */
-	authzDevTools?: {
-		open: () => void;
-	};
 };
 
 export function createShortcutActions(deps: ActionDeps): CmdAction[] {
-	const { navigate, handleThemeChange, aiAssistant, authzDevTools } = deps;
+	const { navigate, handleThemeChange, aiAssistant } = deps;
 
 	const actions: CmdAction[] = [
 		{
@@ -309,17 +302,5 @@ export function createShortcutActions(deps: ActionDeps): CmdAction[] {
 		});
 	}
 
-	if (authzDevTools) {
-		actions.push({
-			id: 'authz-devtools',
-			name: 'AuthZ DevTools',
-			keywords: 'authz permissions rbac debug devtools override testing',
-			section: 'Dev',
-			icon: <Settings size={14} />,
-			roles: ['ADMIN', 'EDITOR', 'VIEWER'],
-			perform: authzDevTools.open,
-		});
-	}
-
 	return actions;
 }

frontend/src/container/RolesSettings/CreateEditRolePage/__tests__/CreateRolePage.authz.test.tsx

@@ -48,8 +48,6 @@ describe('CreateRolePage - AuthZ', () => {
 				isFetching: true,
 				error: null,
 				permissions: null,
-				allowed: false,
-				deniedPermissions: [],
 				refetchPermissions: jest.fn(),
 			});
 

frontend/src/container/RolesSettings/CreateEditRolePage/__tests__/EditRolePage.authz.test.tsx

@@ -77,8 +77,6 @@ describe('EditRolePage - AuthZ', () => {
 				isFetching: true,
 				error: null,
 				permissions: null,
-				allowed: false,
-				deniedPermissions: [],
 				refetchPermissions: jest.fn(),
 			});
 

frontend/src/container/RolesSettings/ViewRolePage/__tests__/ViewRolePage.authz.test.tsx

@@ -409,8 +409,6 @@ describe('ViewRolePage - AuthZ', () => {
 				isFetching: true,
 				error: null,
 				permissions: null,
-				allowed: false,
-				deniedPermissions: [],
 				refetchPermissions: jest.fn(),
 			});
 

frontend/src/lib/authz/devtools/AuthZDevFloatingIndicator/AuthZDevFloatingIndicator.module.css

@@ -1,28 +0,0 @@
-.container {
-	position: fixed;
-	top: 12px;
-	left: 50%;
-	transform: translateX(-50%);
-	z-index: 9998;
-	pointer-events: auto;
-	display: flex;
-	align-items: center;
-	gap: 4px;
-}
-
-.button {
-	box-shadow:
-		0 4px 12px rgb(0 0 0 / 15%),
-		0 0 0 1px rgb(0 0 0 / 5%);
-}
-
-.badge {
-	margin-left: 4px;
-}
-
-.closeButton {
-	border-radius: 4px;
-	box-shadow:
-		0 4px 12px rgb(0 0 0 / 15%),
-		0 0 0 1px rgb(0 0 0 / 5%);
-}

frontend/src/lib/authz/devtools/AuthZDevFloatingIndicator/AuthZDevFloatingIndicator.tsx

@@ -1,61 +0,0 @@
-import { X } from '@signozhq/icons';
-import { Badge } from '@signozhq/ui/badge';
-import { Button } from '@signozhq/ui/button';
-import { useState } from 'react';
-import { createPortal } from 'react-dom';
-
-import { useAuthZDevStore } from '../useAuthZDevStore';
-
-import styles from './AuthZDevFloatingIndicator.module.css';
-
-export function AuthZDevFloatingIndicator(): JSX.Element | null {
-	const overrides = useAuthZDevStore((s) => s.overrides);
-	const isModalOpen = useAuthZDevStore((s) => s.isModalOpen);
-	const openModal = useAuthZDevStore((s) => s.openModal);
-	const [isDismissed, setIsDismissed] = useState(false);
-
-	const overrideCount = Object.keys(overrides).length;
-
-	if (overrideCount === 0 || isModalOpen || isDismissed) {
-		return null;
-	}
-
-	const handleOpen = (): void => {
-		setIsDismissed(false);
-		openModal();
-	};
-
-	const handleDismiss = (e: React.MouseEvent): void => {
-		e.stopPropagation();
-		setIsDismissed(true);
-	};
-
-	return createPortal(
-		<div className={styles.container}>
-			<Button
-				variant="solid"
-				color="warning"
-				size="sm"
-				onClick={handleOpen}
-				className={styles.button}
-				data-testid="authz-dev-floating-indicator"
-			>
-				AuthZ Overrides
-				<Badge color="warning" className={styles.badge}>
-					{overrideCount}
-				</Badge>
-			</Button>
-			<Button
-				variant="ghost"
-				color="secondary"
-				size="sm"
-				onClick={handleDismiss}
-				className={styles.closeButton}
-				aria-label="Dismiss indicator"
-				data-testid="authz-dev-floating-dismiss"
-				prefix={<X />}
-			/>
-		</div>,
-		document.body,
-	);
-}

frontend/src/lib/authz/devtools/AuthZDevModal/AuthZDevModal.module.css

@@ -1,140 +0,0 @@
-.modal {
-	--dialog-width: 640px;
-	--dialog-max-width: 92vw;
-	--dialog-max-height: 78vh;
-	--dialog-description-padding: var(--spacing-4) var(--spacing-4) 0px
-		var(--spacing-4);
-
-	[data-slot='dialog-description'],
-	[data-slot='dialog-header'] {
-		background-color: var(--l2-background);
-	}
-}
-
-.content {
-	display: flex;
-	flex-direction: column;
-	max-height: calc(78vh - 80px);
-}
-
-.header {
-	display: flex;
-	flex-direction: column;
-	flex-shrink: 0;
-	gap: var(--spacing-4);
-	padding-bottom: var(--spacing-4);
-}
-
-.searchRow {
-	display: flex;
-	align-items: center;
-	gap: var(--spacing-4);
-
-	--input-background: var(--l3-background);
-	--input-hover-background: var(--l3-background);
-	--input-focus-background: var(--l3-background);
-	--input-border-color: var(--l3-border);
-	--input-hover-border-color: var(--l3-border);
-	--input-focus-border-color: var(--l3-border);
-
-	--select-trigger-background-color: var(--l3-background);
-	--select-trigger-hover-background: var(--l3-background);
-	--select-trigger-focus-background: var(--l3-background);
-	--select-trigger-border-color: var(--l3-border);
-
-	--select-content-background: var(--l3-background);
-	--select-item-highlight-background: var(--l3-background-hover);
-}
-
-.search {
-	flex: 1 1 auto;
-	min-width: 0;
-
-	--input-width: 100%;
-}
-
-.filter {
-	flex: 0 0 176px;
-	/* Normalize the library trigger height (2.25rem) to match the input. */
-	--select-trigger-height: 2rem;
-}
-
-.search > *,
-.filter > * {
-	box-sizing: border-box;
-}
-
-.searchIcon {
-	display: inline-flex;
-	color: var(--l3-foreground);
-}
-
-.actionsRow {
-	display: flex;
-	gap: var(--spacing-3);
-}
-
-.actionButton {
-	flex: 0 0 auto;
-	height: 2rem;
-}
-
-.list {
-	display: flex;
-	flex: 1 1 auto;
-	flex-direction: column;
-	gap: var(--spacing-6);
-	padding: var(--spacing-4) 0;
-	overflow-y: auto;
-}
-
-.section {
-	display: flex;
-	flex-direction: column;
-	gap: var(--spacing-2);
-}
-
-.sectionHeader {
-	display: flex;
-	align-items: baseline;
-	gap: var(--spacing-3);
-	padding: var(--spacing-3) var(--spacing-2);
-	margin: 0 0 var(--spacing-1);
-	border-bottom: 1px solid var(--l2-border);
-}
-
-.empty {
-	display: flex;
-	align-items: center;
-	justify-content: center;
-	min-height: 160px;
-	padding: var(--spacing-16);
-}
-
-.footer {
-	display: flex;
-	flex-shrink: 0;
-	align-items: center;
-	justify-content: space-between;
-	gap: var(--spacing-4);
-	padding: var(--spacing-4) 0;
-	border-top: 1px solid var(--l2-border);
-}
-
-.hint {
-	display: flex;
-	flex-wrap: wrap;
-	align-items: center;
-	gap: var(--spacing-4);
-}
-
-.hintGroup {
-	display: inline-flex;
-	align-items: center;
-	gap: var(--spacing-2);
-}
-
-.count {
-	flex: 0 0 auto;
-	white-space: nowrap;
-}

frontend/src/lib/authz/devtools/AuthZDevModal/AuthZDevModal.tsx

@@ -1,108 +0,0 @@
-import { DialogWrapper } from '@signozhq/ui/dialog';
-import { useCallback, useRef } from 'react';
-
-import { useAuthZDevStore } from '../useAuthZDevStore';
-import { useAuthZQueryInvalidation } from '../useAuthZQueryInvalidation';
-
-import { AuthZDevModalContent } from './AuthZDevModalContent';
-import { AuthZDevModalFooter } from './AuthZDevModalFooter';
-import { AuthZDevModalHeader } from './AuthZDevModalHeader';
-import { useAuthZDevModalData } from './useAuthZDevModalData';
-import { useModalKeyboard } from './useModalKeyboard';
-
-import styles from './AuthZDevModal.module.css';
-
-export function AuthZDevModal(): JSX.Element | null {
-	const isModalOpen = useAuthZDevStore((s) => s.isModalOpen);
-	const closeModal = useAuthZDevStore((s) => s.closeModal);
-	const observed = useAuthZDevStore((s) => s.observed);
-	const overrides = useAuthZDevStore((s) => s.overrides);
-	const cycleOverride = useAuthZDevStore((s) => s.cycleOverride);
-	const setOverride = useAuthZDevStore((s) => s.setOverride);
-	const clearAllOverrides = useAuthZDevStore((s) => s.clearAllOverrides);
-	const grantAll = useAuthZDevStore((s) => s.grantAll);
-	const denyAll = useAuthZDevStore((s) => s.denyAll);
-
-	useAuthZQueryInvalidation(overrides);
-
-	const searchInputRef = useRef<HTMLInputElement>(null);
-
-	const {
-		search,
-		setSearch,
-		resourceFilter,
-		setResourceFilter,
-		observedList,
-		resourceFilterItems,
-		filteredPermissions,
-		groups,
-		orderedPermissions,
-		indexByPermission,
-		hasActiveFilter,
-		filteredOverrideCount,
-		overrideCount,
-	} = useAuthZDevModalData(observed, overrides);
-
-	const { selectedIndex, setSelectedIndex } = useModalKeyboard({
-		permissions: orderedPermissions,
-		overrides,
-		onCycle: cycleOverride,
-		onSetOverride: setOverride,
-		onClose: closeModal,
-		searchInputRef,
-	});
-
-	const handleOpenChange = useCallback(
-		(open: boolean): void => {
-			if (!open) {
-				closeModal();
-				setSelectedIndex(-1);
-			}
-		},
-		[closeModal, setSelectedIndex],
-	);
-
-	return (
-		<DialogWrapper
-			open={isModalOpen}
-			onOpenChange={handleOpenChange}
-			title="AuthZ DevTools"
-			subTitle="Force permission results locally without touching the backend."
-			className={styles.modal}
-			width="wide"
-		>
-			<div className={styles.content}>
-				<AuthZDevModalHeader
-					searchInputRef={searchInputRef}
-					search={search}
-					setSearch={setSearch}
-					resourceFilter={resourceFilter}
-					setResourceFilter={setResourceFilter}
-					resourceFilterItems={resourceFilterItems}
-					hasActiveFilter={hasActiveFilter}
-					filteredPermissions={filteredPermissions}
-					filteredOverrideCount={filteredOverrideCount}
-					overrideCount={overrideCount}
-					grantAll={grantAll}
-					denyAll={denyAll}
-					clearAllOverrides={clearAllOverrides}
-				/>
-				<AuthZDevModalContent
-					observedListLength={observedList.length}
-					orderedPermissions={orderedPermissions}
-					groups={groups}
-					indexByPermission={indexByPermission}
-					selectedIndex={selectedIndex}
-					setSelectedIndex={setSelectedIndex}
-					observed={observed}
-					overrides={overrides}
-					onSetOverride={setOverride}
-				/>
-				<AuthZDevModalFooter
-					orderedPermissionsCount={orderedPermissions.length}
-					observedListLength={observedList.length}
-				/>
-			</div>
-		</DialogWrapper>
-	);
-}

frontend/src/lib/authz/devtools/AuthZDevModal/AuthZDevModalContent.tsx

@@ -1,85 +0,0 @@
-import { Typography } from '@signozhq/ui/typography';
-import { useCallback } from 'react';
-
-import { BrandedPermission } from '../../hooks/useAuthZ/types';
-import { ObservedPermission, OverrideState } from '../types';
-
-import { PermissionRow } from './PermissionRow';
-
-import styles from './AuthZDevModal.module.css';
-
-export interface PermissionGroup {
-	resource: string;
-	items: string[];
-}
-
-export interface AuthZDevModalContentProps {
-	observedListLength: number;
-	orderedPermissions: string[];
-	groups: PermissionGroup[];
-	indexByPermission: Map<string, number>;
-	selectedIndex: number;
-	setSelectedIndex: (index: number) => void;
-	observed: Record<string, ObservedPermission>;
-	overrides: Record<string, OverrideState>;
-	onSetOverride: (permission: BrandedPermission, state: OverrideState) => void;
-}
-
-export function AuthZDevModalContent({
-	observedListLength,
-	orderedPermissions,
-	groups,
-	indexByPermission,
-	selectedIndex,
-	setSelectedIndex,
-	observed,
-	overrides,
-	onSetOverride,
-}: AuthZDevModalContentProps): JSX.Element {
-	const handleSelectIndex = useCallback(
-		(index: number) => (): void => {
-			setSelectedIndex(index);
-		},
-		[setSelectedIndex],
-	);
-
-	return (
-		<div className={styles.list} data-testid="authz-dev-permission-list">
-			{orderedPermissions.length === 0 ? (
-				<div className={styles.empty}>
-					<Typography.Text align="center" color="muted">
-						{observedListLength === 0
-							? 'No permissions observed yet. Navigate the app to trigger permission checks.'
-							: 'No permissions match your search.'}
-					</Typography.Text>
-				</div>
-			) : (
-				groups.map((group) => (
-					<div key={group.resource} className={styles.section}>
-						<div className={styles.sectionHeader}>
-							<Typography.Text as="span" size="medium" weight="semibold">
-								{group.resource}
-							</Typography.Text>
-							<Typography.Text as="span" size="small" color="muted">
-								{group.items.length}
-							</Typography.Text>
-						</div>
-						{group.items.map((permission) => {
-							const index = indexByPermission.get(permission) ?? 0;
-							return (
-								<PermissionRow
-									key={permission}
-									observed={observed[permission]}
-									override={overrides[permission]}
-									isSelected={index === selectedIndex}
-									onSetOverride={onSetOverride}
-									onSelect={handleSelectIndex(index)}
-								/>
-							);
-						})}
-					</div>
-				))
-			)}
-		</div>
-	);
-}

frontend/src/lib/authz/devtools/AuthZDevModal/AuthZDevModalFooter.tsx

@@ -1,56 +0,0 @@
-import { Kbd } from '@signozhq/ui/kbd';
-import { Typography } from '@signozhq/ui/typography';
-
-import styles from './AuthZDevModal.module.css';
-
-export interface AuthZDevModalFooterProps {
-	orderedPermissionsCount: number;
-	observedListLength: number;
-}
-
-export function AuthZDevModalFooter({
-	orderedPermissionsCount,
-	observedListLength,
-}: AuthZDevModalFooterProps): JSX.Element {
-	return (
-		<div className={styles.footer}>
-			<div className={styles.hint}>
-				<span className={styles.hintGroup}>
-					<Kbd>↑</Kbd>
-					<Kbd>↓</Kbd>
-					<Typography.Text as="span" size="small" color="muted">
-						navigate
-					</Typography.Text>
-				</span>
-				<span className={styles.hintGroup}>
-					<Kbd>←</Kbd>
-					<Kbd>→</Kbd>
-					<Typography.Text as="span" size="small" color="muted">
-						mode
-					</Typography.Text>
-				</span>
-				<span className={styles.hintGroup}>
-					<Kbd>1-5</Kbd>
-					<Typography.Text as="span" size="small" color="muted">
-						set
-					</Typography.Text>
-				</span>
-				<span className={styles.hintGroup}>
-					<Kbd>/</Kbd>
-					<Typography.Text as="span" size="small" color="muted">
-						search
-					</Typography.Text>
-				</span>
-				<span className={styles.hintGroup}>
-					<Kbd>Esc</Kbd>
-					<Typography.Text as="span" size="small" color="muted">
-						close
-					</Typography.Text>
-				</span>
-			</div>
-			<Typography.Text size="small" color="muted" className={styles.count}>
-				{orderedPermissionsCount} of {observedListLength} permissions
-			</Typography.Text>
-		</div>
-	);
-}

frontend/src/lib/authz/devtools/AuthZDevModal/AuthZDevModalHeader.tsx

@@ -1,119 +0,0 @@
-import { Search } from '@signozhq/icons';
-import { Button } from '@signozhq/ui/button';
-import { Input } from '@signozhq/ui/input';
-import { SelectSimple } from '@signozhq/ui/select';
-import { RefObject, useCallback } from 'react';
-
-import { BrandedPermission } from '../../hooks/useAuthZ/types';
-
-import styles from './AuthZDevModal.module.css';
-
-export interface AuthZDevModalHeaderProps {
-	searchInputRef: RefObject<HTMLInputElement>;
-	search: string;
-	setSearch: (value: string) => void;
-	resourceFilter: string;
-	setResourceFilter: (value: string) => void;
-	resourceFilterItems: Array<{ value: string; label: string }>;
-	hasActiveFilter: boolean;
-	filteredPermissions: BrandedPermission[];
-	filteredOverrideCount: number;
-	overrideCount: number;
-	grantAll: (permissions?: BrandedPermission[]) => void;
-	denyAll: (permissions?: BrandedPermission[]) => void;
-	clearAllOverrides: (permissions?: BrandedPermission[]) => void;
-}
-
-export function AuthZDevModalHeader({
-	searchInputRef,
-	search,
-	setSearch,
-	resourceFilter,
-	setResourceFilter,
-	resourceFilterItems,
-	hasActiveFilter,
-	filteredPermissions,
-	filteredOverrideCount,
-	overrideCount,
-	grantAll,
-	denyAll,
-	clearAllOverrides,
-}: AuthZDevModalHeaderProps): JSX.Element {
-	const handleGrantAll = useCallback((): void => {
-		grantAll(hasActiveFilter ? filteredPermissions : undefined);
-	}, [grantAll, hasActiveFilter, filteredPermissions]);
-
-	const handleDenyAll = useCallback((): void => {
-		denyAll(hasActiveFilter ? filteredPermissions : undefined);
-	}, [denyAll, hasActiveFilter, filteredPermissions]);
-
-	const handleClearAll = useCallback((): void => {
-		clearAllOverrides(hasActiveFilter ? filteredPermissions : undefined);
-	}, [clearAllOverrides, hasActiveFilter, filteredPermissions]);
-
-	return (
-		<div className={styles.header}>
-			<div className={styles.searchRow}>
-				<div className={styles.search}>
-					<Input
-						ref={searchInputRef}
-						placeholder="Search permissions..."
-						value={search}
-						onChange={(e): void => setSearch(e.target.value)}
-						prefix={<Search size={14} className={styles.searchIcon} />}
-						aria-label="Search permissions"
-						data-testid="authz-dev-search"
-					/>
-				</div>
-				<div className={styles.filter}>
-					<SelectSimple
-						items={resourceFilterItems}
-						value={resourceFilter}
-						onChange={(value): void => setResourceFilter(value as string)}
-						testId="authz-dev-resource-filter"
-						withPortal={false}
-					/>
-				</div>
-			</div>
-			<div className={styles.actionsRow}>
-				<Button
-					className={styles.actionButton}
-					variant="outlined"
-					color="success"
-					size="sm"
-					onClick={handleGrantAll}
-					disabled={filteredPermissions.length === 0}
-					data-testid="authz-dev-grant-all"
-				>
-					{hasActiveFilter ? 'Grant filtered' : 'Grant all'}
-				</Button>
-				<Button
-					className={styles.actionButton}
-					variant="outlined"
-					color="error"
-					size="sm"
-					onClick={handleDenyAll}
-					disabled={filteredPermissions.length === 0}
-					data-testid="authz-dev-deny-all"
-				>
-					{hasActiveFilter ? 'Deny filtered' : 'Deny all'}
-				</Button>
-				<Button
-					className={styles.actionButton}
-					variant="outlined"
-					color="secondary"
-					size="sm"
-					onClick={handleClearAll}
-					disabled={
-						hasActiveFilter ? filteredOverrideCount === 0 : overrideCount === 0
-					}
-					data-testid="authz-dev-clear-all"
-				>
-					{hasActiveFilter
-						? `Clear filtered (${filteredOverrideCount})`
-						: `Clear all (${overrideCount})`}
-				</Button>
-			</div>
-		</div>
-	);
-}

frontend/src/lib/authz/devtools/AuthZDevModal/OverrideControl.module.css

@@ -1,72 +0,0 @@
-.segmented {
-	display: inline-flex;
-	align-items: center;
-	gap: var(--spacing-1);
-	padding: var(--spacing-1);
-	background: var(--l2-background);
-	border: 1px solid var(--l3-border);
-	border-radius: var(--radius-2);
-}
-
-.segment {
-	--button-height: 22px;
-	--button-padding: 0 var(--spacing-3);
-	--button-secondary-ghost-hover-foreground: var(--l2-foreground);
-	--button-variant-ghost-background-color: transparent;
-	border: none;
-	--button-border-radius: calc(var(--radius-2) - 1px);
-	transition:
-		background 120ms ease,
-		color 120ms ease;
-}
-
-.segment:not(.segmentActive):hover {
-	--button-secondary-ghost-hover-foreground: var(--l2-foreground-hover);
-	--button-variant-ghost-background-color: var(--l2-background);
-}
-
-.segmentIcon {
-	display: inline-flex;
-	align-items: center;
-}
-
-.segment.optAuto {
-	--button-secondary-ghost-hover-foreground: var(--l1-foreground);
-	--button-variant-ghost-background-color: var(--l3-background);
-}
-
-.segment.optGranted {
-	--button-secondary-ghost-hover-foreground: var(--success-foreground);
-	--button-variant-ghost-background-color: color-mix(
-		in srgb,
-		var(--accent-forest) 22%,
-		transparent
-	);
-}
-
-.segment.optDenied {
-	--button-secondary-ghost-hover-foreground: var(--danger-foreground);
-	--button-variant-ghost-background-color: color-mix(
-		in srgb,
-		var(--accent-cherry) 22%,
-		transparent
-	);
-}
-
-.segment.optDelay {
-	--button-secondary-ghost-hover-foreground: var(--warning-foreground);
-	--button-variant-ghost-background-color: color-mix(
-		in srgb,
-		var(--accent-amber) 22%,
-		transparent
-	);
-}
-
-.segment.optError {
-	--button-secondary-ghost-hover-foreground: var(--danger-foreground);
-	--button-variant-ghost-background-color: color-mix(
-		in srgb,
-		var(--accent-cherry) 22%,
-		transparent
-	);
-}

frontend/src/lib/authz/devtools/AuthZDevModal/OverrideControl.tsx

@@ -1,93 +0,0 @@
-import { Check, Clock, RotateCcw, X, Zap } from '@signozhq/icons';
-import { Typography } from '@signozhq/ui/typography';
-import cx from 'classnames';
-
-import type { BrandedPermission } from '../../hooks/useAuthZ/types';
-import { OverrideState } from '../types';
-
-import styles from './OverrideControl.module.css';
-import { Button } from '@signozhq/ui/button';
-
-type OverrideControlProps = {
-	permission: BrandedPermission;
-	value: OverrideState;
-	onSelect: (permission: BrandedPermission, state: OverrideState) => void;
-};
-
-type OverrideOption = {
-	state: OverrideState;
-	label: string;
-	icon: React.ReactNode;
-	activeClassName: string;
-};
-
-const OVERRIDE_OPTIONS: OverrideOption[] = [
-	{
-		state: OverrideState.Reset,
-		label: 'Auto',
-		icon: <RotateCcw size={13} />,
-		activeClassName: styles.optAuto,
-	},
-	{
-		state: OverrideState.Granted,
-		label: 'Grant',
-		icon: <Check size={13} />,
-		activeClassName: styles.optGranted,
-	},
-	{
-		state: OverrideState.Denied,
-		label: 'Deny',
-		icon: <X size={13} />,
-		activeClassName: styles.optDenied,
-	},
-	{
-		state: OverrideState.Delay,
-		label: 'Delay',
-		icon: <Clock size={13} />,
-		activeClassName: styles.optDelay,
-	},
-	{
-		state: OverrideState.Error,
-		label: 'Error',
-		icon: <Zap size={13} />,
-		activeClassName: styles.optError,
-	},
-];
-
-export function OverrideControl({
-	permission,
-	value,
-	onSelect,
-}: OverrideControlProps): JSX.Element {
-	return (
-		<div className={styles.segmented}>
-			{OVERRIDE_OPTIONS.map((option) => {
-				const isActive = value === option.state;
-				return (
-					<Button
-						key={option.state}
-						type="button"
-						aria-pressed={isActive}
-						aria-label={option.label}
-						title={option.label}
-						className={cx(styles.segment, {
-							[styles.segmentActive]: isActive,
-							[option.activeClassName]: isActive,
-						})}
-						variant="ghost"
-						color="secondary"
-						onClick={(): void => onSelect(permission, option.state)}
-						data-testid={`override-${option.state}-${permission}`}
-					>
-						<div className={styles.segmentIcon}>{option.icon}</div>
-						{isActive && (
-							<Typography.Text as="span" size="small" weight="medium">
-								{option.label}
-							</Typography.Text>
-						)}
-					</Button>
-				);
-			})}
-		</div>
-	);
-}

frontend/src/lib/authz/devtools/AuthZDevModal/PermissionRow.module.css

@@ -1,68 +0,0 @@
-.permissionRow {
-	display: flex;
-	align-items: center;
-	justify-content: space-between;
-	gap: var(--spacing-6);
-	padding: var(--spacing-2);
-	border: 1px solid transparent;
-	border-radius: var(--radius-2);
-	cursor: pointer;
-	transition:
-		background 120ms ease,
-		border-color 120ms ease;
-}
-
-.permissionRow:hover {
-	background: var(--l2-background-hover);
-}
-
-/* Overridden rows carry a faint full border in the override color. */
-.permissionRow.rowGranted {
-	border-color: color-mix(in srgb, var(--accent-forest) 45%, transparent);
-}
-
-.permissionRow.rowDenied {
-	border-color: color-mix(in srgb, var(--accent-cherry) 45%, transparent);
-}
-
-.permissionRow.rowDelay {
-	border-color: color-mix(in srgb, var(--accent-amber) 45%, transparent);
-}
-
-.permissionRow.rowError {
-	border-color: color-mix(in srgb, var(--accent-cherry) 45%, transparent);
-}
-
-/* Keyboard selection wins over the override border. */
-.permissionRow.isSelected {
-	border-color: var(--primary);
-}
-
-.permissionInfo {
-	display: flex;
-	flex: 1 1 auto;
-	align-items: baseline;
-	gap: var(--spacing-2);
-	min-width: 0;
-}
-
-.relation {
-	flex: 0 0 auto;
-	--typography-color: var(--accent-primary);
-}
-
-.separator {
-	flex: 0 0 auto;
-}
-
-.object {
-	flex: 0 1 auto;
-	min-width: 0;
-}
-
-.permissionMeta {
-	display: flex;
-	flex: 0 0 auto;
-	align-items: center;
-	gap: var(--spacing-5);
-}

frontend/src/lib/authz/devtools/AuthZDevModal/PermissionRow.tsx

@@ -1,114 +0,0 @@
-import { Badge, BadgeColor } from '@signozhq/ui/badge';
-import { Typography } from '@signozhq/ui/typography';
-import cx from 'classnames';
-import { memo, useCallback, useMemo } from 'react';
-
-import type { BrandedPermission } from '../../hooks/useAuthZ/types';
-import { parsePermission } from '../../hooks/useAuthZ/utils';
-import { OverrideState, type ObservedPermission } from '../types';
-
-import { OverrideControl } from './OverrideControl';
-
-import styles from './PermissionRow.module.css';
-
-type PermissionRowProps = {
-	observed: ObservedPermission;
-	override: OverrideState | undefined;
-	isSelected: boolean;
-	onSetOverride: (permission: BrandedPermission, state: OverrideState) => void;
-	onSelect: () => void;
-};
-
-const ROW_OVERRIDE_CLASSES: Record<OverrideState, string | null> = {
-	[OverrideState.Reset]: null,
-	[OverrideState.Granted]: styles.rowGranted,
-	[OverrideState.Denied]: styles.rowDenied,
-	[OverrideState.Delay]: styles.rowDelay,
-	[OverrideState.Error]: styles.rowError,
-};
-
-export const PermissionRow = memo(function PermissionRow({
-	observed,
-	override,
-	isSelected,
-	onSetOverride,
-	onSelect,
-}: PermissionRowProps): JSX.Element {
-	const currentState = override ?? OverrideState.Reset;
-
-	const { relation, objectId } = useMemo(() => {
-		const parsed = parsePermission(observed.permission);
-		const separatorIndex = parsed.object.indexOf(':');
-		return {
-			relation: parsed.relation,
-			objectId:
-				separatorIndex === -1
-					? parsed.object
-					: parsed.object.slice(separatorIndex + 1),
-		};
-	}, [observed.permission]);
-
-	const handleSetOverride = useCallback(
-		(permission: BrandedPermission, state: OverrideState): void => {
-			onSelect();
-			onSetOverride(permission, state);
-		},
-		[onSelect, onSetOverride],
-	);
-
-	let apiColor: BadgeColor = 'secondary';
-	let apiLabel = 'API ?';
-	if (observed.apiValue === true) {
-		apiColor = 'success';
-		apiLabel = 'API ✓';
-	} else if (observed.apiValue === false) {
-		apiColor = 'error';
-		apiLabel = 'API ✗';
-	}
-
-	return (
-		<div
-			className={cx(styles.permissionRow, ROW_OVERRIDE_CLASSES[currentState], {
-				[styles.isSelected]: isSelected,
-			})}
-			data-testid={`permission-row-${observed.permission}`}
-		>
-			<div className={styles.permissionInfo}>
-				<Typography.Text
-					as="span"
-					size="small"
-					weight="medium"
-					className={styles.relation}
-				>
-					{relation}
-				</Typography.Text>
-				<Typography.Text
-					as="span"
-					size="small"
-					color="muted"
-					className={styles.separator}
-				>
-					:
-				</Typography.Text>
-				<Typography.Text
-					as="span"
-					size="small"
-					truncate={1}
-					className={styles.object}
-				>
-					{objectId}
-				</Typography.Text>
-			</div>
-			<div className={styles.permissionMeta}>
-				<Badge variant="outline" color={apiColor}>
-					{apiLabel}
-				</Badge>
-				<OverrideControl
-					permission={observed.permission}
-					value={currentState}
-					onSelect={handleSetOverride}
-				/>
-			</div>
-		</div>
-	);
-});

frontend/src/lib/authz/devtools/AuthZDevModal/useAuthZDevModalData.ts

@@ -1,172 +0,0 @@
-import { useMemo, useState } from 'react';
-
-import type { BrandedPermission } from '../../hooks/useAuthZ/types';
-import { parsePermission } from '../../hooks/useAuthZ/utils';
-import type { ObservedPermission, OverrideState } from '../types';
-
-type SelectItem = {
-	value: string;
-	label: string;
-};
-
-type PermissionGroup = {
-	resource: string;
-	items: BrandedPermission[];
-};
-
-type UseAuthZDevModalDataResult = {
-	search: string;
-	setSearch: (search: string) => void;
-	resourceFilter: string;
-	setResourceFilter: (filter: string) => void;
-	observedList: ObservedPermission[];
-	resourceFilterItems: SelectItem[];
-	filteredPermissions: BrandedPermission[];
-	groups: PermissionGroup[];
-	orderedPermissions: BrandedPermission[];
-	indexByPermission: Map<string, number>;
-	hasActiveFilter: boolean;
-	filteredOverrideCount: number;
-	overrideCount: number;
-};
-
-export function useAuthZDevModalData(
-	observed: Record<string, ObservedPermission>,
-	overrides: Record<string, OverrideState>,
-): UseAuthZDevModalDataResult {
-	const [search, setSearch] = useState('');
-	const [resourceFilter, setResourceFilter] = useState<string>('all');
-
-	const observedList = useMemo(
-		() =>
-			Object.values(observed).sort((a, b) =>
-				a.permission.localeCompare(b.permission),
-			),
-		[observed],
-	);
-
-	const resources = useMemo(() => {
-		const resourceSet = new Set<string>();
-		for (const obs of observedList) {
-			const { object } = parsePermission(obs.permission);
-			const resource = object.split(':')[0];
-			resourceSet.add(resource);
-		}
-		return Array.from(resourceSet).sort();
-	}, [observedList]);
-
-	const resourceFilterItems = useMemo<SelectItem[]>(
-		() => [
-			{ value: 'all', label: 'All resources' },
-			...resources.map((resource) => ({
-				value: resource,
-				label: resource,
-			})),
-		],
-		[resources],
-	);
-
-	const filteredPermissions = useMemo(() => {
-		let filtered = observedList;
-
-		if (search) {
-			const searchLower = search.toLowerCase();
-			filtered = filtered.filter((obs) =>
-				obs.permission.toLowerCase().includes(searchLower),
-			);
-		}
-
-		if (resourceFilter !== 'all') {
-			filtered = filtered.filter((obs) => {
-				const { object } = parsePermission(obs.permission);
-				const resource = object.split(':')[0];
-				return resource === resourceFilter;
-			});
-		}
-
-		return filtered.map((obs) => obs.permission);
-	}, [observedList, search, resourceFilter]);
-
-	const { groups, orderedPermissions } = useMemo(() => {
-		const groupMap = new Map<string, BrandedPermission[]>();
-		for (const permission of filteredPermissions) {
-			const { object } = parsePermission(permission);
-			const resource = object.split(':')[0] || 'other';
-			const bucket = groupMap.get(resource);
-			if (bucket) {
-				bucket.push(permission);
-			} else {
-				groupMap.set(resource, [permission]);
-			}
-		}
-
-		const sortItems = (items: BrandedPermission[]): BrandedPermission[] =>
-			[...items].sort((a, b) => {
-				const objA = parsePermission(a).object;
-				const objB = parsePermission(b).object;
-				const idA = objA.split(':')[1] ?? '';
-				const idB = objB.split(':')[1] ?? '';
-				const isWildcardA = idA === '*';
-				const isWildcardB = idB === '*';
-
-				// Wildcards first
-				if (isWildcardA && !isWildcardB) {
-					return -1;
-				}
-				if (!isWildcardA && isWildcardB) {
-					return 1;
-				}
-
-				// Then by object ID, then by full permission
-				const idCompare = idA.localeCompare(idB);
-				if (idCompare !== 0) {
-					return idCompare;
-				}
-				return a.localeCompare(b);
-			});
-
-		const sortedGroups = Array.from(groupMap, ([resource, items]) => ({
-			resource,
-			items: sortItems(items),
-		})).sort((a, b) => a.resource.localeCompare(b.resource));
-		return {
-			groups: sortedGroups,
-			orderedPermissions: sortedGroups.flatMap((group) => group.items),
-		};
-	}, [filteredPermissions]);
-
-	const indexByPermission = useMemo(() => {
-		const map = new Map<string, number>();
-		orderedPermissions.forEach((permission, index) => {
-			map.set(permission, index);
-		});
-		return map;
-	}, [orderedPermissions]);
-
-	const hasActiveFilter = search !== '' || resourceFilter !== 'all';
-
-	const filteredOverrideCount = useMemo(() => {
-		if (!hasActiveFilter) {
-			return Object.keys(overrides).length;
-		}
-		return filteredPermissions.filter((p) => p in overrides).length;
-	}, [hasActiveFilter, overrides, filteredPermissions]);
-
-	const overrideCount = Object.keys(overrides).length;
-
-	return {
-		search,
-		setSearch,
-		resourceFilter,
-		setResourceFilter,
-		observedList,
-		resourceFilterItems,
-		filteredPermissions,
-		groups,
-		orderedPermissions,
-		indexByPermission,
-		hasActiveFilter,
-		filteredOverrideCount,
-		overrideCount,
-	};
-}

frontend/src/lib/authz/devtools/AuthZDevModal/useModalKeyboard.ts

@@ -1,174 +0,0 @@
-import { useCallback, useEffect, useState } from 'react';
-
-import type { BrandedPermission } from '../../hooks/useAuthZ/types';
-import { OverrideState, OVERRIDE_CYCLE } from '../types';
-
-type UseModalKeyboardOptions = {
-	permissions: BrandedPermission[];
-	overrides: Record<string, OverrideState>;
-	onCycle: (permission: BrandedPermission) => void;
-	onSetOverride: (permission: BrandedPermission, state: OverrideState) => void;
-	onClose: () => void;
-	searchInputRef: React.RefObject<HTMLInputElement | null>;
-};
-
-type UseModalKeyboardResult = {
-	selectedIndex: number;
-	setSelectedIndex: (index: number) => void;
-};
-
-type KeyContext = {
-	permissions: BrandedPermission[];
-	overrides: Record<string, OverrideState>;
-	selectedIndex: number;
-	setSelectedIndex: React.Dispatch<React.SetStateAction<number>>;
-	onCycle: (permission: BrandedPermission) => void;
-	onSetOverride: (permission: BrandedPermission, state: OverrideState) => void;
-};
-
-const ARROW_KEYS = new Set(['ArrowUp', 'ArrowDown', 'ArrowLeft', 'ArrowRight']);
-
-const NUMBER_KEY_INDEX: Record<string, number> = {
-	'1': 0,
-	'2': 1,
-	'3': 2,
-	'4': 3,
-	'5': 4,
-};
-
-function stepOverrideState(
-	current: OverrideState,
-	direction: number,
-): OverrideState {
-	const currentIndex = OVERRIDE_CYCLE.indexOf(current);
-	const nextIndex =
-		(currentIndex + direction + OVERRIDE_CYCLE.length) % OVERRIDE_CYCLE.length;
-	return OVERRIDE_CYCLE[nextIndex];
-}
-
-// Arrow keys stay active even while the search input is focused so the list can
-// be driven without leaving the search field.
-function handleArrowKey(key: string, ctx: KeyContext): void {
-	if (key === 'ArrowDown') {
-		ctx.setSelectedIndex((prev) =>
-			Math.min(prev + 1, ctx.permissions.length - 1),
-		);
-		return;
-	}
-	if (key === 'ArrowUp') {
-		ctx.setSelectedIndex((prev) => Math.max(prev - 1, 0));
-		return;
-	}
-	const selected = ctx.permissions[ctx.selectedIndex];
-	if (!selected) {
-		return;
-	}
-	const direction = key === 'ArrowLeft' ? -1 : 1;
-	ctx.onSetOverride(
-		selected,
-		stepOverrideState(ctx.overrides[selected] ?? OverrideState.Reset, direction),
-	);
-}
-
-// Number and space/enter shortcuts type into the search field, so they only run
-// when it is not focused. Returns whether the key was handled.
-function handleActionKey(key: string, ctx: KeyContext): boolean {
-	const selected = ctx.permissions[ctx.selectedIndex];
-	const numberIndex = NUMBER_KEY_INDEX[key];
-	if (numberIndex !== undefined) {
-		if (selected) {
-			ctx.onSetOverride(selected, OVERRIDE_CYCLE[numberIndex]);
-		}
-		return true;
-	}
-	if (key === ' ' || key === 'Enter') {
-		if (selected) {
-			ctx.onCycle(selected);
-		}
-		return true;
-	}
-	return false;
-}
-
-export function useModalKeyboard({
-	permissions,
-	overrides,
-	onCycle,
-	onSetOverride,
-	onClose,
-	searchInputRef,
-}: UseModalKeyboardOptions): UseModalKeyboardResult {
-	// Start with no selection (-1) to avoid accidental override changes from
-	// Enter keypress that opened the modal also triggering cycleOverride.
-	const [selectedIndex, setSelectedIndex] = useState(-1);
-
-	const handleKeyDown = useCallback(
-		(e: KeyboardEvent) => {
-			const isSearchFocused = document.activeElement === searchInputRef.current;
-
-			if (e.key === 'Escape') {
-				e.preventDefault();
-				onClose();
-				return;
-			}
-
-			if (e.key === '/') {
-				if (!isSearchFocused) {
-					e.preventDefault();
-					searchInputRef.current?.focus();
-				}
-				return;
-			}
-
-			const ctx: KeyContext = {
-				permissions,
-				overrides,
-				selectedIndex,
-				setSelectedIndex,
-				onCycle,
-				onSetOverride,
-			};
-
-			if (ARROW_KEYS.has(e.key)) {
-				e.preventDefault();
-				handleArrowKey(e.key, ctx);
-				return;
-			}
-
-			if (isSearchFocused) {
-				return;
-			}
-
-			if (handleActionKey(e.key, ctx)) {
-				e.preventDefault();
-			}
-		},
-		[
-			permissions,
-			overrides,
-			selectedIndex,
-			onCycle,
-			onSetOverride,
-			onClose,
-			searchInputRef,
-		],
-	);
-
-	useEffect((): (() => void) => {
-		window.addEventListener('keydown', handleKeyDown);
-		return (): void => {
-			window.removeEventListener('keydown', handleKeyDown);
-		};
-	}, [handleKeyDown]);
-
-	useEffect((): void => {
-		if (selectedIndex >= permissions.length && permissions.length > 0) {
-			setSelectedIndex(permissions.length - 1);
-		}
-	}, [permissions.length, selectedIndex]);
-
-	return {
-		selectedIndex,
-		setSelectedIndex,
-	};
-}

frontend/src/lib/authz/devtools/types.ts

@@ -1,46 +0,0 @@
-import type { BrandedPermission } from '../hooks/useAuthZ/types';
-
-export enum OverrideState {
-	Granted = 'granted',
-	Denied = 'denied',
-	Delay = 'delay',
-	Error = 'error',
-	Reset = 'reset',
-}
-
-export type ObservedPermission = {
-	permission: BrandedPermission;
-	apiValue: boolean | null;
-	lastSeen: number;
-};
-
-export type PermissionOverride = {
-	permission: BrandedPermission;
-	state: OverrideState;
-};
-
-export type AuthZDevStore = {
-	isModalOpen: boolean;
-	observed: Record<string, ObservedPermission>;
-	overrides: Record<string, OverrideState>;
-
-	openModal: () => void;
-	closeModal: () => void;
-	toggleModal: () => void;
-
-	registerObserved: (permission: BrandedPermission, apiValue: boolean) => void;
-	setOverride: (permission: BrandedPermission, state: OverrideState) => void;
-	clearOverride: (permission: BrandedPermission) => void;
-	clearAllOverrides: (permissions?: BrandedPermission[]) => void;
-	grantAll: (permissions?: BrandedPermission[]) => void;
-	denyAll: (permissions?: BrandedPermission[]) => void;
-	cycleOverride: (permission: BrandedPermission) => void;
-};
-
-export const OVERRIDE_CYCLE: OverrideState[] = [
-	OverrideState.Reset,
-	OverrideState.Granted,
-	OverrideState.Denied,
-	OverrideState.Delay,
-	OverrideState.Error,
-];

frontend/src/lib/authz/devtools/useAuthZDevStore.ts

@@ -1,137 +0,0 @@
-import { create } from 'zustand';
-import { persist } from 'zustand/middleware';
-
-import type { BrandedPermission } from '../hooks/useAuthZ/types';
-import { OverrideState, OVERRIDE_CYCLE, type AuthZDevStore } from './types';
-import { getScopedKey } from 'utils/storage';
-
-export const useAuthZDevStore = create<AuthZDevStore>()(
-	persist(
-		(set, get) => ({
-			isModalOpen: false,
-			observed: {},
-			overrides: {},
-
-			openModal: (): void => {
-				set({ isModalOpen: true });
-			},
-
-			closeModal: (): void => {
-				set({ isModalOpen: false });
-			},
-
-			toggleModal: (): void => {
-				set((state) => ({ isModalOpen: !state.isModalOpen }));
-			},
-
-			registerObserved: (
-				permission: BrandedPermission,
-				apiValue: boolean,
-			): void => {
-				set((state) => ({
-					observed: {
-						...state.observed,
-						[permission]: {
-							permission,
-							apiValue,
-							lastSeen: Date.now(),
-						},
-					},
-				}));
-			},
-
-			setOverride: (permission: BrandedPermission, state: OverrideState): void => {
-				if (state === OverrideState.Reset) {
-					get().clearOverride(permission);
-					return;
-				}
-				set((s) => ({
-					overrides: {
-						...s.overrides,
-						[permission]: state,
-					},
-				}));
-			},
-
-			clearOverride: (permission: BrandedPermission): void => {
-				set((state) => {
-					const { [permission]: _, ...rest } = state.overrides;
-					return { overrides: rest };
-				});
-			},
-
-			clearAllOverrides: (permissions?: BrandedPermission[]): void => {
-				if (permissions) {
-					set((state) => {
-						const newOverrides = { ...state.overrides };
-						for (const permission of permissions) {
-							delete newOverrides[permission];
-						}
-						return { overrides: newOverrides };
-					});
-				} else {
-					set({ overrides: {} });
-				}
-			},
-
-			grantAll: (permissions?: BrandedPermission[]): void => {
-				set((state) => {
-					const keys = permissions ?? Object.keys(state.observed);
-					const newOverrides: Record<string, OverrideState> = {
-						...state.overrides,
-					};
-					for (const key of keys) {
-						newOverrides[key] = OverrideState.Granted;
-					}
-					return { overrides: newOverrides };
-				});
-			},
-
-			denyAll: (permissions?: BrandedPermission[]): void => {
-				set((state) => {
-					const keys = permissions ?? Object.keys(state.observed);
-					const newOverrides: Record<string, OverrideState> = {
-						...state.overrides,
-					};
-					for (const key of keys) {
-						newOverrides[key] = OverrideState.Denied;
-					}
-					return { overrides: newOverrides };
-				});
-			},
-
-			cycleOverride: (permission: BrandedPermission): void => {
-				const currentOverride = get().overrides[permission] ?? OverrideState.Reset;
-				const currentIndex = OVERRIDE_CYCLE.indexOf(currentOverride);
-				const nextIndex = (currentIndex + 1) % OVERRIDE_CYCLE.length;
-				const nextState = OVERRIDE_CYCLE[nextIndex];
-				get().setOverride(permission, nextState);
-			},
-		}),
-		{
-			name: `@signoz/${getScopedKey('authz-dev-overrides')}`,
-			partialize: (state) => {
-				// Clear apiValue for permissions without active override (auto mode)
-				// since the API value can change between sessions
-				const observed: typeof state.observed = {};
-				for (const [key, obs] of Object.entries(state.observed)) {
-					observed[key] = {
-						...obs,
-						apiValue: key in state.overrides ? obs.apiValue : null,
-					};
-				}
-				return {
-					observed,
-					overrides: state.overrides,
-				};
-			},
-		},
-	),
-);
-
-export const openAuthZDevModal = (): void =>
-	useAuthZDevStore.getState().openModal();
-export const closeAuthZDevModal = (): void =>
-	useAuthZDevStore.getState().closeModal();
-export const toggleAuthZDevModal = (): void =>
-	useAuthZDevStore.getState().toggleModal();

frontend/src/lib/authz/devtools/useAuthZQueryInvalidation.ts

@@ -1,28 +0,0 @@
-import { useEffect, useRef } from 'react';
-import { useQueryClient } from 'react-query';
-
-import type { OverrideState } from './types';
-
-type Overrides = Record<string, OverrideState>;
-
-export function useAuthZQueryInvalidation(overrides: Overrides): void {
-	const queryClient = useQueryClient();
-	const prevOverridesRef = useRef<Overrides>(overrides);
-
-	useEffect(() => {
-		const prevOverrides = prevOverridesRef.current;
-		prevOverridesRef.current = overrides;
-
-		const allKeys = new Set([
-			...Object.keys(prevOverrides),
-			...Object.keys(overrides),
-		]);
-
-		for (const key of allKeys) {
-			if (prevOverrides[key] !== overrides[key]) {
-				// Reset query to initial state and trigger refetch for active observers
-				void queryClient.resetQueries(['authz', key]);
-			}
-		}
-	}, [overrides, queryClient]);
-}

frontend/src/lib/authz/hooks/useAuthZ/types.ts

@@ -89,13 +89,5 @@ export type UseAuthZResult = {
 	isFetching: boolean;
 	error: Error | null;
 	permissions: AuthZCheckResponse | null;
-	/**
-	 * True if every check is granted. False while loading or on error.
-	 */
-	allowed: boolean;
-	/**
-	 * Checks that resolved as not granted (empty while loading/error).
-	 */
-	deniedPermissions: BrandedPermission[];
 	refetchPermissions: () => void;
 };

frontend/src/lib/authz/hooks/useAuthZ/useAuthZ.authz.test.tsx

@@ -1,6 +1,5 @@
 import { ReactElement } from 'react';
-import { renderHook, waitFor, act } from '@testing-library/react';
-import { useQueryClient } from 'react-query';
+import { renderHook, waitFor } from '@testing-library/react';
 import { server } from 'mocks-server/server';
 import { rest } from 'msw';
 import { AllTheProviders } from 'tests/test-utils';
@@ -47,16 +46,12 @@ describe('useAuthZ', () => {
 
 		expect(result.current.isLoading).toBe(true);
 		expect(result.current.permissions).toBeNull();
-		expect(result.current.allowed).toBe(false);
-		expect(result.current.deniedPermissions).toStrictEqual([]);
 
 		await waitFor(() => {
 			expect(result.current.isLoading).toBe(false);
 		});
 
 		expect(result.current.permissions).toStrictEqual(expectedResponse);
-		expect(result.current.allowed).toBe(false);
-		expect(result.current.deniedPermissions).toStrictEqual([permission2]);
 	});
 
 	it('should return error and null permissions when API errors', async () => {
@@ -78,89 +73,6 @@ describe('useAuthZ', () => {
 
 		expect(result.current.error).not.toBeNull();
 		expect(result.current.permissions).toBeNull();
-		expect(result.current.allowed).toBe(false);
-		expect(result.current.deniedPermissions).toStrictEqual([]);
-	});
-
-	it('should set allowed to true when all permissions are granted', async () => {
-		const permission1 = buildPermission('read', 'role:*');
-		const permission2 = buildPermission('update', 'role:123');
-
-		server.use(
-			rest.post(AUTHZ_CHECK_URL, async (req, res, ctx) => {
-				const payload = await req.json();
-				return res(
-					ctx.status(200),
-					ctx.json(authzMockResponse(payload, [true, true])),
-				);
-			}),
-		);
-
-		const { result } = renderHook(() => useAuthZ([permission1, permission2]), {
-			wrapper,
-		});
-
-		await waitFor(() => {
-			expect(result.current.isLoading).toBe(false);
-		});
-
-		expect(result.current.allowed).toBe(true);
-		expect(result.current.deniedPermissions).toStrictEqual([]);
-	});
-
-	it('should collect all denied permissions when multiple are denied', async () => {
-		const permission1 = buildPermission('read', 'role:*');
-		const permission2 = buildPermission('update', 'role:123');
-
-		server.use(
-			rest.post(AUTHZ_CHECK_URL, async (req, res, ctx) => {
-				const payload = await req.json();
-				return res(
-					ctx.status(200),
-					ctx.json(authzMockResponse(payload, [false, false])),
-				);
-			}),
-		);
-
-		const { result } = renderHook(() => useAuthZ([permission1, permission2]), {
-			wrapper,
-		});
-
-		await waitFor(() => {
-			expect(result.current.isLoading).toBe(false);
-		});
-
-		expect(result.current.allowed).toBe(false);
-		expect(result.current.deniedPermissions).toStrictEqual([
-			permission1,
-			permission2,
-		]);
-	});
-
-	it('should not fetch when enabled is false', async () => {
-		let requestCount = 0;
-		const permission = buildPermission('read', 'role:*');
-
-		server.use(
-			rest.post(AUTHZ_CHECK_URL, async (req, res, ctx) => {
-				requestCount += 1;
-				const payload = await req.json();
-				return res(ctx.status(200), ctx.json(authzMockResponse(payload, [true])));
-			}),
-		);
-
-		const { result } = renderHook(
-			() => useAuthZ([permission], { enabled: false }),
-			{ wrapper },
-		);
-
-		await waitFor(() => {
-			expect(result.current.isLoading).toBe(false);
-		});
-
-		expect(requestCount).toBe(0);
-		expect(result.current.allowed).toBe(false);
-		expect(result.current.permissions).toStrictEqual({});
 	});
 
 	it('should refetch when permissions array changes', async () => {
@@ -562,120 +474,3 @@ describe('useAuthZ', () => {
 		expect(result2.current.permissions).not.toHaveProperty(permission1);
 	});
 });
-
-describe('useAuthZ cache invalidation', () => {
-	it('should re-render with updated data when query is invalidated', async () => {
-		const permission = buildPermission('read', 'role:*');
-
-		let requestCount = 0;
-		let shouldGrant = true;
-
-		server.use(
-			rest.post(AUTHZ_CHECK_URL, async (req, res, ctx) => {
-				requestCount++;
-				const payload = await req.json();
-				return res(
-					ctx.status(200),
-					ctx.json(authzMockResponse(payload, [shouldGrant])),
-				);
-			}),
-		);
-
-		const { result } = renderHook(
-			() => {
-				const queryClient = useQueryClient();
-				const authz = useAuthZ([permission]);
-				return { authz, queryClient };
-			},
-			{ wrapper },
-		);
-
-		await waitFor(() => {
-			expect(result.current.authz.isLoading).toBe(false);
-		});
-
-		expect(requestCount).toBe(1);
-		expect(result.current.authz.allowed).toBe(true);
-		expect(result.current.authz.permissions).toStrictEqual({
-			[permission]: { isGranted: true },
-		});
-
-		// Change server response and reset query (forces refetch)
-		shouldGrant = false;
-
-		await act(async () => {
-			await result.current.queryClient.resetQueries(['authz', permission]);
-		});
-
-		await waitFor(() => {
-			expect(result.current.authz.allowed).toBe(false);
-		});
-
-		expect(requestCount).toBe(2);
-		expect(result.current.authz.permissions).toStrictEqual({
-			[permission]: { isGranted: false },
-		});
-	});
-
-	it('should re-render all components using the same permission when invalidated', async () => {
-		const permission = buildPermission('update', 'role:123');
-
-		let requestCount = 0;
-		let shouldGrant = true;
-
-		server.use(
-			rest.post(AUTHZ_CHECK_URL, async (req, res, ctx) => {
-				requestCount++;
-				const payload = await req.json();
-				return res(
-					ctx.status(200),
-					ctx.json(authzMockResponse(payload, [shouldGrant])),
-				);
-			}),
-		);
-
-		// Two separate hooks using the same permission
-		const { result: result1 } = renderHook(
-			() => {
-				const queryClient = useQueryClient();
-				const authz = useAuthZ([permission]);
-				return { authz, queryClient };
-			},
-			{ wrapper },
-		);
-
-		const { result: result2 } = renderHook(() => useAuthZ([permission]), {
-			wrapper,
-		});
-
-		await waitFor(() => {
-			expect(result1.current.authz.isLoading).toBe(false);
-			expect(result2.current.isLoading).toBe(false);
-		});
-
-		// Both should show granted, single batched request
-		expect(requestCount).toBe(1);
-		expect(result1.current.authz.allowed).toBe(true);
-		expect(result2.current.allowed).toBe(true);
-
-		// Change server response and reset query (forces refetch)
-		shouldGrant = false;
-
-		await act(async () => {
-			await result1.current.queryClient.resetQueries(['authz', permission]);
-		});
-
-		// Both hooks should update
-		await waitFor(() => {
-			expect(result1.current.authz.allowed).toBe(false);
-			expect(result2.current.allowed).toBe(false);
-		});
-
-		expect(result1.current.authz.permissions).toStrictEqual({
-			[permission]: { isGranted: false },
-		});
-		expect(result2.current.permissions).toStrictEqual({
-			[permission]: { isGranted: false },
-		});
-	});
-});

frontend/src/lib/authz/hooks/useAuthZ/useAuthZ.tsx

@@ -1,15 +1,13 @@
 import { useCallback, useMemo } from 'react';
 import { useQueries } from 'react-query';
-import { isAxiosError } from 'axios';
 import { authzCheck } from 'api/generated/services/authz';
 import type {
 	CoretypesObjectDTO,
 	AuthtypesTransactionDTO,
 } from 'api/generated/services/sigNoz.schemas';
-import { IS_DEV, MODE } from 'lib/env';
 
 import { AUTHZ_CACHE_TIME, SINGLE_FLIGHT_WAIT_TIME_MS } from './constants';
-import type {
+import {
 	AuthZCheckResponse,
 	BrandedPermission,
 	UseAuthZOptions,
@@ -19,59 +17,6 @@ import {
 	gettableTransactionToPermission,
 	permissionToTransactionDto,
 } from './utils';
-import { OverrideState } from '../../devtools/types';
-
-let devStoreRef:
-	| typeof import('../../devtools/useAuthZDevStore').useAuthZDevStore
-	| null = null;
-
-if (IS_DEV) {
-	void import('../../devtools/useAuthZDevStore').then((mod) => {
-		devStoreRef = mod.useAuthZDevStore;
-		return mod;
-	});
-}
-
-const DEV_DELAY_MS = 2000;
-
-function getDevOverride(permission: BrandedPermission): OverrideState | null {
-	if (!IS_DEV || !devStoreRef) {
-		return null;
-	}
-	return devStoreRef.getState().overrides[permission] ?? null;
-}
-
-async function applyDevOverrideToQuery(
-	permission: BrandedPermission,
-	fetchFn: () => Promise<AuthZCheckResponse>,
-): Promise<AuthZCheckResponse> {
-	const override = getDevOverride(permission);
-
-	if (override === OverrideState.Error) {
-		throw new Error(`[AuthZ DevTools] Simulated error for: ${permission}`);
-	}
-
-	if (override === OverrideState.Delay) {
-		await new Promise((resolve) => setTimeout(resolve, DEV_DELAY_MS));
-	}
-
-	const response = await fetchFn();
-
-	if (IS_DEV && devStoreRef) {
-		const apiValue = response[permission]?.isGranted ?? false;
-		devStoreRef.getState().registerObserved(permission, apiValue);
-	}
-
-	if (override === OverrideState.Granted) {
-		return { [permission]: { isGranted: true } };
-	}
-
-	if (override === OverrideState.Denied) {
-		return { [permission]: { isGranted: false } };
-	}
-
-	return response;
-}
 
 let ctx: Promise<AuthZCheckResponse> | null;
 let pendingPermissions: BrandedPermission[] = [];
@@ -82,11 +27,10 @@ function dispatchPermission(
 	pendingPermissions.push(permission);
 
 	if (!ctx) {
-		let promiseResolve: (v: AuthZCheckResponse) => void,
-			promiseReject: (reason?: unknown) => void;
-		ctx = new Promise<AuthZCheckResponse>((resolve, reject) => {
-			promiseResolve = resolve;
-			promiseReject = reject;
+		let resolve: (v: AuthZCheckResponse) => void, reject: (reason?: any) => void;
+		ctx = new Promise<AuthZCheckResponse>((r, re) => {
+			resolve = r;
+			reject = re;
 		});
 
 		setTimeout(() => {
@@ -94,9 +38,7 @@ function dispatchPermission(
 			pendingPermissions = [];
 			ctx = null;
 
-			fetchManyPermissions(copiedPermissions)
-				.then(promiseResolve)
-				.catch(promiseReject);
+			fetchManyPermissions(copiedPermissions).then(resolve).catch(reject);
 		}, SINGLE_FLIGHT_WAIT_TIME_MS);
 	}
 
@@ -143,50 +85,19 @@ export function useAuthZ(
 			return {
 				queryKey: ['authz', permission],
 				cacheTime: AUTHZ_CACHE_TIME,
-				staleTime: AUTHZ_CACHE_TIME,
-				// Keep errored state in cache instead of refetching when new observers subscribe
-				retryOnMount: false,
-				// Only override retry in non-test mode to avoid interfering with test-utils QueryClient defaults
-				...(MODE !== 'test' && {
-					retry: (failureCount: number, error: unknown): boolean => {
-						// Don't retry simulated dev errors - they will always fail
-						if (
-							error instanceof Error &&
-							error.message.includes('[AuthZ DevTools]')
-						) {
-							return false;
-						}
-						// Don't retry server errors (5xx) - they won't recover
-						if (
-							isAxiosError(error) &&
-							error.response?.status &&
-							error.response.status >= 500
-						) {
-							return false;
-						}
-						return failureCount < 3;
-					},
-				}),
 				refetchOnMount: false,
 				refetchIntervalInBackground: false,
 				refetchOnWindowFocus: false,
 				refetchOnReconnect: true,
 				enabled,
 				queryFn: async (): Promise<AuthZCheckResponse> => {
-					const fetchFn = async (): Promise<AuthZCheckResponse> => {
-						const response = await dispatchPermission(permission);
-						return {
-							[permission]: {
-								isGranted: response[permission].isGranted,
-							},
-						};
+					const response = await dispatchPermission(permission);
+
+					return {
+						[permission]: {
+							isGranted: response[permission].isGranted,
+						},
 					};
-
-					if (IS_DEV) {
-						return applyDevOverrideToQuery(permission, fetchFn);
-					}
-
-					return fetchFn();
 				},
 			};
 		}),
@@ -196,7 +107,6 @@ export function useAuthZ(
 		() => queryResults.some((q) => q.isLoading),
 		[queryResults],
 	);
-
 	const isFetching = useMemo(
 		() => queryResults.some((q) => q.isFetching),
 		[queryResults],
@@ -229,31 +139,15 @@ export function useAuthZ(
 
 	const refetchPermissions = useCallback(() => {
 		for (const query of queryResults) {
-			void query.refetch();
+			query.refetch();
 		}
 	}, [queryResults]);
 
-	const allowed = useMemo(() => {
-		if (isLoading || error || !data) {
-			return false;
-		}
-		return permissions.every((check) => data[check]?.isGranted === true);
-	}, [permissions, data, isLoading, error]);
-
-	const deniedPermissions = useMemo(() => {
-		if (!data) {
-			return [];
-		}
-		return permissions.filter((check) => data[check]?.isGranted !== true);
-	}, [permissions, data]);
-
 	return {
 		isLoading,
 		isFetching,
 		error,
 		permissions: data ?? null,
-		allowed,
-		deniedPermissions,
 		refetchPermissions,
 	};
 }

frontend/src/lib/authz/utils/authz-test-utils.ts

@@ -168,8 +168,6 @@ export function mockUseAuthZGrantAll(
 		permissions: Object.fromEntries(
 			permissions.map((p) => [p, { isGranted: true }]),
 		) as UseAuthZResult['permissions'],
-		allowed: true,
-		deniedPermissions: [],
 		refetchPermissions: jest.fn(),
 	};
 }
@@ -185,8 +183,6 @@ export function mockUseAuthZDenyAll(
 		permissions: Object.fromEntries(
 			permissions.map((p) => [p, { isGranted: false }]),
 		) as UseAuthZResult['permissions'],
-		allowed: false,
-		deniedPermissions: permissions,
 		refetchPermissions: jest.fn(),
 	};
 }
@@ -197,23 +193,16 @@ export function mockUseAuthZGrantByPrefix(
 	permissions: BrandedPermission[],
 	options?: UseAuthZOptions,
 ) => UseAuthZResult {
-	return (permissions, _options) => {
-		const denied = permissions.filter(
-			(p) => !prefixes.some((prefix) => p.startsWith(prefix)),
-		);
-		return {
-			isLoading: false,
-			isFetching: false,
-			error: null,
-			permissions: Object.fromEntries(
-				permissions.map((p) => [
-					p,
-					{ isGranted: prefixes.some((prefix) => p.startsWith(prefix)) },
-				]),
-			) as UseAuthZResult['permissions'],
-			allowed: denied.length === 0,
-			deniedPermissions: denied,
-			refetchPermissions: jest.fn(),
-		};
-	};
+	return (permissions, _options) => ({
+		isLoading: false,
+		isFetching: false,
+		error: null,
+		permissions: Object.fromEntries(
+			permissions.map((p) => [
+				p,
+				{ isGranted: prefixes.some((prefix) => p.startsWith(prefix)) },
+			]),
+		) as UseAuthZResult['permissions'],
+		refetchPermissions: jest.fn(),
+	});
 }

frontend/src/lib/env.ts

@@ -1,3 +0,0 @@
-export const IS_DEV = import.meta.env.DEV;
-export const IS_PROD = import.meta.env.PROD;
-export const MODE = import.meta.env.MODE;

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/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
@@ -10,37 +11,93 @@ import docker
 import docker.errors
 import pytest
 from testcontainers.clickhouse import ClickHouseContainer
-from testcontainers.core.container import Network
+from testcontainers.core.container import DockerContainer, Network
 
 from fixtures import reuse, types
 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(
+    zookeeper_address: str,
+    zookeeper_port: int,
+    shard: str,
+    remote_servers: str,
+    distributed_ddl_path: str = "/clickhouse/task_queue/ddl",
+) -> str:
+    return f"""
         <clickhouse>
             <logger>
                 <level>information</level>
@@ -55,33 +112,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>{zookeeper_address}</host>
+                    <port>{zookeeper_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 +169,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(
+            zookeeper_address=coordinator.address,
+            zookeeper_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 +238,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 +308,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 +320,334 @@ def clickhouse(
     )
 
 
+@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.
+    """
+    return create_clickhouse(
+        tmpfs=tmpfs,
+        network=network,
+        keeper=zookeeper,
+        request=request,
+        pytestconfig=pytestconfig,
+    )
+
+
+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}"
+
+
+@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()
+
+
+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,
+    )
+
+
+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. `keeper` is any coordination service
+    (ZooKeeper or ClickHouse Keeper).
+    """
+    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(
+                    zookeeper_address=coordinator.address,
+                    zookeeper_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/http.py

@@ -18,19 +18,22 @@ from fixtures.logger import setup_logger
 logger = setup_logger(__name__)
 
 
-@pytest.fixture(name="zeus", scope="package")
-def zeus(
+ZEUS_NETWORK_ALIAS = "signoz-zeus-it"
+
+
+def create_zeus(
     network: Network,
     request: pytest.FixtureRequest,
     pytestconfig: pytest.Config,
+    cache_key: str = "zeus",
+    alias: str | None = None,
 ) -> types.TestContainerDocker:
-    """
-    Package-scoped fixture for running zeus
-    """
 
     def create() -> types.TestContainerDocker:
         container = WireMockContainer(image="wiremock/wiremock:2.35.1-1", secure=False)
         container.with_network(network)
+        if alias:
+            container.with_network_aliases(alias)
         container.start()
 
         return types.TestContainerDocker(
@@ -42,7 +45,7 @@ def zeus(
                     container.get_exposed_port(8080),
                 )
             },
-            container_configs={"8080": types.TestContainerUrlConfig("http", container.get_wrapped_container().name, 8080)},
+            container_configs={"8080": types.TestContainerUrlConfig("http", alias or container.get_wrapped_container().name, 8080)},
         )
 
     def delete(container: types.TestContainerDocker):
@@ -62,7 +65,7 @@ def zeus(
     return reuse.wrap(
         request,
         pytestconfig,
-        "zeus",
+        cache_key,
         lambda: types.TestContainerDocker(id="", host_configs={}, container_configs={}),
         create,
         delete,
@@ -70,6 +73,18 @@ def zeus(
     )
 
 
+@pytest.fixture(name="zeus", scope="package")
+def zeus(
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    """
+    Package-scoped fixture for running zeus
+    """
+    return create_zeus(network=network, request=request, pytestconfig=pytestconfig)
+
+
 @pytest.fixture(name="gateway", scope="package")
 def gateway(
     network: Network,

tests/fixtures/metricreduction.py

@@ -0,0 +1,51 @@
+import datetime
+from collections.abc import Sequence
+
+from fixtures.metrics import MetricsBufferSample, MetricsBufferTimeSeries
+
+
+def build_ruled_gauge_buffer(
+    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/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,47 @@
+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, 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.clickhouse import assert_spans_shards
+from fixtures.metrics import (
+    Metrics,
+    MetricsReducedSampleLast60s,
+    MetricsReducedTimeSeries,
+)
+from fixtures.querier import aligned_epoch, query_metric_values
+
+
+def test_stitch_across_epoch(
+    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 the rule activates, samples live in the raw tables; after, only
+    the reduced 60s tables have data. One query spanning the boundary must
+    stitch the two branches into a continuous series with no gap and no double
+    counting: 32 raw series at 2.0 collapse into 16 groups whose sum_last is
+    4.0, so the summed value stays 320 per step across the epoch. Enough
+    series to guarantee both shards hold data (guarded below), so the totals
+    also prove the raw and reduced joins execute shard-local."""
+    metric_name = "test_reduction_stitch"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+    services = [f"svc-{i:02d}" for i in range(16)]
+
+    # first 30 minutes: raw samples (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 60s buckets (one group per service)
+    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_space_aggregations(
+    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:
+    """Space aggregations read the reduced pre-aggregated columns: sum/avg
+    from sum_last with the count_series weight, min/max from the min/max
+    columns."""
+    metric_name = f"test_reduction_space_{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_dedup_latest_computed_at_wins(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_reduced_metrics: Callable[..., None],
+) -> None:
+    """The refreshable MVs re-emit every bucket on each refresh with a newer
+    computed_at (APPEND mode); reads must dedup to the latest version per
+    (series, bucket). Recompute the same buckets with a newer computed_at and
+    a different value: only the newer value may be counted."""
+    metric_name = "test_reduction_dedup"
+    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 buckets(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 refresh emits 1.0; a later refresh recomputes the same buckets to 5.0
+    insert_reduced_metrics(time_series, buckets(sum_last=1.0, computed_at_offset_seconds=120))
+    insert_reduced_metrics(time_series, buckets(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 buckets x 5.0 per step; 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 buckets 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_buffer_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_ruled_gauge_buffer
+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_window_reads_buffer_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_buffer_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_ruled_gauge_buffer(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
+    # is_reduced=true series rows must not join in (their fingerprints match
+    # no samples, and the ts CTE 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_window_group_by_raw_label(
+    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_buffer_groupby"
+    base_epoch = aligned_epoch(timedelta(minutes=25), step_seconds=300)
+    insert_buffer_metrics(*build_ruled_gauge_buffer(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,114 @@
+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, create_clickhouse_keeper
+from fixtures.http import ZEUS_NETWORK_ALIAS, create_zeus
+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="zeus", scope="package")
+def zeus_metricreduction(
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    return create_zeus(
+        network=network,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="zeus_metricreduction",
+        alias=ZEUS_NETWORK_ALIAS,
+    )
+
+
+@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")