Function upsert_inner

fn upsert_inner<G: Scope, FromTime, F, Fut, US>(
    input: &Collection<G, (UpsertKey, Option<Result<Row, UpsertError>>, FromTime), Diff>,
    key_indices: Vec<usize>,
    resume_upper: Antichain<G::Timestamp>,
    previous: Collection<G, Result<Row, DataflowError>, Diff>,
    previous_token: Option<Vec<PressOnDropButton>>,
    upsert_metrics: UpsertMetrics,
    source_config: RawSourceCreationConfig,
    state: F,
    upsert_config: UpsertConfig,
    prevent_snapshot_buffering: bool,
    snapshot_buffering_max: Option<usize>,
) -> (Collection<G, Result<Row, DataflowError>, Diff>, Stream<G, (usize, HealthStatusUpdate)>, Stream<G, Infallible>, PressOnDropButton)
where
    G::Timestamp: TotalOrder,
    F: FnOnce() -> Fut + 'static,
    Fut: Future<Output = US>,
    US: UpsertStateBackend<Option<FromTime>>,
    FromTime: ExchangeData + Ord,