@Override
    public ParseResult parse() {
        Preconditions.checkNotNull(statements, "sql statement set is null");
        Preconditions.checkNotNull(tableEnv, "tableEnv is null");
        List<String> createTableSqls = new ArrayList<>();
        List<String> insertSqls = new ArrayList<>();
        String[] statementSet = statements.split(";(\\r?\\n|\\r)");
        for (String statement : statementSet) {
            statement = statement.trim();
            if (statement.toUpperCase(Locale.ROOT).startsWith("CREATE TABLE") || statement.toUpperCase(Locale.ROOT)
                    .startsWith("CREATE VIEW")) {
                createTableSqls.add(statement);
            } else if (statement.toUpperCase(Locale.ROOT).startsWith("INSERT INTO")) {
                insertSqls.add(statement);
            } else if (!statement.isEmpty()) {
                log.warn("Not support sql statement: " + statement);
            }
        }
        return new FlinkSqlParseResult(tableEnv, createTableSqls, insertSqls);
    }
parse flink sql script file

@return ParseResult the result of parsing
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/NativeFlinkSqlParser.java