Fix computation of QueuedQueries JMX metrics

[imjalpreet]

Fixes #19929

Test plan

Validated that JMX metrics and Presto Console have the same values for Queued Query post this change.

The motivation behind the change: Due to the discrepancy between the count on UI and JMX metric, this PR tries to modify the way the JMX metric is being computed to make it consistent with the value displayed on Presto UI.

== RELEASE NOTES ==

General Changes
* Fix Queued Query Count JMX Metrics (:issue:`19929`). Expose the QueryManagerStats metrics under the ``com.facebook.presto.dispatcher:name=DispatchManager`` namespace.

[tdcmeehan]

1. I do not think we should remove the metrics from QueryManagerStats, as these are almost certainly being relied upon by many deployments of Presto.
2. If what we're trying to do is separate out queries which are submitted, but not yet queued in a resource group, let's do that by making the appropriate code changes in QueryManagerStats and usages. Let's move the counter update in that class to happen close to when it's queued inside the resource group, for example.

[imjalpreet]

Thank you for your comments @tdcmeehan I have tried to express my views and observations below.

1. I do not think we should remove the metrics from QueryManagerStats, as these are almost certainly being relied upon by many deployments of Presto.

I have tried to keep the change transparent to users by exposing the metrics under the same namespace of QueryManager. Within the code, I tried to take care of any usages of QueryManagerStats but they were only being used in a couple of tests. Please let me know in case I missed something.

2. If what we're trying to do is separate out queries which are submitted, but not yet queued in a resource group, let's do that by making the appropriate code changes in QueryManagerStats and usages. Let's move the counter update in that class to happen close to when it's queued inside the resource group, for example.

The aim of this change was to fix the issue(#19929) mentioned in the description. I have tried to replicate the logic used to display the Queued Query Count and Running Query Count on Presto Console. One of the things I noticed, in the older computation, a queued query count is only decremented and the running query count is incremented when a query is transitioned into Running State which meant all the previous states were considered as Queued. So, to make the counts consistent with Presto console the changes in the PR have been made.

Please let me know if you think we can fix this in a better way.

[yingsu00]

As far as I understood, I think the problem was that the stats was not updated at right location:

• DispatchManager.createQueryInternal() creates a LocalDispatchQuery for each submitted query, and keep them in its member queryTracker.

• Each LocalDispatchQuery has a QueryStateMachine to track the query states (QUEUED, WAITING_FOR_RESOURCES, etc.). The QUEUED state was set when the DispatchManager is trying to queue a query.

• The DispatchManager has a QueryManagerStats object stats , and its queuedQueries was incremented after the dispatchQuery is added to DispatchManager's queryTracker.

• The coordinator UI gets the #queued queries by enumerating all queries in the DispatchManager and check their QueryStateMachine objects to see if it's queued: ClusterStatsResource.getClusterStats ,

• SqlQueryManager has a QueryManagerStats object stats, but this is another stats object than the one in DispatchManager. SqlQueryManager.stats.queuedQueries got incremented in QueryManager.createQuery(). This is AFTER the min worker requirement is met, and when the query starts execution.

So it seems the QueryManagerStats.queuedQueries was not updated correctly. It should be incremented when the stateMachine for the localDispatchQuery enters the QUEUED state. See the stacks below:

			DispatchManager.createQueryInternal(queryId, resourceGroupManager)
				...				
				dispatchQuery = dispatchQueryFactory.createDispatchQuery(…(dq) -> resourceGroupManager.submit(dq, selectionContext, queryExecutor)).  
						
				boolean queryAdded = queryCreated(dispatchQuery);
					queryAdded = queryTracker.addQuery(dispatchQuery);
						queries.putIfAbsent(dispatchQuery.getQueryId(), dispatchQuery)
					If queryAdded
						stats.trackQueryStats(dispatchQuery);    // This is DispatchManager.stats
							submittedQueries.update(1);
							queuedQueries.incrementAndGet();
				clusterStatusSender.registerQuery(dispatchQuery);
				dispatchQuery.startWaitingForPrerequisites();
				prerequisitesFuture = queryPrerequisites.waitForPrerequisites
				LocalDispatchQuery.queueQuery
					stateMachine.transitionToQueued().  // transitioned to QUEUED state. The stats should be updated here
					queryQueuer.accept(this);
						resourceGroupManager.submit(dq, selectionContext, queryExecutor). 
							groups.get(selectionContext.getResourceGroupId()).run(queryExecution);
								InternalResourceGroup.run(ManagedQueryExecution query)
									canQueue &= group.canQueueMore();
									canRun &= group.canRunMore();
									if (canRun && queuedQueries.isEmpty()) {
										startInBackground(query);
											runningQueries.add(query);
											executor.execute(query::startWaitingForResources);  
												query::startWaitingForResources
													if (stateMachine.transitionToWaitingForResources()) {
													            waitForMinimumWorkers();
														// when worker requirement is met, wait for query execution to finish construction and then start the execution
														boolean isDispatching = stateMachine.transitionToDispatching();
														queryExecution -> startExecution(queryExecution, isDispatching)
															LocalDispatchQuery.startExecution(QueryExecution queryExecution, boolean isDispatching)
																querySubmitter.accept(queryExecution);
																	QueryManager::createQuery. 
																		queryTracker.addQuery(queryExecution)
																		stats.trackQueryStats(queryExecution);    // This is SqlQueryManager.stats
																			submittedQueries.update(1);
																			queuedQueries.incrementAndGet();   // queuedQueries incremented here, but the query is not in QUEUED state actually
																		queryExecution::start

I think we should not remove the queuedQueries and runningQueries from QueryManagerStats, but instead, just update it at the right location. I didn't check the decrement callsites, but similarly, it should be updated at the same location that the stateMachine shifted from QUEUED to other states.

[imjalpreet]

@yingsu00 @tdcmeehan thanks for your review, I have made the changes based on your suggestions.

Can you please have a look and let me know in case of any questions?

[imjalpreet]

For now, I have exposed the DispatchManager QueryManagerStats in a separate namespace. But the stats object exposed via this class has more accurate stats(compared to SqlQueryManager) for all metrics including Queued Query and Running Query count.

If we decide to expose only one of them, I would recommend we should expose the DispatchManager stats object. To make the change backwards compatible we can expose it under the same namespace as before so it remains transparent to the users.

[yingsu00]

@imjalpreet Will you be able to add tests to make sure the dispatchQueries' stateMachines match with the QueryManagerStats stats in both SqlQueryManager and DispatchManager?

[yingsu00]

I actually think trackQueryStats() methods in QueryManagerStats should NOT do queuedQueries.incrementAndGet();. Having a query submitted doesn't mean it's already queued. @tdcmeehan Do you know why they were done this way?

[imjalpreet]

I actually think trackQueryStats() methods in QueryManagerStats should NOT do queuedQueries.incrementAndGet();. Having a query submitted doesn't mean it's already queued.

I agree, that doesn't look right for both the methods. In the case of DispatchQuery, it is queued after the call to trackQueryStats, and in the case of SqlQueryManager, the query is not even in the queued state when trackQueryStats is called. It is called after waiting for resources finishes and query execution is about to start.

[yingsu00]

Correct me if I'm wrong, but where is the logic of updating queuedQueries when the new state is QUEUED? In such case the newValue.ordinal() is equal to QUEUED.ordinal(), also the newValue.isDone() is false. Which if branch handles this situation?

[yingsu00]

Should we introduce queryQueued() method that increment queuedQueries count, and add a branch in this method if the new state is QUEUED?

[imjalpreet]

where is the logic of updating queuedQueries when the new state is QUEUED?

In the current implementation, this was not required since every call to trackQueryStats was incrementing queuedQueries.

Should we introduce queryQueued() method that increment queuedQueries count, and add a branch in this method if the new state is QUEUED?

But yes, I think this looks better, I just saw that query is queued after trackQueryStats(dispatchQuery). So we should not be incrementing the queuedQueries in trackQueryStats.

[yingsu00]

In the current implementation, this was not required since every call to trackQueryStats was incrementing queuedQueries.

As far as I can see, trackQueryStats() was only setting up the initial state of the stats. See where it is called. It's only called after queryCreated() was successful. Also it should NOT be increasing the queuedQueries as I mentioned above.

[imjalpreet]

As far as I can see, trackQueryStats() was only setting up the initial state of the stats. See where it is called. It's only called after queryCreated() was successful. Also it should NOT be increasing the queuedQueries as I mentioned above.

Yes, I agree. But I was talking about the current implementation in master. Since in that implementation it was already incrementing queuedQueries in trackQueryStats, I am assuming that was the reason there were no checks for Queued State after that.

I am currently working on making the corrections including not incrementing queuedQueries in trackQueryStats as I mentioned earlier.

[imjalpreet]

Key observations and changes:

1. We observed that queued query count was being incremented in the com.facebook.presto.execution.QueryManagerStats.trackQueryStats method. Queued Query Count should not be incremented at this stage since the Query is not in Queued State when this method is called. In the case of DispatchQuery, the query is yet to be queued and in the case of QueryExecution query has already passed Queued state. This PR stops incrementing the queued query count in this method call.
2. This PR introduces tracking of Queued QueryState in com.facebook.presto.execution.QueryManagerStats.StatisticsListener to increment Queued Query Count based on state change. This ensures that the QueryManagerStats Query Count metric is consistent with QueryStateMachine changes.
3. This PR also exposes the QueryManagerStats under the namespace of com.facebook.presto.dispatcher:name=DispatchManager. This change has been done because only DispatchManagerquery stats account for all the QueryState changes. The stats object part of the SqlQueryManager class cannot account for all QueryState changes including Queued State since these stats only start tracking once the Query is about to begin execution (com.facebook.presto.dispatcher.LocalDispatchQuery#startExecution --> com.facebook.presto.execution.SqlQueryManager#createQuery)

@yingsu00 @tdcmeehan I have updated the PR taking into account all concerns and recommendations. Can you please have a look and let me know in case of any questions.

[yingsu00]

Mostly look good, some nits in addition to the test comment:

Could you please make the commit titles in 50 chars, and commit message lines within 72 chars?

[yingsu00]

This test only tests the stats at one snapshot. Can we enhance it to multiple checks before all queries finish such that all the states match? Something like

while (stats.queuedQueries + stats.runningQueries != 0) {
    assert queuedQueries match;
    assert runningQueries match;
    sleep(x);
}

[imjalpreet]

@yingsu00 I have updated the PR, please have a look and let me know in case anything is missed.

[yingsu00]

Don't have time to read the code further, is the queue size 1 ? How many queries will be running after the queries were created and before waiting? I think the comment is a bit confusing here and wonder if you could clarify why 3 guarantees queueing.

[imjalpreet]

I have updated the comment and the reason queueing is guaranteed with > 3 queries is because max concurrent queries are set to 3 via resource groups.

[yingsu00]

How was 25 chosen? Is it possible the query runs faster than 25msec and this loop never ends?

[imjalpreet]

With reference to other tests in this class and also verified by running the tests, the query will not finish faster than 25 ms

[yingsu00]

Thanks for updating the test. But this only tests 1 queued query. Do you think creating a few more of them would make the tests more meaningful? Something like

create r(=3) queries;
wait until all r queries are running;

create q queries;
assert q queries are queued;
while (dispatchManager.getStats().getQueuedQueries() + dispatchManager.getStats().getRunningQueries() > 0
                && stopwatch.elapsed().toMillis() < 60000) {
    numQueued = count queuedQueries from dispatchManager;
    assert numQueued == stats.queuedQueries;
    assert numQueued <= numQueued of last round 
 
    numRunning = count runningQueries from dispatchManager;
    assert numRunning == stats.runningQueries;
    assert numRunning <= numRunning from last round
    assert numRunning <= r;

    sleep(x); // x be a small number
}

[imjalpreet]

As discussed offline, I have updated the tests accordingly.

[imjalpreet]

@yingsu00 I have updated the test as we discussed offline.

The reason we are only asserting the exact count of queued queries and running queries outside the loop is that flakiness was observed while asserting them over some time in the loop. Flakiness was because queries were changing states while we were trying to assert the counts.

Now within the loop, we are trying to assert that the count of queued queries is decreasing over time and running queries are always <= 3

[tdcmeehan]

LGTM % the suggested simplification to the stateChanged method

[tdcmeehan]

I find the following easier to read:

        @Override
        public void stateChanged(QueryState newValue)
        {
            synchronized (this) {
                if (stopped) {
                    return;
                }

                if (newValue.isDone()) {
                    stopped = true;
                    if (started) {
                        queryStopped();
                    }
                    else if (queued) {
                        queryDequeued();
                    }
                    finalQueryInfoSupplier.get()
                            .ifPresent(QueryManagerStats.this::queryFinished);
                    return;
                }

                if (newValue.ordinal() == QUEUED.ordinal()) {
                    if (!queued) {
                        queued = true;
                        queryQueued();
                    }
                }
                else if (newValue.ordinal() > QUEUED.ordinal()) {
                    if (queued) {
                        queryDequeued();
                        queued = false;
                    }
                    if (newValue.ordinal() >= RUNNING.ordinal()) {
                        if (!started) {
                            started = true;
                            queryStarted();
                        }
                    }
                }
            }
        }
    }
    ```

[imjalpreet]

@tdcmeehan Thanks for the review, I have made the change.