[Feature-609][metadata] Supports generating FlinkSQL with null and not null through metadata

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/driver/Driver.java

@@ -42,7 +42,7 @@ public interface Driver {
         }
         Optional<Driver> optionalDriver = Driver.get(config);
         if (!optionalDriver.isPresent()) {
-            throw new MetaDataException("不支持数据源类型【" + config.getType() + "】，请在 lib 下添加扩展依赖");
+            throw new MetaDataException("缺少数据源类型【" + config.getType() + "】的依赖，请在 lib 下添加对应的扩展依赖");
         }
         Driver driver = optionalDriver.get().connect();
         DriverPool.push(key, driver);

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

@@ -15,38 +15,65 @@ public class ClickHouseTypeConvert implements ITypeConvert {
     // Use mysql now,and welcome to fix it.
     @Override
     public ColumnType convert(Column column) {
+        ColumnType columnType = ColumnType.STRING;
         if (Asserts.isNull(column)) {
-            return ColumnType.STRING;
+            return columnType;
         }
         String t = column.getType().toLowerCase();
+        boolean isNullable = !column.isKeyFlag() && column.isNullable();
         if (t.contains("tinyint")) {
-            return ColumnType.BYTE;
+            columnType = ColumnType.BYTE;
         } else if (t.contains("smallint") || t.contains("tinyint unsigned")) {
-            return ColumnType.SHORT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_SHORT;
+            } else {
+                columnType = ColumnType.SHORT;
+            }
         } else if (t.contains("bigint unsigned") || t.contains("numeric") || t.contains("decimal")) {
-            return ColumnType.DECIMAL;
+            columnType = ColumnType.DECIMAL;
         } else if (t.contains("bigint") || t.contains("int unsigned")) {
-            return ColumnType.LONG;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_LONG;
+            } else {
+                columnType = ColumnType.LONG;
+            }
         } else if (t.contains("float")) {
-            return ColumnType.FLOAT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_FLOAT;
+            } else {
+                columnType = ColumnType.FLOAT;
+            }
         } else if (t.contains("double")) {
-            return ColumnType.DOUBLE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
+            }
         } else if (t.contains("boolean") || t.contains("tinyint(1)")) {
-            return ColumnType.BOOLEAN;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BOOLEAN;
+            } else {
+                columnType = ColumnType.BOOLEAN;
+            }
         } else if (t.contains("datetime")) {
-            return ColumnType.TIMESTAMP;
+            columnType = ColumnType.TIMESTAMP;
         } else if (t.contains("date")) {
-            return ColumnType.DATE;
+            columnType = ColumnType.DATE;
         } else if (t.contains("time")) {
-            return ColumnType.TIME;
+            columnType = ColumnType.TIME;
         } else if (t.contains("char") || t.contains("text")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("binary") || t.contains("blob")) {
-            return ColumnType.BYTES;
+            columnType = ColumnType.BYTES;
         } else if (t.contains("int") || t.contains("mediumint") || t.contains("smallint unsigned")) {
-            return ColumnType.INTEGER;
+            if (isNullable) {
+                columnType = ColumnType.INTEGER;
+            } else {
+                columnType = ColumnType.INT;
+            }
         }
-        return ColumnType.STRING;
+        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
+        return columnType;
     }
 
     @Override
@@ -57,16 +84,21 @@ public class ClickHouseTypeConvert implements ITypeConvert {
             case BYTE:
                 return "tinyint";
             case SHORT:
+            case JAVA_LANG_SHORT:
                 return "smallint";
             case DECIMAL:
                 return "decimal";
             case LONG:
+            case JAVA_LANG_LONG:
                 return "bigint";
             case FLOAT:
+            case JAVA_LANG_FLOAT:
                 return "float";
             case DOUBLE:
+            case JAVA_LANG_DOUBLE:
                 return "double";
             case BOOLEAN:
+            case JAVA_LANG_BOOLEAN:
                 return "boolean";
             case TIMESTAMP:
                 return "datetime";
@@ -77,6 +109,7 @@ public class ClickHouseTypeConvert implements ITypeConvert {
             case BYTES:
                 return "binary";
             case INTEGER:
+            case INT:
                 return "int";
             default:
                 return "varchar";

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

@@ -7,59 +7,73 @@ import com.dlink.model.ColumnType;
 public class DorisTypeConvert implements ITypeConvert {
     @Override
     public ColumnType convert(Column column) {
+        ColumnType columnType = ColumnType.STRING;
         if (Asserts.isNull(column)) {
-            return ColumnType.STRING;
+            return columnType;
         }
         String t = column.getType().toLowerCase();
+        boolean isNullable = !column.isKeyFlag() && column.isNullable();
         if (t.contains("char")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("boolean")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_BOOLEAN;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BOOLEAN;
+            } else {
+                columnType = ColumnType.BOOLEAN;
             }
-            return ColumnType.BOOLEAN;
         } else if (t.contains("tinyint")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_BYTE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BYTE;
+            } else {
+                columnType = ColumnType.BYTE;
             }
-            return ColumnType.BYTE;
         } else if (t.contains("smallint")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_SHORT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_SHORT;
+            } else {
+                columnType = ColumnType.SHORT;
             }
-            return ColumnType.SHORT;
         } else if (t.contains("bigint")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_LONG;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_LONG;
+            } else {
+                columnType = ColumnType.LONG;
             }
-            return ColumnType.LONG;
         } else if (t.contains("largeint")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("int")) {
-            if (column.isNullable()) {
-                return ColumnType.INTEGER;
+            if (isNullable) {
+                columnType = ColumnType.INTEGER;
+            } else {
+                columnType = ColumnType.INT;
             }
-            return ColumnType.INT;
         } else if (t.contains("float")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_FLOAT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_FLOAT;
+            } else {
+                columnType = ColumnType.FLOAT;
             }
-            return ColumnType.FLOAT;
         } else if (t.contains("double")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_DOUBLE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
             }
-            return ColumnType.DOUBLE;
         } else if (t.contains("date")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("datetime")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("decimal")) {
-            return ColumnType.DECIMAL;
+            columnType = ColumnType.DECIMAL;
         } else if (t.contains("time")) {
-            return ColumnType.DOUBLE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
+            }
         }
-        return ColumnType.STRING;
+        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
+        return columnType;
     }
 
     @Override

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

@@ -7,61 +7,75 @@ import com.dlink.model.ColumnType;
 public class HiveTypeConvert implements ITypeConvert {
     @Override
     public ColumnType convert(Column column) {
+        ColumnType columnType = ColumnType.STRING;
         if (Asserts.isNull(column)) {
-            return ColumnType.STRING;
+            return columnType;
         }
         String t = column.getType().toLowerCase().trim();
+        boolean isNullable = !column.isKeyFlag() && column.isNullable();
         if (t.contains("char")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("boolean")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_BOOLEAN;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BOOLEAN;
+            } else {
+                columnType = ColumnType.BOOLEAN;
             }
-            return ColumnType.BOOLEAN;
         } else if (t.contains("tinyint")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_BYTE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BYTE;
+            } else {
+                columnType = ColumnType.BYTE;
             }
-            return ColumnType.BYTE;
         } else if (t.contains("smallint")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_SHORT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_SHORT;
+            } else {
+                columnType = ColumnType.SHORT;
             }
-            return ColumnType.SHORT;
         } else if (t.contains("bigint")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_LONG;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_LONG;
+            } else {
+                columnType = ColumnType.LONG;
             }
-            return ColumnType.LONG;
         } else if (t.contains("largeint")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("int")) {
-            if (column.isNullable()) {
-                return ColumnType.INTEGER;
+            if (isNullable) {
+                columnType = ColumnType.INTEGER;
+            } else {
+                columnType = ColumnType.INT;
             }
-            return ColumnType.INT;
         } else if (t.contains("float")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_FLOAT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_FLOAT;
+            } else {
+                columnType = ColumnType.FLOAT;
             }
-            return ColumnType.FLOAT;
         } else if (t.contains("double")) {
-            if (column.isNullable()) {
-                return ColumnType.JAVA_LANG_DOUBLE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
             }
-            return ColumnType.DOUBLE;
         } else if (t.contains("timestamp")) {
-            return ColumnType.TIMESTAMP;
+            columnType = ColumnType.TIMESTAMP;
         } else if (t.contains("date")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("datetime")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("decimal")) {
-            return ColumnType.DECIMAL;
+            columnType = ColumnType.DECIMAL;
         } else if (t.contains("time")) {
-            return ColumnType.DOUBLE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
+            }
         }
-        return ColumnType.STRING;
+        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
+        return columnType;
     }
 
     @Override

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

@@ -18,16 +18,33 @@ public class MySqlTypeConvert implements ITypeConvert {
             return columnType;
         }
         String t = column.getType().toLowerCase();
+        boolean isNullable = !column.isKeyFlag() && column.isNullable();
         if (t.contains("numeric") || t.contains("decimal")) {
             columnType = ColumnType.DECIMAL;
         } else if (t.contains("bigint")) {
-            columnType = ColumnType.LONG;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_LONG;
+            } else {
+                columnType = ColumnType.LONG;
+            }
         } else if (t.contains("float")) {
-            columnType = ColumnType.FLOAT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_FLOAT;
+            } else {
+                columnType = ColumnType.FLOAT;
+            }
         } else if (t.contains("double")) {
-            columnType = ColumnType.DOUBLE;
-        } else if (t.contains("boolean") || t.contains("tinyint(1)")) {
-            columnType = ColumnType.BOOLEAN;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
+            }
+        } else if (t.contains("boolean") || t.contains("tinyint(1)") || t.contains("bit")) {
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BOOLEAN;
+            } else {
+                columnType = ColumnType.BOOLEAN;
+            }
         } else if (t.contains("datetime")) {
             columnType = ColumnType.TIMESTAMP;
         } else if (t.contains("date")) {
@@ -40,8 +57,12 @@ public class MySqlTypeConvert implements ITypeConvert {
             columnType = ColumnType.STRING;
         } else if (t.contains("binary") || t.contains("blob")) {
             columnType = ColumnType.BYTES;
-        } else if (t.contains("tinyint") || t.contains("mediumint") || t.contains("smallint") || t.contains("int") ) {
-            columnType = ColumnType.INTEGER;
+        } else if (t.contains("tinyint") || t.contains("mediumint") || t.contains("smallint") || t.contains("int")) {
+            if (isNullable) {
+                columnType = ColumnType.INTEGER;
+            } else {
+                columnType = ColumnType.INT;
+            }
         }
         columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
         return columnType;
@@ -59,12 +80,16 @@ public class MySqlTypeConvert implements ITypeConvert {
             case DECIMAL:
                 return "decimal";
             case LONG:
+            case JAVA_LANG_LONG:
                 return "bigint";
             case FLOAT:
+            case JAVA_LANG_FLOAT:
                 return "float";
             case DOUBLE:
+            case JAVA_LANG_DOUBLE:
                 return "double";
             case BOOLEAN:
+            case JAVA_LANG_BOOLEAN:
                 return "boolean";
             case TIMESTAMP:
                 return "datetime";
@@ -75,6 +100,7 @@ public class MySqlTypeConvert implements ITypeConvert {
             case BYTES:
                 return "binary";
             case INTEGER:
+            case INT:
                 return "int";
             default:
                 return "varchar";

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

@@ -18,6 +18,7 @@ public class OracleTypeConvert implements ITypeConvert {
             return columnType;
         }
         String t = column.getType().toLowerCase();
+        boolean isNullable = !column.isKeyFlag() && column.isNullable();
         if (t.contains("char")) {
             columnType = ColumnType.STRING;
         } else if (t.contains("date")) {
@@ -26,14 +27,26 @@ public class OracleTypeConvert implements ITypeConvert {
             columnType = ColumnType.TIMESTAMP;
         } else if (t.contains("number")) {
             if (t.matches("number\\(+\\d\\)")) {
-                columnType = ColumnType.INTEGER;
+                if (isNullable) {
+                    columnType = ColumnType.INTEGER;
+                } else {
+                    columnType = ColumnType.INT;
+                }
             } else if (t.matches("number\\(+\\d{2}+\\)")) {
-                columnType = ColumnType.JAVA_LANG_LONG;
+                if (isNullable) {
+                    columnType = ColumnType.JAVA_LANG_LONG;
+                } else {
+                    columnType = ColumnType.LONG;
+                }
             } else {
                 columnType = ColumnType.DECIMAL;
             }
         } else if (t.contains("float")) {
-            columnType = ColumnType.FLOAT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_FLOAT;
+            } else {
+                columnType = ColumnType.FLOAT;
+            }
         } else if (t.contains("clob")) {
             columnType = ColumnType.STRING;
         } else if (t.contains("blob")) {
@@ -53,10 +66,13 @@ public class OracleTypeConvert implements ITypeConvert {
             case TIMESTAMP:
                 return "timestamp";
             case INTEGER:
+            case INT:
             case LONG:
+            case JAVA_LANG_LONG:
             case DECIMAL:
                 return "number";
             case FLOAT:
+            case JAVA_LANG_FLOAT:
                 return "float";
             case BYTES:
                 return "blob";

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

@@ -7,53 +7,81 @@ import com.dlink.model.ColumnType;
 public class PhoenixTypeConvert implements ITypeConvert {
     @Override
     public ColumnType convert(Column column) {
-        if (Asserts.isNull(column) || Asserts.isNull(column.getType())) {
-            return ColumnType.STRING;
+        ColumnType columnType = ColumnType.STRING;
+        if (Asserts.isNull(column)) {
+            return columnType;
         }
         String t = column.getType().toLowerCase();
+        boolean isNullable = !column.isKeyFlag() && column.isNullable();
         if (t.contains("char") || t.contains("varchar") || t.contains("text") ||
-                t.contains("nchar") || t.contains("nvarchar") || t.contains("ntext")
-                || t.contains("uniqueidentifier") || t.contains("sql_variant")) {
-            return ColumnType.STRING;
+            t.contains("nchar") || t.contains("nvarchar") || t.contains("ntext")
+            || t.contains("uniqueidentifier") || t.contains("sql_variant")) {
+            columnType = ColumnType.STRING;
         } else if (t.contains("bigint")) {
-            return ColumnType.LONG;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_LONG;
+            } else {
+                columnType = ColumnType.LONG;
+            }
         } else if (t.contains("int") || t.contains("tinyint") || t.contains("smallint") || t.contains("integer")) {
-            return ColumnType.INTEGER;
+            if (isNullable) {
+                columnType = ColumnType.INTEGER;
+            } else {
+                columnType = ColumnType.INT;
+            }
         } else if (t.contains("float")) {
-            return ColumnType.FLOAT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_FLOAT;
+            } else {
+                columnType = ColumnType.FLOAT;
+            }
         } else if (t.contains("decimal") || t.contains("money") || t.contains("smallmoney")
-                || t.contains("numeric")) {
-            return ColumnType.DECIMAL;
+            || t.contains("numeric")) {
+            columnType = ColumnType.DECIMAL;
         } else if (t.contains("double")) {
-            return ColumnType.DOUBLE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
+            }
         } else if (t.contains("boolean")) {
-            return ColumnType.BOOLEAN;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BOOLEAN;
+            } else {
+                columnType = ColumnType.BOOLEAN;
+            }
         } else if (t.contains("smalldatetime") || t.contains("datetime")) {
-            return ColumnType.TIMESTAMP;
+            columnType = ColumnType.TIMESTAMP;
         } else if (t.contains("timestamp") || t.contains("binary") || t.contains("varbinary") || t.contains("image")) {
-            return ColumnType.BYTES;
+            columnType = ColumnType.BYTES;
         } else if (t.contains("time")) {
-            return ColumnType.TIME;
+            columnType = ColumnType.TIME;
         } else if (t.contains("date")) {
-            return ColumnType.DATE;
+            columnType = ColumnType.DATE;
         }
-        return ColumnType.STRING;
+        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
+        return columnType;
     }
 
     @Override
     public String convertToDB(ColumnType columnType) {
         switch (columnType) {
             case INTEGER:
+            case INT:
                 return "integer";
             case DOUBLE:
+            case JAVA_LANG_DOUBLE:
                 return "double";
             case LONG:
+            case JAVA_LANG_LONG:
                 return "bigint";
             case FLOAT:
+            case JAVA_LANG_FLOAT:
                 return "float";
             case DECIMAL:
                 return "decimal";
             case BOOLEAN:
+            case JAVA_LANG_BOOLEAN:
                 return "boolean";
             case TIME:
                 return "time";

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

@@ -13,56 +13,89 @@ import com.dlink.model.ColumnType;
 public class PostgreSqlTypeConvert implements ITypeConvert {
     @Override
     public ColumnType convert(Column column) {
+        ColumnType columnType = ColumnType.STRING;
         if (Asserts.isNull(column)) {
-            return ColumnType.STRING;
+            return columnType;
         }
         String t = column.getType().toLowerCase();
+        boolean isNullable = !column.isKeyFlag() && column.isNullable();
         if (t.contains("smallint") || t.contains("int2") || t.contains("smallserial") || t.contains("serial2")) {
-            return ColumnType.SHORT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_SHORT;
+            } else {
+                columnType = ColumnType.SHORT;
+            }
         } else if (t.contains("integer") || t.contains("serial")) {
-            return ColumnType.INTEGER;
+            if (isNullable) {
+                columnType = ColumnType.INTEGER;
+            } else {
+                columnType = ColumnType.INT;
+            }
         } else if (t.contains("bigint") || t.contains("bigserial")) {
-            return ColumnType.LONG;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_LONG;
+            } else {
+                columnType = ColumnType.LONG;
+            }
         } else if (t.contains("real") || t.contains("float4")) {
-            return ColumnType.FLOAT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_FLOAT;
+            } else {
+                columnType = ColumnType.FLOAT;
+            }
         } else if (t.contains("float8") || t.contains("double precision")) {
-            return ColumnType.DOUBLE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
+            }
         } else if (t.contains("numeric") || t.contains("decimal")) {
-            return ColumnType.DECIMAL;
+            columnType = ColumnType.DECIMAL;
         } else if (t.contains("boolean")) {
-            return ColumnType.BOOLEAN;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BOOLEAN;
+            } else {
+                columnType = ColumnType.BOOLEAN;
+            }
         } else if (t.contains("timestamp")) {
-            return ColumnType.TIMESTAMP;
+            columnType = ColumnType.TIMESTAMP;
         } else if (t.contains("date")) {
-            return ColumnType.DATE;
+            columnType = ColumnType.DATE;
         } else if (t.contains("time")) {
-            return ColumnType.TIME;
+            columnType = ColumnType.TIME;
         } else if (t.contains("char") || t.contains("text")) {
-            return ColumnType.STRING;
+            columnType = ColumnType.STRING;
         } else if (t.contains("bytea")) {
-            return ColumnType.BYTES;
+            columnType = ColumnType.BYTES;
         } else if (t.contains("array")) {
-            return ColumnType.T;
+            columnType = ColumnType.T;
         }
-        return ColumnType.STRING;
+        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
+        return columnType;
     }
 
     @Override
     public String convertToDB(ColumnType columnType) {
         switch (columnType) {
             case SHORT:
+            case JAVA_LANG_SHORT:
                 return "int2";
             case INTEGER:
+            case INT:
                 return "integer";
             case LONG:
+            case JAVA_LANG_LONG:
                 return "bigint";
             case FLOAT:
+            case JAVA_LANG_FLOAT:
                 return "float4";
             case DOUBLE:
+            case JAVA_LANG_DOUBLE:
                 return "float8";
             case DECIMAL:
                 return "decimal";
             case BOOLEAN:
+            case JAVA_LANG_BOOLEAN:
                 return "boolean";
             case TIMESTAMP:
                 return "timestamp";

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

@@ -12,33 +12,56 @@ import com.dlink.model.ColumnType;
 public class SqlServerTypeConvert implements ITypeConvert {
     @Override
     public ColumnType convert(Column column) {
+        ColumnType columnType = ColumnType.STRING;
         if (Asserts.isNull(column)) {
-            return ColumnType.STRING;
+            return columnType;
         }
         String t = column.getType().toLowerCase();
+        boolean isNullable = !column.isKeyFlag() && column.isNullable();
         if (t.contains("char") || t.contains("varchar") || t.contains("text") ||
-                t.contains("nchar") || t.contains("nvarchar") || t.contains("ntext")
-                || t.contains("uniqueidentifier") || t.contains("sql_variant")) {
-            return ColumnType.STRING;
+            t.contains("nchar") || t.contains("nvarchar") || t.contains("ntext")
+            || t.contains("uniqueidentifier") || t.contains("sql_variant")) {
+            columnType = ColumnType.STRING;
         } else if (t.contains("bigint")) {
-            return ColumnType.LONG;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_LONG;
+            } else {
+                columnType = ColumnType.LONG;
+            }
         } else if (t.contains("bit")) {
-            return ColumnType.BOOLEAN;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_BOOLEAN;
+            } else {
+                columnType = ColumnType.BOOLEAN;
+            }
         } else if (t.contains("int") || t.contains("tinyint") || t.contains("smallint")) {
-            return ColumnType.INTEGER;
+            if (isNullable) {
+                columnType = ColumnType.INTEGER;
+            } else {
+                columnType = ColumnType.INT;
+            }
         } else if (t.contains("float")) {
-            return ColumnType.DOUBLE;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_DOUBLE;
+            } else {
+                columnType = ColumnType.DOUBLE;
+            }
         } else if (t.contains("decimal") || t.contains("money") || t.contains("smallmoney")
-                || t.contains("numeric")) {
-            return ColumnType.DECIMAL;
+            || t.contains("numeric")) {
+            columnType = ColumnType.DECIMAL;
         } else if (t.contains("real")) {
-            return ColumnType.FLOAT;
+            if (isNullable) {
+                columnType = ColumnType.JAVA_LANG_FLOAT;
+            } else {
+                columnType = ColumnType.FLOAT;
+            }
         } else if (t.contains("smalldatetime") || t.contains("datetime")) {
-            return ColumnType.TIMESTAMP;
+            columnType = ColumnType.TIMESTAMP;
         } else if (t.contains("timestamp") || t.contains("binary") || t.contains("varbinary") || t.contains("image")) {
-            return ColumnType.BYTES;
+            columnType = ColumnType.BYTES;
         }
-        return ColumnType.STRING;
+        columnType.setPrecisionAndScale(column.getPrecision(), column.getScale());
+        return columnType;
     }
 
     @Override
@@ -47,14 +70,19 @@ public class SqlServerTypeConvert implements ITypeConvert {
             case STRING:
                 return "varchar";
             case BOOLEAN:
+            case JAVA_LANG_BOOLEAN:
                 return "bit";
             case LONG:
+            case JAVA_LANG_LONG:
                 return "bigint";
             case INTEGER:
+            case INT:
                 return "int";
             case DOUBLE:
+            case JAVA_LANG_DOUBLE:
                 return "double";
             case FLOAT:
+            case JAVA_LANG_FLOAT:
                 return "float";
             case TIMESTAMP:
                 return "datetime(0)";