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

The same name can exist as an intrinsic column and as an attribute (`name`
on spans, `severity_text` on logs), in two data types under one context,
or as a resource attribute and as a span or log attribute (`service.name`).
These cases pin what the query builder does today for a filter, EXISTS, a
group by, an order by, an aggregation argument, and a raw select on each
shape, so a change to name resolution shows as a failure here, not as a
surprise in a dashboard.

- An explicit context is honored as written. The ambiguity warning fires
  for a bare name with several readings and for an explicit attribute
  context whose attribute exists in two data types.
- A bare name that is both a column and an attribute reads both in a
  filter, orders and groups by the column alone, and counts the column's
  values alone. A string operand reaches a number attribute through a
  text cast.
- A bare name that is both a resource and an attribute reads the resource
  in a filter and in a raw select, with a warning in the filter.
- Under an explicit attribute context, traces order by the attribute and
  logs still order by the column.
- A key under the signal's own context that exists only as an attribute
  corrects to the attribute; on logs the correction also reads the body
  JSON path. A `scope.` key on logs resolves through metadata alone and
  answers "key not found" otherwise, also for the declared `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,98 @@ def keyless_series(insert_metrics: Callable[[list[Metrics]], None]) -> Generator
         ]
     )
     yield start, start + points * 60
+
+
+EXPLICIT_PREFIX = "explicit-ctx"
+# Unambiguous string attribute that names the row. Every assertion reads it back.
+IDENTITY_KEY = "probe.id"
+# Attribute-only key with no same-named column, for the own-context miss. On
+# logs the rows that lack the attribute carry it nested in the body JSON.
+ATTRIBUTE_ONLY_KEY = "route.tag"
+CONTESTED_VALUE = "checkout"
+
+# Row identities, by where the contested name carries the contested value:
+# the intrinsic column (`name` on spans, `severity_text` on logs), the
+# same-named string attribute, both, neither, or a same-named number
+# attribute (a data type that contradicts 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, column carries the value, attribute carries the value,
+#  attribute carries the number, resource service.name, attribute service.name,
+#  carries 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 declared scope path `scope.name` next to a scope attribute
+# that is also named `name`, and a plain scope attribute.
+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]:
+    """One span and one log per identity. `service.name` exists as a resource
+    attribute on every row and as a span or log attribute on some, with
+    different values, so a bare `service.name` is ambiguous. Logs that lack
+    the `route.tag` attribute carry it in the body JSON instead. Logs with
+    the column value carry the scope name; logs with the attribute value
+    carry 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,371 @@
+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,
+)
+
+# Which rows a filter returns when the same name exists as an intrinsic
+# column and as an attribute (`name` on spans, `severity_text` on logs), or
+# as a resource attribute and a span or log attribute (`service.name`).
+# An explicit context is honored as written. A bare name that is both a
+# column and an attribute reads both, with an ambiguity warning. A bare name
+# that is both a resource and an attribute reads the resource, with a
+# warning. The warning also fires for an explicit attribute context when the
+# attribute exists in two data types, and a string operand reaches the
+# number attribute through a text cast. A key under the signal's own context
+# that exists only as an attribute corrects to the attribute; on logs the
+# correction 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
+
+
+# Group by resolves the contested name in the column stage: a bare name that
+# is both a column and an attribute groups by the column alone, an explicit
+# context groups by that context alone.
+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())
+
+
+# A raw select of a bare name that is both a resource and an attribute reads
+# one value per row: the resource value, also on the row whose attribute
+# carries 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",
+    }
+
+
+# Order by resolves the contested name in the column stage, descending, with
+# the timestamp descending as the tie breaker. A bare or own-context name
+# sorts by the column alone. An explicit attribute context sorts by the
+# attribute on traces, where the number attribute reads as text and rows
+# without the attribute come last; on logs it still sorts by the column.
+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]
+
+
+# An aggregation argument resolves the contested name in the column stage: a
+# bare name counts the column's values alone; an attribute context counts the
+# attribute in both of its data types, so the number attribute adds a
+# 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)
+
+
+# Logs only. `body.x` addresses the body JSON and never the same-named
+# attribute; `log.x` reads the attribute and the body JSON path together,
+# even when the attribute exists in metadata. A `scope.` key is a strict
+# context resolved through metadata alone: the declared scope path
+# `scope.name` and a scope attribute both answer "key not found" when
+# metadata does not report them, even when the rows carry them.
+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