[Bug] [jdbc-base] binlog_mysql_transform does not support deleting records with null values.
- Dominant language
- Java
- Stars
- 4.1k
- Forks
- 1.7k
- PR merge metrics
- No merged PRs in 30d
Description
### Search before asking
- [X] I had searched in the [issues](https://github.com/DTStack/chunjun/issues) and found no similar issues.
### What happened
binlog_mysql_transform当源表一行记录某个字段有null值时,source端执行update和delete时,sink端update会多一条记录(目标有主键报主键冲突错误,delete不生效。
### What you expected to happen
binlog_mysql_transform,source端执行update和delete时,sink端执行同样的命令生效。
### How to reproduce
json:
{
"job": {
"content": [
{
"reader": {
"parameter": {
"password": "***",
"splitUpdate": true,
"port": "3306",
"cat": "update,insert,delete",
"jdbcUrl": "jdbc:mysql://127.0.0.1:3306/test?characterEncoding=utf8&allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false",
"host": "127.0.0.1",
"pavingData": true,
"column": [
{
"index": 0,
"name": "test_int",
"type": "int"
},
{
"index": 1,
"name": "test_float",
"type": "float"
},
{
"index": 2,
"name": "test_double",
"type": "double"
},
{
"index": 3,
"name": "test_varchar",
"type": "varchar"
},
{
"index": 4,
"name": "test_char",
"type": "char"
},
{
"index": 5,
"name": "test_datetime",
"type": "datetime"
}
],
"table": [
"src_test_table"
],
"username": "root"
},
"name": "binlogsource",
"table": {
"tableName": "sourceTable"
}
},
"transformer": {
"transformSql": "SELECT test_int,test_float,test_double,test_varchar,test_char,test_datetime FROM sourceTable"
},
"writer": {
"parameter": {
"password": "***",
"column": [
{
"index": 0,
"name": "test_int",
"type": "int"
},
{
"index": 1,
"name": "test_float",
"type": "float"
},
{
"index": 2,
"name": "test_double",
"type": "double"
},
{
"index": 3,
"name": "test_varchar",
"type": "varchar"
},
{
"index": 4,
"name": "test_char",
"type": "char"
},
{
"index": 5,
"name": "test_datetime",
"type": "datetime"
}
],
"connection": [
{
"schema": "test",
"jdbcUrl": "jdbc:mysql://127.0.0.1:3306/test?characterEncoding=utf8&allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false",
"table": [
"sink_test_table"
]
}
],
"writeMode": "insert",
"batchSize": 1024,
"username": "root"
},
"name": "mysqlwriter",
"table": {
"tableName": "sinkTable"
}
}
}
],
"setting": {
"restore": {
"isStream": true,
"isRestore": true
},
"log": {
"path": "",
"level": "debug",
"pattern": "",
"isLogger": false
},
"errorLimit": {},
"speed": {
"readerChannel": 1,
"writerChannel": 1,
"bytes": 0
}
}
}
}
source执行:
INSERT INTO `test`.`src_test_table`(`test_int`, `test_float`, `test_double`, `test_varchar`, `test_char`) VALUES (4, NULL, 2.69, 'DDD', '中华武术');
UPDATE `test`.`src_test_table` SET `test_float` = 1.18, `test_double` = 2.69, `test_varchar` = NULL, `test_char` = '中华武术' WHERE `test_int` = 4;
sink端记录:

### Anything else
_No response_
### Version
master
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
Assessment
This issue has not been assessed yet.