fix 修复hbase-side 别名异常

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -56,7 +56,6 @@
 import org.apache.commons.io.Charsets;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -65,7 +64,6 @@
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.table.sinks.TableSink;
-import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -81,9 +79,10 @@
 import java.util.Set;
 
 /**
- *  任务执行时的流程方法
+ * 任务执行时的流程方法
  * Date: 2020/2/17
  * Company: www.dtstack.com
+ *
  * @author maqi
  */
 public class ExecuteProcessHelper {
@@ -126,11 +125,11 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
                 .setConfProp(confProperties)
                 .setJarUrlList(jarUrlList)
                 .build();
-
     }
 
     /**
-     *   非local模式或者shipfile部署模式，remoteSqlPluginPath必填
+     * 非local模式或者shipfile部署模式，remoteSqlPluginPath必填
+     *
      * @param remoteSqlPluginPath
      * @param deployMode
      * @param pluginLoadMode
@@ -147,7 +146,7 @@ public static boolean checkRemoteSqlPluginPath(String remoteSqlPluginPath, Strin
 
     public static StreamExecutionEnvironment getStreamExecution(ParamsInfo paramsInfo) throws Exception {
         StreamExecutionEnvironment env = ExecuteProcessHelper.getStreamExeEnv(paramsInfo.getConfProp(), paramsInfo.getDeployMode());
-        StreamTableEnvironment tableEnv = getStreamTableEnv(env,paramsInfo.getConfProp());
+        StreamTableEnvironment tableEnv = getStreamTableEnv(env, paramsInfo.getConfProp());
 
         SqlParser.setLocalSqlPluginRoot(paramsInfo.getLocalSqlPluginPath());
         SqlTree sqlTree = SqlParser.parseSql(paramsInfo.getSql());
@@ -188,7 +187,7 @@ public static List<URL> getExternalJarUrls(String addJarListStr) throws java.io.
 
     private static void sqlTranslation(String localSqlPluginPath,
                                        StreamTableEnvironment tableEnv,
-                                       SqlTree sqlTree,Map<String, AbstractSideTableInfo> sideTableMap,
+                                       SqlTree sqlTree, Map<String, AbstractSideTableInfo> sideTableMap,
                                        Map<String, Table> registerTableCache) throws Exception {
 
         SideSqlExec sideSqlExec = new SideSqlExec();
@@ -251,13 +250,14 @@ public static void registerUserDefinedFunction(SqlTree sqlTree, List<URL> jarUrl
     }
 
     /**
-     *    向Flink注册源表和结果表，返回执行时插件包的全路径
+     * 向Flink注册源表和结果表，返回执行时插件包的全路径
+     *
      * @param sqlTree
      * @param env
      * @param tableEnv
      * @param localSqlPluginPath
      * @param remoteSqlPluginPath
-     * @param pluginLoadMode   插件加载模式 classpath or shipfile
+     * @param pluginLoadMode      插件加载模式 classpath or shipfile
      * @param sideTableMap
      * @param registerTableCache
      * @return
@@ -322,7 +322,8 @@ public static Set<URL> registerTable(SqlTree sqlTree, StreamExecutionEnvironment
     }
 
     /**
-     *   perjob模式将job依赖的插件包路径存储到cacheFile，在外围将插件包路径传递给jobgraph
+     * perjob模式将job依赖的插件包路径存储到cacheFile，在外围将插件包路径传递给jobgraph
+     *
      * @param env
      * @param classPathSet
      */

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

@@ -16,8 +16,6 @@
  * limitations under the License.
  */
 
- 
-
 package com.dtstack.flink.sql.side.hbase;
 
 import com.dtstack.flink.sql.enums.ECacheContentType;
@@ -42,9 +40,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -73,9 +69,9 @@ public class HbaseAsyncReqRow extends BaseAsyncReqRow {
 
     private transient AbstractRowKeyModeDealer rowKeyMode;
 
-    private String tableName;
+    private final String tableName;
 
-    private String[] colNames;
+    private final String[] colNames;
 
     public HbaseAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(new HbaseAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
@@ -131,7 +127,7 @@ public void asyncInvoke(Tuple2<Boolean,Row> input, ResultFuture<Tuple2<Boolean,R
                 dealMissKey(inputCopy, resultFuture);
                 return;
             }
-            refData.put(sideInfo.getEqualFieldList().get(i), equalObj);
+            refData.put(getAliasFieldsName(sideInfo.getEqualFieldList().get(i), sideInfo.getSideTableInfo().getPhysicalFields()), equalObj);
         }
 
         String rowKeyStr = ((HbaseAsyncSideInfo)sideInfo).getRowKeyBuilder().getRowKey(refData);
@@ -189,6 +185,23 @@ public Row fillData(Row input, Object sideInput){
         return row;
     }
 
+    // 根据实际字段名获得对应的别名
+    public String getAliasFieldsName(String realFieldName, Map<String, String> physicalFields) {
+        Collection<String> values = physicalFields.values();
+        Set<String> keySet = physicalFields.keySet();
+        if (!values.contains(realFieldName)) {
+            // TODO Error ? or Warn ?
+            LOG.warn(realFieldName + "不存在别名");
+        } else {
+            for (String key : keySet) {
+                if (physicalFields.get(key).equals(realFieldName)) {
+                    return key;
+                }
+            }
+        }
+        return realFieldName;
+    }
+
     @Override
     public void close() throws Exception {
         super.close();

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

@@ -101,7 +101,7 @@ public static String[] splitIgnoreQuotaBrackets(String str, String delimiter){
     public ReplaceInfo getReplaceInfo(String field){
 
         field = field.trim();
-        if(field.length() <= 2){
+        if(field.length() <= 0){
             throw new RuntimeException(field + " \n" +
                     "Format defined exceptions");
         }