[Fix-522] [client] Fix CDCSOURCE OracleCDC number can't be cast to Long

dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -15,6 +15,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.operations.ModifyOperation;
 import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.types.logical.BigIntType;
 import org.apache.flink.table.types.logical.DateType;
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -269,11 +270,25 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
             return null;
         }
         if (logicalType instanceof DateType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDate();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDate();
+            }
         } else if (logicalType instanceof TimestampType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }
         } else if (logicalType instanceof DecimalType) {
             return new BigDecimal((String) value);
+        } else if (logicalType instanceof BigIntType) {
+            if(value instanceof Integer){
+                return ((Integer) value).longValue();
+            }else {
+                return value;
+            }
         } else {
             return value;
         }

dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -15,6 +15,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.operations.ModifyOperation;
 import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.types.logical.BigIntType;
 import org.apache.flink.table.types.logical.DateType;
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -269,11 +270,25 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
             return null;
         }
         if (logicalType instanceof DateType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDate();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDate();
+            }
         } else if (logicalType instanceof TimestampType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }
         } else if (logicalType instanceof DecimalType) {
             return new BigDecimal((String) value);
+        } else if (logicalType instanceof BigIntType) {
+            if(value instanceof Integer){
+                return ((Integer) value).longValue();
+            }else {
+                return value;
+            }
         } else {
             return value;
         }

dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -263,11 +263,25 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
             return null;
         }
         if (logicalType instanceof DateType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDate();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDate();
+            }
         } else if (logicalType instanceof TimestampType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }
         } else if (logicalType instanceof DecimalType) {
             return new BigDecimal((String) value);
+        } else if (logicalType instanceof BigIntType) {
+            if(value instanceof Integer){
+                return ((Integer) value).longValue();
+            }else {
+                return value;
+            }
         } else {
             return value;
         }

dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -15,6 +15,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.operations.ModifyOperation;
 import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.types.logical.BigIntType;
 import org.apache.flink.table.types.logical.DateType;
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -269,11 +270,25 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
             return null;
         }
         if (logicalType instanceof DateType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDate();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDate();
+            }
         } else if (logicalType instanceof TimestampType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }
         } else if (logicalType instanceof DecimalType) {
             return new BigDecimal((String) value);
+        } else if (logicalType instanceof BigIntType) {
+            if(value instanceof Integer){
+                return ((Integer) value).longValue();
+            }else {
+                return value;
+            }
         } else {
             return value;
         }

dlink-client/dlink-client-1.15/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -15,6 +15,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.operations.ModifyOperation;
 import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.types.logical.BigIntType;
 import org.apache.flink.table.types.logical.DateType;
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -269,11 +270,25 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
             return null;
         }
         if (logicalType instanceof DateType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDate();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDate();
+            }
         } else if (logicalType instanceof TimestampType) {
+            if(value instanceof Integer){
+                return Instant.ofEpochMilli(((Integer) value).longValue()).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }else {
                 return Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDateTime();
+            }
         } else if (logicalType instanceof DecimalType) {
             return new BigDecimal((String) value);
+        } else if (logicalType instanceof BigIntType) {
+            if(value instanceof Integer){
+                return ((Integer) value).longValue();
+            }else {
+                return value;
+            }
         } else {
             return value;
         }

dlink-metadata/dlink-metadata-oracle/src/main/java/com/dlink/metadata/convert/OracleTypeConvert.java

@@ -28,7 +28,7 @@ public class OracleTypeConvert implements ITypeConvert {
             if (t.matches("number\\(+\\d\\)")) {
                 columnType = ColumnType.INTEGER;
             } else if (t.matches("number\\(+\\d{2}+\\)")) {
-                columnType = ColumnType.LONG;
+                columnType = ColumnType.JAVA_LANG_LONG;
             } else {
                 columnType = ColumnType.DECIMAL;
             }