feat(logs): scope search() to field contexts

search('needle', body, resource, ...) narrows the keyless full-text search to the named field contexts (body/attribute/resource/log). The first argument is the needle; any remaining arguments are scopes, quoted or bare. Keyless search('needle') still fans across every field.

Adds integration coverage for search() in querierlogs and querier_json_body: keyless and scoped forms, invalid-scope rejection, and the cost-guard trip/recover paths (add a filter, narrow the time range).

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,19 @@ 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 feature flags set to
+// the provided values (keyed by feature name, e.g. flagger.FeatureX.String()).
+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,70 @@ func (q *builderQuery[T]) Execute(ctx context.Context) (*qbtypes.Result, error)
 	return result, nil
 }
 
+// estimateRows runs EXPLAIN ESTIMATE for a cost-guarded statement and returns its
+// estimated total scan rows. guarded=false means there is nothing to enforce (no
+// CostGuard or a non-positive budget). A non-nil error means the query must be
+// rejected: either the estimate itself failed (fail closed — we cannot honor the
+// budget without it) or the parent context was cancelled (surfaced as-is). Callers
+// enforce the budget so it can be applied per-statement or cumulatively.
+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: this is a scan-heavy statement and without an estimate we
+		// cannot bound its cost, so running it unbounded is exactly what the guard
+		// exists to prevent.
+		reason := fmt.Sprintf("This query is too broad to plan within %s; narrow the time range or add a more selective filter.", estimateTimeout)
+		if estCtx.Err() != context.DeadlineExceeded {
+			reason = "Could not estimate this query's scan cost; narrow the time range or add a more selective filter."
+			q.logger.WarnContext(ctx, "EXPLAIN ESTIMATE failed; rejecting scan-heavy query (fail-closed)", errors.Attr(err))
+		}
+		return 0, true, errors.NewInvalidInputf(errors.CodeInvalidInput, "%s", withAdvisory(stmt.CostGuard.Warning, reason))
+	}
+
+	var rows int64
+	for _, e := range entries {
+		rows += e.Rows
+	}
+	return rows, true, nil
+}
+
+// enforceEstimate rejects a single scan-heavy statement (Statement.CostGuard) whose
+// EXPLAIN ESTIMATE rows exceed its 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, "%s",
+			withAdvisory(stmt.CostGuard.Warning, fmt.Sprintf("This query would scan about %d rows in this range, over the limit of %d; narrow the time range or add a more selective filter.", rows, budget)))
+	}
+	return nil
+}
+
+// withAdvisory prefixes reason with the requirement's advisory (e.g. the search()
+// warning) when present, so the rejection leads with why the query is expensive.
+func withAdvisory(advisory, reason string) string {
+	if advisory == "" {
+		return reason
+	}
+	return strings.TrimRight(advisory, ". ") + ". " + reason
+}
+
 // 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 +561,11 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
 	var warnings []string
 	var warningsDocURL string
 
+	// Cost guard is enforced against the cumulative estimate across the buckets we
+	// actually visit — a broad search() that fans every bucket must be bounded by
+	// the per-query budget, not let each bucket pass its slice independently.
+	var estimatedScan int64
+
 	for _, r := range buckets {
 		q.spec.Offset = 0
 		q.spec.Limit = need
@@ -501,6 +576,17 @@ 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, "%s",
+					withAdvisory(stmt.CostGuard.Warning, fmt.Sprintf("This query would scan about %d rows across the time range, over the limit of %d; narrow the time range or add a more selective filter.", estimatedScan, budget)))
+			}
+		}
 		// Execute with proper context for partial value detection
 		res, err := q.executeWithContext(ctx, stmt.Query, stmt.Args)
 		if err != nil {

pkg/querier/config.go

@@ -27,6 +27,8 @@ type Config struct {
 	SkipResourceFingerprint SkipResourceFingerprint `yaml:"skip_resource_fingerprint" mapstructure:"skip_resource_fingerprint"`
 	// 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"`
+	// SearchMaxScanRows caps the rows a search() query may scan, enforced via
+	SearchMaxScanRows int64 `yaml:"search_max_scan_rows" mapstructure:"search_max_scan_rows"`
 }
 
 // NewConfigFactory creates a new config factory for querier.
@@ -45,6 +47,7 @@ func newConfig() factory.Config {
 			Threshold: 100000,
 		},
 		LogTraceIDWindowPadding: 5 * time.Minute,
+		SearchMaxScanRows:       100_000_000,
 	}
 }
 
@@ -65,6 +68,9 @@ 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)
 	}
+	if c.SearchMaxScanRows < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "search_max_scan_rows must not be negative, got %v", c.SearchMaxScanRows)
+	}
 	return nil
 }
 

pkg/querier/signozquerier/provider.go

@@ -124,6 +124,7 @@ func newProvider(
 		telemetryStore,
 		cfg.SkipResourceFingerprint.Enabled,
 		cfg.SkipResourceFingerprint.Threshold,
+		telemetrylogs.WithSearchMaxScanRows(cfg.SearchMaxScanRows),
 	)
 
 	// Create audit statement builder

pkg/querybuilder/constants.go

@@ -8,6 +8,10 @@ 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. search() scans all fields,
+	// so it is slow and expensive; a specific field is cheaper.
+	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,79 @@ 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') and its scoped forms, e.g.
+// search('needle', body, resource). The first arg is the case-insensitive needle; the
+// rest are field-context scopes (bare or quoted). Emits one FilterOperatorSearch per
+// scope (none = keyless, covering every field) and ORs them.
 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 the raw token text of a search() argument (quoted or bare),
+// not the visited value — so a bare word stays literal and search(1000000) doesn't
+// become "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/telemetryaudit/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/telemetrylogs/condition_builder.go

@@ -3,6 +3,7 @@ package telemetrylogs
 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 searchable
+// columns of the key's field context (an unspecified context covers every column).
+func (c *conditionBuilder) conditionForSearch(
+	ctx context.Context,
+	orgID valuer.UUID,
+	key *telemetrytypes.TelemetryFieldKey,
+	value any,
+	sb *sqlbuilder.SelectBuilder,
+) ([]string, []string, error) {
+	// Literal case-insensitive substring match: QuoteMeta the needle and LOWER both sides
+	// so the body path can use the LOWER(toString(body_v2)) skip index (a (?i) regex could not).
+	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.
@@ -378,6 +425,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/telemetrylogs/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 field context. Body is body_v2 JSON when useJSONBody, else 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/telemetrylogs/filter_expr_search_test.go

@@ -0,0 +1,322 @@
+package telemetrylogs
+
+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"
+	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/huandu/go-sqlbuilder"
+	"github.com/stretchr/testify/require"
+)
+
+// searchFanOut returns the WHERE fragment search() fans out to. bodyExpr is the
+// body match expression, which 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 repeated 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 the search('needle') function, which fans 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,
+		},
+		{
+			// A wide window is allowed at build time; scan cost is bounded by the
+			// querier's EXPLAIN ESTIMATE gate, not a window cap in the builder.
+			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,
+		},
+		{
+			// search() is keyless and independent of fullTextColumn (which only
+			// governs bare/quoted free text). It must work even when 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,
+		},
+		{
+			// A context-prefixed bare word must be used as the literal needle, not
+			// normalized into a field key (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,
+		},
+		{
+			// A numeric needle must be the literal digits, not the %v rendering of a
+			// parsed float64 (which would make search(1000000) scan for "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 cost guard; the statement builder
+				// materializes the advisory + budget from config downstream.
+				require.True(t, clause.RequiresCostGuard)
+			}
+		})
+	}
+}
+
+// TestSearchCostGuard covers the search() path end-to-end through Build: the
+// statement carries a CostGuard with its scan budget and the advisory warning.
+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 := NewFieldMapper(fl)
+	cb := NewConditionBuilder(fm, fl)
+	store := telemetrytypestest.NewMockMetadataStore()
+	store.KeysMap = buildCompleteFieldKeyMap(releaseTime)
+	rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
+	sb := NewLogQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		store, fm, cb, rewriter, DefaultFullTextColumn, fl, nil, false, 100000,
+		WithSearchMaxScanRows(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)
+	require.Contains(t, stmt.Warnings, querybuilder.SearchWarning)
+}

pkg/telemetrylogs/statement_builder.go

@@ -29,11 +29,20 @@ type logQueryStatementBuilder struct {
 	fl                             flagger.Flagger
 	skipResourceFingerprintEnabled bool
 
-	fullTextColumn *telemetrytypes.TelemetryFieldKey
+	fullTextColumn    *telemetrytypes.TelemetryFieldKey
+	searchMaxScanRows int64
 }
 
 var _ qbtypes.StatementBuilder[qbtypes.LogAggregation] = (*logQueryStatementBuilder)(nil)
 
+type LogQueryStatementBuilderOption func(*logQueryStatementBuilder)
+
+// WithSearchMaxScanRows sets the estimated-rows budget the querier enforces for
+// search() statements (0 disables the gate).
+func WithSearchMaxScanRows(n int64) LogQueryStatementBuilderOption {
+	return func(b *logQueryStatementBuilder) { b.searchMaxScanRows = n }
+}
+
 func NewLogQueryStatementBuilder(
 	settings factory.ProviderSettings,
 	metadataStore telemetrytypes.MetadataStore,
@@ -45,6 +54,7 @@ func NewLogQueryStatementBuilder(
 	telemetryStore telemetrystore.TelemetryStore,
 	skipResourceFingerprintEnable bool,
 	skipResourceFingerprintThreshold uint64,
+	opts ...LogQueryStatementBuilderOption,
 ) *logQueryStatementBuilder {
 	logsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrylogs")
 
@@ -61,7 +71,7 @@ func NewLogQueryStatementBuilder(
 		skipResourceFingerprintThreshold,
 	)
 
-	return &logQueryStatementBuilder{
+	b := &logQueryStatementBuilder{
 		logger:                         logsSettings.Logger(),
 		metadataStore:                  metadataStore,
 		fm:                             fieldMapper,
@@ -72,6 +82,10 @@ func NewLogQueryStatementBuilder(
 		skipResourceFingerprintEnabled: skipResourceFingerprintEnable,
 		fullTextColumn:                 fullTextColumn,
 	}
+	for _, opt := range opts {
+		opt(b)
+	}
+	return b
 }
 
 // Build builds a SQL query for logs based on the given parameters.
@@ -117,9 +131,23 @@ func (b *logQueryStatementBuilder) Build(
 	}
 
 	stmt.Warnings = append(stmt.Warnings, warnings...)
+	// The search flag is gated in the condition builder; here the advisory rides on
+	// the statement so the querier can enforce the scan budget from the CostGuard.
+	if stmt.CostGuard != nil && stmt.CostGuard.Warning != "" {
+		stmt.Warnings = append(stmt.Warnings, stmt.CostGuard.Warning)
+	}
 	return stmt, nil
 }
 
+// costGuardFor builds the cost guard for a scan-heavy (search()) statement,
+// pairing the advisory with the configured scan budget. Returns nil otherwise.
+func (b *logQueryStatementBuilder) costGuardFor(required bool) *qbtypes.CostGuard {
+	if !required {
+		return nil
+	}
+	return &qbtypes.CostGuard{Warning: querybuilder.SearchWarning, MaxScanRows: b.searchMaxScanRows}
+}
+
 func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], bodyJSONEnabled bool) ([]*telemetrytypes.FieldKeySelector, []string) {
 	var keySelectors []*telemetrytypes.FieldKeySelector
 	var warnings []string
@@ -357,6 +385,7 @@ func (b *logQueryStatementBuilder) buildListQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil
@@ -523,6 +552,7 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil
@@ -650,6 +680,7 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil

pkg/telemetrymetrics/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/telemetrytraces/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'): a keyless search the condition
+	// builder fans out 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). These are logs-only; other signals reject
+// them and the resource-fingerprint builder skips them.
 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 (e.g. "body", "resource") to its
+// FieldContext, applying the same aliases as key parsing (e.g. "tag" -> attribute). ok
+// is false for an unknown word, so callers can reject it instead of getting 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,16 @@ 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 wraps the args in parens and returns the skip
+	// marker if any arg resolves to __all__.
+	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,124 @@
+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
+
+# search() with use_json_body on (see conftest.py): body matches run against body_v2 via
+# LOWER(toString(body_v2)); the map/log fan-out is unchanged from querierlogs/15_search.
+# The response `body` comes back as parsed JSON, so a plain-string body is {"message": <body>}.
+
+Bodies = namedtuple("Bodies", ["a", "b", "c", "d"])
+
+
+@pytest.mark.parametrize(
+    "expression,expected",
+    [
+        # body matches now run through body_v2 (toString of {"message": <body>})
+        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('CHECKOUT')", lambda b: {b.a, b.d}, id="keyless_case_insensitive"),
+        pytest.param("search('login', body)", lambda b: {b.a}, id="scope_body"),
+        pytest.param("search('checkout', body)", lambda b: {b.a}, id="scope_body_excludes_resource"),
+        # the map / log fan-out is flag-independent — sanity that it still works here
+        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('checkout', body, resource)", lambda b: {b.a, b.d}, id="scopes_union"),
+        pytest.param("NOT search('login')", lambda b: {b.b, b.c, b.d}, id="negated"),
+    ],
+)
+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 assert keyless/body-scoped search() matches through the
+    body_v2 JSON column, while resource/attribute/log scopes fan out unchanged."""
+    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
+    # 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"

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

@@ -0,0 +1,211 @@
+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). `expected` is a
+# lambda over the body markers so cases stay declarative while the dataset lives in one place.
+
+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
+
+# search() is scan-heavy, so the querier gates it on EXPLAIN ESTIMATE against
+# search_max_scan_rows (50000 for this instance, see conftest.py). A broad search over a
+# busy range is rejected; the two ways the advisory suggests to recover — add a more
+# selective filter, or narrow the time range — both bring the scan under budget.
+
+
+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 in the current bucket + 1000 'checkout' logs ~45m back (an
+    # earlier ts_bucket bucket), so the checkout fingerprint occupies 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 limit" 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 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,35 @@
+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 conservative search_max_scan_rows (50000) so a broad search() over a
+    busy range trips the cost guard, while a more selective one (by service or by a
+    narrower time range) 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,
+        },
+    )