pub fn reduce_trace<G, T1, Bu, T2, K, V, L>(trace: &Arranged<G, T1>, name: &str, mut logic: L) -> Arranged<G, TraceAgent<T2>>
where
    G: Scope<Timestamp=T1::Time>,
    T1: TraceReader + Clone + 'static,
    for<'a> T1::Key<'a> : IntoOwned<'a, Owned = K>,
    T2: for<'a> Trace<Key<'a>=T1::Key<'a>, Time=T1::Time> + 'static,
    K: Ord + 'static,
    V: Data,
    for<'a> T2::Val<'a> : IntoOwned<'a, Owned = V>,
    T2::Batch: Batch,
    Bu: Builder<Time=T2::Time, Output = T2::Batch>,
    Bu::Input: Container + PushInto<((K, V), T2::Time, T2::Diff)>,
    L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(V,T2::Diff)>, &mut Vec<(V, T2::Diff)>)+'static,
{
    let mut result_trace = None;

    // fabricate a data-parallel operator using the `unary_notify` pattern.
    let stream = {

        let result_trace = &mut result_trace;
        trace.stream.unary_frontier(Pipeline, name, move |_capability, operator_info| {

            // Acquire a logger for arrange events.
            let logger = trace.stream.scope().logger_for::<crate::logging::DifferentialEventBuilder>("differential/arrange").map(Into::into);

            let activator = Some(trace.stream.scope().activator_for(operator_info.address.clone()));
            let mut empty = T2::new(operator_info.clone(), logger.clone(), activator);
            // If there is default exert logic set, install it.
            if let Some(exert_logic) = trace.stream.scope().config().get::<ExertionLogic>("differential/default_exert_logic").cloned() {
                empty.set_exert_logic(exert_logic);
            }


            let mut source_trace = trace.trace.clone();

            let (mut output_reader, mut output_writer) = TraceAgent::new(empty, operator_info, logger);

            // let mut output_trace = TraceRc::make_from(agent).0;
            *result_trace = Some(output_reader.clone());

            // let mut thinker1 = history_replay_prior::HistoryReplayer::<V, V2, G::Timestamp, R, R2>::new();
            // let mut thinker = history_replay::HistoryReplayer::<V, V2, G::Timestamp, R, R2>::new();
            let mut new_interesting_times = Vec::<G::Timestamp>::new();

            // Our implementation maintains a list of outstanding `(key, time)` synthetic interesting times,
            // as well as capabilities for these times (or their lower envelope, at least).
            let mut interesting = Vec::<(K, G::Timestamp)>::new();
            let mut capabilities = Vec::<Capability<G::Timestamp>>::new();

            // buffers and logic for computing per-key interesting times "efficiently".
            let mut interesting_times = Vec::<G::Timestamp>::new();

            // Upper and lower frontiers for the pending input and output batches to process.
            let mut upper_limit = Antichain::from_elem(<G::Timestamp as timely::progress::Timestamp>::minimum());
            let mut lower_limit = Antichain::from_elem(<G::Timestamp as timely::progress::Timestamp>::minimum());

            // Output batches may need to be built piecemeal, and these temp storage help there.
            let mut output_upper = Antichain::from_elem(<G::Timestamp as timely::progress::Timestamp>::minimum());
            let mut output_lower = Antichain::from_elem(<G::Timestamp as timely::progress::Timestamp>::minimum());

            let id = trace.stream.scope().index();

            move |input, output| {

                // The `reduce` operator receives fully formed batches, which each serve as an indication
                // that the frontier has advanced to the upper bound of their description.
                //
                // Although we could act on each individually, several may have been sent, and it makes
                // sense to accumulate them first to coordinate their re-evaluation. We will need to pay
                // attention to which times need to be collected under which capability, so that we can
                // assemble output batches correctly. We will maintain several builders concurrently, and
                // place output updates into the appropriate builder.
                //
                // It turns out we must use notificators, as we cannot await empty batches from arrange to
                // indicate progress, as the arrange may not hold the capability to send such. Instead, we
                // must watch for progress here (and the upper bound of received batches) to tell us how
                // far we can process work.
                //
                // We really want to retire all batches we receive, so we want a frontier which reflects
                // both information from batches as well as progress information. I think this means that
                // we keep times that are greater than or equal to a time in the other frontier, deduplicated.

                let mut batch_cursors = Vec::new();
                let mut batch_storage = Vec::new();

                // Downgrade previous upper limit to be current lower limit.
                lower_limit.clear();
                lower_limit.extend(upper_limit.borrow().iter().cloned());

                // Drain the input stream of batches, validating the contiguity of the batch descriptions and
                // capturing a cursor for each of the batches as well as ensuring we hold a capability for the
                // times in the batch.
                input.for_each(|capability, batches| {

                    for batch in batches.drain(..) {
                        upper_limit.clone_from(batch.upper());
                        batch_cursors.push(batch.cursor());
                        batch_storage.push(batch);
                    }

                    // Ensure that `capabilities` covers the capability of the batch.
                    capabilities.retain(|cap| !capability.time().less_than(cap.time()));
                    if !capabilities.iter().any(|cap| cap.time().less_equal(capability.time())) {
                        capabilities.push(capability.retain());
                    }
                });

                // Pull in any subsequent empty batches we believe to exist.
                source_trace.advance_upper(&mut upper_limit);

                // Only if our upper limit has advanced should we do work.
                if upper_limit != lower_limit {

                    // If we have no capabilities, then we (i) should not produce any outputs and (ii) could not send
                    // any produced outputs even if they were (incorrectly) produced. We cannot even send empty batches
                    // to indicate forward progress, and must hope that downstream operators look at progress frontiers
                    // as well as batch descriptions.
                    //
                    // We can (and should) advance source and output traces if `upper_limit` indicates this is possible.
                    if capabilities.iter().any(|c| !upper_limit.less_equal(c.time())) {

                        // `interesting` contains "warnings" about keys and times that may need to be re-considered.
                        // We first extract those times from this list that lie in the interval we will process.
                        sort_dedup(&mut interesting);
                        // `exposed` contains interesting (key, time)s now below `upper_limit`
                        let exposed = {
                            let (exposed, new_interesting) = interesting.drain(..).partition(|(_, time)| !upper_limit.less_equal(time));
                            interesting = new_interesting;
                            exposed
                        };

                        // Prepare an output buffer and builder for each capability.
                        //
                        // We buffer and build separately, as outputs are produced grouped by time, whereas the
                        // builder wants to see outputs grouped by value. While the per-key computation could
                        // do the re-sorting itself, buffering per-key outputs lets us double check the results
                        // against other implementations for accuracy.
                        //
                        // TODO: It would be better if all updates went into one batch, but timely dataflow prevents
                        //       this as long as it requires that there is only one capability for each message.
                        let mut buffers = Vec::<(G::Timestamp, Vec<(V, G::Timestamp, T2::Diff)>)>::new();
                        let mut builders = Vec::new();
                        for cap in capabilities.iter() {
                            buffers.push((cap.time().clone(), Vec::new()));
                            builders.push(Bu::new());
                        }

                        let mut buffer = Bu::Input::default();

                        // cursors for navigating input and output traces.
                        let (mut source_cursor, source_storage): (T1::Cursor, _) = source_trace.cursor_through(lower_limit.borrow()).expect("failed to acquire source cursor");
                        let source_storage = &source_storage;
                        let (mut output_cursor, output_storage): (T2::Cursor, _) = output_reader.cursor_through(lower_limit.borrow()).expect("failed to acquire output cursor");
                        let output_storage = &output_storage;
                        let (mut batch_cursor, batch_storage) = (CursorList::new(batch_cursors, &batch_storage), batch_storage);
                        let batch_storage = &batch_storage;

                        let mut thinker = history_replay::HistoryReplayer::new();

                        // We now march through the keys we must work on, drawing from `batch_cursors` and `exposed`.
                        //
                        // We only keep valid cursors (those with more data) in `batch_cursors`, and so its length
                        // indicates whether more data remain. We move through `exposed` using (index) `exposed_position`.
                        // There could perhaps be a less provocative variable name.
                        let mut exposed_position = 0;
                        while batch_cursor.key_valid(batch_storage) || exposed_position < exposed.len() {

                            use std::borrow::Borrow;

                            // Determine the next key we will work on; could be synthetic, could be from a batch.
                            let key1 = exposed.get(exposed_position).map(|x| <_ as IntoOwned>::borrow_as(&x.0));
                            let key2 = batch_cursor.get_key(batch_storage);
                            let key = match (key1, key2) {
                                (Some(key1), Some(key2)) => ::std::cmp::min(key1, key2),
                                (Some(key1), None)       => key1,
                                (None, Some(key2))       => key2,
                                (None, None)             => unreachable!(),
                            };

                            // `interesting_times` contains those times between `lower_issued` and `upper_limit`
                            // that we need to re-consider. We now populate it, but perhaps this should be left
                            // to the per-key computation, which may be able to avoid examining the times of some
                            // values (for example, in the case of min/max/topk).
                            interesting_times.clear();

                            // Populate `interesting_times` with synthetic interesting times (below `upper_limit`) for this key.
                            while exposed.get(exposed_position).map(|x| x.0.borrow()).map(|k| key.eq(&<T1::Key<'_> as IntoOwned>::borrow_as(&k))).unwrap_or(false) {
                                interesting_times.push(exposed[exposed_position].1.clone());
                                exposed_position += 1;
                            }

                            // tidy up times, removing redundancy.
                            sort_dedup(&mut interesting_times);

                            // do the per-key computation.
                            let _counters = thinker.compute(
                                key,
                                (&mut source_cursor, source_storage),
                                (&mut output_cursor, output_storage),
                                (&mut batch_cursor, batch_storage),
                                &mut interesting_times,
                                &mut logic,
                                &upper_limit,
                                &mut buffers[..],
                                &mut new_interesting_times,
                            );

                            if batch_cursor.get_key(batch_storage) == Some(key) {
                                batch_cursor.step_key(batch_storage);
                            }

                            // Record future warnings about interesting times (and assert they should be "future").
                            for time in new_interesting_times.drain(..) {
                                debug_assert!(upper_limit.less_equal(&time));
                                interesting.push((key.into_owned(), time));
                            }

                            // Sort each buffer by value and move into the corresponding builder.
                            // TODO: This makes assumptions about at least one of (i) the stability of `sort_by`,
                            //       (ii) that the buffers are time-ordered, and (iii) that the builders accept
                            //       arbitrarily ordered times.
                            for index in 0 .. buffers.len() {
                                buffers[index].1.sort_by(|x,y| x.0.cmp(&y.0));
                                for (val, time, diff) in buffers[index].1.drain(..) {
                                    buffer.push_into(((key.into_owned(), val), time, diff));
                                    builders[index].push(&mut buffer);
                                    buffer.clear();
                                }
                            }
                        }

                        // We start sealing output batches from the lower limit (previous upper limit).
                        // In principle, we could update `lower_limit` itself, and it should arrive at
                        // `upper_limit` by the end of the process.
                        output_lower.clear();
                        output_lower.extend(lower_limit.borrow().iter().cloned());

                        // build and ship each batch (because only one capability per message).
                        for (index, builder) in builders.drain(..).enumerate() {

                            // Form the upper limit of the next batch, which includes all times greater
                            // than the input batch, or the capabilities from i + 1 onward.
                            output_upper.clear();
                            output_upper.extend(upper_limit.borrow().iter().cloned());
                            for capability in &capabilities[index + 1 ..] {
                                output_upper.insert(capability.time().clone());
                            }

                            if output_upper.borrow() != output_lower.borrow() {

                                let description = Description::new(output_lower.clone(), output_upper.clone(), Antichain::from_elem(G::Timestamp::minimum()));
                                let batch = builder.done(description);

                                // ship batch to the output, and commit to the output trace.
                                output.session(&capabilities[index]).give(batch.clone());
                                output_writer.insert(batch, Some(capabilities[index].time().clone()));

                                output_lower.clear();
                                output_lower.extend(output_upper.borrow().iter().cloned());
                            }
                        }

                        // This should be true, as the final iteration introduces no capabilities, and
                        // uses exactly `upper_limit` to determine the upper bound. Good to check though.
                        assert!(output_upper.borrow() == upper_limit.borrow());

                        // Determine the frontier of our interesting times.
                        let mut frontier = Antichain::<G::Timestamp>::new();
                        for (_, time) in &interesting {
                            frontier.insert_ref(time);
                        }

                        // Update `capabilities` to reflect interesting pairs described by `frontier`.
                        let mut new_capabilities = Vec::new();
                        for time in frontier.borrow().iter() {
                            if let Some(cap) = capabilities.iter().find(|c| c.time().less_equal(time)) {
                                new_capabilities.push(cap.delayed(time));
                            }
                            else {
                                println!("{}:\tfailed to find capability less than new frontier time:", id);
                                println!("{}:\t  time: {:?}", id, time);
                                println!("{}:\t  caps: {:?}", id, capabilities);
                                println!("{}:\t  uppr: {:?}", id, upper_limit);
                            }
                        }
                        capabilities = new_capabilities;

                        // ensure that observed progress is reflected in the output.
                        output_writer.seal(upper_limit.clone());
                    }
                    else {
                        output_writer.seal(upper_limit.clone());
                    }

                    // We only anticipate future times in advance of `upper_limit`.
                    source_trace.set_logical_compaction(upper_limit.borrow());
                    output_reader.set_logical_compaction(upper_limit.borrow());

                    // We will only slice the data between future batches.
                    source_trace.set_physical_compaction(upper_limit.borrow());
                    output_reader.set_physical_compaction(upper_limit.borrow());
                }

                // Exert trace maintenance if we have been so requested.
                output_writer.exert();
            }
        }
    )
    };

    Arranged { stream, trace: result_trace.unwrap() }
}
A key-wise reduction of values in an input trace.

This method exists to provide reduce functionality without opinions about qualifying trace types.
differential-dataflow/src/operators/reduce.rs