surrogate

Author: jan-elastic

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

@@ -12,7 +12,6 @@
 import org.antlr.v4.runtime.tree.ParseTree;
 import org.elasticsearch.Build;
 import org.elasticsearch.common.Strings;
-import org.elasticsearch.compute.operator.ChangePointOperator;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.dissect.DissectException;
 import org.elasticsearch.dissect.DissectParser;
@@ -467,21 +466,7 @@ public PlanFactory visitChangePointCommand(EsqlBaseParser.ChangePointCommandCont
             ctx.targetPvalue == null ? "pvalue" : visitQualifiedName(ctx.targetPvalue).name(),
             DataType.DOUBLE
         );
-        return child -> {
-            // ChangePoint should always run on the coordinating node after the data is collected
-            // in sorted order. This is enforced by adding OrderBy here.
-            // Furthermore, ChangePoint should be called with at most 1000 data points. That's
-            // enforced by the Limits here. The first Limit of 1001 data points is necessary to
-            // generate a possible warning, the second Limit of 1000 is to truncate the output.
-            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, ChangePointOperator.INPUT_VALUE_COUNT_LIMIT + 1, DataType.INTEGER),
-                orderBy
-            );
-            ChangePoint changePoint = new ChangePoint(src, limit, value, key, targetType, targetPvalue);
-            return new Limit(src, new Literal(Source.EMPTY, ChangePointOperator.INPUT_VALUE_COUNT_LIMIT, DataType.INTEGER), changePoint);
-        };
+        return child -> new ChangePoint(src, child, 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

@@ -9,22 +9,27 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.compute.operator.ChangePointOperator;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.AttributeSet;
 import org.elasticsearch.xpack.esql.core.expression.Expressions;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.expression.NameId;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.expression.NamedExpressions;
+import org.elasticsearch.xpack.esql.expression.Order;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.plan.GeneratingPlan;
 
 import java.io.IOException;
 import java.util.List;
 import java.util.Objects;
 
-public class ChangePoint extends UnaryPlan implements GeneratingPlan<ChangePoint> {
+public class ChangePoint extends UnaryPlan implements GeneratingPlan<ChangePoint>, SurrogateLogicalPlan {
+
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
         LogicalPlan.class,
         "ChangePoint",
@@ -161,4 +166,22 @@ public boolean equals(Object obj) {
             && Objects.equals(targetType, other.targetType)
             && Objects.equals(targetPvalue, other.targetPvalue);
     }
+
+    @Override
+    public LogicalPlan surrogate() {
+        // ChangePoint should always run on the coordinating node after the data is collected
+        // in sorted order. This is enforced by adding OrderBy here.
+        // Furthermore, ChangePoint should be called with at most 1000 data points. That's
+        // enforced by the Limits here. The first Limit of N+1 data points is necessary to
+        // generate a possible warning, the second Limit of N is to truncate the output.
+        Order order = new Order(source(), key, Order.OrderDirection.ASC, Order.NullsPosition.ANY);
+        OrderBy orderBy = new OrderBy(source(), child(), List.of(order));
+        Limit limit = new Limit(
+            source(),
+            new Literal(Source.EMPTY, ChangePointOperator.INPUT_VALUE_COUNT_LIMIT + 1, DataType.INTEGER),
+            orderBy
+        );
+        ChangePoint changePoint = new ChangePoint(source(), limit, value, key, targetType, targetPvalue);
+        return new Limit(source(), new Literal(Source.EMPTY, ChangePointOperator.INPUT_VALUE_COUNT_LIMIT, DataType.INTEGER), changePoint);
+    }
 }