Function upsert_thinning 

fn upsert_thinning<'scope, T, K, V, FromTime>(
    input: VecCollection<'scope, T, (K, V, FromTime), Diff>,
) -> VecCollection<'scope, T, (K, V, FromTime), Diff>
where
    T: Timestamp + TotalOrder,
    K: ExchangeData + Clone + Eq + Ord,
    V: ExchangeData + Clone,
    FromTime: Timestamp,

Renders an operator that discards updates that are known to not affect the outcome of upsert in a streaming fashion. For each distinct (key, time) in the input it emits the value with the highest from_time. Its purpose is to thin out data as much as possible before exchanging them across workers.