Function build_compute_dataflow

src/compute/src/render.rs:198–553

Assemble the "compute" side of a dataflow, i.e. all but the sources. This method imports sources from provided assets, and then builds the remaining dataflow using "compute-local" assets like shared arrangements, and producing both arrangements and sinks.

(
    timely_worker: &mut TimelyWorker,
    compute_state: &mut ComputeState,
    dataflow: DataflowDescription<RenderPlan, CollectionMetadata>,
    start_signal: StartSignal,
    until: Antichain<mz_

/// dataflow using "compute-local" assets like shared arrangements, and producing
/// both arrangements and sinks.
pub fn build_compute_dataflow(
    timely_worker: &mut TimelyWorker,
    compute_state: &mut ComputeState,
    dataflow: DataflowDescription<RenderPlan, CollectionMetadata>,
    start_signal: StartSignal,
    until: Antichain<mz_repr::Timestamp>,
    dataflow_expiration: Antichain<mz_repr::Timestamp>,
) {
    // Mutually recursive view definitions require special handling.
    let recursive = dataflow
        .objects_to_build
        .iter()
        .any(|object| object.plan.is_recursive());

    // Determine indexes to export, and their dependencies.
    let indexes = dataflow
        .index_exports
        .iter()
        .map(|(idx_id, (idx, _typ))| (*idx_id, dataflow.depends_on(idx.on_id), idx.clone()))
        .collect::<Vec<_>>();

    // Determine sinks to export, and their dependencies.
    let sinks = dataflow
        .sink_exports
        .iter()
        .map(|(sink_id, sink)| (*sink_id, dataflow.depends_on(sink.from), sink.clone()))
        .collect::<Vec<_>>();

    let worker_logging = timely_worker.logger_for("timely").map(Into::into);
    let apply_demands = COMPUTE_APPLY_COLUMN_DEMANDS.get(&compute_state.worker_config);
    let subscribe_snapshot_optimization =
        SUBSCRIBE_SNAPSHOT_OPTIMIZATION.get(&compute_state.worker_config);

    let name = format!("Dataflow: {}", &dataflow.debug_name);
    let input_name = format!("InputRegion: {}", &dataflow.debug_name);
    let build_name = format!("BuildRegion: {}", &dataflow.debug_name);

    timely_worker.dataflow_core(&name, worker_logging, Box::new(()), |_, scope| {
        let scope = scope.with_label();

        // The scope.clone() occurs to allow import in the region.
        // We build a region here to establish a pattern of a scope inside the dataflow,
        // so that other similar uses (e.g. with iterative scopes) do not require weird
        // alternate type signatures.
        let mut imported_sources = Vec::new();
        let mut tokens: BTreeMap<_, Rc<dyn Any>> = BTreeMap::new();
        let output_probe = MzProbeHandle::default();

        scope.clone().region_named(&input_name, |region| {
            // Import declared sources into the rendering context.
            for (source_id, import) in dataflow.source_imports.iter() {
                region.region_named(&format!("Source({:?})", source_id), |inner| {
                    let mut read_schema = None;
                    let mut mfp = import.desc.arguments.operators.clone().map(|mut ops| {
                        // If enabled, we read from Persist with a `RelationDesc` that
                        // omits uneeded columns.
                        if apply_demands {
                            let demands = ops.demand();