test(querier): pin explicit context resolution under ambiguous names

One name can exist in more than one place. `name` is a span column and a
span attribute. `severity_text` is a log column and a log attribute. An
attribute can have two data types. `service.name` is a resource attribute
and a span or log attribute. These tests record what the query builder
does for each shape in a filter, EXISTS, a group by, an order by, an
aggregation argument, and a raw select. A change to name resolution then
fails here first.

- A key with an explicit context reads that context only. A bare key with
  more than one reading returns an ambiguity warning. An `attribute.` key
  returns the warning when the attribute has two data types.
- A bare key that is a column and an attribute reads both in a filter. It
  orders, groups, and counts by the column only. A string operand matches
  a number attribute through a text cast.
- A bare key that is a resource attribute and an attribute reads the
  resource attribute in a filter and in a raw select. The filter returns a
  warning.
- An `attribute.` key in an order by sorts by the attribute on traces and
  by the column on logs.
- A key under the signal's own context that exists only as an attribute
  reads the attribute. On logs it also reads the body JSON path. A
  `scope.` key on logs resolves through metadata only. When metadata does
  not report the key, the query fails with "key not found". This is also
  true for the declared path `scope.name`.

Assisted-by: Claude Fable 5.1
Claude-Session: https://claude.ai/code/session_01RSnZFLSfyi5S4QYQDcxHeW

tests/fixtures/queriercommon.py

@@ -1,4 +1,4 @@
-"""Seed data for the queriercommon keyless-semantics tests.
+"""Seed data for the queriercommon keyless-semantics and explicit-context tests.
 
 Three identities exist in every signal. GOLD and SILVER carry the test keys.
 NONE carries no key at all. The tests assert which identities a filter
@@ -8,6 +8,7 @@ The attribute names are outside every semantic-convention family, so the
 seeded data pins base behavior with any semconv overlay state.
 """
 
+import json
 from collections.abc import Callable, Generator
 from datetime import UTC, datetime, timedelta
 
@@ -122,3 +123,103 @@ def keyless_series(insert_metrics: Callable[[list[Metrics]], None]) -> Generator
         ]
     )
     yield start, start + points * 60
+
+
+EXPLICIT_PREFIX = "explicit-ctx"
+# String attribute that identifies the row. It has one context only. Each
+# assertion reads it back.
+IDENTITY_KEY = "probe.id"
+# Attribute with no column of the same name. It tests a key under the
+# signal's own context that metadata does not know. On logs, the rows without
+# the attribute have the value nested in the body JSON.
+ATTRIBUTE_ONLY_KEY = "route.tag"
+CONTESTED_VALUE = "checkout"
+
+# Row identities. Each row shows where the contested value is:
+# - COLUMN_ONLY: in the column (`name` on spans, `severity_text` on logs).
+# - ATTRIBUTE_ONLY: in the string attribute with the same name.
+# - BOTH: in the column and in the string attribute.
+# - NEITHER: in none of them.
+# - NUMBER_ATTRIBUTE: in a number attribute with the same name. Its data
+#   type is different from the column.
+COLUMN_ONLY = f"{EXPLICIT_PREFIX}-column"
+ATTRIBUTE_ONLY = f"{EXPLICIT_PREFIX}-attribute"
+BOTH = f"{EXPLICIT_PREFIX}-both"
+NEITHER = f"{EXPLICIT_PREFIX}-neither"
+NUMBER_ATTRIBUTE = f"{EXPLICIT_PREFIX}-number"
+NUMBER_VALUE = 42
+
+# (identity, value in the column, value in the string attribute, value in
+#  the number attribute, resource service.name, attribute service.name,
+#  has route.tag, insert offset in seconds)
+ROWS = [
+    (COLUMN_ONLY, True, False, False, "svc-a", None, True, 1),
+    (ATTRIBUTE_ONLY, False, True, False, "svc-b", "svc-a", False, 2),
+    (BOTH, True, True, False, "svc-a", "svc-a", True, 3),
+    (NEITHER, False, False, False, "svc-b", "svc-b", False, 4),
+    (NUMBER_ATTRIBUTE, False, False, True, "svc-b", None, False, 5),
+]
+
+# Logs only. The scope name is a declared path. A scope attribute also has
+# the name `name`. A second scope attribute has a plain name.
+SCOPE_NAME = "scope-a"
+SCOPE_ATTRIBUTE_KEY = "env"
+SCOPE_ATTRIBUTE_VALUE = "prod"
+
+
+@pytest.fixture(name="ambiguous_rows", scope="function")
+def ambiguous_rows(
+    insert_logs: Callable[[list[Logs]], None],
+    insert_traces: Callable[[list[Traces]], None],
+) -> Generator[datetime]:
+    """Inserts one span and one log for each identity. Every row has a
+    resource `service.name`. Some rows also have a span or log attribute
+    `service.name` with a different value. On logs, the rows without the
+    `route.tag` attribute have the value in the body JSON. Logs with the
+    column value have the scope name. Logs with the attribute value have the
+    scope attributes. Yields the base timestamp."""
+    now = datetime.now(tz=UTC).replace(microsecond=0) - timedelta(minutes=1)
+
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(seconds=offset),
+                duration=timedelta(milliseconds=10),
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                name=CONTESTED_VALUE if column else "other",
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                resources={"service.name": resource_service},
+                attributes={
+                    IDENTITY_KEY: identity,
+                    **({"name": CONTESTED_VALUE} if attribute else {}),
+                    **({"name": NUMBER_VALUE} if number else {}),
+                    **({"service.name": attribute_service} if attribute_service else {}),
+                    **({ATTRIBUTE_ONLY_KEY: CONTESTED_VALUE} if tagged else {}),
+                },
+            )
+            for identity, column, attribute, number, resource_service, attribute_service, tagged, offset in ROWS
+        ]
+    )
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=offset),
+                body=json.dumps({} if tagged else {"route": {"tag": CONTESTED_VALUE}}),
+                severity_text="ERROR" if column else "INFO",
+                scope_name=SCOPE_NAME if column else "",
+                scope_attributes={"name": CONTESTED_VALUE, SCOPE_ATTRIBUTE_KEY: SCOPE_ATTRIBUTE_VALUE} if attribute else {},
+                resources={"service.name": resource_service},
+                attributes={
+                    IDENTITY_KEY: identity,
+                    **({"severity_text": "ERROR"} if attribute else {}),
+                    **({"severity_text": NUMBER_VALUE} if number else {}),
+                    **({"service.name": attribute_service} if attribute_service else {}),
+                    **({ATTRIBUTE_ONLY_KEY: CONTESTED_VALUE} if tagged else {}),
+                },
+            )
+            for identity, column, attribute, number, resource_service, attribute_service, tagged, offset in ROWS
+        ]
+    )
+    yield now

tests/integration/tests/queriercommon/07_explicit_context.py

@@ -0,0 +1,381 @@
+from collections.abc import Callable
+from datetime import datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import (
+    RequestType,
+    assert_scalar_value,
+    build_aggregation,
+    build_group_by_field,
+    build_order_by,
+    build_raw_query,
+    build_scalar_query,
+    get_all_warnings,
+    get_column_data_from_response,
+    get_scalar_table_data,
+    make_query_request,
+)
+from fixtures.queriercommon import (
+    ATTRIBUTE_ONLY,
+    BOTH,
+    COLUMN_ONLY,
+    EXPLICIT_PREFIX,
+    IDENTITY_KEY,
+    NEITHER,
+    NUMBER_ATTRIBUTE,
+)
+
+# One name can exist in more than one place. `name` is a span column and a
+# span attribute. `severity_text` is a log column and a log attribute.
+# `service.name` is a resource attribute and a span or log attribute.
+#
+# Rules for a filter:
+# - A key with an explicit context reads that context only.
+# - A bare key that is a column and an attribute reads both. The query
+#   returns an ambiguity warning.
+# - A bare key that is a resource attribute and an attribute reads the
+#   resource attribute. The query returns an ambiguity warning.
+# - An `attribute.` key returns the warning when the attribute has two data
+#   types.
+# - A string operand matches a number attribute through a text cast.
+# - A key under the signal's own context (`span.`, `log.`) that exists only
+#   as an attribute reads the attribute. On logs it also reads the body JSON
+#   path.
+FILTER_MATRIX = [
+    pytest.param("{contested} = '{value}'", {COLUMN_ONLY, ATTRIBUTE_ONLY, BOTH}, True, id="bare_column_and_attribute"),
+    pytest.param("{own}.{contested} = '{value}'", {COLUMN_ONLY, BOTH}, False, id="own_context_column_only"),
+    pytest.param("attribute.{contested} = '{value}'", {ATTRIBUTE_ONLY, BOTH}, True, id="attribute_context_warns_about_two_types"),
+    pytest.param("{contested} != '{value}'", {NEITHER, NUMBER_ATTRIBUTE}, True, id="bare_negative_excludes_every_carrier"),
+    pytest.param("{contested} EXISTS", {COLUMN_ONLY, ATTRIBUTE_ONLY, BOTH, NEITHER, NUMBER_ATTRIBUTE}, True, id="bare_exists_is_the_column"),
+    pytest.param("{contested} NOT EXISTS", set(), True, id="bare_not_exists_is_never"),
+    pytest.param("attribute.{contested} EXISTS", {ATTRIBUTE_ONLY, BOTH, NUMBER_ATTRIBUTE}, True, id="attribute_exists_spans_both_types"),
+    pytest.param("attribute.{contested} NOT EXISTS", {COLUMN_ONLY, NEITHER}, True, id="attribute_not_exists"),
+    pytest.param("{contested} = '42'", {NUMBER_ATTRIBUTE}, True, id="bare_string_operand_reaches_the_number_attribute"),
+    pytest.param("attribute.{contested}:string = '{value}'", {ATTRIBUTE_ONLY, BOTH}, False, id="type_suffix_selects_the_string_attribute"),
+    pytest.param("attribute.{contested}:float64 = 42", {NUMBER_ATTRIBUTE}, False, id="type_suffix_selects_the_number_attribute"),
+    pytest.param("service.name = 'svc-a'", {COLUMN_ONLY, BOTH}, True, id="bare_resource_wins_with_warning"),
+    pytest.param("service.name != 'svc-a'", {ATTRIBUTE_ONLY, NEITHER, NUMBER_ATTRIBUTE}, True, id="bare_resource_negative"),
+    pytest.param("resource.service.name = 'svc-a'", {COLUMN_ONLY, BOTH}, False, id="resource_context_no_warning"),
+    pytest.param("resource.service.name != 'svc-a'", {ATTRIBUTE_ONLY, NEITHER, NUMBER_ATTRIBUTE}, False, id="resource_context_negative"),
+    pytest.param("attribute.service.name = 'svc-a'", {ATTRIBUTE_ONLY, BOTH}, False, id="attribute_context_no_warning"),
+    pytest.param(
+        "{own}.route.tag = 'checkout'",
+        {"traces": {COLUMN_ONLY, BOTH}, "logs": {COLUMN_ONLY, ATTRIBUTE_ONLY, BOTH, NEITHER, NUMBER_ATTRIBUTE}},
+        False,
+        id="own_context_miss_corrects_to_attribute_and_on_logs_to_body",
+    ),
+    pytest.param("route.tag = 'checkout'", {COLUMN_ONLY, BOTH}, False, id="bare_attribute_only_key"),
+]
+
+SIGNALS = [
+    pytest.param("traces", "span", "name", "checkout", "other", id="traces"),
+    pytest.param("logs", "log", "severity_text", "ERROR", "INFO", id="logs"),
+]
+
+
+@pytest.mark.parametrize("expression_template,expected,expects_ambiguity_warning", FILTER_MATRIX)
+@pytest.mark.parametrize("signal,own_context,contested,value,other_value", SIGNALS)
+def test_filter_resolution(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    ambiguous_rows: datetime,
+    signal: str,
+    own_context: str,
+    contested: str,
+    value: str,
+    other_value: str,  # pylint: disable=unused-argument
+    expression_template: str,
+    expected: set[str] | dict[str, set[str]],
+    expects_ambiguity_warning: bool,
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(own=own_context, contested=contested, value=value)
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((ambiguous_rows - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((ambiguous_rows + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                signal,
+                limit=100,
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": IDENTITY_KEY}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    matched = {row for row in get_column_data_from_response(response.json(), IDENTITY_KEY) if row.startswith(EXPLICIT_PREFIX)}
+    assert matched == (expected[signal] if isinstance(expected, dict) else expected), expression
+
+    warnings = [w["message"] for w in get_all_warnings(response.json())]
+    assert any("ambiguous" in w for w in warnings) == expects_ambiguity_warning, warnings
+
+
+# Rules for a group by:
+# - A bare key that is a column and an attribute groups by the column only.
+# - A key with an explicit context groups by that context only.
+GROUP_BY_MATRIX = [
+    pytest.param(None, {"{value}": 2, "{other}": 3}, id="bare_groups_by_the_column"),
+    pytest.param("own", {"{value}": 2, "{other}": 3}, id="own_context_groups_by_the_column"),
+    pytest.param("attribute", {"{value}": 2}, id="attribute_context_groups_by_the_attribute"),
+]
+
+
+@pytest.mark.parametrize("context,expected_template", GROUP_BY_MATRIX)
+@pytest.mark.parametrize("signal,own_context,contested,value,other_value", SIGNALS)
+def test_group_by_resolution(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    ambiguous_rows: datetime,
+    signal: str,
+    own_context: str,
+    contested: str,
+    value: str,
+    other_value: str,
+    context: str | None,
+    expected_template: dict[str, int],
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    field_context = own_context if context == "own" else context
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((ambiguous_rows - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((ambiguous_rows + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.SCALAR,
+        queries=[
+            build_scalar_query(
+                "A",
+                signal,
+                [build_aggregation("count()", "rows")],
+                group_by=[build_group_by_field(contested, "string", field_context) if field_context else {"name": contested}],
+                filter_expression=f"{IDENTITY_KEY} LIKE '{EXPLICIT_PREFIX}%'",
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    expected = {key.format(value=value, other=other_value): count for key, count in expected_template.items()}
+    groups = {row[0]: row[1] for row in get_scalar_table_data(response.json()) if row[0] in expected}
+    assert groups == expected, get_scalar_table_data(response.json())
+
+
+# Rule for a raw select of a bare key that is a resource attribute and an
+# attribute: each row shows the resource value. This is also true for a row
+# where the attribute has a different value.
+@pytest.mark.parametrize("signal", ["traces", "logs"])
+def test_select_of_ambiguous_name(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    ambiguous_rows: datetime,
+    signal: str,
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((ambiguous_rows - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((ambiguous_rows + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                signal,
+                limit=100,
+                filter_expression=f"{IDENTITY_KEY} LIKE '{EXPLICIT_PREFIX}%'",
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": IDENTITY_KEY}, {"name": "service.name"}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    rows = response.json()["data"]["data"]["results"][0]["rows"] or []
+    by_identity = {row["data"][IDENTITY_KEY]: row["data"]["service.name"] for row in rows if row["data"].get(IDENTITY_KEY, "").startswith(EXPLICIT_PREFIX)}
+    assert by_identity == {
+        COLUMN_ONLY: "svc-a",
+        ATTRIBUTE_ONLY: "svc-b",
+        BOTH: "svc-a",
+        NEITHER: "svc-b",
+        NUMBER_ATTRIBUTE: "svc-b",
+    }
+
+
+# Rules for an order by, descending, with the timestamp descending as the
+# second key:
+# - A bare key or a key under the signal's own context sorts by the column
+#   only.
+# - An `attribute.` key sorts by the attribute on traces. The number
+#   attribute sorts as text. Rows without the attribute come last.
+# - An `attribute.` key sorts by the column on logs.
+BY_COLUMN = [ATTRIBUTE_ONLY, NEITHER, NUMBER_ATTRIBUTE, COLUMN_ONLY, BOTH]
+ORDER_BY_MATRIX = [
+    pytest.param(None, {"traces": BY_COLUMN, "logs": BY_COLUMN}, id="bare_orders_by_the_column"),
+    pytest.param("own", {"traces": BY_COLUMN, "logs": BY_COLUMN}, id="own_context_orders_by_the_column"),
+    pytest.param(
+        "attribute",
+        {"traces": [ATTRIBUTE_ONLY, BOTH, NUMBER_ATTRIBUTE, COLUMN_ONLY, NEITHER], "logs": BY_COLUMN},
+        id="attribute_context_orders_by_the_attribute_on_traces_only",
+    ),
+]
+
+
+@pytest.mark.parametrize("context,expected", ORDER_BY_MATRIX)
+@pytest.mark.parametrize("signal,own_context,contested,value,other_value", SIGNALS)
+def test_order_by_resolution(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    ambiguous_rows: datetime,
+    signal: str,
+    own_context: str,
+    contested: str,
+    value: str,  # pylint: disable=unused-argument
+    other_value: str,  # pylint: disable=unused-argument
+    context: str | None,
+    expected: dict[str, list[str]],
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    prefix = f"{own_context}." if context == "own" else f"{context}." if context else ""
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((ambiguous_rows - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((ambiguous_rows + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                signal,
+                limit=100,
+                filter_expression=f"{IDENTITY_KEY} LIKE '{EXPLICIT_PREFIX}%'",
+                order=[build_order_by(f"{prefix}{contested}", "desc"), build_order_by("timestamp", "desc")],
+                select_fields=[{"name": IDENTITY_KEY}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    ordered = [row for row in get_column_data_from_response(response.json(), IDENTITY_KEY) if row.startswith(EXPLICIT_PREFIX)]
+    assert ordered == expected[signal]
+
+
+# Rules for an aggregation argument:
+# - A bare key counts the values of the column only.
+# - An `attribute.` key counts the attribute in both data types. The number
+#   attribute adds one distinct value.
+AGGREGATION_MATRIX = [
+    pytest.param(None, 2, id="bare_counts_the_column"),
+    pytest.param("own", 2, id="own_context_counts_the_column"),
+    pytest.param("attribute", 2, id="attribute_context_counts_both_attribute_types"),
+]
+
+
+@pytest.mark.parametrize("context,expected", AGGREGATION_MATRIX)
+@pytest.mark.parametrize("signal,own_context,contested,value,other_value", SIGNALS)
+def test_aggregation_argument_resolution(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    ambiguous_rows: datetime,
+    signal: str,
+    own_context: str,
+    contested: str,
+    value: str,  # pylint: disable=unused-argument
+    other_value: str,  # pylint: disable=unused-argument
+    context: str | None,
+    expected: int,
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    prefix = f"{own_context}." if context == "own" else f"{context}." if context else ""
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((ambiguous_rows - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((ambiguous_rows + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.SCALAR,
+        queries=[
+            build_scalar_query(
+                "A",
+                signal,
+                [build_aggregation(f"count_distinct({prefix}{contested})", "distinct")],
+                filter_expression=f"{IDENTITY_KEY} LIKE '{EXPLICIT_PREFIX}%'",
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert_scalar_value(response, "A", expected)
+
+
+# Rules for logs only:
+# - A `body.` key reads the body JSON path. It does not read the attribute
+#   with the same name.
+# - A `log.` key reads the attribute and the body JSON path together. This
+#   is also true when metadata reports the attribute.
+# - A `scope.` key resolves through metadata only. When metadata does not
+#   report the key, the query fails with "key not found". This is also true
+#   for the declared path `scope.name` and for rows that have the scope
+#   data.
+LOGS_ONLY_MATRIX = [
+    pytest.param("body.route.tag = 'checkout'", {ATTRIBUTE_ONLY, NEITHER, NUMBER_ATTRIBUTE}, id="body_context_reads_the_body_json"),
+    pytest.param("log.route.tag = 'checkout'", {COLUMN_ONLY, ATTRIBUTE_ONLY, BOTH, NEITHER, NUMBER_ATTRIBUTE}, id="log_context_reads_attribute_and_body"),
+    pytest.param("scope.name = 'scope-a'", "key `name` not found", id="scope_name_needs_metadata"),
+    pytest.param("scope.env = 'prod'", "key `env` not found", id="scope_attribute_needs_metadata"),
+    pytest.param("scope.env EXISTS", "key `env` not found", id="scope_attribute_exists_needs_metadata"),
+]
+
+
+@pytest.mark.parametrize("expression,expected", LOGS_ONLY_MATRIX)
+def test_logs_body_and_scope_contexts(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    ambiguous_rows: datetime,
+    expression: str,
+    expected: set[str] | str,
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((ambiguous_rows - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((ambiguous_rows + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                "logs",
+                limit=100,
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": IDENTITY_KEY}],
+            )
+        ],
+    )
+
+    if isinstance(expected, str):
+        assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+        assert expected in response.text, response.text
+        return
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    matched = {row for row in get_column_data_from_response(response.json(), IDENTITY_KEY) if row.startswith(EXPLICIT_PREFIX)}
+    assert matched == expected, expression