pub fn as_collection_core(
        &self,
        mut mfp: MapFilterProject,
        key_val: Option<(Vec<MirScalarExpr>, Row)>,
    ) -> (
        Collection<S, repr::Row, Diff>,
        Collection<S, DataflowError, Diff>,
    ) {
        if mfp.is_identity() {
            self.as_collection()
        } else {
            mfp.optimize();
            let mut mfp_plan = mfp.into_plan().unwrap();
            let (stream, errors) = self.flat_map(key_val, |permutation| {
                let mut row_builder = Row::default();
                let mut datum_vec = DatumVec::new();

                if let Some(permutation) = &permutation {
                    permutation.permute_mfp_plan(&mut mfp_plan);
                }
                move |row_parts, time, diff| {
                    let temp_storage = RowArena::new();
                    let mut datums_local = datum_vec.borrow_with_many(row_parts);
                    mfp_plan.evaluate(
                        &mut datums_local,
                        &temp_storage,
                        time.clone(),
                        diff.clone(),
                        &mut row_builder,
                    )
                }
            });

            use timely::dataflow::operators::ok_err::OkErr;
            let (oks, errs) = stream.ok_err(|x| x);

            use differential_dataflow::AsCollection;
            let oks = oks.as_collection();
            let errs = errs.as_collection();
            (oks, errors.concat(&errs))
        }
    }
/// Presents `self` as a stream of updates, having been subjected to `mfp`.
///
/// This operator is able to apply the logic of `mfp` early, which can substantially
/// reduce the amount of data produced when `mfp` is non-trivial.
///
/// The `key_val` argument, when present, indicates that a specific arrangement should
/// be used, and that we can seek to the supplied row.
Presents `self` as a stream of updates, having been subjected to `mfp`.
This operator is able to apply the logic of `mfp` early, which can substantially
reduce the amount of data produced when `mfp` is non-trivial.

The `key_val` argument, when present, indicates that a specific arrangement should
be used, and that we can seek to the supplied row.
Presents ` self ` as a stream of updates having been subjected to ` mfp ` . This operator is able to apply the logic of ` mfp ` early which can substantially reduce the amount of data produced when ` mfp ` is non - trivial . The ` key_val ` argument when present indicates that a specific arrangement should be used and that we can seek to the supplied row .
pub fn as_collection_core(
        &self,
        mut mfp: MapFilterProject,
        key_val: Option<(Vec<MirScalarExpr>, Row)>,
    ) -> (
        Collection<S, repr::Row, Diff>,
        Collection<S, DataflowError, Diff>,
    ) {
        if mfp.is_identity() {
            self.as_collection()
        } else {
            mfp.optimize();
            let mut mfp_plan = mfp.into_plan().unwrap();
            let (stream, errors) = self.flat_map(key_val, |permutation| {
                let mut row_builder = Row::default();
                let mut datum_vec = DatumVec::new();
                if let Some(permutation) = &permutation {
                    permutation.permute_mfp_plan(&mut mfp_plan);
                }
                move |row_parts, time, diff| {
                    let temp_storage = RowArena::new();
                    let mut datums_local = datum_vec.borrow_with_many(row_parts);
                    mfp_plan.evaluate(
                        &mut datums_local,
                        &temp_storage,
                        time.clone(),
                        diff.clone(),
                        &mut row_builder,
                    )
                }
            });
            use timely::dataflow::operators::ok_err::OkErr;
            let (oks, errs) = stream.ok_err(|x| x);
            use differential_dataflow::AsCollection;
            let oks = oks.as_collection();
            let errs = errs.as_collection();
            (oks, errors.concat(&errs))
        }
    }
pub fn as_collection_core ( & self , mut mfp : MapFilterProject , key_val : Option < ( Vec < MirScalarExpr > , Row ) > , ) -> ( Collection < S , repr :: Row , Diff > , Collection < S , DataflowError , Diff > , ) { if mfp . is_identity ( ) { self . as_collection ( ) } else { mfp . optimize ( ) ; let mut mfp_plan = mfp . into_plan ( ) . unwrap ( ) ; let ( stream , errors ) = self . flat_map ( key_val , | permutation | { let mut row_builder = Row :: default ( ) ; let mut datum_vec = DatumVec :: new ( ) ; if let Some ( permutation ) = & permutation { permutation . permute_mfp_plan ( & mut mfp_plan ) ; } move | row_parts , time , diff | { let temp_storage = RowArena :: new ( ) ; let mut datums_local = datum_vec . borrow_with_many ( row_parts ) ; mfp_plan . evaluate ( & mut datums_local , & temp_storage , time . clone ( ) , diff . clone ( ) , & mut row_builder , ) } } ) ; use timely :: dataflow :: operators :: ok_err :: OkErr ; let ( oks , errs ) = stream . ok_err ( | x | x ) ; use differential_dataflow :: AsCollection ; let oks = oks . as_collection ( ) ; let errs = errs . as_collection ( ) ; ( oks , errors . concat ( & errs ) ) } }
Presents `self` as a stream of updates, having been subjected to `mfp`.
Presents ` self ` as a stream of updates having been subjected to ` mfp ` .
&self
mfp
MapFilterProject
key_val
Option<(Vec<MirScalarExpr>, Row)>
&self
mfp
MapFilterProject
key_val
Option<(Vec<MirScalarExpr>, Row)>