OPENJPA-1083 Fixed a mapping tool failure caused by the inability to discover and drop multi-column foreign key constraints. Multi-column FK's were not getting dropped, but got added after clearing out the tables. Trying to add an existing FK caused an exception when using Oracle.

git-svn-id: https://svn.apache.org/repos/asf/openjpa/trunk@832587 13f79535-47bb-0310-9956-ffa450edef68

openjpa-jdbc/src/main/java/org/apache/openjpa/jdbc/schema/ForeignKey.java

@@ -26,6 +26,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 
 import org.apache.commons.lang.ObjectUtils;
+import org.apache.commons.lang.StringUtils;
 import org.apache.openjpa.jdbc.sql.DBDictionary;
 import org.apache.openjpa.lib.log.Log;
 import org.apache.openjpa.lib.util.Localizer;
@@ -753,7 +754,7 @@ public class ForeignKey
             Schema schema = getTable().getSchema();
             ForeignKey[] fks = dbdict.getImportedKeys(conn.getMetaData(), 
                 conn.getCatalog(), schema.getName(), 
-                getTable().getName(), conn);
+                getTable().getName(), conn, false);
             for ( int i=0; i< fks.length; i++) {
                 Table localtable = schema.getTable(fks[i].getTableName());
                 Table pkTable = schema.getTable(
@@ -768,10 +769,33 @@ public class ForeignKey
                     fkTemp.setDeferred(fks[i].isDeferred());
                     fkTemp.setDeleteAction(fks[i].getDeleteAction());
                 }
-                if( ! fkTemp.containsColumn(
+                if (fks[i].getColumns() == null || fks[i].getColumns().length == 0) {
-                    localtable.getColumn(fks[i].getColumnName(), dbdict)))
+                    // Singular column foreign key 
-                fkTemp.join(localtable.getColumn(fks[i].getColumnName(), dbdict), 
+                    if( ! fkTemp.containsColumn(
-                    pkTable.getColumn(fks[i].getPrimaryKeyColumnName(), dbdict));
+                        localtable.getColumn(fks[i].getColumnName(), dbdict)))
+                    fkTemp.join(localtable.getColumn(fks[i].getColumnName(), dbdict), 
+                        pkTable.getColumn(fks[i].getPrimaryKeyColumnName(), dbdict));
+                } else {
+                    // Add the multi-column foreign key, joining local and pk columns in
+                    // the temporary key
+                    Column[] locCols = fks[i].getColumns();
+                    Column[] pkCols = fks[i].getPrimaryKeyColumns();
+                    // Column counts must match
+                    if (locCols != null && pkCols != null & 
+                        locCols.length != pkCols.length) {
+                        Log log = dbdict.getLog();
+                        if (log.isTraceEnabled()) {
+                            log.trace(_loc.get("fk-column-mismatch"));
+                        }
+                    }
+                    for (int j = 0; j < locCols.length; j++) {
+                        if( ! fkTemp.containsColumn(
+                            localtable.getColumn(locCols[j].getName(), dbdict))) {
+                            fkTemp.join(localtable.getColumn(locCols[j].getName(), dbdict), 
+                                pkTable.getColumn(pkCols[j].getName(), dbdict));
+                        }
+                    }
+                }
                 if( equalsForeignKey(fkTemp))
                 {
                     if(addFK)
@@ -791,4 +815,95 @@ public class ForeignKey
         return retVal;
     }
 
+    /**
+     * Joins the column of a single column FK to this FK.
+     * @param fk
+     */
+    public void addColumn(ForeignKey fk) {
+        // Convert simple name based fk to a multi-column FK if necessary.
+        if (getColumns() == null || getColumns().length == 0) {
+            // If this FK is single column key, covert to a multi-column key
+            Column[] keyCols = createKeyColumns(this);
+            if (keyCols[0] != null && keyCols[1] != null) {
+                setPrimaryKeyColumnName(null);
+                setColumnName(null);
+                join(keyCols[0], keyCols[1]);
+            }
+        }
+        // Create the local and primary key columns from the fk and add them
+        // to this fk.
+        Column[] keyCols = createKeyColumns(fk);
+        if (keyCols[0] != null && keyCols[1] != null) {
+            join(keyCols[0], keyCols[1]);
+        }
+    }
+    
+    /*
+     * Creates the local and primary key columns for a name-based fk. 
+     * @return Column[] element 0 is local column
+     *                  element 1 is the primary key in another table.
+     */
+    private static Column[] createKeyColumns(ForeignKey fk) {
+        Column fkCol = null;
+        if (!StringUtils.isEmpty(fk.getColumnName())) {
+            fkCol = new Column();
+            fkCol.setName(fk.getColumnName());
+            fkCol.setTableName(fk.getTableName());
+            fkCol.setSchemaName(fk.getSchemaName());
+        }
+        
+        Column pkCol = null;
+        if (!StringUtils.isEmpty(fk.getPrimaryKeyColumnName())) {
+            pkCol = new Column();
+            pkCol.setName(fk.getPrimaryKeyColumnName());
+            pkCol.setTableName(fk.getPrimaryKeyTableName());
+            pkCol.setSchemaName(fk.getPrimaryKeySchemaName());
+        }
+        return new Column[] { fkCol, pkCol };
+    }
+    
+    /*
+     * ForeignKey utility class which determines equality based upon the 
+     * non-column state of the keys.  
+     */
+    public static class FKMapKey {
+        
+        private ForeignKey _fk;
+
+        public FKMapKey(ForeignKey fk) {
+            _fk = fk;
+        }
+        public ForeignKey getFk() {
+            return _fk;
+        }
+
+        public int hashCode() {
+            return getFk().getName() != null ? getFk().getName().hashCode() : getFk().hashCode();
+        }
+        
+        public boolean equals(Object fkObj) {
+            if (fkObj == this) {
+                return true;
+            }
+            if (fkObj == null || !(fkObj instanceof FKMapKey)) {
+                return false;
+            }
+            ForeignKey fk = ((FKMapKey)fkObj).getFk();
+            if (getFk().getDeleteAction() != fk.getDeleteAction())
+                return false;
+            if (getFk().isDeferred() != fk.isDeferred())
+                return false;
+            if (!getFk().getName().equals(fk.getName())) {
+                return false;
+            }
+            // Assert PK table name and schema
+            if (!StringUtils.equals(getFk().getPrimaryKeySchemaName(), fk.getPrimaryKeySchemaName()) ||
+                !StringUtils.equals(getFk().getPrimaryKeyTableName(), fk.getPrimaryKeyTableName()) ||
+                !StringUtils.equals(getFk().getSchemaName(), fk.getSchemaName()) ||
+                !StringUtils.equals(getFk().getTableName(), fk.getTableName())) {
+                return false;
+            }
+            return true;
+        }
+    }
 }

openjpa-jdbc/src/main/java/org/apache/openjpa/jdbc/sql/DBDictionary.java

@@ -86,6 +86,7 @@ import org.apache.openjpa.jdbc.schema.SchemaGroup;
 import org.apache.openjpa.jdbc.schema.Sequence;
 import org.apache.openjpa.jdbc.schema.Table;
 import org.apache.openjpa.jdbc.schema.Unique;
+import org.apache.openjpa.jdbc.schema.ForeignKey.FKMapKey;
 import org.apache.openjpa.kernel.Filters;
 import org.apache.openjpa.kernel.OpenJPAStateManager;
 import org.apache.openjpa.kernel.Seq;
@@ -4118,6 +4119,16 @@ public class DBDictionary
     public ForeignKey[] getImportedKeys(DatabaseMetaData meta, String catalog,
         String schemaName, String tableName, Connection conn)
         throws SQLException {
+        return getImportedKeys(meta, catalog, schemaName, tableName, conn, true);
+    }
+
+    /**
+     * Reflect on the schema to return full foreign keys imported by the given
+     * table pattern.
+     */
+    public ForeignKey[] getImportedKeys(DatabaseMetaData meta, String catalog,
+        String schemaName, String tableName, Connection conn, boolean partialKeys)
+        throws SQLException {
         if (!supportsForeignKeys)
             return null;
         if (tableName == null && !supportsNullTableForGetImportedKeys)
@@ -4130,20 +4141,51 @@ public class DBDictionary
                 getSchemaNameForMetadata(schemaName),
                 getTableNameForMetadata(tableName));
 
-            List importedKeyList = new ArrayList();
+            List<ForeignKey> importedKeyList = new ArrayList<ForeignKey>();
-            while (keys != null && keys.next())
+            Map<FKMapKey, ForeignKey> fkMap = new HashMap<FKMapKey, ForeignKey>();
-                importedKeyList.add(newForeignKey(keys));
+
+            while (keys != null && keys.next()) {
+                ForeignKey nfk = newForeignKey(keys);
+                if (!partialKeys) {
+                    ForeignKey fk = combineForeignKey(fkMap, nfk);
+                    // If the key returned != new key, fk col was combined
+                    // with existing fk.
+                    if (fk != nfk) {
+                        continue;
+                    }
+                }
+                importedKeyList.add(nfk);
+            }
             return (ForeignKey[]) importedKeyList.toArray
                 (new ForeignKey[importedKeyList.size()]);
         } finally {
-            if (keys != null)
+            if (keys != null) {
                 try {
                     keys.close();
                 } catch (Exception e) {
                 }
+            }
         }
     }
     
+    /*
+     * Combines partial foreign keys into singular key
+     */
+    protected ForeignKey combineForeignKey(Map<FKMapKey, ForeignKey> fkMap,
+        ForeignKey fk) {
+        
+        FKMapKey fkmk = new FKMapKey(fk);
+        ForeignKey baseKey = fkMap.get(fkmk);
+        // Found the FK, add the additional column
+        if (baseKey != null) {
+            baseKey.addColumn(fk);
+            return baseKey;
+        }
+        // fkey is new
+        fkMap.put(fkmk, fk);
+        return fk;
+    }
+
     /**
      * Create a new foreign key from the information in the schema metadata.
      */

openjpa-jdbc/src/main/java/org/apache/openjpa/jdbc/sql/OracleDictionary.java

@@ -34,6 +34,7 @@ import java.sql.Types;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Calendar;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -49,6 +50,7 @@ import org.apache.openjpa.jdbc.schema.Index;
 import org.apache.openjpa.jdbc.schema.PrimaryKey;
 import org.apache.openjpa.jdbc.schema.Sequence;
 import org.apache.openjpa.jdbc.schema.Table;
+import org.apache.openjpa.jdbc.schema.ForeignKey.FKMapKey;
 import org.apache.openjpa.lib.jdbc.DelegatingDatabaseMetaData;
 import org.apache.openjpa.lib.jdbc.DelegatingPreparedStatement;
 import org.apache.openjpa.lib.util.J2DoPrivHelper;
@@ -756,7 +758,7 @@ public class OracleDictionary
     }
 
     public ForeignKey[] getImportedKeys(DatabaseMetaData meta, String catalog,
-        String schemaName, String tableName, Connection conn)
+        String schemaName, String tableName, Connection conn, boolean partialKeys)
         throws SQLException {
         StringBuffer delAction = new StringBuffer("DECODE(t1.DELETE_RULE").
             append(", 'NO ACTION', ").append(meta.importedKeyNoAction).
@@ -805,9 +807,20 @@ public class OracleDictionary
                 setString(stmnt, idx++, tableName.toUpperCase(), null);
             setTimeouts(stmnt, conf, false);
             rs = stmnt.executeQuery();
-            List fkList = new ArrayList();
+            List<ForeignKey> fkList = new ArrayList<ForeignKey>();            
-            while (rs != null && rs.next())
+            Map<FKMapKey, ForeignKey> fkMap = new HashMap<FKMapKey, ForeignKey>();
-                fkList.add(newForeignKey(rs));
+
+            while (rs != null && rs.next()) {
+                ForeignKey nfk = newForeignKey(rs);
+                if (!partialKeys) {
+                    ForeignKey fk = combineForeignKey(fkMap, nfk);
+                    // Only add the fk to the import list if it is new
+                    if (fk != nfk) {
+                        continue;
+                    }
+                }
+                fkList.add(nfk);
+            }
             return (ForeignKey[]) fkList.toArray
                 (new ForeignKey[fkList.size()]);
         } finally {

openjpa-jdbc/src/main/resources/org/apache/openjpa/jdbc/schema/localizer.properties

@@ -154,3 +154,5 @@ conn-failed: Failed to connect to DataSource. Verify Driver "{0}", URL "{1}" \
 no-column: Can not find column "{0}" in table "{1}"
 except-read-fk-name: An exception occurred when obtaining the foreign key \
     names from the database.  
+fk-column-mismatch: Unable to create multi-column foreign key. The key \
+    columns do not match primary keys in foreign table.