Function validate_source_privileges

src/sql-server-util/src/inspect.rs:937–1018

Return a Result that is empty if all tables, columns, and capture instances have the necessary permissions to and an error if any table, column, or capture instance does not have the necessary permissions for tracking changes.

(
    client: &mut Client,
    capture_instances: impl IntoIterator<Item = &str>,
)

/// or capture instance does not have the necessary permissions
/// for tracking changes.
pub async fn validate_source_privileges(
    client: &mut Client,
    capture_instances: impl IntoIterator<Item = &str>,
) -> Result<(), SqlServerError> {
    let params: SmallVec<[_; 1]> = capture_instances.into_iter().collect();

    if params.is_empty() {
        return Ok(());
    }

    let params_dyn: SmallVec<[_; 1]> = params
        .iter()
        .map(|instance| {
            let instance: &dyn tiberius::ToSql = instance;
            instance
        })
        .collect();

    let param_indexes = (1..params.len() + 1)
        .map(|idx| format!("@P{}", idx))
        .join(", ");

    // NB(ptravers): we rely on HAS_PERMS_BY_NAME to check both table and column permissions.
    let capture_instance_query = format!(
            "
        SELECT
            SCHEMA_NAME(o.schema_id) + '.' + o.name AS qualified_table_name,
            ct.capture_instance AS capture_instance,
            COALESCE(HAS_PERMS_BY_NAME(SCHEMA_NAME(o.schema_id) + '.' + o.name, 'OBJECT', 'SELECT'), 0) AS table_select,
            COALESCE(HAS_PERMS_BY_NAME('cdc.' + QUOTENAME(ct.capture_instance + '_CT') , 'OBJECT', 'SELECT'), 0) AS capture_table_select
        FROM cdc.change_tables ct
        JOIN sys.objects o ON o.object_id = ct.source_object_id
        WHERE ct.capture_instance IN ({param_indexes});
            "
        );

    let rows = client
        .query(capture_instance_query, &params_dyn[..])
        .await?;

    let mut capture_instances_without_perms = vec![];
    let mut tables_without_perms = vec![];

    for row in rows {
        let table: &str = row
            .try_get("qualified_table_name")
            .context("getting table column")?
            .ok_or_else(|| anyhow::anyhow!("no table column?"))?;

        let capture_instance: &str = row
            .try_get("capture_instance")
            .context("getting capture_instance column")?
            .ok_or_else(|| anyhow::anyhow!("no capture_instance column?"))?;

        let permitted_table: i32 = row
            .try_get("table_select")
            .context("getting table_select column")?
            .ok_or_else(|| anyhow::anyhow!("no table_select column?"))?;