Rindle

The engine

Fold the delta stream yourself (Rust)

The low-level home — attach a change sink and the open engine hands you the raw delta stream. Maintain your own view from the diffs alone, and prove it equals a fresh query.

View as Markdown

This is the low-altitude path, on the open engine (rindle + rindle-sqlite). In the browser and Node the ArrayView folds the change stream into a materialized array for you. Here you do it yourself: instead of adding a View, you attach a change sink and the engine hands you the raw incremental CaughtChange stream. You decide what to do with the diffs — fold them into your own view, forward them over the wire, persist them.

This example does the first, then proves the payoff: a view maintained from the diff stream alone equals a fresh SELECT, with no re-querying. That equivalence — view-after-write == fresh-query — is the whole point of IVM. The complete program ships as a runnable example:

cargo run -p rindle-sqlite --example fold_deltas

Your own view, kept current from diffs

The consumer owns a BTreeMap keyed by primary key. Every delivered CaughtChange folds into it.

use std::collections::BTreeMap;
use serde_json::json;
use rindle::{CaughtChange, OwnedRow, Value};

const COLS: [&str; 4] = ["id", "title", "priority", "open"];

struct Consumer {
    view: BTreeMap<i64, serde_json::Value>, // id -> the row, as you'd render it
}

impl Consumer {
    fn apply(&mut self, changes: &[CaughtChange]) {
        for ch in changes {
            match ch {
                // A row entered the result set.
                CaughtChange::Add(n) => { self.view.insert(pk(&n.row), row_to_json(&n.row)); }
                // A row left it (deleted, or no longer matches the filter).
                CaughtChange::Remove(n) => { self.view.remove(&pk(&n.row)); }
                // A row that stayed but changed. The key can move, so drop old then insert new.
                CaughtChange::Edit { old, row } => {
                    self.view.remove(&pk(old));
                    self.view.insert(pk(row), row_to_json(row));
                }
                // Nested-relationship diffs — only a `sub_as(..)` query emits these.
                CaughtChange::Child { .. } => {}
            }
        }
    }
}

Cells are positional — in schema order — and read with row.col(c), which returns a borrowed Value (Int / Float / Bool / Str / Null). A number may be Int or Float, so match both when you extract a key; Str is raw bytes, so validate to &str at the edge:

fn pk(row: &OwnedRow) -> i64 {
    match row.col(0) {
        Value::Int(i) => i,
        Value::Float(f) => f as i64,
        other => panic!("non-numeric primary key: {other:?}"),
    }
}

fn row_to_json(row: &OwnedRow) -> serde_json::Value {
    let mut obj = serde_json::Map::new();
    for (name, cell) in COLS.iter().zip(row.cells()) {
        let v = match cell {
            Value::Int(i) => json!(i),
            Value::Float(f) if f.fract() == 0.0 => json!(f as i64),
            Value::Float(f) => json!(f),
            Value::Bool(b) => json!(b),
            Value::Str(s) => json!(std::str::from_utf8(s).unwrap()),
            _ => serde_json::Value::Null,
        };
        obj.insert((*name).to_string(), v);
    }
    serde_json::Value::Object(obj)
}

Drive it and prove it

Register the SQLite table as a source, build the query, and — instead of a View — add a change sink over the pipeline’s top with Graph::add_change_sink. hydrate_change_sink returns the cold-start snapshot (all Adds); after each push batch, take_sink_changes drains that transaction’s deltas.

use std::collections::HashMap;
use std::rc::Rc;
use rindle::{build_pipeline, owned_row, table, Graph, OwnedValue, SourceChange, SourceSchema, ValueType};
use rindle_sqlite::{ColumnDef, GraphTableSourceExt, TableSource};
use rusqlite::Connection;

let db = Rc::new(Connection::open_in_memory()?);
db.execute_batch(
    "CREATE TABLE issues (id INTEGER NOT NULL, title TEXT NOT NULL,
                          priority INTEGER NOT NULL, open BOOLEAN NOT NULL);
     CREATE UNIQUE INDEX issues_pk ON issues (id);",
)?;

let columns = vec![
    ColumnDef { name: "id".into(),       ty: ValueType::Number,  optional: false },
    ColumnDef { name: "title".into(),    ty: ValueType::String,  optional: false },
    ColumnDef { name: "priority".into(), ty: ValueType::Number,  optional: false },
    ColumnDef { name: "open".into(),     ty: ValueType::Boolean, optional: false },
];
let schema = SourceSchema::new(vec!["id", "title", "priority", "open"], vec![0], vec![(0, true)]);

let mut graph = Graph::new();
let issues_src = graph.add_table_source(
    TableSource::new_with_schema(db.clone(), "issues", columns, vec![0], schema.clone()),
);
let sources = HashMap::from([("issues", (issues_src, schema))]);
let resolve = |t: &str| sources.get(t).cloned();

// The live query: the OPEN issues, maintained incrementally.
let ast = table("issues").r#where("open", true).build();
let top = build_pipeline(&mut graph, &ast, &resolve).expect("build");

// A change sink hands you the raw delta stream (instead of a materialized View).
let sink = graph.add_change_sink(top);
graph.set_sink_edge(top, sink);

let mut consumer = Consumer { view: BTreeMap::new() };
consumer.apply(&graph.hydrate_change_sink(sink)); // fires once: the cold-start snapshot

Push ordinary changes through the write-through source — each source_push writes SQLite and emits the delta into the sink — then drain and fold:

// Insert three issues (one already closed).
graph.try_source_push(issues_src, SourceChange::Add(owned_row(vec![
    OwnedValue::Int(1), OwnedValue::str("login button misaligned"), OwnedValue::Int(2), OwnedValue::Bool(true),
])))?;
graph.try_source_push(issues_src, SourceChange::Add(owned_row(vec![
    OwnedValue::Int(2), OwnedValue::str("slow dashboard query"), OwnedValue::Int(1), OwnedValue::Bool(true),
])))?;
graph.try_source_push(issues_src, SourceChange::Add(owned_row(vec![
    OwnedValue::Int(3), OwnedValue::str("typo in footer"), OwnedValue::Int(3), OwnedValue::Bool(false), // closed → never enters
])))?;
consumer.apply(&graph.take_sink_changes(sink));

// Close issue 1 — the consumer sees a Remove for #1, no rescan.
graph.try_source_push(issues_src, SourceChange::Edit {
    old: owned_row(vec![OwnedValue::Int(1), OwnedValue::str("login button misaligned"), OwnedValue::Int(2), OwnedValue::Bool(true)]),
    row: owned_row(vec![OwnedValue::Int(1), OwnedValue::str("login button misaligned"), OwnedValue::Int(2), OwnedValue::Bool(false)]),
})?;
consumer.apply(&graph.take_sink_changes(sink));

Finally, the payoff — the delta-folded view must equal a fresh query over the same SQLite database the write-through pushes populated:

let mut stmt = db.prepare(
    "SELECT id, title, priority, open FROM issues WHERE open = 1 ORDER BY id",
)?;
let fresh: Vec<serde_json::Value> = stmt
    .query_map([], |r| Ok(json!({
        "id": r.get::<_, i64>(0)?, "title": r.get::<_, String>(1)?,
        "priority": r.get::<_, i64>(2)?, "open": r.get::<_, bool>(3)?,
    })))?
    .collect::<rusqlite::Result<Vec<_>>>()?;

let folded: Vec<serde_json::Value> = consumer.view.values().cloned().collect();
assert_eq!(folded, fresh, "the delta-folded view must equal a fresh query — the IVM guarantee");

The view we maintained from diffs alone is equal, row for row, to a fresh SELECT — and we never rescanned SQLite to keep it that way.

Why this layer

You drop to the raw change sink when you want the engine without the opinions: the incremental delta stream to route, fold, or persist however you like. It is exactly the stream the higher tiers are built on — the JS ArrayView folds it into an array, and the live-replica runtime re-exports CaughtChange as its ChangeEvent and derives it from ordinary SQL writes for you.

The open rindle-sqlite examples exercise this same TableSource + source_push path end to end — e.g. cargo run -p rindle-sqlite --example bench_zero_throughput_relational_rs.

Next steps