public String buildCreateTableStatement(String database, String table, List<String> primaryKeys,
            CreateTableOperation operation) {
        StringBuilder sb = new StringBuilder();
        sb.append("CREATE TABLE IF NOT EXISTS `").append(database).append("`.`").append(table).append("`(\n");
        Preconditions.checkState(operation.getColumns() != null && !operation.getColumns().isEmpty(),
                String.format("The columns of table: %s.%s is empty", database, table));
        Iterator<Column> iterator = operation.getColumns().iterator();
        StringJoiner joiner = new StringJoiner(",");
        while (iterator.hasNext()) {
            Column column = iterator.next();
            Preconditions.checkNotNull(column, "The column is null");
            Preconditions.checkState(column.getName() != null && !column.getName().trim().isEmpty(),
                    "The column name is blank");
            sb.append("\t`").append(column.getName()).append("` ").append(convert2DorisType(column.getJdbcType(),
                    column.isNullable(), column.getDefinition()));
            if (validDefaultValue(column.getDefaultValue())) {
                sb.append(" DEFAULT ").append(quote(column.getDefaultValue()));
            }
            if (column.getComment() != null) {
                sb.append(" COMMENT ").append(quote(column.getComment()));
            }
            joiner.add(String.format("`%s`", column.getName()));
            if (iterator.hasNext()) {
                sb.append(",\n");
            }
        }
        sb.append("\n)\n");
        String model = "DUPLICATE";
        if (primaryKeys != null && !primaryKeys.isEmpty()) {
            model = "UNIQUE";
            joiner = new StringJoiner(",");
            for (String primaryKey : primaryKeys) {
                joiner.add(String.format("`%s`", primaryKey));
            }
        }
        String keys = joiner.toString();
        sb.append(model).append(" KEY(").append(keys).append(")");
        if (StringUtils.isNotBlank(operation.getComment())) {
            sb.append("\nCOMMENT ").append(quote(operation.getComment()));
        }
        sb.append("\nDISTRIBUTED BY HASH(").append(keys).append(")");
        // Add light schema change support for it if the version of doris is greater than 1.2.0 or equals 1.2.0
        sb.append("\nPROPERTIES (\n\t\"light_schema_change\" = \"true\"\n)");
        return sb.toString();
    }
Build the statement of CreateTable

@param database The database of Doris
@param table The table of Doris
@param primaryKeys The primary key of Doris
@param operation The Operation
@return A statement of CreateTable
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/schema/OperationHelper.java