KuduPartitionScannerCallable implements Callable<Long> { @Override public Long call() throws Exception { long numRowsScanned = 0; KuduScanner aPartitionSpecificScanner = KuduScanToken.deserializeIntoScanner( kuduPartitionScanAssignmentMeta.getSerializedKuduScanToken(), kuduClientHandle); LOG.info("Scanning the following tablet " + KuduScanToken.stringifySerializedToken(kuduPartitionScanAssignmentMeta .getSerializedKuduScanToken(), kuduClientHandle)); KuduRecordWithMeta<T> beginScanRecord = new KuduRecordWithMeta<>(); beginScanRecord.setBeginScanMarker(true); beginScanRecord.setTabletMetadata(kuduPartitionScanAssignmentMeta); bufferForTransmittingRecords.add(beginScanRecord); while ( aPartitionSpecificScanner.hasMoreRows()) { LOG.debug("Number of columns being returned for this read " + aPartitionSpecificScanner.getProjectionSchema().getColumnCount()); RowResultIterator resultIterator = aPartitionSpecificScanner.nextRows(); if (resultIterator == null) { break; } else { while (resultIterator.hasNext()) { KuduRecordWithMeta<T> recordWithMeta = new KuduRecordWithMeta<>(); RowResult aRow = resultIterator.next(); recordWithMeta.setPositionInScan(numRowsScanned); T payload = clazzForResultObject.newInstance(); recordWithMeta.setThePayload(payload); recordWithMeta.setEndOfScanMarker(false); recordWithMeta.setTabletMetadata(kuduPartitionScanAssignmentMeta); setValuesInPOJO(aRow,payload); bufferForTransmittingRecords.add(recordWithMeta); numRowsScanned += 1; } } } aPartitionSpecificScanner.close(); KuduRecordWithMeta<T> endScanRecord = new KuduRecordWithMeta<>(); endScanRecord.setEndOfScanMarker(true); endScanRecord.setTabletMetadata(kuduPartitionScanAssignmentMeta); bufferForTransmittingRecords.add(endScanRecord); LOG.info(" Scanned a total of " + numRowsScanned + " for this scanner thread @tablet " + KuduScanToken.stringifySerializedToken(kuduPartitionScanAssignmentMeta.getSerializedKuduScanToken(), kuduClientHandle)); return numRowsScanned; } KuduPartitionScannerCallable(AbstractKuduInputOperator<T,C> kuduInputOperator,
      KuduPartitionScanAssignmentMeta partitionMeta, ApexKuduConnection apexKuduConnection,Map<String,Object> setters,
      SQLToKuduPredicatesTranslator parsedQueryInstance); void setValuesInPOJO(RowResult aRow, T payload); @Override Long call();  }
@KuduClusterTestContext(kuduClusterBasedTest = true) @Test public void testSettersForPojo() throws Exception { initOperatorState(); AbstractKuduPartitionScanner<UnitTestTablePojo,InputOperatorControlTuple> currentScanner = unitTestStepwiseScanInputOperator.getScanner(); SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator( "select introwkey as intColumn from unittests", new ArrayList<ColumnSchema>(columnDefs.values())); List<KuduPartitionScanAssignmentMeta> scansForThisQuery = currentScanner.preparePlanForScanners(translator); KuduPartitionScannerCallable<UnitTestTablePojo,InputOperatorControlTuple> threadToScan = new KuduPartitionScannerCallable<>(unitTestStepwiseScanInputOperator,scansForThisQuery.get(0), currentScanner.getConnectionPoolForThreads().get(0), unitTestStepwiseScanInputOperator.extractSettersForResultObject(translator),translator); long countOfScan = threadToScan.call(); }
@KuduClusterTestContext(kuduClusterBasedTest = true) @Test public void testRowScansForAllDataAcrossAllPartitions() throws Exception { partitonScanStrategy = KuduPartitionScanStrategy.MANY_TABLETS_PER_OPERATOR; numberOfKuduInputOperatorPartitions = 1; initOperatorState(); AbstractKuduPartitionScanner<UnitTestTablePojo,InputOperatorControlTuple> currentScanner = unitTestStepwiseScanInputOperator.getScanner(); truncateTable(); addTestDataRows(10); assertEquals((KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE * 10 ),countNumRowsInTable()); SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator( "select * from unittests", new ArrayList<ColumnSchema>(columnDefs.values())); List<KuduPartitionScanAssignmentMeta> scansForThisQuery = currentScanner.preparePlanForScanners(translator); long totalRowsRead = 0; unitTestStepwiseScanInputOperator.getBuffer().clear(); for (KuduPartitionScanAssignmentMeta aSegmentToScan : scansForThisQuery) { KuduPartitionScannerCallable<UnitTestTablePojo,InputOperatorControlTuple> threadToScan = new KuduPartitionScannerCallable<>(unitTestStepwiseScanInputOperator,aSegmentToScan, currentScanner.getConnectionPoolForThreads().get(0), unitTestStepwiseScanInputOperator.extractSettersForResultObject(translator),translator); totalRowsRead += threadToScan.call(); } int expectedCount = ( 10 * KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE) + ( 2 * KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE); assertEquals(expectedCount,unitTestStepwiseScanInputOperator.getBuffer().size()); partitonScanStrategy = KuduPartitionScanStrategy.ONE_TABLET_PER_OPERATOR; numberOfKuduInputOperatorPartitions = 5; }
@KuduClusterTestContext(kuduClusterBasedTest = true) @Test public void testRowScansForAllDataInSinglePartition() throws Exception { partitonScanStrategy = KuduPartitionScanStrategy.MANY_TABLETS_PER_OPERATOR; initOperatorState(); AbstractKuduPartitionScanner<UnitTestTablePojo,InputOperatorControlTuple> currentScanner = unitTestStepwiseScanInputOperator.getScanner(); truncateTable(); addTestDataRows(10); assertEquals((10 * KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE),countNumRowsInTable()); int intRowBoundary = Integer.MAX_VALUE / SPLIT_COUNT_FOR_INT_ROW_KEY; SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator( "select * from unittests where introwkey < " + intRowBoundary, new ArrayList<ColumnSchema>(columnDefs.values())); List<KuduPartitionScanAssignmentMeta> scansForThisQuery = currentScanner.preparePlanForScanners(translator); long totalRowsRead = 0; unitTestStepwiseScanInputOperator.getBuffer().clear(); for (KuduPartitionScanAssignmentMeta aSegmentToScan : scansForThisQuery) { KuduPartitionScannerCallable<UnitTestTablePojo,InputOperatorControlTuple> threadToScan = new KuduPartitionScannerCallable<>(unitTestStepwiseScanInputOperator,aSegmentToScan, currentScanner.getConnectionPoolForThreads().get(0), unitTestStepwiseScanInputOperator.extractSettersForResultObject(translator),translator); totalRowsRead += threadToScan.call(); } assertEquals(23L,unitTestStepwiseScanInputOperator.getBuffer().size()); partitonScanStrategy = KuduPartitionScanStrategy.ONE_TABLET_PER_OPERATOR; numberOfKuduInputOperatorPartitions = 5; }