Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/1051#discussion_r155938213
--- Diff:
exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java ---
@@ -502,33 +438,82 @@ private void runFragment(List<PlanFragment>
fragmentsList) throws ExecutionSetup
}
}
+ assert rootFragment != null;
+
final FragmentRoot rootOperator;
try {
rootOperator =
drillbitContext.getPlanReader().readFragmentRoot(rootFragment.getFragmentJson());
} catch (IOException e) {
throw new ExecutionSetupException(String.format("Unable to parse
FragmentRoot from fragment: %s", rootFragment.getFragmentJson()));
}
queryRM.setCost(rootOperator.getCost());
- admit(null);
- drillbitContext.getWorkBus().addFragmentStatusListener(queryId,
queryManager.getFragmentStatusListener());
-
drillbitContext.getClusterCoordinator().addDrillbitStatusListener(queryManager.getDrillbitStatusListener());
- logger.debug("Submitting fragments to run.");
+ fragmentsRunner.setFragmentsInfo(planFragments, rootFragment,
rootOperator);
- // set up the root fragment first so we'll have incoming buffers
available.
- setupRootFragment(rootFragment, rootOperator);
+ startQueryProcessing();
+ }
- setupNonRootFragments(planFragments);
+ /**
+ * Enqueues the query and once enqueued, starts sending out query
fragments for further execution.
+ * Moves query to RUNNING state.
+ */
+ private void startQueryProcessing() {
+ enqueue();
+ runFragments();
+ queryStateProcessor.moveToState(QueryState.RUNNING, null);
+ }
- moveToState(QueryState.RUNNING, null);
- logger.debug("Fragments running.");
+ /**
+ * Move query to ENQUEUED state. Enqueues query if queueing is enabled.
+ * Foreman run will be blocked until query is enqueued.
+ * In case of failures (ex: queue timeout exception) will move query to
FAILED state.
+ */
+ private void enqueue() {
+ queryStateProcessor.moveToState(QueryState.ENQUEUED, null);
+
+ try {
+ queryRM.admit();
+ queryStateProcessor.moveToState(QueryState.STARTING, null);
+ } catch (QueueTimeoutException | QueryQueueException e) {
+ queryStateProcessor.moveToState(QueryState.FAILED, e);
+ } finally {
+ String queueName = queryRM.queueName();
+ queryManager.setQueueName(queueName == null ? "Unknown" : queueName);
--- End diff --
I'm going to violate our code review guidelines a bit and discuss design.
Sorry.
The above is a good place to talk about state design. We have the Foreman
taking actions and setting states. We have the query state processor taking
some actions based on those states and trying to enforce valid state
transitions. I think what we have is a clash of two systems.
The Foreman is a big bunch of code that handles the procedure of parsing,
preparing and launching a query, all in a single thread. The query state
processor appears to handle not just this, but events that arrive after the
query is started.
I wonder rather than having the Foreman do the setup work and the query
state processor try to follow the states, should we just bite the bullet and
make the query state processor be the actual query owner and manager?
The query manager provides events: parse, plan, enqueue, launch, etc. It
uses states to make sure events occur in the expected order.
All the Foreman does is create the environment, create the query manager,
and call the set of methods that we currently expect of the Foreman (from parse
to launch.)
In the future, if we can get queuing to be non-blocking (works by events,
not synchronized calls), then the Foreman would do planning through enqueueing.
The "accept query" action would invoke the launch action.
Then, the query manager not only enforces states, it also takes the actions
required when an event arrives in a given state.
Now, I realize all of the code in Foreman is existing. But, we are doing
refactoring, so now is the time to think through these issues.
---
