From 3ea06a9d85a098d15a357a7ed7cb90ee4bec5d6d Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Wed, 2 Sep 2026 16:56:57 +0200 Subject: [PATCH 1/2] feat(executor): parse RESP3 map replies for FT.SEARCH and FT.AGGREGATE sql-redis sends FT.* commands with client.execute_command and reads the raw reply. redis-py registers no FT.SEARCH / FT.AGGREGATE response callback on the base client (Search.search() calls execute_command and then its own _parse_results), so the executor receives the wire shape verbatim: a flat array on RESP2, a map on RESP3. The parser assumed the array, so on a RESP3 connection its positional strides walked into integer keys that do not exist. FT.SEARCH raised KeyError: 2 and FT.AGGREGATE raised KeyError: slice(1, None, None). Since redis-py 8 defaults to RESP3, this broke on upgrade without the caller passing protocol at all. _parse_reply now dispatches on the shape of the reply, and _resp3_to_resp2 folds a map back into the array shape so one parser handles both protocols. The four branch behaviours (score-alias resolution over the union of row field names, the collision loop, nil field-array tolerance, count extraction) move into _parse_array_reply unchanged, which is what keeps protocol=2 output identical. Both execute() bodies lose their duplicated parse block. Shape dispatch rather than a protocol check on the connection: sniffing connection_kwargs is wrong for cluster and sentinel clients, and cannot distinguish redis-py 8's four callback modes. Routing through client.ft(index) does not help either, since redis-py 7.4 and earlier return the raw map from _parse_results under RESP3, and redis-py 8's protocol=3 callback table has no FT.AGGREGATE entry. Structural map keys arrive as bytes when decode_responses=False, so _map_get probes both key types; reading only str keys fails silently with zero rows, which is the bug redis-py shipped in 8.0.0 (#4107). Document field keys are passed through untouched, as before. FT.HYBRID keeps its own parser: measured against Redis 8.4.6, its RESP3 rows are flat field maps, not the {id, extra_attributes} documents FT.SEARCH returns. FT.INFO and FT._LIST were already shape-tolerant. Verified against Redis 8.4.6 across redis-py {6.3.0, 7.4.0, 8.1.0} and client modes {protocol=2, unset, protocol=3, legacy_responses=False, protocol=3 with decode_responses=False}, sync and async: rows and counts are identical to the protocol=2 baseline in all 30 cells. RESP3 total_results matches the RESP2 leading count for FT.AGGREGATE too. The one accepted divergence is the score's type, float on RESP3 and str on RESP2, since Redis sends a double there. It is passed through rather than normalised; float() is the protocol-safe conversion at the call site and is what the existing tests already use. --- .github/workflows/test.yml | 52 ++++ docs/concepts/result-shape.md | 23 +- docs/for-ais-only/FAILURE_MODES.md | 27 +++ docs/for-ais-only/REPOSITORY_MAP.md | 2 +- sql_redis/executor.py | 308 +++++++++++++----------- tests/conftest.py | 53 ++++ tests/test_nil_field_array.py | 60 ++--- tests/test_protocol_matrix.py | 253 +++++++++++++++++++ tests/test_reply_shapes.py | 360 ++++++++++++++++++++++++++++ 9 files changed, 952 insertions(+), 186 deletions(-) create mode 100644 tests/test_protocol_matrix.py create mode 100644 tests/test_reply_shapes.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 491597d..cce719d 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -46,3 +46,55 @@ jobs: run: | make test + + redis-py-compat: + name: redis-py ${{ matrix.redis-py }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + redis-py: ["6.3.0", "7.4.0", "8.1.0"] + + steps: + - name: Check out repository + uses: actions/checkout@v6 + + - name: Install Python + uses: actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Install uv + uses: astral-sh/setup-uv@v6 + with: + version: ${{ env.UV_VERSION }} + enable-cache: true + python-version: "3.12" # sets UV_PYTHON + cache-dependency-glob: | + pyproject.toml + uv.lock + + - name: Install dependencies + run: | + uv sync + + # Without this, a resolution surprise leaves the job exercising the + # locked redis-py while still reporting green, and the axis rots. + - name: Confirm the redis-py overlay took effect + run: > + uv run --with 'redis==${{ matrix.redis-py }}' + python -c "import redis, sys; + print('redis-py', redis.__version__); + sys.exit(0 if redis.__version__ == '${{ matrix.redis-py }}' else 1)" + + # Reply shapes depend on the redis-py version and the negotiated RESP + # protocol, so these files are run against each supported redis-py. + # 'uv run --with' layers an ephemeral overlay and never writes uv.lock. + - name: Run reply-shape and protocol tests + run: > + uv run --with 'redis==${{ matrix.redis-py }}' python -m pytest + tests/test_reply_shapes.py + tests/test_protocol_matrix.py + tests/test_nil_field_array.py + tests/test_ft_hybrid.py + -q diff --git a/docs/concepts/result-shape.md b/docs/concepts/result-shape.md index 246fb81..a166705 100644 --- a/docs/concepts/result-shape.md +++ b/docs/concepts/result-shape.md @@ -4,14 +4,14 @@ Every `Executor.execute()` call returns a `QueryResult` with two attributes: `ro ## What `count` means -The integer in `count` is what Redis returned at position 0 of the reply. Its meaning depends on which command ran (see [FT.SEARCH vs FT.AGGREGATE](search-vs-aggregate.md)): +The integer in `count` is the total Redis reported for the query: position 0 of a RESP2 array reply, or `total_results` in a RESP3 map reply. The two agree. Its meaning depends on which command ran (see [FT.SEARCH vs FT.AGGREGATE](search-vs-aggregate.md)): - **`FT.SEARCH`** (no aggregation): `count` is the **total number of matching documents** in the index, regardless of `LIMIT`. So `count` can be much larger than `len(rows)`. This is useful for pagination. - **`FT.AGGREGATE`** (any aggregation, GROUP BY, computed field, date function): `count` is the **number of rows in the reply**, which is what you got back. After `LIMIT`, the two are equal. ## What `rows` looks like -A row is always a `dict`, but the *keys* and *values* depend on three orthogonal factors. +A row is always a `dict`, but the *keys* and *values* depend on five orthogonal factors. ### Factor 1: client decoding @@ -55,9 +55,22 @@ If your SELECT contains `score()`, the underlying `FT.SEARCH` runs with `WITHSCO The score column name is whatever alias you wrote (`score() AS relevance` produces `relevance`). If you used `score()` without `AS`, the library falls back to a stable internal name. -The score's value type is `str` (or `bytes` depending on decoding); if you need a `float`, convert at the call site. +The score's value type depends on the wire protocol: `str` (or `bytes`, depending on decoding) on RESP2, and `float` on RESP3, where Redis sends it as a double. `float()` is the protocol-safe conversion at the call site and works on both. -### Factor 4: `RETURN 0` +### Factor 4: wire protocol + +Redis replies to `FT.SEARCH` and `FT.AGGREGATE` with a flat array on RESP2 and +a map on RESP3, and `sql_redis` reads the raw reply, so the protocol your client +negotiated decides which shape arrives. The library parses both, and the rows +come out the same. Note that redis-py 8 defaults to RESP3, where earlier +versions defaulted to RESP2, so this can change under you on an upgrade without +you passing `protocol` at all. + +One value is visibly protocol-dependent: the score, described above. Redis sends +it as text on RESP2 and as a double on RESP3, and the library passes it through +rather than normalising it. + +### Factor 5: `RETURN 0` If your SELECT is just `score()` with no document fields, the underlying command uses `RETURN 0` (no document fields returned). Each row contains only the score column: @@ -81,7 +94,7 @@ You will not normally see this; it is a defensive measure for the unusual case. It would be tempting to normalise everything: always strings, always `float` scores, always uniform shape. The library does not, for three reasons. -1. **Bytes is the right default for Redis.** Some fields legitimately contain binary data. Force-decoding to UTF-8 would corrupt those. +1. **Bytes is the right default for Redis.** Some fields legitimately contain binary data. Force-decoding to UTF-8 would corrupt those. The same argument covers the RESP3 score: stringifying a double cannot reproduce the text Redis would have sent on RESP2 (`repr(1.0)` is `"1.0"`, Redis writes `"1"`), so normalising would invent a third form rather than deliver parity. 2. **The shape difference between `FT.SEARCH` and `FT.AGGREGATE` is intrinsic to the Redis commands.** Hiding it would lie about what is happening. 3. **One-call-site logic is cheap; library-wide policy is expensive.** A user who wants a uniform shape can write a 5-line decoder once, customised to their data. The library shipping that decoder for everyone is the wrong place for the policy. diff --git a/docs/for-ais-only/FAILURE_MODES.md b/docs/for-ais-only/FAILURE_MODES.md index 3cac618..f6e35b3 100644 --- a/docs/for-ais-only/FAILURE_MODES.md +++ b/docs/for-ais-only/FAILURE_MODES.md @@ -13,6 +13,33 @@ not "fix" this in `executor.py` by force-decoding; that breaks users who deliberately want bytes (binary fields, vectors). See [Result shape](../concepts/result-shape.md) for the full story. +## Reply shape depends on the RESP protocol, not on library configuration + +`sql_redis` sends FT.* commands with `client.execute_command` and reads the raw +reply. redis-py registers no `FT.SEARCH` / `FT.AGGREGATE` response callback on +the base client. `Search.search()` calls `execute_command` and then its own +`_parse_results`, so the executor receives the wire shape verbatim: a flat +array on RESP2, a map on RESP3. + +`_parse_reply` therefore dispatches on the shape of the reply +(`isinstance(raw_result, dict)`), and `_resp3_to_resp2` folds a map back into +the array shape so one parser handles both. Do not "simplify" this into a +single unconditional path, and do not replace the shape check with a protocol +check on the connection: sniffing `connection_kwargs["protocol"]` is wrong for +cluster and sentinel clients, and it cannot distinguish redis-py 8's four +callback modes (`protocol=2`, protocol unset, `protocol=3`, +`legacy_responses=False`). + +Routing through `client.ft(index)` does not help. On redis-py 7.4 and earlier +`_parse_results` returns the raw map unchanged under RESP3, and on redis-py 8 +the `protocol=3` callback table has no `FT.AGGREGATE` entry at all. + +Structural map keys (`total_results`, `results`, `id`, `score`, +`extra_attributes`) arrive as `bytes` when the client has +`decode_responses=False`, which is why `_map_get` probes both key types. Reading +only `str` keys fails silently with zero rows rather than raising; that is +exactly the bug redis-py shipped in 8.0.0 (redis-py#4107). + ## Stopwords are silently stripped `WHERE title = 'bank of america'` becomes `@title:"bank america"` because diff --git a/docs/for-ais-only/REPOSITORY_MAP.md b/docs/for-ais-only/REPOSITORY_MAP.md index cbae43e..3cc5ca6 100644 --- a/docs/for-ais-only/REPOSITORY_MAP.md +++ b/docs/for-ais-only/REPOSITORY_MAP.md @@ -69,7 +69,7 @@ tests/ | `IS NULL`, `IS NOT NULL`, `exists()` | `parser.py`, `query_builder.py` | | Parameter substitution | `executor.py::_substitute_params` | | FT.SEARCH vs FT.AGGREGATE branching | `translator.py::translate_parsed` (the `use_aggregate` boolean) | -| Result-row parsing | `executor.py::Executor.execute` (the four parsing branches: WITHSCORES + RETURN 0, WITHSCORES, plain SEARCH, AGGREGATE) | +| Result-row parsing | `executor.py::_ScoreParseMixin` (`_parse_reply` dispatches by reply shape; `_parse_array_reply` holds the four branches: WITHSCORES + RETURN 0, WITHSCORES, plain SEARCH, AGGREGATE; `_resp3_to_resp2` folds a RESP3 map into the array shape) | ## What to read before changing X diff --git a/sql_redis/executor.py b/sql_redis/executor.py index a9549ba..9040f77 100644 --- a/sql_redis/executor.py +++ b/sql_redis/executor.py @@ -9,7 +9,7 @@ import redis from sql_redis.schema import AsyncSchemaRegistry, SchemaRegistry -from sql_redis.translator import Translator +from sql_redis.translator import TranslatedQuery, Translator if TYPE_CHECKING: import redis.asyncio as async_redis @@ -107,6 +107,36 @@ def _substitute_params(sql: str, params: dict[str, Any]) -> str: return "".join(result) +def _map_get(mapping: dict, key: str, default: Any = None) -> Any: + """Look up a RESP3 map key, tolerating bytes keys. + + RESP3 replies arrive as maps whose *structural* keys (``total_results``, + ``results``, ``id``, ``score``, ``extra_attributes``) are ``str`` when the + client decodes responses and ``bytes`` when it does not. Document *field* + keys never go through this helper: they reach the caller exactly as + received, so a ``decode_responses=False`` client still gets bytes keys. + """ + if key in mapping: + return mapping[key] + return mapping.get(key.encode(), default) + + +def _extra_attributes(result_item: Any) -> list: + """Flatten a RESP3 result's ``extra_attributes`` map into a field array. + + Returns the ``[field, value, field, value, ...]`` shape the RESP2 parser + expects. An absent or non-map ``extra_attributes`` (RETURN 0 suppresses it, + and a document expiring mid-query can null it) becomes an empty array, + matching the RESP2 path's tolerance for a nil field-array. + """ + if not isinstance(result_item, dict): + return [] + fields = _map_get(result_item, "extra_attributes") + if not isinstance(fields, dict): + return [] + return [item for pair in fields.items() for item in pair] + + @dataclass class QueryResult: """Result of executing a SQL query.""" @@ -140,13 +170,8 @@ def _parse_hybrid_reply(raw_result) -> tuple[Any, list[dict]]: else: reply = dict(zip(raw_result[::2], raw_result[1::2])) - def _field(name: str): - if name in reply: - return reply[name] - return reply.get(name.encode()) - - count = _field("total_results") or 0 - results = _field("results") or [] + count = _map_get(reply, "total_results") or 0 + results = _map_get(reply, "results") or [] rows = [ dict(row) if isinstance(row, dict) else dict(zip(row[::2], row[1::2])) for row in results @@ -205,6 +230,140 @@ def _resolve_score_alias( alias = f"__score_{alias}" return alias + @staticmethod + def _resp3_to_resp2(reply: dict, translated: TranslatedQuery) -> list: + """Rewrite a RESP3 FT.SEARCH / FT.AGGREGATE map as the RESP2 array. + + A RESP3 reply is a map:: + + {total_results: N, attributes: [...], format: ..., warning: [...], + results: [{id: key, score: 0.5, extra_attributes: {field: val}, + values: [...]}, ...]} + + Rather than parse that shape separately, fold it back into the flat + array RESP2 sends so a single parser handles both wire protocols. The + emitted layout mirrors the branch the array parser will take: the + document key, then the score when WITHSCORES was requested, then the + field array unless RETURN 0 suppressed document fields. ``values`` is + never read, and ``id`` lands in the position the parser discards. + + Values are moved, not converted, so a RESP3 score stays a float and + document field keys keep whatever type the client returned. + """ + response: list = [_map_get(reply, "total_results", 0)] + results = _map_get(reply, "results") or [] + + if translated.command != "FT.SEARCH": + # FT.AGGREGATE rows carry only extra_attributes: no id, no score. + response.extend(_extra_attributes(item) for item in results) + return response + + with_scores = translated.score_alias is not None + no_content = _ScoreParseMixin._has_return_0(translated.args) + for item in results: + response.append(_map_get(item, "id", "") if isinstance(item, dict) else "") + if with_scores: + response.append( + _map_get(item, "score") if isinstance(item, dict) else None + ) + if not no_content: + response.append(_extra_attributes(item)) + return response + + @staticmethod + def _parse_array_reply( + raw_result: Any, translated: TranslatedQuery + ) -> tuple[Any, list[dict]]: + """Parse the flat-array FT.SEARCH / FT.AGGREGATE reply into rows. + + This is the RESP2 wire shape, and the shape ``_resp3_to_resp2`` folds a + RESP3 map into, so it is the only row-assembly path in the executor. + """ + count = raw_result[0] if isinstance(raw_result, list) and raw_result else 0 + rows: list[dict] = [] + + if translated.command == "FT.SEARCH": + # Use the explicit score_alias signal rather than scanning args + # for the literal token "WITHSCORES", which could false-positive + # if a returned field happened to be named "WITHSCORES". + with_scores = translated.score_alias is not None + # RETURN 0 suppresses document fields (like NOCONTENT); + # with WITHSCORES the reply is [count, id, score, id, score, ...] + no_content = _ScoreParseMixin._has_return_0(translated.args) + + if with_scores and no_content: + # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...] + # Stride of 2: key, score (no field array) + score_alias = _ScoreParseMixin._resolve_score_alias( + translated.score_alias, translated.args + ) + for i in range(1, len(raw_result) - 1, 2): + score = raw_result[i + 1] + row = {score_alias: score} + rows.append(row) + elif with_scores: + # WITHSCORES format: [count, key1, score1, [fields1], key2, score2, [fields2], ...] + # Stride of 3: key, score, field_list + # First pass: collect all field names across all rows so the + # alias avoids collisions with any document field, not just + # the first row's fields. + all_field_names: set[str] = set() + parsed_rows: list[tuple[dict, Any]] = [] + for i in range(1, len(raw_result) - 2, 3): + score = raw_result[i + 1] + # A nil field-array (e.g. doc expired mid-query) becomes an + # empty field set, keeping the row's score instead of crashing. + row_data = raw_result[i + 2] or [] + row = dict(zip(row_data[::2], row_data[1::2])) + all_field_names.update(row.keys()) + parsed_rows.append((row, score)) + resolved_alias = _ScoreParseMixin._resolve_score_alias( + translated.score_alias, + translated.args, + first_row_fields=all_field_names, + ) + for row, score in parsed_rows: + row[resolved_alias] = score + rows.append(row) + else: + # Standard format: [count, key1, [fields1], key2, [fields2], ...] + for i in range(2, len(raw_result), 2): + row_data = raw_result[i] or [] + row = dict(zip(row_data[::2], row_data[1::2])) + rows.append(row) + else: + # FT.AGGREGATE format: [count, [fields1], [fields2], ...] + for row_data in raw_result[1:]: + row_data = row_data or [] + row = dict(zip(row_data[::2], row_data[1::2])) + rows.append(row) + + return count, rows + + @staticmethod + def _parse_reply(raw_result: Any, translated: TranslatedQuery) -> QueryResult: + """Turn a raw FT.* reply into a QueryResult, RESP2 or RESP3. + + Dispatch is by reply *shape*, not by asking the connection which + protocol it negotiated. redis-py registers no FT.SEARCH / FT.AGGREGATE + callback on the base client: ``Search.search()`` calls + ``execute_command`` and then its own ``_parse_results``, so a raw + ``execute_command`` receives the wire shape verbatim in every redis-py + version, a flat array on RESP2 and a map on RESP3. Shape dispatch needs no + version check, works for cluster and sentinel clients, and survives a + future redis-py that hands back the flat array on RESP3 too. + + FT.HYBRID keeps its own parser: its RESP3 rows are flat field maps, + not the ``{id, extra_attributes}`` documents FT.SEARCH returns. + """ + if translated.command == "FT.HYBRID": + count, rows = _ScoreParseMixin._parse_hybrid_reply(raw_result) + else: + if isinstance(raw_result, dict): + raw_result = _ScoreParseMixin._resp3_to_resp2(raw_result, translated) + count, rows = _ScoreParseMixin._parse_array_reply(raw_result, translated) + return QueryResult(rows=rows, count=count) + class Executor(_ScoreParseMixin): """Executes SQL queries against Redis.""" @@ -271,74 +430,9 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: ) from e raise - # Parse result based on command type - # FT.SEARCH/FT.AGGREGATE replies are arrays with the count first; - # FT.HYBRID replies are maps (dict) and set count during parsing below. - count = raw_result[0] if isinstance(raw_result, list) and raw_result else 0 - rows: list[dict] = [] - - if translated.command == "FT.HYBRID": - count, rows = self._parse_hybrid_reply(raw_result) - elif translated.command == "FT.SEARCH": - # Use the explicit score_alias signal rather than scanning args - # for the literal token "WITHSCORES", which could false-positive - # if a returned field happened to be named "WITHSCORES". - with_scores = translated.score_alias is not None - # RETURN 0 suppresses document fields (like NOCONTENT); - # with WITHSCORES the reply is [count, id, score, id, score, ...] - no_content = self._has_return_0(translated.args) - - # Pre-resolve score alias; may be deferred for SELECT * - score_alias: str | None = None - - if with_scores and no_content: - # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...] - # Stride of 2: key, score (no field array) - score_alias = self._resolve_score_alias( - translated.score_alias, translated.args - ) - for i in range(1, len(raw_result) - 1, 2): - score = raw_result[i + 1] - row = {score_alias: score} - rows.append(row) - elif with_scores: - # WITHSCORES format: [count, key1, score1, [fields1], key2, score2, [fields2], ...] - # Stride of 3: key, score, field_list - # First pass: collect all field names across all rows so the - # alias avoids collisions with any document field, not just - # the first row's fields. - all_field_names: set[str] = set() - parsed_rows: list[tuple[dict, Any]] = [] - for i in range(1, len(raw_result) - 2, 3): - score = raw_result[i + 1] - # A nil field-array (e.g. doc expired mid-query) becomes an - # empty field set, keeping the row's score instead of crashing. - row_data = raw_result[i + 2] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - all_field_names.update(row.keys()) - parsed_rows.append((row, score)) - resolved_alias = self._resolve_score_alias( - translated.score_alias, - translated.args, - first_row_fields=all_field_names, - ) - for row, score in parsed_rows: - row[resolved_alias] = score - rows.append(row) - else: - # Standard format: [count, key1, [fields1], key2, [fields2], ...] - for i in range(2, len(raw_result), 2): - row_data = raw_result[i] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - else: - # FT.AGGREGATE format: [count, [fields1], [fields2], ...] - for row_data in raw_result[1:]: - row_data = row_data or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - - return QueryResult(rows=rows, count=count) + # Parse the reply. Shape dispatch happens in _parse_reply: RESP2 sends + # a flat array, RESP3 sends a map, and both reach the same row parser. + return self._parse_reply(raw_result, translated) class AsyncExecutor(_ScoreParseMixin): @@ -420,65 +514,9 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: ) from e raise - # Parse result based on command type - # FT.SEARCH/FT.AGGREGATE replies are arrays with the count first; - # FT.HYBRID replies are maps (dict) and set count during parsing below. - count = raw_result[0] if isinstance(raw_result, list) and raw_result else 0 - rows: list[dict] = [] - - if translated.command == "FT.HYBRID": - count, rows = self._parse_hybrid_reply(raw_result) - elif translated.command == "FT.SEARCH": - with_scores = translated.score_alias is not None - no_content = self._has_return_0(translated.args) - - score_alias: str | None = None - - if with_scores and no_content: - # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...] - score_alias = self._resolve_score_alias( - translated.score_alias, translated.args - ) - for i in range(1, len(raw_result) - 1, 2): - score = raw_result[i + 1] - row = {score_alias: score} - rows.append(row) - elif with_scores: - # WITHSCORES format: [count, key1, score1, [fields1], ...] - # First pass: collect all field names across all rows so the - # alias avoids collisions with any document field. - all_field_names: set[str] = set() - parsed_rows: list[tuple[dict, Any]] = [] - for i in range(1, len(raw_result) - 2, 3): - score = raw_result[i + 1] - # A nil field-array (e.g. doc expired mid-query) becomes an - # empty field set, keeping the row's score instead of crashing. - row_data = raw_result[i + 2] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - all_field_names.update(row.keys()) - parsed_rows.append((row, score)) - resolved_alias = self._resolve_score_alias( - translated.score_alias, - translated.args, - first_row_fields=all_field_names, - ) - for row, score in parsed_rows: - row[resolved_alias] = score - rows.append(row) - else: - # Standard format: [count, key1, [fields1], key2, [fields2], ...] - for i in range(2, len(raw_result), 2): - row_data = raw_result[i] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - else: - # FT.AGGREGATE format: [count, [fields1], [fields2], ...] - for row_data in raw_result[1:]: - row_data = row_data or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - - return QueryResult(rows=rows, count=count) + # Parse the reply. Shape dispatch happens in _parse_reply: RESP2 sends + # a flat array, RESP3 sends a map, and both reach the same row parser. + return self._parse_reply(raw_result, translated) def create_executor( diff --git a/tests/conftest.py b/tests/conftest.py index c0e7524..677d116 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,11 +1,15 @@ """Pytest configuration and fixtures.""" import struct +from unittest.mock import AsyncMock, MagicMock import pytest import redis from testcontainers.redis import RedisContainer +from sql_redis.executor import AsyncExecutor, Executor +from sql_redis.translator import TranslatedQuery + @pytest.fixture(scope="module") def redis_container(): @@ -358,3 +362,52 @@ def items_data(redis_client: redis.Redis, items_index: str): binary_client.close() return items_index + + +def make_stub_executor(translated: TranslatedQuery, raw_result) -> Executor: + """Build an Executor whose translation and Redis reply are stubbed. + + Feeds a crafted reply straight through the parser, so every parse branch + can be exercised deterministically without a live Redis server. Exposed as + the ``stub_executor`` fixture. + """ + executor = Executor.__new__(Executor) + executor._client = MagicMock() + executor._client.execute_command.return_value = raw_result + executor._schema_registry = MagicMock() + executor._translator = MagicMock() + executor._translator.translate.return_value = translated + return executor + + +def make_stub_async_executor(translated: TranslatedQuery, raw_result) -> AsyncExecutor: + """Build an AsyncExecutor whose translation and Redis reply are stubbed. + + The async seam differs from the sync one: ``AsyncExecutor.execute`` parses + first, awaits ``ensure_schema`` when the query names an index, and only then + translates the pre-parsed result. Exposed as the ``stub_async_executor`` + fixture. + """ + executor = AsyncExecutor.__new__(AsyncExecutor) + executor._client = MagicMock() + executor._client.execute_command = AsyncMock(return_value=raw_result) + executor._schema_registry = MagicMock() + executor._schema_registry.ensure_schema = AsyncMock() + executor._translator = MagicMock() + parsed = MagicMock() + parsed.index = None # skip ensure_schema() + executor._translator.parse.return_value = parsed + executor._translator.translate_parsed.return_value = translated + return executor + + +@pytest.fixture +def stub_executor(): + """Factory fixture returning :func:`make_stub_executor`.""" + return make_stub_executor + + +@pytest.fixture +def stub_async_executor(): + """Factory fixture returning :func:`make_stub_async_executor`.""" + return make_stub_async_executor diff --git a/tests/test_nil_field_array.py b/tests/test_nil_field_array.py index ddce90d..87e5994 100644 --- a/tests/test_nil_field_array.py +++ b/tests/test_nil_field_array.py @@ -9,44 +9,14 @@ These tests feed crafted replies (with a nil field-array) straight through the parser by mocking the translator and client, so every parse branch is exercised -deterministically without a live Redis server. +deterministically without a live Redis server. The ``stub_executor`` / +``stub_async_executor`` factory fixtures live in ``conftest.py``; the RESP3 +equivalents of these cases are in ``test_reply_shapes.py``. """ -from unittest.mock import AsyncMock, MagicMock - -import pytest - -from sql_redis.executor import AsyncExecutor, Executor from sql_redis.translator import TranslatedQuery -def _make_sync_executor(translated: TranslatedQuery, raw_result): - """Build an Executor whose translation and Redis reply are stubbed.""" - executor = Executor.__new__(Executor) - executor._client = MagicMock() - executor._client.execute_command.return_value = raw_result - executor._schema_registry = MagicMock() - executor._translator = MagicMock() - executor._translator.translate.return_value = translated - return executor - - -def _make_async_executor(translated: TranslatedQuery, raw_result): - """Build an AsyncExecutor whose translation and Redis reply are stubbed.""" - executor = AsyncExecutor.__new__(AsyncExecutor) - executor._client = MagicMock() - executor._client.execute_command = AsyncMock(return_value=raw_result) - executor._schema_registry = MagicMock() - executor._schema_registry.ensure_schema = AsyncMock() - executor._translator = MagicMock() - # AsyncExecutor.execute parses first, then translates the parsed result. - parsed = MagicMock() - parsed.index = None # skip ensure_schema() - executor._translator.parse.return_value = parsed - executor._translator.translate_parsed.return_value = translated - return executor - - class TestStandardSearchNilFields: """Standard FT.SEARCH reply: [count, key1, [fields1], key2, [fields2], ...].""" @@ -57,19 +27,19 @@ def _translated(self) -> TranslatedQuery: query_string="*", ) - def test_sync_tolerates_nil_field_array(self): + def test_sync_tolerates_nil_field_array(self, stub_executor): # Second document's field-array came back nil. raw_result = [2, "product:1", ["title", "Laptop"], "product:2", None] - executor = _make_sync_executor(self._translated(), raw_result) + executor = stub_executor(self._translated(), raw_result) result = executor.execute("SELECT * FROM products") assert result.count == 2 assert result.rows == [{"title": "Laptop"}, {}] - async def test_async_tolerates_nil_field_array(self): + async def test_async_tolerates_nil_field_array(self, stub_async_executor): raw_result = [2, "product:1", ["title", "Laptop"], "product:2", None] - executor = _make_async_executor(self._translated(), raw_result) + executor = stub_async_executor(self._translated(), raw_result) result = await executor.execute("SELECT * FROM products") @@ -88,7 +58,7 @@ def _translated(self) -> TranslatedQuery: score_alias="score", ) - def test_sync_keeps_score_when_fields_nil(self): + def test_sync_keeps_score_when_fields_nil(self, stub_executor): raw_result = [ 2, "product:1", @@ -98,7 +68,7 @@ def test_sync_keeps_score_when_fields_nil(self): "0.9", None, ] - executor = _make_sync_executor(self._translated(), raw_result) + executor = stub_executor(self._translated(), raw_result) result = executor.execute("SELECT * FROM products") @@ -108,7 +78,7 @@ def test_sync_keeps_score_when_fields_nil(self): {"score": "0.9"}, ] - async def test_async_keeps_score_when_fields_nil(self): + async def test_async_keeps_score_when_fields_nil(self, stub_async_executor): raw_result = [ 2, "product:1", @@ -118,7 +88,7 @@ async def test_async_keeps_score_when_fields_nil(self): "0.9", None, ] - executor = _make_async_executor(self._translated(), raw_result) + executor = stub_async_executor(self._translated(), raw_result) result = await executor.execute("SELECT * FROM products") @@ -139,18 +109,18 @@ def _translated(self) -> TranslatedQuery: query_string="*", ) - def test_sync_tolerates_nil_row(self): + def test_sync_tolerates_nil_row(self, stub_executor): raw_result = [2, ["category", "books"], None] - executor = _make_sync_executor(self._translated(), raw_result) + executor = stub_executor(self._translated(), raw_result) result = executor.execute("SELECT category FROM products GROUP BY category") assert result.count == 2 assert result.rows == [{"category": "books"}, {}] - async def test_async_tolerates_nil_row(self): + async def test_async_tolerates_nil_row(self, stub_async_executor): raw_result = [2, ["category", "books"], None] - executor = _make_async_executor(self._translated(), raw_result) + executor = stub_async_executor(self._translated(), raw_result) result = await executor.execute( "SELECT category FROM products GROUP BY category" diff --git a/tests/test_protocol_matrix.py b/tests/test_protocol_matrix.py new file mode 100644 index 0000000..6f5bfac --- /dev/null +++ b/tests/test_protocol_matrix.py @@ -0,0 +1,253 @@ +"""End-to-end tests across RESP protocols and redis-py client modes. + +``sql_redis`` reads raw FT.* replies, so the shape it parses is decided by the +negotiated wire protocol rather than by anything the library configures. Four +client modes matter, not two: + +* ``protocol=2`` - flat arrays, the historical behaviour +* ``protocol`` unset - RESP2 on redis-py 6.x/7.x, RESP3 from redis-py 8 +* ``protocol=3`` - RESP3 maps on every version +* ``legacy_responses=False`` - RESP3, redis-py 8 only + +These tests assert against a real server, because reply shapes are the subject +and a mock would encode the assumption under test. ``test_reply_shapes.py`` +pins the exact map literals; this module proves those are the shapes Redis +actually sends and that the rows come out identical either way. + +The fixtures here are deliberately local. Parametrizing the shared +``redis_client`` fixture in ``conftest.py`` would multiply thirteen other +modules, and three of them assert raw positional replies against that client. +""" + +import pytest +import redis +import redis.asyncio as async_redis + +from sql_redis import create_async_executor, create_executor + +INDEX = "protocol_matrix" +PREFIX = "pm:" + +ROWS = [ + ("pm:1", {"title": "gaming laptop", "category": "electronics", "price": 1499}), + ("pm:2", {"title": "laptop stand", "category": "office", "price": 45}), + ("pm:3", {"title": "sql cookbook", "category": "books", "price": 39}), + ("pm:4", {"title": "redis in action", "category": "books", "price": 32}), +] + +PLAIN_SQL = ( + f"SELECT title, price FROM {INDEX} WHERE category = 'books' ORDER BY price ASC" +) +AGGREGATE_SQL = f"SELECT category, COUNT(*) AS cnt, SUM(price) AS total FROM {INDEX} GROUP BY category" +SCORE_SQL = ( + f"SELECT title, score() AS relevance FROM {INDEX} WHERE fulltext(title, 'laptop')" +) + +_REDIS_PY_MAJOR = int(redis.__version__.split(".")[0]) + +CLIENT_MODES = [ + pytest.param({"protocol": 2}, id="protocol-2"), + pytest.param({}, id="default"), + pytest.param({"protocol": 3}, id="protocol-3"), + pytest.param( + {"legacy_responses": False}, + id="new-response-format", + marks=pytest.mark.skipif( + _REDIS_PY_MAJOR < 8, reason="legacy_responses requires redis-py 8" + ), + ), +] + + +@pytest.fixture(scope="module") +def endpoint(redis_container) -> tuple[str, int]: + """Host and port of the shared test container.""" + return ( + redis_container.get_container_host_ip(), + int(redis_container.get_exposed_port(6379)), + ) + + +@pytest.fixture(scope="module") +def matrix_index(endpoint) -> str: + """Create the index and load its documents over a plain RESP2 client. + + Setup deliberately does not depend on the protocol under test, so the data + is materialized once no matter how many client modes are exercised. + """ + host, port = endpoint + client = redis.Redis(host=host, port=port, protocol=2, decode_responses=True) + try: + client.execute_command("FT.DROPINDEX", INDEX, "DD") + except redis.ResponseError: + pass + client.execute_command( + "FT.CREATE", + INDEX, + "ON", + "HASH", + "PREFIX", + "1", + PREFIX, + "SCHEMA", + "title", + "TEXT", + "SORTABLE", + "category", + "TAG", + "SORTABLE", + "price", + "NUMERIC", + "SORTABLE", + ) + for key, row in ROWS: + client.hset(key, mapping=row) + yield INDEX + try: + client.execute_command("FT.DROPINDEX", INDEX, "DD") + except redis.ResponseError: + pass + client.close() + + +def _client(endpoint, **kwargs) -> redis.Redis: + host, port = endpoint + return redis.Redis(host=host, port=port, decode_responses=True, **kwargs) + + +def _rows(endpoint, sql, **kwargs) -> tuple[list[dict], int]: + client = _client(endpoint, **kwargs) + try: + result = create_executor(client).execute(sql) + return result.rows, result.count + finally: + client.close() + + +class TestReplyShapeCanary: + """The protocols really do send different shapes. + + Without this, the redis-py compatibility CI job could report green having + never produced a map at all. + """ + + @pytest.mark.parametrize("protocol,expected", [(2, list), (3, dict)]) + @pytest.mark.parametrize("command", ["FT.SEARCH", "FT.AGGREGATE"]) + def test_raw_reply_type_follows_protocol( + self, endpoint, matrix_index, protocol, expected, command + ): + client = _client(endpoint, protocol=protocol) + try: + if command == "FT.SEARCH": + raw = client.execute_command( + command, matrix_index, "*", "LIMIT", "0", "2" + ) + else: + raw = client.execute_command( + command, matrix_index, "*", "GROUPBY", "1", "@category" + ) + finally: + client.close() + + assert isinstance(raw, expected) + + +class TestRowsMatchAcrossClientModes: + """Every client mode yields the rows the protocol=2 baseline yields.""" + + @pytest.mark.parametrize("client_kwargs", CLIENT_MODES) + def test_search_rows_match(self, endpoint, matrix_index, client_kwargs): + baseline, baseline_count = _rows(endpoint, PLAIN_SQL, protocol=2) + rows, count = _rows(endpoint, PLAIN_SQL, **client_kwargs) + + assert ( + rows + == baseline + == [ + {"title": "redis in action", "price": "32"}, + {"title": "sql cookbook", "price": "39"}, + ] + ) + assert count == baseline_count == 2 + + @pytest.mark.parametrize("client_kwargs", CLIENT_MODES) + def test_aggregate_rows_match(self, endpoint, matrix_index, client_kwargs): + baseline, baseline_count = _rows(endpoint, AGGREGATE_SQL, protocol=2) + rows, count = _rows(endpoint, AGGREGATE_SQL, **client_kwargs) + + by_category = {row["category"]: row for row in rows} + assert sorted(by_category) == ["books", "electronics", "office"] + assert by_category["books"] == {"category": "books", "cnt": "2", "total": "71"} + assert rows == baseline + assert count == baseline_count == 3 + + @pytest.mark.parametrize("client_kwargs", CLIENT_MODES) + def test_no_bookkeeping_columns_leak(self, endpoint, matrix_index, client_kwargs): + """RESP3 result maps carry id/values/extra_attributes; rows must not.""" + rows, _ = _rows(endpoint, PLAIN_SQL, **client_kwargs) + + for row in rows: + assert not {"id", "values", "extra_attributes", "payload"} & set(row) + + +class TestScoreValueType: + """The score's type differs by protocol, and that is accepted.""" + + def test_score_type_diverges_by_protocol(self, endpoint, matrix_index): + resp2_rows, _ = _rows(endpoint, SCORE_SQL, protocol=2) + resp3_rows, _ = _rows(endpoint, SCORE_SQL, protocol=3) + + assert isinstance(resp2_rows[0]["relevance"], str) + assert isinstance(resp3_rows[0]["relevance"], float) + # float() is the protocol-safe conversion at the call site. + assert [float(r["relevance"]) for r in resp2_rows] == [ + float(r["relevance"]) for r in resp3_rows + ] + assert [r["title"] for r in resp2_rows] == [r["title"] for r in resp3_rows] + + +class TestDecodeResponsesOff: + """A bytes-keyed RESP3 map is the shape that broke redis-py itself.""" + + def test_bytes_keys_still_yield_rows(self, endpoint, matrix_index): + host, port = endpoint + client = redis.Redis(host=host, port=port, protocol=3) + try: + result = create_executor(client).execute(PLAIN_SQL) + finally: + client.close() + + assert result.count == 2 + assert result.rows == [ + {b"title": b"redis in action", b"price": b"32"}, + {b"title": b"sql cookbook", b"price": b"39"}, + ] + + +class TestAsyncAcrossProtocols: + """The async executor has its own parse call site.""" + + @pytest.mark.parametrize("protocol", [2, 3]) + async def test_async_search_and_aggregate(self, endpoint, matrix_index, protocol): + host, port = endpoint + client = async_redis.Redis( + host=host, port=port, protocol=protocol, decode_responses=True + ) + try: + executor = await create_async_executor(client) + search = await executor.execute(PLAIN_SQL) + aggregate = await executor.execute(AGGREGATE_SQL) + finally: + await client.aclose() + + assert search.count == 2 + assert search.rows == [ + {"title": "redis in action", "price": "32"}, + {"title": "sql cookbook", "price": "39"}, + ] + assert aggregate.count == 3 + assert {row["category"] for row in aggregate.rows} == { + "books", + "electronics", + "office", + } diff --git a/tests/test_reply_shapes.py b/tests/test_reply_shapes.py new file mode 100644 index 0000000..09ce903 --- /dev/null +++ b/tests/test_reply_shapes.py @@ -0,0 +1,360 @@ +"""Reply-shape tests for RESP2 arrays and RESP3 maps. + +``sql_redis`` sends FT.* commands with ``client.execute_command`` and reads the +raw reply, because redis-py registers no FT.SEARCH / FT.AGGREGATE callback on +the base client. The reply shape therefore follows the negotiated protocol: a +flat array on RESP2, a map on RESP3. The executor folds the map back into the +array shape (``_resp3_to_resp2``) so a single parser handles both. + +Every canned reply below was captured from a real Redis 8.4.6 server rather +than hand-written, with the producing cell named in a comment. Reply shapes are +the whole subject here, so an invented literal would only encode the assumption +under test. Replies marked "derived" are a captured reply with one documented +mutation, to reach a case the server does not produce on demand. + +The nil-field-array cases for the RESP2 array shape live in +``test_nil_field_array.py``. +""" + +import pytest + +from sql_redis.translator import TranslatedQuery + +# Captured replies. +# +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT title, price FROM verify_products WHERE category = 'books' +# ORDER BY price ASC LIMIT 10 +RESP3_SEARCH = { + "attributes": [], + "format": "STRING", + "results": [ + { + "id": "vp:5", + "extra_attributes": {"price": "32", "title": "redis in action"}, + "values": [], + }, + { + "id": "vp:4", + "extra_attributes": {"price": "39", "title": "sql cookbook"}, + "values": [], + }, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=False, redis:8.4 +# Same query. Structural keys are bytes; this is the shape that made redis-py's +# own Result.from_resp3 return zero rows (redis-py#4107). +RESP3_SEARCH_BYTES = { + b"attributes": [], + b"format": b"STRING", + b"results": [ + { + b"id": b"vp:5", + b"extra_attributes": {b"price": b"32", b"title": b"redis in action"}, + b"values": [], + }, + { + b"id": b"vp:4", + b"extra_attributes": {b"price": b"39", b"title": b"sql cookbook"}, + b"values": [], + }, + ], + b"total_results": 2, + b"warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT title, score() AS relevance FROM verify_products +# WHERE fulltext(title, 'laptop') +RESP3_SEARCH_WITHSCORES = { + "attributes": [], + "format": "STRING", + "results": [ + { + "id": "vp:1", + "score": 0.714285696039395, + "extra_attributes": {"title": "gaming laptop"}, + "values": [], + }, + { + "id": "vp:3", + "score": 0.714285696039395, + "extra_attributes": {"title": "laptop stand"}, + "values": [], + }, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT score() AS relevance FROM verify_products WHERE fulltext(title, 'laptop') +# RETURN 0 suppresses document fields, so results carry no extra_attributes. +RESP3_SEARCH_SCORE_ONLY = { + "attributes": [], + "format": "STRING", + "results": [ + {"id": "vp:1", "score": 0.714285696039395, "values": []}, + {"id": "vp:3", "score": 0.714285696039395, "values": []}, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT category, COUNT(*) AS cnt, SUM(price) AS total, AVG(rating) AS avg_rating +# FROM verify_products GROUP BY category +# Aggregate rows carry only extra_attributes: no id, no score. +RESP3_AGGREGATE = { + "attributes": [], + "format": "STRING", + "results": [ + { + "extra_attributes": { + "category": "books", + "cnt": "2", + "total": "71", + "avg_rating": "4.7", + }, + "values": [], + }, + { + "extra_attributes": { + "category": "office", + "cnt": "2", + "total": "67", + "avg_rating": "3.95", + }, + "values": [], + }, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT title FROM verify_products WHERE category = 'nonexistent' +RESP3_SEARCH_EMPTY = { + "attributes": [], + "format": "STRING", + "results": [], + "total_results": 0, + "warning": [], +} + +# Derived from RESP3_SEARCH: extra_attributes dropped from the second result, +# the RESP3 analogue of the nil field-array of issue #38 (a document expiring +# between id selection and field materialization). +RESP3_SEARCH_MISSING_FIELDS = { + "attributes": [], + "format": "STRING", + "results": [ + { + "id": "vp:5", + "extra_attributes": {"price": "32", "title": "redis in action"}, + "values": [], + }, + {"id": "vp:4", "values": []}, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=2, decode_responses=True, redis:8.4 +# SELECT score() AS relevance FROM verify_products WHERE fulltext(title, 'laptop') +# The one array branch no other test covers, and the only guard on the parser +# having been moved into _parse_array_reply unchanged. +RESP2_SEARCH_SCORE_ONLY = [ + 2, + "vp:1", + "0.714285696039395", + "vp:3", + "0.714285696039395", +] + + +# Translations matching those captures. +def _search() -> TranslatedQuery: + return TranslatedQuery( + command="FT.SEARCH", + index="verify_products", + query_string="@category:{books}", + args=["RETURN", "2", "title", "price", "LIMIT", "0", "10", "DIALECT", "2"], + ) + + +def _search_withscores() -> TranslatedQuery: + return TranslatedQuery( + command="FT.SEARCH", + index="verify_products", + query_string="@title:laptop", + args=["RETURN", "1", "title", "WITHSCORES", "SCORER", "BM25", "DIALECT", "2"], + score_alias="relevance", + ) + + +def _search_score_only() -> TranslatedQuery: + return TranslatedQuery( + command="FT.SEARCH", + index="verify_products", + query_string="@title:laptop", + args=["RETURN", "0", "WITHSCORES", "SCORER", "BM25", "DIALECT", "2"], + score_alias="relevance", + ) + + +def _aggregate() -> TranslatedQuery: + return TranslatedQuery( + command="FT.AGGREGATE", + index="verify_products", + query_string="*", + args=["GROUPBY", "1", "@category", "REDUCE", "COUNT", "0", "AS", "cnt"], + ) + + +@pytest.fixture(params=["sync", "async"]) +def execute_reply(request, stub_executor, stub_async_executor): + """Run one SQL query through either executor against a canned reply. + + Parametrized so each test is written once and asserted for both the sync + and async executors, whose parse paths are separate call sites. + """ + flavour = request.param + + async def _run(translated: TranslatedQuery, raw_result): + sql = f"SELECT * FROM {translated.index}" + if flavour == "sync": + return stub_executor(translated, raw_result).execute(sql) + return await stub_async_executor(translated, raw_result).execute(sql) + + return _run + + +class TestSearchResp3Map: + """A RESP3 FT.SEARCH map parses to the same rows as the RESP2 array.""" + + async def test_standard_search_map_reply_yields_field_rows(self, execute_reply): + result = await execute_reply(_search(), RESP3_SEARCH) + + assert result.count == 2 + assert result.rows == [ + {"price": "32", "title": "redis in action"}, + {"price": "39", "title": "sql cookbook"}, + ] + + async def test_structural_keys_may_be_bytes(self, execute_reply): + """A decode_responses=False client sends bytes structural keys. + + Document field keys stay bytes, matching the RESP2 path; looking up + ``extra_attributes`` as str alone would yield silently empty rows. + """ + result = await execute_reply(_search(), RESP3_SEARCH_BYTES) + + assert result.count == 2 + assert result.rows == [ + {b"price": b"32", b"title": b"redis in action"}, + {b"price": b"39", b"title": b"sql cookbook"}, + ] + + async def test_bookkeeping_keys_do_not_leak_into_rows(self, execute_reply): + """``id``, ``values`` and ``extra_attributes`` are not columns.""" + result = await execute_reply(_search(), RESP3_SEARCH) + + for row in result.rows: + assert not {"id", "values", "extra_attributes", "payload"} & set(row) + + async def test_missing_extra_attributes_yields_empty_row(self, execute_reply): + result = await execute_reply(_search(), RESP3_SEARCH_MISSING_FIELDS) + + assert result.count == 2 + assert result.rows == [{"price": "32", "title": "redis in action"}, {}] + + async def test_empty_results_yields_no_rows(self, execute_reply): + result = await execute_reply(_search(), RESP3_SEARCH_EMPTY) + + assert result.count == 0 + assert result.rows == [] + + +class TestSearchWithScoresResp3Map: + """The score comes from the result's own ``score`` key, not a stride.""" + + async def test_withscores_map_reply_adds_score_column(self, execute_reply): + result = await execute_reply(_search_withscores(), RESP3_SEARCH_WITHSCORES) + + assert result.count == 2 + assert result.rows == [ + {"title": "gaming laptop", "relevance": 0.714285696039395}, + {"title": "laptop stand", "relevance": 0.714285696039395}, + ] + + async def test_score_stays_a_float(self, execute_reply): + """RESP3 scores are doubles and are passed through, not stringified. + + RESP2 delivers the score as text, so the type differs by protocol. The + library does not normalise it: ``float()`` is the protocol-safe + conversion at the call site. See docs/concepts/result-shape.md. + """ + result = await execute_reply(_search_withscores(), RESP3_SEARCH_WITHSCORES) + + assert isinstance(result.rows[0]["relevance"], float) + + async def test_score_alias_collision_is_resolved(self, execute_reply): + """An alias colliding with a document field is renamed, as on RESP2.""" + translated = _search_withscores() + translated.score_alias = "title" + + result = await execute_reply(translated, RESP3_SEARCH_WITHSCORES) + + assert result.rows[0]["title"] == "gaming laptop" + assert result.rows[0]["__score_title"] == 0.714285696039395 + + async def test_withscores_return0_map_reply_yields_score_only_rows( + self, execute_reply + ): + result = await execute_reply(_search_score_only(), RESP3_SEARCH_SCORE_ONLY) + + assert result.count == 2 + assert result.rows == [ + {"relevance": 0.714285696039395}, + {"relevance": 0.714285696039395}, + ] + + +class TestAggregateResp3Map: + """A RESP3 FT.AGGREGATE map parses to rows of reduced values.""" + + async def test_aggregate_map_reply_yields_rows(self, execute_reply): + result = await execute_reply(_aggregate(), RESP3_AGGREGATE) + + assert result.count == 2 + assert result.rows == [ + {"category": "books", "cnt": "2", "total": "71", "avg_rating": "4.7"}, + {"category": "office", "cnt": "2", "total": "67", "avg_rating": "3.95"}, + ] + + async def test_aggregate_count_comes_from_total_results(self, execute_reply): + """RESP3 ``total_results`` matches the RESP2 leading count. + + Measured against Redis 8.4.6: a GROUP BY returning N groups reports + ``total_results: N`` on RESP3 and ``N`` at position 0 on RESP2. + """ + result = await execute_reply(_aggregate(), RESP3_AGGREGATE) + + assert result.count == len(result.rows) == 2 + + +class TestResp2ArrayUnchanged: + """The array path still parses the shape it always did.""" + + async def test_withscores_return0_array_reply(self, execute_reply): + result = await execute_reply(_search_score_only(), RESP2_SEARCH_SCORE_ONLY) + + assert result.count == 2 + assert result.rows == [ + {"relevance": "0.714285696039395"}, + {"relevance": "0.714285696039395"}, + ] From 2033a258c964258862bb764cbf2e805faff14bfd Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Wed, 2 Sep 2026 17:30:42 +0200 Subject: [PATCH 2/2] fix(executor): correct the aggregate count claim and harden reply dispatch Six-perspective review of the RESP3 change. Addresses what it found. Measured, not assumed: RESP2 and RESP3 do NOT agree on QueryResult.count for an FT.AGGREGATE without GROUPBY. A computed-field projection over 5 documents reports count=1 on RESP2, the placeholder the Redis docs call "not a valid value", and count=5 on RESP3. With a GROUP BY the two do agree. The previous docs asserted agreement unconditionally, and the tests happened to use a GROUP BY whose group count equalled its row count, so nothing caught it. Neither value can be derived from the other without changing protocol=2 output, so both are surfaced as Redis sent them, now documented and pinned by tests on both the stub and live paths. The pre-existing claim that the aggregate count equals the row count after LIMIT was also false and is corrected. Reply dispatch now rejects what it cannot parse. A map without a 'results' key (a cluster node-keyed reply, an FT.PROFILE reply) folded to zero rows, and a non-array non-map reply reached positional indexing. Both were silent wrong answers of exactly the kind the new FAILURE_MODES entry warns about; both now raise with the offending shape named. The reply layout is derived once. _fold_map_to_array and _parse_array_reply each re-derived with_scores and has_fields from the translation, and had to agree or the parser would stride over slots the fold never emitted, mis-attributing one document's data to another. A frozen _ReplyLayout is now computed once in _parse_reply and passed to both. Unreachable today, since RETURN 0 is only ever emitted alongside WITHSCORES, but the invariant protecting it lived three modules away. count no longer differs from the hybrid parser when total_results is present but nil, and _field_array accepts the flat form its FT.HYBRID counterpart already accepted. _ScoreParseMixin is gone. Every method was a staticmethod called through the explicit class name, so it offered the syntax of inheritance with none of its semantics, and only two of eight methods concerned scores. The parsers are module-level functions, which is what the new helpers already were; the mixed self./class-name style went with it. Tests: count provenance is now asserted where total_results differs from the row count, which a surviving mutant showed nothing checked. Added the aggregate count divergence, shape rejection, the error re-wrapping paths (the async copy was the last real coverage hole, 16 lines), the aggregate missing-fields branch, and a canary for what an unset protocol negotiates per redis-py major, which a doc claim rested on. Removed two tests subsumed by exact-equality assertions and replaced a bookkeeping denylist with a column allowlist, so a new RESP3 result key fails a test instead of appearing in every row. Unordered aggregate rows are compared by group key rather than positionally, and score parity uses approx. executor.py coverage 90% on main, now 99%. The redis-py-compat job selects by a registered 'protocol' marker rather than a filename list, so a new protocol-sensitive test is picked up without editing the workflow. Docs: two result-shape examples showed a bytes key with a str value, which no client produces; RETURN 0 keeps its Factor 4 heading so the published anchor still resolves, with the protocol material demoted to a note; a stale pointer into Executor.execute and the schema docstrings that conflated redis-py 8 with RESP3 are fixed; the paragraph triplicated across FAILURE_MODES, a docstring and a test module now lives in one place. Verified again after the restructure: 30/30 harness cells across redis-py {6.3.0, 7.4.0, 8.1.0} and five client modes still match the protocol=2 baseline. Suite 677 passed, 3 skipped. Marked subset: 148 on 6.3.0 and 7.4.0, 151 on 8.1.0. --- .github/workflows/test.yml | 12 +- docs/concepts/index.md | 2 +- docs/concepts/result-shape.md | 38 +-- docs/concepts/search-vs-aggregate.md | 2 +- docs/for-ais-only/BUILD_AND_TEST.md | 19 +- docs/for-ais-only/FAILURE_MODES.md | 4 +- docs/for-ais-only/REPOSITORY_MAP.md | 9 +- docs/for-ais-only/index.md | 2 +- pyproject.toml | 3 + sql_redis/executor.py | 486 ++++++++++++++------------- sql_redis/schema.py | 4 +- tests/conftest.py | 19 +- tests/test_ft_hybrid.py | 2 + tests/test_nil_field_array.py | 4 + tests/test_protocol_matrix.py | 73 +++- tests/test_reply_shapes.py | 244 ++++++++++++-- tests/test_schema_registry.py | 2 + 17 files changed, 612 insertions(+), 313 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index cce719d..c275ca9 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -88,13 +88,11 @@ jobs: sys.exit(0 if redis.__version__ == '${{ matrix.redis-py }}' else 1)" # Reply shapes depend on the redis-py version and the negotiated RESP - # protocol, so these files are run against each supported redis-py. + # protocol, so the protocol-marked tests run against each supported + # redis-py. Selecting by marker rather than by filename means a new + # protocol-sensitive test is picked up here without editing this job. # 'uv run --with' layers an ephemeral overlay and never writes uv.lock. - name: Run reply-shape and protocol tests run: > - uv run --with 'redis==${{ matrix.redis-py }}' python -m pytest - tests/test_reply_shapes.py - tests/test_protocol_matrix.py - tests/test_nil_field_array.py - tests/test_ft_hybrid.py - -q + uv run --with 'redis==${{ matrix.redis-py }}' + python -m pytest -m protocol -q diff --git a/docs/concepts/index.md b/docs/concepts/index.md index 118a4fd..e64291b 100644 --- a/docs/concepts/index.md +++ b/docs/concepts/index.md @@ -60,7 +60,7 @@ Foundational reading for sql-redis. Each page explains a single design choice or --- - What `QueryResult.rows` actually contains, why it varies with the command, scoring, and client decoding. + What `QueryResult.rows` actually contains, why it varies with the command, scoring, client decoding and the wire protocol. - :material-test-tube:{ .lg .middle } **[Testing philosophy](testing-philosophy.md)** diff --git a/docs/concepts/result-shape.md b/docs/concepts/result-shape.md index a166705..9fd133e 100644 --- a/docs/concepts/result-shape.md +++ b/docs/concepts/result-shape.md @@ -4,14 +4,16 @@ Every `Executor.execute()` call returns a `QueryResult` with two attributes: `ro ## What `count` means -The integer in `count` is the total Redis reported for the query: position 0 of a RESP2 array reply, or `total_results` in a RESP3 map reply. The two agree. Its meaning depends on which command ran (see [FT.SEARCH vs FT.AGGREGATE](search-vs-aggregate.md)): +`count` is whatever Redis reported for the query: position 0 of a RESP2 array reply, or `total_results` in a RESP3 map reply. Its meaning depends on which command ran (see [FT.SEARCH vs FT.AGGREGATE](search-vs-aggregate.md)): -- **`FT.SEARCH`** (no aggregation): `count` is the **total number of matching documents** in the index, regardless of `LIMIT`. So `count` can be much larger than `len(rows)`. This is useful for pagination. -- **`FT.AGGREGATE`** (any aggregation, GROUP BY, computed field, date function): `count` is the **number of rows in the reply**, which is what you got back. After `LIMIT`, the two are equal. +- **`FT.SEARCH`** (no aggregation): `count` is the **total number of matching documents**, regardless of `LIMIT`. So `count` can be much larger than `len(rows)`, which is what makes it useful for pagination. Both protocols report the same figure. +- **`FT.AGGREGATE`** (any aggregation, GROUP BY, computed field, date function): `count` is **not the number of rows you got back**, and it is the one value that differs by protocol. With a `GROUP BY`, both protocols report the number of groups before `LIMIT`. Without one, a computed-field projection leaves RESP2 reporting `1` (a placeholder the [Redis docs](https://redis.io/docs/latest/commands/ft.aggregate/) call "not a valid value") while RESP3 reports the real figure. Use `len(rows)` if you want the rows you received. + +Neither aggregate value can be derived from the other without changing what `protocol=2` callers already get, so the library reports what Redis sent. ## What `rows` looks like -A row is always a `dict`, but the *keys* and *values* depend on five orthogonal factors. +A row is always a `dict`, but the *keys* and *values* depend on four orthogonal factors. ### Factor 1: client decoding @@ -50,37 +52,31 @@ If your SELECT contains `score()`, the underlying `FT.SEARCH` runs with `WITHSCO ```python # SELECT title, score() AS relevance FROM products WHERE fulltext(...) -{b"title": b"gaming laptop", b"relevance": "0.5"} +{b"title": b"gaming laptop", b"relevance": b"0.5"} # RESP2 +{b"title": b"gaming laptop", b"relevance": 0.5} # RESP3 ``` The score column name is whatever alias you wrote (`score() AS relevance` produces `relevance`). If you used `score()` without `AS`, the library falls back to a stable internal name. The score's value type depends on the wire protocol: `str` (or `bytes`, depending on decoding) on RESP2, and `float` on RESP3, where Redis sends it as a double. `float()` is the protocol-safe conversion at the call site and works on both. -### Factor 4: wire protocol - -Redis replies to `FT.SEARCH` and `FT.AGGREGATE` with a flat array on RESP2 and -a map on RESP3, and `sql_redis` reads the raw reply, so the protocol your client -negotiated decides which shape arrives. The library parses both, and the rows -come out the same. Note that redis-py 8 defaults to RESP3, where earlier -versions defaulted to RESP2, so this can change under you on an upgrade without -you passing `protocol` at all. - -One value is visibly protocol-dependent: the score, described above. Redis sends -it as text on RESP2 and as a double on RESP3, and the library passes it through -rather than normalising it. - -### Factor 5: `RETURN 0` +### Factor 4: `RETURN 0` If your SELECT is just `score()` with no document fields, the underlying command uses `RETURN 0` (no document fields returned). Each row contains only the score column: ```python # SELECT score() AS s FROM products WHERE fulltext(title, 'laptop') -{b"s": "0.5"} +{b"s": b"0.5"} ``` This is mostly useful when you want a relevance-only ranking without paying to ship the document body back. +## A note on the wire protocol + +Redis replies to `FT.SEARCH` and `FT.AGGREGATE` with a flat array on RESP2 and a map on RESP3, and `sql_redis` reads the raw reply, so the protocol your client negotiated decides which shape arrives. The library parses both and the rows come out the same, but two values differ: the score's type (Factor 3) and the aggregate `count` described above. + +Worth knowing because redis-py 8 defaults to RESP3 where earlier versions defaulted to RESP2, so an upgrade can change both without you passing `protocol` at all. + ## Score-column collision avoidance What happens if your document has a field literally named `__score`? The library detects the collision and renames the score column. The deterministic fallback is `__score_`. So: @@ -94,7 +90,7 @@ You will not normally see this; it is a defensive measure for the unusual case. It would be tempting to normalise everything: always strings, always `float` scores, always uniform shape. The library does not, for three reasons. -1. **Bytes is the right default for Redis.** Some fields legitimately contain binary data. Force-decoding to UTF-8 would corrupt those. The same argument covers the RESP3 score: stringifying a double cannot reproduce the text Redis would have sent on RESP2 (`repr(1.0)` is `"1.0"`, Redis writes `"1"`), so normalising would invent a third form rather than deliver parity. +1. **Bytes is the right default for Redis.** Some fields legitimately contain binary data. Force-decoding to UTF-8 would corrupt those. The same argument covers the RESP3 score: stringifying a double cannot reproduce the text Redis would have sent on RESP2, so normalising would invent a third form rather than deliver parity. 2. **The shape difference between `FT.SEARCH` and `FT.AGGREGATE` is intrinsic to the Redis commands.** Hiding it would lie about what is happening. 3. **One-call-site logic is cheap; library-wide policy is expensive.** A user who wants a uniform shape can write a 5-line decoder once, customised to their data. The library shipping that decoder for everyone is the wrong place for the policy. diff --git a/docs/concepts/search-vs-aggregate.md b/docs/concepts/search-vs-aggregate.md index eb26477..ee8e82d 100644 --- a/docs/concepts/search-vs-aggregate.md +++ b/docs/concepts/search-vs-aggregate.md @@ -39,6 +39,6 @@ These constraints are not bugs to fix; they are the cost of the abstraction. The Two practical consequences: 1. **The result-row shape changes.** `FT.SEARCH` returns rows with field-value pairs straight from the indexed documents, possibly with a score column. `FT.AGGREGATE` returns rows of computed fields, group keys, and reduced values; the original document fields are present only if the SQL asks for them. See [Result shape](result-shape.md). -2. **`LIMIT` semantics differ subtly.** Both commands honour `LIMIT`, but the `count` returned by `FT.AGGREGATE` reflects the post-pipeline row count, while `FT.SEARCH`'s `count` is the total match count regardless of the limit. The library exposes both as `QueryResult.count`; the meaning depends on which path ran. +2. **`LIMIT` semantics differ subtly.** Both commands honour `LIMIT`, but `FT.SEARCH`'s `count` is the total match count regardless of the limit, while `FT.AGGREGATE`'s is neither the match count nor the row count, and is the one value that differs between RESP2 and RESP3. The library exposes both as `QueryResult.count`; see [Result shape](result-shape.md#what-count-means) before relying on it. If you need to know which command was issued for a given SQL, call `Translator.translate(sql)` directly and inspect `TranslatedQuery.command`. diff --git a/docs/for-ais-only/BUILD_AND_TEST.md b/docs/for-ais-only/BUILD_AND_TEST.md index 638eded..ba2512f 100644 --- a/docs/for-ais-only/BUILD_AND_TEST.md +++ b/docs/for-ais-only/BUILD_AND_TEST.md @@ -62,7 +62,11 @@ uv run --group docs mkdocs build --strict ## CI gates (target state) -- `make check` (lint + tests) on every PR. +- `make check` (lint + tests) on every PR, across Python 3.9 to 3.13. +- `pytest -m protocol` against redis-py 6.3.0, 7.4.0 and 8.1.0 (the + `redis-py-compat` job). Mark any test whose result depends on the RESP + protocol or the redis-py version with `@pytest.mark.protocol` so it runs + there. - `make docs-build` with `-W` on every PR. - 100% coverage gate. @@ -76,7 +80,18 @@ uv run pytest tests/test_parser.py tests/test_translator.py -x -vv These do not need Redis and run in well under a second. -When changing executor logic: +When changing reply parsing: + +``` +uv run pytest tests/test_reply_shapes.py tests/test_nil_field_array.py -x -vv +``` + +These feed canned replies through the parser with no server, so they are the +fastest signal on a parsing change. Follow with +`tests/test_protocol_matrix.py`, which asserts the same behaviour against a +real server on both protocols. + +When changing other executor logic: ``` uv run pytest tests/test_executor.py tests/test_sql_queries.py -x -vv diff --git a/docs/for-ais-only/FAILURE_MODES.md b/docs/for-ais-only/FAILURE_MODES.md index f6e35b3..f4cd74c 100644 --- a/docs/for-ais-only/FAILURE_MODES.md +++ b/docs/for-ais-only/FAILURE_MODES.md @@ -22,8 +22,8 @@ the base client. `Search.search()` calls `execute_command` and then its own array on RESP2, a map on RESP3. `_parse_reply` therefore dispatches on the shape of the reply -(`isinstance(raw_result, dict)`), and `_resp3_to_resp2` folds a map back into -the array shape so one parser handles both. Do not "simplify" this into a +(`isinstance(raw_result, dict)`), and `_fold_map_to_array` folds a map back +into the array shape so one parser handles both. Do not "simplify" this into a single unconditional path, and do not replace the shape check with a protocol check on the connection: sniffing `connection_kwargs["protocol"]` is wrong for cluster and sentinel clients, and it cannot distinguish redis-py 8's four diff --git a/docs/for-ais-only/REPOSITORY_MAP.md b/docs/for-ais-only/REPOSITORY_MAP.md index 3cc5ca6..504e7f7 100644 --- a/docs/for-ais-only/REPOSITORY_MAP.md +++ b/docs/for-ais-only/REPOSITORY_MAP.md @@ -69,7 +69,7 @@ tests/ | `IS NULL`, `IS NOT NULL`, `exists()` | `parser.py`, `query_builder.py` | | Parameter substitution | `executor.py::_substitute_params` | | FT.SEARCH vs FT.AGGREGATE branching | `translator.py::translate_parsed` (the `use_aggregate` boolean) | -| Result-row parsing | `executor.py::_ScoreParseMixin` (`_parse_reply` dispatches by reply shape; `_parse_array_reply` holds the four branches: WITHSCORES + RETURN 0, WITHSCORES, plain SEARCH, AGGREGATE; `_resp3_to_resp2` folds a RESP3 map into the array shape) | +| Result-row parsing | `executor.py::_parse_reply` (dispatches by reply shape), `_parse_array_reply` (the four branches: WITHSCORES + RETURN 0, WITHSCORES, plain SEARCH, AGGREGATE), `_fold_map_to_array` (folds a RESP3 map into the array shape) | ## What to read before changing X @@ -79,9 +79,10 @@ tests/ - **Analyzer/query-builder changes.** Read `analyzer.py` and the matching branch in `query_builder.py`. The two are tightly coupled: a new field type in the analyzer needs a corresponding emitter in the builder. -- **Executor changes.** The sync and async executors share `_ScoreParseMixin` - and the result-parsing branches. If you change the response shape, both - need updates. +- **Executor changes.** The sync and async executors share the module-level + reply parsers, so a parsing change lands once. Their `execute()` bodies are + still near-identical above the `_parse_reply` call, so a change there needs + applying twice. - **Schema registry changes.** The two registry classes do not share code, but they share semantics. Lazy loading, invalidation, and load coalescing must be consistent across sync and async. diff --git a/docs/for-ais-only/index.md b/docs/for-ais-only/index.md index c25fc29..b8d1372 100644 --- a/docs/for-ais-only/index.md +++ b/docs/for-ais-only/index.md @@ -26,7 +26,7 @@ Use this to find the right starting point fast. | Change schema loading semantics | [Schema-aware translation](../concepts/schema-aware-translation.md), then `schema.py` | | Change async behavior | [Async invariants](../concepts/async-invariants.md), then `executor.py::AsyncExecutor` and `schema.py::AsyncSchemaRegistry` | | Change the FT.SEARCH/FT.AGGREGATE branching | [FT.SEARCH vs FT.AGGREGATE](../concepts/search-vs-aggregate.md), then `translator.py::translate_parsed` | -| Change the result-row shape | [Result shape](../concepts/result-shape.md), then `executor.py` (`Executor.execute` parsing branches) | +| Change the result-row shape | [Result shape](../concepts/result-shape.md), then `executor.py::_parse_reply` and the functions below it | | Run tests | [Build and test](BUILD_AND_TEST.md) | | Diagnose "this looks broken" | [Failure modes](FAILURE_MODES.md) before assuming a bug | diff --git a/pyproject.toml b/pyproject.toml index db0a99e..f3803df 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -82,6 +82,9 @@ exclude = ''' [tool.pytest.ini_options] log_cli = true asyncio_mode = "auto" +markers = [ + "protocol: reply shapes depend on the RESP protocol and the redis-py version (run by the redis-py-compat CI job)", +] [tool.mypy] warn_unused_configs = true diff --git a/sql_redis/executor.py b/sql_redis/executor.py index 9040f77..65e52a3 100644 --- a/sql_redis/executor.py +++ b/sql_redis/executor.py @@ -121,20 +121,19 @@ def _map_get(mapping: dict, key: str, default: Any = None) -> Any: return mapping.get(key.encode(), default) -def _extra_attributes(result_item: Any) -> list: - """Flatten a RESP3 result's ``extra_attributes`` map into a field array. +def _field_array(fields: Any) -> list: + """Return a document's fields as a flat ``[field, value, ...]`` array. - Returns the ``[field, value, field, value, ...]`` shape the RESP2 parser - expects. An absent or non-map ``extra_attributes`` (RETURN 0 suppresses it, - and a document expiring mid-query can null it) becomes an empty array, - matching the RESP2 path's tolerance for a nil field-array. + Accepts the RESP3 map form and the RESP2 flat form, mirroring + ``_parse_hybrid_reply``. A nil or unrecognized value becomes an empty + array, which is the same tolerance the array parser already applies to a + nil field-array (issue #38). """ - if not isinstance(result_item, dict): - return [] - fields = _map_get(result_item, "extra_attributes") - if not isinstance(fields, dict): - return [] - return [item for pair in fields.items() for item in pair] + if isinstance(fields, dict): + return [item for pair in fields.items() for item in pair] + if isinstance(fields, (list, tuple)): + return list(fields) + return [] @dataclass @@ -145,227 +144,248 @@ class QueryResult: count: int -class _ScoreParseMixin: - """Shared helpers for score-related response parsing.""" +def _is_unknown_command(error_msg: str) -> bool: + """Return True when a ResponseError means the command is unsupported.""" + lowered = error_msg.lower() + return "unknown command" in lowered or "unknown subcommand" in lowered - @staticmethod - def _is_unknown_command(error_msg: str) -> bool: - """Return True when a ResponseError means the command is unsupported.""" - lowered = error_msg.lower() - return "unknown command" in lowered or "unknown subcommand" in lowered - @staticmethod - def _parse_hybrid_reply(raw_result) -> tuple[Any, list[dict]]: - """Parse an FT.HYBRID reply into (count, rows). +def _inject_vector_param(cmd: list[str | bytes], vector_param: bytes) -> None: + """Replace the vector PARAMS value with the actual bytes, in place. - FT.HYBRID does not use the FT.AGGREGATE array shape. The reply is a map - ``{total_results: N, results: [...], warnings: [...], ...}`` that arrives - either as a dict (redis-py 8.x / RESP3) or as a flat list - (``[total_results, N, results, [...], ...]``) on RESP2. Each result row - is likewise a dict or a flat ``[field, val, ...]`` list. Keys/values may - be bytes or str depending on the client's decode_responses setting. - """ - if isinstance(raw_result, dict): - reply = raw_result - else: - reply = dict(zip(raw_result[::2], raw_result[1::2])) - - count = _map_get(reply, "total_results") or 0 - results = _map_get(reply, "results") or [] - rows = [ - dict(row) if isinstance(row, dict) else dict(zip(row[::2], row[1::2])) - for row in results - ] - return count, rows - - @staticmethod - def _inject_vector_param(cmd: list[str | bytes], vector_param: bytes) -> None: - """Replace the vector PARAMS value with the actual bytes, in place. - - Only the ``$vector`` token in the PARAMS value position (the one - preceded by the param name ``vector``) is replaced. Query-side - references to ``$vector`` (FT.SEARCH KNN expressions, FT.HYBRID VSIM) - must stay as parameter references so Redis resolves them from PARAMS. - """ - for i, arg in enumerate(cmd): - if arg == "$vector" and i > 0 and cmd[i - 1] == "vector": - cmd[i] = vector_param + Only the ``$vector`` token in the PARAMS value position (the one + preceded by the param name ``vector``) is replaced. Query-side + references to ``$vector`` (FT.SEARCH KNN expressions, FT.HYBRID VSIM) + must stay as parameter references so Redis resolves them from PARAMS. + """ + for i, arg in enumerate(cmd): + if arg == "$vector" and i > 0 and cmd[i - 1] == "vector": + cmd[i] = vector_param + + +def _has_return_0(args: list[str]) -> bool: + """Return True when the args contain 'RETURN 0' (no document fields).""" + try: + idx = args.index("RETURN") + return args[idx + 1] == "0" + except (ValueError, IndexError): + return False + + +def _resolve_score_alias( + score_alias: str | None, + args: list[str], + first_row_fields: set[str] | None = None, +) -> str: + """Determine a stable score column name that won't collide with + document fields. The alias is resolved once and reused for every + row so all rows share the same column name. + + When a RETURN clause is present, the returned field names are used + for collision detection. When RETURN is absent (SELECT *), the + caller should pass ``first_row_fields`` — the union of all field + names across all result rows — so we can detect collisions even + when different documents have different field sets.""" + alias = score_alias or "__score" + # Extract RETURN field names from args to detect collision + try: + idx = args.index("RETURN") + count = int(args[idx + 1]) + return_fields = set(args[idx + 2 : idx + 2 + count]) + except (ValueError, IndexError): + # Normalize bytes keys to str so collision detection works + # regardless of decode_responses setting. + raw = first_row_fields or set() + return_fields = {k.decode() if isinstance(k, bytes) else k for k in raw} + while alias in return_fields: + alias = f"__score_{alias}" + return alias + + +def _parse_hybrid_reply(raw_result) -> tuple[Any, list[dict]]: + """Parse an FT.HYBRID reply into (count, rows). + + FT.HYBRID does not use the FT.AGGREGATE array shape. The reply is a map + ``{total_results: N, results: [...], warnings: [...], ...}`` that arrives + either as a dict (RESP3) or as a flat list + (``[total_results, N, results, [...], ...]``) on RESP2. Each result row is + likewise a dict of *fields* or a flat ``[field, val, ...]`` list, with no + ``extra_attributes`` nesting: unlike FT.SEARCH, a hybrid row's own keys are + the field names. Keys and values may be bytes or str depending on the + client's decode_responses setting. + """ + if isinstance(raw_result, dict): + reply = raw_result + else: + reply = dict(zip(raw_result[::2], raw_result[1::2])) + + count = _map_get(reply, "total_results") or 0 + results = _map_get(reply, "results") or [] + rows = [ + dict(row) if isinstance(row, dict) else dict(zip(row[::2], row[1::2])) + for row in results + ] + return count, rows + + +@dataclass(frozen=True) +class _ReplyLayout: + """How one reply is laid out, derived once per query. + + The array parser strides over the reply, so it and the RESP3 fold have to + agree on which slots each document occupies. Deriving that here and passing + it to both is what keeps them from drifting apart. + """ - @staticmethod - def _has_return_0(args: list[str]) -> bool: - """Return True when the args contain 'RETURN 0' (no document fields).""" - try: - idx = args.index("RETURN") - return args[idx + 1] == "0" - except (ValueError, IndexError): - return False - - @staticmethod - def _resolve_score_alias( - score_alias: str | None, - args: list[str], - first_row_fields: set[str] | None = None, - ) -> str: - """Determine a stable score column name that won't collide with - document fields. The alias is resolved once and reused for every - row so all rows share the same column name. - - When a RETURN clause is present, the returned field names are used - for collision detection. When RETURN is absent (SELECT *), the - caller should pass ``first_row_fields`` — the union of all field - names across all result rows — so we can detect collisions even - when different documents have different field sets.""" - alias = score_alias or "__score" - # Extract RETURN field names from args to detect collision - try: - idx = args.index("RETURN") - count = int(args[idx + 1]) - return_fields = set(args[idx + 2 : idx + 2 + count]) - except (ValueError, IndexError): - # Normalize bytes keys to str so collision detection works - # regardless of decode_responses setting. - raw = first_row_fields or set() - return_fields = {k.decode() if isinstance(k, bytes) else k for k in raw} - while alias in return_fields: - alias = f"__score_{alias}" - return alias - - @staticmethod - def _resp3_to_resp2(reply: dict, translated: TranslatedQuery) -> list: - """Rewrite a RESP3 FT.SEARCH / FT.AGGREGATE map as the RESP2 array. - - A RESP3 reply is a map:: - - {total_results: N, attributes: [...], format: ..., warning: [...], - results: [{id: key, score: 0.5, extra_attributes: {field: val}, - values: [...]}, ...]} - - Rather than parse that shape separately, fold it back into the flat - array RESP2 sends so a single parser handles both wire protocols. The - emitted layout mirrors the branch the array parser will take: the - document key, then the score when WITHSCORES was requested, then the - field array unless RETURN 0 suppressed document fields. ``values`` is - never read, and ``id`` lands in the position the parser discards. - - Values are moved, not converted, so a RESP3 score stays a float and - document field keys keep whatever type the client returned. - """ - response: list = [_map_get(reply, "total_results", 0)] - results = _map_get(reply, "results") or [] + is_search: bool + with_scores: bool + has_fields: bool + score_alias: str | None + args: list[str] + + @classmethod + def of(cls, translated: TranslatedQuery) -> "_ReplyLayout": + return cls( + is_search=translated.command == "FT.SEARCH", + # The explicit score_alias signal, rather than scanning args for + # the literal token "WITHSCORES", which could false-positive if a + # returned field happened to be named "WITHSCORES". + with_scores=translated.score_alias is not None, + # RETURN 0 suppresses document fields (like NOCONTENT), so each + # document arrives without a field array. + has_fields=not _has_return_0(translated.args), + score_alias=translated.score_alias, + args=translated.args, + ) + + +def _fold_map_to_array(reply: dict, layout: _ReplyLayout) -> list: + """Fold a RESP3 FT.SEARCH / FT.AGGREGATE map into the RESP2 array shape. + + A RESP3 reply is a map:: + + {total_results: N, attributes: [...], format: ..., warning: [...], + results: [{id: key, score: 0.5, extra_attributes: {field: val}, + values: [...]}, ...]} + + The output is only meaningful as input to ``_parse_array_reply`` under the + same ``layout``: it emits the document key, the score when WITHSCORES was + requested, and the field array unless RETURN 0 suppressed document fields, + which is exactly the slot layout that parser will stride over. ``id`` lands + in the position the parser discards; ``values``, ``attributes``, ``format`` + and ``warning`` are not read. + + Values are moved, not converted, so a RESP3 score stays a float and + document field keys keep whatever type the client returned. + """ + response: list = [_map_get(reply, "total_results") or 0] + results = _map_get(reply, "results") or [] - if translated.command != "FT.SEARCH": + for item in results: + fields = _map_get(item, "extra_attributes") if isinstance(item, dict) else None + if not layout.is_search: # FT.AGGREGATE rows carry only extra_attributes: no id, no score. - response.extend(_extra_attributes(item) for item in results) - return response - - with_scores = translated.score_alias is not None - no_content = _ScoreParseMixin._has_return_0(translated.args) - for item in results: - response.append(_map_get(item, "id", "") if isinstance(item, dict) else "") - if with_scores: - response.append( - _map_get(item, "score") if isinstance(item, dict) else None - ) - if not no_content: - response.append(_extra_attributes(item)) - return response - - @staticmethod - def _parse_array_reply( - raw_result: Any, translated: TranslatedQuery - ) -> tuple[Any, list[dict]]: - """Parse the flat-array FT.SEARCH / FT.AGGREGATE reply into rows. - - This is the RESP2 wire shape, and the shape ``_resp3_to_resp2`` folds a - RESP3 map into, so it is the only row-assembly path in the executor. - """ - count = raw_result[0] if isinstance(raw_result, list) and raw_result else 0 - rows: list[dict] = [] - - if translated.command == "FT.SEARCH": - # Use the explicit score_alias signal rather than scanning args - # for the literal token "WITHSCORES", which could false-positive - # if a returned field happened to be named "WITHSCORES". - with_scores = translated.score_alias is not None - # RETURN 0 suppresses document fields (like NOCONTENT); - # with WITHSCORES the reply is [count, id, score, id, score, ...] - no_content = _ScoreParseMixin._has_return_0(translated.args) - - if with_scores and no_content: - # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...] - # Stride of 2: key, score (no field array) - score_alias = _ScoreParseMixin._resolve_score_alias( - translated.score_alias, translated.args - ) - for i in range(1, len(raw_result) - 1, 2): - score = raw_result[i + 1] - row = {score_alias: score} - rows.append(row) - elif with_scores: - # WITHSCORES format: [count, key1, score1, [fields1], key2, score2, [fields2], ...] - # Stride of 3: key, score, field_list - # First pass: collect all field names across all rows so the - # alias avoids collisions with any document field, not just - # the first row's fields. - all_field_names: set[str] = set() - parsed_rows: list[tuple[dict, Any]] = [] - for i in range(1, len(raw_result) - 2, 3): - score = raw_result[i + 1] - # A nil field-array (e.g. doc expired mid-query) becomes an - # empty field set, keeping the row's score instead of crashing. - row_data = raw_result[i + 2] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - all_field_names.update(row.keys()) - parsed_rows.append((row, score)) - resolved_alias = _ScoreParseMixin._resolve_score_alias( - translated.score_alias, - translated.args, - first_row_fields=all_field_names, - ) - for row, score in parsed_rows: - row[resolved_alias] = score - rows.append(row) - else: - # Standard format: [count, key1, [fields1], key2, [fields2], ...] - for i in range(2, len(raw_result), 2): - row_data = raw_result[i] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - else: - # FT.AGGREGATE format: [count, [fields1], [fields2], ...] - for row_data in raw_result[1:]: - row_data = row_data or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - + response.append(_field_array(fields)) + continue + response.append(_map_get(item, "id", "") if isinstance(item, dict) else "") + if layout.with_scores: + response.append(_map_get(item, "score") if isinstance(item, dict) else None) + if layout.has_fields: + response.append(_field_array(fields)) + return response + + +def _parse_array_reply( + raw_result: list, layout: _ReplyLayout +) -> tuple[Any, list[dict]]: + """Parse the flat-array FT.SEARCH / FT.AGGREGATE reply into rows. + + This is the RESP2 wire shape, and the shape ``_fold_map_to_array`` folds a + RESP3 map into. + """ + count = raw_result[0] if raw_result else 0 + rows: list[dict] = [] + + if not layout.is_search: + # FT.AGGREGATE format: [count, [fields1], [fields2], ...] + for row_data in raw_result[1:]: + row_data = row_data or [] + rows.append(dict(zip(row_data[::2], row_data[1::2]))) return count, rows - @staticmethod - def _parse_reply(raw_result: Any, translated: TranslatedQuery) -> QueryResult: - """Turn a raw FT.* reply into a QueryResult, RESP2 or RESP3. - - Dispatch is by reply *shape*, not by asking the connection which - protocol it negotiated. redis-py registers no FT.SEARCH / FT.AGGREGATE - callback on the base client: ``Search.search()`` calls - ``execute_command`` and then its own ``_parse_results``, so a raw - ``execute_command`` receives the wire shape verbatim in every redis-py - version, a flat array on RESP2 and a map on RESP3. Shape dispatch needs no - version check, works for cluster and sentinel clients, and survives a - future redis-py that hands back the flat array on RESP3 too. - - FT.HYBRID keeps its own parser: its RESP3 rows are flat field maps, - not the ``{id, extra_attributes}`` documents FT.SEARCH returns. - """ - if translated.command == "FT.HYBRID": - count, rows = _ScoreParseMixin._parse_hybrid_reply(raw_result) - else: - if isinstance(raw_result, dict): - raw_result = _ScoreParseMixin._resp3_to_resp2(raw_result, translated) - count, rows = _ScoreParseMixin._parse_array_reply(raw_result, translated) + if layout.with_scores and not layout.has_fields: + # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...] + # Stride of 2: key, score (no field array) + score_alias = _resolve_score_alias(layout.score_alias, layout.args) + for i in range(1, len(raw_result) - 1, 2): + rows.append({score_alias: raw_result[i + 1]}) + elif layout.with_scores: + # WITHSCORES: [count, key1, score1, [fields1], key2, score2, ...] + # Stride of 3: key, score, field_list + # First pass: collect all field names across all rows so the alias + # avoids collisions with any document field, not just the first row's. + all_field_names: set[str] = set() + parsed_rows: list[tuple[dict, Any]] = [] + for i in range(1, len(raw_result) - 2, 3): + score = raw_result[i + 1] + # A nil field-array (e.g. doc expired mid-query) becomes an empty + # field set, keeping the row's score instead of crashing. + row_data = raw_result[i + 2] or [] + row = dict(zip(row_data[::2], row_data[1::2])) + all_field_names.update(row.keys()) + parsed_rows.append((row, score)) + resolved_alias = _resolve_score_alias( + layout.score_alias, layout.args, first_row_fields=all_field_names + ) + for row, score in parsed_rows: + row[resolved_alias] = score + rows.append(row) + else: + # Standard format: [count, key1, [fields1], key2, [fields2], ...] + for i in range(2, len(raw_result), 2): + row_data = raw_result[i] or [] + rows.append(dict(zip(row_data[::2], row_data[1::2]))) + + return count, rows + + +def _parse_reply(raw_result: Any, translated: TranslatedQuery) -> QueryResult: + """Turn a raw FT.* reply into a QueryResult, RESP2 or RESP3. + + Dispatch is by reply shape rather than by the protocol the connection + negotiated; see ``docs/for-ais-only/FAILURE_MODES.md`` for why. A reply + that is neither shape is rejected rather than parsed into empty rows, + because reading a map that is not a result set (a cluster node-keyed reply, + FT.PROFILE) or an array that is not one (a WITHCURSOR pair) would otherwise + silently return nothing. + + FT.HYBRID keeps its own parser: its RESP3 rows are flat field maps, not the + ``{id, extra_attributes}`` documents FT.SEARCH returns. + """ + if translated.command == "FT.HYBRID": + count, rows = _parse_hybrid_reply(raw_result) return QueryResult(rows=rows, count=count) + layout = _ReplyLayout.of(translated) + if isinstance(raw_result, dict): + if _map_get(raw_result, "results") is None: + raise ValueError( + f"Unrecognized {translated.command} reply: a map without a " + f"'results' key. Got keys {sorted(map(str, raw_result))}." + ) + raw_result = _fold_map_to_array(raw_result, layout) + elif not isinstance(raw_result, list): + raise ValueError( + f"Unrecognized {translated.command} reply of type " + f"{type(raw_result).__name__}; expected an array (RESP2) or a map " + "(RESP3)." + ) + + count, rows = _parse_array_reply(raw_result, layout) + return QueryResult(rows=rows, count=count) -class Executor(_ScoreParseMixin): + +class Executor: """Executes SQL queries against Redis.""" def __init__(self, client: redis.Redis, schema_registry: SchemaRegistry) -> None: @@ -398,16 +418,14 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: # Replace the $vector PARAMS value with actual bytes (query/VSIM # references to $vector stay as parameter references). if vector_param: - self._inject_vector_param(cmd, vector_param) + _inject_vector_param(cmd, vector_param) # Execute command try: raw_result = self._client.execute_command(*cmd) except redis.ResponseError as e: error_msg = str(e) - if translated.command == "FT.HYBRID" and self._is_unknown_command( - error_msg - ): + if translated.command == "FT.HYBRID" and _is_unknown_command(error_msg): raise redis.ResponseError( f"{error_msg}. hybrid_vector_search() translates to FT.HYBRID, " "which requires Redis 8.4+ (RediSearch with hybrid search) " @@ -430,12 +448,10 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: ) from e raise - # Parse the reply. Shape dispatch happens in _parse_reply: RESP2 sends - # a flat array, RESP3 sends a map, and both reach the same row parser. - return self._parse_reply(raw_result, translated) + return _parse_reply(raw_result, translated) -class AsyncExecutor(_ScoreParseMixin): +class AsyncExecutor: """Async version of Executor for use with redis.asyncio clients.""" def __init__( @@ -482,16 +498,14 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: # Replace the $vector PARAMS value with actual bytes (query/VSIM # references to $vector stay as parameter references). if vector_param: - self._inject_vector_param(cmd, vector_param) + _inject_vector_param(cmd, vector_param) # Execute command asynchronously try: raw_result = await self._client.execute_command(*cmd) except redis.ResponseError as e: error_msg = str(e) - if translated.command == "FT.HYBRID" and self._is_unknown_command( - error_msg - ): + if translated.command == "FT.HYBRID" and _is_unknown_command(error_msg): raise redis.ResponseError( f"{error_msg}. hybrid_vector_search() translates to FT.HYBRID, " "which requires Redis 8.4+ (RediSearch with hybrid search) " @@ -514,9 +528,7 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: ) from e raise - # Parse the reply. Shape dispatch happens in _parse_reply: RESP2 sends - # a flat array, RESP3 sends a map, and both reach the same row parser. - return self._parse_reply(raw_result, translated) + return _parse_reply(raw_result, translated) def create_executor( diff --git a/sql_redis/schema.py b/sql_redis/schema.py index 1d89f31..14402c5 100644 --- a/sql_redis/schema.py +++ b/sql_redis/schema.py @@ -20,7 +20,7 @@ def _extract_attributes(info: Any) -> list: """Pull the ``attributes`` section out of an FT.INFO reply. Handles both reply shapes: the RESP2 flat list - (``[..., 'attributes', [...], ...]``) and the redis-py 8.x / RESP3 map + (``[..., 'attributes', [...], ...]``) and the RESP3 map (``{b'attributes': [...], ...}``), whose keys may be bytes or str. """ if isinstance(info, dict): @@ -66,7 +66,7 @@ def _parse_schema_from_info(info: Any) -> dict[str, str]: This is a pure function with no I/O operations, shared by both the sync and async schema registries. It accepts both the RESP2 list reply and the - redis-py 8.x / RESP3 map reply (see ``_extract_attributes``). + RESP3 map reply (see ``_extract_attributes``). Args: info: The raw response from the FT.INFO command (list or dict). diff --git a/tests/conftest.py b/tests/conftest.py index 677d116..d20414f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -368,12 +368,16 @@ def make_stub_executor(translated: TranslatedQuery, raw_result) -> Executor: """Build an Executor whose translation and Redis reply are stubbed. Feeds a crafted reply straight through the parser, so every parse branch - can be exercised deterministically without a live Redis server. Exposed as - the ``stub_executor`` fixture. + can be exercised deterministically without a live Redis server. Pass an + exception as ``raw_result`` to exercise the error paths. Exposed as the + ``stub_executor`` fixture. """ executor = Executor.__new__(Executor) executor._client = MagicMock() - executor._client.execute_command.return_value = raw_result + if isinstance(raw_result, BaseException): + executor._client.execute_command.side_effect = raw_result + else: + executor._client.execute_command.return_value = raw_result executor._schema_registry = MagicMock() executor._translator = MagicMock() executor._translator.translate.return_value = translated @@ -385,12 +389,15 @@ def make_stub_async_executor(translated: TranslatedQuery, raw_result) -> AsyncEx The async seam differs from the sync one: ``AsyncExecutor.execute`` parses first, awaits ``ensure_schema`` when the query names an index, and only then - translates the pre-parsed result. Exposed as the ``stub_async_executor`` - fixture. + translates the pre-parsed result. Pass an exception as ``raw_result`` to + exercise the error paths. Exposed as the ``stub_async_executor`` fixture. """ executor = AsyncExecutor.__new__(AsyncExecutor) executor._client = MagicMock() - executor._client.execute_command = AsyncMock(return_value=raw_result) + if isinstance(raw_result, BaseException): + executor._client.execute_command = AsyncMock(side_effect=raw_result) + else: + executor._client.execute_command = AsyncMock(return_value=raw_result) executor._schema_registry = MagicMock() executor._schema_registry.ensure_schema = AsyncMock() executor._translator = MagicMock() diff --git a/tests/test_ft_hybrid.py b/tests/test_ft_hybrid.py index 46b3a1d..ca3db45 100644 --- a/tests/test_ft_hybrid.py +++ b/tests/test_ft_hybrid.py @@ -48,6 +48,8 @@ from sql_redis.schema import SchemaRegistry from sql_redis.translator import Translator +pytestmark = pytest.mark.protocol + def float_vector_to_bytes(vector: list[float]) -> bytes: """Convert a list of floats to binary format for Redis vector storage.""" diff --git a/tests/test_nil_field_array.py b/tests/test_nil_field_array.py index 87e5994..5e3b9da 100644 --- a/tests/test_nil_field_array.py +++ b/tests/test_nil_field_array.py @@ -14,8 +14,12 @@ equivalents of these cases are in ``test_reply_shapes.py``. """ +import pytest + from sql_redis.translator import TranslatedQuery +pytestmark = pytest.mark.protocol + class TestStandardSearchNilFields: """Standard FT.SEARCH reply: [count, key1, [fields1], key2, [fields2], ...].""" diff --git a/tests/test_protocol_matrix.py b/tests/test_protocol_matrix.py index 6f5bfac..a52a1df 100644 --- a/tests/test_protocol_matrix.py +++ b/tests/test_protocol_matrix.py @@ -25,6 +25,8 @@ from sql_redis import create_async_executor, create_executor +pytestmark = pytest.mark.protocol + INDEX = "protocol_matrix" PREFIX = "pm:" @@ -42,6 +44,8 @@ SCORE_SQL = ( f"SELECT title, score() AS relevance FROM {INDEX} WHERE fulltext(title, 'laptop')" ) +LIMITED_SQL = f"SELECT title FROM {INDEX} ORDER BY price ASC LIMIT 2" +COMPUTED_SQL = f"SELECT price * 2 AS double_price FROM {INDEX}" _REDIS_PY_MAJOR = int(redis.__version__.split(".")[0]) @@ -151,6 +155,22 @@ def test_raw_reply_type_follows_protocol( assert isinstance(raw, expected) + def test_unset_protocol_follows_redis_py_default(self, endpoint, matrix_index): + """redis-py 8 defaults to RESP3; 6.x and 7.x default to RESP2. + + This is the claim docs/concepts/result-shape.md makes about an upgrade + changing the reply shape without the caller passing ``protocol``. + """ + client = _client(endpoint) + try: + raw = client.execute_command( + "FT.SEARCH", matrix_index, "*", "LIMIT", "0", "2" + ) + finally: + client.close() + + assert isinstance(raw, dict if _REDIS_PY_MAJOR >= 8 else list) + class TestRowsMatchAcrossClientModes: """Every client mode yields the rows the protocol=2 baseline yields.""" @@ -175,19 +195,23 @@ def test_aggregate_rows_match(self, endpoint, matrix_index, client_kwargs): baseline, baseline_count = _rows(endpoint, AGGREGATE_SQL, protocol=2) rows, count = _rows(endpoint, AGGREGATE_SQL, **client_kwargs) + # RediSearch does not guarantee group order without a SORTBY, so + # compare by group key rather than positionally. by_category = {row["category"]: row for row in rows} + assert by_category == {row["category"]: row for row in baseline} assert sorted(by_category) == ["books", "electronics", "office"] assert by_category["books"] == {"category": "books", "cnt": "2", "total": "71"} - assert rows == baseline assert count == baseline_count == 3 @pytest.mark.parametrize("client_kwargs", CLIENT_MODES) - def test_no_bookkeeping_columns_leak(self, endpoint, matrix_index, client_kwargs): - """RESP3 result maps carry id/values/extra_attributes; rows must not.""" - rows, _ = _rows(endpoint, PLAIN_SQL, **client_kwargs) + def test_count_is_the_total_not_the_row_count( + self, endpoint, matrix_index, client_kwargs + ): + """LIMIT reduces the rows, not the count callers paginate on.""" + rows, count = _rows(endpoint, LIMITED_SQL, **client_kwargs) - for row in rows: - assert not {"id", "values", "extra_attributes", "payload"} & set(row) + assert count == 4 + assert [row["title"] for row in rows] == ["redis in action", "sql cookbook"] class TestScoreValueType: @@ -200,10 +224,39 @@ def test_score_type_diverges_by_protocol(self, endpoint, matrix_index): assert isinstance(resp2_rows[0]["relevance"], str) assert isinstance(resp3_rows[0]["relevance"], float) # float() is the protocol-safe conversion at the call site. - assert [float(r["relevance"]) for r in resp2_rows] == [ - float(r["relevance"]) for r in resp3_rows - ] - assert [r["title"] for r in resp2_rows] == [r["title"] for r in resp3_rows] + assert [float(r["relevance"]) for r in resp2_rows] == pytest.approx( + [float(r["relevance"]) for r in resp3_rows] + ) + assert {r["title"] for r in resp2_rows} == {r["title"] for r in resp3_rows} + + +class TestAggregateCountDivergesByProtocol: + """``count`` is protocol-dependent for an FT.AGGREGATE without GROUPBY. + + A computed-field projection produces a LOAD/APPLY-only pipeline, where + RESP2 leaves the leading integer at the placeholder the Redis docs call + "not a valid value" while RESP3 reports the real figure in + ``total_results``. Neither can be derived from the other without changing + the protocol=2 output callers already depend on, so both are surfaced as + Redis sent them. Documented in docs/concepts/result-shape.md. + """ + + def test_placeholder_on_resp2_real_total_on_resp3(self, endpoint, matrix_index): + resp2_rows, resp2_count = _rows(endpoint, COMPUTED_SQL, protocol=2) + resp3_rows, resp3_count = _rows(endpoint, COMPUTED_SQL, protocol=3) + + assert len(resp2_rows) == len(resp3_rows) == 4 + assert resp2_count == 1 + assert resp3_count == 4 + + def test_rows_still_match(self, endpoint, matrix_index): + """Only the count differs. The rows are identical.""" + resp2_rows, _ = _rows(endpoint, COMPUTED_SQL, protocol=2) + resp3_rows, _ = _rows(endpoint, COMPUTED_SQL, protocol=3) + + assert sorted(r["double_price"] for r in resp2_rows) == sorted( + r["double_price"] for r in resp3_rows + ) class TestDecodeResponsesOff: diff --git a/tests/test_reply_shapes.py b/tests/test_reply_shapes.py index 09ce903..94097c6 100644 --- a/tests/test_reply_shapes.py +++ b/tests/test_reply_shapes.py @@ -1,10 +1,8 @@ """Reply-shape tests for RESP2 arrays and RESP3 maps. -``sql_redis`` sends FT.* commands with ``client.execute_command`` and reads the -raw reply, because redis-py registers no FT.SEARCH / FT.AGGREGATE callback on -the base client. The reply shape therefore follows the negotiated protocol: a -flat array on RESP2, a map on RESP3. The executor folds the map back into the -array shape (``_resp3_to_resp2``) so a single parser handles both. +The reply shape follows the protocol the client negotiated: a flat array on +RESP2, a map on RESP3. See ``docs/for-ais-only/FAILURE_MODES.md`` for why the +executor sees the raw shape at all. Every canned reply below was captured from a real Redis 8.4.6 server rather than hand-written, with the producing cell named in a comment. Reply shapes are @@ -12,11 +10,14 @@ under test. Replies marked "derived" are a captured reply with one documented mutation, to reach a case the server does not produce on demand. -The nil-field-array cases for the RESP2 array shape live in -``test_nil_field_array.py``. +``test_protocol_matrix.py`` proves these are the shapes Redis really sends; the +nil-field-array cases for the array shape live in ``test_nil_field_array.py``. """ import pytest +import redis + +pytestmark = pytest.mark.protocol from sql_redis.translator import TranslatedQuery @@ -163,10 +164,56 @@ "warning": [], } +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT title FROM cnt_idx ORDER BY price ASC LIMIT 2, over 5 matching documents. +# total_results is the total match count, not the number of rows returned, so +# this is the capture that catches count being derived from len(results). +RESP3_SEARCH_LIMITED = { + "attributes": [], + "format": "STRING", + "results": [ + {"id": "ci:0", "extra_attributes": {"title": "book 0"}, "values": []}, + {"id": "ci:1", "extra_attributes": {"title": "book 1"}, "values": []}, + ], + "total_results": 5, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=2, decode_responses=True, redis:8.4 +# Same query. RESP2 puts the same total at position 0. +RESP2_SEARCH_LIMITED = [ + 5, + "ci:0", + ["title", "book 0"], + "ci:1", + ["title", "book 1"], +] + +# Cells: redis-py 8.1.0, protocol=2 and protocol=3, decode_responses=True, +# redis:8.4. SELECT price * 2 AS dbl FROM cnt_idx, which translates to +# FT.AGGREGATE ... APPLY with no GROUPBY. RESP2 sends the placeholder 1 that +# the Redis docs disclaim as "not a valid value"; RESP3 sends the real count. +# See TestAggregateCountDivergesByProtocol. +RESP2_AGGREGATE_COMPUTED = [ + 1, + ["price", "10", "dbl", "20"], + ["price", "11", "dbl", "22"], +] + +RESP3_AGGREGATE_COMPUTED = { + "attributes": [], + "format": "STRING", + "results": [ + {"extra_attributes": {"price": "10", "dbl": "20"}, "values": []}, + {"extra_attributes": {"price": "11", "dbl": "22"}, "values": []}, + ], + "total_results": 2, + "warning": [], +} + # Cell: redis-py 8.1.0, protocol=2, decode_responses=True, redis:8.4 # SELECT score() AS relevance FROM verify_products WHERE fulltext(title, 'laptop') -# The one array branch no other test covers, and the only guard on the parser -# having been moved into _parse_array_reply unchanged. +# The one array branch no other test covers. RESP2_SEARCH_SCORE_ONLY = [ 2, "vp:1", @@ -206,6 +253,24 @@ def _search_score_only() -> TranslatedQuery: ) +def _search_limited() -> TranslatedQuery: + return TranslatedQuery( + command="FT.SEARCH", + index="cnt_idx", + query_string="*", + args=["RETURN", "1", "title", "LIMIT", "0", "2", "DIALECT", "2"], + ) + + +def _aggregate_computed() -> TranslatedQuery: + return TranslatedQuery( + command="FT.AGGREGATE", + index="cnt_idx", + query_string="*", + args=["APPLY", "@price * 2", "AS", "dbl", "DIALECT", "2"], + ) + + def _aggregate() -> TranslatedQuery: return TranslatedQuery( command="FT.AGGREGATE", @@ -259,12 +324,28 @@ async def test_structural_keys_may_be_bytes(self, execute_reply): {b"price": b"39", b"title": b"sql cookbook"}, ] - async def test_bookkeeping_keys_do_not_leak_into_rows(self, execute_reply): - """``id``, ``values`` and ``extra_attributes`` are not columns.""" + async def test_only_document_fields_become_columns(self, execute_reply): + """A new RESP3 result key must not become a column. + + Asserted as an allowlist rather than a denylist of known bookkeeping + keys, so a key RediSearch adds later fails here instead of silently + appearing in every row. + """ result = await execute_reply(_search(), RESP3_SEARCH) for row in result.rows: - assert not {"id", "values", "extra_attributes", "payload"} & set(row) + assert set(row) == {"price", "title"} + + async def test_count_is_the_total_not_the_row_count(self, execute_reply): + """``count`` comes from ``total_results``, which LIMIT does not reduce. + + Callers paginate on this (docs/concepts/result-shape.md), so deriving + it from the number of returned results would be silently wrong. + """ + result = await execute_reply(_search_limited(), RESP3_SEARCH_LIMITED) + + assert result.count == 5 + assert len(result.rows) == 2 async def test_missing_extra_attributes_yields_empty_row(self, execute_reply): result = await execute_reply(_search(), RESP3_SEARCH_MISSING_FIELDS) @@ -272,6 +353,26 @@ async def test_missing_extra_attributes_yields_empty_row(self, execute_reply): assert result.count == 2 assert result.rows == [{"price": "32", "title": "redis in action"}, {}] + async def test_flat_field_array_is_tolerated(self, execute_reply): + """Defensive: a result whose fields arrive already flat still parses. + + No redis-py version sends this for FT.SEARCH, but ``_parse_hybrid_reply`` + accepts both forms for the same conceptual field, and a total + ``_field_array`` is what keeps an unexpected shape from silently + becoming an empty row. Derived from RESP3_SEARCH. + """ + reply = { + "attributes": [], + "format": "STRING", + "results": [{"id": "vp:5", "extra_attributes": ["title", "flat"]}], + "total_results": 1, + "warning": [], + } + + result = await execute_reply(_search(), reply) + + assert result.rows == [{"title": "flat"}] + async def test_empty_results_yields_no_rows(self, execute_reply): result = await execute_reply(_search(), RESP3_SEARCH_EMPTY) @@ -336,20 +437,58 @@ async def test_aggregate_map_reply_yields_rows(self, execute_reply): {"category": "office", "cnt": "2", "total": "67", "avg_rating": "3.95"}, ] - async def test_aggregate_count_comes_from_total_results(self, execute_reply): - """RESP3 ``total_results`` matches the RESP2 leading count. + async def test_missing_extra_attributes_yields_empty_row(self, execute_reply): + """The aggregate branch reaches the field flattener separately.""" + reply = { + "attributes": [], + "format": "STRING", + "results": [{"extra_attributes": {"category": "books"}}, {"values": []}], + "total_results": 2, + "warning": [], + } - Measured against Redis 8.4.6: a GROUP BY returning N groups reports - ``total_results: N`` on RESP3 and ``N`` at position 0 on RESP2. - """ - result = await execute_reply(_aggregate(), RESP3_AGGREGATE) + result = await execute_reply(_aggregate(), reply) - assert result.count == len(result.rows) == 2 + assert result.rows == [{"category": "books"}, {}] + + +class TestAggregateCountDivergesByProtocol: + """For FT.AGGREGATE, ``count`` is protocol-dependent, and that is accepted. + + An aggregate pipeline with no GROUPBY (a computed field, a date function, a + geo_distance projection) leaves RESP2's leading integer as the placeholder + the Redis docs call "not a valid value", while RESP3's ``total_results`` + reports the real figure. Measured on Redis 8.4.6: the same SQL over 5 + documents gives count=1 on RESP2 and count=5 on RESP3. + + Neither value can be turned into the other without changing the protocol=2 + behaviour callers already depend on, so both are surfaced as Redis sent + them. This test exists to stop someone "fixing" the difference by deriving + count from len(rows), which would silently change RESP2 output. + """ + + async def test_resp2_reports_the_placeholder(self, execute_reply): + result = await execute_reply(_aggregate_computed(), RESP2_AGGREGATE_COMPUTED) + + assert result.count == 1 + assert len(result.rows) == 2 + + async def test_resp3_reports_the_real_total(self, execute_reply): + result = await execute_reply(_aggregate_computed(), RESP3_AGGREGATE_COMPUTED) + + assert result.count == 2 + assert len(result.rows) == 2 class TestResp2ArrayUnchanged: """The array path still parses the shape it always did.""" + async def test_count_is_the_total_not_the_row_count(self, execute_reply): + result = await execute_reply(_search_limited(), RESP2_SEARCH_LIMITED) + + assert result.count == 5 + assert len(result.rows) == 2 + async def test_withscores_return0_array_reply(self, execute_reply): result = await execute_reply(_search_score_only(), RESP2_SEARCH_SCORE_ONLY) @@ -358,3 +497,70 @@ async def test_withscores_return0_array_reply(self, execute_reply): {"relevance": "0.714285696039395"}, {"relevance": "0.714285696039395"}, ] + + +class TestUnrecognizedReplies: + """A reply that is neither shape is rejected, not parsed into empty rows. + + A map that is not a result set (a cluster node-keyed reply, FT.PROFILE) or + an array that is not one (a WITHCURSOR pair) would otherwise fold to zero + rows and look like an empty result set. That silent-empty failure is the + one redis-py shipped in 8.0.0 (redis-py#4107). + """ + + async def test_map_without_results_raises(self, execute_reply): + node_keyed = {"127.0.0.1:6379": {"total_results": 2, "results": []}} + + with pytest.raises(ValueError, match="without a 'results' key"): + await execute_reply(_search(), node_keyed) + + async def test_profile_shaped_map_raises(self, execute_reply): + with pytest.raises(ValueError, match="without a 'results' key"): + await execute_reply(_search(), {"Results": {}, "Profile": {}}) + + async def test_non_array_non_map_reply_raises(self, execute_reply): + with pytest.raises(ValueError, match="Unrecognized FT.SEARCH reply"): + await execute_reply(_search(), "OK") + + async def test_empty_results_list_is_not_rejected(self, execute_reply): + """An empty result set is legitimate and must still parse.""" + result = await execute_reply(_search(), RESP3_SEARCH_EMPTY) + + assert result.count == 0 + assert result.rows == [] + + +class TestErrorRewrapping: + """Both executors re-wrap ResponseError with an actionable hint.""" + + async def test_ismissing_hint_is_added(self, execute_reply): + translated = TranslatedQuery( + command="FT.SEARCH", + index="verify_products", + query_string="ismissing(@email)", + args=["DIALECT", "2"], + ) + error = redis.ResponseError("Unknown function") + + with pytest.raises(redis.ResponseError, match="ismissing\\(\\) requires"): + await execute_reply(translated, error) + + async def test_unrelated_response_error_propagates_unchanged(self, execute_reply): + """Only the two known-cause errors are re-wrapped.""" + error = redis.ResponseError("no such index") + + with pytest.raises(redis.ResponseError, match="^no such index$"): + await execute_reply(_search(), error) + + async def test_hybrid_version_hint_is_added(self, execute_reply): + translated = TranslatedQuery( + command="FT.HYBRID", + index="verify_items", + query_string="", + args=["SEARCH", "@description:smartphone"], + is_hybrid=True, + ) + error = redis.ResponseError("unknown command 'FT.HYBRID'") + + with pytest.raises(redis.ResponseError, match="8.4"): + await execute_reply(translated, error) diff --git a/tests/test_schema_registry.py b/tests/test_schema_registry.py index 91683cc..2274d52 100644 --- a/tests/test_schema_registry.py +++ b/tests/test_schema_registry.py @@ -7,6 +7,8 @@ from sql_redis.schema import SchemaRegistry, _parse_schema_from_info +pytestmark = pytest.mark.protocol + def _create_test_indexes(redis_client: redis.Redis) -> list[str]: """Helper to create test indexes."""