[Feature][Connector-V2][Jdbc] support gbase 8a

docs/en/connector-v2/sink/Jdbc.md

@@ -116,6 +116,7 @@ there are some reference value for params above.
 | phoenix    | org.apache.phoenix.queryserver.client.Driver | jdbc:phoenix:thin:url=http://localhost:8765;serialization=PROTOBUF | /                                                  | https://mvnrepository.com/artifact/com.aliyun.phoenix/ali-phoenix-shaded-thin-client |
 | sqlserver  | com.microsoft.sqlserver.jdbc.SQLServerDriver | jdbc:microsoft:sqlserver://localhost:1433                    | com.microsoft.sqlserver.jdbc.SQLServerXADataSource | https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc |
 | oracle     | oracle.jdbc.OracleDriver                     | jdbc:oracle:thin:@localhost:1521/xepdb1                      | oracle.jdbc.xa.OracleXADataSource                  | https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8 |
+| gbase8a    | com.gbase.jdbc.Driver                        | jdbc:gbase://e2e_gbase8aDb:5258/test                         | /                                                  | https://www.gbase8.cn/wp-content/uploads/2020/10/gbase-connector-java-8.3.81.53-build55.5.7-bin_min_mix.jar |
 
 ## Example
 

docs/en/connector-v2/source/Jdbc.md

@@ -98,6 +98,7 @@ there are some reference value for params above.
 | phoenix    | org.apache.phoenix.queryserver.client.Driver | jdbc:phoenix:thin:url=http://localhost:8765;serialization=PROTOBUF | https://mvnrepository.com/artifact/com.aliyun.phoenix/ali-phoenix-shaded-thin-client |
 | sqlserver  | com.microsoft.sqlserver.jdbc.SQLServerDriver | jdbc:microsoft:sqlserver://localhost:1433                    | https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc |
 | oracle     | oracle.jdbc.OracleDriver                     | jdbc:oracle:thin:@localhost:1521/xepdb1                      | https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8 |
+| gbase8a    | com.gbase.jdbc.Driver                        | jdbc:gbase://e2e_gbase8aDb:5258/test                         | https://www.gbase8.cn/wp-content/uploads/2020/10/gbase-connector-java-8.3.81.53-build55.5.7-bin_min_mix.jar |
 
 ## Example
 

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SqlDateType.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.api.table.type;
+
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Objects;
+
+public class SqlDateType<T> implements SeaTunnelDataType<T> {
+    private static final long serialVersionUID = 2L;
+
+    public static final SqlDateType<Date> SQL_DATE_TYPE = new SqlDateType<>(Date.class, SqlType.DATE);
+    public static final SqlDateType<Time> SQL_TIME_TYPE = new SqlDateType<>(Time.class, SqlType.TIME);
+    public static final SqlDateType<Timestamp> SQL_DATE_TIME_TYPE = new SqlDateType<>(Timestamp.class, SqlType.TIMESTAMP);
+
+    private final Class<T> typeClass;
+    private final SqlType sqlType;
+
+    private SqlDateType(Class<T> typeClass, SqlType sqlType) {
+        this.typeClass = typeClass;
+        this.sqlType = sqlType;
+    }
+
+    @Override
+    public Class<T> getTypeClass() {
+        return typeClass;
+    }
+
+    @Override
+    public SqlType getSqlType() {
+        return this.sqlType;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(typeClass);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == this) {
+            return true;
+        }
+        if (!(obj instanceof SqlDateType)) {
+            return false;
+        }
+        SqlDateType<?> that = (SqlDateType<?>) obj;
+        return Objects.equals(typeClass, that.typeClass);
+    }
+
+    @Override
+    public String toString() {
+        return sqlType.toString();
+    }
+}

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

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.gbase8a;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
+
+public class Gbase8aDialect implements JdbcDialect {
+    @Override
+    public String dialectName() {
+        return "Gbase8a";
+    }
+
+    @Override
+    public JdbcRowConverter getRowConverter() {
+        return new Gbase8aJdbcRowConverter();
+    }
+
+    @Override
+    public JdbcDialectTypeMapper getJdbcDialectTypeMapper() {
+        return new Gbase8aTypeMapper();
+    }
+}

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

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.gbase8a;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectFactory;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(JdbcDialectFactory.class)
+public class Gbase8aDialectFactory implements JdbcDialectFactory {
+    @Override
+    public boolean acceptsURL(String url) {
+        return url.startsWith("jdbc:gbase:");
+    }
+
+    @Override
+    public JdbcDialect create() {
+        return new Gbase8aDialect();
+    }
+}

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

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.gbase8a;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.api.table.type.SqlType;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class Gbase8aJdbcRowConverter extends AbstractJdbcRowConverter {
+    @Override
+    public String converterName() {
+        return "Gbase8a";
+    }
+
+    @SuppressWarnings("checkstyle:MagicNumber")
+    @Override
+    public SeaTunnelRow toInternal(ResultSet rs, ResultSetMetaData metaData, SeaTunnelRowType typeInfo) throws SQLException {
+        List<Object> fields = new ArrayList<>(typeInfo.getFieldTypes().length);
+        SeaTunnelDataType<?>[] seaTunnelDataTypes = typeInfo.getFieldTypes();
+
+        for (int i = 1; i <= seaTunnelDataTypes.length; i++) {
+            Object seatunnelField;
+            SqlType sqlType = seaTunnelDataTypes[i - 1].getSqlType();
+            if (null == rs.getObject(i)) {
+                seatunnelField = null;
+            } else if (SqlType.BOOLEAN.equals(sqlType)) {
+                seatunnelField = rs.getBoolean(i);
+            } else if (SqlType.TINYINT.equals(sqlType)) {
+                seatunnelField = rs.getByte(i);
+            } else if (SqlType.SMALLINT.equals(sqlType)) {
+                seatunnelField = rs.getShort(i);
+            } else if (SqlType.INT.equals(sqlType)) {
+                seatunnelField = rs.getInt(i);
+            } else if (SqlType.BIGINT.equals(sqlType)) {
+                seatunnelField = rs.getLong(i);
+            } else if (SqlType.DECIMAL.equals(sqlType)) {
+                Object value = rs.getObject(i);
+                seatunnelField = value instanceof BigInteger ?
+                    new BigDecimal((BigInteger) value, 0)
+                    : value;
+            } else if (SqlType.FLOAT.equals(sqlType)) {
+                seatunnelField = rs.getFloat(i);
+            } else if (SqlType.DOUBLE.equals(sqlType)) {
+                seatunnelField = rs.getDouble(i);
+            } else if (SqlType.STRING.equals(sqlType)) {
+                seatunnelField = rs.getString(i);
+            } else if (SqlType.TIME.equals(sqlType)) {
+                seatunnelField = rs.getTime(i);
+            } else if (SqlType.DATE.equals(sqlType)) {
+                seatunnelField = rs.getDate(i);
+            } else if (SqlType.TIMESTAMP.equals(sqlType)) {
+                seatunnelField = rs.getTimestamp(i);
+            } else if (SqlType.BYTES.equals(sqlType)) {
+                seatunnelField = rs.getBytes(i);
+            } else {
+                throw new IllegalStateException("Unexpected value: " + sqlType);
+            }
+
+            fields.add(seatunnelField);
+        }
+
+        return new SeaTunnelRow(fields.toArray());
+    }
+
+}

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

@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.gbase8a;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.DecimalType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SqlDateType;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+
+@Slf4j
+public class Gbase8aTypeMapper implements JdbcDialectTypeMapper {
+
+    //ref http://www.gbase.cn/down/4419.html
+    // ============================data types=====================
+    private static final String GBASE8A_UNKNOWN = "UNKNOWN";
+
+    // -------------------------number----------------------------
+    private static final String GBASE8A_INT = "INT";
+    private static final String GBASE8A_TINYINT = "TINYINT";
+    private static final String GBASE8A_SMALLINT = "SMALLINT";
+    private static final String GBASE8A_BIGINT = "BIGINT";
+    private static final String GBASE8A_DECIMAL = "DECIMAL";
+    private static final String GBASE8A_FLOAT = "FLOAT";
+    private static final String GBASE8A_DOUBLE = "DOUBLE";
+
+    // -------------------------string----------------------------
+    private static final String GBASE8A_CHAR = "CHAR";
+    private static final String GBASE8A_VARCHAR = "VARCHAR";
+
+
+    // ------------------------------time-------------------------
+    private static final String GBASE8A_DATE = "DATE";
+    private static final String GBASE8A_TIME = "TIME";
+    private static final String GBASE8A_TIMESTAMP = "TIMESTAMP";
+    private static final String GBASE8A_DATETIME = "DATETIME";
+
+    // ------------------------------blob-------------------------
+    private static final String GBASE8A_BLOB = "BLOB";
+    private static final String GBASE8A_TEXT = "TEXT";
+
+    @SuppressWarnings("checkstyle:MagicNumber")
+    @Override
+    public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex) throws SQLException {
+        String gbase8aType = metadata.getColumnTypeName(colIndex).toUpperCase();
+        int precision = metadata.getPrecision(colIndex);
+        int scale = metadata.getScale(colIndex);
+        switch (gbase8aType) {
+            case GBASE8A_TINYINT:
+                return BasicType.BYTE_TYPE;
+            case GBASE8A_SMALLINT:
+                return BasicType.SHORT_TYPE;
+            case GBASE8A_INT:
+                return BasicType.INT_TYPE;
+            case GBASE8A_BIGINT:
+                return BasicType.LONG_TYPE;
+            case GBASE8A_DECIMAL:
+                if (precision < 38) {
+                    return new DecimalType(precision, scale);
+                }
+                return new DecimalType(38, 18);
+            case GBASE8A_DOUBLE:
+                return BasicType.DOUBLE_TYPE;
+            case GBASE8A_FLOAT:
+                return BasicType.FLOAT_TYPE;
+            case GBASE8A_CHAR:
+            case GBASE8A_VARCHAR:
+                return BasicType.STRING_TYPE;
+            case GBASE8A_DATE:
+                return SqlDateType.SQL_DATE_TYPE;
+            case GBASE8A_TIME:
+                return SqlDateType.SQL_TIME_TYPE;
+            case GBASE8A_TIMESTAMP:
+            case GBASE8A_DATETIME:
+                return SqlDateType.SQL_DATE_TIME_TYPE;
+            case GBASE8A_BLOB:
+            case GBASE8A_TEXT:
+                return PrimitiveByteArrayType.INSTANCE;
+            //Doesn't support yet
+            case GBASE8A_UNKNOWN:
+            default:
+                final String jdbcColumnName = metadata.getColumnName(colIndex);
+                throw new UnsupportedOperationException(
+                    String.format(
+                        "Doesn't support GBASE8A type '%s' on column '%s'  yet.",
+                        gbase8aType, jdbcColumnName));
+        }
+    }
+}

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

@@ -75,12 +75,11 @@ public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex) th
         switch (oracleType) {
             case ORACLE_INTEGER:
                 return BasicType.INT_TYPE;
+            case ORACLE_FLOAT:
             case ORACLE_NUMBER:
-                if (precision < 38) {
-                    return new DecimalType(precision, scale);
-                }
+                //The float type will be converted to DecimalType(10, -127),
+                // which will lose precision in the spark engine
                 return new DecimalType(38, 18);
-            case ORACLE_FLOAT:
             case ORACLE_BINARY_DOUBLE:
                 return BasicType.DOUBLE_TYPE;
             case ORACLE_BINARY_FLOAT: