[hotfix-33089][hbase-side]Compatible with composite primary keys.

Author: HiLany

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableParser.java

@@ -19,13 +19,16 @@
 
 package com.dtstack.flink.sql.table;
 
+import com.dtstack.flink.sql.parser.SqlParser;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -45,6 +48,8 @@
 
 public abstract class AbstractTableParser {
 
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractTableParser.class);
+
     private static final String PRIMARY_KEY = "primaryKey";
     private static final String NEST_JSON_FIELD_KEY = "nestFieldKey";
     private static final String CHAR_TYPE_NO_LENGTH = "CHAR";
@@ -117,7 +122,13 @@ public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
         if (tableInfo instanceof AbstractSideTableInfo) {
             tableInfo.getPrimaryKeys().stream()
                     .filter(pk -> !tableInfo.getFieldList().contains(pk))
-                    .forEach(pk -> handleKeyNotHaveAlias(String.format("%s varchar", pk), tableInfo));
+                    .forEach(pk -> {
+                        try {
+                            handleKeyNotHaveAlias(String.format("%s varchar", pk), tableInfo);
+                        } catch (Exception e) {
+                            LOG.error(String.format("Handle primary key failed. Reason: %s", e.getMessage()));
+                        }
+                    });
         }
 
         tableInfo.finish();

docs/plugin/hbaseSide.md

@@ -146,7 +146,7 @@ CREATE TABLE MyResult(
  CREATE TABLE sideTable (  
         wtz:message varchar as message,
         wtz:info varchar as info , 
-        PRIMARY KEY (rowkey),
+        PRIMARY KEY (md5(rowkey1) + rowkey2 + 'test'),
          PERIOD FOR SYSTEM_TIME
 ) WITH (
         type = 'hbase',
@@ -174,7 +174,7 @@ into
         MyTable a
     left join
         sideTable b
-            on a.id=b.rowkey;
+            on a.id=b.rowkey1 and a.name = b.rowkey2;
 ```
 ### kerberos维表示例
 ```