Function purify_alter_source

src/sql/src/pure.rs:1365–1452

On success, returns the details on new subsources and updated 'options' that sequencing expects for handling `ALTER SOURCE` statements.

(
    catalog: impl SessionCatalog,
    stmt: AlterSourceStatement<Aug>,
    storage_configuration: &StorageConfiguration,
)

/// On success, returns the details on new subsources and updated
/// 'options' that sequencing expects for handling `ALTER SOURCE` statements.
async fn purify_alter_source(
    catalog: impl SessionCatalog,
    stmt: AlterSourceStatement<Aug>,
    storage_configuration: &StorageConfiguration,
) -> Result<PurifiedStatement, PlanError> {
    let scx = StatementContext::new(None, &catalog);
    let AlterSourceStatement {
        source_name: unresolved_source_name,
        action,
        if_exists,
    } = stmt;

    // Get name.
    let item = match scx.resolve_item(RawItemName::Name(unresolved_source_name.clone())) {
        Ok(item) => item,
        Err(_) if if_exists => {
            return Ok(PurifiedStatement::PurifiedAlterSource {
                alter_source_stmt: AlterSourceStatement {
                    source_name: unresolved_source_name,
                    action,
                    if_exists,
                },
            });
        }
        Err(e) => return Err(e),
    };

    // Ensure it's an ingestion-based and alterable source.
    let desc = match item.source_desc()? {
        Some(desc) => desc.clone().into_inline_connection(scx.catalog),
        None => {
            sql_bail!("cannot ALTER this type of source")
        }
    };

    let source_name = item.name();

    let resolved_source_name = ResolvedItemName::Item {
        id: item.id(),
        qualifiers: item.name().qualifiers.clone(),
        full_name: scx.catalog.resolve_full_name(source_name),
        print_id: true,
        version: RelationVersionSelector::Latest,
    };

    let partial_name = scx.catalog.minimal_qualification(source_name);

    match action {
        AlterSourceAction::AddSubsources {
            external_references,
            options,
        } => {
            if scx.catalog.system_vars().enable_create_table_from_source()
                && scx.catalog.system_vars().force_source_table_syntax()
            {
                Err(PlanError::UseTablesForSources(
                    "ALTER SOURCE .. ADD SUBSOURCES ..".to_string(),
                ))?;