ESQL: Don't close over the plan in listener (#134393)

Don't close over the `PhysicalPlan` when executing the compute service.
It can be large.

Author: nik9000

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

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.Page;
+import org.elasticsearch.compute.operator.Driver;
 import org.elasticsearch.compute.operator.DriverCompletionInfo;
 import org.elasticsearch.compute.operator.DriverTaskRunner;
 import org.elasticsearch.compute.operator.FailureCollector;
@@ -658,31 +659,7 @@ public SourceProvider createSourceProvider(SourceFilter sourceFilter) {
                 throw new IllegalStateException("no drivers created");
             }
             LOGGER.debug("using {} drivers", drivers.size());
-            ActionListener<Void> driverListener = listener.map(ignored -> {
-                if (LOGGER.isDebugEnabled()) {
-                    LOGGER.debug(
-                        "finished {}",
-                        DriverCompletionInfo.includingProfiles(
-                            drivers,
-                            context.description(),
-                            clusterService.getClusterName().value(),
-                            transportService.getLocalNode().getName(),
-                            localPlan.toString()
-                        )
-                    );
-                }
-                if (context.configuration().profile()) {
-                    return DriverCompletionInfo.includingProfiles(
-                        drivers,
-                        context.description(),
-                        clusterService.getClusterName().value(),
-                        transportService.getLocalNode().getName(),
-                        localPlan.toString()
-                    );
-                } else {
-                    return DriverCompletionInfo.excludingProfiles(drivers);
-                }
-            });
+            ActionListener<Void> driverListener = addCompletionInfo(listener, drivers, context, localPlan);
             driverRunner.executeDrivers(
                 task,
                 drivers,
@@ -694,6 +671,50 @@ public SourceProvider createSourceProvider(SourceFilter sourceFilter) {
         }
     }
 
+    ActionListener<Void> addCompletionInfo(
+        ActionListener<DriverCompletionInfo> listener,
+        List<Driver> drivers,
+        ComputeContext context,
+        PhysicalPlan localPlan
+    ) {
+        /*
+         * We *really* don't want to close over the localPlan because it can
+         * be quite large, and it isn't tracked.
+         */
+        boolean needPlanString = LOGGER.isDebugEnabled() || context.configuration().profile();
+        String planString = needPlanString ? localPlan.toString() : null;
+        return listener.map(ignored -> {
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug(
+                    "finished {}",
+                    DriverCompletionInfo.includingProfiles(
+                        drivers,
+                        context.description(),
+                        clusterService.getClusterName().value(),
+                        transportService.getLocalNode().getName(),
+                        planString
+                    )
+                );
+                /*
+                 * planString *might* be null if we *just* set DEBUG to *after*
+                 * we built the listener but before we got here. That's something
+                 * we can live with.
+                 */
+            }
+            if (context.configuration().profile()) {
+                return DriverCompletionInfo.includingProfiles(
+                    drivers,
+                    context.description(),
+                    clusterService.getClusterName().value(),
+                    transportService.getLocalNode().getName(),
+                    planString
+                );
+            } else {
+                return DriverCompletionInfo.excludingProfiles(drivers);
+            }
+        });
+    }
+
     static PhysicalPlan reductionPlan(ExchangeSinkExec plan, boolean enable) {
         PhysicalPlan reducePlan = new ExchangeSourceExec(plan.source(), plan.output(), plan.isIntermediateAgg());
         if (enable) {