Merge branch '1.10_release_4.0.x' into 1.10_release_4.1.x

gituser · gituser · commit 89e282dfdc63 · 2020-11-18T19:39:13.000+08:00
diff --git a/core/src/main/java/com/dtstack/flink/sql/exec/FlinkSQLExec.java b/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);
 
diff --git a/core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java b/core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java
@@ -163,10 +163,21 @@ private static void rebuildSelectNode(SqlNodeList selectList, SqlSelect sqlSelec
 
         for (int index = 0; index < selectList.size(); index++) {
             if (selectList.get(index).getKind().equals(SqlKind.AS)
-                    || ((SqlIdentifier) selectList.get(index)).names.size() == 1) {
+                    || (selectList.get(index).getClass().equals(SqlIdentifier.class)
+                    && ((SqlIdentifier) selectList.get(index)).names.size() == 1)) {
                 sqlNodes.add(selectList.get(index));
                 continue;
             }
+
+            if (!selectList.get(index).getClass().equals(SqlIdentifier.class)) {
+                if (selectList.get(index).getKind().equals(SqlKind.LITERAL)) {
+                    throw new IllegalArgumentException(String.format("Constants %s in the SELECT statement must be aliased!",
+                            selectList.get(index).toString()));
+                }
+                throw new RuntimeException(String.format("Illegal statement! Please check the statement: %s",
+                        selectList.get(index).toString()));
+            }
+
             sqlNodes.add(transformToAsNode(selectList.get(index)));
         }
         sqlSelect.setSelectList(sqlNodes);
diff --git a/core/src/main/java/com/dtstack/flink/sql/parser/SqlParser.java b/core/src/main/java/com/dtstack/flink/sql/parser/SqlParser.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.parser;
 
@@ -27,6 +27,8 @@
 import org.apache.commons.lang3.StringUtils;
 import com.google.common.collect.Lists;
 import com.google.common.base.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.Set;
@@ -41,6 +43,7 @@
  */
 
 public class SqlParser {
+    private static final Logger LOG = LoggerFactory.getLogger(SqlParser.class);
 
     private static final char SQL_DELIMITER = ';';
 
@@ -83,13 +86,18 @@ public static SqlTree parseSql(String sql, String pluginLoadMode) throws Excepti
             }
             boolean result = false;
             for(IParser sqlParser : sqlParserList){
-                if(!sqlParser.verify(childSql)){
-                    continue;
-                }
+                try {
+                    if (!sqlParser.verify(childSql)) {
+                        continue;
+                    }
 
-                sqlParser.parseSql(childSql, sqlTree);
-                result = true;
-                break;
+                    sqlParser.parseSql(childSql, sqlTree);
+                    result = true;
+                    break;
+                } catch (Exception e) {
+                    LOG.error("'{}' parser error, detail info: {}", childSql, e.getMessage(), e);
+                    throw e;
+                }
             }
 
             if(!result){
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java b/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);
@@ -310,6 +311,64 @@ private List<String> convertPrimaryAlias(AbstractSideTableInfo sideTableInfo) {
         return res;
     }
 
+    /**
+     * check whether all table fields exist in join condition.
+     * @param conditionNode
+     * @param joinScope
+     */
+    public void checkConditionFieldsInTable(SqlNode conditionNode, JoinScope joinScope) {
+        List<SqlNode> sqlNodeList = Lists.newArrayList();
+        ParseUtils.parseAnd(conditionNode, sqlNodeList);
+        for (SqlNode sqlNode : sqlNodeList) {
+            if (!SqlKind.COMPARISON.contains(sqlNode.getKind())) {
+                throw new RuntimeException("not compare operator.");
+            }
+
+            SqlNode leftNode = ((SqlBasicCall) sqlNode).getOperands()[0];
+            SqlNode rightNode = ((SqlBasicCall) sqlNode).getOperands()[1];
+
+            if (leftNode.getKind() == SqlKind.IDENTIFIER) {
+                checkFieldInTable((SqlIdentifier) leftNode, joinScope, conditionNode);
+            }
+
+            if (rightNode.getKind() == SqlKind.IDENTIFIER) {
+                checkFieldInTable((SqlIdentifier) rightNode, joinScope, conditionNode);
+            }
+
+        }
+    }
+
+    /**
+     * check whether table exists and whether field is in table.
+     * @param sqlNode
+     * @param joinScope
+     * @param conditionNode
+     */
+    private void checkFieldInTable(SqlIdentifier sqlNode, JoinScope joinScope, SqlNode conditionNode) {
+        String tableName = sqlNode.getComponent(0).getSimple();
+        String fieldName = sqlNode.getComponent(1).getSimple();
+        JoinScope.ScopeChild scopeChild = joinScope.getScope(tableName);
+        String tableErrorMsg = "table [%s] is not exist. error condition is [%s]. if you find [%s] is exist, please check AS statement";
+        Preconditions.checkState(
+            scopeChild != null,
+            tableErrorMsg,
+            tableName,
+            conditionNode.toString(),
+            tableName
+        );
+
+        String[] fieldNames = scopeChild.getRowTypeInfo().getFieldNames();
+        boolean hasField = Arrays.asList(fieldNames).contains(fieldName);
+        String fieldErrorMsg = "table [%s] has not [%s] field.\n error join condition is [%s]";
+        Preconditions.checkState(
+            hasField,
+            fieldErrorMsg,
+            tableName,
+            fieldName,
+            conditionNode.toString()
+        );
+    }
+
     public List<String> getConditionFields(SqlNode conditionNode, String specifyTableName, AbstractSideTableInfo sideTableInfo) {
         List<SqlNode> sqlNodeList = Lists.newArrayList();
         ParseUtils.parseAnd(conditionNode, sqlNodeList);
@@ -421,6 +480,7 @@ private void joinFun(Object pollObj,
         joinScope.addScope(rightScopeChild);
 
         HashBasedTable<String, String, String> mappingTable = ((JoinInfo) pollObj).getTableFieldRef();
+        checkConditionFieldsInTable(joinInfo.getCondition(), joinScope);
 
         //获取两个表的所有字段
         List<FieldInfo> sideJoinFieldInfo = ParserJoinField.getRowTypeInfo(joinInfo.getSelectNode(), joinScope, true);
diff --git a/core/src/main/java/com/dtstack/flink/sql/util/SqlCheckUtils.java b/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);
+    }
+
+}
