[Fix-803] [Client] Fix TypeConvert-ColumnType Enumeration usage error (#878)

Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>

dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -212,12 +212,12 @@ public abstract class AbstractSinkBuilder {
     protected void buildColumn(List<String> columnNameList, List<LogicalType> columnTypeList, List<Column> columns) {
         for (Column column : columns) {
             columnNameList.add(column.getName());
-            columnTypeList.add(getLogicalType(column.getJavaType()));
+            columnTypeList.add(getLogicalType(column));
         }
     }
 
-    public LogicalType getLogicalType(ColumnType columnType) {
-        switch (columnType) {
+    public LogicalType getLogicalType(Column column) {
+        switch (column.getJavaType()) {
             case STRING:
                 return new VarCharType();
             case BOOLEAN:
@@ -239,10 +239,10 @@ public abstract class AbstractSinkBuilder {
             case JAVA_LANG_DOUBLE:
                 return new DoubleType();
             case DECIMAL:
-                if (columnType.getPrecision() == null || columnType.getPrecision() == 0) {
-                    return new DecimalType(38, columnType.getScale());
+                if (column.getPrecision() == null || column.getPrecision() == 0) {
+                    return new DecimalType(38, column.getScale());
                 } else {
-                    return new DecimalType(columnType.getPrecision(), columnType.getScale());
+                    return new DecimalType(column.getPrecision(), column.getScale());
                 }
             case INT:
             case INTEGER:

dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -212,12 +212,12 @@ public abstract class AbstractSinkBuilder {
     protected void buildColumn(List<String> columnNameList, List<LogicalType> columnTypeList, List<Column> columns) {
         for (Column column : columns) {
             columnNameList.add(column.getName());
-            columnTypeList.add(getLogicalType(column.getJavaType()));
+            columnTypeList.add(getLogicalType(column));
         }
     }
 
-    public LogicalType getLogicalType(ColumnType columnType) {
-        switch (columnType) {
+    public LogicalType getLogicalType(Column column) {
+        switch (column.getJavaType()) {
             case STRING:
                 return new VarCharType();
             case BOOLEAN:
@@ -239,10 +239,10 @@ public abstract class AbstractSinkBuilder {
             case JAVA_LANG_DOUBLE:
                 return new DoubleType();
             case DECIMAL:
-                if (columnType.getPrecision() == null || columnType.getPrecision() == 0) {
-                    return new DecimalType(38, columnType.getScale());
+                if (column.getPrecision() == null || column.getPrecision() == 0) {
+                    return new DecimalType(38, column.getScale());
                 } else {
-                    return new DecimalType(columnType.getPrecision(), columnType.getScale());
+                    return new DecimalType(column.getPrecision(), column.getScale());
                 }
             case INT:
             case INTEGER:

dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -212,12 +212,12 @@ public abstract class AbstractSinkBuilder {
     protected void buildColumn(List<String> columnNameList, List<LogicalType> columnTypeList, List<Column> columns) {
         for (Column column : columns) {
             columnNameList.add(column.getName());
-            columnTypeList.add(getLogicalType(column.getJavaType()));
+            columnTypeList.add(getLogicalType(column));
         }
     }
 
-    public LogicalType getLogicalType(ColumnType columnType) {
-        switch (columnType) {
+    public LogicalType getLogicalType(Column column) {
+        switch (column.getJavaType()) {
             case STRING:
                 return new VarCharType();
             case BOOLEAN:
@@ -239,10 +239,10 @@ public abstract class AbstractSinkBuilder {
             case JAVA_LANG_DOUBLE:
                 return new DoubleType();
             case DECIMAL:
-                if (columnType.getPrecision() == null || columnType.getPrecision() == 0) {
-                    return new DecimalType(38, columnType.getScale());
+                if (column.getPrecision() == null || column.getPrecision() == 0) {
+                    return new DecimalType(38, column.getScale());
                 } else {
-                    return new DecimalType(columnType.getPrecision(), columnType.getScale());
+                    return new DecimalType(column.getPrecision(), column.getScale());
                 }
             case INT:
             case INTEGER:

dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -212,12 +212,12 @@ public abstract class AbstractSinkBuilder {
     protected void buildColumn(List<String> columnNameList, List<LogicalType> columnTypeList, List<Column> columns) {
         for (Column column : columns) {
             columnNameList.add(column.getName());
-            columnTypeList.add(getLogicalType(column.getJavaType()));
+            columnTypeList.add(getLogicalType(column));
         }
     }
 
-    public LogicalType getLogicalType(ColumnType columnType) {
-        switch (columnType) {
+    public LogicalType getLogicalType(Column column) {
+        switch (column.getJavaType()) {
             case STRING:
                 return new VarCharType();
             case BOOLEAN:
@@ -239,10 +239,10 @@ public abstract class AbstractSinkBuilder {
             case JAVA_LANG_DOUBLE:
                 return new DoubleType();
             case DECIMAL:
-                if (columnType.getPrecision() == null || columnType.getPrecision() == 0) {
-                    return new DecimalType(38, columnType.getScale());
+                if (column.getPrecision() == null || column.getPrecision() == 0) {
+                    return new DecimalType(38, column.getScale());
                 } else {
-                    return new DecimalType(columnType.getPrecision(), columnType.getScale());
+                    return new DecimalType(column.getPrecision(), column.getScale());
                 }
             case INT:
             case INTEGER:

dlink-client/dlink-client-1.15/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -212,12 +212,12 @@ public abstract class AbstractSinkBuilder {
     protected void buildColumn(List<String> columnNameList, List<LogicalType> columnTypeList, List<Column> columns) {
         for (Column column : columns) {
             columnNameList.add(column.getName());
-            columnTypeList.add(getLogicalType(column.getJavaType()));
+            columnTypeList.add(getLogicalType(column));
         }
     }
 
-    public LogicalType getLogicalType(ColumnType columnType) {
-        switch (columnType) {
+    public LogicalType getLogicalType(Column column) {
+        switch (column.getJavaType()) {
             case STRING:
                 return new VarCharType();
             case BOOLEAN:
@@ -239,10 +239,10 @@ public abstract class AbstractSinkBuilder {
             case JAVA_LANG_DOUBLE:
                 return new DoubleType();
             case DECIMAL:
-                if (columnType.getPrecision() == null || columnType.getPrecision() == 0) {
-                    return new DecimalType(38, columnType.getScale());
+                if (column.getPrecision() == null || column.getPrecision() == 0) {
+                    return new DecimalType(38, column.getScale());
                 } else {
-                    return new DecimalType(columnType.getPrecision(), columnType.getScale());
+                    return new DecimalType(column.getPrecision(), column.getScale());
                 }
             case INT:
             case INTEGER:

dlink-client/dlink-client-base/src/main/java/com/dlink/utils/FlinkBaseUtil.java

@@ -75,7 +75,7 @@ public class FlinkBaseUtil {
         sb.append("` (\n");
         List<String> pks = new ArrayList<>();
         for (int i = 0; i < table.getColumns().size(); i++) {
-            String type = table.getColumns().get(i).getJavaType().getFlinkType();
+            String type = table.getColumns().get(i).getFlinkType();
             sb.append("    ");
             if (i > 0) {
                 sb.append(",");

dlink-common/src/main/java/com/dlink/model/Column.java

@@ -52,4 +52,16 @@ public class Column implements Serializable {
     private String characterSet;
     private String collation;
 
+    public String getFlinkType() {
+        String flinkType = javaType.getFlinkType();
+        if (flinkType.equals("DECIMAL")) {
+            if (precision == null || precision == 0) {
+                return flinkType + "(" + 38 + "," + scale + ")";
+            } else {
+                return flinkType + "(" + precision + "," + scale + ")";
+            }
+        } else {
+            return flinkType;
+        }
+    }
 }

dlink-common/src/main/java/com/dlink/model/ColumnType.java

@@ -60,49 +60,17 @@ public enum ColumnType {
 
     private String javaType;
     private String flinkType;
-    private Integer precision;
-    private Integer scale;
 
     ColumnType(String javaType, String flinkType) {
         this.javaType = javaType;
         this.flinkType = flinkType;
     }
 
-    public ColumnType setPrecisionAndScale(Integer precision, Integer scale) {
-        this.precision = precision;
-        this.scale = scale;
-        return this;
-    }
-
     public String getJavaType() {
         return javaType;
     }
 
     public String getFlinkType() {
-        if (flinkType.equals("DECIMAL")) {
-            if (precision == null || precision == 0) {
-                return flinkType + "(" + 38 + "," + scale + ")";
-            } else {
-                return flinkType + "(" + precision + "," + scale + ")";
-            }
-        } else {
-            return flinkType;
-        }
-    }
-
-    public Integer getPrecision() {
-        return precision;
-    }
-
-    public void setPrecision(Integer precision) {
-        this.precision = precision;
-    }
-
-    public Integer getScale() {
-        return scale;
-    }
-
-    public void setScala(Integer scale) {
-        this.scale = scale;
+        return flinkType;
     }
 }

dlink-common/src/main/java/com/dlink/model/Table.java

@@ -107,7 +107,7 @@ public class Table implements Serializable, Comparable<Table> {
         sb.append("CREATE TABLE IF NOT EXISTS " + tableName + " (\n");
         List<String> pks = new ArrayList<>();
         for (int i = 0; i < columns.size(); i++) {
-            String type = columns.get(i).getJavaType().getFlinkType();
+            String type = columns.get(i).getFlinkType();
             sb.append("    ");
             if (i > 0) {
                 sb.append(",");
@@ -158,7 +158,7 @@ public class Table implements Serializable, Comparable<Table> {
         sb.append("CREATE TABLE IF NOT EXISTS " + name + " (\n");
         List<String> pks = new ArrayList<>();
         for (int i = 0; i < columns.size(); i++) {
-            String type = columns.get(i).getJavaType().getFlinkType();
+            String type = columns.get(i).getFlinkType();
             sb.append("    ");
             if (i > 0) {
                 sb.append(",");

dlink-metadata/dlink-metadata-clickhouse/src/main/java/com/dlink/metadata/convert/ClickHouseTypeConvert.java

@@ -92,7 +92,6 @@ public class ClickHouseTypeConvert implements ITypeConvert {
                 columnType = ColumnType.INT;
             }
         }
-        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
     }
 

dlink-metadata/dlink-metadata-doris/src/main/java/com/dlink/metadata/convert/DorisTypeConvert.java

@@ -92,7 +92,6 @@ public class DorisTypeConvert implements ITypeConvert {
                 columnType = ColumnType.DOUBLE;
             }
         }
-        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
     }
 

dlink-metadata/dlink-metadata-hive/src/main/java/com/dlink/metadata/convert/HiveTypeConvert.java

@@ -94,7 +94,6 @@ public class HiveTypeConvert implements ITypeConvert {
                 columnType = ColumnType.DOUBLE;
             }
         }
-        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
     }
 

dlink-metadata/dlink-metadata-mysql/src/main/java/com/dlink/metadata/convert/MySqlTypeConvert.java

@@ -84,7 +84,6 @@ public class MySqlTypeConvert implements ITypeConvert {
                 columnType = ColumnType.INT;
             }
         }
-        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
     }
 

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

@@ -72,7 +72,6 @@ public class OracleTypeConvert implements ITypeConvert {
         } else if (t.contains("blob")) {
             columnType = ColumnType.BYTES;
         }
-        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
     }
 

dlink-metadata/dlink-metadata-phoenix/src/main/java/com/dlink/metadata/convert/PhoenixTypeConvert.java

@@ -79,7 +79,6 @@ public class PhoenixTypeConvert implements ITypeConvert {
         } else if (t.contains("date")) {
             columnType = ColumnType.DATE;
         }
-        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
     }
 

dlink-metadata/dlink-metadata-postgresql/src/main/java/com/dlink/metadata/convert/PostgreSqlTypeConvert.java

@@ -90,7 +90,6 @@ public class PostgreSqlTypeConvert implements ITypeConvert {
         } else if (t.contains("array")) {
             columnType = ColumnType.T;
         }
-        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
     }
 

dlink-metadata/dlink-metadata-sqlserver/src/main/java/com/dlink/metadata/convert/SqlServerTypeConvert.java

@@ -80,7 +80,6 @@ public class SqlServerTypeConvert implements ITypeConvert {
         } else if (t.contains("timestamp") || t.contains("binary") || t.contains("varbinary") || t.contains("image")) {
             columnType = ColumnType.BYTES;
         }
-        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
     }