ESQL - Include thread names in profile output

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -182,6 +182,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INCLUDE_INDEX_MODE_IN_GET_DATA_STREAM = def(9_023_0_00);
     public static final TransportVersion MAX_OPERATION_SIZE_REJECTIONS_ADDED = def(9_024_0_00);
     public static final TransportVersion RETRY_ILM_ASYNC_ACTION_REQUIRE_ERROR = def(9_025_0_00);
+    public static final TransportVersion ESQL_THREAD_NAME_IN_DRIVER_PROFILE = def(9_026_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/DriverSleeps.java

@@ -31,18 +31,28 @@
 public record DriverSleeps(Map<String, Long> counts, List<Sleep> first, List<Sleep> last) implements Writeable, ToXContentObject {
     /**
      * A record of a time the driver slept.
-     * @param reason The reason the driver slept
-     * @param sleep Millis since epoch when the driver slept
-     * @param wake Millis since epoch when the driver woke, or 0 if it is currently sleeping
+     *
+     * @param reason     The reason the driver slept
+     * @param threadName The name of the thread this driver was running on when it went to sleep
+     * @param sleep      Millis since epoch when the driver slept
+     * @param wake       Millis since epoch when the driver woke, or 0 if it is currently sleeping
      */
-    public record Sleep(String reason, long sleep, long wake) implements Writeable, ToXContentObject {
+    public record Sleep(String reason, String threadName, long sleep, long wake) implements Writeable, ToXContentObject {
         Sleep(StreamInput in) throws IOException {
-            this(in.readString(), in.readLong(), in.readLong());
+            this(
+                in.readString(),
+                in.getTransportVersion().onOrAfter(TransportVersions.ESQL_THREAD_NAME_IN_DRIVER_PROFILE) ? in.readString() : "",
+                in.readLong(),
+                in.readLong()
+            );
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeString(reason);
+            if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_THREAD_NAME_IN_DRIVER_PROFILE)) {
+                out.writeString(threadName);
+            }
             out.writeLong(sleep);
             out.writeLong(wake);
         }
@@ -51,7 +61,7 @@ Sleep wake(long now) {
             if (isStillSleeping() == false) {
                 throw new IllegalStateException("Already awake.");
             }
-            return new Sleep(reason, sleep, now);
+            return new Sleep(reason, Thread.currentThread().getName(), sleep, now);
         }
 
         public boolean isStillSleeping() {
@@ -62,6 +72,7 @@ public boolean isStillSleeping() {
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
             builder.field("reason", reason);
+            builder.field("thread_name", threadName);
             builder.timestampFieldsFromUnixEpochMillis("sleep_millis", "sleep", sleep);
             if (wake > 0) {
                 builder.timestampFieldsFromUnixEpochMillis("wake_millis", "wake", wake);
@@ -138,7 +149,7 @@ public DriverSleeps wake(long now) {
     private List<Sleep> append(List<Sleep> old, String reason, long now) {
         List<Sleep> sleeps = new ArrayList<>(old.size() + 1);
         sleeps.addAll(old);
-        sleeps.add(new Sleep(reason, now, 0));
+        sleeps.add(new Sleep(reason, Thread.currentThread().getName(), now, 0));
         return Collections.unmodifiableList(sleeps);
     }
 
@@ -147,7 +158,7 @@ private List<Sleep> rollOnto(List<Sleep> old, String reason, long now) {
         for (int i = 1; i < old.size(); i++) {
             sleeps.add(old.get(i));
         }
-        sleeps.add(new Sleep(reason, now, 0));
+        sleeps.add(new Sleep(reason, Thread.currentThread().getName(), now, 0));
         return Collections.unmodifiableList(sleeps);
     }
 

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/DriverProfileTests.java

@@ -41,8 +41,8 @@ public void testToXContent() {
             ),
             new DriverSleeps(
                 Map.of("driver time", 1L),
-                List.of(new DriverSleeps.Sleep("driver time", 1, 1)),
-                List.of(new DriverSleeps.Sleep("driver time", 1, 1))
+                List.of(new DriverSleeps.Sleep("driver time", "", 1, 1)),
+                List.of(new DriverSleeps.Sleep("driver time", "", 1, 1))
             )
         );
         assertThat(Strings.toString(status, true, true), equalTo("""

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/DriverSleepsTests.java

@@ -17,7 +17,9 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Formatter;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -57,55 +59,64 @@ public void testEmptyToXContent() {
     }
 
     public void testSleepingToXContent() {
-        assertThat(Strings.toString(DriverSleeps.empty().sleep("driver iterations", 1723555763000L), true, true), equalTo("""
+        Formatter formatter = new Formatter(Locale.US);
+        String expected = formatter.format("""
             {
               "counts" : {
                 "driver iterations" : 1
               },
               "first" : [
                 {
                   "reason" : "driver iterations",
+                  "thread_name" : "%1$s",
                   "sleep" : "2024-08-13T13:29:23.000Z",
                   "sleep_millis" : 1723555763000
                 }
               ],
               "last" : [
                 {
                   "reason" : "driver iterations",
+                  "thread_name" : "%1$s",
                   "sleep" : "2024-08-13T13:29:23.000Z",
                   "sleep_millis" : 1723555763000
                 }
               ]
-            }"""));
+            }""", Thread.currentThread().getName()).out().toString();
+        assertThat(Strings.toString(DriverSleeps.empty().sleep("driver iterations", 1723555763000L), true, true), equalTo(expected));
     }
 
     public void testWakingToXContent() {
+        Formatter formatter = new Formatter(Locale.US);
+        String expected = formatter.format("""
+            {
+              "counts" : {
+                "driver iterations" : 1
+              },
+              "first" : [
+                {
+                  "reason" : "driver iterations",
+                  "thread_name" : "%1$s",
+                  "sleep" : "2024-08-13T13:29:23.000Z",
+                  "sleep_millis" : 1723555763000,
+                  "wake" : "2024-08-13T13:31:03.000Z",
+                  "wake_millis" : 1723555863000
+                }
+              ],
+              "last" : [
+                {
+                  "reason" : "driver iterations",
+                  "thread_name" : "%1$s",
+                  "sleep" : "2024-08-13T13:29:23.000Z",
+                  "sleep_millis" : 1723555763000,
+                  "wake" : "2024-08-13T13:31:03.000Z",
+                  "wake_millis" : 1723555863000
+                }
+              ]
+            }""", Thread.currentThread().getName()).out().toString();
+
         assertThat(
             Strings.toString(DriverSleeps.empty().sleep("driver iterations", 1723555763000L).wake(1723555863000L), true, true),
-            equalTo("""
-                {
-                  "counts" : {
-                    "driver iterations" : 1
-                  },
-                  "first" : [
-                    {
-                      "reason" : "driver iterations",
-                      "sleep" : "2024-08-13T13:29:23.000Z",
-                      "sleep_millis" : 1723555763000,
-                      "wake" : "2024-08-13T13:31:03.000Z",
-                      "wake_millis" : 1723555863000
-                    }
-                  ],
-                  "last" : [
-                    {
-                      "reason" : "driver iterations",
-                      "sleep" : "2024-08-13T13:29:23.000Z",
-                      "sleep_millis" : 1723555763000,
-                      "wake" : "2024-08-13T13:31:03.000Z",
-                      "wake_millis" : 1723555863000
-                    }
-                  ]
-                }""")
+            equalTo(expected)
         );
     }
 
@@ -149,13 +160,13 @@ public void testTracking() throws IOException {
             expectedCounts.compute(reason, (k, v) -> v == null ? 1 : v + 1);
 
             sleeps = sleeps.sleep(reason, now);
-            expectedFirst.add(new DriverSleeps.Sleep(reason, now, 0));
+            expectedFirst.add(new DriverSleeps.Sleep(reason, Thread.currentThread().getName(), now, 0));
             assertThat(sleeps, equalTo(new DriverSleeps(expectedCounts, expectedFirst, expectedFirst)));
             assertXContent(sleeps, expectedCounts, expectedFirst, expectedFirst);
 
             now++;
             sleeps = sleeps.wake(now);
-            expectedFirst.set(expectedFirst.size() - 1, new DriverSleeps.Sleep(reason, now - 1, now));
+            expectedFirst.set(expectedFirst.size() - 1, new DriverSleeps.Sleep(reason, Thread.currentThread().getName(), now - 1, now));
             assertThat(sleeps, equalTo(new DriverSleeps(expectedCounts, expectedFirst, expectedFirst)));
             assertXContent(sleeps, expectedCounts, expectedFirst, expectedFirst);
         }
@@ -172,13 +183,13 @@ public void testTracking() throws IOException {
 
             sleeps = sleeps.sleep(reason, now);
             expectedLast.remove(0);
-            expectedLast.add(new DriverSleeps.Sleep(reason, now, 0));
+            expectedLast.add(new DriverSleeps.Sleep(reason, Thread.currentThread().getName(), now, 0));
             assertThat(sleeps, equalTo(new DriverSleeps(expectedCounts, expectedFirst, expectedLast)));
             assertXContent(sleeps, expectedCounts, expectedFirst, expectedLast);
 
             now++;
             sleeps = sleeps.wake(now);
-            expectedLast.set(expectedLast.size() - 1, new DriverSleeps.Sleep(reason, now - 1, now));
+            expectedLast.set(expectedLast.size() - 1, new DriverSleeps.Sleep(reason, Thread.currentThread().getName(), now - 1, now));
             assertThat(sleeps, equalTo(new DriverSleeps(expectedCounts, expectedFirst, expectedLast)));
             assertXContent(sleeps, expectedCounts, expectedFirst, expectedLast);
         }

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/DriverStatusTests.java

@@ -45,8 +45,8 @@ public void testToXContent() {
             List.of(new OperatorStatus("ExchangeSink", ExchangeSinkOperatorStatusTests.simple())),
             new DriverSleeps(
                 Map.of("driver time", 1L),
-                List.of(new DriverSleeps.Sleep("driver time", 1, 1)),
-                List.of(new DriverSleeps.Sleep("driver time", 1, 1))
+                List.of(new DriverSleeps.Sleep("driver time", "", 1, 1)),
+                List.of(new DriverSleeps.Sleep("driver time", "", 1, 1))
             )
         );
         assertThat(Strings.toString(status, true, true), equalTo("""