fn reduce_abelian<L, Bu, T2>(&self, name: &str, mut logic: L) -> Arranged<G, TraceAgent<T2>>
        where
            T2: for<'a> Trace<Key<'a>= &'a K, Time=G::Timestamp>+'static,
            for<'a> T2::Val<'a> : IntoOwned<'a, Owned = V>,
            T2::Diff: Abelian,
            T2::Batch: Batch,
            Bu: Builder<Time=T2::Time, Input = Vec<((K::Owned, V), T2::Time, T2::Diff)>, Output = T2::Batch>,
            L: FnMut(&K, &[(&V, R)], &mut Vec<(V, T2::Diff)>)+'static,
        {
            self.reduce_core::<_,Bu,T2>(name, move |key, input, output, change| {
                if !input.is_empty() {
                    logic(key, input, change);
                }
                change.extend(output.drain(..).map(|(x,mut d)| { d.negate(); (x, d) }));
                crate::consolidation::consolidate(change);
            })
        }
Applies `reduce` to arranged data, and returns an arrangement of output data.

This method is used by the more ergonomic `reduce`, `distinct`, and `count` methods, although
it can be very useful if one needs to manually attach and re-use existing arranged collections.

# Examples

```
use differential_dataflow::input::Input;
use differential_dataflow::operators::reduce::ReduceCore;
use differential_dataflow::trace::Trace;
use differential_dataflow::trace::implementations::{ValBuilder, ValSpine};

::timely::example(|scope| {

let trace =
scope.new_collection_from(1 .. 10u32).1
.map(|x| (x, x))
.reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<_,_,_,_>>(
"Example",
move |_key, src, dst| dst.push((*src[0].0, 1))
)
.trace;
});
```
differential-dataflow/src/operators/reduce.rs