Method implement

src/plan/filter.rs:73–131

(
        &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>,
    {
        let (relation, mut shutdown_handle) =
            self.plan.implement(nested, local_arrangements, context);

        let key_offsets: Vec<usize> = self
            .variables
            .iter()
            .map(|variable| relation.binds(*variable).expect("variable not found"))
            .collect();

        let binary_predicate = match self.predicate {
            Predicate::LT => lt,
            Predicate::LTE => lte,
            Predicate::GT => gt,
            Predicate::GTE => gte,
            Predicate::EQ => eq,
            Predicate::NEQ => neq,
        };

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

        let filtered = if let Some(constant) = self.constants[0].clone() {
            CollectionRelation {
                variables,
                tuples: projected
                    .filter(move |tuple| binary_predicate(&constant, &tuple[key_offsets[0]])),
            }
        } else if let Some(constant) = self.constants[1].clone() {
            CollectionRelation {
                variables,
                tuples: projected
                    .filter(move |tuple| binary_predicate(&tuple[key_offsets[0]], &constant)),
            }
        } else {
            CollectionRelation {
                variables,
                tuples: projected.filter(move |tuple| {
                    binary_predicate(&tuple[key_offsets[0]], &tuple[key_offsets[1]])
                }),
            }
        };

        (Implemented::Collection(filtered), shutdown_handle)