Sqlserver fix 完善字段查询、字段类型判断功能 (#1021)

* Update cdcsource_statements.md

cdcsource增加支持多目标库同步功能。

* sqlserver优化字段属性类型解析.
sqlserver字段sql查询语句优化增加获取字段长度.

* sqlserver优化字段属性类型解析.
sqlserver字段sql查询语句优化增加获取字段长度.
Co-authored-by: 金鑫 <jinyanhui@huansi.net>

dlink-metadata/dlink-metadata-sqlserver/src/main/java/com/dlink/metadata/constant/SqlServerConstant.java

@@ -33,16 +33,17 @@ public interface SqlServerConstant {
      * 查询列信息模板SQL
      */
     String QUERY_COLUMNS_SQL = " SELECT  cast(a.name AS VARCHAR(500)) AS TABLE_NAME,cast(b.name AS VARCHAR(500)) AS COLUMN_NAME,  isnull(CAST ( c.VALUE AS NVARCHAR ( 500 ) ),'') AS COMMENTS, "
-            + " CASE b.is_nullable WHEN 1 THEN 'YES' ELSE 'NO' END as NULLVALUE,cast(sys.types.name AS VARCHAR (500)) AS DATA_TYPE,"
-            + " ( SELECT CASE count(1) WHEN 1 then 'PRI' ELSE '' END FROM syscolumns,sysobjects,sysindexes,sysindexkeys,systypes  WHERE syscolumns.xusertype = systypes.xusertype "
-            + " AND syscolumns.id = object_id (a.name) AND sysobjects.xtype = 'PK' AND sysobjects.parent_obj = syscolumns.id  "
-            + " AND sysindexes.id = syscolumns.id  AND sysobjects.name = sysindexes.name AND sysindexkeys.id = syscolumns.id  AND sysindexkeys.indid = sysindexes.indid  "
-            + "AND syscolumns.colid = sysindexkeys.colid "
-            + " AND syscolumns.name = b.name) as 'KEY',  b.is_identity isIdentity , '' as CHARACTER_SET_NAME, '' as COLLATION_NAME, 0 as ORDINAL_POSITION, 0 as NUMERIC_PRECISION, 0 as NUMERIC_SCALE, "
-            + "'' as AUTO_INCREMENT "
-            + "FROM ( select name,object_id from sys.tables UNION all select name,object_id from sys.views ) a  INNER JOIN sys.columns b "
-            + " ON b.object_id = a.object_id  LEFT JOIN sys.types ON b.user_type_id = sys.types.user_type_id    LEFT JOIN sys.extended_properties c ON c.major_id = b.object_id "
-            + "AND c.minor_id = b.column_id  WHERE a.name = '%s' and sys.types.name !='sysname' ";
+        + " CASE b.is_nullable WHEN 1 THEN 'YES' ELSE 'NO' END as NULLVALUE,cast(sys.types.name AS VARCHAR (500)) AS DATA_TYPE,"
+        + " ( SELECT CASE count(1) WHEN 1 then 'PRI' ELSE '' END FROM syscolumns,sysobjects,sysindexes,sysindexkeys,systypes  WHERE syscolumns.xusertype = systypes.xusertype "
+        + " AND syscolumns.id = object_id (a.name) AND sysobjects.xtype = 'PK' AND sysobjects.parent_obj = syscolumns.id  "
+        + " AND sysindexes.id = syscolumns.id  AND sysobjects.name = sysindexes.name AND sysindexkeys.id = syscolumns.id  AND sysindexkeys.indid = sysindexes.indid  "
+        + "AND syscolumns.colid = sysindexkeys.colid "
+        + " AND syscolumns.name = b.name) as 'KEY',  b.is_identity isIdentity , '' as CHARACTER_SET_NAME, '' as COLLATION_NAME, "
+        + "0 as ORDINAL_POSITION, b.PRECISION as NUMERIC_PRECISION, b.scale as NUMERIC_SCALE,"
+        + "'' as AUTO_INCREMENT "
+        + "FROM ( select name,object_id from sys.tables UNION all select name,object_id from sys.views ) a  INNER JOIN sys.columns b "
+        + " ON b.object_id = a.object_id  LEFT JOIN sys.types ON b.user_type_id = sys.types.user_type_id    LEFT JOIN sys.extended_properties c ON c.major_id = b.object_id "
+        + "AND c.minor_id = b.column_id  WHERE a.name = '%s' and sys.types.name !='sysname' ";
 
     /**
      * 查询schema模板SQL

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

@@ -33,8 +33,8 @@ public class SqlServerTypeConvert implements ITypeConvert {
         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")) {
+            || t.contains("nchar") || t.contains("nvarchar") || t.contains("ntext")
+            || t.contains("uniqueidentifier") || t.contains("sql_variant")) {
             columnType = ColumnType.STRING;
         } else if (t.contains("bigint")) {
             if (isNullable) {
@@ -60,8 +60,7 @@ public class SqlServerTypeConvert implements ITypeConvert {
             } else {
                 columnType = ColumnType.DOUBLE;
             }
-        } else if (t.contains("decimal") || t.contains("money") || t.contains("smallmoney")
-                || t.contains("numeric")) {
+        } else if (t.contains("decimal") || t.contains("money") || t.contains("smallmoney") || t.contains("numeric")) {
             columnType = ColumnType.DECIMAL;
         } else if (t.contains("real")) {
             if (isNullable) {
@@ -69,10 +68,18 @@ public class SqlServerTypeConvert implements ITypeConvert {
             } else {
                 columnType = ColumnType.FLOAT;
             }
-        } else if (t.contains("date")) {
-            columnType = ColumnType.DATE;
-        } else if (t.contains("smalldatetime") || t.contains("datetime")) {
+        } else if (t.equalsIgnoreCase("datetime") || t.equalsIgnoreCase("smalldatetime")) {
             columnType = ColumnType.TIMESTAMP;
+        } else if (t.equalsIgnoreCase("datetime2")) {
+            //这里应该是纳秒
+            columnType = ColumnType.TIMESTAMP;
+        } else if (t.equalsIgnoreCase("datetimeoffset")) {
+            //这里应该是纳秒
+            columnType = ColumnType.TIMESTAMP;
+        } else if (t.equalsIgnoreCase("date")) {
+            columnType = ColumnType.LOCALDATE;
+        } else if (t.equalsIgnoreCase("time")) {
+            columnType = ColumnType.LOCALTIME;
         } else if (t.contains("timestamp") || t.contains("binary") || t.contains("varbinary") || t.contains("image")) {
             columnType = ColumnType.BYTES;
         }