Function purify_statement

src/sql/src/pure.rs:285–322

Purifies a statement, removing any dependencies on external state. See the section on [purification](crate#purification) in the crate documentation for details. Note that this doesn't handle CREATE MATERIALIZED VIEW, which is handled by [purify_create_materialized_view_options] instead. This could be made more consistent by a refactoring discussed here: <https://github.com/MaterializeInc/materia

(
    catalog: impl SessionCatalog,
    now: u64,
    stmt: Statement<Aug>,
    storage_configuration: &StorageConfiguration,
)

/// This could be made more consistent by a refactoring discussed here:
/// <https://github.com/MaterializeInc/materialize/pull/23870#discussion_r1435922709>
pub async fn purify_statement(
    catalog: impl SessionCatalog,
    now: u64,
    stmt: Statement<Aug>,
    storage_configuration: &StorageConfiguration,
) -> (Result<PurifiedStatement, PlanError>, Option<ClusterId>) {
    match stmt {
        Statement::CreateSource(stmt) => {
            let cluster_id = stmt.in_cluster.as_ref().map(|cluster| cluster.id.clone());
            (
                purify_create_source(catalog, now, stmt, storage_configuration).await,
                cluster_id,
            )
        }
        Statement::AlterSource(stmt) => (
            purify_alter_source(catalog, stmt, storage_configuration).await,
            None,
        ),
        Statement::CreateSink(stmt) => {
            let cluster_id = stmt.in_cluster.as_ref().map(|cluster| cluster.id.clone());
            (
                purify_create_sink(catalog, stmt, storage_configuration).await,
                cluster_id,
            )
        }
        Statement::CreateTableFromSource(stmt) => (
            purify_create_table_from_source(catalog, stmt, storage_configuration).await,
            None,
        ),
        o => (
            Err(internal_err!(
                "unexpected statement type in purification: {:?}",
                o
            )),
            None,
        ),
    }
}

/// Injects `DOC ON` comments into all Avro formats that are using a schema
/// registry by finding all SQL `COMMENT`s that are attached to the sink's