(
catalog: impl SessionCatalog,
now: u64,
mut create_source_stmt: CreateSourceStatement<Aug>,
storage_configuration: &StorageConfiguration,
)
| 717 | } |
| 718 | |
| 719 | async fn purify_create_source( |
| 720 | catalog: impl SessionCatalog, |
| 721 | now: u64, |
| 722 | mut create_source_stmt: CreateSourceStatement<Aug>, |
| 723 | storage_configuration: &StorageConfiguration, |
| 724 | ) -> Result<PurifiedStatement, PlanError> { |
| 725 | let CreateSourceStatement { |
| 726 | name: source_name, |
| 727 | col_names, |
| 728 | key_constraint, |
| 729 | connection: source_connection, |
| 730 | format, |
| 731 | envelope, |
| 732 | include_metadata, |
| 733 | external_references, |
| 734 | progress_subsource, |
| 735 | with_options, |
| 736 | .. |
| 737 | } = &mut create_source_stmt; |
| 738 | |
| 739 | let uses_old_syntax = !col_names.is_empty() |
| 740 | || key_constraint.is_some() |
| 741 | || format.is_some() |
| 742 | || envelope.is_some() |
| 743 | || !include_metadata.is_empty() |
| 744 | || external_references.is_some() |
| 745 | || progress_subsource.is_some(); |
| 746 | |
| 747 | if let Some(DeferredItemName::Named(_)) = progress_subsource { |
| 748 | sql_bail!("Cannot manually ID qualify progress subsource") |
| 749 | } |
| 750 | |
| 751 | let mut requested_subsource_map = BTreeMap::new(); |
| 752 | |
| 753 | let progress_desc = match &source_connection { |
| 754 | CreateSourceConnection::Kafka { .. } => { |
| 755 | &mz_storage_types::sources::kafka::KAFKA_PROGRESS_DESC |
| 756 | } |
| 757 | CreateSourceConnection::Postgres { .. } => { |
| 758 | &mz_storage_types::sources::postgres::PG_PROGRESS_DESC |
| 759 | } |
| 760 | CreateSourceConnection::SqlServer { .. } => { |
| 761 | &mz_storage_types::sources::sql_server::SQL_SERVER_PROGRESS_DESC |
| 762 | } |
| 763 | CreateSourceConnection::MySql { .. } => { |
| 764 | &mz_storage_types::sources::mysql::MYSQL_PROGRESS_DESC |
| 765 | } |
| 766 | CreateSourceConnection::LoadGenerator { .. } => { |
| 767 | &mz_storage_types::sources::load_generator::LOAD_GEN_PROGRESS_DESC |
| 768 | } |
| 769 | }; |
| 770 | let scx = StatementContext::new(None, &catalog); |
| 771 | |
| 772 | // Depending on if the user must or can use the `CREATE TABLE .. FROM SOURCE` statement |
| 773 | // to add tables after this source is created we might need to enforce that |
| 774 | // auto-generated subsources are created or not created by this source statement. |
| 775 | let reference_policy = if scx.catalog.system_vars().enable_create_table_from_source() |
| 776 | && scx.catalog.system_vars().force_source_table_syntax() |
no test coverage detected