perf(logs): let the legacy has family use the lower(body) bloom filters

has, hasAll and hasAny over a legacy body path compare extracted JSON, which matches no index
expression, so they read every granule. They now carry what they imply over the indexed
LOWER(body), ANDed onto the extraction that still decides the row:

- the path, since the extraction yields nothing for an absent one;
- each element, since it must appear in the text. has and hasAll require every element, 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, which drops it whole.

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 - the same reason `=` skips them.

The comparison and its literals are built together, so the element type and the element list are
resolved once instead of in two places.

pkg/querybuilder/clickhouse_quote.go

@@ -9,6 +9,14 @@ func ClickHouseStringLiteral(value string) string {
 	return "'" + escaped + "'"
 }
 
+// ClickHouseLikePatternLiteral escapes the LIKE metacharacters so value matches as literal
+// text inside a pattern. Backslash goes first, being the escape character itself.
+func ClickHouseLikePatternLiteral(value string) string {
+	escaped := strings.ReplaceAll(value, `\`, `\\`)
+	escaped = strings.ReplaceAll(escaped, `%`, `\%`)
+	return strings.ReplaceAll(escaped, `_`, `\_`)
+}
+
 // ClickHouseIdentifier quotes a value for a ClickHouse identifier.
 func ClickHouseIdentifier(value string) string {
 	escaped := strings.ReplaceAll(value, `\`, `\\`)

pkg/querybuilder/json_text.go

@@ -0,0 +1,41 @@
+package querybuilder
+
+import (
+	"strings"
+
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+)
+
+// 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. JSON-encoded text holds
+// the producer's own bytes, 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. JSON text that holds s contains
+// each returned run verbatim, in order.
+func JSONTextRuns(s string) []string {
+	return strings.FieldsFunc(s, JSONEscapable)
+}
+
+// JSONComparisonLiterals returns the literals a comparison implies in the JSON text the compared
+// value was decoded from. Only string comparisons qualify: a number is compared after parsing,
+// which reads 1.23e2 as 123, so the digits of the filter value need not appear in the text at all.
+func JSONComparisonLiterals(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:
+		// the pattern's own wildcards split runs like the escapable bytes do; `\` is one of
+		// those, so pattern escapes dissolve with it (an escaped wildcard is merely not required)
+		return strings.FieldsFunc(str, func(r rune) bool {
+			return JSONEscapable(r) || r == '%' || r == '_'
+		})
+	}
+	return nil
+}

pkg/querybuilder/json_text_test.go

@@ -0,0 +1,60 @@
+package querybuilder
+
+import (
+	"testing"
+
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/stretchr/testify/assert"
+)
+
+// The literals have to hold whichever encoder wrote the text, 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))
+		})
+	}
+}
+
+// A LIKE pattern splits at its wildcards on top of the encoder-escapable bytes, so its escape
+// sequences dissolve conservatively; an equality value keeps wildcard characters as literal text.
+func TestJSONComparisonLiterals(t *testing.T) {
+	testCases := []struct {
+		name     string
+		operator qbtypes.FilterOperator
+		value    any
+		expected []string
+	}{
+		{"equality keeps wildcards literal", qbtypes.FilterOperatorEqual, "100%_off", []string{"100%_off"}},
+		{"wildcards split runs", qbtypes.FilterOperatorLike, "%foo%bar_baz%", []string{"foo", "bar", "baz"}},
+		{"pattern escapes dissolve", qbtypes.FilterOperatorLike, `%C:\\tmp\_%`, []string{"C:", "tmp"}},
+		{"a number carries nothing", qbtypes.FilterOperatorEqual, int64(123), nil},
+		{"no literals at all", qbtypes.FilterOperatorLike, "%_%", nil},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			literals := JSONComparisonLiterals(tc.operator, tc.value)
+			if len(tc.expected) == 0 {
+				assert.Empty(t, literals)
+				return
+			}
+			assert.Equal(t, tc.expected, literals)
+		})
+	}
+}

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

@@ -126,20 +126,41 @@ func (c *conditionBuilder) conditionForArrayFunction(
 		return NewJSONConditionBuilder(key, valueType).buildArrayFunctionCondition(operator, needle, 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).
+	return c.legacyArrayFunctionCondition(key, operator, needle, 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,
+	needle 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(needle)
 	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)
-		}
+
+	elements, isList := needle.([]any)
+	if !isList {
+		elements = []any{needle}
+	}
+	vals := make([]any, len(elements))
+	for i, v := range elements {
+		vals[i] = legacyCoerceNeedle(v, elemType)
+	}
+
+	var cond string
+	switch {
+	case isList:
 		// 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)))
 		if !hasScalar {
-			return arrayCond, nil
+			cond = arrayCond
+			break
 		}
 		var membership string
 		if operator == qbtypes.FilterOperatorHasAll {
@@ -151,14 +172,70 @@ 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
+}
+
+// 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 := querybuilder.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(querybuilder.JSONTextRuns(str), sb); predicate != "" {
+			predicates = append(predicates, predicate)
+		}
+	}
+	return predicates
 }
 
 // castNeedleArray pins an Int64 needle array to Array(Int64) so it matches the Array(Nullable(Int64))
@@ -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)
@@ -304,20 +384,63 @@ func (c *conditionBuilder) conditionForResolvedKey(
 		return "", err
 	}
 
-	// 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)) {
+	// Check if this is a body JSON search (legacy string-body path, JSON flag off). Such a filter
+	// compares JSON_VALUE output, which matches no index expression, so it also carries what it
+	// implies over the indexed LOWER(body): the quoted path on the existence assertion, the value
+	// literals on the comparison. Negations carry nothing — they match rows without the path,
+	// which say nothing about the body text.
+	if legacyBodyJSONSearch {
+		if operator == qbtypes.FilterOperatorExists || operator == qbtypes.FilterOperatorNotExists {
+			exists := GetBodyJSONKeyForExists(ctx, key, operator, value)
+			if operator == qbtypes.FilterOperatorNotExists {
+				return "NOT " + exists, nil
+			}
+			return withBodyIndexPredicate(exists, bodyPathLiterals(key), sb), 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 || !legacyBodyJSONSearch {
+		return cond, err
+	}
+	return withBodyIndexPredicate(cond, querybuilder.JSONComparisonLiterals(operator, value), sb), 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 +479,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

@@ -168,9 +168,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 +207,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 +219,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 +619,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 +906,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 +918,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 +954,173 @@ func TestConditionForBodyIn(t *testing.T) {
 		})
 	}
 }
+
+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()
+
+	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)
+
+			cond, err := cb.conditionForResolvedKey(ctx, valuer.UUID{}, 0, 0, key, tc.operator, tc.value, 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,64 +44,64 @@ 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: "",
 		},
 		{
@@ -109,103 +109,104 @@ func TestFilterExprLogsBodyJSON(t *testing.T) {
 			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: "",
 		},
 		{
@@ -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_string.go

@@ -7,8 +7,11 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/SigNoz/signoz/pkg/querybuilder"
 	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 +95,42 @@ func InferDataType(value any, operator qbtypes.FilterOperator, key *telemetrytyp
 	return closure(value, key)
 }
 
+// bodyPathLiterals returns one literal per component of key's JSON path, taking the quoting from
+// getBodyJSONPath so it matches however the path is written. 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(getBodyJSONPath(key), ".") {
+		literal := strings.TrimSuffix(part, "[*]")
+		if !strings.ContainsFunc(strings.Trim(literal, `"`), querybuilder.JSONEscapable) {
+			literals = append(literals, literal)
+		}
+	}
+	return literals
+}
+
+// bodyIndexPredicate asserts the raw body text holds the literals in order ("" for none). ILike
+// renders as LOWER(body) LIKE LOWER(?) on the ClickHouse flavor — the expression both bloom
+// filters index.
+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, querybuilder.ClickHouseLikePatternLiteral(literal))
+	}
+	return sb.ILike(LogsV2BodyColumn, "%"+strings.Join(escaped, "%")+"%")
+}
+
+// withBodyIndexPredicate ANDs the literals' predicate onto cond; with none, cond is returned as is.
+func withBodyIndexPredicate(cond string, literals []string, sb *sqlbuilder.SelectBuilder) string {
+	if predicate := bodyIndexPredicate(literals, sb); predicate != "" {
+		return sb.And(cond, predicate)
+	}
+	return cond
+}
+
 func getBodyJSONPath(key *telemetrytypes.TelemetryFieldKey) string {
 	parts := strings.Split(key.Name, ".")
 	newParts := []string{}
@@ -216,8 +255,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, "[*]"), "[]")

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"