维表all的类型

Author: dapeng

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

@@ -25,6 +25,7 @@
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.hbase.table.HbaseSideTableInfo;
+import com.dtstack.flink.sql.side.hbase.utils.HbaseUtils;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.map.HashedMap;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -165,6 +166,7 @@ public void flatMap(CRow input, Collector<CRow> out) throws Exception {
 
     private void loadData(Map<String, Map<String, Object>> tmpCache) throws SQLException {
         AbstractSideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
+        Map<String, String> colRefType = ((HbaseAllSideInfo)sideInfo).getColRefType();
         HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
         Configuration conf = new Configuration();
         conf.set("hbase.zookeeper.quorum", hbaseSideTableInfo.getHost());
@@ -177,14 +179,12 @@ private void loadData(Map<String, Map<String, Object>> tmpCache) throws SQLExcep
             resultScanner = table.getScanner(new Scan());
             for (Result r : resultScanner) {
                 Map<String, Object> kv = new HashedMap();
-                for (Cell cell : r.listCells())
-                {
+                for (Cell cell : r.listCells()) {
                     String family = Bytes.toString(CellUtil.cloneFamily(cell));
                     String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
-                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                     StringBuilder key = new StringBuilder();
                     key.append(family).append(":").append(qualifier);
-
+                    Object value = HbaseUtils.convertByte(CellUtil.cloneValue(cell), colRefType.get(key.toString()));
                     kv.put(aliasNameInversion.get(key.toString()), value);
                 }
                 tmpCache.put(new String(r.getRow()), kv);

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

@@ -24,17 +24,22 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
+import com.dtstack.flink.sql.side.hbase.table.HbaseSideTableInfo;
 import com.dtstack.flink.sql.util.ParseUtils;
+import com.google.common.collect.Maps;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Lists;
 
 import java.util.List;
+import java.util.Map;
 
 public class HbaseAllSideInfo extends BaseSideInfo {
 
     private RowKeyBuilder rowKeyBuilder;
 
+    private Map<String, String> colRefType;
+
     public HbaseAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
@@ -48,6 +53,14 @@ public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInf
 
         rowKeyBuilder.init(sideTableInfo.getPrimaryKeys().get(0));
 
+        HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
+        colRefType = Maps.newHashMap();
+        for(int i=0; i<hbaseSideTableInfo.getColumnRealNames().length; i++){
+            String realColName = hbaseSideTableInfo.getColumnRealNames()[i];
+            String colType = hbaseSideTableInfo.getFieldTypes()[i];
+            colRefType.put(realColName, colType);
+        }
+
         String sideTableName = joinInfo.getSideTableName();
         SqlNode conditionNode = joinInfo.getCondition();
 
@@ -67,4 +80,8 @@ public void setRowKeyBuilder(RowKeyBuilder rowKeyBuilder) {
         this.rowKeyBuilder = rowKeyBuilder;
     }
 
+    public Map<String, String> getColRefType() {
+        return colRefType;
+    }
+
 }