@Test public void testVisitCube() throws Exception { RawScan rawScan = mockFullScan(gtInfo, getTestConfig()); CoprocessorEnvironment env = PowerMockito.mock(RegionCoprocessorEnvironment.class); PowerMockito.when(env, "getRegion").thenReturn(region); final CubeVisitService service = new CubeVisitService(); service.start(env); CubeVisitProtos.CubeVisitRequest request = mockFullScanRequest(gtInfo, Lists.newArrayList(rawScan)); RpcCallback<CubeVisitProtos.CubeVisitResponse> done = new RpcCallback<CubeVisitProtos.CubeVisitResponse>() { @Override public void run(CubeVisitProtos.CubeVisitResponse result) { CubeVisitProtos.CubeVisitResponse.Stats stats = result.getStats(); Assert.assertEquals(0L, stats.getAggregatedRowCount()); Assert.assertEquals(0L, stats.getFilteredRowCount()); Assert.assertEquals(dateList.size() * userList.size(), stats.getScannedRowCount()); try { byte[] rawData = CompressionUtils .decompress(HBaseZeroCopyByteString.zeroCopyGetBytes(result.getCompressedRows())); PartitionResultIterator iterator = new PartitionResultIterator(rawData, gtInfo, setOf(0, 1, 2, 3)); int nReturn = 0; while (iterator.hasNext()) { iterator.next(); nReturn++; } Assert.assertEquals(dateList.size() * userList.size(), nReturn); } catch (Exception e) { Assert.fail("Fail due to " + e); } } }; service.visitCube(null, request, done); }
@SuppressWarnings("checkstyle:methodlength") @Override public void visitCube(final RpcController controller, final CubeVisitProtos.CubeVisitRequest request, RpcCallback<CubeVisitProtos.CubeVisitResponse> done) { List<RegionScanner> regionScanners = Lists.newArrayList(); HRegion region = null; StringBuilder sb = new StringBuilder(); byte[] allRows; String debugGitTag = ""; CubeVisitProtos.CubeVisitResponse.ErrorInfo errorInfo = null; KylinConfig kylinConfig = KylinConfig.createKylinConfig(request.getKylinProperties()); String queryId = request.hasQueryId() ? request.getQueryId() : "UnknownId"; logger.info("start query {} in thread {}", queryId, Thread.currentThread().getName()); RuntimeException shouldThrow = null; try (SetAndUnsetThreadLocalConfig autoUnset = KylinConfig.setAndUnsetThreadLocalConfig(kylinConfig); SetThreadName ignored = new SetThreadName("Query %s", queryId)) { final long serviceStartTime = System.currentTimeMillis(); region = (HRegion) env.getRegion(); region.startRegionOperation(); debugGitTag = region.getTableDesc().getValue(IRealizationConstants.HTableGitTag); final GTScanRequest scanReq = GTScanRequest.serializer .deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getGtScanRequest()))); final long deadline = scanReq.getStartTime() + scanReq.getTimeout(); checkDeadline(deadline); List<List<Integer>> hbaseColumnsToGT = Lists.newArrayList(); for (IntList intList : request.getHbaseColumnsToGTList()) { hbaseColumnsToGT.add(intList.getIntsList()); } StorageSideBehavior behavior = StorageSideBehavior.valueOf(scanReq.getStorageBehavior()); final List<RawScan> hbaseRawScans = deserializeRawScans( ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getHbaseRawScan()))); appendProfileInfo(sb, "start latency: " + (serviceStartTime - scanReq.getStartTime()), serviceStartTime); final List<InnerScannerAsIterator> cellListsForeachRawScan = Lists.newArrayList(); for (RawScan hbaseRawScan : hbaseRawScans) { if (request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN > 0) { updateRawScanByCurrentRegion(hbaseRawScan, region, request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN); } Scan scan = CubeHBaseRPC.buildScan(hbaseRawScan); RegionScanner innerScanner = region.getScanner(scan); regionScanners.add(innerScanner); InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner); cellListsForeachRawScan.add(cellListIterator); } final Iterator<List<Cell>> allCellLists = Iterators.concat(cellListsForeachRawScan.iterator()); if (behavior.ordinal() < StorageSideBehavior.SCAN.ordinal()) { List<Cell> temp = Lists.newArrayList(); int counter = 0; for (RegionScanner innerScanner : regionScanners) { while (innerScanner.nextRaw(temp)) { counter++; } } appendProfileInfo(sb, "scanned " + counter, serviceStartTime); } if (behavior.ordinal() < StorageSideBehavior.SCAN_FILTER_AGGR_CHECKMEM.ordinal()) { scanReq.disableAggCacheMemCheck(); } final long storagePushDownLimit = scanReq.getStoragePushDownLimit(); ResourceTrackingCellListIterator cellListIterator = new ResourceTrackingCellListIterator(allCellLists, scanReq.getStorageScanRowNumThreshold(), !request.hasMaxScanBytes() ? Long.MAX_VALUE : request.getMaxScanBytes(), deadline); IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScans.get(0).hbaseColumns, hbaseColumnsToGT, request.getRowkeyPreambleSize(), behavior.delayToggledOn(), request.getIsExactAggregate()); IGTScanner rawScanner = store.scan(scanReq); IGTScanner finalScanner = scanReq.decorateScanner(rawScanner, behavior.filterToggledOn(), behavior.aggrToggledOn(), false, request.getSpillEnabled()); ByteBuffer buffer = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); ByteArrayOutputStream outputStream = new ByteArrayOutputStream(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); long finalRowCount = 0L; try { for (GTRecord oneRecord : finalScanner) { buffer.clear(); try { oneRecord.exportColumns(scanReq.getColumns(), buffer); } catch (BufferOverflowException boe) { buffer = ByteBuffer.allocate(oneRecord.sizeOf(scanReq.getColumns()) * 2); oneRecord.exportColumns(scanReq.getColumns(), buffer); } outputStream.write(buffer.array(), 0, buffer.position()); finalRowCount++; if (!scanReq.isDoingStorageAggregation() && (scanReq.getStorageLimitLevel() != StorageLimitLevel.NO_LIMIT && finalRowCount >= storagePushDownLimit)) { logger.info("The finalScanner aborted because storagePushDownLimit is satisfied"); break; } } } catch (KylinTimeoutException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.TIMEOUT).setMessage(e.getMessage()) .build(); } catch (ResourceLimitExceededException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.RESOURCE_LIMIT_EXCEEDED) .setMessage(e.getMessage()).build(); } finally { finalScanner.close(); } long rowCountBeforeAggr = finalScanner instanceof GTAggregateScanner ? ((GTAggregateScanner) finalScanner).getInputRowCount() : finalRowCount; appendProfileInfo(sb, "agg done", serviceStartTime); logger.info("Total scanned {} rows and {} bytes", cellListIterator.getTotalScannedRowCount(), cellListIterator.getTotalScannedRowBytes()); byte[] compressedAllRows; if (errorInfo == null) { allRows = outputStream.toByteArray(); } else { allRows = new byte[0]; } if (!kylinConfig.getCompressionResult()) { compressedAllRows = allRows; } else { compressedAllRows = CompressionUtils.compress(allRows); } appendProfileInfo(sb, "compress done", serviceStartTime); logger.info("Size of final result = {} ({} before compressing)", compressedAllRows.length, allRows.length); OperatingSystemMXBean operatingSystemMXBean = (OperatingSystemMXBean) ManagementFactory .getOperatingSystemMXBean(); double systemCpuLoad = operatingSystemMXBean.getSystemCpuLoad(); double freePhysicalMemorySize = operatingSystemMXBean.getFreePhysicalMemorySize(); double freeSwapSpaceSize = operatingSystemMXBean.getFreeSwapSpaceSize(); appendProfileInfo(sb, "server stats done", serviceStartTime); sb.append(" debugGitTag:" + debugGitTag); CubeVisitProtos.CubeVisitResponse.Builder responseBuilder = CubeVisitProtos.CubeVisitResponse.newBuilder(); if (errorInfo != null) { responseBuilder.setErrorInfo(errorInfo); } done.run(responseBuilder. setCompressedRows(HBaseZeroCopyByteString.wrap(compressedAllRows)). setStats(CubeVisitProtos.CubeVisitResponse.Stats.newBuilder() .setFilteredRowCount(cellListIterator.getTotalScannedRowCount() - rowCountBeforeAggr) .setAggregatedRowCount(rowCountBeforeAggr - finalRowCount) .setScannedRowCount(cellListIterator.getTotalScannedRowCount()) .setScannedBytes(cellListIterator.getTotalScannedRowBytes()) .setServiceStartTime(serviceStartTime).setServiceEndTime(System.currentTimeMillis()) .setSystemCpuLoad(systemCpuLoad).setFreePhysicalMemorySize(freePhysicalMemorySize) .setFreeSwapSpaceSize(freeSwapSpaceSize) .setHostname(InetAddress.getLocalHost().getHostName()).setEtcMsg(sb.toString()) .setNormalComplete(errorInfo == null ? 1 : 0).build()) .build()); } catch (DoNotRetryIOException e) { ResponseConverter.setControllerException(controller, e); } catch (IOException ioe) { logger.error(ioe.toString(), ioe); IOException wrapped = new IOException("Error in coprocessor " + debugGitTag, ioe); ResponseConverter.setControllerException(controller, wrapped); } finally { for (RegionScanner innerScanner : regionScanners) { IOUtils.closeQuietly(innerScanner); } if (region != null) { try { region.closeRegionOperation(); } catch (IOException e) { shouldThrow = new RuntimeException(e); } } } if (null != shouldThrow) { throw shouldThrow; } }
CubeVisitService extends CubeVisitProtos.CubeVisitService implements Coprocessor, CoprocessorService { @SuppressWarnings("checkstyle:methodlength") @Override public void visitCube(final RpcController controller, final CubeVisitProtos.CubeVisitRequest request, RpcCallback<CubeVisitProtos.CubeVisitResponse> done) { List<RegionScanner> regionScanners = Lists.newArrayList(); HRegion region = null; StringBuilder sb = new StringBuilder(); byte[] allRows; String debugGitTag = ""; CubeVisitProtos.CubeVisitResponse.ErrorInfo errorInfo = null; KylinConfig kylinConfig = KylinConfig.createKylinConfig(request.getKylinProperties()); String queryId = request.hasQueryId() ? request.getQueryId() : "UnknownId"; logger.info("start query {} in thread {}", queryId, Thread.currentThread().getName()); RuntimeException shouldThrow = null; try (SetAndUnsetThreadLocalConfig autoUnset = KylinConfig.setAndUnsetThreadLocalConfig(kylinConfig); SetThreadName ignored = new SetThreadName("Query %s", queryId)) { final long serviceStartTime = System.currentTimeMillis(); region = (HRegion) env.getRegion(); region.startRegionOperation(); debugGitTag = region.getTableDesc().getValue(IRealizationConstants.HTableGitTag); final GTScanRequest scanReq = GTScanRequest.serializer .deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getGtScanRequest()))); final long deadline = scanReq.getStartTime() + scanReq.getTimeout(); checkDeadline(deadline); List<List<Integer>> hbaseColumnsToGT = Lists.newArrayList(); for (IntList intList : request.getHbaseColumnsToGTList()) { hbaseColumnsToGT.add(intList.getIntsList()); } StorageSideBehavior behavior = StorageSideBehavior.valueOf(scanReq.getStorageBehavior()); final List<RawScan> hbaseRawScans = deserializeRawScans( ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getHbaseRawScan()))); appendProfileInfo(sb, "start latency: " + (serviceStartTime - scanReq.getStartTime()), serviceStartTime); final List<InnerScannerAsIterator> cellListsForeachRawScan = Lists.newArrayList(); for (RawScan hbaseRawScan : hbaseRawScans) { if (request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN > 0) { updateRawScanByCurrentRegion(hbaseRawScan, region, request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN); } Scan scan = CubeHBaseRPC.buildScan(hbaseRawScan); RegionScanner innerScanner = region.getScanner(scan); regionScanners.add(innerScanner); InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner); cellListsForeachRawScan.add(cellListIterator); } final Iterator<List<Cell>> allCellLists = Iterators.concat(cellListsForeachRawScan.iterator()); if (behavior.ordinal() < StorageSideBehavior.SCAN.ordinal()) { List<Cell> temp = Lists.newArrayList(); int counter = 0; for (RegionScanner innerScanner : regionScanners) { while (innerScanner.nextRaw(temp)) { counter++; } } appendProfileInfo(sb, "scanned " + counter, serviceStartTime); } if (behavior.ordinal() < StorageSideBehavior.SCAN_FILTER_AGGR_CHECKMEM.ordinal()) { scanReq.disableAggCacheMemCheck(); } final long storagePushDownLimit = scanReq.getStoragePushDownLimit(); ResourceTrackingCellListIterator cellListIterator = new ResourceTrackingCellListIterator(allCellLists, scanReq.getStorageScanRowNumThreshold(), !request.hasMaxScanBytes() ? Long.MAX_VALUE : request.getMaxScanBytes(), deadline); IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScans.get(0).hbaseColumns, hbaseColumnsToGT, request.getRowkeyPreambleSize(), behavior.delayToggledOn(), request.getIsExactAggregate()); IGTScanner rawScanner = store.scan(scanReq); IGTScanner finalScanner = scanReq.decorateScanner(rawScanner, behavior.filterToggledOn(), behavior.aggrToggledOn(), false, request.getSpillEnabled()); ByteBuffer buffer = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); ByteArrayOutputStream outputStream = new ByteArrayOutputStream(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); long finalRowCount = 0L; try { for (GTRecord oneRecord : finalScanner) { buffer.clear(); try { oneRecord.exportColumns(scanReq.getColumns(), buffer); } catch (BufferOverflowException boe) { buffer = ByteBuffer.allocate(oneRecord.sizeOf(scanReq.getColumns()) * 2); oneRecord.exportColumns(scanReq.getColumns(), buffer); } outputStream.write(buffer.array(), 0, buffer.position()); finalRowCount++; if (!scanReq.isDoingStorageAggregation() && (scanReq.getStorageLimitLevel() != StorageLimitLevel.NO_LIMIT && finalRowCount >= storagePushDownLimit)) { logger.info("The finalScanner aborted because storagePushDownLimit is satisfied"); break; } } } catch (KylinTimeoutException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.TIMEOUT).setMessage(e.getMessage()) .build(); } catch (ResourceLimitExceededException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.RESOURCE_LIMIT_EXCEEDED) .setMessage(e.getMessage()).build(); } finally { finalScanner.close(); } long rowCountBeforeAggr = finalScanner instanceof GTAggregateScanner ? ((GTAggregateScanner) finalScanner).getInputRowCount() : finalRowCount; appendProfileInfo(sb, "agg done", serviceStartTime); logger.info("Total scanned {} rows and {} bytes", cellListIterator.getTotalScannedRowCount(), cellListIterator.getTotalScannedRowBytes()); byte[] compressedAllRows; if (errorInfo == null) { allRows = outputStream.toByteArray(); } else { allRows = new byte[0]; } if (!kylinConfig.getCompressionResult()) { compressedAllRows = allRows; } else { compressedAllRows = CompressionUtils.compress(allRows); } appendProfileInfo(sb, "compress done", serviceStartTime); logger.info("Size of final result = {} ({} before compressing)", compressedAllRows.length, allRows.length); OperatingSystemMXBean operatingSystemMXBean = (OperatingSystemMXBean) ManagementFactory .getOperatingSystemMXBean(); double systemCpuLoad = operatingSystemMXBean.getSystemCpuLoad(); double freePhysicalMemorySize = operatingSystemMXBean.getFreePhysicalMemorySize(); double freeSwapSpaceSize = operatingSystemMXBean.getFreeSwapSpaceSize(); appendProfileInfo(sb, "server stats done", serviceStartTime); sb.append(" debugGitTag:" + debugGitTag); CubeVisitProtos.CubeVisitResponse.Builder responseBuilder = CubeVisitProtos.CubeVisitResponse.newBuilder(); if (errorInfo != null) { responseBuilder.setErrorInfo(errorInfo); } done.run(responseBuilder. setCompressedRows(HBaseZeroCopyByteString.wrap(compressedAllRows)). setStats(CubeVisitProtos.CubeVisitResponse.Stats.newBuilder() .setFilteredRowCount(cellListIterator.getTotalScannedRowCount() - rowCountBeforeAggr) .setAggregatedRowCount(rowCountBeforeAggr - finalRowCount) .setScannedRowCount(cellListIterator.getTotalScannedRowCount()) .setScannedBytes(cellListIterator.getTotalScannedRowBytes()) .setServiceStartTime(serviceStartTime).setServiceEndTime(System.currentTimeMillis()) .setSystemCpuLoad(systemCpuLoad).setFreePhysicalMemorySize(freePhysicalMemorySize) .setFreeSwapSpaceSize(freeSwapSpaceSize) .setHostname(InetAddress.getLocalHost().getHostName()).setEtcMsg(sb.toString()) .setNormalComplete(errorInfo == null ? 1 : 0).build()) .build()); } catch (DoNotRetryIOException e) { ResponseConverter.setControllerException(controller, e); } catch (IOException ioe) { logger.error(ioe.toString(), ioe); IOException wrapped = new IOException("Error in coprocessor " + debugGitTag, ioe); ResponseConverter.setControllerException(controller, wrapped); } finally { for (RegionScanner innerScanner : regionScanners) { IOUtils.closeQuietly(innerScanner); } if (region != null) { try { region.closeRegionOperation(); } catch (IOException e) { shouldThrow = new RuntimeException(e); } } } if (null != shouldThrow) { throw shouldThrow; } } }
CubeVisitService extends CubeVisitProtos.CubeVisitService implements Coprocessor, CoprocessorService { @SuppressWarnings("checkstyle:methodlength") @Override public void visitCube(final RpcController controller, final CubeVisitProtos.CubeVisitRequest request, RpcCallback<CubeVisitProtos.CubeVisitResponse> done) { List<RegionScanner> regionScanners = Lists.newArrayList(); HRegion region = null; StringBuilder sb = new StringBuilder(); byte[] allRows; String debugGitTag = ""; CubeVisitProtos.CubeVisitResponse.ErrorInfo errorInfo = null; KylinConfig kylinConfig = KylinConfig.createKylinConfig(request.getKylinProperties()); String queryId = request.hasQueryId() ? request.getQueryId() : "UnknownId"; logger.info("start query {} in thread {}", queryId, Thread.currentThread().getName()); RuntimeException shouldThrow = null; try (SetAndUnsetThreadLocalConfig autoUnset = KylinConfig.setAndUnsetThreadLocalConfig(kylinConfig); SetThreadName ignored = new SetThreadName("Query %s", queryId)) { final long serviceStartTime = System.currentTimeMillis(); region = (HRegion) env.getRegion(); region.startRegionOperation(); debugGitTag = region.getTableDesc().getValue(IRealizationConstants.HTableGitTag); final GTScanRequest scanReq = GTScanRequest.serializer .deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getGtScanRequest()))); final long deadline = scanReq.getStartTime() + scanReq.getTimeout(); checkDeadline(deadline); List<List<Integer>> hbaseColumnsToGT = Lists.newArrayList(); for (IntList intList : request.getHbaseColumnsToGTList()) { hbaseColumnsToGT.add(intList.getIntsList()); } StorageSideBehavior behavior = StorageSideBehavior.valueOf(scanReq.getStorageBehavior()); final List<RawScan> hbaseRawScans = deserializeRawScans( ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getHbaseRawScan()))); appendProfileInfo(sb, "start latency: " + (serviceStartTime - scanReq.getStartTime()), serviceStartTime); final List<InnerScannerAsIterator> cellListsForeachRawScan = Lists.newArrayList(); for (RawScan hbaseRawScan : hbaseRawScans) { if (request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN > 0) { updateRawScanByCurrentRegion(hbaseRawScan, region, request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN); } Scan scan = CubeHBaseRPC.buildScan(hbaseRawScan); RegionScanner innerScanner = region.getScanner(scan); regionScanners.add(innerScanner); InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner); cellListsForeachRawScan.add(cellListIterator); } final Iterator<List<Cell>> allCellLists = Iterators.concat(cellListsForeachRawScan.iterator()); if (behavior.ordinal() < StorageSideBehavior.SCAN.ordinal()) { List<Cell> temp = Lists.newArrayList(); int counter = 0; for (RegionScanner innerScanner : regionScanners) { while (innerScanner.nextRaw(temp)) { counter++; } } appendProfileInfo(sb, "scanned " + counter, serviceStartTime); } if (behavior.ordinal() < StorageSideBehavior.SCAN_FILTER_AGGR_CHECKMEM.ordinal()) { scanReq.disableAggCacheMemCheck(); } final long storagePushDownLimit = scanReq.getStoragePushDownLimit(); ResourceTrackingCellListIterator cellListIterator = new ResourceTrackingCellListIterator(allCellLists, scanReq.getStorageScanRowNumThreshold(), !request.hasMaxScanBytes() ? Long.MAX_VALUE : request.getMaxScanBytes(), deadline); IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScans.get(0).hbaseColumns, hbaseColumnsToGT, request.getRowkeyPreambleSize(), behavior.delayToggledOn(), request.getIsExactAggregate()); IGTScanner rawScanner = store.scan(scanReq); IGTScanner finalScanner = scanReq.decorateScanner(rawScanner, behavior.filterToggledOn(), behavior.aggrToggledOn(), false, request.getSpillEnabled()); ByteBuffer buffer = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); ByteArrayOutputStream outputStream = new ByteArrayOutputStream(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); long finalRowCount = 0L; try { for (GTRecord oneRecord : finalScanner) { buffer.clear(); try { oneRecord.exportColumns(scanReq.getColumns(), buffer); } catch (BufferOverflowException boe) { buffer = ByteBuffer.allocate(oneRecord.sizeOf(scanReq.getColumns()) * 2); oneRecord.exportColumns(scanReq.getColumns(), buffer); } outputStream.write(buffer.array(), 0, buffer.position()); finalRowCount++; if (!scanReq.isDoingStorageAggregation() && (scanReq.getStorageLimitLevel() != StorageLimitLevel.NO_LIMIT && finalRowCount >= storagePushDownLimit)) { logger.info("The finalScanner aborted because storagePushDownLimit is satisfied"); break; } } } catch (KylinTimeoutException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.TIMEOUT).setMessage(e.getMessage()) .build(); } catch (ResourceLimitExceededException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.RESOURCE_LIMIT_EXCEEDED) .setMessage(e.getMessage()).build(); } finally { finalScanner.close(); } long rowCountBeforeAggr = finalScanner instanceof GTAggregateScanner ? ((GTAggregateScanner) finalScanner).getInputRowCount() : finalRowCount; appendProfileInfo(sb, "agg done", serviceStartTime); logger.info("Total scanned {} rows and {} bytes", cellListIterator.getTotalScannedRowCount(), cellListIterator.getTotalScannedRowBytes()); byte[] compressedAllRows; if (errorInfo == null) { allRows = outputStream.toByteArray(); } else { allRows = new byte[0]; } if (!kylinConfig.getCompressionResult()) { compressedAllRows = allRows; } else { compressedAllRows = CompressionUtils.compress(allRows); } appendProfileInfo(sb, "compress done", serviceStartTime); logger.info("Size of final result = {} ({} before compressing)", compressedAllRows.length, allRows.length); OperatingSystemMXBean operatingSystemMXBean = (OperatingSystemMXBean) ManagementFactory .getOperatingSystemMXBean(); double systemCpuLoad = operatingSystemMXBean.getSystemCpuLoad(); double freePhysicalMemorySize = operatingSystemMXBean.getFreePhysicalMemorySize(); double freeSwapSpaceSize = operatingSystemMXBean.getFreeSwapSpaceSize(); appendProfileInfo(sb, "server stats done", serviceStartTime); sb.append(" debugGitTag:" + debugGitTag); CubeVisitProtos.CubeVisitResponse.Builder responseBuilder = CubeVisitProtos.CubeVisitResponse.newBuilder(); if (errorInfo != null) { responseBuilder.setErrorInfo(errorInfo); } done.run(responseBuilder. setCompressedRows(HBaseZeroCopyByteString.wrap(compressedAllRows)). setStats(CubeVisitProtos.CubeVisitResponse.Stats.newBuilder() .setFilteredRowCount(cellListIterator.getTotalScannedRowCount() - rowCountBeforeAggr) .setAggregatedRowCount(rowCountBeforeAggr - finalRowCount) .setScannedRowCount(cellListIterator.getTotalScannedRowCount()) .setScannedBytes(cellListIterator.getTotalScannedRowBytes()) .setServiceStartTime(serviceStartTime).setServiceEndTime(System.currentTimeMillis()) .setSystemCpuLoad(systemCpuLoad).setFreePhysicalMemorySize(freePhysicalMemorySize) .setFreeSwapSpaceSize(freeSwapSpaceSize) .setHostname(InetAddress.getLocalHost().getHostName()).setEtcMsg(sb.toString()) .setNormalComplete(errorInfo == null ? 1 : 0).build()) .build()); } catch (DoNotRetryIOException e) { ResponseConverter.setControllerException(controller, e); } catch (IOException ioe) { logger.error(ioe.toString(), ioe); IOException wrapped = new IOException("Error in coprocessor " + debugGitTag, ioe); ResponseConverter.setControllerException(controller, wrapped); } finally { for (RegionScanner innerScanner : regionScanners) { IOUtils.closeQuietly(innerScanner); } if (region != null) { try { region.closeRegionOperation(); } catch (IOException e) { shouldThrow = new RuntimeException(e); } } } if (null != shouldThrow) { throw shouldThrow; } }  }
CubeVisitService extends CubeVisitProtos.CubeVisitService implements Coprocessor, CoprocessorService { @SuppressWarnings("checkstyle:methodlength") @Override public void visitCube(final RpcController controller, final CubeVisitProtos.CubeVisitRequest request, RpcCallback<CubeVisitProtos.CubeVisitResponse> done) { List<RegionScanner> regionScanners = Lists.newArrayList(); HRegion region = null; StringBuilder sb = new StringBuilder(); byte[] allRows; String debugGitTag = ""; CubeVisitProtos.CubeVisitResponse.ErrorInfo errorInfo = null; KylinConfig kylinConfig = KylinConfig.createKylinConfig(request.getKylinProperties()); String queryId = request.hasQueryId() ? request.getQueryId() : "UnknownId"; logger.info("start query {} in thread {}", queryId, Thread.currentThread().getName()); RuntimeException shouldThrow = null; try (SetAndUnsetThreadLocalConfig autoUnset = KylinConfig.setAndUnsetThreadLocalConfig(kylinConfig); SetThreadName ignored = new SetThreadName("Query %s", queryId)) { final long serviceStartTime = System.currentTimeMillis(); region = (HRegion) env.getRegion(); region.startRegionOperation(); debugGitTag = region.getTableDesc().getValue(IRealizationConstants.HTableGitTag); final GTScanRequest scanReq = GTScanRequest.serializer .deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getGtScanRequest()))); final long deadline = scanReq.getStartTime() + scanReq.getTimeout(); checkDeadline(deadline); List<List<Integer>> hbaseColumnsToGT = Lists.newArrayList(); for (IntList intList : request.getHbaseColumnsToGTList()) { hbaseColumnsToGT.add(intList.getIntsList()); } StorageSideBehavior behavior = StorageSideBehavior.valueOf(scanReq.getStorageBehavior()); final List<RawScan> hbaseRawScans = deserializeRawScans( ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getHbaseRawScan()))); appendProfileInfo(sb, "start latency: " + (serviceStartTime - scanReq.getStartTime()), serviceStartTime); final List<InnerScannerAsIterator> cellListsForeachRawScan = Lists.newArrayList(); for (RawScan hbaseRawScan : hbaseRawScans) { if (request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN > 0) { updateRawScanByCurrentRegion(hbaseRawScan, region, request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN); } Scan scan = CubeHBaseRPC.buildScan(hbaseRawScan); RegionScanner innerScanner = region.getScanner(scan); regionScanners.add(innerScanner); InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner); cellListsForeachRawScan.add(cellListIterator); } final Iterator<List<Cell>> allCellLists = Iterators.concat(cellListsForeachRawScan.iterator()); if (behavior.ordinal() < StorageSideBehavior.SCAN.ordinal()) { List<Cell> temp = Lists.newArrayList(); int counter = 0; for (RegionScanner innerScanner : regionScanners) { while (innerScanner.nextRaw(temp)) { counter++; } } appendProfileInfo(sb, "scanned " + counter, serviceStartTime); } if (behavior.ordinal() < StorageSideBehavior.SCAN_FILTER_AGGR_CHECKMEM.ordinal()) { scanReq.disableAggCacheMemCheck(); } final long storagePushDownLimit = scanReq.getStoragePushDownLimit(); ResourceTrackingCellListIterator cellListIterator = new ResourceTrackingCellListIterator(allCellLists, scanReq.getStorageScanRowNumThreshold(), !request.hasMaxScanBytes() ? Long.MAX_VALUE : request.getMaxScanBytes(), deadline); IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScans.get(0).hbaseColumns, hbaseColumnsToGT, request.getRowkeyPreambleSize(), behavior.delayToggledOn(), request.getIsExactAggregate()); IGTScanner rawScanner = store.scan(scanReq); IGTScanner finalScanner = scanReq.decorateScanner(rawScanner, behavior.filterToggledOn(), behavior.aggrToggledOn(), false, request.getSpillEnabled()); ByteBuffer buffer = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); ByteArrayOutputStream outputStream = new ByteArrayOutputStream(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); long finalRowCount = 0L; try { for (GTRecord oneRecord : finalScanner) { buffer.clear(); try { oneRecord.exportColumns(scanReq.getColumns(), buffer); } catch (BufferOverflowException boe) { buffer = ByteBuffer.allocate(oneRecord.sizeOf(scanReq.getColumns()) * 2); oneRecord.exportColumns(scanReq.getColumns(), buffer); } outputStream.write(buffer.array(), 0, buffer.position()); finalRowCount++; if (!scanReq.isDoingStorageAggregation() && (scanReq.getStorageLimitLevel() != StorageLimitLevel.NO_LIMIT && finalRowCount >= storagePushDownLimit)) { logger.info("The finalScanner aborted because storagePushDownLimit is satisfied"); break; } } } catch (KylinTimeoutException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.TIMEOUT).setMessage(e.getMessage()) .build(); } catch (ResourceLimitExceededException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.RESOURCE_LIMIT_EXCEEDED) .setMessage(e.getMessage()).build(); } finally { finalScanner.close(); } long rowCountBeforeAggr = finalScanner instanceof GTAggregateScanner ? ((GTAggregateScanner) finalScanner).getInputRowCount() : finalRowCount; appendProfileInfo(sb, "agg done", serviceStartTime); logger.info("Total scanned {} rows and {} bytes", cellListIterator.getTotalScannedRowCount(), cellListIterator.getTotalScannedRowBytes()); byte[] compressedAllRows; if (errorInfo == null) { allRows = outputStream.toByteArray(); } else { allRows = new byte[0]; } if (!kylinConfig.getCompressionResult()) { compressedAllRows = allRows; } else { compressedAllRows = CompressionUtils.compress(allRows); } appendProfileInfo(sb, "compress done", serviceStartTime); logger.info("Size of final result = {} ({} before compressing)", compressedAllRows.length, allRows.length); OperatingSystemMXBean operatingSystemMXBean = (OperatingSystemMXBean) ManagementFactory .getOperatingSystemMXBean(); double systemCpuLoad = operatingSystemMXBean.getSystemCpuLoad(); double freePhysicalMemorySize = operatingSystemMXBean.getFreePhysicalMemorySize(); double freeSwapSpaceSize = operatingSystemMXBean.getFreeSwapSpaceSize(); appendProfileInfo(sb, "server stats done", serviceStartTime); sb.append(" debugGitTag:" + debugGitTag); CubeVisitProtos.CubeVisitResponse.Builder responseBuilder = CubeVisitProtos.CubeVisitResponse.newBuilder(); if (errorInfo != null) { responseBuilder.setErrorInfo(errorInfo); } done.run(responseBuilder. setCompressedRows(HBaseZeroCopyByteString.wrap(compressedAllRows)). setStats(CubeVisitProtos.CubeVisitResponse.Stats.newBuilder() .setFilteredRowCount(cellListIterator.getTotalScannedRowCount() - rowCountBeforeAggr) .setAggregatedRowCount(rowCountBeforeAggr - finalRowCount) .setScannedRowCount(cellListIterator.getTotalScannedRowCount()) .setScannedBytes(cellListIterator.getTotalScannedRowBytes()) .setServiceStartTime(serviceStartTime).setServiceEndTime(System.currentTimeMillis()) .setSystemCpuLoad(systemCpuLoad).setFreePhysicalMemorySize(freePhysicalMemorySize) .setFreeSwapSpaceSize(freeSwapSpaceSize) .setHostname(InetAddress.getLocalHost().getHostName()).setEtcMsg(sb.toString()) .setNormalComplete(errorInfo == null ? 1 : 0).build()) .build()); } catch (DoNotRetryIOException e) { ResponseConverter.setControllerException(controller, e); } catch (IOException ioe) { logger.error(ioe.toString(), ioe); IOException wrapped = new IOException("Error in coprocessor " + debugGitTag, ioe); ResponseConverter.setControllerException(controller, wrapped); } finally { for (RegionScanner innerScanner : regionScanners) { IOUtils.closeQuietly(innerScanner); } if (region != null) { try { region.closeRegionOperation(); } catch (IOException e) { shouldThrow = new RuntimeException(e); } } } if (null != shouldThrow) { throw shouldThrow; } }  @SuppressWarnings("checkstyle:methodlength") @Override void visitCube(final RpcController controller, final CubeVisitProtos.CubeVisitRequest request,
            RpcCallback<CubeVisitProtos.CubeVisitResponse> done); @Override void start(CoprocessorEnvironment env); @Override void stop(CoprocessorEnvironment env); @Override Service getService(); }
CubeVisitService extends CubeVisitProtos.CubeVisitService implements Coprocessor, CoprocessorService { @SuppressWarnings("checkstyle:methodlength") @Override public void visitCube(final RpcController controller, final CubeVisitProtos.CubeVisitRequest request, RpcCallback<CubeVisitProtos.CubeVisitResponse> done) { List<RegionScanner> regionScanners = Lists.newArrayList(); HRegion region = null; StringBuilder sb = new StringBuilder(); byte[] allRows; String debugGitTag = ""; CubeVisitProtos.CubeVisitResponse.ErrorInfo errorInfo = null; KylinConfig kylinConfig = KylinConfig.createKylinConfig(request.getKylinProperties()); String queryId = request.hasQueryId() ? request.getQueryId() : "UnknownId"; logger.info("start query {} in thread {}", queryId, Thread.currentThread().getName()); RuntimeException shouldThrow = null; try (SetAndUnsetThreadLocalConfig autoUnset = KylinConfig.setAndUnsetThreadLocalConfig(kylinConfig); SetThreadName ignored = new SetThreadName("Query %s", queryId)) { final long serviceStartTime = System.currentTimeMillis(); region = (HRegion) env.getRegion(); region.startRegionOperation(); debugGitTag = region.getTableDesc().getValue(IRealizationConstants.HTableGitTag); final GTScanRequest scanReq = GTScanRequest.serializer .deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getGtScanRequest()))); final long deadline = scanReq.getStartTime() + scanReq.getTimeout(); checkDeadline(deadline); List<List<Integer>> hbaseColumnsToGT = Lists.newArrayList(); for (IntList intList : request.getHbaseColumnsToGTList()) { hbaseColumnsToGT.add(intList.getIntsList()); } StorageSideBehavior behavior = StorageSideBehavior.valueOf(scanReq.getStorageBehavior()); final List<RawScan> hbaseRawScans = deserializeRawScans( ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getHbaseRawScan()))); appendProfileInfo(sb, "start latency: " + (serviceStartTime - scanReq.getStartTime()), serviceStartTime); final List<InnerScannerAsIterator> cellListsForeachRawScan = Lists.newArrayList(); for (RawScan hbaseRawScan : hbaseRawScans) { if (request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN > 0) { updateRawScanByCurrentRegion(hbaseRawScan, region, request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN); } Scan scan = CubeHBaseRPC.buildScan(hbaseRawScan); RegionScanner innerScanner = region.getScanner(scan); regionScanners.add(innerScanner); InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner); cellListsForeachRawScan.add(cellListIterator); } final Iterator<List<Cell>> allCellLists = Iterators.concat(cellListsForeachRawScan.iterator()); if (behavior.ordinal() < StorageSideBehavior.SCAN.ordinal()) { List<Cell> temp = Lists.newArrayList(); int counter = 0; for (RegionScanner innerScanner : regionScanners) { while (innerScanner.nextRaw(temp)) { counter++; } } appendProfileInfo(sb, "scanned " + counter, serviceStartTime); } if (behavior.ordinal() < StorageSideBehavior.SCAN_FILTER_AGGR_CHECKMEM.ordinal()) { scanReq.disableAggCacheMemCheck(); } final long storagePushDownLimit = scanReq.getStoragePushDownLimit(); ResourceTrackingCellListIterator cellListIterator = new ResourceTrackingCellListIterator(allCellLists, scanReq.getStorageScanRowNumThreshold(), !request.hasMaxScanBytes() ? Long.MAX_VALUE : request.getMaxScanBytes(), deadline); IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScans.get(0).hbaseColumns, hbaseColumnsToGT, request.getRowkeyPreambleSize(), behavior.delayToggledOn(), request.getIsExactAggregate()); IGTScanner rawScanner = store.scan(scanReq); IGTScanner finalScanner = scanReq.decorateScanner(rawScanner, behavior.filterToggledOn(), behavior.aggrToggledOn(), false, request.getSpillEnabled()); ByteBuffer buffer = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); ByteArrayOutputStream outputStream = new ByteArrayOutputStream(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); long finalRowCount = 0L; try { for (GTRecord oneRecord : finalScanner) { buffer.clear(); try { oneRecord.exportColumns(scanReq.getColumns(), buffer); } catch (BufferOverflowException boe) { buffer = ByteBuffer.allocate(oneRecord.sizeOf(scanReq.getColumns()) * 2); oneRecord.exportColumns(scanReq.getColumns(), buffer); } outputStream.write(buffer.array(), 0, buffer.position()); finalRowCount++; if (!scanReq.isDoingStorageAggregation() && (scanReq.getStorageLimitLevel() != StorageLimitLevel.NO_LIMIT && finalRowCount >= storagePushDownLimit)) { logger.info("The finalScanner aborted because storagePushDownLimit is satisfied"); break; } } } catch (KylinTimeoutException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.TIMEOUT).setMessage(e.getMessage()) .build(); } catch (ResourceLimitExceededException e) { logger.info("Abort scan: {}", e.getMessage()); errorInfo = CubeVisitProtos.CubeVisitResponse.ErrorInfo.newBuilder() .setType(CubeVisitProtos.CubeVisitResponse.ErrorType.RESOURCE_LIMIT_EXCEEDED) .setMessage(e.getMessage()).build(); } finally { finalScanner.close(); } long rowCountBeforeAggr = finalScanner instanceof GTAggregateScanner ? ((GTAggregateScanner) finalScanner).getInputRowCount() : finalRowCount; appendProfileInfo(sb, "agg done", serviceStartTime); logger.info("Total scanned {} rows and {} bytes", cellListIterator.getTotalScannedRowCount(), cellListIterator.getTotalScannedRowBytes()); byte[] compressedAllRows; if (errorInfo == null) { allRows = outputStream.toByteArray(); } else { allRows = new byte[0]; } if (!kylinConfig.getCompressionResult()) { compressedAllRows = allRows; } else { compressedAllRows = CompressionUtils.compress(allRows); } appendProfileInfo(sb, "compress done", serviceStartTime); logger.info("Size of final result = {} ({} before compressing)", compressedAllRows.length, allRows.length); OperatingSystemMXBean operatingSystemMXBean = (OperatingSystemMXBean) ManagementFactory .getOperatingSystemMXBean(); double systemCpuLoad = operatingSystemMXBean.getSystemCpuLoad(); double freePhysicalMemorySize = operatingSystemMXBean.getFreePhysicalMemorySize(); double freeSwapSpaceSize = operatingSystemMXBean.getFreeSwapSpaceSize(); appendProfileInfo(sb, "server stats done", serviceStartTime); sb.append(" debugGitTag:" + debugGitTag); CubeVisitProtos.CubeVisitResponse.Builder responseBuilder = CubeVisitProtos.CubeVisitResponse.newBuilder(); if (errorInfo != null) { responseBuilder.setErrorInfo(errorInfo); } done.run(responseBuilder. setCompressedRows(HBaseZeroCopyByteString.wrap(compressedAllRows)). setStats(CubeVisitProtos.CubeVisitResponse.Stats.newBuilder() .setFilteredRowCount(cellListIterator.getTotalScannedRowCount() - rowCountBeforeAggr) .setAggregatedRowCount(rowCountBeforeAggr - finalRowCount) .setScannedRowCount(cellListIterator.getTotalScannedRowCount()) .setScannedBytes(cellListIterator.getTotalScannedRowBytes()) .setServiceStartTime(serviceStartTime).setServiceEndTime(System.currentTimeMillis()) .setSystemCpuLoad(systemCpuLoad).setFreePhysicalMemorySize(freePhysicalMemorySize) .setFreeSwapSpaceSize(freeSwapSpaceSize) .setHostname(InetAddress.getLocalHost().getHostName()).setEtcMsg(sb.toString()) .setNormalComplete(errorInfo == null ? 1 : 0).build()) .build()); } catch (DoNotRetryIOException e) { ResponseConverter.setControllerException(controller, e); } catch (IOException ioe) { logger.error(ioe.toString(), ioe); IOException wrapped = new IOException("Error in coprocessor " + debugGitTag, ioe); ResponseConverter.setControllerException(controller, wrapped); } finally { for (RegionScanner innerScanner : regionScanners) { IOUtils.closeQuietly(innerScanner); } if (region != null) { try { region.closeRegionOperation(); } catch (IOException e) { shouldThrow = new RuntimeException(e); } } } if (null != shouldThrow) { throw shouldThrow; } }  @SuppressWarnings("checkstyle:methodlength") @Override void visitCube(final RpcController controller, final CubeVisitProtos.CubeVisitRequest request,
            RpcCallback<CubeVisitProtos.CubeVisitResponse> done); @Override void start(CoprocessorEnvironment env); @Override void stop(CoprocessorEnvironment env); @Override Service getService();  }