Function validate_source_export_names

src/sql/src/pure.rs:144–205

TODO(database-issues#8620): Remove once subsources are removed Validates the requested subsources do not have name conflicts with each other and that the same upstream table is not referenced multiple times.

(
    requested_source_exports: &[RequestedSourceExport<T>],
)

/// Validates the requested subsources do not have name conflicts with each other
/// and that the same upstream table is not referenced multiple times.
fn validate_source_export_names<T>(
    requested_source_exports: &[RequestedSourceExport<T>],
) -> Result<(), PlanError> {
    // This condition would get caught during the catalog transaction, but produces a
    // vague, non-contextual error. Instead, error here so we can suggest to the user
    // how to fix the problem.
    if let Some(name) = requested_source_exports
        .iter()
        .map(|subsource| &subsource.name)
        .duplicates()
        .next()
        .cloned()
    {
        let mut upstream_references: Vec<_> = requested_source_exports
            .into_iter()
            .filter_map(|subsource| {
                if &subsource.name == &name {
                    Some(subsource.external_reference.clone())
                } else {
                    None
                }
            })
            .collect();

        upstream_references.sort();

        Err(PlanError::SubsourceNameConflict {
            name,
            upstream_references,
        })?;
    }

    // We disallow subsource statements from referencing the same upstream table, but we allow
    // `CREATE TABLE .. FROM SOURCE` statements to do so. Since `CREATE TABLE .. FROM SOURCE`
    // purification will only provide 1 `requested_source_export`, we can leave this here without
    // needing to differentiate between the two types of statements.
    // TODO(roshan): Remove this when auto-generated subsources are deprecated.
    if let Some(name) = requested_source_exports
        .iter()
        .map(|export| &export.external_reference)
        .duplicates()
        .next()
        .cloned()
    {
        let mut target_names: Vec<_> = requested_source_exports
            .into_iter()
            .filter_map(|export| {
                if &export.external_reference == &name {
                    Some(export.name.clone())
                } else {
                    None
                }
            })
            .collect();

        target_names.sort();

        Err(PlanError::SubsourceDuplicateReference { name, target_names })?;