Function plan_create_source

src/sql/src/plan/statement/ddl.rs:751–1018

(
    scx: &StatementContext,
    mut stmt: CreateSourceStatement<Aug>,
)

}

pub fn plan_create_source(
    scx: &StatementContext,
    mut stmt: CreateSourceStatement<Aug>,
) -> Result<Plan, PlanError> {
    let CreateSourceStatement {
        name,
        in_cluster: _,
        col_names,
        connection: source_connection,
        envelope,
        if_not_exists,
        format,
        key_constraint,
        include_metadata,
        with_options,
        external_references: referenced_subsources,
        progress_subsource,
    } = &stmt;

    mz_ore::soft_assert_or_log!(
        referenced_subsources.is_none(),
        "referenced subsources must be cleared in purification"
    );

    let force_source_table_syntax = scx.catalog.system_vars().enable_create_table_from_source()
        && scx.catalog.system_vars().force_source_table_syntax();

    // If the new source table syntax is forced all the options related to the primary
    // source output should be un-set.
    if force_source_table_syntax {
        if envelope.is_some() || format.is_some() || !include_metadata.is_empty() {
            Err(PlanError::UseTablesForSources(
                "CREATE SOURCE (ENVELOPE|FORMAT|INCLUDE)".to_string(),
            ))?;
        }
    }

    let envelope = envelope.clone().unwrap_or(ast::SourceEnvelope::None);

    if !matches!(source_connection, CreateSourceConnection::Kafka { .. })
        && include_metadata
            .iter()
            .any(|sic| matches!(sic, SourceIncludeMetadata::Headers { .. }))
    {
        // TODO(guswynn): should this be `bail_unsupported!`?
        sql_bail!("INCLUDE HEADERS with non-Kafka sources not supported");
    }
    if !matches!(
        source_connection,
        CreateSourceConnection::Kafka { .. } | CreateSourceConnection::LoadGenerator { .. }
    ) && !include_metadata.is_empty()
    {
        bail_unsupported!("INCLUDE metadata with non-Kafka sources");
    }

    if !include_metadata.is_empty()
        && !matches!(
            envelope,