diff --git a/src/trace/implementations/merge_batcher_col.rs b/src/trace/implementations/merge_batcher_col.rs
new file mode 100644
index 000000000..16000334a
--- /dev/null
+++ b/src/trace/implementations/merge_batcher_col.rs
@@ -0,0 +1,316 @@
+//! A general purpose `Batcher` implementation based on radix sort for TimelyStack.
+
+use timely::Container;
+use timely::communication::message::RefOrMut;
+use timely::container::columnation::{Columnation, TimelyStack};
+use timely::progress::frontier::Antichain;
+
+use ::difference::Semigroup;
+
+use lattice::Lattice;
+use trace::{Batch, Batcher, Builder};
+
+/// Creates batches from unordered tuples.
+pub struct ColumnatedMergeBatcher<B: Batch>
+    where
+        B::Key: Ord+Clone+Columnation,
+        B::Val: Ord+Clone+Columnation,
+        B::Time: Lattice+timely::progress::Timestamp+Ord+Clone+Columnation,
+        B::R: Semigroup+Columnation,
+{
+    sorter: MergeSorterColumnation<(B::Key, B::Val), B::Time, B::R>,
+    lower: Antichain<B::Time>,
+    frontier: Antichain<B::Time>,
+    phantom: ::std::marker::PhantomData<B>,
+}
+
+impl<B: Batch> Batcher<B> for ColumnatedMergeBatcher<B>
+where
+    B::Key: Ord+Clone+Columnation+'static,
+    B::Val: Ord+Clone+Columnation+'static,
+    B::Time: Lattice+timely::progress::Timestamp+Ord+Clone+Columnation+'static,
+    B::R: Semigroup+Columnation+'static,
+{
+    fn new() -> Self {
+        ColumnatedMergeBatcher {
+            sorter: MergeSorterColumnation::new(),
+            frontier: Antichain::new(),
+            lower: Antichain::from_elem(<B::Time as timely::progress::Timestamp>::minimum()),
+            phantom: ::std::marker::PhantomData,
+        }
+    }
+
+    #[inline(never)]
+    fn push_batch(&mut self, batch: RefOrMut<Vec<((B::Key, B::Val), B::Time, B::R)>>) {
+        // `batch` is either a shared reference or an owned allocations.
+        match batch {
+            RefOrMut::Ref(reference) => {
+                // This is a moment at which we could capture the allocations backing
+                // `batch` into a different form of region, rather than just  cloning.
+                // let mut owned: TimelyStack<((B::Key, B::Val), B::Time, B::R)> = self.sorter.empty();
+                // owned.clone_from(reference);
+                self.sorter.push(reference);
+            },
+            RefOrMut::Mut(reference) => {
+                self.sorter.push(reference);
+            }
+        }
+    }
+
+    // Sealing a batch means finding those updates with times not greater or equal to any time
+    // in `upper`. All updates must have time greater or equal to the previously used `upper`,
+    // which we call `lower`, by assumption that after sealing a batcher we receive no more
+    // updates with times not greater or equal to `upper`.
+    #[inline(never)]
+    fn seal(&mut self, upper: Antichain<B::Time>) -> B {
+
+        let mut builder = B::Builder::new();
+
+        let mut merged = Default::default();
+        self.sorter.finish_into(&mut merged);
+
+        let mut kept = Vec::new();
+        let mut keep = TimelyStack::default();
+
+        self.frontier.clear();
+
+        // TODO: Re-use buffer, rather than dropping.
+        for buffer in merged.drain(..) {
+            for datum @ ((key, val), time, diff) in &buffer[..] {
+                if upper.less_equal(time) {
+                    self.frontier.insert(time.clone());
+                    if keep.len() == keep.capacity() {
+                        if keep.len() > 0 {
+                            kept.push(keep);
+                            keep = self.sorter.empty();
+                        }
+                    }
+                    keep.copy(datum);
+                }
+                else {
+                    builder.push((key.clone(), val.clone(), time.clone(), diff.clone()));
+                }
+            }
+            // buffer.clear();
+            // Recycling buffer.
+            // self.sorter.push(&mut buffer);
+        }
+
+        // Finish the kept data.
+        if keep.len() > 0 {
+            kept.push(keep);
+        }
+        if kept.len() > 0 {
+            self.sorter.push_list(kept);
+        }
+
+        // Drain buffers (fast reclaimation).
+        // TODO : This isn't obviously the best policy, but "safe" wrt footprint.
+        //        In particular, if we are reading serialized input data, we may
+        //        prefer to keep these buffers around to re-fill, if possible.
+        let mut buffer = Default::default();
+        self.sorter.push(&mut buffer);
+        // We recycle buffers with allocations (capacity, and not zero-sized).
+        while buffer.capacity() > 0 && std::mem::size_of::<((B::Key,B::Val),B::Time,B::R)>() > 0 {
+            buffer = Default::default();
+            self.sorter.push(&mut buffer);
+        }
+
+        let seal = builder.done(self.lower.clone(), upper.clone(), Antichain::from_elem(<B::Time as timely::progress::Timestamp>::minimum()));
+        self.lower = upper;
+        seal
+    }
+
+    // the frontier of elements remaining after the most recent call to `self.seal`.
+    fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<B::Time> {
+        self.frontier.borrow()
+    }
+}
+
+pub struct TimelyStackQueue<T: Columnation> {
+    list: TimelyStack<T>,
+    head: usize,
+}
+
+impl<T: Columnation + 'static> TimelyStackQueue<T> {
+    #[inline]
+    pub fn new() -> Self { TimelyStackQueue::from(Default::default()) }
+    #[inline]
+    pub fn pop(&mut self) -> &T {
+        self.head += 1;
+        &self.list[self.head - 1]
+    }
+    #[inline]
+    pub fn peek(&self) -> &T {
+        &self.list[self.head]
+    }
+    #[inline]
+    pub fn from(list: TimelyStack<T>) -> Self {
+        TimelyStackQueue {
+            list,
+            head: 0,
+        }
+    }
+    #[inline]
+    pub fn done(mut self) -> TimelyStack<T> {
+        self.list.clear();
+        self.list
+    }
+    #[inline]
+    pub fn len(&self) -> usize { self.list.len() - self.head }
+    #[inline]
+    pub fn is_empty(&self) -> bool { self.head == self.list.len() }
+}
+
+pub struct MergeSorterColumnation<D: Ord+Columnation, T: Ord+Columnation, R: Semigroup+Columnation> {
+    queue: Vec<Vec<TimelyStack<(D, T, R)>>>,    // each power-of-two length list of allocations.
+    stash: Vec<TimelyStack<(D, T, R)>>,
+}
+
+impl<D: Ord+Clone+Columnation+'static, T: Ord+Clone+Columnation+'static, R: Semigroup+Columnation+'static> MergeSorterColumnation<D, T, R> {
+
+    const BUFFER_SIZE_BYTES: usize = 1 << 13;
+
+    fn buffer_size() -> usize {
+        let size = ::std::mem::size_of::<(D, T, R)>();
+        if size == 0 {
+            Self::BUFFER_SIZE_BYTES
+        } else if size <= Self::BUFFER_SIZE_BYTES {
+            Self::BUFFER_SIZE_BYTES / size
+        } else {
+            1
+        }
+    }
+
+    #[inline]
+    pub fn new() -> Self { MergeSorterColumnation { queue: Vec::new(), stash: Vec::new() } }
+
+    #[inline]
+    pub fn empty(&mut self) -> TimelyStack<(D, T, R)> {
+        self.stash.pop().unwrap_or_else(|| TimelyStack::with_capacity(Self::buffer_size()))
+    }
+
+    #[inline]
+    pub fn push(&mut self, batch: &Vec<(D, T, R)>) {
+
+        if batch.len() > 0 {
+            let mut batch = batch.clone();
+            crate::consolidation::consolidate_updates(&mut batch);
+            let mut stack = TimelyStack::with_capacity(batch.len());
+            for tuple in batch.iter() {
+                stack.copy(tuple);
+            }
+            self.queue.push(vec![stack]);
+            while self.queue.len() > 1 && (self.queue[self.queue.len()-1].len() >= self.queue[self.queue.len()-2].len() / 2) {
+                let list1 = self.queue.pop().unwrap();
+                let list2 = self.queue.pop().unwrap();
+                let merged = self.merge_by(list1, list2);
+                self.queue.push(merged);
+            }
+        }
+    }
+
+    // This is awkward, because it isn't a power-of-two length any more, and we don't want
+    // to break it down to be so.
+    pub fn push_list(&mut self, list: Vec<TimelyStack<(D, T, R)>>) {
+        while self.queue.len() > 1 && self.queue[self.queue.len()-1].len() < list.len() {
+            let list1 = self.queue.pop().unwrap();
+            let list2 = self.queue.pop().unwrap();
+            let merged = self.merge_by(list1, list2);
+            self.queue.push(merged);
+        }
+        self.queue.push(list);
+    }
+
+    #[inline(never)]
+    pub fn finish_into(&mut self, target: &mut Vec<TimelyStack<(D, T, R)>>) {
+        while self.queue.len() > 1 {
+            let list1 = self.queue.pop().unwrap();
+            let list2 = self.queue.pop().unwrap();
+            let merged = self.merge_by(list1, list2);
+            self.queue.push(merged);
+        }
+
+        if let Some(mut last) = self.queue.pop() {
+            ::std::mem::swap(&mut last, target);
+        }
+    }
+
+    // merges two sorted input lists into one sorted output list.
+    #[inline(never)]
+    fn merge_by(&mut self, list1: Vec<TimelyStack<(D, T, R)>>, list2: Vec<TimelyStack<(D, T, R)>>) -> Vec<TimelyStack<(D, T, R)>> {
+
+        use std::cmp::Ordering;
+
+        // TODO: `list1` and `list2` get dropped; would be better to reuse?
+        let mut output = Vec::with_capacity(list1.len() + list2.len());
+        let mut result = self.empty();
+
+        let mut list1 = list1.into_iter().peekable();
+        let mut list2 = list2.into_iter().peekable();
+
+        let mut head1 = if list1.peek().is_some() { TimelyStackQueue::from(list1.next().unwrap()) } else { TimelyStackQueue::new() };
+        let mut head2 = if list2.peek().is_some() { TimelyStackQueue::from(list2.next().unwrap()) } else { TimelyStackQueue::new() };
+
+        // while we have valid data in each input, merge.
+        while !head1.is_empty() && !head2.is_empty() {
+
+            while (result.capacity() - result.len()) > 0 && head1.len() > 0 && head2.len() > 0 {
+
+                let cmp = {
+                    let x = head1.peek();
+                    let y = head2.peek();
+                    (&x.0, &x.1).cmp(&(&y.0, &y.1))
+                };
+                match cmp {
+                    Ordering::Less    => { result.copy(head1.pop()); }
+                    Ordering::Greater => { result.copy(head2.pop()); }
+                    Ordering::Equal   => {
+                        let (data1, time1, diff1) = head1.pop();
+                        let (_data2, _time2, diff2) = head2.pop();
+                        let mut diff1 = diff1.clone();
+                        diff1.plus_equals(&diff2);
+                        if !diff1.is_zero() {
+                            result.copy_destructured(data1, time1, &diff1);
+                        }
+                    }
+                }
+            }
+
+            if result.capacity() == result.len() {
+                output.push(result);
+                result = self.empty();
+            }
+
+            if head1.is_empty() {
+                let done1 = head1.done();
+                if done1.capacity() == Self::buffer_size() { self.stash.push(done1); }
+                head1 = if list1.peek().is_some() { TimelyStackQueue::from(list1.next().unwrap()) } else { TimelyStackQueue::new() };
+            }
+            if head2.is_empty() {
+                let done2 = head2.done();
+                if done2.capacity() == Self::buffer_size() { self.stash.push(done2); }
+                head2 = if list2.peek().is_some() { TimelyStackQueue::from(list2.next().unwrap()) } else { TimelyStackQueue::new() };
+            }
+        }
+
+        if result.len() > 0 { output.push(result); }
+        else if result.capacity() > 0 { self.stash.push(result); }
+
+        if !head1.is_empty() {
+            let mut result = self.empty();
+            for _ in 0 .. head1.len() { result.copy(head1.pop()); }
+            output.push(result);
+        }
+        output.extend(list1);
+
+        if !head2.is_empty() {
+            let mut result = self.empty();
+            for _ in 0 .. head2.len() { result.copy(head2.pop()); }
+            output.push(result);
+        }
+        output.extend(list2);
+
+        output
+    }
+}
diff --git a/src/trace/implementations/mod.rs b/src/trace/implementations/mod.rs
index 4eee120de..1fd7e80eb 100644
--- a/src/trace/implementations/mod.rs
+++ b/src/trace/implementations/mod.rs
@@ -40,7 +40,8 @@
 
 pub mod spine_fueled;
 
-mod merge_batcher;
+pub(crate) mod merge_batcher;
+pub(crate) mod merge_batcher_col;
 
 pub use self::merge_batcher::MergeBatcher as Batcher;
 
diff --git a/src/trace/implementations/ord.rs b/src/trace/implementations/ord.rs
index 5c995e048..9cb566209 100644
--- a/src/trace/implementations/ord.rs
+++ b/src/trace/implementations/ord.rs
@@ -38,10 +38,10 @@ use super::merge_batcher::MergeBatcher;
 use abomonation::abomonated::Abomonated;
 
 /// A trace implementation using a spine of ordered lists.
-pub type OrdValSpine<K, V, T, R, O=usize> = Spine<Rc<OrdValBatch<K, V, T, R, O>>>;
+pub type OrdValSpine<K, V, T, R, O=usize> = Spine<Rc<OrdValBatch<K, V, T, R, Vec<((K,V),T,R)>, O>>>;
 
 /// A trace implementation using a spine of abomonated ordered lists.
-pub type OrdValSpineAbom<K, V, T, R, O=usize> = Spine<Rc<Abomonated<OrdValBatch<K, V, T, R, O>, Vec<u8>>>>;
+pub type OrdValSpineAbom<K, V, T, R, O=usize> = Spine<Rc<Abomonated<OrdValBatch<K, V, T, R, Vec<((K,V),T,R)>,O>, Vec<u8>>>>;
 
 /// A trace implementation for empty values using a spine of ordered lists.
 pub type OrdKeySpine<K, T, R, O=usize> = Spine<Rc<OrdKeyBatch<K, T, R, O>>>;
@@ -50,9 +50,9 @@ pub type OrdKeySpine<K, T, R, O=usize> = Spine<Rc<OrdKeyBatch<K, T, R, O>>>;
 pub type OrdKeySpineAbom<K, T, R, O=usize> = Spine<Rc<Abomonated<OrdKeyBatch<K, T, R, O>, Vec<u8>>>>;
 
 /// A trace implementation backed by columnar storage.
-pub type ColValSpine<K, V, T, R, O=usize> = Spine<Rc<OrdValBatch<K, V, T, R, O, TimelyStack<K>, TimelyStack<V>>>>;
+pub type ColValSpine<K, V, T, R, O=usize> = Spine<Rc<OrdValBatch<K, V, T, R, TimelyStack<((K,V),T,R)>, O, TimelyStack<K>, TimelyStack<V>>>>;
 /// A trace implementation backed by columnar storage.
-pub type ColKeySpine<K, T, R, O=usize> = Spine<Rc<OrdKeyBatch<K,  T, R, O, TimelyStack<K>>>>;
+pub type ColKeySpine<K, T, R, O=usize> = Spine<Rc<OrdKeyBatch<K, T, R, O, TimelyStack<K>>>>;
 
 
 /// A container that can retain/discard from some offset onward.
@@ -87,7 +87,7 @@ impl<T: Columnation> RetainFrom<T> for TimelyStack<T> {
 
 /// An immutable collection of update tuples, from a contiguous interval of logical times.
 #[derive(Debug, Abomonation)]
-pub struct OrdValBatch<K, V, T, R, O=usize, CK=Vec<K>, CV=Vec<V>>
+pub struct OrdValBatch<K, V, T, R, C, O=usize, CK=Vec<K>, CV=Vec<V>>
 where
     K: Ord+Clone,
     V: Ord+Clone,
@@ -101,9 +101,11 @@ where
     pub layer: OrderedLayer<K, OrderedLayer<V, OrderedLeaf<T, R>, O, CV>, O, CK>,
     /// Description of the update times this layer represents.
     pub desc: Description<T>,
+    /// Phantom marker
+    phantom: std::marker::PhantomData<C>
 }
 
-impl<K, V, T, R, O, CK, CV> BatchReader for OrdValBatch<K, V, T, R, O, CK, CV>
+impl<K, V, T, R, O, C, CK, CV> BatchReader for OrdValBatch<K, V, T, R, C, O, CK, CV>
 where
     K: Ord+Clone+'static,
     V: Ord+Clone+'static,
@@ -118,13 +120,13 @@ where
     type Time = T;
     type R = R;
 
-    type Cursor = OrdValCursor<K, V, T, R, O, CK, CV>;
+    type Cursor = OrdValCursor<K, V, T, R, C, O, CK, CV>;
     fn cursor(&self) -> Self::Cursor { OrdValCursor { cursor: self.layer.cursor(), phantom: std::marker::PhantomData } }
     fn len(&self) -> usize { <OrderedLayer<K, OrderedLayer<V, OrderedLeaf<T, R>, O, CV>, O, CK> as Trie>::tuples(&self.layer) }
     fn description(&self) -> &Description<T> { &self.desc }
 }
 
-impl<K, V, T, R, O, CK, CV> Batch for OrdValBatch<K, V, T, R, O, CK, CV>
+impl<K, V, T, R, O, CK, CV> Batch for OrdValBatch<K, V, T, R, Vec<((K, V), T, R)>, O, CK, CV>
 where
     K: Ord+Clone+'static,
     V: Ord+Clone+'static,
@@ -135,15 +137,36 @@ where
     CV: BatchContainer<Item=V>+Deref<Target=[V]>+RetainFrom<V>,
 {
     type Batcher = MergeBatcher<Self>;
-    type Builder = OrdValBuilder<K, V, T, R, O, CK, CV>;
-    type Merger = OrdValMerger<K, V, T, R, O, CK, CV>;
+    type Builder = OrdValBuilder<K, V, T, R, Vec<((K, V), T, R)>, O, CK, CV>;
+    type Merger = OrdValMerger<K, V, T, R, Vec<((K, V), T, R)>, O, CK, CV>;
 
     fn begin_merge(&self, other: &Self, compaction_frontier: Option<AntichainRef<T>>) -> Self::Merger {
         OrdValMerger::new(self, other, compaction_frontier)
     }
 }
 
-impl<K, V, T, R, O, CK, CV> OrdValBatch<K, V, T, R, O, CK, CV>
+use trace::implementations::merge_batcher_col;
+impl<K, V, T, R, O, CK, CV> Batch for OrdValBatch<K, V, T, R, TimelyStack<((K, V), T, R)>, O, CK, CV>
+where
+    K: Ord+Clone+Columnation+'static,
+    V: Ord+Clone+Columnation+'static,
+    T: Lattice+timely::progress::Timestamp+Ord+Clone+::std::fmt::Debug+Columnation+'static,
+    R: Semigroup+Columnation,
+    O: OrdOffset, <O as TryFrom<usize>>::Error: Debug, <O as TryInto<usize>>::Error: Debug,
+    CK: BatchContainer<Item=K>+Deref<Target=[K]>+RetainFrom<K>,
+    CV: BatchContainer<Item=V>+Deref<Target=[V]>+RetainFrom<V>,
+{
+    type Batcher = merge_batcher_col::ColumnatedMergeBatcher<Self>;
+    type Builder = OrdValBuilder<K, V, T, R, TimelyStack<((K, V), T, R)>, O, CK, CV>;
+    type Merger = OrdValMerger<K, V, T, R, TimelyStack<((K, V), T, R)>, O, CK, CV>;
+
+    fn begin_merge(&self, other: &Self, compaction_frontier: Option<AntichainRef<T>>) -> Self::Merger {
+        OrdValMerger::new(self, other, compaction_frontier)
+    }
+}
+
+
+impl<K, V, T, R, O, C, CK, CV> OrdValBatch<K, V, T, R, C, O, CK, CV>
 where
     K: Ord+Clone+'static,
     V: Ord+Clone+'static,
@@ -242,7 +265,7 @@ where
 }
 
 /// State for an in-progress merge.
-pub struct OrdValMerger<K, V, T, R, O=usize, CK=Vec<K>, CV=Vec<V>>
+pub struct OrdValMerger<K, V, T, R, C, O=usize, CK=Vec<K>, CV=Vec<V>>
 where
     K: Ord+Clone+'static,
     V: Ord+Clone+'static,
@@ -262,9 +285,11 @@ where
     result: <OrderedLayer<K, OrderedLayer<V, OrderedLeaf<T, R>, O, CV>, O, CK> as Trie>::MergeBuilder,
     description: Description<T>,
     should_compact: bool,
+    /// Phantom marker
+    phantom: std::marker::PhantomData<C>
 }
 
-impl<K, V, T, R, O, CK, CV> Merger<OrdValBatch<K, V, T, R, O, CK, CV>> for OrdValMerger<K, V, T, R, O, CK, CV>
+impl<K, V, T, R, O, C, CK, CV> Merger<OrdValBatch<K, V, T, R, C, O, CK, CV>> for OrdValMerger<K, V, T, R, C, O, CK, CV>
 where
     K: Ord+Clone+'static,
     V: Ord+Clone+'static,
@@ -273,8 +298,9 @@ where
     O: OrdOffset, <O as TryFrom<usize>>::Error: Debug, <O as TryInto<usize>>::Error: Debug,
     CK: BatchContainer<Item=K>+Deref<Target=[K]>+RetainFrom<K>,
     CV: BatchContainer<Item=V>+Deref<Target=[V]>+RetainFrom<V>,
+    OrdValBatch<K, V, T, R, C, O, CK, CV>: Batch<Key=K, Val=V, Time=T, R=R>,
 {
-    fn new(batch1: &OrdValBatch<K, V, T, R, O, CK, CV>, batch2: &OrdValBatch<K, V, T, R, O, CK, CV>, compaction_frontier: Option<AntichainRef<T>>) -> Self {
+    fn new(batch1: &OrdValBatch<K, V, T, R, C, O, CK, CV>, batch2: &OrdValBatch<K, V, T, R, C, O, CK, CV>, compaction_frontier: Option<AntichainRef<T>>) -> Self {
 
         assert!(batch1.upper() == batch2.lower());
 
@@ -293,9 +319,10 @@ where
             result: <<OrderedLayer<K, OrderedLayer<V, OrderedLeaf<T, R>, O, CV>, O, CK> as Trie>::MergeBuilder as MergeBuilder>::with_capacity(&batch1.layer, &batch2.layer),
             description: description,
             should_compact: compaction_frontier.is_some(),
+            phantom: std::marker::PhantomData,
         }
     }
-    fn done(self) -> OrdValBatch<K, V, T, R, O, CK, CV> {
+    fn done(self) -> OrdValBatch<K, V, T, R, C, O, CK, CV> {
 
         assert!(self.lower1 == self.upper1);
         assert!(self.lower2 == self.upper2);
@@ -303,9 +330,10 @@ where
         OrdValBatch {
             layer: self.result.done(),
             desc: self.description,
+            phantom: std::marker::PhantomData,
         }
     }
-    fn work(&mut self, source1: &OrdValBatch<K,V,T,R,O,CK,CV>, source2: &OrdValBatch<K,V,T,R,O,CK,CV>, fuel: &mut isize) {
+    fn work(&mut self, source1: &OrdValBatch<K,V,T,R,C,O,CK,CV>, source2: &OrdValBatch<K,V,T,R,C,O,CK,CV>, fuel: &mut isize) {
 
         let starting_updates = self.result.vals.vals.vals.len();
         let mut effort = 0isize;
@@ -344,7 +372,7 @@ where
 
         // if we are supplied a frontier, we should compact.
         if self.should_compact {
-            OrdValBatch::<K, V, T, R, O, CK, CV>::advance_builder_from(&mut self.result, self.description.since().borrow(), initial_key_pos);
+            OrdValBatch::<K, V, T, R, C, O, CK, CV>::advance_builder_from(&mut self.result, self.description.since().borrow(), initial_key_pos);
         }
 
         *fuel -= effort;
@@ -357,7 +385,7 @@ where
 
 /// A cursor for navigating a single layer.
 #[derive(Debug)]
-pub struct OrdValCursor<K, V, T, R, O=usize, CK=Vec<K>, CV=Vec<V>>
+pub struct OrdValCursor<K, V, T, R, C, O=usize, CK=Vec<K>, CV=Vec<V>>
 where
     V: Ord+Clone,
     T: Lattice+Ord+Clone,
@@ -366,11 +394,11 @@ where
     CK: BatchContainer<Item=K>+Deref<Target=[K]>+RetainFrom<K>,
     CV: BatchContainer<Item=V>+Deref<Target=[V]>+RetainFrom<V>,
 {
-    phantom: std::marker::PhantomData<(K, CK, CV)>,
+    phantom: std::marker::PhantomData<(K, CK, CV, C)>,
     cursor: OrderedCursor<OrderedLayer<V, OrderedLeaf<T, R>, O, CV>>,
 }
 
-impl<K, V, T, R, O, CK, CV> Cursor for OrdValCursor<K, V, T, R, O, CK, CV>
+impl<K, V, T, R, O, C, CK, CV> Cursor for OrdValCursor<K, V, T, R, C, O, CK, CV>
 where
     K: Ord+Clone,
     V: Ord+Clone,
@@ -385,7 +413,7 @@ where
     type Time = T;
     type R = R;
 
-    type Storage = OrdValBatch<K, V, T, R, O, CK, CV>;
+    type Storage = OrdValBatch<K, V, T, R, C, O, CK, CV>;
 
     fn key<'a>(&self, storage: &'a Self::Storage) -> &'a K { &self.cursor.key(&storage.layer) }
     fn val<'a>(&self, storage: &'a Self::Storage) -> &'a V { &self.cursor.child.key(&storage.layer.vals) }
@@ -408,7 +436,7 @@ where
 
 
 /// A builder for creating layers from unsorted update tuples.
-pub struct OrdValBuilder<K, V, T, R, O=usize, CK=Vec<K>, CV=Vec<V>>
+pub struct OrdValBuilder<K, V, T, R, C, O=usize, CK=Vec<K>, CV=Vec<V>>
 where
     K: Ord+Clone,
     V: Ord+Clone,
@@ -419,9 +447,11 @@ where
     CV: BatchContainer<Item=V>+Deref<Target=[V]>+RetainFrom<V>,
 {
     builder: OrderedBuilder<K, OrderedBuilder<V, OrderedLeafBuilder<T, R>, O, CV>, O, CK>,
+    /// Phantom marker
+    phantom: std::marker::PhantomData<C>
 }
 
-impl<K, V, T, R, O, CK, CV> Builder<OrdValBatch<K, V, T, R, O, CK, CV>> for OrdValBuilder<K, V, T, R, O, CK, CV>
+impl<K, V, T, R, O, C, CK, CV> Builder<OrdValBatch<K, V, T, R, C, O, CK, CV>> for OrdValBuilder<K, V, T, R, C, O, CK, CV>
 where
     K: Ord+Clone+'static,
     V: Ord+Clone+'static,
@@ -430,16 +460,19 @@ where
     O: OrdOffset, <O as TryFrom<usize>>::Error: Debug, <O as TryInto<usize>>::Error: Debug,
     CK: BatchContainer<Item=K>+Deref<Target=[K]>+RetainFrom<K>,
     CV: BatchContainer<Item=V>+Deref<Target=[V]>+RetainFrom<V>,
+    OrdValBatch<K, V, T, R, C, O, CK, CV>: Batch<Key=K, Val=V, Time=T, R=R>,
 {
 
     fn new() -> Self {
         OrdValBuilder {
-            builder: OrderedBuilder::<K, OrderedBuilder<V, OrderedLeafBuilder<T, R>, O, CV>, O, CK>::new()
+            builder: OrderedBuilder::<K, OrderedBuilder<V, OrderedLeafBuilder<T, R>, O, CV>, O, CK>::new(),
+            phantom: std::marker::PhantomData,
         }
     }
     fn with_capacity(cap: usize) -> Self {
         OrdValBuilder {
-            builder: <OrderedBuilder<K, OrderedBuilder<V, OrderedLeafBuilder<T, R>, O, CV>, O, CK> as TupleBuilder>::with_capacity(cap)
+            builder: <OrderedBuilder<K, OrderedBuilder<V, OrderedLeafBuilder<T, R>, O, CV>, O, CK> as TupleBuilder>::with_capacity(cap),
+            phantom: std::marker::PhantomData,
         }
     }
 
@@ -449,10 +482,11 @@ where
     }
 
     #[inline(never)]
-    fn done(self, lower: Antichain<T>, upper: Antichain<T>, since: Antichain<T>) -> OrdValBatch<K, V, T, R, O, CK, CV> {
+    fn done(self, lower: Antichain<T>, upper: Antichain<T>, since: Antichain<T>) -> OrdValBatch<K, V, T, R, C, O, CK, CV> {
         OrdValBatch {
             layer: self.builder.done(),
-            desc: Description::new(lower, upper, since)
+            desc: Description::new(lower, upper, since),
+            phantom: std::marker::PhantomData,
         }
     }
 }
diff --git a/tests/trace.rs b/tests/trace.rs
index d00c4497e..fde37bd4a 100644
--- a/tests/trace.rs
+++ b/tests/trace.rs
@@ -11,11 +11,11 @@ use differential_dataflow::trace::{Trace, TraceReader, Batch, Batcher};
 use differential_dataflow::trace::cursor::Cursor;
 use differential_dataflow::trace::implementations::spine_fueled::Spine;
 
-pub type OrdValSpine<K, V, T, R> = Spine<Rc<OrdValBatch<K, V, T, R>>>;
+pub type OrdValSpine<K, V, T, R> = Spine<Rc<OrdValBatch<K, V, T, R, Vec<((K,V),T,R)>>>>;
 
 type IntegerTrace = OrdValSpine<u64, u64, usize, i64>;
 
-fn get_trace() -> Spine<Rc<OrdValBatch<u64, u64, usize, i64>>> {
+fn get_trace() -> Spine<Rc<OrdValBatch<u64, u64, usize, i64, Vec<((u64,u64),usize,i64)>>>> {
     let op_info = OperatorInfo::new(0, 0, &[]);
     let mut trace = IntegerTrace::new(op_info, None, None);
     {