Struct Engine 

pub struct Engine { /* private fields */ }

The per-thread IVM unit (see the module docs). Internal to the Db/Cluster openers, and public via the crate’s embedded-engine seam (design 308): a host that owns its own write path — the SQLite extension’s commit hook — constructs one over a derivation connection, registers tables/queries, and drives apply_batch itself.

Implementations

impl Engine

pub fn new( worker: Rc<Connection>, plan_queries: bool, operator_storage: OperatorStorage, ) -> Result<Engine, ReplicaError>

pub fn has_source(&self, table: &str) -> bool

pub fn register_table( &mut self, table: &str, ts: TableSchema, ) -> Result<(), ReplicaError>

Build the shared TableSource for table on the worker connection, add it to the graph, and remember its Schema. Idempotent.

pub fn register_query( &mut self, query_id: QueryId, ast: &Ast, ) -> Result<(NodeId, Vec<CaughtChange>), ReplicaError>

Lower ast into the shared graph, terminate it in a fresh change-stream sink, and hydrate it. query_id is the caller’s opaque tag (the engine stores it for correlation but never interprets it — de-dup is the caller’s concern). Returns the sink’s NodeId and the initial change set (the hydration Adds). BuildError (unknown table/column, unsupported shape) is surfaced synchronously.

pub fn register_family( &mut self, query_id: QueryId, stripped: &Ast, params: &[Box<str>], ) -> Result<NodeId, ReplicaError>

Register a parameterized query family (design 310 §4 / impl plan §6.1) under query_id: one pipeline over the family’s stripped template whose params columns are a partition dimension. Mirrors register_query — recording, the panic boundary, scalar-subquery resolution and planning on the template — but builds through build_family_pipeline and does not hydrate: partitions hydrate individually as they are bound (bind_partition). Returns the change-sink id.

pub fn bind_partition( &self, query_id: QueryId, binding: &Binding, ) -> Result<Option<Vec<CaughtChange>>, ReplicaError>

Bind one partition of the family registered under query_id and hydrate only it (design 310 §4.4): the partition’s initial Add set. Ok(None) if query_id is not a registered family; Err if the binding is already bound or the hydrate fails (the binding is then rolled back).

pub fn unbind_partition( &self, query_id: QueryId, binding: &Binding, ) -> Result<bool, ReplicaError>

Unbind one partition of the family registered under query_id (design 310 §4.4, impl plan D5). Ok(false) if query_id is not a registered family or the binding was not bound.

pub fn analyze_query(&self, ast: &Ast) -> Result<AnalyzeReport, ReplicaError>

Run ast cold on a throwaway, instrumented pipeline over a pinned read snapshot and report where its time and rows go (ANALYZE-QUERY-DESIGN.md): the chosen plan, the per-phase timing, and — per source — the rows it emitted into the pipeline beside the SQLite scan work it cost.

Read-only and isolated. It builds and drops a fresh [Graph] with fresh sources (forks over the worker connection), so it never registers a materialization, never dedups on the manager’s QueryKey, never mutates, and never touches self.graph / self.sources / the live materialization set. Every leaf fetch runs inside one deferred read transaction (BEGIN … ROLLBACK) so the whole run sees one consistent snapshot; under wal2 that reader never blocks the writer.

It runs the same planner gate as register_query (plan_queries && has_flippable_exists), so the plan it reports is the one a fresh registration would choose — the “current-stats plan,” which for a long-lived materialization whose stats have drifted can legitimately differ from the frozen plan actually running (§3.1). BuildError (unknown table/column, unsupported shape) surfaces synchronously; an internal-invariant panic (e.g. a cost-model assert — planning runs before build_pipeline validates table names) is contained and surfaced as an error, exactly like register_query.

pub fn view_schema(&self, ast: &Ast) -> Result<Schema, ReplicaError>

The hierarchical view [Schema] ast materializes to — derived from the registered sources exactly as register_query resolves them (so the shape, sort, singular flag, and in-view relationships line up with the change stream). Protocol-agnostic; the flat-change / remote layer turns it into a wire schema + fingerprint. BuildError (unknown table/column) surfaces as Err.

pub fn deregister_query(&mut self, sink: NodeId) -> bool

Tear down a registered query: drop its bookkeeping and reclaim its pipeline (disconnect from the shared sources + free its operator/storage slots). Returns false if sink was not a registered query (already gone). Idempotent and safe against a stale sink id: a non-matching key is simply absent from queries.

pub fn deregister_by_query_id(&mut self, query_id: QueryId) -> usize

Tear down every query registered under query_id and reclaim its pipeline. The parallel Cluster shards one query_id to one worker and tears down by the stable query_id (a sink NodeId does not survive a fault rebuild); duplicates registered under the same id are all removed. Returns the count reclaimed (0 if none — idempotent / unknown id).

pub fn query_id_of(&self, sink: NodeId) -> Option<QueryId>

The caller’s QueryId for a registered query’s sink, or None if sink is not a live query (used to tag derived deltas for channel-out delivery).

pub fn query_ids(&self) -> Vec<QueryId>

The QueryIds of every query currently registered on this engine (used by the parallel runtime to notify subscribers when the whole engine is torn down and rebuilt after a derivation fault).

pub fn read_snapshot( &self, query_id: QueryId, ) -> Result<Option<Vec<CaughtChange>>, ReplicaError>

Fetch query_id through its pipeline as a fresh set of hydration Adds. This traverses current sources rather than reading a cached change-stream fold. try_hydrate_change_sink leaves the push buffer unchanged, so it is safe to call on an already-hydrated query between commits. Ok(None) if query_id is not registered on this engine; Err when the read boundary raises (a parked leaf error, or the §5.3 sum-overflow state a prior — itself typed-errored — write left out of range: reads over unrepresentable state error every time, never render a NULL aggregate as data).

pub fn plan_queries(&self) -> bool

Whether opt-in cost-based query planning is enabled (so a post-fault rebuild can inherit the setting).

pub fn operator_storage(&self) -> OperatorStorage

The operator-scratch-state backend this engine’s graph uses (so a post-fault rebuild re-creates the graph with the same backend).

pub fn set_max_delta_bytes(&self, max_bytes: usize)

Set the D4 derivation memory budget — the estimated bytes of folded rows ONE transaction may hold across every table it touches. Every registered delta already shares this budget, so the new ceiling is live immediately; tables registered later — including by a shed/fault rebuild — inherit it.

pub fn max_delta_bytes(&self) -> usize

The engine’s current D4 budget ceiling (so a rebuild can carry it over).

pub fn set_join_precheck_bounds( &self, per_join: Option<usize>, per_graph: usize, )

Set the join membership pre-check bounds on this engine’s graph (designs/311-JOIN-MEMBERSHIP-PRECHECK-DESIGN.md §8): the per-join distinct-key bound (None = off, the default) and the per-graph key budget. A change resets every join’s set, which then rebuilds by observing its next hydrate-shaped fetch.

pub fn join_precheck_bounds(&self) -> JoinPrecheckBounds

The engine’s current pre-check bounds (so a rebuild can carry them over).

pub fn storage_report( &self, query_id: QueryId, dump: bool, ) -> Option<StorageReport>

The operator scratch storage held by the pipeline(s) registered under query_id — the entry count always, and every (slot, key, value) when dump. None if nothing is registered under that id. Inspection-only (a full scan per storage slot); the leak probe of design 310 impl plan D5 — after unbinding every partition of a family, its pipeline’s scratch state must be back at the count it held when the bindings were all bound minus their slots, not carrying zombie partitions.

pub fn join_precheck_report(&self) -> JoinPrecheckReport

A read-only report of the join membership pre-check on this engine: the bounds in force, the graph-wide tracked-key charge, and — per registered query, sorted by QueryId — the state of every join its pipeline owns. The inspection hook the schedule tests use to compare a live worker against a fresh graph (design 311 §14.1: the knob used to land only at the next commit barrier, after the hydrate that builds the sets, which no lane could see).

pub fn worker_conn(&self) -> Rc<Connection>

The connection this engine derives against, as the shared handle a shed/fault rebuild needs to construct the replacement engine.

pub fn conn(&self) -> &Connection

The connection this engine derives against (for ad-hoc reads/probes).

pub fn set_push_deadline(&self, deadline: Option<Instant>)

Arm (or clear) the wall-clock deadline for the next push (FOLLOWER-LAG-SHED §6.6 — the runaway-push bail). The worker brackets each apply_and_drain with it; the graph’s fan-out checkpoints park a PushDeadlineExceeded on expiry.

pub fn open_snapshot(&self) -> Result<(), ReplicaError>

Phase 1. Open the derivation snapshot: a transaction plus a forcing read that PINS the read snapshot at the current (pre-commit = T-1) state. A WAL read snapshot is established at the first read, not at BEGIN, so the pin must happen here — before the writer commits the transaction being derived — or a read during apply_and_drain (after the writer commits) could slide forward to the post-commit state and double-count. Reads sqlite_schema, so it needs no registered table.

The transaction is a plain deferred BEGIN — the derivation never writes; batch state lives in each source’s [BatchDelta], activated here (design 306).

Self-healing: nothing but this method opens a transaction on the worker connection, so a BEGIN that fails with one already open can only mean an earlier derivation leaked its snapshot. Close it and retry once rather than leaving the connection pinned at that stale WAL snapshot for the life of the process (every later open_snapshot failing the same way, the shard serving ever-older hydrations, and the WAL never checkpointing past the held read mark).

pub fn apply_and_drain( &self, batch: &[Captured], ) -> Result<Vec<(NodeId, Vec<CaughtChange>)>, ReplicaError>

Phase 2. Push every hosted change in batch through the shared sources — fanning each out to every dependent query — then drain each affected sink. A change for a table this engine does not host is skipped (another worker hosts it; the single-thread engine hosts every captured table, so the skip is unreachable there). Returns (sink, events) per changed query. On error, partial sink buffers are discarded so a failed tx can’t leak into the next. The snapshot must be open (Phase 1).

The whole push+drain is run under a panic boundary so an internal-invariant panic (an engine bug, never valid input) becomes an Err rather than unwinding the worker thread: the per-push source_push_isolated already catches the push fan-out, and this outer guard additionally covers the drain loop. An Err here routes the worker to its in-place tear-down-on-fault (discard + rebuild the engine) — keeping the worker thread alive instead of escalating to a full thread death + respawn. Effective under panic = "unwind".

pub fn push(&self, cap: &Captured) -> Result<(), ReplicaError>

Apply one captured change to the shared sources, fanning it out to every dependent query. A change for a table this engine does not host is skipped (another worker hosts it; the single-thread engine hosts every captured table, so the skip is unreachable there). Does not drain — the deltas this change produces accumulate in the affected sinks until drain. The snapshot must be open (Phase 1).

CONTRACT: a SourceChange::Edit is pk-stable — the primary key is row identity, never mutated in place. The CDC captures an arbitrary SQL statement, so an UPDATE … SET <pk> = … (legal, if pathological) arrives here as a single Edit whose old/new pks differ. Normalize that ONE case into Remove(old)+Add(new) at this single ingress, so every consumer — especially the pk-keyed NormalizeFold, which would otherwise silently drop it — sees the honest row-identity change (adversarial-review #13). All other producers honor the contract by construction. The batch is shared (Arc) across workers; cloning a SourceChange is a refcount bump, not a row copy. source_push_isolated catch_unwinds the operator-graph fan-out + drains parked errors, so an internal-invariant panic surfaces as Err.

pub fn drain(&self) -> Vec<(NodeId, Vec<CaughtChange>)>

Take whatever deltas are pending in the sinks: drain every registered query’s change-sink and return (sink, events) for each that produced output (empty sinks are skipped). Called once after a batch of pushes in the batch model; per-change in the streaming model (where each push is followed immediately by a drain so the worker holds nothing across a transaction).

pub fn rollback(&self)

Phase 3. Discard the derivation state (the writer’s COMMIT is the durable copy): clear every table’s [BatchDelta], then close the snapshot transaction. Safe to call once the snapshot was opened (Phase 1).

pub fn apply_batch( &self, captured: Vec<Captured>, ) -> Result<Vec<(NodeId, Vec<CaughtChange>)>, ReplicaError>

Single-thread convenience (the Db path): the three phases back-to-back on one thread — open the snapshot, derive, roll back. Returns (sink, events) per changed query.