Function flatten_results_prepend_keys 

fn flatten_results_prepend_keys<'scope, T: Timestamp>(
    none_envelope: &NoneEnvelope,
    results: VecCollection<'scope, T, KV, Diff>,
) -> VecCollection<'scope, T, Result<Row, DataflowError>, Diff>

Convert from streams of DecodeResult to Rows, inserting the Key according to KeyEnvelope