pub fn half_join<G, K, V, R, Tr, FF, CF, DOut, S>(
    stream: &Collection<G, (K, V, G::Timestamp), R>,
    arrangement: Arranged<G, Tr>,
    frontier_func: FF,
    comparison: CF,
    mut output_func: S,
) -> Collection<G, (DOut, G::Timestamp), <R as Mul<Tr::Diff>>::Output>
where
    G: Scope<Timestamp = Tr::Time>,
    K: Hashable + ExchangeData,
    V: ExchangeData,
    R: ExchangeData + Monoid,
    Tr: TraceReader+Clone+'static,
    for<'a> Tr::Key<'a> : IntoOwned<'a, Owned = K>,
    R: Mul<Tr::Diff>,
    <R as Mul<Tr::Diff>>::Output: Semigroup,
    FF: Fn(&G::Timestamp, &mut Antichain<G::Timestamp>) + 'static,
    CF: Fn(Tr::TimeGat<'_>, &G::Timestamp) -> bool + 'static,
    DOut: Clone+'static,
    S: FnMut(&K, &V, Tr::Val<'_>)->DOut+'static,
{
    let output_func = move |k: &K, v1: &V, v2: Tr::Val<'_>, initial: &G::Timestamp, time: &G::Timestamp, diff1: &R, diff2: &Tr::Diff| {
        let diff = diff1.clone() * diff2.clone();
        let dout = (output_func(k, v1, v2), time.clone());
        Some((dout, initial.clone(), diff))
    };
    half_join_internal_unsafe(stream, arrangement, frontier_func, comparison, |_timer, _count| false, output_func)
}
A binary equijoin that responds to updates on only its first input.

This operator responds to inputs of the form

```ignore
((key, val1, time1), initial_time, diff1)
```

where `initial_time` is less or equal to `time1`, and produces as output

```ignore
((output_func(key, val1, val2), lub(time1, time2)), initial_time, diff1 * diff2)
```

for each `((key, val2), time2, diff2)` present in `arrangement`, where
`time2` is less than `initial_time` *UNDER THE TOTAL ORDER ON TIMES*.
This last constraint is important to ensure that we correctly produce
all pairs of output updates across multiple `half_join` operators.

Notice that the time is hoisted up into data. The expectation is that
once out of the "delta flow region", the updates will be `delay`d to the
times specified in the payloads.
dogsdogsdogs/src/operators/half_join.rs