Function normalize_column_refs

src/sql/src/pure/mysql.rs:254–288

Normalize column references to a sorted, deduplicated options list of column names.

(
    cols: Vec<UnresolvedItemName>,
    reference_resolver: &SourceReferenceResolver,
    tables: &[RequestedSourceExport<MySqlTableDesc>],
    option_name: &str,
)

/// Normalize column references to a sorted, deduplicated options list of column names.
pub(super) fn normalize_column_refs(
    cols: Vec<UnresolvedItemName>,
    reference_resolver: &SourceReferenceResolver,
    tables: &[RequestedSourceExport<MySqlTableDesc>],
    option_name: &str,
) -> Result<Vec<WithOptionValue<Aug>>, MySqlSourcePurificationError> {
    let (seq, unknown): (Vec<_>, Vec<_>) = cols.into_iter().partition(|name| {
        let (column_name, qual) = name.0.split_last().expect("non-empty");
        match reference_resolver.resolve_idx(qual) {
            // TODO: this needs to also introduce the maximum qualification on
            // the columns, i.e. ensure they have the schema name.
            Ok(idx) => tables[idx]
                .meta
                .columns
                .iter()
                .any(|n| &n.name == column_name.as_str()),
            Err(_) => false,
        }
    });

    if !unknown.is_empty() {
        return Err(MySqlSourcePurificationError::DanglingColumns {
            option_name: option_name.to_string(),
            items: unknown,
        });
    }

    let mut seq: Vec<_> = seq
        .into_iter()
        .map(WithOptionValue::UnresolvedItemName)
        .collect();
    seq.sort();
    seq.dedup();
    Ok(seq)
}

pub(super) async fn validate_requested_references_privileges(
    requested_external_references: impl Iterator<Item = &UnresolvedItemName>,