Function plan_create_webhook_source

src/sql/src/plan/statement/ddl.rs:560–749

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

);

pub fn plan_create_webhook_source(
    scx: &StatementContext,
    mut stmt: CreateWebhookSourceStatement<Aug>,
) -> Result<Plan, PlanError> {
    if stmt.is_table {
        scx.require_feature_flag(&ENABLE_CREATE_TABLE_FROM_SOURCE)?;
    }

    // We will rewrite the cluster if one is not provided, so we must use the `in_cluster` value
    // we plan to normalize when we canonicalize the create statement.
    let in_cluster = source_sink_cluster_config(scx, &mut stmt.in_cluster)?;
    let create_sql =
        normalize::create_statement(scx, Statement::CreateWebhookSource(stmt.clone()))?;

    let CreateWebhookSourceStatement {
        name,
        if_not_exists,
        body_format,
        include_headers,
        validate_using,
        is_table,
        // We resolved `in_cluster` above, so we want to ignore it here.
        in_cluster: _,
    } = stmt;

    let validate_using = validate_using
        .map(|stmt| query::plan_webhook_validate_using(scx, stmt))
        .transpose()?;
    if let Some(WebhookValidation { expression, .. }) = &validate_using {
        // If the validation expression doesn't reference any part of the request, then we should
        // return an error because it's almost definitely wrong.
        if !expression.contains_column() {
            return Err(PlanError::WebhookValidationDoesNotUseColumns);
        }
        // Validation expressions cannot contain unmaterializable functions, except `now()`. We
        // allow calls to `now()` because some webhook providers recommend rejecting requests that
        // are older than a certain threshold.
        if expression.contains_unmaterializable_except(&[UnmaterializableFunc::CurrentTimestamp]) {
            return Err(PlanError::WebhookValidationNonDeterministic);
        }
    }

    let body_format = match body_format {
        Format::Bytes => WebhookBodyFormat::Bytes,
        Format::Json { array } => WebhookBodyFormat::Json { array },
        Format::Text => WebhookBodyFormat::Text,
        // TODO(parkmycar): Make an issue to support more types, or change this to NeverSupported.
        ty => {
            return Err(PlanError::Unsupported {
                feature: format!("{ty} is not a valid BODY FORMAT for a WEBHOOK source"),
                discussion_no: None,
            });
        }
    };

    let mut column_ty = vec![
        // Always include the body of the request as the first column.
        SqlColumnType {