public String getControlFileContents( GPBulkLoaderMeta meta, RowMetaInterface rm, Object[] r ) throws KettleException {
    DatabaseMeta dm = meta.getDatabaseMeta();
    String inputName = "'" + environmentSubstitute( meta.getDataFile() ) + "'";

    // if ( GPBulkLoaderMeta.METHOD_AUTO_CONCURRENT.equals(meta.getLoadMethod()) )
    // {
    // // if loading is concurrent, the filename has to be a * as sqlldr will
    // // read from stdin.
    // inputName = "*";
    // }

    String loadAction = meta.getLoadAction();

    StringBuffer contents = new StringBuffer( 500 );

    String tableName =
      dm.getQuotedSchemaTableCombination(
        environmentSubstitute( meta.getSchemaName() ), environmentSubstitute( meta.getTableName() ) );

    // Create a Postgres / Greenplum COPY string for use with a psql client
    if ( loadAction.equalsIgnoreCase( "truncate" ) ) {
      contents.append( loadAction + " " );
      contents.append( tableName + ";" );
      contents.append( Const.CR );
    }
    contents.append( "\\COPY " );
    // Table name

    contents.append( tableName );

    // Names of columns

    contents.append( " ( " );

    String[] streamFields = meta.getFieldStream();
    String[] tableFields = meta.getFieldTable();

    if ( streamFields == null || streamFields.length == 0 ) {
      throw new KettleException( "No fields defined to load to database" );
    }

    for ( int i = 0; i < streamFields.length; i++ ) {
      if ( i != 0 ) {
        contents.append( ", " );
      }
      contents.append( dm.quoteField( tableFields[i] ) );
    }

    contents.append( " ) " );

    // The "FROM" filename
    contents.append( " FROM " );
    contents.append( inputName );

    // The "FORMAT" clause
    contents.append( " WITH CSV " );

    // The single row error handling
    contents.append( "LOG ERRORS INTO " );
    contents.append( tableName + "_errors " );
    contents.append( " SEGMENT REJECT LIMIT " );
    contents.append( meta.getMaxErrors() );

    // contents.append("  ERRORS=\'").append(meta.getMaxErrors()).append("\'").append(Const.CR);

    // contents.append("LOAD DATA").append(Const.CR).append(
    // "INFILE ").append(inputName).append(Const.CR).append(
    // "INTO TABLE ").append(dm.getQuotedSchemaTableCombination(environmentSubstitute(meta.getSchemaName()),
    // environmentSubstitute(meta.getTableName()))).append(
    // Const.CR).append(loadAction).append(Const.CR).append(
    // "FIELDS TERMINATED BY ',' ENCLOSED BY '\"'").append(Const.CR).append(
    // "(");
    //
    // String streamFields[] = meta.getFieldStream();
    // String tableFields[] = meta.getFieldTable();
    // String dateMask[] = meta.getDateMask();
    //
    // if ( streamFields == null || streamFields.length == 0 )
    // {
    // throw new KettleException("No fields defined to load to database");
    // }
    //
    // for (int i = 0; i < streamFields.length; i++)
    // {
    // if ( i!=0 ) contents.append(", ").append(Const.CR);
    // contents.append(dm.quoteField(tableFields[i]));
    //
    // int pos = rm.indexOfValue(streamFields[i]);
    // if (pos<0)
    // {
    // throw new KettleException("Could not find field " +
    // streamFields[i] + " in stream");
    // }
    // ValueMetaInterface v = rm.getValueMeta(pos);
    // switch ( v.getType() )
    // {
    // case ValueMetaInterface.TYPE_STRING:
    // if ( v.getLength() > 255 )
    // {
    // contents.append(" CHAR(").append(v.getLength()).append(")");
    // }
    // else
    // {
    // contents.append(" CHAR");
    // }
    // break;
    // case ValueMetaInterface.TYPE_INTEGER:
    // case ValueMetaInterface.TYPE_NUMBER:
    // case ValueMetaInterface.TYPE_BIGNUMBER:
    // break;
    // case ValueMetaInterface.TYPE_DATE:
    // if ( GPBulkLoaderMeta.DATE_MASK_DATE.equals(dateMask[i]) )
    // {
    // contents.append(" DATE 'yyyy-mm-dd'");
    // }
    // else if ( GPBulkLoaderMeta.DATE_MASK_DATETIME.equals(dateMask[i]) )
    // {
    // contents.append(" TIMESTAMP 'yyyy-mm-dd hh24:mi:ss.ff'");
    // }
    // else
    // {
    // // If not specified the default is date.
    // contents.append(" DATE 'yyyy-mm-dd'");
    // }
    // break;
    // case ValueMetaInterface.TYPE_BINARY:
    // contents.append(" ENCLOSED BY '<startlob>' AND '<endlob>'");
    // break;
    // }
    // }
    // contents.append(")");

    // if ( GPBulkLoaderMeta.METHOD_AUTO_CONCURRENT.equals(meta.getLoadMethod()) )
    // {
    // contents.append(Const.CR).append("BEGINDATA").append(Const.CR);
    // }

    return contents.toString();
  }
@Test
  public void testInputFileSurroundedBySingleQuotes() throws Exception {

    String datafile = "test-data-file";

    loader =
      new GPBulkLoader( mockHelper.stepMeta, mockHelper.stepDataInterface, 0, mockHelper.transMeta, mockHelper.trans );

    DatabaseMeta dbMetaMock = mock( DatabaseMeta.class );
    doReturn( "" ).when( dbMetaMock ).getQuotedSchemaTableCombination( anyString(), anyString() );
    doReturn( "" ).when( dbMetaMock ).quoteField( anyString() );

    GPBulkLoaderMeta meta = new GPBulkLoaderMeta();
    meta.setLoadAction( "" );
    meta.setFieldStream( new String[] { "" } );
    meta.setFieldTable( new String[] { "" } );
    meta.setDatabaseMeta( dbMetaMock );
    meta.setDataFile( datafile );

    String actual = loader.getControlFileContents( meta, null, null );

    int first = actual.indexOf( datafile );
    if ( first > 0 ) {
      if ( actual.charAt( first - 1 ) != '\'' || actual.charAt( first + datafile.length() ) != '\'' ) {
        Assert.fail( "Datafile name is not surrounded by single quotes. Actual control file: " + actual );
      }
    } else {
      Assert.fail( "Datafile name not found in control file. Actual control file: " + actual );
    }

  }