feature: hbase compatible 2.x (#242)

* support hbase scan renewLease (#211)

* hbase tablename bug fix (#237)

* hbase empty family scan error (#239)

* hbase tablename bug fix

* hbase empty family scan error

---------

Co-authored-by: stuBirdFly <[email protected]>

Author: 2 people

src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java

@@ -3007,16 +3007,13 @@ public ObPayload execute(final ObTableAbstractOperationRequest request) throws E
         } else if (request instanceof ObTableQueryRequest) {
             // TableGroup -> TableName
             String tableName = request.getTableName();
-            tableName = getPhyTableNameFromTableGroup(((ObTableQueryRequest) request), tableName);
             ObTableClientQueryImpl tableQuery = new ObTableClientQueryImpl(tableName,
                 ((ObTableQueryRequest) request).getTableQuery(), this);
             tableQuery.setEntityType(request.getEntityType());
             return new ObClusterTableQuery(tableQuery).executeInternal();
         } else if (request instanceof ObTableQueryAsyncRequest) {
             // TableGroup -> TableName
             String tableName = request.getTableName();
-            tableName = getPhyTableNameFromTableGroup(
-                ((ObTableQueryAsyncRequest) request).getObTableQueryRequest(), tableName);
             ObTableClientQueryImpl tableQuery = new ObTableClientQueryImpl(tableName,
                 ((ObTableQueryAsyncRequest) request).getObTableQueryRequest().getTableQuery(), this);
             tableQuery.setEntityType(request.getEntityType());

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java

@@ -274,6 +274,13 @@ protected ObPayload commonExecute(ObTableClient client, Logger logger,
         return result;
     }
 
+    /*
+     * RenewLease.
+     */
+    public void renewLease() throws Exception {
+        throw new IllegalStateException("renew only support stream query");
+    }
+
     /*
      * Next.
      */

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/syncquery/ObQueryOperationType.java

@@ -21,7 +21,7 @@
 import java.util.Map;
 
 public enum ObQueryOperationType {
-    QUERY_START(0), QUERY_NEXT(1), QUERY_END(2);
+    QUERY_START(0), QUERY_NEXT(1), QUERY_END(2), QUERY_RENEW(3);
 
     private int                                       value;
     private static Map<Integer, ObQueryOperationType> map = new HashMap<Integer, ObQueryOperationType>();

src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java

@@ -182,6 +182,32 @@ protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery ta
         return buildPartitions(client, tableQuery, tableName);
     }
 
+    // This function is designed for HBase-type requests.
+    // It is used to extend the session duration of a scan
+    @Override
+    public void renewLease() throws Exception {
+        if (!isEnd() && !expectant.isEmpty()) {
+            Iterator<Map.Entry<Long, ObPair<Long, ObTableParam>>> it = expectant.entrySet()
+                    .iterator();
+            Map.Entry<Long, ObPair<Long, ObTableParam>> lastEntry = it.next();
+            ObPair<Long, ObTableParam> partIdWithObTable = lastEntry.getValue();
+            // try access new partition, async will not remove useless expectant
+            ObTableParam obTableParam = partIdWithObTable.getRight();
+            ObTableQueryRequest queryRequest = asyncRequest.getObTableQueryRequest();
+
+            // refresh request info
+            queryRequest.setPartitionId(obTableParam.getPartitionId());
+            queryRequest.setTableId(obTableParam.getTableId());
+
+            // refresh async query request
+            asyncRequest.setQueryType(ObQueryOperationType.QUERY_RENEW);
+            asyncRequest.setQuerySessionId(sessionId);
+            executeAsync(partIdWithObTable, asyncRequest);
+        } else {
+            throw new ObTableException("query end or expectant is null");
+        }
+    }
+
     @Override
     public boolean next() throws Exception {
         checkStatus();

src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientQueryImpl.java

@@ -24,6 +24,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.ObPayload;
 import com.alipay.oceanbase.rpc.protocol.payload.ResultCodes;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObRowKey;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableEntityType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.aggregation.ObTableAggregationType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.*;
 import com.alipay.oceanbase.rpc.stream.ObTableClientQueryAsyncStreamResult;
@@ -192,14 +193,6 @@ private AbstractQueryStreamResult commonExecute(InitQueryResultCallback<Abstract
             }
         }
 
-        // set correct table group name for hbase
-        if (tableQuery.isHbaseQuery()
-            && obTableClient.getTableGroupInverted().containsKey(tableName)
-            && tableName.equalsIgnoreCase(obTableClient.getTableGroupCache().get(
-                obTableClient.getTableGroupInverted().get(tableName)))) {
-            tableName = obTableClient.getTableGroupInverted().get(tableName);
-        }
-
         // init query stream result
         AbstractQueryStreamResult streamResult = callable.execute();
 
@@ -267,7 +260,9 @@ public Map<Long, ObPair<Long, ObTableParam>> initPartitions(ObTableQuery tableQu
                 end[i] = endKey.getObj(i).isMinObj() || endKey.getObj(i).isMaxObj() ?
                         endKey.getObj(i) : endKey.getObj(i).getValue();
             }
-
+            if (this.entityType == ObTableEntityType.HKV && obTableClient.isTableGroupName(tableName)) {
+                indexTableName = obTableClient.tryGetTableNameFromTableGroupCache(tableName, false);
+            }
             ObBorderFlag borderFlag = range.getBorderFlag();
             List<ObPair<Long, ObTableParam>> pairs = this.obTableClient.getTables(indexTableName,
                     tableQuery, start, borderFlag.isInclusiveStart(), end, borderFlag.isInclusiveEnd(),