Function dataflow::render::upsert::upsert_core

fn upsert_core<G>(
    stream: &Stream<G, DecodeResult>, 
    key_arity: usize, 
    source_arity: usize, 
    predicates: Vec<MirScalarExpr>, 
    position_or: Vec<Option<usize>>, 
    as_of_frontier: Antichain<Timestamp>
) -> Stream<G, (Result<Row, DataflowError>, u64, isize)> where
    G: Scope<Timestamp = Timestamp>, 

Internal core upsert logic.