@Override
  public boolean checkIndexExists( Database database, String schemaName, String tableName, String[] idxFields ) throws KettleDatabaseException {

    String tablename = database.getDatabaseMeta().getQuotedSchemaTableCombination( schemaName, tableName );

    boolean[] exists = new boolean[ idxFields.length];
    for ( int i = 0; i < exists.length; i++ ) {
      exists[i] = false;
    }

    try {
      //
      // Get the info from the data dictionary...
      //
      StringBuilder sql = new StringBuilder( 128 );
      sql.append( "select i.name table_name, c.name column_name " );
      sql.append( "from     sysindexes i, sysindexkeys k, syscolumns c " );
      sql.append( "where    i.name = '" + tablename + "' " );
      sql.append( "AND      i.id = k.id " );
      sql.append( "AND      i.id = c.id " );
      sql.append( "AND      k.colid = c.colid " );

      ResultSet res = null;
      try {
        res = database.openQuery( sql.toString() );
        if ( res != null ) {
          Object[] row = database.getRow( res );
          while ( row != null ) {
            String column = database.getReturnRowMeta().getString( row, "column_name", "" );
            int idx = Const.indexOfString( column, idxFields );
            if ( idx >= 0 ) {
              exists[idx] = true;
            }

            row = database.getRow( res );
          }
        } else {
          return false;
        }
      } finally {
        if ( res != null ) {
          database.closeQuery( res );
        }
      }

      // See if all the fields are indexed...
      boolean all = true;
      for ( int i = 0; i < exists.length && all; i++ ) {
        if ( !exists[i] ) {
          all = false;
        }
      }

      return all;
    } catch ( Exception e ) {
      throw new KettleDatabaseException( "Unable to determine if indexes exists on table [" + tablename + "]", e );
    }
  }
@Test
  public void testCheckIndexExists() throws Exception {
    String expectedSQL = "select i.name table_name, c.name column_name from     sysindexes i, sysindexkeys k, syscolumns c where    i.name = 'FOO' AND      i.id = k.id AND      i.id = c.id AND      k.colid = c.colid "; // yes, space at the end like in the dbmeta
    Database db = Mockito.mock(  Database.class );
    RowMetaInterface rm = Mockito.mock( RowMetaInterface.class );
    ResultSet rs = Mockito.mock( ResultSet.class );
    DatabaseMeta dm = Mockito.mock( DatabaseMeta.class );
    Mockito.when( dm.getQuotedSchemaTableCombination( "", "FOO" ) ).thenReturn( "FOO" );
    Mockito.when( rs.next() ).thenReturn( rowCnt < 2 );
    Mockito.when( db.openQuery( expectedSQL ) ).thenReturn( rs );
    Mockito.when( db.getReturnRowMeta() ).thenReturn( rm );
    Mockito.when( rm.getString( row1, "column_name", "" ) ).thenReturn( "ROW1COL2" );
    Mockito.when( rm.getString( row2, "column_name", "" ) ).thenReturn( "ROW2COL2" );
    Mockito.when( db.getRow( rs ) ).thenAnswer( new Answer<Object[]>() {
        @Override
        public Object[] answer( InvocationOnMock invocation ) throws Throwable {
          rowCnt++;
          if ( rowCnt == 1 ) {
            return row1;
          } else if ( rowCnt == 2 ) {
            return row2;
          } else {
            return null;
          }
        }
    } );
    Mockito.when(  db.getDatabaseMeta() ).thenReturn( dm );
    assertTrue( nativeMeta.checkIndexExists( db, "", "FOO", new String[] { "ROW1COL2", "ROW2COL2" } ) );
    assertFalse( nativeMeta.checkIndexExists( db, "", "FOO", new String[] { "ROW2COL2", "NOTTHERE" } ) );
    assertFalse( nativeMeta.checkIndexExists( db, "", "FOO", new String[] { "NOTTHERE", "ROW1COL2" } ) );

  }