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

Author: gituser

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

@@ -22,15 +22,11 @@
 
 import com.dtstack.flink.sql.side.cache.AbstractSideCache;
 import com.google.common.base.Preconditions;
-import org.apache.calcite.sql.JoinType;
-import org.apache.calcite.sql.SqlBasicCall;
-import org.apache.calcite.sql.SqlIdentifier;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlLiteral;
-import org.apache.calcite.sql.SqlNode;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import org.apache.calcite.sql.*;
+import org.apache.calcite.util.NlsString;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.table.runtime.typeutils.BaseRowTypeInfo;
 
 import java.io.Serializable;
@@ -46,6 +42,9 @@
 
 public abstract class BaseSideInfo implements Serializable{
 
+    protected static final String QUOTE = "'";
+    protected static final String ESCAPEQUOTE = "''";
+
     protected RowTypeInfo rowTypeInfo;
 
     protected List<FieldInfo> outFieldInfoList;
@@ -186,12 +185,18 @@ private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier,
         checkSupport(identifier);
         String fieldName = identifier.getComponent(1).getSimple();
         Object constant = literal.getValue();
+        String condition;
+        if(constant instanceof NlsString){
+            condition = QUOTE + ((NlsString) constant).getValue().replace(QUOTE,ESCAPEQUOTE)+ QUOTE;
+        }else {
+            condition = constant.toString();
+        }
         PredicateInfo predicate = PredicateInfo.builder()
             .setOperatorName(sqlKind.sql)
             .setOperatorKind(sqlKind.name())
             .setOwnerTable(tableName)
             .setFieldName(fieldName)
-            .setCondition(constant.toString())
+            .setCondition(condition)
             .build();
         sideTableInfo.addPredicateInfo(predicate);
         sideTableInfo.addFullPredicateInfoes(predicate);