feat: add SQL support to alter fragment parallelism

[shanicky]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Summary
This PR introduces SQL support for altering the parallelism of streaming fragments via the new ALTER FRAGMENT ... SET PARALLELISM statement, enabling fine-grained runtime adjustment of fragment execution. It extends the SQL, frontend, meta service, and stream manager layers to support direct fragment-level parallelism changes, and refactors related APIs for improved clarity and maintainability.

Details

• Adds SQL parser, binder, and handler support for ALTER FRAGMENT <fragment_id> SET PARALLELISM = <value>, allowing users to modify fragment parallelism (fixed or adaptive) at runtime for specific fragments.
• Extends the meta service API (gRPC and protobuf definitions) with new request/response types and a dedicated RPC to handle fragment parallelism changes, ensuring end-to-end support from frontend to meta.
• Implements fragment-level rescheduling logic in both the DDL controller and stream manager, allowing targeted scaling and operational flexibility for streaming fragments.
• Refactors and cleans up internal APIs, removing obsolete types and exposing more direct runtime information for streaming jobs.
• Updates the frontend meta client trait and implementation to include the new RPC method, and amends test/mock clients for compatibility.
• Ensures robust input validation and error handling for parallelism settings, accepting only supported (fixed/adaptive) configurations and reporting errors for others.

Checklist

• I have written necessary rustdoc comments.

Documentation

• My PR needs documentation updates.

Release note

Added an SQL interface for alter fragment {fragmnt_id} set parallelism = {parallelism}

[graphite-app]

Looks like this PR extends new SQL syntax or updates existing ones. Make sure that:

• Test cases about the new/updated syntax are added in src/sqlparser/tests/testdata. Especially, double check the formatted_sql is still a valid SQL #20713
• The meaning of each enum variant is documented in PR description. Additionally, document what it means when each optional clause is omitted.

[Copilot]

Pull Request Overview

This PR adds support for altering fragment parallelism through a new ALTER FRAGMENT <id> SET PARALLELISM = <value> SQL command. This extends the existing ALTER FRAGMENT functionality, which previously only supported altering backfill rate limits.

Key changes:

• Added ALTER_FRAGMENT statement type to the PostgreSQL wire protocol response types
• Extended the SQL parser to handle SET PARALLELISM operation alongside existing SET RATE_LIMIT operation
• Implemented backend handlers and RPC endpoints for altering fragment parallelism
• Added test coverage for the new SQL syntax

Reviewed Changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
src/utils/pgwire/src/pg_response.rs	Adds ALTER_FRAGMENT statement type enum variant and mapping
src/sqlparser/src/ast/ddl.rs	Adds SetParallelism variant to AlterFragmentOperation enum with Display impl
src/sqlparser/src/parser.rs	Extends parser to handle SET PARALLELISM in addition to SET RATE_LIMIT
src/sqlparser/tests/sqlparser_common.rs	Adds test for parsing ALTER FRAGMENT SET PARALLELISM syntax
proto/ddl_service.proto	Defines new RPC messages for alter fragment parallelism requests
src/rpc_client/src/meta_client.rs	Adds client method for calling alter fragment parallelism RPC
src/meta/service/src/ddl_service.rs	Implements RPC service handler for alter fragment parallelism
src/meta/src/rpc/ddl_controller.rs	Adds DDL controller method to delegate to stream manager
src/meta/src/stream/stream_manager.rs	Implements core logic for rescheduling fragments with new parallelism
src/meta/src/stream/scale.rs	Improves error message for missing fragment entry in reschedule request
src/frontend/src/handler/mod.rs	Routes ALTER FRAGMENT SET PARALLELISM statements to appropriate handler
src/frontend/src/handler/alter_parallelism.rs	Implements frontend handler for alter fragment parallelism
src/frontend/src/meta_client.rs	Adds frontend meta client interface method
src/frontend/src/test_utils.rs	Adds unimplemented stub for mock frontend meta client

Comments suppressed due to low confidence (2)

src/sqlparser/src/parser.rs:3835

• The error message is redundant with the word 'expected' appearing twice. Change to 'RATE_LIMIT after SET' or 'expected RATE_LIMIT or PARALLELISM after ALTER FRAGMENT SET'.

        if !self.parse_word("RATE_LIMIT") {
            return self.expected("expected RATE_LIMIT after SET");

src/sqlparser/src/parser.rs:3838

• The error message should be consistent with the full context. Change to 'TO or = after ALTER FRAGMENT SET RATE_LIMIT' to match the pattern used in line 3819 and other similar error messages in the file.

        if self.expect_keyword(Keyword::TO).is_err() && self.expect_token(&Token::Eq).is_err() {
            return self.expected("TO or = after RATE_LIMIT");