Method implement

src/plan/mod.rs:371–538

(
        &self,
        nested: &mut Iterative<'b, S, u64>,
        local_arrangements: &VariableMap<Iterative<'b, S, u64>>,
        context: &mut I,
    )

    }

    fn implement<'b, T, I, S>(
        &self,
        nested: &mut Iterative<'b, S, u64>,
        local_arrangements: &VariableMap<Iterative<'b, S, u64>>,
        context: &mut I,
    ) -> (Implemented<'b, S>, ShutdownHandle)
    where
        T: Timestamp + Lattice,
        I: ImplContext<T>,
        S: Scope<Timestamp = T>,
    {
        match *self {
            Plan::Project(ref projection) => {
                projection.implement(nested, local_arrangements, context)
            }
            Plan::Aggregate(ref aggregate) => {
                aggregate.implement(nested, local_arrangements, context)
            }
            Plan::Union(ref union) => union.implement(nested, local_arrangements, context),
            Plan::Join(ref join) => join.implement(nested, local_arrangements, context),
            Plan::Hector(ref hector) => hector.implement(nested, local_arrangements, context),
            Plan::Antijoin(ref antijoin) => antijoin.implement(nested, local_arrangements, context),
            Plan::Negate(ref plan) => {
                let (relation, mut shutdown_handle) =
                    plan.implement(nested, local_arrangements, context);
                let variables = relation.variables();

                let tuples = {
                    let (projected, shutdown) = relation.projected(nested, context, &variables);
                    shutdown_handle.merge_with(shutdown);

                    projected.negate()
                };

                (
                    Implemented::Collection(CollectionRelation { variables, tuples }),
                    shutdown_handle,
                )
            }
            Plan::Filter(ref filter) => filter.implement(nested, local_arrangements, context),
            Plan::Transform(ref transform) => {
                transform.implement(nested, local_arrangements, context)
            }
            Plan::MatchA(e, ref a, v) => {
                let binding = AttributeBinding {
                    variables: (e, v),
                    source_attribute: a.to_string(),
                };

                (Implemented::Attribute(binding), ShutdownHandle::empty())
            }
            Plan::MatchEA(match_e, ref a, sym1) => {
                let (tuples, shutdown_propose) = match context.forward_propose(a) {
                    None => panic!("attribute {:?} does not exist", a),
                    Some(propose_trace) => {
                        let frontier: Vec<T> = propose_trace.advance_frontier().to_vec();
                        let (propose, shutdown_propose) =
                            propose_trace.import_core(&nested.parent, a);