perf(logs): let body equality use the lower(body) bloom filters

logs_v2 indexes lower(body) with a token and an ngram bloom filter, so a plain
`body = ?` matches no index expression and reads every granule. AND in the
lowered comparison as a redundant predicate: the bloom filters can prune on it
and the exact comparison still decides the row. On 1M rows that is 123/123
granules down to 1/123.

IN picks this up for free through the `=` delegation, so a list of bodies
prunes per arm.

ClickHouse folds LOWER(?) on the bound value, so the predicate stays a constant
the index can prune on, and the fold matches lower() on the column by
construction rather than by a reimplementation on our side.

Scoped to the legacy body column: body_v2 keeps the value in a JSON column
indexed on lower(toString(body_v2)), which needs a different predicate.

The integration cases run the same expressions with the flag off and on. The
companion is case-insensitive where the equality is not, so they pin that down:
a body differing only in case must not come back, and the flag-on path — which
skips the companion — has to answer identically.

pkg/statementbuilder/logsstatementbuilder/stmt_builder_test.go

@@ -469,6 +469,8 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
 			expectedErr: nil,
 		},
 		{
+			// The `[*]` path is extracted per value, not as an Array(String) compared to a
+			// scalar — ClickHouse rejects that outright (code 130).
 			name:        "IN operator with json search",
 			requestType: qbtypes.RequestTypeRaw,
 			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
@@ -479,7 +481,7 @@ 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 ((JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(String)') = ?) AND JSON_EXISTS(body, '$.\"user_names\"[*]')) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				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},
 				Warnings: []string{querybuilder.NewKeyNotFoundWarning("user_names[*]")},
 			},
@@ -1009,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/statementbuilder/resourcefilter/condition_builder.go

@@ -180,20 +180,16 @@ func (b *defaultConditionBuilder) conditionForKey(
 		if !ok {
 			return "", qbtypes.ErrInValues
 		}
+		// each value carries its own index filter, since `=` derives one from the value
 		inConditions := make([]string, 0, len(values))
 		for _, v := range values {
-			inConditions = append(inConditions, sb.E(fieldName, querybuilder.FormatValueForContains(v)))
-		}
-		mainCondition := sb.Or(inConditions...)
-		valConditions := make([]string, 0, len(values))
-		if valuesForIndexFilter, ok := valueForIndexFilter.([]string); ok {
-			for _, v := range valuesForIndexFilter {
-				valConditions = append(valConditions, sb.Like(column.Name, v))
+			cond, err := b.conditionForKey(ctx, startNs, endNs, key, qbtypes.FilterOperatorEqual, v, sb)
+			if err != nil {
+				return "", err
 			}
+			inConditions = append(inConditions, cond)
 		}
-		mainCondition = sb.And(mainCondition, keyIdxFilter, sb.Or(valConditions...))
-
-		return mainCondition, nil
+		return sb.Or(inConditions...), nil
 	case qbtypes.FilterOperatorNotIn:
 		values, ok := value.([]any)
 		if !ok {
@@ -201,17 +197,13 @@ func (b *defaultConditionBuilder) conditionForKey(
 		}
 		notInConditions := make([]string, 0, len(values))
 		for _, v := range values {
-			notInConditions = append(notInConditions, sb.NE(fieldName, querybuilder.FormatValueForContains(v)))
-		}
-		mainCondition := sb.And(notInConditions...)
-		valConditions := make([]string, 0, len(values))
-		if valuesForIndexFilter, ok := valueForIndexFilter.([]string); ok {
-			for _, v := range valuesForIndexFilter {
-				valConditions = append(valConditions, sb.NotLike(column.Name, v))
+			cond, err := b.conditionForKey(ctx, startNs, endNs, key, qbtypes.FilterOperatorNotEqual, v, sb)
+			if err != nil {
+				return "", err
 			}
+			notInConditions = append(notInConditions, cond)
 		}
-		mainCondition = sb.And(mainCondition, sb.And(valConditions...))
-		return mainCondition, nil
+		return sb.And(notInConditions...), nil
 
 	case qbtypes.FilterOperatorExists:
 		return sb.And(

pkg/statementbuilder/resourcefilter/condition_builder_test.go

@@ -109,8 +109,8 @@ func TestConditionBuilder(t *testing.T) {
 			},
 			op:           qbtypes.FilterOperatorIn,
 			value:        []any{"watch", "redis"},
-			expected:     "(simpleJSONExtractString(labels, 'k8s.namespace.name') = ? OR simpleJSONExtractString(labels, 'k8s.namespace.name') = ?) AND labels LIKE ? AND (labels LIKE ? OR labels LIKE ?)",
-			expectedArgs: []any{"watch", "redis", "%k8s.namespace.name%", "%k8s.namespace.name\":\"watch%", "%k8s.namespace.name\":\"redis%"},
+			expected:     "((simpleJSONExtractString(labels, 'k8s.namespace.name') = ? AND labels LIKE ? AND labels LIKE ?) OR (simpleJSONExtractString(labels, 'k8s.namespace.name') = ? AND labels LIKE ? AND labels LIKE ?))",
+			expectedArgs: []any{"watch", "%k8s.namespace.name%", "%k8s.namespace.name\":\"watch%", "redis", "%k8s.namespace.name%", "%k8s.namespace.name\":\"redis%"},
 		},
 		{
 			name: "string_not_in",
@@ -120,8 +120,8 @@ func TestConditionBuilder(t *testing.T) {
 			},
 			op:           qbtypes.FilterOperatorNotIn,
 			value:        []any{"watch", "redis"},
-			expected:     "(simpleJSONExtractString(labels, 'k8s.namespace.name') <> ? AND simpleJSONExtractString(labels, 'k8s.namespace.name') <> ?) AND (labels NOT LIKE ? AND labels NOT LIKE ?)",
-			expectedArgs: []any{"watch", "redis", "%k8s.namespace.name\":\"watch%", "%k8s.namespace.name\":\"redis%"},
+			expected:     "((simpleJSONExtractString(labels, 'k8s.namespace.name') <> ? AND labels NOT LIKE ?) AND (simpleJSONExtractString(labels, 'k8s.namespace.name') <> ? AND labels NOT LIKE ?))",
+			expectedArgs: []any{"watch", "%k8s.namespace.name\":\"watch%", "redis", "%k8s.namespace.name\":\"redis%"},
 		},
 		{
 			name: "string_exists",
@@ -173,8 +173,8 @@ func TestConditionBuilder(t *testing.T) {
 			},
 			op:           qbtypes.FilterOperatorIn,
 			value:        []any{1, 2},
-			expected:     "(simpleJSONExtractString(labels, 'test_num') = ? OR simpleJSONExtractString(labels, 'test_num') = ?) AND labels LIKE ? AND (labels LIKE ? OR labels LIKE ?)",
-			expectedArgs: []any{"1", "2", "%test_num%", "%test_num\":\"1%", "%test_num\":\"2%"},
+			expected:     "((simpleJSONExtractString(labels, 'test_num') = ? AND labels LIKE ? AND labels LIKE ?) OR (simpleJSONExtractString(labels, 'test_num') = ? AND labels LIKE ? AND labels LIKE ?))",
+			expectedArgs: []any{"1", "%test_num%", "%test_num\":\"1%", "2", "%test_num%", "%test_num\":\"2%"},
 		},
 		{
 			name: "number_between",

pkg/statementbuilder/resourcefilter/statement_builder_test.go

@@ -229,8 +229,8 @@ func TestResourceFilterStatementBuilder_Traces(t *testing.T) {
 			start: testStartNs,
 			end:   testEndNs,
 			expected: &qbtypes.Statement{
-				Query: "SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE ((simpleJSONExtractString(labels, 'service.name') = ? OR simpleJSONExtractString(labels, 'service.name') = ?) AND labels LIKE ? AND (labels LIKE ? OR labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint",
-				Args:  []any{"redis", "postgres", "%service.name%", "%service.name\":\"redis%", "%service.name\":\"postgres%", expectedBucketStart, expectedBucketEnd},
+				Query: "SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE ((simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) OR (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint",
+				Args:  []any{"redis", "%service.name%", "%service.name\":\"redis%", "postgres", "%service.name%", "%service.name\":\"postgres%", expectedBucketStart, expectedBucketEnd},
 			},
 		},
 		{
@@ -244,8 +244,8 @@ func TestResourceFilterStatementBuilder_Traces(t *testing.T) {
 			start: testStartNs,
 			end:   testEndNs,
 			expected: &qbtypes.Statement{
-				Query: "SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE ((simpleJSONExtractString(labels, 'service.name') <> ? AND simpleJSONExtractString(labels, 'service.name') <> ?) AND (labels NOT LIKE ? AND labels NOT LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint",
-				Args:  []any{"redis", "postgres", "%service.name\":\"redis%", "%service.name\":\"postgres%", expectedBucketStart, expectedBucketEnd},
+				Query: "SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE ((simpleJSONExtractString(labels, 'service.name') <> ? AND labels NOT LIKE ?) AND (simpleJSONExtractString(labels, 'service.name') <> ? AND labels NOT LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint",
+				Args:  []any{"redis", "%service.name\":\"redis%", "postgres", "%service.name\":\"postgres%", expectedBucketStart, expectedBucketEnd},
 			},
 		},
 		{

pkg/telemetryschema/audittelemetryschema/condition_builder.go

@@ -94,7 +94,11 @@ func (c *conditionBuilder) conditionFor(
 		}
 		conditions := []string{}
 		for _, value := range values {
-			conditions = append(conditions, sb.E(fieldExpression, value))
+			cond, err := c.conditionFor(ctx, orgID, startNs, endNs, key, qbtypes.FilterOperatorEqual, value, sb)
+			if err != nil {
+				return "", err
+			}
+			conditions = append(conditions, cond)
 		}
 		return sb.Or(conditions...), nil
 	case qbtypes.FilterOperatorNotIn:
@@ -104,7 +108,11 @@ func (c *conditionBuilder) conditionFor(
 		}
 		conditions := []string{}
 		for _, value := range values {
-			conditions = append(conditions, sb.NE(fieldExpression, value))
+			cond, err := c.conditionFor(ctx, orgID, startNs, endNs, key, qbtypes.FilterOperatorNotEqual, value, sb)
+			if err != nil {
+				return "", err
+			}
+			conditions = append(conditions, cond)
 		}
 		return sb.And(conditions...), nil
 	case qbtypes.FilterOperatorExists, qbtypes.FilterOperatorNotExists:

pkg/telemetryschema/logstelemetryschema/condition_builder.go

@@ -314,6 +314,14 @@ func (c *conditionBuilder) conditionForResolvedKey(
 	// 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:
@@ -410,7 +418,11 @@ func (c *conditionBuilder) conditionForResolvedKey(
 		// instead of using IN, we use `=` + `OR` to make use of index
 		conditions := []string{}
 		for _, value := range values {
-			conditions = append(conditions, sb.E(fieldExpression, value))
+			cond, err := c.conditionForResolvedKey(ctx, orgID, startNs, endNs, key, qbtypes.FilterOperatorEqual, value, sb)
+			if err != nil {
+				return "", err
+			}
+			conditions = append(conditions, cond)
 		}
 		return sb.Or(conditions...), nil
 	case qbtypes.FilterOperatorNotIn:
@@ -421,7 +433,11 @@ func (c *conditionBuilder) conditionForResolvedKey(
 		// instead of using NOT IN, we use `!=` + `AND` to make use of index
 		conditions := []string{}
 		for _, value := range values {
-			conditions = append(conditions, sb.NE(fieldExpression, value))
+			cond, err := c.conditionForResolvedKey(ctx, orgID, startNs, endNs, key, qbtypes.FilterOperatorNotEqual, value, sb)
+			if err != nil {
+				return "", err
+			}
+			conditions = append(conditions, cond)
 		}
 		return sb.And(conditions...), nil
 

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,
 		},
 		{
@@ -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,
 		},
 	}
@@ -905,3 +905,52 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		})
 	}
 }
+
+// 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
+		values       []any
+		expectedSQL  string
+		expectedArgs []any
+	}{
+		{
+			name:         "strings",
+			values:       []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 = ? AND LOWER(body) = LOWER(?)) OR (body = ? AND LOWER(body) = LOWER(?)) OR (body = ? AND LOWER(body) = LOWER(?)))",
+			expectedArgs: []any{"alpha", "alpha", "1", "1", "true", "true"},
+		},
+	}
+
+	fl := flaggertest.New(t)
+	fm := NewFieldMapper(fl)
+	conditionBuilder := NewConditionBuilder(fm, fl)
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			key := telemetrytypes.TelemetryFieldKey{
+				Name:          "body",
+				FieldContext:  telemetrytypes.FieldContextLog,
+				FieldDataType: telemetrytypes.FieldDataTypeString,
+			}
+			sb := sqlbuilder.NewSelectBuilder()
+			sb.Select("1").From("t")
+			cond, _, err := conditionBuilder.ConditionFor(context.Background(), valuer.UUID{}, 0, 0, &key,
+				map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, qbtypes.ConditionBuilderOptions{},
+				qbtypes.FilterOperatorIn, tc.values, sb)
+			require.NoError(t, err)
+			sb.Where(cond...)
+
+			sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+			assert.Contains(t, sql, tc.expectedSQL)
+			assert.Equal(t, tc.expectedArgs, args)
+		})
+	}
+}

pkg/telemetryschema/tracestelemetryschema/condition_builder.go

@@ -135,7 +135,11 @@ func (c *conditionBuilder) conditionFor(
 		// instead of using IN, we use `=` + `OR` to make use of index
 		conditions := []string{}
 		for _, value := range values {
-			conditions = append(conditions, sb.E(fieldExpression, value))
+			cond, err := c.conditionFor(ctx, orgID, startNs, endNs, key, qbtypes.FilterOperatorEqual, value, sb)
+			if err != nil {
+				return "", err
+			}
+			conditions = append(conditions, cond)
 		}
 		return sb.Or(conditions...), nil
 	case qbtypes.FilterOperatorNotIn:
@@ -146,7 +150,11 @@ func (c *conditionBuilder) conditionFor(
 		// instead of using NOT IN, we use `!=` + `AND` to make use of index
 		conditions := []string{}
 		for _, value := range values {
-			conditions = append(conditions, sb.NE(fieldExpression, value))
+			cond, err := c.conditionFor(ctx, orgID, startNs, endNs, key, qbtypes.FilterOperatorNotEqual, value, sb)
+			if err != nil {
+				return "", err
+			}
+			conditions = append(conditions, cond)
 		}
 		return sb.And(conditions...), nil
 

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/06_json_body.py

@@ -3,11 +3,13 @@ from collections.abc import Callable
 from datetime import UTC, datetime, timedelta
 from http import HTTPStatus
 
+import pytest
 import requests
 
 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
 
 
 def test_logs_json_body_simple_searches(
@@ -911,3 +913,61 @@ def test_logs_json_body_listing(
     assert len(results) == 1
     count = results[0]["data"][0][0]
     assert count == 4  # 4 logs have status="success"
+
+
+@pytest.mark.parametrize(
+    "expression,expected_services",
+    [
+        pytest.param("body.service IN ['auth', 'payment']", {"auth", "payment"}, id="in_scalar_path"),
+        pytest.param("body.status IN [200, 500]", {"auth", "payment"}, id="in_number_path"),
+        pytest.param("body.service NOT IN ['auth']", {"payment", "search"}, id="not_in_scalar_path"),
+        # An `[]` path is extracted as an array. Comparing that array to each scalar in the
+        # list is something ClickHouse rejects outright (code 130), so this shape used to
+        # fail the whole query; per-value extraction reads the first element instead.
+        pytest.param("body.user_names[*] IN ['alpha', 'gamma']", {"auth", "payment"}, id="in_array_path"),
+    ],
+)
+def test_logs_json_body_in_operator(
+    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_services: set[str],
+) -> None:
+    """IN over a body JSON path fans out to one comparison per value."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    specs = [("auth", 200, ["alpha", "beta"]), ("payment", 500, ["gamma"]), ("search", 404, ["beta", "alpha"])]
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": "api"},
+                body=json.dumps({"service": service, "status": status, "user_names": user_names}),
+            )
+            for i, (service, status, user_names) in enumerate(specs)
+        ]
+    )
+
+    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")],
+                limit=100,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["status"] == "success"
+    # flag off: the body comes back as the raw JSON string
+    assert {json.loads(row["data"]["body"])["service"] for row in get_rows(response)} == expected_services

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