Function upsert_thinning 

fn upsert_thinning<G, K, V, FromTime>(
    input: VecCollection<G, (K, V, FromTime), Diff>,
) -> VecCollection<G, (K, V, FromTime), Diff>
where
    G: Scope,
    G::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.