feat: add rw_actor_splits

[shanicky]

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

What's changed and what's your intention?

This pull request adds the rw_actor_splits system table, which provides the columns actor_id, split_id, source_id, and fragment_id.

dev=> create source s1 (v1 int) with (
  connector = 'kafka',
  properties.bootstrap.server = 'localhost:29092',
  topic = 't',
  scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;
CREATE_SOURCE
dev=> create materialized view m1 as select count(*) from s1 ;
CREATE_MATERIALIZED_VIEW
dev=> SET rw_enable_shared_source TO true;
SET_VARIABLE
dev=> create source s2 (v1 int) with (
  connector = 'kafka', properties.bootstrap.server = 'localhost:29092',
  topic = 't',
  scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;
CREATE_SOURCE
dev=> select * from rw_actor_splits;
 actor_id | split_id | source_id | fragment_id |   fragment_type
----------+----------+-----------+-------------+-------------------
        2 | 0        |         6 |           2 | NON_SHARED_SOURCE
        3 | 1        |         6 |           2 | NON_SHARED_SOURCE
        5 | 2        |         6 |           2 | NON_SHARED_SOURCE
        7 | 0        |        10 |           3 | SHARED_SOURCE
        8 | 1        |        10 |           3 | SHARED_SOURCE
        9 | 2        |        10 |           3 | SHARED_SOURCE
(6 rows)
dev=> create materialized view m2 as select count(*) from s2 ;
CREATE_MATERIALIZED_VIEW
dev=> select * from rw_actor_splits;
 actor_id | split_id | source_id | fragment_id |     fragment_type
----------+----------+-----------+-------------+------------------------
        2 | 0        |         6 |           2 | NON_SHARED_SOURCE
        3 | 1        |         6 |           2 | NON_SHARED_SOURCE
        5 | 2        |         6 |           2 | NON_SHARED_SOURCE
        7 | 0        |        10 |           3 | SHARED_SOURCE
        8 | 1        |        10 |           3 | SHARED_SOURCE
        9 | 2        |        10 |           3 | SHARED_SOURCE
       12 | 0        |        10 |           5 | SHARED_SOURCE_BACKFILL
       13 | 1        |        10 |           5 | SHARED_SOURCE_BACKFILL
       14 | 2        |        10 |           5 | SHARED_SOURCE_BACKFILL
(9 rows)

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• All checks passed in ./risedev check (or alias, ./risedev c)

[xxchan]

Others LGTM. Thanks

Can we add some tests btw? (Perhaps we can do sth like select count(*) group by source_id to avoid nondeterministic results including actor ids)

[xxchan]

Can we add a column type to show whether it's source fragment or backfill fragment here?

[shanicky]

Do we need to differentiate between the backfill source and the downstream?

[shanicky]

added

[xxchan]

Maybe this need a release note btw