Function purify_create_table_from_source

src/sql/src/pure.rs:1759–2259

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

}

async fn purify_create_table_from_source(
    catalog: impl SessionCatalog,
    mut stmt: CreateTableFromSourceStatement<Aug>,
    storage_configuration: &StorageConfiguration,
) -> Result<PurifiedStatement, PlanError> {
    let scx = StatementContext::new(None, &catalog);
    let CreateTableFromSourceStatement {
        name: _,
        columns,
        constraints,
        source: source_name,
        if_not_exists: _,
        external_reference,
        format,
        envelope,
        include_metadata: _,
        with_options,
    } = &mut stmt;

    // Columns and constraints cannot be specified by the user but will be populated below.
    if matches!(columns, TableFromSourceColumns::Defined(_)) {
        sql_bail!("CREATE TABLE .. FROM SOURCE column definitions cannot be specified directly");
    }
    if !constraints.is_empty() {
        sql_bail!(
            "CREATE TABLE .. FROM SOURCE constraint definitions cannot be specified directly"
        );
    }

    // Get the source item
    let item = match scx.get_item_by_resolved_name(source_name) {
        Ok(item) => item,
        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 unresolved_source_name: UnresolvedItemName = source_name.full_item_name().clone().into();

    let crate::plan::statement::ddl::TableFromSourceOptionExtracted {
        text_columns,
        exclude_columns,
        retain_history: _,
        details,
        partition_by: _,
        seen: _,
    } = with_options.clone().try_into()?;
    if details.is_some() {
        sql_bail!("DETAILS option cannot be explicitly set");
    }

    // Our text column values are unqualified (just column names), but the purification methods below
    // expect to match the fully-qualified names against the full set of tables in upstream, so we