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

Author: gituser

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

@@ -189,6 +189,10 @@ public List<PredicateInfo> getPredicateInfoes() {
         return predicateInfoes;
     }
 
+    public void addPredicateInfo(PredicateInfo predicateInfo) {
+        this.predicateInfoes.add(predicateInfo);
+    }
+
     public Long getAsyncFailMaxNum(Long defaultValue) {
         return Objects.isNull(asyncFailMaxNum) ? defaultValue : asyncFailMaxNum;
     }

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

@@ -134,12 +134,14 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
         if (leftNode.getKind() == SqlKind.LITERAL) {
             evalConstantEquation(
                 (SqlLiteral) leftNode,
-                (SqlIdentifier) rightNode
+                (SqlIdentifier) rightNode,
+                sqlNode.getKind()
             );
         } else if (rightNode.getKind() == SqlKind.LITERAL) {
             evalConstantEquation(
                 (SqlLiteral) rightNode,
-                (SqlIdentifier) leftNode
+                (SqlIdentifier) leftNode,
+                sqlNode.getKind()
             );
         } else {
             SqlIdentifier left = (SqlIdentifier) leftNode;
@@ -176,22 +178,21 @@ private void evalEquation(SqlIdentifier left, SqlIdentifier right, String sideTa
      * @param literal
      * @param identifier
      */
-    private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier) {
+    private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier, SqlKind sqlKind) {
         String tableName = identifier.getComponent(0).getSimple();
         String sideTableName = sideTableInfo.getName();
         String errorMsg = "only support set side table constant field, error field " + identifier;
         Preconditions.checkState(tableName.equals(sideTableName), errorMsg);
         String fieldName = identifier.getComponent(1).getSimple();
         Object constant = literal.getValue();
-        List<PredicateInfo> predicateInfos = sideTableInfo.getPredicateInfoes();
         PredicateInfo predicate = PredicateInfo.builder()
-            .setOperatorName("=")
-            .setOperatorKind("EQUALS")
+            .setOperatorName(sqlKind.sql)
+            .setOperatorKind(sqlKind.name())
             .setOwnerTable(tableName)
             .setFieldName(fieldName)
             .setCondition(constant.toString())
             .build();
-        predicateInfos.add(predicate);
+        sideTableInfo.addPredicateInfo(predicate);
     }
 
     private void associateField(String sourceTableField, String sideTableField, SqlNode sqlNode) {