Fix for top

Author: julian-elastic

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

@@ -26,6 +26,20 @@ date:date | double:double | integer:integer | long:long
 [1985-02-18T00:00:00.000Z,1985-02-24T00:00:00.000Z] | [-9.81,-9.28] | [25324,25945] | [-9,-9]
 ;
 
+topFoldableExpressions
+required_capability: agg_top
+FROM employees
+| STATS
+    date = TOP(hire_date, 1+1, "dEsc"),
+    double = TOP(salary_change, 100-98, REVERSE("csed")),
+    integer = TOP(salary, 4-(1+1), Substring("Ascending",0,3)),
+    long = TOP(salary_change.long, 10 - 4*2, Concat("as","c"))
+;
+
+date:date | double:double | integer:integer | long:long
+[1999-04-30T00:00:00.000Z, 1997-05-19T00:00:00.000Z] | [14.74, 14.68] | [25324,25945] | [-9,-9]
+;
+
 topAllTypesDesc
 required_capability: agg_top
 FROM employees

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Top.java

@@ -20,8 +20,9 @@
 import org.elasticsearch.compute.aggregation.TopIpAggregatorFunctionSupplier;
 import org.elasticsearch.compute.aggregation.TopLongAggregatorFunctionSupplier;
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
+import org.elasticsearch.xpack.esql.capabilities.PostOptimizationVerificationAware;
+import org.elasticsearch.xpack.esql.common.Failures;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
-import org.elasticsearch.xpack.esql.core.expression.FoldContext;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
@@ -39,14 +40,15 @@
 
 import static java.util.Arrays.asList;
 import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
+import static org.elasticsearch.xpack.esql.common.Failure.fail;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.FIRST;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.SECOND;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.THIRD;
-import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isNotNullAndFoldable;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isNotNull;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isString;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isType;
 
-public class Top extends AggregateFunction implements ToAggregator, SurrogateExpression {
+public class Top extends AggregateFunction implements ToAggregator, SurrogateExpression, PostOptimizationVerificationAware {
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(Expression.class, "Top", Top::new);
 
     private static final String ORDER_ASC = "ASC";
@@ -116,16 +118,26 @@ Expression orderField() {
         return parameters().get(1);
     }
 
-    private int limitValue() {
-        return (int) limitField().fold(FoldContext.small() /* TODO remove me */);
-    }
-
-    private String orderRawValue() {
-        return BytesRefs.toString(orderField().fold(FoldContext.small() /* TODO remove me */));
+    private Integer limitValue() {
+        if (limitField() instanceof Literal literal) {
+            Object value = literal.value();
+            if (value instanceof Integer intValue) {
+                return intValue;
+            }
+        }
+        throw new EsqlIllegalArgumentException(
+            format(null, "Limit value must be an integer in [{}], found [{}]", sourceText(), limitField())
+        );
     }
 
     private boolean orderValue() {
-        return orderRawValue().equalsIgnoreCase(ORDER_ASC);
+        if (orderField() instanceof Literal literal) {
+            String order = BytesRefs.toString(literal.value());
+            if (ORDER_ASC.equalsIgnoreCase(order) || ORDER_DESC.equalsIgnoreCase(order)) {
+                return order.equalsIgnoreCase(ORDER_ASC);
+            }
+        }
+        throw new EsqlIllegalArgumentException("Order value must be a literal, found: " + orderField());
     }
 
     @Override
@@ -148,31 +160,114 @@ protected TypeResolution resolveType() {
             "ip",
             "string",
             "numeric except unsigned_long or counter types"
-        ).and(isNotNullAndFoldable(limitField(), sourceText(), SECOND))
+        ).and(isNotNull(limitField(), sourceText(), SECOND))
             .and(isType(limitField(), dt -> dt == DataType.INTEGER, sourceText(), SECOND, "integer"))
-            .and(isNotNullAndFoldable(orderField(), sourceText(), THIRD))
+            .and(isNotNull(orderField(), sourceText(), THIRD))
             .and(isString(orderField(), sourceText(), THIRD));
 
         if (typeResolution.unresolved()) {
             return typeResolution;
         }
 
-        var limit = limitValue();
-        var order = orderRawValue();
-
-        if (limit <= 0) {
-            return new TypeResolution(format(null, "Limit must be greater than 0 in [{}], found [{}]", sourceText(), limit));
+        TypeResolution result = resolveTypeLimit();
+        if (result.equals(TypeResolution.TYPE_RESOLVED) == false) {
+            return result;
+        }
+        result = resolveTypeOrder();
+        if (result.equals(TypeResolution.TYPE_RESOLVED) == false) {
+            return result;
         }
+        return TypeResolution.TYPE_RESOLVED;
+    }
 
-        if (order.equalsIgnoreCase(ORDER_ASC) == false && order.equalsIgnoreCase(ORDER_DESC) == false) {
-            return new TypeResolution(
-                format(null, "Invalid order value in [{}], expected [{}, {}] but got [{}]", sourceText(), ORDER_ASC, ORDER_DESC, order)
-            );
+    /**
+     * We check that the limit is not null and that if it is a literal, it is a positive integer
+     * We will do a more thorough check in the postOptimizationVerification once folding is done.
+     */
+    private TypeResolution resolveTypeLimit() {
+        Expression limit = limitField();
+        if (limit == null) {
+            return new TypeResolution(format(null, "Limit must be a constant integer in [{}], found [{}]", sourceText(), limit));
+        }
+        if (limit instanceof Literal literal) {
+            if (literal.value() == null) {
+                return new TypeResolution(format(null, "Limit must be a constant integer in [{}], found [{}]", sourceText(), limit));
+            }
+            int value = (Integer) literal.value();
+            if (value <= 0) {
+                return new TypeResolution(format(null, "Limit must be greater than 0 in [{}], found [{}]", sourceText(), value));
+            }
         }
+        return TypeResolution.TYPE_RESOLVED;
+    }
 
+    /**
+     * We check that the order is not null and that if it is a literal, it is one of the two valid values: "asc" or "desc".
+     * We will do a more thorough check in the postOptimizationVerification once folding is done.
+     */
+    private TypeResolution resolveTypeOrder() {
+        Expression order = orderField();
+        if (order == null) {
+            return new TypeResolution(format(null, "Order must be a valid string in [{}], found [{}]", sourceText(), order));
+        }
+        if (order instanceof Literal literal) {
+            if (literal.value() == null) {
+                return new TypeResolution(
+                    format(null, "Invalid order value in [{}], expected [{}, {}] but got [{}]", sourceText(), ORDER_ASC, ORDER_DESC, order)
+                );
+            }
+            String value = BytesRefs.toString(literal.value());
+            if (value == null || value.equalsIgnoreCase(ORDER_ASC) == false && value.equalsIgnoreCase(ORDER_DESC) == false) {
+                return new TypeResolution(
+                    format(null, "Invalid order value in [{}], expected [{}, {}] but got [{}]", sourceText(), ORDER_ASC, ORDER_DESC, order)
+                );
+            }
+        }
         return TypeResolution.TYPE_RESOLVED;
     }
 
+    @Override
+    public void postOptimizationVerification(Failures failures) {
+        postOptimizationVerificationLimit(failures);
+        postOptimizationVerificationOrder(failures);
+    }
+
+    private void postOptimizationVerificationLimit(Failures failures) {
+        Expression limit = limitField();
+        if (limit == null) {
+            failures.add(fail(limit, "Limit must be a constant integer in [{}], found [{}]", sourceText(), limit));
+        }
+        if (limit instanceof Literal literal) {
+            int value = (Integer) literal.value();
+            if (value <= 0) {
+                failures.add(fail(limit, "Limit must be greater than 0 in [{}], found [{}]", sourceText(), value));
+            }
+        } else {
+            // it is expected that the expression is a literal after folding
+            // we fail if it is not a literal
+            failures.add(fail(limit, "Limit must be a constant integer in [{}], found [{}]", sourceText(), limit));
+        }
+    }
+
+    private void postOptimizationVerificationOrder(Failures failures) {
+        Expression order = orderField();
+        if (order == null) {
+            failures.add(fail(order, "Order must be a valid string in [{}], found [{}]", sourceText(), order));
+        }
+        if (order instanceof Literal literal) {
+            String value = BytesRefs.toString(literal.value());
+            if (value == null || value.equalsIgnoreCase(ORDER_ASC) == false && value.equalsIgnoreCase(ORDER_DESC) == false) {
+                failures.add(
+                    fail(order, "Invalid order value in [{}], expected [{}, {}] but got [{}]", sourceText(), ORDER_ASC, ORDER_DESC, order)
+                );
+            }
+        } else {
+            // it is expected that the expression is a literal after folding
+            // we fail if it is not a literal
+            failures.add(fail(order, "Order must be a valid string in [{}], found [{}]", sourceText(), order));
+        }
+    }
+
     @Override
     public DataType dataType() {
         return field().dataType().noText();
@@ -215,15 +310,20 @@ public AggregatorFunctionSupplier supplier() {
     @Override
     public Expression surrogate() {
         var s = source();
-
-        if (limitValue() == 1) {
-            if (orderValue()) {
-                return new Min(s, field());
-            } else {
-                return new Max(s, field());
+        try {
+            if (limitValue() == 1) {
+                if (orderValue()) {
+                    return new Min(s, field());
+                } else {
+                    return new Max(s, field());
+                }
             }
+        } catch (EsqlIllegalArgumentException e) {
+            // If the limit is not a literal or is not a positive integer, we cannot create a surrogate
+            // so we return null to indicate that no surrogate can be created.
+            // This is possible if the limit is an expression, and folding has not been done yet.
+            return null;
         }
-
         return null;
     }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/TopTests.java

@@ -262,7 +262,7 @@ public static Iterable<Object[]> parameters() {
                             new TestCaseSupplier.TypedData(null, DataType.INTEGER, "limit").forceLiteral(),
                             new TestCaseSupplier.TypedData(new BytesRef("desc"), DataType.KEYWORD, "order").forceLiteral()
                         ),
-                        "second argument of [source] cannot be null, received [limit]"
+                        "Limit must be a constant integer in [source], found [null]"
                     )
                 ),
                 new TestCaseSupplier(
@@ -273,7 +273,7 @@ public static Iterable<Object[]> parameters() {
                             new TestCaseSupplier.TypedData(1, DataType.INTEGER, "limit").forceLiteral(),
                             new TestCaseSupplier.TypedData(null, DataType.KEYWORD, "order").forceLiteral()
                         ),
-                        "third argument of [source] cannot be null, received [order]"
+                        "Invalid order value in [source], expected [ASC, DESC] but got [null]"
                     )
                 )
             )