private boolean readLookupValues() throws KettleException {
    data.infoStream = meta.getStepIOMeta().getInfoStreams().get( 0 );
    if ( data.infoStream.getStepMeta() == null ) {
      logError( BaseMessages.getString( PKG, "FuzzyMatch.Log.NoLookupStepSpecified" ) );
      return false;
    }

    if ( isDetailed() ) {
      logDetailed( BaseMessages.getString( PKG, "FuzzyMatch.Log.ReadingFromStream" )
        + data.infoStream.getStepname() + "]" );
    }

    boolean firstRun = true;
    // Which row set do we read from?
    //
    RowSet rowSet = findInputRowSet( data.infoStream.getStepname() );
    Object[] rowData = getRowFrom( rowSet ); // rows are originating from "lookup_from"

    while ( rowData != null ) {
      if ( firstRun ) {
        data.infoMeta = rowSet.getRowMeta().clone();
        // Check lookup field
        int indexOfLookupField = data.infoMeta.indexOfValue( environmentSubstitute( meta.getLookupField() ) );
        if ( indexOfLookupField < 0 ) {
          // The field is unreachable !
          throw new KettleException( BaseMessages.getString(
            PKG, "FuzzyMatch.Exception.CouldnotFindLookField", meta.getLookupField() ) );
        }
        data.infoCache = new RowMeta();
        ValueMetaInterface keyValueMeta = data.infoMeta.getValueMeta( indexOfLookupField );
        keyValueMeta.setStorageType( ValueMetaInterface.STORAGE_TYPE_NORMAL );
        data.infoCache.addValueMeta( keyValueMeta );
        // Add key
        data.indexOfCachedFields[0] = indexOfLookupField;

        // Check additional fields
        if ( data.addAdditionalFields ) {
          ValueMetaInterface additionalFieldValueMeta;
          for ( int i = 0; i < meta.getValue().length; i++ ) {
            int fi = i + 1;
            data.indexOfCachedFields[fi] = data.infoMeta.indexOfValue( meta.getValue()[i] );
            if ( data.indexOfCachedFields[fi] < 0 ) {
              // The field is unreachable !
              throw new KettleException( BaseMessages.getString(
                PKG, "FuzzyMatch.Exception.CouldnotFindLookField", meta.getValue()[i] ) );
            }
            additionalFieldValueMeta = data.infoMeta.getValueMeta( data.indexOfCachedFields[fi] );
            additionalFieldValueMeta.setStorageType( ValueMetaInterface.STORAGE_TYPE_NORMAL );
            data.infoCache.addValueMeta( additionalFieldValueMeta );
          }
          data.nrCachedFields += meta.getValue().length;
        }
      }
      if ( log.isRowLevel() ) {
        logRowlevel( BaseMessages.getString( PKG, "FuzzyMatch.Log.ReadLookupRow" )
          + rowSet.getRowMeta().getString( rowData ) );
      }

      // Look up the keys in the source rows
      // and store values in cache

      Object[] storeData = new Object[data.nrCachedFields];
      // Add key field
      if ( rowData[data.indexOfCachedFields[0]] == null ) {
        storeData[0] = "";
      } else {
        ValueMetaInterface fromStreamRowMeta = rowSet.getRowMeta().getValueMeta( data.indexOfCachedFields[0] );
        if ( fromStreamRowMeta.isStorageBinaryString() ) {
          storeData[0] = fromStreamRowMeta.convertToNormalStorageType( rowData[data.indexOfCachedFields[0]] );
        } else {
          storeData[0] = rowData[data.indexOfCachedFields[0]];
        }
      }

      // Add additional fields?
      for ( int i = 1; i < data.nrCachedFields; i++ ) {
        ValueMetaInterface fromStreamRowMeta = rowSet.getRowMeta().getValueMeta( data.indexOfCachedFields[i] );
        if ( fromStreamRowMeta.isStorageBinaryString() ) {
          storeData[i] = fromStreamRowMeta.convertToNormalStorageType( rowData[data.indexOfCachedFields[i]] );
        } else {
          storeData[i] = rowData[data.indexOfCachedFields[i]];
        }
      }
      if ( isDebug() ) {
        logDebug( BaseMessages.getString( PKG, "FuzzyMatch.Log.AddingValueToCache", data.infoCache
          .getString( storeData ) ) );
      }

      addToCache( storeData );

      rowData = getRowFrom( rowSet );

      if ( firstRun ) {
        firstRun = false;
      }
    }

    return true;
  }
@Test
  public void testReadLookupValues() throws Exception {
    FuzzyMatchData data = spy( new FuzzyMatchData() );
    data.indexOfCachedFields = new int[2];
    data.minimalDistance = 0;
    data.maximalDistance = 5;
    FuzzyMatchMeta meta = spy( new FuzzyMatchMeta() );
    meta.setOutputMatchField( "I don't want NPE here!" );
    data.readLookupValues = true;
    fuzzyMatch =
        new FuzzyMatchHandler( mockHelper.stepMeta, mockHelper.stepDataInterface, 0, mockHelper.transMeta,
            mockHelper.trans );

    fuzzyMatch.init( meta, data );
    RowSet lookupRowSet = mockHelper.getMockInputRowSet( binaryLookupRows );
    fuzzyMatch.addRowSetToInputRowSets( mockHelper.getMockInputRowSet( binaryRows ) );
    fuzzyMatch.addRowSetToInputRowSets( lookupRowSet );
    fuzzyMatch.rowset = lookupRowSet;

    RowMetaInterface rowMetaInterface = new RowMeta();
    ValueMetaInterface valueMeta = new ValueMetaString( "field1" );
    valueMeta.setStorageMetadata( new ValueMetaString( "field1" ) );
    valueMeta.setStorageType( ValueMetaInterface.STORAGE_TYPE_BINARY_STRING );
    rowMetaInterface.addValueMeta( valueMeta );
    when( lookupRowSet.getRowMeta() ).thenReturn( rowMetaInterface );
    when( meta.getLookupField() ).thenReturn( "field1" );
    when( meta.getMainStreamField() ).thenReturn( "field1" );
    fuzzyMatch.setInputRowMeta( rowMetaInterface.clone() );

    when( meta.getAlgorithmType() ).thenReturn( 1 );
    StepIOMetaInterface stepIOMetaInterface = mock( StepIOMetaInterface.class );
    when( meta.getStepIOMeta() ).thenReturn( stepIOMetaInterface );
    StreamInterface streamInterface = mock( StreamInterface.class );
    List<StreamInterface> streamInterfaceList = new ArrayList<StreamInterface>();
    streamInterfaceList.add( streamInterface );
    when( streamInterface.getStepMeta() ).thenReturn( mockHelper.stepMeta );

    when( stepIOMetaInterface.getInfoStreams() ).thenReturn( streamInterfaceList );

    fuzzyMatch.processRow( meta, data );
    Assert.assertEquals( rowMetaInterface.getString( row3B, 0 ),
        data.outputRowMeta.getString( fuzzyMatch.resultRow, 1 ) );
  }