densa_arrow/
series.rs

1//! Two numeric columns of a live table's view as a chart series.
2
3use std::fmt;
4
5use arrow_array::cast::AsArray;
6use arrow_array::types::{
7    ArrowPrimitiveType, Date32Type, Date64Type, Decimal128Type, Float32Type, Float64Type,
8    Int16Type, Int32Type, Int64Type, Int8Type, TimestampMicrosecondType, TimestampMillisecondType,
9    TimestampNanosecondType, TimestampSecondType, UInt16Type, UInt32Type, UInt64Type, UInt8Type,
10};
11use arrow_array::{Array, ArrayRef};
12use arrow_schema::{DataType, TimeUnit};
13use densa_chart::Points;
14use densa_data::{LiveTable, View, ViewId};
15
16/// Why a series could not be made.
17#[derive(Debug, Clone, PartialEq, Eq)]
18#[non_exhaustive]
19pub enum SeriesError {
20    /// The table has no column of this name.
21    NoSuchColumn(String),
22    /// The column of this name holds values that are not numbers, times or dates.
23    NotNumeric(String),
24}
25
26impl fmt::Display for SeriesError {
27    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
28        match self {
29            Self::NoSuchColumn(name) => write!(f, "no column named {name:?}"),
30            Self::NotNumeric(name) => write!(
31                f,
32                "column {name:?} is not numeric: a series reads integers, f32 and f64 floats, \
33                 decimals, timestamps and dates"
34            ),
35        }
36    }
37}
38
39impl std::error::Error for SeriesError {}
40
41/// Reads one value of a column as a number: NaN for a null, or for an array of another type.
42type NumberFn = fn(&dyn Array, usize) -> f64;
43
44/// A column found by name, with its reader.
45#[derive(Debug, Clone, Copy)]
46struct Column {
47    index: usize,
48    read: NumberFn,
49}
50
51/// Two columns of a `densa_data::LiveTable` read as numbers, `x` and `y`, so a view of the
52/// table is a `densa_chart` series ([`points`](Self::points)).
53///
54/// Integers, `Float32`, `Float64` and `Decimal128` read as their value; timestamps as
55/// seconds since the epoch (whatever their unit and time zone: instants are UTC), and dates
56/// as the seconds of their midnight, so time columns go straight onto a time axis. Other
57/// types (`Float16`, `Decimal256`, times of day, durations, dictionaries) are not numeric
58/// here. Nulls are gaps.
59///
60/// A chart needs `x` ascending: give a view sorted by the `x` column, ascending. Its nulls
61/// sort last, past every point a chart looks for, so they cost nothing; a descending view
62/// draws wrong.
63///
64/// Like [`LiveGrid`](crate::LiveGrid), it keeps a handle to each batch's two arrays, brought up
65/// to date by [`sync`](Self::sync), which allocates only when batches were added or
66/// compacted; reading points allocates nothing. Until it is synced after a change,
67/// [`points`](Self::points) gives nothing (never values from the wrong batch): sync it right
68/// after changing the table, and before painting. Given another table, it finds its columns
69/// again by name.
70#[derive(Debug, Clone)]
71pub struct LiveSeries {
72    names: (String, String),
73    /// `None` while the table has no numeric columns of those names.
74    columns: Option<(Column, Column)>,
75    /// The `x` and `y` arrays of each batch of the table.
76    arrays: Vec<(ArrayRef, ArrayRef)>,
77    /// The table object (`LiveTable::instance`) and generation the arrays are from.
78    table: (u64, u64),
79}
80
81impl LiveSeries {
82    /// A series of `table`'s columns named `x` and `y`.
83    ///
84    /// # Errors
85    /// [`SeriesError::NoSuchColumn`] or [`SeriesError::NotNumeric`] for either column.
86    pub fn new(table: &LiveTable, x: &str, y: &str) -> Result<Self, SeriesError> {
87        let columns = (column(table, x)?, column(table, y)?);
88        let mut series = Self {
89            names: (x.to_owned(), y.to_owned()),
90            columns: Some(columns),
91            arrays: Vec::new(),
92            table: (table.instance(), table.generation()),
93        };
94        series.sync(table);
95        Ok(series)
96    }
97
98    /// Takes in `table`'s batches added or compacted since the last call (or another
99    /// table: its columns are found again by name). Call it after the table changed, before
100    /// reading points.
101    pub fn sync(&mut self, table: &LiveTable) {
102        let current = (table.instance(), table.generation());
103        if current != self.table {
104            if self.table.0 != current.0 {
105                self.columns = column(table, &self.names.0)
106                    .and_then(|x| Ok((x, column(table, &self.names.1)?)))
107                    .ok();
108            }
109            self.arrays.clear();
110            self.table = current;
111        }
112        let Some((x, y)) = self.columns else {
113            return;
114        };
115        let new = table.batches().get(self.arrays.len()..).unwrap_or_default();
116        for batch in new {
117            self.arrays
118                .push((batch.column(x.index).clone(), batch.column(y.index).clone()));
119        }
120    }
121
122    /// The name of the `x` column.
123    pub fn x_name(&self) -> &str {
124        &self.names.0
125    }
126
127    /// The name of the `y` column.
128    pub fn y_name(&self) -> &str {
129        &self.names.1
130    }
131
132    /// Whether the table last synced has numeric columns of these names (after another
133    /// table, it may not).
134    pub fn is_resolved(&self) -> bool {
135        self.columns.is_some()
136    }
137
138    /// The rows of `view` as points, in the view's order, or `None` if the view does not
139    /// exist, the table changed since [`sync`](Self::sync) (never values from the wrong
140    /// batch), or the table has no numeric columns of the names given.
141    pub fn points<'a>(&'a self, table: &'a LiveTable, view: ViewId) -> Option<SeriesPoints<'a>> {
142        let (x, y) = self.columns?;
143        let synced = self.table == (table.instance(), table.generation())
144            && self.arrays.len() == table.batches().len();
145        if !synced {
146            return None;
147        }
148        Some(SeriesPoints {
149            arrays: &self.arrays,
150            x: x.read,
151            y: y.read,
152            rows: table.view(view)?,
153        })
154    }
155}
156
157/// A view's rows as points of a [`LiveSeries`]. Cheap to make each frame; reading allocates
158/// nothing.
159#[derive(Debug, Clone, Copy)]
160pub struct SeriesPoints<'a> {
161    arrays: &'a [(ArrayRef, ArrayRef)],
162    x: NumberFn,
163    y: NumberFn,
164    rows: View<'a>,
165}
166
167impl SeriesPoints<'_> {
168    /// The lowest and highest `x`, for a view sorted by `x` (nulls sort last, so the
169    /// finite values come first), or `None` if there is none: found by binary search.
170    pub fn x_extent(&self) -> Option<(f64, f64)> {
171        let first = self.x(0);
172        if !first.is_finite() {
173            return None;
174        }
175        // The rows with a finite `x` come first: find where they end.
176        let (mut low, mut high) = (0, self.rows.len());
177        while low < high {
178            let middle = low + (high - low) / 2;
179            if self.x(middle).is_finite() {
180                low = middle + 1;
181            } else {
182                high = middle;
183            }
184        }
185        let last = self.x(low - 1);
186        Some((first.min(last), first.max(last)))
187    }
188
189    /// Reads row `index` of the view with `read` from the array `pick` chooses.
190    fn read(
191        &self,
192        index: usize,
193        pick: fn(&(ArrayRef, ArrayRef)) -> &ArrayRef,
194        read: NumberFn,
195    ) -> f64 {
196        self.rows
197            .row(index)
198            .and_then(|at| Some((self.arrays.get(at.batch)?, at.row)))
199            .map_or(f64::NAN, |(arrays, row)| read(pick(arrays).as_ref(), row))
200    }
201}
202
203impl Points for SeriesPoints<'_> {
204    fn len(&self) -> usize {
205        self.rows.len()
206    }
207
208    fn x(&self, index: usize) -> f64 {
209        self.read(index, |arrays| &arrays.0, self.x)
210    }
211
212    fn y(&self, index: usize) -> f64 {
213        self.read(index, |arrays| &arrays.1, self.y)
214    }
215}
216
217/// The column of `table` named `name`, with a reader for its type.
218fn column(table: &LiveTable, name: &str) -> Result<Column, SeriesError> {
219    let schema = table.schema();
220    let (index, field) = schema
221        .column_with_name(name)
222        .ok_or_else(|| SeriesError::NoSuchColumn(name.to_owned()))?;
223    let read =
224        number_reader(field.data_type()).ok_or_else(|| SeriesError::NotNumeric(name.to_owned()))?;
225    Ok(Column { index, read })
226}
227
228/// The reader for values of `data_type`, or `None` if they are not numbers, times or dates.
229fn number_reader(data_type: &DataType) -> Option<NumberFn> {
230    Some(match data_type {
231        DataType::Int8 => read::<Int8Type>,
232        DataType::Int16 => read::<Int16Type>,
233        DataType::Int32 => read::<Int32Type>,
234        DataType::Int64 => read::<Int64Type>,
235        DataType::UInt8 => read::<UInt8Type>,
236        DataType::UInt16 => read::<UInt16Type>,
237        DataType::UInt32 => read::<UInt32Type>,
238        DataType::UInt64 => read::<UInt64Type>,
239        DataType::Float32 => read::<Float32Type>,
240        DataType::Float64 => read::<Float64Type>,
241        DataType::Decimal128(_, _) => read_decimal,
242        DataType::Timestamp(TimeUnit::Second, _) => read::<TimestampSecondType>,
243        DataType::Timestamp(TimeUnit::Millisecond, _) => read_scaled::<TimestampMillisecondType, 3>,
244        DataType::Timestamp(TimeUnit::Microsecond, _) => read_scaled::<TimestampMicrosecondType, 6>,
245        DataType::Timestamp(TimeUnit::Nanosecond, _) => read_scaled::<TimestampNanosecondType, 9>,
246        DataType::Date32 => read_days,
247        DataType::Date64 => read_scaled::<Date64Type, 3>,
248        _ => return None,
249    })
250}
251
252/// A value that converts to a number for a chart.
253trait ToNumber {
254    fn to_number(self) -> f64;
255}
256
257macro_rules! to_number {
258    ($($t:ty),*) => {$(
259        impl ToNumber for $t {
260            fn to_number(self) -> f64 {
261                // Lossless enough: a chart draws to a pixel, far coarser than 2^-53.
262                self as f64
263            }
264        }
265    )*};
266}
267
268to_number!(i8, i16, i32, i64, u8, u16, u32, u64, f32, f64);
269
270/// Reads a primitive value as a number.
271fn read<T>(array: &dyn Array, row: usize) -> f64
272where
273    T: ArrowPrimitiveType,
274    T::Native: ToNumber,
275{
276    match array.as_primitive_opt::<T>() {
277        Some(values) if row < values.len() && values.is_valid(row) => values.value(row).to_number(),
278        _ => f64::NAN,
279    }
280}
281
282/// Reads a primitive value counted in `10^-DIGITS` of a unit as a number of units.
283fn read_scaled<T, const DIGITS: i32>(array: &dyn Array, row: usize) -> f64
284where
285    T: ArrowPrimitiveType,
286    T::Native: ToNumber,
287{
288    read::<T>(array, row) / 10f64.powi(DIGITS)
289}
290
291/// Reads days since the epoch as the seconds of that midnight.
292fn read_days(array: &dyn Array, row: usize) -> f64 {
293    read::<Date32Type>(array, row) * 86_400.0
294}
295
296/// Reads a decimal as its value: the integer over ten to its scale.
297fn read_decimal(array: &dyn Array, row: usize) -> f64 {
298    match array.as_primitive_opt::<Decimal128Type>() {
299        Some(values) if row < values.len() && values.is_valid(row) => {
300            // Lossless enough: a chart draws to a pixel.
301            let value = values.value(row) as f64;
302            let scale = i32::from(values.scale());
303            // Dividing by an exact power of ten, rather than multiplying by an inexact
304            // tenth, keeps 3 at scale -2 and 1.5 at scale 2 exact.
305            if scale >= 0 {
306                value / 10f64.powi(scale)
307            } else {
308                value * 10f64.powi(-scale)
309            }
310        }
311        _ => f64::NAN,
312    }
313}
densa_arrow/series.rs

densa_arrow/
series.rs
