[Bug] [jdbc-connector] MysqlSource 中 tinyint 被处理成了 BIT

[PorterXie]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

使用 Mysql 作为输入输出时，source 配置中仅设置了 query，在 处理 tinyint(1) 类型的 column， 被解析成了 BIT 类型，导致数据值丢失

源表：
CREATE TABLE person_3 (
id INT(11) NOT NULL AUTO_INCREMENT,
mob VARCHAR(20) DEFAULT NULL COMMENT '',
card VARCHAR(18) DEFAULT NULL COMMENT '',
nation VARCHAR(10) DEFAULT NULL COMMENT '',
test TINYINT(1) DEFAULT NULL,
phone VARCHAR(20) DEFAULT NULL COMMENT '',
PRIMARY KEY (id)
) ENGINE=INNODB AUTO_INCREMENT=723784 DEFAULT CHARSET=utf8

目标表：

CREATE TABLE xy_person_3 (
id INT(11) NOT NULL AUTO_INCREMENT,
mob VARCHAR(20) DEFAULT NULL COMMENT '',
card VARCHAR(18) DEFAULT NULL COMMENT '',
nation VARCHAR(10) DEFAULT NULL COMMENT '',
test TINYINT(1) DEFAULT NULL,
phone VARCHAR(20) DEFAULT NULL COMMENT '',
PRIMARY KEY (id)
) ENGINE=INNODB AUTO_INCREMENT=3454 DEFAULT CHARSET=utf8

错误原因：

错误分析代码：

问题分析：
我尝试了 5.x 及 8.x 的JDBC 驱动，通过 querySQL 获取的方式都会将 tinyint(1) 处理成 BIT，因此我认为不是驱动的问题，而通过 querySQL 获取元信息是不太合理的。

只设置了 querySql，那么肯定会出问题，如果同时设置了 table_path，在做 merge 的时候也会因为类型不匹配会被过滤掉
因此我采用修改了 JdbcCatalogUtils 类 mergeCatalogTable 的方法，去掉了类型相等的判断，因为按照目前这种情况来看，对类型相等的判断是不正确的，我同时设置在参数中设置 table_path，让其走 mergeCatalogTable 的流程

SeaTunnel Version

2.3.7

SeaTunnel Config

{
    "env":
    {
        "job.mode": "BATCH",
        "parallelism": 1,
        "job.name": "tinyint验证"
    },
    "source":
    [
        {
            "url": "jdbc:mysql://192.168.19.183:3306/s_end?CatalogMeansCurrent=true&characterEncoding=UTF-8",
            "user": "root",
            "password": "123456",
            "query": "select `phone`,`nation`,`card`,`id`,`mob`,`test` from `s_end`.`person_3`",
            "plugin_name": "Jdbc",
            "result_table_name": "person_3_845084699",
            "fetch_size": 5000
        }
    ],
    "sink":
    [
        {
            "url": "jdbc:mysql://192.168.19.183:3306/s1_end?CatalogMeansCurrent=true&characterEncoding=UTF-8",
            "user": "root",
            "password": "123456",
            "database": "s1_end",
            "table": "xy_person_3",
            "primary_keys":
            [
                "id"
            ],
            "batch_size": 5000,
            "generate_sink_sql": true,
            "schema_save_mode": "ERROR_WHEN_SCHEMA_NOT_EXIST",
            "data_save_mode": "DROP_DATA",
            "enable_upsert": false,
            "plugin_name": "Jdbc",
            "source_table_name": "person_3_845084699_t"
        }
    ],
    "transform":
    [
        {
            "query": "SELECT id,mob,card,nation,test,phone FROM person_3_845084699",
            "plugin_name": "Sql",
            "source_table_name": "person_3_845084699",
            "result_table_name": "person_3_845084699_t"
        }
    ]
}

Running Command

FlinkExecution flinkExecution = new FlinkExecution(config);
flinkExecution.execute();

Error Exception

没有错误，但是同步之后数据值丢失

Zeta or Flink or Spark Version

No response

Java or Scala Version

No response

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Carl-Zhou-CN]

You can try the mysql parameter 'tinyInt1isBit=false'

[PorterXie]

You can try the mysql parameter 'tinyInt1isBit=false'

是的，这样是可以的

[github-actions]

This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs.

[github-actions]

This issue has been closed because it has not received response for too long time. You could reopen it if you encountered similar problems in the future.