refactor(qb): build IN as an OR of equalities

The IN and NOT IN cases route each value back through the condition builder
with `=` / `!=`, instead of assembling the comparisons themselves. Whatever a
builder does for a scalar comparison then applies to the list form without
being restated.

Applied to logs, traces, audit and resourcefilter, which all fanned a list out
into per-value comparisons already. Metrics and rulestatehistory build a real
sb.In, so there is nothing to delegate to. telemetrymetadata is left alone as
well: it applies a key-existence guard at a single exit, so a recursed arm
comes back already wrapped, and either the guard nests or the case has to skip
the shared tail and lose the invariant that every case is guarded.

This fixes `body.<path>[*] IN [...]` with use_json_body off, which returned a
500. The list shape made the path extract as Array(String), and comparing that
to each scalar is something ClickHouse rejects outright (code 130); extracting
per value reads the field instead. Covered end-to-end by the new case in
querierlogs/06_json_body.py, which fails on main and passes here.

resourcefilter changes shape without changing results: each value is paired
with its own index filter — `(e1 AND k AND l1) OR (e2 AND k AND l2)` rather
than `(e1 OR e2) AND k AND (l1 OR l2)` — which selects the same rows because
each equality implies its own filter.

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[*]")},
 			},

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

@@ -410,7 +410,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 +425,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

@@ -905,3 +905,52 @@ 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.
+func TestConditionForBodyIn(t *testing.T) {
+	testCases := []struct {
+		name         string
+		values       []any
+		expectedSQL  string
+		expectedArgs []any
+	}{
+		{
+			name:         "strings",
+			values:       []any{"alpha", "beta"},
+			expectedSQL:  "(body = ? OR body = ?)",
+			expectedArgs: []any{"alpha", "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"},
+		},
+	}
+
+	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/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