Method parse_create_kafka_sink

src/sql-parser/src/parser.rs:3512–3558

(
        &mut self,
        name: Option<UnresolvedItemName>,
        in_cluster: Option<RawClusterName>,
        from: RawItemName,
        if_not_exists: bool,
        connection: CreateSinkConnect

    }

    fn parse_create_kafka_sink(
        &mut self,
        name: Option<UnresolvedItemName>,
        in_cluster: Option<RawClusterName>,
        from: RawItemName,
        if_not_exists: bool,
        connection: CreateSinkConnection<Raw>,
    ) -> Result<CreateSinkStatement<Raw>, ParserError> {
        let format = match &self.parse_one_of_keywords(&[KEY, FORMAT]) {
            Some(KEY) => {
                self.expect_keyword(FORMAT)?;
                let key = self.parse_format()?;
                self.expect_keywords(&[VALUE, FORMAT])?;
                let value = self.parse_format()?;
                Some(FormatSpecifier::KeyValue { key, value })
            }
            Some(FORMAT) => Some(FormatSpecifier::Bare(self.parse_format()?)),
            Some(_) => unreachable!("parse_one_of_keywords returns None for this"),
            None => None,
        };
        let envelope = if self.parse_keyword(ENVELOPE) {
            Some(self.parse_sink_envelope()?)
        } else {
            None
        };

        let with_options = if self.parse_keyword(WITH) {
            self.expect_token(&Token::LParen)?;
            let options = self.parse_comma_separated(Parser::parse_create_sink_option)?;
            self.expect_token(&Token::RParen)?;
            options
        } else {
            vec![]
        };

        Ok(CreateSinkStatement {
            name,
            in_cluster,
            from,
            connection,
            format,
            envelope,
            mode: None,
            if_not_exists,
            with_options,
        })
    }

    fn parse_create_sink(&mut self) -> Result<Statement<Raw>, ParserError> {
        self.expect_keyword(SINK)?;