Function purify_source_format

src/sql/src/pure.rs:2266–2294

(
    catalog: &dyn SessionCatalog,
    format: &mut Option<FormatSpecifier<Aug>>,
    options: &SourceFormatOptions,
    envelope: &Option<SourceEnvelope>,
    storage_configuration: &StorageConfigur

}

async fn purify_source_format(
    catalog: &dyn SessionCatalog,
    format: &mut Option<FormatSpecifier<Aug>>,
    options: &SourceFormatOptions,
    envelope: &Option<SourceEnvelope>,
    storage_configuration: &StorageConfiguration,
) -> Result<(), PlanError> {
    if matches!(format, Some(FormatSpecifier::KeyValue { .. }))
        && !matches!(options, SourceFormatOptions::Kafka { .. })
    {
        sql_bail!("Kafka sources are the only source type that can provide KEY/VALUE formats")
    }

    match format.as_mut() {
        None => {}
        Some(FormatSpecifier::Bare(format)) => {
            purify_source_format_single(catalog, format, options, envelope, storage_configuration)
                .await?;
        }

        Some(FormatSpecifier::KeyValue { key, value: val }) => {
            purify_source_format_single(catalog, key, options, envelope, storage_configuration)
                .await?;
            purify_source_format_single(catalog, val, options, envelope, storage_configuration)
                .await?;
        }
    }
    Ok(())
}

async fn purify_source_format_single(
    catalog: &dyn SessionCatalog,