[hotfix-407][all] fix bugs and add dirty-plugins.

Author: FlechazoW

README.md

@@ -31,7 +31,6 @@ FlinkStreamSQL
 
 ## 目录
 
-[ 1.0 变更记录](docs/changelog.md)  
 [ 1.1 demo](docs/demo.md)  
 [ 1.2 快速开始](docs/quickStart.md)  
 [ 1.3 参数配置](docs/config.md)  

core/src/main/java/com/dtstack/flink/sql/exception/ExceptionTrace.java

@@ -20,4 +20,21 @@ public static String traceOriginalCause(Throwable e) {
         }
         return errorMsg;
     }
+
+    /**
+     * 根据异常的种类来判断是否需要强制跳过Flink的重启{@link SuppressRestartsException}
+      * @param e exception
+     * @param errorMsg 需要抛出的异常信息
+     */
+    public static void dealExceptionWithSuppressStart(Exception e, String errorMsg) {
+        if (e instanceof SuppressRestartsException) {
+            throw new SuppressRestartsException(
+                new Throwable(
+                    errorMsg
+                )
+            );
+        } else {
+            throw new RuntimeException(errorMsg);
+        }
+    }
 }

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

@@ -218,6 +218,22 @@ private void checkSupport(SqlIdentifier identifier) {
         Preconditions.checkState(isSide, errorMsg);
     }
 
+    private void checkSupport(SqlIdentifier identifier) {
+        String tableName = identifier.getComponent(0).getSimple();
+        String sideTableName;
+        String sideTableAlias;
+        if (joinInfo.isLeftIsSideTable()) {
+            sideTableName = joinInfo.getLeftTableName();
+            sideTableAlias = joinInfo.getLeftTableAlias();
+        } else {
+            sideTableName = joinInfo.getRightTableName();
+            sideTableAlias = joinInfo.getRightTableAlias();
+        }
+        boolean isSide = tableName.equals(sideTableName) || tableName.equals(sideTableAlias);
+        String errorMsg = "only support set side table constant field, error field " + identifier;
+        Preconditions.checkState(isSide, errorMsg);
+    }
+
     private void associateField(String sourceTableField, String sideTableField, SqlNode sqlNode) {
         String errorMsg = "can't deal equal field: " + sqlNode;
         equalFieldList.add(sideTableField);