Hand the rows over as the engine makes them

README.md

@@ -126,6 +126,39 @@ result.record_batches()  # a reader, for a result larger than memory
 
 `Result` implements `__arrow_c_stream__`, so anything that reads the protocol reads a result directly and none of the four methods above is needed: `pyarrow.table(result)` and `polars.DataFrame(result)` both work. Batches are 65,536 rows. A column holds one type, which the values decide, and integers beside floats are the one mixture that widens rather than being refused. Nodes, rels and paths go across as structs. The copy runs with the GIL released, and on this machine 300,000 rows across three columns take 44 ms as Arrow against 67 ms as Python objects, and a single integer column takes 13.8 ms against 44.5 ms.
 
+## Reading a result as it arrives
+
+`execute` runs the statement to the end and hands back every row. `stream` hands back the rows as the engine makes them, which is what you want when the result is bigger than the memory you meant to spend on it, or when the first rows are worth having before the last are made.
+
+```python
+with conn.stream("MATCH (p:person) RETURN p.uid AS uid, p.name AS name") as rows:
+    for uid, name in rows:
+        write(uid, name)
+```
+
+```python
+with conn.stream("MATCH (p:person) RETURN p.uid AS uid", batch_rows=10_000) as rows:
+    for batch in rows.batches():
+        write_many(batch)
+    rows.summary.rows  # how many were read
+```
+
+`batch_rows` is a ceiling rather than a size. The engine fills whole vectors and a batch holds as many of them as fit under the number, so asking for 10,000 gives batches of 9,216 and asking for 500 gives batches of exactly 500, since a vector divides evenly by that.
+
+The statement runs on a thread of its own with the GIL down and hands each batch over a queue two batches deep, so the engine is filling the next batch while your loop is reading this one and neither waits for the other for long. On this machine, over a million people in two columns, the first row arrives 0.7 ms after the call against 30 ms for `execute`, and reading all of them takes 214 ms streamed against 256 ms in one piece, which is 4.7 million rows a second against 3.9 million. Reading them a batch at a time takes 176 ms, or 5.7 million a second, because a batch is one call where a row is one call. Streaming being the faster of the two is not a trick: the rows are made and consumed while the cache still has them, and nothing has to hold a million tuples at once. Held is the whole difference: the Python side of `fetchall` peaks at 153 MB for that result and the same read streamed peaks at nothing worth measuring, since every tuple is freed as the loop moves past it.
+
+A stream holds the connection until it ends, because a connection runs one statement at a time. Reading it to the end frees the connection, so does `close`, and so does the end of a `with` block however it was left. A statement run on the connection while a stream is open is refused with a message that says so rather than queued behind a loop that may never finish, which is the deadlock the refusal exists to prevent. If a program needs to read a stream and run statements at the same time, that is two connections.
+
+`summary` is `None` while the statement runs and afterwards says what it did: the columns, how many rows were handed over, whether the reader stopped it early, and whether the rows arrived as they were made. That last one is worth reading. A statement that has to see every row before it can give one, which is `ORDER BY`, `DISTINCT` and the aggregates, is run whole by the engine and handed over in batches afterwards, and `streamed` is `False` for it. The loop over it is the same loop and what differs is what it cost.
+
+`zudb.aio` has the same thing, where the waits go off the loop like every other wait there:
+
+```python
+async with conn.stream("MATCH (p:person) RETURN p.name AS name") as rows:
+    async for (name,) in rows:
+        await write(name)
+```
+
 ## Preparing a statement
 
 A statement a program runs many times with different values can be compiled once and kept.
@@ -286,7 +319,7 @@ Half of this package is compiled, which is the one thing an inspection cannot se
 
 ## What works today
 
-The list above is what this client is for. What it does so far is the core of it: `connect`, `execute` and `sql` with named parameters, results that iterate and fetch, values as Python objects both ways including dates, times, datetimes and durations, `Node`, `Rel` and `Path` as classes, `load` for building a graph with edges in it, an appender for growing one, transactions as a context manager that commits at the end of a block and rolls back when it raises, every condition as an exception class carrying its code, its position and its documentation link, results as Arrow columns and as pandas and polars frames, `register` for putting a frame under a name a statement can match on and reading it where it lies, stubs inside the wheel with a gate that keeps them true, the GIL released around every statement, every load and every copy out, `Ctrl-C` and `interrupt()` stopping a statement without touching the connection under it, `zudb.aio` for the same calls awaited on an event loop, results, nodes, rels and paths that draw themselves in a notebook with `%gql` and `%%gql` to run statements in one, `zudb.dbapi` for code written against PEP 249, and `prepare`, `explain` and `profile` for a statement compiled once, the plan it would run and the plan it did. Each one landed with the tests that say it works.
+The list above is what this client is for. What it does so far is the core of it: `connect`, `execute` and `sql` with named parameters, results that iterate and fetch, values as Python objects both ways including dates, times, datetimes and durations, `Node`, `Rel` and `Path` as classes, `load` for building a graph with edges in it, an appender for growing one, transactions as a context manager that commits at the end of a block and rolls back when it raises, every condition as an exception class carrying its code, its position and its documentation link, results as Arrow columns and as pandas and polars frames, `register` for putting a frame under a name a statement can match on and reading it where it lies, stubs inside the wheel with a gate that keeps them true, the GIL released around every statement, every load and every copy out, `Ctrl-C` and `interrupt()` stopping a statement without touching the connection under it, `zudb.aio` for the same calls awaited on an event loop, results, nodes, rels and paths that draw themselves in a notebook with `%gql` and `%%gql` to run statements in one, `zudb.dbapi` for code written against PEP 249, `prepare`, `explain` and `profile` for a statement compiled once, the plan it would run and the plan it did, and `stream` for a result read as the engine makes it rather than after it has made all of it. Each one landed with the tests that say it works.
 
 ## Wheels
 

python/zudb/__init__.py

@@ -34,6 +34,9 @@
     Rel,
     Result,
     ScalarPlan,
+    Stream,
+    StreamBatches,
+    StreamSummary,
     Transaction,
     __abi_version__,
     connect,
@@ -61,6 +64,9 @@
     "Appender",
     "Prepared",
     "Result",
+    "Stream",
+    "StreamBatches",
+    "StreamSummary",
     "Plan",
     "PlanNode",
     "ScalarPlan",

python/zudb/_zudb.pyi

@@ -85,6 +85,15 @@ class Connection:
     def profile(self, statement: str, params: Mapping[str, Value] | None = None) -> Profile:
         """Runs the statement with the counters on and answers what its operators really did."""
 
+    def stream(
+        self,
+        statement: str,
+        params: Mapping[str, Value] | None = None,
+        *,
+        batch_rows: int | None = None,
+    ) -> Stream:
+        """Runs one statement and hands back its rows as the engine makes them."""
+
     def transaction(self, *, read_only: bool = False) -> Transaction:
         """Starts a transaction and hands it back for a `with` block."""
 
@@ -199,6 +208,65 @@ class Prepared:
     def __exit__(self, *_exception: object) -> bool: ...
     def __repr__(self) -> str: ...
 
+class Stream:
+    """Rows arriving one batch at a time, while the statement is still running."""
+
+    @property
+    def columns(self) -> list[str]:
+        """The column names, in the order the statement projects them."""
+
+    @property
+    def summary(self) -> StreamSummary | None:
+        """What the statement did, once it has done it, and `None` while it is still running."""
+
+    def batches(self) -> StreamBatches:
+        """The rows in the batches they arrived in, as lists of tuples."""
+
+    def close(self) -> None:
+        """Stops the statement and gives the connection back."""
+
+    @property
+    def closed(self) -> bool:
+        """Whether the statement is over, by running out of rows or by being closed."""
+
+    def __iter__(self) -> Iterator[tuple[Value, ...]]: ...
+    def __next__(self) -> tuple[Value, ...]: ...
+    def __enter__(self) -> Stream: ...
+    def __exit__(self, *_exception: object) -> bool: ...
+    def __repr__(self) -> str: ...
+
+class StreamBatches:
+    """The same rows, in the batches they arrived in."""
+
+    def __iter__(self) -> Iterator[list[tuple[Value, ...]]]: ...
+    def __next__(self) -> list[tuple[Value, ...]]: ...
+    def __repr__(self) -> str: ...
+
+class StreamSummary:
+    """What a streamed statement did, known once it has ended."""
+
+    @property
+    def columns(self) -> list[str]:
+        """The column names, in the order the statement projected them."""
+
+    @property
+    def rows(self) -> int:
+        """How many rows were handed over."""
+
+    @property
+    def stopped(self) -> bool:
+        """Whether the reader stopped it before it ran out of rows."""
+
+    @property
+    def streamed(self) -> bool:
+        """Whether the rows arrived as they were made."""
+
+    @property
+    def notices(self) -> list[dict[str, str]]:
+        """The warnings the statement raised, in the shape a result reports them."""
+
+    def __repr__(self) -> str: ...
+
 class PlanNode:
     """One operator of a plan."""
 

python/zudb/aio.py

@@ -49,7 +49,18 @@
 from typing import Any, Generic, TypeVar
 
 from . import _zudb
-from ._zudb import Appender, Connection, Plan, Prepared, Profile, Result, Transaction
+from ._zudb import (
+    Appender,
+    Connection,
+    Plan,
+    Prepared,
+    Profile,
+    Result,
+    Stream,
+    StreamBatches,
+    StreamSummary,
+    Transaction,
+)
 from .types import Value
 
 __all__ = [
@@ -58,10 +69,18 @@
     "AsyncTransaction",
     "AsyncAppender",
     "AsyncPrepared",
+    "AsyncStream",
+    "AsyncStreamBatches",
 ]
 
 T = TypeVar("T")
 
+#: Handed to `next` so that the end of a stream arrives as a value on the
+#: connection's thread. A `StopIteration` raised there would cross a
+#: future on its way back, and a `StopIteration` crossing a future is the
+#: one exception asyncio cannot let through.
+_NOTHING = object()
+
 
 def connect(
     path: str | os.PathLike[str],
@@ -252,6 +271,45 @@ async def profile(self, statement: str, params: Mapping[str, Value] | None = Non
         """
         return await self._call(functools.partial(self._conn.profile, statement, params))
 
+    def stream(
+        self,
+        statement: str,
+        params: Mapping[str, Value] | None = None,
+        *,
+        batch_rows: int | None = None,
+    ) -> _Opening[AsyncStream]:
+        """Runs one statement and hands back its rows as the engine
+        makes them.
+
+            async with conn.stream("MATCH (p:person) RETURN p.name AS name") as rows:
+                async for (name,) in rows:
+                    await write(name)
+
+        This is the call for a result too big to want in memory and for
+        one whose first rows are worth having before the last are made.
+        The statement runs on a thread of its own, not this connection's,
+        and every wait for a batch is handed off the loop like every
+        other wait here, so a task reading a stream leaves the loop free
+        between batches.
+
+        A stream holds the connection until it ends, so a statement run
+        on the same connection while one is open is refused rather than
+        queued behind a loop that may never finish. Read it to the end,
+        close it, or open it with `async with`.
+        """
+        return _Opening(functools.partial(self._stream, statement, params, batch_rows))
+
+    async def _stream(
+        self,
+        statement: str,
+        params: Mapping[str, Value] | None,
+        batch_rows: int | None,
+    ) -> AsyncStream:
+        opened = await self._call(
+            functools.partial(self._conn.stream, statement, params, batch_rows=batch_rows)
+        )
+        return AsyncStream(self, opened)
+
     def transaction(self, *, read_only: bool = False) -> _Opening[AsyncTransaction]:
         """Starts a transaction and hands it back for an `async with`
         block.
@@ -597,3 +655,113 @@ async def __aexit__(self, *_exception: Any) -> bool:
     def __repr__(self) -> str:
         closed = ", closed" if self.closed else ""
         return f"<zudb.aio.AsyncPrepared {self.statement!r}{closed}>"
+
+
+class AsyncStream:
+    """Rows arriving one batch at a time, awaited a row at a time.
+
+    Take one with `AsyncConnection.stream`. The stream underneath is
+    `zudb.Stream` and the rules are its rules: it holds the connection
+    until it ends, closing it twice does nothing, and the summary is
+    there once the statement is over whether it ran out of rows or was
+    stopped.
+
+    Waiting for the next batch is handed to the connection's thread, so
+    the loop is free while the engine is working. The rows of a batch
+    already in hand are turned into tuples on the loop's thread, which is
+    the same work `Result` does when it is read, and there is nothing to
+    wait for in it.
+    """
+
+    __slots__ = ("_conn", "_stream")
+
+    def __init__(self, conn: AsyncConnection, stream: Stream) -> None:
+        self._conn = conn
+        self._stream = stream
+
+    async def columns(self) -> list[str]:
+        """The column names, in the order the statement projects them.
+
+        A method where the sync client has a property, because answering
+        reads the first batch and reading a batch can wait.
+        """
+        return await self._conn._call(lambda: self._stream.columns)
+
+    @property
+    def summary(self) -> StreamSummary | None:
+        """What the statement did, once it has done it, and `None` while
+        it is still running.
+
+        A property and not a coroutine: the answer is beside the queue
+        rather than behind the engine, so asking reaches nothing that
+        could wait.
+        """
+        return self._stream.summary
+
+    @property
+    def closed(self) -> bool:
+        """Whether the statement is over, by running out of rows or by
+        being closed.
+        """
+        return self._stream.closed
+
+    def batches(self) -> AsyncStreamBatches:
+        """The rows in the batches they arrived in, as lists of tuples.
+
+        For a writer with a size of its own: one await per batch rather
+        than one per row, and one call into whatever is being written to.
+        """
+        return AsyncStreamBatches(self._conn, self._stream.batches())
+
+    async def close(self) -> None:
+        """Stops the statement and gives the connection back.
+
+        Awaited because it waits for the statement to stop, so the
+        connection is free by the time the call returns. Doing it twice
+        is not an error.
+        """
+        await self._conn._call(self._stream.close)
+
+    def __aiter__(self) -> AsyncStream:
+        return self
+
+    async def __anext__(self) -> tuple[Value, ...]:
+        row = await self._conn._call(functools.partial(next, self._stream, _NOTHING))
+        if row is _NOTHING:
+            raise StopAsyncIteration
+        return row  # type: ignore[return-value]
+
+    async def __aenter__(self) -> AsyncStream:
+        return self
+
+    async def __aexit__(self, *_exception: Any) -> bool:
+        """Closes on the way out, whether the block ended well or badly,
+        so the connection comes back either way.
+        """
+        await self.close()
+        return False
+
+    def __repr__(self) -> str:
+        return f"<zudb.aio.AsyncStream of {self._stream!r}>"
+
+
+class AsyncStreamBatches:
+    """The same rows, in the batches they arrived in."""
+
+    __slots__ = ("_conn", "_batches")
+
+    def __init__(self, conn: AsyncConnection, batches: StreamBatches) -> None:
+        self._conn = conn
+        self._batches = batches
+
+    def __aiter__(self) -> AsyncStreamBatches:
+        return self
+
+    async def __anext__(self) -> list[tuple[Value, ...]]:
+        batch = await self._conn._call(functools.partial(next, self._batches, _NOTHING))
+        if batch is _NOTHING:
+            raise StopAsyncIteration
+        return batch  # type: ignore[return-value]
+
+    def __repr__(self) -> str:
+        return f"<zudb.aio.AsyncStreamBatches of {self._batches!r}>"