Binary datapoints: read, stream, dataframe ingest and compression 2

AGENTS.md

@@ -199,9 +199,13 @@ 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 3; 9 compressed noisy floats no smaller than 1 and took several times as long) and posted through `execute_post_bytes_request`. `BinaryIngestOptions::compression` picks where the zstd goes: `FrameCompression::Buffers` (compression 2, the default: arrow-rs compresses each buffer in-band, the form a read answers with) or `FrameCompression::Payload` (compression 1: one stream around the Arrow payload). Buffers is the default because the consumer relays those batches into ClickHouse without decompressing them (platform `c9e9351dd`), while a compression 1 frame is decoded first: on 100M float points from polars, time until the last point was readable fell from 15.5 s to 13.2 s, and the client's cost was the same. At compression 2 the envelope's raw length is what the server recounts — every record batch buffer uncompressed, padded to 8 — so `in_band_raw_len` reads it back from the written stream rather than predicting which buffers arrow-rs writes; a mismatch is a 400. The api only takes compression 2 on ingest from platform `357794cce` on. 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` and `compression` (`"buffers"`, the default, or `"payload"`); 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`. Read frames are **compression 2**: the payload is an Arrow IPC stream whose record batches zstd-compress each buffer in-band (Arrow's body compression), because the api relays ClickHouse's compressed batches without decoding them; `arrow-ipc`'s `zstd` feature is what decodes that, and a frame holds whole batches up to the row cap, so most hold fewer rows than the cap. Compression 1 (one zstd stream around the payload, what ingest sends) is still read. At compression 2 the envelope's raw length is the batches' buffers uncompressed, which arrow-rs does not report, so only its cap is checked. 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.relayStream` (`read_*_zstd.bin` are the compression 1 bodies its former `wrap` wrote), 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.
+
+`TimeSeriesService::insert_table_binary` (`src/timeseries/binary_table.rs`) is the third ingest entry point, Arrow in: a wide `RecordBatch` (a timestamp column plus one column per series, named by external id). It shares series resolution and its cache, request packing and the stale-series retry with the string path (`resolve_series`, `post_frame_requests`, `assemble_frame` and `ipc_stream` in `binary.rs`), and stores what the string path stores: its offline tests decode both paths' frames with the read parser and compare. Columns already in the canon's type are sliced, not copied; nulls are dropped as "no point"; unsorted rows are sorted and a repeated timestamp keeps its last value; timestamps must carry a time zone and whole milliseconds. Frames hold one value type, series in ascending id order, short series sharing a frame, and are compressed on scoped threads. A `mixed` series takes a column as typed (numeric → `value_numeric`, string → `value_text`), unlike the string path's per-value classification. In Python it is `insert_dataframe_binary` (sync client), importing anything with `__arrow_c_stream__` zero-copy; with no `timestamp=` it takes the pandas index if that is a timestamp (pandas records it in the schema metadata, and other index columns are dropped as not series), then a `timestamp` column, then the only timestamp-typed column. 100M float points from polars took ~3 s with the client using ~3.6 cores, so the api, not the client, sets that pace; `python_tests/test_timeseries_binary_dataframe.py` round-trips offline through a stand-in that replays the posted frames, and live through `to_polars(wide=True)`.
 
 ### 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"
@@ -44,7 +45,8 @@ geojson = "1"
 base64 = "0.22"
 arrow-array = "59.3"
 arrow-schema = "59.3"
-arrow-ipc = "59.3"
+arrow-ipc = { version = "59.3", features = ["zstd"] }
+arrow-select = "59.3"
 
 
 #[lib]

README.md

@@ -124,7 +124,7 @@ the same collections as `insert_datapoints` and sends them through
 `POST /timeseries/data/binary`: each series is resolved once to its id and value type and
 cached, values are checked against that type locally, sorted and de-duplicated, cut into Arrow
 IPC frames at the contract's caps (100 000 points per numeric frame, 10 000 per text or mixed
-frame, 32 frames per request), compressed with zstd (level 9 by default, 1 and 3 are the other
+frame, 32 frames per request), compressed with zstd (level 3 by default, 1 and 9 are the other
 choices) and posted. A 204 means every frame was accepted.
 
 A series that does not exist is a 404 before anything is sent, a value that does not fit its
@@ -139,6 +139,81 @@ 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.
 
+### From a dataframe
+
+`time_series.insert_table_binary(&batches, "timestamp", &options)` takes Arrow `RecordBatch`es in
+the wide shape: a timestamp column and one column per series, named by its external id. In Python
+that is `insert_dataframe_binary`, which takes a polars or pandas DataFrame or a pyarrow table as
+it is, through the Arrow C stream:
+
+```python
+df = pl.DataFrame({"timestamp": stamps, "pump.temp": temps, "pump.count": counts})
+client.timeseries.insert_dataframe_binary(df)        # or a pandas frame with a timestamp index
+```
+
+Columns are used without copying where they already have their series' type; nulls are "no
+point here", so a ragged table works; values are stored as the string path stores them. 100M float
+points from a polars frame took about 3 s, against about 75 s through Python lists. What
+`to_polars(wide=True)` returns from a read goes back in unchanged.
+
+## 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.
+
+`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/` wraps this SDK as the Python package `intellistream-datahub-sdk` (import name

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"
@@ -25,3 +26,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"