chore: remove redundent comparison

* feat: adding query params in cloud integration APIs

* refactor: create account HTTP status change from OK to CREATED

cmd/community/server.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/SigNoz/signoz/cmd"
 	"github.com/SigNoz/signoz/pkg/analytics"
+	"github.com/SigNoz/signoz/pkg/auditor"
 	"github.com/SigNoz/signoz/pkg/authn"
 	"github.com/SigNoz/signoz/pkg/authz"
 	"github.com/SigNoz/signoz/pkg/authz/openfgaauthz"
@@ -93,6 +94,9 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 		func(_ licensing.Licensing) factory.ProviderFactory[gateway.Gateway, gateway.Config] {
 			return noopgateway.NewProviderFactory()
 		},
+		func(_ licensing.Licensing) factory.NamedMap[factory.ProviderFactory[auditor.Auditor, auditor.Config]] {
+			return signoz.NewAuditorProviderFactories()
+		},
 		func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
 			return querier.NewHandler(ps, q, a)
 		},

cmd/enterprise/server.go

@@ -8,6 +8,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/SigNoz/signoz/cmd"
+	"github.com/SigNoz/signoz/ee/auditor/otlphttpauditor"
 	"github.com/SigNoz/signoz/ee/authn/callbackauthn/oidccallbackauthn"
 	"github.com/SigNoz/signoz/ee/authn/callbackauthn/samlcallbackauthn"
 	"github.com/SigNoz/signoz/ee/authz/openfgaauthz"
@@ -24,6 +25,7 @@ import (
 	enterprisezeus "github.com/SigNoz/signoz/ee/zeus"
 	"github.com/SigNoz/signoz/ee/zeus/httpzeus"
 	"github.com/SigNoz/signoz/pkg/analytics"
+	"github.com/SigNoz/signoz/pkg/auditor"
 	"github.com/SigNoz/signoz/pkg/authn"
 	"github.com/SigNoz/signoz/pkg/authz"
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -133,6 +135,13 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 		func(licensing licensing.Licensing) factory.ProviderFactory[gateway.Gateway, gateway.Config] {
 			return httpgateway.NewProviderFactory(licensing)
 		},
+		func(licensing licensing.Licensing) factory.NamedMap[factory.ProviderFactory[auditor.Auditor, auditor.Config]] {
+			factories := signoz.NewAuditorProviderFactories()
+			if err := factories.Add(otlphttpauditor.NewFactory(licensing, version.Info)); err != nil {
+				panic(err)
+			}
+			return factories
+		},
 		func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
 			communityHandler := querier.NewHandler(ps, q, a)
 			return eequerier.NewHandler(ps, q, communityHandler)

conf/example.yaml

@@ -364,3 +364,34 @@ serviceaccount:
   analytics:
     # toggle service account analytics
     enabled: true
+
+##################### Auditor #####################
+auditor:
+  # Specifies the auditor provider to use.
+  # noop: discards all audit events (community default).
+  # otlphttp: exports audit events via OTLP HTTP (enterprise).
+  provider: noop
+  # The async channel capacity for audit events. Events are dropped when full (fail-open).
+  buffer_size: 1000
+  # The maximum number of events per export batch.
+  batch_size: 100
+  # The maximum time between export flushes.
+  flush_interval: 1s
+  otlphttp:
+    # The target scheme://host:port/path of the OTLP HTTP endpoint.
+    endpoint: http://localhost:4318/v1/logs
+    # Whether to use HTTP instead of HTTPS.
+    insecure: false
+    # The maximum duration for an export attempt.
+    timeout: 10s
+    # Additional HTTP headers sent with every export request.
+    headers: {}
+    retry:
+      # Whether to retry on transient failures.
+      enabled: true
+      # The initial wait time before the first retry.
+      initial_interval: 5s
+      # The upper bound on backoff interval.
+      max_interval: 30s
+      # The total maximum time spent retrying.
+      max_elapsed_time: 60s

docs/api/openapi.yml

@@ -2787,146 +2787,6 @@ components:
             type: string
           type: array
       type: object
-    TracedetailtypesEvent:
-      properties:
-        attributeMap:
-          additionalProperties: {}
-          type: object
-        isError:
-          type: boolean
-        name:
-          type: string
-        timeUnixNano:
-          minimum: 0
-          type: integer
-      type: object
-    TracedetailtypesWaterfallRequest:
-      properties:
-        limit:
-          minimum: 0
-          type: integer
-        selectedSpanId:
-          type: string
-        uncollapsedSpans:
-          items:
-            type: string
-          nullable: true
-          type: array
-      type: object
-    TracedetailtypesWaterfallResponse:
-      properties:
-        endTimestampMillis:
-          minimum: 0
-          type: integer
-        hasMissingSpans:
-          type: boolean
-        hasMore:
-          type: boolean
-        rootServiceEntryPoint:
-          type: string
-        rootServiceName:
-          type: string
-        serviceNameToTotalDurationMap:
-          additionalProperties:
-            minimum: 0
-            type: integer
-          nullable: true
-          type: object
-        spans:
-          items:
-            $ref: '#/components/schemas/TracedetailtypesWaterfallSpan'
-          nullable: true
-          type: array
-        startTimestampMillis:
-          minimum: 0
-          type: integer
-        totalErrorSpansCount:
-          minimum: 0
-          type: integer
-        totalSpansCount:
-          minimum: 0
-          type: integer
-        uncollapsedSpans:
-          items:
-            type: string
-          nullable: true
-          type: array
-      type: object
-    TracedetailtypesWaterfallSpan:
-      properties:
-        attributes:
-          additionalProperties: {}
-          nullable: true
-          type: object
-        db_name:
-          type: string
-        db_operation:
-          type: string
-        duration_nano:
-          minimum: 0
-          type: integer
-        events:
-          items:
-            $ref: '#/components/schemas/TracedetailtypesEvent'
-          nullable: true
-          type: array
-        external_http_method:
-          type: string
-        external_http_url:
-          type: string
-        flags:
-          minimum: 0
-          type: integer
-        has_children:
-          type: boolean
-        has_error:
-          type: boolean
-        http_host:
-          type: string
-        http_method:
-          type: string
-        http_url:
-          type: string
-        is_remote:
-          type: string
-        kind:
-          format: int32
-          type: integer
-        kind_string:
-          type: string
-        level:
-          minimum: 0
-          type: integer
-        name:
-          type: string
-        parent_span_id:
-          type: string
-        resource:
-          additionalProperties:
-            type: string
-          nullable: true
-          type: object
-        response_status_code:
-          type: string
-        span_id:
-          type: string
-        status_code:
-          type: integer
-        status_code_string:
-          type: string
-        status_message:
-          type: string
-        sub_tree_node_count:
-          minimum: 0
-          type: integer
-        timestamp:
-          minimum: 0
-          type: integer
-        trace_id:
-          type: string
-        trace_state:
-          type: string
-      type: object
     TypesChangePasswordRequest:
       properties:
         newPassword:
@@ -3449,7 +3309,7 @@ paths:
             schema:
               $ref: '#/components/schemas/CloudintegrationtypesPostableAccount'
       responses:
-        "200":
+        "201":
           content:
             application/json:
               schema:
@@ -3462,7 +3322,7 @@ paths:
                 - status
                 - data
                 type: object
-          description: OK
+          description: Created
         "401":
           content:
             application/json:
@@ -3823,6 +3683,11 @@ paths:
         provider
       operationId: ListServicesMetadata
       parameters:
+      - in: query
+        name: cloud_integration_id
+        required: false
+        schema:
+          type: string
       - in: path
         name: cloud_provider
         required: true
@@ -3875,6 +3740,11 @@ paths:
       description: This endpoint gets a service for the specified cloud provider
       operationId: GetService
       parameters:
+      - in: query
+        name: cloud_integration_id
+        required: false
+        schema:
+          type: string
       - in: path
         name: cloud_provider
         required: true
@@ -9644,76 +9514,6 @@ paths:
       summary: Put profile in Zeus for a deployment.
       tags:
       - zeus
-  /api/v3/traces/{traceID}/waterfall:
-    post:
-      deprecated: false
-      description: Returns the waterfall view of spans for a given trace ID with tree
-        structure, metadata, and windowed pagination
-      operationId: GetWaterfall
-      parameters:
-      - in: path
-        name: traceID
-        required: true
-        schema:
-          type: string
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: '#/components/schemas/TracedetailtypesWaterfallRequest'
-      responses:
-        "200":
-          content:
-            application/json:
-              schema:
-                properties:
-                  data:
-                    $ref: '#/components/schemas/TracedetailtypesWaterfallResponse'
-                  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
-        "404":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Not Found
-        "500":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Internal Server Error
-      security:
-      - api_key:
-        - VIEWER
-      - tokenizer:
-        - VIEWER
-      summary: Get waterfall view for a trace
-      tags:
-      - tracedetail
   /api/v5/query_range:
     post:
       deprecated: false

ee/query-service/app/server.go

@@ -227,7 +227,7 @@ func (s *Server) createPublicServer(apiHandler *api.APIHandler, web web.Web) (*h
 		s.config.APIServer.Timeout.Default,
 		s.config.APIServer.Timeout.Max,
 	).Wrap)
-	r.Use(middleware.NewAudit(s.signoz.Instrumentation.Logger(), s.config.APIServer.Logging.ExcludedRoutes, nil).Wrap)
+	r.Use(middleware.NewAudit(s.signoz.Instrumentation.Logger(), s.config.APIServer.Logging.ExcludedRoutes, s.signoz.Auditor).Wrap)
 	r.Use(middleware.NewComment().Wrap)
 
 	apiHandler.RegisterRoutes(r, am)

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

@@ -28,7 +28,7 @@ import type {
 	CloudintegrationtypesPostableAgentCheckInDTO,
 	CloudintegrationtypesUpdatableAccountDTO,
 	CloudintegrationtypesUpdatableServiceDTO,
-	CreateAccount200,
+	CreateAccount201,
 	CreateAccountPathParameters,
 	DisconnectAccountPathParameters,
 	GetAccount200,
@@ -36,10 +36,12 @@ import type {
 	GetConnectionCredentials200,
 	GetConnectionCredentialsPathParameters,
 	GetService200,
+	GetServiceParams,
 	GetServicePathParameters,
 	ListAccounts200,
 	ListAccountsPathParameters,
 	ListServicesMetadata200,
+	ListServicesMetadataParams,
 	ListServicesMetadataPathParameters,
 	RenderErrorResponseDTO,
 	UpdateAccountPathParameters,
@@ -260,7 +262,7 @@ export const createAccount = (
 	cloudintegrationtypesPostableAccountDTO: BodyType<CloudintegrationtypesPostableAccountDTO>,
 	signal?: AbortSignal,
 ) => {
-	return GeneratedAPIInstance<CreateAccount200>({
+	return GeneratedAPIInstance<CreateAccount201>({
 		url: `/api/v1/cloud_integrations/${cloudProvider}/accounts`,
 		method: 'POST',
 		headers: { 'Content-Type': 'application/json' },
@@ -940,19 +942,25 @@ export const invalidateGetConnectionCredentials = async (
  */
 export const listServicesMetadata = (
 	{ cloudProvider }: ListServicesMetadataPathParameters,
+	params?: ListServicesMetadataParams,
 	signal?: AbortSignal,
 ) => {
 	return GeneratedAPIInstance<ListServicesMetadata200>({
 		url: `/api/v1/cloud_integrations/${cloudProvider}/services`,
 		method: 'GET',
+		params,
 		signal,
 	});
 };
 
-export const getListServicesMetadataQueryKey = ({
-	cloudProvider,
-}: ListServicesMetadataPathParameters) => {
-	return [`/api/v1/cloud_integrations/${cloudProvider}/services`] as const;
+export const getListServicesMetadataQueryKey = (
+	{ cloudProvider }: ListServicesMetadataPathParameters,
+	params?: ListServicesMetadataParams,
+) => {
+	return [
+		`/api/v1/cloud_integrations/${cloudProvider}/services`,
+		...(params ? [params] : []),
+	] as const;
 };
 
 export const getListServicesMetadataQueryOptions = <
@@ -960,6 +968,7 @@ export const getListServicesMetadataQueryOptions = <
 	TError = ErrorType<RenderErrorResponseDTO>
 >(
 	{ cloudProvider }: ListServicesMetadataPathParameters,
+	params?: ListServicesMetadataParams,
 	options?: {
 		query?: UseQueryOptions<
 			Awaited<ReturnType<typeof listServicesMetadata>>,
@@ -971,11 +980,12 @@ export const getListServicesMetadataQueryOptions = <
 	const { query: queryOptions } = options ?? {};
 
 	const queryKey =
-		queryOptions?.queryKey ?? getListServicesMetadataQueryKey({ cloudProvider });
+		queryOptions?.queryKey ??
+		getListServicesMetadataQueryKey({ cloudProvider }, params);
 
 	const queryFn: QueryFunction<
 		Awaited<ReturnType<typeof listServicesMetadata>>
-	> = ({ signal }) => listServicesMetadata({ cloudProvider }, signal);
+	> = ({ signal }) => listServicesMetadata({ cloudProvider }, params, signal);
 
 	return {
 		queryKey,
@@ -1003,6 +1013,7 @@ export function useListServicesMetadata<
 	TError = ErrorType<RenderErrorResponseDTO>
 >(
 	{ cloudProvider }: ListServicesMetadataPathParameters,
+	params?: ListServicesMetadataParams,
 	options?: {
 		query?: UseQueryOptions<
 			Awaited<ReturnType<typeof listServicesMetadata>>,
@@ -1013,6 +1024,7 @@ export function useListServicesMetadata<
 ): UseQueryResult<TData, TError> & { queryKey: QueryKey } {
 	const queryOptions = getListServicesMetadataQueryOptions(
 		{ cloudProvider },
+		params,
 		options,
 	);
 
@@ -1031,10 +1043,11 @@ export function useListServicesMetadata<
 export const invalidateListServicesMetadata = async (
 	queryClient: QueryClient,
 	{ cloudProvider }: ListServicesMetadataPathParameters,
+	params?: ListServicesMetadataParams,
 	options?: InvalidateOptions,
 ): Promise<QueryClient> => {
 	await queryClient.invalidateQueries(
-		{ queryKey: getListServicesMetadataQueryKey({ cloudProvider }) },
+		{ queryKey: getListServicesMetadataQueryKey({ cloudProvider }, params) },
 		options,
 	);
 
@@ -1047,21 +1060,24 @@ export const invalidateListServicesMetadata = async (
  */
 export const getService = (
 	{ cloudProvider, serviceId }: GetServicePathParameters,
+	params?: GetServiceParams,
 	signal?: AbortSignal,
 ) => {
 	return GeneratedAPIInstance<GetService200>({
 		url: `/api/v1/cloud_integrations/${cloudProvider}/services/${serviceId}`,
 		method: 'GET',
+		params,
 		signal,
 	});
 };
 
-export const getGetServiceQueryKey = ({
-	cloudProvider,
-	serviceId,
-}: GetServicePathParameters) => {
+export const getGetServiceQueryKey = (
+	{ cloudProvider, serviceId }: GetServicePathParameters,
+	params?: GetServiceParams,
+) => {
 	return [
 		`/api/v1/cloud_integrations/${cloudProvider}/services/${serviceId}`,
+		...(params ? [params] : []),
 	] as const;
 };
 
@@ -1070,6 +1086,7 @@ export const getGetServiceQueryOptions = <
 	TError = ErrorType<RenderErrorResponseDTO>
 >(
 	{ cloudProvider, serviceId }: GetServicePathParameters,
+	params?: GetServiceParams,
 	options?: {
 		query?: UseQueryOptions<
 			Awaited<ReturnType<typeof getService>>,
@@ -1081,11 +1098,12 @@ export const getGetServiceQueryOptions = <
 	const { query: queryOptions } = options ?? {};
 
 	const queryKey =
-		queryOptions?.queryKey ?? getGetServiceQueryKey({ cloudProvider, serviceId });
+		queryOptions?.queryKey ??
+		getGetServiceQueryKey({ cloudProvider, serviceId }, params);
 
 	const queryFn: QueryFunction<Awaited<ReturnType<typeof getService>>> = ({
 		signal,
-	}) => getService({ cloudProvider, serviceId }, signal);
+	}) => getService({ cloudProvider, serviceId }, params, signal);
 
 	return {
 		queryKey,
@@ -1111,6 +1129,7 @@ export function useGetService<
 	TError = ErrorType<RenderErrorResponseDTO>
 >(
 	{ cloudProvider, serviceId }: GetServicePathParameters,
+	params?: GetServiceParams,
 	options?: {
 		query?: UseQueryOptions<
 			Awaited<ReturnType<typeof getService>>,
@@ -1121,6 +1140,7 @@ export function useGetService<
 ): UseQueryResult<TData, TError> & { queryKey: QueryKey } {
 	const queryOptions = getGetServiceQueryOptions(
 		{ cloudProvider, serviceId },
+		params,
 		options,
 	);
 
@@ -1139,10 +1159,11 @@ export function useGetService<
 export const invalidateGetService = async (
 	queryClient: QueryClient,
 	{ cloudProvider, serviceId }: GetServicePathParameters,
+	params?: GetServiceParams,
 	options?: InvalidateOptions,
 ): Promise<QueryClient> => {
 	await queryClient.invalidateQueries(
-		{ queryKey: getGetServiceQueryKey({ cloudProvider, serviceId }) },
+		{ queryKey: getGetServiceQueryKey({ cloudProvider, serviceId }, params) },
 		options,
 	);
 

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

@@ -3196,248 +3196,6 @@ export interface TelemetrytypesTelemetryFieldValuesDTO {
 	stringValues?: string[];
 }
 
-export type TracedetailtypesEventDTOAttributeMap = { [key: string]: unknown };
-
-export interface TracedetailtypesEventDTO {
-	/**
-	 * @type object
-	 */
-	attributeMap?: TracedetailtypesEventDTOAttributeMap;
-	/**
-	 * @type boolean
-	 */
-	isError?: boolean;
-	/**
-	 * @type string
-	 */
-	name?: string;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	timeUnixNano?: number;
-}
-
-export interface TracedetailtypesWaterfallRequestDTO {
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	limit?: number;
-	/**
-	 * @type string
-	 */
-	selectedSpanId?: string;
-	/**
-	 * @type array
-	 * @nullable true
-	 */
-	uncollapsedSpans?: string[] | null;
-}
-
-/**
- * @nullable
- */
-export type TracedetailtypesWaterfallResponseDTOServiceNameToTotalDurationMap = {
-	[key: string]: number;
-} | null;
-
-export interface TracedetailtypesWaterfallResponseDTO {
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	endTimestampMillis?: number;
-	/**
-	 * @type boolean
-	 */
-	hasMissingSpans?: boolean;
-	/**
-	 * @type boolean
-	 */
-	hasMore?: boolean;
-	/**
-	 * @type string
-	 */
-	rootServiceEntryPoint?: string;
-	/**
-	 * @type string
-	 */
-	rootServiceName?: string;
-	/**
-	 * @type object
-	 * @nullable true
-	 */
-	serviceNameToTotalDurationMap?: TracedetailtypesWaterfallResponseDTOServiceNameToTotalDurationMap;
-	/**
-	 * @type array
-	 * @nullable true
-	 */
-	spans?: TracedetailtypesWaterfallSpanDTO[] | null;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	startTimestampMillis?: number;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	totalErrorSpansCount?: number;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	totalSpansCount?: number;
-	/**
-	 * @type array
-	 * @nullable true
-	 */
-	uncollapsedSpans?: string[] | null;
-}
-
-/**
- * @nullable
- */
-export type TracedetailtypesWaterfallSpanDTOAttributes = {
-	[key: string]: unknown;
-} | null;
-
-/**
- * @nullable
- */
-export type TracedetailtypesWaterfallSpanDTOResource = {
-	[key: string]: string;
-} | null;
-
-export interface TracedetailtypesWaterfallSpanDTO {
-	/**
-	 * @type object
-	 * @nullable true
-	 */
-	attributes?: TracedetailtypesWaterfallSpanDTOAttributes;
-	/**
-	 * @type string
-	 */
-	db_name?: string;
-	/**
-	 * @type string
-	 */
-	db_operation?: string;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	duration_nano?: number;
-	/**
-	 * @type array
-	 * @nullable true
-	 */
-	events?: TracedetailtypesEventDTO[] | null;
-	/**
-	 * @type string
-	 */
-	external_http_method?: string;
-	/**
-	 * @type string
-	 */
-	external_http_url?: string;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	flags?: number;
-	/**
-	 * @type boolean
-	 */
-	has_children?: boolean;
-	/**
-	 * @type boolean
-	 */
-	has_error?: boolean;
-	/**
-	 * @type string
-	 */
-	http_host?: string;
-	/**
-	 * @type string
-	 */
-	http_method?: string;
-	/**
-	 * @type string
-	 */
-	http_url?: string;
-	/**
-	 * @type string
-	 */
-	is_remote?: string;
-	/**
-	 * @type integer
-	 * @format int32
-	 */
-	kind?: number;
-	/**
-	 * @type string
-	 */
-	kind_string?: string;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	level?: number;
-	/**
-	 * @type string
-	 */
-	name?: string;
-	/**
-	 * @type string
-	 */
-	parent_span_id?: string;
-	/**
-	 * @type object
-	 * @nullable true
-	 */
-	resource?: TracedetailtypesWaterfallSpanDTOResource;
-	/**
-	 * @type string
-	 */
-	response_status_code?: string;
-	/**
-	 * @type string
-	 */
-	span_id?: string;
-	/**
-	 * @type integer
-	 */
-	status_code?: number;
-	/**
-	 * @type string
-	 */
-	status_code_string?: string;
-	/**
-	 * @type string
-	 */
-	status_message?: string;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	sub_tree_node_count?: number;
-	/**
-	 * @type integer
-	 * @minimum 0
-	 */
-	timestamp?: number;
-	/**
-	 * @type string
-	 */
-	trace_id?: string;
-	/**
-	 * @type string
-	 */
-	trace_state?: string;
-}
-
 export interface TypesChangePasswordRequestDTO {
 	/**
 	 * @type string
@@ -3831,7 +3589,7 @@ export type ListAccounts200 = {
 export type CreateAccountPathParameters = {
 	cloudProvider: string;
 };
-export type CreateAccount200 = {
+export type CreateAccount201 = {
 	data: CloudintegrationtypesGettableAccountWithConnectionArtifactDTO;
 	/**
 	 * @type string
@@ -3889,6 +3647,14 @@ export type GetConnectionCredentials200 = {
 export type ListServicesMetadataPathParameters = {
 	cloudProvider: string;
 };
+export type ListServicesMetadataParams = {
+	/**
+	 * @type string
+	 * @description undefined
+	 */
+	cloud_integration_id?: string;
+};
+
 export type ListServicesMetadata200 = {
 	data: CloudintegrationtypesGettableServicesMetadataDTO;
 	/**
@@ -3901,6 +3667,14 @@ export type GetServicePathParameters = {
 	cloudProvider: string;
 	serviceId: string;
 };
+export type GetServiceParams = {
+	/**
+	 * @type string
+	 * @description undefined
+	 */
+	cloud_integration_id?: string;
+};
+
 export type GetService200 = {
 	data: CloudintegrationtypesServiceDTO;
 	/**
@@ -5099,17 +4873,6 @@ export type GetHosts200 = {
 	status: string;
 };
 
-export type GetWaterfallPathParameters = {
-	traceID: string;
-};
-export type GetWaterfall200 = {
-	data: TracedetailtypesWaterfallResponseDTO;
-	/**
-	 * @type string
-	 */
-	status: string;
-};
-
 export type QueryRangeV5200 = {
 	data: Querybuildertypesv5QueryRangeResponseDTO;
 	/**

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

@@ -1,121 +0,0 @@
-/**
- * ! Do not edit manually
- * * The file has been auto-generated using Orval for SigNoz
- * * regenerate with 'yarn generate:api'
- * SigNoz
- */
-import type {
-	MutationFunction,
-	UseMutationOptions,
-	UseMutationResult,
-} from 'react-query';
-import { useMutation } from 'react-query';
-
-import type { BodyType, ErrorType } from '../../../generatedAPIInstance';
-import { GeneratedAPIInstance } from '../../../generatedAPIInstance';
-import type {
-	GetWaterfall200,
-	GetWaterfallPathParameters,
-	RenderErrorResponseDTO,
-	TracedetailtypesWaterfallRequestDTO,
-} from '../sigNoz.schemas';
-
-/**
- * Returns the waterfall view of spans for a given trace ID with tree structure, metadata, and windowed pagination
- * @summary Get waterfall view for a trace
- */
-export const getWaterfall = (
-	{ traceID }: GetWaterfallPathParameters,
-	tracedetailtypesWaterfallRequestDTO: BodyType<TracedetailtypesWaterfallRequestDTO>,
-	signal?: AbortSignal,
-) => {
-	return GeneratedAPIInstance<GetWaterfall200>({
-		url: `/api/v3/traces/${traceID}/waterfall`,
-		method: 'POST',
-		headers: { 'Content-Type': 'application/json' },
-		data: tracedetailtypesWaterfallRequestDTO,
-		signal,
-	});
-};
-
-export const getGetWaterfallMutationOptions = <
-	TError = ErrorType<RenderErrorResponseDTO>,
-	TContext = unknown
->(options?: {
-	mutation?: UseMutationOptions<
-		Awaited<ReturnType<typeof getWaterfall>>,
-		TError,
-		{
-			pathParams: GetWaterfallPathParameters;
-			data: BodyType<TracedetailtypesWaterfallRequestDTO>;
-		},
-		TContext
-	>;
-}): UseMutationOptions<
-	Awaited<ReturnType<typeof getWaterfall>>,
-	TError,
-	{
-		pathParams: GetWaterfallPathParameters;
-		data: BodyType<TracedetailtypesWaterfallRequestDTO>;
-	},
-	TContext
-> => {
-	const mutationKey = ['getWaterfall'];
-	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 getWaterfall>>,
-		{
-			pathParams: GetWaterfallPathParameters;
-			data: BodyType<TracedetailtypesWaterfallRequestDTO>;
-		}
-	> = (props) => {
-		const { pathParams, data } = props ?? {};
-
-		return getWaterfall(pathParams, data);
-	};
-
-	return { mutationFn, ...mutationOptions };
-};
-
-export type GetWaterfallMutationResult = NonNullable<
-	Awaited<ReturnType<typeof getWaterfall>>
->;
-export type GetWaterfallMutationBody = BodyType<TracedetailtypesWaterfallRequestDTO>;
-export type GetWaterfallMutationError = ErrorType<RenderErrorResponseDTO>;
-
-/**
- * @summary Get waterfall view for a trace
- */
-export const useGetWaterfall = <
-	TError = ErrorType<RenderErrorResponseDTO>,
-	TContext = unknown
->(options?: {
-	mutation?: UseMutationOptions<
-		Awaited<ReturnType<typeof getWaterfall>>,
-		TError,
-		{
-			pathParams: GetWaterfallPathParameters;
-			data: BodyType<TracedetailtypesWaterfallRequestDTO>;
-		},
-		TContext
-	>;
-}): UseMutationResult<
-	Awaited<ReturnType<typeof getWaterfall>>,
-	TError,
-	{
-		pathParams: GetWaterfallPathParameters;
-		data: BodyType<TracedetailtypesWaterfallRequestDTO>;
-	},
-	TContext
-> => {
-	const mutationOptions = getGetWaterfallMutationOptions(options);
-
-	return useMutation(mutationOptions);
-};

frontend/src/container/NewWidget/index.tsx

@@ -677,6 +677,18 @@ function NewWidget({
 			queryType: currentQuery.queryType,
 			isNewPanel,
 			dataSource: currentQuery?.builder?.queryData?.[0]?.dataSource,
+			...(currentQuery.queryType === EQueryType.CLICKHOUSE && {
+				clickhouseQueryCount: currentQuery.clickhouse_sql.length,
+				clickhouseQueries: currentQuery.clickhouse_sql.map((q) => ({
+					name: q.name,
+					query: (q.query ?? '')
+						.replace(/--[^\n]*/g, '') // strip line comments
+						.replace(/\/\*[\s\S]*?\*\//g, '') // strip block comments
+						.replace(/'(?:[^'\\]|\\.|'')*'/g, "'?'") // replace single-quoted strings (handles \' and '' escapes)
+						.replace(/\b\d+(?:\.\d+)?(?:[eE][+-]?\d+)?\b/g, '?'), // replace numeric literals (int, float, scientific)
+					disabled: q.disabled,
+				})),
+			}),
 		});
 		setSaveModal(true);
 		// eslint-disable-next-line react-hooks/exhaustive-deps

pkg/apiserver/signozapiserver/cloudintegration.go

@@ -41,7 +41,7 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			RequestContentType:  "application/json",
 			Response:            new(citypes.GettableAccountWithConnectionArtifact),
 			ResponseContentType: "application/json",
-			SuccessStatusCode:   http.StatusOK,
+			SuccessStatusCode:   http.StatusCreated,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
 			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
@@ -138,6 +138,7 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			Summary:             "List services metadata",
 			Description:         "This endpoint lists the services metadata for the specified cloud provider",
 			Request:             nil,
+			RequestQuery:        new(citypes.ListServicesMetadataParams),
 			RequestContentType:  "",
 			Response:            new(citypes.GettableServicesMetadata),
 			ResponseContentType: "application/json",
@@ -158,6 +159,7 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			Summary:             "Get service",
 			Description:         "This endpoint gets a service for the specified cloud provider",
 			Request:             nil,
+			RequestQuery:        new(citypes.GetServiceParams),
 			RequestContentType:  "",
 			Response:            new(citypes.Service),
 			ResponseContentType: "application/json",

pkg/apiserver/signozapiserver/provider.go

@@ -23,7 +23,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/rulestatehistory"
 	"github.com/SigNoz/signoz/pkg/modules/serviceaccount"
 	"github.com/SigNoz/signoz/pkg/modules/session"
-	"github.com/SigNoz/signoz/pkg/modules/tracedetail"
 	"github.com/SigNoz/signoz/pkg/modules/user"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/types"
@@ -58,7 +57,6 @@ type provider struct {
 	factoryHandler          factory.Handler
 	cloudIntegrationHandler cloudintegration.Handler
 	ruleStateHistoryHandler rulestatehistory.Handler
-	traceDetailHandler      tracedetail.Handler
 }
 
 func NewFactory(
@@ -85,7 +83,6 @@ func NewFactory(
 	factoryHandler factory.Handler,
 	cloudIntegrationHandler cloudintegration.Handler,
 	ruleStateHistoryHandler rulestatehistory.Handler,
-	traceDetailHandler tracedetail.Handler,
 ) factory.ProviderFactory[apiserver.APIServer, apiserver.Config] {
 	return factory.NewProviderFactory(factory.MustNewName("signoz"), func(ctx context.Context, providerSettings factory.ProviderSettings, config apiserver.Config) (apiserver.APIServer, error) {
 		return newProvider(
@@ -115,7 +112,6 @@ func NewFactory(
 			factoryHandler,
 			cloudIntegrationHandler,
 			ruleStateHistoryHandler,
-			traceDetailHandler,
 		)
 	})
 }
@@ -147,7 +143,6 @@ func newProvider(
 	factoryHandler factory.Handler,
 	cloudIntegrationHandler cloudintegration.Handler,
 	ruleStateHistoryHandler rulestatehistory.Handler,
-	traceDetailHandler tracedetail.Handler,
 ) (apiserver.APIServer, error) {
 	settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/apiserver/signozapiserver")
 	router := mux.NewRouter().UseEncodedPath()
@@ -177,7 +172,6 @@ func newProvider(
 		factoryHandler:          factoryHandler,
 		cloudIntegrationHandler: cloudIntegrationHandler,
 		ruleStateHistoryHandler: ruleStateHistoryHandler,
-		traceDetailHandler:      traceDetailHandler,
 	}
 
 	provider.authZ = middleware.NewAuthZ(settings.Logger(), orgGetter, authz)
@@ -278,10 +272,6 @@ func (provider *provider) AddToRouter(router *mux.Router) error {
 		return err
 	}
 
-	if err := provider.addTraceDetailRoutes(router); err != nil {
-		return err
-	}
-
 	return nil
 }
 

pkg/apiserver/signozapiserver/tracedetail.go

@@ -1,33 +0,0 @@
-package signozapiserver
-
-import (
-	"net/http"
-
-	"github.com/SigNoz/signoz/pkg/http/handler"
-	"github.com/SigNoz/signoz/pkg/types"
-	"github.com/SigNoz/signoz/pkg/types/tracedetailtypes"
-	"github.com/gorilla/mux"
-)
-
-func (provider *provider) addTraceDetailRoutes(router *mux.Router) error {
-	if err := router.Handle("/api/v3/traces/{traceID}/waterfall", handler.New(
-		provider.authZ.ViewAccess(provider.traceDetailHandler.GetWaterfall),
-		handler.OpenAPIDef{
-			ID:                  "GetWaterfall",
-			Tags:                []string{"tracedetail"},
-			Summary:             "Get waterfall view for a trace",
-			Description:         "Returns the waterfall view of spans for a given trace ID with tree structure, metadata, and windowed pagination",
-			Request:             new(tracedetailtypes.WaterfallRequest),
-			RequestContentType:  "application/json",
-			Response:            new(tracedetailtypes.WaterfallResponse),
-			ResponseContentType: "application/json",
-			SuccessStatusCode:   http.StatusOK,
-			ErrorStatusCodes:    []int{http.StatusBadRequest, http.StatusNotFound},
-			SecuritySchemes:     newSecuritySchemes(types.RoleViewer),
-		},
-	)).Methods(http.MethodPost).GetError(); err != nil {
-		return err
-	}
-
-	return nil
-}

pkg/auditor/config.go

@@ -63,6 +63,7 @@ type RetryConfig struct {
 
 func newConfig() factory.Config {
 	return Config{
+		Provider:      "noop",
 		BufferSize:    1000,
 		BatchSize:     100,
 		FlushInterval: time.Second,

pkg/instrumentation/sdk.go

@@ -14,7 +14,7 @@ import (
 	sdkmetric "go.opentelemetry.io/otel/metric"
 	sdkmetricnoop "go.opentelemetry.io/otel/metric/noop"
 	sdkresource "go.opentelemetry.io/otel/sdk/resource"
-	semconv "go.opentelemetry.io/otel/semconv/v1.39.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.40.0"
 	sdktrace "go.opentelemetry.io/otel/trace"
 )
 

pkg/modules/tracedetail/impltracedetail/handler.go

@@ -1,56 +0,0 @@
-package impltracedetail
-
-import (
-	"net/http"
-
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/http/binding"
-	"github.com/SigNoz/signoz/pkg/http/render"
-	"github.com/SigNoz/signoz/pkg/modules/tracedetail"
-	"github.com/SigNoz/signoz/pkg/types/authtypes"
-	"github.com/SigNoz/signoz/pkg/types/tracedetailtypes"
-	"github.com/SigNoz/signoz/pkg/valuer"
-	"github.com/gorilla/mux"
-)
-
-type handler struct {
-	module tracedetail.Module
-}
-
-func NewHandler(module tracedetail.Module) tracedetail.Handler {
-	return &handler{module: module}
-}
-
-func (h *handler) GetWaterfall(rw http.ResponseWriter, r *http.Request) {
-	claims, err := authtypes.ClaimsFromContext(r.Context())
-	if err != nil {
-		render.Error(rw, err)
-		return
-	}
-
-	orgID, err := valuer.NewUUID(claims.OrgID)
-	if err != nil {
-		render.Error(rw, err)
-		return
-	}
-
-	traceID := mux.Vars(r)["traceID"]
-	if traceID == "" {
-		render.Error(rw, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "traceId is required"))
-		return
-	}
-
-	var req tracedetailtypes.WaterfallRequest
-	if err := binding.JSON.BindBody(r.Body, &req); err != nil {
-		render.Error(rw, err)
-		return
-	}
-
-	result, err := h.module.GetWaterfall(r.Context(), orgID, traceID, &req)
-	if err != nil {
-		render.Error(rw, err)
-		return
-	}
-
-	render.Success(rw, http.StatusOK, result)
-}

pkg/modules/tracedetail/impltracedetail/module.go

@@ -1,272 +0,0 @@
-package impltracedetail
-
-import (
-	"context"
-	"database/sql"
-	"fmt"
-	"log/slog"
-	"sort"
-	"strconv"
-	"strings"
-	"time"
-
-	"github.com/SigNoz/signoz/pkg/cache"
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/factory"
-	"github.com/SigNoz/signoz/pkg/modules/tracedetail"
-	"github.com/SigNoz/signoz/pkg/telemetrystore"
-	"github.com/SigNoz/signoz/pkg/types/tracedetailtypes"
-	"github.com/SigNoz/signoz/pkg/valuer"
-
-	tracedetailv2 "github.com/SigNoz/signoz/pkg/query-service/app/traces/tracedetail"
-)
-
-const (
-	traceDB           = "signoz_traces"
-	traceTable        = "distributed_signoz_index_v3"
-	traceSummaryTable = "distributed_trace_summary"
-	cacheTTL          = 5 * time.Minute
-	fluxInterval      = 2 * time.Minute
-)
-
-type module struct {
-	telemetryStore telemetrystore.TelemetryStore
-	cache          cache.Cache
-	logger         *slog.Logger
-}
-
-func NewModule(telemetryStore telemetrystore.TelemetryStore, cache cache.Cache, providerSettings factory.ProviderSettings) tracedetail.Module {
-	return &module{
-		telemetryStore: telemetryStore,
-		cache:          cache,
-		logger:         providerSettings.Logger,
-	}
-}
-
-func (m *module) GetWaterfall(ctx context.Context, orgID valuer.UUID, traceID string, req *tracedetailtypes.WaterfallRequest) (*tracedetailtypes.WaterfallResponse, error) {
-	response := new(tracedetailtypes.WaterfallResponse)
-	var startTime, endTime, durationNano, totalErrorSpans, totalSpans uint64
-	var spanIDToSpanNodeMap = map[string]*tracedetailtypes.WaterfallSpan{}
-	var traceRoots []*tracedetailtypes.WaterfallSpan
-	var serviceNameToTotalDurationMap = map[string]uint64{}
-	var serviceNameIntervalMap = map[string][]tracedetailv2.Interval{}
-	var hasMissingSpans bool
-
-	// Try cache first
-	cachedTraceData, err := m.getFromCache(ctx, orgID, traceID)
-	if err == nil {
-		startTime = cachedTraceData.StartTime
-		endTime = cachedTraceData.EndTime
-		durationNano = cachedTraceData.DurationNano
-		spanIDToSpanNodeMap = cachedTraceData.SpanIDToSpanNodeMap
-		serviceNameToTotalDurationMap = cachedTraceData.ServiceNameToTotalDurationMap
-		traceRoots = cachedTraceData.TraceRoots
-		totalSpans = cachedTraceData.TotalSpans
-		totalErrorSpans = cachedTraceData.TotalErrorSpans
-		hasMissingSpans = cachedTraceData.HasMissingSpans
-	}
-
-	if err != nil {
-		m.logger.InfoContext(ctx, "cache miss for v3 waterfall", slog.String("trace_id", traceID))
-
-		// Query trace summary for time boundaries
-		var summary tracedetailtypes.TraceSummary
-		summaryQuery := fmt.Sprintf(
-			"SELECT trace_id, min(start) AS start, max(end) AS end, sum(num_spans) AS num_spans FROM %s.%s WHERE trace_id=$1 GROUP BY trace_id",
-			traceDB, traceSummaryTable,
-		)
-		err := m.telemetryStore.ClickhouseDB().QueryRow(ctx, summaryQuery, traceID).Scan(
-			&summary.TraceID, &summary.Start, &summary.End, &summary.NumSpans,
-		)
-		if err != nil {
-			if err == sql.ErrNoRows {
-				return response, nil
-			}
-			return nil, errors.Newf(errors.TypeInternal, errors.CodeInternal, "error querying trace summary: %v", err)
-		}
-
-		// Query span details
-		detailsQuery := fmt.Sprintf(
-			"SELECT DISTINCT ON (span_id) "+
-				"timestamp, duration_nano, span_id, trace_id, has_error, kind, "+
-				"resource_string_service$$name, name, links as references, "+
-				"attributes_string, attributes_number, attributes_bool, resources_string, "+
-				"events, status_message, status_code_string, kind_string, parent_span_id, "+
-				"flags, is_remote, trace_state, status_code, "+
-				"db_name, db_operation, http_method, http_url, http_host, "+
-				"external_http_method, external_http_url, response_status_code "+
-				"FROM %s.%s WHERE trace_id=$1 AND ts_bucket_start>=$2 AND ts_bucket_start<=$3 "+
-				"ORDER BY timestamp ASC, name ASC",
-			traceDB, traceTable,
-		)
-
-		var spanItems []tracedetailtypes.SpanModel
-		err = m.telemetryStore.ClickhouseDB().Select(
-			ctx, &spanItems, detailsQuery,
-			traceID,
-			strconv.FormatInt(summary.Start.Unix()-1800, 10),
-			strconv.FormatInt(summary.End.Unix(), 10),
-		)
-		if err != nil {
-			return nil, errors.Newf(errors.TypeInternal, errors.CodeInternal, "error querying trace spans: %v", err)
-		}
-
-		if len(spanItems) == 0 {
-			return response, nil
-		}
-
-		totalSpans = uint64(len(spanItems))
-		spanIDToSpanNodeMap = make(map[string]*tracedetailtypes.WaterfallSpan, len(spanItems))
-
-		// Build span nodes
-		for _, item := range spanItems {
-			span := item.ToSpan()
-			startTimeUnixNano := span.TimeUnixMilli
-
-			// Metadata calculation
-			if startTime == 0 || startTimeUnixNano < startTime {
-				startTime = startTimeUnixNano
-			}
-			if endTime == 0 || (startTimeUnixNano+span.DurationNano) > endTime {
-				endTime = startTimeUnixNano + span.DurationNano
-			}
-			if durationNano == 0 || span.DurationNano > durationNano {
-				durationNano = span.DurationNano
-			}
-			if span.HasError {
-				totalErrorSpans++
-			}
-
-			// Collect intervals for service execution time calculation
-			serviceNameIntervalMap[span.ServiceName] = append(
-				serviceNameIntervalMap[span.ServiceName],
-				tracedetailv2.Interval{StartTime: startTimeUnixNano, Duration: span.DurationNano, Service: span.ServiceName},
-			)
-
-			spanIDToSpanNodeMap[span.SpanID] = span
-		}
-
-		// Build tree: parent-child relationships and missing spans
-		for _, spanNode := range spanIDToSpanNodeMap {
-			if spanNode.ParentSpanID != "" {
-				if parentNode, exists := spanIDToSpanNodeMap[spanNode.ParentSpanID]; exists {
-					parentNode.Children = append(parentNode.Children, spanNode)
-				} else {
-					// Insert missing span
-					missingSpan := &tracedetailtypes.WaterfallSpan{
-						SpanID:        spanNode.ParentSpanID,
-						TraceID:       spanNode.TraceID,
-						Name:          "Missing Span",
-						TimeUnixMilli: spanNode.TimeUnixMilli,
-						DurationNano:  spanNode.DurationNano,
-						Events:        make([]tracedetailtypes.Event, 0),
-						Children:      make([]*tracedetailtypes.WaterfallSpan, 0),
-						Attributes:    make(map[string]any),
-						Resource:      make(map[string]string),
-					}
-					missingSpan.Children = append(missingSpan.Children, spanNode)
-					spanIDToSpanNodeMap[missingSpan.SpanID] = missingSpan
-					traceRoots = append(traceRoots, missingSpan)
-					hasMissingSpans = true
-				}
-			} else if !containsSpan(traceRoots, spanNode) {
-				traceRoots = append(traceRoots, spanNode)
-			}
-		}
-
-		// Sort children of each span for consistent ordering across requests.
-		for _, root := range traceRoots {
-			SortSpanChildren(root)
-		}
-
-		// Sort trace roots
-		sort.Slice(traceRoots, func(i, j int) bool {
-			if traceRoots[i].TimeUnixMilli == traceRoots[j].TimeUnixMilli {
-				return traceRoots[i].Name < traceRoots[j].Name
-			}
-			return traceRoots[i].TimeUnixMilli < traceRoots[j].TimeUnixMilli
-		})
-
-		serviceNameToTotalDurationMap = tracedetailv2.CalculateServiceTime(serviceNameIntervalMap)
-
-		// Cache the processed data
-		traceCache := &tracedetailtypes.WaterfallCache{
-			StartTime:                     startTime,
-			EndTime:                       endTime,
-			DurationNano:                  durationNano,
-			TotalSpans:                    totalSpans,
-			TotalErrorSpans:               totalErrorSpans,
-			SpanIDToSpanNodeMap:           spanIDToSpanNodeMap,
-			ServiceNameToTotalDurationMap: serviceNameToTotalDurationMap,
-			TraceRoots:                    traceRoots,
-			HasMissingSpans:               hasMissingSpans,
-		}
-		cacheKey := strings.Join([]string{"v3_waterfall", traceID}, "-")
-		if cacheErr := m.cache.Set(ctx, orgID, cacheKey, traceCache, cacheTTL); cacheErr != nil {
-			m.logger.DebugContext(ctx, "failed to store v3 waterfall cache", slog.String("trace_id", traceID), errors.Attr(cacheErr))
-		}
-	}
-
-	// Span selection: all spans or windowed
-	limit := min(req.Limit, MaxLimitToSelectAllSpans)
-	selectAllSpans := totalSpans <= uint64(limit)
-
-	var (
-		selectedSpans                          []*tracedetailtypes.WaterfallSpan
-		uncollapsedSpans                       []string
-		rootServiceName, rootServiceEntryPoint string
-	)
-	if selectAllSpans {
-		selectedSpans, rootServiceName, rootServiceEntryPoint = GetAllSpans(traceRoots)
-	} else {
-		selectedSpans, uncollapsedSpans, rootServiceName, rootServiceEntryPoint = GetSelectedSpans(
-			req.UncollapsedSpans, req.SelectedSpanID, traceRoots, spanIDToSpanNodeMap,
-		)
-	}
-
-	// Convert timestamps to milliseconds for service duration map
-	for serviceName, totalDuration := range serviceNameToTotalDurationMap {
-		serviceNameToTotalDurationMap[serviceName] = totalDuration / 1000000
-	}
-
-	response.Spans = selectedSpans
-	response.UncollapsedSpans = uncollapsedSpans
-	response.StartTimestampMillis = startTime / 1000000
-	response.EndTimestampMillis = endTime / 1000000
-	response.TotalSpansCount = totalSpans
-	response.TotalErrorSpansCount = totalErrorSpans
-	response.RootServiceName = rootServiceName
-	response.RootServiceEntryPoint = rootServiceEntryPoint
-	response.ServiceNameToTotalDurationMap = serviceNameToTotalDurationMap
-	response.HasMissingSpans = hasMissingSpans
-	response.HasMore = !selectAllSpans
-
-	return response, nil
-}
-
-func (m *module) getFromCache(ctx context.Context, orgID valuer.UUID, traceID string) (*tracedetailtypes.WaterfallCache, error) {
-	cachedData := new(tracedetailtypes.WaterfallCache)
-	cacheKey := strings.Join([]string{"v3_waterfall", traceID}, "-")
-	err := m.cache.Get(ctx, orgID, cacheKey, cachedData)
-	if err != nil {
-		return nil, err
-	}
-
-	// Skip cache if trace end time falls within flux interval
-	if time.Since(time.UnixMilli(int64(cachedData.EndTime))) < fluxInterval {
-		m.logger.InfoContext(ctx, "trace end time within flux interval, skipping v3 waterfall cache", slog.String("trace_id", traceID))
-		return nil, errors.Newf(errors.TypeInternal, errors.CodeInternal, "trace end time within flux interval, traceID: %s", traceID)
-	}
-
-	m.logger.InfoContext(ctx, "cache hit for v3 waterfall", slog.String("trace_id", traceID))
-	return cachedData, nil
-}
-
-func containsSpan(spans []*tracedetailtypes.WaterfallSpan, target *tracedetailtypes.WaterfallSpan) bool {
-	for _, s := range spans {
-		if s.SpanID == target.SpanID {
-			return true
-		}
-	}
-	return false
-}

pkg/modules/tracedetail/impltracedetail/waterfall.go

@@ -1,191 +0,0 @@
-package impltracedetail
-
-import (
-	"maps"
-	"slices"
-	"sort"
-
-	"github.com/SigNoz/signoz/pkg/types/tracedetailtypes"
-)
-
-var (
-	spanLimitPerRequest         float64 = 500
-	maxDepthForSelectedChildren int     = 5
-	MaxLimitToSelectAllSpans    uint    = 10_000
-)
-
-type traverseOpts struct {
-	uncollapsedSpans map[string]struct{}
-	selectedSpanID   string
-	selectAll        bool
-}
-
-func traverseTrace(
-	span *tracedetailtypes.WaterfallSpan,
-	opts traverseOpts,
-	level uint64,
-	isPartOfPreOrder bool,
-	autoExpandDepth int,
-) ([]*tracedetailtypes.WaterfallSpan, []string) {
-
-	preOrderTraversal := []*tracedetailtypes.WaterfallSpan{}
-	autoExpandedSpans := []string{}
-
-	span.SubTreeNodeCount = 0
-	nodeWithoutChildren := span.CopyWithoutChildren(level)
-
-	if isPartOfPreOrder {
-		preOrderTraversal = append(preOrderTraversal, nodeWithoutChildren)
-	}
-
-	remainingAutoExpandDepth := 0
-	_, isSelectedSpanUncollapsed := opts.uncollapsedSpans[opts.selectedSpanID]
-	if span.SpanID == opts.selectedSpanID && isSelectedSpanUncollapsed {
-		remainingAutoExpandDepth = maxDepthForSelectedChildren
-	} else if autoExpandDepth > 0 {
-		remainingAutoExpandDepth = autoExpandDepth - 1
-	}
-
-	_, isAlreadyUncollapsed := opts.uncollapsedSpans[span.SpanID]
-	for _, child := range span.Children {
-		isChildWithinMaxDepth := remainingAutoExpandDepth > 0
-		childIsPartOfPreOrder := opts.selectAll || (isPartOfPreOrder && (isAlreadyUncollapsed || isChildWithinMaxDepth))
-
-		if isPartOfPreOrder && isChildWithinMaxDepth && !isAlreadyUncollapsed {
-			if !slices.Contains(autoExpandedSpans, span.SpanID) {
-				autoExpandedSpans = append(autoExpandedSpans, span.SpanID)
-			}
-		}
-
-		childTraversal, childAutoExpanded := traverseTrace(child, opts, level+1, childIsPartOfPreOrder, remainingAutoExpandDepth)
-		preOrderTraversal = append(preOrderTraversal, childTraversal...)
-		autoExpandedSpans = append(autoExpandedSpans, childAutoExpanded...)
-		nodeWithoutChildren.SubTreeNodeCount += child.SubTreeNodeCount + 1
-		span.SubTreeNodeCount += child.SubTreeNodeCount + 1
-	}
-
-	nodeWithoutChildren.SubTreeNodeCount += 1
-	return preOrderTraversal, autoExpandedSpans
-}
-
-func GetSelectedSpans(uncollapsedSpans []string, selectedSpanID string, traceRoots []*tracedetailtypes.WaterfallSpan, spanIDToSpanNodeMap map[string]*tracedetailtypes.WaterfallSpan) ([]*tracedetailtypes.WaterfallSpan, []string, string, string) {
-	var preOrderTraversal = make([]*tracedetailtypes.WaterfallSpan, 0)
-	var rootServiceName, rootServiceEntryPoint string
-
-	uncollapsedSpanMap := make(map[string]struct{})
-	for _, spanID := range uncollapsedSpans {
-		uncollapsedSpanMap[spanID] = struct{}{}
-	}
-
-	selectedSpanIndex := -1
-	for _, rootSpanID := range traceRoots {
-		if rootNode, exists := spanIDToSpanNodeMap[rootSpanID.SpanID]; exists {
-			present, spansFromRootToNode := getPathFromRootToSelectedSpanID(rootNode, selectedSpanID)
-			if present {
-				for _, spanID := range spansFromRootToNode {
-					if selectedSpanID == spanID {
-						continue
-					}
-					uncollapsedSpanMap[spanID] = struct{}{}
-				}
-			}
-
-			opts := traverseOpts{
-				uncollapsedSpans: uncollapsedSpanMap,
-				selectedSpanID:   selectedSpanID,
-			}
-			traversal, autoExpanded := traverseTrace(rootNode, opts, 0, true, 0)
-			for _, spanID := range autoExpanded {
-				uncollapsedSpanMap[spanID] = struct{}{}
-			}
-			idx := findIndexForSelectedSpan(traversal, selectedSpanID)
-
-			if idx != -1 {
-				selectedSpanIndex = idx + len(preOrderTraversal)
-			}
-
-			preOrderTraversal = append(preOrderTraversal, traversal...)
-
-			if rootServiceName == "" {
-				rootServiceName = rootNode.ServiceName
-			}
-			if rootServiceEntryPoint == "" {
-				rootServiceEntryPoint = rootNode.Name
-			}
-		}
-	}
-
-	if selectedSpanIndex == -1 && selectedSpanID != "" {
-		selectedSpanIndex = 0
-	}
-
-	// Window: 40% before, 60% after selected span
-	startIndex := selectedSpanIndex - int(spanLimitPerRequest*0.4)
-	endIndex := selectedSpanIndex + int(spanLimitPerRequest*0.6)
-
-	if startIndex < 0 {
-		endIndex = endIndex - startIndex
-		startIndex = 0
-	}
-	if endIndex > len(preOrderTraversal) {
-		startIndex = startIndex - (endIndex - len(preOrderTraversal))
-		endIndex = len(preOrderTraversal)
-	}
-	if startIndex < 0 {
-		startIndex = 0
-	}
-
-	return preOrderTraversal[startIndex:endIndex], slices.Collect(maps.Keys(uncollapsedSpanMap)), rootServiceName, rootServiceEntryPoint
-}
-
-func GetAllSpans(traceRoots []*tracedetailtypes.WaterfallSpan) (spans []*tracedetailtypes.WaterfallSpan, rootServiceName, rootEntryPoint string) {
-	if len(traceRoots) > 0 {
-		rootServiceName = traceRoots[0].ServiceName
-		rootEntryPoint = traceRoots[0].Name
-	}
-	for _, root := range traceRoots {
-		childSpans, _ := traverseTrace(root, traverseOpts{selectAll: true}, 0, true, 0)
-		spans = append(spans, childSpans...)
-	}
-	return
-}
-
-func getPathFromRootToSelectedSpanID(node *tracedetailtypes.WaterfallSpan, selectedSpanID string) (bool, []string) {
-	path := []string{node.SpanID}
-	if node.SpanID == selectedSpanID {
-		return true, path
-	}
-
-	for _, child := range node.Children {
-		found, childPath := getPathFromRootToSelectedSpanID(child, selectedSpanID)
-		if found {
-			path = append(path, childPath...)
-			return true, path
-		}
-	}
-	return false, nil
-}
-
-func findIndexForSelectedSpan(spans []*tracedetailtypes.WaterfallSpan, selectedSpanID string) int {
-	for i, span := range spans {
-		if span.SpanID == selectedSpanID {
-			return i
-		}
-	}
-	return -1
-}
-
-// SortSpanChildren recursively sorts children of each span by TimeUnixNano then Name.
-// Must be called once after the span tree is fully built so that traverseTrace
-// sees a consistent ordering without needing to re-sort on every call.
-func SortSpanChildren(span *tracedetailtypes.WaterfallSpan) {
-	sort.Slice(span.Children, func(i, j int) bool {
-		if span.Children[i].TimeUnixMilli == span.Children[j].TimeUnixMilli {
-			return span.Children[i].Name < span.Children[j].Name
-		}
-		return span.Children[i].TimeUnixMilli < span.Children[j].TimeUnixMilli
-	})
-	for _, child := range span.Children {
-		SortSpanChildren(child)
-	}
-}

pkg/modules/tracedetail/impltracedetail/waterfall_test.go

@@ -1,559 +0,0 @@
-// Package impltracedetail tests — waterfall
-//
-// # Background
-//
-// The waterfall view renders a trace as a scrollable list of spans in
-// pre-order (parent before children, siblings left-to-right). Because a trace
-// can have thousands of spans, only a window of ~500 is returned per request.
-// The window is centred on the selected span.
-//
-// # Key concepts
-//
-// uncollapsedSpans
-//
-//	The set of span IDs the user has manually expanded in the UI.
-//	Only the direct children of an uncollapsed span are included in the
-//	output; grandchildren stay hidden until their parent is also uncollapsed.
-//	When multiple spans are uncollapsed their children are all visible at once.
-//
-// selectedSpanID
-//
-//	The span currently focused — set when the user clicks a span in the
-//	waterfall or selects one from the flamegraph.  The output window is always
-//	centred on this span.  The path from the trace root down to the selected
-//	span is automatically uncollapsed so ancestors are visible even if they are
-//	not in uncollapsedSpans.
-//
-//
-// traceRoots
-//
-//	Root spans of the trace — spans with no parent in the current dataset.
-//	Normally one, but multiple roots are common when upstream services are
-//	not instrumented or their spans were not sampled/exported.
-
-package impltracedetail
-
-import (
-	"fmt"
-	"testing"
-
-	"github.com/SigNoz/signoz/pkg/types/tracedetailtypes"
-	"github.com/stretchr/testify/assert"
-)
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Helpers
-// ─────────────────────────────────────────────────────────────────────────────
-
-func mkSpan(id, service string, children ...*tracedetailtypes.WaterfallSpan) *tracedetailtypes.WaterfallSpan {
-	return &tracedetailtypes.WaterfallSpan{
-		SpanID:      id,
-		ServiceName: service,
-		Name:        id + "-op",
-		Children:    children,
-	}
-}
-
-func spanIDs(spans []*tracedetailtypes.WaterfallSpan) []string {
-	ids := make([]string, len(spans))
-	for i, s := range spans {
-		ids[i] = s.SpanID
-	}
-	return ids
-}
-
-func buildSpanMap(roots ...*tracedetailtypes.WaterfallSpan) map[string]*tracedetailtypes.WaterfallSpan {
-	m := map[string]*tracedetailtypes.WaterfallSpan{}
-	var walk func(*tracedetailtypes.WaterfallSpan)
-	walk = func(s *tracedetailtypes.WaterfallSpan) {
-		m[s.SpanID] = s
-		for _, c := range s.Children {
-			walk(c)
-		}
-	}
-	for _, r := range roots {
-		SortSpanChildren(r)
-		walk(r)
-	}
-	return m
-}
-
-// makeChain builds a linear trace: span0 → span1 → … → span(n-1).
-func makeChain(n int) (*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan, []string) {
-	spans := make([]*tracedetailtypes.WaterfallSpan, n)
-	for i := n - 1; i >= 0; i-- {
-		if i == n-1 {
-			spans[i] = mkSpan(fmt.Sprintf("span%d", i), "svc")
-		} else {
-			spans[i] = mkSpan(fmt.Sprintf("span%d", i), "svc", spans[i+1])
-		}
-	}
-	uncollapsed := make([]string, n)
-	for i := range spans {
-		uncollapsed[i] = fmt.Sprintf("span%d", i)
-	}
-	return spans[0], buildSpanMap(spans[0]), uncollapsed
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// GetSelectedSpans — span ordering and visibility
-// ─────────────────────────────────────────────────────────────────────────────
-
-func TestGetSelectedSpans_SpanOrdering(t *testing.T) {
-	tests := []struct {
-		name             string
-		buildRoots       func() ([]*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan)
-		uncollapsedSpans []string
-		selectedSpanID   string
-		wantSpanIDs      []string
-	}{
-		{
-			// Pre-order traversal is preserved: parent before children, siblings left-to-right.
-			name: "pre_order_traversal",
-			buildRoots: func() ([]*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("child1", "svc", mkSpan("grandchild", "svc")),
-					mkSpan("child2", "svc"),
-				)
-				return []*tracedetailtypes.WaterfallSpan{root}, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{"root", "child1"},
-			selectedSpanID:   "root",
-			wantSpanIDs:      []string{"root", "child1", "grandchild", "child2"},
-		},
-		{
-			// Multiple spans uncollapsed simultaneously: children of all uncollapsed spans are visible at once.
-			//
-			//	root
-			//	  ├─ childA (uncollapsed) → grandchildA ✓
-			//	  └─ childB (uncollapsed) → grandchildB ✓
-			name: "multiple_uncollapsed",
-			buildRoots: func() ([]*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("childA", "svc", mkSpan("grandchildA", "svc")),
-					mkSpan("childB", "svc", mkSpan("grandchildB", "svc")),
-				)
-				return []*tracedetailtypes.WaterfallSpan{root}, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{"root", "childA", "childB"},
-			selectedSpanID:   "root",
-			wantSpanIDs:      []string{"root", "childA", "grandchildA", "childB", "grandchildB"},
-		},
-		{
-			// Collapsing a span with other uncollapsed spans.
-			//
-			// root
-			// ├─ childA  (previously expanded — in uncollapsedSpans)
-			// │    ├─ grandchild1 ✓
-			// │    │    └─ greatGrandchild ✗  (grandchild1 not in uncollapsedSpans)
-			// │    └─ grandchild2 ✓
-			// └─ childB                        ← selected (not expanded)
-			name: "manual_uncollapse",
-			buildRoots: func() ([]*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("childA", "svc",
-						mkSpan("grandchild1", "svc", mkSpan("greatGrandchild", "svc")),
-						mkSpan("grandchild2", "svc"),
-					),
-					mkSpan("childB", "svc"),
-				)
-				return []*tracedetailtypes.WaterfallSpan{root}, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{"childA"},
-			selectedSpanID:   "childB",
-			wantSpanIDs:      []string{"root", "childA", "grandchild1", "grandchild2", "childB"},
-		},
-		{
-			// A collapsed span hides all children.
-			name: "collapsed_span_hides_children",
-			buildRoots: func() ([]*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("child1", "svc"),
-					mkSpan("child2", "svc"),
-				)
-				return []*tracedetailtypes.WaterfallSpan{root}, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{},
-			selectedSpanID:   "root",
-			wantSpanIDs:      []string{"root"},
-		},
-		{
-			// Selecting a span auto-uncollpases the path from root to that span so it is visible.
-			//
-			//	root → parent → selected
-			name: "path_to_selected_is_uncollapsed",
-			buildRoots: func() ([]*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("parent", "svc",
-						mkSpan("selected", "svc"),
-					),
-				)
-				return []*tracedetailtypes.WaterfallSpan{root}, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{},
-			selectedSpanID:   "selected",
-			wantSpanIDs:      []string{"root", "parent", "selected"},
-		},
-		{
-			// Siblings of ancestors are rendered as collapsed nodes but their subtrees must NOT be expanded.
-			//
-			//	root
-			//	  ├─ unrelated → unrelated-child (✗)
-			//	  └─ parent → selected
-			name: "siblings_not_expanded",
-			buildRoots: func() ([]*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("unrelated", "svc", mkSpan("unrelated-child", "svc")),
-					mkSpan("parent", "svc",
-						mkSpan("selected", "svc"),
-					),
-				)
-				return []*tracedetailtypes.WaterfallSpan{root}, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{},
-			selectedSpanID:   "selected",
-			// children of root sort alphabetically: parent < unrelated; unrelated-child stays hidden
-			wantSpanIDs: []string{"root", "parent", "selected", "unrelated"},
-		},
-		{
-			// An unknown selectedSpanID must not panic; returns a window from index 0.
-			name: "unknown_selected_span",
-			buildRoots: func() ([]*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc", mkSpan("child", "svc"))
-				return []*tracedetailtypes.WaterfallSpan{root}, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{},
-			selectedSpanID:   "nonexistent",
-			wantSpanIDs:      []string{"root"},
-		},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			roots, spanMap := tc.buildRoots()
-			spans, _, _, _ := GetSelectedSpans(tc.uncollapsedSpans, tc.selectedSpanID, roots, spanMap)
-			assert.Equal(t, tc.wantSpanIDs, spanIDs(spans))
-		})
-	}
-}
-
-// Multiple roots: both trees are flattened into a single pre-order list with
-// root1's subtree before root2's. Service/entry-point come from the first root.
-//
-//	root1 svc-a          ← selected
-//	  └─ child1
-//	root2 svc-b
-//	  └─ child2
-//
-// Expected output order: root1 → child1 → root2 → child2.
-func TestGetSelectedSpans_MultipleRoots(t *testing.T) {
-	root1 := mkSpan("root1", "svc-a", mkSpan("child1", "svc-a"))
-	root2 := mkSpan("root2", "svc-b", mkSpan("child2", "svc-b"))
-	spanMap := buildSpanMap(root1, root2)
-
-	spans, _, svcName, entryPoint := GetSelectedSpans([]string{"root1", "root2"}, "root1", []*tracedetailtypes.WaterfallSpan{root1, root2}, spanMap)
-
-	assert.Equal(t, []string{"root1", "child1", "root2", "child2"}, spanIDs(spans), "root1 subtree must precede root2 subtree")
-	assert.Equal(t, "svc-a", svcName, "metadata comes from first root")
-	assert.Equal(t, "root1-op", entryPoint, "metadata comes from first root")
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// GetSelectedSpans — uncollapsed span tracking
-// ─────────────────────────────────────────────────────────────────────────────
-
-func TestGetSelectedSpans_UncollapsedTracking(t *testing.T) {
-	tests := []struct {
-		name             string
-		buildRoot        func() (*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan)
-		uncollapsedSpans []string
-		selectedSpanID   string
-		wantSpanIDs      []string
-		checkUncollapsed func(t *testing.T, uncollapsed []string)
-	}{
-		{
-			// The path-to-selected spans are returned in updatedUncollapsedSpans.
-			name: "path_returned_in_uncollapsed",
-			buildRoot: func() (*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("parent", "svc",
-						mkSpan("selected", "svc"),
-					),
-				)
-				return root, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{},
-			selectedSpanID:   "selected",
-			wantSpanIDs:      []string{"root", "parent", "selected"},
-			checkUncollapsed: func(t *testing.T, uncollapsed []string) {
-				assert.ElementsMatch(t, []string{"root", "parent"}, uncollapsed)
-			},
-		},
-		{
-			// Siblings of ancestors are not tracked as uncollapsed.
-			name: "siblings_not_in_uncollapsed",
-			buildRoot: func() (*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("unrelated", "svc", mkSpan("unrelated-child", "svc")),
-					mkSpan("parent", "svc",
-						mkSpan("selected", "svc"),
-					),
-				)
-				return root, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{},
-			selectedSpanID:   "selected",
-			wantSpanIDs:      []string{"root", "parent", "selected", "unrelated"},
-			checkUncollapsed: func(t *testing.T, uncollapsed []string) {
-				assert.ElementsMatch(t, []string{"root", "parent"}, uncollapsed)
-			},
-		},
-		{
-			// Auto-expanded span IDs from ALL branches are returned in updatedUncollapsedSpans.
-			// Only internal nodes (spans with children) are tracked — leaf spans are never added.
-			// root is in uncollapsedSpans, so its children are auto-expanded.
-			//
-			//	root (selected, expanded)
-			//	  ├─ childA (internal ✓)
-			//	  │    └─ grandchildA (internal ✓)
-			//	  │         └─ leafA (leaf ✗)
-			//	  └─ childB (internal ✓)
-			//	       └─ grandchildB (internal ✓)
-			//	            └─ leafB (leaf ✗)
-			name: "auto_expanded_spans_returned",
-			buildRoot: func() (*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("childA", "svc",
-						mkSpan("grandchildA", "svc",
-							mkSpan("leafA", "svc"),
-						),
-					),
-					mkSpan("childB", "svc",
-						mkSpan("grandchildB", "svc",
-							mkSpan("leafB", "svc"),
-						),
-					),
-				)
-				return root, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{"root"},
-			selectedSpanID:   "root",
-			checkUncollapsed: func(t *testing.T, uncollapsed []string) {
-				assert.Contains(t, uncollapsed, "root")
-				assert.Contains(t, uncollapsed, "childA", "internal node depth 1, branch A")
-				assert.Contains(t, uncollapsed, "childB", "internal node depth 1, branch B")
-				assert.Contains(t, uncollapsed, "grandchildA", "internal node depth 2, branch A")
-				assert.Contains(t, uncollapsed, "grandchildB", "internal node depth 2, branch B")
-				assert.NotContains(t, uncollapsed, "leafA", "leaf spans are never added to uncollapsedSpans")
-				assert.NotContains(t, uncollapsed, "leafB", "leaf spans are never added to uncollapsedSpans")
-			},
-		},
-		{
-			// If the selected span is already in uncollapsedSpans,
-			// it should appear exactly once in the result.
-			name: "duplicate_in_uncollapsed",
-			buildRoot: func() (*tracedetailtypes.WaterfallSpan, map[string]*tracedetailtypes.WaterfallSpan) {
-				root := mkSpan("root", "svc",
-					mkSpan("selected", "svc", mkSpan("child", "svc")),
-				)
-				return root, buildSpanMap(root)
-			},
-			uncollapsedSpans: []string{"selected"}, // already present
-			selectedSpanID:   "selected",
-			checkUncollapsed: func(t *testing.T, uncollapsed []string) {
-				count := 0
-				for _, id := range uncollapsed {
-					if id == "selected" {
-						count++
-					}
-				}
-				assert.Equal(t, 1, count, "should appear once")
-			},
-		},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			root, spanMap := tc.buildRoot()
-			spans, uncollapsed, _, _ := GetSelectedSpans(tc.uncollapsedSpans, tc.selectedSpanID, []*tracedetailtypes.WaterfallSpan{root}, spanMap)
-			if tc.wantSpanIDs != nil {
-				assert.Equal(t, tc.wantSpanIDs, spanIDs(spans))
-			}
-			if tc.checkUncollapsed != nil {
-				tc.checkUncollapsed(t, uncollapsed)
-			}
-		})
-	}
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// GetSelectedSpans — span metadata
-// ─────────────────────────────────────────────────────────────────────────────
-
-// Test to check if Level, HasChildren, and SubTreeNodeCount are populated correctly.
-//
-//	root          level=0, hasChildren=true,  subTree=4
-//	  child1      level=1, hasChildren=true,  subTree=2
-//	    grandchild level=2, hasChildren=false, subTree=1
-//	  child2      level=1, hasChildren=false, subTree=1
-func TestGetSelectedSpans_SpanMetadata(t *testing.T) {
-	root := mkSpan("root", "svc",
-		mkSpan("child1", "svc", mkSpan("grandchild", "svc")),
-		mkSpan("child2", "svc"),
-	)
-	spanMap := buildSpanMap(root)
-	spans, _, _, _ := GetSelectedSpans([]string{"root", "child1"}, "root", []*tracedetailtypes.WaterfallSpan{root}, spanMap)
-
-	byID := map[string]*tracedetailtypes.WaterfallSpan{}
-	for _, s := range spans {
-		byID[s.SpanID] = s
-	}
-
-	tests := []struct {
-		spanID          string
-		wantLevel       uint64
-		wantHasChildren bool
-		wantSubTree     uint64
-	}{
-		{"root", 0, true, 4},
-		{"child1", 1, true, 2},
-		{"child2", 1, false, 1},
-		{"grandchild", 2, false, 1},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.spanID, func(t *testing.T) {
-			s := byID[tc.spanID]
-			assert.Equal(t, tc.wantLevel, s.Level)
-			assert.Equal(t, tc.wantHasChildren, s.HasChildren)
-			assert.Equal(t, tc.wantSubTree, s.SubTreeNodeCount)
-		})
-	}
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// GetSelectedSpans — windowing
-// ─────────────────────────────────────────────────────────────────────────────
-
-func TestGetSelectedSpans_Window(t *testing.T) {
-	tests := []struct {
-		name            string
-		selectedSpanID  string
-		wantLen         int
-		wantFirst       string
-		wantLast        string
-		wantSelectedPos int
-	}{
-		{
-			// The selected span is centred: 200 spans before it, 300 after (0.4 / 0.6 split).
-			name:            "centred_on_selected",
-			selectedSpanID:  "span300",
-			wantLen:         500,
-			wantFirst:       "span100",
-			wantLast:        "span599",
-			wantSelectedPos: 200,
-		},
-		{
-			// When the selected span is near the start, the window shifts right so no
-			// negative index is used — the result is still 500 spans.
-			name:            "shifts_at_start",
-			selectedSpanID:  "span10",
-			wantLen:         500,
-			wantFirst:       "span0",
-			wantSelectedPos: 10,
-		},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			root, spanMap, uncollapsed := makeChain(600)
-			spans, _, _, _ := GetSelectedSpans(uncollapsed, tc.selectedSpanID, []*tracedetailtypes.WaterfallSpan{root}, spanMap)
-
-			assert.Equal(t, tc.wantLen, len(spans), "window size")
-			assert.Equal(t, tc.wantFirst, spans[0].SpanID, "first span in window")
-			if tc.wantLast != "" {
-				assert.Equal(t, tc.wantLast, spans[len(spans)-1].SpanID, "last span in window")
-			}
-			assert.Equal(t, tc.selectedSpanID, spans[tc.wantSelectedPos].SpanID, "selected span position")
-		})
-	}
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// GetSelectedSpans — depth limit
-// ─────────────────────────────────────────────────────────────────────────────
-
-// Depth is measured from the selected span, not the trace root.
-// Ancestors appear via the path-to-root logic, not the depth limit.
-// Each depth level has two children to confirm the limit is enforced on all
-// branches, not just the first.
-//
-//	root
-//	  └─ A                           ancestor ✓ (path-to-root)
-//	       └─ selected
-//	            ├─ d1a               depth 1 ✓
-//	            │    ├─ d2a          depth 2 ✓
-//	            │    │    ├─ d3a     depth 3 ✓
-//	            │    │    │    ├─ d4a   depth 4 ✓
-//	            │    │    │    │    ├─ d5a  depth 5 ✓
-//	            │    │    │    │    │    └─ d6a  depth 6 ✗
-//	            │    │    │    │    └─ d5b  depth 5 ✓
-//	            │    │    │    └─ d4b   depth 4 ✓
-//	            │    │    └─ d3b     depth 3 ✓
-//	            │    └─ d2b          depth 2 ✓
-//	            └─ d1b               depth 1 ✓
-func TestGetSelectedSpans_DepthCountedFromSelectedSpan(t *testing.T) {
-	selected := mkSpan("selected", "svc",
-		mkSpan("d1a", "svc",
-			mkSpan("d2a", "svc",
-				mkSpan("d3a", "svc",
-					mkSpan("d4a", "svc",
-						mkSpan("d5a", "svc",
-							mkSpan("d6a", "svc"), // depth 6 — excluded
-						),
-						mkSpan("d5b", "svc"), // depth 5 — included
-					),
-					mkSpan("d4b", "svc"), // depth 4 — included
-				),
-				mkSpan("d3b", "svc"), // depth 3 — included
-			),
-			mkSpan("d2b", "svc"), // depth 2 — included
-		),
-		mkSpan("d1b", "svc"), // depth 1 — included
-	)
-	root := mkSpan("root", "svc", mkSpan("A", "svc", selected))
-
-	spanMap := buildSpanMap(root)
-	spans, _, _, _ := GetSelectedSpans([]string{"selected"}, "selected", []*tracedetailtypes.WaterfallSpan{root}, spanMap)
-	ids := spanIDs(spans)
-
-	assert.Contains(t, ids, "root", "ancestor shown via path-to-root")
-	assert.Contains(t, ids, "A", "ancestor shown via path-to-root")
-	for _, id := range []string{"d1a", "d1b", "d2a", "d2b", "d3a", "d3b", "d4a", "d4b", "d5a", "d5b"} {
-		assert.Contains(t, ids, id, "depth ≤ 5 — must be included")
-	}
-	assert.NotContains(t, ids, "d6a", "depth 6 > limit — excluded")
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// GetAllSpans
-// ─────────────────────────────────────────────────────────────────────────────
-
-func TestGetAllSpans(t *testing.T) {
-	root := mkSpan("root", "svc",
-		mkSpan("childA", "svc",
-			mkSpan("grandchildA", "svc",
-				mkSpan("leafA", "svc2"),
-			),
-		),
-		mkSpan("childB", "svc3",
-			mkSpan("grandchildB", "svc",
-				mkSpan("leafB", "svc2"),
-			),
-		),
-	)
-	spans, rootServiceName, rootEntryPoint := GetAllSpans([]*tracedetailtypes.WaterfallSpan{root})
-	assert.ElementsMatch(t, spanIDs(spans), []string{"root", "childA", "grandchildA", "leafA", "childB", "grandchildB", "leafB"})
-	assert.Equal(t, "svc", rootServiceName)
-	assert.Equal(t, "root-op", rootEntryPoint)
-}

pkg/querier/querier.go

@@ -40,6 +40,7 @@ type querier struct {
 	promEngine               prometheus.Prometheus
 	traceStmtBuilder         qbtypes.StatementBuilder[qbtypes.TraceAggregation]
 	logStmtBuilder           qbtypes.StatementBuilder[qbtypes.LogAggregation]
+	auditStmtBuilder         qbtypes.StatementBuilder[qbtypes.LogAggregation]
 	metricStmtBuilder        qbtypes.StatementBuilder[qbtypes.MetricAggregation]
 	meterStmtBuilder         qbtypes.StatementBuilder[qbtypes.MetricAggregation]
 	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder
@@ -56,6 +57,7 @@ func New(
 	promEngine prometheus.Prometheus,
 	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
+	auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder,
@@ -69,6 +71,7 @@ func New(
 		promEngine:               promEngine,
 		traceStmtBuilder:         traceStmtBuilder,
 		logStmtBuilder:           logStmtBuilder,
+		auditStmtBuilder:         auditStmtBuilder,
 		metricStmtBuilder:        metricStmtBuilder,
 		meterStmtBuilder:         meterStmtBuilder,
 		traceOperatorStmtBuilder: traceOperatorStmtBuilder,
@@ -361,7 +364,11 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 			case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
 				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
-				bq := newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
+				stmtBuilder := q.logStmtBuilder
+				if spec.Source == telemetrytypes.SourceAudit {
+					stmtBuilder = q.auditStmtBuilder
+				}
+				bq := newBuilderQuery(q.logger, q.telemetryStore, stmtBuilder, spec, timeRange, req.RequestType, tmplVars)
 				queries[spec.Name] = bq
 				steps[spec.Name] = spec.StepInterval
 			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
@@ -550,7 +557,11 @@ func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qb
 		case <-tick:
 			// timestamp end is not specified here
 			timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: tsStart}, req.RequestType)
-			bq := newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
+			liveTailStmtBuilder := q.logStmtBuilder
+			if spec.Source == telemetrytypes.SourceAudit {
+				liveTailStmtBuilder = q.auditStmtBuilder
+			}
+			bq := newBuilderQuery(q.logger, q.telemetryStore, liveTailStmtBuilder, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
 				"id": {
 					Value: updatedLogID,
 				},
@@ -850,7 +861,11 @@ func (q *querier) createRangedQuery(originalQuery qbtypes.Query, timeRange qbtyp
 		specCopy := qt.spec.Copy()
 		specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
 		adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
-		return newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
+		shiftStmtBuilder := q.logStmtBuilder
+		if qt.spec.Source == telemetrytypes.SourceAudit {
+			shiftStmtBuilder = q.auditStmtBuilder
+		}
+		return newBuilderQuery(q.logger, q.telemetryStore, shiftStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
 
 	case *builderQuery[qbtypes.MetricAggregation]:
 		specCopy := qt.spec.Copy()

pkg/querier/querier_test.go

@@ -47,6 +47,7 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
 		nil, // prometheus
 		nil, // traceStmtBuilder
 		nil, // logStmtBuilder
+		nil, // auditStmtBuilder
 		nil, // metricStmtBuilder
 		nil, // meterStmtBuilder
 		nil, // traceOperatorStmtBuilder
@@ -110,6 +111,7 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 		nil,                      // prometheus
 		nil,                      // traceStmtBuilder
 		nil,                      // logStmtBuilder
+		nil,                      // auditStmtBuilder
 		&mockMetricStmtBuilder{}, // metricStmtBuilder
 		nil,                      // meterStmtBuilder
 		nil,                      // traceOperatorStmtBuilder

pkg/querier/signozquerier/provider.go

@@ -9,6 +9,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/prometheus"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymetadata"
 	"github.com/SigNoz/signoz/pkg/telemetrymeter"
@@ -63,6 +64,11 @@ func newProvider(
 		telemetrylogs.TagAttributesV2TableName,
 		telemetrylogs.LogAttributeKeysTblName,
 		telemetrylogs.LogResourceKeysTblName,
+		telemetryaudit.DBName,
+		telemetryaudit.AuditLogsTableName,
+		telemetryaudit.TagAttributesTableName,
+		telemetryaudit.LogAttributeKeysTblName,
+		telemetryaudit.LogResourceKeysTblName,
 		telemetrymetadata.DBName,
 		telemetrymetadata.AttributesMetadataLocalTableName,
 		telemetrymetadata.ColumnEvolutionMetadataTableName,
@@ -82,13 +88,13 @@ func newProvider(
 		telemetryStore,
 	)
 
-	// ADD: Create trace operator statement builder
+	// Create trace operator statement builder
 	traceOperatorStmtBuilder := telemetrytraces.NewTraceOperatorStatementBuilder(
 		settings,
 		telemetryMetadataStore,
 		traceFieldMapper,
 		traceConditionBuilder,
-		traceStmtBuilder, // Pass the regular trace statement builder
+		traceStmtBuilder,
 		traceAggExprRewriter,
 	)
 
@@ -112,6 +118,26 @@ func newProvider(
 		telemetrylogs.GetBodyJSONKey,
 	)
 
+	// Create audit statement builder
+	auditFieldMapper := telemetryaudit.NewFieldMapper()
+	auditConditionBuilder := telemetryaudit.NewConditionBuilder(auditFieldMapper)
+	auditAggExprRewriter := querybuilder.NewAggExprRewriter(
+		settings,
+		telemetryaudit.DefaultFullTextColumn,
+		auditFieldMapper,
+		auditConditionBuilder,
+		nil,
+	)
+	auditStmtBuilder := telemetryaudit.NewAuditQueryStatementBuilder(
+		settings,
+		telemetryMetadataStore,
+		auditFieldMapper,
+		auditConditionBuilder,
+		auditAggExprRewriter,
+		telemetryaudit.DefaultFullTextColumn,
+		nil,
+	)
+
 	// Create metric statement builder
 	metricFieldMapper := telemetrymetrics.NewFieldMapper()
 	metricConditionBuilder := telemetrymetrics.NewConditionBuilder(metricFieldMapper)
@@ -148,6 +174,7 @@ func newProvider(
 		prometheus,
 		traceStmtBuilder,
 		logStmtBuilder,
+		auditStmtBuilder,
 		metricStmtBuilder,
 		meterStmtBuilder,
 		traceOperatorStmtBuilder,

pkg/query-service/app/server.go

@@ -208,7 +208,7 @@ func (s *Server) createPublicServer(api *APIHandler, web web.Web) (*http.Server,
 		s.config.APIServer.Timeout.Default,
 		s.config.APIServer.Timeout.Max,
 	).Wrap)
-	r.Use(middleware.NewAudit(s.signoz.Instrumentation.Logger(), s.config.APIServer.Logging.ExcludedRoutes, nil).Wrap)
+	r.Use(middleware.NewAudit(s.signoz.Instrumentation.Logger(), s.config.APIServer.Logging.ExcludedRoutes, s.signoz.Auditor).Wrap)
 	r.Use(middleware.NewComment().Wrap)
 
 	am := middleware.NewAuthZ(s.signoz.Instrumentation.Logger(), s.signoz.Modules.OrgGetter, s.signoz.Authz)

pkg/query-service/rules/setups_test.go

@@ -46,6 +46,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		nil, // prometheus
 		nil, // traceStmtBuilder
 		nil, // logStmtBuilder
+		nil, // auditStmtBuilder
 		metricStmtBuilder,
 		nil, // meterStmtBuilder
 		nil, // traceOperatorStmtBuilder
@@ -91,6 +92,7 @@ func prepareQuerierForLogs(telemetryStore telemetrystore.TelemetryStore, keysMap
 		nil,            // prometheus
 		nil,            // traceStmtBuilder
 		logStmtBuilder, // logStmtBuilder
+		nil,            // auditStmtBuilder
 		nil,            // metricStmtBuilder
 		nil,            // meterStmtBuilder
 		nil,            // traceOperatorStmtBuilder
@@ -131,6 +133,7 @@ func prepareQuerierForTraces(telemetryStore telemetrystore.TelemetryStore, keysM
 		nil,              // prometheus
 		traceStmtBuilder, // traceStmtBuilder
 		nil,              // logStmtBuilder
+		nil,              // auditStmtBuilder
 		nil,              // metricStmtBuilder
 		nil,              // meterStmtBuilder
 		nil,              // traceOperatorStmtBuilder

pkg/querybuilder/collision.go

@@ -207,23 +207,16 @@ func AdjustKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemet
 		indexes := []telemetrytypes.JSONDataTypeIndex{}
 		fieldContextsSeen := map[telemetrytypes.FieldContext]bool{}
 		dataTypesSeen := map[telemetrytypes.FieldDataType]bool{}
-		jsonTypesSeen := map[string]*telemetrytypes.JSONDataType{}
 		for _, matchingKey := range matchingKeys {
 			materialized = materialized && matchingKey.Materialized
 			fieldContextsSeen[matchingKey.FieldContext] = true
 			dataTypesSeen[matchingKey.FieldDataType] = true
-			if matchingKey.JSONDataType != nil {
-				jsonTypesSeen[matchingKey.JSONDataType.StringValue()] = matchingKey.JSONDataType
-			}
 			indexes = append(indexes, matchingKey.Indexes...)
 		}
 		for _, matchingKey := range contextPrefixedMatchingKeys {
 			materialized = materialized && matchingKey.Materialized
 			fieldContextsSeen[matchingKey.FieldContext] = true
 			dataTypesSeen[matchingKey.FieldDataType] = true
-			if matchingKey.JSONDataType != nil {
-				jsonTypesSeen[matchingKey.JSONDataType.StringValue()] = matchingKey.JSONDataType
-			}
 			indexes = append(indexes, matchingKey.Indexes...)
 		}
 		key.Materialized = materialized
@@ -248,15 +241,6 @@ func AdjustKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemet
 				break
 			}
 		}
-
-		if len(jsonTypesSeen) == 1 && key.JSONDataType == nil {
-			// all matching keys have same JSON data type, use it
-			for _, jt := range jsonTypesSeen {
-				actions = append(actions, fmt.Sprintf("Adjusting key %s to have JSON data type %s", key, jt.StringValue()))
-				key.JSONDataType = jt
-				break
-			}
-		}
 	}
 
 	return actions

pkg/signoz/config.go

@@ -11,6 +11,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/alertmanager"
 	"github.com/SigNoz/signoz/pkg/analytics"
 	"github.com/SigNoz/signoz/pkg/apiserver"
+	"github.com/SigNoz/signoz/pkg/auditor"
 	"github.com/SigNoz/signoz/pkg/cache"
 	"github.com/SigNoz/signoz/pkg/config"
 	"github.com/SigNoz/signoz/pkg/emailing"
@@ -123,6 +124,9 @@ type Config struct {
 
 	// ServiceAccount config
 	ServiceAccount serviceaccount.Config `mapstructure:"serviceaccount"`
+
+	// Auditor config
+	Auditor auditor.Config `mapstructure:"auditor"`
 }
 
 func NewConfig(ctx context.Context, logger *slog.Logger, resolverConfig config.ResolverConfig) (Config, error) {
@@ -153,6 +157,7 @@ func NewConfig(ctx context.Context, logger *slog.Logger, resolverConfig config.R
 		user.NewConfigFactory(),
 		identn.NewConfigFactory(),
 		serviceaccount.NewConfigFactory(),
+		auditor.NewConfigFactory(),
 	}
 
 	conf, err := config.New(ctx, resolverConfig, configFactories)

pkg/signoz/handler.go

@@ -34,8 +34,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/services/implservices"
 	"github.com/SigNoz/signoz/pkg/modules/spanpercentile"
 	"github.com/SigNoz/signoz/pkg/modules/spanpercentile/implspanpercentile"
-	"github.com/SigNoz/signoz/pkg/modules/tracedetail"
-	"github.com/SigNoz/signoz/pkg/modules/tracedetail/impltracedetail"
 	"github.com/SigNoz/signoz/pkg/modules/tracefunnel"
 	"github.com/SigNoz/signoz/pkg/modules/tracefunnel/impltracefunnel"
 	"github.com/SigNoz/signoz/pkg/querier"
@@ -64,7 +62,6 @@ type Handlers struct {
 	RegistryHandler         factory.Handler
 	CloudIntegrationHandler cloudintegration.Handler
 	RuleStateHistory        rulestatehistory.Handler
-	TraceDetail             tracedetail.Handler
 }
 
 func NewHandlers(
@@ -102,6 +99,5 @@ func NewHandlers(
 		RegistryHandler:         registryHandler,
 		CloudIntegrationHandler: implcloudintegration.NewHandler(),
 		RuleStateHistory:        implrulestatehistory.NewHandler(modules.RuleStateHistory),
-		TraceDetail:             impltracedetail.NewHandler(modules.TraceDetail),
 	}
 }

pkg/signoz/module.go

@@ -37,8 +37,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/session/implsession"
 	"github.com/SigNoz/signoz/pkg/modules/spanpercentile"
 	"github.com/SigNoz/signoz/pkg/modules/spanpercentile/implspanpercentile"
-	"github.com/SigNoz/signoz/pkg/modules/tracedetail"
-	"github.com/SigNoz/signoz/pkg/modules/tracedetail/impltracedetail"
 	"github.com/SigNoz/signoz/pkg/modules/tracefunnel"
 	"github.com/SigNoz/signoz/pkg/modules/tracefunnel/impltracefunnel"
 	"github.com/SigNoz/signoz/pkg/modules/user"
@@ -74,7 +72,6 @@ type Modules struct {
 	Promote          promote.Module
 	ServiceAccount   serviceaccount.Module
 	RuleStateHistory rulestatehistory.Module
-	TraceDetail      tracedetail.Module
 }
 
 func NewModules(
@@ -122,6 +119,5 @@ func NewModules(
 		Promote:          implpromote.NewModule(telemetryMetadataStore, telemetryStore),
 		ServiceAccount:   implserviceaccount.NewModule(implserviceaccount.NewStore(sqlstore), authz, cache, analytics, providerSettings, config.ServiceAccount),
 		RuleStateHistory: implrulestatehistory.NewModule(implrulestatehistory.NewStore(telemetryStore, telemetryMetadataStore, providerSettings.Logger)),
-		TraceDetail:      impltracedetail.NewModule(telemetryStore, cache, providerSettings),
 	}
 }

pkg/signoz/openapi.go

@@ -28,7 +28,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/rulestatehistory"
 	"github.com/SigNoz/signoz/pkg/modules/serviceaccount"
 	"github.com/SigNoz/signoz/pkg/modules/session"
-	"github.com/SigNoz/signoz/pkg/modules/tracedetail"
 	"github.com/SigNoz/signoz/pkg/modules/user"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/types/authtypes"
@@ -70,7 +69,6 @@ func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumenta
 		struct{ factory.Handler }{},
 		struct{ cloudintegration.Handler }{},
 		struct{ rulestatehistory.Handler }{},
-		struct{ tracedetail.Handler }{},
 	).New(ctx, instrumentation.ToProviderSettings(), apiserver.Config{})
 	if err != nil {
 		return nil, err

pkg/signoz/provider.go

@@ -3,6 +3,8 @@ package signoz
 import (
 	"github.com/SigNoz/signoz/pkg/alertmanager"
 	"github.com/SigNoz/signoz/pkg/alertmanager/nfmanager"
+	"github.com/SigNoz/signoz/pkg/auditor"
+	"github.com/SigNoz/signoz/pkg/auditor/noopauditor"
 	"github.com/SigNoz/signoz/pkg/alertmanager/nfmanager/rulebasednotification"
 	"github.com/SigNoz/signoz/pkg/alertmanager/signozalertmanager"
 	"github.com/SigNoz/signoz/pkg/analytics"
@@ -286,7 +288,6 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
 			handlers.RegistryHandler,
 			handlers.CloudIntegrationHandler,
 			handlers.RuleStateHistory,
-			handlers.TraceDetail,
 		),
 	)
 }
@@ -313,6 +314,12 @@ func NewGlobalProviderFactories(identNConfig identn.Config) factory.NamedMap[fac
 	)
 }
 
+func NewAuditorProviderFactories() factory.NamedMap[factory.ProviderFactory[auditor.Auditor, auditor.Config]] {
+	return factory.MustNewNamedMap(
+		noopauditor.NewFactory(),
+	)
+}
+
 func NewFlaggerProviderFactories(registry featuretypes.Registry) factory.NamedMap[factory.ProviderFactory[flagger.FlaggerProvider, flagger.Config]] {
 	return factory.MustNewNamedMap(
 		configflagger.NewFactory(registry),

pkg/signoz/signoz.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/alertmanager"
 	"github.com/SigNoz/signoz/pkg/alertmanager/nfmanager"
+	"github.com/SigNoz/signoz/pkg/auditor"
 	"github.com/SigNoz/signoz/pkg/alertmanager/nfmanager/nfroutingstore/sqlroutingstore"
 	"github.com/SigNoz/signoz/pkg/analytics"
 	"github.com/SigNoz/signoz/pkg/apiserver"
@@ -33,6 +34,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/sqlschema"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/statsreporter"
+	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymetadata"
 	"github.com/SigNoz/signoz/pkg/telemetrymeter"
@@ -74,6 +76,7 @@ type SigNoz struct {
 	QueryParser            queryparser.QueryParser
 	Flagger                flagger.Flagger
 	Gateway                gateway.Gateway
+	Auditor                auditor.Auditor
 }
 
 func New(
@@ -93,6 +96,7 @@ func New(
 	authzCallback func(context.Context, sqlstore.SQLStore, licensing.Licensing, dashboard.Module) (factory.ProviderFactory[authz.AuthZ, authz.Config], error),
 	dashboardModuleCallback func(sqlstore.SQLStore, factory.ProviderSettings, analytics.Analytics, organization.Getter, queryparser.QueryParser, querier.Querier, licensing.Licensing) dashboard.Module,
 	gatewayProviderFactory func(licensing.Licensing) factory.ProviderFactory[gateway.Gateway, gateway.Config],
+	auditorProviderFactories func(licensing.Licensing) factory.NamedMap[factory.ProviderFactory[auditor.Auditor, auditor.Config]],
 	querierHandlerCallback func(factory.ProviderSettings, querier.Querier, analytics.Analytics) querier.Handler,
 ) (*SigNoz, error) {
 	// Initialize instrumentation
@@ -370,6 +374,12 @@ func New(
 		return nil, err
 	}
 
+	// Initialize auditor from the variant-specific provider factories
+	auditor, err := factory.NewProviderFromNamedMap(ctx, providerSettings, config.Auditor, auditorProviderFactories(licensing), config.Auditor.Provider)
+	if err != nil {
+		return nil, err
+	}
+
 	// Initialize authns
 	store := sqlauthnstore.NewStore(sqlstore)
 	authNs, err := authNsCallback(ctx, providerSettings, store, licensing)
@@ -395,6 +405,11 @@ func New(
 		telemetrylogs.TagAttributesV2TableName,
 		telemetrylogs.LogAttributeKeysTblName,
 		telemetrylogs.LogResourceKeysTblName,
+		telemetryaudit.DBName,
+		telemetryaudit.AuditLogsTableName,
+		telemetryaudit.TagAttributesTableName,
+		telemetryaudit.LogAttributeKeysTblName,
+		telemetryaudit.LogResourceKeysTblName,
 		telemetrymetadata.DBName,
 		telemetrymetadata.AttributesMetadataLocalTableName,
 		telemetrymetadata.ColumnEvolutionMetadataTableName,
@@ -464,6 +479,7 @@ func New(
 		factory.NewNamedService(factory.MustNewName("tokenizer"), tokenizer),
 		factory.NewNamedService(factory.MustNewName("authz"), authz),
 		factory.NewNamedService(factory.MustNewName("user"), userService, factory.MustNewName("authz")),
+		factory.NewNamedService(factory.MustNewName("auditor"), auditor),
 	)
 	if err != nil {
 		return nil, err
@@ -510,5 +526,6 @@ func New(
 		QueryParser:            queryParser,
 		Flagger:                flagger,
 		Gateway:                gateway,
+		Auditor:                auditor,
 	}, nil
 }

pkg/telemetryaudit/condition_builder.go

@@ -0,0 +1,200 @@
+package telemetryaudit
+
+import (
+	"context"
+	"fmt"
+
+	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+type conditionBuilder struct {
+	fm qbtypes.FieldMapper
+}
+
+func NewConditionBuilder(fm qbtypes.FieldMapper) *conditionBuilder {
+	return &conditionBuilder{fm: fm}
+}
+
+func (c *conditionBuilder) conditionFor(
+	ctx context.Context,
+	startNs, endNs uint64,
+	key *telemetrytypes.TelemetryFieldKey,
+	operator qbtypes.FilterOperator,
+	value any,
+	sb *sqlbuilder.SelectBuilder,
+) (string, error) {
+	columns, err := c.fm.ColumnFor(ctx, startNs, endNs, key)
+	if err != nil {
+		return "", err
+	}
+
+	if operator.IsStringSearchOperator() {
+		value = querybuilder.FormatValueForContains(value)
+	}
+
+	fieldExpression, err := c.fm.FieldFor(ctx, startNs, endNs, key)
+	if err != nil {
+		return "", err
+	}
+
+	fieldExpression, value = querybuilder.DataTypeCollisionHandledFieldName(key, value, fieldExpression, operator)
+
+	switch operator {
+	case qbtypes.FilterOperatorEqual:
+		return sb.E(fieldExpression, value), nil
+	case qbtypes.FilterOperatorNotEqual:
+		return sb.NE(fieldExpression, value), nil
+	case qbtypes.FilterOperatorGreaterThan:
+		return sb.G(fieldExpression, value), nil
+	case qbtypes.FilterOperatorGreaterThanOrEq:
+		return sb.GE(fieldExpression, value), nil
+	case qbtypes.FilterOperatorLessThan:
+		return sb.LT(fieldExpression, value), nil
+	case qbtypes.FilterOperatorLessThanOrEq:
+		return sb.LE(fieldExpression, value), nil
+	case qbtypes.FilterOperatorLike:
+		return sb.Like(fieldExpression, value), nil
+	case qbtypes.FilterOperatorNotLike:
+		return sb.NotLike(fieldExpression, value), nil
+	case qbtypes.FilterOperatorILike:
+		return sb.ILike(fieldExpression, value), nil
+	case qbtypes.FilterOperatorNotILike:
+		return sb.NotILike(fieldExpression, value), nil
+	case qbtypes.FilterOperatorContains:
+		return sb.ILike(fieldExpression, fmt.Sprintf("%%%s%%", value)), nil
+	case qbtypes.FilterOperatorNotContains:
+		return sb.NotILike(fieldExpression, fmt.Sprintf("%%%s%%", value)), nil
+	case qbtypes.FilterOperatorRegexp:
+		return fmt.Sprintf(`match(%s, %s)`, sqlbuilder.Escape(fieldExpression), sb.Var(value)), nil
+	case qbtypes.FilterOperatorNotRegexp:
+		return fmt.Sprintf(`NOT match(%s, %s)`, sqlbuilder.Escape(fieldExpression), sb.Var(value)), nil
+	case qbtypes.FilterOperatorBetween:
+		values, ok := value.([]any)
+		if !ok {
+			return "", qbtypes.ErrBetweenValues
+		}
+		if len(values) != 2 {
+			return "", qbtypes.ErrBetweenValues
+		}
+		return sb.Between(fieldExpression, values[0], values[1]), nil
+	case qbtypes.FilterOperatorNotBetween:
+		values, ok := value.([]any)
+		if !ok {
+			return "", qbtypes.ErrBetweenValues
+		}
+		if len(values) != 2 {
+			return "", qbtypes.ErrBetweenValues
+		}
+		return sb.NotBetween(fieldExpression, values[0], values[1]), nil
+	case qbtypes.FilterOperatorIn:
+		values, ok := value.([]any)
+		if !ok {
+			return "", qbtypes.ErrInValues
+		}
+		conditions := []string{}
+		for _, value := range values {
+			conditions = append(conditions, sb.E(fieldExpression, value))
+		}
+		return sb.Or(conditions...), nil
+	case qbtypes.FilterOperatorNotIn:
+		values, ok := value.([]any)
+		if !ok {
+			return "", qbtypes.ErrInValues
+		}
+		conditions := []string{}
+		for _, value := range values {
+			conditions = append(conditions, sb.NE(fieldExpression, value))
+		}
+		return sb.And(conditions...), nil
+	case qbtypes.FilterOperatorExists, qbtypes.FilterOperatorNotExists:
+		var value any
+		column := columns[0]
+
+		switch column.Type.GetType() {
+		case schema.ColumnTypeEnumJSON:
+			if operator == qbtypes.FilterOperatorExists {
+				return sb.IsNotNull(fieldExpression), nil
+			}
+			return sb.IsNull(fieldExpression), nil
+		case schema.ColumnTypeEnumLowCardinality:
+			switch elementType := column.Type.(schema.LowCardinalityColumnType).ElementType; elementType.GetType() {
+			case schema.ColumnTypeEnumString:
+				value = ""
+				if operator == qbtypes.FilterOperatorExists {
+					return sb.NE(fieldExpression, value), nil
+				}
+				return sb.E(fieldExpression, value), nil
+			default:
+				return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "exists operator is not supported for low cardinality column type %s", elementType)
+			}
+		case schema.ColumnTypeEnumString:
+			value = ""
+			if operator == qbtypes.FilterOperatorExists {
+				return sb.NE(fieldExpression, value), nil
+			}
+			return sb.E(fieldExpression, value), nil
+		case schema.ColumnTypeEnumUInt64, schema.ColumnTypeEnumUInt32, schema.ColumnTypeEnumUInt8:
+			value = 0
+			if operator == qbtypes.FilterOperatorExists {
+				return sb.NE(fieldExpression, value), nil
+			}
+			return sb.E(fieldExpression, value), nil
+		case schema.ColumnTypeEnumMap:
+			keyType := column.Type.(schema.MapColumnType).KeyType
+			if _, ok := keyType.(schema.LowCardinalityColumnType); !ok {
+				return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "key type %s is not supported for map column type %s", keyType, column.Type)
+			}
+
+			switch valueType := column.Type.(schema.MapColumnType).ValueType; valueType.GetType() {
+			case schema.ColumnTypeEnumString, schema.ColumnTypeEnumBool, schema.ColumnTypeEnumFloat64:
+				leftOperand := fmt.Sprintf("mapContains(%s, '%s')", column.Name, key.Name)
+				if key.Materialized {
+					leftOperand = telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key)
+				}
+				if operator == qbtypes.FilterOperatorExists {
+					return sb.E(leftOperand, true), nil
+				}
+				return sb.NE(leftOperand, true), nil
+			default:
+				return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "exists operator is not supported for map column type %s", valueType)
+			}
+		default:
+			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "exists operator is not supported for column type %s", column.Type)
+		}
+	}
+	return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported operator: %v", operator)
+}
+
+func (c *conditionBuilder) ConditionFor(
+	ctx context.Context,
+	startNs uint64,
+	endNs uint64,
+	key *telemetrytypes.TelemetryFieldKey,
+	operator qbtypes.FilterOperator,
+	value any,
+	sb *sqlbuilder.SelectBuilder,
+) (string, error) {
+	condition, err := c.conditionFor(ctx, startNs, endNs, key, operator, value, sb)
+	if err != nil {
+		return "", err
+	}
+
+	if key.FieldContext == telemetrytypes.FieldContextLog || key.FieldContext == telemetrytypes.FieldContextScope {
+		return condition, nil
+	}
+
+	if operator.AddDefaultExistsFilter() {
+		existsCondition, err := c.conditionFor(ctx, startNs, endNs, key, qbtypes.FilterOperatorExists, nil, sb)
+		if err != nil {
+			return "", err
+		}
+		return sb.And(condition, existsCondition), nil
+	}
+
+	return condition, nil
+}

pkg/telemetryaudit/const.go

@@ -0,0 +1,129 @@
+package telemetryaudit
+
+import (
+	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+const (
+	// Internal Columns.
+	IDColumn                   = "id"
+	TimestampBucketStartColumn = "ts_bucket_start"
+	ResourceFingerPrintColumn  = "resource_fingerprint"
+
+	// Intrinsic Columns.
+	TimestampColumn         = "timestamp"
+	ObservedTimestampColumn = "observed_timestamp"
+	BodyColumn              = "body"
+	EventNameColumn         = "event_name"
+	TraceIDColumn           = "trace_id"
+	SpanIDColumn            = "span_id"
+	TraceFlagsColumn        = "trace_flags"
+	SeverityTextColumn      = "severity_text"
+	SeverityNumberColumn    = "severity_number"
+	ScopeNameColumn         = "scope_name"
+	ScopeVersionColumn      = "scope_version"
+
+	// Contextual Columns.
+	AttributesStringColumn = "attributes_string"
+	AttributesNumberColumn = "attributes_number"
+	AttributesBoolColumn   = "attributes_bool"
+	ResourceColumn         = "resource"
+	ScopeStringColumn      = "scope_string"
+)
+
+var (
+	DefaultFullTextColumn = &telemetrytypes.TelemetryFieldKey{
+		Name:          "body",
+		Signal:        telemetrytypes.SignalLogs,
+		FieldContext:  telemetrytypes.FieldContextLog,
+		FieldDataType: telemetrytypes.FieldDataTypeString,
+	}
+
+	IntrinsicFields = map[string]telemetrytypes.TelemetryFieldKey{
+		"body": {
+			Name:          "body",
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  telemetrytypes.FieldContextLog,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+		},
+		"trace_id": {
+			Name:          "trace_id",
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  telemetrytypes.FieldContextLog,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+		},
+		"span_id": {
+			Name:          "span_id",
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  telemetrytypes.FieldContextLog,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+		},
+		"trace_flags": {
+			Name:          "trace_flags",
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  telemetrytypes.FieldContextLog,
+			FieldDataType: telemetrytypes.FieldDataTypeNumber,
+		},
+		"severity_text": {
+			Name:          "severity_text",
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  telemetrytypes.FieldContextLog,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+		},
+		"severity_number": {
+			Name:          "severity_number",
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  telemetrytypes.FieldContextLog,
+			FieldDataType: telemetrytypes.FieldDataTypeNumber,
+		},
+		"event_name": {
+			Name:          "event_name",
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  telemetrytypes.FieldContextLog,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+		},
+	}
+
+	DefaultSortingOrder = []qbtypes.OrderBy{
+		{
+			Key: qbtypes.OrderByKey{
+				TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+					Name: TimestampColumn,
+				},
+			},
+			Direction: qbtypes.OrderDirectionDesc,
+		},
+		{
+			Key: qbtypes.OrderByKey{
+				TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+					Name: IDColumn,
+				},
+			},
+			Direction: qbtypes.OrderDirectionDesc,
+		},
+	}
+)
+
+var auditLogColumns = map[string]*schema.Column{
+	"ts_bucket_start":      {Name: "ts_bucket_start", Type: schema.ColumnTypeUInt64},
+	"resource_fingerprint": {Name: "resource_fingerprint", Type: schema.ColumnTypeString},
+	"timestamp":            {Name: "timestamp", Type: schema.ColumnTypeUInt64},
+	"observed_timestamp":   {Name: "observed_timestamp", Type: schema.ColumnTypeUInt64},
+	"id":                   {Name: "id", Type: schema.ColumnTypeString},
+	"trace_id":             {Name: "trace_id", Type: schema.ColumnTypeString},
+	"span_id":              {Name: "span_id", Type: schema.ColumnTypeString},
+	"trace_flags":          {Name: "trace_flags", Type: schema.ColumnTypeUInt32},
+	"severity_text":        {Name: "severity_text", Type: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}},
+	"severity_number":      {Name: "severity_number", Type: schema.ColumnTypeUInt8},
+	"body":                 {Name: "body", Type: schema.ColumnTypeString},
+	"attributes_string":    {Name: "attributes_string", Type: schema.MapColumnType{KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}, ValueType: schema.ColumnTypeString}},
+	"attributes_number":    {Name: "attributes_number", Type: schema.MapColumnType{KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}, ValueType: schema.ColumnTypeFloat64}},
+	"attributes_bool":      {Name: "attributes_bool", Type: schema.MapColumnType{KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}, ValueType: schema.ColumnTypeBool}},
+	"resource":             {Name: "resource", Type: schema.JSONColumnType{}},
+	"event_name":           {Name: "event_name", Type: schema.ColumnTypeString},
+	"scope_name":           {Name: "scope_name", Type: schema.ColumnTypeString},
+	"scope_version":        {Name: "scope_version", Type: schema.ColumnTypeString},
+	"scope_string":         {Name: "scope_string", Type: schema.MapColumnType{KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}, ValueType: schema.ColumnTypeString}},
+}

pkg/telemetryaudit/field_mapper.go

@@ -0,0 +1,124 @@
+package telemetryaudit
+
+import (
+	"context"
+	"fmt"
+
+	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
+	"github.com/SigNoz/signoz/pkg/errors"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/huandu/go-sqlbuilder"
+
+	"golang.org/x/exp/maps"
+)
+
+type fieldMapper struct{}
+
+func NewFieldMapper() qbtypes.FieldMapper {
+	return &fieldMapper{}
+}
+
+func (m *fieldMapper) getColumn(_ context.Context, key *telemetrytypes.TelemetryFieldKey) ([]*schema.Column, error) {
+	switch key.FieldContext {
+	case telemetrytypes.FieldContextResource:
+		return []*schema.Column{auditLogColumns["resource"]}, nil
+	case telemetrytypes.FieldContextScope:
+		switch key.Name {
+		case "name", "scope.name", "scope_name":
+			return []*schema.Column{auditLogColumns["scope_name"]}, nil
+		case "version", "scope.version", "scope_version":
+			return []*schema.Column{auditLogColumns["scope_version"]}, nil
+		}
+		return []*schema.Column{auditLogColumns["scope_string"]}, nil
+	case telemetrytypes.FieldContextAttribute:
+		switch key.FieldDataType {
+		case telemetrytypes.FieldDataTypeString:
+			return []*schema.Column{auditLogColumns["attributes_string"]}, nil
+		case telemetrytypes.FieldDataTypeInt64, telemetrytypes.FieldDataTypeFloat64, telemetrytypes.FieldDataTypeNumber:
+			return []*schema.Column{auditLogColumns["attributes_number"]}, nil
+		case telemetrytypes.FieldDataTypeBool:
+			return []*schema.Column{auditLogColumns["attributes_bool"]}, nil
+		}
+	case telemetrytypes.FieldContextLog, telemetrytypes.FieldContextUnspecified:
+		col, ok := auditLogColumns[key.Name]
+		if !ok {
+			return nil, qbtypes.ErrColumnNotFound
+		}
+		return []*schema.Column{col}, nil
+	}
+
+	return nil, qbtypes.ErrColumnNotFound
+}
+
+func (m *fieldMapper) FieldFor(ctx context.Context, _, _ uint64, key *telemetrytypes.TelemetryFieldKey) (string, error) {
+	columns, err := m.getColumn(ctx, key)
+	if err != nil {
+		return "", err
+	}
+	if len(columns) != 1 {
+		return "", errors.Newf(errors.TypeInternal, errors.CodeInternal, "expected exactly 1 column, got %d", len(columns))
+	}
+	column := columns[0]
+
+	switch column.Type.GetType() {
+	case schema.ColumnTypeEnumJSON:
+		if key.FieldContext != telemetrytypes.FieldContextResource {
+			return "", errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "only resource context fields are supported for json columns in audit, got %s", key.FieldContext.String)
+		}
+		return fmt.Sprintf("%s.`%s`::String", column.Name, key.Name), nil
+	case schema.ColumnTypeEnumLowCardinality:
+		return column.Name, nil
+	case schema.ColumnTypeEnumString, schema.ColumnTypeEnumUInt64, schema.ColumnTypeEnumUInt32, schema.ColumnTypeEnumUInt8:
+		return column.Name, nil
+	case schema.ColumnTypeEnumMap:
+		keyType := column.Type.(schema.MapColumnType).KeyType
+		if _, ok := keyType.(schema.LowCardinalityColumnType); !ok {
+			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "key type %s is not supported for map column type %s", keyType, column.Type)
+		}
+
+		switch valueType := column.Type.(schema.MapColumnType).ValueType; valueType.GetType() {
+		case schema.ColumnTypeEnumString, schema.ColumnTypeEnumBool, schema.ColumnTypeEnumFloat64:
+			if key.Materialized {
+				return telemetrytypes.FieldKeyToMaterializedColumnName(key), nil
+			}
+			return fmt.Sprintf("%s['%s']", column.Name, key.Name), nil
+		default:
+			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported map value type %s", valueType)
+		}
+	}
+
+	return column.Name, nil
+}
+
+func (m *fieldMapper) ColumnFor(ctx context.Context, _, _ uint64, key *telemetrytypes.TelemetryFieldKey) ([]*schema.Column, error) {
+	return m.getColumn(ctx, key)
+}
+
+func (m *fieldMapper) ColumnExpressionFor(
+	ctx context.Context,
+	tsStart, tsEnd uint64,
+	field *telemetrytypes.TelemetryFieldKey,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+) (string, error) {
+	fieldExpression, err := m.FieldFor(ctx, tsStart, tsEnd, field)
+	if errors.Is(err, qbtypes.ErrColumnNotFound) {
+		keysForField := keys[field.Name]
+		if len(keysForField) == 0 {
+			if _, ok := auditLogColumns[field.Name]; ok {
+				field.FieldContext = telemetrytypes.FieldContextLog
+				fieldExpression, _ = m.FieldFor(ctx, tsStart, tsEnd, field)
+			} else {
+				correction, found := telemetrytypes.SuggestCorrection(field.Name, maps.Keys(keys))
+				if found {
+					return "", errors.Wrap(err, errors.TypeInvalidInput, errors.CodeInvalidInput, correction)
+				}
+				return "", errors.Wrapf(err, errors.TypeInvalidInput, errors.CodeInvalidInput, "field `%s` not found", field.Name)
+			}
+		} else {
+			fieldExpression, _ = m.FieldFor(ctx, tsStart, tsEnd, keysForField[0])
+		}
+	}
+
+	return fmt.Sprintf("%s AS `%s`", sqlbuilder.Escape(fieldExpression), field.Name), nil
+}

pkg/telemetryaudit/statement_builder.go

@@ -0,0 +1,612 @@
+package telemetryaudit
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetryresourcefilter"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+type auditQueryStatementBuilder struct {
+	logger                    *slog.Logger
+	metadataStore             telemetrytypes.MetadataStore
+	fm                        qbtypes.FieldMapper
+	cb                        qbtypes.ConditionBuilder
+	resourceFilterStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
+	aggExprRewriter           qbtypes.AggExprRewriter
+	fullTextColumn            *telemetrytypes.TelemetryFieldKey
+	jsonKeyToKey              qbtypes.JsonKeyToFieldFunc
+}
+
+var _ qbtypes.StatementBuilder[qbtypes.LogAggregation] = (*auditQueryStatementBuilder)(nil)
+
+func NewAuditQueryStatementBuilder(
+	settings factory.ProviderSettings,
+	metadataStore telemetrytypes.MetadataStore,
+	fieldMapper qbtypes.FieldMapper,
+	conditionBuilder qbtypes.ConditionBuilder,
+	aggExprRewriter qbtypes.AggExprRewriter,
+	fullTextColumn *telemetrytypes.TelemetryFieldKey,
+	jsonKeyToKey qbtypes.JsonKeyToFieldFunc,
+) *auditQueryStatementBuilder {
+	auditSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryaudit")
+
+	resourceFilterStmtBuilder := telemetryresourcefilter.New[qbtypes.LogAggregation](
+		settings,
+		DBName,
+		LogsResourceTableName,
+		telemetrytypes.SignalLogs,
+		telemetrytypes.SourceAudit,
+		metadataStore,
+		fullTextColumn,
+		jsonKeyToKey,
+	)
+
+	return &auditQueryStatementBuilder{
+		logger:                    auditSettings.Logger(),
+		metadataStore:             metadataStore,
+		fm:                        fieldMapper,
+		cb:                        conditionBuilder,
+		resourceFilterStmtBuilder: resourceFilterStmtBuilder,
+		aggExprRewriter:           aggExprRewriter,
+		fullTextColumn:            fullTextColumn,
+		jsonKeyToKey:              jsonKeyToKey,
+	}
+}
+
+func (b *auditQueryStatementBuilder) Build(
+	ctx context.Context,
+	start uint64,
+	end uint64,
+	requestType qbtypes.RequestType,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	start = querybuilder.ToNanoSecs(start)
+	end = querybuilder.ToNanoSecs(end)
+
+	keySelectors := getKeySelectors(query)
+	keys, _, err := b.metadataStore.GetKeysMulti(ctx, keySelectors)
+	if err != nil {
+		return nil, err
+	}
+
+	query = b.adjustKeys(ctx, keys, query, requestType)
+
+	q := sqlbuilder.NewSelectBuilder()
+
+	var stmt *qbtypes.Statement
+	switch requestType {
+	case qbtypes.RequestTypeRaw, qbtypes.RequestTypeRawStream:
+		stmt, err = b.buildListQuery(ctx, q, query, start, end, keys, variables)
+	case qbtypes.RequestTypeTimeSeries:
+		stmt, err = b.buildTimeSeriesQuery(ctx, q, query, start, end, keys, variables)
+	case qbtypes.RequestTypeScalar:
+		stmt, err = b.buildScalarQuery(ctx, q, query, start, end, keys, false, variables)
+	default:
+		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported request type: %s", requestType)
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return stmt, nil
+}
+
+func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]) []*telemetrytypes.FieldKeySelector {
+	var keySelectors []*telemetrytypes.FieldKeySelector
+
+	for idx := range query.Aggregations {
+		aggExpr := query.Aggregations[idx]
+		selectors := querybuilder.QueryStringToKeysSelectors(aggExpr.Expression)
+		keySelectors = append(keySelectors, selectors...)
+	}
+
+	if query.Filter != nil && query.Filter.Expression != "" {
+		whereClauseSelectors := querybuilder.QueryStringToKeysSelectors(query.Filter.Expression)
+		keySelectors = append(keySelectors, whereClauseSelectors...)
+	}
+
+	for idx := range query.GroupBy {
+		groupBy := query.GroupBy[idx]
+		keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
+			Name:          groupBy.Name,
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  groupBy.FieldContext,
+			FieldDataType: groupBy.FieldDataType,
+		})
+	}
+
+	for idx := range query.SelectFields {
+		selectField := query.SelectFields[idx]
+		keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
+			Name:          selectField.Name,
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  selectField.FieldContext,
+			FieldDataType: selectField.FieldDataType,
+		})
+	}
+
+	for idx := range query.Order {
+		keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
+			Name:          query.Order[idx].Key.Name,
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  query.Order[idx].Key.FieldContext,
+			FieldDataType: query.Order[idx].Key.FieldDataType,
+		})
+	}
+
+	for idx := range keySelectors {
+		keySelectors[idx].Signal = telemetrytypes.SignalLogs
+		keySelectors[idx].Source = telemetrytypes.SourceAudit
+		keySelectors[idx].SelectorMatchType = telemetrytypes.FieldSelectorMatchTypeExact
+	}
+
+	return keySelectors
+}
+
+func (b *auditQueryStatementBuilder) adjustKeys(ctx context.Context, keys map[string][]*telemetrytypes.TelemetryFieldKey, query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], requestType qbtypes.RequestType) qbtypes.QueryBuilderQuery[qbtypes.LogAggregation] {
+	keys["id"] = append([]*telemetrytypes.TelemetryFieldKey{{
+		Name:          "id",
+		Signal:        telemetrytypes.SignalLogs,
+		FieldContext:  telemetrytypes.FieldContextLog,
+		FieldDataType: telemetrytypes.FieldDataTypeString,
+	}}, keys["id"]...)
+
+	keys["timestamp"] = append([]*telemetrytypes.TelemetryFieldKey{{
+		Name:          "timestamp",
+		Signal:        telemetrytypes.SignalLogs,
+		FieldContext:  telemetrytypes.FieldContextLog,
+		FieldDataType: telemetrytypes.FieldDataTypeNumber,
+	}}, keys["timestamp"]...)
+
+	actions := querybuilder.AdjustKeysForAliasExpressions(&query, requestType)
+	actions = append(actions, querybuilder.AdjustDuplicateKeys(&query)...)
+
+	for idx := range query.SelectFields {
+		actions = append(actions, b.adjustKey(&query.SelectFields[idx], keys)...)
+	}
+	for idx := range query.GroupBy {
+		actions = append(actions, b.adjustKey(&query.GroupBy[idx].TelemetryFieldKey, keys)...)
+	}
+	for idx := range query.Order {
+		actions = append(actions, b.adjustKey(&query.Order[idx].Key.TelemetryFieldKey, keys)...)
+	}
+
+	for _, action := range actions {
+		b.logger.InfoContext(ctx, "key adjustment action", slog.String("action", action))
+	}
+
+	return query
+}
+
+func (b *auditQueryStatementBuilder) adjustKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) []string {
+	if _, ok := IntrinsicFields[key.Name]; ok {
+		intrinsicField := IntrinsicFields[key.Name]
+		return querybuilder.AdjustKey(key, keys, &intrinsicField)
+	}
+	return querybuilder.AdjustKey(key, keys, nil)
+}
+
+func (b *auditQueryStatementBuilder) buildListQuery(
+	ctx context.Context,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	var (
+		cteFragments []string
+		cteArgs      [][]any
+	)
+
+	if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
+		return nil, err
+	} else if frag != "" {
+		cteFragments = append(cteFragments, frag)
+		cteArgs = append(cteArgs, args)
+	}
+
+	sb.Select(TimestampColumn)
+	sb.SelectMore(IDColumn)
+	if len(query.SelectFields) == 0 {
+		sb.SelectMore(TraceIDColumn)
+		sb.SelectMore(SpanIDColumn)
+		sb.SelectMore(TraceFlagsColumn)
+		sb.SelectMore(SeverityTextColumn)
+		sb.SelectMore(SeverityNumberColumn)
+		sb.SelectMore(ScopeNameColumn)
+		sb.SelectMore(ScopeVersionColumn)
+		sb.SelectMore(BodyColumn)
+		sb.SelectMore(EventNameColumn)
+		sb.SelectMore(AttributesStringColumn)
+		sb.SelectMore(AttributesNumberColumn)
+		sb.SelectMore(AttributesBoolColumn)
+		sb.SelectMore(ResourceColumn)
+		sb.SelectMore(ScopeStringColumn)
+	} else {
+		for index := range query.SelectFields {
+			if query.SelectFields[index].Name == TimestampColumn || query.SelectFields[index].Name == IDColumn {
+				continue
+			}
+
+			colExpr, err := b.fm.ColumnExpressionFor(ctx, start, end, &query.SelectFields[index], keys)
+			if err != nil {
+				return nil, err
+			}
+			sb.SelectMore(colExpr)
+		}
+	}
+
+	sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
+
+	preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, orderBy := range query.Order {
+		colExpr, err := b.fm.ColumnExpressionFor(ctx, start, end, &orderBy.Key.TelemetryFieldKey, keys)
+		if err != nil {
+			return nil, err
+		}
+		sb.OrderBy(fmt.Sprintf("%s %s", colExpr, orderBy.Direction.StringValue()))
+	}
+
+	if query.Limit > 0 {
+		sb.Limit(query.Limit)
+	} else {
+		sb.Limit(100)
+	}
+
+	if query.Offset > 0 {
+		sb.Offset(query.Offset)
+	}
+
+	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL
+	finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
+
+	stmt := &qbtypes.Statement{
+		Query: finalSQL,
+		Args:  finalArgs,
+	}
+	if preparedWhereClause != nil {
+		stmt.Warnings = preparedWhereClause.Warnings
+		stmt.WarningsDocURL = preparedWhereClause.WarningsDocURL
+	}
+
+	return stmt, nil
+}
+
+func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
+	ctx context.Context,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	var (
+		cteFragments []string
+		cteArgs      [][]any
+	)
+
+	if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
+		return nil, err
+	} else if frag != "" {
+		cteFragments = append(cteFragments, frag)
+		cteArgs = append(cteArgs, args)
+	}
+
+	sb.SelectMore(fmt.Sprintf(
+		"toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL %d SECOND) AS ts",
+		int64(query.StepInterval.Seconds()),
+	))
+
+	var allGroupByArgs []any
+
+	fieldNames := make([]string, 0, len(query.GroupBy))
+	for _, gb := range query.GroupBy {
+		expr, args, err := querybuilder.CollisionHandledFinalExpr(ctx, start, end, &gb.TelemetryFieldKey, b.fm, b.cb, keys, telemetrytypes.FieldDataTypeString, b.jsonKeyToKey)
+		if err != nil {
+			return nil, err
+		}
+
+		colExpr := fmt.Sprintf("toString(%s) AS `%s`", expr, gb.Name)
+		allGroupByArgs = append(allGroupByArgs, args...)
+		sb.SelectMore(colExpr)
+		fieldNames = append(fieldNames, fmt.Sprintf("`%s`", gb.Name))
+	}
+
+	allAggChArgs := make([]any, 0)
+	for i, agg := range query.Aggregations {
+		rewritten, chArgs, err := b.aggExprRewriter.Rewrite(ctx, start, end, agg.Expression, uint64(query.StepInterval.Seconds()), keys)
+		if err != nil {
+			return nil, err
+		}
+		allAggChArgs = append(allAggChArgs, chArgs...)
+		sb.SelectMore(fmt.Sprintf("%s AS __result_%d", rewritten, i))
+	}
+
+	sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
+
+	preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
+	if err != nil {
+		return nil, err
+	}
+
+	var finalSQL string
+	var finalArgs []any
+
+	if query.Limit > 0 && len(query.GroupBy) > 0 {
+		cteSB := sqlbuilder.NewSelectBuilder()
+		cteStmt, err := b.buildScalarQuery(ctx, cteSB, query, start, end, keys, true, variables)
+		if err != nil {
+			return nil, err
+		}
+
+		cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", cteStmt.Query))
+		cteArgs = append(cteArgs, cteStmt.Args)
+
+		tuple := fmt.Sprintf("(%s)", strings.Join(fieldNames, ", "))
+		sb.Where(fmt.Sprintf("%s GLOBAL IN (SELECT %s FROM __limit_cte)", tuple, strings.Join(fieldNames, ", ")))
+
+		sb.GroupBy("ts")
+		sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+		if query.Having != nil && query.Having.Expression != "" {
+			rewriter := querybuilder.NewHavingExpressionRewriter()
+			rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)
+			if err != nil {
+				return nil, err
+			}
+			sb.Having(rewrittenExpr)
+		}
+
+		if len(query.Order) != 0 {
+			for _, orderBy := range query.Order {
+				_, ok := aggOrderBy(orderBy, query)
+				if !ok {
+					sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
+				}
+			}
+			sb.OrderBy("ts desc")
+		}
+
+		combinedArgs := append(allGroupByArgs, allAggChArgs...)
+		mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
+
+		finalSQL = querybuilder.CombineCTEs(cteFragments) + mainSQL
+		finalArgs = querybuilder.PrependArgs(cteArgs, mainArgs)
+	} else {
+		sb.GroupBy("ts")
+		sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+		if query.Having != nil && query.Having.Expression != "" {
+			rewriter := querybuilder.NewHavingExpressionRewriter()
+			rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)
+			if err != nil {
+				return nil, err
+			}
+			sb.Having(rewrittenExpr)
+		}
+
+		if len(query.Order) != 0 {
+			for _, orderBy := range query.Order {
+				_, ok := aggOrderBy(orderBy, query)
+				if !ok {
+					sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
+				}
+			}
+			sb.OrderBy("ts desc")
+		}
+
+		combinedArgs := append(allGroupByArgs, allAggChArgs...)
+		mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
+
+		finalSQL = querybuilder.CombineCTEs(cteFragments) + mainSQL
+		finalArgs = querybuilder.PrependArgs(cteArgs, mainArgs)
+	}
+
+	stmt := &qbtypes.Statement{
+		Query: finalSQL,
+		Args:  finalArgs,
+	}
+	if preparedWhereClause != nil {
+		stmt.Warnings = preparedWhereClause.Warnings
+		stmt.WarningsDocURL = preparedWhereClause.WarningsDocURL
+	}
+
+	return stmt, nil
+}
+
+func (b *auditQueryStatementBuilder) buildScalarQuery(
+	ctx context.Context,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	start, end uint64,
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	skipResourceCTE bool,
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	var (
+		cteFragments []string
+		cteArgs      [][]any
+	)
+
+	if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
+		return nil, err
+	} else if frag != "" && !skipResourceCTE {
+		cteFragments = append(cteFragments, frag)
+		cteArgs = append(cteArgs, args)
+	}
+
+	allAggChArgs := []any{}
+
+	var allGroupByArgs []any
+
+	for _, gb := range query.GroupBy {
+		expr, args, err := querybuilder.CollisionHandledFinalExpr(ctx, start, end, &gb.TelemetryFieldKey, b.fm, b.cb, keys, telemetrytypes.FieldDataTypeString, b.jsonKeyToKey)
+		if err != nil {
+			return nil, err
+		}
+
+		colExpr := fmt.Sprintf("toString(%s) AS `%s`", expr, gb.Name)
+		allGroupByArgs = append(allGroupByArgs, args...)
+		sb.SelectMore(colExpr)
+	}
+
+	rateInterval := (end - start) / querybuilder.NsToSeconds
+
+	if len(query.Aggregations) > 0 {
+		for idx := range query.Aggregations {
+			aggExpr := query.Aggregations[idx]
+			rewritten, chArgs, err := b.aggExprRewriter.Rewrite(ctx, start, end, aggExpr.Expression, rateInterval, keys)
+			if err != nil {
+				return nil, err
+			}
+			allAggChArgs = append(allAggChArgs, chArgs...)
+			sb.SelectMore(fmt.Sprintf("%s AS __result_%d", rewritten, idx))
+		}
+	}
+
+	sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
+
+	preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
+	if err != nil {
+		return nil, err
+	}
+
+	sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+
+	if query.Having != nil && query.Having.Expression != "" {
+		rewriter := querybuilder.NewHavingExpressionRewriter()
+		rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)
+		if err != nil {
+			return nil, err
+		}
+		sb.Having(rewrittenExpr)
+	}
+
+	for _, orderBy := range query.Order {
+		idx, ok := aggOrderBy(orderBy, query)
+		if ok {
+			sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
+		} else {
+			sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
+		}
+	}
+
+	if len(query.Order) == 0 {
+		sb.OrderBy("__result_0 DESC")
+	}
+
+	if query.Limit > 0 {
+		sb.Limit(query.Limit)
+	}
+
+	combinedArgs := append(allGroupByArgs, allAggChArgs...)
+
+	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
+
+	finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL
+	finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
+
+	stmt := &qbtypes.Statement{
+		Query: finalSQL,
+		Args:  finalArgs,
+	}
+	if preparedWhereClause != nil {
+		stmt.Warnings = preparedWhereClause.Warnings
+		stmt.WarningsDocURL = preparedWhereClause.WarningsDocURL
+	}
+
+	return stmt, nil
+}
+
+func (b *auditQueryStatementBuilder) addFilterCondition(
+	ctx context.Context,
+	sb *sqlbuilder.SelectBuilder,
+	start, end uint64,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+) (*querybuilder.PreparedWhereClause, error) {
+	var preparedWhereClause *querybuilder.PreparedWhereClause
+	var err error
+
+	if query.Filter != nil && query.Filter.Expression != "" {
+		preparedWhereClause, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
+			Context:            ctx,
+			Logger:             b.logger,
+			FieldMapper:        b.fm,
+			ConditionBuilder:   b.cb,
+			FieldKeys:          keys,
+			SkipResourceFilter: true,
+			FullTextColumn:     b.fullTextColumn,
+			JsonKeyToKey:       b.jsonKeyToKey,
+			Variables:          variables,
+			StartNs:            start,
+			EndNs:              end,
+		})
+
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if preparedWhereClause != nil {
+		sb.AddWhereClause(preparedWhereClause.WhereClause)
+	}
+
+	startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
+	var endBucket uint64
+	if end != 0 {
+		endBucket = end / querybuilder.NsToSeconds
+	}
+
+	if start != 0 {
+		sb.Where(sb.GE("timestamp", fmt.Sprintf("%d", start)), sb.GE("ts_bucket_start", startBucket))
+	}
+	if end != 0 {
+		sb.Where(sb.L("timestamp", fmt.Sprintf("%d", end)), sb.LE("ts_bucket_start", endBucket))
+	}
+
+	return preparedWhereClause, nil
+}
+
+func aggOrderBy(k qbtypes.OrderBy, q qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]) (int, bool) {
+	for i, agg := range q.Aggregations {
+		if k.Key.Name == agg.Alias || k.Key.Name == agg.Expression || k.Key.Name == fmt.Sprintf("%d", i) {
+			return i, true
+		}
+	}
+	return 0, false
+}
+
+func (b *auditQueryStatementBuilder) maybeAttachResourceFilter(
+	ctx context.Context,
+	sb *sqlbuilder.SelectBuilder,
+	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
+	start, end uint64,
+	variables map[string]qbtypes.VariableItem,
+) (cteSQL string, cteArgs []any, err error) {
+	stmt, err := b.resourceFilterStmtBuilder.Build(ctx, start, end, qbtypes.RequestTypeRaw, query, variables)
+	if err != nil {
+		return "", nil, err
+	}
+
+	sb.Where("resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
+
+	return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args, nil
+}

pkg/telemetryaudit/statement_builder_test.go

@@ -0,0 +1,223 @@
+package telemetryaudit
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"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/types/telemetrytypes/telemetrytypestest"
+	"github.com/stretchr/testify/require"
+)
+
+func auditFieldKeyMap() map[string][]*telemetrytypes.TelemetryFieldKey {
+	key := func(name string, ctx telemetrytypes.FieldContext, dt telemetrytypes.FieldDataType, materialized bool) *telemetrytypes.TelemetryFieldKey {
+		return &telemetrytypes.TelemetryFieldKey{
+			Name:          name,
+			Signal:        telemetrytypes.SignalLogs,
+			FieldContext:  ctx,
+			FieldDataType: dt,
+			Materialized:  materialized,
+		}
+	}
+
+	attr := telemetrytypes.FieldContextAttribute
+	res := telemetrytypes.FieldContextResource
+	str := telemetrytypes.FieldDataTypeString
+	i64 := telemetrytypes.FieldDataTypeInt64
+
+	return map[string][]*telemetrytypes.TelemetryFieldKey{
+		"service.name":                 {key("service.name", res, str, false)},
+		"signoz.audit.action":          {key("signoz.audit.action", attr, str, true)},
+		"signoz.audit.outcome":         {key("signoz.audit.outcome", attr, str, true)},
+		"signoz.audit.principal.email": {key("signoz.audit.principal.email", attr, str, true)},
+		"signoz.audit.principal.id":    {key("signoz.audit.principal.id", attr, str, true)},
+		"signoz.audit.principal.type":  {key("signoz.audit.principal.type", attr, str, true)},
+		"signoz.audit.resource.kind":   {key("signoz.audit.resource.kind", res, str, false)},
+		"signoz.audit.resource.id":     {key("signoz.audit.resource.id", res, str, false)},
+		"signoz.audit.action_category": {key("signoz.audit.action_category", attr, str, false)},
+		"signoz.audit.error.type":      {key("signoz.audit.error.type", attr, str, false)},
+		"signoz.audit.error.code":      {key("signoz.audit.error.code", attr, str, false)},
+		"http.request.method":          {key("http.request.method", attr, str, false)},
+		"http.response.status_code":    {key("http.response.status_code", attr, i64, false)},
+	}
+}
+
+func newTestAuditStatementBuilder() *auditQueryStatementBuilder {
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore.KeysMap = auditFieldKeyMap()
+
+	fm := NewFieldMapper()
+	cb := NewConditionBuilder(fm)
+	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
+
+	return NewAuditQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		mockMetadataStore,
+		fm,
+		cb,
+		aggExprRewriter,
+		DefaultFullTextColumn,
+		nil,
+	)
+}
+
+func TestStatementBuilder(t *testing.T) {
+	statementBuilder := newTestAuditStatementBuilder()
+	ctx := context.Background()
+
+	testCases := []struct {
+		name        string
+		requestType qbtypes.RequestType
+		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
+		expected    qbtypes.Statement
+		expectedErr error
+	}{
+		// List: all actions by a specific user (materialized principal.id filter)
+		{
+			name:        "ListByPrincipalID",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{
+					Expression: "signoz.audit.principal.id = '019a-1234-abcd-5678'",
+				},
+				Limit: 100,
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) 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 resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$principal$$id` = ? AND `attribute_string_signoz$$audit$$principal$$id_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{uint64(1747945619), uint64(1747983448), "019a-1234-abcd-5678", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
+			},
+		},
+		// List: all failed actions (materialized outcome filter)
+		{
+			name:        "ListByOutcomeFailure",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{
+					Expression: "signoz.audit.outcome = 'failure'",
+				},
+				Limit: 100,
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) 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 resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{uint64(1747945619), uint64(1747983448), "failure", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
+			},
+		},
+		// List: change history of a specific dashboard (two materialized column AND)
+		{
+			name:        "ListByResourceKindAndID",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{
+					Expression: "signoz.audit.resource.kind = 'dashboard' AND signoz.audit.resource.id = '019b-5678-efgh-9012'",
+				},
+				Limit: 100,
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE ((simpleJSONExtractString(labels, 'signoz.audit.resource.kind') = ? AND labels LIKE ? AND labels LIKE ?) AND (simpleJSONExtractString(labels, 'signoz.audit.resource.id') = ? AND labels LIKE ? AND labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) 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 resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND true AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"dashboard", "%signoz.audit.resource.kind%", "%signoz.audit.resource.kind\":\"dashboard%", "019b-5678-efgh-9012", "%signoz.audit.resource.id%", "%signoz.audit.resource.id\":\"019b-5678-efgh-9012%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
+			},
+		},
+		// List: all dashboard deletions (compliance — resource.kind + action AND)
+		{
+			name:        "ListByResourceKindAndAction",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{
+					Expression: "signoz.audit.resource.kind = 'dashboard' AND signoz.audit.action = 'delete'",
+				},
+				Limit: 100,
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE (simpleJSONExtractString(labels, 'signoz.audit.resource.kind') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) 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 resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$action` = ? AND `attribute_string_signoz$$audit$$action_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"dashboard", "%signoz.audit.resource.kind%", "%signoz.audit.resource.kind\":\"dashboard%", uint64(1747945619), uint64(1747983448), "delete", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
+			},
+		},
+		// List: all actions by service accounts (materialized principal.type)
+		{
+			name:        "ListByPrincipalType",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{
+					Expression: "signoz.audit.principal.type = 'service_account'",
+				},
+				Limit: 100,
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) 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 resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$principal$$type` = ? AND `attribute_string_signoz$$audit$$principal$$type_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{uint64(1747945619), uint64(1747983448), "service_account", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
+			},
+		},
+		// Scalar: alert — count forbidden errors (outcome + action AND)
+		{
+			name:        "ScalarCountByOutcomeAndAction",
+			requestType: qbtypes.RequestTypeScalar,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				Source:       telemetrytypes.SourceAudit,
+				StepInterval: qbtypes.Step{Duration: 60 * time.Second},
+				Filter: &qbtypes.Filter{
+					Expression: "signoz.audit.outcome = 'failure' AND signoz.audit.action = 'update'",
+				},
+				Aggregations: []qbtypes.LogAggregation{
+					{Expression: "count()"},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT count() AS __result_0 FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND ((`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = ?) AND (`attribute_string_signoz$$audit$$action` = ? AND `attribute_string_signoz$$audit$$action_exists` = ?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY __result_0 DESC",
+				Args:  []any{uint64(1747945619), uint64(1747983448), "failure", true, "update", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
+			},
+		},
+		// TimeSeries: failures grouped by principal email with top-N limit
+		{
+			name:        "TimeSeriesFailuresGroupedByPrincipal",
+			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"}},
+				},
+				Limit: 5,
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?), __limit_cte AS (SELECT toString(multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = ?, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = ?) 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` = ?, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = ?) 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{uint64(1747945619), uint64(1747983448), true, "failure", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 5, true, "failure", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			q, err := statementBuilder.Build(ctx, 1747947419000, 1747983448000, testCase.requestType, testCase.query, nil)
+			if testCase.expectedErr != nil {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), testCase.expectedErr.Error())
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, testCase.expected.Query, q.Query)
+				require.Equal(t, testCase.expected.Args, q.Args)
+			}
+		})
+	}
+}

pkg/telemetryaudit/tables.go

@@ -0,0 +1,12 @@
+package telemetryaudit
+
+const (
+	DBName                      = "signoz_audit"
+	AuditLogsTableName          = "distributed_logs"
+	AuditLogsLocalTableName     = "logs"
+	TagAttributesTableName      = "distributed_tag_attributes"
+	TagAttributesLocalTableName = "tag_attributes"
+	LogAttributeKeysTblName     = "distributed_logs_attribute_keys"
+	LogResourceKeysTblName      = "distributed_logs_resource_keys"
+	LogsResourceTableName       = "distributed_logs_resource"
+)

pkg/telemetrylogs/field_mapper.go

@@ -276,11 +276,12 @@ func (m *fieldMapper) FieldFor(ctx context.Context, tsStart, tsEnd uint64, key *
 					continue
 				}
 
-				if key.JSONDataType == nil {
+				if key.FieldDataType == telemetrytypes.FieldDataTypeUnspecified {
 					return "", qbtypes.ErrColumnNotFound
 				}
 
-				if key.KeyNameContainsArray() && !key.JSONDataType.IsArray {
+				jdt := key.GetJSONDataType()
+				if key.KeyNameContainsArray() && !jdt.IsArray {
 					return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "FieldFor not supported for nested fields; only supported for flat paths (e.g. body.status.detail) and paths of Array type: %s(%s)", key.Name, key.FieldDataType)
 				}
 

pkg/telemetrylogs/json_stmt_builder_test.go

@@ -220,7 +220,7 @@ func TestJSONStmtBuilder_PrimitivePaths(t *testing.T) {
 			expected: TestExpected{
 				WhereClause: "(((LOWER(toString(dynamicElement(body_v2.`user.age`, 'Int64'))) LIKE LOWER(?)) AND has(JSONAllPaths(body_v2), 'user.age')) OR ((LOWER(dynamicElement(body_v2.`user.age`, 'String')) LIKE LOWER(?)) AND has(JSONAllPaths(body_v2), 'user.age')))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), "%25%", "%25%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings:    []string{"Key `user.age` is ambiguous, found 2 different combinations of field context / data type: [name=user.age,context=body,datatype=int64,jsondatatype=Int64 name=user.age,context=body,datatype=string,jsondatatype=String]."},
+				Warnings:    []string{"Key `user.age` is ambiguous, found 2 different combinations of field context / data type: [name=user.age,context=body,datatype=int64 name=user.age,context=body,datatype=string]."},
 			},
 		},
 		{
@@ -414,7 +414,7 @@ func TestStatementBuilderListQueryBodyPromoted(t *testing.T) {
 			},
 			expected: TestExpected{
 				Args:     []any{uint64(1747945619), uint64(1747983448), "%1.65%", 1.65, "%1.65%", 1.65, "%1.65%", 1.65, "%1.65%", 1.65, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings: []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,materialized=true,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,materialized=true,jsondatatype=Array(Dynamic)]."},
+				Warnings: []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,materialized=true name=education[].parameters,context=body,datatype=[]dynamic,materialized=true]."},
 			},
 			expectedErr: nil,
 		},
@@ -441,7 +441,7 @@ func TestStatementBuilderListQueryBodyPromoted(t *testing.T) {
 			},
 			expected: TestExpected{
 				Args:     []any{uint64(1747945619), uint64(1747983448), "%true%", true, "%true%", true, "%true%", true, "%true%", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings: []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,materialized=true,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,materialized=true,jsondatatype=Array(Dynamic)]."},
+				Warnings: []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,materialized=true name=education[].parameters,context=body,datatype=[]dynamic,materialized=true]."},
 			},
 			expectedErr: nil,
 		},
@@ -455,7 +455,7 @@ func TestStatementBuilderListQueryBodyPromoted(t *testing.T) {
 			},
 			expected: TestExpected{
 				Args:     []any{uint64(1747945619), uint64(1747983448), "%passed%", "passed", "%passed%", "passed", "%passed%", "passed", "%passed%", "passed", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings: []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,materialized=true,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,materialized=true,jsondatatype=Array(Dynamic)]."},
+				Warnings: []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,materialized=true name=education[].parameters,context=body,datatype=[]dynamic,materialized=true]."},
 			},
 			expectedErr: nil,
 		},
@@ -549,7 +549,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			expected: TestExpected{
 				WhereClause: "((NOT arrayExists(`body_v2.education`-> toFloat64OrNull(dynamicElement(`body_v2.education`.`type`, 'String')) = ?, dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND (NOT arrayExists(`body_v2.education`-> dynamicElement(`body_v2.education`.`type`, 'Int64') = ?, dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), int64(10001), int64(10001), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings:    []string{"Key `education[].type` is ambiguous, found 2 different combinations of field context / data type: [name=education[].type,context=body,datatype=string,jsondatatype=String name=education[].type,context=body,datatype=int64,jsondatatype=Int64]."},
+				Warnings:    []string{"Key `education[].type` is ambiguous, found 2 different combinations of field context / data type: [name=education[].type,context=body,datatype=string name=education[].type,context=body,datatype=int64]."},
 			},
 		},
 		{
@@ -576,7 +576,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			expected: TestExpected{
 				WhereClause: "(((arrayExists(`body_v2.education`-> (arrayExists(x -> LOWER(toString(x)) LIKE LOWER(?), dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))')) OR arrayExists(x -> toFloat64(x) = ?, dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))'))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'education')) OR ((arrayExists(`body_v2.education`-> (arrayExists(x -> LOWER(toString(x)) LIKE LOWER(?), arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)'))) OR arrayExists(x -> accurateCastOrNull(x, 'Float64') = ?, arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)')))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'education')))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), "%1.65%", 1.65, "%1.65%", 1.65, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,jsondatatype=Array(Dynamic)]."},
+				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64 name=education[].parameters,context=body,datatype=[]dynamic]."},
 			},
 		},
 		{
@@ -585,7 +585,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			expected: TestExpected{
 				WhereClause: "(((arrayExists(`body_v2.education`-> arrayExists(x -> toString(x) = ?, dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))')), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'education')) OR ((arrayExists(`body_v2.education`-> arrayExists(x -> toString(x) = ?, arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)'))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'education')))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), "passed", "passed", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,jsondatatype=Array(Dynamic)]."},
+				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64 name=education[].parameters,context=body,datatype=[]dynamic]."},
 			},
 		},
 		{
@@ -594,7 +594,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			expected: TestExpected{
 				WhereClause: "(((arrayExists(`body_v2.education`-> (arrayExists(x -> LOWER(toString(x)) LIKE LOWER(?), dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))')) OR arrayExists(x -> toString(x) = ?, dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))'))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'education')) OR ((arrayExists(`body_v2.education`-> (arrayExists(x -> LOWER(toString(x)) LIKE LOWER(?), arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)'))) OR arrayExists(x -> toString(x) = ?, arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)')))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'education')))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), "%passed%", "passed", "%passed%", "passed", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,jsondatatype=Array(Dynamic)]."},
+				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64 name=education[].parameters,context=body,datatype=[]dynamic]."},
 			},
 		},
 		{
@@ -603,7 +603,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			expected: TestExpected{
 				WhereClause: "(((arrayExists(`body_v2.education`-> arrayExists(x -> toFloat64(x) IN (?, ?), dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))')), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'education')) OR ((arrayExists(`body_v2.education`-> arrayExists(x -> toString(x) IN (?, ?), arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)'))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'education')))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), 1.65, 1.99, "1.65", "1.99", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,jsondatatype=Array(Dynamic)]."},
+				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64 name=education[].parameters,context=body,datatype=[]dynamic]."},
 			},
 		},
 		{
@@ -612,7 +612,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			expected: TestExpected{
 				WhereClause: "((NOT arrayExists(`body_v2.education`-> (arrayExists(x -> LOWER(toString(x)) LIKE LOWER(?), dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))')) OR arrayExists(x -> toFloat64(x) = ?, dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))'))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND (NOT arrayExists(`body_v2.education`-> (arrayExists(x -> LOWER(toString(x)) LIKE LOWER(?), arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)'))) OR arrayExists(x -> accurateCastOrNull(x, 'Float64') = ?, arrayFilter(x->(dynamicType(x) IN ('String', 'Int64', 'Float64', 'Bool')), dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)')))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), "%1.65%", float64(1.65), "%1.65%", float64(1.65), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,jsondatatype=Array(Dynamic)]."},
+				Warnings:    []string{"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64 name=education[].parameters,context=body,datatype=[]dynamic]."},
 			},
 		},
 		{
@@ -622,7 +622,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 				WhereClause: "(has(arrayFlatten(arrayConcat(arrayMap(`body_v2.education`->dynamicElement(`body_v2.education`.`parameters`, 'Array(Nullable(Float64))'), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))), ?) OR has(arrayFlatten(arrayConcat(arrayMap(`body_v2.education`->dynamicElement(`body_v2.education`.`parameters`, 'Array(Dynamic)'), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))), ?))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), 1.65, 1.65, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 				Warnings: []string{
-					"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64,jsondatatype=Array(Nullable(Float64)) name=education[].parameters,context=body,datatype=[]dynamic,jsondatatype=Array(Dynamic)].",
+					"Key `education[].parameters` is ambiguous, found 2 different combinations of field context / data type: [name=education[].parameters,context=body,datatype=[]float64 name=education[].parameters,context=body,datatype=[]dynamic].",
 				},
 			},
 		},
@@ -702,7 +702,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			expected: TestExpected{
 				WhereClause: "(((arrayExists(`body_v2.interests`-> arrayExists(`body_v2.interests[].entities`-> arrayExists(`body_v2.interests[].entities[].reviews`-> arrayExists(`body_v2.interests[].entities[].reviews[].entries`-> arrayExists(`body_v2.interests[].entities[].reviews[].entries[].metadata`-> arrayExists(`body_v2.interests[].entities[].reviews[].entries[].metadata[].positions`-> (arrayExists(x -> LOWER(toString(x)) LIKE LOWER(?), dynamicElement(`body_v2.interests[].entities[].reviews[].entries[].metadata[].positions`.`ratings`, 'Array(Nullable(Int64))')) OR arrayExists(x -> toFloat64(x) = ?, dynamicElement(`body_v2.interests[].entities[].reviews[].entries[].metadata[].positions`.`ratings`, 'Array(Nullable(Int64))'))), dynamicElement(`body_v2.interests[].entities[].reviews[].entries[].metadata`.`positions`, 'Array(JSON(max_dynamic_types=0, max_dynamic_paths=0))')), dynamicElement(`body_v2.interests[].entities[].reviews[].entries`.`metadata`, 'Array(JSON(max_dynamic_types=1, max_dynamic_paths=0))')), dynamicElement(`body_v2.interests[].entities[].reviews`.`entries`, 'Array(JSON(max_dynamic_types=2, max_dynamic_paths=0))')), dynamicElement(`body_v2.interests[].entities`.`reviews`, 'Array(JSON(max_dynamic_types=4, max_dynamic_paths=0))')), dynamicElement(`body_v2.interests`.`entities`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')), dynamicElement(body_v2.`interests`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'interests')) OR ((arrayExists(`body_v2.interests`-> arrayExists(`body_v2.interests[].entities`-> arrayExists(`body_v2.interests[].entities[].reviews`-> arrayExists(`body_v2.interests[].entities[].reviews[].entries`-> arrayExists(`body_v2.interests[].entities[].reviews[].entries[].metadata`-> arrayExists(`body_v2.interests[].entities[].reviews[].entries[].metadata[].positions`-> (arrayExists(x -> LOWER(x) LIKE LOWER(?), dynamicElement(`body_v2.interests[].entities[].reviews[].entries[].metadata[].positions`.`ratings`, 'Array(Nullable(String))')) OR arrayExists(x -> toFloat64OrNull(x) = ?, dynamicElement(`body_v2.interests[].entities[].reviews[].entries[].metadata[].positions`.`ratings`, 'Array(Nullable(String))'))), dynamicElement(`body_v2.interests[].entities[].reviews[].entries[].metadata`.`positions`, 'Array(JSON(max_dynamic_types=0, max_dynamic_paths=0))')), dynamicElement(`body_v2.interests[].entities[].reviews[].entries`.`metadata`, 'Array(JSON(max_dynamic_types=1, max_dynamic_paths=0))')), dynamicElement(`body_v2.interests[].entities[].reviews`.`entries`, 'Array(JSON(max_dynamic_types=2, max_dynamic_paths=0))')), dynamicElement(`body_v2.interests[].entities`.`reviews`, 'Array(JSON(max_dynamic_types=4, max_dynamic_paths=0))')), dynamicElement(`body_v2.interests`.`entities`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')), dynamicElement(body_v2.`interests`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND has(JSONAllPaths(body_v2), 'interests')))",
 				Args:        []any{uint64(1747945619), uint64(1747983448), "%4%", float64(4), "%4%", float64(4), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings:    []string{"Key `interests[].entities[].reviews[].entries[].metadata[].positions[].ratings` is ambiguous, found 2 different combinations of field context / data type: [name=interests[].entities[].reviews[].entries[].metadata[].positions[].ratings,context=body,datatype=[]int64,jsondatatype=Array(Nullable(Int64)) name=interests[].entities[].reviews[].entries[].metadata[].positions[].ratings,context=body,datatype=[]string,jsondatatype=Array(Nullable(String))]."},
+				Warnings:    []string{"Key `interests[].entities[].reviews[].entries[].metadata[].positions[].ratings` is ambiguous, found 2 different combinations of field context / data type: [name=interests[].entities[].reviews[].entries[].metadata[].positions[].ratings,context=body,datatype=[]int64 name=interests[].entities[].reviews[].entries[].metadata[].positions[].ratings,context=body,datatype=[]string]."},
 			},
 		},
 		{
@@ -946,16 +946,16 @@ func buildTestTelemetryMetadataStore(t *testing.T, addIndexes bool) *telemetryty
 	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
 	mockMetadataStore.SetStaticFields(IntrinsicFields)
 	types, _ := telemetrytypes.TestJSONTypeSet()
-	for path, jsonTypes := range types {
-		for _, jsonType := range jsonTypes {
+	for path, fieldDataTypes := range types {
+		for _, fdt := range fieldDataTypes {
 			key := &telemetrytypes.TelemetryFieldKey{
 				Name:          path,
 				Signal:        telemetrytypes.SignalLogs,
 				FieldContext:  telemetrytypes.FieldContextBody,
-				FieldDataType: telemetrytypes.MappingJSONDataTypeToFieldDataType[jsonType],
-				JSONDataType:  &jsonType,
+				FieldDataType: fdt,
 			}
 			if addIndexes {
+				jsonType := telemetrytypes.MappingFieldDataTypeToJSONDataType[fdt]
 				idx := slices.IndexFunc(telemetrytypes.TestIndexedPaths, func(entry telemetrytypes.TestIndexedPathEntry) bool {
 					return entry.Path == path && entry.Type == jsonType
 				})

pkg/telemetrylogs/statement_builder.go

@@ -45,6 +45,7 @@ func NewLogQueryStatementBuilder(
 		DBName,
 		LogsResourceV2TableName,
 		telemetrytypes.SignalLogs,
+		telemetrytypes.SourceUnspecified,
 		metadataStore,
 		fullTextColumn,
 		jsonKeyToKey,

pkg/telemetrylogs/stmt_builder_test.go

@@ -875,7 +875,6 @@ func TestAdjustKey(t *testing.T) {
 			require.Equal(t, c.expectedKey.FieldContext, key.FieldContext, "field context should match")
 			require.Equal(t, c.expectedKey.FieldDataType, key.FieldDataType, "field data type should match")
 			require.Equal(t, c.expectedKey.Materialized, key.Materialized, "materialized should match")
-			require.Equal(t, c.expectedKey.JSONDataType, key.JSONDataType, "json data type should match")
 			require.Equal(t, c.expectedKey.Indexes, key.Indexes, "json exists should match")
 		})
 	}

pkg/telemetrymetadata/body_json_metadata.go

@@ -21,133 +21,72 @@ import (
 )
 
 var (
-	defaultPathLimit = 100 // Default limit to prevent full table scans
-
-	CodeUnknownJSONDataType     = errors.MustNewCode("unknown_json_data_type")
 	CodeFailLoadPromotedPaths   = errors.MustNewCode("fail_load_promoted_paths")
 	CodeFailCheckPathPromoted   = errors.MustNewCode("fail_check_path_promoted")
-	CodeFailIterateBodyJSONKeys = errors.MustNewCode("fail_iterate_body_json_keys")
-	CodeFailExtractBodyJSONKeys = errors.MustNewCode("fail_extract_body_json_keys")
 	CodeFailLoadLogsJSONIndexes = errors.MustNewCode("fail_load_logs_json_indexes")
 	CodeFailListJSONValues      = errors.MustNewCode("fail_list_json_values")
 	CodeFailScanJSONValue       = errors.MustNewCode("fail_scan_json_value")
 	CodeFailScanVariant         = errors.MustNewCode("fail_scan_variant")
-	CodeFailBuildJSONPathsQuery = errors.MustNewCode("fail_build_json_paths_query")
-	CodeNoPathsToQueryIndexes   = errors.MustNewCode("no_paths_to_query_indexes_provided")
 
 	CodeFailedToPrepareBatch = errors.MustNewCode("failed_to_prepare_batch_promoted_paths")
 	CodeFailedToSendBatch    = errors.MustNewCode("failed_to_send_batch_promoted_paths")
 	CodeFailedToAppendPath   = errors.MustNewCode("failed_to_append_path_promoted_paths")
 )
 
-// fetchBodyJSONPaths extracts body JSON paths from the path_types table
-// This function can be used by both JSONQueryBuilder and metadata extraction
-// uniquePathLimit: 0 for no limit, >0 for maximum number of unique paths to return
-//   - For startup load: set to 10000 to get top 10k unique paths
-//   - For lookup: set to 0 (no limit needed for single path)
-//   - For metadata API: set to desired pagination limit
-//
-// searchOperator: LIKE for pattern matching, EQUAL for exact match.
-func (t *telemetryMetaStore) fetchBodyJSONPaths(ctx context.Context,
-	fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, []string, bool, error) {
-	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
-		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
-		instrumentationtypes.CodeNamespace:    "metadata",
-		instrumentationtypes.CodeFunctionName: "fetchBodyJSONPaths",
-	})
-
-	query, args, limit := buildGetBodyJSONPathsQuery(fieldKeySelectors)
-	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
-	if err != nil {
-		return nil, nil, false, errors.WrapInternalf(err, CodeFailExtractBodyJSONKeys, "failed to extract body JSON keys")
-	}
-	defer rows.Close()
-
-	fieldKeys := []*telemetrytypes.TelemetryFieldKey{}
-	paths := []string{}
-	rowCount := 0
-	for rows.Next() {
-		var path string
-		var typesArray []string // ClickHouse returns array as []string
-		var lastSeen uint64
-
-		err = rows.Scan(&path, &typesArray, &lastSeen)
-		if err != nil {
-			return nil, nil, false, errors.WrapInternalf(err, CodeFailExtractBodyJSONKeys, "failed to scan body JSON key row")
-		}
-
-		for _, typ := range typesArray {
-			mapping, found := telemetrytypes.MappingStringToJSONDataType[typ]
-			if !found {
-				t.logger.ErrorContext(ctx, "failed to map type string to JSON data type", slog.String("type", typ), slog.String("path", path))
-				continue
-			}
-			fieldKeys = append(fieldKeys, &telemetrytypes.TelemetryFieldKey{
-				Name:          path,
-				Signal:        telemetrytypes.SignalLogs,
-				FieldContext:  telemetrytypes.FieldContextBody,
-				FieldDataType: telemetrytypes.MappingJSONDataTypeToFieldDataType[mapping],
-				JSONDataType:  &mapping,
-			})
-		}
-
-		paths = append(paths, path)
-		rowCount++
-	}
-	if rows.Err() != nil {
-		return nil, nil, false, errors.WrapInternalf(rows.Err(), CodeFailIterateBodyJSONKeys, "error iterating body JSON keys")
+// enrichBodyKeys enriches body-context keys with promoted path info, indexes,
+// and JSON access plans.
+// parentTypeCache contains parent array types (ArrayJSON/ArrayDynamic) pre-fetched in the main UNION query.
+func (t *telemetryMetaStore) enrichBodyKeys(ctx context.Context, keys []*telemetrytypes.TelemetryFieldKey, parentTypeCache map[string][]telemetrytypes.FieldDataType) error {
+	if len(keys) == 0 {
+		return nil
 	}
 
-	return fieldKeys, paths, rowCount <= limit, nil
-}
-
-func (t *telemetryMetaStore) buildBodyJSONPaths(ctx context.Context,
-	fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
-
-	fieldKeys, paths, finished, err := t.fetchBodyJSONPaths(ctx, fieldKeySelectors)
-	if err != nil {
-		return nil, false, err
-	}
-
-	promoted, err := t.GetPromotedPaths(ctx, paths...)
-	if err != nil {
-		return nil, false, err
-	}
-
-	indexes, err := t.getJSONPathIndexes(ctx, paths...)
-	if err != nil {
-		return nil, false, err
-	}
-
-	for _, fieldKey := range fieldKeys {
-		promotedKey := strings.Split(fieldKey.Name, telemetrytypes.ArraySep)[0]
-		fieldKey.Materialized = promoted[promotedKey]
-		fieldKey.Indexes = indexes[fieldKey.Name]
-	}
-
-	return fieldKeys, finished, t.buildJSONPlans(ctx, fieldKeys)
-}
-
-func (t *telemetryMetaStore) buildJSONPlans(ctx context.Context, keys []*telemetrytypes.TelemetryFieldKey) error {
-	parentSelectors := make([]*telemetrytypes.FieldKeySelector, 0, len(keys))
+	var filteredKeys []*telemetrytypes.TelemetryFieldKey
 	for _, key := range keys {
-		parentSelectors = append(parentSelectors, key.ArrayParentSelectors()...)
+		if key.FieldContext == telemetrytypes.FieldContextBody {
+			filteredKeys = append(filteredKeys, key)
+		}
 	}
 
-	parentKeys, _, _, err := t.fetchBodyJSONPaths(ctx, parentSelectors)
+	// collect paths for batch queries
+	paths := make([]string, 0, len(filteredKeys))
+	for _, key := range filteredKeys {
+		paths = append(paths, key.Name)
+	}
+
+	// fetch promoted paths
+	promoted, err := t.GetPromotedPaths(ctx, paths...)
 	if err != nil {
 		return err
 	}
 
-	typeCache := make(map[string][]telemetrytypes.JSONDataType)
-	for _, key := range parentKeys {
-		typeCache[key.Name] = append(typeCache[key.Name], *key.JSONDataType)
+	// fetch JSON path indexes
+	indexes, err := t.getJSONPathIndexes(ctx, paths...)
+	if err != nil {
+		return err
 	}
 
-	// build plans for keys now
+	// apply promoted/index metadata to keys
+	for _, key := range filteredKeys {
+		promotedKey := strings.Split(key.Name, telemetrytypes.ArraySep)[0]
+		key.Materialized = promoted[promotedKey]
+		key.Indexes = indexes[key.Name]
+	}
+
+	// build JSON access plans using the pre-fetched parent type cache
+	return t.buildJSONPlans(ctx, filteredKeys, parentTypeCache)
+}
+
+// buildJSONPlans builds JSON access plans for the given keys
+// using the provided parent type cache (pre-fetched in the main UNION query).
+func (t *telemetryMetaStore) buildJSONPlans(_ context.Context, keys []*telemetrytypes.TelemetryFieldKey, typeCache map[string][]telemetrytypes.FieldDataType) error {
+	if len(keys) == 0 {
+		return nil
+	}
+
+	columnMeta := t.jsonColumnMetadata[telemetrytypes.SignalLogs][telemetrytypes.FieldContextBody]
 	for _, key := range keys {
-		err = key.SetJSONAccessPlan(t.jsonColumnMetadata[telemetrytypes.SignalLogs][telemetrytypes.FieldContextBody], typeCache)
-		if err != nil {
+		if err := key.SetJSONAccessPlan(columnMeta, typeCache); err != nil {
 			return err
 		}
 	}
@@ -155,51 +94,6 @@ func (t *telemetryMetaStore) buildJSONPlans(ctx context.Context, keys []*telemet
 	return nil
 }
 
-func buildGetBodyJSONPathsQuery(fieldKeySelectors []*telemetrytypes.FieldKeySelector) (string, []any, int) {
-	if len(fieldKeySelectors) == 0 {
-		return "", nil, defaultPathLimit
-	}
-	from := fmt.Sprintf("%s.%s", DBName, PathTypesTableName)
-
-	// Build a better query using GROUP BY to deduplicate at database level
-	// This aggregates all types per path and gets the max last_seen, then applies LIMIT
-	sb := sqlbuilder.Select(
-		"path",
-		"groupArray(DISTINCT type) AS types",
-		"max(last_seen) AS last_seen",
-	).From(from)
-
-	limit := 0
-	// Add search filter if provided
-	orClauses := []string{}
-	for _, fieldKeySelector := range fieldKeySelectors {
-		// replace [*] with []
-		fieldKeySelector.Name = strings.ReplaceAll(fieldKeySelector.Name, telemetrytypes.ArrayAnyIndex, telemetrytypes.ArraySep)
-		// Extract search text for body JSON keys
-		keyName := CleanPathPrefixes(fieldKeySelector.Name)
-		if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
-			orClauses = append(orClauses, sb.Equal("path", keyName))
-		} else {
-			// Pattern matching for metadata API (defaults to LIKE behavior for other operators)
-			orClauses = append(orClauses, sb.ILike("path", fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(keyName))))
-		}
-		limit += fieldKeySelector.Limit
-	}
-	sb.Where(sb.Or(orClauses...))
-	// Group by path to get unique paths with aggregated types
-	sb.GroupBy("path")
-
-	// Order by max last_seen to get most recent paths first
-	sb.OrderBy("last_seen DESC")
-	if limit == 0 {
-		limit = defaultPathLimit
-	}
-	sb.Limit(limit)
-
-	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
-	return query, args, limit
-}
-
 func (t *telemetryMetaStore) getJSONPathIndexes(ctx context.Context, paths ...string) (map[string][]telemetrytypes.JSONDataTypeIndex, error) {
 	filteredPaths := []string{}
 	for _, path := range paths {

pkg/telemetrymetadata/body_json_metadata_test.go

@@ -7,99 +7,9 @@ import (
 	"github.com/SigNoz/signoz-otel-collector/constants"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/stretchr/testify/require"
 )
 
-func TestBuildGetBodyJSONPathsQuery(t *testing.T) {
-	testCases := []struct {
-		name              string
-		fieldKeySelectors []*telemetrytypes.FieldKeySelector
-		expectedSQL       string
-		expectedArgs      []any
-		expectedLimit     int
-	}{
-
-		{
-			name: "Single search text with EQUAL operator",
-			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
-				{
-					Name:              "user.name",
-					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
-				},
-			},
-			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (path = ?) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
-			expectedArgs:  []any{"user.name", defaultPathLimit},
-			expectedLimit: defaultPathLimit,
-		},
-		{
-			name: "Single search text with LIKE operator",
-			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
-				{
-					Name:              "user",
-					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeFuzzy,
-				},
-			},
-			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (LOWER(path) LIKE LOWER(?)) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
-			expectedArgs:  []any{"%user%", 100},
-			expectedLimit: 100,
-		},
-		{
-			name: "Multiple search texts with EQUAL operator",
-			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
-				{
-					Name:              "user.name",
-					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
-				},
-				{
-					Name:              "user.age",
-					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
-				},
-			},
-			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (path = ? OR path = ?) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
-			expectedArgs:  []any{"user.name", "user.age", defaultPathLimit},
-			expectedLimit: defaultPathLimit,
-		},
-		{
-			name: "Multiple search texts with LIKE operator",
-			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
-				{
-					Name:              "user",
-					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeFuzzy,
-				},
-				{
-					Name:              "admin",
-					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeFuzzy,
-				},
-			},
-			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (LOWER(path) LIKE LOWER(?) OR LOWER(path) LIKE LOWER(?)) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
-			expectedArgs:  []any{"%user%", "%admin%", defaultPathLimit},
-			expectedLimit: defaultPathLimit,
-		},
-		{
-			name: "Search with Contains operator (should default to LIKE)",
-			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
-				{
-					Name:              "test",
-					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeFuzzy,
-				},
-			},
-			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (LOWER(path) LIKE LOWER(?)) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
-			expectedArgs:  []any{"%test%", defaultPathLimit},
-			expectedLimit: defaultPathLimit,
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			query, args, limit := buildGetBodyJSONPathsQuery(tc.fieldKeySelectors)
-			require.Equal(t, tc.expectedSQL, query)
-			require.Equal(t, tc.expectedArgs, args)
-			require.Equal(t, tc.expectedLimit, limit)
-		})
-	}
-}
-
 func TestBuildListLogsJSONIndexesQuery(t *testing.T) {
 	testCases := []struct {
 		name         string

pkg/telemetrymetadata/metadata.go

@@ -13,6 +13,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymetrics"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
@@ -27,6 +28,7 @@ import (
 var (
 	ErrFailedToGetTracesKeys    = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get traces keys")
 	ErrFailedToGetLogsKeys      = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get logs keys")
+	ErrFailedToGetAuditKeys     = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get audit keys")
 	ErrFailedToGetTblStatement  = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get tbl statement")
 	ErrFailedToGetMetricsKeys   = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get metrics keys")
 	ErrFailedToGetMeterKeys     = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get meter keys")
@@ -50,6 +52,11 @@ type telemetryMetaStore struct {
 	logAttributeKeysTblName        string
 	logResourceKeysTblName         string
 	logsV2TblName                  string
+	auditDBName                    string
+	auditLogsTblName               string
+	auditFieldsTblName             string
+	auditAttributeKeysTblName      string
+	auditResourceKeysTblName       string
 	relatedMetadataDBName          string
 	relatedMetadataTblName         string
 	columnEvolutionMetadataTblName string
@@ -79,6 +86,11 @@ func NewTelemetryMetaStore(
 	logsFieldsTblName string,
 	logAttributeKeysTblName string,
 	logResourceKeysTblName string,
+	auditDBName string,
+	auditLogsTblName string,
+	auditFieldsTblName string,
+	auditAttributeKeysTblName string,
+	auditResourceKeysTblName string,
 	relatedMetadataDBName string,
 	relatedMetadataTblName string,
 	columnEvolutionMetadataTblName string,
@@ -101,6 +113,11 @@ func NewTelemetryMetaStore(
 		logsFieldsTblName:              logsFieldsTblName,
 		logAttributeKeysTblName:        logAttributeKeysTblName,
 		logResourceKeysTblName:         logResourceKeysTblName,
+		auditDBName:                    auditDBName,
+		auditLogsTblName:               auditLogsTblName,
+		auditFieldsTblName:             auditFieldsTblName,
+		auditAttributeKeysTblName:      auditAttributeKeysTblName,
+		auditResourceKeysTblName:       auditResourceKeysTblName,
 		relatedMetadataDBName:          relatedMetadataDBName,
 		relatedMetadataTblName:         relatedMetadataTblName,
 		columnEvolutionMetadataTblName: columnEvolutionMetadataTblName,
@@ -351,6 +368,19 @@ func (t *telemetryMetaStore) logsTblStatementToFieldKeys(ctx context.Context) ([
 	return materialisedKeys, nil
 }
 
+// logKeysUnionArm declares one arm of the UNION ALL in getLogsKeys.
+// All per-table variance is captured here so the loop body can stay uniform.
+type logKeysUnionArm struct {
+	shouldQuery        bool
+	fieldContext       telemetrytypes.FieldContext
+	table              string
+	dataTypeColumn     string                                    // column used in WHERE/GROUP BY
+	dataTypeSelectExpr string                                    // expression used in SELECT (may wrap with lower())
+	addBaseFilters     func(sb *sqlbuilder.SelectBuilder)        // mandatory WHERE filters (e.g., signal, field_context)
+	encodeDataType     func(telemetrytypes.FieldDataType) string // how to render a FieldDataType in WHERE values
+	extraOrBranch      func(sb *sqlbuilder.SelectBuilder) string // optional extra OR branch (e.g., body parent-types)
+}
+
 // getLogsKeys returns the keys from the spans that match the field selection criteria.
 func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
 	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
@@ -380,90 +410,149 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 	// tables to query based on field selectors
 	queryAttributeTable := false
 	queryResourceTable := false
+	queryBodyTable := false
 
 	for _, selector := range fieldKeySelectors {
 		if selector.FieldContext == telemetrytypes.FieldContextUnspecified {
-			// unspecified context, query both tables
+			// unspecified context, query all tables
 			queryAttributeTable = true
 			queryResourceTable = true
+			queryBodyTable = true
 			break
 		} else if selector.FieldContext == telemetrytypes.FieldContextAttribute {
 			queryAttributeTable = true
 		} else if selector.FieldContext == telemetrytypes.FieldContextResource {
 			queryResourceTable = true
+		} else if selector.FieldContext == telemetrytypes.FieldContextBody && querybuilder.BodyJSONQueryEnabled {
+			queryBodyTable = true
 		}
 	}
 
-	tablesToQuery := []struct {
-		fieldContext telemetrytypes.FieldContext
-		shouldQuery  bool
-	}{
-		{telemetrytypes.FieldContextAttribute, queryAttributeTable},
-		{telemetrytypes.FieldContextResource, queryResourceTable},
+	// body keys are gated behind the feature flag
+	queryBodyTable = queryBodyTable && querybuilder.BodyJSONQueryEnabled
+
+	// pre-compute parent array path names from body selectors for JSON plan building;
+	// these will be fetched as a separate UNION arm filtered to ArrayJSON/ArrayDynamic only
+	parentPaths := make(map[string]bool)
+	if queryBodyTable {
+		for _, sel := range fieldKeySelectors {
+			if sel.FieldContext != telemetrytypes.FieldContextBody &&
+				sel.FieldContext != telemetrytypes.FieldContextUnspecified {
+				continue
+			}
+			if !strings.Contains(sel.Name, telemetrytypes.ArraySep) {
+				continue
+			}
+			key := &telemetrytypes.TelemetryFieldKey{
+				Name:         sel.Name,
+				Signal:       telemetrytypes.SignalLogs,
+				FieldContext: telemetrytypes.FieldContextBody,
+			}
+			for _, ps := range key.ArrayParentSelectors() {
+				parentPaths[ps.Name] = true
+			}
+		}
 	}
 
-	for _, table := range tablesToQuery {
-		if !table.shouldQuery {
+	// Each UNION arm differs only in: table, data-type column name and SELECT
+	// expression (lower-wrapped for historical mixed-case in attr/resource),
+	// base WHERE filters, the per-selector data-type encoding, and (for body)
+	// an extra OR branch that fetches parent array types for JSON plan building.
+	// All other logic is shared by the loop below.
+	tablesToQuery := []logKeysUnionArm{
+		{
+			shouldQuery:        queryAttributeTable,
+			fieldContext:       telemetrytypes.FieldContextAttribute,
+			table:              t.logsDBName + "." + t.logAttributeKeysTblName,
+			dataTypeColumn:     "datatype",
+			dataTypeSelectExpr: "lower(datatype)",
+			addBaseFilters:     func(*sqlbuilder.SelectBuilder) {},
+			encodeDataType:     func(ft telemetrytypes.FieldDataType) string { return ft.TagDataType() },
+			extraOrBranch:      func(*sqlbuilder.SelectBuilder) string { return "" },
+		},
+		{
+			shouldQuery:        queryResourceTable,
+			fieldContext:       telemetrytypes.FieldContextResource,
+			table:              t.logsDBName + "." + t.logResourceKeysTblName,
+			dataTypeColumn:     "datatype",
+			dataTypeSelectExpr: "lower(datatype)",
+			addBaseFilters:     func(*sqlbuilder.SelectBuilder) {},
+			encodeDataType:     func(ft telemetrytypes.FieldDataType) string { return ft.TagDataType() },
+			extraOrBranch:      func(*sqlbuilder.SelectBuilder) string { return "" },
+		},
+		{
+			shouldQuery:        queryBodyTable,
+			fieldContext:       telemetrytypes.FieldContextBody,
+			table:              fmt.Sprintf("%s.%s", DBName, FieldKeysTable),
+			dataTypeColumn:     "field_data_type",
+			dataTypeSelectExpr: "field_data_type",
+			addBaseFilters: func(sb *sqlbuilder.SelectBuilder) {
+				sb.Where(sb.E("signal", telemetrytypes.SignalLogs.StringValue()))
+				sb.Where(sb.E("field_context", telemetrytypes.FieldContextBody.StringValue()))
+			},
+			encodeDataType: func(ft telemetrytypes.FieldDataType) string { return ft.StringValue() },
+			extraOrBranch: func(sb *sqlbuilder.SelectBuilder) string {
+				if len(parentPaths) == 0 {
+					return ""
+				}
+				names := make([]any, 0, len(parentPaths))
+				for n := range parentPaths {
+					names = append(names, n)
+				}
+				return sb.And(
+					sb.In("name", names...),
+					sb.In("field_data_type",
+						telemetrytypes.FieldDataTypeArrayDynamic.StringValue(),
+						telemetrytypes.FieldDataTypeArrayJSON.StringValue(),
+					),
+				)
+			},
+		},
+	}
+
+	for _, arm := range tablesToQuery {
+		if !arm.shouldQuery {
 			continue
 		}
 
-		fieldContext := table.fieldContext
-
-		// table name based on field context
-		var tblName string
-		if fieldContext == telemetrytypes.FieldContextAttribute {
-			tblName = t.logsDBName + "." + t.logAttributeKeysTblName
-		} else {
-			tblName = t.logsDBName + "." + t.logResourceKeysTblName
-		}
-
 		sb := sqlbuilder.Select(
 			"name AS tag_key",
-			fmt.Sprintf("'%s' AS tag_type", fieldContext.TagType()),
-			"lower(datatype) AS tag_data_type", // in logs, we had some historical data with capital and small case
-			fmt.Sprintf(`%d AS priority`, getPriorityForContext(fieldContext)),
-		).From(tblName)
+			fmt.Sprintf("'%s' AS tag_type", arm.fieldContext.TagType()),
+			arm.dataTypeSelectExpr+" AS tag_data_type",
+			fmt.Sprintf("%d AS priority", getPriorityForContext(arm.fieldContext)),
+		).From(arm.table)
 
-		var limit int
-		conds := []string{}
+		arm.addBaseFilters(sb)
 
-		for _, fieldKeySelector := range fieldKeySelectors {
-			// Include this selector if:
-			// 1. It has unspecified context (matches all tables)
-			// 2. Its context matches the current table's context
-			if fieldKeySelector.FieldContext != telemetrytypes.FieldContextUnspecified &&
-				fieldKeySelector.FieldContext != fieldContext {
+		branches := []string{}
+		for _, sel := range fieldKeySelectors {
+			if sel.FieldContext != telemetrytypes.FieldContextUnspecified && sel.FieldContext != arm.fieldContext {
 				continue
 			}
-
-			// key part of the selector
-			fieldKeyConds := []string{}
-			if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
-				fieldKeyConds = append(fieldKeyConds, sb.E("name", fieldKeySelector.Name))
+			parts := []string{}
+			if sel.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+				parts = append(parts, sb.E("name", sel.Name))
 			} else {
-				fieldKeyConds = append(fieldKeyConds, sb.ILike("name", "%"+escapeForLike(fieldKeySelector.Name)+"%"))
+				parts = append(parts, sb.ILike("name", "%"+escapeForLike(sel.Name)+"%"))
 			}
-
-			// now look at the field data type
-			if fieldKeySelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
-				fieldKeyConds = append(fieldKeyConds, sb.E("datatype", fieldKeySelector.FieldDataType.TagDataType()))
+			if sel.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+				parts = append(parts, sb.E(arm.dataTypeColumn, arm.encodeDataType(sel.FieldDataType)))
 			}
-
-			if len(fieldKeyConds) > 0 {
-				conds = append(conds, sb.And(fieldKeyConds...))
+			if len(parts) > 0 {
+				branches = append(branches, sb.And(parts...))
 			}
-			limit += fieldKeySelector.Limit
 		}
 
-		if len(conds) > 0 {
-			sb.Where(sb.Or(conds...))
+		if extra := arm.extraOrBranch(sb); extra != "" {
+			branches = append(branches, extra)
 		}
 
-		sb.GroupBy("name", "datatype")
-		if limit == 0 {
-			limit = 1000
+		if len(branches) > 0 {
+			sb.Where(sb.Or(branches...))
 		}
 
+		sb.GroupBy("name", arm.dataTypeColumn)
+
 		query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
 		queries = append(queries, query)
 		allArgs = append(allArgs, args...)
@@ -500,6 +589,7 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 	defer rows.Close()
 
 	keys := []*telemetrytypes.TelemetryFieldKey{}
+	parentTypeCache := make(map[string][]telemetrytypes.FieldDataType)
 	rowCount := 0
 	searchTexts := []string{}
 
@@ -523,6 +613,17 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 		if err != nil {
 			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
 		}
+
+		// body keys with ArrayJSON/ArrayDynamic types are internal container types
+		// used only for JSON access plan building; route to parentTypeCache, not to results
+		switch fieldDataType {
+		case telemetrytypes.FieldDataTypeArrayJSON, telemetrytypes.FieldDataTypeArrayDynamic:
+			if fieldContext == telemetrytypes.FieldContextBody && parentPaths[name] {
+				parentTypeCache[name] = append(parentTypeCache[name], fieldDataType)
+				continue
+			}
+		}
+
 		key, ok := mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
 
 		// if there is no materialised column, create a key with the field context and data type
@@ -576,13 +677,11 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 		}
 	}
 
+	// enrich body keys with promoted paths, indexes, and JSON access plans
 	if querybuilder.BodyJSONQueryEnabled {
-		bodyJSONPaths, finished, err := t.buildBodyJSONPaths(ctx, fieldKeySelectors) // LIKE for pattern matching
-		if err != nil {
-			t.logger.ErrorContext(ctx, "failed to extract body JSON paths", errors.Attr(err))
+		if err := t.enrichBodyKeys(ctx, keys, parentTypeCache); err != nil {
+			t.logger.ErrorContext(ctx, "failed to enrich body JSON keys", errors.Attr(err))
 		}
-		keys = append(keys, bodyJSONPaths...)
-		complete = complete && finished
 	}
 
 	if _, err := t.updateColumnEvolutionMetadataForKeys(ctx, keys); err != nil {
@@ -592,6 +691,227 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 	return keys, complete, nil
 }
 
+func (t *telemetryMetaStore) auditTblStatementToFieldKeys(ctx context.Context) ([]*telemetrytypes.TelemetryFieldKey, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "auditTblStatementToFieldKeys",
+	})
+
+	query := fmt.Sprintf("SHOW CREATE TABLE %s.%s", t.auditDBName, t.auditLogsTblName)
+	statements := []telemetrytypes.ShowCreateTableStatement{}
+	err := t.telemetrystore.ClickhouseDB().Select(ctx, &statements, query)
+	if err != nil {
+		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTblStatement.Error())
+	}
+
+	materialisedKeys, err := ExtractFieldKeysFromTblStatement(statements[0].Statement)
+	if err != nil {
+		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
+	}
+
+	for idx := range materialisedKeys {
+		materialisedKeys[idx].Signal = telemetrytypes.SignalLogs
+	}
+
+	return materialisedKeys, nil
+}
+
+func (t *telemetryMetaStore) getAuditKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getAuditKeys",
+	})
+
+	if len(fieldKeySelectors) == 0 {
+		return nil, true, nil
+	}
+
+	matKeys, err := t.auditTblStatementToFieldKeys(ctx)
+	if err != nil {
+		return nil, false, err
+	}
+	mapOfKeys := make(map[string]*telemetrytypes.TelemetryFieldKey)
+	for _, key := range matKeys {
+		mapOfKeys[key.Name+";"+key.FieldContext.StringValue()+";"+key.FieldDataType.StringValue()] = key
+	}
+
+	var queries []string
+	var allArgs []any
+
+	queryAttributeTable := false
+	queryResourceTable := false
+
+	for _, selector := range fieldKeySelectors {
+		if selector.FieldContext == telemetrytypes.FieldContextUnspecified {
+			queryAttributeTable = true
+			queryResourceTable = true
+			break
+		} else if selector.FieldContext == telemetrytypes.FieldContextAttribute {
+			queryAttributeTable = true
+		} else if selector.FieldContext == telemetrytypes.FieldContextResource {
+			queryResourceTable = true
+		}
+	}
+
+	tablesToQuery := []struct {
+		fieldContext telemetrytypes.FieldContext
+		shouldQuery  bool
+		tblName      string
+	}{
+		{telemetrytypes.FieldContextAttribute, queryAttributeTable, t.auditDBName + "." + t.auditAttributeKeysTblName},
+		{telemetrytypes.FieldContextResource, queryResourceTable, t.auditDBName + "." + t.auditResourceKeysTblName},
+	}
+
+	for _, table := range tablesToQuery {
+		if !table.shouldQuery {
+			continue
+		}
+
+		fieldContext := table.fieldContext
+		tblName := table.tblName
+
+		sb := sqlbuilder.Select(
+			"name AS tag_key",
+			fmt.Sprintf("'%s' AS tag_type", fieldContext.TagType()),
+			"lower(datatype) AS tag_data_type",
+			fmt.Sprintf("%d AS priority", getPriorityForContext(fieldContext)),
+		).From(tblName)
+
+		var limit int
+		conds := []string{}
+
+		for _, fieldKeySelector := range fieldKeySelectors {
+			if fieldKeySelector.FieldContext != telemetrytypes.FieldContextUnspecified && fieldKeySelector.FieldContext != fieldContext {
+				continue
+			}
+
+			fieldKeyConds := []string{}
+			if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+				fieldKeyConds = append(fieldKeyConds, sb.E("name", fieldKeySelector.Name))
+			} else {
+				fieldKeyConds = append(fieldKeyConds, sb.ILike("name", "%"+escapeForLike(fieldKeySelector.Name)+"%"))
+			}
+
+			if fieldKeySelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+				fieldKeyConds = append(fieldKeyConds, sb.E("datatype", fieldKeySelector.FieldDataType.TagDataType()))
+			}
+
+			if len(fieldKeyConds) > 0 {
+				conds = append(conds, sb.And(fieldKeyConds...))
+			}
+			limit += fieldKeySelector.Limit
+		}
+
+		if len(conds) > 0 {
+			sb.Where(sb.Or(conds...))
+		}
+
+		sb.GroupBy("name", "datatype")
+		if limit == 0 {
+			limit = 1000
+		}
+
+		query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+		queries = append(queries, query)
+		allArgs = append(allArgs, args...)
+	}
+
+	if len(queries) == 0 {
+		return []*telemetrytypes.TelemetryFieldKey{}, true, nil
+	}
+
+	var limit int
+	for _, fieldKeySelector := range fieldKeySelectors {
+		limit += fieldKeySelector.Limit
+	}
+	if limit == 0 {
+		limit = 1000
+	}
+
+	mainQuery := fmt.Sprintf(`
+		SELECT tag_key, tag_type, tag_data_type, max(priority) as priority
+		FROM (
+			%s
+		) AS combined_results
+		GROUP BY tag_key, tag_type, tag_data_type
+		ORDER BY priority
+		LIMIT %d
+	`, strings.Join(queries, " UNION ALL "), limit+1)
+
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, mainQuery, allArgs...)
+	if err != nil {
+		return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
+	}
+	defer rows.Close()
+
+	keys := []*telemetrytypes.TelemetryFieldKey{}
+	rowCount := 0
+	searchTexts := []string{}
+
+	for _, fieldKeySelector := range fieldKeySelectors {
+		searchTexts = append(searchTexts, fieldKeySelector.Name)
+	}
+
+	for rows.Next() {
+		rowCount++
+		if rowCount > limit {
+			break
+		}
+
+		var name string
+		var fieldContext telemetrytypes.FieldContext
+		var fieldDataType telemetrytypes.FieldDataType
+		var priority uint8
+		err = rows.Scan(&name, &fieldContext, &fieldDataType, &priority)
+		if err != nil {
+			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
+		}
+		key, ok := mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
+
+		if !ok {
+			key = &telemetrytypes.TelemetryFieldKey{
+				Name:          name,
+				Signal:        telemetrytypes.SignalLogs,
+				FieldContext:  fieldContext,
+				FieldDataType: fieldDataType,
+			}
+		}
+
+		keys = append(keys, key)
+		mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()] = key
+	}
+
+	if rows.Err() != nil {
+		return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
+	}
+
+	complete := rowCount <= limit
+
+	// Add intrinsic audit fields (same as logs intrinsics: body, severity_text, etc.)
+	staticKeys := maps.Keys(telemetryaudit.IntrinsicFields)
+	for _, key := range staticKeys {
+		found := false
+		for _, v := range searchTexts {
+			if v == "" || strings.Contains(key, v) {
+				found = true
+				break
+			}
+		}
+
+		if found {
+			if field, exists := telemetryaudit.IntrinsicFields[key]; exists {
+				if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
+					keys = append(keys, &field)
+				}
+			}
+		}
+	}
+
+	return keys, complete, nil
+}
+
 func getPriorityForContext(ctx telemetrytypes.FieldContext) int {
 	switch ctx {
 	case telemetrytypes.FieldContextLog:
@@ -889,7 +1209,11 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, fieldKeySelector *tele
 	case telemetrytypes.SignalTraces:
 		keys, complete, err = t.getTracesKeys(ctx, selectors)
 	case telemetrytypes.SignalLogs:
-		keys, complete, err = t.getLogsKeys(ctx, selectors)
+		if fieldKeySelector.Source == telemetrytypes.SourceAudit {
+			keys, complete, err = t.getAuditKeys(ctx, selectors)
+		} else {
+			keys, complete, err = t.getLogsKeys(ctx, selectors)
+		}
 	case telemetrytypes.SignalMetrics:
 		if fieldKeySelector.Source == telemetrytypes.SourceMeter {
 			keys, complete, err = t.getMeterSourceMetricKeys(ctx, selectors)
@@ -938,6 +1262,7 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, fieldKeySelector *tele
 func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) (map[string][]*telemetrytypes.TelemetryFieldKey, bool, error) {
 
 	logsSelectors := []*telemetrytypes.FieldKeySelector{}
+	auditSelectors := []*telemetrytypes.FieldKeySelector{}
 	tracesSelectors := []*telemetrytypes.FieldKeySelector{}
 	metricsSelectors := []*telemetrytypes.FieldKeySelector{}
 	meterSourceMetricsSelectors := []*telemetrytypes.FieldKeySelector{}
@@ -945,7 +1270,11 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
 	for _, fieldKeySelector := range fieldKeySelectors {
 		switch fieldKeySelector.Signal {
 		case telemetrytypes.SignalLogs:
-			logsSelectors = append(logsSelectors, fieldKeySelector)
+			if fieldKeySelector.Source == telemetrytypes.SourceAudit {
+				auditSelectors = append(auditSelectors, fieldKeySelector)
+			} else {
+				logsSelectors = append(logsSelectors, fieldKeySelector)
+			}
 		case telemetrytypes.SignalTraces:
 			tracesSelectors = append(tracesSelectors, fieldKeySelector)
 		case telemetrytypes.SignalMetrics:
@@ -965,6 +1294,10 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
 	if err != nil {
 		return nil, false, err
 	}
+	auditKeys, auditComplete, err := t.getAuditKeys(ctx, auditSelectors)
+	if err != nil {
+		return nil, false, err
+	}
 	tracesKeys, tracesComplete, err := t.getTracesKeys(ctx, tracesSelectors)
 	if err != nil {
 		return nil, false, err
@@ -979,12 +1312,15 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
 		return nil, false, err
 	}
 	// Complete only if all queries are complete
-	complete := logsComplete && tracesComplete && metricsComplete
+	complete := logsComplete && auditComplete && tracesComplete && metricsComplete
 
 	mapOfKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey)
 	for _, key := range logsKeys {
 		mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
 	}
+	for _, key := range auditKeys {
+		mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
+	}
 	for _, key := range tracesKeys {
 		mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
 	}
@@ -1338,6 +1674,97 @@ func (t *telemetryMetaStore) getLogFieldValues(ctx context.Context, fieldValueSe
 	return values, complete, nil
 }
 
+func (t *telemetryMetaStore) getAuditFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getAuditFieldValues",
+	})
+
+	limit := fieldValueSelector.Limit
+	if limit == 0 {
+		limit = 50
+	}
+
+	sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(t.auditDBName + "." + t.auditFieldsTblName)
+
+	if fieldValueSelector.Name != "" {
+		sb.Where(sb.E("tag_key", fieldValueSelector.Name))
+	}
+
+	if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
+		sb.Where(sb.E("tag_type", fieldValueSelector.FieldContext.TagType()))
+	}
+
+	if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+		sb.Where(sb.E("tag_data_type", fieldValueSelector.FieldDataType.TagDataType()))
+	}
+
+	if fieldValueSelector.Value != "" {
+		switch fieldValueSelector.FieldDataType {
+		case telemetrytypes.FieldDataTypeString:
+			sb.Where(sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
+		case telemetrytypes.FieldDataTypeNumber:
+			sb.Where(sb.IsNotNull("number_value"))
+			sb.Where(sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
+		case telemetrytypes.FieldDataTypeUnspecified:
+			sb.Where(sb.Or(
+				sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
+				sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
+			))
+		}
+	}
+
+	// fetch one extra row to detect whether the result set is complete
+	sb.Limit(limit + 1)
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
+	}
+	defer rows.Close()
+
+	values := &telemetrytypes.TelemetryFieldValues{}
+	seen := make(map[string]bool)
+	rowCount := 0
+	totalCount := 0
+
+	for rows.Next() {
+		rowCount++
+
+		var stringValue string
+		var numberValue float64
+		err = rows.Scan(&stringValue, &numberValue)
+		if err != nil {
+			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
+		}
+		if stringValue != "" && !seen[stringValue] {
+			if totalCount >= limit {
+				break
+			}
+			values.StringValues = append(values.StringValues, stringValue)
+			seen[stringValue] = true
+			totalCount++
+		}
+		if numberValue != 0 {
+			if totalCount >= limit {
+				break
+			}
+			if !seen[fmt.Sprintf("%f", numberValue)] {
+				values.NumberValues = append(values.NumberValues, numberValue)
+				seen[fmt.Sprintf("%f", numberValue)] = true
+				totalCount++
+			}
+		}
+	}
+
+	complete := rowCount <= limit
+
+	return values, complete, nil
+}
+
 // getMetricFieldValues returns field values and whether the result is complete.
 func (t *telemetryMetaStore) getMetricFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
 	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
@@ -1628,7 +2055,11 @@ func (t *telemetryMetaStore) GetAllValues(ctx context.Context, fieldValueSelecto
 	case telemetrytypes.SignalTraces:
 		values, complete, err = t.getSpanFieldValues(ctx, fieldValueSelector)
 	case telemetrytypes.SignalLogs:
-		values, complete, err = t.getLogFieldValues(ctx, fieldValueSelector)
+		if fieldValueSelector.Source == telemetrytypes.SourceAudit {
+			values, complete, err = t.getAuditFieldValues(ctx, fieldValueSelector)
+		} else {
+			values, complete, err = t.getLogFieldValues(ctx, fieldValueSelector)
+		}
 	case telemetrytypes.SignalMetrics:
 		if fieldValueSelector.Source == telemetrytypes.SourceMeter {
 			values, complete, err = t.getMeterSourceMetricFieldValues(ctx, fieldValueSelector)

pkg/telemetrymetadata/metadata_query_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymeter"
 	"github.com/SigNoz/signoz/pkg/telemetrymetrics"
@@ -37,6 +38,11 @@ func TestGetFirstSeenFromMetricMetadata(t *testing.T) {
 		telemetrylogs.TagAttributesV2TableName,
 		telemetrylogs.LogAttributeKeysTblName,
 		telemetrylogs.LogResourceKeysTblName,
+		telemetryaudit.DBName,
+		telemetryaudit.AuditLogsTableName,
+		telemetryaudit.TagAttributesTableName,
+		telemetryaudit.LogAttributeKeysTblName,
+		telemetryaudit.LogResourceKeysTblName,
 		DBName,
 		AttributesMetadataLocalTableName,
 		ColumnEvolutionMetadataTableName,

pkg/telemetrymetadata/metadata_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymeter"
 	"github.com/SigNoz/signoz/pkg/telemetrymetrics"
@@ -36,6 +37,11 @@ func newTestTelemetryMetaStoreTestHelper(store telemetrystore.TelemetryStore) te
 		telemetrylogs.TagAttributesV2TableName,
 		telemetrylogs.LogAttributeKeysTblName,
 		telemetrylogs.LogResourceKeysTblName,
+		telemetryaudit.DBName,
+		telemetryaudit.AuditLogsTableName,
+		telemetryaudit.TagAttributesTableName,
+		telemetryaudit.LogAttributeKeysTblName,
+		telemetryaudit.LogResourceKeysTblName,
 		DBName,
 		AttributesMetadataLocalTableName,
 		ColumnEvolutionMetadataTableName,

pkg/telemetrymetadata/tables.go

@@ -1,14 +1,12 @@
 package telemetrymetadata
 
-import otelcollectorconst "github.com/SigNoz/signoz-otel-collector/constants"
-
 const (
 	DBName                           = "signoz_metadata"
 	AttributesMetadataTableName      = "distributed_attributes_metadata"
 	AttributesMetadataLocalTableName = "attributes_metadata"
 	ColumnEvolutionMetadataTableName = "distributed_column_evolution_metadata"
-	PathTypesTableName               = otelcollectorconst.DistributedPathTypesTable
+	FieldKeysTable                   = "distributed_field_keys"
 	// Column Evolution table stores promoted paths as (signal, column_name, field_context, field_name); see signoz-otel-collector metadata_migrations.
-	PromotedPathsTableName           = "distributed_column_evolution_metadata"
-	SkipIndexTableName               = "system.data_skipping_indices"
+	PromotedPathsTableName = "distributed_column_evolution_metadata"
+	SkipIndexTableName     = "system.data_skipping_indices"
 )

pkg/telemetryresourcefilter/statement_builder.go

@@ -21,6 +21,7 @@ type resourceFilterStatementBuilder[T any] struct {
 	conditionBuilder qbtypes.ConditionBuilder
 	metadataStore    telemetrytypes.MetadataStore
 	signal           telemetrytypes.Signal
+	source           telemetrytypes.Source
 
 	fullTextColumn *telemetrytypes.TelemetryFieldKey
 	jsonKeyToKey   qbtypes.JsonKeyToFieldFunc
@@ -37,6 +38,7 @@ func New[T any](
 	dbName string,
 	tableName string,
 	signal telemetrytypes.Signal,
+	source telemetrytypes.Source,
 	metadataStore telemetrytypes.MetadataStore,
 	fullTextColumn *telemetrytypes.TelemetryFieldKey,
 	jsonKeyToKey qbtypes.JsonKeyToFieldFunc,
@@ -52,6 +54,7 @@ func New[T any](
 		conditionBuilder: cb,
 		metadataStore:    metadataStore,
 		signal:           signal,
+		source:           source,
 		fullTextColumn:   fullTextColumn,
 		jsonKeyToKey:     jsonKeyToKey,
 	}
@@ -72,6 +75,7 @@ func (b *resourceFilterStatementBuilder[T]) getKeySelectors(query qbtypes.QueryB
 			continue
 		}
 		keySelectors[idx].Signal = b.signal
+		keySelectors[idx].Source = b.source
 		keySelectors[idx].SelectorMatchType = telemetrytypes.FieldSelectorMatchTypeExact
 		filteredKeySelectors = append(filteredKeySelectors, keySelectors[idx])
 	}

pkg/telemetryresourcefilter/statement_builder_test.go

@@ -375,6 +375,7 @@ func TestResourceFilterStatementBuilder_Traces(t *testing.T) {
 		"signoz_traces",
 		"distributed_traces_v3_resource",
 		telemetrytypes.SignalTraces,
+		telemetrytypes.SourceUnspecified,
 		mockMetadataStore,
 		nil,
 		nil,
@@ -592,6 +593,7 @@ func TestResourceFilterStatementBuilder_Logs(t *testing.T) {
 		"signoz_logs",
 		"distributed_logs_v2_resource",
 		telemetrytypes.SignalLogs,
+		telemetrytypes.SourceUnspecified,
 		mockMetadataStore,
 		nil,
 		nil,
@@ -653,6 +655,7 @@ func TestResourceFilterStatementBuilder_Variables(t *testing.T) {
 		"signoz_traces",
 		"distributed_traces_v3_resource",
 		telemetrytypes.SignalTraces,
+		telemetrytypes.SourceUnspecified,
 		mockMetadataStore,
 		nil,
 		nil,

pkg/telemetrytraces/statement_builder.go

@@ -49,6 +49,7 @@ func NewTraceQueryStatementBuilder(
 		DBName,
 		TracesResourceV3TableName,
 		telemetrytypes.SignalTraces,
+		telemetrytypes.SourceUnspecified,
 		metadataStore,
 		nil,
 		nil,

pkg/telemetrytraces/trace_operator_statement_builder.go

@@ -39,6 +39,7 @@ func NewTraceOperatorStatementBuilder(
 		DBName,
 		TracesResourceV3TableName,
 		telemetrytypes.SignalTraces,
+		telemetrytypes.SourceUnspecified,
 		metadataStore,
 		nil,
 		nil,

pkg/types/cloudintegrationtypes/service.go

@@ -62,6 +62,10 @@ type GettableServicesMetadata struct {
 	Services []*ServiceMetadata `json:"services" required:"true" nullable:"false"`
 }
 
+type ListServicesMetadataParams struct {
+	CloudIntegrationID valuer.UUID `query:"cloud_integration_id" required:"false"`
+}
+
 // Service represents a cloud integration service with its definition,
 // cloud integration service is non nil only when the service entry exists in DB with ANY config (enabled or disabled).
 type Service struct {
@@ -69,6 +73,10 @@ type Service struct {
 	CloudIntegrationService *CloudIntegrationService `json:"cloudIntegrationService" required:"true" nullable:"true"`
 }
 
+type GetServiceParams struct {
+	CloudIntegrationID valuer.UUID `query:"cloud_integration_id" required:"false"`
+}
+
 type UpdatableService struct {
 	Config *ServiceConfig `json:"config" required:"true" nullable:"false"`
 }

pkg/types/telemetrytypes/field.go

@@ -37,7 +37,6 @@ type TelemetryFieldKey struct {
 	FieldContext  FieldContext  `json:"fieldContext,omitzero"`
 	FieldDataType FieldDataType `json:"fieldDataType,omitzero"`
 
-	JSONDataType *JSONDataType       `json:"-"`
 	JSONPlan     JSONAccessPlan      `json:"-"`
 	Indexes      []JSONDataTypeIndex `json:"-"`
 	Materialized bool                `json:"-"` // refers to promoted in case of body.... fields
@@ -80,6 +79,12 @@ func (f *TelemetryFieldKey) ArrayParentSelectors() []*FieldKeySelector {
 	return selectors
 }
 
+// GetJSONDataType derives the JSONDataType from FieldDataType.
+// Callers should check FieldDataType != FieldDataTypeUnspecified before calling.
+func (f *TelemetryFieldKey) GetJSONDataType() JSONDataType {
+	return MappingFieldDataTypeToJSONDataType[f.FieldDataType]
+}
+
 func (f TelemetryFieldKey) String() string {
 	var sb strings.Builder
 	fmt.Fprintf(&sb, "name=%s", f.Name)
@@ -92,9 +97,6 @@ func (f TelemetryFieldKey) String() string {
 	if f.Materialized {
 		sb.WriteString(",materialized=true")
 	}
-	if f.JSONDataType != nil {
-		fmt.Fprintf(&sb, ",jsondatatype=%s", f.JSONDataType.StringValue())
-	}
 	if len(f.Indexes) > 0 {
 		sb.WriteString(",indexes=[")
 		for i, index := range f.Indexes {
@@ -117,7 +119,6 @@ func (f TelemetryFieldKey) Text() string {
 func (f *TelemetryFieldKey) OverrideMetadataFrom(src *TelemetryFieldKey) {
 	f.FieldContext = src.FieldContext
 	f.FieldDataType = src.FieldDataType
-	f.JSONDataType = src.JSONDataType
 	f.Indexes = src.Indexes
 	f.Materialized = src.Materialized
 	f.JSONPlan = src.JSONPlan

pkg/types/telemetrytypes/field_datatype.go

@@ -31,7 +31,7 @@ var (
 	FieldDataTypeArrayInt64  = FieldDataType{valuer.NewString("[]int64")}
 	FieldDataTypeArrayNumber = FieldDataType{valuer.NewString("[]number")}
 
-	FieldDataTypeArrayObject  = FieldDataType{valuer.NewString("[]object")}
+	FieldDataTypeArrayJSON    = FieldDataType{valuer.NewString("[]object")}
 	FieldDataTypeArrayDynamic = FieldDataType{valuer.NewString("[]dynamic")}
 
 	// Map string representations to FieldDataType values
@@ -72,6 +72,8 @@ var (
 		"[]float64": FieldDataTypeArrayFloat64,
 		"[]number":  FieldDataTypeArrayNumber,
 		"[]bool":    FieldDataTypeArrayBool,
+		"[]json":    FieldDataTypeArrayJSON,
+		"[]dynamic": FieldDataTypeArrayDynamic,
 
 		// c-style array types
 		"string[]":  FieldDataTypeArrayString,
@@ -79,6 +81,8 @@ var (
 		"float64[]": FieldDataTypeArrayFloat64,
 		"number[]":  FieldDataTypeArrayNumber,
 		"bool[]":    FieldDataTypeArrayBool,
+		"json[]":    FieldDataTypeArrayJSON,
+		"dynamic[]": FieldDataTypeArrayDynamic,
 	}
 
 	fieldDataTypeToCHDataType = map[FieldDataType]string{

pkg/types/telemetrytypes/json_access_plan.go

@@ -43,7 +43,7 @@ type JSONAccessNode struct {
 	isRoot     bool // marked true for only body_v2 and body_promoted
 
 	// Precomputed type information (single source of truth)
-	AvailableTypes []JSONDataType
+	AvailableTypes []FieldDataType
 
 	// Array type branches (Array(JSON) vs Array(Dynamic))
 	Branches map[JSONAccessBranchType]*JSONAccessNode
@@ -106,7 +106,7 @@ type planBuilder struct {
 	paths      []string // cumulative paths for type cache lookups
 	segments   []string // individual path segments for node names
 	isPromoted bool
-	typeCache  map[string][]JSONDataType
+	typeCache  map[string][]FieldDataType
 }
 
 // buildPlan recursively builds the path plan tree.
@@ -155,14 +155,14 @@ func (pb *planBuilder) buildPlan(index int, parent *JSONAccessNode, isDynArrChil
 		MaxDynamicPaths: maxPaths,
 	}
 
-	hasJSON := slices.Contains(node.AvailableTypes, ArrayJSON)
-	hasDynamic := slices.Contains(node.AvailableTypes, ArrayDynamic)
+	hasJSON := slices.Contains(node.AvailableTypes, FieldDataTypeArrayJSON)
+	hasDynamic := slices.Contains(node.AvailableTypes, FieldDataTypeArrayDynamic)
 
 	// Configure terminal if this is the last part
 	if isTerminal {
 		node.TerminalConfig = &TerminalConfig{
 			Key:      pb.key,
-			ElemType: *pb.key.JSONDataType,
+			ElemType: pb.key.GetJSONDataType(),
 		}
 	} else {
 		var err error
@@ -185,7 +185,7 @@ func (pb *planBuilder) buildPlan(index int, parent *JSONAccessNode, isDynArrChil
 
 // buildJSONAccessPlan builds a tree structure representing the complete JSON path traversal
 // that precomputes all possible branches and their types.
-func (key *TelemetryFieldKey) SetJSONAccessPlan(columnInfo JSONColumnMetadata, typeCache map[string][]JSONDataType,
+func (key *TelemetryFieldKey) SetJSONAccessPlan(columnInfo JSONColumnMetadata, typeCache map[string][]FieldDataType,
 ) error {
 	// if path is empty, return nil
 	if key.Name == "" {

pkg/types/telemetrytypes/json_access_plan_test.go

@@ -19,11 +19,11 @@ const (
 // ============================================================================
 
 // makeKey creates a TelemetryFieldKey for testing.
-func makeKey(name string, dataType JSONDataType, materialized bool) *TelemetryFieldKey {
+func makeKey(name string, dataType FieldDataType, materialized bool) *TelemetryFieldKey {
 	return &TelemetryFieldKey{
-		Name:         name,
-		JSONDataType: &dataType,
-		Materialized: materialized,
+		Name:          name,
+		FieldDataType: dataType,
+		Materialized:  materialized,
 	}
 }
 
@@ -242,7 +242,7 @@ func TestPlanJSON_BasicStructure(t *testing.T) {
 	}{
 		{
 			name: "Simple path not promoted",
-			key:  makeKey("user.name", String, false),
+			key:  makeKey("user.name", FieldDataTypeString, false),
 			expectedYAML: fmt.Sprintf(`
 - name: user.name
   column: %s
@@ -255,7 +255,7 @@ func TestPlanJSON_BasicStructure(t *testing.T) {
 		},
 		{
 			name: "Simple path promoted",
-			key:  makeKey("user.name", String, true),
+			key:  makeKey("user.name", FieldDataTypeString, true),
 			expectedYAML: fmt.Sprintf(`
 - name: user.name
   column: %s
@@ -276,7 +276,7 @@ func TestPlanJSON_BasicStructure(t *testing.T) {
 		},
 		{
 			name:         "Empty path returns error",
-			key:          makeKey("", String, false),
+			key:          makeKey("", FieldDataTypeString, false),
 			expectErr:    true,
 			expectedYAML: "",
 		},
@@ -431,7 +431,7 @@ func TestPlanJSON_ArrayPaths(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			key := makeKey(tt.path, String, false)
+			key := makeKey(tt.path, FieldDataTypeString, false)
 			err := key.SetJSONAccessPlan(JSONColumnMetadata{
 				BaseColumn:     bodyV2Column,
 				PromotedColumn: bodyPromotedColumn,
@@ -450,7 +450,7 @@ func TestPlanJSON_PromotedVsNonPromoted(t *testing.T) {
 	path := "education[].awards[].type"
 
 	t.Run("Non-promoted plan", func(t *testing.T) {
-		key := makeKey(path, String, false)
+		key := makeKey(path, FieldDataTypeString, false)
 		err := key.SetJSONAccessPlan(JSONColumnMetadata{
 			BaseColumn:     bodyV2Column,
 			PromotedColumn: bodyPromotedColumn,
@@ -493,7 +493,7 @@ func TestPlanJSON_PromotedVsNonPromoted(t *testing.T) {
 	})
 
 	t.Run("Promoted plan", func(t *testing.T) {
-		key := makeKey(path, String, true)
+		key := makeKey(path, FieldDataTypeString, true)
 		err := key.SetJSONAccessPlan(JSONColumnMetadata{
 			BaseColumn:     bodyV2Column,
 			PromotedColumn: bodyPromotedColumn,
@@ -666,12 +666,12 @@ func TestPlanJSON_EdgeCases(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			// Choose key type based on path; operator does not affect the tree shape asserted here.
-			keyType := String
+			keyType := FieldDataTypeString
 			switch tt.path {
 			case "education":
-				keyType = ArrayJSON
+				keyType = FieldDataTypeArrayJSON
 			case "education[].type":
-				keyType = String
+				keyType = FieldDataTypeString
 			}
 			key := makeKey(tt.path, keyType, false)
 			err := key.SetJSONAccessPlan(JSONColumnMetadata{
@@ -692,7 +692,7 @@ func TestPlanJSON_EdgeCases(t *testing.T) {
 func TestPlanJSON_TreeStructure(t *testing.T) {
 	types, _ := TestJSONTypeSet()
 	path := "education[].awards[].participated[].team[].branch"
-	key := makeKey(path, String, false)
+	key := makeKey(path, FieldDataTypeString, false)
 	err := key.SetJSONAccessPlan(JSONColumnMetadata{
 		BaseColumn:     bodyV2Column,
 		PromotedColumn: bodyPromotedColumn,

pkg/types/telemetrytypes/json_datatype.go

@@ -46,14 +46,6 @@ var MappingStringToJSONDataType = map[string]JSONDataType{
 	"Array(JSON)":              ArrayJSON,
 }
 
-var ScalerTypeToArrayType = map[JSONDataType]JSONDataType{
-	String:  ArrayString,
-	Int64:   ArrayInt64,
-	Float64: ArrayFloat64,
-	Bool:    ArrayBool,
-	Dynamic: ArrayDynamic,
-}
-
 var MappingFieldDataTypeToJSONDataType = map[FieldDataType]JSONDataType{
 	FieldDataTypeString:       String,
 	FieldDataTypeInt64:        Int64,
@@ -63,18 +55,8 @@ var MappingFieldDataTypeToJSONDataType = map[FieldDataType]JSONDataType{
 	FieldDataTypeArrayString:  ArrayString,
 	FieldDataTypeArrayInt64:   ArrayInt64,
 	FieldDataTypeArrayFloat64: ArrayFloat64,
+	FieldDataTypeArrayNumber:  ArrayFloat64,
 	FieldDataTypeArrayBool:    ArrayBool,
-}
-
-var MappingJSONDataTypeToFieldDataType = map[JSONDataType]FieldDataType{
-	String:       FieldDataTypeString,
-	Int64:        FieldDataTypeInt64,
-	Float64:      FieldDataTypeFloat64,
-	Bool:         FieldDataTypeBool,
-	ArrayString:  FieldDataTypeArrayString,
-	ArrayInt64:   FieldDataTypeArrayInt64,
-	ArrayFloat64: FieldDataTypeArrayFloat64,
-	ArrayBool:    FieldDataTypeArrayBool,
-	ArrayDynamic: FieldDataTypeArrayDynamic,
-	ArrayJSON:    FieldDataTypeArrayObject,
+	FieldDataTypeArrayDynamic: ArrayDynamic,
+	FieldDataTypeArrayJSON:    ArrayJSON,
 }

pkg/types/telemetrytypes/source.go

@@ -7,11 +7,13 @@ type Source struct {
 }
 
 var (
+	SourceAudit       = Source{valuer.NewString("audit")}
 	SourceMeter       = Source{valuer.NewString("meter")}
 	SourceUnspecified = Source{valuer.NewString("")}
 )
 
 // Enum returns the acceptable values for Source.
+// TODO: Add SourceAudit once the frontend is ready for consumption.
 func (Source) Enum() []any {
 	return []any{
 		SourceMeter,

pkg/types/telemetrytypes/test_data.go

@@ -4,69 +4,69 @@ package telemetrytypes
 // Test JSON Type Set Data Setup
 // ============================================================================
 
-// TestJSONTypeSet returns a map of path->types for testing.
+// TestJSONTypeSet returns a map of path->field data types for testing.
 // This represents the type information available in the test JSON structure.
-func TestJSONTypeSet() (map[string][]JSONDataType, MetadataStore) {
-	types := map[string][]JSONDataType{
+func TestJSONTypeSet() (map[string][]FieldDataType, MetadataStore) {
+	types := map[string][]FieldDataType{
 
 		// ── user (primitives) ─────────────────────────────────────────────
-		"user.name":        {String},
-		"user.permissions": {ArrayString},
-		"user.age":         {Int64, String}, // Int64/String ambiguity
-		"user.height":      {Float64},
-		"user.active":      {Bool}, // Bool — not IndexSupported
+		"user.name":        {FieldDataTypeString},
+		"user.permissions": {FieldDataTypeArrayString},
+		"user.age":         {FieldDataTypeInt64, FieldDataTypeString}, // Int64/String ambiguity
+		"user.height":      {FieldDataTypeFloat64},
+		"user.active":      {FieldDataTypeBool}, // Bool — not IndexSupported
 
 		// Deeper non-array nesting (a.b.c — no array hops)
-		"user.address.zip": {Int64},
+		"user.address.zip": {FieldDataTypeInt64},
 
 		// ── education[] ───────────────────────────────────────────────────
 		// Pattern: x[].y
-		"education":              {ArrayJSON},
-		"education[].name":       {String},
-		"education[].type":       {String, Int64},
-		"education[].year":       {Int64},
-		"education[].scores":     {ArrayInt64},
-		"education[].parameters": {ArrayFloat64, ArrayDynamic},
+		"education":              {FieldDataTypeArrayJSON},
+		"education[].name":       {FieldDataTypeString},
+		"education[].type":       {FieldDataTypeString, FieldDataTypeInt64},
+		"education[].year":       {FieldDataTypeInt64},
+		"education[].scores":     {FieldDataTypeArrayInt64},
+		"education[].parameters": {FieldDataTypeArrayFloat64, FieldDataTypeArrayDynamic},
 
 		// Pattern: x[].y[]
-		"education[].awards": {ArrayDynamic, ArrayJSON},
+		"education[].awards": {FieldDataTypeArrayDynamic, FieldDataTypeArrayJSON},
 
 		// Pattern: x[].y[].z
-		"education[].awards[].name":     {String},
-		"education[].awards[].type":     {String},
-		"education[].awards[].semester": {Int64},
+		"education[].awards[].name":     {FieldDataTypeString},
+		"education[].awards[].type":     {FieldDataTypeString},
+		"education[].awards[].semester": {FieldDataTypeInt64},
 
 		// Pattern: x[].y[].z[]
-		"education[].awards[].participated": {ArrayDynamic, ArrayJSON},
+		"education[].awards[].participated": {FieldDataTypeArrayDynamic, FieldDataTypeArrayJSON},
 
 		// Pattern: x[].y[].z[].w
-		"education[].awards[].participated[].members": {ArrayString},
+		"education[].awards[].participated[].members": {FieldDataTypeArrayString},
 
 		// Pattern: x[].y[].z[].w[]
-		"education[].awards[].participated[].team": {ArrayJSON},
+		"education[].awards[].participated[].team": {FieldDataTypeArrayJSON},
 
 		// Pattern: x[].y[].z[].w[].v
-		"education[].awards[].participated[].team[].branch": {String},
+		"education[].awards[].participated[].team[].branch": {FieldDataTypeString},
 
 		// ── interests[] ───────────────────────────────────────────────────
-		"interests":                                                                 {ArrayJSON},
-		"interests[].entities":                                                      {ArrayJSON},
-		"interests[].entities[].reviews":                                            {ArrayJSON},
-		"interests[].entities[].reviews[].entries":                                  {ArrayJSON},
-		"interests[].entities[].reviews[].entries[].metadata":                       {ArrayJSON},
-		"interests[].entities[].reviews[].entries[].metadata[].positions":           {ArrayJSON},
-		"interests[].entities[].reviews[].entries[].metadata[].positions[].name":    {String},
-		"interests[].entities[].reviews[].entries[].metadata[].positions[].ratings": {ArrayInt64, ArrayString},
-		"http-events":                     {ArrayJSON},
-		"http-events[].request-info.host": {String},
-		"ids":                             {ArrayDynamic},
+		"interests":                                                                 {FieldDataTypeArrayJSON},
+		"interests[].entities":                                                      {FieldDataTypeArrayJSON},
+		"interests[].entities[].reviews":                                            {FieldDataTypeArrayJSON},
+		"interests[].entities[].reviews[].entries":                                  {FieldDataTypeArrayJSON},
+		"interests[].entities[].reviews[].entries[].metadata":                       {FieldDataTypeArrayJSON},
+		"interests[].entities[].reviews[].entries[].metadata[].positions":           {FieldDataTypeArrayJSON},
+		"interests[].entities[].reviews[].entries[].metadata[].positions[].name":    {FieldDataTypeString},
+		"interests[].entities[].reviews[].entries[].metadata[].positions[].ratings": {FieldDataTypeArrayInt64, FieldDataTypeArrayString},
+		"http-events":                     {FieldDataTypeArrayJSON},
+		"http-events[].request-info.host": {FieldDataTypeString},
+		"ids":                             {FieldDataTypeArrayDynamic},
 
 		// ── top-level primitives ──────────────────────────────────────────
-		"message":     {String},
-		"http-status": {Int64, String}, // hyphen in root key, ambiguous
+		"message":     {FieldDataTypeString},
+		"http-status": {FieldDataTypeInt64, FieldDataTypeString}, // hyphen in root key, ambiguous
 
 		// ── top-level nested objects (no array hops) ───────────────────────
-		"response.time-taken": {Float64}, // hyphen inside nested key
+		"response.time-taken": {FieldDataTypeFloat64}, // hyphen inside nested key
 	}
 
 	return types, nil

pkg/types/tracedetailtypes/waterfall.go

@@ -19,6 +19,7 @@ type WaterfallRequest struct {
 type WaterfallResponse struct {
 	StartTimestampMillis          uint64            `json:"startTimestampMillis"`
 	EndTimestampMillis            uint64            `json:"endTimestampMillis"`
+	DurationNano                  uint64            `json:"durationNano"`
 	RootServiceName               string            `json:"rootServiceName"`
 	RootServiceEntryPoint         string            `json:"rootServiceEntryPoint"`
 	TotalSpansCount               uint64            `json:"totalSpansCount"`
@@ -27,7 +28,6 @@ type WaterfallResponse struct {
 	Spans                         []*WaterfallSpan  `json:"spans"`
 	HasMissingSpans               bool              `json:"hasMissingSpans"`
 	UncollapsedSpans              []string          `json:"uncollapsedSpans"`
-	HasMore                       bool              `json:"hasMore"`
 }
 
 // Event represents a span event.
@@ -58,15 +58,16 @@ type WaterfallSpan struct {
 	IsRemote           string            `json:"is_remote"`
 	Kind               int32             `json:"kind"`
 	KindString         string            `json:"kind_string"`
+	Links              string            `json:"links"`
 	Name               string            `json:"name"`
 	ParentSpanID       string            `json:"parent_span_id"`
 	Resource           map[string]string `json:"resource"`
 	ResponseStatusCode string            `json:"response_status_code"`
 	SpanID             string            `json:"span_id"`
-	StatusCode         int16             `json:"status_code"`
+	StatusCode         int32             `json:"status_code"`
 	StatusCodeString   string            `json:"status_code_string"`
 	StatusMessage      string            `json:"status_message"`
-	TimeUnixMilli      uint64            `json:"timestamp"`
+	Timestamp          string            `json:"timestamp"`
 	TraceID            string            `json:"trace_id"`
 	TraceState         string            `json:"trace_state"`
 
@@ -76,7 +77,10 @@ type WaterfallSpan struct {
 	HasChildren      bool             `json:"has_children"`
 	Level            uint64           `json:"level"`
 
-	// used only for service time calculation
+	// timeUnixNano is an internal field used for tree building and sorting.
+	// It is not serialized in the JSON response.
+	TimeUnixNano uint64 `json:"-"`
+	// serviceName is an internal field used for service time calculation.
 	ServiceName string `json:"-"`
 }
 
@@ -113,7 +117,7 @@ type SpanModel struct {
 	Flags              uint32             `ch:"flags"`
 	IsRemote           string             `ch:"is_remote"`
 	TraceState         string             `ch:"trace_state"`
-	StatusCode         int16              `ch:"status_code"`
+	StatusCode         int32              `ch:"status_code"`
 	DBName             string             `ch:"db_name"`
 	DBOperation        string             `ch:"db_operation"`
 	HTTPMethod         string             `ch:"http_method"`
@@ -145,7 +149,7 @@ func (item *SpanModel) ToSpan() *WaterfallSpan {
 	for _, eventStr := range item.Events {
 		var event Event
 		if err := json.Unmarshal([]byte(eventStr), &event); err != nil {
-			continue // skipping malformed events
+			continue
 		}
 		events = append(events, event)
 	}
@@ -166,6 +170,7 @@ func (item *SpanModel) ToSpan() *WaterfallSpan {
 		IsRemote:           item.IsRemote,
 		Kind:               int32(item.Kind),
 		KindString:         item.SpanKind,
+		Links:              item.References,
 		Name:               item.Name,
 		ParentSpanID:       item.ParentSpanID,
 		Resource:           resources,
@@ -174,10 +179,11 @@ func (item *SpanModel) ToSpan() *WaterfallSpan {
 		StatusCode:         item.StatusCode,
 		StatusCodeString:   item.StatusCodeString,
 		StatusMessage:      item.StatusMessage,
+		Timestamp:          item.TimeUnixNano.Format(time.RFC3339Nano),
 		TraceID:            item.TraceID,
 		TraceState:         item.TraceState,
 		Children:           make([]*WaterfallSpan, 0),
-		TimeUnixMilli:      uint64(item.TimeUnixNano.UnixNano() / 1000_000),
+		TimeUnixNano:       uint64(item.TimeUnixNano.UnixNano()),
 		ServiceName:        item.ServiceName,
 	}
 }

tests/integration/conftest.py

@@ -12,6 +12,7 @@ pytest_plugins = [
     "fixtures.sqlite",
     "fixtures.zookeeper",
     "fixtures.signoz",
+    "fixtures.audit",
     "fixtures.logs",
     "fixtures.traces",
     "fixtures.metrics",

tests/integration/fixtures/audit.py

@@ -0,0 +1,404 @@
+import datetime
+import json
+from abc import ABC
+from typing import Any, Callable, Generator, List, Optional
+
+import numpy as np
+import pytest
+from ksuid import KsuidMs
+
+from fixtures import types
+from fixtures.fingerprint import LogsOrTracesFingerprint
+
+
+class AuditResource(ABC):
+    labels: str
+    fingerprint: str
+    seen_at_ts_bucket_start: np.int64
+
+    def __init__(
+        self,
+        labels: dict[str, str],
+        fingerprint: str,
+        seen_at_ts_bucket_start: np.int64,
+    ) -> None:
+        self.labels = json.dumps(labels, separators=(",", ":"))
+        self.fingerprint = fingerprint
+        self.seen_at_ts_bucket_start = seen_at_ts_bucket_start
+
+    def np_arr(self) -> np.array:
+        return np.array(
+            [
+                self.labels,
+                self.fingerprint,
+                self.seen_at_ts_bucket_start,
+            ]
+        )
+
+
+class AuditResourceOrAttributeKeys(ABC):
+    name: str
+    datatype: str
+
+    def __init__(self, name: str, datatype: str) -> None:
+        self.name = name
+        self.datatype = datatype
+
+    def np_arr(self) -> np.array:
+        return np.array([self.name, self.datatype])
+
+
+class AuditTagAttributes(ABC):
+    unix_milli: np.int64
+    tag_key: str
+    tag_type: str
+    tag_data_type: str
+    string_value: str
+    int64_value: Optional[np.int64]
+    float64_value: Optional[np.float64]
+
+    def __init__(
+        self,
+        timestamp: datetime.datetime,
+        tag_key: str,
+        tag_type: str,
+        tag_data_type: str,
+        string_value: Optional[str],
+        int64_value: Optional[np.int64],
+        float64_value: Optional[np.float64],
+    ) -> None:
+        self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
+        self.tag_key = tag_key
+        self.tag_type = tag_type
+        self.tag_data_type = tag_data_type
+        self.string_value = string_value or ""
+        self.int64_value = int64_value
+        self.float64_value = float64_value
+
+    def np_arr(self) -> np.array:
+        return np.array(
+            [
+                self.unix_milli,
+                self.tag_key,
+                self.tag_type,
+                self.tag_data_type,
+                self.string_value,
+                self.int64_value,
+                self.float64_value,
+            ]
+        )
+
+
+class AuditLog(ABC):
+    """Represents a single audit log event in signoz_audit.
+
+    Matches the ClickHouse DDL from the schema migration (ticket #1936):
+    - Database: signoz_audit
+    - Local table: logs
+    - Distributed table: distributed_logs
+    - No resources_string column (resource JSON only)
+    - Has event_name column
+    - 7 materialized columns auto-populated from attributes_string at INSERT time
+    """
+
+    ts_bucket_start: np.uint64
+    resource_fingerprint: str
+    timestamp: np.uint64
+    observed_timestamp: np.uint64
+    id: str
+    trace_id: str
+    span_id: str
+    trace_flags: np.uint32
+    severity_text: str
+    severity_number: np.uint8
+    body: str
+    scope_name: str
+    scope_version: str
+    scope_string: dict[str, str]
+    attributes_string: dict[str, str]
+    attributes_number: dict[str, np.float64]
+    attributes_bool: dict[str, bool]
+    resource_json: dict[str, str]
+    event_name: str
+
+    resource: List[AuditResource]
+    tag_attributes: List[AuditTagAttributes]
+    resource_keys: List[AuditResourceOrAttributeKeys]
+    attribute_keys: List[AuditResourceOrAttributeKeys]
+
+    def __init__(
+        self,
+        timestamp: Optional[datetime.datetime] = None,
+        resources: dict[str, Any] = {},
+        attributes: dict[str, Any] = {},
+        body: str = "",
+        event_name: str = "",
+        severity_text: str = "INFO",
+        trace_id: str = "",
+        span_id: str = "",
+        trace_flags: np.uint32 = 0,
+        scope_name: str = "signoz.audit",
+        scope_version: str = "",
+    ) -> None:
+        if timestamp is None:
+            timestamp = datetime.datetime.now()
+        self.tag_attributes = []
+        self.attribute_keys = []
+        self.resource_keys = []
+
+        self.timestamp = np.uint64(int(timestamp.timestamp() * 1e9))
+        self.observed_timestamp = self.timestamp
+
+        minute = timestamp.minute
+        bucket_minute = 0 if minute < 30 else 30
+        bucket_start = timestamp.replace(minute=bucket_minute, second=0, microsecond=0)
+        self.ts_bucket_start = np.uint64(int(bucket_start.timestamp()))
+
+        self.id = str(KsuidMs(datetime=timestamp))
+
+        self.trace_id = trace_id
+        self.span_id = span_id
+        self.trace_flags = trace_flags
+
+        self.severity_text = severity_text
+        self.severity_number = np.uint8(9 if severity_text == "INFO" else 17)
+
+        self.body = body
+        self.event_name = event_name
+
+        # Resources — JSON column only (no resources_string in audit DDL)
+        self.resource_json = {k: str(v) for k, v in resources.items()}
+        for k, v in self.resource_json.items():
+            self.tag_attributes.append(
+                AuditTagAttributes(
+                    timestamp=timestamp,
+                    tag_key=k,
+                    tag_type="resource",
+                    tag_data_type="string",
+                    string_value=str(v),
+                    int64_value=None,
+                    float64_value=None,
+                )
+            )
+            self.resource_keys.append(
+                AuditResourceOrAttributeKeys(name=k, datatype="string")
+            )
+
+        self.resource_fingerprint = LogsOrTracesFingerprint(
+            self.resource_json
+        ).calculate()
+
+        # Process attributes by type
+        self.attributes_string = {}
+        self.attributes_number = {}
+        self.attributes_bool = {}
+
+        for k, v in attributes.items():
+            if isinstance(v, bool):
+                self.attributes_bool[k] = v
+                self.tag_attributes.append(
+                    AuditTagAttributes(
+                        timestamp=timestamp,
+                        tag_key=k,
+                        tag_type="tag",
+                        tag_data_type="bool",
+                        string_value=None,
+                        int64_value=None,
+                        float64_value=None,
+                    )
+                )
+                self.attribute_keys.append(
+                    AuditResourceOrAttributeKeys(name=k, datatype="bool")
+                )
+            elif isinstance(v, int):
+                self.attributes_number[k] = np.float64(v)
+                self.tag_attributes.append(
+                    AuditTagAttributes(
+                        timestamp=timestamp,
+                        tag_key=k,
+                        tag_type="tag",
+                        tag_data_type="int64",
+                        string_value=None,
+                        int64_value=np.int64(v),
+                        float64_value=None,
+                    )
+                )
+                self.attribute_keys.append(
+                    AuditResourceOrAttributeKeys(name=k, datatype="int64")
+                )
+            elif isinstance(v, float):
+                self.attributes_number[k] = np.float64(v)
+                self.tag_attributes.append(
+                    AuditTagAttributes(
+                        timestamp=timestamp,
+                        tag_key=k,
+                        tag_type="tag",
+                        tag_data_type="float64",
+                        string_value=None,
+                        int64_value=None,
+                        float64_value=np.float64(v),
+                    )
+                )
+                self.attribute_keys.append(
+                    AuditResourceOrAttributeKeys(name=k, datatype="float64")
+                )
+            else:
+                self.attributes_string[k] = str(v)
+                self.tag_attributes.append(
+                    AuditTagAttributes(
+                        timestamp=timestamp,
+                        tag_key=k,
+                        tag_type="tag",
+                        tag_data_type="string",
+                        string_value=str(v),
+                        int64_value=None,
+                        float64_value=None,
+                    )
+                )
+                self.attribute_keys.append(
+                    AuditResourceOrAttributeKeys(name=k, datatype="string")
+                )
+
+        self.scope_name = scope_name
+        self.scope_version = scope_version
+        self.scope_string = {}
+
+        self.resource = [
+            AuditResource(
+                labels=self.resource_json,
+                fingerprint=self.resource_fingerprint,
+                seen_at_ts_bucket_start=self.ts_bucket_start,
+            )
+        ]
+
+    def np_arr(self) -> np.array:
+        return np.array(
+            [
+                self.ts_bucket_start,
+                self.resource_fingerprint,
+                self.timestamp,
+                self.observed_timestamp,
+                self.id,
+                self.trace_id,
+                self.span_id,
+                self.trace_flags,
+                self.severity_text,
+                self.severity_number,
+                self.body,
+                self.scope_name,
+                self.scope_version,
+                self.scope_string,
+                self.attributes_string,
+                self.attributes_number,
+                self.attributes_bool,
+                self.resource_json,
+                self.event_name,
+            ]
+        )
+
+
+@pytest.fixture(name="insert_audit_logs", scope="function")
+def insert_audit_logs(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[Callable[[List[AuditLog]], None], Any, None]:
+    def _insert_audit_logs(logs: List[AuditLog]) -> None:
+        resources: List[AuditResource] = []
+        for log in logs:
+            resources.extend(log.resource)
+
+        if len(resources) > 0:
+            clickhouse.conn.insert(
+                database="signoz_audit",
+                table="distributed_logs_resource",
+                data=[resource.np_arr() for resource in resources],
+                column_names=[
+                    "labels",
+                    "fingerprint",
+                    "seen_at_ts_bucket_start",
+                ],
+            )
+
+        tag_attributes: List[AuditTagAttributes] = []
+        for log in logs:
+            tag_attributes.extend(log.tag_attributes)
+
+        if len(tag_attributes) > 0:
+            clickhouse.conn.insert(
+                database="signoz_audit",
+                table="distributed_tag_attributes",
+                data=[ta.np_arr() for ta in tag_attributes],
+                column_names=[
+                    "unix_milli",
+                    "tag_key",
+                    "tag_type",
+                    "tag_data_type",
+                    "string_value",
+                    "int64_value",
+                    "float64_value",
+                ],
+            )
+
+        attribute_keys: List[AuditResourceOrAttributeKeys] = []
+        for log in logs:
+            attribute_keys.extend(log.attribute_keys)
+
+        if len(attribute_keys) > 0:
+            clickhouse.conn.insert(
+                database="signoz_audit",
+                table="distributed_logs_attribute_keys",
+                data=[ak.np_arr() for ak in attribute_keys],
+                column_names=["name", "datatype"],
+            )
+
+        resource_keys: List[AuditResourceOrAttributeKeys] = []
+        for log in logs:
+            resource_keys.extend(log.resource_keys)
+
+        if len(resource_keys) > 0:
+            clickhouse.conn.insert(
+                database="signoz_audit",
+                table="distributed_logs_resource_keys",
+                data=[rk.np_arr() for rk in resource_keys],
+                column_names=["name", "datatype"],
+            )
+
+        clickhouse.conn.insert(
+            database="signoz_audit",
+            table="distributed_logs",
+            data=[log.np_arr() for log in logs],
+            column_names=[
+                "ts_bucket_start",
+                "resource_fingerprint",
+                "timestamp",
+                "observed_timestamp",
+                "id",
+                "trace_id",
+                "span_id",
+                "trace_flags",
+                "severity_text",
+                "severity_number",
+                "body",
+                "scope_name",
+                "scope_version",
+                "scope_string",
+                "attributes_string",
+                "attributes_number",
+                "attributes_bool",
+                "resource",
+                "event_name",
+            ],
+        )
+
+    yield _insert_audit_logs
+
+    cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
+    for table in [
+        "logs",
+        "logs_resource",
+        "tag_attributes",
+        "logs_attribute_keys",
+        "logs_resource_keys",
+    ]:
+        clickhouse.conn.query(
+            f"TRUNCATE TABLE signoz_audit.{table} ON CLUSTER '{cluster}' SYNC"
+        )

tests/integration/fixtures/querier.py

@@ -38,6 +38,7 @@ class OrderBy:
 class BuilderQuery:
     signal: str
     name: str = "A"
+    source: Optional[str] = None
     limit: Optional[int] = None
     filter_expression: Optional[str] = None
     select_fields: Optional[List[TelemetryFieldKey]] = None
@@ -48,6 +49,8 @@ class BuilderQuery:
             "signal": self.signal,
             "name": self.name,
         }
+        if self.source:
+            spec["source"] = self.source
         if self.limit is not None:
             spec["limit"] = self.limit
         if self.filter_expression:
@@ -55,7 +58,9 @@ class BuilderQuery:
         if self.select_fields:
             spec["selectFields"] = [f.to_dict() for f in self.select_fields]
         if self.order:
-            spec["order"] = [o.to_dict() for o in self.order]
+            spec["order"] = [
+                o.to_dict() if hasattr(o, "to_dict") else o for o in self.order
+            ]
         return {"type": "builder_query", "spec": spec}
 
 
@@ -76,7 +81,9 @@ class TraceOperatorQuery:
         if self.limit is not None:
             spec["limit"] = self.limit
         if self.order:
-            spec["order"] = [o.to_dict() for o in self.order]
+            spec["order"] = [
+                o.to_dict() if hasattr(o, "to_dict") else o for o in self.order
+            ]
         return {"type": "builder_trace_operator", "spec": spec}
 
 
@@ -442,6 +449,7 @@ def build_scalar_query(
     signal: str,
     aggregations: List[Dict],
     *,
+    source: Optional[str] = None,
     group_by: Optional[List[Dict]] = None,
     order: Optional[List[Dict]] = None,
     limit: Optional[int] = None,
@@ -458,6 +466,9 @@ def build_scalar_query(
         "aggregations": aggregations,
     }
 
+    if source:
+        spec["source"] = source
+
     if group_by:
         spec["groupBy"] = group_by
 

tests/integration/src/auditquerier/01_audit_logs.py

@@ -0,0 +1,441 @@
+from datetime import datetime, timedelta, timezone
+from http import HTTPStatus
+from typing import Callable, List
+
+import pytest
+
+from fixtures import types
+from fixtures.audit import AuditLog
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import (
+    BuilderQuery,
+    build_logs_aggregation,
+    build_order_by,
+    build_scalar_query,
+    make_query_request,
+)
+
+
+def test_audit_list_all(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_audit_logs: Callable[[List[AuditLog]], None],
+) -> None:
+    """List audit events across multiple resource types — verify count, ordering, and fields."""
+    now = datetime.now(tz=timezone.utc)
+    insert_audit_logs(
+        [
+            AuditLog(
+                timestamp=now - timedelta(seconds=3),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "alert-rule",
+                    "signoz.audit.resource.id": "alert-001",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-010",
+                    "signoz.audit.principal.email": "ops@acme.com",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "create",
+                    "signoz.audit.outcome": "success",
+                },
+                body="ops@acme.com (user-010) created alert-rule (alert-001)",
+                event_name="alert-rule.created",
+                severity_text="INFO",
+            ),
+            AuditLog(
+                timestamp=now - timedelta(seconds=2),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "saved-view",
+                    "signoz.audit.resource.id": "view-001",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-010",
+                    "signoz.audit.principal.email": "ops@acme.com",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "update",
+                    "signoz.audit.outcome": "success",
+                },
+                body="ops@acme.com (user-010) updated saved-view (view-001)",
+                event_name="saved-view.updated",
+                severity_text="INFO",
+            ),
+            AuditLog(
+                timestamp=now - timedelta(seconds=1),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "user",
+                    "signoz.audit.resource.id": "user-020",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-010",
+                    "signoz.audit.principal.email": "ops@acme.com",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "update",
+                    "signoz.audit.action_category": "access_control",
+                    "signoz.audit.outcome": "success",
+                },
+                body="ops@acme.com (user-010) updated user (user-020)",
+                event_name="user.role.changed",
+                severity_text="INFO",
+            ),
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    now = datetime.now(tz=timezone.utc)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        queries=[
+            BuilderQuery(
+                signal="logs",
+                source="audit",
+                limit=100,
+                order=[build_order_by("timestamp"), build_order_by("id")],
+            ).to_dict()
+        ],
+        request_type="raw",
+    )
+
+    assert response.status_code == HTTPStatus.OK
+    assert response.json()["status"] == "success"
+
+    rows = response.json()["data"]["data"]["results"][0]["rows"]
+    assert len(rows) == 3
+
+    # Most recent first
+    assert rows[0]["data"]["event_name"] == "user.role.changed"
+    assert rows[1]["data"]["event_name"] == "saved-view.updated"
+    assert rows[2]["data"]["event_name"] == "alert-rule.created"
+
+    # Verify event_name and body are present
+    assert rows[0]["data"]["body"] == "ops@acme.com (user-010) updated user (user-020)"
+    assert rows[0]["data"]["severity_text"] == "INFO"
+
+
+@pytest.mark.parametrize(
+    "filter_expression,expected_count,expected_event_names",
+    [
+        pytest.param(
+            "signoz.audit.principal.id = 'user-001'",
+            3,
+            {"session.login", "dashboard.updated", "dashboard.created"},
+            id="filter_by_principal_id",
+        ),
+        pytest.param(
+            "signoz.audit.outcome = 'failure'",
+            1,
+            {"dashboard.deleted"},
+            id="filter_by_outcome_failure",
+        ),
+        pytest.param(
+            "signoz.audit.resource.kind = 'dashboard'"
+            " AND signoz.audit.resource.id = 'dash-001'",
+            3,
+            {"dashboard.deleted", "dashboard.updated", "dashboard.created"},
+            id="filter_by_resource_kind_and_id",
+        ),
+        pytest.param(
+            "signoz.audit.principal.type = 'service_account'",
+            1,
+            {"serviceaccount.apikey.created"},
+            id="filter_by_principal_type",
+        ),
+        pytest.param(
+            "signoz.audit.resource.kind = 'dashboard'"
+            " AND signoz.audit.action = 'delete'",
+            1,
+            {"dashboard.deleted"},
+            id="filter_by_resource_kind_and_action",
+        ),
+    ],
+)
+def test_audit_filter(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_audit_logs: Callable[[List[AuditLog]], None],
+    filter_expression: str,
+    expected_count: int,
+    expected_event_names: set,
+) -> None:
+    """Parametrized audit filter tests covering the documented query patterns."""
+    now = datetime.now(tz=timezone.utc)
+    insert_audit_logs(
+        [
+            AuditLog(
+                timestamp=now - timedelta(seconds=5),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "dashboard",
+                    "signoz.audit.resource.id": "dash-001",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-001",
+                    "signoz.audit.principal.email": "alice@acme.com",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "create",
+                    "signoz.audit.action_category": "configuration_change",
+                    "signoz.audit.outcome": "success",
+                },
+                body="alice@acme.com created dashboard",
+                event_name="dashboard.created",
+            ),
+            AuditLog(
+                timestamp=now - timedelta(seconds=4),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "dashboard",
+                    "signoz.audit.resource.id": "dash-001",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-001",
+                    "signoz.audit.principal.email": "alice@acme.com",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "update",
+                    "signoz.audit.action_category": "configuration_change",
+                    "signoz.audit.outcome": "success",
+                },
+                body="alice@acme.com updated dashboard",
+                event_name="dashboard.updated",
+            ),
+            AuditLog(
+                timestamp=now - timedelta(seconds=3),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "dashboard",
+                    "signoz.audit.resource.id": "dash-001",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-002",
+                    "signoz.audit.principal.email": "viewer@acme.com",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "delete",
+                    "signoz.audit.action_category": "configuration_change",
+                    "signoz.audit.outcome": "failure",
+                    "signoz.audit.error.type": "forbidden",
+                    "signoz.audit.error.code": "authz_forbidden",
+                },
+                body="viewer@acme.com failed to delete dashboard",
+                event_name="dashboard.deleted",
+                severity_text="ERROR",
+            ),
+            AuditLog(
+                timestamp=now - timedelta(seconds=2),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "serviceaccount",
+                    "signoz.audit.resource.id": "sa-001",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "sa-001",
+                    "signoz.audit.principal.email": "",
+                    "signoz.audit.principal.type": "service_account",
+                    "signoz.audit.action": "create",
+                    "signoz.audit.action_category": "access_control",
+                    "signoz.audit.outcome": "success",
+                },
+                body="sa-001 created serviceaccount",
+                event_name="serviceaccount.apikey.created",
+            ),
+            AuditLog(
+                timestamp=now - timedelta(seconds=1),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "session",
+                    "signoz.audit.resource.id": "*",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-001",
+                    "signoz.audit.principal.email": "alice@acme.com",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "login",
+                    "signoz.audit.action_category": "access_control",
+                    "signoz.audit.outcome": "success",
+                },
+                body="alice@acme.com login session",
+                event_name="session.login",
+            ),
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    now = datetime.now(tz=timezone.utc)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        queries=[
+            BuilderQuery(
+                signal="logs",
+                source="audit",
+                limit=100,
+                filter_expression=filter_expression,
+                order=[build_order_by("timestamp"), build_order_by("id")],
+            ).to_dict()
+        ],
+        request_type="raw",
+    )
+
+    assert response.status_code == HTTPStatus.OK
+
+    rows = response.json()["data"]["data"]["results"][0]["rows"]
+    assert len(rows) == expected_count
+
+    actual_event_names = {row["data"]["event_name"] for row in rows}
+    assert actual_event_names == expected_event_names
+
+
+def test_audit_scalar_count_failures(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_audit_logs: Callable[[List[AuditLog]], None],
+) -> None:
+    """Alert query — count multiple failures from different principals."""
+    now = datetime.now(tz=timezone.utc)
+    insert_audit_logs(
+        [
+            AuditLog(
+                timestamp=now - timedelta(seconds=3),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "dashboard",
+                    "signoz.audit.resource.id": "dash-100",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-050",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "delete",
+                    "signoz.audit.outcome": "failure",
+                },
+                body="user-050 failed to delete dashboard",
+                event_name="dashboard.deleted",
+                severity_text="ERROR",
+            ),
+            AuditLog(
+                timestamp=now - timedelta(seconds=2),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "alert-rule",
+                    "signoz.audit.resource.id": "alert-200",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-060",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "update",
+                    "signoz.audit.outcome": "failure",
+                },
+                body="user-060 failed to update alert-rule",
+                event_name="alert-rule.updated",
+                severity_text="ERROR",
+            ),
+            AuditLog(
+                timestamp=now - timedelta(seconds=1),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "dashboard",
+                    "signoz.audit.resource.id": "dash-100",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-050",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "update",
+                    "signoz.audit.outcome": "success",
+                },
+                body="user-050 updated dashboard",
+                event_name="dashboard.updated",
+            ),
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    now = datetime.now(tz=timezone.utc)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        queries=[
+            build_scalar_query(
+                name="A",
+                signal="logs",
+                source="audit",
+                aggregations=[build_logs_aggregation("count()")],
+                filter_expression="signoz.audit.outcome = 'failure'",
+            )
+        ],
+        request_type="scalar",
+    )
+
+    assert response.status_code == HTTPStatus.OK
+    assert response.json()["status"] == "success"
+
+    scalar_data = response.json()["data"]["data"]["results"][0].get("data", [])
+    assert len(scalar_data) == 1
+    assert scalar_data[0][0] == 2
+
+
+def test_audit_does_not_leak_into_logs(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_audit_logs: Callable[[List[AuditLog]], None],
+) -> None:
+    """A single audit event in signoz_audit must not appear in regular log queries."""
+    now = datetime.now(tz=timezone.utc)
+    insert_audit_logs(
+        [
+            AuditLog(
+                timestamp=now - timedelta(seconds=1),
+                resources={
+                    "service.name": "signoz",
+                    "signoz.audit.resource.kind": "organization",
+                    "signoz.audit.resource.id": "org-999",
+                },
+                attributes={
+                    "signoz.audit.principal.id": "user-admin",
+                    "signoz.audit.principal.type": "user",
+                    "signoz.audit.action": "update",
+                    "signoz.audit.outcome": "success",
+                },
+                body="user-admin updated organization (org-999)",
+                event_name="organization.updated",
+            ),
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    now = datetime.now(tz=timezone.utc)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        queries=[
+            BuilderQuery(
+                signal="logs",
+                limit=100,
+                order=[build_order_by("timestamp"), build_order_by("id")],
+            ).to_dict()
+        ],
+        request_type="raw",
+    )
+
+    assert response.status_code == HTTPStatus.OK
+
+    rows = response.json()["data"]["data"]["results"][0].get("rows") or []
+
+    audit_bodies = [
+        row["data"]["body"]
+        for row in rows
+        if "signoz.audit"
+        in row["data"].get("attributes_string", {}).get("signoz.audit.action", "")
+    ]
+    assert len(audit_bodies) == 0