Method sequence_alter_source

src/adapter/src/coord/sequencer/inner.rs:3792–4124

(
        &mut self,
        session: &Session,
        plan::AlterSourcePlan {
            item_id,
            ingestion_id,
            action,
        }: plan::AlterSourcePlan,
    )

    #[instrument]
    pub(super) async fn sequence_alter_source(
        &mut self,
        session: &Session,
        plan::AlterSourcePlan {
            item_id,
            ingestion_id,
            action,
        }: plan::AlterSourcePlan,
    ) -> Result<ExecuteResponse, AdapterError> {
        let cur_entry = self.catalog().get_entry(&item_id);
        let cur_source = cur_entry.source().expect("known to be source");

        let create_sql_to_stmt_deps = |coord: &Coordinator, err_cx, create_source_sql| {
            // Parse statement.
            let create_source_stmt = match mz_sql::parse::parse(create_source_sql)
                .expect("invalid create sql persisted to catalog")
                .into_element()
                .ast
            {
                Statement::CreateSource(stmt) => stmt,
                _ => unreachable!("proved type is source"),
            };

            let catalog = coord.catalog().for_system_session();

            // Resolve items in statement
            mz_sql::names::resolve(&catalog, create_source_stmt)
                .map_err(|e| AdapterError::internal(err_cx, e))
        };

        match action {
            plan::AlterSourceAction::AddSubsourceExports {
                subsources,
                options,
            } => {
                const ALTER_SOURCE: &str = "ALTER SOURCE...ADD SUBSOURCES";

                let mz_sql::plan::AlterSourceAddSubsourceOptionExtracted {
                    text_columns: mut new_text_columns,
                    exclude_columns: mut new_exclude_columns,
                    ..
                } = options.try_into()?;

                // Resolve items in statement
                let (mut create_source_stmt, resolved_ids) =
                    create_sql_to_stmt_deps(self, ALTER_SOURCE, cur_entry.create_sql())?;

                // Get all currently referred-to items
                let catalog = self.catalog();
                let curr_references: BTreeSet<_> = catalog
                    .get_entry(&item_id)
                    .used_by()
                    .into_iter()
                    .filter_map(|subsource| {
                        catalog
                            .get_entry(subsource)
                            .subsource_details()
                            .map(|(_id, reference, _details)| reference)