Function arrange_join_input 

fn arrange_join_input<'s, T>(
    edge: Collection<'s, T, Column<(Row, T, Diff)>>,
    stream_key: Vec<LirScalarExpr>,
    stream_thinning: Vec<usize>,
    batcher: ArrangementBatcher,
) -> (Arranged<'s, TraceAgent<RowRowSpine<T, Diff>>>, VecCollection<'s, T, DataflowErrorSer, Diff>)
where
    T: Lattice + RenderTimestamp,

Forms the source arrangement for a streamed join input off a collection edge.

Pushes the key and value borrowed into the ColumnBuilder the Col2Val batcher consumes, so the ok path holds no owned Row per record. Only the error path owns a time and diff.