Add sorting

Author: jan-elastic

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java

@@ -703,6 +703,7 @@ private PhysicalOperation planMvExpand(MvExpandExec mvExpandExec, LocalExecution
     }
 
     private PhysicalOperation planChangePoint(ChangePointExec changePoint, LocalExecutionPlannerContext context) {
+        // TODO: should this be planned locally?
         PhysicalOperation source = plan(changePoint.child(), context);
         Layout layout = source.layout.builder()
             .append(changePoint.targetType())

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/LocalMapper.java

@@ -11,6 +11,10 @@
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.expression.Order;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
 import org.elasticsearch.xpack.esql.plan.logical.BinaryPlan;
 import org.elasticsearch.xpack.esql.plan.logical.ChangePoint;
@@ -92,6 +96,11 @@ private PhysicalPlan mapUnary(UnaryPlan unary) {
         }
 
         if (unary instanceof ChangePoint changePoint) {
+            // TODO: ChangePoint shouldn't run on the local node
+            // TODO: fix hardcoded 1000
+            mappedChild = new TopNExec(changePoint.source(), mappedChild,
+                List.of(new Order(changePoint.source(), changePoint.key(), Order.OrderDirection.ASC, Order.NullsPosition.ANY)),
+                new Literal(Source.EMPTY, 1000, DataType.INTEGER), null);
             return new ChangePointExec(
                 changePoint.source(),
                 mappedChild,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/Mapper.java

@@ -11,7 +11,11 @@
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.Holder;
+import org.elasticsearch.xpack.esql.expression.Order;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
 import org.elasticsearch.xpack.esql.plan.logical.BinaryPlan;
 import org.elasticsearch.xpack.esql.plan.logical.ChangePoint;
@@ -175,6 +179,9 @@ private PhysicalPlan mapUnary(UnaryPlan unary) {
 
         if (unary instanceof ChangePoint changePoint) {
             mappedChild = addExchangeForFragment(changePoint, mappedChild);
+            mappedChild = new TopNExec(changePoint.source(), mappedChild,
+                List.of(new Order(changePoint.source(), changePoint.key(), Order.OrderDirection.ASC, Order.NullsPosition.ANY)),
+                new Literal(Source.EMPTY, 1000, DataType.INTEGER), null);
             return new ChangePointExec(
                 changePoint.source(),
                 mappedChild,