@Override
  protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm, PartitionedRegion pr,
      long startTime)
      throws CacheException, QueryException, ForceReattemptException, InterruptedException {
    // calculate trace start time if trace is on this is because the start time is only set if
    // enableClock stats is on in this case we still want to see trace time even if clock is not
    // enabled
    long traceStartTime = 0;
    if (traceOn) {
      traceStartTime = NanoTimer.getTime();
    }
    if (Thread.interrupted()) {
      throw new InterruptedException();
    }
    if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
      logger.trace(LogMarker.DM_VERBOSE, "QueryMessage operateOnPartitionedRegion: {} buckets {}",
          pr.getFullPath(), buckets);
    }

    pr.waitOnInitialization();

    if (QueryMonitor.isLowMemory()) {
      String reason = String.format(
          "Query execution canceled due to memory threshold crossed in system, memory used: %s bytes.",
          QueryMonitor.getMemoryUsedBytes());
      // throw query exception to piggyback on existing error handling as qp.executeQuery also
      // throws the same error for low memory
      throw new QueryExecutionLowMemoryException(reason);
    }

    DefaultQuery query = new DefaultQuery(queryString, pr.getCache(), false);
    final ExecutionContext executionContext = new QueryExecutionContext(null, pr.getCache(), query);
    // Remote query, use the PDX types in serialized form.
    Boolean initialPdxReadSerialized = pr.getCache().getPdxReadSerializedOverride();
    pr.getCache().setPdxReadSerializedOverride(true);
    // In case of "select *" queries we can keep the results in serialized form and send
    query.setRemoteQuery(true);
    QueryObserver indexObserver = query.startTrace();
    boolean isQueryTraced = false;
    List queryTraceList = null;

    try {
      query.setIsCqQuery(cqQuery);
      PRQueryProcessor qp = new PRQueryProcessor(pr, query, parameters, buckets);
      if (logger.isDebugEnabled()) {
        logger.debug("Started executing query from remote node: {}", query.getQueryString());
      }
      isQueryTraced =
          query.isTraced()
              && sender.getVersion().isNotOlderThan(KnownVersion.GFE_81);

      // Adds a query trace info object to the results list for remote queries
      PRQueryTraceInfo queryTraceInfo = null;
      if (isQueryTraced) {
        isTraceInfoIteration = true;
        if (DefaultQuery.testHook != null) {
          DefaultQuery.testHook
              .doTestHook(DefaultQuery.TestHook.SPOTS.CREATE_PR_QUERY_TRACE_INFO_FOR_REMOTE_QUERY,
                  null, null);
        }
        queryTraceInfo = new PRQueryTraceInfo();
        queryTraceList = Collections.singletonList(queryTraceInfo);
      }

      isStructType = qp.executeQuery(resultCollector);
      // Add the trace info list object after the NWayMergeResults is created so as to exclude it
      // from the sorted collection of NWayMergeResults
      if (isQueryTraced) {
        resultCollector.add(0, queryTraceList);
      }
      currentSelectResultIterator = resultCollector.iterator();

      // If trace is enabled, we will generate a trace object to send back. The time info will be
      // slightly different than the one logged on this node due to generating the trace object
      // information here rather than the finally block.
      if (isQueryTraced) {
        if (DefaultQuery.testHook != null) {
          DefaultQuery.testHook
              .doTestHook(DefaultQuery.TestHook.SPOTS.POPULATING_TRACE_INFO_FOR_REMOTE_QUERY, null,
                  null);
        }

        // calculate the number of rows being sent
        int traceSize = queryTraceInfo.calculateNumberOfResults(resultCollector);
        // subtract the query trace info object
        traceSize -= 1;
        queryTraceInfo.setTimeInMillis((NanoTimer.getTime() - traceStartTime) / 1.0e6f);
        queryTraceInfo.setNumResults(traceSize);

        // created the indexes used string
        if (indexObserver instanceof IndexTrackingQueryObserver) {
          Map indexesUsed = ((IndexTrackingQueryObserver) indexObserver).getUsedIndexes();
          StringBuilder sb = new StringBuilder();
          sb.append(" indexesUsed(").append(indexesUsed.size()).append(")");
          if (indexesUsed.size() > 0) {
            sb.append(":");
            for (Iterator itr = indexesUsed.entrySet().iterator(); itr.hasNext();) {
              Map.Entry entry = (Map.Entry) itr.next();
              sb.append(entry.getKey()).append(entry.getValue());
              if (itr.hasNext()) {
                sb.append(",");
              }
            }
          }
          queryTraceInfo.setIndexesUsed(sb.toString());
        }
      }

      if (QueryMonitor.isLowMemory()) {
        String reason = String.format(
            "Query execution canceled due to memory threshold crossed in system, memory used: %s bytes.",
            QueryMonitor.getMemoryUsedBytes());
        throw new QueryExecutionLowMemoryException(reason);
      } else if (executionContext.isCanceled()) {
        throw executionContext.getQueryCanceledException();
      }
      super.operateOnPartitionedRegion(dm, pr, startTime);
    } finally {
      // remove trace info so that it is not included in the num results when logged
      if (isQueryTraced) {
        resultCollector.remove(queryTraceList);
      }
      pr.getCache().setPdxReadSerializedOverride(initialPdxReadSerialized);
      query.setRemoteQuery(false);
      query.endTrace(indexObserver, traceStartTime, resultCollector);
    }

    // Unless there was an exception thrown, this message handles sending the response
    return false;
  }
Provide results to send back to requestor. terminate by returning END_OF_STREAM token object
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/QueryMessage.java