You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "cygnusdark (via GitHub)" <gi...@apache.org> on 2023/10/08 08:49:08 UTC
[PR] [Improve] optimize mysql schema sync [doris-flink-connector]
cygnusdark opened a new pull request, #204:
URL: https://github.com/apache/doris-flink-connector/pull/204
# Proposed changes
Optimize mysql schema sync when creating doris table or making light schema change.
## Problem Summary:
Currently, when the sync tool creates doris table or makes light schema changes, some schema information like default value, nullable and comment is lost. The PR fixed this issue.
## Checklist(Required)
1. Does it affect the original behavior: (No)
2. Has unit tests been added: (Yes)
3. Has document been added or modified: (No)
4. Does it need to update dependencies: (No)
5. Are there any changes that cannot be rolled back: (No)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve] optimize mysql schema sync [doris-flink-connector]
Posted by "cygnusdark (via GitHub)" <gi...@apache.org>.
cygnusdark closed pull request #204: [Improve] optimize mysql schema sync
URL: https://github.com/apache/doris-flink-connector/pull/204
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve] optimize mysql schema sync [doris-flink-connector]
Posted by "DongLiang-0 (via GitHub)" <gi...@apache.org>.
DongLiang-0 commented on PR #204:
URL: https://github.com/apache/doris-flink-connector/pull/204#issuecomment-1794663680
The code seems to be a little behind the master branch. You need to rebase the master branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve] optimize mysql schema sync [doris-flink-connector]
Posted by "cygnusdark (via GitHub)" <gi...@apache.org>.
cygnusdark commented on code in PR #204:
URL: https://github.com/apache/doris-flink-connector/pull/204#discussion_r1349946175
##########
flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/JsonDebeziumSchemaSerializer.java:
##########
@@ -406,12 +408,40 @@ public String extractDDL(JsonNode record) throws JsonProcessingException {
if (!Objects.isNull(ddl)) {
//filter add/drop operation
Matcher matcher = addDropDDLPattern.matcher(ddl);
- if(matcher.find()){
+ if (matcher.find()) {
String op = matcher.group(1);
String col = matcher.group(3);
String type = matcher.group(5);
type = handleType(type);
- ddl = String.format(EXECUTE_DDL, dorisOptions.getTableIdentifier(), op, col, type);
+
+ String nullable = "";
+ String defaultValue = "";
+ String comment = "";
+ Pattern pattern = Pattern.compile(defaultNullableCommentRegex, Pattern.CASE_INSENSITIVE);
+ Matcher defaultNullableCommentMatcher = pattern.matcher(ddl);
+
+ while (defaultNullableCommentMatcher.find()) {
+ if (defaultNullableCommentMatcher.group(1) != null) {
+ nullable = "NOT NULL";
+ }
+ if (defaultNullableCommentMatcher.group(2) != null) {
+ String originalDefaultValue = defaultNullableCommentMatcher.group(3);
+ if ("NULL".equalsIgnoreCase(originalDefaultValue.trim())
+ || "EMPTY".equalsIgnoreCase(originalDefaultValue.trim())
+ || "CURRENT_TIMESTAMP".equalsIgnoreCase(originalDefaultValue.trim())
+ || originalDefaultValue.startsWith("'")
+ || originalDefaultValue.startsWith("\"")) {
+ defaultValue = defaultNullableCommentMatcher.group(2);
+ } else {
+ defaultValue = "DEFAULT '" + defaultNullableCommentMatcher.group(3) + "'";
+ }
+ }
+ if (defaultNullableCommentMatcher.group(4) != null) {
+ comment = defaultNullableCommentMatcher.group(4);
+ }
+ }
+
+ ddl = String.format(EXECUTE_DDL, dorisOptions.getTableIdentifier(), op, col, type, nullable, defaultValue, comment).trim();
Review Comment:
Our cdc jobs used new SchemaChange feature before. Due to debezium and mysql-cdc limitations, using newSchemaChange only guarantees defaults to be synchronized (Old SchemaChange extracts defaults, nulables and coments from DDL, and new SchemaChange extracts them from debezium records). So I made this pr and switched back to old schemaChange.
I will think about how to strengthen the generality.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve] optimize mysql schema sync [doris-flink-connector]
Posted by "DongLiang-0 (via GitHub)" <gi...@apache.org>.
DongLiang-0 commented on code in PR #204:
URL: https://github.com/apache/doris-flink-connector/pull/204#discussion_r1349826947
##########
flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/JsonDebeziumSchemaSerializer.java:
##########
@@ -406,12 +408,40 @@ public String extractDDL(JsonNode record) throws JsonProcessingException {
if (!Objects.isNull(ddl)) {
//filter add/drop operation
Matcher matcher = addDropDDLPattern.matcher(ddl);
- if(matcher.find()){
+ if (matcher.find()) {
String op = matcher.group(1);
String col = matcher.group(3);
String type = matcher.group(5);
type = handleType(type);
- ddl = String.format(EXECUTE_DDL, dorisOptions.getTableIdentifier(), op, col, type);
+
+ String nullable = "";
+ String defaultValue = "";
+ String comment = "";
+ Pattern pattern = Pattern.compile(defaultNullableCommentRegex, Pattern.CASE_INSENSITIVE);
+ Matcher defaultNullableCommentMatcher = pattern.matcher(ddl);
+
+ while (defaultNullableCommentMatcher.find()) {
+ if (defaultNullableCommentMatcher.group(1) != null) {
+ nullable = "NOT NULL";
+ }
+ if (defaultNullableCommentMatcher.group(2) != null) {
+ String originalDefaultValue = defaultNullableCommentMatcher.group(3);
+ if ("NULL".equalsIgnoreCase(originalDefaultValue.trim())
+ || "EMPTY".equalsIgnoreCase(originalDefaultValue.trim())
+ || "CURRENT_TIMESTAMP".equalsIgnoreCase(originalDefaultValue.trim())
+ || originalDefaultValue.startsWith("'")
+ || originalDefaultValue.startsWith("\"")) {
+ defaultValue = defaultNullableCommentMatcher.group(2);
+ } else {
+ defaultValue = "DEFAULT '" + defaultNullableCommentMatcher.group(3) + "'";
+ }
+ }
+ if (defaultNullableCommentMatcher.group(4) != null) {
+ comment = defaultNullableCommentMatcher.group(4);
+ }
+ }
+
+ ddl = String.format(EXECUTE_DDL, dorisOptions.getTableIdentifier(), op, col, type, nullable, defaultValue, comment).trim();
Review Comment:
The new schema change already supports mysql field changes.https://github.com/apache/doris-flink-connector/pull/167
And I think it is not very general to extract default and comment values through regular matching. For example, does oracle, sqlserver, and postgres also have this syntax?
##########
flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/JsonDebeziumSchemaSerializer.java:
##########
@@ -406,12 +408,40 @@ public String extractDDL(JsonNode record) throws JsonProcessingException {
if (!Objects.isNull(ddl)) {
//filter add/drop operation
Matcher matcher = addDropDDLPattern.matcher(ddl);
- if(matcher.find()){
+ if (matcher.find()) {
String op = matcher.group(1);
String col = matcher.group(3);
String type = matcher.group(5);
type = handleType(type);
- ddl = String.format(EXECUTE_DDL, dorisOptions.getTableIdentifier(), op, col, type);
+
+ String nullable = "";
+ String defaultValue = "";
+ String comment = "";
+ Pattern pattern = Pattern.compile(defaultNullableCommentRegex, Pattern.CASE_INSENSITIVE);
+ Matcher defaultNullableCommentMatcher = pattern.matcher(ddl);
+
+ while (defaultNullableCommentMatcher.find()) {
+ if (defaultNullableCommentMatcher.group(1) != null) {
+ nullable = "NOT NULL";
+ }
+ if (defaultNullableCommentMatcher.group(2) != null) {
+ String originalDefaultValue = defaultNullableCommentMatcher.group(3);
+ if ("NULL".equalsIgnoreCase(originalDefaultValue.trim())
+ || "EMPTY".equalsIgnoreCase(originalDefaultValue.trim())
+ || "CURRENT_TIMESTAMP".equalsIgnoreCase(originalDefaultValue.trim())
+ || originalDefaultValue.startsWith("'")
+ || originalDefaultValue.startsWith("\"")) {
+ defaultValue = defaultNullableCommentMatcher.group(2);
+ } else {
+ defaultValue = "DEFAULT '" + defaultNullableCommentMatcher.group(3) + "'";
+ }
+ }
+ if (defaultNullableCommentMatcher.group(4) != null) {
+ comment = defaultNullableCommentMatcher.group(4);
+ }
+ }
+
+ ddl = String.format(EXECUTE_DDL, dorisOptions.getTableIdentifier(), op, col, type, nullable, defaultValue, comment).trim();
Review Comment:
The new schema change already supports mysql field changes.https://github.com/apache/doris-flink-connector/pull/167
And I think it is not very general to extract default and comment values through regular matching. For example, does oracle, sqlserver, and postgres also have this syntax?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org