densa_data/
lib.rs

1//! Live Arrow tables: rows identified by a key, updated by record batches, and views that
2//! stay sorted and filtered as the rows change.
3//!
4//! [`LiveTable`] takes Arrow [`RecordBatch`](arrow_array::RecordBatch)es from a feed, a
5//! query engine or a file, and upserts them by a key column; [`ViewSpec`] describes a view
6//! (a [`Filter`] and sort columns), which the table keeps up to date incrementally instead of
7//! recomputing it; [`GroupSpec`] does the same for grouped aggregates (count, sum, min, max,
8//! mean), grouped by key columns or by buckets of one (counts per minute,
9//! [`GroupSpec::group_by_time`]). A view's rows are [`RowRef`]s into
10//! [`LiveTable::batches`], so a display layer (such as `densa-arrow`'s grid) reads values
11//! straight from the Arrow arrays. Each key's row is a [`RecordId`] while the key lives: a
12//! selection can follow it through updates, compactions and re-sorted views
13//! ([`View::position`]). [`LiveTable::snapshot`] hands the live rows to anything else that
14//! reads Arrow, such as a SQL engine. A [`Window`] keeps a table to its recent rows, by a
15//! time column. A [`LookupJoin`] gives each row of one table the columns of the row of
16//! another that it names (a trade its instrument), kept up to date as either changes.
17//!
18//! This crate has no user-interface dependency.
19//!
20//! ```
21//! use std::sync::Arc;
22//! use arrow_array::{ArrayRef, Float64Array, RecordBatch, StringArray};
23//! use densa_data::{CompareOp, Filter, LiveTable, Order, Scalar, ViewSpec};
24//!
25//! let quotes = |symbols: Vec<&str>, prices: Vec<f64>| {
26//!     RecordBatch::try_from_iter([
27//!         ("symbol", Arc::new(StringArray::from(symbols)) as ArrayRef),
28//!         ("price", Arc::new(Float64Array::from(prices)) as ArrayRef),
29//!     ])
30//! };
31//! let mut table = LiveTable::new(quotes(vec![], vec![])?.schema(), "symbol")?;
32//! let expensive = table.add_view(
33//!     ViewSpec::new()
34//!         .filter(Filter::compare("price", CompareOp::Gt, Scalar::Float(100.0)))
35//!         .sort_by("price", Order::Descending),
36//! )?;
37//!
38//! table.upsert(quotes(vec!["AAA", "BBB", "CCC"], vec![150.0, 90.0, 120.0])?)?;
39//! table.upsert(quotes(vec!["BBB"], vec![200.0])?)?; // BBB rises: it enters the view
40//!
41//! let view = table.view(expensive).unwrap();
42//! assert_eq!(view.len(), 3); // BBB 200, AAA 150, CCC 120
43//! # Ok::<(), Box<dyn std::error::Error>>(())
44//! ```
45
46mod bucket;
47mod dictionary;
48mod error;
49mod filter;
50mod group;
51mod join;
52mod key;
53mod snapshot;
54mod sorted;
55mod table;
56mod value;
57mod view;
58mod window;
59
60pub use arrow_array;
61pub use arrow_schema;
62pub use error::DataError;
63pub use filter::{CompareOp, Filter, Scalar};
64pub use group::{Aggregate, GroupSpec, Groups};
65pub use join::LookupJoin;
66pub use table::{Changes, GroupsId, LiveTable, RecordId, RowRef, TableOptions, View, ViewId};
67pub use view::{Order, ViewSpec};
68pub use window::Window;
69
70/// A row's internal identity: stable for as long as its key is live.
71type RowId = u32;
densa_data/lib.rs

densa_data/
lib.rs
