From 76d08f3204635835efb349941bcf18b7e28bd4f5 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 15:24:38 +0300 Subject: [PATCH] perf: claim rows with ordered per-queue scans; add id to _pending_idx --- docs/adr/0003-fetch-query-shape.md | 18 ++++++ docs/introduction/how-it-works.md | 46 ++++++++------ docs/operations/alembic.md | 73 +++++++++++++++++++--- faststream_outbox/client.py | 98 ++++++++++++++++-------------- faststream_outbox/schema.py | 6 +- tests/test_client_contract.py | 19 ++++++ tests/test_integration.py | 74 +++++++++++++++++++++- tests/test_unit.py | 25 ++++---- 8 files changed, 267 insertions(+), 92 deletions(-) create mode 100644 docs/adr/0003-fetch-query-shape.md diff --git a/docs/adr/0003-fetch-query-shape.md b/docs/adr/0003-fetch-query-shape.md new file mode 100644 index 0000000..6d47177 --- /dev/null +++ b/docs/adr/0003-fetch-query-shape.md @@ -0,0 +1,18 @@ +# The fetch claims rows with ordered per-queue scans, not one filtered sort + +The fetch query takes each subscribed queue separately through `CROSS JOIN LATERAL` and runs two +branches per queue, unleased rows and expired leases. Each branch is an +`ORDER BY next_attempt_at, id LIMIT n FOR UPDATE SKIP LOCKED` scan, the branches are combined with +`UNION ALL`, and an outer `ORDER BY ... LIMIT n` picks the rows to claim. The unleased branch reads +`_pending_idx`, whose key is `(queue, next_attempt_at, id)`, so the scan stops after `n` +unlocked rows and a fetch costs the same with ten ready rows or a million. The earlier single query, +`WHERE queue = ANY(:queues) AND (acquired_token IS NULL OR )`, claimed the same rows in the +same order, but Postgres could serve neither the OR nor `= ANY` with an ordered index scan, and it +doesn't push the row limit through `LockRows`, so every fetch read and sorted the whole backlog: +about 200 ms per fetch at a million ready rows, against 0.1 ms for this shape, on PostgreSQL 13 and +17 alike. `id` is in the index because rows published in one transaction share `next_attempt_at`, +and without it a fetch reads the entire tie group. The costs accepted are a statement that is harder +to read, up to `2 × n` rows per queue locked briefly while only `n` are claimed (other fetchers skip +them until the claim commits, which affects fairness, not correctness), and a concurrent index +rebuild when upgrading from 0.14. The expired-lease branch still sorts every expired lease of its +queue; an index for it was left out until it is measured against a realistic number of hung handlers. diff --git a/docs/introduction/how-it-works.md b/docs/introduction/how-it-works.md index 6cd5651..47de409 100644 --- a/docs/introduction/how-it-works.md +++ b/docs/introduction/how-it-works.md @@ -63,37 +63,43 @@ bulk publish costs one NOTIFY, not one per row. Per subscriber, two loops run concurrently: -The fetch loop owns a long-lived `AsyncConnection` for the fetch CTE and -a separate raw asyncpg connection for `LISTEN outbox_
`. A single CTE +The fetch loop owns a long-lived `AsyncConnection` for the fetch query and +a separate raw asyncpg connection for `LISTEN outbox_
`. One statement claims rows: ```sql -WITH claimed AS ( - SELECT id FROM outbox - WHERE queue = :queue - AND next_attempt_at <= now() - AND ( - acquired_token IS NULL - OR acquired_at < now() - make_interval(secs => :lease_ttl) - ) - ORDER BY next_attempt_at, id - LIMIT :batch - FOR UPDATE SKIP LOCKED +WITH queues AS (SELECT DISTINCT unnest(:queues) AS name), +pending AS ( + SELECT p.id, p.next_attempt_at FROM queues JOIN LATERAL ( + SELECT id, next_attempt_at FROM outbox + WHERE queue = queues.name AND next_attempt_at <= now() + AND acquired_token IS NULL + ORDER BY next_attempt_at, id LIMIT :batch + FOR UPDATE SKIP LOCKED + ) p ON true +), +expired AS ( + -- the same per-queue scan for leases older than :lease_ttl +), +ready AS ( + SELECT id FROM (SELECT * FROM pending UNION ALL SELECT * FROM expired) c + ORDER BY next_attempt_at, id LIMIT :batch ) UPDATE outbox SET acquired_token = :uuid, acquired_at = now(), deliveries_count = deliveries_count + 1 -WHERE id IN (SELECT id FROM claimed) +WHERE id IN (SELECT id FROM ready) RETURNING * ``` -This is simplified for illustration. The real query writes each `OR` -disjunct with its own partial-index predicate spelled out as a conjunct, so -Postgres can use the `outbox_pending_idx` / `outbox_lease_idx` partial -indexes instead of a seq-scan. The naive `OR` above is the exact shape the -code avoids. +Each queue gets its own ordered scan of `outbox_pending_idx`, which stops +after `:batch` unlocked rows, so a fetch costs the same with ten ready rows +or a million. A single `WHERE ... OR ...` over `queue = ANY(:queues)` would +claim the same rows, but Postgres would have to read and sort the whole +backlog first. [ADR 0003](https://github.com/modern-python/faststream-outbox/blob/main/docs/adr/0003-fetch-query-shape.md) +records why the query has this shape. -The CTE reclaims both unleased rows AND rows whose lease has expired +The query reclaims both unleased rows AND rows whose lease has expired (`acquired_at < now() - lease_ttl_seconds`), so there is no separate stuck-row reaper. The idle-sleep is short-circuited by NOTIFY via an `asyncio.Event`, so idle dispatch latency drops from up to `max_fetch_interval` (default 10s) to diff --git a/docs/operations/alembic.md b/docs/operations/alembic.md index dd13ee6..01bb736 100644 --- a/docs/operations/alembic.md +++ b/docs/operations/alembic.md @@ -44,7 +44,7 @@ op.create_index( op.create_index( "outbox_pending_idx", "outbox", - ["queue", "next_attempt_at"], + ["queue", "next_attempt_at", "id"], unique=False, postgresql_where=sa.text("acquired_token IS NULL"), ) @@ -56,14 +56,15 @@ op.create_index( The three indexes carry load-bearing partial predicates: -- `outbox_pending_idx`: `(queue, next_attempt_at) WHERE acquired_token IS NULL`. - Branch A of the fetch CTE (unleased rows) is served by this index. The - WHERE clause is written so Postgres' planner recognizes the implied - predicate and uses the partial index; drop the predicate and the planner - falls back to a seq-scan as the table grows. +- `outbox_pending_idx`: `(queue, next_attempt_at, id) WHERE acquired_token IS NULL`. + The fetch's unleased branch reads this index in order, one queue at a + time, and stops after `limit` unlocked rows. `id` is in the key because + rows published in one transaction share `next_attempt_at`; without it a + fetch reads the whole tie group. Drop the predicate and the planner falls + back to a seq-scan as the table grows. - `outbox_lease_idx`: `(queue, acquired_at) WHERE acquired_token IS NOT NULL`. - Branch B of the fetch CTE (expired-lease reclaim) is served by this - index. Here too, the predicate is load-bearing for fetch performance. + The fetch's expired-lease branch is served by this index. Here too, the + predicate is load-bearing for fetch performance. - `outbox_timer_id_uq`: unique `(queue, timer_id) WHERE timer_id IS NOT NULL`. Backs the `timer_id` dedup contract via `pg_insert(...).on_conflict_do_nothing(...)`. Without the partial @@ -92,6 +93,60 @@ indexes are exactly what the broker depends on. When you wire the `/health` probe or CI gate, it fails when the live DB drifts from this declaration. (It is opt-in; it never runs at `broker.start()`.) +## Upgrading to 0.15: rebuild `outbox_pending_idx` { #upgrading-to-015 } + +0.15 claims rows with an ordered scan of `outbox_pending_idx` per queue +(see [ADR 0003](https://github.com/modern-python/faststream-outbox/blob/main/docs/adr/0003-fetch-query-shape.md)), +and the index gains `id` as its last column. A table migrated before 0.15 +still has `(queue, next_attempt_at)`, and `validate_schema()` reports it as +drift. + +Rebuild the index without blocking writes, and run the migration **before** +deploying 0.15. Earlier versions work on the new index, so the order is +safe; 0.15 on the old index works too, but each fetch then reads every row +that shares the oldest `next_attempt_at`. + +```python +import sqlalchemy as sa +from alembic import op + + +def upgrade() -> None: + with op.get_context().autocommit_block(): + op.create_index( + "outbox_pending_idx_new", + "outbox", + ["queue", "next_attempt_at", "id"], + postgresql_where=sa.text("acquired_token IS NULL"), + postgresql_concurrently=True, + ) + op.drop_index("outbox_pending_idx", table_name="outbox", postgresql_concurrently=True) + op.execute("ALTER INDEX outbox_pending_idx_new RENAME TO outbox_pending_idx") + + +def downgrade() -> None: + with op.get_context().autocommit_block(): + op.create_index( + "outbox_pending_idx_old", + "outbox", + ["queue", "next_attempt_at"], + postgresql_where=sa.text("acquired_token IS NULL"), + postgresql_concurrently=True, + ) + op.drop_index("outbox_pending_idx", table_name="outbox", postgresql_concurrently=True) + op.execute("ALTER INDEX outbox_pending_idx_old RENAME TO outbox_pending_idx") +``` + +`CONCURRENTLY` can't run inside a transaction, hence `autocommit_block()`. +For a table in a non-default schema, pass `schema=` to both index calls and +qualify the name in the `ALTER INDEX`. If a concurrent build fails, it +leaves an `INVALID` index behind: drop it and run the migration again. + +Don't apply what `alembic revision --autogenerate` produces for this +change. It drops and recreates the index in one transaction without +`CONCURRENTLY`, which blocks every publish and fetch on the table for the +whole build. + ## Adding the DLQ after the fact To opt into [Dead-letter queue](../usage/dlq.md) audit, add @@ -296,7 +351,7 @@ three indexes and their expected shape: | Index | Columns | Unique | `postgresql_where` | | --- | --- | --- | --- | -| `outbox_pending_idx` | `queue, next_attempt_at` | no | `acquired_token IS NULL` | +| `outbox_pending_idx` | `queue, next_attempt_at, id` | no | `acquired_token IS NULL` | | `outbox_lease_idx` | `queue, acquired_at` | no | `acquired_token IS NOT NULL` | | `outbox_timer_id_uq` | `queue, timer_id` | yes | `timer_id IS NOT NULL` | diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index bf69e7b..790e782 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -23,17 +23,17 @@ from sqlalchemy import ( ARRAY, Float, + Integer, String, - and_, - any_, bindparam, cast, delete, func, insert, - or_, select, text, + true, + union_all, update, ) from sqlalchemy.ext.asyncio import AsyncSession @@ -51,7 +51,7 @@ import typing from collections.abc import Mapping, Sequence - from sqlalchemy import Insert, Table + from sqlalchemy import CTE, ColumnElement, Insert, Table, Update from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine @@ -141,6 +141,50 @@ async def validate_schema(self, *, check_autovacuum: bool = False) -> None: ... async def ping(self) -> bool: ... +def _build_fetch_stmt(outbox: "Table") -> "Update": + """Build the claim statement once; queues, limit, lease TTL and the new token are bind parameters. + + Each queue gets two ``LATERAL`` branches, unleased rows and expired leases, and each branch is an + ordered ``LIMIT n FOR UPDATE SKIP LOCKED`` scan, so Postgres stops after ``n`` unlocked rows instead of + sorting the whole backlog. See ADR 0003 for why the query has this shape. + """ + t = outbox + limit = bindparam("limit", type_=Integer) + # ``make_interval(secs => :lease_ttl)`` keeps the cutoff server-side, immune to worker/DB clock skew. + lease_cutoff = func.now() - func.make_interval(0, 0, 0, 0, 0, 0, bindparam("lease_ttl", type_=Float)) + queue_array = ARRAY(String) + queue_name = func.unnest(cast(bindparam("queues", type_=queue_array), queue_array)).label("name") + queues = select(queue_name).distinct().cte("queues") + + def branch(name: str, *lease_state: "ColumnElement[bool]") -> "CTE": + per_queue = ( + select(t.c.id, t.c.next_attempt_at) + .where(t.c.queue == queues.c.name, t.c.next_attempt_at <= func.now(), *lease_state) + .order_by(t.c.next_attempt_at, t.c.id) + .limit(limit) + .with_for_update(skip_locked=True) + .lateral(f"{name}_rows") + ) + return select(per_queue.c.id, per_queue.c.next_attempt_at).select_from(queues.join(per_queue, true())).cte(name) + + pending = branch("pending", t.c.acquired_token.is_(None)) + expired = branch("expired", t.c.acquired_token.is_not(None), t.c.acquired_at < lease_cutoff) + candidates = union_all( + select(pending.c.id, pending.c.next_attempt_at), select(expired.c.id, expired.c.next_attempt_at) + ).subquery("candidates") + ready = select(candidates.c.id).order_by(candidates.c.next_attempt_at, candidates.c.id).limit(limit).cte("ready") + return ( + update(t) + .where(t.c.id.in_(select(ready.c.id))) + .values( + acquired_at=func.now(), + acquired_token=bindparam("token", type_=t.c.acquired_token.type), + deliveries_count=t.c.deliveries_count + 1, + ) + .returning(*t.c) + ) + + def _build_dlq_cte(outbox: "Table", dlq: "Table") -> "Insert": """Build the DELETE-then-archive CTE once; the row id, lease token and failure context are bind parameters.""" deleted = ( @@ -171,6 +215,7 @@ def __init__( self._engine = engine self._table = outbox_table self._dlq_table = dlq_table + self._fetch_stmt = _build_fetch_stmt(outbox_table) self._dlq_cte = None if dlq_table is None else _build_dlq_cte(outbox_table, dlq_table) @property @@ -212,50 +257,9 @@ async def fetch( raise TypeError(msg) if not queues: return [] - token = uuid.uuid4() - t = self._table - - # ``make_interval(secs => :lease_ttl)`` keeps the cutoff computation server-side - # so lease expiry is immune to clock skew between worker and DB hosts. - lease_cutoff = func.now() - func.make_interval(0, 0, 0, 0, 0, 0, bindparam("lease_ttl", type_=Float)) - ready = ( - select(t.c.id) - .where( - t.c.next_attempt_at <= func.now(), - # P11: ``queue = ANY(:queues)`` binds a single array param, so the SQL - # text is stable regardless of the number of queues — asyncpg can reuse - # the prepared statement across ticks instead of recompiling an - # OR-of-N-equalities whose text changes with the queue count. - t.c.queue == any_(bindparam("queues", list(queues), type_=ARRAY(String))), - # The OR is split into two index-implying disjuncts so Postgres' - # partial-index inference picks up both `_pending_idx` (Branch A, - # `acquired_token IS NULL`) and `_lease_idx` (Branch B, - # `acquired_token IS NOT NULL AND acquired_at < lease_cutoff`) and - # combines them via BitmapOr. The naive form `acquired_at < lease_cutoff` - # alone does not imply `acquired_token IS NOT NULL`, so the planner - # cannot prove `_lease_idx` applies and falls back to a seq-scan. - or_( - t.c.acquired_token.is_(None), - and_(t.c.acquired_token.is_not(None), t.c.acquired_at < lease_cutoff), - ), - ) - .order_by(t.c.next_attempt_at, t.c.id) - .limit(limit) - .with_for_update(skip_locked=True) - .cte("ready") - ) - stmt = ( - update(t) - .where(t.c.id.in_(select(ready.c.id))) - .values( - acquired_at=func.now(), - acquired_token=token, - deliveries_count=t.c.deliveries_count + 1, - ) - .returning(*t.c) - ) + params = {"queues": list(queues), "limit": limit, "lease_ttl": lease_ttl_seconds, "token": uuid.uuid4()} async with conn.begin(): - result = await conn.execute(stmt, {"lease_ttl": lease_ttl_seconds}) + result = await conn.execute(self._fetch_stmt, params) rows = result.mappings().all() return [_row_to_message(dict(row)) for row in rows] diff --git a/faststream_outbox/schema.py b/faststream_outbox/schema.py index 5796852..3b2a1d7 100644 --- a/faststream_outbox/schema.py +++ b/faststream_outbox/schema.py @@ -121,13 +121,15 @@ def make_outbox_table(metadata: "MetaData", table_name: str = "outbox") -> Table name=f"{table_name}{_LEASE_CK_SUFFIX}", ), ) - # Partial index that backs the fetch query's hot branch - # (`WHERE acquired_token IS NULL AND queue = ? AND next_attempt_at <= now()`). + # Partial index that backs the fetch query's unleased branch: an ordered scan per queue on + # (next_attempt_at, id) that stops after `limit` unlocked rows. `id` is part of the key because + # rows published in one transaction share `next_attempt_at`. See ADR 0003. # The expired-lease branch is covered by `_lease_idx` below. Index( f"{table_name}{_PENDING_IDX_SUFFIX}", table.c.queue, table.c.next_attempt_at, + table.c.id, postgresql_where=table.c.acquired_token.is_(None), ) # Partial unique index that backs `publish(..., timer_id=...)`'s ON CONFLICT DO NOTHING diff --git a/tests/test_client_contract.py b/tests/test_client_contract.py index 7bc1388..1c57060 100644 --- a/tests/test_client_contract.py +++ b/tests/test_client_contract.py @@ -316,6 +316,25 @@ async def test_fetch_selects_oldest_under_limit(contract: _Harness) -> None: assert rid_a not in {m.id for m in msgs} +async def test_fetch_selects_oldest_across_queues_under_limit(contract: _Harness) -> None: + rid_a_old = await contract.seed(queue="a", next_attempt_offset=-30.0) + rid_b_mid = await contract.seed(queue="b", next_attempt_offset=-20.0) + await contract.seed(queue="a", next_attempt_offset=-10.0) + await contract.seed(queue="b", next_attempt_offset=-5.0) + msgs = await contract.fetch(["a", "b"], limit=2, lease_ttl_seconds=60.0) + assert {m.id for m in msgs} == {rid_a_old, rid_b_mid} + + +async def test_fetch_selects_oldest_across_pending_and_expired_leases(contract: _Harness) -> None: + rid_expired = await contract.seed( + queue="orders", next_attempt_offset=-30.0, leased_token=uuid.uuid4(), acquired_age=120.0, deliveries_count=1 + ) + rid_pending = await contract.seed(queue="orders", next_attempt_offset=-20.0) + await contract.seed(queue="orders", next_attempt_offset=-10.0) + msgs = await contract.fetch(["orders"], limit=2, lease_ttl_seconds=60.0) + assert {m.id for m in msgs} == {rid_expired, rid_pending} + + async def test_fetch_respects_limit(contract: _Harness) -> None: for _ in range(3): await contract.seed(queue="orders") diff --git a/tests/test_integration.py b/tests/test_integration.py index 3b5f981..35a8ed8 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -12,7 +12,7 @@ import pytest from faststream import BaseMiddleware, Context, Depends from faststream.kafka import KafkaBroker, TestKafkaBroker -from sqlalchemy import MetaData, Table, event, insert, select, text +from sqlalchemy import MetaData, Table, event, func, insert, literal, select, text from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine @@ -103,6 +103,22 @@ async def test_validate_schema_detects_non_unique_timer_id_index( await client.validate_schema() +async def test_validate_schema_reports_the_pre_0_15_pending_index_shape( + pg_engine: AsyncEngine, + outbox_table: Table, +) -> None: + """An outbox migrated before 0.15 has ``_pending_idx`` on ``(queue, next_attempt_at)`` and must fail validation.""" + idx = f"{outbox_table.name}_pending_idx" + async with pg_engine.begin() as conn: + await conn.execute(text(f'DROP INDEX "{idx}"')) + await conn.execute( + text(f'CREATE INDEX "{idx}" ON {outbox_table.name} (queue, next_attempt_at) WHERE acquired_token IS NULL'), + ) + client = OutboxClient(pg_engine, outbox_table) + with pytest.raises(RuntimeError, match="pending_idx"): + await client.validate_schema() + + async def test_validate_schema_passes_for_correct_table(pg_engine, outbox_table) -> None: client = OutboxClient(pg_engine, outbox_table) await client.validate_schema() # should not raise @@ -559,6 +575,62 @@ async def handle(body: dict[str, Any], session: AsyncSession = session_ctx) -> O assert queues == ["downstream"] +def _scanned_rows(plan: dict[str, Any], table: str) -> list[float]: + """Rows each plan node read from *table* (actual rows times loops), skipping the UPDATE target itself.""" + found = [] + if plan.get("Relation Name") == table and plan.get("Node Type") != "ModifyTable": + found.append(plan["Actual Rows"] * plan["Actual Loops"]) + for child in plan.get("Plans", []): + found.extend(_scanned_rows(child, table)) + return found + + +async def test_fetch_reads_only_the_rows_it_claims_from_a_backlog( + pg_engine: AsyncEngine, + outbox_table: Table, +) -> None: + """INVARIANT: a fetch reads about ``limit`` rows from the outbox, however large the ready backlog is. + + The claim must come off an ordered scan of ``_pending_idx`` that stops after ``limit`` unlocked rows. + An OR across the pending and expired-lease branches, ``queue = ANY(...)``, or an index without + ``id`` (rows published in one transaction share ``next_attempt_at``) each force Postgres to read + and sort every ready row, so draining a backlog slows down with the square of its size. + """ + limit = 10 + async with pg_engine.begin() as conn: + backlog = select(literal("orders"), literal(b"x")).select_from(func.generate_series(1, 20000)) + await conn.execute(insert(outbox_table).from_select(["queue", "payload"], backlog)) + await conn.execute(text(f"ANALYZE {outbox_table.name}")) + + captured: list[tuple[str, Any]] = [] + + def capture(_conn: Any, _cursor: Any, statement: str, parameters: Any, _context: Any, _many: bool) -> None: + if "FOR UPDATE SKIP LOCKED" in statement: + captured.append((statement, parameters)) + + client = OutboxClient(pg_engine, outbox_table) + event.listen(pg_engine.sync_engine, "before_cursor_execute", capture) + try: + async with pg_engine.connect() as conn: + await client.fetch(conn, ["orders"], limit=limit, lease_ttl_seconds=60.0) + finally: + event.remove(pg_engine.sync_engine, "before_cursor_execute", capture) + statement, parameters = captured[0] + + async with pg_engine.connect() as conn: + raw = (await conn.get_raw_connection()).driver_connection + assert raw is not None + tx = raw.transaction() + await tx.start() + try: + explained = await raw.fetchval(f"EXPLAIN (ANALYZE, FORMAT JSON) {statement}", *parameters) + finally: + await tx.rollback() + plan = explained[0]["Plan"] + + assert max(_scanned_rows(plan, outbox_table.name)) <= 2 * limit + + async def test_publish_payload_is_decodable_by_subscriber(pg_engine, outbox_table) -> None: received: list[dict] = [] broker = OutboxBroker(pg_engine, outbox_table=outbox_table) diff --git a/tests/test_unit.py b/tests/test_unit.py index c4fb41d..aa9d423 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -3789,14 +3789,14 @@ async def test_delete_with_lease_raises_when_dlq_payload_but_no_dlq_table() -> N await engine.dispose() -async def test_fetch_cte_carries_partial_index_predicates_as_conjuncts() -> None: - """INVARIANT: each OR arm of the fetch WHERE carries its partial-index predicate as a conjunct. - - Branch A is ``acquired_token IS NULL``; Branch B is ``acquired_token IS NOT NULL AND acquired_at - < cutoff``. The naive single-OR form (``acquired_at < cutoff`` without the ``IS NOT NULL`` - conjunct) returns the same rows -- the fake/real parity suite cannot tell them apart -- but - Postgres only uses a partial index when the query implies the index's own WHERE clause, so the - naive form silently seq-scans. This compiles the real statement and pins the shape. +async def test_fetch_branches_carry_partial_index_predicates_and_one_queue_array() -> None: + """INVARIANT: each fetch branch carries its partial-index predicate, and the queues bind as one array. + + The unleased branch filters ``acquired_token IS NULL`` and the expired-lease branch ``acquired_token IS + NOT NULL AND acquired_at < cutoff``. Dropping the ``IS NOT NULL`` conjunct returns the same rows, so the + fake/real parity suite can't tell, but Postgres only uses a partial index when the query implies its + WHERE clause. Binding the queues as one ``unnest``-ed array keeps the SQL text the same for any number + of queues, so asyncpg reuses one prepared statement instead of recompiling per queue count. """ engine = create_async_engine("postgresql+asyncpg://u:p@localhost/db") try: @@ -3821,11 +3821,10 @@ async def execute(self, stmt: object, _params: object) -> object: await client.fetch(_CapturingConn(), ["orders"], limit=10, lease_ttl_seconds=60.0) # ty: ignore[invalid-argument-type] sql = str(captured["stmt"].compile(dialect=postgresql.dialect())).lower() - assert "acquired_token is null" in sql # Branch A predicate - assert "acquired_token is not null" in sql # Branch B conjunct (the naive form drops this) - # P11: queues bound as a single ANY(:queues) array (stable SQL across queue counts), - # not an OR-of-N-equalities. - assert "= any (" in sql # ANY(:queues) array, not OR-of-equalities + assert "acquired_token is null" in sql + assert "acquired_token is not null" in sql + assert sql.count("unnest(") == 1 + assert sql.count("join lateral") == 2 finally: await engine.dispose()