Merge pull request #145 from zhu-mingye/dev

[fixbug]修复flinksqlDDL注释中存在单双引号引起语法校验失败的问题 优化SQLDDL中字段添加`` 防止关键字问题造成校验失败

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

@@ -82,9 +82,13 @@ public class Table implements Serializable, Comparable<Table> {
             if (i > 0) {
                 sb.append(",");
             }
-            sb.append(columns.get(i).getName() + " " + type);
+            sb.append("`" + columns.get(i).getName() + "` " + type);
             if (Asserts.isNotNullString(columns.get(i).getComment())) {
-                sb.append(" COMMENT '"+columns.get(i).getComment() + "'");
+                if(columns.get(i).getComment().contains("\'") | columns.get(i).getComment().contains("\"")) {
+                    sb.append(" COMMENT '" + columns.get(i).getComment().replaceAll("\"|'","") + "'");
+                }else {
+                    sb.append(" COMMENT '" + columns.get(i).getComment() + "'");
+                }
             }
             sb.append("\n");
             if (columns.get(i).isKeyFlag()) {
@@ -96,7 +100,7 @@ public class Table implements Serializable, Comparable<Table> {
             if (i > 0) {
                 pksb.append(",");
             }
-            pksb.append(pks.get(i));
+            pksb.append("`"+pks.get(i)+"`");
         }
         pksb.append(" ) NOT ENFORCED\n");
         if (pks.size() > 0) {
@@ -105,7 +109,11 @@ public class Table implements Serializable, Comparable<Table> {
         }
         sb.append(")");
         if(Asserts.isNotNullString(comment)){
-            sb.append(" COMMENT '"+comment+"'\n");
+            if(comment.contains("\'") | comment.contains("\"")) {
+                sb.append(" COMMENT '" + comment.replaceAll("\"|'","") + "'\n");
+            }else {
+                sb.append(" COMMENT '" + comment + "'\n");
+            }
         }
         sb.append(" WITH (\n");
         sb.append(getFlinkTableWith(flinkConfig));
@@ -113,6 +121,7 @@ public class Table implements Serializable, Comparable<Table> {
         return sb.toString();
     }
 
+
     public String getSqlSelect(String catalogName) {
         StringBuilder sb = new StringBuilder("SELECT\n");
         for (int i = 0; i < columns.size(); i++) {
@@ -120,9 +129,22 @@ public class Table implements Serializable, Comparable<Table> {
             if (i > 0) {
                 sb.append(",");
             }
-            sb.append(columns.get(i).getName() + "\n");
+            String columnComment= columns.get(i).getComment();
+            if(columnComment.contains("\'") | columnComment.contains("\"")) {
+                columnComment = columnComment.replaceAll("\"|'","");
+            }
+            if(Asserts.isNotNullString(columnComment)){
+                sb.append("`"+columns.get(i).getName() + "`  --  " + columnComment + " \n");
+            }else {
+                sb.append("`"+columns.get(i).getName() + "` \n");
+
+            }
+        }
+        if(Asserts.isNotNullString(comment)){
+            sb.append(" FROM " + catalogName + "." + schema + "." + name + ";" + " -- " + comment + "\n");
+        }else {
+            sb.append(" FROM " + catalogName + "." + schema + "." + name +";\n");
         }
-        sb.append(" FROM " + catalogName + "." + schema + "." + name + ";\n");
         return sb.toString();
     }
 }

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

@@ -166,7 +166,10 @@ public abstract class AbstractJdbcDriver extends AbstractDriver {
                 field.setName(columnName);
                 field.setType(results.getString(dbQuery.columnType()));
                 field.setJavaType(getTypeConvert().convert(field.getType()).getType());
-                field.setComment(results.getString(dbQuery.columnComment()));
+
+                String columnComment=results.getString(dbQuery.columnComment()).replaceAll("\"|'","");
+                field.setComment(columnComment);
+
                 field.setNullable(Asserts.isEqualsIgnoreCase(results.getString(dbQuery.isNullable()),"YES"));
                 field.setCharacterSet(results.getString(dbQuery.characterSet()));
                 field.setCollation(results.getString(dbQuery.collation()));