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

* feat: add resizable quick filters panel with persistent width settings

* feat: add tooltip to checkbox value display for better user experience

* feat: add grip handle to resizable box for improved user interaction

.github/workflows/integrationci.yaml

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

conf/example.yaml

@@ -155,8 +155,8 @@ querier:
   cache_ttl: 168h
   # The interval for recent data that should not be cached.
   flux_interval: 5m
-  # The maximum number of concurrent queries for missing ranges.
-  max_concurrent_queries: 4
+  # The maximum number of queries a single query range request runs at once.
+  max_concurrent_queries: 8
   # When filtering logs by trace_id, clamp the query window to the trace time
   # range with padding to include slightly delayed log exports. Logs only; set
   # to 0 to disable.

docs/api/openapi.yml

@@ -24118,17 +24118,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - logs:read
-        - traces:read
-        - metrics:read
-        - audit-logs:read
-        - meter-metrics:read
+        - VIEWER
       - tokenizer:
-        - logs:read
-        - traces:read
-        - metrics:read
-        - audit-logs:read
-        - meter-metrics:read
+        - VIEWER
       summary: Query range
       tags:
       - querier
@@ -24195,17 +24187,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - logs:read
-        - traces:read
-        - metrics:read
-        - audit-logs:read
-        - meter-metrics:read
+        - VIEWER
       - tokenizer:
-        - logs:read
-        - traces:read
-        - metrics:read
-        - audit-logs:read
-        - meter-metrics:read
+        - VIEWER
       summary: Query range preview
       tags:
       - querier

frontend/src/components/QuickFilters/FilterRenderers/Checkbox/CheckboxValueRow.tsx

@@ -2,6 +2,7 @@ import { Button } from 'antd';
 import { Checkbox } from '@signozhq/ui/checkbox';
 import { Typography } from '@signozhq/ui/typography';
 import cx from 'classnames';
+import { TooltipSimple } from '@signozhq/ui/tooltip';
 
 interface CheckboxValueRowProps {
 	value: string;
@@ -46,9 +47,11 @@ function CheckboxValueRow({
 				{customRendererForValue ? (
 					customRendererForValue(value)
 				) : (
-					<Typography.Text className="value-string" truncate={1}>
-						{String(value)}
-					</Typography.Text>
+					<TooltipSimple title={value} side="top" align="center" arrow>
+						<Typography.Text className="value-string" truncate={1}>
+							{String(value)}
+						</Typography.Text>
+					</TooltipSimple>
 				)}
 				<Button type="text" className="only-btn">
 					{onlyButtonLabel}

frontend/src/constants/localStorage.ts

@@ -31,6 +31,7 @@ export enum LOCALSTORAGE {
 	METRICS_LIST_OPTIONS = 'METRICS_LIST_OPTIONS',
 	SHOW_EXCEPTIONS_QUICK_FILTERS = 'SHOW_EXCEPTIONS_QUICK_FILTERS',
 	QUICK_FILTERS_SETTINGS_ANNOUNCEMENT = 'QUICK_FILTERS_SETTINGS_ANNOUNCEMENT',
+	QUICK_FILTERS_WIDTH_LOGS = 'QUICK_FILTERS_WIDTH_LOGS',
 	FUNNEL_STEPS = 'FUNNEL_STEPS',
 	SPAN_DETAILS_PINNED_ATTRIBUTES = 'SPAN_DETAILS_PINNED_ATTRIBUTES',
 	LAST_USED_CUSTOM_TIME_RANGES = 'LAST_USED_CUSTOM_TIME_RANGES',

frontend/src/pages/LogsExplorer/LogsExplorer.styles.scss

@@ -33,15 +33,19 @@
 	}
 
 	&.filter-visible {
+		// Width is owned by ResizableBox (inline style); this section is the
+		// ResizableBox root, so it stays position: relative for the drag handle.
 		.log-quick-filter-left-section {
-			width: 260px;
 			height: 100%;
 			overflow: visible;
 			min-height: 0;
-			position: relative;
 			z-index: 2;
-			display: flex;
-			flex-direction: column;
+
+			.resizable-box__content {
+				display: flex;
+				flex-direction: column;
+				overflow: visible;
+			}
 
 			.quick-filters-container {
 				flex: 1;
@@ -50,7 +54,9 @@
 		}
 
 		.log-module-right-section {
-			width: calc(100% - 260px);
+			flex: 1;
+			width: auto;
+			min-width: 0;
 		}
 	}
 }

frontend/src/pages/LogsExplorer/index.tsx

@@ -26,6 +26,8 @@ import {
 import { useIsAIAssistantEnabled } from 'hooks/useIsAIAssistantEnabled';
 import { defaultTo, isEmpty, isNull } from 'lodash-es';
 import ErrorBoundaryFallback from 'pages/ErrorBoundaryFallback/ErrorBoundaryFallback';
+import { ResizableBox } from 'periscope/components/ResizableBox';
+import usePanelWidth from 'periscope/components/ResizableBox/usePanelWidth';
 import { EventSourceProvider } from 'providers/EventSource';
 import { Warning } from 'types/api';
 import { DataSource } from 'types/common/queryBuilder';
@@ -44,9 +46,23 @@ import { ExplorerViews } from './utils';
 
 import './LogsExplorer.styles.scss';
 
+const QUICK_FILTERS_DEFAULT_WIDTH = 260;
+const QUICK_FILTERS_MIN_WIDTH = 240;
+const QUICK_FILTERS_MAX_WIDTH = 500;
+
 function LogsExplorer(): JSX.Element {
 	const [showLiveLogs, setShowLiveLogs] = useState<boolean>(false);
 
+	const {
+		initialWidth: quickFiltersInitialWidth,
+		persistWidth: persistQuickFiltersWidth,
+	} = usePanelWidth({
+		storageKey: LOCALSTORAGE.QUICK_FILTERS_WIDTH_LOGS,
+		defaultWidth: QUICK_FILTERS_DEFAULT_WIDTH,
+		minWidth: QUICK_FILTERS_MIN_WIDTH,
+		maxWidth: QUICK_FILTERS_MAX_WIDTH,
+	});
+
 	// Get panel type from URL
 	const panelTypesFromUrl = useGetPanelTypesQueryParam(PANEL_TYPES.LIST);
 
@@ -226,14 +242,25 @@ function LogsExplorer(): JSX.Element {
 					className={cx('logs-module-page', showFilters ? 'filter-visible' : '')}
 				>
 					{showFilters && (
-						<section className={cx('log-quick-filter-left-section')}>
+						<ResizableBox
+							handle="right"
+							defaultWidth={QUICK_FILTERS_DEFAULT_WIDTH}
+							initialWidth={quickFiltersInitialWidth}
+							minWidth={QUICK_FILTERS_MIN_WIDTH}
+							maxWidth={QUICK_FILTERS_MAX_WIDTH}
+							onResize={persistQuickFiltersWidth}
+							resetToDefaultOnDoubleClick
+							withHandle
+							className="log-quick-filter-left-section"
+							handleTestId="quick-filters-resize-handle"
+						>
 							<QuickFilters
 								className="qf-logs-explorer"
 								signal={SignalType.LOGS}
 								source={QuickFiltersSource.LOGS_EXPLORER}
 								handleFilterVisibilityChange={handleFilterVisibilityChange}
 							/>
-						</section>
+						</ResizableBox>
 					)}
 					<section className={cx('log-module-right-section')}>
 						<Toolbar

frontend/src/periscope/components/ResizableBox/ResizableBox.styles.scss

@@ -18,6 +18,17 @@
 		z-index: 10;
 		background: var(--l2-border);
 
+		// Extend the interactive area beyond the 1px visual line so the handle
+		// is easy to grab and double-click, without changing its appearance.
+		&::before {
+			content: '';
+			position: absolute;
+			top: -4px;
+			right: -4px;
+			bottom: -4px;
+			left: -4px;
+		}
+
 		&:hover,
 		&:active {
 			background: var(--primary);
@@ -55,4 +66,29 @@
 			right: 0;
 		}
 	}
+
+	// Visible grip indicator (opt-in via the `withHandle` prop). Purely visual —
+	// pointer events fall through to the handle so it still owns drag + reset.
+	&__grip {
+		position: absolute;
+		top: 50%;
+		left: 50%;
+		transform: translate(-50%, -50%);
+		display: flex;
+		align-items: center;
+		justify-content: center;
+		width: 14px;
+		height: 26px;
+		border: 1px solid var(--l2-border);
+		border-radius: 4px;
+		background: var(--l1-background);
+		color: var(--l2-foreground);
+		pointer-events: none;
+	}
+
+	&__handle:hover &__grip,
+	&__handle:active &__grip {
+		border-color: var(--primary);
+		color: var(--primary);
+	}
 }

frontend/src/periscope/components/ResizableBox/ResizableBox.tsx

@@ -1,3 +1,4 @@
+import { GripVertical } from '@signozhq/icons';
 import { useCallback, useRef, useState } from 'react';
 
 import './ResizableBox.styles.scss';
@@ -11,15 +12,28 @@ export interface ResizableBoxProps {
 	// resize (width). Dragging the handle away from the content grows the box;
 	// dragging it toward the content shrinks it.
 	handle?: ResizableBoxHandle;
+	// Canonical default size, and the target that double-click reset restores to.
 	defaultHeight?: number;
 	minHeight?: number;
 	maxHeight?: number;
 	defaultWidth?: number;
 	minWidth?: number;
 	maxWidth?: number;
+	// Starting size when different from the default (e.g. a persisted value).
+	// Falls back to defaultWidth/defaultHeight when omitted, preserving the
+	// behavior of callers that don't opt in.
+	initialWidth?: number;
+	initialHeight?: number;
+	// When true, double-clicking the handle resets the size to
+	// defaultWidth/defaultHeight and fires onResize with that value.
+	resetToDefaultOnDoubleClick?: boolean;
+	// When true, renders a visible grip indicator on the handle so it is
+	// discoverable as a draggable affordance.
+	withHandle?: boolean;
 	onResize?: (size: number) => void;
 	disabled?: boolean;
 	className?: string;
+	handleTestId?: string;
 }
 
 function ResizableBox({
@@ -31,13 +45,22 @@ function ResizableBox({
 	defaultWidth = 200,
 	minWidth = 50,
 	maxWidth = Infinity,
+	initialWidth,
+	initialHeight,
+	resetToDefaultOnDoubleClick = false,
+	withHandle = false,
 	onResize,
 	disabled = false,
 	className,
+	handleTestId,
 }: ResizableBoxProps): JSX.Element {
 	const isHorizontal = handle === 'left' || handle === 'right';
 	const isStartHandle = handle === 'top' || handle === 'left';
-	const [size, setSize] = useState(isHorizontal ? defaultWidth : defaultHeight);
+	const [size, setSize] = useState(
+		isHorizontal
+			? (initialWidth ?? defaultWidth)
+			: (initialHeight ?? defaultHeight),
+	);
 	const containerRef = useRef<HTMLDivElement>(null);
 
 	const handleMouseDown = useCallback(
@@ -83,6 +106,21 @@ function ResizableBox({
 		],
 	);
 
+	const handleDoubleClick = useCallback((): void => {
+		if (!resetToDefaultOnDoubleClick) {
+			return;
+		}
+		const nextSize = isHorizontal ? defaultWidth : defaultHeight;
+		setSize(nextSize);
+		onResize?.(nextSize);
+	}, [
+		resetToDefaultOnDoubleClick,
+		isHorizontal,
+		defaultWidth,
+		defaultHeight,
+		onResize,
+	]);
+
 	const containerStyle = disabled
 		? undefined
 		: isHorizontal
@@ -99,7 +137,22 @@ function ResizableBox({
 			style={containerStyle}
 		>
 			<div className="resizable-box__content">{children}</div>
-			{!disabled && <div className={handleClass} onMouseDown={handleMouseDown} />}
+			{!disabled && (
+				<div
+					role="separator"
+					aria-orientation={isHorizontal ? 'vertical' : 'horizontal'}
+					className={handleClass}
+					onMouseDown={handleMouseDown}
+					onDoubleClick={handleDoubleClick}
+					data-testid={handleTestId}
+				>
+					{withHandle && (
+						<span className="resizable-box__grip">
+							<GripVertical size={12} />
+						</span>
+					)}
+				</div>
+			)}
 		</div>
 	);
 }

frontend/src/periscope/components/ResizableBox/__tests__/ResizableBox.test.tsx

@@ -0,0 +1,137 @@
+import { fireEvent, render, screen } from '@testing-library/react';
+
+import ResizableBox from '../ResizableBox';
+
+const HANDLE_TEST_ID = 'resize-handle';
+
+describe('ResizableBox', () => {
+	it('starts at defaultWidth when initialWidth is omitted', () => {
+		render(
+			<ResizableBox
+				handle="right"
+				defaultWidth={260}
+				handleTestId={HANDLE_TEST_ID}
+			>
+				<div>content</div>
+			</ResizableBox>,
+		);
+
+		const box = screen.getByTestId(HANDLE_TEST_ID).parentElement as HTMLElement;
+		expect(box.style.width).toBe('260px');
+	});
+
+	it('starts at initialWidth when provided', () => {
+		render(
+			<ResizableBox
+				handle="right"
+				defaultWidth={260}
+				initialWidth={340}
+				handleTestId={HANDLE_TEST_ID}
+			>
+				<div>content</div>
+			</ResizableBox>,
+		);
+
+		const box = screen.getByTestId(HANDLE_TEST_ID).parentElement as HTMLElement;
+		expect(box.style.width).toBe('340px');
+	});
+
+	it('resets to defaultWidth and fires onResize on double-click when enabled', () => {
+		const onResize = jest.fn();
+		render(
+			<ResizableBox
+				handle="right"
+				defaultWidth={260}
+				initialWidth={480}
+				onResize={onResize}
+				resetToDefaultOnDoubleClick
+				handleTestId={HANDLE_TEST_ID}
+			>
+				<div>content</div>
+			</ResizableBox>,
+		);
+
+		const handle = screen.getByTestId(HANDLE_TEST_ID);
+		const box = handle.parentElement as HTMLElement;
+		expect(box.style.width).toBe('480px');
+
+		fireEvent.doubleClick(handle);
+
+		expect(box.style.width).toBe('260px');
+		expect(onResize).toHaveBeenCalledWith(260);
+	});
+
+	it('does nothing on double-click when reset is not enabled', () => {
+		const onResize = jest.fn();
+		render(
+			<ResizableBox
+				handle="right"
+				defaultWidth={260}
+				initialWidth={480}
+				onResize={onResize}
+				handleTestId={HANDLE_TEST_ID}
+			>
+				<div>content</div>
+			</ResizableBox>,
+		);
+
+		const handle = screen.getByTestId(HANDLE_TEST_ID);
+		const box = handle.parentElement as HTMLElement;
+
+		fireEvent.doubleClick(handle);
+
+		expect(box.style.width).toBe('480px');
+		expect(onResize).not.toHaveBeenCalled();
+	});
+
+	it('renders a visible grip only when withHandle is set', () => {
+		const { rerender, container } = render(
+			<ResizableBox
+				handle="right"
+				defaultWidth={260}
+				handleTestId={HANDLE_TEST_ID}
+			>
+				<div>content</div>
+			</ResizableBox>,
+		);
+		expect(container.querySelector('.resizable-box__grip')).toBeNull();
+
+		rerender(
+			<ResizableBox
+				handle="right"
+				defaultWidth={260}
+				withHandle
+				handleTestId={HANDLE_TEST_ID}
+			>
+				<div>content</div>
+			</ResizableBox>,
+		);
+		expect(container.querySelector('.resizable-box__grip')).not.toBeNull();
+	});
+
+	it('clamps drag to maxWidth and reports the clamped size via onResize', () => {
+		const onResize = jest.fn();
+		render(
+			<ResizableBox
+				handle="right"
+				defaultWidth={260}
+				minWidth={240}
+				maxWidth={500}
+				onResize={onResize}
+				handleTestId={HANDLE_TEST_ID}
+			>
+				<div>content</div>
+			</ResizableBox>,
+		);
+
+		const handle = screen.getByTestId(HANDLE_TEST_ID);
+		const box = handle.parentElement as HTMLElement;
+
+		fireEvent.mouseDown(handle, { clientX: 0 });
+		fireEvent.mouseMove(document, { clientX: 1000 });
+		fireEvent.mouseUp(document);
+
+		expect(box.style.width).toBe('500px');
+		expect(onResize).toHaveBeenLastCalledWith(500);
+	});
+});

frontend/src/periscope/components/ResizableBox/__tests__/usePanelWidth.test.ts

@@ -0,0 +1,89 @@
+import { act, renderHook } from '@testing-library/react';
+import getLocalStorageKey from 'api/browser/localstorage/get';
+import setLocalStorageKey from 'api/browser/localstorage/set';
+import { LOCALSTORAGE } from 'constants/localStorage';
+
+import usePanelWidth from '../usePanelWidth';
+
+jest.mock('api/browser/localstorage/get');
+jest.mock('api/browser/localstorage/set');
+
+const mockedGet = getLocalStorageKey as jest.MockedFunction<
+	typeof getLocalStorageKey
+>;
+const mockedSet = setLocalStorageKey as jest.MockedFunction<
+	typeof setLocalStorageKey
+>;
+
+const ARGS = {
+	storageKey: LOCALSTORAGE.QUICK_FILTERS_WIDTH_LOGS,
+	defaultWidth: 260,
+	minWidth: 240,
+	maxWidth: 500,
+};
+
+describe('usePanelWidth', () => {
+	beforeEach(() => {
+		jest.clearAllMocks();
+		jest.useFakeTimers();
+	});
+
+	afterEach(() => {
+		jest.runOnlyPendingTimers();
+		jest.useRealTimers();
+	});
+
+	it('returns defaultWidth when nothing is persisted', () => {
+		mockedGet.mockReturnValue(null);
+		const { result } = renderHook(() => usePanelWidth(ARGS));
+		expect(result.current.initialWidth).toBe(260);
+	});
+
+	it('returns the persisted width when present', () => {
+		mockedGet.mockReturnValue('340');
+		const { result } = renderHook(() => usePanelWidth(ARGS));
+		expect(result.current.initialWidth).toBe(340);
+	});
+
+	it('clamps an out-of-bounds persisted width on read', () => {
+		mockedGet.mockReturnValue('9999');
+		const { result } = renderHook(() => usePanelWidth(ARGS));
+		expect(result.current.initialWidth).toBe(500);
+	});
+
+	it('falls back to defaultWidth for an invalid persisted value', () => {
+		mockedGet.mockReturnValue('not-a-number');
+		const { result } = renderHook(() => usePanelWidth(ARGS));
+		expect(result.current.initialWidth).toBe(260);
+	});
+
+	it('persists a clamped width (debounced)', () => {
+		mockedGet.mockReturnValue(null);
+		const { result } = renderHook(() => usePanelWidth(ARGS));
+
+		act(() => {
+			result.current.persistWidth(320);
+			jest.advanceTimersByTime(200);
+		});
+
+		expect(mockedSet).toHaveBeenCalledWith(
+			LOCALSTORAGE.QUICK_FILTERS_WIDTH_LOGS,
+			'320',
+		);
+	});
+
+	it('clamps below-min widths before persisting', () => {
+		mockedGet.mockReturnValue(null);
+		const { result } = renderHook(() => usePanelWidth(ARGS));
+
+		act(() => {
+			result.current.persistWidth(10);
+			jest.advanceTimersByTime(200);
+		});
+
+		expect(mockedSet).toHaveBeenCalledWith(
+			LOCALSTORAGE.QUICK_FILTERS_WIDTH_LOGS,
+			'240',
+		);
+	});
+});

frontend/src/periscope/components/ResizableBox/usePanelWidth.ts

@@ -0,0 +1,68 @@
+import getLocalStorageKey from 'api/browser/localstorage/get';
+import setLocalStorageKey from 'api/browser/localstorage/set';
+import { LOCALSTORAGE } from 'constants/localStorage';
+import debounce from 'lodash-es/debounce';
+import { useCallback, useMemo, useRef } from 'react';
+
+const PERSIST_DEBOUNCE_MS = 150;
+
+interface UsePanelWidthArgs {
+	/** Per-page localStorage key the width is persisted under. */
+	storageKey: LOCALSTORAGE;
+	/** Canonical default width, used when nothing is persisted. */
+	defaultWidth: number;
+	minWidth: number;
+	maxWidth: number;
+}
+
+interface UsePanelWidthReturn {
+	/** Width to start from: the persisted value (clamped) or the default. */
+	initialWidth: number;
+	/** Clamp and persist a width. Debounced to avoid a write per mousemove. */
+	persistWidth: (width: number) => void;
+}
+
+const clamp = (value: number, min: number, max: number): number =>
+	Math.min(max, Math.max(min, value));
+
+/**
+ * Per-page localStorage persistence for a resizable panel width. Mirrors the
+ * getLocalStorageKey/setLocalStorageKey idiom used for the trace span-details
+ * panel position. Pairs with ResizableBox: feed initialWidth into its
+ * initialWidth prop and persistWidth into its onResize.
+ */
+function usePanelWidth({
+	storageKey,
+	defaultWidth,
+	minWidth,
+	maxWidth,
+}: UsePanelWidthArgs): UsePanelWidthReturn {
+	// Read once on mount. Kept in a ref so a re-render doesn't re-read storage.
+	const initialWidthRef = useRef<number | null>(null);
+	if (initialWidthRef.current === null) {
+		const stored = getLocalStorageKey(storageKey);
+		const parsed = stored !== null && stored !== '' ? Number(stored) : NaN;
+		initialWidthRef.current = Number.isFinite(parsed)
+			? clamp(parsed, minWidth, maxWidth)
+			: defaultWidth;
+	}
+
+	const debouncedWrite = useMemo(
+		() =>
+			debounce((width: number): void => {
+				setLocalStorageKey(storageKey, String(width));
+			}, PERSIST_DEBOUNCE_MS),
+		[storageKey],
+	);
+
+	const persistWidth = useCallback(
+		(width: number): void => {
+			debouncedWrite(clamp(width, minWidth, maxWidth));
+		},
+		[debouncedWrite, minWidth, maxWidth],
+	);
+
+	return { initialWidth: initialWidthRef.current, persistWidth };
+}
+
+export default usePanelWidth;

pkg/apiserver/signozapiserver/querier.go

@@ -5,25 +5,12 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/http/handler"
 	"github.com/SigNoz/signoz/pkg/types"
-	"github.com/SigNoz/signoz/pkg/types/authtypes"
-	"github.com/SigNoz/signoz/pkg/types/coretypes"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/gorilla/mux"
 )
 
-func telemetryReadScopes() []string {
-	return []string{
-		coretypes.ResourceTelemetryResourceLogs.Scope(coretypes.VerbRead),
-		coretypes.ResourceTelemetryResourceTraces.Scope(coretypes.VerbRead),
-		coretypes.ResourceTelemetryResourceMetrics.Scope(coretypes.VerbRead),
-		coretypes.ResourceTelemetryResourceAuditLogs.Scope(coretypes.VerbRead),
-		coretypes.ResourceTelemetryResourceMeterMetrics.Scope(coretypes.VerbRead),
-	}
-}
-
 func (provider *provider) addQuerierRoutes(router *mux.Router) error {
-	if err := router.Handle("/api/v5/query_range", handler.New(provider.authzMiddleware.CheckResources(provider.querierHandler.QueryRange, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName, authtypes.SigNozViewerRoleName), handler.OpenAPIDef{
+	if err := router.Handle("/api/v5/query_range", handler.New(provider.authzMiddleware.ViewAccess(provider.querierHandler.QueryRange), handler.OpenAPIDef{
 		ID:                 "QueryRangeV5",
 		Tags:               []string{"querier"},
 		Summary:            "Query range",
@@ -459,17 +446,12 @@ func (provider *provider) addQuerierRoutes(router *mux.Router) error {
 		ResponseContentType: "application/json",
 		SuccessStatusCode:   http.StatusOK,
 		ErrorStatusCodes:    []int{http.StatusBadRequest},
-		SecuritySchemes:     newScopedSecuritySchemes(telemetryReadScopes()),
-	}, handler.WithResourceDefs(handler.TelemetryResourceDef{
-		Verb:      coretypes.VerbRead,
-		Category:  coretypes.ActionCategoryDataAccess,
-		Selector:  telemetrytypes.PrefixSelector,
-		Resources: telemetrytypes.QueryRangeResources,
-	}))).Methods(http.MethodPost).GetError(); err != nil {
+		SecuritySchemes:     newSecuritySchemes(types.RoleViewer),
+	})).Methods(http.MethodPost).GetError(); err != nil {
 		return err
 	}
 
-	if err := router.Handle("/api/v5/query_range/preview", handler.New(provider.authzMiddleware.CheckResources(provider.querierHandler.QueryRangePreview, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName, authtypes.SigNozViewerRoleName), handler.OpenAPIDef{
+	if err := router.Handle("/api/v5/query_range/preview", handler.New(provider.authzMiddleware.ViewAccess(provider.querierHandler.QueryRangePreview), handler.OpenAPIDef{
 		ID:                  "QueryRangePreviewV5",
 		Tags:                []string{"querier"},
 		Summary:             "Query range preview",
@@ -481,13 +463,8 @@ func (provider *provider) addQuerierRoutes(router *mux.Router) error {
 		ResponseContentType: "application/json",
 		SuccessStatusCode:   http.StatusOK,
 		ErrorStatusCodes:    []int{http.StatusBadRequest},
-		SecuritySchemes:     newScopedSecuritySchemes(telemetryReadScopes()),
-	}, handler.WithResourceDefs(handler.TelemetryResourceDef{
-		Verb:      coretypes.VerbRead,
-		Category:  coretypes.ActionCategoryDataAccess,
-		Selector:  telemetrytypes.PrefixSelector,
-		Resources: telemetrytypes.QueryRangeResources,
-	}))).Methods(http.MethodPost).GetError(); err != nil {
+		SecuritySchemes:     newSecuritySchemes(types.RoleViewer),
+	})).Methods(http.MethodPost).GetError(); err != nil {
 		return err
 	}
 

pkg/http/handler/resourcedef.go

@@ -1,9 +1,6 @@
 package handler
 
-import (
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/types/coretypes"
-)
+import "github.com/SigNoz/signoz/pkg/types/coretypes"
 
 type ResourceDef interface {
 	// resolveRequest is unexported to seal the interface. It returns a slice so a
@@ -100,31 +97,3 @@ func (def AttachDetachParentChildResourceDef) resolveRequest(ec coretypes.Extrac
 		),
 	}
 }
-
-type TelemetryResourceDef struct {
-	Verb      coretypes.Verb
-	Category  coretypes.ActionCategory
-	Selector  coretypes.SelectorFunc
-	Resources coretypes.ResourceExtractor
-}
-
-func (def TelemetryResourceDef) resolveRequest(ec coretypes.ExtractorContext) []coretypes.ResolvedResource {
-	refs, err := def.Resources(ec)
-	if err != nil {
-		return []coretypes.ResolvedResource{coretypes.NewResolvedResourceWithError(def.Verb, def.Category, err)}
-	}
-	if len(refs) == 0 {
-		return []coretypes.ResolvedResource{coretypes.NewResolvedResourceWithError(
-			def.Verb,
-			def.Category,
-			errors.NewInvalidInputf(errors.CodeInvalidInput, "request resolved to no resources"),
-		)}
-	}
-
-	resolved := make([]coretypes.ResolvedResource, 0, len(refs))
-	for _, ref := range refs {
-		resolved = append(resolved, coretypes.NewResolvedResourceWithID(def.Verb, def.Category, ref.Resource, ref.ID, def.Selector))
-	}
-
-	return resolved
-}

pkg/http/middleware/audit.go

@@ -118,10 +118,6 @@ func (middleware *Audit) emitAuditEvent(req *http.Request, writer responseCaptur
 	extractorCtx := coretypes.ExtractorContext{Request: req, ResponseBody: writer.BodyBytes()}
 
 	for _, resource := range resolved {
-		if err := resource.Err(); err != nil {
-			continue
-		}
-
 		resource.ResolveResponse(extractorCtx)
 		verb, category := resource.Verb(), resource.Category()
 

pkg/querier/config.go

@@ -7,6 +7,8 @@ import (
 	"github.com/SigNoz/signoz/pkg/factory"
 )
 
+const DefaultMaxConcurrentQueries = 8
+
 type SkipResourceFingerprint struct {
 	Enabled bool `yaml:"enabled" mapstructure:"enabled"`
 	// If count of fingerprint is above threshold, skip the fingerprint subquery and filter on main table instead.
@@ -37,7 +39,7 @@ func newConfig() factory.Config {
 		// Default values
 		CacheTTL:             168 * time.Hour,
 		FluxInterval:         5 * time.Minute,
-		MaxConcurrentQueries: 4,
+		MaxConcurrentQueries: DefaultMaxConcurrentQueries,
 		SkipResourceFingerprint: SkipResourceFingerprint{
 			Enabled:   false,
 			Threshold: 100000,

pkg/querier/querier.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/dustin/go-humanize"
 	"golang.org/x/exp/maps"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
@@ -35,20 +36,21 @@ var (
 )
 
 type querier struct {
-	logger                    *slog.Logger
-	fl                        flagger.Flagger
-	telemetryStore            telemetrystore.TelemetryStore
-	metadataStore             telemetrytypes.MetadataStore
-	promEngine                prometheus.Prometheus
-	traceStmtBuilder          qbtypes.StatementBuilder[qbtypes.TraceAggregation]
-	logStmtBuilder            qbtypes.StatementBuilder[qbtypes.LogAggregation]
-	auditStmtBuilder          qbtypes.StatementBuilder[qbtypes.LogAggregation]
-	metricStmtBuilder         qbtypes.StatementBuilder[qbtypes.MetricAggregation]
-	meterStmtBuilder          qbtypes.StatementBuilder[qbtypes.MetricAggregation]
-	traceOperatorStmtBuilder  qbtypes.TraceOperatorStatementBuilder
-	bucketCache               BucketCache
-	liveDataRefresh           time.Duration
-	builderConfig             builderConfig
+	logger                   *slog.Logger
+	fl                       flagger.Flagger
+	telemetryStore           telemetrystore.TelemetryStore
+	metadataStore            telemetrytypes.MetadataStore
+	promEngine               prometheus.Prometheus
+	traceStmtBuilder         qbtypes.StatementBuilder[qbtypes.TraceAggregation]
+	logStmtBuilder           qbtypes.StatementBuilder[qbtypes.LogAggregation]
+	auditStmtBuilder         qbtypes.StatementBuilder[qbtypes.LogAggregation]
+	metricStmtBuilder        qbtypes.StatementBuilder[qbtypes.MetricAggregation]
+	meterStmtBuilder         qbtypes.StatementBuilder[qbtypes.MetricAggregation]
+	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder
+	bucketCache              BucketCache
+	liveDataRefresh          time.Duration
+	builderConfig            builderConfig
+	maxConcurrentQueries     int
 }
 
 var _ Querier = (*querier)(nil)
@@ -67,25 +69,30 @@ func New(
 	bucketCache BucketCache,
 	flagger flagger.Flagger,
 	logTraceIDWindowPadding time.Duration,
+	maxConcurrentQueries int,
 ) *querier {
 	querierSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/querier")
+	if maxConcurrentQueries <= 0 {
+		maxConcurrentQueries = DefaultMaxConcurrentQueries
+	}
 	return &querier{
-		logger:                    querierSettings.Logger(),
-		fl:                        flagger,
-		telemetryStore:            telemetryStore,
-		metadataStore:             metadataStore,
-		promEngine:                promEngine,
-		traceStmtBuilder:          traceStmtBuilder,
-		logStmtBuilder:            logStmtBuilder,
-		auditStmtBuilder:          auditStmtBuilder,
-		metricStmtBuilder:         metricStmtBuilder,
-		meterStmtBuilder:          meterStmtBuilder,
-		traceOperatorStmtBuilder:  traceOperatorStmtBuilder,
-		bucketCache:               bucketCache,
-		liveDataRefresh:           5 * time.Second,
+		logger:                   querierSettings.Logger(),
+		fl:                       flagger,
+		telemetryStore:           telemetryStore,
+		metadataStore:            metadataStore,
+		promEngine:               promEngine,
+		traceStmtBuilder:         traceStmtBuilder,
+		logStmtBuilder:           logStmtBuilder,
+		auditStmtBuilder:         auditStmtBuilder,
+		metricStmtBuilder:        metricStmtBuilder,
+		meterStmtBuilder:         meterStmtBuilder,
+		traceOperatorStmtBuilder: traceOperatorStmtBuilder,
+		bucketCache:              bucketCache,
+		liveDataRefresh:          5 * time.Second,
 		builderConfig: builderConfig{
 			logTraceIDWindowPaddingMS: uint64(logTraceIDWindowPadding.Milliseconds()),
 		},
+		maxConcurrentQueries: maxConcurrentQueries,
 	}
 }
 
@@ -607,30 +614,40 @@ func (q *querier) run(
 		return false
 	}
 
-	for name, query := range qs {
-		// Skip cache if NoCache is set, or if cache is not available
-		if req.NoCache || q.bucketCache == nil || query.Fingerprint() == "" {
-			if req.NoCache {
-				q.logger.DebugContext(ctx, "NoCache flag set, bypassing cache", slog.String("query", name))
-			} else {
-				q.logger.InfoContext(ctx, "no bucket cache or fingerprint, executing query", slog.String("fingerprint", query.Fingerprint()))
+	names := maps.Keys(qs)
+	slices.Sort(names)
+	queryResults := make([]*qbtypes.Result, len(names))
+
+	// sem limits how many queries run at once for this request. The same
+	// limit covers the missing-range queries in executeWithCache. sem is held
+	// only while a query is running, never while waiting for other
+	// goroutines, so the two levels cannot deadlock.
+	sem := make(chan struct{}, q.maxConcurrentQueries)
+
+	eg, egCtx := errgroup.WithContext(ctx)
+	for i, name := range names {
+		query := qs[name]
+		eg.Go(func() error {
+			// Skip cache if NoCache is set, or if cache is not available
+			if req.NoCache || q.bucketCache == nil || query.Fingerprint() == "" {
+				if req.NoCache {
+					q.logger.DebugContext(egCtx, "NoCache flag set, bypassing cache", slog.String("query", name))
+				} else {
+					q.logger.InfoContext(egCtx, "no bucket cache or fingerprint, executing query", slog.String("fingerprint", query.Fingerprint()))
+				}
+				sem <- struct{}{}
+				result, err := query.Execute(egCtx)
+				<-sem
+				if err != nil {
+					return err
+				}
+				queryResults[i] = result
+				return nil
 			}
-			result, err := query.Execute(ctx)
-			qbEvent.HasData = qbEvent.HasData || hasData(result)
+
+			result, err := q.executeWithCache(egCtx, orgID, query, steps[name], sem)
 			if err != nil {
-				return nil, err
-			}
-			results[name] = result.Value
-			warnings = append(warnings, result.Warnings...)
-			warningsDocURL = result.WarningsDocURL
-			stats.RowsScanned += result.Stats.RowsScanned
-			stats.BytesScanned += result.Stats.BytesScanned
-			stats.DurationMS += result.Stats.DurationMS
-		} else {
-			result, err := q.executeWithCache(ctx, orgID, query, steps[name], req.NoCache)
-			qbEvent.HasData = qbEvent.HasData || hasData(result)
-			if err != nil {
-				return nil, err
+				return err
 			}
 			switch v := result.Value.(type) {
 			case *qbtypes.TimeSeriesData:
@@ -640,14 +657,23 @@ func (q *querier) run(
 			case *qbtypes.RawData:
 				v.QueryName = name
 			}
+			queryResults[i] = result
+			return nil
+		})
+	}
+	if err := eg.Wait(); err != nil {
+		return nil, err
+	}
 
-			results[name] = result.Value
-			warnings = append(warnings, result.Warnings...)
-			warningsDocURL = result.WarningsDocURL
-			stats.RowsScanned += result.Stats.RowsScanned
-			stats.BytesScanned += result.Stats.BytesScanned
-			stats.DurationMS += result.Stats.DurationMS
-		}
+	for i, name := range names {
+		result := queryResults[i]
+		qbEvent.HasData = qbEvent.HasData || hasData(result)
+		results[name] = result.Value
+		warnings = append(warnings, result.Warnings...)
+		warningsDocURL = result.WarningsDocURL
+		stats.RowsScanned += result.Stats.RowsScanned
+		stats.BytesScanned += result.Stats.BytesScanned
+		stats.DurationMS += result.Stats.DurationMS
 	}
 
 	gomaps.Copy(results, preseededResults)
@@ -707,8 +733,9 @@ func (q *querier) run(
 	return resp, nil
 }
 
-// executeWithCache executes a query using the bucket cache.
-func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query qbtypes.Query, step qbtypes.Step, _ bool) (*qbtypes.Result, error) {
+// executeWithCache executes a query using the bucket cache. sem limits how
+// many queries run at once for the whole request.
+func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query qbtypes.Query, step qbtypes.Step, sem chan struct{}) (*qbtypes.Result, error) {
 	// Get cached data and missing ranges
 	cachedResult, missingRanges := q.bucketCache.GetMissRanges(ctx, orgID, query, step)
 
@@ -721,7 +748,9 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 	if cachedResult == nil && len(missingRanges) == 1 {
 		startMs, endMs := query.Window()
 		if missingRanges[0].From == startMs && missingRanges[0].To == endMs {
+			sem <- struct{}{}
 			result, err := query.Execute(ctx)
+			<-sem
 			if err != nil {
 				return nil, err
 			}
@@ -740,7 +769,6 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 		slog.Int("missing_ranges_count", len(missingRanges)),
 		slog.Any("ranges", missingRanges))
 
-	sem := make(chan struct{}, 4)
 	var wg sync.WaitGroup
 
 	for i, timeRange := range missingRanges {
@@ -777,7 +805,9 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 		if err != nil {
 			// If any query failed, fall back to full execution
 			q.logger.ErrorContext(ctx, "parallel query execution failed", errors.Attr(err))
+			sem <- struct{}{}
 			result, err := query.Execute(ctx)
+			<-sem
 			if err != nil {
 				return nil, err
 			}

pkg/querier/querier_test.go

@@ -2,11 +2,13 @@ package querier
 
 import (
 	"context"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	cmock "github.com/SigNoz/clickhouse-go-mock"
 
+	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
@@ -54,7 +56,8 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
 		nil,                // traceOperatorStmtBuilder
 		nil,                // bucketCache
 		flaggertest.New(t), // flagger
-		0,
+		0,                  // logTraceIDWindowPadding
+		0,                  // maxConcurrentQueries
 	)
 
 	req := &qbtypes.QueryRangeRequest{
@@ -125,7 +128,8 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 		nil,                      // traceOperatorStmtBuilder
 		nil,                      // bucketCache
 		flaggertest.New(t),       // flagger
-		0,
+		0,                        // logTraceIDWindowPadding
+		0,                        // maxConcurrentQueries
 	)
 
 	req := &qbtypes.QueryRangeRequest{
@@ -155,3 +159,149 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, resp)
 }
+
+type fakeQuery struct {
+	execute func(ctx context.Context) (*qbtypes.Result, error)
+}
+
+func (f *fakeQuery) Fingerprint() string                                  { return "" }
+func (f *fakeQuery) Window() (uint64, uint64)                             { return 0, 0 }
+func (f *fakeQuery) Execute(ctx context.Context) (*qbtypes.Result, error) { return f.execute(ctx) }
+
+func chQueryEnvelopes(names []string) []qbtypes.QueryEnvelope {
+	envelopes := make([]qbtypes.QueryEnvelope, 0, len(names))
+	for _, name := range names {
+		envelopes = append(envelopes, qbtypes.QueryEnvelope{
+			Type: qbtypes.QueryTypeClickHouseSQL,
+			Spec: qbtypes.ClickHouseQuery{Name: name},
+		})
+	}
+	return envelopes
+}
+
+func TestRunExecutesQueriesConcurrently(t *testing.T) {
+	names := []string{"A", "B", "C", "D", "E"}
+	numQueries := len(names)
+
+	q := &querier{
+		logger:               instrumentationtest.New().Logger(),
+		maxConcurrentQueries: numQueries,
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var started atomic.Int32
+	allStarted := make(chan struct{})
+
+	qs := make(map[string]qbtypes.Query, numQueries)
+	for _, name := range names {
+		qs[name] = &fakeQuery{execute: func(ctx context.Context) (*qbtypes.Result, error) {
+			if int(started.Add(1)) == numQueries {
+				close(allStarted)
+			}
+			select {
+			case <-allStarted:
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+			return &qbtypes.Result{
+				Type:  qbtypes.RequestTypeScalar,
+				Value: &qbtypes.ScalarData{QueryName: name},
+				Stats: qbtypes.ExecStats{RowsScanned: 1, BytesScanned: 2, DurationMS: 3},
+			}, nil
+		}}
+	}
+
+	req := &qbtypes.QueryRangeRequest{
+		RequestType:    qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{Queries: chQueryEnvelopes(names)},
+	}
+	resp, err := q.run(ctx, valuer.GenerateUUID(), qs, req, nil, &qbtypes.QBEvent{}, nil)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Len(t, resp.Data.Results, numQueries)
+	assert.Equal(t, uint64(numQueries), resp.Meta.RowsScanned)
+	assert.Equal(t, uint64(2*numQueries), resp.Meta.BytesScanned)
+	assert.Equal(t, uint64(3*numQueries), resp.Meta.DurationMS)
+}
+
+func TestRunRespectsMaxConcurrentQueries(t *testing.T) {
+	const limit = 2
+	names := []string{"A", "B", "C", "D", "E", "F", "G", "H"}
+
+	q := &querier{
+		logger:               instrumentationtest.New().Logger(),
+		maxConcurrentQueries: limit,
+	}
+
+	var running, maxRunning atomic.Int32
+	qs := make(map[string]qbtypes.Query, len(names))
+	for _, name := range names {
+		qs[name] = &fakeQuery{execute: func(ctx context.Context) (*qbtypes.Result, error) {
+			cur := running.Add(1)
+			defer running.Add(-1)
+			for {
+				m := maxRunning.Load()
+				if cur <= m || maxRunning.CompareAndSwap(m, cur) {
+					break
+				}
+			}
+			time.Sleep(20 * time.Millisecond)
+			return &qbtypes.Result{
+				Type:  qbtypes.RequestTypeScalar,
+				Value: &qbtypes.ScalarData{QueryName: name},
+			}, nil
+		}}
+	}
+
+	req := &qbtypes.QueryRangeRequest{
+		RequestType:    qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{Queries: chQueryEnvelopes(names)},
+	}
+	resp, err := q.run(context.Background(), valuer.GenerateUUID(), qs, req, nil, &qbtypes.QBEvent{}, nil)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Len(t, resp.Data.Results, len(names))
+	assert.LessOrEqual(t, maxRunning.Load(), int32(limit), "running queries must not exceed maxConcurrentQueries")
+}
+
+func TestRunQueryErrorCancelsSiblings(t *testing.T) {
+	q := &querier{
+		logger:               instrumentationtest.New().Logger(),
+		maxConcurrentQueries: 4,
+	}
+
+	bStarted := make(chan struct{})
+	var bCanceled atomic.Bool
+
+	qs := map[string]qbtypes.Query{
+		// fails once B is running.
+		"A": &fakeQuery{execute: func(ctx context.Context) (*qbtypes.Result, error) {
+			select {
+			case <-bStarted:
+			case <-ctx.Done():
+			}
+			return nil, errors.NewInternalf(errors.CodeInternal, "query A failed")
+		}},
+		// blocks until its context is canceled by A's failure.
+		"B": &fakeQuery{execute: func(ctx context.Context) (*qbtypes.Result, error) {
+			close(bStarted)
+			select {
+			case <-ctx.Done():
+				bCanceled.Store(true)
+				return nil, ctx.Err()
+			case <-time.After(10 * time.Second):
+				return nil, errors.NewInternalf(errors.CodeInternal, "query B was never canceled")
+			}
+		}},
+	}
+
+	req := &qbtypes.QueryRangeRequest{
+		RequestType:    qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{Queries: chQueryEnvelopes([]string{"A", "B"})},
+	}
+	_, err := q.run(context.Background(), valuer.GenerateUUID(), qs, req, nil, &qbtypes.QBEvent{}, nil)
+	require.ErrorContains(t, err, "query A failed")
+	assert.True(t, bCanceled.Load(), "query B should be canceled once query A fails")
+}

pkg/querier/signozquerier/provider.go

@@ -193,5 +193,6 @@ func newProvider(
 		bucketCache,
 		flagger,
 		cfg.LogTraceIDWindowPadding,
+		cfg.MaxConcurrentQueries,
 	), nil
 }

pkg/query-service/rules/setups_test.go

@@ -56,6 +56,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		nil, // bucketCache
 		flagger,
 		0,
+		0, // maxConcurrentQueries (0 means default)
 	), metadataStore
 }
 
@@ -110,6 +111,7 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
 		nil,            // bucketCache
 		fl,
 		5*time.Minute, // logTraceIDWindowPadding
+		0,             // maxConcurrentQueries (0 means default)
 	)
 }
 
@@ -158,5 +160,6 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
 		nil,              // bucketCache
 		fl,
 		0,
+		0, // maxConcurrentQueries (0 means default)
 	)
 }

pkg/signoz/provider.go

@@ -218,7 +218,6 @@ func NewSQLMigrationProviderFactories(
 		sqlmigration.NewMigrateSSORoleMappingNamesFactory(sqlstore),
 		sqlmigration.NewAddMetricReductionRulesFactory(sqlstore, sqlschema),
 		sqlmigration.NewRemoveOrganizationTuplesFactory(sqlstore),
-		sqlmigration.NewAddTelemetryTuplesFactory(sqlstore),
 	)
 }
 

pkg/sqlmigration/099_add_telemetry_tuples.go

@@ -1,144 +0,0 @@
-package sqlmigration
-
-import (
-	"context"
-	"time"
-
-	"github.com/SigNoz/signoz/pkg/factory"
-	"github.com/SigNoz/signoz/pkg/sqlstore"
-	"github.com/SigNoz/signoz/pkg/types/authtypes"
-	"github.com/oklog/ulid/v2"
-	"github.com/uptrace/bun"
-	"github.com/uptrace/bun/dialect"
-	"github.com/uptrace/bun/migrate"
-)
-
-type addTelemetryTuples struct {
-	sqlstore sqlstore.SQLStore
-}
-
-func NewAddTelemetryTuplesFactory(sqlstore sqlstore.SQLStore) factory.ProviderFactory[SQLMigration, Config] {
-	return factory.NewProviderFactory(factory.MustNewName("add_telemetry_tuples"), func(ctx context.Context, ps factory.ProviderSettings, c Config) (SQLMigration, error) {
-		return &addTelemetryTuples{sqlstore: sqlstore}, nil
-	})
-}
-
-func (migration *addTelemetryTuples) Register(migrations *migrate.Migrations) error {
-	return migrations.Register(migration.Up, migration.Down)
-}
-
-func (migration *addTelemetryTuples) Up(ctx context.Context, db *bun.DB) error {
-	tx, err := db.BeginTx(ctx, nil)
-	if err != nil {
-		return err
-	}
-	defer func() { _ = tx.Rollback() }()
-
-	var storeID string
-	err = tx.QueryRowContext(ctx, `SELECT id FROM store WHERE name = ? LIMIT 1`, "signoz").Scan(&storeID)
-	if err != nil {
-		return err
-	}
-
-	var orgIDs []string
-	rows, err := tx.QueryContext(ctx, `SELECT id FROM organizations`)
-	if err != nil {
-		return err
-	}
-	defer rows.Close()
-	for rows.Next() {
-		var orgID string
-		if err := rows.Scan(&orgID); err != nil {
-			return err
-		}
-		orgIDs = append(orgIDs, orgID)
-	}
-
-	isPG := migration.sqlstore.BunDB().Dialect().Name() == dialect.PG
-
-	tuples := []migrationTuple{
-		{authtypes.SigNozAdminRoleName, "telemetryresource", "logs", "read"},
-		{authtypes.SigNozAdminRoleName, "telemetryresource", "traces", "read"},
-		{authtypes.SigNozAdminRoleName, "telemetryresource", "metrics", "read"},
-		{authtypes.SigNozAdminRoleName, "telemetryresource", "audit-logs", "read"},
-		{authtypes.SigNozAdminRoleName, "telemetryresource", "meter-metrics", "read"},
-		{authtypes.SigNozEditorRoleName, "telemetryresource", "logs", "read"},
-		{authtypes.SigNozEditorRoleName, "telemetryresource", "traces", "read"},
-		{authtypes.SigNozEditorRoleName, "telemetryresource", "metrics", "read"},
-		{authtypes.SigNozViewerRoleName, "telemetryresource", "logs", "read"},
-		{authtypes.SigNozViewerRoleName, "telemetryresource", "traces", "read"},
-		{authtypes.SigNozViewerRoleName, "telemetryresource", "metrics", "read"},
-	}
-
-	for _, orgID := range orgIDs {
-		for _, tuple := range tuples {
-			entropy := ulid.DefaultEntropy()
-			now := time.Now().UTC()
-			tupleID := ulid.MustNew(ulid.Timestamp(now), entropy).String()
-
-			objectID := "organization/" + orgID + "/" + tuple.objectName + "/*"
-			roleSubject := "organization/" + orgID + "/role/" + tuple.roleName
-
-			if isPG {
-				user := "role:" + roleSubject + "#assignee"
-				result, err := tx.ExecContext(ctx, `
-					INSERT INTO tuple (store, object_type, object_id, relation, _user, user_type, ulid, inserted_at)
-					VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-					ON CONFLICT (store, object_type, object_id, relation, _user) DO NOTHING`,
-					storeID, tuple.objectType, objectID, tuple.relation, user, "userset", tupleID, now,
-				)
-				if err != nil {
-					return err
-				}
-				rowsAffected, err := result.RowsAffected()
-				if err != nil {
-					return err
-				}
-				if rowsAffected == 0 {
-					continue
-				}
-				_, err = tx.ExecContext(ctx, `
-					INSERT INTO changelog (store, object_type, object_id, relation, _user, operation, ulid, inserted_at)
-					VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-					ON CONFLICT (store, ulid, object_type) DO NOTHING`,
-					storeID, tuple.objectType, objectID, tuple.relation, user, "TUPLE_OPERATION_WRITE", tupleID, now,
-				)
-				if err != nil {
-					return err
-				}
-			} else {
-				result, err := tx.ExecContext(ctx, `
-					INSERT INTO tuple (store, object_type, object_id, relation, user_object_type, user_object_id, user_relation, user_type, ulid, inserted_at)
-					VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-					ON CONFLICT (store, object_type, object_id, relation, user_object_type, user_object_id, user_relation) DO NOTHING`,
-					storeID, tuple.objectType, objectID, tuple.relation, "role", roleSubject, "assignee", "userset", tupleID, now,
-				)
-				if err != nil {
-					return err
-				}
-				rowsAffected, err := result.RowsAffected()
-				if err != nil {
-					return err
-				}
-				if rowsAffected == 0 {
-					continue
-				}
-				_, err = tx.ExecContext(ctx, `
-					INSERT INTO changelog (store, object_type, object_id, relation, user_object_type, user_object_id, user_relation, operation, ulid, inserted_at)
-					VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-					ON CONFLICT (store, ulid, object_type) DO NOTHING`,
-					storeID, tuple.objectType, objectID, tuple.relation, "role", roleSubject, "assignee", 0, tupleID, now,
-				)
-				if err != nil {
-					return err
-				}
-			}
-		}
-	}
-
-	return tx.Commit()
-}
-
-func (migration *addTelemetryTuples) Down(context.Context, *bun.DB) error {
-	return nil
-}

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
 }
 

pkg/types/coretypes/extractor.go

@@ -55,13 +55,6 @@ func OneID(extractor ResourceIDExtractor) ResourceIDsExtractor {
 	}}
 }
 
-type ResourceWithID struct {
-	Resource Resource
-	ID       string
-}
-
-type ResourceExtractor func(ExtractorContext) ([]ResourceWithID, error)
-
 func PathParam(name string) ResourceIDExtractor {
 	return ResourceIDExtractor{Phase: PhaseRequest, Fn: func(ec ExtractorContext) (string, error) {
 		if ec.Request == nil {

pkg/types/coretypes/registry_type.go

@@ -23,5 +23,5 @@ var (
 	TypeRole              = Type{valuer.NewString("role"), regexp.MustCompile(`^([a-z-]{1,50}|\*)$`), []Verb{VerbAssignee, VerbCreate, VerbList, VerbRead, VerbUpdate, VerbDelete, VerbAttach, VerbDetach}}
 	TypeOrganization      = Type{valuer.NewString("organization"), regexp.MustCompile(`^(^[0-9a-f]{8}(?:\-[0-9a-f]{4}){3}-[0-9a-f]{12}$|\*)$`), []Verb{VerbRead, VerbUpdate}}
 	TypeMetaResource      = Type{valuer.NewString("metaresource"), regexp.MustCompile(`^(^[0-9a-f]{8}(?:\-[0-9a-f]{4}){3}-[0-9a-f]{12}$|\*)$`), []Verb{VerbCreate, VerbList, VerbRead, VerbUpdate, VerbDelete, VerbAttach, VerbDetach}}
-	TypeTelemetryResource = Type{valuer.NewString("telemetryresource"), regexp.MustCompile(`^(\*|(builder_query|builder_sub_query|builder_trace_operator|promql|clickhouse_sql)(/([a-f0-9]{32}|\*)){0,2})$`), []Verb{VerbRead}}
+	TypeTelemetryResource = Type{valuer.NewString("telemetryresource"), regexp.MustCompile(`^\*$`), []Verb{VerbRead}}
 )

pkg/types/coretypes/resolved_resource.go

@@ -30,19 +30,6 @@ func NewResolvedResource(
 	return resolved
 }
 
-func NewResolvedResourceWithID(verb Verb, category ActionCategory, resource Resource, id string, selector SelectorFunc) ResolvedResource {
-	resolved := &resolvedResource{verb: verb, category: category, resource: resource, selector: selector}
-	if id != "" {
-		resolved.ids = []string{id}
-	}
-
-	return resolved
-}
-
-func NewResolvedResourceWithError(verb Verb, category ActionCategory, err error) ResolvedResource {
-	return &resolvedResource{verb: verb, category: category, err: err}
-}
-
 func (resolved *resolvedResource) fill(phase ExtractPhase, ec ExtractorContext) {
 	if !resolved.idExtractor.IsPhase(phase) {
 		return

pkg/types/coretypes/selector_test.go

@@ -1,42 +0,0 @@
-package coretypes
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestTelemetryResourceSelectorRegex(t *testing.T) {
-	segment := "abcdef0123456789abcdef0123456789"
-
-	valid := []string{
-		"*",
-		"builder_query",
-		"promql",
-		"clickhouse_sql",
-		"builder_trace_operator",
-		"builder_sub_query",
-		"builder_query/" + segment,
-		"builder_query/*",
-		"builder_query/" + segment + "/" + segment,
-		"builder_query/" + segment + "/*",
-	}
-	for _, value := range valid {
-		_, err := TypeTelemetryResource.Selector(value)
-		assert.NoError(t, err, "expected %q to be a valid telemetry selector", value)
-	}
-
-	invalid := []string{
-		"",
-		"builder_formula",
-		"builder_join",
-		"trace_operator",
-		"builder_query/abc",
-		"builder_query/" + segment + "/" + segment + "/" + segment,
-		"unknown_type/" + segment,
-	}
-	for _, value := range invalid {
-		_, err := TypeTelemetryResource.Selector(value)
-		assert.Error(t, err, "expected %q to be rejected as a telemetry selector", value)
-	}
-}

pkg/types/telemetrytypes/resources.go

@@ -1,93 +0,0 @@
-package telemetrytypes
-
-import (
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/types/coretypes"
-	"github.com/tidwall/gjson"
-)
-
-func QueryRangeResources(ec coretypes.ExtractorContext) ([]coretypes.ResourceWithID, error) {
-	queries := gjson.GetBytes(ec.RequestBody, "compositeQuery.queries")
-	if !queries.IsArray() || len(queries.Array()) == 0 {
-		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "composite query has no queries")
-	}
-
-	refs := make([]coretypes.ResourceWithID, 0, len(queries.Array()))
-	seen := make(map[string]struct{})
-	for _, query := range queries.Array() {
-		queryRefs, err := resourcesForQuery(query)
-		if err != nil {
-			return nil, err
-		}
-
-		for _, ref := range queryRefs {
-			key := ref.Resource.Kind().String() + ":" + ref.ID
-			if _, ok := seen[key]; ok {
-				continue
-			}
-			seen[key] = struct{}{}
-			refs = append(refs, ref)
-		}
-	}
-
-	return refs, nil
-}
-
-func resourcesForQuery(query gjson.Result) ([]coretypes.ResourceWithID, error) {
-	queryType := query.Get("type").String()
-
-	switch queryType {
-	case "builder_query", "builder_sub_query":
-		return resourcesForBuilderQuery(queryType, query.Get("spec"))
-	case "builder_trace_operator":
-		return []coretypes.ResourceWithID{{Resource: coretypes.ResourceTelemetryResourceTraces, ID: queryType}}, nil
-	case "promql":
-		return []coretypes.ResourceWithID{{Resource: coretypes.ResourceTelemetryResourceMetrics, ID: queryType}}, nil
-	case "clickhouse_sql":
-		return []coretypes.ResourceWithID{
-			{Resource: coretypes.ResourceTelemetryResourceLogs, ID: queryType},
-			{Resource: coretypes.ResourceTelemetryResourceTraces, ID: queryType},
-			{Resource: coretypes.ResourceTelemetryResourceMetrics, ID: queryType},
-		}, nil
-	case "builder_formula", "builder_join":
-		return nil, nil
-	default:
-		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported query type %q", queryType)
-	}
-}
-
-func resourcesForBuilderQuery(queryType string, spec gjson.Result) ([]coretypes.ResourceWithID, error) {
-	whereSegment := selectorSegment(spec.Get("filter.expression").String())
-	source := spec.Get("source").String()
-
-	switch spec.Get("signal").String() {
-	case SignalTraces.StringValue():
-		return []coretypes.ResourceWithID{{Resource: coretypes.ResourceTelemetryResourceTraces, ID: queryType + "/" + whereSegment}}, nil
-	case SignalLogs.StringValue():
-		resource := coretypes.ResourceTelemetryResourceLogs
-		if source == SourceAudit.StringValue() {
-			resource = coretypes.ResourceTelemetryResourceAuditLogs
-		}
-		return []coretypes.ResourceWithID{{Resource: resource, ID: queryType + "/" + whereSegment}}, nil
-	case SignalMetrics.StringValue():
-		resource := coretypes.ResourceTelemetryResourceMetrics
-		if source == SourceMeter.StringValue() {
-			resource = coretypes.ResourceTelemetryResourceMeterMetrics
-		}
-
-		refs := make([]coretypes.ResourceWithID, 0, 1)
-		for _, aggregation := range spec.Get("aggregations").Array() {
-			refs = append(refs, coretypes.ResourceWithID{
-				Resource: resource,
-				ID:       queryType + "/" + selectorSegment(aggregation.Get("metricName").String()) + "/" + whereSegment,
-			})
-		}
-		if len(refs) == 0 {
-			return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "metrics query has no aggregations")
-		}
-
-		return refs, nil
-	default:
-		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported signal %q", spec.Get("signal").String())
-	}
-}

pkg/types/telemetrytypes/resources_test.go

@@ -1,133 +0,0 @@
-package telemetrytypes
-
-import (
-	"testing"
-
-	"github.com/SigNoz/signoz/pkg/types/coretypes"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestQueryRangeResources(t *testing.T) {
-	whereSegment := selectorSegment("service.name = 'frontend'")
-	emptyWhereSegment := selectorSegment("")
-	metricSegment := selectorSegment("http.server.duration.count")
-
-	testCases := []struct {
-		name     string
-		body     string
-		expected map[string]string
-	}{
-		{
-			name: "traces builder query with filter",
-			body: `{"compositeQuery":{"queries":[{"type":"builder_query","spec":{"signal":"traces","filter":{"expression":"service.name = 'frontend'"}}}]}}`,
-			expected: map[string]string{
-				"traces": "builder_query/" + whereSegment,
-			},
-		},
-		{
-			name: "logs builder query without filter",
-			body: `{"compositeQuery":{"queries":[{"type":"builder_query","spec":{"signal":"logs"}}]}}`,
-			expected: map[string]string{
-				"logs": "builder_query/" + emptyWhereSegment,
-			},
-		},
-		{
-			name: "audit logs",
-			body: `{"compositeQuery":{"queries":[{"type":"builder_query","spec":{"signal":"logs","source":"audit"}}]}}`,
-			expected: map[string]string{
-				"audit-logs": "builder_query/" + emptyWhereSegment,
-			},
-		},
-		{
-			name: "metrics builder query",
-			body: `{"compositeQuery":{"queries":[{"type":"builder_query","spec":{"signal":"metrics","filter":{"expression":"service.name = 'frontend'"},"aggregations":[{"metricName":"http.server.duration.count"}]}}]}}`,
-			expected: map[string]string{
-				"metrics": "builder_query/" + metricSegment + "/" + whereSegment,
-			},
-		},
-		{
-			name: "meter metrics",
-			body: `{"compositeQuery":{"queries":[{"type":"builder_query","spec":{"signal":"metrics","source":"meter","aggregations":[{"metricName":"http.server.duration.count"}]}}]}}`,
-			expected: map[string]string{
-				"meter-metrics": "builder_query/" + metricSegment + "/" + emptyWhereSegment,
-			},
-		},
-		{
-			name: "promql",
-			body: `{"compositeQuery":{"queries":[{"type":"promql","spec":{"query":"sum(rate(foo[5m]))"}}]}}`,
-			expected: map[string]string{
-				"metrics": "promql",
-			},
-		},
-		{
-			name: "trace operator",
-			body: `{"compositeQuery":{"queries":[{"type":"builder_trace_operator","spec":{"expression":"A => B"}}]}}`,
-			expected: map[string]string{
-				"traces": "builder_trace_operator",
-			},
-		},
-		{
-			name: "clickhouse sql fans out to core signals",
-			body: `{"compositeQuery":{"queries":[{"type":"clickhouse_sql","spec":{"query":"SELECT 1"}}]}}`,
-			expected: map[string]string{
-				"logs":    "clickhouse_sql",
-				"traces":  "clickhouse_sql",
-				"metrics": "clickhouse_sql",
-			},
-		},
-		{
-			name: "duplicate queries deduplicated",
-			body: `{"compositeQuery":{"queries":[
-				{"type":"builder_query","spec":{"signal":"traces","filter":{"expression":"service.name = 'frontend'"}}},
-				{"type":"builder_query","spec":{"signal":"traces","filter":{"expression":"service.name = 'frontend'"}}},
-				{"type":"builder_formula","spec":{"name":"error_rate","expression":"A / B * 100"}}
-			]}}`,
-			expected: map[string]string{
-				"traces": "builder_query/" + whereSegment,
-			},
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			refs, err := QueryRangeResources(coretypes.ExtractorContext{RequestBody: []byte(testCase.body)})
-			require.NoError(t, err)
-			require.Len(t, refs, len(testCase.expected))
-
-			actual := make(map[string]string, len(refs))
-			for _, ref := range refs {
-				actual[ref.Resource.Kind().String()] = ref.ID
-			}
-			assert.Equal(t, testCase.expected, actual)
-		})
-	}
-}
-
-func TestQueryRangeResourcesErrors(t *testing.T) {
-	testCases := []struct {
-		name string
-		body string
-	}{
-		{name: "empty body", body: ``},
-		{name: "empty object", body: `{}`},
-		{name: "empty queries", body: `{"compositeQuery":{"queries":[]}}`},
-		{name: "unknown query type", body: `{"compositeQuery":{"queries":[{"type":"magic","spec":{}}]}}`},
-		{name: "missing signal", body: `{"compositeQuery":{"queries":[{"type":"builder_query","spec":{}}]}}`},
-		{name: "unknown signal", body: `{"compositeQuery":{"queries":[{"type":"builder_query","spec":{"signal":"events"}}]}}`},
-		{name: "metrics without aggregations", body: `{"compositeQuery":{"queries":[{"type":"builder_query","spec":{"signal":"metrics"}}]}}`},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			_, err := QueryRangeResources(coretypes.ExtractorContext{RequestBody: []byte(testCase.body)})
-			assert.Error(t, err)
-		})
-	}
-}
-
-func TestQueryRangeResourcesFormulaOnly(t *testing.T) {
-	refs, err := QueryRangeResources(coretypes.ExtractorContext{RequestBody: []byte(`{"compositeQuery":{"queries":[{"type":"builder_formula","spec":{"expression":"A / B"}}]}}`)})
-	require.NoError(t, err)
-	assert.Empty(t, refs)
-}

pkg/types/telemetrytypes/selector.go

@@ -1,51 +0,0 @@
-package telemetrytypes
-
-import (
-	"context"
-	"crypto/sha256"
-	"encoding/hex"
-	"strings"
-
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/types/coretypes"
-	"github.com/SigNoz/signoz/pkg/valuer"
-)
-
-var errCodeInvalidResourceID = errors.MustNewCode("invalid_resource_id")
-
-var PrefixSelector coretypes.SelectorFunc = func(_ context.Context, resource coretypes.Resource, id string, _ valuer.UUID) ([]coretypes.Selector, error) {
-	if id == "" {
-		return nil, errors.Newf(
-			errors.TypeInvalidInput,
-			errCodeInvalidResourceID,
-			"resource id is required for %s",
-			resource.Kind().String(),
-		)
-	}
-
-	segments := strings.Split(id, "/")
-	values := make([]string, 0, len(segments)+1)
-	values = append(values, id)
-	for level := len(segments) - 1; level >= 1; level-- {
-		values = append(values, strings.Join(segments[:level], "/")+"/*")
-	}
-	values = append(values, coretypes.WildCardSelectorString)
-
-	selectors := make([]coretypes.Selector, 0, len(values))
-	for _, value := range values {
-		selector, err := resource.Type().Selector(value)
-		if err != nil {
-			return nil, err
-		}
-		selectors = append(selectors, selector)
-	}
-
-	return selectors, nil
-}
-
-// Must stay stable: grant-time callers rely on producing the same segment for the same input.
-func selectorSegment(input string) string {
-	normalized := strings.Join(strings.Fields(input), " ")
-	sum := sha256.Sum256([]byte(normalized))
-	return hex.EncodeToString(sum[:16])
-}

pkg/types/telemetrytypes/selector_test.go

@@ -1,78 +0,0 @@
-package telemetrytypes
-
-import (
-	"context"
-	"testing"
-
-	"github.com/SigNoz/signoz/pkg/types/coretypes"
-	"github.com/SigNoz/signoz/pkg/valuer"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestSelectorSegment(t *testing.T) {
-	assert.Len(t, selectorSegment("service.name = 'frontend'"), 32)
-	assert.Equal(t,
-		selectorSegment("service.name    =  'frontend'"),
-		selectorSegment("service.name = 'frontend'"),
-	)
-	assert.NotEqual(t,
-		selectorSegment("service.name = 'frontend'"),
-		selectorSegment("service.name = 'backend'"),
-	)
-	assert.Equal(t, selectorSegment(""), selectorSegment("  "))
-}
-
-func TestPrefixSelector(t *testing.T) {
-	ctx := context.Background()
-	orgID := valuer.GenerateUUID()
-	metricSegment := selectorSegment("http.server.duration.count")
-	whereSegment := selectorSegment("service.name = 'frontend'")
-
-	testCases := []struct {
-		name     string
-		id       string
-		expected []string
-	}{
-		{
-			name: "two segments",
-			id:   "builder_query/" + metricSegment + "/" + whereSegment,
-			expected: []string{
-				"builder_query/" + metricSegment + "/" + whereSegment,
-				"builder_query/" + metricSegment + "/*",
-				"builder_query/*",
-				"*",
-			},
-		},
-		{
-			name: "one segment",
-			id:   "builder_query/" + whereSegment,
-			expected: []string{
-				"builder_query/" + whereSegment,
-				"builder_query/*",
-				"*",
-			},
-		},
-		{
-			name:     "query type only",
-			id:       "promql",
-			expected: []string{"promql", "*"},
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			selectors, err := PrefixSelector(ctx, coretypes.ResourceTelemetryResourceMetrics, testCase.id, orgID)
-			require.NoError(t, err)
-
-			values := make([]string, len(selectors))
-			for idx, selector := range selectors {
-				values[idx] = selector.String()
-			}
-			assert.Equal(t, testCase.expected, values)
-		})
-	}
-
-	_, err := PrefixSelector(ctx, coretypes.ResourceTelemetryResourceMetrics, "", orgID)
-	assert.Error(t, err)
-}

tests/conftest.py

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

tests/fixtures/clickhouse.py

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

tests/fixtures/keeper.py

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

tests/fixtures/metricreduction.py

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

tests/fixtures/metrics.py

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

tests/fixtures/migrator.py

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

tests/fixtures/querier.py

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

tests/fixtures/types.py

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

tests/fixtures/zookeeper.py

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

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

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

tests/integration/tests/clickhousecluster/conftest.py

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

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

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

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

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

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

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

tests/integration/tests/metricreduction/conftest.py

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