fix: table with fs connector cannot work with rate limit

[tabVersion]

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

What's changed and what's your intention?

resolve #19296

The pr fix two places:

• plan for fs fetch exec does not access rate limit from the catalog (src/frontend/src/optimizer/plan_node/stream_fs_fetch.rs)
• FsFetch Exec is not assigned to the correct source_id when building fragments (src/meta/src/stream/stream_graph/fragment.rs)

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[tabVersion]

another question about the generated plan

---

for table with connector

dev=> explain  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 = ',');
                                                                    QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------------
 StreamMaterialize { columns: [carat, cut, color, depth, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: Overwrite }
 └─StreamRowIdGen { row_id_index: 4 }
   └─StreamUnion { all: true }
     ├─StreamExchange [no_shuffle] { dist: SomeShard }
     │ └─StreamFsFetch { columns: [carat, cut, color, depth, _row_id] }
     │   └─StreamAppendOnlyDedup { dedup_cols: [filename] }
     │     └─StreamExchange { dist: HashShard(filename) }
     │       └─StreamSource { source: diamonds, columns: [filename, last_edit_time, file_size] }
     └─StreamExchange { dist: HashShard(_row_id) }
       └─StreamDml { columns: [carat, cut, color, depth, _row_id] }
         └─StreamSource
(11 rows)

and I get the following 3 fragments

Materialize(MaterializeNode { table_id: 6, ... }), mask: 2
StreamFsFetch(StreamFsFetchNode { node_inner: Some(StreamFsFetch { source_id: 7, ... ) }), mask: 0
Source(SourceNode { source_inner: Some(StreamSource { source_id: 7,  ... }), mask: 129

but for creating source + building mv on the source

dev=>  CREATE source diamonds1 (
    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_SOURCE
dev=> explain create materialized view mv as select * from diamonds1 ;
                                                                   QUERY PLAN
------------------------------------------------------------------------------------------------------------------------------------------------
 StreamMaterialize { columns: [carat, cut, color, depth, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
 └─StreamProject { exprs: [carat, cut, color, depth, _row_id] }
   └─StreamRowIdGen { row_id_index: 4 }
     └─StreamFsFetch { columns: [carat, cut, color, depth, _row_id] }
       └─StreamAppendOnlyDedup { dedup_cols: [filename] }
         └─StreamExchange { dist: HashShard(filename) }
           └─StreamSource { source: diamonds1, columns: [filename, last_edit_time, file_size] }
(7 rows)

I only have two fragments

Materialize(MaterializeNode { table_id: 12, ... }), mask: 2
Source(SourceNode { source_inner: Some(StreamSource { source_id: 11, ...  }), mask: 1

---

is it expected that the 2nd approach have no fragment for StreamFsFetch while the plan contains it? cc @st1page

[st1page]

another question about the generated plan

for table with connector

dev=> explain  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 = ',');
                                                                    QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------------
 StreamMaterialize { columns: [carat, cut, color, depth, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: Overwrite }
 └─StreamRowIdGen { row_id_index: 4 }
   └─StreamUnion { all: true }
     ├─StreamExchange [no_shuffle] { dist: SomeShard }
     │ └─StreamFsFetch { columns: [carat, cut, color, depth, _row_id] }
     │   └─StreamAppendOnlyDedup { dedup_cols: [filename] }
     │     └─StreamExchange { dist: HashShard(filename) }
     │       └─StreamSource { source: diamonds, columns: [filename, last_edit_time, file_size] }
     └─StreamExchange { dist: HashShard(_row_id) }
       └─StreamDml { columns: [carat, cut, color, depth, _row_id] }
         └─StreamSource
(11 rows)

and I get the following 3 fragments

Materialize(MaterializeNode { table_id: 6, ... }), mask: 2
StreamFsFetch(StreamFsFetchNode { node_inner: Some(StreamFsFetch { source_id: 7, ... ) }), mask: 0
Source(SourceNode { source_inner: Some(StreamSource { source_id: 7,  ... }), mask: 129

but for creating source + building mv on the source

dev=>  CREATE source diamonds1 (
    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_SOURCE
dev=> explain create materialized view mv as select * from diamonds1 ;
                                                                   QUERY PLAN
------------------------------------------------------------------------------------------------------------------------------------------------
 StreamMaterialize { columns: [carat, cut, color, depth, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
 └─StreamProject { exprs: [carat, cut, color, depth, _row_id] }
   └─StreamRowIdGen { row_id_index: 4 }
     └─StreamFsFetch { columns: [carat, cut, color, depth, _row_id] }
       └─StreamAppendOnlyDedup { dedup_cols: [filename] }
         └─StreamExchange { dist: HashShard(filename) }
           └─StreamSource { source: diamonds1, columns: [filename, last_edit_time, file_size] }
(7 rows)

I only have two fragments

Materialize(MaterializeNode { table_id: 12, ... }), mask: 2
Source(SourceNode { source_inner: Some(StreamSource { source_id: 11, ...  }), mask: 1

is it expected that the 2nd approach have no fragment for StreamFsFetch while the plan contains it? cc @st1page

That's expected. The first plan's fragments is for multiple data sources like sink into table and DML statments.

[tabVersion]

found a new issue #19346
merge this one first

[xxchan]

This seems still incomplete to me: I guess MV on fs source still won't work?

[tabVersion]

It works but it breaks in another way.

Test locally

dev=> CREATE source 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_SOURCE
dev=> create materialized view mv as select * from diamonds;
CREATE_MATERIALIZED_VIEW
dev=> select * from mv ;
 carat | cut | color | depth
-------+-----+-------+-------
(0 rows)

dev=> ALTER source  diamonds SET source_rate_limit TO DEFAULT;
ALTER_SOURCE
dev=>
dev=> flush;
FLUSH
dev=>  select * from mv ; <- still not consuming 
 carat | cut | color | depth
-------+-----+-------+-------
(0 rows)