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

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

[Fix-522] [client] Fix CDCSOURCE OracleCDC number can't be cast to Long
55e50632 · aiwenmo · GitHub · 63ed66e8 · 45ea7bdf · 55e50632
Unverified Commit 55e50632 authored May 19, 2022 by aiwenmo Committed by GitHub May 19, 2022
6 changed files
--- a/dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java
+++ b/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;
        }

--- a/dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java
+++ b/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;
        }

--- a/dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java
+++ b/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;
        }

--- a/dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java
+++ b/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;
        }

--- a/dlink-client/dlink-client-1.15/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java
+++ b/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;
        }

--- a/dlink-metadata/dlink-metadata-oracle/src/main/java/com/dlink/metadata/convert/OracleTypeConvert.java
+++ b/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;
            }