[Fix-3478][CDCSOURCE] Fix Oracle CDC could not deal with the date and…

… number type properly (#3562) Co-authored-by: wenmo <[email protected]>
DataLinkDC · Jun 5, 2024 · 34eacc2 · 34eacc2
1 parent 8b9bfd1
commit 34eacc2
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 2 deletions.
diff --git a/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/AbstractSinkBuilder.java b/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/AbstractSinkBuilder.java
@@ -404,7 +404,7 @@ protected Optional<Object> convertBigIntType(Object value, LogicalType logicalTy
                 return Optional.of(((Integer) value).longValue());
             }
 
-            return Optional.of(value);
+            return Optional.of(Long.parseLong(value.toString()));
         }
         return Optional.empty();
     }
@@ -484,7 +484,7 @@ protected Optional<Object> convertTimeType(Object target, LogicalType logicalTyp
 
     protected Optional<Object> convertVarCharType(Object target, LogicalType logicalType) {
         if (logicalType instanceof VarCharType) {
-            return Optional.of(StringData.fromString((String) target));
+            return Optional.of((String) target);
         }
         return Optional.empty();
     }

diff --git a/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/SQLSinkBuilder.java b/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/SQLSinkBuilder.java
@@ -58,6 +58,7 @@ private SQLSinkBuilder(FlinkCDCConfig config) {
     @Override
     protected void initTypeConverterList() {
         typeConverterList = Arrays.asList(
+                this::convertVarCharType,
                 this::convertDateType,
                 this::convertTimestampType,
                 this::convertFloatType,

diff --git a/...ata/dinky-metadata-base/src/main/java/org/dinky/metadata/convert/AbstractTypeConvert.java b/...ata/dinky-metadata-base/src/main/java/org/dinky/metadata/convert/AbstractTypeConvert.java
@@ -48,6 +48,7 @@ public ColumnType convert(Column column, DriverConfig<T> driverConfig) {
         if (Asserts.isNull(column)) {
             return ColumnType.STRING;
         }
+        column.setType(column.getType().toLowerCase());
         for (Map.Entry<String, BiFunction<Column, DriverConfig<T>, Optional<ColumnType>>> entry :
                 convertMap.entrySet()) {
             if (column.getType().contains(entry.getKey())) {