Method render_plan_expr

src/compute/src/render.rs:1148–1386

Renders a [`render_plan::Expr`], producing the collection of results. # Panics Panics if any of the expr's inputs is not found in `collections`. Callers must ensure that input nodes have been rendered previously.

(
        &self,
        expr: render_plan::Expr,
        collections: &BTreeMap<LirId, CollectionBundle<'scope, T>>,
    )

    /// Panics if any of the expr's inputs is not found in `collections`.
    /// Callers must ensure that input nodes have been rendered previously.
    fn render_plan_expr(
        &self,
        expr: render_plan::Expr,
        collections: &BTreeMap<LirId, CollectionBundle<'scope, T>>,
    ) -> CollectionBundle<'scope, T> {
        use render_plan::Expr::*;

        let expect_input = |id| {
            collections
                .get(&id)
                .cloned()
                .unwrap_or_else(|| panic!("missing input collection: {id}"))
        };

        match expr {
            Constant { rows } => {
                // Produce both rows and errs to avoid conditional dataflow construction.
                let (rows, errs) = match rows {
                    Ok(rows) => (rows, Vec::new()),
                    Err(e) => (Vec::new(), vec![e]),
                };

                // We should advance times in constant collections to start from `as_of`.
                let as_of_frontier = self.as_of_frontier.clone();
                let until = self.until.clone();
                let ok_collection = rows
                    .into_iter()
                    .filter_map(move |(row, mut time, diff)| {
                        time.advance_by(as_of_frontier.borrow());
                        if !until.less_equal(&time) {
                            Some((
                                row,
                                <T as Refines<mz_repr::Timestamp>>::to_inner(time),
                                diff,
                            ))
                        } else {
                            None
                        }
                    })
                    .to_stream(self.scope)
                    .as_collection();

                let mut error_time: mz_repr::Timestamp = Timestamp::minimum();
                error_time.advance_by(self.as_of_frontier.borrow());
                let err_collection = errs
                    .into_iter()
                    .map(move |e| {
                        (
                            DataflowErrorSer::from(e),
                            <T as Refines<mz_repr::Timestamp>>::to_inner(error_time),
                            Diff::ONE,
                        )
                    })
                    .to_stream(self.scope)
                    .as_collection();

                CollectionBundle::from_collections(ok_collection, err_collection)
            }