public interface MetadataFactory extends Closeable {
  /**
   * @return database metadata.
   */
  DatabaseMetaData getMetadata();
} public class CatalogSchema {
  /**
   * Default schema
   */
  public static final String DEFAULT_SCHEMA = "public";

  /**
   * Default catalog
   */
  public static final String DEFAULT_CATALOG = null;

  /**
   * @return default catalog/schema.
   */
  public static CatalogSchema defaultCatalogSchema() {
    return new CatalogSchema(DEFAULT_CATALOG, DEFAULT_SCHEMA);
  }

  private final String m_catalog;
  private final String m_schema;

  /**
   * Construct a new {@link CatalogSchema}.
   * @param catalog catalog.
   * @param schema schema.
   */
  public CatalogSchema(String catalog, String schema) {
    m_catalog = catalog;
    m_schema = schema;
  }

  /**
   * @return the catalog.
   */
  public String getCatalog() {
    return m_catalog;
  }

  /**
   * @return the schema.
   */
  public String getSchema() {
    return m_schema;
  }

  @Override
  public boolean equals(Object obj) {
    if (obj == null) {
      return false;
    } else if (getClass() != obj.getClass()) {
      return false;
    } else {
      CatalogSchema other = (CatalogSchema) obj;
      return Objects.equals(getCatalog(), other.getCatalog()) && Objects.equals(getSchema(), other.getSchema());
    }
  }

  @Override
  public int hashCode() {
    return Objects.hash(getClass(), getCatalog(), getSchema());
  }

  @Override
  public String toString() {
    return "[" + m_catalog + "." + m_schema + "]";
  }
} public class ColumnType {
  private final int m_type;
  private final String m_typeCode;

  /**
   * @param type an integer description of the data type in SQL
   * @param typeCode a String description of the data type in Java
   */
  public ColumnType(final int type, final String typeCode) {
    m_type = type;
    m_typeCode = typeCode;
  }

  /**
   *
   * {@inheritDoc}
   */
  @Override
  public boolean equals(Object t) {
    if (t == null || getClass() != t.getClass()) {
      return false;
    } else {
      ColumnType other = (ColumnType) t;
      return m_type == other.getType() && Objects.equals(m_typeCode, other.getTypeCode());
    }
  }

  /**
   *
   * {@inheritDoc}
   */
  @Override
  public int hashCode() {
    return Objects.hash(m_type, m_typeCode);
  }

  /**
   * @return an integer that defines a data type, the sql description
   */
  public int getType() {
    return m_type;
  }

  /**
   * @return a String that describes a data type, the Java description
   */
  public String getTypeCode() {
    return m_typeCode;
  }
} public enum TableType {
  /** An INDEX-style table */
  INDEX,

  /** A regular user table */
  TABLE,

  /** A View */
  VIEW;

  //"SYSTEM TABLE", "GLOBAL TEMPORARY", "LOCAL TEMPORARY", "ALIAS", and "SYNONYM" not supported.


} public class Column extends NamedSchemaItem implements Comparable<Column> {
  private final String m_table;

  // Column properties
  private String m_default;
  private Boolean m_isNullable;
  private Integer m_columnSize;
  private Integer m_ordinal;
  private ColumnType m_columnType;

  /**
   * Construct a new Column.
   * @param catalogSchema catalog/schema.
   * @param name column name.
   * @param table name of the table that contains the column.
   */
  public Column(CatalogSchema catalogSchema, String name, String table) {
    super(catalogSchema, name);
    m_table = table;
  }

  /**
   * Create a new Column.
   * @param catalog catalog.
   * @param schema schema.
   * @param name column name.
   * @param table name of the table that contains the column.
   */
  public Column(String catalog, String schema, String name, String table) {
    super(catalog, schema, name);
    m_table = table;
  }

  /**
   * Get the table.
   * @return Returns the table
   */
  public String getTable() {
    return m_table;
  }

  /**
   * Set the default.
   * @param defaultVal The default to set
   */
  public void setDefault(String defaultVal) {
    m_default = defaultVal;
  }

  /**
   * Get the default.
   * @return Returns the default
   */
  public String getDefault() {
    return m_default;
  }
  /**
   * Set the isNullable.
   * @param isNullable The isNullable to set
   */
  public void setIsNullable(Boolean isNullable) {
    m_isNullable = isNullable;
  }
  /**
   * Get the isNullable.
   * @return Returns the isNullable
   */
  public Boolean getIsNullable() {
    return m_isNullable;
  }
  /**
   * Set the size of the column.  For char or date types this is the maximum number of characters, for numeric or decimal types
   * this is precision.
   * @param columnSize The columnSize to set
   */
  public void setColumnSize(Integer columnSize) {
    m_columnSize = columnSize;
  }
  /**
   * Get the size of the column.  For char or date types this is the maximum number of characters, for numeric or decimal types
   * this is precision.
   * @return Returns the columnSize
   */
  public Integer getColumnSize() {
    return m_columnSize;
  }
  /**
   * Set the ordinal.
   * @param ordinal The ordinal to set
   */
  public void setOrdinal(Integer ordinal) {
    m_ordinal = ordinal;
  }
  /**
   * Get the ordinal.
   * @return Returns the ordinal
   */
  public Integer getOrdinal() {
    return m_ordinal;
  }

  /**
   * Get the type, corresponds to java.sql.Types
   * @return Returns the type
   */
  public int getType() {
    return m_columnType.getType();
  }

  /**
   * @return the type name (e.g. float4, bigint, varchar(255))
   */
  public String getTypeName() {
    return m_columnType.getTypeCode();
  }

  /**
   * @return the column type object
   */
  public ColumnType getColumnType() {
    return m_columnType;
  }

  /**
   * set the column type
   * @param columnType the value to set
   */
  public void setColumnType(ColumnType columnType) {
    m_columnType = columnType;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean equals(Object o) {
    if (!(o instanceof Column)) {
      return false;
    }
    Column other = (Column)o;
    return getOrdinal().equals(other.getOrdinal()) && getName().equals(other.getName());
  }

  @Override
  public int hashCode() {
    return Objects.hash(m_ordinal, getName());
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public int compareTo(Column o) {
    int ordComp = getOrdinal().compareTo(o.getOrdinal());
    if (ordComp == 0) {
      return getName().compareTo(o.getName());
    }
    return ordComp;
  }
} public class ForeignKey {
  //Constraint name
  private String m_fkName;
  private String m_keySeq;

  //Table and column the FK belongs to, ResultSet 5-8
  private CatalogSchema m_fkCatalogSchema;
  private String m_fkTable;
  private String m_fkColumn;

  //Table and column the FK points to, ResultSet 1-4
  private CatalogSchema m_pkCatalogSchema;
  private String m_pkTable;
  private String m_pkColumn;
  /**
   * Set the fkName.
   * @param fkName The fkName to set
   */
  public void setFkName(String fkName) {
    m_fkName = fkName;
  }
  /**
   * Get the fkName.
   * @return Returns the fkName
   */
  public String getFkName() {
    return m_fkName;
  }
  /**
   * Set the keySeq.
   * @param keySeq The keySeq to set
   */
  public void setKeySeq(String keySeq) {
    m_keySeq = keySeq;
  }
  /**
   * Get the keySeq.
   * @return Returns the keySeq
   */
  public String getKeySeq() {
    return m_keySeq;
  }

  /**
   * Set the fkTable.
   * @param fkTable The fkTable to set
   */
  public void setFkTable(String fkTable) {
    m_fkTable = fkTable;
  }
  /**
   * Get the fkTable.
   * @return Returns the fkTable
   */
  public String getFkTable() {
    return m_fkTable;
  }
  /**
   * Set the fkColumn.
   * @param fkColumn The fkColumn to set
   */
  public void setFkColumn(String fkColumn) {
    m_fkColumn = fkColumn;
  }
  /**
   * Get the fkColumn.
   * @return Returns the fkColumn
   */
  public String getFkColumn() {
    return m_fkColumn;
  }

  /**
   * Set the pkTable.
   * @param pkTable The pkTable to set
   */
  public void setPkTable(String pkTable) {
    m_pkTable = pkTable;
  }
  /**
   * Get the pkTable.
   * @return Returns the pkTable
   */
  public String getPkTable() {
    return m_pkTable;
  }
  /**
   * Set the pkColumn.
   * @param pkColumn The pkColumn to set
   */
  public void setPkColumn(String pkColumn) {
    m_pkColumn = pkColumn;
  }
  /**
   * Get the pkColumn.
   * @return Returns the pkColumn
   */
  public String getPkColumn() {
    return m_pkColumn;
  }

  public CatalogSchema getFkCatalogSchema() {
    return m_fkCatalogSchema;
  }

  public CatalogSchema getPkCatalogSchema() {
    return m_pkCatalogSchema;
  }

  public void setFkCatalogSchema(CatalogSchema fkCatalogSchema) {
    m_fkCatalogSchema = fkCatalogSchema;
  }

  public void setPkCatalogSchema(CatalogSchema pkCatalogSchema) {
    m_pkCatalogSchema = pkCatalogSchema;
  }

  @Override
  public int hashCode() {
    return Objects.hashCode(37, m_pkCatalogSchema, m_pkTable, m_pkColumn,
                            m_fkCatalogSchema, m_fkTable, m_fkColumn,
                            m_fkName, m_keySeq);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean equals(Object o) {
    if (!(o instanceof ForeignKey)) {
      return false;
    }
    ForeignKey other = (ForeignKey)o;
    return equalsReference(other)
           && equalsFrom(other)
           && Objects.equal(m_fkName, other.getFkName())
           && Objects.equal(m_keySeq, other.getKeySeq());
  }

  /**
   * Returns true if this foreign key refers to the same table/column as another foreign key
   * (ie the referenced catalog, schema, table, and column of the two fk's are equal)
   * @param other Another fk to check
   * @return True if the two fk's point to the same column
   */
  public boolean equalsReference(ForeignKey other) {
    return Objects.equal(m_pkCatalogSchema, other.getPkCatalogSchema())
            && Objects.equal(m_pkTable, other.getPkTable())
            && Objects.equal(m_pkColumn, other.getPkColumn());
  }

  /**
   * Returns true if this foreign key is based off of the same table/column as another foreign key
   * (ie the catalog, schema, table, and column that define the fk reference are equal)
   * @param other Another fk to check
   * @return True if the two fk's point from the same column
   */
  public boolean equalsFrom(ForeignKey other) {
    return Objects.equal(m_fkCatalogSchema, other.getFkCatalogSchema())
            && Objects.equal(m_fkTable, other.getFkTable())
            && Objects.equal(m_fkColumn, other.getFkColumn());
  }

  @Override
  public String toString() {
    return getFkName() + "(" + getKeySeq() + "): " + getFkTable() + "(" + getFkColumn() + ")-->" + getPkTable()
    + "(" + getPkColumn() + ")";
  }
} @SuppressWarnings("serial")
public class InconsistentSchemaException extends RuntimeException {
  /**
   * Create a new exception.
   * @param message message.
   */
  public InconsistentSchemaException(String message) {
    super(message);
  }
} public class RelationalDatabase {
  private final Map<String, RelationalTable> m_tablesByName;

  /**
   * Construct a new instance.
   * @param tables ordered collection of tables.
   */
  public RelationalDatabase(Collection<RelationalTable> tables) {
    m_tablesByName = new LinkedHashMap<>(tables.size());
    for (RelationalTable rt : tables) {
      if (m_tablesByName.containsKey(rt.getName())) {
        throw new InconsistentSchemaException("A RelationalDatabase supports only unique table names of tables of the same "
            + "catalog/schema. Non-unique name found: " + rt.getName());
      }
      m_tablesByName.put(rt.getName(), rt);
    }
  }

  /**
   * @return ordered collection of tables in this database.
   */
  public Collection<RelationalTable> getTables() {
    return Collections.unmodifiableCollection(m_tablesByName.values());
  }

  /**
   * Gets a specific table by name.
   * @param tableName the name of the table.
   * @return the table with the given name.
   */
  public RelationalTable getTableByName(String tableName) {
    return m_tablesByName.get(tableName);
  }
} public class RelationalIndex extends BaseColumnContainer {
  /**
   * Create a new index.
   * @param catalogSchema catalog/schema.
   * @param name index name.
   */
  public RelationalIndex(CatalogSchema catalogSchema, String name) {
    super(catalogSchema, name);
  }

  /**
   * Create a new index.
   * @param catalog catalog.
   * @param schema schema.
   * @param name index name.
   */
  public RelationalIndex(String catalog, String schema, String name) {
    super(catalog, schema, name);
  }
} public class RelationalTable extends BaseColumnContainer {
  /**
   * Create a new table.
   * @param catalogSchema catalog/schema.
   * @param name table name.
   */
  public RelationalTable(CatalogSchema catalogSchema, String name) {
    super(catalogSchema, name);
  }

  /**
   * Create a new table.
   * @param catalog catalog.
   * @param schema schema.
   * @param name table name.
   */
  public RelationalTable(String catalog, String schema, String name) {
    super(catalog, schema, name);
  }

  private String m_type;
  private String m_typeName;

  private Multimap<List<String>, RelationalIndex> m_indicesByColumns;

  private Set<ForeignKey> m_fks;

  private SetMultimap<String, ForeignKey> m_fksByName; //An internal search index of fk's by name
                                                    //Note that composite keys have the same name
  private SetMultimap<String, ForeignKey> m_fksByTableColumn; //An internal search index of fk's by the "catalog.schema.table.col"
                                                           //being referenced
  private List<String> m_pkColumns;

  /**
   * Set the indices.
   * @param indices The indices to set
   * @throws InconsistentSchemaException If adding an index not recognized by the current table
   */
  public void setIndices(List<RelationalIndex> indices) throws InconsistentSchemaException {
    ImmutableMultimap.Builder<List<String>, RelationalIndex> indexMapBuilder = ImmutableListMultimap.builder();

    if (getName() == null) {
      throw new InconsistentSchemaException("Trying to add indices without setting a table!");
    } else {
      for (RelationalIndex ri : indices) {
        if (!getCatalogSchema().equals(ri.getCatalogSchema())) {
          throw new InconsistentSchemaException("Index " + ri.getName() + " and table " + getName() + " belong to different catalogs or schemas.");
        }

        indexMapBuilder.put(ri.getColumnNames(), ri);
      }
    }
    m_indicesByColumns = indexMapBuilder.build();
  }
  /**
   * Get the indices. Do NOT modify these indices -- doing so will mess up internal search indexes!
   * @return Returns the indices
   */
  public Collection<RelationalIndex> getIndices() {
    return m_indicesByColumns.values();
  }

  /**
   * Get the multimap that maps lists of column names to the indices spanning those columns
   * @return the multimap that maps lists of column names to the indices spanning those columns
   */
  public Multimap<List<String>, RelationalIndex> getIndicesByColumns() {
    return m_indicesByColumns;
  }

  /**
   * Set the fks.
   * @param fks The fks to set
   * @throws InconsistentSchemaException If mismatch between the added fk's and the table
   */
  public void setFks(Set<ForeignKey> fks) throws InconsistentSchemaException {
    m_fks = fks;
    m_fksByName = HashMultimap.create();
    m_fksByTableColumn = HashMultimap.create();

    // Check validity of fk's and then add them to the search index
    for (ForeignKey fk : fks) {
      if (!getCatalogSchema().equals(fk.getFkCatalogSchema()) || !getName().equals(fk.getFkTable())) {
        throw new InconsistentSchemaException("Foreign key " + fk + " does not match table " + getName());
      }

      //The fk is valid.  Add it to the search indices
      m_fksByName.put(fk.getFkName(), fk);

      String key = fk.getPkCatalogSchema().getCatalog() + "." + fk.getPkCatalogSchema().getSchema() + "." + fk.getPkTable() + "." + fk.getPkColumn();
      m_fksByTableColumn.put(key, fk);
    }
  }

  public String getType() {
    return m_type;
  }

  public void setType(String type) {
    m_type = type;
  }

  public String getTypeName() {
    return m_typeName;
  }

  public void setTypeName(String typeName) {
    m_typeName = typeName;
  }

  /**
   * Gets a COPY of the current table's fk.  If you add/remove any, you must set the new collection so that internal search indexes
   * can be reconstructed.
   * @return Returns a copy of the table's fks
   */
  public List<ForeignKey> getFks() {
    return new LinkedList<>(m_fks);
  }

  /**
   * Retrieve a table's foreign keys by a constraint name.
   * @param name A foreign key name
   * @return a set containing matching ForeignKeys
   */
  public Set<ForeignKey> getFksByName(String name) {
    return m_fksByName.get(name);
  }

  /**
   * Get primary key columns
   * @return primary key columns
   */
  public List<String> getPkColumns() {
    return m_pkColumns;
  }

  /**
   * Set primary key columns
   * @param pkColumns the primary key columns to set
   */
  public void setPkColumns(List<String> pkColumns) {
    m_pkColumns = pkColumns;
  }

  /**
   * Retrieve a table's foreign keys by specifying the column the constraint refers to
   * @param catalog The catalog of the referenced table.  Must be exact match (ie no wildcards)
   * @param schema The schema of the referenced table.  Must be exact match (ie no wildcards)
   * @param table The name of the referenced table.  Must be exact match (ie no wildcards)
   * @param column The column being referenced.  Must be exact match (ie no wildcards)
   * @return All foreign keys matching the specified reference.
   */
  public Set<ForeignKey> getFksByReferencedCol(String catalog, String schema, String table, String column) {
    return m_fksByTableColumn.get(catalog + "." + schema + "." + table + "." + column);
  }
}
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import com.google.common.base.Function;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Collections2;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.vecna.dbDiff.jdbc.MetadataFactory;
import com.vecna.dbDiff.model.CatalogSchema;
import com.vecna.dbDiff.model.ColumnType;
import com.vecna.dbDiff.model.TableType;
import com.vecna.dbDiff.model.db.Column;
import com.vecna.dbDiff.model.db.ForeignKey;
import com.vecna.dbDiff.model.relationalDb.InconsistentSchemaException;
import com.vecna.dbDiff.model.relationalDb.RelationalDatabase;
import com.vecna.dbDiff.model.relationalDb.RelationalIndex;
import com.vecna.dbDiff.model.relationalDb.RelationalTable;
/**
 * Copyright 2011 Vecna Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you
 * may not use this file except in compliance with the License.  You may
 * obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 * implied.  See the License for the specific language governing
 * permissions and limitations under the License.
 */

package com.vecna.dbDiff.builder;



/**
 * Builds a {@link RelationalDatabase} representation of a live database schema.
 *
 * @author dlopuch@vecna.com
 * @author ogolberg@vecna.com
 */
public class RelationalDatabaseBuilderImpl implements RelationalDatabaseBuilder {
  private final MetadataFactory m_metadataFactory;
  private ExecutorService m_executor = new ForkJoinPool();

  /**
   * Execute multiple tasks in parallel (scaling to the number of available cores). If an exception is thrown by one of the tasks, it is converted as specified below.
   * @param <T> task return type.
   * @param tasks tasks to execute.
   * @throws RelationalDatabaseReadException if one of the tasks throws a {@link SQLException} or a {@link RelationalDatabaseReadException}.
   * @throws InconsistentSchemaException if one of the tasks throws an {@link InconsistentSchemaException}.
   * @throws RuntimeException if one of the tasks throws any other exception.
   */
  private <T> void runInParallel(Collection<? extends Callable<T>> tasks) throws RelationalDatabaseReadException, InconsistentSchemaException, RuntimeException {
    Collection<Future<T>> futures;
    try {
      futures = m_executor.invokeAll(tasks);
    } catch (InterruptedException e) {
      throw new IllegalStateException(e);
    }

    for (Future<T> future : futures) {
      try {
        future.get();
      } catch (ExecutionException e) {
        Throwable cause = e.getCause();
        if (cause instanceof RuntimeException) {
          throw (RuntimeException) cause;
        } else if (cause instanceof SQLException) {
          throw new RelationalDatabaseReadException(cause);
        } else {
          throw new RuntimeException(e);
        }
      } catch (InterruptedException e) {
        throw new IllegalStateException(e);
      }
    }
  }

  /**
   * Constructor that sets metadata based on a JDBC connection
   * @param metadataFactory a {@link MetadataFactory}.
   */
  public RelationalDatabaseBuilderImpl(MetadataFactory metadataFactory) {
    m_metadataFactory = metadataFactory;
  }

  /**
   * Retrieve all tables from a schema.
   * @param catalogSchema catalog/schema.
   * @return the tables.
   * @throws SQLException if thrown by the jdbc driver.
   */
  private List<RelationalTable> getTables(final CatalogSchema catalogSchema) throws SQLException {
    // Get the ResultSet of tables
    String[] tableTypes = {TableType.TABLE.name()};
    ResultSet rs = doGetTablesQuery(catalogSchema, tableTypes);

    // Build a set of Tables
    List<RelationalTable> tables = new ArrayList<RelationalTable>();

    while (rs.next()) {
      RelationalTable table = new RelationalTable(new CatalogSchema(rs.getString(1), rs.getString(2)), rs.getString(3));

      table.setType(rs.getString(4));
      table.setTypeName(rs.getString(5));

      tables.add(table);
    }

    return tables;
  }

  /**
   * Performs a metaData.getTables() query.
   * @param catalogSchema the desired catalog and schema names.
   * @param tableTypes the desired table types, specific for the particular implementation.
   * @return The ResultSet of the getTables() call.
   * @throws SQLException if thrown by the jdbc driver.
   */
  protected ResultSet doGetTablesQuery(CatalogSchema catalogSchema, String[] tableTypes) throws SQLException {
    return m_metadataFactory.getMetadata().getTables(catalogSchema.getCatalog(), catalogSchema.getSchema(), null, tableTypes);
  }

  /**
   * Retrieve column information for a table.
   * @param table the table.
   * @return ordered list of columns.
   * @throws SQLException if thrown by the jdbc driver.
   */
  private List<Column> getColumns(RelationalTable table) throws SQLException {
    ResultSet columnResultSet = m_metadataFactory.getMetadata().getColumns(table.getCatalogSchema().getCatalog(), table.getCatalogSchema().getSchema(), table.getName(), null);

    List<Column> columns = new LinkedList<Column>();
    while (columnResultSet.next()) {
      Column column = new Column(columnResultSet.getString(1), columnResultSet.getString(2),
                                 columnResultSet.getString(4), columnResultSet.getString(3));

      column.setColumnType(new ColumnType(columnResultSet.getInt(5), columnResultSet.getString(6)));
      column.setColumnSize(columnResultSet.getInt(7));

      //Nullability
      int nullable = columnResultSet.getInt(11);
      column.setIsNullable((DatabaseMetaData.columnNullable == nullable ? true
          : (DatabaseMetaData.columnNoNulls == nullable ? false : null)));

      column.setDefault(columnResultSet.getString(13));
      column.setOrdinal(columnResultSet.getInt(17));
      columns.add(column);
    }
    return columns;
  }

  /**
   * Retrieve foreign keys for a table.
   * @param table table.
   * @return ordered list of foreign keys.
   * @throws SQLException if thrown by the jdbc driver.
   */
  private List<ForeignKey> getForeignKeys(RelationalTable table) throws SQLException {
    ResultSet fkResultSet = m_metadataFactory.getMetadata().getImportedKeys(table.getCatalogSchema().getCatalog(), table.getCatalogSchema().getSchema(), table.getName());
    List<ForeignKey> fks = new LinkedList<ForeignKey>();
    while (fkResultSet.next()) {
      ForeignKey fk = new ForeignKey();
      fk.setFkName(fkResultSet.getString(12));

      fk.setFkCatalogSchema(new CatalogSchema(fkResultSet.getString(5), fkResultSet.getString(6)));
      fk.setFkTable(fkResultSet.getString(7));
      fk.setFkColumn(fkResultSet.getString(8));

      fk.setPkCatalogSchema(new CatalogSchema(fkResultSet.getString(1), fkResultSet.getString(2)));
      fk.setPkTable(fkResultSet.getString(3));
      fk.setPkColumn(fkResultSet.getString(4));

      fk.setKeySeq(fkResultSet.getString(9));
      fks.add(fk);
    }
    return fks;
  }

  /**
   * Retrieve index information for a table.
   * @param table the table.
   * @return list of indices.
   * @throws SQLException if thrown by the jdbc driver.
   */
  private List<RelationalIndex> getIndices(RelationalTable table) throws SQLException {
    List<RelationalIndex> indices = new ArrayList<>();

    // maps index name to column names
    Multimap<String, String> idxColumns = ArrayListMultimap.create();

    // one row per index-column pair
    ResultSet rs = m_metadataFactory.getMetadata().getIndexInfo(table.getCatalogSchema().getCatalog(),
                                                                table.getCatalogSchema().getSchema(),
                                                                table.getName(), false, false);

    while (rs.next()) {
      String idxName = rs.getString(6);
      Collection<String> columns = idxColumns.get(idxName);
      if (columns.isEmpty()) {
        // build a new index
        RelationalIndex index = new RelationalIndex(table.getCatalogSchema(), rs.getString(6));
        indices.add(index);
      }

      columns.add(rs.getString(9));
    }

    for (RelationalIndex index : indices) {
      List<Column> columns = new ArrayList<>(idxColumns.size());
      for (String idxColumnName : idxColumns.get(index.getName())) {
        // Some db preserved names are double-quoted
        String columnName = idxColumnName.replaceAll("^\"|\"$", "");
        Column column = table.getColumnByName(columnName);

        if (column == null) {
          throw new InconsistentSchemaException("cannot find column " + columnName + " referenced by index " + index.getName() + " in table " + table.getName());
        }

        columns.add(column);
      }

      index.setColumns(columns);
    }

    return indices;
  }

  /**
   * Retrieve primary key information for a table.
   * @param table the table.
   * @return ordered list of primary key column names.
   * @throws SQLException if thrown by the jdbc driver.
   */
  private List<String> getPrimaryKeyColumns(RelationalTable table) throws SQLException {
    Map<Short, String> primaryKeys = new TreeMap<>();
    ResultSet rs = m_metadataFactory.getMetadata().getPrimaryKeys(table.getCatalogSchema().getCatalog(), table.getCatalogSchema().getSchema(), table.getName());
    while (rs.next()) {
      primaryKeys.put(rs.getShort(5), rs.getString(4));
    }
    return Lists.newArrayList(primaryKeys.values());
  }

  @Override
  public RelationalDatabase createRelationalDatabase(CatalogSchema catalogSchema) {