Make timestamp optional

Author: jan-elastic

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/ChangePointLongAggregator.java

@@ -249,8 +249,7 @@ Block toBlock(Function<SingleState, LongArray> getArray, BlockFactory blockFacto
             }
         }
 
-        void enableGroupIdTracking(SeenGroupIds seenGroupIds) {
-        }
+        void enableGroupIdTracking(SeenGroupIds seenGroupIds) {}
 
         @Override
         public void close() {

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

@@ -1,4 +1,4 @@
-change point
+change point with timestamp argument
 # required_capability: change_point
 
 FROM k8s
@@ -12,3 +12,18 @@ cp:keyword
 "{""type"":{""stationary"":{}}}"                                                                                                    | qa
 "{""type"":{""indeterminable"":{""reason"":""not enough buckets to calculate change_point. Requires at least [22]; found [18]""}}}" | staging
 ;
+
+change point without timestamp argument
+# required_capability: change_point
+
+FROM k8s
+  | STATS count=COUNT() BY @timestamp=BUCKET(@timestamp, 1 minute), cluster
+  | STATS cp=CHANGE_POINT(count) BY cluster
+  | SORT cluster
+;
+
+cp:keyword                                                                                                                          | cluster:keyword
+"{""type"":{""indeterminable"":{""reason"":""not enough buckets to calculate change_point. Requires at least [22]; found [21]""}}}" | prod
+"{""type"":{""stationary"":{}}}"                                                                                                    | qa
+"{""type"":{""indeterminable"":{""reason"":""not enough buckets to calculate change_point. Requires at least [22]; found [18]""}}}" | staging
+;

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

@@ -14,10 +14,12 @@
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
+import org.elasticsearch.xpack.esql.core.expression.UnresolvedAttribute;
 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.function.FunctionInfo;
+import org.elasticsearch.xpack.esql.expression.function.OptionalArgument;
 import org.elasticsearch.xpack.esql.expression.function.Param;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.planner.ToAggregator;
@@ -29,21 +31,20 @@
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.SECOND;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isType;
 
-public class ChangePoint extends AggregateFunction implements ToAggregator {
+public class ChangePoint extends AggregateFunction implements OptionalArgument, ToAggregator {
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
         Expression.class,
         "ChangePoint",
         ChangePoint::new
     );
 
-    // TODO: make "timestamp" field optional
     @FunctionInfo(returnType = { "string" }, description = "...", isAggregation = true)
     public ChangePoint(
         Source source,
         @Param(name = "field", type = { "double", "integer", "long" }, description = "field") Expression field,
-        @Param(name = "timestamp", type = { "date_nanos", "datetime", "double", "integer", "long" }) Expression timestamp
+        @Param(optional = true, name = "timestamp", type = { "date_nanos", "datetime", "double", "integer", "long" }) Expression timestamp
     ) {
-        this(source, field, Literal.TRUE, timestamp);
+        this(source, field, Literal.TRUE, timestamp != null ? timestamp : new UnresolvedAttribute(source, "@timestamp"));
     }
 
     public ChangePoint(Source source, Expression field, Expression filter, Expression timestamp) {
@@ -98,10 +99,6 @@ public AggregateFunction withFilter(Expression filter) {
 
     @Override
     public AggregatorFunctionSupplier supplier(List<Integer> inputChannels) {
-        // if (inputChannels.size() != 2 && inputChannels.size() != 3) {
-        // throw new IllegalArgumentException("change point requires two for raw input or three channels for partial input; got " +
-        // inputChannels);
-        // }
         final DataType type = field().dataType();
         return switch (type) {
             case LONG -> new ChangePointLongAggregatorFunctionSupplier(inputChannels);