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-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
