source_rate_limit doesn't work for file source

[xxchan]

Describe the bug

FsFetch is not handled in update_source_rate_limit_by_source_id

Error message/log

No response

To Reproduce

Add source_rate_limit = 0 in e2e_test/source_inline/fs/posix_fs.slt, you can see data still produced

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response

[kwannoel]

What's the corresponding streaming source executor for it? Is it the fetch_executor?

Edit: Yes it is.

[fuyufjh]

Is someone working on this? cc @kwannoel @tabVersion

[kwannoel]

Is someone working on this? cc @kwannoel @tabVersion

I'll check with @tabVersion, no bandwidth to work on this recently.

Edit: He will manage this issue. Many thanks 🙏.

[xxchan]

Add link for original customer message. Please notify them after the fix.

https://risingwave-community.slack.com/archives/C06LMDNJRMK/p1730981695290009?thread_ts=1730973353.631599&cid=C06LMDNJRMK

[tabVersion]

Is someone working on this? cc @kwannoel @tabVersion

I'll check with @tabVersion, no bandwidth to work on this recently.

Edit: He will manage this issue. Many thanks 🙏.

I am doing a quick check on code. Will fix this ASAP.

[tabVersion]

#16472 seems to forget apply the same change as StreamSource to StreamFsFetch

After the simple fix, create table ... source_rate_limit = 0 ... format ... encode ... can work but alter cannot.

---

Let me explain further on the behavior and why we meet this bug.

take this sql as example

dev=> CREATE TABLE diamonds (
    carat FLOAT,
    cut TEXT,
    color TEXT,
    depth FLOAT,
) WITH (
    connector = 'posix_fs',
    match_pattern = 'data*.csv',
    posix_fs.root = 'e2e_test/source_inline/fs/data',
    source_rate_limit = 0
) FORMAT PLAIN ENCODE CSV ( without_header = 'false', delimiter = ',');
CREATE_TABLE
dev=>
dev=> alter table diamonds set source_rate_limit to default ;
ALTER_SOURCE

obvervations:

• Fs fetch exec starts with actor_id [9, 10, 11, 12] and fs list exec starts with actor_id 13 as singleton.
• When handling alter command, meta sends Mutation to actor 13 instead of [9, 10, 11, 12]
  • run command: {3: {13: None}} stands for doing Mutation::Throttle to actor 13 and change rate_limit to None

reason:

• FS source has a sightly different compute graph as MQ source.
  • Fs list exec in FS source takes the position of source exec in MQ source. We manually add fs fetch exec behind the fs fetch exec.
• Meta treats the FS source and MQ source the same, so it alwasys tries to alter actor the fs list exec instead of the fetch ones.
• So we need to tell if the source is FS source or MQ source when handling alter source_rate_limit.