Snapshot: sql and scan return a streaming reader; fix rows() stopping early

[nhobin219]

Closes #108. There are three commits, so the release notes list the bug fix separately from the two breaking changes.

fix(log): rows() no longer stops early

_published.Table.scan now streams on a cursor of its own (Owning closes the cursor together with its reader). Before, it streamed on the snapshot's shared connection, so any other query on that snapshot ended the replay at the next batch boundary, and rows() finished as if it had read everything. test_a_query_while_rows_are_read_does_not_end_them fails against the old code, cut off at 50 of 600 rows because the test uses small batches, and passes with the fix.

feat(stream)!: sql and scan return a Reader

async with snapshot.sql("SELECT id, max(litelink_offset) AS last FROM log GROUP BY id") as result:
    async for batch in result:     # pa.RecordBatch, batch_size rows
        ...
table = await snapshot.sql("SELECT …").read_all()

• Where: Snapshot.sql/scan and Live.sql/scan. They are now plain methods (not async), and batch_size= defaults to DuckDB's 1,000,000.
• How a reader runs: its query runs once, when the reader is first read. It uses a DuckDB cursor of its own, so it is unaffected by other reads, and works in a worker thread, so the event loop never waits on it. The tail and end it reads are recorded when the reader is created, because Live._view changes them on the snapshot for each later query. streamcast_tail, streamcast_empty and the view log are registered on the reader's own cursor, so two sql calls no longer overwrite each other's log.
• Lifetime: a reader is valid while its snapshot is open. A batch asked for after the snapshot closed raises; it never ends the result early. rows() now checks this on every row. The snapshot counts its open readers and closes its connection when the last one finishes.
• Live: rebase retires the old base instead of closing it, so a reader keeps reading it and the base closes when its last reader finishes. Since _view and the swap in rebase contain no await, the asyncio lock had nothing left to protect, so I removed it.
• Exports: streamcast.Reader is now exported.
• Call sites: I added .read_all() everywhere a full table is wanted: _live.wait_for, examples/otel/analytics.py, benchmarks/snapshot.py, the tests, README.md, docs/API.md and the examples/README.md snippet from docs(examples): query the demo's log with a snapshot, not litelink #107.

Tests. Every test below fails when the behaviour it covers is broken, and passes with this branch:

• Shared connection used for readers: test_two_readers_on_one_snapshot_both_finish_whole and test_read_all_is_the_batches_put_together.
• No release on close: test_leaving_it_early_releases_its_cursor and test_a_batch_after_the_snapshot_closed_raises.
• No closed check: test_a_batch_after_the_snapshot_closed_raises.
• Reads on the event loop: test_the_event_loop_runs_while_it_reads.
• rows() checking per log instead of per row: test_rows_after_the_snapshot_closed_raises.
• Rebase closing the old base: test_a_rebase_does_not_end_it.
• Reader not recording its view: test_rows_received_while_it_is_read_are_not_in_it.

test_catchup.py::test_the_reads_cross_into_a_thread now inspects Snapshot._shut_if_idle, which is where the connection is closed now.

feat(stream)!: Stream.sql/Stream.scan return a reader too

These now return a reader, as litelink's scan and sql do. It is a streamcast.Reader, not a pyarrow.RecordBatchReader: a pyarrow reader is synchronous, so each read_next_batch() would run DuckDB's work on the event loop.

• No longer async: use await Stream.sql(uri, q).read_all(), or async for batch in Stream.scan(uri).
• Its own snapshot: the reader opens a snapshot on its first read, so a refusal such as SnapshotUnavailable is raised there rather than at the call. The snapshot closes once the reader is done.
• Closed unread: a reader closed before it was ever read never opens a snapshot.
• Tests: four new ones in TestTheOneShotForms. The close test fails when the reader doesn't close its own snapshot.

Benchmark

just bench-snapshot --rows 100000 1000000 --logs 1 4 16 on main and on this branch, both on this VM.

• Single run on each side: most rows were within noise, but the 16-log full scans and aggregates looked slower on the branch. For example, the full scan at 1M rows went from 402 ms to 477 ms.
• Repeated, alternating runs: main itself moved as much as that between runs. It took 426 and 474 ms for the full 1M-row scan, and 64 and 90 ms for the group-by. The branch took 421 and 370 ms, and 87 and 75 ms, which is inside the same ranges.
• The suspected cause: to_arrow_reader().read_all() and to_arrow_table() cost the same when measured on their own: 103 ms against 100 ms for a sort of 3M rows.

So I see no regression.

Breaking: await snapshot.sql(...), and the same on Live and Stream, must become await ….read_all(). This is a minor version bump under 0.x. The CHANGELOG has Breaking and Fixed entries.

🤖 Generated with Claude Code

https://claude.ai/code/session_01FsSDkeb5rVAxA1FSmKKfQi