diff --git a/AGENTS.md b/AGENTS.md index 6a153fa..7205115 100644 --- a/AGENTS.md +++ b/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` 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` 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` 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`) diff --git a/Cargo.toml b/Cargo.toml index 885d051..fc2916f 100644 --- a/Cargo.toml +++ b/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" diff --git a/README.md b/README.md index a228fe0..0b1c6d7 100644 --- a/README.md +++ b/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 diff --git a/datahub_python_bindings/Cargo.toml b/datahub_python_bindings/Cargo.toml index cae7243..b6cbb3a 100644 --- a/datahub_python_bindings/Cargo.toml +++ b/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" diff --git a/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi b/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi index 0359bba..d2e39e6 100644 --- a/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi +++ b/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( diff --git a/datahub_python_bindings/python/intellistream_datahub_sdk/_tables.py b/datahub_python_bindings/python/intellistream_datahub_sdk/_tables.py new file mode 100644 index 0000000..81fba15 --- /dev/null +++ b/datahub_python_bindings/python/intellistream_datahub_sdk/_tables.py @@ -0,0 +1,60 @@ +"""Polars tables from a binary read, shared by the buffered result and the streamed reader so the +two cannot drift: the long table, and its wide form with one column per series.""" + + +def _polars(method): + try: + import polars + except ImportError: + raise ImportError(f"{method}() needs polars; install it with `pip install polars`") from None + return polars + + +def to_polars(read, wide): + """`read` is a `BinaryReadResult` or a `BinaryDatapointReader`: an Arrow stream of the long + table, and an iterable of `BinaryDatapoints` (whole series, or a reader's per-frame chunks).""" + pl = _polars("to_polars") + if not wide: + return pl.DataFrame(read) + return _wide(pl, read) + + +def _wide(pl, parts): + # Chunks of one series arrive consecutively; a dict keyed by id keeps the read's order. + series = {} + for part in parts: + external_id, frames = series.setdefault(part.id, (part.external_id, [])) + frames.append(part.to_polars()) + if not series: + raise ValueError("the read returned no rows, so there are no columns for a wide table") + + columns = [] + for external_id, frames in series.values(): + frame = pl.concat(frames, rechunk=False) if len(frames) > 1 else frames[0] + if not frame["timestamp"].is_sorted() or frame["timestamp"].n_unique() != frame.height: + raise ValueError( + f"series {external_id} has repeated timestamps, as when one series is asked for " + f"twice with overlapping windows; a wide table needs one value per timestamp" + ) + values = [c for c in frame.columns if c != "timestamp"] + names = {values[0]: external_id} if len(values) == 1 else {c: f"{external_id}.{c}" for c in values} + columns.append(frame.rename(names)) + + first = columns[0]["timestamp"] + if all(c.height == first.len() and c["timestamp"].equals(first) for c in columns[1:]): + # Every series on the same timestamps: the columns side by side, nothing to match up. + return columns[0].hstack([s for c in columns[1:] for s in c.drop("timestamp").get_columns()]) + + single_values = [c.columns[1:] for c in columns] + if all(len(v) == 1 for v in single_values) and len({c.dtypes[1] for c in columns}) == 1: + # One value column of one type: a pivot on a numeric key is the fastest way to match rows. + names = [v[0] for v in single_values] + long = pl.concat( + [c.rename({n: "value"}).with_columns(pl.lit(i, pl.Int32).alias("series")) for i, (c, n) in enumerate(zip(columns, names))], + rechunk=False, + ) + wide = long.pivot(on="series", index="timestamp", values="value").sort("timestamp") + return wide.rename({str(i): n for i, n in enumerate(names)}).select(["timestamp", *names]) + + # Different value types or several value columns: matched on timestamp, each keeping its type. + return pl.concat(columns, how="align") diff --git a/datahub_python_bindings/src/timeseries/binary_read.rs b/datahub_python_bindings/src/timeseries/binary_read.rs index 8a044f6..e4aefd5 100644 --- a/datahub_python_bindings/src/timeseries/binary_read.rs +++ b/datahub_python_bindings/src/timeseries/binary_read.rs @@ -1,15 +1,21 @@ use std::sync::Arc; use arrow_array::ffi_stream::FFI_ArrowArrayStream; +use std::sync::Mutex; + use arrow_array::{ - ArrayRef, Decimal128Array, Float32Array, Float64Array, Int64Array, RecordBatch, - RecordBatchIterator, StringArray, TimestampMillisecondArray, + ArrayRef, Decimal128Array, Float32Array, Float64Array, Int64Array, + RecordBatch, RecordBatchIterator, RecordBatchReader, StringArray, TimestampMillisecondArray, +}; +use arrow_schema::{ArrowError, DataType, Field, Schema, SchemaRef, TimeUnit}; +use futures::StreamExt; +use intellistream_datahub_sdk::timeseries::{ + BinaryDatapointStream, BinaryDatapoints, BinaryValues, DatapointValueType, }; -use arrow_schema::{ArrowError, DataType, Field, Schema, TimeUnit}; -use intellistream_datahub_sdk::timeseries::{BinaryDatapoints, BinaryValues, DatapointValueType}; -use pyo3::exceptions::{PyImportError, PyValueError}; +use tokio::runtime::Runtime; +use pyo3::exceptions::{PyImportError, PyRuntimeError, PyStopIteration, PyValueError}; use pyo3::prelude::*; -use pyo3::types::PyCapsule; +use pyo3::types::{PyCapsule, PyDict, PyList}; /// One series' rows from `retrieve_datapoints_binary`, as an Arrow table: `timestamp` and the /// value columns of its type. Read it with anything that takes the Arrow PyCapsule protocol. @@ -22,49 +28,59 @@ pub struct PyBinaryDatapoints { batch: RecordBatch, } +/// A series' `timestamp` and value columns, as Arrow arrays that take over its vectors. +fn columns( + value_type: DatapointValueType, + timestamps: Vec, + values: BinaryValues, +) -> Result<(Vec, Vec), ArrowError> { + let timestamp = Field::new( + "timestamp", + DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())), + false, + ); + let mut fields = vec![timestamp]; + let mut columns: Vec = + vec![Arc::new(TimestampMillisecondArray::from(timestamps).with_timezone("UTC"))]; + let mut push = |name: &str, nullable: bool, array: ArrayRef| { + fields.push(Field::new(name, array.data_type().clone(), nullable)); + columns.push(array); + }; + match values { + BinaryValues::Bigint(v) => push("value", false, Arc::new(Int64Array::from(v))), + BinaryValues::Float(v) => push("value", false, Arc::new(Float64Array::from(v))), + BinaryValues::Float32(v) => push("value", false, Arc::new(Float32Array::from(v))), + BinaryValues::Decimal { unscaled, scale } => { + let precision = if value_type == DatapointValueType::Decimal32 { 9 } else { 18 }; + let array = Decimal128Array::from(unscaled).with_precision_and_scale(precision, scale)?; + push("value", false, Arc::new(array)) + } + BinaryValues::Text(v) => push("value", false, Arc::new(StringArray::from(v))), + BinaryValues::Mixed { numeric, text } => { + push("value_numeric", true, Arc::new(Float64Array::from(numeric))); + push("value_text", true, Arc::new(StringArray::from(text))); + } + BinaryValues::Aggregates { avg, min, max, sum } => { + push("avg", false, Arc::new(Float64Array::from(avg))); + push("min", false, Arc::new(Float64Array::from(min))); + push("max", false, Arc::new(Float64Array::from(max))); + push("sum", false, Arc::new(Float64Array::from(sum))); + } + other => { + return Err(ArrowError::NotYetImplemented(format!( + "binary values {other:?} have no Arrow mapping" + ))) + } + } + Ok((fields, columns)) +} + impl TryFrom for PyBinaryDatapoints { type Error = ArrowError; fn try_from(series: BinaryDatapoints) -> Result { let aggregates = series.is_aggregates(); - let timestamp = Field::new( - "timestamp", - DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())), - false, - ); - let mut fields = vec![timestamp]; - let mut columns: Vec = - vec![Arc::new(TimestampMillisecondArray::from(series.timestamps).with_timezone("UTC"))]; - let mut push = |name: &str, nullable: bool, array: ArrayRef| { - fields.push(Field::new(name, array.data_type().clone(), nullable)); - columns.push(array); - }; - match series.values { - BinaryValues::Bigint(v) => push("value", false, Arc::new(Int64Array::from(v))), - BinaryValues::Float(v) => push("value", false, Arc::new(Float64Array::from(v))), - BinaryValues::Float32(v) => push("value", false, Arc::new(Float32Array::from(v))), - BinaryValues::Decimal { unscaled, scale } => { - let precision = if series.value_type == DatapointValueType::Decimal32 { 9 } else { 18 }; - let array = Decimal128Array::from(unscaled).with_precision_and_scale(precision, scale)?; - push("value", false, Arc::new(array)) - } - BinaryValues::Text(v) => push("value", false, Arc::new(StringArray::from(v))), - BinaryValues::Mixed { numeric, text } => { - push("value_numeric", true, Arc::new(Float64Array::from(numeric))); - push("value_text", true, Arc::new(StringArray::from(text))); - } - BinaryValues::Aggregates { avg, min, max, sum } => { - push("avg", false, Arc::new(Float64Array::from(avg))); - push("min", false, Arc::new(Float64Array::from(min))); - push("max", false, Arc::new(Float64Array::from(max))); - push("sum", false, Arc::new(Float64Array::from(sum))); - } - other => { - return Err(ArrowError::NotYetImplemented(format!( - "binary values {other:?} have no Arrow mapping" - ))) - } - } + let (fields, columns) = columns(series.value_type, series.timestamps, series.values)?; let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?; Ok(Self { id: series.id, @@ -76,6 +92,19 @@ impl TryFrom for PyBinaryDatapoints { } } +/// One frame's rows as a batch of the long table: `timeseries_id` ahead of the series' own +/// columns. No external id column: as a dictionary column polars spent 0.8 s per 100M rows making +/// it categorical, more than the rest of the conversion; `external_ids` maps the ids instead. +fn long_batch(series: BinaryDatapoints) -> Result { + let id = Arc::new(Int64Array::from(vec![series.id as i64; series.len()])) as ArrayRef; + let (fields, columns) = columns(series.value_type, series.timestamps, series.values)?; + let mut all_fields = vec![Field::new("timeseries_id", DataType::Int64, false)]; + all_fields.extend(fields); + let mut all_columns = vec![id]; + all_columns.extend(columns); + RecordBatch::try_new(Arc::new(Schema::new(all_fields)), all_columns) +} + #[pymethods] impl PyBinaryDatapoints { #[getter] @@ -149,6 +178,294 @@ impl PyBinaryDatapoints { } } +/// A binary read as it arrives, from `retrieve_datapoints_binary_stream`. Single use: read it +/// either by iterating, one `BinaryDatapoints` per frame, or as an Arrow stream of the long table. +#[pyclass(module = "intellistream_datahub_sdk", name = "BinaryDatapointReader", frozen)] +pub struct PyBinaryDatapointReader { + runtime: Arc, + state: Mutex, + seen: Seen, +} + +/// The series read so far, by id, in the order they first came. +type Seen = Arc>>; + +fn note(seen: &Seen, series: &BinaryDatapoints) { + let mut seen = seen.lock().unwrap(); + if !seen.iter().any(|(id, _)| *id == series.id) { + seen.push((series.id, series.external_id.clone())); + } +} + +enum ReaderState { + Unread(BinaryDatapointStream), + Iterating(BinaryDatapointStream), + Done, +} + +impl PyBinaryDatapointReader { + pub(crate) fn new(runtime: Arc, stream: BinaryDatapointStream) -> Self { + Self { + runtime, + state: Mutex::new(ReaderState::Unread(stream)), + seen: Arc::default(), + } + } +} + +#[pymethods] +impl PyBinaryDatapointReader { + fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> { + slf + } + + /// The next frame's rows: a chunk of one series, which may continue in the next item. + fn __next__(&self, py: Python<'_>) -> PyResult { + let mut state = self.state.lock().unwrap(); + let mut stream = match std::mem::replace(&mut *state, ReaderState::Done) { + ReaderState::Unread(stream) | ReaderState::Iterating(stream) => stream, + ReaderState::Done => return Err(PyStopIteration::new_err(())), + }; + let item = py.detach(|| self.runtime.block_on(stream.next())); + match item { + Some(Ok(part)) => { + *state = ReaderState::Iterating(stream); + note(&self.seen, &part); + PyBinaryDatapoints::try_from(part).map_err(|e| PyValueError::new_err(e.to_string())) + } + Some(Err(error)) => Err(crate::datahub_err(error)), + None => Err(PyStopIteration::new_err(())), + } + } + + /// `{timeseries_id: external_id}` for the series read so far, in the order they came: the key + /// for the long table's `timeseries_id`. Complete once the reader is exhausted. + #[getter] + fn external_ids<'py>(&self, py: Python<'py>) -> PyResult> { + let ids = PyDict::new(py); + for (id, external_id) in self.seen.lock().unwrap().iter() { + ids.set_item(id, external_id)?; + } + Ok(ids) + } + + /// The Arrow PyCapsule stream protocol, as one long table: `timeseries_id`, `timestamp` and + /// the value columns, a batch per frame. Every series in the read has to + /// share those columns; one that does not is an error at its batch. A read with no rows has + /// no schema to declare and raises. + #[pyo3(signature = (requested_schema=None))] + fn __arrow_c_stream__<'py>( + &self, + py: Python<'py>, + requested_schema: Option>, + ) -> PyResult> { + let _ = requested_schema; + let mut stream = { + let mut state = self.state.lock().unwrap(); + match std::mem::replace(&mut *state, ReaderState::Done) { + ReaderState::Unread(stream) => stream, + other => { + *state = other; + return Err(PyRuntimeError::new_err( + "this reader has already been read; a binary read streams once", + )); + } + } + }; + let first = match py.detach(|| self.runtime.block_on(stream.next())) { + Some(Ok(part)) => { + note(&self.seen, &part); + long_batch(part).map_err(|e| PyValueError::new_err(e.to_string()))? + } + Some(Err(error)) => return Err(crate::datahub_err(error)), + None => { + return Err(PyValueError::new_err( + "the read returned no rows, so there is no schema for an Arrow stream", + )) + } + }; + let reader = LongTable { + schema: first.schema(), + first: Some(first), + stream, + runtime: self.runtime.clone(), + seen: self.seen.clone(), + }; + let ffi = FFI_ArrowArrayStream::new(Box::new(reader)); + PyCapsule::new_with_value(py, ffi, c"arrow_array_stream") + } + + /// A `polars.DataFrame`: the long table, or with `wide` one column per series, matched on + /// timestamp. Needs polars. Reads the reader, like any other use of it. + #[pyo3(signature = (wide=false))] + fn to_polars<'py>(slf: &Bound<'py, Self>, wide: bool) -> PyResult> { + tables_to_polars(slf.as_any(), wide) + } +} + +fn tables_to_polars<'py>(read: &Bound<'py, PyAny>, wide: bool) -> PyResult> { + read.py() + .import("intellistream_datahub_sdk._tables")? + .call_method1("to_polars", (read, wide)) +} + +/// A buffered binary read, from `retrieve_datapoints_binary`: the series it returned, one +/// `BinaryDatapoints` each, which it behaves as a list of, and the same tables the streamed +/// reader gives. +#[pyclass(module = "intellistream_datahub_sdk", name = "BinaryReadResult", frozen, sequence)] +pub struct PyBinaryReadResult { + series: Vec>, +} + +impl PyBinaryReadResult { + pub(crate) fn new(py: Python<'_>, series: Vec) -> PyResult { + let series = series + .into_iter() + .map(|s| { + let s = PyBinaryDatapoints::try_from(s).map_err(|e| PyValueError::new_err(e.to_string()))?; + Py::new(py, s) + }) + .collect::>()?; + Ok(Self { series }) + } +} + +#[pymethods] +impl PyBinaryReadResult { + fn __len__(&self) -> usize { + self.series.len() + } + + fn __getitem__(&self, py: Python<'_>, index: isize) -> PyResult> { + let len = self.series.len() as isize; + let at = if index < 0 { index + len } else { index }; + if at < 0 || at >= len { + return Err(pyo3::exceptions::PyIndexError::new_err("series index out of range")); + } + Ok(self.series[at as usize].clone_ref(py)) + } + + fn __iter__<'py>(&self, py: Python<'py>) -> PyResult> { + Ok(PyList::new(py, &self.series)?.into_any().try_iter()?.into_any()) + } + + fn __repr__(&self) -> String { + let rows: usize = self.series.iter().map(|s| s.get().batch.num_rows()).sum(); + format!("BinaryReadResult(series={}, rows={rows})", self.series.len()) + } + + /// `{timeseries_id: external_id}` for the series in the read, in order: the key to the long + /// table's `timeseries_id`. + #[getter] + fn external_ids<'py>(&self, py: Python<'py>) -> PyResult> { + let ids = PyDict::new(py); + for s in &self.series { + ids.set_item(s.get().id, &s.get().external_id)?; + } + Ok(ids) + } + + /// The Arrow PyCapsule stream protocol, as the long table: `timeseries_id`, `timestamp` and + /// the value columns, a batch per series, which all have to share those columns. A read with + /// no rows has no schema and raises. + #[pyo3(signature = (requested_schema=None))] + fn __arrow_c_stream__<'py>( + &self, + py: Python<'py>, + requested_schema: Option>, + ) -> PyResult> { + let _ = requested_schema; + let mut batches = Vec::with_capacity(self.series.len()); + for s in &self.series { + let s = s.get(); + let id = Arc::new(Int64Array::from(vec![s.id as i64; s.batch.num_rows()])) as ArrayRef; + let mut fields = vec![Arc::new(Field::new("timeseries_id", DataType::Int64, false))]; + fields.extend(s.batch.schema().fields().iter().cloned()); + let mut columns = vec![id]; + columns.extend(s.batch.columns().iter().cloned()); + let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns) + .map_err(|e| PyValueError::new_err(e.to_string()))?; + if let Some(first) = batches.first() { + let first: &RecordBatch = first; + if first.schema().fields() != batch.schema().fields() { + return Err(PyValueError::new_err(format!( + "series {} has columns {:?}, the table's are {:?}; series of different value \ + types do not share one long table, but to_polars(wide=True) holds them", + s.external_id, + batch.schema().fields().iter().map(|f| f.name()).collect::>(), + first.schema().fields().iter().map(|f| f.name()).collect::>(), + ))); + } + } + batches.push(batch); + } + let Some(first) = batches.first() else { + return Err(PyValueError::new_err( + "the read returned no rows, so there is no schema for an Arrow stream", + )); + }; + let schema = first.schema(); + let reader = RecordBatchIterator::new(batches.into_iter().map(Ok), schema); + PyCapsule::new_with_value(py, FFI_ArrowArrayStream::new(Box::new(reader)), c"arrow_array_stream") + } + + /// A `polars.DataFrame`: the long table, or with `wide` one column per series, matched on + /// timestamp. Needs polars. + #[pyo3(signature = (wide=false))] + fn to_polars<'py>(slf: &Bound<'py, Self>, wide: bool) -> PyResult> { + tables_to_polars(slf.as_any(), wide) + } +} + +/// The long table behind `__arrow_c_stream__`: the frame peeked for its schema, then the rest as +/// the consumer asks. Waits for the network with the GIL held, as the C stream gives no way to +/// release it; the reader belongs to the sync client. +struct LongTable { + schema: SchemaRef, + first: Option, + stream: BinaryDatapointStream, + runtime: Arc, + seen: Seen, +} + +impl Iterator for LongTable { + type Item = Result; + + fn next(&mut self) -> Option { + if let Some(first) = self.first.take() { + return Some(Ok(first)); + } + let part = match self.runtime.block_on(self.stream.next())? { + Ok(part) => part, + Err(error) => { + let message = format!("{}: {}", error.get_status(), error.get_message()); + return Some(Err(ArrowError::ExternalError(message.into()))); + } + }; + note(&self.seen, &part); + let external_id = part.external_id.clone(); + let batch = match long_batch(part) { + Ok(batch) => batch, + Err(error) => return Some(Err(error)), + }; + if batch.schema().fields() != self.schema.fields() { + return Some(Err(ArrowError::SchemaError(format!( + "series {external_id} has columns {:?}, the stream's are {:?}; read series of \ + different value types with to_polars(wide=True), or by iterating", + batch.schema().fields().iter().map(|f| f.name()).collect::>(), + self.schema.fields().iter().map(|f| f.name()).collect::>(), + )))); + } + Some(Ok(batch)) + } +} + +impl RecordBatchReader for LongTable { + fn schema(&self) -> SchemaRef { + self.schema.clone() + } +} + fn import<'py>(py: Python<'py>, module: &str, method: &str) -> PyResult> { py.import(module).map_err(|_| { PyImportError::new_err(format!( @@ -157,9 +474,4 @@ fn import<'py>(py: Python<'py>, module: &str, method: &str) -> PyResult) -> PyResult> { - series - .into_iter() - .map(|s| PyBinaryDatapoints::try_from(s).map_err(|e| PyValueError::new_err(e.to_string()))) - .collect() -} + diff --git a/datahub_python_bindings/src/timeseries/mod.rs b/datahub_python_bindings/src/timeseries/mod.rs index 53422a7..6f15aff 100644 --- a/datahub_python_bindings/src/timeseries/mod.rs +++ b/datahub_python_bindings/src/timeseries/mod.rs @@ -471,6 +471,8 @@ pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/datahub_python_bindings/src/timeseries/sync_service.rs b/datahub_python_bindings/src/timeseries/sync_service.rs index 4303ea2..0d6f05a 100644 --- a/datahub_python_bindings/src/timeseries/sync_service.rs +++ b/datahub_python_bindings/src/timeseries/sync_service.rs @@ -4,7 +4,7 @@ use crate::timeseries::datapoints::{ }; use crate::{DatahubIdentity, Identifiable}; use crate::{PyIdCollection, PyRetrieveFilter}; -use crate::timeseries::binary_read::{self, PyBinaryDatapoints}; +use crate::timeseries::binary_read::{PyBinaryDatapointReader, PyBinaryReadResult}; use intellistream_datahub_sdk::generic::{DataWrapper, IdAndExtId}; use intellistream_datahub_sdk::{ApiService, TimeSeriesUpdateCollection}; use pyo3_async_runtimes::tokio::future_into_py; @@ -368,7 +368,7 @@ impl PyTimeSeriesServiceSync { &self, py: Python<'py>, input: RetrieveFilters, - ) -> PyResult> { + ) -> PyResult { let service = self.api_service.clone(); let filters: Vec = match input { RetrieveFilters::One(f) => vec![f.into()], @@ -380,7 +380,27 @@ impl PyTimeSeriesServiceSync { .block_on(service.time_series.retrieve_datapoints_binary(&wrapper)) .map_err(crate::datahub_err) })?; - binary_read::to_py(series) + PyBinaryReadResult::new(py, series) + } + /// The read `retrieve_datapoints_binary` makes, handed over as it arrives: a single-use + /// `BinaryDatapointReader`, iterated frame by frame or read as an Arrow stream. + fn retrieve_datapoints_binary_stream<'py>( + &self, + py: Python<'py>, + input: RetrieveFilters, + ) -> PyResult { + let service = self.api_service.clone(); + let filters: Vec = match input { + RetrieveFilters::One(f) => vec![f.into()], + RetrieveFilters::Many(fs) => fs.into_iter().map(Into::into).collect(), + }; + let wrapper = DataWrapper::::from_vec(filters); + let stream = py.detach(|| { + self.runtime + .block_on(service.time_series.stream_datapoints_binary(&wrapper)) + .map_err(crate::datahub_err) + })?; + Ok(PyBinaryDatapointReader::new(self.runtime.clone(), stream)) } fn delete_datapoints<'py>(&self, py: Python<'py>, input: Vec) -> PyResult<()> { let service = self.api_service.clone(); diff --git a/python_tests/test_timeseries_binary_read.py b/python_tests/test_timeseries_binary_read.py index 8e9c03d..669331b 100644 --- a/python_tests/test_timeseries_binary_read.py +++ b/python_tests/test_timeseries_binary_read.py @@ -9,6 +9,7 @@ import decimal import http.server import pathlib +import struct import threading import pytest @@ -117,7 +118,7 @@ def test_aggregates_convert_to_pandas(stand_in): def test_no_points_is_no_series(stand_in): client, server = stand_in server.body = b"DHDE" - assert client.timeseries.retrieve_datapoints_binary(_filter()) == [] + assert len(client.timeseries.retrieve_datapoints_binary(_filter())) == 0 def test_a_response_cut_short_raises_rather_than_returning_fewer_points(stand_in): @@ -141,6 +142,216 @@ def test_a_refusal_is_the_problem_the_server_sent(stand_in): assert excinfo.value.status_code == 400 +# --- streaming, offline ----------------------------------------------------------------------- + + +def _frames(body): + """The frames of a response body, each as its own bytes, without the end marker.""" + frames, pos = [], 0 + while body[pos:pos + 4] == b"DHDP": + directory, payload = struct.unpack_from(", + ) -> Result { + let stream = self + .rt + .block_on(self.api.time_series.stream_datapoints_binary(json))?; + Ok(BinaryDatapointIter { + stream, + rt: self.rt.clone(), + }) + } +} + +/// A binary read frame by frame; see +/// [`BinaryDatapointStream`](crate::timeseries::BinaryDatapointStream) for what an item is and how +/// a response that breaks off shows up. +pub struct BinaryDatapointIter { + stream: BinaryDatapointStream, + rt: Arc, +} + +impl Iterator for BinaryDatapointIter { + type Item = Result; + + fn next(&mut self) -> Option { + self.rt.block_on(futures::StreamExt::next(&mut self.stream)) + } } /// Blocking counterpart of [`crate::ResourceService`]. diff --git a/src/generic.rs b/src/generic.rs index 0220427..697eb6f 100644 --- a/src/generic.rs +++ b/src/generic.rs @@ -883,18 +883,19 @@ pub trait ApiServiceProvider { } /// `POST` a JSON body to an endpoint that answers with bytes of type `accept` rather than JSON - /// (currently only `/timeseries/data/list/binary`), and return the whole body. + /// (currently only `/timeseries/data/list/binary`), returning the response to read as it + /// arrives. /// /// `application/problem+json` is offered beside `accept` so a refusal reaches the caller as the /// problem document it is, rather than as Spring's 406 for a type the handler does not produce. - /// A body that breaks off while it is being read is a 502: the server answered 200 and then - /// failed to finish, which is not the caller's fault and is safe to send again. - async fn execute_post_request_for_bytes( + /// The status is checked here, before any of the body is read, so a refusal is always an + /// `Err` and never a stream. + async fn execute_post_request_for_stream( &self, path: &str, json: &J, accept: &str, - ) -> Result, ResponseError> { + ) -> Result { let token = self.get_token().await?; let response = self .get_api_service() @@ -907,30 +908,23 @@ pub trait ApiServiceProvider { .await .map_err(ResponseError::from_err)?; let status = response.status(); - if !status.is_success() { - let content_type = response - .headers() - .get(reqwest::header::CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - .map(str::to_string); - let error = ResponseError { - status, - message: response - .text() - .await - .unwrap_or_else(|_| "Failed to read response body".to_string()), - content_type, - }; - return Err(self.on_request_error(error, &token).await); - } - match response.bytes().await { - Ok(body) => Ok(body.into()), - Err(error) => Err(ResponseError { - status: http::StatusCode::BAD_GATEWAY, - message: format!("the response was cut short while it was being read: {error}"), - content_type: None, - }), + if status.is_success() { + return Ok(response); } + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_string); + let error = ResponseError { + status, + message: response + .text() + .await + .unwrap_or_else(|_| "Failed to read response body".to_string()), + content_type, + }; + Err(self.on_request_error(error, &token).await) } /// `GET` an endpoint that answers with bytes rather than JSON (currently only diff --git a/src/timeseries/binary_read.rs b/src/timeseries/binary_read.rs index c57a76a..9b71658 100644 --- a/src/timeseries/binary_read.rs +++ b/src/timeseries/binary_read.rs @@ -5,13 +5,16 @@ //! the marker was cut short, whether inside a frame or between two, and is an error rather than a //! shorter answer. Codec 2 frames carry bucketed aggregates instead of values. //! -//! [`parse_read_response`] turns a body into one [`BinaryDatapoints`] per series, columnar, with -//! the frames of one series merged. [`TimeSeriesService::retrieve_datapoints_binary`] posts the -//! request and parses the answer. +//! [`TimeSeriesService::stream_datapoints_binary`] hands the frames over as they arrive, decoded +//! on blocking threads while the rest download; [`TimeSeriesService::retrieve_datapoints_binary`] +//! collects that stream into one [`BinaryDatapoints`] per series. [`parse_read_response`] does the +//! same for a body already in memory. use std::io::Cursor; +use std::pin::Pin; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::task::{Context, Poll}; use arrow_array::{ Array, Decimal128Array, Float32Array, Float64Array, Int64Array, RecordBatch, StringArray, @@ -19,6 +22,9 @@ use arrow_array::{ }; use arrow_ipc::reader::StreamReader; use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; +use bytes::{Buf, Bytes, BytesMut}; +use futures::stream::{FuturesOrdered, Stream, StreamExt}; +use tokio::sync::mpsc; use oauth2::http::StatusCode; use crate::generic::{ApiServiceProvider, DataWrapper, RetrieveFilter}; @@ -182,43 +188,51 @@ fn read_varint(body: &[u8], pos: &mut usize) -> Option { /// Errors on anything short of a complete, well-formed answer: a missing end marker, bytes after /// it, a frame that breaks the envelope, or a payload outside its canon. pub fn parse_read_response(body: &[u8]) -> Result, String> { - let frames = scan_frames(body)?; + let mut assembler = FrameAssembler::new(); + assembler.push(body); + let mut frames = Vec::new(); + while let Some(frame) = assembler.next_frame()? { + frames.push(frame); + } + assembler.finish()?; let decoded = decode_frames(&frames)?; + merge_series(decoded.into_iter().flatten()) +} + +/// Merges consecutive parts of one series, the way they come off the frames, into one entry each. +fn merge_series( + parts: impl IntoIterator, +) -> Result, String> { // Grouped before merging so each series is allocated once at its final size: extending part // by part regrows and recopies a long series dozens of times. - let mut groups: Vec> = Vec::new(); - for (index, parts) in decoded.into_iter().enumerate() { - for part in parts { - let continues = groups.last().and_then(|g| g.last()).is_some_and(|(_, last)| { - last.id == part.id - && last.is_aggregates() == part.is_aggregates() - && last.timestamps.last() < part.timestamps.first() - }); - if continues { - groups.last_mut().unwrap().push((index, part)); - } else { - groups.push(vec![(index, part)]); - } + let mut groups: Vec> = Vec::new(); + for part in parts { + let continues = groups.last().and_then(|g| g.last()).is_some_and(|last| { + last.id == part.id + && last.is_aggregates() == part.is_aggregates() + && last.timestamps.last() < part.timestamps.first() + }); + if continues { + groups.last_mut().unwrap().push(part); + } else { + groups.push(vec![part]); } } groups.into_iter().map(merge).collect() } /// One series from its consecutive parts, in order. -fn merge(parts: Vec<(usize, BinaryDatapoints)>) -> Result { - let rows: usize = parts.iter().map(|(_, p)| p.len()).sum(); +fn merge(parts: Vec) -> Result { + let rows: usize = parts.iter().map(BinaryDatapoints::len).sum(); let mut parts = parts.into_iter(); - let (_, mut series) = parts.next().expect("a group has at least one part"); + let mut series = parts.next().expect("a group has at least one part"); series.timestamps.reserve_exact(rows - series.len()); series.values.reserve_exact(rows - series.len()); - for (index, part) in parts { + for part in parts { let BinaryDatapoints { timestamps, values, .. } = part; series.timestamps.extend(timestamps); if !series.values.append(values) { - return Err(format!( - "frame {index}: series {} changes shape between frames", - series.id - )); + return Err(format!("series {} changes shape between frames", series.id)); } } Ok(series) @@ -228,65 +242,80 @@ fn cut_short(index: usize) -> String { format!("the response ends without its end marker, at frame {index}; it was cut short") } -/// One frame's envelope, checked, with its payload still compressed. -struct FrameView<'a> { +/// Cuts a response into frames as its bytes arrive, in chunks of any size: a frame's header, its +/// directory and its payload may each be split anywhere. +struct FrameAssembler { + buf: BytesMut, index: usize, - value_type: DatapointValueType, - aggregates: bool, - rows: usize, - names: Vec<(i64, String)>, - payload: &'a [u8], - raw_len: usize, + ended: bool, } -/// Every frame up to the end marker, in order. -fn scan_frames(body: &[u8]) -> Result>, String> { - let mut frames = Vec::new(); - let mut pos = 0; - loop { - let rest = &body[pos..]; - if rest.starts_with(END_OF_RESPONSE) { - if rest.len() != END_OF_RESPONSE.len() { - return Err(format!( - "{} bytes after the end marker", - rest.len() - END_OF_RESPONSE.len() - )); +impl FrameAssembler { + fn new() -> Self { + FrameAssembler { + buf: BytesMut::new(), + index: 0, + ended: false, + } + } + + fn push(&mut self, chunk: &[u8]) { + self.buf.extend_from_slice(chunk); + } + + /// The next complete frame; `None` until more bytes arrive, and after the end marker. + fn next_frame(&mut self) -> Result, String> { + if !self.ended && self.buf.starts_with(END_OF_RESPONSE) { + self.ended = true; + self.buf.advance(END_OF_RESPONSE.len()); + } + if self.ended { + if !self.buf.is_empty() { + return Err(format!("{} bytes after the end marker", self.buf.len())); } - return Ok(frames); + return Ok(None); } - if rest.len() < HEADER_BYTES { - return Err(cut_short(frames.len())); + if self.buf.len() >= MAGIC.len() && !self.buf.starts_with(MAGIC) { + return Err(format!("frame {}: does not start with DHDP", self.index)); } - let (consumed, frame) = scan_frame(rest, frames.len())?; - frames.push(frame); - pos += consumed; + if self.buf.len() < HEADER_BYTES { + return Ok(None); + } + let total = check_header(&self.buf[..HEADER_BYTES], self.index)?; + if self.buf.len() < total { + return Ok(None); + } + let frame = FrameView::parse(self.buf.split_to(total).freeze(), self.index)?; + self.index += 1; + Ok(Some(frame)) + } + + /// After the last byte: the body has to have ended with its marker and nothing after it. + fn finish(&mut self) -> Result<(), String> { + if self.next_frame()?.is_some() || !self.ended { + return Err(cut_short(self.index)); + } + Ok(()) } } -/// One frame's envelope and directory: how many bytes it takes, and the view of it. -fn scan_frame(body: &[u8], index: usize) -> Result<(usize, FrameView<'_>), String> { +/// Checks a frame's 28-byte header and returns the frame's whole length. +fn check_header(header: &[u8], index: usize) -> Result { let fail = |message: String| format!("frame {index}: {message}"); - if &body[0..4] != MAGIC { - return Err(fail("does not start with DHDP".to_string())); + if header[4] != VERSION { + return Err(fail(format!("version {}", header[4]))); } - if body[4] != VERSION { - return Err(fail(format!("version {}", body[4]))); - } - let value_type = DatapointValueType::from_id(body[5]) - .ok_or_else(|| fail(format!("value type id {}", body[5])))?; - let aggregates = match body[6] { + let value_type = DatapointValueType::from_id(header[5]) + .ok_or_else(|| fail(format!("value type id {}", header[5])))?; + let aggregates = match header[6] { CODEC_ARROW_IPC => false, CODEC_ARROW_IPC_AGGREGATES => true, codec => return Err(fail(format!("codec {codec}"))), }; - if body[7] != COMPRESSION_ZSTD { - return Err(fail(format!("compression {}", body[7]))); - } - let rows = u32_at(body, 8); - let series_count = u32_at(body, 12); - let directory_len = u32_at(body, 16); - let payload_len = u32_at(body, 20); - let raw_len = u32_at(body, 24); + if header[7] != COMPRESSION_ZSTD { + return Err(fail(format!("compression {}", header[7]))); + } + let rows = u32_at(header, 8); let max_rows = if aggregates { MAX_AGGREGATE_ROWS } else { @@ -295,53 +324,72 @@ fn scan_frame(body: &[u8], index: usize) -> Result<(usize, FrameView<'_>), Strin if rows < 1 || rows > max_rows { return Err(fail(format!("{rows} rows, allowed 1 to {max_rows}"))); } + let raw_len = u32_at(header, 24); if raw_len > MAX_FRAME_RAW_BYTES { return Err(fail(format!( "payload is {raw_len} bytes raw, the cap is {MAX_FRAME_RAW_BYTES}" ))); } - let total = HEADER_BYTES + directory_len + payload_len; - if body.len() < total { - return Err(cut_short(index)); - } - - let directory = &body[HEADER_BYTES..HEADER_BYTES + directory_len]; - let mut names: Vec<(i64, String)> = Vec::with_capacity(series_count); - let mut at = 0; - for _ in 0..series_count { - let id = directory - .get(at..at + 8) - .map(|b| i64::from_le_bytes(b.try_into().unwrap())) - .ok_or_else(|| fail("directory is shorter than its series count".to_string()))?; - at += 8; - let len = read_varint(directory, &mut at) - .ok_or_else(|| fail("directory is shorter than its series count".to_string()))?; - let name = directory - .get(at..at + len) - .ok_or_else(|| fail("directory is shorter than its series count".to_string()))?; - at += len; - let name = String::from_utf8(name.to_vec()) - .map_err(|_| fail(format!("external id of series {id} is not UTF-8")))?; - names.push((id, name)); - } - if at != directory_len { - return Err(fail(format!("{} bytes after the directory", directory_len - at))); - } - let frame = FrameView { - index, - value_type, - aggregates, - rows, - names, - payload: &body[HEADER_BYTES + directory_len..total], - raw_len, - }; - Ok((total, frame)) + Ok(HEADER_BYTES + u32_at(header, 16) + u32_at(header, 20)) +} + +/// One complete frame whose header passed [`check_header`], its payload still compressed. Owns its +/// bytes, so it can be decoded on another thread. +struct FrameView { + index: usize, + value_type: DatapointValueType, + aggregates: bool, + rows: usize, + names: Vec<(i64, String)>, + bytes: Bytes, + payload_start: usize, + raw_len: usize, +} + +impl FrameView { + fn parse(bytes: Bytes, index: usize) -> Result { + let fail = |message: String| format!("frame {index}: {message}"); + let series_count = u32_at(&bytes, 12); + let directory_len = u32_at(&bytes, 16); + let directory = &bytes[HEADER_BYTES..HEADER_BYTES + directory_len]; + let mut names: Vec<(i64, String)> = Vec::with_capacity(series_count); + let mut at = 0; + for _ in 0..series_count { + let id = directory + .get(at..at + 8) + .map(|b| i64::from_le_bytes(b.try_into().unwrap())) + .ok_or_else(|| fail("directory is shorter than its series count".to_string()))?; + at += 8; + let len = read_varint(directory, &mut at) + .ok_or_else(|| fail("directory is shorter than its series count".to_string()))?; + let name = directory + .get(at..at + len) + .ok_or_else(|| fail("directory is shorter than its series count".to_string()))?; + at += len; + let name = String::from_utf8(name.to_vec()) + .map_err(|_| fail(format!("external id of series {id} is not UTF-8")))?; + names.push((id, name)); + } + if at != directory_len { + return Err(fail(format!("{} bytes after the directory", directory_len - at))); + } + let aggregates = bytes[6] == CODEC_ARROW_IPC_AGGREGATES; + Ok(FrameView { + index, + value_type: DatapointValueType::from_id(bytes[5]).expect("checked with the header"), + aggregates, + rows: u32_at(&bytes, 8), + names, + payload_start: HEADER_BYTES + directory_len, + raw_len: u32_at(&bytes, 24), + bytes, + }) + } } /// Decodes every frame, on as many threads as there are cores and frames, keeping their order. /// On failure, the error of the earliest frame that failed. -fn decode_frames(frames: &[FrameView<'_>]) -> Result>, String> { +fn decode_frames(frames: &[FrameView]) -> Result>, String> { let threads = std::thread::available_parallelism() .map_or(1, |n| n.get()) .min(frames.len()); @@ -349,32 +397,33 @@ fn decode_frames(frames: &[FrameView<'_>]) -> Result>, return frames.iter().map(decode_frame).collect(); } let next = AtomicUsize::new(0); - let mut decoded: Vec<(usize, Result, String>)> = std::thread::scope(|scope| { - let workers: Vec<_> = (0..threads) - .map(|_| { - scope.spawn(|| { - let mut done = Vec::new(); - loop { - let i = next.fetch_add(1, Ordering::Relaxed); - let Some(frame) = frames.get(i) else { return done }; - done.push((i, decode_frame(frame))); - } + let mut decoded: Vec<(usize, Result, String>)> = + std::thread::scope(|scope| { + let workers: Vec<_> = (0..threads) + .map(|_| { + scope.spawn(|| { + let mut done = Vec::new(); + loop { + let i = next.fetch_add(1, Ordering::Relaxed); + let Some(frame) = frames.get(i) else { return done }; + done.push((i, decode_frame(frame))); + } + }) }) - }) - .collect(); - workers - .into_iter() - .flat_map(|w| w.join().expect("a frame decoder panicked")) - .collect() - }); + .collect(); + workers + .into_iter() + .flat_map(|w| w.join().expect("a frame decoder panicked")) + .collect() + }); decoded.sort_unstable_by_key(|(i, _)| *i); decoded.into_iter().map(|(_, result)| result).collect() } /// One frame's rows, split into one part per series it names. -fn decode_frame(frame: &FrameView<'_>) -> Result, String> { +fn decode_frame(frame: &FrameView) -> Result, String> { let fail = |message: String| format!("frame {}: {message}", frame.index); - let raw = zstd::bulk::decompress(frame.payload, frame.raw_len) + let raw = zstd::bulk::decompress(&frame.bytes[frame.payload_start..], frame.raw_len) .map_err(|e| fail(format!("zstd: {e}")))?; if raw.len() != frame.raw_len { return Err(fail(format!( @@ -418,6 +467,122 @@ fn decode_frame(frame: &FrameView<'_>) -> Result, String> Ok(parts) } +type Item = Result; + +/// A binary read as it arrives: one [`BinaryDatapoints`] per frame, in the order the frames come, +/// which is the order of the request's items. Consecutive items may belong to the same series — +/// a series longer than one frame comes as several, oldest first — so an item is a *chunk* of a +/// series, not all of it. +/// +/// Frames are decoded on blocking threads, several at a time, while the rest are still +/// downloading. A refusal is not an item: it is the error [`TimeSeriesService::stream_datapoints_binary`] +/// returns before there is a stream. **A response that breaks off is an error item after the +/// items that did arrive**, so a consumer that acts on items as they come has to treat what it +/// already took as incomplete when one appears; the stream ends after it. Dropping the stream +/// closes the connection. +/// +/// Read it with [`futures::StreamExt::next`]. +pub struct BinaryDatapointStream { + items: mpsc::Receiver, + task: tokio::task::JoinHandle<()>, +} + +impl Stream for BinaryDatapointStream { + type Item = Item; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.items.poll_recv(cx) + } +} + +impl Drop for BinaryDatapointStream { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// Frames decoded at once. Past a handful the decoders only compete for memory bandwidth. +fn decoders() -> usize { + std::thread::available_parallelism().map_or(1, |n| n.get()).clamp(1, 8) +} + +fn unreadable(path: &str, message: String) -> ResponseError { + ResponseError { + status: StatusCode::BAD_GATEWAY, + message: format!("POST {path} returned an unreadable answer: {message}"), + content_type: None, + } +} + +/// Reads the body, cuts it into frames, decodes up to [`decoders`] of them at once and sends their +/// parts on in frame order. Whatever fails, the parts of the frames before it are sent first. +async fn produce(response: reqwest::Response, path: String, items: mpsc::Sender) { + let limit = decoders(); + let mut body = response.bytes_stream(); + let mut assembler = FrameAssembler::new(); + let mut pending = FuturesOrdered::new(); + let mut failure: Option = None; + let mut eof = false; + loop { + while failure.is_none() && pending.len() < limit { + match assembler.next_frame() { + Ok(Some(frame)) => { + pending.push_back(tokio::task::spawn_blocking(move || decode_frame(&frame))) + } + Ok(None) => break, + Err(message) => failure = Some(unreadable(&path, message)), + } + } + let reading = !eof && failure.is_none() && pending.len() < limit; + if pending.is_empty() && !reading { + let last = match failure { + Some(error) => Some(error), + None => assembler.finish().err().map(|m| unreadable(&path, m)), + }; + if let Some(error) = last { + let _ = items.send(Err(error)).await; + } + return; + } + tokio::select! { + biased; + Some(decoded) = pending.next(), if !pending.is_empty() => match decoded { + Ok(Ok(parts)) => { + for part in parts { + if items.send(Ok(part)).await.is_err() { + return; + } + } + } + // The earliest frame still in flight, so its error comes before any other. + Ok(Err(message)) => { + failure = Some(unreadable(&path, message)); + pending = FuturesOrdered::new(); + } + Err(join) => { + failure = Some(ResponseError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("decoding a frame failed: {join}"), + content_type: None, + }); + pending = FuturesOrdered::new(); + } + }, + chunk = body.next(), if reading => match chunk { + Some(Ok(bytes)) => assembler.push(&bytes), + Some(Err(error)) => { + failure = Some(ResponseError { + status: StatusCode::BAD_GATEWAY, + message: format!("the response was cut short while it was being read: {error}"), + content_type: None, + }) + } + None => eof = true, + }, + } + } +} + /// Appends a batch's rows to `parts`, one part per run of a series id. fn split_batch( batch: &RecordBatch, @@ -554,26 +719,49 @@ impl TimeSeriesService { /// read is a 400 for the whole request, not an empty result. A response the server could not /// finish is a 502 whose message says it was cut short, never fewer points; sending the /// request again is safe. + /// + /// The frames are decoded as they arrive, so this costs little more than the download; for + /// rows as they come, or a window too large to hold, use + /// [`stream_datapoints_binary`](Self::stream_datapoints_binary). pub async fn retrieve_datapoints_binary( &self, json: &DataWrapper, ) -> Result, ResponseError> { let path = format!("{}/data/list/binary", self.base_url); - let body = self - .execute_post_request_for_bytes(&path, json, MEDIA_TYPE) - .await?; - let parsed = tokio::task::spawn_blocking(move || parse_read_response(&body)) + let mut stream = self.stream_datapoints_binary(json).await?; + let mut parts = Vec::new(); + while let Some(part) = stream.next().await { + parts.push(part?); + } + tokio::task::spawn_blocking(move || merge_series(parts)) .await .map_err(|e| ResponseError { status: StatusCode::INTERNAL_SERVER_ERROR, - message: format!("decoding the binary read failed: {e}"), + message: format!("merging the binary read failed: {e}"), content_type: None, - })?; - parsed.map_err(|message| ResponseError { - status: StatusCode::BAD_GATEWAY, - message: format!("POST {path} returned an unreadable answer: {message}"), - content_type: None, - }) + })? + .map_err(|message| unreadable(&path, message)) + } + + /// The same read as [`retrieve_datapoints_binary`](Self::retrieve_datapoints_binary), handed + /// over frame by frame as the response arrives instead of collected: see + /// [`BinaryDatapointStream`] for what an item is and how a broken-off response shows up. + /// + /// The request is checked before there is a stream, so a refusal (a 400 for an unknown + /// series, a `cursor`, aggregates on a text series) is the `Err` here rather than an item. + /// Holding the stream holds the connection, and the server's query, open: a consumer that + /// stops reading for long can run into the server's timeouts. + pub async fn stream_datapoints_binary( + &self, + json: &DataWrapper, + ) -> Result { + let path = format!("{}/data/list/binary", self.base_url); + let response = self + .execute_post_request_for_stream(&path, json, MEDIA_TYPE) + .await?; + let (tx, items) = mpsc::channel(2 * decoders()); + let task = tokio::spawn(produce(response, path, tx)); + Ok(BinaryDatapointStream { items, task }) } } @@ -739,6 +927,129 @@ mod tests { assert!(error.starts_with("frame 17:"), "{error}"); } + fn assemble_in_chunks(body: &[u8], chunk: usize) -> Result, String> { + let mut assembler = FrameAssembler::new(); + let mut frames = Vec::new(); + for piece in body.chunks(chunk) { + assembler.push(piece); + while let Some(frame) = assembler.next_frame()? { + frames.push(frame); + } + } + assembler.finish()?; + merge_series(decode_frames(&frames)?.into_iter().flatten()) + } + + #[test] + fn a_frame_may_be_split_anywhere() { + for body in [JAVA_VALUES, JAVA_AGGREGATES] { + let whole = parse_read_response(body).unwrap(); + for chunk in [1, 3, 7, 28, 1000, 65536] { + assert_eq!(assemble_in_chunks(body, chunk).unwrap(), whole, "chunks of {chunk}"); + } + } + // Cut inside the marker itself: "DH" then nothing. + let mut body = response(&[float_frame(1, "a", 0..2)]); + body.truncate(body.len() - 2); + assert!(assemble_in_chunks(&body, 1).unwrap_err().contains("cut short")); + } + + /// A local HTTP server answering with `chunks` as a chunked body, each written and flushed on + /// its own; the chunk at `gate_after` waits for a permit on `permits`. + async fn serve(chunks: Vec>, permits: tokio::sync::mpsc::Receiver<()>, gate_after: usize) -> String { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + let mut permits = permits; + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = vec![0u8; 4096]; + let _ = socket.read(&mut request).await; + socket + .write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/vnd.intellistream.datapoint-block\r\nTransfer-Encoding: chunked\r\n\r\n") + .await + .unwrap(); + for (i, chunk) in chunks.iter().enumerate() { + if i == gate_after { + permits.recv().await; + } + socket.write_all(format!("{:x}\r\n", chunk.len()).as_bytes()).await.unwrap(); + socket.write_all(chunk).await.unwrap(); + socket.write_all(b"\r\n").await.unwrap(); + socket.flush().await.unwrap(); + } + let _ = socket.write_all(b"0\r\n\r\n").await; + let _ = socket.shutdown().await; + }); + format!("http://{address}/") + } + + async fn stream_of(url: &str) -> BinaryDatapointStream { + let response = reqwest::get(url).await.unwrap(); + let (tx, items) = mpsc::channel(4); + let task = tokio::spawn(produce(response, "test".to_string(), tx)); + BinaryDatapointStream { items, task } + } + + #[tokio::test] + async fn items_arrive_while_the_body_is_still_coming() { + let first = float_frame(1, "a", 0..3); + let rest = [float_frame(1, "a", 3..5), float_frame(2, "b", 0..2), END_OF_RESPONSE.to_vec()].concat(); + let (release, permits) = tokio::sync::mpsc::channel(1); + let url = serve(vec![first, rest], permits, 1).await; + let mut stream = stream_of(&url).await; + + // The server sends nothing more until this item is in hand. + let item = tokio::time::timeout(std::time::Duration::from_secs(5), stream.next()) + .await + .expect("the first frame did not arrive before the rest of the body") + .unwrap() + .unwrap(); + assert_eq!((item.id, item.timestamps.clone()), (1, vec![0, 1, 2])); + release.send(()).await.unwrap(); + + let rest: Vec = stream.map(|i| i.unwrap()).collect().await; + assert_eq!(rest.iter().map(|s| s.id).collect::>(), [1, 2], "one item per frame"); + let merged = merge_series(std::iter::once(item).chain(rest)).unwrap(); + assert_eq!(merged.len(), 2); + assert_eq!(merged[0].timestamps, [0, 1, 2, 3, 4]); + } + + #[tokio::test] + async fn a_stream_cut_short_ends_with_an_error_after_what_arrived() { + let frames = vec![float_frame(1, "a", 0..2), float_frame(2, "b", 0..2)]; + // Every frame intact and the chunked body properly ended, but no end marker. + let (_release, permits) = tokio::sync::mpsc::channel(1); + let url = serve(frames, permits, usize::MAX).await; + let items: Vec = stream_of(&url).await.collect().await; + assert_eq!(items.len(), 3); + assert!(items[0].is_ok() && items[1].is_ok()); + let error = items[2].as_ref().unwrap_err(); + assert_eq!(error.get_status(), StatusCode::BAD_GATEWAY); + assert!(error.get_message().contains("cut short"), "{}", error.get_message()); + } + + #[tokio::test] + async fn a_bad_frame_ends_the_stream_after_the_frames_before_it() { + let batch = RecordBatch::try_new( + DatapointValueType::Float.schema(), + vec![ + Arc::new(Int64Array::from(vec![1])), + Arc::new(TimestampMillisecondArray::from(vec![0]).with_timezone("UTC")), + Arc::new(Float64Array::from(vec![1.0])), + ], + ) + .unwrap(); + let bad = frame_around(DatapointValueType::Float, CODEC_ARROW_IPC_AGGREGATES, 2, "b", &batch); + let body = response(&[float_frame(1, "a", 0..2), bad, float_frame(3, "c", 0..2)]); + let (_release, permits) = tokio::sync::mpsc::channel(1); + let url = serve(body.chunks(5).map(<[u8]>::to_vec).collect(), permits, usize::MAX).await; + let items: Vec = stream_of(&url).await.collect().await; + assert_eq!(items.len(), 2, "the good frame, then the error; nothing after it"); + assert_eq!(items[0].as_ref().unwrap().id, 1); + assert!(items[1].as_ref().unwrap_err().get_message().contains("frame 1:")); + } + /// The same series asked for twice, the second window starting before the first ended: two /// entries, since merging them would put the timestamps out of order. #[test] diff --git a/src/timeseries/mod.rs b/src/timeseries/mod.rs index f8a64dc..df2035a 100644 --- a/src/timeseries/mod.rs +++ b/src/timeseries/mod.rs @@ -4,7 +4,7 @@ pub mod datapoint_listen; mod test; pub use binary::{BinaryIngestOptions, DatapointValueType, Frame, FrameWriter, ResolvedSeries}; -pub use binary_read::{BinaryDatapoints, BinaryValues}; +pub use binary_read::{BinaryDatapointStream, BinaryDatapoints, BinaryValues}; pub use datapoint_listen::{DatapointListener, LiveDatapoint}; use crate::buffer::DurableSpool; diff --git a/src/timeseries/test.rs b/src/timeseries/test.rs index 18fba9e..bfcecff 100644 --- a/src/timeseries/test.rs +++ b/src/timeseries/test.rs @@ -779,8 +779,9 @@ mod tests { whole_window.set_external_id(&float_ext_id); whole_window.set_start(start); whole_window.set_end(start + Duration::seconds(count)); + let request = DataWrapper::from_vec(vec![whole_window, item(&text_ext_id, text_count as u64)]); let first = api_service.time_series - .retrieve_datapoints_binary(&DataWrapper::from_vec(vec![whole_window, item(&text_ext_id, text_count as u64)])) + .retrieve_datapoints_binary(&request) .await .unwrap_or_else(|e| panic!("binary read failed with {}: {}", e.get_status(), e.get_message())); assert_eq!(first.len(), 2, "one entry per series, the frames of one merged"); @@ -792,6 +793,19 @@ mod tests { let want_values: Vec = (0..count).map(value).collect(); assert_eq!(float.values, BinaryValues::Float(want_values)); + println!("The same read, streamed frame by frame..."); + let mut stream = api_service.time_series.stream_datapoints_binary(&request).await?; + let (mut frames, mut streamed) = (Vec::new(), Vec::new()); + while let Some(chunk) = futures::StreamExt::next(&mut stream).await { + let chunk = chunk?; + if chunk.id == float.id { + streamed.extend_from_slice(&chunk.timestamps); + } + frames.push((chunk.len(), chunk.external_id)); + } + assert_eq!(frames.len(), 5, "two float frames and three text frames, one item each: {frames:?}"); + assert_eq!(streamed, want_timestamps); + let texts = &first[1]; assert_eq!(texts.external_id, text_ext_id); assert_eq!(texts.timestamps, want_timestamps[..text_count as usize]);