@Override
  public ScheduledQueryPollResponse scheduledQueryPoll(ScheduledQueryPollRequest request) throws MetaException {
    ensureScheduledQueriesEnabled();
    boolean commited = false;
    ScheduledQueryPollResponse ret = new ScheduledQueryPollResponse();
    Query q = null;
    try {
      openTransaction();
      q = pm.newQuery(MScheduledQuery.class,
          "nextExecution <= now && enabled && clusterNamespace == ns && activeExecution == null");
      q.setSerializeRead(true);
      q.declareParameters("java.lang.Integer now, java.lang.String ns");
      q.setOrdering("nextExecution");
      int now = (int) (System.currentTimeMillis() / 1000);
      List<MScheduledQuery> results = (List<MScheduledQuery>) q.execute(now, request.getClusterNamespace());
      if (results == null || results.isEmpty()) {
        return new ScheduledQueryPollResponse();
      }
      MScheduledQuery schq = results.get(0);
      schq.setNextExecution(computeNextExecutionTime(schq.getSchedule()));

      MScheduledExecution execution = new MScheduledExecution();
      execution.setScheduledQuery(schq);
      execution.setState(QueryState.INITED);
      execution.setStartTime(now);
      execution.setLastUpdateTime(now);
      schq.setActiveExecution(execution);
      pm.makePersistent(execution);
      pm.makePersistent(schq);
      ObjectStoreTestHook.onScheduledQueryPoll();
      commited = commitTransaction();
      ret.setScheduleKey(schq.getScheduleKey());
      ret.setQuery("/* schedule: " + schq.getScheduleName() + " */" + schq.getQuery());
      ret.setUser(schq.getUser());
      int executionId = ((IntIdentity) pm.getObjectId(execution)).getKey();
      ret.setExecutionId(executionId);
    } catch (JDOException e) {
      LOG.debug("Caught jdo exception; exclusive", e);
      commited = false;
    } finally {
      rollbackAndCleanup(commited, q);
      return commited ? ret : new ScheduledQueryPollResponse();
    }
  }
Return true if the current statistics in the Metastore is valid
for the query of the given "txnId" and "queryValidWriteIdList".

Note that a statistics entity is valid iff
the stats is written by the current query or
the conjunction of the following two are true:
~ COLUMN_STATE_ACCURATE(CSA) state is true
~ Isolation-level (snapshot) compliant with the query
@param part                   MPartition of the stats entity
@param queryValidWriteIdList  valid writeId list of the query
@Precondition   "part" should be retrieved from the PARTITIONS table.
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java