@Transient
    public String getFlinkDDL(String flinkConfig, String tableName) {
        String columnStrs = columns.stream()
                .map(column -> {
                    String comment = "";
                    if (Asserts.isNotNullString(column.getComment())) {
                        comment = String.format(
                                " COMMENT '%s'", column.getComment().replaceAll("[\"']", ""));
                    }
                    return String.format("    `%s` %s%s", column.getName(), column.getFlinkType(), comment);
                })
                .collect(Collectors.joining(",\n"));

        List<String> columnKeys = columns.stream()
                .filter(Column::isKeyFlag)
                .map(Column::getName)
                .map(t -> String.format("`%s`", t))
                .collect(Collectors.toList());

        String primaryKeyStr = columnKeys.isEmpty()
                ? ""
                : columnKeys.stream().collect(Collectors.joining(",", ",\n    PRIMARY KEY ( ", " ) NOT ENFORCED\n"));

        return MessageFormat.format(
                "CREATE TABLE IF NOT EXISTS {0} (\n{1}{2}) WITH (\n{3})\n",
                tableName, columnStrs, primaryKeyStr, flinkConfig);
    }
@Test
    void getFlinkDDL() {
        String result = table.getFlinkDDL(flinkConfig, "NewTableName");
        assertThat(
                result,
                equalTo("CREATE TABLE IF NOT EXISTS NewTableName (\n"
                        + "    `column1` INT NOT NULL COMMENT 'comment abc',\n"
                        + "    `column2` STRING COMMENT 'comment abc',\n"
                        + "    `column3` DOUBLE NOT NULL COMMENT 'comment abc',\n"
                        + "    PRIMARY KEY ( `column1`,`column2` ) NOT ENFORCED\n"
                        + ") WITH (\n"
                        + "#{schemaName}=schemaName, #{tableName}=tableName, #{abc}=abc,"
                        + " #{}=null, bcd=bcd)\n"));

        result = tableWithoutKey.getFlinkDDL(flinkConfig, "NewTableNameWithoutKey");
        assertThat(
                result,
                equalTo("CREATE TABLE IF NOT EXISTS NewTableNameWithoutKey (\n"
                        + "    `column1` INT NOT NULL COMMENT 'comment abc',\n"
                        + "    `column2` STRING COMMENT 'comment abc',\n"
                        + "    `column3` DOUBLE NOT NULL COMMENT 'comment abc') WITH (\n"
                        + "#{schemaName}=schemaName, #{tableName}=tableName, #{abc}=abc,"
                        + " #{}=null, bcd=bcd)\n"));
    }