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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
