add hbase plugin constant join support

Author: a49a

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

@@ -172,6 +172,9 @@ private void evalEquation(SqlIdentifier left, SqlIdentifier right, String sideTa
      */
     private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier) {
         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();

hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllSideInfo.java

@@ -46,7 +46,7 @@ public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInf
             throw new RuntimeException("Primary key dimension table must be filled");
         }
 
-        rowKeyBuilder.init(sideTableInfo.getPrimaryKeys().get(0));
+        rowKeyBuilder.init(sideTableInfo.getPrimaryKeys().get(0), sideTableInfo);
 
         String sideTableName = joinInfo.getSideTableName();
         SqlNode conditionNode = joinInfo.getCondition();

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncSideInfo.java

@@ -41,7 +41,7 @@ public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInf
         }
 
         HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
-        rowKeyBuilder.init(sideTableInfo.getPrimaryKeys().get(0));
+        rowKeyBuilder.init(sideTableInfo.getPrimaryKeys().get(0), sideTableInfo);
 
         colRefType = Maps.newHashMap();
         for(int i=0; i<hbaseSideTableInfo.getColumnRealNames().length; i++){

hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/RowKeyBuilder.java

@@ -20,7 +20,10 @@
 
 package com.dtstack.flink.sql.side.hbase;
 
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
+import com.dtstack.flink.sql.side.PredicateInfo;
 import com.dtstack.flink.sql.side.hbase.enums.EReplaceType;
+import com.dtstack.flink.sql.util.DtStringUtil;
 import com.dtstack.flink.sql.util.MD5Utils;
 import com.google.common.collect.Lists;
 import org.apache.commons.collections.CollectionUtils;
@@ -46,20 +49,43 @@ public class RowKeyBuilder implements Serializable{
 
     private List<ReplaceInfo> operatorChain = Lists.newArrayList();
 
-    public void init(String rowKeyTempl){
+    private AbstractSideTableInfo sideTableInfo;
+
+    public void init(String rowKeyTempl) {
     	operatorChain.addAll(makeFormula(rowKeyTempl));
     }
 
+    public void init(String rowKeyTempl, AbstractSideTableInfo sideTableInfo) {
+        this.init(rowKeyTempl);
+        this.sideTableInfo = sideTableInfo;
+    }
+
     /**
      *
      * @param refData
      * @return
      */
-    public String getRowKey(Map<String, Object> refData){
+    public String getRowKey(Map<String, Object> refData) {
+        addConstant(refData);
     	return buildStr(operatorChain, refData);
     }
 
-
+    /**
+     * add constant join fields
+     * @param rowkeyMap
+     */
+    private void addConstant(Map<String, Object> rowkeyMap) {
+        List<PredicateInfo> predicateInfos = sideTableInfo.getPredicateInfoes();
+        final String name = sideTableInfo.getName();
+        for (PredicateInfo info : predicateInfos) {
+            if (info.getOwnerTable().equals(name) &&
+                info.getOperatorName().equals("=")) {
+                String condition = info.getCondition();
+                String conditionWithoutQuota = condition.replaceAll("['\"]", "");
+                rowkeyMap.put(info.getFieldName(), conditionWithoutQuota);
+            }
+        }
+    }
 
     private String buildStr(List<ReplaceInfo> fieldList, Map<String, Object> refData){
         if(CollectionUtils.isEmpty(fieldList)){

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -18,8 +18,7 @@
 
 package com.dtstack.flink.sql.side.redis;
 
-import com.dtstack.flink.sql.side.AbstractSideTableInfo;
-import com.dtstack.flink.sql.side.BaseAsyncReqRow;
+import com.dtstack.flink.sql.side.*;
 import io.lettuce.core.KeyValue;
 import io.lettuce.core.api.async.RedisStringAsyncCommands;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -29,9 +28,6 @@
 import org.apache.flink.types.Row;
 
 import com.dtstack.flink.sql.enums.ECacheContentType;
-import com.dtstack.flink.sql.side.CacheMissVal;
-import com.dtstack.flink.sql.side.FieldInfo;
-import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.cache.CacheObj;
 import com.dtstack.flink.sql.side.redis.enums.RedisType;
 import com.dtstack.flink.sql.side.redis.table.RedisSideReqRow;
@@ -154,6 +150,7 @@ public void accept(Map<String, String> values) {
         });
     }
 
+    // TODO 升级对常量JOIN的支持
     @Override
     public String buildCacheKey(Map<String, Object> refData) {
         StringBuilder keyBuilder = new StringBuilder(redisSideTableInfo.getTableName());