/*
 * 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.util.Objects.hash;

import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.query.IndexMaintenanceException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.TypeMismatchException;
import org.apache.geode.cache.query.internal.CompiledComparison;
import org.apache.geode.cache.query.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.index.AbstractIndex.InternalIndexStatistics;
import org.apache.geode.cache.query.internal.parse.OQLLexerTokenTypes;
import org.apache.geode.cache.query.internal.types.TypeUtils;
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.pdx.internal.PdxInstanceImpl;

/**
 * The in-memory index storage
 *
 * @since GemFire 8.0
 */
public class MemoryIndexStore implements IndexStore {
  /**
   * Map for valueOf(indexedExpression)=>RegionEntries. SortedMap<Object, (RegionEntry |
   * List<RegionEntry>)>. Package access for unit tests.
   */
  final ConcurrentNavigableMap valueToEntriesMap =
      new ConcurrentSkipListMap(TypeUtils.getExtendedNumericComparator());

  // number of keys
  private final AtomicInteger numIndexKeys = new AtomicInteger(0);

  // Map for RegionEntries=>value of indexedExpression (reverse map)
  private ConcurrentMap entryToValuesMap;

  private final InternalIndexStatistics internalIndexStats;

  private final InternalCache cache;

  private final Region region;

  private boolean indexOnRegionKeys;

  private boolean indexOnValues;

  // Used as a place holder for an indexkey collection for when a thread is about to change
  // the collection from index elem to concurrent hash set. Solution for #47475 where
  // we could be missing removes due to different threads grabbing occurrences of the same
  // index elem (or collection), one would transition and addAll to a new collection,
  // while the other would execute a remove on the index elem.
  // both would complete but the remove would have been lost because we had already added it to the
  // new collection
  private final Object TRANSITIONING_TOKEN = new IndexElemArray(1);

  MemoryIndexStore(Region region, InternalIndexStatistics internalIndexStats, InternalCache cache) {
    this.region = region;
    RegionAttributes ra = region.getAttributes();
    // Initialize the reverse-map if in-place modification is set by the
    // application.
    if (IndexManager.isObjectModificationInplace()) {
      entryToValuesMap = new ConcurrentHashMap(ra.getInitialCapacity(), ra.getLoadFactor(),
          ra.getConcurrencyLevel());
    }
    this.internalIndexStats = internalIndexStats;
    this.cache = cache;
  }

  @Override
  public void updateMapping(Object indexKey, Object oldKey, RegionEntry re, Object oldValue)
      throws IMQException {
    try {

      if (DefaultQuery.testHook != null) {
        DefaultQuery.testHook.doTestHook(
            DefaultQuery.TestHook.SPOTS.BEFORE_ADD_OR_UPDATE_MAPPING_OR_DESERIALIZING_NTH_STREAMINGOPERATION,
            null, null);
      }

      // Check if reverse-map is present.
      if (IndexManager.isObjectModificationInplace()) {
        // If reverse map get the old index key from reverse map.
        if (entryToValuesMap.containsKey(re)) {
          oldKey = entryToValuesMap.get(re);
        }
      } else {
        // Check if the old value and new value same.
        // If they are same, that means the value got updated in place.
        // In the absence of reverse-map find the old index key from
        // forward map.
        if (oldValue != null && oldValue == getTargetObjectInVM(re)) {
          oldKey = getOldKey(indexKey, re);
        }
      }

      // No need to update the map if new and old index key are same.
      if (oldKey != null && oldKey.equals(TypeUtils.indexKeyFor(indexKey))) {
        return;
      }

      boolean retry = false;
      indexKey = TypeUtils.indexKeyFor(indexKey);
      if (indexKey.equals(QueryService.UNDEFINED)) {
        Object targetObject = getTargetObjectForUpdate(re);
        if (Token.isInvalidOrRemoved(targetObject)) {
          if (oldKey != null) {
            basicRemoveMapping(oldKey, re, false);
          }
          return;
        }
      }

      do {
        retry = false;
        Object regionEntries = valueToEntriesMap.putIfAbsent(indexKey, re);
        if (regionEntries == TRANSITIONING_TOKEN) {
          retry = true;
          continue;
        } else if (regionEntries == null) {
          if (indexKey != IndexManager.NULL && indexKey != QueryService.UNDEFINED) {
            internalIndexStats.incNumKeys(1);
          }
          numIndexKeys.incrementAndGet();
        } else if (regionEntries instanceof RegionEntry) {
          IndexElemArray elemArray = new IndexElemArray();
          if (DefaultQuery.testHook != null) {
            DefaultQuery.testHook.doTestHook(
                DefaultQuery.TestHook.SPOTS.BEGIN_TRANSITION_FROM_REGION_ENTRY_TO_ELEMARRAY, null,
                null);
          }
          elemArray.add(regionEntries);
          elemArray.add(re);
          if (!valueToEntriesMap.replace(indexKey, regionEntries, elemArray)) {
            retry = true;
          }
          if (DefaultQuery.testHook != null) {
            DefaultQuery.testHook
                .doTestHook(DefaultQuery.TestHook.SPOTS.TRANSITIONED_FROM_REGION_ENTRY_TO_ELEMARRAY,
                    null, null);
          }
          if (DefaultQuery.testHook != null) {
            DefaultQuery.testHook.doTestHook(
                DefaultQuery.TestHook.SPOTS.COMPLETE_TRANSITION_FROM_REGION_ENTRY_TO_ELEMARRAY,
                null, null);
          }
        } else if (regionEntries instanceof IndexConcurrentHashSet) {
          // This synchronized is for avoiding conflcts with remove of
          // ConcurrentHashSet when set size becomes zero during
          // basicRemoveMapping();
          synchronized (regionEntries) {
            ((IndexConcurrentHashSet) regionEntries).add(re);
          }
          if (regionEntries != valueToEntriesMap.get(indexKey)) {
            retry = true;
          }
        } else {
          IndexElemArray elemArray = (IndexElemArray) regionEntries;
          synchronized (elemArray) {
            int threshold = IndexManager.INDEX_ELEMARRAY_THRESHOLD;
            if (IndexManager.INDEX_ELEMARRAY_THRESHOLD_FOR_TESTING > 0) {
              threshold = IndexManager.INDEX_ELEMARRAY_THRESHOLD_FOR_TESTING;
            }
            if (elemArray.size() >= threshold) {
              IndexConcurrentHashSet set =
                  new IndexConcurrentHashSet(IndexManager.INDEX_ELEMARRAY_THRESHOLD + 20, 0.75f, 1);
              // Replace first so that we are sure that the set is placed in
              // index then we should add old elements in the new set.

              if (DefaultQuery.testHook != null) {
                DefaultQuery.testHook.doTestHook(
                    DefaultQuery.TestHook.SPOTS.BEGIN_TRANSITION_FROM_ELEMARRAY_TO_CONCURRENT_HASH_SET,
                    null, null);
              }
              // on a remove from the elem array, another thread could start and complete its remove
              // at this point, that is why we need to replace before adding the elem array elements
              // once we put this set into the forward map, we know any future removes are either
              // captured
              // by our instance of the elem array, or the remove operations will need to do a
              // retry?
              if (!valueToEntriesMap.replace(indexKey, regionEntries, TRANSITIONING_TOKEN)) {
                retry = true;
              } else {
                if (DefaultQuery.testHook != null) {
                  DefaultQuery.testHook
                      .doTestHook(DefaultQuery.TestHook.SPOTS.TRANSITIONED_FROM_ELEMARRAY_TO_TOKEN,
                          null, null);
                }
                set.add(re);
                set.addAll(elemArray);
                if (!valueToEntriesMap.replace(indexKey, TRANSITIONING_TOKEN, set)) {
                  // This should never happen. If we see this in the log, then something is wrong
                  // with the TRANSITIONING TOKEN and synchronization of changing collection types
                  // we should then just go from RE to CHS and completely remove the Elem Array.
                  region.getCache().getLogger().warning(
                      "Unable to transition from index elem to concurrent hash set.  Index needs to be recreated");
                  throw new IndexMaintenanceException(
                      "Unable to transition from index elem to concurrent hash set.  Index needs to be recreated");
                }
                if (DefaultQuery.testHook != null) {
                  DefaultQuery.testHook.doTestHook(
                      DefaultQuery.TestHook.SPOTS.COMPLETE_TRANSITION_FROM_ELEMARRAY_TO_CONCURRENT_HASH_SET,
                      null, null);
                }
              }
            } else {
              elemArray.add(re);
              if (regionEntries != valueToEntriesMap.get(indexKey)) {
                retry = true;
              }
            }
          }
        }

        // Add to reverse Map with the new value.
        if (!retry) {

          // remove from forward map in case of update
          // oldKey is not null only for an update
          if (oldKey != null) {
            basicRemoveMapping(oldKey, re, false);
          }

          if (IndexManager.isObjectModificationInplace()) {
            entryToValuesMap.put(re, indexKey);
          }
        }
      } while (retry);
    } catch (TypeMismatchException ex) {
      throw new IMQException("Could not add object of type " + indexKey.getClass().getName(), ex);
    }
    internalIndexStats.incNumValues(1);
  }

  /**
   * Find the old key by traversing the forward map in case of in-place update modification If not
   * found it means the value object was modified with same value. So oldKey is same as newKey.
   *
   */
  private Object getOldKey(Object newKey, RegionEntry entry) throws TypeMismatchException {
    for (Object mapEntry : valueToEntriesMap.entrySet()) {
      Object regionEntries = ((Entry) mapEntry).getValue();
      Object indexKey = ((Entry) mapEntry).getKey();
      // if more than one index key maps to the same RegionEntry that
      // means there has been an in-place modification
      if (TypeUtils.compare(indexKey, newKey, CompiledComparison.TOK_NE).equals(Boolean.TRUE)) {
        if (regionEntries instanceof RegionEntry && regionEntries.equals(entry)) {
          return indexKey;
        } else if (regionEntries instanceof Collection) {
          Collection coll = (Collection) regionEntries;
          if (coll.contains(entry)) {
            return indexKey;
          }
        }
      }
    }
    return newKey;
  }

  @Override
  public void addMapping(Object indexKey, RegionEntry re) throws IMQException {
    // for add, oldkey is null
    updateMapping(indexKey, null, re, null);
  }

  @Override
  public void removeMappingGII(Object indexKey, RegionEntry re) throws IMQException {
    doRemoveMapping(indexKey, re, false);
  }

  @Override
  public void removeMapping(Object indexKey, RegionEntry re) throws IMQException {
    doRemoveMapping(indexKey, re, true);
  }

  private void doRemoveMapping(Object indexKey, RegionEntry re, boolean findOldKey)
      throws IMQException {
    // Remove from forward map
    boolean found = basicRemoveMapping(indexKey, re, findOldKey);
    // Remove from reverse map.
    // We do NOT need to synchronize here as different RegionEntries will be
    // operating concurrently i.e. different keys in entryToValuesMap which
    // is a concurrent map.
    if (found && IndexManager.isObjectModificationInplace()) {
      entryToValuesMap.remove(re);
    }
  }

  private boolean basicRemoveMapping(Object key, RegionEntry entry, boolean findOldKey)
      throws IMQException {
    boolean found = false;
    boolean possiblyAlreadyRemoved = false;
    try {
      Object newKey = convertToIndexKey(key, entry);
      if (DefaultQuery.testHook != null) {
        DefaultQuery.testHook.doTestHook(DefaultQuery.TestHook.SPOTS.ATTEMPT_REMOVE, null, null);
      }
      boolean retry = false;
      do {
        retry = false;
        Object regionEntries = valueToEntriesMap.get(newKey);
        if (regionEntries == TRANSITIONING_TOKEN) {
          if (DefaultQuery.testHook != null) {
            DefaultQuery.testHook.doTestHook(DefaultQuery.TestHook.SPOTS.ATTEMPT_RETRY, null, null);
          }
          retry = true;
          continue;
        } else if (regionEntries != null) {
          if (regionEntries instanceof RegionEntry) {
            found = regionEntries == entry;
            if (found) {
              if (valueToEntriesMap.remove(newKey, regionEntries)) {
                if (newKey != IndexManager.NULL && newKey != QueryService.UNDEFINED) {
                  internalIndexStats.incNumKeys(-1);
                }
                numIndexKeys.decrementAndGet();
              } else {
                // is another thread has since done an add and shifted us into a collection
                retry = true;
              }
            }
          } else {
            Collection entries = (Collection) regionEntries;
            if (DefaultQuery.testHook != null) {
              DefaultQuery.testHook
                  .doTestHook(DefaultQuery.TestHook.SPOTS.BEGIN_REMOVE_FROM_ELEM_ARRAY, null, null);
            }
            found = entries.remove(entry);
            if (DefaultQuery.testHook != null) {
              DefaultQuery.testHook
                  .doTestHook(DefaultQuery.TestHook.SPOTS.REMOVE_CALLED_FROM_ELEM_ARRAY, null,
                      null);
            }
            // This could be IndexElementArray and might be changing to Set
            // If the remove occurred before changing to a set, then next time it will not be
            // "found"
            // However the end effect would be that it was removed
            if (entries instanceof IndexElemArray) {
              if (!valueToEntriesMap.replace(newKey, entries, entries)) {
                retry = true;
                possiblyAlreadyRemoved = found;
                continue;
              }
            }

            if (entries.isEmpty()) {
              // because entries collection is empty, remove and decrement
              // value
              synchronized (entries) {
                if (entries.isEmpty()) {
                  if (valueToEntriesMap.remove(newKey, entries)) {
                    if (newKey != IndexManager.NULL && newKey != QueryService.UNDEFINED) {
                      internalIndexStats.incNumKeys(-1);
                    }
                    numIndexKeys.decrementAndGet();
                  }
                }
              }
            }
            if (DefaultQuery.testHook != null) {
              DefaultQuery.testHook
                  .doTestHook(DefaultQuery.TestHook.SPOTS.COMPLETE_REMOVE_FROM_ELEM_ARRAY, null,
                      null);
            }
          }
        }
      } while (retry);
    } catch (TypeMismatchException ex) {
      throw new IMQException("Could not add object of type " + key.getClass().getName(), ex);
    }
    if (found) {
      // Update stats if entry was actually removed
      internalIndexStats.incNumValues(-1);
    } else if (!found && !possiblyAlreadyRemoved && !IndexManager.isObjectModificationInplace()
        && key != null) {
      // if there is an inplace-modification find old key by iterating
      // over fwd map and then remove the mapping
      if (findOldKey) {
        try {
          Object oldKey = getOldKey(key, entry);
          found = basicRemoveMapping(oldKey, entry, false);
        } catch (TypeMismatchException e) {
          throw new IMQException("Could not find old key: " + key.getClass().getName(), e);
        }
      }
    }
    return found;
  }

  private Object convertToIndexKey(Object key, RegionEntry entry) throws TypeMismatchException {
    Object newKey;
    if (IndexManager.isObjectModificationInplace() && entryToValuesMap.containsKey(entry)) {
      newKey = entryToValuesMap.get(entry);
    } else {
      newKey = TypeUtils.indexKeyFor(key);
    }
    return newKey;
  }

  @Override
  public CloseableIterator<IndexStoreEntry> get(Object indexKey) {
    return new MemoryIndexStoreIterator(
        valueToEntriesMap.subMap(indexKey, true, indexKey, true), indexKey, null);
  }

  @Override
  public CloseableIterator<IndexStoreEntry> iterator(Object start, boolean startInclusive,
      Object end, boolean endInclusive, Collection keysToRemove) {
    if (start == null) {
      return new MemoryIndexStoreIterator(valueToEntriesMap.headMap(end, endInclusive), null,
          keysToRemove);
    }
    return new MemoryIndexStoreIterator(
        valueToEntriesMap.subMap(start, startInclusive, end, endInclusive), null,
        keysToRemove);
  }

  @Override
  public CloseableIterator<IndexStoreEntry> iterator(Object start, boolean startInclusive,
      Collection keysToRemove) {
    return new MemoryIndexStoreIterator(valueToEntriesMap.tailMap(start, startInclusive), null,
        keysToRemove);
  }

  public Iterator<IndexStoreEntry> getKeysIterator() {
    return new MemoryIndexStoreKeyIterator(valueToEntriesMap);
  }

  @Override
  public CloseableIterator<IndexStoreEntry> iterator(Collection keysToRemove) {
    return new MemoryIndexStoreIterator(valueToEntriesMap, null, keysToRemove);
  }

  @Override
  public CloseableIterator<IndexStoreEntry> descendingIterator(Object start, boolean startInclusive,
      Object end, boolean endInclusive, Collection keysToRemove) {
    if (start == null) {
      return new MemoryIndexStoreIterator(
          valueToEntriesMap.headMap(end, endInclusive).descendingMap(), null, keysToRemove);
    }
    return new MemoryIndexStoreIterator(
        valueToEntriesMap.subMap(start, startInclusive, end, endInclusive).descendingMap(),
        null, keysToRemove);
  }

  @Override
  public CloseableIterator<IndexStoreEntry> descendingIterator(Object start, boolean startInclusive,
      Collection keysToRemove) {
    return new MemoryIndexStoreIterator(
        valueToEntriesMap.tailMap(start, startInclusive).descendingMap(), null, keysToRemove);
  }

  @Override
  public CloseableIterator<IndexStoreEntry> descendingIterator(Collection keysToRemove) {
    return new MemoryIndexStoreIterator(valueToEntriesMap.descendingMap(), null, keysToRemove);
  }

  @Override
  public boolean isIndexOnRegionKeys() {
    return indexOnRegionKeys;
  }

  @Override
  public void setIndexOnRegionKeys(boolean indexOnRegionKeys) {
    this.indexOnRegionKeys = indexOnRegionKeys;
  }

  @Override
  public boolean isIndexOnValues() {
    return indexOnValues;
  }

  @Override
  public void setIndexOnValues(boolean indexOnValues) {
    this.indexOnValues = indexOnValues;
  }

  /**
   * Get the object of interest from the region entry. For now it always gets the deserialized
   * value.
   */
  @Override
  public Object getTargetObject(RegionEntry entry) {
    if (indexOnValues) {
      Object o = entry.getValue((LocalRegion) region);
      try {
        if (o == Token.INVALID) {
          return null;
        }
        if (o instanceof CachedDeserializable) {
          return ((CachedDeserializable) o).getDeserializedValue(region, entry);
        }
      } catch (EntryDestroyedException ignore) {
        return null;
      }
      return o;
    } else if (indexOnRegionKeys) {
      return entry.getKey();
    }
    return new CachedEntryWrapper(new NonTXEntry((LocalRegion) region, entry));
  }

  @Override
  public Object getTargetObjectInVM(RegionEntry entry) {
    if (indexOnValues) {
      Object o = entry.getValueInVM((LocalRegion) region);
      try {
        if (o == Token.INVALID) {
          return null;
        }
        if (o instanceof CachedDeserializable) {
          return ((CachedDeserializable) o).getDeserializedValue(region, entry);
        }
      } catch (EntryDestroyedException ede) {
        return null;
      }
      return o;
    } else if (indexOnRegionKeys) {
      return entry.getKey();
    }
    return new NonTXEntry((LocalRegion) region, entry);
  }

  private Object getTargetObjectForUpdate(RegionEntry entry) {
    if (indexOnValues) {
      Object o = entry.getValue((LocalRegion) region);
      try {
        if (o == Token.INVALID) {
          return Token.INVALID;
        }
        if (o instanceof CachedDeserializable) {
          return ((CachedDeserializable) o).getDeserializedValue(region, entry);
        }
      } catch (EntryDestroyedException ede) {
        return Token.INVALID;
      }
      return o;
    } else if (indexOnRegionKeys) {
      return entry.getKey();
    }
    return new NonTXEntry((LocalRegion) region, entry);
  }

  @Override
  public boolean clear() {
    valueToEntriesMap.clear();
    if (IndexManager.isObjectModificationInplace()) {
      entryToValuesMap.clear();
    }
    numIndexKeys.set(0);
    return true;
  }

  @Override
  public int size(Object key) {
    Object obj = valueToEntriesMap.get(key);
    if (obj != null) {
      return obj instanceof RegionEntry ? 1 : ((Collection) obj).size();
    } else {
      return 0;
    }
  }

  @Override
  public int size() {
    return numIndexKeys.get();
  }

  private class MemoryIndexStoreKeyIterator implements Iterator<IndexStoreEntry> {

    private final Map valuesToEntriesMap;
    private Object currKey;
    private Iterator<Map.Entry> mapIterator;

    public MemoryIndexStoreKeyIterator(Map valuesToEntriesMap) {
      this.valuesToEntriesMap = valuesToEntriesMap;
    }

    @Override
    public boolean hasNext() {
      if (mapIterator == null) {
        mapIterator = valuesToEntriesMap.entrySet().iterator();
      }
      if (mapIterator.hasNext()) {
        Map.Entry currentEntry = mapIterator.next();
        currKey = currentEntry.getKey();
        if (currKey == IndexManager.NULL || currKey == QueryService.UNDEFINED) {
          return hasNext();
        }
        return currKey != null;
      }
      return false;
    }

    @Override
    public MemoryIndexStoreKey next() {
      return new MemoryIndexStoreKey(currKey);
    }
  }

  /**
   * A bi-directional iterator over the CSL. Iterates over the entries of CSL where entry is a
   * mapping (value -> Collection) as well as over the Collection.
   */
  private class MemoryIndexStoreIterator implements CloseableIterator<IndexStoreEntry> {
    final Map map;
    Object indexKey;
    Collection keysToRemove;
    Iterator<Map.Entry> mapIterator;
    Iterator valuesIterator;
    Object currKey;
    Object currValue; // RegionEntry
    final long iteratorStartTime;
    MemoryIndexStoreEntry currentEntry;

    MemoryIndexStoreIterator(Map submap, Object indexKey, Collection keysToRemove) {
      this(submap, indexKey, keysToRemove, cache.cacheTimeMillis());
    }

    private MemoryIndexStoreIterator(Map submap, Object indexKey, Collection keysToRemove,
        long iteratorStartTime) {
      map = submap;
      this.indexKey = indexKey;
      this.keysToRemove = keysToRemove == null ? null : new HashSet(keysToRemove);
      this.iteratorStartTime = iteratorStartTime;
      currentEntry = new MemoryIndexStoreEntry(iteratorStartTime);
    }

    /**
     * This iterator iterates over the CSL map as well as on the collection of values for each entry
     * in the map. If the map has next element, check if the previous valuesIterator (iterator on
     * the Collection) has been created or has finished iterating. If not created, create an
     * iterator on the Collection. If created and not finished, return, so that the next() call
     * would create a new InMemoryIndexStorageEntry for the current key and next RE from the
     * collection of values.
     */
    @Override
    public boolean hasNext() {
      // return previous collection of values if not over
      if (valuesIterator != null && valuesIterator.hasNext()) {
        return true;
      }
      // sets the next values iterator
      if (mapIterator == null) {
        mapIterator = map.entrySet().iterator();
      }
      if (mapIterator.hasNext()) {
        // set the next entry in the map as current
        Map.Entry currentMapEntry = mapIterator.next();
        // set the index key
        currKey = currentMapEntry.getKey();
        // if the index key in currentIndexEntry is present in the
        // keysToRemove collection or is Undefined or Null
        // skip the current map entry and advance to the
        // next entry in the index map.
        // skipping null & undefined is required so that they do not get
        // into results of range queries.
        // equality and not equality for null/undefined would create
        // valuesIterator so it would not come here
        if (currKey != indexKey
            && (currKey == QueryService.UNDEFINED || currKey == IndexManager.NULL
                || keysToRemove != null && removeFromKeysToRemove(keysToRemove, currKey))) {
          return hasNext();
        }
        // iterator on the collection of values for the index key
        Object values = currentMapEntry.getValue();
        if (values instanceof Collection) {
          valuesIterator = ((Collection) values).iterator();
        } else {
          valuesIterator = null;
          currValue = values;
        }
        return values != null && (values instanceof RegionEntry || valuesIterator.hasNext());
      }

      currKey = null;
      return false;
    }

    /**
     * returns a new InMemoryIndexStorageEntry with current index key and its next value which is
     * the next RE in the collection. Make sure hasNext() has been called before calling this method
     */
    @Override
    public MemoryIndexStoreEntry next() {
      if (valuesIterator == null) {
        currentEntry.setMemoryIndexStoreEntry(currKey, (RegionEntry) currValue);
        return currentEntry;
      }

      RegionEntry re = (RegionEntry) valuesIterator.next();
      if (re == null) {
        throw new NoSuchElementException();
      }

      currentEntry.setMemoryIndexStoreEntry(currKey, re);
      return currentEntry;
    }

    @Override
    public void remove() {
      mapIterator.remove();
    }

    @Override
    public void close() {
      // do nothing
    }

    public boolean removeFromKeysToRemove(Collection keysToRemove, Object key) {
      Iterator iterator = keysToRemove.iterator();
      while (iterator.hasNext()) {
        try {
          if (TypeUtils.compare(key, iterator.next(), OQLLexerTokenTypes.TOK_EQ)
              .equals(Boolean.TRUE)) {
            iterator.remove();
            return true;
          }
        } catch (TypeMismatchException e) {
          // they are not equals, so we just continue iterating
        }
      }
      return false;
    }
  }

  @Override
  public String printAll() {
    StringBuilder sb = new StringBuilder();
    for (final Object item : valueToEntriesMap.entrySet()) {
      Entry mapEntry = (Entry) item;
      sb.append("Key: " + mapEntry.getKey());
      Object value = mapEntry.getValue();
      if (value instanceof Collection) {
        for (final Object o : (Collection) value) {
          sb.append(" Value:" + getTargetObject((RegionEntry) o));
        }
      } else {
        sb.append(" Value:" + getTargetObject((RegionEntry) value));
      }

      sb.append("\n");

    }
    return sb.toString();
  }

  class MemoryIndexStoreKey implements IndexStoreEntry {
    private final Object indexKey;

    public MemoryIndexStoreKey(Object indexKey) {
      this.indexKey = indexKey;
    }

    @Override

    public Object getDeserializedKey() {
      return indexKey;
    }

    @Override
    public Object getDeserializedValue() {
      throw new UnsupportedOperationException();
    }

    @Override
    public Object getDeserializedRegionKey() {
      throw new UnsupportedOperationException();
    }

    @Override
    public boolean isUpdateInProgress() {
      throw new UnsupportedOperationException();
    }
  }
  /**
   * A wrapper over the entry in the CSL index map. It maps IndexKey -> RegionEntry
   */
  class MemoryIndexStoreEntry implements IndexStoreEntry {
    private Object deserializedIndexKey;
    private RegionEntry regionEntry;
    private boolean updateInProgress;
    private Object value;
    private final long iteratorStartTime;

    private MemoryIndexStoreEntry(long iteratorStartTime) {
      this.iteratorStartTime = iteratorStartTime;
    }

    public void setMemoryIndexStoreEntry(Object deserializedIndexKey, RegionEntry regionEntry) {
      this.deserializedIndexKey = deserializedIndexKey;
      this.regionEntry = regionEntry;
      updateInProgress = regionEntry.isUpdateInProgress();
      value = getTargetObject(regionEntry);
    }

    @Override
    public Object getDeserializedKey() {
      return deserializedIndexKey;
    }

    @Override
    public Object getDeserializedValue() {
      return value;
    }

    @Override
    public Object getDeserializedRegionKey() {
      return regionEntry.getKey();
    }

    public RegionEntry getRegionEntry() {
      return regionEntry;
    }

    @Override
    public boolean isUpdateInProgress() {
      return updateInProgress || regionEntry.isUpdateInProgress()
      // The index update could have started just before the iterator was created. The entry still
      // needs to be re-evaluated in this case.
          || IndexManager.needsRecalculation(iteratorStartTime, regionEntry.getLastModified());
    }
  }

  public class CachedEntryWrapper {

    private final Object key;
    private final Object value;

    public CachedEntryWrapper(NonTXEntry entry) {
      if (IndexManager.testHook != null) {
        IndexManager.testHook.hook(201);
      }
      key = entry.getKey();
      value = entry.getValue();
    }

    public Object getKey() {
      return key;
    }

    public Object getValue() {
      return value;
    }

    public String toString() {
      return "CachedEntryWrapper@"
          + Integer.toHexString(System.identityHashCode(this)) + ' ' + key
          + ' ' + value;
    }

    @Override
    public boolean equals(Object obj) {
      if (!(obj instanceof CachedEntryWrapper)) {
        return false;
      }
      CachedEntryWrapper object = (CachedEntryWrapper) obj;
      if (!getKey().equals(object.getKey())) {
        if (!(getKey() instanceof PdxInstanceImpl)) {
          return false;
        }
        if (!(object.getKey() instanceof PdxInstanceImpl)) {
          return false;
        }
        PdxInstanceImpl pdxkey1 = (PdxInstanceImpl) getKey();
        PdxInstanceImpl pdxkey2 = (PdxInstanceImpl) object.getKey();
        if (!pdxkey1.equals(pdxkey2)) {
          return false;
        }
      }
      if (!getValue().equals(object.getValue())) {
        if (!(getValue() instanceof PdxInstanceImpl)) {
          return false;
        }
