hbase 解析PRIMARY KEY优化

Author: dapeng

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

@@ -20,9 +20,10 @@
 
 package com.dtstack.flink.sql.side.hbase;
 
-import com.dtstack.flink.sql.side.hbase.enums.EReplaceType;
-
 import java.io.Serializable;
+import java.util.List;
+
+import com.dtstack.flink.sql.side.hbase.enums.EReplaceType;
 
 /**
  * Reason:
@@ -39,6 +40,8 @@ public class ReplaceInfo implements Serializable {
     private EReplaceType type;
 
     private String param;
+    
+    private List<ReplaceInfo> subReplaceInfos;
 
     public ReplaceInfo(EReplaceType type){
         this.type = type;
@@ -59,4 +62,12 @@ public String getParam() {
     public void setParam(String param) {
         this.param = param;
     }
+    
+    public List<ReplaceInfo> getSubReplaceInfos() {
+        return subReplaceInfos;
+    }
+
+    public void setSubReplaceInfos(List<ReplaceInfo> subReplaceInfos) {
+        this.subReplaceInfos = subReplaceInfos;
+    }
 }

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

@@ -21,7 +21,9 @@
 package com.dtstack.flink.sql.side.hbase;
 
 import com.dtstack.flink.sql.side.hbase.enums.EReplaceType;
+import com.dtstack.flink.sql.util.MD5Utils;
 import com.google.common.collect.Lists;
+import org.apache.commons.collections.CollectionUtils;
 
 import java.io.Serializable;
 import java.util.List;
@@ -42,31 +44,10 @@ public class RowKeyBuilder implements Serializable{
 
     private static Pattern Md5Operator = Pattern.compile("(?i)^md5\\(\\s*(.*)\\s*\\)$");
 
-    private List<List<ReplaceInfo>> operatorChain = Lists.newArrayList();
+    private List<ReplaceInfo> operatorChain = Lists.newArrayList();
 
     public void init(String rowKeyTempl){
-
-        String[] strArr = splitIgnoreQuotaBrackets(rowKeyTempl, "\\+");
-
-        for(String infoAlias : strArr){
-            infoAlias = infoAlias.trim();
-            Matcher matcher = Md5Operator.matcher(infoAlias);
-            String fieldCols = null;
-            if(matcher.find()){
-                fieldCols = matcher.group(1);
-            }else{
-                fieldCols = infoAlias;
-            }
-
-            String[] fieldArr = fieldCols.split("\\+");
-            List<ReplaceInfo> fieldList = Lists.newArrayList();
-            for(String oneField : fieldArr){
-                ReplaceInfo replaceInfo =  getReplaceInfo(oneField);
-                fieldList.add(replaceInfo);
-            }
-
-            operatorChain.add(fieldList);
-        }
+    	operatorChain.addAll(makeFormula(rowKeyTempl));
     }
 
     /**
@@ -75,18 +56,15 @@ public void init(String rowKeyTempl){
      * @return
      */
     public String getRowKey(Map<String, Object> refData){
-
-        StringBuilder sb = new StringBuilder("");
-        for(List<ReplaceInfo> fieldList : operatorChain){
-            sb.append(buildStr(fieldList, refData));
-        }
-
-        return sb.toString();
+    	return buildStr(operatorChain, refData);
     }
 
 
 
     private String buildStr(List<ReplaceInfo> fieldList, Map<String, Object> refData){
+        if(CollectionUtils.isEmpty(fieldList)){
+            return "";
+        }
         StringBuffer sb = new StringBuffer("");
         for(ReplaceInfo replaceInfo : fieldList){
 
@@ -95,6 +73,10 @@ private String buildStr(List<ReplaceInfo> fieldList, Map<String, Object> refData
                 continue;
             }
 
+            if(replaceInfo.getType() == EReplaceType.FUNC){
+                sb.append(MD5Utils.getMD5String(buildStr(replaceInfo.getSubReplaceInfos(), refData)));
+                continue;
+            }
             String replaceName = replaceInfo.getParam();
             if(!refData.containsKey(replaceName)){
                 throw new RuntimeException(String.format("build rowKey with field %s which value not found.", replaceName));
@@ -137,4 +119,21 @@ public ReplaceInfo getReplaceInfo(String field){
         return replaceInfo;
     }
 
+    private List<ReplaceInfo> makeFormula(String formula){
+        if(formula == null || formula.length() <= 0){
+            Lists.newArrayList();
+        }
+        List<ReplaceInfo> result = Lists.newArrayList();
+        for(String meta: splitIgnoreQuotaBrackets(formula, "\\+")){
+            Matcher matcher = Md5Operator.matcher(meta.trim());
+            if(matcher.find()){
+                ReplaceInfo replaceInfo = new ReplaceInfo(EReplaceType.FUNC);
+                replaceInfo.setSubReplaceInfos(makeFormula(matcher.group(1)));
+                result.add(replaceInfo);
+            } else {
+                result.add(getReplaceInfo(meta));
+            }
+        }
+        return result;
+    }
 }

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

@@ -29,5 +29,6 @@
  */
 public enum EReplaceType {
     PARAM,
+    FUNC,
     CONSTANT;
 }