/*
 * 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.index;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.query.FunctionDomainException;
import org.apache.geode.cache.query.IndexStatistics;
import org.apache.geode.cache.query.IndexType;
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.TypeMismatchException;
import org.apache.geode.cache.query.internal.CompiledBindArgument;
import org.apache.geode.cache.query.internal.CompiledIteratorDef;
import org.apache.geode.cache.query.internal.CompiledLiteral;
import org.apache.geode.cache.query.internal.CompiledPath;
import org.apache.geode.cache.query.internal.CompiledSortCriterion;
import org.apache.geode.cache.query.internal.CompiledValue;
import org.apache.geode.cache.query.internal.CqEntry;
import org.apache.geode.cache.query.internal.ExecutionContext;
import org.apache.geode.cache.query.internal.IndexInfo;
import org.apache.geode.cache.query.internal.QRegion;
import org.apache.geode.cache.query.internal.QueryMonitor;
import org.apache.geode.cache.query.internal.QueryObserver;
import org.apache.geode.cache.query.internal.QueryObserverHolder;
import org.apache.geode.cache.query.internal.QueryUtils;
import org.apache.geode.cache.query.internal.RuntimeIterator;
import org.apache.geode.cache.query.internal.StructImpl;
import org.apache.geode.cache.query.internal.Support;
import org.apache.geode.cache.query.internal.index.IndexManager.TestHook;
import org.apache.geode.cache.query.internal.index.IndexStore.IndexStoreEntry;
import org.apache.geode.cache.query.internal.parse.OQLLexerTokenTypes;
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.ObjectType;
import org.apache.geode.cache.query.types.StructType;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.NonTXEntry;
import org.apache.geode.internal.cache.RegionEntry;
import org.apache.geode.internal.cache.RegionEntryContext;
import org.apache.geode.internal.cache.entries.VMThinRegionEntryHeap;
import org.apache.geode.internal.cache.persistence.query.CloseableIterator;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.internal.PdxString;

/**
 * A CompactRangeIndex is a range index that has simple data structures to minimize its footprint,
 * at the expense of doing extra work at index maintenance. It is selected as the index
 * implementation when the indexed expression is a path expression and the from clause has only one
 * iterator. This implies there is only one value in the index for each region entry.
 *
 * This index does not support the storage of projection attributes.
 *
 * Currently this implementation only supports an index on a region path.
 *
 * @since GemFire 6.0
 */
public class CompactRangeIndex extends AbstractIndex {
  private static final Logger logger = LogService.getLogger();

  @MutableForTesting
  private static TestHook testHook;

  protected ThreadLocal<OldKeyValuePair> oldKeyValue;

  private final IndexStore indexStore;

  @MutableForTesting
  static boolean TEST_ALWAYS_UPDATE_IN_PROGRESS = false;

  public CompactRangeIndex(InternalCache cache, String indexName, Region region, String fromClause,
      String indexedExpression, String projectionAttributes, String origFromClause,
      String origIndexExpr, String[] definitions, IndexStatistics stats) {
    super(cache, indexName, region, fromClause, indexedExpression, projectionAttributes,
        origFromClause, origIndexExpr, definitions, stats);
    if (IndexManager.IS_TEST_LDM) {
      indexStore = new MapIndexStore(
          ((LocalRegion) region).getIndexMap(indexName, indexedExpression, origFromClause), region);
    } else {
      indexStore =
          new MemoryIndexStore(region, internalIndexStats, (InternalCache) region.getCache());
    }
  }

  public IndexStore getIndexStorage() {
    return indexStore;
  }

  /**
   * Get the index type
   *
   * @return the type of index
   */
  @Override
  public IndexType getType() {
    return IndexType.FUNCTIONAL;
  }

  @Override
  protected boolean isCompactRangeIndex() {
    return true;
  }

  @Override
  public void initializeIndex(boolean loadEntries) throws IMQException {
    long startTime = System.nanoTime();
    evaluator.initializeIndex(loadEntries);
    internalIndexStats.incNumUpdates(((IMQEvaluator) evaluator).getTotalEntriesUpdated());
    long endTime = System.nanoTime();
    internalIndexStats.incUpdateTime(endTime - startTime);
  }

  @Override
  void addMapping(RegionEntry entry) throws IMQException {
    evaluator.evaluate(entry, true);
    internalIndexStats.incNumUpdates();
  }

  /**
   * @param opCode one of OTHER_OP, BEFORE_UPDATE_OP, AFTER_UPDATE_OP.
   */
  @Override
  void removeMapping(RegionEntry entry, int opCode) throws IMQException {
    if (opCode == BEFORE_UPDATE_OP) {
      // Either take key from reverse map OR evaluate it using IMQEvaluator.
      if (!IndexManager.isObjectModificationInplace()) {
        // It will always contain 1 element only, for this thread.
        if (oldKeyValue == null) {
          oldKeyValue = new ThreadLocal<>();
        }
        oldKeyValue.set(new OldKeyValuePair());
        evaluator.evaluate(entry, false);
      }
    } else if (opCode == REMOVE_DUE_TO_GII_TOMBSTONE_CLEANUP) {
      // we know in this specific case, that a before op was called and stored oldKey/value
      // we also know that a regular remove won't work due to the entry no longer being present
      // We know the old key so let's just remove mapping from the old key
      if (oldKeyValue != null) {
        if (oldKeyValue.get() == null) {
          return;
        }
        indexStore.removeMappingGII(oldKeyValue.get().getOldKey(), entry);
      } else {
        // rely on reverse map in the index store to figure out the real key
        indexStore.removeMappingGII(IndexManager.NULL, entry);
      }
    } else if (opCode == CLEAN_UP_THREAD_LOCALS) {
      if (oldKeyValue != null) {
        oldKeyValue.remove();
      }
    } else {
      // Need to reset the thread-local map as many puts and destroys might
      // happen in same thread.
      if (oldKeyValue != null) {
        oldKeyValue.remove();
      }
      evaluator.evaluate(entry, false);
      internalIndexStats.incNumUpdates();
    }
  }

  void removeMapping(Object key, RegionEntry entry) throws IMQException {
    indexStore.removeMapping(key, entry);
  }

  @Override
  public boolean clear() {
    return indexStore.clear();
  }


  @Override
  public List queryEquijoinCondition(IndexProtocol indx, ExecutionContext context)
      throws TypeMismatchException, FunctionDomainException, NameResolutionException,
      QueryInvocationTargetException {
    // get a read lock when doing a lookup
    long start = updateIndexUseStats();
    ((AbstractIndex) indx).updateIndexUseStats();
    List data = new ArrayList();
    Iterator<IndexStoreEntry> outer = null;
    Iterator inner = null;
    try {
      // We will iterate over each of the index Map to obtain the keys
      outer = ((MemoryIndexStore) indexStore).getKeysIterator();

      if (indx instanceof CompactRangeIndex) {
        IndexStore indexStore = ((CompactRangeIndex) indx).getIndexStorage();
        inner = ((MemoryIndexStore) indexStore).getKeysIterator();

      } else {
        inner = ((RangeIndex) indx).getValueToEntriesMap().entrySet().iterator();
      }
      IndexStoreEntry outerEntry = null;
      Object innerEntry = null;
      Object outerKey = null;
      Object innerKey = null;
      boolean incrementInner = true;
      outer: while (outer.hasNext()) {
        outerEntry = outer.next();
        outerKey = outerEntry.getDeserializedKey();
        // TODO: eliminate all labels
        inner: while (!incrementInner || inner.hasNext()) {
          if (incrementInner) {
            innerEntry = inner.next();
            if (innerEntry instanceof IndexStoreEntry) {
              innerKey = ((IndexStoreEntry) innerEntry).getDeserializedKey();
            } else {
              innerKey = ((Map.Entry) innerEntry).getKey();
            }
          }
          int compare = ((Comparable) outerKey).compareTo(innerKey);
          if (compare == 0) {
            Object innerValue = null;
            CloseableIterator<IndexStoreEntry> iter = null;
            try {
              if (innerEntry instanceof IndexStoreEntry) {
                innerValue = ((CompactRangeIndex) indx).getIndexStorage().get(outerKey);
              } else {
                innerValue = ((Map.Entry) innerEntry).getValue();
              }
              iter = indexStore.get(outerKey);
              populateListForEquiJoin(data, iter, innerValue, context, innerKey);
            } finally {
              if (iter != null) {
                iter.close();
              }
              if (innerValue != null && innerValue instanceof CloseableIterator) {
                ((CloseableIterator<IndexStoreEntry>) innerValue).close();
              }
            }

            incrementInner = true;
            continue outer;
          } else if (compare < 0) {
            // The outer key is smaller than the inner key. That means
            // that we need
            // to increment the outer loop without moving inner loop.
            // incrementOuter = true;
            incrementInner = false;
            continue outer;
          } else {
            // The outer key is greater than inner key , so increment the
            // inner loop without changing outer
            incrementInner = true;
          }
        }
        break;
      }
      return data;
    } finally {
      ((AbstractIndex) indx).updateIndexUseEndStats(start);
      updateIndexUseEndStats(start);
      if (inner != null && indx instanceof CompactRangeIndex
          && inner instanceof CloseableIterator) {
        ((CloseableIterator<IndexStoreEntry>) inner).close();
      }
    }
  }

  /**
   * This evaluates the left and right side of a EQUI-JOIN where condition for which this Index was
   * used. Like, if condition is "p.ID = e.ID", {@link IndexInfo} will contain Left as p.ID, Right
   * as e.ID and operator as TOK_EQ. This method will evaluate p.ID OR e.ID based on if it is inner
   * or outer RegionEntry, and verify the p.ID = e.ID.
   *
   * This method is called only for Memory indexstore
   *
   * @return true if entry value and index value are consistent.
   */
  protected boolean verifyInnerAndOuterEntryValues(IndexStoreEntry entry, ExecutionContext context,
      IndexInfo indexInfo, Object keyVal) throws FunctionDomainException, TypeMismatchException,
      NameResolutionException, QueryInvocationTargetException {
    // Verify index key in value only for memory index store
    CompactRangeIndex index = (CompactRangeIndex) indexInfo._getIndex();
    RuntimeIterator runtimeItr = index.getRuntimeIteratorForThisIndex(context, indexInfo);
    if (runtimeItr != null) {
      runtimeItr.setCurrent(entry.getDeserializedValue());
    }
    return evaluateEntry(indexInfo, context, keyVal);
  }

  @Override
  public int getSizeEstimate(Object key, int operator, int matchLevel)
      throws TypeMismatchException {
    // Get approx size;
    int size = 0;
    if (key == null) {
      key = IndexManager.NULL;
    }
    long start = updateIndexUseStats(false);
    try {
      switch (operator) {
        case OQLLexerTokenTypes.TOK_EQ: {
          key = TypeUtils.indexKeyFor(key);
          key = getPdxStringForIndexedPdxKeys(key);
          size = indexStore.size(key);
          break;
        }
        case OQLLexerTokenTypes.TOK_NE_ALT:
        case OQLLexerTokenTypes.TOK_NE:
          size = region.size();
          key = TypeUtils.indexKeyFor(key);
          key = getPdxStringForIndexedPdxKeys(key);
          size -= indexStore.size(key);
          break;
        case OQLLexerTokenTypes.TOK_LE:
        case OQLLexerTokenTypes.TOK_LT:
          if (matchLevel <= 0 && (key instanceof Number)) {

            int totalSize = indexStore.size();
            if (CompactRangeIndex.testHook != null) {
              CompactRangeIndex.testHook.hook(1);
            }
            if (totalSize > 1) {
              Number keyAsNum = (Number) key;
              int x = 0;
              IndexStoreEntry firstEntry = null;
              CloseableIterator<IndexStoreEntry> iter1 = null;
              CloseableIterator<IndexStoreEntry> iter2 = null;

              try {
                iter1 = indexStore.iterator(null);
                if (iter1.hasNext()) {
                  firstEntry = iter1.next();

                  IndexStoreEntry lastEntry = null;

                  iter2 = indexStore.descendingIterator(null);
                  if (iter2.hasNext()) {
                    lastEntry = iter2.next();
                  }

                  if (firstEntry != null && lastEntry != null) {
                    Number first = (Number) firstEntry.getDeserializedKey();
                    Number last = (Number) lastEntry.getDeserializedKey();
                    if (first.doubleValue() != last.doubleValue()) {
                      // Shobhit: Now without ReadLoack on index we can end up with 0
                      // in denominator if the numbers are floating-point and
                      // truncated with conversion to long, and the first and last
                      // truncate to the same long, so safest calculation is to
                      // convert to doubles.
                      x = (int) (((keyAsNum.doubleValue() - first.doubleValue()) * totalSize)
                          / (last.doubleValue() - first.doubleValue()));
                    }
                  }

                  if (x < 0) {
                    x = 0;
                  }
                  size = x;
                }
              } finally {
                if (iter1 != null) {
                  iter1.close();
                }
                if (iter2 != null) {
                  iter1.close();
                }
              }

            } else {
              // not attempting to differentiate between LT & LE
              size = indexStore.size(key) > 0 ? 1 : 0;
            }
          } else {
            size = Integer.MAX_VALUE;
          }
          break;

        case OQLLexerTokenTypes.TOK_GE:
        case OQLLexerTokenTypes.TOK_GT:
          if (matchLevel <= 0 && (key instanceof Number)) {
            int totalSize = indexStore.size();
            if (CompactRangeIndex.testHook != null) {
              CompactRangeIndex.testHook.hook(2);
            }
            if (totalSize > 1) {
              Number keyAsNum = (Number) key;
              int x = 0;
              IndexStoreEntry firstEntry = null;
              CloseableIterator<IndexStoreEntry> iter1 = null;
              CloseableIterator<IndexStoreEntry> iter2 = null;

              try {
                iter1 = indexStore.iterator(null);
                if (iter1.hasNext()) {
                  firstEntry = iter1.next();
                }

                IndexStoreEntry lastEntry = null;

                iter2 = indexStore.descendingIterator(null);
                if (iter2.hasNext()) {
                  lastEntry = iter2.next();
                }


                if (firstEntry != null && lastEntry != null) {
                  Number first = (Number) firstEntry.getDeserializedKey();
                  Number last = (Number) lastEntry.getDeserializedKey();
                  if (first.doubleValue() != last.doubleValue()) {
                    // Shobhit: Now without ReadLoack on index we can end up with 0
                    // in denominator if the numbers are floating-point and
                    // truncated with conversion to long, and the first and last
                    // truncate to the same long, so safest calculation is to
                    // convert to doubles.
                    x = (int) (((last.doubleValue() - keyAsNum.doubleValue()) * totalSize)
                        / (last.doubleValue() - first.doubleValue()));
                  }
                }
                if (x < 0) {
                  x = 0;
                }
                size = x;
              } finally {
                if (iter1 != null) {
                  iter1.close();
                }
              }
            } else {
              // not attempting to differentiate between GT & GE
              size = indexStore.size(key) > 0 ? 1 : 0;
            }
          } else {
            size = Integer.MAX_VALUE;
          }
          break;
      }
    } catch (ClassCastException e) {
      // no values will match in this index because the key types are not the same
      // This means that there will be 0 results and it will be fast to use this index
      // because it has filtered everything out
      return 0;
    } catch (EntryDestroyedException ignore) {
      return Integer.MAX_VALUE;
    } finally {
      updateIndexUseEndStats(start, false);
    }
    return size;
  }

  /** Method called while appropriate lock held on index */
  private void lockedQueryPrivate(Object key, int operator, Collection results,
      CompiledValue iterOps, RuntimeIterator runtimeItr, ExecutionContext context, Set keysToRemove,
      List projAttrib, SelectResults intermediateResults, boolean isIntersection)
      throws TypeMismatchException, FunctionDomainException, NameResolutionException,
      QueryInvocationTargetException {
    if (keysToRemove == null) {
      keysToRemove = new HashSet(0);
    }
    int limit = -1;

    Boolean applyLimit = (Boolean) context.cacheGet(CompiledValue.CAN_APPLY_LIMIT_AT_INDEX);
    if (applyLimit != null && applyLimit) {
      limit = (Integer) context.cacheGet(CompiledValue.RESULT_LIMIT);
    }

    Boolean orderByClause = (Boolean) context.cacheGet(CompiledValue.CAN_APPLY_ORDER_BY_AT_INDEX);
    boolean applyOrderBy = false;
    List orderByAttrs = null;
    if (orderByClause != null && orderByClause) {
      orderByAttrs = (List) context.cacheGet(CompiledValue.ORDERBY_ATTRIB);
      CompiledSortCriterion csc = (CompiledSortCriterion) orderByAttrs.get(0);
      applyOrderBy = true;
    }
    if (isEmpty()) {
      return;
    }
    key = getPdxStringForIndexedPdxKeys(key);
    evaluate(key, operator, results, iterOps, runtimeItr, context, keysToRemove, projAttrib,
        intermediateResults, isIntersection, limit, applyOrderBy, orderByAttrs);
  }

  /** Method called while appropriate lock held on index */
  @Override
  void lockedQuery(Object lowerBoundKey, int lowerBoundOperator, Object upperBoundKey,
      int upperBoundOperator, Collection results, Set keysToRemove, ExecutionContext context)
      throws TypeMismatchException, FunctionDomainException, NameResolutionException,
      QueryInvocationTargetException {
    lowerBoundKey = TypeUtils.indexKeyFor(lowerBoundKey);
    upperBoundKey = TypeUtils.indexKeyFor(upperBoundKey);
    boolean lowerBoundInclusive = lowerBoundOperator == OQLLexerTokenTypes.TOK_GE;
    boolean upperBoundInclusive = upperBoundOperator == OQLLexerTokenTypes.TOK_LE;
    // LowerBound Key inclusive , Upper bound key exclusive

    int limit = -1;
    Boolean applyLimit = (Boolean) context.cacheGet(CompiledValue.CAN_APPLY_LIMIT_AT_INDEX);
    if (applyLimit != null && applyLimit) {
      limit = (Integer) context.cacheGet(CompiledValue.RESULT_LIMIT);
    }
    Boolean orderByClause = (Boolean) context.cacheGet(CompiledValue.CAN_APPLY_ORDER_BY_AT_INDEX);

    List orderByAttrs;
    boolean asc = true;
    boolean multiColOrderBy = false;
    if (orderByClause != null && orderByClause) {
      orderByAttrs = (List) context.cacheGet(CompiledValue.ORDERBY_ATTRIB);
      CompiledSortCriterion csc = (CompiledSortCriterion) orderByAttrs.get(0);
      asc = !csc.getCriterion();
      multiColOrderBy = orderByAttrs.size() > 1;
    }
    // return if the index map is still empty at this stage
    if (isEmpty()) {
      return;
    }
    lowerBoundKey = getPdxStringForIndexedPdxKeys(lowerBoundKey);
    upperBoundKey = getPdxStringForIndexedPdxKeys(upperBoundKey);
    if (keysToRemove == null) {
      keysToRemove = new HashSet();
    }
    CloseableIterator<IndexStoreEntry> iterator = null;
    try {
      if (asc) {
        iterator = indexStore.iterator(lowerBoundKey, lowerBoundInclusive, upperBoundKey,
            upperBoundInclusive, keysToRemove);
      } else {
        iterator = indexStore.descendingIterator(lowerBoundKey, lowerBoundInclusive, upperBoundKey,
            upperBoundInclusive, keysToRemove);
      }
      addToResultsFromEntries(lowerBoundKey, upperBoundKey, lowerBoundOperator, upperBoundOperator,
          iterator, results, null, null, context, null, null, true, multiColOrderBy ? -1 : limit);
    } finally {
      if (iterator != null) {
        iterator.close();
      }
    }
  }


  private void evaluate(Object key, int operator, Collection results, CompiledValue iterOps,
      RuntimeIterator runtimeItr, ExecutionContext context, Set keysToRemove, List projAttrib,
      SelectResults intermediateResults, boolean isIntersection, int limit, boolean applyOrderBy,
      List orderByAttribs) throws TypeMismatchException, FunctionDomainException,
      NameResolutionException, QueryInvocationTargetException {
    boolean multiColOrderBy = false;
    if (keysToRemove == null) {
      keysToRemove = new HashSet(0);
    }
    key = TypeUtils.indexKeyFor(key);
    if (key == null) {
      key = IndexManager.NULL;
    }
    boolean asc = true;
    if (applyOrderBy) {
      CompiledSortCriterion csc = (CompiledSortCriterion) orderByAttribs.get(0);
      asc = !csc.getCriterion();
      multiColOrderBy = orderByAttribs.size() > 1;
    }
    CloseableIterator<IndexStoreEntry> iterator = null;
    try {
      switch (operator) {
        case OQLLexerTokenTypes.TOK_EQ:
          assert keysToRemove.isEmpty();
          iterator = indexStore.get(key);
          addToResultsFromEntries(key, operator, iterator, results, iterOps, runtimeItr, context,
              projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit);
          break;
        case OQLLexerTokenTypes.TOK_LT: {
          if (asc) {
            iterator = indexStore.iterator(null, true, key, false, keysToRemove);
          } else {
            iterator = indexStore.descendingIterator(null, true, key, false, keysToRemove);
          }
          addToResultsFromEntries(key, operator, iterator, results, iterOps, runtimeItr, context,
              projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit);
        }
          break;
        case OQLLexerTokenTypes.TOK_LE: {
          if (asc) {
            iterator = indexStore.iterator(null, true, key, true, keysToRemove);
          } else {
            iterator = indexStore.descendingIterator(null, true, key, true, keysToRemove);
          }

          addToResultsFromEntries(key, operator, iterator, results, iterOps, runtimeItr, context,
              projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit);
        }
          break;
        case OQLLexerTokenTypes.TOK_GT: {
          if (asc) {
            iterator = indexStore.iterator(key, false, keysToRemove);
          } else {
            iterator = indexStore.descendingIterator(key, false, keysToRemove);
          }
          addToResultsFromEntries(key, operator, iterator, results, iterOps, runtimeItr, context,
              projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit);
        }
          break;
        case OQLLexerTokenTypes.TOK_GE: {
          if (asc) {
            iterator = indexStore.iterator(key, true, keysToRemove);
          } else {
            iterator = indexStore.descendingIterator(key, true, keysToRemove);
          }

          addToResultsFromEntries(key, operator, iterator, results, iterOps, runtimeItr, context,
              projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit);
        }
          break;
        case OQLLexerTokenTypes.TOK_NE_ALT:
        case OQLLexerTokenTypes.TOK_NE: {
          keysToRemove.add(key);
          if (asc) {
            iterator = indexStore.iterator(keysToRemove);
          } else {
            iterator = indexStore.descendingIterator(keysToRemove);
          }
          addToResultsFromEntries(key, operator, iterator, results, iterOps, runtimeItr, context,
              projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit);
          // If the key is not null, then add the nulls to the results as this is a not equals query
          if (!IndexManager.NULL.equals(key)) {
            // we pass in the operator TOK_EQ because we want to add results where the key is equal
            // to NULL
            addToResultsFromEntries(IndexManager.NULL, OQLLexerTokenTypes.TOK_EQ,
                indexStore.get(IndexManager.NULL), results, iterOps, runtimeItr, context,
                projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit);
          }
          // If the key is not undefined, then add the undefineds to the results as this is a not
          // equals query
          if (!QueryService.UNDEFINED.equals(key)) {
            // we pass in the operator TOK_EQ because we want to add results where the key is equal
            // to UNDEFINED
            addToResultsFromEntries(QueryService.UNDEFINED, OQLLexerTokenTypes.TOK_EQ,
                indexStore.get(QueryService.UNDEFINED), results, iterOps, runtimeItr, context,
                projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit);
          }
        }
          break;
        default:
          throw new AssertionError("Operator = " + operator);
      } // end switch
    } catch (ClassCastException ex) {
      if (operator == OQLLexerTokenTypes.TOK_EQ) { // result is empty set
        return;
      } else if (operator == OQLLexerTokenTypes.TOK_NE
          || operator == OQLLexerTokenTypes.TOK_NE_ALT) { // put all in result
        keysToRemove.add(key);
        try {
          if (asc) {
            iterator = indexStore.iterator(keysToRemove);
          } else {
            iterator = indexStore.descendingIterator(keysToRemove);
          }
          addToResultsFromEntries(key, OQLLexerTokenTypes.TOK_NE, iterator, results, iterOps,
              runtimeItr, context, projAttrib, intermediateResults, isIntersection,
              multiColOrderBy ? -1 : limit);
        } finally {
          if (iterator != null) {
            iterator.close();
          }
        }
      } else { // otherwise throw exception
        throw new TypeMismatchException("", ex);
      }
    } finally {
      if (iterator != null) {
        iterator.close();
      }
    }
  }

  @Override
  void instantiateEvaluator(IndexCreationHelper indexCreationHelper) {
    evaluator = new IMQEvaluator(indexCreationHelper);
  }

  // Only used by CompactMapRangeIndex. This is due to the way the index initialization happens
  // first we use the IMQEvaluator for CompactMapRangeIndex
  // Each index in CMRI is a CRI that has the CRI.IMQ and not AbstractIndex.IMQ
  // So instead we create create the evaluator
  // because we are not doing index init as usual (each value is just put directly?)
  // we must set the result type to match
  void instantiateEvaluator(IndexCreationHelper ich, ObjectType objectType) {
    instantiateEvaluator(ich);
    ((IMQEvaluator) evaluator).indexResultSetType = objectType;
  }

  @Override
  public ObjectType getResultSetType() {
    return evaluator.getIndexResultSetType();
  }

  /*
   *
   * @param lowerBoundKey the index key to match on
   *
   * @param lowerBoundOperator the operator to use to determine a match
   */
  private void addToResultsFromEntries(Object lowerBoundKey, int lowerBoundOperator,
      CloseableIterator<IndexStoreEntry> entriesIter, Collection result, CompiledValue iterOps,
      RuntimeIterator runtimeItr, ExecutionContext context, List projAttrib,
      SelectResults intermediateResults, boolean isIntersection, int limit)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {
    addToResultsFromEntries(lowerBoundKey, null, lowerBoundOperator, -1, entriesIter, result,
        iterOps, runtimeItr, context, projAttrib, intermediateResults, isIntersection, limit);
  }

  /*
   *
   * @param lowerBoundKey the index key to match on for a lower bound on a ranged query, otherwise
   * the key to match on
   *
   * @param upperBoundKey the index key to match on for an upper bound on a ranged query, otherwise
   * null
   *
   * @param lowerBoundOperator the operator to use to determine a match against the lower bound
   *
   * @param upperBoundOperator the operator to use to determine a match against the upper bound
   */
  private void addToResultsFromEntries(Object lowerBoundKey, Object upperBoundKey,
      int lowerBoundOperator, int upperBoundOperator,
      CloseableIterator<IndexStoreEntry> entriesIter, Collection result, CompiledValue iterOps,
      RuntimeIterator runtimeItr, ExecutionContext context, List projAttrib,
      SelectResults intermediateResults, boolean isIntersection, int limit)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {

    QueryObserver observer = QueryObserverHolder.getInstance();
    boolean limitApplied = false;
    if (entriesIter == null || (limitApplied = verifyLimit(result, limit))) {
      if (limitApplied) {
        if (observer != null) {
          observer.limitAppliedAtIndexLevel(this, limit, result);
        }
      }
      return;
    }

    Set seenKey = null;
    if (IndexManager.IS_TEST_EXPANSION) {
      seenKey = new HashSet();
    }

    while (entriesIter.hasNext()) {
      try {
        // Check if query execution on this thread is canceled.
        QueryMonitor.throwExceptionIfQueryOnCurrentThreadIsCanceled();
        if (IndexManager.testHook != null) {
          if (region.getCache().getLogger().fineEnabled()) {
            region.getCache().getLogger()
                .fine("IndexManager TestHook is set in addToResultsFromEntries.");
          }
          IndexManager.testHook.hook(11);
        }

        IndexStoreEntry indexEntry = null;
        try {
          indexEntry = entriesIter.next();
        } catch (NoSuchElementException ignore) {
          // We are done with all the elements in array.
          // Continue from while.
          continue;
        }

        Object value = indexEntry.getDeserializedValue();

        if (IndexManager.IS_TEST_EXPANSION) {
          Object rk = indexEntry.getDeserializedRegionKey();
          if (seenKey.contains(rk)) {
            continue;
          }
          seenKey.add(rk);

          List expandedResults = expandValue(context, lowerBoundKey, upperBoundKey,
              lowerBoundOperator, upperBoundOperator, value);
          for (final Object o : (Collection) expandedResults) {
            value = o;
            if (value != null) {
              boolean ok = true;

              if (runtimeItr != null) {
