Function collection_attribute

src/plan/join.rs:183–227

(
    nested: &mut Iterative<'b, S, u64>,
    context: &mut I,
    target_variables: &[Var],
    left: CollectionRelation<'b, S>,
    right: AttributeBinding,
)

}

fn collection_attribute<'b, T, S, I>(
    nested: &mut Iterative<'b, S, u64>,
    context: &mut I,
    target_variables: &[Var],
    left: CollectionRelation<'b, S>,
    right: AttributeBinding,
) -> (Implemented<'b, S>, ShutdownHandle)
where
    T: Timestamp + Lattice,
    I: ImplContext<T>,
    S: Scope<Timestamp = T>,
{
    // @TODO specialized implementation

    let (tuples, shutdown_propose) = match context.forward_propose(&right.source_attribute) {
        None => panic!("attribute {:?} does not exist", &right.source_attribute),
        Some(propose_trace) => {
            let frontier: Vec<T> = propose_trace.advance_frontier().to_vec();
            let (propose, shutdown_propose) =
                propose_trace.import_core(&nested.parent, &right.source_attribute);

            let tuples = propose
                .enter_at(nested, move |_, _, time| {
                    let mut forwarded = time.clone();
                    forwarded.advance_by(&frontier);
                    Product::new(forwarded, 0)
                })
                .as_collection(|e, v| vec![e.clone(), v.clone()]);

            (tuples, shutdown_propose)
        }
    };

    let right_collected = CollectionRelation {
        variables: vec![right.variables.0, right.variables.1],
        tuples,
    };

    let (implemented, mut shutdown_handle) =
        collection_collection(nested, context, target_variables, left, right_collected);

    shutdown_handle.add_button(shutdown_propose);

    (implemented, shutdown_handle)
}

//             Some(var) => {
//                 assert!(*var == self.variables.1);