Function decode

src/storage/src/source/sql_server/replication.rs:744–767

Helper method to decode a row from a [`tiberius::Row`] (or 2 of them in the case of update) to a [`Row`]. This centralizes the decode and mapping to result.

(
    decoder: &SqlServerRowDecoder,
    row: &tiberius::Row,
    mz_row: &mut Row,
    arena: &RowArena,
    new_row: Option<&tiberius::Row>,
)

/// Helper method to decode a row from a [`tiberius::Row`] (or 2 of them in the case of update)
/// to a [`Row`]. This centralizes the decode and mapping to result.
fn decode(
    decoder: &SqlServerRowDecoder,
    row: &tiberius::Row,
    mz_row: &mut Row,
    arena: &RowArena,
    new_row: Option<&tiberius::Row>,
) -> Result<SourceMessage, DataflowError> {
    match decoder.decode(row, mz_row, arena, new_row) {
        Ok(()) => Ok(SourceMessage {
            key: Row::default(),
            value: mz_row.clone(),
            metadata: Row::default(),
        }),
        Err(e) => {
            let kind = DecodeErrorKind::Text(e.to_string().into());
            // TODO(sql_server2): Get the raw bytes from `tiberius`.
            let raw = format!("{row:?}");
            Err(DataflowError::DecodeError(Box::new(DecodeError {
                kind,
                raw: raw.as_bytes().to_vec(),
            })))
        }
    }
}

/// Helper method to return a "definite" error upstream.
async fn return_definite_error(