chore: change tests to reflect non-normalised support for scope

For scope name and version fields

.gitignore

@@ -231,4 +231,5 @@ cython_debug/
 # LSP config files
 pyrightconfig.json
 
-
+# agents
+*settings.local.json

docs/api/openapi.yml

@@ -3109,6 +3109,32 @@ components:
       - end
       - limit
       type: object
+    InframonitoringtypesPostableStatefulSets:
+      properties:
+        end:
+          format: int64
+          type: integer
+        filter:
+          $ref: '#/components/schemas/Querybuildertypesv5Filter'
+        groupBy:
+          items:
+            $ref: '#/components/schemas/Querybuildertypesv5GroupByKey'
+          nullable: true
+          type: array
+        limit:
+          type: integer
+        offset:
+          type: integer
+        orderBy:
+          $ref: '#/components/schemas/Querybuildertypesv5OrderBy'
+        start:
+          format: int64
+          type: integer
+      required:
+      - start
+      - end
+      - limit
+      type: object
     InframonitoringtypesPostableVolumes:
       properties:
         end:
@@ -3150,6 +3176,76 @@ components:
       - list
       - grouped_list
       type: string
+    InframonitoringtypesStatefulSetRecord:
+      properties:
+        currentPods:
+          type: integer
+        desiredPods:
+          type: integer
+        meta:
+          additionalProperties:
+            type: string
+          nullable: true
+          type: object
+        podCountsByPhase:
+          $ref: '#/components/schemas/InframonitoringtypesPodCountsByPhase'
+        statefulSetCPU:
+          format: double
+          type: number
+        statefulSetCPULimit:
+          format: double
+          type: number
+        statefulSetCPURequest:
+          format: double
+          type: number
+        statefulSetMemory:
+          format: double
+          type: number
+        statefulSetMemoryLimit:
+          format: double
+          type: number
+        statefulSetMemoryRequest:
+          format: double
+          type: number
+        statefulSetName:
+          type: string
+      required:
+      - statefulSetName
+      - statefulSetCPU
+      - statefulSetCPURequest
+      - statefulSetCPULimit
+      - statefulSetMemory
+      - statefulSetMemoryRequest
+      - statefulSetMemoryLimit
+      - desiredPods
+      - currentPods
+      - podCountsByPhase
+      - meta
+      type: object
+    InframonitoringtypesStatefulSets:
+      properties:
+        endTimeBeforeRetention:
+          type: boolean
+        records:
+          items:
+            $ref: '#/components/schemas/InframonitoringtypesStatefulSetRecord'
+          nullable: true
+          type: array
+        requiredMetricsCheck:
+          $ref: '#/components/schemas/InframonitoringtypesRequiredMetricsCheck'
+        total:
+          type: integer
+        type:
+          $ref: '#/components/schemas/InframonitoringtypesResponseType'
+        warning:
+          $ref: '#/components/schemas/Querybuildertypesv5QueryWarnData'
+      required:
+      - type
+      - records
+      - total
+      - requiredMetricsCheck
+      - endTimeBeforeRetention
+      type: object
     InframonitoringtypesVolumeRecord:
       properties:
         meta:
@@ -12498,6 +12594,81 @@ paths:
       summary: List Volumes for Infra Monitoring
       tags:
       - inframonitoring
+  /api/v2/infra_monitoring/statefulsets:
+    post:
+      deprecated: false
+      description: 'Returns a paginated list of Kubernetes StatefulSets with key aggregated
+        pod metrics: CPU usage and memory working set summed across pods owned by
+        the statefulset, plus average CPU/memory request and limit utilization (statefulSetCPURequest,
+        statefulSetCPULimit, statefulSetMemoryRequest, statefulSetMemoryLimit). Each
+        row also reports the latest known desiredPods (k8s.statefulset.desired_pods)
+        and currentPods (k8s.statefulset.current_pods) replica counts and per-group
+        podCountsByPhase ({ pending, running, succeeded, failed, unknown } from each
+        pod''s latest k8s.pod.phase value). Each statefulset includes metadata attributes
+        (k8s.statefulset.name, k8s.namespace.name, k8s.cluster.name). The response
+        type is ''list'' for the default k8s.statefulset.name grouping or ''grouped_list''
+        for custom groupBy keys; in both modes every row aggregates pods owned by
+        statefulsets in the group. Supports filtering via a filter expression, custom
+        groupBy, ordering by cpu / cpu_request / cpu_limit / memory / memory_request
+        / memory_limit / desired_pods / current_pods, and pagination via offset/limit.
+        Also reports missing required metrics and whether the requested time range
+        falls before the data retention boundary. Numeric metric fields (statefulSetCPU,
+        statefulSetCPURequest, statefulSetCPULimit, statefulSetMemory, statefulSetMemoryRequest,
+        statefulSetMemoryLimit, desiredPods, currentPods) return -1 as a sentinel
+        when no data is available for that field.'
+      operationId: ListStatefulSets
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/InframonitoringtypesPostableStatefulSets'
+      responses:
+        "200":
+          content:
+            application/json:
+              schema:
+                properties:
+                  data:
+                    $ref: '#/components/schemas/InframonitoringtypesStatefulSets'
+                  status:
+                    type: string
+                required:
+                - status
+                - data
+                type: object
+          description: OK
+        "400":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Bad Request
+        "401":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Unauthorized
+        "403":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Forbidden
+        "500":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Internal Server Error
+      security:
+      - api_key:
+        - VIEWER
+      - tokenizer:
+        - VIEWER
+      summary: List StatefulSets for Infra Monitoring
+      tags:
+      - inframonitoring
   /api/v2/livez:
     get:
       deprecated: false

frontend/src/api/generated/services/inframonitoring/index.ts

@@ -18,6 +18,7 @@ import type {
 	InframonitoringtypesPostableNamespacesDTO,
 	InframonitoringtypesPostableNodesDTO,
 	InframonitoringtypesPostablePodsDTO,
+	InframonitoringtypesPostableStatefulSetsDTO,
 	InframonitoringtypesPostableVolumesDTO,
 	ListClusters200,
 	ListDeployments200,
@@ -25,6 +26,7 @@ import type {
 	ListNamespaces200,
 	ListNodes200,
 	ListPods200,
+	ListStatefulSets200,
 	ListVolumes200,
 	RenderErrorResponseDTO,
 } from '../sigNoz.schemas';
@@ -620,3 +622,87 @@ export const useListVolumes = <
 
 	return useMutation(mutationOptions);
 };
+/**
+ * Returns a paginated list of Kubernetes StatefulSets with key aggregated pod metrics: CPU usage and memory working set summed across pods owned by the statefulset, plus average CPU/memory request and limit utilization (statefulSetCPURequest, statefulSetCPULimit, statefulSetMemoryRequest, statefulSetMemoryLimit). Each row also reports the latest known desiredPods (k8s.statefulset.desired_pods) and currentPods (k8s.statefulset.current_pods) replica counts and per-group podCountsByPhase ({ pending, running, succeeded, failed, unknown } from each pod's latest k8s.pod.phase value). Each statefulset includes metadata attributes (k8s.statefulset.name, k8s.namespace.name, k8s.cluster.name). The response type is 'list' for the default k8s.statefulset.name grouping or 'grouped_list' for custom groupBy keys; in both modes every row aggregates pods owned by statefulsets in the group. Supports filtering via a filter expression, custom groupBy, ordering by cpu / cpu_request / cpu_limit / memory / memory_request / memory_limit / desired_pods / current_pods, and pagination via offset/limit. Also reports missing required metrics and whether the requested time range falls before the data retention boundary. Numeric metric fields (statefulSetCPU, statefulSetCPURequest, statefulSetCPULimit, statefulSetMemory, statefulSetMemoryRequest, statefulSetMemoryLimit, desiredPods, currentPods) return -1 as a sentinel when no data is available for that field.
+ * @summary List StatefulSets for Infra Monitoring
+ */
+export const listStatefulSets = (
+	inframonitoringtypesPostableStatefulSetsDTO: BodyType<InframonitoringtypesPostableStatefulSetsDTO>,
+	signal?: AbortSignal,
+) => {
+	return GeneratedAPIInstance<ListStatefulSets200>({
+		url: `/api/v2/infra_monitoring/statefulsets`,
+		method: 'POST',
+		headers: { 'Content-Type': 'application/json' },
+		data: inframonitoringtypesPostableStatefulSetsDTO,
+		signal,
+	});
+};
+
+export const getListStatefulSetsMutationOptions = <
+	TError = ErrorType<RenderErrorResponseDTO>,
+	TContext = unknown,
+>(options?: {
+	mutation?: UseMutationOptions<
+		Awaited<ReturnType<typeof listStatefulSets>>,
+		TError,
+		{ data: BodyType<InframonitoringtypesPostableStatefulSetsDTO> },
+		TContext
+	>;
+}): UseMutationOptions<
+	Awaited<ReturnType<typeof listStatefulSets>>,
+	TError,
+	{ data: BodyType<InframonitoringtypesPostableStatefulSetsDTO> },
+	TContext
+> => {
+	const mutationKey = ['listStatefulSets'];
+	const { mutation: mutationOptions } = options
+		? options.mutation &&
+			'mutationKey' in options.mutation &&
+			options.mutation.mutationKey
+			? options
+			: { ...options, mutation: { ...options.mutation, mutationKey } }
+		: { mutation: { mutationKey } };
+
+	const mutationFn: MutationFunction<
+		Awaited<ReturnType<typeof listStatefulSets>>,
+		{ data: BodyType<InframonitoringtypesPostableStatefulSetsDTO> }
+	> = (props) => {
+		const { data } = props ?? {};
+
+		return listStatefulSets(data);
+	};
+
+	return { mutationFn, ...mutationOptions };
+};
+
+export type ListStatefulSetsMutationResult = NonNullable<
+	Awaited<ReturnType<typeof listStatefulSets>>
+>;
+export type ListStatefulSetsMutationBody =
+	BodyType<InframonitoringtypesPostableStatefulSetsDTO>;
+export type ListStatefulSetsMutationError = ErrorType<RenderErrorResponseDTO>;
+
+/**
+ * @summary List StatefulSets for Infra Monitoring
+ */
+export const useListStatefulSets = <
+	TError = ErrorType<RenderErrorResponseDTO>,
+	TContext = unknown,
+>(options?: {
+	mutation?: UseMutationOptions<
+		Awaited<ReturnType<typeof listStatefulSets>>,
+		TError,
+		{ data: BodyType<InframonitoringtypesPostableStatefulSetsDTO> },
+		TContext
+	>;
+}): UseMutationResult<
+	Awaited<ReturnType<typeof listStatefulSets>>,
+	TError,
+	{ data: BodyType<InframonitoringtypesPostableStatefulSetsDTO> },
+	TContext
+> => {
+	const mutationOptions = getListStatefulSetsMutationOptions(options);
+
+	return useMutation(mutationOptions);
+};

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

@@ -5190,6 +5190,34 @@ export interface InframonitoringtypesPostablePodsDTO {
 	start: number;
 }
 
+export interface InframonitoringtypesPostableStatefulSetsDTO {
+	/**
+	 * @type integer
+	 * @format int64
+	 */
+	end: number;
+	filter?: Querybuildertypesv5FilterDTO;
+	/**
+	 * @type array
+	 * @nullable true
+	 */
+	groupBy?: Querybuildertypesv5GroupByKeyDTO[] | null;
+	/**
+	 * @type integer
+	 */
+	limit: number;
+	/**
+	 * @type integer
+	 */
+	offset?: number;
+	orderBy?: Querybuildertypesv5OrderByDTO;
+	/**
+	 * @type integer
+	 * @format int64
+	 */
+	start: number;
+}
+
 export interface InframonitoringtypesPostableVolumesDTO {
 	/**
 	 * @type integer
@@ -5230,6 +5258,83 @@ export enum InframonitoringtypesResponseTypeDTO {
 	list = 'list',
 	grouped_list = 'grouped_list',
 }
+/**
+ * @nullable
+ */
+export type InframonitoringtypesStatefulSetRecordDTOMeta = {
+	[key: string]: string;
+} | null;
+
+export interface InframonitoringtypesStatefulSetRecordDTO {
+	/**
+	 * @type integer
+	 */
+	currentPods: number;
+	/**
+	 * @type integer
+	 */
+	desiredPods: number;
+	/**
+	 * @type object
+	 * @nullable true
+	 */
+	meta: InframonitoringtypesStatefulSetRecordDTOMeta;
+	podCountsByPhase: InframonitoringtypesPodCountsByPhaseDTO;
+	/**
+	 * @type number
+	 * @format double
+	 */
+	statefulSetCPU: number;
+	/**
+	 * @type number
+	 * @format double
+	 */
+	statefulSetCPULimit: number;
+	/**
+	 * @type number
+	 * @format double
+	 */
+	statefulSetCPURequest: number;
+	/**
+	 * @type number
+	 * @format double
+	 */
+	statefulSetMemory: number;
+	/**
+	 * @type number
+	 * @format double
+	 */
+	statefulSetMemoryLimit: number;
+	/**
+	 * @type number
+	 * @format double
+	 */
+	statefulSetMemoryRequest: number;
+	/**
+	 * @type string
+	 */
+	statefulSetName: string;
+}
+
+export interface InframonitoringtypesStatefulSetsDTO {
+	/**
+	 * @type boolean
+	 */
+	endTimeBeforeRetention: boolean;
+	/**
+	 * @type array
+	 * @nullable true
+	 */
+	records: InframonitoringtypesStatefulSetRecordDTO[] | null;
+	requiredMetricsCheck: InframonitoringtypesRequiredMetricsCheckDTO;
+	/**
+	 * @type integer
+	 */
+	total: number;
+	type: InframonitoringtypesResponseTypeDTO;
+	warning?: Querybuildertypesv5QueryWarnDataDTO;
+}
+
 /**
  * @nullable
  */
@@ -9662,6 +9767,14 @@ export type ListVolumes200 = {
 	status: string;
 };
 
+export type ListStatefulSets200 = {
+	data: InframonitoringtypesStatefulSetsDTO;
+	/**
+	 * @type string
+	 */
+	status: string;
+};
+
 export type Livez200 = {
 	data: FactoryResponseDTO;
 	/**

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

@@ -437,11 +437,16 @@ export function convertTraceOperatorToV5(
 				panelType,
 			);
 
-			// Skip aggregation for raw request type
+			// Skip aggregation for raw request type. Force dataSource to traces so
+			// createAggregation never takes the metrics branch (which would emit a
+			// metricName field the backend rejects for trace operators).
 			const aggregations =
 				requestType === 'raw'
 					? undefined
-					: createAggregation(traceOperatorData, panelType);
+					: createAggregation(
+							{ ...traceOperatorData, dataSource: DataSource.TRACES },
+							panelType,
+						);
 
 			const spec: QueryEnvelope['spec'] = {
 				name: queryName,

frontend/src/components/CustomTimePicker/CustomTimePicker.tsx

@@ -596,6 +596,7 @@ function CustomTimePicker({
 				>
 					<Input
 						ref={inputRef}
+						autoComplete="off"
 						className={cx(
 							'timeSelection-input',
 							inputStatus === CustomTimePickerInputStatus.ERROR ? 'error' : '',

frontend/src/container/CreateAlertV2/Footer/utils.tsx

@@ -220,6 +220,7 @@ export function buildCreateThresholdAlertRulePayload(
 		builderQueries: {
 			...mapQueryDataToApi(query.builder.queryData, 'queryName').data,
 			...mapQueryDataToApi(query.builder.queryFormulas, 'queryName').data,
+			...mapQueryDataToApi(query.builder.queryTraceOperator, 'queryName').data,
 		},
 		promQueries: mapQueryDataToApi(query.promql, 'name').data,
 		chQueries: mapQueryDataToApi(query.clickhouse_sql, 'name').data,

frontend/src/container/ListAlertRules/ToggleAlertState.tsx

@@ -1,6 +1,8 @@
 import { Dispatch, SetStateAction, useState } from 'react';
+import { useQueryClient } from 'react-query';
 import { patchRulePartial } from 'api/alerts/patchRulePartial';
 import { convertToApiError } from 'api/ErrorResponseHandlerForGeneratedAPIs';
+import { invalidateGetRuleByID } from 'api/generated/services/rules';
 import type {
 	RenderErrorResponseDTO,
 	RuletypesRuleDTO,
@@ -28,6 +30,7 @@ function ToggleAlertState({
 
 	const { notifications } = useNotifications();
 	const { showErrorModal } = useErrorModal();
+	const queryClient = useQueryClient();
 
 	const onToggleHandler = async (
 		id: string,
@@ -60,6 +63,9 @@ function ToggleAlertState({
 				loading: false,
 				payload: updatedRule,
 			}));
+
+			invalidateGetRuleByID(queryClient, { id });
+
 			notifications.success({
 				message: 'Success',
 			});

frontend/src/lib/newQueryBuilder/queryBuilderMappers/mapQueryDataFromApi.ts

@@ -53,7 +53,9 @@ const mapQueryFromV5 = (compositeQuery: ICompositeMetricQuery): Query => {
 			}
 		} else if (q.type === 'builder_trace_operator') {
 			if (spec.name) {
-				builderQueries[spec.name] = spec as unknown as IBuilderTraceOperator;
+				builderQueries[spec.name] = convertBuilderQueryToIBuilderQuery(
+					spec as BuilderQuery,
+				) as IBuilderTraceOperator;
 				builderQueryTypes[spec.name] = 'builder_trace_operator';
 			}
 		} else if (q.type === 'promql') {

frontend/src/pages/AlertDetails/AlertHeader/AlertState/AlertState.tsx

@@ -17,7 +17,7 @@ export default function AlertState({
 	let label;
 	const isDarkMode = useIsDarkMode();
 	switch (state) {
-		case 'no-data':
+		case 'nodata':
 			icon = (
 				<CircleOff
 					size={18}

frontend/src/pages/AlertDetails/hooks.tsx

@@ -12,6 +12,7 @@ import { convertToApiError } from 'api/ErrorResponseHandlerForGeneratedAPIs';
 import {
 	createRule,
 	deleteRuleByID,
+	invalidateGetRuleByID,
 	updateRuleByID,
 	useGetRuleByID,
 	useListRules,
@@ -408,6 +409,7 @@ export const useAlertRuleStatusToggle = ({
 		{
 			onSuccess: (data) => {
 				setAlertRuleState(data.data.state);
+				invalidateGetRuleByID(queryClient, { id: ruleId });
 				queryClient.refetchQueries([REACT_QUERY_KEY.ALERT_RULE_DETAILS, ruleId]);
 				notifications.success({
 					message: `Alert has been ${
@@ -416,6 +418,7 @@ export const useAlertRuleStatusToggle = ({
 				});
 			},
 			onError: (error) => {
+				invalidateGetRuleByID(queryClient, { id: ruleId });
 				queryClient.refetchQueries([REACT_QUERY_KEY.ALERT_RULE_DETAILS, ruleId]);
 				showErrorModal(
 					convertToApiError(error as AxiosError<RenderErrorResponseDTO>) as APIError,

frontend/src/types/api/alerts/def.ts

@@ -109,7 +109,8 @@ export type AlertRuleTimelineTableResponsePayload = {
 		labels: AlertLabelsProps['labels'];
 	};
 };
-type AlertState = 'firing' | 'normal' | 'no-data' | 'muted';
+
+type AlertState = 'firing' | 'normal' | 'nodata' | 'muted';
 
 export interface AlertRuleTimelineGraphResponse {
 	start: number;

go.mod

@@ -11,7 +11,6 @@ require (
 	github.com/SigNoz/signoz-otel-collector v0.144.3
 	github.com/antlr4-go/antlr/v4 v4.13.1
 	github.com/antonmedv/expr v1.15.3
-	github.com/bytedance/sonic v1.14.1
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/coreos/go-oidc/v3 v3.17.0
 	github.com/dgraph-io/ristretto/v2 v2.3.0
@@ -113,6 +112,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.6 // indirect
 	github.com/aws/smithy-go v1.24.2 // indirect
 	github.com/bytedance/gopkg v0.1.3 // indirect
+	github.com/bytedance/sonic v1.14.1 // indirect
 	github.com/bytedance/sonic/loader v0.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // indirect

pkg/apiserver/signozapiserver/inframonitoring.go

@@ -143,5 +143,24 @@ func (provider *provider) addInfraMonitoringRoutes(router *mux.Router) error {
 		return err
 	}
 
+	if err := router.Handle("/api/v2/infra_monitoring/statefulsets", handler.New(
+		provider.authzMiddleware.ViewAccess(provider.infraMonitoringHandler.ListStatefulSets),
+		handler.OpenAPIDef{
+			ID:                  "ListStatefulSets",
+			Tags:                []string{"inframonitoring"},
+			Summary:             "List StatefulSets for Infra Monitoring",
+			Description:         "Returns a paginated list of Kubernetes StatefulSets with key aggregated pod metrics: CPU usage and memory working set summed across pods owned by the statefulset, plus average CPU/memory request and limit utilization (statefulSetCPURequest, statefulSetCPULimit, statefulSetMemoryRequest, statefulSetMemoryLimit). Each row also reports the latest known desiredPods (k8s.statefulset.desired_pods) and currentPods (k8s.statefulset.current_pods) replica counts and per-group podCountsByPhase ({ pending, running, succeeded, failed, unknown } from each pod's latest k8s.pod.phase value). Each statefulset includes metadata attributes (k8s.statefulset.name, k8s.namespace.name, k8s.cluster.name). The response type is 'list' for the default k8s.statefulset.name grouping or 'grouped_list' for custom groupBy keys; in both modes every row aggregates pods owned by statefulsets in the group. Supports filtering via a filter expression, custom groupBy, ordering by cpu / cpu_request / cpu_limit / memory / memory_request / memory_limit / desired_pods / current_pods, and pagination via offset/limit. Also reports missing required metrics and whether the requested time range falls before the data retention boundary. Numeric metric fields (statefulSetCPU, statefulSetCPURequest, statefulSetCPULimit, statefulSetMemory, statefulSetMemoryRequest, statefulSetMemoryLimit, desiredPods, currentPods) return -1 as a sentinel when no data is available for that field.",
+			Request:             new(inframonitoringtypes.PostableStatefulSets),
+			RequestContentType:  "application/json",
+			Response:            new(inframonitoringtypes.StatefulSets),
+			ResponseContentType: "application/json",
+			SuccessStatusCode:   http.StatusOK,
+			ErrorStatusCodes:    []int{http.StatusBadRequest, http.StatusUnauthorized},
+			Deprecated:          false,
+			SecuritySchemes:     newSecuritySchemes(types.RoleViewer),
+		})).Methods(http.MethodPost).GetError(); err != nil {
+		return err
+	}
+
 	return nil
 }

pkg/modules/inframonitoring/implinframonitoring/handler.go

@@ -189,3 +189,27 @@ func (h *handler) ListDeployments(rw http.ResponseWriter, req *http.Request) {
 
 	render.Success(rw, http.StatusOK, result)
 }
+
+func (h *handler) ListStatefulSets(rw http.ResponseWriter, req *http.Request) {
+	claims, err := authtypes.ClaimsFromContext(req.Context())
+	if err != nil {
+		render.Error(rw, err)
+		return
+	}
+
+	orgID := valuer.MustNewUUID(claims.OrgID)
+
+	var parsedReq inframonitoringtypes.PostableStatefulSets
+	if err := binding.JSON.BindBody(req.Body, &parsedReq); err != nil {
+		render.Error(rw, err)
+		return
+	}
+
+	result, err := h.module.ListStatefulSets(req.Context(), orgID, &parsedReq)
+	if err != nil {
+		render.Error(rw, err)
+		return
+	}
+
+	render.Success(rw, http.StatusOK, result)
+}

pkg/modules/inframonitoring/implinframonitoring/module.go

@@ -706,3 +706,100 @@ func (m *module) ListDeployments(ctx context.Context, orgID valuer.UUID, req *in
 
 	return resp, nil
 }
+
+func (m *module) ListStatefulSets(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableStatefulSets) (*inframonitoringtypes.StatefulSets, error) {
+	if err := req.Validate(); err != nil {
+		return nil, err
+	}
+
+	resp := &inframonitoringtypes.StatefulSets{}
+
+	if req.OrderBy == nil {
+		req.OrderBy = &qbtypes.OrderBy{
+			Key: qbtypes.OrderByKey{
+				TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+					Name: inframonitoringtypes.StatefulSetsOrderByCPU,
+				},
+			},
+			Direction: qbtypes.OrderDirectionDesc,
+		}
+	}
+
+	if len(req.GroupBy) == 0 {
+		req.GroupBy = []qbtypes.GroupByKey{statefulSetNameGroupByKey}
+		resp.Type = inframonitoringtypes.ResponseTypeList
+	} else {
+		resp.Type = inframonitoringtypes.ResponseTypeGroupedList
+	}
+
+	// Bake the workload base filter into req.Filter so all downstream helpers pick it up.
+	if req.Filter == nil {
+		req.Filter = &qbtypes.Filter{}
+	}
+	req.Filter.Expression = mergeFilterExpressions(statefulSetsBaseFilterExpr, req.Filter.Expression)
+
+	missingMetrics, minFirstReportedUnixMilli, err := m.getMetricsExistenceAndEarliestTime(ctx, statefulSetsTableMetricNamesList)
+	if err != nil {
+		return nil, err
+	}
+	if len(missingMetrics) > 0 {
+		resp.RequiredMetricsCheck = inframonitoringtypes.RequiredMetricsCheck{MissingMetrics: missingMetrics}
+		resp.Records = []inframonitoringtypes.StatefulSetRecord{}
+		resp.Total = 0
+		return resp, nil
+	}
+	if req.End < int64(minFirstReportedUnixMilli) {
+		resp.EndTimeBeforeRetention = true
+		resp.Records = []inframonitoringtypes.StatefulSetRecord{}
+		resp.Total = 0
+		return resp, nil
+	}
+	resp.RequiredMetricsCheck = inframonitoringtypes.RequiredMetricsCheck{MissingMetrics: []string{}}
+
+	metadataMap, err := m.getStatefulSetsTableMetadata(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	resp.Total = len(metadataMap)
+
+	pageGroups, err := m.getTopStatefulSetGroups(ctx, orgID, req, metadataMap)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(pageGroups) == 0 {
+		resp.Records = []inframonitoringtypes.StatefulSetRecord{}
+		return resp, nil
+	}
+
+	filterExpr := ""
+	if req.Filter != nil {
+		filterExpr = req.Filter.Expression
+	}
+
+	fullQueryReq := buildFullQueryRequest(req.Start, req.End, filterExpr, req.GroupBy, pageGroups, m.newStatefulSetsTableListQuery())
+	queryResp, err := m.querier.QueryRange(ctx, orgID, fullQueryReq)
+	if err != nil {
+		return nil, err
+	}
+
+	// Reuse the pods phase-counts CTE function via a temp struct — it reads only
+	// Start/End/Filter/GroupBy from PostablePods. Pods owned by a StatefulSet carry
+	// k8s.statefulset.name as a resource attribute, so default-groupBy gives
+	// per-statefulset phase counts automatically.
+	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, &inframonitoringtypes.PostablePods{
+		Start:   req.Start,
+		End:     req.End,
+		Filter:  req.Filter,
+		GroupBy: req.GroupBy,
+	}, pageGroups)
+	if err != nil {
+		return nil, err
+	}
+
+	resp.Records = buildStatefulSetRecords(queryResp, pageGroups, req.GroupBy, metadataMap, phaseCounts)
+	resp.Warning = queryResp.Warning
+
+	return resp, nil
+}

pkg/modules/inframonitoring/implinframonitoring/statefulsets.go

@@ -0,0 +1,148 @@
+package implinframonitoring
+
+import (
+	"context"
+	"slices"
+
+	"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// buildStatefulSetRecords assembles the page records. Pod phase counts come from
+// phaseCounts in both modes; every row is a group of pods (one statefulset in
+// list mode, an arbitrary roll-up in grouped_list mode), so there's no
+// per-row "current phase" concept.
+func buildStatefulSetRecords(
+	resp *qbtypes.QueryRangeResponse,
+	pageGroups []map[string]string,
+	groupBy []qbtypes.GroupByKey,
+	metadataMap map[string]map[string]string,
+	phaseCounts map[string]podPhaseCounts,
+) []inframonitoringtypes.StatefulSetRecord {
+	metricsMap := parseFullQueryResponse(resp, groupBy)
+
+	records := make([]inframonitoringtypes.StatefulSetRecord, 0, len(pageGroups))
+	for _, labels := range pageGroups {
+		compositeKey := compositeKeyFromLabels(labels, groupBy)
+		statefulSetName := labels[statefulSetNameAttrKey]
+
+		record := inframonitoringtypes.StatefulSetRecord{ // initialize with default values
+			StatefulSetName:          statefulSetName,
+			StatefulSetCPU:           -1,
+			StatefulSetCPURequest:    -1,
+			StatefulSetCPULimit:      -1,
+			StatefulSetMemory:        -1,
+			StatefulSetMemoryRequest: -1,
+			StatefulSetMemoryLimit:   -1,
+			DesiredPods:              -1,
+			CurrentPods:              -1,
+			Meta:                     map[string]string{},
+		}
+
+		if metrics, ok := metricsMap[compositeKey]; ok {
+			if v, exists := metrics["A"]; exists {
+				record.StatefulSetCPU = v
+			}
+			if v, exists := metrics["B"]; exists {
+				record.StatefulSetCPURequest = v
+			}
+			if v, exists := metrics["C"]; exists {
+				record.StatefulSetCPULimit = v
+			}
+			if v, exists := metrics["D"]; exists {
+				record.StatefulSetMemory = v
+			}
+			if v, exists := metrics["E"]; exists {
+				record.StatefulSetMemoryRequest = v
+			}
+			if v, exists := metrics["F"]; exists {
+				record.StatefulSetMemoryLimit = v
+			}
+			if v, exists := metrics["H"]; exists {
+				record.DesiredPods = int(v)
+			}
+			if v, exists := metrics["I"]; exists {
+				record.CurrentPods = int(v)
+			}
+		}
+
+		if phaseCountsForGroup, ok := phaseCounts[compositeKey]; ok {
+			record.PodCountsByPhase = inframonitoringtypes.PodCountsByPhase{
+				Pending:   phaseCountsForGroup.Pending,
+				Running:   phaseCountsForGroup.Running,
+				Succeeded: phaseCountsForGroup.Succeeded,
+				Failed:    phaseCountsForGroup.Failed,
+				Unknown:   phaseCountsForGroup.Unknown,
+			}
+		}
+
+		if attrs, ok := metadataMap[compositeKey]; ok {
+			for k, v := range attrs {
+				record.Meta[k] = v
+			}
+		}
+
+		records = append(records, record)
+	}
+	return records
+}
+
+func (m *module) getTopStatefulSetGroups(
+	ctx context.Context,
+	orgID valuer.UUID,
+	req *inframonitoringtypes.PostableStatefulSets,
+	metadataMap map[string]map[string]string,
+) ([]map[string]string, error) {
+	orderByKey := req.OrderBy.Key.Name
+	queryNamesForOrderBy := orderByToStatefulSetsQueryNames[orderByKey]
+	rankingQueryName := queryNamesForOrderBy[len(queryNamesForOrderBy)-1]
+
+	topReq := &qbtypes.QueryRangeRequest{
+		Start:       uint64(req.Start),
+		End:         uint64(req.End),
+		RequestType: qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{
+			Queries: make([]qbtypes.QueryEnvelope, 0, len(queryNamesForOrderBy)),
+		},
+	}
+
+	for _, envelope := range m.newStatefulSetsTableListQuery().CompositeQuery.Queries {
+		if !slices.Contains(queryNamesForOrderBy, envelope.GetQueryName()) {
+			continue
+		}
+		copied := envelope
+		if copied.Type == qbtypes.QueryTypeBuilder {
+			existingExpr := ""
+			if f := copied.GetFilter(); f != nil {
+				existingExpr = f.Expression
+			}
+			reqFilterExpr := ""
+			if req.Filter != nil {
+				reqFilterExpr = req.Filter.Expression
+			}
+			merged := mergeFilterExpressions(existingExpr, reqFilterExpr)
+			copied.SetFilter(&qbtypes.Filter{Expression: merged})
+			copied.SetGroupBy(req.GroupBy)
+		}
+		topReq.CompositeQuery.Queries = append(topReq.CompositeQuery.Queries, copied)
+	}
+
+	resp, err := m.querier.QueryRange(ctx, orgID, topReq)
+	if err != nil {
+		return nil, err
+	}
+
+	allMetricGroups := parseAndSortGroups(resp, rankingQueryName, req.GroupBy, req.OrderBy.Direction)
+	return paginateWithBackfill(allMetricGroups, metadataMap, req.GroupBy, req.Offset, req.Limit), nil
+}
+
+func (m *module) getStatefulSetsTableMetadata(ctx context.Context, req *inframonitoringtypes.PostableStatefulSets) (map[string]map[string]string, error) {
+	var nonGroupByAttrs []string
+	for _, key := range statefulSetAttrKeysForMetadata {
+		if !isKeyInGroupByAttrs(req.GroupBy, key) {
+			nonGroupByAttrs = append(nonGroupByAttrs, key)
+		}
+	}
+	return m.getMetadata(ctx, statefulSetsTableMetricNamesList, req.GroupBy, nonGroupByAttrs, req.Filter, req.Start, req.End)
+}

pkg/modules/inframonitoring/implinframonitoring/statefulsets_constants.go

@@ -0,0 +1,254 @@
+package implinframonitoring
+
+import (
+	"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+const (
+	statefulSetNameAttrKey     = "k8s.statefulset.name"
+	statefulSetsBaseFilterExpr = "k8s.statefulset.name != ''"
+)
+
+var statefulSetNameGroupByKey = qbtypes.GroupByKey{
+	TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+		Name:          statefulSetNameAttrKey,
+		FieldContext:  telemetrytypes.FieldContextResource,
+		FieldDataType: telemetrytypes.FieldDataTypeString,
+	},
+}
+
+// statefulSetsTableMetricNamesList drives the existence/retention check.
+// Includes k8s.pod.phase even though phase isn't part of the QB composite query —
+// it is queried separately via getPerGroupPodPhaseCounts, and we want the
+// response to short-circuit cleanly when the phase metric is absent.
+var statefulSetsTableMetricNamesList = []string{
+	"k8s.pod.phase",
+	"k8s.pod.cpu.usage",
+	"k8s.pod.cpu_request_utilization",
+	"k8s.pod.cpu_limit_utilization",
+	"k8s.pod.memory.working_set",
+	"k8s.pod.memory_request_utilization",
+	"k8s.pod.memory_limit_utilization",
+	"k8s.statefulset.desired_pods",
+	"k8s.statefulset.current_pods",
+}
+
+// Carried forward from v1 statefulSetAttrsToEnrich
+// (pkg/query-service/app/inframetrics/statefulsets.go:29-33).
+var statefulSetAttrKeysForMetadata = []string{
+	"k8s.statefulset.name",
+	"k8s.namespace.name",
+	"k8s.cluster.name",
+}
+
+// orderByToStatefulSetsQueryNames maps the orderBy column to the query name
+// used for ranking statefulset groups. v2 B/C/E/F are direct metrics, no
+// formula deps — so unlike v1 we don't carry A/D.
+var orderByToStatefulSetsQueryNames = map[string][]string{
+	inframonitoringtypes.StatefulSetsOrderByCPU:           {"A"},
+	inframonitoringtypes.StatefulSetsOrderByCPURequest:    {"B"},
+	inframonitoringtypes.StatefulSetsOrderByCPULimit:      {"C"},
+	inframonitoringtypes.StatefulSetsOrderByMemory:        {"D"},
+	inframonitoringtypes.StatefulSetsOrderByMemoryRequest: {"E"},
+	inframonitoringtypes.StatefulSetsOrderByMemoryLimit:   {"F"},
+	inframonitoringtypes.StatefulSetsOrderByDesiredPods:   {"H"},
+	inframonitoringtypes.StatefulSetsOrderByCurrentPods:   {"I"},
+}
+
+// newStatefulSetsTableListQuery builds the composite QB v5 request for the statefulsets list.
+// Eight builder queries: A..F roll up pod-level metrics by statefulset, H/I take the
+// latest statefulset-level desired/current counts. Restarts (v1 query G) is intentionally
+// omitted to match the v2 pods/deployments pattern.
+//
+// Every builder query carries a base filter `k8s.statefulset.name != ”`.
+// Reason: pod-level metrics (A..F) are emitted for every pod regardless of whether the
+// pod belongs to a StatefulSet; only StatefulSet-owned pods carry the
+// `k8s.statefulset.name` resource attribute. Without this filter, standalone pods and
+// pods owned by other workloads (Deployment/DaemonSet/Job/...) collapse into a single
+// empty-string group under the default groupBy. v1's GetStatefulSetList applied the same
+// filter via FilterOperatorExists; this matches v1 parity. The base filter merges
+// cleanly with user filters via mergeFilterExpressions / buildFullQueryRequest.
+func (m *module) newStatefulSetsTableListQuery() *qbtypes.QueryRangeRequest {
+	queries := []qbtypes.QueryEnvelope{
+		// Query A: k8s.pod.cpu.usage — sum of pod CPU within the group.
+		{
+			Type: qbtypes.QueryTypeBuilder,
+			Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Name:   "A",
+				Signal: telemetrytypes.SignalMetrics,
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "k8s.pod.cpu.usage",
+						TimeAggregation:  metrictypes.TimeAggregationAvg,
+						SpaceAggregation: metrictypes.SpaceAggregationSum,
+						ReduceTo:         qbtypes.ReduceToAvg,
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: statefulSetsBaseFilterExpr,
+				},
+				GroupBy:  []qbtypes.GroupByKey{statefulSetNameGroupByKey},
+				Disabled: false,
+			},
+		},
+		// Query B: k8s.pod.cpu_request_utilization — avg across pods in the group.
+		{
+			Type: qbtypes.QueryTypeBuilder,
+			Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Name:   "B",
+				Signal: telemetrytypes.SignalMetrics,
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "k8s.pod.cpu_request_utilization",
+						TimeAggregation:  metrictypes.TimeAggregationAvg,
+						SpaceAggregation: metrictypes.SpaceAggregationAvg,
+						ReduceTo:         qbtypes.ReduceToAvg,
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: statefulSetsBaseFilterExpr,
+				},
+				GroupBy:  []qbtypes.GroupByKey{statefulSetNameGroupByKey},
+				Disabled: false,
+			},
+		},
+		// Query C: k8s.pod.cpu_limit_utilization — avg across pods in the group.
+		{
+			Type: qbtypes.QueryTypeBuilder,
+			Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Name:   "C",
+				Signal: telemetrytypes.SignalMetrics,
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "k8s.pod.cpu_limit_utilization",
+						TimeAggregation:  metrictypes.TimeAggregationAvg,
+						SpaceAggregation: metrictypes.SpaceAggregationAvg,
+						ReduceTo:         qbtypes.ReduceToAvg,
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: statefulSetsBaseFilterExpr,
+				},
+				GroupBy:  []qbtypes.GroupByKey{statefulSetNameGroupByKey},
+				Disabled: false,
+			},
+		},
+		// Query D: k8s.pod.memory.working_set — sum of pod memory within the group.
+		{
+			Type: qbtypes.QueryTypeBuilder,
+			Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Name:   "D",
+				Signal: telemetrytypes.SignalMetrics,
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "k8s.pod.memory.working_set",
+						TimeAggregation:  metrictypes.TimeAggregationAvg,
+						SpaceAggregation: metrictypes.SpaceAggregationSum,
+						ReduceTo:         qbtypes.ReduceToAvg,
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: statefulSetsBaseFilterExpr,
+				},
+				GroupBy:  []qbtypes.GroupByKey{statefulSetNameGroupByKey},
+				Disabled: false,
+			},
+		},
+		// Query E: k8s.pod.memory_request_utilization — avg across pods in the group.
+		{
+			Type: qbtypes.QueryTypeBuilder,
+			Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Name:   "E",
+				Signal: telemetrytypes.SignalMetrics,
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "k8s.pod.memory_request_utilization",
+						TimeAggregation:  metrictypes.TimeAggregationAvg,
+						SpaceAggregation: metrictypes.SpaceAggregationAvg,
+						ReduceTo:         qbtypes.ReduceToAvg,
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: statefulSetsBaseFilterExpr,
+				},
+				GroupBy:  []qbtypes.GroupByKey{statefulSetNameGroupByKey},
+				Disabled: false,
+			},
+		},
+		// Query F: k8s.pod.memory_limit_utilization — avg across pods in the group.
+		{
+			Type: qbtypes.QueryTypeBuilder,
+			Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Name:   "F",
+				Signal: telemetrytypes.SignalMetrics,
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "k8s.pod.memory_limit_utilization",
+						TimeAggregation:  metrictypes.TimeAggregationAvg,
+						SpaceAggregation: metrictypes.SpaceAggregationAvg,
+						ReduceTo:         qbtypes.ReduceToAvg,
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: statefulSetsBaseFilterExpr,
+				},
+				GroupBy:  []qbtypes.GroupByKey{statefulSetNameGroupByKey},
+				Disabled: false,
+			},
+		},
+		// Query H: k8s.statefulset.desired_pods — latest known desired replica count per group.
+		// v1 used TimeAggregationAnyLast (v3) → mapped to TimeAggregationLatest in v5;
+		// SpaceAggregationSum + ReduceToLast preserve v1's "latest, summed across the group".
+		{
+			Type: qbtypes.QueryTypeBuilder,
+			Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Name:   "H",
+				Signal: telemetrytypes.SignalMetrics,
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "k8s.statefulset.desired_pods",
+						TimeAggregation:  metrictypes.TimeAggregationLatest,
+						SpaceAggregation: metrictypes.SpaceAggregationSum,
+						ReduceTo:         qbtypes.ReduceToLast,
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: statefulSetsBaseFilterExpr,
+				},
+				GroupBy:  []qbtypes.GroupByKey{statefulSetNameGroupByKey},
+				Disabled: false,
+			},
+		},
+		// Query I: k8s.statefulset.current_pods — latest known current replica count per group.
+		{
+			Type: qbtypes.QueryTypeBuilder,
+			Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Name:   "I",
+				Signal: telemetrytypes.SignalMetrics,
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "k8s.statefulset.current_pods",
+						TimeAggregation:  metrictypes.TimeAggregationLatest,
+						SpaceAggregation: metrictypes.SpaceAggregationSum,
+						ReduceTo:         qbtypes.ReduceToLast,
+					},
+				},
+				Filter: &qbtypes.Filter{
+					Expression: statefulSetsBaseFilterExpr,
+				},
+				GroupBy:  []qbtypes.GroupByKey{statefulSetNameGroupByKey},
+				Disabled: false,
+			},
+		},
+	}
+
+	return &qbtypes.QueryRangeRequest{
+		RequestType: qbtypes.RequestTypeScalar,
+		CompositeQuery: qbtypes.CompositeQuery{
+			Queries: queries,
+		},
+	}
+}

pkg/modules/inframonitoring/inframonitoring.go

@@ -16,6 +16,7 @@ type Handler interface {
 	ListClusters(http.ResponseWriter, *http.Request)
 	ListVolumes(http.ResponseWriter, *http.Request)
 	ListDeployments(http.ResponseWriter, *http.Request)
+	ListStatefulSets(http.ResponseWriter, *http.Request)
 }
 
 type Module interface {
@@ -26,4 +27,5 @@ type Module interface {
 	ListClusters(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableClusters) (*inframonitoringtypes.Clusters, error)
 	ListVolumes(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableVolumes) (*inframonitoringtypes.Volumes, error)
 	ListDeployments(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableDeployments) (*inframonitoringtypes.Deployments, error)
+	ListStatefulSets(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableStatefulSets) (*inframonitoringtypes.StatefulSets, error)
 }

pkg/modules/spanpercentile/implspanpercentile/module.go

@@ -52,7 +52,7 @@ func (m *module) GetSpanPercentile(ctx context.Context, orgID valuer.UUID, req *
 
 func transformToSpanPercentileResponse(queryResult *qbtypes.QueryRangeResponse) (*spanpercentiletypes.SpanPercentileResponse, error) {
 	if len(queryResult.Data.Results) == 0 {
-		return nil, errors.New(errors.TypeInternal, errors.CodeInternal, "no data returned from query")
+		return nil, errors.New(errors.TypeNotFound, errors.CodeNotFound, "no spans found matching the specified criteria")
 	}
 
 	scalarData, ok := queryResult.Data.Results[0].(*qbtypes.ScalarData)
@@ -61,7 +61,7 @@ func transformToSpanPercentileResponse(queryResult *qbtypes.QueryRangeResponse)
 	}
 
 	if len(scalarData.Data) == 0 {
-		return nil, errors.New(errors.TypeInternal, errors.CodeInternal, "no rows returned from query")
+		return nil, errors.New(errors.TypeNotFound, errors.CodeNotFound, "no spans found matching the specified criteria")
 	}
 
 	row := scalarData.Data[0]

pkg/querier/consume.go

@@ -12,10 +12,8 @@ import (
 	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
-	"github.com/SigNoz/signoz/pkg/errors"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-	"github.com/bytedance/sonic"
 )
 
 var (
@@ -24,8 +22,6 @@ var (
 	// written clickhouse query. The column alias indcate which value is
 	// to be considered as final result (or target).
 	legacyReservedColumnTargetAliases = []string{"__result", "__value", "result", "res", "value"}
-
-	CodeFailUnmarshalJSONColumn = errors.MustNewCode("fail_unmarshal_json_column")
 )
 
 // consume reads every row and shapes it into the payload expected for the
@@ -397,16 +393,11 @@ func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
 
 			// de-reference the typed pointer to any
 			val := reflect.ValueOf(cellPtr).Elem().Interface()
-			// Post-process JSON columns: unmarshal bytes into map[string]any
+			// Post-process JSON columns: normalize into String value
 			if strings.HasPrefix(strings.ToUpper(colTypes[i].DatabaseTypeName()), "JSON") {
 				switch x := val.(type) {
 				case []byte:
-					var m map[string]any
-					err := sonic.Unmarshal(x, &m)
-					if err != nil {
-						return nil, errors.WrapInternalf(err, CodeFailUnmarshalJSONColumn, "failed to unmarshal JSON column %s", name)
-					}
-					val = m
+					val = string(x)
 				default:
 					// already a structured type (map[string]any, []any, etc.)
 				}

pkg/querier/postprocess.go

@@ -12,12 +12,9 @@ import (
 	"github.com/SigNoz/govaluate"
 
 	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/flagger"
-	"github.com/SigNoz/signoz/pkg/types/featuretypes"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
 // queryInfo holds common query properties.
@@ -53,7 +50,7 @@ func getQueryName(spec any) string {
 	return getqueryInfo(spec).Name
 }
 
-func (q *querier) postProcessResults(ctx context.Context, orgID valuer.UUID, results map[string]any, req *qbtypes.QueryRangeRequest) (map[string]any, error) {
+func (q *querier) postProcessResults(ctx context.Context, results map[string]any, req *qbtypes.QueryRangeRequest) (map[string]any, error) {
 	// Convert results to typed format for processing
 	typedResults := make(map[string]*qbtypes.Result)
 	for name, result := range results {
@@ -72,7 +69,6 @@ func (q *querier) postProcessResults(ctx context.Context, orgID valuer.UUID, res
 		case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
 			if result, ok := typedResults[spec.Name]; ok {
 				result = postProcessBuilderQuery(q, result, spec, req)
-				result = q.postProcessLogBody(ctx, orgID, result, req)
 				typedResults[spec.Name] = result
 			}
 		case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
@@ -1049,33 +1045,3 @@ func (q *querier) calculateFormulaStep(expression string, req *qbtypes.QueryRang
 
 	return result
 }
-
-// postProcessLogBody removes the "message" key from the body map when it is empty.
-// Only runs for raw list queries with the use_json_body feature enabled.
-func (q *querier) postProcessLogBody(ctx context.Context, orgID valuer.UUID, result *qbtypes.Result, req *qbtypes.QueryRangeRequest) *qbtypes.Result {
-	if req.RequestType != qbtypes.RequestTypeRaw {
-		return result
-	}
-	if !q.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
-		return result
-	}
-	rawData, ok := result.Value.(*qbtypes.RawData)
-	if !ok {
-		return result
-	}
-	for _, row := range rawData.Rows {
-		bodyMap, ok := row.Data["body"].(map[string]any)
-		if !ok {
-			continue
-		}
-		if msg, exists := bodyMap["message"]; exists {
-			switch v := msg.(type) {
-			case string:
-				if v == "" {
-					delete(bodyMap, "message")
-				}
-			}
-		}
-	}
-	return result
-}

pkg/querier/querier.go

@@ -16,7 +16,6 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
-	"github.com/SigNoz/signoz/pkg/flagger"
 	"github.com/SigNoz/signoz/pkg/prometheus"
 	"github.com/SigNoz/signoz/pkg/query-service/utils"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
@@ -36,7 +35,6 @@ var (
 
 type querier struct {
 	logger                   *slog.Logger
-	fl                       flagger.Flagger
 	telemetryStore           telemetrystore.TelemetryStore
 	metadataStore            telemetrytypes.MetadataStore
 	promEngine               prometheus.Prometheus
@@ -64,12 +62,10 @@ func New(
 	meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder,
 	bucketCache BucketCache,
-	flagger flagger.Flagger,
 ) *querier {
 	querierSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/querier")
 	return &querier{
 		logger:                   querierSettings.Logger(),
-		fl:                       flagger,
 		telemetryStore:           telemetryStore,
 		metadataStore:            metadataStore,
 		promEngine:               promEngine,
@@ -688,7 +684,7 @@ func (q *querier) run(
 	}
 
 	gomaps.Copy(results, preseededResults)
-	processedResults, err := q.postProcessResults(ctx, orgID, results, req)
+	processedResults, err := q.postProcessResults(ctx, results, req)
 	if err != nil {
 		return nil, err
 	}

pkg/querier/querier_test.go

@@ -7,7 +7,6 @@ import (
 
 	cmock "github.com/srikanthccv/ClickHouse-go-mock"
 
-	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
 	"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
@@ -45,15 +44,14 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
 		providerSettings,
 		nil, // telemetryStore
 		metadataStore,
-		nil,                // prometheus
-		nil,                // traceStmtBuilder
-		nil,                // logStmtBuilder
-		nil,                // auditStmtBuilder
-		nil,                // metricStmtBuilder
-		nil,                // meterStmtBuilder
-		nil,                // traceOperatorStmtBuilder
-		nil,                // bucketCache
-		flaggertest.New(t), // flagger
+		nil, // prometheus
+		nil, // traceStmtBuilder
+		nil, // logStmtBuilder
+		nil, // auditStmtBuilder
+		nil, // metricStmtBuilder
+		nil, // meterStmtBuilder
+		nil, // traceOperatorStmtBuilder
+		nil, // bucketCache
 	)
 
 	req := &qbtypes.QueryRangeRequest{
@@ -118,7 +116,6 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 		nil,                      // meterStmtBuilder
 		nil,                      // traceOperatorStmtBuilder
 		nil,                      // bucketCache
-		flaggertest.New(t),       // flagger
 	)
 
 	req := &qbtypes.QueryRangeRequest{

pkg/querier/signozquerier/provider.go

@@ -186,6 +186,5 @@ func newProvider(
 		meterStmtBuilder,
 		traceOperatorStmtBuilder,
 		bucketCache,
-		flagger,
 	), nil
 }

pkg/query-service/model/alerting.go

@@ -134,7 +134,7 @@ type RuleStateHistory struct {
 	// One of ["normal", "firing"]
 	OverallState        AlertState `json:"overallState" ch:"overall_state"`
 	OverallStateChanged bool       `json:"overallStateChanged" ch:"overall_state_changed"`
-	// One of ["normal", "firing", "no_data", "muted"]
+	// One of ["normal", "firing", "nodata", "muted"]
 	State        AlertState   `json:"state" ch:"state"`
 	StateChanged bool         `json:"stateChanged" ch:"state_changed"`
 	UnixMilli    int64        `json:"unixMilli" ch:"unix_milli"`

pkg/query-service/rules/setups_test.go

@@ -53,7 +53,6 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		nil, // meterStmtBuilder
 		nil, // traceOperatorStmtBuilder
 		nil, // bucketCache
-		flagger,
 	), metadataStore
 }
 
@@ -103,7 +102,6 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
 		nil,            // meterStmtBuilder
 		nil,            // traceOperatorStmtBuilder
 		nil,            // bucketCache
-		fl,
 	)
 }
 
@@ -148,6 +146,5 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
 		nil,              // meterStmtBuilder
 		nil,              // traceOperatorStmtBuilder
 		nil,              // bucketCache
-		fl,
 	)
 }

pkg/telemetrymetadata/metadata.go

@@ -200,7 +200,7 @@ func (t *telemetryMetaStore) getTracesKeys(ctx context.Context, fieldKeySelector
 		`CASE
 			// WHEN tagType = 'spanfield' THEN 1
 			WHEN tagType = 'resource' THEN 2
-			// WHEN tagType = 'scope' THEN 3
+			WHEN tagType = 'scope' THEN 3
 			WHEN tagType = 'tag' THEN 4
 			ELSE 5
 		END as priority`,

pkg/telemetrytraces/field_mapper.go

@@ -51,6 +51,7 @@ var (
 			ValueType: schema.ColumnTypeString,
 		}},
 		"resource": {Name: "resource", Type: schema.JSONColumnType{}},
+		"scope":    {Name: "scope", Type: schema.JSONColumnType{}},
 
 		"events": {Name: "events", Type: schema.ArrayColumnType{
 			ElementType: schema.ColumnTypeString,
@@ -176,7 +177,7 @@ func (m *defaultFieldMapper) getColumn(
 	case telemetrytypes.FieldContextResource:
 		return []*schema.Column{indexV3Columns["resource"]}, nil
 	case telemetrytypes.FieldContextScope:
-		return []*schema.Column{}, qbtypes.ErrColumnNotFound
+		return []*schema.Column{indexV3Columns["scope"]}, nil
 	case telemetrytypes.FieldContextAttribute:
 		switch key.FieldDataType {
 		case telemetrytypes.FieldDataTypeString:
@@ -261,21 +262,29 @@ func (m *defaultFieldMapper) FieldFor(
 
 	switch column.Type.GetType() {
 	case schema.ColumnTypeEnumJSON:
-		// json is only supported for resource context as of now
-		if key.FieldContext != telemetrytypes.FieldContextResource {
-			return "", errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "only resource context fields are supported for json columns, got %s", key.FieldContext.String)
-		}
-		oldColumn := indexV3Columns["resources_string"]
-		oldKeyName := fmt.Sprintf("%s['%s']", oldColumn.Name, key.Name)
-		// have to add ::string as clickHouse throws an error :- data types Variant/Dynamic are not allowed in GROUP BY
-		// once clickHouse dependency is updated, we need to check if we can remove it.
-		if key.Materialized {
-			oldKeyName = telemetrytypes.FieldKeyToMaterializedColumnName(key)
-			oldKeyNameExists := telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key)
-			return fmt.Sprintf("multiIf(%s.`%s` IS NOT NULL, %s.`%s`::String, %s==true, %s, NULL)", column.Name, key.Name, column.Name, key.Name, oldKeyNameExists, oldKeyName), nil
-		} else {
-			return fmt.Sprintf("multiIf(%s.`%s` IS NOT NULL, %s.`%s`::String, mapContains(%s, '%s'), %s, NULL)", column.Name, key.Name, column.Name, key.Name, oldColumn.Name, key.Name, oldKeyName), nil
+		switch key.FieldContext {
+		case telemetrytypes.FieldContextScope:
+			scopeKey, _ := strings.CutPrefix(key.Name, "scope.") // required for support current implementation of select fields
+			switch scopeKey {
+			case "name", "version":
+				return fmt.Sprintf("%s.%s::String", column.Name, scopeKey), nil
+			default:
+				return fmt.Sprintf("%s.attributes.`%s`::String", column.Name, scopeKey), nil
+			}
+		case telemetrytypes.FieldContextResource:
+			oldColumn := indexV3Columns["resources_string"]
+			oldKeyName := fmt.Sprintf("%s['%s']", oldColumn.Name, key.Name)
+			// have to add ::string as clickHouse throws an error :- data types Variant/Dynamic are not allowed in GROUP BY
+			// once clickHouse dependency is updated, we need to check if we can remove it.
+			if key.Materialized {
+				oldKeyName = telemetrytypes.FieldKeyToMaterializedColumnName(key)
+				oldKeyNameExists := telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key)
+				return fmt.Sprintf("multiIf(%s.`%s` IS NOT NULL, %s.`%s`::String, %s==true, %s, NULL)", column.Name, key.Name, column.Name, key.Name, oldKeyNameExists, oldKeyName), nil
+			} else {
+				return fmt.Sprintf("multiIf(%s.`%s` IS NOT NULL, %s.`%s`::String, mapContains(%s, '%s'), %s, NULL)", column.Name, key.Name, column.Name, key.Name, oldColumn.Name, key.Name, oldKeyName), nil
+			}
 		}
+		return "", errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "json column type only supported for resource and scope context, got %s", key.FieldContext.String)
 	case schema.ColumnTypeEnumString,
 		schema.ColumnTypeEnumUInt64,
 		schema.ColumnTypeEnumUInt32,

pkg/telemetrytraces/field_mapper_test.go

@@ -78,6 +78,51 @@ func TestGetFieldKeyName(t *testing.T) {
 			expectedResult: "multiIf(resource.`deployment.environment` IS NOT NULL, resource.`deployment.environment`::String, `resource_string_deployment$$environment_exists`==true, `resource_string_deployment$$environment`, NULL)",
 			expectedError:  nil,
 		},
+		{
+			name: "Scope field - name (normalized)",
+			key: telemetrytypes.TelemetryFieldKey{
+				Name:         "name",
+				FieldContext: telemetrytypes.FieldContextScope,
+			},
+			expectedResult: "scope.name::String",
+			expectedError:  nil,
+		},
+		{
+			name: "Scope field - name (un-normalized)",
+			key: telemetrytypes.TelemetryFieldKey{
+				Name:         "scope.name",
+				FieldContext: telemetrytypes.FieldContextScope,
+			},
+			expectedResult: "scope.name::String",
+			expectedError:  nil,
+		},
+		{
+			name: "Scope field - version (normalized)",
+			key: telemetrytypes.TelemetryFieldKey{
+				Name:         "version",
+				FieldContext: telemetrytypes.FieldContextScope,
+			},
+			expectedResult: "scope.version::String",
+			expectedError:  nil,
+		},
+		{
+			name: "Scope field - version (un-normalized)",
+			key: telemetrytypes.TelemetryFieldKey{
+				Name:         "scope.version",
+				FieldContext: telemetrytypes.FieldContextScope,
+			},
+			expectedResult: "scope.version::String",
+			expectedError:  nil,
+		},
+		{
+			name: "Scope field - custom attribute",
+			key: telemetrytypes.TelemetryFieldKey{
+				Name:         "custom.attr",
+				FieldContext: telemetrytypes.FieldContextScope,
+			},
+			expectedResult: "scope.attributes.`custom.attr`::String",
+			expectedError:  nil,
+		},
 		{
 			name: "Non-existent column",
 			key: telemetrytypes.TelemetryFieldKey{

pkg/telemetrytraces/stmt_builder_test.go

@@ -350,6 +350,66 @@ func TestStatementBuilder(t *testing.T) {
 			},
 			expectedErr: nil,
 		},
+		{
+			name:        "scope.name filter and group by",
+			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: "scope.name = 'opentelemetry-io'",
+				},
+				Limit: 10,
+				GroupBy: []qbtypes.GroupByKey{
+					{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name:         "scope.name",
+							FieldContext: telemetrytypes.FieldContextScope,
+						},
+					},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __limit_cte AS (SELECT toString(multiIf(scope.name::String IS NOT NULL, scope.name::String, NULL)) AS `scope.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.name::String = ? AND scope.name::String IS NOT NULL) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `scope.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, toString(multiIf(scope.name::String IS NOT NULL, scope.name::String, NULL)) AS `scope.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.name::String = ? AND scope.name::String IS NOT NULL) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND (`scope.name`) GLOBAL IN (SELECT `scope.name` FROM __limit_cte) GROUP BY ts, `scope.name`",
+				Args:  []any{"opentelemetry-io", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10, "opentelemetry-io", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
+			},
+			expectedErr: nil,
+		},
+		{
+			name:        "scope.version filter with scope.name group by",
+			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: "scope.version = '1.0.0'",
+				},
+				Limit: 10,
+				GroupBy: []qbtypes.GroupByKey{
+					{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name:         "scope.name",
+							FieldContext: telemetrytypes.FieldContextScope,
+						},
+					},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __limit_cte AS (SELECT toString(multiIf(scope.name::String IS NOT NULL, scope.name::String, NULL)) AS `scope.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String IS NOT NULL) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `scope.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, toString(multiIf(scope.name::String IS NOT NULL, scope.name::String, NULL)) AS `scope.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String IS NOT NULL) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND (`scope.name`) GLOBAL IN (SELECT `scope.name` FROM __limit_cte) GROUP BY ts, `scope.name`",
+				Args:  []any{"1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10, "1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
+			},
+			expectedErr: nil,
+		},
 	}
 
 	fm := NewFieldMapper()

pkg/telemetrytraces/test_data.go

@@ -111,6 +111,20 @@ func buildCompleteFieldKeyMap() map[string][]*telemetrytypes.TelemetryFieldKey {
 				FieldDataType: telemetrytypes.FieldDataTypeBool,
 			},
 		},
+		"scope.name": {
+			{
+				Name:          "scope.name",
+				FieldContext:  telemetrytypes.FieldContextScope,
+				FieldDataType: telemetrytypes.FieldDataTypeString,
+			},
+		},
+		"scope.version": {
+			{
+				Name:          "scope.version",
+				FieldContext:  telemetrytypes.FieldContextScope,
+				FieldDataType: telemetrytypes.FieldDataTypeString,
+			},
+		},
 	}
 	for _, keys := range keysMap {
 		for _, key := range keys {

pkg/types/inframonitoringtypes/statefulsets.go

@@ -0,0 +1,105 @@
+package inframonitoringtypes
+
+import (
+	"encoding/json"
+	"slices"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+)
+
+type StatefulSets struct {
+	Type                   ResponseType           `json:"type" required:"true"`
+	Records                []StatefulSetRecord    `json:"records" required:"true"`
+	Total                  int                    `json:"total" required:"true"`
+	RequiredMetricsCheck   RequiredMetricsCheck   `json:"requiredMetricsCheck" required:"true"`
+	EndTimeBeforeRetention bool                   `json:"endTimeBeforeRetention" required:"true"`
+	Warning                *qbtypes.QueryWarnData `json:"warning,omitempty"`
+}
+
+type StatefulSetRecord struct {
+	StatefulSetName          string            `json:"statefulSetName" required:"true"`
+	StatefulSetCPU           float64           `json:"statefulSetCPU" required:"true"`
+	StatefulSetCPURequest    float64           `json:"statefulSetCPURequest" required:"true"`
+	StatefulSetCPULimit      float64           `json:"statefulSetCPULimit" required:"true"`
+	StatefulSetMemory        float64           `json:"statefulSetMemory" required:"true"`
+	StatefulSetMemoryRequest float64           `json:"statefulSetMemoryRequest" required:"true"`
+	StatefulSetMemoryLimit   float64           `json:"statefulSetMemoryLimit" required:"true"`
+	DesiredPods              int               `json:"desiredPods" required:"true"`
+	CurrentPods              int               `json:"currentPods" required:"true"`
+	PodCountsByPhase         PodCountsByPhase  `json:"podCountsByPhase" required:"true"`
+	Meta                     map[string]string `json:"meta" required:"true"`
+}
+
+// PostableStatefulSets is the request body for the v2 statefulsets list API.
+type PostableStatefulSets struct {
+	Start   int64                `json:"start" required:"true"`
+	End     int64                `json:"end" required:"true"`
+	Filter  *qbtypes.Filter      `json:"filter"`
+	GroupBy []qbtypes.GroupByKey `json:"groupBy"`
+	OrderBy *qbtypes.OrderBy     `json:"orderBy"`
+	Offset  int                  `json:"offset"`
+	Limit   int                  `json:"limit" required:"true"`
+}
+
+// Validate ensures PostableStatefulSets contains acceptable values.
+func (req *PostableStatefulSets) Validate() error {
+	if req == nil {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "request is nil")
+	}
+
+	if req.Start <= 0 {
+		return errors.NewInvalidInputf(
+			errors.CodeInvalidInput,
+			"invalid start time %d: start must be greater than 0",
+			req.Start,
+		)
+	}
+
+	if req.End <= 0 {
+		return errors.NewInvalidInputf(
+			errors.CodeInvalidInput,
+			"invalid end time %d: end must be greater than 0",
+			req.End,
+		)
+	}
+
+	if req.Start >= req.End {
+		return errors.NewInvalidInputf(
+			errors.CodeInvalidInput,
+			"invalid time range: start (%d) must be less than end (%d)",
+			req.Start,
+			req.End,
+		)
+	}
+
+	if req.Limit < 1 || req.Limit > 5000 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be between 1 and 5000")
+	}
+
+	if req.Offset < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "offset cannot be negative")
+	}
+
+	if req.OrderBy != nil {
+		if !slices.Contains(StatefulSetsValidOrderByKeys, req.OrderBy.Key.Name) {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order by key: %s", req.OrderBy.Key.Name)
+		}
+		if req.OrderBy.Direction != qbtypes.OrderDirectionAsc && req.OrderBy.Direction != qbtypes.OrderDirectionDesc {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order by direction: %s", req.OrderBy.Direction)
+		}
+	}
+
+	return nil
+}
+
+// UnmarshalJSON validates input immediately after decoding.
+func (req *PostableStatefulSets) UnmarshalJSON(data []byte) error {
+	type raw PostableStatefulSets
+	var decoded raw
+	if err := json.Unmarshal(data, &decoded); err != nil {
+		return err
+	}
+	*req = PostableStatefulSets(decoded)
+	return req.Validate()
+}

pkg/types/inframonitoringtypes/statefulsets_constants.go

@@ -0,0 +1,23 @@
+package inframonitoringtypes
+
+const (
+	StatefulSetsOrderByCPU           = "cpu"
+	StatefulSetsOrderByCPURequest    = "cpu_request"
+	StatefulSetsOrderByCPULimit      = "cpu_limit"
+	StatefulSetsOrderByMemory        = "memory"
+	StatefulSetsOrderByMemoryRequest = "memory_request"
+	StatefulSetsOrderByMemoryLimit   = "memory_limit"
+	StatefulSetsOrderByDesiredPods   = "desired_pods"
+	StatefulSetsOrderByCurrentPods   = "current_pods"
+)
+
+var StatefulSetsValidOrderByKeys = []string{
+	StatefulSetsOrderByCPU,
+	StatefulSetsOrderByCPURequest,
+	StatefulSetsOrderByCPULimit,
+	StatefulSetsOrderByMemory,
+	StatefulSetsOrderByMemoryRequest,
+	StatefulSetsOrderByMemoryLimit,
+	StatefulSetsOrderByDesiredPods,
+	StatefulSetsOrderByCurrentPods,
+}

pkg/types/inframonitoringtypes/statefulsets_test.go

@@ -0,0 +1,273 @@
+package inframonitoringtypes
+
+import (
+	"testing"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPostableStatefulSets_Validate(t *testing.T) {
+	tests := []struct {
+		name    string
+		req     *PostableStatefulSets
+		wantErr bool
+	}{
+		{
+			name: "valid request",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+			},
+			wantErr: false,
+		},
+		{
+			name:    "nil request",
+			req:     nil,
+			wantErr: true,
+		},
+		{
+			name: "start time zero",
+			req: &PostableStatefulSets{
+				Start:  0,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "start time negative",
+			req: &PostableStatefulSets{
+				Start:  -1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "end time zero",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    0,
+				Limit:  100,
+				Offset: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "start time greater than end time",
+			req: &PostableStatefulSets{
+				Start:  2000,
+				End:    1000,
+				Limit:  100,
+				Offset: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "start time equal to end time",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    1000,
+				Limit:  100,
+				Offset: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "limit zero",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  0,
+				Offset: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "limit negative",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  -10,
+				Offset: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "limit exceeds max",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  5001,
+				Offset: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "offset negative",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: -5,
+			},
+			wantErr: true,
+		},
+		{
+			name: "orderBy nil is valid",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+			},
+			wantErr: false,
+		},
+		{
+			name: "orderBy with valid key cpu and direction asc",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+				OrderBy: &qbtypes.OrderBy{
+					Key: qbtypes.OrderByKey{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: StatefulSetsOrderByCPU,
+						},
+					},
+					Direction: qbtypes.OrderDirectionAsc,
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "orderBy with valid key memory_limit and direction desc",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+				OrderBy: &qbtypes.OrderBy{
+					Key: qbtypes.OrderByKey{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: StatefulSetsOrderByMemoryLimit,
+						},
+					},
+					Direction: qbtypes.OrderDirectionDesc,
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "orderBy with valid key desired_pods and direction desc",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+				OrderBy: &qbtypes.OrderBy{
+					Key: qbtypes.OrderByKey{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: StatefulSetsOrderByDesiredPods,
+						},
+					},
+					Direction: qbtypes.OrderDirectionDesc,
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "orderBy with valid key current_pods and direction asc",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+				OrderBy: &qbtypes.OrderBy{
+					Key: qbtypes.OrderByKey{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: StatefulSetsOrderByCurrentPods,
+						},
+					},
+					Direction: qbtypes.OrderDirectionAsc,
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "orderBy with restarts key is rejected",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+				OrderBy: &qbtypes.OrderBy{
+					Key: qbtypes.OrderByKey{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: "restarts",
+						},
+					},
+					Direction: qbtypes.OrderDirectionDesc,
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "orderBy with invalid key",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+				OrderBy: &qbtypes.OrderBy{
+					Key: qbtypes.OrderByKey{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: "unknown",
+						},
+					},
+					Direction: qbtypes.OrderDirectionDesc,
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "orderBy with valid key but invalid direction",
+			req: &PostableStatefulSets{
+				Start:  1000,
+				End:    2000,
+				Limit:  100,
+				Offset: 0,
+				OrderBy: &qbtypes.OrderBy{
+					Key: qbtypes.OrderByKey{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: StatefulSetsOrderByCPU,
+						},
+					},
+					Direction: qbtypes.OrderDirection{String: valuer.NewString("invalid")},
+				},
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.req.Validate()
+			if tt.wantErr {
+				require.Error(t, err)
+				require.True(t, errors.Ast(err, errors.TypeInvalidInput), "expected error to be of type InvalidInput")
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}

tests/fixtures/querier.py

@@ -69,7 +69,7 @@ class BuilderQuery:
 class TraceOperatorQuery:
     name: str
     expression: str
-    return_spans_from: str
+    return_spans_from: str | None = None
     limit: int | None = None
     order: list[OrderBy] | None = None
 
@@ -77,8 +77,9 @@ class TraceOperatorQuery:
         spec: dict[str, Any] = {
             "name": self.name,
             "expression": self.expression,
-            "returnSpansFrom": self.return_spans_from,
         }
+        if self.return_spans_from is not None:
+            spec["returnSpansFrom"] = self.return_spans_from
         if self.limit is not None:
             spec["limit"] = self.limit
         if self.order:

tests/integration/tests/querier_json_body/01_logs_json_body_new_qb.py

@@ -20,7 +20,7 @@ from fixtures.querier import (
 
 
 def _get_bodies(response: requests.Response) -> list[dict[str, Any]]:
-    return [row["data"]["body"] for row in get_rows(response)]
+    return [json.loads(row["data"]["body"]) for row in get_rows(response)]
 
 
 def _run_query_case(signoz: types.SigNoz, token: str, now: datetime, case: dict[str, Any]) -> None:
@@ -1183,7 +1183,7 @@ def test_message_searches(
     token = get_token(email=USER_ADMIN_EMAIL, password=USER_ADMIN_PASSWORD)
 
     def _body_messages(response: requests.Response) -> list[str]:
-        return [row["data"]["body"].get("message", "") for row in get_rows(response)]
+        return [json.loads(row["data"]["body"]).get("message", "") for row in get_rows(response)]
 
     payment_messages = {
         "Payment processed successfully",

tests/integration/tests/rawexportdata/02_traces.py

@@ -625,7 +625,6 @@ def test_export_traces_with_composite_query_trace_operator(
     query_c = TraceOperatorQuery(
         name="C",
         expression="A => B",
-        return_spans_from="A",
         limit=1000,
         order=[OrderBy(TelemetryFieldKey("timestamp", "string", "span"), "desc")],
     )
@@ -652,17 +651,15 @@ def test_export_traces_with_composite_query_trace_operator(
 
     # Parse JSONL content
     jsonl_lines = response.text.strip().split("\n")
-    assert len(jsonl_lines) == 1, f"Expected at least 1 line, got {len(jsonl_lines)}"
+    assert len(jsonl_lines) >= 1, f"Expected at least 1 line, got {len(jsonl_lines)}"
 
-    # Verify all returned spans belong to the matched trace
+    # Verify all returned spans belong to the matched trace.
+    # The direct-descendant JOIN emits one row per matching child, so the parent
+    # span may appear more than once (once per child that satisfies the condition).
     json_objects = [json.loads(line) for line in jsonl_lines]
     trace_ids = [obj.get("trace_id") for obj in json_objects]
     assert all(tid == parent_trace_id for tid in trace_ids)
 
-    # Verify the parent span (returnSpansFrom = "A") is present
-    span_names = [obj.get("name") for obj in json_objects]
-    assert "parent-operation" in span_names
-
 
 def test_export_traces_with_select_fields(
     signoz: types.SigNoz,