refactor(logs): drop the needle term from the logs condition builder

The word carried three different meanings in one package: the substring a filter asserts over
the indexed text, the value a has-family filter looks for in an array, and the single token
hasToken matches. Each is now named for what it is - literal, element, token - so the reader
does not have to infer which one a given `needle` refers to.

Renames only. `castNeedleArray` and `legacyCoerceNeedle` follow their parameters, and the
hasToken error keeps its wording, which was always about tokens.

pkg/statementbuilder/logsstatementbuilder/stmt_builder_test.go

@@ -462,8 +462,8 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
 				Limit: 10,
 			},
 			expected: qbtypes.Statement{
-				Query:    "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE (JSON_VALUE(body, '$.\"status\"') = ? AND JSON_EXISTS(body, '$.\"status\"')) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:     []any{"success", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+				Query:    "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE ((JSON_VALUE(body, '$.\"status\"') = ? AND LOWER(body) LIKE LOWER(?)) AND (JSON_EXISTS(body, '$.\"status\"') AND LOWER(body) LIKE LOWER(?))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:     []any{"success", "%success%", "%\"status\"%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 				Warnings: []string{querybuilder.NewKeyNotFoundWarning("status")},
 			},
 			expectedErr: nil,
@@ -481,8 +481,8 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
 				Limit: 10,
 			},
 			expected: qbtypes.Statement{
-				Query:    "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE ((JSON_VALUE(body, '$.\"user_names\"[*]') = ?) AND JSON_EXISTS(body, '$.\"user_names\"[*]')) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:     []any{"john_doe", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+				Query:    "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE (((JSON_VALUE(body, '$.\"user_names\"[*]') = ? AND LOWER(body) LIKE LOWER(?))) AND (JSON_EXISTS(body, '$.\"user_names\"[*]') AND LOWER(body) LIKE LOWER(?))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:     []any{"john_doe", "%john\\_doe%", "%\"user\\_names\"%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 				Warnings: []string{querybuilder.NewKeyNotFoundWarning("user_names[*]")},
 			},
 			expectedErr: nil,
@@ -498,8 +498,8 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
 				Limit: 10,
 			},
 			expected: qbtypes.Statement{
-				Query:    "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE (has(JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$.\"user_names\"') = ? AND JSONType(body, 'user_names') NOT IN ('Array', 'Object', 'Null')), false)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:     []any{"john_doe", "john_doe", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+				Query:    "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE ((has(JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$.\"user_names\"') = ? AND JSONType(body, 'user_names') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:     []any{"john_doe", "john_doe", "%\"user\\_names\"%", "%john\\_doe%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 				Warnings: []string{querybuilder.NewKeyNotFoundWarning("user_names[*]")},
 			},
 			expectedErr: nil,
@@ -1011,8 +1011,8 @@ func TestStmtBuilderBodyField(t *testing.T) {
 			},
 			enableUseJSONBody: false,
 			expected: qbtypes.Statement{
-				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE body = ? AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{"", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+				Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE (body = ? AND LOWER(body) = LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{"", "", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 			},
 			expectedErr: nil,
 		},

pkg/telemetryschema/logstelemetryschema/condition_builder.go

@@ -105,14 +105,14 @@ func (c *conditionBuilder) conditionForArrayFunction(
 			"function `%s` supports only body JSON search", operator.FunctionName()).WithUrl(functionBodyJSONSearchDocURL)
 	}
 
-	needle := value
+	element := value
 	if args, ok := value.([]any); ok && len(args) > 0 {
-		needle = args[0]
+		element = args[0]
 	}
 
 	if c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
 		// JSON access plan: data-type collision handling, nested array paths.
-		valueType, needle := InferDataType(needle, operator, key)
+		valueType, element := InferDataType(element, operator, key)
 		// A not-found (synthesized) body path carries no metadata plan; build an exhaustive
 		// one so the query runs against the underlying data (with the not-found warning)
 		// instead of erroring, matching the regular-operator path.
@@ -123,23 +123,44 @@ func (c *conditionBuilder) conditionForArrayFunction(
 			}
 			key = keyCopy
 		}
-		return NewJSONConditionBuilder(key, valueType).buildArrayFunctionCondition(operator, needle, sb)
+		return NewJSONConditionBuilder(key, valueType).buildArrayFunctionCondition(operator, element, sb)
 	}
 
-	// legacy string-body path: type-matched array extraction, OR-ed with a scalar comparison
-	// for a scalar body value (coalesced to false so NOT has() matches missing-key rows).
-	elemType := legacyElemType(needle)
+	return c.legacyArrayFunctionCondition(key, operator, element, sb), nil
+}
+
+// legacyArrayFunctionCondition builds the has-family comparison over the plain body string, with
+// what it implies over the indexed LOWER(body): the path, since the extraction yields nothing for an
+// absent one, and each element, since it must appear in the text. The comparison decides the row.
+func (c *conditionBuilder) legacyArrayFunctionCondition(
+	key *telemetrytypes.TelemetryFieldKey,
+	operator qbtypes.FilterOperator,
+	element any,
+	sb *sqlbuilder.SelectBuilder,
+) string {
+	// type-matched array extraction, OR-ed with a scalar comparison for a scalar body value
+	// (coalesced to false so NOT has() matches missing-key rows).
+	elemType := legacyElemType(element)
 	arrayExpr := getBodyJSONArrayKey(key, elemType)
 	scalarExpr, scalarGuard, hasScalar := getBodyJSONScalarKey(key, elemType)
-	if list, ok := needle.([]any); ok {
-		vals := make([]any, len(list))
-		for i, v := range list {
-			vals[i] = legacyCoerceNeedle(v, elemType)
-		}
-		// Pin the needle array type to the haystack; scalar fallback below coerces value-level.
-		arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, castNeedleArray(elemType, sb.Var(vals)))
+
+	elements, isList := element.([]any)
+	if !isList {
+		elements = []any{element}
+	}
+	vals := make([]any, len(elements))
+	for i, v := range elements {
+		vals[i] = legacyCoerceElement(v, elemType)
+	}
+
+	var cond string
+	switch {
+	case isList:
+		// Pin the element array type to the array it is tested against; scalar fallback below coerces value-level.
+		arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, castElementArray(elemType, sb.Var(vals)))
 		if !hasScalar {
-			return arrayCond, nil
+			cond = arrayCond
+			break
 		}
 		var membership string
 		if operator == qbtypes.FilterOperatorHasAll {
@@ -151,19 +172,75 @@ func (c *conditionBuilder) conditionForArrayFunction(
 		} else {
 			membership = sb.In(scalarExpr, vals...)
 		}
-		return fmt.Sprintf("(%s OR ifNull(%s, false))", arrayCond, sb.And(membership, scalarGuard)), nil
+		cond = fmt.Sprintf("(%s OR ifNull(%s, false))", arrayCond, sb.And(membership, scalarGuard))
+	default:
+		arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, sb.Var(vals[0]))
+		if !hasScalar {
+			cond = arrayCond
+			break
+		}
+		cond = fmt.Sprintf("(%s OR ifNull(%s, false))", arrayCond, sb.And(sb.E(scalarExpr, vals[0]), scalarGuard))
 	}
-	typedNeedle := legacyCoerceNeedle(needle, elemType)
-	arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, sb.Var(typedNeedle))
-	if !hasScalar {
-		return arrayCond, nil
+
+	predicates := []string{cond}
+	if predicate := bodyIndexPredicate(bodyPathLiterals(key), sb); predicate != "" {
+		predicates = append(predicates, predicate)
 	}
-	return fmt.Sprintf("(%s OR ifNull(%s, false))", arrayCond, sb.And(sb.E(scalarExpr, typedNeedle), scalarGuard)), nil
+	// only strings imply text: the family compares at the element type it infers, so a quoted
+	// number is still a number here and its digits need not appear in the body
+	if elemType == telemetrytypes.FieldDataTypeString {
+		predicates = append(predicates, elementLiterals(operator, elements, sb)...)
+	}
+	if len(predicates) > 1 {
+		return sb.And(predicates...)
+	}
+	return cond
 }
 
-// castNeedleArray pins an Int64 needle array to Array(Int64) so it matches the Array(Nullable(Int64))
-// haystack; without it a needle >= 2^32 binds as Array(UInt64) and hasAny/hasAll error (code 386).
-func castNeedleArray(elemType telemetrytypes.FieldDataType, arg string) string {
+// elementLiterals returns what the elements of a has-family filter imply about the body text. has
+// and hasAll require every one, so each becomes its own predicate; hasAny requires only one, so the
+// arms are ORed - and an element yielding no literal leaves that OR unassertable.
+func elementLiterals(operator qbtypes.FilterOperator, elements []any, sb *sqlbuilder.SelectBuilder) []string {
+	if operator == qbtypes.FilterOperatorHasAny {
+		// resolve every element before binding anything: one unusable element voids the whole OR
+		runSets := make([][]string, 0, len(elements))
+		for _, v := range elements {
+			str, ok := v.(string)
+			if !ok {
+				return nil
+			}
+			runs := jsonTextRuns(str)
+			if len(runs) == 0 {
+				return nil
+			}
+			runSets = append(runSets, runs)
+		}
+		arms := make([]string, 0, len(runSets))
+		for _, runs := range runSets {
+			arms = append(arms, bodyIndexPredicate(runs, sb))
+		}
+		if len(arms) == 0 {
+			return nil
+		}
+		return []string{sb.Or(arms...)}
+	}
+
+	var predicates []string
+	for _, v := range elements {
+		str, ok := v.(string)
+		if !ok {
+			continue
+		}
+		if predicate := bodyIndexPredicate(jsonTextRuns(str), sb); predicate != "" {
+			predicates = append(predicates, predicate)
+		}
+	}
+	return predicates
+}
+
+// castElementArray pins an Int64 element array to Array(Int64) so it matches the Array(Nullable(Int64))
+// it is tested against; without it an element >= 2^32 binds as Array(UInt64) and hasAny/hasAll error (code 386).
+func castElementArray(elemType telemetrytypes.FieldDataType, arg string) string {
 	if elemType == telemetrytypes.FieldDataTypeInt64 {
 		return fmt.Sprintf("CAST(%s AS Array(Int64))", arg)
 	}
@@ -191,24 +268,24 @@ func (c *conditionBuilder) conditionForHasToken(
 	value any,
 	sb *sqlbuilder.SelectBuilder,
 ) (string, error) {
-	// hasToken takes a single needle; unwrap it from the function-argument slice.
-	needle := value
+	// hasToken takes a single token; unwrap it from the function-argument slice.
+	token := value
 	if args, ok := value.([]any); ok && len(args) > 0 {
-		needle = args[0]
+		token = args[0]
 	}
 
 	// hasToken matches string tokens only.
-	needleStr, ok := needle.(string)
+	tokenStr, ok := token.(string)
 	if !ok {
 		return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
 			"function `hasToken` expects value parameter to be a string").WithUrl(hasTokenFunctionDocURL)
 	}
 
-	// A multi-token needle makes CH hasToken error (code 36); reject up front as a 400. Both modes flow here.
-	if sep, found := firstTokenSeparator(needleStr); found {
+	// A multi-token value makes CH hasToken error (code 36); reject up front as a 400. Both modes flow here.
+	if sep, found := firstTokenSeparator(tokenStr); found {
 		return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
 			"function `hasToken` matches a single whole token, but %q contains the separator %q; use a substring filter (e.g. `body CONTAINS '%s'`) to search across separators",
-			needleStr, sep, needleStr).WithUrl(hasTokenFunctionDocURL)
+			tokenStr, sep, tokenStr).WithUrl(hasTokenFunctionDocURL)
 	}
 
 	bodyJSONEnabled := c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID))
@@ -219,7 +296,7 @@ func (c *conditionBuilder) conditionForHasToken(
 			return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
 				"function `hasToken` only supports body field as first parameter").WithUrl(hasTokenFunctionDocURL)
 		}
-		return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", LogsV2BodyColumn, sb.Var(needle)), nil
+		return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", LogsV2BodyColumn, sb.Var(token)), nil
 	}
 
 	// JSON mode: a bare body/body.message key searches the body.message column; any other body
@@ -228,7 +305,7 @@ func (c *conditionBuilder) conditionForHasToken(
 	// falls through and emits dynamicElement over the already-typed String column, which errors.
 	if key.Name == LogsV2BodyColumn || key.Name == bodyMessageField ||
 		(key.FieldContext == telemetrytypes.FieldContextBody && key.Name == messageSubField) {
-		return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", bodyMessageField, sb.Var(needle)), nil
+		return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", bodyMessageField, sb.Var(token)), nil
 	}
 	if key.FieldContext == telemetrytypes.FieldContextBody {
 		// A not-found (synthesized) body path carries no metadata plan; build an exhaustive
@@ -240,7 +317,7 @@ func (c *conditionBuilder) conditionForHasToken(
 			}
 			key = keyCopy
 		}
-		return NewJSONConditionBuilder(key, telemetrytypes.FieldDataTypeString).buildTokenFunctionCondition(needle, sb)
+		return NewJSONConditionBuilder(key, telemetrytypes.FieldDataTypeString).buildTokenFunctionCondition(token, sb)
 	}
 	return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
 		"function `hasToken` only supports the body field or a body JSON string field as first parameter").WithUrl(hasTokenFunctionDocURL)
@@ -269,6 +346,9 @@ func (c *conditionBuilder) conditionForResolvedKey(
 		return "", err
 	}
 
+	useJSONBody := c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID))
+	legacyBodyJSONSearch := isBodyJSONSearch(key, columns) && !useJSONBody
+
 	// has/hasAny/hasAll take the body-JSON path, not the normal operator paths.
 	if operator.IsArrayFunctionOperator() {
 		return c.conditionForArrayFunction(ctx, orgID, key, operator, value, columns, sb)
@@ -276,7 +356,7 @@ func (c *conditionBuilder) conditionForResolvedKey(
 
 	// TODO(Piyush): Update this to support multiple JSON columns based on evolutions
 	for _, column := range columns {
-		if column.Type.GetType() == schema.ColumnTypeEnumJSON && isBodyJSONSearch(key, columns) && c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) && key.Name != messageSubField {
+		if column.Type.GetType() == schema.ColumnTypeEnumJSON && isBodyJSONSearch(key, columns) && useJSONBody && key.Name != messageSubField {
 			valueType, value := InferDataType(value, operator, key)
 			if len(key.JSONPlan) == 0 {
 				keyCopy := telemetrytypes.NewTelemetryFieldKey(key.Name, key.FieldContext, key.FieldDataType)
@@ -305,19 +385,83 @@ func (c *conditionBuilder) conditionForResolvedKey(
 	}
 
 	// Check if this is a body JSON search (legacy string-body path, JSON flag off).
-	if isBodyJSONSearch(key, columns) && !c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
-		fieldExpression, value = GetBodyJSONKey(ctx, key, operator, value)
+	if legacyBodyJSONSearch {
+		return c.conditionForLegacyBodyJSON(ctx, orgID, startNs, endNs, key, operator, value, columns, sb)
 	}
 
 	fieldExpression, value = querybuilder.DataTypeCollisionHandledFieldName(key, value, fieldExpression, operator)
 
+	return c.conditionForOperator(ctx, orgID, startNs, endNs, key, operator, value, columns, fieldExpression, sb)
+}
+
+// conditionForLegacyBodyJSON renders a filter over a path inside the plain string body, with what it
+// implies over the indexed LOWER(body) - nothing such a filter compares matches an index expression.
+func (c *conditionBuilder) conditionForLegacyBodyJSON(
+	ctx context.Context,
+	orgID valuer.UUID,
+	startNs, endNs uint64,
+	key *telemetrytypes.TelemetryFieldKey,
+	operator qbtypes.FilterOperator,
+	value any,
+	columns []*schema.Column,
+	sb *sqlbuilder.SelectBuilder,
+) (string, error) {
+	if operator == qbtypes.FilterOperatorExists || operator == qbtypes.FilterOperatorNotExists {
+		exists := GetBodyJSONKeyForExists(ctx, key, operator, value)
+		if operator == qbtypes.FilterOperatorNotExists {
+			// matches the rows without the path, which say nothing about the body text
+			return "NOT " + exists, nil
+		}
+		if predicate := bodyIndexPredicate(bodyPathLiterals(key), sb); predicate != "" {
+			return sb.And(exists, predicate), nil
+		}
+		return exists, nil
+	}
+
+	fieldExpression, value := GetBodyJSONKey(ctx, key, operator, value)
+	fieldExpression, value = querybuilder.DataTypeCollisionHandledFieldName(key, value, fieldExpression, operator)
+	cond, err := c.conditionForOperator(ctx, orgID, startNs, endNs, key, operator, value, columns, fieldExpression, sb)
+	if err != nil {
+		return "", err
+	}
+	if predicate := bodyIndexPredicate(bodyValueLiterals(operator, value), sb); predicate != "" {
+		return sb.And(cond, predicate), nil
+	}
+	return cond, nil
+}
+
+// conditionForOperator renders the comparison itself, once the field expression and value have been
+// resolved for the column the key landed on.
+func (c *conditionBuilder) conditionForOperator(
+	ctx context.Context,
+	orgID valuer.UUID,
+	startNs, endNs uint64,
+	key *telemetrytypes.TelemetryFieldKey,
+	operator qbtypes.FilterOperator,
+	value any,
+	columns []*schema.Column,
+	fieldExpression string,
+	sb *sqlbuilder.SelectBuilder,
+) (string, error) {
+
 	// make use of case insensitive index for body
 	if fieldExpression == "body" || fieldExpression == messageSubColumn {
 		switch operator {
+		case qbtypes.FilterOperatorEqual:
+			// Bloom filters index lower(body), not the column; `=` still decides the row.
+			if _, ok := value.(string); ok && fieldExpression == LogsV2BodyColumn {
+				return sb.And(
+					sb.E(fieldExpression, value),
+					fmt.Sprintf("LOWER(%s) = LOWER(%s)", fieldExpression, sb.Var(value)),
+				), nil
+			}
 		case qbtypes.FilterOperatorLike:
-			return sb.ILike(fieldExpression, value), nil
-		case qbtypes.FilterOperatorNotLike:
-			return sb.NotILike(fieldExpression, value), nil
+			if _, ok := value.(string); ok && fieldExpression == LogsV2BodyColumn {
+				return sb.And(
+					sb.Like(fieldExpression, value),
+					sb.ILike(fieldExpression, value),
+				), nil
+			}
 		case qbtypes.FilterOperatorRegexp:
 			// Note: Escape $$ to $$$$ to avoid sqlbuilder interpreting materialized $ signs
 			// Only needed because we are using sprintf instead of sb.Match (not implemented in sqlbuilder)
@@ -356,12 +500,6 @@ func (c *conditionBuilder) conditionForResolvedKey(
 		return sb.NotILike(fieldExpression, value), nil
 
 	case qbtypes.FilterOperatorExists, qbtypes.FilterOperatorNotExists:
-		if isBodyJSONSearch(key, columns) && !c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
-			if operator == qbtypes.FilterOperatorExists {
-				return GetBodyJSONKeyForExists(ctx, key, operator, value), nil
-			}
-			return "NOT " + GetBodyJSONKeyForExists(ctx, key, operator, value), nil
-		}
 		pred, err := querybuilder.ExistsExpression(columns, key, startNs, endNs, fieldExpression, operator == qbtypes.FilterOperatorExists)
 		if err != nil {
 			return "", err

pkg/telemetryschema/logstelemetryschema/condition_builder_test.go

@@ -2,6 +2,7 @@ package logstelemetryschema
 
 import (
 	"context"
+	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"testing"
 	"time"
 
@@ -168,9 +169,9 @@ func TestConditionFor(t *testing.T) {
 				FieldContext: telemetrytypes.FieldContextLog,
 			},
 			operator:      qbtypes.FilterOperatorEqual,
-			value:         "error message",
-			expectedSQL:   "body = ?",
-			expectedArgs:  []any{"error message"},
+			value:         "Error Message",
+			expectedSQL:   "(body = ? AND LOWER(body) = LOWER(?))",
+			expectedArgs:  []any{"Error Message", "Error Message"},
 			expectedError: nil,
 		},
 		{
@@ -207,8 +208,8 @@ func TestConditionFor(t *testing.T) {
 			},
 			operator:      qbtypes.FilterOperatorLike,
 			value:         "%error%",
-			expectedSQL:   "LOWER(body) LIKE LOWER(?)",
-			expectedArgs:  []any{"%error%"},
+			expectedSQL:   "(body LIKE ? AND LOWER(body) LIKE LOWER(?))",
+			expectedArgs:  []any{"%error%", "%error%"},
 			expectedError: nil,
 		},
 		{
@@ -219,7 +220,7 @@ func TestConditionFor(t *testing.T) {
 			},
 			operator:      qbtypes.FilterOperatorNotLike,
 			value:         "%error%",
-			expectedSQL:   "LOWER(body) NOT LIKE LOWER(?)",
+			expectedSQL:   "body NOT LIKE ?",
 			expectedArgs:  []any{"%error%"},
 			expectedError: nil,
 		},
@@ -619,8 +620,8 @@ func TestConditionForMultipleKeys(t *testing.T) {
 			},
 			operator:      qbtypes.FilterOperatorEqual,
 			value:         "error message",
-			expectedSQL:   "body = ? AND severity_text = ?",
-			expectedArgs:  []any{"error message", "error message"},
+			expectedSQL:   "(body = ? AND LOWER(body) = LOWER(?)) AND severity_text = ?",
+			expectedArgs:  []any{"error message", "error message", "error message"},
 			expectedError: nil,
 		},
 	}
@@ -906,8 +907,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 	}
 }
 
-// IN on the body column routes each value back through the `=` path; the SQL it produces
-// must stay what the shared IN handling produced before, including for a mixed-type list.
+// IN on the body column routes each value back through the `=` path, so every arm picks up
+// the lower(body) companion — including the values a mixed-type list stringifies.
 func TestConditionForBodyIn(t *testing.T) {
 	testCases := []struct {
 		name         string
@@ -918,14 +919,14 @@ func TestConditionForBodyIn(t *testing.T) {
 		{
 			name:         "strings",
 			values:       []any{"alpha", "beta"},
-			expectedSQL:  "(body = ? OR body = ?)",
-			expectedArgs: []any{"alpha", "beta"},
+			expectedSQL:  "((body = ? AND LOWER(body) = LOWER(?)) OR (body = ? AND LOWER(body) = LOWER(?)))",
+			expectedArgs: []any{"alpha", "alpha", "beta", "beta"},
 		},
 		{
 			name:         "mixed types are stringified before they reach the column",
 			values:       []any{"alpha", float64(1), true},
-			expectedSQL:  "(body = ? OR body = ? OR body = ?)",
-			expectedArgs: []any{"alpha", "1", "true"},
+			expectedSQL:  "((body = ? AND LOWER(body) = LOWER(?)) OR (body = ? AND LOWER(body) = LOWER(?)) OR (body = ? AND LOWER(body) = LOWER(?)))",
+			expectedArgs: []any{"alpha", "alpha", "1", "1", "true", "true"},
 		},
 	}
 
@@ -954,3 +955,228 @@ func TestConditionForBodyIn(t *testing.T) {
 		})
 	}
 }
+
+// ClickHouse treats `\` as an escape only before `%`, `_` and itself.
+func TestLikePatternLiterals(t *testing.T) {
+	testCases := []struct {
+		name     string
+		pattern  string
+		expected []string
+	}{
+		{"contains wraps a plain value", "%error%", []string{"error"}},
+		{"wildcards split runs", "%foo%bar%", []string{"foo", "bar"}},
+		{"underscore splits too", "a_b", []string{"a", "b"}},
+		{"escaped wildcards stay literal", `%100\%\_off%`, []string{`100%_off`}},
+		{"escaped backslash collapses", `%C:\\tmp%`, []string{`C:\tmp`}},
+		{"backslash before other chars is literal", `%C:\tmp%`, []string{`C:\tmp`}},
+		{"trailing backslash is literal", `%path\`, []string{`path\`}},
+		{"no literals at all", "%_%", nil},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, likePatternLiterals(tc.pattern))
+		})
+	}
+}
+
+// The literals have to hold whichever encoder wrote the body, so the runs stop at every byte
+// one of them may rewrite.
+func TestJSONTextRuns(t *testing.T) {
+	testCases := []struct {
+		name     string
+		value    string
+		expected []string
+	}{
+		{"plain text is one run", "checkout failed", []string{"checkout failed"}},
+		{"a run with nothing to split on is kept whole", "abc", []string{"abc"}},
+		{"quote splits the run", `say "hello there"`, []string{"say ", "hello there"}},
+		{"slash splits the run, PHP escapes it", "/api/v1/users", []string{"api", "v1", "users"}},
+		{"ampersand and angles split, Go escapes them", "a&b<c>dddd", []string{"a", "b", "c", "dddd"}},
+		{"non-ascii splits, Python escapes it", "order café latte", []string{"order caf", " latte"}},
+		{"newline splits", "line one\nline two", []string{"line one", "line two"}},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, jsonTextRuns(tc.value))
+		})
+	}
+}
+
+func TestBodyPathLiterals(t *testing.T) {
+	testCases := []struct {
+		name     string
+		key      string
+		expected []string
+	}{
+		{"quoting lifts a short name over the ngram size", "id", []string{`"id"`}},
+		{"one literal per component", "response.status_code", []string{`"response"`, `"status_code"`}},
+		{"array suffixes are trimmed", "items[*].sku", []string{`"items"`, `"sku"`}},
+		{"every component is carried", "a.b.count", []string{`"a"`, `"b"`, `"count"`}},
+		{"a component an encoder may rewrite is dropped", "user/name.email", []string{`"email"`}},
+		{"nothing usable", "us/er.na/me", nil},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			key := telemetrytypes.NewTelemetryFieldKey(tc.key, telemetrytypes.FieldContextBody, telemetrytypes.FieldDataTypeUnspecified)
+			assert.Equal(t, tc.expected, bodyPathLiterals(key))
+		})
+	}
+}
+
+// The path literals ride on the existence assertion and the value literals on the comparison, so
+// a filter carries each at most once. Nothing rides on a negated operator: it matches rows
+// without the path, which say nothing about the body text.
+func TestLegacyBodyIndexPredicates(t *testing.T) {
+	testCases := []struct {
+		name         string
+		key          string
+		operator     qbtypes.FilterOperator
+		value        any
+		expected     string
+		expectedArgs []any
+	}{
+		{
+			name:         "exists carries the path",
+			key:          "user_id",
+			operator:     qbtypes.FilterOperatorExists,
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"user\_id"%`},
+		},
+		{
+			name:         "equality carries the value",
+			key:          "status",
+			operator:     qbtypes.FilterOperatorEqual,
+			value:        "timeout_error",
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%timeout\_error%`},
+		},
+		{
+			name:         "contains carries the value",
+			key:          "message",
+			operator:     qbtypes.FilterOperatorContains,
+			value:        "upstream refused",
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%upstream refused%`},
+		},
+		{
+			name:         "like carries one literal per run of the pattern",
+			key:          "message",
+			operator:     qbtypes.FilterOperatorLike,
+			value:        "%conn%refused%",
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%conn%refused%`},
+		},
+		{
+			name:     "has carries the path and the element",
+			key:      "tags[*]",
+			operator: qbtypes.FilterOperatorHas,
+			value:    []any{"production"},
+			// The element rides on its own predicate rather than being pinned next to the key:
+			// has() over the extracted array says nothing about where in the text it sits.
+			expected:     `LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"tags"%`, "%production%"},
+		},
+		{
+			name:         "hasAll carries one literal per element",
+			key:          "tags[*]",
+			operator:     qbtypes.FilterOperatorHasAll,
+			value:        []any{[]any{"production", "webserver"}},
+			expected:     `LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"tags"%`, "%production%", "%webserver%"},
+		},
+		{
+			// hasAny asks for one of the elements, so the arms are ORed — ANDing them would
+			// demand every element be present.
+			name:         "hasAny ORs the element literals",
+			key:          "tags[*]",
+			operator:     qbtypes.FilterOperatorHasAny,
+			value:        []any{[]any{"production", "webserver"}},
+			expected:     `LOWER(body) LIKE LOWER(?) AND (LOWER(body) LIKE LOWER(?) OR LOWER(body) LIKE LOWER(?))`,
+			expectedArgs: []any{`%"tags"%`, "%production%", "%webserver%"},
+		},
+		{
+			// one element with no usable literal voids the whole OR: the filter can still match
+			// through that element, so nothing about the text is implied.
+			name:         "hasAny drops the OR when an element carries no literal",
+			key:          "tags[*]",
+			operator:     qbtypes.FilterOperatorHasAny,
+			value:        []any{[]any{"production", "/"}},
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"tags"%`},
+		},
+		{
+			name:         "numeric elements carry nothing",
+			key:          "ids[*]",
+			operator:     qbtypes.FilterOperatorHasAny,
+			value:        []any{[]any{"9007199254740993", "9007199254740994"}},
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"ids"%`},
+		},
+		{
+			name:     "a number carries nothing",
+			key:      "user_id",
+			operator: qbtypes.FilterOperatorEqual,
+			value:    int64(123),
+		},
+		{
+			// IN delegates to `=`, so the literal rides each arm rather than the IN itself
+			name:         "IN carries one literal per arm it delegates to",
+			key:          "status",
+			operator:     qbtypes.FilterOperatorIn,
+			value:        []any{"timeout_error", "conn_refused"},
+			expected:     `(JSON_VALUE(body, '$."status"') = ? AND LOWER(body) LIKE LOWER(?)) OR (JSON_VALUE(body, '$."status"') = ? AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs: []any{`%timeout\_error%`, `%conn\_refused%`},
+		},
+		{
+			name:     "not equal carries nothing",
+			key:      "status",
+			operator: qbtypes.FilterOperatorNotEqual,
+			value:    "timeout_error",
+		},
+		{
+			name:     "not exists carries nothing",
+			key:      "user_id",
+			operator: qbtypes.FilterOperatorNotExists,
+		},
+		{
+			name:     "not contains carries nothing",
+			key:      "message",
+			operator: qbtypes.FilterOperatorNotContains,
+			value:    "upstream refused",
+		},
+	}
+
+	fl := flaggertest.New(t)
+	cb := NewConditionBuilder(NewFieldMapper(fl), fl)
+	ctx := context.Background()
+	columns := []*schema.Column{logsV2Columns[LogsV2BodyColumn]}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			sb := sqlbuilder.NewSelectBuilder()
+			sb.Select("1").From("t")
+			key := telemetrytypes.NewTelemetryFieldKey(tc.key, telemetrytypes.FieldContextBody, telemetrytypes.FieldDataTypeUnspecified)
+
+			var cond string
+			var err error
+			if tc.operator.IsArrayFunctionOperator() {
+				cond, err = cb.conditionForArrayFunction(ctx, valuer.UUID{}, key, tc.operator, tc.value, columns, sb)
+			} else {
+				cond, err = cb.conditionForLegacyBodyJSON(ctx, valuer.UUID{}, 0, 0, key, tc.operator, tc.value, columns, sb)
+			}
+			require.NoError(t, err)
+
+			sb.Where(cond)
+			query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+			if tc.expected == "" {
+				assert.NotContains(t, query, "LOWER(body) LIKE")
+				return
+			}
+			assert.Contains(t, query, tc.expected)
+			assert.Subset(t, args, tc.expectedArgs)
+		})
+	}
+}

pkg/telemetryschema/logstelemetryschema/filter_expr_logs_body_json_test.go

@@ -44,168 +44,169 @@ func TestFilterExprLogsBodyJSON(t *testing.T) {
 			category:              "json",
 			query:                 "has(body.requestor_list[*], 'index_service')",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (has(JSONExtract(JSON_QUERY(body, '$."requestor_list"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$."requestor_list"') = ? AND JSONType(body, 'requestor_list') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{"index_service", "index_service"},
+			expectedQuery:         `WHERE ((has(JSONExtract(JSON_QUERY(body, '$."requestor_list"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$."requestor_list"') = ? AND JSONType(body, 'requestor_list') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{"index_service", "index_service", "%\"requestor\\_list\"%", "%index\\_service%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "has(body.int_numbers[*], 2)",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (has(JSONExtract(JSON_QUERY(body, '$."int_numbers"[*]'), 'Array(Nullable(Float64))'), ?) OR ifNull((JSONExtract(JSON_VALUE(body, '$."int_numbers"'), 'Nullable(Float64)') = ? AND JSONType(body, 'int_numbers') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{float64(2), float64(2)},
+			expectedQuery:         `WHERE ((has(JSONExtract(JSON_QUERY(body, '$."int_numbers"[*]'), 'Array(Nullable(Float64))'), ?) OR ifNull((JSONExtract(JSON_VALUE(body, '$."int_numbers"'), 'Nullable(Float64)') = ? AND JSONType(body, 'int_numbers') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{float64(2), float64(2), "%\"int\\_numbers\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "has(body.bool[*], true)",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (has(JSONExtract(JSON_QUERY(body, '$."bool"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$."bool"') = ? AND JSONType(body, 'bool') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{"true", "true"},
+			expectedQuery:         `WHERE ((has(JSONExtract(JSON_QUERY(body, '$."bool"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$."bool"') = ? AND JSONType(body, 'bool') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{"true", "true", "%\"bool\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "NOT has(body.nested_num[*].float_nums[*], 2.2)",
 			shouldPass:            true,
-			expectedQuery:         `WHERE NOT (has(JSONExtract(JSON_QUERY(body, '$."nested_num"[*]."float_nums"[*]'), 'Array(Nullable(Float64))'), ?))`,
-			expectedArgs:          []any{float64(2.2)},
+			expectedQuery:         `WHERE NOT ((has(JSONExtract(JSON_QUERY(body, '$."nested_num"[*]."float_nums"[*]'), 'Array(Nullable(Float64))'), ?) AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{float64(2.2), "%\"nested\\_num\"%\"float\\_nums\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "has(body.tags, 'production')",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (has(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$."tags"') = ? AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{"production", "production"},
+			expectedQuery:         `WHERE ((has(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$."tags"') = ? AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{"production", "production", "%\"tags\"%", "%production%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "hasAny(body.tags, ['critical', 'test'])",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (hasAny(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$."tags"') IN (?, ?) AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{[]any{"critical", "test"}, "critical", "test"},
+			expectedQuery:         `WHERE ((hasAny(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$."tags"') IN (?, ?) AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?) AND (LOWER(body) LIKE LOWER(?) OR LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{[]any{"critical", "test"}, "critical", "test", "%\"tags\"%", "%critical%", "%test%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "hasAll(body.tags, ['production', 'web'])",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (hasAll(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR ifNull(((JSON_VALUE(body, '$."tags"') = ? AND JSON_VALUE(body, '$."tags"') = ?) AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{[]any{"production", "web"}, "production", "web"},
+			expectedQuery:         `WHERE ((hasAll(JSONExtract(JSON_QUERY(body, '$."tags"[*]'), 'Array(Nullable(String))'), ?) OR ifNull(((JSON_VALUE(body, '$."tags"') = ? AND JSON_VALUE(body, '$."tags"') = ?) AND JSONType(body, 'tags') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{[]any{"production", "web"}, "production", "web", "%\"tags\"%", "%production%", "%web%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "has(body.ids, \"200\")",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (has(JSONExtract(JSON_QUERY(body, '$."ids"[*]'), 'Array(Nullable(Int64))'), ?) OR ifNull((JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') = ? AND JSONType(body, 'ids') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{int64(200), int64(200)},
+			expectedQuery:         `WHERE ((has(JSONExtract(JSON_QUERY(body, '$."ids"[*]'), 'Array(Nullable(Int64))'), ?) OR ifNull((JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') = ? AND JSONType(body, 'ids') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{int64(200), int64(200), "%\"ids\"%"},
 			expectedErrorContains: "",
 		},
 		{
-			// Big-int needle CAST to Array(Int64) to match the haystack (else 386).
+			// Big-int element CAST to Array(Int64) to match the array it is tested against (else 386).
 			category:              "json",
 			query:                 `hasAny(body.ids, ['9007199254740993', '9007199254740994'])`,
 			shouldPass:            true,
-			expectedQuery:         `WHERE (hasAny(JSONExtract(JSON_QUERY(body, '$."ids"[*]'), 'Array(Nullable(Int64))'), CAST(? AS Array(Int64))) OR ifNull((JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') IN (?, ?) AND JSONType(body, 'ids') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{[]any{int64(9007199254740993), int64(9007199254740994)}, int64(9007199254740993), int64(9007199254740994)},
+			expectedQuery:         `WHERE ((hasAny(JSONExtract(JSON_QUERY(body, '$."ids"[*]'), 'Array(Nullable(Int64))'), CAST(? AS Array(Int64))) OR ifNull((JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') IN (?, ?) AND JSONType(body, 'ids') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{[]any{int64(9007199254740993), int64(9007199254740994)}, int64(9007199254740993), int64(9007199254740994), "%\"ids\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 `hasAll(body.ids, ['9007199254740993', '9007199254740994'])`,
 			shouldPass:            true,
-			expectedQuery:         `WHERE (hasAll(JSONExtract(JSON_QUERY(body, '$."ids"[*]'), 'Array(Nullable(Int64))'), CAST(? AS Array(Int64))) OR ifNull(((JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') = ? AND JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') = ?) AND JSONType(body, 'ids') NOT IN ('Array', 'Object', 'Null')), false))`,
-			expectedArgs:          []any{[]any{int64(9007199254740993), int64(9007199254740994)}, int64(9007199254740993), int64(9007199254740994)},
+			expectedQuery:         `WHERE ((hasAll(JSONExtract(JSON_QUERY(body, '$."ids"[*]'), 'Array(Nullable(Int64))'), CAST(? AS Array(Int64))) OR ifNull(((JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') = ? AND JSONExtract(JSON_VALUE(body, '$."ids"'), 'Nullable(Int64)') = ?) AND JSONType(body, 'ids') NOT IN ('Array', 'Object', 'Null')), false)) AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{[]any{int64(9007199254740993), int64(9007199254740994)}, int64(9007199254740993), int64(9007199254740994), "%\"ids\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "body.message = hello",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (JSON_VALUE(body, '$."message"') = ? AND JSON_EXISTS(body, '$."message"'))`,
-			expectedArgs:          []any{"hello"},
+			expectedQuery:         `WHERE ((JSON_VALUE(body, '$."message"') = ? AND LOWER(body) LIKE LOWER(?)) AND (JSON_EXISTS(body, '$."message"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{"hello", "%hello%", "%\"message\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "body.status = 1",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."status"'), 'Float64') = ? AND JSON_EXISTS(body, '$."status"'))`,
-			expectedArgs:          []any{float64(1)},
+			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."status"'), 'Float64') = ? AND (JSON_EXISTS(body, '$."status"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{float64(1), "%\"status\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "body.status = 1.1",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."status"'), 'Float64') = ? AND JSON_EXISTS(body, '$."status"'))`,
-			expectedArgs:          []any{float64(1.1)},
+			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."status"'), 'Float64') = ? AND (JSON_EXISTS(body, '$."status"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{float64(1.1), "%\"status\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "body.boolkey = true",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."boolkey"'), 'Bool') = ? AND JSON_EXISTS(body, '$."boolkey"'))`,
-			expectedArgs:          []any{true},
+			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."boolkey"'), 'Bool') = ? AND (JSON_EXISTS(body, '$."boolkey"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{true, "%\"boolkey\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "body.status > 200",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."status"'), 'Float64') > ? AND JSON_EXISTS(body, '$."status"'))`,
-			expectedArgs:          []any{float64(200)},
+			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."status"'), 'Float64') > ? AND (JSON_EXISTS(body, '$."status"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{float64(200), "%\"status\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "body.message REGEXP 'a*'",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (match(JSON_VALUE(body, '$."message"'), ?) AND JSON_EXISTS(body, '$."message"'))`,
-			expectedArgs:          []any{"a*"},
+			expectedQuery:         `WHERE (match(JSON_VALUE(body, '$."message"'), ?) AND (JSON_EXISTS(body, '$."message"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{"a*", "%\"message\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 `body.message CONTAINS "hello 'world'"`,
 			shouldPass:            true,
-			expectedQuery:         `WHERE (LOWER(JSON_VALUE(body, '$."message"')) LIKE LOWER(?) AND JSON_EXISTS(body, '$."message"'))`,
-			expectedArgs:          []any{"%hello 'world'%"},
+			expectedQuery:         `WHERE ((LOWER(JSON_VALUE(body, '$."message"')) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?)) AND (JSON_EXISTS(body, '$."message"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{"%hello 'world'%", "%hello 'world'%", "%\"message\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 `body.message EXISTS`,
 			shouldPass:            true,
-			expectedQuery:         `WHERE JSON_EXISTS(body, '$."message"')`,
+			expectedQuery:         `WHERE (JSON_EXISTS(body, '$."message"') AND LOWER(body) LIKE LOWER(?))`,
+			expectedArgs:          []any{"%\"message\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 `body.name IN ('hello', 'world')`,
 			shouldPass:            true,
-			expectedQuery:         `WHERE ((JSON_VALUE(body, '$."name"') = ? OR JSON_VALUE(body, '$."name"') = ?) AND JSON_EXISTS(body, '$."name"'))`,
-			expectedArgs:          []any{"hello", "world"},
+			expectedQuery:         `WHERE (((JSON_VALUE(body, '$."name"') = ? AND LOWER(body) LIKE LOWER(?)) OR (JSON_VALUE(body, '$."name"') = ? AND LOWER(body) LIKE LOWER(?))) AND (JSON_EXISTS(body, '$."name"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{"hello", "%hello%", "world", "%world%", "%\"name\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 `body.value IN (200, 300)`,
 			shouldPass:            true,
-			expectedQuery:         `WHERE ((JSONExtract(JSON_VALUE(body, '$."value"'), 'Float64') = ? OR JSONExtract(JSON_VALUE(body, '$."value"'), 'Float64') = ?) AND JSON_EXISTS(body, '$."value"'))`,
-			expectedArgs:          []any{float64(200), float64(300)},
+			expectedQuery:         `WHERE ((JSONExtract(JSON_VALUE(body, '$."value"'), 'Float64') = ? OR JSONExtract(JSON_VALUE(body, '$."value"'), 'Float64') = ?) AND (JSON_EXISTS(body, '$."value"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{float64(200), float64(300), "%\"value\"%"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "json",
 			query:                 "body.key-with-hyphen = true",
 			shouldPass:            true,
-			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."key-with-hyphen"'), 'Bool') = ? AND JSON_EXISTS(body, '$."key-with-hyphen"'))`,
-			expectedArgs:          []any{true},
+			expectedQuery:         `WHERE (JSONExtract(JSON_VALUE(body, '$."key-with-hyphen"'), 'Bool') = ? AND (JSON_EXISTS(body, '$."key-with-hyphen"') AND LOWER(body) LIKE LOWER(?)))`,
+			expectedArgs:          []any{true, "%\"key-with-hyphen\"%"},
 			expectedErrorContains: "",
 		},
 	}

pkg/telemetryschema/logstelemetryschema/filter_expr_logs_test.go

@@ -737,8 +737,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Key-operator-value boundary",
 			query:                 "greater>than",
 			shouldPass:            true,
-			expectedQuery:         `WHERE ((attributes_string['greater'] > ? AND mapContains(attributes_string, 'greater')) OR (JSON_VALUE(body, '$."greater"') > ? AND JSON_EXISTS(body, '$."greater"')))`,
-			expectedArgs:          []any{"than", "than"},
+			expectedQuery:         `WHERE ((attributes_string['greater'] > ? AND mapContains(attributes_string, 'greater')) OR (JSON_VALUE(body, '$."greater"') > ? AND (JSON_EXISTS(body, '$."greater"') AND LOWER(body) LIKE LOWER(?))))`,
+			expectedArgs:          []any{"than", "than", "%\"greater\"%"},
 			expectedErrorContains: "",
 		},
 		{
@@ -753,8 +753,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Key-operator-value boundary",
 			query:                 "less<than",
 			shouldPass:            true,
-			expectedQuery:         `WHERE ((attributes_string['less'] < ? AND mapContains(attributes_string, 'less')) OR (JSON_VALUE(body, '$."less"') < ? AND JSON_EXISTS(body, '$."less"')))`,
-			expectedArgs:          []any{"than", "than"},
+			expectedQuery:         `WHERE ((attributes_string['less'] < ? AND mapContains(attributes_string, 'less')) OR (JSON_VALUE(body, '$."less"') < ? AND (JSON_EXISTS(body, '$."less"') AND LOWER(body) LIKE LOWER(?))))`,
+			expectedArgs:          []any{"than", "than", "%\"less\"%"},
 			expectedErrorContains: "",
 		},
 		{
@@ -809,8 +809,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Key-operator-value boundary",
 			query:                 "user=admin",
 			shouldPass:            true,
-			expectedQuery:         `WHERE ((attributes_string['user'] = ? AND mapContains(attributes_string, 'user')) OR (JSON_VALUE(body, '$."user"') = ? AND JSON_EXISTS(body, '$."user"')))`,
-			expectedArgs:          []any{"admin", "admin"},
+			expectedQuery:         `WHERE ((attributes_string['user'] = ? AND mapContains(attributes_string, 'user')) OR ((JSON_VALUE(body, '$."user"') = ? AND LOWER(body) LIKE LOWER(?)) AND (JSON_EXISTS(body, '$."user"') AND LOWER(body) LIKE LOWER(?))))`,
+			expectedArgs:          []any{"admin", "admin", "%admin%", "%\"user\"%"},
 			expectedErrorContains: "",
 		},
 		{
@@ -1561,15 +1561,15 @@ func TestFilterExprLogs(t *testing.T) {
 			expectedArgs:          []any{"download"},
 			expectedErrorContains: "function `hasToken` expects value parameter to be a string",
 		},
-		// A multi-token needle (separator/whitespace) is a clean 400, not a CH execution error.
+		// A multi-token value (separator/whitespace) is a clean 400, not a CH execution error.
 		{
-			category:              "hasTokenUnderscoreNeedle",
+			category:              "hasTokenUnderscoreSeparator",
 			query:                 "hasToken(body, \"user_id\")",
 			shouldPass:            false,
 			expectedErrorContains: "function `hasToken` matches a single whole token",
 		},
 		{
-			category:              "hasTokenWhitespaceNeedle",
+			category:              "hasTokenWhitespaceSeparator",
 			query:                 "hasToken(body, \"production node\")",
 			shouldPass:            false,
 			expectedErrorContains: "function `hasToken` matches a single whole token",
@@ -2052,8 +2052,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Nested object paths",
 			query:                 "response.body.data.items[].id=123",
 			shouldPass:            true,
-			expectedQuery:         `WHERE ((toFloat64(attributes_number['response.body.data.items[].id']) = ? AND mapContains(attributes_number, 'response.body.data.items[].id')) OR (JSONExtract(JSON_VALUE(body, '$."response"."body"."data"."items"[*]."id"'), 'Float64') = ? AND JSON_EXISTS(body, '$."response"."body"."data"."items"[*]."id"')))`,
-			expectedArgs:          []any{float64(123), float64(123)},
+			expectedQuery:         `WHERE ((toFloat64(attributes_number['response.body.data.items[].id']) = ? AND mapContains(attributes_number, 'response.body.data.items[].id')) OR (JSONExtract(JSON_VALUE(body, '$."response"."body"."data"."items"[*]."id"'), 'Float64') = ? AND (JSON_EXISTS(body, '$."response"."body"."data"."items"[*]."id"') AND LOWER(body) LIKE LOWER(?))))`,
+			expectedArgs:          []any{float64(123), float64(123), "%\"response\"%\"body\"%\"data\"%\"items\"%\"id\"%"},
 			expectedErrorContains: "",
 		},
 		{
@@ -2521,7 +2521,7 @@ func TestFilterExprLogsConflictNegation(t *testing.T) {
 			query:      "body NOT LIKE 'done'",
 			shouldPass: true,
 			// lower index search on body even for LIKE
-			expectedQuery:         "WHERE (LOWER(body) NOT LIKE LOWER(?) AND attributes_string['body'] NOT LIKE ?)",
+			expectedQuery:         "WHERE (body NOT LIKE ? AND attributes_string['body'] NOT LIKE ?)",
 			expectedArgs:          []any{"done", "done"},
 			expectedErrorContains: "",
 		},

pkg/telemetryschema/logstelemetryschema/json_condition_builder.go

@@ -439,27 +439,27 @@ func (c *jsonConditionBuilder) arrayFuncScalarLeaf(node *telemetrytypes.JSONAcce
 // buildTokenFunctionCondition builds a hasToken search over a body JSON string field:
 // hasToken(LOWER(<elem>), LOWER(?)) wrapped in arrayExists over any array hops between the
 // root and the terminal. The field must resolve to a String leaf or a String array.
-func (c *jsonConditionBuilder) buildTokenFunctionCondition(needle any, sb *sqlbuilder.SelectBuilder) (string, error) {
+func (c *jsonConditionBuilder) buildTokenFunctionCondition(token any, sb *sqlbuilder.SelectBuilder) (string, error) {
 	if len(c.key.JSONPlan) == 0 {
 		return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "function `hasToken` could not resolve a JSON access plan for field `%s`", c.key.Name)
 	}
 
 	return c.buildOredRootChains(func(node *telemetrytypes.JSONAccessNode) (string, error) {
-		return c.tokenLeaf(node, needle, sb)
+		return c.tokenLeaf(node, token, sb)
 	}, sb)
 }
 
 // tokenLeaf builds the hasToken match at a terminal node: a direct match for a String leaf
 // (coalesced to false, as in arrayFuncScalarLeaf), or an arrayExists over the elements for a
 // String array leaf. hasToken is string-only, so any other element type is rejected.
-func (c *jsonConditionBuilder) tokenLeaf(node *telemetrytypes.JSONAccessNode, needle any, sb *sqlbuilder.SelectBuilder) (string, error) {
+func (c *jsonConditionBuilder) tokenLeaf(node *telemetrytypes.JSONAccessNode, token any, sb *sqlbuilder.SelectBuilder) (string, error) {
 	switch node.TerminalConfig.ElemType {
 	case telemetrytypes.String:
 		fieldExpr := fmt.Sprintf("dynamicElement(%s, 'String')", node.FieldPath())
-		return fmt.Sprintf("ifNull(hasToken(LOWER(%s), LOWER(%s)), false)", fieldExpr, sb.Var(needle)), nil
+		return fmt.Sprintf("ifNull(hasToken(LOWER(%s), LOWER(%s)), false)", fieldExpr, sb.Var(token)), nil
 	case telemetrytypes.ArrayString:
 		arrayExpr := fmt.Sprintf("dynamicElement(%s, '%s')", node.FieldPath(), node.TerminalConfig.ElemType.StringValue())
-		return fmt.Sprintf("arrayExists(x -> hasToken(LOWER(x), LOWER(%s)), %s)", sb.Var(needle), arrayExpr), nil
+		return fmt.Sprintf("arrayExists(x -> hasToken(LOWER(x), LOWER(%s)), %s)", sb.Var(token), arrayExpr), nil
 	default:
 		return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "function `hasToken` only supports string fields; field `%s` is `%s`", c.key.Name, node.TerminalConfig.Key.FieldDataType.StringValue())
 	}

pkg/telemetryschema/logstelemetryschema/json_string.go

@@ -9,6 +9,8 @@ import (
 
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+
+	"github.com/huandu/go-sqlbuilder"
 )
 
 func parseStrValue(valueStr string, operator qbtypes.FilterOperator) (telemetrytypes.FieldDataType, any) {
@@ -92,6 +94,114 @@ func InferDataType(value any, operator qbtypes.FilterOperator, key *telemetrytyp
 	return closure(value, key)
 }
 
+// likePatternLiterals returns the runs of pattern between unescaped wildcards, with `\` escapes
+// resolved; every value the pattern matches holds each run verbatim. ClickHouse treats `\` as an
+// escape only before `%`, `_` and itself, so dropping it elsewhere would yield a run it never requires.
+func likePatternLiterals(pattern string) []string {
+	var (
+		literals []string
+		run      strings.Builder
+	)
+	for i := 0; i < len(pattern); i++ {
+		switch c := pattern[i]; c {
+		case '%', '_':
+			if run.Len() > 0 {
+				literals = append(literals, run.String())
+				run.Reset()
+			}
+		case '\\':
+			if i+1 >= len(pattern) {
+				run.WriteByte('\\')
+				continue
+			}
+			i++
+			if escaped := pattern[i]; escaped != '%' && escaped != '_' && escaped != '\\' {
+				run.WriteByte('\\')
+			}
+			run.WriteByte(pattern[i])
+		default:
+			run.WriteByte(c)
+		}
+	}
+	if run.Len() > 0 {
+		literals = append(literals, run.String())
+	}
+	return literals
+}
+
+// jsonEscapable reports whether a JSON encoder is free to rewrite r: `"` and `\` always, `/` by
+// PHP, `<` `>` `&` by Go, non-printable ASCII by Python's ensure_ascii. The legacy body holds the
+// producer's own text, so a literal spanning one of these may not be there to find.
+func jsonEscapable(r rune) bool {
+	return r < 0x20 || r > 0x7e || strings.ContainsRune(`"\/<>&`, r)
+}
+
+// jsonTextRuns splits s at every byte an encoder may rewrite. A body whose JSON holds s contains
+// each returned run verbatim, in order.
+func jsonTextRuns(s string) []string {
+	return strings.FieldsFunc(s, jsonEscapable)
+}
+
+// bodyPathLiterals returns one literal per component of key's JSON path, quoted the way JSON writes
+// an object key. A component holding a byte an encoder may rewrite is dropped; JSON writes a parent
+// first, so the order carries.
+func bodyPathLiterals(key *telemetrytypes.TelemetryFieldKey) []string {
+	var literals []string
+	for _, part := range strings.Split(key.Name, ".") {
+		if idx := strings.Index(part, "["); idx >= 0 {
+			part = part[:idx]
+		}
+		if literal := `"` + part + `"`; !strings.ContainsFunc(part, jsonEscapable) {
+			literals = append(literals, literal)
+		}
+	}
+	return literals
+}
+
+// bodyValueLiterals returns the literals a comparison implies in the body text. Only string
+// comparisons qualify: a number is compared after JSONExtract parses it, which reads 1.23e2
+// as 123, so the digits of the filter value need not appear in the body at all.
+func bodyValueLiterals(operator qbtypes.FilterOperator, value any) []string {
+	str, ok := value.(string)
+	if !ok {
+		return nil
+	}
+	switch operator {
+	case qbtypes.FilterOperatorEqual, qbtypes.FilterOperatorContains:
+		return jsonTextRuns(str)
+	case qbtypes.FilterOperatorLike, qbtypes.FilterOperatorILike:
+		var literals []string
+		for _, literal := range likePatternLiterals(str) {
+			literals = append(literals, jsonTextRuns(literal)...)
+		}
+		return literals
+	}
+	return nil
+}
+
+// escapeLikeLiteral escapes the LIKE metacharacters so s matches as literal text. Backslash
+// goes first, being the escape character itself.
+func escapeLikeLiteral(s string) string {
+	s = strings.ReplaceAll(s, `\`, `\\`)
+	s = strings.ReplaceAll(s, "%", `\%`)
+	return strings.ReplaceAll(s, "_", `\_`)
+}
+
+// bodyIndexPredicate asserts the raw body text holds the literals in order. ILike renders as
+// LOWER(body) LIKE LOWER(?) on the ClickHouse flavor — the expression both bloom filters index.
+// They are plain text and backslash-free by construction, so only the LIKE wildcards need escaping.
+func bodyIndexPredicate(literals []string, sb *sqlbuilder.SelectBuilder) string {
+	if len(literals) == 0 {
+		return ""
+	}
+	escaped := make([]string, 0, len(literals))
+	for _, literal := range literals {
+		escaped = append(escaped, escapeLikeLiteral(literal))
+	}
+	pattern := "%" + strings.Join(escaped, "%") + "%"
+	return sb.ILike(LogsV2BodyColumn, pattern)
+}
+
 func getBodyJSONPath(key *telemetrytypes.TelemetryFieldKey) string {
 	parts := strings.Split(key.Name, ".")
 	newParts := []string{}
@@ -139,14 +249,14 @@ func GetBodyJSONKeyForExists(_ context.Context, key *telemetrytypes.TelemetryFie
 	return fmt.Sprintf("JSON_EXISTS(body, '$.%s')", getBodyJSONPath(key))
 }
 
-// legacyElemType infers the has-family element type from the needle (legacy has no schema). It
-// scans EVERY value so the chosen array type and all coerced needles agree — else ClickHouse
+// legacyElemType infers the has-family element type from the arg (legacy has no schema). It
+// scans EVERY value so the chosen array type and all coerced args agree — else ClickHouse
 // raises "no supertype ... String" (code 386). Int64 stays distinct from Float64 so a quoted
 // integer is exact past 2^53 (unquoted literals already arrive as float64, parsed upstream).
-func legacyElemType(needle any) telemetrytypes.FieldDataType {
-	list, ok := needle.([]any)
+func legacyElemType(arg any) telemetrytypes.FieldDataType {
+	list, ok := arg.([]any)
 	if !ok {
-		list = []any{needle}
+		list = []any{arg}
 	}
 	if len(list) == 0 {
 		return telemetrytypes.FieldDataTypeString
@@ -167,7 +277,7 @@ func legacyElemType(needle any) telemetrytypes.FieldDataType {
 			}
 		default:
 			// booleans (and anything else) -> String; a bool renders to 'true'/'false', so a
-			// bool needle only matches genuine JSON booleans, not truthy numbers/strings.
+			// bool arg only matches genuine JSON booleans, not truthy numbers/strings.
 			allInt, allNumeric = false, false
 		}
 	}
@@ -181,9 +291,9 @@ func legacyElemType(needle any) telemetrytypes.FieldDataType {
 	}
 }
 
-// legacyCoerceNeedle coerces a needle to elem type dt so its bound-arg type matches the
+// legacyCoerceElement coerces an element to elem type dt so its bound-arg type matches the
 // extracted column (legacyElemType guarantees it's coercible).
-func legacyCoerceNeedle(v any, dt telemetrytypes.FieldDataType) any {
+func legacyCoerceElement(v any, dt telemetrytypes.FieldDataType) any {
 	switch dt {
 	case telemetrytypes.FieldDataTypeInt64:
 		if s, ok := v.(string); ok {
@@ -199,7 +309,7 @@ func legacyCoerceNeedle(v any, dt telemetrytypes.FieldDataType) any {
 		}
 		return v
 	default:
-		return bodyArrayNeedleString(v)
+		return bodyArrayElementString(v)
 	}
 }
 
@@ -216,8 +326,8 @@ func getBodyJSONArrayKey(key *telemetrytypes.TelemetryFieldKey, dt telemetrytype
 
 // getBodyJSONScalarKey builds the single-element-set fallback for a scalar body value: the leaf
 // extracted as a scalar of type dt, plus a guard restricting it to a genuinely scalar body. The
-// guard is required because JSON_VALUE returns '' for an array/object/missing value, which would
-// otherwise zero-value match (has(x,0) / has(x,false) / has(x,'') on any array). ok=false when
+// guard is required because JSON_VALUE returns ” for an array/object/missing value, which would
+// otherwise zero-value match (has(x,0) / has(x,false) / has(x,”) on any array). ok=false when
 // the path still traverses an array ([*]/[]).
 func getBodyJSONScalarKey(key *telemetrytypes.TelemetryFieldKey, dt telemetrytypes.FieldDataType) (expr string, guard string, ok bool) {
 	name := strings.TrimSuffix(strings.TrimSuffix(key.Name, "[*]"), "[]")
@@ -242,7 +352,7 @@ func getBodyJSONScalarKey(key *telemetrytypes.TelemetryFieldKey, dt telemetrytyp
 	return expr, guard, true
 }
 
-func bodyArrayNeedleString(v any) string {
+func bodyArrayElementString(v any) string {
 	switch t := v.(type) {
 	case string:
 		return t

tests/fixtures/querier.py

@@ -253,6 +253,27 @@ def get_preview_sql(response: requests.Response, name: str) -> str:
     return statements[0]["db.statement.query"]
 
 
+def get_preview_skip_indexes(response: requests.Response, name: str) -> dict[str, dict[str, Any]]:
+    """The skip-index steps of the named query's read funnel, keyed by index name.
+
+    Needs a verbose preview. ClickHouse lists a skip index only when the predicate matches its
+    expression, so an absent entry means it was never consulted."""
+    statements = get_preview_statements(response, name)
+    assert len(statements) == 1, f"expected 1 statement for query {name}, got {len(statements)}"
+    granules = statements[0]["granules"]
+    assert granules is not None, f"query {name} reads no MergeTree table: {statements[0]}"
+    return {step["name"]: step for read in granules["reads"] for step in read["steps"] if step["type"] == "Skip"}
+
+
+def get_preview_selected_granules(response: requests.Response, name: str) -> int:
+    """Granules surviving every index step of the named query's read funnel."""
+    statements = get_preview_statements(response, name)
+    assert len(statements) == 1, f"expected 1 statement for query {name}, got {len(statements)}"
+    granules = statements[0]["granules"]
+    assert granules is not None, f"query {name} reads no MergeTree table: {statements[0]}"
+    return granules["selected"]
+
+
 def aligned_epoch(ago: timedelta, step_seconds: int = DEFAULT_STEP_INTERVAL) -> int:
     """Epoch seconds for `now - ago`, floored to a step boundary so seeded
     points land exactly on the query's toStartOfInterval buckets."""

tests/integration/tests/querier_json_body/06_body_equality.py

@@ -0,0 +1,90 @@
+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
+
+LOWER = "alpha"
+UPPER = "ALPHA"
+PLAIN = "beta"
+NON_ASCII = "Mixed CASE Ünïcode"
+SLASH = "GET /api/v1/users"
+SUPERSTRING = "GET /api/v1/users/42"
+QUOTE = 'say "hi" now'
+BACKSLASH = "C:\\tmp\\log"
+LIKE_META = "100% _off"
+TAB = "tab\there"
+CTRL = "ctrl\x01here"
+
+BODIES = [LOWER, UPPER, PLAIN, NON_ASCII, SLASH, SUPERSTRING, QUOTE, BACKSLASH, LIKE_META, TAB, CTRL]
+
+
+# querierlogs/16_body_equality.py with use_json_body on: `body` resolves to body_v2.message,
+# which the lower(body) companion skips, and the same expressions must still answer alike.
+@pytest.mark.parametrize(
+    "expression,expected_bodies",
+    [
+        pytest.param(f"body = '{LOWER}'", {LOWER}, id="equality_exact"),
+        pytest.param(f"body = '{UPPER}'", {UPPER}, id="equality_other_case"),
+        pytest.param("body = 'Alpha'", set(), id="equality_case_must_match"),
+        pytest.param(f"body = '{NON_ASCII}'", {NON_ASCII}, id="equality_non_ascii"),
+        pytest.param("body = 'gamma'", set(), id="equality_no_match"),
+        pytest.param(f"body = '{SLASH}'", {SLASH}, id="equality_slash"),
+        pytest.param("body = 'say \"hi\" now'", {QUOTE}, id="equality_quote"),
+        pytest.param(r"body = 'C:\\tmp\\log'", {BACKSLASH}, id="equality_backslash"),
+        pytest.param(f"body = '{LIKE_META}'", {LIKE_META}, id="equality_like_metacharacters"),
+        pytest.param("body = 'tab\there'", {TAB}, id="equality_tab"),
+        pytest.param("body = 'ctrl\x01here'", {CTRL}, id="equality_control_char"),
+        pytest.param("body = 'GET /api/v1'", set(), id="equality_prefix_does_not_match"),
+        pytest.param(f"body IN ('{LOWER}', '{PLAIN}')", {LOWER, PLAIN}, id="in_excludes_other_case"),
+        pytest.param(f"body IN ('{SLASH}', '{LIKE_META}')", {SLASH, LIKE_META}, id="in_escaped_values"),
+        pytest.param(f"body NOT IN ('{LOWER}', '{UPPER}')", set(BODIES) - {LOWER, UPPER}, id="not_in"),
+    ],
+)
+def test_logs_body_equality_json(
+    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_bodies: set[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": "api"},
+                body=body,
+            )
+            for i, body in enumerate(BODIES)
+        ]
+    )
+
+    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_bodies

tests/integration/tests/querierlogs/16_body_equality.py

@@ -0,0 +1,92 @@
+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, make_query_request
+
+LOWER = "alpha"
+UPPER = "ALPHA"
+PLAIN = "beta"
+NON_ASCII = "Mixed CASE Ünïcode"
+SLASH = "GET /api/v1/users"
+SUPERSTRING = "GET /api/v1/users/42"
+QUOTE = 'say "hi" now'
+BACKSLASH = "C:\\tmp\\log"
+LIKE_META = "100% _off"
+TAB = "tab\there"
+CTRL = "ctrl\x01here"
+
+BODIES = [LOWER, UPPER, PLAIN, NON_ASCII, SLASH, SUPERSTRING, QUOTE, BACKSLASH, LIKE_META, TAB, CTRL]
+
+
+# `body = ?` carries a case-insensitive LOWER(body) companion for the bloom filters, so a
+# body differing only in case must still not come back.
+@pytest.mark.parametrize(
+    "expression,expected_bodies",
+    [
+        pytest.param(f"body = '{LOWER}'", {LOWER}, id="equality_exact"),
+        pytest.param(f"body = '{UPPER}'", {UPPER}, id="equality_other_case"),
+        pytest.param("body = 'Alpha'", set(), id="equality_case_must_match"),
+        pytest.param(f"body = '{NON_ASCII}'", {NON_ASCII}, id="equality_non_ascii"),
+        pytest.param("body = ''", set(), id="equality_empty"),
+        pytest.param("body = 'gamma'", set(), id="equality_no_match"),
+        # the companion is a LIKE-free equality, so none of these are metacharacters to it
+        pytest.param(f"body = '{SLASH}'", {SLASH}, id="equality_slash"),
+        pytest.param("body = 'say \"hi\" now'", {QUOTE}, id="equality_quote"),
+        pytest.param(r"body = 'C:\\tmp\\log'", {BACKSLASH}, id="equality_backslash"),
+        pytest.param(f"body = '{LIKE_META}'", {LIKE_META}, id="equality_like_metacharacters"),
+        pytest.param("body = 'tab\there'", {TAB}, id="equality_tab"),
+        pytest.param("body = 'ctrl\x01here'", {CTRL}, id="equality_control_char"),
+        # a prefix of another body must not match it
+        pytest.param("body = 'GET /api/v1'", set(), id="equality_prefix_does_not_match"),
+        pytest.param(f"body IN ('{LOWER}', '{PLAIN}')", {LOWER, PLAIN}, id="in_excludes_other_case"),
+        pytest.param(f"body IN ('{SLASH}', '{LIKE_META}')", {SLASH, LIKE_META}, id="in_escaped_values"),
+        pytest.param(f"body NOT IN ('{LOWER}', '{UPPER}')", set(BODIES) - {LOWER, UPPER}, id="not_in"),
+    ],
+)
+def test_logs_body_equality(
+    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_bodies: set[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": "api"},
+                body=body,
+            )
+            for i, body in enumerate(BODIES)
+        ]
+    )
+
+    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_bodies

tests/integration/tests/querierlogs/17_index_pruning.py

@@ -0,0 +1,226 @@
+import json
+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_preview_selected_granules,
+    get_preview_skip_indexes,
+    get_rows,
+    make_preview_query_request,
+    make_query_request,
+)
+
+# The legacy body holds the text the producer wrote, so the same value reaches ClickHouse under
+# different encodings: PHP escapes `/`, Go escapes `&` `<` `>`, Python escapes non-ASCII. The
+# LOWER(body) predicates the filters carry for the bloom filters must find all of them.
+BODIES = {
+    "plain": '{"tag":"plain","url":"https://signoz.io/docs","user_id":4242,"status":"timeout_error"}',
+    "php": '{"tag":"php","url":"https:\\/\\/signoz.io\\/docs"}',
+    "go": '{"tag":"go","note":"connection reset \\u0026 retry aborted"}',
+    "python": '{"tag":"python","city":"caf\\u00e9 municipal district"}',
+    "other_case": '{"tag":"other_case","status":"TIMEOUT_ERROR"}',
+    "no_user_id": '{"tag":"no_user_id","status":"ok","url":"https://signoz.io/pricing"}',
+    "tagged": '{"tag":"tagged","labels":["production","webserver"]}',
+    "tagged_escaped": '{"tag":"tagged_escaped","labels":["batch \\u0026 stream","webserver"]}',
+}
+
+
+@pytest.mark.parametrize(
+    "expression,expected_tags",
+    [
+        pytest.param("body.user_id = 4242", {"plain"}, id="numeric_equality"),
+        pytest.param("body.user_id EXISTS", {"plain"}, id="exists"),
+        # a negated comparison matches the rows without the path, so it carries no predicate
+        pytest.param("body.user_id != 4242", set(BODIES) - {"plain"}, id="not_equal_keeps_pathless_rows"),
+        pytest.param("body.status NOT EXISTS", {"php", "go", "python", "tagged", "tagged_escaped"}, id="not_exists"),
+        pytest.param("body.url = 'https://signoz.io/docs'", {"plain", "php"}, id="equality_escaped_slashes"),
+        pytest.param("body.url CONTAINS 'signoz.io/docs'", {"plain", "php"}, id="contains_escaped_slashes"),
+        pytest.param("body.note = 'connection reset & retry aborted'", {"go"}, id="equality_escaped_ampersand"),
+        pytest.param("body.city = 'café municipal district'", {"python"}, id="equality_escaped_non_ascii"),
+        # the value predicate is case-insensitive where the equality is not
+        pytest.param("body.status = 'timeout_error'", {"plain"}, id="equality_underscore"),
+        pytest.param("body.status = 'TIMEOUT_ERROR'", {"other_case"}, id="equality_other_case"),
+        pytest.param("body.status IN ('timeout_error', 'ok')", {"plain", "no_user_id"}, id="in_carries_one_value_per_arm"),
+        # has and hasAll assert every element, hasAny only one of them
+        pytest.param("has(body.labels[*], 'production')", {"tagged"}, id="has_element"),
+        pytest.param("has(body.labels[*], 'batch & stream')", {"tagged_escaped"}, id="has_escaped_element"),
+        pytest.param("hasAll(body.labels[*], ['production', 'webserver'])", {"tagged"}, id="has_all_needs_every_element"),
+        pytest.param(
+            "hasAny(body.labels[*], ['production', 'batch & stream'])",
+            {"tagged", "tagged_escaped"},
+            id="has_any_needs_one_element",
+        ),
+        # 'webserver' is in both, so an ORed literal set must not exclude either row
+        pytest.param(
+            "hasAny(body.labels[*], ['webserver', 'nothing here'])",
+            {"tagged", "tagged_escaped"},
+            id="has_any_across_both",
+        ),
+    ],
+)
+def test_logs_body_json_index_predicates(
+    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_tags: set[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": "api"},
+                body=body,
+            )
+            for i, body in enumerate(BODIES.values())
+        ]
+    )
+
+    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 {json.loads(row["data"]["body"])["tag"] for row in get_rows(response)} == expected_tags
+
+
+# JSON_VALUE matches no index expression, so the literals are what get the bloom filters consulted
+# at all; the read funnel is what catches one that stops matching.
+BODY_BLOOM_FILTERS = {"body_index_v2_token", "body_index_v2_ngram"}
+
+
+@pytest.mark.parametrize(
+    "expression,prunes_every_granule",
+    [
+        pytest.param("body.status = 'timeout_error'", False, id="value_needle_present"),
+        pytest.param("body.status = 'zz_no_seeded_body_holds_this'", True, id="value_needle_absent"),
+        pytest.param("body.zz_no_seeded_body_holds_this EXISTS", True, id="path_needle_absent"),
+        # a number is compared after JSONExtract parses it, so it carries no value literal - only
+        # its path, which every row holding the key satisfies
+        pytest.param("body.user_id = 999999", False, id="number_carries_only_its_path"),
+    ],
+)
+def test_logs_body_json_index_prunes_granules(
+    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,
+    prunes_every_granule: bool,
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": "api"},
+                body=body,
+            )
+            for i, body in enumerate(BODIES.values())
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_preview_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
+    skip_indexes = get_preview_skip_indexes(response, "A")
+    assert BODY_BLOOM_FILTERS <= set(skip_indexes), f"body bloom filters not consulted, only: {sorted(skip_indexes)}"
+
+    selected = get_preview_selected_granules(response, "A")
+    if prunes_every_granule:
+        assert selected == 0, f"expected every granule pruned, {selected} survived"
+    else:
+        assert selected > 0, "the granule holding the match must survive"
+
+
+# `body = ?` matches no index expression on its own; the lowered companion is what the filters
+# prune on, so its absence from the funnel is the regression this catches.
+@pytest.mark.parametrize(
+    "expression,prunes_every_granule",
+    [
+        pytest.param("body = 'alpha'", False, id="equality_present_value"),
+        pytest.param("body = 'zz_no_seeded_body_holds_this'", True, id="equality_absent_value"),
+        # IN delegates to the equalities, so every arm carries its own companion
+        pytest.param("body IN ('alpha', 'beta')", False, id="in_present_values"),
+        pytest.param("body IN ('zz_absent_one', 'zz_absent_two')", True, id="in_absent_values"),
+    ],
+)
+def test_logs_body_equality_prunes_granules(
+    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,
+    prunes_every_granule: bool,
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs([Logs(timestamp=now - timedelta(seconds=i + 1), resources={"service.name": "api"}, body=body) for i, body in enumerate(["alpha", "ALPHA", "beta"])])
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_preview_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
+    skip_indexes = get_preview_skip_indexes(response, "A")
+    assert BODY_BLOOM_FILTERS <= set(skip_indexes), f"body bloom filters not consulted, only: {sorted(skip_indexes)}"
+
+    selected = get_preview_selected_granules(response, "A")
+    if prunes_every_granule:
+        assert selected == 0, f"expected every granule pruned, {selected} survived"
+    else:
+        assert selected > 0, "the granule holding the match must survive"