Read datapoints through the binary endpoint

AGENTS.md

@@ -199,9 +199,11 @@ at-least-once path is still `SubscriptionsService::listen`. Two things differ fr
 `recommend_value_type` sits beside it: a bare `ValueTypeRecommendation`, and an unknown unit is
 the generic default with `recognized: false`, not an error.
 
-### Binary datapoint ingest (`src/timeseries/binary.rs`)
+### Binary datapoints (`src/timeseries/binary.rs`, `src/timeseries/binary_read.rs`)
 
-`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`) exist for this path and are the seed of the Arrow read path. 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.
+`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 `ApiServiceProvider` trait (`src/generic.rs`)
 

README.md

@@ -139,6 +139,41 @@ and sends again, once. Resolving by external id goes through
 The durable spool does not cover this path. `binary::FrameWriter` is public for producers that
 build frames themselves.
 
+## Binary datapoint read
+
+`time_series.retrieve_datapoints_binary(&filters)` takes the request `retrieve_datapoints`
+takes and reads the same frames back from `POST /timeseries/data/list/binary`, so a large read
+costs no JSON and no object per point. It returns one `BinaryDatapoints` per series that has
+points, in the order of the request's items: `timestamps` in epoch milliseconds and `values` as
+one typed column (`BinaryValues::Float(Vec<f64>)`, exact `Decimal { unscaled, scale }` for
+`numeric` and `decimal32`, two optional columns for `mixed`, or `Aggregates { avg, min, max,
+sum }` for an item with `aggregates` and a `granularity`).
+
+```rust
+let mut filter = RetrieveFilter::default();
+filter.external_id = Some("engine.temp".to_string());
+filter.start = Some(Utc.with_ymd_and_hms(2026, 10, 1, 0, 0, 0).unwrap());
+filter.end = Some(Utc.with_ymd_and_hms(2026, 10, 2, 0, 0, 0).unwrap());
+for series in api.time_series.retrieve_datapoints_binary(&DataWrapper::from_vec(vec![filter])).await? {
+    if let BinaryValues::Float(values) = &series.values {
+        println!("{}: {} points", series.external_id, values.len());
+    }
+}
+```
+
+An item with a `limit` (at most 100 000) returns that many of the oldest points from `start`;
+an item without one returns its whole window, so bound a long series with `end`, as the whole
+answer is held in memory. There is no cursor: to read on in pieces, send the request again with
+`start` one millisecond past the last timestamp you got. A series that does not exist is a 400 for the whole request, where
+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.
+
+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.
+
 ## Python bindings
 
 `datahub_python_bindings/` wraps this SDK as the Python package `intellistream-datahub-sdk` (import name

datahub_python_bindings/Cargo.toml

@@ -25,3 +25,6 @@ pyo3-async-runtimes = { version = "0.29", features = ["tokio-runtime"] }
 numpy = "0.29"
 uuid = { version = "1.22.0", features = ["v4"] }
 strum = { version = "0.28", features = ["derive"] }
+# `ffi` for the Arrow PyCapsule stream; the SDK already depends on both crates.
+arrow-array = { version = "59.3", features = ["ffi"] }
+arrow-schema = "59.3"

datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi

@@ -590,6 +590,43 @@ class DatapointsCollectionDatapoints:
     def id(self) -> int | None: ...
 
 
+class BinaryDatapoints:
+    """One series' rows from ``TimeSeriesServiceSync.retrieve_datapoints_binary``, as an Arrow table.
+
+    Implements the Arrow PyCapsule stream protocol, so ``polars.DataFrame(series)``,
+    ``pyarrow.table(series)`` and ``pandas.DataFrame.from_arrow(series)`` (pandas 3) take it
+    without copying. Every call makes a new stream, so one object converts more than once.
+
+    Columns are ``timestamp`` (``timestamp[ms, UTC]``; the bucket start for aggregates) and then,
+    by value type: ``value`` as int64 (``bigint``), float64 (``float``), float32 (``float32``),
+    ``decimal128(18, 6)`` (``numeric``), ``decimal128(9, 4)`` (``decimal32``) or utf8 (``text``);
+    nullable ``value_numeric`` and ``value_text`` for ``mixed``, each row setting one; or
+    ``avg``, ``min``, ``max`` and ``sum`` as float64 when :attr:`is_aggregates`.
+    """
+
+    @property
+    def id(self) -> int: ...
+    @property
+    def external_id(self) -> str: ...
+    @property
+    def value_type(self) -> str:
+        """The series' value type: ``bigint``, ``float``, ``float32``, ``numeric``, ``decimal32``,
+        ``text`` or ``mixed``. An aggregate read reports the series' own type."""
+    @property
+    def is_aggregates(self) -> bool:
+        """Whether the rows are buckets with ``avg``, ``min``, ``max`` and ``sum``."""
+    @property
+    def columns(self) -> list[str]: ...
+    def __len__(self) -> int: ...
+    def __arrow_c_stream__(self, requested_schema: object | None = None) -> object:
+        """The Arrow PyCapsule stream protocol. ``requested_schema`` is ignored."""
+    def to_polars(self) -> Any:
+        """A ``polars.DataFrame``. Needs polars; raises ``ImportError`` without it."""
+    def to_pandas(self) -> Any:
+        """A ``pandas.DataFrame``, converted through pyarrow. Needs pandas and pyarrow; raises
+        ``ImportError`` without pyarrow."""
+
+
 class RetrieveFilter:
     def __init__(
         self,
@@ -748,6 +785,39 @@ class TimeSeriesServiceSync:
         ts: Identifiable,
     ) -> list[str]: ...
     def retrieve_datapoints(self, input: RetrieveFilter) -> list[DatapointsCollectionDatapoints]: ...
+    def retrieve_datapoints_binary(
+        self, input: RetrieveFilter | list[RetrieveFilter]
+    ) -> list[BinaryDatapoints]:
+        """Read datapoints through ``POST /timeseries/data/list/binary``: the request
+        ``retrieve_datapoints`` takes, answered as compressed Arrow frames instead of JSON.
+
+        Parameters
+        ----------
+        input
+            One filter or a list. Each needs a ``start`` or an ``end``. With a ``limit`` (at most
+            100 000) it returns that many of the oldest points from there; without one it returns
+            every point in its window, so bound a long series with ``end``, as the whole answer is
+            held in memory. With ``aggregates`` and a ``granularity``
+            the series comes back as buckets carrying all four aggregates, whichever were named.
+
+        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.
+
+        Raises
+        ------
+        DataHubException
+            400 for a ``cursor`` (there is no paging: send the request again with ``start`` one
+            millisecond past the last timestamp read), for aggregates on a text series, and for
+            a series that does not exist or that you cannot read, where the JSON read answers
+            empty. 502 when the response was cut short; sending it again is safe.
+
+        Notes
+        -----
+        Sync client only, like the binary insert.
+        """
     def delete_datapoints(self, input: list[DeleteFilter]) -> None: ...
     def retrieve_latest_datapoints(
         self, input: list[Identifiable]