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

densa_arrow/
series.rs
