Sometimes EventListenerMangager.queryCompleted() called too early

For happy path and parsable queries, we have the following order of calls:
- query is added to the QueryTracker
- EventListener.queryCompleted() is called

For really bad queries, the ones that do not even parse to a proper SQL, the order is inverted.
As a result, when implementing an EventListener that wants to access query detailed information
we have to use a workaround for this edge case and artificially invert the sequence of calls.
EventListener will typically use a retry mechanism combined with switching to a separate thread,
which adds unnecessary complexity and multi-threading to a code that otherwise would be
single-threaded.

core/trino-main/src/main/java/io/trino/dispatcher/DispatchManager.java

@@ -22,6 +22,7 @@
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.context.Context;
 import io.trino.Session;
+import io.trino.event.QueryMonitor;
 import io.trino.execution.QueryIdGenerator;
 import io.trino.execution.QueryInfo;
 import io.trino.execution.QueryManagerConfig;
@@ -56,6 +57,7 @@
 import static io.trino.execution.QueryState.RUNNING;
 import static io.trino.spi.StandardErrorCode.QUERY_TEXT_TOO_LARGE;
 import static io.trino.tracing.ScopedSpan.scopedSpan;
+import static io.trino.util.Failures.toFailure;
 import static io.trino.util.StatementUtils.getQueryType;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
@@ -80,6 +82,7 @@ public class DispatchManager
     private final QueryTracker<DispatchQuery> queryTracker;
 
     private final QueryManagerStats stats = new QueryManagerStats();
+    private final QueryMonitor queryMonitor;
 
     @Inject
     public DispatchManager(
@@ -94,7 +97,8 @@ public DispatchManager(
             SessionPropertyManager sessionPropertyManager,
             Tracer tracer,
             QueryManagerConfig queryManagerConfig,
-            DispatchExecutor dispatchExecutor)
+            DispatchExecutor dispatchExecutor,
+            QueryMonitor queryMonitor)
     {
         this.queryIdGenerator = requireNonNull(queryIdGenerator, "queryIdGenerator is null");
         this.queryPreparer = requireNonNull(queryPreparer, "queryPreparer is null");
@@ -112,6 +116,7 @@ public DispatchManager(
         this.dispatchExecutor = dispatchExecutor.getExecutor();
 
         this.queryTracker = new QueryTracker<>(queryManagerConfig, dispatchExecutor.getScheduledExecutor());
+        this.queryMonitor = requireNonNull(queryMonitor, "queryMonitor is null");
     }
 
     @PostConstruct
@@ -236,6 +241,11 @@ private <C> void createQueryInternal(QueryId queryId, Span querySpan, Slug slug,
             Optional<String> preparedSql = Optional.ofNullable(preparedQuery).flatMap(PreparedQuery::getPrepareSql);
             DispatchQuery failedDispatchQuery = failedDispatchQueryFactory.createFailedDispatchQuery(session, query, preparedSql, Optional.empty(), throwable);
             queryCreated(failedDispatchQuery);
+            // maintain proper order of calls such that EventListener has access to QueryInfo
+            // - add query to tracker
+            // - fire query created event
+            // - fire query completed event
+            queryMonitor.queryImmediateFailureEvent(failedDispatchQuery.getBasicQueryInfo(), toFailure(throwable));
             querySpan.setStatus(StatusCode.ERROR, throwable.getMessage())
                     .recordException(throwable)
                     .end();

core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQueryFactory.java

@@ -24,7 +24,6 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 
-import static io.trino.util.Failures.toFailure;
 import static java.util.Objects.requireNonNull;
 
 public class FailedDispatchQueryFactory
@@ -58,7 +57,6 @@ public FailedDispatchQuery createFailedDispatchQuery(Session session, String que
         BasicQueryInfo queryInfo = failedDispatchQuery.getBasicQueryInfo();
 
         queryMonitor.queryCreatedEvent(queryInfo);
-        queryMonitor.queryImmediateFailureEvent(queryInfo, toFailure(throwable));
 
         return failedDispatchQuery;
     }