Repository: risingwavelabs/risingwave
File: src/sqlparser/src/parser.rs (Rust)

Code before:
                AlterSinkOperation::SetStreamingEnableUnalignedJoin { enable: value }
            } else if self.parse_keyword(Keyword::PARALLELISM) {
                if self.expect_keyword(Keyword::TO).is_err()
                    && self.expect_token(&Token::Eq).is_err()
                {
                    return self.expected("TO or = after ALTER TABLE SET PARALLELISM");
                }

                let value = self.parse_set_variable()?;
                let deferred = self.parse_keyword(Keyword::DEFERRED);

                AlterSinkOperation::SetParallelism {
                    parallelism: value,
                    deferred,
                }
            } else if let Some(rate_limit) = self.parse_alter_sink_rate_limit()? {
                AlterSinkOperation::SetSinkRateLimit { rate_limit }
            } else if let Some(rate_limit) = self.parse_alter_backfill_rate_limit()? {
                AlterSinkOperation::SetBackfillRateLimit { rate_limit }
            } else if self.parse_keyword(Keyword::CONFIG) {
                let entries = self.parse_options()?;
                AlterSinkOperation::SetConfig { entries }
            } else {
                return self.expected(
                    "SCHEMA/PARALLELISM/SINK_RATE_LIMIT/STREAMING_ENABLE_UNALIGNED_JOIN/CONFIG after SET",
                );
            }
        } else if self.parse_keyword(Keyword::RESET) {
            if self.parse_keyword(Keyword::CONFIG) {
                let keys = self.parse_parenthesized_object_name_list()?;
                AlterSinkOperation::ResetConfig { keys }
            } else {
                return self.expected("CONFIG after RESET");
            }
        } else if self.parse_keywords(&[Keyword::SWAP, Keyword::WITH]) {
            let target_sink = self.parse_object_name()?;
            AlterSinkOperation::SwapRenameSink { target_sink }
        } else if self.parse_keyword(Keyword::CONNECTOR) {
            let changed_props = self.parse_with_properties()?;
            AlterSinkOperation::AlterConnectorProps {
                alter_props: changed_props,
            }
        } else {
            return self
                .expected("RENAME or OWNER TO or SET or RESET or CONNECTOR WITH after ALTER SINK");
        };

        Ok(Statement::AlterSink {
            name: sink_name,
            operation,
        })

Review comment (suggestion):
The error message does not include BACKFILL_RATE_LIMIT in the list of expected options after SET. Since BACKFILL_RATE_LIMIT parsing was added at line 3632, the error message should be updated to include this new option to help users understand what operations are available.

Code after:
                let schema_name = self.parse_object_name()?;
                AlterSinkOperation::SetSchema {
                    new_schema_name: schema_name,
                }
            } else if self.parse_word("STREAMING_ENABLE_UNALIGNED_JOIN") {
                self.expect_keyword(Keyword::TO)?;
                let value = self.parse_boolean()?;
                AlterSinkOperation::SetStreamingEnableUnalignedJoin { enable: value }
            } else if self.parse_keyword(Keyword::PARALLELISM) {
                if self.expect_keyword(Keyword::TO).is_err()
                    && self.expect_token(&Token::Eq).is_err()
                {
                    return self.expected("TO or = after ALTER TABLE SET PARALLELISM");
                }

                let value = self.parse_set_variable()?;
                let deferred = self.parse_keyword(Keyword::DEFERRED);

                AlterSinkOperation::SetParallelism {
                    parallelism: value,
                    deferred,
                }
            } else if let Some(rate_limit) = self.parse_alter_sink_rate_limit()? {
                AlterSinkOperation::SetSinkRateLimit { rate_limit }
            } else if let Some(rate_limit) = self.parse_alter_backfill_rate_limit()? {
                AlterSinkOperation::SetBackfillRateLimit { rate_limit }
            } else if self.parse_keyword(Keyword::CONFIG) {
                let entries = self.parse_options()?;
                AlterSinkOperation::SetConfig { entries }
            } else {
                return self.expected(
                    "SCHEMA/PARALLELISM/SINK_RATE_LIMIT/BACKFILL_RATE_LIMIT/STREAMING_ENABLE_UNALIGNED_JOIN/CONFIG after SET",
                );
            }
        } else if self.parse_keyword(Keyword::RESET) {
            if self.parse_keyword(Keyword::CONFIG) {
                let keys = self.parse_parenthesized_object_name_list()?;
                AlterSinkOperation::ResetConfig { keys }
            } else {
                return self.expected("CONFIG after RESET");
            }
        } else if self.parse_keywords(&[Keyword::SWAP, Keyword::WITH]) {
            let target_sink = self.parse_object_name()?;
            AlterSinkOperation::SwapRenameSink { target_sink }
        } else if self.parse_keyword(Keyword::CONNECTOR) {
            let changed_props = self.parse_with_properties()?;
            AlterSinkOperation::AlterConnectorProps {
                alter_props: changed_props,
            }
        } else {
            return self