/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.cache.query.internal;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.LongAdder;

import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.client.internal.ProxyCache;
import org.apache.geode.cache.client.internal.ServerProxy;
import org.apache.geode.cache.client.internal.UserAttributes;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.RegionFunctionContext;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.cache.query.FunctionDomainException;
import org.apache.geode.cache.query.NameResolutionException;
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.QueryStatistics;
import org.apache.geode.cache.query.RegionNotFoundException;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.TypeMismatchException;
import org.apache.geode.cache.query.internal.cq.InternalCqQuery;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.internal.NanoTimer;
import org.apache.geode.internal.cache.BucketRegion;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalDataSet;
import org.apache.geode.internal.cache.PRQueryProcessor;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.TXStateProxy;

/**
 * Thread-safe implementation of org.apache.persistence.query.Query
 */
public class DefaultQuery implements Query {

  private final CompiledValue compiledQuery;

  private final String queryString;

  private final InternalCache cache;

  private ServerProxy serverProxy;

  private final LongAdder numExecutions = new LongAdder();

  private final LongAdder totalExecutionTime = new LongAdder();

  private final QueryStatistics stats;

  private boolean traceOn = false;

  @Immutable
  private static final Object[] EMPTY_ARRAY = new Object[0];

  @MutableForTesting
  public static boolean QUERY_VERBOSE =
      Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "Query.VERBOSE");

  /**
   * System property to cleanup the compiled query. The compiled query will be removed if it is not
   * used for more than the set value. By default its set to 10 minutes, the time is set in
   * MilliSecs.
   */
  public static final int COMPILED_QUERY_CLEAR_TIME = Integer.getInteger(
      DistributionConfig.GEMFIRE_PREFIX + "Query.COMPILED_QUERY_CLEAR_TIME", 10 * 60 * 1000);

  @MutableForTesting
  public static int TEST_COMPILED_QUERY_CLEAR_TIME = -1;

  /**
   * Use to represent null result. Used while adding PR results to the results-queue, which is a
   * blocking queue.
   */
  public static final Object NULL_RESULT = new Object();

  private ProxyCache proxyCache;

  private boolean isCqQuery = false;

  private boolean isQueryWithFunctionContext = false;

  /**
   * Holds the CQ reference. In cases of peer PRs this will be set to null even though isCqQuery is
   * set to true.
   */
  private InternalCqQuery cqQuery = null;

  private volatile boolean lastUsed = true;

  @MakeNotStatic
  public static TestHook testHook;

  /** indicates query executed remotely */
  private boolean isRemoteQuery = false;

  // to prevent objects from getting deserialized
  private boolean keepSerialized = false;


  /**
   * Caches the fields not found in any Pdx version. This threadlocal will be cleaned up after query
   * execution completes in {@linkplain #executeUsingContext(ExecutionContext)}
   */
  private static final ThreadLocal<Map<String, Set<String>>> pdxClassToFieldsMap =
      ThreadLocal.withInitial(HashMap::new);

  public static Map<String, Set<String>> getPdxClasstofieldsmap() {
    return pdxClassToFieldsMap.get();
  }


  /**
   * Caches the methods not found in any Pdx version. This threadlocal will be cleaned up after
   * query execution completes in {@linkplain #executeUsingContext(ExecutionContext)}
   */
  private static final ThreadLocal<Map<String, Set<String>>> pdxClassToMethodsMap =
      ThreadLocal.withInitial(HashMap::new);

  public static void setPdxClasstoMethodsmap(Map<String, Set<String>> map) {
    pdxClassToMethodsMap.set(map);
  }

  public static Map<String, Set<String>> getPdxClasstoMethodsmap() {
    return pdxClassToMethodsMap.get();
  }

  /**
   * Should be constructed from DefaultQueryService
   *
   * @see QueryService#newQuery
   */
  public DefaultQuery(String queryString, InternalCache cache, boolean isForRemote) {
    this.queryString = queryString;
    QCompiler compiler = new QCompiler();
    this.compiledQuery = compiler.compileQuery(queryString);
    CompiledSelect cs = getSimpleSelect();
    if (cs != null && !isForRemote && (cs.isGroupBy() || cs.isOrderBy())) {
      QueryExecutionContext ctx = new QueryExecutionContext(null, cache);
      try {
        cs.computeDependencies(ctx);
      } catch (QueryException qe) {
        throw new QueryInvalidException("", qe);
      }
    }
    this.traceOn = compiler.isTraceRequested() || QUERY_VERBOSE;
    this.cache = cache;
    this.stats = new DefaultQueryStatistics();
  }

  /**
   * Get statistics information for this query.
   */
  @Override
  public QueryStatistics getStatistics() {
    return this.stats;
  }

  @Override
  public String getQueryString() {
    return this.queryString;
  }

  @Override
  public Object execute() throws FunctionDomainException, TypeMismatchException,
      NameResolutionException, QueryInvocationTargetException {
    return execute(EMPTY_ARRAY);
  }

  /**
   * namespace or parameters can be null
   */
  @Override
  public Object execute(Object[] params) throws FunctionDomainException, TypeMismatchException,
      NameResolutionException, QueryInvocationTargetException {

    // Local Query.
    if (params == null) {
      throw new IllegalArgumentException(
          "'parameters' cannot be null");
    }

    // If pool is associated with the Query; execute the query on pool. ServerSide query.
    if (this.serverProxy != null) {
      // Execute Query using pool.
      return executeOnServer(params);
    }

    long startTime = 0L;
    if (this.traceOn && this.cache != null) {
      startTime = NanoTimer.getTime();
    }

    QueryObserver indexObserver = null;
    QueryMonitor queryMonitor = null;
    QueryExecutor qe = checkQueryOnPR(params);

    Object result = null;
    Boolean initialPdxReadSerialized = this.cache.getPdxReadSerializedOverride();
    final ExecutionContext context = new QueryExecutionContext(params, this.cache, this);

    try {
      // Setting the readSerialized flag for local queries
      this.cache.setPdxReadSerializedOverride(true);
      indexObserver = this.startTrace();
      if (qe != null) {
        if (DefaultQuery.testHook != null) {
          DefaultQuery.testHook.doTestHook(DefaultQuery.TestHook.SPOTS.BEFORE_QUERY_EXECUTION,
              this, context);
        }

        result = qe.executeQuery(this, context, params, null);
        // For local queries returning pdx objects wrap the resultset with
        // ResultsCollectionPdxDeserializerWrapper
        // which deserializes these pdx objects.
        if (needsPDXDeserializationWrapper(true /* is query on PR */)
            && result instanceof SelectResults) {
          // we use copy on read false here because the copying has already taken effect earlier in
          // the PartitionedRegionQueryEvaluator
          result = new ResultsCollectionPdxDeserializerWrapper((SelectResults) result, false);
        }
        return result;
      }

      queryMonitor = this.cache.getQueryMonitor();

      // If QueryMonitor is enabled add query to be monitored.
      if (queryMonitor != null) {
        // Add current thread to be monitored by QueryMonitor.
        // In case of partitioned region it will be added before the query execution
        // starts on the Local Buckets.
        queryMonitor.monitorQueryExecution(context);
      }

      result = executeUsingContext(context);
      // Only wrap/copy results when copy on read is set and an index is used
      // This is because when an index is used, the results are actual references to values in the
      // cache
      // Currently as 7.0.1 when indexes are not used, iteration uses non tx entries to retrieve the
      // value.
      // The non tx entry already checks copy on read and returns a copy.
      // We only wrap actual results and not UNDEFINED.

      // Takes into consideration that isRemoteQuery is already being checked with the if checks
      // this flag is true if copy on read is set to true and we are copying at the entry level for
      // queries is set to false (default)
      // OR copy on read is true and we used an index where copy on entry level for queries is set
      // to true.
      // Due to bug#46970 index usage does not actually copy at the entry level so that is why we
      // have the OR condition
      boolean needsCopyOnReadWrapper =
          this.cache.getCopyOnRead() && !DefaultQueryService.COPY_ON_READ_AT_ENTRY_LEVEL
              || (((QueryExecutionContext) context).isIndexUsed()
                  && DefaultQueryService.COPY_ON_READ_AT_ENTRY_LEVEL);
      // For local queries returning pdx objects wrap the resultset with
      // ResultsCollectionPdxDeserializerWrapper
      // which deserializes these pdx objects.
      if (needsPDXDeserializationWrapper(false /* is query on PR */)
          && result instanceof SelectResults) {
        result = new ResultsCollectionPdxDeserializerWrapper((SelectResults) result,
            needsCopyOnReadWrapper);
      } else if (!isRemoteQuery() && this.cache.getCopyOnRead()
          && result instanceof SelectResults) {
        if (needsCopyOnReadWrapper) {
          result = new ResultsCollectionCopyOnReadWrapper((SelectResults) result);
        }
      }
      return result;
    } catch (QueryExecutionCanceledException ignore) {
      return context.reinterpretQueryExecutionCanceledException();
    } finally {
      this.cache.setPdxReadSerializedOverride(initialPdxReadSerialized);
      if (queryMonitor != null) {
        queryMonitor.stopMonitoringQueryExecution(context);
      }
      this.endTrace(indexObserver, startTime, result);
    }
  }

  /**
   * For Order by queries ,since they are already ordered by the comparator && it takes care of
   * conversion, we do not have to wrap it in a wrapper
   */
  private boolean needsPDXDeserializationWrapper(boolean isQueryOnPR) {
    return !isRemoteQuery() && !this.cache.getPdxReadSerialized();
  }

  private Object executeOnServer(Object[] parameters) {
    long startTime = CachePerfStats.getStatTime();
    Object result = null;
    try {
      if (this.proxyCache != null) {
        if (this.proxyCache.isClosed()) {
          throw proxyCache.getCacheClosedException("Cache is closed for this user.");
        }
        UserAttributes.userAttributes.set(this.proxyCache.getUserAttributes());
      }
      result = this.serverProxy.query(this.queryString, parameters);
    } finally {
      UserAttributes.userAttributes.set(null);
      long endTime = CachePerfStats.getStatTime();
      updateStatistics(endTime - startTime);
    }
    return result;
  }

  /**
   * Execute a PR Query on the specified bucket. Assumes query already meets restrictions for PR
   * Query, and the first iterator in the FROM clause can be replaced with the BucketRegion.
   */
  public Object prExecuteOnBucket(Object[] parameters, PartitionedRegion pr, BucketRegion bukRgn)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {
    if (parameters == null) {
      parameters = EMPTY_ARRAY;
    }

    long startTime = 0L;
    if (this.traceOn && this.cache != null) {
      startTime = NanoTimer.getTime();
    }

    IndexTrackingQueryObserver indexObserver = null;
    String otherObserver = null;
    if (this.traceOn) {
      QueryObserver qo = QueryObserverHolder.getInstance();
      if (qo instanceof IndexTrackingQueryObserver) {
        indexObserver = (IndexTrackingQueryObserver) qo;
      } else if (!QueryObserverHolder.hasObserver()) {
        indexObserver = new IndexTrackingQueryObserver();
        QueryObserverHolder.setInstance(indexObserver);
      } else {
        otherObserver = qo.getClass().getName();
      }
    }

    final ExecutionContext context = new QueryExecutionContext(parameters, this.cache, this);
    context.setBucketRegion(pr, bukRgn);

    // Check if QueryMonitor is enabled, if enabled add query to be monitored.
    QueryMonitor queryMonitor = this.cache.getQueryMonitor();

    // PRQueryProcessor executes the query using single thread(in-line) or ThreadPool.
    // In case of threadPool each individual threads needs to be added into
    // QueryMonitor Service.
    if (queryMonitor != null && PRQueryProcessor.NUM_THREADS > 1) {
      // Add current thread to be monitored by QueryMonitor.
      queryMonitor.monitorQueryExecution(context);
    }

    Object result = null;
    try {
      result = executeUsingContext(context);
    } finally {
      if (queryMonitor != null && PRQueryProcessor.NUM_THREADS > 1) {
        queryMonitor.stopMonitoringQueryExecution(context);
      }

      int resultSize = 0;
      if (this.traceOn) {
        if (result instanceof Collection) {
          resultSize = ((Collection) result).size();
        }
      }

      String queryVerboseMsg = DefaultQuery.getLogMessage(indexObserver, startTime, otherObserver,
          resultSize, this.queryString, bukRgn);

      if (this.traceOn) {
        if (this.cache.getLogger().fineEnabled()) {
          this.cache.getLogger().fine(queryVerboseMsg);
        }
      }
    }
    return result;
  }

  public Object executeUsingContext(ExecutionContext context) throws FunctionDomainException,
      TypeMismatchException, NameResolutionException, QueryInvocationTargetException {
    QueryObserver observer = QueryObserverHolder.getInstance();

    long startTime = CachePerfStats.getStatTime();
    TXStateProxy tx = ((TXManagerImpl) this.cache.getCacheTransactionManager()).pauseTransaction();
    try {
      observer.startQuery(this);
      observer.beforeQueryEvaluation(this.compiledQuery, context);

      if (DefaultQuery.testHook != null) {
        DefaultQuery.testHook.doTestHook(TestHook.SPOTS.BEFORE_QUERY_DEPENDENCY_COMPUTATION, this,
            context);
      }
      Object results = null;
      try {
        // two-pass evaluation.
        // first pre-compute dependencies, cached in the context.
        this.compiledQuery.computeDependencies(context);
        if (testHook != null) {
          testHook.doTestHook(DefaultQuery.TestHook.SPOTS.BEFORE_QUERY_EXECUTION, this, context);
        }
        results = this.compiledQuery.evaluate(context);
      } catch (QueryExecutionCanceledException ignore) {
        context.reinterpretQueryExecutionCanceledException();
      } finally {
        observer.afterQueryEvaluation(results);
      }
      return results;
    } finally {
      observer.endQuery();
      long endTime = CachePerfStats.getStatTime();
      updateStatistics(endTime - startTime);
      pdxClassToFieldsMap.remove();
      pdxClassToMethodsMap.remove();
      ExecutionContext.isCanceled.remove();
      ((TXManagerImpl) this.cache.getCacheTransactionManager()).unpauseTransaction(tx);
    }
  }

  private QueryExecutor checkQueryOnPR(Object[] parameters)
      throws RegionNotFoundException, PartitionOfflineException {

    // check for PartitionedRegions. If a PartitionedRegion is referred to in the query,
    // then the following restrictions apply:
    // 1) the query must be just a SELECT expression; (preceded by zero or more IMPORT statements)
    // 2) the first FROM clause iterator cannot contain a subquery;
    // 3) PR reference can only be in the first FROM clause

    List<QueryExecutor> prs = new ArrayList<>();
    for (final Object o : getRegionsInQuery(parameters)) {
      String regionPath = (String) o;
      Region rgn = this.cache.getRegion(regionPath);
      if (rgn == null) {
        this.cache.getCancelCriterion().checkCancelInProgress(null);
        throw new RegionNotFoundException(
            String.format("Region not found: %s", regionPath));
      }
      if (rgn instanceof QueryExecutor) {
        ((PartitionedRegion) rgn).checkPROffline();
        prs.add((QueryExecutor) rgn);
      }
    }
    if (prs.size() == 1) {
      return prs.get(0);
    } else if (prs.size() > 1) {
      // colocation checks; valid for more the one PRs

      // First query has to be executed in a Function.
      if (!this.isQueryWithFunctionContext()) {
        throw new UnsupportedOperationException(
            String.format(
                "A query on a Partitioned Region ( %s ) may not reference any other region if query is NOT executed within a Function",
                prs.get(0).getName()));
      }

      // If there are more than one PRs they have to be co-located.
      QueryExecutor other = null;
      for (QueryExecutor eachPR : prs) {
        boolean colocated = false;

        for (QueryExecutor allPRs : prs) {
          if (eachPR == allPRs) {
            continue;
          }
          other = allPRs;
          if (((PartitionedRegion) eachPR).getColocatedByList().contains(allPRs)
              || ((PartitionedRegion) allPRs).getColocatedByList().contains(eachPR)) {
            colocated = true;
            break;
          }
        } // allPrs

        if (!colocated) {
          throw new UnsupportedOperationException(
              String.format(
                  "A query on a Partitioned Region ( %s ) may not reference any other region except Co-located Partitioned Region. PR region %s is not collocated with other PR region in the query.",
                  eachPR.getName(), other.getName()));
        }

      } // eachPR

      // this is a query on a PR, check to make sure it is only a SELECT
      CompiledSelect select = getSimpleSelect();
      if (select == null) {
        throw new UnsupportedOperationException(
            "query must be a simple select when referencing a Partitioned Region");
      }

      // make sure the where clause references no regions
      Set regions = new HashSet();
      CompiledValue whereClause = select.getWhereClause();
      if (whereClause != null) {
        whereClause.getRegionsInQuery(regions, parameters);
        if (!regions.isEmpty()) {
          throw new UnsupportedOperationException(
              "The WHERE clause cannot refer to a region when querying on a Partitioned Region");
        }
      }
      List fromClause = select.getIterators();

      // the first iterator in the FROM clause must be just a reference to the Partitioned Region
      Iterator fromClauseIterator = fromClause.iterator();
      CompiledIteratorDef itrDef = (CompiledIteratorDef) fromClauseIterator.next();

      // By process of elimination, we know that the first iterator contains a reference
      // to the PR. Check to make sure there are no subqueries in this first iterator
      itrDef.visitNodes(new CompiledValue.NodeVisitor() {
        @Override
        public boolean visit(CompiledValue node) {
          if (node instanceof CompiledSelect) {
            throw new UnsupportedOperationException(
                "When querying a PartitionedRegion, the first FROM clause iterator must not contain a subquery");
          }
          return true;
        }
      });

      // the rest of the FROM clause iterators must not reference any regions
      if (!this.isQueryWithFunctionContext()) {
        while (fromClauseIterator.hasNext()) {
          itrDef = (CompiledIteratorDef) fromClauseIterator.next();
          itrDef.getRegionsInQuery(regions, parameters);
          if (!regions.isEmpty()) {
            throw new UnsupportedOperationException(
                "When querying a Partitioned Region, the FROM clause iterators other than the first one must not reference any regions");
          }
        }

        // check the projections, must not reference any regions
        List projs = select.getProjectionAttributes();
        if (projs != null) {
          for (Object proj1 : projs) {
            Object[] rawProj = (Object[]) proj1;
            CompiledValue proj = (CompiledValue) rawProj[1];
            proj.getRegionsInQuery(regions, parameters);
            if (!regions.isEmpty()) {
              throw new UnsupportedOperationException(
                  "When querying a Partitioned Region, the projections must not reference any regions");
            }
          }
        }
        // check the orderByAttrs, must not reference any regions
        List<CompiledSortCriterion> orderBys = select.getOrderByAttrs();
        if (orderBys != null) {
          for (CompiledSortCriterion orderBy : orderBys) {
            orderBy.getRegionsInQuery(regions, parameters);
            if (!regions.isEmpty()) {
              throw new UnsupportedOperationException(
                  "When querying a Partitioned Region, the order-by attributes must not reference any regions");
            }
          }
        }
      }
      return prs.get(0); // PR query is okay
    }
    return null;
  }

  private void updateStatistics(long executionTime) {
    this.numExecutions.increment();
    this.totalExecutionTime.add(executionTime);
    this.cache.getCachePerfStats().endQueryExecution(executionTime);
  }

  // TODO: Implement the function. Toggle the isCompiled flag accordingly
  @Override
  public void compile() {
    throw new UnsupportedOperationException(
        "not yet implemented");
  }

  @Override
  public boolean isCompiled() {
    return false;
  }

  public boolean isTraced() {
    return this.traceOn;
  }

  class DefaultQueryStatistics implements QueryStatistics {

    /**
     * Returns the total amount of time (in nanoseconds) spent executing the query.
     */
    @Override
    public long getTotalExecutionTime() {
      return DefaultQuery.this.totalExecutionTime.longValue();
    }

    /**
     * Returns the total number of times the query has been executed.
     */
    @Override
    public long getNumExecutions() {
      return DefaultQuery.this.numExecutions.longValue();
    }
  }

  /**
   * Returns an unmodifiable Set containing the Region names which are present in the Query. A
   * region which is associated with the query as a bind parameter will not be included in the list.
   * The Region names returned by the query do not indicate anything about the state of the region
   * or whether the region actually exists in the GemfireCache etc.
   *
   * @param parameters the parameters to be passed in to the query when executed
   * @return Unmodifiable List containing the region names.
   */
  public Set<String> getRegionsInQuery(Object[] parameters) {
    Set<String> regions = new HashSet<>();
    this.compiledQuery.getRegionsInQuery(regions, parameters);
    return Collections.unmodifiableSet(regions);
  }

  /**
   * Returns the CompiledSelect if this query consists of only a SELECT expression (possibly with
   * IMPORTS as well). Otherwise, returns null
   */
  public CompiledSelect getSimpleSelect() {
    if (this.compiledQuery instanceof CompiledSelect) {
      return (CompiledSelect) this.compiledQuery;
    }
    return null;
  }

  public CompiledSelect getSelect() {
    return (CompiledSelect) this.compiledQuery;
  }

  /**
   * @return int identifying the limit. A value of -1 indicates that no limit is imposed or the
   *         query is not a select query
   */
  public int getLimit(Object[] bindArguments) throws FunctionDomainException, TypeMismatchException,
      NameResolutionException, QueryInvocationTargetException {
    return this.compiledQuery instanceof CompiledSelect
        ? ((CompiledSelect) this.compiledQuery).getLimitValue(bindArguments) : -1;
  }

  void setServerProxy(ServerProxy serverProxy) {
    this.serverProxy = serverProxy;
  }

  public void setIsCqQuery(boolean isCqQuery) {
    this.isCqQuery = isCqQuery;
  }

  public boolean isCqQuery() {
    return this.isCqQuery;
  }

  public void setCqQuery(InternalCqQuery cqQuery) {
    this.cqQuery = cqQuery;
  }

  public void setLastUsed(boolean lastUsed) {
    this.lastUsed = lastUsed;
  }

  public boolean getLastUsed() {
    return this.lastUsed;
  }

  public InternalCqQuery getCqQuery() {
    return this.cqQuery;
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder("Query String = ");
    sb.append(this.queryString);
    sb.append("; Total Executions = ");
    sb.append(this.numExecutions);
    sb.append("; Total Execution Time = ");
    sb.append(this.totalExecutionTime);
    return sb.toString();
  }

  void setProxyCache(ProxyCache proxyCache) {
    this.proxyCache = proxyCache;
  }

  /**
   * Used for test purpose.
   */
  public static void setTestCompiledQueryClearTime(int val) {
    DefaultQuery.TEST_COMPILED_QUERY_CLEAR_TIME = val;
  }

  private static String getLogMessage(QueryObserver observer, long startTime, int resultSize,
      String query) {
    float time = (NanoTimer.getTime() - startTime) / 1.0e6f;

    String usedIndexesString = null;
    if (observer instanceof IndexTrackingQueryObserver) {
      IndexTrackingQueryObserver indexObserver = (IndexTrackingQueryObserver) observer;
      Map usedIndexes = indexObserver.getUsedIndexes();
      indexObserver.reset();
      StringBuilder sb = new StringBuilder();
      sb.append(" indexesUsed(");
      sb.append(usedIndexes.size());
      sb.append(')');
      if (usedIndexes.size() > 0) {
        sb.append(':');
        for (Iterator itr = usedIndexes.entrySet().iterator(); itr.hasNext();) {
          Map.Entry entry = (Map.Entry) itr.next();
          sb.append(entry.getKey()).append(entry.getValue());
          if (itr.hasNext()) {
            sb.append(',');
          }
        }
      }
      usedIndexesString = sb.toString();
    } else if (DefaultQuery.QUERY_VERBOSE) {
      usedIndexesString = " indexesUsed(NA due to other observer in the way: "
          + observer.getClass().getName() + ')';
    }

    String rowCountString = null;
    if (resultSize != -1) {
      rowCountString = " rowCount = " + resultSize + ';';
    }
    return "Query Executed in " + time + " ms;" + (rowCountString != null ? rowCountString : "")
        + (usedIndexesString != null ? usedIndexesString : "") + " \"" + query + '"';
  }

  private static String getLogMessage(IndexTrackingQueryObserver indexObserver, long startTime,
      String otherObserver, int resultSize, String query, BucketRegion bucket) {
    float time = 0.0f;

    if (startTime > 0L) {
      time = (NanoTimer.getTime() - startTime) / 1.0e6f;
    }

    String usedIndexesString = null;
    if (indexObserver != null) {
      Map usedIndexes = indexObserver.getUsedIndexes(bucket.getFullPath());
      StringBuilder sb = new StringBuilder();
      sb.append(" indexesUsed(");
      sb.append(usedIndexes.size());
      sb.append(')');
      if (!usedIndexes.isEmpty()) {
        sb.append(':');
        for (Iterator itr = usedIndexes.entrySet().iterator(); itr.hasNext();) {
          Map.Entry entry = (Map.Entry) itr.next();
          sb.append(entry.getKey()).append("(Results: ").append(entry.getValue())
              .append(", Bucket: ").append(bucket.getId()).append(")");
          if (itr.hasNext()) {
            sb.append(',');
          }
        }
      }
      usedIndexesString = sb.toString();
    } else if (DefaultQuery.QUERY_VERBOSE) {
      usedIndexesString =
          " indexesUsed(NA due to other observer in the way: " + otherObserver + ')';
    }

    String rowCountString = " rowCount = " + resultSize + ';';
    return "Query Executed" + (startTime > 0L ? " in " + time + " ms;" : ";") + rowCountString
        + (usedIndexesString != null ? usedIndexesString : "") + " \"" + query + '"';
  }

  @Override
  public Object execute(RegionFunctionContext context) throws FunctionDomainException,
      TypeMismatchException, NameResolutionException, QueryInvocationTargetException {
    return execute(context, EMPTY_ARRAY);
  }

  @Override
  public Object execute(RegionFunctionContext context, Object[] params)
      throws FunctionDomainException, TypeMismatchException, NameResolutionException,
      QueryInvocationTargetException {

    // Supported only with RegionFunctionContext
    if (context == null) {
      throw new IllegalArgumentException(
          "'Function Context' cannot be null");
    }
    this.isQueryWithFunctionContext = true;

    if (params == null) {
      throw new IllegalArgumentException(
          "'parameters' cannot be null");
    }

    long startTime = 0L;
    if (this.traceOn && this.cache != null) {
      startTime = NanoTimer.getTime();
    }

    QueryObserver indexObserver = null;
    QueryExecutor qe = checkQueryOnPR(params);

    Object result = null;
    try {
      indexObserver = startTrace();
      if (qe != null) {
        LocalDataSet localDataSet =
            (LocalDataSet) PartitionRegionHelper.getLocalDataForContext(context);
        Set<Integer> buckets = localDataSet.getBucketSet();
        final ExecutionContext executionContext = new ExecutionContext(null, cache);
        result = qe.executeQuery(this, executionContext, params, buckets);
        return result;
      } else {
        // Not supported on regions other than PartitionRegion.
        throw new IllegalArgumentException(
            "This query API can only be used for Partition Region Queries.");
      }

    } finally {
      this.endTrace(indexObserver, startTime, result);
    }
  }

  /**
   * For queries which are executed from a Function "with a Filter".
   *
   * @return returns if this query is coming from a {@link Function}.
   */
  public boolean isQueryWithFunctionContext() {
    return this.isQueryWithFunctionContext;
  }

  public QueryObserver startTrace() {
    QueryObserver queryObserver = null;
    if (this.traceOn && this.cache != null) {

      QueryObserver qo = QueryObserverHolder.getInstance();
      if (qo instanceof IndexTrackingQueryObserver) {
        queryObserver = qo;
      } else if (!QueryObserverHolder.hasObserver()) {
        queryObserver = new IndexTrackingQueryObserver();
        QueryObserverHolder.setInstance(queryObserver);
      } else {
        queryObserver = qo;
      }
    }
    return queryObserver;
  }

  public void endTrace(QueryObserver indexObserver, long startTime, Object result) {
    if (this.traceOn && this.cache != null) {
      int resultSize = -1;

      if (result instanceof Collection) {
        resultSize = ((Collection) result).size();
      }

      String queryVerboseMsg =
          DefaultQuery.getLogMessage(indexObserver, startTime, resultSize, this.queryString);
      this.cache.getLogger().info(queryVerboseMsg);
    }
  }

  public void endTrace(QueryObserver indexObserver, long startTime, Collection<Collection> result) {
    if (this.cache != null && this.cache.getLogger().infoEnabled() && this.traceOn) {
      int resultSize = 0;

      for (Collection aResult : result) {
        resultSize += aResult.size();
      }

      String queryVerboseMsg =
          DefaultQuery.getLogMessage(indexObserver, startTime, resultSize, this.queryString);
      if (this.cache.getLogger().infoEnabled()) {
        this.cache.getLogger().info(queryVerboseMsg);
      }
    }
  }

  public boolean isRemoteQuery() {
    return this.isRemoteQuery;
  }

  public void setRemoteQuery(boolean isRemoteQuery) {
    this.isRemoteQuery = isRemoteQuery;
  }

  /**
   * set keepSerialized flag for remote queries of type 'select *' having independent operators
   */
  void keepResultsSerialized(CompiledSelect cs, ExecutionContext context) {
    if (isRemoteQuery()) {
      // for dependent iterators, deserialization is required
      if (cs.getIterators().size() == context.getAllIndependentIteratorsOfCurrentScope().size()
          && cs.getWhereClause() == null && cs.getProjectionAttributes() == null && !cs.isDistinct()
          && cs.getOrderByAttrs() == null) {
        setKeepSerialized();
      }
    }
  }

  public boolean isKeepSerialized() {
    return this.keepSerialized;
  }

  private void setKeepSerialized() {
    this.keepSerialized = true;
  }

  /**
   * Test logic sets DefaultQuery.testHook to an implementation of this interface,
   * to facilitate white-box testing.
   *
   * DefaultQuery and other classes in query* packages invoke doTestHook() at various points
   * during query processing--identifying the location with a SPOT value.
   */
  @FunctionalInterface
  public interface TestHook {
    enum SPOTS {

      /*
       * These spots pass a DefaultQuery
       */
      BEFORE_QUERY_EXECUTION, /* was 1 */
      BEFORE_QUERY_DEPENDENCY_COMPUTATION, /* was 6 */

      /*
       * These spots do not pass a DefaultQuery
       */
      LOW_MEMORY_WHEN_DESERIALIZING_STREAMINGOPERATION, /* was 2 */
      BEFORE_ADD_OR_UPDATE_MAPPING_OR_DESERIALIZING_NTH_STREAMINGOPERATION, /* was 3 */
      BEFORE_BUILD_CUMULATIVE_RESULT, /* was 4 */
      BEFORE_THROW_QUERY_CANCELED_EXCEPTION, /* was 5 */
      BEGIN_TRANSITION_FROM_REGION_ENTRY_TO_ELEMARRAY,
      TRANSITIONED_FROM_REGION_ENTRY_TO_ELEMARRAY,
      COMPLETE_TRANSITION_FROM_REGION_ENTRY_TO_ELEMARRAY,
      BEGIN_TRANSITION_FROM_ELEMARRAY_TO_CONCURRENT_HASH_SET,
      TRANSITIONED_FROM_ELEMARRAY_TO_TOKEN,
      COMPLETE_TRANSITION_FROM_ELEMARRAY_TO_CONCURRENT_HASH_SET,
      ATTEMPT_REMOVE,
      ATTEMPT_RETRY,
      BEGIN_REMOVE_FROM_ELEM_ARRAY,
      REMOVE_CALLED_FROM_ELEM_ARRAY,
      COMPLETE_REMOVE_FROM_ELEM_ARRAY,
      PULL_OFF_PR_QUERY_TRACE_INFO,
      CREATE_PR_QUERY_TRACE_STRING,
      CREATE_PR_QUERY_TRACE_INFO_FROM_LOCAL_NODE,
      CREATE_PR_QUERY_TRACE_INFO_FOR_REMOTE_QUERY,
      POPULATING_TRACE_INFO_FOR_REMOTE_QUERY
    };

    /**
     * Called (for side-effects) at various points in query processing, to facilitate
     * white-box testing.
     *
     * @param spot identifies the (logical) calling code location. Some SPOT values represent
     *        more than one physical location in the query processing code.
     * @param query nullable, DefaultQuery, for SPOTS in the DefaultQuery class
     */
    void doTestHook(SPOTS spot, DefaultQuery query,
        ExecutionContext executionContext);
  }

}