FIX: change converting type to bigDecimal

Author: suhwan-cheon

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtils.java

@@ -22,6 +22,7 @@
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.TableId;
 
+import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -79,14 +80,14 @@ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
 
     // PreparedStatement#setObject method will be converted to long type when handling bigint
     // unsigned, which poses a data overflow issue.
-    // Therefore, we need to handle the overflow issue by converting the long value to BigInteger.
+    // Therefore, we need to handle the overflow issue by converting the bigint unsigned to BigDecimal.
     public static void setSafeObject(PreparedStatement ps, int parameterIndex, Object value)
             throws SQLException {
-        if (value instanceof Long && (Long) value < 0L) {
-            ps.setObject(parameterIndex, new BigInteger(Long.toUnsignedString((Long) value)));
-            return;
+        if (value instanceof BigInteger) {
+            ps.setBigDecimal(parameterIndex, new BigDecimal((BigInteger) value));
+        } else {
+            ps.setObject(parameterIndex, value);
         }
-        ps.setObject(parameterIndex, value);
     }
 
     public static Object queryMin(

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java

@@ -1513,6 +1513,7 @@ void testUnsignedBigintPrimaryKeyChunking() throws Exception {
                         .deserializer(deserializer)
                         .startupOptions(StartupOptions.initial())
                         .chunkKeyColumn(new ObjectPath(db, table), "order_id")
+                        .splitSize(2)
                         .build();
 
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();