Function upsert_v2 

pub(crate) fn upsert_v2<'scope, T, FromTime>(
    input: VecCollection<'scope, T, (UpsertKey, Option<Result<Row, Box<UpsertError>>>, FromTime), Diff>,
    upsert_envelope: UpsertEnvelope,
    resume_upper: Antichain<T>,
    previous: VecCollection<'scope, T, Result<Row, DataflowError>, Diff>,
    previous_token: Option<Vec<PressOnDropButton>>,
    source_config: SourceExportCreationConfig,
    backpressure_metrics: Option<BackpressureMetrics>,
) -> (VecCollection<'scope, T, Result<Row, DataflowError>, Diff>, StreamVec<'scope, T, (Option<GlobalId>, HealthStatusUpdate)>, StreamVec<'scope, T, Infallible>, PressOnDropButton)
where
    T: Timestamp + TotalOrder + Sync + Refines<Timestamp> + Lattice,
    FromTime: Timestamp + Clone + Sync,

An experimental upsert implementation loosely described in this doc: Upsert V2 Much Simpler Boogaloo

Instead of using rocksdb as a state backend, this implementation uses a differential dataflow collection to hold the key state, and performs consolidation of updates with matching keys and MZ timestamps, using max FromTime to choose winners, resulting in only one record per key per time.