feat(connector): support ingest Citus distributed table

[StrikeW]

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

What's changed and what's your intention?

Support citus-cdc connector for the CREATE TABLE statement, which allows users to ingest a distributed table stored in Citus to RisingWave. For example, we can ingest the github_events table via the following SQL (don't forget to specify a primary key for the table):

CREATE TABLE github_events_rw
(
    event_id bigint,
    event_type text,
    event_public boolean,
    repo_id bigint,
    payload jsonb,
    repo jsonb,
    user_id bigint,
    org jsonb,
    created_at timestamp,
    PRIMARY KEY (event_id, user_id)
) WITH (
    connector = 'citus-cdc',
    hostname = '127.0.0.1',
    port = '5432',
    username = 'dbz',
    password = '123456',
    database.servers = '172.31.29.245:5432,172.31.31.177:5432',
    database.name = 'postgres',
    schema.name = 'public',
    table.name = 'github_events',
    slot.name = 'github_events_dbz_slot1',
);

Users need to specify the host and port for Citus worker nodes by database.servers in the WITH clause.

Prerequisites

1. Create a superuser role in the Citus cluster

# create role on coordinator node
create role dbz Superuser LOGIN;

# create role on worker nodes
SELECT run_command_on_workers($cmd$
  create role dbz Superuser LOGIN;
$cmd$);

2. Set wal_level to logical on each worker node (e.g. modify the postgresql.conf)
3. Set the replica identity to FULL for the table you want to ingest

# execute on coordinator node
ALTER TABLE github_events REPLICA IDENTITY FULL;

Limitations

• The connector requires a Postgres superuser role
• Only distributed table is supported
• Cannot detect newly added worker nodes (can be improved in future PR)

Checklist For Contributors

• 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).
• I have demonstrated that backward compatibility is not broken by breaking changes and created issues to track deprecated features to be removed in the future. (Please refer to the issue)
• All checks passed in ./risedev check (or alias, ./risedev c)

Checklist For Reviewers

• I have requested macro/micro-benchmarks as this PR can affect performance substantially, and the results are shown.

Documentation

Click here for Documentation

Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

• Connector (sources & sinks)

Release note

Support citus-cdc connector for the CREATE TABLE statement, which allows users to ingest a distributed table stored in Citus to RisingWave.

[tabVersion]

@fuyufjh I am a little skeptical about this approach. Can we try to deliver the poc on this branch rather than main branch? We can merge the pr after they accept this.

[codecov]

Codecov Report

Merging #8988 (9d60c43) into main (ff8de5c) will decrease coverage by 0.02%.
The diff coverage is 29.90%.

@@            Coverage Diff             @@
##             main    #8988      +/-   ##
==========================================
- Coverage   70.83%   70.81%   -0.02%     
==========================================
  Files        1201     1201              
  Lines      200454   200533      +79     
==========================================
+ Hits       141992   142013      +21     
- Misses      58462    58520      +58     

Flag	Coverage Δ
rust	70.81% <29.90%> (-0.02%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
src/connector/src/source/base.rs	65.26% <0.00%> (-2.33%)	⬇️
src/connector/src/source/cdc/enumerator/mod.rs	0.00% <0.00%> (ø)
src/connector/src/source/cdc/mod.rs	20.00% <0.00%> (+5.71%)	⬆️
src/connector/src/source/cdc/source/reader.rs	0.00% <0.00%> (ø)
src/connector/src/source/cdc/split.rs	58.33% <33.33%> (-10.09%)	⬇️
...c/connector/src/parser/maxwell/simd_json_parser.rs	82.35% <80.00%> (-0.18%)	⬇️
src/connector/src/parser/common.rs	71.69% <85.00%> (+1.48%)	⬆️
.../connector/src/parser/debezium/simd_json_parser.rs	91.26% <100.00%> (+0.08%)	⬆️
src/connector/src/parser/json_parser.rs	95.73% <100.00%> (+0.02%)	⬆️
src/frontend/src/handler/create_source.rs	49.49% <100.00%> (+0.08%)	⬆️

... and 6 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[fuyufjh]

@fuyufjh I am a little skeptical about this approach. Can we try to deliver the poc on this branch rather than main branch? We can merge the pr after they accept this.

I am 50-50 on this.

Except the refactoring around PostgresValidator (which I think is good), this PR doesn't look very invasive to me. The newly-introduced server_addr in Split also LGTM.

Let the author @StrikeW make this decision :)

[fuyufjh]

Generally LGTM

[fuyufjh]

Just curious: why MySQL doesn't need a "Validator"?

[StrikeW]

It will but not that urgent. Just because I want to implement the feature ASAP. I will refactor the validation part when I have bandwidth.

[StrikeW]

Performance results with 1 coordinator 2 workers setup, ingest a TPCH lineitem table with 10M rows.
The result conforms to our bench result of the Postgres connector, that is each split can reach 2w rows/sec throughput.

[StrikeW]

@fuyufjh I am a little skeptical about this approach. Can we try to deliver the poc on this branch rather than main branch? We can merge the pr after they accept this.

Could you elaborate more about your concerns? @tabVersion