Merge pull request #154 from zhu-mingye/dev

优化FlinkDDL

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

@@ -6,11 +6,9 @@ import lombok.Getter;
 import lombok.Setter;
 
 import java.io.Serializable;
-import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
-import java.util.Map;
 
 /**
  * Table
@@ -72,8 +70,10 @@ public class Table implements Serializable, Comparable<Table> {
     }
 
     public String getFlinkTableSql(String catalogName, String flinkConfig) {
-        StringBuilder sb = new StringBuilder("CREATE TABLE ");
-        sb.append(catalogName + "." + schema + "." + name + " (\n");
+        StringBuilder sb = new StringBuilder("DROP TABLE IF EXISTS ");
+        String fullSchemaName = catalogName + "." + schema + "." + name;
+        sb.append(fullSchemaName + ";\n");
+        sb.append("CREATE TABLE IF NOT EXISTS " + fullSchemaName + " (\n");
         List<String> pks = new ArrayList<>();
         for (int i = 0; i < columns.size(); i++) {
             String type = columns.get(i).getJavaType().getFlinkType();