feat: add support for offset in all query range request types

#### Description

- `app-shell/sidenav` TC-03: dnd-kit stops every click for 50 ms after a
drop, so the Save click that follows the drag was sometimes lost and no
preference request went out. `saveShortcuts` now clicks again while the
dialog is still open.
- `messaging-queues/kafka` TC-01: the health check button is a `loading`
antd `Button` while its onboarding queries fetch, and a loading antd
button ignores clicks without being disabled. `openHealthCheck` now
clicks again until the dialog opens.
- Both tests came in with #13189 and have been failing or needing
retries in CI on unrelated PRs. Notes added to
`tests/e2e/docs/flake-notes.md`.

docs/api/openapi.yml

@@ -4027,7 +4027,9 @@ components:
             the query they draw.
         cursor:
           description: The opaque nextCursor value from a previous raw result, used
-            to fetch the next page.
+            to fetch the next page. Honored only by raw requests whose order starts
+            with timestamp, for logs followed by id in the same direction; every other
+            request ignores it.
           type: string
         disabled:
           description: When true, the result is excluded from the response; the query
@@ -4064,10 +4066,11 @@ components:
             are replaced by group-by label values.
           type: string
         limit:
-          description: The maximum number of rows for raw requests, or of series or
-            rows for grouped aggregations, keeping the top entries by order. When
-            0, aggregations are unlimited while raw requests ordered by timestamp
-            return a page of 10000 rows; follow nextCursor for the rest.
+          description: The maximum number of rows for raw and trace requests, or of
+            series or rows for grouped scalar and time_series requests, keeping the
+            top entries by order. When 0, scalar and time_series requests are unlimited,
+            raw requests ordered by timestamp return a page of 10000 rows to continue
+            with nextCursor, and other raw and trace requests return 100 rows.
           type: integer
         limitBy:
           allOf:
@@ -4079,8 +4082,10 @@ components:
             operators and selectedQueryName refer to it.
           type: string
         offset:
-          description: The number of rows to skip before returning results. Prefer
-            cursor for raw requests.
+          description: The number of rows for raw and trace requests, or of ordered
+            series or rows for scalar and time_series requests, to skip before the
+            page starts. On raw requests the skip counts from where a cursor resumes,
+            and the skipped rows are still read, so prefer cursor for paging.
           type: integer
         order:
           description: The sort order of the result. Order by a group-by field, an
@@ -10102,6 +10107,10 @@ components:
           description: The unique name of the formula, such as F1, used by selectedQueryName
             and by panels. Referencing a formula from another formula is not reliable.
           type: string
+        offset:
+          description: The number of ordered series or rows to skip before the page
+            starts.
+          type: integer
         order:
           description: The sort order of the formula result.
           items:
@@ -10127,7 +10136,9 @@ components:
             the query they draw.
         cursor:
           description: The opaque nextCursor value from a previous raw result, used
-            to fetch the next page.
+            to fetch the next page. Honored only by raw requests whose order starts
+            with timestamp, for logs followed by id in the same direction; every other
+            request ignores it.
           type: string
         disabled:
           description: When true, the result is excluded from the response; the query
@@ -10163,10 +10174,11 @@ components:
             are replaced by group-by label values.
           type: string
         limit:
-          description: The maximum number of rows for raw requests, or of series or
-            rows for grouped aggregations, keeping the top entries by order. When
-            0, aggregations are unlimited while raw requests ordered by timestamp
-            return a page of 10000 rows; follow nextCursor for the rest.
+          description: The maximum number of rows for raw and trace requests, or of
+            series or rows for grouped scalar and time_series requests, keeping the
+            top entries by order. When 0, scalar and time_series requests are unlimited,
+            raw requests ordered by timestamp return a page of 10000 rows to continue
+            with nextCursor, and other raw and trace requests return 100 rows.
           type: integer
         limitBy:
           allOf:
@@ -10178,8 +10190,10 @@ components:
             operators and selectedQueryName refer to it.
           type: string
         offset:
-          description: The number of rows to skip before returning results. Prefer
-            cursor for raw requests.
+          description: The number of rows for raw and trace requests, or of ordered
+            series or rows for scalar and time_series requests, to skip before the
+            page starts. On raw requests the skip counts from where a cursor resumes,
+            and the skipped rows are still read, so prefer cursor for paging.
           type: integer
         order:
           description: The sort order of the result. Order by a group-by field, an
@@ -10241,7 +10255,9 @@ components:
             the query they draw.
         cursor:
           description: The opaque nextCursor value from a previous raw result, used
-            to fetch the next page.
+            to fetch the next page. Honored only by raw requests whose order starts
+            with timestamp, for logs followed by id in the same direction; every other
+            request ignores it.
           type: string
         disabled:
           description: When true, the result is excluded from the response; the query
@@ -10278,10 +10294,11 @@ components:
             are replaced by group-by label values.
           type: string
         limit:
-          description: The maximum number of rows for raw requests, or of series or
-            rows for grouped aggregations, keeping the top entries by order. When
-            0, aggregations are unlimited while raw requests ordered by timestamp
-            return a page of 10000 rows; follow nextCursor for the rest.
+          description: The maximum number of rows for raw and trace requests, or of
+            series or rows for grouped scalar and time_series requests, keeping the
+            top entries by order. When 0, scalar and time_series requests are unlimited,
+            raw requests ordered by timestamp return a page of 10000 rows to continue
+            with nextCursor, and other raw and trace requests return 100 rows.
           type: integer
         limitBy:
           allOf:
@@ -10293,8 +10310,10 @@ components:
             operators and selectedQueryName refer to it.
           type: string
         offset:
-          description: The number of rows to skip before returning results. Prefer
-            cursor for raw requests.
+          description: The number of rows for raw and trace requests, or of ordered
+            series or rows for scalar and time_series requests, to skip before the
+            page starts. On raw requests the skip counts from where a cursor resumes,
+            and the skipped rows are still read, so prefer cursor for paging.
           type: integer
         order:
           description: The sort order of the result. Order by a group-by field, an
@@ -10356,7 +10375,9 @@ components:
             the query they draw.
         cursor:
           description: The opaque nextCursor value from a previous raw result, used
-            to fetch the next page.
+            to fetch the next page. Honored only by raw requests whose order starts
+            with timestamp, for logs followed by id in the same direction; every other
+            request ignores it.
           type: string
         disabled:
           description: When true, the result is excluded from the response; the query
@@ -10393,10 +10414,11 @@ components:
             are replaced by group-by label values.
           type: string
         limit:
-          description: The maximum number of rows for raw requests, or of series or
-            rows for grouped aggregations, keeping the top entries by order. When
-            0, aggregations are unlimited while raw requests ordered by timestamp
-            return a page of 10000 rows; follow nextCursor for the rest.
+          description: The maximum number of rows for raw and trace requests, or of
+            series or rows for grouped scalar and time_series requests, keeping the
+            top entries by order. When 0, scalar and time_series requests are unlimited,
+            raw requests ordered by timestamp return a page of 10000 rows to continue
+            with nextCursor, and other raw and trace requests return 100 rows.
           type: integer
         limitBy:
           allOf:
@@ -10408,8 +10430,10 @@ components:
             operators and selectedQueryName refer to it.
           type: string
         offset:
-          description: The number of rows to skip before returning results. Prefer
-            cursor for raw requests.
+          description: The number of rows for raw and trace requests, or of ordered
+            series or rows for scalar and time_series requests, to skip before the
+            page starts. On raw requests the skip counts from where a cursor resumes,
+            and the skipped rows are still read, so prefer cursor for paging.
           type: integer
         order:
           description: The sort order of the result. Order by a group-by field, an
@@ -10510,8 +10534,8 @@ components:
           description: The unique name of the trace operator query.
           type: string
         offset:
-          description: The number of raw results to skip before returning. Aggregation
-            requests ignore it.
+          description: The number of raw or trace rows, or of ordered series or rows
+            for time_series and scalar requests, to skip before the page starts.
           type: integer
         order:
           description: The sort order of the matched traces; only span_count and trace_duration

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

@@ -5273,7 +5273,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
 	bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
 	/**
 	 * @type string
-	 * @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
+	 * @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
 	 */
 	cursor?: string;
 	/**
@@ -5306,7 +5306,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
 	legend?: string;
 	/**
 	 * @type integer
-	 * @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
+	 * @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
 	 */
 	limit?: number;
 	/**
@@ -5320,7 +5320,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
 	name?: string;
 	/**
 	 * @type integer
-	 * @description The number of rows to skip before returning results. Prefer cursor for raw requests.
+	 * @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
 	 */
 	offset?: number;
 	/**
@@ -5665,7 +5665,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
 	/**
 	 * @type string
-	 * @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
+	 * @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
 	 */
 	cursor?: string;
 	/**
@@ -5698,7 +5698,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	legend?: string;
 	/**
 	 * @type integer
-	 * @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
+	 * @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
 	 */
 	limit?: number;
 	/**
@@ -5712,7 +5712,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	name?: string;
 	/**
 	 * @type integer
-	 * @description The number of rows to skip before returning results. Prefer cursor for raw requests.
+	 * @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
 	 */
 	offset?: number;
 	/**
@@ -5845,7 +5845,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
 	/**
 	 * @type string
-	 * @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
+	 * @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
 	 */
 	cursor?: string;
 	/**
@@ -5878,7 +5878,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	legend?: string;
 	/**
 	 * @type integer
-	 * @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
+	 * @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
 	 */
 	limit?: number;
 	/**
@@ -5892,7 +5892,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	name?: string;
 	/**
 	 * @type integer
-	 * @description The number of rows to skip before returning results. Prefer cursor for raw requests.
+	 * @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
 	 */
 	offset?: number;
 	/**
@@ -5944,7 +5944,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
 	/**
 	 * @type string
-	 * @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
+	 * @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
 	 */
 	cursor?: string;
 	/**
@@ -5977,7 +5977,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	legend?: string;
 	/**
 	 * @type integer
-	 * @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
+	 * @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
 	 */
 	limit?: number;
 	/**
@@ -5991,7 +5991,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
 	name?: string;
 	/**
 	 * @type integer
-	 * @description The number of rows to skip before returning results. Prefer cursor for raw requests.
+	 * @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
 	 */
 	offset?: number;
 	/**
@@ -7064,6 +7064,11 @@ export interface Querybuildertypesv5QueryBuilderFormulaDTO {
 	 * @description The unique name of the formula, such as F1, used by selectedQueryName and by panels. Referencing a formula from another formula is not reliable.
 	 */
 	name?: string;
+	/**
+	 * @type integer
+	 * @description The number of ordered series or rows to skip before the page starts.
+	 */
+	offset?: number;
 	/**
 	 * @type array,null
 	 * @description The sort order of the formula result.
@@ -7146,7 +7151,7 @@ export interface Querybuildertypesv5QueryBuilderTraceOperatorDTO {
 	name?: string;
 	/**
 	 * @type integer
-	 * @description The number of raw results to skip before returning. Aggregation requests ignore it.
+	 * @description The number of raw or trace rows, or of ordered series or rows for time_series and scalar requests, to skip before the page starts.
 	 */
 	offset?: number;
 	/**

frontend/src/components/InputWithLabel/InputWithLabel.tsx

@@ -18,6 +18,7 @@ function InputWithLabel({
 	className,
 	closeIcon,
 	disabled,
+	sanitize,
 }: {
 	label: string;
 	initialValue?: string | number | null;
@@ -29,14 +30,17 @@ function InputWithLabel({
 	className?: string;
 	closeIcon?: React.ReactNode;
 	disabled?: boolean;
+	/** Rewrites the typed value before it is shown and reported. */
+	sanitize?: (value: string) => string;
 }): JSX.Element {
 	const [inputValue, setInputValue] = useState<string>(
 		initialValue ? initialValue.toString() : '',
 	);
 
 	const handleChange = (e: React.ChangeEvent<HTMLInputElement>): void => {
-		setInputValue(e.target.value);
-		onChange?.(e.target.value);
+		const value = sanitize ? sanitize(e.target.value) : e.target.value;
+		setInputValue(value);
+		onChange?.(value);
 	};
 
 	return (
@@ -78,6 +82,7 @@ InputWithLabel.defaultProps = {
 	initialValue: undefined,
 	className: undefined,
 	closeIcon: undefined,
+	sanitize: undefined,
 };
 
 export default InputWithLabel;

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

@@ -26,6 +26,7 @@ import {
 	RAW_QUERY_FIELDS,
 	resolveQueryBuilderFields,
 } from '../../queryBuilderFields.utils';
+import { parseLimit, sanitizeLimit } from '../../utils';
 
 import BucketOptions from './BucketOptions/BucketOptions';
 import HavingFilter from './HavingFilter/HavingFilter';
@@ -278,7 +279,7 @@ function QueryAddOns({
 
 	const handleChangeLimit = useCallback(
 		(value: string) => {
-			handleChangeQueryData('limit', Number(value) || null);
+			handleChangeQueryData('limit', parseLimit(value));
 		},
 		[handleChangeQueryData],
 	);
@@ -374,6 +375,7 @@ function QueryAddOns({
 									label="Limit"
 									type="number"
 									onChange={handleChangeLimit}
+									sanitize={sanitizeLimit}
 									initialValue={query?.limit ?? undefined}
 									placeholder={placeholderOf(QueryBuilderField.Limit)}
 									onClose={(): void => handleRemoveView(QueryBuilderField.Limit)}

frontend/src/components/QueryBuilderV2/__tests__/QueryAddOns.test.tsx

@@ -169,8 +169,15 @@ describe('QueryAddOns', () => {
 		expect(screen.getByTestId('limit-content')).toBeInTheDocument();
 		expect(input.value).toBe('5');
 
-		fireEvent.change(input, { target: { value: '10' } });
-		expect(mockHandleChangeQueryData).toHaveBeenCalledWith('limit', 10);
+		const user = userEvent.setup();
+		await user.clear(input);
+		await user.type(input, '10');
+		expect(mockHandleChangeQueryData).toHaveBeenLastCalledWith('limit', 10);
+
+		await user.clear(input);
+		await user.type(input, '-5');
+		expect(input.value).toBe('5');
+		expect(mockHandleChangeQueryData).toHaveBeenLastCalledWith('limit', 5);
 	});
 
 	it.each([

frontend/src/components/QueryBuilderV2/__tests__/utils.test.ts

@@ -12,10 +12,62 @@ import {
 	convertExpressionToFilters,
 	convertFiltersToExpression,
 	convertFiltersToExpressionWithExistingQuery,
+	convertHavingToExpression,
 	formatValueForExpression,
+	parseLimit,
 	removeKeysFromExpression,
+	sanitizeLimit,
 } from '../utils';
 
+describe('convertHavingToExpression', () => {
+	it.each([
+		['COUNT()', 'count() > 5'],
+		['COUNT_DISTINCT(service.name)', 'count_distinct(service.name) > 5'],
+		['P99(duration_nano)', 'p99(duration_nano) > 5'],
+		['SUM(http.Request.Size)', 'sum(http.Request.Size) > 5'],
+		['count()', 'count() > 5'],
+	])('references %s by its lower-case function', (columnName, expected) => {
+		expect(
+			convertHavingToExpression([{ columnName, op: '>', value: 5 }]),
+		).toStrictEqual({ expression: expected });
+	});
+
+	it('joins several conditions and formats list values', () => {
+		expect(
+			convertHavingToExpression([
+				{ columnName: 'COUNT()', op: '>', value: 5 },
+				{ columnName: 'COUNT()', op: 'NOT IN', value: [1, 2] },
+			]),
+		).toStrictEqual({ expression: 'count() > 5 AND count() NOT IN [1, 2]' });
+	});
+});
+
+describe('sanitizeLimit', () => {
+	it.each([
+		['10', '10'],
+		['-10', '10'],
+		['2.5', '25'],
+		['1e3', '13'],
+		['', ''],
+	])('sanitizes %j to %j', (value, expected) => {
+		expect(sanitizeLimit(value)).toBe(expected);
+	});
+});
+
+describe('parseLimit', () => {
+	it.each([
+		['10', 10],
+		['1', 1],
+		['', null],
+		['0', null],
+		['-5', null],
+		['2.5', null],
+		['abc', null],
+	])('parses %j as %j', (value, expected) => {
+		expect(parseLimit(value)).toBe(expected);
+	});
+});
+
 describe('convertFiltersToExpression', () => {
 	beforeEach(() => {
 		jest.clearAllMocks();

frontend/src/components/QueryBuilderV2/utils.ts

@@ -840,6 +840,16 @@ export const appendAndClause = (
 	return hasTopLevelOr ? `(${base}) AND ${clause}` : `${base} AND ${clause}`;
 };
 
+/** Strips everything but digits, so a limit input can't hold a sign or decimal. */
+export const sanitizeLimit = (value: string): string =>
+	value.replace(/\D/g, '');
+
+/** A query or formula limit: a positive integer, otherwise none. */
+export const parseLimit = (value: string): number | null => {
+	const limit = Number(value);
+	return Number.isInteger(limit) && limit > 0 ? limit : null;
+};
+
 /**
  * Convert old having format to new having format
  * @param having - Array of old having objects with columnName, op, and value
@@ -871,7 +881,14 @@ export const convertHavingToExpression = (
 				formattedValue = String(value);
 			}
 
-			return `${columnName} ${op} ${formattedValue}`;
+			// The old having filter upper-cased the function (`COUNT()`); v5 only
+			// resolves the aggregation's own lower-case expression.
+			const reference = columnName.replace(
+				/^\s*([A-Za-z_]\w*)\(/,
+				(_, fn: string) => `${fn.toLowerCase()}(`,
+			);
+
+			return `${reference} ${op} ${formattedValue}`;
 		})
 		.filter((expression) => expression !== ''); // Remove empty expressions
 

frontend/src/container/AIAssistant/components/ActionsSection/ActionsSection.tsx

@@ -26,6 +26,7 @@ import {
 } from 'api/ai-assistant/chat';
 import ROUTES from 'constants/routes';
 import { QueryParams } from 'constants/query';
+import { buildExplorerNavigationUrl } from 'utils/buildExplorerNavigationUrl';
 import { openInNewTab } from 'utils/navigation';
 import {
 	ArchiveRestore,
@@ -47,6 +48,10 @@ import { AIAssistantEvents, SuggestedPromptCategory } from '../../events';
 import { useAIAssistantAnalyticsContext } from '../../hooks/useAIAssistantAnalyticsContext';
 import { useAIAssistantStore } from '../../store/useAIAssistantStore';
 
+import {
+	getPanelTypeForRequestType,
+	requestTypeFromActionQuery,
+} from './utils/getPanelTypeForRequestType';
 import { openSavedViewByKey } from './utils/openSavedView';
 import {
 	isSavedViewOpenAction,
@@ -333,6 +338,11 @@ function applyFilter(action: MessageActionDTO, deps: ApplyFilterDeps): void {
 		);
 		return;
 	}
+	// Read before toUrlCompositeQuery, which drops the requestType envelope.
+	const panelType = getPanelTypeForRequestType(
+		requestTypeFromActionQuery(action.query),
+		action.signal,
+	);
 	const normalized = withDerivedFilterExpressions(urlQuery as unknown as Query);
 	// eslint-disable-next-line no-console
 	console.log('[apply_filter] normalized', normalized);
@@ -342,7 +352,9 @@ function applyFilter(action: MessageActionDTO, deps: ApplyFilterDeps): void {
 		normalized.builder.queryData.forEach((q, i) => {
 			deps.handleSetQueryData(i, q);
 		});
-		deps.redirectWithQueryBuilderData(normalized);
+		deps.redirectWithQueryBuilderData(normalized, {
+			[QueryParams.panelTypes]: panelType,
+		});
 		return;
 	}
 	const base = explorerRouteForSignal(action.signal);
@@ -353,8 +365,10 @@ function applyFilter(action: MessageActionDTO, deps: ApplyFilterDeps): void {
 	}
 	// eslint-disable-next-line no-console
 	console.log('[apply_filter] off-page → navigate', base);
-	const encoded = encodeURIComponent(JSON.stringify(normalized));
-	navigate(`${base}?${QueryParams.compositeQuery}=${encoded}`);
+	const url = buildExplorerNavigationUrl(base, normalized, {
+		[QueryParams.panelTypes]: panelType,
+	});
+	navigate(url);
 }
 
 /** Picks the right rollback API call for a given action kind. */
@@ -613,6 +627,7 @@ export default function ActionsSection({
 							disabled={isSuccess}
 							loading={isLoading}
 							aria-label={action.label}
+							testId={`ai-action-chip-${action.kind}`}
 							prefix={icon}
 						>
 							<span className={styles.chipLabel}>{action.label}</span>

frontend/src/container/AIAssistant/components/ActionsSection/__tests__/ActionsSection.test.tsx

@@ -1,7 +1,16 @@
 import { undoExecution } from 'api/ai-assistant/chat';
 import type { MessageActionDTO } from 'api/ai-assistant/sigNozAIAssistantAPI.schemas';
-import { MessageActionKindDTO } from 'api/ai-assistant/sigNozAIAssistantAPI.schemas';
+import {
+	ApplyFilterSignalDTO,
+	MessageActionKindDTO,
+} from 'api/ai-assistant/sigNozAIAssistantAPI.schemas';
+import { Querybuildertypesv5RequestTypeDTO } from 'api/generated/services/sigNoz.schemas';
+import { QueryParams } from 'constants/query';
+import { PANEL_TYPES } from 'constants/queryBuilder';
+import ROUTES from 'constants/routes';
+import { getCurrentLocation } from 'lib/router/navigation';
 import { render, screen, userEvent, waitFor } from 'tests/test-utils';
+import { QueryBuilderContextType } from 'types/common/queryBuilder';
 
 import ActionsSection from '../ActionsSection';
 
@@ -28,6 +37,37 @@ function renderUndoChip(): HTMLElement {
 	return screen.getByRole('button', { name: 'Undo change' });
 }
 
+// Labels differ per test: auto-applied filters are deduped module-wide by label.
+function countByServiceAction(label: string): MessageActionDTO {
+	return {
+		kind: MessageActionKindDTO.apply_filter,
+		label,
+		signal: ApplyFilterSignalDTO.logs,
+		query: {
+			requestType: Querybuildertypesv5RequestTypeDTO.scalar,
+			compositeQuery: {
+				queries: [
+					{
+						type: 'builder_query',
+						spec: {
+							name: 'A',
+							signal: 'logs',
+							aggregations: [{ expression: 'count()' }],
+							groupBy: [{ name: 'service.name' }],
+						},
+					},
+				],
+			},
+		},
+	};
+}
+
+function currentPanelTypesParam(): string | null {
+	return new URLSearchParams(getCurrentLocation().search).get(
+		QueryParams.panelTypes,
+	);
+}
+
 describe('ActionsSection rollback chip', () => {
 	beforeEach(() => {
 		mockUndo.mockReset();
@@ -81,3 +121,52 @@ describe('ActionsSection rollback chip', () => {
 		).resolves.toBeInTheDocument();
 	});
 });
+
+describe('ActionsSection apply_filter', () => {
+	it('opens the explorer in the Table view when the chip is clicked from another page', async () => {
+		const user = userEvent.setup({ pointerEventsCheck: 0 });
+		render(
+			<ActionsSection
+				actions={[countByServiceAction('Open in Logs Explorer')]}
+				messageId="message-1"
+			/>,
+			undefined,
+			{ initialRoute: ROUTES.HOME },
+		);
+
+		await user.click(screen.getByTestId('ai-action-chip-apply_filter'));
+
+		await waitFor(() =>
+			expect(getCurrentLocation().pathname).toBe(ROUTES.LOGS_EXPLORER),
+		);
+		expect(currentPanelTypesParam()).toBe(JSON.stringify(PANEL_TYPES.TABLE));
+	});
+
+	it('switches the open explorer to the Table view when applied on it', async () => {
+		const redirectWithQueryBuilderData = jest.fn() as jest.MockedFunction<
+			QueryBuilderContextType['redirectWithQueryBuilderData']
+		>;
+		const handleSetQueryData = jest.fn() as jest.MockedFunction<
+			QueryBuilderContextType['handleSetQueryData']
+		>;
+		render(
+			<ActionsSection
+				actions={[countByServiceAction('Apply on Logs Explorer')]}
+				messageId="message-2"
+			/>,
+			undefined,
+			{
+				initialRoute: ROUTES.LOGS_EXPLORER,
+				queryBuilderOverrides: { redirectWithQueryBuilderData, handleSetQueryData },
+			},
+		);
+
+		await waitFor(() =>
+			expect(redirectWithQueryBuilderData).toHaveBeenCalledTimes(1),
+		);
+		expect(redirectWithQueryBuilderData).toHaveBeenCalledWith(
+			expect.objectContaining({ builder: expect.anything() }),
+			{ [QueryParams.panelTypes]: PANEL_TYPES.TABLE },
+		);
+	});
+});

frontend/src/container/AIAssistant/components/ActionsSection/utils/__tests__/getPanelTypeForRequestType.test.ts

@@ -0,0 +1,106 @@
+import { ApplyFilterSignalDTO } from 'api/ai-assistant/sigNozAIAssistantAPI.schemas';
+import { Querybuildertypesv5RequestTypeDTO } from 'api/generated/services/sigNoz.schemas';
+import { PANEL_TYPES } from 'constants/queryBuilder';
+
+import {
+	getPanelTypeForRequestType,
+	requestTypeFromActionQuery,
+} from '../getPanelTypeForRequestType';
+
+describe('getPanelTypeForRequestType', () => {
+	it.each([
+		[
+			Querybuildertypesv5RequestTypeDTO.scalar,
+			ApplyFilterSignalDTO.logs,
+			PANEL_TYPES.TABLE,
+		],
+		[
+			Querybuildertypesv5RequestTypeDTO.time_series,
+			ApplyFilterSignalDTO.logs,
+			PANEL_TYPES.TIME_SERIES,
+		],
+		[
+			Querybuildertypesv5RequestTypeDTO.raw,
+			ApplyFilterSignalDTO.logs,
+			PANEL_TYPES.LIST,
+		],
+		[
+			Querybuildertypesv5RequestTypeDTO.trace,
+			ApplyFilterSignalDTO.traces,
+			PANEL_TYPES.TRACE,
+		],
+		[
+			Querybuildertypesv5RequestTypeDTO.trace,
+			ApplyFilterSignalDTO.logs,
+			PANEL_TYPES.LIST,
+		],
+		[
+			Querybuildertypesv5RequestTypeDTO.trace,
+			ApplyFilterSignalDTO.metrics,
+			PANEL_TYPES.LIST,
+		],
+	])('maps %s on the %s signal to the %s view', (requestType, signal, view) => {
+		expect(getPanelTypeForRequestType(requestType, signal)).toBe(view);
+	});
+
+	it.each([
+		undefined,
+		null,
+		'',
+		Querybuildertypesv5RequestTypeDTO.raw_stream,
+		Querybuildertypesv5RequestTypeDTO.heatmap,
+		'distribution',
+		'nonsense',
+		'toString',
+		'constructor',
+		42,
+		{},
+	])(
+		'falls back to the List view for unmapped or invalid requestType (%p)',
+		(value) => {
+			expect(getPanelTypeForRequestType(value, ApplyFilterSignalDTO.traces)).toBe(
+				PANEL_TYPES.LIST,
+			);
+		},
+	);
+});
+
+describe('requestTypeFromActionQuery', () => {
+	it('reads the requestType envelope field and returns undefined when absent', () => {
+		expect(
+			requestTypeFromActionQuery({
+				requestType: 'scalar',
+				schemaVersion: 'v5',
+				compositeQuery: { queries: [] },
+			}),
+		).toBe('scalar');
+		expect(requestTypeFromActionQuery({})).toBeUndefined();
+		expect(requestTypeFromActionQuery(null)).toBeUndefined();
+		expect(requestTypeFromActionQuery(undefined)).toBeUndefined();
+	});
+
+	it('maps a grouped count() payload to the Table view', () => {
+		const query = {
+			requestType: 'scalar',
+			schemaVersion: 'v5',
+			compositeQuery: {
+				queries: [
+					{
+						type: 'builder_query',
+						spec: {
+							signal: 'logs',
+							aggregations: [{ expression: 'count()' }],
+							groupBy: [{ name: 'service.name' }],
+						},
+					},
+				],
+			},
+		};
+		expect(
+			getPanelTypeForRequestType(
+				requestTypeFromActionQuery(query),
+				ApplyFilterSignalDTO.logs,
+			),
+		).toBe(PANEL_TYPES.TABLE);
+	});
+});

frontend/src/container/AIAssistant/components/ActionsSection/utils/getPanelTypeForRequestType.ts

@@ -0,0 +1,30 @@
+import {
+	ApplyFilterSignalDTO,
+	type MessageActionDTOQuery,
+} from 'api/ai-assistant/sigNozAIAssistantAPI.schemas';
+import { Querybuildertypesv5RequestTypeDTO } from 'api/generated/services/sigNoz.schemas';
+import { PANEL_TYPES } from 'constants/queryBuilder';
+
+export function getPanelTypeForRequestType(
+	requestType: unknown,
+	signal: ApplyFilterSignalDTO,
+): PANEL_TYPES {
+	switch (requestType) {
+		case Querybuildertypesv5RequestTypeDTO.scalar:
+			return PANEL_TYPES.TABLE;
+		case Querybuildertypesv5RequestTypeDTO.time_series:
+			return PANEL_TYPES.TIME_SERIES;
+		case Querybuildertypesv5RequestTypeDTO.trace:
+			return signal === ApplyFilterSignalDTO.traces
+				? PANEL_TYPES.TRACE
+				: PANEL_TYPES.LIST;
+		default:
+			return PANEL_TYPES.LIST;
+	}
+}
+
+export function requestTypeFromActionQuery(
+	query: MessageActionDTOQuery | undefined,
+): unknown {
+	return query?.requestType;
+}

frontend/src/container/QueryBuilder/components/Formula/Formula.tsx

@@ -2,6 +2,7 @@ import { ChangeEvent, useCallback, useMemo, useState } from 'react';
 import { Col, Input, Row, Select } from 'antd';
 import InputWithLabel from 'components/InputWithLabel/InputWithLabel';
 import BucketOptions from 'components/QueryBuilderV2/QueryV2/QueryAddOns/BucketOptions/BucketOptions';
+import { parseLimit, sanitizeLimit } from 'components/QueryBuilderV2/utils';
 import { LEGEND } from 'constants/global';
 import { PANEL_TYPES } from 'constants/queryBuilder';
 // ** Hooks
@@ -171,7 +172,8 @@ export function Formula({
 								<InputWithLabel
 									label="Limit"
 									type="number"
-									onChange={(value): void => handleChangeLimit(Number(value))}
+									onChange={(value): void => handleChangeLimit(parseLimit(value))}
+									sanitize={sanitizeLimit}
 									initialValue={formula?.limit ?? undefined}
 									placeholder="Enter limit"
 								/>

frontend/src/providers/QueryBuilder.tsx

@@ -475,12 +475,17 @@ export function QueryBuilderProvider({
 	const createNewBuilderQuery = useCallback(
 		(queries: IBuilderQuery[]): IBuilderQuery => {
 			const existNames = queries.map((item) => item.queryName);
+			// Hosts without a fixed data source (the panel editor) continue the
+			// previous query's signal rather than resetting to metrics.
+			const previous = queries[queries.length - 1];
 			const initialBuilderQuery =
-				initialQueryBuilderFormValuesMap[initialDataSource || DataSource.METRICS];
+				initialQueryBuilderFormValuesMap[
+					initialDataSource || previous?.dataSource || DataSource.METRICS
+				];
 
 			const newQuery: IBuilderQuery = {
 				...initialBuilderQuery,
-				source: queries?.[0]?.source || '',
+				source: previous?.source || '',
 				builderQueryType: queries?.[0]?.builderQueryType,
 				queryName: createNewBuilderItemName({ existNames, sourceNames: alphabet }),
 				expression: createNewBuilderItemName({

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

@@ -0,0 +1,104 @@
+import {
+	initialQueriesMap,
+	initialQueryBuilderFormValuesMap,
+	PANEL_TYPES,
+} from 'constants/queryBuilder';
+import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
+import { act, AllTheProviders, renderHook } from 'tests/test-utils';
+import { Query } from 'types/api/queryBuilder/queryBuilderData';
+import { DataSource } from 'types/common/queryBuilder';
+
+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 data source', () => {
+	it.each([DataSource.LOGS, DataSource.TRACES, DataSource.METRICS])(
+		'continues the previous %s query when the host fixes no data source',
+		(dataSource) => {
+			const { result } = renderQueryBuilder(initialQueriesMap[dataSource]);
+
+			act(() => {
+				result.current.addNewBuilderQuery();
+			});
+
+			const [, added] = result.current.currentQuery.builder.queryData;
+			expect(added.dataSource).toBe(dataSource);
+			expect(added.aggregations).toStrictEqual(
+				initialQueryBuilderFormValuesMap[dataSource].aggregations,
+			);
+		},
+	);
+
+	it('follows the last query when queries mix signals', () => {
+		const { result } = renderQueryBuilder(initialQueriesMap.logs);
+		act(() => {
+			result.current.addNewBuilderQuery();
+		});
+		act(() => {
+			result.current.handleSetQueryData(1, {
+				...initialQueryBuilderFormValuesMap.traces,
+				queryName: 'B',
+				expression: 'B',
+			});
+		});
+
+		act(() => {
+			result.current.addNewBuilderQuery();
+		});
+
+		expect(result.current.currentQuery.builder.queryData[2].dataSource).toBe(
+			DataSource.TRACES,
+		);
+	});
+
+	it('takes the source from the last query, not the first', () => {
+		const [metricsQuery] = initialQueriesMap.metrics.builder.queryData;
+		const { result } = renderQueryBuilder({
+			...initialQueriesMap.metrics,
+			builder: {
+				...initialQueriesMap.metrics.builder,
+				queryData: [
+					{ ...metricsQuery, source: '' },
+					{ ...metricsQuery, queryName: 'B', expression: 'B', source: 'meter' },
+				],
+			},
+		});
+
+		act(() => {
+			result.current.addNewBuilderQuery();
+		});
+
+		expect(result.current.currentQuery.builder.queryData[2]).toMatchObject({
+			dataSource: DataSource.METRICS,
+			source: 'meter',
+		});
+	});
+
+	it('uses the host’s fixed data source over the previous query', () => {
+		const { result } = renderQueryBuilder(initialQueriesMap.metrics);
+		act(() => {
+			result.current.handleSetConfig(PANEL_TYPES.TIME_SERIES, DataSource.LOGS);
+		});
+
+		act(() => {
+			result.current.addNewBuilderQuery();
+		});
+
+		expect(result.current.currentQuery.builder.queryData[1].dataSource).toBe(
+			DataSource.LOGS,
+		);
+	});
+});

frontend/src/utils/__tests__/buildExplorerNavigationUrl.test.ts

@@ -1,6 +1,7 @@
 import ROUTES from 'constants/routes';
 import { QueryParams } from 'constants/query';
-import { PANEL_TYPES } from 'constants/queryBuilder';
+import { initialQueriesMap, PANEL_TYPES } from 'constants/queryBuilder';
+import { DataSource } from 'types/common/queryBuilder';
 
 import { buildExplorerNavigationUrl } from '../buildExplorerNavigationUrl';
 
@@ -19,4 +20,14 @@ describe('buildExplorerNavigationUrl', () => {
 		expect(url).toContain(`${QueryParams.compositeQuery}=`);
 		expect(url).toContain(`${QueryParams.viewKey}=`);
 	});
+
+	it('JSON-encodes panelTypes so the explorer opens the right view', () => {
+		const url = buildExplorerNavigationUrl(
+			ROUTES.LOGS_EXPLORER,
+			initialQueriesMap[DataSource.LOGS],
+			{ [QueryParams.panelTypes]: PANEL_TYPES.TABLE },
+		);
+
+		expect(url).toContain(`${QueryParams.panelTypes}=%22table%22`);
+	});
 });

pkg/modules/llmpricingrule/impllmpricingrule/module.go

@@ -4,9 +4,13 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log/slog"
+	"maps"
+	"sync"
 	"time"
 
 	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/modules/llmpricingrule"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/query-service/agentConf"
@@ -22,12 +26,14 @@ import (
 const unmappedModelsLookback = time.Hour
 
 type module struct {
-	store   llmpricingruletypes.Store
-	querier querier.Querier
+	store                  llmpricingruletypes.Store
+	querier                querier.Querier
+	telemetryMetadataStore telemetrytypes.MetadataStore
+	logger                 *slog.Logger
 }
 
-func NewModule(store llmpricingruletypes.Store, querier querier.Querier) llmpricingrule.Module {
-	return &module{store: store, querier: querier}
+func NewModule(store llmpricingruletypes.Store, querier querier.Querier, telemetryMetadataStore telemetrytypes.MetadataStore, providerSettings factory.ProviderSettings) llmpricingrule.Module {
+	return &module{store: store, querier: querier, telemetryMetadataStore: telemetryMetadataStore, logger: providerSettings.Logger}
 }
 
 func (module *module) List(ctx context.Context, orgID valuer.UUID, offset, limit int, search string, isOverride *bool) ([]*llmpricingruletypes.LLMPricingRule, int, error) {
@@ -51,13 +57,47 @@ func (module *module) ListUnmappedModels(ctx context.Context, orgID valuer.UUID)
 		return nil, err
 	}
 
-	unmapped := make([]*llmpricingruletypes.UnmappedModel, 0, len(models))
-	for _, m := range models {
-		if !llmpricingruletypes.ModelMatchesAnyRule(m.ModelName, rules) {
-			unmapped = append(unmapped, m)
+	return llmpricingruletypes.UnmappedModels(models, rules), nil
+}
+
+// Collect keeps the rule counts even when ClickHouse is unreachable: the
+// counts that depend on it are skipped so the reporter retains their last value.
+// Models come from the traces tag table rather than a span scan, so the
+// unmapped count covers the metadata lookback instead of the last hour.
+func (module *module) Collect(ctx context.Context, orgID valuer.UUID) (map[string]any, error) {
+	rules, err := module.listAllRules(ctx, orgID)
+	if err != nil {
+		return nil, err
+	}
+	stats := llmpricingruletypes.NewStatsFromRules(rules)
+
+	selectors := aiobservabilitytypes.StatsFieldValueSelectors(time.Now())
+	results := make([]*telemetrytypes.TelemetryFieldValues, len(selectors))
+	var lookups sync.WaitGroup
+	for i, selector := range selectors {
+		lookups.Go(func() {
+			fieldValues, _, err := module.telemetryMetadataStore.GetAllValues(ctx, orgID, selector)
+			if err != nil {
+				module.logger.DebugContext(ctx, "failed to collect gen_ai attribute stats", slog.String("key", selector.Name), errors.Attr(err))
+				return
+			}
+			results[i] = fieldValues
+		})
+	}
+	lookups.Wait()
+
+	values := make(map[string]*telemetrytypes.TelemetryFieldValues, len(selectors))
+	for i, selector := range selectors {
+		if results[i] != nil {
+			values[selector.Name] = results[i]
 		}
 	}
-	return unmapped, nil
+	maps.Copy(stats, aiobservabilitytypes.NewStatsFromFieldValues(values))
+	if models := values[aiobservabilitytypes.GenAIRequestModel]; models != nil {
+		maps.Copy(stats, llmpricingruletypes.NewStatsFromModels(models.StringValues, rules))
+	}
+
+	return stats, nil
 }
 
 // CreateOrUpdate saves a batch of pricing rules. isOverride decides how a rule

pkg/modules/llmpricingrule/llmpricingrule.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 
 	"github.com/SigNoz/signoz/pkg/query-service/agentConf"
+	"github.com/SigNoz/signoz/pkg/statsreporter"
 	"github.com/SigNoz/signoz/pkg/types/llmpricingruletypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
@@ -13,6 +14,8 @@ type Module interface {
 	// Since this module interacts with OpAMP, it must implement the AgentFeature interface.
 	agentConf.AgentFeature
 
+	statsreporter.StatsCollector
+
 	List(ctx context.Context, orgID valuer.UUID, offset, limit int, search string, isOverride *bool) ([]*llmpricingruletypes.LLMPricingRule, int, error)
 	Get(ctx context.Context, orgID valuer.UUID, id valuer.UUID) (*llmpricingruletypes.LLMPricingRule, error)
 	CreateOrUpdate(ctx context.Context, orgID valuer.UUID, userEmail string, rules []*llmpricingruletypes.UpdatableLLMPricingRule) (err error)

pkg/querier/postprocess.go

@@ -182,7 +182,7 @@ func postProcessBuilderQuery[T any](
 	req *qbtypes.QueryRangeRequest,
 ) *qbtypes.Result {
 
-	result = q.applySeriesLimit(result, query.Limit, query.Order)
+	result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
 
 	// Apply functions
 	if len(query.Functions) > 0 {
@@ -234,7 +234,7 @@ func postProcessMetricQuery(
 	reducesToScalar := req.RequestType == qbtypes.RequestTypeScalar && config.ReduceTo != qbtypes.ReduceToUnknown
 
 	if !reducesToScalar {
-		result = q.applySeriesLimit(result, query.Limit, query.Order)
+		result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
 	}
 
 	if len(query.Functions) > 0 {
@@ -246,7 +246,7 @@ func postProcessMetricQuery(
 	// Apply reduce to for scalar request type
 	if reducesToScalar {
 		result = q.applyMetricReduceTo(result, config.ReduceTo)
-		result = q.applySeriesLimit(result, query.Limit, query.Order)
+		result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
 		if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok {
 			result.Value = convertTimeSeriesDataToScalar(tsData, tsData.QueryName)
 		}
@@ -263,7 +263,7 @@ func postProcessTraceOperator(
 	req *qbtypes.QueryRangeRequest,
 ) *qbtypes.Result {
 
-	result = q.applySeriesLimit(result, query.Limit, query.Order)
+	result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
 
 	// Apply functions if any
 	if len(query.Functions) > 0 {
@@ -296,7 +296,7 @@ func (q *querier) applyMetricReduceTo(result *qbtypes.Result, reduceOp qbtypes.R
 }
 
 // applySeriesLimit limits the number of series in the result.
-func (q *querier) applySeriesLimit(result *qbtypes.Result, limit int, orderBy []qbtypes.OrderBy) *qbtypes.Result {
+func (q *querier) applySeriesLimit(result *qbtypes.Result, limit, offset int, orderBy []qbtypes.OrderBy) *qbtypes.Result {
 	tsData, ok := result.Value.(*qbtypes.TimeSeriesData)
 	if !ok {
 		return result
@@ -305,7 +305,7 @@ func (q *querier) applySeriesLimit(result *qbtypes.Result, limit int, orderBy []
 	if tsData != nil {
 		for _, agg := range tsData.Aggregations {
 			// Use the ApplySeriesLimit function from querybuildertypes
-			agg.Series = qbtypes.ApplySeriesLimit(agg.Series, orderBy, limit)
+			agg.Series = qbtypes.ApplySeriesLimit(agg.Series, orderBy, limit, offset)
 		}
 	}
 
@@ -357,7 +357,7 @@ func (q *querier) applyFormulas(ctx context.Context, results map[string]*qbtypes
 		case qbtypes.RequestTypeTimeSeries:
 			result := q.processTimeSeriesFormula(ctx, results, formula, req)
 			if result != nil {
-				result = q.applySeriesLimit(result, formula.Limit, formula.Order)
+				result = q.applySeriesLimit(result, formula.Limit, formula.Offset, formula.Order)
 				results[name] = result
 			}
 		case qbtypes.RequestTypeHeatmap:
@@ -370,7 +370,7 @@ func (q *querier) applyFormulas(ctx context.Context, results map[string]*qbtypes
 					bucketFormulaOutputAsHeatmap(tsData, bucketing)
 					resolveHeatmapBucketAxis(tsData, bucketing)
 				}
-				result = q.applySeriesLimit(result, formula.Limit, formula.Order)
+				result = q.applySeriesLimit(result, formula.Limit, formula.Offset, formula.Order)
 				results[name] = result
 			}
 		case qbtypes.RequestTypeScalar:
@@ -651,7 +651,7 @@ func (q *querier) processScalarFormula(
 	}
 
 	// Apply ordering (and limit) before converting to scalar format.
-	formulaSeries = qbtypes.ApplySeriesLimit(formulaSeries, formula.Order, formula.Limit)
+	formulaSeries = qbtypes.ApplySeriesLimit(formulaSeries, formula.Order, formula.Limit, formula.Offset)
 
 	// Convert back to scalar format
 	scalarResult := &qbtypes.ScalarData{

pkg/querier/postprocess_test.go

@@ -186,6 +186,7 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
 		name     string
 		reduceTo qbtypes.ReduceTo
 		order    []qbtypes.OrderBy
+		offset   int
 		expected [][]any
 	}{
 		{
@@ -211,6 +212,13 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
 			reduceTo: qbtypes.ReduceToMax,
 			expected: [][]any{{"payment", 40.0}, {"auth", 30.0}},
 		},
+		{
+			name:     "Max_Desc_Offset1_SkipsHighestPeak",
+			reduceTo: qbtypes.ReduceToMax,
+			order:    orderByValue(qbtypes.OrderDirectionDesc),
+			offset:   1,
+			expected: [][]any{{"auth", 30.0}, {"checkout", 10.0}},
+		},
 	}
 
 	for _, testCase := range testCases {
@@ -220,6 +228,7 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
 				Aggregations: []qbtypes.MetricAggregation{{MetricName: "http_requests_total", ReduceTo: testCase.reduceTo}},
 				Order:        testCase.order,
 				Limit:        2,
+				Offset:       testCase.offset,
 			}
 			req := &qbtypes.QueryRangeRequest{RequestType: qbtypes.RequestTypeScalar}
 

pkg/querybuilder/offset.go

@@ -0,0 +1,17 @@
+package querybuilder
+
+import "github.com/huandu/go-sqlbuilder"
+
+// AddOffset skips offset rows. Call it after the ORDER BY and LIMIT clauses:
+// the ClickHouse flavor renders OFFSET only inside a LIMIT, so an unlimited
+// page writes the clause itself at the current position.
+func AddOffset(sb *sqlbuilder.SelectBuilder, limit, offset int) {
+	if offset <= 0 {
+		return
+	}
+	if limit > 0 {
+		sb.Offset(offset)
+		return
+	}
+	sb.SQL("OFFSET " + sb.Var(offset))
+}

pkg/signoz/module.go

@@ -164,7 +164,7 @@ func NewModules(
 		CloudIntegration:     cloudIntegrationModule,
 		TraceDetail:          impltracedetail.NewModule(impltracedetail.NewTraceStore(telemetryStore, telemetryMetadataStore, schemaMetadataStore, fl), providerSettings, config.TraceDetail),
 		SpanMapper:           spanMapper,
-		LLMPricingRule:       impllmpricingrule.NewModule(impllmpricingrule.NewStore(sqlstore), querier),
+		LLMPricingRule:       impllmpricingrule.NewModule(impllmpricingrule.NewStore(sqlstore), querier, telemetryMetadataStore, providerSettings),
 		Tag:                  tagModule,
 	}
 }

pkg/signoz/signoz.go

@@ -572,6 +572,7 @@ func New(
 		cloudIntegrationModule,
 		modules.LogsPipeline,
 		modules.InfraMonitoring,
+		modules.LLMPricingRule,
 		querier,
 		authz,
 	}

pkg/statementbuilder/auditstatementbuilder/statement_builder.go

@@ -374,7 +374,11 @@ func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
 
 	if query.Limit > 0 && len(query.GroupBy) > 0 {
 		cteSB := sqlbuilder.NewSelectBuilder()
-		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, query, start, end, keys, true, variables)
+		// the series skip happens after the query, so the top-N must still include the skipped groups
+		cteQuery := query
+		cteQuery.Limit += query.Offset
+		cteQuery.Offset = 0
+		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, true, variables)
 		if err != nil {
 			return nil, err
 		}
@@ -536,6 +540,8 @@ func (b *auditQueryStatementBuilder) buildScalarQuery(
 		sb.Limit(query.Limit)
 	}
 
+	querybuilder.AddOffset(sb, query.Limit, query.Offset)
+
 	combinedArgs := append(allGroupByArgs, allAggChArgs...)
 
 	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)

pkg/statementbuilder/logsstatementbuilder/statement_builder.go

@@ -506,7 +506,11 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
 	if query.Limit > 0 && len(query.GroupBy) > 0 {
 		// build the scalar “top/bottom-N” query in its own builder.
 		cteSB := sqlbuilder.NewSelectBuilder()
-		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, query, start, end, keys, true, variables)
+		// the series skip happens after the query, so the top-N must still include the skipped groups
+		cteQuery := query
+		cteQuery.Limit += query.Offset
+		cteQuery.Offset = 0
+		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, true, variables)
 		if err != nil {
 			return nil, err
 		}
@@ -707,6 +711,8 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
 		sb.Limit(query.Limit)
 	}
 
+	querybuilder.AddOffset(sb, query.Limit, query.Offset)
+
 	combinedArgs := allAggChArgs
 
 	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)

pkg/statementbuilder/scopedtracesstatementbuilder/trace_aggregation.go

@@ -654,7 +654,10 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
 		cteFragments = append(cteFragments, fmt.Sprintf("__scoped_traces_total AS (%s)", totalSQL))
 		cteArgs = append(cteArgs, totalArgs)
 
-		limitSQL, limitArgs := outerLimitSQL(query, traceAggs, groupNames, windowSeconds)
+		// the series skip happens after the query, so the top-N must still include the skipped groups
+		cteQuery := query
+		cteQuery.Limit += query.Offset
+		limitSQL, limitArgs := outerLimitSQL(cteQuery, traceAggs, groupNames, windowSeconds)
 		cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", limitSQL))
 		cteArgs = append(cteArgs, limitArgs)
 
@@ -722,6 +725,7 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
 		if query.Limit > 0 {
 			sb.Limit(query.Limit)
 		}
+		querybuilder.AddOffset(sb, query.Limit, query.Offset)
 	}
 
 	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)

pkg/statementbuilder/tracesstatementbuilder/statement_builder.go

@@ -632,7 +632,11 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
 	if query.Limit > 0 && len(query.GroupBy) > 0 {
 		// build the scalar “top/bottom-N” query in its own builder.
 		cteSB := sqlbuilder.NewSelectBuilder()
-		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, query, start, end, keys, variables, true, true)
+		// the series skip happens after the query, so the top-N must still include the skipped groups
+		cteQuery := query
+		cteQuery.Limit += query.Offset
+		cteQuery.Offset = 0
+		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, variables, true, true)
 		if err != nil {
 			return nil, err
 		}
@@ -835,6 +839,8 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
 		sb.Limit(query.Limit)
 	}
 
+	querybuilder.AddOffset(sb, query.Limit, query.Offset)
+
 	combinedArgs := allAggChArgs
 
 	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)

pkg/statementbuilder/tracesstatementbuilder/trace_operator_cte_builder.go

@@ -841,6 +841,8 @@ func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFro
 		sb.Limit(b.operator.Limit)
 	}
 
+	querybuilder.AddOffset(sb, b.operator.Limit, b.operator.Offset)
+
 	combinedArgs := allAggChArgs
 
 	sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
@@ -922,6 +924,8 @@ func (b *traceOperatorCTEBuilder) buildScalarQuery(ctx context.Context, selectFr
 		sb.Limit(b.operator.Limit)
 	}
 
+	querybuilder.AddOffset(sb, b.operator.Limit, b.operator.Offset)
+
 	combinedArgs := allAggChArgs
 
 	// Add HAVING clause if specified

pkg/types/aiobservabilitytypes/stats.go

@@ -0,0 +1,55 @@
+package aiobservabilitytypes
+
+import (
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+const (
+	// statsLookback reaches into the previous day; the metadata store rounds
+	// the start down to a day boundary.
+	statsLookback = 24 * time.Hour
+	// statsModelValuesLimit caps the model list, which also feeds the
+	// unmapped model count. One value is enough for the other keys.
+	statsModelValuesLimit = 100
+)
+
+// statsAttributeKeys are the gen_ai span attributes whose recent presence the
+// stats reporter tracks under telemetry.traces.<key>.exists.
+var statsAttributeKeys = []string{GenAIRequestModel, GenAIToolName, GenAIAgentName}
+
+// StatsFieldValueSelectors look up each tracked attribute's string values
+// since the start of the previous day.
+func StatsFieldValueSelectors(now time.Time) []*telemetrytypes.FieldValueSelector {
+	selectors := make([]*telemetrytypes.FieldValueSelector, 0, len(statsAttributeKeys))
+	for _, key := range statsAttributeKeys {
+		limit := 1
+		if key == GenAIRequestModel {
+			limit = statsModelValuesLimit
+		}
+		selectors = append(selectors, &telemetrytypes.FieldValueSelector{
+			FieldKeySelector: &telemetrytypes.FieldKeySelector{
+				StartUnixMilli: now.Add(-statsLookback).UnixMilli(),
+				Signal:         telemetrytypes.SignalTraces,
+				FieldContext:   telemetrytypes.FieldContextAttribute,
+				FieldDataType:  telemetrytypes.FieldDataTypeString,
+				Name:           key,
+			},
+			Limit: limit,
+		})
+	}
+	return selectors
+}
+
+// NewStatsFromFieldValues reports whether each looked-up key had a value. Keys
+// absent from values are left out, so a failed lookup never reports false.
+func NewStatsFromFieldValues(values map[string]*telemetrytypes.TelemetryFieldValues) map[string]any {
+	stats := make(map[string]any, len(values))
+	for _, key := range statsAttributeKeys {
+		if fieldValues := values[key]; fieldValues != nil {
+			stats["telemetry.traces."+key+".exists"] = len(fieldValues.StringValues) > 0
+		}
+	}
+	return stats
+}

pkg/types/aiobservabilitytypes/stats_test.go

@@ -0,0 +1,73 @@
+package aiobservabilitytypes
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+func TestNewStatsFromFieldValues(t *testing.T) {
+	testCases := []struct {
+		name   string
+		values map[string]*telemetrytypes.TelemetryFieldValues
+		want   map[string]any
+	}{
+		{
+			name: "NoLookups_NothingReported",
+			want: map[string]any{},
+		},
+		{
+			name: "EveryKeySeen_AllTrue",
+			values: map[string]*telemetrytypes.TelemetryFieldValues{
+				GenAIRequestModel: {StringValues: []string{"gpt-5-nano"}},
+				GenAIToolName:     {StringValues: []string{"search"}},
+				GenAIAgentName:    {StringValues: []string{"planner"}},
+			},
+			want: map[string]any{
+				"telemetry.traces.gen_ai.request.model.exists": true,
+				"telemetry.traces.gen_ai.tool.name.exists":     true,
+				"telemetry.traces.gen_ai.agent.name.exists":    true,
+			},
+		},
+		{
+			name: "EmptyAndMissingLookups_OnlyLookedUpKeysReported",
+			values: map[string]*telemetrytypes.TelemetryFieldValues{
+				GenAIRequestModel:  {StringValues: []string{"gpt-5-nano"}},
+				GenAIToolName:      {},
+				GenAIOperationName: {StringValues: []string{"chat"}},
+			},
+			want: map[string]any{
+				"telemetry.traces.gen_ai.request.model.exists": true,
+				"telemetry.traces.gen_ai.tool.name.exists":     false,
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, NewStatsFromFieldValues(testCase.values))
+		})
+	}
+}
+
+func TestStatsFieldValueSelectors(t *testing.T) {
+	now := time.Date(2026, 11, 3, 10, 0, 0, 0, time.UTC)
+
+	selectors := StatsFieldValueSelectors(now)
+
+	require.Len(t, selectors, 3)
+	for _, selector := range selectors {
+		assert.Equal(t, now.Add(-24*time.Hour).UnixMilli(), selector.StartUnixMilli, selector.Name)
+		assert.Equal(t, telemetrytypes.SignalTraces, selector.Signal, selector.Name)
+		assert.Equal(t, telemetrytypes.FieldContextAttribute, selector.FieldContext, selector.Name)
+		assert.Equal(t, telemetrytypes.FieldDataTypeString, selector.FieldDataType, selector.Name)
+	}
+	assert.Equal(t, GenAIRequestModel, selectors[0].Name)
+	assert.Equal(t, 100, selectors[0].Limit)
+	assert.Equal(t, 1, selectors[1].Limit)
+	assert.Equal(t, 1, selectors[2].Limit)
+}

pkg/types/dashboardtypes/perses_dashboard_stats.go

@@ -1,6 +1,7 @@
 package dashboardtypes
 
 import (
+	qb "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
 
@@ -10,6 +11,7 @@ const (
 	statKeyPanelTracesCount  = "dashboard.panels.traces.count"
 	statKeyPanelMetricsCount = "dashboard.panels.metrics.count"
 	statKeyPanelLogsCount    = "dashboard.panels.logs.count"
+	statKeyPanelAICount      = "dashboard.panels.ai.count"
 )
 
 // panelSignalStatKeys maps a builder query's signal to the stat it contributes
@@ -22,14 +24,17 @@ var panelSignalStatKeys = map[telemetrytypes.Signal]string{
 }
 
 // NewStatsFromStorableDashboards reports the stats of stored dashboards. Rows that
-// do not decode as v2 contribute to dashboard.count only.
+// do not decode as v2 contribute to dashboard.count only. System dashboards are
+// provisioned into every org, so they are left out of every count.
 func NewStatsFromStorableDashboards(dashboards []*StorableDashboard) map[string]any {
-	stats := newPanelStats()
+	stats := newStats()
 
 	for _, dashboard := range dashboards {
-		if dashboard == nil {
+		if dashboard == nil || dashboard.Source == SourceSystem {
 			continue
 		}
+		incrementStat(stats, statKeyDashboardCount)
+
 		dashboardV2, err := dashboard.ToDashboardV2(nil)
 		if err != nil {
 			continue
@@ -37,7 +42,6 @@ func NewStatsFromStorableDashboards(dashboards []*StorableDashboard) map[string]
 		addPanelStats(&dashboardV2.Spec, stats)
 	}
 
-	stats[statKeyDashboardCount] = int64(len(dashboards))
 	return stats
 }
 
@@ -45,24 +49,26 @@ func NewStatsFromStorableDashboards(dashboards []*StorableDashboard) map[string]
 // created, straight off the postable spec — the create path has no reason to make
 // a storable round-trip just to be counted.
 func NewStatsFromPostableDashboardV2(postable PostableDashboardV2) map[string]any {
-	stats := newPanelStats()
+	stats := newStats()
+	stats[statKeyDashboardCount] = int64(1)
 	addPanelStats(&postable.Spec, stats)
 
-	stats[statKeyDashboardCount] = int64(1)
 	return stats
 }
 
-func newPanelStats() map[string]any {
+func newStats() map[string]any {
 	return map[string]any{
+		statKeyDashboardCount:    int64(0),
 		statKeyPanelCount:        int64(0),
 		statKeyPanelTracesCount:  int64(0),
 		statKeyPanelMetricsCount: int64(0),
 		statKeyPanelLogsCount:    int64(0),
+		statKeyPanelAICount:      int64(0),
 	}
 }
 
 // addPanelStats counts the panels of a v2 spec, and each panel's queries against
-// the signal they read.
+// the signal they read. AI queries count only towards the AI stat.
 func addPanelStats(spec *DashboardSpec, stats map[string]any) {
 	for _, panel := range spec.Panels {
 		if panel == nil {
@@ -76,6 +82,10 @@ func addPanelStats(spec *DashboardSpec, stats map[string]any) {
 				continue
 			}
 			for _, envelope := range composite.Queries {
+				if envelope.Type == qb.QueryTypeBuilderAI {
+					incrementStat(stats, statKeyPanelAICount)
+					continue
+				}
 				if key, ok := panelSignalStatKeys[envelope.GetSignal()]; ok {
 					incrementStat(stats, key)
 				}

pkg/types/dashboardtypes/perses_dashboard_stats_test.go

@@ -117,20 +117,60 @@ func TestNewStatsFromStorableDashboardsCountsCompositeSubQueries(t *testing.T) {
 	assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
 }
 
-// An AI builder query is always a traces query, so it counts towards traces.
-func TestNewStatsFromStorableDashboardsCountsAIBuilderQueries(t *testing.T) {
-	aiBuilder := `{
+func statsAIBuilderQuery() string {
+	return `{
 		"kind": "time_series",
 		"spec": {"plugin": {"kind": "signoz/AIBuilderQuery", "spec": {"name": "A", "aggregations": [{"expression": "count()"}]}}}
 	}`
-	dashboard := newStatsStorableV2(t, `"p1": `+statsPanel(aiBuilder))
+}
+
+// An AI builder query reads traces but counts only towards the AI stat.
+func TestNewStatsFromStorableDashboardsCountsAIBuilderQueries(t *testing.T) {
+	dashboard := newStatsStorableV2(t, `"p1": `+statsPanel(statsAIBuilderQuery()))
 
 	stats := NewStatsFromStorableDashboards([]*StorableDashboard{dashboard})
 
 	assert.Equal(t, int64(1), stats[statKeyPanelCount])
-	assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelTracesCount])
 	assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
 	assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelAICount])
+}
+
+func TestNewStatsFromStorableDashboardsCountsAICompositeSubQueries(t *testing.T) {
+	composite := `{
+		"kind": "time_series",
+		"spec": {"plugin": {"kind": "signoz/CompositeQuery", "spec": {"queries": [
+			{"type": "builder_ai_query", "spec": {"name": "A", "aggregations": [{"expression": "count()"}]}},
+			{"type": "builder_query", "spec": ` + statsBuilderQuerySpec("B", "logs") + `}
+		]}}}
+	}`
+	dashboard := newStatsStorableV2(t, `
+		"p1": `+statsPanel(composite)+`,
+		"p2": `+statsPanel(statsAIBuilderQuery())+`
+	`)
+
+	stats := NewStatsFromStorableDashboards([]*StorableDashboard{dashboard})
+
+	assert.Equal(t, int64(2), stats[statKeyPanelCount])
+	assert.Equal(t, int64(2), stats[statKeyPanelAICount])
+	assert.Equal(t, int64(0), stats[statKeyPanelTracesCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
+}
+
+func TestNewStatsFromStorableDashboardsSkipsSystemDashboards(t *testing.T) {
+	system := newStatsStorableV2(t, `"p1": `+statsPanel(statsAIBuilderQuery()))
+	system.Source = SourceSystem
+	system.Name = SystemDashboardNamePrefix + "overview"
+	user := newStatsStorableV2(t, `"p1": `+statsPanel(statsBuilderQuery("logs")))
+
+	stats := NewStatsFromStorableDashboards([]*StorableDashboard{system, user})
+
+	assert.Equal(t, int64(1), stats[statKeyDashboardCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelTracesCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelAICount])
+	assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
 }
 
 // promql and clickhouse queries carry no signal, so they land in the panel total
@@ -215,6 +255,17 @@ func TestNewStatsFromPostableDashboardV2(t *testing.T) {
 	assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
 	assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
 	assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelAICount])
+}
+
+func TestNewStatsFromPostableDashboardV2CountsAIBuilderQueries(t *testing.T) {
+	postable := newStatsPostableV2(t, `"p1": `+statsPanel(statsAIBuilderQuery()))
+
+	stats := NewStatsFromPostableDashboardV2(postable)
+
+	assert.Equal(t, int64(1), stats[statKeyDashboardCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelAICount])
 }
 
 func TestNewStatsFromPostableDashboardV2WithNoPanels(t *testing.T) {
@@ -233,4 +284,5 @@ func TestNewStatsFromStorableDashboardsWithNoDashboards(t *testing.T) {
 	assert.Equal(t, int64(0), stats[statKeyPanelTracesCount])
 	assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
 	assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelAICount])
 }

pkg/types/llmpricingruletypes/pricing.go

@@ -277,3 +277,13 @@ func ModelMatchesAnyRule(model string, rules []*LLMPricingRule) bool {
 	}
 	return false
 }
+
+func UnmappedModels(models []*UnmappedModel, rules []*LLMPricingRule) []*UnmappedModel {
+	unmapped := make([]*UnmappedModel, 0, len(models))
+	for _, model := range models {
+		if !ModelMatchesAnyRule(model.ModelName, rules) {
+			unmapped = append(unmapped, model)
+		}
+	}
+	return unmapped
+}

pkg/types/llmpricingruletypes/stats.go

@@ -0,0 +1,31 @@
+package llmpricingruletypes
+
+// NewStatsFromRules reports the rules users added themselves and the synced
+// defaults users pinned.
+func NewStatsFromRules(rules []*LLMPricingRule) map[string]any {
+	custom, overrides := int64(0), int64(0)
+	for _, rule := range rules {
+		switch {
+		case rule.SourceID == nil:
+			custom++
+		case rule.IsOverride:
+			overrides++
+		}
+	}
+
+	return map[string]any{
+		"llmpricingrule.custom.count":   custom,
+		"llmpricingrule.override.count": overrides,
+	}
+}
+
+func NewStatsFromModels(models []string, rules []*LLMPricingRule) map[string]any {
+	unmapped := int64(0)
+	for _, model := range models {
+		if !ModelMatchesAnyRule(model, rules) {
+			unmapped++
+		}
+	}
+
+	return map[string]any{"llmpricingrule.unmapped_model.count": unmapped}
+}

pkg/types/llmpricingruletypes/stats_test.go

@@ -0,0 +1,65 @@
+package llmpricingruletypes
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+func TestNewStatsFromRules(t *testing.T) {
+	sourceID := valuer.GenerateUUID()
+
+	testCases := []struct {
+		name  string
+		rules []*LLMPricingRule
+		want  map[string]any
+	}{
+		{
+			name: "NoRules",
+			want: map[string]any{
+				"llmpricingrule.custom.count":   int64(0),
+				"llmpricingrule.override.count": int64(0),
+			},
+		},
+		{
+			name: "CustomPinnedAndSyncedRules_CountedSeparately",
+			rules: []*LLMPricingRule{
+				{IsOverride: true},
+				{IsOverride: false},
+				{SourceID: &sourceID, IsOverride: true},
+				{SourceID: &sourceID, IsOverride: false},
+			},
+			want: map[string]any{
+				"llmpricingrule.custom.count":   int64(2),
+				"llmpricingrule.override.count": int64(1),
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, NewStatsFromRules(testCase.rules))
+		})
+	}
+}
+
+func TestNewStatsFromModels(t *testing.T) {
+	rules := []*LLMPricingRule{{ModelPattern: StringSlice{"gpt-*"}}}
+
+	testCases := []struct {
+		name   string
+		models []string
+		want   int64
+	}{
+		{name: "NoModels", want: 0},
+		{name: "PricedAndUnpricedModels_UnpricedCounted", models: []string{"gpt-5-nano", "claude-haiku-5-5", "gemini-3"}, want: 2},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, map[string]any{"llmpricingrule.unmapped_model.count": testCase.want}, NewStatsFromModels(testCase.models, rules))
+		})
+	}
+}

pkg/types/querybuildertypes/querybuildertypesv5/builder_query.go

@@ -34,13 +34,13 @@ type QueryBuilderQuery[T any] struct {
 
 	SelectFields []telemetrytypes.TelemetryFieldKey `json:"selectFields,omitzero" description:"The fields to return for each row in raw requests, in addition to the columns always present: timestamp and id for logs, timestamp, trace_id and span_id for traces. Ignored by aggregating requests."`
 
-	Limit int `json:"limit,omitempty" description:"The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest."`
+	Limit int `json:"limit,omitempty" description:"The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows."`
 
 	LimitBy *LimitBy `json:"limitBy,omitempty" description:"Reserved for applying limit within each combination of the given group-by keys. Accepted but currently ignored."`
 
-	Offset int `json:"offset,omitempty" description:"The number of rows to skip before returning results. Prefer cursor for raw requests."`
+	Offset int `json:"offset,omitempty" description:"The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging."`
 
-	Cursor string `json:"cursor,omitempty" description:"The opaque nextCursor value from a previous raw result, used to fetch the next page."`
+	Cursor string `json:"cursor,omitempty" description:"The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it."`
 
 	Having *Having `json:"having,omitempty" description:"A filter applied to aggregation results, referencing aliases or expressions, for example count() > 100."`
 

pkg/types/querybuildertypes/querybuildertypesv5/formula.go

@@ -26,6 +26,8 @@ type QueryBuilderFormula struct {
 
 	Limit int `json:"limit,omitempty" description:"The maximum number of series or rows to return. Unlimited when 0."`
 
+	Offset int `json:"offset,omitempty" description:"The number of ordered series or rows to skip before the page starts."`
+
 	Having *Having `json:"having,omitempty" description:"Accepted but currently ignored; the computed values are returned unfiltered."`
 
 	Functions []Function `json:"functions,omitzero" description:"Post-processing functions applied in order to each result series."`
@@ -94,6 +96,14 @@ func (f QueryBuilderFormula) Validate() error {
 		)
 	}
 
+	if f.Offset < 0 {
+		return errors.NewInvalidInputf(
+			errors.CodeInvalidInput,
+			"offset must be non-negative, got %d",
+			f.Offset,
+		)
+	}
+
 	// Validate expression is parseable
 	if _, err := govaluate.NewEvaluableExpressionWithFunctions(f.Expression, EvalFuncs()); err != nil {
 		return errors.NewInvalidInputf(

pkg/types/querybuildertypes/querybuildertypesv5/req_getters.go

@@ -180,6 +180,8 @@ func (q *QueryEnvelope) GetOffset() int {
 		return spec.Offset
 	case QueryBuilderQuery[MetricAggregation]:
 		return spec.Offset
+	case QueryBuilderFormula:
+		return spec.Offset
 	}
 	return 0
 }

pkg/types/querybuildertypes/querybuildertypesv5/req_setters.go

@@ -209,6 +209,9 @@ func (q *QueryEnvelope) SetOffset(offset int) {
 	case QueryBuilderQuery[MetricAggregation]:
 		spec.Offset = offset
 		q.Spec = spec
+	case QueryBuilderFormula:
+		spec.Offset = offset
+		q.Spec = spec
 	}
 }
 

pkg/types/querybuildertypes/querybuildertypesv5/series_limit.go

@@ -13,9 +13,8 @@ const (
 	DefaultOrderByKey = "__result"
 )
 
-// ApplyLimit applies limit and ordering to a list of time series
-// This function sorts the series based on the provided order criteria and applies the limit.
-func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit int) []*TimeSeries {
+// ApplySeriesLimit sorts the series by the order criteria, skips the first offset of them and keeps at most limit.
+func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit, offset int) []*TimeSeries {
 	if len(series) == 0 {
 		return series
 	}
@@ -65,6 +64,10 @@ func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit int) []*Tim
 		return 0
 	})
 
+	if offset > 0 {
+		series = series[min(offset, len(series)):]
+	}
+
 	// Apply limit if specified
 	if limit > 0 && len(series) > limit {
 		return series[:limit]

pkg/types/querybuildertypes/querybuildertypesv5/series_limit_test.go

@@ -84,7 +84,7 @@ func TestApplySeriesLimit(t *testing.T) {
 			},
 		}
 
-		result := ApplySeriesLimit(series, orderBy, 2)
+		result := ApplySeriesLimit(series, orderBy, 2, 0)
 
 		// Should return top 2 series by value: service-b (50.0), service-c (30.0)
 		assert.Len(t, result, 2)
@@ -209,7 +209,7 @@ func TestApplySeriesLimit(t *testing.T) {
 			},
 		}
 
-		result := ApplySeriesLimit(series, orderBy, 3)
+		result := ApplySeriesLimit(series, orderBy, 3, 0)
 
 		// Should return 3 series sorted by service (asc), then environment (desc)
 		// Expected order:
@@ -233,6 +233,25 @@ func TestApplySeriesLimit(t *testing.T) {
 		assert.Equal(t, "prod", result[2].Labels[1].Value)
 		assert.Equal(t, 40.0, result[2].Values[0].Value)
 	})
+
+	t.Run("Offset skips the top series before the limit applies", func(t *testing.T) {
+		hostKey := telemetrytypes.TelemetryFieldKey{Name: "host", FieldDataType: telemetrytypes.FieldDataTypeString}
+		series := []*TimeSeries{
+			{Labels: []*Label{{Key: hostKey, Value: "h1"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 10.0}}},
+			{Labels: []*Label{{Key: hostKey, Value: "h2"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 50.0}}},
+			{Labels: []*Label{{Key: hostKey, Value: "h3"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 30.0}}},
+			{Labels: []*Label{{Key: hostKey, Value: "h4"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 20.0}}},
+		}
+
+		// Default order is value descending: h2, h3, h4, h1. Skipping one leaves h3, h4.
+		result := ApplySeriesLimit(series, nil, 2, 1)
+
+		require.Len(t, result, 2)
+		assert.Equal(t, "h3", result[0].Labels[0].Value)
+		assert.Equal(t, "h4", result[1].Labels[0].Value)
+
+		assert.Empty(t, ApplySeriesLimit(series, nil, 2, 4))
+	})
 }
 
 func TestApplySeriesLimitRanksHeatmapSeriesByBucketTotals(t *testing.T) {
@@ -271,7 +290,7 @@ func TestApplySeriesLimitRanksHeatmapSeriesByBucketTotals(t *testing.T) {
 		},
 	}
 
-	result := ApplySeriesLimit(series, nil, 2)
+	result := ApplySeriesLimit(series, nil, 2, 0)
 
 	require.Len(t, result, 2)
 	assert.Equal(t, "busy", result[0].Labels[0].Value)

pkg/types/querybuildertypes/querybuildertypesv5/trace_operator.go

@@ -55,7 +55,7 @@ type QueryBuilderTraceOperator struct {
 	Having *Having `json:"having,omitempty" description:"A condition on the aggregation results, for example count() > 10."`
 
 	Limit  int    `json:"limit,omitempty" description:"The maximum number of traces or groups to return. For raw requests 0 means 100 rows; for aggregations 0 means unlimited."`
-	Offset int    `json:"offset,omitempty" description:"The number of raw results to skip before returning. Aggregation requests ignore it."`
+	Offset int    `json:"offset,omitempty" description:"The number of raw or trace rows, or of ordered series or rows for time_series and scalar requests, to skip before the page starts."`
 	Cursor string `json:"cursor,omitempty" description:"Accepted but currently ignored by the trace operator; page raw results with offset."`
 
 	Legend string `json:"legend" description:"The display name template for each series; {{field}} placeholders are replaced by group-by label values."`

tests/e2e/docs/flake-notes.md

@@ -259,3 +259,20 @@ neither antd 5.11 nor `@rc-component/trigger` closes it on Escape, so
 it, but when the popover covered the Advanced Options header, that click waited
 on "intercepts pointer events" for the rest of the budget. The driver now closes
 the popover through its trigger and waits for it to go.
+
+## dnd-kit stops clicks right after a drop
+
+`app-shell/sidenav` TC-03 needed a retry in half its CI jobs and failed all
+three attempts in four of 22. dnd-kit 6.1 adds a capture-phase click handler on
+`document` that stops propagation once a drag activates, and removes it 50 ms
+after the drop. `saveShortcuts` clicked Save inside that window, so the modal
+stayed open and no preference request went out. It now clicks again while the
+dialog is still open.
+
+## The Kafka configuration button drops clicks while loading
+
+`messaging-queues/kafka` TC-01 failed its first attempt in four of 20 CI jobs.
+The health check button passes `loading` to an antd 5.11 `Button` while the
+three onboarding status queries fetch, and a loading antd button ignores clicks
+without being disabled, so Playwright clicks it and nothing opens.
+`openHealthCheck` clicks again until the dialog shows.

tests/e2e/drivers/messaging-queues.ts

@@ -258,9 +258,15 @@ export function messagingQueuesDriver(page: Page): MessagingQueues {
 
 		openHealthCheck: async () => {
 			// "Configuration", or "Missing Configuration (N)" when checks fail.
-			await page.getByRole('button', { name: /Configuration/ }).click();
+			const button = page.getByRole('button', { name: /Configuration/ });
 			const dialog = page.getByRole('dialog');
-			await expect(dialog).toBeVisible();
+			// The button drops clicks while its checks load, and is not disabled.
+			await expect(async () => {
+				if (!(await dialog.isVisible())) {
+					await button.click({ timeout: 5_000 });
+				}
+				await expect(dialog).toBeVisible({ timeout: 1_000 });
+			}).toPass();
 			return dialog;
 		},
 

tests/e2e/drivers/shell.ts

@@ -635,7 +635,16 @@ export function shellDriver(page: Page): Shell {
 
 			saveShortcuts: async () => {
 				const saved = page.waitForResponse(isPreferencePut('nav_shortcuts'));
-				await page.getByTestId('save-changes-btn').click();
+				const dialog = page
+					.getByRole('dialog')
+					.filter({ hasText: 'Manage Shortcuts' });
+				// dnd-kit stops every click for 50 ms after a drop.
+				await expect(async () => {
+					if (await dialog.isVisible()) {
+						await page.getByTestId('save-changes-btn').click({ timeout: 5_000 });
+					}
+					await expect(dialog).toBeHidden({ timeout: 1_000 });
+				}).toPass();
 				const response = await saved;
 				expect(response.ok()).toBe(true);
 				return (response.request().postDataJSON() as { value: string[] }).value;

tests/fixtures/querier.py

@@ -300,6 +300,9 @@ def build_builder_query(
     functions: list[dict] | None = None,
     bucket_options: dict | None = None,
     disabled: bool = False,
+    order: list[dict] | None = None,
+    limit: int | None = None,
+    offset: int | None = None,
 ) -> dict:
     spec: dict[str, Any] = {
         "name": name,
@@ -336,6 +339,15 @@ def build_builder_query(
     if functions:
         spec["functions"] = functions
 
+    if order:
+        spec["order"] = order
+
+    if limit is not None:
+        spec["limit"] = limit
+
+    if offset is not None:
+        spec["offset"] = offset
+
     return {"type": "builder_query", "spec": spec}
 
 
@@ -348,6 +360,7 @@ def build_formula_query(
     disabled: bool = False,
     order: list[dict] | None = None,
     limit: int | None = None,
+    offset: int | None = None,
 ) -> dict:
     spec: dict[str, Any] = {
         "name": name,
@@ -362,6 +375,8 @@ def build_formula_query(
         spec["order"] = order
     if limit is not None:
         spec["limit"] = limit
+    if offset is not None:
+        spec["offset"] = offset
     return {"type": "builder_formula", "spec": spec}
 
 
@@ -668,6 +683,7 @@ def build_scalar_query(
     group_by: list[dict] | None = None,
     order: list[dict] | None = None,
     limit: int | None = None,
+    offset: int | None = None,
     filter_expression: str | None = None,
     having_expression: str | None = None,
     step_interval: int = DEFAULT_STEP_INTERVAL,
@@ -694,6 +710,9 @@ def build_scalar_query(
     if limit is not None:
         spec["limit"] = limit
 
+    if offset is not None:
+        spec["offset"] = offset
+
     if filter_expression:
         spec["filter"] = {"expression": filter_expression}
 

tests/integration/tests/querierlogs/16_offset.py

@@ -0,0 +1,81 @@
+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.logs import Logs
+from fixtures.querier import (
+    Aggregation,
+    BuilderQuery,
+    RequestType,
+    TelemetryFieldKey,
+    get_all_series,
+    make_query_request,
+)
+
+
+@pytest.mark.parametrize(
+    "limit, offset, expected_services",
+    [
+        # limit=2 selects the top two groups in SQL first; the skipped group must still count toward that page
+        pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_limited_page"),
+        pytest.param(None, 1, ["service-a", "service-b", "service-d"], id="offset_without_limit"),
+        pytest.param(2, 4, [], id="offset_past_the_end"),
+    ],
+)
+def test_logs_time_series_offset_skips_series_in_order(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    limit: int | None,
+    offset: int,
+    expected_services: list[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # Ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1, all in the bucket before now.
+    service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": service},
+                body=f"{service} log {i}",
+            )
+            for service, count in service_counts.items()
+            for i in range(count)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type=RequestType.TIME_SERIES,
+        queries=[
+            BuilderQuery(
+                signal="logs",
+                name="A",
+                aggregations=[Aggregation("count()")],
+                group_by=[TelemetryFieldKey("service.name", "string", "resource")],
+                step_interval=60,
+                limit=limit,
+                offset=offset,
+            ).to_dict()
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    bucket_ms = int((now - timedelta(minutes=1)).timestamp() * 1000)
+    assert [
+        (
+            next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
+            [(point["timestamp"], point["value"]) for point in series["values"]],
+        )
+        for series in get_all_series(response.json(), "A")
+    ] == [(service, [(bucket_ms, service_counts[service])]) for service in expected_services]

tests/integration/tests/queriermetrics/19_offset.py

@@ -0,0 +1,75 @@
+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 Metrics
+from fixtures.querier import build_builder_query, get_all_series, make_query_request
+
+
+@pytest.mark.parametrize(
+    "limit, offset, expected_services",
+    [
+        pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
+        pytest.param(None, 1, ["service-a", "service-b", "service-d"], id="offset_without_limit"),
+        pytest.param(2, 4, [], id="offset_past_the_end"),
+    ],
+)
+def test_metrics_time_series_offset_skips_series_in_order(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+    limit: int | None,
+    offset: int,
+    expected_services: list[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # Ordered by value, descending: service-c 70, service-a 50, service-b 30, service-d 10, one sample each in the bucket before now.
+    service_values = {"service-a": 50.0, "service-b": 30.0, "service-c": 70.0, "service-d": 10.0}
+    insert_metrics(
+        [
+            Metrics(
+                metric_name="offset.gauge",
+                labels={"service.name": service},
+                timestamp=now - timedelta(seconds=1),
+                temporality="Unspecified",
+                type_="Gauge",
+                is_monotonic=False,
+                value=value,
+            )
+            for service, value in service_values.items()
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        queries=[
+            build_builder_query(
+                "A",
+                "offset.gauge",
+                "avg",
+                "sum",
+                temporality="unspecified",
+                group_by=["service.name"],
+                limit=limit,
+                offset=offset,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert [
+        (
+            next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
+            [point["value"] for point in series["values"]],
+        )
+        for series in get_all_series(response.json(), "A")
+    ] == [(service, [service_values[service]]) for service in expected_services]

tests/integration/tests/querierscalar/09_offset.py

@@ -0,0 +1,249 @@
+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.logs import Logs
+from fixtures.metrics import Metrics
+from fixtures.querier import (
+    RequestType,
+    build_aggregation,
+    build_formula_query,
+    build_group_by_field,
+    build_metrics_aggregation,
+    build_order_by,
+    build_scalar_query,
+    find_named_result,
+    get_scalar_table_data,
+    make_query_request,
+    make_scalar_query_request,
+)
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
+
+# Four services ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1.
+service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
+
+
+@pytest.mark.parametrize(
+    "limit, offset, expected_services",
+    [
+        pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
+        pytest.param(None, 3, ["service-d"], id="offset_without_limit"),
+        pytest.param(2, 4, [], id="offset_past_the_end"),
+    ],
+)
+def test_logs_scalar_offset_skips_rows_in_order(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    limit: int | None,
+    offset: int,
+    expected_services: list[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": service},
+                body=f"{service} log {i}",
+            )
+            for service, count in service_counts.items()
+            for i in range(count)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_scalar_query_request(
+        signoz,
+        token,
+        now,
+        [
+            build_scalar_query(
+                name="A",
+                signal="logs",
+                aggregations=[build_aggregation("count()")],
+                group_by=[build_group_by_field("service.name")],
+                order=[build_order_by("count()", "desc")],
+                limit=limit,
+                offset=offset,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert get_scalar_table_data(response.json()) == [[service, service_counts[service]] for service in expected_services]
+
+
+@pytest.mark.parametrize(
+    "limit, offset, expected_services",
+    [
+        pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
+        pytest.param(None, 3, ["service-d"], id="offset_without_limit"),
+        pytest.param(2, 4, [], id="offset_past_the_end"),
+    ],
+)
+def test_traces_scalar_offset_skips_rows_in_order(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+    limit: int | None,
+    offset: int,
+    expected_services: list[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(seconds=i + 1),
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                resources={"service.name": service},
+                name=f"{service} span {i}",
+            )
+            for service, count in service_counts.items()
+            for i in range(count)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_scalar_query_request(
+        signoz,
+        token,
+        now,
+        [
+            build_scalar_query(
+                name="A",
+                signal="traces",
+                aggregations=[build_aggregation("count()")],
+                group_by=[build_group_by_field("service.name")],
+                order=[build_order_by("count()", "desc")],
+                limit=limit,
+                offset=offset,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert get_scalar_table_data(response.json()) == [[service, service_counts[service]] for service in expected_services]
+
+
+def test_metrics_scalar_offset_skips_rows_in_order(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # Ordered by value, descending: service-c 70, service-a 50, service-b 30, service-d 10.
+    insert_metrics(
+        [
+            Metrics(
+                metric_name="offset.gauge",
+                labels={"service.name": service},
+                timestamp=now - timedelta(seconds=1),
+                temporality="Unspecified",
+                type_="Gauge",
+                is_monotonic=False,
+                value=value,
+            )
+            for service, value in {"service-a": 50.0, "service-b": 30.0, "service-c": 70.0, "service-d": 10.0}.items()
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_scalar_query_request(
+        signoz,
+        token,
+        now,
+        [
+            build_scalar_query(
+                name="A",
+                signal="metrics",
+                aggregations=[build_metrics_aggregation("offset.gauge", "latest", "sum", "unspecified", reduce_to="last")],
+                group_by=[build_group_by_field("service.name", "string", "attribute")],
+                order=[build_order_by("sum(offset.gauge)", "desc")],
+                limit=2,
+                offset=1,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert get_scalar_table_data(response.json()) == [["service-a", 50.0], ["service-b", 30.0]]
+
+
+def test_formula_scalar_offset_skips_rows_in_order(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": service},
+                body=f"{service} log {i}",
+            )
+            for service, count in service_counts.items()
+            for i in range(count)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type=RequestType.SCALAR,
+        queries=[
+            build_scalar_query(
+                name="A",
+                signal="logs",
+                aggregations=[build_aggregation("count()")],
+                group_by=[build_group_by_field("service.name")],
+                disabled=True,
+            ),
+            build_formula_query(
+                "F1",
+                "A * 2",
+                order=[build_order_by("__result", "desc")],
+                limit=2,
+                offset=1,
+            ),
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    result = find_named_result(response.json()["data"]["data"]["results"], "F1")
+    assert result is not None, "formula result F1 not found"
+    service_column = next(i for i, column in enumerate(result["columns"]) if column["name"] == "service.name")
+    value_column = next(i for i, column in enumerate(result["columns"]) if column["name"] == "__result")
+    assert [[row[service_column], row[value_column]] for row in result["data"]] == [["service-a", 10.0], ["service-b", 6.0]]
+
+
+def test_negative_offset_is_rejected(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_scalar_query_request(
+        signoz,
+        token,
+        datetime.now(tz=UTC).replace(second=0, microsecond=0),
+        [build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=-1)],
+    )
+
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert response.json()["error"]["message"] == "invalid query 'A': offset must be non-negative, got -1"

tests/integration/tests/queriertraces/15_offset.py

@@ -0,0 +1,152 @@
+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,
+    RequestType,
+    TelemetryFieldKey,
+    find_named_result,
+    get_all_series,
+    make_query_request,
+)
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
+
+
+@pytest.mark.parametrize(
+    "limit, offset, expected_services",
+    [
+        # limit=2 selects the top two groups in SQL first; the skipped group must still count toward that page
+        pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_limited_page"),
+        pytest.param(None, 1, ["service-a", "service-b", "service-d"], id="offset_without_limit"),
+        pytest.param(2, 4, [], id="offset_past_the_end"),
+    ],
+)
+def test_traces_time_series_offset_skips_series_in_order(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+    limit: int | None,
+    offset: int,
+    expected_services: list[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # Ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1, all in the bucket before now.
+    service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(seconds=i + 1),
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                resources={"service.name": service},
+                name=f"{service} span {i}",
+            )
+            for service, count in service_counts.items()
+            for i in range(count)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type=RequestType.TIME_SERIES,
+        queries=[
+            BuilderQuery(
+                signal="traces",
+                name="A",
+                aggregations=[Aggregation("count()")],
+                group_by=[TelemetryFieldKey("service.name", "string", "resource")],
+                step_interval=60,
+                limit=limit,
+                offset=offset,
+            ).to_dict()
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    bucket_ms = int((now - timedelta(minutes=1)).timestamp() * 1000)
+    assert [
+        (
+            next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
+            [(point["timestamp"], point["value"]) for point in series["values"]],
+        )
+        for series in get_all_series(response.json(), "A")
+    ] == [(service, [(bucket_ms, service_counts[service])]) for service in expected_services]
+
+
+def test_trace_operator_scalar_offset_skips_rows_in_order(
+    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)
+    span_counts = {"offset-a": 4, "offset-b": 3, "offset-c": 2, "offset-d": 1}
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(seconds=10),
+                duration=timedelta(seconds=1),
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                parent_span_id="",
+                name="scalar-offset",
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                status_message="",
+                resources={"service.name": service},
+                attributes={"operation.type": "scalar-offset"},
+            )
+            for service, count in span_counts.items()
+            for _ in range(count)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type=RequestType.SCALAR,
+        queries=[
+            {
+                "type": "builder_query",
+                "spec": {
+                    "name": "A",
+                    "signal": "traces",
+                    "disabled": True,
+                    "filter": {"expression": "operation.type = 'scalar-offset'"},
+                    "aggregations": [{"expression": "count()"}],
+                },
+            },
+            {
+                "type": "builder_trace_operator",
+                "spec": {
+                    "name": "C",
+                    "expression": "A",
+                    "aggregations": [{"expression": "count()"}],
+                    "groupBy": [{"name": "service.name", "fieldContext": "resource"}],
+                    "limit": 2,
+                    "offset": 1,
+                },
+            },
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    result = find_named_result(response.json()["data"]["data"]["results"], "C")
+    assert result is not None, "trace_operator result C not found"
+    assert result["data"] == [["offset-b", 3], ["offset-c", 2]]