Merge branch '1.5_v3.8.0_bugfix_joinnpe' into '1.5_v3.8.0'

[flinksql][维表join npe，错误提示不友好][19240]



See merge request !85

Author: zoudaokoulife

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

@@ -356,7 +356,16 @@ private SqlNode replaceOrderByTableName(SqlNode orderNode, String tableAlias) {
     private SqlNode replaceNodeInfo(SqlNode groupNode, HashBasedTable<String, String, String> mappingTable, String tableAlias){
         if(groupNode.getKind() == IDENTIFIER){
             SqlIdentifier sqlIdentifier = (SqlIdentifier) groupNode;
+            // 如果没有表别名前缀，直接返回字段名称
+            if (sqlIdentifier.names.size() == 1) {
+                return groupNode;
+            }
             String mappingFieldName = mappingTable.get(sqlIdentifier.getComponent(0).getSimple(), sqlIdentifier.getComponent(1).getSimple());
+            // 如果有表别名前缀，但是在宽表中找不到映射，只需要设置别名，不需要替换映射
+            if (null == mappingFieldName){
+                // return sqlIdentifier.setName(0, tableAlias);
+                throw new RuntimeException("Column '" + sqlIdentifier.getComponent(1).getSimple() + "' not found in table '" + sqlIdentifier.getComponent(0).getSimple() + "'");
+            }
             sqlIdentifier = sqlIdentifier.setName(0, tableAlias);
             return sqlIdentifier.setName(1, mappingFieldName);
         }else if(groupNode instanceof  SqlBasicCall){