private String genCreateSql(Node node) {
        if (node instanceof TransformNode) {
            return genCreateTransformSql(node);
        }
        if (node instanceof HbaseLoadNode) {
            return genCreateHbaseLoadSql((HbaseLoadNode) node);
        }
        StringBuilder sb = new StringBuilder("CREATE TABLE `");
        sb.append(node.genTableName()).append("`(\n");
        String filterPrimaryKey = getFilterPrimaryKey(node);
        sb.append(genPrimaryKey(node.getPrimaryKey(), filterPrimaryKey));
        sb.append(parseFields(node.getFields(), node, filterPrimaryKey));
        if (node instanceof ExtractNode) {
            ExtractNode extractNode = (ExtractNode) node;
            if (extractNode.getWatermarkField() != null) {
                sb.append(",\n     ").append(extractNode.getWatermarkField().format());
            }
        }
        sb.append(")");
        if (node.getPartitionFields() != null && !node.getPartitionFields().isEmpty()) {
            sb.append(String.format("\nPARTITIONED BY (%s)",
                    StringUtils.join(formatFields(node.getPartitionFields()), ",")));
        }
        sb.append(parseOptions(node.tableOptions()));
        return sb.toString();
    }
Generate create sql

@param node The abstract of extract, transform, load
@return The creation sql pf table
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java