[Improve]Schema change parses ddl sql using jsqlparser framework #422

DongLiang-0 · 2024-07-10T08:56:18Z

Proposed changes

Problem Summary:

Introduce the com.github.jsqlparser framework to parse DDL SQL statements when schema changes.
The original change principle is to infer the columns that need to be added, deleted or modified in this change based on the last schema structure. In some special scenarios, this method will have certain bugs.
After the jsqlparser framework is introduced, the content of this schema change can be parsed based on the ddl passed by the upstream flink-cdc. Very accurate and no need to rely on the last schema.
At the same time, we also deprecated org.apache.doris.flink.sink.writer.serializer.jsondebezium.JsonDebeziumSchemaChangeImpl, which uses regular expressions to match DDL. It is recommended to use org.apache.doris.flink.sink.writer.serializer.jsondebezium .SQLParserService.

Through schema-change-mode, you can specify which way to parse DDL:

specify sql_parser to use JSQLParser to parse DDL statements;
the default is the debezium_structure parameter, which infers schema change changes by parsing the upstream Debezium structure.

Checklist(Required)

Does it affect the original behavior: (Yes/No/I Don't know)
Has unit tests been added: (Yes/No/No Need)
Has document been added or modified: (Yes/No/No Need)
Does it need to update dependencies: (Yes/No)
Are there any changes that cannot be rolled back: (Yes/No)

Further comments

If this is a relatively large or complex change, kick off the discussion at dev@doris.apache.org by explaining why you chose the solution you did and what alternatives you considered, etc...

JNSimba · 2024-07-11T13:05:22Z

Thanks to @DongLiang-0 for his contribution. It is suggested to control it through a switch here.

...rc/main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java

JNSimba · 2024-07-16T07:01:36Z

...rc/main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java

+            if (org.apache.commons.lang3.StringUtils.isEmpty(schemaChangeMode)) {
+                return this;
+            }


Which schemachangemode is used when it is empty?

JsonDebeziumSchemaChangeImplV2 is created when empty.

JNSimba · 2024-07-16T07:07:24Z

...doris-connector/src/test/java/org/apache/doris/flink/tools/cdc/CdcMysqlSyncDatabaseCase.java

@@ -90,6 +92,7 @@ public static void main(String[] args) throws Exception {
                .setTableConfig(tableConfig)
                .setCreateTableOnly(false)
                .setNewSchemaChange(useNewSchemaChange)
+                .setSchemaChangeMode(schemaChangeMode)


What will happen if this is not set?

Same as above.

flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SQLParserManager.java

JNSimba · 2024-07-16T07:15:49Z

...ava/org/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChange.java

+
+    protected void extractSourceConnector(JsonNode record) {
+        if (Objects.isNull(sourceConnector)) {
+            sourceConnector =


Will sourceConnector be null?

Yes, the program is null when it is started, because the sourceConnector is extracted from the schema change data.

JNSimba · 2024-07-16T07:17:01Z

...g/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java

-            } else {
-                LOG.info("Unsupported event type {}", eventType);


It seems that this branch is missing?

extractEventType(JsonNode record)only returns three types of events, namely ALTER, CREATE, and null. where null has been handled here

JNSimba · 2024-07-16T07:19:42Z

...c/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/SQLParserService.java

+                }
+                List<String> ddlList = tryParserAlterDDLs(recordRoot);
+                status = executeAlterDDLs(ddlList, recordRoot, dorisTableTuple, status);
+            }


Missing an else?

Same as above.

...c/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/SQLParserService.java

JNSimba · 2024-07-16T10:17:32Z

...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SQLParserSchemaManager.java

+        if (columnSpecs.contains("default")) {
+            int defaultIndex = columnSpecs.indexOf("default");
+            defaultValue = removeQuotes(columnSpecs.get(defaultIndex + 1));
+        } else if (columnSpecs.contains("DEFAULT")) {
+            int defaultIndex = columnSpecs.indexOf("DEFAULT");
+            defaultValue = removeQuotes(columnSpecs.get(defaultIndex + 1));


Will there be any problem with this judgment?

For example, in this example, there is no special method to obtain default and comment, only through the columnSpecs collection.
In order to avoid NPE, I also added some prevention methods.

JNSimba · 2024-07-16T10:32:20Z

...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SQLParserSchemaManager.java

+    private String extractComment(List<String> columnSpecs) {
+        String comment = null;
+        if (columnSpecs.contains("comment")) {
+            int commentIndex = columnSpecs.indexOf("comment");
+            comment = removeQuotes(columnSpecs.get(commentIndex + 1));
+        }
+        if (columnSpecs.contains("COMMENT")) {
+            int commentIndex = columnSpecs.indexOf("COMMENT");
+            comment = removeQuotes(columnSpecs.get(commentIndex + 1));
+        }
+        return comment;
+    }


same above, does JSqlParser have any suspicious methods to directly obtain the Default value and Comment?

...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SQLParserSchemaManager.java

JNSimba

LGTM

DongLiang-0 marked this pull request as draft July 10, 2024 10:19

DongLiang-0 marked this pull request as ready for review July 11, 2024 10:11

DongLiang-0 marked this pull request as draft July 15, 2024 03:38

DongLiang-0 force-pushed the improve-schema branch 2 times, most recently from 2c60f2d to 4c5595e Compare July 15, 2024 09:40

[Improve]Schema change parses ddl sql using jsqlparser framework

9b39600

DongLiang-0 force-pushed the improve-schema branch from 4c5595e to 9b39600 Compare July 16, 2024 02:13

fix

301e057

DongLiang-0 marked this pull request as ready for review July 16, 2024 06:34

fix

0d296a5

JNSimba reviewed Jul 16, 2024

View reviewed changes

...rc/main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java Show resolved Hide resolved

JNSimba reviewed Jul 16, 2024

View reviewed changes

flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SQLParserManager.java Outdated Show resolved Hide resolved

JNSimba reviewed Jul 16, 2024

View reviewed changes

...c/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/SQLParserService.java Outdated Show resolved Hide resolved

fix comment

cc5f3ac

JNSimba reviewed Jul 16, 2024

View reviewed changes

DongLiang-0 added 3 commits July 17, 2024 10:16

fix comment

b44947d

fix

1618075

fix

e973481

JNSimba reviewed Jul 17, 2024

View reviewed changes

...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SQLParserSchemaManager.java Outdated Show resolved Hide resolved

JNSimba reviewed Jul 17, 2024

View reviewed changes

...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SQLParserSchemaManager.java Outdated Show resolved Hide resolved

DongLiang-0 added 2 commits July 17, 2024 12:29

fix

ed0b162

add unit test

086082b

JNSimba approved these changes Jul 17, 2024

View reviewed changes

JNSimba merged commit 13f1fcd into apache:master Jul 17, 2024
6 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Improve]Schema change parses ddl sql using jsqlparser framework #422

[Improve]Schema change parses ddl sql using jsqlparser framework #422

DongLiang-0 commented Jul 10, 2024 •

edited

Loading

JNSimba commented Jul 11, 2024

JNSimba Jul 16, 2024

DongLiang-0 Jul 16, 2024 •

edited

Loading

JNSimba Jul 16, 2024

DongLiang-0 Jul 16, 2024

JNSimba Jul 16, 2024

DongLiang-0 Jul 16, 2024

JNSimba Jul 16, 2024

DongLiang-0 Jul 16, 2024

JNSimba Jul 16, 2024

DongLiang-0 Jul 16, 2024

JNSimba Jul 16, 2024

DongLiang-0 Jul 17, 2024 •

edited

Loading

JNSimba Jul 16, 2024

JNSimba left a comment

[Improve]Schema change parses ddl sql using jsqlparser framework #422

[Improve]Schema change parses ddl sql using jsqlparser framework #422

Conversation

DongLiang-0 commented Jul 10, 2024 • edited Loading

Proposed changes

Problem Summary:

Checklist(Required)

Further comments

JNSimba commented Jul 11, 2024

Choose a reason for hiding this comment

DongLiang-0 Jul 16, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

DongLiang-0 Jul 17, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JNSimba left a comment

Choose a reason for hiding this comment

DongLiang-0 commented Jul 10, 2024 •

edited

Loading

DongLiang-0 Jul 16, 2024 •

edited

Loading

DongLiang-0 Jul 17, 2024 •

edited

Loading