[ES|QL] Add support for counters to First/LastOverTime (#135676)

This commit adds support for running FirstOverTime and LastOverTime on
counter fields (counter_long, counter_integer, counter_double).

The return value is the non-counter variation.
For example, the following is a valid query and will return a double:
```
TS my_tsdb
| STATS max(first_over_time(my_counter_double_field))
  BY my_dimension, bucket(@timestamp, 10minute)
```

Author: limotova

docs/reference/query-languages/esql/_snippets/functions/types/first_over_time.md

@@ -796,6 +796,15 @@ public DataType noText() {
         return isString(this) ? KEYWORD : this;
     }
 
+    public DataType noCounter() {
+        return switch (this) {
+            case COUNTER_DOUBLE -> DOUBLE;
+            case COUNTER_INTEGER -> INTEGER;
+            case COUNTER_LONG -> LONG;
+            default -> this;
+        };
+    }
+
     public boolean isDate() {
         return switch (this) {
             case DATETIME, DATE_NANOS -> true;

docs/reference/query-languages/esql/_snippets/functions/types/last_over_time.md

@@ -295,12 +295,15 @@ yield switch ((randomIntBetween(0, 6))) {
                 if (counterField == null) {
                     yield null;
                 }
-                yield "rate(" + counterField + ")";
+                yield switch ((randomIntBetween(0, 2))) {
+                    case 0 -> "rate(" + counterField + ")";
+                    case 1 -> "first_over_time(" + counterField + ")";
+                    default -> "last_over_time(" + counterField + ")";
+                };
             }
             case 2 -> {
                 // numerics except aggregate_metric_double
                 // TODO: add to case 0 when support for aggregate_metric_double is added to these functions
-                // TODO: add to case 1 when support for counters is added
                 String numericFieldName = randomNumericField(previousOutput);
                 if (numericFieldName == null) {
                     yield null;

docs/reference/query-languages/esql/kibana/definition/functions/first_over_time.json

@@ -126,3 +126,42 @@ events:long | pod:keyword | time_bucket:datetime
 6           | two         | 2024-05-10T00:20:00.000Z
 ;
 
+first_over_time_counter_double
+required_capability: ts_command_v0
+required_capability: first_last_over_time_counter_support
+TS k8s
+| STATS sum = sum(first_over_time(network.total_cost)) by pod, time_bucket = bucket(@timestamp, 10minute)
+| SORT time_bucket, pod
+;
+
+sum:double | pod:keyword | time_bucket:datetime
+35.75      | one         | 2024-05-10T00:00:00.000Z
+34.375     | three       | 2024-05-10T00:00:00.000Z
+30.375     | two         | 2024-05-10T00:00:00.000Z
+96.5       | one         | 2024-05-10T00:10:00.000Z
+191.75     | three       | 2024-05-10T00:10:00.000Z
+163.25     | two         | 2024-05-10T00:10:00.000Z
+224.875    | one         | 2024-05-10T00:20:00.000Z
+142.875    | three       | 2024-05-10T00:20:00.000Z
+163.625    | two         | 2024-05-10T00:20:00.000Z
+;
+
+first_over_time_counter_long
+required_capability: ts_command_v0
+required_capability: first_last_over_time_counter_support
+TS k8s
+| STATS max = max(first_over_time(network.total_bytes_in)) by pod, time_bucket = bucket(@timestamp, 10minute)
+| SORT time_bucket, pod
+;
+
+max:long | pod:keyword | time_bucket:datetime
+1103     | one         | 2024-05-10T00:00:00.000Z
+1441     | three       | 2024-05-10T00:00:00.000Z
+1395     | two         | 2024-05-10T00:00:00.000Z
+6077     | one         | 2024-05-10T00:10:00.000Z
+4200     | three       | 2024-05-10T00:10:00.000Z
+3478     | two         | 2024-05-10T00:10:00.000Z
+10207    | one         | 2024-05-10T00:20:00.000Z
+9969     | three       | 2024-05-10T00:20:00.000Z
+8709     | two         | 2024-05-10T00:20:00.000Z
+;

docs/reference/query-languages/esql/kibana/definition/functions/last_over_time.json

@@ -276,3 +276,42 @@ events:long | pod:keyword | time_bucket:datetime
 5           | two         | 2024-05-10T00:20:00.000Z
 ;
 
+last_over_time_counter_double
+required_capability: ts_command_v0
+required_capability: first_last_over_time_counter_support
+TS k8s
+| STATS sum = sum(last_over_time(network.total_cost)) by pod, time_bucket = bucket(@timestamp, 10minute)
+| SORT time_bucket, pod
+;
+
+sum:double | pod:keyword | time_bucket:datetime
+190.125    | one         | 2024-05-10T00:00:00.000Z
+158.75     | three       | 2024-05-10T00:00:00.000Z
+148.25     | two         | 2024-05-10T00:00:00.000Z
+220.125    | one         | 2024-05-10T00:10:00.000Z
+271.75     | three       | 2024-05-10T00:10:00.000Z
+160.75     | two         | 2024-05-10T00:10:00.000Z
+190.625    | one         | 2024-05-10T00:20:00.000Z
+180.375    | three       | 2024-05-10T00:20:00.000Z
+191.0      | two         | 2024-05-10T00:20:00.000Z
+;
+
+last_over_time_counter_long
+required_capability: ts_command_v0
+required_capability: first_last_over_time_counter_support
+TS k8s
+| STATS max = max(last_over_time(network.total_bytes_in)) by pod, time_bucket = bucket(@timestamp, 10minute)
+| SORT time_bucket, pod
+;
+
+max:long | pod:keyword | time_bucket:datetime
+5963     | one         | 2024-05-10T00:00:00.000Z
+4169     | three       | 2024-05-10T00:00:00.000Z
+7900     | two         | 2024-05-10T00:00:00.000Z
+9254     | one         | 2024-05-10T00:10:00.000Z
+9195     | three       | 2024-05-10T00:10:00.000Z
+8031     | two         | 2024-05-10T00:10:00.000Z
+7286     | one         | 2024-05-10T00:20:00.000Z
+10797    | three       | 2024-05-10T00:20:00.000Z
+10277    | two         | 2024-05-10T00:20:00.000Z
+;

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/DataType.java

@@ -1562,6 +1562,11 @@ public enum Cap {
          */
         TS_COMMAND_V0(),
 
+        /**
+         * Add support for counter doubles, ints, and longs in first_ and last_over_time
+         */
+        FIRST_LAST_OVER_TIME_COUNTER_SUPPORT,
+
         FIX_ALIAS_ID_WHEN_DROP_ALL_AGGREGATES,
 
         /**

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/generator/EsqlQueryGenerator.java

@@ -56,7 +56,10 @@ public class FirstOverTime extends TimeSeriesAggregateFunction implements Option
         preview = true,
         examples = { @Example(file = "k8s-timeseries", tag = "first_over_time") }
     )
-    public FirstOverTime(Source source, @Param(name = "field", type = { "long", "integer", "double" }) Expression field) {
+    public FirstOverTime(
+        Source source,
+        @Param(name = "field", type = { "counter_long", "counter_integer", "counter_double", "long", "integer", "double" }) Expression field
+    ) {
         this(source, field, new UnresolvedAttribute(source, "@timestamp"));
     }
 
@@ -109,21 +112,20 @@ public FirstOverTime withFilter(Expression filter) {
 
     @Override
     public DataType dataType() {
-        return field().dataType();
+        return field().dataType().noCounter();
     }
 
     @Override
     protected TypeResolution resolveType() {
-        return isType(field(), dt -> dt.isNumeric() && dt != DataType.UNSIGNED_LONG, sourceText(), DEFAULT, "numeric except unsigned_long")
-            .and(
-                isType(
-                    timestamp,
-                    dt -> dt == DataType.DATETIME || dt == DataType.DATE_NANOS,
-                    sourceText(),
-                    SECOND,
-                    "date_nanos or datetime"
-                )
-            );
+        return isType(
+            field(),
+            dt -> (dt.noCounter().isNumeric() && dt != DataType.UNSIGNED_LONG) || dt == DataType.AGGREGATE_METRIC_DOUBLE,
+            sourceText(),
+            DEFAULT,
+            "numeric except unsigned_long"
+        ).and(
+            isType(timestamp, dt -> dt == DataType.DATETIME || dt == DataType.DATE_NANOS, sourceText(), SECOND, "date_nanos or datetime")
+        );
     }
 
     @Override
@@ -132,9 +134,9 @@ public AggregatorFunctionSupplier supplier() {
         // we can read the first encountered value for each group of `_tsid` and time bucket.
         final DataType type = field().dataType();
         return switch (type) {
-            case LONG -> new FirstLongByTimestampAggregatorFunctionSupplier();
-            case INTEGER -> new FirstIntByTimestampAggregatorFunctionSupplier();
-            case DOUBLE -> new FirstDoubleByTimestampAggregatorFunctionSupplier();
+            case LONG, COUNTER_LONG -> new FirstLongByTimestampAggregatorFunctionSupplier();
+            case INTEGER, COUNTER_INTEGER -> new FirstIntByTimestampAggregatorFunctionSupplier();
+            case DOUBLE, COUNTER_DOUBLE -> new FirstDoubleByTimestampAggregatorFunctionSupplier();
             case FLOAT -> new FirstFloatByTimestampAggregatorFunctionSupplier();
             default -> throw EsqlIllegalArgumentException.illegalDataType(type);
         };