Method log_dataflow_errors

src/compute/src/logging/compute.rs:1437–1450

(self, logger: Logger, export_id: GlobalId)

    D: Clone + 'static,
{
    fn log_dataflow_errors(self, logger: Logger, export_id: GlobalId) -> Self {
        self.inner
            .unary(Pipeline, "LogDataflowErrorsCollection", |_cap, _info| {
                move |input, output| {
                    input.for_each(|cap, data| {
                        let diff = data.iter().map(|(_d, _t, r)| *r).sum::<Diff>();
                        logger.log(&ComputeEvent::ErrorCount(ErrorCount { export_id, diff }));

                        output.session(&cap).give_container(data);
                    });
                }
            })
            .as_collection()
    }
}

impl<'scope, T, B> LogDataflowErrors for StreamVec<'scope, T, B>