Move OrderBy/Limit to the logical plan

Author: jan-elastic

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/LogicalPlanBuilder.java

@@ -457,7 +457,13 @@ public PlanFactory visitChangePointCommand(EsqlBaseParser.ChangePointCommandCont
         NamedExpression key = visitQualifiedName(ctx.key);
         Attribute targetType = new ReferenceAttribute(src, visitQualifiedName(ctx.targetType).name(), DataType.TEXT);
         Attribute targetPvalue = new ReferenceAttribute(src, visitQualifiedName(ctx.targetPvalue).name(), DataType.DOUBLE);
-        return child -> new ChangePoint(src, child, value, key, targetType, targetPvalue);
+        return child -> {
+            // ChangePoint should always run on the coordinating node after all data is collected
+            // in sorted order. This is enforced by adding OrderBy and Limit here.
+            OrderBy orderBy = new OrderBy(src, child, List.of(new Order(src, key, Order.OrderDirection.ASC, Order.NullsPosition.ANY)));
+            Limit limit = new Limit(src, new Literal(Source.EMPTY, 1000, DataType.INTEGER), orderBy);
+            return new ChangePoint(src, limit, value, key, targetType, targetPvalue);
+        };
     }
 
     private static Tuple<Mode, String> parsePolicyName(Token policyToken) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/ChangePoint.java

@@ -66,12 +66,7 @@ private ChangePoint(StreamInput in) throws IOException {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        Source.EMPTY.writeTo(out);
-        out.writeNamedWriteable(child());
-        out.writeNamedWriteable(value);
-        out.writeNamedWriteable(key);
-        out.writeNamedWriteable(targetType);
-        out.writeNamedWriteable(targetPvalue);
+        throw new UnsupportedOperationException("ChangePoint should run on the coordinating node, so never be serialized.");
     }
 
     @Override

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ChangePointExec.java

@@ -64,12 +64,7 @@ private ChangePointExec(StreamInput in) throws IOException {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        Source.EMPTY.writeTo(out);
-        out.writeNamedWriteable(child());
-        out.writeNamedWriteable(value);
-        out.writeNamedWriteable(key);
-        out.writeNamedWriteable(targetType);
-        out.writeNamedWriteable(targetPvalue);
+        throw new UnsupportedOperationException("ChangePoint should run on the coordinating node, so never be serialized.");
     }
 
     @Override

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

@@ -703,7 +703,6 @@ 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()).append(changePoint.targetPvalue()).build();
         return source.with(new ChangePointOperator.Factory(layout.get(changePoint.value().id()).channel()), layout);

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

@@ -95,26 +95,6 @@ private PhysicalPlan mapUnary(UnaryPlan unary) {
             return new TopNExec(topN.source(), mappedChild, topN.order(), topN.limit(), null);
         }
 
-        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,
-                changePoint.value(),
-                changePoint.key(),
-                changePoint.targetType(),
-                changePoint.targetPvalue()
-            );
-        }
-
         //
         // Pipeline operators
         //

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

@@ -177,25 +177,6 @@ private PhysicalPlan mapUnary(UnaryPlan unary) {
             return new TopNExec(topN.source(), mappedChild, topN.order(), topN.limit(), null);
         }
 
-        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,
-                changePoint.value(),
-                changePoint.key(),
-                changePoint.targetType(),
-                changePoint.targetPvalue()
-            );
-        }
-
         //
         // Pipeline operators
         //

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

@@ -13,6 +13,7 @@
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.FoldContext;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
+import org.elasticsearch.xpack.esql.plan.logical.ChangePoint;
 import org.elasticsearch.xpack.esql.plan.logical.Dissect;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.plan.logical.Eval;
@@ -26,6 +27,7 @@
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalRelation;
 import org.elasticsearch.xpack.esql.plan.logical.show.ShowInfo;
 import org.elasticsearch.xpack.esql.plan.physical.AggregateExec;
+import org.elasticsearch.xpack.esql.plan.physical.ChangePointExec;
 import org.elasticsearch.xpack.esql.plan.physical.DissectExec;
 import org.elasticsearch.xpack.esql.plan.physical.EnrichExec;
 import org.elasticsearch.xpack.esql.plan.physical.EvalExec;
@@ -98,6 +100,17 @@ static PhysicalPlan mapUnary(UnaryPlan p, PhysicalPlan child) {
             return new MvExpandExec(mvExpand.source(), child, mvExpand.target(), mvExpand.expanded());
         }
 
+        if (p instanceof ChangePoint changePoint) {
+            return new ChangePointExec(
+                changePoint.source(),
+                child,
+                changePoint.value(),
+                changePoint.key(),
+                changePoint.targetType(),
+                changePoint.targetPvalue()
+            );
+        }
+
         return unsupported(p);
     }
 

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

@@ -74,6 +74,7 @@
 import org.elasticsearch.xpack.esql.parser.EsqlParser;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.esql.plan.physical.ChangePointExec;
 import org.elasticsearch.xpack.esql.plan.physical.HashJoinExec;
 import org.elasticsearch.xpack.esql.plan.physical.LocalSourceExec;
 import org.elasticsearch.xpack.esql.plan.physical.OutputExec;
@@ -567,12 +568,9 @@ private Throwable reworkException(Throwable th) {
 
     // Asserts that the serialization and deserialization of the plan creates an equivalent plan.
     private void opportunisticallyAssertPlanSerialization(PhysicalPlan plan) {
-
-        // skip plans with localSourceExec
-        if (plan.anyMatch(p -> p instanceof LocalSourceExec || p instanceof HashJoinExec)) {
+        if (plan.anyMatch(p -> p instanceof LocalSourceExec || p instanceof HashJoinExec || p instanceof ChangePointExec)) {
             return;
         }
-
         SerializationTestUtils.assertSerialization(plan, configuration);
     }