[feat-1583] using chinese in where clause after dim table join.

Author: a49a

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

@@ -22,6 +22,7 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlCharStringLiteral;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlJoin;
@@ -35,7 +36,9 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.calcite.sql.SqlKind.*;
+import static org.apache.calcite.sql.SqlKind.IDENTIFIER;
+import static org.apache.calcite.sql.SqlKind.LITERAL;
+import static org.apache.calcite.sql.SqlKind.OR;
 
 /**
  *
@@ -47,6 +50,7 @@
 public class SidePredicatesParser {
 
     private FlinkPlanner flinkPlanner = new FlinkPlanner();
+    private static final String QUOTE = "'";
 
     public void fillPredicatesForSideTable(String exeSql, Map<String, AbstractSideTableInfo> sideTableMap) throws SqlParseException {
         SqlNode sqlNode = flinkPlanner.getParser().parse(exeSql);
@@ -140,26 +144,57 @@ private void extractPredicateInfo(SqlNode whereNode, List<PredicateInfo> predica
         }
     }
 
-    private void  fillPredicateInfoToList(SqlBasicCall whereNode, List<PredicateInfo> predicatesInfoList, String operatorName, SqlKind operatorKind,
-                                         int fieldIndex, int conditionIndex) {
+    private void fillPredicateInfoToList(
+            SqlBasicCall whereNode,
+            List<PredicateInfo> predicatesInfoList,
+            String operatorName,
+            SqlKind operatorKind,
+            int fieldIndex,
+            int conditionIndex) {
         SqlNode sqlNode = whereNode.getOperands()[fieldIndex];
         if (sqlNode.getKind() == SqlKind.IDENTIFIER) {
             SqlIdentifier fieldFullPath = (SqlIdentifier) sqlNode;
             if (fieldFullPath.names.size() == 2) {
                 String ownerTable = fieldFullPath.names.get(0);
                 String fieldName = fieldFullPath.names.get(1);
-                String content = (operatorKind == SqlKind.BETWEEN) ? whereNode.getOperands()[conditionIndex].toString() + " AND " +
-                        whereNode.getOperands()[2].toString() : whereNode.getOperands()[conditionIndex].toString();
 
-                if (StringUtils.containsIgnoreCase(content,SqlKind.CASE.toString())) {
+                String condition;
+                SqlNode[] conditionNodes = whereNode.getOperands();
+                SqlNode literal = conditionNodes[conditionIndex];
+                if (operatorKind == SqlKind.BETWEEN) {
+                    condition = literal.toString() + " AND " + conditionNodes[2].toString();
+                } else {
+                    if (literal instanceof SqlCharStringLiteral) {
+                        condition = removeCoding((SqlCharStringLiteral) literal);
+                    } else {
+                        condition = literal.toString();
+                    }
+                }
+
+                if (StringUtils.containsIgnoreCase(condition, SqlKind.CASE.toString())) {
                     return;
                 }
 
-                PredicateInfo predicateInfo = PredicateInfo.builder().setOperatorName(operatorName).setOperatorKind(operatorKind.toString())
-                        .setOwnerTable(ownerTable).setFieldName(fieldName).setCondition(content).build();
+                PredicateInfo predicateInfo =
+                        PredicateInfo.builder()
+                                .setOperatorName(operatorName)
+                                .setOperatorKind(operatorKind.toString())
+                                .setOwnerTable(ownerTable)
+                                .setFieldName(fieldName)
+                                .setCondition(condition)
+                                .build();
                 predicatesInfoList.add(predicateInfo);
             }
         }
     }
 
+    /**
+     * 去掉_UTF16前缀，只获取字符本身。要不然中文编码会有问题。例如 _UTF16'甲' 去掉后返回结果为 '甲'。
+     *
+     * @param stringLiteral
+     * @return
+     */
+    private String removeCoding(SqlCharStringLiteral stringLiteral) {
+        return QUOTE + stringLiteral.getNlsString().getValue() + QUOTE;
+    }
 }

core/src/test/java/com/dtstack/flink/sql/side/SidePredicatesParserTest.java

@@ -2,10 +2,13 @@
 
 import com.google.common.collect.Lists;
 import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlCharStringLiteral;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.util.NlsString;
 import org.apache.flink.calcite.shaded.com.google.common.collect.ImmutableList;
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.lang.reflect.Method;
@@ -17,10 +20,9 @@
 public class SidePredicatesParserTest {
 
     private SidePredicatesParser sidePredicatesParser = new SidePredicatesParser();
-    
 
     @Test
-    public void fillPredicateInfoToList() throws Exception {
+    public void testFillPredicateInfoToList() throws Exception {
         Method method = SidePredicatesParser.class.getDeclaredMethod("fillPredicateInfoToList", SqlBasicCall.class, List.class, String.class, SqlKind.class, int.class, int.class);
         method.setAccessible(true);
         SqlBasicCall sqlBasicCall = mock(SqlBasicCall.class);
@@ -33,4 +35,15 @@ public void fillPredicateInfoToList() throws Exception {
         method.invoke(sidePredicatesParser, sqlBasicCall, Lists.newArrayList(), "a", SqlKind.IN, 0, 0);
     }
 
+    @Test
+    public void testRemoveCoding() throws Exception {
+        Method method = SidePredicatesParser.class.getDeclaredMethod("removeCoding", SqlCharStringLiteral.class);
+        method.setAccessible(true);
+        SqlCharStringLiteral stringLiteral = mock(SqlCharStringLiteral.class);
+        when(stringLiteral.getNlsString()).thenReturn(new NlsString("甲", "UTF16", null));
+
+        String val = (String) method.invoke(sidePredicatesParser, stringLiteral);
+        Assert.assertEquals("'甲'", val);
+    }
+
 }