Function plan_create_subsource

src/sql/src/plan/statement/ddl.rs:1589–1737

(
    scx: &StatementContext,
    stmt: CreateSubsourceStatement<Aug>,
)

);

pub fn plan_create_subsource(
    scx: &StatementContext,
    stmt: CreateSubsourceStatement<Aug>,
) -> Result<Plan, PlanError> {
    let CreateSubsourceStatement {
        name,
        columns,
        of_source,
        constraints,
        if_not_exists,
        with_options,
    } = &stmt;

    let CreateSubsourceOptionExtracted {
        progress,
        retain_history,
        external_reference,
        text_columns,
        exclude_columns,
        details,
        seen: _,
    } = with_options.clone().try_into()?;

    // This invariant is enforced during purification; we are responsible for
    // creating the AST for subsources as a response to CREATE SOURCE
    // statements, so this would fire in integration testing if we failed to
    // uphold it.
    if !(progress ^ (external_reference.is_some() && of_source.is_some())) {
        bail_internal!(
            "CREATE SUBSOURCE statement must specify either PROGRESS or REFERENCES option"
        );
    }

    let desc = plan_source_export_desc(scx, name, columns, constraints)?;

    let data_source = if let Some(source_reference) = of_source {
        // If the new source table syntax is forced we should not be creating any non-progress
        // subsources.
        if scx.catalog.system_vars().enable_create_table_from_source()
            && scx.catalog.system_vars().force_source_table_syntax()
        {
            Err(PlanError::UseTablesForSources(
                "CREATE SUBSOURCE".to_string(),
            ))?;
        }

        // This is a subsource with the "natural" dependency order, i.e. it is
        // not a legacy subsource with the inverted structure.
        let ingestion_id = *source_reference.item_id();
        let external_reference = external_reference.ok_or_else(|| {
            sql_err!("CREATE SUBSOURCE with REFERENCES requires EXTERNAL REFERENCE option")
        })?;

        // Decode the details option stored on the subsource statement, which contains information
        // created during the purification process.
        let details = details
            .as_ref()
            .ok_or_else(|| internal_err!("source-export subsource missing details"))?;