Stream the binary read frame by frame

AGENTS.md

@@ -203,7 +203,7 @@ the generic default with `recognized: false`, not an error.
 
 `TimeSeriesService::insert_datapoints_binary` is the second ingest path, to `POST /timeseries/data/binary`: the same `DatapointsCollection<DatapointString>` input, resolved through `/timeseries/byids` (cached per service instance; needs read access on the dataset), checked against each series' value type locally, cut into Arrow IPC frames at the contract's caps, zstd-compressed per frame (mandatory: level 1, 3 or 9, default 9) and posted through `execute_post_bytes_request`. Every refusal on this path is a problem document typed for its status — `invalid-frame` (400), `unknown-timeseries` (404), `request-too-large` (413), `unsupported-media-type` (415), `value-type-mismatch` and `external-id-mismatch` (422), `too-many-in-flight` (429) — with the kebab-case sub-case in a `reason` extension beside it. It was one type, `datapoint-block-rejected`, answering with all six statuses until the api split it (platform #120); the SDK matches the new slugs only, and nothing here recognises the old one — the binary path has never been in a release, so no caller can be on the other side of that change. `is_stale_series_rejection` reads the slug to decide the one re-resolve-and-retry, and reads it from the problem rather than the body text because the SDK's own pre-flight 404 names the series it could not resolve — an external id spelling `unknown-timeseries` used to trigger the retry. `FrameWriter` builds one frame and is public; `cut_into_writers` and `pack_requests` hold the caps. The byte layout is the platform's `binary_datapoints_format.md`. The arrow-rs crates (`arrow-array`, `arrow-schema`, `arrow-ipc`) serve both directions. In Python it is `insert_datapoints_binary` and `insert_from_lists_binary` on the **sync** client only, each taking an optional `zstd_level`; the async client has no binary methods. The ignored `timeseries::tests::test_datapoints_binary` is the live twin of `test_datapoints` and needs a backend that serves the endpoint; the writer's own tests in `binary.rs` run offline. `python_tests/test_timeseries_binary.py` round-trips the Python binding through the JSON read path.
 
-The read is `TimeSeriesService::retrieve_datapoints_binary`, to `POST /timeseries/data/list/binary`: the JSON retrieve body in, the same frames out, one series per frame, then the 4-byte marker `DHDE`. A body without the marker was cut short (the server streams, so a failure after the first bytes can only end the body early) and is a **502** saying so, never fewer points; bytes after it are refused too. Codec 2 frames carry bucketed aggregates in their own schema (`avg`/`min`/`max`/`sum`, Float64, 50 000 buckets per frame) while the envelope's type byte stays the series' own. `parse_read_response` returns one columnar `BinaryDatapoints` per series, merging consecutive frames of one series; decimals stay exact as unscaled `i128` plus scale. It walks the envelopes first (cheap, and where a cut-short body is caught), then decompresses and decodes the frames on scoped threads — zstd was 90% of a single-threaded parse, 1.3 s per 100M float rows — and `retrieve_datapoints_binary` runs it on `spawn_blocking`. What remains (about 0.6 s per 100M rows) is allocation and copying, not zstd; streaming the read would hide it behind the download. Differences from the JSON read worth knowing: no `cursor` (a 400; advance `start` past the last timestamp), a series with no points is absent rather than empty, and an unknown or unreadable series is a 400 for the whole request — `DataWrapper<RetrieveFilter>` serialises only `items`, so the request-level `ignoreUnknownIds` and `limit` are not reachable. `src/timeseries/fixtures/read_*.bin` are response bodies written by the platform's own `DatapointFrameWriter.wrap`, so the offline tests in `binary_read.rs` prove arrow-rs reads what the Java writer emits; regenerate them when the platform's writer changes. The ignored `timeseries::tests::test_datapoints_binary_read` is the live check against the JSON read. In Python it is `retrieve_datapoints_binary` on the sync client, returning `BinaryDatapoints` objects that implement the Arrow PyCapsule stream protocol (`__arrow_c_stream__`, via `arrow-array`'s `ffi` feature enabled in the bindings crate only), so polars, pyarrow and pandas 3 read them zero-copy; the package still declares no runtime dependency, and `to_polars`/`to_pandas` import theirs lazily. `python_tests/test_timeseries_binary_read.py` covers it offline against a stand-in server and live when the backend serves the endpoint.
+The read is `TimeSeriesService::retrieve_datapoints_binary`, to `POST /timeseries/data/list/binary`: the JSON retrieve body in, the same frames out, one series per frame, then the 4-byte marker `DHDE`. A body without the marker was cut short (the server streams, so a failure after the first bytes can only end the body early) and is a **502** saying so, never fewer points; bytes after it are refused too. Codec 2 frames carry bucketed aggregates in their own schema (`avg`/`min`/`max`/`sum`, Float64, 50 000 buckets per frame) while the envelope's type byte stays the series' own. `parse_read_response` returns one columnar `BinaryDatapoints` per series, merging consecutive frames of one series; decimals stay exact as unscaled `i128` plus scale. `TimeSeriesService::stream_datapoints_binary` is the primitive: a `FrameAssembler` cuts the body into frames as chunks arrive (a frame may be split anywhere), each frame is an owned `Bytes` decoded on `spawn_blocking`, at most `min(cores, 8)` in flight through a `FuturesOrdered` so items stay in frame order, and a producer task feeds a bounded channel behind the `BinaryDatapointStream` (dropping it aborts the task and closes the connection). One item per frame, so a long series is several consecutive items. A refusal is the `Err` before there is a stream; a response that breaks off is an error item after the items that arrived, and that is the semantic difference stream consumers must handle. `retrieve_datapoints_binary` collects the stream and merges consecutive parts (each series allocated once at its final size); `parse_read_response` does the same for a body in memory, decoding on scoped threads. Measured on 100M float rows, 10 series: the server takes ~6.5 s; the buffered call adds ~0.24 s (the merge copy) and the stream into polars ~0.02 s. Single-threaded zstd used to add 1.3 s on its own. Differences from the JSON read worth knowing: no `cursor` (a 400; advance `start` past the last timestamp), a series with no points is absent rather than empty, and an unknown or unreadable series is a 400 for the whole request — `DataWrapper<RetrieveFilter>` serialises only `items`, so the request-level `ignoreUnknownIds` and `limit` are not reachable. `src/timeseries/fixtures/read_*.bin` are response bodies written by the platform's own `DatapointFrameWriter.wrap`, so the offline tests in `binary_read.rs` prove arrow-rs reads what the Java writer emits; regenerate them when the platform's writer changes. The ignored `timeseries::tests::test_datapoints_binary_read` is the live check against the JSON read. In Python it is `retrieve_datapoints_binary` on the sync client, plus `retrieve_datapoints_binary_stream`, whose single-use `BinaryDatapointReader` iterates per frame or is an Arrow stream of one long table (`timeseries_id`, `timestamp`, values; one schema, so a read mixing value types must be iterated). The long table carries no external id column on purpose: as a dictionary column, polars spent ~0.8 s per 100M rows making it categorical, more than everything else; `reader.external_ids` maps the ids instead. The buffered `retrieve_datapoints_binary` returns a `BinaryReadResult`, a sequence of per-series `BinaryDatapoints` with the same `external_ids`, long-table Arrow stream and `to_polars(wide=False)` as the reader: both call `python/intellistream_datahub_sdk/_tables.py`, so the two tables cannot drift. `wide=True` puts series side by side when every one has the same timestamps (free), pivots on a numeric key when they share one value column type (~1.7 s per 100M rows; joining names first doubled it, `concat(how="align")` tripled it), and aligns otherwise, which is how mixed value types sit side by side. Each per-series `BinaryDatapoints` implements the Arrow PyCapsule stream protocol (`__arrow_c_stream__`, via `arrow-array`'s `ffi` feature enabled in the bindings crate only), so polars, pyarrow and pandas 3 read them zero-copy; the package still declares no runtime dependency, and `to_polars`/`to_pandas` import theirs lazily. `python_tests/test_timeseries_binary_read.py` covers it offline against a stand-in server and live when the backend serves the endpoint.
 
 ### The `ApiServiceProvider` trait (`src/generic.rs`)
 

Cargo.toml

@@ -30,6 +30,7 @@ reqwest = { version = "0.12", default-features = false, features = ["blocking",
 chrono = { version ="0.4", features = ["serde"] }
 rand = "0.9"
 futures = "0.3"
+bytes = "1"
 dotenv = "0.15"
 uuid = { version = "1.16", features = ["v4", "v7", "serde"] }
 regex = "1.11.1"

README.md

@@ -168,11 +168,34 @@ answer is held in memory. There is no cursor: to read on in pieces, send the req
 the JSON read answers empty. A response the server could not finish is a 502 rather than fewer
 points, and is safe to send again.
 
+`time_series.stream_datapoints_binary(&filters)` is the same read handed over as it arrives: a
+`Stream` of `BinaryDatapoints`, one per frame, decoded on blocking threads while the rest of the
+response downloads, so nothing waits for the whole body and memory holds what the consumer keeps.
+A series longer than one frame comes as consecutive items. A refusal is the `Err` before there is
+a stream; a response that breaks off is an error item **after** the items that arrived, so a
+consumer acting on items as they come has to treat what it took as incomplete. The blocking
+client has it as an iterator. `retrieve_datapoints_binary` collects this stream, which is where
+all-or-nothing lives.
+
 In Python, `client.timeseries.retrieve_datapoints_binary(filter_or_list)` on the sync client
 returns objects that implement the Arrow PyCapsule stream protocol, so
 `polars.DataFrame(series)`, `pyarrow.table(series)` and `pandas.DataFrame.from_arrow(series)`
 take them without copying; `series.to_polars()` and `series.to_pandas()` do the same. The package
 adds no dependency for this — bring polars, or pyarrow for pandas.
+`retrieve_datapoints_binary_stream(filter_or_list)` returns a single-use reader of the same read
+as it arrives: iterate it for one `BinaryDatapoints` per frame. Both the buffered result and the
+reader turn into one table the same way:
+
+```python
+read = client.timeseries.retrieve_datapoints_binary(filters)        # or ..._binary_stream(filters)
+long = read.to_polars()           # timeseries_id, timestamp, value: one row per point
+wide = read.to_polars(wide=True)  # timestamp, then one column per series named by external id
+read.external_ids                 # {timeseries_id: external_id}
+```
+
+The wide table matches series on `timestamp`, leaving nulls where one has no point, and holds
+series of different value types side by side. Reading 100M float rows into either costs within
+half a second of the download itself.
 
 ## Python bindings
 

datahub_python_bindings/Cargo.toml

@@ -16,6 +16,7 @@ crate-type = ["cdylib"]
 pyo3 = { version = "0.29", features = ["extension-module", "chrono","chrono-tz","uuid","experimental-inspect", "abi3-py39","multiple-pymethods"] }
 intellistream-datahub-sdk = { path = ".." }
 tokio = { version = "1", features = ["full"] }
+futures = "0.3"
 chrono = { version = "0.4", features = ["serde"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi

@@ -590,6 +590,108 @@ class DatapointsCollectionDatapoints:
     def id(self) -> int | None: ...
 
 
+class BinaryDatapointReader:
+    """A binary read as it arrives, from ``TimeSeriesServiceSync.retrieve_datapoints_binary_stream``.
+
+    Single use, in one of two ways; a second use raises ``RuntimeError``.
+
+    **Iterating** yields one :class:`BinaryDatapoints` per frame, as the frames arrive. A series
+    longer than one frame comes as several consecutive items, oldest first, so an item is a chunk
+    of a series rather than all of it; series of different value types mix freely.
+
+    **As an Arrow stream** (``polars.DataFrame(reader)``, :meth:`to_polars`,
+    ``pyarrow.RecordBatchReader.from_stream(reader)``) it is one long table: ``timeseries_id``,
+    ``timestamp`` and the value columns, a batch per frame. Every series in the read has to share
+    those columns, so a read mixing value types (or values and aggregates) raises at the first
+    batch that differs; read such a read with ``to_polars(wide=True)`` or by iterating. A read
+    with no rows has no schema and raises
+    ``ValueError``. The network is waited on with the GIL held, as the C stream protocol gives no
+    way to release it.
+
+    A response that breaks off raises ``DataHubException`` (502, "cut short") from ``next()`` or
+    from the Arrow stream, **after** the rows that did arrive were handed out: treat what was
+    already taken as incomplete. ``retrieve_datapoints_binary`` is the all-or-nothing form.
+    """
+
+    def __iter__(self) -> Iterator[BinaryDatapoints]: ...
+    def __next__(self) -> BinaryDatapoints: ...
+    @property
+    def external_ids(self) -> dict[int, str]:
+        """``{timeseries_id: external_id}`` for the series read so far, in the order they came:
+        the key to the long table's ``timeseries_id``. Complete once the reader is exhausted."""
+    def __arrow_c_stream__(self, requested_schema: object | None = None) -> object:
+        """The Arrow PyCapsule stream protocol, as the long table. ``requested_schema`` is ignored."""
+    def to_polars(self, wide: bool = False) -> Any:
+        """A ``polars.DataFrame`` of the read; needs polars, and raises ``ImportError`` without it.
+
+        Parameters
+        ----------
+        wide
+            ``False`` (default): the long table, ``timeseries_id``, ``timestamp`` and the value
+            columns, one row per point. Every series has to share those columns.
+            ``True``: one row per timestamp and one column per series, named after its external
+            id (``{external_id}.{column}`` for a series with several value columns: ``mixed``, or
+            aggregates), matched on ``timestamp`` with nulls where a series has no point. Series
+            of different value types sit side by side, each keeping its type.
+
+        Raises
+        ------
+        ValueError
+            The read returned no rows; or, for the long table, its series do not share their
+            columns; or, for the wide one, a series has repeated timestamps.
+
+        Notes
+        -----
+        The same table whether the read was buffered or streamed. The wide table costs nothing
+        beyond the long one when every series has the same timestamps, and a pivot (about 1.7 s
+        per 100M rows) when they do not.
+        """
+
+
+class BinaryReadResult(Sequence[BinaryDatapoints]):
+    """A buffered binary read, from ``TimeSeriesServiceSync.retrieve_datapoints_binary``.
+
+    A sequence of :class:`BinaryDatapoints`, one per series that had points, in the order of the
+    request; and the same tables a :class:`BinaryDatapointReader` gives, through
+    :meth:`to_polars` and the Arrow stream of the long table.
+    """
+
+    def __len__(self) -> int: ...
+    def __getitem__(self, index: int) -> BinaryDatapoints: ...  # type: ignore[override]
+    def __iter__(self) -> Iterator[BinaryDatapoints]: ...
+    @property
+    def external_ids(self) -> dict[int, str]:
+        """``{timeseries_id: external_id}`` for the series in the read: the key to the long
+        table's ``timeseries_id``."""
+    def __arrow_c_stream__(self, requested_schema: object | None = None) -> object:
+        """The Arrow PyCapsule stream protocol, as the long table. ``requested_schema`` is ignored."""
+    def to_polars(self, wide: bool = False) -> Any:
+        """A ``polars.DataFrame`` of the read; needs polars, and raises ``ImportError`` without it.
+
+        Parameters
+        ----------
+        wide
+            ``False`` (default): the long table, ``timeseries_id``, ``timestamp`` and the value
+            columns, one row per point. Every series has to share those columns.
+            ``True``: one row per timestamp and one column per series, named after its external
+            id (``{external_id}.{column}`` for a series with several value columns: ``mixed``, or
+            aggregates), matched on ``timestamp`` with nulls where a series has no point. Series
+            of different value types sit side by side, each keeping its type.
+
+        Raises
+        ------
+        ValueError
+            The read returned no rows; or, for the long table, its series do not share their
+            columns; or, for the wide one, a series has repeated timestamps.
+
+        Notes
+        -----
+        The same table whether the read was buffered or streamed. The wide table costs nothing
+        beyond the long one when every series has the same timestamps, and a pivot (about 1.7 s
+        per 100M rows) when they do not.
+        """
+
+
 class BinaryDatapoints:
     """One series' rows from ``TimeSeriesServiceSync.retrieve_datapoints_binary``, as an Arrow table.
 
@@ -787,7 +889,7 @@ class TimeSeriesServiceSync:
     def retrieve_datapoints(self, input: RetrieveFilter) -> list[DatapointsCollectionDatapoints]: ...
     def retrieve_datapoints_binary(
         self, input: RetrieveFilter | list[RetrieveFilter]
-    ) -> list[BinaryDatapoints]:
+    ) -> BinaryReadResult:
         """Read datapoints through ``POST /timeseries/data/list/binary``: the request
         ``retrieve_datapoints`` takes, answered as compressed Arrow frames instead of JSON.
 
@@ -802,9 +904,10 @@ class TimeSeriesServiceSync:
 
         Returns
         -------
-        list[BinaryDatapoints]
-            One per series that has points in its window, in the order of ``input``; a series
-            with none is left out rather than returned empty.
+        BinaryReadResult
+            A sequence of :class:`BinaryDatapoints`, one per series that has points in its window,
+            in the order of ``input`` (a series with none is left out rather than returned empty),
+            with ``to_polars(wide=False)`` for one table of them all.
 
         Raises
         ------
@@ -816,7 +919,37 @@ class TimeSeriesServiceSync:
 
         Notes
         -----
-        Sync client only, like the binary insert.
+        Sync client only, like the binary insert. The frames are decoded while the rest of the
+        response downloads; for rows as they arrive, or a window too large to hold, use
+        :meth:`retrieve_datapoints_binary_stream`.
+        """
+    def retrieve_datapoints_binary_stream(
+        self, input: RetrieveFilter | list[RetrieveFilter]
+    ) -> BinaryDatapointReader:
+        """The read :meth:`retrieve_datapoints_binary` makes, handed over as it arrives.
+
+        Parameters
+        ----------
+        input
+            As for :meth:`retrieve_datapoints_binary`.
+
+        Returns
+        -------
+        BinaryDatapointReader
+            Single use: iterate it for one :class:`BinaryDatapoints` per frame, or read it as an
+            Arrow stream of one long table.
+
+        Raises
+        ------
+        DataHubException
+            A refusal (400 for an unknown series, a ``cursor``, aggregates on a text series)
+            raises here, before there is a reader. A response that breaks off raises from the
+            reader instead, after the rows that did arrive: see :class:`BinaryDatapointReader`.
+
+        Notes
+        -----
+        Sync client only. Holding the reader holds the connection and the server's query open,
+        so read it through rather than leaving it half-consumed.
         """
     def delete_datapoints(self, input: list[DeleteFilter]) -> None: ...
     def retrieve_latest_datapoints(