[incubator-seatunnel] branch dev updated: [JDBC] [ORACLE] Improve Oracle Type to SeaTunnel Type Mapping (#3486)

[wuchunfu]

This is an automated email from the ASF dual-hosted git repository.

wuchunfu pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new 8fe0dda6e [JDBC] [ORACLE] Improve Oracle Type to SeaTunnel Type 
Mapping (#3486)
8fe0dda6e is described below

commit 8fe0dda6e20d527fb20a645d69aa35a1342ff8cd
Author: Hisoka <fanjiaemi...@qq.com>
AuthorDate: Tue Nov 22 13:34:17 2022 +0800

    [JDBC] [ORACLE] Improve Oracle Type to SeaTunnel Type Mapping (#3486)
    
    * [JDBC] [ORACLE] Improve Oracle Number to SeaTunnel Type
    
    * [JDBC] [ORACLE] Improve Oracle Number to SeaTunnel Type
    
    * [JDBC] [ORACLE] Improve Oracle Number to SeaTunnel Type
    
    * [Core] [Plugin] Fix Ci problem
---
 .../jdbc/internal/dialect/oracle/OracleDialect.java    | 18 ++++++++++++++++++
 .../jdbc/internal/dialect/oracle/OracleTypeMapper.java | 12 +++++++++++-
 2 files changed, 29 insertions(+), 1 deletion(-)

diff --git 
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java
 
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java
index 96d86115b..f3fe59ab6 100644
--- 
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java
+++ 
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java
@@ -21,12 +21,19 @@ import 
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRow
 import 
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
 import 
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
 
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
 public class OracleDialect implements JdbcDialect {
+
+    private static final int DEFAULT_ORACLE_FETCH_SIZE = 128;
+
     @Override
     public String dialectName() {
         return "Oracle";
@@ -84,4 +91,15 @@ public class OracleDialect implements JdbcDialect {
 
         return Optional.of(upsertSQL);
     }
+
+    @Override
+    public PreparedStatement creatPreparedStatement(Connection connection, 
String queryTemplate, int fetchSize) throws SQLException {
+        PreparedStatement statement = 
connection.prepareStatement(queryTemplate, ResultSet.TYPE_FORWARD_ONLY, 
ResultSet.CONCUR_READ_ONLY);
+        if (fetchSize > 0) {
+            statement.setFetchSize(fetchSize);
+        } else {
+            statement.setFetchSize(DEFAULT_ORACLE_FETCH_SIZE);
+        }
+        return statement;
+    }
 }
diff --git 
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleTypeMapper.java
 
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleTypeMapper.java
index eb1f9097f..aa739a3b7 100644
--- 
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleTypeMapper.java
+++ 
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleTypeMapper.java
@@ -76,10 +76,19 @@ public class OracleTypeMapper implements 
JdbcDialectTypeMapper {
             case ORACLE_INTEGER:
                 return BasicType.INT_TYPE;
             case ORACLE_FLOAT:
-            case ORACLE_NUMBER:
                 //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_NUMBER:
+                if (scale == 0) {
+                    if (precision <= 9) {
+                        return BasicType.INT_TYPE;
+                    }
+                    if (precision <= 18) {
+                        return BasicType.LONG_TYPE;
+                    }
+                }
+                return new DecimalType(38, 18);
             case ORACLE_BINARY_DOUBLE:
                 return BasicType.DOUBLE_TYPE;
             case ORACLE_BINARY_FLOAT:
@@ -95,6 +104,7 @@ public class OracleTypeMapper implements 
JdbcDialectTypeMapper {
             case ORACLE_CLOB:
                 return BasicType.STRING_TYPE;
             case ORACLE_DATE:
+                return LocalTimeType.LOCAL_DATE_TYPE;
             case ORACLE_TIMESTAMP:
             case ORACLE_TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                 return LocalTimeType.LOCAL_DATE_TIME_TYPE;