nl_tokens: Process all records in queue . All records have same operation to same table . Generate a query and set parameters from each record . INSERT and DELETE can be multi - row operation but UPDATE is single - row operation . If maxStatement
pl_tokens: private void processQueue ( LinkedList < Record > queue , List < OnRecordErrorException > errorRecords , Connection connection , int maxRowsPerBatch , int opCode ) throws StageException { if ( queue . isEmpty ( ) ) { return ; } int rowCount = 0 ; // Assume that columns are all same for the same operation to the same table // If some columns are missing in record, the record goes to error. final Record first = queue . getFirst ( ) ; SortedMap < String , String > columnsToParameters = recordReader . getColumnsToParameters ( first , opCode , getColumnsToParameters ( ) , opCode == OperationType . UPDATE_CODE ? getColumnsToFieldNoPK ( ) : getColumnsToFields ( ) ) ; if ( columnsToParameters . isEmpty ( ) ) { // no parameters found for configured columns if ( LOG . isWarnEnabled ( ) ) { LOG . warn ( "No parameters found for record with ID {}; skipping" , first . getHeader ( ) . getSourceId ( ) ) ; } return ; } String query = generateQueryForMultiRow ( opCode , columnsToParameters , getPrimaryKeyColumns ( ) , // the next batch will have either the max number of records, or however many are left. Math . <mask> ( maxRowsPerBatch , queue . size ( ) ) ) ; // Need to store removed records from queue, because we might need to add newly generated columns // to records for Jdbc Tee Processor. LinkedList < Record > removed = new LinkedList <> ( ) ; try ( PreparedStatement statement = jdbcUtil . getPreparedStatement ( getGeneratedColumnMappings ( ) , query , connection ) ) { int paramIdx = 1 ; // Start processing records in queue. All records have the same operation to the same table. while ( ! queue . isEmpty ( ) ) { Record r = queue . removeFirst ( ) ; if ( opCode != DELETE_CODE ) { paramIdx = setParamsToStatement ( paramIdx , statement , columnsToParameters , r , connection , opCode ) ; } if ( opCode != OperationType . INSERT_CODE ) { paramIdx = setPrimaryKeys ( paramIdx , r , statement , opCode ) ; } removed . add ( r ) ; ++ rowCount ; if ( rowCount == maxRowsPerBatch ) { // time to execute the current batch processBatch ( removed , errorRecords , statement , connection ) ; // reset our counters rowCount = 0 ; paramIdx = 1 ; removed . clear ( ) ; } } } catch ( SQLException e ) { handleSqlException ( e , removed , errorRecords ) ; } // Process the rest of the records that are removed from queue but haven't processed yet // this happens when rowCount is still less than maxRowsPerBatch. // This is a bit of an ugly fix as its not very DRY but sufficient until there's a larger // refactoring of this code. if ( ! removed . isEmpty ( ) ) { query = generateQueryForMultiRow ( opCode , columnsToParameters , getPrimaryKeyColumns ( ) , removed . size ( ) // always the remainder ) ; try ( PreparedStatement statement = jdbcUtil . getPreparedStatement ( getGeneratedColumnMappings ( ) , query , connection ) ) { int paramIdx = 1 ; for ( Record r : removed ) { if ( opCode != DELETE_CODE ) { paramIdx = setParamsToStatement ( paramIdx , statement , columnsToParameters , r , connection , opCode ) ; } if ( opCode != OperationType . INSERT_CODE ) { paramIdx = setPrimaryKeys ( paramIdx , r , statement , opCode ) ; } } processBatch ( removed , errorRecords , statement , connection ) ; } catch ( SQLException e ) { handleSqlException ( e , removed , errorRecords ) ; } } }