class GroupBy extends BaseStep implements StepInterface {

  @Override
  public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {
    meta = (GroupByMeta) smi;
    data = (GroupByData) sdi;

    Object[] r = getRow(); // get row!

    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 );

      // What is the output looking like?
      //
      data.inputRowMeta = getInputRowMeta();

      // In case we have 0 input rows, we still want to send out a single row aggregate
      // However... the problem then is that we don't know the layout from receiving it from the previous step over the
      // row set.
      // So we need to calculated based on the metadata...
      //
      if ( data.inputRowMeta == null ) {
        data.inputRowMeta = getTransMeta().getPrevStepFields( getStepMeta() );
      }

      data.outputRowMeta = data.inputRowMeta.clone();
      meta.getFields( data.outputRowMeta, getStepname(), null, null, this, repository, metaStore );

      // Do all the work we can beforehand
      // Calculate indexes, loop up fields, etc.
      //
      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 ] );

          // The position of the target in the output row is the input row size + i
          //
          data.cumulativeSumTargetIndexes.add( data.inputRowMeta.size() + i );
        }
        if ( meta.getAggregateType()[ i ] == GroupByMeta.TYPE_GROUP_CUMULATIVE_AVERAGE ) {
          data.cumulativeAvgSourceIndexes.add( data.subjectnrs[ i ] );

          // The position of the target in the output row is the input row size + 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;
        }
      }

      // Create a metadata value for the counter Integers
      //
      data.valueMetaInteger = new ValueMetaInteger( "count" );
      data.valueMetaNumber = new ValueMetaNumber( "sum" );

      // Initialize the group metadata
      //
      initGroupMeta( data.inputRowMeta );
    }

    if ( first || data.newBatch ) {
      // Create a new group aggregate (init)
      //
      newAggregate( r );
    }

    if ( first ) {
      // for speed: groupMeta+aggMeta
      //
      data.groupAggMeta = new RowMeta();
      data.groupAggMeta.addRowMeta( data.groupMeta );
      data.groupAggMeta.addRowMeta( data.aggMeta );
    }

    if ( r == null ) { // no more input to be expected... (or none received in the first place)
      handleLastOfGroup();
      setOutputDone();
      return false;
    }

    if ( first || data.newBatch ) {
      first = false;
      data.newBatch = false;

      data.previous = data.inputRowMeta.cloneRow( r ); // copy the row to previous
    } else {
      calcAggregate( data.previous );

      if ( meta.passAllRows() ) {
        addToBuffer( data.previous );
      }
    }

    if ( !sameGroup( data.previous, r ) ) {
      if ( meta.passAllRows() ) {
        // Not the same group: close output (if any)
        closeOutput();
        // Get all rows from the buffer!
        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 );
            // ValueMetaInterface lineNrValueMeta = new ValueMeta(meta.getLineNrInGroupField(),
            // ValueMetaInterface.TYPE_INTEGER);
            // lineNrValueMeta.setLength(9);
            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 ); // copy row to possible alternate rowset(s).
        }
      }
      newAggregate( r ); // Create a new group aggregate (init)
    }

    data.previous = data.inputRowMeta.cloneRow( r );

    if ( checkFeedback( getLinesRead() ) ) {
      if ( log.isBasic() ) {
        logBasic( BaseMessages.getString( PKG, "GroupBy.LineNumber" ) + getLinesRead() );
      }
    }

    return true;
  }

  public  GroupBy( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
      Trans trans );

}