You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "zhangjun0x01 (via GitHub)" <gi...@apache.org> on 2023/07/21 03:45:08 UTC
[GitHub] [incubator-paimon] zhangjun0x01 opened a new pull request, #1621: [flink] parse primary key and schema by historyRecord
zhangjun0x01 opened a new pull request, #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close https://github.com/apache/incubator-paimon/issues/1613
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] yuzelin commented on a diff in pull request #1621: [flink] parse primary key and schema by historyRecord
Posted by "yuzelin (via GitHub)" <gi...@apache.org>.
yuzelin commented on code in PR #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621#discussion_r1271658190
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTableSchemaBuilder.java:
##########
@@ -53,50 +47,40 @@ public MySqlTableSchemaBuilder(Map<String, String> tableConfig, boolean caseSens
}
@Override
- public Optional<Schema> build(MySqlCreateTableStatement statement) {
- List<SQLTableElement> columns = statement.getTableElementList();
- LinkedHashMap<String, Tuple2<DataType, String>> fields = new LinkedHashMap<>();
-
- for (SQLTableElement element : columns) {
- if (element instanceof SQLColumnDefinition) {
- SQLColumnDefinition column = (SQLColumnDefinition) element;
- SQLName name = column.getName();
- SQLDataType dataType = column.getDataType();
- List<SQLExpr> arguments = dataType.getArguments();
- Integer precision = null;
- Integer scale = null;
- if (arguments.size() >= 1) {
- precision = (int) (((SQLIntegerExpr) arguments.get(0)).getValue());
- }
-
- if (arguments.size() >= 2) {
- scale = (int) (((SQLIntegerExpr) arguments.get(1)).getValue());
- }
-
- SQLCharExpr comment = (SQLCharExpr) column.getComment();
- fields.put(
- name.getSimpleName(),
- Tuple2.of(
- MySqlTypeUtils.toDataType(
- column.getDataType().getName(),
- precision,
- scale,
- MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue()),
- comment == null ? null : String.valueOf(comment.getValue())));
- }
+ public Optional<Schema> build(JsonNode tableChange) {
+ JsonNode jsonTable = tableChange.get("table");
+ String tableName = tableChange.get("id").asText();
+ ArrayNode columns = (ArrayNode) jsonTable.get("columns");
+ LinkedHashMap<String, DataType> fields = new LinkedHashMap<>();
+
+ for (JsonNode element : columns) {
+ Integer precision = element.has("length") ? element.get("length").asInt() : null;
+ Integer scale = element.has("scale") ? element.get("scale").asInt() : null;
+ fields.put(
+ element.get("name").asText(),
+ MySqlTypeUtils.toDataType(
+ element.get("typeExpression").asText(),
+ precision,
+ scale,
+ MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue())
+ .copy(element.get("optional").asBoolean()));
Review Comment:
cool～
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTableSchemaBuilder.java:
##########
@@ -53,50 +47,40 @@ public MySqlTableSchemaBuilder(Map<String, String> tableConfig, boolean caseSens
}
@Override
- public Optional<Schema> build(MySqlCreateTableStatement statement) {
- List<SQLTableElement> columns = statement.getTableElementList();
- LinkedHashMap<String, Tuple2<DataType, String>> fields = new LinkedHashMap<>();
-
- for (SQLTableElement element : columns) {
- if (element instanceof SQLColumnDefinition) {
- SQLColumnDefinition column = (SQLColumnDefinition) element;
- SQLName name = column.getName();
- SQLDataType dataType = column.getDataType();
- List<SQLExpr> arguments = dataType.getArguments();
- Integer precision = null;
- Integer scale = null;
- if (arguments.size() >= 1) {
- precision = (int) (((SQLIntegerExpr) arguments.get(0)).getValue());
- }
-
- if (arguments.size() >= 2) {
- scale = (int) (((SQLIntegerExpr) arguments.get(1)).getValue());
- }
-
- SQLCharExpr comment = (SQLCharExpr) column.getComment();
- fields.put(
- name.getSimpleName(),
- Tuple2.of(
- MySqlTypeUtils.toDataType(
- column.getDataType().getName(),
- precision,
- scale,
- MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue()),
- comment == null ? null : String.valueOf(comment.getValue())));
- }
+ public Optional<Schema> build(JsonNode tableChange) {
+ JsonNode jsonTable = tableChange.get("table");
+ String tableName = tableChange.get("id").asText();
+ ArrayNode columns = (ArrayNode) jsonTable.get("columns");
+ LinkedHashMap<String, DataType> fields = new LinkedHashMap<>();
+
+ for (JsonNode element : columns) {
+ Integer precision = element.has("length") ? element.get("length").asInt() : null;
+ Integer scale = element.has("scale") ? element.get("scale").asInt() : null;
+ fields.put(
+ element.get("name").asText(),
+ MySqlTypeUtils.toDataType(
+ element.get("typeExpression").asText(),
+ precision,
+ scale,
+ MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue())
+ .copy(element.get("optional").asBoolean()));
Review Comment:
cool～
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] zhangjun0x01 commented on a diff in pull request #1621: [flink] parse primary key and schema by historyRecord
Posted by "zhangjun0x01 (via GitHub)" <gi...@apache.org>.
zhangjun0x01 commented on code in PR #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621#discussion_r1270763271
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlDebeziumJsonEventParser.java:
##########
@@ -231,20 +227,26 @@ public Optional<Schema> parseNewTable() {
try {
String historyRecordString = historyRecord.asText();
- String ddl = objectMapper.readTree(historyRecordString).get("ddl").asText();
- if (Strings.isNullOrEmpty(ddl)) {
- return Optional.empty();
+ JsonNode tableChanges = objectMapper.readTree(historyRecordString).get("tableChanges");
+ if (tableChanges.size() != 1) {
+ throw new IllegalArgumentException(
+ "Invalid historyRecord, because tableChanges should contain exactly 1 item.\n"
+ + historyRecord.asText());
}
- SQLStatement statement = SQLUtils.parseSingleStatement(ddl, JdbcConstants.MYSQL);
- if (!(statement instanceof MySqlCreateTableStatement)) {
+ if (!tableChanges
+ .get(0)
+ .get("type")
+ .asText()
+ .equals(TableChanges.TableChangeType.CREATE.name())) {
return Optional.empty();
}
- MySqlCreateTableStatement createTableStatement = (MySqlCreateTableStatement) statement;
- List<String> primaryKeys = createTableStatement.getPrimaryKeyNames();
- String tableName = createTableStatement.getTableName();
- if (primaryKeys.isEmpty()) {
+ JsonNode primaryKeyColumnNames =
+ tableChanges.get(0).get("table").get("primaryKeyColumnNames");
+ if (primaryKeyColumnNames.size() == 0) {
+ String id = tableChanges.get(0).get("id").asText();
+ String tableName = id.replaceAll("\"", "").split("\\.")[1];
Review Comment:
yes, when I write other PR (https://github.com/apache/incubator-paimon/pull/1282), sync multiple mysql database to paimon, we need to distinguish tables with the same name under different databases， so we need to use the format of `dbName.tableName`. I will update the format in that PR.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlDebeziumJsonEventParser.java:
##########
@@ -231,20 +227,26 @@ public Optional<Schema> parseNewTable() {
try {
String historyRecordString = historyRecord.asText();
- String ddl = objectMapper.readTree(historyRecordString).get("ddl").asText();
- if (Strings.isNullOrEmpty(ddl)) {
- return Optional.empty();
+ JsonNode tableChanges = objectMapper.readTree(historyRecordString).get("tableChanges");
+ if (tableChanges.size() != 1) {
+ throw new IllegalArgumentException(
+ "Invalid historyRecord, because tableChanges should contain exactly 1 item.\n"
+ + historyRecord.asText());
}
- SQLStatement statement = SQLUtils.parseSingleStatement(ddl, JdbcConstants.MYSQL);
- if (!(statement instanceof MySqlCreateTableStatement)) {
+ if (!tableChanges
+ .get(0)
+ .get("type")
+ .asText()
+ .equals(TableChanges.TableChangeType.CREATE.name())) {
return Optional.empty();
}
- MySqlCreateTableStatement createTableStatement = (MySqlCreateTableStatement) statement;
- List<String> primaryKeys = createTableStatement.getPrimaryKeyNames();
- String tableName = createTableStatement.getTableName();
- if (primaryKeys.isEmpty()) {
+ JsonNode primaryKeyColumnNames =
+ tableChanges.get(0).get("table").get("primaryKeyColumnNames");
+ if (primaryKeyColumnNames.size() == 0) {
+ String id = tableChanges.get(0).get("id").asText();
+ String tableName = id.replaceAll("\"", "").split("\\.")[1];
Review Comment:
yes, when I write other PR (https://github.com/apache/incubator-paimon/pull/1282), sync multiple mysql database to paimon, we need to distinguish tables with the same name under different databases， so we need to use the format of `dbName.tableName`. I will update the format in that PR.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] yuzelin commented on a diff in pull request #1621: [flink] parse primary key and schema by historyRecord
Posted by "yuzelin (via GitHub)" <gi...@apache.org>.
yuzelin commented on code in PR #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621#discussion_r1270557733
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTableSchemaBuilder.java:
##########
@@ -53,50 +47,40 @@ public MySqlTableSchemaBuilder(Map<String, String> tableConfig, boolean caseSens
}
@Override
- public Optional<Schema> build(MySqlCreateTableStatement statement) {
- List<SQLTableElement> columns = statement.getTableElementList();
- LinkedHashMap<String, Tuple2<DataType, String>> fields = new LinkedHashMap<>();
-
- for (SQLTableElement element : columns) {
- if (element instanceof SQLColumnDefinition) {
- SQLColumnDefinition column = (SQLColumnDefinition) element;
- SQLName name = column.getName();
- SQLDataType dataType = column.getDataType();
- List<SQLExpr> arguments = dataType.getArguments();
- Integer precision = null;
- Integer scale = null;
- if (arguments.size() >= 1) {
- precision = (int) (((SQLIntegerExpr) arguments.get(0)).getValue());
- }
-
- if (arguments.size() >= 2) {
- scale = (int) (((SQLIntegerExpr) arguments.get(1)).getValue());
- }
-
- SQLCharExpr comment = (SQLCharExpr) column.getComment();
- fields.put(
- name.getSimpleName(),
- Tuple2.of(
- MySqlTypeUtils.toDataType(
- column.getDataType().getName(),
- precision,
- scale,
- MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue()),
- comment == null ? null : String.valueOf(comment.getValue())));
- }
+ public Optional<Schema> build(JsonNode tableChange) {
+ JsonNode jsonTable = tableChange.get("table");
+ String tableName = tableChange.get("id").asText();
+ ArrayNode columns = (ArrayNode) jsonTable.get("columns");
+ LinkedHashMap<String, DataType> fields = new LinkedHashMap<>();
+
+ for (JsonNode element : columns) {
+ Integer precision = element.has("length") ? element.get("length").asInt() : null;
+ Integer scale = element.has("scale") ? element.get("scale").asInt() : null;
+ fields.put(
+ element.get("name").asText(),
+ MySqlTypeUtils.toDataType(
+ element.get("typeExpression").asText(),
+ precision,
+ scale,
+ MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue())
+ .copy(element.get("optional").asBoolean()));
Review Comment:
I'm not familiar with debezium json, I want to know here can we find field comments? If we can, we should also get the comments.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlDebeziumJsonEventParser.java:
##########
@@ -231,20 +227,26 @@ public Optional<Schema> parseNewTable() {
try {
String historyRecordString = historyRecord.asText();
- String ddl = objectMapper.readTree(historyRecordString).get("ddl").asText();
- if (Strings.isNullOrEmpty(ddl)) {
- return Optional.empty();
+ JsonNode tableChanges = objectMapper.readTree(historyRecordString).get("tableChanges");
+ if (tableChanges.size() != 1) {
+ throw new IllegalArgumentException(
+ "Invalid historyRecord, because tableChanges should contain exactly 1 item.\n"
+ + historyRecord.asText());
}
- SQLStatement statement = SQLUtils.parseSingleStatement(ddl, JdbcConstants.MYSQL);
- if (!(statement instanceof MySqlCreateTableStatement)) {
+ if (!tableChanges
+ .get(0)
+ .get("type")
+ .asText()
+ .equals(TableChanges.TableChangeType.CREATE.name())) {
return Optional.empty();
}
- MySqlCreateTableStatement createTableStatement = (MySqlCreateTableStatement) statement;
- List<String> primaryKeys = createTableStatement.getPrimaryKeyNames();
- String tableName = createTableStatement.getTableName();
- if (primaryKeys.isEmpty()) {
+ JsonNode primaryKeyColumnNames =
+ tableChanges.get(0).get("table").get("primaryKeyColumnNames");
+ if (primaryKeyColumnNames.size() == 0) {
+ String id = tableChanges.get(0).get("id").asText();
+ String tableName = id.replaceAll("\"", "").split("\\.")[1];
Review Comment:
I think it's OK to use `id` in `LOG.debug`.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlDebeziumJsonEventParser.java:
##########
@@ -231,20 +227,26 @@ public Optional<Schema> parseNewTable() {
try {
String historyRecordString = historyRecord.asText();
- String ddl = objectMapper.readTree(historyRecordString).get("ddl").asText();
- if (Strings.isNullOrEmpty(ddl)) {
- return Optional.empty();
+ JsonNode tableChanges = objectMapper.readTree(historyRecordString).get("tableChanges");
+ if (tableChanges.size() != 1) {
+ throw new IllegalArgumentException(
+ "Invalid historyRecord, because tableChanges should contain exactly 1 item.\n"
+ + historyRecord.asText());
}
- SQLStatement statement = SQLUtils.parseSingleStatement(ddl, JdbcConstants.MYSQL);
- if (!(statement instanceof MySqlCreateTableStatement)) {
+ if (!tableChanges
+ .get(0)
Review Comment:
Too many tableChanges.get(0).
You can use a variable.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] zhangjun0x01 commented on a diff in pull request #1621: [flink] parse primary key and schema by historyRecord
Posted by "zhangjun0x01 (via GitHub)" <gi...@apache.org>.
zhangjun0x01 commented on code in PR #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621#discussion_r1270754045
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlDebeziumJsonEventParser.java:
##########
@@ -231,20 +227,26 @@ public Optional<Schema> parseNewTable() {
try {
String historyRecordString = historyRecord.asText();
- String ddl = objectMapper.readTree(historyRecordString).get("ddl").asText();
- if (Strings.isNullOrEmpty(ddl)) {
- return Optional.empty();
+ JsonNode tableChanges = objectMapper.readTree(historyRecordString).get("tableChanges");
+ if (tableChanges.size() != 1) {
+ throw new IllegalArgumentException(
+ "Invalid historyRecord, because tableChanges should contain exactly 1 item.\n"
+ + historyRecord.asText());
}
- SQLStatement statement = SQLUtils.parseSingleStatement(ddl, JdbcConstants.MYSQL);
- if (!(statement instanceof MySqlCreateTableStatement)) {
+ if (!tableChanges
+ .get(0)
Review Comment:
I update it
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] zhangjun0x01 commented on a diff in pull request #1621: [flink] parse primary key and schema by historyRecord
Posted by "zhangjun0x01 (via GitHub)" <gi...@apache.org>.
zhangjun0x01 commented on code in PR #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621#discussion_r1270753843
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTableSchemaBuilder.java:
##########
@@ -53,50 +47,40 @@ public MySqlTableSchemaBuilder(Map<String, String> tableConfig, boolean caseSens
}
@Override
- public Optional<Schema> build(MySqlCreateTableStatement statement) {
- List<SQLTableElement> columns = statement.getTableElementList();
- LinkedHashMap<String, Tuple2<DataType, String>> fields = new LinkedHashMap<>();
-
- for (SQLTableElement element : columns) {
- if (element instanceof SQLColumnDefinition) {
- SQLColumnDefinition column = (SQLColumnDefinition) element;
- SQLName name = column.getName();
- SQLDataType dataType = column.getDataType();
- List<SQLExpr> arguments = dataType.getArguments();
- Integer precision = null;
- Integer scale = null;
- if (arguments.size() >= 1) {
- precision = (int) (((SQLIntegerExpr) arguments.get(0)).getValue());
- }
-
- if (arguments.size() >= 2) {
- scale = (int) (((SQLIntegerExpr) arguments.get(1)).getValue());
- }
-
- SQLCharExpr comment = (SQLCharExpr) column.getComment();
- fields.put(
- name.getSimpleName(),
- Tuple2.of(
- MySqlTypeUtils.toDataType(
- column.getDataType().getName(),
- precision,
- scale,
- MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue()),
- comment == null ? null : String.valueOf(comment.getValue())));
- }
+ public Optional<Schema> build(JsonNode tableChange) {
+ JsonNode jsonTable = tableChange.get("table");
+ String tableName = tableChange.get("id").asText();
+ ArrayNode columns = (ArrayNode) jsonTable.get("columns");
+ LinkedHashMap<String, DataType> fields = new LinkedHashMap<>();
+
+ for (JsonNode element : columns) {
+ Integer precision = element.has("length") ? element.get("length").asInt() : null;
+ Integer scale = element.has("scale") ? element.get("scale").asInt() : null;
+ fields.put(
+ element.get("name").asText(),
+ MySqlTypeUtils.toDataType(
+ element.get("typeExpression").asText(),
+ precision,
+ scale,
+ MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue())
+ .copy(element.get("optional").asBoolean()));
Review Comment:
the new version debezium history record contains the table comment and column ,I add `TODO`, when we upgrade to flink cdc 2.4 , we can add the comment
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] zhangjun0x01 commented on a diff in pull request #1621: [flink] parse primary key and schema by historyRecord
Posted by "zhangjun0x01 (via GitHub)" <gi...@apache.org>.
zhangjun0x01 commented on code in PR #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621#discussion_r1270753843
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTableSchemaBuilder.java:
##########
@@ -53,50 +47,40 @@ public MySqlTableSchemaBuilder(Map<String, String> tableConfig, boolean caseSens
}
@Override
- public Optional<Schema> build(MySqlCreateTableStatement statement) {
- List<SQLTableElement> columns = statement.getTableElementList();
- LinkedHashMap<String, Tuple2<DataType, String>> fields = new LinkedHashMap<>();
-
- for (SQLTableElement element : columns) {
- if (element instanceof SQLColumnDefinition) {
- SQLColumnDefinition column = (SQLColumnDefinition) element;
- SQLName name = column.getName();
- SQLDataType dataType = column.getDataType();
- List<SQLExpr> arguments = dataType.getArguments();
- Integer precision = null;
- Integer scale = null;
- if (arguments.size() >= 1) {
- precision = (int) (((SQLIntegerExpr) arguments.get(0)).getValue());
- }
-
- if (arguments.size() >= 2) {
- scale = (int) (((SQLIntegerExpr) arguments.get(1)).getValue());
- }
-
- SQLCharExpr comment = (SQLCharExpr) column.getComment();
- fields.put(
- name.getSimpleName(),
- Tuple2.of(
- MySqlTypeUtils.toDataType(
- column.getDataType().getName(),
- precision,
- scale,
- MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue()),
- comment == null ? null : String.valueOf(comment.getValue())));
- }
+ public Optional<Schema> build(JsonNode tableChange) {
+ JsonNode jsonTable = tableChange.get("table");
+ String tableName = tableChange.get("id").asText();
+ ArrayNode columns = (ArrayNode) jsonTable.get("columns");
+ LinkedHashMap<String, DataType> fields = new LinkedHashMap<>();
+
+ for (JsonNode element : columns) {
+ Integer precision = element.has("length") ? element.get("length").asInt() : null;
+ Integer scale = element.has("scale") ? element.get("scale").asInt() : null;
+ fields.put(
+ element.get("name").asText(),
+ MySqlTypeUtils.toDataType(
+ element.get("typeExpression").asText(),
+ precision,
+ scale,
+ MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue())
+ .copy(element.get("optional").asBoolean()));
Review Comment:
the new version debezium history record contains the table comment and column comment,I add `TODO`, when we upgrade to flink cdc 2.4 , we can add the comment
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] zhangjun0x01 commented on a diff in pull request #1621: [flink] parse primary key and schema by historyRecord
Posted by "zhangjun0x01 (via GitHub)" <gi...@apache.org>.
zhangjun0x01 commented on code in PR #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621#discussion_r1271662773
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTableSchemaBuilder.java:
##########
@@ -53,50 +47,40 @@ public MySqlTableSchemaBuilder(Map<String, String> tableConfig, boolean caseSens
}
@Override
- public Optional<Schema> build(MySqlCreateTableStatement statement) {
- List<SQLTableElement> columns = statement.getTableElementList();
- LinkedHashMap<String, Tuple2<DataType, String>> fields = new LinkedHashMap<>();
-
- for (SQLTableElement element : columns) {
- if (element instanceof SQLColumnDefinition) {
- SQLColumnDefinition column = (SQLColumnDefinition) element;
- SQLName name = column.getName();
- SQLDataType dataType = column.getDataType();
- List<SQLExpr> arguments = dataType.getArguments();
- Integer precision = null;
- Integer scale = null;
- if (arguments.size() >= 1) {
- precision = (int) (((SQLIntegerExpr) arguments.get(0)).getValue());
- }
-
- if (arguments.size() >= 2) {
- scale = (int) (((SQLIntegerExpr) arguments.get(1)).getValue());
- }
-
- SQLCharExpr comment = (SQLCharExpr) column.getComment();
- fields.put(
- name.getSimpleName(),
- Tuple2.of(
- MySqlTypeUtils.toDataType(
- column.getDataType().getName(),
- precision,
- scale,
- MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue()),
- comment == null ? null : String.valueOf(comment.getValue())));
- }
+ public Optional<Schema> build(JsonNode tableChange) {
+ JsonNode jsonTable = tableChange.get("table");
+ String tableName = tableChange.get("id").asText();
+ ArrayNode columns = (ArrayNode) jsonTable.get("columns");
+ LinkedHashMap<String, DataType> fields = new LinkedHashMap<>();
+
+ for (JsonNode element : columns) {
+ Integer precision = element.has("length") ? element.get("length").asInt() : null;
+ Integer scale = element.has("scale") ? element.get("scale").asInt() : null;
+ fields.put(
+ element.get("name").asText(),
+ MySqlTypeUtils.toDataType(
+ element.get("typeExpression").asText(),
+ precision,
+ scale,
+ MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue())
+ .copy(element.get("optional").asBoolean()));
Review Comment:
new version debezium history record also contains the column default value , we can sync it also.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi merged pull request #1621: [flink] parse primary key and schema by historyRecord
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #1621:
URL: https://github.com/apache/incubator-paimon/pull/1621
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org