Struct DataflowBuilder 

pub struct DataflowBuilder { /* private fields */ }

Builds a compute dataflow from generic parts, hiding the lowering and persist wiring mechanism.

Contract

By default the caller supplies MIR and the builder lowers it faithfully, attaching the persist wiring without optimizing — so a hand-built minimal plan lowers exactly as written. Optimization — fusion, predicate pushdown, and notably join-implementation selection — is opt-in via Self::optimize, paid for only by callers that need it. A Join whose implementation is left Unimplemented is rejected by the LIR lowering, so a plan containing one requires optimize, which runs mz_transform::optimize_dataflow to fill the implementation first. When optimizing, the builder hands the optimizer an index oracle built from its own index_imports (ImportedIndexOracle), so imported arrangements are recognized — the same index information environmentd’s catalog oracle would supply for these imports.

Construction strategy

The builder deliberately does not hand-roll the RenderPlan: the LirIds used to stitch nodes together have no public constructor, and the LetFreePlan invariants (notably a valid topological_order) are easy to get wrong. Instead it mirrors exactly what the real compute controller does:

1. Accumulate a MIR-level DataflowDescription<OptimizedMirRelationExpr, ()> using the same import_source / insert_plan / export_index helpers the optimizer uses.
2. Lower it to LIR via Plan::finalize_dataflow, yielding DataflowDescription<Plan, ()>.
3. Augment it into DataflowDescription<RenderPlan, CollectionMetadata> by converting each object’s Plan via RenderPlan::try_from and attaching the storage CollectionMetadata to each source import — the same step performed in compute-client’s Instance::create_dataflow.

This guarantees the emitted plan is structurally identical to one produced by a live environmentd, at the cost of running the (cheap, deterministic) lowering in-process.

Implementations

impl DataflowBuilder

pub fn new(name: impl Into<String>) -> Self

Start an empty builder. name becomes the dataflow’s debug name.

pub fn import_persist(&mut self, id: GlobalId, source: PersistSource) -> Input

Import a persist-backed storage collection as id.

Registers the source on the MIR description and records the persist metadata for the augment step. Returns an Input handle whose Input::get yields a correctly typed Get node, so callers never construct a ReprRelationType by hand.

pub fn import_index( &mut self, index_id: GlobalId, on_id: GlobalId, key_cols: Vec<usize>, on_type: ReprRelationType, monotonic: bool, ) -> Input

Import a previously-exported index, making the collection it arranges (on_id) available to this dataflow as an in-memory arrangement.

Unlike Self::import_persist, this imports no storage collection: the arrangement is served from the replica’s existing, hydrated index, so the dataflow needs no CollectionMetadata and the augment step leaves the index import untouched. The MIR-to-LIR lowering registers the imported arrangement under Get(on_id) automatically, so a faithful (unoptimized) Get(on_id) picks it up. Returns an Input referencing on_id — the id a computation Gets, not the index id itself.

pub fn get(&self, id: GlobalId) -> Result<MirRelationExpr>

A typed Get of an already-imported or built id, for callers that assemble MIR by id rather than threading Input handles — notably the JSON MIR translator. Errors if id was never imported or built, so a bad reference surfaces cleanly instead of constructing an ill-typed Get.

pub fn build(&mut self, id: GlobalId, expr: MirRelationExpr) -> &mut Self

Insert a MIR object to compute, bound to id.

expr is wrapped via OptimizedMirRelationExpr::declare_optimized; the caller is responsible for any optimization (see the type-level contract). The object’s relation type is recorded so a later Self::export_index over id can derive its on_type.

pub fn export_index( &mut self, index_id: GlobalId, on_id: GlobalId, key_cols: Vec<usize>, ) -> &mut Self

Export an index index_id arranging on_id by key_cols.

on_id may be an imported source or a built object; either way the lowering synthesizes the ArrangeBy. The on_type is derived from the referenced id, which must have been imported or built first.

pub fn export_materialized_view( &mut self, sink_id: GlobalId, from_id: GlobalId, value_desc: RelationDesc, target: PersistSink, ) -> &mut Self

Export a materialized-view persist sink sink_id writing the collection from_id to a target persist shard (a materialized view).

value_desc is the output relation schema; it must match from_id’s type (validated by the caller). The target shard is identified by target, whose CollectionMetadata the augment step splices into the sink connection — the compute persist sink opens it as SourceData/()/Timestamp/StorageDiff, the same codec a storage collection uses, so the shard reads back like any other.

up_to is always the empty antichain: the persist sink does not implement UP TO (it panics during rendering otherwise), and the real optimizer likewise leaves a materialized view’s up_to empty — it is a subscribe-only concept.

pub fn export_subscribe( &mut self, sink_id: GlobalId, from_id: GlobalId, value_desc: RelationDesc, up_to: Antichain<Timestamp>, ) -> &mut Self

Export a subscribe sink sink_id streaming changes of the collection from_id back as ComputeResponse::SubscribeResponse batches.

Unlike a materialized view, a subscribe writes no shard, so it needs no storage metadata. value_desc is the output schema (must match from_id’s type); up_to is the exclusive upper at which the subscribe completes. The empty output ordering leaves intra-timestamp order unconstrained — the driver consolidates and sorts the updates for a deterministic golden.

pub fn as_of(&mut self, t: Timestamp) -> &mut Self

Set the dataflow’s as_of (the read frontier hydration starts from).

pub fn until(&mut self, t: Timestamp) -> &mut Self

Set the dataflow’s until (the exclusive upper bound past which output is dropped). Defaults to the empty antichain (no bound).

pub fn optimize(&mut self) -> &mut Self

Run the MIR dataflow optimizer in Self::finish before lowering.

Off by default: the builder otherwise lowers the caller’s MIR faithfully (the contract above). Enable it for plans that don’t lower from raw MIR — notably a Join, whose implementation defaults to Unimplemented and is rejected by the LIR lowering until mz_transform::optimize_dataflow’s JoinImplementation fills it in — or to reproduce the plan environmentd would ship for a logical expression rather than the literal one written.

pub fn finish( self, ) -> Result<DataflowDescription<RenderPlan, CollectionMetadata>>

Lower the accumulated MIR and attach persist wiring, producing the DataflowDescription the compute protocol expects.

Returns an error rather than panicking on a malformed plan (e.g. a key column out of range, or an unbalanced object graph), so a caller driving this from external input — notably the script reader — can surface a clean error instead of crashing the process.