perf(logs): let attribute filters use the mapValues bloom filters

attributes_string and attributes_number each carry a bloom filter over mapValues, but the
subscript a filter reads matches no index expression, so a filter on a key every row carries -
a status code, a client address - reads every granule. The mapKeys index prunes only when the
key itself is rare. Both cases below AND in a predicate over mapValues that the original filter
already implies. On 1M rows, 1000001 rows read down to 8192, returning the same rows.

- A numeric equality carries the membership it implies. conditionForKey pairs a positive
  comparison with mapContains, so the key is present and the value compared is one of the map's
  own - which is what makes the assertion hold for the zero a subscript returns for an absent
  key. A non-numeric value means the collision handler compared the column as text, where an
  Array(Float64) membership check has no supertype, so it carries no predicate.
- A case-insensitive match reaches the raw-valued index only for a pattern holding no ASCII
  letter, LOWER being the identity on those bytes; a letter breaks the implication, since an `a`
  in the pattern may have come from an `A` in the value. That covers the values case folding
  cannot alter: addresses, timestamps, ports, numeric ids.

Both read a column the query already reads - a Map subscript decompresses keys and values
either way - so when the filter does not prune, bytes read are unchanged.

frontend/src/components/LogDetail/__tests__/LogDetail.test.tsx

@@ -18,9 +18,10 @@ jest.mock('periscope/components/DataViewer', () => ({
 	DataViewer: (): JSX.Element => <div data-testid="overview-data-viewer" />,
 }));
 
-// Force v2 for these tests regardless of route.
-jest.mock('../useIsLogDetailsV2', () => ({
-	useIsLogDetailsV2: (): boolean => true,
+// The flag to be removed later
+jest.mock('../constants', () => ({
+	...jest.requireActual('../constants'),
+	isLogDetailsV2: true,
 }));
 
 const mockLog: ILog = {

frontend/src/components/LogDetail/constants.ts

@@ -1,3 +1,6 @@
+// temporary flag to be removed with old log details code.
+export const isLogDetailsV2 = true;
+
 export const VIEW_TYPES = {
 	OVERVIEW: 'OVERVIEW',
 	JSON: 'JSON',

frontend/src/components/LogDetail/index.tsx

@@ -51,12 +51,11 @@ import { ILogBody } from 'types/api/logs/log';
 import { Query, TagFilter } from 'types/api/queryBuilder/queryBuilderData';
 import { DataSource, StringOperators } from 'types/common/queryBuilder';
 
-import { RESOURCE_KEYS, VIEW_TYPES, VIEWS } from './constants';
+import { isLogDetailsV2, RESOURCE_KEYS, VIEW_TYPES, VIEWS } from './constants';
 import { LogDetailInnerProps, LogDetailProps } from './LogDetail.interfaces';
 import LogDetailsHeader from './LogDetailsHeader/LogDetailsHeader';
 import { useLogNavigation } from './LogDetailsHeader/useLogNavigation';
 import LogHighlights from './LogHighlights/LogHighlights';
-import { useIsLogDetailsV2 } from './useIsLogDetailsV2';
 
 import './LogDetails.styles.scss';
 
@@ -93,8 +92,6 @@ function LogDetailInner({
 	const [isEdit, setIsEdit] = useState<boolean>(false);
 	const { stagedQuery } = useQueryBuilder();
 
-	const isLogDetailsV2 = useIsLogDetailsV2();
-
 	// Handle clicks outside to close drawer, except on explicitly ignored regions
 	useEffect(() => {
 		const handleClickOutside = (e: MouseEvent): void => {

frontend/src/components/LogDetail/useIsLogDetailsV2.ts

@@ -1,9 +0,0 @@
-import ROUTES from 'constants/routes';
-import { useLocation } from 'react-router-dom';
-
-// v2 is rolled out only on the logs explorer route for now; every other surface
-// (dashboards, infra monitoring, etc.) keeps the v1 log details view.
-export function useIsLogDetailsV2(): boolean {
-	const { pathname } = useLocation();
-	return pathname === ROUTES.LOGS_EXPLORER;
-}

frontend/src/container/Home/AlertRules/AlertRules.tsx

@@ -139,6 +139,11 @@ export default function AlertRules({
 			encodeURIComponent(JSON.stringify(compositeQuery)),
 		);
 
+		const panelType = record.condition.compositeQuery.panelType;
+		if (panelType) {
+			params.set(QueryParams.panelTypes, panelType);
+		}
+
 		params.set(QueryParams.ruleId, record.id);
 
 		history.push(`${ROUTES.ALERT_OVERVIEW}?${params.toString()}`);

frontend/src/container/ListAlertRules/__tests__/ListAlertRules.rowClick.test.tsx

@@ -26,7 +26,7 @@ describe('ListAlertRules — row click navigation', () => {
 		const [url] = safeNavigateMock.mock.calls[0];
 		expect(url).toContain('/alerts/overview?');
 		expect(url).toContain('ruleId=rule-1');
-		expect(url).not.toContain('panelTypes');
+		expect(url).toContain('panelTypes=graph');
 		expect(url).toContain('compositeQuery=');
 	});
 

frontend/src/container/ListAlertRules/useAlertRulesHandlers.ts

@@ -36,6 +36,11 @@ export function useAlertRulesHandlers(
 				encodeURIComponent(JSON.stringify(compositeQuery)),
 			);
 
+			const panelType = rule.condition.compositeQuery.panelType;
+			if (panelType) {
+				params.set(QueryParams.panelTypes, panelType);
+			}
+
 			params.set(QueryParams.ruleId, rule.id);
 
 			return `${ROUTES.ALERT_OVERVIEW}?${params.toString()}`;

frontend/src/container/LogDetailedView/Overview.tsx

@@ -13,7 +13,7 @@ import { ChangeViewFunctionType } from 'container/ExplorerOptions/types';
 import { OptionsQuery } from 'container/OptionsMenu/types';
 import { useIsDarkMode } from 'hooks/useDarkMode';
 import { ChevronDown, ChevronRight, Search } from '@signozhq/icons';
-import { useIsLogDetailsV2 } from 'components/LogDetail/useIsLogDetailsV2';
+import { isLogDetailsV2 } from 'components/LogDetail/constants';
 import { DataViewer } from 'periscope/components/DataViewer';
 import { IField } from 'types/api/logs/fields';
 import { ILog } from 'types/api/logs/log';
@@ -69,8 +69,6 @@ function Overview({
 		isListViewPanel,
 	});
 
-	const isLogDetailsV2 = useIsLogDetailsV2();
-
 	if (isLogDetailsV2) {
 		const raw = aggregateAttributesResourcesToObject(logData);
 		const prettyData = buildPrettyViewData(raw);

frontend/src/hooks/queryBuilder/useCreateAlerts.tsx

@@ -95,6 +95,7 @@ const useCreateAlerts = (widget?: Widgets, caller?: string): VoidFunction => {
 					QueryParams.compositeQuery,
 					encodeURIComponent(JSON.stringify(updatedQuery)),
 				);
+				params.set(QueryParams.panelTypes, widget.panelTypes);
 				params.set(QueryParams.version, ENTITY_VERSION_V5);
 				params.set(QueryParams.source, YAxisSource.DASHBOARDS);
 

frontend/src/hooks/queryBuilder/useGetPanelTypesQueryParam.test.tsx

@@ -1,54 +0,0 @@
-import { Router } from 'react-router-dom';
-import { renderHook } from '@testing-library/react';
-import { PANEL_TYPES } from 'constants/queryBuilder';
-import { createMemoryHistory } from 'history';
-
-import { useGetPanelTypesQueryParam } from './useGetPanelTypesQueryParam';
-
-const renderWithSearch = (
-	search: string,
-	defaultPanelType?: PANEL_TYPES,
-): PANEL_TYPES | null => {
-	const history = createMemoryHistory({
-		initialEntries: [`/logs/logs-explorer${search}`],
-	});
-
-	const { result } = renderHook(
-		() => useGetPanelTypesQueryParam(defaultPanelType),
-		{
-			wrapper: ({ children }) => <Router history={history}>{children}</Router>,
-		},
-	);
-
-	return result.current;
-};
-
-describe('useGetPanelTypesQueryParam', () => {
-	it('reads a JSON encoded panel type, as written by the explorers', () => {
-		expect(renderWithSearch('?panelTypes=%22table%22', PANEL_TYPES.LIST)).toBe(
-			PANEL_TYPES.TABLE,
-		);
-	});
-
-	it('reads a plain string panel type, as written by the alerts flow', () => {
-		expect(renderWithSearch('?panelTypes=graph', PANEL_TYPES.LIST)).toBe(
-			PANEL_TYPES.TIME_SERIES,
-		);
-	});
-
-	it('falls back to the default for an unparseable panel type', () => {
-		expect(renderWithSearch('?panelTypes=%7Bfoo', PANEL_TYPES.LIST)).toBe(
-			PANEL_TYPES.LIST,
-		);
-	});
-
-	it('falls back to the default for a value that is not a panel type', () => {
-		expect(renderWithSearch('?panelTypes=%22nope%22', PANEL_TYPES.LIST)).toBe(
-			PANEL_TYPES.LIST,
-		);
-	});
-
-	it('falls back to the default when the param is absent', () => {
-		expect(renderWithSearch('', PANEL_TYPES.LIST)).toBe(PANEL_TYPES.LIST);
-	});
-});

frontend/src/hooks/queryBuilder/useGetPanelTypesQueryParam.ts

@@ -3,24 +3,6 @@ import { QueryParams } from 'constants/query';
 import { PANEL_TYPES } from 'constants/queryBuilder';
 import useUrlQuery from 'hooks/useUrlQuery';
 
-const PANEL_TYPE_VALUES = new Set<string>(Object.values(PANEL_TYPES));
-
-// The param is JSON encoded by the explorers and written as a plain string by the
-// alerts flow, so accept both and treat anything unrecognised as absent.
-const parsePanelType = (value: string): PANEL_TYPES | null => {
-	let parsed: unknown = value;
-
-	try {
-		parsed = JSON.parse(value);
-	} catch {
-		parsed = value;
-	}
-
-	return typeof parsed === 'string' && PANEL_TYPE_VALUES.has(parsed)
-		? (parsed as PANEL_TYPES)
-		: null;
-};
-
 export const useGetPanelTypesQueryParam = <T extends PANEL_TYPES | undefined>(
 	defaultPanelType?: T,
 ): T extends undefined ? PANEL_TYPES | null : PANEL_TYPES => {
@@ -29,10 +11,6 @@ export const useGetPanelTypesQueryParam = <T extends PANEL_TYPES | undefined>(
 	return useMemo(() => {
 		const panelTypeQuery = urlQuery.get(QueryParams.panelTypes);
 
-		return (
-			(panelTypeQuery ? parsePanelType(panelTypeQuery) : null) ?? defaultPanelType
-		);
-	}, [urlQuery, defaultPanelType]) as T extends undefined
-		? PANEL_TYPES | null
-		: PANEL_TYPES;
+		return panelTypeQuery ? JSON.parse(panelTypeQuery) : defaultPanelType;
+	}, [urlQuery, defaultPanelType]);
 };

frontend/src/pages/DashboardPageV2/DashboardContainer/PanelsAndSectionsLayout/Panel/hooks/__tests__/useCreateAlertFromPanel.test.ts

@@ -168,6 +168,7 @@ describe('useCreateAlertFromPanel', () => {
 			// The resolved query is seeded with the panel-derived alert prefill.
 			expect(mockBuildAlertUrl).toHaveBeenCalledWith(
 				{ resolved: 'query' },
+				PANEL_TYPES.TIME_SERIES,
 				undefined,
 				mockPrefill,
 			);

frontend/src/pages/DashboardPageV2/DashboardContainer/PanelsAndSectionsLayout/Panel/hooks/useCreateAlertFromPanel.ts

@@ -79,6 +79,7 @@ export function useCreateAlertFromPanel(): (
 						const unit = readPanelUnit(panel.spec.plugin);
 						const url = buildAlertUrl(
 							query,
+							panelType,
 							unit,
 							deriveAlertPrefill(panel, query, unit),
 						);

frontend/src/pages/DashboardPageV2/DashboardContainer/PanelsAndSectionsLayout/Panel/utils/__tests__/buildCreateAlertUrl.test.ts

@@ -65,19 +65,14 @@ describe('buildCreateAlertUrl', () => {
 		);
 	});
 
-	it('tags the URL with the v5 version and the dashboards source', () => {
+	it('tags the URL with panel type, v5 version, and the dashboards source', () => {
 		const params = parse(buildCreateAlertUrl(makePanel()));
 
+		expect(params.get(QueryParams.panelTypes)).toBe(PANEL_TYPES.TIME_SERIES);
 		expect(params.get(QueryParams.version)).toBe(ENTITY_VERSION_V5);
 		expect(params.get(QueryParams.source)).toBe('dashboards');
 	});
 
-	it('does not tag the URL with a panel type, which the alert page ignores', () => {
-		const params = parse(buildCreateAlertUrl(makePanel()));
-
-		expect(params.get(QueryParams.panelTypes)).toBeNull();
-	});
-
 	it('encodes the translated query as the compositeQuery param', () => {
 		const params = parse(buildCreateAlertUrl(makePanel()));
 

frontend/src/pages/DashboardPageV2/DashboardContainer/PanelsAndSectionsLayout/Panel/utils/buildCreateAlertUrl.ts

@@ -5,6 +5,7 @@ import type {
 import { YAxisSource } from 'components/YAxisUnitSelector/types';
 import { ENTITY_VERSION_V5 } from 'constants/app';
 import { QueryParams } from 'constants/query';
+import { PANEL_TYPES } from 'constants/queryBuilder';
 import ROUTES from 'constants/routes';
 import { PANEL_KIND_TO_PANEL_TYPE } from 'pages/DashboardPageV2/DashboardContainer/Panels/types/panelKind';
 import { fromPerses } from 'pages/DashboardPageV2/DashboardContainer/queryV5/persesQueryAdapters';
@@ -33,6 +34,7 @@ export function readPanelUnit(
  */
 export function buildAlertUrl(
 	query: Query,
+	panelType: PANEL_TYPES,
 	unit?: string,
 	prefill?: PanelAlertPrefill,
 ): string {
@@ -46,6 +48,7 @@ export function buildAlertUrl(
 		QueryParams.compositeQuery,
 		encodeURIComponent(JSON.stringify(query)),
 	);
+	params.set(QueryParams.panelTypes, panelType);
 	params.set(QueryParams.version, ENTITY_VERSION_V5);
 	params.set(QueryParams.source, YAxisSource.DASHBOARDS);
 
@@ -73,5 +76,10 @@ export function buildCreateAlertUrl(panel: DashboardtypesPanelDTO): string {
 	const panelType = PANEL_KIND_TO_PANEL_TYPE[panel.spec.plugin.kind];
 	const query = fromPerses(panel.spec.queries, panelType);
 	const unit = readPanelUnit(panel.spec.plugin);
-	return buildAlertUrl(query, unit, deriveAlertPrefill(panel, query, unit));
+	return buildAlertUrl(
+		query,
+		panelType,
+		unit,
+		deriveAlertPrefill(panel, query, unit),
+	);
 }

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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"strings"
 
 	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -74,6 +75,35 @@ func (c *conditionBuilder) conditionForSearch(
 	return []string{sb.Or(conditions...)}, nil, nil
 }
 
+// numberAttributeIndexPredicate returns what an equality on a numeric attribute implies over
+// mapValues(attributes_number), which its bloom filter indexes while the subscript the comparison
+// reads matches nothing. The paired mapContains is what makes membership hold for the zero default.
+func numberAttributeIndexPredicate(columns []*schema.Column, value any, sb *sqlbuilder.SelectBuilder) string {
+	if len(columns) != 1 || columns[0].Name != LogsV2AttributesNumberColumn {
+		return ""
+	}
+	// a non-numeric value means the collision handler compared the column as text, where an
+	// Array(Float64) membership check has no supertype
+	switch value.(type) {
+	case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64:
+		return fmt.Sprintf("has(mapValues(%s), %s)", LogsV2AttributesNumberColumn, sb.Var(value))
+	}
+	return ""
+}
+
+// stringAttributeIndexPredicate returns the raw-value match a case-insensitive one implies when
+// the pattern holds no ASCII letter, LOWER being the identity on those bytes. A letter breaks it:
+// an `a` in the pattern may have come from an `A` in the value.
+func stringAttributeIndexPredicate(columns []*schema.Column, fieldExpression, pattern string, sb *sqlbuilder.SelectBuilder) string {
+	if len(columns) != 1 || columns[0].Name != LogsV2AttributesStringColumn {
+		return ""
+	}
+	if strings.ContainsFunc(pattern, func(r rune) bool { return (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') }) {
+		return ""
+	}
+	return sb.Like(fieldExpression, pattern)
+}
+
 // isBodyJSONSearch reports whether a key addresses a path within the body JSON. Only
 // an explicit Body context qualifies; a bare, context-less `body` (e.g. full-text
 // `count_distinct(body)` or `body EXISTS`) is a full-text match, not a `$.body` path.
@@ -105,14 +135,14 @@ func (c *conditionBuilder) conditionForArrayFunction(
 			"function `%s` supports only body JSON search", operator.FunctionName()).WithUrl(functionBodyJSONSearchDocURL)
 	}
 
-	needle := value
+	element := value
 	if args, ok := value.([]any); ok && len(args) > 0 {
-		needle = args[0]
+		element = args[0]
 	}
 
 	if c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
 		// JSON access plan: data-type collision handling, nested array paths.
-		valueType, needle := InferDataType(needle, operator, key)
+		valueType, element := InferDataType(element, operator, key)
 		// A not-found (synthesized) body path carries no metadata plan; build an exhaustive
 		// one so the query runs against the underlying data (with the not-found warning)
 		// instead of erroring, matching the regular-operator path.
@@ -123,21 +153,21 @@ func (c *conditionBuilder) conditionForArrayFunction(
 			}
 			key = keyCopy
 		}
-		return NewJSONConditionBuilder(key, valueType).buildArrayFunctionCondition(operator, needle, sb)
+		return NewJSONConditionBuilder(key, valueType).buildArrayFunctionCondition(operator, element, sb)
 	}
 
 	// legacy string-body path: type-matched array extraction, OR-ed with a scalar comparison
 	// for a scalar body value (coalesced to false so NOT has() matches missing-key rows).
-	elemType := legacyElemType(needle)
+	elemType := legacyElemType(element)
 	arrayExpr := getBodyJSONArrayKey(key, elemType)
 	scalarExpr, scalarGuard, hasScalar := getBodyJSONScalarKey(key, elemType)
-	if list, ok := needle.([]any); ok {
+	if list, ok := element.([]any); ok {
 		vals := make([]any, len(list))
 		for i, v := range list {
-			vals[i] = legacyCoerceNeedle(v, elemType)
+			vals[i] = legacyCoerceElement(v, elemType)
 		}
-		// Pin the needle array type to the haystack; scalar fallback below coerces value-level.
-		arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, castNeedleArray(elemType, sb.Var(vals)))
+		// Pin the element array type to the array it is tested against; scalar fallback below coerces value-level.
+		arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, castElementArray(elemType, sb.Var(vals)))
 		if !hasScalar {
 			return arrayCond, nil
 		}
@@ -153,17 +183,17 @@ func (c *conditionBuilder) conditionForArrayFunction(
 		}
 		return fmt.Sprintf("(%s OR ifNull(%s, false))", arrayCond, sb.And(membership, scalarGuard)), nil
 	}
-	typedNeedle := legacyCoerceNeedle(needle, elemType)
-	arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, sb.Var(typedNeedle))
+	typedElement := legacyCoerceElement(element, elemType)
+	arrayCond := fmt.Sprintf("%s(%s, %s)", operator.FunctionName(), arrayExpr, sb.Var(typedElement))
 	if !hasScalar {
 		return arrayCond, nil
 	}
-	return fmt.Sprintf("(%s OR ifNull(%s, false))", arrayCond, sb.And(sb.E(scalarExpr, typedNeedle), scalarGuard)), nil
+	return fmt.Sprintf("(%s OR ifNull(%s, false))", arrayCond, sb.And(sb.E(scalarExpr, typedElement), scalarGuard)), nil
 }
 
-// castNeedleArray pins an Int64 needle array to Array(Int64) so it matches the Array(Nullable(Int64))
-// haystack; without it a needle >= 2^32 binds as Array(UInt64) and hasAny/hasAll error (code 386).
-func castNeedleArray(elemType telemetrytypes.FieldDataType, arg string) string {
+// castElementArray pins an Int64 element array to Array(Int64) so it matches the Array(Nullable(Int64))
+// it is tested against; without it an element >= 2^32 binds as Array(UInt64) and hasAny/hasAll error (code 386).
+func castElementArray(elemType telemetrytypes.FieldDataType, arg string) string {
 	if elemType == telemetrytypes.FieldDataTypeInt64 {
 		return fmt.Sprintf("CAST(%s AS Array(Int64))", arg)
 	}
@@ -191,24 +221,24 @@ func (c *conditionBuilder) conditionForHasToken(
 	value any,
 	sb *sqlbuilder.SelectBuilder,
 ) (string, error) {
-	// hasToken takes a single needle; unwrap it from the function-argument slice.
-	needle := value
+	// hasToken takes a single token; unwrap it from the function-argument slice.
+	token := value
 	if args, ok := value.([]any); ok && len(args) > 0 {
-		needle = args[0]
+		token = args[0]
 	}
 
 	// hasToken matches string tokens only.
-	needleStr, ok := needle.(string)
+	tokenStr, ok := token.(string)
 	if !ok {
 		return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
 			"function `hasToken` expects value parameter to be a string").WithUrl(hasTokenFunctionDocURL)
 	}
 
-	// A multi-token needle makes CH hasToken error (code 36); reject up front as a 400. Both modes flow here.
-	if sep, found := firstTokenSeparator(needleStr); found {
+	// A multi-token value makes CH hasToken error (code 36); reject up front as a 400. Both modes flow here.
+	if sep, found := firstTokenSeparator(tokenStr); found {
 		return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
 			"function `hasToken` matches a single whole token, but %q contains the separator %q; use a substring filter (e.g. `body CONTAINS '%s'`) to search across separators",
-			needleStr, sep, needleStr).WithUrl(hasTokenFunctionDocURL)
+			tokenStr, sep, tokenStr).WithUrl(hasTokenFunctionDocURL)
 	}
 
 	bodyJSONEnabled := c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID))
@@ -219,7 +249,7 @@ func (c *conditionBuilder) conditionForHasToken(
 			return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
 				"function `hasToken` only supports body field as first parameter").WithUrl(hasTokenFunctionDocURL)
 		}
-		return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", LogsV2BodyColumn, sb.Var(needle)), nil
+		return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", LogsV2BodyColumn, sb.Var(token)), nil
 	}
 
 	// JSON mode: a bare body/body.message key searches the body.message column; any other body
@@ -228,7 +258,7 @@ func (c *conditionBuilder) conditionForHasToken(
 	// falls through and emits dynamicElement over the already-typed String column, which errors.
 	if key.Name == LogsV2BodyColumn || key.Name == bodyMessageField ||
 		(key.FieldContext == telemetrytypes.FieldContextBody && key.Name == messageSubField) {
-		return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", bodyMessageField, sb.Var(needle)), nil
+		return fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", bodyMessageField, sb.Var(token)), nil
 	}
 	if key.FieldContext == telemetrytypes.FieldContextBody {
 		// A not-found (synthesized) body path carries no metadata plan; build an exhaustive
@@ -240,7 +270,7 @@ func (c *conditionBuilder) conditionForHasToken(
 			}
 			key = keyCopy
 		}
-		return NewJSONConditionBuilder(key, telemetrytypes.FieldDataTypeString).buildTokenFunctionCondition(needle, sb)
+		return NewJSONConditionBuilder(key, telemetrytypes.FieldDataTypeString).buildTokenFunctionCondition(token, sb)
 	}
 	return "", errors.NewInvalidInputf(errors.CodeInvalidInput,
 		"function `hasToken` only supports the body field or a body JSON string field as first parameter").WithUrl(hasTokenFunctionDocURL)
@@ -254,12 +284,21 @@ func (c *conditionBuilder) conditionForResolvedKey(
 	operator qbtypes.FilterOperator,
 	value any,
 	sb *sqlbuilder.SelectBuilder,
-) (string, error) {
+) (condition string, err error) {
 	// hasToken resolves from the key name + flag alone (no column resolution), so handle it first.
 	if operator == qbtypes.FilterOperatorHasToken {
 		return c.conditionForHasToken(ctx, orgID, key, value, sb)
 	}
 
+	// What the legacy body JSON path implies over the indexed LOWER(body), ANDed onto whichever
+	// condition the operator builds below — which still decides the row.
+	var bodyIndexPredicates []string
+	defer func() {
+		if err == nil && len(bodyIndexPredicates) > 0 {
+			condition = sb.And(append([]string{condition}, bodyIndexPredicates...)...)
+		}
+	}()
+
 	columns, err := c.fm.ColumnFor(ctx, orgID, startNs, endNs, key)
 	if errors.Is(err, qbtypes.ErrColumnNotFound) && key.FieldContext == telemetrytypes.FieldContextUnspecified {
 		key = telemetrytypes.NewTelemetryFieldKey(key.Name, telemetrytypes.FieldContextBody, key.FieldDataType)
@@ -269,14 +308,20 @@ 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() {
+		if legacyBodyJSONSearch {
+			bodyIndexPredicates = legacyBodyIndexPredicates(key, operator, value, sb)
+		}
 		return c.conditionForArrayFunction(ctx, orgID, key, operator, value, columns, sb)
 	}
 
 	// TODO(Piyush): Update this to support multiple JSON columns based on evolutions
 	for _, column := range columns {
-		if column.Type.GetType() == schema.ColumnTypeEnumJSON && isBodyJSONSearch(key, columns) && c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) && key.Name != messageSubField {
+		if column.Type.GetType() == schema.ColumnTypeEnumJSON && isBodyJSONSearch(key, columns) && useJSONBody && key.Name != messageSubField {
 			valueType, value := InferDataType(value, operator, key)
 			if len(key.JSONPlan) == 0 {
 				keyCopy := telemetrytypes.NewTelemetryFieldKey(key.Name, key.FieldContext, key.FieldDataType)
@@ -305,8 +350,9 @@ func (c *conditionBuilder) conditionForResolvedKey(
 	}
 
 	// Check if this is a body JSON search (legacy string-body path, JSON flag off).
-	if isBodyJSONSearch(key, columns) && !c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
+	if legacyBodyJSONSearch {
 		fieldExpression, value = GetBodyJSONKey(ctx, key, operator, value)
+		bodyIndexPredicates = legacyBodyIndexPredicates(key, operator, value, sb)
 	}
 
 	fieldExpression, value = querybuilder.DataTypeCollisionHandledFieldName(key, value, fieldExpression, operator)
@@ -314,10 +360,21 @@ 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:
-			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)
@@ -333,6 +390,9 @@ func (c *conditionBuilder) conditionForResolvedKey(
 	switch operator {
 	// regular operators
 	case qbtypes.FilterOperatorEqual:
+		if predicate := numberAttributeIndexPredicate(columns, value, sb); predicate != "" {
+			return sb.And(sb.E(fieldExpression, value), predicate), nil
+		}
 		return sb.E(fieldExpression, value), nil
 	case qbtypes.FilterOperatorNotEqual:
 		return sb.NE(fieldExpression, value), nil
@@ -351,12 +411,17 @@ func (c *conditionBuilder) conditionForResolvedKey(
 	case qbtypes.FilterOperatorNotLike:
 		return sb.NotLike(fieldExpression, value), nil
 	case qbtypes.FilterOperatorILike:
+		if pattern, ok := value.(string); ok {
+			if predicate := stringAttributeIndexPredicate(columns, fieldExpression, pattern, sb); predicate != "" {
+				return sb.And(sb.ILike(fieldExpression, pattern), predicate), nil
+			}
+		}
 		return sb.ILike(fieldExpression, value), nil
 	case qbtypes.FilterOperatorNotILike:
 		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 legacyBodyJSONSearch {
 			if operator == qbtypes.FilterOperatorExists {
 				return GetBodyJSONKeyForExists(ctx, key, operator, value), nil
 			}
@@ -369,7 +434,13 @@ func (c *conditionBuilder) conditionForResolvedKey(
 		return sqlbuilder.Escape(pred), nil
 
 	case qbtypes.FilterOperatorContains:
-		return sb.ILike(fieldExpression, fmt.Sprintf("%%%s%%", value)), nil
+		// The map value indexes are over raw mapValues, which a case-insensitive match reaches only
+		// for the patterns stringAttributeIndexPredicate can assert the raw value from.
+		pattern := fmt.Sprintf("%%%s%%", value)
+		if predicate := stringAttributeIndexPredicate(columns, fieldExpression, pattern, sb); predicate != "" {
+			return sb.And(sb.ILike(fieldExpression, pattern), predicate), nil
+		}
+		return sb.ILike(fieldExpression, pattern), nil
 	case qbtypes.FilterOperatorNotContains:
 		return sb.NotILike(fieldExpression, fmt.Sprintf("%%%s%%", value)), 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,
 		},
 		{
@@ -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,
 		},
@@ -258,8 +258,8 @@ func TestConditionFor(t *testing.T) {
 			},
 			operator:      qbtypes.FilterOperatorContains,
 			value:         521509198310,
-			expectedSQL:   "LOWER(attributes_string['user.id']) LIKE LOWER(?)",
-			expectedArgs:  []any{"%521509198310%"},
+			expectedSQL:   "(LOWER(attributes_string['user.id']) LIKE LOWER(?) AND attributes_string['user.id'] LIKE ?)",
+			expectedArgs:  []any{"%521509198310%", "%521509198310%"},
 			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,213 @@ func TestConditionForBodyIn(t *testing.T) {
 		})
 	}
 }
+
+// ClickHouse treats `\` as an escape only before `%`, `_` and itself.
+func TestLikePatternLiterals(t *testing.T) {
+	testCases := []struct {
+		name     string
+		pattern  string
+		expected []string
+	}{
+		{"contains wraps a plain value", "%error%", []string{"error"}},
+		{"wildcards split runs", "%foo%bar%", []string{"foo", "bar"}},
+		{"underscore splits too", "a_b", []string{"a", "b"}},
+		{"escaped wildcards stay literal", `%100\%\_off%`, []string{`100%_off`}},
+		{"escaped backslash collapses", `%C:\\tmp%`, []string{`C:\tmp`}},
+		{"backslash before other chars is literal", `%C:\tmp%`, []string{`C:\tmp`}},
+		{"trailing backslash is literal", `%path\`, []string{`path\`}},
+		{"no literals at all", "%_%", nil},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, likePatternLiterals(tc.pattern))
+		})
+	}
+}
+
+// The literals have to hold whichever encoder wrote the body, so the runs stop at every byte
+// one of them may rewrite.
+func TestJSONTextRuns(t *testing.T) {
+	testCases := []struct {
+		name     string
+		value    string
+		expected []string
+	}{
+		{"plain text is one run", "checkout failed", []string{"checkout failed"}},
+		{"a run with nothing to split on is kept whole", "abc", []string{"abc"}},
+		{"quote splits the run", `say "hello there"`, []string{"say ", "hello there"}},
+		{"slash splits the run, PHP escapes it", "/api/v1/users", []string{"api", "v1", "users"}},
+		{"ampersand and angles split, Go escapes them", "a&b<c>dddd", []string{"a", "b", "c", "dddd"}},
+		{"non-ascii splits, Python escapes it", "order café latte", []string{"order caf", " latte"}},
+		{"newline splits", "line one\nline two", []string{"line one", "line two"}},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, jsonTextRuns(tc.value))
+		})
+	}
+}
+
+func TestBodyPathLiterals(t *testing.T) {
+	testCases := []struct {
+		name     string
+		key      string
+		expected []string
+	}{
+		{"quoting lifts a short name over the ngram size", "id", []string{`"id"`}},
+		{"one literal per component", "response.status_code", []string{`"response"`, `"status_code"`}},
+		{"array suffixes are trimmed", "items[*].sku", []string{`"items"`, `"sku"`}},
+		{"every component is carried", "a.b.count", []string{`"a"`, `"b"`, `"count"`}},
+		{"a component an encoder may rewrite is dropped", "user/name.email", []string{`"email"`}},
+		{"nothing usable", "us/er.na/me", nil},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			key := telemetrytypes.NewTelemetryFieldKey(tc.key, telemetrytypes.FieldContextBody, telemetrytypes.FieldDataTypeUnspecified)
+			assert.Equal(t, tc.expected, bodyPathLiterals(key))
+		})
+	}
+}
+
+// The path literals ride on the existence assertion and the value literals on the comparison, so
+// a filter carries each at most once. Nothing rides on a negated operator: it matches rows
+// without the path, which say nothing about the body text.
+func TestLegacyBodyIndexPredicates(t *testing.T) {
+	testCases := []struct {
+		name         string
+		key          string
+		operator     qbtypes.FilterOperator
+		value        any
+		expected     string
+		expectedArgs []any
+	}{
+		{
+			name:         "exists carries the path",
+			key:          "user_id",
+			operator:     qbtypes.FilterOperatorExists,
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"user\_id"%`},
+		},
+		{
+			name:         "equality carries the value",
+			key:          "status",
+			operator:     qbtypes.FilterOperatorEqual,
+			value:        "timeout_error",
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%timeout\_error%`},
+		},
+		{
+			name:         "contains carries the value",
+			key:          "message",
+			operator:     qbtypes.FilterOperatorContains,
+			value:        "upstream refused",
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%upstream refused%`},
+		},
+		{
+			name:         "like carries one literal per run of the pattern",
+			key:          "message",
+			operator:     qbtypes.FilterOperatorLike,
+			value:        "%conn%refused%",
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%conn%refused%`},
+		},
+		{
+			name:     "has carries the path and the element",
+			key:      "tags[*]",
+			operator: qbtypes.FilterOperatorHas,
+			value:    []any{"production"},
+			// The element rides on its own predicate rather than being pinned next to the key:
+			// has() over the extracted array says nothing about where in the text it sits.
+			expected:     `LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"tags"%`, "%production%"},
+		},
+		{
+			name:         "hasAll carries one literal per element",
+			key:          "tags[*]",
+			operator:     qbtypes.FilterOperatorHasAll,
+			value:        []any{[]any{"production", "webserver"}},
+			expected:     `LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?) AND LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"tags"%`, "%production%", "%webserver%"},
+		},
+		{
+			// hasAny asks for one of the elements, so the arms are ORed — ANDing them would
+			// demand every element be present.
+			name:         "hasAny ORs the element literals",
+			key:          "tags[*]",
+			operator:     qbtypes.FilterOperatorHasAny,
+			value:        []any{[]any{"production", "webserver"}},
+			expected:     `LOWER(body) LIKE LOWER(?) AND (LOWER(body) LIKE LOWER(?) OR LOWER(body) LIKE LOWER(?))`,
+			expectedArgs: []any{`%"tags"%`, "%production%", "%webserver%"},
+		},
+		{
+			// one element with no usable literal voids the whole OR: the filter can still match
+			// through that element, so nothing about the text is implied.
+			name:         "hasAny drops the OR when an element carries no literal",
+			key:          "tags[*]",
+			operator:     qbtypes.FilterOperatorHasAny,
+			value:        []any{[]any{"production", "/"}},
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"tags"%`},
+		},
+		{
+			name:         "numeric elements carry nothing",
+			key:          "ids[*]",
+			operator:     qbtypes.FilterOperatorHasAny,
+			value:        []any{[]any{"9007199254740993", "9007199254740994"}},
+			expected:     `LOWER(body) LIKE LOWER(?)`,
+			expectedArgs: []any{`%"ids"%`},
+		},
+		{
+			name:     "a number carries nothing",
+			key:      "user_id",
+			operator: qbtypes.FilterOperatorEqual,
+			value:    int64(123),
+		},
+		{
+			name:     "IN leaves it to the equalities it delegates to",
+			key:      "status",
+			operator: qbtypes.FilterOperatorIn,
+			value:    []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",
+		},
+	}
+
+	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)
+
+			predicates := legacyBodyIndexPredicates(key, tc.operator, tc.value, sb)
+			if tc.expected == "" {
+				assert.Empty(t, predicates)
+				return
+			}
+
+			sb.Where(predicates...)
+			query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+			assert.Contains(t, query, tc.expected)
+			assert.Equal(t, tc.expectedArgs, args)
+		})
+	}
+}

pkg/telemetryschema/logstelemetryschema/filter_expr_logs_body_json_test.go

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

pkg/telemetryschema/logstelemetryschema/filter_expr_logs_test.go

@@ -495,32 +495,32 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "FREETEXT with parentheses",
 			query:                 "error (status.code=500 OR status.code=503)",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (match(LOWER(body), LOWER(?)) AND (((toFloat64(attributes_number['status.code']) = ? AND mapContains(attributes_number, 'status.code')) OR (toFloat64(attributes_number['status.code']) = ? AND mapContains(attributes_number, 'status.code')))))",
-			expectedArgs:          []any{"error", float64(500), float64(503)},
+			expectedQuery:         "WHERE (match(LOWER(body), LOWER(?)) AND ((((toFloat64(attributes_number['status.code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status.code')) OR ((toFloat64(attributes_number['status.code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status.code')))))",
+			expectedArgs:          []any{"error", float64(500), float64(500), float64(503), float64(503)},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "FREETEXT with parentheses",
 			query:                 "(status.code=500 OR status.code=503) error",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((((toFloat64(attributes_number['status.code']) = ? AND mapContains(attributes_number, 'status.code')) OR (toFloat64(attributes_number['status.code']) = ? AND mapContains(attributes_number, 'status.code')))) AND match(LOWER(body), LOWER(?)))",
-			expectedArgs:          []any{float64(500), float64(503), "error"},
+			expectedQuery:         "WHERE (((((toFloat64(attributes_number['status.code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status.code')) OR ((toFloat64(attributes_number['status.code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status.code')))) AND match(LOWER(body), LOWER(?)))",
+			expectedArgs:          []any{float64(500), float64(500), float64(503), float64(503), "error"},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "FREETEXT with parentheses",
 			query:                 "error AND (status.code=500 OR status.code=503)",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (match(LOWER(body), LOWER(?)) AND (((toFloat64(attributes_number['status.code']) = ? AND mapContains(attributes_number, 'status.code')) OR (toFloat64(attributes_number['status.code']) = ? AND mapContains(attributes_number, 'status.code')))))",
-			expectedArgs:          []any{"error", float64(500), float64(503)},
+			expectedQuery:         "WHERE (match(LOWER(body), LOWER(?)) AND ((((toFloat64(attributes_number['status.code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status.code')) OR ((toFloat64(attributes_number['status.code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status.code')))))",
+			expectedArgs:          []any{"error", float64(500), float64(500), float64(503), float64(503)},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "FREETEXT with parentheses",
 			query:                 "(status.code=500 OR status.code=503) AND error",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((((toFloat64(attributes_number['status.code']) = ? AND mapContains(attributes_number, 'status.code')) OR (toFloat64(attributes_number['status.code']) = ? AND mapContains(attributes_number, 'status.code')))) AND match(LOWER(body), LOWER(?)))",
-			expectedArgs:          []any{float64(500), float64(503), "error"},
+			expectedQuery:         "WHERE (((((toFloat64(attributes_number['status.code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status.code')) OR ((toFloat64(attributes_number['status.code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status.code')))) AND match(LOWER(body), LOWER(?)))",
+			expectedArgs:          []any{float64(500), float64(500), float64(503), float64(503), "error"},
 			expectedErrorContains: "",
 		},
 
@@ -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: "",
 		},
 		{
@@ -827,16 +827,16 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Basic equality",
 			query:                 "status=200",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status'))",
-			expectedArgs:          []any{float64(200)},
+			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status'))",
+			expectedArgs:          []any{float64(200), float64(200)},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "Basic equality",
 			query:                 "code=400",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (toFloat64(attributes_number['code']) = ? AND mapContains(attributes_number, 'code'))",
-			expectedArgs:          []any{float64(400)},
+			expectedQuery:         "WHERE ((toFloat64(attributes_number['code']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'code'))",
+			expectedArgs:          []any{float64(400), float64(400)},
 			expectedErrorContains: "",
 		},
 		{
@@ -867,8 +867,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Basic equality",
 			query:                 "count=0",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (toFloat64(attributes_number['count']) = ? AND mapContains(attributes_number, 'count'))",
-			expectedArgs:          []any{float64(0)},
+			expectedQuery:         "WHERE ((toFloat64(attributes_number['count']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'count'))",
+			expectedArgs:          []any{float64(0), float64(0)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1187,16 +1187,16 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "IN operator (parentheses)",
 			query:                 "status IN (200, 201, 202)",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? OR toFloat64(attributes_number['status']) = ? OR toFloat64(attributes_number['status']) = ?) AND mapContains(attributes_number, 'status'))",
-			expectedArgs:          []any{float64(200), float64(201), float64(202)},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) OR (toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) OR (toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?))) AND mapContains(attributes_number, 'status'))",
+			expectedArgs:          []any{float64(200), float64(200), float64(201), float64(201), float64(202), float64(202)},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "IN operator (parentheses)",
 			query:                 "error.code IN (404, 500, 503)",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['error.code']) = ? OR toFloat64(attributes_number['error.code']) = ? OR toFloat64(attributes_number['error.code']) = ?) AND mapContains(attributes_number, 'error.code'))",
-			expectedArgs:          []any{float64(404), float64(500), float64(503)},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['error.code']) = ? AND has(mapValues(attributes_number), ?)) OR (toFloat64(attributes_number['error.code']) = ? AND has(mapValues(attributes_number), ?)) OR (toFloat64(attributes_number['error.code']) = ? AND has(mapValues(attributes_number), ?))) AND mapContains(attributes_number, 'error.code'))",
+			expectedArgs:          []any{float64(404), float64(404), float64(500), float64(500), float64(503), float64(503)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1221,16 +1221,16 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "IN operator (brackets)",
 			query:                 "status IN [200, 201, 202]",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? OR toFloat64(attributes_number['status']) = ? OR toFloat64(attributes_number['status']) = ?) AND mapContains(attributes_number, 'status'))",
-			expectedArgs:          []any{float64(200), float64(201), float64(202)},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) OR (toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) OR (toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?))) AND mapContains(attributes_number, 'status'))",
+			expectedArgs:          []any{float64(200), float64(200), float64(201), float64(201), float64(202), float64(202)},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "IN operator (brackets)",
 			query:                 "error.code IN [404, 500, 503]",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['error.code']) = ? OR toFloat64(attributes_number['error.code']) = ? OR toFloat64(attributes_number['error.code']) = ?) AND mapContains(attributes_number, 'error.code'))",
-			expectedArgs:          []any{float64(404), float64(500), float64(503)},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['error.code']) = ? AND has(mapValues(attributes_number), ?)) OR (toFloat64(attributes_number['error.code']) = ? AND has(mapValues(attributes_number), ?)) OR (toFloat64(attributes_number['error.code']) = ? AND has(mapValues(attributes_number), ?))) AND mapContains(attributes_number, 'error.code'))",
+			expectedArgs:          []any{float64(404), float64(404), float64(500), float64(500), float64(503), float64(503)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1561,15 +1561,15 @@ func TestFilterExprLogs(t *testing.T) {
 			expectedArgs:          []any{"download"},
 			expectedErrorContains: "function `hasToken` expects value parameter to be a string",
 		},
-		// A multi-token needle (separator/whitespace) is a clean 400, not a CH execution error.
+		// A multi-token value (separator/whitespace) is a clean 400, not a CH execution error.
 		{
-			category:              "hasTokenUnderscoreNeedle",
+			category:              "hasTokenUnderscoreSeparator",
 			query:                 "hasToken(body, \"user_id\")",
 			shouldPass:            false,
 			expectedErrorContains: "function `hasToken` matches a single whole token",
 		},
 		{
-			category:              "hasTokenWhitespaceNeedle",
+			category:              "hasTokenWhitespaceSeparator",
 			query:                 "hasToken(body, \"production node\")",
 			shouldPass:            false,
 			expectedErrorContains: "function `hasToken` matches a single whole token",
@@ -1609,8 +1609,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Explicit AND",
 			query:                 "status=200 AND service.name=\"api\"",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
-			expectedArgs:          []any{float64(200), "api"},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
+			expectedArgs:          []any{float64(200), float64(200), "api"},
 			expectedErrorContains: "",
 		},
 		{
@@ -1635,8 +1635,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Explicit OR",
 			query:                 "status=200 OR status=201",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) OR (toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')))",
-			expectedArgs:          []any{float64(200), float64(201)},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) OR ((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')))",
+			expectedArgs:          []any{float64(200), float64(200), float64(201), float64(201)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1661,8 +1661,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "NOT with expressions",
 			query:                 "NOT status=200",
 			shouldPass:            true,
-			expectedQuery:         "WHERE NOT ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')))",
-			expectedArgs:          []any{float64(200)},
+			expectedQuery:         "WHERE NOT (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')))",
+			expectedArgs:          []any{float64(200), float64(200)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1687,8 +1687,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "AND + OR combinations",
 			query:                 "status=200 AND (service.name=\"api\" OR service.name=\"web\")",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))))",
-			expectedArgs:          []any{float64(200), "api", "web"},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))))",
+			expectedArgs:          []any{float64(200), float64(200), "api", "web"},
 			expectedErrorContains: "",
 		},
 		{
@@ -1713,8 +1713,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "AND + NOT combinations",
 			query:                 "status=200 AND NOT service.name=\"api\"",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND NOT ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))",
-			expectedArgs:          []any{float64(200), "api"},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND NOT ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))",
+			expectedArgs:          []any{float64(200), float64(200), "api"},
 			expectedErrorContains: "",
 		},
 		{
@@ -1731,8 +1731,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "OR + NOT combinations",
 			query:                 "NOT status=200 OR NOT service.name=\"api\"",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (NOT ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status'))) OR NOT ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))",
-			expectedArgs:          []any{float64(200), "api"},
+			expectedQuery:         "WHERE (NOT (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status'))) OR NOT ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))",
+			expectedArgs:          []any{float64(200), float64(200), "api"},
 			expectedErrorContains: "",
 		},
 		{
@@ -1749,8 +1749,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "AND + OR + NOT combinations",
 			query:                 "status=200 AND (service.name=\"api\" OR NOT duration>1000)",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR NOT ((toFloat64(attributes_number['duration']) > ? AND mapContains(attributes_number, 'duration'))))))",
-			expectedArgs:          []any{float64(200), "api", float64(1000)},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR NOT ((toFloat64(attributes_number['duration']) > ? AND mapContains(attributes_number, 'duration'))))))",
+			expectedArgs:          []any{float64(200), float64(200), "api", float64(1000)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1765,8 +1765,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "AND + OR + NOT combinations",
 			query:                 "NOT (status=200 AND service.name=\"api\") OR count>0",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (NOT ((((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))) OR (toFloat64(attributes_number['count']) > ? AND mapContains(attributes_number, 'count')))",
-			expectedArgs:          []any{float64(200), "api", float64(0)},
+			expectedQuery:         "WHERE (NOT (((((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))) OR (toFloat64(attributes_number['count']) > ? AND mapContains(attributes_number, 'count')))",
+			expectedArgs:          []any{float64(200), float64(200), "api", float64(0)},
 			expectedErrorContains: "",
 		},
 
@@ -1775,8 +1775,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Implicit AND",
 			query:                 "status=200 service.name=\"api\"",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
-			expectedArgs:          []any{float64(200), "api"},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
+			expectedArgs:          []any{float64(200), float64(200), "api"},
 			expectedErrorContains: "",
 		},
 		{
@@ -1801,8 +1801,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Mixed implicit/explicit AND",
 			query:                 "status=200 AND service.name=\"api\" duration<1000",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) AND (toFloat64(attributes_number['duration']) < ? AND mapContains(attributes_number, 'duration')))",
-			expectedArgs:          []any{float64(200), "api", float64(1000)},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) AND (toFloat64(attributes_number['duration']) < ? AND mapContains(attributes_number, 'duration')))",
+			expectedArgs:          []any{float64(200), float64(200), "api", float64(1000)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1819,8 +1819,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Simple grouping",
 			query:                 "(status=200)",
 			shouldPass:            true,
-			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')))",
-			expectedArgs:          []any{float64(200)},
+			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')))",
+			expectedArgs:          []any{float64(200), float64(200)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1845,8 +1845,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Nested grouping",
 			query:                 "((status=200))",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status'))))",
-			expectedArgs:          []any{float64(200)},
+			expectedQuery:         "WHERE ((((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status'))))",
+			expectedArgs:          []any{float64(200), float64(200)},
 			expectedErrorContains: "",
 		},
 		{
@@ -1871,8 +1871,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Complex nested grouping",
 			query:                 "(status=200 AND (service.name=\"api\" OR service.name=\"web\"))",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))))",
-			expectedArgs:          []any{float64(200), "api", "web"},
+			expectedQuery:         "WHERE ((((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))))",
+			expectedArgs:          []any{float64(200), float64(200), "api", "web"},
 			expectedErrorContains: "",
 		},
 		{
@@ -1897,8 +1897,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Deep nesting",
 			query:                 "(((status=200 OR status=201) AND service.name=\"api\") OR ((status=202 OR status=203) AND service.name=\"web\"))",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (((((((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) OR (toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')))) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))) OR (((((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) OR (toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')))) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))))",
-			expectedArgs:          []any{float64(200), float64(201), "api", float64(202), float64(203), "web"},
+			expectedQuery:         "WHERE ((((((((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) OR ((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')))) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))) OR ((((((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) OR ((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')))) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))))",
+			expectedArgs:          []any{float64(200), float64(200), float64(201), float64(201), "api", float64(202), float64(202), float64(203), float64(203), "web"},
 			expectedErrorContains: "",
 		},
 		{
@@ -1949,32 +1949,32 @@ func TestFilterExprLogs(t *testing.T) {
 			category:              "Numeric values",
 			query:                 "status=200",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status'))",
-			expectedArgs:          []any{float64(200)},
+			expectedQuery:         "WHERE ((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status'))",
+			expectedArgs:          []any{float64(200), float64(200)},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "Numeric values",
 			query:                 "count=0",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (toFloat64(attributes_number['count']) = ? AND mapContains(attributes_number, 'count'))",
-			expectedArgs:          []any{float64(0)},
+			expectedQuery:         "WHERE ((toFloat64(attributes_number['count']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'count'))",
+			expectedArgs:          []any{float64(0), float64(0)},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "Numeric values",
 			query:                 "duration=1000.5",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (toFloat64(attributes_number['duration']) = ? AND mapContains(attributes_number, 'duration'))",
-			expectedArgs:          []any{float64(1000.5)},
+			expectedQuery:         "WHERE ((toFloat64(attributes_number['duration']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'duration'))",
+			expectedArgs:          []any{float64(1000.5), float64(1000.5)},
 			expectedErrorContains: "",
 		},
 		{
 			category:              "Numeric values",
 			query:                 "amount=-10.25",
 			shouldPass:            true,
-			expectedQuery:         "WHERE (toFloat64(attributes_number['amount']) = ? AND mapContains(attributes_number, 'amount'))",
-			expectedArgs:          []any{float64(-10.25)},
+			expectedQuery:         "WHERE ((toFloat64(attributes_number['amount']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'amount'))",
+			expectedArgs:          []any{float64(-10.25), float64(-10.25)},
 			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 has(mapValues(attributes_number), ?)) 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), float64(123), "%\"response\"%\"body\"%\"data\"%\"items\"%\"id\"%"},
 			expectedErrorContains: "",
 		},
 		{
@@ -2083,29 +2083,29 @@ func TestFilterExprLogs(t *testing.T) {
 			category:      "Operator precedence",
 			query:         "NOT status=200 AND service.name=\"api\"",
 			shouldPass:    true,
-			expectedQuery: "WHERE (NOT ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status'))) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
-			expectedArgs:  []any{float64(200), "api"}, // Should be (NOT status=200) AND service.name="api"
+			expectedQuery: "WHERE (NOT (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status'))) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
+			expectedArgs:  []any{float64(200), float64(200), "api"}, // Should be (NOT status=200) AND service.name="api"
 		},
 		{
 			category:      "Operator precedence",
 			query:         "status=200 AND service.name=\"api\" OR service.name=\"web\"",
 			shouldPass:    true,
-			expectedQuery: "WHERE (((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)) OR (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
-			expectedArgs:  []any{float64(200), "api", "web"}, // Should be (status=200 AND service.name="api") OR service.name="web"
+			expectedQuery: "WHERE ((((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)) OR (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
+			expectedArgs:  []any{float64(200), float64(200), "api", "web"}, // Should be (status=200 AND service.name="api") OR service.name="web"
 		},
 		{
 			category:      "Operator precedence",
 			query:         "NOT status=200 OR NOT service.name=\"api\"",
 			shouldPass:    true,
-			expectedQuery: "WHERE (NOT ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status'))) OR NOT ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))",
-			expectedArgs:  []any{float64(200), "api"}, // Should be (NOT status=200) OR (NOT service.name="api")
+			expectedQuery: "WHERE (NOT (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status'))) OR NOT ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))",
+			expectedArgs:  []any{float64(200), float64(200), "api"}, // Should be (NOT status=200) OR (NOT service.name="api")
 		},
 		{
 			category:      "Operator precedence",
 			query:         "status=200 OR service.name=\"api\" AND level=\"ERROR\"",
 			shouldPass:    true,
-			expectedQuery: "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) OR ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) AND (attributes_string['level'] = ? AND mapContains(attributes_string, 'level'))))",
-			expectedArgs:  []any{float64(200), "api", "ERROR"}, // Should be status=200 OR (service.name="api" AND level="ERROR")
+			expectedQuery: "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) OR ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) AND (attributes_string['level'] = ? AND mapContains(attributes_string, 'level'))))",
+			expectedArgs:  []any{float64(200), float64(200), "api", "ERROR"}, // Should be status=200 OR (service.name="api" AND level="ERROR")
 		},
 
 		// Different whitespace patterns
@@ -2129,8 +2129,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:      "Whitespace patterns",
 			query:         "status=200  AND     service.name=\"api\"",
 			shouldPass:    true,
-			expectedQuery: "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
-			expectedArgs:  []any{float64(200), "api"}, // Multiple spaces
+			expectedQuery: "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
+			expectedArgs:  []any{float64(200), float64(200), "api"}, // Multiple spaces
 		},
 
 		// More Unicode characters
@@ -2365,8 +2365,8 @@ func TestFilterExprLogs(t *testing.T) {
 			category:      "Unusual whitespace",
 			query:         "status   =    200    AND   service.name    =   \"api\"",
 			shouldPass:    true,
-			expectedQuery: "WHERE ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
-			expectedArgs:  []any{float64(200), "api"},
+			expectedQuery: "WHERE (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status')) AND (multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL))",
+			expectedArgs:  []any{float64(200), float64(200), "api"},
 		},
 		{
 			category:              "Unusual whitespace",
@@ -2426,9 +2426,9 @@ func TestFilterExprLogs(t *testing.T) {
 		  )
 		`,
 			shouldPass:    true,
-			expectedQuery: "WHERE ((((((((toFloat64(attributes_number['status']) >= ? AND mapContains(attributes_number, 'status')) AND (toFloat64(attributes_number['status']) < ? AND mapContains(attributes_number, 'status')))) OR (((toFloat64(attributes_number['status']) >= ? AND mapContains(attributes_number, 'status')) AND (toFloat64(attributes_number['status']) < ? AND mapContains(attributes_number, 'status')) AND NOT ((toFloat64(attributes_number['status']) = ? AND mapContains(attributes_number, 'status'))))))) AND ((((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? OR multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? OR multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ?) AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR (((multiIf(resource.`service.type` IS NOT NULL, resource.`service.type`::String, mapContains(resources_string, 'service.type'), resources_string['service.type'], NULL) = ? AND multiIf(resource.`service.type` IS NOT NULL, resource.`service.type`::String, mapContains(resources_string, 'service.type'), resources_string['service.type'], NULL) IS NOT NULL) AND NOT ((multiIf(resource.`service.deprecated` IS NOT NULL, resource.`service.deprecated`::String, mapContains(resources_string, 'service.deprecated'), resources_string['service.deprecated'], NULL) = ? AND multiIf(resource.`service.deprecated` IS NOT NULL, resource.`service.deprecated`::String, mapContains(resources_string, 'service.deprecated'), resources_string['service.deprecated'], NULL) IS NOT NULL)))))))) AND (((((toFloat64(attributes_number['duration']) < ? AND mapContains(attributes_number, 'duration')) OR ((toFloat64(attributes_number['duration']) BETWEEN ? AND ? AND mapContains(attributes_number, 'duration'))))) AND ((multiIf(resource.`environment` IS NOT NULL, resource.`environment`::String, mapContains(resources_string, 'environment'), resources_string['environment'], NULL) <> ? OR (((multiIf(resource.`environment` IS NOT NULL, resource.`environment`::String, mapContains(resources_string, 'environment'), resources_string['environment'], NULL) = ? AND multiIf(resource.`environment` IS NOT NULL, resource.`environment`::String, mapContains(resources_string, 'environment'), resources_string['environment'], NULL) IS NOT NULL) AND (attributes_bool['is_automated_test'] = ? AND mapContains(attributes_bool, 'is_automated_test')))))))) AND NOT ((((((LOWER(attributes_string['message']) LIKE LOWER(?) AND mapContains(attributes_string, 'message')) OR (LOWER(attributes_string['message']) LIKE LOWER(?) AND mapContains(attributes_string, 'message')))) AND (attributes_string['severity'] = ? AND mapContains(attributes_string, 'severity'))))))",
+			expectedQuery: "WHERE ((((((((toFloat64(attributes_number['status']) >= ? AND mapContains(attributes_number, 'status')) AND (toFloat64(attributes_number['status']) < ? AND mapContains(attributes_number, 'status')))) OR (((toFloat64(attributes_number['status']) >= ? AND mapContains(attributes_number, 'status')) AND (toFloat64(attributes_number['status']) < ? AND mapContains(attributes_number, 'status')) AND NOT (((toFloat64(attributes_number['status']) = ? AND has(mapValues(attributes_number), ?)) AND mapContains(attributes_number, 'status'))))))) AND ((((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? OR multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? OR multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ?) AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR (((multiIf(resource.`service.type` IS NOT NULL, resource.`service.type`::String, mapContains(resources_string, 'service.type'), resources_string['service.type'], NULL) = ? AND multiIf(resource.`service.type` IS NOT NULL, resource.`service.type`::String, mapContains(resources_string, 'service.type'), resources_string['service.type'], NULL) IS NOT NULL) AND NOT ((multiIf(resource.`service.deprecated` IS NOT NULL, resource.`service.deprecated`::String, mapContains(resources_string, 'service.deprecated'), resources_string['service.deprecated'], NULL) = ? AND multiIf(resource.`service.deprecated` IS NOT NULL, resource.`service.deprecated`::String, mapContains(resources_string, 'service.deprecated'), resources_string['service.deprecated'], NULL) IS NOT NULL)))))))) AND (((((toFloat64(attributes_number['duration']) < ? AND mapContains(attributes_number, 'duration')) OR ((toFloat64(attributes_number['duration']) BETWEEN ? AND ? AND mapContains(attributes_number, 'duration'))))) AND ((multiIf(resource.`environment` IS NOT NULL, resource.`environment`::String, mapContains(resources_string, 'environment'), resources_string['environment'], NULL) <> ? OR (((multiIf(resource.`environment` IS NOT NULL, resource.`environment`::String, mapContains(resources_string, 'environment'), resources_string['environment'], NULL) = ? AND multiIf(resource.`environment` IS NOT NULL, resource.`environment`::String, mapContains(resources_string, 'environment'), resources_string['environment'], NULL) IS NOT NULL) AND (attributes_bool['is_automated_test'] = ? AND mapContains(attributes_bool, 'is_automated_test')))))))) AND NOT ((((((LOWER(attributes_string['message']) LIKE LOWER(?) AND mapContains(attributes_string, 'message')) OR (LOWER(attributes_string['message']) LIKE LOWER(?) AND mapContains(attributes_string, 'message')))) AND (attributes_string['severity'] = ? AND mapContains(attributes_string, 'severity'))))))",
 			expectedArgs: []any{
-				float64(200), float64(300), float64(400), float64(500), float64(404),
+				float64(200), float64(300), float64(400), float64(500), float64(404), float64(404),
 				"api", "web", "auth",
 				"internal", true,
 				float64(1000), float64(1000), float64(5000),
@@ -2521,7 +2521,7 @@ func TestFilterExprLogsConflictNegation(t *testing.T) {
 			query:      "body NOT LIKE 'done'",
 			shouldPass: true,
 			// lower index search on body even for LIKE
-			expectedQuery:         "WHERE (LOWER(body) NOT LIKE LOWER(?) AND attributes_string['body'] NOT LIKE ?)",
+			expectedQuery:         "WHERE (body NOT LIKE ? AND attributes_string['body'] NOT LIKE ?)",
 			expectedArgs:          []any{"done", "done"},
 			expectedErrorContains: "",
 		},

pkg/telemetryschema/logstelemetryschema/json_condition_builder.go

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

pkg/telemetryschema/logstelemetryschema/json_string.go

@@ -9,6 +9,8 @@ import (
 
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+
+	"github.com/huandu/go-sqlbuilder"
 )
 
 func parseStrValue(valueStr string, operator qbtypes.FilterOperator) (telemetrytypes.FieldDataType, any) {
@@ -92,6 +94,190 @@ func InferDataType(value any, operator qbtypes.FilterOperator, key *telemetrytyp
 	return closure(value, key)
 }
 
+// likePatternLiterals returns the runs of pattern between unescaped wildcards, with `\` escapes
+// resolved; every value the pattern matches holds each run verbatim. ClickHouse treats `\` as an
+// escape only before `%`, `_` and itself, so dropping it elsewhere would yield a run it never requires.
+func likePatternLiterals(pattern string) []string {
+	var (
+		literals []string
+		run      strings.Builder
+	)
+	for i := 0; i < len(pattern); i++ {
+		switch c := pattern[i]; c {
+		case '%', '_':
+			if run.Len() > 0 {
+				literals = append(literals, run.String())
+				run.Reset()
+			}
+		case '\\':
+			if i+1 >= len(pattern) {
+				run.WriteByte('\\')
+				continue
+			}
+			i++
+			if escaped := pattern[i]; escaped != '%' && escaped != '_' && escaped != '\\' {
+				run.WriteByte('\\')
+			}
+			run.WriteByte(pattern[i])
+		default:
+			run.WriteByte(c)
+		}
+	}
+	if run.Len() > 0 {
+		literals = append(literals, run.String())
+	}
+	return literals
+}
+
+// jsonEscapable reports whether a JSON encoder is free to rewrite r: `"` and `\` always, `/` by
+// PHP, `<` `>` `&` by Go, non-printable ASCII by Python's ensure_ascii. The legacy body holds the
+// producer's own text, so a literal spanning one of these may not be there to find.
+func jsonEscapable(r rune) bool {
+	return r < 0x20 || r > 0x7e || strings.ContainsRune(`"\/<>&`, r)
+}
+
+// jsonTextRuns splits s at every byte an encoder may rewrite. A body whose JSON holds s contains
+// each returned run verbatim, in order.
+func jsonTextRuns(s string) []string {
+	return strings.FieldsFunc(s, jsonEscapable)
+}
+
+// bodyPathLiterals returns one literal per component of key's JSON path, quoted the way JSON writes
+// an object key. A component holding a byte an encoder may rewrite is dropped; JSON writes a parent
+// first, so the order carries.
+func bodyPathLiterals(key *telemetrytypes.TelemetryFieldKey) []string {
+	var literals []string
+	for _, part := range strings.Split(key.Name, ".") {
+		if idx := strings.Index(part, "["); idx >= 0 {
+			part = part[:idx]
+		}
+		if literal := `"` + part + `"`; !strings.ContainsFunc(part, jsonEscapable) {
+			literals = append(literals, literal)
+		}
+	}
+	return literals
+}
+
+// bodyValueLiterals returns the literals a comparison implies in the body text. Only string
+// comparisons qualify: a number is compared after JSONExtract parses it, which reads 1.23e2
+// as 123, so the digits of the filter value need not appear in the body at all.
+func bodyValueLiterals(operator qbtypes.FilterOperator, value any) []string {
+	str, ok := value.(string)
+	if !ok {
+		return nil
+	}
+	switch operator {
+	case qbtypes.FilterOperatorEqual, qbtypes.FilterOperatorContains:
+		return jsonTextRuns(str)
+	case qbtypes.FilterOperatorLike, qbtypes.FilterOperatorILike:
+		var literals []string
+		for _, literal := range likePatternLiterals(str) {
+			literals = append(literals, jsonTextRuns(literal)...)
+		}
+		return literals
+	}
+	return nil
+}
+
+// escapeLikeLiteral escapes the LIKE metacharacters so s matches as literal text. Backslash
+// goes first, being the escape character itself.
+func escapeLikeLiteral(s string) string {
+	s = strings.ReplaceAll(s, `\`, `\\`)
+	s = strings.ReplaceAll(s, "%", `\%`)
+	return strings.ReplaceAll(s, "_", `\_`)
+}
+
+// bodyIndexPredicate asserts the raw body text holds the literals in order. ILike renders as
+// LOWER(body) LIKE LOWER(?) on the ClickHouse flavor — the expression both bloom filters index.
+// They are plain text and backslash-free by construction, so only the LIKE wildcards need escaping.
+func bodyIndexPredicate(literals []string, sb *sqlbuilder.SelectBuilder) string {
+	if len(literals) == 0 {
+		return ""
+	}
+	escaped := make([]string, 0, len(literals))
+	for _, literal := range literals {
+		escaped = append(escaped, escapeLikeLiteral(literal))
+	}
+	pattern := "%" + strings.Join(escaped, "%") + "%"
+	return sb.ILike(LogsV2BodyColumn, pattern)
+}
+
+// legacyBodyIndexPredicates returns what a legacy body JSON filter implies over LOWER(body),
+// which nothing it compares matches. Path literals ride on the existence assertion and value
+// literals on the comparison, so each appears once; a negation carries none, matching rows that
+// lack the path entirely.
+func legacyBodyIndexPredicates(key *telemetrytypes.TelemetryFieldKey, operator qbtypes.FilterOperator, value any, sb *sqlbuilder.SelectBuilder) []string {
+	var predicates []string
+	if operator == qbtypes.FilterOperatorExists || operator.IsArrayFunctionOperator() {
+		if predicate := bodyIndexPredicate(bodyPathLiterals(key), sb); predicate != "" {
+			predicates = append(predicates, predicate)
+		}
+	}
+	if operator.IsArrayFunctionOperator() {
+		return append(predicates, bodyArrayFunctionPredicates(operator, value, sb)...)
+	}
+	if predicate := bodyIndexPredicate(bodyValueLiterals(operator, value), sb); predicate != "" {
+		predicates = append(predicates, predicate)
+	}
+	return predicates
+}
+
+// bodyArrayFunctionPredicates returns what a has-family filter implies about the body text. has
+// and hasAll require every element, so each becomes its own predicate; hasAny requires one, so its
+// arms are ORed, and an element yielding no literal leaves that OR unassertable.
+func bodyArrayFunctionPredicates(operator qbtypes.FilterOperator, value any, sb *sqlbuilder.SelectBuilder) []string {
+	element := value
+	if args, ok := value.([]any); ok && len(args) > 0 {
+		element = args[0]
+	}
+	// the has 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 — same reason `=` skips them
+	if legacyElemType(element) != telemetrytypes.FieldDataTypeString {
+		return nil
+	}
+
+	values, ok := element.([]any)
+	if !ok {
+		values = []any{element}
+	}
+
+	if operator == qbtypes.FilterOperatorHasAny {
+		// resolve every value before binding anything: one unusable value voids the whole OR
+		runSets := make([][]string, 0, len(values))
+		for _, v := range values {
+			str, ok := v.(string)
+			if !ok {
+				return nil
+			}
+			runs := jsonTextRuns(str)
+			if len(runs) == 0 {
+				return nil
+			}
+			runSets = append(runSets, runs)
+		}
+		if len(runSets) == 0 {
+			return nil
+		}
+		arms := make([]string, 0, len(runSets))
+		for _, runs := range runSets {
+			arms = append(arms, bodyIndexPredicate(runs, sb))
+		}
+		return []string{sb.Or(arms...)}
+	}
+
+	var predicates []string
+	for _, v := range values {
+		str, ok := v.(string)
+		if !ok {
+			continue
+		}
+		if predicate := bodyIndexPredicate(jsonTextRuns(str), sb); predicate != "" {
+			predicates = append(predicates, predicate)
+		}
+	}
+	return predicates
+}
+
 func getBodyJSONPath(key *telemetrytypes.TelemetryFieldKey) string {
 	parts := strings.Split(key.Name, ".")
 	newParts := []string{}
@@ -139,14 +325,14 @@ func GetBodyJSONKeyForExists(_ context.Context, key *telemetrytypes.TelemetryFie
 	return fmt.Sprintf("JSON_EXISTS(body, '$.%s')", getBodyJSONPath(key))
 }
 
-// legacyElemType infers the has-family element type from the needle (legacy has no schema). It
-// scans EVERY value so the chosen array type and all coerced needles agree — else ClickHouse
+// legacyElemType infers the has-family element type from the arg (legacy has no schema). It
+// scans EVERY value so the chosen array type and all coerced args agree — else ClickHouse
 // raises "no supertype ... String" (code 386). Int64 stays distinct from Float64 so a quoted
 // integer is exact past 2^53 (unquoted literals already arrive as float64, parsed upstream).
-func legacyElemType(needle any) telemetrytypes.FieldDataType {
-	list, ok := needle.([]any)
+func legacyElemType(arg any) telemetrytypes.FieldDataType {
+	list, ok := arg.([]any)
 	if !ok {
-		list = []any{needle}
+		list = []any{arg}
 	}
 	if len(list) == 0 {
 		return telemetrytypes.FieldDataTypeString
@@ -167,7 +353,7 @@ func legacyElemType(needle any) telemetrytypes.FieldDataType {
 			}
 		default:
 			// booleans (and anything else) -> String; a bool renders to 'true'/'false', so a
-			// bool needle only matches genuine JSON booleans, not truthy numbers/strings.
+			// bool arg only matches genuine JSON booleans, not truthy numbers/strings.
 			allInt, allNumeric = false, false
 		}
 	}
@@ -181,9 +367,9 @@ func legacyElemType(needle any) telemetrytypes.FieldDataType {
 	}
 }
 
-// legacyCoerceNeedle coerces a needle to elem type dt so its bound-arg type matches the
+// legacyCoerceElement coerces an element to elem type dt so its bound-arg type matches the
 // extracted column (legacyElemType guarantees it's coercible).
-func legacyCoerceNeedle(v any, dt telemetrytypes.FieldDataType) any {
+func legacyCoerceElement(v any, dt telemetrytypes.FieldDataType) any {
 	switch dt {
 	case telemetrytypes.FieldDataTypeInt64:
 		if s, ok := v.(string); ok {
@@ -199,7 +385,7 @@ func legacyCoerceNeedle(v any, dt telemetrytypes.FieldDataType) any {
 		}
 		return v
 	default:
-		return bodyArrayNeedleString(v)
+		return bodyArrayElementString(v)
 	}
 }
 
@@ -242,7 +428,7 @@ func getBodyJSONScalarKey(key *telemetrytypes.TelemetryFieldKey, dt telemetrytyp
 	return expr, guard, true
 }
 
-func bodyArrayNeedleString(v any) string {
+func bodyArrayElementString(v any) string {
 	switch t := v.(type) {
 	case string:
 		return t

tests/fixtures/querier.py

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

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

@@ -0,0 +1,90 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logs import Logs
+from fixtures.querier import build_order_by, build_raw_query, get_rows, make_query_request
+
+LOWER = "alpha"
+UPPER = "ALPHA"
+PLAIN = "beta"
+NON_ASCII = "Mixed CASE Ünïcode"
+SLASH = "GET /api/v1/users"
+SUPERSTRING = "GET /api/v1/users/42"
+QUOTE = 'say "hi" now'
+BACKSLASH = "C:\\tmp\\log"
+LIKE_META = "100% _off"
+TAB = "tab\there"
+CTRL = "ctrl\x01here"
+
+BODIES = [LOWER, UPPER, PLAIN, NON_ASCII, SLASH, SUPERSTRING, QUOTE, BACKSLASH, LIKE_META, TAB, CTRL]
+
+
+# querierlogs/16_body_equality.py with use_json_body on: `body` resolves to body_v2.message,
+# which the lower(body) companion skips, and the same expressions must still answer alike.
+@pytest.mark.parametrize(
+    "expression,expected_bodies",
+    [
+        pytest.param(f"body = '{LOWER}'", {LOWER}, id="equality_exact"),
+        pytest.param(f"body = '{UPPER}'", {UPPER}, id="equality_other_case"),
+        pytest.param("body = 'Alpha'", set(), id="equality_case_must_match"),
+        pytest.param(f"body = '{NON_ASCII}'", {NON_ASCII}, id="equality_non_ascii"),
+        pytest.param("body = 'gamma'", set(), id="equality_no_match"),
+        pytest.param(f"body = '{SLASH}'", {SLASH}, id="equality_slash"),
+        pytest.param("body = 'say \"hi\" now'", {QUOTE}, id="equality_quote"),
+        pytest.param(r"body = 'C:\\tmp\\log'", {BACKSLASH}, id="equality_backslash"),
+        pytest.param(f"body = '{LIKE_META}'", {LIKE_META}, id="equality_like_metacharacters"),
+        pytest.param("body = 'tab\there'", {TAB}, id="equality_tab"),
+        pytest.param("body = 'ctrl\x01here'", {CTRL}, id="equality_control_char"),
+        pytest.param("body = 'GET /api/v1'", set(), id="equality_prefix_does_not_match"),
+        pytest.param(f"body IN ('{LOWER}', '{PLAIN}')", {LOWER, PLAIN}, id="in_excludes_other_case"),
+        pytest.param(f"body IN ('{SLASH}', '{LIKE_META}')", {SLASH, LIKE_META}, id="in_escaped_values"),
+        pytest.param(f"body NOT IN ('{LOWER}', '{UPPER}')", set(BODIES) - {LOWER, UPPER}, id="not_in"),
+    ],
+)
+def test_logs_body_equality_json(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    expression: str,
+    expected_bodies: set[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": "api"},
+                body=body,
+            )
+            for i, body in enumerate(BODIES)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[
+            build_raw_query(
+                "A",
+                "logs",
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "desc"), build_order_by("id", "desc")],
+                limit=100,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["status"] == "success"
+    # body_v2 comes back parsed; a plain-string body is {"message": <body>}.
+    assert {row["data"]["body"]["message"] for row in get_rows(response)} == expected_bodies

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

@@ -0,0 +1,92 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logs import Logs
+from fixtures.querier import build_order_by, build_raw_query, get_column_data_from_response, make_query_request
+
+LOWER = "alpha"
+UPPER = "ALPHA"
+PLAIN = "beta"
+NON_ASCII = "Mixed CASE Ünïcode"
+SLASH = "GET /api/v1/users"
+SUPERSTRING = "GET /api/v1/users/42"
+QUOTE = 'say "hi" now'
+BACKSLASH = "C:\\tmp\\log"
+LIKE_META = "100% _off"
+TAB = "tab\there"
+CTRL = "ctrl\x01here"
+
+BODIES = [LOWER, UPPER, PLAIN, NON_ASCII, SLASH, SUPERSTRING, QUOTE, BACKSLASH, LIKE_META, TAB, CTRL]
+
+
+# `body = ?` carries a case-insensitive LOWER(body) companion for the bloom filters, so a
+# body differing only in case must still not come back.
+@pytest.mark.parametrize(
+    "expression,expected_bodies",
+    [
+        pytest.param(f"body = '{LOWER}'", {LOWER}, id="equality_exact"),
+        pytest.param(f"body = '{UPPER}'", {UPPER}, id="equality_other_case"),
+        pytest.param("body = 'Alpha'", set(), id="equality_case_must_match"),
+        pytest.param(f"body = '{NON_ASCII}'", {NON_ASCII}, id="equality_non_ascii"),
+        pytest.param("body = ''", set(), id="equality_empty"),
+        pytest.param("body = 'gamma'", set(), id="equality_no_match"),
+        # the companion is a LIKE-free equality, so none of these are metacharacters to it
+        pytest.param(f"body = '{SLASH}'", {SLASH}, id="equality_slash"),
+        pytest.param("body = 'say \"hi\" now'", {QUOTE}, id="equality_quote"),
+        pytest.param(r"body = 'C:\\tmp\\log'", {BACKSLASH}, id="equality_backslash"),
+        pytest.param(f"body = '{LIKE_META}'", {LIKE_META}, id="equality_like_metacharacters"),
+        pytest.param("body = 'tab\there'", {TAB}, id="equality_tab"),
+        pytest.param("body = 'ctrl\x01here'", {CTRL}, id="equality_control_char"),
+        # a prefix of another body must not match it
+        pytest.param("body = 'GET /api/v1'", set(), id="equality_prefix_does_not_match"),
+        pytest.param(f"body IN ('{LOWER}', '{PLAIN}')", {LOWER, PLAIN}, id="in_excludes_other_case"),
+        pytest.param(f"body IN ('{SLASH}', '{LIKE_META}')", {SLASH, LIKE_META}, id="in_escaped_values"),
+        pytest.param(f"body NOT IN ('{LOWER}', '{UPPER}')", set(BODIES) - {LOWER, UPPER}, id="not_in"),
+    ],
+)
+def test_logs_body_equality(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    expression: str,
+    expected_bodies: set[str],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=i + 1),
+                resources={"service.name": "api"},
+                body=body,
+            )
+            for i, body in enumerate(BODIES)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[
+            build_raw_query(
+                "A",
+                "logs",
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "desc"), build_order_by("id", "desc")],
+                limit=100,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["status"] == "success"
+    assert set(get_column_data_from_response(response.json(), "body")) == expected_bodies

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

@@ -0,0 +1,346 @@
+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"
+
+
+# The attribute maps carry a bloom filter over mapValues, which the subscript the comparison reads
+# matches no more than the body column did. mapContains prunes on the key alone, so these use a key
+# every row carries to isolate what the value predicate contributes.
+ATTRIBUTE_NUMBER_VALUE_INDEX = "attributes_number_idx_val"
+ATTRIBUTE_STRING_VALUE_INDEX = "attributes_string_idx_val"
+
+
+@pytest.mark.parametrize(
+    "expression,prunes_every_granule",
+    [
+        pytest.param("attribute.resp_code = 503", False, id="value_present"),
+        pytest.param("attribute.resp_code = 60599", True, id="value_absent"),
+        # IN delegates to the equalities, so every arm carries its own membership assertion
+        pytest.param("attribute.resp_code IN (503, 200)", False, id="in_present_values"),
+        pytest.param("attribute.resp_code IN (60599, 60600)", True, id="in_absent_values"),
+    ],
+)
+def test_attribute_number_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"}, attributes={"resp_code": code}) for i, code in enumerate([200, 200, 503])])
+
+    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 ATTRIBUTE_NUMBER_VALUE_INDEX in skip_indexes, f"mapValues filter 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"
+
+
+# The mapValues filter indexes the values raw, so a case-insensitive match reaches it only for a
+# pattern holding no ASCII letter, where LOWER changes nothing. A letter leaves it unconsulted.
+@pytest.mark.parametrize(
+    "expression,value_index_consulted,prunes_every_granule",
+    [
+        pytest.param("attribute.client.ip CONTAINS '192.168.77'", True, False, id="letter_free_value_present"),
+        pytest.param("attribute.client.ip CONTAINS '192.168.99'", True, True, id="letter_free_value_absent"),
+        pytest.param("attribute.env CONTAINS 'production'", False, False, id="letters_leave_it_unconsulted"),
+        # the filter is consulted for any letter-free pattern, but a run below the index ngram
+        # leaves it nothing to check
+        pytest.param("attribute.client.ip CONTAINS '.7'", True, False, id="run_shorter_than_the_ngram"),
+    ],
+)
+def test_attribute_letter_free_match_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,
+    value_index_consulted: bool,
+    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"},
+                attributes={"client.ip": ip, "env": "production"},
+            )
+            for i, ip in enumerate(["10.0.0.1", "10.0.0.2", "192.168.77.31"])
+        ]
+    )
+
+    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 (ATTRIBUTE_STRING_VALUE_INDEX in skip_indexes) == value_index_consulted, f"consulted: {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"