fn ast_rewrite_kafka_protobuf_source_text_to_compiled_0_9_13(
    stmt: &mut sql::ast::Statement<Raw>,
) -> Result<(), anyhow::Error> {
    fn compile_proto(schema: &str) -> Result<CsrSeedCompiledEncoding, anyhow::Error> {
        let temp_schema_name: String = Uuid::new_v4().to_simple().to_string();
        let include_dir = tempfile::tempdir()?;
        let schema_path = include_dir.path().join(&temp_schema_name);
        let schema_bytes = strconv::parse_bytes(schema)?;
        fs::write(&schema_path, &schema_bytes)?;

        match Protoc::new()
            .include(include_dir.path())
            .input(schema_path)
            .parse()
        {
            Ok(fds) => {
                let message_name = fds
                    .file
                    .iter()
                    .find(|f| f.get_name() == temp_schema_name)
                    .map(|file| file.message_type.first())
                    .flatten()
                    .map(|message| format!(".{}", message.get_name()))
                    .ok_or_else(|| anyhow!("unable to compile temporary schema"))?;
                let mut schema = String::new();
                strconv::format_bytes(&mut schema, &fds.write_to_bytes()?);
                Ok(CsrSeedCompiledEncoding {
                    schema,
                    message_name,
                })
            }
            Err(e) => {
                lazy_static! {
                    static ref MISSING_IMPORT_ERROR: Regex = Regex::new(
                        r#"protobuf path \\"(?P<reference>.*)\\" is not found in import path"#
                    )
                    .unwrap();
                }

                // Make protobuf import errors more user-friendly.
                if let Some(captures) = MISSING_IMPORT_ERROR.captures(&e.to_string()) {
                    bail!(
                        "unsupported protobuf schema reference {}",
                        &captures["reference"]
                    )
                } else {
                    Err(e)
                }
            }
        }
    }

    fn do_upgrade(seed: &mut CsrSeedCompiledOrLegacy) -> Result<(), anyhow::Error> {
        match seed {
            CsrSeedCompiledOrLegacy::Legacy(CsrSeed {
                key_schema,
                value_schema,
            }) => {
                let key = match key_schema {
                    Some(k) => Some(compile_proto(k)?),
                    None => None,
                };
                *seed = CsrSeedCompiledOrLegacy::Compiled(CsrSeedCompiled {
                    value: compile_proto(value_schema)?,
                    key,
                });
            }
            CsrSeedCompiledOrLegacy::Compiled(_) => (),
        }
        Ok(())
    }

    if let Statement::CreateSource(CreateSourceStatement { format, .. }) = stmt {
        match format {
            CreateSourceFormat::Bare(value) => {
                if let Format::Protobuf(ProtobufSchema::Csr {
                    csr_connector: CsrConnectorProto { seed: Some(s), .. },
                }) = value
                {
                    do_upgrade(s)?;
                }
            }
            CreateSourceFormat::KeyValue { key, value } => {
                if let Format::Protobuf(ProtobufSchema::Csr {
                    csr_connector: CsrConnectorProto { seed: Some(s), .. },
                }) = key
                {
                    do_upgrade(s)?;
                }
                if let Format::Protobuf(ProtobufSchema::Csr {
                    csr_connector: CsrConnectorProto { seed: Some(s), .. },
                }) = value
                {
                    do_upgrade(s)?;
                }
            }
            CreateSourceFormat::None => {}
        }
    }
    Ok(())
}
/// Rewrites Protobuf sources to store the compiled bytes rather than the text
/// of the schema.
Rewrites Protobuf sources to store the compiled bytes rather than the text
of the schema.
Rewrites Protobuf sources to store the compiled bytes rather than the text of the schema .
fn ast_rewrite_kafka_protobuf_source_text_to_compiled_0_9_13(
    stmt: &mut sql::ast::Statement<Raw>,
) -> Result<(), anyhow::Error> {
    fn compile_proto(schema: &str) -> Result<CsrSeedCompiledEncoding, anyhow::Error> {
        let temp_schema_name: String = Uuid::new_v4().to_simple().to_string();
        let include_dir = tempfile::tempdir()?;
        let schema_path = include_dir.path().join(&temp_schema_name);
        let schema_bytes = strconv::parse_bytes(schema)?;
        fs::write(&schema_path, &schema_bytes)?;
        match Protoc::new()
            .include(include_dir.path())
            .input(schema_path)
            .parse()
        {
            Ok(fds) => {
                let message_name = fds
                    .file
                    .iter()
                    .find(|f| f.get_name() == temp_schema_name)
                    .map(|file| file.message_type.first())
                    .flatten()
                    .map(|message| format!(".{}", message.get_name()))
                    .ok_or_else(|| anyhow!("unable to compile temporary schema"))?;
                let mut schema = String::new();
                strconv::format_bytes(&mut schema, &fds.write_to_bytes()?);
                Ok(CsrSeedCompiledEncoding {
                    schema,
                    message_name,
                })
            }
            Err(e) => {
                lazy_static! {
                    static ref MISSING_IMPORT_ERROR: Regex = Regex::new(
                        r#"protobuf path \\"(?P<reference>.*)\\" is not found in import path"#
                    )
                    .unwrap();
                }
                // Make protobuf import errors more user-friendly.
                if let Some(captures) = MISSING_IMPORT_ERROR.captures(&e.to_string()) {
                    bail!(
                        "unsupported protobuf schema reference {}",
                        &captures["reference"]
                    )
                } else {
                    Err(e)
                }
            }
        }
    }
    fn do_upgrade(seed: &mut CsrSeedCompiledOrLegacy) -> Result<(), anyhow::Error> {
        match seed {
            CsrSeedCompiledOrLegacy::Legacy(CsrSeed {
                key_schema,
                value_schema,
            }) => {
                let key = match key_schema {
                    Some(k) => Some(compile_proto(k)?),
                    None => None,
                };
                *seed = CsrSeedCompiledOrLegacy::Compiled(CsrSeedCompiled {
                    value: compile_proto(value_schema)?,
                    key,
                });
            }
            CsrSeedCompiledOrLegacy::Compiled(_) => (),
        }
        Ok(())
    }
    if let Statement::CreateSource(CreateSourceStatement { format, .. }) = stmt {
        match format {
            CreateSourceFormat::Bare(value) => {
                if let Format::Protobuf(ProtobufSchema::Csr {
                    csr_connector: CsrConnectorProto { seed: Some(s), .. },
                }) = value
                {
                    do_upgrade(s)?;
                }
            }
            CreateSourceFormat::KeyValue { key, value } => {
                if let Format::Protobuf(ProtobufSchema::Csr {
                    csr_connector: CsrConnectorProto { seed: Some(s), .. },
                }) = key
                {
                    do_upgrade(s)?;
                }
                if let Format::Protobuf(ProtobufSchema::Csr {
                    csr_connector: CsrConnectorProto { seed: Some(s), .. },
                }) = value
                {
                    do_upgrade(s)?;
                }
            }
            CreateSourceFormat::None => {}
        }
    }
    Ok(())
}
fn ast_rewrite_kafka_protobuf_source_text_to_compiled_0_9_13 ( stmt : & mut sql :: ast :: Statement < Raw > , ) -> Result < ( ) , anyhow :: Error > { fn compile_proto ( schema : & str ) -> Result < CsrSeedCompiledEncoding , anyhow :: Error > { let temp_schema_name : String = Uuid :: new_v4 ( ) . to_simple ( ) . to_string ( ) ; let include_dir = tempfile :: tempdir ( ) ? ; let schema_path = include_dir . path ( ) . join ( & temp_schema_name ) ; let schema_bytes = strconv :: parse_bytes ( schema ) ? ; fs :: write ( & schema_path , & schema_bytes ) ? ; match Protoc :: new ( ) . include ( include_dir . path ( ) ) . input ( schema_path ) . parse ( ) { Ok ( fds ) => { let message_name = fds . file . iter ( ) . find ( | f | f . get_name ( ) == temp_schema_name ) . map ( | file | file . message_type . first ( ) ) . flatten ( ) . map ( | message | format ! ( " " message get_name ( ) ) ) . ok_or_else ( | | anyhow ! ( " " ) ) ? ; let mut schema = String :: new ( ) ; strconv :: format_bytes ( & mut schema , & fds . write_to_bytes ( ) ? ) ; Ok ( CsrSeedCompiledEncoding { schema , message_name , } ) } Err ( e ) => { lazy_static ! { static ref MISSING_IMPORT_ERROR Regex Regex new ( r#"protobuf path \\"(?P<reference>.*)\\" is not found in import path"# ) unwrap ( ) } // Make protobuf import errors more user-friendly. if let Some ( captures ) = MISSING_IMPORT_ERROR . captures ( & e . to_string ( ) ) { bail ! ( " " captures [ " " ] ) } else { Err ( e ) } } } } fn do_upgrade ( seed : & mut CsrSeedCompiledOrLegacy ) -> Result < ( ) , anyhow :: Error > { match seed { CsrSeedCompiledOrLegacy :: Legacy ( CsrSeed { key_schema , value_schema , } ) => { let key = match key_schema { Some ( k ) => Some ( compile_proto ( k ) ? ) , None => None , } ; * seed = CsrSeedCompiledOrLegacy :: Compiled ( CsrSeedCompiled { value : compile_proto ( value_schema ) ? , key , } ) ; } CsrSeedCompiledOrLegacy :: Compiled ( _ ) => ( ) , } Ok ( ( ) ) } if let Statement :: CreateSource ( CreateSourceStatement { format , .. } ) = stmt { match format { CreateSourceFormat :: Bare ( value ) => { if let Format :: Protobuf ( ProtobufSchema :: Csr { csr_connector : CsrConnectorProto { seed : Some ( s ) , .. } , } ) = value { do_upgrade ( s ) ? ; } } CreateSourceFormat :: KeyValue { key , value } => { if let Format :: Protobuf ( ProtobufSchema :: Csr { csr_connector : CsrConnectorProto { seed : Some ( s ) , .. } , } ) = key { do_upgrade ( s ) ? ; } if let Format :: Protobuf ( ProtobufSchema :: Csr { csr_connector : CsrConnectorProto { seed : Some ( s ) , .. } , } ) = value { do_upgrade ( s ) ? ; } } CreateSourceFormat :: None => { } } } Ok ( ( ) ) }
Rewrites Protobuf sources to store the compiled bytes rather than the text
of the schema.
Rewrites Protobuf sources to store the compiled bytes rather than the text of the schema .
stmt
&mut sql::ast::Statement<Raw>
stmt
&mut sql::ast::Statement<Raw>