public static void sqlInsert(StreamTableEnvironment tableEnv, RichSqlInsert insert) throws Exception {
        StreamTableEnvironmentImpl tableEnvImpl = ((StreamTableEnvironmentImpl) tableEnv);
        StreamPlanner streamPlanner = (StreamPlanner) tableEnvImpl.getPlanner();
        FlinkPlannerImpl flinkPlanner = streamPlanner.createFlinkPlanner();

        TableImpl queryResult = extractQueryTableFromInsertCaluse(tableEnvImpl, flinkPlanner, insert);
        String targetTableName = ((SqlIdentifier) insert.getTargetTable()).names.get(0);
        TableSink tableSink = getTableSinkByPlanner(streamPlanner, targetTableName);

        String[] sinkFieldNames = tableSink.getTableSchema().getFieldNames();
        String[] queryFieldNames = queryResult.getSchema().getFieldNames();
        Table newTable;
        try {
            newTable = queryResult.select(String.join(",", sinkFieldNames));
            tableEnv.insertInto(targetTableName, newTable);
        } catch (Exception e) {
            LOG.warn(String.format("Query result and registered TableSink do not match \n input field list:%s \n output field list:%s ", Arrays.toString(queryFieldNames), Arrays.toString(sinkFieldNames)));
            newTable = queryResult.select(String.join(",", ignoreCase(queryFieldNames, sinkFieldNames)));
            tableEnv.insertInto(targetTableName, newTable);
        }
    }
@description: mapping by name when insert into sink table
@author: maqi
@create: 2019/08/15 11:09
core/src/main/java/com/dtstack/flink/sql/exec/FlinkSQLExec.java