Function mz_storage::render::upsert::upsert_core

fn upsert_core<G>(
    input: &Collection<G, DecodeResult, Diff>,
    predicates: Vec<MirScalarExpr>,
    position_or: Vec<Option<usize>>,
    as_of_frontier: Antichain<Timestamp>,
    upsert_envelope: UpsertEnvelope,
    previous: Collection<G, Result<Row, UpsertError>, Diff>,
    previous_token: Option<Rc<dyn Any>>
) -> Stream<G, (Result<Row, DataflowError>, Timestamp, Diff)>where
    G: Scope<Timestamp = Timestamp>,

Internal core upsert logic.