Merge pull request #143 from oceanbase/getscanners_2.0

[CP] Patch getscanners to hbase_2.x

Author: stuBirdFly

src/main/java/com/alipay/oceanbase/hbase/OHTable.java

@@ -27,6 +27,7 @@
 import com.alipay.oceanbase.rpc.ObTableClient;
 import com.alipay.oceanbase.rpc.exception.ObTableException;
 import com.alipay.oceanbase.rpc.exception.ObTableUnexpectedException;
+import com.alipay.oceanbase.rpc.location.model.partition.Partition;
 import com.alipay.oceanbase.rpc.mutation.BatchOperation;
 import com.alipay.oceanbase.rpc.mutation.result.BatchOperationResult;
 import com.alipay.oceanbase.rpc.mutation.result.MutationResult;
@@ -1086,6 +1087,105 @@ public ResultScanner call() throws IOException {
         return executeServerCallable(serverCallable);
     }
 
+    public List<ResultScanner> getScanners(final Scan scan) throws IOException {
+
+        if (scan.getFamilyMap().keySet().isEmpty()) {
+            // check nothing, use table group;
+        } else {
+            checkFamilyViolation(scan.getFamilyMap().keySet(), false);
+        }
+
+        //be careful about the packet size ,may the packet exceed the max result size ,leading to error
+        ServerCallable<List<ResultScanner>> serverCallable = new ServerCallable<List<ResultScanner>>(
+                configuration, obTableClient, tableNameString, scan.getStartRow(), scan.getStopRow(),
+                operationTimeout) {
+            public List<ResultScanner> call() throws IOException {
+                byte[] family = new byte[] {};
+                List<ResultScanner> resultScanners = new ArrayList<ResultScanner>();
+                ObTableClientQueryAsyncStreamResult clientQueryAsyncStreamResult;
+                ObTableQueryAsyncRequest request;
+                ObTableQuery obTableQuery;
+                ObHTableFilter filter;
+                try {
+                    if (scan.getFamilyMap().keySet() == null
+                            || scan.getFamilyMap().keySet().isEmpty()
+                            || scan.getFamilyMap().size() > 1) {
+                        // In a Scan operation where the family map is greater than 1 or equal to 0,
+                        // we handle this by appending the column family to the qualifier on the client side.
+                        // The server can then use this information to filter the appropriate column families and qualifiers.
+                        if (!scan.getColumnFamilyTimeRange().isEmpty()) {
+                            throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now");
+                        }
+                        NavigableSet<byte[]> columnFilters = new TreeSet<>(Bytes.BYTES_COMPARATOR);
+                        processColumnFilters(columnFilters, scan.getFamilyMap());
+                        filter = buildObHTableFilter(scan.getFilter(), scan.getTimeRange(),
+                                scan.getMaxVersions(), columnFilters);
+                        obTableQuery = buildObTableQuery(filter, scan);
+
+                        request = buildObTableQueryAsyncRequest(obTableQuery,
+                                getTargetTableName(tableNameString));
+                        String phyTableName = obTableClient.getPhyTableNameFromTableGroup(
+                                request.getObTableQueryRequest(), tableNameString);
+                        List<Partition> partitions = obTableClient.getPartition(phyTableName, false);
+                        for (Partition partition : partitions) {
+                            request.getObTableQueryRequest().setTableQueryPartId(
+                                    partition.getPartId());
+                            clientQueryAsyncStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient
+                                    .execute(request);
+                            ClientStreamScanner clientScanner = new ClientStreamScanner(
+                                    clientQueryAsyncStreamResult, tableNameString, scan, true);
+                            resultScanners.add(clientScanner);
+                        }
+                        return resultScanners;
+                    } else {
+                        for (Map.Entry<byte[], NavigableSet<byte[]>> entry : scan.getFamilyMap()
+                                .entrySet()) {
+                            family = entry.getKey();
+                            if (!scan.getColumnFamilyTimeRange().isEmpty()) {
+                                Map<byte[], TimeRange> colFamTimeRangeMap = scan.getColumnFamilyTimeRange();
+                                if (colFamTimeRangeMap.size() > 1) {
+                                    throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now");
+                                } else if (colFamTimeRangeMap.get(family) == null) {
+                                    throw new IllegalArgumentException("Scan family is not matched in ColumnFamilyTimeRange");
+                                } else {
+                                    TimeRange tr = colFamTimeRangeMap.get(family);
+                                    scan.setTimeRange(tr.getMin(), tr.getMax());
+                                }
+                            }
+                            filter = buildObHTableFilter(scan.getFilter(), scan.getTimeRange(),
+                                    scan.getMaxVersions(), entry.getValue());
+                            obTableQuery = buildObTableQuery(filter, scan);
+
+                            String targetTableName = getTargetTableName(tableNameString, Bytes.toString(family),
+                                    configuration);
+                            request = buildObTableQueryAsyncRequest(obTableQuery, targetTableName);
+                            List<Partition> partitions = obTableClient
+                                    .getPartition(targetTableName, false);
+                            for (Partition partition : partitions) {
+                                request.getObTableQueryRequest().setTableQueryPartId(
+                                        partition.getPartId());
+                                clientQueryAsyncStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient
+                                        .execute(request);
+                                ClientStreamScanner clientScanner = new ClientStreamScanner(
+                                        clientQueryAsyncStreamResult, tableNameString, scan, false);
+                                resultScanners.add(clientScanner);
+                            }
+                            return resultScanners;
+                        }
+                    }
+                } catch (Exception e) {
+                    logger.error(LCD.convert("01-00003"), tableNameString, Bytes.toString(family),
+                            e);
+                    throw new IOException("scan table:" + tableNameString + " family "
+                            + Bytes.toString(family) + " error.", e);
+                }
+
+                throw new IOException("scan table:" + tableNameString + "has no family");
+            }
+        };
+        return executeServerCallable(serverCallable);
+    }
+
     @Override
     public ResultScanner getScanner(final byte[] family) throws IOException {
         Scan scan = new Scan();

src/main/java/com/alipay/oceanbase/hbase/OHTableClient.java

@@ -290,6 +290,11 @@ public ResultScanner getScanner(byte[] family, byte[] qualifier) throws IOExcept
         return ohTable.getScanner(family, qualifier);
     }
 
+    public List<ResultScanner> getScanners(Scan scan) throws IOException {
+        checkStatus();
+        return ohTable.getScanners(scan);
+    }
+
     @Override
     public void put(Put put) throws IOException {
         checkStatus();

src/main/java/com/alipay/oceanbase/hbase/OHTablePool.java

@@ -769,6 +769,16 @@ public ResultScanner getScanner(byte[] family, byte[] qualifier) throws IOExcept
             return table.getScanner(family, qualifier);
         }
 
+        public List<ResultScanner> getScanners(Scan scan) throws IOException {
+            if (table instanceof OHTableClient) {
+                return ((OHTableClient) table).getScanners(scan);
+            } else if (table instanceof OHTable) {
+                return ((OHTable) table).getScanners(scan);
+            }
+
+            throw new IllegalArgumentException("just support for OHTable and OHTableClient");
+        }
+
         @Override
         public void put(Put put) throws IOException {
             table.put(put);