ESQL: Shrink toString for OutputOperator (#102499)

We sync a status object every thousand iterations or so and if the
`toString` for the operators is super big they take a bunch of time.
Like, the pages have to be small and the `toString` has to be really
expensive for it to matter. Well, `OutputOperator` can have a very big
`toString` because it lists every column it'll write. If there are
thousands of columns this can cost. Well, a little. In a kind of broken
benchmark it's about 3% of the runtime. But shrinking the `toString` is
pretty easy, so let's do it.

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

@@ -13,8 +13,6 @@ import java.util.List;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
-import static java.util.stream.Collectors.joining;
-
 /**
  * Sink operator that calls a given listener for each page received. The listener receives both the page as well as schema information,
  * i.e. the names of the rows that are outputted.
@@ -36,7 +34,7 @@ public class OutputOperator extends SinkOperator {
 
         @Override
         public String describe() {
-            return "OutputOperator[columns = " + columns.stream().collect(joining(", ")) + "]";
+            return OutputOperator.describe(columns);
         }
     }
 
@@ -75,10 +73,18 @@ public class OutputOperator extends SinkOperator {
 
     @Override
     public String toString() {
+        return describe(columns);
+    }
+
+    private static String describe(List<String> columns) {
         StringBuilder sb = new StringBuilder();
-        sb.append(this.getClass().getSimpleName()).append("[");
-        sb.append("columns=").append(columns).append(", ");
-        sb.append("pageConsumer=").append(pageConsumer);
+        sb.append("OutputOperator").append("[");
+        sb.append("columns = ");
+        if (columns.size() <= 10) {
+            sb.append(columns);
+        } else {
+            sb.append('[').append(columns.size()).append(" columns").append(']');
+        }
         sb.append("]");
         return sb.toString();
     }

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

@@ -26,7 +26,7 @@ public abstract class SinkOperator implements Operator {
     /**
      * A factory for creating sink operators.
      */
-    public interface SinkOperatorFactory extends Describable {
+    public interface SinkOperatorFactory extends OperatorFactory, Describable {
         /** Creates a new sink operator. */
         SinkOperator get(DriverContext driverContext);
     }

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

@@ -0,0 +1,50 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.compute.operator;
+
+import org.elasticsearch.common.util.BigArrays;
+
+import java.util.List;
+import java.util.stream.IntStream;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class OutputOperatorTests extends AnyOperatorTestCase {
+    @Override
+    protected Operator.OperatorFactory simple(BigArrays bigArrays) {
+        return new OutputOperator.OutputOperatorFactory(List.of("a"), p -> p, p -> {});
+    }
+
+    @Override
+    protected String expectedDescriptionOfSimple() {
+        return "OutputOperator[columns = [a]]";
+    }
+
+    @Override
+    protected String expectedToStringOfSimple() {
+        return expectedDescriptionOfSimple();
+    }
+
+    private Operator.OperatorFactory big() {
+        return new OutputOperator.OutputOperatorFactory(IntStream.range(0, 20).mapToObj(i -> "a" + i).toList(), p -> p, p -> {});
+    }
+
+    private String expectedDescriptionOfBig() {
+        return "OutputOperator[columns = [20 columns]]";
+    }
+
+    public void testBigToString() {
+        try (Operator operator = big().get(driverContext())) {
+            assertThat(operator.toString(), equalTo(expectedDescriptionOfBig()));
+        }
+    }
+
+    public void testBigDescription() {
+        assertThat(big().describe(), equalTo(expectedDescriptionOfBig()));
+    }
+}

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlActionTaskIT.java

@@ -99,7 +99,7 @@ public class EsqlActionTaskIT extends AbstractEsqlIntegTestCase {
             \\_AggregationOperator[mode = FINAL, aggs = sum of longs]
             \\_ProjectOperator[projection = [0]]
             \\_LimitOperator[limit = 500]
-            \\_OutputOperator[columns = sum(pause_me)]""";
+            \\_OutputOperator[columns = [sum(pause_me)]]""";
 
         XContentBuilder mapping = JsonXContent.contentBuilder().startObject();
         mapping.startObject("runtime");