apache / apache/paimon

[Feature][Question] CDC schema evolution in synchronizing databases?

Open
#3,581 2 comments 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Java
Stars
3.4k
Forks
1.4k
Avg merge
1d 11h
Merged PRs (30d)
396

Description

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.

### Motivation

I was using Flink CDC to sync some tables in a MySQL database and I suddenly noticed that the job failed, I didn't look much into it since the job often fails due to OOM (I am testing with loads of table, around 90ish), and when I restart the job, it comes with the below exception
```
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Incompatible schema found.
Paimon table is: ...
MySQL table is: ...
If you want to ignore the incompatible tables, please specify --ignore-incompatible to true.
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:105)
at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.runApplicationEntryPoint(ApplicationDispatcherBootstrap.java:301)
at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.lambda$runApplicationAsync$2(ApplicationDispatcherBootstrap.java:254)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
at org.apache.flink.runtime.concurrent.pekko.ActorSystemScheduledExecutorAdapter$ScheduledFutureTask.run(ActorSystemScheduledExecutorAdapter.java:172)
at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
at org.apache.flink.runtime.concurrent.ClassLoadingUtils.lambda$withContextClassLoader$0(ClassLoadingUtils.java:41)
at org.apache.pekko.dispatch.TaskInvocation.run(AbstractDispatcher.scala:59)
at org.apache.pekko.dispatch.ForkJoinExecutorConfigurator$PekkoForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:57)
at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source)
at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source)
```
The table schema in MySQL has indeed changed with a column added. Originally I had the assumption that database synchronization would also have schema evolution like table synchronization, but upon further reading into the [documents](https://paimon.apache.org/docs/0.8/flink/cdc-ingestion/overview/#what-is-schema-evolution):
> Suppose we have a MySQL table named tableA, it has three fields: field_1, field_2, field_3. When we want to load this MySQL table to Paimon, we can do this in Flink SQL, or use [MySqlSyncTableAction](https://paimon.apache.org/docs/0.8/api/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction).

With it specifically mentioning MySqlSyncTableAction, it seems schema evolution is not supported in database synchronization, is this true? If so, I have 2 questions:
1. What would be the recommended action if I want to continue the database synchronization with that table included and updated? Do I need to drop said table and load it again? Or is there another easier way?
2. Would schema evolution be supported in database synchronization in the future?

Some additional info for the builds I am using:
- Flink 1.18.1
- paimon-flink-action 0.8.1
- paimon-flink 1.18-0.8.1
- flink-sql-connector-mysql-cdc 3.0.1

### Solution

_No response_

### Anything else?

_No response_

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with the schema-evolution documentation and the MySqlSyncTableAction entry point cited in the issue, then reproduce the incompatible-schema failure using the listed Flink, Paimon, and MySQL CDC versions. Determine whether database synchronization supports an added column, document the recommended recovery path, and define the expected behavior for future support.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, mysql
Domain
data-engineering, databases
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.