pub async fn purify_create_source(
    now: u64,
    mut stmt: CreateSourceStatement<Raw>,
    connector_context: ConnectorContext,
) -> Result<CreateSourceStatement<Raw>, anyhow::Error> {
    let CreateSourceStatement {
        connector,
        format,
        envelope,
        with_options,
        include_metadata: _,
        ..
    } = &mut stmt;

    let mut with_options_map = normalize::options(with_options);
    let mut config_options = BTreeMap::new();

    match connector {
        CreateSourceConnector::Kafka(KafkaSourceConnector {
            connector, topic, ..
        }) => {
            let (broker, connector_options) = match connector {
                KafkaConnector::Reference {
                    broker,
                    with_options,
                    connector,
                } => {
                    let broker_url = match broker {
                        Some(url) => url.to_owned(),
                        None => {
                            bail!("Unable to find connector named {}", connector.to_string())
                        }
                    };
                    let options = match with_options {
                        Some(opts) => opts,
                        None => {
                            bail!("Unable to find connector named {}", connector.to_string())
                        }
                    };
                    (broker_url, Some(options))
                }
                KafkaConnector::Inline { broker } => (broker.to_string(), None),
            };
            config_options = if let Some(options) = connector_options {
                options
                    .iter()
                    .map(|(k, v)| (k.to_owned(), v.to_owned()))
                    .collect::<BTreeMap<String, String>>()
            } else {
                // Verify that the provided security options are valid and then test them.
                kafka_util::extract_config(&mut with_options_map)?
            };
            let consumer = kafka_util::create_consumer(
                &broker,
                &topic,
                &config_options,
                connector_context.librdkafka_log_level,
            )
            .await
            .map_err(|e| anyhow!("Failed to create and connect Kafka consumer: {}", e))?;

            // Translate `kafka_time_offset` to `start_offset`.
            match kafka_util::lookup_start_offsets(
                Arc::clone(&consumer),
                &topic,
                &with_options_map,
                now,
            )
            .await?
            {
                Some(start_offsets) => {
                    // Drop `kafka_time_offset`
                    with_options.retain(|val| match val {
                        WithOption { key, .. } => key.as_str() != "kafka_time_offset",
                    });
                    info!("add start_offset {:?}", start_offsets);
                    // Add `start_offset`
                    with_options.push(WithOption {
                        key: Ident::new("start_offset"),
                        value: Some(WithOptionValue::Value(Value::Array(
                            start_offsets
                                .iter()
                                .map(|offset| Value::Number(offset.to_string()))
                                .collect(),
                        ))),
                    });
                }
                None => {}
            }
        }
        CreateSourceConnector::S3 { .. } => {
            let aws_config = normalize::aws_config(&mut with_options_map, None)?;
            validate_aws_credentials(
                &aws_config,
                connector_context.aws_external_id_prefix.as_ref(),
            )
            .await?;
        }
        CreateSourceConnector::Kinesis { arn } => {
            let region = arn
                .parse::<ARN>()
                .context("Unable to parse provided ARN")?
                .region
                .ok_or_else(|| anyhow!("Provided ARN does not include an AWS region"))?;

            let aws_config = normalize::aws_config(&mut with_options_map, Some(region.into()))?;
            validate_aws_credentials(
                &aws_config,
                connector_context.aws_external_id_prefix.as_ref(),
            )
            .await?;
        }
        CreateSourceConnector::Postgres {
            conn,
            publication,
            slot,
            details,
        } => {
            slot.get_or_insert_with(|| {
                format!(
                    "materialize_{}",
                    Uuid::new_v4().to_string().replace('-', "")
                )
            });

            // verify that we can connect upstream and snapshot publication metadata
            let tables = mz_postgres_util::publication_info(&conn, &publication).await?;

            let details_proto = PostgresSourceDetails {
                tables: tables.into_iter().map(|t| t.into()).collect(),
                slot: slot.clone().expect("slot must exist"),
            };
            *details = Some(hex::encode(details_proto.encode_to_vec()));
        }
        CreateSourceConnector::PubNub { .. } => (),
        CreateSourceConnector::Persist { .. } => (),
    }

    purify_source_format(format, connector, &envelope, &config_options, with_options).await?;

    Ok(stmt)
}
/// Purifies a statement, removing any dependencies on external state.
///
/// See the section on [purification](crate#purification) in the crate
/// documentation for details.
///
/// Note that purification is asynchronous, and may take an unboundedly long
/// time to complete. As a result purification does *not* have access to a
/// [`SessionCatalog`](crate::catalog::SessionCatalog), as that would require
/// locking access to the catalog for an unbounded amount of time.
Purifies a statement, removing any dependencies on external state.
See the section on [purification](crate#purification) in the crate
documentation for details.

Note that purification is asynchronous, and may take an unboundedly long
time to complete. As a result purification does *not* have access to a
[`SessionCatalog`](crate::catalog::SessionCatalog), as that would require
locking access to the catalog for an unbounded amount of time.
Purifies a statement removing any dependencies on external state . See the section on [ purification ] ( crate#purification ) in the crate documentation for details . Note that purification is asynchronous and may take an unboundedly long time to complete . As a result purification does * not * have access to a [ ` SessionCatalog ` ] ( crate :: catalog :: SessionCatalog ) as that would require locking access to the catalog for an unbounded amount of time .
pub async fn purify_create_source(
    now: u64,
    mut stmt: CreateSourceStatement<Raw>,
    connector_context: ConnectorContext,
) -> Result<CreateSourceStatement<Raw>, anyhow::Error> {
    let CreateSourceStatement {
        connector,
        format,
        envelope,
        with_options,
        include_metadata: _,
        ..
    } = &mut stmt;
    let mut with_options_map = normalize::options(with_options);
    let mut config_options = BTreeMap::new();
    match connector {
        CreateSourceConnector::Kafka(KafkaSourceConnector {
            connector, topic, ..
        }) => {
            let (broker, connector_options) = match connector {
                KafkaConnector::Reference {
                    broker,
                    with_options,
                    connector,
                } => {
                    let broker_url = match broker {
                        Some(url) => url.to_owned(),
                        None => {
                            bail!("Unable to find connector named {}", connector.to_string())
                        }
                    };
                    let options = match with_options {
                        Some(opts) => opts,
                        None => {
                            bail!("Unable to find connector named {}", connector.to_string())
                        }
                    };
                    (broker_url, Some(options))
                }
                KafkaConnector::Inline { broker } => (broker.to_string(), None),
            };
            config_options = if let Some(options) = connector_options {
                options
                    .iter()
                    .map(|(k, v)| (k.to_owned(), v.to_owned()))
                    .collect::<BTreeMap<String, String>>()
            } else {
                // Verify that the provided security options are valid and then test them.
                kafka_util::extract_config(&mut with_options_map)?
            };
            let consumer = kafka_util::create_consumer(
                &broker,
                &topic,
                &config_options,
                connector_context.librdkafka_log_level,
            )
            .await
            .map_err(|e| anyhow!("Failed to create and connect Kafka consumer: {}", e))?;
            // Translate `kafka_time_offset` to `start_offset`.
            match kafka_util::lookup_start_offsets(
                Arc::clone(&consumer),
                &topic,
                &with_options_map,
                now,
            )
            .await?
            {
                Some(start_offsets) => {
                    // Drop `kafka_time_offset`
                    with_options.retain(|val| match val {
                        WithOption { key, .. } => key.as_str() != "kafka_time_offset",
                    });
                    info!("add start_offset {:?}", start_offsets);
                    // Add `start_offset`
                    with_options.push(WithOption {
                        key: Ident::new("start_offset"),
                        value: Some(WithOptionValue::Value(Value::Array(
                            start_offsets
                                .iter()
                                .map(|offset| Value::Number(offset.to_string()))
                                .collect(),
                        ))),
                    });
                }
                None => {}
            }
        }
        CreateSourceConnector::S3 { .. } => {
            let aws_config = normalize::aws_config(&mut with_options_map, None)?;
            validate_aws_credentials(
                &aws_config,
                connector_context.aws_external_id_prefix.as_ref(),
            )
            .await?;
        }
        CreateSourceConnector::Kinesis { arn } => {
            let region = arn
                .parse::<ARN>()
                .context("Unable to parse provided ARN")?
                .region
                .ok_or_else(|| anyhow!("Provided ARN does not include an AWS region"))?;
            let aws_config = normalize::aws_config(&mut with_options_map, Some(region.into()))?;
            validate_aws_credentials(
                &aws_config,
                connector_context.aws_external_id_prefix.as_ref(),
            )
            .await?;
        }
        CreateSourceConnector::Postgres {
            conn,
            publication,
            slot,
            details,
        } => {
            slot.get_or_insert_with(|| {
                format!(
                    "materialize_{}",
                    Uuid::new_v4().to_string().replace('-', "")
                )
            });
            // verify that we can connect upstream and snapshot publication metadata
            let tables = mz_postgres_util::publication_info(&conn, &publication).await?;
            let details_proto = PostgresSourceDetails {
                tables: tables.into_iter().map(|t| t.into()).collect(),
                slot: slot.clone().expect("slot must exist"),
            };
            *details = Some(hex::encode(details_proto.encode_to_vec()));
        }
        CreateSourceConnector::PubNub { .. } => (),
        CreateSourceConnector::Persist { .. } => (),
    }
    purify_source_format(format, connector, &envelope, &config_options, with_options).await?;
    Ok(stmt)
}
pub async fn purify_create_source ( now : u64 , mut stmt : CreateSourceStatement < Raw > , connector_context : ConnectorContext , ) -> Result < CreateSourceStatement < Raw > , anyhow :: Error > { let CreateSourceStatement { connector , format , envelope , with_options , include_metadata : _ , .. } = & mut stmt ; let mut with_options_map = normalize :: options ( with_options ) ; let mut config_options = BTreeMap :: new ( ) ; match connector { CreateSourceConnector :: Kafka ( KafkaSourceConnector { connector , topic , .. } ) => { let ( broker , connector_options ) = match connector { KafkaConnector :: Reference { broker , with_options , connector , } => { let broker_url = match broker { Some ( url ) => url . to_owned ( ) , None => { bail ! ( " " connector to_string ( ) ) } } ; let options = match with_options { Some ( opts ) => opts , None => { bail ! ( " " connector to_string ( ) ) } } ; ( broker_url , Some ( options ) ) } KafkaConnector :: Inline { broker } => ( broker . to_string ( ) , None ) , } ; config_options = if let Some ( options ) = connector_options { options . iter ( ) . map ( | ( k , v ) | ( k . to_owned ( ) , v . to_owned ( ) ) ) . collect :: < BTreeMap < String , String > > ( ) } else { // Verify that the provided security options are valid and then test them. kafka_util :: extract_config ( & mut with_options_map ) ? } ; let consumer = kafka_util :: create_consumer ( & broker , & topic , & config_options , connector_context . librdkafka_log_level , ) . await . map_err ( | e | anyhow ! ( " " e ) ) ? ; // Translate `kafka_time_offset` to `start_offset`. match kafka_util :: lookup_start_offsets ( Arc :: clone ( & consumer ) , & topic , & with_options_map , now , ) . await ? { Some ( start_offsets ) => { // Drop `kafka_time_offset` with_options . retain ( | val | match val { WithOption { key , .. } => key . as_str ( ) != " " , } ) ; info ! ( " " start_offsets ) ; // Add `start_offset` with_options . push ( WithOption { key : Ident :: new ( " " ) , value : Some ( WithOptionValue :: Value ( Value :: Array ( start_offsets . iter ( ) . map ( | offset | Value :: Number ( offset . to_string ( ) ) ) . collect ( ) , ) ) ) , } ) ; } None => { } } } CreateSourceConnector :: S3 { .. } => { let aws_config = normalize :: aws_config ( & mut with_options_map , None ) ? ; validate_aws_credentials ( & aws_config , connector_context . aws_external_id_prefix . as_ref ( ) , ) . await ? ; } CreateSourceConnector :: Kinesis { arn } => { let region = arn . parse :: < ARN > ( ) . context ( " " ) ? . region . ok_or_else ( | | anyhow ! ( " " ) ) ? ; let aws_config = normalize :: aws_config ( & mut with_options_map , Some ( region . into ( ) ) ) ? ; validate_aws_credentials ( & aws_config , connector_context . aws_external_id_prefix . as_ref ( ) , ) . await ? ; } CreateSourceConnector :: Postgres { conn , publication , slot , details , } => { slot . get_or_insert_with ( | | { format ! ( " " Uuid new_v4 ( ) to_string ( ) replace ( '-' " " ) ) } ) ; // verify that we can connect upstream and snapshot publication metadata let tables = mz_postgres_util :: publication_info ( & conn , & publication ) . await ? ; let details_proto = PostgresSourceDetails { tables : tables . into_iter ( ) . map ( | t | t . into ( ) ) . collect ( ) , slot : slot . clone ( ) . expect ( " " ) , } ; * details = Some ( hex :: encode ( details_proto . encode_to_vec ( ) ) ) ; } CreateSourceConnector :: PubNub { .. } => ( ) , CreateSourceConnector :: Persist { .. } => ( ) , } purify_source_format ( format , connector , & envelope , & config_options , with_options ) . await ? ; Ok ( stmt ) }
Purifies a statement, removing any dependencies on external state.
Purifies a statement removing any dependencies on external state .
now
u64
stmt
CreateSourceStatement<Raw>
connector_context
ConnectorContext
now
u64
stmt
CreateSourceStatement<Raw>
connector_context
ConnectorContext