/*
 * 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.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.query.AmbiguousNameException;
import org.apache.geode.cache.query.FunctionDomainException;
import org.apache.geode.cache.query.Index;
import org.apache.geode.cache.query.NameNotFoundException;
import org.apache.geode.cache.query.NameResolutionException;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryInvalidException;
import org.apache.geode.cache.query.QueryInvocationTargetException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.RegionNotFoundException;
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.PartitionedIndex;
import org.apache.geode.cache.query.internal.types.ObjectTypeImpl;
import org.apache.geode.cache.query.internal.types.StructTypeImpl;
import org.apache.geode.cache.query.internal.types.TypeUtils;
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.cache.InternalCache;
import org.apache.geode.pdx.PdxInstance;
import org.apache.geode.pdx.internal.PdxString;

public class CompiledSelect extends AbstractCompiledValue {

  protected List<CompiledSortCriterion> orderByAttrs; // order by attributes: list of CompiledValue
  private final CompiledValue whereClause; // can be null if there isn't one
  private final List iterators; // fromClause: list of CompiledIteratorDefs
  protected List projAttrs; // projection attributes: list of Object[2]:
  // 0 is projection name, 1 is the CompiledValue for the expression
  private boolean distinct;
  private boolean count;
  // limits the SelectResults by the number specified.
  private final CompiledValue limit;
  // counts the no of results satisfying where condition for
  // count(*) non-distinct queries where no indexes are used.
  private int countStartQueryResult = 0;

  protected List<CompiledValue> groupBy = null;
  // Are not serialized and are recreated when compiling the query
  private final List<String> hints;
  protected boolean transformationDone = false;
  protected ObjectType cachedElementTypeForOrderBy = null;
  private boolean hasUnmappedOrderByCols = false;

  // used as a key in a context to identify the scope of this CompiledSelect
  private final Object scopeID = new Object();

  /*
   * Set in context for the where clause to signify that it has been evaluated at least one time for
   * any other CompiledValue that may use precalculated indexes we want to mark this as Evaluated so
   * that we don't unlock locks that don't belong to this iteration of evaluate. This is similar to
   * how CompiledComparisons store their IndexInfo in the context but for example a CompiledJunction
   * that uses 2 Comparisons would have unlocked the readlocks because we check to see if the clause
   * has a mapped value in the context. Because CompiledJunctions did not, we unlocked the read
   * locks. Now we set a value so that it will not do this. See where we use this value to see how
   * unlock is determined
   */
  private static final String CLAUSE_EVALUATED = "Evaluated";

  public CompiledSelect(boolean distinct, boolean count, CompiledValue whereClause, List iterators,
      List projAttrs, List<CompiledSortCriterion> orderByAttrs, CompiledValue limit,
      List<String> hints, List<CompiledValue> groupByClause) {
    this.orderByAttrs = orderByAttrs;
    this.whereClause = whereClause;
    this.iterators = iterators;
    this.projAttrs = projAttrs;
    this.distinct = distinct;
    this.count = count;
    this.limit = limit;
    this.hints = hints;
    groupBy = groupByClause;

  }

  @Override
  public List getChildren() {
    List list = new ArrayList();
    if (whereClause != null) {
      list.add(whereClause);
    }

    list.addAll(iterators);

    if (projAttrs != null) {
      // extract the CompiledValues out of the projAttrs (each of which are Object[2])
      for (final Object projAttr : projAttrs) {
        list.add(((Object[]) projAttr)[1]);
      }
    }

    if (orderByAttrs != null) {
      list.addAll(orderByAttrs);
    }

    return list;
  }

  public boolean isDistinct() {
    return distinct;
  }

  public boolean isGroupBy() {
    return groupBy != null;
  }

  public boolean isOrderBy() {
    return orderByAttrs != null;
  }

  public void setDistinct(boolean distinct) {
    this.distinct = distinct;
  }

  public boolean isCount() {
    return count;
  }

  public void setCount(boolean count) {
    this.count = count;
  }

  @Override
  public int getType() {
    return LITERAL_select;
  }

  public CompiledValue getWhereClause() {
    return whereClause;
  }

  public List getIterators() {
    return iterators;
  }

  public List getProjectionAttributes() {
    return projAttrs;
  }

  public List<CompiledSortCriterion> getOrderByAttrs() {
    return orderByAttrs;
  }

  @Override
  public Set computeDependencies(ExecutionContext context)
      throws TypeMismatchException, NameResolutionException {
    // bind iterators in new scope in order to determine dependencies
    context.cachePut(scopeID, context.associateScopeID());
    context.newScope((Integer) context.cacheGet(scopeID));
    context.pushExecCache((Integer) context.cacheGet(scopeID));
    try {
      for (final Object iterator : iterators) {

        CompiledIteratorDef iterDef = (CompiledIteratorDef) iterator;
        // compute dependencies on this iter first before adding its
        // RuntimeIterator to the current scope.
        // this makes sure it doesn't bind attributes to itself
        context.addDependencies(this, iterDef.computeDependencies(context));
        RuntimeIterator rIter = iterDef.getRuntimeIterator(context);
        context.addToIndependentRuntimeItrMap(iterDef);
        context.bindIterator(rIter);
      }

      // is the where clause dependent on itr?
      if (whereClause != null) {
        context.addDependencies(this, whereClause.computeDependencies(context));
      }
      // are any projections dependent on itr?
      if (projAttrs != null) {
        Set totalDependencySet = null;
        for (Iterator iter = projAttrs.iterator(); iter.hasNext();) {
          // unwrap the projection expressions, they are in 2-element Object[]
          // with first element the fieldName and second the projection
          // expression
          Object[] prj = (Object[]) TypeUtils.checkCast(iter.next(), Object[].class);
          CompiledValue prjExpr = (CompiledValue) TypeUtils.checkCast(prj[1], CompiledValue.class);
          totalDependencySet = context.addDependencies(this, prjExpr.computeDependencies(context));
        }
        doTreeTransformation(context);

        return totalDependencySet;
      } else {
        doTreeTransformation(context);
        return context.getDependencySet(this, true);
      }
      // is the where clause dependent on itr?
      /*
       * if (this.whereClause != null) { return context.addDependencies(this,
       * this.whereClause.computeDependencies(context)); } else { return
       * context.getDependencySet(this, true); }
       */
    } finally {
      context.popExecCache();
      context.popScope();
    }
  }

  protected void doTreeTransformation(ExecutionContext context)
      throws TypeMismatchException, NameResolutionException {
    if (!transformationDone) {
      cachedElementTypeForOrderBy = prepareResultType(context);
      mapOrderByColumns(context);
      transformGroupByIfPossible(context);
    }
    transformationDone = true;
  }

  /**
   * Transforms the group by clause into distinct order by clause, if possible
   */
  private void transformGroupByIfPossible(ExecutionContext context)
      throws TypeMismatchException, NameResolutionException {
    // for time being assume that the group by cols are explicitly mentioned in proj
    if (groupBy != null) {
      List projAttribs = projAttrs;
      if (projAttribs == null) {
        projAttribs = new ArrayList();
        List currentIters = context.getCurrentIterators();
        for (Object o : currentIters) {
          RuntimeIterator rIter = (RuntimeIterator) o;
          String name = rIter.getName();
          projAttribs.add(new Object[] {name, rIter});
        }
      }

      if (projAttribs != null && projAttribs.size() != groupBy.size()) {
        throw new QueryInvalidException(
            "Query contains projected column not present in group by clause or "
                + "Query contains group by columns not present in projected fields");
      }

      boolean shouldTransform = true;
      StringBuilder lhsBuffer = new StringBuilder();
      StringBuilder rhsBuffer = new StringBuilder();

      outer: for (int i = 0; i < projAttribs.size(); ++i) {
        Object[] prj = (Object[]) TypeUtils.checkCast(projAttribs.get(i), Object[].class);
        CompiledValue groupByAttr = groupBy.get(i);
        if (prj[0] != null) {
          if (groupByAttr instanceof CompiledID) {
            if (prj[0].equals(((CompiledID) groupByAttr).getId())) {
              lhsBuffer.delete(0, lhsBuffer.length());
              rhsBuffer.delete(0, rhsBuffer.length());
              continue;
            }
          }
        }
        CompiledValue cvProj = (CompiledValue) TypeUtils.checkCast(prj[1], CompiledValue.class);
        cvProj.generateCanonicalizedExpression(lhsBuffer, context);
        groupByAttr.generateCanonicalizedExpression(rhsBuffer, context);
        if (lhsBuffer.length() == rhsBuffer.length()) {
          for (int indx = 0; indx < lhsBuffer.length(); ++indx) {
            if (lhsBuffer.charAt(indx) != rhsBuffer.charAt(indx)) {
              shouldTransform = false;
              break outer;
            }
          }
        } else {
          shouldTransform = false;
          break;
        }

        lhsBuffer.delete(0, lhsBuffer.length());
        rhsBuffer.delete(0, rhsBuffer.length());

      }
      // check if the order by clause is null or order by clause is same as proj.
      // for now check if order by is null
      if (shouldTransform && orderByAttrs == null) {
        modifyGroupByToOrderBy(true, context);
      } else {
        throw new QueryInvalidException(
            "Query contains projected column not present in group by clause or "
                + "Query contains group by columns not present in projected fields");
      }
    }
  }

  protected void modifyGroupByToOrderBy(boolean setDistinct, ExecutionContext context)
      throws TypeMismatchException, NameResolutionException {
    if (setDistinct) {
      distinct = setDistinct;
    }
    orderByAttrs = new ArrayList<>(groupBy.size());
    int colIndex = 0;
    for (CompiledValue cv : groupBy) {
      CompiledSortCriterion csc = new CompiledSortCriterion(false, cv);
      csc.mapExpressionToProjectionField(projAttrs, context);
      orderByAttrs.add(csc);
    }
    groupBy = null;
  }

  private void mapOrderByColumns(ExecutionContext context)
      throws TypeMismatchException, NameResolutionException {
    if (orderByAttrs != null) {
      for (final CompiledSortCriterion csc : orderByAttrs) {
        // Ideally for replicated regions, the requirement that
        // projected columns should
        // contain order by fields ( directly or derivable on it),
        // is not needed. But for PR , the query gathers only projected
        // columns, so applying order by on the query node
        // will need order by values ( which we dont send). So this
        // restriction is needed.
        // Also if this restriction is assumed to be correct, then the order
        // by comparator can be optimized as
        // it does not need to keep the mapping of evaluated order by clause,
        // for comparison
        if (!csc.mapExpressionToProjectionField(projAttrs, context)) {
          hasUnmappedOrderByCols = true;
        }
      }
    }
  }

  private void evalCanonicalizedExpressionForCSC(CompiledSortCriterion csc,
      ExecutionContext context, StringBuilder buffer)
      throws TypeMismatchException, NameResolutionException {
    csc.getExpr().generateCanonicalizedExpression(buffer, context);
  }

  /*
   * Gets the appropriate empty results set when outside of actual query evalutaion.
   *
   * @param parameters the parameters that will be passed into the query when evaluated
   *
   * @param cache the cache the query will be executed in the context of
   *
   * @return the empty result set of the appropriate type
   */
  public SelectResults getEmptyResultSet(Object[] parameters, InternalCache cache, Query query)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {
    ExecutionContext context = new QueryExecutionContext(parameters, cache, query);
    computeDependencies(context);
    context.newScope((Integer) context.cacheGet(scopeID));
    context.pushExecCache((Integer) context.cacheGet(scopeID));
    SelectResults results = null;
    try {
      for (final Object iterator : iterators) {
        CompiledIteratorDef iterDef = (CompiledIteratorDef) iterator;
        RuntimeIterator rIter = iterDef.getRuntimeIterator(context);
        context.bindIterator(rIter);
      }
      results = prepareEmptyResultSet(context, false);
    } finally {
      context.popScope();
      context.popExecCache();
    }
    return results;
  }

  public ObjectType getElementTypeForOrderByQueries() {
    return cachedElementTypeForOrderBy;
  }

  @Override
  public SelectResults evaluate(ExecutionContext context) throws FunctionDomainException,
      TypeMismatchException, NameResolutionException, QueryInvocationTargetException {
    context.newScope((Integer) context.cacheGet(scopeID));
    context.pushExecCache((Integer) context.cacheGet(scopeID));
    boolean prevDistinctState = context.isDistinct();
    context.setDistinct(distinct);
    if (hasUnmappedOrderByCols && context.getBucketList() != null) {
      throw new QueryInvalidException(
          "Query contains atleast one order by field which is not present in projected fields.");
    }
    if (hints != null) {
      context.cachePut(QUERY_INDEX_HINTS, hints);
    }

    try {
      // set flag to keep objects serialized for "select *" queries
      if (context.getQuery() != null) {
        ((DefaultQuery) context.getQuery()).keepResultsSerialized(this, context);
      }
      for (final Object iterator : iterators) {
        CompiledIteratorDef iterDef = (CompiledIteratorDef) iterator;
        RuntimeIterator rIter = iterDef.getRuntimeIterator(context);
        context.bindIterator(rIter);
        // Ideally the function below should always be called after binding has occurred
        // So that the internal ID gets set during binding to the scope. If not so then chances
        // are that internal_id is still null causing index_internal_id to be null.
        // Though in our case it may not be an issue as the compute dependency phase must have
        // already set the index id
      }
      Integer limitValue = evaluateLimitValue(context, limit);
      SelectResults result = null;
      boolean evalAsFilters = false;
      if (whereClause == null) {
        result = doIterationEvaluate(context, false);
      } else {
        if (!whereClause.isDependentOnCurrentScope(context)) { // independent
                                                               // where
                                                               // @todo
                                                               // check
                                                               // for
                                                               // dependency
                                                               // on
                                                               // current
                                                               // scope
                                                               // only?
          // clause
          Object b = whereClause.evaluate(context);
          if (b == null || b == QueryService.UNDEFINED) {
            // treat as if all elements are undefined
            result = prepareEmptyResultSet(context, false);
            // ResultsSet.emptyResultsSet(resultSet, 0);
            // return result;
          } else if (!(b instanceof Boolean)) {
            throw new TypeMismatchException(
                String.format("The WHERE clause was type ' %s ' instead of boolean",
                    b.getClass().getName()));
          } else if ((Boolean) b) {
            result = doIterationEvaluate(context, false);
          } else {
            result = prepareEmptyResultSet(context, false);
            // ResultsSet.emptyResultsSet(resultSet, 0);
            // return result;
          }
        } else {
          // Check the numer of independent iterators
          int numInd = context.getAllIndependentIteratorsOfCurrentScope().size();
          // If order by clause is defined, then the first column should be the preferred index
          if (orderByAttrs != null && numInd == 1) {
            CompiledSortCriterion csc = orderByAttrs.get(0);
            StringBuilder preferredIndexCondn = new StringBuilder();
            evalCanonicalizedExpressionForCSC(csc, context, preferredIndexCondn);
            context.cachePut(PREF_INDEX_COND, preferredIndexCondn.toString());
          }
          boolean unlock = true;
          Object obj = context.cacheGet(whereClause);
          if (obj != null && (obj instanceof IndexInfo[] || obj.equals(CLAUSE_EVALUATED))) {
            // if indexinfo is cached means the read lock
            // is not being taken this time, so releasing
            // the lock is not required
            unlock = false;
          }
          // see if we should evaluate as filters,
          // and count how many actual index lookups will be performed
          PlanInfo planInfo = whereClause.getPlanInfo(context);
          if (context.cacheGet(whereClause) == null) {
            context.cachePut(whereClause, CLAUSE_EVALUATED);
          }
          try {
            evalAsFilters = planInfo.evalAsFilter;
            // let context know if there is exactly one index lookup
            context.setOneIndexLookup(planInfo.indexes.size() == 1);
            if (evalAsFilters) {
              ((QueryExecutionContext) context).setIndexUsed(true);
              // Ignore order by attribs for a while

              boolean canApplyOrderByAtIndex = false;
              if (limitValue >= 0 && numInd == 1
                  && ((Filter) whereClause).isLimitApplicableAtIndexLevel(context)) {
                context.cachePut(CAN_APPLY_LIMIT_AT_INDEX, Boolean.TRUE);
              }
              StringBuilder temp = null;
              if (orderByAttrs != null) {
                temp = new StringBuilder();
                CompiledSortCriterion csc = orderByAttrs.get(0);
                evalCanonicalizedExpressionForCSC(csc, context, temp);
              }

              boolean needsTopLevelOrdering = true;
              if (temp != null && numInd == 1 && ((Filter) whereClause)
                  .isOrderByApplicableAtIndexLevel(context, temp.toString())) {
                context.cachePut(CAN_APPLY_ORDER_BY_AT_INDEX, Boolean.TRUE);
                context.cachePut(ORDERBY_ATTRIB, orderByAttrs);
                canApplyOrderByAtIndex = true;
                if (orderByAttrs.size() == 1) {
                  needsTopLevelOrdering = false;
                  // If there is a limit present and we are executing on a partitioned region
                  // we should use a sorted set
                  if (limit != null) {
                    // Currently check bucket list to determine if it's a pr query
                    if (context.getBucketList() != null && context.getBucketList().size() > 0) {
                      needsTopLevelOrdering = true;
                    }
                  }
                }
              } else if (temp != null) {
                // If order by is present but cannot be applied at index level,
                // then limit also cannot be applied
                // at index level
                context.cachePut(CAN_APPLY_LIMIT_AT_INDEX, Boolean.FALSE);
              }

              context.cachePut(RESULT_LIMIT, limitValue);
              if (numInd == 1
                  && ((Filter) whereClause).isProjectionEvaluationAPossibility(context)
                  && (orderByAttrs == null
                      || (canApplyOrderByAtIndex && !needsTopLevelOrdering))
                  && projAttrs != null) {
                // Possibility of evaluating the resultset as filter itself
                ObjectType resultType = cachedElementTypeForOrderBy != null
                    ? cachedElementTypeForOrderBy : prepareResultType(context);
                context.cachePut(RESULT_TYPE, resultType);
                context.cachePut(PROJ_ATTRIB, projAttrs);
              }


              result = ((Filter) whereClause).filterEvaluate(context, null);
              if (!(context.cacheGet(RESULT_TYPE) instanceof Boolean)) {
                QueryObserverHolder.getInstance()
                    .beforeApplyingProjectionOnFilterEvaluatedResults(result);
                result = applyProjectionOnCollection(result, context, !needsTopLevelOrdering);
              }
            } else {
              // otherwise iterate over the single from var to evaluate
              result = doIterationEvaluate(context, true);
            }
          } finally {
            // The Read lock is acquired in {@link
            // IndexManager#getBestMatchIndex()},
            // because we need to select index which can be read-locked.
            if (unlock) {
              releaseReadLockOnUsedIndex(planInfo);
            }
          }
        }
      }
      // TODO: It does not appear that results would be null ever.
      // if (result == null) { return QueryService.UNDEFINED; }
      assert result != null;
      // drop duplicates if this is DISTINCT
      if (result instanceof SelectResults) {
        SelectResults sr = result;
        CollectionType colnType = sr.getCollectionType();
        // if (this.distinct && colnType.allowsDuplicates()) {
        if (distinct) {
          Collection r;
          // Set s = sr.asSet();
          if (colnType.allowsDuplicates()) {
            // don't just convert to a ResultsSet (or StructSet), since
            // the bags can convert themselves to a Set more efficiently
            r = sr.asSet();
          } else {
            r = sr;
          }

          result = new ResultsCollectionWrapper(colnType.getElementType(), r, limitValue);
          if (r instanceof Bag.SetView) {
            ((ResultsCollectionWrapper) result).setModifiable(false);
          }
        } else {
          // SelectResults is of type
          if (limitValue > -1) {
            ((Bag) sr).applyLimit(limitValue);
          }
        }

        /*
         * We still have to get size of SelectResults in some cases like, if index was used OR query
         * is a distinct query.
         *
         * If SelectResult size is zero then we need to put Integer for 0 count.
         */
        if (count) {
          SelectResults res = result;

          if ((distinct || evalAsFilters || countStartQueryResult == 0)) {
            // Retrun results as it is as distinct is applied
            // at coordinator node for PR queries.
            if (context.getBucketList() != null && distinct) {
              return result;
            }
            // Take size and empty the results
            int resultCount = res.size();
            res.clear();

            ResultsBag countResult =
                new ResultsBag(new ObjectTypeImpl(Integer.class), context.getCachePerfStats());
            countResult.addAndGetOccurence(resultCount);
            result = countResult;

          } else {
            ((Bag) res).addAndGetOccurence(countStartQueryResult);
          }
        }
      }
      return result;
    } finally {
      context.setDistinct(prevDistinctState);
      context.popScope();
      context.popExecCache();
    }
  }

  /**
   * The index is locked during query to prevent it from being removed by another thread. So we have
   * to release the lock only after whole query is finished as one query can use an index multiple
   * times.
   */
  private void releaseReadLockOnUsedIndex(PlanInfo planInfo) {
    List inds = planInfo.indexes;
    for (Object obj : inds) {
      Index index = (Index) obj;
      Index prIndex = ((AbstractIndex) index).getPRIndex();
      if (prIndex != null) {
        ((PartitionedIndex) prIndex).releaseIndexReadLockForRemove();
      } else {
        ((AbstractIndex) index).releaseIndexReadLockForRemove();
      }
    }
  }

  /**
   * Returns the size of region iterator for count(*) on a region without whereclause.
   *
   * @since GemFire 6.6.2
   */
  private int getRegionIteratorSize(ExecutionContext context, CompiledValue collExpr)
      throws RegionNotFoundException {
    Region region;
    String regionPath = ((CompiledRegion) collExpr).getRegionPath();
    if (context.getBucketRegion() == null) {
      region = context.getCache().getRegion(regionPath);
    } else {
      region = context.getBucketRegion();
    }
    if (region != null) {
      return region.size();
    } else {
      // if we couldn't find the region because the cache is closed, throw
      // a CacheClosedException
      Cache cache = context.getCache();
      if (cache.isClosed()) {
        throw new CacheClosedException();
      }
      throw new RegionNotFoundException(
          String.format("Region not found: %s", regionPath));
    }
  }

  public int getLimitValue(Object[] bindArguments) throws FunctionDomainException,
      TypeMismatchException, NameResolutionException, QueryInvocationTargetException {
    // if evaluation of the limit fails, we default to no limit
    return evaluateLimitValue(bindArguments);
  }

  // returns null if result is UNDEFINED
  private SelectResults doIterationEvaluate(ExecutionContext context, boolean evaluateWhereClause)
      throws TypeMismatchException, FunctionDomainException, NameResolutionException,
      QueryInvocationTargetException {

    SelectResults results = prepareEmptyResultSet(context, false);

    // TODO: SELF : Work on limit implementation on bulk get
    // check for bulk get optimization
    if (evaluateWhereClause) {
      List tmpResults = optimizeBulkGet(context);
      if (tmpResults != null) {
        // (has only one iterator)
        RuntimeIterator rIter = (RuntimeIterator) context.getCurrentIterators().get(0);
        for (Object currObj : tmpResults) {
          rIter.setCurrent(currObj);
          QueryObserver observer = QueryObserverHolder.getInstance();
          observer.beforeIterationEvaluation(rIter, currObj);
          applyProjectionAndAddToResultSet(context, results, orderByAttrs == null);
        }
        return results;
      }
    }
    int numElementsInResult = 0;
    try {
      doNestedIterations(0, results, context, evaluateWhereClause, numElementsInResult);
    } catch (CompiledSelect.NullIteratorException ignore) {
      return null;
    }
    return results;
  }

  // TODO: make this more general to work for any kind of map, not just regions
  /**
   * Check for the bulk-get pattern and if it applies do an optimized execution. The pattern is:
   * SELECT ?? FROM <Region>.entrySet e WHERE e.key IN <Collection>.
   *
   * @return a List of entries if optimization was executed, or null if it wasn't because the
   *         optimization pattern didn't match
   */
  private List optimizeBulkGet(ExecutionContext context) throws TypeMismatchException,
      FunctionDomainException, NameResolutionException, QueryInvocationTargetException {
    List iterList = context.getCurrentIterators();
    // must be one iterator
    if (iterList.size() != 1) {
      return null;
    }

    // where clause must be an IN operation
    if (!(whereClause instanceof CompiledIn)) {
      return null;
    }

    RuntimeIterator rIter = (RuntimeIterator) iterList.get(0);
    CompiledIteratorDef cIterDef = rIter.getCmpIteratorDefn();
    CompiledValue colnExpr = cIterDef.getCollectionExpr();

    // check for region.entrySet or region.entrySet()
    boolean match = false;
    CompiledRegion rgn = null;
    if (colnExpr instanceof CompiledPath) {
      CompiledPath cPath = (CompiledPath) colnExpr;
      CompiledValue rcvr = cPath.getReceiver();
      if (rcvr instanceof CompiledRegion) {
        rgn = (CompiledRegion) rcvr;
        String attr = cPath.getTailID();
        match = attr.equals("entrySet");
      }
    }
    if (!match && (colnExpr instanceof CompiledOperation)) {
      CompiledOperation cOp = (CompiledOperation) colnExpr;
      CompiledValue rcvr = cOp.getReceiver(context);
      if (rcvr instanceof CompiledRegion) {
        rgn = (CompiledRegion) rcvr;
        match = cOp.getMethodName().equals("entrySet");
      }
    }
    if (!match) {
      return null;
    }

    // check for IN expression
    CompiledIn cIn = (CompiledIn) whereClause;
    // defer to the CompiledIn for rest of pattern match and
    // evaluation
    return cIn.optimizeBulkGet(rgn, context);
  }

  // returns the number of elements added in the return ResultSet
  private int doNestedIterations(int level, SelectResults results, ExecutionContext context,
      boolean evaluateWhereClause, int numElementsInResult)
      throws TypeMismatchException, FunctionDomainException, NameResolutionException,
      QueryInvocationTargetException, CompiledSelect.NullIteratorException {
    List iterList = context.getCurrentIterators();
    if (level == iterList.size()) {
      boolean addToResults = true;
      if (evaluateWhereClause) {
        Object result = whereClause.evaluate(context);
        QueryObserver observer = QueryObserverHolder.getInstance();
        observer.afterIterationEvaluation(result);
        if (result == null) {
          addToResults = false;
        } else if (result instanceof Boolean) {
          addToResults = (Boolean) result;
        } else if (result == QueryService.UNDEFINED) {
          // add UNDEFINED to results only for NOT EQUALS queries
          if (whereClause.getType() == COMPARISON) {
            int operator = ((Filter) whereClause).getOperator();
            if ((operator != TOK_NE && operator != TOK_NE_ALT)) {
              addToResults = false;
            }
          } else {
            addToResults = false;
          }
        } else {
          throw new TypeMismatchException(
              String.format("The WHERE clause was type ' %s ' instead of boolean",
                  result.getClass().getName()));
        }
      }
      if (addToResults) {
        int occurrence =
            applyProjectionAndAddToResultSet(context, results, orderByAttrs == null);
        // If the occurrence is greater than 1, then only in case of
        // non distinct query should it be treated as contributing to size
        // else duplication will be eliminated when making it distinct using
        // ResultsCollectionWrapper and we will fall short of limit
        if (occurrence == 1 || (occurrence > 1 && !distinct)) {
