[8.x] ESQL: Preserve single aggregate when all attributes are pruned (#126397) (#127126)

* ESQL: Preserve single aggregate when all attributes are pruned (#126397)

* Avoid using `EmptyAttribute`

* fix compile error

---------

Co-authored-by: kanoshiou <[email protected]>

Author: fang-xing-esql

docs/changelog/126397.yaml

@@ -0,0 +1,6 @@
+pr: 126397
+summary: "ESQL: Preserve single aggregate when all attributes are pruned"
+area: ES|QL
+type: bug
+issues:
+ - 126392

x-pack/plugin/esql/qa/testFixtures/src/main/resources/eval.csv-spec

@@ -637,3 +637,39 @@ foo:keyword
 ;
 
 
+evalAfterAvgUsingSameName
+required_capability: remove_empty_attribute_in_merging_output
+from employees 
+| stats avg = avg(salary) 
+| eval avg = 12
+;
+
+avg:integer
+12
+;
+
+evalAfterStatsUsingSameName
+required_capability: remove_empty_attribute_in_merging_output
+from employees 
+| stats count = count(emp_no), median = median(salary), top_salaries = TOP(salary, 3, "desc")
+| keep median, top_salaries
+| rename top_salaries as median
+| eval median = 12
+;
+
+median:integer
+12
+;
+
+evalAfterStatsUsingSameName2
+required_capability: remove_empty_attribute_in_merging_output
+ROW foo = [10, 11, 12]
+| mv_expand foo
+| stats sum = sum(foo), max = max(foo)
+| rename sum as max
+| eval max = 13
+;
+
+max:integer
+13
+;

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

@@ -277,6 +277,12 @@ public enum Cap {
          */
         RENAME_SEQUENTIAL_PROCESSING,
 
+        /**
+         * Support for removing empty attribute in merging output.
+         * See <a href="https://github.com/elastic/elasticsearch/issues/126392"> EVAL after STATS produces an empty column #126392 </a>
+         */
+        REMOVE_EMPTY_ATTRIBUTE_IN_MERGING_OUTPUT,
+
         /**
          * Fix for union-types when some indexes are missing the required field. Done in #111932.
          */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PruneColumns.java

@@ -11,7 +11,6 @@
 import org.elasticsearch.compute.data.BlockUtils;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.xpack.esql.core.expression.AttributeSet;
-import org.elasticsearch.xpack.esql.core.expression.EmptyAttribute;
 import org.elasticsearch.xpack.esql.core.expression.Expressions;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
@@ -65,7 +64,7 @@ public LogicalPlan apply(LogicalPlan plan) {
                             if (aggregate.groupings().isEmpty()) {
                                 p = new LocalRelation(
                                     aggregate.source(),
-                                    List.of(new EmptyAttribute(aggregate.source())),
+                                    List.of(Expressions.attribute(aggregate.aggregates().get(0))),
                                     LocalSupplier.of(
                                         new Block[] { BlockUtils.constantBlock(PlannerUtils.NON_BREAKING_BLOCK_FACTORY, null, 1) }
                                     )

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java

@@ -2725,6 +2725,24 @@ private static List<String> orderNames(TopN topN) {
         return topN.order().stream().map(o -> as(o.child(), NamedExpression.class).name()).toList();
     }
 
+    /**
+     * Expects
+     * Eval[[2[INTEGER] AS x]]
+     * \_Limit[1000[INTEGER],false]
+     *   \_LocalRelation[[{e}#9],[ConstantNullBlock[positions=1]]]
+     */
+    public void testEvalAfterStats() {
+        var plan = optimizedPlan("""
+            ROW foo = 1
+            | STATS x = max(foo)
+            | EVAL x = 2
+            """);
+        var eval = as(plan, Eval.class);
+        var limit = as(eval.child(), Limit.class);
+        var localRelation = as(limit.child(), LocalRelation.class);
+        assertThat(Expressions.names(eval.output()), contains("x"));
+    }
+
     public void testCombineLimitWithOrderByThroughFilterAndEval() {
         LogicalPlan plan = optimizedPlan("""
             from test