ESQL: Use less memory in listener (#114358)

Use less memory in the top level listener by fetching the output
attributes from the plan before starting rather than after finishing.
The plan *can* be very large so let's not hold on to it longer than we
have to.

Author: nik9000

docs/changelog/114358.yaml

@@ -0,0 +1,5 @@
+pr: 114358
+summary: "ESQL: Use less memory in listener"
+area: ES|QL
+type: enhancement
+issues: []

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java

@@ -60,6 +60,7 @@
 import org.elasticsearch.xpack.esql.action.EsqlExecutionInfo;
 import org.elasticsearch.xpack.esql.action.EsqlQueryAction;
 import org.elasticsearch.xpack.esql.action.EsqlSearchShardsAction;
+import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.enrich.EnrichLookupService;
 import org.elasticsearch.xpack.esql.plan.physical.ExchangeSinkExec;
 import org.elasticsearch.xpack.esql.plan.physical.ExchangeSourceExec;
@@ -206,13 +207,19 @@ public void execute(
         );
         long start = configuration.getQueryStartTimeNanos();
         String local = RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY;
+        /*
+         * Grab the output attributes here, so we can pass them to
+         * the listener without holding on to a reference to the
+         * entire plan.
+         */
+        List<Attribute> outputAttributes = physicalPlan.output();
         try (
             Releasable ignored = exchangeSource.addEmptySink();
             // this is the top level ComputeListener called once at the end (e.g., once all clusters have finished for a CCS)
             var computeListener = ComputeListener.create(local, transportService, rootTask, execInfo, start, listener.map(r -> {
                 long tookTimeNanos = System.nanoTime() - configuration.getQueryStartTimeNanos();
                 execInfo.overallTook(new TimeValue(tookTimeNanos, TimeUnit.NANOSECONDS));
-                return new Result(physicalPlan.output(), collectedPages, r.getProfiles(), execInfo);
+                return new Result(outputAttributes, collectedPages, r.getProfiles(), execInfo);
             }))
         ) {
             // run compute on the coordinator