Struct LookupJoin 

pub struct LookupJoin { /* private fields */ }

Joins the rows of a live table to the rows of another, the right table, that they name: each output row is a left row followed by the columns of the right row whose key is in the left row’s on column (nulls when there is none: a left outer join). The join keeps the right table; the output is an ordinary LiveTable the application owns (made by table), so views, groups, windows and grids work on it as on any table.

Left rows go in through upsert_left, joined as they arrive; right rows through upsert_right and delete_right, which join again every output row naming a changed right key. Remove left rows from the output table itself (delete, or a Window): the join notices. To keep only rows that found a match, view the output with Filter::is_not_null on a right column.

Rows reach the output only through upsert_left: delete from it, window it, compact it, view and group it, but do not upsert into it directly, or those rows are not joined again when their right row changes.

A right row’s change rewrites every output row naming it, so a right key named by many rows (one instrument of most trades) costs that many rows each time it changes. The join allocates as it works (its index, the joined batches).

use std::sync::Arc;
use densa_data::arrow_array::cast::AsArray;
use densa_data::arrow_array::{ArrayRef, Float64Array, Int64Array, RecordBatch, StringArray};
use densa_data::LookupJoin;

let trades = |ids: Vec<i64>, symbols: Vec<&str>, prices: Vec<f64>| {
    RecordBatch::try_from_iter([
        ("id", Arc::new(Int64Array::from(ids)) as ArrayRef),
        ("symbol", Arc::new(StringArray::from(symbols)) as ArrayRef),
        ("price", Arc::new(Float64Array::from(prices)) as ArrayRef),
    ])
};
let instruments = |symbols: Vec<&str>, names: Vec<&str>| {
    RecordBatch::try_from_iter([
        ("symbol", Arc::new(StringArray::from(symbols)) as ArrayRef),
        ("name", Arc::new(StringArray::from(names)) as ArrayRef),
    ])
};

// Trades keyed by id name their instrument in `symbol`; instruments are keyed by symbol.
let left = trades(vec![], vec![], vec![])?.schema();
let right = instruments(vec![], vec![])?.schema();
let mut join = LookupJoin::new(left, "id", "symbol", right, "symbol")?;
let mut joined = join.table()?; // id, symbol, price, name

join.upsert_right(&mut joined, instruments(vec!["AAA"], vec!["Alpha"])?)?;
join.upsert_left(&mut joined, trades(vec![1, 2], vec!["AAA", "BBB"], vec![10.0, 20.0])?)?;
// BBB is not an instrument yet: trade 2's name is null until it is.
join.upsert_right(&mut joined, instruments(vec!["BBB"], vec!["Beta"])?)?;

let names: Vec<String> = joined
    .snapshot()?
    .iter()
    .flat_map(|batch| batch.column(3).as_string::<i32>().iter().flatten().map(String::from).collect::<Vec<_>>())
    .collect();
assert_eq!(names.len(), 2);
assert!(names.contains(&"Alpha".to_owned()) && names.contains(&"Beta".to_owned()));

Implementations

impl LookupJoin

pub fn new( left: SchemaRef, left_key: &str, on: &str, right: SchemaRef, right_key: &str, ) -> Result<Self, DataError>

A join of rows of left’s columns (keyed by left_key) to the rows of a right table of right’s columns (keyed by right_key) whose key is in the left column on. The output has the left columns, then the right ones but the right key (its value is in on).

Errors

DataError::UnknownColumn for a column neither side has, DataError::UnsupportedKey for a key column of another type than integers or text, DataError::JoinKey if on does not hold the right key’s kind of value, DataError::DuplicateName if a left column and a right one share a name.

pub fn schema(&self) -> &SchemaRef

The output’s columns: the left ones, then the right ones but the right key; every one nullable.

pub fn table(&self) -> Result<LiveTable, DataError>

An empty output table, keyed by the left key: give it to every call of this join. Add views, groups or a window to it as to any table, or hand it to a grid.

Errors

None in practice: the columns were checked when the join was made.

pub fn right(&self) -> &LiveTable

The right rows, as the join holds them.

pub fn upsert_left( &mut self, out: &mut LiveTable, rows: RecordBatch, ) -> Result<Changes, DataError>

Joins rows (of the left columns) and upserts them into out, replacing the output rows of their keys. The changes are out’s.

Errors

A column error if rows has other columns than the left ones, DataError::JoinTable if out is not a table of this join’s output, and out’s own errors (DataError::NullKey, …). Nothing changes on error.

pub fn upsert_right( &mut self, out: &mut LiveTable, rows: RecordBatch, ) -> Result<Changes, DataError>

Upserts rows (of the right columns) into the right table, then joins again every row of out naming one of their keys. The changes are out’s.

Errors

The right table’s errors for rows (other columns, a null key, …) and DataError::JoinTable if out is not a table of this join’s output: nothing changes then. DataError::Arrow if Arrow cannot join the rows again (a dictionary column whose key type cannot index all the values joined) comes after the right table took rows: the output rows naming those keys keep their old right columns until the keys change again.

pub fn delete_right( &mut self, out: &mut LiveTable, keys: &dyn Array, ) -> Result<Changes, DataError>

Deletes the right rows of keys (an array of the right key’s kind), then joins again every row of out naming one of them: their right columns become null. The changes are out’s.

Errors

DataError::KeyType if keys is not of the right key’s kind and DataError::JoinTable if out is not a table of this join’s output: nothing changes then. DataError::Arrow as for upsert_right.

Trait Implementations

impl Debug for LookupJoin

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.