pub(crate) fn consolidate_updates_kv<K, V, I>(
    rows: I
) -> impl Iterator<Item = ((K, V), Timestamp, Diff)>where
    I: IntoIterator<Item = ((Value, Value), Timestamp, Diff)>,
    K: Data,
    V: Data,