fix: keep the shared key-suggestion call site on the response envelope

fetchFieldKeysForQuery returned the unwrapped keys map, so QuerySearch guarded
on `keys` instead of `response.data.data`. That silently changed behaviour for
every explorer: a response carrying an envelope with no keys used to throw, and
had started skipping instead.

The fetcher now returns the response envelope, so the call site in the shared
component is unchanged from before apart from the fetcher swap. The generic
branch passes its response straight through; only the ai_observability branch
adapts, normalizing its nullable keys map.

docs/api/openapi.yml

@@ -8010,6 +8010,7 @@ components:
       - logs
       - metrics
       - meter
+      - ai_observability
       type: string
     SavedviewtypesUpdatableSavedView:
       properties:

frontend/src/api/generated/services/sigNoz.schemas.ts

@@ -9021,6 +9021,7 @@ export enum SavedviewtypesSourceDTO {
 	logs = 'logs',
 	metrics = 'metrics',
 	meter = 'meter',
+	ai_observability = 'ai_observability',
 }
 export interface SavedviewtypesSavedViewSpecDTO {
 	display?: SavedviewtypesDisplayDTO;

frontend/src/api/v5/queryRange/convertV5Response.ts

@@ -2,8 +2,10 @@ import { cloneDeep, isEmpty } from 'lodash-es';
 import { SuccessResponse, Warning } from 'types/api';
 import { MetricRangePayloadV3 } from 'types/api/metrics/getQueryRange';
 import {
+	BuilderQuery,
 	DistributionData,
 	MetricRangePayloadV5,
+	QueryEnvelope,
 	QueryRangeRequestV5,
 	RawData,
 	ScalarData,
@@ -11,6 +13,12 @@ import {
 } from 'types/api/v5/queryRange';
 import { QueryDataV3 } from 'types/api/widgets/getQuery';
 
+/** builder_query and builder_ai_query both carry a BuilderQuery spec, hence aggregation metadata. */
+const isBuilderQueryEnvelope = (
+	envelope: QueryEnvelope,
+): envelope is QueryEnvelope & { spec: BuilderQuery } =>
+	envelope.type === 'builder_query' || envelope.type === 'builder_ai_query';
+
 function getColName(
 	col: ScalarData['columns'][number],
 	legendMap: Record<string, string>,
@@ -409,21 +417,15 @@ export function convertV5ResponseToLegacy(
 	const v5Data = payload?.data;
 
 	const aggregationPerQuery =
-		params?.compositeQuery?.queries
-			?.filter((query) => query.type === 'builder_query')
-			.reduce(
-				(acc, query) => {
-					if (
-						query.type === 'builder_query' &&
-						'aggregations' in query.spec &&
-						query.spec.name
-					) {
-						acc[query.spec.name] = query.spec.aggregations;
-					}
-					return acc;
-				},
-				{} as Record<string, any>,
-			) || {};
+		params?.compositeQuery?.queries?.filter(isBuilderQueryEnvelope).reduce(
+			(acc, query) => {
+				if ('aggregations' in query.spec && query.spec.name) {
+					acc[query.spec.name] = query.spec.aggregations;
+				}
+				return acc;
+			},
+			{} as Record<string, any>,
+		) || {};
 
 	// clickhouse_sql queries have no aggregation metadata; their value columns
 	// are named/keyed by the real SQL alias the response carries (see getColId).

frontend/src/api/v5/queryRange/prepareQueryRangePayloadV5.test.ts

@@ -14,6 +14,7 @@ import {
 	QueryBuilderFormula as V5QueryBuilderFormula,
 	QueryEnvelope,
 	QueryRangePayloadV5,
+	RequestType,
 } from 'types/api/v5/queryRange';
 import { EQueryType } from 'types/common/dashboard';
 import { DataSource, ReduceOperators } from 'types/common/queryBuilder';
@@ -935,3 +936,41 @@ describe('convertBuilderQueriesToV5 having normalization', () => {
 		});
 	});
 });
+
+describe('convertBuilderQueriesToV5 builder query type', () => {
+	const buildEnvelope = (
+		builderQueryType: IBuilderQuery['builderQueryType'],
+		requestType: RequestType,
+	): QueryEnvelope => {
+		const [envelope] = convertBuilderQueriesToV5(
+			{
+				A: {
+					dataSource: DataSource.TRACES,
+					queryName: 'A',
+					builderQueryType,
+				} as unknown as IBuilderQuery,
+			},
+			requestType,
+		);
+		return envelope;
+	};
+
+	it.each<[RequestType]>([
+		['trace'],
+		['raw'],
+		['time_series'],
+		['scalar'],
+		['distribution'],
+	])('sends builder_ai_query for the %s request type', (requestType) => {
+		expect(buildEnvelope('builder_ai_query', requestType).type).toBe(
+			'builder_ai_query',
+		);
+	});
+
+	it.each<[string, IBuilderQuery['builderQueryType']]>([
+		['an unmarked query', undefined],
+		['an explicitly generic query', 'builder_query'],
+	])('sends builder_query for %s', (_label, builderQueryType) => {
+		expect(buildEnvelope(builderQueryType, 'trace').type).toBe('builder_query');
+	});
+});

frontend/src/api/v5/queryRange/prepareQueryRangePayloadV5.ts

@@ -364,7 +364,7 @@ export function convertBuilderQueriesToV5(
 			}
 
 			return {
-				type: 'builder_query' as QueryType,
+				type: queryData.builderQueryType ?? 'builder_query',
 				spec,
 			};
 		},

frontend/src/components/QueryBuilderV2/QueryV2/QuerySearch/QuerySearch.tsx

@@ -16,8 +16,6 @@ import { githubLight } from '@uiw/codemirror-theme-github';
 import CodeMirror, { EditorView, keymap, Prec } from '@uiw/react-codemirror';
 import { Button, Card, Collapse, Popover, Tooltip } from 'antd';
 import { Badge } from '@signozhq/ui/badge';
-import { getKeySuggestions } from 'api/querySuggestions/getKeySuggestions';
-import { getValueSuggestions } from 'api/querySuggestions/getValueSuggestion';
 import cx from 'classnames';
 import {
 	negationQueryOperatorSuggestions,
@@ -54,6 +52,12 @@ import {
 	SUGGESTION_FETCH_DEBOUNCE_MS,
 	SUGGESTIONS_SECTION,
 } from './constants';
+import {
+	fetchFieldKeysForQuery,
+	fetchFieldValuesForQuery,
+	SuggestedFieldKey,
+	SuggestedFieldKeysByName,
+} from './fieldSuggestions';
 import {
 	combineInitialAndUserExpression,
 	dedupeOptionsByLabel,
@@ -244,6 +248,12 @@ function QuerySearch({
 		[keySuggestions],
 	);
 
+	// read inside fetchValueSuggestions so resolving a key's context does not churn its deps
+	const keySuggestionsRef = useRef<QueryKeyDataSuggestionsProps[]>([]);
+	useEffect(() => {
+		keySuggestionsRef.current = keySuggestions || [];
+	}, [keySuggestions]);
+
 	const [showExamples] = useState(false);
 
 	const [cursorPos, setCursorPos] = useState({ line: 0, ch: 0 });
@@ -264,10 +274,8 @@ function QuerySearch({
 	);
 
 	// Add back the generateOptions function and useEffect
-	const generateOptions = (keys: {
-		[key: string]: QueryKeyDataSuggestionsProps[];
-	}): any[] =>
-		Object.values(keys).flatMap((items: QueryKeyDataSuggestionsProps[]) =>
+	const generateOptions = (keys: SuggestedFieldKeysByName): any[] =>
+		Object.values(keys).flatMap((items: SuggestedFieldKey[]) =>
 			items.map(({ name, fieldDataType, fieldContext }) => ({
 				label: name,
 				type: fieldDataType === 'string' ? 'keyword' : fieldDataType,
@@ -320,8 +328,9 @@ function QuerySearch({
 
 			lastFetchedKeyRef.current = searchText || '';
 
-			const response = await getKeySuggestions({
-				signal: dataSource,
+			const response = await fetchFieldKeysForQuery({
+				builderQueryType: queryData.builderQueryType,
+				dataSource,
 				searchText: searchText || '',
 				metricName: debouncedMetricName ?? undefined,
 				signalSource: signalSource as 'meter' | '',
@@ -363,6 +372,7 @@ function QuerySearch({
 			hardcodedAttributeKeys,
 			showFilterSuggestionsWithoutMetric,
 			metricNamespace,
+			queryData.builderQueryType,
 		],
 	);
 
@@ -496,21 +506,16 @@ function QuerySearch({
 			try {
 				const values = valueSuggestionsOverride
 					? await valueSuggestionsOverride(key, sanitizedSearchText)
-					: await getValueSuggestions({
+					: await fetchFieldValuesForQuery({
+							builderQueryType: queryData.builderQueryType,
+							dataSource,
 							key,
 							searchText: sanitizedSearchText,
-							signal: dataSource,
+							fieldContext: keySuggestionsRef.current.find(
+								(option) => option.label === key,
+							)?.fieldContext,
 							signalSource: signalSource as 'meter' | '',
 							metricName: debouncedMetricName ?? undefined,
-						}).then((response) => {
-							const responseData = response.data as any;
-							const data = responseData.data || {};
-							const values = data.values || {};
-							return {
-								stringValues: values.stringValues || [],
-								numberValues: values.numberValues || [],
-								complete: data.complete ?? false,
-							};
 						});
 
 				// Skip updates if component unmounted or key changed
@@ -604,6 +609,7 @@ function QuerySearch({
 			signalSource,
 			toggleSuggestions,
 			valueSuggestionsOverride,
+			queryData.builderQueryType,
 		],
 	);
 

frontend/src/components/QueryBuilderV2/QueryV2/QuerySearch/__tests__/fieldSuggestions.test.ts

@@ -0,0 +1,246 @@
+import {
+	getAIObservabilityFieldsKeys,
+	getAIObservabilityFieldsValues,
+} from 'api/generated/services/ai-observability';
+import { TelemetrytypesFieldContextDTO } from 'api/generated/services/sigNoz.schemas';
+import { getKeySuggestions } from 'api/querySuggestions/getKeySuggestions';
+import { getValueSuggestions } from 'api/querySuggestions/getValueSuggestion';
+import { DataSource } from 'types/common/queryBuilder';
+
+import {
+	fetchFieldKeysForQuery,
+	fetchFieldValuesForQuery,
+} from '../fieldSuggestions';
+
+jest.mock('api/generated/services/ai-observability', () => ({
+	getAIObservabilityFieldsKeys: jest.fn(),
+	getAIObservabilityFieldsValues: jest.fn(),
+}));
+
+jest.mock('api/querySuggestions/getKeySuggestions', () => ({
+	getKeySuggestions: jest.fn(),
+}));
+
+jest.mock('api/querySuggestions/getValueSuggestion', () => ({
+	getValueSuggestions: jest.fn(),
+}));
+
+const mockedAIKeys = getAIObservabilityFieldsKeys as jest.MockedFunction<
+	typeof getAIObservabilityFieldsKeys
+>;
+const mockedGenericKeys = getKeySuggestions as jest.MockedFunction<
+	typeof getKeySuggestions
+>;
+const mockedAIValues = getAIObservabilityFieldsValues as jest.MockedFunction<
+	typeof getAIObservabilityFieldsValues
+>;
+const mockedGenericValues = getValueSuggestions as jest.MockedFunction<
+	typeof getValueSuggestions
+>;
+
+const aiValuesResponse = (
+	values: { stringValues?: string[]; numberValues?: number[] } | null,
+	complete = true,
+): Awaited<ReturnType<typeof getAIObservabilityFieldsValues>> =>
+	({
+		status: 'success',
+		data: { complete, values },
+	}) as Awaited<ReturnType<typeof getAIObservabilityFieldsValues>>;
+
+describe('fetchFieldKeysForQuery', () => {
+	beforeEach(() => {
+		jest.clearAllMocks();
+	});
+
+	it('reads the ai_observability endpoint for a builder_ai_query', async () => {
+		mockedAIKeys.mockResolvedValue({
+			status: 'success',
+			data: {
+				complete: true,
+				keys: { llm_call_count: [{ name: 'llm_call_count' }] },
+			},
+		} as Awaited<ReturnType<typeof getAIObservabilityFieldsKeys>>);
+
+		const keys = await fetchFieldKeysForQuery({
+			builderQueryType: 'builder_ai_query',
+			dataSource: DataSource.TRACES,
+			searchText: 'llm',
+		});
+
+		expect(mockedAIKeys).toHaveBeenCalledWith({ searchText: 'llm' });
+		expect(mockedGenericKeys).not.toHaveBeenCalled();
+		expect(keys.data.data).toStrictEqual({
+			complete: true,
+			keys: { llm_call_count: [{ name: 'llm_call_count' }] },
+		});
+	});
+
+	it.each<[string, 'builder_query' | undefined]>([
+		['an unmarked query', undefined],
+		['an explicitly generic query', 'builder_query'],
+	])('reads the generic endpoint for %s', async (_label, builderQueryType) => {
+		mockedGenericKeys.mockResolvedValue({
+			data: { status: 'success', data: { complete: true, keys: {} } },
+		} as Awaited<ReturnType<typeof getKeySuggestions>>);
+
+		await fetchFieldKeysForQuery({
+			builderQueryType,
+			dataSource: DataSource.TRACES,
+			searchText: 'svc',
+		});
+
+		expect(mockedAIKeys).not.toHaveBeenCalled();
+		expect(mockedGenericKeys).toHaveBeenCalledWith(
+			expect.objectContaining({ signal: DataSource.TRACES, searchText: 'svc' }),
+		);
+	});
+
+	it('normalizes a null ai_observability keys payload to an empty map', async () => {
+		mockedAIKeys.mockResolvedValue({
+			status: 'success',
+			data: { complete: false, keys: null },
+		} as Awaited<ReturnType<typeof getAIObservabilityFieldsKeys>>);
+
+		const response = await fetchFieldKeysForQuery({
+			builderQueryType: 'builder_ai_query',
+			dataSource: DataSource.TRACES,
+			searchText: '',
+		});
+
+		expect(response.data.data).toStrictEqual({ complete: false, keys: {} });
+	});
+
+	it('passes the generic response through untouched', async () => {
+		const genericResponse = {
+			data: { status: 'success', data: { complete: true, keys: {} } },
+		} as unknown as Awaited<ReturnType<typeof getKeySuggestions>>;
+		mockedGenericKeys.mockResolvedValue(genericResponse);
+
+		await expect(
+			fetchFieldKeysForQuery({
+				builderQueryType: 'builder_query',
+				dataSource: DataSource.TRACES,
+				searchText: '',
+			}),
+		).resolves.toBe(genericResponse);
+	});
+});
+
+describe('fetchFieldValuesForQuery', () => {
+	beforeEach(() => {
+		jest.clearAllMocks();
+	});
+
+	it('reads the ai_observability endpoint for a builder_ai_query', async () => {
+		mockedAIValues.mockResolvedValue(
+			aiValuesResponse({ stringValues: ['gpt-4o'], numberValues: [] }),
+		);
+
+		const values = await fetchFieldValuesForQuery({
+			builderQueryType: 'builder_ai_query',
+			dataSource: DataSource.TRACES,
+			key: 'gen_ai.request.model',
+			searchText: 'gpt',
+			fieldContext: 'attribute',
+		});
+
+		expect(mockedGenericValues).not.toHaveBeenCalled();
+		expect(values).toStrictEqual({
+			stringValues: ['gpt-4o'],
+			numberValues: [],
+			complete: true,
+		});
+	});
+
+	it('forwards fieldContext so the endpoint can short-circuit trace aggregates', async () => {
+		mockedAIValues.mockResolvedValue(aiValuesResponse({}));
+
+		const values = await fetchFieldValuesForQuery({
+			builderQueryType: 'builder_ai_query',
+			dataSource: DataSource.TRACES,
+			key: 'total_tokens',
+			searchText: '',
+			fieldContext: 'trace',
+		});
+
+		expect(mockedAIValues).toHaveBeenCalledWith({
+			name: 'total_tokens',
+			searchText: '',
+			fieldContext: TelemetrytypesFieldContextDTO.trace,
+		});
+		expect(values).toStrictEqual({
+			stringValues: [],
+			numberValues: [],
+			complete: true,
+		});
+	});
+
+	it('normalizes a null ai_observability values payload to empty lists', async () => {
+		mockedAIValues.mockResolvedValue(aiValuesResponse(null, false));
+
+		await expect(
+			fetchFieldValuesForQuery({
+				builderQueryType: 'builder_ai_query',
+				dataSource: DataSource.TRACES,
+				key: 'llm_call_count',
+				searchText: '',
+				fieldContext: 'trace',
+			}),
+		).resolves.toStrictEqual({
+			stringValues: [],
+			numberValues: [],
+			complete: false,
+		});
+	});
+
+	it.each<[string, 'builder_query' | undefined]>([
+		['an unmarked query', undefined],
+		['an explicitly generic query', 'builder_query'],
+	])('reads the generic endpoint for %s', async (_label, builderQueryType) => {
+		mockedGenericValues.mockResolvedValue({
+			data: {
+				data: { complete: false, values: { stringValues: ['frontend'] } },
+			},
+		} as unknown as Awaited<ReturnType<typeof getValueSuggestions>>);
+
+		const values = await fetchFieldValuesForQuery({
+			builderQueryType,
+			dataSource: DataSource.TRACES,
+			key: 'service.name',
+			searchText: 'front',
+		});
+
+		expect(mockedAIValues).not.toHaveBeenCalled();
+		expect(mockedGenericValues).toHaveBeenCalledWith(
+			expect.objectContaining({
+				signal: DataSource.TRACES,
+				key: 'service.name',
+				searchText: 'front',
+			}),
+		);
+		expect(values).toStrictEqual({
+			stringValues: ['frontend'],
+			numberValues: [],
+			complete: false,
+		});
+	});
+
+	it('falls back to empty lists when the generic endpoint returns no data', async () => {
+		mockedGenericValues.mockResolvedValue({ data: {} } as unknown as Awaited<
+			ReturnType<typeof getValueSuggestions>
+		>);
+
+		await expect(
+			fetchFieldValuesForQuery({
+				builderQueryType: 'builder_query',
+				dataSource: DataSource.TRACES,
+				key: 'service.name',
+				searchText: '',
+			}),
+		).resolves.toStrictEqual({
+			stringValues: [],
+			numberValues: [],
+			complete: false,
+		});
+	});
+});

frontend/src/components/QueryBuilderV2/QueryV2/QuerySearch/fieldSuggestions.ts

@@ -0,0 +1,142 @@
+import {
+	getAIObservabilityFieldsKeys,
+	getAIObservabilityFieldsValues,
+} from 'api/generated/services/ai-observability';
+import { TelemetrytypesFieldContextDTO } from 'api/generated/services/sigNoz.schemas';
+import { getKeySuggestions } from 'api/querySuggestions/getKeySuggestions';
+import { getValueSuggestions } from 'api/querySuggestions/getValueSuggestion';
+import { IBuilderQuery } from 'types/api/queryBuilder/queryBuilderData';
+import { DataSource } from 'types/common/queryBuilder';
+
+export interface SuggestedFieldKey {
+	name: string;
+	fieldContext?: string;
+	fieldDataType?: string;
+}
+
+export type SuggestedFieldKeysByName = Record<string, SuggestedFieldKey[]>;
+
+export interface SuggestedFieldKeysPayload {
+	complete: boolean;
+	keys: SuggestedFieldKeysByName;
+}
+
+export interface SuggestedFieldKeysResponse {
+	data: { data?: SuggestedFieldKeysPayload };
+}
+
+export interface SuggestedFieldValues {
+	stringValues: string[];
+	numberValues: number[];
+	complete: boolean;
+}
+
+interface FetchFieldKeysParams {
+	builderQueryType: IBuilderQuery['builderQueryType'];
+	dataSource: DataSource;
+	searchText: string;
+	metricName?: string;
+	signalSource?: 'meter' | '';
+	metricNamespace?: string;
+}
+
+interface FetchFieldValuesParams {
+	builderQueryType: IBuilderQuery['builderQueryType'];
+	dataSource: DataSource;
+	key: string;
+	searchText: string;
+	fieldContext?: SuggestedFieldKey['fieldContext'];
+	metricName?: string;
+	signalSource?: 'meter' | '';
+}
+
+interface LegacyFieldValuesResponseData {
+	complete?: boolean;
+	values?: {
+		stringValues?: string[] | null;
+		numberValues?: number[] | null;
+	} | null;
+}
+
+const EMPTY_FIELD_VALUES: SuggestedFieldValues = {
+	stringValues: [],
+	numberValues: [],
+	complete: false,
+};
+
+export const fetchFieldKeysForQuery = async ({
+	builderQueryType,
+	dataSource,
+	searchText,
+	metricName,
+	signalSource,
+	metricNamespace,
+}: FetchFieldKeysParams): Promise<SuggestedFieldKeysResponse> => {
+	if (builderQueryType === 'builder_ai_query') {
+		const response = await getAIObservabilityFieldsKeys({ searchText });
+
+		return {
+			data: {
+				data: response.data
+					? { complete: response.data.complete, keys: response.data.keys ?? {} }
+					: undefined,
+			},
+		};
+	}
+
+	return getKeySuggestions({
+		signal: dataSource,
+		searchText,
+		metricName,
+		signalSource,
+		metricNamespace,
+	});
+};
+
+export const fetchFieldValuesForQuery = async ({
+	builderQueryType,
+	dataSource,
+	key,
+	searchText,
+	fieldContext,
+	metricName,
+	signalSource,
+}: FetchFieldValuesParams): Promise<SuggestedFieldValues> => {
+	if (builderQueryType === 'builder_ai_query') {
+		const response = await getAIObservabilityFieldsValues({
+			name: key,
+			searchText,
+			fieldContext: fieldContext as TelemetrytypesFieldContextDTO | undefined,
+		});
+
+		return {
+			stringValues: response.data?.values?.stringValues ?? [],
+			numberValues: response.data?.values?.numberValues ?? [],
+			complete: response.data?.complete ?? false,
+		};
+	}
+
+	const response = await getValueSuggestions({
+		signal: dataSource,
+		key,
+		searchText,
+		signalSource,
+		metricName,
+	});
+
+	const data = (
+		response.data as unknown as {
+			data?: LegacyFieldValuesResponseData;
+		}
+	)?.data;
+
+	if (!data) {
+		return EMPTY_FIELD_VALUES;
+	}
+
+	return {
+		stringValues: data.values?.stringValues ?? [],
+		numberValues: data.values?.numberValues ?? [],
+		complete: data.complete ?? false,
+	};
+};

frontend/src/components/QueryBuilderV2/QueryV2/QueryV2.tsx

@@ -54,7 +54,7 @@ export const QueryV2 = forwardRef(function QueryV2(
 	const { cloneQuery, panelType } = useQueryBuilder();
 
 	const showFunctions = query?.functions?.length > 0;
-	const { dataSource } = query;
+	const { dataSource, builderQueryType } = query;
 
 	const [isCollapsed, setIsCollapsed] = useState(false);
 
@@ -94,8 +94,9 @@ export const QueryV2 = forwardRef(function QueryV2(
 	);
 
 	const showSpanScopeSelector = useMemo(
-		() => dataSource === DataSource.TRACES,
-		[dataSource],
+		() =>
+			dataSource === DataSource.TRACES && builderQueryType !== 'builder_ai_query',
+		[dataSource, builderQueryType],
 	);
 
 	const showInlineQuerySearch = useMemo(() => {

frontend/src/constants/queryBuilder.ts

@@ -348,6 +348,19 @@ export const initialQueryMeterWithType: Query = {
 	},
 };
 
+export const initialQueryAIWithType: Query = {
+	...initialQueryWithType,
+	builder: {
+		...initialQueryWithType.builder,
+		queryData: [
+			{
+				...initialQueryBuilderFormValuesMap.traces,
+				builderQueryType: 'builder_ai_query',
+			},
+		],
+	},
+};
+
 export const operatorsByTypes: Record<LocalDataType, string[]> = {
 	string: Object.values(StringOperators),
 	number: Object.values(NumberOperators),

frontend/src/container/LLMObservability/Explorer/Explorer.module.scss

@@ -1,11 +0,0 @@
-.explorer {
-	display: flex;
-	flex-direction: column;
-	gap: var(--spacing-4);
-	padding: var(--spacing-2) var(--spacing-0);
-}
-
-.placeholder {
-	color: var(--l2-foreground);
-	font-size: var(--periscope-font-size-base);
-}

frontend/src/container/LLMObservability/Explorer/Explorer.styles.scss

@@ -0,0 +1,106 @@
+.trace-explorer-header {
+	.trace-explorer-run-query {
+		display: flex;
+		flex-direction: row-reverse;
+		align-items: center;
+		margin: 8px 16px;
+		gap: 8px;
+	}
+
+	.filter-outlined-btn {
+		border-radius: 0px 2px 2px 0px;
+		border-top: 1px solid var(--l1-border);
+		border-right: 1px solid var(--l1-border);
+		border-bottom: 1px solid var(--l1-border);
+		background: var(--l2-background);
+		box-shadow: 0px 0px 8px 0px rgba(0, 0, 0, 0.1);
+	}
+}
+
+.trace-explorer-header.single-child {
+	justify-content: flex-end;
+}
+
+.traces-explorer-views {
+	padding: 8px;
+	padding-bottom: 60px;
+	margin-bottom: 24px;
+
+	.ant-tabs-tabpane {
+		padding: 0 8px;
+	}
+}
+
+.qb-search-view-container {
+	padding: 8px;
+
+	display: flex;
+	flex-direction: row;
+	align-items: center;
+	gap: 8px;
+
+	.ant-select-selector {
+		border-radius: 2px;
+		border: 1px solid var(--l1-border) !important;
+		background: var(--l2-background) !important;
+		height: 34px !important;
+		box-sizing: border-box !important;
+	}
+}
+
+.trace-explorer-list-view {
+	flex: 1;
+}
+
+.trace-explorer-traces-view {
+	flex: 1;
+}
+
+.trace-explorer-table-view {
+	flex: 1;
+}
+
+.trace-explorer-time-series-view {
+	flex: 1;
+}
+
+.trace-explorer-page {
+	display: flex;
+
+	// Meant to fix the query builder colors
+	--input-background: var(--l2-background);
+	--input-hover-background: var(--l2-background);
+	--input-focus-background: var(--l2-background);
+	--input-border-color: var(--l2-border);
+	--input-hover-border-color: var(--internal-ant-border-color-hover);
+	--input-focus-border-color: var(--internal-ant-border-color-hover);
+
+	.filter {
+		width: 260px;
+		height: 100%;
+		min-height: 100vh;
+
+		border-right: 0px;
+		border: 1px solid var(--l1-border);
+		background-color: var(--l1-background);
+
+		> .ant-card-body {
+			padding: 0;
+			width: 258px;
+		}
+	}
+
+	.trace-explorer {
+		width: 100%;
+		background: var(--l1-background);
+
+		> .ant-card-body {
+			padding: 0;
+		}
+
+		border-color: var(--l1-border);
+	}
+	.trace-explorer.filters-expanded {
+		width: calc(100% - 260px);
+	}
+}

frontend/src/container/LLMObservability/Explorer/Explorer.tsx

@@ -1,13 +1,373 @@
-import styles from './Explorer.module.scss';
+import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
+import { useQueryClient } from 'react-query';
+import { useSearchParams } from 'react-router-dom-v5-compat';
+import * as Sentry from '@sentry/react';
+import { Card } from 'antd';
+import logEvent from 'api/common/logEvent';
+import cx from 'classnames';
+import ExplorerCard from 'components/ExplorerCard/ExplorerCard';
+import QueryCancelledPlaceholder from 'components/QueryCancelledPlaceholder';
+import QuickFilters from 'components/QuickFilters/QuickFilters';
+import { QuickFiltersSource, SignalType } from 'components/QuickFilters/types';
+import WarningPopover from 'components/WarningPopover/WarningPopover';
+import { AVAILABLE_EXPORT_PANEL_TYPES } from 'constants/panelTypes';
+import { initialQueryAIWithType, PANEL_TYPES } from 'constants/queryBuilder';
+import { usePageActions } from 'container/AIAssistant/pageActions/usePageActions';
+import ExplorerOptionWrapper from 'container/ExplorerOptions/ExplorerOptionWrapper';
+import { useOptionsMenu } from 'container/OptionsMenu';
+import LeftToolbarActions from 'container/QueryBuilder/components/ToolbarActions/LeftToolbarActions';
+import RightToolbarActions from 'container/QueryBuilder/components/ToolbarActions/RightToolbarActions';
+import Toolbar from 'container/Toolbar/Toolbar';
+import {
+	getExportQueryData,
+	getQueryByPanelType,
+} from 'container/TracesExplorer/explorerUtils';
+import { ExportDashboard } from 'hooks/dashboard/useExportDashboards';
+import { useGetExportToDashboardLink } from 'hooks/dashboard/useGetExportToDashboardLink';
+import { useGetPanelTypesQueryParam } from 'hooks/queryBuilder/useGetPanelTypesQueryParam';
+import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
+import { useShareBuilderUrl } from 'hooks/queryBuilder/useShareBuilderUrl';
+import {
+	ICurrentQueryData,
+	useHandleExplorerTabChange,
+} from 'hooks/useHandleExplorerTabChange';
+import { useIsAIAssistantEnabled } from 'hooks/useIsAIAssistantEnabled';
+import { useSafeNavigate } from 'hooks/useSafeNavigate';
+import { isEmpty } from 'lodash-es';
+import ErrorBoundaryFallback from 'pages/ErrorBoundaryFallback/ErrorBoundaryFallback';
+import { ExplorerViews } from 'pages/LogsExplorer/utils';
+import {
+	tracesAddFilterAction,
+	tracesChangeViewAction,
+	tracesRunQueryAction,
+	tracesSaveViewAction,
+} from 'pages/TracesExplorer/aiActions';
+import { Warning } from 'types/api';
+import { Query } from 'types/api/queryBuilder/queryBuilderData';
+import { DataSource } from 'types/common/queryBuilder';
+import {
+	explorerViewToPanelType,
+	getExplorerViewFromUrl,
+} from 'utils/explorerUtils';
+import { v4 } from 'uuid';
+
+import { DEFAULT_PANEL_TYPE, TOOLBAR_VIEWS } from './constants';
+import ListView from './ListView/ListView';
+import { defaultSelectedColumns } from './ListView/configs';
+import QuerySection from './QuerySection/QuerySection';
+import TableView from './TableView/TableView';
+import TimeSeriesView from './TimeSeriesView/TimeSeriesView';
+import TracesView from './TracesView/TracesView';
+
+import './Explorer.styles.scss';
 
-// Shell for the AI Observability Explorer tab. Owns the
-// /ai-observability/explorer route and is intentionally empty for now: the
-// query builder + results surface land in a follow-up.
 function Explorer(): JSX.Element {
+	const {
+		panelType,
+		updateAllQueriesOperators,
+		handleRunQuery,
+		stagedQuery,
+		handleSetConfig,
+		currentQuery,
+		handleSetQueryData,
+		redirectWithQueryBuilderData,
+	} = useQueryBuilder();
+
+	const isAIAssistantEnabled = useIsAIAssistantEnabled();
+
+	const { options } = useOptionsMenu({
+		dataSource: DataSource.TRACES,
+		aggregateOperator: 'noop',
+		initialOptions: {
+			selectColumns: defaultSelectedColumns,
+		},
+	});
+
+	const [searchParams] = useSearchParams();
+	const queryClient = useQueryClient();
+	const listQueryKeyRef = useRef<any>();
+
+	// Get panel type from URL
+	const panelTypesFromUrl = useGetPanelTypesQueryParam(DEFAULT_PANEL_TYPE);
+	const [isLoadingQueries, setIsLoadingQueries] = useState<boolean>(false);
+	const [isCancelled, setIsCancelled] = useState(false);
+
+	useEffect(() => {
+		if (isLoadingQueries) {
+			setIsCancelled(false);
+		}
+	}, [isLoadingQueries]);
+
+	const handleCancelQuery = useCallback(() => {
+		if (listQueryKeyRef.current) {
+			queryClient.cancelQueries(listQueryKeyRef.current);
+		}
+		setIsCancelled(true);
+		// Reset loading state — the active view unmounts when cancelled, so no
+		// child will call setIsLoadingQueries(false) otherwise.
+		setIsLoadingQueries(false);
+	}, [queryClient]);
+
+	const [selectedView, setSelectedView] = useState<ExplorerViews>(() =>
+		getExplorerViewFromUrl(searchParams, panelTypesFromUrl),
+	);
+
+	const [warning, setWarning] = useState<Warning | undefined>();
+	const [isOpen, setOpen] = useState<boolean>(true);
+
+	const defaultQuery = useMemo(
+		(): Query =>
+			updateAllQueriesOperators(
+				initialQueryAIWithType,
+				DEFAULT_PANEL_TYPE,
+				DataSource.TRACES,
+			),
+		[updateAllQueriesOperators],
+	);
+
+	const { handleExplorerTabChange } = useHandleExplorerTabChange();
+	const { safeNavigate } = useSafeNavigate();
+	const getExportToDashboardLink = useGetExportToDashboardLink();
+
+	const handleChangeSelectedView = useCallback(
+		(view: ExplorerViews, querySearchParameters?: ICurrentQueryData): void => {
+			handleSetConfig(explorerViewToPanelType[view], DataSource.TRACES);
+
+			setSelectedView(view);
+
+			handleExplorerTabChange(
+				explorerViewToPanelType[view],
+				querySearchParameters,
+			);
+		},
+		[handleExplorerTabChange, handleSetConfig],
+	);
+
+	// ─── AI Assistant page actions (only when license feature is on) ───────────
+	const aiActions = useMemo(
+		() =>
+			isAIAssistantEnabled
+				? [
+						tracesRunQueryAction({
+							currentQuery,
+							handleSetQueryData,
+							redirectWithQueryBuilderData,
+						}),
+						tracesAddFilterAction({
+							currentQuery,
+							handleSetQueryData,
+							redirectWithQueryBuilderData,
+						}),
+						tracesChangeViewAction({
+							onChangeView: (view) => handleChangeSelectedView(view as ExplorerViews),
+						}),
+						tracesSaveViewAction({
+							// POC stub — logs a save request; wire to real API when available
+							onSaveView: async (name) => {
+								// eslint-disable-next-line no-console
+								console.info('[AI Assistant] Save view requested:', name);
+							},
+						}),
+					]
+				: [],
+		// eslint-disable-next-line react-hooks/exhaustive-deps
+		[
+			isAIAssistantEnabled,
+			currentQuery,
+			handleSetQueryData,
+			redirectWithQueryBuilderData,
+			handleChangeSelectedView,
+		],
+	);
+	usePageActions('traces-explorer', aiActions);
+	// ───────────────────────────────────────────────────────────────────────────
+
+	const exportDefaultQuery = useMemo(
+		() =>
+			getQueryByPanelType(
+				stagedQuery || initialQueryAIWithType,
+				panelType || DEFAULT_PANEL_TYPE,
+			),
+		[stagedQuery, panelType],
+	);
+
+	const handleExport = useCallback(
+		(dashboard: ExportDashboard | null, isNewDashboard?: boolean): void => {
+			if (!dashboard || !panelType) {
+				return;
+			}
+
+			const panelTypeParam = AVAILABLE_EXPORT_PANEL_TYPES.includes(panelType)
+				? panelType
+				: PANEL_TYPES.TIME_SERIES;
+
+			const widgetId = v4();
+
+			const query = getExportQueryData(
+				exportDefaultQuery,
+				panelTypeParam,
+				options,
+			);
+
+			logEvent('Traces Explorer: Add to dashboard successful', {
+				panelType,
+				isNewDashboard,
+				dashboardName: dashboard?.title,
+			});
+
+			const dashboardEditView = getExportToDashboardLink({
+				query,
+				panelType: panelTypeParam,
+				dashboardId: dashboard.id,
+				widgetId,
+			});
+
+			if (dashboardEditView) {
+				safeNavigate(dashboardEditView);
+			}
+		},
+		[
+			exportDefaultQuery,
+			panelType,
+			safeNavigate,
+			options,
+			getExportToDashboardLink,
+		],
+	);
+
+	useShareBuilderUrl({ defaultValue: defaultQuery });
+
+	const logEventCalledRef = useRef(false);
+
+	useEffect(() => {
+		if (!logEventCalledRef.current) {
+			logEvent('Traces Explorer: Page visited', {});
+			logEventCalledRef.current = true;
+		}
+	}, []);
+
+	const isFilterApplied = useMemo(() => {
+		// if any of the non-disabled queries has filters applied, return true
+		const result = stagedQuery?.builder?.queryData?.filter(
+			(item) => !isEmpty(item.filters?.items) && !item.disabled,
+		);
+		return !!result?.length;
+	}, [stagedQuery]);
+
 	return (
-		<div className={styles.explorer} data-testid="llm-observability-explorer">
-			<div className={styles.placeholder}>Explorer coming soon.</div>
-		</div>
+		<Sentry.ErrorBoundary fallback={<ErrorBoundaryFallback />}>
+			<div
+				className="trace-explorer-page"
+				data-testid="llm-observability-explorer"
+			>
+				<Card className="filter" hidden={!isOpen}>
+					<QuickFilters
+						className="qf-traces-explorer"
+						source={QuickFiltersSource.TRACES_EXPLORER}
+						signal={SignalType.TRACES}
+						handleFilterVisibilityChange={(): void => {
+							setOpen(!isOpen);
+						}}
+					/>
+				</Card>
+				<div
+					className={cx('trace-explorer', {
+						'filters-expanded': isOpen,
+					})}
+				>
+					<div className="trace-explorer-header">
+						<Toolbar
+							showAutoRefresh
+							leftActions={
+								<LeftToolbarActions
+									showFilter={isOpen}
+									handleFilterVisibilityChange={(): void => setOpen(!isOpen)}
+									items={TOOLBAR_VIEWS}
+									selectedView={selectedView}
+									onChangeSelectedView={handleChangeSelectedView}
+								/>
+							}
+							warningElement={
+								!isEmpty(warning) ? <WarningPopover warningData={warning} /> : <div />
+							}
+							rightActions={
+								<RightToolbarActions
+									onStageRunQuery={(): void => {
+										setIsCancelled(false);
+										handleRunQuery();
+									}}
+									isLoadingQueries={isLoadingQueries}
+									handleCancelQuery={handleCancelQuery}
+								/>
+							}
+						/>
+					</div>
+					<ExplorerCard sourcepage={DataSource.TRACES}>
+						<div className="query-section-container">
+							<QuerySection />
+						</div>
+					</ExplorerCard>
+
+					<div className="traces-explorer-views">
+						{isCancelled && (
+							<QueryCancelledPlaceholder subText='Click "Run Query" to load traces.' />
+						)}
+
+						{!isCancelled && selectedView === ExplorerViews.LIST && (
+							<div className="trace-explorer-list-view">
+								<ListView
+									isFilterApplied={isFilterApplied}
+									setWarning={setWarning}
+									setIsLoadingQueries={setIsLoadingQueries}
+									queryKeyRef={listQueryKeyRef}
+								/>
+							</div>
+						)}
+
+						{!isCancelled && selectedView === ExplorerViews.TRACE && (
+							<div className="trace-explorer-traces-view">
+								<TracesView
+									isFilterApplied={isFilterApplied}
+									setWarning={setWarning}
+									setIsLoadingQueries={setIsLoadingQueries}
+									queryKeyRef={listQueryKeyRef}
+								/>
+							</div>
+						)}
+
+						{!isCancelled && selectedView === ExplorerViews.TIMESERIES && (
+							<div className="trace-explorer-time-series-view">
+								<TimeSeriesView
+									dataSource={DataSource.TRACES}
+									isFilterApplied={isFilterApplied}
+									setWarning={setWarning}
+									setIsLoadingQueries={setIsLoadingQueries}
+									queryKeyRef={listQueryKeyRef}
+								/>
+							</div>
+						)}
+
+						{!isCancelled && selectedView === ExplorerViews.TABLE && (
+							<div className="trace-explorer-table-view">
+								<TableView
+									setWarning={setWarning}
+									setIsLoadingQueries={setIsLoadingQueries}
+									queryKeyRef={listQueryKeyRef}
+								/>
+							</div>
+						)}
+					</div>
+
+					<ExplorerOptionWrapper
+						disabled={!stagedQuery}
+						query={exportDefaultQuery}
+						sourcepage={DataSource.TRACES}
+						onExport={handleExport}
+						handleChangeSelectedView={handleChangeSelectedView}
+					/>
+				</div>
+			</div>
+		</Sentry.ErrorBoundary>
 	);
 }
 

frontend/src/container/LLMObservability/Explorer/ListView/ListView.module.scss

@@ -0,0 +1,8 @@
+.container {
+	display: flex;
+	flex-direction: column;
+	flex: 1;
+	min-height: 0;
+	height: calc(100vh - 240px);
+	min-height: 400px;
+}

frontend/src/container/LLMObservability/Explorer/ListView/ListView.styles.scss

@@ -0,0 +1,34 @@
+.trace-explorer-controls {
+	display: flex;
+	justify-content: flex-end;
+	align-items: center;
+	gap: 8px;
+
+	.order-by-container {
+		display: flex;
+		align-items: center;
+		gap: 8px;
+
+		.order-by-label {
+			color: var(--muted-foreground);
+			font-size: 12px;
+			font-style: normal;
+			font-weight: 400;
+			line-height: 16px; /* 133.333% */
+
+			display: flex;
+			align-items: center;
+			gap: 4px;
+		}
+
+		.order-by-select {
+			width: 100px;
+
+			.ant-select-selector {
+				border: none;
+				box-shadow: none;
+				background-color: transparent;
+			}
+		}
+	}
+}

frontend/src/container/LLMObservability/Explorer/ListView/ListView.tsx

@@ -0,0 +1,272 @@
+import {
+	Dispatch,
+	memo,
+	MutableRefObject,
+	SetStateAction,
+	useCallback,
+	useEffect,
+	useMemo,
+	useState,
+} from 'react';
+import { QueryKey } from 'react-query';
+// eslint-disable-next-line no-restricted-imports
+import { useSelector } from 'react-redux';
+import logEvent from 'api/common/logEvent';
+import DownloadOptionsMenu from 'components/DownloadOptionsMenu/DownloadOptionsMenu';
+import ListViewOrderBy from 'components/OrderBy/ListViewOrderBy';
+import type { TableColumnDef } from 'components/TanStackTableView/types';
+import { ENTITY_VERSION_V5 } from 'constants/app';
+import { QueryParams } from 'constants/query';
+import { initialQueryAIWithType, PANEL_TYPES } from 'constants/queryBuilder';
+import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
+import { useOptionsMenu } from 'container/OptionsMenu';
+import { CustomTimeType } from 'container/TopNav/DateTimeSelectionV2/types';
+import TraceExplorerControls from 'container/TracesExplorer/Controls';
+import { getListViewQuery } from 'container/TracesExplorer/explorerUtils';
+import {
+	getTraceLink,
+	transformSpanRows,
+} from 'container/TracesExplorer/ListView/utils';
+import {
+	getFieldColumn,
+	TracesTableRow,
+} from 'container/TracesExplorer/TracesTable/getFieldColumn';
+import TracesTable from 'container/TracesExplorer/TracesTable/TracesTable';
+import { useGetQueryRange } from 'hooks/queryBuilder/useGetQueryRange';
+import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
+import { Pagination } from 'hooks/queryPagination';
+import { getDefaultPaginationConfig } from 'hooks/queryPagination/utils';
+import useUrlQueryData from 'hooks/useUrlQueryData';
+import { ArrowUp10, Minus } from '@signozhq/icons';
+import { AppState } from 'store/reducers';
+import { Warning } from 'types/api';
+import { DataSource } from 'types/common/queryBuilder';
+import { GlobalReducer } from 'types/reducer/globalTime';
+
+import {
+	defaultSelectedColumns,
+	PER_PAGE_OPTIONS,
+	TIMESTAMP_FIELD,
+} from './configs';
+import './ListView.styles.scss';
+
+import styles from './ListView.module.scss';
+
+interface ListViewProps {
+	isFilterApplied: boolean;
+	setWarning: Dispatch<SetStateAction<Warning | undefined>>;
+	setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
+	queryKeyRef?: MutableRefObject<QueryKey | undefined>;
+}
+
+function ListView({
+	isFilterApplied,
+	setWarning,
+	setIsLoadingQueries,
+	queryKeyRef,
+}: ListViewProps): JSX.Element {
+	const { stagedQuery, panelType: panelTypeFromQueryBuilder } =
+		useQueryBuilder();
+
+	const panelType = panelTypeFromQueryBuilder || PANEL_TYPES.LIST;
+
+	const [orderBy, setOrderBy] = useState<string>('timestamp:desc');
+
+	const {
+		selectedTime: globalSelectedTime,
+		maxTime,
+		minTime,
+		loading: timeRangeUpdateLoading,
+	} = useSelector<AppState, GlobalReducer>((state) => state.globalTime);
+
+	const { options, config } = useOptionsMenu({
+		dataSource: DataSource.TRACES,
+		aggregateOperator: 'count',
+		initialOptions: {
+			selectColumns: defaultSelectedColumns,
+		},
+	});
+
+	const { queryData: paginationQueryData } = useUrlQueryData<Pagination>(
+		QueryParams.pagination,
+	);
+	const paginationConfig =
+		paginationQueryData ?? getDefaultPaginationConfig(PER_PAGE_OPTIONS);
+
+	const requestQuery = useMemo(
+		() => getListViewQuery(stagedQuery || initialQueryAIWithType, orderBy),
+		[stagedQuery, orderBy],
+	);
+
+	// Stable sorted-name signature for the queryKey.
+	// - Drag updates selectColumns; raw queryKey would churn on reorder.
+	// - Trace API fetches only listed columns → add/remove must refetch.
+	// - Sorted-name signature: stable on reorder, changes on add/remove.
+	const selectColumnsSignature = useMemo(
+		() =>
+			(options?.selectColumns ?? [])
+				.map((c) => c.name)
+				.sort()
+				.join(','),
+		[options?.selectColumns],
+	);
+
+	const queryKey = useMemo(
+		() => [
+			REACT_QUERY_KEY.GET_QUERY_RANGE,
+			globalSelectedTime,
+			maxTime,
+			minTime,
+			stagedQuery,
+			panelType,
+			paginationConfig,
+			selectColumnsSignature,
+			orderBy,
+		],
+		[
+			stagedQuery,
+			panelType,
+			globalSelectedTime,
+			paginationConfig,
+			selectColumnsSignature,
+			maxTime,
+			minTime,
+			orderBy,
+		],
+	);
+
+	if (queryKeyRef) {
+		queryKeyRef.current = queryKey;
+	}
+
+	const { data, isFetching, isLoading, isError, error } = useGetQueryRange(
+		{
+			query: requestQuery,
+			graphType: panelType,
+			selectedTime: 'GLOBAL_TIME' as const,
+			globalSelectedInterval: globalSelectedTime as CustomTimeType,
+			params: {
+				dataSource: 'traces',
+			},
+			tableParams: {
+				pagination: paginationConfig,
+				selectColumns: options?.selectColumns,
+			},
+		},
+		ENTITY_VERSION_V5,
+		{
+			queryKey,
+			enabled:
+				// don't make api call while the time range state in redux is loading
+				!timeRangeUpdateLoading &&
+				!!stagedQuery &&
+				panelType === PANEL_TYPES.LIST &&
+				!!options?.selectColumns?.length,
+		},
+	);
+
+	useEffect(() => {
+		if (data?.payload) {
+			setWarning(data?.warning);
+		}
+		// eslint-disable-next-line react-hooks/exhaustive-deps
+	}, [data?.payload, data?.warning]);
+
+	useEffect(() => {
+		if (isLoading || isFetching) {
+			setIsLoadingQueries(true);
+		} else {
+			setIsLoadingQueries(false);
+		}
+	}, [isLoading, isFetching, setIsLoadingQueries]);
+
+	const queryTableDataResult = data?.payload?.data?.newResult?.data?.result;
+	const queryTableData = useMemo(
+		() => queryTableDataResult || [],
+		[queryTableDataResult],
+	);
+
+	const columns = useMemo<TableColumnDef<TracesTableRow>[]>(() => {
+		const fields = [
+			TIMESTAMP_FIELD,
+			...(options?.selectColumns ?? []).filter(
+				(field) => field.name !== TIMESTAMP_FIELD.name,
+			),
+		];
+		return fields.map((field) => getFieldColumn(field));
+	}, [options?.selectColumns]);
+
+	const rows = useMemo(
+		() => transformSpanRows(queryTableData),
+		[queryTableData],
+	);
+
+	const handleColumnOrderChange = useCallback(
+		(reordered: TableColumnDef<TracesTableRow>[]): void => {
+			config?.addColumn?.onReorder(reordered.map((column) => column.id));
+		},
+		[config],
+	);
+
+	const handleOrderChange = useCallback((value: string) => {
+		setOrderBy(value);
+	}, []);
+
+	useEffect(() => {
+		if (!isLoading && !isFetching && !isError && rows.length !== 0) {
+			void logEvent('Traces Explorer: Data present', {
+				panelType,
+			});
+		}
+	}, [isLoading, isFetching, isError, rows, panelType]);
+
+	return (
+		<div className={styles.container}>
+			<div className="trace-explorer-controls">
+				<div className="order-by-container">
+					<div className="order-by-label">
+						Order by <Minus size={14} /> <ArrowUp10 size={14} />
+					</div>
+
+					<ListViewOrderBy
+						value={orderBy}
+						onChange={handleOrderChange}
+						dataSource={DataSource.TRACES}
+					/>
+				</div>
+
+				<DownloadOptionsMenu
+					dataSource={DataSource.TRACES}
+					selectedColumns={options?.selectColumns}
+				/>
+
+				<TraceExplorerControls
+					isLoading={isFetching}
+					totalCount={rows.length}
+					config={config}
+					perPageOptions={PER_PAGE_OPTIONS}
+				/>
+			</div>
+
+			<TracesTable
+				data={rows}
+				columns={columns}
+				panelType="LIST"
+				getRowHref={getTraceLink}
+				isLoading={isLoading}
+				isFetching={isFetching}
+				isError={isError}
+				error={error}
+				isFilterApplied={isFilterApplied}
+				onColumnOrderChange={handleColumnOrderChange}
+				onColumnRemove={config?.addColumn?.onRemove}
+			/>
+		</div>
+	);
+}
+
+ListView.defaultProps = {
+	queryKeyRef: undefined,
+};
+
+export default memo(ListView);

frontend/src/container/LLMObservability/Explorer/ListView/configs.ts

@@ -0,0 +1,19 @@
+import type { TelemetryFieldKey } from 'api/v5/v5';
+import { DEFAULT_PER_PAGE_OPTIONS } from 'hooks/queryPagination';
+
+export const defaultSelectedColumns: string[] = [
+	'service.name',
+	'name',
+	'duration_nano',
+	'http_method',
+	'response_status_code',
+	'timestamp',
+];
+
+export const PER_PAGE_OPTIONS: number[] = [10, ...DEFAULT_PER_PAGE_OPTIONS];
+
+// Pinned timestamp column
+export const TIMESTAMP_FIELD = {
+	name: 'timestamp',
+	fieldContext: 'span',
+} as TelemetryFieldKey;

frontend/src/container/LLMObservability/Explorer/QuerySection/QuerySection.tsx

@@ -0,0 +1,40 @@
+import { memo, useMemo } from 'react';
+import { QueryBuilderV2 } from 'components/QueryBuilderV2/QueryBuilderV2';
+import { PANEL_TYPES } from 'constants/queryBuilder';
+import { QueryBuilderProps } from 'container/QueryBuilder/QueryBuilder.interfaces';
+import { useGetPanelTypesQueryParam } from 'hooks/queryBuilder/useGetPanelTypesQueryParam';
+import { DataSource } from 'types/common/queryBuilder';
+
+import { DEFAULT_PANEL_TYPE } from '../constants';
+
+function QuerySection(): JSX.Element {
+	const panelTypes = useGetPanelTypesQueryParam(DEFAULT_PANEL_TYPE);
+
+	// Only reaches the builder for timeseries/table; list/trace panels use QueryBuilderV2's listViewTracesFilterConfigs.
+	const filterConfigs: QueryBuilderProps['filterConfigs'] = useMemo(
+		() => ({
+			stepInterval: { isHidden: false, isDisabled: false },
+			limit: { isHidden: false, isDisabled: true },
+			having: { isHidden: false, isDisabled: true },
+		}),
+		[],
+	);
+
+	const isListViewPanel = useMemo(
+		() => panelTypes === PANEL_TYPES.LIST || panelTypes === PANEL_TYPES.TRACE,
+		[panelTypes],
+	);
+
+	return (
+		<QueryBuilderV2
+			isListViewPanel={isListViewPanel}
+			config={{ initialDataSource: DataSource.TRACES, queryVariant: 'static' }}
+			panelType={panelTypes}
+			filterConfigs={filterConfigs}
+			showOnlyWhereClause={isListViewPanel}
+			version="v3" // setting this to v3 as we this is rendered in logs explorer
+		/>
+	);
+}
+
+export default memo(QuerySection);

frontend/src/container/LLMObservability/Explorer/TableView/TableView.styles.scss

@@ -0,0 +1,7 @@
+.traces-table-view-header {
+	display: flex;
+	justify-content: flex-end;
+	align-items: center;
+	padding: 12px;
+	flex-shrink: 0;
+}

frontend/src/container/LLMObservability/Explorer/TableView/TableView.tsx

@@ -0,0 +1,130 @@
+import {
+	Dispatch,
+	memo,
+	MutableRefObject,
+	SetStateAction,
+	useEffect,
+	useMemo,
+} from 'react';
+// eslint-disable-next-line no-restricted-imports
+import { useSelector } from 'react-redux';
+import { Space } from 'antd';
+import ErrorInPlace from 'components/ErrorInPlace/ErrorInPlace';
+import ExportMenu from 'components/ExportMenu/ExportMenu';
+import { ENTITY_VERSION_V5 } from 'constants/app';
+import { initialQueriesMap, PANEL_TYPES } from 'constants/queryBuilder';
+import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
+import { QueryTable } from 'container/QueryTable';
+import { useGetQueryRange } from 'hooks/queryBuilder/useGetQueryRange';
+import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
+import { AppState } from 'store/reducers';
+import { Warning } from 'types/api';
+import APIError from 'types/api/error';
+import { QueryDataV3 } from 'types/api/widgets/getQuery';
+import { DataSource } from 'types/common/queryBuilder';
+import { GlobalReducer } from 'types/reducer/globalTime';
+
+import './TableView.styles.scss';
+
+function TableView({
+	setWarning,
+	setIsLoadingQueries,
+	queryKeyRef,
+}: {
+	setWarning: Dispatch<SetStateAction<Warning | undefined>>;
+	setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
+	queryKeyRef?: MutableRefObject<any>;
+}): JSX.Element {
+	const { stagedQuery, panelType } = useQueryBuilder();
+
+	const {
+		selectedTime: globalSelectedTime,
+		maxTime,
+		minTime,
+	} = useSelector<AppState, GlobalReducer>((state) => state.globalTime);
+
+	const queryKey = useMemo(
+		() => [
+			REACT_QUERY_KEY.GET_QUERY_RANGE,
+			globalSelectedTime,
+			maxTime,
+			minTime,
+			stagedQuery,
+		],
+		[globalSelectedTime, maxTime, minTime, stagedQuery],
+	);
+
+	if (queryKeyRef) {
+		queryKeyRef.current = queryKey;
+	}
+
+	const { data, isLoading, isFetching, isError, error } = useGetQueryRange(
+		{
+			query: stagedQuery || initialQueriesMap.traces,
+			graphType: panelType || PANEL_TYPES.TABLE,
+			selectedTime: 'GLOBAL_TIME',
+			globalSelectedInterval: globalSelectedTime,
+			params: {
+				dataSource: 'traces',
+			},
+		},
+		ENTITY_VERSION_V5,
+		{
+			queryKey,
+			enabled: !!stagedQuery && panelType === PANEL_TYPES.TABLE,
+		},
+	);
+
+	useEffect(() => {
+		if (isLoading || isFetching) {
+			setIsLoadingQueries(true);
+		} else {
+			setIsLoadingQueries(false);
+		}
+	}, [isLoading, isFetching, setIsLoadingQueries]);
+
+	const queryTableData = useMemo(
+		() =>
+			data?.payload?.data?.newResult?.data?.result ||
+			data?.payload.data.result ||
+			[],
+		[data],
+	);
+
+	useEffect(() => {
+		if (data?.payload) {
+			setWarning(data.warning);
+		}
+		// eslint-disable-next-line react-hooks/exhaustive-deps
+	}, [data?.payload, data?.warning]);
+
+	return (
+		<Space.Compact block direction="vertical">
+			{isError && error && <ErrorInPlace error={error as APIError} />}
+			{!isError && data && (
+				<div className="traces-table-view-header">
+					<ExportMenu
+						dataSource={DataSource.TRACES}
+						data={data}
+						query={stagedQuery || initialQueriesMap.traces}
+						fileName="traces-table"
+					/>
+				</div>
+			)}
+			{!isError && (
+				<QueryTable
+					query={stagedQuery || initialQueriesMap.traces}
+					queryTableData={queryTableData as QueryDataV3[]}
+					loading={isLoading}
+					sticky
+				/>
+			)}
+		</Space.Compact>
+	);
+}
+
+TableView.defaultProps = {
+	queryKeyRef: undefined,
+};
+
+export default memo(TableView);

frontend/src/container/LLMObservability/Explorer/TimeSeriesView/TimeSeriesView.styles.scss

@@ -0,0 +1,8 @@
+.trace-explorer-time-series-view-container {
+	&-header {
+		display: flex;
+		justify-content: flex-start;
+		align-items: center;
+		padding: 12px;
+	}
+}

frontend/src/container/LLMObservability/Explorer/TimeSeriesView/TimeSeriesView.tsx

@@ -0,0 +1,147 @@
+import {
+	Dispatch,
+	MutableRefObject,
+	SetStateAction,
+	useEffect,
+	useMemo,
+} from 'react';
+// eslint-disable-next-line no-restricted-imports
+import { useSelector } from 'react-redux';
+import { ENTITY_VERSION_V5 } from 'constants/app';
+import { initialQueriesMap, PANEL_TYPES } from 'constants/queryBuilder';
+import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
+import TimeSeriesView from 'container/TimeSeriesView/TimeSeriesView';
+import { convertDataValueToMs } from 'container/TimeSeriesView/utils';
+import { useGetQueryRange } from 'hooks/queryBuilder/useGetQueryRange';
+import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
+import useUrlYAxisUnit from 'hooks/useUrlYAxisUnit';
+import { AppState } from 'store/reducers';
+import { Warning } from 'types/api';
+import APIError from 'types/api/error';
+import { DataSource } from 'types/common/queryBuilder';
+import { GlobalReducer } from 'types/reducer/globalTime';
+
+import './TimeSeriesView.styles.scss';
+
+function TimeSeriesViewContainer({
+	dataSource = DataSource.TRACES,
+	isFilterApplied,
+	setWarning,
+	setIsLoadingQueries,
+	queryKeyRef,
+}: TimeSeriesViewProps): JSX.Element {
+	const { stagedQuery, currentQuery, panelType } = useQueryBuilder();
+
+	const isValidToConvertToMs = useMemo(() => {
+		const isValid: boolean[] = [];
+
+		currentQuery.builder.queryData.forEach(
+			({ aggregateAttribute, aggregateOperator }) => {
+				const isExistDurationNanoAttribute =
+					aggregateAttribute?.key === 'durationNano' ||
+					aggregateAttribute?.key === 'duration_nano';
+
+				const isCountOperator =
+					aggregateOperator === 'count' || aggregateOperator === 'count_distinct';
+
+				isValid.push(!isCountOperator && isExistDurationNanoAttribute);
+			},
+		);
+
+		return isValid.every(Boolean);
+	}, [currentQuery]);
+
+	const defaultUnit = isValidToConvertToMs ? 'ms' : 'short';
+	const { yAxisUnit, onUnitChange } = useUrlYAxisUnit(defaultUnit);
+
+	const {
+		selectedTime: globalSelectedTime,
+		maxTime,
+		minTime,
+	} = useSelector<AppState, GlobalReducer>((state) => state.globalTime);
+
+	const queryKey = useMemo(
+		() => [
+			REACT_QUERY_KEY.GET_QUERY_RANGE,
+			globalSelectedTime,
+			maxTime,
+			minTime,
+			stagedQuery,
+		],
+		[globalSelectedTime, maxTime, minTime, stagedQuery],
+	);
+
+	if (queryKeyRef) {
+		queryKeyRef.current = queryKey;
+	}
+
+	const { data, isLoading, isFetching, isError, error } = useGetQueryRange(
+		{
+			query: stagedQuery || initialQueriesMap[dataSource],
+			graphType: panelType || PANEL_TYPES.TIME_SERIES,
+			selectedTime: 'GLOBAL_TIME',
+			globalSelectedInterval: globalSelectedTime,
+			params: {
+				dataSource,
+			},
+		},
+		// ENTITY_VERSION_V4,
+		ENTITY_VERSION_V5,
+		{
+			queryKey,
+			enabled: !!stagedQuery && panelType === PANEL_TYPES.TIME_SERIES,
+		},
+	);
+
+	useEffect(() => {
+		if (data?.payload) {
+			setWarning(data?.warning);
+		}
+		// eslint-disable-next-line react-hooks/exhaustive-deps
+	}, [data?.payload, data?.warning]);
+
+	const responseData = useMemo(
+		() => (isValidToConvertToMs ? convertDataValueToMs(data) : data),
+		[data, isValidToConvertToMs],
+	);
+
+	useEffect(() => {
+		if (isLoading || isFetching) {
+			setIsLoadingQueries(true);
+		} else {
+			setIsLoadingQueries(false);
+		}
+	}, [isLoading, isFetching, setIsLoadingQueries]);
+
+	return (
+		<div className="trace-explorer-time-series-view-container">
+			<TimeSeriesView
+				isFilterApplied={isFilterApplied}
+				isError={isError}
+				error={error as APIError}
+				isLoading={isLoading || isFetching}
+				data={responseData}
+				yAxisUnit={yAxisUnit}
+				onYAxisUnitChange={onUnitChange}
+				dataSource={dataSource}
+				setWarning={setWarning}
+				allowExport
+			/>
+		</div>
+	);
+}
+
+interface TimeSeriesViewProps {
+	dataSource?: DataSource;
+	isFilterApplied: boolean;
+	setWarning: Dispatch<SetStateAction<Warning | undefined>>;
+	setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
+	queryKeyRef?: MutableRefObject<any>;
+}
+
+TimeSeriesViewContainer.defaultProps = {
+	dataSource: DataSource.TRACES,
+	queryKeyRef: undefined,
+};
+
+export default TimeSeriesViewContainer;

frontend/src/container/LLMObservability/Explorer/TracesView/TracesView.module.scss

@@ -0,0 +1,15 @@
+.container {
+	display: flex;
+	flex-direction: column;
+	flex: 1;
+	min-height: 0;
+	// Page chain isn't a flex column, so anchor the virtualized table against the viewport.
+	height: calc(100vh - 240px);
+	min-height: 400px;
+}
+
+.actionsContainer {
+	display: flex;
+	justify-content: space-between;
+	align-items: center;
+}

frontend/src/container/LLMObservability/Explorer/TracesView/TracesView.tsx

@@ -0,0 +1,190 @@
+import {
+	Dispatch,
+	memo,
+	MutableRefObject,
+	SetStateAction,
+	useEffect,
+	useMemo,
+} from 'react';
+import { QueryKey } from 'react-query';
+// eslint-disable-next-line no-restricted-imports
+import { useSelector } from 'react-redux';
+import { Typography } from '@signozhq/ui/typography';
+import logEvent from 'api/common/logEvent';
+import DownloadOptionsMenu from 'components/DownloadOptionsMenu/DownloadOptionsMenu';
+import { ENTITY_VERSION_V5 } from 'constants/app';
+import { QueryParams } from 'constants/query';
+import { initialQueryAIWithType, PANEL_TYPES } from 'constants/queryBuilder';
+import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
+import TraceExplorerControls from 'container/TracesExplorer/Controls';
+import { getListViewQuery } from 'container/TracesExplorer/explorerUtils';
+import { getTraceLink } from 'container/TracesExplorer/ListView/utils';
+import { TracesTableRow } from 'container/TracesExplorer/TracesTable/getFieldColumn';
+import TracesTable from 'container/TracesExplorer/TracesTable/TracesTable';
+import { useGetQueryRange } from 'hooks/queryBuilder/useGetQueryRange';
+import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
+import { Pagination } from 'hooks/queryPagination';
+import useUrlQueryData from 'hooks/useUrlQueryData';
+import { AppState } from 'store/reducers';
+import { Warning } from 'types/api';
+import { DataSource } from 'types/common/queryBuilder';
+import { GlobalReducer } from 'types/reducer/globalTime';
+import DOCLINKS from 'utils/docLinks';
+
+import { columns, PER_PAGE_OPTIONS } from './configs';
+import styles from './TracesView.module.scss';
+
+interface TracesViewProps {
+	isFilterApplied: boolean;
+	setWarning: Dispatch<SetStateAction<Warning | undefined>>;
+	setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
+	queryKeyRef?: MutableRefObject<QueryKey | undefined>;
+}
+
+function TracesView({
+	isFilterApplied,
+	setWarning,
+	setIsLoadingQueries,
+	queryKeyRef,
+}: TracesViewProps): JSX.Element {
+	const { stagedQuery, panelType } = useQueryBuilder();
+
+	const {
+		selectedTime: globalSelectedTime,
+		maxTime,
+		minTime,
+	} = useSelector<AppState, GlobalReducer>((state) => state.globalTime);
+
+	const { queryData: paginationQueryData } = useUrlQueryData<Pagination>(
+		QueryParams.pagination,
+	);
+
+	const transformedQuery = useMemo(
+		() => getListViewQuery(stagedQuery || initialQueryAIWithType),
+		[stagedQuery],
+	);
+
+	const queryKey = useMemo(
+		() => [
+			REACT_QUERY_KEY.GET_QUERY_RANGE,
+			globalSelectedTime,
+			maxTime,
+			minTime,
+			stagedQuery,
+			panelType,
+			paginationQueryData,
+		],
+		[
+			globalSelectedTime,
+			maxTime,
+			minTime,
+			stagedQuery,
+			panelType,
+			paginationQueryData,
+		],
+	);
+
+	if (queryKeyRef) {
+		queryKeyRef.current = queryKey;
+	}
+
+	const { data, isLoading, isFetching, isError, error } = useGetQueryRange(
+		{
+			query: transformedQuery,
+			graphType: panelType || PANEL_TYPES.TRACE,
+			selectedTime: 'GLOBAL_TIME',
+			globalSelectedInterval: globalSelectedTime,
+			params: {
+				dataSource: 'traces',
+			},
+			tableParams: {
+				pagination: paginationQueryData,
+			},
+		},
+		ENTITY_VERSION_V5,
+		{
+			queryKey,
+			enabled: !!stagedQuery && panelType === PANEL_TYPES.TRACE,
+		},
+	);
+
+	useEffect(() => {
+		if (data?.payload) {
+			setWarning(data?.warning);
+		}
+		// eslint-disable-next-line react-hooks/exhaustive-deps
+	}, [data?.payload, data?.warning]);
+
+	const responseData = data?.payload?.data?.newResult?.data?.result[0]?.list;
+
+	const rows = useMemo<TracesTableRow[]>(
+		() =>
+			(responseData ?? []).map((item) => {
+				const row = item.data;
+				return { ...row, id: row.trace_id };
+			}) as TracesTableRow[],
+		[responseData],
+	);
+
+	useEffect(() => {
+		if (isLoading || isFetching) {
+			setIsLoadingQueries(true);
+		} else {
+			setIsLoadingQueries(false);
+		}
+	}, [isLoading, isFetching, setIsLoadingQueries]);
+
+	useEffect(() => {
+		if (!isLoading && !isFetching && !isError && rows.length !== 0) {
+			void logEvent('Traces Explorer: Data present', {
+				panelType: 'TRACE',
+			});
+		}
+	}, [isLoading, isFetching, isError, rows.length]);
+
+	return (
+		<div className={styles.container}>
+			<div className={styles.actionsContainer}>
+				<Typography>
+					This tab only shows Root Spans. More details
+					<Typography.Link href={DOCLINKS.TRACES_DETAILS_LINK} target="_blank">
+						{' '}
+						here
+					</Typography.Link>
+				</Typography>
+
+				<div className="trace-explorer-controls">
+					<DownloadOptionsMenu
+						dataSource={DataSource.TRACES}
+						panelType={PANEL_TYPES.TRACE}
+					/>
+
+					<TraceExplorerControls
+						isLoading={isLoading}
+						totalCount={rows.length}
+						perPageOptions={PER_PAGE_OPTIONS}
+					/>
+				</div>
+			</div>
+
+			<TracesTable
+				data={rows}
+				columns={columns}
+				respectColumnOrder
+				panelType="TRACE"
+				getRowHref={getTraceLink}
+				isLoading={isLoading}
+				isFetching={isFetching}
+				isError={isError}
+				error={error}
+				isFilterApplied={isFilterApplied}
+			/>
+		</div>
+	);
+}
+
+TracesView.defaultProps = {
+	queryKeyRef: undefined,
+};
+
+export default memo(TracesView);

frontend/src/container/LLMObservability/Explorer/TracesView/configs.ts

@@ -0,0 +1,25 @@
+import { TelemetryFieldKey } from 'api/v5/v5';
+import type { TableColumnDef } from 'components/TanStackTableView/types';
+import {
+	getFieldColumn,
+	TracesTableRow,
+} from 'container/TracesExplorer/TracesTable/getFieldColumn';
+import { DEFAULT_PER_PAGE_OPTIONS } from 'hooks/queryPagination';
+
+export const PER_PAGE_OPTIONS: number[] = [10, ...DEFAULT_PER_PAGE_OPTIONS];
+
+const TRACE_FIELDS = [
+	{ name: 'service.name', fieldContext: 'resource' },
+	{ name: 'name' },
+	{ name: 'duration_nano' },
+	{ name: 'span_count' },
+	{ name: 'trace_id' },
+] as TelemetryFieldKey[];
+
+export const columns: TableColumnDef<TracesTableRow>[] = TRACE_FIELDS.map(
+	(field) => ({
+		...getFieldColumn(field),
+		enableRemove: false,
+		canBeHidden: false,
+	}),
+);

frontend/src/container/LLMObservability/Explorer/constants.ts

@@ -0,0 +1,40 @@
+import { PANEL_TYPES } from 'constants/queryBuilder';
+
+export const DEFAULT_PANEL_TYPE = PANEL_TYPES.TRACE;
+
+export const TOOLBAR_VIEWS = {
+	list: {
+		name: 'list',
+		label: 'List',
+		show: true,
+		key: 'list',
+	},
+	timeseries: {
+		name: 'timeseries',
+		label: 'Timeseries',
+		disabled: false,
+		show: true,
+		key: 'timeseries',
+	},
+	trace: {
+		name: 'trace',
+		label: 'Trace',
+		disabled: false,
+		show: true,
+		key: 'trace',
+	},
+	table: {
+		name: 'table',
+		label: 'Table',
+		disabled: false,
+		show: true,
+		key: 'table',
+	},
+	clickhouse: {
+		name: 'clickhouse',
+		label: 'Clickhouse',
+		disabled: false,
+		show: false,
+		key: 'clickhouse',
+	},
+};

frontend/src/container/LLMObservability/__tests__/LLMObservability.test.tsx

@@ -18,6 +18,12 @@ jest.mock('pages/DashboardPageV2/DashboardContainer', () => ({
 	default: (): JSX.Element => <div data-testid="llm-overview-dashboard" />,
 }));
 
+// Same data-router gap as the dashboard above: the Explorer toolbar calls useNavigationType.
+jest.mock('container/LLMObservability/Explorer/Explorer', () => ({
+	__esModule: true,
+	default: (): JSX.Element => <div data-testid="llm-observability-explorer" />,
+}));
+
 function setupList(items = mockRules): void {
 	server.use(
 		rest.get(LLM_PRICING_ENDPOINT, (_req, res, ctx) =>

frontend/src/providers/QueryBuilder.tsx

@@ -470,6 +470,7 @@ export function QueryBuilderProvider({
 			const newQuery: IBuilderQuery = {
 				...initialBuilderQuery,
 				source: queries?.[0]?.source || '',
+				builderQueryType: queries?.[0]?.builderQueryType,
 				queryName: createNewBuilderItemName({ existNames, sourceNames: alphabet }),
 				expression: createNewBuilderItemName({
 					existNames,

frontend/src/providers/__tests__/QueryBuilder.createNewBuilderQuery.test.tsx

@@ -0,0 +1,55 @@
+import {
+	initialQueriesMap,
+	initialQueryAIWithType,
+} from 'constants/queryBuilder';
+import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
+import { act, AllTheProviders, renderHook } from 'tests/test-utils';
+import { Query } from 'types/api/queryBuilder/queryBuilderData';
+
+const renderQueryBuilder = (
+	initialQuery: Query,
+): ReturnType<
+	typeof renderHook<ReturnType<typeof useQueryBuilder>, unknown>
+> => {
+	const hook = renderHook(() => useQueryBuilder(), {
+		wrapper: AllTheProviders,
+	});
+
+	act(() => {
+		hook.result.current.initQueryBuilderData(initialQuery);
+	});
+
+	return hook;
+};
+
+describe('createNewBuilderQuery builderQueryType propagation', () => {
+	it('carries builderQueryType from the first query onto an added query', () => {
+		const { result } = renderQueryBuilder(initialQueryAIWithType);
+
+		expect(
+			result.current.currentQuery.builder.queryData[0].builderQueryType,
+		).toBe('builder_ai_query');
+
+		act(() => {
+			result.current.addNewBuilderQuery();
+		});
+
+		expect(result.current.currentQuery.builder.queryData).toHaveLength(2);
+		expect(
+			result.current.currentQuery.builder.queryData[1].builderQueryType,
+		).toBe('builder_ai_query');
+	});
+
+	it('leaves builderQueryType unset when the first query has none', () => {
+		const { result } = renderQueryBuilder(initialQueriesMap.traces);
+
+		act(() => {
+			result.current.addNewBuilderQuery();
+		});
+
+		expect(result.current.currentQuery.builder.queryData).toHaveLength(2);
+		expect(
+			result.current.currentQuery.builder.queryData[1].builderQueryType,
+		).toBeUndefined();
+	});
+});

frontend/src/types/api/queryBuilder/queryBuilderData.ts

@@ -8,6 +8,7 @@ import {
 } from 'types/common/queryBuilder';
 
 import {
+	BuilderQueryType,
 	Filter,
 	Having as HavingV5,
 	LogAggregation,
@@ -90,6 +91,8 @@ export type IBuilderQuery = {
 	offset?: number;
 	selectColumns?: BaseAutocompleteData[] | TelemetryFieldKey[];
 	source?: 'meter' | '';
+	/** Envelope type this query serializes to in the v5 payload; absent means builder_query. */
+	builderQueryType?: BuilderQueryType;
 };
 
 export interface IClickHouseQuery {

frontend/src/types/api/v5/queryRange.ts

@@ -16,6 +16,7 @@ export type RequestType =
 
 export type QueryType =
 	| 'builder_query'
+	| 'builder_ai_query'
 	| 'builder_trace_operator'
 	| 'builder_formula'
 	| 'builder_sub_query'
@@ -23,6 +24,12 @@ export type QueryType =
 	| 'clickhouse_sql'
 	| 'promql';
 
+/** Envelope types carrying a BuilderQuery spec; builder_ai_query routes to the gen_ai-scoped trace builder. */
+export type BuilderQueryType = Extract<
+	QueryType,
+	'builder_query' | 'builder_ai_query'
+>;
+
 export type OrderDirection = 'asc' | 'desc';
 
 export type JoinType = 'inner' | 'left' | 'right' | 'full' | 'cross';

pkg/querier/builder_query.go

@@ -82,6 +82,12 @@ func (q *builderQuery[T]) Fingerprint() string {
 		return ""
 	}
 
+	// AI trace aggregations qualify and rank traces on whole-window per-trace
+	// values, which do not decompose into cacheable time buckets.
+	if q.queryType == qbtypes.QueryTypeBuilderAI {
+		return ""
+	}
+
 	// Create a deterministic fingerprint for builder queries
 	// This needs to include all fields that affect the query results
 	parts := []string{q.queryType.StringValue()}

pkg/querier/builder_query_test.go

@@ -117,8 +117,7 @@ func TestBuilderQueryFingerprintQueryType(t *testing.T) {
 	}
 
 	assert.True(t, strings.HasPrefix(regular.Fingerprint(), qbtypes.QueryTypeBuilder.StringValue()+"&"))
-	assert.True(t, strings.HasPrefix(ai.Fingerprint(), qbtypes.QueryTypeBuilderAI.StringValue()+"&"))
-	assert.NotEqual(t, regular.Fingerprint(), ai.Fingerprint())
+	assert.Empty(t, ai.Fingerprint())
 }
 
 func TestMakeBucketsOrder(t *testing.T) {

pkg/querier/querier.go

@@ -470,7 +470,7 @@ func (q *querier) resolveMetricMetadata(ctx context.Context, orgID valuer.UUID,
 				continue
 			}
 			// Type is resolved now; validate aggregation compatibility against it.
-			if err := spec.Aggregations[i].ValidateForType(); err != nil {
+			if err := spec.Aggregations[i].ValidateForTypeAndTemporality(); err != nil {
 				return nil, nil, err
 			}
 			if reducedMetricsSet[spec.Aggregations[i].MetricName] {

pkg/statementbuilder/aistatementbuilder/statement_builder_test.go

@@ -994,8 +994,8 @@ func TestBuild_TraceList_MultiVariantGateKey(t *testing.T) {
 	assert.Contains(t, got, "mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_number, 'gen_ai.tool.name')")
 }
 
-// `trace.` marks a trace-level aggregate; `tracefield.` routes trace-level too but is
-// not a rewritable alias, so the HAVING rewriter rejects it.
+// A `trace.`-prefixed aggregate in the filter box and the same condition in the
+// explicit Having box build the same query; output-only aggregates are rejected.
 func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
 	b := newTestBuilder(t)
 	build := func(q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) (*qbtypes.Statement, error) {
@@ -1003,19 +1003,14 @@ func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
 		return b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTrace, q, nil)
 	}
 
-	_, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+	viaTrace, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
 		Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"}})
 	require.NoError(t, err)
 
-	_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
-		Filter: &qbtypes.Filter{Expression: "tracefield.output_tokens > 1000"}})
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "Invalid references in `Having` expression: [tracefield.output_tokens]")
-
-	_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
-		Having: &qbtypes.Having{Expression: "tracefield.output_tokens > 1000"}})
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "Invalid references in `Having` expression: [tracefield.output_tokens]")
+	viaHaving, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+		Having: &qbtypes.Having{Expression: "trace.output_tokens > 1000"}})
+	require.NoError(t, err)
+	assert.Equal(t, viaTrace.Query, viaHaving.Query)
 
 	_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
 		Filter: &qbtypes.Filter{Expression: "trace.span_count > 3"}})
@@ -1023,7 +1018,8 @@ func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
 	assert.Contains(t, err.Error(), "cannot be used")
 }
 
-// Query variables in a trace-level condition are substituted into the HAVING.
+// Query variables in a trace-level condition resolve like span filters: bound args,
+// list/IN handling, dynamic __all__ dropping the condition.
 func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
 	b := newTestBuilder(t)
 	build := func(expr string, vars map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
@@ -1035,17 +1031,18 @@ func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
 			}, vars)
 	}
 
-	// scalar variable -> literal in HAVING
+	// scalar variable -> bound arg via the filter pipeline
 	stmt, err := build("trace.output_tokens > $threshold",
 		map[string]qbtypes.VariableItem{"threshold": {Value: 700}})
 	require.NoError(t, err)
-	assert.Contains(t, stmt.Query, "HAVING output_tokens > 700")
+	assert.Contains(t, stmt.Query, "HAVING output_tokens > ?")
+	assert.Contains(t, stmt.Args, float64(700))
 
 	// list variable with IN
 	stmt, err = build("trace.llm_call_count IN $counts",
 		map[string]qbtypes.VariableItem{"counts": {Value: []any{1, 2}}})
 	require.NoError(t, err)
-	assert.Contains(t, stmt.Query, "HAVING llm_call_count IN")
+	assert.Contains(t, stmt.Query, "HAVING llm_call_count IN (?, ?)")
 
 	// dynamic __all__ -> condition dropped, no HAVING at all
 	stmt, err = build("trace.output_tokens > $threshold",
@@ -1053,7 +1050,7 @@ func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
 	require.NoError(t, err)
 	assert.NotContains(t, stmt.Query, "HAVING")
 
-	// unresolved variable -> rejected, not compared as a literal
+	// unresolved variable -> rejected, though only as an unknown aggregate today
 	_, err = build("trace.output_tokens > $missing", map[string]qbtypes.VariableItem{"other": {Value: 1}})
 	require.Error(t, err)
 }

pkg/statementbuilder/aistatementbuilder/trace_aggregation_test.go

@@ -0,0 +1,753 @@
+package aistatementbuilder
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The builder assumes at least one aggregation; request validation is what enforces it.
+func TestBuild_Aggregation_NoAggregations_RejectedByRequestValidation(t *testing.T) {
+	for _, rt := range []qbtypes.RequestType{qbtypes.RequestTypeScalar, qbtypes.RequestTypeTimeSeries} {
+		req := qbtypes.QueryRangeRequest{
+			Start:       testStartMs,
+			End:         testEndMs,
+			RequestType: rt,
+			CompositeQuery: qbtypes.CompositeQuery{
+				Queries: []qbtypes.QueryEnvelope{{
+					Type: qbtypes.QueryTypeBuilderAI,
+					Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+						Name:         "A",
+						Signal:       telemetrytypes.SignalTraces,
+						StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+					},
+				}},
+			},
+		}
+		require.ErrorContains(t, req.Validate(), "at least one aggregation is required", rt.StringValue())
+	}
+}
+
+// Traces without token spans yield NULL, which the outer avg skips.
+func TestBuild_FullSQL_Scalar_TraceAgg(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __scoped_traces AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id
+)
+SELECT avg(output_tokens) AS __result_0
+FROM __scoped_traces
+ORDER BY __result_0 DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// A span-level filter is ANDed into the per-trace scan's WHERE, next to the gate mask.
+func TestBuild_FullSQL_Scalar_SpanFilter(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+			Filter:       &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini'"},
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __scoped_traces AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model'))
+    GROUP BY trace_id
+)
+SELECT avg(output_tokens) AS __result_0
+FROM __scoped_traces
+ORDER BY __result_0 DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// A trace-level filter qualifies first: __qualified holds the trace ids whose
+// whole-window value passes, and the per-trace scan is constrained to them.
+func TestBuild_FullSQL_Scalar_TraceFilter(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+			Filter:       &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __qualified AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id
+    HAVING output_tokens > 1000
+),
+__scoped_traces AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
+    GROUP BY trace_id
+)
+SELECT avg(output_tokens) AS __result_0
+FROM __scoped_traces
+ORDER BY __result_0 DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Grouping by an intrinsic: the positional alias keeps `toString(name) AS name` (a cyclic
+// alias) from forming, and an order key on the dimension resolves to that alias.
+func TestBuild_FullSQL_Scalar_GroupByIntrinsic(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+			GroupBy:      []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "name"}}},
+			Order:        []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "name"}}, Direction: qbtypes.OrderDirectionAsc}},
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __scoped_traces AS (
+    SELECT trace_id,
+        toString(multiIf(name <> '', toString(name), NULL)) AS __GROUP_BY_KEY_0_name,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id, __GROUP_BY_KEY_0_name
+)
+SELECT __GROUP_BY_KEY_0_name, avg(output_tokens) AS __result_0
+FROM __scoped_traces
+GROUP BY __GROUP_BY_KEY_0_name
+ORDER BY __GROUP_BY_KEY_0_name asc
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Every dimension at once; the HAVING on the alias is rewritten to __result_0.
+func TestBuild_FullSQL_Scalar_FullCombo(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{
+				{Expression: "avg(trace.output_tokens)", Alias: "avg_out"},
+				{Expression: "count(trace.trace_id)"},
+			},
+			Filter:  &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND trace.total_tokens > 100"},
+			GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}}},
+			Having:  &qbtypes.Having{Expression: "avg_out > 50"},
+			Order:   []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "avg_out"}}, Direction: qbtypes.OrderDirectionDesc}},
+			Limit:   5,
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __qualified AS (
+    SELECT trace_id,
+        coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens'), toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id
+    HAVING total_tokens > 100
+),
+__scoped_traces AS (
+    SELECT trace_id,
+        toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model'))
+      AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
+    GROUP BY trace_id, __GROUP_BY_KEY_0_gen_ai.request.model
+)
+SELECT __GROUP_BY_KEY_0_gen_ai.request.model, avg(output_tokens) AS __result_0, count(trace_id) AS __result_1
+FROM __scoped_traces
+GROUP BY __GROUP_BY_KEY_0_gen_ai.request.model
+HAVING __result_0 > 50
+ORDER BY __result_0 desc
+LIMIT 5
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Time series: the per-trace scan buckets by span time, the outer aggregation per bucket.
+func TestBuild_FullSQL_TimeSeries_TraceAgg(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __scoped_traces AS (
+    SELECT trace_id,
+        toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id, ts
+)
+SELECT ts, avg(output_tokens) AS __result_0
+FROM __scoped_traces
+GROUP BY ts
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// A grouped, limited time series ranks groups on unbucketed whole-window values
+// (__scoped_traces_total), so a non-composable aggregate like avg ranks exactly.
+func TestBuild_FullSQL_TimeSeries_GroupLimit(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(trace.output_tokens)", Alias: "total_out"}},
+			GroupBy:      []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}}},
+			Having:       &qbtypes.Having{Expression: "total_out > 500"},
+			Order:        []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "total_out"}}, Direction: qbtypes.OrderDirectionDesc}},
+			Limit:        3,
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __scoped_traces_total AS (
+    SELECT trace_id,
+        toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id, __GROUP_BY_KEY_0_gen_ai.request.model
+),
+__limit_cte AS (
+    SELECT __GROUP_BY_KEY_0_gen_ai.request.model, sum(output_tokens) AS __result_0
+    FROM __scoped_traces_total
+    GROUP BY __GROUP_BY_KEY_0_gen_ai.request.model
+    ORDER BY __result_0 desc
+    LIMIT 3
+),
+__scoped_traces AS (
+    SELECT trace_id,
+        toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
+        toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND (toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL))) GLOBAL IN (SELECT __GROUP_BY_KEY_0_gen_ai.request.model FROM __limit_cte)
+    GROUP BY trace_id, ts, __GROUP_BY_KEY_0_gen_ai.request.model
+)
+SELECT ts, __GROUP_BY_KEY_0_gen_ai.request.model, sum(output_tokens) AS __result_0
+FROM __scoped_traces
+GROUP BY ts, __GROUP_BY_KEY_0_gen_ai.request.model
+HAVING __result_0 > 500
+ORDER BY ts desc
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// A span-level scalar delegates to the trace builder, constrained by __trace_scope;
+// the shape is the delegate's own, hence no SETTINGS suffix.
+func TestBuild_FullSQL_Scalar_SpanAgg_TraceScoped(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(gen_ai.usage.output_tokens)"}},
+			Filter:       &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __trace_scope AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id
+    HAVING output_tokens > 1000
+)
+SELECT sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS __result_0
+FROM signoz_traces.distributed_signoz_index_v3
+WHERE trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)
+  AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+  AND timestamp >= '1747947419000000000'
+  AND timestamp < '1747983448000000000'
+  AND ts_bucket_start >= 1747945619
+  AND ts_bucket_start <= 1747983448
+ORDER BY __result_0 DESC
+`, stmt)
+}
+
+// Two group keys make the top-N prune a 2-tuple GLOBAL IN, and the qualification plus
+// span predicate apply to the ranking scan and the main scan alike.
+func TestBuild_FullSQL_TimeSeries_GroupLimit_MultiKey(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+			Aggregations: []qbtypes.TraceAggregation{
+				{Expression: "sum(trace.output_tokens)"},
+				{Expression: "count(trace.trace_id)"},
+			},
+			Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND trace.total_tokens > 100"},
+			GroupBy: []qbtypes.GroupByKey{
+				{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}},
+				{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.user.id"}},
+			},
+			Limit: 2,
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __qualified AS (
+    SELECT trace_id,
+        coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens'), toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id
+    HAVING total_tokens > 100
+),
+__scoped_traces_total AS (
+    SELECT trace_id,
+        toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
+        toString(multiIf(mapContains(attributes_string, 'gen_ai.user.id'), attributes_string['gen_ai.user.id'], NULL)) AS __GROUP_BY_KEY_1_gen_ai.user.id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model'))
+      AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
+    GROUP BY trace_id, __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id
+),
+__limit_cte AS (
+    SELECT __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id, sum(output_tokens) AS __result_0, count(trace_id) AS __result_1
+    FROM __scoped_traces_total
+    GROUP BY __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id
+    ORDER BY __result_0 DESC
+    LIMIT 2
+),
+__scoped_traces AS (
+    SELECT trace_id,
+        toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
+        toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
+        toString(multiIf(mapContains(attributes_string, 'gen_ai.user.id'), attributes_string['gen_ai.user.id'], NULL)) AS __GROUP_BY_KEY_1_gen_ai.user.id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model'))
+      AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
+      AND (toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)), toString(multiIf(mapContains(attributes_string, 'gen_ai.user.id'), attributes_string['gen_ai.user.id'], NULL))) GLOBAL IN (SELECT __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id FROM __limit_cte)
+    GROUP BY trace_id, ts, __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id
+)
+SELECT ts, __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id, sum(output_tokens) AS __result_0, count(trace_id) AS __result_1
+FROM __scoped_traces
+GROUP BY ts, __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// A time-series limit without group-by has nothing to rank: it is ignored, matching
+// the trace builder — the query equals its unlimited form.
+func TestBuild_TimeSeries_LimitWithoutGroupByIgnored(t *testing.T) {
+	b := newTestBuilder(t)
+	build := func(limit int) *qbtypes.Statement {
+		stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
+			qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal:       telemetrytypes.SignalTraces,
+				StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+				Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+				Limit:        limit,
+			}, nil)
+		require.NoError(t, err)
+		return stmt
+	}
+	assert.Equal(t, build(0).Query, build(5).Query)
+}
+
+// ---------------------------------------------------------------------------
+// Behavior / branch tests not covered by the goldens above
+// ---------------------------------------------------------------------------
+
+// Mixing span- and trace-level aggregations across one query is rejected.
+func TestBuild_Aggregation_MixedDomainsRejected(t *testing.T) {
+	b := newTestBuilder(t)
+	_, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{
+				{Expression: "avg(trace.output_tokens)"},
+				{Expression: "sum(gen_ai.usage.output_tokens)"},
+			},
+		}, nil)
+	require.ErrorContains(t, err, "cannot be mixed")
+}
+
+// Output-only aggregates are rejected in trace-level filters on the aggregation
+// path too (the raw and trace-list paths are covered elsewhere).
+func TestBuild_Aggregation_OutputOnlyFilterRejected(t *testing.T) {
+	b := newTestBuilder(t)
+	_, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
+			Filter:       &qbtypes.Filter{Expression: "trace.span_count > 3"},
+		}, nil)
+	require.ErrorContains(t, err, `aggregate "span_count" cannot be used`)
+}
+
+// Trace-level columns are rejected as group-by keys; order keys never reach the builder,
+// since request validation only admits group keys and aggregation aliases/expressions.
+func TestBuild_Aggregation_GroupByOrderValidation(t *testing.T) {
+	b := newTestBuilder(t)
+	ctx := context.Background()
+
+	_, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+			GroupBy:      []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.llm_call_count"}}},
+		}, nil)
+	require.ErrorContains(t, err, `grouping by trace-level aggregate "trace.llm_call_count" is not supported`)
+
+	req := qbtypes.QueryRangeRequest{
+		Start:       testStartMs,
+		End:         testEndMs,
+		RequestType: qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{
+			Queries: []qbtypes.QueryEnvelope{{
+				Type: qbtypes.QueryTypeBuilderAI,
+				Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+					Name:         "A",
+					Signal:       telemetrytypes.SignalTraces,
+					Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+					Order:        []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.total_tokens"}}, Direction: qbtypes.OrderDirectionDesc}},
+				},
+			}},
+		},
+	}
+	require.ErrorContains(t, req.Validate(), "invalid order by key")
+
+	_, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)", Alias: "avg_out"}},
+			Order:        []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "avg_out"}}, Direction: qbtypes.OrderDirectionAsc}},
+		}, nil)
+	require.NoError(t, err)
+}
+
+// Variables in trace-level conditions resolve as bound args; a dynamic __all__ drops the
+// condition, and an unresolved $var is rejected only as an unknown aggregate today.
+func TestBuild_FullSQL_Aggregation_VariablesInTraceFilter(t *testing.T) {
+	b := newTestBuilder(t)
+	ctx := context.Background()
+
+	q := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+		Signal:       telemetrytypes.SignalTraces,
+		Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+		Filter:       &qbtypes.Filter{Expression: "trace.output_tokens > $threshold"},
+	}
+	stmt, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, q,
+		map[string]qbtypes.VariableItem{"threshold": {Type: qbtypes.TextBoxVariableType, Value: float64(1000)}})
+	require.NoError(t, err)
+	assertSQLEqual(t, `
+WITH __qualified AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id
+    HAVING output_tokens > 1000
+),
+__scoped_traces AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
+    GROUP BY trace_id
+)
+SELECT avg(output_tokens) AS __result_0
+FROM __scoped_traces
+ORDER BY __result_0 DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+
+	// an unresolved $var is only rejected as an unknown aggregate today; a targeted
+	// "unknown variable" error is a separate concern
+	_, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, q, nil)
+	require.ErrorContains(t, err, `aggregate "$threshold" cannot be used`)
+
+	// __all__ drops the condition: the query equals its unfiltered form
+	stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, q,
+		map[string]qbtypes.VariableItem{"threshold": {Type: qbtypes.DynamicVariableType, Value: "__all__"}})
+	require.NoError(t, err)
+	unfiltered := q
+	unfiltered.Filter = nil
+	want, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, unfiltered, nil)
+	require.NoError(t, err)
+	assert.Equal(t, want.Query, stmt.Query)
+
+	// list variables render as IN with bound args; the scan selects only trace_id
+	// since no aggregation touches a per-trace column
+	stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "count(trace.trace_id)"}},
+			Filter:       &qbtypes.Filter{Expression: "trace.llm_call_count IN $counts"},
+		}, map[string]qbtypes.VariableItem{
+			"counts": {Type: qbtypes.QueryVariableType, Value: []any{float64(1), float64(2)}},
+		})
+	require.NoError(t, err)
+	assertSQLEqual(t, `
+WITH __qualified AS (
+    SELECT trace_id,
+        countIf(mapContains(attributes_string, 'gen_ai.request.model')) AS llm_call_count
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+    GROUP BY trace_id
+    HAVING llm_call_count IN (1, 2)
+),
+__scoped_traces AS (
+    SELECT trace_id
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
+    GROUP BY trace_id
+)
+SELECT count(trace_id) AS __result_0
+FROM __scoped_traces
+ORDER BY __result_0 DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Resource conditions on the native path: the __resource_filter CTE prunes the
+// qualification scan and the per-trace scan by fingerprint.
+func TestBuild_FullSQL_Aggregation_ResourceFilter_Native(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
+			Filter:       &qbtypes.Filter{Expression: "service.name = 'api' AND trace.output_tokens > 1000"},
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __resource_filter AS (
+    SELECT fingerprint
+    FROM signoz_traces.distributed_traces_v3_resource
+    WHERE (simpleJSONExtractString(labels, 'service.name') = 'api' AND labels LIKE '%service.name%' AND labels LIKE '%service.name":"api%')
+      AND seen_at_ts_bucket_start >= 1747945619
+      AND seen_at_ts_bucket_start <= 1747983448
+    GROUP BY fingerprint
+),
+__qualified AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
+    GROUP BY trace_id
+    HAVING output_tokens > 1000
+),
+__scoped_traces AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
+      AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
+    GROUP BY trace_id
+)
+SELECT avg(output_tokens) AS __result_0
+FROM __scoped_traces
+ORDER BY __result_0 DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// On the delegated path __trace_scope and the main query share one __resource_filter
+// CTE, so the resource table is scanned once.
+func TestBuild_FullSQL_Aggregation_ResourceFilter_Delegated(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(gen_ai.usage.output_tokens)"}},
+			Filter:       &qbtypes.Filter{Expression: "service.name = 'api' AND trace.output_tokens > 1000"},
+		}, nil)
+	require.NoError(t, err)
+
+	assertSQLEqual(t, `
+WITH __resource_filter AS (
+    SELECT fingerprint
+    FROM signoz_traces.distributed_traces_v3_resource
+    WHERE ((simpleJSONExtractString(labels, 'service.name') = 'api' AND labels LIKE '%service.name%' AND labels LIKE '%service.name":"api%'))
+      AND seen_at_ts_bucket_start >= 1747945619
+      AND seen_at_ts_bucket_start <= 1747983448
+    GROUP BY fingerprint
+),
+__trace_scope AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
+      AND resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
+    GROUP BY trace_id
+    HAVING output_tokens > 1000
+)
+SELECT sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS __result_0
+FROM signoz_traces.distributed_signoz_index_v3
+WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
+  AND trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)
+  AND (((mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))) AND ((multiIf(resource.service.name IS NOT NULL, resource.service.name::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = 'api' AND multiIf(resource.service.name IS NOT NULL, resource.service.name::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))
+  AND timestamp >= '1747947419000000000'
+  AND timestamp < '1747983448000000000'
+  AND ts_bucket_start >= 1747945619
+  AND ts_bucket_start <= 1747983448
+ORDER BY __result_0 DESC
+`, stmt)
+}
+
+// rate() divides by the window (scalar) / step (series). Per AggreFuncMap it counts
+// per-trace rows per second; it does not sum the column.
+func TestBuild_Aggregation_RateDividesByInterval(t *testing.T) {
+	b := newTestBuilder(t)
+	ctx := context.Background()
+	q := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+		Signal:       telemetrytypes.SignalTraces,
+		Aggregations: []qbtypes.TraceAggregation{{Expression: "rate(trace.llm_call_count)"}},
+	}
+
+	stmt, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, q, nil)
+	require.NoError(t, err)
+	assert.Contains(t, stmt.Query, "count(llm_call_count)/36029 AS __result_0") // (end-start) seconds
+
+	q.StepInterval = qbtypes.Step{Duration: 60 * time.Second}
+	stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries, q, nil)
+	require.NoError(t, err)
+	assert.Contains(t, stmt.Query, "count(llm_call_count)/60 AS __result_0")
+
+	// a sub-second window clamps the divisor instead of truncating it to zero
+	stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testStartMs+500, qbtypes.RequestTypeScalar, q, nil)
+	require.NoError(t, err)
+	assert.Contains(t, stmt.Query, "count(llm_call_count)/1 AS __result_0")
+}

pkg/statementbuilder/metricsstatementbuilder/statement_builder.go

@@ -428,20 +428,24 @@ func (b *StatementBuilder) buildTemporalAggDeltaFastPath(
 		sb.SelectMore(fmt.Sprintf("`%s`", GroupByColumnAlias(i, g.Name)))
 	}
 
-	aggCol, err := metricstelemetryschema.AggregationColumnForSamplesTable(
-		samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation,
-	)
-	if err != nil {
-		return "", nil, err
-	}
-	if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
-		// TODO(srikanthccv): should it be step interval or use [start_time_unix_nano](https://github.com/open-telemetry/opentelemetry-proto/blob/d3fb76d70deb0874692bd0ebe03148580d85f3bb/opentelemetry/proto/metrics/v1/metrics.proto#L400C11-L400C31)?
-		aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
-	}
-
+	var aggCol string
 	if query.Aggregations[0].SpaceAggregation.IsPercentile() &&
 		query.Aggregations[0].Type == metrictypes.ExpHistogramType {
+		// merging sketches already spans every series in the step, so neither a
+		// samples-table value column nor the rate divisor applies
 		aggCol = fmt.Sprintf("quantilesDDMerge(0.01, %f)(sketch)[1]", query.Aggregations[0].SpaceAggregation.Percentile())
+	} else {
+		col, err := metricstelemetryschema.AggregationColumnForSamplesTable(
+			samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation,
+		)
+		if err != nil {
+			return "", nil, err
+		}
+		aggCol = col
+		if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
+			// TODO(srikanthccv): should it be step interval or use [start_time_unix_nano](https://github.com/open-telemetry/opentelemetry-proto/blob/d3fb76d70deb0874692bd0ebe03148580d85f3bb/opentelemetry/proto/metrics/v1/metrics.proto#L400C11-L400C31)?
+			aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
+		}
 	}
 
 	sb.SelectMore(fmt.Sprintf("%s AS value", aggCol))

pkg/statementbuilder/metricsstatementbuilder/stmt_builder_test.go

@@ -126,6 +126,64 @@ func TestStatementBuilder(t *testing.T) {
 			},
 			expectedErr: nil,
 		},
+		{
+			name:        "test_exp_histogram_percentile_delta",
+			requestType: qbtypes.RequestTypeTimeSeries,
+			query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Signal:       telemetrytypes.SignalMetrics,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "signoz_latency",
+						Type:             metrictypes.ExpHistogramType,
+						Temporality:      metrictypes.Delta,
+						SpaceAggregation: metrictypes.SpaceAggregationPercentile95,
+					},
+				},
+				GroupBy: []qbtypes.GroupByKey{
+					{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: "service.name",
+						},
+					},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `__GROUP_BY_KEY_0_service.name`, quantilesDDMerge(0.01, 0.950000)(sketch)[1] AS value FROM signoz_metrics.distributed_exp_hist AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `__GROUP_BY_KEY_0_service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, `__GROUP_BY_KEY_0_service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `__GROUP_BY_KEY_0_service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `__GROUP_BY_KEY_0_service.name`, ts",
+				Args:  []any{"signoz_latency", uint64(1747936800000), uint64(1747983420000), "delta", "signoz_latency", uint64(1747947390000), uint64(1747983420000)},
+			},
+			expectedErr: nil,
+		},
+		{
+			// the sketch merge spans the whole step, so `rate` must not add a /step divisor
+			name:        "test_exp_histogram_percentile_delta_rate_time_aggregation",
+			requestType: qbtypes.RequestTypeTimeSeries,
+			query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Signal:       telemetrytypes.SignalMetrics,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "signoz_latency",
+						Type:             metrictypes.ExpHistogramType,
+						Temporality:      metrictypes.Delta,
+						TimeAggregation:  metrictypes.TimeAggregationRate,
+						SpaceAggregation: metrictypes.SpaceAggregationPercentile95,
+					},
+				},
+				GroupBy: []qbtypes.GroupByKey{
+					{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: "service.name",
+						},
+					},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `__GROUP_BY_KEY_0_service.name`, quantilesDDMerge(0.01, 0.950000)(sketch)[1] AS value FROM signoz_metrics.distributed_exp_hist AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `__GROUP_BY_KEY_0_service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, `__GROUP_BY_KEY_0_service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `__GROUP_BY_KEY_0_service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `__GROUP_BY_KEY_0_service.name`, ts",
+				Args:  []any{"signoz_latency", uint64(1747936800000), uint64(1747983420000), "delta", "signoz_latency", uint64(1747947390000), uint64(1747983420000)},
+			},
+			expectedErr: nil,
+		},
 		{
 			name:        "test_histogram_percentile1",
 			requestType: qbtypes.RequestTypeTimeSeries,

pkg/statementbuilder/scopedtracesstatementbuilder/statement_builder.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
-	"sort"
 	"strings"
 
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -19,7 +18,6 @@ import (
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
-	qbvariables "github.com/SigNoz/signoz/pkg/variables"
 	"github.com/huandu/go-sqlbuilder"
 )
 
@@ -117,6 +115,8 @@ func (b *scopedTraceStatementBuilder) Build(
 		return b.buildTraceListQuery(ctx, orgID, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), query, variables)
 	case qbtypes.RequestTypeRaw:
 		return b.buildDelegated(ctx, orgID, start, end, requestType, query, variables)
+	case qbtypes.RequestTypeScalar, qbtypes.RequestTypeTimeSeries:
+		return b.buildAggregation(ctx, orgID, start, end, requestType, query, variables)
 	default:
 		return nil, ErrUnsupportedRequestType
 	}
@@ -145,6 +145,63 @@ func (b *scopedTraceStatementBuilder) buildDelegated(
 	return b.traceStmtBuilder.Build(ctx, orgID, start, end, requestType, gated, variables)
 }
 
+// traceScopedStatementBuilder is the delegate's optional capability of constraining a
+// query to a set of trace ids (implemented by the traces statement builder).
+// traceScopeResource is the __resource_filter CTE traceScope's predicate references,
+// shared with the delegate's own resource filter so the table is scanned once.
+type traceScopedStatementBuilder interface {
+	BuildTraceScoped(ctx context.Context, orgID valuer.UUID, start, end uint64, requestType qbtypes.RequestType, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], variables map[string]qbtypes.VariableItem, traceScope, traceScopeResource *qbtypes.Statement) (*qbtypes.Statement, error)
+}
+
+// buildDelegatedAggregation serves span-level scalar/time-series through the standard
+// trace builder, with the gate ANDed into the span-level filter part; a trace-level
+// part becomes a qualification the delegate constrains trace_id by.
+func (b *scopedTraceStatementBuilder) buildDelegatedAggregation(
+	ctx context.Context,
+	orgID valuer.UUID,
+	start, end uint64,
+	requestType qbtypes.RequestType,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	var spanExpr, traceExpr string
+	var err error
+	if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
+		spanExpr, traceExpr, err = querybuilder.SplitFilterForAggregates(query.Filter.Expression, b.aggregateAliasSet())
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	gate := b.scope.FilterExpression
+	expr := gate
+	if strings.TrimSpace(spanExpr) != "" {
+		expr = fmt.Sprintf("(%s) AND (%s)", gate, spanExpr)
+	}
+
+	// shallow copy; only Filter is replaced, caller's query untouched
+	gated := query
+	gated.Filter = &qbtypes.Filter{Expression: expr}
+
+	if strings.TrimSpace(traceExpr) == "" {
+		return b.traceStmtBuilder.Build(ctx, orgID, start, end, requestType, gated, variables)
+	}
+
+	scoped, ok := b.traceStmtBuilder.(traceScopedStatementBuilder)
+	if !ok {
+		return nil, errors.NewInternalf(errors.CodeInternal, "trace statement builder does not support trace-scoped queries")
+	}
+	scope, scopeResource, err := b.buildQualifiedStatement(ctx, orgID, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), traceExpr, query, variables)
+	if err != nil {
+		return nil, err
+	}
+	if scope == nil {
+		// every trace-level condition was dropped by variable resolution
+		return b.traceStmtBuilder.Build(ctx, orgID, start, end, requestType, gated, variables)
+	}
+	return scoped.BuildTraceScoped(ctx, orgID, start, end, requestType, gated, variables, scope, scopeResource)
+}
+
 // buildTraceListQuery wires the CTE pipeline (start/end are nanoseconds):
 // matched (windowed, mask-pruned top-N trace_ids) → ranked (their [start,end] from
 // the summary table) → buckets (ts_bucket_start prune) → enrichment (every per-trace
@@ -166,9 +223,13 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
 		limit = 100
 	}
 
+	filterExpr := ""
+	if query.Filter != nil {
+		filterExpr = query.Filter.Expression
+	}
 	// Condition args bind into the builder an expression is embedded in, so the
 	// matched and enrichment passes each resolve against their own builder.
-	keys, err := b.fetchKeys(ctx, orgID)
+	keys, err := b.fetchKeys(ctx, orgID, spanFilterSelectors(filterExpr)...)
 	if err != nil {
 		return nil, err
 	}
@@ -186,23 +247,17 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
 	if err != nil {
 		return nil, err
 	}
-	orderableSet := orderableAliasSet(resolved)
-	filterableSet := filterableAliasSet(resolved)
-
 	resourceFrag, resourceArgs, resourcePred, err := b.maybeAttachResourceFilter(ctx, orgID, query, start, end, variables)
 	if err != nil {
 		return nil, err
 	}
 
-	fp, err := b.splitFilter(ctx, orgID, query, b.aggregateAliasSet(), filterableSet, start, end, variables, matchedSB)
+	fp, err := b.splitFilter(ctx, orgID, query, b.aggregateAliasSet(), keys, start, end, variables, matchedSB)
 	if err != nil {
 		return nil, err
 	}
 
-	matchedFrag, matchedArgs, err := b.buildMatchedCTE(matchedSB, start, end, startBucket, endBucket, resolved, orders, orderableSet, filterableSet, maskExpr, fp, resourcePred, limit, query.Offset)
-	if err != nil {
-		return nil, err
-	}
+	matchedFrag, matchedArgs := b.buildMatchedCTE(matchedSB, start, end, startBucket, endBucket, resolved, orders, maskExpr, fp, resourcePred, limit, query.Offset)
 	rankedFrag, rankedArgs := b.buildRankedCTE(start, end)
 
 	adj := querybuilder.BucketAdjustment // 30-min bucket width in seconds
@@ -258,9 +313,10 @@ func (b *scopedTraceStatementBuilder) maybeAttachResourceFilter(
 		"resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)", nil
 }
 
-func (b *scopedTraceStatementBuilder) fetchKeys(ctx context.Context, orgID valuer.UUID) (map[string][]*telemetrytypes.TelemetryFieldKey, error) {
+func (b *scopedTraceStatementBuilder) fetchKeys(ctx context.Context, orgID valuer.UUID, extra ...*telemetrytypes.FieldKeySelector) (map[string][]*telemetrytypes.TelemetryFieldKey, error) {
 	fields := b.resolverFieldKeys()
-	selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(fields))
+	selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(fields)+len(extra))
+	selectors = append(selectors, extra...)
 	for _, k := range fields {
 		selectors = append(selectors, &telemetrytypes.FieldKeySelector{
 			Name:              k.Name,
@@ -329,10 +385,9 @@ func (b *scopedTraceStatementBuilder) resolveMask(ctx context.Context, orgID val
 }
 
 type resolvedColumn struct {
-	alias      string
-	expr       string
-	orderable  bool
-	filterable bool
+	alias     string
+	expr      string
+	orderable bool
 }
 
 func (b *scopedTraceStatementBuilder) resolveColumns(ctx context.Context, orgID valuer.UUID, start, end uint64, cols *columnResolver, preds *predicateResolver) ([]resolvedColumn, error) {
@@ -342,7 +397,7 @@ func (b *scopedTraceStatementBuilder) resolveColumns(ctx context.Context, orgID
 		if err != nil {
 			return nil, err
 		}
-		out = append(out, resolvedColumn{alias: c.Alias, expr: expr, orderable: c.Orderable, filterable: c.Filterable})
+		out = append(out, resolvedColumn{alias: c.Alias, expr: expr, orderable: c.Orderable})
 	}
 	return out, nil
 }
@@ -384,29 +439,30 @@ func (b *scopedTraceStatementBuilder) resolveListOrders(order []qbtypes.OrderBy,
 	return orders, nil
 }
 
-// filterParts is the user filter split into a span-level predicate and a trace-level
-// HAVING expression.
+// filterParts is the user filter split into a span-level predicate and the resolved
+// trace-level HAVING (nil when there is none).
 type filterParts struct {
 	spanPred      string
 	hasSpanFilter bool
-	havingExpr    string
+	having        *traceHaving
 	warnings      []string
 	warningsURL   string
 }
 
-// splitFilter splits query.Filter into a span-level predicate (args bound into sb)
-// and a trace-level HAVING (explicit query.Having ANDed on), then validates the
-// trace-level part against the matched-pass aggregates.
-func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID valuer.UUID, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet, filterableSet map[string]struct{}, start, end uint64, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (filterParts, error) {
+// splitFilter splits query.Filter into a span-level predicate and a trace-level
+// HAVING (explicit query.Having ANDed on before resolution); args bind into sb.
+// keys must cover the filter's span-level selectors.
+func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID valuer.UUID, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet map[string]struct{}, keys map[string][]*telemetrytypes.TelemetryFieldKey, start, end uint64, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (filterParts, error) {
 	var fp filterParts
+	havingExpr := ""
 	if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
 		spanExpr, traceExpr, err := querybuilder.SplitFilterForAggregates(query.Filter.Expression, classifySet)
 		if err != nil {
 			return fp, err
 		}
-		fp.havingExpr = traceExpr
+		havingExpr = traceExpr
 		if strings.TrimSpace(spanExpr) != "" {
-			pred, warnings, url, err := b.resolveSpanPredicate(ctx, orgID, start, end, spanExpr, variables, sb)
+			pred, warnings, url, err := b.resolveSpanPredicate(ctx, orgID, start, end, spanExpr, keys, variables, sb)
 			if err != nil {
 				return fp, err
 			}
@@ -419,37 +475,23 @@ func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID val
 		}
 	}
 	if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
-		if fp.havingExpr != "" {
-			fp.havingExpr = fmt.Sprintf("(%s) AND (%s)", fp.havingExpr, query.Having.Expression)
+		if havingExpr != "" {
+			havingExpr = fmt.Sprintf("(%s) AND (%s)", havingExpr, query.Having.Expression)
 		} else {
-			fp.havingExpr = query.Having.Expression
+			havingExpr = query.Having.Expression
 		}
 	}
-	// the HAVING is a plain text rewrite, so substitute variables here
-	if strings.TrimSpace(fp.havingExpr) != "" && len(variables) > 0 {
-		replaced, err := qbvariables.ReplaceVariablesInExpression(fp.havingExpr, variables)
-		if err != nil {
-			return fp, err
-		}
-		fp.havingExpr = replaced
-	}
-	if err := validateAggregateFilter(fp.havingExpr, filterableSet); err != nil {
+	having, err := b.resolveTraceHaving(ctx, havingExpr, variables, sb)
+	if err != nil {
 		return fp, err
 	}
+	fp.having = having
 	return fp, nil
 }
 
 // resolveSpanPredicate resolves a span-level filter expression to a bare boolean
-// predicate, args bound into sb.
-func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context, orgID valuer.UUID, start, end uint64, expr string, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (string, []string, string, error) {
-	selectors := querybuilder.QueryStringToKeysSelectors(expr)
-	for i := range selectors {
-		selectors[i].Signal = telemetrytypes.SignalTraces
-	}
-	keys, _, err := b.metadataStore.GetKeysMulti(ctx, orgID, querybuilder.ExpandKeySelectorsForFamilies(ctx, orgID, b.fl, selectors))
-	if err != nil {
-		return "", nil, "", err
-	}
+// predicate, args bound into sb; keys must cover the expression's selectors.
+func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context, orgID valuer.UUID, start, end uint64, expr string, keys map[string][]*telemetrytypes.TelemetryFieldKey, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (string, []string, string, error) {
 	prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
 		Context:          ctx,
 		OrgID:            orgID,
@@ -478,8 +520,8 @@ func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context,
 // span filter + HAVING + ORDER BY + LIMIT/OFFSET, selecting only the aliases ORDER BY
 // / HAVING reference. Expressions carry $n markers bound to sb, so each can appear
 // several times and every occurrence resolves to the same arg.
-func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, orderableSet, filterableSet map[string]struct{}, maskExpr string, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
-	needed := neededMatchedAliases(orders, fp.havingExpr, orderableSet)
+func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, maskExpr string, fp filterParts, resourcePred string, limit, offset int) (string, []any) {
+	needed := neededMatchedAliases(orders, fp.having)
 	selects := []string{"trace_id"}
 	for _, rc := range resolved {
 		if _, ok := needed[rc.alias]; !ok {
@@ -516,22 +558,8 @@ func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuild
 		having = append(having, "countIf("+maskExpr+") > 0")
 		having = append(having, "countIf("+fp.spanPred+") > 0")
 	}
-	if strings.TrimSpace(fp.havingExpr) != "" {
-		// the rewriter matches raw key text, so map the trace. form alongside the bare name
-		columnMap := make(map[string]string, len(filterableSet)*2)
-		for a := range filterableSet {
-			columnMap[a] = quoteAlias(a)
-			columnMap[telemetrytypes.FieldContextTrace.StringValue()+"."+a] = quoteAlias(a)
-		}
-		hv, err := querybuilder.NewHavingExpressionRewriter().Rewrite(fp.havingExpr, columnMap)
-		if err != nil {
-			return "", nil, err
-		}
-		if hv != "" {
-			// escape user text so a literal $ isn't read as an arg marker; the countIf
-			// entries hold live $n markers and must stay unescaped
-			having = append(having, sqlbuilder.Escape(hv))
-		}
+	if fp.having != nil {
+		having = append(having, fp.having.pred)
 	}
 	if len(having) > 0 {
 		sb.Having(strings.Join(having, " AND "))
@@ -544,7 +572,7 @@ func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuild
 	}
 
 	sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
-	return fmt.Sprintf("matched AS (%s)", sql), args, nil
+	return fmt.Sprintf("matched AS (%s)", sql), args
 }
 
 // buildRankedCTE builds `ranked`: [start,end] bounds per matched trace from the
@@ -585,8 +613,9 @@ func (b *scopedTraceStatementBuilder) buildEnrichmentSelect(sb *sqlbuilder.Selec
 	return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
 }
 
-// aggregateAliasSet is every trace-level column alias, used to classify filter keys;
-// SpanLevel columns are filtered span-level, so skip them.
+// aggregateAliasSet recognises trace-level keys — display-only aliases included, so one
+// gets a targeted error instead of falling through as a span attribute (what a predicate
+// may actually use is filterableColumnSet). SpanLevel columns are filtered span-level.
 func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
 	set := make(map[string]struct{}, len(b.scope.Columns))
 	for _, c := range b.scope.Columns {
@@ -597,70 +626,36 @@ func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
 	return set
 }
 
-// orderableAliasSet is the subset of aliases computable in the matched pass.
-func orderableAliasSet(resolved []resolvedColumn) map[string]struct{} {
-	set := make(map[string]struct{})
-	for _, rc := range resolved {
-		if rc.orderable {
-			set[rc.alias] = struct{}{}
-		}
-	}
-	return set
-}
-
-// filterableAliasSet is the subset of aliases usable in the trace-level filter.
-func filterableAliasSet(resolved []resolvedColumn) map[string]struct{} {
-	set := make(map[string]struct{})
-	for _, rc := range resolved {
-		if rc.filterable {
-			set[rc.alias] = struct{}{}
-		}
-	}
-	return set
-}
-
 // neededMatchedAliases is the minimal alias set the matched pass must select: those
-// in ORDER BY plus those in the aggregate HAVING.
-func neededMatchedAliases(orders []listOrder, havingExpr string, orderableSet map[string]struct{}) map[string]struct{} {
+// in ORDER BY plus those the resolved trace-level HAVING touches.
+func neededMatchedAliases(orders []listOrder, having *traceHaving) map[string]struct{} {
 	needed := make(map[string]struct{})
 	for _, o := range orders {
 		needed[o.alias] = struct{}{}
 	}
-	for _, name := range traceAggregateNames(havingExpr) {
-		if _, ok := orderableSet[name]; ok {
+	if having != nil {
+		for name := range having.used {
 			needed[name] = struct{}{}
 		}
 	}
 	return needed
 }
 
-// traceAggregateNames extracts the aggregate names a trace-level HAVING references;
-// only unspecified- and trace-context selectors name aggregates.
-func traceAggregateNames(havingExpr string) []string {
-	var names []string
-	for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
-		if sel.FieldContext == telemetrytypes.FieldContextUnspecified || sel.FieldContext == telemetrytypes.FieldContextTrace {
-			names = append(names, sel.Name)
-		}
-	}
-	return names
-}
-
-// validateAggregateFilter rejects a trace-level filter referencing an aggregate that
-// is not filterable.
+// validateAggregateFilter rejects filters on aggregates that are not filterable
+// (e.g. span_count) upfront, since inside the where-clause visitor the error would
+// surface only as a detail of a combined one. Only unspecified- and trace-context
+// selectors name aggregates.
 func validateAggregateFilter(havingExpr string, filterableSet map[string]struct{}) error {
 	if strings.TrimSpace(havingExpr) == "" {
 		return nil
 	}
-	allowed := make([]string, 0, len(filterableSet))
-	for a := range filterableSet {
-		allowed = append(allowed, a)
-	}
-	sort.Strings(allowed)
-	for _, name := range traceAggregateNames(havingExpr) {
-		if _, ok := filterableSet[name]; !ok {
+	for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
+		if sel.FieldContext != telemetrytypes.FieldContextUnspecified && sel.FieldContext != telemetrytypes.FieldContextTrace {
+			continue
+		}
+		if _, ok := filterableSet[sel.Name]; !ok {
 			return errors.NewInvalidInputf(errors.CodeInvalidInput,
-				"aggregate %q cannot be used in the trace-list filter; filterable aggregates: %s", name, strings.Join(allowed, ", "))
+				"aggregate %q cannot be used in a trace-level filter; filterable aggregates: %s", sel.Name, strings.Join(sortedAliases(filterableSet), ", "))
 		}
 	}
 	return nil
@@ -675,6 +670,19 @@ func orderClause(orders []listOrder) []string {
 	return append(out, "trace_id DESC")
 }
 
+// spanFilterSelectors are the metadata selectors for every key a filter expression
+// references, for batching into a single GetKeysMulti fetch.
+func spanFilterSelectors(expr string) []*telemetrytypes.FieldKeySelector {
+	if strings.TrimSpace(expr) == "" {
+		return nil
+	}
+	selectors := querybuilder.QueryStringToKeysSelectors(expr)
+	for i := range selectors {
+		selectors[i].Signal = telemetrytypes.SignalTraces
+	}
+	return selectors
+}
+
 // quoteAlias backticks an alias containing characters special to the SQL builder.
 func quoteAlias(alias string) string {
 	if strings.ContainsAny(alias, ".$`") {

pkg/statementbuilder/scopedtracesstatementbuilder/trace_aggregation.go

@@ -0,0 +1,783 @@
+package scopedtracesstatementbuilder
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+
+	chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+// The per-trace values these aggregations read are window-clipped and span-filtered,
+// unlike the list's enrichment pass over every span of the whole trace, so the same
+// column reads differently in each.
+
+// traceAggregation is one aggregation rewritten to run over the per-trace scan.
+type traceAggregation struct {
+	expr   string              // rewritten SQL over the per-trace column aliases
+	used   map[string]struct{} // per-trace aliases referenced
+	isRate bool
+}
+
+// buildAggregation routes by aggregation domain: bare keys delegate to the standard
+// trace builder, trace.-prefixed aggregates run over the per-trace scan.
+func (b *scopedTraceStatementBuilder) buildAggregation(
+	ctx context.Context,
+	orgID valuer.UUID,
+	start, end uint64,
+	requestType qbtypes.RequestType,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	traceAggs, err := b.classifyAggregations(query.Aggregations)
+	if err != nil {
+		return nil, err
+	}
+	if err := b.validateGroupBy(query); err != nil {
+		return nil, err
+	}
+	if len(traceAggs) == 0 {
+		return b.buildDelegatedAggregation(ctx, orgID, start, end, requestType, query, variables)
+	}
+	return b.buildTraceAggregationQuery(ctx, orgID, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), requestType, query, variables, traceAggs)
+}
+
+// classifyAggregations returns the rewritten trace-domain aggregations, nil when all
+// are span-domain; mixing the two domains is rejected.
+func (b *scopedTraceStatementBuilder) classifyAggregations(aggs []qbtypes.TraceAggregation) ([]traceAggregation, error) {
+	// permission, not recognition: unknown names are reported against exactly this set
+	traceCols := b.orderableColumnSet()
+	var out []traceAggregation
+	spanCount := 0
+	for _, agg := range aggs {
+		ta, isTrace, err := rewriteTraceAggregation(agg.Expression, traceCols)
+		if err != nil {
+			return nil, err
+		}
+		if isTrace {
+			out = append(out, *ta)
+		} else {
+			spanCount++
+		}
+	}
+	if len(out) > 0 && spanCount > 0 {
+		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"span-level and trace-level (trace.) aggregations cannot be mixed in one query")
+	}
+	return out, nil
+}
+
+// orderableColumnSet is what a trace-level aggregation may use;
+// recognising a key as trace-level is aggregateAliasSet's job.
+func (b *scopedTraceStatementBuilder) orderableColumnSet() map[string]struct{} {
+	set := make(map[string]struct{})
+	for _, c := range b.scope.Columns {
+		if c.Orderable {
+			set[c.Alias] = struct{}{}
+		}
+	}
+	return set
+}
+
+// filterableColumnSet is what a trace-level filter predicate may use.
+func (b *scopedTraceStatementBuilder) filterableColumnSet() map[string]struct{} {
+	set := make(map[string]struct{})
+	for _, c := range b.scope.Columns {
+		if c.Filterable {
+			set[c.Alias] = struct{}{}
+		}
+	}
+	return set
+}
+
+// validateGroupBy rejects trace-level columns as group-by keys with a targeted error
+// (not the field mapper's generic "field not found"). Order keys need no check here:
+// request validation only admits group keys and aggregation aliases/expressions.
+func (b *scopedTraceStatementBuilder) validateGroupBy(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) error {
+	// recognition, not permission: a display-only alias must be named here to be rejected
+	// rather than reaching the field mapper as a span attribute
+	aliases := b.aggregateAliasSet()
+	for _, gb := range query.GroupBy {
+		key := gb.TelemetryFieldKey
+		key.Normalize()
+		// a bare name may be a span column sharing the alias (duration_nano, timestamp)
+		if key.FieldContext != telemetrytypes.FieldContextTrace {
+			continue
+		}
+		if _, ok := aliases[key.Name]; ok {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput,
+				"grouping by trace-level aggregate %q is not supported; group by span attributes instead (e.g. service.name)", gb.Name)
+		}
+	}
+	return nil
+}
+
+// rewriteTraceAggregation rewrites an aggregation over trace.-prefixed columns to run
+// on the per-trace scan (trace.output_tokens → output_tokens, functions mapped via
+// AggreFuncMap); a pure span-level expression returns isTrace=false for the delegate.
+func rewriteTraceAggregation(expr string, traceCols map[string]struct{}) (*traceAggregation, bool, error) {
+	p := chparser.NewParser("SELECT " + expr)
+	stmts, err := p.ParseStmts()
+	if err != nil {
+		return nil, false, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "failed to parse aggregation expression %q", expr)
+	}
+	if len(stmts) == 0 {
+		return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid aggregation expression %q", expr)
+	}
+	sel, ok := stmts[0].(*chparser.SelectQuery)
+	if !ok || len(sel.SelectItems) == 0 {
+		return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid aggregation expression %q", expr)
+	}
+
+	v := &traceAggVisitor{traceCols: traceCols, used: make(map[string]struct{})}
+	if err := sel.SelectItems[0].Accept(v); err != nil {
+		return nil, false, err
+	}
+	if !v.hasTrace {
+		return nil, false, nil
+	}
+	if v.hasSpan {
+		return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"aggregation %q mixes trace-level (trace.) and span-level columns; use one domain per aggregation", expr)
+	}
+	// the interval divides the rendered expression as a whole, so a second aggregation
+	// alongside the rate would be divided too
+	if v.isRate && v.aggCount > 1 {
+		return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"aggregation %q combines a rate with another aggregation; the rate interval would divide both, so give each its own aggregation", expr)
+	}
+	return &traceAggregation{expr: chparser.Format(sel.SelectItems[0]), used: v.used, isRate: v.isRate}, true, nil
+}
+
+// traceAggVisitor classifies column references and rewrites trace.-prefixed ones in
+// place; the ancestor stack tells a column identifier from a path segment, function
+// name, or alias, and rejects trace. columns inside *If combinators.
+type traceAggVisitor struct {
+	chparser.DefaultASTVisitor
+	traceCols map[string]struct{}
+	used      map[string]struct{}
+	stack     []chparser.Expr
+	aggCount  int
+	hasTrace  bool
+	hasSpan   bool
+	isRate    bool
+}
+
+func (v *traceAggVisitor) Enter(expr chparser.Expr) { v.stack = append(v.stack, expr) }
+func (v *traceAggVisitor) Leave(expr chparser.Expr) { v.stack = v.stack[:len(v.stack)-1] }
+
+// parent is the node enclosing the one currently being visited (the visited node
+// itself is the stack top).
+func (v *traceAggVisitor) parent() chparser.Expr {
+	if len(v.stack) < 2 {
+		return nil
+	}
+	return v.stack[len(v.stack)-2]
+}
+
+func (v *traceAggVisitor) enclosingCombinator() (string, bool) {
+	for _, e := range v.stack {
+		fn, ok := e.(*chparser.FunctionExpr)
+		if !ok {
+			continue
+		}
+		if agg, known := querybuilder.AggreFuncMap[valuer.NewString(strings.ToLower(fn.Name.Name))]; known && agg.FuncCombinator {
+			return fn.Name.Name, true
+		}
+	}
+	return "", false
+}
+
+// enclosingAggregate walks the ancestor stack; AggreFuncMap holds only aggregates and
+// VisitFunctionExpr rejects any name missing from it, so a known name is enough.
+func (v *traceAggVisitor) enclosingAggregate() bool {
+	for _, e := range v.stack {
+		fn, ok := e.(*chparser.FunctionExpr)
+		if !ok {
+			continue
+		}
+		if _, known := querybuilder.AggreFuncMap[valuer.NewString(strings.ToLower(fn.Name.Name))]; known {
+			return true
+		}
+	}
+	return false
+}
+
+// VisitPath classifies a dotted reference (trace.output_tokens); trace-level ones are
+// rewritten in place to the bare per-trace alias.
+func (v *traceAggVisitor) VisitPath(p *chparser.Path) error {
+	col, isTrace := traceColumnFromPath(p)
+	if !isTrace {
+		v.hasSpan = true
+		return nil
+	}
+	if err := v.acceptTraceColumn(chparser.Format(p), col); err != nil {
+		return err
+	}
+	p.Fields = p.Fields[len(p.Fields)-1:]
+	p.Fields[0].Name = col
+	return nil
+}
+
+// VisitIdent classifies a plain identifier (a backquoted `trace.output_tokens` is
+// trace-level); path segments, function names, and aliases are structural, not columns.
+func (v *traceAggVisitor) VisitIdent(i *chparser.Ident) error {
+	switch parent := v.parent().(type) {
+	case *chparser.Path:
+		return nil // segments are classified whole by VisitPath
+	case *chparser.FunctionExpr:
+		if parent.Name == i {
+			return nil
+		}
+	case *chparser.ColumnExpr:
+		if parent.Alias == i {
+			return nil
+		}
+	}
+	key := telemetrytypes.GetFieldKeyFromKeyText(i.Name)
+	if key.FieldContext != telemetrytypes.FieldContextTrace || key.Name == "" {
+		v.hasSpan = true
+		return nil
+	}
+	if err := v.acceptTraceColumn(i.Name, key.Name); err != nil {
+		return err
+	}
+	i.Name = key.Name
+	return nil
+}
+
+func (v *traceAggVisitor) acceptTraceColumn(ref, col string) error {
+	if name, in := v.enclosingCombinator(); in {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"%q over trace-level (trace.) columns is not supported; put the trace-level condition in the filter expression instead", name)
+	}
+	// trace_id is always selected by the per-trace scan (count(trace.trace_id)
+	// counts traces); everything else must be a scope column.
+	if col != "trace_id" {
+		if _, known := v.traceCols[col]; !known {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput,
+				"unknown trace-level aggregation column %q; usable columns: %s", ref, strings.Join(sortedAliases(v.traceCols), ", "))
+		}
+		v.used[col] = struct{}{}
+	}
+	// ungrouped, a bare per-trace column would make the outer SELECT emit one row per
+	// trace instead of one aggregated row
+	if !v.enclosingAggregate() {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"trace-level column %q must be inside an aggregation function (e.g. avg(%s))", ref, ref)
+	}
+	v.hasTrace = true
+	return nil
+}
+
+// VisitFunctionExpr validates and maps the function name. Children were already
+// visited (post-order), so classification is complete for this subtree.
+func (v *traceAggVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
+	name := strings.ToLower(fn.Name.Name)
+	aggFunc, ok := querybuilder.AggreFuncMap[valuer.NewString(name)]
+	if !ok {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "unrecognized function: %s", name)
+	}
+	if fn.Params != nil && fn.Params.Items != nil && len(fn.Params.Items.Items) > 0 && aggFunc.FuncCombinator {
+		// combinator predicates over span columns stay span-level (countIf(has_error=true))
+		v.hasSpan = true
+		return nil
+	}
+	fn.Name.Name = aggFunc.FuncName
+	v.aggCount++
+	if aggFunc.Rate {
+		v.isRate = true
+	}
+	return nil
+}
+
+// traceColumnFromPath returns the per-trace column a dotted reference names
+// (trace.output_tokens -> output_tokens, trace.a.b -> a.b).
+func traceColumnFromPath(p *chparser.Path) (string, bool) {
+	key := telemetrytypes.GetFieldKeyFromKeyText(chparser.Format(p))
+	if key.FieldContext != telemetrytypes.FieldContextTrace || key.Name == "" {
+		return "", false
+	}
+	return key.Name, true
+}
+
+func sortedAliases(set map[string]struct{}) []string {
+	out := make([]string, 0, len(set))
+	for a := range set {
+		out = append(out, a)
+	}
+	sort.Strings(out)
+	return out
+}
+
+// ---------------------------------------------------------------------------
+// Qualification + per-trace scan
+// ---------------------------------------------------------------------------
+
+// buildQualifiedStatement selects the trace ids whose window-clipped aggregates satisfy
+// the trace-level filter. The second statement (nil without resource conditions) is the
+// __resource_filter CTE the scope's predicate references; the embedder emits it exactly
+// once, shared with its own resource filter. start/end are ns; both statements are nil
+// when variable resolution dropped every condition.
+func (b *scopedTraceStatementBuilder) buildQualifiedStatement(
+	ctx context.Context,
+	orgID valuer.UUID,
+	start, end uint64,
+	traceExpr string,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, *qbtypes.Statement, error) {
+	keys, err := b.fetchKeys(ctx, orgID)
+	if err != nil {
+		return nil, nil, err
+	}
+	sb := sqlbuilder.NewSelectBuilder()
+	maskExpr, resolved, err := b.resolveFor(ctx, orgID, start, end, keys, sb)
+	if err != nil {
+		return nil, nil, err
+	}
+	having, err := b.resolveTraceHaving(ctx, traceExpr, variables, sb)
+	if err != nil {
+		return nil, nil, err
+	}
+	if having == nil {
+		return nil, nil, nil
+	}
+	// nil when the filter has no resource-attribute conditions
+	resourceStmt, err := b.resourceFilterStmtBuilder.Build(ctx, orgID, start, end, qbtypes.RequestTypeRaw, query, variables)
+	if err != nil {
+		return nil, nil, err
+	}
+	var resourcePred string
+	if resourceStmt != nil {
+		resourcePred = "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)"
+	}
+	sql, args := b.buildPerTraceScan(sb, start, end, resolved, maskExpr, perTraceScanOpts{
+		needed:       having.used,
+		havingPred:   having.pred,
+		resourcePred: resourcePred,
+	})
+	return &qbtypes.Statement{Query: sql, Args: args}, resourceStmt, nil
+}
+
+// groupColumn holds a resolved, arg-free span-attribute expression.
+type groupColumn struct {
+	alias string
+	expr  string
+}
+
+// groupByColumnAlias prefixes the i-th group-by dimension so the alias cannot shadow the
+// span column its expression reads; the querier (stripKeyAlias) strips it back off.
+func groupByColumnAlias(i int, name string) string {
+	return fmt.Sprintf("__GROUP_BY_KEY_%d_%s", i, name)
+}
+
+// orderColumn is the SQL identifier a non-aggregation order key sorts by: the
+// positional alias when the key names a group-by dimension, else the key itself.
+func orderColumn(orderKey string, groupBy []qbtypes.GroupByKey) string {
+	for i := range groupBy {
+		if groupBy[i].Name == orderKey {
+			return groupByColumnAlias(i, groupBy[i].Name)
+		}
+	}
+	return orderKey
+}
+
+// perTraceScanOpts parametrize one windowed, mask-pruned GROUP BY trace_id scan.
+// All expressions are already resolved against the scan's builder.
+type perTraceScanOpts struct {
+	stepSeconds  int64 // >0 → bucket per-trace values by time (ts column)
+	groupCols    []groupColumn
+	needed       map[string]struct{} // per-trace aliases to select
+	spanPred     string              // resolved span-level filter, ANDed per span
+	resourcePred string              // resource-fingerprint prune (CTE reference or inline subquery)
+	qualified    bool                // constrain to __qualified
+	limitPred    string              // top-N group prune (GLOBAL IN __limit_cte)
+	havingPred   string              // resolved HAVING predicate over the selected aliases
+}
+
+func (b *scopedTraceStatementBuilder) buildPerTraceScan(sb *sqlbuilder.SelectBuilder, start, end uint64, resolved []resolvedColumn, maskExpr string, o perTraceScanOpts) (string, []any) {
+	startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
+	endBucket := end / querybuilder.NsToSeconds
+
+	selects := []string{"trace_id"}
+	if o.stepSeconds > 0 {
+		selects = append(selects, fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts", o.stepSeconds))
+	}
+	for _, gc := range o.groupCols {
+		selects = append(selects, fmt.Sprintf("toString(%s) AS `%s`", gc.expr, gc.alias))
+	}
+	for _, rc := range resolved {
+		if _, ok := o.needed[rc.alias]; !ok {
+			continue
+		}
+		selects = append(selects, rc.expr+" AS "+quoteAlias(rc.alias))
+	}
+	sb.Select(selects...)
+	sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
+
+	where := []string{
+		sb.GE("timestamp", fmt.Sprintf("%d", start)),
+		sb.L("timestamp", fmt.Sprintf("%d", end)),
+		sb.GE("ts_bucket_start", startBucket),
+		sb.LE("ts_bucket_start", endBucket),
+		maskExpr,
+	}
+	if strings.TrimSpace(o.spanPred) != "" {
+		where = append(where, o.spanPred)
+	}
+	if o.resourcePred != "" {
+		where = append(where, o.resourcePred)
+	}
+	if o.qualified {
+		where = append(where, "trace_id GLOBAL IN (SELECT trace_id FROM __qualified)")
+	}
+	if o.limitPred != "" {
+		where = append(where, o.limitPred)
+	}
+	sb.Where(where...)
+
+	groupBy := []string{"trace_id"}
+	if o.stepSeconds > 0 {
+		groupBy = append(groupBy, "ts")
+	}
+	for _, gc := range o.groupCols {
+		groupBy = append(groupBy, "`"+gc.alias+"`")
+	}
+	sb.GroupBy(groupBy...)
+	if strings.TrimSpace(o.havingPred) != "" {
+		sb.Having(o.havingPred)
+	}
+	return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+}
+
+// groupBySelectors are the metadata selectors for the group-by keys, for batching
+// into a single GetKeysMulti fetch.
+func groupBySelectors(groupBy []qbtypes.GroupByKey) []*telemetrytypes.FieldKeySelector {
+	selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(groupBy))
+	for i := range groupBy {
+		selectors = append(selectors, &telemetrytypes.FieldKeySelector{
+			Name:              groupBy[i].Name,
+			Signal:            telemetrytypes.SignalTraces,
+			FieldContext:      groupBy[i].FieldContext,
+			FieldDataType:     groupBy[i].FieldDataType,
+			SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
+		})
+	}
+	return selectors
+}
+
+// resolveGroupColumns resolves group-by keys through the field mapper for selection
+// inside the per-trace scan; keys must cover the group-by selectors.
+func (b *scopedTraceStatementBuilder) resolveGroupColumns(ctx context.Context, orgID valuer.UUID, start, end uint64, groupBy []qbtypes.GroupByKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) ([]groupColumn, error) {
+	if len(groupBy) == 0 {
+		return nil, nil
+	}
+	out := make([]groupColumn, 0, len(groupBy))
+	for i := range groupBy {
+		expr, err := b.fm.ColumnExpressionFor(ctx, orgID, start, end, &groupBy[i].TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, groupColumn{alias: groupByColumnAlias(i, groupBy[i].Name), expr: sqlbuilder.Escape(expr)})
+	}
+	return out, nil
+}
+
+// ---------------------------------------------------------------------------
+// Native trace-domain aggregation query
+// ---------------------------------------------------------------------------
+
+// scanContext is one per-scan resolution: a fresh builder with the mask, columns,
+// span predicate, and optionally the trace-level HAVING resolved against it.
+type scanContext struct {
+	sb       *sqlbuilder.SelectBuilder
+	maskExpr string
+	resolved []resolvedColumn
+	spanPred string
+	having   *traceHaving
+	warnings []string
+	warnURL  string
+}
+
+func (b *scopedTraceStatementBuilder) newScanContext(
+	ctx context.Context,
+	orgID valuer.UUID,
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	spanExpr, traceExpr string,
+	variables map[string]qbtypes.VariableItem,
+) (*scanContext, error) {
+	sc := &scanContext{sb: sqlbuilder.NewSelectBuilder()}
+	var err error
+	sc.maskExpr, sc.resolved, err = b.resolveFor(ctx, orgID, start, end, keys, sc.sb)
+	if err != nil {
+		return nil, err
+	}
+	if strings.TrimSpace(spanExpr) != "" {
+		pred, warns, url, err := b.resolveSpanPredicate(ctx, orgID, start, end, spanExpr, keys, variables, sc.sb)
+		if err != nil {
+			return nil, err
+		}
+		sc.spanPred, sc.warnings, sc.warnURL = pred, warns, url
+	}
+	if strings.TrimSpace(traceExpr) != "" {
+		sc.having, err = b.resolveTraceHaving(ctx, traceExpr, variables, sc.sb)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return sc, nil
+}
+
+// buildTraceAggregationQuery aggregates over the per-trace scan: __qualified (when the
+// filter has a trace-level part) → __scoped_traces → outer aggregation. start/end are ns.
+func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
+	ctx context.Context,
+	orgID valuer.UUID,
+	start, end uint64,
+	requestType qbtypes.RequestType,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	variables map[string]qbtypes.VariableItem,
+	traceAggs []traceAggregation,
+) (*qbtypes.Statement, error) {
+	var spanExpr, traceExpr string
+	var err error
+	if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
+		// the broad set so a condition on a display-only alias still lands in the
+		// trace-level part, where resolveTraceHaving rejects it by name
+		spanExpr, traceExpr, err = querybuilder.SplitFilterForAggregates(query.Filter.Expression, b.aggregateAliasSet())
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	keys, err := b.fetchKeys(ctx, orgID, append(spanFilterSelectors(spanExpr), groupBySelectors(query.GroupBy)...)...)
+	if err != nil {
+		return nil, err
+	}
+
+	resourceFrag, resourceArgs, resourcePred, err := b.maybeAttachResourceFilter(ctx, orgID, query, start, end, variables)
+	if err != nil {
+		return nil, err
+	}
+
+	var cteFragments []string
+	var cteArgs [][]any
+	if resourceFrag != "" {
+		cteFragments = append(cteFragments, resourceFrag)
+		cteArgs = append(cteArgs, resourceArgs)
+	}
+
+	// __qualified: its own scan resolution, HAVING = the trace-level filter part
+	qualified := false
+	if strings.TrimSpace(traceExpr) != "" {
+		qsc, err := b.newScanContext(ctx, orgID, start, end, keys, "", traceExpr, variables)
+		if err != nil {
+			return nil, err
+		}
+		if qsc.having != nil {
+			qsql, qargs := b.buildPerTraceScan(qsc.sb, start, end, qsc.resolved, qsc.maskExpr, perTraceScanOpts{
+				needed:       qsc.having.used,
+				havingPred:   qsc.having.pred,
+				resourcePred: resourcePred,
+			})
+			cteFragments = append(cteFragments, fmt.Sprintf("__qualified AS (%s)", qsql))
+			cteArgs = append(cteArgs, qargs)
+			qualified = true
+		}
+	}
+
+	groupCols, err := b.resolveGroupColumns(ctx, orgID, start, end, query.GroupBy, keys)
+	if err != nil {
+		return nil, err
+	}
+	groupNames := make([]string, 0, len(groupCols))
+	for _, gc := range groupCols {
+		groupNames = append(groupNames, "`"+gc.alias+"`")
+	}
+
+	needed := make(map[string]struct{})
+	for _, ta := range traceAggs {
+		for a := range ta.used {
+			needed[a] = struct{}{}
+		}
+	}
+
+	// a window or step under one second would truncate to a zero divisor
+	windowSeconds := max((end-start)/querybuilder.NsToSeconds, 1)
+	stepSeconds := int64(0)
+	rateInterval := windowSeconds
+	if requestType == qbtypes.RequestTypeTimeSeries {
+		stepSeconds = int64(query.StepInterval.Seconds())
+		rateInterval = max(uint64(stepSeconds), 1)
+	}
+
+	// outer aggregation over the per-trace rows
+	sb := sqlbuilder.NewSelectBuilder()
+	selects := []string{}
+	if stepSeconds > 0 {
+		selects = append(selects, "ts")
+	}
+	selects = append(selects, groupNames...)
+	for i, ta := range traceAggs {
+		selects = append(selects, fmt.Sprintf("%s AS __result_%d", ta.rendered(rateInterval), i))
+	}
+	sb.Select(selects...)
+	sb.From("__scoped_traces")
+
+	// grouped, limited time series → rank groups on whole-window per-trace values
+	// (exact for non-composable aggregates) and prune the main scan to the top-N.
+	limitPred := ""
+	if requestType == qbtypes.RequestTypeTimeSeries && query.Limit > 0 && len(groupCols) > 0 {
+		tsc, err := b.newScanContext(ctx, orgID, start, end, keys, spanExpr, "", variables)
+		if err != nil {
+			return nil, err
+		}
+		totalSQL, totalArgs := b.buildPerTraceScan(tsc.sb, start, end, tsc.resolved, tsc.maskExpr, perTraceScanOpts{
+			groupCols:    groupCols,
+			needed:       needed,
+			spanPred:     tsc.spanPred,
+			resourcePred: resourcePred,
+			qualified:    qualified,
+		})
+		cteFragments = append(cteFragments, fmt.Sprintf("__scoped_traces_total AS (%s)", totalSQL))
+		cteArgs = append(cteArgs, totalArgs)
+
+		limitSQL, limitArgs := outerLimitSQL(query, traceAggs, groupNames, windowSeconds)
+		cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", limitSQL))
+		cteArgs = append(cteArgs, limitArgs)
+
+		exprs := make([]string, 0, len(groupCols))
+		for _, gc := range groupCols {
+			exprs = append(exprs, "toString("+gc.expr+")")
+		}
+		limitPred = fmt.Sprintf("(%s) GLOBAL IN (SELECT %s FROM __limit_cte)",
+			strings.Join(exprs, ", "), strings.Join(groupNames, ", "))
+	}
+
+	msc, err := b.newScanContext(ctx, orgID, start, end, keys, spanExpr, "", variables)
+	if err != nil {
+		return nil, err
+	}
+	perTraceSQL, perTraceArgs := b.buildPerTraceScan(msc.sb, start, end, msc.resolved, msc.maskExpr, perTraceScanOpts{
+		stepSeconds:  stepSeconds,
+		groupCols:    groupCols,
+		needed:       needed,
+		spanPred:     msc.spanPred,
+		resourcePred: resourcePred,
+		qualified:    qualified,
+		limitPred:    limitPred,
+	})
+	cteFragments = append(cteFragments, fmt.Sprintf("__scoped_traces AS (%s)", perTraceSQL))
+	cteArgs = append(cteArgs, perTraceArgs)
+
+	groupBys := []string{}
+	if stepSeconds > 0 {
+		groupBys = append(groupBys, "ts")
+	}
+	groupBys = append(groupBys, groupNames...)
+	if len(groupBys) > 0 {
+		sb.GroupBy(groupBys...)
+	}
+
+	if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
+		rewritten, err := querybuilder.NewHavingExpressionRewriter().RewriteForTraces(query.Having.Expression, query.Aggregations)
+		if err != nil {
+			return nil, err
+		}
+		sb.Having(sqlbuilder.Escape(rewritten))
+	}
+
+	if requestType == qbtypes.RequestTypeTimeSeries {
+		if len(query.Order) != 0 {
+			for _, orderBy := range query.Order {
+				if _, ok := traceAggOrderIndex(orderBy, query); !ok {
+					sb.OrderBy(fmt.Sprintf("`%s` %s", orderColumn(orderBy.Key.Name, query.GroupBy), orderBy.Direction.StringValue()))
+				}
+			}
+			sb.OrderBy("ts desc")
+		}
+	} else {
+		for _, orderBy := range query.Order {
+			if idx, ok := traceAggOrderIndex(orderBy, query); ok {
+				sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
+			} else {
+				sb.OrderBy(fmt.Sprintf("`%s` %s", orderColumn(orderBy.Key.Name, query.GroupBy), orderBy.Direction.StringValue()))
+			}
+		}
+		if len(query.Order) == 0 {
+			sb.OrderBy("__result_0 DESC")
+		}
+		if query.Limit > 0 {
+			sb.Limit(query.Limit)
+		}
+	}
+
+	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL + " SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000"
+	finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
+
+	return &qbtypes.Statement{
+		Query:          finalSQL,
+		Args:           finalArgs,
+		Warnings:       msc.warnings,
+		WarningsDocURL: msc.warnURL,
+	}, nil
+}
+
+// rendered divides a rate aggregation by the interval (step for time series, window
+// length for scalar); the divisor applies to the whole expression, which holds only
+// because a rate must be the sole aggregation.
+func (ta traceAggregation) rendered(rateInterval uint64) string {
+	if ta.isRate {
+		return fmt.Sprintf("%s/%d", ta.expr, rateInterval)
+	}
+	return ta.expr
+}
+
+// outerLimitSQL ranks groups on whole-window per-trace values, so a non-composable
+// aggregate (avg) ranks exactly rather than over bucketed rows.
+func outerLimitSQL(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], traceAggs []traceAggregation, groupNames []string, windowSeconds uint64) (string, []any) {
+	sb := sqlbuilder.NewSelectBuilder()
+	selects := append([]string{}, groupNames...)
+	for i, ta := range traceAggs {
+		selects = append(selects, fmt.Sprintf("%s AS __result_%d", ta.rendered(windowSeconds), i))
+	}
+	sb.Select(selects...)
+	sb.From("__scoped_traces_total")
+	sb.GroupBy(groupNames...)
+	for _, orderBy := range query.Order {
+		if idx, ok := traceAggOrderIndex(orderBy, query); ok {
+			sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
+		} else {
+			sb.OrderBy(fmt.Sprintf("`%s` %s", orderColumn(orderBy.Key.Name, query.GroupBy), orderBy.Direction.StringValue()))
+		}
+	}
+	if len(query.Order) == 0 {
+		sb.OrderBy("__result_0 DESC")
+	}
+	sb.Limit(query.Limit)
+	return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+}
+
+// traceAggOrderIndex reports whether an order key refers to the i-th aggregation
+// (by alias, expression, or index), mirroring the trace builder.
+func traceAggOrderIndex(k qbtypes.OrderBy, q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) (int, bool) {
+	for i, agg := range q.Aggregations {
+		if k.Key.Name == agg.Alias ||
+			k.Key.Name == agg.Expression ||
+			k.Key.Name == fmt.Sprintf("%d", i) {
+			return i, true
+		}
+	}
+	return 0, false
+}

pkg/statementbuilder/scopedtracesstatementbuilder/trace_aggregation_test.go

@@ -0,0 +1,75 @@
+package scopedtracesstatementbuilder
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRewriteTraceAggregation(t *testing.T) {
+	cols := map[string]struct{}{
+		"input_tokens": {}, "output_tokens": {}, "total_tokens": {}, "llm_call_count": {}, "max_llm_latency_ns": {},
+	}
+
+	cases := []struct {
+		name    string
+		expr    string
+		isTrace bool
+		want    string // rewritten expr, only checked when isTrace
+		used    []string
+		wantErr string
+	}{
+		{name: "avg trace col", expr: "avg(trace.output_tokens)", isTrace: true, want: "avg(output_tokens)", used: []string{"output_tokens"}},
+		{name: "sum trace col", expr: "sum(trace.total_tokens)", isTrace: true, want: "sum(total_tokens)", used: []string{"total_tokens"}},
+		{name: "count traces", expr: "count(trace.trace_id)", isTrace: true, want: "count(trace_id)"},
+		{name: "p90 trace col", expr: "p90(trace.max_llm_latency_ns)", isTrace: true, want: "quantile(0.90)(max_llm_latency_ns)", used: []string{"max_llm_latency_ns"}},
+		{name: "arithmetic between trace cols", expr: "avg(trace.output_tokens + trace.input_tokens)", isTrace: true, want: "avg(output_tokens + input_tokens)", used: []string{"output_tokens", "input_tokens"}},
+		{name: "arithmetic with constant", expr: "sum(trace.output_tokens * 1.5)", isTrace: true, want: "sum(output_tokens * 1.5)", used: []string{"output_tokens"}},
+		{name: "ratio of two aggregations", expr: "sum(trace.output_tokens)/count(trace.trace_id)", isTrace: true, want: "sum(output_tokens) / count(trace_id)", used: []string{"output_tokens"}},
+		{name: "backquoted trace col", expr: "avg(`trace.output_tokens`)", isTrace: true, want: "avg(`output_tokens`)", used: []string{"output_tokens"}},
+		{name: "bare count is span-level", expr: "count()", isTrace: false},
+		{name: "span attribute is span-level", expr: "sum(gen_ai.usage.output_tokens)", isTrace: false},
+		{name: "countIf span predicate is span-level", expr: "countIf(has_error = true)", isTrace: false},
+		{name: "mixed domains in one expression", expr: "sum(trace.output_tokens) + sum(gen_ai.usage.input_tokens)", wantErr: "mixes trace-level"},
+		{name: "mixed domains in one function", expr: "sum(trace.output_tokens + gen_ai.usage.input_tokens)", wantErr: "mixes trace-level"},
+		{name: "output-only column rejected", expr: "avg(trace.span_count)", wantErr: "unknown trace-level aggregation column"},
+		{name: "unknown column rejected", expr: "avg(trace.bogus)", wantErr: "unknown trace-level aggregation column"},
+		// a dotted column keeps every segment after the prefix, so it is reported whole
+		{name: "multi segment column rejected by full name", expr: "avg(trace.service.name)", wantErr: `"trace.service.name"`},
+		{name: "bare trace identifier is span-level", expr: "avg(trace)", isTrace: false},
+		{name: "countIf over trace col rejected", expr: "countIf(trace.output_tokens > 1000)", wantErr: "not supported"},
+		{name: "bare trace col rejected", expr: "trace.output_tokens", wantErr: "must be inside an aggregation function"},
+		{name: "backquoted bare trace col rejected", expr: "`trace.output_tokens`", wantErr: "must be inside an aggregation function"},
+		{name: "bare trace_id rejected", expr: "trace.trace_id", wantErr: "must be inside an aggregation function"},
+		{name: "arithmetic outside an aggregation rejected", expr: "trace.output_tokens + trace.input_tokens", wantErr: "must be inside an aggregation function"},
+		{name: "trace col beside an aggregation rejected", expr: "sum(trace.output_tokens) + trace.input_tokens", wantErr: "must be inside an aggregation function"},
+		{name: "aggregation scaled by a constant", expr: "sum(trace.output_tokens) * 2", isTrace: true, want: "sum(output_tokens) * 2", used: []string{"output_tokens"}},
+		{name: "rate over traces", expr: "rate(trace.trace_id)", isTrace: true, want: "count(trace_id)"},
+		{name: "rate_sum trace col", expr: "rate_sum(trace.output_tokens)", isTrace: true, want: "sum(output_tokens)", used: []string{"output_tokens"}},
+		// the interval divides the whole rendered expression, so a second aggregation
+		// alongside a rate would be divided too
+		{name: "rate mixed with another aggregation rejected", expr: "rate(trace.trace_id) + avg(trace.output_tokens)", wantErr: "combines a rate with another aggregation"},
+		{name: "ratio of two rates rejected", expr: "rate_sum(trace.output_tokens)/rate_sum(trace.input_tokens)", wantErr: "combines a rate with another aggregation"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ta, isTrace, err := rewriteTraceAggregation(tc.expr, cols)
+			if tc.wantErr != "" {
+				require.ErrorContains(t, err, tc.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.isTrace, isTrace)
+			if !tc.isTrace {
+				return
+			}
+			assert.Equal(t, tc.want, ta.expr)
+			for _, u := range tc.used {
+				assert.Contains(t, ta.used, u)
+			}
+			assert.Len(t, ta.used, len(tc.used))
+		})
+	}
+}

pkg/statementbuilder/scopedtracesstatementbuilder/trace_having.go

@@ -0,0 +1,144 @@
+package scopedtracesstatementbuilder
+
+import (
+	"context"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	qbvariables "github.com/SigNoz/signoz/pkg/variables"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+// traceHaving is the resolved trace-level filter part: a HAVING predicate over the
+// per-trace aliases plus the aliases it references (so scans select only those).
+type traceHaving struct {
+	pred string
+	used map[string]struct{}
+}
+
+// resolveTraceHaving runs a trace-level filter through the standard where-clause
+// pipeline against the per-trace aliases, so operators, bound args, and __all__ behave
+// as in span filters. Returns nil when nothing is left to filter; args bind into sb.
+func (b *scopedTraceStatementBuilder) resolveTraceHaving(ctx context.Context, expr string, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (*traceHaving, error) {
+	if strings.TrimSpace(expr) == "" {
+		return nil, nil //nolint:nilnil
+	}
+	// replaced before validation so variable literals are not mistaken for aggregate
+	// names; an unresolved $var is left in place and fails validation as an unknown one
+	if len(variables) > 0 {
+		replaced, err := qbvariables.ReplaceVariablesInExpression(expr, variables)
+		if err != nil {
+			return nil, err
+		}
+		expr = replaced
+		if strings.TrimSpace(expr) == "" {
+			return nil, nil //nolint:nilnil
+		}
+	}
+	allowed := b.filterableColumnSet()
+	// upfront targeted errors; the visitor folds them into a combined "Found N errors"
+	if err := validateAggregateFilter(expr, allowed); err != nil {
+		return nil, err
+	}
+
+	// both spellings resolve here: the key parser strips the trace. prefix into
+	// FieldContextTrace, which matches this entry's context
+	fieldKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey, len(allowed))
+	for alias := range allowed {
+		key := &telemetrytypes.TelemetryFieldKey{Name: alias, FieldContext: telemetrytypes.FieldContextTrace}
+		fieldKeys[alias] = []*telemetrytypes.TelemetryFieldKey{key}
+	}
+
+	cb := &aliasConditionBuilder{allowed: allowed, used: make(map[string]struct{})}
+	prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
+		Context:          ctx,
+		Logger:           b.logger,
+		ConditionBuilder: cb,
+		FieldKeys:        fieldKeys,
+		Variables:        variables,
+		Builder:          sb,
+	})
+	if err != nil {
+		return nil, err
+	}
+	if prepared.IsEmpty() {
+		return nil, nil //nolint:nilnil
+	}
+	return &traceHaving{pred: prepared.Expr, used: cb.used}, nil
+}
+
+// aliasConditionBuilder renders filter conditions directly against the per-trace
+// aliases, recording the ones it touches; a key resolving to no alias is an error.
+type aliasConditionBuilder struct {
+	allowed map[string]struct{}
+	used    map[string]struct{}
+}
+
+var _ qbtypes.ConditionBuilder = (*aliasConditionBuilder)(nil)
+
+func (c *aliasConditionBuilder) ConditionFor(
+	_ context.Context,
+	_ valuer.UUID,
+	_, _ uint64,
+	key *telemetrytypes.TelemetryFieldKey,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	_ qbtypes.ConditionBuilderOptions,
+	op qbtypes.FilterOperator,
+	value any,
+	sb *sqlbuilder.SelectBuilder,
+) ([]string, []string, error) {
+	matching := keys[key.Name]
+	if len(matching) == 0 {
+		return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"aggregate %q cannot be used in a trace-level filter; filterable aggregates: %s",
+			key.Name, strings.Join(sortedAliases(c.allowed), ", "))
+	}
+	alias := matching[0].Name
+	c.used[alias] = struct{}{}
+	col := quoteAlias(alias)
+
+	var cond string
+	switch op {
+	case qbtypes.FilterOperatorEqual:
+		cond = sb.E(col, value)
+	case qbtypes.FilterOperatorNotEqual:
+		cond = sb.NE(col, value)
+	case qbtypes.FilterOperatorGreaterThan:
+		cond = sb.G(col, value)
+	case qbtypes.FilterOperatorGreaterThanOrEq:
+		cond = sb.GE(col, value)
+	case qbtypes.FilterOperatorLessThan:
+		cond = sb.L(col, value)
+	case qbtypes.FilterOperatorLessThanOrEq:
+		cond = sb.LE(col, value)
+	case qbtypes.FilterOperatorIn, qbtypes.FilterOperatorNotIn:
+		values, ok := value.([]any)
+		if !ok {
+			values = []any{value}
+		}
+		if op == qbtypes.FilterOperatorIn {
+			cond = sb.In(col, values...)
+		} else {
+			cond = sb.NotIn(col, values...)
+		}
+	case qbtypes.FilterOperatorBetween, qbtypes.FilterOperatorNotBetween:
+		values, ok := value.([]any)
+		if !ok || len(values) != 2 {
+			return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
+				"between on trace-level aggregate %q requires exactly two values", alias)
+		}
+		if op == qbtypes.FilterOperatorBetween {
+			cond = sb.Between(col, values[0], values[1])
+		} else {
+			cond = sb.NotBetween(col, values[0], values[1])
+		}
+	default:
+		return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"trace-level aggregate %q supports only comparison operators (=, !=, <, <=, >, >=, in, between)", alias)
+	}
+	return []string{cond}, nil, nil
+}

pkg/statementbuilder/tracesstatementbuilder/statement_builder.go

@@ -33,6 +33,12 @@ type traceQueryStatementBuilder struct {
 	aggExprRewriter                qbtypes.AggExprRewriter
 	fl                             flagger.Flagger
 	skipResourceFingerprintEnabled bool
+	// traceScope, set only on the per-call copy made by BuildTraceScoped, constrains
+	// queries to spans whose trace_id is in the __trace_scope CTE.
+	traceScope *qbtypes.Statement
+	// traceScopeResource is the __resource_filter CTE traceScope's predicate references,
+	// emitted only when this builder's own resource filter did not already emit it.
+	traceScopeResource *qbtypes.Statement
 }
 
 var _ qbtypes.StatementBuilder[qbtypes.TraceAggregation] = (*traceQueryStatementBuilder)(nil)
@@ -97,6 +103,41 @@ func NewTraceQueryStatementBuilder(
 	}
 }
 
+// BuildTraceScoped is Build constrained to trace_ids selected by traceScope; the
+// receiver is copied so the shared builder stays stateless.
+func (b *traceQueryStatementBuilder) BuildTraceScoped(
+	ctx context.Context,
+	orgID valuer.UUID,
+	start uint64,
+	end uint64,
+	requestType qbtypes.RequestType,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	variables map[string]qbtypes.VariableItem,
+	traceScope, traceScopeResource *qbtypes.Statement,
+) (*qbtypes.Statement, error) {
+	scoped := *b
+	scoped.traceScope = traceScope
+	scoped.traceScopeResource = traceScopeResource
+	return scoped.Build(ctx, orgID, start, end, requestType, query, variables)
+}
+
+// attachTraceScope adds the trace-scope condition to sb and returns the CTE fragments
+// + args to prepend; resourceEmitted reports whether the query already carries the
+// __resource_filter CTE, so the scope's copy is emitted only when it does not.
+func (b *traceQueryStatementBuilder) attachTraceScope(sb *sqlbuilder.SelectBuilder, resourceEmitted bool) ([]string, [][]any) {
+	if b.traceScope == nil {
+		return nil, nil
+	}
+	sb.Where("trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)")
+	var frags []string
+	var args [][]any
+	if b.traceScopeResource != nil && !resourceEmitted {
+		frags = append(frags, fmt.Sprintf("__resource_filter AS (%s)", b.traceScopeResource.Query))
+		args = append(args, b.traceScopeResource.Args)
+	}
+	return append(frags, fmt.Sprintf("__trace_scope AS (%s)", b.traceScope.Query)), append(args, b.traceScope.Args)
+}
+
 // Build builds a SQL query for traces based on the given parameters.
 func (b *traceQueryStatementBuilder) Build(
 	ctx context.Context,
@@ -521,6 +562,11 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
 		cteArgs = append(cteArgs, args)
 	}
 
+	if scopeFrags, scopeArgs := b.attachTraceScope(sb, frag != ""); len(scopeFrags) > 0 {
+		cteFragments = append(cteFragments, scopeFrags...)
+		cteArgs = append(cteArgs, scopeArgs...)
+	}
+
 	sb.SelectMore(fmt.Sprintf(
 		"toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts",
 		int64(query.StepInterval.Seconds()),
@@ -681,6 +727,13 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
 		cteArgs = append(cteArgs, args)
 	}
 
+	// skipResourceCTE means this scalar is embedded as a CTE of a time-series query,
+	// which has already emitted the __trace_scope fragment — add only the condition.
+	if scopeFrags, scopeArgs := b.attachTraceScope(sb, frag != ""); len(scopeFrags) > 0 && !skipResourceCTE {
+		cteFragments = append(cteFragments, scopeFrags...)
+		cteArgs = append(cteArgs, scopeArgs...)
+	}
+
 	allAggChArgs := []any{}
 
 	fieldNames := make([]string, 0, len(query.GroupBy))

pkg/types/querybuildertypes/querybuildertypesv5/validation.go

@@ -374,7 +374,7 @@ func (q *QueryBuilderQuery[T]) validateAggregations(cfg validationConfig) error
 	return nil
 }
 
-func (m MetricAggregation) ValidateForType() error {
+func (m MetricAggregation) ValidateForTypeAndTemporality() error {
 	if m.SpaceAggregation.IsPercentile() && !m.Type.IsPercentileSpaceAggregationAllowed() {
 		return errors.Newf(
 			errors.TypeInvalidInput,
@@ -384,6 +384,17 @@ func (m MetricAggregation) ValidateForType() error {
 			m.Type.StringValue(),
 		)
 	}
+	// reading a step's distribution out of a cumulative sketch would mean
+	// subtracting the previous point's sketch, which ClickHouse cannot do
+	if m.Type == metrictypes.ExpHistogramType && m.Temporality != metrictypes.Delta {
+		return errors.Newf(
+			errors.TypeUnsupported,
+			errors.CodeUnsupported,
+			"metric `%s` is an exponential histogram recorded with `%s` temporality, which cannot be queried; only `delta` exponential histograms are supported",
+			m.MetricName,
+			m.Temporality.StringValue(),
+		)
+	}
 	return nil
 }
 

pkg/types/querybuildertypes/querybuildertypesv5/validation_test.go

@@ -1517,10 +1517,11 @@ func TestNonAggregationFieldsSkipped(t *testing.T) {
 	})
 }
 
-func TestMetricAggregationValidateForType(t *testing.T) {
+func TestMetricAggregationValidateForTypeAndTemporality(t *testing.T) {
 	cases := []struct {
 		name             string
 		metricType       metrictypes.Type
+		temporality      metrictypes.Temporality
 		spaceAggregation metrictypes.SpaceAggregation
 		comparisonParam  *metrictypes.ComparisonSpaceAggregationParam
 		wantErr          bool
@@ -1532,11 +1533,33 @@ func TestMetricAggregationValidateForType(t *testing.T) {
 			wantErr:          false,
 		},
 		{
-			name:             "percentile on exponential histogram is allowed",
+			name:             "percentile on delta exponential histogram is allowed",
 			metricType:       metrictypes.ExpHistogramType,
+			temporality:      metrictypes.Delta,
 			spaceAggregation: metrictypes.SpaceAggregationPercentile99,
 			wantErr:          false,
 		},
+		{
+			name:             "cumulative exponential histogram is not allowed",
+			metricType:       metrictypes.ExpHistogramType,
+			temporality:      metrictypes.Cumulative,
+			spaceAggregation: metrictypes.SpaceAggregationPercentile99,
+			wantErr:          true,
+		},
+		{
+			name:             "exponential histogram with unresolved temporality is not allowed",
+			metricType:       metrictypes.ExpHistogramType,
+			temporality:      metrictypes.Unknown,
+			spaceAggregation: metrictypes.SpaceAggregationPercentile99,
+			wantErr:          true,
+		},
+		{
+			name:             "cumulative histogram is unaffected by the exponential histogram rule",
+			metricType:       metrictypes.HistogramType,
+			temporality:      metrictypes.Cumulative,
+			spaceAggregation: metrictypes.SpaceAggregationPercentile95,
+			wantErr:          false,
+		},
 		{
 			name:             "percentile on summary is not allowed",
 			metricType:       metrictypes.SummaryType,
@@ -1562,10 +1585,11 @@ func TestMetricAggregationValidateForType(t *testing.T) {
 			agg := MetricAggregation{
 				MetricName:                      "test_metric",
 				Type:                            tc.metricType,
+				Temporality:                     tc.temporality,
 				SpaceAggregation:                tc.spaceAggregation,
 				ComparisonSpaceAggregationParam: tc.comparisonParam,
 			}
-			err := agg.ValidateForType()
+			err := agg.ValidateForTypeAndTemporality()
 			if tc.wantErr && err == nil {
 				t.Errorf("expected error, got nil")
 			}

pkg/types/savedviewtypes/savedview.go

@@ -23,10 +23,11 @@ var (
 const savedViewNameSuffixLen = 8
 
 var (
-	SourceTraces  = Source{valuer.NewString("traces")}
-	SourceLogs    = Source{valuer.NewString("logs")}
-	SourceMetrics = Source{valuer.NewString("metrics")}
-	SourceMeter   = Source{valuer.NewString("meter")}
+	SourceTraces          = Source{valuer.NewString("traces")}
+	SourceLogs            = Source{valuer.NewString("logs")}
+	SourceMetrics         = Source{valuer.NewString("metrics")}
+	SourceMeter           = Source{valuer.NewString("meter")}
+	SourceAIObservability = Source{valuer.NewString("ai_observability")}
 )
 
 type SavedView struct {
@@ -117,12 +118,13 @@ func (Source) Enum() []any {
 		SourceLogs,
 		SourceMetrics,
 		SourceMeter,
+		SourceAIObservability,
 	}
 }
 
 func (s Source) Validate() error {
 	switch s {
-	case SourceTraces, SourceLogs, SourceMetrics, SourceMeter:
+	case SourceTraces, SourceLogs, SourceMetrics, SourceMeter, SourceAIObservability:
 		return nil
 	default:
 		return errors.NewInvalidInputf(ErrCodeSavedViewInvalidInput, "invalid source: %s", s.StringValue())

pkg/types/savedviewtypes/savedview_test.go

@@ -39,6 +39,7 @@ func TestSourceValidate(t *testing.T) {
 		{name: "logs", source: SourceLogs},
 		{name: "metrics", source: SourceMetrics},
 		{name: "meter", source: SourceMeter},
+		{name: "ai_observability", source: SourceAIObservability},
 		{name: "unknown is rejected", source: Source{valuer.NewString("bogus")}, expectError: true},
 	}
 

pkg/types/savedviewtypes/spec_test.go

@@ -173,6 +173,21 @@ func TestSavedViewSpecValidate(t *testing.T) {
 			},
 			expectError: false,
 		},
+		{
+			name: "builder_ai_query is valid",
+			spec: SavedViewSpec{
+				DisplayName: "My View",
+				PanelType:   PanelTypeList,
+				RequestType: qbtypes.RequestTypeRaw,
+				Queries: []qbtypes.QueryEnvelope{{
+					Type: qbtypes.QueryTypeBuilderAI,
+					Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+						Signal: telemetrytypes.SignalTraces,
+					},
+				}},
+			},
+			expectError: false,
+		},
 		{
 			name: "graph panel query with no aggregation is still rejected",
 			spec: SavedViewSpec{

tests/fixtures/metrics.py

@@ -132,7 +132,15 @@ class MetricsSample(ABC):
 
 
 class MetricsExpHist(ABC):
-    """Represents a row in the exp_hist table for exponential histograms."""
+    """Represents a row in the exp_hist table for exponential histograms.
+
+    `observations` must be non-empty; ClickHouse folds it into the `sketch`
+    AggregateFunction state on insert.
+
+    TODO: take an ExponentialHistogramDataPoint and compute the sketch bytes the
+    way the collector does, so the fixture exercises the real write path instead
+    of having ClickHouse build the state.
+    """
 
     env: str
     temporality: str
@@ -143,7 +151,7 @@ class MetricsExpHist(ABC):
     sum: np.float64
     min: np.float64
     max: np.float64
-    sketch: bytes
+    observations: list[int]
     flags: np.uint32
 
     def __init__(
@@ -151,11 +159,7 @@ class MetricsExpHist(ABC):
         metric_name: str,
         fingerprint: np.uint64,
         timestamp: datetime.datetime,
-        count: int,
-        sum_value: float,
-        min_value: float,
-        max_value: float,
-        sketch: bytes = b"",
+        observations: list[int],
         temporality: str = "Unspecified",
         env: str = "default",
         flags: int = 0,
@@ -165,28 +169,13 @@ class MetricsExpHist(ABC):
         self.metric_name = metric_name
         self.fingerprint = fingerprint
         self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
-        self.count = np.uint64(count)
-        self.sum = np.float64(sum_value)
-        self.min = np.float64(min_value)
-        self.max = np.float64(max_value)
-        self.sketch = sketch
+        self.observations = observations
+        self.count = np.uint64(len(observations))
+        self.sum = np.float64(sum(observations))
+        self.min = np.float64(min(observations))
+        self.max = np.float64(max(observations))
         self.flags = np.uint32(flags)
 
-    def to_row(self) -> list:
-        return [
-            self.env,
-            self.temporality,
-            self.metric_name,
-            self.fingerprint,
-            self.unix_milli,
-            self.count,
-            self.sum,
-            self.min,
-            self.max,
-            self.sketch,
-            self.flags,
-        ]
-
 
 class MetricsMetadata(ABC):
     """Represents a row in the metadata table for metric metadata."""
@@ -429,6 +418,73 @@ class Metrics(ABC):
         return metrics
 
 
+class ExpHistogramMetrics(ABC):
+    """High-level exponential histogram representation. Produces both time series
+    and exp_hist entries."""
+
+    metric_name: str
+    labels: dict[str, str]
+    temporality: str
+    timestamp: datetime.datetime
+    observations: list[int]
+
+    @property
+    def time_series(self) -> MetricsTimeSeries:
+        return self._time_series
+
+    @property
+    def exp_hist(self) -> MetricsExpHist:
+        return self._exp_hist
+
+    def __init__(
+        self,
+        metric_name: str,
+        observations: list[int],
+        labels: dict[str, str] = {},
+        timestamp: datetime.datetime | None = None,
+        temporality: str = "Delta",
+        flags: int = 0,
+        description: str = "",
+        unit: str = "",
+        env: str = "default",
+        resource_attributes: dict[str, str] = {},
+        scope_attributes: dict[str, str] = {},
+    ) -> None:
+        if timestamp is None:
+            timestamp = datetime.datetime.now()
+        self.metric_name = metric_name
+        self.labels = labels
+        self.temporality = temporality
+        self.timestamp = timestamp
+        self.observations = observations
+
+        self._time_series = MetricsTimeSeries(
+            metric_name=metric_name,
+            labels=labels,
+            timestamp=timestamp,
+            temporality=temporality,
+            description=description,
+            unit=unit,
+            # the querier resolves the metric type from this column, and only an
+            # ExponentialHistogram here routes the query to the sketch read
+            type_="ExponentialHistogram",
+            is_monotonic=False,
+            env=env,
+            resource_attrs=resource_attributes,
+            scope_attrs=scope_attributes,
+        )
+
+        self._exp_hist = MetricsExpHist(
+            metric_name=metric_name,
+            fingerprint=self._time_series.fingerprint,
+            timestamp=timestamp,
+            observations=observations,
+            temporality=temporality,
+            env=env,
+            flags=flags,
+        )
+
+
 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
@@ -686,6 +742,47 @@ class MetricsBufferSample(ABC):
         ]
 
 
+def insert_time_series_to_clickhouse(conn, time_series: list[MetricsTimeSeries]) -> None:
+    """
+    Insert one distributed_time_series_v4 registration row per (series, hour
+    bucket), unix_milli floored to the hour — the exporter's exact shape.
+    Readers floor lookup windows to these buckets: skipping per-bucket
+    re-registration or keeping raw mid-hour timestamps hides series in ways
+    production never sees.
+    """
+    time_series_map: dict[tuple[int, int], MetricsTimeSeries] = {}
+    for ts in time_series:
+        fp = int(ts.fingerprint)
+        hour_bucket = int(ts.unix_milli) // 3_600_000
+        if (fp, hour_bucket) not in time_series_map:
+            ts.unix_milli = np.int64(hour_bucket * 3_600_000)
+            time_series_map[(fp, hour_bucket)] = ts
+
+    if len(time_series_map) == 0:
+        return
+
+    conn.insert(
+        database="signoz_metrics",
+        table="distributed_time_series_v4",
+        column_names=[
+            "env",
+            "temporality",
+            "metric_name",
+            "description",
+            "unit",
+            "type",
+            "is_monotonic",
+            "fingerprint",
+            "unix_milli",
+            "labels",
+            "attrs",
+            "scope_attrs",
+            "resource_attrs",
+        ],
+        data=[ts.to_row() for ts in time_series_map.values()],
+    )
+
+
 def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
     """
     Insert metrics into ClickHouse tables.
@@ -697,39 +794,7 @@ def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
     Pure function so the seeder container can reuse the exact insert path
     used by the pytest fixture. `conn` is a clickhouse-connect Client.
     """
-    # One registration row per (series, hour bucket), unix_milli floored to
-    # the hour — the exporter's exact shape. Readers floor lookup windows to
-    # these buckets: skipping per-bucket re-registration or keeping raw
-    # mid-hour timestamps hides series in ways production never sees.
-    time_series_map: dict[tuple[int, int], MetricsTimeSeries] = {}
-    for metric in metrics:
-        fp = int(metric.time_series.fingerprint)
-        hour_bucket = int(metric.time_series.unix_milli) // 3_600_000
-        if (fp, hour_bucket) not in time_series_map:
-            metric.time_series.unix_milli = np.int64(hour_bucket * 3_600_000)
-            time_series_map[(fp, hour_bucket)] = metric.time_series
-
-    if len(time_series_map) > 0:
-        conn.insert(
-            database="signoz_metrics",
-            table="distributed_time_series_v4",
-            column_names=[
-                "env",
-                "temporality",
-                "metric_name",
-                "description",
-                "unit",
-                "type",
-                "is_monotonic",
-                "fingerprint",
-                "unix_milli",
-                "labels",
-                "attrs",
-                "scope_attrs",
-                "resource_attrs",
-            ],
-            data=[ts.to_row() for ts in time_series_map.values()],
-        )
+    insert_time_series_to_clickhouse(conn, [metric.time_series for metric in metrics])
 
     samples = [metric.sample for metric in metrics]
     if len(samples) > 0:
@@ -748,6 +813,15 @@ def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
             data=[sample.to_row() for sample in samples],
         )
 
+    insert_metrics_metadata_to_clickhouse(conn, metrics)
+
+
+def insert_metrics_metadata_to_clickhouse(conn, metrics: list) -> None:
+    """
+    Insert the distributed_metadata rows describing each metric's point, resource
+    and scope attributes. Accepts anything exposing `time_series`, `labels` and
+    `timestamp`.
+    """
     # (metric_name, attr_type, attr_name, attr_value) -> MetricsMetadata
     metadata_map: dict[tuple, MetricsMetadata] = {}
     for metric in metrics:
@@ -853,6 +927,61 @@ def insert_metrics(
     )
 
 
+def insert_exp_histogram_metrics_to_clickhouse(conn, metrics: list[ExpHistogramMetrics]) -> None:
+    """
+    Insert exponential histograms into ClickHouse tables.
+    Handles insertion into:
+    - distributed_time_series_v4 (time series metadata)
+    - distributed_exp_hist (per-point sketches)
+    - distributed_metadata (metric attribute metadata)
+    """
+    insert_time_series_to_clickhouse(conn, [metric.time_series for metric in metrics])
+
+    # `sketch` is AggregateFunction(quantilesDD(...), UInt64) — the state has to be
+    # folded server-side, it cannot be sent as a literal. The quantilesDDState
+    # parameters must match the column's exactly or the INSERT is rejected.
+    for metric in metrics:
+        hist = metric.exp_hist
+        conn.command(
+            "INSERT INTO signoz_metrics.distributed_exp_hist "
+            "(env, temporality, metric_name, fingerprint, unix_milli, count, sum, min, max, sketch, flags) "
+            "SELECT %(env)s, %(temporality)s, %(metric_name)s, %(fingerprint)s, %(unix_milli)s, "
+            "%(count)s, %(sum)s, %(min)s, %(max)s, "
+            "quantilesDDState(0.01, 0.5, 0.75, 0.9, 0.95, 0.99)(toUInt64(observation)), %(flags)s "
+            "FROM (SELECT arrayJoin(%(observations)s) AS observation)",
+            parameters={
+                "env": hist.env,
+                "temporality": hist.temporality,
+                "metric_name": hist.metric_name,
+                "fingerprint": int(hist.fingerprint),
+                "unix_milli": int(hist.unix_milli),
+                "count": int(hist.count),
+                "sum": float(hist.sum),
+                "min": float(hist.min),
+                "max": float(hist.max),
+                "observations": hist.observations,
+                "flags": int(hist.flags),
+            },
+        )
+
+    insert_metrics_metadata_to_clickhouse(conn, metrics)
+
+
+@pytest.fixture(name="insert_exp_histogram_metrics", scope="function")
+def insert_exp_histogram_metrics(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[Callable[[list[ExpHistogramMetrics]], None], Any]:
+    def _insert_exp_histogram_metrics(metrics: list[ExpHistogramMetrics]) -> None:
+        insert_exp_histogram_metrics_to_clickhouse(clickhouse.conn, metrics)
+
+    yield _insert_exp_histogram_metrics
+
+    truncate_metrics_tables(
+        clickhouse.conn,
+        clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"],
+    )
+
+
 def insert_reduced_metrics_to_clickhouse(
     conn,
     time_series: list[MetricsReducedTimeSeries],

tests/fixtures/querierai.py

@@ -1,5 +1,16 @@
 from datetime import datetime, timedelta
+from http import HTTPStatus
 
+from fixtures import types
+from fixtures.querier import (
+    Aggregation,
+    BuilderQuery,
+    OrderBy,
+    RequestType,
+    TelemetryFieldKey,
+    get_scalar_table_data,
+    make_query_request,
+)
 from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
 
 
@@ -32,10 +43,10 @@ def ai_trace(
     *,
     now: datetime,
     service: str,
-    user: str,
     in_tokens: int | None,
     out_tokens: int,
-    cost: float,
+    user: str = "user",
+    cost: float = 0.1,
     model: str = "gpt-4o-mini",
     environment: str = "production",
 ) -> list[Traces]:
@@ -74,6 +85,28 @@ def ai_trace(
     ]
 
 
+def tool_only_trace(*, now: datetime, service: str) -> list[Traces]:
+    """Root + one tool span: passes the gen_ai gate but has NO LLM span."""
+    trace_id = TraceIdGenerator.trace_id()
+    root_id = TraceIdGenerator.span_id()
+    resources = {"service.name": service}
+    return [
+        root_span(now=now, trace_id=trace_id, span_id=root_id, resources=resources, duration_s=2),
+        Traces(
+            timestamp=now - timedelta(seconds=4),
+            duration=timedelta(seconds=0.5),
+            trace_id=trace_id,
+            span_id=TraceIdGenerator.span_id(),
+            parent_span_id=root_id,
+            name="execute_tool",
+            kind=TracesKind.SPAN_KIND_INTERNAL,
+            status_code=TracesStatusCode.STATUS_CODE_OK,
+            resources=resources,
+            attributes={"gen_ai.tool.name": "get_weather", "gen_ai.tool.type": "function"},
+        ),
+    ]
+
+
 def ai_trace_mixed_spans(*, now: datetime, service: str, user: str) -> list[Traces]:
     """Root + LLM + tool + agent spans; only the LLM span carries gen_ai.request.model."""
     trace_id = TraceIdGenerator.trace_id()
@@ -116,3 +149,48 @@ def ai_trace_mixed_spans(*, now: datetime, service: str, user: str) -> list[Trac
         ),
         child("agent.step", TracesKind.SPAN_KIND_INTERNAL, {"gen_ai.agent.name": "chat-agent"}, 2),
     ]
+
+
+def ai_aggregation_query(
+    service: str,
+    expression: str,
+    *,
+    filter_extra: str = "",
+    group_by: list[TelemetryFieldKey] | None = None,
+    alias: str | None = None,
+    having: str | None = None,
+    order: list[OrderBy] | None = None,
+    limit: int | None = None,
+    step_interval: int | None = None,
+) -> dict:
+    filter_expression = f"service.name = '{service}'"
+    if filter_extra:
+        filter_expression += f" AND {filter_extra}"
+    return BuilderQuery(
+        signal="traces",
+        query_type="builder_ai_query",
+        name="A",
+        filter_expression=filter_expression,
+        aggregations=[Aggregation(expression=expression, alias=alias)],
+        group_by=group_by,
+        having_expression=having,
+        order=order,
+        limit=limit,
+        step_interval=step_interval,
+    ).to_dict()
+
+
+def scalar_value(signoz: types.SigNoz, token: str, start_ms: int, end_ms: int, service: str, expression: str, filter_extra: str = "") -> float:
+    """The single cell of a one-aggregation, ungrouped scalar query."""
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [ai_aggregation_query(service, expression, filter_extra=filter_extra)],
+        request_type=RequestType.SCALAR,
+    )
+    assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
+    data = get_scalar_table_data(response.json())
+    assert len(data) == 1, f"{expression}: expected one row, got {data}"
+    return float(data[0][-1])

tests/integration/tests/querierai/01_ai_traces.py

@@ -72,8 +72,8 @@ def test_ai_list_having_aggregate_filter(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
-    """Span + aggregate condition in one filter box splits into WHERE + HAVING; bare
-    and `trace.` spellings behave identically; an output-only aggregate is rejected."""
+    """One filter box splits into WHERE + HAVING; bare and `trace.` spellings behave
+    identically; an output-only aggregate is rejected."""
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     service = "ai-it-having"
 
@@ -322,9 +322,8 @@ def test_ai_list_nested_group_span_or_and_aggregate(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
-    """service.name = X AND (has_error = true OR gen_ai.request.model = 'gpt-4o') AND
-    total_tokens > 100: the nested OR group must not flatten, span predicates go to
-    WHERE, the aggregate to HAVING."""
+    """A nested (span OR span) group ANDed with an aggregate must not flatten: span
+    predicates go to WHERE, the aggregate to HAVING."""
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     service = "ai-it-nested"
 

tests/integration/tests/querierai/02_ai_aggregations.py

@@ -0,0 +1,502 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import (
+    Aggregation,
+    BuilderQuery,
+    OrderBy,
+    RequestType,
+    TelemetryFieldKey,
+    get_all_series,
+    get_scalar_columns,
+    get_scalar_table_data,
+    get_series_values,
+    make_query_request,
+)
+from fixtures.querierai import ai_aggregation_query, ai_trace, query_window, scalar_value, tool_only_trace
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
+
+
+def test_ai_scalar_trace_level_aggregations(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """Scalars over per-trace values, and the bare-key span domain through the same request type."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-scalar"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    def value(expression: str) -> float:
+        return scalar_value(signoz, token, start_ms, end_ms, service, expression)
+
+    assert value("avg(trace.output_tokens)") == pytest.approx(200)
+    assert value("count(trace.trace_id)") == 2
+    assert value("max(trace.total_tokens)") == pytest.approx(330)
+    assert value("p50(trace.output_tokens)") == pytest.approx(200)  # AggreFuncMap -> quantile(0.50)
+    # arithmetic inside one function and between functions
+    assert value("avg(trace.output_tokens + trace.input_tokens)") == pytest.approx(220)
+    assert value("sum(trace.output_tokens)/count(trace.trace_id)") == pytest.approx(200)
+    assert value("count()") == 2  # the two LLM spans; roots are not gen_ai
+    assert value("sum(gen_ai.usage.output_tokens)") == pytest.approx(400)
+
+    # multiple trace-level aggregations in one query -> one column per aggregation
+    multi = BuilderQuery(
+        signal="traces",
+        query_type="builder_ai_query",
+        name="A",
+        filter_expression=f"service.name = '{service}'",
+        aggregations=[Aggregation(expression="avg(trace.output_tokens)"), Aggregation(expression="count(trace.trace_id)")],
+    )
+    resp = make_query_request(signoz, token, start_ms, end_ms, [multi.to_dict()], request_type=RequestType.SCALAR)
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    data = get_scalar_table_data(resp.json())
+    assert len(data) == 1 and [float(v) for v in data[0]] == [pytest.approx(200), 2], data
+
+
+def test_ai_scalar_trace_level_filter_qualifies_traces(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """A trace-level condition qualifies whole traces before aggregation, on both domains."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-qualify"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    for expression in (
+        "sum(trace.output_tokens)",  # native trace-domain path
+        "sum(gen_ai.usage.output_tokens)",  # delegated span-domain path (__trace_scope)
+    ):
+        got = scalar_value(signoz, token, start_ms, end_ms, service, expression, filter_extra="trace.output_tokens > 100")
+        assert got == pytest.approx(300), expression
+
+    # the qualification also constrains delegated (span-domain) time series
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [
+            ai_aggregation_query(
+                service,
+                "sum(gen_ai.usage.output_tokens)",
+                filter_extra="trace.output_tokens > 100",
+                step_interval=60,
+            )
+        ],
+        request_type=RequestType.TIME_SERIES,
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    assert [v["value"] for v in get_series_values(resp.json(), "A")] == [pytest.approx(300)]
+
+
+def test_ai_scalar_group_by_model(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """Trace-level aggregation grouped by a span attribute."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-groupby"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [ai_aggregation_query(service, "avg(trace.output_tokens)", group_by=[TelemetryFieldKey(name="gen_ai.request.model")])],
+        request_type=RequestType.SCALAR,
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    data = get_scalar_table_data(resp.json())
+    by_model = {row[0]: float(row[-1]) for row in data}
+    assert by_model == {"gpt-4o": pytest.approx(200), "gpt-4o-mini": pytest.approx(50)}, data
+
+
+def test_ai_scalar_group_by_intrinsic_span_column(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """Grouping by an intrinsic must not alias the group column to the span column it reads
+    (`toString(name) AS name` is a cyclic alias ClickHouse rejects)."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-groupby-intrinsic"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=10, out_tokens=300) + tool_only_trace(now=now, service=service))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [
+            ai_aggregation_query(
+                service,
+                "count(trace.trace_id)",
+                group_by=[TelemetryFieldKey(name="name")],
+                order=[OrderBy(key=TelemetryFieldKey(name="name"), direction="asc")],
+            )
+        ],
+        request_type=RequestType.SCALAR,
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    columns = get_scalar_columns(resp.json())
+    assert columns[0]["name"] == "name", columns
+    data = get_scalar_table_data(resp.json())
+    # the root spans are gated out, so each trace groups under its gen_ai span name
+    assert [(row[0], int(row[-1])) for row in data] == [("chat gpt-4o-mini", 2), ("execute_tool", 1)], data
+
+
+def test_ai_timeseries_trace_level_aggregation(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-ts"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    # all spans fall in one step bucket
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [ai_aggregation_query(service, "avg(trace.output_tokens)", step_interval=60)],
+        request_type=RequestType.TIME_SERIES,
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    assert [v["value"] for v in get_series_values(resp.json(), "A")] == [pytest.approx(200)]
+
+
+def test_ai_timeseries_top_n_groups(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """A grouped, limited time series ranks groups on whole-window per-trace values in
+    the requested order."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-topn"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    def top_series(order: list[OrderBy] | None) -> dict:
+        resp = make_query_request(
+            signoz,
+            token,
+            start_ms,
+            end_ms,
+            [
+                ai_aggregation_query(
+                    service,
+                    "sum(trace.output_tokens)",
+                    group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
+                    alias="total_out",
+                    order=order,
+                    limit=1,
+                    step_interval=60,
+                )
+            ],
+            request_type=RequestType.TIME_SERIES,
+        )
+        assert resp.status_code == HTTPStatus.OK, resp.text
+        series = get_all_series(resp.json(), "A")
+        assert len(series) == 1, f"limit=1 must keep exactly one group, got {len(series)} series"
+        return series[0]
+
+    top = top_series(None)  # default ranking: first aggregation desc
+    assert top["labels"][0]["value"] == "gpt-4o", top["labels"]
+    assert [v["value"] for v in top["values"]] == [pytest.approx(400)]
+
+    bottom = top_series([OrderBy(key=TelemetryFieldKey(name="total_out"), direction="asc")])
+    assert bottom["labels"][0]["value"] == "gpt-4o-mini", bottom["labels"]
+    assert [v["value"] for v in bottom["values"]] == [pytest.approx(50)]
+
+
+def test_ai_scalar_group_order_limit(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """Scalar limit is a plain top-N over the grouped rows."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-scalar-limit"
+    insert_traces(
+        ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o")
+        + ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o")
+        + ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini")
+        + ai_trace(now=now, service=service, in_tokens=10, out_tokens=10, model="gpt-4")
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [
+            ai_aggregation_query(
+                service,
+                "sum(trace.output_tokens)",
+                group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
+                alias="total_out",
+                order=[OrderBy(key=TelemetryFieldKey(name="total_out"), direction="desc")],
+                limit=2,
+            )
+        ],
+        request_type=RequestType.SCALAR,
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    data = get_scalar_table_data(resp.json())
+    assert [(row[0], float(row[-1])) for row in data] == [("gpt-4o", pytest.approx(400)), ("gpt-4o-mini", pytest.approx(50))], data
+
+
+def test_ai_timeseries_span_time_bucketing(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """Per-trace values are clipped per (bucket, trace), so a trace spanning two buckets
+    contributes each call's tokens to its own bucket, not the total to both."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-buckets"
+
+    trace_id = TraceIdGenerator.trace_id()
+    root_id = TraceIdGenerator.span_id()
+    resources = {"service.name": service}
+
+    def llm(offset_s: float, out_tokens: int) -> Traces:
+        return Traces(
+            timestamp=now - timedelta(seconds=offset_s),
+            duration=timedelta(seconds=1),
+            trace_id=trace_id,
+            span_id=TraceIdGenerator.span_id(),
+            parent_span_id=root_id,
+            name="chat",
+            kind=TracesKind.SPAN_KIND_CLIENT,
+            status_code=TracesStatusCode.STATUS_CODE_OK,
+            resources=resources,
+            attributes={"gen_ai.request.model": "gpt-4o-mini", "gen_ai.usage.output_tokens": out_tokens},
+        )
+
+    root = Traces(
+        timestamp=now - timedelta(seconds=130),
+        duration=timedelta(seconds=130),
+        trace_id=trace_id,
+        span_id=root_id,
+        parent_span_id="",
+        name="POST /api/chat",
+        kind=TracesKind.SPAN_KIND_SERVER,
+        status_code=TracesStatusCode.STATUS_CODE_OK,
+        resources=resources,
+        attributes={"http.request.method": "POST"},
+    )
+    # two LLM calls two minutes apart
+    insert_traces([root, llm(124, 100), llm(4, 300)])
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [ai_aggregation_query(service, "avg(trace.output_tokens)", step_interval=60)],
+        request_type=RequestType.TIME_SERIES,
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+
+    series = get_all_series(resp.json(), "A")
+    assert len(series) == 1, series
+    assert sorted(v["value"] for v in series[0]["values"]) == [pytest.approx(100), pytest.approx(300)], series
+
+
+def test_ai_scalar_variables_in_trace_level_filter(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """Variables resolve inside trace-level conditions with span-filter semantics."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-vars"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+    query = ai_aggregation_query(service, "sum(trace.output_tokens)", filter_extra="trace.output_tokens > $threshold")
+
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [query],
+        request_type=RequestType.SCALAR,
+        variables={"threshold": {"type": "text", "value": 100}},
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    data = get_scalar_table_data(resp.json())
+    assert len(data) == 1 and float(data[0][-1]) == pytest.approx(300), data
+
+    # an unresolvable $var is a 400 today via aggregate validation
+    resp = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.SCALAR)
+    assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
+    # quotes in the message are JSON-escaped, so match the halves separately
+    assert "$threshold" in resp.text and "cannot be used in a trace-level filter" in resp.text, resp.text
+
+    # a dynamic variable resolved to __all__ drops the condition (both traces count)
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [query],
+        request_type=RequestType.SCALAR,
+        variables={"threshold": {"type": "dynamic", "value": "__all__"}},
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    data = get_scalar_table_data(resp.json())
+    assert len(data) == 1 and float(data[0][-1]) == pytest.approx(400), data
+
+
+def test_ai_scalar_tool_only_trace_null_semantics(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """A tool-only trace (in the gate, no LLM span) follows plain SQL NULL semantics."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-toolonly"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + tool_only_trace(now=now, service=service))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    def value(expression: str, filter_extra: str = "") -> float:
+        return scalar_value(signoz, token, start_ms, end_ms, service, expression, filter_extra)
+
+    assert value("count(trace.trace_id)") == 2, "tool-only trace is an AI trace and must be counted"
+    assert value("avg(trace.output_tokens)") == pytest.approx(100), "NULL tokens are skipped by avg"
+    assert value("avg(trace.tool_call_count)") == pytest.approx(0.5), "tool-only trace feeds tool aggregates (1 and 0 calls)"
+    assert value("count()") == 2, "span-level count sees the LLM and the tool span"
+
+    # filtering on LLM activity is explicit, not implicit
+    assert value("count(trace.trace_id)", filter_extra="trace.llm_call_count > 0") == 1
+
+
+def test_ai_scalar_having_on_aggregation(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """The outer having filters aggregation results per group (by alias)."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-having"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    resp = make_query_request(
+        signoz,
+        token,
+        start_ms,
+        end_ms,
+        [
+            ai_aggregation_query(
+                service,
+                "avg(trace.output_tokens)",
+                group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
+                alias="avg_out",
+                having="avg_out > 100",
+            )
+        ],
+        request_type=RequestType.SCALAR,
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    data = get_scalar_table_data(resp.json())
+    assert len(data) == 1 and data[0][0] == "gpt-4o", data
+
+
+def test_ai_aggregation_rejections(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-agg-reject"
+    insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100))
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = query_window(now)
+
+    def expect_bad_request(query: dict, message: str) -> None:
+        resp = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.SCALAR)
+        assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
+        assert message in resp.text, resp.text
+
+    # span-level and trace-level aggregations cannot be mixed in one query
+    mixed = BuilderQuery(
+        signal="traces",
+        query_type="builder_ai_query",
+        name="A",
+        filter_expression=f"service.name = '{service}'",
+        aggregations=[Aggregation(expression="avg(trace.output_tokens)"), Aggregation(expression="count()")],
+    )
+    expect_bad_request(mixed.to_dict(), "cannot be mixed")
+
+    expect_bad_request(
+        ai_aggregation_query(service, "avg(trace.output_tokens)", group_by=[TelemetryFieldKey(name="trace.llm_call_count")]),
+        "grouping by trace-level aggregate",
+    )
+
+    # a bare per-trace column would emit one row per trace instead of one aggregated row
+    expect_bad_request(ai_aggregation_query(service, "trace.output_tokens"), "must be inside an aggregation function")
+
+    # the rate interval divides the whole expression, so it may not carry a second aggregation
+    expect_bad_request(ai_aggregation_query(service, "rate(trace.trace_id) + avg(trace.output_tokens)"), "combines a rate with another aggregation")
+
+    # order-by is stopped earlier, by request validation
+    expect_bad_request(
+        ai_aggregation_query(service, "avg(trace.output_tokens)", order=[OrderBy(key=TelemetryFieldKey(name="trace.total_tokens"), direction="desc")]),
+        "invalid order by key",
+    )

tests/integration/tests/queriermetrics/14_exp_histogram.py

@@ -0,0 +1,172 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metrics import ExpHistogramMetrics
+from fixtures.querier import (
+    build_builder_query,
+    get_all_series,
+    get_all_warnings,
+    get_error_message,
+    get_series_values,
+    index_series_by_label,
+    make_query_request,
+)
+
+# quantilesDD carries 0.01 relative accuracy and the log-spaced observations put
+# neighbouring ranks ~1.25% apart, so a percentile can land a few percent off
+PERCENTILE_TOLERANCE = 0.05
+
+
+@pytest.mark.parametrize(
+    "space_aggregation, frontend_first, frontend_last, backend_first, backend_last",
+    [
+        ("p50", 118, 153, 711, 921),
+        ("p95", 1108, 1435, 6651, 8613),
+        ("p99", 1352, 1751, 8113, 10507),
+    ],
+)
+@pytest.mark.parametrize("time_aggregation", ["", "rate"])
+def test_exp_histogram_percentile_delta_grouped(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_exp_histogram_metrics: Callable[[list[ExpHistogramMetrics]], None],
+    time_aggregation: str,
+    space_aggregation: str,
+    frontend_first: float,
+    frontend_last: float,
+    backend_first: float,
+    backend_last: float,
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    start_ms = int((now - timedelta(minutes=65)).timestamp() * 1000)
+    end_ms = int(now.timestamp() * 1000)
+    metric_name = "test_exp_histogram_latency"
+
+    insert_exp_histogram_metrics(
+        [
+            ExpHistogramMetrics(
+                metric_name=metric_name,
+                # log-spaced latencies with a long tail, drifting ~30% higher across
+                # the hour so each point carries a distinct distribution
+                observations=[round(base * 1.0125**rank * (1 + minute / 200)) for rank in range(400)],
+                labels={"service.name": service},
+                timestamp=now - timedelta(minutes=60 - minute),
+                temporality="Delta",
+            )
+            for service, base in (("frontend", 10), ("backend", 60))
+            for minute in range(60)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    query = build_builder_query(
+        "A",
+        metric_name,
+        time_aggregation,
+        space_aggregation,
+        temporality="delta",
+        group_by=["service.name"],
+    )
+
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    data = response.json()
+    assert get_all_warnings(data) == [], f"unexpected warnings: {get_all_warnings(data)}"
+
+    series_by_service = index_series_by_label(get_all_series(data, "A"), "service.name")
+    assert set(series_by_service.keys()) == {"frontend", "backend"}, f"got series {set(series_by_service.keys())}"
+
+    for service, first, last in (
+        ("frontend", frontend_first, frontend_last),
+        ("backend", backend_first, backend_last),
+    ):
+        values = [point["value"] for point in sorted(series_by_service[service]["values"], key=lambda point: point["timestamp"])]
+        assert len(values) == 60, f"{service}: expected a point per minute, got {len(values)}"
+        assert values[0] == pytest.approx(first, rel=PERCENTILE_TOLERANCE), f"{service} {space_aggregation} at the oldest point: got {values[0]}, want ~{first}"
+        assert values[-1] == pytest.approx(last, rel=PERCENTILE_TOLERANCE), f"{service} {space_aggregation} at the newest point: got {values[-1]}, want ~{last}"
+        # every observation drifts up minute over minute, so the sketch must too
+        assert values == sorted(values), f"{service} {space_aggregation} is not non-decreasing: {values}"
+
+
+def test_exp_histogram_percentile_delta_merges_across_series(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_exp_histogram_metrics: Callable[[list[ExpHistogramMetrics]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    start_ms = int((now - timedelta(minutes=65)).timestamp() * 1000)
+    end_ms = int(now.timestamp() * 1000)
+    metric_name = "test_exp_histogram_latency_merged"
+
+    insert_exp_histogram_metrics(
+        [
+            ExpHistogramMetrics(
+                metric_name=metric_name,
+                observations=[round(base * 1.0125**rank * (1 + minute / 200)) for rank in range(400)],
+                labels={"service.name": service},
+                timestamp=now - timedelta(minutes=60 - minute),
+                temporality="Delta",
+            )
+            for service, base in (("frontend", 10), ("backend", 60))
+            for minute in range(60)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    query = build_builder_query("A", metric_name, "", "p95", temporality="delta")
+
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    data = response.json()
+    assert get_all_warnings(data) == [], f"unexpected warnings: {get_all_warnings(data)}"
+
+    # both services' sketches merge into one, so p95 sits well above the frontend's
+    # own p95 (~1108) and below the backend's (~6651)
+    values = [point["value"] for point in sorted(get_series_values(data, "A"), key=lambda point: point["timestamp"])]
+    assert len(values) == 60, f"expected a point per minute, got {len(values)}"
+    assert values[0] == pytest.approx(5188, rel=PERCENTILE_TOLERANCE), f"oldest point: got {values[0]}, want ~5188"
+    assert values[-1] == pytest.approx(6718, rel=PERCENTILE_TOLERANCE), f"newest point: got {values[-1]}, want ~6718"
+
+
+def test_exp_histogram_percentile_cumulative_is_rejected(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_exp_histogram_metrics: Callable[[list[ExpHistogramMetrics]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    start_ms = int((now - timedelta(minutes=65)).timestamp() * 1000)
+    end_ms = int(now.timestamp() * 1000)
+    metric_name = "test_exp_histogram_latency_cumulative"
+
+    insert_exp_histogram_metrics(
+        [
+            ExpHistogramMetrics(
+                metric_name=metric_name,
+                observations=[round(10 * 1.0125**rank * (1 + minute / 200)) for rank in range(400)],
+                labels={"service.name": "frontend"},
+                timestamp=now - timedelta(minutes=60 - minute),
+                temporality="Cumulative",
+            )
+            for minute in range(60)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    query = build_builder_query("A", metric_name, "", "p95", temporality="cumulative")
+
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+
+    # the request is well formed; it is the stored metric that cannot be read, so
+    # this is reported as unsupported rather than as bad input
+    assert response.status_code == HTTPStatus.NOT_IMPLEMENTED, response.text
+    assert "only `delta` exponential histograms are supported" in get_error_message(response.json()), response.text

tests/integration/tests/savedview/01_saved_view.py

@@ -592,6 +592,66 @@ def test_saved_view_lifecycle(
     assert response.status_code == HTTPStatus.NOT_FOUND
 
 
+def test_ai_observability_view_with_builder_ai_query_roundtrip(
+    signoz: SigNoz,
+    create_user_admin: Operation,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+):
+    """builder_ai_query implies the traces signal -- the spec is sent without
+    one and must read back with signal pinned to "traces"."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    headers = {"Authorization": f"Bearer {token}"}
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get(BASE_URL),
+        json={
+            "name": "ai-observability-overview",
+            "generateName": False,
+            "source": "ai_observability",
+            "schemaVersion": "v2",
+            "spec": {
+                "displayName": "ai-observability-overview",
+                "requestType": "scalar",
+                "queries": [{"type": "builder_ai_query", "spec": {"name": "A", "aggregations": [{"expression": "count()"}], "disabled": False, "legend": ""}}],
+                "selectedFields": [],
+                "panelType": "table",
+                "display": {"maxLines": 0, "fontSize": "", "format": "", "color": ""},
+            },
+        },
+        headers=headers,
+        timeout=5,
+    )
+    assert response.status_code == HTTPStatus.CREATED, response.text
+    view_id = response.json()["data"]["id"]
+
+    try:
+        response = requests.get(
+            signoz.self.host_configs["8080"].get(f"{BASE_URL}/{view_id}"),
+            headers=headers,
+            timeout=5,
+        )
+        assert response.status_code == HTTPStatus.OK, response.text
+        got = response.json()["data"]
+        assert got["source"] == "ai_observability"
+        assert got["spec"]["queries"][0]["type"] == "builder_ai_query"
+        assert got["spec"]["queries"][0]["spec"]["signal"] == "traces"
+
+        response = requests.get(
+            signoz.self.host_configs["8080"].get(BASE_URL),
+            params={"source": "ai_observability"},
+            headers=headers,
+            timeout=5,
+        )
+        assert response.status_code == HTTPStatus.OK, response.text
+        assert {v["name"] for v in response.json()["data"]} == {"ai-observability-overview"}
+    finally:
+        requests.delete(
+            signoz.self.host_configs["8080"].get(f"{BASE_URL}/{view_id}"),
+            headers=headers,
+            timeout=5,
+        )
+
+
 def test_empty_name_derives_a_slug_from_display_name(
     signoz: SigNoz,
     create_user_admin: Operation,  # pylint: disable=unused-argument