refactor: making querier changes for audit schema

pkg/querier/builder_query.go

@@ -488,7 +488,7 @@ func (q *builderQuery[T]) executeWithContext(ctx context.Context, query string,
 		kind = qbtypes.RequestTypeTimeSeries
 	}
 
-	payload, err := consume(rows, kind, queryWindow, q.spec.StepInterval, q.spec.Name)
+	payload, err := consume(rows, kind, queryWindow, q.spec.StepInterval, q.spec.Name, flatJSONColumns(q.spec.Signal, q.spec.Source))
 	if err != nil {
 		return nil, err
 	}

pkg/querier/clickhouse_query.go

@@ -147,7 +147,7 @@ func (q *chSQLQuery) Execute(ctx context.Context) (*qbtypes.Result, error) {
 	}
 	defer rows.Close()
 
-	payload, err := consume(rows, q.kind, nil, qbtypes.Step{}, q.query.Name)
+	payload, err := consume(rows, q.kind, nil, qbtypes.Step{}, q.query.Name, defaultFlatJSONColumns)
 	if err != nil {
 		return nil, err
 	}

pkg/querier/consume.go

@@ -32,8 +32,18 @@ var (
 	// written clickhouse query. The column alias indcate which value is
 	// to be considered as final result (or target).
 	legacyReservedColumnTargetAliases = []string{"__result", "__value", "result", "res", "value"}
+	defaultFlatJSONColumns            = []string{"attributes", "scope"}
+	// auditFlatJSONColumns adds resource, which only audit returns as a JSON column.
+	auditFlatJSONColumns = []string{"attributes", "scope", "resource"}
 )
 
+func flatJSONColumns(signal telemetrytypes.Signal, source telemetrytypes.Source) []string {
+	if signal == telemetrytypes.SignalLogs && source == telemetrytypes.SourceAudit {
+		return auditFlatJSONColumns
+	}
+	return defaultFlatJSONColumns
+}
+
 // stripKeyAlias removes the __SELECT_KEY_<n>_ / __GROUP_BY_KEY_<n>_ prefix from a result
 // column name, recovering the field name; unprefixed names are returned unchanged.
 func stripKeyAlias(name string) string {
@@ -45,13 +55,12 @@ func stripKeyAlias(name string) string {
 // bag decodes flat so its dotted keys merge with the legacy attribute maps and a key stored at both a
 // scalar and an object path stays two keys; the scope decodes flat too, keeping its attributes dotted;
 // every other JSON column decodes nested.
-func unwrapVariant(name string, val any) any {
+func unwrapVariant(name string, val any, flatColumns []string) any {
 	switch v := val.(type) {
 	case chcol.Variant:
 		return v.Any()
 	case telemetrystoretypes.JSONValue:
-		switch name {
-		case "attributes", "scope":
+		if slices.Contains(flatColumns, name) {
 			return v.Flatten()
 		}
 		return v.Nest()
@@ -64,7 +73,7 @@ func unwrapVariant(name string, val any) any {
 // series. JSON goes through encoding/json for its sorted map keys: ClickHouse groups documents by
 // structure, so two rows it considers equal have to produce the same label.
 func labelValue(val any) string {
-	val = unwrapVariant("", val)
+	val = unwrapVariant("", val, nil)
 	if val == nil {
 		return ""
 	}
@@ -83,7 +92,7 @@ func labelValue(val any) string {
 // * Scalar      - *qbtypes.ScalarData
 // * Raw         - *qbtypes.RawData
 // * Distribution- *qbtypes.DistributionData.
-func consume(rows driver.Rows, kind qbtypes.RequestType, queryWindow *qbtypes.TimeRange, step qbtypes.Step, queryName string) (any, error) {
+func consume(rows driver.Rows, kind qbtypes.RequestType, queryWindow *qbtypes.TimeRange, step qbtypes.Step, queryName string, flatColumns []string) (any, error) {
 	var (
 		payload any
 		err     error
@@ -97,7 +106,7 @@ func consume(rows driver.Rows, kind qbtypes.RequestType, queryWindow *qbtypes.Ti
 	case qbtypes.RequestTypeHeatmap:
 		payload, err = readAsHeatmap(rows, queryWindow, step, queryName)
 	case qbtypes.RequestTypeRaw, qbtypes.RequestTypeTrace, qbtypes.RequestTypeRawStream:
-		payload, err = readAsRaw(rows, queryName)
+		payload, err = readAsRaw(rows, queryName, flatColumns)
 		// TODO: add support for other request types
 	}
 
@@ -488,7 +497,7 @@ func readAsScalar(rows driver.Rows, queryName string) (*qbtypes.ScalarData, erro
 		// 2. deref each slot into the output row
 		row := make([]any, len(scan))
 		for i, cell := range scan {
-			row[i] = unwrapVariant(stripKeyAlias(colNames[i]), derefValue(cell))
+			row[i] = unwrapVariant(stripKeyAlias(colNames[i]), derefValue(cell), defaultFlatJSONColumns)
 		}
 		data = append(data, row)
 	}
@@ -520,7 +529,7 @@ func derefValue(v any) any {
 	return val.Interface()
 }
 
-func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
+func readAsRaw(rows driver.Rows, queryName string, flatColumns []string) (*qbtypes.RawData, error) {
 	colNames := rows.Columns()
 	colTypes := rows.ColumnTypes()
 	colCnt := len(colNames)
@@ -546,7 +555,7 @@ func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
 			name := stripKeyAlias(colNames[i])
 
 			// de-reference the typed pointer to any
-			val := unwrapVariant(name, reflect.ValueOf(cellPtr).Elem().Interface())
+			val := unwrapVariant(name, reflect.ValueOf(cellPtr).Elem().Interface(), flatColumns)
 
 			// special-case: timestamp column
 			if name == "timestamp" || name == "timestamp_datetime" {

pkg/querier/consume_test.go

@@ -98,24 +98,31 @@ func TestUnwrapVariant(t *testing.T) {
 	scopeAttrs := chcol.NewJSON()
 	scopeAttrs.SetValueAtPath("lib.lang", "go")
 	scopeCol := jsonColumn(map[string]any{"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": scopeAttrs})
+	resourceAttrs := chcol.NewJSON()
+	resourceAttrs.SetValueAtPath("service.name", "signoz")
+	resourceAttrs.SetValueAtPath("signoz.audit.resource.kind", "role")
+	resourceCol := jsonColumn(map[string]any{"attributes": resourceAttrs})
 
 	testCases := []struct {
-		name   string
-		column string
-		input  any
-		want   any
+		name        string
+		column      string
+		input       any
+		flatColumns []string
+		want        any
 	}{
-		{name: "VariantScalar", column: "", input: chcol.NewDynamicWithType("error", "String"), want: "error"},
-		{name: "EmptyDynamic", column: "", input: chcol.Dynamic{}, want: nil},
-		{name: "PlainValuePassthrough", column: "", input: uint64(3), want: uint64(3)},
-		{name: "JSONColumnNested", column: "body_v2", input: jsonCol, want: map[string]any{"level": "error", "attrs": map[string]any{"code": float64(500)}}},
-		{name: "AttributesColumnFlat", column: "attributes", input: jsonCol, want: map[string]any{"level": "error", "attrs.code": float64(500)}},
-		{name: "AttributesColumnDropsNullPaths", column: "attributes", input: nullCol, want: map[string]any{"level": "error"}},
-		{name: "ScopeColumnFlat", column: "scope", input: scopeCol, want: map[string]any{"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": map[string]any{"lib.lang": "go"}}},
+		{name: "VariantScalar", column: "", input: chcol.NewDynamicWithType("error", "String"), flatColumns: defaultFlatJSONColumns, want: "error"},
+		{name: "EmptyDynamic", column: "", input: chcol.Dynamic{}, flatColumns: defaultFlatJSONColumns, want: nil},
+		{name: "PlainValuePassthrough", column: "", input: uint64(3), flatColumns: defaultFlatJSONColumns, want: uint64(3)},
+		{name: "JSONColumnNested", column: "body_v2", input: jsonCol, flatColumns: defaultFlatJSONColumns, want: map[string]any{"level": "error", "attrs": map[string]any{"code": float64(500)}}},
+		{name: "AttributesColumnFlat", column: "attributes", input: jsonCol, flatColumns: defaultFlatJSONColumns, want: map[string]any{"level": "error", "attrs.code": float64(500)}},
+		{name: "AttributesColumnDropsNullPaths", column: "attributes", input: nullCol, flatColumns: defaultFlatJSONColumns, want: map[string]any{"level": "error"}},
+		{name: "ScopeColumnFlat", column: "scope", input: scopeCol, flatColumns: defaultFlatJSONColumns, want: map[string]any{"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": map[string]any{"lib.lang": "go"}}},
+		{name: "ResourceColumnNested_Default", column: "resource", input: resourceCol, flatColumns: defaultFlatJSONColumns, want: map[string]any{"attributes": map[string]any{"service": map[string]any{"name": "signoz"}, "signoz": map[string]any{"audit": map[string]any{"resource": map[string]any{"kind": "role"}}}}}},
+		{name: "ResourceColumnFlat_Audit", column: "resource", input: resourceCol, flatColumns: auditFlatJSONColumns, want: map[string]any{"attributes": map[string]any{"service.name": "signoz", "signoz.audit.resource.kind": "role"}}},
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			assert.Equal(t, testCase.want, unwrapVariant(testCase.column, testCase.input))
+			assert.Equal(t, testCase.want, unwrapVariant(testCase.column, testCase.input, testCase.flatColumns))
 		})
 	}
 }

pkg/querier/trace_operator_query.go

@@ -89,7 +89,7 @@ func (q *traceOperatorQuery) executeWithContext(ctx context.Context, query strin
 	queryWindow := &qbtypes.TimeRange{From: q.fromMS, To: q.toMS}
 
 	// Use the consume function like builderQuery does
-	payload, err := consume(rows, q.kind, queryWindow, q.spec.StepInterval, q.spec.Name)
+	payload, err := consume(rows, q.kind, queryWindow, q.spec.StepInterval, q.spec.Name, defaultFlatJSONColumns)
 	if err != nil {
 		return nil, err
 	}

pkg/statementbuilder/auditstatementbuilder/statement_builder.go

@@ -248,15 +248,11 @@ func (b *auditQueryStatementBuilder) buildListQuery(
 		sb.SelectMore(audittelemetryschema.TraceFlagsColumn)
 		sb.SelectMore(audittelemetryschema.SeverityTextColumn)
 		sb.SelectMore(audittelemetryschema.SeverityNumberColumn)
-		sb.SelectMore(audittelemetryschema.ScopeNameColumn)
-		sb.SelectMore(audittelemetryschema.ScopeVersionColumn)
 		sb.SelectMore(audittelemetryschema.BodyColumn)
 		sb.SelectMore(audittelemetryschema.EventNameColumn)
-		sb.SelectMore(audittelemetryschema.AttributesStringColumn)
-		sb.SelectMore(audittelemetryschema.AttributesNumberColumn)
-		sb.SelectMore(audittelemetryschema.AttributesBoolColumn)
+		sb.SelectMore(audittelemetryschema.AttributesColumn)
 		sb.SelectMore(audittelemetryschema.ResourceColumn)
-		sb.SelectMore(audittelemetryschema.ScopeStringColumn)
+		sb.SelectMore(audittelemetryschema.ScopeColumn)
 	} else {
 		for index := range query.SelectFields {
 			if query.SelectFields[index].Name == audittelemetryschema.TimestampColumn || query.SelectFields[index].Name == audittelemetryschema.IDColumn {

pkg/statementbuilder/auditstatementbuilder/statement_builder_test.go

@@ -13,17 +13,17 @@ import (
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
 	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
 	"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 {
+	key := func(name string, ctx telemetrytypes.FieldContext, dt telemetrytypes.FieldDataType) *telemetrytypes.TelemetryFieldKey {
 		return &telemetrytypes.TelemetryFieldKey{
 			Name:          name,
 			Signal:        telemetrytypes.SignalLogs,
 			FieldContext:  ctx,
 			FieldDataType: dt,
-			Materialized:  materialized,
 		}
 	}
 
@@ -33,27 +33,27 @@ func auditFieldKeyMap() map[string][]*telemetrytypes.TelemetryFieldKey {
 	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)},
+		"service.name":                 {key("service.name", res, str)},
+		"signoz.audit.action":          {key("signoz.audit.action", attr, str)},
+		"signoz.audit.outcome":         {key("signoz.audit.outcome", attr, str)},
+		"signoz.audit.principal.email": {key("signoz.audit.principal.email", attr, str)},
+		"signoz.audit.principal.id":    {key("signoz.audit.principal.id", attr, str)},
+		"signoz.audit.principal.type":  {key("signoz.audit.principal.type", attr, str)},
+		"signoz.audit.resource.kind":   {key("signoz.audit.resource.kind", res, str)},
+		"signoz.audit.resource.id":     {key("signoz.audit.resource.id", res, str)},
+		"signoz.audit.action_category": {key("signoz.audit.action_category", attr, str)},
+		"signoz.audit.error.type":      {key("signoz.audit.error.type", attr, str)},
+		"signoz.audit.error.code":      {key("signoz.audit.error.code", attr, str)},
+		"http.request.method":          {key("http.request.method", attr, str)},
+		"http.response.status_code":    {key("http.response.status_code", attr, i64)},
 	}
 }
 
-func newTestAuditStatementBuilder(t *testing.T) *auditQueryStatementBuilder {
+func newTestAuditStatementBuilder(t *testing.T, keysMap map[string][]*telemetrytypes.TelemetryFieldKey) *auditQueryStatementBuilder {
 	t.Helper()
 	fl := flaggertest.New(t)
 	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
-	mockMetadataStore.KeysMap = auditFieldKeyMap()
+	mockMetadataStore.KeysMap = keysMap
 
 	storage := audittelemetryschema.NewStorage()
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
@@ -69,7 +69,7 @@ func newTestAuditStatementBuilder(t *testing.T) *auditQueryStatementBuilder {
 }
 
 func TestStatementBuilder(t *testing.T) {
-	statementBuilder := newTestAuditStatementBuilder(t)
+	statementBuilder := newTestAuditStatementBuilder(t, auditFieldKeyMap())
 	ctx := context.Background()
 
 	testCases := []struct {
@@ -79,7 +79,7 @@ func TestStatementBuilder(t *testing.T) {
 		expected    qbtypes.Statement
 		expectedErr error
 	}{
-		// List: all actions by a specific user (materialized principal.id filter)
+		// List: all actions by a specific user
 		{
 			name:        "ListByPrincipalID",
 			requestType: qbtypes.RequestTypeRaw,
@@ -92,11 +92,11 @@ func TestStatementBuilder(t *testing.T) {
 				Limit: 100,
 			},
 			expected: qbtypes.Statement{
-				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE (`attribute_string_signoz$$audit$$principal$$id` = ? AND `attribute_string_signoz$$audit$$principal$$id_exists` = true) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE (attributes.`signoz.audit.principal.id`::String = ? AND attributes.`signoz.audit.principal.id` IS NOT NULL) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
 				Args:  []any{"019a-1234-abcd-5678", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
 			},
 		},
-		// List: all failed actions (materialized outcome filter)
+		// List: all failed actions
 		{
 			name:        "ListByOutcomeFailure",
 			requestType: qbtypes.RequestTypeRaw,
@@ -109,11 +109,11 @@ func TestStatementBuilder(t *testing.T) {
 				Limit: 100,
 			},
 			expected: qbtypes.Statement{
-				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = true) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE (attributes.`signoz.audit.outcome`::String = ? AND attributes.`signoz.audit.outcome` IS NOT NULL) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
 				Args:  []any{"failure", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
 			},
 		},
-		// List: change history of a specific dashboard (two materialized column AND)
+		// List: change history of a specific dashboard (two resource keys AND)
 		{
 			name:        "ListByResourceKindAndID",
 			requestType: qbtypes.RequestTypeRaw,
@@ -126,7 +126,7 @@ func TestStatementBuilder(t *testing.T) {
 				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 <= ? GROUP BY fingerprint) 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 timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				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 <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) 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},
 			},
 		},
@@ -143,11 +143,11 @@ func TestStatementBuilder(t *testing.T) {
 				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 <= ? GROUP BY fingerprint) 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` = true) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				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 <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (attributes.`signoz.audit.action`::String = ? AND attributes.`signoz.audit.action` IS NOT NULL) 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", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
 			},
 		},
-		// List: all actions by service accounts (materialized principal.type)
+		// List: all actions by service accounts
 		{
 			name:        "ListByPrincipalType",
 			requestType: qbtypes.RequestTypeRaw,
@@ -160,7 +160,7 @@ func TestStatementBuilder(t *testing.T) {
 				Limit: 100,
 			},
 			expected: qbtypes.Statement{
-				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE (`attribute_string_signoz$$audit$$principal$$type` = ? AND `attribute_string_signoz$$audit$$principal$$type_exists` = true) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE (attributes.`signoz.audit.principal.type`::String = ? AND attributes.`signoz.audit.principal.type` IS NOT NULL) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
 				Args:  []any{"service_account", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
 			},
 		},
@@ -180,7 +180,7 @@ func TestStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "SELECT count() AS __result_0 FROM signoz_audit.distributed_logs WHERE ((`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = true) AND (`attribute_string_signoz$$audit$$action` = ? AND `attribute_string_signoz$$audit$$action_exists` = true)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY __result_0 DESC",
+				Query: "SELECT count() AS __result_0 FROM signoz_audit.distributed_logs WHERE ((attributes.`signoz.audit.outcome`::String = ? AND attributes.`signoz.audit.outcome` IS NOT NULL) AND (attributes.`signoz.audit.action`::String = ? AND attributes.`signoz.audit.action` IS NOT NULL)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY __result_0 DESC",
 				Args:  []any{"failure", "update", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
 			},
 		},
@@ -204,7 +204,7 @@ func TestStatementBuilder(t *testing.T) {
 				Limit: 5,
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __limit_cte AS (SELECT toString(multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = true, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = true) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `signoz.audit.principal.email` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 60 SECOND) AS ts, toString(multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = true, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = true) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`signoz.audit.principal.email`) GLOBAL IN (SELECT `signoz.audit.principal.email` FROM __limit_cte) GROUP BY ts, `signoz.audit.principal.email`",
+				Query: "WITH __limit_cte AS (SELECT toString(multiIf(attributes.`signoz.audit.principal.email` IS NOT NULL, attributes.`signoz.audit.principal.email`::String, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE (attributes.`signoz.audit.outcome`::String = ? AND attributes.`signoz.audit.outcome` IS NOT NULL) 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(attributes.`signoz.audit.principal.email` IS NOT NULL, attributes.`signoz.audit.principal.email`::String, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE (attributes.`signoz.audit.outcome`::String = ? AND attributes.`signoz.audit.outcome` IS NOT NULL) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`signoz.audit.principal.email`) GLOBAL IN (SELECT `signoz.audit.principal.email` FROM __limit_cte) GROUP BY ts, `signoz.audit.principal.email`",
 				Args:  []any{"failure", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 5, "failure", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
 			},
 		},
@@ -224,3 +224,205 @@ func TestStatementBuilder(t *testing.T) {
 		})
 	}
 }
+
+// Audit has no keys tables yet, so every key outside the intrinsic fields
+// resolves through the storage fallback.
+func TestStatementBuilder_WithoutMetadata(t *testing.T) {
+	statementBuilder := newTestAuditStatementBuilder(t, map[string][]*telemetrytypes.TelemetryFieldKey{})
+	ctx := context.Background()
+
+	testCases := []struct {
+		name        string
+		requestType qbtypes.RequestType
+		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
+		expected    qbtypes.Statement
+	}{
+		{
+			name:        "ListByAttribute",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{Expression: "signoz.audit.outcome = 'failure'"},
+				Limit:  10,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE (attributes.`signoz.audit.outcome`::String = ? AND attributes.`signoz.audit.outcome` IS NOT NULL) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"failure", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+			},
+		},
+		{
+			name:        "ListByResourceAttribute",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{Expression: "resource.signoz.audit.resource.kind = 'dashboard'"},
+				Limit:  20,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE (resource.attributes.`signoz.audit.resource.kind`::String = ? AND resource.attributes.`signoz.audit.resource.kind` IS NOT NULL) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"dashboard", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 20},
+			},
+		},
+		{
+			name:        "ListByScopeNameAndBody",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{Expression: "scope.name = 'signoz.audit' AND body CONTAINS 'deleted'"},
+				Limit:  30,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE ((scope.name = ? AND scope.name <> '') AND LOWER(body.message) LIKE LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"signoz.audit", "%deleted%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 30},
+			},
+		},
+		{
+			name:        "ListByFullText",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{Expression: "'dashboard.deleted'"},
+				Limit:  40,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE match(body.message, ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"dashboard.deleted", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 40},
+			},
+		},
+		{
+			name:        "ListByNumericAttribute",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{Expression: "http.response.status_code >= 400"},
+				Limit:  50,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE toFloat64OrNull(attributes.`http.response.status_code`::String) >= ? AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{float64(400), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 50},
+			},
+		},
+		{
+			name:        "ListByAttributeExists_OrderByTimestamp",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{Expression: "signoz.audit.error.code EXISTS"},
+				Order: []qbtypes.OrderBy{
+					{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "timestamp"}}, Direction: qbtypes.OrderDirectionDesc},
+				},
+				Limit: 60,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE attributes.`signoz.audit.error.code` IS NOT NULL AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY timestamp desc LIMIT ?",
+				Args:  []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 60},
+			},
+		},
+		{
+			name:        "ListSelectFields",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				SelectFields: []telemetrytypes.TelemetryFieldKey{
+					{Name: "signoz.audit.principal.email", FieldContext: telemetrytypes.FieldContextAttribute},
+					{Name: "service.name", FieldContext: telemetrytypes.FieldContextResource},
+					{Name: "body"},
+				},
+				Limit: 70,
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT timestamp, id, multiIf(attributes.`signoz.audit.principal.email` IS NOT NULL, attributes.`signoz.audit.principal.email`::String, NULL) AS `signoz.audit.principal.email`, multiIf(resource.attributes.`service.name` IS NOT NULL, resource.attributes.`service.name`::String, NULL) AS `service.name`, body.message AS `body` FROM signoz_audit.distributed_logs WHERE timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 70},
+			},
+		},
+		{
+			name:        "ScalarCountGroupedByAttribute",
+			requestType: qbtypes.RequestTypeScalar,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				Source:       telemetrytypes.SourceAudit,
+				Aggregations: []qbtypes.LogAggregation{{Expression: "count()"}},
+				GroupBy: []qbtypes.GroupByKey{
+					{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "signoz.audit.action"}},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT toString(multiIf(attributes.`signoz.audit.action` IS NOT NULL, attributes.`signoz.audit.action`::String, NULL)) AS `signoz.audit.action`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `signoz.audit.action` ORDER BY __result_0 DESC",
+				Args:  []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			q, err := statementBuilder.Build(ctx, valuer.UUID{}, 1747947419000, 1747983448000, testCase.requestType, testCase.query, nil)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expected.Query, q.Query)
+			assert.Equal(t, testCase.expected.Args, q.Args)
+		})
+	}
+}
+
+func TestStatementBuilder_KnownResourceFields(t *testing.T) {
+	keysMap := map[string][]*telemetrytypes.TelemetryFieldKey{}
+	for name, field := range audittelemetryschema.ResourceFields {
+		keysMap[name] = []*telemetrytypes.TelemetryFieldKey{&field}
+	}
+	statementBuilder := newTestAuditStatementBuilder(t, keysMap)
+	ctx := context.Background()
+
+	testCases := []struct {
+		name        string
+		requestType qbtypes.RequestType
+		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
+		expected    qbtypes.Statement
+	}{
+		{
+			name:        "ListByBareResourceKey",
+			requestType: qbtypes.RequestTypeRaw,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal: telemetrytypes.SignalLogs,
+				Source: telemetrytypes.SourceAudit,
+				Filter: &qbtypes.Filter{Expression: "signoz.audit.resource.kind = 'role'"},
+				Limit:  10,
+			},
+			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 <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, body, event_name, attributes, resource, scope FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"role", "%signoz.audit.resource.kind%", "%signoz.audit.resource.kind\":\"role%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+			},
+		},
+		{
+			name:        "ScalarCountGroupedByBareResourceKey",
+			requestType: qbtypes.RequestTypeScalar,
+			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+				Signal:       telemetrytypes.SignalLogs,
+				Source:       telemetrytypes.SourceAudit,
+				Aggregations: []qbtypes.LogAggregation{{Expression: "count()"}},
+				GroupBy: []qbtypes.GroupByKey{
+					{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "signoz.audit.resource.kind"}},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "SELECT toString(multiIf(resource.attributes.`signoz.audit.resource.kind` IS NOT NULL, resource.attributes.`signoz.audit.resource.kind`::String, NULL)) AS `signoz.audit.resource.kind`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `signoz.audit.resource.kind` ORDER BY __result_0 DESC",
+				Args:  []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			q, err := statementBuilder.Build(ctx, valuer.UUID{}, 1747947419000, 1747983448000, testCase.requestType, testCase.query, nil)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expected.Query, q.Query)
+			assert.Equal(t, testCase.expected.Args, q.Args)
+			assert.Empty(t, q.Warnings)
+		})
+	}
+}

pkg/telemetrymetadata/metadata.go

@@ -691,6 +691,7 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, orgID valuer.UUID,
 	return keys, complete, nil
 }
 
+//nolint:unused // not called until audit keys are recorded in signoz_metadata
 func (t *telemetryMetaStore) auditTblStatementToFieldKeys(ctx context.Context) ([]*telemetrytypes.TelemetryFieldKey, error) {
 	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
 		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
@@ -717,6 +718,7 @@ func (t *telemetryMetaStore) auditTblStatementToFieldKeys(ctx context.Context) (
 	return materialisedKeys, nil
 }
 
+//nolint:unused // not called until audit keys are recorded in signoz_metadata
 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(),
@@ -884,18 +886,25 @@ func (t *telemetryMetaStore) getAuditKeys(ctx context.Context, fieldKeySelectors
 
 	complete := rowCount <= limit
 
-	// Add the matching intrinsic audit fields (same as logs intrinsics: body, severity_text, etc.)
-	for _, field := range maps.Values(audittelemetryschema.IntrinsicFields) {
-		if !staticFieldMatchesAny(field, fieldKeySelectors) {
-			continue
+	return appendStaticAuditKeys(keys, mapOfKeys, fieldKeySelectors), complete, nil
+}
+
+// appendStaticAuditKeys adds the matching intrinsic audit fields (same as logs intrinsics: body, severity_text, etc.)
+// and the known resource fields.
+func appendStaticAuditKeys(keys []*telemetrytypes.TelemetryFieldKey, mapOfKeys map[string]*telemetrytypes.TelemetryFieldKey, fieldKeySelectors []*telemetrytypes.FieldKeySelector) []*telemetrytypes.TelemetryFieldKey {
+	for _, fields := range []map[string]telemetrytypes.TelemetryFieldKey{audittelemetryschema.IntrinsicFields, audittelemetryschema.ResourceFields} {
+		for _, field := range fields {
+			if !staticFieldMatchesAny(field, fieldKeySelectors) {
+				continue
+			}
+			if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; added {
+				continue
+			}
+			keys = append(keys, &field)
 		}
-		if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; added {
-			continue
-		}
-		keys = append(keys, &field)
 	}
 
-	return keys, complete, nil
+	return keys
 }
 
 func getPriorityForContext(ctx telemetrytypes.FieldContext) int {
@@ -1209,7 +1218,7 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, orgID valuer.UUID, fie
 		keys, complete, err = t.getTracesKeys(ctx, selectors)
 	case telemetrytypes.SignalLogs:
 		if fieldKeySelector.Source == telemetrytypes.SourceAudit {
-			keys, complete, err = t.getAuditKeys(ctx, selectors)
+			keys = appendStaticAuditKeys(nil, map[string]*telemetrytypes.TelemetryFieldKey{}, selectors)
 		} else {
 			keys, complete, err = t.getLogsKeys(ctx, orgID, selectors)
 		}
@@ -1293,10 +1302,7 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, orgID valuer.UUID
 	if err != nil {
 		return nil, false, err
 	}
-	auditKeys, auditComplete, err := t.getAuditKeys(ctx, auditSelectors)
-	if err != nil {
-		return nil, false, err
-	}
+	auditKeys := appendStaticAuditKeys(nil, map[string]*telemetrytypes.TelemetryFieldKey{}, auditSelectors)
 	tracesKeys, tracesComplete, err := t.getTracesKeys(ctx, tracesSelectors)
 	if err != nil {
 		return nil, false, err
@@ -1311,7 +1317,7 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, orgID valuer.UUID
 		return nil, false, err
 	}
 	// Complete only if all queries are complete
-	complete := logsComplete && auditComplete && tracesComplete && metricsComplete
+	complete := logsComplete && tracesComplete && metricsComplete
 
 	mapOfKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey)
 	for _, key := range logsKeys {
@@ -1693,6 +1699,7 @@ func (t *telemetryMetaStore) getTagTableValues(ctx context.Context, table string
 	return values, complete, nil
 }
 
+//nolint:unused // not called until audit values are recorded in signoz_metadata
 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(),
@@ -2163,7 +2170,7 @@ func (t *telemetryMetaStore) GetAllValues(ctx context.Context, orgID valuer.UUID
 		values, complete, err = t.getSpanFieldValues(ctx, orgID, fieldValueSelector)
 	case telemetrytypes.SignalLogs:
 		if fieldValueSelector.Source == telemetrytypes.SourceAudit {
-			values, complete, err = t.getAuditFieldValues(ctx, fieldValueSelector)
+			values = &telemetrytypes.TelemetryFieldValues{}
 		} else {
 			values, complete, err = t.getLogFieldValues(ctx, orgID, fieldValueSelector)
 		}

pkg/telemetryschema/audittelemetryschema/const.go

@@ -22,15 +22,11 @@ const (
 	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"
+	AttributesColumn = "attributes"
+	ResourceColumn   = "resource"
+	ScopeColumn      = "scope"
 )
 
 var (
@@ -86,6 +82,18 @@ var (
 		},
 	}
 
+	// ResourceFields are the resource keys the audit writer emits, so a bare name resolves to resource.
+	ResourceFields = map[string]telemetrytypes.TelemetryFieldKey{
+		"service.name":                        newResourceField("service.name"),
+		"service.version":                     newResourceField("service.version"),
+		"signoz.audit.resource.kind":          newResourceField("signoz.audit.resource.kind"),
+		"signoz.audit.resource.id":            newResourceField("signoz.audit.resource.id"),
+		"signoz.audit.resource.object":        newResourceField("signoz.audit.resource.object"),
+		"signoz.audit.resource.target.kind":   newResourceField("signoz.audit.resource.target.kind"),
+		"signoz.audit.resource.target.id":     newResourceField("signoz.audit.resource.target.id"),
+		"signoz.audit.resource.target.object": newResourceField("signoz.audit.resource.target.object"),
+	}
+
 	DefaultSortingOrder = []qbtypes.OrderBy{
 		{
 			Key: qbtypes.OrderByKey{
@@ -117,15 +125,11 @@ var auditLogColumns = map[string]*schema.Column{
 	"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}},
+	"body":                 {Name: "body", Type: schema.JSONColumnType{}},
+	"scope":                {Name: "scope", Type: schema.JSONColumnType{}},
+	"attributes":           {Name: "attributes", Type: schema.JSONColumnType{}},
 	"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}},
 }
 
 const (
@@ -138,3 +142,12 @@ const (
 	LogResourceKeysTblName      = "distributed_logs_resource_keys"
 	LogsResourceTableName       = "distributed_logs_resource"
 )
+
+func newResourceField(name string) telemetrytypes.TelemetryFieldKey {
+	return telemetrytypes.TelemetryFieldKey{
+		Name:          name,
+		Signal:        telemetrytypes.SignalLogs,
+		FieldContext:  telemetrytypes.FieldContextResource,
+		FieldDataType: telemetrytypes.FieldDataTypeString,
+	}
+}

pkg/telemetryschema/audittelemetryschema/storage.go

@@ -24,24 +24,11 @@ func NewStorage() qbtypes.Storage {
 func (m *storage) getColumn(_ context.Context, key *telemetrytypes.TelemetryFieldKey) ([]*schema.Column, error) {
 	switch key.FieldContext {
 	case telemetrytypes.FieldContextResource:
-		return []*schema.Column{auditLogColumns["resource"]}, nil
+		return []*schema.Column{auditLogColumns[ResourceColumn]}, 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
+		return []*schema.Column{auditLogColumns[ScopeColumn]}, 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
-		}
+		return []*schema.Column{auditLogColumns[AttributesColumn]}, nil
 	case telemetrytypes.FieldContextLog, telemetrytypes.FieldContextUnspecified:
 		col, ok := auditLogColumns[key.Name]
 		if !ok {
@@ -53,46 +40,6 @@ func (m *storage) getColumn(_ context.Context, key *telemetrytypes.TelemetryFiel
 	return nil, qbtypes.ErrColumnNotFound
 }
 
-func (m *storage) read(ctx context.Context, _ qbtypes.QueryInfo, key *telemetrytypes.TelemetryFieldKey) (string, error) {
-	columns, err := m.getColumn(ctx, key)
-	if err != nil {
-		return "", err
-	}
-	if len(columns) != 1 {
-		return "", errors.NewInternalf(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.NewInvalidInputf(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, clickhousesql.Identifier(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, clickhousesql.StringLiteral(key.Name)), nil
-		default:
-			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported map value type %s", valueType)
-		}
-	}
-
-	return column.Name, nil
-}
-
 // Read composes the bare read of one key with its membership test and what
 // an absent row reads.
 func (m *storage) Read(ctx context.Context, q qbtypes.QueryInfo, key *telemetrytypes.TelemetryFieldKey) (qbtypes.Read, error) {
@@ -100,46 +47,43 @@ func (m *storage) Read(ctx context.Context, q qbtypes.QueryInfo, key *telemetryt
 	if err != nil {
 		return qbtypes.Read{}, err
 	}
-	sql, err := m.read(ctx, q, key)
+	if len(columns) != 1 {
+		return qbtypes.Read{}, errors.NewInternalf(errors.CodeInternal, "expected exactly 1 column, got %d", len(columns))
+	}
+	column := columns[0]
+
+	if column.Type.GetType() == schema.ColumnTypeEnumJSON {
+		return jsonRead(column, key)
+	}
+
+	presence, err := querybuilder.ExistsExpression(columns, key, q.StartNs, q.EndNs, column.Name, true)
 	if err != nil {
 		return qbtypes.Read{}, err
 	}
-	presence, err := querybuilder.ExistsExpression(columns, key, q.StartNs, q.EndNs, sql, true)
+	absence, err := querybuilder.ExistsExpression(columns, key, q.StartNs, q.EndNs, column.Name, false)
 	if err != nil {
 		return qbtypes.Read{}, err
 	}
-	absence, err := querybuilder.ExistsExpression(columns, key, q.StartNs, q.EndNs, sql, false)
-	if err != nil {
-		return qbtypes.Read{}, err
-	}
-	return qbtypes.Read{SQL: sql, Presence: presence, Absence: absence, WhenAbsent: absentReads(columns[0])}, nil
+	return qbtypes.Read{SQL: column.Name, Presence: presence, Absence: absence, WhenAbsent: qbtypes.AlwaysPresent}, nil
 }
 
-// absentReads tells what a row without the key reads from the column. A
-// map reads its empty value. A JSON path reads the empty string, because
-// its ::String cast folds NULL. Every other column reads a real value.
-func absentReads(column *schema.Column) qbtypes.Absent {
-	switch column.Type.GetType() {
-	case schema.ColumnTypeEnumMap, schema.ColumnTypeEnumJSON:
-		return qbtypes.AbsentIsSentinel
+// Fallback answers a column name with the column, and synthesizes any other
+// key from the operand, since audit has no keys tables to report it.
+func (m *storage) Fallback(_ context.Context, _ qbtypes.QueryInfo, key *telemetrytypes.TelemetryFieldKey, _ qbtypes.FilterOperator, value any) ([]*telemetrytypes.LogicalField, error) {
+	var keys []*telemetrytypes.TelemetryFieldKey
+	switch key.FieldContext {
+	case telemetrytypes.FieldContextUnspecified, telemetrytypes.FieldContextLog:
+		if column, ok := auditLogColumns[key.Name]; ok {
+			keys = []*telemetrytypes.TelemetryFieldKey{telemetrytypes.NewTelemetryFieldKey(key.Name, telemetrytypes.FieldContextLog, columnDataType(key, column))}
+			break
+		}
+		if key.FieldContext == telemetrytypes.FieldContextUnspecified {
+			keys = synthesizeKeys(key, value)
+		}
+	case telemetrytypes.FieldContextAttribute, telemetrytypes.FieldContextResource, telemetrytypes.FieldContextScope:
+		keys = synthesizeKeys(key, value)
 	}
-	return qbtypes.AlwaysPresent
-}
-
-// Fallback answers a column name with the column. Audit has no attribute
-// synthesis, so any other unknown key stays unresolved and the caller
-// errors.
-func (m *storage) Fallback(_ context.Context, _ qbtypes.QueryInfo, key *telemetrytypes.TelemetryFieldKey, _ qbtypes.FilterOperator, _ any) ([]*telemetrytypes.LogicalField, error) {
-	schemaColumn, ok := auditLogColumns[key.Name]
-	if !ok {
-		return nil, nil
-	}
-	dataType := key.FieldDataType
-	if dataType == telemetrytypes.FieldDataTypeUnspecified {
-		dataType = querybuilder.ColumnDataType(schemaColumn)
-	}
-	column := telemetrytypes.NewTelemetryFieldKey(key.Name, telemetrytypes.FieldContextLog, dataType)
-	return querybuilder.WrapAsLogicalFields(key.Name, []*telemetrytypes.TelemetryFieldKey{column}), nil
+	return querybuilder.WrapAsLogicalFields(key.Name, keys), nil
 }
 
 func (m *storage) Traits() qbtypes.Traits {
@@ -149,3 +93,81 @@ func (m *storage) Traits() qbtypes.Traits {
 func (m *storage) Compile(ctx context.Context, q qbtypes.QueryInfo, logical *telemetrytypes.LogicalField, operator qbtypes.FilterOperator, value any, sb *sqlbuilder.SelectBuilder) (qbtypes.Compiled, error) {
 	return querybuilder.SharedCondition(ctx, q, m, logical, operator, value, sb)
 }
+
+// jsonRead reads a key from one of the JSON columns. The body text and the
+// scope name and version are typed paths, so they read without a cast.
+// Resource and scope attributes sit under an `attributes` path.
+func jsonRead(column *schema.Column, key *telemetrytypes.TelemetryFieldKey) (qbtypes.Read, error) {
+	switch column.Name {
+	case BodyColumn:
+		if key.FieldContext != telemetrytypes.FieldContextLog && key.FieldContext != telemetrytypes.FieldContextUnspecified {
+			break
+		}
+		path := BodyColumn + ".message"
+		return qbtypes.Read{SQL: path, Presence: path + " <> ''", Absence: path + " = ''", WhenAbsent: qbtypes.AlwaysPresent}, nil
+	case ScopeColumn:
+		switch key.Name {
+		case "name", "scope.name", "scope_name":
+			return declaredStringRead(ScopeColumn + ".name"), nil
+		case "version", "scope.version", "scope_version":
+			return declaredStringRead(ScopeColumn + ".version"), nil
+		}
+		return stringPathRead(fmt.Sprintf("%s.attributes.%s", ScopeColumn, clickhousesql.Identifier(key.Name))), nil
+	case ResourceColumn:
+		if key.FieldContext != telemetrytypes.FieldContextResource {
+			break
+		}
+		return stringPathRead(fmt.Sprintf("%s.attributes.%s", ResourceColumn, clickhousesql.Identifier(key.Name))), nil
+	case AttributesColumn:
+		if key.FieldContext != telemetrytypes.FieldContextAttribute {
+			break
+		}
+		path := fmt.Sprintf("%s.%s", AttributesColumn, clickhousesql.Identifier(key.Name))
+		switch key.FieldDataType {
+		case telemetrytypes.FieldDataTypeInt64, telemetrytypes.FieldDataTypeFloat64, telemetrytypes.FieldDataTypeNumber:
+			return typedPathRead(fmt.Sprintf("toFloat64OrNull(%s::String)", path)), nil
+		case telemetrytypes.FieldDataTypeBool:
+			return typedPathRead(fmt.Sprintf("if(dynamicType(%s) = 'Bool', accurateCastOrNull(%s, 'Bool'), NULL)", path, path)), nil
+		}
+		return stringPathRead(path), nil
+	}
+
+	return qbtypes.Read{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "field %s in context %s cannot be read from the json column %s", key.Name, key.FieldContext.StringValue(), column.Name)
+}
+
+// declaredStringRead reads a typed String path, which reads an empty string when absent.
+func declaredStringRead(path string) qbtypes.Read {
+	return qbtypes.Read{SQL: path, Presence: path + " <> ''", Absence: path + " = ''", WhenAbsent: qbtypes.AbsentIsSentinel}
+}
+
+// stringPathRead reads a dynamic path as String. The cast folds NULL to an
+// empty string, so presence tests the raw path.
+func stringPathRead(path string) qbtypes.Read {
+	return qbtypes.Read{SQL: path + "::String", Presence: path + " IS NOT NULL", Absence: path + " IS NULL", WhenAbsent: qbtypes.AbsentIsSentinel}
+}
+
+// typedPathRead reads a number or bool path through a cast that is NULL when
+// the key is absent or stored as another type.
+// TODO: share with tracestelemetryschema.AttributeJSONValueExpr from a common package.
+func typedPathRead(expr string) qbtypes.Read {
+	return qbtypes.Read{SQL: expr, Presence: expr + " IS NOT NULL", Absence: expr + " IS NULL", WhenAbsent: qbtypes.AbsentIsNull}
+}
+
+// synthesizeKeys reads an untyped key with no operand as String only. A JSON
+// path casts every stored type to String, so the other types add nothing.
+func synthesizeKeys(key *telemetrytypes.TelemetryFieldKey, value any) []*telemetrytypes.TelemetryFieldKey {
+	if value == nil && key.FieldDataType == telemetrytypes.FieldDataTypeUnspecified {
+		key = telemetrytypes.NewTelemetryFieldKey(key.Name, key.FieldContext, telemetrytypes.FieldDataTypeString)
+	}
+	return querybuilder.SynthesizeKeys(key, value)
+}
+
+func columnDataType(key *telemetrytypes.TelemetryFieldKey, column *schema.Column) telemetrytypes.FieldDataType {
+	if key.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+		return key.FieldDataType
+	}
+	if field, ok := IntrinsicFields[key.Name]; ok {
+		return field.FieldDataType
+	}
+	return querybuilder.ColumnDataType(column)
+}

tests/fixtures/audit.py

@@ -37,71 +37,7 @@ class AuditResource(ABC):
         )
 
 
-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: np.int64 | None
-    float64_value: np.float64 | None
-
-    def __init__(
-        self,
-        timestamp: datetime.datetime,
-        tag_key: str,
-        tag_type: str,
-        tag_data_type: str,
-        string_value: str | None,
-        int64_value: np.int64 | None,
-        float64_value: np.float64 | None,
-    ) -> 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
@@ -112,20 +48,13 @@ class AuditLog(ABC):
     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]
+    body: dict[str, str]
+    scope: dict[str, Any]
+    attributes: dict[str, Any]
+    resource_json: dict[str, Any]
     event_name: str
 
     resource: list[AuditResource]
-    tag_attributes: list[AuditTagAttributes]
-    resource_keys: list[AuditResourceOrAttributeKeys]
-    attribute_keys: list[AuditResourceOrAttributeKeys]
 
     def __init__(
         self,
@@ -143,9 +72,6 @@ class AuditLog(ABC):
     ) -> 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
@@ -164,97 +90,19 @@ class AuditLog(ABC):
         self.severity_text = severity_text
         self.severity_number = np.uint8(9 if severity_text == "INFO" else 17)
 
-        self.body = body
+        self.body = {"message": 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"))
+        resource_attributes = {k: str(v) for k, v in resources.items()}
+        self.resource_json = {"attributes": resource_attributes}
+        self.resource_fingerprint = LogsOrTracesFingerprint(resource_attributes).calculate()
 
-        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.attributes = dict(attributes)
+        self.scope = {"name": scope_name, "version": scope_version, "attributes": {}}
 
         self.resource = [
             AuditResource(
-                labels=self.resource_json,
+                labels=resource_attributes,
                 fingerprint=self.resource_fingerprint,
                 seen_at_ts_bucket_start=self.ts_bucket_start,
             )
@@ -274,15 +122,12 @@ class AuditLog(ABC):
                 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.scope,
+                self.attributes,
                 self.resource_json,
                 self.event_name,
-            ]
+            ],
+            dtype=object,
         )
 
 
@@ -307,50 +152,6 @@ def insert_audit_logs(
                 ],
             )
 
-        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",
@@ -367,12 +168,8 @@ def insert_audit_logs(
                 "severity_text",
                 "severity_number",
                 "body",
-                "scope_name",
-                "scope_version",
-                "scope_string",
-                "attributes_string",
-                "attributes_number",
-                "attributes_bool",
+                "scope",
+                "attributes",
                 "resource",
                 "event_name",
             ],
@@ -381,11 +178,5 @@ def insert_audit_logs(
     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",
-    ]:
+    for table in ["logs", "logs_resource"]:
         clickhouse.conn.query(f"TRUNCATE TABLE signoz_audit.{table} ON CLUSTER '{cluster}' SYNC")

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

@@ -114,9 +114,19 @@ def test_audit_list_all(
     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"]["body"] == {"message": "ops@acme.com (user-010) updated user (user-020)"}
     assert rows[0]["data"]["severity_text"] == "INFO"
 
+    assert rows[0]["data"]["resource"] == {
+        "attributes": {
+            "service.name": "signoz",
+            "signoz.audit.resource.kind": "user",
+            "signoz.audit.resource.id": "user-020",
+        }
+    }
+    assert rows[0]["data"]["attributes"]["signoz.audit.action_category"] == "access_control"
+    assert rows[0]["data"]["scope"]["name"] == "signoz.audit"
+
 
 @pytest.mark.parametrize(
     "filter_expression,expected_count,expected_event_names",
@@ -151,6 +161,18 @@ def test_audit_list_all(
             {"dashboard.deleted"},
             id="filter_by_resource_kind_and_action",
         ),
+        pytest.param(
+            "resource.signoz.audit.resource.id = 'sa-001'",
+            1,
+            {"serviceaccount.apikey.created"},
+            id="filter_by_resource_prefix",
+        ),
+        pytest.param(
+            "body CONTAINS 'failed'",
+            1,
+            {"dashboard.deleted"},
+            id="filter_by_body_contains",
+        ),
     ],
 )
 def test_audit_filter(