feat(frontend): support sink backfill rate limit

[kwannoel]

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

What's changed and what's your intention?

1. Decouple ThrottleTarget into ThrottleTarget and ThrottleType. ThrottleType refers to the type of rate limit we want to apply. ThrottleTarget refers to the actual object we are targeting (sink, index, fragment, etc...).
2. This change is required so we can alter both backfill and sink rate limit for sinks.
3. This is because the sink executor and the scan executor can lie on the same fragment, so we need to provide the throttle type at runtime, through the barrier mutation, to make sure only the rate limit type we are targeting gets altered.
4. Proto breaking changes are just for the RPC layer, and do not change the persisted state of rate limit in fragments.

• I avoided doing mass renaming of Throttle -> RateLimit, but we should make this change subsequently for consistency
• I also avoided specializing alter fragment rate limit. After this PR we can have another one to do the specialization to different rate limit types.

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

• My PR needs documentation updates.

Release note

[Copilot]

Pull request overview

This PR introduces support for separate backfill rate limiting on sinks by decoupling ThrottleTarget (what to throttle) from ThrottleType (how to throttle). This allows users to independently control both the sink execution rate and the sink backfill rate.

Key Changes:

• Introduced ThrottleType enum to distinguish between DML, Backfill, Source, and Sink rate limit types
• Added ALTER SINK SET BACKFILL_RATE_LIMIT SQL syntax support
• Updated all executors to check throttle_type before applying rate limits
• Modified RPC interfaces to accept both throttle_target and throttle_type parameters

Reviewed changes

Copilot reviewed 26 out of 26 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
proto/meta.proto	Added ThrottleType enum and updated ApplyThrottleRequest to include both target and type
proto/stream_plan.proto	Added nested ThrottleType enum in ThrottleMutation and added throttle_type field
src/sqlparser/src/ast/ddl.rs	Added SetBackfillRateLimit variant to AlterSinkOperation enum
src/sqlparser/src/parser.rs	Added parsing logic for ALTER SINK SET BACKFILL_RATE_LIMIT syntax
src/frontend/src/meta_client.rs	Updated apply_throttle method signature to accept both target and type
src/frontend/src/test_utils.rs	Updated mock implementation to match new signature
src/frontend/src/handler/mod.rs	Added handler routing for sink backfill rate limit operations
src/frontend/src/handler/alter_streaming_rate_limit.rs	Updated logic to handle different throttle target/type combinations
src/rpc_client/src/meta_client.rs	Updated apply_throttle to pass both target and type parameters
src/meta/service/src/stream_service.rs	Updated throttle application logic to match on (type, target) tuples
src/meta/service/src/ddl_service.rs	Updated to pass ThrottleType::Source when creating throttle commands
src/meta/src/barrier/command.rs	Modified Command::Throttle to include ThrottleType parameter
src/meta/src/barrier/info.rs	Updated pattern matching for throttle commands
src/meta/src/barrier/context/context_impl.rs	Updated pattern matching for throttle commands
src/stream/src/executor/mod.rs	Updated Mutation::Throttle to include throttle_type field and conversion logic
src/stream/src/executor/sink.rs	Added throttle type check to only apply sink rate limits
src/stream/src/executor/dml.rs	Added throttle type check to only apply DML rate limits
src/stream/src/executor/source/source_executor.rs	Added throttle type check to only apply source rate limits
src/stream/src/executor/source/source_backfill_executor.rs	Added throttle type check to only apply backfill rate limits
src/stream/src/executor/source/iceberg_fetch_executor.rs	Added throttle type check to only apply source rate limits
src/stream/src/executor/source/fs_fetch_executor.rs	Added throttle type check to only apply source rate limits
src/stream/src/executor/backfill/no_shuffle_backfill.rs	Added throttle type check to only apply backfill rate limits
src/stream/src/executor/backfill/arrangement_backfill.rs	Added throttle type check to only apply backfill rate limits
src/stream/src/executor/backfill/cdc/cdc_backfill.rs	Added throttle type check to only apply backfill rate limits
src/stream/src/executor/backfill/cdc/cdc_backill_v2.rs	Added throttle type check to only apply backfill rate limits
src/ctl/src/cmd_impl/throttle.rs	Added logic to infer throttle type from target for backward compatibility

[kwannoel]

@copilot please submit a pr to fix the review comments.

[Copilot]

@kwannoel I've opened a new pull request, #24083, to work on those changes. Once the pull request is ready, I'll request review from you.

[Copilot]

@kwannoel I've opened a new pull request, #24111, to work on those changes. Once the pull request is ready, I'll request review from you.

[kwannoel]

@copilot please fix the integration test test_throttle_mv:

�[31;1m        FAIL�[0m [  16.156s] �[35;1mrisingwave_simulation::integration_tests�[0m �[36mthrottle�[0m�[36m::�[0m�[34;1mtest_throttle_mv�[0m
�[31;1m �[0m �[31;1mstdout�[0m �[31;1m───�[0m

    running 1 test
    seed = 1
    Configuration {
        config_path: Temp(
            "/tmp/.tmpA0u7ih",
        ),
        frontend_nodes: 1,
        compute_nodes: 1,
        meta_nodes: 1,
        compactor_nodes: 1,
        compute_node_cores: 4,
        per_session_queries: [],
        compute_resource_groups: {},
    }�[0m
�[31;1m �[0m �[31;1mstderr�[0m �[31;1m───�[0m
    error: unexpected argument '7' found

    Usage: ctl throttle mv [OPTIONS] --id <ID> --throttle-type <THROTTLE_TYPE>

    For more information, try '--help'.�[0m

    (test �[31;1mfailed�[0m with exit code �[1m2�[0m)

�[31;1merror�[0m: test run failed

[Copilot]

@kwannoel I've opened a new pull request, #24190, to work on those changes. Once the pull request is ready, I'll request review from you.

[tabVersion]

we do need some review on rate limit mechanism, distinguish source_rate_limit and backfill_rate_limit are poor designs, TBH. But introducing ThrottleType may have it deeper nested with prev impl.

[tabVersion]

so one for SINK_RATE_LIMIT and one for backfill_rate_limit

[kwannoel]

Yep

[tabVersion]

Type -> Object, as you are rate limit database objects

[kwannoel]

This refers to the type of rate limit we are applying. Source, backfill, sink, dml these are all different types of rate limits.

[tabVersion]

also if using id, there is no need for ThrottleTypeArg. Id is unique in the cluster, using id can find the object and determine the throttle_type.

[kwannoel]

Id is not specified at the executor level, it's done at the fragment level in the current implementation. In such a scenario, consider when sink executor and scan executor are collocated in the same fragment. When altering either rate limit (sink or backfill), both rate limits will be changed, since the provided id is fragment id. You can see that the slt test was changed, because this was incorrect behaviour that existed due to the current design.

[tabVersion]

I dont think ThrottleTarget and ThrottleType are orthogonal in concept.
ThrottleType in your design stands for things you want to apply rate limit, but ThrottleTarget also contain similar concepts.

Maybe we can change to ThrottleObject and ThrottleDirection,
eg. ThrottleObject.sink + ThrottleDirection.out = sink_rate_limit and ThrottleObject.sink + ThrottleDirection.in = backfill_rate_limit

[kwannoel]

If we support streaming rate limit in the future, direction also becomes ambiguous. streaming and source share the same semantics. Further consider altering a cdc table rate limit. Lets say we specify the direction "in". Are we changing the dml rate limit? Or the source rate limit? Or the backfill rate limit? Decoupling the rate limit type from the target makes this unambiguous.

[Chiro11]

Basically LGTM