flinksql字段名忽略大小写，增加warn日志

Author: dengdejiushini

core/src/main/java/com/dtstack/flink/sql/exec/FlinkSQLExec.java

@@ -27,6 +27,8 @@
 import org.apache.flink.table.plan.logical.LogicalRelNode;
 import org.apache.flink.table.plan.schema.TableSinkTable;
 import org.apache.flink.table.plan.schema.TableSourceSinkTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import scala.Option;
 
 import java.lang.reflect.Method;
@@ -38,6 +40,8 @@
  */
 public class FlinkSQLExec {
 
+    private static final Logger LOG = LoggerFactory.getLogger(FlinkSQLExec.class);
+
     public static void sqlUpdate(StreamTableEnvironment tableEnv, String stmt, StreamQueryConfig queryConfig) throws Exception {
 
         FlinkPlannerImpl planner = new FlinkPlannerImpl(tableEnv.getFrameworkConfig(), tableEnv.getPlanner(), tableEnv.getTypeFactory());
@@ -89,6 +93,7 @@ public static void sqlUpdate(StreamTableEnvironment tableEnv, String stmt, Strea
         try {
             tableEnv.insertInto(newTable, targetTableName, config);
         } catch (Exception ex) {
+            LOG.warn("Field name case of query result and registered TableSink " + targetTableName + "do not match. " + ex.getMessage());
             newTable = queryResult.select(String.join(",", ignoreCase(queryFieldNames, sinkFieldNames)));
             tableEnv.insertInto(newTable, targetTableName, config);
         }