[Refactor] query

oceanbase · Mar 21, 2024 · cdec3de · cdec3de
1 parent 3501f55
commit cdec3de
Show file tree

Hide file tree

Showing 28 changed files with 562 additions and 2,109 deletions.
diff --git a/src/main/java/com/alipay/oceanbase/rpc/ObClusterTableAsyncQuery.java b/src/main/java/com/alipay/oceanbase/rpc/ObClusterTableAsyncQuery.java
diff --git a/src/main/java/com/alipay/oceanbase/rpc/ObClusterTableQuery.java b/src/main/java/com/alipay/oceanbase/rpc/ObClusterTableQuery.java
@@ -17,17 +17,16 @@
 
 package com.alipay.oceanbase.rpc;
 
-import com.alipay.oceanbase.rpc.location.model.partition.ObPair;
 import com.alipay.oceanbase.rpc.mutation.Row;
 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.ObHTableFilter;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObTableQuery;
+import com.alipay.oceanbase.rpc.stream.ObTableClientQueryAsyncStreamResult;
 import com.alipay.oceanbase.rpc.stream.ObTableClientQueryStreamResult;
 import com.alipay.oceanbase.rpc.stream.QueryResultSet;
 import com.alipay.oceanbase.rpc.table.AbstractTableQuery;
 import com.alipay.oceanbase.rpc.table.ObTableClientQueryImpl;
-import com.alipay.oceanbase.rpc.table.ObTableParam;
 import com.alipay.oceanbase.rpc.table.api.TableQuery;
 
 import java.util.List;
@@ -85,14 +84,12 @@ public QueryResultSet execute() throws Exception {
         return tableClientQuery.execute();
     }
 
+    /*
+     * Execute.
+     */
     @Override
-    public QueryResultSet executeInit(ObPair<Long, ObTableParam> entry) throws Exception {
-        throw new IllegalArgumentException("not support executeInit");
-    }
-
-    @Override
-    public QueryResultSet executeNext(ObPair<Long, ObTableParam> entry) throws Exception {
-        throw new IllegalArgumentException("not support executeNext");
+    public QueryResultSet asyncExecute() throws Exception {
+        return tableClientQuery.asyncExecute();
     }
 
     /*
@@ -102,6 +99,13 @@ public ObTableClientQueryStreamResult executeInternal() throws Exception {
         return tableClientQuery.executeInternal();
     }
 
+    /*
+     * Async execute internal.
+     */
+    public ObTableClientQueryAsyncStreamResult asyncExecuteInternal() throws Exception {
+        return tableClientQuery.asyncExecuteInternal();
+    }
+
     /*
      * Select.
      */

diff --git a/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java b/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java
@@ -21,7 +21,6 @@
 import com.alipay.oceanbase.rpc.checkandmutate.CheckAndInsUp;
 import com.alipay.oceanbase.rpc.constant.Constants;
 import com.alipay.oceanbase.rpc.exception.*;
-import com.alipay.oceanbase.rpc.batch.QueryByBatch;
 import com.alipay.oceanbase.rpc.location.LocationUtil;
 import com.alipay.oceanbase.rpc.filter.ObTableFilter;
 import com.alipay.oceanbase.rpc.location.model.*;
@@ -1820,17 +1819,6 @@ public TableQuery query(String tableName) {
         return new ObClusterTableQuery(tableQuery);
     }
 
-    @Override
-    public TableQuery queryByBatchV2(String tableName) {
-        ObTableClientQueryAsyncImpl queryAsync = new ObTableClientQueryAsyncImpl(tableName, this);
-        return new ObClusterTableAsyncQuery(queryAsync);
-    }
-
-    @Override
-    public TableQuery queryByBatch(String tableName) throws Exception {
-        return new QueryByBatch(query(tableName));
-    }
-
     /**
      * Batch.
      */
@@ -2707,12 +2695,10 @@ && isTableGroupName(tableName)) {
                     && isTableGroupName(tableName)) {
                 tableName = tryGetTableNameFromTableGroupCache(tableName, false);
             }
-            ObTableClientQueryAsyncImpl tableClientQueryAsync = new ObTableClientQueryAsyncImpl(
-                tableName, ((ObTableQueryAsyncRequest) request)
-                    .getObTableQueryRequest().getTableQuery(), this);
-            tableClientQueryAsync.setEntityType(request.getEntityType());
-            return new ObClusterTableAsyncQuery(tableClientQueryAsync)
-                .executeInternal(((ObTableQueryAsyncRequest) request).getQueryType());
+            ObTableClientQueryImpl tableQuery = new ObTableClientQueryImpl(
+                    tableName, ((ObTableQueryAsyncRequest) request).getObTableQueryRequest().getTableQuery(), this);
+            tableQuery.setEntityType(request.getEntityType());
+            return new ObClusterTableQuery(tableQuery).asyncExecuteInternal();
         } else if (request instanceof ObTableBatchOperationRequest) {
             ObTableClientBatchOpsImpl batchOps = new ObTableClientBatchOpsImpl(
                 request.getTableName(),