Method implement

src/plan/pull.rs:90–233

(
        &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>,
    {
        use differential_dataflow::operators::arrange::{Arrange, Arranged, TraceAgent};
        use differential_dataflow::operators::JoinCore;
        use differential_dataflow::trace::implementations::ord::OrdValSpine;
        use differential_dataflow::trace::TraceReader;

        let (input, mut shutdown_handle) = self.plan.implement(nested, local_arrangements, context);

        if self.pull_attributes.is_empty() {
            if self.path_attributes.is_empty() {
                // nothing to pull
                (input, shutdown_handle)
            } else {
                let path_attributes = self.path_attributes.clone();
                let tuples = {
                    let (tuples, shutdown) = input.tuples(nested, context);
                    shutdown_handle.merge_with(shutdown);

                    tuples.map(move |tuple| interleave(&tuple, &path_attributes))
                };

                (
                    Implemented::Collection(CollectionRelation {
                        variables: self.variables.to_vec(),
                        tuples,
                    }),
                    shutdown_handle,
                )
            }
        } else {
            // Arrange input entities by eid.
            let e_offset = input
                .binds(self.pull_variable)
                .expect("input relation doesn't bind pull_variable");

            let paths = {
                let (tuples, shutdown) = input.tuples(nested, context);
                shutdown_handle.merge_with(shutdown);
                tuples
            };

            let e_path: Arranged<
                Iterative<S, u64>,
                TraceAgent<OrdValSpine<Value, Vec<Value>, Product<T, u64>, isize>>,
            > = paths.map(move |t| (t[e_offset].clone(), t)).arrange();

            let mut shutdown_handle = shutdown_handle;
            let streams = self.pull_attributes.iter().map(|a| {