/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You 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 org.apache.geode.cache.query.internal;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.cache.query.FunctionDomainException;
import org.apache.geode.cache.query.Index;
import org.apache.geode.cache.query.NameResolutionException;
import org.apache.geode.cache.query.QueryInvocationTargetException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.Struct;
import org.apache.geode.cache.query.TypeMismatchException;
import org.apache.geode.cache.query.internal.index.AbstractIndex;
import org.apache.geode.cache.query.internal.index.IndexData;
import org.apache.geode.cache.query.internal.index.IndexManager;
import org.apache.geode.cache.query.internal.index.IndexProtocol;
import org.apache.geode.cache.query.internal.index.IndexUtils;
import org.apache.geode.cache.query.internal.index.MemoryIndexStore;
import org.apache.geode.cache.query.internal.index.MemoryIndexStore.CachedEntryWrapper;
import org.apache.geode.cache.query.internal.index.PartitionedIndex;
import org.apache.geode.cache.query.internal.parse.OQLLexerTokenTypes;
import org.apache.geode.cache.query.internal.types.ObjectTypeImpl;
import org.apache.geode.cache.query.internal.types.StructTypeImpl;
import org.apache.geode.cache.query.types.CollectionType;
import org.apache.geode.cache.query.types.ObjectType;
import org.apache.geode.cache.query.types.StructType;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.BucketRegion;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.logging.internal.log4j.api.LogService;

public class QueryUtils {
  private static final Logger logger = LogService.getLogger();

  /**
   * Return a SelectResults that is the intersection of c1 and c2. May or may not return a modified
   * c1 or c2.
   */
  public static SelectResults intersection(SelectResults c1, SelectResults c2,
      ExecutionContext contextOrNull) {
    QueryObserverHolder.getInstance().invokedQueryUtilsIntersection(c1, c2);
    assertCompatible(c1, c2);
    if (c1.isEmpty()) {
      return c1;
    }
    if (c2.isEmpty()) {
      return c2;
    }
    // iterate on the smallest one
    if (c1.size() < c2.size()) {
      return sizeSortedIntersection(c1, c2, contextOrNull);
    } else {
      return sizeSortedIntersection(c2, c1, contextOrNull);
    }
  }

  /**
   * Return a SelectResults that is the union of c1 and c2. May or may not return a modified c1 or
   * c2.
   */
  public static SelectResults union(SelectResults c1, SelectResults c2,
      ExecutionContext contextOrNull) {
    QueryObserverHolder.getInstance().invokedQueryUtilsUnion(c1, c2);
    assertCompatible(c1, c2);
    // iterate on the smallest one
    if (c1.size() < c2.size()) {
      return sizeSortedUnion(c1, c2, contextOrNull);
    } else {
      return sizeSortedUnion(c2, c1, contextOrNull);
    }
  }

  private static void assertCompatible(SelectResults sr1, SelectResults sr2) {
    Assert.assertTrue(
        sr1.getCollectionType().getElementType().equals(sr2.getCollectionType().getElementType()));
  }


  public static SelectResults createResultCollection(ExecutionContext context,
      ObjectType elementType) {
    return context.isDistinct() ? new ResultsSet(elementType)
        : new ResultsBag(elementType, context.getCachePerfStats());
  }

  public static SelectResults createStructCollection(ExecutionContext context,
      StructType elementType) {
    return context.isDistinct() ? new StructSet(elementType)
        : new StructBag(elementType, context.getCachePerfStats());
  }

  public static SelectResults createResultCollection(boolean distinct, ObjectType elementType,
      ExecutionContext context) {
    return distinct ? new ResultsSet(elementType)
        : new ResultsBag(elementType, context.getCachePerfStats());
  }

  public static SelectResults createStructCollection(boolean distinct, StructType elementType,
      ExecutionContext context) {
    return distinct ? new StructSet(elementType)
        : new StructBag(elementType, context.getCachePerfStats());
  }

  /**
   * Returns an appropriate, empty {@code SelectResults}
   *
   * @param objectType The {@code ObjectType} of the query results
   * @return an appropriate, empty {@code SelectResults}
   */
  public static SelectResults getEmptySelectResults(ObjectType objectType,
      CachePerfStats statsOrNull) {
    SelectResults emptyResults = null;
    if (objectType instanceof StructType) {
      emptyResults = new StructBag((StructTypeImpl) objectType, statsOrNull);
    } else {
      emptyResults = new ResultsBag(objectType, statsOrNull);
    }
    return emptyResults;
  }

  /**
   * Returns an appropriate, empty {@code SelectResults}
   *
   * TODO: statsOrNull is always null
   *
   * @param collectionType The {@code CollectionType} of the query results
   * @return an appropriate, empty {@code SelectResults}
   */
  public static SelectResults getEmptySelectResults(CollectionType collectionType,
      CachePerfStats statsOrNull) {
    SelectResults emptyResults = null;
    if (collectionType.isOrdered()) {
      // The collectionType is ordered.
      // The 'order by' clause was used in the query.
      // Wrap an ArrayList with a ResultsCollectionWrapper
      emptyResults = new ResultsCollectionWrapper(collectionType.getElementType(), new ArrayList());
    } else if (!collectionType.allowsDuplicates()) {
      // The collectionType does not allow duplicates.
      // The distinct keyword was used in the query.
      // Wrap a HashSet with a ResultsCollectionWrapper
      emptyResults = new ResultsCollectionWrapper(collectionType.getElementType(), new HashSet());
    } else {
      // Use ObjectType to determine the correct SelectResults implementation
      emptyResults = getEmptySelectResults(collectionType.getElementType(), statsOrNull);
    }
    return emptyResults;
  }

  // convenience method
  private static boolean isBag(SelectResults coln) {
    return coln.getCollectionType().allowsDuplicates();
  }

  /** collections are passed in from smallest to largest */
  // consider: taking intersection of two bags, only retain
  // the number of occurrences in the intersection equal to the
  // minimum number between the two bags
  private static SelectResults sizeSortedIntersection(SelectResults small, SelectResults large,
      ExecutionContext contextOrNull) {
    // if one is a set and one is a bag, then treat the set like a bag (and return a bag)
    boolean smallModifiable = small.isModifiable() && (isBag(small) || !isBag(large));
    boolean largeModifiable = large.isModifiable() && (isBag(large) || !isBag(small));
    if (smallModifiable) {
      try {
        for (Iterator itr = small.iterator(); itr.hasNext();) {
          Object element = itr.next();
          if (element instanceof MemoryIndexStore.CachedEntryWrapper) {
            element = ((CachedEntryWrapper) element).getKey();
          }
          int count = large.occurrences(element);
          if (small.occurrences(element) > count) {
            // bag intersection: only retain smaller number of dups
            itr.remove();
          }
        }
        return small;
      } catch (UnsupportedOperationException ignore) {
        // didn't succeed because small is actually unmodifiable
      }
    }
    if (largeModifiable) {
      try {
        for (Iterator itr = large.iterator(); itr.hasNext();) {
          Object element = itr.next();
          if (element instanceof MemoryIndexStore.CachedEntryWrapper) {
            element = ((CachedEntryWrapper) element).getKey();
          }
          int count = small.occurrences(element);
          if (large.occurrences(element) > count) {
            // bag intersection: only retain smaller number of dups
            itr.remove();
          }
        }
        return large;
      } catch (UnsupportedOperationException ignore) {
        // didn't succeed because large is actually unmodifiable
      }
    }

    SelectResults rs;
    if (contextOrNull != null) {
      rs = contextOrNull.isDistinct() ? new ResultsSet(small)
          : new ResultsBag(small, contextOrNull.getCachePerfStats());
    } else {
      rs = new ResultsBag(small, null);
    }

    for (Iterator itr = rs.iterator(); itr.hasNext();) {
      Object element = itr.next();
      int count = large.occurrences(element);
      if (rs.occurrences(element) > count) { // bag intersection: only retain smaller number of dups
        itr.remove();
      }
    }
    return rs;
  }

  /** collections are passed in from smallest to largest */
  // assume we're dealing with bags and/or sets here, number of occurrences in the
  // union should be the sum of the occurrences in the two bags
  // Is this Ok? There may be tuples which are actually common to both set so
  // union in such cases should not increase count. right.?
  private static SelectResults sizeSortedUnion(SelectResults small, SelectResults large,
      ExecutionContext contextOrNull) {
    // if one is a set and one is a bag, then treat the set like a bag (and return a bag)
    boolean smallModifiable = small.isModifiable() && (isBag(small) || !isBag(large));
    boolean largeModifiable = large.isModifiable() && (isBag(large) || !isBag(small));
    if (largeModifiable) {
      try {
        for (Object element : small) {
          int count = small.occurrences(element);
          if (large.occurrences(element) < count) {
            large.add(element);
          }
        }
        return large;
      } catch (UnsupportedOperationException ignore) {
        // didn't succeed because large is actually unmodifiable
      }
    }
    if (smallModifiable) {
      try {
        for (Object element : large) {
          int count = large.occurrences(element);
          if (small.occurrences(element) < count) {
            small.add(element);
          }
        }
        return small;
      } catch (UnsupportedOperationException ignore) {
        // didn't succeed because small is actually unmodifiable
      }
    }
    SelectResults rs;
    if (contextOrNull != null) {
      rs = contextOrNull.isDistinct() ? new ResultsSet(large)
          : new ResultsBag(large, contextOrNull.getCachePerfStats());
    } else {
      rs = new ResultsBag(large, null);
    }

    for (Object element : small) {
      rs.add(element);
    }
    return rs;
  }

  /**
   * This function returns a list of runtime iterators in current scope which are exclusively
   * dependent on given independent RuntimeIterators. The order of dependent iterators in the List
   * is based on the order of independent Iterators present in the array . For each group the first
   * iterator is its independent iterator
   *
   * @param indpndntItrs array of independent RuntimeIterators
   */
  static List getDependentItrChainForIndpndntItrs(RuntimeIterator[] indpndntItrs,
      ExecutionContext context) {
    List ret = new ArrayList();
    for (RuntimeIterator indpndntItr : indpndntItrs) {
      ret.addAll(context.getCurrScopeDpndntItrsBasedOnSingleIndpndntItr(indpndntItr));
    }
    return ret;
  }

  /**
   * This util function does a cartesian of the array of SelectResults object , expanding the
   * resultant set to the number of iterators passed in expansionList. The position of the iterator
   * fields in the final result is governed by the order of RuntimeIterators present in the
   * finalList. If any condition needs to be evaluated during cartesian , it can be passed as
   * operand
   *
   * @param results Array of SelectResults object which are to be cartesianed
   * @param itrsForResultFields A two dimensional array of RuntimeIterator. Each row of this two
   *        dimensional RuntimeIterator array , maps to a SelectResults object in the results array.
   *        Thus the 0th row of two dimensional RuntimeIterator array will map to the 0th element of
   *        the SelectResults array. The order of RuntimeIterator in a row will map to the fields in
   *        the SelectResults object. The 0th RuntimeIterator will map to the 0th field of the
   *        corresponding SelectResults object. The number of rows in the two dimensional array of
   *        RuntimeIterator should be equal to the size of array of SelectResults object passed and
   *        the number of RuntimeIterators in each row should be equal to the number of fields in
   *        the SelectResults object . The SelectResults object itself may be a ResultBag object or
   *        a StructBag object.
   *
   * @param expansionList List containing RunimeIterators to which the final Results should be
   *        expanded to.
   * @param finalList List containing RuntimeIterators which define the number of fields to be
   *        present in the resultant SelectResults and their relative positions. The Runtime
   *        Iterators present in the List should be either available in the expansion List or should
   *        be present in each row of the two dimensional RuntimeIterator array.
   *
   * @param context ExecutionContext object
   * @param operand The CompiledValue which needs to be iter evaluated during cartesian. Only those
   *        tuples will be selected in the final Result for which oerand evaluates to true.
   * @return SelectResults object representing the final result of the cartesian
   */
  public static SelectResults cartesian(SelectResults[] results,
      RuntimeIterator[][] itrsForResultFields, List expansionList, List finalList,
      ExecutionContext context, CompiledValue operand) throws FunctionDomainException,
      TypeMismatchException, NameResolutionException, QueryInvocationTargetException {
    SelectResults returnSet = null;
    if (finalList.size() == 1) {
      ObjectType type = ((RuntimeIterator) finalList.iterator().next()).getElementType();
      if (type instanceof StructType) {
        returnSet = QueryUtils.createStructCollection(context, (StructTypeImpl) type);
      } else {
        return QueryUtils.createResultCollection(context, type);
      }
    } else {
      StructType structType = createStructTypeForRuntimeIterators(finalList);
      returnSet = QueryUtils.createStructCollection(context, structType);
    }
    ListIterator expnItr = expansionList.listIterator();
    doNestedIterations(0, returnSet, results, itrsForResultFields, finalList, expnItr,
        results.length + expansionList.size(), context, operand);
    return returnSet;
  }

  // TODO:Optimize the function further in terms of reducing the
  // parameters passed in the function, if possible
  private static void doNestedIterations(int level, SelectResults returnSet,
      SelectResults[] results, RuntimeIterator[][] itrsForResultFields, List finalItrs,
      ListIterator expansionItrs, int finalLevel, ExecutionContext context, CompiledValue operand)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {
    if (level == finalLevel) {
      // end recusrion
      boolean select = true;
      if (operand != null) {
        select = applyCondition(operand, context);
      }
      Iterator itr = finalItrs.iterator();
      int len = finalItrs.size();
      if (len > 1) {
        Object[] values = new Object[len];
        int j = 0;
        while (itr.hasNext()) {
          values[j++] = ((RuntimeIterator) itr.next()).evaluate(context);
        }
        if (select) {

          ((StructFields) returnSet).addFieldValues(values);

        }
      } else {
        if (select) {
          returnSet.add(((RuntimeIterator) itr.next()).evaluate(context));
        }
      }
    } else if (level < results.length) {
      SelectResults individualResultSet = results[level];
      RuntimeIterator[] itrsForFields = itrsForResultFields[level];
      int len = itrsForFields.length;
      for (Object anIndividualResultSet : individualResultSet) {
        // Check if query execution on this thread is canceled.
        QueryMonitor.throwExceptionIfQueryOnCurrentThreadIsCanceled();
        if (len == 1) {
          // this means we have a ResultSet
          itrsForFields[0].setCurrent(anIndividualResultSet);
        } else {
          Struct struct = (Struct) anIndividualResultSet;
          Object[] fieldValues = struct.getFieldValues();
          int size = fieldValues.length;
          for (int i = 0; i < size; ++i) {
            itrsForFields[i].setCurrent(fieldValues[i]);
          }
        }
        doNestedIterations(level + 1, returnSet, results, itrsForResultFields, finalItrs,
            expansionItrs, finalLevel, context, operand);
      }
    } else {
      RuntimeIterator currLevel = (RuntimeIterator) expansionItrs.next();
      SelectResults c = currLevel.evaluateCollection(context);
      if (c == null) {
        expansionItrs.previous();
        return;
      }
      for (Object aC : c) {
        currLevel.setCurrent(aC);
        doNestedIterations(level + 1, returnSet, results, itrsForResultFields, finalItrs,
            expansionItrs, finalLevel, context, operand);
      }
      expansionItrs.previous();
    }
  }

  public static boolean applyCondition(CompiledValue operand, ExecutionContext context)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {
    if (operand == null) {
      return true;
    }
    Object result = operand.evaluate(context);
    if (result instanceof Boolean) {
      return (Boolean) result;
    } else if (result != null && result != QueryService.UNDEFINED) {
      throw new TypeMismatchException(
          String.format("AND/OR operands must be of type boolean, not type ' %s '",
              result.getClass().getName()));
    } else {
      return false;
    }
  }

  /**
   * NOTE: intermediateResults should be a single element array
   * <p>
   * NOTE: itrsForIntermediateResults should be a two dimensional array but with only one row
   * <p>
   * TODO: This function is used to do cartesian of index resultset while
   * <p>
   * expanding/cutting down index resultset with the intermediate resultset
   */
  private static void mergeRelationshipIndexResultsWithIntermediateResults(SelectResults returnSet,
      SelectResults[] intermediateResults, RuntimeIterator[][] itrsForIntermediateResults,
      Object[][] indexResults, RuntimeIterator[][] indexFieldToItrsMapping,
      ListIterator expansionListItr, List finalItrs, ExecutionContext context, List[] checkList,
      CompiledValue iterOps, IndexCutDownExpansionHelper[] icdeh, int level,
      int maxExpnCartesianDepth) throws FunctionDomainException, TypeMismatchException,
      NameResolutionException, QueryInvocationTargetException {

    int resultSize = indexResults[level].length;
    // TODO: Since this is constant for a given merge call, pass it as a
    // parameter to
    // the function rather than calling it every time
    for (int j = 0; j < resultSize; ++j) {
      if (setIndexFieldValuesInRespectiveIterators(indexResults[level][j],
          indexFieldToItrsMapping[level], icdeh[level])) {
        if (level == indexResults.length - 1) {
          // Set the values in the Intermedaite Resultset
          doNestedIterations(0, returnSet, intermediateResults, itrsForIntermediateResults,
              finalItrs, expansionListItr, maxExpnCartesianDepth, context, iterOps);
        } else {
          mergeRelationshipIndexResultsWithIntermediateResults(returnSet, intermediateResults,
              itrsForIntermediateResults, indexResults, indexFieldToItrsMapping, expansionListItr,
              finalItrs, context, checkList, iterOps, icdeh, level + 1, maxExpnCartesianDepth);
          if (icdeh[level + 1].cutDownNeeded) {
            icdeh[level + 1].checkSet.clear();
          }
        }
      }
    }
  }

  // TODO: Test the function & write expnanation of the parameters
  private static void mergeAndExpandCutDownRelationshipIndexResults(Object[][] values,
      SelectResults result, RuntimeIterator[][] indexFieldToItrsMapping,
      ListIterator expansionListIterator, List finalItrs, ExecutionContext context,
      CompiledValue iterOps, IndexCutDownExpansionHelper[] icdeh, int level)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {

    int resultSize = values[level].length;
    // We do not know if the first X results might or might not fulfill all operands.
    Boolean applyLimit = (Boolean) context.cacheGet(CompiledValue.CAN_APPLY_LIMIT_AT_INDEX);
    int limit = (applyLimit != null && applyLimit) ? getLimitValue(context) : -1;

    // Stops recursion if limit has already been met AND limit can be applied to index.
    if (limit != -1 && result.size() >= limit) {
      return;
    }

    for (int j = 0; j < resultSize; ++j) {
      // Check if query execution on this thread is canceled.
      QueryMonitor.throwExceptionIfQueryOnCurrentThreadIsCanceled();

      if (setIndexFieldValuesInRespectiveIterators(values[level][j], indexFieldToItrsMapping[level],
          icdeh[level])) {
        if (level == values.length - 1) {
          doNestedIterationsForIndex(expansionListIterator.hasNext(), result, finalItrs,
              expansionListIterator, context, iterOps, limit, null);
          if (limit != -1 && result.size() >= limit) {
            break;
          }
        } else {
          mergeAndExpandCutDownRelationshipIndexResults(values, result, indexFieldToItrsMapping,
              expansionListIterator, finalItrs, context, iterOps, icdeh, level + 1);
          if (icdeh[level + 1].cutDownNeeded) {
            icdeh[level + 1].checkSet.clear();
          }
        }
      }
    }
  }

  // TODO: Explain the function & write test cases. A boolean false means by pass i.e the set value
  // to be ignored
  // End result if we have not already expanded is that we have created a new struct and added to a
  // set to prevent future expansions of the same object
  // It also advances the current object for the iterator.
  private static boolean setIndexFieldValuesInRespectiveIterators(Object value,
      RuntimeIterator[] indexFieldToItrsMapping, IndexCutDownExpansionHelper icdeh) {
    boolean select = true;
    int len = indexFieldToItrsMapping.length;
    if (len == 1) {
      // this means we have a ResultSet
      Support.Assert(!icdeh.cutDownNeeded,
          "If the index fields to iter mapping is of of size 1 then cut down cannot occur");
      indexFieldToItrsMapping[0].setCurrent(value);
    } else {
      Struct struct = (Struct) value;
      Object[] fieldValues = struct.getFieldValues();
      int size = fieldValues.length;
      Object[] checkFields = null;
      if (icdeh.cutDownNeeded) {
        checkFields = new Object[icdeh.checkSize];
      }
      // Object values[] = new Object[numItersInResultSet];
      int j = 0;
      RuntimeIterator rItr = null;
      for (int i = 0; i < size; i++) {
        rItr = indexFieldToItrsMapping[i];
        if (rItr != null) {
          rItr.setCurrent(fieldValues[i]);
          if (icdeh.cutDownNeeded) {
            checkFields[j++] = fieldValues[i];
          }
        }
      }
      if (icdeh.cutDownNeeded) {
        Object temp = null;
        if (icdeh.checkSize == 1) {
          temp = checkFields[0];
        } else {
          temp = new StructImpl((StructTypeImpl) icdeh.checkType, checkFields);
        }
        if (icdeh.checkSet.contains(temp)) {
          select = false;
        } else {
          icdeh.checkSet.add(temp);
        }
      }
    }
    return select;
  }

  // creates the returned set and then calls other methods to do actual work
  private static SelectResults cutDownAndExpandIndexResults(SelectResults result,
      RuntimeIterator[] indexFieldToItrsMapping, List expansionList, List finalItrs,
      ExecutionContext context, List checkList, CompiledValue iterOps, IndexInfo theFilteringIndex)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {

    boolean useLinkedDataStructure = false;
    boolean nullValuesAtStart = true;
    Boolean orderByClause = (Boolean) context.cacheGet(CompiledValue.CAN_APPLY_ORDER_BY_AT_INDEX);
    if (orderByClause != null && orderByClause) {
      List orderByAttrs = (List) context.cacheGet(CompiledValue.ORDERBY_ATTRIB);
      useLinkedDataStructure = orderByAttrs.size() == 1;
      nullValuesAtStart = !((CompiledSortCriterion) orderByAttrs.get(0)).getCriterion();
    }
    SelectResults returnSet = null;
    if (finalItrs.size() == 1) {
      ObjectType resultType = ((RuntimeIterator) finalItrs.iterator().next()).getElementType();
      if (useLinkedDataStructure) {
        returnSet = context.isDistinct() ? new LinkedResultSet(resultType)
            : new SortedResultsBag(resultType, nullValuesAtStart);
      } else {
        returnSet = QueryUtils.createResultCollection(context, resultType);
      }

    } else {
      StructTypeImpl resultType = (StructTypeImpl) createStructTypeForRuntimeIterators(finalItrs);
      if (useLinkedDataStructure) {
        returnSet = context.isDistinct() ? new LinkedStructSet(resultType)
            : new SortedResultsBag<>(resultType, nullValuesAtStart);
      } else {
        returnSet = QueryUtils.createStructCollection(context, resultType);
      }

    }
    cutDownAndExpandIndexResults(returnSet, result, indexFieldToItrsMapping, expansionList,
        finalItrs, context, checkList, iterOps, theFilteringIndex);
    return returnSet;
  }

  // TODO: Explain the parameters passed
  private static void cutDownAndExpandIndexResults(SelectResults returnSet, SelectResults result,
      RuntimeIterator[] indexFieldToItrsMapping, List expansionList, List finalItrs,
      ExecutionContext context, List checkList, CompiledValue iterOps, IndexInfo theFilteringIndex)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {

    IndexCutDownExpansionHelper icdeh = new IndexCutDownExpansionHelper(checkList, context);
    int len = indexFieldToItrsMapping.length;
    // don't call instanceof ResultsBag here, since a StructBag is a subtype of ResultsBag
    if (result.getClass() == ResultsBag.class) {
      Support.Assert(len == 1, "The array size of iterators should be 1 here, but got " + len);
    }
    Iterator itr = result.iterator();
    ListIterator expansionListIterator = expansionList.listIterator();
    int limit = getLimitValue(context);

    while (itr.hasNext()) {
      DerivedInfo derivedInfo = null;
      if (IndexManager.JOIN_OPTIMIZATION) {
        derivedInfo = new DerivedInfo();
        derivedInfo.setExpansionList(expansionList);
      }
      Object value = itr.next();
      if (setIndexFieldValuesInRespectiveIterators(value, indexFieldToItrsMapping, icdeh)) {
        // does that mean we don't get dupes even if they exist in the index?
        // DO NESTED LOOPING
        if (IndexManager.JOIN_OPTIMIZATION) {
          derivedInfo.computeDerivedJoinResults(theFilteringIndex, context, iterOps);
        }
        doNestedIterationsForIndex(expansionListIterator.hasNext(), returnSet, finalItrs,
            expansionListIterator, context, iterOps, limit, derivedInfo.derivedResults);
        if (limit != -1 && returnSet.size() >= limit) {
          break;
        }
      }
    }
  }

  // returns the limit value from the context. This was set in CompiledSelect evaluate
  // We do not apply limit if we have an order by attribute at this time
  // it may be possible but we need better understanding of when ordering is taking place
  // If it's at the index level, we may be able to apply limits at this point
  // however a lot of the code in this class is fragile/unreadable/hard to maintain
  private static int getLimitValue(ExecutionContext context) {
    int limit = -1;
    if (context.cacheGet(CompiledValue.ORDERBY_ATTRIB) == null) {
      limit = context.cacheGet(CompiledValue.RESULT_LIMIT) != null
          ? (Integer) context.cacheGet(CompiledValue.RESULT_LIMIT) : -1;
    }
    return limit;
  }

  static CompiledID getCompiledIdFromPath(CompiledValue path) {
    int type = path.getType();
    if (type == OQLLexerTokenTypes.Identifier) {
      return (CompiledID) path;
    }
    return getCompiledIdFromPath(path.getReceiver());
  }

  private static void doNestedIterationsForIndex(boolean continueRecursion, SelectResults resultSet,
      List finalItrs, ListIterator expansionItrs, ExecutionContext context, CompiledValue iterOps,
      int limit, Map<String, SelectResults> derivedResults) throws FunctionDomainException,
      TypeMismatchException, NameResolutionException, QueryInvocationTargetException {

    if (!continueRecursion) {
      // end recusrion
      Iterator itr = finalItrs.iterator();
      int len = finalItrs.size();
      boolean select = true;
      if (iterOps != null) {
        select = applyCondition(iterOps, context);
      }
      if (len > 1) {
        boolean isOrdered = resultSet.getCollectionType().isOrdered();
        StructTypeImpl elementType =
            (StructTypeImpl) resultSet.getCollectionType().getElementType();

        // TODO: Optimize the LinkedStructSet implementation so that Object[] can be added rather
        // than Struct

        Object[] values = new Object[len];
        int j = 0;
        // creates tuple
        while (itr.hasNext()) {
          // Check if query execution on this thread is canceled.
          QueryMonitor.throwExceptionIfQueryOnCurrentThreadIsCanceled();

          values[j++] = ((RuntimeIterator) itr.next()).evaluate(context);
        }
        if (select) {
          if (isOrdered) {
            // ((LinkedStructSet) resultSet).add(new StructImpl(elementType, values));
            // Can be LinkedStructSet or SortedResultsBag ( containing underlying LinkedHashMap)
            resultSet.add(new StructImpl(elementType, values));
          } else {
            ((StructFields) resultSet).addFieldValues(values);
          }
        }
      } else {
        if (select) {
          resultSet.add(((RuntimeIterator) itr.next()).evaluate(context));
        }
      }
    } else {
      RuntimeIterator currentLevel = (RuntimeIterator) expansionItrs.next();
      // Calculate the key to find the derived join results. If we are a non nested lookup it will
      // be a Compiled Region otherwise it will be a CompiledPath that
      // we can extract the id from. In the end the result will be the alias which is used as a
      // prefix
      String key = null;
      boolean useDerivedResults = true;
      if (currentLevel.getCmpIteratorDefn().getCollectionExpr()
          .getType() == OQLLexerTokenTypes.RegionPath) {
        key = currentLevel.getCmpIteratorDefn().getName() + ':' + currentLevel.getDefinition();
      } else if (currentLevel.getCmpIteratorDefn().getCollectionExpr()
          .getType() == OQLLexerTokenTypes.LITERAL_select) {
        useDerivedResults = false;
      }
      SelectResults c;
      CompiledValue path = currentLevel.getCmpIteratorDefn().getCollectionExpr();
      if (useDerivedResults && derivedResults != null && path.hasIdentifierAtLeafNode()) {
        key = getCompiledIdFromPath(path).getId()
            + ':' + currentLevel.getDefinition();
        if (derivedResults.containsKey(key)) {
          c = derivedResults.get(key);
        } else {
          c = currentLevel.evaluateCollection(context);
        }
      } else if (useDerivedResults && derivedResults != null && key != null
