ZhaoRuidong opened a new issue, #5176: URL: https://github.com/apache/seatunnel/issues/5176
### Search before asking - [X] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened Create an ETL job like [ mysql source -> filter -> click sink] . Row data's column order will change after Filter transform component, but Clickhouse sink writer still set column to prepare statement by database column index. For example, here is a ck table ： ```sql CREATE TABLE test.emps2 ( `EMPNO` Int32, `ENAME` String, `JOB` String, `MGR` Int32, `HIREDATE` DateTime, `SAL` Int32, `COMM` Int32, `DEPTNO` Int32 ) ENGINE = MergeTree PRIMARY KEY EMPNO ORDER BY EMPNO SETTINGS index_granularity = 8192; ``` 1 . Extract data use query like 'select DEPTNO, EMPNO, ENAME from emps1' from mysql table with the same structure. 2. Add filter after mysql source, collect [DEPTNO, EMPNO] 3. Sink data to Clickhouse table. After sink writer it will set DEPTNO's value to EMPNO, EMPNO's to DEPTNO. It caused by JdbcRowConverter inject prepare statement values in the order of column index in table ```java public PreparedStatement toExternal(SeaTunnelRow row, PreparedStatement statement) throws SQLException { for (int i = 0; i < projectionFields.length; i++) { String fieldName = projectionFields[i]; Object fieldValue = fieldGetterMap.get(fieldName).apply(row); if (fieldValue == null) { // field does not exist in row // todo: do we need to transform to default value of each type statement.setObject(i + 1, null); continue; } fieldInjectFunctionMap .getOrDefault(fieldName, DEFAULT_INJECT_FUNCTION) .injectFields(statement, i + 1, fieldValue); } return statement; } ``` ### SeaTunnel Version 2.3.1 ### SeaTunnel Config ```conf { "env" : { "job.mode" : "BATCH", "execution.parallelism" : 1 }, "source" : [ { "url" : "jdbc:mysql://127.0.0.1:3306/demo?useSSL=false", "driver" : "com.mysql.cj.jdbc.Driver", "user" : "root", "password" : "root", "query" : "SELECT DEPTNO,EMPNO, ENAME, JOB, MGR, HIREDATE, SAL, COMM FROM emps", "plugin_name" : "Jdbc", "result_table_name" : "m1", "connection_check_timeout_sec" : 30, "fetch_size" : 0 } ], "transform" : [ { "fields" : [ "ENAME", "DEPTNO", "EMPNO" ], "plugin_name" : "Filter", "result_table_name" : "f11", "source_table_name" : "m1" } ], "sink" : [ { "host" : "127.0.0.1:8123", "database" : "test", "table" : "emps2", "username" : "default", "password" : "root", "plugin_name" : "Clickhouse", "source_table_name" : "f11", "clickhouse.config" : {}, "bulk_size" : "20000", "split_mode" : "false", "support_upsert" : false, "allow_experimental_lightweight_delete" : false } ] } ``` ### Running Command ```shell I run in debug local mode. ``` ### Error Exception ```log Not necessary. ``` ### Zeta or Flink or Spark Version _No response_ ### Java or Scala Version _No response_ ### Screenshots _No response_ ### 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) -- 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]
