gt: 				if (dn.isLast() && dn.isObjectValue()) {
context: /*
   * JBoss, Home of Professional Open Source.
   * See the COPYRIGHT.txt file distributed with this work for information
   * regarding copyright ownership.  Some portions may be licensed
   * to Red Hat, Inc. under one or more contributor license agreements.
   * 
   * This library is free software; you can redistribute it and/or
   * modify it under the terms of the GNU Lesser General Public
   * License as published by the Free Software Foundation; either
   * version 2.1 of the License, or (at your option) any later version.
   * 
   * This library is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   * Lesser General Public License for more details.
   * 
   * You should have received a copy of the GNU Lesser General Public
   * License along with this library; if not, write to the Free Software
   * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
   * 02110-1301 USA.
   */
  
  package org.teiid.translator.infinispan.dsl;
  
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  
  import javax.script.CompiledScript;
  import javax.script.ScriptContext;
  import javax.script.ScriptException;
  import javax.script.SimpleScriptContext;
  
  import org.teiid.core.util.StringUtil;
  import org.teiid.language.ColumnReference;
  import org.teiid.language.DerivedColumn;
  import org.teiid.language.NamedTable;
  import org.teiid.language.Select;
  import org.teiid.language.TableReference;
  import org.teiid.logging.LogConstants;
  import org.teiid.logging.LogManager;
  import org.teiid.metadata.Column;
  import org.teiid.metadata.ForeignKey;
  import org.teiid.metadata.RuntimeMetadata;
  import org.teiid.query.eval.TeiidScriptEngine;
  import org.teiid.translator.DataNotAvailableException;
  import org.teiid.translator.ExecutionContext;
  import org.teiid.translator.ResultSetExecution;
  import org.teiid.translator.TranslatorException;
  
  /**
   * Execution of the SELECT Command
   */
  public class InfinispanExecution implements ResultSetExecution {
  	public enum  OBJECT_TYPE {
  		COLLECTION,
  		ARRAY,
  		MAP,
  		VALUE
  	}
  
  	
  	class DepthNode {
  		
  		protected List<String> nodes = null;
  		protected String depthNodes = null;
  		protected int nodeSize = 0;
  		protected int columnLoc = 0;
  		
  		// values that change during processing
  		protected int nodePosition=0;
  		protected Object value = null;  
  		protected OBJECT_TYPE ot;
  		
  		private DepthNode() {
  			
  		}
  		DepthNode(String name, int colLocation) {
  			this();
  			nodes = StringUtil.split(name, ".");
  			nodeSize = nodes.size();
  			depthNodes = name.substring(0, name.lastIndexOf("."));
  			columnLoc = colLocation;
  	
  		}
  		
  		int getColumnLocation() {
  			return columnLoc;
  		}
  		
  		String getNodeName(int position) {
  			return nodes.get(position);
  		}
  		
  		String getCurrentNodeName() {
  			return nodes.get(nodePosition);
  		}
  		
  		int getNumberOfNodes() {
  			return nodeSize;
  		}
  		
  		String getDepthNodes() {
  			return depthNodes;
  		}
  		
  		String getDepthMethodName() {
  			return nodes.get(nodeSize -1);
  		}
  		
  		boolean hasNext() {
  			return (nodePosition < (nodeSize -1) ?true:false);
  		}
  		
  		boolean isLast() {
  			return ( nodePosition == (nodeSize - 1) ? true:false);
  		}
  		
  		void incrementPosition() {
  			++nodePosition;
  		}
  		
  		Object getValue() {
  			return value;
  		}
  		
  		boolean isCollection() {
  			return (ot == OBJECT_TYPE.COLLECTION);
  		}
  		
  		boolean isMap() {
  			return (ot == OBJECT_TYPE.MAP);
  		}
  
  		boolean isArray() {
  			return (ot == OBJECT_TYPE.ARRAY);
  		}
  		
  		boolean isObjectValue() {
  			return (ot == OBJECT_TYPE.VALUE);
  		}
  		
  		/**
  		 * Called to reset pointer to what it was before 
  		 * being processed.  This is so the this node can be reused
  		 * for each object in the object result set for this column.
  		 */
  		void reset() {
  			nodePosition=0;
  			value = null;
  			ot=null;
  		}
  		
  		void setValue(Object v) {
  			value = v;
  			if (Collection.class.isAssignableFrom(v.getClass())) {
  				ot = OBJECT_TYPE.COLLECTION;	
  			} else if (Map.class.isAssignableFrom(v.getClass())) {
  				ot = OBJECT_TYPE.MAP;
  			} else if (v.getClass().isArray()) {
  				ot = OBJECT_TYPE.ARRAY;
  			} else {
  				ot = OBJECT_TYPE.VALUE;
  			}
  		}
  
  		@Override
  		protected Object clone() {
  			DepthNode dn = new DepthNode();
  			dn.nodePosition = this.nodePosition;
  			dn.nodes = this.nodes;
  			dn.depthNodes = this.depthNodes;
  			dn.nodeSize = this.nodeSize;
  			dn.columnLoc = this.columnLoc;
  			return dn;
  		}
  	}
  
  	protected Select query;
  	protected InfinispanConnection connection;
  	private Object[] colObjects;
  	private ScriptContext sc = new SimpleScriptContext();
  	private TeiidScriptEngine scriptEngine;
  	private Iterator<Object> objResultsItr = null;
  	private Iterator<Object> cacheResultsIt = null;
  	private InfinispanExecutionFactory factory;
  	private ExecutionContext executionContext;
  	private int depth = 0; // the bottom depth to go, not all depths may retrieve data/
  	private int colSize = 0;
  
  	public InfinispanExecution(Select query, RuntimeMetadata metadata,
  			InfinispanExecutionFactory factory, InfinispanConnection connection, ExecutionContext executionContext) throws TranslatorException {
  		this.factory = factory;
  		this.query = query;
  		this.connection = connection;
  		this.executionContext = executionContext;
  		this.scriptEngine = connection.getClassRegistry().getReadScriptEngine();
  
  		colSize = query.getDerivedColumns().size();
  
  		int numForeignKeys = 0;
  		ForeignKey fk = null;
  		// there should only be 1 table with a foreign key in the query
  		for (TableReference tr:query.getFrom()) {
  			List<ForeignKey> fkeys = ((NamedTable) tr).getMetadataObject().getForeignKeys();
  			if (fkeys.size() > 0) { 
  				numForeignKeys++;
  				fk = fkeys.get(0);
  			}
  		}
  		
  		if (numForeignKeys > 1) {
  			throw new TranslatorException(InfinispanPlugin.Util.gs(InfinispanPlugin.Event.TEIID25000, new Object[] {query}));
  		}
  		
  		colObjects = new Object[colSize];
  		
  		int col = 0;
  		for (DerivedColumn dc : query.getDerivedColumns()) {
  			ColumnReference cr = (ColumnReference) dc.getExpression();
  			String name = cr.getName();
  			
  			Column c = cr.getMetadataObject();
  			String nis = c.getSourceName();
  
  			if (nis.equalsIgnoreCase("this")) { //$NON-NLS-1$
  					// the object in cache is being requested
  					colObjects[col] = null;
  					
  					
  					// nis with a period indicates an internal class to the root
  			} else if (nis.indexOf(".") > 0)  {
  					if (fk == null) {
  						throw new TranslatorException(InfinispanPlugin.Util.gs(InfinispanPlugin.Event.TEIID25001, new Object[] {query.getFrom()}));
  					}
  						DepthNode dn = new DepthNode(fk.getSourceName() + "." + name, col);
   						colObjects[col] = dn;
  						 						
   						int n =dn.getNumberOfNodes();
  
  						if (n > depth) depth = n;
  					
  				} else {
  					
  					try {
  						colObjects[col] = getCompiledNode(name);
  					} catch (ScriptException e) {
  						throw new TranslatorException(e);
  					}
  				}
  			col++;
  		}
  	}
  
  	@Override
  	public void execute() throws TranslatorException {
  	
  		String nameInSource = ((NamedTable)query.getFrom().get(0)).getMetadataObject().getSourceName();
  	    
  		// column NIS for a column will be used to query the cache
  	    List<Object> objResults = factory.search(query, nameInSource, connection, executionContext);
  	    
  		if (objResults == null) {
  			objResults = Collections.emptyList();
  		}
  		LogManager.logDetail(LogConstants.CTX_CONNECTOR,
  				"InfinsipanExecution number of returned objects is :", objResults.size()); //$NON-NLS-1$
  
  		this.objResultsItr = objResults.iterator();
  	}
  
  	@SuppressWarnings("unchecked")
  	@Override
  	public List<Object> next() throws TranslatorException,
  			DataNotAvailableException {
  		if  (cacheResultsIt != null && cacheResultsIt.hasNext()) {
  		
  			// return the next row in the cache 
  			return (List<Object>) cacheResultsIt.next();
  		} 
  		
  		cacheResultsIt = null;
  		
  		// process the next object in the search result set
  		while (objResultsItr.hasNext()) {
  			List<Object> r = new ArrayList<Object>(colSize);
  			final Object o = objResultsItr.next();
  			sc.setAttribute(ClassRegistry.OBJECT_NAME, o, ScriptContext.ENGINE_SCOPE);			
  			
  			if (depth > 0) {
  				// this contains the object returned from a node that has depth
  				// key=value   ---  nodename=Object
  				// it should contain 2 types of relationships
  				// 1.  Collection based object (list, map, arraylist)
  				// 2.  non-primitive type object
  				Map<String, DepthNode> loadedDepths = new HashMap<String, DepthNode>();  // nodename ==> value
  				// process all the columns for a row, before processing depth
  				for (int i = 0; i < colSize; i++) {
  					if (colObjects[i] == null) {
  						r.add(o);
  						continue;
  					}					
  					
  					try {
  						if (colObjects[i] instanceof CompiledScript) {
  							CompiledScript cs = (CompiledScript) colObjects[i];
  							r.add(cs.eval(sc));
  						} else {
  							DepthNode dn = (DepthNode) colObjects[i];	
  							dn.reset();
  							Object depthObject = getCompiledNode(dn.getCurrentNodeName()).eval(sc);
  							dn.setValue(depthObject);
  							
  								// only include those that contain multiple objects
  							loadedDepths.put(dn.getDepthNodes(), dn);
  							r.add(dn);
  						}
  						
  					} catch (ScriptException e) {
  						throw new TranslatorException(e);
  					}
  				}
  				
  				final List<Object> rows = processDepth( r, loadedDepths);
  				
  				if (rows != null && rows.size() > 0) {
  				
  					if (rows.size() < 2) return (List<Object>) rows.get(0);
  					
  					cacheResultsIt = rows.iterator();
  					if (cacheResultsIt != null && cacheResultsIt.hasNext()) {
  						// return the next row in the cache 
  						return (List<Object>) cacheResultsIt.next();
  					} 
  
  				}
  			} else {
  
  				for (int i = 0; i < colSize; i++) {
  					if (colObjects[i] == null) {
  						r.add(o);
  						continue;
  					}
  					try {
  						CompiledScript cs = (CompiledScript) colObjects[i];					
  						r.add(cs.eval(sc));
  					} catch (ScriptException e) {
  						throw new TranslatorException(e);
  					}
  				}	
  				
  				return r;
  			}
  
  		}
  		return null;
  	}
  	
  	// the recursive logic will traverse to the bottom depth and work its way back up
  	// expanding the the number of rows
  	@SuppressWarnings("rawtypes")
  	private List<Object> processDepth(List<Object> mainRow, Map<String, DepthNode> loadedDepths) throws TranslatorException {
  		
  		// assumptions:
  		// -  only 1 1-to-many relationship can be specified per query, so if more than one column maps to a collection, 
  		//		its assumed they all map to the same collection
  		
  		Object[] cols = new Object[colSize];
  		Arrays.fill(cols,null);
  
  		int rows = 0;
  		for (Object r: mainRow) {
  			if (r instanceof DepthNode) {
  				DepthNode dn = (DepthNode) r;
  				dn.incrementPosition();
  