# 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. This is the low-altitude path, on the **open** engine (`rindle` + `rindle-sqlite`). In the browser and Node the [`ArrayView`](/docs/wasm-client) 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`](/docs/change-model) 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: ```bash 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. ```rust 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, // 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: ```rust 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 `Add`s); after each push batch, `take_sink_changes` drains that transaction's deltas. ```rust 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: ```rust // 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: ```rust let mut stmt = db.prepare( "SELECT id, title, priority, open FROM issues WHERE open = 1 ORDER BY id", )?; let fresh: Vec = 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::>>()?; let folded: Vec = 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`](/docs/wasm-client) folds it into an array, and the [live-replica runtime](/docs/replica-and-views) re-exports `CaughtChange` as its `ChangeEvent` and derives it from ordinary SQL writes for you. The open [`rindle-sqlite` examples](https://github.com/rindle-sh/rindle) 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 - [Rust quickstart](/docs/quickstart) — the same engine with a materialized `View` instead of a raw sink. - [The change model](/docs/change-model) — `Add` / `Remove` / `Edit` / `Child` and the replay-equivalence invariant. - [The live-replica runtime](/docs/replica-and-views) — the same deltas, derived from SQL writes and scaled across readers. --- [View this page on Rindle](https://rindle.sh/docs/example-rust)