pub fn join_traces<G, T1, T2, L, CB>(arranged1: &Arranged<G,T1>, arranged2: &Arranged<G,T2>, mut result: L) -> StreamCore<G, CB::Container>
where
    G: Scope<Timestamp=T1::Time>,
    T1: TraceReader+Clone+'static,
    T2: for<'a> TraceReader<Key<'a>=T1::Key<'a>, Time=T1::Time>+Clone+'static,
    L: FnMut(T1::Key<'_>,T1::Val<'_>,T2::Val<'_>,&G::Timestamp,&T1::Diff,&T2::Diff,&mut JoinSession<T1::Time, CB, CB::Container>)+'static,
    CB: ContainerBuilder + 'static,
{
    // Rename traces for symmetry from here on out.
    let mut trace1 = arranged1.trace.clone();
    let mut trace2 = arranged2.trace.clone();

    arranged1.stream.binary_frontier(&arranged2.stream, Pipeline, Pipeline, "Join", move |capability, info| {

        // Acquire an activator to reschedule the operator when it has unfinished work.
        use timely::scheduling::Activator;
        let activations = arranged1.stream.scope().activations().clone();
        let activator = Activator::new(info.address, activations);

        // Our initial invariants are that for each trace, physical compaction is less or equal the trace's upper bound.
        // These invariants ensure that we can reference observed batch frontiers from `_start_upper` onward, as long as
        // we maintain our physical compaction capabilities appropriately. These assertions are tested as we load up the
        // initial work for the two traces, and before the operator is constructed.

        // Acknowledged frontier for each input.
        // These two are used exclusively to track batch boundaries on which we may want/need to call `cursor_through`.
        // They will drive our physical compaction of each trace, and we want to maintain at all times that each is beyond
        // the physical compaction frontier of their corresponding trace.
        // Should we ever *drop* a trace, these are 1. much harder to maintain correctly, but 2. no longer used.
        use timely::progress::frontier::Antichain;
        let mut acknowledged1 = Antichain::from_elem(<G::Timestamp>::minimum());
        let mut acknowledged2 = Antichain::from_elem(<G::Timestamp>::minimum());

        // deferred work of batches from each input.
        let mut todo1 = std::collections::VecDeque::new();
        let mut todo2 = std::collections::VecDeque::new();

        // We'll unload the initial batches here, to put ourselves in a less non-deterministic state to start.
        trace1.map_batches(|batch1| {
            acknowledged1.clone_from(batch1.upper());
            // No `todo1` work here, because we haven't accepted anything into `batches2` yet.
            // It is effectively "empty", because we choose to drain `trace1` before `trace2`.
            // Once we start streaming batches in, we will need to respond to new batches from
            // `input1` with logic that would have otherwise been here. Check out the next loop
            // for the structure.
        });
        // At this point, `ack1` should exactly equal `trace1.read_upper()`, as they are both determined by
        // iterating through batches and capturing the upper bound. This is a great moment to assert that
        // `trace1`'s physical compaction frontier is before the frontier of completed times in `trace1`.
        // TODO: in the case that this does not hold, instead "upgrade" the physical compaction frontier.
        assert!(PartialOrder::less_equal(&trace1.get_physical_compaction(), &acknowledged1.borrow()));

        // We capture batch2 cursors first and establish work second to avoid taking a `RefCell` lock
        // on both traces at the same time, as they could be the same trace and this would panic.
        let mut batch2_cursors = Vec::new();
        trace2.map_batches(|batch2| {
            acknowledged2.clone_from(batch2.upper());
            batch2_cursors.push((batch2.cursor(), batch2.clone()));
        });
        // At this point, `ack2` should exactly equal `trace2.read_upper()`, as they are both determined by
        // iterating through batches and capturing the upper bound. This is a great moment to assert that
        // `trace2`'s physical compaction frontier is before the frontier of completed times in `trace2`.
        // TODO: in the case that this does not hold, instead "upgrade" the physical compaction frontier.
        assert!(PartialOrder::less_equal(&trace2.get_physical_compaction(), &acknowledged2.borrow()));

        // Load up deferred work using trace2 cursors and batches captured just above.
        for (batch2_cursor, batch2) in batch2_cursors.into_iter() {
            // It is safe to ask for `ack1` because we have confirmed it to be in advance of `distinguish_since`.
            let (trace1_cursor, trace1_storage) = trace1.cursor_through(acknowledged1.borrow()).unwrap();
            // We could downgrade the capability here, but doing so is a bit complicated mathematically.
            // TODO: downgrade the capability by searching out the one time in `batch2.lower()` and not
            // in `batch2.upper()`. Only necessary for non-empty batches, as empty batches may not have
            // that property.
            todo2.push_back(Deferred::new(trace1_cursor, trace1_storage, batch2_cursor, batch2.clone(), capability.clone()));
        }

        // Droppable handles to shared trace data structures.
        let mut trace1_option = Some(trace1);
        let mut trace2_option = Some(trace2);

        move |input1, input2, output| {

            // 1. Consuming input.
            //
            // The join computation repeatedly accepts batches of updates from each of its inputs.
            //
            // For each accepted batch, it prepares a work-item to join the batch against previously "accepted"
            // updates from its other input. It is important to track which updates have been accepted, because
            // we use a shared trace and there may be updates present that are in advance of this accepted bound.
            //
            // Batches are accepted: 1. in bulk at start-up (above), 2. as we observe them in the input stream,
            // and 3. if the trace can confirm a region of empty space directly following our accepted bound.
            // This last case is a consequence of our inability to transmit empty batches, as they may be formed
            // in the absence of timely dataflow capabilities.

            // Drain input 1, prepare work.
            input1.for_each(|capability, data| {
                // This test *should* always pass, as we only drop a trace in response to the other input emptying.
                if let Some(ref mut trace2) = trace2_option {
                    let capability = capability.retain();
                    for batch1 in data.drain(..) {
                        // Ignore any pre-loaded data.
                        if PartialOrder::less_equal(&acknowledged1, batch1.lower()) {
                            if !batch1.is_empty() {
                                // It is safe to ask for `ack2` as we validated that it was at least `get_physical_compaction()`
                                // at start-up, and have held back physical compaction ever since.
                                let (trace2_cursor, trace2_storage) = trace2.cursor_through(acknowledged2.borrow()).unwrap();
                                let batch1_cursor = batch1.cursor();
                                todo1.push_back(Deferred::new(trace2_cursor, trace2_storage, batch1_cursor, batch1.clone(), capability.clone()));
                            }

                            // To update `acknowledged1` we might presume that `batch1.lower` should equal it, but we
                            // may have skipped over empty batches. Still, the batches are in-order, and we should be
                            // able to just assume the most recent `batch1.upper`
                            debug_assert!(PartialOrder::less_equal(&acknowledged1, batch1.upper()));
                            acknowledged1.clone_from(batch1.upper());
                        }
                    }
                }
                else { panic!("`trace2_option` dropped before `input1` emptied!"); }
            });

            // Drain input 2, prepare work.
            input2.for_each(|capability, data| {
                // This test *should* always pass, as we only drop a trace in response to the other input emptying.
                if let Some(ref mut trace1) = trace1_option {
                    let capability = capability.retain();
                    for batch2 in data.drain(..) {
                        // Ignore any pre-loaded data.
                        if PartialOrder::less_equal(&acknowledged2, batch2.lower()) {
                            if !batch2.is_empty() {
                                // It is safe to ask for `ack1` as we validated that it was at least `get_physical_compaction()`
                                // at start-up, and have held back physical compaction ever since.
                                let (trace1_cursor, trace1_storage) = trace1.cursor_through(acknowledged1.borrow()).unwrap();
                                let batch2_cursor = batch2.cursor();
                                todo2.push_back(Deferred::new(trace1_cursor, trace1_storage, batch2_cursor, batch2.clone(), capability.clone()));
                            }

                            // To update `acknowledged2` we might presume that `batch2.lower` should equal it, but we
                            // may have skipped over empty batches. Still, the batches are in-order, and we should be
                            // able to just assume the most recent `batch2.upper`
                            debug_assert!(PartialOrder::less_equal(&acknowledged2, batch2.upper()));
                            acknowledged2.clone_from(batch2.upper());
                        }
                    }
                }
                else { panic!("`trace1_option` dropped before `input2` emptied!"); }
            });

            // Advance acknowledged frontiers through any empty regions that we may not receive as batches.
            if let Some(trace1) = trace1_option.as_mut() {
                trace1.advance_upper(&mut acknowledged1);
            }
            if let Some(trace2) = trace2_option.as_mut() {
                trace2.advance_upper(&mut acknowledged2);
            }

            // 2. Join computation.
            //
            // For each of the inputs, we do some amount of work (measured in terms of number
            // of output records produced). This is meant to yield control to allow downstream
            // operators to consume and reduce the output, but it it also means to provide some
            // degree of responsiveness. There is a potential risk here that if we fall behind
            // then the increasing queues hold back physical compaction of the underlying traces
            // which results in unintentionally quadratic processing time (each batch of either
            // input must scan all batches from the other input).

            // Perform some amount of outstanding work.
            let mut fuel = 1_000_000;
            while !todo1.is_empty() && fuel > 0 {
                todo1.front_mut().unwrap().work(
                    output,
                    |k,v2,v1,t,r2,r1,c| result(k,v1,v2,t,r1,r2,c),
                    &mut fuel
                );
                if !todo1.front().unwrap().work_remains() { todo1.pop_front(); }
            }

            // Perform some amount of outstanding work.
            let mut fuel = 1_000_000;
            while !todo2.is_empty() && fuel > 0 {
                todo2.front_mut().unwrap().work(
                    output,
                    |k,v1,v2,t,r1,r2,c| result(k,v1,v2,t,r1,r2,c),
                    &mut fuel
                );
                if !todo2.front().unwrap().work_remains() { todo2.pop_front(); }
            }

            // Re-activate operator if work remains.
            if !todo1.is_empty() || !todo2.is_empty() {
                activator.activate();
            }

            // 3. Trace maintenance.
            //
            // Importantly, we use `input.frontier()` here rather than `acknowledged` to track
            // the progress of an input, because should we ever drop one of the traces we will
            // lose the ability to extract information from anything other than the input.
            // For example, if we dropped `trace2` we would not be able to use `advance_upper`
            // to keep `acknowledged2` up to date wrt empty batches, and would hold back logical
            // compaction of `trace1`.

            // Maintain `trace1`. Drop if `input2` is empty, or advance based on future needs.
            if let Some(trace1) = trace1_option.as_mut() {
                if input2.frontier().is_empty() { trace1_option = None; }
                else {
                    // Allow `trace1` to compact logically up to the frontier we may yet receive,
                    // in the opposing input (`input2`). All `input2` times will be beyond this
                    // frontier, and joined times only need to be accurate when advanced to it.
                    trace1.set_logical_compaction(input2.frontier().frontier());
                    // Allow `trace1` to compact physically up to the upper bound of batches we
                    // have received in its input (`input1`). We will not require a cursor that
                    // is not beyond this bound.
                    trace1.set_physical_compaction(acknowledged1.borrow());
                }
            }

            // Maintain `trace2`. Drop if `input1` is empty, or advance based on future needs.
            if let Some(trace2) = trace2_option.as_mut() {
                if input1.frontier().is_empty() { trace2_option = None;}
                else {
                    // Allow `trace2` to compact logically up to the frontier we may yet receive,
                    // in the opposing input (`input1`). All `input1` times will be beyond this
                    // frontier, and joined times only need to be accurate when advanced to it.
                    trace2.set_logical_compaction(input1.frontier().frontier());
                    // Allow `trace2` to compact physically up to the upper bound of batches we
                    // have received in its input (`input2`). We will not require a cursor that
                    // is not beyond this bound.
                    trace2.set_physical_compaction(acknowledged2.borrow());
                }
            }
        }
    })
}
An equijoin of two traces, sharing a common key type.

This method exists to provide join functionality without opinions on the specific input types, keys and values,
that should be presented. The two traces here can have arbitrary key and value types, which can be unsized and
even potentially unrelated to the input collection data. Importantly, the key and value types could be generic
associated types (GATs) of the traces, and we would seemingly struggle to frame these types as trait arguments.

The implementation produces a caller-specified container. Implementations can use [`AsCollection`] to wrap the
output stream in a collection.

The "correctness" of this method depends heavily on the behavior of the supplied `result` function.

[`AsCollection`]: crate::collection::AsCollection
differential-dataflow/src/operators/join.rs