Function normalize_column_refs

src/sql/src/pure/sql_server.rs:565–601

Normalize the provided `cols` references to a sorted and deduplicated list of [`WithOptionValue`]s. Returns an error if any column reference in `cols` is not part of a table in the provided `tables`.

(
    cols: &[UnresolvedItemName],
    reference_resolver: &SourceReferenceResolver,
    tables: &[RequestedSourceExport<SqlServerTableDesc>],
    option_name: SqlServerConfigOptionName,
)

/// Returns an error if any column reference in `cols` is not part of a table in the
/// provided `tables`.
fn normalize_column_refs(
    cols: &[UnresolvedItemName],
    reference_resolver: &SourceReferenceResolver,
    tables: &[RequestedSourceExport<SqlServerTableDesc>],
    option_name: SqlServerConfigOptionName,
) -> Result<Vec<WithOptionValue<Aug>>, SqlServerSourcePurificationError> {
    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(sql_server3): 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(SqlServerSourcePurificationError::DanglingColumns {
            option_name: option_name.to_string(),
            items: unknown.into_iter().cloned().collect(),
        });
    }

    let mut seq: Vec<_> = seq
        .into_iter()
        .cloned()
        .map(WithOptionValue::UnresolvedItemName)
        .collect();

    seq.sort();
    seq.dedup();
    Ok(seq)
}