public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {
    meta = (SelectValuesMeta) smi;
    data = (SelectValuesData) sdi;

    Object[] rowData = getRow(); // get row from rowset, wait for our turn, indicate busy!
    if ( rowData == null ) { // no more input to be expected...

      setOutputDone();
      return false;
    }

    Object[] rowCopy = null;
    if ( getStepMeta().isDoingErrorHandling() ) {
      rowCopy = getInputRowMeta().cloneRow( rowData );
    }

    if ( log.isRowLevel() ) {
      logRowlevel( BaseMessages.getString( PKG, "SelectValues.Log.GotRowFromPreviousStep" )
        + getInputRowMeta().getString( rowData ) );
    }

    if ( first ) {
      first = false;

      data.selectRowMeta = getInputRowMeta().clone();
      meta.getSelectFields( data.selectRowMeta, getStepname() );
      data.deselectRowMeta = data.selectRowMeta.clone();
      meta.getDeleteFields( data.deselectRowMeta );
      data.metadataRowMeta = data.deselectRowMeta.clone();
      meta.getMetadataFields( data.metadataRowMeta, getStepname(), this );
    }

    try {
      Object[] outputData = rowData;

      if ( data.select ) {
        outputData = selectValues( getInputRowMeta(), outputData );
      }
      if ( data.deselect ) {
        outputData = removeValues( data.selectRowMeta, outputData );
      }
      if ( data.metadata ) {
        outputData = metadataValues( data.deselectRowMeta, outputData );
      }

      if ( outputData == null ) {
        setOutputDone(); // signal end to receiver(s)
        return false;
      }

      // Send the row on its way
      //
      putRow( data.metadataRowMeta, outputData );
      if ( log.isRowLevel() ) {
        logRowlevel( BaseMessages.getString( PKG, "SelectValues.Log.WroteRowToNextStep" )
          + data.metadataRowMeta.getString( outputData ) );
      }

    } catch ( KettleException e ) {
      if ( getStepMeta().isDoingErrorHandling() ) {
        String field;
        if ( e instanceof KettleConversionException ) {
          List<ValueMetaInterface> fields = ( (KettleConversionException) e ).getFields();
          field = fields.isEmpty() ? null : fields.get( 0 ).getName();
        } else {
          field = null;
        }
        putError( getInputRowMeta(), rowCopy, 1, e.getMessage(), field, "SELECT001" );
      } else {
        throw e;
      }
    }

    if ( checkFeedback( getLinesRead() ) ) {
      logBasic( BaseMessages.getString( PKG, "SelectValues.Log.LineNumber" ) + getLinesRead() );
    }

    return true;
  }
@Test
  public void testPDI16368() throws Exception {
    // This tests that the fix for PDI-16388 doesn't get re-broken.
    //

    SelectValuesHandler  step2;
    Object[] inputRow2;
    RowMeta inputRowMeta;
    SelectValuesMeta stepMeta;
    SelectValuesData stepData;
    ValueMetaInterface vmi;
    // First, test current behavior (it's worked this way since 5.x or so)
    //
    step2 = new SelectValuesHandler( helper.stepMeta, helper.stepDataInterface, 1, helper.transMeta, helper.trans );
    step2 = spy( step2 );
    inputRow2 = new Object[] { new BigDecimal( "589" ) }; // Starting with a BigDecimal (no places)
    doReturn( inputRow2 ).when( step2 ).getRow();
    doNothing().when( step2 )
        .putError( any( RowMetaInterface.class ), any( Object[].class ), anyLong(), anyString(), anyString(),
          anyString() );

    inputRowMeta = new RowMeta();
    inputRowMeta.addValueMeta( new ValueMetaBigNumber( SELECTED_FIELD ) );
    step2.setInputRowMeta( inputRowMeta );
    stepMeta = new SelectValuesMeta();
    stepMeta.allocate( 1, 0, 1 );
    stepMeta.getSelectFields()[0] = new SelectField();
    stepMeta.getSelectFields()[0].setName( SELECTED_FIELD );
    stepMeta.getMeta()[ 0 ] =
      new SelectMetadataChange( stepMeta, SELECTED_FIELD, null, ValueMetaInterface.TYPE_INTEGER, -2, -2,
        ValueMetaInterface.STORAGE_TYPE_NORMAL, null, false, null, null, false, null, null, null ); // no specified conversion type so should have default conversion mask.

    stepData = new SelectValuesData();
    stepData.select = true;
    stepData.metadata = true;
    stepData.firstselect = true;
    stepData.firstmetadata = true;
    step2.processRow( stepMeta, stepData );

    vmi = step2.rowMeta.getValueMeta( 0 );
    assertEquals( ValueMetaBase.DEFAULT_BIG_NUMBER_FORMAT_MASK, vmi.getConversionMask() );

    step2 = new SelectValuesHandler( helper.stepMeta, helper.stepDataInterface, 1, helper.transMeta, helper.trans );
    step2 = spy( step2 );
    doReturn( inputRow2 ).when( step2 ).getRow();
    doNothing().when( step2 )
        .putError( any( RowMetaInterface.class ), any( Object[].class ), anyLong(), anyString(), anyString(),
          anyString() );

    inputRowMeta = new RowMeta();
    inputRowMeta.addValueMeta( new ValueMetaBigNumber( SELECTED_FIELD ) );
    step2.setInputRowMeta( inputRowMeta );
    stepMeta = new SelectValuesMeta();
    stepMeta.allocate( 1, 0, 1 );
    stepMeta.getSelectFields()[0] = new SelectField();
    stepMeta.getSelectFields()[0].setName( SELECTED_FIELD );
    stepMeta.getMeta()[ 0 ] =
      new SelectMetadataChange( stepMeta, SELECTED_FIELD, null, ValueMetaInterface.TYPE_NUMBER, -2, -2,
        ValueMetaInterface.STORAGE_TYPE_NORMAL, null, false, null, null, false, null, null, null ); // no specified conversion type so should have default conversion mask for Double.

    stepData = new SelectValuesData();
    stepData.select = true;
    stepData.metadata = true;
    stepData.firstselect = true;
    stepData.firstmetadata = true;
    step2.processRow( stepMeta, stepData );

    vmi = step2.rowMeta.getValueMeta( 0 );
    assertEquals( ValueMetaBase.DEFAULT_BIG_NUMBER_FORMAT_MASK, vmi.getConversionMask() );


    step2 = new SelectValuesHandler( helper.stepMeta, helper.stepDataInterface, 1, helper.transMeta, helper.trans );
    step2 = spy( step2 );
    inputRow2 = new Object[] { 589L }; // Starting with a Long
    doReturn( inputRow2 ).when( step2 ).getRow();
    doNothing().when( step2 )
        .putError( any( RowMetaInterface.class ), any( Object[].class ), anyLong(), anyString(), anyString(),
          anyString() );

    inputRowMeta = new RowMeta();
    inputRowMeta.addValueMeta( new ValueMetaInteger( SELECTED_FIELD ) );
    step2.setInputRowMeta( inputRowMeta );
    stepMeta = new SelectValuesMeta();
    stepMeta.allocate( 1, 0, 1 );
    stepMeta.getSelectFields()[0] = new SelectField();
    stepMeta.getSelectFields()[0].setName( SELECTED_FIELD );
    // no specified conversion type so should have default conversion mask for BigNumber
    stepMeta.getMeta()[ 0 ] =
      new SelectMetadataChange( stepMeta, SELECTED_FIELD, null, ValueMetaInterface.TYPE_BIGNUMBER, -2, -2,
        ValueMetaInterface.STORAGE_TYPE_NORMAL, null, false, null, null, false, null, null, null );

    stepData = new SelectValuesData();
    stepData.select = true;
    stepData.metadata = true;
    stepData.firstselect = true;
    stepData.firstmetadata = true;
    step2.processRow( stepMeta, stepData );

    vmi = step2.rowMeta.getValueMeta( 0 );
    assertEquals( ValueMetaBase.DEFAULT_INTEGER_FORMAT_MASK, vmi.getConversionMask() );

    // Now, test that setting the variable results in getting the default conversion mask
    step2 = new SelectValuesHandler( helper.stepMeta, helper.stepDataInterface, 1, helper.transMeta, helper.trans );
    step2.setVariable( Const.KETTLE_COMPATIBILITY_SELECT_VALUES_TYPE_CHANGE_USES_TYPE_DEFAULTS, "Y" );
    step2 = spy( step2 );
    inputRow2 = new Object[] { new BigDecimal( "589" ) }; // Starting with a BigDecimal (no places)
    doReturn( inputRow2 ).when( step2 ).getRow();
    doNothing().when( step2 )
        .putError( any( RowMetaInterface.class ), any( Object[].class ), anyLong(), anyString(), anyString(),
          anyString() );

    inputRowMeta = new RowMeta();
    inputRowMeta.addValueMeta( new ValueMetaBigNumber( SELECTED_FIELD ) );
    step2.setInputRowMeta( inputRowMeta );
    stepMeta = new SelectValuesMeta();
    stepMeta.allocate( 1, 0, 1 );
    stepMeta.getSelectFields()[0] = new SelectField();
    stepMeta.getSelectFields()[0].setName( SELECTED_FIELD );
    stepMeta.getMeta()[ 0 ] =
      new SelectMetadataChange( stepMeta, SELECTED_FIELD, null, ValueMetaInterface.TYPE_INTEGER, -2, -2,
        ValueMetaInterface.STORAGE_TYPE_NORMAL, null, false, null, null, false, null, null, null ); // no specified conversion type so should have default conversion mask.

    stepData = new SelectValuesData();
    stepData.select = true;
    stepData.metadata = true;
    stepData.firstselect = true;
    stepData.firstmetadata = true;
    step2.processRow( stepMeta, stepData );

    vmi = step2.rowMeta.getValueMeta( 0 );
    assertEquals( ValueMetaBase.DEFAULT_INTEGER_FORMAT_MASK, vmi.getConversionMask() );

    step2 = new SelectValuesHandler( helper.stepMeta, helper.stepDataInterface, 1, helper.transMeta, helper.trans );
    step2.setVariable( Const.KETTLE_COMPATIBILITY_SELECT_VALUES_TYPE_CHANGE_USES_TYPE_DEFAULTS, "Y" );
    step2 = spy( step2 );
    doReturn( inputRow2 ).when( step2 ).getRow();
    doNothing().when( step2 )
        .putError( any( RowMetaInterface.class ), any( Object[].class ), anyLong(), anyString(), anyString(),
          anyString() );

    inputRowMeta = new RowMeta();
    inputRowMeta.addValueMeta( new ValueMetaBigNumber( SELECTED_FIELD ) );
    step2.setInputRowMeta( inputRowMeta );
    stepMeta = new SelectValuesMeta();
    stepMeta.allocate( 1, 0, 1 );
    stepMeta.getSelectFields()[0] = new SelectField();
    stepMeta.getSelectFields()[0].setName( SELECTED_FIELD );
    stepMeta.getMeta()[ 0 ] =
      new SelectMetadataChange( stepMeta, SELECTED_FIELD, null, ValueMetaInterface.TYPE_NUMBER, -2, -2,
        ValueMetaInterface.STORAGE_TYPE_NORMAL, null, false, null, null, false, null, null, null ); // no specified conversion type so should have default conversion mask for Double.

    stepData = new SelectValuesData();
    stepData.select = true;
    stepData.metadata = true;
    stepData.firstselect = true;
    stepData.firstmetadata = true;
    step2.processRow( stepMeta, stepData );

    vmi = step2.rowMeta.getValueMeta( 0 );
    assertEquals( ValueMetaBase.DEFAULT_NUMBER_FORMAT_MASK, vmi.getConversionMask() );


    step2 = new SelectValuesHandler( helper.stepMeta, helper.stepDataInterface, 1, helper.transMeta, helper.trans );
    step2.setVariable( Const.KETTLE_COMPATIBILITY_SELECT_VALUES_TYPE_CHANGE_USES_TYPE_DEFAULTS, "Y" );
    step2 = spy( step2 );
    inputRow2 = new Object[] { 589L }; // Starting with a Long
    doReturn( inputRow2 ).when( step2 ).getRow();
    doNothing().when( step2 )
        .putError( any( RowMetaInterface.class ), any( Object[].class ), anyLong(), anyString(), anyString(),
          anyString() );

    inputRowMeta = new RowMeta();
    inputRowMeta.addValueMeta( new ValueMetaInteger( SELECTED_FIELD ) );
    step2.setInputRowMeta( inputRowMeta );
    stepMeta = new SelectValuesMeta();
    stepMeta.allocate( 1, 0, 1 );
    stepMeta.getSelectFields()[0] = new SelectField();
    stepMeta.getSelectFields()[0].setName( SELECTED_FIELD );
    // no specified conversion type so should have default conversion mask for BigNumber
    stepMeta.getMeta()[ 0 ] =
      new SelectMetadataChange( stepMeta, SELECTED_FIELD, null, ValueMetaInterface.TYPE_BIGNUMBER, -2, -2,
        ValueMetaInterface.STORAGE_TYPE_NORMAL, null, false, null, null, false, null, null, null );

    stepData = new SelectValuesData();
    stepData.select = true;
    stepData.metadata = true;
    stepData.firstselect = true;
    stepData.firstmetadata = true;
    step2.processRow( stepMeta, stepData );

    vmi = step2.rowMeta.getValueMeta( 0 );
    assertEquals( ValueMetaBase.DEFAULT_BIG_NUMBER_FORMAT_MASK, vmi.getConversionMask() );

  }