public void getFields( RowMetaInterface row, String origin, RowMetaInterface[] info, StepMeta nextStep,
    VariableSpace space, Repository repository, IMetaStore metaStore ) throws KettleStepException {
    if ( databaseMeta == null ) {
      return; // TODO: throw an exception here
    }

    if ( cachedRowMetaActive ) {
      row.addRowMeta( cachedRowMeta );
      return;
    }

    boolean param = false;

    Database db = getDatabase();
    super.databases = new Database[] { db }; // keep track of it for canceling purposes...

    // First try without connecting to the database... (can be S L O W)
    String sNewSQL = sql;
    if ( isVariableReplacementActive() ) {
      sNewSQL = db.environmentSubstitute( sql );
      if ( space != null ) {
        sNewSQL = space.environmentSubstitute( sNewSQL );
      }
    }

    RowMetaInterface add = null;
    try {
      add = db.getQueryFields( sNewSQL, param );
    } catch ( KettleDatabaseException dbe ) {
      throw new KettleStepException( "Unable to get queryfields for SQL: " + Const.CR + sNewSQL, dbe );
    }

    if ( add != null ) {
      attachOrigin( add, origin );
      row.addRowMeta( add );
    } else {
      try {
        db.connect();

        RowMetaInterface paramRowMeta = null;
        Object[] paramData = null;

        StreamInterface infoStream = getStepIOMeta().getInfoStreams().get( 0 );
        if ( !Utils.isEmpty( infoStream.getStepname() ) ) {
          param = true;
          if ( info.length > 0 && info[ 0 ] != null ) {
            paramRowMeta = info[ 0 ];
            paramData = RowDataUtil.allocateRowData( paramRowMeta.size() );
          }
        }

        add = db.getQueryFields( sNewSQL, param, paramRowMeta, paramData );

        if ( add == null ) {
          return;
        }
        attachOrigin( add, origin );
        row.addRowMeta( add );
      } catch ( KettleException ke ) {
        throw new KettleStepException( "Unable to get queryfields for SQL: " + Const.CR + sNewSQL, ke );
      } finally {
        db.disconnect();
      }
    }
    if ( isLazyConversionActive() ) {
      for ( int i = 0; i < row.size(); i++ ) {
        ValueMetaInterface v = row.getValueMeta( i );
        try {
          if ( v.getType() == ValueMetaInterface.TYPE_STRING ) {
            ValueMetaInterface storageMeta = ValueMetaFactory.cloneValueMeta( v );
            storageMeta.setStorageType( ValueMetaInterface.STORAGE_TYPE_NORMAL );
            v.setStorageMetadata( storageMeta );
            v.setStorageType( ValueMetaInterface.STORAGE_TYPE_BINARY_STRING );
          }
        } catch ( KettlePluginException e ) {
          throw new KettleStepException( "Unable to clone meta for lazy conversion: " + Const.CR + v, e );
        }
      }
    }
  }
@Test
  public void testGetFields() throws Exception {
    TableInputMetaHandler meta = new TableInputMetaHandler();
    meta.setLazyConversionActive( true );
    DatabaseMeta dbMeta = mock( DatabaseMeta.class );
    meta.setDatabaseMeta( dbMeta );
    Database mockDB = meta.getDatabase();
    when( mockDB.getQueryFields( anyString(), anyBoolean() ) ).thenReturn( createMockFields() );

    RowMetaInterface expectedRowMeta = new RowMeta();
    ValueMetaInterface valueMeta = new ValueMetaString( "field1" );
    valueMeta.setStorageMetadata( new ValueMetaString( "field1" ) );
    valueMeta.setStorageType( ValueMetaInterface.STORAGE_TYPE_BINARY_STRING );
    expectedRowMeta.addValueMeta( valueMeta );

    VariableSpace space = mock( VariableSpace.class );
    RowMetaInterface rowMetaInterface = new RowMeta();
    meta.getFields( rowMetaInterface, "TABLE_INPUT_META", null, null, space, null, null );

    verify( mockDB).getQueryFields( any(), anyBoolean() ) ;
  }