Merge branch 'hotfix_1.10_4.0.x_31007' into 1.10_release_4.0.x

Author: gituser

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

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.exec;
 
+import com.dtstack.flink.sql.util.SqlCheckUtils;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.flink.sql.parser.dml.RichSqlInsert;
@@ -58,7 +59,7 @@ public static void sqlUpdate(StreamTableEnvironment tableEnv, String stmt) throw
         StreamTableEnvironmentImpl tableEnvImpl = ((StreamTableEnvironmentImpl) tableEnv);
         StreamPlanner streamPlanner = (StreamPlanner)tableEnvImpl.getPlanner();
         FlinkPlannerImpl flinkPlanner = streamPlanner.createFlinkPlanner();
-
+        SqlCheckUtils.check(stmt);
         RichSqlInsert insert = (RichSqlInsert) flinkPlanner.validate(flinkPlanner.parser().parse(stmt));
         TableImpl queryResult = extractQueryTableFromInsertCaluse(tableEnvImpl, flinkPlanner, insert);
 

core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

@@ -28,6 +28,7 @@
 import com.dtstack.flink.sql.side.operator.SideWithAllCacheOperator;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.ParseUtils;
+import com.dtstack.flink.sql.util.SqlCheckUtils;
 import com.dtstack.flink.sql.util.TableUtils;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.HashBasedTable;
@@ -151,7 +152,7 @@ public void exec(String sql,
 
                 } else if (pollSqlNode.getKind() == SELECT) {
                     Preconditions.checkState(createView != null, "select sql must included by create view");
-                    Table table = tableEnv.sqlQuery(pollObj.toString());
+                    Table table = SqlCheckUtils.sqlQueryWithCheck(tableEnv, pollObj.toString());
 
                     if (createView.getFieldsInfoStr() == null) {
                         tableEnv.registerTable(createView.getTableName(), table);

core/src/main/java/com/dtstack/flink/sql/util/SqlCheckUtils.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.util;
+
+import com.google.common.base.Preconditions;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/10/13
+ **/
+public class SqlCheckUtils {
+
+    private final static Pattern NULL_AS_PATTERN = Pattern.compile("(?i)NULL\\s+AS");
+
+    /**
+     * check SQL before call sqlQuery
+     * @param tEnv
+     * @param query
+     * @return
+     */
+    public static Table sqlQueryWithCheck(TableEnvironment tEnv, String query) {
+        check(query);
+        return tEnv.sqlQuery(query);
+    }
+
+    /**
+     * check SQL before pass into flink planner
+     * 在传入原生Flink之前校验SQL合法性。
+     * @param stmt
+     */
+    public static void check(String stmt) {
+        Matcher matcher = NULL_AS_PATTERN.matcher(stmt);
+        Preconditions.checkState(!matcher.find(),"NULL AS is not supported. error SQL is [%s]", stmt);
+    }
+
+}