Function upsert_operator 

fn upsert_operator<'scope, T, FromTime, F, Fut, US>(
    input: VecCollection<'scope, T, (UpsertKey, Option<Result<Row, Box<UpsertError>>>, FromTime), Diff>,
    key_indices: Vec<usize>,
    resume_upper: Antichain<T>,
    persist_input: VecCollection<'scope, T, Result<Row, DataflowError>, Diff>,
    persist_token: Option<Vec<PressOnDropButton>>,
    upsert_metrics: UpsertMetrics,
    source_config: SourceExportCreationConfig,
    state: F,
    upsert_config: UpsertConfig,
    _storage_configuration: &StorageConfiguration,
    prevent_snapshot_buffering: bool,
    snapshot_buffering_max: Option<usize>,
) -> (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>,
    F: FnOnce() -> Fut + 'static,
    Fut: Future<Output = US>,
    US: UpsertStateBackend<T, FromTime>,
    FromTime: Debug + ExchangeData + Clone + Ord + Sync,