@Test public void testProcessRow() throws KettleException { GroupByMeta groupByMeta = mock( GroupByMeta.class ); GroupByData groupByData = mock( GroupByData.class ); GroupBy groupBySpy = Mockito.spy( new GroupBy( mockHelper.stepMeta, mockHelper.stepDataInterface, 0, mockHelper.transMeta, mockHelper.trans ) ); doReturn( null ).when( groupBySpy ).getRow(); doReturn( null ).when( groupBySpy ).getInputRowMeta(); RowMetaInterface rowMeta = new RowMeta(); rowMeta.addValueMeta( new ValueMetaInteger( "ROWNR" ) ); List<RowSet> outputRowSets = new ArrayList<RowSet>(); BlockingRowSet rowSet = new BlockingRowSet( 1 ); rowSet.putRow( rowMeta, new Object[] { new Long( 0 ) } ); outputRowSets.add( rowSet ); groupBySpy.setOutputRowSets( outputRowSets ); final String[] sub = { "b" }; doReturn( sub ).when( groupByMeta ).getSubjectField(); final String[] groupField = { "a" }; doReturn( groupField ).when( groupByMeta ).getGroupField(); final String[] aggFields = { "b_g" }; doReturn( aggFields ).when( groupByMeta ).getAggregateField(); final int[] aggType = { GroupByMeta.TYPE_GROUP_CONCAT_COMMA }; doReturn( aggType ).when( groupByMeta ).getAggregateType(); when( mockHelper.transMeta.getPrevStepFields( mockHelper.stepMeta ) ).thenReturn( new RowMeta() ); groupBySpy.processRow( groupByMeta, groupByData ); assertTrue( groupBySpy.getOutputRowSets().get( 0 ).isDone() ); }
@Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (GroupByMeta) smi; data = (GroupByData) sdi; Object[] r = getRow(); if ( first ) { String val = getVariable( Const.KETTLE_AGGREGATION_ALL_NULLS_ARE_ZERO, "N" ); allNullsAreZero = ValueMetaBase.convertStringToBoolean( val ); val = getVariable( Const.KETTLE_AGGREGATION_MIN_NULL_IS_VALUED, "N" ); minNullIsValued = ValueMetaBase.convertStringToBoolean( val ); data.inputRowMeta = getInputRowMeta(); if ( data.inputRowMeta == null ) { data.inputRowMeta = getTransMeta().getPrevStepFields( getStepMeta() ); } data.outputRowMeta = data.inputRowMeta.clone(); meta.getFields( data.outputRowMeta, getStepname(), null, null, this, repository, metaStore ); data.counts = new long[ meta.getSubjectField().length ]; data.subjectnrs = new int[ meta.getSubjectField().length ]; data.cumulativeSumSourceIndexes = new ArrayList<>(); data.cumulativeSumTargetIndexes = new ArrayList<>(); data.cumulativeAvgSourceIndexes = new ArrayList<>(); data.cumulativeAvgTargetIndexes = new ArrayList<>(); for ( int i = 0; i < meta.getSubjectField().length; i++ ) { if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_COUNT_ANY ) { data.subjectnrs[ i ] = 0; } else { data.subjectnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getSubjectField()[i] ); } if ( ( r != null ) && ( data.subjectnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.AggregateSubjectFieldCouldNotFound", meta.getSubjectField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_SUM ) { data.cumulativeSumSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeSumTargetIndexes.add( data.inputRowMeta.size() + i ); } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_AVERAGE ) { data.cumulativeAvgSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeAvgTargetIndexes.add( data.inputRowMeta.size() + i ); } } data.previousSums = new Object[ data.cumulativeSumTargetIndexes.size() ]; data.previousAvgSum = new Object[ data.cumulativeAvgTargetIndexes.size() ]; data.previousAvgCount = new long[ data.cumulativeAvgTargetIndexes.size() ]; data.groupnrs = new int[ meta.getGroupField().length ]; for ( int i = 0; i < meta.getGroupField().length; i++ ) { data.groupnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getGroupField()[i] ); if ( ( r != null ) && ( data.groupnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.GroupFieldCouldNotFound", meta.getGroupField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } } data.valueMetaInteger = new ValueMetaInteger( "count" ); data.valueMetaNumber = new ValueMetaNumber( "sum" ); initGroupMeta( data.inputRowMeta ); } if ( first || data.newBatch ) { newAggregate( r ); } if ( first ) { data.groupAggMeta = new RowMeta(); data.groupAggMeta.addRowMeta( data.groupMeta ); data.groupAggMeta.addRowMeta( data.aggMeta ); } if ( r == null ) { handleLastOfGroup(); setOutputDone(); return false; } if ( first || data.newBatch ) { first = false; data.newBatch = false; data.previous = data.inputRowMeta.cloneRow( r ); } else { calcAggregate( data.previous ); if ( meta.passAllRows() ) { addToBuffer( data.previous ); } } if ( !sameGroup( data.previous, r ) ) { if ( meta.passAllRows() ) { closeOutput(); data.groupResult = getAggregateResult(); Object[] row = getRowFromBuffer(); long lineNr = 0; while ( row != null ) { int size = data.inputRowMeta.size(); row = RowDataUtil.addRowData( row, size, data.groupResult ); size += data.groupResult.length; lineNr++; if ( meta.isAddingLineNrInGroup() && !Utils.isEmpty( meta.getLineNrInGroupField() ) ) { Object lineNrValue = new Long( lineNr ); row = RowDataUtil.addValueData( row, size, lineNrValue ); size++; } addCumulativeSums( row ); addCumulativeAverages( row ); putRow( data.outputRowMeta, row ); row = getRowFromBuffer(); } closeInput(); } else { Object[] result = buildResult( data.previous ); if ( result != null ) { putRow( data.groupAggMeta, result ); } } newAggregate( r ); } data.previous = data.inputRowMeta.cloneRow( r ); if ( checkFeedback( getLinesRead() ) ) { if ( log.isBasic() ) { logBasic( BaseMessages.getString( PKG, "GroupBy.LineNumber" ) + getLinesRead() ); } } return true; }
GroupBy extends BaseStep implements StepInterface { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (GroupByMeta) smi; data = (GroupByData) sdi; Object[] r = getRow(); if ( first ) { String val = getVariable( Const.KETTLE_AGGREGATION_ALL_NULLS_ARE_ZERO, "N" ); allNullsAreZero = ValueMetaBase.convertStringToBoolean( val ); val = getVariable( Const.KETTLE_AGGREGATION_MIN_NULL_IS_VALUED, "N" ); minNullIsValued = ValueMetaBase.convertStringToBoolean( val ); data.inputRowMeta = getInputRowMeta(); if ( data.inputRowMeta == null ) { data.inputRowMeta = getTransMeta().getPrevStepFields( getStepMeta() ); } data.outputRowMeta = data.inputRowMeta.clone(); meta.getFields( data.outputRowMeta, getStepname(), null, null, this, repository, metaStore ); data.counts = new long[ meta.getSubjectField().length ]; data.subjectnrs = new int[ meta.getSubjectField().length ]; data.cumulativeSumSourceIndexes = new ArrayList<>(); data.cumulativeSumTargetIndexes = new ArrayList<>(); data.cumulativeAvgSourceIndexes = new ArrayList<>(); data.cumulativeAvgTargetIndexes = new ArrayList<>(); for ( int i = 0; i < meta.getSubjectField().length; i++ ) { if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_COUNT_ANY ) { data.subjectnrs[ i ] = 0; } else { data.subjectnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getSubjectField()[i] ); } if ( ( r != null ) && ( data.subjectnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.AggregateSubjectFieldCouldNotFound", meta.getSubjectField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_SUM ) { data.cumulativeSumSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeSumTargetIndexes.add( data.inputRowMeta.size() + i ); } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_AVERAGE ) { data.cumulativeAvgSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeAvgTargetIndexes.add( data.inputRowMeta.size() + i ); } } data.previousSums = new Object[ data.cumulativeSumTargetIndexes.size() ]; data.previousAvgSum = new Object[ data.cumulativeAvgTargetIndexes.size() ]; data.previousAvgCount = new long[ data.cumulativeAvgTargetIndexes.size() ]; data.groupnrs = new int[ meta.getGroupField().length ]; for ( int i = 0; i < meta.getGroupField().length; i++ ) { data.groupnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getGroupField()[i] ); if ( ( r != null ) && ( data.groupnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.GroupFieldCouldNotFound", meta.getGroupField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } } data.valueMetaInteger = new ValueMetaInteger( "count" ); data.valueMetaNumber = new ValueMetaNumber( "sum" ); initGroupMeta( data.inputRowMeta ); } if ( first || data.newBatch ) { newAggregate( r ); } if ( first ) { data.groupAggMeta = new RowMeta(); data.groupAggMeta.addRowMeta( data.groupMeta ); data.groupAggMeta.addRowMeta( data.aggMeta ); } if ( r == null ) { handleLastOfGroup(); setOutputDone(); return false; } if ( first || data.newBatch ) { first = false; data.newBatch = false; data.previous = data.inputRowMeta.cloneRow( r ); } else { calcAggregate( data.previous ); if ( meta.passAllRows() ) { addToBuffer( data.previous ); } } if ( !sameGroup( data.previous, r ) ) { if ( meta.passAllRows() ) { closeOutput(); data.groupResult = getAggregateResult(); Object[] row = getRowFromBuffer(); long lineNr = 0; while ( row != null ) { int size = data.inputRowMeta.size(); row = RowDataUtil.addRowData( row, size, data.groupResult ); size += data.groupResult.length; lineNr++; if ( meta.isAddingLineNrInGroup() && !Utils.isEmpty( meta.getLineNrInGroupField() ) ) { Object lineNrValue = new Long( lineNr ); row = RowDataUtil.addValueData( row, size, lineNrValue ); size++; } addCumulativeSums( row ); addCumulativeAverages( row ); putRow( data.outputRowMeta, row ); row = getRowFromBuffer(); } closeInput(); } else { Object[] result = buildResult( data.previous ); if ( result != null ) { putRow( data.groupAggMeta, result ); } } newAggregate( r ); } data.previous = data.inputRowMeta.cloneRow( r ); if ( checkFeedback( getLinesRead() ) ) { if ( log.isBasic() ) { logBasic( BaseMessages.getString( PKG, "GroupBy.LineNumber" ) + getLinesRead() ); } } return true; } }
GroupBy extends BaseStep implements StepInterface { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (GroupByMeta) smi; data = (GroupByData) sdi; Object[] r = getRow(); if ( first ) { String val = getVariable( Const.KETTLE_AGGREGATION_ALL_NULLS_ARE_ZERO, "N" ); allNullsAreZero = ValueMetaBase.convertStringToBoolean( val ); val = getVariable( Const.KETTLE_AGGREGATION_MIN_NULL_IS_VALUED, "N" ); minNullIsValued = ValueMetaBase.convertStringToBoolean( val ); data.inputRowMeta = getInputRowMeta(); if ( data.inputRowMeta == null ) { data.inputRowMeta = getTransMeta().getPrevStepFields( getStepMeta() ); } data.outputRowMeta = data.inputRowMeta.clone(); meta.getFields( data.outputRowMeta, getStepname(), null, null, this, repository, metaStore ); data.counts = new long[ meta.getSubjectField().length ]; data.subjectnrs = new int[ meta.getSubjectField().length ]; data.cumulativeSumSourceIndexes = new ArrayList<>(); data.cumulativeSumTargetIndexes = new ArrayList<>(); data.cumulativeAvgSourceIndexes = new ArrayList<>(); data.cumulativeAvgTargetIndexes = new ArrayList<>(); for ( int i = 0; i < meta.getSubjectField().length; i++ ) { if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_COUNT_ANY ) { data.subjectnrs[ i ] = 0; } else { data.subjectnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getSubjectField()[i] ); } if ( ( r != null ) && ( data.subjectnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.AggregateSubjectFieldCouldNotFound", meta.getSubjectField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_SUM ) { data.cumulativeSumSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeSumTargetIndexes.add( data.inputRowMeta.size() + i ); } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_AVERAGE ) { data.cumulativeAvgSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeAvgTargetIndexes.add( data.inputRowMeta.size() + i ); } } data.previousSums = new Object[ data.cumulativeSumTargetIndexes.size() ]; data.previousAvgSum = new Object[ data.cumulativeAvgTargetIndexes.size() ]; data.previousAvgCount = new long[ data.cumulativeAvgTargetIndexes.size() ]; data.groupnrs = new int[ meta.getGroupField().length ]; for ( int i = 0; i < meta.getGroupField().length; i++ ) { data.groupnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getGroupField()[i] ); if ( ( r != null ) && ( data.groupnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.GroupFieldCouldNotFound", meta.getGroupField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } } data.valueMetaInteger = new ValueMetaInteger( "count" ); data.valueMetaNumber = new ValueMetaNumber( "sum" ); initGroupMeta( data.inputRowMeta ); } if ( first || data.newBatch ) { newAggregate( r ); } if ( first ) { data.groupAggMeta = new RowMeta(); data.groupAggMeta.addRowMeta( data.groupMeta ); data.groupAggMeta.addRowMeta( data.aggMeta ); } if ( r == null ) { handleLastOfGroup(); setOutputDone(); return false; } if ( first || data.newBatch ) { first = false; data.newBatch = false; data.previous = data.inputRowMeta.cloneRow( r ); } else { calcAggregate( data.previous ); if ( meta.passAllRows() ) { addToBuffer( data.previous ); } } if ( !sameGroup( data.previous, r ) ) { if ( meta.passAllRows() ) { closeOutput(); data.groupResult = getAggregateResult(); Object[] row = getRowFromBuffer(); long lineNr = 0; while ( row != null ) { int size = data.inputRowMeta.size(); row = RowDataUtil.addRowData( row, size, data.groupResult ); size += data.groupResult.length; lineNr++; if ( meta.isAddingLineNrInGroup() && !Utils.isEmpty( meta.getLineNrInGroupField() ) ) { Object lineNrValue = new Long( lineNr ); row = RowDataUtil.addValueData( row, size, lineNrValue ); size++; } addCumulativeSums( row ); addCumulativeAverages( row ); putRow( data.outputRowMeta, row ); row = getRowFromBuffer(); } closeInput(); } else { Object[] result = buildResult( data.previous ); if ( result != null ) { putRow( data.groupAggMeta, result ); } } newAggregate( r ); } data.previous = data.inputRowMeta.cloneRow( r ); if ( checkFeedback( getLinesRead() ) ) { if ( log.isBasic() ) { logBasic( BaseMessages.getString( PKG, "GroupBy.LineNumber" ) + getLinesRead() ); } } return true; } GroupBy( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
      Trans trans ); }
GroupBy extends BaseStep implements StepInterface { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (GroupByMeta) smi; data = (GroupByData) sdi; Object[] r = getRow(); if ( first ) { String val = getVariable( Const.KETTLE_AGGREGATION_ALL_NULLS_ARE_ZERO, "N" ); allNullsAreZero = ValueMetaBase.convertStringToBoolean( val ); val = getVariable( Const.KETTLE_AGGREGATION_MIN_NULL_IS_VALUED, "N" ); minNullIsValued = ValueMetaBase.convertStringToBoolean( val ); data.inputRowMeta = getInputRowMeta(); if ( data.inputRowMeta == null ) { data.inputRowMeta = getTransMeta().getPrevStepFields( getStepMeta() ); } data.outputRowMeta = data.inputRowMeta.clone(); meta.getFields( data.outputRowMeta, getStepname(), null, null, this, repository, metaStore ); data.counts = new long[ meta.getSubjectField().length ]; data.subjectnrs = new int[ meta.getSubjectField().length ]; data.cumulativeSumSourceIndexes = new ArrayList<>(); data.cumulativeSumTargetIndexes = new ArrayList<>(); data.cumulativeAvgSourceIndexes = new ArrayList<>(); data.cumulativeAvgTargetIndexes = new ArrayList<>(); for ( int i = 0; i < meta.getSubjectField().length; i++ ) { if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_COUNT_ANY ) { data.subjectnrs[ i ] = 0; } else { data.subjectnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getSubjectField()[i] ); } if ( ( r != null ) && ( data.subjectnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.AggregateSubjectFieldCouldNotFound", meta.getSubjectField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_SUM ) { data.cumulativeSumSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeSumTargetIndexes.add( data.inputRowMeta.size() + i ); } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_AVERAGE ) { data.cumulativeAvgSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeAvgTargetIndexes.add( data.inputRowMeta.size() + i ); } } data.previousSums = new Object[ data.cumulativeSumTargetIndexes.size() ]; data.previousAvgSum = new Object[ data.cumulativeAvgTargetIndexes.size() ]; data.previousAvgCount = new long[ data.cumulativeAvgTargetIndexes.size() ]; data.groupnrs = new int[ meta.getGroupField().length ]; for ( int i = 0; i < meta.getGroupField().length; i++ ) { data.groupnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getGroupField()[i] ); if ( ( r != null ) && ( data.groupnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.GroupFieldCouldNotFound", meta.getGroupField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } } data.valueMetaInteger = new ValueMetaInteger( "count" ); data.valueMetaNumber = new ValueMetaNumber( "sum" ); initGroupMeta( data.inputRowMeta ); } if ( first || data.newBatch ) { newAggregate( r ); } if ( first ) { data.groupAggMeta = new RowMeta(); data.groupAggMeta.addRowMeta( data.groupMeta ); data.groupAggMeta.addRowMeta( data.aggMeta ); } if ( r == null ) { handleLastOfGroup(); setOutputDone(); return false; } if ( first || data.newBatch ) { first = false; data.newBatch = false; data.previous = data.inputRowMeta.cloneRow( r ); } else { calcAggregate( data.previous ); if ( meta.passAllRows() ) { addToBuffer( data.previous ); } } if ( !sameGroup( data.previous, r ) ) { if ( meta.passAllRows() ) { closeOutput(); data.groupResult = getAggregateResult(); Object[] row = getRowFromBuffer(); long lineNr = 0; while ( row != null ) { int size = data.inputRowMeta.size(); row = RowDataUtil.addRowData( row, size, data.groupResult ); size += data.groupResult.length; lineNr++; if ( meta.isAddingLineNrInGroup() && !Utils.isEmpty( meta.getLineNrInGroupField() ) ) { Object lineNrValue = new Long( lineNr ); row = RowDataUtil.addValueData( row, size, lineNrValue ); size++; } addCumulativeSums( row ); addCumulativeAverages( row ); putRow( data.outputRowMeta, row ); row = getRowFromBuffer(); } closeInput(); } else { Object[] result = buildResult( data.previous ); if ( result != null ) { putRow( data.groupAggMeta, result ); } } newAggregate( r ); } data.previous = data.inputRowMeta.cloneRow( r ); if ( checkFeedback( getLinesRead() ) ) { if ( log.isBasic() ) { logBasic( BaseMessages.getString( PKG, "GroupBy.LineNumber" ) + getLinesRead() ); } } return true; } GroupBy( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
      Trans trans ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); @Override boolean init( StepMetaInterface smi, StepDataInterface sdi ); @Override void dispose( StepMetaInterface smi, StepDataInterface sdi ); @Override void batchComplete(); GroupByMeta getMeta(); }
GroupBy extends BaseStep implements StepInterface { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (GroupByMeta) smi; data = (GroupByData) sdi; Object[] r = getRow(); if ( first ) { String val = getVariable( Const.KETTLE_AGGREGATION_ALL_NULLS_ARE_ZERO, "N" ); allNullsAreZero = ValueMetaBase.convertStringToBoolean( val ); val = getVariable( Const.KETTLE_AGGREGATION_MIN_NULL_IS_VALUED, "N" ); minNullIsValued = ValueMetaBase.convertStringToBoolean( val ); data.inputRowMeta = getInputRowMeta(); if ( data.inputRowMeta == null ) { data.inputRowMeta = getTransMeta().getPrevStepFields( getStepMeta() ); } data.outputRowMeta = data.inputRowMeta.clone(); meta.getFields( data.outputRowMeta, getStepname(), null, null, this, repository, metaStore ); data.counts = new long[ meta.getSubjectField().length ]; data.subjectnrs = new int[ meta.getSubjectField().length ]; data.cumulativeSumSourceIndexes = new ArrayList<>(); data.cumulativeSumTargetIndexes = new ArrayList<>(); data.cumulativeAvgSourceIndexes = new ArrayList<>(); data.cumulativeAvgTargetIndexes = new ArrayList<>(); for ( int i = 0; i < meta.getSubjectField().length; i++ ) { if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_COUNT_ANY ) { data.subjectnrs[ i ] = 0; } else { data.subjectnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getSubjectField()[i] ); } if ( ( r != null ) && ( data.subjectnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.AggregateSubjectFieldCouldNotFound", meta.getSubjectField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_SUM ) { data.cumulativeSumSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeSumTargetIndexes.add( data.inputRowMeta.size() + i ); } if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_AVERAGE ) { data.cumulativeAvgSourceIndexes.add( data.subjectnrs[ i ] ); data.cumulativeAvgTargetIndexes.add( data.inputRowMeta.size() + i ); } } data.previousSums = new Object[ data.cumulativeSumTargetIndexes.size() ]; data.previousAvgSum = new Object[ data.cumulativeAvgTargetIndexes.size() ]; data.previousAvgCount = new long[ data.cumulativeAvgTargetIndexes.size() ]; data.groupnrs = new int[ meta.getGroupField().length ]; for ( int i = 0; i < meta.getGroupField().length; i++ ) { data.groupnrs[ i ] = data.inputRowMeta.indexOfValue( meta.getGroupField()[i] ); if ( ( r != null ) && ( data.groupnrs[ i ] < 0 ) ) { logError( BaseMessages.getString( PKG, "GroupBy.Log.GroupFieldCouldNotFound", meta.getGroupField()[ i ] ) ); setErrors( 1 ); stopAll(); return false; } } data.valueMetaInteger = new ValueMetaInteger( "count" ); data.valueMetaNumber = new ValueMetaNumber( "sum" ); initGroupMeta( data.inputRowMeta ); } if ( first || data.newBatch ) { newAggregate( r ); } if ( first ) { data.groupAggMeta = new RowMeta(); data.groupAggMeta.addRowMeta( data.groupMeta ); data.groupAggMeta.addRowMeta( data.aggMeta ); } if ( r == null ) { handleLastOfGroup(); setOutputDone(); return false; } if ( first || data.newBatch ) { first = false; data.newBatch = false; data.previous = data.inputRowMeta.cloneRow( r ); } else { calcAggregate( data.previous ); if ( meta.passAllRows() ) { addToBuffer( data.previous ); } } if ( !sameGroup( data.previous, r ) ) { if ( meta.passAllRows() ) { closeOutput(); data.groupResult = getAggregateResult(); Object[] row = getRowFromBuffer(); long lineNr = 0; while ( row != null ) { int size = data.inputRowMeta.size(); row = RowDataUtil.addRowData( row, size, data.groupResult ); size += data.groupResult.length; lineNr++; if ( meta.isAddingLineNrInGroup() && !Utils.isEmpty( meta.getLineNrInGroupField() ) ) { Object lineNrValue = new Long( lineNr ); row = RowDataUtil.addValueData( row, size, lineNrValue ); size++; } addCumulativeSums( row ); addCumulativeAverages( row ); putRow( data.outputRowMeta, row ); row = getRowFromBuffer(); } closeInput(); } else { Object[] result = buildResult( data.previous ); if ( result != null ) { putRow( data.groupAggMeta, result ); } } newAggregate( r ); } data.previous = data.inputRowMeta.cloneRow( r ); if ( checkFeedback( getLinesRead() ) ) { if ( log.isBasic() ) { logBasic( BaseMessages.getString( PKG, "GroupBy.LineNumber" ) + getLinesRead() ); } } return true; } GroupBy( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
      Trans trans ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); @Override boolean init( StepMetaInterface smi, StepDataInterface sdi ); @Override void dispose( StepMetaInterface smi, StepDataInterface sdi ); @Override void batchComplete(); GroupByMeta getMeta();  }