Method parse_create_sink

src/sql-parser/src/parser.rs:3560–3597

(&mut self)

    }

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

        let mut name = Some(self.parse_item_name()?);

        // Sniff out `CREATE SINK IN CLUSTER <c> ...` and `CREATE SINK FROM
        // <view>...`  and ensure they are parsed as nameless `CREATE SINK`
        // commands.
        //
        // This is a bit gross, but we didn't have the foresight to make
        // `IN` and `FROM` reserved keywords for sink names.
        if (name == Some(UnresolvedItemName::unqualified(ident!("in")))
            && self.peek_keyword(CLUSTER))
            || (name == Some(UnresolvedItemName::unqualified(ident!("from")))
                && !self.peek_keyword(FROM))
        {
            name = None;
            self.prev_token();
        }

        let in_cluster = self.parse_optional_in_cluster()?;
        self.expect_keyword(FROM)?;
        let from = self.parse_raw_name()?;
        self.expect_keyword(INTO)?;
        let connection = self.parse_create_sink_connection()?;

        let statement = match connection {
            conn @ CreateSinkConnection::Kafka { .. } => {
                self.parse_create_kafka_sink(name, in_cluster, from, if_not_exists, conn)
            }
            conn @ CreateSinkConnection::Iceberg { .. } => {
                self.parse_create_iceberg_sink(name, in_cluster, from, if_not_exists, conn)
            }
        }?;

        Ok(Statement::CreateSink(statement))
    }

    /// Parse the name of a CREATE SINK optional parameter
    fn parse_create_sink_option_name(&mut self) -> Result<CreateSinkOptionName, ParserError> {