Support throttling an already created source/table

[hzxa21]

Currently user can use SET RW_STREAMING_RATE_LIMIT = <rate_limit_per_actor> to rate limit source/table created in the session but we don't have a way to throttle a specific source/table if it is not created with SET RW_STREAMING_RATE_LIMIT.

This is useful when there are data accumulated in the source and processing these data in one barrier without throttling will cause endless OOM.

[hzxa21]

cc @tabVersion @fuyufjh @chenzl25 @kwannoel

[hzxa21]

@tabVersion and I was thinking whether we can easily support throttling an already created source by introducing a config to limit "number of messages allowed" per barrier in source. It seems that this needs a bigger refactoring and I think that we can adopt the existing FlowControlExeuctor approach.

[fuyufjh]

Agree.

In my mind, the user-interface is via altering a property of materialized view. Previsouly discussed in #11929

[chenzl25]

Technically, we can modify the rate_limit property in the ChainNode of the Mv and restart the cluster to make it work. That is the simplest way.

[tabVersion]

Technically, we can modify the rate_limit property in the ChainNode of the Mv and restart the cluster to make it work. That is the simplest way.

I think making it an individual executor will give us more freedom placing it anywhere in the graph if we want to make some quick fix. I am drafting a rfc explaining how what the expected behavior is.

[fuyufjh]

Technically, we can modify the rate_limit property in the ChainNode of the Mv and restart the cluster to make it work. That is the simplest way.

I think making it an individual executor will give us more freedom placing it anywhere in the graph if we want to make some quick fix. I am drafting a rfc explaining how what the expected behavior is.

FYI. This part is perhaps done in #11919 and #12295

[tabVersion]

Technically, we can modify the rate_limit property in the ChainNode of the Mv and restart the cluster to make it work. That is the simplest way.

I think making it an individual executor will give us more freedom placing it anywhere in the graph if we want to make some quick fix. I am drafting a rfc explaining how what the expected behavior is.

FYI. This part is perhaps done in #11919 and #12295

That's true. The work left is on the ChainNode and the frontend, giving more ways to modify the control rate.

[BugenZhao]

I'm unsure if it's really necessary to make it so "general", that is, utilizing the Alter MV approach to achieve the rate modification. From my prospective, we may simply wrap each executor with a RateLimitExecutor without the awareness in the persisted stream graph. A (developer) system parameter will encode rate limit settings for all actors (if present), which is subscribed by those RateLimitExecutor.

[tabVersion]

I'm unsure if it's really necessary to make it so "general", that is, utilizing the Alter MV approach to achieve the rate modification. From my prospective, we may simply wrap each executor with a RateLimitExecutor without the awareness in the persisted stream graph. A (developer) system parameter will encode rate limit settings for all actors (if present), which is subscribed by those RateLimitExecutor.

I don't think it is necessary in most cases but it is essential to reduce the data into the cluster, preventing from OOM loop.
let's make it a dev tool, a risectl func, rather than a common prop in the first version.

[kwannoel]

Hi, seems @tabVersion is working on this issue. Could you please share more of the overall direction you will take?

[BugenZhao]

Could you please share more of the overall direction you will take?

+1. So what's the eventual plan for updating the flow control rate online?

[tabVersion]

Could you please share more of the overall direction you will take?

+1. So what's the eventual plan for updating the flow control rate online?

I am proposing a config change solution, risectl will send a Throttle Command to apply throttle args on given table_idorsource_id related actors. PR #13166

[tabVersion]

As FlowControl Executor does not require an additional actor_id, I think all Source exec (including fetch exec) and Chain exec will get the following FlowControl exec after #13057.
Thus, the throttle config change proposed in #13166 is also valid for existing tables(and requires an upgrade)

[kwannoel]

I'm unsure if it's really necessary to make it so "general", that is, utilizing the Alter MV approach to achieve the rate modification. From my prospective, we may simply wrap each executor with a RateLimitExecutor without the awareness in the persisted stream graph. A (developer) system parameter will encode rate limit settings for all actors (if present), which is subscribed by those RateLimitExecutor.

I'm thinking if we can let the system parameter be granular enough. Seems kind of weird, because system parameter is system-wide setting.

Whereas here, we may want to specify rate limit for a specific stream job, rather than the whole stream graph. So config change seems to be more reasonable..

[kwannoel]

Actually I think the two approach can be compatible? Both have their usecases.

• System parameter to throttle entire stream graph, if we don't know error source, and just want to make it stable ASAP.
• Config change once we have narrowed down the source.

Within the flow control executor we have the following behaviour:

1. Some meta endpoint which can temporarily throttle throughput of entire stream graph or stop it as suggested by @yezizp2012 . I don't think we need a system variable per-se. Because system variable is persisted, and makes the entire stream graph follow some rate_limit.
2. Config change support as done by @tabVersion in feat: use risectl to throttle Source and Chain by changing FlowControl params via config change #13166

[tabVersion]

will continue to push forward after closing #14384

[tabVersion]

will continue moving the throttle inside source exec and backfill exec

[tabVersion]

#15948 close as completed