refactor(statementbuilder): ship a noop audit builder in community

Community declares the audit source with a builder that refuses the query,
rather than leaving it unregistered, matching how the other optional
subsystems are wired. LogSourceQuery.Resolver is optional since community
registers no audit resolver.

cmd/community/server.go

@@ -44,6 +44,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/ruler/signozruler"
 	"github.com/SigNoz/signoz/pkg/signoz"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
+	"github.com/SigNoz/signoz/pkg/statementbuilder/noopauditstatementbuilder"
 	"github.com/SigNoz/signoz/pkg/subscription"
 	"github.com/SigNoz/signoz/pkg/subscription/noopsubscription"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
@@ -121,6 +122,10 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 		func(_ context.Context, _ factory.ProviderSettings, _ flagger.Flagger, _ licensing.Licensing, _ telemetrystore.TelemetryStore, _ retention.Getter, _ organization.Getter, _ zeus.Zeus) (factory.NamedMap[factory.ProviderFactory[meterreporter.Reporter, meterreporter.Config]], string) {
 			return signoz.NewMeterReporterProviderFactories(), "noop"
 		},
+		[]signoz.LogSourceQuery{{
+			Source:                  telemetrytypes.SourceAudit,
+			StatementBuilderFactory: noopauditstatementbuilder.NewFactory,
+		}},
 		func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
 			return querier.NewHandler(ps, q, a)
 		},

cmd/enterprise/server.go

@@ -28,7 +28,9 @@ import (
 	eequerier "github.com/SigNoz/signoz/ee/querier"
 	enterpriseapp "github.com/SigNoz/signoz/ee/query-service/app"
 	eerules "github.com/SigNoz/signoz/ee/query-service/rules"
+	"github.com/SigNoz/signoz/ee/statementbuilder/auditstatementbuilder"
 	"github.com/SigNoz/signoz/ee/subscription/httpsubscription"
+	"github.com/SigNoz/signoz/ee/telemetrymetadata/audittelemetrymetadata"
 	enterprisezeus "github.com/SigNoz/signoz/ee/zeus"
 	"github.com/SigNoz/signoz/ee/zeus/httpzeus"
 	"github.com/SigNoz/signoz/pkg/alertmanager"
@@ -172,6 +174,13 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 
 			return factories, "http"
 		},
+		[]signoz.LogSourceQuery{{
+			Source: telemetrytypes.SourceAudit,
+			Resolver: func(ts telemetrystore.TelemetryStore) telemetrytypes.FieldKeyResolver {
+				return audittelemetrymetadata.New(ts)
+			},
+			StatementBuilderFactory: auditstatementbuilder.NewFactory,
+		}},
 		func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
 			communityHandler := querier.NewHandler(ps, q, a)
 			return eequerier.NewHandler(ps, q, communityHandler)

ee/statementbuilder/auditstatementbuilder/statement_builder.go

@@ -6,6 +6,7 @@ import (
 	"log/slog"
 	"strings"
 
+	"github.com/SigNoz/signoz/ee/telemetryschema/audittelemetryschema"
 	"github.com/SigNoz/signoz/pkg/clickhousesql"
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
@@ -13,7 +14,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
-	"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -58,7 +58,7 @@ func NewAuditQueryStatementBuilder(
 	fullTextColumn *telemetrytypes.TelemetryFieldKey,
 	flagger flagger.Flagger,
 ) *auditQueryStatementBuilder {
-	auditSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema")
+	auditSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/ee/telemetryschema/audittelemetryschema")
 
 	resourceFilterStmtBuilder := resourcefilter.New[qbtypes.LogAggregation](
 		settings,

ee/statementbuilder/auditstatementbuilder/statement_builder_test.go

@@ -5,10 +5,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/SigNoz/signoz/ee/telemetryschema/audittelemetryschema"
 	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
-	"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"

ee/telemetrymetadata/audittelemetrymetadata/resolver.go

@@ -0,0 +1,333 @@
+package audittelemetrymetadata
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/huandu/go-sqlbuilder"
+	"golang.org/x/exp/maps"
+
+	"github.com/SigNoz/signoz/ee/telemetryschema/audittelemetryschema"
+	"github.com/SigNoz/signoz/pkg/clickhousesql"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/telemetrymetadata/parser"
+	"github.com/SigNoz/signoz/pkg/telemetrymetadata/resolvercore"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
+	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+var errFailedToGetAuditKeys = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get audit keys")
+
+// Resolver answers key and value lookups for the audit log tables. It is
+// registered against telemetrytypes.SourceAudit; without it the metadata store
+// carries no audit knowledge.
+type Resolver struct {
+	store telemetrystore.TelemetryStore
+}
+
+var _ telemetrytypes.FieldKeyResolver = (*Resolver)(nil)
+
+func New(store telemetrystore.TelemetryStore) *Resolver {
+	return &Resolver{store: store}
+}
+
+func (r *Resolver) tblStatementToFieldKeys(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", audittelemetryschema.DBName, audittelemetryschema.AuditLogsTableName)
+	statements := []telemetrytypes.ShowCreateTableStatement{}
+	err := r.store.ClickhouseDB().Select(ctx, &statements, query)
+	if err != nil {
+		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, resolvercore.ErrFailedToGetTblStatement.Error())
+	}
+
+	materialisedKeys, err := parser.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 (r *Resolver) GetKeys(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 := r.tblStatementToFieldKeys(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, audittelemetryschema.DBName + "." + audittelemetryschema.LogAttributeKeysTblName},
+		{telemetrytypes.FieldContextResource, queryResourceTable, audittelemetryschema.DBName + "." + audittelemetryschema.LogResourceKeysTblName},
+	}
+
+	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", resolvercore.PriorityForContext(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", "%"+clickhousesql.LikePattern(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...)
+	}
+
+	var limit int
+	for _, fieldKeySelector := range fieldKeySelectors {
+		limit += fieldKeySelector.Limit
+	}
+	if limit == 0 {
+		limit = 1000
+	}
+
+	keys := []*telemetrytypes.TelemetryFieldKey{}
+	rowCount := 0
+
+	// the log and scope contexts have no keys table; they are served by the
+	// static fields appended below
+	if len(queries) > 0 {
+		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 := r.store.ClickhouseDB().Query(ctx, mainQuery, allArgs...)
+		if err != nil {
+			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, errFailedToGetAuditKeys.Error())
+		}
+		defer rows.Close()
+
+		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 the matching intrinsic audit fields (same as logs intrinsics: body, severity_text, etc.)
+	for _, field := range maps.Values(audittelemetryschema.IntrinsicFields) {
+		if !resolvercore.StaticFieldMatchesAny(field, fieldKeySelectors) {
+			continue
+		}
+		if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; added {
+			continue
+		}
+		keys = append(keys, &field)
+	}
+
+	return keys, complete, nil
+}
+
+func (r *Resolver) GetAllValues(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(audittelemetryschema.DBName + "." + audittelemetryschema.TagAttributesTableName)
+
+	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", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
+		case telemetrytypes.FieldDataTypeNumber:
+			sb.Where(sb.IsNotNull("number_value"))
+			sb.Where(sb.ILike("toString(number_value)", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
+		case telemetrytypes.FieldDataTypeUnspecified:
+			sb.Where(sb.Or(
+				sb.ILike("string_value", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"),
+				sb.ILike("toString(number_value)", "%"+clickhousesql.LikePattern(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 := r.store.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
+}

pkg/querier/log_source.go

@@ -0,0 +1,32 @@
+package querier
+
+import (
+	"context"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// logStmtBuilderFor returns the statement builder for a log query's source. A
+// source this build registered no builder for resolves to one that reports it
+// unavailable, so the query fails with that instead of silently reading the
+// plain logs tables.
+func (q *querier) logStmtBuilderFor(source telemetrytypes.Source) qbtypes.StatementBuilder[qbtypes.LogAggregation] {
+	if source == telemetrytypes.SourceUnspecified {
+		return q.logStmtBuilder
+	}
+	if builder, ok := q.sourceLogStmtBuilders[source]; ok {
+		return builder
+	}
+	return unavailableSourceStmtBuilder{source: source}
+}
+
+type unavailableSourceStmtBuilder struct {
+	source telemetrytypes.Source
+}
+
+func (b unavailableSourceStmtBuilder) Build(_ context.Context, _ valuer.UUID, _, _ uint64, _ qbtypes.RequestType, _ qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], _ map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
+	return nil, errors.Newf(errors.TypeInvalidInput, errors.CodeUnsupported, "log source %q is not available in this build", b.source.StringValue())
+}

pkg/querier/querier.go

@@ -54,7 +54,7 @@ type querier struct {
 	traceStmtBuilder         qbtypes.StatementBuilder[qbtypes.TraceAggregation]
 	aiTraceStmtBuilder       qbtypes.StatementBuilder[qbtypes.TraceAggregation]
 	logStmtBuilder           qbtypes.StatementBuilder[qbtypes.LogAggregation]
-	auditStmtBuilder         qbtypes.StatementBuilder[qbtypes.LogAggregation]
+	sourceLogStmtBuilders    map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation]
 	metricStmtBuilder        qbtypes.StatementBuilder[qbtypes.MetricAggregation]
 	meterStmtBuilder         qbtypes.StatementBuilder[qbtypes.MetricAggregation]
 	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder
@@ -74,7 +74,7 @@ func New(
 	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
-	auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
+	sourceLogStmtBuilders map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder,
@@ -96,7 +96,7 @@ func New(
 		traceStmtBuilder:         traceStmtBuilder,
 		aiTraceStmtBuilder:       aiTraceStmtBuilder,
 		logStmtBuilder:           logStmtBuilder,
-		auditStmtBuilder:         auditStmtBuilder,
+		sourceLogStmtBuilders:    sourceLogStmtBuilders,
 		metricStmtBuilder:        metricStmtBuilder,
 		meterStmtBuilder:         meterStmtBuilder,
 		traceOperatorStmtBuilder: traceOperatorStmtBuilder,
@@ -266,10 +266,7 @@ func (q *querier) buildQueries(
 			case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
 				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
-				stmtBuilder := q.logStmtBuilder
-				if spec.Source == telemetrytypes.SourceAudit {
-					stmtBuilder = q.auditStmtBuilder
-				}
+				stmtBuilder := q.logStmtBuilderFor(spec.Source)
 				bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, stmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, q.builderConfig)
 				queries[spec.Name] = bq
 				steps[spec.Name] = spec.StepInterval
@@ -602,10 +599,7 @@ 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)
-			liveTailStmtBuilder := q.logStmtBuilder
-			if spec.Source == telemetrytypes.SourceAudit {
-				liveTailStmtBuilder = q.auditStmtBuilder
-			}
+			liveTailStmtBuilder := q.logStmtBuilderFor(spec.Source)
 			bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, liveTailStmtBuilder, query.Type, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
 				"id": {
 					Value: updatedLogID,
@@ -937,10 +931,7 @@ func (q *querier) createRangedQuery(_ valuer.UUID, originalQuery qbtypes.Query,
 		specCopy := qt.spec.Copy()
 		specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
 		adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
-		shiftStmtBuilder := q.logStmtBuilder
-		if qt.spec.Source == telemetrytypes.SourceAudit {
-			shiftStmtBuilder = q.auditStmtBuilder
-		}
+		shiftStmtBuilder := q.logStmtBuilderFor(qt.spec.Source)
 		return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, shiftStmtBuilder, qt.queryType, specCopy, adjustedTimeRange, qt.kind, qt.variables, q.builderConfig)
 
 	case *builderQuery[qbtypes.MetricAggregation]:

pkg/querier/signozquerier/provider.go

@@ -22,7 +22,7 @@ func NewFactory(
 	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
-	auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
+	sourceLogStmtBuilders map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder,
@@ -44,7 +44,7 @@ func NewFactory(
 				traceStmtBuilder,
 				aiTraceStmtBuilder,
 				logStmtBuilder,
-				auditStmtBuilder,
+				sourceLogStmtBuilders,
 				metricStmtBuilder,
 				meterStmtBuilder,
 				traceOperatorStmtBuilder,

pkg/query-service/rules/managertestfactory.go

@@ -20,7 +20,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/sqlstore/sqlstoretest"
 	"github.com/SigNoz/signoz/pkg/statementbuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/aistatementbuilder"
-	"github.com/SigNoz/signoz/pkg/statementbuilder/auditstatementbuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/logsstatementbuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/meterstatementbuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
@@ -113,7 +112,7 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
 	}
 
 	// Create querier with test values
-	metadataStore := telemetrymetadata.NewTelemetryMetaStore(providerSettings, telemetryStore, flagger)
+	metadataStore := telemetrymetadata.NewTelemetryMetaStore(providerSettings, telemetryStore, flagger, nil)
 	cfg := statementbuilder.Config{}
 	ctx := context.Background()
 	traceStmtBuilder, err := tracesstatementbuilder.NewFactory(telemetryStore, metadataStore, flagger).New(ctx, providerSettings, cfg)
@@ -124,14 +123,12 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
 	require.NoError(t, err)
 	logStmtBuilder, err := logsstatementbuilder.NewFactory(telemetryStore, metadataStore, flagger).New(ctx, providerSettings, cfg)
 	require.NoError(t, err)
-	auditStmtBuilder, err := auditstatementbuilder.NewFactory(metadataStore, flagger).New(ctx, providerSettings, cfg)
-	require.NoError(t, err)
 	metricStmtBuilder, err := metricsstatementbuilder.NewFactory(metadataStore, flagger).New(ctx, providerSettings, cfg)
 	require.NoError(t, err)
 	meterStmtBuilder, err := meterstatementbuilder.NewFactory(metadataStore, flagger).New(ctx, providerSettings, cfg)
 	require.NoError(t, err)
 	bucketCache := querier.NewBucketCache(providerSettings, cache, 0, 0)
-	providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger)
+	providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, nil, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger)
 	mockQuerier, err := providerFactory.New(context.Background(), providerSettings, querier.Config{})
 	require.NoError(t, err)
 

pkg/signoz/log_source_query.go

@@ -0,0 +1,20 @@
+package signoz
+
+import (
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/statementbuilder"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+// LogSourceQuery supplies the query stack for one source within the logs
+// signal. The resolver is built before the metadata store and the statement
+// builder after it, so they arrive as two functions rather than two values.
+// Resolver may be nil for a source that serves no key or value suggestions.
+type LogSourceQuery struct {
+	Source                  telemetrytypes.Source
+	Resolver                func(telemetrystore.TelemetryStore) telemetrytypes.FieldKeyResolver
+	StatementBuilderFactory func(telemetrytypes.MetadataStore, flagger.Flagger) factory.ProviderFactory[qbtypes.StatementBuilder[qbtypes.LogAggregation], statementbuilder.Config]
+}

pkg/signoz/provider.go

@@ -324,9 +324,9 @@ func NewStatsReporterProviderFactories(aggregator statsreporter.Aggregator, orgG
 	)
 }
 
-func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], aiTraceStmtBuilder 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, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
+func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], sourceLogStmtBuilders map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation], metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
 	return factory.MustNewNamedMap(
-		signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
+		signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, sourceLogStmtBuilders, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
 	)
 }
 

pkg/signoz/signoz.go

@@ -50,7 +50,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/sqlschema"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/aistatementbuilder"
-	"github.com/SigNoz/signoz/pkg/statementbuilder/auditstatementbuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/logsstatementbuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/meterstatementbuilder"
 	"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
@@ -103,7 +102,8 @@ type SigNoz struct {
 
 // newQueryStack assembles the query stack once and returns, in order: the shared
 // telemetry metadata store (reused elsewhere in signoz.New), the per-signal
-// statement builders (trace, ai-trace, log, audit, metric, meter, trace-operator),
+// statement builders (trace, ai-trace, log, per-source log, metric, meter,
+// trace-operator),
 // and the bucket cache. It is the only place that imports the concrete
 // statement-builder sub-packages.
 func newQueryStack(
@@ -113,19 +113,27 @@ func newQueryStack(
 	telemetryStore telemetrystore.TelemetryStore,
 	cache cache.Cache,
 	fl flagger.Flagger,
+	logSources []LogSourceQuery,
 ) (
 	telemetrytypes.MetadataStore,
 	qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	qbtypes.StatementBuilder[qbtypes.LogAggregation],
-	qbtypes.StatementBuilder[qbtypes.LogAggregation],
+	map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	qbtypes.TraceOperatorStatementBuilder,
 	querier.BucketCache,
 	error,
 ) {
-	metadataStore := telemetrymetadata.NewTelemetryMetaStore(settings, telemetryStore, fl)
+	resolvers := map[telemetrytypes.Source]telemetrytypes.FieldKeyResolver{}
+	for _, logSource := range logSources {
+		if logSource.Resolver == nil {
+			continue
+		}
+		resolvers[logSource.Source] = logSource.Resolver(telemetryStore)
+	}
+	metadataStore := telemetrymetadata.NewTelemetryMetaStore(settings, telemetryStore, fl, resolvers)
 
 	cfg := config.Querier.Config
 	traceStmtBuilder, err := tracesstatementbuilder.NewFactory(telemetryStore, metadataStore, fl).New(ctx, settings, cfg)
@@ -144,9 +152,13 @@ func newQueryStack(
 	if err != nil {
 		return nil, nil, nil, nil, nil, nil, nil, nil, nil, err
 	}
-	auditStmtBuilder, err := auditstatementbuilder.NewFactory(metadataStore, fl).New(ctx, settings, cfg)
-	if err != nil {
-		return nil, nil, nil, nil, nil, nil, nil, nil, nil, err
+	sourceLogStmtBuilders := map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation]{}
+	for _, logSource := range logSources {
+		builder, err := logSource.StatementBuilderFactory(metadataStore, fl).New(ctx, settings, cfg)
+		if err != nil {
+			return nil, nil, nil, nil, nil, nil, nil, nil, nil, err
+		}
+		sourceLogStmtBuilders[logSource.Source] = builder
 	}
 	metricStmtBuilder, err := metricsstatementbuilder.NewFactory(metadataStore, fl).New(ctx, settings, cfg)
 	if err != nil {
@@ -159,7 +171,7 @@ func newQueryStack(
 
 	bucketCache := querier.NewBucketCache(settings, cache, config.Querier.CacheTTL, config.Querier.FluxInterval)
 
-	return metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, nil
+	return metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, sourceLogStmtBuilders, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, nil
 }
 
 func New(
@@ -182,6 +194,7 @@ func New(
 	gatewayProviderFactory func(licensing.Licensing) factory.ProviderFactory[gateway.Gateway, gateway.Config],
 	auditorProviderFactories func(licensing.Licensing) factory.NamedMap[factory.ProviderFactory[auditor.Auditor, auditor.Config]],
 	meterReporterProviderFactories func(context.Context, factory.ProviderSettings, flagger.Flagger, licensing.Licensing, telemetrystore.TelemetryStore, retention.Getter, organization.Getter, zeus.Zeus) (factory.NamedMap[factory.ProviderFactory[meterreporter.Reporter, meterreporter.Config]], string),
+	logSources []LogSourceQuery,
 	querierHandlerCallback func(factory.ProviderSettings, querier.Querier, analytics.Analytics) querier.Handler,
 	cloudIntegrationCallback func(sqlstore.SQLStore, dashboard.Module, global.Global, zeus.Zeus, gateway.Gateway, licensing.Licensing, serviceaccount.Module, cloudintegration.Config) (cloudintegration.Module, error),
 	metricReductionRuleModuleCallback func(sqlstore.SQLStore, telemetrystore.TelemetryStore, dashboard.Module, queryparser.QueryParser, licensing.Licensing, flagger.Flagger, telemetrytypes.MetadataStore, factory.ProviderSettings, int) metricreductionrule.Module,
@@ -323,7 +336,7 @@ func New(
 
 	// Assemble the query stack (metadata store, statement builders, bucket cache) once,
 	// and reuse the single metadata store everywhere downstream.
-	telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, err := newQueryStack(ctx, providerSettings, config, telemetrystore, cache, flagger)
+	telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, sourceLogStmtBuilders, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, err := newQueryStack(ctx, providerSettings, config, telemetrystore, cache, flagger, logSources)
 	if err != nil {
 		return nil, err
 	}
@@ -333,7 +346,7 @@ func New(
 		ctx,
 		providerSettings,
 		config.Querier,
-		NewQuerierProviderFactories(telemetrystore, prometheus, telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
+		NewQuerierProviderFactories(telemetrystore, prometheus, telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, sourceLogStmtBuilders, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
 		config.Querier.Provider(),
 	)
 	if err != nil {

pkg/statementbuilder/noopauditstatementbuilder/statement_builder.go

@@ -0,0 +1,32 @@
+package noopauditstatementbuilder
+
+import (
+	"context"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/statementbuilder"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// NewFactory builds the audit statement builder for builds that do not ship
+// audit logging. Audit records are written by an enterprise-only auditor, so
+// there is nothing for a query to read and it is refused rather than served
+// from the plain logs tables.
+func NewFactory(_ telemetrytypes.MetadataStore, _ flagger.Flagger) factory.ProviderFactory[qbtypes.StatementBuilder[qbtypes.LogAggregation], statementbuilder.Config] {
+	return factory.NewProviderFactory(
+		factory.MustNewName("noop"),
+		func(_ context.Context, _ factory.ProviderSettings, _ statementbuilder.Config) (qbtypes.StatementBuilder[qbtypes.LogAggregation], error) {
+			return statementBuilder{}, nil
+		},
+	)
+}
+
+type statementBuilder struct{}
+
+func (statementBuilder) Build(_ context.Context, _ valuer.UUID, _, _ uint64, _ qbtypes.RequestType, _ qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], _ map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
+	return nil, errors.Newf(errors.TypeInvalidInput, errors.CodeUnsupported, "audit logs are not available in this build")
+}

pkg/telemetrymetadata/family_values_test.go

@@ -15,45 +15,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// The flagger provider registration is process-global and keyed by provider
-// name, so each flagger must be used before the next one is created.
-func TestFamilyValueNames(t *testing.T) {
-	selector := &telemetrytypes.FieldValueSelector{
-		FieldKeySelector: &telemetrytypes.FieldKeySelector{Name: "deployment.environment"},
-	}
-
-	off := &telemetryMetaStore{fl: flaggertest.WithBooleanFlags(t, map[string]bool{})}
-	assert.Equal(t,
-		[]string{"deployment.environment"},
-		off.familyValueNames(context.Background(), valuer.UUID{}, telemetrytypes.SignalLogs, selector),
-		"the flag default keeps values literal")
-
-	on := &telemetryMetaStore{fl: flaggertest.WithBooleanFlags(t, map[string]bool{
-		flagger.FeatureResolveSemconvFamilies.String(): true,
-	})}
-	assert.Equal(t,
-		[]string{"deployment.environment.name", "deployment.environment"},
-		on.familyValueNames(context.Background(), valuer.UUID{}, telemetrytypes.SignalLogs, selector),
-		"values for one spelling must cover the whole family")
-	assert.Equal(t,
-		[]string{"deployment.environment.name", "deployment.environment"},
-		on.familyValueNames(context.Background(), valuer.UUID{}, telemetrytypes.SignalMetrics, selector),
-		"metric values cover the family members")
-	spanMetric := &telemetrytypes.FieldValueSelector{
-		FieldKeySelector: &telemetrytypes.FieldKeySelector{
-			Name:          "deployment.environment",
-			MetricContext: &telemetrytypes.MetricContext{MetricName: "signoz_calls_total"},
-		},
-	}
-	assert.Equal(t,
-		[]string{
-			"deployment.environment.name", "resource_deployment.environment.name",
-			"deployment.environment", "resource_deployment.environment",
-		},
-		on.familyValueNames(context.Background(), valuer.UUID{}, telemetrytypes.SignalMetrics, spanMetric),
-		"span-metrics values cover the resource_ spellings too")
-}
-
 // A family condition on the related values table follows the shared guard
 // rule: the operator applies to the current-first merge, a positive operator
 // takes the presence guard, and a negative operator keeps the keyless rows.

pkg/telemetrymetadata/logs.go

@@ -0,0 +1,356 @@
+package telemetrymetadata
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/huandu/go-sqlbuilder"
+	"golang.org/x/exp/maps"
+
+	"github.com/SigNoz/signoz/pkg/clickhousesql"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/telemetrymetadata/parser"
+	"github.com/SigNoz/signoz/pkg/telemetrymetadata/resolvercore"
+	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
+	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
+	"github.com/SigNoz/signoz/pkg/types/featuretypes"
+	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// logsTblStatementToFieldKeys returns materialised attribute/resource/scope keys from the logs table.
+func (t *telemetryMetaStore) logsTblStatementToFieldKeys(ctx context.Context) ([]*telemetrytypes.TelemetryFieldKey, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "logsTblStatementToFieldKeys",
+	})
+	query := fmt.Sprintf("SHOW CREATE TABLE %s.%s", t.logsDBName, t.logsV2TblName)
+	statements := []telemetrytypes.ShowCreateTableStatement{}
+	err := t.telemetrystore.ClickhouseDB().Select(ctx, &statements, query)
+	if err != nil {
+		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, resolvercore.ErrFailedToGetTblStatement.Error())
+	}
+
+	materialisedKeys, err := parser.ExtractFieldKeysFromTblStatement(statements[0].Statement)
+	if err != nil {
+		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
+	}
+
+	for idx := range materialisedKeys {
+		materialisedKeys[idx].Signal = telemetrytypes.SignalLogs
+	}
+
+	return materialisedKeys, nil
+}
+
+// getLogsKeys returns the keys from the spans that match the field selection criteria.
+func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, orgID valuer.UUID, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getLogsKeys",
+	})
+
+	if len(fieldKeySelectors) == 0 {
+		return nil, true, nil
+	}
+
+	// pre-fetch the materialised keys from the logs table
+	matKeys, err := t.logsTblStatementToFieldKeys(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
+	}
+
+	// queries for both attribute and resource keys tables
+	var queries []string
+	var allArgs []any
+
+	// tables to query based on field selectors
+	queryAttributeTable := false
+	queryResourceTable := false
+	queryBodyTable := false
+
+	for _, selector := range fieldKeySelectors {
+		if selector.FieldContext == telemetrytypes.FieldContextUnspecified {
+			// unspecified context, query all tables
+			queryAttributeTable = true
+			queryResourceTable = true
+			queryBodyTable = true
+			break
+		} else if selector.FieldContext == telemetrytypes.FieldContextAttribute {
+			queryAttributeTable = true
+		} else if selector.FieldContext == telemetrytypes.FieldContextResource {
+			queryResourceTable = true
+		} else if selector.FieldContext == telemetrytypes.FieldContextBody {
+			queryBodyTable = true
+		}
+	}
+
+	// body keys are gated behind the feature flag
+	queryBodyTable = queryBodyTable && t.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID))
+
+	// requestedFieldKeySelectors is the set of names the user explicitly asked for.
+	// Used to ensure a name that is both a parent path AND a directly requested field still surfaces
+	// in the result keys (e.g. "a.b[].c" is a parent of "a.b[].c[].d" but also a queried field).
+	mapOfRequestedSelectors := make(map[string]bool)
+	for _, sel := range fieldKeySelectors {
+		if sel.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+			mapOfRequestedSelectors[sel.Name] = true
+		}
+	}
+
+	// pre-compute parent array path names from body selectors for JSON plan building;
+	// these will be fetched as a separate UNION arm filtered to ArrayJSON/ArrayDynamic only.
+	parentPaths := make(map[string]bool)
+	if queryBodyTable {
+		for _, sel := range fieldKeySelectors {
+			if sel.FieldContext != telemetrytypes.FieldContextBody &&
+				sel.FieldContext != telemetrytypes.FieldContextUnspecified {
+				continue
+			}
+			key := &telemetrytypes.TelemetryFieldKey{
+				Name:         sel.Name,
+				Signal:       telemetrytypes.SignalLogs,
+				FieldContext: telemetrytypes.FieldContextBody,
+			}
+			if !key.KeyNameContainsArray() {
+				continue
+			}
+			for _, parent := range key.ArrayParentPaths() {
+				parentPaths[parent] = true
+			}
+		}
+	}
+
+	// attribute and resource tables share identical schema (name/datatype columns, lower-cased select,
+	// TagDataType encoding) — only the table name and field context differ.
+	addTagTableQuery := func(tblName string, fieldContext telemetrytypes.FieldContext) {
+		sb := sqlbuilder.Select(
+			"name AS tag_key",
+			fmt.Sprintf("'%s' AS tag_type", fieldContext.TagType()),
+			"lower(datatype) AS tag_data_type", // logs had historical mixed-case data
+			fmt.Sprintf("%d AS priority", resolvercore.PriorityForContext(fieldContext)),
+		).From(tblName)
+
+		conds := []string{}
+		for _, sel := range fieldKeySelectors {
+			// Include this selector if:
+			// 1. It has unspecified context (matches all tables)
+			// 2. Its context matches the current table's context
+			if sel.FieldContext != telemetrytypes.FieldContextUnspecified && sel.FieldContext != fieldContext {
+				continue
+			}
+			fieldKeyConds := []string{}
+			if sel.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+				fieldKeyConds = append(fieldKeyConds, sb.E("name", sel.Name))
+			} else {
+				fieldKeyConds = append(fieldKeyConds, sb.ILike("name", "%"+clickhousesql.LikePattern(sel.Name)+"%"))
+			}
+			if sel.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+				fieldKeyConds = append(fieldKeyConds, sb.E("datatype", sel.FieldDataType.TagDataType()))
+			}
+			if len(fieldKeyConds) > 0 {
+				conds = append(conds, sb.And(fieldKeyConds...))
+			}
+		}
+		if len(conds) > 0 {
+			sb.Where(sb.Or(conds...))
+		}
+		sb.GroupBy("name", "datatype")
+		query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+		queries = append(queries, query)
+		allArgs = append(allArgs, args...)
+	}
+
+	if queryAttributeTable {
+		addTagTableQuery(t.logsDBName+"."+t.logAttributeKeysTblName, telemetrytypes.FieldContextAttribute)
+	}
+	if queryResourceTable {
+		addTagTableQuery(t.logsDBName+"."+t.logResourceKeysTblName, telemetrytypes.FieldContextResource)
+	}
+
+	// body context uses a different table/schema (field_name, field_data_type) and requires
+	// signal+context base filters. It also fetches parent array container types (ArrayJSON/ArrayDynamic)
+	// needed for JSON access plan building.
+	if queryBodyTable {
+		sb := sqlbuilder.Select(
+			"field_name AS tag_key",
+			fmt.Sprintf("'%s' AS tag_type", telemetrytypes.FieldContextBody.TagType()),
+			"field_data_type AS tag_data_type",
+			fmt.Sprintf("%d AS priority", resolvercore.PriorityForContext(telemetrytypes.FieldContextBody)),
+		).From(fmt.Sprintf("%s.%s", DBName, FieldKeysTable))
+
+		sb.Where(sb.E("signal", telemetrytypes.SignalLogs.StringValue()))
+		sb.Where(sb.E("field_context", telemetrytypes.FieldContextBody.StringValue()))
+
+		branches := []string{}
+		for _, sel := range fieldKeySelectors {
+			if sel.FieldContext != telemetrytypes.FieldContextUnspecified && sel.FieldContext != telemetrytypes.FieldContextBody {
+				continue
+			}
+			fieldKeyConds := []string{}
+			if sel.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+				fieldKeyConds = append(fieldKeyConds, sb.E("field_name", sel.Name))
+			} else {
+				fieldKeyConds = append(fieldKeyConds, sb.ILike("field_name", "%"+clickhousesql.LikePattern(sel.Name)+"%"))
+			}
+			if sel.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+				fieldKeyConds = append(fieldKeyConds, sb.E("field_data_type", sel.FieldDataType.StringValue()))
+			}
+			if len(fieldKeyConds) > 0 {
+				branches = append(branches, sb.And(fieldKeyConds...))
+			}
+		}
+		if len(parentPaths) > 0 {
+			names := make([]any, 0, len(parentPaths))
+			for n := range parentPaths {
+				names = append(names, n)
+			}
+			branches = append(branches, sb.And(
+				sb.In("field_name", names...),
+				sb.In("field_data_type",
+					telemetrytypes.FieldDataTypeArrayDynamic.StringValue(),
+					telemetrytypes.FieldDataTypeArrayJSON.StringValue(),
+				),
+			))
+		}
+		if len(branches) > 0 {
+			sb.Where(sb.Or(branches...))
+		}
+		sb.GroupBy("field_name", "field_data_type")
+		query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+		queries = append(queries, query)
+		allArgs = append(allArgs, args...)
+	}
+
+	var limit int
+	for _, fieldKeySelector := range fieldKeySelectors {
+		limit += fieldKeySelector.Limit
+	}
+	if limit == 0 {
+		limit = 1000
+	}
+
+	keys := []*telemetrytypes.TelemetryFieldKey{}
+	parentTypes := make(map[string][]telemetrytypes.FieldDataType)
+	rowCount := 0
+
+	// the log and scope contexts have no keys table; they are served by the
+	// static fields appended below
+	if len(queries) > 0 {
+		// Combine queries with UNION ALL
+		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, ErrFailedToGetLogsKeys.Error())
+		}
+		defer rows.Close()
+
+		for rows.Next() {
+			rowCount++
+			// reached the limit, we know there are more results
+			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, ErrFailedToGetLogsKeys.Error())
+			}
+
+			// ArrayJSON/ArrayDynamic body rows for parent paths are needed by the JSON access plan
+			// builder (enrichJSONKeys). Always record them in parentTypes. Only skip adding to keys
+			// if the user did not also directly request this name — a field like "education" can be
+			// both a parent of "education[].name" and an explicitly queried field in its own right.
+			switch fieldDataType {
+			case telemetrytypes.FieldDataTypeArrayJSON, telemetrytypes.FieldDataTypeArrayDynamic:
+				if fieldContext == telemetrytypes.FieldContextBody && parentPaths[name] {
+					parentTypes[name] = append(parentTypes[name], fieldDataType)
+					if !mapOfRequestedSelectors[name] {
+						continue // skip; don't register the key.
+					}
+				}
+			}
+
+			key, ok := mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
+
+			// if there is no materialised column, create a key with the field context and data type
+			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, ErrFailedToGetLogsKeys.Error())
+		}
+	}
+
+	// hit the limit? (only counting DB results)
+	complete := rowCount <= limit
+
+	// Add the matching intrinsic columns. These don't count towards the limit
+	for _, field := range maps.Values(logstelemetryschema.IntrinsicFields) {
+		if !resolvercore.StaticFieldMatchesAny(field, fieldKeySelectors) {
+			continue
+		}
+		if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; added {
+			continue
+		}
+		keys = append(keys, &field)
+	}
+
+	// enrich body keys with promoted paths, indexes, and JSON access plans
+	if t.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
+		if err := t.enrichJSONKeys(ctx, fieldKeySelectors, keys, parentTypes); err != nil {
+			return nil, false, err
+		}
+	}
+
+	if err := t.updateColumnEvolutionMetadataForKeys(ctx, keys); err != nil {
+		return nil, false, err
+	}
+
+	return keys, complete, nil
+}
+
+func (t *telemetryMetaStore) getLogFieldValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getLogFieldValues",
+	})
+
+	knownBool := resolvercore.IsKnownBoolField(fieldValueSelector, logstelemetryschema.IntrinsicFields)
+	names := resolvercore.FamilyValueNames(ctx, orgID, t.fl, telemetrytypes.SignalLogs, fieldValueSelector)
+	// unix_milli is the hour the log was ingested, not the log's own timestamp
+	return resolvercore.TagTableValues(ctx, t.telemetrystore, t.logsDBName+"."+t.logsFieldsTblName, fieldValueSelector, names, knownBool)
+}

pkg/telemetrymetadata/metadata_query_test.go

@@ -38,6 +38,7 @@ func TestGetFirstSeenFromMetricMetadata(t *testing.T) {
 		instrumentationtest.New().ToProviderSettings(),
 		mockTelemetryStore,
 		flaggertest.New(t),
+		nil,
 	)
 
 	lookupKeys := []telemetrytypes.MetricMetadataLookupKey{

pkg/telemetrymetadata/meter.go

@@ -0,0 +1,215 @@
+package telemetrymetadata
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/huandu/go-sqlbuilder"
+
+	"github.com/SigNoz/signoz/pkg/clickhousesql"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
+	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+// getMeterKeys returns the keys from the meter metrics that match the field selection criteria.
+func (t *telemetryMetaStore) getMeterSourceMetricKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getMeterSourceMetricKeys",
+	})
+
+	if len(fieldKeySelectors) == 0 {
+		return nil, true, nil
+	}
+
+	sb := sqlbuilder.Select("DISTINCT arrayJoin(JSONExtractKeys(labels)) as attr_name").From(t.meterDBName + "." + t.meterFieldsTblName)
+	conds := []string{}
+	var limit int
+	for _, fieldKeySelector := range fieldKeySelectors {
+		fieldConds := []string{}
+		if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+			fieldConds = append(fieldConds, sb.E("attr_name", fieldKeySelector.Name))
+		} else {
+			fieldConds = append(fieldConds, sb.Like("attr_name", "%"+fieldKeySelector.Name+"%"))
+		}
+		fieldConds = append(fieldConds, sb.NotLike("attr_name", "\\_\\_%"))
+
+		if fieldKeySelector.MetricContext != nil {
+			if fieldKeySelector.MetricContext.MetricName != "" {
+				fieldConds = append(fieldConds, sb.E("metric_name", fieldKeySelector.MetricContext.MetricName))
+			}
+			if fieldKeySelector.MetricContext.MetricNamespace != "" {
+				fieldConds = append(fieldConds, sb.Like("metric_name", clickhousesql.LikePattern(fieldKeySelector.MetricContext.MetricNamespace)+"%"))
+			}
+		}
+
+		conds = append(conds, sb.And(fieldConds...))
+		limit += fieldKeySelector.Limit
+	}
+	sb.Where(sb.Or(conds...))
+	if limit == 0 {
+		limit = 1000
+	}
+
+	sb.Limit(limit)
+	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, ErrFailedToGetMeterKeys.Error())
+	}
+	defer rows.Close()
+
+	keys := []*telemetrytypes.TelemetryFieldKey{}
+	rowCount := 0
+	for rows.Next() {
+		rowCount++
+		// reached the limit, we know there are more results
+		if rowCount > limit {
+			break
+		}
+
+		var name string
+		err = rows.Scan(&name)
+		if err != nil {
+			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterKeys.Error())
+		}
+		// meter labels are stored as strings in the labels JSON and have no
+		// attribute context, so only the data type is known
+		keys = append(keys, &telemetrytypes.TelemetryFieldKey{
+			Name:          name,
+			Signal:        telemetrytypes.SignalMetrics,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+		})
+	}
+
+	if rows.Err() != nil {
+		return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterKeys.Error())
+	}
+
+	// hit the limit?
+	complete := rowCount <= limit
+
+	return keys, complete, nil
+
+}
+
+func (t *telemetryMetaStore) getMeterSourceMetricFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getMeterSourceMetricFieldValues",
+	})
+	sb := sqlbuilder.Select("DISTINCT arrayJoin(JSONExtractKeysAndValues(labels, 'String')) AS attr").
+		From(t.meterDBName + "." + t.meterFieldsTblName)
+
+	if fieldValueSelector.Name != "" {
+		sb.Where(sb.E("attr.1", fieldValueSelector.Name))
+	}
+	sb.Where(sb.NotLike("attr.1", "\\_\\_%"))
+
+	if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
+		sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
+	}
+	if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
+		sb.Where(sb.Like("metric_name", clickhousesql.LikePattern(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
+	}
+
+	if fieldValueSelector.Value != "" {
+		if fieldValueSelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+			sb.Where(sb.E("attr.2", fieldValueSelector.Value))
+		} else {
+			sb.Where(sb.Like("attr.2", "%"+fieldValueSelector.Value+"%"))
+		}
+	}
+	sb.Where(sb.NE("attr.2", ""))
+
+	limit := fieldValueSelector.Limit
+	if limit == 0 {
+		limit = 50
+	}
+	// query one extra to check if we hit the limit
+	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, ErrFailedToGetMeterValues.Error())
+	}
+	defer rows.Close()
+
+	values := &telemetrytypes.TelemetryFieldValues{}
+	rowCount := 0
+	for rows.Next() {
+		rowCount++
+		// reached the limit, we know there are more results
+		if rowCount > limit {
+			break
+		}
+
+		var attribute []string
+		if err := rows.Scan(&attribute); err != nil {
+			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterValues.Error())
+		}
+		if len(attribute) > 1 {
+			values.StringValues = append(values.StringValues, attribute[1])
+		}
+	}
+
+	// hit the limit?
+	complete := rowCount <= limit
+	return values, complete, nil
+}
+
+func (t *telemetryMetaStore) fetchMeterSourceMetricsTemporalityAndType(ctx context.Context, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "fetchMeterSourceMetricsTemporalityAndType",
+	})
+	temporalities := make(map[string]metrictypes.Temporality)
+	types := make(map[string]metrictypes.Type)
+
+	sb := sqlbuilder.Select(
+		"metric_name",
+		"argMax(temporality, unix_milli) as temporality",
+		"any(type) AS type",
+		"argMax(is_monotonic, unix_milli) as is_monotonic",
+	).From(t.meterDBName + "." + t.meterFieldsTblName)
+
+	// Filter by metric names (in the temporality column due to data mix-up)
+	sb.Where(sb.In("metric_name", metricNames))
+
+	// Group by metric name to get one temporality per metric
+	sb.GroupBy("metric_name")
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	t.logger.DebugContext(ctx, "fetching meter metrics temporality", slog.String("query", query), slog.Any("args", args))
+
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to fetch meter metric temporality")
+	}
+	defer rows.Close()
+
+	// Process results
+	for rows.Next() {
+		var metricName string
+		var temporality metrictypes.Temporality
+		var metricType metrictypes.Type
+		var isMonotonic bool
+		if err := rows.Scan(&metricName, &temporality, &metricType, &isMonotonic); err != nil {
+			return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan temporality result")
+		}
+		metricType = resolveMetricType(metricType, isMonotonic, temporality)
+		temporalities[metricName] = temporality
+		types[metricName] = metricType
+	}
+
+	return temporalities, types, nil
+}

pkg/telemetrymetadata/metrics.go

@@ -0,0 +1,651 @@
+package telemetrymetadata
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/huandu/go-sqlbuilder"
+
+	"github.com/SigNoz/signoz/pkg/clickhousesql"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetrymetadata/resolvercore"
+	"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
+	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
+	"github.com/SigNoz/signoz/pkg/types/featuretypes"
+	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// chunkSizeFirstSeenMetricMetadata limits the number of tuples per SQL query to avoid hitting the max_query_size limit.
+//
+// Calculation Logic:
+//
+//  1. The Ceiling: 250 KB (256,000 bytes). A conservative safety limit set just below the common DB max_query_size (262KB)
+//     to guarantee the database does not reject the query.
+//     Reference: https://clickhouse.com/docs/operations/settings/settings#max_query_size
+//
+//  2. Unit Cost: ~150 bytes per tuple. The estimated "weight" of a single lookup key, summing MetricName (40B),
+//     AttrName (30B), AttrValue (64B), and SQL syntax overhead (16B).
+//
+//  3. Theoretical Max: ~1,706 tuples. The absolute maximum capacity if all keys adhere to the average size (256,000 / 150).
+//
+//  4. Final Limit: 1600. Rounds down to provide a ~6% safety buffer, accounting for data outliers
+//     (unusually long attribute values) and multi-byte character expansion.
+const chunkSizeFirstSeenMetricMetadata = 1600
+
+// getMetricsKeys returns the keys from the metrics that match the field selection criteria.
+func (t *telemetryMetaStore) getMetricsKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getMetricsKeys",
+	})
+	if len(fieldKeySelectors) == 0 {
+		return nil, true, nil
+	}
+
+	sb := sqlbuilder.
+		Select("attr_name as name", "attr_type as field_context", "attr_datatype as field_data_type", `
+			CASE
+				WHEN attr_type = 'resource' THEN 1
+				WHEN attr_type = 'scope' THEN 2
+				WHEN attr_type = 'point' THEN 3
+				ELSE 4
+			END as priority`).
+		From(t.metricsDBName + "." + t.metricsFieldsTblName)
+
+	var limit int
+
+	conds := []string{}
+	for _, fieldKeySelector := range fieldKeySelectors {
+		fieldConds := []string{}
+		if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+			fieldConds = append(fieldConds, sb.E("attr_name", fieldKeySelector.Name))
+		} else {
+			fieldConds = append(fieldConds, sb.ILike("attr_name", "%"+clickhousesql.LikePattern(fieldKeySelector.Name)+"%"))
+		}
+		fieldConds = append(fieldConds, sb.NotLike("attr_name", "\\_\\_%"))
+
+		// note: type and datatype do not have much significance in metrics
+
+		// if fieldKeySelector.FieldContext != telemetrytypes.FieldContextUnspecified {
+		// 	fieldConds = append(fieldConds, sb.E("attr_type", fieldKeySelector.FieldContext.TagType()))
+		// }
+
+		// if fieldKeySelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+		// 	fieldConds = append(fieldConds, sb.E("attr_datatype", fieldKeySelector.FieldDataType.TagDataType()))
+		// }
+
+		if fieldKeySelector.MetricContext != nil {
+			if fieldKeySelector.MetricContext.MetricName != "" {
+				fieldConds = append(fieldConds, sb.E("metric_name", fieldKeySelector.MetricContext.MetricName))
+			}
+			if fieldKeySelector.MetricContext.MetricNamespace != "" {
+				fieldConds = append(fieldConds, sb.Like("metric_name", clickhousesql.LikePattern(fieldKeySelector.MetricContext.MetricNamespace)+"%"))
+			}
+		}
+
+		conds = append(conds, sb.And(fieldConds...))
+		limit += fieldKeySelector.Limit
+	}
+	sb.Where(sb.Or(conds...))
+	sb.GroupBy("name", "field_context", "field_data_type")
+
+	if limit == 0 {
+		limit = 1000
+	}
+
+	mainSb := sqlbuilder.Select("name", "field_context", "field_data_type", "max(priority) as priority")
+	mainSb.From(mainSb.BuilderAs(sb, "sub_query"))
+	mainSb.GroupBy("name", "field_context", "field_data_type")
+	mainSb.OrderBy("priority")
+	// query one extra to check if we hit the limit
+	mainSb.Limit(limit + 1)
+
+	query, args := mainSb.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, ErrFailedToGetMetricsKeys.Error())
+	}
+	defer rows.Close()
+
+	keys := []*telemetrytypes.TelemetryFieldKey{}
+	rowCount := 0
+	for rows.Next() {
+		rowCount++
+		// reached the limit, we know there are more results
+		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, ErrFailedToGetMetricsKeys.Error())
+		}
+		keys = append(keys, &telemetrytypes.TelemetryFieldKey{
+			Name:          name,
+			Signal:        telemetrytypes.SignalMetrics,
+			FieldContext:  fieldContext,
+			FieldDataType: fieldDataType,
+		})
+	}
+
+	if rows.Err() != nil {
+		return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
+	}
+
+	// hit the limit?
+	complete := rowCount <= limit
+
+	return keys, complete, nil
+}
+
+// getMetricFieldValues returns field values and whether the result is complete.
+func (t *telemetryMetaStore) getMetricFieldValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getMetricFieldValues",
+	})
+	limit := fieldValueSelector.Limit
+	if limit == 0 {
+		limit = 50
+	}
+
+	values, err := t.getIntrinsicMetricFieldValues(ctx, orgID, fieldValueSelector, limit)
+	if err != nil {
+		return nil, false, err
+	}
+
+	var remainingLimit int
+	if values == nil {
+		values = &telemetrytypes.TelemetryFieldValues{}
+		remainingLimit = limit
+	} else {
+		remainingLimit = limit - values.NumValues()
+	}
+
+	if remainingLimit == 0 {
+		return values, values.NumValues() < limit, nil
+	}
+
+	sb := sqlbuilder.
+		Select("DISTINCT attr_string_value").
+		From(t.metricsDBName + "." + t.metricsFieldsTblName)
+
+	if fieldValueSelector.Name != "" {
+		if names := resolvercore.FamilyValueNames(ctx, orgID, t.fl, telemetrytypes.SignalMetrics, fieldValueSelector); len(names) > 1 {
+			sb.Where(sb.In("attr_name", sqlbuilder.List(names)))
+		} else {
+			sb.Where(sb.E("attr_name", fieldValueSelector.Name))
+		}
+	}
+
+	if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
+		sb.Where(sb.E("attr_type", fieldValueSelector.FieldContext.TagType()))
+	}
+
+	if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+		sb.Where(sb.E("attr_datatype", fieldValueSelector.FieldDataType.TagDataType()))
+	}
+
+	if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
+		if metricNames := querybuilder.FamilyMetricNames(ctx, orgID, t.fl, fieldValueSelector.MetricContext.MetricName); len(metricNames) > 1 {
+			sb.Where(sb.In("metric_name", sqlbuilder.List(metricNames)))
+		} else {
+			sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
+		}
+	}
+	if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
+		sb.Where(sb.Like("metric_name", clickhousesql.LikePattern(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
+	}
+
+	if fieldValueSelector.StartUnixMilli > 0 {
+		sb.Where(sb.GE("last_reported_unix_milli", fieldValueSelector.StartUnixMilli))
+	}
+
+	if fieldValueSelector.EndUnixMilli > 0 {
+		sb.Where(sb.LE("first_reported_unix_milli", fieldValueSelector.EndUnixMilli))
+	}
+
+	if fieldValueSelector.Value != "" {
+		if fieldValueSelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+			sb.Where(sb.E("attr_string_value", fieldValueSelector.Value))
+		} else {
+			sb.Where(sb.ILike("attr_string_value", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
+		}
+	}
+	// query one extra to check if we hit the limit
+	sb.Limit(remainingLimit + 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, ErrFailedToGetMetricsKeys.Error())
+	}
+	defer rows.Close()
+
+	rowCount := 0
+	for rows.Next() {
+		rowCount++
+		if rowCount > remainingLimit {
+			break
+		}
+
+		var stringValue string
+		if err := rows.Scan(&stringValue); err != nil {
+			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
+		}
+		values.StringValues = append(values.StringValues, stringValue)
+	}
+	// hit the limit?
+	complete := values.NumValues() < limit // decision on total limit
+	return values, complete, nil
+}
+
+// getIntrinsicMetricFieldValues returns values, isSearchComplete, error.
+func (t *telemetryMetaStore) getIntrinsicMetricFieldValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *telemetrytypes.FieldValueSelector, limit int) (*telemetrytypes.TelemetryFieldValues, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getIntrinsicMetricFieldValues",
+	})
+	key, ok := metricstelemetryschema.IntrinsicMetricFieldDefinitions[fieldValueSelector.Name]
+	if !ok {
+		return &telemetrytypes.TelemetryFieldValues{}, nil
+	}
+
+	if fieldValueSelector.Signal != telemetrytypes.SignalMetrics && fieldValueSelector.Signal != telemetrytypes.SignalUnspecified {
+		return &telemetrytypes.TelemetryFieldValues{}, nil
+	}
+
+	// if the field context or data type does not match for the requested key (and is not unspecified as well),
+	// we should return false to enable searching in metadata table.
+	if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified && fieldValueSelector.FieldContext != key.FieldContext {
+		return &telemetrytypes.TelemetryFieldValues{}, nil
+	}
+	if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified && fieldValueSelector.FieldDataType != key.FieldDataType {
+		return &telemetrytypes.TelemetryFieldValues{}, nil
+	}
+
+	// no values are surfaced for intrinsic boolean fields.
+	if key.FieldDataType == telemetrytypes.FieldDataTypeBool {
+		return &telemetrytypes.TelemetryFieldValues{}, nil
+	}
+
+	// Reduced metrics live in time_series_v4_reduced + the buffer (is_reduced=false), not the raw
+	// tables; query those only when reduction is enabled (the tables may not exist otherwise).
+	reductionEnabled := t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID))
+	sources := []struct {
+		tableName  string
+		extraConds []string
+	}{
+		{metricstelemetryschema.TimeseriesV41weekTableName, nil},
+	}
+	if reductionEnabled {
+		sources = append(sources,
+			struct {
+				tableName  string
+				extraConds []string
+			}{metricstelemetryschema.TimeseriesV4BufferTableName, []string{"is_reduced = false"}},
+			struct {
+				tableName  string
+				extraConds []string
+			}{metricstelemetryschema.TimeseriesV4ReducedTableName, nil},
+		)
+	}
+
+	values := &telemetrytypes.TelemetryFieldValues{}
+	seen := make(map[string]struct{})
+	for _, src := range sources {
+		if len(values.StringValues) >= limit {
+			break
+		}
+		tableValues, err := t.getIntrinsicMetricFieldValuesForTable(ctx, fieldValueSelector, key, src.tableName, limit, src.extraConds...)
+		if err != nil {
+			return nil, err
+		}
+		for _, str := range tableValues {
+			if _, ok := seen[str]; ok {
+				continue
+			}
+			seen[str] = struct{}{}
+			values.StringValues = append(values.StringValues, str)
+			if len(values.StringValues) >= limit {
+				break
+			}
+		}
+	}
+	return values, nil
+}
+
+// getIntrinsicMetricFieldValuesForTable returns the distinct intrinsic field values for the
+// given key from a single time series table. extraConds are appended verbatim (e.g. the
+// buffer's "is_reduced = false" filter).
+func (t *telemetryMetaStore) getIntrinsicMetricFieldValuesForTable(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector, key telemetrytypes.TelemetryFieldKey, tableName string, limit int, extraConds ...string) ([]string, error) {
+	sb := sqlbuilder.Select(sqlbuilder.Escape(key.Name)).
+		From(t.metricsDBName + "." + tableName)
+
+	if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
+		sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
+	}
+	if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
+		sb.Where(sb.Like("metric_name", clickhousesql.LikePattern(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
+	}
+
+	if fieldValueSelector.StartUnixMilli > 0 {
+		sb.Where(sb.GE("unix_milli", fieldValueSelector.StartUnixMilli))
+	}
+
+	if fieldValueSelector.EndUnixMilli > 0 {
+		sb.Where(sb.LE("unix_milli", fieldValueSelector.EndUnixMilli))
+	}
+
+	if fieldValueSelector.Value != "" {
+		if fieldValueSelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+			sb.Where(sb.E(key.Name, fieldValueSelector.Value))
+		} else {
+			sb.Where(sb.ILike(key.Name, "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
+		}
+	}
+
+	for _, cond := range extraConds {
+		sb.Where(cond)
+	}
+
+	sb.GroupBy(sqlbuilder.Escape(key.Name))
+	sb.Limit(limit + 1)
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
+	}
+	defer rows.Close()
+
+	var values []string
+	rowCount := 0
+	for rows.Next() {
+		rowCount++
+		if rowCount > limit {
+			break
+		}
+
+		var str string
+		if err := rows.Scan(&str); err != nil {
+			return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
+		}
+		values = append(values, str)
+	}
+	return values, nil
+}
+
+func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, reductionEnabled bool, metricNames ...string) (map[string][]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "fetchMetricsTemporalityAndType",
+	})
+	adjustedStartTs, adjustedEndTs, tsTableName, _ := metricstelemetryschema.WhichTSTableToUse(queryTimeRangeStartTs, queryTimeRangeEndTs, false, nil)
+
+	temporalities, types, err := t.fetchTemporalityTypeForTable(ctx, tsTableName, adjustedStartTs, adjustedEndTs, metricNames)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	reduced := make(map[string]bool)
+	if reductionEnabled {
+		bufferTemporalities, bufferTypes, err := t.fetchTemporalityTypeForTable(ctx, metricstelemetryschema.TimeseriesV4BufferTableName, adjustedStartTs, adjustedEndTs, metricNames, "is_reduced = false")
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		for metricName, bufferTemps := range bufferTemporalities {
+			temporalities[metricName] = unionTemporalities(temporalities[metricName], bufferTemps)
+		}
+		for metricName, metricType := range bufferTypes {
+			if _, ok := types[metricName]; !ok {
+				types[metricName] = metricType
+			}
+		}
+
+		reducedTemporalities, reducedTypes, err := t.fetchTemporalityTypeForTable(ctx, metricstelemetryschema.TimeseriesV4ReducedTableName, adjustedStartTs, adjustedEndTs, metricNames)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		for metricName := range reducedTypes {
+			reduced[metricName] = true
+		}
+		for metricName, reducedTemps := range reducedTemporalities {
+			if len(temporalities[metricName]) == 0 {
+				temporalities[metricName] = reducedTemps
+			}
+		}
+		for metricName, metricType := range reducedTypes {
+			if _, ok := types[metricName]; !ok {
+				types[metricName] = metricType
+			}
+		}
+	}
+
+	return temporalities, types, reduced, nil
+}
+
+// unionTemporalities appends the distinct temporalities from additional to existing.
+func unionTemporalities(existing, additional []metrictypes.Temporality) []metrictypes.Temporality {
+	for _, t := range additional {
+		found := false
+		for _, e := range existing {
+			if e == t {
+				found = true
+				break
+			}
+		}
+		if !found {
+			existing = append(existing, t)
+		}
+	}
+	return existing
+}
+
+func (t *telemetryMetaStore) fetchTemporalityTypeForTable(ctx context.Context, tableName string, adjustedStartTs, adjustedEndTs uint64, metricNames []string, extraConds ...string) (map[string][]metrictypes.Temporality, map[string]metrictypes.Type, error) {
+	temporalities := make(map[string][]metrictypes.Temporality)
+	types := make(map[string]metrictypes.Type)
+
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select("metric_name", "temporality", "any(type) AS type", "argMax(is_monotonic, unix_milli) as is_monotonic")
+	sb.From(t.metricsDBName + "." + tableName)
+	conds := []string{
+		sb.In("metric_name", metricNames),
+		sb.GTE("unix_milli", adjustedStartTs),
+		sb.LT("unix_milli", adjustedEndTs),
+	}
+	conds = append(conds, extraConds...)
+	sb.Where(conds...)
+	sb.GroupBy("metric_name", "temporality")
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	t.logger.DebugContext(ctx, "fetching metric temporality", slog.String("query", query), slog.Any("args", args))
+
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to fetch metric temporality")
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var metricName string
+		var temporality metrictypes.Temporality
+		var metricType metrictypes.Type
+		var isMonotonic bool
+		if err := rows.Scan(&metricName, &temporality, &metricType, &isMonotonic); err != nil {
+			return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan temporality result")
+		}
+		if temporality != metrictypes.Unknown {
+			temporalities[metricName] = append(temporalities[metricName], temporality)
+		}
+		metricType = resolveMetricType(metricType, isMonotonic, temporality)
+		types[metricName] = metricType
+	}
+	if err := rows.Err(); err != nil {
+		return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "error iterating over metrics temporality rows")
+	}
+
+	return temporalities, types, nil
+}
+
+// GetFirstSeenFromMetricMetadata queries the metadata table to get the first_seen timestamp
+// for each metric-attribute-value combination.
+// Returns a map where key is `telemetrytypes.MetricMetadataLookupKey` and value is first_seen in milliseconds.
+func (t *telemetryMetaStore) GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []telemetrytypes.MetricMetadataLookupKey) (map[telemetrytypes.MetricMetadataLookupKey]int64, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "GetFirstSeenFromMetricMetadata",
+	})
+	result := make(map[telemetrytypes.MetricMetadataLookupKey]int64)
+
+	for i := 0; i < len(lookupKeys); i += chunkSizeFirstSeenMetricMetadata {
+		end := i + chunkSizeFirstSeenMetricMetadata
+		if end > len(lookupKeys) {
+			end = len(lookupKeys)
+		}
+		chunk := lookupKeys[i:end]
+
+		sb := sqlbuilder.Select(
+			"metric_name",
+			"attr_name",
+			"attr_string_value",
+			"min(first_reported_unix_milli) AS first_seen",
+		).From(t.metricsDBName + "." + t.metricsFieldsTblName)
+
+		lookupItems := make([]interface{}, 0, len(chunk))
+		for _, key := range chunk {
+			lookupItems = append(lookupItems, sqlbuilder.Tuple(key.MetricName, key.AttributeName, key.AttributeValue))
+		}
+		sb.Where(
+			sb.In(
+				sqlbuilder.TupleNames("metric_name", "attr_name", "attr_string_value"),
+				lookupItems...,
+			),
+		)
+		sb.GroupBy("metric_name", "attr_name", "attr_string_value")
+		sb.OrderBy("first_seen")
+
+		query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+		rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
+		if err != nil {
+			return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to query metadata for first_seen")
+		}
+
+		for rows.Next() {
+			var metricName, attrName, attrValue string
+			var firstSeen uint64
+			if err := rows.Scan(&metricName, &attrName, &attrValue, &firstSeen); err != nil {
+				rows.Close()
+				return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan metadata first_seen result")
+			}
+			result[telemetrytypes.MetricMetadataLookupKey{
+				MetricName:     metricName,
+				AttributeName:  attrName,
+				AttributeValue: attrValue,
+			}] = int64(firstSeen)
+		}
+
+		if err := rows.Err(); err != nil {
+			rows.Close()
+			return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to iterate metadata first_seen results")
+		}
+		rows.Close()
+	}
+
+	return result, nil
+}
+
+func (t *telemetryMetaStore) FetchLastSeenInfoMulti(ctx context.Context, orgID valuer.UUID, metricNames ...string) (map[string]int64, error) {
+	lastSeenInfo, err := t.fetchLastSeenInfoForTable(ctx, metricstelemetryschema.TimeseriesV4TableName, metricNames)
+	if err != nil {
+		return nil, err
+	}
+
+	if t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID)) {
+		reducedLastSeen, err := t.fetchLastSeenInfoForTable(ctx, metricstelemetryschema.TimeseriesV4ReducedTableName, metricNames)
+		if err != nil {
+			return nil, err
+		}
+		for metricName, ts := range reducedLastSeen {
+			if ts > lastSeenInfo[metricName] {
+				lastSeenInfo[metricName] = ts
+			}
+		}
+	}
+
+	return lastSeenInfo, nil
+}
+
+func (t *telemetryMetaStore) fetchLastSeenInfoForTable(ctx context.Context, tableName string, metricNames []string) (map[string]int64, error) {
+	sb := sqlbuilder.Select(
+		"metric_name",
+		"max(unix_milli)",
+	).
+		From(t.metricsDBName + "." + tableName)
+	sb.Where(sb.In("metric_name", metricNames))
+	sb.GroupBy("metric_name")
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	t.logger.DebugContext(ctx, "fetching metric last seen timestamp", slog.String("query", query), slog.Any("args", args))
+
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to fetch metric last seen info")
+	}
+	defer rows.Close()
+
+	lastSeenInfo := make(map[string]int64)
+	for rows.Next() {
+		var metricName string
+		var unix_milli int64
+		if err := rows.Scan(&metricName, &unix_milli); err != nil {
+			return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan last seen info result")
+		}
+		lastSeenInfo[metricName] = unix_milli
+	}
+	if err := rows.Err(); err != nil {
+		return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "error iterating over metrics temporality rows")
+	}
+	return lastSeenInfo, nil
+}
+
+// containsConditions compiles a contains search over the key and its family
+// spellings, which stand as their own metadata.
+func (t *telemetryMetaStore) containsConditions(ctx context.Context, q qbtypes.QueryInfo, key *telemetrytypes.TelemetryFieldKey, names []string, value string, sb *sqlbuilder.SelectBuilder) ([]string, error) {
+	// A family forms only over string keys with a signal. The table stores
+	// only string maps, so an unspecified data type is a string here.
+	dataType := key.FieldDataType
+	if dataType == telemetrytypes.FieldDataTypeUnspecified {
+		dataType = telemetrytypes.FieldDataTypeString
+	}
+	fieldKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey, len(names))
+	for _, name := range names {
+		fieldKeys[name] = []*telemetrytypes.TelemetryFieldKey{{
+			Name:          name,
+			Signal:        key.Signal,
+			FieldContext:  key.FieldContext,
+			FieldDataType: dataType,
+		}}
+	}
+	conds, _, err := querybuilder.Conditions(ctx, q, t.storage, key, qbtypes.FilterOperatorContains, value, fieldKeys, false, sb)
+	return conds, err
+}

pkg/telemetrymetadata/parser/parser.go

@@ -1,15 +1,15 @@
-package telemetrymetadata
+package parser
 
 import (
 	"strings"
 
-	"github.com/AfterShip/clickhouse-sql-parser/parser"
+	chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
 
 // TelemetryFieldVisitor is an AST visitor for extracting telemetry fields.
 type TelemetryFieldVisitor struct {
-	parser.DefaultASTVisitor
+	chparser.DefaultASTVisitor
 	Fields []*telemetrytypes.TelemetryFieldKey
 }
 
@@ -20,14 +20,14 @@ func NewTelemetryFieldVisitor() *TelemetryFieldVisitor {
 }
 
 // VisitColumnDef is called when visiting a column definition.
-func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
+func (v *TelemetryFieldVisitor) VisitColumnDef(expr *chparser.ColumnDef) error {
 	// Check if this is a materialized column with DEFAULT expression
 	if expr.DefaultExpr == nil {
 		return nil
 	}
 
 	// Parse column name to extract context and data type
-	columnName := parser.Format(expr.Name)
+	columnName := chparser.Format(expr.Name)
 
 	// Remove backticks if present
 	columnName = strings.TrimPrefix(columnName, "`")
@@ -75,7 +75,7 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
 	// Extract field name from the DEFAULT expression
 	// The DEFAULT expression should be something like: resources_string['k8s.cluster.name']
 	// We need to extract the key inside the square brackets
-	defaultExprStr := parser.Format(expr.DefaultExpr)
+	defaultExprStr := chparser.Format(expr.DefaultExpr)
 
 	// Look for the pattern: map['key']
 	startIdx := strings.Index(defaultExprStr, "['")
@@ -101,7 +101,7 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
 
 func ExtractFieldKeysFromTblStatement(statement string) ([]*telemetrytypes.TelemetryFieldKey, error) {
 	// Parse the CREATE TABLE statement using the ClickHouse parser
-	p := parser.NewParser(statement)
+	p := chparser.NewParser(statement)
 	stmts, err := p.ParseStmts()
 	if err != nil {
 		return nil, err
@@ -113,7 +113,7 @@ func ExtractFieldKeysFromTblStatement(statement string) ([]*telemetrytypes.Telem
 	// Visit each statement
 	for _, stmt := range stmts {
 		// We're looking for CreateTable statements
-		createTable, ok := stmt.(*parser.CreateTable)
+		createTable, ok := stmt.(*chparser.CreateTable)
 		if !ok {
 			continue
 		}

pkg/telemetrymetadata/parser/parser_test.go

@@ -1,4 +1,4 @@
-package telemetrymetadata
+package parser
 
 import (
 	"slices"

pkg/telemetrymetadata/resolvercore/bool_values.go

@@ -1,10 +1,9 @@
-package telemetrymetadata
+package resolvercore
 
 import (
 	"strconv"
 	"strings"
 
-	"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
 
@@ -21,27 +20,10 @@ func boolFieldValues(searchText string) *telemetrytypes.TelemetryFieldValues {
 	return values
 }
 
-// spanSearchScopeFieldValues is the suggestion set for a search-scope selector
-// (isRoot, isEntryPoint), which only filters with true. ok is false for any
-// other name.
-func spanSearchScopeFieldValues(name, searchText string) (*telemetrytypes.TelemetryFieldValues, bool) {
-	for scopeName := range tracestelemetryschema.SpanSearchScopeFields {
-		if !strings.EqualFold(scopeName, name) {
-			continue
-		}
-		values := &telemetrytypes.TelemetryFieldValues{}
-		if needle := strings.ToLower(searchText); needle == "" || strings.Contains("true", needle) {
-			values.BoolValues = []bool{true}
-		}
-		return values, true
-	}
-	return nil, false
-}
-
-// isKnownBoolField is true when the caller asked for the bool data type, or
+// IsKnownBoolField is true when the caller asked for the bool data type, or
 // when the name is one of the signal's static bool fields and the requested
 // context does not rule that static field out.
-func isKnownBoolField(selector *telemetrytypes.FieldValueSelector, staticFields ...map[string]telemetrytypes.TelemetryFieldKey) bool {
+func IsKnownBoolField(selector *telemetrytypes.FieldValueSelector, staticFields ...map[string]telemetrytypes.TelemetryFieldKey) bool {
 	if selector.FieldDataType == telemetrytypes.FieldDataTypeBool {
 		return true
 	}

pkg/telemetrymetadata/resolvercore/family_values.go

@@ -0,0 +1,29 @@
+package resolvercore
+
+import (
+	"context"
+
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/semconv"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// FamilyValueNames returns the spellings whose values merge into the
+// suggestions. With the flag off, the requested name alone.
+func FamilyValueNames(ctx context.Context, orgID valuer.UUID, fl flagger.Flagger, signal telemetrytypes.Signal, fieldValueSelector *telemetrytypes.FieldValueSelector) []string {
+	if !querybuilder.SemconvFamiliesEnabled(ctx, orgID, fl) {
+		return []string{fieldValueSelector.Name}
+	}
+	selector := telemetrytypes.FieldKeySelector{
+		Name:          fieldValueSelector.Name,
+		Signal:        signal,
+		FieldContext:  fieldValueSelector.FieldContext,
+		MetricContext: fieldValueSelector.MetricContext,
+	}
+	if signal == telemetrytypes.SignalMetrics {
+		return querybuilder.MetricLabelSpellings(selector)
+	}
+	return semconv.Members(semconv.KindAttribute, selector)
+}

pkg/telemetrymetadata/resolvercore/family_values_test.go

@@ -0,0 +1,52 @@
+package resolvercore
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// The flagger provider registration is process-global and keyed by provider
+// name, so each flagger must be used before the next one is created.
+func TestFamilyValueNames(t *testing.T) {
+	selector := &telemetrytypes.FieldValueSelector{
+		FieldKeySelector: &telemetrytypes.FieldKeySelector{Name: "deployment.environment"},
+	}
+
+	off := flaggertest.WithBooleanFlags(t, map[string]bool{})
+	assert.Equal(t,
+		[]string{"deployment.environment"},
+		FamilyValueNames(context.Background(), valuer.UUID{}, off, telemetrytypes.SignalLogs, selector),
+		"the flag default keeps values literal")
+
+	on := flaggertest.WithBooleanFlags(t, map[string]bool{
+		flagger.FeatureResolveSemconvFamilies.String(): true,
+	})
+	assert.Equal(t,
+		[]string{"deployment.environment.name", "deployment.environment"},
+		FamilyValueNames(context.Background(), valuer.UUID{}, on, telemetrytypes.SignalLogs, selector),
+		"values for one spelling must cover the whole family")
+	assert.Equal(t,
+		[]string{"deployment.environment.name", "deployment.environment"},
+		FamilyValueNames(context.Background(), valuer.UUID{}, on, telemetrytypes.SignalMetrics, selector),
+		"metric values cover the family members")
+	spanMetric := &telemetrytypes.FieldValueSelector{
+		FieldKeySelector: &telemetrytypes.FieldKeySelector{
+			Name:          "deployment.environment",
+			MetricContext: &telemetrytypes.MetricContext{MetricName: "signoz_calls_total"},
+		},
+	}
+	assert.Equal(t,
+		[]string{
+			"deployment.environment.name", "resource_deployment.environment.name",
+			"deployment.environment", "resource_deployment.environment",
+		},
+		FamilyValueNames(context.Background(), valuer.UUID{}, on, telemetrytypes.SignalMetrics, spanMetric),
+		"span-metrics values cover the resource_ spellings too")
+}

pkg/telemetrymetadata/resolvercore/priority.go

@@ -0,0 +1,25 @@
+package resolvercore
+
+import (
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+var ErrFailedToGetTblStatement = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get tbl statement")
+
+// PriorityForContext orders the contexts a key can resolve to when the same
+// name exists in more than one, lowest first.
+func PriorityForContext(ctx telemetrytypes.FieldContext) int {
+	switch ctx {
+	case telemetrytypes.FieldContextLog:
+		return 1
+	case telemetrytypes.FieldContextResource:
+		return 2
+	case telemetrytypes.FieldContextScope:
+		return 3
+	case telemetrytypes.FieldContextAttribute:
+		return 4
+	default:
+		return 5
+	}
+}

pkg/telemetrymetadata/resolvercore/static_fields.go

@@ -1,4 +1,4 @@
-package telemetrymetadata
+package resolvercore
 
 import (
 	"strings"
@@ -6,7 +6,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
 
-func staticFieldMatchesAny(field telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) bool {
+func StaticFieldMatchesAny(field telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) bool {
 	for _, selector := range selectors {
 		if staticFieldMatches(field, selector) {
 			return true

pkg/telemetrymetadata/resolvercore/tag_table.go

@@ -0,0 +1,161 @@
+package resolvercore
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/huandu/go-sqlbuilder"
+
+	"github.com/SigNoz/signoz/pkg/clickhousesql"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+// errFailedToGetTagValues keeps the message the tag table lookups have always
+// wrapped with, which names logs keys for every signal.
+var errFailedToGetTagValues = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get logs keys")
+
+// TagTableValues returns the string and number values of the key from a
+// tag table, and true and false when the key is a known bool field or the
+// table holds bool rows for it. Bool rows do not count towards the limit.
+func TagTableValues(ctx context.Context, store telemetrystore.TelemetryStore, table string, fieldValueSelector *telemetrytypes.FieldValueSelector, names []string, knownBool bool) (*telemetrytypes.TelemetryFieldValues, bool, error) {
+	limit := fieldValueSelector.Limit
+	if limit == 0 {
+		limit = 50
+	}
+
+	values := &telemetrytypes.TelemetryFieldValues{}
+	if knownBool {
+		values.BoolValues = boolFieldValues(fieldValueSelector.Value).BoolValues
+		if fieldValueSelector.FieldDataType == telemetrytypes.FieldDataTypeBool {
+			return values, true, nil
+		}
+	} else if fieldValueSelector.FieldDataType == telemetrytypes.FieldDataTypeUnspecified {
+		hasBoolRows, err := tagTableHasBoolRows(ctx, store, table, fieldValueSelector, names)
+		if err != nil {
+			return nil, false, err
+		}
+		if hasBoolRows {
+			values.BoolValues = boolFieldValues(fieldValueSelector.Value).BoolValues
+		}
+	}
+
+	sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(table)
+
+	if fieldValueSelector.Name != "" {
+		sb.Where(tagKeyCondition(sb, fieldValueSelector.Name, names))
+	}
+	sb.Where(sb.NE("tag_data_type", telemetrytypes.FieldDataTypeBool.TagDataType()))
+
+	tagTableSinceDay(sb, fieldValueSelector.StartUnixMilli)
+
+	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", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
+		case telemetrytypes.FieldDataTypeNumber:
+			sb.Where(sb.IsNotNull("number_value"))
+			sb.Where(sb.ILike("toString(number_value)", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
+		case telemetrytypes.FieldDataTypeUnspecified:
+			// or b/w string and number
+			sb.Where(sb.Or(
+				sb.ILike("string_value", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"),
+				sb.ILike("toString(number_value)", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"),
+			))
+		}
+	}
+
+	// query one extra to check if we hit the limit
+	sb.Limit(limit + 1)
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	rows, err := store.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, errFailedToGetTagValues.Error())
+	}
+	defer rows.Close()
+
+	seen := make(map[string]bool)
+	rowCount := 0
+	totalCount := 0 // Track total unique values
+
+	for rows.Next() {
+		rowCount++
+
+		var stringValue string
+		var numberValue float64
+		if err := rows.Scan(&stringValue, &numberValue); err != nil {
+			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, errFailedToGetTagValues.Error())
+		}
+
+		// Only add values if we haven't hit the limit yet
+		if totalCount < limit {
+			if _, ok := seen[stringValue]; !ok && stringValue != "" {
+				values.StringValues = append(values.StringValues, stringValue)
+				seen[stringValue] = true
+				totalCount++
+			}
+			if _, ok := seen[fmt.Sprintf("%f", numberValue)]; !ok && numberValue != 0 && totalCount < limit {
+				values.NumberValues = append(values.NumberValues, numberValue)
+				seen[fmt.Sprintf("%f", numberValue)] = true
+				totalCount++
+			}
+		}
+	}
+
+	// hit the limit?
+	complete := rowCount <= limit
+
+	return values, complete, nil
+}
+
+// tagKeyCondition matches the requested key, or every spelling of its
+// family when there is more than one.
+func tagKeyCondition(sb *sqlbuilder.SelectBuilder, name string, names []string) string {
+	if len(names) > 1 {
+		return sb.In("tag_key", sqlbuilder.List(names))
+	}
+	return sb.E("tag_key", name)
+}
+
+// tagTableSinceDay restricts rows to the tag table's day partitions from the
+// start's day on. Partitions are toDate(unix_milli / 1000) in the server's
+// timezone, and a value's surviving row within a day carries whichever hour
+// was inserted last, so the day is the finest safe unit.
+func tagTableSinceDay(sb *sqlbuilder.SelectBuilder, startUnixMilli int64) {
+	if startUnixMilli != 0 {
+		sb.Where(fmt.Sprintf("toDate(unix_milli / 1000) >= toDate(%d)", startUnixMilli/1000))
+	}
+}
+
+// tagTableHasBoolRows reports whether the tag table holds a bool row for the
+// key. Bool rows carry no value, so one row is enough to know the key takes
+// the values true and false.
+func tagTableHasBoolRows(ctx context.Context, store telemetrystore.TelemetryStore, table string, selector *telemetrytypes.FieldValueSelector, names []string) (bool, error) {
+	sb := sqlbuilder.Select("1").From(table)
+	sb.Where(tagKeyCondition(sb, selector.Name, names))
+	sb.Where(sb.E("tag_data_type", telemetrytypes.FieldDataTypeBool.TagDataType()))
+	if selector.FieldContext != telemetrytypes.FieldContextUnspecified {
+		sb.Where(sb.E("tag_type", selector.FieldContext.TagType()))
+	}
+	tagTableSinceDay(sb, selector.StartUnixMilli)
+	sb.Limit(1)
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	rows, err := store.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, errFailedToGetTagValues.Error())
+	}
+	defer rows.Close()
+	return rows.Next(), rows.Err()
+}

pkg/telemetrymetadata/traces.go

@@ -0,0 +1,235 @@
+package telemetrymetadata
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/huandu/go-sqlbuilder"
+	"golang.org/x/exp/maps"
+
+	"github.com/SigNoz/signoz/pkg/clickhousesql"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/telemetrymetadata/parser"
+	"github.com/SigNoz/signoz/pkg/telemetrymetadata/resolvercore"
+	"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
+	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
+	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// tracesTblStatementToFieldKeys returns materialised attribute/resource/scope keys from the traces table.
+func (t *telemetryMetaStore) tracesTblStatementToFieldKeys(ctx context.Context) ([]*telemetrytypes.TelemetryFieldKey, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalTraces.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "tracesTblStatementToFieldKeys",
+	})
+	query := fmt.Sprintf("SHOW CREATE TABLE %s.%s", t.tracesDBName, t.indexV3TblName)
+	statements := []telemetrytypes.ShowCreateTableStatement{}
+	err := t.telemetrystore.ClickhouseDB().Select(ctx, &statements, query)
+	if err != nil {
+		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, resolvercore.ErrFailedToGetTblStatement.Error())
+	}
+
+	materialisedKeys, err := parser.ExtractFieldKeysFromTblStatement(statements[0].Statement)
+	if err != nil {
+		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTracesKeys.Error())
+	}
+
+	for idx := range materialisedKeys {
+		materialisedKeys[idx].Signal = telemetrytypes.SignalTraces
+	}
+
+	return materialisedKeys, nil
+}
+
+// getTracesKeys returns the keys from the spans that match the field selection criteria.
+func (t *telemetryMetaStore) getTracesKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalTraces.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getTracesKeys",
+	})
+
+	if len(fieldKeySelectors) == 0 {
+		return nil, true, nil
+	}
+
+	// pre-fetch the materialised keys from the traces table
+	matKeys, err := t.tracesTblStatementToFieldKeys(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
+	}
+
+	sb := sqlbuilder.Select(
+		"tagKey AS tag_key",
+		"tagType AS tag_type",
+		"dataType AS tag_data_type",
+		`CASE
+			// WHEN tagType = 'spanfield' THEN 1
+			WHEN tagType = 'resource' THEN 2
+			WHEN tagType = 'scope' THEN 3
+			WHEN tagType = 'tag' THEN 4
+			ELSE 5
+		END as priority`,
+	).From(t.tracesDBName + "." + t.spanAttributesKeysTblName)
+	var limit int
+
+	conds := []string{}
+	for _, fieldKeySelector := range fieldKeySelectors {
+
+		// TODO(srikanthccv): support time filtering for span attribute keys
+		// if fieldKeySelector.StartUnixMilli != 0 {
+		// 	conds = append(conds, sb.GE("unix_milli", fieldKeySelector.StartUnixMilli))
+		// }
+		// if fieldKeySelector.EndUnixMilli != 0 {
+		// 	conds = append(conds, sb.LE("unix_milli", fieldKeySelector.EndUnixMilli))
+		// }
+
+		// key part of the selector
+		fieldKeyConds := []string{}
+		if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+			fieldKeyConds = append(fieldKeyConds, sb.E("tagKey", fieldKeySelector.Name))
+		} else {
+			fieldKeyConds = append(fieldKeyConds, sb.ILike("tagKey", "%"+clickhousesql.LikePattern(fieldKeySelector.Name)+"%"))
+		}
+
+		// now look at the field context
+		// we don't write most of intrinsic fields to keys table
+		// for this reason we don't want to apply tagType if the field context
+		// is not attribute, resource attribute or scope
+		switch fieldKeySelector.FieldContext {
+		case telemetrytypes.FieldContextAttribute, telemetrytypes.FieldContextResource, telemetrytypes.FieldContextScope:
+			fieldKeyConds = append(fieldKeyConds, sb.E("tagType", fieldKeySelector.FieldContext.TagType()))
+		}
+
+		// now look at the field data type
+		if fieldKeySelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
+			fieldKeyConds = append(fieldKeyConds, sb.E("dataType", fieldKeySelector.FieldDataType.TagDataType()))
+		}
+
+		conds = append(conds, sb.And(fieldKeyConds...))
+		limit += fieldKeySelector.Limit
+	}
+	// the span_attribute_keys has historically pushed the top level column as attributes
+	sb.Where(sb.Or(conds...)).Where("isColumn = false")
+	sb.GroupBy("tagKey", "tagType", "dataType")
+	if limit == 0 {
+		limit = 1000
+	}
+
+	mainSb := sqlbuilder.Select("tag_key", "tag_type", "tag_data_type", "max(priority) as priority")
+	mainSb.From(mainSb.BuilderAs(sb, "sub_query"))
+	mainSb.GroupBy("tag_key", "tag_type", "tag_data_type")
+	mainSb.OrderBy("priority")
+	// query one extra to check if we hit the limit
+	mainSb.Limit(limit + 1)
+
+	query, args := mainSb.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, ErrFailedToGetTracesKeys.Error())
+	}
+	defer rows.Close()
+	keys := []*telemetrytypes.TelemetryFieldKey{}
+	rowCount := 0
+	for rows.Next() {
+		rowCount++
+		// reached the limit, we know there are more results
+		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, ErrFailedToGetTracesKeys.Error())
+		}
+		key, ok := mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
+
+		// if there is no materialised column, create a key with the field context and data type
+		if !ok {
+			key = &telemetrytypes.TelemetryFieldKey{
+				Name:          name,
+				Signal:        telemetrytypes.SignalTraces,
+				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, ErrFailedToGetTracesKeys.Error())
+	}
+
+	// hit the limit? (only counting DB results)
+	complete := rowCount <= limit
+
+	// Add the matching static fields: the span scope selectors, the intrinsic
+	// columns and the calculated columns. These don't count towards the limit
+	staticFields := maps.Values(tracestelemetryschema.SpanSearchScopeFields)
+	staticFields = append(staticFields, maps.Values(tracestelemetryschema.IntrinsicFields)...)
+	staticFields = append(staticFields, maps.Values(tracestelemetryschema.CalculatedFields)...)
+
+	for _, field := range staticFields {
+		if !resolvercore.StaticFieldMatchesAny(field, fieldKeySelectors) {
+			continue
+		}
+		if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; added {
+			continue
+		}
+		keys = append(keys, &field)
+	}
+
+	if err = t.updateColumnEvolutionMetadataForKeys(ctx, keys); err != nil {
+		return nil, false, err
+	}
+
+	return keys, complete, nil
+}
+
+func (t *telemetryMetaStore) getSpanFieldValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalTraces.StringValue(),
+		instrumentationtypes.CodeNamespace:    "metadata",
+		instrumentationtypes.CodeFunctionName: "getSpanFieldValues",
+	})
+
+	if values, ok := spanSearchScopeFieldValues(fieldValueSelector.Name, fieldValueSelector.Value); ok {
+		return values, true, nil
+	}
+	knownBool := resolvercore.IsKnownBoolField(fieldValueSelector, tracestelemetryschema.IntrinsicFields, tracestelemetryschema.CalculatedFields)
+	names := resolvercore.FamilyValueNames(ctx, orgID, t.fl, telemetrytypes.SignalTraces, fieldValueSelector)
+	// unix_milli is the hour of the span start
+	return resolvercore.TagTableValues(ctx, t.telemetrystore, t.tracesDBName+"."+t.tracesFieldsTblName, fieldValueSelector, names, knownBool)
+}
+
+// spanSearchScopeFieldValues is the suggestion set for a search-scope selector
+// (isRoot, isEntryPoint), which only filters with true. ok is false for any
+// other name.
+func spanSearchScopeFieldValues(name, searchText string) (*telemetrytypes.TelemetryFieldValues, bool) {
+	for scopeName := range tracestelemetryschema.SpanSearchScopeFields {
+		if !strings.EqualFold(scopeName, name) {
+			continue
+		}
+		values := &telemetrytypes.TelemetryFieldValues{}
+		if needle := strings.ToLower(searchText); needle == "" || strings.Contains("true", needle) {
+			values.BoolValues = []bool{true}
+		}
+		return values, true
+	}
+	return nil, false
+}

pkg/types/telemetrytypes/resolver.go

@@ -0,0 +1,49 @@
+package telemetrytypes
+
+import (
+	"context"
+
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+// FieldKeyResolver resolves keys and values for one source within a signal.
+// Implementations never dispatch; a MetadataStore owns the routing, registers
+// them by Source and merges what they return. Keys come back flat rather than
+// grouped by name because the store does the grouping across sources.
+type FieldKeyResolver interface {
+	GetKeys(ctx context.Context, fieldKeySelectors []*FieldKeySelector) ([]*TelemetryFieldKey, bool, error)
+
+	GetAllValues(ctx context.Context, fieldValueSelector *FieldValueSelector) (*TelemetryFieldValues, bool, error)
+}
+
+// MetricMetadata covers the lookups that only apply to the metrics signal.
+type MetricMetadata interface {
+	// FetchTemporality fetches the temporality for metric
+	FetchTemporality(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error)
+
+	// FetchTemporalityMulti fetches the temporality for multiple metrics
+	FetchTemporalityMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error)
+
+	FetchTemporalityAndTypeMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error)
+
+	// GetFirstSeenFromMetricMetadata gets the first seen timestamp for a metric metadata lookup key.
+	GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []MetricMetadataLookupKey) (map[MetricMetadataLookupKey]int64, error)
+
+	FetchLastSeenInfoMulti(ctx context.Context, orgID valuer.UUID, metricNames ...string) (map[string]int64, error)
+}
+
+// LogsMetadata covers the JSON column evolution lookups, which today only the
+// logs signal materialises.
+type LogsMetadata interface {
+	// ListJSONIndexes lists the per-path JSON skip indexes of the given source.
+	ListJSONIndexes(ctx context.Context, lookup JSONIndexLookup, filters ...string) ([]TelemetryFieldKeySkipIndex, error)
+
+	// GetPromotedPaths lists the promoted paths recorded in the column
+	// evolution table for the entry's signal, column and field context.
+	GetPromotedPaths(ctx context.Context, entry EvolutionEntry, paths ...string) (map[string]bool, error)
+
+	// PromotePaths records promoted paths in the column evolution table as
+	// rows templated by entry; FieldName and ReleaseTime are set per path.
+	PromotePaths(ctx context.Context, entry EvolutionEntry, paths ...string) error
+}

pkg/types/telemetrytypes/store.go

@@ -3,11 +3,12 @@ package telemetrytypes
 import (
 	"context"
 
-	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
-// MetadataStore is the interface for the telemetry metadata store.
+// MetadataStore is the interface for the telemetry metadata store. The key and
+// value lookups route to a per-signal FieldKeyResolver; the embedded interfaces
+// are signal-scoped and pass straight through.
 type MetadataStore interface {
 	// GetKeys returns a map of field keys types.TelemetryFieldKey by name, there can be multiple keys with the same name
 	// if they have different types or data types.
@@ -26,32 +27,12 @@ type MetadataStore interface {
 	// GetAllValues returns a list of all values.
 	GetAllValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *FieldValueSelector) (*TelemetryFieldValues, bool, error)
 
-	// FetchTemporality fetches the temporality for metric
-	FetchTemporality(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error)
-
-	// FetchTemporalityMulti fetches the temporality for multiple metrics
-	FetchTemporalityMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error)
-
-	FetchTemporalityAndTypeMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error)
-
-	// ListJSONIndexes lists the per-path JSON skip indexes of the given source.
-	ListJSONIndexes(ctx context.Context, lookup JSONIndexLookup, filters ...string) ([]TelemetryFieldKeySkipIndex, error)
-
-	// GetPromotedPaths lists the promoted paths recorded in the column
-	// evolution table for the entry's signal, column and field context.
-	GetPromotedPaths(ctx context.Context, entry EvolutionEntry, paths ...string) (map[string]bool, error)
-
-	// PromotePaths records promoted paths in the column evolution table as
-	// rows templated by entry; FieldName and ReleaseTime are set per path.
-	PromotePaths(ctx context.Context, entry EvolutionEntry, paths ...string) error
-
 	// GetMaterializedKeys lists the keys materialized as dedicated columns.
 	GetMaterializedKeys(ctx context.Context, signal Signal) ([]*TelemetryFieldKey, error)
 
-	// GetFirstSeenFromMetricMetadata gets the first seen timestamp for a metric metadata lookup key.
-	GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []MetricMetadataLookupKey) (map[MetricMetadataLookupKey]int64, error)
+	MetricMetadata
 
-	FetchLastSeenInfoMulti(ctx context.Context, orgID valuer.UUID, metricNames ...string) (map[string]int64, error)
+	LogsMetadata
 }
 
 type MetricMetadataLookupKey struct {