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

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.test.util.ResourceUtils.createTempFileFromResource;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import util.TestException;

import org.apache.geode.CancelException;
import org.apache.geode.LogWriter;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.EntryExistsException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.RegionFunctionContext;
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.MultiIndexCreationException;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryException;
import org.apache.geode.cache.query.QueryInvalidException;
import org.apache.geode.cache.query.QueryInvocationTargetException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.RegionNotFoundException;
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.index.PartitionedIndex;
import org.apache.geode.cache.query.types.ObjectType;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.internal.cache.ForceReattemptException;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.SerializableRunnableIF;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;

/**
 * This is a helper class for the various Partitioned Query DUnit Test Cases
 *
 * TODO: inline and then delete class PRQueryDUnitHelper
 */
public class PRQueryDUnitHelper implements Serializable {

  static Cache cache;

  public static void setCache(Cache cache) {
    PRQueryDUnitHelper.cache = cache;
  }

  public static Cache getCache() {
    return cache;
  }

  CacheSerializableRunnable getCacheSerializableRunnableForLocalRegionCreation(
      final String regionName, final Class constraint) {
    return new CacheSerializableRunnable(regionName) {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        Region localRegion = null;
        try {
          AttributesFactory attr = new AttributesFactory();
          attr.setValueConstraint(constraint);
          attr.setScope(Scope.LOCAL);
          localRegion = cache.createRegion(regionName, attr.create());
        } catch (IllegalStateException ex) {
          LogWriterUtils.getLogWriter().warning(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Creation caught IllegalStateException",
              ex);
        }
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region ref null",
            localRegion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region ref claims to be destroyed",
            !localRegion.isDestroyed());
      }
    };
  }

  CacheSerializableRunnable getCacheSerializableRunnableForLocalRegionWithAsyncIndexCreation(
      final String regionName, final Class constraint) {
    return new CacheSerializableRunnable(regionName) {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();

        AttributesFactory attr = new AttributesFactory();
        attr.setValueConstraint(constraint);
        attr.setScope(Scope.LOCAL);
        attr.setIndexMaintenanceSynchronous(false);
        Region localRegion = cache.createRegion(regionName, attr.create());

        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region ref null",
            localRegion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region ref claims to be destroyed",
            !localRegion.isDestroyed());
      }
    };
  }

  public CacheSerializableRunnable getCacheSerializableRunnableForReplicatedRegionCreation(
      final String regionName) {
    return new CacheSerializableRunnable(regionName) {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();

        Region localRegion = cache.createRegionFactory(RegionShortcut.REPLICATE).create(regionName);

        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region ref null",
            localRegion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreate: Partitioned Region ref claims to be destroyed",
            !localRegion.isDestroyed());
      }
    };
  }

  /**
   * This function creates a appropriate region PR given the scope & the redundancy parameters *
   *
   * @return cacheSerializable object
   */
  public CacheSerializableRunnable getCacheSerializableRunnableForPRCreate(final String regionName,
      final int redundancy, final Class constraint) {
    return new CacheSerializableRunnable(regionName) {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        AttributesFactory attr = new AttributesFactory();
        attr.setValueConstraint(constraint);

        PartitionAttributesFactory paf = new PartitionAttributesFactory();
        PartitionAttributes prAttr = paf.setRedundantCopies(redundancy).create();

        attr.setPartitionAttributes(prAttr);

        Region partitionedregion = cache.createRegion(regionName, attr.create());
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref null",
            partitionedregion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref claims to be destroyed",
            !partitionedregion.isDestroyed());
      }
    };
  }

  /**
   * This function creates a colocated pair of PR's given the scope & the redundancy parameters for
   * the parent *
   *
   * @return cacheSerializable object
   */
  CacheSerializableRunnable getCacheSerializableRunnableForColocatedPRCreate(
      final String regionName, final int redundancy, final Class constraint,
      boolean makePersistent) {
    String childRegionName = regionName + "Child";
    String diskName = "disk";

    return new CacheSerializableRunnable(regionName) {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        AttributesFactory attr = new AttributesFactory();
        attr.setValueConstraint(constraint);
        if (makePersistent) {
          DiskStore ds = cache.findDiskStore(diskName);
          if (ds == null) {
            ds = cache.createDiskStoreFactory().setDiskDirs(JUnit4CacheTestCase.getDiskDirs())
                .create(diskName);
          }
          attr.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
          attr.setDiskStoreName(diskName);
        } else {
          attr.setDataPolicy(DataPolicy.PARTITION);
          attr.setDiskStoreName(null);
        }

        PartitionAttributesFactory paf = new PartitionAttributesFactory();
        paf.setRedundantCopies(redundancy);
        attr.setPartitionAttributes(paf.create());

        // parent region
        Region partitionedregion = cache.createRegion(regionName, attr.create());
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref null",
            partitionedregion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref claims to be destroyed",
            !partitionedregion.isDestroyed());

        // child region
        attr.setValueConstraint(constraint);
        paf.setColocatedWith(regionName);
        attr.setPartitionAttributes(paf.create());

        cache.createRegion(childRegionName, attr.create());
      }
    };
  }

  /**
   * This function creates the parent region of colocated pair of PR's given the scope & the
   * redundancy parameters for the parent *
   *
   * @return cacheSerializable object
   */
  CacheSerializableRunnable getCacheSerializableRunnableForColocatedParentCreate(
      final String regionName, final int redundancy, final Class constraint,
      boolean makePersistent) {
    String diskName = "disk";

    return new CacheSerializableRunnable(regionName + "-NoChildRegion") {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        AttributesFactory attr = new AttributesFactory();
        attr.setValueConstraint(constraint);
        if (makePersistent) {
          DiskStore ds = cache.findDiskStore(diskName);
          if (ds == null) {
            ds = cache.createDiskStoreFactory().setDiskDirs(JUnit4CacheTestCase.getDiskDirs())
                .create(diskName);
          }
          attr.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
          attr.setDiskStoreName(diskName);
        } else {
          attr.setDataPolicy(DataPolicy.PARTITION);
          attr.setDiskStoreName(null);
        }

        PartitionAttributesFactory paf = new PartitionAttributesFactory();
        paf.setRedundantCopies(redundancy);
        attr.setPartitionAttributes(paf.create());

        // parent region
        Region partitionedregion = cache.createRegion(regionName, attr.create());
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref null",
            partitionedregion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref claims to be destroyed",
            !partitionedregion.isDestroyed());
      }
    };
  }

  /**
   * This function creates the parent region of colocated pair of PR's given the scope & the
   * redundancy parameters for the parent *
   *
   * @return cacheSerializable object
   */
  CacheSerializableRunnable getCacheSerializableRunnableForColocatedChildCreate(
      final String regionName, final int redundancy, final Class constraint, boolean isPersistent) {

    final String childRegionName = regionName + "Child";
    final String diskName = "disk";
    SerializableRunnable createPrRegion =
        new CacheSerializableRunnable(regionName + "-ChildRegion") {
          @Override
          public void run2() throws CacheException {

            Cache cache = getCache();
            Region partitionedregion = null;
            AttributesFactory attr = new AttributesFactory();
            attr.setValueConstraint(constraint);
            if (isPersistent) {
              DiskStore ds = cache.findDiskStore(diskName);
              if (ds == null) {
                // ds = cache.createDiskStoreFactory().setDiskDirs(getDiskDirs())
                ds = cache.createDiskStoreFactory()
                    .setDiskDirs(
                        JUnit4CacheTestCase.getDiskDirs())
                    .create(diskName);
              }
              attr.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
              attr.setDiskStoreName(diskName);
            } else {
              attr.setDataPolicy(DataPolicy.PARTITION);
              attr.setDiskStoreName(null);
            }

            PartitionAttributesFactory paf = new PartitionAttributesFactory();
            paf.setRedundantCopies(redundancy);
            attr.setPartitionAttributes(paf.create());

            // skip parent region creation
            // partitionedregion = cache.createRegion(regionName, attr.create());

            // child region
            attr.setValueConstraint(constraint);
            paf.setColocatedWith(regionName);
            attr.setPartitionAttributes(paf.create());
            Region childRegion = cache.createRegion(childRegionName, attr.create());
          }
        };

    return (CacheSerializableRunnable) createPrRegion;
  }

  CacheSerializableRunnable getCacheSerializableRunnableForPRCreateLimitedBuckets(
      final String regionName, final int redundancy, final int buckets) {

    SerializableRunnable createPrRegion = new CacheSerializableRunnable(regionName) {
      @Override
      public void run2() throws CacheException {

        Cache cache = getCache();
        Region partitionedregion = null;
        try {
          AttributesFactory attr = new AttributesFactory();
          PartitionAttributesFactory paf = new PartitionAttributesFactory();
          PartitionAttributes prAttr =
              paf.setRedundantCopies(redundancy).setTotalNumBuckets(buckets).create();
          attr.setPartitionAttributes(prAttr);
          partitionedregion = cache.createRegion(regionName, attr.create());
        } catch (IllegalStateException ex) {
          LogWriterUtils.getLogWriter().warning(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Creation caught IllegalStateException",
              ex);
        }
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref null",
            partitionedregion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref claims to be destroyed",
            !partitionedregion.isDestroyed());
      }
    };

    return (CacheSerializableRunnable) createPrRegion;
  }

  CacheSerializableRunnable getCacheSerializableRunnableForPersistentPRCreate(
      final String regionName, final int redundancy, final Class constraint) {

    SerializableRunnable createPrRegion = new CacheSerializableRunnable(regionName) {
      @Override
      public void run2() throws CacheException {

        Cache cache = getCache();
        Region partitionedregion = null;
        try {
          cache.createDiskStoreFactory().setDiskDirs(JUnit4CacheTestCase.getDiskDirs())
              .create("diskstore");
          AttributesFactory attr = new AttributesFactory();
          attr.setValueConstraint(constraint);
          attr.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
          attr.setDiskStoreName("diskstore");

          PartitionAttributesFactory paf = new PartitionAttributesFactory();
          PartitionAttributes prAttr = paf.setRedundantCopies(redundancy).create();

          attr.setPartitionAttributes(prAttr);

          partitionedregion = cache.createRegion(regionName, attr.create());
        } catch (IllegalStateException ex) {
          LogWriterUtils.getLogWriter().warning(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Creation caught IllegalStateException",
              ex);
        }
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref null",
            partitionedregion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref claims to be destroyed",
            !partitionedregion.isDestroyed());
      }
    };

    return (CacheSerializableRunnable) createPrRegion;
  }

  /**
   * This function creates a colocated region PR given the oher colocated region.
   *
   * @return cacheSerializable object
   */

  CacheSerializableRunnable getCacheSerializableRunnableForPRColocatedCreate(
      final String regionName, final int redundancy, final String coloRegionName) {

    SerializableRunnable createPrRegion = new CacheSerializableRunnable(regionName) {
      @Override
      public void run2() throws CacheException {

        Cache cache = getCache();
        Region partitionedregion = null;
        try {
          Region colocatedRegion = cache.getRegion(coloRegionName);
          assertNotNull(colocatedRegion);
          AttributesFactory attr = new AttributesFactory();

          PartitionAttributesFactory paf = new PartitionAttributesFactory();
          PartitionAttributes prAttr = paf.setRedundantCopies(redundancy)
              .setColocatedWith(colocatedRegion.getFullPath()).create();
          attr.setPartitionAttributes(prAttr);

          partitionedregion = cache.createRegion(regionName, attr.create());
        } catch (IllegalStateException ex) {
          LogWriterUtils.getLogWriter().warning(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Creation caught IllegalStateException",
              ex);
        }
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region "
                + regionName + " not in cache",
            cache.getRegion(regionName));
        assertNotNull(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref null",
            partitionedregion);
        assertTrue(
            "PRQueryDUnitHelper#getCacheSerializableRunnableForPRCreateWithRedundancy: Partitioned Region ref claims to be destroyed",
            !partitionedregion.isDestroyed());
      }
    };

    return (CacheSerializableRunnable) createPrRegion;
  }


  /**
   * This function puts portfolio objects into the created Region (PR or Local) *
   *
   * @return cacheSerializable object
   */
  public CacheSerializableRunnable getCacheSerializableRunnableForPRPuts(final String regionName,
      final Object[] portfolio, final int from, final int to) {
    SerializableRunnable prPuts = new CacheSerializableRunnable("PRPuts") {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        Region region = cache.getRegion(regionName);
        for (int j = from; j < to; j++) {
          region.put(j, portfolio[j]);
        }
      }
    };
    return (CacheSerializableRunnable) prPuts;
  }


  /**
   * This function puts portfolio objects into the created Region (PR or RR). Also, other operation
   * like, invalidate, destroy and create are performed in random manner based on
   * {@link Random#nextInt(int)}.
   *
   * @return cacheSerializable object
   */
  public CacheSerializableRunnable getCacheSerializableRunnableForPRRandomOps(
      final String regionName, final int from, final int to) {
    SerializableRunnable prPuts = new CacheSerializableRunnable("PRPuts") {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        Region region = cache.getRegion(regionName);

        for (int i = 0; i < 3; i++) {
          for (int j = from; j < to; j++) {
            int op = new Random().nextInt(4);
            try {
              switch (op) {
                case 0:
                  // Put operation
                  region.put(j, new Portfolio(j));
                  break;
                case 1:
                  // invalidate
                  if (region.containsKey(j)) {
                    region.invalidate(j);
                  }
                  break;
                case 2:
                  if (region.containsKey(j)) {
                    region.destroy(j);
                  }
                  break;
                case 3:

                  if (!region.containsKey(j)) {
                    region.create(j, null);
                  }

                  break;
                default:
                  break;
              }
            } catch (EntryExistsException e) {
              // Do nothing let it go
              LogWriterUtils.getLogWriter()
                  .info("EntryExistsException was thrown for key " + j);
            } catch (EntryNotFoundException e) {
              // Do nothing let it go
              LogWriterUtils.getLogWriter()
                  .info("EntryNotFoundException was thrown for key " + j);
            }
          }
        }
      }
    };
    return (CacheSerializableRunnable) prPuts;
  }

  /**
   * This function puts portfolio objects into the created Region (PR or Local) *
   *
   * @return cacheSerializable object
   */
  CacheSerializableRunnable getCacheSerializableRunnableForPRDuplicatePuts(
      final String regionName, final Object[] portfolio, final int from, final int to) {
    SerializableRunnable prPuts = new CacheSerializableRunnable("PRPuts") {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        Region region = cache.getRegion(regionName);
        for (int j = from, i = to; j < to; j++, i++) {
          region.put(i, portfolio[j]);
        }
      }
    };
    return (CacheSerializableRunnable) prPuts;
  }

  /**
   * This function puts portfolio objects into the created Region (PR or Local) *
   *
   * @return cacheSerializable object
   */
  CacheSerializableRunnable getCacheSerializableRunnableForPRPutsKeyValue(
      final String regionName, final Object[] portfolio, final int from, final int to) {
    SerializableRunnable prPuts = new CacheSerializableRunnable("PRPuts") {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        Region region = cache.getRegion(regionName);
        for (int j = from; j < to; j++) {
          region.put(portfolio[j], portfolio[j]);
        }
      }
    };
    return (CacheSerializableRunnable) prPuts;
  }

  /**
   * This function <br>
   * 1. Creates & executes a query with Logical Operators on the given PR Region 2. Executes the
   * same query on the local region <br>
   * 3. Compares the appropriate resultSet <br>
   */

  CacheSerializableRunnable getCacheSerializableRunnableForPRQueryAndCompareResults(
      final String regionName, final String localRegion) {
    return getCacheSerializableRunnableForPRQueryAndCompareResults(regionName, localRegion, false);
  }

  CacheSerializableRunnable getCacheSerializableRunnableForPRQueryAndCompareResults(
      final String regionName, final String localRegion,
      final boolean fullQueryOnPortfolioPositions) {

    SerializableRunnable PrRegion = new CacheSerializableRunnable("PRQuery") {
      @Override
      public void run2() throws CacheException {

        Cache cache = getCache();
        // Querying the localRegion and the PR region

        String[] queries;

        if (fullQueryOnPortfolioPositions) {
          queries = new String[] {"import org.apache.geode.cache.\"query\".data.Position;"
              + "select distinct r.ID, status, mktValue "
              + "from $1 r, r.positions.values pVal TYPE Position "
              + "where r.status = 'active' AND pVal.mktValue >= 25.00",

              "import org.apache.geode.cache.\"query\".data.Position;" + "select distinct * "
                  + "from $1 r, r.positions.values pVal TYPE Position "
                  + "where r.status = 'active' AND pVal.mktValue >= 25.00",

              "import org.apache.geode.cache.\"query\".data.Position;" + "select distinct ID "
                  + "from $1 r, r.positions.values pVal TYPE Position "
                  + "where r.status = 'active' AND pVal.mktValue >= 25.00",

              "select distinct * " + "from $1 " + "where status = 'active'",

              "import org.apache.geode.cache.\"query\".data.Position;"
                  + "select distinct r from $1 r, "
                  + "r.positions.values pVal TYPE Position where pVal.mktValue < $2",

              "select p.positions.get('acc') from $1 p",

          };
        } else {
          queries = new String[] {"ID = 0 OR ID = 1", "ID > 4 AND ID < 9", "ID = 5", "ID < 5 ",
              "ID <= 5"};
        }

        Object[][] r = new Object[queries.length][2];
        Region local = cache.getRegion(localRegion);
        Region region = cache.getRegion(regionName);
        assertNotNull(region);

        final String[] expectedExceptions =
            new String[] {RegionDestroyedException.class.getName(), ReplyException.class.getName(),
                CacheClosedException.class.getName(), ForceReattemptException.class.getName(),
                QueryInvocationTargetException.class.getName()};

        for (final String exception : expectedExceptions) {
          getCache().getLogger().info(
              "<ExpectedException action=add>" + exception + "</ExpectedException>");
        }

        QueryService qs = getCache().getQueryService();
        try {
          for (int j = 0; j < queries.length; j++) {
            synchronized (region) {
              Object[] params;
              if (fullQueryOnPortfolioPositions) {
                params = new Object[] {local, (j % 25) * 1.0 + 1};
                r[j][0] = qs.newQuery(queries[j]).execute(params);
              } else {
                r[j][0] = local.query(queries[j]);
              }
              if (fullQueryOnPortfolioPositions) {
                params = new Object[] {region, (j % 25) * 1.0 + 1};
                r[j][1] = qs.newQuery(queries[j]).execute(params);
              } else {
                r[j][1] = region.query(queries[j]);
              }
            }
          }
          compareTwoQueryResults(r, queries.length);
        } catch (QueryInvocationTargetException e) {
          // If cause is RegionDestroyedException then its ok
          Throwable cause = e.getCause();
          if (!(cause instanceof RegionDestroyedException)) {
            // throw an unchecked exception so the controller can examine the cause and see whether
            // or not it's okay
            throw new TestException(
                "PRQueryDUnitHelper#getCacheSerializableRunnableForPRQueryAndCompareResults: Caught unexpected query exception",
                e);
          }
        }

        catch (QueryException e) {
          LogWriterUtils.getLogWriter().error(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRQueryAndCompareResults: Caught QueryException while querying"
                  + e,
              e);
          throw new TestException(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRQueryAndCompareResults: Caught unexpected query exception",
              e);
        }

        catch (RegionDestroyedException rde) {
          LogWriterUtils.getLogWriter().info(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRQueryAndCompareResults: Caught a RegionDestroyedException while querying as expected ",
              rde);

        } catch (CancelException cce) {
          LogWriterUtils.getLogWriter().info(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRQueryAndCompareResults: Caught a CancelException while querying as expected ",
              cce);

        } finally {
          for (final String expectedException : expectedExceptions) {
            getCache().getLogger().info(
                "<ExpectedException action=remove>" + expectedException + "</ExpectedException>");
          }
        }

      }

    };
    return (CacheSerializableRunnable) PrRegion;
  }


  CacheSerializableRunnable getCacheSerializableRunnableForPROrderByQueryAndCompareResults(
      final String regionName, final String localRegion) {
    SerializableRunnable PrRegion = new CacheSerializableRunnable("PRQuery") {
      @Override
      public void run2() throws CacheException {

        Cache cache = getCache();
        // Querying the localRegion and the PR region

        String[] queries =
            new String[] {"p.status from " + SEPARATOR + "REGION_NAME p order by p.status",
                "* from " + SEPARATOR + "REGION_NAME order by status, ID desc",
                "status, ID from " + SEPARATOR + "REGION_NAME order by status",
                "p.status, p.ID from " + SEPARATOR + "REGION_NAME p order by p.status",
                "p.position1.secId, p.ID from " + SEPARATOR
                    + "REGION_NAME p order by p.position1.secId",
                "key from " + SEPARATOR + "REGION_NAME.keys key order by key.status",
                "key.ID from " + SEPARATOR + "REGION_NAME.keys key order by key.ID",
                "key.ID, key.status from " + SEPARATOR + "REGION_NAME.keys key order by key.status",
                "key.ID, key.status from " + SEPARATOR
                    + "REGION_NAME.keys key order by key.status, key.ID",
                "key.ID, key.status from " + SEPARATOR
                    + "REGION_NAME.keys key order by key.status desc, key.ID",
                "key.ID, key.status from " + SEPARATOR
                    + "REGION_NAME.keys key order by key.status, key.ID desc",
