fn upsert_commands<'scope, T: Timestamp, FromTime: Timestamp>(
input: VecCollection<'scope, T, DecodeResult<FromTime>, Diff>,
upsert_envelope: UpsertEnvelope,
) -> VecCollection<'scope, T, (UpsertKey, Option<Result<Row, Box<UpsertError>>>, FromTime), Diff>Expand description
Convert from streams of DecodeResult to UpsertCommands, inserting the Key according to KeyEnvelope