[Fix][Connector-V2] Make JDBC dialect type converters locale independent

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/dm/DamengDataTypeConvertor.java

@@ -30,6 +30,7 @@
 import com.google.auto.service.AutoService;
 
 import java.util.Collections;
+import java.util.Locale;
 import java.util.Map;
 
 import static org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.dm.DmdbTypeConverter.DM_DEC;
@@ -61,7 +62,7 @@ public SeaTunnelDataType<?> toSeaTunnelType(
             String field, String dataType, Map<String, Object> properties) {
         Integer precision = null;
         Integer scale = null;
-        switch (dataType.toUpperCase()) {
+        switch (dataType.toUpperCase(Locale.ROOT)) {
             case DM_NUMERIC:
             case DM_NUMBER:
             case DM_DECIMAL:

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/duckdb/DuckDBCatalog.java

@@ -42,6 +42,7 @@
 import java.util.Enumeration;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
@@ -262,6 +263,6 @@ protected String getOptionTableName(TablePath tablePath) {
     }
 
     private boolean isDuckDBDecimal(String typeName) {
-        return typeName.toUpperCase().startsWith(DuckDBTypeConverter.DUCKDB_DECIMAL);
+        return typeName.toUpperCase(Locale.ROOT).startsWith(DuckDBTypeConverter.DUCKDB_DECIMAL);
     }
 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseOracleCreateTableSqlBuilder.java

@@ -25,6 +25,8 @@
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter;
 
+import java.util.Locale;
+
 public class OceanBaseOracleCreateTableSqlBuilder extends OracleCreateTableSqlBuilder {
 
     public OceanBaseOracleCreateTableSqlBuilder(CatalogTable catalogTable, boolean createIndex) {
@@ -46,7 +48,7 @@ protected String buildColumnSql(Column column) {
                 // handle OceanBase Oracle compatible mode unsupported types, please refer
                 // https://www.oceanbase.com/docs/enterprise-oceanbase-database-cn-10000000000355002
                 // and https://www.oceanbase.com/docs/enterprise-oms-doc-cn-1000000002530110
-                switch (column.getSourceType().toUpperCase()) {
+                switch (column.getSourceType().toUpperCase(Locale.ROOT)) {
                     case OracleTypeConverter.ORACLE_LONG:
                         columnType = OracleTypeConverter.ORACLE_CLOB;
                         break;

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/redshift/RedshiftCatalog.java

@@ -33,6 +33,7 @@
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.Locale;
 
 @Slf4j
 public class RedshiftCatalog extends AbstractJdbcCatalog {
@@ -82,7 +83,7 @@ protected String getTableName(ResultSet rs) throws SQLException {
                 .append(".")
                 .append(rs.getString(2))
                 .toString()
-                .toLowerCase();
+                .toLowerCase(Locale.ROOT);
     }
 
     @Override
@@ -133,7 +134,7 @@ protected TableIdentifier getTableIdentifier(TablePath tablePath) {
     @Override
     protected Column buildColumn(ResultSet resultSet) throws SQLException {
         String columnName = resultSet.getString("COLUMN_NAME");
-        String typeName = resultSet.getString("DATA_TYPE").toUpperCase();
+        String typeName = resultSet.getString("DATA_TYPE").toUpperCase(Locale.ROOT);
         long precision = resultSet.getLong("NUMERIC_PRECISION");
         int scale = resultSet.getInt("NUMERIC_SCALE");
         long columnLength = resultSet.getLong("CHARACTER_MAXIMUM_LENGTH");

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/redshift/RedshiftDataTypeConvertor.java

@@ -31,6 +31,7 @@
 import com.google.auto.service.AutoService;
 
 import java.util.Collections;
+import java.util.Locale;
 import java.util.Map;
 
 import static org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
@@ -58,7 +59,7 @@ public SeaTunnelDataType<?> toSeaTunnelType(
 
         Integer precision = null;
         Integer scale = null;
-        switch (connectorDataType.toUpperCase()) {
+        switch (connectorDataType.toUpperCase(Locale.ROOT)) {
             case RedshiftTypeConverter.PG_NUMERIC:
                 precision = MapUtils.getInteger(dataTypeProperties, PRECISION, DEFAULT_PRECISION);
                 scale = MapUtils.getInteger(dataTypeProperties, SCALE, DEFAULT_SCALE);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/xugu/XuguCatalog.java

@@ -41,6 +41,7 @@
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
@@ -157,13 +158,13 @@ public boolean databaseExists(String databaseName) throws CatalogException {
         }
         try {
             return querySQLResultExists(
-                    defaultUrl, getDatabaseWithConditionSql(databaseName.toUpperCase()));
+                    defaultUrl, getDatabaseWithConditionSql(databaseName.toUpperCase(Locale.ROOT)));
         } catch (SeaTunnelRuntimeException e) {
             if (e.getSeaTunnelErrorCode().getCode().equals(UNSUPPORTED_METHOD.getCode())) {
                 log.warn(
                         "The catalog: {} is not supported the getDatabaseWithConditionSql for databaseExists",
                         this.catalogName);
-                return listDatabases().contains(databaseName.toUpperCase());
+                return listDatabases().contains(databaseName.toUpperCase(Locale.ROOT));
             }
             throw e;
         } catch (SQLException e) {

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/yashandb/YashanDbCatalog.java

@@ -44,6 +44,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Locale;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -153,7 +154,11 @@ protected Column buildColumn(ResultSet resultSet) throws SQLException {
         // DATA_TYPE from all_tab_columns may include precision like TIMESTAMP(6),
         // strip parenthesized digits to get the pure type name
         String typeName =
-                resultSet.getString("DATA_TYPE").toUpperCase().replaceAll("\\(\\d+\\)", "").trim();
+                resultSet
+                        .getString("DATA_TYPE")
+                        .toUpperCase(Locale.ROOT)
+                        .replaceAll("\\(\\d+\\)", "")
+                        .trim();
         // e.g NUMBER(10, 2)
         String fullTypeName = resultSet.getString("FULL_TYPE_NAME");
         long columnLength = resultSet.getLong("COLUMN_LENGTH");

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2TypeConverter.java

@@ -32,6 +32,8 @@
 import com.google.auto.service.AutoService;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Locale;
+
 // reference https://www.ibm.com/docs/en/db2/11.5?topic=statements-create-table#r0000927__title__52
 @Slf4j
 @AutoService(TypeConverter.class)
@@ -106,7 +108,7 @@ public Column convert(BasicTypeDefine typeDefine) {
                         .defaultValue(typeDefine.getDefaultValue())
                         .comment(typeDefine.getComment());
 
-        String db2Type = typeDefine.getDataType().toUpperCase();
+        String db2Type = typeDefine.getDataType().toUpperCase(Locale.ROOT);
         switch (db2Type) {
             case DB2_BOOLEAN:
                 builder.sourceType(DB2_BOOLEAN);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverter.java

@@ -32,6 +32,8 @@
 import com.google.auto.service.AutoService;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Locale;
+
 // reference https://eco.dameng.com/document/dm/zh-cn/sql-dev/dmpl-sql-datatype.html
 @Slf4j
 @AutoService(TypeConverter.class)
@@ -125,7 +127,7 @@ public Column convert(BasicTypeDefine typeDefine) {
                         .defaultValue(typeDefine.getDefaultValue())
                         .comment(typeDefine.getComment());
 
-        String dmType = typeDefine.getDataType().toUpperCase();
+        String dmType = typeDefine.getDataType().toUpperCase(Locale.ROOT);
         if (dmType.startsWith("TIMESTAMP(") && dmType.endsWith(" WITH LOCAL TIME ZONE")) {
             dmType = DM_TIMESTAMP_WITH_LOCAL_TIME_ZONE;
         }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/gbase8a/Gbase8aTypeMapper.java

@@ -30,6 +30,7 @@
 
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
+import java.util.Locale;
 
 @Slf4j
 public class Gbase8aTypeMapper implements JdbcDialectTypeMapper {
@@ -64,7 +65,7 @@ public class Gbase8aTypeMapper implements JdbcDialectTypeMapper {
     @Override
     public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex)
             throws SQLException {
-        String gbase8aType = metadata.getColumnTypeName(colIndex).toUpperCase();
+        String gbase8aType = metadata.getColumnTypeName(colIndex).toUpperCase(Locale.ROOT);
         int precision = metadata.getPrecision(colIndex);
         int scale = metadata.getScale(colIndex);
         switch (gbase8aType) {

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/hive/HiveTypeMapper.java

@@ -30,6 +30,7 @@
 
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
+import java.util.Locale;
 
 public class HiveTypeMapper implements JdbcDialectTypeMapper {
 
@@ -68,7 +69,7 @@ public class HiveTypeMapper implements JdbcDialectTypeMapper {
     @Override
     public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex)
             throws SQLException {
-        String columnType = metadata.getColumnTypeName(colIndex).toUpperCase();
+        String columnType = metadata.getColumnTypeName(colIndex).toUpperCase(Locale.ROOT);
         int precision = metadata.getPrecision(colIndex);
         switch (columnType) {
             case HIVE_TINYINT:

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/iris/IrisTypeConverter.java

@@ -31,6 +31,7 @@
 import com.google.auto.service.AutoService;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Locale;
 import java.util.Objects;
 
 /**
@@ -144,7 +145,7 @@ public Column convert(BasicTypeDefine typeDefine) {
                         .nullable(typeDefine.isNullable())
                         .defaultValue(typeDefine.getDefaultValue())
                         .comment(typeDefine.getComment());
-        String irisDataType = typeDefine.getDataType().toUpperCase();
+        String irisDataType = typeDefine.getDataType().toUpperCase(Locale.ROOT);
         long charOrBinaryLength =
                 Objects.nonNull(typeDefineLength) && typeDefineLength > 0 ? typeDefineLength : 1;
         switch (irisDataType) {

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/kingbase/KingbaseTypeConverter.java

@@ -37,6 +37,7 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.sql.Types;
+import java.util.Locale;
 
 // reference https://help.kingbase.com.cn/v8/development/sql-plsql/sql/datatype.html#id2
 @Slf4j
@@ -74,7 +75,7 @@ public Column convert(BasicTypeDefine typeDefine) {
                             .defaultValue(typeDefine.getDefaultValue())
                             .comment(typeDefine.getComment());
 
-            String kingbaseDataType = typeDefine.getDataType().toUpperCase();
+            String kingbaseDataType = typeDefine.getDataType().toUpperCase(Locale.ROOT);
             switch (kingbaseDataType) {
                     // MySQL compatibility - only types not in PostgresTypeConverter
                     // int not in PG (PG has SMALLINT/INTEGER/BIGINT)

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java

@@ -205,7 +205,7 @@ public Long approximateRowCntStatement(Connection connection, JdbcSourceTable ta
 
         boolean useTableStats =
                 StringUtils.isBlank(table.getQuery())
-                        || (!table.getQuery().toLowerCase().contains("where")
+                        || (!table.getQuery().toLowerCase(Locale.ROOT).contains("where")
                                 && table.getTablePath() != null
                                 && !TablePath.DEFAULT
                                         .getFullName()

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlDialect.java

@@ -42,6 +42,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
@@ -204,7 +205,7 @@ public Long approximateRowCntStatement(Connection connection, JdbcSourceTable ta
 
         boolean useTableStats =
                 StringUtils.isBlank(table.getQuery())
-                        || (!table.getQuery().toLowerCase().contains("where")
+                        || (!table.getQuery().toLowerCase(Locale.ROOT).contains("where")
                                 && table.getTablePath() != null
                                 && !TablePath.DEFAULT
                                         .getFullName()

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java

@@ -51,6 +51,7 @@
 import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -242,7 +243,7 @@ public Long approximateRowCntStatement(Connection connection, JdbcSourceTable ta
 
         boolean useTableStats =
                 StringUtils.isBlank(query)
-                        || (!query.toLowerCase().contains("where")
+                        || (!query.toLowerCase(Locale.ROOT).contains("where")
                                 && table.getTablePath() != null
                                 && !TablePath.DEFAULT
                                         .getFullName()

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleTypeConverter.java

@@ -33,6 +33,8 @@
 import com.google.auto.service.AutoService;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Locale;
+
 // reference https://docs.oracle.com/en/database/oracle/oracle-database/19/sqlrf/Data-Types.html
 @Slf4j
 @AutoService(TypeConverter.class)
@@ -127,7 +129,7 @@ public Column convert(BasicTypeDefine typeDefine) {
                         .defaultValue(typeDefine.getDefaultValue())
                         .comment(typeDefine.getComment());
 
-        String oracleType = typeDefine.getDataType().toUpperCase();
+        String oracleType = typeDefine.getDataType().toUpperCase(Locale.ROOT);
         if (oracleType.startsWith(ORACLE_INTERVAL)) {
             // The JDBC driver may report precision qualifiers in the type name,
             // e.g. "INTERVAL DAY(2) TO SECOND(6)", strip them for exact matching.

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/phoenix/PhoenixTypeConverter.java

@@ -32,6 +32,8 @@
 
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Locale;
+
 import static org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier.PHOENIX;
 
 // reference https://phoenix.apache.org/language/datatypes.html
@@ -98,7 +100,7 @@ public Column convert(BasicTypeDefine typeDefine) {
                         .defaultValue(typeDefine.getDefaultValue())
                         .comment(typeDefine.getComment());
 
-        String phoenixDataType = typeDefine.getDataType().toUpperCase();
+        String phoenixDataType = typeDefine.getDataType().toUpperCase(Locale.ROOT);
         switch (phoenixDataType) {
             case PHOENIX_BOOLEAN:
                 builder.dataType(BasicType.BOOLEAN_TYPE);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/presto/PrestoTypeMapper.java

@@ -28,6 +28,7 @@
 
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
+import java.util.Locale;
 
 public class PrestoTypeMapper implements JdbcDialectTypeMapper {
     // ============================data types=====================
@@ -66,7 +67,7 @@ public class PrestoTypeMapper implements JdbcDialectTypeMapper {
     @Override
     public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex)
             throws SQLException {
-        String columnType = metadata.getColumnTypeName(colIndex).toUpperCase();
+        String columnType = metadata.getColumnTypeName(colIndex).toUpperCase(Locale.ROOT);
         // VARCHAR(x)      --->      VARCHAR
         if (columnType.indexOf("(") > -1) {
             columnType = columnType.split("\\(")[0];

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresDialect.java

@@ -50,6 +50,7 @@
 import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -275,7 +276,7 @@ public Long approximateRowCntStatement(Connection connection, JdbcSourceTable ta
 
         boolean useTableStats =
                 StringUtils.isBlank(table.getQuery())
-                        || (!table.getQuery().toLowerCase().contains("where")
+                        || (!table.getQuery().toLowerCase(Locale.ROOT).contains("where")
                                 && table.getTablePath() != null
                                 && !TablePath.DEFAULT
                                         .getFullName()
@@ -360,7 +361,7 @@ public void applySchemaChange(
 
     @Override
     public boolean needsQuotesWithDefaultValue(BasicTypeDefine columnDefine) {
-        String pgDataType = columnDefine.getDataType().toLowerCase();
+        String pgDataType = columnDefine.getDataType().toLowerCase(Locale.ROOT);
         switch (pgDataType) {
             case PG_CHAR:
             case PG_VARCHAR:

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresTypeConverter.java

@@ -35,6 +35,7 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.sql.Types;
+import java.util.Locale;
 import java.util.regex.Pattern;
 
 // reference http://www.postgres.cn/docs/13/datatype.html
@@ -141,7 +142,7 @@ public Column convert(BasicTypeDefine typeDefine) {
                         .defaultValue(typeDefine.getDefaultValue())
                         .comment(typeDefine.getComment());
 
-        String pgDataType = typeDefine.getDataType().toLowerCase();
+        String pgDataType = typeDefine.getDataType().toLowerCase(Locale.ROOT);
         switch (pgDataType) {
             case PG_BOOLEAN:
                 builder.dataType(BasicType.BOOLEAN_TYPE);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/redshift/RedshiftDialect.java

@@ -33,6 +33,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Locale;
 import java.util.Optional;
 
 public class RedshiftDialect implements JdbcDialect {
@@ -145,7 +146,7 @@ public Long approximateRowCntStatement(Connection connection, JdbcSourceTable ta
 
         boolean useTableStats =
                 StringUtils.isBlank(table.getQuery())
-                        || (!table.getQuery().toLowerCase().contains("where")
+                        || (!table.getQuery().toLowerCase(Locale.ROOT).contains("where")
                                 && table.getTablePath() != null
                                 && !TablePath.DEFAULT
                                         .getFullName()