Method parse_create_kafka_sink_connection

src/sql-parser/src/parser.rs:3956–4005

(
        &mut self,
    )

    }

    fn parse_create_kafka_sink_connection(
        &mut self,
    ) -> Result<CreateSinkConnection<Raw>, ParserError> {
        self.expect_keyword(CONNECTION)?;

        let connection = self.parse_raw_name()?;

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

        // one token of lookahead:
        // * `KEY (` means we're parsing a list of columns for the key
        // * `KEY FORMAT` means there is no key, we'll parse a KeyValueFormat later
        let key =
            if self.peek_keyword(KEY) && self.peek_nth_token(1) != Some(Token::Keyword(FORMAT)) {
                let _ = self.expect_keyword(KEY);
                let key_columns = self.parse_parenthesized_column_list(Mandatory)?;

                let not_enforced = if self.peek_keywords(&[NOT, ENFORCED]) {
                    self.expect_keywords(&[NOT, ENFORCED])?;
                    true
                } else {
                    false
                };
                Some(SinkKey {
                    key_columns,
                    not_enforced,
                })
            } else {
                None
            };

        let headers = if self.parse_keyword(HEADERS) {
            Some(self.parse_identifier()?)
        } else {
            None
        };

        Ok(CreateSinkConnection::Kafka {
            connection,
            options,
            key,
            headers,
        })
    }

    fn parse_create_iceberg_sink_connection(
        &mut self,