/*
 * 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.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.util.Comparator;
import java.util.Iterator;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;

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.Region;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.ClientCache;
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.IndexType;
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.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.QueryObserverAdapter;
import org.apache.geode.cache.query.internal.QueryObserverHolder;
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.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.DistributedTestUtils;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.junit.categories.OQLQueryTest;

/**
 * Tests remote (client/server) query execution.
 *
 * @since GemFire 5.0.1
 */
@Category({OQLQueryTest.class})
public class QueryUsingPoolDUnitTest extends JUnit4CacheTestCase {
  private static final Logger logger = LogService.getLogger();

  /**
   * The port on which the cache server was started in this VM
   */
  private static int bridgeServerPort;

  private String rootRegionName;
  private String regionName;
  private String regName;

  // Used with compiled queries.
  private String[] queryString;

  @Override
  public final void postSetUp() throws Exception {
    rootRegionName = "root";
    regionName = getName();
    regName = SEPARATOR + rootRegionName + SEPARATOR + regionName;

    queryString =
        new String[] {"SELECT itr.value FROM " + regName + ".entries itr where itr.key = $1", // 0
            "SELECT DISTINCT * FROM " + regName + " WHERE id < $1 ORDER BY   id", // 1
            "SELECT DISTINCT * FROM " + regName + " WHERE id < $1 ORDER BY id", // 2
            "(SELECT DISTINCT * FROM " + regName + " WHERE id < $1).size", // 3
            "SELECT * FROM " + regName + " WHERE id = $1 and Ticker = $2", // 4
            "SELECT * FROM " + regName + " WHERE id < $1 and Ticker = $2", // 5
        };

    disconnectAllFromDS();
    IgnoredException.addIgnoredException("Connection reset");
    IgnoredException.addIgnoredException("Socket input is shutdown");
    IgnoredException.addIgnoredException("Connection refused");
  }

  @Override
  public final void postTearDownCacheTestCase() throws Exception {
    disconnectAllFromDS();
  }

  public void createPool(String poolName, String server, int port, boolean subscriptionEnabled) {
    createPool(poolName, new String[] {server}, new int[] {port}, subscriptionEnabled);
  }

  public void createPool(String poolName, String server, int port) {
    createPool(poolName, new String[] {server}, new int[] {port}, false);
  }

  public void createPool(final String poolName, final String[] servers, final int[] ports,
      final boolean subscriptionEnabled) {
    // Create Cache.
    getCache(true);

    PoolFactory poolFactory = PoolManager.createFactory();
    poolFactory.setSubscriptionEnabled(subscriptionEnabled);
    for (int i = 0; i < servers.length; i++) {
      logger.info("### Adding to Pool. ### Server : " + servers[i] + " Port : " + ports[i]);
      poolFactory.addServer(servers[i], ports[i]);
    }
    poolFactory.setMinConnections(1);
    poolFactory.setMaxConnections(5);
    poolFactory.setRetryAttempts(5);
    poolFactory.create(poolName);
  }

  public void validateCompiledQuery(final long compiledQueryCount) {
    await().timeout(60, TimeUnit.SECONDS)
        .until(() -> CacheClientNotifier.getInstance().getStats()
            .getCompiledQueryCount() == compiledQueryCount);
  }

  /**
   * Tests remote import query execution.
   */
  @Test
  public void testRemoteImportQueries() throws CacheException {

    final String name = getName();
    final String rootRegionName = "root";

    final Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    final int numberOfEntries = 100;

    // Start server
    vm0.invoke("Create cache server", () -> {
      createAndStartBridgeServer();
      AttributesFactory factory = new AttributesFactory();
      factory.setScope(Scope.LOCAL);
      createRegion(name, rootRegionName, factory.create());
    });

    // Initialize server region
    vm0.invoke("Create cache server", () -> {
      Region region = getRootRegion().getSubregion(name);
      for (int i = 0; i < numberOfEntries; i++) {
        region.put("key-" + i, new TestObject(i, "ibm"));
      }
    });

    final int port =
        vm0.invoke("GetCacheServerPort", QueryUsingPoolDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
    final String regionName = SEPARATOR + rootRegionName + SEPARATOR + name;

    // Create client pool.
    final String poolName = "testRemoteImportQueries";
    vm1.invoke("createPool", () -> createPool(poolName, host0, port));

    // Execute client queries
    vm1.invoke("Execute queries", () -> {
      String queryString = null;
      SelectResults results = null;

      QueryService qService = null;

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

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct * from "
              + regionName;

      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }

      assertEquals(numberOfEntries, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct * from "
              + regionName + " where ticker = 'ibm'";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(numberOfEntries, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct * from "
              + regionName + " where ticker = 'IBM'";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(0, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct * from "
              + regionName + " where price > 49";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(numberOfEntries / 2, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct * from "
              + regionName + " where price = 50";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(1, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct * from "
              + regionName + " where ticker = 'ibm' and price = 50";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(1, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates());
    });

    // Stop server
    vm0.invoke("Stop CacheServer", () -> stopBridgeServer(getCache()));
  }

  /**
   * Tests remote struct query execution.
   */
  @Test
  public void testRemoteStructQueries() throws CacheException {

    final String name = getName();
    final String rootRegionName = "root";

    final Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    final int numberOfEntries = 100;

    // Start server
    final int port = vm0.invoke("Create cache server", () -> {
      setupBridgeServerAndCreateData(name, numberOfEntries);
      return getCacheServerPort();
    });

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

    final String regionName = SEPARATOR + rootRegionName + SEPARATOR + name;

    // Create client pool.
    final String poolName = "testRemoteStructQueries";
    vm1.invoke("createPool", () -> createPool(poolName, host0, port));

    // Execute client queries
    vm1.invoke("Execute queries", () -> {
      String queryString = null;
      SelectResults results = null;

      QueryService qService = null;

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

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct ticker, price from "
              + regionName;
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(numberOfEntries, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates()
          && results.getCollectionType().getElementType().isStructType());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct ticker, price from "
              + regionName + " where ticker = 'ibm'";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(numberOfEntries, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates()
          && results.getCollectionType().getElementType().isStructType());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct ticker, price from "
              + regionName + " where ticker = 'IBM'";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(0, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates()
          && results.getCollectionType().getElementType().isStructType());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct ticker, price from "
              + regionName + " where price > 49";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(numberOfEntries / 2, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates()
          && results.getCollectionType().getElementType().isStructType());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct ticker, price from "
              + regionName + " where price = 50";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(1, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates()
          && results.getCollectionType().getElementType().isStructType());

      queryString =
          "import org.apache.geode.admin.QueryUsingPoolDUnitTest.TestObject; select distinct ticker, price from "
              + regionName + " where ticker = 'ibm' and price = 50";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(1, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates()
          && results.getCollectionType().getElementType().isStructType());
    });

    // Stop server
    vm0.invoke("Stop CacheServer", () -> stopBridgeServer(getCache()));
  }

  private void setupBridgeServerAndCreateData(String name, int numberOfEntries) {
    createAndStartBridgeServer();
    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.LOCAL);
    createRegion(regionName, rootRegionName, factory.create());
    Region region = getRootRegion().getSubregion(name);
    for (int i = 0; i < numberOfEntries; i++) {
      region.put("key-" + i, new TestObject(i, "ibm"));
    }
  }


  /**
   * Tests remote full region query execution.
   */
  @Test
  public void testRemoteFullRegionQueries() throws CacheException {

    final String name = getName();
    final String rootRegionName = "root";

    final Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    final int numberOfEntries = 100;

    // Start server
    vm0.invoke("Create cache server", () -> {
      createAndStartBridgeServer();
      AttributesFactory factory = new AttributesFactory();
      factory.setScope(Scope.LOCAL);
      createRegion(name, factory.create());
    });

    // Initialize server region
    vm0.invoke("Create cache server", () -> {
      Region region = getRootRegion().getSubregion(name);
      for (int i = 0; i < numberOfEntries; i++) {
        region.put("key-" + i, new TestObject(i, "ibm"));
      }
    });

    // Create client region
    final int port = vm0.invoke(QueryUsingPoolDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());

    final String regionName = SEPARATOR + rootRegionName + SEPARATOR + name;

    // Create client pool.
    final String poolName = "testRemoteFullRegionQueries";
    vm1.invoke("createPool", () -> createPool(poolName, host0, port));

    // Execute client queries
    vm1.invoke("Execute queries", () -> {
      String queryString = null;
      SelectResults results = null;
      Comparator comparator = null;
      Object[] resultsArray = null;
      QueryService qService = null;

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

      // value query
      queryString =
          "SELECT DISTINCT itr.value FROM " + regionName + ".entries itr where itr.key = 'key-1'";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(1, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates());
      assertTrue(results.asList().get(0) instanceof TestObject);

      // key query
      queryString =
          "SELECT DISTINCT itr.key FROM " + regionName + ".entries itr where itr.key = 'key-1'";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(1, results.size());
      assertTrue(!results.getCollectionType().allowsDuplicates());
      assertEquals("key-1", results.asList().get(0));

      // order by value query
      queryString = "SELECT DISTINCT * FROM " + regionName + " WHERE id < 101 ORDER BY id";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(numberOfEntries, results.size());
      // All order-by query results are stored in a ResultsCollectionWrapper
      // wrapping a list, so the assertion below is not correct even though
      // it should be.
      // assertTrue(!results.getCollectionType().allowsDuplicates());
      assertTrue(results.getCollectionType().isOrdered());
      comparator = new IdComparator();
      resultsArray = results.toArray();
      for (int i = 0; i < resultsArray.length; i++) {
        if (i + 1 != resultsArray.length) {
          // The id of the current element in the result set must be less
          // than the id of the next one to pass.
          assertTrue(
              "The id for " + resultsArray[i] + " should be less than the id for "
                  + resultsArray[i + 1],
              comparator.compare(resultsArray[i], resultsArray[i + 1]) == -1);
        }
      }

      // order by struct query
      queryString =
          "SELECT DISTINCT id, ticker, price FROM " + regionName + " WHERE id < 101 ORDER BY id";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(numberOfEntries, results.size());
      // All order-by query results are stored in a ResultsCollectionWrapper
      // wrapping a list, so the assertion below is not correct even though
      // it should be.
      // assertTrue(!results.getCollectionType().allowsDuplicates());
      assertTrue(results.getCollectionType().isOrdered());
      comparator = new StructIdComparator();
      resultsArray = results.toArray();
      for (int i = 0; i < resultsArray.length; i++) {
        if (i + 1 != resultsArray.length) {
          // The id of the current element in the result set must be less
          // than the id of the next one to pass.
          assertTrue(
              "The id for " + resultsArray[i] + " should be less than the id for "
                  + resultsArray[i + 1],
              comparator.compare(resultsArray[i], resultsArray[i + 1]) == -1);
        }
      }

      // size query
      queryString = "(SELECT DISTINCT * FROM " + regionName + " WHERE id < 101).size";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(1, results.size());
      Object result = results.iterator().next();
      assertTrue(result instanceof Integer);
      int resultInt = (Integer) result;
      assertEquals(resultInt, 100);

      // query with leading/trailing spaces
      queryString =
          " SELECT DISTINCT itr.key FROM " + regionName + ".entries itr where itr.key = 'key-1' ";
      try {
        Query query = qService.newQuery(queryString);
        results = (SelectResults) query.execute();
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString, e);
      }
      assertEquals(1, results.size());
      assertEquals("key-1", results.asList().get(0));
    });

    // Stop server
    vm0.invoke("Stop CacheServer", () -> stopBridgeServer(getCache()));
  }

  private int createAndStartBridgeServer() {
    Properties config = new Properties();
    config.setProperty(LOCATORS, "localhost[" + DistributedTestUtils.getDUnitLocatorPort() + "]");
    getSystem(config);
    try {
      return startBridgeServer(0, false);
    } catch (Exception ex) {
      Assert.fail("While starting CacheServer", ex);
      return -1;
    }
  }

  /**
   * Tests client-server query using parameters (compiled queries).
   */
  @Test
  public void testClientServerQueriesWithParams() throws CacheException {
    final Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    final int numberOfEntries = 100;

    final Object[][] params = new Object[][] {{"key-1"}, // 0
        {101}, // 1
        {101}, // 2
        {101}, // 3
        {50, "ibm"}, // 4
        {50, "ibm"}, // 5
    };

    final int[] expectedResults = new int[] {1, // 0
        100, // 1
        100, // 2
        1, // 3
        1, // 4
        50, // 5
    };

    assertNotNull(regionName);

    // Start server
    final int port = vm0.invoke("Create cache server", () -> {
      setupBridgeServerAndCreateData(regionName, numberOfEntries);
      return getCacheServerPort();
    });

    // Create client region
    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());

    // Create client pool.
    final String poolName = "testClientServerQueriesWithParams";
    vm1.invoke("createPool", () -> createPool(poolName, host0, port));

    // Execute client queries
    vm1.invoke(new CacheSerializableRunnable("Execute queries") {
      @Override
      public void run2() throws CacheException {
        SelectResults results = null;
        QueryService qService = null;

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

        executeQueriesForClientServerQueriesWithParams(results, qService, params, expectedResults);
      }
    });

    final int useMaintainedCompiledQueries = queryString.length;

    // Execute the same compiled queries multiple time
    vm1.invoke("Execute queries", () -> {
      SelectResults results = null;
      QueryService qService = null;

      try {
        qService = (PoolManager.find(poolName)).getQueryService();
      } catch (Exception e) {
        Assert.fail("Failed to get QueryService.", e);
      }
      for (int x = 0; x < useMaintainedCompiledQueries; x++) {
        executeQueriesForClientServerQueriesWithParams(results, qService, params, expectedResults);
      }
    });

    // Validate maintained compiled queries.
    // There should be only queryString.length compiled queries registered.
    vm0.invoke("validate compiled query.", () -> {
      long compiledQueryCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryCount();
      assertEquals(queryString.length, compiledQueryCount);
    });

    // Check to see if maintained compiled queries are used.
    vm0.invoke("validate compiled query.", () -> {
      long compiledQueryUsedCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryUsedCount();
      int numTimesUsed = (useMaintainedCompiledQueries + 1) * queryString.length;
      assertEquals(numTimesUsed, compiledQueryUsedCount);
    });

    // Stop server
    vm0.invoke("Stop CacheServer", () -> stopBridgeServer(getCache()));
  }

  private void executeQueriesForClientServerQueriesWithParams(SelectResults results,
      QueryService qService, Object[][] params, int[] expectedResults) {
    for (int i = 0; i < queryString.length; i++) {
      try {
        logger.info("### Executing Query :" + queryString[i]);
        Query query = qService.newQuery(queryString[i]);
        results = (SelectResults) query.execute(params[i]);
      } catch (Exception e) {
        Assert.fail("Failed executing " + queryString[i], e);
      }
      try {
        assertEquals(expectedResults[i], results.size());
      } catch (Throwable th) {
        fail("Result mismatch for query= " + queryString[i] + " expected = " + expectedResults[i]
            + " actual=" + results.size());
      }
    }
  }

  /**
   * Tests client-server query using parameters (compiled queries).
   */
  @Test
  public void testMulitipleClientServerQueriesWithParams() throws CacheException {
    final Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    VM vm2 = host.getVM(2);
    VM vm3 = host.getVM(3);
    final int numberOfEntries = 100;

    final Object[][] params = new Object[][] {{"key-1"}, // 0
        {101}, // 1
        {101}, // 2
        {101}, // 3
        {50, "ibm"}, // 4
        {50, "ibm"}, // 5
    };

    final int[] expectedResults = new int[] {1, // 0
        100, // 1
        100, // 2
        1, // 3
        1, // 4
        50, // 5
    };

    // Start server1
    final int port0 = vm0.invoke("Create cache server", () -> {
      setupBridgeServerAndCreateData(regionName, numberOfEntries);
      return getCacheServerPort();
    });

    // Start server2
    final int port1 = vm1.invoke("Create cache server", () -> {
      setupBridgeServerAndCreateData(regionName, numberOfEntries);
      return getCacheServerPort();
    });

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

    // Create client pool.
    final String poolName = "testClientServerQueriesWithParams";
    vm2.invoke("createPool",
        () -> createPool(poolName, new String[] {host0}, new int[] {port0}, true));
    vm3.invoke("createPool",
        () -> createPool(poolName, new String[] {host0}, new int[] {port1}, true));

    // Execute client queries
    SerializableRunnable executeQueries = new CacheSerializableRunnable("Execute queries") {
      @Override
      public void run2() throws CacheException {
        SelectResults results = null;
        QueryService qService = null;

        try {
          qService = (PoolManager.find(poolName)).getQueryService();
        } catch (Exception e) {
          Assert.fail("Failed to get QueryService.", e);
        }
        for (int j = 0; j < queryString.length; j++) {
          executeQueriesForClientServerQueriesWithParams(results, qService, params,
              expectedResults);
        }
      }
    };

    vm2.invoke(executeQueries);
    vm3.invoke(executeQueries);

    // Validate maintained compiled queries.
    // There should be only queryString.length compiled queries registered.
    vm0.invoke("validate compiled query.", () -> {
      long compiledQueryCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryCount();
      assertEquals(queryString.length, compiledQueryCount);
    });

    vm2.invoke("closeClient", this::closeClient);
    vm3.invoke("closeClient", this::closeClient);

    // Validate maintained compiled queries.
    // All the queries will be still present in the server.
    // They will be cleaned up periodically1
    vm0.invoke("validate compiled query.", () -> {
      long compiledQueryCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryCount();
      assertEquals(queryString.length, compiledQueryCount);
    });

    vm1.invoke("validate compiled query.", () -> {
      long compiledQueryCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryCount();
      assertEquals(queryString.length, compiledQueryCount);
    });

    // recreate clients and execute queries.
    vm2.invoke("createPool",
        () -> createPool(poolName, new String[] {host0, host0}, new int[] {port1, port0}, true));
    vm3.invoke("createPool",
        () -> createPool(poolName, new String[] {host0, host0}, new int[] {port0, port1}, true));

    vm2.invoke(executeQueries);
    vm3.invoke(executeQueries);

    // Validate maintained compiled queries.
    // All the queries will be still present in the server.
    vm0.invoke("validate compiled query.", () -> {
      long compiledQueryCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryCount();
      assertEquals(queryString.length, compiledQueryCount);
    });

    vm1.invoke("validate compiled query.", () -> {
      long compiledQueryCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryCount();
      assertEquals(queryString.length, compiledQueryCount);
    });

    // Stop server
    vm0.invoke("Stop CacheServer", () -> stopBridgeServer(getCache()));
    vm1.invoke("Stop CacheServer", () -> stopBridgeServer(getCache()));
  }

  /**
   * Tests client-server compiled query register and cleanup.
   */
  @Test
  public void testClientServerCompiledQueryRegisterAndCleanup() throws CacheException {

    final String name = getName();
    final Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    VM vm2 = host.getVM(2);
    final int numberOfEntries = 100;

    final Object[][] params = new Object[][] {{"key-1"}, // 0
        {101}, // 1
        {101}, // 2
        {101}, // 3
        {50, "ibm"}, // 4
        {50, "ibm"}, // 5
    };

    // Start server
    final int port = vm0.invoke("Create cache server", () -> {
      setupBridgeServerAndCreateData(name, numberOfEntries);
      return getCacheServerPort();
    });

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

    // Create client pool.
    final String poolName = "testClientServerQueriesWithParams";
    vm1.invoke("createPool", () -> createPool(poolName, host0, port));
    vm2.invoke("createPool", () -> createPool(poolName, host0, port));

    // Execute client queries
    vm1.invoke("executeCompiledQueries", () -> executeCompiledQueries(poolName, params));

    // Execute client queries
    vm2.invoke("executeCompiledQueries", () -> executeCompiledQueries(poolName, params));

    // Validate maintained compiled queries.
    vm0.invoke("validate compiled query.", () -> {
      long compiledQueryCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryCount();
      assertEquals(queryString.length, compiledQueryCount);
    });

    vm1.invoke("closeClient", this::closeClient);

    // Validate maintained compiled queries.
    vm0.invoke("validate compiled query.", () -> {
      long compiledQueryCount =
          CacheClientNotifier.getInstance().getStats().getCompiledQueryCount();
