TUSCANY-3089 - Applying Florian Pinel patch to improve DAS Read command performance

git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@787796 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 290 insertions, 111 deletions

diff --git a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/config/wrapper/MappingWrapper.java b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/config/wrapper/MappingWrapper.java
index 6386df1927..6c010b3ee8 100644
--- a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/config/wrapper/MappingWrapper.java
+++ b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/config/wrapper/MappingWrapper.java
@@ -582,6 +582,21 @@ public class MappingWrapper {
         }
         return false;
     }
+    
+    public Set getRecursiveTypeNames() {
+    	Set typeNames = new HashSet();
+        if (config != null) {
+            Iterator i = getConfig().getRelationship().iterator();
+            while (i.hasNext()) {
+                Relationship r = (Relationship) i.next();
+                if (r.getPrimaryKeyTable().equals(r.getForeignKeyTable())) {
+                	typeNames.add(getTableTypeName(r.getPrimaryKeyTable()));
+                }
+            }
+        }
+        return typeNames;
+    	
+    }
 
     /*Parameter name is SDO Type name and so requires mapping in table<->type, if one is available in Config*/
     public Collection getRelationshipsByChildTable(String name) {
diff --git a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/DataObjectMaker.java b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/DataObjectMaker.java
index 495d401947..0e5fc48c0c 100644
--- a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/DataObjectMaker.java
+++ b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/DataObjectMaker.java
@@ -18,7 +18,9 @@
  */
 package org.apache.tuscany.das.rdb.graphbuilder.impl;
 
+import java.util.HashMap;
 import java.util.Iterator;
+import java.util.Map;
 
 import org.apache.log4j.Logger;
 
@@ -30,11 +32,25 @@ import commonj.sdo.helper.DataFactory;
 public final class DataObjectMaker {
 
     private final DataObject rootObject;
+    private final Map containmentPropertyMap;
+    private final Map typeMap;
 
     private static final Logger logger = Logger.getLogger(DataObjectMaker.class);
 
     public DataObjectMaker(DataObject root) {
         this.rootObject = root;
+        containmentPropertyMap = new HashMap();
+        typeMap = new HashMap();
+        Iterator i = this.rootObject.getType().getProperties().iterator();
+        while (i.hasNext()) {
+            Property p = (Property) i.next();
+            Type type = p.getType();
+            String typeName = type.getName();
+            typeMap.put(typeName, type);
+            if (p.isContainment()) {
+            	containmentPropertyMap.put(typeName, p);
+            }
+        }
     }
 
     /**
@@ -48,7 +64,8 @@ public final class DataObjectMaker {
             this.logger.debug("Looking for Type for " + tableData.getTableName());
         }
 
-        Type tableClass = findTableTypeByPropertyName(tableData.getTableName());
+        String tableName = tableData.getTableName();
+        Type tableClass = (Type) typeMap.get(tableName);
 
         if (tableClass == null) {
             throw new RuntimeException("An SDO Type with name " + tableData.getTableName() + " was not found");
@@ -58,31 +75,29 @@ public final class DataObjectMaker {
 
         // Now, check to see if the root data object has a containment reference
         // to this EClass. If so, add it to the graph. If not, it will be taken
-        // care
-        // of when we process relationships
-
-        Iterator i = this.rootObject.getType().getProperties().iterator();
-        while (i.hasNext()) {
-            Property p = (Property) i.next();
-
-            if (p.isContainment() && p.getType().equals(tableClass)) {
-                if (p.isMany()) {
-                    rootObject.getList(p).add(obj);
-                } else {
-                    this.rootObject.set(p, obj);
-                }
+        // care of when we process relationships
+        Property containmentProp = (Property) containmentPropertyMap.get(tableName);
+        if (containmentProp != null) {
+            if (containmentProp.isMany()) {
+                rootObject.getList(containmentProp).add(obj);
+            } else {
+                this.rootObject.set(containmentProp, obj);
             }
-
         }
 
+        // Set the column values
         Iterator columnNames = resultMetadata.getPropertyNames(tableData.getTableName()).iterator();
+        Type objType = obj.getType();
         while (columnNames.hasNext()) {
             String propertyName = (String) columnNames.next();
-
-            Property p = findProperty(obj.getType(), propertyName);
+            Property p = objType.getProperty(propertyName);
             if (p == null) {
-                throw new RuntimeException("Type " + obj.getType().getName() 
-                        + " does not contain a property named " + propertyName);
+            	// Try again, ignoring case
+            	p = findProperty(objType, propertyName);
+                if (p == null) {
+                    throw new RuntimeException("Type " + obj.getType().getName() 
+                            + " does not contain a property named " + propertyName);
+                }
             }
 
             Object value = tableData.getColumnData(propertyName);
@@ -105,16 +120,4 @@ public final class DataObjectMaker {
         return null;
     }
 
-    private Type findTableTypeByPropertyName(String tableName) {
-        Iterator i = rootObject.getType().getProperties().iterator();
-        while (i.hasNext()) {
-            Property p = (Property) i.next();
-            if (tableName.equals(p.getType().getName())) {
-                return p.getType();
-            }
-        }
-
-        return null;
-    }
-
 }
diff --git a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultMetadata.java b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultMetadata.java
index fdd1845be9..39f35150ec 100644
--- a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultMetadata.java
+++ b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultMetadata.java
@@ -28,6 +28,7 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.tuscany.das.rdb.Converter;
 import org.apache.tuscany.das.rdb.config.Column;
@@ -36,6 +37,7 @@ import org.apache.tuscany.das.rdb.config.Table;
 import org.apache.tuscany.das.rdb.config.wrapper.MappingWrapper;
 import org.apache.tuscany.das.rdb.config.wrapper.TableWrapper;
 import org.apache.tuscany.das.rdb.impl.ResultSetShape;
+import org.apache.tuscany.das.rdb.impl.SDODataTypes;
 
 import commonj.sdo.Type;
 
@@ -122,7 +124,7 @@ public final class ResultMetadata {
             	converterName = configWrapper.getConverter(tableName, resultSetShape.getColumnName(i));	
             }
 
-            converters[i - 1] = loadConverter(converterName);
+            converters[i - 1] = loadConverter(converterName, resultSetShape.getColumnType(i));
 
             typeNames.add(typeName);
             propertyNames.add(propertyName);
@@ -226,7 +228,7 @@ public final class ResultMetadata {
 	    }
     }
     
-    private Converter loadConverter(String converterName) {
+    private Converter loadConverter(String converterName, Type type) {
         if (converterName != null) {
 
             try {
@@ -248,7 +250,11 @@ public final class ResultMetadata {
                 throw new RuntimeException(ex);
             }
         }
-        return new DefaultConverter();
+        if (SDODataTypes.BYTES.getName().equals(type.getName()) && SDODataTypes.BYTES.getURI().equals(type.getURI())) {
+            return new DefaultConverter();
+        } else {
+        	return null;
+        }
     }
 
     public String getColumnPropertyName(int i) {
@@ -393,9 +399,13 @@ public final class ResultMetadata {
     public boolean isRecursive() {
         return configWrapper.hasRecursiveRelationships();
     }
+    
+    public Set getRecursiveTypeNames() {
+    	return configWrapper.getRecursiveTypeNames();
+    }
 
-    public Converter getConverter(int i) {
-        return converters[i - 1];
+    public Object convert(int i, Object data) {
+    	return (converters[i - 1] == null) ? data : converters[i - 1].getPropertyValue(data);
     }
 
 }
diff --git a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultSetProcessor.java b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultSetProcessor.java
index 03a77c000f..fa17ebb098 100644
--- a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultSetProcessor.java
+++ b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultSetProcessor.java
@@ -113,6 +113,8 @@ public class ResultSetProcessor {
      */
     private int addRowToGraph(ResultSetRow row, ResultMetadata resultMetadata) throws SQLException {
     	int rootRowsCreated = 0;
+    	int objectsCreated = 0;
+    	boolean recursive = row.isRecursive();
     	Set rootTableNames = metadata.getConfigWrapper().getRootTableNames();
         tableObjects.clear();
     	Iterator tables = row.getAllTableData().iterator();
@@ -131,11 +133,15 @@ public class ResultSetProcessor {
             DataObject tableObject = registry.get(tableName, rawDataFromRow.getPrimaryKeyValues());
             boolean newlyCreated = (tableObject == null);
             // check whether row is a new root row
-            if (newlyCreated && rootTableNames.contains(tableName)) rootRowsCreated++;
-            if (newlyCreated
-            		&& !rawDataFromRow.hasNullPrimaryKey()) {//2nd check for null data in PK,
-            	//as TableData.addData() - hasValidPrimaryKey=false is commented for a reason
-            	//with this, DataObjs with null PK will not be added to registry and tableObjects
+            if (newlyCreated) {
+            	objectsCreated++;
+            	// increment root row count
+            	// in case of recursive table, assuming that first table occurrence is the root
+                if (rootTableNames.contains(tableName) && rawDataFromRow.getIndex() == 0) rootRowsCreated++;
+                // get whole table data 
+                // (only for non-recursive statements; recursive statements already have the whole table data)
+                if (!recursive) rawDataFromRow = row.processRowForTable(tableName);
+            	// create data object
             	tableObject = doMaker.createAndAddDataObject(rawDataFromRow, resultMetadata);
                 if (this.logger.isDebugEnabled()) {
                     this.logger.debug("Putting table " + tableName + " with PK "
@@ -146,7 +152,7 @@ public class ResultSetProcessor {
             }
             else{
                 if (this.logger.isDebugEnabled()) {
-                    this.logger.debug("Not Null tableObject or NULL PK");
+                    this.logger.debug("Not Null tableObject");
                 }
             }
 
@@ -157,8 +163,15 @@ public class ResultSetProcessor {
             	tableObjects.put(tableName, tableObject, newlyCreated);
             }
         }
-
-        tableObjects.processRelationships();
+        
+        if (objectsCreated == 0) {
+            // duplicated row
+        	if (this.logger.isDebugEnabled()) {
+        		this.logger.debug("Found duplicated row");
+        	}
+        } else {
+        	tableObjects.processRelationships();
+        }
         
         return rootRowsCreated;
 
diff --git a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultSetRow.java b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultSetRow.java
index 6ef815530d..70e0961b09 100644
--- a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultSetRow.java
+++ b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/ResultSetRow.java
@@ -42,12 +42,16 @@ public class ResultSetRow {
     private final boolean recursive;
     private final int resultSetSize;
     private Collection allTableNames;
+    private Set recursiveTablePropertyNames;
     private Set tablesWithNoPK = new HashSet();
     private String[] tablePropertyNames;
     private String[] columnPropertyNames;
     private boolean[] isPKColumn;
+    private int[] indexesForPKs;
+    private Map indexesByTablePropertyName = new HashMap();
     private Map tableMap = new HashMap();
-    private List allTableData;
+    private List allTableData = new ArrayList();
+    private ResultSet currentResultSet;
 
     /**
      * Method ResultSetRow.
@@ -57,7 +61,8 @@ public class ResultSetRow {
      */
     public ResultSetRow(ResultMetadata m) throws SQLException {
         this.metadata = m;
-        this.recursive = m.isRecursive();
+        this.recursiveTablePropertyNames = m.getRecursiveTypeNames();
+        this.recursive = (recursiveTablePropertyNames.size() > 0);
         this.resultSetSize = m.getResultSetSize();
         cacheMetadata();
         getAllTableNamesForRS();
@@ -70,36 +75,74 @@ public class ResultSetRow {
      * @param rs
      *            A ResultSet positioned on the desired row
      */
-    public void processRow(ResultSet rs)  throws SQLException {
-    	// clear previous data 
-    	for (Iterator itTableData = tableMap.values().iterator(); itTableData.hasNext(); ) {
+    public final void processRow(ResultSet rs)  throws SQLException {
+    	// clear previous data
+    	for (Iterator itTableData = allTableData.iterator(); itTableData.hasNext(); ) {
     		TableData tableData = (TableData) itTableData.next();
     		tableData.clear();
     	}
-        allTableData = null;
-
+        allTableData = new ArrayList();
+        // set current resultSet
+        currentResultSet = rs;
         // process row
         if (recursive) {
-            processRecursiveRow(rs);
+            processRecursiveRow();
         } else {
-        	processNonRecursiveRow(rs);
+        	processNonRecursiveRow();
         }
     }
+    
+    public final TableData processRowForTable(String tablePropertyName) throws SQLException {
+        int[] indexes = (int[]) indexesByTablePropertyName.get(tablePropertyName);
+		TableData table = getRawData(tablePropertyName);
+        int count = indexes.length;
+        for (int j = 0; j < count; j++) {
+        	int i = indexes[j];
+        	if (!isPKColumn[i]) {
+        		// skipping primary key columns since they've already been processed by processRow()
+	        	Object data = getObject(currentResultSet, i);
+	            if (this.logger.isDebugEnabled()) {
+	                this.logger.debug("Adding column: " + columnPropertyNames[i] + "\tValue: " 
+	                        + data + "\tTable: "
+	                        + tablePropertyNames[i]);
+	            }
+	            table.addData(columnPropertyNames[i], false, data);
+        	}
+		}
+        return table;
+    }
 
-    private void processNonRecursiveRow(ResultSet rs) throws SQLException {
-
-        if (this.logger.isDebugEnabled()) {
-            this.logger.debug("");
-        }
-        for (int i = 1; i <= resultSetSize; i++) {
-        	Object data = getObject(rs, i);
-			TableData table = getRawData(tablePropertyNames[i]);
-            if (this.logger.isDebugEnabled()) {
-                this.logger.debug("Adding column: " + columnPropertyNames[i] + "\tValue: " 
-                        + data + "\tTable: "
-                        + tablePropertyNames[i]);
-            }
-            table.addData(columnPropertyNames[i], isPKColumn[i], data);
+    private void processNonRecursiveRow() throws SQLException {
+        // parse primary keys only
+        // the rest will be parsed as needed
+        int count = indexesForPKs.length;
+        for (int j = 0; j < count; j++) {
+        	int i = indexesForPKs[j];
+        	Object data = getObject(currentResultSet, i);
+        	if (data == null) {
+        		// primary key is null, check other columns
+        		String tablePropertyName = tablePropertyNames[i];
+        		// if table data already exists then this has already been done
+        		if (!tableMap.containsKey(tablePropertyName)) {
+        			TableData table = getRawData(tablePropertyName);
+        			processRowForTable(tablePropertyName);
+        			// add table data only if not empty
+        			if (!table.isTableEmpty()) {
+        	            table.addData(columnPropertyNames[i], true, data);
+        				allTableData.add(table);
+        			}
+        		}
+        	} else {
+        		// add table data
+				TableData table = getRawData(tablePropertyNames[i]);
+				if (!allTableData.contains(table)) allTableData.add(table);
+	            if (this.logger.isDebugEnabled()) {
+	                this.logger.debug("Adding column: " + columnPropertyNames[i] + "\tValue: " 
+	                        + data + "\tTable: "
+	                        + tablePropertyNames[i]);
+	            }
+	            table.addData(columnPropertyNames[i], true, data);
+        	}
 		}
         checkResultSetMissesPK();
         }
@@ -118,10 +161,42 @@ public class ResultSetRow {
     	tablePropertyNames = new String[resultSetSize + 1];
     	columnPropertyNames = new String[resultSetSize + 1];
     	isPKColumn = new boolean[resultSetSize + 1];
+    	String tablePropertyName = null;
+    	boolean isPK;
+    	List pkColumnList = new ArrayList();
+    	// loop thru indexes
         for (int i = 1; i <= resultSetSize; i++) {
-        	tablePropertyNames[i] = metadata.getTablePropertyName(i);
         	columnPropertyNames[i] = metadata.getColumnPropertyName(i);
-        	isPKColumn[i] = metadata.isPKColumn(i);
+        	tablePropertyName = metadata.getTablePropertyName(i);
+        	tablePropertyNames[i] = tablePropertyName;
+        	List indexes = (List) indexesByTablePropertyName.get(tablePropertyName);
+        	if (indexes == null) {
+        		indexes = new ArrayList();
+        		indexesByTablePropertyName.put(tablePropertyName, indexes);
+        	}
+        	indexes.add(new Integer(i));
+        	isPK = metadata.isPKColumn(i);
+        	isPKColumn[i] = isPK;
+        	if (isPK) {
+        		pkColumnList.add(new Integer(i));
+        	}
+        }
+        // reorganize indexes by table property name
+        for (Iterator itTablePropertyNames = indexesByTablePropertyName.keySet().iterator(); itTablePropertyNames.hasNext(); ) {
+        	tablePropertyName = (String) itTablePropertyNames.next();
+        	List indexes = (List) indexesByTablePropertyName.get(tablePropertyName);
+        	int count = indexes.size();
+        	int[] indexArray = new int[count];
+        	for (int i = 0; i < count; i++) {
+        		indexArray[i] = ((Integer) indexes.get(i)).intValue();
+        	}
+        	indexesByTablePropertyName.put(tablePropertyName, indexArray);
+        }
+        // reorganize indexes for PKs
+        int count = pkColumnList.size();
+        indexesForPKs = new int[count];
+        for (int i = 0; i < count; i++) {
+        	indexesForPKs[i] = ((Integer) pkColumnList.get(i)).intValue();
         }
     }
     	
@@ -191,19 +266,36 @@ public class ResultSetRow {
         	String currentTableName = (String)itr.next();
            	TableData table = getRawData(currentTableName);
            	table.setValidPrimaryKey(false);
+			allTableData.add(table);
         }
     }
     
-    private void processRecursiveRow(ResultSet rs) throws SQLException {
-        this.allTableData = new ArrayList();
+    private void processRecursiveRow() throws SQLException {
         int i = 1;
-
+        // create map to keep track of recursive indexes
+        // each recursive table contains a 0-based index to keep track of the sequence
+        Map recursiveIndexes = new HashMap();
+        for (Iterator itTablePropertyNames = recursiveTablePropertyNames.iterator(); itTablePropertyNames.hasNext(); ) {
+        	recursiveIndexes.put(itTablePropertyNames.next(), new Integer(-1));
+        }
+        
+        // loop thru result set columns
+        // assuming that the columns of each recursive table are grouped together (metadata do not allow for further granularity)
         while (i <= resultSetSize) {
-            TableData table = new TableData(tablePropertyNames[i]);
-            this.allTableData.add(table);
-
+        	TableData table;
+        	String tablePropertyName = tablePropertyNames[i];
+        	if (recursiveTablePropertyNames.contains(tablePropertyName)) {
+        		// increment current recursive index for table
+        		int recursiveIndex = ((Integer) recursiveIndexes.get(tablePropertyName)).intValue() + 1;
+        		recursiveIndexes.put(tablePropertyName, new Integer(recursiveIndex));
+        		// get table data
+                table = getRecursiveRawData(tablePropertyName, recursiveIndex);
+        	} else {
+                table = getRawData(tablePropertyNames[i]);
+        	}
+ 
             while ((i <= resultSetSize) && (isPKColumn[i])) {
-                Object data = getObject(rs, i);
+                Object data = getObject(currentResultSet, i);
                 if (this.logger.isDebugEnabled()) {
                     this.logger.debug("Adding column: " + columnPropertyNames[i]
                             + "\tValue: " + data + "\tTable: "
@@ -214,7 +306,7 @@ public class ResultSetRow {
             }
 
             while ((i <= resultSetSize) && (!isPKColumn[i])) {
-                Object data = getObject(rs, i);
+                Object data = getObject(currentResultSet, i);
                 if (this.logger.isDebugEnabled()) {
                     this.logger.debug("Adding column: " + columnPropertyNames[i] 
                             + "\tValue: " + data + "\tTable: "
@@ -223,6 +315,14 @@ public class ResultSetRow {
                 table.addData(columnPropertyNames[i], false, data);
                 i++;
             }
+            
+            // skip table if empty
+            if (table.isTableEmpty()) {
+            	table.clear();
+            } else {
+            	this.allTableData.add(table);
+            }
+
         }
         
         checkResultSetMissesPK();        
@@ -242,22 +342,11 @@ public class ResultSetRow {
             return null;
         } 
             
-        return metadata.getConverter(i).getPropertyValue(data);
+        return metadata.convert(i, data);
         
     }
 
     /**
-     * Returns a HashMap that holds data for the specified table
-     * 
-     * @param tableName
-     *            The name of the table
-     * @return HashMap
-     */
-    public TableData getTable(String tableName) {
-        return (TableData) tableMap.get(tableName);
-    }
-
-    /**
      * Returns a HashMap that holds data for the specified table If the HashMap 
      * doesn't exist, it will be created. This is used internally to build
      * the ResultSetRow, whereas getTable is used externally to retrieve existing table data.
@@ -278,17 +367,24 @@ public class ResultSetRow {
         return table;
     }
     
-    public List getAllTableData() {
-        if (this.allTableData == null) {
-            this.allTableData = new ArrayList();
-            this.allTableData.addAll(tableMap.values());
-        }
+    private TableData getRecursiveRawData(String tableName, int recursiveIndex) {
+    	String key = tableName + ":" + recursiveIndex;
+        TableData table = (TableData) tableMap.get(key);
 
-        if (this.logger.isDebugEnabled()) {
-            this.logger.debug(allTableData);
+        if (table == null) {
+            table = new TableData(tableName, recursiveIndex);
+            tableMap.put(key, table);
         }
 
-        return this.allTableData;
+        return table;
     }
 
+    public List getAllTableData() {
+        return this.allTableData;
+    }
+    
+    public boolean isRecursive() {
+    	return recursive;
+    }
+    
 }
diff --git a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/RowObjects.java b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/RowObjects.java
index 13060a3ca6..b41b7379c8 100644
--- a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/RowObjects.java
+++ b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/RowObjects.java
@@ -44,6 +44,8 @@ public class RowObjects {
     private Set newTableObjectNames;
 
     private Map tableTypeNames;
+    
+    private Map relationshipMap;
 
     private final GraphBuilderMetadata metadata;
 
@@ -62,6 +64,31 @@ public class RowObjects {
         this.registry = registry;
         this.configWrapper = metadata.getConfigWrapper();
         this.hasRecursiveRelationships = configWrapper.hasRecursiveRelationships();
+        if (!hasRecursiveRelationships) buildRelationshipMap();
+    }
+    
+    private void buildRelationshipMap() {
+    	relationshipMap = new HashMap();
+        Iterator i = metadata.getRelationships().iterator();
+        while (i.hasNext()) {
+            Relationship r = (Relationship) i.next();
+            String parentTypeName = getTableTypeName(r.getPrimaryKeyTable());
+            String childTypeName = getTableTypeName(r.getForeignKeyTable());
+            // Add relationship under child type name
+            List relationships = (List) relationshipMap.get(childTypeName);
+            if (relationships == null) {
+            	relationships = new ArrayList();
+            	relationshipMap.put(childTypeName, relationships);
+            }
+            relationships.add(r);
+            // Add relationship under parent type name
+            relationships = (List) relationshipMap.get(parentTypeName);
+            if (relationships == null) {
+            	relationships = new ArrayList();
+            	relationshipMap.put(parentTypeName, relationships);
+            }
+            relationships.add(r);
+        }
     }
     
     public void clear() {
@@ -94,27 +121,31 @@ public class RowObjects {
             processRecursiveRelationships(configWrapper);
             return;
         }
-
-        Iterator i = metadata.getRelationships().iterator();
+        
+        // the relationship needs to be set only if the parent or the child is newly created
+        // otherwise the relationship has already been set
+        Set relationshipsToSet = new HashSet();
+        Iterator itNewTableObjectNames = newTableObjectNames.iterator();
+        while (itNewTableObjectNames.hasNext()) {
+        	List relationships = (List) relationshipMap.get((String) itNewTableObjectNames.next());
+        	if (relationships != null) relationshipsToSet.addAll(relationships);
+        }
+        
+        Iterator i = relationshipsToSet.iterator();
         while (i.hasNext()) {
             Relationship r = (Relationship) i.next();
 
             String parentTypeName = getTableTypeName(r.getPrimaryKeyTable());
             String childTypeName = getTableTypeName(r.getForeignKeyTable());
+            DataObject parent = get(parentTypeName);
+            DataObject child = get(childTypeName);
 
-            // the relationship needs to be set only if the parent or the child is newly created
-            // otherwise the relationship has already been set
-            if (newTableObjectNames.contains(parentTypeName) || newTableObjectNames.contains(childTypeName)) {
-                DataObject parent = get(parentTypeName);
-                DataObject child = get(childTypeName);
-
-                if (this.logger.isDebugEnabled()) {
-                    this.logger.debug("Parent table: " + parent);
-                    this.logger.debug("Child table: " + child);
-                }
-
-                setOrAdd(parent, child, r.getName());
+            if (this.logger.isDebugEnabled()) {
+                this.logger.debug("Parent table: " + parent);
+                this.logger.debug("Child table: " + child);
             }
+
+            setOrAdd(parent, child, r.getName());
         }
     }
 
diff --git a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/TableData.java b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/TableData.java
index 959a07b26f..f9084f4014 100644
--- a/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/TableData.java
+++ b/java/das/rdb/src/main/java/org/apache/tuscany/das/rdb/graphbuilder/impl/TableData.java
@@ -33,6 +33,8 @@ public final class TableData {
     private List primaryKey;;
 
     private final String name;
+    
+    private int index = 0;
 
     private boolean hasValidPrimaryKey = true;
 
@@ -48,6 +50,11 @@ public final class TableData {
         this.primaryKey = new ArrayList();
     }
     
+    public TableData(String tableName, int index) {
+    	this(tableName);
+    	this.index = index;
+    }
+    
     public void clear() {
     	columnData.clear();
     	primaryKey.clear();
@@ -82,6 +89,10 @@ public final class TableData {
     public String getTableName() {
         return this.name;
     }
+    
+    public int getIndex() {
+    	return index;
+    }
 
     /**
      * @return