AbstractKuduPartitionScanner { public List<KuduPartitionScanAssignmentMeta> preparePlanForScanners(SQLToKuduPredicatesTranslator parsedQuery) throws IOException { List<KuduPredicate> predicateList = parsedQuery.getKuduSQLParseTreeListener().getKuduPredicateList(); ApexKuduConnection apexKuduConnection = verifyConnectionStaleness(0); KuduScanToken.KuduScanTokenBuilder builder = apexKuduConnection.getKuduClient().newScanTokenBuilder( apexKuduConnection.getKuduTable()); builder = builder.setProjectedColumnNames(new ArrayList<>( parsedQuery.getKuduSQLParseTreeListener().getListOfColumnsUsed())); for (KuduPredicate aPredicate : predicateList) { builder = builder.addPredicate(aPredicate); } builder.setFaultTolerant(parentOperator.isFaultTolerantScanner()); Map<String,String> optionsUsedForThisQuery = parentOperator.getOptionsEnabledForCurrentQuery(); if ( optionsUsedForThisQuery.containsKey(KuduSQLParseTreeListener.READ_SNAPSHOT_TIME)) { try { long readSnapShotTime = Long.valueOf(optionsUsedForThisQuery.get(KuduSQLParseTreeListener.READ_SNAPSHOT_TIME)); builder = builder.readMode(AsyncKuduScanner.ReadMode.READ_AT_SNAPSHOT); builder = builder.snapshotTimestampMicros(readSnapShotTime); LOG.info("Using read snapshot for this query as " + readSnapShotTime); } catch ( Exception ex) { LOG.error("Cannot parse the Read snaptshot time " + ex.getMessage(), ex); } } List<KuduScanToken> allPossibleScanTokens = builder.build(); Collections.sort(allPossibleScanTokens, new Comparator<KuduScanToken>() { @Override public int compare(KuduScanToken left, KuduScanToken right) { return left.compareTo(right); } }); LOG.info(" Query will scan " + allPossibleScanTokens.size() + " tablets"); if ( LOG.isDebugEnabled()) { LOG.debug(" Predicates scheduled for this query are " + predicateList.size()); for ( int i = 0; i < allPossibleScanTokens.size(); i++) { LOG.debug("A tablet scheduled for all operators scanning is " + allPossibleScanTokens.get(i).getTablet()); } } List<KuduPartitionScanAssignmentMeta> partitionPieForThisOperator = parentOperator.getPartitionPieAssignment(); List<KuduPartitionScanAssignmentMeta> returnOfAssignments = new ArrayList<>(); int totalScansForThisQuery = allPossibleScanTokens.size(); int counterForPartAssignments = 0; for (KuduPartitionScanAssignmentMeta aPartofThePie : partitionPieForThisOperator) { if ( aPartofThePie.getOrdinal() < totalScansForThisQuery) { KuduPartitionScanAssignmentMeta aMetaForThisQuery = new KuduPartitionScanAssignmentMeta(); aMetaForThisQuery.setTotalSize(totalScansForThisQuery); aMetaForThisQuery.setOrdinal(counterForPartAssignments); counterForPartAssignments += 1; aMetaForThisQuery.setCurrentQuery(parsedQuery.getSqlExpresssion()); KuduScanToken aTokenForThisOperator = allPossibleScanTokens.get(aPartofThePie.getOrdinal()); aMetaForThisQuery.setSerializedKuduScanToken(aTokenForThisOperator.serialize()); returnOfAssignments.add(aMetaForThisQuery); LOG.debug("Added query scan for this operator " + aMetaForThisQuery + " with scan tablet as " + allPossibleScanTokens.get(aPartofThePie.getOrdinal()).getTablet()); } } LOG.info(" A total of " + returnOfAssignments.size() + " have been scheduled for this operator"); return returnOfAssignments; } AbstractKuduPartitionScanner(AbstractKuduInputOperator<T,C> parentOperatorRunningThisScanner); abstract int scanAllRecords(SQLToKuduPredicatesTranslator parsedQuery,
      Map<String,Object> setters); void initScannerCommons(); ApexKuduConnection verifyConnectionStaleness(int indexPos); List<KuduPartitionScanAssignmentMeta> preparePlanForScanners(SQLToKuduPredicatesTranslator parsedQuery); void close(); AbstractKuduInputOperator<T, C> getParentOperator(); void setParentOperator(AbstractKuduInputOperator<T, C> parentOperator); Map<Integer, ApexKuduConnection> getConnectionPoolForThreads(); void setConnectionPoolForThreads(Map<Integer, ApexKuduConnection> connectionPoolForThreads);  }
@KuduClusterTestContext(kuduClusterBasedTest = true) @Test public void textPrepaprePlanForScanners() throws Exception { initOperatorState(); AbstractKuduPartitionScanner<UnitTestTablePojo, InputOperatorControlTuple> scanner = unitTestStepwiseScanInputOperator.getScanner(); SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator( "select introwkey as intColumn from unittests", new ArrayList<ColumnSchema>(columnDefs.values())); List<KuduPartitionScanAssignmentMeta> scansForThisQuery = scanner.preparePlanForScanners(translator); assertEquals(1,scansForThisQuery.size()); numberOfKuduInputOperatorPartitions = 4; partitonScanStrategy = KuduPartitionScanStrategy.MANY_TABLETS_PER_OPERATOR; initOperatorState(); translator = new SQLToKuduPredicatesTranslator( "select introwkey as intColumn from unittests", new ArrayList<ColumnSchema>(columnDefs.values())); scanner = unitTestStepwiseScanInputOperator.getScanner(); scansForThisQuery = scanner.preparePlanForScanners(translator); assertEquals(3,scansForThisQuery.size()); numberOfKuduInputOperatorPartitions = 4; partitonScanStrategy = KuduPartitionScanStrategy.MANY_TABLETS_PER_OPERATOR; initOperatorState(); translator = new SQLToKuduPredicatesTranslator( "select introwkey as intColumn from unittests where introwkey = 1", new ArrayList<ColumnSchema>(columnDefs.values())); scanner = unitTestStepwiseScanInputOperator.getScanner(); scansForThisQuery = scanner.preparePlanForScanners(translator); assertEquals(1,scansForThisQuery.size()); numberOfKuduInputOperatorPartitions = 5; partitonScanStrategy = KuduPartitionScanStrategy.ONE_TABLET_PER_OPERATOR; }