diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
index e59776b6f95..f98f2cb3129 100644
--- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
+++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
@@ -40,8 +40,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.mysql.cj.MysqlType;
-
 import java.io.Serializable;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -532,8 +530,7 @@ default String buildAlterTableSql(
                     "ALTER TABLE %s drop column %s", tableName, quoteIdentifier(oldColumnName));
         }
         TypeConverter<?> typeConverter = ConverterLoader.loadTypeConverter(dialectName());
-        BasicTypeDefine<MysqlType> typeBasicTypeDefine =
-                (BasicTypeDefine<MysqlType>) typeConverter.reconvert(newColumn);
+        BasicTypeDefine typeBasicTypeDefine = (BasicTypeDefine) typeConverter.reconvert(newColumn);
 
         String basicSql = buildAlterTableBasicSql(alterOperation, tableName);
         basicSql =
@@ -616,8 +613,7 @@ default String decorateWithColumnNameAndType(
      * @param typeBasicTypeDefine type basic type define of new column
      * @return alter table sql with nullable for sink table
      */
-    default String decorateWithNullable(
-            String basicSql, BasicTypeDefine<MysqlType> typeBasicTypeDefine) {
+    default String decorateWithNullable(String basicSql, BasicTypeDefine typeBasicTypeDefine) {
         StringBuilder sql = new StringBuilder(basicSql);
         if (typeBasicTypeDefine.isNullable()) {
             sql.append("NULL ");
@@ -634,8 +630,7 @@ default String decorateWithNullable(
      * @param typeBasicTypeDefine type basic type define of new column
      * @return alter table sql with default value for sink table
      */
-    default String decorateWithDefaultValue(
-            String basicSql, BasicTypeDefine<MysqlType> typeBasicTypeDefine) {
+    default String decorateWithDefaultValue(String basicSql, BasicTypeDefine typeBasicTypeDefine) {
         Object defaultValue = typeBasicTypeDefine.getDefaultValue();
         if (Objects.nonNull(defaultValue)
                 && needsQuotesWithDefaultValue(typeBasicTypeDefine.getColumnType())
@@ -656,8 +651,7 @@ && needsQuotesWithDefaultValue(typeBasicTypeDefine.getColumnType())
      * @param typeBasicTypeDefine type basic type define of new column
      * @return alter table sql with comment for sink table
      */
-    default String decorateWithComment(
-            String basicSql, BasicTypeDefine<MysqlType> typeBasicTypeDefine) {
+    default String decorateWithComment(String basicSql, BasicTypeDefine typeBasicTypeDefine) {
         String comment = typeBasicTypeDefine.getComment();
         StringBuilder sql = new StringBuilder(basicSql);
         if (StringUtils.isNotBlank(comment)) {
diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
index 73ef12bc47b..fd0af3d9ffd 100644
--- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
+++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
@@ -245,13 +245,12 @@ public void refreshTableSchemaBySchemaChangeEvent(
     }
 
     @Override
-    public String decorateWithComment(
-            String basicSql, BasicTypeDefine<MysqlType> mysqlTypeBasicTypeDefine) {
-        MysqlType nativeType = mysqlTypeBasicTypeDefine.getNativeType();
+    public String decorateWithComment(String basicSql, BasicTypeDefine typeBasicTypeDefine) {
+        MysqlType nativeType = (MysqlType) typeBasicTypeDefine.getNativeType();
         if (NOT_SUPPORTED_DEFAULT_VALUES.contains(nativeType)) {
             return basicSql;
         }
-        return JdbcDialect.super.decorateWithComment(basicSql, mysqlTypeBasicTypeDefine);
+        return JdbcDialect.super.decorateWithComment(basicSql, typeBasicTypeDefine);
     }
 
     @Override