Function upsert 

pub(crate) fn upsert<'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,
    instance_context: &StorageInstanceContext,
    storage_configuration: &StorageConfiguration,
    dataflow_paramters: &DataflowParameters,
    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>,
    FromTime: Timestamp + Clone + Sync,

Resumes an upsert computation at resume_upper given as inputs a collection of upsert commands and the collection of the previous output of this operator. Returns a tuple of

• A collection of the computed upsert operator and,
• A health update stream to propagate errors