feat(logs): implement search() across all log fields

Replace the search() stub with the real implementation. The logs condition
builder fans a case-insensitive match of the needle across every searchable
column - log columns, body/body_v2, and the attribute + resource maps - gated
by the allow_logs_search feature flag. Scoped forms such as
search('needle', body, resource) narrow the fan-out to the named field contexts.

The where-clause visitor emits FilterOperatorSearch and flags the statement as
scan-heavy; the logs statement builder attaches a CostGuard that the querier
enforces via EXPLAIN ESTIMATE against a configurable per-shard scan budget.
body_v2 gets its own, lower budget (search_max_scan_rows_json_body) since
toString() rebuilds every document and no skip index can prune. Both keys are
squashed into the querier config namespace. Rejections lead with the search()
advisory and say how to get back under budget.

grammar/FilterQuery.g4

@@ -112,15 +112,12 @@ functionCall
     ;
 
 /*
- * Full-text search call: search('needle')
- *
- * Uses the shared functionParamList so future scoped forms like
- * search(body, 'abc') / search(attribute, 'abc') need no grammar change. Today
- * only a single needle is supported. Unlike bare/quoted free text (`fullText`),
- * which only targets the body column, search() fans out across every field.
+ * Full-text search: search('needle') or scoped search('needle', body, ...).
+ * First param is the needle; the rest are field-context scopes (body/attribute/
+ * resource/log), quoted or bare. Handled in the visitor — no grammar change.
  */
 searchCall
-    : SEARCH LPAREN functionParamList RPAREN
+    : SEARCH LPAREN valueList RPAREN
     ;
 
 // Function parameters can be keys, single scalar values, or arrays

pkg/flagger/flaggertest/flaggertest.go

@@ -29,13 +29,18 @@ func New(t *testing.T) flagger.Flagger {
 
 // WithUseJSONBody returns a Flagger with use_json_body set to the given value.
 func WithUseJSONBody(t *testing.T, enabled bool) flagger.Flagger {
+	return WithBooleanFlags(t, map[string]bool{
+		flagger.FeatureUseJSONBody.String(): enabled,
+	})
+}
+
+// WithBooleanFlags returns a Flagger with the given boolean flags, keyed by feature name.
+func WithBooleanFlags(t *testing.T, flags map[string]bool) flagger.Flagger {
 	t.Helper()
 	registry := flagger.MustNewRegistry()
 	cfg := flagger.Config{}
-	if enabled {
-		cfg.Config.Boolean = map[string]bool{
-			flagger.FeatureUseJSONBody.String(): true,
-		}
+	if len(flags) > 0 {
+		cfg.Config.Boolean = flags
 	}
 	fl, err := flagger.New(
 		context.Background(),

pkg/modules/rulestatehistory/implrulestatehistory/condition_builder.go

@@ -35,7 +35,7 @@ func (c *conditionBuilder) ConditionFor(
 	sb *sqlbuilder.SelectBuilder,
 ) ([]string, []string, error) {
 
-	// has/hasAny/hasAll/hasToken are logs-body-only; reject for rule state history.
+	// has/hasAny/hasAll/hasToken/search are logs-only functions; reject for rule state history.
 	if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
 		return nil, nil, err
 	}

pkg/parser/filterquery/grammar/filterquery_parser.go

@@ -137,8 +137,8 @@ func filterqueryParserInit() {
 		0, 0, 191, 19, 1, 0, 0, 0, 192, 190, 1, 0, 0, 0, 193, 194, 7, 2, 0, 0,
 		194, 21, 1, 0, 0, 0, 195, 196, 7, 3, 0, 0, 196, 197, 5, 1, 0, 0, 197, 198,
 		3, 26, 13, 0, 198, 199, 5, 2, 0, 0, 199, 23, 1, 0, 0, 0, 200, 201, 5, 27,
-		0, 0, 201, 202, 5, 1, 0, 0, 202, 203, 3, 26, 13, 0, 203, 204, 5, 2, 0,
-		0, 204, 25, 1, 0, 0, 0, 205, 210, 3, 28, 14, 0, 206, 207, 5, 5, 0, 0, 207,
+		0, 0, 201, 202, 5, 1, 0, 0, 202, 203, 3, 18, 9, 0, 203, 204, 5, 2, 0, 0,
+		204, 25, 1, 0, 0, 0, 205, 210, 3, 28, 14, 0, 206, 207, 5, 5, 0, 0, 207,
 		209, 3, 28, 14, 0, 208, 206, 1, 0, 0, 0, 209, 212, 1, 0, 0, 0, 210, 208,
 		1, 0, 0, 0, 210, 211, 1, 0, 0, 0, 211, 27, 1, 0, 0, 0, 212, 210, 1, 0,
 		0, 0, 213, 217, 3, 34, 17, 0, 214, 217, 3, 32, 16, 0, 215, 217, 3, 30,
@@ -2945,7 +2945,7 @@ type ISearchCallContext interface {
 	// Getter signatures
 	SEARCH() antlr.TerminalNode
 	LPAREN() antlr.TerminalNode
-	FunctionParamList() IFunctionParamListContext
+	ValueList() IValueListContext
 	RPAREN() antlr.TerminalNode
 
 	// IsSearchCallContext differentiates from other interfaces.
@@ -2992,10 +2992,10 @@ func (s *SearchCallContext) LPAREN() antlr.TerminalNode {
 	return s.GetToken(FilterQueryParserLPAREN, 0)
 }
 
-func (s *SearchCallContext) FunctionParamList() IFunctionParamListContext {
+func (s *SearchCallContext) ValueList() IValueListContext {
 	var t antlr.RuleContext
 	for _, ctx := range s.GetChildren() {
-		if _, ok := ctx.(IFunctionParamListContext); ok {
+		if _, ok := ctx.(IValueListContext); ok {
 			t = ctx.(antlr.RuleContext)
 			break
 		}
@@ -3005,7 +3005,7 @@ func (s *SearchCallContext) FunctionParamList() IFunctionParamListContext {
 		return nil
 	}
 
-	return t.(IFunctionParamListContext)
+	return t.(IValueListContext)
 }
 
 func (s *SearchCallContext) RPAREN() antlr.TerminalNode {
@@ -3064,7 +3064,7 @@ func (p *FilterQueryParser) SearchCall() (localctx ISearchCallContext) {
 	}
 	{
 		p.SetState(202)
-		p.FunctionParamList()
+		p.ValueList()
 	}
 	{
 		p.SetState(203)

pkg/querier/builder_query.go

@@ -20,6 +20,8 @@ import (
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
+const estimateTimeout = 5 * time.Second
+
 const traceOutsideRangeWarn = "Query %s references a trace_id that exists between %s and %s (UTC) but lies outside the selected time range; adjust the time range to see results"
 
 type builderQuery[T any] struct {
@@ -247,6 +249,10 @@ func (q *builderQuery[T]) Execute(ctx context.Context) (*qbtypes.Result, error)
 		return nil, err
 	}
 
+	if err := q.enforceEstimate(ctx, stmt); err != nil {
+		return nil, err
+	}
+
 	// Execute the query with proper context for partial value detection
 	result, err := q.executeWithContext(ctx, stmt.Query, stmt.Args)
 	if err != nil {
@@ -258,6 +264,68 @@ func (q *builderQuery[T]) Execute(ctx context.Context) (*qbtypes.Result, error)
 	return result, nil
 }
 
+// estimateRows returns the per-shard EXPLAIN ESTIMATE scan rows for a cost-guarded
+// statement. guarded=false means nothing to enforce; a non-nil error means reject.
+// Callers own the budget comparison (per-statement or cumulative).
+func (q *builderQuery[T]) estimateRows(ctx context.Context, stmt *qbtypes.Statement) (int64, bool, error) {
+	if stmt.CostGuard == nil || stmt.CostGuard.MaxScanRows <= 0 {
+		return 0, false, nil
+	}
+
+	estCtx, cancel := context.WithTimeout(ctx, estimateTimeout)
+	defer cancel()
+
+	entries, err := q.telemetryStore.Estimate(estCtx, stmt.Query, stmt.Args...)
+	if err != nil {
+		// Parent cancellation isn't the budget's concern — surface it unchanged.
+		if ctx.Err() != nil {
+			return 0, true, ctx.Err()
+		}
+		// Fail closed: without an estimate we cannot bound a scan-heavy query.
+		reason := fmt.Sprintf("This query is too broad to plan within %s", estimateTimeout)
+		if estCtx.Err() != context.DeadlineExceeded {
+			reason = "Could not estimate this query's scan cost"
+			q.logger.WarnContext(ctx, "EXPLAIN ESTIMATE failed; rejecting scan-heavy query (fail-closed)", errors.Attr(err))
+		}
+		return 0, true, errors.NewInvalidInputf(errors.CodeInvalidInput, "%s", reason).
+			WithSuggestions(costGuardSuggestions(stmt.CostGuard.Warning)...)
+	}
+
+	var rows int64
+	for _, e := range entries {
+		rows += e.Rows
+	}
+	return rows, true, nil
+}
+
+// enforceEstimate rejects a scan-heavy statement whose estimate exceeds its own
+// budget, before executing. Budget 0 disables.
+func (q *builderQuery[T]) enforceEstimate(ctx context.Context, stmt *qbtypes.Statement) error {
+	rows, guarded, err := q.estimateRows(ctx, stmt)
+	if err != nil {
+		return err
+	}
+	if !guarded {
+		return nil
+	}
+	if budget := stmt.CostGuard.MaxScanRows; rows > budget {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"This query would scan about %d rows per shard in this range, over the per-shard limit of %d", rows, budget).
+			WithSuggestions(costGuardSuggestions(stmt.CostGuard.Warning)...)
+	}
+	return nil
+}
+
+// costGuardSuggestions leads with the requirement's advisory (e.g. the search() hint),
+// then how to get under budget.
+func costGuardSuggestions(advisory string) []string {
+	suggestions := make([]string, 0, 2)
+	if advisory != "" {
+		suggestions = append(suggestions, advisory)
+	}
+	return append(suggestions, "Narrow the time range or add a more selective filter.")
+}
+
 // narrowWindowByTraceID inspects the filter for trace_id predicates and clamps
 // [fromMS,toMS] to the time range stored in signoz_traces.distributed_trace_summary.
 // Returns the (possibly narrowed) window, overlap=false when the trace lies
@@ -491,6 +559,10 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
 	var warnings []string
 	var warningsDocURL string
 
+	// Cumulative across visited buckets: the budget bounds the whole query's per-shard
+	// scan, not each bucket independently.
+	var estimatedScan int64
+
 	for _, r := range buckets {
 		q.spec.Offset = 0
 		q.spec.Limit = need
@@ -501,6 +573,18 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
 		}
 		warnings = stmt.Warnings
 		warningsDocURL = stmt.WarningsDocURL
+		rowsEst, guarded, err := q.estimateRows(ctx, stmt)
+		if err != nil {
+			return nil, err
+		}
+		if guarded {
+			estimatedScan += rowsEst
+			if budget := stmt.CostGuard.MaxScanRows; estimatedScan > budget {
+				return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
+					"This query would scan about %d rows per shard across the time range, over the per-shard limit of %d", estimatedScan, budget).
+					WithSuggestions(costGuardSuggestions(stmt.CostGuard.Warning)...)
+			}
+		}
 		// Execute with proper context for partial value detection
 		res, err := q.executeWithContext(ctx, stmt.Query, stmt.Args)
 		if err != nil {

pkg/querier/config.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/statementbuilder"
 )
 
 const DefaultMaxConcurrentQueries = 8
@@ -19,6 +20,9 @@ type Config struct {
 	MaxConcurrentQueries int `yaml:"max_concurrent_queries" mapstructure:"max_concurrent_queries"`
 	// LogTraceIDWindowPadding is the padding added to narrowed down timerange from trace summary to logs with trace_id filter.
 	LogTraceIDWindowPadding time.Duration `yaml:"log_trace_id_window_padding" mapstructure:"log_trace_id_window_padding"`
+
+	// Squashed so its keys live directly under the querier namespace, e.g. querier.search_max_scan_rows.
+	statementbuilder.Config `mapstructure:",squash" yaml:",squash"`
 }
 
 // NewConfigFactory creates a new config factory for querier.
@@ -29,10 +33,11 @@ func NewConfigFactory() factory.ConfigFactory {
 func newConfig() factory.Config {
 	return Config{
 		// Default values
-		CacheTTL:             168 * time.Hour,
-		FluxInterval:         5 * time.Minute,
+		CacheTTL:                168 * time.Hour,
+		FluxInterval:            5 * time.Minute,
 		MaxConcurrentQueries:    DefaultMaxConcurrentQueries,
 		LogTraceIDWindowPadding: 5 * time.Minute,
+		Config:                  statementbuilder.NewConfig(),
 	}
 }
 
@@ -50,6 +55,10 @@ func (c Config) Validate() error {
 	if c.LogTraceIDWindowPadding < 0 {
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "log_trace_id_window_padding must not be negative, got %v", c.LogTraceIDWindowPadding)
 	}
+	// The embedded config's Validate is shadowed by this one; call it explicitly.
+	if err := c.Config.Validate(); err != nil {
+		return err
+	}
 	return nil
 }
 

pkg/querybuilder/constants.go

@@ -8,6 +8,9 @@ const (
 	// BodyFullTextSearchDefaultWarning is emitted when a full-text search or "body" searches are hit
 	// with New JSON Body enhancements.
 	BodyFullTextSearchDefaultWarning = "Full text searches default to `body.message:string`. Use `body.<key>` to search a different field inside body"
+
+	// SearchWarning is emitted on every search() call — it scans all fields.
+	SearchWarning = "search() runs across all fields and can be slow and expensive. Prefer a specific field, e.g. `<context>.<field_key>:<type>`"
 )
 
 var (

pkg/querybuilder/key_resolution.go

@@ -161,14 +161,16 @@ func inferDataTypesFromList(values []any) []telemetrytypes.FieldDataType {
 	return out
 }
 
-// NewFunctionUnsupportedError returns the error for a has/hasAny/hasAll/hasToken operator
-// on a builder that doesn't support it (logs body only), or nil for other operators.
+// NewFunctionUnsupportedError returns the error for a has/hasAny/hasAll/hasToken/search
+// operator on a builder that doesn't support it (logs only), or nil for other operators.
 func NewFunctionUnsupportedError(operator qbtypes.FilterOperator) error {
 	switch operator {
 	case qbtypes.FilterOperatorHasToken:
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "function `hasToken` only supports body field as first parameter").WithUrl(hasTokenFunctionDocURL)
 	case qbtypes.FilterOperatorHas, qbtypes.FilterOperatorHasAny, qbtypes.FilterOperatorHasAll:
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "function `%s` supports only body JSON search", operator.FunctionName()).WithUrl(functionBodyJSONSearchDocURL)
+	case qbtypes.FilterOperatorSearch:
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "function `search` is only supported for logs")
 	default:
 		return nil
 	}

pkg/querybuilder/where_clause_visitor.go

@@ -43,6 +43,8 @@ type filterExpressionVisitor struct {
 	keysWithWarnings map[string]bool
 	startNs          uint64
 	endNs            uint64
+
+	requiresCostGuard bool
 }
 
 type FilterExprVisitorOpts struct {
@@ -81,9 +83,10 @@ func newFilterExpressionVisitor(opts FilterExprVisitorOpts) *filterExpressionVis
 }
 
 type PreparedWhereClause struct {
-	WhereClause    *sqlbuilder.WhereClause
-	Warnings       []string
-	WarningsDocURL string
+	WhereClause       *sqlbuilder.WhereClause
+	Warnings          []string
+	WarningsDocURL    string
+	RequiresCostGuard bool
 }
 
 func (p PreparedWhereClause) IsEmpty() bool {
@@ -165,12 +168,12 @@ func PrepareWhereClause(query string, opts FilterExprVisitorOpts) (PreparedWhere
 
 	// Return empty where clause so callers can skip the WHERE clause
 	if cond == "" || cond == SkipConditionLiteral {
-		return PreparedWhereClause{WhereClause: nil, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL}, nil
+		return PreparedWhereClause{WhereClause: nil, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL, RequiresCostGuard: visitor.requiresCostGuard}, nil
 	}
 
 	whereClause := sqlbuilder.NewWhereClause().AddWhereExpr(visitor.builder.Args, cond)
 
-	return PreparedWhereClause{WhereClause: whereClause, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL}, nil
+	return PreparedWhereClause{WhereClause: whereClause, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL, RequiresCostGuard: visitor.requiresCostGuard}, nil
 }
 
 // Visit dispatches to the specific visit method based on node type.
@@ -776,11 +779,77 @@ func normalizeFunctionValue(operator qbtypes.FilterOperator, functionName string
 	return valueParams, nil
 }
 
-// VisitSearchCall handles search('needle'). The search() function is parsed but
-// not yet implemented; reject it with a clear invalid-input error.
+// VisitSearchCall handles search('needle'[, body, resource, …]): a case-insensitive
+// needle plus optional field-context scopes, ORing one FilterOperatorSearch per scope
+// (no scope = keyless, covering every field).
 func (v *filterExpressionVisitor) VisitSearchCall(ctx *grammar.SearchCallContext) any {
-	v.errors = append(v.errors, "function `search` is not yet supported")
-	return ErrorConditionLiteral
+	// Flag scan-heavy so the statement builder attaches the cost guard.
+	v.requiresCostGuard = true
+
+	valueList := ctx.ValueList()
+	if valueList == nil {
+		v.errors = append(v.errors, "function `search` expects a needle, e.g. search('error')")
+		return ErrorConditionLiteral
+	}
+	params := valueList.AllValue()
+	if len(params) == 0 {
+		v.errors = append(v.errors, "function `search` expects a needle, e.g. search('error')")
+		return ErrorConditionLiteral
+	}
+
+	searchText, ok := searchParamText(params[0])
+	if !ok {
+		v.errors = append(v.errors, "function `search` expects a needle as its first argument, e.g. search('error')")
+		return ErrorConditionLiteral
+	}
+
+	var fieldContexts []telemetrytypes.FieldContext
+	if len(params) == 1 {
+		fieldContexts = []telemetrytypes.FieldContext{telemetrytypes.FieldContextUnspecified}
+	} else {
+		for _, p := range params[1:] {
+			scopeText, sok := searchParamText(p)
+			if !sok {
+				v.errors = append(v.errors, "function `search` expects each scope to be a context, e.g. search('error', body, resource)")
+				return ErrorConditionLiteral
+			}
+			fc, fok := telemetrytypes.FieldContextFromText(scopeText)
+			if !fok {
+				v.errors = append(v.errors, fmt.Sprintf("invalid search scope %q; expected a field context: body, attribute, resource, or log", scopeText))
+				return ErrorConditionLiteral
+			}
+			fieldContexts = append(fieldContexts, fc)
+		}
+	}
+
+	var conds []string
+	for _, fieldContext := range fieldContexts {
+		key := telemetrytypes.NewTelemetryFieldKey("", fieldContext, telemetrytypes.FieldDataTypeUnspecified)
+		scoped, cok := v.buildConditions(key, nil, qbtypes.FilterOperatorSearch, searchText)
+		if !cok {
+			return ErrorConditionLiteral
+		}
+		conds = append(conds, scoped...)
+	}
+	if len(conds) == 0 {
+		return SkipConditionLiteral
+	}
+	if len(conds) == 1 {
+		return conds[0]
+	}
+	return v.builder.Or(conds...)
+}
+
+// searchParamText returns an argument's raw token text (quoted or bare) rather than its
+// visited value, so a bare word stays literal and search(1000000) isn't "1e+06".
+func searchParamText(val grammar.IValueContext) (string, bool) {
+	if val == nil {
+		return "", false
+	}
+	if val.QUOTED_TEXT() != nil {
+		return trimQuotes(val.QUOTED_TEXT().GetText()), true
+	}
+	return val.GetText(), true
 }
 
 // VisitFunctionParamList handles the parameter list for function calls.

pkg/signoz/config.go

@@ -39,7 +39,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/sqlmigrator"
 	"github.com/SigNoz/signoz/pkg/sqlschema"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
-	"github.com/SigNoz/signoz/pkg/statementbuilder"
 	"github.com/SigNoz/signoz/pkg/statsreporter"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
 	"github.com/SigNoz/signoz/pkg/tokenizer"
@@ -95,12 +94,9 @@ type Config struct {
 	// Alertmanager config
 	Alertmanager alertmanager.Config `mapstructure:"alertmanager" yaml:"alertmanager"`
 
-	// Querier config
+	// Querier config (the statement-builder config is embedded under it)
 	Querier querier.Config `mapstructure:"querier"`
 
-	// StatementBuilder config
-	StatementBuilder statementbuilder.Config `mapstructure:"statementbuilder"`
-
 	// Ruler config
 	Ruler ruler.Config `mapstructure:"ruler"`
 
@@ -170,7 +166,6 @@ func NewConfig(ctx context.Context, logger *slog.Logger, resolverConfig config.R
 		prometheus.NewConfigFactory(),
 		alertmanager.NewConfigFactory(),
 		querier.NewConfigFactory(),
-		statementbuilder.NewConfigFactory(),
 		ruler.NewConfigFactory(),
 		emailing.NewConfigFactory(),
 		sharder.NewConfigFactory(),

pkg/signoz/signoz.go

@@ -122,7 +122,7 @@ func newQueryStack(
 ) {
 	metadataStore := telemetrymetadata.NewTelemetryMetaStore(settings, telemetryStore, fl)
 
-	cfg := config.StatementBuilder
+	cfg := config.Querier.Config
 	traceStmtBuilder, err := tracesstatementbuilder.NewFactory(telemetryStore, metadataStore, fl).New(ctx, settings, cfg)
 	if err != nil {
 		return nil, nil, nil, nil, nil, nil, nil, nil, err

pkg/statementbuilder/config.go

@@ -2,7 +2,6 @@ package statementbuilder
 
 import (
 	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/factory"
 )
 
 // SkipResourceFingerprint configures when the resource fingerprint subquery is skipped in favor of main-table filtering.
@@ -16,19 +15,24 @@ type SkipResourceFingerprint struct {
 type Config struct {
 	// SkipResourceFingerprint configures when the resource fingerprint subquery is skipped in favor of main-table filtering.
 	SkipResourceFingerprint SkipResourceFingerprint `yaml:"skip_resource_fingerprint" mapstructure:"skip_resource_fingerprint"`
+	// SearchMaxScanRows caps per-shard rows a search() may scan, enforced by the querier
+	// via EXPLAIN ESTIMATE (0 disables).
+	SearchMaxScanRows int64 `yaml:"search_max_scan_rows" mapstructure:"search_max_scan_rows"`
+	// SearchMaxScanRowsJSONBody is the same budget for body_v2, where each row costs far
+	// more: toString() rebuilds every document and no skip index prunes (0 disables).
+	SearchMaxScanRowsJSONBody int64 `yaml:"search_max_scan_rows_json_body" mapstructure:"search_max_scan_rows_json_body"`
 }
 
-// NewConfigFactory creates a new config factory for the statement builders.
-func NewConfigFactory() factory.ConfigFactory {
-	return factory.NewConfigFactory(factory.MustNewName("statementbuilder"), newConfig)
-}
-
-func newConfig() factory.Config {
+// NewConfig returns the default config. It is squashed into querier.Config rather than
+// registered as its own section, so querier's newConfig seeds these defaults.
+func NewConfig() Config {
 	return Config{
 		SkipResourceFingerprint: SkipResourceFingerprint{
 			Enabled:   false,
 			Threshold: 100000,
 		},
+		SearchMaxScanRows:         60_000_000,
+		SearchMaxScanRowsJSONBody: 6_000_000,
 	}
 }
 
@@ -37,5 +41,11 @@ func (c Config) Validate() error {
 	if c.SkipResourceFingerprint.Enabled && c.SkipResourceFingerprint.Threshold == 0 {
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "skip_resource_fingerprint.threshold must be > 0 when enabled")
 	}
+	if c.SearchMaxScanRows < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "search_max_scan_rows must not be negative, got %v", c.SearchMaxScanRows)
+	}
+	if c.SearchMaxScanRowsJSONBody < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "search_max_scan_rows_json_body must not be negative, got %v", c.SearchMaxScanRowsJSONBody)
+	}
 	return nil
 }

pkg/statementbuilder/logsstatementbuilder/groupby_unknown_key_test.go

@@ -8,6 +8,7 @@ import (
 	"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/statementbuilder"
 	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -51,7 +52,8 @@ func TestStatementBuilderGroupByUnknownKey(t *testing.T) {
 			statementBuilder := NewLogQueryStatementBuilder(
 				instrumentationtest.New().ToProviderSettings(),
 				mockMetadataStore, fm, cb, aggExprRewriter,
-				logstelemetryschema.DefaultFullTextColumn, fl, nil, false, 100000,
+				logstelemetryschema.DefaultFullTextColumn, fl, nil,
+				statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 			)
 
 			query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{

pkg/statementbuilder/logsstatementbuilder/json_stmt_builder_test.go

@@ -11,6 +11,7 @@ import (
 	"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/statementbuilder"
 	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -1346,8 +1347,7 @@ func buildJSONTestStatementBuilder(t *testing.T, addIndexes bool) (*logQueryStat
 		logstelemetryschema.DefaultFullTextColumn,
 		fl,
 		nil,
-		false,
-		100000,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 	)
 
 	return statementBuilder, mockMetadataStore

pkg/statementbuilder/logsstatementbuilder/search_cost_guard_test.go

@@ -0,0 +1,85 @@
+package logsstatementbuilder
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"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/statementbuilder"
+	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestSearchCostGuard asserts Build attaches the CostGuard budget and its advisory.
+func TestSearchCostGuard(t *testing.T) {
+	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
+	ctx := context.Background()
+	start := uint64(releaseTime.Add(-5 * time.Minute).UnixMilli())
+	end := uint64(releaseTime.UnixMilli())
+
+	fl := flaggertest.WithBooleanFlags(t, map[string]bool{})
+	fm := logstelemetryschema.NewFieldMapper(fl)
+	cb := logstelemetryschema.NewConditionBuilder(fm, fl)
+	store := telemetrytypestest.NewMockMetadataStore()
+	store.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
+	rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
+	sb := NewLogQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		store, fm, cb, rewriter, logstelemetryschema.DefaultFullTextColumn, fl, nil,
+		statementbuilder.Config{SearchMaxScanRows: 100000, SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
+	)
+	query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+		Signal: telemetrytypes.SignalLogs,
+		Filter: &qbtypes.Filter{Expression: "search('error')"},
+		Limit:  1,
+	}
+
+	stmt, err := sb.Build(ctx, valuer.UUID{}, start, end, qbtypes.RequestTypeRaw, query, nil)
+	require.NoError(t, err)
+	require.NotNil(t, stmt.CostGuard)
+	assert.Equal(t, int64(100000), stmt.CostGuard.MaxScanRows)
+	assert.Contains(t, stmt.Warnings, querybuilder.SearchWarning)
+}
+
+// TestSearchCostGuardJSONBody asserts body_v2 gets its own, lower budget.
+func TestSearchCostGuardJSONBody(t *testing.T) {
+	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
+	ctx := context.Background()
+	start := uint64(releaseTime.Add(-5 * time.Minute).UnixMilli())
+	end := uint64(releaseTime.UnixMilli())
+
+	fl := flaggertest.WithUseJSONBody(t, true)
+	fm := logstelemetryschema.NewFieldMapper(fl)
+	cb := logstelemetryschema.NewConditionBuilder(fm, fl)
+	store := telemetrytypestest.NewMockMetadataStore()
+	store.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
+	rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
+	sb := NewLogQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		store, fm, cb, rewriter, logstelemetryschema.DefaultFullTextColumn, fl, nil,
+		statementbuilder.Config{
+			SearchMaxScanRows:         100000,
+			SearchMaxScanRowsJSONBody: 10000,
+			SkipResourceFingerprint:   statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000},
+		},
+	)
+	query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+		Signal: telemetrytypes.SignalLogs,
+		Filter: &qbtypes.Filter{Expression: "search('error')"},
+		Limit:  1,
+	}
+
+	stmt, err := sb.Build(ctx, valuer.UUID{}, start, end, qbtypes.RequestTypeRaw, query, nil)
+	require.NoError(t, err)
+	require.NotNil(t, stmt.CostGuard)
+	assert.Equal(t, int64(10000), stmt.CostGuard.MaxScanRows)
+	assert.Contains(t, stmt.Warnings, querybuilder.SearchWarning)
+}

pkg/statementbuilder/logsstatementbuilder/statement_builder.go

@@ -41,14 +41,16 @@ type logQueryStatementBuilder struct {
 	fl                             flagger.Flagger
 	skipResourceFingerprintEnabled bool
 
-	fullTextColumn *telemetrytypes.TelemetryFieldKey
+	fullTextColumn            *telemetrytypes.TelemetryFieldKey
+	searchMaxScanRows         int64
+	searchMaxScanRowsJSONBody int64
 }
 
 var _ qbtypes.StatementBuilder[qbtypes.LogAggregation] = (*logQueryStatementBuilder)(nil)
 
 // NewFactory returns a provider factory for the logs statement builder. Its New
 // internalizes the FieldMapper, ConditionBuilder, and AggExprRewriter, and reads
-// SkipResourceFingerprint from the config.
+// SkipResourceFingerprint and the search() scan budgets from the config.
 func NewFactory(
 	telemetryStore telemetrystore.TelemetryStore,
 	metadataStore telemetrytypes.MetadataStore,
@@ -62,7 +64,7 @@ func NewFactory(
 			aggExprRewriter := querybuilder.NewAggExprRewriter(settings, logstelemetryschema.DefaultFullTextColumn, fm, cb, fl)
 			return NewLogQueryStatementBuilder(
 				settings, metadataStore, fm, cb, aggExprRewriter, logstelemetryschema.DefaultFullTextColumn,
-				fl, telemetryStore, cfg.SkipResourceFingerprint.Enabled, cfg.SkipResourceFingerprint.Threshold,
+				fl, telemetryStore, cfg,
 			), nil
 		},
 	)
@@ -77,8 +79,7 @@ func NewLogQueryStatementBuilder(
 	fullTextColumn *telemetrytypes.TelemetryFieldKey,
 	fl flagger.Flagger,
 	telemetryStore telemetrystore.TelemetryStore,
-	skipResourceFingerprintEnable bool,
-	skipResourceFingerprintThreshold uint64,
+	cfg statementbuilder.Config,
 ) *logQueryStatementBuilder {
 	logsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema")
 
@@ -92,10 +93,10 @@ func NewLogQueryStatementBuilder(
 		fullTextColumn,
 		fl,
 		telemetryStore,
-		skipResourceFingerprintThreshold,
+		cfg.SkipResourceFingerprint.Threshold,
 	)
 
-	return &logQueryStatementBuilder{
+	b := &logQueryStatementBuilder{
 		logger:                         logsSettings.Logger(),
 		metadataStore:                  metadataStore,
 		fm:                             fieldMapper,
@@ -103,9 +104,12 @@ func NewLogQueryStatementBuilder(
 		resourceFilterResolver:         resourceFilterResolver,
 		aggExprRewriter:                aggExprRewriter,
 		fl:                             fl,
-		skipResourceFingerprintEnabled: skipResourceFingerprintEnable,
+		skipResourceFingerprintEnabled: cfg.SkipResourceFingerprint.Enabled,
+		searchMaxScanRows:              cfg.SearchMaxScanRows,
+		searchMaxScanRowsJSONBody:      cfg.SearchMaxScanRowsJSONBody,
 		fullTextColumn:                 fullTextColumn,
 	}
+	return b
 }
 
 // Build builds a SQL query for logs based on the given parameters.
@@ -151,9 +155,26 @@ func (b *logQueryStatementBuilder) Build(
 	}
 
 	stmt.Warnings = append(stmt.Warnings, warnings...)
+	// Surface the guard's advisory to the user alongside the other warnings.
+	if stmt.CostGuard != nil && stmt.CostGuard.Warning != "" {
+		stmt.Warnings = append(stmt.Warnings, stmt.CostGuard.Warning)
+	}
 	return stmt, nil
 }
 
+// costGuardFor pairs the search() advisory with the budget for the body path taken —
+// body_v2 has its own, lower one. nil when the statement needs no guard.
+func (b *logQueryStatementBuilder) costGuardFor(ctx context.Context, orgID valuer.UUID, required bool) *qbtypes.CostGuard {
+	if !required {
+		return nil
+	}
+	maxScanRows := b.searchMaxScanRows
+	if b.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
+		maxScanRows = b.searchMaxScanRowsJSONBody
+	}
+	return &qbtypes.CostGuard{Warning: querybuilder.SearchWarning, MaxScanRows: maxScanRows}
+}
+
 func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], bodyJSONEnabled bool) ([]*telemetrytypes.FieldKeySelector, []string) {
 	var keySelectors []*telemetrytypes.FieldKeySelector
 	var warnings []string
@@ -391,6 +412,7 @@ func (b *logQueryStatementBuilder) buildListQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(ctx, orgID, preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil
@@ -557,6 +579,7 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(ctx, orgID, preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil
@@ -684,6 +707,7 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(ctx, orgID, preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil

pkg/statementbuilder/logsstatementbuilder/stmt_builder_test.go

@@ -11,6 +11,7 @@ import (
 	"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/statementbuilder"
 	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
 	"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
@@ -231,8 +232,7 @@ func TestStatementBuilderTimeSeries(t *testing.T) {
 		logstelemetryschema.DefaultFullTextColumn,
 		fl,
 		nil,
-		false,
-		100000,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 	)
 
 	for _, c := range cases {
@@ -374,8 +374,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
 		logstelemetryschema.DefaultFullTextColumn,
 		fl,
 		nil,
-		false,
-		100000,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 	)
 
 	for _, c := range cases {
@@ -525,8 +524,7 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
 		logstelemetryschema.DefaultFullTextColumn,
 		fl,
 		nil,
-		false,
-		100000,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 	)
 
 	for _, c := range cases {
@@ -603,8 +601,7 @@ func TestStatementBuilderTimeSeriesBodyGroupBy(t *testing.T) {
 		logstelemetryschema.DefaultFullTextColumn,
 		fl,
 		nil,
-		false,
-		100000,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 	)
 
 	for _, c := range cases {
@@ -700,8 +697,7 @@ func TestStatementBuilderListQueryServiceCollision(t *testing.T) {
 		logstelemetryschema.DefaultFullTextColumn,
 		fl,
 		nil,
-		false,
-		100000,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 	)
 
 	for _, c := range cases {
@@ -926,8 +922,7 @@ func TestAdjustKey(t *testing.T) {
 		logstelemetryschema.DefaultFullTextColumn,
 		fl,
 		nil,
-		false,
-		100000,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 	)
 
 	for _, c := range cases {
@@ -1073,8 +1068,7 @@ func TestStmtBuilderBodyField(t *testing.T) {
 				logstelemetryschema.DefaultFullTextColumn,
 				fl,
 				nil,
-				false,
-				100000,
+				statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 			)
 
 			q, err := statementBuilder.Build(context.Background(), valuer.UUID{}, 1747947419000, 1747983448000, c.requestType, c.query, nil)
@@ -1174,8 +1168,7 @@ func TestStmtBuilderBodyFullTextSearch(t *testing.T) {
 				logstelemetryschema.DefaultFullTextColumn,
 				fl,
 				nil,
-				false,
-				100000,
+				statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
 			)
 
 			q, err := statementBuilder.Build(context.Background(), valuer.UUID{}, 1747947419000, 1747983448000, c.requestType, c.query, nil)
@@ -1296,7 +1289,6 @@ func newSkipResourceFingerprintLogsBuilder(
 		logstelemetryschema.DefaultFullTextColumn,
 		fl,
 		telemetryStore,
-		skipEnable,
-		threshold,
+		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: skipEnable, Threshold: threshold}},
 	)
 }

pkg/telemetryschema/audittelemetryschema/condition_builder.go

@@ -134,7 +134,7 @@ func (c *conditionBuilder) ConditionFor(
 	sb *sqlbuilder.SelectBuilder,
 ) ([]string, []string, error) {
 
-	// has/hasAny/hasAll/hasToken are logs-body-only; reject for audit.
+	// has/hasAny/hasAll/hasToken/search are logs-only functions; reject for audit.
 	if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
 		return nil, nil, err
 	}

pkg/telemetryschema/logstelemetryschema/condition_builder.go

@@ -3,6 +3,7 @@ package logstelemetryschema
 import (
 	"context"
 	"fmt"
+	"regexp"
 
 	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -27,6 +28,52 @@ func NewConditionBuilder(fm qbtypes.FieldMapper, fl flagger.Flagger) *conditionB
 	return &conditionBuilder{fm: fm, fl: fl}
 }
 
+// conditionForSearch ORs a case-insensitive match of the needle across the key context's
+// searchable columns (unspecified context = every column).
+func (c *conditionBuilder) conditionForSearch(
+	ctx context.Context,
+	orgID valuer.UUID,
+	key *telemetrytypes.TelemetryFieldKey,
+	value any,
+	sb *sqlbuilder.SelectBuilder,
+) ([]string, []string, error) {
+	// QuoteMeta + LOWER on both sides, not (?i): a literal match that can still use the
+	// LOWER(toString(body_v2)) skip index.
+	needle := regexp.QuoteMeta(fmt.Sprintf("%v", value))
+
+	useJSONBody := c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID))
+
+	var conditions []string
+
+	for _, col := range searchColumns(key.FieldContext, useJSONBody) {
+		switch col.Type.GetType() {
+		case schema.ColumnTypeEnumMap:
+			keysExpr := fmt.Sprintf("mapKeys(%s)", col.Name)
+			valsExpr := fmt.Sprintf("mapValues(%s)", col.Name)
+			// match() needs a String array; cast non-string map values first.
+			if mc, ok := col.Type.(schema.MapColumnType); ok && mc.ValueType.GetType() != schema.ColumnTypeEnumString {
+				valsExpr = fmt.Sprintf("arrayMap(x -> toString(x), mapValues(%s))", col.Name)
+			}
+			conditions = append(conditions, sb.Or(
+				fmt.Sprintf("arrayExists(x -> match(LOWER(x), LOWER(%s)), %s)", sb.Var(needle), keysExpr),
+				fmt.Sprintf("arrayExists(x -> match(LOWER(x), LOWER(%s)), %s)", sb.Var(needle), valsExpr),
+			))
+		case schema.ColumnTypeEnumJSON:
+			conditions = append(conditions, fmt.Sprintf("match(LOWER(toString(%s)), LOWER(%s))", col.Name, sb.Var(needle)))
+		case schema.ColumnTypeEnumString, schema.ColumnTypeEnumLowCardinality:
+			conditions = append(conditions, fmt.Sprintf("match(LOWER(%s), LOWER(%s))", col.Name, sb.Var(needle)))
+		default:
+			return nil, nil, errors.NewInternalf(errors.CodeInternal, "search does not support the column type of %q", col.Name)
+		}
+	}
+
+	if len(conditions) == 0 {
+		return nil, nil, nil
+	}
+	// The advisory rides on CostGuard (set by the visitor), not warnings.
+	return []string{sb.Or(conditions...)}, nil, nil
+}
+
 // isBodyJSONSearch reports whether a key addresses a path within the body JSON. Only
 // an explicit Body context qualifies; a bare, context-less `body` (e.g. full-text
 // `count_distinct(body)` or `body EXISTS`) is a full-text match, not a `$.body` path.
@@ -408,6 +455,11 @@ func (c *conditionBuilder) ConditionFor(
 	matches := querybuilder.MatchingFieldKeys(key, fieldKeys)
 	skipResourceFilter := options.SkipResourceFilter
 
+	// search() resolves its own (optional) scope; handle it before key resolution.
+	if operator == qbtypes.FilterOperatorSearch {
+		return c.conditionForSearch(ctx, orgID, key, value, sb)
+	}
+
 	keys, warning := querybuilder.ResolveKeys(key, matches)
 	var warnings []string
 	if warning != "" {

pkg/telemetryschema/logstelemetryschema/field_mapper.go

@@ -635,3 +635,37 @@ func (m *fieldMapper) existsExpressionFor(
 	}
 	return querybuilder.ExistsExpression(columns, key, tsStart, tsEnd, fieldExpression, exists)
 }
+
+// searchColumns is the single source of truth for the columns search() fans out across,
+// by context; body is body_v2 when useJSONBody, else the legacy body string.
+func searchColumns(fieldContext telemetrytypes.FieldContext, useJSONBody bool) []*schema.Column {
+	switch fieldContext {
+	case telemetrytypes.FieldContextLog:
+		return []*schema.Column{
+			logsV2Columns[LogsV2SeverityTextColumn],
+			logsV2Columns[LogsV2TraceIDColumn],
+			logsV2Columns[LogsV2SpanIDColumn],
+		}
+	case telemetrytypes.FieldContextBody:
+		if useJSONBody {
+			return []*schema.Column{logsV2Columns[LogsV2BodyV2Column]}
+		}
+		return []*schema.Column{logsV2Columns[LogsV2BodyColumn]}
+	case telemetrytypes.FieldContextAttribute:
+		return []*schema.Column{
+			logsV2Columns[LogsV2AttributesStringColumn],
+			logsV2Columns[LogsV2AttributesNumberColumn],
+			logsV2Columns[LogsV2AttributesBoolColumn],
+		}
+	case telemetrytypes.FieldContextResource:
+		return []*schema.Column{
+			logsV2Columns[LogsV2ResourcesStringColumn],
+		}
+	default:
+		columns := searchColumns(telemetrytypes.FieldContextLog, useJSONBody)
+		columns = append(columns, searchColumns(telemetrytypes.FieldContextBody, useJSONBody)...)
+		columns = append(columns, searchColumns(telemetrytypes.FieldContextAttribute, useJSONBody)...)
+		columns = append(columns, searchColumns(telemetrytypes.FieldContextResource, useJSONBody)...)
+		return columns
+	}
+}

pkg/telemetryschema/logstelemetryschema/filter_expr_search_test.go

@@ -0,0 +1,284 @@
+package logstelemetryschema
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"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/types/telemetrytypes"
+	"github.com/huandu/go-sqlbuilder"
+	"github.com/stretchr/testify/require"
+)
+
+// searchFanOut returns the WHERE fragment search() fans out to; bodyExpr differs
+// between the legacy string body and the body_v2 JSON column.
+func searchFanOut(bodyExpr string) string {
+	return "(match(LOWER(severity_text), LOWER(?)) OR match(LOWER(trace_id), LOWER(?)) OR match(LOWER(span_id), LOWER(?)) OR " +
+		bodyExpr + " OR " +
+		"(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(attributes_string)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), mapValues(attributes_string))) OR " +
+		"(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(attributes_number)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), arrayMap(x -> toString(x), mapValues(attributes_number)))) OR " +
+		"(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(attributes_bool)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), arrayMap(x -> toString(x), mapValues(attributes_bool)))) OR " +
+		"(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(resources_string)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), mapValues(resources_string))))"
+}
+
+// searchArgs returns v once per bound parameter search() emits — one per searchable
+// column expression (currently 12).
+func searchArgs(v any) []any {
+	const searchColumnParams = 12
+	args := make([]any, searchColumnParams)
+	for i := range args {
+		args[i] = v
+	}
+	return args
+}
+
+// TestFilterExprSearch covers search('needle') fanning out across every searchable
+// column via FilterOperatorSearch.
+func TestFilterExprSearch(t *testing.T) {
+	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
+	inWindowStart := uint64(releaseTime.Add(-5 * time.Minute).UnixNano())
+	inWindowEnd := uint64(releaseTime.Add(5 * time.Minute).UnixNano())
+
+	legacyBody := "match(LOWER(body), LOWER(?))"
+	jsonBody := "match(LOWER(toString(body_v2)), LOWER(?))"
+
+	// Single-context scope fragments (the fan-out narrowed to one context).
+	logScope := "(match(LOWER(severity_text), LOWER(?)) OR match(LOWER(trace_id), LOWER(?)) OR match(LOWER(span_id), LOWER(?)))"
+	resourceScope := "(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(resources_string)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), mapValues(resources_string)))"
+
+	serviceNameEq := "(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? " +
+		"AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)"
+
+	testCases := []struct {
+		name                  string
+		query                 string
+		jsonBodyEnabled       bool
+		fullTextColumn        *telemetrytypes.TelemetryFieldKey
+		startNs               uint64
+		endNs                 uint64
+		shouldPass            bool
+		expectedQuery         string
+		expectedArgs          []any
+		expectWarning         bool
+		expectedErrorContains string
+	}{
+		{
+			name:           "quoted, legacy body",
+			query:          "search('error')",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("error"),
+			expectWarning:  true,
+		},
+		{
+			name:            "quoted, json body",
+			query:           "search('error')",
+			jsonBodyEnabled: true,
+			fullTextColumn:  DefaultFullTextColumn,
+			startNs:         inWindowStart,
+			endNs:           inWindowEnd,
+			shouldPass:      true,
+			expectedQuery:   "WHERE " + searchFanOut(jsonBody),
+			expectedArgs:    searchArgs("error"),
+			expectWarning:   true,
+		},
+		{
+			name:           "bare word",
+			query:          "search(timeout)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("timeout"),
+			expectWarning:  true,
+		},
+		{
+			name:           "negated",
+			query:          "NOT search('error')",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE NOT (" + searchFanOut(legacyBody) + ")",
+			expectedArgs:   searchArgs("error"),
+			expectWarning:  true,
+		},
+		{
+			name:           "combined with field filter",
+			query:          "search('error') AND service.name=\"api\"",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE (" + searchFanOut(legacyBody) + " AND " + serviceNameEq + ")",
+			expectedArgs:   append(searchArgs("error"), "api"),
+			expectWarning:  true,
+		},
+		{
+			// The builder caps no window; the querier's estimate gate bounds scan cost.
+			name:           "wide window builds (estimate gate lives in querier)",
+			query:          "search('error')",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        uint64(releaseTime.Add(-10 * time.Hour).UnixNano()),
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("error"),
+			expectWarning:  true,
+		},
+		{
+			// fullTextColumn governs only bare/quoted free text, so search() must
+			// work when it is unset.
+			name:           "independent of full text column",
+			query:          "search('error')",
+			fullTextColumn: nil,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("error"),
+			expectWarning:  true,
+		},
+		{
+			// The bare word is the literal needle; Normalize would strip "resource.".
+			name:           "bare word with context prefix is not normalized",
+			query:          "search(resource.deployment)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("resource\\.deployment"),
+			expectWarning:  true,
+		},
+		{
+			// Literal digits, not %v of a parsed float64 (which would scan "1e+06").
+			name:           "numeric needle is not scientific notation",
+			query:          "search(1000000)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("1000000"),
+			expectWarning:  true,
+		},
+		{
+			name:           "scoped to body, legacy",
+			query:          "search('error', body)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE (" + legacyBody + ")",
+			expectedArgs:   []any{"error"},
+			expectWarning:  true,
+		},
+		{
+			name:            "scoped to body, json",
+			query:           "search('error', body)",
+			jsonBodyEnabled: true,
+			fullTextColumn:  DefaultFullTextColumn,
+			startNs:         inWindowStart,
+			endNs:           inWindowEnd,
+			shouldPass:      true,
+			expectedQuery:   "WHERE (" + jsonBody + ")",
+			expectedArgs:    []any{"error"},
+			expectWarning:   true,
+		},
+		{
+			name:           "scoped to resource (quoted scope)",
+			query:          "search('error', 'resource')",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE (" + resourceScope + ")",
+			expectedArgs:   []any{"error", "error"},
+			expectWarning:  true,
+		},
+		{
+			name:           "scoped to log fields",
+			query:          "search('error', log)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + logScope,
+			expectedArgs:   []any{"error", "error", "error"},
+			expectWarning:  true,
+		},
+		{
+			name:           "scoped to multiple contexts",
+			query:          "search('error', body, resource)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE ((" + legacyBody + ") OR (" + resourceScope + "))",
+			expectedArgs:   []any{"error", "error", "error"},
+			expectWarning:  true,
+		},
+		{
+			name:                  "invalid scope",
+			query:                 "search('error', 'timeout')",
+			fullTextColumn:        DefaultFullTextColumn,
+			startNs:               inWindowStart,
+			endNs:                 inWindowEnd,
+			shouldPass:            false,
+			expectedErrorContains: "invalid search scope",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fl := flaggertest.WithBooleanFlags(t, map[string]bool{
+				flagger.FeatureUseJSONBody.String(): tc.jsonBodyEnabled,
+			})
+			fm := NewFieldMapper(fl)
+			cb := NewConditionBuilder(fm, fl)
+			keys := BuildCompleteFieldKeyMap(releaseTime)
+
+			opts := querybuilder.FilterExprVisitorOpts{
+				Context:          context.Background(),
+				Logger:           instrumentationtest.New().Logger(),
+				FieldMapper:      fm,
+				ConditionBuilder: cb,
+				FieldKeys:        keys,
+				FullTextColumn:   tc.fullTextColumn,
+				StartNs:          tc.startNs,
+				EndNs:            tc.endNs,
+			}
+
+			clause, err := querybuilder.PrepareWhereClause(tc.query, opts)
+
+			if !tc.shouldPass {
+				require.Error(t, err)
+				require.True(t, detailContains(err, tc.expectedErrorContains),
+					"error %v should contain %q", err, tc.expectedErrorContains)
+				return
+			}
+
+			require.NoError(t, err)
+			require.False(t, clause.IsEmpty())
+
+			sql, args := clause.WhereClause.BuildWithFlavor(sqlbuilder.ClickHouse)
+			require.Equal(t, tc.expectedQuery, sql)
+			require.Equal(t, tc.expectedArgs, args)
+
+			if tc.expectWarning {
+				// The visitor only flags the guard; the statement builder
+				// materializes it from config.
+				require.True(t, clause.RequiresCostGuard)
+			}
+		})
+	}
+}

pkg/telemetryschema/metricstelemetryschema/condition_builder.go

@@ -157,7 +157,7 @@ func (c *conditionBuilder) ConditionFor(
 	sb *sqlbuilder.SelectBuilder,
 ) ([]string, []string, error) {
 
-	// has/hasAny/hasAll/hasToken are logs-body-only; reject for metrics.
+	// has/hasAny/hasAll/hasToken/search are logs-only functions; reject for metrics.
 	if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
 		return nil, nil, err
 	}

pkg/telemetryschema/tracestelemetryschema/condition_builder.go

@@ -205,7 +205,7 @@ func (c *conditionBuilder) ConditionFor(
 	sb *sqlbuilder.SelectBuilder,
 ) ([]string, []string, error) {
 
-	// has/hasAny/hasAll/hasToken are logs-body-only; reject for traces.
+	// has/hasAny/hasAll/hasToken/search are logs-only functions; reject for traces.
 	if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
 		return nil, nil, err
 	}

pkg/types/querybuildertypes/querybuildertypesv5/builder_elements.go

@@ -118,6 +118,10 @@ const (
 	FilterOperatorHasToken
 	FilterOperatorHasAny
 	FilterOperatorHasAll
+
+	// FilterOperatorSearch backs search('needle'): keyless, fanned out by the condition
+	// builder across every searchable column.
+	FilterOperatorSearch
 )
 
 var operatorInverseMapping = map[FilterOperator]FilterOperator{
@@ -186,7 +190,8 @@ func (f FilterOperator) IsNegativeOperator() bool {
 		FilterOperatorIn,
 		FilterOperatorExists,
 		FilterOperatorRegexp,
-		FilterOperatorContains:
+		FilterOperatorContains,
+		FilterOperatorSearch:
 		return false
 	}
 	return true
@@ -243,10 +248,11 @@ func (f FilterOperator) IsArrayFunctionOperator() bool {
 }
 
 // IsFunctionOperator reports whether the operator is a query function
-// (has/hasAny/hasAll/hasToken); these apply to the logs body column only.
+// (has/hasAny/hasAll/hasToken/search) — logs-only, and skipped by the
+// resource-fingerprint builder.
 func (f FilterOperator) IsFunctionOperator() bool {
 	switch f {
-	case FilterOperatorHas, FilterOperatorHasAny, FilterOperatorHasAll, FilterOperatorHasToken:
+	case FilterOperatorHas, FilterOperatorHasAny, FilterOperatorHasAll, FilterOperatorHasToken, FilterOperatorSearch:
 		return true
 	default:
 		return false
@@ -265,6 +271,8 @@ func (f FilterOperator) FunctionName() string {
 		return "hasAll"
 	case FilterOperatorHasToken:
 		return "hasToken"
+	case FilterOperatorSearch:
+		return "search"
 	default:
 		return ""
 	}

pkg/types/querybuildertypes/querybuildertypesv5/qb.go

@@ -57,6 +57,12 @@ type Statement struct {
 	Args           []any
 	Warnings       []string
 	WarningsDocURL string
+	CostGuard      *CostGuard
+}
+
+type CostGuard struct {
+	Warning     string
+	MaxScanRows int64
 }
 
 // StatementBuilder builds the query.

pkg/types/telemetrytypes/field_context.go

@@ -79,6 +79,14 @@ var (
 	}
 )
 
+// FieldContextFromText resolves a context word with the same aliases as key parsing
+// ("tag" -> attribute). ok is false for an unknown word, so callers can reject it
+// rather than get unspecified.
+func FieldContextFromText(text string) (FieldContext, bool) {
+	fc, ok := fieldContexts[strings.ToLower(strings.TrimSpace(text))]
+	return fc, ok
+}
+
 // UnmarshalJSON implements the json.Unmarshaler interface.
 func (f *FieldContext) UnmarshalJSON(data []byte) error {
 	var str string

pkg/variables/variable_replace_visitor.go

@@ -412,10 +412,15 @@ func (v *variableReplacementVisitor) VisitFunctionCall(ctx *grammar.FunctionCall
 }
 
 func (v *variableReplacementVisitor) VisitSearchCall(ctx *grammar.SearchCallContext) any {
-	if ctx.FunctionParamList() == nil {
+	if ctx.ValueList() == nil {
 		return "search()"
 	}
-	return "search(" + v.Visit(ctx.FunctionParamList()).(string) + ")"
+	// VisitValueList already parenthesizes the args and propagates the __all__ marker.
+	result := v.Visit(ctx.ValueList()).(string)
+	if result == specialSkipMarker {
+		return specialSkipMarker
+	}
+	return "search" + result
 }
 
 func (v *variableReplacementVisitor) VisitFunctionParamList(ctx *grammar.FunctionParamListContext) any {

tests/integration/tests/querier_json_body/05_search.py

@@ -0,0 +1,253 @@
+import json
+from collections import namedtuple
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logs import Logs
+from fixtures.querier import build_order_by, build_raw_query, get_rows, make_query_request
+
+# querierlogs/15_search.py with use_json_body on (see conftest.py): body matches run against
+# body_v2, the map/log fan-out is unchanged, and the response `body` comes back parsed — a
+# plain-string body is {"message": <body>}.
+
+Bodies = namedtuple("Bodies", ["a", "b", "c", "d"])
+
+
+@pytest.mark.parametrize(
+    "expression,expected",
+    [
+        # ── keyless: fans across every field ────────────────────────────────
+        pytest.param("search('login')", lambda b: {b.a}, id="keyless_body"),
+        pytest.param("search('checkout')", lambda b: {b.a, b.d}, id="keyless_body_and_resource"),
+        pytest.param("search('useast')", lambda b: {b.a, b.c}, id="keyless_resource_value"),
+        pytest.param("search('acme')", lambda b: {b.a, b.c}, id="keyless_attribute_value"),
+        pytest.param("search('tenant')", lambda b: {b.a, b.b, b.c, b.d}, id="keyless_attribute_key"),
+        pytest.param("search('error')", lambda b: {b.b, b.d}, id="keyless_severity_case_insensitive"),
+        pytest.param("search('CHECKOUT')", lambda b: {b.a, b.d}, id="keyless_needle_case_insensitive"),
+        # ── scoped: narrows to one context ──────────────────────────────────
+        pytest.param("search('login', body)", lambda b: {b.a}, id="scope_body"),
+        pytest.param("search('login', 'body')", lambda b: {b.a}, id="scope_body_quoted"),
+        pytest.param("search('checkout', body)", lambda b: {b.a}, id="scope_body_excludes_resource"),
+        pytest.param("search('checkout', resource)", lambda b: {b.a, b.d}, id="scope_resource"),
+        pytest.param("search('acme', attribute)", lambda b: {b.a, b.c}, id="scope_attribute"),
+        pytest.param("search('error', log)", lambda b: {b.b, b.d}, id="scope_log_severity"),
+        pytest.param("search('acme', body)", lambda b: set(), id="scope_body_no_match"),
+        pytest.param("search('checkout', attribute)", lambda b: set(), id="scope_attribute_no_match"),
+        # ── multiple scopes: union of the named contexts ────────────────────
+        pytest.param("search('login', body, resource)", lambda b: {b.a}, id="scopes_body_resource_body_only"),
+        pytest.param("search('checkout', body, resource)", lambda b: {b.a, b.d}, id="scopes_body_resource_union"),
+        # ── composition with boolean / field filters ────────────────────────
+        pytest.param("NOT search('login')", lambda b: {b.b, b.c, b.d}, id="negated"),
+        pytest.param("search('useast') AND severity_text = 'INFO'", lambda b: {b.a}, id="and_field_filter"),
+    ],
+)
+def test_search(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    expression: str,
+    expected: Callable[[Bodies], set[str]],
+) -> None:
+    """Four self-naming logs, each with a token planted in a distinct place (body,
+    resource, attribute, severity), assert search() reaches exactly the right ones."""
+    body = Bodies(
+        a="alpha checkout login ok",  # service checkout / region useast / tenant acme    / INFO
+        b="bravo declined",  # service payment  / region euwest / tenant globex  / ERROR
+        c="charlie miss",  # service cart     / region useast / tenant acme    / WARN
+        d="delta slow",  # service checkout / region apac   / tenant initech / ERROR
+    )
+    # (body, resources, attributes, severity_text)
+    specs = [
+        (body.a, {"service.name": "checkout", "region": "useast"}, {"tenant": "acme"}, "INFO"),
+        (body.b, {"service.name": "payment", "region": "euwest"}, {"tenant": "globex"}, "ERROR"),
+        (body.c, {"service.name": "cart", "region": "useast"}, {"tenant": "acme"}, "WARN"),
+        (body.d, {"service.name": "checkout", "region": "apac"}, {"tenant": "initech"}, "ERROR"),
+    ]
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    logs = [Logs(timestamp=now - timedelta(seconds=i + 1), resources=res, attributes=attrs, body=b, severity_text=sev) for i, (b, res, attrs, sev) in enumerate(specs)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[
+            build_raw_query(
+                "A",
+                "logs",
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "desc"), build_order_by("id", "desc")],
+                limit=100,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["status"] == "success"
+    # body_v2 comes back parsed; a plain-string body is {"message": <body>}.
+    assert {row["data"]["body"]["message"] for row in get_rows(response)} == expected(body)
+
+
+@pytest.mark.parametrize(
+    "needle",
+    [
+        pytest.param("eve@acme.io", id="nested_string_value"),
+        pytest.param("503", id="nested_numeric_value"),
+        pytest.param("status", id="nested_key"),
+    ],
+)
+def test_search_body_reaches_nested_json(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    needle: str,
+) -> None:
+    """A body-scoped search matches values and keys nested inside the body_v2 JSON."""
+    # searchable content lives only in nested fields, not a top-level message
+    nested_body = json.dumps({"user": {"email": "eve@acme.io"}, "http": {"status": 503}}, separators=(",", ":"))
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs([Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "api"}, body=nested_body, severity_text="INFO")])
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[build_raw_query("A", "logs", filter_expression=f"search('{needle}', body)", order=[build_order_by("timestamp", "desc")], limit=100)],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    rows = get_rows(response)
+    assert len(rows) == 1
+    assert rows[0]["data"]["body"]["user"]["email"] == "eve@acme.io"
+
+
+@pytest.mark.parametrize(
+    "expression",
+    [
+        pytest.param("search('login', bogus)", id="unknown_scope_word"),
+        pytest.param("search('login', body.message)", id="qualified_field_not_a_scope"),
+    ],
+)
+def test_search_invalid_scope_rejected(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    expression: str,
+) -> None:
+    """A scope that is not a field context (an unknown word, or a qualified
+    `context.field`) is rejected at build time with a 400 — even when, as with
+    `body.message`, it names a real body path."""
+    now = datetime.now(tz=UTC)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[build_raw_query("A", "logs", filter_expression=expression, limit=100)],
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "invalid search scope" in response.text
+
+
+# The querier gates search() on EXPLAIN ESTIMATE against search_max_scan_rows_json_body on
+# this path (50000 here, see conftest.py). Both recoveries the advisory suggests — a
+# selective filter, a narrower range — are exercised below.
+
+
+def test_search_cost_guard_trips_then_passes_with_filter(
+    signoz_search_scan_budget: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """A broad search() over ~61000 logs exceeds the 50000-row budget and is rejected;
+    the same search narrowed to the 1000-log 'checkout' service scans under budget and
+    succeeds — the advisory's "add a more selective filter" made real."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # 60000 'catalog' logs now + 1000 'checkout' logs ~45m back, in an earlier ts_bucket, so
+    # the checkout fingerprint owns its own marks and a resource filter on it prunes the scan.
+    logs = [Logs(timestamp=now - timedelta(seconds=1 + i % 30), resources={"service.name": "catalog"}, body="log line") for i in range(60000)]
+    logs += [Logs(timestamp=now - timedelta(minutes=45, seconds=i % 30), resources={"service.name": "checkout"}, body="log line") for i in range(1000)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=60)).timestamp() * 1000)
+    end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
+
+    def run(expression: str):
+        return make_query_request(
+            signoz_search_scan_budget,
+            token,
+            start_ms=start_ms,
+            end_ms=end_ms,
+            request_type="raw",
+            queries=[build_raw_query("A", "logs", filter_expression=expression, order=[build_order_by("timestamp", "desc")], limit=100)],
+        )
+
+    # Broad search over the whole range is over budget -> rejected before executing.
+    over_budget = run("search('log')")
+    assert over_budget.status_code == HTTPStatus.BAD_REQUEST, over_budget.text
+    assert "over the per-shard limit" in over_budget.text
+    # The advisory leads the suggestions, then how to get under budget.
+    assert "runs across all fields" in over_budget.text
+    assert "Narrow the time range or add a more selective filter." in over_budget.text
+
+    # Adding a selective resource filter prunes the scan under budget -> runs.
+    within_budget = run("search('log') AND resource.service.name = 'checkout'")
+    assert within_budget.status_code == HTTPStatus.OK, within_budget.text
+    assert len(get_rows(within_budget)) > 0
+
+
+def test_search_cost_guard_passes_with_narrower_time_range(
+    signoz_search_scan_budget: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """A steady stream of ~80000 logs (~4/sec over the last ~5.5h). A search() over the
+    whole window is over the 50000-row budget and rejected; the same search over the last
+    15 minutes scans only a few thousand rows and runs — the advisory's "narrow the time
+    range" made real."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # ~4 logs/sec, oldest ~5.5h back, newest ~now.
+    logs = [Logs(timestamp=now - timedelta(seconds=i // 4), resources={"service.name": "app"}, body="log line") for i in range(80000)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    def run(lookback_minutes: int):
+        return make_query_request(
+            signoz_search_scan_budget,
+            token,
+            start_ms=int((now - timedelta(minutes=lookback_minutes)).timestamp() * 1000),
+            end_ms=int((now + timedelta(minutes=1)).timestamp() * 1000),
+            request_type="raw",
+            queries=[build_raw_query("A", "logs", filter_expression="search('log')", order=[build_order_by("timestamp", "desc")], limit=100)],
+        )
+
+    # The last 6 hours cover the whole stream (~80000) -> over budget -> rejected.
+    wide = run(360)
+    assert wide.status_code == HTTPStatus.BAD_REQUEST, wide.text
+    assert "over the per-shard limit" in wide.text
+
+    # The last 15 minutes hold only ~3600 logs -> under budget -> runs, returning rows.
+    narrow = run(15)
+    assert narrow.status_code == HTTPStatus.OK, narrow.text
+    assert len(get_rows(narrow)) > 0

tests/integration/tests/querier_json_body/conftest.py

@@ -55,3 +55,34 @@ def signoz_json_body(
             "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_USE__JSON__BODY": True,
         },
     )
+
+
+@pytest.fixture(name="signoz_search_scan_budget", scope="package")
+def signoz_search_scan_budget(
+    network: Network,
+    migrator: types.Operation,  # pylint: disable=unused-argument
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    """The querierlogs budget instance over body_v2: same 50000 rows, but on
+    search_max_scan_rows_json_body. search_max_scan_rows keeps its 60M default, so only the
+    body_v2 budget can trip here. Shares the default instance's sqlstore + clickhouse, so
+    the same admin token and seeded logs work against it."""
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-json-body-search-scan-budget-50k",
+        env_overrides={
+            "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_USE__JSON__BODY": True,
+            "SIGNOZ_QUERIER_SEARCH__MAX__SCAN__ROWS__JSON__BODY": 50000,
+        },
+    )

tests/integration/tests/querier_skip_resource_fingerprint/conftest.py

@@ -31,8 +31,8 @@ def signoz_skip_resource_fingerprint(
         pytestconfig=pytestconfig,
         cache_key="signoz-skip-resource-fingerprint",
         env_overrides={
-            "SIGNOZ_STATEMENTBUILDER_SKIP__RESOURCE__FINGERPRINT_ENABLED": True,
-            "SIGNOZ_STATEMENTBUILDER_SKIP__RESOURCE__FINGERPRINT_THRESHOLD": 2,
+            "SIGNOZ_QUERIER_SKIP__RESOURCE__FINGERPRINT_ENABLED": True,
+            "SIGNOZ_QUERIER_SKIP__RESOURCE__FINGERPRINT_THRESHOLD": 2,
         },
     )
 
@@ -66,6 +66,6 @@ def signoz_fingerprint(
         pytestconfig=pytestconfig,
         cache_key="signoz-skip-resource-fingerprint-baseline",
         env_overrides={
-            "SIGNOZ_STATEMENTBUILDER_SKIP__RESOURCE__FINGERPRINT_ENABLED": False,
+            "SIGNOZ_QUERIER_SKIP__RESOURCE__FINGERPRINT_ENABLED": False,
         },
     )

tests/integration/tests/querierlogs/15_search.py

@@ -0,0 +1,212 @@
+from collections import namedtuple
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logs import Logs
+from fixtures.querier import build_order_by, build_raw_query, get_column_data_from_response, get_rows, make_query_request
+
+# search(): keyless fans across every field; scoped search('needle', <ctx>...) narrows to
+# the named contexts (body/attribute/resource/log). Flag off here, so body matches the
+# `body` String column (querier_json_body mirrors this over body_v2).
+
+Bodies = namedtuple("Bodies", ["a", "b", "c", "d"])
+
+
+@pytest.mark.parametrize(
+    "expression,expected",
+    [
+        # ── keyless: fans across every field ────────────────────────────────
+        pytest.param("search('login')", lambda b: {b.a}, id="keyless_body"),
+        pytest.param("search('checkout')", lambda b: {b.a, b.d}, id="keyless_body_and_resource"),
+        pytest.param("search('useast')", lambda b: {b.a, b.c}, id="keyless_resource_value"),
+        pytest.param("search('acme')", lambda b: {b.a, b.c}, id="keyless_attribute_value"),
+        pytest.param("search('tenant')", lambda b: {b.a, b.b, b.c, b.d}, id="keyless_attribute_key"),
+        pytest.param("search('error')", lambda b: {b.b, b.d}, id="keyless_severity_case_insensitive"),
+        pytest.param("search('CHECKOUT')", lambda b: {b.a, b.d}, id="keyless_needle_case_insensitive"),
+        # ── scoped: narrows to one context ──────────────────────────────────
+        pytest.param("search('login', body)", lambda b: {b.a}, id="scope_body"),
+        pytest.param("search('login', 'body')", lambda b: {b.a}, id="scope_body_quoted"),
+        pytest.param("search('checkout', body)", lambda b: {b.a}, id="scope_body_excludes_resource"),
+        pytest.param("search('checkout', resource)", lambda b: {b.a, b.d}, id="scope_resource"),
+        pytest.param("search('acme', attribute)", lambda b: {b.a, b.c}, id="scope_attribute"),
+        pytest.param("search('error', log)", lambda b: {b.b, b.d}, id="scope_log_severity"),
+        pytest.param("search('acme', body)", lambda b: set(), id="scope_body_no_match"),
+        pytest.param("search('checkout', attribute)", lambda b: set(), id="scope_attribute_no_match"),
+        # ── multiple scopes: union of the named contexts ────────────────────
+        pytest.param("search('login', body, resource)", lambda b: {b.a}, id="scopes_body_resource_body_only"),
+        pytest.param("search('checkout', body, resource)", lambda b: {b.a, b.d}, id="scopes_body_resource_union"),
+        # ── composition with boolean / field filters ────────────────────────
+        pytest.param("NOT search('login')", lambda b: {b.b, b.c, b.d}, id="negated"),
+        pytest.param("search('useast') AND severity_text = 'INFO'", lambda b: {b.a}, id="and_field_filter"),
+    ],
+)
+def test_search(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    expression: str,
+    expected: Callable[[Bodies], set[str]],
+) -> None:
+    """Four self-naming logs, each with a token planted in a distinct place (body,
+    resource, attribute, severity), assert search() reaches exactly the right ones."""
+    body = Bodies(
+        a="alpha checkout login ok",  # service checkout / region useast / tenant acme    / INFO
+        b="bravo declined",  # service payment  / region euwest / tenant globex  / ERROR
+        c="charlie miss",  # service cart     / region useast / tenant acme    / WARN
+        d="delta slow",  # service checkout / region apac   / tenant initech / ERROR
+    )
+    # (body, resources, attributes, severity_text)
+    specs = [
+        (body.a, {"service.name": "checkout", "region": "useast"}, {"tenant": "acme"}, "INFO"),
+        (body.b, {"service.name": "payment", "region": "euwest"}, {"tenant": "globex"}, "ERROR"),
+        (body.c, {"service.name": "cart", "region": "useast"}, {"tenant": "acme"}, "WARN"),
+        (body.d, {"service.name": "checkout", "region": "apac"}, {"tenant": "initech"}, "ERROR"),
+    ]
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    logs = [Logs(timestamp=now - timedelta(seconds=i + 1), resources=res, attributes=attrs, body=b, severity_text=sev) for i, (b, res, attrs, sev) in enumerate(specs)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[
+            build_raw_query(
+                "A",
+                "logs",
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "desc"), build_order_by("id", "desc")],
+                limit=100,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["status"] == "success"
+    assert set(get_column_data_from_response(response.json(), "body")) == expected(body)
+
+
+@pytest.mark.parametrize(
+    "expression",
+    [
+        pytest.param("search('login', bogus)", id="unknown_scope_word"),
+        pytest.param("search('login', body.message)", id="qualified_field_not_a_scope"),
+    ],
+)
+def test_search_invalid_scope_rejected(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    expression: str,
+) -> None:
+    """A scope that is not a field context (an unknown word, or a qualified
+    `context.field`) is rejected at build time with a 400."""
+    now = datetime.now(tz=UTC)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[build_raw_query("A", "logs", filter_expression=expression, limit=100)],
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "invalid search scope" in response.text
+
+
+# The querier gates search() on EXPLAIN ESTIMATE against search_max_scan_rows (50000 here,
+# see conftest.py). Both recoveries the advisory suggests — a selective filter, a narrower
+# range — are exercised below.
+
+
+def test_search_cost_guard_trips_then_passes_with_filter(
+    signoz_search_scan_budget: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """A broad search() over ~61000 logs exceeds the 50000-row budget and is rejected;
+    the same search narrowed to the 1000-log 'checkout' service scans under budget and
+    succeeds — the advisory's "add a more selective filter" made real."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # 60000 'catalog' logs now + 1000 'checkout' logs ~45m back, in an earlier ts_bucket, so
+    # the checkout fingerprint owns its own marks and a resource filter on it prunes the scan.
+    logs = [Logs(timestamp=now - timedelta(seconds=1 + i % 30), resources={"service.name": "catalog"}, body="log line") for i in range(60000)]
+    logs += [Logs(timestamp=now - timedelta(minutes=45, seconds=i % 30), resources={"service.name": "checkout"}, body="log line") for i in range(1000)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=60)).timestamp() * 1000)
+    end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
+
+    def run(expression: str):
+        return make_query_request(
+            signoz_search_scan_budget,
+            token,
+            start_ms=start_ms,
+            end_ms=end_ms,
+            request_type="raw",
+            queries=[build_raw_query("A", "logs", filter_expression=expression, order=[build_order_by("timestamp", "desc")], limit=100)],
+        )
+
+    # Broad search over the whole range is over budget -> rejected before executing.
+    over_budget = run("search('log')")
+    assert over_budget.status_code == HTTPStatus.BAD_REQUEST, over_budget.text
+    assert "over the per-shard limit" in over_budget.text
+    # The advisory leads the suggestions, then how to get under budget.
+    assert "runs across all fields" in over_budget.text
+    assert "Narrow the time range or add a more selective filter." in over_budget.text
+
+    # Adding a selective resource filter prunes the scan under budget -> runs.
+    within_budget = run("search('log') AND resource.service.name = 'checkout'")
+    assert within_budget.status_code == HTTPStatus.OK, within_budget.text
+    assert len(get_rows(within_budget)) > 0
+
+
+def test_search_cost_guard_passes_with_narrower_time_range(
+    signoz_search_scan_budget: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """A steady stream of ~80000 logs (~4/sec over the last ~5.5h). A search() over the
+    whole window is over the 50000-row budget and rejected; the same search over the last
+    15 minutes scans only a few thousand rows and runs — the advisory's "narrow the time
+    range" made real."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # ~4 logs/sec, oldest ~5.5h back, newest ~now.
+    logs = [Logs(timestamp=now - timedelta(seconds=i // 4), resources={"service.name": "app"}, body="log line") for i in range(80000)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    def run(lookback_minutes: int):
+        return make_query_request(
+            signoz_search_scan_budget,
+            token,
+            start_ms=int((now - timedelta(minutes=lookback_minutes)).timestamp() * 1000),
+            end_ms=int((now + timedelta(minutes=1)).timestamp() * 1000),
+            request_type="raw",
+            queries=[build_raw_query("A", "logs", filter_expression="search('log')", order=[build_order_by("timestamp", "desc")], limit=100)],
+        )
+
+    # The last 6 hours cover the whole stream (~80000) -> over budget -> rejected.
+    wide = run(360)
+    assert wide.status_code == HTTPStatus.BAD_REQUEST, wide.text
+    assert "over the per-shard limit" in wide.text
+
+    # The last 15 minutes hold only ~3600 logs -> under budget -> runs, returning rows.
+    narrow = run(15)
+    assert narrow.status_code == HTTPStatus.OK, narrow.text
+    assert len(get_rows(narrow)) > 0

tests/integration/tests/querierlogs/conftest.py

@@ -0,0 +1,34 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.signoz import create_signoz
+
+
+@pytest.fixture(name="signoz_search_scan_budget", scope="package")
+def signoz_search_scan_budget(
+    network: Network,
+    migrator: types.Operation,  # pylint: disable=unused-argument
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    """SigNoz with a low search_max_scan_rows (50000) so a broad search() trips the cost
+    guard while a selective one stays under it. Shares the default instance's sqlstore +
+    clickhouse, so the same admin token and seeded logs work against it."""
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-search-scan-budget-50k",
+        env_overrides={
+            "SIGNOZ_QUERIER_SEARCH__MAX__SCAN__ROWS": 50000,
+        },
+    )