test(wal): log the leaky queries

* feat(audit): add telemetry audit query infrastructure

Add pkg/telemetryaudit/ with tables, field mapper, condition builder,
and statement builder for querying audit logs from signoz_audit database.
Add SourceAudit to source enum and integrate audit key resolution
into the metadata store.

* chore: address review comments

Comment out SourceAudit from Enum() until frontend is ready.
Use actual audit table constants in metadata test helpers.

* fix(audit): align field mapper with actual audit DDL schema

Remove resources_string (not in audit table DDL).
Add event_name as intrinsic column.
Resource context resolves only through the resource JSON column.

* feat(audit): add audit field value autocomplete support

Wire distributed_tag_attributes_v2 for signoz_audit into the
metadata store. Add getAuditFieldValues() and route SignalLogs +
SourceAudit to it in GetFieldValues().

* test(audit): add statement builder tests

Cover all three request types (list, time series, scalar) with
audit-specific query patterns: materialized column filters, AND/OR
conditions, limit CTEs, and group-by expressions.

* refactor(audit): inline field key map into test file

Remove test_data.go and inline the audit field key map directly
into statement_builder_test.go with a compact helper function.

* style(audit): move column map to const.go, use sqlbuilder.As in metadata

Move logsV2Columns from field_mapper.go to const.go to colocate all
column definitions. Switch getAuditKeys() to use sb.As() instead of
raw string formatting. Fix FieldContext alignment.

* fix(audit): align table names with schema migration

Migration uses logs/distributed_logs (not logs_v2/distributed_logs_v2).
Rename LogsV2TableName to LogsTableName and LogsV2LocalTableName to
LogsLocalTableName to match the actual signoz_audit DDL.

* feat(audit): add integration test fixture for audit logs

AuditLog fixture inserts into all 5 signoz_audit tables matching
the schema migration DDL: distributed_logs (no resources_string,
has event_name), distributed_logs_resource, distributed_tag_attributes_v2,
distributed_logs_attribute_keys, distributed_logs_resource_keys.

* fix(audit): rename tag_attributes_v2 to tag_attributes

Migration uses tag_attributes/distributed_tag_attributes (no _v2
suffix). Rename constants and update all references including the
integration test fixture.

* feat(audit): wire audit statement builder into querier

Add auditStmtBuilder to querier struct and route LogAggregation
queries with source=audit to it in all three dispatch locations
(main query, live tail, shiftedQuery). Create and wire the full
audit query stack in signozquerier provider.

* test(audit): add integration tests for audit log querying

Cover the documented query patterns: list all events, filter by
principal ID, filter by outcome, filter by resource name+ID,
filter by principal type, scalar count for alerting, and
isolation test ensuring audit data doesn't leak into regular logs.

* fix(audit): revert sb.As in getAuditKeys, fix fixture column_names

Revert getAuditKeys to use raw SQL strings instead of sb.As() which
incorrectly treated string literals as column references. Add explicit
column_names to all ClickHouse insert calls in the audit fixture.

* fix(audit): remove debug assertion from integration test

* feat(audit): internalize resource filter in audit statement builder

Build the resource filter internally pointing at
signoz_audit.distributed_logs_resource. Add LogsResourceTableName
constant. Remove resourceFilterStmtBuilder from constructor params.
Update test expectations to use the audit resource table.

* fix(audit): rename resource.name to resource.kind, move to resource attributes

Align with schema change from SigNoz/signoz#10826:
- signoz.audit.resource.name renamed to signoz.audit.resource.kind
- resource.kind and resource.id moved from event attributes to OTel
  Resource attributes (resource JSON column)
- Materialized columns reduced from 7 to 5 (resource.kind and
  resource.id no longer materialized)

* refactor(audit): use pytest.mark.parametrize for filter integration tests

Consolidate filter test functions into a single parametrized test.
6/8 tests passing; resource kind+ID filter and scalar count need
further investigation (resource filter JSON key extraction with
dotted keys, scalar response format).

* fix(audit): add source to resource filter for correct metadata routing

Add source param to telemetryresourcefilter.New so the resource
filter's key selectors include Source when calling GetKeysMulti.
Without this, audit resource keys route to signoz_logs metadata
tables instead of signoz_audit. Fix scalar test to use table
response format (columns+data, not rows).

* refactor(audit): reuse querier fixtures in integration tests

Add source param to BuilderQuery and build_scalar_query in the
querier fixture. Replace custom _build_audit_query and
_build_audit_ts_query helpers with BuilderQuery and
build_scalar_query from the shared fixtures.

* refactor(audit): remove wrapper helpers, inline make_query_request calls

Remove _query_audit_raw and _query_audit_scalar helpers. Use
make_query_request, BuilderQuery, and build_scalar_query directly.
Compute time window at test execution time via _time_window() to
avoid stale module-level timestamps.

* refactor(audit): inline _time_window into test functions

* style(audit): use snake_case for pytest parametrize IDs

* refactor(audit): inline DEFAULT_ORDER using build_order_by

Use build_order_by from querier fixtures instead of OrderBy/
TelemetryFieldKey dataclasses. Allow BuilderQuery.order to accept
plain dicts alongside OrderBy objects.

* refactor(audit): inline all data setup, use distinct scenarios per test

Remove _insert_standard_audit_events helper. Each test now owns its
data: list_all uses alert-rule/saved-view/user resource types,
scalar_count uses multiple failures from different principals (count=2),
leak test uses a single organization event. Parametrized filter tests
keep the original 5-event dataset.

* fix(audit): remove silent empty-string guards in metadata store

Remove guards that silently returned nil/empty when audit DB params
were empty. All call sites now pass real constants, so misconfiguration
should fail loudly rather than produce silent empty results.

* style(audit): remove module docstring from integration test

* style: formatting fix in tables file

* style: formatting fix in tables file

* fix: add auditStmtBuilder nil param to querier_test.go

* fix: fix fmt

docs/api/openapi.yml

@@ -2787,147 +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:
-          format: int32
-          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:
@@ -9645,76 +9504,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

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

@@ -3196,249 +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
-	 * @format int32
-	 */
-	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
@@ -5100,17 +4857,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);
-};

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/modules/tracedetail/impltracedetail/handler.go

@@ -1,61 +0,0 @@
-package impltracedetail
-
-import (
-	"context"
-	"net/http"
-	"time"
-
-	"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) {
-	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
-	defer cancel()
-
-	claims, err := authtypes.ClaimsFromContext(ctx)
-	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(ctx, 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/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/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

@@ -286,7 +286,6 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
 			handlers.RegistryHandler,
 			handlers.CloudIntegrationHandler,
 			handlers.RuleStateHistory,
-			handlers.TraceDetail,
 		),
 	)
 }

pkg/signoz/signoz.go

@@ -33,6 +33,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"
@@ -395,6 +396,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,

pkg/sqlstore/sqlitesqlstore/provider.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"log/slog"
 	"net/url"
+	"os"
+	"time"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
@@ -23,6 +25,7 @@ type provider struct {
 	bundb     *sqlstore.BunDB
 	dialect   *dialect
 	formatter sqlstore.SQLFormatter
+	done      chan struct{}
 }
 
 func NewFactory(hookFactories ...factory.ProviderFactory[sqlstore.SQLStoreHook, sqlstore.Config]) factory.ProviderFactory[sqlstore.SQLStore, sqlstore.Config] {
@@ -59,13 +62,19 @@ func New(ctx context.Context, providerSettings factory.ProviderSettings, config
 
 	sqliteDialect := sqlitedialect.New()
 	bunDB := sqlstore.NewBunDB(settings, sqldb, sqliteDialect, hooks)
-	return &provider{
+
+	done := make(chan struct{})
+	p := &provider{
 		settings:  settings,
 		sqldb:     sqldb,
 		bundb:     bunDB,
 		dialect:   new(dialect),
 		formatter: newFormatter(bunDB.Dialect()),
-	}, nil
+		done:      done,
+	}
+	go p.walDiagnosticLoop(config.Sqlite.Path)
+
+	return p, nil
 }
 
 func (provider *provider) BunDB() *bun.DB {
@@ -109,3 +118,73 @@ func (provider *provider) WrapAlreadyExistsErrf(err error, code errors.Code, for
 
 	return err
 }
+
+// walDiagnosticLoop periodically logs pool stats, WAL file size, and busy prepared statements
+// to help diagnose WAL checkpoint failures caused by permanent read locks.
+func (provider *provider) walDiagnosticLoop(dbPath string) {
+	ticker := time.NewTicker(60 * time.Second)
+	defer ticker.Stop()
+
+	logger := provider.settings.Logger()
+	walPath := dbPath + "-wal"
+
+	for {
+		select {
+		case <-provider.done:
+			return
+		case <-ticker.C:
+			// 1. Log pool stats (no SQL needed)
+			stats := provider.sqldb.Stats()
+			logger.Info("sqlite_pool_stats",
+				slog.Int("max_open", stats.MaxOpenConnections),
+				slog.Int("open", stats.OpenConnections),
+				slog.Int("in_use", stats.InUse),
+				slog.Int("idle", stats.Idle),
+				slog.Int64("wait_count", stats.WaitCount),
+				slog.String("wait_duration", stats.WaitDuration.String()),
+				slog.Int64("max_idle_closed", stats.MaxIdleClosed),
+				slog.Int64("max_idle_time_closed", stats.MaxIdleTimeClosed),
+				slog.Int64("max_lifetime_closed", stats.MaxLifetimeClosed),
+			)
+
+			// 2. Log WAL file size (no SQL needed)
+			if info, err := os.Stat(walPath); err == nil {
+				logger.Info("sqlite_wal_size",
+					slog.Int64("bytes", info.Size()),
+					slog.String("path", walPath),
+				)
+			}
+
+			// 3. Check for busy prepared statements on a single pool connection
+			provider.checkBusyStatements(logger)
+		}
+	}
+}
+
+func (provider *provider) checkBusyStatements(logger *slog.Logger) {
+	conn, err := provider.sqldb.Conn(context.Background())
+	if err != nil {
+		logger.Warn("sqlite_diag_conn_error", slog.String("error", err.Error()))
+		return
+	}
+	defer conn.Close()
+
+	rows, err := conn.QueryContext(context.Background(), "SELECT sql FROM sqlite_stmt WHERE busy")
+	if err != nil {
+		logger.Warn("sqlite_diag_query_error", slog.String("error", err.Error()))
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var stmtSQL string
+		if err := rows.Scan(&stmtSQL); err != nil {
+			logger.Warn("sqlite_diag_scan_error", slog.String("error", err.Error()))
+			continue
+		}
+		logger.Warn("leaked_busy_statement", slog.String("sql", stmtSQL))
+	}
+	if err := rows.Err(); err != nil {
+		logger.Warn("sqlite_diag_rows_error", slog.String("error", err.Error()))
+	}
+}

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/statement_builder.go

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

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,
@@ -592,6 +609,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 +1127,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 +1180,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 +1188,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 +1212,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 +1230,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 +1592,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 +1973,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

@@ -9,6 +9,6 @@ const (
 	ColumnEvolutionMetadataTableName = "distributed_column_evolution_metadata"
 	PathTypesTableName               = otelcollectorconst.DistributedPathTypesTable
 	// 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/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/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,6 +58,7 @@ 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"`
@@ -66,7 +67,7 @@ type WaterfallSpan struct {
 	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:"-"`
 }
 
@@ -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