feat(querier): apply limitBy to raw queries

A raw query with limitBy now keeps at most value rows for each
combination of the keys, in the query order, before limit applies. The
logs, traces, and audit list builders resolve each key as they resolve
an order-by key and add LIMIT n BY after ORDER BY. A key can be any
field.

A raw query ordered by timestamp runs once per time window and pages by
cursor, so LIMIT n BY in each window keeps n rows of a key in every
window. With limitBy, the querier skips the windows and runs one query
over the whole range with offset paging.

Time series, heatmap, raw stream, and trace requests still reject
limitBy. The limit description now says which raw requests page by
cursor and which return 100 rows when limit is 0.

Assisted-by: Claude Opus 5.5

docs/api/openapi.yml

@@ -4066,14 +4066,20 @@ components:
         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.
+            0, aggregations are unlimited. A raw request ordered by timestamp, with
+            id next for logs, and without limitBy returns a page of 10000 rows; follow
+            nextCursor for the rest. Any other raw request returns 100 rows; use offset
+            for the rest.
           type: integer
         limitBy:
           allOf:
           - $ref: '#/components/schemas/Querybuildertypesv5LimitBy'
-          description: Reserved for applying limit within each combination of the
-            given group-by keys. Accepted but currently ignored.
+          description: Keeps at most value rows for each combination of the given
+            keys, in the query order, before limit applies. A scalar request partitions
+            its groups, so each key must be a group-by key. A raw request partitions
+            its rows, so a key can be any field, and the request runs as one query
+            over the whole range with no nextCursor. Time series and heatmap requests
+            reject it.
         name:
           description: The unique name of the query, such as A or B; formulas, trace
             operators and selectedQueryName refer to it.
@@ -9876,15 +9882,15 @@ components:
     Querybuildertypesv5LimitBy:
       properties:
         keys:
-          description: The group-by field names that partition the result before the
-            limit is applied.
+          description: The field names that partition the result before the limit
+            is applied. For a scalar request, each one must be a group-by key.
           items:
             type: string
           nullable: true
           type: array
         value:
-          description: The aggregation alias or expression whose value ranks entries
-            within each partition.
+          description: The number of rows to keep in each partition, a positive integer
+            sent as a string. The query order decides which rows are kept.
           type: string
       type: object
     Querybuildertypesv5LinearBucketsSpec:
@@ -10165,14 +10171,20 @@ components:
         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.
+            0, aggregations are unlimited. A raw request ordered by timestamp, with
+            id next for logs, and without limitBy returns a page of 10000 rows; follow
+            nextCursor for the rest. Any other raw request returns 100 rows; use offset
+            for the rest.
           type: integer
         limitBy:
           allOf:
           - $ref: '#/components/schemas/Querybuildertypesv5LimitBy'
-          description: Reserved for applying limit within each combination of the
-            given group-by keys. Accepted but currently ignored.
+          description: Keeps at most value rows for each combination of the given
+            keys, in the query order, before limit applies. A scalar request partitions
+            its groups, so each key must be a group-by key. A raw request partitions
+            its rows, so a key can be any field, and the request runs as one query
+            over the whole range with no nextCursor. Time series and heatmap requests
+            reject it.
         name:
           description: The unique name of the query, such as A or B; formulas, trace
             operators and selectedQueryName refer to it.
@@ -10280,14 +10292,20 @@ components:
         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.
+            0, aggregations are unlimited. A raw request ordered by timestamp, with
+            id next for logs, and without limitBy returns a page of 10000 rows; follow
+            nextCursor for the rest. Any other raw request returns 100 rows; use offset
+            for the rest.
           type: integer
         limitBy:
           allOf:
           - $ref: '#/components/schemas/Querybuildertypesv5LimitBy'
-          description: Reserved for applying limit within each combination of the
-            given group-by keys. Accepted but currently ignored.
+          description: Keeps at most value rows for each combination of the given
+            keys, in the query order, before limit applies. A scalar request partitions
+            its groups, so each key must be a group-by key. A raw request partitions
+            its rows, so a key can be any field, and the request runs as one query
+            over the whole range with no nextCursor. Time series and heatmap requests
+            reject it.
         name:
           description: The unique name of the query, such as A or B; formulas, trace
             operators and selectedQueryName refer to it.
@@ -10395,14 +10413,20 @@ components:
         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.
+            0, aggregations are unlimited. A raw request ordered by timestamp, with
+            id next for logs, and without limitBy returns a page of 10000 rows; follow
+            nextCursor for the rest. Any other raw request returns 100 rows; use offset
+            for the rest.
           type: integer
         limitBy:
           allOf:
           - $ref: '#/components/schemas/Querybuildertypesv5LimitBy'
-          description: Reserved for applying limit within each combination of the
-            given group-by keys. Accepted but currently ignored.
+          description: Keeps at most value rows for each combination of the given
+            keys, in the query order, before limit applies. A scalar request partitions
+            its groups, so each key must be a group-by key. A raw request partitions
+            its rows, so a key can be any field, and the request runs as one query
+            over the whole range with no nextCursor. Time series and heatmap requests
+            reject it.
         name:
           description: The unique name of the query, such as A or B; formulas, trace
             operators and selectedQueryName refer to it.

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

@@ -5131,12 +5131,12 @@ export interface Querybuildertypesv5HavingDTO {
 export interface Querybuildertypesv5LimitByDTO {
 	/**
 	 * @type array,null
-	 * @description The group-by field names that partition the result before the limit is applied.
+	 * @description The field names that partition the result before the limit is applied. For a scalar request, each one must be a group-by key.
 	 */
 	keys?: string[] | null;
 	/**
 	 * @type string
-	 * @description The aggregation alias or expression whose value ranks entries within each partition.
+	 * @description The number of rows to keep in each partition, a positive integer sent as a string. The query order decides which rows are kept.
 	 */
 	value?: string;
 }
@@ -5306,11 +5306,11 @@ 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 requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited. A raw request ordered by timestamp, with id next for logs, and without limitBy returns a page of 10000 rows; follow nextCursor for the rest. Any other raw request returns 100 rows; use offset for the rest.
 	 */
 	limit?: number;
 	/**
-	 * @description Reserved for applying limit within each combination of the given group-by keys. Accepted but currently ignored.
+	 * @description Keeps at most value rows for each combination of the given keys, in the query order, before limit applies. A scalar request partitions its groups, so each key must be a group-by key. A raw request partitions its rows, so a key can be any field, and the request runs as one query over the whole range with no nextCursor. Time series and heatmap requests reject it.
 	 */
 	limitBy?: Querybuildertypesv5LimitByDTO;
 	/**
@@ -5698,11 +5698,11 @@ 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 requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited. A raw request ordered by timestamp, with id next for logs, and without limitBy returns a page of 10000 rows; follow nextCursor for the rest. Any other raw request returns 100 rows; use offset for the rest.
 	 */
 	limit?: number;
 	/**
-	 * @description Reserved for applying limit within each combination of the given group-by keys. Accepted but currently ignored.
+	 * @description Keeps at most value rows for each combination of the given keys, in the query order, before limit applies. A scalar request partitions its groups, so each key must be a group-by key. A raw request partitions its rows, so a key can be any field, and the request runs as one query over the whole range with no nextCursor. Time series and heatmap requests reject it.
 	 */
 	limitBy?: Querybuildertypesv5LimitByDTO;
 	/**
@@ -5878,11 +5878,11 @@ 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 requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited. A raw request ordered by timestamp, with id next for logs, and without limitBy returns a page of 10000 rows; follow nextCursor for the rest. Any other raw request returns 100 rows; use offset for the rest.
 	 */
 	limit?: number;
 	/**
-	 * @description Reserved for applying limit within each combination of the given group-by keys. Accepted but currently ignored.
+	 * @description Keeps at most value rows for each combination of the given keys, in the query order, before limit applies. A scalar request partitions its groups, so each key must be a group-by key. A raw request partitions its rows, so a key can be any field, and the request runs as one query over the whole range with no nextCursor. Time series and heatmap requests reject it.
 	 */
 	limitBy?: Querybuildertypesv5LimitByDTO;
 	/**
@@ -5977,11 +5977,11 @@ 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 requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited. A raw request ordered by timestamp, with id next for logs, and without limitBy returns a page of 10000 rows; follow nextCursor for the rest. Any other raw request returns 100 rows; use offset for the rest.
 	 */
 	limit?: number;
 	/**
-	 * @description Reserved for applying limit within each combination of the given group-by keys. Accepted but currently ignored.
+	 * @description Keeps at most value rows for each combination of the given keys, in the query order, before limit applies. A scalar request partitions its groups, so each key must be a group-by key. A raw request partitions its rows, so a key can be any field, and the request runs as one query over the whole range with no nextCursor. Time series and heatmap requests reject it.
 	 */
 	limitBy?: Querybuildertypesv5LimitByDTO;
 	/**

pkg/querier/builder_query.go

@@ -229,6 +229,12 @@ func (q *builderQuery[T]) isWindowList() bool {
 		return false
 	}
 
+	// LIMIT n BY must see the whole range in one query; each window would keep n
+	// rows per key again
+	if q.spec.LimitBy != nil {
+		return false
+	}
+
 	// first ORDER BY must be `timestamp`
 	if q.spec.Order[0].Key.Name != "timestamp" {
 		return false

pkg/querier/postprocess.go

@@ -246,6 +246,12 @@ func postProcessMetricQuery(
 	// Apply reduce to for scalar request type
 	if reducesToScalar {
 		result = q.applyMetricReduceTo(result, config.ReduceTo)
+		// metrics reduce to scalar in Go, so limitBy runs here, before limit as in SQL
+		if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok && query.LimitBy != nil {
+			for _, agg := range tsData.Aggregations {
+				agg.Series = qbtypes.ApplySeriesLimitBy(agg.Series, query.Order, query.LimitBy)
+			}
+		}
 		result = q.applySeriesLimit(result, query.Limit, query.Order)
 		if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok {
 			result.Value = convertTimeSeriesDataToScalar(tsData, tsData.QueryName)

pkg/statementbuilder/auditstatementbuilder/statement_builder.go

@@ -166,6 +166,18 @@ func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]) []
 		})
 	}
 
+	if query.LimitBy != nil {
+		for _, name := range query.LimitBy.Keys {
+			key := telemetrytypes.GetFieldKeyFromKeyText(name)
+			keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
+				Name:          key.Name,
+				Signal:        telemetrytypes.SignalLogs,
+				FieldContext:  key.FieldContext,
+				FieldDataType: key.FieldDataType,
+			})
+		}
+	}
+
 	for idx := range keySelectors {
 		keySelectors[idx].Signal = telemetrytypes.SignalLogs
 		keySelectors[idx].Source = telemetrytypes.SourceAudit
@@ -286,6 +298,21 @@ func (b *auditQueryStatementBuilder) buildListQuery(
 		sb.OrderBy(fmt.Sprintf("%s %s", sqlbuilder.Escape(colExpr), orderBy.Direction.StringValue()))
 	}
 
+	// LIMIT BY runs after ORDER BY and before LIMIT, so the order decides which rows
+	// each key keeps
+	if query.LimitBy != nil {
+		columns := make([]string, len(query.LimitBy.Keys))
+		for i, name := range query.LimitBy.Keys {
+			key := telemetrytypes.GetFieldKeyFromKeyText(name)
+			b.adjustKey(&key, keys)
+			columns[i], err = querybuilder.ResolveColumn(ctx, info, b.storage, &key, telemetrytypes.FieldDataTypeUnspecified, keys)
+			if err != nil {
+				return nil, err
+			}
+		}
+		sb.SQL(sqlbuilder.Escape(fmt.Sprintf("LIMIT %d BY %s", query.LimitBy.Rows(), strings.Join(columns, ", "))))
+	}
+
 	if query.Limit > 0 {
 		sb.Limit(query.Limit)
 	} else {
@@ -374,7 +401,10 @@ 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)
+		// time series requests do not apply limitBy; a stored alert skips validation and can carry one
+		cteQuery := query
+		cteQuery.LimitBy = nil
+		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, true, variables)
 		if err != nil {
 			return nil, err
 		}
@@ -532,6 +562,16 @@ func (b *auditQueryStatementBuilder) buildScalarQuery(
 		sb.OrderBy("__result_0 DESC")
 	}
 
+	// LIMIT BY runs after ORDER BY and before LIMIT; validation keeps every key a
+	// group-by key, which this query selects under its own name
+	if query.LimitBy != nil {
+		columns := make([]string, len(query.LimitBy.Keys))
+		for i, key := range query.LimitBy.Keys {
+			columns[i] = clickhousesql.Identifier(key)
+		}
+		sb.SQL(sqlbuilder.Escape(fmt.Sprintf("LIMIT %d BY %s", query.LimitBy.Rows(), strings.Join(columns, ", "))))
+	}
+
 	if query.Limit > 0 {
 		sb.Limit(query.Limit)
 	}

pkg/statementbuilder/auditstatementbuilder/statement_builder_test.go

@@ -164,6 +164,25 @@ func TestStatementBuilder(t *testing.T) {
 				Args:  []any{"service_account", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
 			},
 		},
+		// List: the newest event of each principal
+		{
+			name:        "List_LimitBy_NewestPerPrincipal",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Order: []qbtypes.OrderBy{
+					{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "timestamp"}}, Direction: qbtypes.OrderDirectionDesc},
+					{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "id"}}, Direction: qbtypes.OrderDirectionDesc},
+				},
+				LimitBy: &qbtypes.LimitBy{Keys: []string{"signoz.audit.principal.email"}, Value: "1"},
+				Limit:   100,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY timestamp desc, id desc LIMIT 1 BY multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = true, `attribute_string_signoz$$audit$$principal$$email`, NULL) LIMIT ?",
+				Args:  []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
+			},
+		},
 		// Scalar: alert — count forbidden errors (outcome + action AND)
 		{
 			name:        "ScalarCountByOutcomeAndAction",
@@ -184,6 +203,54 @@ func TestStatementBuilder(t *testing.T) {
 				Args:  []any{"failure", "update", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
 			},
 		},
+		// Scalar: the 2 busiest principals for each action
+		{
+			name:        "Scalar_LimitBy_TopPrincipalsPerAction",
+			requestType: qbtypes.RequestTypeScalar,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				Source:       telemetrytypes.SourceAudit,
+				StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+				Aggregations: []qbtypes.LogAggregation{
+					{Expression: "count()"},
+				},
+				GroupBy: []qbtypes.GroupByKey{
+					{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "signoz.audit.action"}},
+					{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "signoz.audit.principal.email"}},
+				},
+				LimitBy: &qbtypes.LimitBy{Keys: []string{"signoz.audit.action"}, Value: "2"},
+				Limit:   50,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT toString(multiIf(`attribute_string_signoz$$audit$$action_exists` = true, `attribute_string_signoz$$audit$$action`, NULL)) AS `signoz.audit.action`, toString(multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = true, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `signoz.audit.action`, `signoz.audit.principal.email` ORDER BY __result_0 DESC LIMIT 2 BY `signoz.audit.action` LIMIT ?",
+				Args:  []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 50},
+			},
+		},
+		// TimeSeries: time series requests do not apply limitBy, so the top-N limit CTE leaves it out
+		{
+			name:        "TimeSeries_LimitBy_NotInLimitCTE",
+			requestType: qbtypes.RequestTypeTimeSeries,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				Source:       telemetrytypes.SourceAudit,
+				StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+				Aggregations: []qbtypes.LogAggregation{
+					{Expression: "count()"},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: "signoz.audit.outcome = 'failure'",
+				},
+				GroupBy: []qbtypes.GroupByKey{
+					{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "signoz.audit.principal.email"}},
+				},
+				LimitBy: &qbtypes.LimitBy{Keys: []string{"signoz.audit.principal.email"}, Value: "1"},
+				Limit:   5,
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __limit_cte AS (SELECT toString(multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = true, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = true) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `signoz.audit.principal.email` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 60 SECOND) AS ts, toString(multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = true, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = true) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`signoz.audit.principal.email`) GLOBAL IN (SELECT `signoz.audit.principal.email` FROM __limit_cte) GROUP BY ts, `signoz.audit.principal.email`",
+				Args:  []any{"failure", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 5, "failure", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
+			},
+		},
 		// TimeSeries: failures grouped by principal email with top-N limit
 		{
 			name:        "TimeSeriesFailuresGroupedByPrincipal",

pkg/statementbuilder/logsstatementbuilder/statement_builder.go

@@ -217,6 +217,18 @@ func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], bo
 		})
 	}
 
+	if query.LimitBy != nil {
+		for _, name := range query.LimitBy.Keys {
+			key := telemetrytypes.GetFieldKeyFromKeyText(name)
+			keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
+				Name:          key.Name,
+				Signal:        telemetrytypes.SignalLogs,
+				FieldContext:  key.FieldContext,
+				FieldDataType: key.FieldDataType,
+			})
+		}
+	}
+
 	for idx := range keySelectors {
 		keySelectors[idx].Signal = telemetrytypes.SignalLogs
 		keySelectors[idx].SelectorMatchType = telemetrytypes.FieldSelectorMatchTypeExact
@@ -403,6 +415,21 @@ func (b *logQueryStatementBuilder) buildListQuery(
 		sb.OrderBy(fmt.Sprintf("%s %s", sqlbuilder.Escape(colExpr), orderBy.Direction.StringValue()))
 	}
 
+	// LIMIT BY runs after ORDER BY and before LIMIT, so the order decides which rows
+	// each key keeps
+	if query.LimitBy != nil {
+		columns := make([]string, len(query.LimitBy.Keys))
+		for i, name := range query.LimitBy.Keys {
+			key := telemetrytypes.GetFieldKeyFromKeyText(name)
+			b.adjustKey(&key, keys)
+			columns[i], err = querybuilder.ResolveColumn(ctx, info, b.storage, columnKey(&key), telemetrytypes.FieldDataTypeUnspecified, keys)
+			if err != nil {
+				return nil, err
+			}
+		}
+		sb.SQL(sqlbuilder.Escape(fmt.Sprintf("LIMIT %d BY %s", query.LimitBy.Rows(), strings.Join(columns, ", "))))
+	}
+
 	// Add limit and offset
 	if query.Limit > 0 {
 		sb.Limit(query.Limit)
@@ -506,7 +533,10 @@ 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)
+		// time series requests do not apply limitBy; a stored alert skips validation and can carry one
+		cteQuery := query
+		cteQuery.LimitBy = nil
+		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, true, variables)
 		if err != nil {
 			return nil, err
 		}
@@ -702,6 +732,16 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
 		sb.OrderBy("__result_0 DESC")
 	}
 
+	// LIMIT BY runs after ORDER BY and before LIMIT; validation keeps every key a
+	// group-by key
+	if query.LimitBy != nil {
+		columns := make([]string, len(query.LimitBy.Keys))
+		for i, key := range query.LimitBy.Keys {
+			columns[i], _ = groupByOrderAlias(key, query.GroupBy)
+		}
+		sb.SQL(sqlbuilder.Escape(fmt.Sprintf("LIMIT %d BY %s", query.LimitBy.Rows(), strings.Join(columns, ", "))))
+	}
+
 	// Add limit and offset
 	if query.Limit > 0 {
 		sb.Limit(query.Limit)

pkg/statementbuilder/logsstatementbuilder/stmt_builder_test.go

@@ -333,6 +333,26 @@ func TestStatementBuilderListQuery(t *testing.T) {
 			},
 			expectedErr: nil,
 		},
+		{
+			name:        "list query with limitBy keeps the newest row per key",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Limit:  10,
+				Order: []qbtypes.OrderBy{
+					{
+						Key:       qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "timestamp"}},
+						Direction: qbtypes.OrderDirectionDesc,
+					},
+				},
+				LimitBy: &qbtypes.LimitBy{Keys: []string{"service.name", "materialized.key.name"}, Value: "1"},
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY timestamp desc LIMIT 1 BY multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL), multiIf(`attribute_string_materialized$$key$$name_exists` = true, `attribute_string_materialized$$key$$name`, NULL) LIMIT ?",
+				Args:  []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+			},
+			expectedErr: nil,
+		},
 		{
 			name:        "filter skips entirely but emits LIKE-without-wildcards warning",
 			requestType: qbtypes.RequestTypeRaw,

pkg/statementbuilder/scopedtracesstatementbuilder/trace_aggregation.go

@@ -719,6 +719,15 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
 		if len(query.Order) == 0 {
 			sb.OrderBy("__result_0 DESC")
 		}
+		// LIMIT BY runs after ORDER BY and before LIMIT; validation keeps every key a
+		// group-by key
+		if query.LimitBy != nil {
+			columns := make([]string, len(query.LimitBy.Keys))
+			for i, key := range query.LimitBy.Keys {
+				columns[i] = orderColumn(key, query.GroupBy)
+			}
+			sb.SQL(sqlbuilder.Escape(fmt.Sprintf("LIMIT %d BY %s", query.LimitBy.Rows(), strings.Join(columns, ", "))))
+		}
 		if query.Limit > 0 {
 			sb.Limit(query.Limit)
 		}

pkg/statementbuilder/tracesstatementbuilder/statement_builder.go

@@ -213,6 +213,12 @@ func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation])
 		keySelectors = append(keySelectors, keySelectorsForField(query.Order[idx].Key.TelemetryFieldKey)...)
 	}
 
+	if query.LimitBy != nil {
+		for _, name := range query.LimitBy.Keys {
+			keySelectors = append(keySelectors, keySelectorsForField(telemetrytypes.GetFieldKeyFromKeyText(name))...)
+		}
+	}
+
 	for idx := range keySelectors {
 		keySelectors[idx].Signal = telemetrytypes.SignalTraces
 		keySelectors[idx].SelectorMatchType = telemetrytypes.FieldSelectorMatchTypeExact
@@ -413,6 +419,21 @@ func (b *traceQueryStatementBuilder) buildListQuery(
 		sb.OrderBy(fmt.Sprintf("%s %s", sqlbuilder.Escape(expr), orderBy.Direction.StringValue()))
 	}
 
+	// LIMIT BY runs after ORDER BY and before LIMIT, so the order decides which rows
+	// each key keeps
+	if query.LimitBy != nil {
+		columns := make([]string, len(query.LimitBy.Keys))
+		for i, name := range query.LimitBy.Keys {
+			key := telemetrytypes.GetFieldKeyFromKeyText(name)
+			adjustTraceKey(&key, keys)
+			columns[i], err = querybuilder.ResolveColumn(ctx, info, b.storage, &key, telemetrytypes.FieldDataTypeUnspecified, keys)
+			if err != nil {
+				return nil, err
+			}
+		}
+		sb.SQL(sqlbuilder.Escape(fmt.Sprintf("LIMIT %d BY %s", query.LimitBy.Rows(), strings.Join(columns, ", "))))
+	}
+
 	// Add limit and offset
 	if query.Limit > 0 {
 		sb.Limit(query.Limit)
@@ -632,7 +653,10 @@ 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)
+		// time series requests do not apply limitBy; a stored alert skips validation and can carry one
+		cteQuery := query
+		cteQuery.LimitBy = nil
+		cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, variables, true, true)
 		if err != nil {
 			return nil, err
 		}
@@ -830,6 +854,16 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
 		sb.OrderBy("__result_0 DESC")
 	}
 
+	// LIMIT BY runs after ORDER BY and before LIMIT; validation keeps every key a
+	// group-by key
+	if query.LimitBy != nil {
+		columns := make([]string, len(query.LimitBy.Keys))
+		for i, key := range query.LimitBy.Keys {
+			columns[i], _ = groupByOrderAlias(key, query.GroupBy)
+		}
+		sb.SQL(sqlbuilder.Escape(fmt.Sprintf("LIMIT %d BY %s", query.LimitBy.Rows(), strings.Join(columns, ", "))))
+	}
+
 	// Add limit and offset
 	if query.Limit > 0 {
 		sb.Limit(query.Limit)

pkg/statementbuilder/tracesstatementbuilder/stmt_builder_test.go

@@ -76,6 +76,60 @@ func TestStatementBuilder(t *testing.T) {
 			},
 			expectedErr: nil,
 		},
+		{
+			name:        "Scalar_LimitBy_TopOperationsPerService",
+			requestType: qbtypes.RequestTypeScalar,
+			query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal:       telemetrytypes.SignalTraces,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.TraceAggregation{
+					{
+						Expression: "p99(duration_nano)",
+					},
+				},
+				GroupBy: []qbtypes.GroupByKey{
+					{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "service.name"}},
+					{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "name"}},
+				},
+				LimitBy: &qbtypes.LimitBy{Keys: []string{"service.name"}, Value: "3"},
+				Limit:   20,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL)) AS `__GROUP_BY_KEY_0_service.name`, toString(name) AS `__GROUP_BY_KEY_1_name`, quantile(0.99)(toFloat64(duration_nano)) AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name`, `__GROUP_BY_KEY_1_name` ORDER BY __result_0 DESC LIMIT 3 BY `__GROUP_BY_KEY_0_service.name` LIMIT ?",
+				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 20},
+			},
+			expectedErr: nil,
+		},
+		{
+			name:        "TimeSeries_LimitBy_NotInLimitCTE",
+			requestType: qbtypes.RequestTypeTimeSeries,
+			query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal:       telemetrytypes.SignalTraces,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.TraceAggregation{
+					{
+						Expression: "count()",
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: "service.name = 'redis-manual'",
+				},
+				Limit: 10,
+				GroupBy: []qbtypes.GroupByKey{
+					{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: "service.name",
+						},
+					},
+				},
+				LimitBy: &qbtypes.LimitBy{Keys: []string{"service.name"}, Value: "1"},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), __limit_cte AS (SELECT toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name`",
+				Args:  []any{"redis-manual", "%service.name%", "%service.name\":\"redis-manual%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10, "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
+			},
+			expectedErr: nil,
+		},
 		{
 			name:        "OR b/w resource attr and attribute",
 			requestType: qbtypes.RequestTypeTimeSeries,

pkg/types/querybuildertypes/querybuildertypesv5/builder_elements.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"math"
 	"slices"
+	"strconv"
 	"time"
 
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -714,8 +715,18 @@ func (f Function) Validate() error {
 }
 
 type LimitBy struct {
-	Keys  []string `json:"keys" description:"The group-by field names that partition the result before the limit is applied."`
-	Value string   `json:"value" description:"The aggregation alias or expression whose value ranks entries within each partition."`
+	Keys  []string `json:"keys" description:"The field names that partition the result before the limit is applied. For a scalar request, each one must be a group-by key."`
+	Value string   `json:"value" description:"The number of rows to keep in each partition, a positive integer sent as a string. The query order decides which rows are kept."`
+}
+
+// Rows returns the number of rows each partition keeps, or 0 when the value is
+// not a positive integer.
+func (l *LimitBy) Rows() int {
+	rows, err := strconv.Atoi(l.Value)
+	if err != nil || rows < 0 {
+		return 0
+	}
+	return rows
 }
 
 // Copy creates a deep copy of LimitBy.

pkg/types/querybuildertypes/querybuildertypesv5/builder_query.go

@@ -34,9 +34,9 @@ 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 requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited. A raw request ordered by timestamp, with id next for logs, and without limitBy returns a page of 10000 rows; follow nextCursor for the rest. Any other raw request returns 100 rows; use offset for the rest."`
 
-	LimitBy *LimitBy `json:"limitBy,omitempty" description:"Reserved for applying limit within each combination of the given group-by keys. Accepted but currently ignored."`
+	LimitBy *LimitBy `json:"limitBy,omitempty" description:"Keeps at most value rows for each combination of the given keys, in the query order, before limit applies. A scalar request partitions its groups, so each key must be a group-by key. A raw request partitions its rows, so a key can be any field, and the request runs as one query over the whole range with no nextCursor. Time series and heatmap requests reject it."`
 
 	Offset int `json:"offset,omitempty" description:"The number of rows to skip before returning results. Prefer cursor for raw requests."`
 

pkg/types/querybuildertypes/querybuildertypesv5/series_limit.go

@@ -73,6 +73,41 @@ func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit int) []*Tim
 	return series
 }
 
+// ApplySeriesLimitBy sorts the series as ApplySeriesLimit does, then keeps the
+// first limitBy.Rows() series of each combination of the limitBy key labels, as
+// ClickHouse LIMIT n BY does for the queries it runs.
+func ApplySeriesLimitBy(series []*TimeSeries, orderBy []OrderBy, limitBy *LimitBy) []*TimeSeries {
+	if limitBy == nil || len(series) == 0 {
+		return series
+	}
+
+	series = ApplySeriesLimit(series, orderBy, 0)
+
+	rows := limitBy.Rows()
+	kept := series[:0]
+	keptByPartition := make(map[string]int)
+	partitionValues := make([]string, len(limitBy.Keys))
+	for _, s := range series {
+		for i, key := range limitBy.Keys {
+			partitionValues[i] = ""
+			for _, label := range s.Labels {
+				if label.Key.Name == key {
+					// quoted so that distinct value lists never join to the same key
+					partitionValues[i] = strconv.Quote(convertValueToString(label.Value))
+					break
+				}
+			}
+		}
+		partition := strings.Join(partitionValues, ",")
+		if keptByPartition[partition] < rows {
+			keptByPartition[partition]++
+			kept = append(kept, s)
+		}
+	}
+
+	return kept
+}
+
 // compareSeries compares two time series based on the order criteria
 // Returns true if series i should come before series j.
 func compareSeries(seriesI, seriesJ *TimeSeries, orderBy []OrderBy, seriesValues map[*TimeSeries]float64, seriesLabels map[*TimeSeries]map[string]string) bool {

pkg/types/querybuildertypes/querybuildertypesv5/validation.go

@@ -86,6 +86,7 @@ type validationConfig struct {
 	skipGroupByValidation          bool
 	withTimestampGroupByValidation bool
 	withReduceToValidation         bool
+	withLimitByValidation          bool
 }
 
 func applyValidationOptions(opts []ValidationOption) validationConfig {
@@ -160,6 +161,14 @@ func WithReduceToValidation() ValidationOption {
 	}
 }
 
+// WithLimitByValidation accepts a limitBy and checks it. Without this option a
+// limitBy is rejected, because only scalar and raw requests apply it.
+func WithLimitByValidation() ValidationOption {
+	return func(cfg *validationConfig) {
+		cfg.withLimitByValidation = true
+	}
+}
+
 // Validate performs preliminary validation on QueryBuilderQuery.
 func (q *QueryBuilderQuery[T]) Validate(opts ...ValidationOption) error {
 	cfg := applyValidationOptions(opts)
@@ -176,6 +185,10 @@ func (q *QueryBuilderQuery[T]) Validate(opts ...ValidationOption) error {
 		return err
 	}
 
+	if err := q.validateLimitBy(cfg); err != nil {
+		return err
+	}
+
 	if err := q.validateLimitAndPagination(cfg); err != nil {
 		return err
 	}
@@ -235,6 +248,43 @@ func (q *QueryBuilderQuery[T]) validateGroupBy(cfg validationConfig) error {
 	return nil
 }
 
+func (q *QueryBuilderQuery[T]) validateLimitBy(cfg validationConfig) error {
+	if q.LimitBy == nil {
+		return nil
+	}
+
+	if !cfg.withLimitByValidation {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "limitBy is supported only for scalar and raw requests")
+	}
+
+	if q.LimitBy.Rows() == 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "limitBy value must be a positive integer, got %q", q.LimitBy.Value)
+	}
+
+	if len(q.LimitBy.Keys) == 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "limitBy keys cannot be empty")
+	}
+
+	// a raw list has no groups, so its keys can name any field
+	if cfg.skipGroupByValidation {
+		return nil
+	}
+
+	// the partitions of a scalar result are its groups, so each key must be a group-by key
+	groupByNames := make([]string, len(q.GroupBy))
+	for i, groupBy := range q.GroupBy {
+		groupByNames[i] = groupBy.Name
+	}
+	for _, key := range q.LimitBy.Keys {
+		if !slices.Contains(groupByNames, key) {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "limitBy key %q is not a group-by key", key).
+				WithSuggestions(errors.NewValidReferences(errors.NounKeys, groupByNames...))
+		}
+	}
+
+	return nil
+}
+
 func (q *QueryBuilderQuery[T]) validateSignal() error {
 	// Signal validation is handled during unmarshaling in req.go
 	// Valid signals are: metrics, traces, logs
@@ -1033,8 +1083,10 @@ func GetValidationOptions(requestType RequestType) []ValidationOption {
 	case RequestTypeTimeSeries, RequestTypeHeatmap:
 		return []ValidationOption{WithSkipSelectFieldValidation(), WithTimestampGroupByValidation()}
 	case RequestTypeScalar:
-		return []ValidationOption{WithSkipSelectFieldValidation(), WithReduceToValidation()}
-	case RequestTypeRaw, RequestTypeRawStream, RequestTypeTrace:
+		return []ValidationOption{WithSkipSelectFieldValidation(), WithReduceToValidation(), WithLimitByValidation()}
+	case RequestTypeRaw:
+		return []ValidationOption{WithSkipAggregationValidation(), WithSkipHavingValidation(), WithSkipAggregationOrderBy(), WithSkipGroupByValidation(), WithLimitByValidation()}
+	case RequestTypeRawStream, RequestTypeTrace:
 		return []ValidationOption{WithSkipAggregationValidation(), WithSkipHavingValidation(), WithSkipAggregationOrderBy(), WithSkipGroupByValidation()}
 	default:
 		return []ValidationOption{}

tests/fixtures/querier.py

@@ -670,6 +670,7 @@ def build_scalar_query(
     limit: int | None = None,
     filter_expression: str | None = None,
     having_expression: str | None = None,
+    limit_by: dict | None = None,
     step_interval: int = DEFAULT_STEP_INTERVAL,
     disabled: bool = False,
     functions: list[dict] | None = None,
@@ -700,6 +701,9 @@ def build_scalar_query(
     if having_expression:
         spec["having"] = {"expression": having_expression}
 
+    if limit_by:
+        spec["limitBy"] = limit_by
+
     if functions:
         spec["functions"] = functions
 
@@ -743,6 +747,7 @@ def build_raw_query(
     limit: int | None = None,
     filter_expression: str | None = None,
     select_fields: list[dict] | None = None,
+    limit_by: dict | None = None,
     step_interval: int = DEFAULT_STEP_INTERVAL,
     disabled: bool = False,
 ) -> dict:
@@ -765,6 +770,9 @@ def build_raw_query(
     if select_fields:
         spec["selectFields"] = select_fields
 
+    if limit_by:
+        spec["limitBy"] = limit_by
+
     return {"type": "builder_query", "spec": spec}
 
 

tests/integration/tests/queriercommon/10_limit_by.py

@@ -0,0 +1,301 @@
+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 (
+    assert_scalar_result_order,
+    build_aggregation,
+    build_group_by_field,
+    build_metrics_aggregation,
+    build_order_by,
+    build_raw_query,
+    build_scalar_query,
+    get_error_message,
+    get_rows,
+    get_scalar_table_data,
+    make_query_request,
+    make_scalar_query_request,
+)
+from fixtures.traces import TraceIdGenerator, Traces
+
+# Ordered by value: b/checkout 7, c/login 6, a/login 5, a/checkout 4, a/search 3,
+# b/login 2, b/search 1. Each case below returns rows that a plain limit cannot.
+LIMIT_BY_VALUES = {
+    ("service-a", "login"): 5,
+    ("service-a", "checkout"): 4,
+    ("service-a", "search"): 3,
+    ("service-b", "checkout"): 7,
+    ("service-b", "login"): 2,
+    ("service-b", "search"): 1,
+    ("service-c", "login"): 6,
+}
+
+LIMIT_BY_CASES = [
+    pytest.param(
+        "1",
+        None,
+        [("service-b", "checkout", 7), ("service-c", "login", 6), ("service-a", "login", 5)],
+        id="one_per_service",
+    ),
+    pytest.param(
+        "2",
+        5,
+        [("service-b", "checkout", 7), ("service-c", "login", 6), ("service-a", "login", 5), ("service-a", "checkout", 4), ("service-b", "login", 2)],
+        id="two_per_service_then_limit",
+    ),
+]
+
+
+@pytest.mark.parametrize("rows_per_service,limit,expected_rows", LIMIT_BY_CASES)
+def test_logs_scalar_limit_by(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    rows_per_service: str,
+    limit: int | None,
+    expected_rows: list[tuple],
+) -> None:
+    """limitBy keeps the top rows of each service, in the query order, before the limit applies."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs([Logs(timestamp=now - timedelta(seconds=i + 1), resources={"service.name": service}, attributes={"operation": operation}, body=f"{service} {operation} {i}") for (service, operation), count in LIMIT_BY_VALUES.items() for i in range(count)])
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    query = build_scalar_query(
+        name="A",
+        signal="logs",
+        aggregations=[build_aggregation("count()")],
+        group_by=[build_group_by_field("service.name", "string", "resource"), build_group_by_field("operation", "string", "attribute")],
+        limit_by={"keys": ["service.name"], "value": rows_per_service},
+        limit=limit,
+    )
+    response = make_scalar_query_request(signoz, token, now, [query])
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_scalar_result_order(get_scalar_table_data(response.json()), expected_rows, "logs limitBy")
+
+
+@pytest.mark.parametrize("rows_per_service,limit,expected_rows", LIMIT_BY_CASES)
+def test_traces_scalar_limit_by(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+    rows_per_service: str,
+    limit: int | None,
+    expected_rows: list[tuple],
+) -> None:
+    """limitBy keeps the top rows of each service, in the query order, before the limit applies."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(seconds=i + 1),
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                resources={"service.name": service},
+                attributes={"operation": operation},
+                name=f"{service} {operation}",
+            )
+            for (service, operation), count in LIMIT_BY_VALUES.items()
+            for i in range(count)
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    query = build_scalar_query(
+        name="A",
+        signal="traces",
+        aggregations=[build_aggregation("count()")],
+        group_by=[build_group_by_field("service.name", "string", "resource"), build_group_by_field("operation", "string", "attribute")],
+        limit_by={"keys": ["service.name"], "value": rows_per_service},
+        limit=limit,
+    )
+    response = make_scalar_query_request(signoz, token, now, [query])
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_scalar_result_order(get_scalar_table_data(response.json()), expected_rows, "traces limitBy")
+
+
+@pytest.mark.parametrize("rows_per_service,limit,expected_rows", LIMIT_BY_CASES)
+def test_metrics_scalar_limit_by(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+    rows_per_service: str,
+    limit: int | None,
+    expected_rows: list[tuple],
+) -> None:
+    """Metrics reduce to scalar in the querier, so limitBy runs there; the result must match the SQL behaviour."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_metrics(
+        [
+            Metrics(
+                metric_name="test.limit_by",
+                labels={"service.name": service, "operation": operation},
+                timestamp=now - timedelta(seconds=1),
+                temporality="Unspecified",
+                type_="Gauge",
+                is_monotonic=False,
+                value=float(value),
+            )
+            for (service, operation), value in LIMIT_BY_VALUES.items()
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    query = build_scalar_query(
+        name="A",
+        signal="metrics",
+        aggregations=[build_metrics_aggregation("test.limit_by", "latest", "sum", "unspecified", reduce_to="last")],
+        group_by=[build_group_by_field("service.name", "string", "attribute"), build_group_by_field("operation", "string", "attribute")],
+        limit_by={"keys": ["service.name"], "value": rows_per_service},
+        limit=limit,
+    )
+    response = make_scalar_query_request(signoz, token, now, [query])
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_scalar_result_order(get_scalar_table_data(response.json()), expected_rows, "metrics limitBy")
+
+
+# Rows as (key, minutes ago). A raw request over the last 3 hours that pages by
+# cursor reads the newest hour first and the rest next, so a limitBy applied per
+# window would return key a twice and key c twice.
+RAW_LIMIT_BY_ROWS = [("c", 5), ("a", 10), ("a", 20), ("a", 120), ("b", 130), ("c", 140), ("d", 150)]
+
+RAW_LIMIT_BY_CASES = [
+    pytest.param("1", None, ["c@5", "a@10", "b@130", "d@150"], id="newest_per_key"),
+    pytest.param("2", None, ["c@5", "a@10", "a@20", "b@130", "c@140", "d@150"], id="two_newest_per_key"),
+    pytest.param("1", 3, ["c@5", "a@10", "b@130"], id="newest_per_key_then_limit"),
+]
+
+
+def make_raw_limit_by_request(signoz: types.SigNoz, token: str, now: datetime, query: dict) -> list[str]:
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(hours=3)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[query],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    # the request runs as one query over the whole range, so there is no cursor to follow
+    assert not response.json()["data"]["data"]["results"][0].get("nextCursor"), response.text
+    return [row["data"]["body" if query["spec"]["signal"] == "logs" else "name"] for row in get_rows(response)]
+
+
+@pytest.mark.parametrize(
+    "order",
+    [
+        pytest.param([build_order_by("timestamp")], id="timestamp"),
+        pytest.param([build_order_by("timestamp"), build_order_by("id")], id="timestamp_id"),
+    ],
+)
+@pytest.mark.parametrize("rows_per_key,limit,expected_rows", RAW_LIMIT_BY_CASES)
+def test_logs_raw_limit_by(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    order: list[dict],
+    rows_per_key: str,
+    limit: int | None,
+    expected_rows: list[str],
+) -> None:
+    """limitBy keeps the newest logs of each fingerprint over the whole range, then the limit applies."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs([Logs(timestamp=now - timedelta(minutes=minutes), resources={"service.name": "service-a"}, attributes={"exception.fingerprint": key}, body=f"{key}@{minutes}") for key, minutes in RAW_LIMIT_BY_ROWS])
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    query = build_raw_query(name="A", signal="logs", order=order, limit=limit, limit_by={"keys": ["exception.fingerprint"], "value": rows_per_key})
+
+    assert make_raw_limit_by_request(signoz, token, now, query) == expected_rows
+
+
+@pytest.mark.parametrize("rows_per_key,limit,expected_rows", RAW_LIMIT_BY_CASES)
+def test_traces_raw_limit_by(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+    rows_per_key: str,
+    limit: int | None,
+    expected_rows: list[str],
+) -> None:
+    """limitBy keeps the newest spans of each operation over the whole range, then the limit applies."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(minutes=minutes),
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                resources={"service.name": "service-a"},
+                attributes={"operation": key},
+                name=f"{key}@{minutes}",
+            )
+            for key, minutes in RAW_LIMIT_BY_ROWS
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    query = build_raw_query(
+        name="A",
+        signal="traces",
+        order=[build_order_by("timestamp")],
+        limit=limit,
+        select_fields=[{"name": "name"}],
+        limit_by={"keys": ["operation"], "value": rows_per_key},
+    )
+
+    assert make_raw_limit_by_request(signoz, token, now, query) == expected_rows
+
+
+@pytest.mark.parametrize(
+    "request_type,limit_by,expected_message",
+    [
+        pytest.param("time_series", {"keys": ["service.name"], "value": "1"}, "invalid query 'A': limitBy is supported only for scalar and raw requests", id="time_series"),
+        pytest.param("scalar", {"keys": ["operation"], "value": "1"}, "invalid query 'A': limitBy key \"operation\" is not a group-by key", id="key_not_grouped"),
+        pytest.param("scalar", {"keys": ["service.name"], "value": "0"}, "invalid query 'A': limitBy value must be a positive integer, got \"0\"", id="zero_value"),
+        pytest.param("scalar", {"keys": [], "value": "1"}, "invalid query 'A': limitBy keys cannot be empty", id="no_keys"),
+    ],
+)
+def test_limit_by_bad_request(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    request_type: str,
+    limit_by: dict,
+    expected_message: str,
+) -> None:
+    """A limitBy that the query cannot apply is a bad request, never ignored."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    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=request_type,
+        queries=[
+            build_scalar_query(
+                name="A",
+                signal="logs",
+                aggregations=[build_aggregation("count()")],
+                group_by=[build_group_by_field("service.name", "string", "resource")],
+                limit_by=limit_by,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert get_error_message(response.json()) == expected_message