Added TopNAgg csv tests for count and multiple aggs, which can be pushed down to lucene

Author: ivancea

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

@@ -0,0 +1,37 @@
+count
+from employees | stats c = count(height) | sort c | limit 100;
+
+c:long
+100
+;
+
+countGrouping
+from employees | stats c = count(height) by languages.long | sort languages.long | limit 100;
+
+c:long | languages.long:long
+15     | 1
+19     | 2
+17     | 3
+18     | 4
+21     | 5
+10     | null
+;
+
+multipleAggs
+from employees | stats c = count(height), m = max(height) | sort c | limit 100;
+
+c:long | m:double
+100    | 2.1
+;
+
+multipleAggsGrouping
+from employees | stats c = count(height), m = max(height) by languages.long | sort languages.long | limit 100;
+
+c:long | m:double | languages.long:long
+15     | 2.06     | 1
+19     | 2.1      | 2
+17     | 2.1      | 3
+18     | 2.0      | 4
+21     | 2.1      | 5
+10     | 2.1      | null
+; 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/PushStatsToSource.java

@@ -22,7 +22,7 @@
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Count;
 import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext;
 import org.elasticsearch.xpack.esql.optimizer.PhysicalOptimizerRules;
-import org.elasticsearch.xpack.esql.plan.physical.AggregateExec;
+import org.elasticsearch.xpack.esql.plan.physical.AbstractAggregateExec;
 import org.elasticsearch.xpack.esql.plan.physical.EsQueryExec;
 import org.elasticsearch.xpack.esql.plan.physical.EsStatsQueryExec;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
@@ -41,10 +41,10 @@
 /**
  * Looks for the case where certain stats exist right before the query and thus can be pushed down.
  */
-public class PushStatsToSource extends PhysicalOptimizerRules.ParameterizedOptimizerRule<AggregateExec, LocalPhysicalOptimizerContext> {
+public class PushStatsToSource extends PhysicalOptimizerRules.ParameterizedOptimizerRule<AbstractAggregateExec, LocalPhysicalOptimizerContext> {
 
     @Override
-    protected PhysicalPlan rule(AggregateExec aggregateExec, LocalPhysicalOptimizerContext context) {
+    protected PhysicalPlan rule(AbstractAggregateExec aggregateExec, LocalPhysicalOptimizerContext context) {
         PhysicalPlan plan = aggregateExec;
         if (aggregateExec.child() instanceof EsQueryExec queryExec) {
             var tuple = pushableStats(aggregateExec, context);
@@ -72,7 +72,7 @@ protected PhysicalPlan rule(AggregateExec aggregateExec, LocalPhysicalOptimizerC
     }
 
     private Tuple<List<Attribute>, List<EsStatsQueryExec.Stat>> pushableStats(
-        AggregateExec aggregate,
+        AbstractAggregateExec aggregate,
         LocalPhysicalOptimizerContext context
     ) {
         AttributeMap.Builder<EsStatsQueryExec.Stat> statsBuilder = AttributeMap.builder();