gt:         result = batchedObjects.get(sourceKey);
context: /*******************************************************************************
   * Copyright (c) 1998, 2013 Oracle and/or its affiliates. All rights reserved.
   * This program and the accompanying materials are made available under the 
   * terms of the Eclipse Public License v1.0 and Eclipse Distribution License v. 1.0 
   * which accompanies this distribution. 
   * The Eclipse Public License is available at http://www.eclipse.org/legal/epl-v10.html
   * and the Eclipse Distribution License is available at 
   * http://www.eclipse.org/org/documents/edl-v10.php.
   *
   * Contributors:
   *     Oracle - initial API and implementation from Oracle TopLink
   *     08/23/2010-2.2 Michael O'Brien 
   *        - 323043: application.xml module ordering may cause weaving not to occur causing an NPE.
   *                       warn if expected "_persistence_*_vh" method not found
   *                       instead of throwing NPE during deploy validation.
   *     11/19/2012-2.5 Guy Pelletier 
   *       - 389090: JPA 2.1 DDL Generation Support (foreign key metadata support)
   *     12/07/2012-2.5 Guy Pelletier 
   *       - 389090: JPA 2.1 DDL Generation Support (foreign key metadata support)
   ******************************************************************************/  
  package org.eclipse.persistence.mappings;
  
  import java.security.AccessController;
  import java.security.PrivilegedActionException;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Hashtable;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.Vector;
  
  import org.eclipse.persistence.annotations.BatchFetchType;
  import org.eclipse.persistence.descriptors.ClassDescriptor;
  import org.eclipse.persistence.descriptors.partitioning.PartitioningPolicy;
  import org.eclipse.persistence.exceptions.DatabaseException;
  import org.eclipse.persistence.exceptions.DescriptorException;
  import org.eclipse.persistence.exceptions.QueryException;
  import org.eclipse.persistence.exceptions.ValidationException;
  import org.eclipse.persistence.expressions.Expression;
  import org.eclipse.persistence.expressions.ExpressionBuilder;
  import org.eclipse.persistence.history.AsOfClause;
  import org.eclipse.persistence.indirection.ValueHolder;
  import org.eclipse.persistence.indirection.ValueHolderInterface;
  import org.eclipse.persistence.internal.descriptors.DescriptorIterator;
  import org.eclipse.persistence.internal.descriptors.InstanceVariableAttributeAccessor;
  import org.eclipse.persistence.internal.descriptors.MethodAttributeAccessor;
  import org.eclipse.persistence.internal.expressions.ForUpdateOfClause;
  import org.eclipse.persistence.internal.expressions.ObjectExpression;
  import org.eclipse.persistence.internal.helper.ClassConstants;
  import org.eclipse.persistence.internal.helper.DatabaseField;
  import org.eclipse.persistence.internal.helper.Helper;
  import org.eclipse.persistence.internal.helper.NonSynchronizedSubVector;
  import org.eclipse.persistence.internal.helper.NonSynchronizedVector;
  import org.eclipse.persistence.internal.identitymaps.CacheId;
  import org.eclipse.persistence.internal.identitymaps.CacheKey;
  import org.eclipse.persistence.internal.indirection.BasicIndirectionPolicy;
  import org.eclipse.persistence.internal.indirection.ContainerIndirectionPolicy;
  import org.eclipse.persistence.internal.indirection.DatabaseValueHolder;
  import org.eclipse.persistence.internal.indirection.IndirectionPolicy;
  import org.eclipse.persistence.internal.indirection.NoIndirectionPolicy;
  import org.eclipse.persistence.internal.indirection.WeavedObjectBasicIndirectionPolicy;
  import org.eclipse.persistence.internal.queries.AttributeItem;
  import org.eclipse.persistence.internal.queries.JoinedAttributeManager;
  import org.eclipse.persistence.internal.security.PrivilegedAccessHelper;
  import org.eclipse.persistence.internal.security.PrivilegedClassForName;
  import org.eclipse.persistence.internal.sessions.AbstractRecord;
  import org.eclipse.persistence.internal.sessions.AbstractSession;
  import org.eclipse.persistence.internal.sessions.ChangeRecord;
  import org.eclipse.persistence.internal.sessions.MergeManager;
  import org.eclipse.persistence.internal.sessions.UnitOfWorkChangeSet;
  import org.eclipse.persistence.internal.sessions.UnitOfWorkImpl;
  import org.eclipse.persistence.internal.sessions.remote.RemoteSessionController;
  import org.eclipse.persistence.internal.sessions.remote.RemoteValueHolder;
  import org.eclipse.persistence.logging.SessionLog;
  import org.eclipse.persistence.queries.BatchFetchPolicy;
  import org.eclipse.persistence.queries.Call;
  import org.eclipse.persistence.queries.DatabaseQuery;
  import org.eclipse.persistence.queries.FetchGroup;
  import org.eclipse.persistence.queries.ObjectBuildingQuery;
  import org.eclipse.persistence.queries.ObjectLevelModifyQuery;
  import org.eclipse.persistence.queries.ObjectLevelReadQuery;
  import org.eclipse.persistence.queries.ReadAllQuery;
  import org.eclipse.persistence.queries.ReadObjectQuery;
  import org.eclipse.persistence.queries.ReadQuery;
  import org.eclipse.persistence.queries.ReportQuery;
  import org.eclipse.persistence.sessions.DatabaseRecord;
  import org.eclipse.persistence.sessions.remote.DistributedSession;
  
  /**
   * <b>Purpose</b>: Abstract class for relationship mappings
   */
  public abstract class ForeignReferenceMapping extends DatabaseMapping {
      
      /** Query parameter name used for IN batch ids. */
      public static final String QUERY_BATCH_PARAMETER = "query-batch-parameter";
      
      /** This is used only in descriptor proxy in remote session */
      protected Class referenceClass;
      protected String referenceClassName;
  
      /** The session is temporarily used for initialization. Once used, it is set to null */
      protected transient AbstractSession tempInitSession;
  
      /** The descriptor of the reference class. */
      protected transient ClassDescriptor referenceDescriptor;
  
      /** This query is used to read referenced objects for this mapping. */
      protected ReadQuery selectionQuery;
  
      /** Indicates whether the referenced object is privately owned or not. */
      protected boolean isPrivateOwned;
  
      /**
       * Indicates whether the referenced object should always be batch read on read all queries,
       * and defines the type of batch fetch to use.
       */
      protected BatchFetchType batchFetchType;
  
      /** Implements indirection behavior */
      protected IndirectionPolicy indirectionPolicy;
  
      /** Indicates whether the selection query is TopLink generated or defined by the user. */
      protected transient boolean hasCustomSelectionQuery;
  
      /** Used to reference the other half of a bi-directional relationship. */
      protected DatabaseMapping relationshipPartner;
  
      /** Set by users, used to retrieve the backpointer for this mapping */
      protected String relationshipPartnerAttributeName;
  
      /** Cascading flags used by the EntityManager */
      protected boolean cascadePersist;
      protected boolean cascadeMerge;
      protected boolean cascadeRefresh;
      protected boolean cascadeRemove;
      protected boolean cascadeDetach;
      
      /** Flag used to determine if we need to weave the transient annotation on weaved fields.*/
      protected boolean requiresTransientWeavedFields;
      
      /** Define if the relationship should always be join fetched. */
      protected int joinFetch = NONE;
      /** Specify any INNER join on a join fetch. */
      public static final int INNER_JOIN = 1;
      /** Specify any OUTER join on a join fetch. */
      public static final int OUTER_JOIN = 2;
      /** Specify no join fetch, this is the default. */
      public static final int NONE = 0;
  
      /** This is a way (after cloning) to force the initialization of the selection criteria */
      protected boolean forceInitializationOfSelectionCriteria;
      
      /**
       * Indicates whether and how pessimistic lock scope should be extended 
       */
      enum ExtendPessimisticLockScope {
          // should not be extended.
          NONE,
          // should be extended in mapping's selectQuery.
          TARGET_QUERY,
          // should be extended in the source query.
          SOURCE_QUERY,
          // should be extended in a dedicated query (which doesn't do anything else).
          DEDICATED_QUERY
      }
      ExtendPessimisticLockScope extendPessimisticLockScope;
  
      /** Support delete cascading on the database relationship constraint. */
      protected boolean isCascadeOnDeleteSetOnDatabase;
  
      /** Allow the mapping's queries to be targeted at specific connection pools. */
      protected PartitioningPolicy partitioningPolicy;
      
      /** Allow the mapping's queries to be targeted at specific connection pools. */
      protected String partitioningPolicyName;
      
      /** Stores JPA metadata about whether another mapping is the owning mapping.  Only populated for JPA models **/
      protected String mappedBy;
      
      protected ForeignReferenceMapping() {
          this.isPrivateOwned = false;
          this.hasCustomSelectionQuery = false;
          this.useBasicIndirection();
          this.cascadePersist = false;
          this.cascadeMerge = false;
          this.cascadeRefresh = false;
          this.cascadeRemove = false;
          this.requiresTransientWeavedFields = true;
          this.forceInitializationOfSelectionCriteria = false;
          this.extendPessimisticLockScope = ExtendPessimisticLockScope.NONE;
      }
     
      /**
       * ADVANCED: Allows the retrieval of the owning mapping for a particular
       * mapping. Note: This will only be set for JPA models
       * 
       * @return
       */
      public String getMappedBy() {
          return mappedBy;
      }
  
      /**
       * PUBLIC:
       * Return the mapping's partitioning policy.
       */
      public PartitioningPolicy getPartitioningPolicy() {
          return partitioningPolicy;
      }
      
      /**
       * PUBLIC:
       * Set the mapping's partitioning policy.
       * A PartitioningPolicy is used to partition, load-balance or replicate data across multiple difference databases
       * or across a database cluster such as Oracle RAC.
       * Partitioning can provide improved scalability by allowing multiple database machines to service requests.
       * Setting a policy on a mapping will set the policy on all of its mappings.
       */
      public void setPartitioningPolicy(PartitioningPolicy partitioningPolicy) {
          this.partitioningPolicy = partitioningPolicy;
      }
      
      /**
       * PUBLIC:
       * Return the name of the mapping's partitioning policy.
       * A PartitioningPolicy with the same name must be defined on the Project.
       * A PartitioningPolicy is used to partition the data for a class across multiple difference databases
       * or across a database cluster such as Oracle RAC.
       * Partitioning can provide improved scalability by allowing multiple database machines to service requests.
       */
      public String getPartitioningPolicyName() {
          return partitioningPolicyName;
      }
      
      /**
       * PUBLIC:
       * Set the name of the mapping's partitioning policy.
       * A PartitioningPolicy with the same name must be defined on the Project.
       * A PartitioningPolicy is used to partition the data for a class across multiple difference databases
       * or across a database cluster such as Oracle RAC.
       * Partitioning can provide improved scalability by allowing multiple database machines to service requests.
       */
      public void setPartitioningPolicyName(String partitioningPolicyName) {
          this.partitioningPolicyName = partitioningPolicyName;
      }
  
      /**
       * INTERNAL:
       * Retrieve the value through using batch reading.
       * This executes a single query to read the target for all of the objects and stores the
       * result of the batch query in the original query to allow the other objects to share the results.
       */
      protected Object batchedValueFromRow(AbstractRecord row, ObjectLevelReadQuery query, CacheKey parentCacheKey) {
          ReadQuery batchQuery = (ReadQuery)query.getProperty(this);
          if (batchQuery == null) {
              if (query.hasBatchReadAttributes()) {
                  Map<DatabaseMapping, ReadQuery> queries = query.getBatchFetchPolicy().getMappingQueries();
                  if (queries != null) {
                      batchQuery = queries.get(this);
                  }
              }
              if (batchQuery == null) {
                  batchQuery = prepareNestedBatchQuery(query);
                  batchQuery.setIsExecutionClone(true);
              } else {
                  batchQuery = (ReadQuery)batchQuery.clone();
                  batchQuery.setIsExecutionClone(true);
              }
              query.setProperty(this, batchQuery);
          }
          return this.indirectionPolicy.valueFromBatchQuery(batchQuery, row, query, parentCacheKey);
      }
  
      /**
       * INTERNAL:
       * Clone the attribute from the clone and assign it to the backup.
       */
      @Override
      public void buildBackupClone(Object clone, Object backup, UnitOfWorkImpl unitOfWork) {
          Object attributeValue = getAttributeValueFromObject(clone);
          Object clonedAttributeValue = this.indirectionPolicy.backupCloneAttribute(attributeValue, clone, backup, unitOfWork);
          setAttributeValueInObject(backup, clonedAttributeValue);
      }
  
      /**
       * INTERNAL:
       * Used during building the backup shallow copy to copy the
       * target object without re-registering it.
       */
      @Override
      public abstract Object buildBackupCloneForPartObject(Object attributeValue, Object clone, Object backup, UnitOfWorkImpl unitOfWork);
  
      /**
       * INTERNAL:
       * Clone the attribute from the original and assign it to the clone.
       */
      @Override
      public void buildClone(Object original, CacheKey cacheKey, Object clone, Integer refreshCascade, AbstractSession cloningSession) {
          Object attributeValue = null;
          if (!this.isCacheable && (cacheKey != null && !cacheKey.isIsolated())){
              ReadObjectQuery query = new ReadObjectQuery(descriptor.getJavaClass());
              query.setSession(cloningSession);
              attributeValue = valueFromRow(cacheKey.getProtectedForeignKeys(), null, query, cacheKey, cloningSession, true, null);
          }else{
              attributeValue = getAttributeValueFromObject(original);
          }
          attributeValue = this.indirectionPolicy.cloneAttribute(attributeValue, original, cacheKey, clone, refreshCascade, cloningSession, false); // building clone from an original not a row.
          //GFBug#404 - fix moved to ObjectBuildingQuery.registerIndividualResult 
          setAttributeValueInObject(clone, attributeValue);
      }
  
      /**
       * INTERNAL:
       * A combination of readFromRowIntoObject and buildClone.
       * <p>
       * buildClone assumes the attribute value exists on the original and can
       * simply be copied.
       * <p>
       * readFromRowIntoObject assumes that one is building an original.
       * <p>
       * Both of the above assumptions are false in this method, and actually
       * attempts to do both at the same time.
       * <p>
       * Extract value from the row and set the attribute to this value in the
       * working copy clone.
       * In order to bypass the shared cache when in transaction a UnitOfWork must
       * be able to populate working copies directly from the row.
       */
      @Override
      public void buildCloneFromRow(AbstractRecord databaseRow, JoinedAttributeManager joinManager, Object clone, CacheKey sharedCacheKey, ObjectBuildingQuery sourceQuery, UnitOfWorkImpl unitOfWork, AbstractSession executionSession) {
          Boolean[] wasCacheUsed = new Boolean[]{Boolean.FALSE};
          Object attributeValue = valueFromRow(databaseRow, joinManager, sourceQuery, sharedCacheKey, executionSession, true, wasCacheUsed);
          Object clonedAttributeValue = this.indirectionPolicy.cloneAttribute(attributeValue, null, sharedCacheKey,clone, null, unitOfWork, !wasCacheUsed[0]);// building clone directly from row.
          if (executionSession.isUnitOfWork() && sourceQuery.shouldRefreshIdentityMapResult()){
              // check whether the attribute is fully build before calling getAttributeValueFromObject because that
              // call may fully build the attribute
              boolean wasAttributeValueFullyBuilt = isAttributeValueFullyBuilt(clone);
              Object oldAttribute = this.getAttributeValueFromObject(clone);
              setAttributeValueInObject(clone, clonedAttributeValue); // set this first to prevent infinite recursion
               if (wasAttributeValueFullyBuilt && this.indirectionPolicy.objectIsInstantiatedOrChanged(oldAttribute)){
                  this.indirectionPolicy.instantiateObject(clone, clonedAttributeValue);
              }
          }else{
              setAttributeValueInObject(clone, clonedAttributeValue);
          }
          if((joinManager != null && joinManager.isAttributeJoined(this.descriptor, this)) || (isExtendingPessimisticLockScope(sourceQuery) && extendPessimisticLockScope == ExtendPessimisticLockScope.TARGET_QUERY)) {
              // need to instantiate to extended the lock beyond the source object table(s).
              this.indirectionPolicy.instantiateObject(clone, clonedAttributeValue);
          }
      }
  
      /**
       * INTERNAL:
       * Require for cloning, the part must be cloned.
       */
      public abstract Object buildCloneForPartObject(Object attributeValue, Object original, CacheKey cacheKey, Object clone, AbstractSession cloningSession, Integer refreshCascade, boolean isExisting, boolean isFromSharedCache);
  
      /**
       * INTERNAL:
       * The mapping clones itself to create deep copy.
       */
      @Override
      public Object clone() {
          ForeignReferenceMapping clone = (ForeignReferenceMapping)super.clone();
  
          clone.setIndirectionPolicy((IndirectionPolicy)indirectionPolicy.clone());
          clone.setSelectionQuery((ReadQuery)getSelectionQuery().clone());
  
          return clone;
      }
  
      /**
       * INTERNAL:
       * This method will access the target relationship and create a list of information to rebuild the relationship.
       * This method is used in combination with the CachedValueHolder to store references to PK's to be loaded from
       * a cache instead of a query.
       */
      public abstract Object[] buildReferencesPKList(Object entity, Object attribute, AbstractSession session);
  
      /**
       * INTERNAL: Compare the attributes belonging to this mapping for the
       * objects.
       */
      @Override
      public boolean compareObjects(Object firstObject, Object secondObject, AbstractSession session) {
          if (isPrivateOwned()) {
              return compareObjectsWithPrivateOwned(firstObject, secondObject, session);
          } else {
              return compareObjectsWithoutPrivateOwned(firstObject, secondObject, session);
          }
      }
  
      /**
       * Compare two objects if their parts are not private owned
       */
      protected abstract boolean compareObjectsWithoutPrivateOwned(Object first, Object second, AbstractSession session);
  
      /**
       * Compare two objects if their parts are private owned
       */
      protected abstract boolean compareObjectsWithPrivateOwned(Object first, Object second, AbstractSession session);
  
      /**
       * INTERNAL:
       * Convert all the class-name-based settings in this mapping to actual class-based
       * settings. This method is used when converting a project that has been built
       * with class names to a project with classes.
       */
      @Override
      public void convertClassNamesToClasses(ClassLoader classLoader){
          super.convertClassNamesToClasses(classLoader);
  
          // DirectCollection mappings don't require a reference class.
          if (getReferenceClassName() != null) {
              Class referenceClass = null;
              try{
                  if (PrivilegedAccessHelper.shouldUsePrivilegedAccess()){
                      try {
                          referenceClass = (Class)AccessController.doPrivileged(new PrivilegedClassForName(getReferenceClassName(), true, classLoader));
                      } catch (PrivilegedActionException exception) {
                          throw ValidationException.classNotFoundWhileConvertingClassNames(getReferenceClassName(), exception.getException());
                      }
                  } else {
                      referenceClass = org.eclipse.persistence.internal.security.PrivilegedAccessHelper.getClassForName(getReferenceClassName(), true, classLoader);
                  }
              } catch (ClassNotFoundException exc){
                  throw ValidationException.classNotFoundWhileConvertingClassNames(getReferenceClassName(), exc);
              }
              setReferenceClass(referenceClass);
          }
          if (getSelectionQuery() != null) {
              getSelectionQuery().convertClassNamesToClasses(classLoader);
          }
      }
  
      /**
       * INTERNAL:
       * Builder the unit of work value holder.
       * Ignore the original object.
       * @param buildDirectlyFromRow indicates that we are building the clone directly
       * from a row as opposed to building the original from the row, putting it in
       * the shared cache, and then cloning the original.
       */
      public DatabaseValueHolder createCloneValueHolder(ValueHolderInterface attributeValue, Object original, Object clone, AbstractRecord row, AbstractSession cloningSession, boolean buildDirectlyFromRow) {
          return cloningSession.createCloneQueryValueHolder(attributeValue, clone, row, this);
      }
  
      /**
       * INTERNAL:
       * Return true if the merge should be bypassed. This would be the case for several reasons, depending on
       * the kind of merge taking place.
       */
      protected boolean dontDoMerge(Object target, Object source, MergeManager mergeManager) {
          if (!shouldMergeCascadeReference(mergeManager)) {
              return true;
          }
          if (mergeManager.isForRefresh()) {
              // For reverts we are more concerned about the target than the source.
              if (!isAttributeValueInstantiated(target)) {
                  return true;
              }
          } else {
              if (mergeManager.shouldRefreshRemoteObject() && shouldMergeCascadeParts(mergeManager) && usesIndirection()) {
                  return true;
              } else {
                  if (!isAttributeValueInstantiated(source)) {
                      return true;
                  }
              }
          }
          return false;
      }
  
      /**
       * PUBLIC:
       * Indicates whether the referenced object should always be batch read on read all queries.
       * Batch reading will read all of the related objects in a single query when accessed from an originating read all.
       * This should only be used if it is know that the related objects are always required with the source object, or indirection is not used.
       */
      public void dontUseBatchReading() {
          setUsesBatchReading(false);
      }
  
      /**
       * PUBLIC:
       * Indirection means that a ValueHolder will be put in-between the attribute and the real object.
       * This allows for the reading of the target from the database to be delayed until accessed.
       * This defaults to true and is strongly suggested as it give a huge performance gain.
       */
      public void dontUseIndirection() {
          setIndirectionPolicy(new NoIndirectionPolicy());
      }
  
      /**
       * INTERNAL:
       * Called if shouldExtendPessimisticLockScopeInTargetQuery() is true.
       * Adds locking clause to the target query to extend pessimistic lock scope.
       */
      protected void extendPessimisticLockScopeInTargetQuery(ObjectLevelReadQuery targetQuery, ObjectBuildingQuery sourceQuery) {
          targetQuery.setLockMode(sourceQuery.getLockMode());
      }
      
      /**
       * INTERNAL:
       * Called if shouldExtendPessimisticLockScopeInSourceQuery is true.
       * Adds fields to be locked to the where clause of the source query.
       * Note that the sourceQuery must be ObjectLevelReadQuery so that it has ExpressionBuilder.
       * 
       * This method must be implemented in subclasses that allow
       * setting shouldExtendPessimisticLockScopeInSourceQuery to true.
       */
      public void extendPessimisticLockScopeInSourceQuery(ObjectLevelReadQuery sourceQuery) {
      }
      
      /**
       * INTERNAL:
       * Extract the value from the batch optimized query, this should be supported by most query types.
       */
      public Object extractResultFromBatchQuery(ReadQuery batchQuery, CacheKey parentCacheKey, AbstractRecord sourceRow, AbstractSession session, ObjectLevelReadQuery originalQuery) throws QueryException {
          Map<Object, Object> batchedObjects = null;
          Object result = null;
          Object sourceKey = extractBatchKeyFromRow(sourceRow, session);
          if (sourceKey == null) {
              // If the foreign key was null, then just return null.
              return null;
          }
          Object cachedObject = checkCacheForBatchKey(sourceRow, sourceKey, batchedObjects, batchQuery, originalQuery, session);
          if (cachedObject != null) {
              // If the object is already in the cache, then just return it.
              return cachedObject;
          }
          // Ensure the query is only executed once.
          synchronized (batchQuery) {
              // Check if query was already executed.
              batchedObjects = batchQuery.getBatchObjects();
              BatchFetchPolicy originalPolicy = originalQuery.getBatchFetchPolicy();
              if (batchedObjects == null) {
                  batchedObjects = new Hashtable();
                  batchQuery.setBatchObjects(batchedObjects);
              } else {
                  result = batchedObjects.get(sourceKey);
                  if (result == Helper.NULL_VALUE) {
                      return null;
                  // If IN may not have that batch yet, or it may have been null.
                  } else if ((result != null) || (!originalPolicy.isIN())) {
                      return result;
                  }
              }
              // In case of IN the batch including this row may not have been executed yet.
              if (result == null) {
                  AbstractRecord translationRow = originalQuery.getTranslationRow();
                  // Execute query and index resulting object sets by key.
                  if (originalPolicy.isIN()) {
                      // Need to extract all foreign key values from all parent rows for IN parameter.
                      List<AbstractRecord> parentRows = originalPolicy.getDataResults(this);
                      // Execute queries by batch if too many rows.
                      int rowsSize = parentRows.size();
                      int size = Math.min(rowsSize, originalPolicy.getSize());
                      if (size == 0) {
                          return null;
                      }
                      int startIndex = 0;
                      if (size != rowsSize) {
                          // If only fetching a page, need to make sure the row we want is in the page.
                          startIndex = parentRows.indexOf(sourceRow);
                      }
                      List foreignKeyValues = new ArrayList(size);
                      Set foreignKeys = new HashSet(size);
                      int index = 0;
                      int offset = startIndex;
                      for (int count = 0; count < size; count++) {
                          if (index >= rowsSize) {
                              // Processed all rows, done.
                              break;
                          } else if ((offset + index) >= rowsSize) {
                              // If passed the end, go back to start.
                              offset = index * -1;
                          }
                          AbstractRecord row = parentRows.get(offset + index);
                          // Handle duplicate rows in the ComplexQueryResult being replaced with null, as a
                          // result of duplicate filtering being true for constructing the ComplexQueryResult
                          if (row != null) {
                              Object foreignKey = extractBatchKeyFromRow(row, session);
                              if (foreignKey == null) {
                                  // Ignore null foreign keys.
                                  count--;
                              } else {
                                  cachedObject = checkCacheForBatchKey(row, foreignKey, batchedObjects, batchQuery, originalQuery, session);
                                  if (cachedObject != null) {
                                      // Avoid fetching things a cache hit occurs for.
                                      count--;                                
                                  } else {
                                      // Ensure the same id is not selected twice.
                                      if (foreignKeys.contains(foreignKey)) {
                                          count--;
                                      } else {
                                          Object[] key = ((CacheId)foreignKey).getPrimaryKey();
                                          Object foreignKeyValue = key[0];
                                          // Support composite keys using nested IN.
                                          if (key.length > 1) {
                                              foreignKeyValue = Arrays.asList(key);
                                          }
                                          foreignKeyValues.add(foreignKeyValue);
                                          foreignKeys.add(foreignKey);
                                      }
                                  }
                              }
                          }
                          index++;
                      }
                      // Need to compute remaining rows, this is tricky because a page in the middle could have been processed.
                      List<AbstractRecord> remainingParentRows = null;
                      if (startIndex == 0) {
                          // Tail
                          remainingParentRows = new ArrayList(parentRows.subList(index, rowsSize));
                      } else if (startIndex == offset) {
                          // Head and tail.
                          remainingParentRows = new ArrayList(parentRows.subList(0, startIndex - 1));
                          remainingParentRows.addAll(parentRows.subList(index, rowsSize));
                      } else {
                          // Middle
                          // Check if empty,
                          if ((offset + index) >= (startIndex - 1)) {
                              remainingParentRows = new ArrayList(0);
                          } else {
                              remainingParentRows = new ArrayList(parentRows.subList(offset + index, startIndex - 1));
                          }
                      }
                      originalPolicy.setDataResults(this, remainingParentRows);
                      translationRow = translationRow.clone();
                      translationRow.put(QUERY_BATCH_PARAMETER, foreignKeyValues);
                      // Register each id as null, in case it has no relationship.
                      for (Object foreignKey : foreignKeys) {
                          batchedObjects.put(foreignKey, Helper.NULL_VALUE);
                      }
                  } else if (batchQuery.isReadAllQuery() && ((ReadAllQuery)batchQuery).getBatchFetchPolicy().isIN()) {
                      throw QueryException.originalQueryMustUseBatchIN(this, originalQuery);
                  }
                  executeBatchQuery(batchQuery, parentCacheKey, batchedObjects, session, translationRow);
                  batchQuery.setSession(null);
              }
          }
  