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

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;

import org.apache.logging.log4j.Logger;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.query.Index;
import org.apache.geode.cache.query.IndexExistsException;
import org.apache.geode.cache.query.IndexNameConflictException;
import org.apache.geode.cache.query.IndexType;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.data.Portfolio;
import org.apache.geode.cache.query.functional.StructSetOrResultsSet;
import org.apache.geode.cache.query.internal.QueryObserverAdapter;
import org.apache.geode.cache.query.internal.QueryObserverHolder;
import org.apache.geode.cache.query.internal.index.IndexManager;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.ThreadUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.junit.categories.OQLIndexTest;


@Category({OQLIndexTest.class})
public class QueryIndexDUnitTest extends JUnit4CacheTestCase {
  public static final Logger logger = LogService.getLogger();

  public QueryIndexDUnitTest() {
    super();
  }

  @Override
  public final void postSetUp() throws Exception {
    int hostCount = Host.getHostCount();
    SerializableRunnable createRegion = new SerializableRunnable("createRegion") {
      @Override
      public void run() {
        Cache cache = getCache();
        Region region = cache.getRegion("portfolios");
        if (region == null) {
          AttributesFactory attributesFactory = new AttributesFactory();
          attributesFactory.setValueConstraint(Portfolio.class);
          attributesFactory.setDataPolicy(DataPolicy.REPLICATE);
          attributesFactory.setIndexMaintenanceSynchronous(true);
          attributesFactory.setScope(Scope.GLOBAL);
          RegionAttributes regionAttributes = attributesFactory.create();
          region = cache.createRegion("portfolios", regionAttributes);
          region.put("0", new Portfolio(0));
        }
      }
    };
    logger.info("Number of hosts = " + hostCount);
    for (int i = 0; i < hostCount; i++) {
      Host host = Host.getHost(i);
      int vmCount = host.getVMCount();
      logger.info("Host number :" + i + "contains " + vmCount + "Vms");

      for (int j = 0; j < vmCount; j++) {
        VM vm = host.getVM(j);
        vm.invoke(createRegion);
      }
    }
  }

  private static void doPut() {
    Region region = basicGetCache().getRegion("portfolios");
    if (region == null) {
      logger.info("REGION IS NULL");
    }
    try {
      region.put("1", new Portfolio(1));
      region.put("2", new Portfolio(2));
      region.put("3", new Portfolio(3));
      region.put("4", new Portfolio(4));
    } catch (Exception e) {
      Assert.fail("Caught exception while trying to do put operation", e);
    }
  }

  private static void doDestroy() {
    Region region = basicGetCache().getRegion("portfolios");
    if (region == null) {
      logger.info("REGION IS NULL");
    }
    try {
      region.destroy("2");
      region.destroy("3");
      region.destroy("4");
    } catch (Exception e) {
      Assert.fail("Caught exception while trying to do put operation", e);
    }
  }

  private static void doInPlaceUpdate(String str) {
    Region region = basicGetCache().getRegion("portfolios");
    Portfolio p = null;
    if (region == null) {
      logger.info("REGION IS NULL");
    }
    try {
      for (int i = 0; i <= 4; i++) {

        p = (Portfolio) region.get("" + i);
        p.status = "active";
        region.put("" + i, p);
      }
    } catch (Exception e) {
      Assert.fail("Caught exception while trying to do put operation", e);
    }
  }

  @Test
  public void createIndexesAndUpdatesAndThenValidate() throws InterruptedException {
    Integer[] intArr = new Integer[2];
    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);

    AsyncInvocation<Void> ai1 = null;
    AsyncInvocation<Void> ai2 = null;

    vm0.invoke(QueryIndexDUnitTest::createIndex);
    vm0.invoke(QueryIndexDUnitTest::validateIndexUsage);
    VM vm3 = host.getVM(3);


    vm0.invoke(QueryIndexDUnitTest::removeIndex);
    ai1 = vm3.invokeAsync(QueryIndexDUnitTest::doPut);
    ai2 = vm0.invokeAsync(QueryIndexDUnitTest::createIndex);
    ThreadUtils.join(ai1, 30 * 1000);
    ThreadUtils.join(ai2, 30 * 1000);
    intArr[0] = 3;
    intArr[1] = 2;
    vm0.invoke(() -> validateIndexUpdate(intArr[0], intArr[1]));


    vm0.invoke(QueryIndexDUnitTest::removeIndex);
    ai1 = vm0.invokeAsync(QueryIndexDUnitTest::doDestroy);
    ai2 = vm0.invokeAsync(QueryIndexDUnitTest::createIndex);
    ThreadUtils.join(ai1, 30 * 1000);
    ThreadUtils.join(ai2, 30 * 1000);
    intArr[0] = 1;
    intArr[1] = 1;
    vm0.invoke(() -> validateIndexUpdate(intArr[0], intArr[1]));


    vm0.invoke(QueryIndexDUnitTest::removeIndex);
    ai1 = vm0.invokeAsync(QueryIndexDUnitTest::doPut);
    ai2 = vm0.invokeAsync(QueryIndexDUnitTest::createIndex);
    ThreadUtils.join(ai1, 30 * 1000);
    ThreadUtils.join(ai2, 30 * 1000);
    intArr[0] = 3;
    intArr[1] = 2;
    vm0.invoke(() -> validateIndexUpdate(intArr[0], intArr[1]));


    vm0.invoke(QueryIndexDUnitTest::removeIndex);
    ai1 = vm3.invokeAsync(QueryIndexDUnitTest::doDestroy);
    ai2 = vm0.invokeAsync(QueryIndexDUnitTest::createIndex);
    ThreadUtils.join(ai1, 30 * 1000);
    ThreadUtils.join(ai2, 30 * 1000);
    intArr[0] = 1;
    intArr[1] = 1;
    vm0.invoke(() -> validateIndexUpdate(intArr[0], intArr[1]));

    // Test for in-place update.
    vm0.invoke(QueryIndexDUnitTest::removeIndex);
    ai1 = vm0.invokeAsync(QueryIndexDUnitTest::doPut);
    ai2 = vm0.invokeAsync(QueryIndexDUnitTest::createIndex);
    ThreadUtils.join(ai1, 30 * 1000);
    ThreadUtils.join(ai2, 30 * 1000);
    intArr[0] = 3;
    intArr[1] = 2;
    vm0.invoke(() -> validateIndexUpdate(intArr[0], intArr[1]));

    try {
      Thread.sleep(2000);
    } catch (Exception ignored) {
    }
    // Do an in-place update of the region entries.
    // This will set the Portfolio objects status to "active".
    String str = "To get Update in synch thread."; // Else test was exiting before the validation
                                                   // could finish.
    vm0.invoke(() -> doInPlaceUpdate(str));
    intArr[0] = 5;
    intArr[1] = 0;
    vm0.invoke(() -> validateIndexUpdate(intArr[0], intArr[1]));
  }

  @Test
  public void createIndexOnOverflowRegionsAndExecuteQueries() throws InterruptedException {
    Host host = Host.getHost(0);
    VM[] vms = new VM[] {host.getVM(0), host.getVM(1), host.getVM(2), host.getVM(3),};

    // Create and load regions on all vms.
    for (int i = 0; i < vms.length; i++) {
      int finalI = i;
      vms[i].invoke(() -> QueryIndexDUnitTest.createAndLoadOverFlowRegions("testOf" + "vm" + finalI,
          Boolean.TRUE, Boolean.TRUE));
    }

    // Create index on the regions.
    for (final VM item : vms) {
      item.invoke(QueryIndexDUnitTest::createIndexOnOverFlowRegions);
    }

    // execute query.
    for (final VM value : vms) {
      value.invoke(QueryIndexDUnitTest::executeQueriesUsingIndexOnOverflowRegions);
    }

    // reload the regions after index creation.
    for (int i = 0; i < vms.length; i++) {
      int finalI = i;
      vms[i].invoke(() -> QueryIndexDUnitTest.createAndLoadOverFlowRegions("testOf" + "vm" + finalI,
          Boolean.FALSE, Boolean.TRUE));
    }

    // reexecute the query.
    for (final VM vm : vms) {
      vm.invoke(QueryIndexDUnitTest::executeQueriesUsingIndexOnOverflowRegions);
    }
  }

  @Test
  public void createIndexOnOverflowRegionsAndValidateResults() throws Exception {
    Host host = Host.getHost(0);
    VM[] vms = new VM[] {host.getVM(0), host.getVM(1),};

    // Create and load regions on all vms.
    for (int i = 0; i < vms.length; i++) {
      int finalI = i;
      vms[i].invoke(() -> QueryIndexDUnitTest.createAndLoadOverFlowRegions(
          "testOfValid" + "vm" + finalI, Boolean.TRUE, Boolean.FALSE));
    }

    vms[0].invoke(new CacheSerializableRunnable("Execute query validate results") {
      @Override
      public void run2() throws CacheException {
        Cache cache = basicGetCache();
        String[] regionNames = new String[] {"replicateOverFlowRegion",
            "replicatePersistentOverFlowRegion", "prOverFlowRegion", "prPersistentOverFlowRegion",};

        QueryService qs = cache.getQueryService();
        Region region = null;

        int numObjects = 10;
        // Update the region and re-execute the query.
        // The index should get updated accordingly.
        for (final String value : regionNames) {
          region = cache.getRegion(value);
          for (int cnt = 1; cnt < numObjects; cnt++) {
            region.put(new Portfolio(cnt), new Portfolio(cnt));
          }
        }

        String[] qString =
            new String[] {"SELECT * FROM " + SEPARATOR + "REGION_NAME pf WHERE pf.ID = 1",
                "SELECT ID FROM " + SEPARATOR + "REGION_NAME pf WHERE pf.ID = 1",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME pf WHERE pf.ID > 5",
                "SELECT ID FROM " + SEPARATOR + "REGION_NAME pf WHERE pf.ID > 5",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID = 1",
                "SELECT ID FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID = 1",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID > 5",
                "SELECT ID FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID > 5",};

        // Execute Query without index.
        SelectResults[] srWithoutIndex = new SelectResults[qString.length * regionNames.length];
        String[] queryString = new String[qString.length * regionNames.length];

        int r = 0;
        try {
          for (final String s : qString) {
            for (final String regionName : regionNames) {
              String queryStr = s.replace("REGION_NAME", regionName);
              Query query = qs.newQuery(queryStr);
              queryString[r] = queryStr;
              srWithoutIndex[r] = (SelectResults) query.execute();
              r++;
            }
          }
        } catch (Exception ex) {
          LogWriterUtils.getLogWriter().info("Failed to Execute query", ex);
          fail("Failed to Execute query.");
        }

        // Create index.
        try {
          for (final String regionName : regionNames) {
            region = cache.getRegion(regionName);
            String indexName = "idIndex" + regionName;
            cache.getLogger()
                .fine("createIndexOnOverFlowRegions() checking for index: " + indexName);
            try {
              if (qs.getIndex(region, indexName) == null) {
                cache.getLogger()
                    .fine("createIndexOnOverFlowRegions() Index doesn't exist, creating index: "
                        + indexName);
                Index i1 = qs.createIndex(indexName, "pf.ID", SEPARATOR + regionName + " pf");
              }
              indexName = "keyIdIndex" + regionName;
              if (qs.getIndex(region, indexName) == null) {
                cache.getLogger()
                    .fine("createIndexOnOverFlowRegions() Index doesn't exist, creating index: "
                        + indexName);
                Index i2 =
                    qs.createIndex(indexName, "key.ID", SEPARATOR + regionName + ".keys key");
              }
            } catch (IndexNameConflictException ice) {
              // Ignore. The pr may have created the index through
              // remote index create message from peer.
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to create index", ex);
          fail("Failed to create index.");
        }

        // Execute Query with index.
        SelectResults[] srWithIndex = new SelectResults[qString.length * regionNames.length];
        try {
          r = 0;
          for (final String s : qString) {
            for (final String regionName : regionNames) {
              QueryObserverImpl observer = new QueryObserverImpl();
              QueryObserverHolder.setInstance(observer);
              String queryStr = s.replace("REGION_NAME", regionName);
              Query query = qs.newQuery(queryStr);
              srWithIndex[r++] = (SelectResults) query.execute();
              if (!observer.isIndexesUsed) {
                fail("Index not used for query. " + queryStr);
              }
            }
          }
        } catch (Exception ex) {
          LogWriterUtils.getLogWriter().info("Failed to Execute query", ex);
          fail("Failed to Execute query.");
        }

        // Compare results with and without index.
        StructSetOrResultsSet ssORrs = new StructSetOrResultsSet();
        SelectResults[][] sr = new SelectResults[1][2];

        for (int i = 0; i < srWithIndex.length; i++) {
          sr[0][0] = srWithoutIndex[i];
          sr[0][1] = srWithIndex[i];
          logger.info("Comparing the result for the query : " + queryString[i]
              + " Index in ResultSet is: " + i);
          ssORrs.CompareQueryResultsWithoutAndWithIndexes(sr, 1, queryString);
        }

        // Update the region and re-execute the query.
        // The index should get updated accordingly.
        for (final String name : regionNames) {
          region = cache.getRegion(name);
          for (int cnt = 1; cnt < numObjects; cnt++) {
            if (cnt % 2 == 0) {
              region.destroy(new Portfolio(cnt));
            }
          }
          for (int cnt = 10; cnt < numObjects; cnt++) {
            if (cnt % 2 == 0) {
              region.put(new Portfolio(cnt), new Portfolio(cnt));
            }
          }
        }

        // Execute Query with index.
        srWithIndex = new SelectResults[qString.length * regionNames.length];
        try {
          r = 0;
          for (final String s : qString) {
            for (final String regionName : regionNames) {
              QueryObserverImpl observer = new QueryObserverImpl();
              QueryObserverHolder.setInstance(observer);
              String queryStr = s.replace("REGION_NAME", regionName);
              Query query = qs.newQuery(queryStr);
              srWithIndex[r++] = (SelectResults) query.execute();
              if (!observer.isIndexesUsed) {
                fail("Index not used for query. " + queryStr);
              }
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to Execute query", ex);
          fail("Failed to Execute query.");
        }

      }
    });
  }

  @Test
  public void createIndexOnOverflowRegionsWithMultipleWhereClauseAndValidateResults()
      throws Exception {
    Host host = Host.getHost(0);
    VM[] vms = new VM[] {host.getVM(0), host.getVM(1),};

    // Create and load regions on all vms.
    for (int i = 0; i < vms.length; i++) {
      int finalI = i;
      vms[i].invoke(() -> QueryIndexDUnitTest.createAndLoadOverFlowRegions(
          "testOfValid2" + "vm" + finalI, Boolean.TRUE, Boolean.FALSE));
    }

    vms[0].invoke(new CacheSerializableRunnable("Execute query validate results") {
      @Override
      public void run2() throws CacheException {
        Cache cache = basicGetCache();
        String[] regionNames = new String[] {"replicateOverFlowRegion",
            "replicatePersistentOverFlowRegion", "prOverFlowRegion", "prPersistentOverFlowRegion",};

        QueryService qs = cache.getQueryService();
        Region region = null;

        int numObjects = 10;
        // Update the region and re-execute the query.
        // The index should get updated accordingly.
        for (final String value : regionNames) {
          region = cache.getRegion(value);
          for (int cnt = 1; cnt < numObjects; cnt++) {
            region.put(new Portfolio(cnt), "XX" + cnt);
          }
        }

        String[] qString =
            new String[] {"SELECT * FROM " + SEPARATOR + "REGION_NAME pf WHERE pf = 'XX1'",
                "SELECT * FROM " + SEPARATOR
                    + "REGION_NAME pf WHERE pf IN SET( 'XX5', 'XX6', 'XX7')",
                "SELECT * FROM " + SEPARATOR
                    + "REGION_NAME.values pf WHERE pf IN SET( 'XX5', 'XX6', 'XX7')",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME.keys k WHERE k.ID = 1",
                "SELECT key.ID FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID = 1",
                "SELECT ID, status FROM " + SEPARATOR + "REGION_NAME.keys WHERE ID = 1",
                "SELECT k.ID, k.status FROM " + SEPARATOR
                    + "REGION_NAME.keys k WHERE k.ID = 1 and k.status = 'active'",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID > 5",
                "SELECT key.ID FROM " + SEPARATOR
                    + "REGION_NAME.keys key WHERE key.ID > 5 and key.status = 'active'",};

        // Execute Query without index.
        SelectResults[] srWithoutIndex = new SelectResults[qString.length * regionNames.length];
        String[] queryString = new String[qString.length * regionNames.length];

        int r = 0;
        try {
          for (final String s : qString) {
            for (final String regionName : regionNames) {
              String queryStr = s.replace("REGION_NAME", regionName);
              Query query = qs.newQuery(queryStr);
              queryString[r] = queryStr;
              srWithoutIndex[r] = (SelectResults) query.execute();
              r++;
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to Execute query", ex);
          fail("Failed to Execute query.");
        }

        // Create index.
        String indexName = "";
        try {
          for (final String regionName : regionNames) {
            region = cache.getRegion(regionName);
            indexName = "idIndex" + regionName;
            cache.getLogger()
                .fine("createIndexOnOverFlowRegions() checking for index: " + indexName);
            try {
              if (qs.getIndex(region, indexName) == null) {
                cache.getLogger()
                    .fine("createIndexOnOverFlowRegions() Index doesn't exist, creating index: "
                        + indexName);
                Index i1 = qs.createIndex(indexName, "pf", SEPARATOR + regionName + " pf");
              }
              indexName = "valueIndex" + regionName;
              if (qs.getIndex(region, indexName) == null) {
                cache.getLogger()
                    .fine("createIndexOnOverFlowRegions() Index doesn't exist, creating index: "
                        + indexName);
                Index i1 =
                    qs.createIndex(indexName, "pf", SEPARATOR + regionName + ".values pf");
              }
              indexName = "keyIdIndex" + regionName;
              if (qs.getIndex(region, indexName) == null) {
                cache.getLogger()
                    .fine("createIndexOnOverFlowRegions() Index doesn't exist, creating index: "
                        + indexName);
                Index i2 =
                    qs.createIndex(indexName, "key.ID", SEPARATOR + regionName + ".keys key");
              }
              indexName = "keyIdIndex2" + regionName;


            } catch (IndexNameConflictException ice) {
              // Ignore. The pr may have created the index through
              // remote index create message from peer.
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to create index", ex);
          fail("Failed to create index." + indexName);
        }

        // Execute Query with index.
        SelectResults[] srWithIndex = new SelectResults[qString.length * regionNames.length];
        try {
          r = 0;
          for (final String s : qString) {
            for (final String regionName : regionNames) {
              QueryObserverImpl observer = new QueryObserverImpl();
              QueryObserverHolder.setInstance(observer);
              String queryStr = s.replace("REGION_NAME", regionName);
              Query query = qs.newQuery(queryStr);
              srWithIndex[r++] = (SelectResults) query.execute();
              if (!observer.isIndexesUsed) {
                fail("Index not used for query. " + queryStr);
              }
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to Execute query", ex);
          fail("Failed to Execute query.");
        }

        // Compare results with and without index.
        StructSetOrResultsSet ssORrs = new StructSetOrResultsSet();
        SelectResults[][] sr = new SelectResults[1][2];

        for (int i = 0; i < srWithIndex.length; i++) {
          sr[0][0] = srWithoutIndex[i];
          sr[0][1] = srWithIndex[i];
          logger.info("Comparing the result for the query : " + queryString[i]
              + " Index in ResultSet is: " + i);
          ssORrs.CompareQueryResultsWithoutAndWithIndexes(sr, 1, queryString);
        }

        // Update the region and re-execute the query.
        // The index should get updated accordingly.
        for (final String name : regionNames) {
          region = cache.getRegion(name);
          for (int cnt = 1; cnt < numObjects; cnt++) {
            if (cnt % 2 == 0) {
              region.destroy(new Portfolio(cnt));
            }
          }
          for (int cnt = 10; cnt < numObjects; cnt++) {
            if (cnt % 2 == 0) {
              region.put(new Portfolio(cnt), "XX" + cnt);
            }
          }
        }

        // Execute Query with index.
        srWithIndex = new SelectResults[qString.length * regionNames.length];
        try {
          r = 0;
          for (final String s : qString) {
            for (final String regionName : regionNames) {
              QueryObserverImpl observer = new QueryObserverImpl();
              QueryObserverHolder.setInstance(observer);
              String queryStr = s.replace("REGION_NAME", regionName);
              Query query = qs.newQuery(queryStr);
              srWithIndex[r++] = (SelectResults) query.execute();
              if (!observer.isIndexesUsed) {
                fail("Index not used for query. " + queryStr);
              }
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to Execute query", ex);
          fail("Failed to Execute query.");
        }

      }
    });
  }

  @Test
  public void createIndexAndUpdatesOnOverflowRegionsAndValidateResults() throws Exception {
    Host host = Host.getHost(0);
    VM[] vms = new VM[] {host.getVM(0), host.getVM(1),};

    // Create and load regions on all vms.
    for (int i = 0; i < vms.length; i++) {
      int finalI = i;
      vms[i].invoke(() -> QueryIndexDUnitTest.createAndLoadOverFlowRegions(
          "testOfValid3" + "vm" + finalI, Boolean.TRUE, Boolean.TRUE));
    }

    vms[0].invoke(new CacheSerializableRunnable("Execute query validate results") {
      @Override
      public void run2() throws CacheException {
        Cache cache = basicGetCache();
        String[] regionNames = new String[] {"replicateOverFlowRegion",
            "replicatePersistentOverFlowRegion", "prOverFlowRegion", "prPersistentOverFlowRegion",};

        QueryService qs = cache.getQueryService();
        Region region = null;


        String[] qString =
            new String[] {"SELECT * FROM " + SEPARATOR + "REGION_NAME pf WHERE pf.ID = 1",
                "SELECT ID FROM " + SEPARATOR + "REGION_NAME pf WHERE pf.ID = 1",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME pf WHERE pf.ID > 5",
                "SELECT ID FROM " + SEPARATOR + "REGION_NAME pf WHERE pf.ID > 5",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID = 1",
                "SELECT ID FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID = 1",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID > 5",
                "SELECT ID FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID > 5",};

        // Execute Query without index.
        SelectResults[] srWithoutIndex = new SelectResults[qString.length * regionNames.length];
        String[] queryString = new String[qString.length * regionNames.length];

        int r = 0;
        try {
          for (final String s : qString) {
            for (final String regionName : regionNames) {
              String queryStr = s.replace("REGION_NAME", regionName);
              Query query = qs.newQuery(queryStr);
              queryString[r] = queryStr;
              srWithoutIndex[r] = (SelectResults) query.execute();
              r++;
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to Execute query", ex);
          fail("Failed to Execute query.");
        }

        // Create index.
        try {
          for (final String regionName : regionNames) {
            region = cache.getRegion(regionName);
            String indexName = "idIndex" + regionName;
            cache.getLogger()
                .fine("createIndexOnOverFlowRegions() checking for index: " + indexName);
            try {
              if (qs.getIndex(region, indexName) == null) {
                cache.getLogger()
                    .fine("createIndexOnOverFlowRegions() Index doesn't exist, creating index: "
                        + indexName);
                Index i1 = qs.createIndex(indexName, "pf.ID", SEPARATOR + regionName + " pf");
              }
              indexName = "keyIdIndex" + regionName;
              if (qs.getIndex(region, indexName) == null) {
                cache.getLogger()
                    .fine("createIndexOnOverFlowRegions() Index doesn't exist, creating index: "
                        + indexName);
                Index i2 =
                    qs.createIndex(indexName, "key.ID", SEPARATOR + regionName + ".keys key");
              }
            } catch (IndexNameConflictException ice) {
              // Ignore. The pr may have created the index through
              // remote index create message from peer.
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to create index", ex);
          fail("Failed to create index.");
        }

        int numObjects = 50;
        // Update the region and re-execute the query.
        // The index should get updated accordingly.

        for (final String value : regionNames) {
          region = cache.getRegion(value);
          for (int cnt = 0; cnt < numObjects; cnt++) {
            region.put(new Portfolio(cnt), new Portfolio(cnt));
          }
        }

        for (final String name : regionNames) {
          region = cache.getRegion(name);
          String indexName = "idIndex" + name;
          Index i1, i2;
          if ((i1 = qs.getIndex(region, indexName)) != null) {
            assertEquals("Unexpected number of keys in the index ", numObjects,
                i1.getStatistics().getNumberOfKeys());
            assertEquals("Unexpected number of values in the index ", numObjects,
                i1.getStatistics().getNumberOfValues());
          }
          indexName = "keyIdIndex" + name;
          if ((i2 = qs.getIndex(region, indexName)) != null) {
            assertEquals("Unexpected number of keys in the index ", numObjects,
                i2.getStatistics().getNumberOfKeys());
            assertEquals("Unexpected number of values in the index ", numObjects,
                i2.getStatistics().getNumberOfValues());
          }
        }

        // Execute Query with index.
        SelectResults[] srWithIndex = new SelectResults[qString.length * regionNames.length];
        try {
          r = 0;
          for (final String s : qString) {
            for (final String regionName : regionNames) {
              QueryObserverImpl observer = new QueryObserverImpl();
              QueryObserverHolder.setInstance(observer);
              String queryStr = s.replace("REGION_NAME", regionName);
              Query query = qs.newQuery(queryStr);
              srWithIndex[r++] = (SelectResults) query.execute();
              if (!observer.isIndexesUsed) {
                fail("Index not used for query. " + queryStr);
              }
            }
          }
        } catch (Exception ex) {
          logger.info("Failed to Execute query", ex);
          fail("Failed to Execute query.");
        }

        // Compare results with and without index.
        StructSetOrResultsSet ssORrs = new StructSetOrResultsSet();
        SelectResults[][] sr = new SelectResults[1][2];

        for (int i = 0; i < srWithIndex.length; i++) {
          sr[0][0] = srWithoutIndex[i];
          sr[0][1] = srWithIndex[i];
          logger.info("Comparing the result for the query : " + queryString[i]
              + " Index in ResultSet is: " + i);
          ssORrs.CompareQueryResultsWithoutAndWithIndexes(sr, 1, queryString);
        }

      }
    });
  }

  @Test
  public void createIndexOnOverflowRegionsAndValidateResultsUsingParams() throws Exception {
    Host host = Host.getHost(0);
    VM[] vms = new VM[] {host.getVM(0), host.getVM(1),};

    // Create and load regions on all vms.
    for (int i = 0; i < vms.length; i++) {
      int finalI = i;
      vms[i].invoke(() -> QueryIndexDUnitTest.createAndLoadOverFlowRegions(
          "testOfValidUseParams" + "vm" + finalI, Boolean.TRUE, Boolean.FALSE));
    }

    vms[0].invoke(new CacheSerializableRunnable("Execute query validate results") {
      @Override
      public void run2() throws CacheException {
        Cache cache = basicGetCache();
        String[] regionNames = new String[] {"replicateOverFlowRegion",
            "replicatePersistentOverFlowRegion", "prOverFlowRegion", "prPersistentOverFlowRegion",};

        QueryService qs = cache.getQueryService();
        Region region = null;

        int numObjects = 10;
        // Update the region and re-execute the query.
        // The index should get updated accordingly.
        for (final String value : regionNames) {
          region = cache.getRegion(value);
          for (int cnt = 1; cnt < numObjects; cnt++) {
            region.put(new Portfolio(cnt), cnt + 100);
          }
        }

        String[] qString =
            new String[] {"SELECT * FROM " + SEPARATOR + "REGION_NAME pf WHERE pf = $1",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME pf WHERE pf > $1",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME.values pf WHERE pf < $1",
                "SELECT * FROM " + SEPARATOR + "REGION_NAME.keys k WHERE k.ID = $1",
                "SELECT key.ID FROM " + SEPARATOR + "REGION_NAME.keys key WHERE key.ID = $1",
                "SELECT ID, status FROM " + SEPARATOR + "REGION_NAME.keys WHERE ID = $1",
