Function walk_sink_arrangement 

fn walk_sink_arrangement<'scope>(
    name: String,
    batches: StreamVec<'scope, Timestamp, <TraceAgent<OrdValSpine<Option<Row>, Row, Timestamp, Diff>> as TraceReader>::Batch>,
    sink_id: GlobalId,
    from_id: GlobalId,
    key_is_synthetic: bool,
) -> (VecCollection<'scope, Timestamp, (Option<Row>, DiffPair<Row>), Diff>, PressOnDropButton)

Walks each arrangement batch and emits a stream of individual (key, DiffPair) records that feeds the rest of the Iceberg sink pipeline.

Tracks per-(key, timestamp) group sizes and rate-limits a warning when a non-synthetic key has more than one DiffPair. When key_is_synthetic the arrangement’s hash-based key is stripped before emission.