Function join_function 

pub fn join_function<U, I, L>(
    input: Collection<'_, U::Time, RecordedUpdates<U>>,
    logic: L,
) -> Collection<'_, U::Time, RecordedUpdates<U>>
where
    U::Time: Lattice,
    U: ColumnarUpdate<Diff: Multiply<U::Diff, Output = U::Diff>>,
    I: IntoIterator<Item = (U::Key, U::Val, U::Time, U::Diff)>,
    L: FnMut(Ref<'_, U::Key>, Ref<'_, U::Val>, Ref<'_, U::Time>, Ref<'_, U::Diff>) -> I + 'static,

A columnar flat_map: iterates RecordedUpdates, calls logic per (key, val, time, diff), joins output times with input times, multiplies output diffs with input diffs.

This subsumes map, filter, negate, and enter_at for columnar collections.