Function encode_feedback 

fn encode_feedback<'scope, T>(
    persist_input: VecCollection<'scope, T, Result<Row, DataflowError>, Diff>,
    key_indices: Vec<usize>,
    source_statistics: SourceStatistics,
) -> Stream<'scope, T, Column<((UpsertKey, Row), T, Diff)>>
where
    T: Timestamp + TotalOrder + Sync + Refines<Timestamp> + Lattice + Columnation + Columnar + Default,
    for<'a> Ref<'a, T>: Copy + Ord,

Key the persist feedback by UpsertKey, record source statistics, and encode (UpsertKey, UpsertValue) as (UpsertKey, Row) Column chunks, the input both flavors’ feedback arrangements consume. Built with Pipeline downstream of an UpsertKey::hashed exchange, so the arrangement keeps that locality.