ESQL - Include thread names in profile output

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

@@ -211,6 +211,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion MULTI_PROJECT = def(9_018_0_00);
     public static final TransportVersion STORED_SCRIPT_CONTENT_LENGTH = def(9_019_0_00);
     public static final TransportVersion JINA_AI_EMBEDDING_TYPE_SUPPORT_ADDED = def(9_020_0_00);
+    public static final TransportVersion ESQL_THREAD_NAME_IN_DRIVER_PROFILE = def(9_021_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
+     * @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

@@ -149,13 +149,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, "", 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, "", now - 1, now));
             assertThat(sleeps, equalTo(new DriverSleeps(expectedCounts, expectedFirst, expectedFirst)));
             assertXContent(sleeps, expectedCounts, expectedFirst, expectedFirst);
         }
@@ -172,13 +172,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, "", 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, "", 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("""