Function optimize_dataflow_monotonic

src/transform/src/dataflow.rs:421–454

(
    dataflow: &mut DataflowDesc,
    ctx: &mut TransformCtx,
)

    fields(path.segment ="monotonic")
)]
pub fn optimize_dataflow_monotonic(
    dataflow: &mut DataflowDesc,
    ctx: &mut TransformCtx,
) -> Result<(), TransformError> {
    let mut monotonic_ids = BTreeSet::new();
    for (source_id, source_import) in dataflow.source_imports.iter() {
        if source_import.monotonic {
            monotonic_ids.insert(source_id.clone());
        }
    }
    for (
        _index_id,
        IndexImport {
            desc: index_desc,
            monotonic,
            ..
        },
    ) in dataflow.index_imports.iter()
    {
        if *monotonic {
            monotonic_ids.insert(index_desc.on_id.clone());
        }
    }

    let monotonic_flag = MonotonicFlag::default();

    for build_desc in dataflow.objects_to_build.iter_mut() {
        monotonic_flag.transform(build_desc.plan.as_inner_mut(), ctx, &monotonic_ids)?;
    }

    mz_repr::explain::trace_plan(dataflow);

    Ok(())
}

/// Determine whether we require snapshots from our durable source imports.
/// (For example, these can often be skipped for simple subscribe queries.)