[FLINK-35242] Supports per-SE type configuration & "lenient" evolution behavior #3339

yuxiqian · 2024-05-21T09:47:05Z

Adds TRY_EVOLVE behavior that tolerates exception during metadata applying process
Supports configuring various behavior for each schema evolution type with include.schema.changes and exclude.schema.changes sink option
Adds schema evolution IT cases

With this change, Schema Operator now stores both "upstream" schema (which always keep up with the structure coming from pipeline source after the transformation) and "evolved" schema (the schema that actually applied to sink). They might differ when:

Schema change events are partially or fully ignored (by fine-grained options / IGNORE behavior)
Some schema change events failed to apply but pipeline keeps executing (in TRY_EVOLVE mode)

If these cases occur, SchemaOperator needs to cast upstream schema to match the actual evolved version (by adding / removing columns, converting types, etc.) with tolerance (fill in null if meaningful casts are not possible.)

hk-lrzy · 2024-05-24T06:07:14Z

...r/src/main/java/org/apache/flink/cdc/composer/flink/translator/SchemaOperatorTranslator.java

            case EXCEPTION:
-                return exceptionOnSchemaChange(input, parallelism);
+                return exceptionOnSchemaChange(


I have some questions about the behavior for IGNORE and EXCEPTION.
Now when we setting the behavior as IGNORE or EXCEPTION, the job will be failed, should it be fixed it or create a new PR for it?

Thanks for pointing it out! I think it would be better fix it in this PR since SchemaOperator & Registry will be greatly modified to implement FLINK-35242.

(Seems #3352 / FLINK-35432 is about an irrelevant problem about MySQL ddl parsing, and should be fine to be reviewed and merged independently.)

I agree it, and i also point it in the issue with JIRA / https://issues.apache.org/jira/browse/FLINK-35436 and have another #3355

I think we can merge it into this PR.

yuxiqian · 2024-05-24T08:54:51Z

@PatrickRen @leonardBang PTAL

yuxiqian · 2024-06-26T08:27:47Z

Just added LENIENT schema evolve behavior & rebased to master. Could @loserwang1024 please take a look at this?

Since this PR has grown too big, hopefully this could be reviewed & merged soon to avoid any conflicts hassle in the future. cc @leonardBang

leonardBang · 2024-07-12T09:46:37Z

@yuxiqian Thanks for the great work, could you rebase to latest master?

yuxiqian · 2024-07-12T10:23:41Z

Done, rebased with master branch

leonardBang

Thanks @yuxiqian for the great work, I finished the first round review and left some comments and I think you can start to address my comments.

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/DataSink.java

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/MetadataApplier.java

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventType.java

leonardBang · 2024-07-23T12:17:58Z

...-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventTypeFamily.java

+    public static final SchemaChangeEventType[] ADD = {SchemaChangeEventType.ADD_COLUMN};
+
+    public static final SchemaChangeEventType[] ALTER = {SchemaChangeEventType.ALTER_COLUMN_TYPE};
+
+    public static final SchemaChangeEventType[] CREATE = {SchemaChangeEventType.CREATE_TABLE};
+
+    public static final SchemaChangeEventType[] DROP = {SchemaChangeEventType.DROP_COLUMN};
+
+    public static final SchemaChangeEventType[] RENAME = {SchemaChangeEventType.RENAME_COLUMN};
+
+    public static final SchemaChangeEventType[] TABLE = {SchemaChangeEventType.CREATE_TABLE};


Could you share the basis of the categories ? current hierarchy confuse me a bit.

For class names like AddColumnEvent, I've splitten them like add.column, and puts them into both add and column family.

...k-cdc-common/src/main/java/org/apache/flink/cdc/common/exceptions/SchemaEvolveException.java

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/SchemaChangeBehavior.java

...-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisDataSink.java

flink-cdc-runtime/src/test/resources/log4j2-test.properties

leonardBang · 2024-07-23T12:27:26Z

...c/main/java/org/apache/flink/cdc/runtime/operators/schema/metrics/SchemaOperatorMetrics.java

+import java.util.Map;
+
+/** A collection class for handling metrics in {@link SchemaOperator}. */
+public class SchemaOperatorMetrics {


We can split the metric logic to another PR in next time, it's totally independent with SE behavior. But, I like the idea to report metrics for SchemaOperator.

Thanks for the reminder, I'll take notice on this next time.

yuxiqian · 2024-08-07T08:57:12Z

Squashed & Rebased with master.

Since this also changes SchemaManager serialization format, could @lvyanquan please take a look? (Serialization version isn't updated since it's expected that no version will be released between this PR and FLINK-34638.

leonardBang

Thanks @yuxiqian for the great work, the PR generally looks good to me, I just left some minor comments

...-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java

...k-cdc-common/src/main/java/org/apache/flink/cdc/common/exceptions/SchemaEvolveException.java

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventType.java

...va/org/apache/flink/cdc/runtime/operators/schema/event/ApplyEvolvedSchemaChangeResponse.java

...va/org/apache/flink/cdc/runtime/operators/schema/event/ApplyUpstreamSchemaChangeRequest.java

...ava/org/apache/flink/cdc/runtime/operators/schema/event/ApplyEvolvedSchemaChangeRequest.java

...eline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolveE2eITCase.java

… evolution behavior & Add schema operator metrics # Conflicts: # flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaManager.java # flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TableChangeInfo.java

…text => cause

...va/org/apache/flink/cdc/runtime/operators/schema/event/ApplyOriginalSchemaChangeRequest.java

leonardBang

Thanks @yuxiqian for the great work, LGTM

… schema evolution behavior This closes apache#3339.

github-actions bot added values-pipeline-connector composer common runtime cli e2e-tests doris-pipeline-connector starrocks-pipeline-connector labels May 21, 2024

yuxiqian marked this pull request as draft May 22, 2024 01:22

hk-lrzy reviewed May 24, 2024

View reviewed changes

yuxiqian force-pushed the FLINK-35242 branch 2 times, most recently from 4e4dbf5 to a3c9efe Compare May 24, 2024 07:14

yuxiqian marked this pull request as ready for review May 24, 2024 08:46

yuxiqian force-pushed the FLINK-35242 branch 3 times, most recently from 6b13fe7 to 6959c47 Compare May 27, 2024 01:15

yuxiqian changed the title ~~[FLINK-35242] Optimize schema evolution & add SE IT cases~~ [FLINK-35242] Support per-type configuration & tolerance SE behaviors May 31, 2024

yuxiqian changed the title ~~[FLINK-35242] Support per-type configuration & tolerance SE behaviors~~ [FLINK-35242] Supports per-SE type configuration & tolerance evolution behavior May 31, 2024

yuxiqian mentioned this pull request May 31, 2024

[FLINK-35243][cdc] Supports more schema change event types & backfill pre-schema payload for sink #3296

Closed

yuxiqian force-pushed the FLINK-35242 branch from 6959c47 to 71b661e Compare June 12, 2024 03:50

github-actions bot added the migration-tests label Jun 12, 2024

yuxiqian force-pushed the FLINK-35242 branch 3 times, most recently from 7470fc4 to 079d531 Compare June 19, 2024 11:06

github-actions bot added paimon-pipeline-connector kafka-pipeline-connector labels Jun 19, 2024

yuxiqian force-pushed the FLINK-35242 branch 3 times, most recently from a252a61 to 9ac25ef Compare June 26, 2024 08:25

yuxiqian changed the title ~~[FLINK-35242] Supports per-SE type configuration & tolerance evolution behavior~~ [FLINK-35242] Supports per-SE type configuration & "lenient" evolution behavior Jul 8, 2024

yuxiqian force-pushed the FLINK-35242 branch from 9ac25ef to 83a6174 Compare July 12, 2024 10:18

yuxiqian force-pushed the FLINK-35242 branch from 83a6174 to 0ad032c Compare July 12, 2024 11:01

leonardBang self-requested a review July 22, 2024 11:53

leonardBang reviewed Jul 23, 2024

View reviewed changes

yuxiqian force-pushed the FLINK-35242 branch from afa70ae to 7c97bad Compare July 24, 2024 01:14

yuxiqian force-pushed the FLINK-35242 branch 3 times, most recently from b3cc000 to 1c6face Compare August 7, 2024 08:53

leonardBang reviewed Aug 7, 2024

View reviewed changes

yuxiqian added 6 commits August 8, 2024 09:46

Resolve conflicts

5281e71

mark api @PublicEvolving

a18def4

rename SchemaEvolveException fields: problem => exceptionMessage, con…

18a5f39

…text => cause

grand rename UpstreamSchema => OriginalSchema

7a2a562

refactor: schema evolve e2e case

d9a525c

yuxiqian force-pushed the FLINK-35242 branch from 1c6face to d9a525c Compare August 8, 2024 03:28

leonardBang reviewed Aug 8, 2024

View reviewed changes

...va/org/apache/flink/cdc/runtime/operators/schema/event/ApplyOriginalSchemaChangeRequest.java Outdated Show resolved Hide resolved

fix: java docs

862db2c

leonardBang approved these changes Aug 8, 2024

View reviewed changes

github-actions bot added approved reviewed labels Aug 8, 2024

leonardBang merged commit 5ed9e05 into apache:master Aug 8, 2024
20 checks passed

yuxiqian mentioned this pull request Aug 13, 2024

[hotfix][doris] DorisMetadataApplier fails to reject unsupported schema change events #3503

Closed

qiaozongmi pushed a commit to qiaozongmi/flink-cdc that referenced this pull request Sep 23, 2024

[FLINK-35242][cdc-common][cdc-runtime] Support TRY_EVOLVE and LENIENT…

774cfa7

… schema evolution behavior This closes apache#3339.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-35242] Supports per-SE type configuration & "lenient" evolution behavior #3339

[FLINK-35242] Supports per-SE type configuration & "lenient" evolution behavior #3339

yuxiqian commented May 21, 2024 •

edited

Loading

hk-lrzy May 24, 2024 •

edited

Loading

yuxiqian May 24, 2024 •

edited

Loading

hk-lrzy May 25, 2024

yuxiqian commented May 24, 2024

yuxiqian commented Jun 26, 2024 •

edited

Loading

leonardBang commented Jul 12, 2024

yuxiqian commented Jul 12, 2024

leonardBang left a comment

leonardBang Jul 23, 2024

yuxiqian Jul 24, 2024

leonardBang Jul 23, 2024

yuxiqian Jul 24, 2024

yuxiqian commented Aug 7, 2024

leonardBang left a comment

leonardBang left a comment

[FLINK-35242] Supports per-SE type configuration & "lenient" evolution behavior #3339

[FLINK-35242] Supports per-SE type configuration & "lenient" evolution behavior #3339

Conversation

yuxiqian commented May 21, 2024 • edited Loading

hk-lrzy May 24, 2024 • edited Loading

Choose a reason for hiding this comment

yuxiqian May 24, 2024 • edited Loading

Choose a reason for hiding this comment

hk-lrzy May 25, 2024

Choose a reason for hiding this comment

yuxiqian commented May 24, 2024

yuxiqian commented Jun 26, 2024 • edited Loading

leonardBang commented Jul 12, 2024

yuxiqian commented Jul 12, 2024

leonardBang left a comment

Choose a reason for hiding this comment

leonardBang Jul 23, 2024

Choose a reason for hiding this comment

yuxiqian Jul 24, 2024

Choose a reason for hiding this comment

leonardBang Jul 23, 2024

Choose a reason for hiding this comment

yuxiqian Jul 24, 2024

Choose a reason for hiding this comment

yuxiqian commented Aug 7, 2024

leonardBang left a comment

Choose a reason for hiding this comment

leonardBang left a comment

Choose a reason for hiding this comment

yuxiqian commented May 21, 2024 •

edited

Loading

hk-lrzy May 24, 2024 •

edited

Loading

yuxiqian May 24, 2024 •

edited

Loading

yuxiqian commented Jun 26, 2024 •

edited

Loading