risingwave_frontend/handler/create_source/external_schema/
protobuf.rs1use risingwave_connector::parser::PROTOBUF_MESSAGES_AS_JSONB;
16
17use super::*;
18
19pub async fn extract_protobuf_table_schema(
21 info: &StreamSourceInfo,
22 with_properties: &WithOptionsSecResolved,
23 format_encode_options: &mut BTreeMap<String, String>,
24) -> Result<Vec<ColumnCatalog>> {
25 let parser_config = SpecificParserConfig::new(info, with_properties)?;
26 try_consume_schema_registry_config_from_options(format_encode_options);
27 try_consume_string_from_options(format_encode_options, PROTOBUF_MESSAGES_AS_JSONB);
28 consume_aws_config_from_options(format_encode_options);
29
30 let conf = ProtobufParserConfig::new(parser_config.encoding_config).await?;
31
32 let column_descs = conf.map_to_columns()?;
33
34 Ok(column_descs
35 .into_iter()
36 .map(|col| ColumnCatalog {
37 column_desc: ColumnDesc::from_field_without_column_id(&col),
38 is_hidden: false,
39 })
40 .collect_vec())
41}