[flink] parse primary key and schema by historyRecord #1621
Conversation
| if (!tableChanges | ||
| .get(0) |
There was a problem hiding this comment.
Too many tableChanges.get(0).
You can use a variable.
| String id = tableChanges.get(0).get("id").asText(); | ||
| String tableName = id.replaceAll("\"", "").split("\\.")[1]; |
There was a problem hiding this comment.
I think it's OK to use id in LOG.debug.
There was a problem hiding this comment.
yes, when I write other PR (#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.
| precision, | ||
| scale, | ||
| MYSQL_CONVERTER_TINYINT1_BOOL.defaultValue()) | ||
| .copy(element.get("optional").asBoolean())); |
There was a problem hiding this 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.
There was a problem hiding this 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
There was a problem hiding this comment.
new version debezium history record also contains the column default value , we can sync it also.
zhangjun0x01
force-pushed
the
parse_by_history_record
branch
from
63a6468 to
00defaa
Compare
zhangjun0x01
force-pushed
the
parse_by_history_record
branch
from
c4705b4 to
6ad0973
Compare
Purpose
Linked issue: close #1613
Tests
API and Format
Documentation