Method parse_kafka_sink_config_option

src/sql-parser/src/parser.rs:2848–2915

(
        &mut self,
    )

    }

    fn parse_kafka_sink_config_option(
        &mut self,
    ) -> Result<KafkaSinkConfigOption<Raw>, ParserError> {
        let name = match self.expect_one_of_keywords(&[
            COMPRESSION,
            PARTITION,
            PROGRESS,
            TOPIC,
            LEGACY,
            TRANSACTIONAL,
        ])? {
            COMPRESSION => {
                self.expect_keyword(TYPE)?;
                KafkaSinkConfigOptionName::CompressionType
            }
            PARTITION => {
                self.expect_keyword(BY)?;
                let _ = self.consume_token(&Token::Eq);
                return Ok(KafkaSinkConfigOption {
                    name: KafkaSinkConfigOptionName::PartitionBy,
                    value: Some(WithOptionValue::Expr(self.parse_expr()?)),
                });
            }
            PROGRESS => {
                self.expect_keywords(&[GROUP, ID, PREFIX])?;
                KafkaSinkConfigOptionName::ProgressGroupIdPrefix
            }
            TOPIC => {
                match self.parse_one_of_keywords(&[METADATA, PARTITION, REPLICATION, CONFIG]) {
                    None => KafkaSinkConfigOptionName::Topic,
                    Some(METADATA) => {
                        self.expect_keywords(&[REFRESH, INTERVAL])?;
                        KafkaSinkConfigOptionName::TopicMetadataRefreshInterval
                    }
                    Some(PARTITION) => {
                        self.expect_keyword(COUNT)?;
                        KafkaSinkConfigOptionName::TopicPartitionCount
                    }
                    Some(REPLICATION) => {
                        self.expect_keyword(FACTOR)?;
                        KafkaSinkConfigOptionName::TopicReplicationFactor
                    }
                    Some(CONFIG) => KafkaSinkConfigOptionName::TopicConfig,
                    Some(other) => {
                        return parser_err!(
                            self,
                            self.peek_prev_pos(),
                            "unexpected keyword {}",
                            other
                        );
                    }
                }
            }
            TRANSACTIONAL => {
                self.expect_keywords(&[ID, PREFIX])?;
                KafkaSinkConfigOptionName::TransactionalIdPrefix
            }
            LEGACY => {