public static void sqlInsert(StreamTableEnvironment tableEnv, String stmt, Collection<String> newRegisterTableList) throws Exception{
        StreamTableEnvironmentImpl tableEnvImpl = ((StreamTableEnvironmentImpl) tableEnv);
        StreamPlanner streamPlanner = (StreamPlanner) tableEnvImpl.getPlanner();
        FlinkPlannerImpl flinkPlanner = streamPlanner.createFlinkPlanner();

        RichSqlInsert insert = (RichSqlInsert) flinkPlanner.validate(flinkPlanner.parser().parse(stmt));
        boolean isGroupByTimeWindow = TableUtils.checkIsDimTableGroupBy(insert, newRegisterTableList);
        if(isGroupByTimeWindow){
            QueryOperationConverter.setProducesUpdates(true);
        }

        sqlInsert(tableEnv, insert);
        QueryOperationConverter.setProducesUpdates(false);
    }
@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