class PartitionedRegionQueryEvaluator extends StreamingPartitionOperation {

  public SelectResults queryBuckets(final TestHook th) throws QueryException, InterruptedException {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    if (Thread.interrupted()) {
      throw new InterruptedException();
    }
    if (isDebugEnabled) {
      logger.debug("PRQE query :{}", this.query.getQueryString());
    }
    Assert.assertTrue(!(this.bucketsToQuery == null || this.bucketsToQuery.isEmpty()),
        "bucket set is empty.");
    this.node2bucketIds = buildNodeToBucketMap();
    Assert.assertTrue(!this.node2bucketIds.isEmpty(),
        " There are no data stores hosting any of the buckets.");

    boolean needsRetry = true;
    int retry = 0;
    while (needsRetry && retry < MAX_PR_QUERY_RETRIES) {
      // Shobhit: Now on if buckets to be queried are on remote as well as local node,
      // request will be sent to remote node first to run query in parallel on local and
      // remote node.
      // Note: if Any Exception is thrown on local and some remote node, local exception
      // will be given priority and will be visible to query executor and remote exception
      // will be swallowed.
      needsRetry = executeQueryOnRemoteAndLocalNodes(th);
      if (th != null) {
        th.hook(1);
      }

      if (needsRetry) {
        // Shobhit: Only one chance is allowed for Function queries.
        if (query.isQueryWithFunctionContext()) {
          if (isDebugEnabled) {
            logger.debug("No of retry attempts are: {}", retry);
          }
          break;
        }
        Map b2n = buildNodeToBucketMapForBuckets(calculateRetryBuckets());
        if (th != null) {
          th.hook(2);
        }
        this.node2bucketIds = b2n;
        if (isDebugEnabled) {
          logger.debug("PR Query retry: {} total: {}", retry,
              this.pr.getCachePerfStats().getPRQueryRetries());
        }
        this.pr.getCachePerfStats().incPRQueryRetries();
        retry++;
        // Shobhit: Wait for sometime as rebalancing might be happening
        waitBeforeRetry();
      }
      if (th != null) {
        th.hook(3);
      }
    }

    if (needsRetry) {
      String msg = "Failed to query all the partitioned region " + "dataset (buckets) after "
          + retry + " attempts.";

      if (isDebugEnabled) {
        logger.debug("{} Unable to query some of the buckets from the set :{}", msg,
            this.calculateRetryBuckets());
      }
      throw new QueryException(msg);
    }

    return addResultsToResultSet();
  }

  public  PartitionedRegionQueryEvaluator(final InternalDistributedSystem sys,
      final PartitionedRegion pr,
      final DefaultQuery query,
      final ExecutionContext executionContext,
      final Object[] parameters,
      final SelectResults cumulativeResults,
      final Set<Integer> bucketsToQuery);

}