[hotfix]oracle type modify on sql & adjust to get unique key method (#1262)

Author: yanghuaiGit

chunjun-connectors/chunjun-connector-jdbc-base/src/main/java/com/dtstack/chunjun/connector/jdbc/sink/JdbcOutputFormat.java

@@ -98,7 +98,7 @@ protected void openInternal(int taskNumber, int numTasks) {
                 List<String> updateKey = jdbcConf.getUniqueKey();
                 if (CollectionUtils.isEmpty(updateKey)) {
                     List<String> tableIndex =
-                            JdbcUtil.getTableIndex(
+                            JdbcUtil.getTableUniqueIndex(
                                     jdbcConf.getSchema(), jdbcConf.getTable(), dbConn);
                     jdbcConf.setUniqueKey(tableIndex);
                     LOG.info("updateKey = {}", JsonUtil.toJson(tableIndex));

chunjun-connectors/chunjun-connector-jdbc-base/src/main/java/com/dtstack/chunjun/connector/jdbc/util/JdbcUtil.java

@@ -32,6 +32,7 @@
 
 import org.apache.flink.table.types.logical.LogicalType;
 
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
@@ -177,6 +178,22 @@ public static List<String> getTableIndex(String schema, String tableName, Connec
         return indexList;
     }
 
+    public static List<String> getTableUniqueIndex(
+            String schema, String tableName, Connection dbConn) throws SQLException {
+        List<String> tablePrimaryKey = getTablePrimaryKey(schema, tableName, dbConn);
+        if (CollectionUtils.isNotEmpty(tablePrimaryKey)) {
+            return tablePrimaryKey;
+        }
+
+        ResultSet rs = dbConn.getMetaData().getIndexInfo(null, schema, tableName, true, false);
+        List<String> indexList = new LinkedList<>();
+        while (rs.next()) {
+            String index = rs.getString(9);
+            if (StringUtils.isNotBlank(index)) indexList.add(index);
+        }
+        return indexList;
+    }
+
     /**
      * get primarykey
      *

chunjun-connectors/chunjun-connector-oracle/src/main/java/com/dtstack/chunjun/connector/oracle/converter/OracleRowConverter.java

@@ -152,15 +152,17 @@ protected IDeserializationConverter createInternalConverter(LogicalType type) {
                     }
                 };
             case CHAR:
-                if (((CharType) type).getLength() > CLOB_LENGTH) {
+                if (((CharType) type).getLength() > CLOB_LENGTH
+                        && ((VarCharType) type).getLength() != Integer.MAX_VALUE) {
                     return val -> {
                         oracle.sql.CLOB clob = (oracle.sql.CLOB) val;
                         return StringData.fromString(ConvertUtil.convertClob(clob));
                     };
                 }
                 return val -> StringData.fromString(val.toString());
             case VARCHAR:
-                if (((VarCharType) type).getLength() > CLOB_LENGTH) {
+                if (((VarCharType) type).getLength() > CLOB_LENGTH
+                        && ((VarCharType) type).getLength() != Integer.MAX_VALUE) {
                     return val -> {
                         oracle.sql.CLOB clob = (oracle.sql.CLOB) val;
                         return StringData.fromString(ConvertUtil.convertClob(clob));
@@ -200,7 +202,8 @@ protected ISerializationConverter<FieldNamedPreparedStatement> createExternalCon
             case DOUBLE:
                 return (val, index, statement) -> statement.setDouble(index, val.getDouble(index));
             case CHAR:
-                if (((CharType) type).getLength() > CLOB_LENGTH) {
+                if (((CharType) type).getLength() > CLOB_LENGTH
+                        && ((VarCharType) type).getLength() != Integer.MAX_VALUE) {
                     return (val, index, statement) -> {
                         try (StringReader reader =
                                 new StringReader(val.getString(index).toString())) {
@@ -213,7 +216,8 @@ protected ISerializationConverter<FieldNamedPreparedStatement> createExternalCon
                 };
             case VARCHAR:
                 // value is BinaryString
-                if (((VarCharType) type).getLength() > CLOB_LENGTH) {
+                if (((VarCharType) type).getLength() > CLOB_LENGTH
+                        && ((VarCharType) type).getLength() != Integer.MAX_VALUE) {
                     return (val, index, statement) -> {
                         try (StringReader reader =
                                 new StringReader(val.getString(index).toString())) {

chunjun-connectors/chunjun-connector-oraclelogminer/src/main/java/com/dtstack/chunjun/connector/oraclelogminer/converter/OracleRawTypeConverter.java

@@ -52,14 +52,15 @@ public static DataType apply(String type) {
             case "NCHAR":
             case "NVARCHAR2":
             case "LONG":
-            case "RAW":
-            case "LONG RAW":
             case "BLOB":
             case "CLOB":
             case "NCLOB":
             case "INTERVAL YEAR":
             case "INTERVAL DAY":
                 return DataTypes.STRING();
+            case "RAW":
+            case "LONG RAW":
+                return DataTypes.BYTES();
             case "INT":
             case "INTEGER":
                 return DataTypes.INT();