Function try_run_sql

src/testdrive/src/action/sql.rs:211–358

(
    state: &mut State,
    query: &str,
    expected_output: &SqlOutput,
    should_retry: bool,
)

}

async fn try_run_sql(
    state: &mut State,
    query: &str,
    expected_output: &SqlOutput,
    should_retry: bool,
) -> Result<(), anyhow::Error> {
    let stmt = state
        .materialize
        .pgclient
        .prepare(query)
        .await
        .context("preparing query failed")?;

    let query_with_timeout = tokio::time::timeout(
        state.timeout.clone(),
        query_prepared(&state.materialize.pgclient, &stmt, &[]),
    )
    .await;

    if query_with_timeout.is_err() {
        bail!("query timed out\n")
    }

    let rows: Vec<_> = query_with_timeout
        .unwrap()
        .context("executing query failed")?
        .into_iter()
        .map(|row| decode_row(state, row))
        .collect::<Result<_, _>>()?;

    let (mut actual, raw_actual): (Vec<_>, Vec<_>) = rows.into_iter().unzip();

    let raw_actual: Option<Vec<_>> = if raw_actual.iter().any(|r| r.is_some()) {
        // TODO(guswynn): Note we don't sort the raw rows, because
        // there is no easy way of ensuring they sort the same way as actual.
        Some(
            actual
                .iter()
                .zip_eq(raw_actual)
                .map(|(actual, unreplaced)| match unreplaced {
                    Some(raw_row) => raw_row,
                    None => actual.clone(),
                })
                .collect(),
        )
    } else {
        None
    };

    actual.sort();
    let actual_columns: Vec<_> = stmt.columns().iter().map(|c| c.name()).collect();

    match expected_output {
        SqlOutput::Full {
            expected_rows,
            column_names,
        } => {
            if let Some(column_names) = column_names {