/*
 * 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 static java.lang.System.lineSeparator;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import it.unimi.dsi.fastutil.objects.Object2ObjectOpenHashMap;
import org.apache.logging.log4j.Logger;

import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
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.QueryException;
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.CompiledIteratorDef;
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.DefaultQuery;
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.Support;
import org.apache.geode.cache.query.internal.index.HashIndex.IMQEvaluator.HashIndexComparator;
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.internal.cache.CachedDeserializable;
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.Token;
import org.apache.geode.internal.cache.persistence.query.CloseableIterator;
import org.apache.geode.internal.offheap.StoredObject;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * A HashIndex is an index that can be used for equal and not equals queries It is created only when
 * called explicitly with createHashIndex It requires the indexed expression be 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.
 * <p>
 * This index does not support the storage of projection attributes.
 * <p>
 * Currently this implementation only supports an index on a region path.
 *
 * @deprecated Due to the overhead caused by rehashing while expanding the backing array, Hash Index
 *             has been deprecated since Apache Geode 1.4.0. Instead the use of
 *             {@link CompactRangeIndex} is recommended
 */
@Deprecated
public class HashIndex extends AbstractIndex {
  private static final Logger logger = LogService.getLogger();

  /**
   * ThreadLocal for Map for under update RegionEntries=>oldKey (reverse map) if
   * {@link IndexManager#INPLACE_OBJECT_MODIFICATION} is false.
   */
  protected ThreadLocal<Object2ObjectOpenHashMap> entryToOldKeysMap;

  /**
   * Map for valueOf(indexedExpression)=>RegionEntries. SortedMap<Object, (RegionEntry |
   * List<RegionEntry>)>. Package access for unit tests.
   */
  final HashIndexSet entriesSet;

  /**
   * Map for RegionEntries=>value of indexedExpression (reverse map) maintained by the HashIndexSet
   * entrieSet
   */
  private ConcurrentMap<Object, Object> entryToValuesMap = null;

  private boolean indexOnRegionKeys = false;

  private boolean indexOnValues = false;

  // used for sorting asc and desc queries
  private HashIndexComparator comparator;

  /**
   * Create a HashIndex that can be used when executing queries.
   *
   * @param indexName the name of this index, used for statistics collection
   * @param indexedExpression the expression to index on, a function dependent on region entries
   *        individually, limited to a path expression.
   * @param fromClause expression that evaluates to the collection(s) that will be queried over,
   *        must contain one and only one region path, and only one iterator.
   * @param projectionAttributes not used
   * @param definitions the canonicalized definitions
   */
  public HashIndex(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);
    RegionAttributes ra = region.getAttributes();


    if (IndexManager.isObjectModificationInplace()) {
      entryToValuesMap = new ConcurrentHashMap(ra.getInitialCapacity(), ra.getLoadFactor(),
          ra.getConcurrencyLevel());
    } else {
      if (entryToOldKeysMap == null) {
        entryToOldKeysMap = new ThreadLocal<>();
      }
    }

    entriesSet = new HashIndexSet();
  }

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

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

  @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();
  }

  /**
   * Add/Updates the index forward and reverse map. If index key for a RegionEntry is found same as
   * previous key no update is performed.
   *
   * This also updates the {@link IndexStatistics} numKeys and numValues as and when appropriate.
   * One thing to notice though is no increment in numValues is performed if old key and new index
   * key are found equal using {@link Object#equals(Object)}.
   */
  private void basicAddMapping(Object key, RegionEntry entry) throws IMQException {

    try {
      if (DefaultQuery.testHook != null) {
        DefaultQuery.testHook.doTestHook(
            DefaultQuery.TestHook.SPOTS.BEFORE_ADD_OR_UPDATE_MAPPING_OR_DESERIALIZING_NTH_STREAMINGOPERATION,
            null, null);
      }
      Object newKey = TypeUtils.indexKeyFor(key);
      if (newKey.equals(QueryService.UNDEFINED)) {
        Object targetObject = getTargetObjectForUpdate(entry);
        if (Token.isInvalidOrRemoved(targetObject)) {
          // This should not happen as a token should only be added during gii
          // meaning we do not have an old mapping
          // we will continue to remove the old mapping to be safe and log a fine level message
          Object oldKey = null;
          if (IndexManager.isObjectModificationInplace()
              && entryToValuesMap.containsKey(entry)) {
            oldKey = entryToValuesMap.get(entry);
          } else if (!IndexManager.isObjectModificationInplace()
              && entryToOldKeysMap != null) {
            Map oldKeyMap = entryToOldKeysMap.get();
            if (oldKeyMap != null) {
              oldKey = TypeUtils.indexKeyFor(oldKeyMap.get(entry));
            }
          }
          if (oldKey != null) {
            if (logger.isDebugEnabled()) {
              logger
                  .debug("A removed or invalid token was being added, and we had an old mapping.");
            }
            removeFromEntriesSet(oldKey, entry, true);
          }
          return;
        }
      }

      // Before adding the entry with new value, remove it from reverse map and
      // using the oldValue remove entry from the forward map.
      // Reverse-map is used based on the system property
      Object oldKey = getOldKey(entry);

      int indexSlot = entriesSet.add(newKey, entry);

      if (indexSlot >= 0) {
        // Update the reverse map
        if (IndexManager.isObjectModificationInplace()) {
          entryToValuesMap.put(entry, newKey);
        }
        if (newKey != null && oldKey != null) {
          removeFromEntriesSet(oldKey, entry, false, indexSlot);
        }
        // Update Stats after real addition
        internalIndexStats.incNumValues(1);

      }
    } catch (TypeMismatchException ex) {
      throw new IMQException("Could not add object of type " + key.getClass().getName(), ex);
    }
  }

  private Object getOldKey(RegionEntry entry) throws TypeMismatchException {
    Object oldKey = null;
    if (IndexManager.isObjectModificationInplace() && entryToValuesMap.containsKey(entry)) {
      oldKey = entryToValuesMap.get(entry);
    } else if (!IndexManager.isObjectModificationInplace() && entryToOldKeysMap != null) {
      Map oldKeyMap = entryToOldKeysMap.get();
      if (oldKeyMap != null) {
        oldKey = TypeUtils.indexKeyFor(oldKeyMap.get(entry));
      }
    }
    return oldKey;
  }

  /**
   * @param opCode one of OTHER_OP, BEFORE_UPDATE_OP, AFTER_UPDATE_OP.
   */
  @Override
  void removeMapping(RegionEntry entry, int opCode) throws IMQException {
    // logger.debug("##### In RemoveMapping: entry : "
    // + entry );
    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.
        entryToOldKeysMap.set(new Object2ObjectOpenHashMap(1));
        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 (entryToOldKeysMap != null) {
        basicRemoveMapping(entryToOldKeysMap.get().get(entry), entry, true);
      }
    } else if (opCode == CLEAN_UP_THREAD_LOCALS) {
      if (entryToOldKeysMap != null) {
        entryToOldKeysMap.remove();
      }
    } else {
      // Need to reset the thread-local map as many puts and destroys might
      // happen in same thread.
      if (entryToOldKeysMap != null) {
        entryToOldKeysMap.remove();
      }
      evaluator.evaluate(entry, false);
      internalIndexStats.incNumUpdates();
    }
  }

  /**
   * Remove an index entry for a RegionEntry when invalidate/destroy is called OR new index key is
   * inserted for the RegionEntry. In case of update only forward map is cleared of old key and NO
   * update is performed on reverse map as that has already been done during
   * {@link HashIndex#basicAddMapping(Object, RegionEntry)}.
   *
   * @param key - Index key.
   * @param entry RegionEntry for which is being updated by user.
   * @param updateReverseMap true only when RegionEntry is invalidated/destroyed.
   */
  private void basicRemoveMapping(Object key, RegionEntry entry, boolean updateReverseMap)
      throws IMQException {
    // after removal, trim the ArrayList to prevent
    // too much extra space.
    // Ideally we would only trim if there is excessive
    // space, but there is no way to ask an ArrayList what
    // it's current capacity is..so we trim after every
    // removal
    try {
      Object newKey = TypeUtils.indexKeyFor(key);
      removeFromEntriesSet(newKey, entry, updateReverseMap);
    } catch (TypeMismatchException ex) {
      throw new IMQException("Could not add object of type " + key.getClass().getName(), ex);
    }
  }

  private void removeFromEntriesSet(Object newKey, RegionEntry entry, boolean updateReverseMap) {
    removeFromEntriesSet(newKey, entry, updateReverseMap, -1);
  }

  private void removeFromEntriesSet(Object newKey, RegionEntry entry, boolean updateReverseMap,
      int ignoreThisSlot) {
    if (entriesSet.remove(newKey, entry, ignoreThisSlot)) {
      if (updateReverseMap && IndexManager.isObjectModificationInplace()) {
        entryToValuesMap.remove(entry);
      }
      internalIndexStats.incNumValues(-1);
    }
  }

  // // IndexProtocol interface implementation
  @Override
  public boolean clear() throws QueryException {
    throw new UnsupportedOperationException("Not yet implemented");
  }

  /**
   * computes the resultset of an equijoin query
   */
  @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 inner = null;
    try {
      // We will iterate over each of the valueToEntries Map to obtain the keys
      Iterator outer = entriesSet.iterator();
      if (indx instanceof CompactRangeIndex) {
        inner = ((CompactRangeIndex) indx).getIndexStorage().iterator(null);
      } else {
        inner = ((RangeIndex) indx).getValueToEntriesMap().entrySet().iterator();
      }
      Map.Entry outerEntry = null;
      Object innerEntry = null;
      Object outerKey = null;
      Object innerKey = null;
      // boolean incrementOuter = true;
      boolean incrementInner = true;
      outer: while (outer.hasNext()) {
        // if (incrementOuter) {
        outerEntry = (Map.Entry) outer.next();
        // }
        outerKey = outerEntry.getKey();
        // TODO: eliminate use of 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;
            if (innerEntry instanceof IndexStoreEntry) {
              innerValue = ((CompactRangeIndex) indx).getIndexStorage().get(outerKey);
            } else {
              innerValue = ((Map.Entry) innerEntry).getValue();
            }
            populateListForEquiJoin(data, outerEntry.getValue(), innerValue, context, innerKey);

            incrementInner = true;
            continue outer;
          } else if (compare < 0) {
            // Asif :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) {
        ((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.
   *
   * @return true if entry value and index value are consistent.
   */
  private boolean verifyInnerAndOuterEntryValues(RegionEntry entry, ExecutionContext context,
      IndexInfo indexInfo, Object keyVal) throws FunctionDomainException, TypeMismatchException,
      NameResolutionException, QueryInvocationTargetException {
    // Verify index key in value.
    CompactRangeIndex index = (CompactRangeIndex) indexInfo._getIndex();
    RuntimeIterator runtimeItr = index.getRuntimeIteratorForThisIndex(context, indexInfo);
    if (runtimeItr != null) {
      runtimeItr.setCurrent(getTargetObject(entry));
    }
    return evaluateEntry(indexInfo, context, keyVal);
  }

  @Override
  public int getSizeEstimate(Object key, int operator, int matchLevel)
      throws TypeMismatchException {
    // Get approx size;
    int size = 0;
    long start = updateIndexUseStats(false);
    try {
      switch (operator) {
        case OQLLexerTokenTypes.TOK_EQ: {
          key = TypeUtils.indexKeyFor(key);
          size = entriesSet.size(key);
        }
          break;
        case OQLLexerTokenTypes.TOK_NE_ALT:
        case OQLLexerTokenTypes.TOK_NE:
          size = region.size();
          key = TypeUtils.indexKeyFor(key);
          size = entriesSet.size(key);
          break;
      }
    } finally {
      updateIndexUseEndStats(start, false);
    }
    return size;
  }

  /**
   * Convert a RegionEntry or THashSet<RegionEntry> to be consistently a Collection
   */
  private Collection regionEntryCollection(Object regionEntries) {
    if (regionEntries == null) {
      return null;
    }
    if (regionEntries instanceof RegionEntry) {
      return Collections.singleton(regionEntries);
    }
    return (Collection) regionEntries;
  }

  /** 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;
    }
    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 {
    throw new UnsupportedOperationException(
        "Range grouping for HashIndex condition is not supported");

  }

  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;
    }

    try {
      long iteratorCreationTime = cache.cacheTimeMillis();

      switch (operator) {
        case OQLLexerTokenTypes.TOK_EQ:
          assert keysToRemove.isEmpty();
          addToResultsFromEntries(entriesSet.get(key), results, iterOps, runtimeItr, context,
              projAttrib, intermediateResults, isIntersection, multiColOrderBy ? -1 : limit,
              keysToRemove, applyOrderBy, asc, iteratorCreationTime);
          break;
        case OQLLexerTokenTypes.TOK_NE_ALT:
        case OQLLexerTokenTypes.TOK_NE: {
          keysToRemove.add(key);
          addToResultsFromEntries(entriesSet.getAllNotMatching(keysToRemove), results, iterOps,
              runtimeItr, context, projAttrib, intermediateResults, isIntersection,
              multiColOrderBy ? -1 : limit, keysToRemove, applyOrderBy, asc, iteratorCreationTime);
        }
          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
        keysToRemove.add(key);
        long iteratorCreationTime = cache.cacheTimeMillis();
        addToResultsFromEntries(entriesSet.getAllNotMatching(keysToRemove), results, iterOps,
            runtimeItr, context, projAttrib, intermediateResults, isIntersection,
            multiColOrderBy ? -1 : limit, keysToRemove, applyOrderBy, asc, iteratorCreationTime);
      } else { // otherwise throw exception
        throw new TypeMismatchException("", ex);
      }
    }
  }

  @Override
  void instantiateEvaluator(IndexCreationHelper indexCreationHelper) {
    evaluator = new IMQEvaluator(indexCreationHelper);
    entriesSet.setEvaluator((HashIndex.IMQEvaluator) evaluator);
    comparator = ((IMQEvaluator) evaluator).comparator;
  }

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


  /**
   * @param entriesIter is Iterable<RegionEntry>
   */
  private void addToResultsFromEntries(Iterator entriesIter, Collection result,
      CompiledValue iterOps, RuntimeIterator runtimeItr, ExecutionContext context, List projAttrib,
      SelectResults intermediateResults, boolean isIntersection, int limit, Set keysToRemove,
      boolean applyOrderBy, boolean asc, long iteratorCreationTime) throws FunctionDomainException,
      TypeMismatchException, NameResolutionException, QueryInvocationTargetException {
    QueryObserver observer = QueryObserverHolder.getInstance();
    if (result == null || limit != -1 && result.size() == limit) {
      return;
    }
    List orderedKeys = null;
    List orderedResults = null;
    if (applyOrderBy) {
      orderedKeys = new ArrayList();
      orderedResults = new ArrayList();
    }
    int i = 0;
    while (entriesIter.hasNext()) {
      // Check if query execution on this thread is canceled.
      QueryMonitor.throwExceptionIfQueryOnCurrentThreadIsCanceled();
      if (IndexManager.testHook != null) {
        if (logger.isDebugEnabled()) {
          logger.debug("IndexManager TestHook is set in addToResultsFromEntries.");
        }
        IndexManager.testHook.hook(11);
      }
      Object obj = entriesIter.next();
      Object key = null;
      if (obj != null && obj != HashIndexSet.REMOVED) {
        RegionEntry re = (RegionEntry) obj;
        if (applyOrderBy) {
          key = ((HashIndex.IMQEvaluator) evaluator).evaluateKey(obj);
          orderedKeys.add(new Object[] {key, i++});
          addValueToResultSet(re, orderedResults, iterOps, runtimeItr, context, projAttrib,
              intermediateResults, isIntersection, limit, observer, iteratorCreationTime);
        } else {
          addValueToResultSet(re, result, iterOps, runtimeItr, context, projAttrib,
              intermediateResults, isIntersection, limit, observer, iteratorCreationTime);
        }
      }
    }
    if (applyOrderBy) {
      /*
       * For orderby queries, 1. Store the keys in a list along with the order. 2. Store the results
       * in another temp list. 3. Sort the keys. The order will also get sorted. 4. Fetch the result
       * objects from the temp list according to the sorted orders from the sorted list and add to
       * the result collection.
       */
      Collections.sort(orderedKeys, comparator);
      if (!asc) {
        Collections.reverse(orderedKeys);
      }
      Object[] temp = orderedResults.toArray();
      List tempResults = new ArrayList(temp.length);
      for (Object o : orderedKeys) {
        int index = (Integer) ((Object[]) o)[1];
        tempResults.add(temp[index]);
      }
      result.addAll(tempResults);
    }
  }

  private void addValueToResultSet(RegionEntry re, Collection result, CompiledValue iterOps,
      RuntimeIterator runtimeItr, ExecutionContext context, List projAttrib,
      SelectResults intermediateResults, boolean isIntersection, int limit, QueryObserver observer,
      long iteratorCreationTime) throws FunctionDomainException, TypeMismatchException,
      NameResolutionException, QueryInvocationTargetException {
    Object value = getTargetObject(re);
    if (value != null) {
      boolean ok = true;
      // If the region entry is currently being updated or it has been modified since starting
      // iteration
      // we will reevaluate to be sure the value still matches the key
      if (re.isUpdateInProgress()
          || IndexManager.needsRecalculation(iteratorCreationTime, re.getLastModified())) {
        IndexInfo indexInfo = (IndexInfo) context.cacheGet(CompiledValue.INDEX_INFO);
        if (runtimeItr == null) {
          runtimeItr = getRuntimeIteratorForThisIndex(context, indexInfo);
          if (runtimeItr == null) {
            // could not match index with iterator
            throw new QueryInvocationTargetException("Query alias's must be used consistently");
          }
        }
        runtimeItr.setCurrent(value);
        // Verify index key in region entry value.
        ok = evaluateEntry(indexInfo, context, null);
      }
      if (runtimeItr != null) {
        runtimeItr.setCurrent(value);
      }
      if (ok && runtimeItr != null && iterOps != null) {
        ok = QueryUtils.applyCondition(iterOps, context);
      }
      if (ok) {
        applyCqOrProjection(projAttrib, context, result, value, intermediateResults,
            isIntersection, re.getKey());
        if (limit != -1 && result.size() == limit) {
          observer.limitAppliedAtIndexLevel(this, limit, result);
          return;
        }
      }
    }
  }

  /**
   * This evaluates the left and right side of a where condition for which this Index was used.
   * Like, if condition is "ID > 1", {@link IndexInfo} will contain Left as ID, Right as '1' and
   * operator as TOK_GT. This method will evaluate ID from region entry value and verify the ID > 1.
   *
   * Note: IndexInfo is created for each query separately based on the condition being evaluated
   * using the Index.
   *
   * @return true if RegionEntry value satisfies the where condition (contained in IndexInfo).
   */
  private boolean evaluateEntry(IndexInfo indexInfo, ExecutionContext context, Object keyVal)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {
    CompiledValue path = indexInfo._path();
    Object left = path.evaluate(context);
    CompiledValue key = indexInfo._key();
    Object right = null;

    // For CompiledUndefined indexInfo has null key.
    if (keyVal == null && key == null) {
      return left == QueryService.UNDEFINED;
    }

    if (key != null) {
      right = key.evaluate(context);
    } else {
      right = keyVal;
    }

    int operator = indexInfo._operator();
    if (left == null && right == null) {
      return Boolean.TRUE;
    } else {
      return (Boolean) TypeUtils.compare(left, right, operator);
    }
  }

  /**
   * Get the object of interest from the region entry. For now it always gets the deserialized
   * value.
   */
  private Object getTargetObject(RegionEntry entry) {
    if (indexOnValues) {
      // OFFHEAP: incrc, deserialize, decrc
      Object o = entry.getValue((LocalRegion) getRegion());
      try {
        if (o == Token.INVALID) {
          return null;
        }
        if (o instanceof CachedDeserializable) {
          return ((CachedDeserializable) o).getDeserializedForReading();
        }
      } catch (EntryDestroyedException ignored) {
        return null;
      }
      return o;
    } else if (indexOnRegionKeys) {
      return entry.getKey();
    }
    return new NonTXEntry((LocalRegion) getRegion(), entry);
  }

  private Object getTargetObjectForUpdate(RegionEntry entry) {
    if (indexOnValues) {
      Object o = entry.getValueOffHeapOrDiskWithoutFaultIn((LocalRegion) getRegion());
      try {
        if (o instanceof StoredObject) {
          StoredObject ohval = (StoredObject) o;
          try {
            o = ohval.getDeserializedForReading();
          } finally {
            ohval.release();
          }
        } else if (o instanceof CachedDeserializable) {
          o = ((CachedDeserializable) o).getDeserializedForReading();
        }
      } catch (EntryDestroyedException ignored) {
        return Token.INVALID;
      }
      return o;
    } else if (indexOnRegionKeys) {
      return entry.getKey();
    }
    return new NonTXEntry((LocalRegion) getRegion(), entry);
  }

  @Override
  void recreateIndexData() throws IMQException {
    // Mark the data maps to null & call the initialization code of index
    entriesSet.clear();
    if (IndexManager.isObjectModificationInplace()) {
      entryToValuesMap.clear();
    }
    int numKeys = (int) internalIndexStats.getNumberOfKeys();
    if (numKeys > 0) {
      internalIndexStats.incNumKeys(-numKeys);
    }
    int numValues = (int) internalIndexStats.getNumberOfValues();
    if (numValues > 0) {
      internalIndexStats.incNumValues(-numValues);
    }
    int updates = (int) internalIndexStats.getNumUpdates();
    if (updates > 0) {
      internalIndexStats.incNumUpdates(updates);
    }
    initializeIndex(true);
  }

  public String dump() {
    StringBuilder sb = new StringBuilder(toString()).append(" {").append(lineSeparator());
    sb.append(" -----------------------------------------------").append(lineSeparator());
    for (Object anEntriesSet : entriesSet) {
      Entry indexEntry = (Entry) anEntriesSet;
      sb.append(" Key = ").append(indexEntry.getKey()).append(lineSeparator());
