pub fn render_plan(
        &mut self,
        plan: plan::Plan,
        scope: &mut G,
        worker_index: usize,
    ) -> CollectionBundle<G, Row, G::Timestamp> {
        match plan {
            Plan::Constant { rows } => {
                // Produce both rows and errs to avoid conditional dataflow construction.
                let (mut 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`.
                use differential_dataflow::lattice::Lattice;
                for (_, time, _) in rows.iter_mut() {
                    time.advance_by(self.as_of_frontier.borrow());
                }
                let mut error_time: G::Timestamp = timely::progress::Timestamp::minimum();
                error_time.advance_by(self.as_of_frontier.borrow());

                let ok_collection = rows.into_iter().to_stream(scope).as_collection();

                let err_collection = errs
                    .into_iter()
                    .map(move |e| (DataflowError::from(e), error_time, 1))
                    .to_stream(scope)
                    .as_collection();

                CollectionBundle::from_collections(ok_collection, err_collection)
            }
            Plan::Get {
                id,
                keys,
                mfp,
                key_val,
            } => {
                // Recover the collection from `self` and then apply `mfp` to it.
                // If `mfp` happens to be trivial, we can just return the collection.
                let mut collection = self
                    .lookup_id(id)
                    .unwrap_or_else(|| panic!("Get({:?}) not found at render time", id));
                if mfp.is_identity() {
                    // Assert that each of `keys` are present in `collection`.
                    assert!(keys.iter().all(|key| collection.arranged.contains_key(key)));
                    // Retain only those keys we want to import.
                    collection.arranged.retain(|key, _value| keys.contains(key));
                    collection
                } else {
                    let (oks, errs) = collection.as_collection_core(mfp, key_val);
                    CollectionBundle::from_collections(oks, errs)
                }
            }
            Plan::Let { id, value, body } => {
                // Render `value` and bind it to `id`. Complain if this shadows an id.
                let value = self.render_plan(*value, scope, worker_index);
                let prebound = self.insert_id(Id::Local(id), value);
                assert!(prebound.is_none());

                let body = self.render_plan(*body, scope, worker_index);
                self.remove_id(Id::Local(id));
                body
            }
            Plan::Mfp {
                input,
                mfp,
                key_val,
            } => {
                // If `mfp` is non-trivial, we should apply it and produce a collection.
                let input = self.render_plan(*input, scope, worker_index);
                if mfp.is_identity() {
                    input
                } else {
                    let (oks, errs) = input.as_collection_core(mfp, key_val);
                    CollectionBundle::from_collections(oks, errs)
                }
            }
            Plan::FlatMap {
                input,
                func,
                exprs,
                mfp,
            } => {
                let input = self.render_plan(*input, scope, worker_index);
                self.render_flat_map(input, func, exprs, mfp)
            }
            Plan::Join { inputs, plan } => {
                let inputs = inputs
                    .into_iter()
                    .map(|input| self.render_plan(input, scope, worker_index))
                    .collect();
                match plan {
                    dataflow_types::plan::join::JoinPlan::Linear(linear_plan) => {
                        self.render_join(inputs, linear_plan, scope)
                    }
                    dataflow_types::plan::join::JoinPlan::Delta(delta_plan) => {
                        self.render_delta_join(inputs, delta_plan, scope)
                    }
                }
            }
            Plan::Reduce {
                input,
                key_val_plan,
                plan,
                permutation,
            } => {
                let input = self.render_plan(*input, scope, worker_index);
                self.render_reduce(input, key_val_plan, plan, permutation)
            }
            Plan::TopK { input, top_k_plan } => {
                let input = self.render_plan(*input, scope, worker_index);
                self.render_topk(input, top_k_plan)
            }
            Plan::Negate { input } => {
                let input = self.render_plan(*input, scope, worker_index);
                let (oks, errs) = input.as_collection();
                CollectionBundle::from_collections(oks.negate(), errs)
            }
            Plan::Threshold {
                input,
                threshold_plan,
            } => {
                let input = self.render_plan(*input, scope, worker_index);
                self.render_threshold(input, threshold_plan)
            }
            Plan::Union { inputs } => {
                let mut oks = Vec::new();
                let mut errs = Vec::new();
                for input in inputs.into_iter() {
                    let (os, es) = self.render_plan(input, scope, worker_index).as_collection();
                    oks.push(os);
                    errs.push(es);
                }
                let oks = differential_dataflow::collection::concatenate(scope, oks);
                let errs = differential_dataflow::collection::concatenate(scope, errs);
                CollectionBundle::from_collections(oks, errs)
            }
            Plan::ArrangeBy {
                input,
                ensure_arrangements,
            } => {
                let input = self.render_plan(*input, scope, worker_index);
                input.ensure_arrangements(ensure_arrangements)
            }
        }
    }
/// Renders a plan to a differential dataflow, producing the collection of results.
///
/// The return type reflects the uncertainty about the data representation, perhaps
/// as a stream of data, perhaps as an arrangement, perhaps as a stream of batches.
Renders a plan to a differential dataflow, producing the collection of results.
The return type reflects the uncertainty about the data representation, perhaps
as a stream of data, perhaps as an arrangement, perhaps as a stream of batches.
Renders a plan to a differential dataflow producing the collection of results . The return type reflects the uncertainty about the data representation perhaps as a stream of data perhaps as an arrangement perhaps as a stream of batches .
pub fn render_plan(
        &mut self,
        plan: plan::Plan,
        scope: &mut G,
        worker_index: usize,
    ) -> CollectionBundle<G, Row, G::Timestamp> {
        match plan {
            Plan::Constant { rows } => {
                // Produce both rows and errs to avoid conditional dataflow construction.
                let (mut 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`.
                use differential_dataflow::lattice::Lattice;
                for (_, time, _) in rows.iter_mut() {
                    time.advance_by(self.as_of_frontier.borrow());
                }
                let mut error_time: G::Timestamp = timely::progress::Timestamp::minimum();
                error_time.advance_by(self.as_of_frontier.borrow());
                let ok_collection = rows.into_iter().to_stream(scope).as_collection();
                let err_collection = errs
                    .into_iter()
                    .map(move |e| (DataflowError::from(e), error_time, 1))
                    .to_stream(scope)
                    .as_collection();
                CollectionBundle::from_collections(ok_collection, err_collection)
            }
            Plan::Get {
                id,
                keys,
                mfp,
                key_val,
            } => {
                // Recover the collection from `self` and then apply `mfp` to it.
                // If `mfp` happens to be trivial, we can just return the collection.
                let mut collection = self
                    .lookup_id(id)
                    .unwrap_or_else(|| panic!("Get({:?}) not found at render time", id));
                if mfp.is_identity() {
                    // Assert that each of `keys` are present in `collection`.
                    assert!(keys.iter().all(|key| collection.arranged.contains_key(key)));
                    // Retain only those keys we want to import.
                    collection.arranged.retain(|key, _value| keys.contains(key));
                    collection
                } else {
                    let (oks, errs) = collection.as_collection_core(mfp, key_val);
                    CollectionBundle::from_collections(oks, errs)
                }
            }
            Plan::Let { id, value, body } => {
                // Render `value` and bind it to `id`. Complain if this shadows an id.
                let value = self.render_plan(*value, scope, worker_index);
                let prebound = self.insert_id(Id::Local(id), value);
                assert!(prebound.is_none());
                let body = self.render_plan(*body, scope, worker_index);
                self.remove_id(Id::Local(id));
                body
            }
            Plan::Mfp {
                input,
                mfp,
                key_val,
            } => {
                // If `mfp` is non-trivial, we should apply it and produce a collection.
                let input = self.render_plan(*input, scope, worker_index);
                if mfp.is_identity() {
                    input
                } else {
                    let (oks, errs) = input.as_collection_core(mfp, key_val);
                    CollectionBundle::from_collections(oks, errs)
                }
            }
            Plan::FlatMap {
                input,
                func,
                exprs,
                mfp,
            } => {
                let input = self.render_plan(*input, scope, worker_index);
                self.render_flat_map(input, func, exprs, mfp)
            }
            Plan::Join { inputs, plan } => {
                let inputs = inputs
                    .into_iter()
                    .map(|input| self.render_plan(input, scope, worker_index))
                    .collect();
                match plan {
                    dataflow_types::plan::join::JoinPlan::Linear(linear_plan) => {
                        self.render_join(inputs, linear_plan, scope)
                    }
                    dataflow_types::plan::join::JoinPlan::Delta(delta_plan) => {
                        self.render_delta_join(inputs, delta_plan, scope)
                    }
                }
            }
            Plan::Reduce {
                input,
                key_val_plan,
                plan,
                permutation,
            } => {
                let input = self.render_plan(*input, scope, worker_index);
                self.render_reduce(input, key_val_plan, plan, permutation)
            }
            Plan::TopK { input, top_k_plan } => {
                let input = self.render_plan(*input, scope, worker_index);
                self.render_topk(input, top_k_plan)
            }
            Plan::Negate { input } => {
                let input = self.render_plan(*input, scope, worker_index);
                let (oks, errs) = input.as_collection();
                CollectionBundle::from_collections(oks.negate(), errs)
            }
            Plan::Threshold {
                input,
                threshold_plan,
            } => {
                let input = self.render_plan(*input, scope, worker_index);
                self.render_threshold(input, threshold_plan)
            }
            Plan::Union { inputs } => {
                let mut oks = Vec::new();
                let mut errs = Vec::new();
                for input in inputs.into_iter() {
                    let (os, es) = self.render_plan(input, scope, worker_index).as_collection();
                    oks.push(os);
                    errs.push(es);
                }
                let oks = differential_dataflow::collection::concatenate(scope, oks);
                let errs = differential_dataflow::collection::concatenate(scope, errs);
                CollectionBundle::from_collections(oks, errs)
            }
            Plan::ArrangeBy {
                input,
                ensure_arrangements,
            } => {
                let input = self.render_plan(*input, scope, worker_index);
                input.ensure_arrangements(ensure_arrangements)
            }
        }
    }
pub fn render_plan ( & mut self , plan : plan :: Plan , scope : & mut G , worker_index : usize , ) -> CollectionBundle < G , Row , G :: Timestamp > { match plan { Plan :: Constant { rows } => { // Produce both rows and errs to avoid conditional dataflow construction. let ( mut 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`. use differential_dataflow :: lattice :: Lattice ; for ( _ , time , _ ) in rows . iter_mut ( ) { time . advance_by ( self . as_of_frontier . borrow ( ) ) ; } let mut error_time : G :: Timestamp = timely :: progress :: Timestamp :: minimum ( ) ; error_time . advance_by ( self . as_of_frontier . borrow ( ) ) ; let ok_collection = rows . into_iter ( ) . to_stream ( scope ) . as_collection ( ) ; let err_collection = errs . into_iter ( ) . map ( move | e | ( DataflowError :: from ( e ) , error_time , 1 ) ) . to_stream ( scope ) . as_collection ( ) ; CollectionBundle :: from_collections ( ok_collection , err_collection ) } Plan :: Get { id , keys , mfp , key_val , } => { // Recover the collection from `self` and then apply `mfp` to it. // If `mfp` happens to be trivial, we can just return the collection. let mut collection = self . lookup_id ( id ) . unwrap_or_else ( | | panic ! ( " " id ) ) ; if mfp . is_identity ( ) { // Assert that each of `keys` are present in `collection`. assert ! ( keys iter ( ) all ( key collection arranged contains_key ( key ) ) ) ; // Retain only those keys we want to import. collection . arranged . retain ( | key , _value | keys . contains ( key ) ) ; collection } else { let ( oks , errs ) = collection . as_collection_core ( mfp , key_val ) ; CollectionBundle :: from_collections ( oks , errs ) } } Plan :: Let { id , value , body } => { // Render `value` and bind it to `id`. Complain if this shadows an id. let value = self . render_plan ( * value , scope , worker_index ) ; let prebound = self . insert_id ( Id :: Local ( id ) , value ) ; assert ! ( prebound is_none ( ) ) ; let body = self . render_plan ( * body , scope , worker_index ) ; self . remove_id ( Id :: Local ( id ) ) ; body } Plan :: Mfp { input , mfp , key_val , } => { // If `mfp` is non-trivial, we should apply it and produce a collection. let input = self . render_plan ( * input , scope , worker_index ) ; if mfp . is_identity ( ) { input } else { let ( oks , errs ) = input . as_collection_core ( mfp , key_val ) ; CollectionBundle :: from_collections ( oks , errs ) } } Plan :: FlatMap { input , func , exprs , mfp , } => { let input = self . render_plan ( * input , scope , worker_index ) ; self . render_flat_map ( input , func , exprs , mfp ) } Plan :: Join { inputs , plan } => { let inputs = inputs . into_iter ( ) . map ( | input | self . render_plan ( input , scope , worker_index ) ) . collect ( ) ; match plan { dataflow_types :: plan :: join :: JoinPlan :: Linear ( linear_plan ) => { self . render_join ( inputs , linear_plan , scope ) } dataflow_types :: plan :: join :: JoinPlan :: Delta ( delta_plan ) => { self . render_delta_join ( inputs , delta_plan , scope ) } } } Plan :: Reduce { input , key_val_plan , plan , permutation , } => { let input = self . render_plan ( * input , scope , worker_index ) ; self . render_reduce ( input , key_val_plan , plan , permutation ) } Plan :: TopK { input , top_k_plan } => { let input = self . render_plan ( * input , scope , worker_index ) ; self . render_topk ( input , top_k_plan ) } Plan :: Negate { input } => { let input = self . render_plan ( * input , scope , worker_index ) ; let ( oks , errs ) = input . as_collection ( ) ; CollectionBundle :: from_collections ( oks . negate ( ) , errs ) } Plan :: Threshold { input , threshold_plan , } => { let input = self . render_plan ( * input , scope , worker_index ) ; self . render_threshold ( input , threshold_plan ) } Plan :: Union { inputs } => { let mut oks = Vec :: new ( ) ; let mut errs = Vec :: new ( ) ; for input in inputs . into_iter ( ) { let ( os , es ) = self . render_plan ( input , scope , worker_index ) . as_collection ( ) ; oks . push ( os ) ; errs . push ( es ) ; } let oks = differential_dataflow :: collection :: concatenate ( scope , oks ) ; let errs = differential_dataflow :: collection :: concatenate ( scope , errs ) ; CollectionBundle :: from_collections ( oks , errs ) } Plan :: ArrangeBy { input , ensure_arrangements , } => { let input = self . render_plan ( * input , scope , worker_index ) ; input . ensure_arrangements ( ensure_arrangements ) } } }
Renders a plan to a differential dataflow, producing the collection of results.
Renders a plan to a differential dataflow producing the collection of results .
&mut self
plan
plan::Plan
scope
&mut G
worker_index
usize
&mut self
plan
plan::Plan
scope
&mut G
worker_index
usize