Function get_encoding_inner

src/sql/src/plan/statement/ddl.rs:2315–2539

(
    scx: &StatementContext,
    format: &Format<Aug>,
)

}

fn get_encoding_inner(
    scx: &StatementContext,
    format: &Format<Aug>,
) -> Result<SourceDataEncoding<ReferencedConnection>, PlanError> {
    let value = match format {
        Format::Bytes => DataEncoding::Bytes,
        Format::Avro(schema) => {
            let Schema {
                key_schema,
                value_schema,
                key_reference_schemas,
                value_reference_schemas,
                wire_format,
            } = match schema {
                // TODO(jldlaughlin): we need a way to pass in primary key information
                // when building a source from a string or file.
                AvroSchema::InlineSchema {
                    schema: ast::Schema { schema },
                    with_options,
                } => {
                    let AvroSchemaOptionExtracted {
                        confluent_wire_format,
                        ..
                    } = with_options.clone().try_into()?;
                    let wire_format = if confluent_wire_format {
                        WireFormat::Confluent { registry: None }
                    } else {
                        WireFormat::None
                    };
                    Schema {
                        key_schema: None,
                        value_schema: schema.clone(),
                        key_reference_schemas: vec![],
                        value_reference_schemas: vec![],
                        wire_format,
                    }
                }
                AvroSchema::Csr {
                    csr_connection:
                        CsrConnectionAvro {
                            connection,
                            seed,
                            key_strategy: _,
                            value_strategy: _,
                        },
                } => {
                    let item = scx.get_item_by_resolved_name(&connection.connection)?;
                    let csr_connection = match item.connection()? {
                        Connection::Csr(_) => item.id(),
                        _ => {
                            sql_bail!(
                                "{} is not a Confluent Schema Registry connection",
                                scx.catalog
                                    .resolve_full_name(item.name())
                                    .to_string()
                                    .quoted()
                            )
                        }