public void processRows ( RowProcessingMetrics rowProcessingMetrics ) { final RowProcessingQueryOptimizer queryOptimizer = getQueryOptimizer ( ) ; final Query finalQuery = queryOptimizer . getOptimizedQuery ( ) ; final RowIdGenerator idGenerator ; if ( finalQuery . getFirstRow ( ) == null ) { idGenerator = new SimpleRowIdGenerator ( ) ; } else { idGenerator = new SimpleRowIdGenerator ( finalQuery . getFirstRow ( ) ) ; } final AnalysisJob analysisJob = _publishers . getAnalysisJob ( ) ; final AnalysisListener analysisListener = _publishers . getAnalysisListener ( ) ; final TaskRunner taskRunner = _publishers . getTaskRunner ( ) ; for ( RowProcessingConsumer rowProcessingConsumer : _consumers ) { if ( rowProcessingConsumer instanceof AnalyzerConsumer ) { final AnalyzerConsumer analyzerConsumer = ( AnalyzerConsumer ) rowProcessingConsumer ; final AnalyzerJob analyzerJob = analyzerConsumer . getComponentJob ( ) ; final AnalyzerMetrics metrics = rowProcessingMetrics . getAnalysisJobMetrics ( ) . getAnalyzerMetrics ( analyzerJob ) ; analysisListener . analyzerBegin ( analysisJob , analyzerJob , metrics ) ; } if ( rowProcessingConsumer instanceof TransformerConsumer ) { ( ( TransformerConsumer ) rowProcessingConsumer ) . setRowIdGenerator ( idGenerator ) ; } } final List < RowProcessingConsumer > consumers = queryOptimizer . getOptimizedConsumers ( ) ; final Collection < ? extends FilterOutcome > availableOutcomes = queryOptimizer . getOptimizedAvailableOutcomes ( ) ; analysisListener . rowProcessingBegin ( analysisJob , rowProcessingMetrics ) ; final RowConsumerTaskListener taskListener = new RowConsumerTaskListener ( analysisJob , analysisListener , taskRunner ) ; final Datastore datastore = _publishers . getDatastore ( ) ; try ( final DatastoreConnection con = datastore . openConnection ( ) ) { final DataContext dataContext = con . getDataContext ( ) ; if ( logger . isDebugEnabled ( ) ) { final String queryString ; if ( dataContext instanceof JdbcDataContext ) { final JdbcDataContext jdbcDataContext = ( JdbcDataContext ) dataContext ; queryString = jdbcDataContext . getQueryRewriter ( ) . rewriteQuery ( finalQuery ) ; } else { queryString = finalQuery . toSql ( ) ; } logger . debug ( "Final query: {}" , queryString ) ; logger . debug ( "Final query firstRow={}, maxRows={}" , finalQuery . getFirstRow ( ) , finalQuery . getMaxRows ( ) ) ; } // represents the distinct count of rows as well as the number of // tasks to execute int numTasks = 0 ; try ( final DataSet dataSet = dataContext . executeQuery ( finalQuery ) ) { final ConsumeRowHandler consumeRowHandler = new ConsumeRowHandler ( consumers , availableOutcomes ) ; while ( dataSet . next ( ) ) { if ( taskListener . isErrornous ( ) ) { break ; } numTasks ++ ; final Row metaModelRow = dataSet . getRow ( ) ; final int rowId = idGenerator . nextPhysicalRowId ( ) ; final MetaModelInputRow inputRow = new MetaModelInputRow ( rowId , metaModelRow ) ; final ConsumeRowTask task = new ConsumeRowTask ( consumeRowHandler , rowProcessingMetrics , inputRow , analysisListener , numTasks ) ; taskRunner . run ( task , taskListener ) ; } } taskListener . awaitTasks ( numTasks ) ; } if ( taskListener . isErrornous ( ) ) { _successful . set ( false ) ; return ; } analysisListener . rowProcessingSuccess ( analysisJob , rowProcessingMetrics ) ; }
Fires the actual row processing . This method assumes that consumers have been initialized and the publisher is ready to start processing .