feat(cdc): share a changelog stream for multiple cdc tables

[StrikeW]

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

What's changed and what's your intention?

Part of #11079, related: #11545

Interface

1. create a CDC Source job

create source mysql_mydb with (
 connector = 'mysql-cdc',
 hostname = '127.0.0.1',
 port = '8306',
 username = 'root',
 password = '123456',
 database.name = 'mydb',
 server.id = 5888
);

• The source job is singleton
• The format is fixed to FORMAT PLAIN ENCODE JSON so it can be eliminated.

2. create CDC Tables based on the above source

CREATE TABLE t1_rw (
    v1 int,
    v2 int,
    PRIMARY KEY(v1)
) FROM mysql_mydb TABLE 'mydb.t1';

CREATE TABLE t3_rw (
  v1 INTEGER,
  v2 timestamptz,
  PRIMARY KEY (v1)
) FROM mysql_mydb TABLE 'mydb.t3';

• The parallelism of these table jobs is set to 1
• The table name behind the second TABLE keyword should be qualified with the database name in the MySQL.

Implementation

• Add a new Source type streaming job to support the cdc source job
• The output schema of the cdc source job is fixed to (payload jsonb, _rw_offset varchar, _rw_table_name varchar, _row_id ).
• The dispatcher of the cdc source job will dispatch cdc event chunks to downstream table jobs based on the _rw_table_name column
• The CdcBackfillExecutor will transform the upstream chunk to the specific schema of a table job (e.g. (v1 int, v2 int)) before doing backfill
• Right now the backfill state doesn't persist in this case, so it will trigger the backfill again upon recovery
• Here is a figure to demonstrate the workflow:
  

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• 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.

• Allows users to create multiple CDC tables that share a same Source subscribing to a upstream database.
• The new syntax is described in the PR description.
• User needs to enable this feature by set cdc_backfill='true'

[StrikeW]

It should be fine now. 😋

[StrikeW]

I have modified the drop_relation() to support dropping the source streaming job if necessary.
https://github.com/risingwavelabs/risingwave/pull/12535/files#diff-1da683135f143ab95455c5fee12eca2dccaafe7d90b9e2cbc8c0ba330204659fR1263-R1265

[StrikeW]

Still in use:

risingwave/src/meta/src/stream/stream_graph/actor.rs

Line 168 in 27ea58e

assert_matches!(batch_plan_node.node_body, Some(NodeBody::BatchPlan(_)));

[StrikeW]

Will refine the dispatcher part later, I have fired an issue.

[fuyufjh]

Overall LGTM. Please kindly check these suggestions before merging:

#12535 (comment)
#12535 (comment)

[BugenZhao]

Generally LGTM. Not dived into the executor implementation. Mainly focused on other parts.