sink: Add schema/table routing support for mysql and redo sinks

[a-cong]

What problem does this PR solve?

Issue Number: close #3700

What is changed and how it works?

Porting changes from pingcap/tiflow#12435 to new ticdc architecture.

The dispatchers config is the same as in tiflow -

[[sink.dispatchers]]
matcher = ["uds_000.*"]
schema = "tidb_failover_test"
table = "{table}"

[[sink.dispatchers]]
matcher = ["oyster_production.*"]
schema = "oyster_production"
table = "{table}"

We also add the TargetSchema and TargetTable fields to the TableName struct. Since there is now a shared TableInfo object that can be accessed by multiple changefeeds and a new concept of Dispatcher objects, we clone the TableInfo in each dispatcher with the routing information for that table in handleHandshakeEvent (which is executed when the dispatcher starts), and then store that cloned/routed TableInfo for the dispatcher. Subsequent DMLs then look up the stored info. Subsequent DDLs also explicitly update routing in handleSingleDataEvents.

Another difference between ticdc and tiflow is that DDLs are now handled per-sink instead of in one single entry point. Because of this, we need to apply routing in both the mysql and redo sinks separately - the shared rewriting functionality is in pkg/sink/util/ddl_routing.go.

We also copied FetchDDLTables and RenameDDLTable (and the corresponding tests) from https://github.com/pingcap/tiflow/blob/master/dm/pkg/parser/common.go so we could use them to rewrite DDLs.

Check List

Tests

• Unit tests
• Integration tests
• Manual test - deployed to internal dev cluster and bootstrapped a migration shadow with schema mapping, verified changefeed routes DDLs/DMLs correctly

Questions

Do you need to update user documentation, design documentation or monitoring documentation?

Yes - will do so in separate PR

Release note

Added support for schema and table routing for mysql-compatible sinks by extending the `dispatchers` config

Summary by CodeRabbit

• New Features

  • Schema/table routing added to dispatch rules with {schema}/{table} placeholders; routing applied to DML and DDL and exposed via sink routers
  • DDL rewrite support so routed DDLs target configured destinations
  • Redo logs now preserve and restore routing targets during replay
  • Routing expression validation with clear error types

• Testing

  • Extensive unit and integration tests covering routing, DDL rewrite, SQL generation, and redo replay scenarios

[ti-chi-bot]

Adding the "do-not-merge/release-note-label-needed" label because no release-note block was detected, please follow our release note process to remove it.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository.

[ti-chi-bot]

Hi @a-cong. Thanks for your PR.

I'm waiting for a pingcap member to verify that this patch is reasonable to test. If it is, they should reply with /ok-to-test on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should join the org to skip this step.

Once the patch is verified, the new status will be reflected by the ok-to-test label.

I understand the commands that are listed here.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository.

[pingcap-cla-assistant]

All committers have signed the CLA.

[gemini-code-assist]

Summary of Changes

Hello @a-cong, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request significantly enhances the data replication capabilities by introducing schema and table routing for MySQL-compatible and Redo sinks. It allows users to dynamically remap database and table names from the source to the target, providing greater flexibility for data integration and consolidation. The changes involve updating configuration structures, implementing DDL query rewriting logic, and ensuring concurrency safety through event cloning, ultimately enabling seamless replication to custom target schemas and tables.

Highlights

• Schema and Table Routing for MySQL and Redo Sinks: Introduces comprehensive support for schema and table routing, allowing users to define rules to map source database and table names to different target names in MySQL-compatible and Redo sinks. This enables flexible data migration and consolidation scenarios.
• DDL Query Rewriting: Implements a mechanism to parse and rewrite DDL (Data Definition Language) queries based on configured routing rules. This ensures that DDL statements like CREATE TABLE, ALTER TABLE, and RENAME TABLE are executed against the correct target schema and table names in the downstream.
• Event Cloning for Concurrency Safety: Adds cloning logic for BatchDMLEvent and DDLEvent objects before applying routing. This prevents race conditions and unintended mutations when the same event is processed by multiple dispatchers (e.g., the main event dispatcher and the redo log dispatcher), ensuring each can apply routing independently.
• Enhanced TableName and TableInfo: Extends the TableName struct with TargetSchema and TargetTable fields to store routed names. The TableInfo struct now includes a CloneWithRouting method to create a routed copy, and DML SQL generation functions (buildInsert, buildDelete, buildUpdate) are updated to use these target names.
• Integration with Redo Log Replay: Ensures that DDL routing is applied before writing events to the redo log. This means that when redo logs are replayed (e.g., during disaster recovery), the DDLs will correctly target the routed schema and table names, maintaining data consistency.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request introduces a significant new feature: schema and table routing for MySQL and redo log sinks. The changes are extensive, touching upon the API, configuration, dispatchers, and sinks to support this functionality. The design thoughtfully addresses potential concurrency issues by cloning event objects to prevent race conditions, especially when events are shared between the main sink and the redo log sink. A comprehensive set of unit and integration tests has been added to validate the new routing logic. Overall, this is a well-structured implementation of a complex feature. I've identified one critical issue in BatchDMLEvent.CloneForRouting that could lead to data corruption or mis-routing due to improper cloning of nested event objects. Please address this to ensure the correctness of the feature.

[wk989898]

/ok-to-test

[a-cong]

/test all

[3AceShowHand]

/test all

[a-cong]

/test all

[a-cong]

/test all

[a-cong]

/test all

[hongyunyan]

Based on the calculation, ddlTableID appears to represent the logical ID of the DDL table; however, dispatcherID may correspond to the physical ID of a partitioned table, which would explain the mismatch. I am not certain whether this aligns with the intended design.

[hongyunyan]

Based on the current review, the changes related to the router span a significant number of distinct components—such as the event collector, dispatcher, and sink. Could we consider consolidating all unified router behaviors within a single module to prevent the logic from becoming scattered across various locations?

[3AceShowHand]

/test all

[3AceShowHand]

/test all

[3AceShowHand]

/test all

[3AceShowHand]

/test all

[wk989898]

Please use syncdiff to compare the data https://docs.pingcap.com/tidb/stable/sync-diff-inspector-overview/#configuration-file-description

[3AceShowHand]

/test all

[ti-chi-bot]

@a-cong: The following tests failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
pull-cdc-kafka-integration-heavy	4c57fc9	link	true	/test pull-cdc-kafka-integration-heavy
pull-error-log-review	37664ca	link	true	/test pull-error-log-review

Full PR test history. Your PR dashboard.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here.