pub fn upsert_format(
    dps: Vec<DiffPair<Row>>,
    sink_id: GlobalId,
    from: GlobalId
) -> Option<Row>