/*
 * 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.assertFalse;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

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.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.query.Index;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.Struct;
import org.apache.geode.cache.query.data.Portfolio;
import org.apache.geode.cache.query.data.PortfolioPdx;
import org.apache.geode.cache.query.data.PositionPdx;
import org.apache.geode.cache.query.internal.index.CompactRangeIndex;
import org.apache.geode.cache.query.internal.index.IndexManager;
import org.apache.geode.cache.query.internal.index.IndexStore.IndexStoreEntry;
import org.apache.geode.cache.query.internal.index.PartitionedIndex;
import org.apache.geode.cache.query.internal.index.RangeIndex;
import org.apache.geode.cache.query.types.CollectionType;
import org.apache.geode.cache.query.types.ObjectType;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.cache30.ClientServerTestCase;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.persistence.query.CloseableIterator;
import org.apache.geode.pdx.internal.PdxString;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.junit.categories.OQLQueryTest;

@Category({OQLQueryTest.class})
public class PdxStringQueryDUnitTest extends JUnit4CacheTestCase {
  final String rootRegionName = "root";
  final String regionName = "PdxTest";
  final String regName = SEPARATOR + rootRegionName + SEPARATOR + regionName;
  private static int bridgeServerPort;

  private static final int orderByQueryIndex = 11;
  private static final int[] groupByQueryIndex = new int[] {7, 8, 9, 10};

  private final String[] queryString = new String[] {
      "SELECT pos.secId FROM " + regName + " p, p.positions.values pos WHERE pos.secId LIKE '%L'",
      // 0
      "SELECT pos.secId FROM " + regName + " p, p.positions.values pos where pos.secId = 'IBM'",
      // 1
      "SELECT pos.secId, p.status FROM " + regName
          + " p, p.positions.values pos where pos.secId > 'APPL'", // 2
      "SELECT pos.secId FROM " + regName
          + " p, p.positions.values pos WHERE pos.secId > 'APPL' and pos.secId < 'SUN'", // 3
      "select pos.secId from " + regName
          + " p, p.positions.values pos where pos.secId  IN SET ('YHOO', 'VMW')", // 4
      "select pos.secId from " + regName
          + " p, p.positions.values pos where NOT (pos.secId = 'VMW')", // 5
      "select pos.secId from " + regName
          + " p, p.positions.values pos where NOT (pos.secId IN SET('SUN', 'ORCL')) ", // 6
      "select pos.secId , count(pos.id) from " + regName
          + " p, p.positions.values pos where  pos.secId > 'APPL' group by pos.secId ", // 7
      "select pos.secId , sum(pos.id) from " + regName
          + " p, p.positions.values pos where  pos.secId > 'APPL' group by pos.secId ", // 8,
      "select pos.secId , count(distinct pos.secId) from " + regName
          + " p, p.positions.values pos where  pos.secId > 'APPL' group by pos.secId ", // 9
      "select  count(distinct pos.secId) from " + regName
          + " p, p.positions.values pos where  pos.secId > 'APPL' ", // 10
      "SELECT distinct pos.secId FROM " + regName + " p, p.positions.values pos order by pos.secId",
      // 11
      "SELECT distinct pos.secId FROM " + regName
          + " p, p.positions.values pos WHERE p.ID > 1 order by pos.secId limit 5",// 12
  };

  private final String[] queryString2 = new String[] {
      "SELECT pos.secIdIndexed FROM " + regName
          + " p, p.positions.values pos WHERE pos.secIdIndexed LIKE '%L'", // 0
      "SELECT pos.secIdIndexed FROM " + regName
          + " p, p.positions.values pos where pos.secIdIndexed = 'IBM'", // 1
      "SELECT pos.secIdIndexed, p.status FROM " + regName
          + " p, p.positions.values pos where pos.secIdIndexed > 'APPL'", // 2
      "SELECT pos.secIdIndexed FROM " + regName
          + " p, p.positions.values pos WHERE pos.secIdIndexed > 'APPL' and pos.secIdIndexed < 'SUN'",
      // 3
      "select pos.secIdIndexed from " + regName
          + " p, p.positions.values pos where pos.secIdIndexed  IN SET ('YHOO', 'VMW')", // 4
      "select pos.secIdIndexed from " + regName
          + " p, p.positions.values pos where NOT (pos.secIdIndexed = 'VMW')", // 5
      "select pos.secIdIndexed from " + regName
          + " p, p.positions.values pos where NOT (pos.secIdIndexed IN SET('SUN', 'ORCL')) ", // 6
      "select pos.secIdIndexed , count(pos.id) from " + regName
          + " p, p.positions.values pos where  pos.secIdIndexed > 'APPL' group by pos.secIdIndexed ",
      // 7
      "select pos.secIdIndexed , sum(pos.id) from " + regName
          + " p, p.positions.values pos where  pos.secIdIndexed > 'APPL' group by pos.secIdIndexed ",
      // 8
      "select pos.secIdIndexed , count(distinct pos.secIdIndexed) from " + regName
          + " p, p.positions.values pos where  pos.secIdIndexed > 'APPL' group by pos.secIdIndexed ",
      // 9
      "select  count(distinct pos.secIdIndexed) from " + regName
          + " p, p.positions.values pos where  pos.secIdIndexed > 'APPL'  ", // 10
      "SELECT distinct pos.secIdIndexed FROM " + regName
          + " p, p.positions.values pos order by pos.secIdIndexed", // 11
      "SELECT distinct pos.secIdIndexed FROM " + regName
          + " p, p.positions.values pos WHERE p.ID > 1 order by pos.secIdIndexed limit 5",// 12
  };

  @Test
  public void testReplicatedRegionNoIndex() throws CacheException {
    final Host host = Host.getHost(0);
    VM server0 = host.getVM(0);
    VM server1 = host.getVM(1);
    VM server2 = host.getVM(2);
    VM client = host.getVM(3);
    final int numberOfEntries = 10;

    // Start server1 and create index
    server0.invoke(new CacheSerializableRunnable("Create Server1") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        // create a local query service
        QueryService localQueryService = null;
        try {
          localQueryService = getCache().getQueryService();
        } catch (Exception e) {
          Assert.fail("Failed to get QueryService.", e);
        }
        Index index = null;
        // create an index on statusIndexed is created
        try {
          index = localQueryService.createIndex("secIdIndex2", "pos.secIdIndexed",
              regName + " p, p.positions.values pos");
          if (!(index instanceof RangeIndex)) {
            fail("Range Index should have been created instead of " + index.getClass());
          }
        } catch (Exception ex) {
          fail("Failed to create index." + ex.getMessage());
        }
      }
    });

    // Start server2
    server1.invoke(new CacheSerializableRunnable("Create Server2") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        Region region = getRootRegion().getSubregion(regionName);
      }
    });

    // Start server3
    server2.invoke(new CacheSerializableRunnable("Create Server3") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        Region region = getRootRegion().getSubregion(regionName);
      }
    });

    // Client pool.
    final int port0 = server0.invoke(PdxStringQueryDUnitTest::getCacheServerPort);
    final int port1 = server1.invoke(PdxStringQueryDUnitTest::getCacheServerPort);
    final int port2 = server2.invoke(PdxStringQueryDUnitTest::getCacheServerPort);

    final String host0 = NetworkUtils.getServerHostName(server0.getHost());

    // Create client pool.
    final String poolName = "testClientServerQueryPool";
    createPool(client, poolName, new String[] {host0}, new int[] {port0, port1, port2}, true);

    // Create client region and put PortfolioPdx objects (PdxInstances)
    client.invoke(new CacheSerializableRunnable("Create client") {
      @Override
      public void run2() throws CacheException {
        AttributesFactory factory = new AttributesFactory();
        factory.setScope(Scope.LOCAL);
        ClientServerTestCase.configureConnectionPool(factory, host0, port1, -1, true, -1, -1, null);
        Region region = createRegion(regionName, rootRegionName, factory.create());
        LogWriterUtils.getLogWriter().info("Put PortfolioPdx");
        for (int i = 0; i < numberOfEntries; i++) {
          region.put("key-" + i, new PortfolioPdx(i));
        }
      }
    });

    // Execute queries from client to server and locally on client
    SerializableRunnable executeQueries = new CacheSerializableRunnable("Execute queries") {
      @Override
      public void run2() throws CacheException {
        QueryService remoteQueryService = null;
        QueryService localQueryService = null;
        SelectResults[][] rs = new SelectResults[1][2];
        SelectResults[] resWithoutIndexRemote = new SelectResults[queryString.length];
        SelectResults[] resWithIndexRemote = new SelectResults[queryString2.length];
        SelectResults[] resWithoutIndexLocal = new SelectResults[queryString.length];
        SelectResults[] resWithIndexLocal = new SelectResults[queryString2.length];

        try {
          remoteQueryService = (PoolManager.find(poolName)).getQueryService();
          localQueryService = getCache().getQueryService();
        } catch (Exception e) {
          Assert.fail("Failed to get QueryService.", e);
        }

        for (int i = 0; i < queryString.length; i++) {
          try {
            LogWriterUtils.getLogWriter()
                .info("### Executing Query on remote server:" + queryString[i]);
            Query query = remoteQueryService.newQuery(queryString[i]);
            rs[0][0] = (SelectResults) query.execute();
            resWithoutIndexRemote[i] = rs[0][0];
            LogWriterUtils.getLogWriter().info("RR remote indexType: no index  size of resultset: "
                + rs[0][0].size() + " for query: " + queryString[i]);
            checkForPdxString(rs[0][0].asList(), queryString[i]);

            LogWriterUtils.getLogWriter()
                .info("### Executing Query locally on client:" + queryString[i]);
            query = localQueryService.newQuery(queryString[i]);
            rs[0][1] = (SelectResults) query.execute();
            resWithoutIndexLocal[i] = rs[0][1];
            LogWriterUtils.getLogWriter()
                .info("RR  client local indexType:no index size of resultset: " + rs[0][1].size()
                    + " for query: " + queryString[i]);
            checkForPdxString(rs[0][1].asList(), queryString[i]);
          } catch (Exception e) {
            Assert.fail("Failed executing " + queryString[i], e);
          }
          try {
            // to compare remote query results with and without index
            LogWriterUtils.getLogWriter()
                .info("### Executing Query on remote server for region2:" + queryString2[i]);
            Query query = remoteQueryService.newQuery(queryString2[i]);
            resWithIndexRemote[i] = (SelectResults) query.execute();
            LogWriterUtils.getLogWriter().info("RR  remote region2 size of resultset: "
                + resWithIndexRemote[i].size() + " for query: " + queryString2[i]);
            checkForPdxString(resWithIndexRemote[i].asList(), queryString2[i]);

            // to compare local query results with and without index
            LogWriterUtils.getLogWriter()
                .info("### Executing Query on local for region2:" + queryString2[i]);
            query = localQueryService.newQuery(queryString2[i]);
            resWithIndexLocal[i] = (SelectResults) query.execute();
            LogWriterUtils.getLogWriter().info("RR  local region2 size of resultset: "
                + resWithIndexLocal[i].size() + " for query: " + queryString2[i]);
            checkForPdxString(resWithIndexLocal[i].asList(), queryString2[i]);
          } catch (Exception e) {
            Assert.fail("Failed executing " + queryString2[i], e);
          }

          if (i < orderByQueryIndex) {
            // Compare local and remote query results.
            if (!compareResultsOfWithAndWithoutIndex(rs)) {
              fail("Local and Remote Query Results are not matching for query :" + queryString[i]);
            }
          } else {
            // compare the order of results returned
            compareResultsOrder(rs, false);
          }

        }
        // compare remote query results with and without index
        for (int i = 0; i < queryString.length; i++) {
          rs[0][0] = resWithoutIndexRemote[i];
          rs[0][1] = resWithIndexRemote[i];
          if (i < orderByQueryIndex) {
            if (!compareResultsOfWithAndWithoutIndex(rs)) {
              fail("Results with and without index are not matching for query :" + queryString2[i]);
            }
          } else {
            // compare the order of results returned
            compareResultsOrder(rs, false);
          }
        }
        // compare local query results with and without index
        for (int i = 0; i < queryString.length; i++) {
          rs[0][0] = resWithoutIndexLocal[i];
          rs[0][1] = resWithIndexLocal[i];
          if (i < orderByQueryIndex) {
            if (!compareResultsOfWithAndWithoutIndex(rs)) {
              fail("Results with and without index are not matching for query :" + queryString2[i]);
            }
          } else {
            // compare the order of results returned
            compareResultsOrder(rs, false);
          }
        }

      }
    };

    client.invoke(executeQueries);

    // Put Non Pdx objects on server execute queries locally
    server0.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion().getSubregion(regionName);

        LogWriterUtils.getLogWriter().info("Put Objects locally on server");
        for (int i = numberOfEntries; i < numberOfEntries * 2; i++) {
          region.put("key-" + i, new Portfolio(i));
        }
        QueryService localQueryService = getCache().getQueryService();

        // Query server1 locally to check if PdxString is not being returned
        for (int i = 0; i < queryString.length; i++) {
          try {
            LogWriterUtils.getLogWriter()
                .info("### Executing Query locally on server:" + queryString[i]);
            SelectResults rs = (SelectResults) localQueryService.newQuery(queryString[i]).execute();
            LogWriterUtils.getLogWriter().info("RR server local indexType: no  size of resultset: "
                + rs.size() + " for query: " + queryString[i]);
            // The results should not be PdxString
            checkForPdxString(rs.asList(), queryString[i]);

          } catch (Exception e) {
            Assert.fail("Failed executing " + queryString[i], e);
          }
          try {
            SelectResults rs =
                (SelectResults) localQueryService.newQuery(queryString2[i]).execute();
            LogWriterUtils.getLogWriter().info("RR server local indexType: no size of resultset: "
                + rs.size() + " for query: " + queryString2[i]);
            // The results should not be PdxString
            checkForPdxString(rs.asList(), queryString2[i]);
          } catch (Exception e) {
            Assert.fail("Failed executing " + queryString2[i], e);
          }
        }
      }
    });

    // test for readSerialized flag
    server0.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
        cache.setReadSerializedForTest(true);
        QueryService localQueryService = getCache().getQueryService();

        // Query server1 locally to check if PdxString is not being returned
        for (final String s : queryString) {
          try {
            LogWriterUtils.getLogWriter()
                .info("### Executing Query locally on server:" + s);
            SelectResults rs = (SelectResults) localQueryService.newQuery(s).execute();
            LogWriterUtils.getLogWriter().info(
                "isPR: false server local readSerializedTrue: indexType: false size of resultset: "
                    + rs.size() + " for query: " + s);
            // The results should not be PdxString
            checkForPdxString(rs.asList(), s);
          } catch (Exception e) {
            Assert.fail("Failed executing " + s, e);
          }
        }
      }
    });

    // test for readSerialized flag on client
    client.invoke(new CacheSerializableRunnable("Create client") {
      @Override
      public void run2() throws CacheException {
        GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
        cache.setReadSerializedForTest(true);
        QueryService remoteQueryService = (PoolManager.find(poolName)).getQueryService();

        // Query server1 remotely to check if PdxString is not being returned
        for (final String s : queryString) {
          try {
            LogWriterUtils.getLogWriter()
                .info("### Executing Query locally on server:" + s);
            SelectResults rs =
                (SelectResults) remoteQueryService.newQuery(s).execute();
            LogWriterUtils.getLogWriter()
                .info("RR server remote readSerializedTrue: indexType: false size of resultset: "
                    + rs.size() + " for query: " + s);
            // The results should not be PdxString
            checkForPdxString(rs.asList(), s);
          } catch (Exception e) {
            Assert.fail("Failed executing " + s, e);
          }
        }
      }
    });

    closeClient(server2);
    closeClient(client);
    closeClient(server1);
    closeClient(server0);
  }

  @Test
  public void testRepliacatedRegionCompactRangeIndex() throws CacheException {
    final Host host = Host.getHost(0);
    VM server0 = host.getVM(0);
    VM server1 = host.getVM(1);
    VM server2 = host.getVM(2);
    VM client = host.getVM(3);
    final int numberOfEntries = 10;

    // Start server1 and create index
    server0.invoke(new CacheSerializableRunnable("Create Server1") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        // create a local query service
        QueryService localQueryService = null;
        try {
          localQueryService = getCache().getQueryService();
        } catch (Exception e) {
          Assert.fail("Failed to get QueryService.", e);
        }
        // Verify the type of index created
        Index index = null;
        try {
          index = localQueryService.createIndex("statusIndex", "status", regName);
          if (!(index instanceof CompactRangeIndex)) {
            fail("CompactRange Index should have been created instead of " + index.getClass());
          }
        } catch (Exception ex) {
          fail("Failed to create index." + ex.getMessage());
        }
      }
    });

    // Start server2
    server1.invoke(new CacheSerializableRunnable("Create Server2") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        Region region = getRootRegion().getSubregion(regionName);
      }
    });

    // Start server3
    server2.invoke(new CacheSerializableRunnable("Create Server3") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        Region region = getRootRegion().getSubregion(regionName);
      }
    });

    // Client pool.
    final int port0 = server0.invoke(PdxStringQueryDUnitTest::getCacheServerPort);
    final int port1 = server1.invoke(PdxStringQueryDUnitTest::getCacheServerPort);
    final int port2 = server2.invoke(PdxStringQueryDUnitTest::getCacheServerPort);

    final String host0 = NetworkUtils.getServerHostName(server0.getHost());

    // Create client pool.
    final String poolName = "testClientServerQueryPool";
    createPool(client, poolName, new String[] {host0}, new int[] {port0, port1, port2}, true);

    // Create client region and put PortfolioPdx objects (PdxInstances)
    client.invoke(new CacheSerializableRunnable("Create client") {
      @Override
      public void run2() throws CacheException {
        AttributesFactory factory = new AttributesFactory();
        factory.setScope(Scope.LOCAL);
        ClientServerTestCase.configureConnectionPool(factory, host0, port1, -1, true, -1, -1, null);
        Region region = createRegion(regionName, rootRegionName, factory.create());
        LogWriterUtils.getLogWriter().info("Put PortfolioPdx");
        for (int i = 0; i < numberOfEntries; i++) {
          region.put("key-" + i, new PortfolioPdx(i));
        }
      }
    });

    // Verify if all the index keys are PdxStrings
    server0.invoke(new CacheSerializableRunnable("Create Server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion().getSubregion(regionName);
        QueryService localQueryService = getCache().getQueryService();

        Index index = localQueryService.getIndex(region, "statusIndex");
        CloseableIterator<IndexStoreEntry> iter =
            ((CompactRangeIndex) index).getIndexStorage().iterator(null);
        while (iter.hasNext()) {
          Object key = iter.next().getDeserializedKey();
          if (!(key instanceof PdxString)) {
            fail(
                "All keys of the CompactRangeIndex should be PdxStrings and not " + key.getClass());
          }
        }
      }
    });

    // Execute queries from client to server and locally on client
    SerializableRunnable executeQueries = new CacheSerializableRunnable("Execute queries") {
      @Override
      public void run2() throws CacheException {
        QueryService remoteQueryService = null;
        QueryService localQueryService = null;
        SelectResults[][] rs = new SelectResults[1][2];
        try {
          remoteQueryService = (PoolManager.find(poolName)).getQueryService();
          localQueryService = getCache().getQueryService();
        } catch (Exception e) {
          Assert.fail("Failed to get QueryService.", e);
        }

        for (int i = 0; i < queryString.length; i++) {
          try {
            LogWriterUtils.getLogWriter()
                .info("### Executing Query on remote server:" + queryString[i]);
            Query query = remoteQueryService.newQuery(queryString[i]);
            rs[0][0] = (SelectResults) query.execute();
            LogWriterUtils.getLogWriter()
                .info("RR remote indexType: CompactRange size of resultset: " + rs[0][0].size()
                    + " for query: " + queryString[i]);
            checkForPdxString(rs[0][0].asList(), queryString[i]);

            LogWriterUtils.getLogWriter()
                .info("### Executing Query locally on client:" + queryString[i]);
            query = localQueryService.newQuery(queryString[i]);
            rs[0][1] = (SelectResults) query.execute();
            LogWriterUtils.getLogWriter()
                .info("RR  client local indexType: CompactRange size of resultset: "
                    + rs[0][1].size() + " for query: " + queryString[i]);
            checkForPdxString(rs[0][1].asList(), queryString[i]);

            if (i < orderByQueryIndex) {
              // Compare local and remote query results.
              if (!compareResultsOfWithAndWithoutIndex(rs)) {
                fail(
                    "Local and Remote Query Results are not matching for query :" + queryString[i]);
              }
            } else {
              // compare the order of results returned
              compareResultsOrder(rs, false);
            }
          } catch (Exception e) {
            Assert.fail("Failed executing " + queryString[i], e);
          }
        }

      }
    };

    client.invoke(executeQueries);

    // Put Non Pdx objects on server execute queries locally
    server0.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion().getSubregion(regionName);

        LogWriterUtils.getLogWriter().info("Put Objects locally on server");
        for (int i = numberOfEntries; i < numberOfEntries * 2; i++) {
          region.put("key-" + i, new Portfolio(i));
        }
        QueryService localQueryService = getCache().getQueryService();

        // Query server1 locally to check if PdxString is not being returned
        for (final String s : queryString) {
          try {
            LogWriterUtils.getLogWriter()
                .info("### Executing Query locally on server:" + s);
            SelectResults rs = (SelectResults) localQueryService.newQuery(s).execute();
            LogWriterUtils.getLogWriter()
                .info("RR server local indexType:Range  size of resultset: " + rs.size()
                    + " for query: " + s);
            // The results should not be PdxString
            checkForPdxString(rs.asList(), s);
          } catch (Exception e) {
            Assert.fail("Failed executing " + s, e);
          }
        }
      }
    });

    // test for readSerialized flag
    server0.invoke(new CacheSerializableRunnable("Create cache server") {
      @Override
      public void run2() throws CacheException {
        GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
        cache.setReadSerializedForTest(true);
        QueryService localQueryService = getCache().getQueryService();
        // Query server1 locally to check if PdxString is not being returned
        for (final String s : queryString) {
          try {
            LogWriterUtils.getLogWriter()
                .info("### Executing Query locally on server:" + s);
            SelectResults rs = (SelectResults) localQueryService.newQuery(s).execute();
            LogWriterUtils.getLogWriter().info(
                "RR server local readSerializedTrue: indexType: CompactRange size of resultset: "
                    + rs.size() + " for query: " + s);
            // The results should not be PdxString
            checkForPdxString(rs.asList(), s);
          } catch (Exception e) {
            Assert.fail("Failed executing " + s, e);
          }
        }
      }
    });

    // test for readSerialized flag on client
    client.invoke(new CacheSerializableRunnable("Create client") {
      @Override
      public void run2() throws CacheException {
        GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
        cache.setReadSerializedForTest(true);
        QueryService remoteQueryService = (PoolManager.find(poolName)).getQueryService();

        // Query server1 remotely to check if PdxString is not being returned
        for (final String s : queryString) {
          try {
            LogWriterUtils.getLogWriter()
                .info("### Executing Query locally on server:" + s);
            SelectResults rs =
                (SelectResults) remoteQueryService.newQuery(s).execute();
            LogWriterUtils.getLogWriter().info(
                "RR server remote readSerializedTrue: indexType:CompactRange size of resultset: "
                    + rs.size() + " for query: " + s);
            // The results should not be PdxString
            checkForPdxString(rs.asList(), s);
          } catch (Exception e) {
            Assert.fail("Failed executing " + s, e);
          }
        }
      }
    });

    closeClient(server2);
    closeClient(client);
    closeClient(server1);
    closeClient(server0);
  }

  @Test
  public void testReplicatedRegionRangeIndex() throws CacheException {
    final Host host = Host.getHost(0);
    VM server0 = host.getVM(0);
    VM server1 = host.getVM(1);
    VM server2 = host.getVM(2);
    VM client = host.getVM(3);
    final int numberOfEntries = 10;
    // Start server1 and create index
    server0.invoke(new CacheSerializableRunnable("Create Server1") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        // create a local query service
        QueryService localQueryService = null;
        try {
          localQueryService = getCache().getQueryService();
        } catch (Exception e) {
          Assert.fail("Failed to get QueryService.", e);
        }
        // Verify the type of index created
        Index index = null;
        try {
          index = localQueryService.createIndex("secIdIndex", "pos.secId",
              regName + " p, p.positions.values pos");
          if (!(index instanceof RangeIndex)) {
            fail("Range Index should have been created instead of " + index.getClass());
          }
        } catch (Exception ex) {
          fail("Failed to create index." + ex.getMessage());
        }
      }
    });

    // Start server2
    server1.invoke(new CacheSerializableRunnable("Create Server2") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        Region region = getRootRegion().getSubregion(regionName);
      }
    });

    // Start server3
    server2.invoke(new CacheSerializableRunnable("Create Server3") {
      @Override
      public void run2() throws CacheException {
        configAndStartBridgeServer(false, false, false);
        Region region = getRootRegion().getSubregion(regionName);
      }
    });

    // Client pool.
    final int port0 = server0.invoke(PdxStringQueryDUnitTest::getCacheServerPort);
    final int port1 = server1.invoke(PdxStringQueryDUnitTest::getCacheServerPort);
    final int port2 = server2.invoke(PdxStringQueryDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(server0.getHost());

    // Create client pool.
    final String poolName = "testClientServerQueryPool";
    createPool(client, poolName, new String[] {host0}, new int[] {port0, port1, port2}, true);

    // Create client region and put PortfolioPdx objects (PdxInstances)
    client.invoke(new CacheSerializableRunnable("Create client") {
      @Override
      public void run2() throws CacheException {
