From edc73920a5eb73f245a7d712c61a1068bd182552 Mon Sep 17 00:00:00 2001 From: jgjesdal Date: Mon, 12 Oct 2026 14:14:26 +0200 Subject: [PATCH] Read datapoints through the binary endpoint, decoded in parallel retrieve_datapoints_binary posts the JSON retrieve body to POST /timeseries/data/list/binary and parses the frames that come back into one columnar BinaryDatapoints per series. A body without the DHDE end marker was cut short and is a 502, never fewer points. Aggregate frames (codec 2) carry avg, min, max and sum; decimals stay exact. The envelopes are walked first and the frames then decompressed and decoded on scoped threads: zstd was 90% of a single-threaded parse. For 100M float rows the parse went from 1.46s to 0.60s and the SDK's share of an end-to-end read from 1.57s to 0.90s. In Python, the sync client's retrieve_datapoints_binary returns objects implementing the Arrow PyCapsule stream protocol, so polars, pyarrow and pandas 3 take them without copying. The package adds no runtime dependency. The offline tests read fixtures written by the platform's own frame writer. Co-Authored-By: Claude Opus 5.5 Signed-off-by: jgjesdal --- AGENTS.md | 6 +- README.md | 35 + datahub_python_bindings/Cargo.toml | 3 + .../intellistream_datahub_sdk/__init__.pyi | 70 ++ .../src/timeseries/binary_read.rs | 165 ++++ datahub_python_bindings/src/timeseries/mod.rs | 2 + .../src/timeseries/sync_service.rs | 27 + python_tests/requirements.txt | 2 + python_tests/test_timeseries_binary_read.py | 212 +++++ src/blocking.rs | 3 +- src/generic.rs | 51 ++ src/timeseries/binary.rs | 14 + src/timeseries/binary_read.rs | 836 ++++++++++++++++++ .../fixtures/GenerateReadFixtures.java | 115 +++ src/timeseries/fixtures/read_aggregates.bin | Bin 0 -> 401 bytes src/timeseries/fixtures/read_values.bin | Bin 0 -> 2594 bytes src/timeseries/mod.rs | 2 + src/timeseries/test.rs | 160 +++- 18 files changed, 1699 insertions(+), 4 deletions(-) create mode 100644 datahub_python_bindings/src/timeseries/binary_read.rs create mode 100644 python_tests/test_timeseries_binary_read.py create mode 100644 src/timeseries/binary_read.rs create mode 100644 src/timeseries/fixtures/GenerateReadFixtures.java create mode 100644 src/timeseries/fixtures/read_aggregates.bin create mode 100644 src/timeseries/fixtures/read_values.bin diff --git a/AGENTS.md b/AGENTS.md index 7b432db..6a153fa 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -199,9 +199,11 @@ at-least-once path is still `SubscriptionsService::listen`. Two things differ fr `recommend_value_type` sits beside it: a bare `ValueTypeRecommendation`, and an unknown unit is the generic default with `recognized: false`, not an error. -### Binary datapoint ingest (`src/timeseries/binary.rs`) +### Binary datapoints (`src/timeseries/binary.rs`, `src/timeseries/binary_read.rs`) -`TimeSeriesService::insert_datapoints_binary` is the second ingest path, to `POST /timeseries/data/binary`: the same `DatapointsCollection` 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` 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 `ApiServiceProvider` trait (`src/generic.rs`) diff --git a/README.md b/README.md index 0653c0b..a228fe0 100644 --- a/README.md +++ b/README.md @@ -139,6 +139,41 @@ and sends again, once. Resolving by external id goes through The durable spool does not cover this path. `binary::FrameWriter` is public for producers that build frames themselves. +## Binary datapoint read + +`time_series.retrieve_datapoints_binary(&filters)` takes the request `retrieve_datapoints` +takes and reads the same frames back from `POST /timeseries/data/list/binary`, so a large read +costs no JSON and no object per point. It returns one `BinaryDatapoints` per series that has +points, in the order of the request's items: `timestamps` in epoch milliseconds and `values` as +one typed column (`BinaryValues::Float(Vec)`, exact `Decimal { unscaled, scale }` for +`numeric` and `decimal32`, two optional columns for `mixed`, or `Aggregates { avg, min, max, +sum }` for an item with `aggregates` and a `granularity`). + +```rust +let mut filter = RetrieveFilter::default(); +filter.external_id = Some("engine.temp".to_string()); +filter.start = Some(Utc.with_ymd_and_hms(2026, 10, 1, 0, 0, 0).unwrap()); +filter.end = Some(Utc.with_ymd_and_hms(2026, 10, 2, 0, 0, 0).unwrap()); +for series in api.time_series.retrieve_datapoints_binary(&DataWrapper::from_vec(vec![filter])).await? { + if let BinaryValues::Float(values) = &series.values { + println!("{}: {} points", series.external_id, values.len()); + } +} +``` + +An item with a `limit` (at most 100 000) returns that many of the oldest points from `start`; +an item without one returns its whole window, so bound a long series with `end`, as the whole +answer is held in memory. There is no cursor: to read on in pieces, send the request again with +`start` one millisecond past the last timestamp you got. A series that does not exist is a 400 for the whole request, where +the JSON read answers empty. A response the server could not finish is a 502 rather than fewer +points, and is safe to send again. + +In Python, `client.timeseries.retrieve_datapoints_binary(filter_or_list)` on the sync client +returns objects that implement the Arrow PyCapsule stream protocol, so +`polars.DataFrame(series)`, `pyarrow.table(series)` and `pandas.DataFrame.from_arrow(series)` +take them without copying; `series.to_polars()` and `series.to_pandas()` do the same. The package +adds no dependency for this — bring polars, or pyarrow for pandas. + ## Python bindings `datahub_python_bindings/` wraps this SDK as the Python package `intellistream-datahub-sdk` (import name diff --git a/datahub_python_bindings/Cargo.toml b/datahub_python_bindings/Cargo.toml index 96212f4..cae7243 100644 --- a/datahub_python_bindings/Cargo.toml +++ b/datahub_python_bindings/Cargo.toml @@ -25,3 +25,6 @@ pyo3-async-runtimes = { version = "0.29", features = ["tokio-runtime"] } numpy = "0.29" uuid = { version = "1.22.0", features = ["v4"] } strum = { version = "0.28", features = ["derive"] } +# `ffi` for the Arrow PyCapsule stream; the SDK already depends on both crates. +arrow-array = { version = "59.3", features = ["ffi"] } +arrow-schema = "59.3" diff --git a/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi b/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi index 7a233fa..0359bba 100644 --- a/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi +++ b/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi @@ -590,6 +590,43 @@ class DatapointsCollectionDatapoints: def id(self) -> int | None: ... +class BinaryDatapoints: + """One series' rows from ``TimeSeriesServiceSync.retrieve_datapoints_binary``, as an Arrow table. + + Implements the Arrow PyCapsule stream protocol, so ``polars.DataFrame(series)``, + ``pyarrow.table(series)`` and ``pandas.DataFrame.from_arrow(series)`` (pandas 3) take it + without copying. Every call makes a new stream, so one object converts more than once. + + Columns are ``timestamp`` (``timestamp[ms, UTC]``; the bucket start for aggregates) and then, + by value type: ``value`` as int64 (``bigint``), float64 (``float``), float32 (``float32``), + ``decimal128(18, 6)`` (``numeric``), ``decimal128(9, 4)`` (``decimal32``) or utf8 (``text``); + nullable ``value_numeric`` and ``value_text`` for ``mixed``, each row setting one; or + ``avg``, ``min``, ``max`` and ``sum`` as float64 when :attr:`is_aggregates`. + """ + + @property + def id(self) -> int: ... + @property + def external_id(self) -> str: ... + @property + def value_type(self) -> str: + """The series' value type: ``bigint``, ``float``, ``float32``, ``numeric``, ``decimal32``, + ``text`` or ``mixed``. An aggregate read reports the series' own type.""" + @property + def is_aggregates(self) -> bool: + """Whether the rows are buckets with ``avg``, ``min``, ``max`` and ``sum``.""" + @property + def columns(self) -> list[str]: ... + def __len__(self) -> int: ... + def __arrow_c_stream__(self, requested_schema: object | None = None) -> object: + """The Arrow PyCapsule stream protocol. ``requested_schema`` is ignored.""" + def to_polars(self) -> Any: + """A ``polars.DataFrame``. Needs polars; raises ``ImportError`` without it.""" + def to_pandas(self) -> Any: + """A ``pandas.DataFrame``, converted through pyarrow. Needs pandas and pyarrow; raises + ``ImportError`` without pyarrow.""" + + class RetrieveFilter: def __init__( self, @@ -748,6 +785,39 @@ class TimeSeriesServiceSync: ts: Identifiable, ) -> list[str]: ... def retrieve_datapoints(self, input: RetrieveFilter) -> list[DatapointsCollectionDatapoints]: ... + def retrieve_datapoints_binary( + self, input: RetrieveFilter | list[RetrieveFilter] + ) -> list[BinaryDatapoints]: + """Read datapoints through ``POST /timeseries/data/list/binary``: the request + ``retrieve_datapoints`` takes, answered as compressed Arrow frames instead of JSON. + + Parameters + ---------- + input + One filter or a list. Each needs a ``start`` or an ``end``. With a ``limit`` (at most + 100 000) it returns that many of the oldest points from there; without one it returns + every point in its window, so bound a long series with ``end``, as the whole answer is + held in memory. With ``aggregates`` and a ``granularity`` + the series comes back as buckets carrying all four aggregates, whichever were named. + + Returns + ------- + list[BinaryDatapoints] + One per series that has points in its window, in the order of ``input``; a series + with none is left out rather than returned empty. + + Raises + ------ + DataHubException + 400 for a ``cursor`` (there is no paging: send the request again with ``start`` one + millisecond past the last timestamp read), for aggregates on a text series, and for + a series that does not exist or that you cannot read, where the JSON read answers + empty. 502 when the response was cut short; sending it again is safe. + + Notes + ----- + Sync client only, like the binary insert. + """ def delete_datapoints(self, input: list[DeleteFilter]) -> None: ... def retrieve_latest_datapoints( self, input: list[Identifiable] diff --git a/datahub_python_bindings/src/timeseries/binary_read.rs b/datahub_python_bindings/src/timeseries/binary_read.rs new file mode 100644 index 0000000..8a044f6 --- /dev/null +++ b/datahub_python_bindings/src/timeseries/binary_read.rs @@ -0,0 +1,165 @@ +use std::sync::Arc; + +use arrow_array::ffi_stream::FFI_ArrowArrayStream; +use arrow_array::{ + ArrayRef, Decimal128Array, Float32Array, Float64Array, Int64Array, RecordBatch, + RecordBatchIterator, StringArray, TimestampMillisecondArray, +}; +use arrow_schema::{ArrowError, DataType, Field, Schema, TimeUnit}; +use intellistream_datahub_sdk::timeseries::{BinaryDatapoints, BinaryValues, DatapointValueType}; +use pyo3::exceptions::{PyImportError, PyValueError}; +use pyo3::prelude::*; +use pyo3::types::PyCapsule; + +/// 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. +#[pyclass(module = "intellistream_datahub_sdk", name = "BinaryDatapoints", frozen)] +pub struct PyBinaryDatapoints { + id: u64, + external_id: String, + value_type: DatapointValueType, + aggregates: bool, + batch: RecordBatch, +} + +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 batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?; + Ok(Self { + id: series.id, + external_id: series.external_id, + value_type: series.value_type, + aggregates, + batch, + }) + } +} + +#[pymethods] +impl PyBinaryDatapoints { + #[getter] + fn id(&self) -> u64 { + self.id + } + + #[getter] + fn external_id(&self) -> &str { + &self.external_id + } + + /// The series' value type, as a timeseries read reports it: `float`, `bigint`, `text`, ... + #[getter] + fn value_type(&self) -> String { + format!("{:?}", self.value_type).to_lowercase() + } + + /// Whether the rows are buckets with `avg`, `min`, `max` and `sum` rather than values. + #[getter] + fn is_aggregates(&self) -> bool { + self.aggregates + } + + /// The column names, in order. + #[getter] + fn columns(&self) -> Vec { + self.batch.schema().fields().iter().map(|f| f.name().clone()).collect() + } + + fn __len__(&self) -> usize { + self.batch.num_rows() + } + + fn __repr__(&self) -> String { + format!( + "BinaryDatapoints(external_id={:?}, value_type={:?}, rows={}{})", + self.external_id, + self.value_type(), + self.batch.num_rows(), + if self.aggregates { ", aggregates" } else { "" } + ) + } + + /// The Arrow PyCapsule stream protocol. Each call is a new stream over the same buffers, so + /// the object can be converted more than once. `requested_schema` is ignored, as the protocol + /// allows: the columns come in their own types. + #[pyo3(signature = (requested_schema=None))] + fn __arrow_c_stream__<'py>( + &self, + py: Python<'py>, + requested_schema: Option>, + ) -> PyResult> { + let _ = requested_schema; + let reader = RecordBatchIterator::new(vec![Ok(self.batch.clone())], self.batch.schema()); + let stream = FFI_ArrowArrayStream::new(Box::new(reader)); + PyCapsule::new_with_value(py, stream, c"arrow_array_stream") + } + + /// A `polars.DataFrame`, without copying. Needs polars. + fn to_polars<'py>(slf: &Bound<'py, Self>) -> PyResult> { + let polars = import(slf.py(), "polars", "to_polars")?; + polars.call_method1("DataFrame", (slf,)) + } + + /// A `pandas.DataFrame`. Needs pandas and pyarrow, which pandas converts Arrow through. + fn to_pandas<'py>(slf: &Bound<'py, Self>) -> PyResult> { + let py = slf.py(); + let pyarrow = import(py, "pyarrow", "to_pandas")?; + pyarrow.call_method1("table", (slf,))?.call_method0("to_pandas") + } +} + +fn import<'py>(py: Python<'py>, module: &str, method: &str) -> PyResult> { + py.import(module).map_err(|_| { + PyImportError::new_err(format!( + "BinaryDatapoints.{method}() needs {module}; install it with `pip install {module}`" + )) + }) +} + +pub(crate) fn to_py(series: Vec) -> 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 5cd6d94..53422a7 100644 --- a/datahub_python_bindings/src/timeseries/mod.rs +++ b/datahub_python_bindings/src/timeseries/mod.rs @@ -33,6 +33,7 @@ use std::sync::Arc; use strum::{Display, EnumString}; pub mod async_service; +pub mod binary_read; mod construction; pub mod datapoints; pub mod general; @@ -469,6 +470,7 @@ 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::()?; diff --git a/datahub_python_bindings/src/timeseries/sync_service.rs b/datahub_python_bindings/src/timeseries/sync_service.rs index 0d84537..4303ea2 100644 --- a/datahub_python_bindings/src/timeseries/sync_service.rs +++ b/datahub_python_bindings/src/timeseries/sync_service.rs @@ -4,11 +4,18 @@ use crate::timeseries::datapoints::{ }; use crate::{DatahubIdentity, Identifiable}; use crate::{PyIdCollection, PyRetrieveFilter}; +use crate::timeseries::binary_read::{self, PyBinaryDatapoints}; use intellistream_datahub_sdk::generic::{DataWrapper, IdAndExtId}; use intellistream_datahub_sdk::{ApiService, TimeSeriesUpdateCollection}; use pyo3_async_runtimes::tokio::future_into_py; use std::sync::Arc; +#[derive(FromPyObject)] +enum RetrieveFilters { + One(PyRetrieveFilter), + Many(Vec), +} + #[pyclass(module = "intellistream_datahub_sdk", name = "TimeSeriesServiceSync")] pub struct PyTimeSeriesServiceSync { pub api_service: Arc, @@ -355,6 +362,26 @@ impl PyTimeSeriesServiceSync { Ok(result) }) } + /// `POST /timeseries/data/list/binary`: the request `retrieve_datapoints` takes, one filter or + /// a list, answered as Arrow frames. One `BinaryDatapoints` per series that has points. + fn retrieve_datapoints_binary<'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 series = py.detach(|| { + self.runtime + .block_on(service.time_series.retrieve_datapoints_binary(&wrapper)) + .map_err(crate::datahub_err) + })?; + binary_read::to_py(series) + } fn delete_datapoints<'py>(&self, py: Python<'py>, input: Vec) -> PyResult<()> { let service = self.api_service.clone(); let wrapper = diff --git a/python_tests/requirements.txt b/python_tests/requirements.txt index c73f18a..5d91ece 100644 --- a/python_tests/requirements.txt +++ b/python_tests/requirements.txt @@ -14,3 +14,5 @@ pytest-asyncio numpy pandas polars +# pandas converts Arrow through pyarrow (BinaryDatapoints.to_pandas). +pyarrow diff --git a/python_tests/test_timeseries_binary_read.py b/python_tests/test_timeseries_binary_read.py new file mode 100644 index 0000000..8e9c03d --- /dev/null +++ b/python_tests/test_timeseries_binary_read.py @@ -0,0 +1,212 @@ +"""Binary datapoint read (`POST /timeseries/data/list/binary`) through the Python bindings. + +Two halves. The offline one answers the client from a stand-in server with response bodies the +platform's own Java frame writer produced (`src/timeseries/fixtures/`), so it checks the binding +and the Arrow PyCapsule stream without a backend. The live one round-trips through a backend and +skips when that backend does not serve the endpoint yet. +""" +import datetime +import decimal +import http.server +import pathlib +import threading + +import pytest + +import intellistream_datahub_sdk +from intellistream_datahub_sdk import DataHubException +from polling import poll_until +from python_tests.fixtures import * # noqa: F401,F403 + +FIXTURES = pathlib.Path(__file__).resolve().parent.parent / "src" / "timeseries" / "fixtures" +T0 = datetime.datetime.fromtimestamp(1_700_000_000, tz=datetime.timezone.utc) +MEDIA_TYPE = "application/vnd.intellistream.datapoint-block" + + +class _StandIn(http.server.BaseHTTPRequestHandler): + status = 200 + body = b"" + content_type = MEDIA_TYPE + seen = [] + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + type(self).seen.append((self.path, self.headers.get("Accept"), self.rfile.read(length))) + self.send_response(type(self).status) + self.send_header("Content-Type", type(self).content_type) + self.send_header("Content-Length", str(len(type(self).body))) + self.end_headers() + self.wfile.write(type(self).body) + + def log_message(self, *args): + pass + + +@pytest.fixture +def stand_in(): + _StandIn.status, _StandIn.content_type, _StandIn.seen = 200, MEDIA_TYPE, [] + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _StandIn) + threading.Thread(target=server.serve_forever, daemon=True).start() + client = intellistream_datahub_sdk.DataHubClient( + base_url=f"http://127.0.0.1:{server.server_address[1]}", token="test-token" + ) + yield client, _StandIn + server.shutdown() + + +def _filter(ts="rd_float"): + return intellistream_datahub_sdk.RetrieveFilter(ts=ts, start=T0, limit=1000) + + +def test_every_value_type_converts_through_the_arrow_stream(stand_in): + pl = pytest.importorskip("polars") + client, server = stand_in + server.body = (FIXTURES / "read_values.bin").read_bytes() + + series = client.timeseries.retrieve_datapoints_binary([_filter("rd_bigint"), _filter("rd_float")]) + + path, accept, _ = server.seen[0] + assert path == "/timeseries/data/list/binary" + assert accept.startswith(MEDIA_TYPE) and "application/problem+json" in accept + assert [s.external_id for s in series] == [ + "rd_bigint", "rd_float", "rd_float32", "rd_numeric", "rd_decimal32", "rd_text", "rd_mixed", "rd_mixed_numbers", + ] + by_name = {s.external_id: s for s in series} + + bigint = by_name["rd_bigint"] + assert (bigint.id, bigint.value_type, len(bigint), bigint.is_aggregates) == (11, "bigint", 3, False) + assert bigint.columns == ["timestamp", "value"] + frame = pl.DataFrame(bigint) + assert frame["value"].to_list() == [1, 42, -9_000_000_000] + assert frame["timestamp"].dtype == pl.Datetime("ms", "UTC") + assert frame["timestamp"][0] == T0 + # A stream is made per call, so one object converts more than once. + assert bigint.to_polars().equals(frame) + + assert pl.DataFrame(by_name["rd_float"])["value"].to_list() == [1.5, -2.25] + assert pl.DataFrame(by_name["rd_float32"])["value"].dtype == pl.Float32 + numeric = pl.DataFrame(by_name["rd_numeric"])["value"] + assert numeric.dtype == pl.Decimal(18, 6) + assert numeric.to_list() == [decimal.Decimal("123456789012.345678"), decimal.Decimal("-0.000001")] + assert pl.DataFrame(by_name["rd_decimal32"])["value"].dtype == pl.Decimal(9, 4) + assert pl.DataFrame(by_name["rd_text"])["value"].to_list() == ["open", "blåbær"] + + mixed = by_name["rd_mixed"] + assert mixed.columns == ["timestamp", "value_numeric", "value_text"] + mixed_frame = pl.DataFrame(mixed) + assert mixed_frame["value_numeric"].to_list() == [3.5, None, -7.0] + assert mixed_frame["value_text"].to_list() == [None, "fault", None] + + +def test_aggregates_convert_to_pandas(stand_in): + pytest.importorskip("pyarrow") + pytest.importorskip("pandas") + client, server = stand_in + server.body = (FIXTURES / "read_aggregates.bin").read_bytes() + + (series,) = client.timeseries.retrieve_datapoints_binary(_filter("rd_aggregates")) + + assert series.is_aggregates and series.value_type == "float" + assert series.columns == ["timestamp", "avg", "min", "max", "sum"] + frame = series.to_pandas() + assert frame["avg"].tolist() == [1.5, 2.5, 3.5] + assert frame["sum"].tolist() == [6.0, 10.0, 14.0] + assert frame["timestamp"].iloc[1] == T0 + datetime.timedelta(hours=1) + + +def test_no_points_is_no_series(stand_in): + client, server = stand_in + server.body = b"DHDE" + assert client.timeseries.retrieve_datapoints_binary(_filter()) == [] + + +def test_a_response_cut_short_raises_rather_than_returning_fewer_points(stand_in): + client, server = stand_in + full = (FIXTURES / "read_values.bin").read_bytes() + server.body = full[:-4] # every frame intact, only the end marker missing + with pytest.raises(DataHubException, match="cut short") as excinfo: + client.timeseries.retrieve_datapoints_binary(_filter()) + assert excinfo.value.status_code == 502 + + +def test_a_refusal_is_the_problem_the_server_sent(stand_in): + client, server = stand_in + server.status, server.content_type = 400, "application/problem+json" + server.body = ( + b'{"type":"https://intellistream.ai/errors/bad-request","title":"Bad Request","status":400,' + b'"detail":"The binary read has no cursors; page by moving `start` past the last timestamp read."}' + ) + with pytest.raises(DataHubException, match="no cursors") as excinfo: + client.timeseries.retrieve_datapoints_binary(_filter()) + assert excinfo.value.status_code == 400 + + +# --- live --------------------------------------------------------------------------------------- + +START = datetime.datetime(2024, 3, 1, tzinfo=datetime.timezone.utc) +TIMESTAMPS = [START + datetime.timedelta(minutes=i) for i in range(120)] + + +@pytest.fixture +def binary_read(sync_client, make_ts): + """Skips unless the backend serves the binary read: it is newer than the binary insert.""" + probe = intellistream_datahub_sdk.RetrieveFilter(ts=make_ts(value_type="float"), start=START, limit=1) + try: + sync_client.timeseries.retrieve_datapoints_binary(probe) + except DataHubException as e: + if e.status_code in (404, 405): + pytest.skip("the backend does not serve POST /timeseries/data/list/binary") + raise + return sync_client.timeseries.retrieve_datapoints_binary + + +def _wait_for(client, ts, expected): + rf = intellistream_datahub_sdk.RetrieveFilter(ts=ts, start=START, limit=10_000) + poll_until( + lambda: client.timeseries.retrieve_datapoints(rf), + lambda collections: bool(collections) and len(collections[0].get_datapoints()) >= expected, + ) + + +def test_binary_read_matches_the_json_read(sync_client, make_ts, binary_read): + pl = pytest.importorskip("polars") + ts = make_ts(value_type="float") + values = [i * 0.5 - 30.0 for i in range(len(TIMESTAMPS))] + sync_client.timeseries.insert_from_lists_binary(TIMESTAMPS, values, ts) + _wait_for(sync_client, ts, len(TIMESTAMPS)) + + (series,) = binary_read(intellistream_datahub_sdk.RetrieveFilter(ts=ts, start=START, limit=10_000)) + + assert series.external_id == ts.external_id and series.value_type == "float" + frame = series.to_polars() + assert frame["timestamp"].to_list() == TIMESTAMPS + assert frame["value"].to_list() == values + + +def test_binary_read_of_hourly_aggregates(sync_client, make_ts, binary_read): + pytest.importorskip("pyarrow") + ts = make_ts(value_type="float") + sync_client.timeseries.insert_from_lists_binary(TIMESTAMPS, [float(i) for i in range(len(TIMESTAMPS))], ts) + _wait_for(sync_client, ts, len(TIMESTAMPS)) + + rf = intellistream_datahub_sdk.RetrieveFilter( + ts=ts, start=START, end=TIMESTAMPS[-1] + datetime.timedelta(minutes=1), + aggregates=["avg", "min", "max", "sum"], granularity="1h", + ) + (series,) = binary_read(rf) + json_buckets = sync_client.timeseries.retrieve_datapoints(rf)[0].get_datapoints() + + frame = series.to_pandas() + assert series.is_aggregates and len(frame) == len(json_buckets) == 2 + # Zero-order hold with carry-in, so the second hour's min is the 59 carried into it. + assert frame["min"].tolist() == [dp.min for dp in json_buckets] + assert frame["max"].tolist() == [dp.max for dp in json_buckets] + assert frame["sum"].tolist() == pytest.approx([dp.sum for dp in json_buckets]) + assert frame["avg"].tolist() == pytest.approx([dp.average for dp in json_buckets]) + + +def test_binary_read_of_an_unknown_series_is_a_bad_request(make_ts, binary_read): + missing = unique_id("ts_missing") + with pytest.raises(DataHubException, match=missing) as excinfo: + binary_read(intellistream_datahub_sdk.RetrieveFilter(ts=missing, start=START)) + assert excinfo.value.status_code == 400 diff --git a/src/blocking.rs b/src/blocking.rs index 1dd827a..057baf7 100644 --- a/src/blocking.rs +++ b/src/blocking.rs @@ -49,7 +49,7 @@ use crate::resources::{ ResourceNetwork, ResourceUpdate, }; use crate::timeseries::{ - BinaryIngestOptions, TimeSeries, TimeSeriesFilter, TimeSeriesUpdateCollection, + BinaryDatapoints, BinaryIngestOptions, TimeSeries, TimeSeriesFilter, TimeSeriesUpdateCollection, ValueTypeRecommendation, }; use crate::unit::Unit; @@ -182,6 +182,7 @@ impl TimeSeriesService { fn insert_datapoints(json: &mut DataWrapper>) -> Result, ResponseError>; fn insert_datapoints_binary(json: &DataWrapper>, options: &BinaryIngestOptions) -> Result, ResponseError>; fn retrieve_datapoints(json: &DataWrapper) -> Result>, ResponseError>; + fn retrieve_datapoints_binary(json: &DataWrapper) -> Result, ResponseError>; fn delete_datapoints(json: &DataWrapper) -> Result, ResponseError>; fn retrieve_latest_datapoint(json: &DataWrapper) -> Result>, ResponseError>; } diff --git a/src/generic.rs b/src/generic.rs index 7faad32..0220427 100644 --- a/src/generic.rs +++ b/src/generic.rs @@ -882,6 +882,57 @@ 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. + /// + /// `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( + &self, + path: &str, + json: &J, + accept: &str, + ) -> Result, ResponseError> { + let token = self.get_token().await?; + let response = self + .get_api_service() + .http_client + .post(path) + .json(json) + .header(http::header::ACCEPT, format!("{accept}, application/problem+json")) + .bearer_auth(token.clone()) + .send() + .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, + }), + } + } + /// `GET` an endpoint that answers with bytes rather than JSON (currently only /// `/files/download/{id}`). /// diff --git a/src/timeseries/binary.rs b/src/timeseries/binary.rs index 45cbfac..f156c22 100644 --- a/src/timeseries/binary.rs +++ b/src/timeseries/binary.rs @@ -86,6 +86,20 @@ impl DatapointValueType { self as u8 } + /// The value type a frame envelope numbers `id`. + pub fn from_id(id: u8) -> Option { + Some(match id { + 1 => Self::Bigint, + 2 => Self::Float, + 3 => Self::Numeric, + 4 => Self::Text, + 5 => Self::Decimal32, + 6 => Self::Mixed, + 7 => Self::Float32, + _ => return None, + }) + } + pub fn carries_text(self) -> bool { matches!(self, Self::Text | Self::Mixed) } diff --git a/src/timeseries/binary_read.rs b/src/timeseries/binary_read.rs new file mode 100644 index 0000000..c57a76a --- /dev/null +++ b/src/timeseries/binary_read.rs @@ -0,0 +1,836 @@ +//! The binary datapoint read behind `POST /timeseries/data/list/binary`. +//! +//! The request is the JSON retrieve body; the answer is the frames `POST /timeseries/data/binary` +//! takes, one series per frame, concatenated, and ended by the 4-byte marker `DHDE`. A body without +//! 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. + +use std::io::Cursor; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +use arrow_array::{ + Array, Decimal128Array, Float32Array, Float64Array, Int64Array, RecordBatch, StringArray, + TimestampMillisecondArray, +}; +use arrow_ipc::reader::StreamReader; +use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; +use oauth2::http::StatusCode; + +use crate::generic::{ApiServiceProvider, DataWrapper, RetrieveFilter}; +use crate::http::ResponseError; +use crate::timeseries::binary::{ + DatapointValueType, CODEC_ARROW_IPC, COMPRESSION_ZSTD, HEADER_BYTES, MAGIC, MAX_FRAME_RAW_BYTES, + MEDIA_TYPE, VERSION, +}; +use crate::timeseries::TimeSeriesService; + +/// Ends a binary read's response, after the last frame. +pub const END_OF_RESPONSE: &[u8; 4] = b"DHDE"; +/// A response frame whose payload holds bucketed aggregates in [`aggregates_schema`]. +pub const CODEC_ARROW_IPC_AGGREGATES: u8 = 2; +/// Buckets per aggregate frame, whatever the series' type. +pub const MAX_AGGREGATE_ROWS: usize = 50_000; + +/// The schema of an aggregate frame: the bucket start and four Float64s, for every value type. +pub fn aggregates_schema() -> SchemaRef { + let float = |name: &str| Field::new(name, DataType::Float64, false); + Arc::new(Schema::new(vec![ + Field::new("timeseries_id", DataType::Int64, false), + Field::new( + "timestamp", + DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())), + false, + ), + float("avg"), + float("min"), + float("max"), + float("sum"), + ])) +} + +/// One series' rows from a binary read, a column per field rather than an object per point. +#[derive(Debug, Clone, PartialEq)] +pub struct BinaryDatapoints { + pub id: u64, + pub external_id: String, + pub value_type: DatapointValueType, + /// Epoch milliseconds, oldest first; for aggregates, the start of each bucket. + pub timestamps: Vec, + pub values: BinaryValues, +} + +impl BinaryDatapoints { + pub fn len(&self) -> usize { + self.timestamps.len() + } + + pub fn is_empty(&self) -> bool { + self.timestamps.is_empty() + } + + pub fn is_aggregates(&self) -> bool { + matches!(self.values, BinaryValues::Aggregates { .. }) + } +} + +/// The value column of a [`BinaryDatapoints`], typed by the series' value type. +#[derive(Debug, Clone, PartialEq)] +#[non_exhaustive] +pub enum BinaryValues { + Bigint(Vec), + Float(Vec), + Float32(Vec), + /// `numeric` and `decimal32`, exact: each value is `unscaled / 10^scale`. + Decimal { unscaled: Vec, scale: i8 }, + Text(Vec), + /// Each row sets exactly one side. + Mixed { + numeric: Vec>, + text: Vec>, + }, + /// Time-weighted `avg`, `min`, `max` and `sum` per bucket. All four come back whichever the + /// request named. + Aggregates { + avg: Vec, + min: Vec, + max: Vec, + sum: Vec, + }, +} + +impl BinaryValues { + fn reserve_exact(&mut self, additional: usize) { + match self { + Self::Bigint(v) => v.reserve_exact(additional), + Self::Float(v) => v.reserve_exact(additional), + Self::Float32(v) => v.reserve_exact(additional), + Self::Decimal { unscaled, .. } => unscaled.reserve_exact(additional), + Self::Text(v) => v.reserve_exact(additional), + Self::Mixed { numeric, text } => { + numeric.reserve_exact(additional); + text.reserve_exact(additional); + } + Self::Aggregates { avg, min, max, sum } => { + for v in [avg, min, max, sum] { + v.reserve_exact(additional); + } + } + } + } + + /// Appends `other`, which must be the same variant; false if it is not. + fn append(&mut self, other: BinaryValues) -> bool { + match (self, other) { + (Self::Bigint(a), Self::Bigint(b)) => a.extend(b), + (Self::Float(a), Self::Float(b)) => a.extend(b), + (Self::Float32(a), Self::Float32(b)) => a.extend(b), + ( + Self::Decimal { unscaled: a, scale: sa }, + Self::Decimal { unscaled: b, scale: sb }, + ) if *sa == sb => a.extend(b), + (Self::Text(a), Self::Text(b)) => a.extend(b), + ( + Self::Mixed { numeric: an, text: at }, + Self::Mixed { numeric: bn, text: bt }, + ) => { + an.extend(bn); + at.extend(bt); + } + ( + Self::Aggregates { avg: aa, min: ai, max: ax, sum: asum }, + Self::Aggregates { avg: ba, min: bi, max: bx, sum: bsum }, + ) => { + aa.extend(ba); + ai.extend(bi); + ax.extend(bx); + asum.extend(bsum); + } + _ => return false, + } + true + } +} + +fn u32_at(body: &[u8], offset: usize) -> usize { + u32::from_le_bytes(body[offset..offset + 4].try_into().unwrap()) as usize +} + +fn read_varint(body: &[u8], pos: &mut usize) -> Option { + let mut value = 0usize; + for shift in (0..35).step_by(7) { + let byte = *body.get(*pos)?; + *pos += 1; + value |= ((byte & 0x7F) as usize) << shift; + if byte & 0x80 == 0 { + return Some(value); + } + } + None +} + +/// Parses a binary read's whole response body: one entry per series in the order the frames came, +/// which is the order of the request's items, with consecutive frames of one series merged. +/// +/// The envelopes are walked first, which is cheap and finds a body that was cut short; the frames +/// are then decompressed and decoded on several threads, since zstd is most of the cost. +/// +/// 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 decoded = decode_frames(&frames)?; + // 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)]); + } + } + } + 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(); + let mut parts = parts.into_iter(); + 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 { + 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 + )); + } + } + Ok(series) +} + +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> { + index: usize, + value_type: DatapointValueType, + aggregates: bool, + rows: usize, + names: Vec<(i64, String)>, + payload: &'a [u8], + raw_len: usize, +} + +/// 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() + )); + } + return Ok(frames); + } + if rest.len() < HEADER_BYTES { + return Err(cut_short(frames.len())); + } + let (consumed, frame) = scan_frame(rest, frames.len())?; + frames.push(frame); + pos += consumed; + } +} + +/// 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> { + let fail = |message: String| format!("frame {index}: {message}"); + if &body[0..4] != MAGIC { + return Err(fail("does not start with DHDP".to_string())); + } + 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] { + 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); + let max_rows = if aggregates { + MAX_AGGREGATE_ROWS + } else { + value_type.max_rows() + }; + if rows < 1 || rows > max_rows { + return Err(fail(format!("{rows} rows, allowed 1 to {max_rows}"))); + } + 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)) +} + +/// 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> { + let threads = std::thread::available_parallelism() + .map_or(1, |n| n.get()) + .min(frames.len()); + if threads <= 1 { + 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))); + } + }) + }) + .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> { + let fail = |message: String| format!("frame {}: {message}", frame.index); + let raw = zstd::bulk::decompress(frame.payload, frame.raw_len) + .map_err(|e| fail(format!("zstd: {e}")))?; + if raw.len() != frame.raw_len { + return Err(fail(format!( + "payload decompressed to {} bytes, the envelope says {}", + raw.len(), + frame.raw_len + ))); + } + let reader = + StreamReader::try_new(Cursor::new(raw), None).map_err(|e| fail(format!("arrow: {e}")))?; + let expected = if frame.aggregates { + aggregates_schema() + } else { + frame.value_type.schema() + }; + let schema = reader.schema(); + let same_columns = schema.fields().len() == expected.fields().len() + && schema + .fields() + .iter() + .zip(expected.fields()) + .all(|(got, want)| got.name() == want.name() && got.data_type() == want.data_type()); + if !same_columns { + return Err(fail(format!("schema {schema:?} is not the canon {expected:?}"))); + } + + let mut parts: Vec = Vec::new(); + let mut read = 0; + for batch in reader { + let batch = batch.map_err(|e| fail(format!("arrow: {e}")))?; + read += batch.num_rows(); + split_batch(&batch, frame.value_type, frame.aggregates, &frame.names, &mut parts) + .map_err(fail)?; + } + if read != frame.rows { + return Err(fail(format!( + "{read} rows in the payload, the envelope says {}", + frame.rows + ))); + } + Ok(parts) +} + +/// Appends a batch's rows to `parts`, one part per run of a series id. +fn split_batch( + batch: &RecordBatch, + value_type: DatapointValueType, + aggregates: bool, + names: &[(i64, String)], + parts: &mut Vec, +) -> Result<(), String> { + let ids = column::(batch, 0)?; + let timestamps = column::(batch, 1)?; + let values = batch_values(batch, value_type, aggregates)?; + let mut start = 0; + while start < batch.num_rows() { + let id = ids.value(start); + let mut end = start + 1; + while end < batch.num_rows() && ids.value(end) == id { + end += 1; + } + let external_id = names + .iter() + .find(|(named, _)| *named == id) + .map(|(_, name)| name.clone()) + .ok_or_else(|| format!("series {id} is not in the directory"))?; + let part = BinaryDatapoints { + id: id as u64, + external_id, + value_type, + timestamps: timestamps.values()[start..end].to_vec(), + values: slice_values(&values, start, end), + }; + match parts.last_mut() { + Some(last) if last.id == part.id => { + last.timestamps.extend(part.timestamps); + last.values.append(part.values); + } + _ => parts.push(part), + } + start = end; + } + Ok(()) +} + +fn column(batch: &RecordBatch, index: usize) -> Result<&T, String> { + batch + .column(index) + .as_any() + .downcast_ref::() + .ok_or_else(|| format!("column {index} is not of its canonical type")) +} + +fn batch_values( + batch: &RecordBatch, + value_type: DatapointValueType, + aggregates: bool, +) -> Result { + let floats = |index| column::(batch, index).map(|a| a.values().to_vec()); + if aggregates { + return Ok(BinaryValues::Aggregates { + avg: floats(2)?, + min: floats(3)?, + max: floats(4)?, + sum: floats(5)?, + }); + } + Ok(match value_type { + DatapointValueType::Bigint => { + BinaryValues::Bigint(column::(batch, 2)?.values().to_vec()) + } + DatapointValueType::Float => BinaryValues::Float(floats(2)?), + DatapointValueType::Float32 => { + BinaryValues::Float32(column::(batch, 2)?.values().to_vec()) + } + DatapointValueType::Numeric | DatapointValueType::Decimal32 => { + let array = column::(batch, 2)?; + BinaryValues::Decimal { + unscaled: array.values().to_vec(), + scale: array.scale(), + } + } + DatapointValueType::Text => { + let array = column::(batch, 2)?; + BinaryValues::Text(array.iter().map(|v| v.unwrap_or_default().to_string()).collect()) + } + DatapointValueType::Mixed => { + let numeric = column::(batch, 2)?; + let text = column::(batch, 3)?; + BinaryValues::Mixed { + numeric: numeric.iter().collect(), + text: text.iter().map(|v| v.map(str::to_string)).collect(), + } + } + }) +} + +fn slice_values(values: &BinaryValues, start: usize, end: usize) -> BinaryValues { + match values { + BinaryValues::Bigint(v) => BinaryValues::Bigint(v[start..end].to_vec()), + BinaryValues::Float(v) => BinaryValues::Float(v[start..end].to_vec()), + BinaryValues::Float32(v) => BinaryValues::Float32(v[start..end].to_vec()), + BinaryValues::Decimal { unscaled, scale } => BinaryValues::Decimal { + unscaled: unscaled[start..end].to_vec(), + scale: *scale, + }, + BinaryValues::Text(v) => BinaryValues::Text(v[start..end].to_vec()), + BinaryValues::Mixed { numeric, text } => BinaryValues::Mixed { + numeric: numeric[start..end].to_vec(), + text: text[start..end].to_vec(), + }, + BinaryValues::Aggregates { avg, min, max, sum } => BinaryValues::Aggregates { + avg: avg[start..end].to_vec(), + min: min[start..end].to_vec(), + max: max[start..end].to_vec(), + sum: sum[start..end].to_vec(), + }, + } +} + +impl TimeSeriesService { + /// Reads datapoints through `POST /timeseries/data/list/binary`, the high-throughput twin of + /// [`retrieve_datapoints`](Self::retrieve_datapoints): the same request, answered as + /// zstd-compressed Arrow frames instead of JSON, and returned as one columnar + /// [`BinaryDatapoints`] per series, in the order of the request's items. + /// + /// Each item needs a `start` or an `end`. With a `limit` (at most 100 000) it returns that many + /// of the oldest points from there; without one it returns **every** point in its window, + /// however many, so bound a long series with `end`: the whole answer is held in memory. A + /// series with no points in its window is left out of the answer rather than returned empty. + /// There is no cursor (sending one is a 400): to read on, send the request again with `start` + /// one millisecond past the last timestamp received. An item with `aggregates` and a + /// `granularity` comes back as [`BinaryValues::Aggregates`], carrying all four whichever were + /// named; text series have none. + /// + /// Unlike the JSON read, a series that does not exist or sits in a dataset the caller cannot + /// 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. + 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)) + .await + .map_err(|e| ResponseError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("decoding 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, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::timeseries::binary::FrameWriter; + use arrow_ipc::writer::StreamWriter; + + /// Written by the platform's own frame writer (`DatapointFrameWriter.wrap`, the code that + /// answers a binary read), from `feat/binary-datapoint-read`: one series per value type, then + /// the end marker. What the Rust reader must accept from the Java writer. + const JAVA_VALUES: &[u8] = include_bytes!("fixtures/read_values.bin"); + /// One codec-2 frame of three hourly buckets for series 21, `rd_aggregates`. + const JAVA_AGGREGATES: &[u8] = include_bytes!("fixtures/read_aggregates.bin"); + + const T0: i64 = 1_700_000_000_000; + + fn response(frames: &[Vec]) -> Vec { + let mut body: Vec = frames.concat(); + body.extend_from_slice(END_OF_RESPONSE); + body + } + + fn float_frame(id: u64, external_id: &str, timestamps: std::ops::Range) -> Vec { + let mut writer = FrameWriter::new(DatapointValueType::Float); + writer.series(id, external_id).unwrap(); + for t in timestamps { + writer.add_float(id, t, t as f64 / 2.0).unwrap(); + } + writer.build(1).unwrap().bytes + } + + /// A frame around any batch, with any codec byte: how a test makes a codec-2 frame. + fn frame_around(value_type: DatapointValueType, codec: u8, id: i64, name: &str, batch: &RecordBatch) -> Vec { + let mut raw = Vec::new(); + let mut writer = StreamWriter::try_new(&mut raw, batch.schema().as_ref()).unwrap(); + writer.write(batch).unwrap(); + writer.finish().unwrap(); + drop(writer); + let compressed = zstd::bulk::compress(&raw, 1).unwrap(); + let mut directory = id.to_le_bytes().to_vec(); + directory.push(name.len() as u8); + directory.extend_from_slice(name.as_bytes()); + let mut frame = MAGIC.to_vec(); + frame.extend_from_slice(&[VERSION, value_type.id(), codec, COMPRESSION_ZSTD]); + for n in [batch.num_rows(), 1, directory.len(), compressed.len(), raw.len()] { + frame.extend_from_slice(&(n as u32).to_le_bytes()); + } + frame.extend_from_slice(&directory); + frame.extend_from_slice(&compressed); + frame + } + + #[test] + fn reads_every_value_type_the_java_writer_answers_with() { + let series = parse_read_response(JAVA_VALUES).unwrap(); + let names: Vec<&str> = series.iter().map(|s| s.external_id.as_str()).collect(); + assert_eq!( + names, + ["rd_bigint", "rd_float", "rd_float32", "rd_numeric", "rd_decimal32", "rd_text", "rd_mixed", "rd_mixed_numbers"] + ); + + let bigint = &series[0]; + assert_eq!((bigint.id, bigint.value_type), (11, DatapointValueType::Bigint)); + assert_eq!(bigint.timestamps, [T0, T0 + 1000, T0 + 2000]); + assert_eq!(bigint.values, BinaryValues::Bigint(vec![1, 42, -9_000_000_000])); + + assert_eq!(series[1].values, BinaryValues::Float(vec![1.5, -2.25])); + assert_eq!(series[2].values, BinaryValues::Float32(vec![0.5])); + assert_eq!( + series[3].values, + BinaryValues::Decimal { unscaled: vec![123_456_789_012_345_678, -1], scale: 6 } + ); + assert_eq!( + series[4].values, + BinaryValues::Decimal { unscaled: vec![123_456_789], scale: 4 } + ); + assert_eq!( + series[5].values, + BinaryValues::Text(vec!["open".to_string(), "blåbær".to_string()]) + ); + assert_eq!( + series[6].values, + BinaryValues::Mixed { + numeric: vec![Some(3.5), None, Some(-7.0)], + text: vec![None, Some("fault".to_string()), None], + } + ); + // No text at all: the writer leaves the validity bitmap out of the numeric column. + assert_eq!( + series[7].values, + BinaryValues::Mixed { numeric: vec![Some(1.0), Some(2.0)], text: vec![None, None] } + ); + } + + #[test] + fn reads_an_aggregate_frame_the_java_writer_answers_with() { + let series = parse_read_response(JAVA_AGGREGATES).unwrap(); + assert_eq!(series.len(), 1); + let s = &series[0]; + assert_eq!((s.id, s.external_id.as_str()), (21, "rd_aggregates")); + assert_eq!(s.value_type, DatapointValueType::Float, "the series' own type, not the canon's"); + assert!(s.is_aggregates()); + assert_eq!(s.timestamps, [T0, T0 + 3_600_000, T0 + 7_200_000]); + assert_eq!( + s.values, + BinaryValues::Aggregates { + avg: vec![1.5, 2.5, 3.5], + min: vec![1.0, 2.0, 3.0], + max: vec![2.0, 3.0, 4.0], + sum: vec![6.0, 10.0, 14.0], + } + ); + } + + #[test] + fn consecutive_frames_of_one_series_merge_in_order() { + let body = response(&[ + float_frame(1, "a", 0..3), + float_frame(1, "a", 3..5), + float_frame(2, "b", 0..2), + ]); + let series = parse_read_response(&body).unwrap(); + assert_eq!(series.len(), 2); + assert_eq!(series[0].timestamps, [0, 1, 2, 3, 4]); + assert_eq!(series[0].values, BinaryValues::Float(vec![0.0, 0.5, 1.0, 1.5, 2.0])); + assert_eq!(series[1].external_id, "b"); + } + + /// Decoded on several threads, merged in the order the frames came, and a failure reported for + /// the earliest frame that has one, whichever thread reached it first. + #[test] + fn many_frames_decode_in_parallel_and_keep_their_order() { + let mut frames: Vec> = Vec::new(); + for id in 1..=64u64 { + let start = id as i64 * 1_000; + frames.push(float_frame(id, &format!("s{id}"), start..start + 3)); + frames.push(float_frame(id, &format!("s{id}"), start + 3..start + 5)); + } + let series = parse_read_response(&response(&frames)).unwrap(); + assert_eq!(series.len(), 64); + for (i, s) in series.iter().enumerate() { + let id = i as u64 + 1; + assert_eq!((s.id, s.external_id.as_str()), (id, format!("s{id}").as_str())); + let start = id as i64 * 1_000; + assert_eq!(s.timestamps, (start..start + 5).collect::>()); + } + + 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, 1, "a", &batch); + frames[100] = bad.clone(); + frames[17] = bad; + let error = parse_read_response(&response(&frames)).unwrap_err(); + assert!(error.starts_with("frame 17:"), "{error}"); + } + + /// 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] + fn a_series_whose_timestamps_restart_is_a_second_entry() { + let body = response(&[float_frame(1, "a", 0..3), float_frame(1, "a", 1..2)]); + let series = parse_read_response(&body).unwrap(); + assert_eq!(series.len(), 2); + assert_eq!(series[1].timestamps, [1]); + } + + #[test] + fn the_end_marker_alone_is_no_points() { + assert!(parse_read_response(END_OF_RESPONSE).unwrap().is_empty()); + } + + #[test] + fn a_response_cut_short_is_an_error_not_fewer_points() { + let first = float_frame(1, "a", 0..3); + let full = response(&[first.clone(), float_frame(2, "b", 0..3)]); + + // Between two frames: every byte that arrived parses, only the marker is missing. + let error = parse_read_response(&full[..first.len()]).unwrap_err(); + assert!(error.contains("cut short"), "{error}"); + // Inside a frame, in the header and in the payload. + for cut in [first.len() + 10, first.len() + HEADER_BYTES + 12, full.len() - 5] { + let error = parse_read_response(&full[..cut]).unwrap_err(); + assert!(error.contains("cut short"), "cut at {cut}: {error}"); + } + // Nothing at all. + assert!(parse_read_response(&[]).unwrap_err().contains("cut short")); + } + + #[test] + fn bytes_after_the_end_marker_are_refused() { + let mut body = response(&[float_frame(1, "a", 0..2)]); + body.push(0); + assert!(parse_read_response(&body).unwrap_err().contains("after the end marker")); + } + + #[test] + fn a_payload_outside_its_canon_is_refused() { + // Values in a codec-2 frame: right envelope, wrong schema. + 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 body = response(&[frame_around(DatapointValueType::Float, CODEC_ARROW_IPC_AGGREGATES, 1, "a", &batch)]); + assert!(parse_read_response(&body).unwrap_err().contains("canon")); + + let body = response(&[frame_around(DatapointValueType::Float, 9, 1, "a", &batch)]); + assert!(parse_read_response(&body).unwrap_err().contains("codec 9")); + } + + #[test] + fn a_row_not_named_in_the_directory_is_refused() { + let batch = RecordBatch::try_new( + DatapointValueType::Float.schema(), + vec![ + Arc::new(Int64Array::from(vec![2])), + Arc::new(TimestampMillisecondArray::from(vec![0]).with_timezone("UTC")), + Arc::new(Float64Array::from(vec![1.0])), + ], + ) + .unwrap(); + let body = response(&[frame_around(DatapointValueType::Float, CODEC_ARROW_IPC, 1, "a", &batch)]); + assert!(parse_read_response(&body).unwrap_err().contains("not in the directory")); + } + + #[test] + fn the_aggregate_row_cap_is_its_own() { + let n = MAX_AGGREGATE_ROWS; + let floats = || Arc::new(Float64Array::from(vec![1.0; n])) as Arc; + let batch = RecordBatch::try_new( + aggregates_schema(), + vec![ + Arc::new(Int64Array::from(vec![1; n])), + Arc::new(TimestampMillisecondArray::from((0..n as i64).collect::>()).with_timezone("UTC")), + floats(), + floats(), + floats(), + floats(), + ], + ) + .unwrap(); + // Text's value cap is 10 000; an aggregate frame of a mixed series may still hold 50 000. + let body = response(&[frame_around(DatapointValueType::Mixed, CODEC_ARROW_IPC_AGGREGATES, 1, "a", &batch)]); + let series = parse_read_response(&body).unwrap(); + assert_eq!(series[0].len(), n); + } +} diff --git a/src/timeseries/fixtures/GenerateReadFixtures.java b/src/timeseries/fixtures/GenerateReadFixtures.java new file mode 100644 index 0000000..aab854a --- /dev/null +++ b/src/timeseries/fixtures/GenerateReadFixtures.java @@ -0,0 +1,115 @@ +// Regenerates read_values.bin and read_aggregates.bin with the platform's own frame writer, the +// code that answers POST /timeseries/data/list/binary. Run from a platform checkout that has built +// datahub-api-model, with arrow-format, flatbuffers-java, zero-allocation-hashing, zstd-jni and +// slf4j-api on the classpath: +// +// java -cp ":" \ +// GenerateReadFixtures.java src/timeseries/fixtures +// +// The Rust tests in binary_read.rs assert on the exact values written here. +import ai.intellistream.datahub.api.binary.*; +import ai.intellistream.datahub.api.binary.ArrowIpc.*; + +import java.io.ByteArrayOutputStream; +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +/** Writes binary-read response bodies with the platform's own writer, for the Rust reader to parse. */ +public class GenerateReadFixtures { + static final long T0 = 1_700_000_000_000L; + static final PayloadCodec CODEC = new ZstdPayloadCodec(3); + + /** An ingest frame's batches, rewrapped the way the API answers a read. */ + static byte[] response(DatapointValueType type, long id, String ext, DatapointFrameWriter w) { + byte[] ingest = w.build(CODEC); + int dirLen = ByteBuffer.wrap(ingest).order(ByteOrder.LITTLE_ENDIAN).getInt(16); + int payLen = ByteBuffer.wrap(ingest).order(ByteOrder.LITTLE_ENDIAN).getInt(20); + int rawLen = ByteBuffer.wrap(ingest).order(ByteOrder.LITTLE_ENDIAN).getInt(24); + byte[] payload = java.util.Arrays.copyOfRange(ingest, 28 + dirLen, 28 + dirLen + payLen); + byte[] raw = CODEC.decompress(payload, rawLen); + List data = new ArrayList<>(); + for (Batch b : ArrowIpc.readStream(ArrowSchemaCanon.of(type), raw, 0, 100_000)) { + data.add(b.toData()); + } + return DatapointFrameWriter.wrap(type, false, id, ext, data, CODEC); + } + + static ByteBuffer longs(long... v) { + ByteBuffer b = ByteBuffer.allocate(v.length * 8).order(ByteOrder.LITTLE_ENDIAN); + for (long x : v) b.putLong(x); + return b.flip(); + } + + static ByteBuffer doubles(double... v) { + ByteBuffer b = ByteBuffer.allocate(v.length * 8).order(ByteOrder.LITTLE_ENDIAN); + for (double x : v) b.putDouble(x); + return b.flip(); + } + + public static void main(String[] args) throws Exception { + Path out = Path.of(args[0]); + ByteArrayOutputStream values = new ByteArrayOutputStream(); + + var bigint = DatapointFrameWriter.forType(DatapointValueType.BIGINT).series(11, "rd_bigint"); + bigint.addBigint(11, T0 + 2000, -9_000_000_000L); + bigint.addBigint(11, T0, 1); + bigint.addBigint(11, T0 + 1000, 42); + values.writeBytes(response(DatapointValueType.BIGINT, 11, "rd_bigint", bigint)); + + var flt = DatapointFrameWriter.forType(DatapointValueType.FLOAT).series(12, "rd_float"); + flt.addFloat(12, T0, 1.5); + flt.addFloat(12, T0 + 1000, -2.25); + values.writeBytes(response(DatapointValueType.FLOAT, 12, "rd_float", flt)); + + var f32 = DatapointFrameWriter.forType(DatapointValueType.FLOAT32).series(13, "rd_float32"); + f32.addFloat32(13, T0, 0.5f); + values.writeBytes(response(DatapointValueType.FLOAT32, 13, "rd_float32", f32)); + + var numeric = DatapointFrameWriter.forType(DatapointValueType.NUMERIC).series(14, "rd_numeric"); + numeric.addNumeric(14, T0, new BigDecimal("123456789012.345678")); + numeric.addNumeric(14, T0 + 1000, new BigDecimal("-0.000001")); + values.writeBytes(response(DatapointValueType.NUMERIC, 14, "rd_numeric", numeric)); + + var dec32 = DatapointFrameWriter.forType(DatapointValueType.DECIMAL32).series(15, "rd_decimal32"); + dec32.addDecimal32(15, T0, new BigDecimal("12345.6789")); + values.writeBytes(response(DatapointValueType.DECIMAL32, 15, "rd_decimal32", dec32)); + + var text = DatapointFrameWriter.forType(DatapointValueType.TEXT).series(16, "rd_text"); + text.addText(16, T0, "open"); + text.addText(16, T0 + 1000, "blåbær"); + values.writeBytes(response(DatapointValueType.TEXT, 16, "rd_text", text)); + + var mixed = DatapointFrameWriter.forType(DatapointValueType.MIXED).series(17, "rd_mixed"); + mixed.addMixed(17, T0, "3.5"); + mixed.addMixed(17, T0 + 1000, "fault"); + mixed.addMixed(17, T0 + 2000, "-7"); + values.writeBytes(response(DatapointValueType.MIXED, 17, "rd_mixed", mixed)); + + // Only numbers: the API leaves the validity bitmap out of a column with no nulls. + var mixedNumbers = DatapointFrameWriter.forType(DatapointValueType.MIXED).series(18, "rd_mixed_numbers"); + mixedNumbers.addMixed(18, T0, "1"); + mixedNumbers.addMixed(18, T0 + 1000, "2"); + values.writeBytes(response(DatapointValueType.MIXED, 18, "rd_mixed_numbers", mixedNumbers)); + + values.writeBytes(FrameLimits.END_OF_RESPONSE); + Files.write(out.resolve("read_values.bin"), values.toByteArray()); + + ByteArrayOutputStream aggregates = new ByteArrayOutputStream(); + BatchData buckets = new BatchData(3, new ColumnData[]{ + new ColumnData(null, 0, null, longs(21, 21, 21)), + new ColumnData(null, 0, null, longs(T0, T0 + 3_600_000, T0 + 7_200_000)), + new ColumnData(null, 0, null, doubles(1.5, 2.5, 3.5)), + new ColumnData(null, 0, null, doubles(1, 2, 3)), + new ColumnData(null, 0, null, doubles(2, 3, 4)), + new ColumnData(null, 0, null, doubles(6, 10, 14)), + }); + aggregates.writeBytes(DatapointFrameWriter.wrap(DatapointValueType.FLOAT, true, 21, "rd_aggregates", List.of(buckets), CODEC)); + aggregates.writeBytes(FrameLimits.END_OF_RESPONSE); + Files.write(out.resolve("read_aggregates.bin"), aggregates.toByteArray()); + } +} diff --git a/src/timeseries/fixtures/read_aggregates.bin b/src/timeseries/fixtures/read_aggregates.bin new file mode 100644 index 0000000000000000000000000000000000000000..adab85f38dd0ec815623a6129f14ca2bd4af309f GIT binary patch literal 401 zcmV;C0dD?8NJLNp0s;X8000010000M000160RRA~0{{RO000000000Da%5j&XJ>L{ zXJK?@b11bh{a~m9jS2ulPAxHTYtk6xSicJtN&Wq+D$#iCK3(;1O=W+dTYj{$~@XBizK5mzefr4Ndr7=0jb z-Mw*pegZ#s?e4CD4ps#zs(^oDQ%;NIh$1}^sf|@dxcLD#P*`A}7uOlWf`U#0jV0iR z=8K9P1;+~Of0khU?e+P0SHk=aKj-f}lYZ#ipXdK>%$-k$e!HV*oZa15s9{-TGhqsv z!ZHT%0-2d94-!BlF(eidC5nvg{~L_1HS3W}gWNCXeVGVOQy+$HMi{rY-}ZdOXdk|U zYYQ%8e(BG<8NGf~!V*^U8}cT1tFTZRN4qcTbYG>UBg3@xIrOA$)ioTkr*K}d#0}#O v>M*nJQ6L%4B_x9@Z*S~->O;vYS)2n}i%)xZ8LiK!|E>c5QGXhyL`XzMbOx+Yocf{mjgzRepu7A{z=F(q&_FUukkixtj_(xg=Q%$t6b_ zAxy{2y(FR1$(%WQNaa?XBZu_+PMt^Z$K(C@d>)@apO5GJ_5P4tNS**#FbD(!fUN?y z5_oZeLLf-+E-3_yej*?=IP~Nh(?+W=R2RTb79w;rvPcT&!imjh|TP_ZMNa>uSl|w1gd7KgVa=GGt zL+ZTf`T2QIyO64k6zDZN%;WJJmO<*PkScQH4u%lg$vt))!z`-4DDZV)ak^t01WWu* zLUT`P+5K5&!|jOMbc@Ee2@%{LBS=9Td?2LuT*~O6HlY6<%ZRx$8eW|RC zbIb{cL0l)qn`Pdqu@)GBB%ucd01(zDJ!SBOgY*dCgS-SCSh66! zpomknZ}13!NCq;DG&IY$cmlM%eb(rnTlj2tRc#rObZIn_xP6J@moWDpRyLIKW+OM^ z8@kF4hnbk>8mv7*2j8@}tch`tx~&vmaE439@W(~WYntZ6M*46ar?jK60qOKhC1gvr zbiCgdoDEG!Nt-Fw^fo-o=TL*W_weH^S7vg`H&&OkUDf%k+ zVHjyep3I}|Xq&uxi_Rb|0d#dR;_Cg^=pZyS*rcl>0O5nFps}~tDb*hyh%ay-my4>W z4zGV zBqM1Q*lS>8Kw=8uDJi-Ch}qybg~I?@GLZU57_%g|VJ#cX#hn~$n1%mT^UU^|vqigp z8y!N+*>HwQSt#-0{WaU){uiH1l6C;(Uutgy}j! znk_U*fuwTFxy85WQB6)SvNj}fb`(@_u3B?-hqLXvs&g8YKdjmmq zHXE?Ul%yzY--k&7wtZb_*NIG~O{RL^gEWn4;&cb@^!Z%Z!#K?9d9$+$A&c|-EIFgL zoH=H7md~4sdSc6`tW}itqn>DP+0vnq*%a9a7jTs^PwJo=sS^{G&ayfmay@8EHM8Y+ zq?fKbLpR-m{cuY{3JD^UOy-tNPTxqKjAk-ep@H8=Q$ZGD;AXfvnj=Xt*TSrW*Yn>U zEz5MV)E5Tp#bzi*Lxl7hG-6KM?|Pr;VLv#U;f64$GO^(#ZK-u;F_KW&;A`C1a_dla z9Ok;9sfB-z*qvQ!j*!DVD)J(slUqXYsujaWKRME93d$}zmBtvX>Sdj?j*m_CE+(+m zrF#i)-ddhpvozC2Fv~|WMIyK@h6mLG^}0r?;4CD6qVx>ho_)kC1dwmtpwMZzX&@lB zw?t_y9J1({pDJJP%3IiHQ8RfyVDVH~OOJWi)`G^i*CvU3BB&b8INg%wuDTa{d&V16 z6;}QluArX#{Kg0r2_k;IOc&1WWH2CMJtq9naHlxw3T+( z(AH6e7GG|BqLw$9;(^c36u)og^NFL>WCMh$V`J}r*L~im8c6SSEJ6^`MdvJXi9hj36Tz(~40%9ImW*maR1G}mBcHEz zWWL4U(VC&5(zGWbM~C-nUHA7Sx0U@x1ufhcwJ5|UEU6!wDS?1cpk|&REXiF`G9auo zjF@i;<_S<_AuObkxg`E?)Zl7PJLeg=c%F)#yFSp{^SLLm))gnMbtffkdBbJ4mihCM zJTUzWI%YhvJJk<#adMKxzCIxl4p($-p=Qh1Mh@)LplgrqoS)z6t!3-A%?9D%lvbaz zACW=D=4A!7dfO{j@dPD7Ioc$2v9To>Wa0S^?Yoeis4cNK7^ey{>OGQ}-bWttkpxNl zbEg2#xGV1=a0ap~JI+986pLhC3LAp1^;=pS4O|$2)nTlq6n%Fy`~Qj!h842@yXH)j+AokKwCpIG1`!-qB3-x6nX z(0*Tip65d@6gTNFb0U1EM;7g#LAhxyLHAaF$2qZ2zWybT(nJQnqHOjmg%QPO3nuo; zdbOH5x&_-*%Adl+hU7>?FwOp$PSN94_<94Cy^@%%PPtO}S#>MR&~s0kDur_8&AOq9 zC2sxK1e?h4^(}Z*xhs^~J?60i?APy`{fB;}D>NNtTpgpJ_eCT!UDrtLt!)==tQ=ym z6@K}JSeq{rxmTS&$Rg5I{Lb4ALAj;xl`(gh+-c-Nplo$?wV`Q|m$+DS%lgdn(Z7Qu z=1DyfH@sl)zqrRw&~%dLa%O<2?s2byKC7xvlDgDR%1elKj3|7=8`!V1c0wT;Zvy`~Ibj=gsa_Z!BEIMy}d{A#Z zDzkR?pr+d8{{eTYF+^09{rp5j|Ac=9xJ+UfKZ;NDhi1TZbOj!st!eDs-@b@G{5%DmHZWI+@H> zGrMu9KEvYiQIwKa#xuFw@oepD{zX$WBk!J99jweAHhf7af;h}}E7yM|Od+)?|I6Cn5uJ#R%NUot3mZ9V6k99%ja-Vq zbb4vS@v9&7T6590e~y$17zTT~LzMGcHR5=xbMM$2NX9cw`K#U9T^+UMHx-%A5e3n` V7xw2RkCR+JDds7XCI6ef{|B(lG0y-1 literal 0 HcmV?d00001 diff --git a/src/timeseries/mod.rs b/src/timeseries/mod.rs index e34231f..f8a64dc 100644 --- a/src/timeseries/mod.rs +++ b/src/timeseries/mod.rs @@ -1,8 +1,10 @@ pub mod binary; +pub mod binary_read; pub mod datapoint_listen; mod test; pub use binary::{BinaryIngestOptions, DatapointValueType, Frame, FrameWriter, ResolvedSeries}; +pub use binary_read::{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 2f7bb38..18fba9e 100644 --- a/src/timeseries/test.rs +++ b/src/timeseries/test.rs @@ -12,7 +12,8 @@ mod tests { use crate::generic::{DataWrapper, Datapoint, DatapointString, DatapointsCollection, DeleteFilter, IdAndExtId, RetrieveFilter}; use crate::http::ResponseError; use crate::timeseries::{TimeSeries, TimeSeriesFilter, TimeSeriesFilterForm, TimeSeriesUpdate, TimeSeriesUpdateCollection, TimeSeriesUpdateFields}; - use crate::timeseries::binary::BinaryIngestOptions; + use crate::timeseries::binary::{BinaryIngestOptions, DatapointValueType}; + use crate::timeseries::binary_read::{BinaryDatapoints, BinaryValues}; use crate::tests::cleanup::cleanup_timeseries; use crate::tests::ids::{unique_id, unique_token}; use crate::tests::polling::poll_until_for; @@ -712,6 +713,163 @@ mod tests { Ok(()) } + /// `POST /timeseries/data/list/binary` against the JSON read of the same series: every point + /// of a float series past one frame's row cap, hourly aggregates, a text series, an empty + /// window, and the two refusals the JSON read does not have. Needs a backend that serves the + /// binary read. + #[tokio::test] + #[ignore] + async fn test_datapoints_binary_read() -> Result<(), Box> { + let api_service = create_api_service(); + let float_ext_id = unique_id("bin_read_float"); + let text_ext_id = unique_id("bin_read_text"); + let mut ts_collection = DataWrapper::new(); + for (external_id, value_type) in [(&float_ext_id, "float"), (&text_ext_id, "text")] { + ts_collection.add_item( + TimeSeries::builder() + .set_external_id(external_id) + .set_name(external_id) + .set_unit("celsius") + .set_value_type(value_type) + .clone(), + ); + } + let mut cleanup = cleanup_timeseries(vec![float_ext_id.clone(), text_ext_id.clone()]); + api_service.time_series.create(&ts_collection).await.expect("could not create the series"); + + // 150 000 one-second floats: past the 100 000-row numeric frame cap and past the 100 000 + // an item's `limit` may ask for, so they come back whole only from an item without one, + // as two frames. 25 000 texts: past the 10 000-row text frame cap, so three frames. + let start = Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap(); + let count: i64 = 150_000; + let value = |i: i64| (i % 3600) as f64 * 0.25 - 100.0; + let mut floats = DatapointsCollection::from_external_id(&float_ext_id); + floats.datapoints = (0..count) + .map(|i| DatapointString::new(&(start.timestamp_millis() + i * 1000).to_string(), &value(i).to_string())) + .collect(); + let mut texts = DatapointsCollection::from_external_id(&text_ext_id); + let text_count: i64 = 25_000; + let text = |i: i64| format!("blåbær_{}", i % 97); + texts.datapoints = (0..text_count) + .map(|i| DatapointString::new(&(start.timestamp_millis() + i * 1000).to_string(), &text(i))) + .collect(); + api_service.time_series + .insert_datapoints_binary(&DataWrapper::from_vec(vec![floats, texts]), &BinaryIngestOptions::default()) + .await + .unwrap_or_else(|e| panic!("binary insert failed with {}: {}", e.get_status(), e.get_message())); + let item = |external_id: &str, limit: u64| { + let mut rf = RetrieveFilter::new(); + rf.set_external_id(external_id); + rf.set_start(start); + rf.set_limit(limit); + rf + }; + poll_all_points(&api_service, &float_ext_id, count as usize).await; + // Not poll_all_points: the JSON `Datapoint` holds only numbers and cannot read text. + let text_request = DataWrapper::from_vec(vec![item(&text_ext_id, text_count as u64)]); + poll_until_for( + std::time::Duration::from_secs(60), + || api_service.time_series.retrieve_datapoints_binary(&text_request), + |read| matches!(read, Ok(series) if series.first().is_some_and(|s| s.len() == text_count as usize)), + ) + .await; + + println!("Every point, in order, across frames..."); + let mut whole_window = RetrieveFilter::new(); + whole_window.set_external_id(&float_ext_id); + whole_window.set_start(start); + whole_window.set_end(start + Duration::seconds(count)); + let first = api_service.time_series + .retrieve_datapoints_binary(&DataWrapper::from_vec(vec![whole_window, item(&text_ext_id, text_count as u64)])) + .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"); + let float = first[0].clone(); + assert_eq!(float.external_id, float_ext_id); + assert_eq!(float.value_type, DatapointValueType::Float); + let want_timestamps: Vec = (0..count).map(|i| start.timestamp_millis() + i * 1000).collect(); + assert_eq!(float.timestamps, want_timestamps); + let want_values: Vec = (0..count).map(value).collect(); + assert_eq!(float.values, BinaryValues::Float(want_values)); + + let texts = &first[1]; + assert_eq!(texts.external_id, text_ext_id); + assert_eq!(texts.timestamps, want_timestamps[..text_count as usize]); + assert_eq!(texts.values, BinaryValues::Text((0..text_count).map(text).collect())); + + println!("The same window as the JSON read..."); + let json = api_service.time_series + .retrieve_datapoints(&DataWrapper::from_vec(vec![item(&float_ext_id, 1000)])) + .await?; + let json_points = &json.get_items()[0].datapoints; + assert_eq!(json_points.len(), 1000); + for (i, dp) in json_points.iter().enumerate() { + assert_eq!(dp.timestamp.timestamp_millis(), float.timestamps[i]); + assert_eq!(dp.value, Some(want_value_at(&float, i))); + } + + println!("Hourly aggregates against the JSON aggregates..."); + let hourly = |mut rf: RetrieveFilter| { + rf.set_aggregates(vec!["avg".to_string(), "min".to_string(), "max".to_string(), "sum".to_string()]); + rf.set_granularity("1h"); + rf.set_end(start + Duration::seconds(count)); + DataWrapper::from_vec(vec![rf]) + }; + let mut rf = RetrieveFilter::new(); + rf.set_external_id(&float_ext_id); + rf.set_start(start); + let binary_aggs = api_service.time_series.retrieve_datapoints_binary(&hourly(rf.clone())).await?; + let json_aggs = api_service.time_series.retrieve_datapoints(&hourly(rf)).await?; + let json_buckets = &json_aggs.get_items()[0].datapoints; + let BinaryValues::Aggregates { avg, min, max, sum } = &binary_aggs[0].values else { + panic!("an aggregate read came back as {:?}", binary_aggs[0].values); + }; + assert_eq!(binary_aggs[0].len(), json_buckets.len(), "the two reads answered a different number of buckets"); + assert_eq!(json_buckets.len(), (count as usize).div_ceil(3600)); + for (i, bucket) in json_buckets.iter().enumerate() { + assert_eq!(binary_aggs[0].timestamps[i], bucket.timestamp.timestamp_millis()); + assert_eq!((Some(min[i]), Some(max[i])), (bucket.min, bucket.max), "bucket {i}"); + assert!((avg[i] - bucket.average.unwrap()).abs() < AVG_TOLERANCE, "bucket {i}: avg"); + assert!((sum[i] - bucket.sum.unwrap()).abs() < 1e-6 * sum[i].abs().max(1.0), "bucket {i}: sum"); + } + + println!("An empty window is no entry..."); + let mut empty = item(&float_ext_id, 10); + empty.set_start(start - Duration::days(10)); + empty.set_end(start - Duration::days(9)); + let none = api_service.time_series.retrieve_datapoints_binary(&DataWrapper::from_vec(vec![empty])).await?; + assert!(none.is_empty(), "{none:?}"); + + println!("Refusals..."); + let mut with_cursor = item(&float_ext_id, 10); + with_cursor.cursor = Some("anything".to_string()); + let error = api_service.time_series + .retrieve_datapoints_binary(&DataWrapper::from_vec(vec![with_cursor])) + .await + .unwrap_err(); + assert_eq!(error.get_status(), StatusCode::BAD_REQUEST, "{}", error.get_message()); + assert!(error.problem().is_some(), "a refusal is a problem document: {}", error.get_message()); + + let unknown = unique_id("bin_read_missing"); + let error = api_service.time_series + .retrieve_datapoints_binary(&DataWrapper::from_vec(vec![item(&unknown, 10)])) + .await + .unwrap_err(); + assert_eq!(error.get_status(), StatusCode::BAD_REQUEST, "{}", error.get_message()); + assert!(error.get_message().contains(&unknown), "{}", error.get_message()); + + delete_timeseries(&api_service, &[&float_ext_id, &text_ext_id]).await; + cleanup.disarm(); + Ok(()) + } + + fn want_value_at(series: &BinaryDatapoints, i: usize) -> f64 { + match &series.values { + BinaryValues::Float(values) => values[i], + other => panic!("not a float series: {other:?}"), + } + } + /// JSON against binary ingest from Rust, reporting the same columns as the Java benchmark in /// the platform's `datahub-e2e` module and the Python one in `python_tests`, so the three /// clients can be compared.