Function plan_webhook_validate_using

src/sql/src/plan/query.rs:1094–1249

Plans an expression in the CHECK position of a `CREATE SOURCE ... FROM WEBHOOK`.

(
    scx: &StatementContext,
    validate_using: CreateWebhookSourceCheck<Aug>,
)

/// Plans an expression in the CHECK position of a `CREATE SOURCE ... FROM WEBHOOK`.
pub fn plan_webhook_validate_using(
    scx: &StatementContext,
    validate_using: CreateWebhookSourceCheck<Aug>,
) -> Result<WebhookValidation, PlanError> {
    let qcx = QueryContext::root(scx, QueryLifetime::Source);

    let CreateWebhookSourceCheck {
        options,
        using: mut expr,
    } = validate_using;

    let mut column_typs = vec![];
    let mut column_names = vec![];

    let (bodies, headers, secrets) = options
        .map(|o| (o.bodies, o.headers, o.secrets))
        .unwrap_or_default();

    // Append all of the bodies so they can be used in the expression.
    let mut body_tuples = vec![];
    for CreateWebhookSourceBody { alias, use_bytes } in bodies {
        let scalar_type = use_bytes
            .then_some(SqlScalarType::Bytes)
            .unwrap_or(SqlScalarType::String);
        let name = alias
            .map(|a| a.into_string())
            .unwrap_or_else(|| "body".to_string());

        column_typs.push(SqlColumnType {
            scalar_type,
            nullable: false,
        });
        column_names.push(name);

        // Store the column index so we can be sure to provide this body correctly.
        let column_idx = column_typs.len() - 1;
        // Double check we're consistent with column names.
        assert_eq!(
            column_idx,
            column_names.len() - 1,
            "body column names and types don't match"
        );
        body_tuples.push((column_idx, use_bytes));
    }

    // Append all of the headers so they can be used in the expression.
    let mut header_tuples = vec![];

    for CreateWebhookSourceHeader { alias, use_bytes } in headers {
        let value_type = use_bytes
            .then_some(SqlScalarType::Bytes)
            .unwrap_or(SqlScalarType::String);
        let name = alias
            .map(|a| a.into_string())
            .unwrap_or_else(|| "headers".to_string());

        column_typs.push(SqlColumnType {
            scalar_type: SqlScalarType::Map {