Hand the rows over as the engine makes them

[tamnd]

execute runs a statement to the end and gives back every row. That is the wrong shape for a result bigger than the memory meant for it, and for a reader that wants the first rows before the last are made. conn.stream is the other shape.

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)

How it is put together

The statement runs on a thread of its own rather than on the connection's runner, because a stream lasts as long as its reader and the runner is for statements that end. Rows are copied out of the borrowed batch on that thread with the GIL down, and turned into Python objects on the reader's thread. The two are joined by a queue two batches deep, so the engine is filling the next batch while the loop reads this one and neither waits for the other for long. It is the design zu-node already uses, on purpose, so that the two clients behave the same way and the reasoning is written down once in each module.

A stream holds the connection until it ends, since a connection runs one statement at a time. A statement run on that 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. Reading a stream to the end frees the connection, so does close, and so does the end of a with block however it was left. Closing the connection hangs the stream up first, which is what lets the close return rather than wait for a scan nobody is reading. A reader that stopped a stream early can still ask what it stopped, because the summary is kept beside the queue rather than in it and hanging up throws away the queue.

summary is None while the statement runs and afterwards says what it did, including whether the rows arrived as they were made. 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 says streamed is False. The loop over it is the same loop and what differs is what it cost.

zudb.aio gets the same surface as AsyncStream and AsyncStreamBatches, with async for and an awaited close, where waiting for a batch is handed off the loop like every other wait there.

Measured

One million people in two columns, on this machine.

	execute	stream	stream(...).batches()
first row	29.9 ms	0.70 ms
whole read	256 ms	214 ms	176 ms
rows a second	3.9 M	4.7 M	5.7 M
Python peak	153 MB	nothing worth measuring

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. Reading a batch at a time is faster again because a batch is one call where a row is one call.

Tests

Twenty seven new ones, 671 passed and 5 skipped in 201 s. They cover the rows and their order, the columns answered before a row is read, a first row in hand while the summary is still None, the batch sizes asked for and the ones the engine chose, the refusal and the connection coming back after it, the summary of a run that ended and of one that was stopped, the buffered case, a statement that does not compile failing at the first row asked for and failing again at the second, a stream nobody read, a connection closed underneath one, and the memory a full read holds.

The tests about a stream that is still running are written on a cross join rather than on a scan, because the engine hands over 2,048 rows at a time and a queue two deep means a scan of five thousand people is finished before a test can look at it. That one bit me while writing them and is worth knowing when adding more.

A note on batch_rows

It is a ceiling rather than a size. The engine fills whole vectors and hands over as many of them as fit under the number, so batch_rows=10_000 gives batches of 9,216 and batch_rows=500 gives batches of exactly 500, since a vector splits evenly at that size. Asking for zero is a ValueError rather than a loop that never moves.

Closes part of tamnd/zu#169.