Drafting SumOverTime transform for timeseries

Author: pabloem

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

@@ -274,3 +274,23 @@ max_cost:double | cluster:keyword | time_bucket:datetime
 11.625          | prod            | 2024-05-10T00:12:00.000Z
 11.5            | staging         | 2024-05-10T00:16:00.000Z
 ;
+
+sum_over_time
+required_capability: metrics_command
+required_capability: sum_over_time
+
+TS k8s | STATS cost=sum(sum_over_time(network.cost)) BY cluster, time_bucket = bucket(@timestamp,1minute) | SORT cost DESC, time_bucket DESC, cluster | LIMIT 10;
+
+cost:double | cluster:keyword | time_bucket:datetime
+12.375          | prod            | 2024-05-10T00:17:00.000Z
+12.375          | qa              | 2024-05-10T00:01:00.000Z
+12.25           | prod            | 2024-05-10T00:19:00.000Z
+12.125          | qa              | 2024-05-10T00:07:00.000Z
+12.125          | staging         | 2024-05-10T00:03:00.000Z
+11.875          | prod            | 2024-05-10T00:15:00.000Z
+11.875          | qa              | 2024-05-10T00:09:00.000Z
+11.75           | qa              | 2024-05-10T00:06:00.000Z
+11.625          | prod            | 2024-05-10T00:12:00.000Z
+11.5            | staging         | 2024-05-10T00:16:00.000Z
+;
+

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -1066,6 +1066,11 @@ public enum Cap {
          */
         FIRST_OVER_TIME(Build.current().isSnapshot()),
 
+        /**
+         * Support sum_over_time aggregation that gets evaluated per time-series
+         */
+        SUM_OVER_TIME(Build.current().isSnapshot()),
+
         /**
          * Resolve groupings before resolving references to groupings in the aggregations.
          */

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

@@ -37,6 +37,7 @@
 import org.elasticsearch.xpack.esql.expression.function.aggregate.SpatialExtent;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.StdDev;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Sum;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.SumOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Top;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Values;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.WeightedAvg;
@@ -449,6 +450,7 @@ private static FunctionDefinition[][] snapshotFunctions() {
                 def(Rate.class, Rate::withUnresolvedTimestamp, "rate"),
                 def(MaxOverTime.class, uni(MaxOverTime::new), "max_over_time"),
                 def(MinOverTime.class, uni(MinOverTime::new), "min_over_time"),
+                def(SumOverTime.class, uni(SumOverTime::new), "sum_over_time"),
                 def(AvgOverTime.class, uni(AvgOverTime::new), "avg_over_time"),
                 def(LastOverTime.class, LastOverTime::withUnresolvedTimestamp, "last_over_time"),
                 def(FirstOverTime.class, FirstOverTime::withUnresolvedTimestamp, "first_over_time"),

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

@@ -36,6 +36,7 @@ public static List<NamedWriteableRegistry.Entry> getNamedWriteables() {
             AvgOverTime.ENTRY,
             LastOverTime.ENTRY,
             FirstOverTime.ENTRY,
+            SumOverTime.ENTRY,
             // internal functions
             ToPartial.ENTRY,
             FromPartial.ENTRY,

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

@@ -0,0 +1,90 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.expression.function.aggregate;
+
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
+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.FunctionType;
+import org.elasticsearch.xpack.esql.expression.function.Param;
+
+import java.io.IOException;
+import java.util.List;
+
+import static java.util.Collections.emptyList;
+
+/**
+ * Similar to {@link Sum}, but it is used to calculate the sum of values over a time series from the given field.
+ */
+public class SumOverTime extends TimeSeriesAggregateFunction {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
+        Expression.class,
+        "SumOverTime",
+        SumOverTime::new
+    );
+
+    @FunctionInfo(
+        returnType = { "double", "integer", "long" },
+        description = "The sum over time value of a field.",
+        type = FunctionType.AGGREGATE
+    )
+    public SumOverTime(
+        Source source,
+        @Param(name = "field", type = { "aggregate_metric_double", "double", "integer", "long" }) Expression field
+    ) {
+        this(source, field, Literal.TRUE);
+    }
+
+    public SumOverTime(Source source, Expression field, Expression filter) {
+        super(source, field, filter, emptyList());
+    }
+
+    private SumOverTime(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    public SumOverTime withFilter(Expression filter) {
+        return new SumOverTime(source(), field(), filter);
+    }
+
+    @Override
+    protected NodeInfo<SumOverTime> info() {
+        return NodeInfo.create(this, SumOverTime::new, field(), filter());
+    }
+
+    @Override
+    public SumOverTime replaceChildren(List<Expression> newChildren) {
+        return new SumOverTime(source(), newChildren.get(0), newChildren.get(1));
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        return perTimeSeriesAggregation().resolveType();
+    }
+
+    @Override
+    public DataType dataType() {
+        return perTimeSeriesAggregation().dataType();
+    }
+
+    @Override
+    public Sum perTimeSeriesAggregation() {
+        return new Sum(source(), field(), filter());
+    }
+}

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/60_usage.yml

@@ -33,7 +33,7 @@ setup:
           path: /_query
           parameters: []
           # A snapshot function was removed in match_function_options, it can't work on mixed cluster tests otherwise.
-          capabilities: [ snapshot_test_for_telemetry, fn_byte_length, match_function_options, first_over_time]
+          capabilities: [ snapshot_test_for_telemetry, fn_byte_length, match_function_options, first_over_time, sum_over_time]
       reason: "Test that should only be executed on snapshot versions"
 
   - do: {xpack.usage: {}}